Integrate matching websocket in BE and FE

Author: ruiqi7

backend/matching-service/config/rabbitmq.ts

@@ -1,6 +1,7 @@
 import amqplib, { Connection } from "amqplib";
 import dotenv from "dotenv";
-import { matchUsers } from "../utils/mq_utils";
+import { matchUsers } from "../src/utils/mq_utils";
+import { MatchItem } from "../src/types/matchTypes";
 
 dotenv.config();
 
@@ -15,11 +16,7 @@ export const connectRabbitMq = async () => {
 
     consumerChannel.consume(queue, async (msg) => {
       if (msg !== null) {
-        try {
-          await matchUsers(msg.content.toString());
-        } catch (error) {
-          console.error(error);
-        }
+        matchUsers(msg.content.toString());
         consumerChannel.ack(msg);
       }
     });
@@ -29,20 +26,13 @@ export const connectRabbitMq = async () => {
   }
 };
 
-type MatchRequestMessage = {
-  userId: string;
-  categories: string[] | string;
-  complexities: string[] | string;
-  sentTimestamp: number;
-  ttlInSecs: number;
-};
-
-export const sendRabbitMq = async (data: MatchRequestMessage) => {
+export const sendRabbitMq = async (data: MatchItem): Promise<boolean> => {
   try {
     const senderChannel = await mrConnection.createChannel();
     senderChannel.sendToQueue(queue, Buffer.from(JSON.stringify(data)));
+    return true;
   } catch (error) {
     console.log(error);
-    throw new Error("Failed to send match request");
+    return false;
   }
 };

backend/matching-service/server.ts

@@ -10,7 +10,7 @@ export const io = new Server(server, {
     origin: allowedOrigins,
     methods: ["GET", "POST"],
   },
-  connectionStateRecovery: {},
+  connectionStateRecovery: {}, // TODO: rejoin room?
 });
 
 io.on("connection", (socket) => {
@@ -29,24 +29,6 @@ if (process.env.NODE_ENV !== "test") {
           `Matching service server listening on http://localhost:${PORT}`
         );
       });
-
-      //can use this to test if rabbitmq works for you (import sendRabbitMq from rabbitmq.ts first)
-      /*const message1 = {
-        userId: "1",
-        categories: "Algorithms",
-        complexities: "Easy",
-        sentTimestamp: Date.now(),
-        ttlInSecs: 30,
-      };
-      sendRabbitMq(message1);
-      const message2 = {
-        userId: "2",
-        categories: "Algorithms",
-        complexities: "Medium",
-        sentTimestamp: Date.now(),
-        ttlInSecs: 30,
-      };
-      sendRabbitMq(message2);*/
     })
     .catch((err) => {
       console.error("Failed to connect to RabbitMq");

backend/matching-service/src/handlers/matchHandler.ts

@@ -2,48 +2,50 @@ import { io } from "../../server";
 import { Match, MatchItem, MatchRequest } from "../types/matchTypes";
 import { v4 as uuidv4 } from "uuid";
 import {
-  MATCH_ACCEPTANCE_TIMEOUT,
   MATCH_FOUND,
   MATCH_IN_PROGRESS,
   MATCH_SUCCESSFUL,
-  MATCH_TIMEOUT,
   MATCH_UNSUCCESSFUL,
+  MATCH_REQUEST_ERROR,
 } from "../utils/constants";
-import { appendToMatchQueue } from "./queueHandler";
 import { Socket } from "socket.io";
+import { sendRabbitMq } from "../../config/rabbitmq";
 
 const matches: Match = {};
+export const userSockets: Map<string, Socket> = new Map<string, Socket>();
 
-export const createMatchItem = (socket: Socket, matchRequest: MatchRequest) => {
+export const createMatchItem = async (
+  socket: Socket,
+  matchRequest: MatchRequest
+): Promise<boolean> => {
   const { user, complexities, categories, languages, timeout } = matchRequest;
 
-  const matchTimeout = setTimeout(() => {
-    socket.emit(MATCH_TIMEOUT);
-  }, timeout * 1000);
+  if (userSockets.has(user.id)) {
+    console.log(`user request exists: ${user.username}`);
+    socket.emit(MATCH_IN_PROGRESS);
+    return false;
+  }
+
+  userSockets.set(user.id, socket);
 
   const matchQueueItem: MatchItem = {
-    socket: socket,
     user: user,
     complexities: complexities,
     categories: categories,
     languages: languages,
-    timeout: matchTimeout,
+    sentTimestamp: Date.now(),
+    ttlInSecs: timeout,
     acceptedMatch: false,
   };
 
-  const result = appendToMatchQueue(matchQueueItem);
+  const result = await sendRabbitMq(matchQueueItem);
   if (!result) {
-    socket.emit(MATCH_IN_PROGRESS);
+    socket.emit(MATCH_REQUEST_ERROR);
   }
+  return result;
 };
 
-export const createMatch = (matchItems: MatchItem[]) => {
-  const matchItem1 = matchItems[0];
-  const matchItem2 = matchItems[1];
-
-  clearTimeout(matchItem1.timeout);
-  clearTimeout(matchItem2.timeout);
-
+export const createMatch = (matchItem1: MatchItem, matchItem2: MatchItem) => {
   const matchId = uuidv4();
   matches[matchId] = {
     item1: matchItem1,
@@ -52,57 +54,56 @@ export const createMatch = (matchItems: MatchItem[]) => {
     accepted: false,
   };
 
-  matchItem1.socket.join(matchId);
-  matchItem2.socket.join(matchId);
+  // check for disconnection? or just send the match (disconnected user will timeout anyway)
+  userSockets.get(matchItem1.user.id)!.join(matchId);
+  userSockets.get(matchItem2.user.id)!.join(matchId);
   io.to(matchId).emit(MATCH_FOUND, {
     matchId: matchId,
     user1: matchItem1.user,
     user2: matchItem2.user,
   });
 };
 
-export const setMatchTimeout = (matchId: string) => {
-  const match = matches[matchId];
-  if (!match) {
-    return;
-  }
-
-  const timeout = setTimeout(() => {
-    io.to(matchId).emit(MATCH_UNSUCCESSFUL);
-    delete matches[matchId];
-  }, MATCH_ACCEPTANCE_TIMEOUT);
-
-  match.timeout = timeout;
-};
-
 export const handleMatchAcceptance = (matchId: string) => {
   const match = matches[matchId];
   if (!match) {
     return;
   }
 
   if (match.accepted) {
-    clearTimeout(match.timeout!);
     io.to(matchId).emit(MATCH_SUCCESSFUL);
-    delete matches[matchId];
   } else {
     match.accepted = true;
   }
 };
 
-export const handleMatchDecline = (matchId: string) => {
+export const handleRematch = (
+  socket: Socket,
+  matchId: string,
+  rematchRequest: MatchRequest
+) => {
   const match = matches[matchId];
-  if (!match) {
-    return;
+  if (match) {
+    delete matches[matchId];
+    socket.to(matchId).emit(MATCH_UNSUCCESSFUL);
   }
 
-  clearTimeout(match.timeout!);
-  io.to(matchId).emit(MATCH_UNSUCCESSFUL);
-  delete matches[matchId];
+  createMatchItem(socket, rematchRequest);
 };
 
-export const isUserMatched = (userId: string): boolean => {
-  return !!Object.values(matches).find(
-    (match) => match.item1.user.id === userId || match.item2.user.id === userId
-  );
+export const handleMatchTermination = (terminatedSocket: Socket) => {
+  for (const [uid, socket] of userSockets) {
+    if (socket.id === terminatedSocket.id) {
+      userSockets.delete(uid);
+      break;
+    }
+  }
+
+  // TODO: no access to rooms
+  const matchId = Array.from(terminatedSocket.rooms)[1];
+  const match = matches[matchId];
+  if (match) {
+    delete matches[matchId];
+    terminatedSocket.to(matchId).emit(MATCH_UNSUCCESSFUL);
+  }
 };

backend/matching-service/src/handlers/queueHandler.ts

@@ -1,32 +1,41 @@
 import { Socket } from "socket.io";
 import {
   MATCH_ACCEPTED,
-  MATCH_DECLINED,
-  MATCH_RECEIVED,
+  REMATCH_REQUEST,
   MATCH_REQUEST,
+  SOCKET_CLIENT_DISCONNECT,
+  SOCKET_DISCONNECT,
 } from "../utils/constants";
 import { MatchRequest } from "../types/matchTypes";
 import {
   createMatchItem,
   handleMatchAcceptance,
-  handleMatchDecline,
-  setMatchTimeout,
+  handleMatchTermination,
+  handleRematch,
 } from "./matchHandler";
 
 export const handleWebsocketMatchEvents = (socket: Socket) => {
-  socket.on(MATCH_REQUEST, (matchRequest: MatchRequest) => {
-    createMatchItem(socket, matchRequest);
-  });
+  socket.on(
+    MATCH_REQUEST,
+    async (matchRequest: MatchRequest, callback: (result: boolean) => void) => {
+      const result = await createMatchItem(socket, matchRequest);
+      callback(result);
+    }
+  );
 
-  socket.on(MATCH_RECEIVED, (matchId: string) => {
-    setMatchTimeout(matchId);
-  });
+  socket.on(MATCH_ACCEPTED, (matchId: string) =>
+    handleMatchAcceptance(matchId)
+  );
 
-  socket.on(MATCH_ACCEPTED, (matchId: string) => {
-    handleMatchAcceptance(matchId);
-  });
+  socket.on(REMATCH_REQUEST, (matchId: string, rematchRequest: MatchRequest) =>
+    handleRematch(socket, matchId, rematchRequest)
+  );
 
-  socket.on(MATCH_DECLINED, (matchId: string) => {
-    handleMatchDecline(matchId);
+  // TODO: handle client reconnect failure
+  socket.on(SOCKET_DISCONNECT, (reason) => {
+    if (reason === SOCKET_CLIENT_DISCONNECT) {
+      console.log("Client manually disconnected");
+      handleMatchTermination(socket);
+    }
   });
 };

backend/matching-service/src/handlers/websocketHandler.ts

@@ -1,9 +1,7 @@
-import { Socket } from "socket.io";
-
 export interface MatchUser {
   id: string;
   username: string;
-  profile: string;
+  profile?: string;
 }
 
 export interface MatchRequest {
@@ -15,12 +13,12 @@ export interface MatchRequest {
 }
 
 export interface MatchItem {
-  socket: Socket;
   user: MatchUser;
   complexities: string[];
   categories: string[];
   languages: string[];
-  timeout: NodeJS.Timeout;
+  sentTimestamp: number;
+  ttlInSecs: number;
   acceptedMatch: boolean;
 }
 

backend/matching-service/src/types/matchTypes.ts

@@ -1,12 +1,12 @@
 /* Websocket Match Events */
 export const MATCH_REQUEST = "match_request";
-export const MATCH_TIMEOUT = "match_timeout";
+export const MATCH_REQUEST_ERROR = "match_request_error";
 export const MATCH_FOUND = "match_found";
 export const MATCH_IN_PROGRESS = "match_in_progress";
-export const MATCH_RECEIVED = "match_received";
 export const MATCH_ACCEPTED = "match_accepted";
-export const MATCH_DECLINED = "match_declined";
+export const REMATCH_REQUEST = "rematch_request";
 export const MATCH_SUCCESSFUL = "match_successful";
 export const MATCH_UNSUCCESSFUL = "match_unsuccessful";
 
-export const MATCH_ACCEPTANCE_TIMEOUT = 10000;
+export const SOCKET_DISCONNECT = "disconnect";
+export const SOCKET_CLIENT_DISCONNECT = "client namespace disconnect";