Refactor redis service

Author: Jajared

backend/gateway-service/src/modules/match/match.controller.ts

@@ -1,43 +1,34 @@
 import {
   WebSocketGateway,
   WebSocketServer,
-  OnGatewayInit,
-  ConnectedSocket,
   SubscribeMessage,
+  ConnectedSocket,
   MessageBody,
+  OnGatewayInit,
 } from '@nestjs/websockets';
-import { Server, Socket } from 'socket.io';
-import { RedisService } from './redis.service';
+import { Socket, Server } from 'socket.io';
 import { ClientProxy } from '@nestjs/microservices';
 import { Inject } from '@nestjs/common';
+import { firstValueFrom } from 'rxjs';
+import { RedisService } from './redis.service';
 
 @WebSocketGateway({ namespace: '/match' })
 export class MatchGateway implements OnGatewayInit {
   @WebSocketServer() server: Server;
+  private userSockets: Map<string, string> = new Map();
 
   constructor(
-    private redisService: RedisService,
     @Inject('MATCHING_SERVICE') private matchingClient: ClientProxy,
+    private redisService: RedisService,
   ) {}
 
   afterInit() {
-    // Subscribe to Redis Pub/Sub for user match notifications
-    this.redisService.subscribeToUserMatchEvents((payload) => {
-      this.notifyUsers(payload);
+    // Subscribe to Redis Pub/Sub for match notifications
+    this.redisService.subscribeToMatchEvents((matchedUsers) => {
+      this.notifyUsers(matchedUsers);
     });
   }
 
-  // Handle user connection and add them to a room based on their userId
-  handleConnection(@ConnectedSocket() client: Socket) {
-    const userId = client.handshake.query.userId; // Extract userId from query
-    if (userId) {
-      client.join(userId); // Make the client join a room based on userId
-      console.log(`User ${userId} connected and joined room ${userId}`);
-    } else {
-      console.log('User connected without userId');
-    }
-  }
-
   @SubscribeMessage('requestMatch')
   async handleRequestMatch(
     @ConnectedSocket() client: Socket,
@@ -49,33 +40,59 @@ export class MatchGateway implements OnGatewayInit {
       selectedDifficulty: payload.selectedDifficulty,
     };
 
-    // Emit match request event to the microservice
-    const timeoutRef = setTimeout(() => {
-      client.emit('matchTimeout', 'No match found within 30 seconds');
-    }, 30000); // Timeout after 30 seconds if no match is found
+    let isMatched = false;
 
-    this.matchingClient.emit('match.request', matchPayload).subscribe({
-      next: () => {
-        // If a match is found, clear the timeout
-        clearTimeout(timeoutRef);
-        client.emit('matchInProgress', 'Finding a match...');
-      },
-      error: (err) => {
-        clearTimeout(timeoutRef);  // Also clear the timeout in case of an error
-        client.emit('matchError', 'Error finding match');
-      },
-    });
+    // Send the match request to the microservice
+    try {
+      await firstValueFrom(this.matchingClient.send('match.request', matchPayload));
+    } catch (error) {
+      client.emit('matchError', `Error requesting match: ${error.message}`);
+      return;
+    }
   }
 
-  // Notify matched users via WebSocket
-  notifyUsers(userIds: string[]) {
-    userIds.forEach((userId, index) => {
-      const matchedUserId = userIds[index === 0 ? 1 : 0]; // Get the other matched user's ID
-      console.log(`Notifying user ${userId} about match with ${matchedUserId}`);
-      this.server.to(userId).emit('matchNotification', {
-        message: `You have been matched with user ${matchedUserId}`,
-        matchedUserId: matchedUserId,
+  // Notify both matched users via WebSocket
+  notifyUsers(matchedUsers: string[]) {
+    const [user1, user2] = matchedUsers;
+    const user1SocketId = this.getUserSocketId(user1);
+    const user2SocketId = this.getUserSocketId(user2);
+    if (user1SocketId) {
+      this.server.to(user1SocketId).emit('matchNotification', {
+        message: `You have been matched with user ${user2}`,
+        matchedUserId: user2,
       });
-    });
+    }
+
+    if (user2SocketId) {
+      this.server.to(user2SocketId).emit('matchNotification', {
+        message: `You have been matched with user ${user1}`,
+        matchedUserId: user1,
+      });
+    }
+  }
+
+  
+
+  handleConnection(@ConnectedSocket() client: Socket) {
+    const userId = client.handshake.query.userId;
+    if (userId) {
+      this.userSockets.set(userId as string, client.id);
+      console.log(`User ${userId} connected with socket ID ${client.id}`);
+    }
+  }
+
+  handleDisconnect(@ConnectedSocket() client: Socket) {
+    const userId = [...this.userSockets.entries()].find(
+      ([, socketId]) => socketId === client.id,
+    )?.[0];
+
+    if (userId) {
+      this.userSockets.delete(userId);
+      console.log(`User ${userId} disconnected`);
+    }
+  }
+
+  getUserSocketId(userId: string): string | undefined {
+    return this.userSockets.get(userId);
   }
 }

backend/gateway-service/src/modules/match/redis.service.ts

@@ -6,16 +6,14 @@ export class RedisService {
   private redisSubscriber: Redis;
 
   constructor() {
-    // Initialize Redis for subscribing to match notifications
     this.redisSubscriber = new Redis({
-      host: 'backend-redis-1',
+      host: 'backend-redis-1', // Your Redis host
       port: 6379,
     });
   }
 
-  // Subscribe to Redis Pub/Sub events for match notifications
-  subscribeToUserMatchEvents(matchCallback: (users: any) => void) {
-    // Subscribe to the Redis Pub/Sub channel 'matchChannel'
+  // Subscribe to the Redis Pub/Sub channel for match events
+  subscribeToMatchEvents(callback: (matchedUsers: any) => void): void {
     this.redisSubscriber.subscribe('matchChannel', (err, count) => {
       if (err) {
         console.error('Error subscribing to Redis channel:', err);
@@ -24,11 +22,11 @@ export class RedisService {
       console.log('Subscribed to Redis channel matchChannel.');
     });
 
-    // Listen for messages on the 'matchChannel' and trigger the callback
+    // Listen for published messages on the channel
     this.redisSubscriber.on('message', (channel, message) => {
       if (channel === 'matchChannel') {
-        const matchedUsers = JSON.parse(message);  // Parse the JSON message
-        matchCallback(matchedUsers);  // Call the provided callback with the matched users
+        const matchedUsers = JSON.parse(message);
+        callback(matchedUsers);
       }
     });
   }

backend/matching-service/src/app.controller.ts

@@ -1,14 +1,14 @@
 import { Controller, Get } from '@nestjs/common';
 import { AppService } from './app.service';
-import { EventPattern, MessagePattern, Payload } from '@nestjs/microservices';
+import { MessagePattern, Payload } from '@nestjs/microservices';
 import { MatchRequestDto } from './dto';
 
 @Controller()
 export class AppController {
   constructor(private readonly appService: AppService) {}
 
-  @EventPattern('match.request')
+  @MessagePattern('match.request')
   async handleMatchRequest(@Payload() data: MatchRequestDto) {
-    await this.appService.requestMatch(data);
+    await this.appService.requestMatch(data); 
   }
 }

backend/matching-service/src/app.module.ts

@@ -4,7 +4,6 @@ import { AppService } from './app.service';
 import { BullModule } from '@nestjs/bull';
 import { ClientsModule, Transport } from '@nestjs/microservices';
 import { MatchWorkerService } from './match-worker.service';
-import { NotificationService } from './notification.service';
 import { RedisService } from './redis.service';
 
 @Module({
@@ -33,7 +32,6 @@ import { RedisService } from './redis.service';
   providers: [
     AppService,
     MatchWorkerService,
-    NotificationService,
     RedisService,
   ],
 })

backend/matching-service/src/app.service.ts

@@ -1,16 +1,16 @@
-import { Inject, Injectable } from '@nestjs/common';
-import { ClientProxy, RpcException } from '@nestjs/microservices';
-import { MatchRequestDto } from './dto';
+import { Injectable } from '@nestjs/common';
+import Redis from 'ioredis';
+import { MatchRequestDto } from './dto/match-request.dto';
 import { RedisService } from './redis.service';
 
 @Injectable()
 export class AppService {
-  constructor(
-    @Inject('USER_SERVICE') private readonly userClient: ClientProxy,
-    private redisService: RedisService,
-  ) {}
+  constructor(private redisService: RedisService) {}
 
-  async requestMatch(data: MatchRequestDto) {
+  // Add user to the Redis pool
+  async requestMatch(data: MatchRequestDto): Promise<void> {
     await this.redisService.addUserToPool(data);
   }
 }
+
+

backend/matching-service/src/match-worker.service.ts

@@ -1,6 +1,7 @@
 import { Injectable } from '@nestjs/common';
-import { RedisService } from './redis.service';
 import { MatchRequestDto } from './dto';
+import { RedisService } from './redis.service';
+import { PriorityQueue } from './helper/priority-queue';
 
 @Injectable()
 export class MatchWorkerService {
@@ -9,22 +10,20 @@ export class MatchWorkerService {
   // Poll for matches at a regular interval
   async pollForMatches() {
     setInterval(async () => {
-      const users = await this.redisService.getUsersFromPool(); // Get users from Redis pool
+      const users = await this.redisService.getUsersFromPool();
       console.log('polling', users)
 
       if (users.length >= 2) {
         const matches = this.rankUsers(users);
-        const bestMatch = matches[0];  // Find the best match
-
-        console.log(`Matched users: ${bestMatch.user1.userId} and ${bestMatch.user2.userId}`);
+        const bestMatch = matches[0];
 
         // Notify the gateway via Redis Pub/Sub
         await this.notifyGateway([bestMatch.user1.userId, bestMatch.user2.userId]);
 
         // Remove the matched users from the Redis pool
         await this.redisService.removeUsersFromPool([bestMatch.user1.userId, bestMatch.user2.userId]);
       }
-    }, 5000); // Poll every 5 seconds
+    }, 5000);
   }
 
   // Ranking logic for matches
@@ -36,7 +35,7 @@ export class MatchWorkerService {
         matches.push({ user1: users[i], user2: users[j], score });
       }
     }
-    return matches.sort((a, b) => b.score - a.score);  // Sort matches by score
+    return matches.sort((a, b) => b.score - a.score)
   }
 
   private calculateScore(user1: MatchRequestDto, user2: MatchRequestDto): number {
@@ -51,7 +50,6 @@ export class MatchWorkerService {
 
   // Notify the gateway service about the match via Redis Pub/Sub
   async notifyGateway(matchedUserIds: string[]) {
-    // Publish matched users to the Redis Pub/Sub channel
     await this.redisService.publishMatchedUsers(matchedUserIds);
   }
 }

backend/matching-service/src/notification.service.ts

@@ -1,69 +1,49 @@
 import { Injectable } from '@nestjs/common';
 import Redis from 'ioredis';
-import { MatchRequestDto } from 'src/dto';
+import { MatchRequestDto } from './dto/match-request.dto';
 
 @Injectable()
 export class RedisService {
-  private redisPublisher: Redis; 
+  private redisPublisher: Redis;
   private redisSubscriber: Redis;
 
   constructor() {
     this.redisPublisher = new Redis({
-      host: 'backend-redis-1',  // Use your Redis host (or Docker network host if in Docker)
+      host: 'backend-redis-1',
       port: 6379,
     });
-
     this.redisSubscriber = new Redis({
       host: 'backend-redis-1',
       port: 6379,
     });
   }
 
-  // Add user to Redis set (matching pool)
-  async addUserToPool(data: MatchRequestDto) {
+  // Add user to Redis pool
+  async addUserToPool(data: MatchRequestDto): Promise<void> {
     await this.redisPublisher.sadd('userPool', JSON.stringify(data));
   }
 
-   // Get users from Redis pool
-   async getUsersFromPool(): Promise<MatchRequestDto[]> {
+  // Get users from Redis pool
+  async getUsersFromPool(): Promise<MatchRequestDto[]> {
     const users = await this.redisPublisher.smembers('userPool');
-    return users.map(user => JSON.parse(user) as MatchRequestDto);
+    return users.map(user => JSON.parse(user));
   }
 
-   // Remove users from the Redis pool by userId
-   async removeUsersFromPool(userIds: string[]) {
-    const users = await this.getUsersFromPool();  // Get all users in the pool
+  // Remove users from Redis pool
+  async removeUsersFromPool(userIds: string[]) {
+    const users = await this.getUsersFromPool();
 
     // Find and remove users whose userId matches the provided userIds
     userIds.forEach(async (userId) => {
       const userToRemove = users.find(user => user.userId === userId);
       if (userToRemove) {
-        await this.redisPublisher.srem('userPool', JSON.stringify(userToRemove));  // Remove user from pool
-        console.log(`Removed user ${userId} from the pool`);
+        await this.redisPublisher.srem('userPool', JSON.stringify(userToRemove));
       }
     });
   }
 
-  // Publish matched users via Redis Pub/Sub
-  async publishMatchedUsers(matchedUsers: any) {
-    await this.redisPublisher.publish('matchChannel', JSON.stringify(matchedUsers));
-  }
-
-  // Subscribe to Redis Pub/Sub for user match notifications
-  subscribeToUserMatchEvents(matchCallback: (users: any) => void) {
-    this.redisSubscriber.subscribe('matchChannel', (err, count) => {
-      if (err) {
-        console.error('Error subscribing to Redis channel:', err);
-        return;
-      }
-      console.log('Subscribed to Redis channel matchChannel.');
-    });
-
-    this.redisSubscriber.on('message', (channel, message) => {
-      if (channel === 'matchChannel') {
-        const matchedUsers = JSON.parse(message);
-        matchCallback(matchedUsers);
-      }
-    });
+  // Publish matched users to Redis Pub/Sub channel
+  async publishMatchedUsers(matchedUserIds: string[]): Promise<void> {
+    await this.redisPublisher.publish('matchChannel', JSON.stringify(matchedUserIds));
   }
 }