Merge pull request #81 from CS3219-AY2425S1/improvement/amqp_over_ws

Improvement/amqp over ws

Author: tyouwei

.gitignore

@@ -17,6 +17,9 @@
 # production
 /build
 
+# pnpm stores
+*/.pnpm-store
+
 # misc
 .DS_Store
 *.pem
@@ -28,6 +31,7 @@ yarn-error.log*
 
 # local env files
 .env*.local
+*/.env
 
 # vercel
 .vercel

matching-service/package.json

@@ -19,14 +19,12 @@
     "cors": "^2.8.5",
     "dotenv": "^16.4.5",
     "express": "^4.21.1",
-    "redis": "^4.7.0",
-    "ws": "^8.18.0"
+    "redis": "^4.7.0"
   },
   "devDependencies": {
     "@types/amqplib": "^0.10.5",
     "@types/cors": "^2.8.17",
     "@types/express": "^5.0.0",
-    "@types/ws": "^8.5.12",
     "nodemon": "^3.1.7",
     "prettier": "^3.3.3",
     "ts-node": "^10.9.2",

matching-service/src/app.ts

@@ -1,15 +1,11 @@
 import express from "express";
 import cors from "cors";
-import http from "http";
 import messageRoutes from "./routes/messageRoute";
 import { initRabbitMQ } from "./services/rabbitMqService";
 import redisClient from "./config/redisConfig";
 import { startBackgroundTransfer } from "./services/matchingService";
-import WebSocketService from "./services/webSocketService";
 
 const app = express();
-const server = http.createServer(app);
-const webSocketService = new WebSocketService(server);
 
 const PORT = process.env.PORT || 5001; // 5001 to prevent conflicts
 
@@ -25,15 +21,7 @@ app.use(cors({ origin: true, credentials: true }));
 // Mainly to check health or state of service
 app.use(`${apiVersion}/`, messageRoutes);
 
-export const notifyMatch = async (
-  user1Id: string,
-  user2Id: string,
-  matchData: any,
-) => {
-  await webSocketService.notifyMatch(user1Id, user2Id, matchData);
-};
-
-server.listen(PORT, async () => {
+app.listen(PORT, async () => {
   await initRabbitMQ();
 
   try {

matching-service/src/model/CustomWebSocket.ts

@@ -61,3 +61,24 @@ export const transferToTopicQueue = async (user: User): Promise<void> => {
     console.error(`Error storing user in topic queue for topic: ${user.topic}`);
   }
 };
+
+export const removeUserFromQueues = async (userId: string) => {
+  const allTopics = await redisClient.keys("topic:*");
+  const allDifficulties = await redisClient.keys("difficulty:*");
+
+  const removeFromQueue = async (key: string) => {
+    const users = await redisClient.zRange(key, 0, -1);
+    for (const userString of users) {
+      const user = JSON.parse(userString);
+      if (user._id === userId) {
+        await removeUserFromKey(key, userString);
+        console.log(`Removed user ${userId} from queue ${key}`);
+        break;
+      }
+    }
+  };
+
+  for (const key of [...allTopics, ...allDifficulties]) {
+    await removeFromQueue(key);
+  }
+};

matching-service/src/model/userModel.ts

@@ -4,11 +4,10 @@ import {
   transferToTopicQueue,
   removeUserFromKey,
   transferToDifficultyQueue,
+  removeUserFromQueues,
 } from "../model/userModel";
 import { User } from "../types";
 import { sendToQueue } from "./rabbitMqService";
-import { notifyMatch } from "../app";
-
 /**
  * Check match in queue, if there is a match, we will remove that user from the queue, and return
  * a response to rabbit.
@@ -63,12 +62,8 @@ export const processOldUsers = async (): Promise<void> => {
           status: "matched",
           match: match,
         });
-        console.log(
-          `Notifying match for OLD users: ${match._id} and ${user._id}`,
-        );
-        notifyMatch(match._id, user._id, { user1: match, user2: user });
-        return;
-        // continue;
+
+        continue;
       }
 
       // Transfer to difficulty queue if there is no match
@@ -98,24 +93,29 @@ export const processOldUsers = async (): Promise<void> => {
 };
 
 // Process a new user and attempt matching
-export const processNewUser = async (user: User): Promise<void> => {
-  const match = await checkMatch(`topic:${user.topic}`);
+export const processNewMessage = async (user: User): Promise<void> => {
+  if (user.type == "match") {
+    const match = await checkMatch(`topic:${user.topic}`);
 
-  // match found
-  if (match) {
-    await sendToQueue(match._id, {
-      status: "matched",
-      match: user,
-    });
-    await sendToQueue(user._id, {
-      status: "matched",
-      match: match,
-    });
-    // Call notifyMatch here
-    notifyMatch(match._id, user._id, { user1: match, user2: user });
-  } else {
-    // Add to the topic queue if no match
-    await transferToTopicQueue(user);
+    // match found
+    if (match) {
+      await sendToQueue(match._id, {
+        status: "matched",
+        match: user,
+      });
+      await sendToQueue(user._id, {
+        status: "matched",
+        match: match,
+      });
+    } else {
+      // Add to the topic queue if no match
+      await transferToTopicQueue(user);
+    }
+  } else if (user.type == "cancel") {
+    //handle cancel request
+    console.log(`Cancellation request received for user: ${user._id}`);
+    // Remove the user from topic and difficulty queues
+    await removeUserFromQueues(user._id);
   }
 };
 

matching-service/src/services/matchingService.ts

@@ -1,6 +1,6 @@
 import amqp, { Channel } from "amqplib";
 import { User } from "../types";
-import { processNewUser } from "./matchingService";
+import { processNewMessage } from "./matchingService";
 import { SECONDS } from "../lib/constants";
 
 let channel: Channel;
@@ -17,7 +17,7 @@ export const initRabbitMQ = async (): Promise<void> => {
     console.log("Connected to RabbitMQ");
 
     // Subscribe to the queue upon successful connection
-    await subscribeToQueue<User>(QUEUE, processNewUser);
+    await subscribeToQueue<User>(QUEUE, processNewMessage);
 
     // when connection close, initiate reconnection
     connection.on("close", async () => {

matching-service/src/services/rabbitMqService.ts

@@ -1,6 +1,7 @@
 export interface User {
   _id: string;
-  name: string;
-  difficulty: string;
-  topic: string;
+  name?: string;
+  difficulty?: string;
+  topic?: string;
+  type: string;
 }

matching-service/src/services/webSocketService.ts

@@ -19,6 +19,7 @@
     "@radix-ui/react-scroll-area": "^1.2.0",
     "@radix-ui/react-select": "^2.1.1",
     "@radix-ui/react-slot": "^1.1.0",
+    "@stomp/stompjs": "^7.0.0",
     "@types/js-cookie": "^3.0.6",
     "amqplib": "^0.10.4",
     "axios": "^1.7.7",