From ec28b21cf81ff49421db46826d6e24f69559249c Mon Sep 17 00:00:00 2001 From: AmirHossein Mahmoodi Date: Sat, 10 May 2025 15:04:19 +0330 Subject: [PATCH] change stracture --- src/app.module.ts | 2 + src/redis/redis.module.ts | 8 ++++ src/redis/redis.service.ts | 20 ++++++++++ src/socket/dto/clientConnection.dto.ts | 3 +- src/socket/socket.entity.ts | 15 ++++--- src/socket/socket.gateway.ts | 27 +++++++++---- src/socket/socket.module.ts | 7 +++- src/socket/socket.service.ts | 55 +++++++++++++++++++++----- 8 files changed, 112 insertions(+), 25 deletions(-) create mode 100644 src/redis/redis.module.ts create mode 100644 src/redis/redis.service.ts diff --git a/src/app.module.ts b/src/app.module.ts index 0f436b9..c6136f4 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -3,6 +3,7 @@ import { ConfigModule } from '@nestjs/config'; import { TypeOrmModule } from '@nestjs/typeorm'; import { SocketModule } from './socket/socket.module'; import { NotificationModule } from './notifications/notification.module'; +import { RedisModule } from './redis/redis.module'; @Module({ imports: [ @@ -20,6 +21,7 @@ import { NotificationModule } from './notifications/notification.module'; }), SocketModule, NotificationModule, + RedisModule, ], }) export class AppModule {} diff --git a/src/redis/redis.module.ts b/src/redis/redis.module.ts new file mode 100644 index 0000000..b495868 --- /dev/null +++ b/src/redis/redis.module.ts @@ -0,0 +1,8 @@ +import { Module } from '@nestjs/common'; +import { RedisService } from './redis.service'; + +@Module({ + providers: [RedisService], + exports: [RedisService], +}) +export class RedisModule {} diff --git a/src/redis/redis.service.ts b/src/redis/redis.service.ts new file mode 100644 index 0000000..5365662 --- /dev/null +++ b/src/redis/redis.service.ts @@ -0,0 +1,20 @@ +import { Injectable, OnModuleInit } from '@nestjs/common'; +import { createClient } from 'redis'; + +@Injectable() +export class RedisService implements OnModuleInit { + private client; + + async onModuleInit() { + this.client = createClient({ + url: `redis://${process.env.REDIS_HOST}:${process.env.REDIS_PORT}`, + }); + + this.client.on('error', (err) => console.error('Redis Error:', err)); + await this.client.connect(); + } + + getClient() { + return this.client; + } +} diff --git a/src/socket/dto/clientConnection.dto.ts b/src/socket/dto/clientConnection.dto.ts index 077284e..49fdd73 100644 --- a/src/socket/dto/clientConnection.dto.ts +++ b/src/socket/dto/clientConnection.dto.ts @@ -1,5 +1,6 @@ export class ClientConnectionDto { - client_id: string; + user_id: number; + telephone_id: string; socket_id: string; event: string; description?: string; diff --git a/src/socket/socket.entity.ts b/src/socket/socket.entity.ts index 2820689..9c2f2a4 100644 --- a/src/socket/socket.entity.ts +++ b/src/socket/socket.entity.ts @@ -1,14 +1,14 @@ -import { Column, Entity, PrimaryGeneratedColumn } from 'typeorm'; +import { Column, Entity, PrimaryColumn } from 'typeorm'; @Entity() export class SocketConnection { - @PrimaryGeneratedColumn() - id: number; + @PrimaryColumn() + user_id: number; - @Column() - client_id: string; + @PrimaryColumn() + telephone_id: string; - @Column() + @PrimaryColumn() socket_id: string; @Column() @@ -16,4 +16,7 @@ export class SocketConnection { @Column({ nullable: true }) description: string; + + @Column({ type: 'timestamp', default: () => 'CURRENT_TIMESTAMP' }) + created_at: Date; } diff --git a/src/socket/socket.gateway.ts b/src/socket/socket.gateway.ts index 8bca864..ce9994f 100644 --- a/src/socket/socket.gateway.ts +++ b/src/socket/socket.gateway.ts @@ -6,7 +6,7 @@ import { } from '@nestjs/websockets'; import { Server, Socket } from 'socket.io'; import { SendMessageDto } from './dto/sendMessage.dto'; -import { SocketService } from './socket.service'; +import { ClientToken, SocketService } from './socket.service'; @WebSocketGateway({ cors: { @@ -21,27 +21,38 @@ export class SocketGateway implements OnGatewayConnection { constructor(private socketService: SocketService) {} async handleConnection(client: Socket) { - const clientId = client.handshake.auth.token; + const token: ClientToken = client.handshake.auth.token; + + if (!this.socketService.isValidToken(token)) { + client.emit( + 'error', + 'Invalid token: user_id or telephone_id missing or malformed.', + ); + client.disconnect(true); + return; + } await this.socketService.clientConnection({ - client_id: JSON.stringify(clientId), + user_id: token.user_id, + telephone_id: token.telephone_id, socket_id: client.id, event: 'connected', }); - this.socketService.setOnlineClient(clientId, client.id); + await this.socketService.addOnlineClient(token, client.id); this.broadcastOnlineClients(); client.on('disconnect', async (reason) => { await this.socketService.clientConnection({ - client_id: JSON.stringify(clientId), + user_id: token.user_id, + telephone_id: token.telephone_id, socket_id: client.id, event: 'disconnected', description: reason, }); - this.socketService.removeOnlineClient(clientId); + await this.socketService.removeOnlineClient(token, client.id); this.broadcastOnlineClients(); }); @@ -66,8 +77,8 @@ export class SocketGateway implements OnGatewayConnection { return result; } - broadcastOnlineClients() { - const onlineClients = this.socketService.getOnlineClients(); + async broadcastOnlineClients() { + const onlineClients = await this.socketService.getAllOnlineClients(); this.server.emit('onlineClientsUpdated', { clients: onlineClients, diff --git a/src/socket/socket.module.ts b/src/socket/socket.module.ts index aec9078..6b613cd 100644 --- a/src/socket/socket.module.ts +++ b/src/socket/socket.module.ts @@ -4,9 +4,14 @@ import { SocketConnection } from './socket.entity'; import { SocketGateway } from './socket.gateway'; import { SocketService } from './socket.service'; import { HttpModule } from '@nestjs/axios'; +import { RedisModule } from 'src/redis/redis.module'; @Module({ - imports: [TypeOrmModule.forFeature([SocketConnection]), HttpModule], + imports: [ + TypeOrmModule.forFeature([SocketConnection]), + HttpModule, + RedisModule, + ], providers: [SocketGateway, SocketService], exports: [SocketGateway], }) diff --git a/src/socket/socket.service.ts b/src/socket/socket.service.ts index 187ac75..7d1d307 100644 --- a/src/socket/socket.service.ts +++ b/src/socket/socket.service.ts @@ -6,18 +6,28 @@ import { Socket } from 'socket.io'; import { Repository } from 'typeorm'; import { ClientConnectionDto } from './dto/clientConnection.dto'; import { SocketConnection } from './socket.entity'; +import { RedisService } from 'src/redis/redis.service'; + +export interface ClientToken { + user_id: number; + telephone_id: string; +} @Injectable() export class SocketService { private readonly logger = new Logger(SocketService.name); - private readonly onlineClients = new Map(); constructor( @InjectRepository(SocketConnection) private socketConnectionRepository: Repository, private readonly httpService: HttpService, + private readonly redisService: RedisService, ) {} + private getRedisKey(token: ClientToken): string { + return `online_clients:${token.user_id}-${token.telephone_id}`; + } + async clientConnection(data: ClientConnectionDto) { const client = this.socketConnectionRepository.create(data); @@ -29,14 +39,14 @@ export class SocketService { const endpoint = eventEndpoints[data.event]; if (endpoint) { - const url = `${process.env.SERVICE_URL}${endpoint}?telephone_id=${client.client_id}`; + const url = `${process.env.SERVICE_URL}${endpoint}?telephone_id=${client.telephone_id}`; try { await firstValueFrom( this.httpService.get(url, { timeout: 3000 }).pipe(retry(3)), ); } catch (error) { this.logger.error( - `❌ Failed to notify ${data.event} for client ${client.client_id}:`, + `❌ Failed to notify ${data.event} for client ${client.telephone_id}:`, error.message || error, ); } @@ -77,15 +87,42 @@ export class SocketService { }; } - setOnlineClient(clientId: string, socketId: string) { - this.onlineClients.set(clientId, socketId); + getClientKey(token: ClientToken): string { + return `${token.user_id}-${token.telephone_id}`; } - removeOnlineClient(clientId: string) { - this.onlineClients.delete(clientId); + async addOnlineClient(token: any, socketId: string) { + const client = this.redisService.getClient(); + const key = this.getRedisKey(token); + await client.sAdd(key, socketId); } - getOnlineClients(): string[] { - return [...this.onlineClients.keys()]; + async removeOnlineClient(token: any, socketId: string) { + const client = this.redisService.getClient(); + const key = this.getRedisKey(token); + await client.sRem(key, socketId); + const count = await client.sCard(key); + if (count === 0) { + await client.del(key); + } + } + + async getAllOnlineClients(): Promise { + const client = this.redisService.getClient(); + const keys = await client.keys('online_clients:*'); + return keys.map((key) => { + const [user_id, telephone_id] = key.split(':')[1].split('-').map(Number); + return { user_id, telephone_id }; + }); + } + + isValidToken(token: any): token is ClientToken { + return ( + token && + typeof token.user_id === 'number' && + typeof token.telephone_id === 'string' && + !isNaN(token.user_id) && + !isNaN(token.telephone_id) + ); } }