Merge pull request #995 from Hexastack/993-bug-console-chat-widget-doesnt-show-quick-replies

fix: Show quick replies in Console Chat Widget

api/src/chat/services/message.service.spec.ts

@@ -31,6 +31,9 @@ import {
   rootMongooseTestModule,
 } from '@/utils/test/test';
 import { buildTestingMocks } from '@/utils/test/utils';
+import { IOOutgoingSubscribeMessage } from '@/websocket/pipes/io-message.pipe';
+import { Room } from '@/websocket/types';
+import { WebsocketGateway } from '@/websocket/websocket.gateway';
 
 import { MessageRepository } from '../repositories/message.repository';
 import { Message, MessageModel } from '../schemas/message.schema';
@@ -53,6 +56,13 @@ describe('MessageService', () => {
   let recipient: Subscriber;
   let messagesWithSenderAndRecipient: Message[];
   let user: User;
+  let mockGateway: Partial<WebsocketGateway>;
+  let mockMessageService: MessageService;
+  const SESSION_ID = 'session-123';
+  const SUCCESS_PAYLOAD: IOOutgoingSubscribeMessage = {
+    success: true,
+    subscribe: Room.MESSAGE,
+  };
 
   beforeAll(async () => {
     const { getMocks } = await buildTestingMocks({
@@ -102,11 +112,34 @@ describe('MessageService', () => {
       recipient: allSubscribers.find(({ id }) => id === message.recipient)?.id,
       sentBy: allUsers.find(({ id }) => id === message.sentBy)?.id,
     }));
+    mockGateway = {
+      joinNotificationSockets: jest.fn(),
+    };
+    mockMessageService = new MessageService({} as any, mockGateway as any);
   });
 
   afterEach(jest.clearAllMocks);
   afterAll(closeInMongodConnection);
 
+  describe('subscribe', () => {
+    it('should join Notification sockets message room and return a success response', async () => {
+      const req = { sessionID: SESSION_ID };
+      const res = {
+        json: jest.fn(),
+        status: jest.fn().mockReturnThis(),
+      };
+
+      await mockMessageService.subscribe(req as any, res as any);
+
+      expect(mockGateway.joinNotificationSockets).toHaveBeenCalledWith(
+        SESSION_ID,
+        Room.MESSAGE,
+      );
+      expect(res.status).toHaveBeenCalledWith(200);
+      expect(res.json).toHaveBeenCalledWith(SUCCESS_PAYLOAD);
+    });
+  });
+
   describe('findOneAndPopulate', () => {
     it('should find message by id, and populate its corresponding sender and recipient', async () => {
       jest.spyOn(messageRepository, 'findOneAndPopulate');

api/src/chat/services/message.service.ts

@@ -19,6 +19,7 @@ import {
 } from '@/websocket/decorators/socket-method.decorator';
 import { SocketReq } from '@/websocket/decorators/socket-req.decorator';
 import { SocketRes } from '@/websocket/decorators/socket-res.decorator';
+import { IOOutgoingSubscribeMessage } from '@/websocket/pipes/io-message.pipe';
 import { Room } from '@/websocket/types';
 import { SocketRequest } from '@/websocket/utils/socket-request';
 import { SocketResponse } from '@/websocket/utils/socket-response';
@@ -53,11 +54,16 @@ export class MessageService extends BaseService<
    */
   @SocketGet('/message/subscribe/')
   @SocketPost('/message/subscribe/')
-  subscribe(@SocketReq() req: SocketRequest, @SocketRes() res: SocketResponse) {
+  async subscribe(
+    @SocketReq() req: SocketRequest,
+    @SocketRes() res: SocketResponse,
+  ): Promise<IOOutgoingSubscribeMessage> {
     try {
-      this.gateway.io.socketsJoin(Room.MESSAGE);
+      await this.gateway.joinNotificationSockets(req.sessionID, Room.MESSAGE);
+
       return res.status(200).json({
         success: true,
+        subscribe: Room.MESSAGE,
       });
     } catch (e) {
       this.logger.error('Websocket subscription', e);

api/src/chat/services/subscriber.service.spec.ts

@@ -38,6 +38,9 @@ import {
   rootMongooseTestModule,
 } from '@/utils/test/test';
 import { buildTestingMocks } from '@/utils/test/utils';
+import { IOOutgoingSubscribeMessage } from '@/websocket/pipes/io-message.pipe';
+import { Room } from '@/websocket/types';
+import { WebsocketGateway } from '@/websocket/websocket.gateway';
 
 import { LabelRepository } from '../repositories/label.repository';
 import { SubscriberRepository } from '../repositories/subscriber.repository';
@@ -58,6 +61,13 @@ describe('SubscriberService', () => {
   let allSubscribers: Subscriber[];
   let allLabels: Label[];
   let allUsers: User[];
+  let mockGateway: Partial<WebsocketGateway>;
+  let mockSubscriberService: SubscriberService;
+  const SESSION_ID = 'session-123';
+  const SUCCESS_PAYLOAD: IOOutgoingSubscribeMessage = {
+    success: true,
+    subscribe: Room.SUBSCRIBER,
+  };
 
   beforeAll(async () => {
     const { getMocks } = await buildTestingMocks({
@@ -103,11 +113,38 @@ describe('SubscriberService', () => {
     allSubscribers = await subscriberRepository.findAll();
     allLabels = await labelRepository.findAll();
     allUsers = await userRepository.findAll();
+    mockGateway = {
+      joinNotificationSockets: jest.fn(),
+    };
+    mockSubscriberService = new SubscriberService(
+      {} as any,
+      {} as any,
+      mockGateway as any,
+    );
   });
 
   afterEach(jest.clearAllMocks);
   afterAll(closeInMongodConnection);
 
+  describe('subscribe', () => {
+    it('should join Notification sockets subscriber room and return a success response', async () => {
+      const req = { sessionID: SESSION_ID };
+      const res = {
+        json: jest.fn(),
+        status: jest.fn().mockReturnThis(),
+      };
+
+      await mockSubscriberService.subscribe(req as any, res as any);
+
+      expect(mockGateway.joinNotificationSockets).toHaveBeenCalledWith(
+        SESSION_ID,
+        Room.SUBSCRIBER,
+      );
+      expect(res.status).toHaveBeenCalledWith(200);
+      expect(res.json).toHaveBeenCalledWith(SUCCESS_PAYLOAD);
+    });
+  });
+
   describe('findOneAndPopulate', () => {
     it('should find subscribers, and foreach subscriber populate its corresponding labels', async () => {
       jest.spyOn(subscriberService, 'findOneAndPopulate');

api/src/chat/services/subscriber.service.ts

@@ -30,6 +30,7 @@ import {
 } from '@/websocket/decorators/socket-method.decorator';
 import { SocketReq } from '@/websocket/decorators/socket-req.decorator';
 import { SocketRes } from '@/websocket/decorators/socket-res.decorator';
+import { IOOutgoingSubscribeMessage } from '@/websocket/pipes/io-message.pipe';
 import { Room } from '@/websocket/types';
 import { SocketRequest } from '@/websocket/utils/socket-request';
 import { SocketResponse } from '@/websocket/utils/socket-response';
@@ -71,10 +72,17 @@ export class SubscriberService extends BaseService<
    */
   @SocketGet('/subscriber/subscribe/')
   @SocketPost('/subscriber/subscribe/')
-  subscribe(@SocketReq() req: SocketRequest, @SocketRes() res: SocketResponse) {
+  async subscribe(
+    @SocketReq() req: SocketRequest,
+    @SocketRes() res: SocketResponse,
+  ): Promise<IOOutgoingSubscribeMessage> {
     try {
-      this.gateway.io.socketsJoin(Room.SUBSCRIBER);
-      return res.json({
+      await this.gateway.joinNotificationSockets(
+        req.sessionID,
+        Room.SUBSCRIBER,
+      );
+
+      return res.status(200).json({
         success: true,
         subscribe: Room.SUBSCRIBER,
       });

api/src/websocket/pipes/io-message.pipe.ts

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2024 Hexastack. All rights reserved.
+ * Copyright © 2025 Hexastack. All rights reserved.
  *
  * Licensed under the GNU Affero General Public License v3.0 (AGPLv3) with the following additional terms:
  * 1. The name "Hexabot" is a trademark of Hexastack. You may not use this name in derivative works without express written permission.
@@ -15,6 +15,8 @@ import {
 
 import { config } from '@/config';
 
+import { Room } from '../types';
+
 export interface IOOutgoingMessage {
   statusCode: number;
   body: any;
@@ -29,6 +31,11 @@ export interface IOIncomingMessage {
   url: string;
 }
 
+export interface IOOutgoingSubscribeMessage {
+  success: boolean;
+  subscribe: Room;
+}
+
 @Injectable()
 export class IOMessagePipe implements PipeTransform<string, IOIncomingMessage> {
   transform(value: string, _metadata: ArgumentMetadata): IOIncomingMessage {

api/src/websocket/utils/gateway-options.ts

@@ -52,9 +52,11 @@ export const buildWebSocketGatewayOptions = (): Partial<ServerOptions> => {
       allowUpgrades: config.sockets.allowUpgrades,
     }),
     ...(config.sockets.cookie && { cookie: config.sockets.cookie }),
-    ...(config.sockets.onlyAllowOrigins && {
-      cors: {
-        origin: async (origin, cb) => {
+    cors: {
+      origin: async (origin, cb) => {
+        if (config.env === 'test') {
+          cb(null, true);
+        } else {
           // Retrieve the allowed origins from the settings
           const app = AppInstance.getApp();
           const settingService = app.get<SettingService>(SettingService);
@@ -74,9 +76,9 @@ export const buildWebSocketGatewayOptions = (): Partial<ServerOptions> => {
               }
             })
             .catch(cb);
-        },
+        }
       },
-    }),
+    },
   };
 
   return opts;

api/src/websocket/utils/socket-response.ts

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2024 Hexastack. All rights reserved.
+ * Copyright © 2025 Hexastack. All rights reserved.
  *
  * Licensed under the GNU Affero General Public License v3.0 (AGPLv3) with the following additional terms:
  * 1. The name "Hexabot" is a trademark of Hexastack. You may not use this name in derivative works without express written permission.
@@ -70,9 +70,9 @@ export class SocketResponse {
     return response;
   }
 
-  json(data: any) {
+  json<T = Partial<IOOutgoingMessage>>(data: T): T {
     this.set('Content-Type', 'application/json');
-    return this.send(data);
+    return this.send(data) as T;
   }
 
   public getPromise() {

api/src/websocket/websocket.gateway.spec.ts

@@ -7,8 +7,8 @@
  */
 
 import { INestApplication } from '@nestjs/common';
-import { EventEmitter2 } from '@nestjs/event-emitter';
 import { Socket, io } from 'socket.io-client';
+import { v4 as uuidv4 } from 'uuid';
 
 import {
   closeInMongodConnection,
@@ -17,21 +17,22 @@ import {
 import { buildTestingMocks } from '@/utils/test/utils';
 
 import { SocketEventDispatcherService } from './services/socket-event-dispatcher.service';
+import { Room } from './types';
 import { WebsocketGateway } from './websocket.gateway';
 
 describe('WebsocketGateway', () => {
   let gateway: WebsocketGateway;
   let app: INestApplication;
-  let ioClient: Socket;
+  let createSocket: (index: number) => Socket;
+  let sockets: Socket[];
+  let messageRoomSockets: Socket[];
+  let uuids: string[];
+  let validUuids: string[];
 
   beforeAll(async () => {
     // Instantiate the app
     const { module } = await buildTestingMocks({
-      providers: [
-        WebsocketGateway,
-        EventEmitter2,
-        SocketEventDispatcherService,
-      ],
+      providers: [WebsocketGateway, SocketEventDispatcherService],
       imports: [
         rootMongooseTestModule(({ uri, dbName }) => {
           process.env.MONGO_URI = uri;
@@ -44,14 +45,25 @@ describe('WebsocketGateway', () => {
     // Get the gateway instance from the app instance
     gateway = app.get<WebsocketGateway>(WebsocketGateway);
     // Create a new client that will interact with the gateway
-    ioClient = io('http://localhost:3000', {
-      autoConnect: false,
-      transports: ['websocket', 'polling'],
-      // path: '/socket.io/?EIO=4&transport=websocket&channel=web-channel',
-      query: { EIO: '4', transport: 'websocket', channel: 'web-channel' },
-    });
 
-    app.listen(3000);
+    uuids = [uuidv4(), uuidv4(), uuidv4()];
+    validUuids = [uuids[0], uuids[2]];
+
+    createSocket = (index: number) =>
+      io('http://localhost:3000', {
+        autoConnect: false,
+        transports: ['websocket', 'polling'],
+        // path: '/socket.io/?EIO=4&transport=websocket&channel=web-channel',
+        query: { EIO: '4', transport: 'websocket', channel: 'web-channel' },
+        extraHeaders: { uuid: uuids[index] },
+      });
+
+    sockets = uuids.map((_e, index) => createSocket(index));
+    messageRoomSockets = sockets.filter((socket) =>
+      validUuids.includes(socket?.io.opts.extraHeaders?.['uuid'] || ''),
+    );
+
+    await app.listen(3000);
   });
 
   afterAll(async () => {
@@ -64,22 +76,95 @@ describe('WebsocketGateway', () => {
   });
 
   it('should connect successfully', async () => {
-    ioClient.on('connect', () => {
-      expect(true).toBe(true);
+    const [socket1] = sockets;
+    socket1.connect();
+
+    await new Promise<void>((resolve) => {
+      socket1.on('connect', async () => {
+        expect(true).toBe(true);
+        resolve();
+      });
     });
-    ioClient.connect();
-    ioClient.disconnect();
+
+    socket1.disconnect();
   });
 
   it('should emit "OK" on "healthcheck"', async () => {
-    ioClient.on('connect', () => {
-      ioClient.emit('healthcheck', 'Hello world');
+    const [socket1] = sockets;
 
-      ioClient.on('event', (data) => {
-        expect(data).toBe('OK');
+    socket1.connect();
+
+    await new Promise<void>((resolve) => {
+      socket1.on('connect', () => {
+        socket1.emit('healthcheck', 'Hello world!');
+        socket1.on('event', (data) => {
+          expect(data).toBe('OK');
+          resolve();
+        });
       });
     });
-    ioClient.connect();
-    ioClient.disconnect();
+
+    socket1.disconnect();
+  });
+
+  describe('joinNotificationSockets', () => {
+    it('should make socket1 and socket3 join the room MESSAGE', async () => {
+      messageRoomSockets.forEach((socket) => socket.connect());
+
+      for (const socket of messageRoomSockets) {
+        await new Promise<void>((resolve) => socket.on('connect', resolve));
+      }
+
+      const serverSockets = await gateway.io.fetchSockets();
+
+      expect(serverSockets.length).toBe(2);
+
+      jest.spyOn(gateway, 'getNotificationSockets').mockResolvedValueOnce(
+        serverSockets.filter(({ handshake: { headers } }) => {
+          const uuid = headers.uuid?.toString() || '';
+
+          return validUuids.includes(uuid);
+        }),
+      );
+
+      await gateway.joinNotificationSockets('sessionId', Room.MESSAGE);
+
+      gateway.io.to(Room.MESSAGE).emit('message', { data: 'OK' });
+
+      for (const socket of messageRoomSockets) {
+        await new Promise<void>((resolve) => {
+          socket.on('message', async ({ data }) => {
+            expect(data).toBe('OK');
+            resolve();
+          });
+        });
+      }
+
+      messageRoomSockets.forEach((socket) => socket.disconnect());
+    });
+
+    it('should throw an error when socket array is empty', async () => {
+      jest.spyOn(gateway, 'getNotificationSockets').mockResolvedValueOnce([]);
+
+      await expect(
+        gateway.joinNotificationSockets('sessionId', Room.MESSAGE),
+      ).rejects.toThrow('No notification sockets found!');
+
+      expect(gateway.getNotificationSockets).toHaveBeenCalledWith('sessionId');
+    });
+
+    it('should throw an error with empty sessionId', async () => {
+      await expect(
+        gateway.joinNotificationSockets('', Room.MESSAGE),
+      ).rejects.toThrow('SessionId is required!');
+    });
+  });
+
+  describe('getNotificationSockets', () => {
+    it('should throw an error with empty sessionId', async () => {
+      await expect(gateway.getNotificationSockets('')).rejects.toThrow(
+        'SessionId is required!',
+      );
+    });
   });
 });

api/src/websocket/websocket.gateway.ts

@@ -21,7 +21,8 @@ import cookie from 'cookie';
 import * as cookieParser from 'cookie-parser';
 import signature from 'cookie-signature';
 import { Session as ExpressSession, SessionData } from 'express-session';
-import { Server, Socket } from 'socket.io';
+import { RemoteSocket, Server, Socket } from 'socket.io';
+import { DefaultEventsMap } from 'socket.io/dist/typed-events';
 import { sync as uid } from 'uid-safe';
 
 import { MessageFull } from '@/chat/schemas/message.schema';
@@ -113,53 +114,56 @@ export class WebsocketGateway
     this.io.to(subscriber.foreign_id).emit(type, content);
   }
 
-  createAndStoreSession(client: Socket, next: (err?: Error) => void): void {
-    const sid = uid(24); // Sign the sessionID before sending
-    const signedSid = 's:' + signature.sign(sid, config.session.secret);
-    // Send session ID to client to set cookie
-    const cookies = cookie.serialize(
-      config.session.name,
-      signedSid,
-      config.session.cookie,
-    );
-    const newSession: SessionData<SubscriberStub> = {
-      cookie: {
-        // Prevent access from client-side javascript
-        httpOnly: true,
+  async createAndStoreSession(client: Socket): Promise<void> {
+    return new Promise<void>((resolve, reject) => {
+      const sid = uid(24); // Sign the sessionID before sending
+      const signedSid = 's:' + signature.sign(sid, config.session.secret);
+      // Send session ID to client to set cookie
+      const cookies = cookie.serialize(
+        config.session.name,
+        signedSid,
+        config.session.cookie,
+      );
+      const newSession: SessionData<SubscriberStub> = {
+        cookie: {
+          // Prevent access from client-side javascript
+          httpOnly: true,
 
-        // Restrict to path
-        path: '/',
+          // Restrict to path
+          path: '/',
 
-        originalMaxAge: config.session.cookie.maxAge,
-      },
-      passport: { user: {} },
-    }; // Initialize your session object as needed
-    getSessionStore().set(sid, newSession, (err) => {
-      if (err) {
-        this.logger.error('Error saving session:', err);
-        return next(new Error('Unable to establish a new socket session'));
-      }
+          originalMaxAge: config.session.cookie.maxAge,
+        },
+        passport: { user: {} },
+      }; // Initialize your session object as needed
+      getSessionStore().set(sid, newSession, (err) => {
+        if (err) {
+          this.logger.error('Error saving session:', err);
+          return reject(new Error('Unable to establish a new socket session'));
+        }
 
-      client.emit('set-cookie', cookies);
-      // Optionally set the cookie on the client's handshake object if needed
-      client.handshake.headers.cookie = cookies;
-      client.data.session = newSession;
-      this.logger.verbose(`
-        Could not fetch session, since connecting socket has no cookie in its handshake.
-        Generated a one-time-use cookie:
-        ${client.handshake.headers.cookie}
-        and saved it on the socket handshake.
-
-        > This means the socket started off with an empty session, i.e. (req.session === {})
-        > That "anonymous" session will only last until the socket is disconnected. To work around this,
-        > make sure the socket sends a 'cookie' header or query param when it initially connects.
-        > (This usually arises due to using a non-browser client such as a native iOS/Android app,
-        > React Native, a Node.js script, or some other connected device. It can also arise when
-        > attempting to connect a cross-origin socket in the browser, particularly for Safari users.
-        > To work around this, either supply a cookie manually, or ignore this message and use an
-        > approach other than sessions-- e.g. an auth token.)
-      `);
-      return next();
+        client.emit('set-cookie', cookies);
+        // Optionally set the cookie on the client's handshake object if needed
+        client.handshake.headers.cookie = cookies;
+        client.data.session = newSession;
+        client.data.sessionID = sid;
+        this.logger.verbose(`
+          Could not fetch session, since connecting socket has no cookie in its handshake.
+          Generated a one-time-use cookie:
+          ${client.handshake.headers.cookie}
+          and saved it on the socket handshake.
+  
+          > This means the socket started off with an empty session, i.e. (req.session === {})
+          > That "anonymous" session will only last until the socket is disconnected. To work around this,
+          > make sure the socket sends a 'cookie' header or query param when it initially connects.
+          > (This usually arises due to using a non-browser client such as a native iOS/Android app,
+          > React Native, a Node.js script, or some other connected device. It can also arise when
+          > attempting to connect a cross-origin socket in the browser, particularly for Safari users.
+          > To work around this, either supply a cookie manually, or ignore this message and use an
+          > approach other than sessions-- e.g. an auth token.)
+        `);
+        return resolve();
+      });
     });
   }
 
@@ -205,11 +209,14 @@ export class WebsocketGateway
     this.logger.log('Initialized websocket gateway');
 
     // Handle session
-    this.io.use((client, next) => {
+    this.io.use(async (client, next) => {
       this.logger.verbose('Client connected, attempting to load session.');
       try {
         const { searchParams } = new URL(`ws://localhost${client.request.url}`);
 
+        if (config.env === 'test') {
+          await this.createAndStoreSession(client);
+        }
         if (client.request.headers.cookie) {
           const cookies = cookie.parse(client.request.headers.cookie);
           if (cookies && config.session.name in cookies) {
@@ -218,14 +225,19 @@ export class WebsocketGateway
               config.session.secret,
             );
             if (sessionID) {
-              return this.loadSession(sessionID, (err, session) => {
+              return this.loadSession(sessionID, async (err, session) => {
                 if (err || !session) {
                   this.logger.warn(
                     'Unable to load session, creating a new one ...',
                     err,
                   );
                   if (searchParams.get('channel') !== 'console-channel') {
-                    return this.createAndStoreSession(client, next);
+                    try {
+                      await this.createAndStoreSession(client);
+                      next();
+                    } catch (e) {
+                      next(e);
+                    }
                   } else {
                     return next(new Error('Unauthorized: Unknown session ID'));
                   }
@@ -235,17 +247,22 @@ export class WebsocketGateway
                 next();
               });
             } else {
-              return next(new Error('Unable to parse session ID from cookie'));
+              next(new Error('Unable to parse session ID from cookie'));
             }
           }
         } else if (searchParams.get('channel') === 'web-channel') {
-          return this.createAndStoreSession(client, next);
+          try {
+            await this.createAndStoreSession(client);
+            next();
+          } catch (e) {
+            next(e);
+          }
         } else {
-          return next(new Error('Unauthorized to connect to WS'));
+          next(new Error('Unauthorized to connect to WS'));
         }
       } catch (e) {
         this.logger.warn('Something unexpected happening');
-        return next(e);
+        next(e);
       }
     });
   }
@@ -268,7 +285,7 @@ export class WebsocketGateway
   }
 
   async handleDisconnect(client: Socket): Promise<void> {
-    this.logger.log(`Client id:${client.id} disconnected`);
+    this.logger.log(`Client id: ${client.id} disconnected`);
     // Configurable custom afterDisconnect logic here
     // (default: do nothing)
     if (!config.sockets.afterDisconnect) {
@@ -405,4 +422,46 @@ export class WebsocketGateway
     );
     return response.getPromise();
   }
+
+  /**
+   * Retrieves notification sockets based on session id.
+   *
+   * @param sessionId - The session id
+   * @returns An RemoteSocket array
+   */
+  async getNotificationSockets(
+    sessionId: string,
+  ): Promise<RemoteSocket<DefaultEventsMap, any>[]> {
+    if (!sessionId) {
+      throw new Error('SessionId is required!');
+    }
+    const allSockets = await this.io.fetchSockets();
+
+    return allSockets.filter(
+      ({ handshake, data }) =>
+        !handshake.query.channel && data.sessionID === sessionId,
+    );
+  }
+
+  /**
+   * Join notification sockets based on session id.
+   *
+   * @param sessionId - The session id
+   * @param room - the joined room name
+   */
+  async joinNotificationSockets(sessionId: string, room: Room): Promise<void> {
+    if (!sessionId) {
+      throw new Error('SessionId is required!');
+    }
+
+    const notificationSockets = await this.getNotificationSockets(sessionId);
+
+    if (!notificationSockets.length) {
+      throw new Error('No notification sockets found!');
+    }
+
+    notificationSockets.forEach((notificationSocket) =>
+      notificationSocket.join(room),
+    );
+  }
 }