Refactor redis service

Author: Jajared

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

@@ -11,6 +11,8 @@ import { ClientProxy } from '@nestjs/microservices';
 import { Inject } from '@nestjs/common';
 import { firstValueFrom } from 'rxjs';
 import { RedisService } from './redis.service';
+import { MatchRequestDto } from './dto';
+import { MATCH_SUCCESS, MATCH_TIMEOUT } from './match.event';
 
 @WebSocketGateway({ namespace: '/match' })
 export class MatchGateway implements OnGatewayInit {
@@ -25,23 +27,25 @@ export class MatchGateway implements OnGatewayInit {
   afterInit() {
     // Subscribe to Redis Pub/Sub for match notifications
     this.redisService.subscribeToMatchEvents((matchedUsers) => {
-      this.notifyUsers(matchedUsers);
+      this.notifyUsersWithMatch(matchedUsers);
+    }); 
+
+    this.redisService.subscribeToTimeoutEvents((timedOutUsers) => {
+      this.notifyUsersWithTimeout(timedOutUsers);
     });
   }
 
-  @SubscribeMessage('requestMatch')
+  @SubscribeMessage('matchRequest')
   async handleRequestMatch(
     @ConnectedSocket() client: Socket,
-    @MessageBody() payload: any,
+    @MessageBody() payload: MatchRequestDto,
   ) {
     const matchPayload = {
       userId: payload.userId,
       selectedTopic: payload.selectedTopic,
       selectedDifficulty: payload.selectedDifficulty,
     };
 
-    let isMatched = false;
-
     // Send the match request to the microservice
     try {
       await firstValueFrom(this.matchingClient.send('match.request', matchPayload));
@@ -52,26 +56,36 @@ export class MatchGateway implements OnGatewayInit {
   }
 
   // Notify both matched users via WebSocket
-  notifyUsers(matchedUsers: string[]) {
+  notifyUsersWithMatch(matchedUsers: string[]) {
     const [user1, user2] = matchedUsers;
     const user1SocketId = this.getUserSocketId(user1);
     const user2SocketId = this.getUserSocketId(user2);
     if (user1SocketId) {
-      this.server.to(user1SocketId).emit('matchNotification', {
+      this.server.to(user1SocketId).emit(MATCH_SUCCESS, {
         message: `You have been matched with user ${user2}`,
         matchedUserId: user2,
       });
     }
 
     if (user2SocketId) {
-      this.server.to(user2SocketId).emit('matchNotification', {
+      this.server.to(user2SocketId).emit(MATCH_SUCCESS, {
         message: `You have been matched with user ${user1}`,
         matchedUserId: user1,
       });
     }
   }
 
-  
+  notifyUsersWithTimeout(timedOutUsers: string[]) {
+    timedOutUsers.forEach(user => {
+      const socketId = this.getUserSocketId(user);
+      if (socketId) {
+        this.server.to(socketId).emit(MATCH_TIMEOUT, {
+          message: `You have been timed out.`,
+          timedOutUserId: user,
+        });
+      }
+    });
+  }
 
   handleConnection(@ConnectedSocket() client: Socket) {
     const userId = client.handshake.query.userId;

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

@@ -0,0 +1,3 @@
+export const MATCH_TIMEOUT = 'matchTimeout';
+export const MATCH_SUCCESS = 'matchSuccess';
+

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

@@ -7,7 +7,7 @@ export class RedisService {
 
   constructor() {
     this.redisSubscriber = new Redis({
-      host: 'backend-redis-1', // Your Redis host
+      host: 'backend-redis-1',
       port: 6379,
     });
   }
@@ -30,4 +30,21 @@ export class RedisService {
       }
     });
   }
+
+  subscribeToTimeoutEvents(callback: (matchedUsers: any) => void): void {
+    this.redisSubscriber.subscribe('timeoutChannel', (err, count) => {
+      if (err) {
+        console.error('Error subscribing to Redis channel:', err);
+        return;
+      }
+      console.log('Subscribed to Redis channel timeoutChannel.');
+    });
+
+    this.redisSubscriber.on('message', (channel, message) => {
+      if (channel === 'timeoutChannel') {
+        const timedOutUsers = JSON.parse(message);
+        callback(timedOutUsers);
+      }
+    });
+  }
 }

backend/matching-service/src/interfaces/match-job.interface.ts

@@ -3,4 +3,5 @@ export interface MatchJob {
   userProficiency: string;
   selectedTopic: string[];
   selectedDifficulty: string;
+  timestamp: number;
 }

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

@@ -11,10 +11,20 @@ export class MatchWorkerService {
   async pollForMatches() {
     setInterval(async () => {
       const users = await this.redisService.getUsersFromPool();
-      console.log('polling', users)
+      const currentTime = Date.now();
+      const timeout = 29000; // Slightly less than 30 seconds to account for processing time
+      console.log('Polling', users);
+      // Filter out users who have timed out
+      const activeUsers = users.filter(user => currentTime - user.timestamp < timeout);
 
-      if (users.length >= 2) {
-        const matches = this.rankUsers(users);
+      if (activeUsers.length < users.length) {
+        const timedOutUsers = users.filter(user => currentTime - user.timestamp >= timeout);
+        await this.notifyGatewayTimeout(timedOutUsers.map(user => user.userId));
+        await this.redisService.removeUsersFromPool(timedOutUsers.map(user => user.userId));
+      }
+
+      if (activeUsers.length >= 2) {
+        const matches = this.rankUsers(activeUsers);
         const bestMatch = matches[0];
 
         // Notify the gateway via Redis Pub/Sub
@@ -23,7 +33,7 @@ export class MatchWorkerService {
         // Remove the matched users from the Redis pool
         await this.redisService.removeUsersFromPool([bestMatch.user1.userId, bestMatch.user2.userId]);
       }
-    }, 5000);
+    }, 20000);
   }
 
   // Ranking logic for matches
@@ -52,5 +62,9 @@ export class MatchWorkerService {
   async notifyGateway(matchedUserIds: string[]) {
     await this.redisService.publishMatchedUsers(matchedUserIds);
   }
+
+  async notifyGatewayTimeout(userIds: string[]) {
+    await this.redisService.publishTimedOutUsers(userIds);
+  }
 }
 

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

@@ -1,6 +1,7 @@
 import { Injectable } from '@nestjs/common';
 import Redis from 'ioredis';
 import { MatchRequestDto } from './dto/match-request.dto';
+import { MatchJob } from './interfaces/match-job.interface';
 
 @Injectable()
 export class RedisService {
@@ -20,11 +21,18 @@ export class RedisService {
 
   // Add user to Redis pool
   async addUserToPool(data: MatchRequestDto): Promise<void> {
-    await this.redisPublisher.sadd('userPool', JSON.stringify(data));
+    const payload: MatchJob = {
+      userId: data.userId,
+      userProficiency: 'beginner',
+      selectedTopic: data.selectedTopic,
+      selectedDifficulty: data.selectedDifficulty,
+      timestamp: Date.now(),
+    };
+    await this.redisPublisher.sadd('userPool', JSON.stringify(payload));
   }
 
   // Get users from Redis pool
-  async getUsersFromPool(): Promise<MatchRequestDto[]> {
+  async getUsersFromPool(): Promise<MatchJob[]> {
     const users = await this.redisPublisher.smembers('userPool');
     return users.map(user => JSON.parse(user));
   }
@@ -46,4 +54,8 @@ export class RedisService {
   async publishMatchedUsers(matchedUserIds: string[]): Promise<void> {
     await this.redisPublisher.publish('matchChannel', JSON.stringify(matchedUserIds));
   }
+
+  async publishTimedOutUsers(timedOutUserIds: string[]): Promise<void> {
+    await this.redisPublisher.publish('timeoutChannel', JSON.stringify(timedOutUserIds));
+  }
 }