Imp node clustering

.cluster-clients-mocharc.json

@@ -0,0 +1,17 @@
+{
+  "recursive": true,
+  "sort": false,
+  "color": true,
+  "allow-uncaught": true,
+  "parallel": false,
+  "reporter": "list",
+  "require": [
+    "dotenv/config"
+  ],
+  "exit": true,
+  "spec": [
+    "./test/cluster/cluster-clients.spec.js"
+  ],
+  "exclude": "test/**/*.spec.js",
+  "timeout": 120000
+}

.cluster-mocharc.json

@@ -0,0 +1,22 @@
+{
+  "recursive": true,
+  "sort": false,
+  "color": true,
+  "allow-uncaught": true,
+  "parallel": false,
+  "reporter": "list",
+  "require": [
+    "dotenv/config",
+    "./test/cluster/utils.js",
+    "./test/cluster/fixtures.js"
+  ],
+  "exit": true,
+  "spec": [
+    "./test/cluster/same-node.spec.js",
+    "./test/cluster/cross-node.spec.js",
+    "./test/cluster/multi-devices.spec.js",
+    "./test/cluster/node-crash.spec.js"
+  ],
+  "exclude": "test/**/*.spec.js",
+  "timeout": 120000
+}

.env.docker

@@ -1,4 +1,4 @@
-NODE_ENV=development
+NODE_ENV=testing
 
 APP_PORT=9001
 
@@ -19,23 +19,38 @@ CONVERSATION_MAX_PARTICIPANTS=50
 CONVERSATION_PRELOAD_COUNT=30
 MESSAGE_PRELOAD_COUNT=30
 SEARCH_PRELOAD_COUNT=30
+
 JWT_ACCESS_SECRET=4c09be6a35bec6d4089cde4be5ca57a6ee85da5739579cbecb0b8253f268eca2
 JWT_ACCESS_TOKEN_EXPIRES_IN=10800
 JWT_REFRESH_TOKEN_EXPIRES_IN=1209600
 JWT_REFRESH_SECRET=4c09be6a35bec6d4089cde4be5ca57a6ee85da5739579cbecb0b8253f268eca2
+
 ENCRYPTION_MESSAGE_EXPIRED_IN=2592000 # 30 days
 ENCRYPTION_DEVICE_TOKEN_EXPIRES_IN=2592000 # 30 days
+ENCRYPTION_MESSAGE_EXPIRED_IN=2592000
+ENCRYPTION_DEVICE_TOKEN_EXPIRES_IN=2592000
+
 OPERATIONS_LOG_EXPIRES_IN=1209600
-NODE_CLUSTER_DATA_EXPIRES_IN=30000
+NODE_CLUSTER_DATA_EXPIRES_IN=10000
+
 FILE_UPLOAD_URL_EXPIRES_IN=3600
 FILE_DOWNLOAD_URL_EXPIRES_IN=604800
-ENCRYPTION_MESSAGE_EXPIRED_IN=2592000
-ENCRYPTION_DEVICE_TOKEN_EXPIRES_IN=2592000
+
 SSL_KEY_FILE_NAME=
 SSL_CERT_FILE_NAME=
 
 # set this env if you want to override the os.hostname() value
 HOSTNAME=NodeName
 CORS_ORIGIN=http://localhost:10000
 COOKIE_SECRET=4c09be6a35bec6d4089cde4be5ca57a6ee85da5739579cbecb0b8253f268eca2
+HTTP_ADMIN_API_KEY=docker-secret-key
+
+# Testing
+RUN_NODE_1_CMD='APP_PORT=9001 APP_TCP_PORT=8001 CORS_ORIGIN=http://localhost:3001 npm start'
+RUN_NODE_2_CMD='APP_PORT=9002 APP_TCP_PORT=8002 CORS_ORIGIN=http://localhost:3002 npm start'
+
+NODE_1_WS_ENDPOINT=ws://localhost:9001
+NODE_1_HTTP_ENDPOINT=http://localhost:9001
 
+NODE_2_WS_ENDPOINT=ws://localhost:9002
+NODE_2_HTTP_ENDPOINT=http://localhost:9002

.env.example

@@ -101,3 +101,19 @@ GOOGLE_GENERATIVE_AI_API_KEY=
 SERVICE_OTP_TOKEN_EXPIRES_IN=300000
 RESEND_SENDER="SAMASupport <onboarding@resend.dev>"
 RESEND_API_KEY=
+
+
+# Testing
+RUN_NODE_1_CMD='APP_PORT=9001 APP_TCP_PORT=8001 CORS_ORIGIN=http://localhost:3001 npm start'
+RUN_NODE_2_CMD='APP_PORT=9002 APP_TCP_PORT=8002 CORS_ORIGIN=http://localhost:3002 npm start'
+
+NODE_1_WS_ENDPOINT=ws://localhost:9001
+NODE_1_HTTP_ENDPOINT=http://localhost:9001
+
+NODE_2_WS_ENDPOINT=ws://localhost:9002
+NODE_2_HTTP_ENDPOINT=http://localhost:9002
+
+TEST_CLIENTS_COUNT=10
+TEST_CLIENT_ORG_ID=683db99d3874471b4dd36c69
+TEST_CLIENT_WS_ENDPOINT=ws://localhost:9001
+TEST_CLIENT_HTTP_ENDPOINT=http://localhost:9001

.github/workflows/tests.yml

@@ -5,18 +5,22 @@ on:
     branches:
       - main
 
+  workflow_dispatch:
+
 jobs:
   run_tests:
     runs-on: self-hosted
     steps:
       - uses: actions/checkout@v3
       - name: Build the SAMA resources
+        env:
+          NODE_ENV: testing
         run: docker compose -f docker-compose-full.yml up -d --build
 
       - name: Run migrations
         run: docker compose exec sama-server sh -c "MONGODB_URL=mongodb://172.25.0.4/samatests npm run migrate-mongo-up"
 
-      - name: Run tests
+      - name: Run tests single node
         run: docker compose exec sama-server sh -c "MONGODB_URL=mongodb://172.25.0.4/samatests npm run test"
 
       - name: Stop the SAMA resources

.mocharc.json

@@ -6,5 +6,6 @@
   "node-option": ["experimental-loader=./sama-loader.mjs", "import=dotenv/config"],
   "exit": true,
   "spec": "test/**/*.spec.js",
+  "exclude": "test/cluster/*.spec.js",
   "timeout": 60000
 }

APIs/JSON/controllers/http/activity.js

@@ -13,7 +13,7 @@ class HttpActivityController extends BaseHttpController {
     const response = new Response()
 
     const HttpActivityOnlineListOperation = ServiceLocatorContainer.use("HttpActivityOnlineListOperation")
-    const onlineListResponse = await HttpActivityOnlineListOperation.perform(res.fakeWsSessionKey, payload)
+    const onlineListResponse = await HttpActivityOnlineListOperation.perform(res, payload)
 
     const responsePayload = isCountRequest ? { count: onlineListResponse } : { users: onlineListResponse }
 

APIs/JSON/controllers/http/auth.js

@@ -16,7 +16,7 @@ class HttpAuthController extends BaseHttpController {
     const httpAuthOperation = ServiceLocatorContainer.use("HttpUserAuthOperation")
 
     const { user, newAccessToken, accessTokenExpiredAt, newRefreshToken } = await httpAuthOperation.perform(
-      res.fakeWsSessionKey,
+      res,
       res.parsedHeaders,
       res.parsedSignedCookies,
       payload
@@ -43,7 +43,7 @@ class HttpAuthController extends BaseHttpController {
   async logout(res, req) {
     const httpLogoutOperation = ServiceLocatorContainer.use("HttpUserLogoutOperation")
 
-    const refreshTokenRecord = await httpLogoutOperation.perform(res.fakeWsSessionKey, res.parsedHeaders, res.parsedSignedCookies)
+    const { refreshTokenRecord, isWasLastUserSession } = await httpLogoutOperation.perform(res, res.parsedHeaders, res.parsedSignedCookies)
 
     const httpResponse = new HttpResponse(200, {}, { success: true }).addCookie("refresh_token", refreshTokenRecord.token, {
       maxAge: 0,
@@ -52,15 +52,19 @@ class HttpAuthController extends BaseHttpController {
       sameSite: "lax",
     })
 
-    return new Response()
-      .setHttpResponse(httpResponse)
-      .updateLastActivityStatus(
+    const response = new Response().setHttpResponse(httpResponse)
+
+    if (isWasLastUserSession) {
+      response.updateLastActivityStatus(
         new LastActivityStatusResponse(
           refreshTokenRecord.organization_id,
           refreshTokenRecord.user_id,
           MAIN_CONSTANTS.LAST_ACTIVITY_STATUS.OFFLINE
         )
       )
+    }
+
+    return response
   }
 }
 

APIs/JSON/controllers/http/message.js

@@ -20,7 +20,7 @@ class HttpMessageController extends BaseHttpController {
     const response = new Response()
 
     const httpMessageCreateOperation = ServiceLocatorContainer.use("HttpMessageCreateOperation")
-    const createMessageResponse = await httpMessageCreateOperation.perform(res.fakeWsSessionKey, payload)
+    const createMessageResponse = await httpMessageCreateOperation.perform(res, payload)
     const { organizationId, messageId, message, deliverMessages, cId, participantIds, modifiedFields, botMessage } = createMessageResponse
 
     deliverMessages.forEach((event) => {
@@ -48,7 +48,7 @@ class HttpMessageController extends BaseHttpController {
     const payload = res.parsedBody
 
     const messageSendSystemOperation = ServiceLocatorContainer.use("HttpMessageSendSystemOperation")
-    const { organizationId, cId, recipientsIds, systemMessage } = await messageSendSystemOperation.perform(res.fakeWsSessionKey, payload)
+    const { organizationId, cId, recipientsIds, systemMessage } = await messageSendSystemOperation.perform(res, payload)
 
     const deliverMessage = new DeliverMessage(organizationId, new SystemMessageResponse(systemMessage), true)
       .setConversationDestination(cId)
@@ -63,7 +63,7 @@ class HttpMessageController extends BaseHttpController {
     const payload = res.parsedBody
 
     const messageReadOperation = ServiceLocatorContainer.use("HttpMessageReadOperation")
-    const { organizationId, readMessagesGroups } = await messageReadOperation.perform(res.fakeWsSessionKey, payload)
+    const { organizationId, readMessagesGroups } = await messageReadOperation.perform(res, payload)
 
     const response = new Response()
 
@@ -79,7 +79,7 @@ class HttpMessageController extends BaseHttpController {
     const payload = res.parsedBody
 
     const messageEditOperation = ServiceLocatorContainer.use("HttpMessageEditOperation")
-    const { organizationId, cId, participantsIds, editedMessage } = await messageEditOperation.perform(res.fakeWsSessionKey, payload)
+    const { organizationId, cId, participantsIds, editedMessage } = await messageEditOperation.perform(res, payload)
 
     return new Response()
       .setHttpResponse(new HttpResponse(200, {}, { success: true }))
@@ -94,10 +94,7 @@ class HttpMessageController extends BaseHttpController {
     const payload = res.parsedBody
 
     const messageReactionOperation = ServiceLocatorContainer.use("HttpMessageReactionOperation")
-    const { organizationId, cId, participantsIds, messageReactionsUpdate } = await messageReactionOperation.perform(
-      res.fakeWsSessionKey,
-      payload
-    )
+    const { organizationId, cId, participantsIds, messageReactionsUpdate } = await messageReactionOperation.perform(res, payload)
 
     return new Response()
       .setHttpResponse(new HttpResponse(200, {}, { success: true }))
@@ -112,7 +109,7 @@ class HttpMessageController extends BaseHttpController {
     const payload = res.parsedBody
 
     const messageDeleteOperation = ServiceLocatorContainer.use("HttpMessageDeleteOperation")
-    const { organizationId, cId, participantsIds, deletedMessages } = await messageDeleteOperation.perform(res.fakeWsSessionKey, payload)
+    const { organizationId, cId, participantsIds, deletedMessages } = await messageDeleteOperation.perform(res, payload)
 
     const response = new Response()
 

APIs/JSON/controllers/http/organization.js

@@ -12,7 +12,7 @@ class HttpOrganizationController extends BaseHttpController {
     const response = new Response()
 
     const httpOrganizationCreateOperation = ServiceLocatorContainer.use("HttpOrganizationCreateOperation")
-    const organization = await httpOrganizationCreateOperation.perform(res.fakeWsSessionKey, payload)
+    const organization = await httpOrganizationCreateOperation.perform(res, payload)
 
     return response.setHttpResponse(new HttpResponse(200, {}, { organization: organization.visibleParams() }))
   }

APIs/JSON/controllers/http/stats.js

@@ -12,7 +12,7 @@ class HttpServerStatsController extends BaseHttpController {
     const response = new Response()
 
     const httpOrganizationCreateOperation = ServiceLocatorContainer.use("HttpStatsCollectOperation")
-    const stats = await httpOrganizationCreateOperation.perform(res.fakeWsSessionKey, payload)
+    const stats = await httpOrganizationCreateOperation.perform(res, payload)
 
     return response.setHttpResponse(new HttpResponse(200, {}, stats))
   }

APIs/JSON/controllers/users.js

@@ -56,11 +56,15 @@ class UsersController extends BaseJSONController {
     const { id: requestId } = data
 
     const userLogoutOperation = ServiceLocatorContainer.use("UserLogoutOperation")
-    const { organizationId, userId } = await userLogoutOperation.perform(ws)
+    const { organizationId, userId, isWasLastUserSession } = await userLogoutOperation.perform(ws)
 
-    return new Response()
-      .addBackMessage({ response: { id: requestId, success: true } })
-      .updateLastActivityStatus(new LastActivityStatusResponse(organizationId, userId, MAIN_CONSTANTS.LAST_ACTIVITY_STATUS.OFFLINE))
+    const response = new Response().addBackMessage({ response: { id: requestId, success: true } })
+
+    if (isWasLastUserSession) {
+      response.updateLastActivityStatus(new LastActivityStatusResponse(organizationId, userId, MAIN_CONSTANTS.LAST_ACTIVITY_STATUS.OFFLINE))
+    }
+
+    return response
   }
 
   async send_otp(ws, data) {

APIs/JSON/routes/packet_processor.js

@@ -33,11 +33,13 @@ class PacketJsonProcessor extends BasePacketProcessor {
   async processMessageOrError(ws, message) {
     let responseData
     let json
+    let requestTrackId
     try {
       json = this.#parseMessage(message)
+      requestTrackId = json?.request?.id ?? json?.message?.id ?? json?.system_message?.id ?? json?.id
       updateStoreContext(
         MAIN_CONSTANTS.LOGGER_BINDINGS_NAMES.REQUEST_ID,
-        json?.request?.id ?? json?.message?.id ?? json?.system_message?.id ?? json?.id ?? MAIN_CONSTANTS.LOGGER_BINDINGS_NAMES.NO_REQUEST_ID
+        requestTrackId ?? MAIN_CONSTANTS.LOGGER_BINDINGS_NAMES.NO_REQUEST_ID
       )
       responseData = await this.#processMessage(ws, json)
       if (!responseData) {
@@ -49,15 +51,15 @@ class PacketJsonProcessor extends BasePacketProcessor {
       if (json?.request) {
         errorBackMessage = {
           response: {
-            id: json.request?.id,
+            id: json.request?.id ?? requestTrackId,
             error: error.cause || error.message,
           },
         }
       } else {
         const topLevelElement = json ? Object.keys(json)[0] : void 0
         errorBackMessage = {
           [topLevelElement]: {
-            id: json?.[topLevelElement]?.id,
+            id: json?.[topLevelElement]?.id ?? requestTrackId,
             error: error.cause || error.message,
           },
         }

Dockerfile

@@ -4,7 +4,7 @@ WORKDIR /app
 
 COPY package*.json ./
 
-RUN npm install 
+RUN npm install --omit=dev
 
 COPY . .
 

Dockerfile.local

@@ -4,7 +4,7 @@ WORKDIR /app
 
 COPY package*.json ./
 
-RUN npm install 
+RUN npm install
 
 COPY . .