diff --git a/package-lock.json b/package-lock.json index 8b08c71..f775147 100644 --- a/package-lock.json +++ b/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", diff --git a/package.json b/package.json index 93d414e..457ba09 100644 --- a/package.json +++ b/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", diff --git a/src/app.module.ts b/src/app.module.ts index 7405927..e23a52b 100644 --- a/src/app.module.ts +++ b/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([ { diff --git a/src/common/adapters/redis-io.adapter.ts b/src/common/adapters/redis-io.adapter.ts new file mode 100644 index 0000000..c02b3c5 --- /dev/null +++ b/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; + + constructor( + app: INestApplicationContext, + private readonly configService: ConfigService, + ) { + super(app); + } + + async connectToRedis(): Promise { + const host = this.configService.get('redis.host', 'localhost'); + const port = this.configService.get('redis.port', 6379); + const password = this.configService.get('redis.password') || undefined; + const db = this.configService.get('redis.db', 0); + + const pubClient = new Redis({ host, port, password, db }); + const subClient = new Redis({ host, port, password, db }); + + await Promise.all([ + new Promise((resolve) => pubClient.on('connect', resolve)), + new Promise((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; + } +} diff --git a/src/common/redis/redis.module.ts b/src/common/redis/redis.module.ts new file mode 100644 index 0000000..6c84c96 --- /dev/null +++ b/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 {} diff --git a/src/common/redis/redis.provider.ts b/src/common/redis/redis.provider.ts new file mode 100644 index 0000000..a5e834f --- /dev/null +++ b/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('redis.host', 'localhost'), + port: configService.get('redis.port', 6379), + password: configService.get('redis.password') || undefined, + db: configService.get('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], +}; diff --git a/src/common/services/redis-presence.service.ts b/src/common/services/redis-presence.service.ts new file mode 100644 index 0000000..cea68e1 --- /dev/null +++ b/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 { + await this.redis.hset(PRESENCE_KEY, userId, new Date().toISOString()); + } + + async setOffline(userId: string): Promise { + await this.redis.hdel(PRESENCE_KEY, userId); + } + + async isOnline(userId: string): Promise { + const exists = await this.redis.hexists(PRESENCE_KEY, userId); + return exists === 1; + } + + async getOnlineUsers(userIds: string[]): Promise> { + if (userIds.length === 0) return new Set(); + const results = await this.redis.hmget(PRESENCE_KEY, ...userIds); + const online = new Set(); + userIds.forEach((id, i) => { + if (results[i] !== null) online.add(id); + }); + return online; + } + + async clearAll(): Promise { + await this.redis.del(PRESENCE_KEY); + } +} diff --git a/src/main.ts b/src/main.ts index 769a564..74123d2 100644 --- a/src/main.ts +++ b/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}`); diff --git a/src/messages/messages.gateway.ts b/src/messages/messages.gateway.ts index 28baf4b..0a46753 100644 --- a/src/messages/messages.gateway.ts +++ b/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 { + return this.redisPresence.isOnline(userId); } } diff --git a/src/messages/messages.module.ts b/src/messages/messages.module.ts index dbabddb..09e26ea 100644 --- a/src/messages/messages.module.ts +++ b/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 {}