Merge pull request #1066 from Hexastack/1064-bug---api-duplicate-subscriber-issue

fix: resolve duplicated subscriber issue

api/package-lock.json

@@ -27,6 +27,7 @@
         "@resvg/resvg-js": "^2.6.2",
         "@socket.io/redis-adapter": "^8.3.0",
         "@tekuconcept/nestjs-csrf": "^1.1.0",
+        "async-mutex": "^0.5.0",
         "bcryptjs": "^2.4.3",
         "cache-manager": "^5.3.2",
         "cache-manager-redis-yet": "^4.1.2",
@@ -7411,10 +7412,9 @@
       "optional": true
     },
     "node_modules/async-mutex": {
-      "version": "0.4.1",
-      "resolved": "https://registry.npmjs.org/async-mutex/-/async-mutex-0.4.1.tgz",
-      "integrity": "sha512-WfoBo4E/TbCX1G95XTjbWTE3X2XLG0m1Xbv2cwOtuPdyH9CZvnaA5nCt1ucjaKEgW2A5IF71hxrRhr83Je5xjA==",
-      "dev": true,
+      "version": "0.5.0",
+      "resolved": "https://registry.npmjs.org/async-mutex/-/async-mutex-0.5.0.tgz",
+      "integrity": "sha512-1A94B18jkJ3DYq284ohPxoXbfTA5HsQ7/Mf4DEhcyLx3Bz27Rh59iScbB6EPiP+B+joue6YCxcMXSbFC1tZKwA==",
       "dependencies": {
         "tslib": "^2.4.0"
       }
@@ -15003,6 +15003,15 @@
         "node": ">=14.20.1"
       }
     },
+    "node_modules/mongodb-memory-server-core/node_modules/async-mutex": {
+      "version": "0.4.1",
+      "resolved": "https://registry.npmjs.org/async-mutex/-/async-mutex-0.4.1.tgz",
+      "integrity": "sha512-WfoBo4E/TbCX1G95XTjbWTE3X2XLG0m1Xbv2cwOtuPdyH9CZvnaA5nCt1ucjaKEgW2A5IF71hxrRhr83Je5xjA==",
+      "dev": true,
+      "dependencies": {
+        "tslib": "^2.4.0"
+      }
+    },
     "node_modules/mongodb-memory-server-core/node_modules/bson": {
       "version": "5.5.1",
       "resolved": "https://registry.npmjs.org/bson/-/bson-5.5.1.tgz",

api/package.json

@@ -62,6 +62,7 @@
     "@resvg/resvg-js": "^2.6.2",
     "@socket.io/redis-adapter": "^8.3.0",
     "@tekuconcept/nestjs-csrf": "^1.1.0",
+    "async-mutex": "^0.5.0",
     "bcryptjs": "^2.4.3",
     "cache-manager": "^5.3.2",
     "cache-manager-redis-yet": "^4.1.2",

api/src/channel/channel.service.ts

@@ -120,13 +120,13 @@ export class ChannelService {
    */
   @SocketGet(`/webhook/${WEB_CHANNEL_NAME}/`)
   @SocketPost(`/webhook/${WEB_CHANNEL_NAME}/`)
-  handleWebsocketForWebChannel(
+  async handleWebsocketForWebChannel(
     @SocketReq() req: SocketRequest,
     @SocketRes() res: SocketResponse,
   ) {
     this.logger.log('Channel notification (Web Socket) : ', req.method);
     const handler = this.getChannelHandler(WEB_CHANNEL_NAME);
-    return handler.handle(req, res);
+    return await handler.handle(req, res);
   }
 
   /**
@@ -195,6 +195,6 @@ export class ChannelService {
     }
 
     const handler = this.getChannelHandler(CONSOLE_CHANNEL_NAME);
-    return handler.handle(req, res);
+    return await handler.handle(req, res);
   }
 }

api/src/websocket/services/socket-event-dispatcher.service.ts

@@ -14,7 +14,9 @@ import {
 } from '@nestjs/common';
 import { ModulesContainer } from '@nestjs/core';
 import { InstanceWrapper } from '@nestjs/core/injector/instance-wrapper';
-import { EventEmitter2 } from '@nestjs/event-emitter';
+import { EventEmitter2, OnEvent } from '@nestjs/event-emitter';
+import { Mutex } from 'async-mutex';
+import { Socket } from 'socket.io';
 
 import { SocketEventMetadataStorage } from '../storage/socket-event-metadata.storage';
 import { SocketRequest } from '../utils/socket-request';
@@ -39,12 +41,21 @@ export class SocketEventDispatcherService implements OnModuleInit {
     private readonly modulesContainer: ModulesContainer,
   ) {}
 
+  @OnEvent('hook:websocket:connection')
+  handleConnection(client: Socket) {
+    client.data.mutex = new Mutex();
+  }
+
   async handleEvent(
     socketMethod: SocketMethod,
     path: string,
     req: SocketRequest,
     res: SocketResponse,
   ) {
+    // Prevent racing conditions from the same socket
+    const socketData = req.socket.data;
+    const release = await socketData.mutex.acquire();
+
     try {
       const handlers = this.routeHandlers[socketMethod];
       const foundHandler = Array.from(handlers.entries()).find(([key, _]) => {
@@ -62,6 +73,8 @@ export class SocketEventDispatcherService implements OnModuleInit {
       return await handler(req, res);
     } catch (error) {
       return this.handleException(error, res);
+    } finally {
+      release();
     }
   }
 

frontend/src/app-components/widget/ChatWidget.tsx

@@ -9,7 +9,7 @@
 import { Avatar, Box } from "@mui/material";
 import UiChatWidget from "hexabot-chat-widget/src/UiChatWidget";
 import { useRouter } from "next/router";
-import { useEffect, useState } from "react";
+import { useMemo } from "react";
 
 import { getAvatarSrc } from "@/components/inbox/helpers/mapMessages";
 import { useAuth } from "@/hooks/useAuth";
@@ -17,7 +17,6 @@ import { useConfig } from "@/hooks/useConfig";
 import { useSetting } from "@/hooks/useSetting";
 import i18n from "@/i18n/config";
 import { EntityType, RouterType } from "@/services/types";
-import { generateId } from "@/utils/generateId";
 
 import { ChatWidgetHeader } from "./ChatWidgetHeader";
 
@@ -30,11 +29,10 @@ export const ChatWidget = () => {
   const isVisualEditor = pathname.startsWith(`/${RouterType.VISUAL_EDITOR}`);
   const allowedDomainsSetting = useSetting(SETTING_TYPE, "allowed_domains");
   const themeColorSetting = useSetting(SETTING_TYPE, "theme_color");
-  const [key, setKey] = useState(generateId());
-
-  useEffect(() => {
-    setKey(generateId());
-  }, [allowedDomainsSetting, themeColorSetting]);
+  const key = useMemo(
+    () => `${allowedDomainsSetting}_${themeColorSetting}`,
+    [allowedDomainsSetting, themeColorSetting],
+  );
 
   return isAuthenticated ? (
     <Box