abstract subscribe to queue and add comments

Author: Kurtyjlee

docker-compose.prod.yml

@@ -3,7 +3,7 @@ services:
     image: redis:latest
     container_name: redis-server
     ports:
-      - "6379:6379"  # Expose Redis port
+      - "6379:6379"
     networks:
       - peerprep-network
   frontend:

matching-service/src/app.ts

@@ -3,9 +3,13 @@ import cors from "cors";
 import * as dotenv from "dotenv";
 
 import messageRoutes from "./routes/messageRoute";
-import { initRabbitMQ } from "./services/rabbitMqService";
+import { initRabbitMQ, subscribeToQueue } from "./services/rabbitMqService";
 import redisClient from "./config/redisConfig";
-import { startBackgroundTransfer } from "./services/matchingService";
+import {
+  processNewUser,
+  startBackgroundTransfer,
+} from "./services/matchingService";
+import { User } from "./types";
 
 dotenv.config({ path: ".env.dev" });
 
@@ -22,21 +26,22 @@ const apiVersion = "/api/v1";
 // Allow frontend to talk to service
 app.use(cors({ origin: true, credentials: true }));
 
-app.use(`${apiVersion}/message`, messageRoutes);
-
-// ping server
-app.get(`${apiVersion}/ping`, (req, res) => {
-  res.status(200).json({ data: "pong" });
-});
+// Mainly to check health or state of service
+app.use(`${apiVersion}/`, messageRoutes);
 
 // Start the server
 app.listen(PORT, async () => {
-  await initRabbitMQ(QUEUE);
+  await initRabbitMQ();
+
+  // Subscribe to the matching service queue
+  await subscribeToQueue<User>(QUEUE, processNewUser);
+
   try {
     await redisClient.connect();
   } catch (err) {
     console.error("Error connecting to Redis:", err);
   }
+
   console.log(`Server is running on port ${PORT}`);
 
   startBackgroundTransfer();

matching-service/src/model/userModel.ts

@@ -1,19 +1,15 @@
 import redisClient from "../config/redisConfig";
 import { User } from "../types";
 
-// Store user in Redis using async/await
-export const storeUser = async (user: User): Promise<string | null> => {
-  try {
-    // Store the user in Redis, Redis 4.x+ returns a Promise, no need for callbacks
-    const reply = await redisClient.set(user._id, JSON.stringify(user));
-    console.log("User stored in Redis:", reply);
-    return reply;
-  } catch (err) {
-    console.error("Redis set error:", err);
-    throw err;
-  }
-};
-
+/**
+ * Store a user in an ordered set in Redis, where the key represents the location
+ * and the user is stored with a score based on the current timestamp.
+ *
+ * @param key key to store user in
+ * @param user The user object to store, which will be serialized to JSON.
+ * @returns The number of elements added to the sorted set (not including all the elements already present).
+ *          If an error occurs, it returns null.
+ */
 export const storeUserToKey = async (
   key: string,
   user: User,
@@ -24,6 +20,16 @@ export const storeUserToKey = async (
   ]);
 };
 
+/**
+ * Remove a user from an ordered set in Redis.
+ *
+ * This function removes the specified user from the ordered set identified by the given key.
+ *
+ * @param key - The key of the ordered set from which the user will be removed.
+ * @param user - The string representation of the user to be removed from the ordered set.
+ * @returns The number of elements removed from the sorted set (1 if the user was removed, 0 if the user was not found).
+ *          Returns null if an error occurs during the operation.
+ */
 export const removeUserFromKey = async (
   key: string,
   user: string,
@@ -32,6 +38,9 @@ export const removeUserFromKey = async (
   return redisClient.zRem(key, user);
 };
 
+/**
+ * Transfers user to the difficulty queue based on the user's difficulty level.
+ */
 export const transferToDifficultyQueue = async (user: User): Promise<void> => {
   const difficultyKey = `difficulty:${user.difficulty}`;
   const res = await storeUserToKey(difficultyKey, user);
@@ -42,6 +51,9 @@ export const transferToDifficultyQueue = async (user: User): Promise<void> => {
   }
 };
 
+/**
+ * Transfers user to the topic queue based on the user's topic.
+ */
 export const transferToTopicQueue = async (user: User): Promise<void> => {
   const topicKey = `topic:${user.topic}`;
   const res = await storeUserToKey(topicKey, user);

matching-service/src/routes/messageRoute.ts

@@ -3,6 +3,7 @@ import { returnPing } from "../controller/messageController";
 
 const router = express.Router();
 
+// Check the health of the service
 router.get("/ping", returnPing);
 
 export default router;

matching-service/src/services/matchingService.ts

@@ -14,7 +14,7 @@ import { sendToQueue } from "./rabbitMqService";
  * If there is no match, we will return null.
  *
  * @param topic
- * @returns
+ * @returns User if there is a match, null if there is no match
  */
 export const checkMatch = async (key: string): Promise<User | null> => {
   const match = await redisClient.zRange(key, 0, 0); // Get the first user in the topic queue
@@ -26,7 +26,6 @@ export const checkMatch = async (key: string): Promise<User | null> => {
     if (res === null) {
       console.error(`Error removing user from queue for key: ${key}`);
     }
-
     return user;
   }
   return null;
@@ -111,12 +110,13 @@ export const processNewUser = async (user: User): Promise<void> => {
       user2: user,
     });
   } else {
+    // Add to the topic queue if no match
     await transferToTopicQueue(user);
   }
 };
 
 // Start background transfer process, polling every 5 sec
-export const startBackgroundTransfer = (): void => {
+export const startBackgroundTransfer = () => {
   setInterval(async () => {
     await processOldUsers();
   }, 5 * SECONDS);

matching-service/src/services/rabbitMqService.ts

@@ -1,47 +1,68 @@
 import amqp, { Channel } from "amqplib";
-import { storeUser } from "../model/userModel";
-import { User } from "../types";
-import { processNewUser } from "./matchingService";
 
 let channel: Channel;
 const rabbit_url = process.env.RABBITMQ_URL || "amqp://localhost";
 
-export const initRabbitMQ = async (queue: string): Promise<void> => {
+/**
+ * Establish a connection to RabbitMQ
+ */
+export const initRabbitMQ = async (): Promise<void> => {
   try {
     const connection = await amqp.connect(rabbit_url);
     channel = await connection.createChannel();
-    await channel.assertQueue(queue, { durable: true });
-    console.log(`RabbitMQ initialised and subscribed to ${queue}`);
-
-    // Start consuming messages from matching-service
-    channel.consume(queue, async (msg) => {
-      if (msg !== null) {
-        const user: User = JSON.parse(msg.content.toString());
-        console.log("Received message from queue:", user);
-
-        // Store the user into Redis
-        try {
-          await processNewUser(user);
-        } catch (err) {
-          console.error("Error storing user in Redis:", err);
-        }
-
-        channel.ack(msg);
-      }
-    });
+    console.log("Connected to RabbitMQ");
   } catch (err) {
     console.error("RabbitMQ connection error:", err);
   }
 };
 
+/**
+ * Get the RabbitMQ channel
+ * @returns RabbitMQ channel
+ */
 export const getChannel = (): amqp.Channel => {
   if (!channel) {
     throw new Error("RabbitMQ channel is not initialized");
   }
   return channel;
 };
 
-// Send a payload to a specified RabbitMQ queue
+/**
+ * Subscribes to a RabbitMQ queue and calls the callback function when a message is received.
+ * @param queue queue to subscribe to
+ * @param callback callback when message is received
+ */
+export const subscribeToQueue = async <T>(
+  queue: string,
+  callback: (message: T) => void,
+): Promise<void> => {
+  try {
+    if (!channel) throw new Error("RabbitMQ channel is not initialized");
+
+    await channel.assertQueue(queue);
+    console.log(`Subscribed to RabbitMQ queue "${queue}"`);
+
+    channel.consume(queue, (msg) => {
+      if (msg !== null) {
+        const message: T = JSON.parse(msg.content.toString());
+        console.log(
+          `Received message from RabbitMQ queue "${queue}":`,
+          message,
+        );
+        callback(message);
+        channel.ack(msg);
+      }
+    });
+  } catch (err) {
+    console.error(`Error subscribing to RabbitMQ queue "${queue}":`, err);
+  }
+};
+
+/**
+ * Sends a payload to a RabbitMQ queue
+ * @param queue queue to send payload to
+ * @param payload payload object to send
+ */
 export const sendToQueue = async (
   queue: string,
   payload: Object,