Refactor code and include type use for websocket

Author: ketamethane

Backend/MatchingService/rabbitmq/publisher.js

@@ -8,9 +8,6 @@ async function publishToQueue({userId, difficulty, language}) {
         const channel = await connection.createChannel();
         const routingKey = `${difficulty}.${language}`;
 
-        // Connect to the exchange (just in case it does not exist)
-        await channel.assertExchange(matching_exchange_name, 'topic', { durable: false });
-
         // Publish the message to the exchange
         const messageSent = channel.publish(
             matching_exchange_name,

Backend/MatchingService/rabbitmq/setup.js

@@ -46,6 +46,9 @@ async function setupRabbitMQ() {
             await channel.bindQueue(queueName, matching_exchange_name, queueName); // Bind to exchange
         }
 
+        // Delete DLQ before asserting it
+        await channel.deleteQueue(dead_letter_queue_name);
+
         // Declare the dead-letter queue and bind it to the dead-letter exchange
         await channel.assertQueue(dead_letter_queue_name, { durable: false });
         await channel.bindQueue(dead_letter_queue_name, dead_letter_exchange_name, ''); // Bind with no routing key

Backend/MatchingService/rabbitmq/subscriber.js

@@ -10,71 +10,10 @@ const { notifyUsers } = require('../websocket/websocket');
 
 const dead_letter_queue_name = "dead_letter_queue";
 const timeoutMap = {};
+
 // Local dictionary to store waiting users
 const waitingUsers = {};
 
-// manual rejection of message that is constantly being consumed. Not using TTL feature.
-// Race condition: upon matching user 2 with user 1, the flag for whether user 1 is matched is still false.
-// async function consumeQueue() {
-//   try {
-//       // Connect
-//       const connection = await amqp.connect(process.env.RABBITMQ_URL);
-//       const channel = await connection.createChannel();
-
-//       console.log("Waiting for users...");
-
-//       // Process + subscribe to each matchmaking queue
-//       for (let queueName of queueNames) {
-//           await channel.consume(queueName, (msg) => {
-//               if (msg !== null) {
-//                   const userData = JSON.parse(msg.content.toString());
-//                   const { userId, language, difficulty } = userData;
-
-//                   // Perform the matching logic
-//                   console.log(`Received user ${userId} with ${language} and ${difficulty}`);
-//                   const matched = matchUsers(userId, language, difficulty);
-
-//                   // Flag to track if a match is found
-//                   // let isMatched = false;
-//                   if (timeoutMap[userId]) {
-//                     clearTimeout(timeoutMap[userId]);
-//                     delete timeoutMap[userId];
-//                 }
-
-//                   // Only acknowledge if a match was found
-//                   if (matched) {
-//                       // isMatched = true; // Set the flag
-//                       channel.ack(msg);
-//                       console.log(`Matched user ${userId}`);
-
-//                       // clearTimeout(timeoutMap[userId]);
-//                       // delete timeoutMap[userId];
-//                   } else {
-//                       console.log(`No match for ${userId}, waiting for TTL to expire.`);
-
-//                       // Set a timeout for rejection only if not matched
-//                       const timeoutId = setTimeout(() => {
-//                           // if (!isMatched) { // Check if matched after timeout
-//                           console.log(`is the user matched upon timeout? ${matched}`);
-//                           console.log(`Rejecting user ${userId} after 10 seconds.`);
-//                           channel.reject(msg, false); // Reject without requeuing
-//                           // }
-//                       }, 10000); // 10 seconds delay
-
-//                       timeoutMap[userId] = timeoutId;
-//                   }
-//               }
-//           });
-//       }
-
-//       console.log("Listening to matchmaking queues");
-//   } catch (error) {
-//       console.error('Error consuming RabbitMQ queue:', error);
-//   }
-// }
-
-
-
 // using promises to handle errors and ensure clearing of timer.
 function matchUsers(channel, msg, userId, language, difficulty) {
     const criteriaKey = `${difficulty}.${language}`;
@@ -93,7 +32,8 @@ function matchUsers(channel, msg, userId, language, difficulty) {
         console.log(`Matched users: ${matchedUsers.map(user => user.userId)}`);
 
         // Send match success (this could trigger WebSocket communication)
-        notifyUsers(matchedUsers.map(user => user.userId));
+        // notifyUsers(matchedUsers.map(user => user.userId));
+        notifyUsers(matchedUsers.map(user => user.userId), 'Match found!', 'match');
 
         // Acknowledge the messages for both matched users
         matchedUsers.forEach(({ msg }) => {
@@ -191,7 +131,7 @@ async function consumeDLQ() {
               console.log(`Received message from DLQ for user: ${userId}`);
 
               // Notify the user via WebSocket
-              notifyUsers(userId, `Match not found for ${difficulty} ${language}, please try again.`);
+              notifyUsers(userId, `Match not found for ${difficulty} ${language}, please try again.`, 'rejection');
 
               // Acknowledge the message (so it's removed from the DLQ)
               channel.ack(msg);

Backend/MatchingService/services/matchingService.js

@@ -1,5 +1,3 @@
-// TODO: Write socket logic to connect backend to frontend here 
-
 const WebSocket = require('ws');
 const wss = new WebSocket.Server({ port: 8080 });
 
@@ -15,11 +13,23 @@ wss.on('connection', (ws) => {
     });
 });
 
-function notifyUsers(userId, message) {
-  console.log(`Notifying user: ${userId}, Message: ${message}`); // Add this log to track notifications
+/**
+ * Notify users through WebSocket.
+ * @param {string|array} userId - User ID or an array of user IDs to notify.
+ * @param {string} message - The message to send.
+ * @param {string} type - The type of message (e.g., 'match' or 'rejection').
+ */
+function notifyUsers(userId, message, type) {
+    console.log(`Notifying user: ${userId}, Message: ${message}, Type: ${type}`); // Log message details
+    
     wss.clients.forEach((client) => {
         if (client.readyState === WebSocket.OPEN) {
-            client.send(JSON.stringify({ userId, message }));
+            // Construct the payload to include userId, message, and type
+            client.send(JSON.stringify({
+                userId, 
+                message,
+                type // This allows the frontend to differentiate between match and rejection messages
+            }));
         }
     });
 }