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.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

.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/auth.js

@@ -43,7 +43,11 @@ 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.fakeWsSessionKey,
+      res.parsedHeaders,
+      res.parsedSignedCookies
+    )
 
     const httpResponse = new HttpResponse(200, {}, { success: true }).addCookie("refresh_token", refreshTokenRecord.token, {
       maxAge: 0,
@@ -52,15 +56,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/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) {

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 --omit=dev
 
 COPY . .
 

README.md

@@ -77,3 +77,68 @@ Want to support us?
 <a href="https://www.buymeacoffee.com/khomenkoigor" target="_blank"><img src="https://cdn.buymeacoffee.com/buttons/v2/default-blue.png" alt="Buy Me A Coffee" style="height: 60px !important;width: 217px !important;" ></a>
 
 <!-- GitAds-Verify: 35EQ2ZS2ZWQJ4EF324OI3WJ83GY8BDGX -->
+
+## Node Clustering
+
+### What data is stored in Redis by each node ?
+
+#### sama-node-data
+
+Record key in redis like `sama-node-data:{node-endpoint}` where `{node-endpoint}` ws url like `ws://192.168.1.13:55495/`, so redis record will be `sama-node-data:ws://192.168.1.13:55495/`. The record is HSET with ttl that equal config value `ws.cluster.nodeExpiresIn` in seconds + 5 seconds. The record has key/values like `ip/port/host` etc.
+
+#### sama-node-users
+
+Record key in redis like `sama-node-users:{node-endpoint}`. The record is SET that contains list user users connected to this node, user item is string like `{organizationId}:{userId}:{deviceId}`. Item example: `683db8ecdb9dee54f53304c0:683db9ed29aecb9feb4afb73:38400000-8cf0-11bd-b23e-10b96e40000d`
+
+#### sama-user-devices
+
+Record key in redis like `sama-user-devices:{organizationId}:{userId}`. The record is SET that contains list user devices ids connected to all nodes. Item example: `38400000-8cf0-11bd-b23e-10b96e40000d`
+
+#### sama-user-data
+
+Record key in redis like `sama-user-data:{userId}:{deviceId}`. The record in HSET that contains extra user session data e.g. active/inactive status
+
+### How nodes discover each other ?
+
+Each node write record key in Redis like `sama-node-data:{node-endpoint}` where `{node-endpoint}` it is own ws url. Node create this record on start and update on sync cluster (check method `syncCluster`). In sync method node retrieve all records from Redis with prefix `sama-node-data:`, the parse record ws url part (`{node-endpoint}`) and try establish ws connection if it doesn't already exist.
+
+
+### How nodes connect to each other ?
+
+If node-A does not have establish connection with node-B, node-A will create ws connection (check method `createConnectionWithNode`). When ws with node-B endpoint (`node-endpoint`) successfully opened - node-A send like 'ping' message with own network data (`ip/port/host` etc.) to node-B (check method `shareCurrentNodeInfo`), node-B send to node-A like 'pong' message with own network data, node-A after receiving 'pong' message - connection successfully established and handshake finished, app resolve promise and save ws connection in local object (check prop `clusterNodesConnections`). Node-B after send 'pong' message start connecting with node-A by same flow
+
+#### What happens when connection breaks between 2 nodes ?
+
+If connecting breaks between 2 nodes, node detect it by ws event close, and start reconnecting (check method `startNodeReconnecting` and `promiseQueueWithJittering`, prop `closeNodesConnections`), this method try 3 time create connection until the first successful with delay '1/4/9 seconds', flow like 1 second delay - try connect (if success return) - 4 seconds delay - try connect (if success return) - 9 seconds delay - try connect (if success return) - failed
+
+### What happens when new node added to cluster ?
+
+2 Ways (node-A,node-B - running, node-C - new node in cluster)
+
+- Flow `How nodes discover each other` and then `How nodes connect to each other`. Node-C retrieve redis records (`sama-node-data`) created by node-A,node-B and create ws connections with each node (open ws and then connect flow with 'ping'/'pong').
+
+- Node-A,node-B in method `syncCluster` with call with interval that equal config value `ws.cluster.nodeExpiresIn` in ms, retrieve `sama-node-data` records and see new node record created by node-C and then create connection with node-C
+
+### What happens when existing node removed from cluster ?
+
+Example with node-A and node-B
+
+When node-B disconnect from cluster, ws connection breaks and node-A start reconnecting with node-B but it will not be success.
+Then on next `syncCluster` iteration (or next + 1 iteration) node-A can't find node-B record in `sama-node-data` records, but has active reconnecting state with node-B, then mean that node-B destroyed because no-one update node-B `sama-node-data` record ttl, then node-A cancel reconnecting and clean redis data creates by node-B (`sama-node-users`/`sama-user-devices`/`sama-user-data`), `sama-node-data` already deleted by redis ttl.
+
+#### Conditions when node mark other nodes like destroyed and should clean data ?
+
+It can be detected no `syncCluster` iteration
+
+- If have active node record `sama-node-users` w/o ttl, but do not have record `sama-node-data` - no-one update record so node is destroyed
+- If do not have node record `sama-node-data` but has reconnecting state
+- If have connection (check prop `clusterNodesConnections`) but do not have record `sama-node-data` (case when some how ws close was not called)
+
+#### How its data is cleared from Redis ?
+
+Check method `cleanDestroyedNodeData` that clean destroyed node data by node-endpoint.
+Flow:
+- retrieve `sama-node-users` - list of user who was connected to destroyed node
+- delete 'smembers' from `sama-user-devices`
+- delete `sama-user-data` record
+- delete `sama-node-users`