Refactor

Author: ruiqi7

backend/matching-service/package-lock.json

@@ -19,9 +19,7 @@
     "dotenv": "^16.4.5",
     "express": "^4.21.1",
     "socket.io": "^4.8.0",
-    "swagger-ui-express": "^5.0.1",
-    "uuid": "^10.0.0",
-    "yaml": "^2.5.1"
+    "uuid": "^10.0.0"
   },
   "devDependencies": {
     "@eslint/js": "^9.12.0",

backend/matching-service/package.json

@@ -1,28 +1,18 @@
 import express, { Request, Response } from "express";
 import dotenv from "dotenv";
-import swaggerUi from "swagger-ui-express";
-import yaml from "yaml";
-import fs from "fs";
 import cors from "cors";
 
-import matchingRoutes from "./routes/matchingRoutes.ts";
-
 dotenv.config();
 
 export const allowedOrigins = process.env.ORIGINS
   ? process.env.ORIGINS.split(",")
   : ["http://localhost:5173", "http://127.0.0.1:5173"];
 
-const file = fs.readFileSync("./swagger.yml", "utf-8");
-const swaggerDocument = yaml.parse(file);
-
 const app = express();
 
 app.use(cors({ origin: allowedOrigins, credentials: true }));
 app.options("*", cors({ origin: allowedOrigins, credentials: true }));
 
-app.use("/api/matching", matchingRoutes);
-app.use("/docs", swaggerUi.serve, swaggerUi.setup(swaggerDocument));
 app.get("/", (req: Request, res: Response) => {
   res.status(200).json({ message: "Hello world from matching service" });
 });

backend/matching-service/src/app.ts

@@ -1,72 +1,113 @@
 import amqplib, { Connection } from "amqplib";
 import dotenv from "dotenv";
-import { matchUsers } from "../utils/messageQueue";
-import { MatchRequestItem } from "../handlers/matchHandler";
+import { matchUsers } from "../handlers/matchHandler";
 import { Complexities, Categories, Languages } from "../utils/constants";
+import { MatchRequest, MatchRequestItem } from "../utils/types";
 
 dotenv.config();
 
 const RABBITMQ_ADDR = process.env.RABBITMQ_ADDR || "amqp://localhost:5672";
+const QUEUE_NAME_DELIMITER = "_";
 
 let mrConnection: Connection;
-const queues: string[] = [];
-const pendingQueueRequests = new Map<string, Map<string, MatchRequestItem>>();
+const waitingLists = new Map<string, Map<string, MatchRequestItem>>();
 
-const initQueueNames = () => {
-  for (const complexity of Object.values(Complexities)) {
-    for (const category of Object.values(Categories)) {
-      for (const language of Object.values(Languages)) {
-        queues.push(`${complexity}_${category}_${language}`);
-      }
+export const connectToRabbitMq = async () => {
+  try {
+    mrConnection = await amqplib.connect(RABBITMQ_ADDR);
+    const queues = setUpQueueNames();
+    for (const queue of queues) {
+      await setUpConsumer(queue);
+      getWaitingList(queue);
     }
+  } catch (error) {
+    console.error(error);
+    process.exit(1);
   }
 };
 
-const setUpQueue = async (queueName: string) => {
+export const sendToProducer = async (
+  matchRequest: MatchRequest,
+  requestId: string,
+  rejectedPartnerId?: string
+): Promise<boolean> => {
+  const { user, complexity, category, language, timeout } = matchRequest;
+
+  const requestItem: MatchRequestItem = {
+    id: requestId,
+    user: user,
+    sentTimestamp: Date.now(),
+    ttlInSecs: timeout,
+    rejectedPartnerId: rejectedPartnerId,
+  };
+
+  const sent = await routeToQueue(
+    [complexity, category, language],
+    requestItem
+  );
+  return sent;
+};
+
+const setUpConsumer = async (queueName: string) => {
   const consumerChannel = await mrConnection.createChannel();
-  await consumerChannel.assertQueue(queueName);
+  await consumerChannel.assertQueue(queueName, { durable: true });
 
   consumerChannel.consume(queueName, (msg) => {
     if (msg !== null) {
-      matchUsers(queueName, msg.content.toString());
+      const matchRequestItem = JSON.parse(msg.content.toString());
+      const waitingList = getWaitingList(queueName);
+      const [complexity, category] = deconstructQueueName(queueName);
+      matchUsers(matchRequestItem, waitingList, complexity, category);
       consumerChannel.ack(msg);
     }
   });
 };
 
-export const connectToRabbitMq = async () => {
-  try {
-    initQueueNames();
-    mrConnection = await amqplib.connect(RABBITMQ_ADDR);
-    for (const queue of queues) {
-      await setUpQueue(queue);
-      pendingQueueRequests.set(queue, new Map<string, MatchRequestItem>());
-    }
-  } catch (error) {
-    console.error(error);
-    process.exit(1);
-  }
-};
-
-export const sendToQueue = async (
-  complexity: string,
-  category: string,
-  language: string,
-  data: MatchRequestItem
+const routeToQueue = async (
+  criterias: string[],
+  requestItem: MatchRequestItem
 ): Promise<boolean> => {
   try {
-    const queueName = `${complexity}_${category}_${language}`;
+    const queueName = constructQueueName(criterias);
     const senderChannel = await mrConnection.createChannel();
-    senderChannel.sendToQueue(queueName, Buffer.from(JSON.stringify(data)));
+    senderChannel.sendToQueue(
+      queueName,
+      Buffer.from(JSON.stringify(requestItem)),
+      {
+        persistent: true,
+      }
+    );
     return true;
   } catch (error) {
     console.log(error);
     return false;
   }
 };
 
-export const getPendingRequests = (
-  queueName: string
-): Map<string, MatchRequestItem> => {
-  return pendingQueueRequests.get(queueName)!;
+const setUpQueueNames = () => {
+  const queues = [];
+  for (const complexity of Object.values(Complexities)) {
+    for (const category of Object.values(Categories)) {
+      for (const language of Object.values(Languages)) {
+        const queueName = constructQueueName([complexity, category, language]);
+        queues.push(queueName);
+      }
+    }
+  }
+  return queues;
+};
+
+const constructQueueName = (criterias: string[]) => {
+  return criterias.join(QUEUE_NAME_DELIMITER);
+};
+
+const deconstructQueueName = (queueName: string) => {
+  return queueName.split(QUEUE_NAME_DELIMITER);
+};
+
+const getWaitingList = (queueName: string): Map<string, MatchRequestItem> => {
+  if (!waitingLists.has(queueName)) {
+    waitingLists.set(queueName, new Map<string, MatchRequestItem>());
+  }
+  return waitingLists.get(queueName)!;
 };

backend/matching-service/src/config/rabbitmq.ts

@@ -1,80 +1,60 @@
 import { v4 as uuidv4 } from "uuid";
-import { sendToQueue } from "../config/rabbitmq";
-import { sendMatchFound } from "./websocketHandler";
+import {
+  isActiveRequest,
+  isUserConnected,
+  sendMatchFound,
+} from "./websocketHandler";
+import { MatchRequestItem, MatchUser } from "../utils/types";
 
 interface Match {
   matchUser1: MatchUser;
   matchUser2: MatchUser;
   accepted: boolean;
   complexity: string;
   category: string;
-  language: string;
-}
-
-export interface MatchUser {
-  id: string;
-  username: string;
-  profile?: string;
-}
-
-export interface MatchRequest {
-  user: MatchUser;
-  complexity: string;
-  category: string;
-  language: string;
-  timeout: number;
-}
-
-export interface MatchRequestItem {
-  id: string;
-  user: MatchUser;
-  sentTimestamp: number;
-  ttlInSecs: number;
-  rejectedPartnerId?: string;
 }
 
 const matches = new Map<string, Match>();
 
-export const sendMatchRequest = async (
-  matchRequest: MatchRequest,
-  requestId: string,
-  rejectedPartnerId?: string
-): Promise<boolean> => {
-  const { user, complexity, category, language, timeout } = matchRequest;
-
-  const matchItem: MatchRequestItem = {
-    id: requestId,
-    user: user,
-    sentTimestamp: Date.now(),
-    ttlInSecs: timeout,
-    rejectedPartnerId: rejectedPartnerId,
-  };
-
-  const sent = await sendToQueue(complexity, category, language, matchItem);
-  return sent;
-};
-
-export const createMatch = (
-  requestItem1: MatchRequestItem,
-  requestItem2: MatchRequestItem,
+export const matchUsers = (
+  newRequest: MatchRequestItem,
+  waitingList: Map<string, MatchRequestItem>,
   complexity: string,
-  category: string,
-  language: string
+  category: string
 ) => {
-  const matchId = uuidv4();
-  const matchUser1 = requestItem1.user;
-  const matchUser2 = requestItem2.user;
+  const newRequestUid = newRequest.user.id;
+
+  for (const [uid, waitListRequest] of waitingList) {
+    if (
+      isExpired(waitListRequest) ||
+      !isUserConnected(uid) ||
+      !isActiveRequest(uid, waitListRequest.id) ||
+      uid === newRequestUid
+    ) {
+      waitingList.delete(uid);
+      continue;
+    }
 
-  matches.set(matchId, {
-    matchUser1: matchUser1,
-    matchUser2: matchUser2,
-    accepted: false,
-    complexity,
-    category,
-    language,
-  });
+    if (
+      isExpired(newRequest) ||
+      !isUserConnected(newRequestUid) ||
+      !isActiveRequest(newRequestUid, newRequest.id)
+    ) {
+      return;
+    }
 
-  sendMatchFound(matchId, matchUser1, matchUser2);
+    if (
+      uid === newRequest.rejectedPartnerId ||
+      newRequestUid === waitListRequest.rejectedPartnerId
+    ) {
+      continue;
+    }
+
+    waitingList.delete(uid);
+    createMatch(waitListRequest.user, newRequest.user, complexity, category);
+    return;
+  }
+  waitingList.set(newRequestUid, newRequest);
 };
 
 export const handleMatchAccept = (matchId: string): boolean => {
@@ -116,3 +96,26 @@ export const getMatchByUid = (
 export const getMatchById = (matchId: string): Match | undefined => {
   return matches.get(matchId);
 };
+
+const createMatch = (
+  matchUser1: MatchUser,
+  matchUser2: MatchUser,
+  complexity: string,
+  category: string
+) => {
+  const matchId = uuidv4();
+
+  matches.set(matchId, {
+    matchUser1: matchUser1,
+    matchUser2: matchUser2,
+    accepted: false,
+    complexity,
+    category,
+  });
+
+  sendMatchFound(matchId, matchUser1, matchUser2);
+};
+
+const isExpired = (data: MatchRequestItem): boolean => {
+  return Date.now() - data.sentTimestamp >= data.ttlInSecs * 1000;
+};