PEER-232: Add queuing event schema

Signed-off-by: SeeuSim <[email protected]>

Author: SeeuSim

backend/matching/src/types/index.ts

@@ -1,4 +1,5 @@
 import { client } from '@/lib/db';
+import { MATCHING_EVENT } from '@/ws/events';
 
 const DIFFICULTIES = ['Easy', 'Medium', 'Hard'] as const;
 
@@ -36,3 +37,10 @@ export type IStreamMessage = {
   };
   value?: Awaited<ReturnType<(typeof client)['ft']['search']>>['documents'][number]['value'];
 };
+
+export type IMatchEvent = (typeof MATCHING_EVENT)[keyof typeof MATCHING_EVENT];
+export type IChildProcessMessage = {
+  rooms: Array<string>;
+  event: IMatchEvent;
+  message?: unknown;
+};

backend/matching/src/workers/cleaner.ts

@@ -1,7 +1,7 @@
 import { client, logQueueStatus } from '@/lib/db';
 import { STREAM_CLEANER, STREAM_GROUP, STREAM_NAME } from '@/lib/db/constants';
 import { decodePoolTicket, getPoolKey } from '@/lib/utils';
-import { MATCH_SVC_EVENT } from '@/ws/main';
+import { MATCHING_EVENT } from '@/ws/events';
 
 import { connectClient, sendNotif } from './common';
 
@@ -65,8 +65,8 @@ async function clean() {
 
     if (socketRoom) {
       // Notify client
-      sendNotif([socketRoom], MATCH_SVC_EVENT.FAILED);
-      sendNotif([socketRoom], MATCH_SVC_EVENT.DISCONNECT);
+      sendNotif([socketRoom], MATCHING_EVENT.FAILED);
+      sendNotif([socketRoom], MATCHING_EVENT.DISCONNECT);
     }
 
     await logQueueStatus(logger, redisClient, `Queue Status after Expiring Request: <PLACEHOLDER>`);

backend/matching/src/workers/common.ts

@@ -2,7 +2,7 @@
 
 import { client } from '@/lib/db';
 import { logger } from '@/lib/utils';
-import type { IChildProcessMessage, IMatchEvent } from '@/ws/main';
+import type { IChildProcessMessage, IMatchEvent } from '@/types';
 
 export const sendNotif = (roomIds: Array<string>, event: IMatchEvent, message?: unknown) => {
   if (process.send) {

backend/matching/src/workers/index.ts

@@ -5,7 +5,8 @@ import path from 'path';
 import type { Server } from 'socket.io';
 
 import { logger } from '@/lib/utils';
-import { type IChildProcessMessage, MATCH_SVC_EVENT } from '@/ws/main';
+import type { IChildProcessMessage } from '@/types';
+import { MATCHING_EVENT } from '@/ws/events';
 
 let nWorkers = 0; // For tracking graceful exit of main process
 
@@ -27,7 +28,7 @@ export const initWorker = (name: string, io: Server) => {
     );
     const { rooms, event, message: payload } = messagePayload as IChildProcessMessage;
 
-    if (event === MATCH_SVC_EVENT.DISCONNECT) {
+    if (event === MATCHING_EVENT.DISCONNECT) {
       io.sockets.in(rooms).socketsLeave(rooms);
       return;
     }

backend/matching/src/workers/matcher.ts

@@ -2,7 +2,7 @@ import { client, logQueueStatus } from '@/lib/db';
 import { POOL_INDEX, STREAM_GROUP, STREAM_NAME, STREAM_WORKER } from '@/lib/db/constants';
 import { decodePoolTicket, getPoolKey, getStreamId } from '@/lib/utils';
 import { getMatchItems } from '@/services';
-import { MATCH_SVC_EVENT } from '@/ws/main';
+import { MATCHING_EVENT } from '@/ws/events';
 
 import { connectClient, sendNotif } from './common';
 
@@ -56,7 +56,7 @@ async function processMatch(
       }
 
       // To block cancellation
-      sendNotif([matchedSocketPort], MATCH_SVC_EVENT.MATCHING);
+      sendNotif([matchedSocketPort], MATCHING_EVENT.MATCHING);
 
       const matchedStreamId = getStreamId(timestamp);
 
@@ -73,8 +73,8 @@ async function processMatch(
       const { ...matchItems } = getMatchItems({ userId1: requestorUserId, userId2: matchedUserId });
 
       const sendMatchLogic = (socketPort: string) => {
-        sendNotif([socketPort], MATCH_SVC_EVENT.SUCCESS, matchItems);
-        sendNotif([socketPort], MATCH_SVC_EVENT.DISCONNECT);
+        sendNotif([socketPort], MATCHING_EVENT.SUCCESS, matchItems);
+        sendNotif([socketPort], MATCHING_EVENT.DISCONNECT);
       };
 
       // TODO: If client disconnected, stop sending to requestor
@@ -127,7 +127,7 @@ async function match() {
       } = decodePoolTicket(matchRequest);
 
       // To Block Cancellation
-      sendNotif([requestorSocketPort], MATCH_SVC_EVENT.MATCHING);
+      sendNotif([requestorSocketPort], MATCHING_EVENT.MATCHING);
 
       const clause = [`-@userId:(${requestorUserId})`];
 
@@ -190,7 +190,7 @@ async function match() {
 
       if (!hasDifficultyMatch) {
         // To allow cancellation
-        sendNotif([requestorSocketPort], MATCH_SVC_EVENT.PENDING);
+        sendNotif([requestorSocketPort], MATCHING_EVENT.PENDING);
       }
     }
   }

backend/matching/src/ws/events.ts

@@ -0,0 +1,17 @@
+export const WS_EVENT = {
+  JOIN_ROOM: 'joinRoom',
+  CANCEL_ROOM: 'cancelRoom',
+  LEAVE_ROOM: 'leave',
+  START_QUEUING: 'startQueuing',
+  DISCONNECT: 'disconnect',
+} as const;
+
+export const MATCHING_EVENT = {
+  ERROR: 'ERROR', // When match encounters error
+  QUEUED: 'QUEUED', // When match joins pool
+  MATCHING: 'MATCHING', // When matching in progress
+  PENDING: 'PENDING', // When waiting for match
+  SUCCESS: 'SUCCESS', // When match successful
+  FAILED: 'FAILED', // When match failed
+  DISCONNECT: 'DISCONNECT', // To disconnect all sockets in room
+} as const;

backend/matching/src/ws/handlers.ts

@@ -5,20 +5,11 @@ import { logger } from '@/lib/utils';
 import { queueingService } from '@/services';
 import type { IRedisClient, IRequestMatchEvent } from '@/types';
 
-import { MATCH_SVC_EVENT } from './main';
+import { MATCHING_EVENT,WS_EVENT } from './events';
 
 type ISocketIOServer<T> = Server<DefaultEventsMap, DefaultEventsMap, DefaultEventsMap, T>;
 type ISocketIOSocket<T> = Socket<DefaultEventsMap, DefaultEventsMap, DefaultEventsMap, T>;
 
-export const EVENTS = {
-  ERROR: 'ERROR',
-  JOIN_ROOM: 'joinRoom',
-  CANCEL_ROOM: 'cancelRoom',
-  LEAVE_ROOM: 'leave',
-  START_QUEUE: 'startQueue',
-  DISCONNECT: 'disconnect',
-} as const;
-
 export const joinRoomHandler =
   <T>(socket: ISocketIOSocket<T>) =>
   (roomId?: string) => {
@@ -53,7 +44,7 @@ export const queueEventHandler =
     if (!payload.roomId) {
       const errorMessage = 'Queuing Event triggered without room.';
       logger.warn(errorMessage);
-      socket.emit(EVENTS.ERROR, errorMessage);
+      socket.emit(MATCHING_EVENT.ERROR, errorMessage);
       return;
     }
 
@@ -65,7 +56,7 @@ export const queueEventHandler =
       (!payload.topic && !payload.difficulty) ||
       (payload.topic && !Array.isArray(payload.topic))
     ) {
-      socket.emit(EVENTS.ERROR, `Payload for ${EVENTS.START_QUEUE} is invalid.`);
+      socket.emit(MATCHING_EVENT.ERROR, `Payload for ${WS_EVENT.START_QUEUING} is invalid.`);
       return;
     }
 
@@ -85,11 +76,11 @@ export const queueEventHandler =
       } catch (error) {
         const { name, message, stack, cause } = error as Error;
         logger.error({ name, message, stack, cause }, `An error occurred connecting to the client`);
-        socket.emit(EVENTS.ERROR, 'Error connecting to client');
+        socket.emit(MATCHING_EVENT.ERROR, 'Error connecting to client');
         return;
       }
     }
 
-    socket.emit(MATCH_SVC_EVENT.QUEUED);
+    socket.emit(MATCHING_EVENT.QUEUED);
     logQueueStatus(logger, redisClient, `Queue Status Before Matching: <PLACEHOLDER>`);
   };

backend/matching/src/ws/main.ts

@@ -5,7 +5,8 @@ import { Server } from 'socket.io';
 import { UI_HOST } from '@/config';
 import { logger } from '@/lib/utils';
 
-import { cancelRoomHandler, EVENTS, joinRoomHandler, queueEventHandler } from './handlers';
+import { WS_EVENT } from './events';
+import { cancelRoomHandler, joinRoomHandler, queueEventHandler } from './handlers';
 
 export const createWs = (server: ReturnType<typeof createServer>) => {
   const io = new Server(server, {
@@ -18,30 +19,18 @@ export const createWs = (server: ReturnType<typeof createServer>) => {
   io.on('connection', (socket) => {
     logger.info(`Socket ${socket.id} connected`);
 
-    socket.on(EVENTS.JOIN_ROOM, joinRoomHandler(socket));
-    socket.on(EVENTS.CANCEL_ROOM, cancelRoomHandler(io, socket));
-    socket.on(EVENTS.LEAVE_ROOM, (room) => {
-      socket.leave(room);
+    socket.on(WS_EVENT.JOIN_ROOM, joinRoomHandler(socket));
+    socket.on(WS_EVENT.CANCEL_ROOM, cancelRoomHandler(io, socket));
+    socket.on(WS_EVENT.LEAVE_ROOM, (room?: string) => {
+      if (room) {
+        socket.leave(room);
+      }
     });
-    socket.on(EVENTS.DISCONNECT, () => {
+    socket.on(WS_EVENT.START_QUEUING, queueEventHandler(socket));
+    socket.on(WS_EVENT.DISCONNECT, () => {
       logger.info(`Client disconnected: ${socket.id}`);
+      socket.disconnect();
     });
-    socket.on(EVENTS.START_QUEUE, queueEventHandler(socket));
   });
   return io;
 };
-
-export const MATCH_SVC_EVENT = {
-  QUEUED: 'QUEUED', // When match joins pool
-  SUCCESS: 'SUCCESS', // When match successful
-  FAILED: 'FAILED', // When match failed
-  PENDING: 'PENDING', // When waiting for match
-  MATCHING: 'MATCHING', // When matching in progress
-  DISCONNECT: 'DISCONNECT', // To disconnect all sockets in room
-} as const;
-export type IMatchEvent = (typeof MATCH_SVC_EVENT)[keyof typeof MATCH_SVC_EVENT];
-export type IChildProcessMessage = {
-  rooms: Array<string>;
-  event: IMatchEvent;
-  message?: unknown;
-};

frontend/src/lib/ws/events.ts

@@ -0,0 +1 @@
+export * from './events';