fix online list request (#204)

* fix online list request

* add basic cmd ws test commands

* update check cmd connections

* add close

* add socketCloseWatchdog

* add remove session

* on close logs debug

* add repl service

* update docker-file

* add ping socket

* update package

* fix tcp send failed

* update watchdogPingSocket

* add logs

* update addUserDeviceConnection

* update submodule

Author: banshiAnton

.env.example

@@ -1,5 +1,6 @@
 NODE_ENV=development
 STANDALONE_NODE=false
+WATCHDOG_PING_SOCKET_INTERVAL=60000 # 1 minute
 
 APP_PORT=9001
 APP_TCP_PORT=8001
@@ -11,6 +12,16 @@ MONGODB_URL=mongodb://127.0.0.1/samadb
 
 REDIS_URL=redis://127.0.0.1:6379
 
+# REPL
+# http
+APP_REPL_HTTP_PORT=5010
+APP_REPL_HTTP_ACCESS_KEY=repl-wow-key
+# socket
+APP_REPL_SOCKET_HANDLER=/tmp/net-repl.socket
+# file
+APP_REPL_FILE_IN=/tmp/pipe-repl.in
+APP_REPL_FILE_OUT=/tmp/pipe-repl.out
+
 STORAGE_DRIVER=minio
 
 # If you set STORAGE_DRIVER=s3, then fill the below envs

APIs/JSON/index.js

@@ -35,4 +35,10 @@ export default class JsonAPI extends BaseAPI {
     }
     return this.stringifyMessage(message)
   }
+
+  pingPackage() {
+    const ping = { response: { ping: {} } }
+
+    return this.stringifyMessage(ping)
+  }
 }

APIs/JSON/routes/packet_processor.js

@@ -46,18 +46,18 @@ class PacketJsonProcessor extends BasePacketProcessor {
     } catch (error) {
       logger.error(error)
       let errorBackMessage = null
-      if (json.request) {
+      if (json?.request) {
         errorBackMessage = {
           response: {
-            id: json.request.id,
+            id: json.request?.id,
             error: error.cause || error.message,
           },
         }
       } else {
-        const topLevelElement = Object.keys(json)[0]
+        const topLevelElement = json ? Object.keys(json)[0] : void 0
         errorBackMessage = {
           [topLevelElement]: {
-            id: json[topLevelElement].id,
+            id: json?.[topLevelElement]?.id,
             error: error.cause || error.message,
           },
         }

APIs/XMPP

@@ -11,7 +11,7 @@ COPY . .
 FROM node:22-slim
 
 RUN apt-get update && apt-get install -y --no-install-recommends \
-    curl \
+    curl netcat-openbsd \
  && rm -rf /var/lib/apt/lists/*
 
 WORKDIR /app

Dockerfile

@@ -7,6 +7,7 @@ const CONFIG = {
     name: process.env.APP_NAME ?? "SAMA",
     hostName: process.env.HOSTNAME,
     isStandAloneNode: process.env.STANDALONE_NODE === CONSTANTS.ENV_TRUE,
+    watchdogPingSocketInterval: +process.env.WATCHDOG_PING_SOCKET_INTERVAL,
   },
   logger: {
     logLevel: process.env.LOG_LEVEL ?? "debug",
@@ -102,6 +103,19 @@ const CONFIG = {
       apiKey: process.env.HTTP_ADMIN_API_KEY,
     },
   },
+  repl: {
+    http: {
+      port: +(process.env.APP_REPL_HTTP_PORT ?? 5010),
+      accessKey: process.env.APP_REPL_HTTP_ACCESS_KEY,
+    },
+    socket: {
+      handler: process.env.APP_REPL_SOCKET_HANDLER,
+    },
+    file: {
+      in: process.env.APP_REPL_FILE_IN,
+      out: process.env.APP_REPL_FILE_OUT,
+    },
+  },
   conversation: {
     disableChannelsLogic: process.env.CONVERSATION_DISABLE_CHANNELS_LOGIC === CONSTANTS.ENV_TRUE,
     isEventsEnabled: process.env.CONVERSATION_NOTIFICATIONS_ENABLED === CONSTANTS.ENV_TRUE,

app/config/default.js

@@ -0,0 +1,144 @@
+import fs from "node:fs"
+import fsPromises from "node:fs/promises"
+import vm from "node:vm"
+import { exec } from "node:child_process"
+import repl from "node:repl"
+import net from "node:net"
+import http from "node:http"
+
+import { CONSTANTS } from "../constants/constants.js"
+
+// example: curl --no-progress-meter -sSNT. -H "api-key: ****" localhost:5010
+
+const httpReplService = async (replOptions, httpOptions) => {
+  const { ctx } = replOptions
+  const { accessKey, port } = httpOptions
+
+  const server = http.createServer((req, res) => {
+    if (req.headers[CONSTANTS.HTTP_REPL_ACCESS_KEY_HEADER] !== accessKey) {
+      return res.end("Invalid access-key")
+    }
+
+    res.setHeader("content-type", "multipart/octet-stream")
+
+    const replService = repl.start({
+      prompt: "curl repl> ",
+      input: req,
+      output: res,
+      terminal: false,
+      useColors: true,
+      useGlobal: false,
+    })
+
+    const context = vm.createContext({ ...ctx })
+
+    replService.context = context
+
+    req.on("error", (error) => replService.close())
+    req.on("end", () => replService.close())
+
+    res.on("error", (error) => replService.close())
+    res.on("end", () => replService.close())
+
+    replService.on("close", () => !res.closed && res.end("REPL closed"))
+    replService.on("error", (error) => {
+      replService.close()
+      !res.closed && res.end("REPL closed")
+      !req.closed && req.destroy()
+    })
+  })
+
+  return new Promise((resolve, reject) => {
+    server.listen(port, () => resolve(port))
+  })
+}
+
+// example write: cat > ./pipe-repl.in
+// example read: tail -f ./pipe-repl.out
+
+const fileReplService = async (replOptions, fileOptions) => {
+  const { ctx } = replOptions
+  const { fileIn, fileOut } = fileOptions
+
+  await fsPromises.rm(fileIn, { force: true }).catch((error) => {})
+  await fsPromises.rm(fileOut, { force: true }).catch((error) => {})
+
+  await new Promise((resolve, reject) => {
+    exec(`mkfifo ${fileIn}`, (error, out, outErr) => {
+      if (error) {
+        return reject(error)
+      }
+      resolve()
+    })
+  })
+
+  const cmdInputPipe = fs.createReadStream(fileIn, { encoding: "utf8" })
+  const cmdOutPipe = fs.createWriteStream(fileOut, { encoding: "utf8" })
+
+  const replService = repl.start({
+    prompt: "file repl> ",
+    input: cmdInputPipe,
+    output: cmdOutPipe,
+    terminal: false,
+    useColors: true,
+    useGlobal: false,
+  })
+
+  const context = vm.createContext({ ...ctx })
+
+  replService.context = context
+
+  replService.on("error", (error) => {
+    replService.close()
+    cmdInputPipe.close()
+  })
+
+  cmdInputPipe.on("end", () => {
+    replService.close()
+    fileReplService(replOptions, fileOptions)
+  })
+}
+
+// example: nc -U ./net-repl.socket
+
+const netReplService = async (replOptions, fileOptions) => {
+  const { ctx } = replOptions
+  const { socketHandler } = fileOptions
+
+  await fsPromises.rm(socketHandler, { force: true }).catch((error) => {})
+
+  const server = net.createServer((socket) => {
+    const replService = repl.start({
+      prompt: "socket repl> ",
+      input: socket,
+      output: socket,
+      terminal: false,
+      useColors: true,
+      useGlobal: false,
+    })
+
+    const context = vm.createContext({ ...ctx })
+
+    replService.context = context
+
+    socket.on("end", () => replService.close())
+    socket.on("error", (error) => replService.close())
+    replService.on("close", () => !socket.closed && socket.end("REPL closed"))
+  })
+
+  return new Promise((resolve, reject) => {
+    server.listen(socketHandler, () => resolve(socketHandler))
+  })
+}
+
+export const startReplServices = async (replOptions, httpOptions, netOptions, fileOptions) => {
+  if (httpOptions.accessKey) {
+    await httpReplService(replOptions, httpOptions)
+  }
+  if (netOptions.socketHandler) {
+    await netReplService(replOptions, netOptions)
+  }
+  if (fileOptions.fileIn && fileOptions.fileOut) {
+    await fileReplService(replOptions, fileOptions)
+  }
+}

app/lib/repl-tools.js

@@ -200,7 +200,7 @@ class BaseProtocolProcessor {
   }
 
   async onClose(socket, code) {
-    logger.trace("[Close] IP: %s CLIENT_ID: %s CODE: %s", this.socketAddress(socket), socket.clientId, code)
+    logger.debug("[Close] IP: %s CLIENT_ID: %s CODE: %s", this.socketAddress(socket), socket.clientId, code)
 
     socket.isAlive = false
 
@@ -214,7 +214,7 @@ class BaseProtocolProcessor {
   async updateLastUserLastActivityOnClose(socket) {
     const { organizationId, userId } = this.sessionService.getSession(socket) ?? {}
 
-    logger.trace("[UPDATE_LAST_ACTIVITY][CLOSE] OrgId: %s UserId: %s", organizationId, userId)
+    logger.debug("[UPDATE_LAST_ACTIVITY][CLOSE] OrgId: %s UserId: %s", organizationId, userId)
 
     if (!userId) {
       return

app/networking/protocol_processors/base.js

@@ -8,9 +8,10 @@ const name = "SessionService"
 class SessionServiceRegisterProvider extends RegisterProvider {
   register(slc) {
     const config = slc.use("Config")
+    const logger = slc.use("Logger").child("[SessionService]")
     const redisClient = slc.use("RedisClient")
 
-    return new SessionService(ACTIVE, config, redisClient)
+    return new SessionService(ACTIVE, config, logger, redisClient)
   }
 }
 

app/providers/services/session/Provider.js

@@ -10,9 +10,10 @@ import { CONSTANTS } from "../../../constants/constants.js"
 */
 
 class SessionService {
-  constructor(activeSessions, config, redisConnection) {
+  constructor(activeSessions, config, logger, redisConnection) {
     this.activeSessions = activeSessions
     this.config = config
+    this.logger = logger
     this.redisConnection = redisConnection
   }
 
@@ -21,21 +22,25 @@ class SessionService {
   }
 
   addUserDeviceConnection(socket, organizationId, userId, deviceId) {
-    const activeConnections = this.activeSessions.DEVICES[userId]
     const socketsToClose = []
 
+    let activeConnections = this.getUserDevices(userId)
+    const filterNotSameSocket = activeConnections.filter(connection => connection.socket !== socket)
+    this.activeSessions.DEVICES[userId] = filterNotSameSocket
+    activeConnections = this.getUserDevices(userId)
+
     const connection = { socket: socket, deviceId, organizationId }
 
     if (activeConnections) {
-      const devices = activeConnections.filter((connection) => {
+      const otherDeviceConnections = activeConnections.filter((connection) => {
         if (connection.deviceId !== deviceId) {
           return true
         } else {
           socketsToClose.push(connection.socket)
           return false
         }
       })
-      this.activeSessions.DEVICES[userId] = [...devices, connection]
+      this.activeSessions.DEVICES[userId] = [...otherDeviceConnections, connection]
     } else {
       this.activeSessions.DEVICES[userId] = [connection]
     }
@@ -102,9 +107,7 @@ class SessionService {
 
   async listUserDevice(organizationId, userId) {
     if (this.config.get("app.isStandAloneNode")) {
-      return this.getUserDevices(userId)
-        .map((connection) => connection?.deviceId)
-        .filter((deviceId) => deviceId !== CONSTANTS.HTTP_DEVICE_ID)
+      return this.listUserDeviceLocal(userId)
     }
 
     const userKey = this.#usersSetCacheKey(organizationId, userId)
@@ -113,6 +116,12 @@ class SessionService {
     return deviceIds ?? []
   }
 
+  listUserDeviceLocal(userId) {
+    return this.getUserDevices(userId)
+      .map((connection) => connection?.deviceId)
+      .filter((deviceId) => deviceId !== CONSTANTS.HTTP_DEVICE_ID)
+  }
+
   async deleteUserDevices(organizationId, userId) {
     const userKey = this.#usersSetCacheKey(organizationId, userId)
 
@@ -312,10 +321,21 @@ class SessionService {
   }
 
   async removeUserSession(socket, userId, deviceId) {
+    this.logger.debug("[removeUserSession][args]: %o", { socket: socket?.isAlive, userId, deviceId })
+
     userId = userId ?? this.getSessionUserId(socket)
     deviceId = deviceId ?? this.getDeviceId(socket, userId)
     const orgId = this.getSession(socket)?.organizationId
 
+    this.logger.debug("[removeUserSession][vars]: %o [session]: %o [device]: %s", { orgId, userId, deviceId }, this.getSession(socket), this.getDeviceId(socket, userId))
+
+    const devicesBefore = this.getUserDevices(userId).map((connection) => {
+      const { socket, ...connectionData } = connection
+      return { ...connectionData, socket: socket?.clientId }
+    })
+
+    this.logger.debug("[removeUserSession][devices][before]: %o %s", devicesBefore, devicesBefore?.length)
+
     const leftActiveConnections = this.getUserDevices(userId).filter(({ deviceId: activeDeviceId }) => activeDeviceId !== deviceId)
 
     if (leftActiveConnections?.length) {
@@ -325,6 +345,14 @@ class SessionService {
     }
     this.activeSessions.SESSIONS.delete(socket)
 
+
+    const devicesAfter = this.getUserDevices(userId).map((connection) => {
+      const { socket, ...connectionData } = connection
+      return { ...connectionData, socket: socket?.clientId }
+    })
+
+    this.logger.debug("[removeUserSession][devices][after]: %o %s", devicesAfter, devicesAfter?.length)
+
     if (!deviceId) {
       return
     }
@@ -393,7 +421,7 @@ class SessionService {
         (session) =>
           session?.organizationId === organizationId &&
           session?.extraParams[CONSTANTS.SESSION_DEVICE_ID_KEY] !== CONSTANTS.HTTP_DEVICE_ID &&
-          session?.userId
+          session?.userId && this.listUserDeviceLocal(session?.userId)?.length
       )
       .map((session) => session.userId)
       .sort((userIdA, userIdB) => userIdA - userIdB)