feat: implement Redis-backed Socket.io adapter and user presence service for cross-instance communication

package-lock.json

@@ -33,6 +33,7 @@
         "@prisma/adapter-pg": "^7.2.0",
         "@prisma/client": "^7.2.0",
         "@sendgrid/mail": "^8.1.6",
+        "@socket.io/redis-adapter": "^8.3.0",
         "argon2": "^0.44.0",
         "axios": "^1.13.2",
         "bull": "^4.16.5",
@@ -6360,6 +6361,49 @@
       "integrity": "sha512-9BCxFwvbGg/RsZK9tjXd8s4UcwR0MWeFQ1XEKIQVVvAGJyINdrqKMcTRyLoK8Rse1GjzLV9cwjWV1olXRWEXVA==",
       "license": "MIT"
     },
+    "node_modules/@socket.io/redis-adapter": {
+      "version": "8.3.0",
+      "resolved": "https://registry.npmjs.org/@socket.io/redis-adapter/-/redis-adapter-8.3.0.tgz",
+      "integrity": "sha512-ly0cra+48hDmChxmIpnESKrc94LjRL80TEmZVscuQ/WWkRP81nNj8W8cCGMqbI4L6NCuAaPRSzZF1a9GlAxxnA==",
+      "license": "MIT",
+      "dependencies": {
+        "debug": "~4.3.1",
+        "notepack.io": "~3.0.1",
+        "uid2": "1.0.0"
+      },
+      "engines": {
+        "node": ">=10.0.0"
+      },
+      "peerDependencies": {
+        "socket.io-adapter": "^2.5.4"
+      }
+    },
+    "node_modules/@socket.io/redis-adapter/node_modules/debug": {
+      "version": "4.3.7",
+      "resolved": "https://registry.npmjs.org/debug/-/debug-4.3.7.tgz",
+      "integrity": "sha512-Er2nc/H7RrMXZBFCEim6TCmMk02Z8vLC2Rbi1KEBggpo0fS6l0S1nnapwmIi3yW/+GOJap1Krg4w0Hg80oCqgQ==",
+      "license": "MIT",
+      "dependencies": {
+        "ms": "^2.1.3"
+      },
+      "engines": {
+        "node": ">=6.0"
+      },
+      "peerDependenciesMeta": {
+        "supports-color": {
+          "optional": true
+        }
+      }
+    },
+    "node_modules/@socket.io/redis-adapter/node_modules/uid2": {
+      "version": "1.0.0",
+      "resolved": "https://registry.npmjs.org/uid2/-/uid2-1.0.0.tgz",
+      "integrity": "sha512-+I6aJUv63YAcY9n4mQreLUt0d4lvwkkopDNmpomkAUz0fAkEMV9pRWxN0EjhW1YfRhcuyHg2v3mwddCDW1+LFQ==",
+      "license": "MIT",
+      "engines": {
+        "node": ">= 4.0.0"
+      }
+    },
     "node_modules/@standard-schema/spec": {
       "version": "1.1.0",
       "resolved": "https://registry.npmjs.org/@standard-schema/spec/-/spec-1.1.0.tgz",
@@ -15052,6 +15096,12 @@
         "node": ">=0.10.0"
       }
     },
+    "node_modules/notepack.io": {
+      "version": "3.0.1",
+      "resolved": "https://registry.npmjs.org/notepack.io/-/notepack.io-3.0.1.tgz",
+      "integrity": "sha512-TKC/8zH5pXIAMVQio2TvVDTtPRX+DJPHDqjRbxogtFiByHyzKmy96RA0JtCQJ+WouyyL4A10xomQzgbUT+1jCg==",
+      "license": "MIT"
+    },
     "node_modules/npm-run-path": {
       "version": "4.0.1",
       "resolved": "https://registry.npmjs.org/npm-run-path/-/npm-run-path-4.0.1.tgz",

package.json

@@ -51,6 +51,7 @@
     "@prisma/adapter-pg": "^7.2.0",
     "@prisma/client": "^7.2.0",
     "@sendgrid/mail": "^8.1.6",
+    "@socket.io/redis-adapter": "^8.3.0",
     "argon2": "^0.44.0",
     "axios": "^1.13.2",
     "bull": "^4.16.5",

src/app.module.ts

@@ -6,6 +6,7 @@ import { ScheduleModule } from '@nestjs/schedule';
 import { EventEmitterModule } from '@nestjs/event-emitter';
 
 import { configuration } from './config';
+import { RedisModule } from './common/redis/redis.module';
 import { PrismaModule } from './prisma';
 import { AuthModule, JwtAuthGuard, RolesGuard } from './auth';
 import { UsersModule } from './users';
@@ -35,6 +36,9 @@ import { AppService } from './app.service';
       envFilePath: ['.env.local', '.env'],
     }),
 
+    // Redis (global)
+    RedisModule,
+
     // Rate Limiting
     ThrottlerModule.forRoot([
       {

src/common/adapters/redis-io.adapter.ts

@@ -0,0 +1,42 @@
+import { IoAdapter } from '@nestjs/platform-socket.io';
+import { INestApplicationContext, Logger } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import { ServerOptions } from 'socket.io';
+import { createAdapter } from '@socket.io/redis-adapter';
+import Redis from 'ioredis';
+
+export class RedisIoAdapter extends IoAdapter {
+  private readonly logger = new Logger(RedisIoAdapter.name);
+  private adapterConstructor: ReturnType<typeof createAdapter>;
+
+  constructor(
+    app: INestApplicationContext,
+    private readonly configService: ConfigService,
+  ) {
+    super(app);
+  }
+
+  async connectToRedis(): Promise<void> {
+    const host = this.configService.get<string>('redis.host', 'localhost');
+    const port = this.configService.get<number>('redis.port', 6379);
+    const password = this.configService.get<string>('redis.password') || undefined;
+    const db = this.configService.get<number>('redis.db', 0);
+
+    const pubClient = new Redis({ host, port, password, db });
+    const subClient = new Redis({ host, port, password, db });
+
+    await Promise.all([
+      new Promise<void>((resolve) => pubClient.on('connect', resolve)),
+      new Promise<void>((resolve) => subClient.on('connect', resolve)),
+    ]);
+
+    this.adapterConstructor = createAdapter(pubClient, subClient);
+    this.logger.log(`Redis adapter connected to ${host}:${port}`);
+  }
+
+  createIOServer(port: number, options?: ServerOptions) {
+    const server = super.createIOServer(port, options);
+    server.adapter(this.adapterConstructor);
+    return server;
+  }
+}

src/common/redis/redis.module.ts

@@ -0,0 +1,9 @@
+import { Global, Module } from '@nestjs/common';
+import { redisProvider, REDIS_CLIENT } from './redis.provider';
+
+@Global()
+@Module({
+  providers: [redisProvider],
+  exports: [REDIS_CLIENT],
+})
+export class RedisModule {}

src/common/redis/redis.provider.ts

@@ -0,0 +1,31 @@
+import { ConfigService } from '@nestjs/config';
+import Redis from 'ioredis';
+
+export const REDIS_CLIENT = 'REDIS_CLIENT';
+
+export const redisProvider = {
+  provide: REDIS_CLIENT,
+  useFactory: (configService: ConfigService) => {
+    const client = new Redis({
+      host: configService.get<string>('redis.host', 'localhost'),
+      port: configService.get<number>('redis.port', 6379),
+      password: configService.get<string>('redis.password') || undefined,
+      db: configService.get<number>('redis.db', 0),
+      retryStrategy: (times: number) => {
+        if (times > 10) return null; // Stop retrying after 10 attempts
+        return Math.min(times * 200, 2000);
+      },
+    });
+
+    client.on('connect', () => {
+      console.log('[Redis] Client connected');
+    });
+
+    client.on('error', (err) => {
+      console.error('[Redis] Client error:', err.message);
+    });
+
+    return client;
+  },
+  inject: [ConfigService],
+};

src/common/services/redis-presence.service.ts

@@ -0,0 +1,45 @@
+import { Inject, Injectable, Logger, OnModuleInit } from '@nestjs/common';
+import Redis from 'ioredis';
+import { REDIS_CLIENT } from '../redis/redis.provider';
+
+const PRESENCE_KEY = 'presence:users';
+
+@Injectable()
+export class RedisPresenceService implements OnModuleInit {
+  private readonly logger = new Logger(RedisPresenceService.name);
+
+  constructor(@Inject(REDIS_CLIENT) private readonly redis: Redis) {}
+
+  async onModuleInit() {
+    // Clear stale presence data on startup (single-server safe)
+    await this.clearAll();
+    this.logger.log('Cleared stale presence data on startup');
+  }
+
+  async setOnline(userId: string): Promise<void> {
+    await this.redis.hset(PRESENCE_KEY, userId, new Date().toISOString());
+  }
+
+  async setOffline(userId: string): Promise<void> {
+    await this.redis.hdel(PRESENCE_KEY, userId);
+  }
+
+  async isOnline(userId: string): Promise<boolean> {
+    const exists = await this.redis.hexists(PRESENCE_KEY, userId);
+    return exists === 1;
+  }
+
+  async getOnlineUsers(userIds: string[]): Promise<Set<string>> {
+    if (userIds.length === 0) return new Set();
+    const results = await this.redis.hmget(PRESENCE_KEY, ...userIds);
+    const online = new Set<string>();
+    userIds.forEach((id, i) => {
+      if (results[i] !== null) online.add(id);
+    });
+    return online;
+  }
+
+  async clearAll(): Promise<void> {
+    await this.redis.del(PRESENCE_KEY);
+  }
+}

src/main.ts

@@ -8,6 +8,7 @@ import cookieParser from 'cookie-parser';
 import { AppModule } from './app.module';
 import { AllExceptionsFilter } from './common/filters';
 import { TransformInterceptor } from './common/interceptors';
+import { RedisIoAdapter } from './common/adapters/redis-io.adapter';
 
 async function bootstrap() {
   const logger = new Logger('Bootstrap');
@@ -101,6 +102,11 @@ async function bootstrap() {
     logger.log(`Swagger documentation available at http://localhost:${port}/api/docs`);
   }
 
+  // Socket.IO Redis adapter for cross-instance event broadcasting
+  const redisIoAdapter = new RedisIoAdapter(app, configService);
+  await redisIoAdapter.connectToRedis();
+  app.useWebSocketAdapter(redisIoAdapter);
+
   await app.listen(port);
   logger.log(`Application is running on: http://localhost:${port}`);
   logger.log(`Environment: ${appEnv}`);

src/messages/messages.gateway.ts

@@ -16,6 +16,7 @@ import { ConnectionRequestsService } from '../connection-requests/connection-req
 import { PrismaService } from '../prisma/prisma.service';
 import { CreateMessageDto } from './dto';
 import { Logger } from '@nestjs/common';
+import { RedisPresenceService } from '../common/services/redis-presence.service';
 
 interface AuthenticatedSocket extends Socket {
   userId?: string;
@@ -44,6 +45,7 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
     private readonly prisma: PrismaService,
     private readonly jwtService: JwtService,
     private readonly configService: ConfigService,
+    private readonly redisPresence: RedisPresenceService,
   ) {
     // Periodic cleanup of userSocketMap and rateLimitMap every 5 minutes
     setInterval(() => {
@@ -112,6 +114,9 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
 
     this.logger.log(`Client connected successfully: ${client.id} (User: ${userId}, Role: ${payload.role})`);
 
+    // Join per-user room for Redis-backed cross-instance message routing
+    client.join(`user:${userId}`);
+
     // Everything below is best-effort — don't let DB errors kill the socket connection
     try {
       // Auto-join admin users to admin_room for support chat notifications
@@ -123,7 +128,11 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
       this.logger.warn(`Failed to join admin room for ${userId}: ${err.message}`);
     }
 
-    // Update user online status in background (don't await)
+    // Update user online status in Redis + DB (don't await)
+    this.redisPresence.setOnline(userId).catch((err) => {
+      this.logger.warn(`Failed to set Redis presence for ${userId}: ${err.message}`);
+    });
+
     this.messagesService.updateOnlineStatus(userId, true)
       .then(() => {
         this.broadcastUserStatus(userId, true);
@@ -158,13 +167,14 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
    */
   async handleDisconnect(client: AuthenticatedSocket) {
     if (client.userId) {
-      // Remove socket from tracking
+      // Remove socket from local tracking
       const sockets = this.userSocketMap.get(client.userId);
       if (sockets) {
         sockets.delete(client.id);
-        // Only mark offline if no more sockets connected
+        // Only mark offline if no more local sockets connected
         if (sockets.size === 0) {
           this.userSocketMap.delete(client.userId);
+          await this.redisPresence.setOffline(client.userId);
           await this.messagesService.updateOnlineStatus(client.userId, false);
           this.broadcastUserStatus(client.userId, false);
         }
@@ -245,7 +255,7 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
           this.sendToUser(otherUserId, 'new_message', message);
 
           // Auto-mark as DELIVERED if recipient is online
-          if (this.isUserOnline(otherUserId)) {
+          if (await this.isUserOnline(otherUserId)) {
             const deliveredAt = new Date().toISOString();
             await this.messagesService.markMessageDelivered(message.id);
             // Update message object so sender gets correct status in callback
@@ -502,25 +512,17 @@ export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect
   }
 
   /**
-   * Send message to a specific user (by userId)
+   * Send message to a specific user via per-user room (Redis-backed cross-instance)
    */
   sendToUser(userId: string, event: string, data: any) {
-    const sockets = this.userSocketMap.get(userId);
-    if (sockets && sockets.size > 0) {
-      sockets.forEach((socketId) => {
-        this.server.to(socketId).emit(event, data);
-      });
-      this.logger.debug(`Sent ${event} to user ${userId} via ${sockets.size} socket(s)`);
-    } else {
-      this.logger.debug(`User ${userId} has no active sockets for ${event} - message saved in DB`);
-    }
+    this.server.to(`user:${userId}`).emit(event, data);
+    this.logger.debug(`Sent ${event} to user:${userId} room`);
   }
 
   /**
-   * Check if a user is online
+   * Check if a user is online (Redis-backed)
    */
-  isUserOnline(userId: string): boolean {
-    const sockets = this.userSocketMap.get(userId);
-    return sockets ? sockets.size > 0 : false;
+  async isUserOnline(userId: string): Promise<boolean> {
+    return this.redisPresence.isOnline(userId);
   }
 }

src/messages/messages.module.ts

@@ -7,6 +7,7 @@ import { MessagesGateway } from './messages.gateway';
 import { PrismaModule } from '../prisma/prisma.module';
 import { SupportChatModule } from '../support-chat/support-chat.module';
 import { ConnectionRequestsModule } from '../connection-requests/connection-requests.module';
+import { RedisPresenceService } from '../common/services/redis-presence.service';
 
 @Module({
   imports: [
@@ -26,7 +27,7 @@ import { ConnectionRequestsModule } from '../connection-requests/connection-requ
     }),
   ],
   controllers: [MessagesController],
-  providers: [MessagesService, MessagesGateway],
-  exports: [MessagesService, MessagesGateway],
+  providers: [MessagesService, MessagesGateway, RedisPresenceService],
+  exports: [MessagesService, MessagesGateway, RedisPresenceService],
 })
 export class MessagesModule {}