From 3b1a7b999e5d3cd60fba297863d1edd215fa2a58 Mon Sep 17 00:00:00 2001 From: pradeepkumar Date: Sun, 8 Feb 2026 22:44:16 +0530 Subject: [PATCH] feat: Implement real-time messaging module with REST API endpoints and WebSocket gateway. --- package-lock.json | 45 +- package.json | 6 +- prisma/schema.prisma | 83 +++ src/app.module.ts | 2 + src/auth/guards/jwt-auth.guard.ts | 5 + src/auth/guards/roles.guard.ts | 5 + src/messages/dto/create-message.dto.ts | 35 ++ src/messages/dto/index.ts | 2 + src/messages/dto/start-conversation.dto.ts | 10 + src/messages/index.ts | 5 + src/messages/messages.controller.ts | 218 ++++++++ src/messages/messages.gateway.ts | 278 ++++++++++ src/messages/messages.module.ts | 28 + src/messages/messages.service.ts | 586 +++++++++++++++++++++ 14 files changed, 1274 insertions(+), 34 deletions(-) create mode 100644 src/messages/dto/create-message.dto.ts create mode 100644 src/messages/dto/index.ts create mode 100644 src/messages/dto/start-conversation.dto.ts create mode 100644 src/messages/index.ts create mode 100644 src/messages/messages.controller.ts create mode 100644 src/messages/messages.gateway.ts create mode 100644 src/messages/messages.module.ts create mode 100644 src/messages/messages.service.ts diff --git a/package-lock.json b/package-lock.json index 4a34ec4..9fe0926 100644 --- a/package-lock.json +++ b/package-lock.json @@ -24,12 +24,12 @@ "@nestjs/jwt": "^11.0.2", "@nestjs/passport": "^11.0.5", "@nestjs/platform-express": "^11.0.1", - "@nestjs/platform-socket.io": "^11.1.9", + "@nestjs/platform-socket.io": "^11.1.13", "@nestjs/schedule": "^6.1.0", "@nestjs/swagger": "^11.2.3", "@nestjs/terminus": "^11.0.0", "@nestjs/throttler": "^6.5.0", - "@nestjs/websockets": "^11.1.9", + "@nestjs/websockets": "^11.1.13", "@prisma/adapter-pg": "^7.2.0", "@prisma/client": "^7.2.0", "@sendgrid/mail": "^8.1.6", @@ -70,7 +70,7 @@ "rxjs": "^7.8.1", "sharp": "^0.34.5", "slugify": "^1.6.6", - "socket.io": "^4.8.1", + "socket.io": "^4.8.3", "speakeasy": "^2.0.0", "stripe": "^20.1.0", "uuid": "^13.0.0", @@ -4822,12 +4822,12 @@ } }, "node_modules/@nestjs/platform-socket.io": { - "version": "11.1.9", - "resolved": "https://registry.npmjs.org/@nestjs/platform-socket.io/-/platform-socket.io-11.1.9.tgz", - "integrity": "sha512-OaAW+voXo5BXbFKd9Ot3SL05tEucRMhZRdw5wdWZf/RpIl9hB6G6OHr8DDxNbUGvuQWzNnZHCDHx3EQJzjcIyA==", + "version": "11.1.13", + "resolved": "https://registry.npmjs.org/@nestjs/platform-socket.io/-/platform-socket.io-11.1.13.tgz", + "integrity": "sha512-04Rh16IopZzHRXt0ZjFASqt9oNFV/0m0NsYe4kVOSaTEoef3cH7cTFpNpHsfNHcc4QpYL963XE8SvIRcZs5L8A==", "license": "MIT", "dependencies": { - "socket.io": "4.8.1", + "socket.io": "4.8.3", "tslib": "2.8.1" }, "funding": { @@ -5094,9 +5094,9 @@ } }, "node_modules/@nestjs/websockets": { - "version": "11.1.9", - "resolved": "https://registry.npmjs.org/@nestjs/websockets/-/websockets-11.1.9.tgz", - "integrity": "sha512-kkkdeTVcc3X7ZzvVqUVpOAJoh49kTRUjWNUXo5jmG+27OvZoHfs/vuSiqxidrrbIgydSqN15HUsf1wZwQUrxCQ==", + "version": "11.1.13", + "resolved": "https://registry.npmjs.org/@nestjs/websockets/-/websockets-11.1.13.tgz", + "integrity": "sha512-8r8EadqBkrTYtH2uog42HfIb5fcP5a3iXymH/ityd9bO/gDson5Q1qbtCQRjuU++6NY12YYteKRu4eP/iErbLw==", "license": "MIT", "dependencies": { "iterare": "1.2.1", @@ -17492,15 +17492,15 @@ } }, "node_modules/socket.io": { - "version": "4.8.1", - "resolved": "https://registry.npmjs.org/socket.io/-/socket.io-4.8.1.tgz", - "integrity": "sha512-oZ7iUCxph8WYRHHcjBEc9unw3adt5CmSNlppj/5Q4k2RIrhl8Z5yY2Xr4j9zj0+wzVZ0bxmYoGSzKJnRl6A4yg==", + "version": "4.8.3", + "resolved": "https://registry.npmjs.org/socket.io/-/socket.io-4.8.3.tgz", + "integrity": "sha512-2Dd78bqzzjE6KPkD5fHZmDAKRNe3J15q+YHDrIsy9WEkqttc7GY+kT9OBLSMaPbQaEd0x1BjcmtMtXkfpc+T5A==", "license": "MIT", "dependencies": { "accepts": "~1.3.4", "base64id": "~2.0.0", "cors": "~2.8.5", - "debug": "~4.3.2", + "debug": "~4.4.1", "engine.io": "~6.6.0", "socket.io-adapter": "~2.5.2", "socket.io-parser": "~4.2.4" @@ -17579,23 +17579,6 @@ "node": ">= 0.6" } }, - "node_modules/socket.io/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/node_modules/mime-db": { "version": "1.52.0", "resolved": "https://registry.npmjs.org/mime-db/-/mime-db-1.52.0.tgz", diff --git a/package.json b/package.json index 091c02b..aacf45c 100644 --- a/package.json +++ b/package.json @@ -42,12 +42,12 @@ "@nestjs/jwt": "^11.0.2", "@nestjs/passport": "^11.0.5", "@nestjs/platform-express": "^11.0.1", - "@nestjs/platform-socket.io": "^11.1.9", + "@nestjs/platform-socket.io": "^11.1.13", "@nestjs/schedule": "^6.1.0", "@nestjs/swagger": "^11.2.3", "@nestjs/terminus": "^11.0.0", "@nestjs/throttler": "^6.5.0", - "@nestjs/websockets": "^11.1.9", + "@nestjs/websockets": "^11.1.13", "@prisma/adapter-pg": "^7.2.0", "@prisma/client": "^7.2.0", "@sendgrid/mail": "^8.1.6", @@ -88,7 +88,7 @@ "rxjs": "^7.8.1", "sharp": "^0.34.5", "slugify": "^1.6.6", - "socket.io": "^4.8.1", + "socket.io": "^4.8.3", "speakeasy": "^2.0.0", "stripe": "^20.1.0", "uuid": "^13.0.0", diff --git a/prisma/schema.prisma b/prisma/schema.prisma index abffa5a..f462bd7 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -62,6 +62,19 @@ enum ConnectionStatus { REJECTED } +enum MessageType { + TEXT + FILE + IMAGE + SYSTEM // For system messages like "connection accepted" +} + +enum MessageStatus { + SENT + DELIVERED + READ +} + // =========================================== // USER - Authentication Only @@ -94,6 +107,10 @@ model User { updatedAt DateTime @updatedAt lastLoginAt DateTime? + // Online Status (for messaging) + isOnline Boolean @default(false) + lastSeenAt DateTime? + // Relations - Profile based on role userProfile UserProfile? agentProfile AgentProfile? @@ -104,9 +121,14 @@ model User { // Connection Requests connectionRequests ConnectionRequest[] + // Messaging Relations + conversations Conversation[] + messages Message[] + @@index([email]) @@index([role]) @@index([status]) + @@index([isOnline]) @@map("users") } @@ -204,6 +226,7 @@ model AgentProfile { agentType AgentType? @relation(fields: [agentTypeId], references: [id]) fieldValues AgentProfileFieldValue[] connectionRequests ConnectionRequest[] + conversations Conversation[] @@index([userId]) @@index([agentTypeId]) @@ -409,3 +432,63 @@ model ConnectionRequest { @@index([status]) @@map("connection_requests") } + +// =========================================== +// MESSAGING SYSTEM +// =========================================== + +model Conversation { + id String @id @default(uuid()) + userId String // Regular user in the conversation + agentProfileId String // Agent in the conversation + lastMessageAt DateTime? + lastMessageText String? @db.VarChar(255) + userUnreadCount Int @default(0) // Unread count for the user + agentUnreadCount Int @default(0) // Unread count for the agent + + // Timestamps + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + // Relations + user User @relation(fields: [userId], references: [id], onDelete: Cascade) + agentProfile AgentProfile @relation(fields: [agentProfileId], references: [id], onDelete: Cascade) + messages Message[] + + @@unique([userId, agentProfileId]) // One conversation per user-agent pair + @@index([userId, lastMessageAt]) + @@index([agentProfileId, lastMessageAt]) + @@map("conversations") +} + +model Message { + id String @id @default(uuid()) + conversationId String + senderId String // User ID of the sender (can be user or agent's user) + content String @db.Text + messageType MessageType @default(TEXT) + + // File attachment fields + fileUrl String? + fileName String? + fileSize Int? // File size in bytes + mimeType String? + + // Message status + status MessageStatus @default(SENT) + deliveredAt DateTime? + readAt DateTime? + + // Timestamps + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + // Relations + conversation Conversation @relation(fields: [conversationId], references: [id], onDelete: Cascade) + sender User @relation(fields: [senderId], references: [id], onDelete: Cascade) + + @@index([conversationId, createdAt]) + @@index([senderId]) + @@index([status]) + @@map("messages") +} diff --git a/src/app.module.ts b/src/app.module.ts index 8cdc2c5..9eeca47 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -15,6 +15,7 @@ import { AgentsModule } from './agents'; import { UploadModule } from './upload'; import { ProfileFieldsModule } from './profile-fields'; import { ConnectionRequestsModule } from './connection-requests'; +import { MessagesModule } from './messages'; import { AppController } from './app.controller'; import { AppService } from './app.service'; @@ -53,6 +54,7 @@ import { AppService } from './app.service'; UploadModule, ProfileFieldsModule, ConnectionRequestsModule, + MessagesModule, ], controllers: [AppController], providers: [ diff --git a/src/auth/guards/jwt-auth.guard.ts b/src/auth/guards/jwt-auth.guard.ts index 3bc9480..2a7cdf6 100644 --- a/src/auth/guards/jwt-auth.guard.ts +++ b/src/auth/guards/jwt-auth.guard.ts @@ -14,6 +14,11 @@ export class JwtAuthGuard extends AuthGuard('jwt') { } canActivate(context: ExecutionContext) { + // Skip WebSocket connections - they have their own auth in the gateway + if (context.getType() === 'ws') { + return true; + } + // Check if route is marked as public const isPublic = this.reflector.getAllAndOverride(IS_PUBLIC_KEY, [ context.getHandler(), diff --git a/src/auth/guards/roles.guard.ts b/src/auth/guards/roles.guard.ts index 4158696..71ec958 100644 --- a/src/auth/guards/roles.guard.ts +++ b/src/auth/guards/roles.guard.ts @@ -13,6 +13,11 @@ export class RolesGuard implements CanActivate { constructor(private reflector: Reflector) {} canActivate(context: ExecutionContext): boolean { + // Skip WebSocket connections - they have their own auth in the gateway + if (context.getType() === 'ws') { + return true; + } + const requiredRoles = this.reflector.getAllAndOverride( ROLES_KEY, [context.getHandler(), context.getClass()], diff --git a/src/messages/dto/create-message.dto.ts b/src/messages/dto/create-message.dto.ts new file mode 100644 index 0000000..5627528 --- /dev/null +++ b/src/messages/dto/create-message.dto.ts @@ -0,0 +1,35 @@ +import { IsString, IsNotEmpty, IsOptional, IsEnum, MaxLength } from 'class-validator'; +import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; +import { MessageType } from '@prisma/client'; + +export class CreateMessageDto { + @ApiProperty({ description: 'Message content', maxLength: 5000 }) + @IsString() + @IsNotEmpty() + @MaxLength(5000) + content: string; + + @ApiPropertyOptional({ description: 'Message type', enum: MessageType, default: MessageType.TEXT }) + @IsOptional() + @IsEnum(MessageType) + messageType?: MessageType = MessageType.TEXT; + + @ApiPropertyOptional({ description: 'File URL for file/image messages' }) + @IsOptional() + @IsString() + fileUrl?: string; + + @ApiPropertyOptional({ description: 'Original file name' }) + @IsOptional() + @IsString() + fileName?: string; + + @ApiPropertyOptional({ description: 'File size in bytes' }) + @IsOptional() + fileSize?: number; + + @ApiPropertyOptional({ description: 'MIME type of the file' }) + @IsOptional() + @IsString() + mimeType?: string; +} diff --git a/src/messages/dto/index.ts b/src/messages/dto/index.ts new file mode 100644 index 0000000..39218d2 --- /dev/null +++ b/src/messages/dto/index.ts @@ -0,0 +1,2 @@ +export * from './create-message.dto'; +export * from './start-conversation.dto'; diff --git a/src/messages/dto/start-conversation.dto.ts b/src/messages/dto/start-conversation.dto.ts new file mode 100644 index 0000000..cbe322d --- /dev/null +++ b/src/messages/dto/start-conversation.dto.ts @@ -0,0 +1,10 @@ +import { IsString, IsNotEmpty, IsUUID } from 'class-validator'; +import { ApiProperty } from '@nestjs/swagger'; + +export class StartConversationDto { + @ApiProperty({ description: 'Agent profile ID to start conversation with' }) + @IsString() + @IsNotEmpty() + @IsUUID() + agentProfileId: string; +} diff --git a/src/messages/index.ts b/src/messages/index.ts new file mode 100644 index 0000000..ecda4ad --- /dev/null +++ b/src/messages/index.ts @@ -0,0 +1,5 @@ +export * from './messages.module'; +export * from './messages.service'; +export * from './messages.controller'; +export * from './messages.gateway'; +export * from './dto'; diff --git a/src/messages/messages.controller.ts b/src/messages/messages.controller.ts new file mode 100644 index 0000000..bb327d7 --- /dev/null +++ b/src/messages/messages.controller.ts @@ -0,0 +1,218 @@ +import { + Controller, + Get, + Post, + Patch, + Body, + Param, + Query, + UseGuards, + HttpCode, + HttpStatus, +} from '@nestjs/common'; +import { + ApiTags, + ApiOperation, + ApiResponse, + ApiBearerAuth, + ApiQuery, + ApiParam, +} from '@nestjs/swagger'; +import { MessagesService } from './messages.service'; +import { CreateMessageDto, StartConversationDto } from './dto'; +import { JwtAuthGuard } from '../auth/guards'; +import { CurrentUser } from '../auth/decorators'; +import { UserRole } from '@prisma/client'; + +@ApiTags('Messages') +@Controller('messages') +@UseGuards(JwtAuthGuard) +@ApiBearerAuth('JWT-auth') +export class MessagesController { + constructor(private readonly messagesService: MessagesService) {} + + // ========================================== + // GET CONVERSATIONS LIST + // ========================================== + @Get('conversations') + @ApiOperation({ summary: 'Get all conversations for the current user' }) + @ApiResponse({ + status: 200, + description: 'List of conversations', + schema: { + example: [ + { + id: 'conv-uuid', + lastMessageAt: '2024-01-01T00:00:00.000Z', + lastMessageText: 'Hello!', + unreadCount: 2, + otherParty: { + id: 'agent-uuid', + name: 'John Doe', + avatar: 'https://...', + headline: 'Real Estate Agent', + isOnline: true, + }, + }, + ], + }, + }) + async getConversations( + @CurrentUser('id') userId: string, + @CurrentUser('role') role: UserRole, + ) { + return this.messagesService.getConversations(userId, role); + } + + // ========================================== + // START OR GET CONVERSATION + // ========================================== + @Post('conversations') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Start or get existing conversation with an agent' }) + @ApiResponse({ + status: 200, + description: 'Conversation created or retrieved', + }) + @ApiResponse({ + status: 403, + description: 'Connection not accepted', + }) + async startConversation( + @CurrentUser('id') userId: string, + @Body() dto: StartConversationDto, + ) { + return this.messagesService.getOrCreateConversation(userId, dto.agentProfileId); + } + + // ========================================== + // GET CONVERSATION DETAILS + // ========================================== + @Get('conversations/:id') + @ApiOperation({ summary: 'Get a specific conversation' }) + @ApiParam({ name: 'id', description: 'Conversation ID' }) + @ApiResponse({ + status: 200, + description: 'Conversation details', + }) + @ApiResponse({ + status: 404, + description: 'Conversation not found', + }) + async getConversation( + @Param('id') conversationId: string, + @CurrentUser('id') userId: string, + ) { + return this.messagesService.getConversationById(conversationId, userId); + } + + // ========================================== + // GET MESSAGES FOR CONVERSATION + // ========================================== + @Get('conversations/:id/messages') + @ApiOperation({ summary: 'Get messages for a conversation' }) + @ApiParam({ name: 'id', description: 'Conversation ID' }) + @ApiQuery({ name: 'page', required: false, type: Number }) + @ApiQuery({ name: 'limit', required: false, type: Number }) + @ApiResponse({ + status: 200, + description: 'List of messages with pagination', + schema: { + example: { + messages: [ + { + id: 'msg-uuid', + content: 'Hello!', + messageType: 'TEXT', + status: 'READ', + createdAt: '2024-01-01T00:00:00.000Z', + sender: { + id: 'user-uuid', + role: 'USER', + userProfile: { + firstName: 'Jane', + lastName: 'Doe', + }, + }, + }, + ], + pagination: { + page: 1, + limit: 50, + total: 100, + pages: 2, + }, + }, + }, + }) + async getMessages( + @Param('id') conversationId: string, + @CurrentUser('id') userId: string, + @Query('page') page?: string, + @Query('limit') limit?: string, + ) { + return this.messagesService.getMessages( + conversationId, + userId, + page ? parseInt(page, 10) : 1, + limit ? parseInt(limit, 10) : 50, + ); + } + + // ========================================== + // SEND MESSAGE (REST FALLBACK) + // ========================================== + @Post('conversations/:id/messages') + @HttpCode(HttpStatus.CREATED) + @ApiOperation({ summary: 'Send a message (REST fallback - prefer WebSocket)' }) + @ApiParam({ name: 'id', description: 'Conversation ID' }) + @ApiResponse({ + status: 201, + description: 'Message created', + }) + async sendMessage( + @Param('id') conversationId: string, + @CurrentUser('id') userId: string, + @Body() dto: CreateMessageDto, + ) { + return this.messagesService.createMessage(conversationId, userId, dto); + } + + // ========================================== + // MARK MESSAGES AS READ + // ========================================== + @Patch('conversations/:id/read') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Mark all messages in conversation as read' }) + @ApiParam({ name: 'id', description: 'Conversation ID' }) + @ApiResponse({ + status: 200, + description: 'Messages marked as read', + }) + async markAsRead( + @Param('id') conversationId: string, + @CurrentUser('id') userId: string, + ) { + return this.messagesService.markMessagesAsRead(conversationId, userId); + } + + // ========================================== + // GET UNREAD COUNT + // ========================================== + @Get('unread-count') + @ApiOperation({ summary: 'Get total unread message count' }) + @ApiResponse({ + status: 200, + description: 'Unread count', + schema: { + example: { unreadCount: 5 }, + }, + }) + async getUnreadCount( + @CurrentUser('id') userId: string, + @CurrentUser('role') role: UserRole, + ) { + const unreadCount = await this.messagesService.getTotalUnreadCount(userId, role); + return { unreadCount }; + } +} diff --git a/src/messages/messages.gateway.ts b/src/messages/messages.gateway.ts new file mode 100644 index 0000000..4332ec6 --- /dev/null +++ b/src/messages/messages.gateway.ts @@ -0,0 +1,278 @@ +import { + WebSocketGateway, + WebSocketServer, + SubscribeMessage, + OnGatewayConnection, + OnGatewayDisconnect, + ConnectedSocket, + MessageBody, +} from '@nestjs/websockets'; +import { Server, Socket } from 'socket.io'; +import { JwtService } from '@nestjs/jwt'; +import { ConfigService } from '@nestjs/config'; +import { MessagesService } from './messages.service'; +import { CreateMessageDto } from './dto'; +import { Logger } from '@nestjs/common'; + +interface AuthenticatedSocket extends Socket { + userId?: string; + userRole?: string; +} + +@WebSocketGateway({ + cors: { + origin: '*', // Configure in production + credentials: true, + }, + transports: ['websocket', 'polling'], +}) +export class MessagesGateway implements OnGatewayConnection, OnGatewayDisconnect { + @WebSocketServer() + server: Server; + + private readonly logger = new Logger(MessagesGateway.name); + private userSocketMap = new Map>(); // userId -> Set of socketIds + + constructor( + private readonly messagesService: MessagesService, + private readonly jwtService: JwtService, + private readonly configService: ConfigService, + ) {} + + /** + * Handle new WebSocket connections + */ + async handleConnection(client: AuthenticatedSocket) { + this.logger.log(`New connection attempt: ${client.id}`); + + // Extract token from handshake + const token = + client.handshake.auth?.token || + client.handshake.headers?.authorization?.replace('Bearer ', ''); + + if (!token) { + this.logger.warn(`Connection rejected: No token provided for ${client.id}`); + client.disconnect(true); + return; + } + + // Verify JWT token + let payload; + try { + payload = await this.jwtService.verifyAsync(token, { + secret: this.configService.get('JWT_SECRET'), + }); + this.logger.log(`Token verified for ${client.id}, user: ${payload.sub}`); + } catch (jwtError) { + this.logger.warn(`Connection rejected: Invalid token for ${client.id} - ${jwtError.message}`); + client.disconnect(true); + return; + } + + // Attach user info to socket + const userId = payload.sub as string; + client.userId = userId; + client.userRole = payload.role; + + // Track socket connection + if (!this.userSocketMap.has(userId)) { + this.userSocketMap.set(userId, new Set()); + } + this.userSocketMap.get(userId)!.add(client.id); + + this.logger.log(`Client connected successfully: ${client.id} (User: ${userId})`); + + // Update user online status in background (don't await) + this.messagesService.updateOnlineStatus(userId, true) + .then(() => { + this.broadcastUserStatus(userId, true); + this.logger.log(`Online status updated for ${userId}`); + }) + .catch((err) => { + this.logger.warn(`Failed to update online status for ${userId}: ${err.message}`); + }); + } + + /** + * Handle WebSocket disconnections + */ + async handleDisconnect(client: AuthenticatedSocket) { + if (client.userId) { + // Remove socket from tracking + const sockets = this.userSocketMap.get(client.userId); + if (sockets) { + sockets.delete(client.id); + // Only mark offline if no more sockets connected + if (sockets.size === 0) { + this.userSocketMap.delete(client.userId); + await this.messagesService.updateOnlineStatus(client.userId, false); + this.broadcastUserStatus(client.userId, false); + } + } + + this.logger.log(`Client disconnected: ${client.id} (User: ${client.userId})`); + } + } + + /** + * Join a conversation room + */ + @SubscribeMessage('join_conversation') + async handleJoinConversation( + @ConnectedSocket() client: AuthenticatedSocket, + @MessageBody() data: { conversationId: string }, + ) { + if (!client.userId) { + return { error: 'Unauthorized' }; + } + + try { + // Verify user is part of the conversation + await this.messagesService.getConversationById(data.conversationId, client.userId); + + // Join the room + client.join(`conversation:${data.conversationId}`); + + this.logger.log(`User ${client.userId} joined conversation ${data.conversationId}`); + + return { success: true }; + } catch (error) { + return { error: error.message }; + } + } + + /** + * Leave a conversation room + */ + @SubscribeMessage('leave_conversation') + handleLeaveConversation( + @ConnectedSocket() client: AuthenticatedSocket, + @MessageBody() data: { conversationId: string }, + ) { + client.leave(`conversation:${data.conversationId}`); + this.logger.log(`User ${client.userId} left conversation ${data.conversationId}`); + return { success: true }; + } + + /** + * Send a message via WebSocket + */ + @SubscribeMessage('send_message') + async handleSendMessage( + @ConnectedSocket() client: AuthenticatedSocket, + @MessageBody() data: { conversationId: string; message: CreateMessageDto }, + ) { + if (!client.userId) { + return { error: 'Unauthorized' }; + } + + try { + const message = await this.messagesService.createMessage( + data.conversationId, + client.userId, + data.message, + ); + + // Broadcast to all clients in the conversation room + this.server + .to(`conversation:${data.conversationId}`) + .emit('new_message', message); + + return { success: true, message }; + } catch (error) { + return { error: error.message }; + } + } + + /** + * Handle typing indicator start + */ + @SubscribeMessage('typing_start') + handleTypingStart( + @ConnectedSocket() client: AuthenticatedSocket, + @MessageBody() data: { conversationId: string }, + ) { + if (!client.userId) return; + + // Broadcast to others in the room + client.to(`conversation:${data.conversationId}`).emit('typing_start', { + userId: client.userId, + conversationId: data.conversationId, + }); + } + + /** + * Handle typing indicator stop + */ + @SubscribeMessage('typing_stop') + handleTypingStop( + @ConnectedSocket() client: AuthenticatedSocket, + @MessageBody() data: { conversationId: string }, + ) { + if (!client.userId) return; + + client.to(`conversation:${data.conversationId}`).emit('typing_stop', { + userId: client.userId, + conversationId: data.conversationId, + }); + } + + /** + * Mark messages as read + */ + @SubscribeMessage('mark_read') + async handleMarkRead( + @ConnectedSocket() client: AuthenticatedSocket, + @MessageBody() data: { conversationId: string }, + ) { + if (!client.userId) { + return { error: 'Unauthorized' }; + } + + try { + await this.messagesService.markMessagesAsRead(data.conversationId, client.userId); + + // Notify the other party that messages have been read + client.to(`conversation:${data.conversationId}`).emit('messages_read', { + conversationId: data.conversationId, + readBy: client.userId, + readAt: new Date(), + }); + + return { success: true }; + } catch (error) { + return { error: error.message }; + } + } + + /** + * Broadcast user online/offline status + */ + private broadcastUserStatus(userId: string, isOnline: boolean) { + this.server.emit('user_status_change', { + userId, + isOnline, + lastSeenAt: isOnline ? null : new Date(), + }); + } + + /** + * Send message to a specific user (by userId) + */ + sendToUser(userId: string, event: string, data: any) { + const sockets = this.userSocketMap.get(userId); + if (sockets) { + sockets.forEach((socketId) => { + this.server.to(socketId).emit(event, data); + }); + } + } + + /** + * Check if a user is online + */ + isUserOnline(userId: string): boolean { + const sockets = this.userSocketMap.get(userId); + return sockets ? sockets.size > 0 : false; + } +} diff --git a/src/messages/messages.module.ts b/src/messages/messages.module.ts new file mode 100644 index 0000000..05a94be --- /dev/null +++ b/src/messages/messages.module.ts @@ -0,0 +1,28 @@ +import { Module } from '@nestjs/common'; +import { JwtModule } from '@nestjs/jwt'; +import { ConfigModule, ConfigService } from '@nestjs/config'; +import { MessagesController } from './messages.controller'; +import { MessagesService } from './messages.service'; +import { MessagesGateway } from './messages.gateway'; +import { PrismaModule } from '../prisma/prisma.module'; + +@Module({ + imports: [ + PrismaModule, + ConfigModule, + JwtModule.registerAsync({ + imports: [ConfigModule], + inject: [ConfigService], + useFactory: (configService: ConfigService) => ({ + secret: configService.get('JWT_SECRET') || 'fallback-secret', + signOptions: { + expiresIn: '15m', + }, + }), + }), + ], + controllers: [MessagesController], + providers: [MessagesService, MessagesGateway], + exports: [MessagesService, MessagesGateway], +}) +export class MessagesModule {} diff --git a/src/messages/messages.service.ts b/src/messages/messages.service.ts new file mode 100644 index 0000000..8786f22 --- /dev/null +++ b/src/messages/messages.service.ts @@ -0,0 +1,586 @@ +import { + Injectable, + NotFoundException, + ForbiddenException, + BadRequestException, +} from '@nestjs/common'; +import { PrismaService } from '../prisma/prisma.service'; +import { CreateMessageDto } from './dto'; +import { ConnectionStatus, MessageStatus, MessageType, UserRole } from '@prisma/client'; + +@Injectable() +export class MessagesService { + constructor(private readonly prisma: PrismaService) {} + + /** + * Validate if a user can message an agent (must have ACCEPTED connection) + */ + async validateCanMessage(userId: string, agentProfileId: string): Promise { + const connection = await this.prisma.connectionRequest.findUnique({ + where: { + userId_agentProfileId: { + userId, + agentProfileId, + }, + }, + }); + + return connection?.status === ConnectionStatus.ACCEPTED; + } + + /** + * Get or create a conversation between user and agent + */ + async getOrCreateConversation(userId: string, agentProfileId: string) { + // Validate connection first + const canMessage = await this.validateCanMessage(userId, agentProfileId); + if (!canMessage) { + throw new ForbiddenException( + 'You can only message agents with accepted connection requests', + ); + } + + // Check if conversation already exists + let conversation = await this.prisma.conversation.findUnique({ + where: { + userId_agentProfileId: { + userId, + agentProfileId, + }, + }, + include: { + agentProfile: { + select: { + id: true, + firstName: true, + lastName: true, + avatar: true, + headline: true, + userId: true, + user: { + select: { + isOnline: true, + lastSeenAt: true, + }, + }, + }, + }, + user: { + select: { + id: true, + isOnline: true, + lastSeenAt: true, + userProfile: { + select: { + firstName: true, + lastName: true, + avatar: true, + }, + }, + }, + }, + }, + }); + + // Create new conversation if it doesn't exist + if (!conversation) { + conversation = await this.prisma.conversation.create({ + data: { + userId, + agentProfileId, + }, + include: { + agentProfile: { + select: { + id: true, + firstName: true, + lastName: true, + avatar: true, + headline: true, + userId: true, + user: { + select: { + isOnline: true, + lastSeenAt: true, + }, + }, + }, + }, + user: { + select: { + id: true, + isOnline: true, + lastSeenAt: true, + userProfile: { + select: { + firstName: true, + lastName: true, + avatar: true, + }, + }, + }, + }, + }, + }); + } + + // Transform to include otherParty (for the user, other party is the agent) + return { + ...conversation, + unreadCount: conversation.userUnreadCount, + otherParty: { + id: conversation.agentProfile.id, + userId: conversation.agentProfile.userId, + name: `${conversation.agentProfile.firstName || ''} ${conversation.agentProfile.lastName || ''}`.trim() || 'Agent', + avatar: conversation.agentProfile.avatar, + headline: conversation.agentProfile.headline, + isOnline: conversation.agentProfile.user?.isOnline || false, + lastSeenAt: conversation.agentProfile.user?.lastSeenAt || null, + }, + }; + } + + /** + * Create a new message in a conversation + */ + async createMessage( + conversationId: string, + senderId: string, + dto: CreateMessageDto, + ) { + // Verify conversation exists and user is part of it + const conversation = await this.prisma.conversation.findUnique({ + where: { id: conversationId }, + include: { + agentProfile: { + select: { userId: true }, + }, + }, + }); + + if (!conversation) { + throw new NotFoundException('Conversation not found'); + } + + // Check if sender is part of the conversation + const isUser = conversation.userId === senderId; + const isAgent = conversation.agentProfile.userId === senderId; + + if (!isUser && !isAgent) { + throw new ForbiddenException('You are not part of this conversation'); + } + + // Create the message + const message = await this.prisma.message.create({ + data: { + conversationId, + senderId, + content: dto.content, + messageType: dto.messageType || MessageType.TEXT, + fileUrl: dto.fileUrl, + fileName: dto.fileName, + fileSize: dto.fileSize, + mimeType: dto.mimeType, + status: MessageStatus.SENT, + }, + include: { + sender: { + select: { + id: true, + role: true, + userProfile: { + select: { + firstName: true, + lastName: true, + avatar: true, + }, + }, + agentProfile: { + select: { + firstName: true, + lastName: true, + avatar: true, + }, + }, + }, + }, + }, + }); + + // Update conversation with last message info and unread counts + const truncatedContent = + dto.content.length > 255 ? dto.content.substring(0, 252) + '...' : dto.content; + + await this.prisma.conversation.update({ + where: { id: conversationId }, + data: { + lastMessageAt: new Date(), + lastMessageText: truncatedContent, + // Increment unread count for the other party + ...(isUser + ? { agentUnreadCount: { increment: 1 } } + : { userUnreadCount: { increment: 1 } }), + }, + }); + + return message; + } + + /** + * Get all conversations for a user (either as user or agent) + */ + async getConversations(userId: string, role: UserRole) { + let conversations; + + if (role === UserRole.AGENT) { + // Get agent profile for this user + const agentProfile = await this.prisma.agentProfile.findUnique({ + where: { userId }, + }); + + if (!agentProfile) { + return []; + } + + conversations = await this.prisma.conversation.findMany({ + where: { agentProfileId: agentProfile.id }, + include: { + user: { + select: { + id: true, + isOnline: true, + lastSeenAt: true, + userProfile: { + select: { + firstName: true, + lastName: true, + avatar: true, + }, + }, + }, + }, + messages: { + take: 1, + orderBy: { createdAt: 'desc' }, + }, + }, + orderBy: { lastMessageAt: 'desc' }, + }); + + // Map to include unread count from agent perspective + return conversations.map((conv) => ({ + ...conv, + unreadCount: conv.agentUnreadCount, + otherParty: { + id: conv.user.id, + name: `${conv.user.userProfile?.firstName || ''} ${conv.user.userProfile?.lastName || ''}`.trim() || 'User', + avatar: conv.user.userProfile?.avatar, + isOnline: conv.user.isOnline, + lastSeenAt: conv.user.lastSeenAt, + }, + })); + } else { + // Regular user + conversations = await this.prisma.conversation.findMany({ + where: { userId }, + include: { + agentProfile: { + select: { + id: true, + firstName: true, + lastName: true, + avatar: true, + headline: true, + userId: true, + user: { + select: { + isOnline: true, + lastSeenAt: true, + }, + }, + }, + }, + messages: { + take: 1, + orderBy: { createdAt: 'desc' }, + }, + }, + orderBy: { lastMessageAt: 'desc' }, + }); + + // Map to include unread count from user perspective + return conversations.map((conv) => ({ + ...conv, + unreadCount: conv.userUnreadCount, + otherParty: { + id: conv.agentProfile.id, + userId: conv.agentProfile.userId, + name: `${conv.agentProfile.firstName || ''} ${conv.agentProfile.lastName || ''}`.trim() || 'Agent', + avatar: conv.agentProfile.avatar, + headline: conv.agentProfile.headline, + isOnline: conv.agentProfile.user.isOnline, + lastSeenAt: conv.agentProfile.user.lastSeenAt, + }, + })); + } + } + + /** + * Get messages for a conversation with pagination + */ + async getMessages( + conversationId: string, + userId: string, + page: number = 1, + limit: number = 50, + ) { + // Verify user is part of the conversation + const conversation = await this.prisma.conversation.findUnique({ + where: { id: conversationId }, + include: { + agentProfile: { + select: { userId: true }, + }, + }, + }); + + if (!conversation) { + throw new NotFoundException('Conversation not found'); + } + + const isUser = conversation.userId === userId; + const isAgent = conversation.agentProfile.userId === userId; + + if (!isUser && !isAgent) { + throw new ForbiddenException('You are not part of this conversation'); + } + + const skip = (page - 1) * limit; + + const [messages, total] = await Promise.all([ + this.prisma.message.findMany({ + where: { conversationId }, + include: { + sender: { + select: { + id: true, + role: true, + userProfile: { + select: { + firstName: true, + lastName: true, + avatar: true, + }, + }, + agentProfile: { + select: { + firstName: true, + lastName: true, + avatar: true, + }, + }, + }, + }, + }, + orderBy: { createdAt: 'desc' }, + skip, + take: limit, + }), + this.prisma.message.count({ where: { conversationId } }), + ]); + + return { + messages: messages.reverse(), // Return in chronological order + pagination: { + page, + limit, + total, + pages: Math.ceil(total / limit), + }, + }; + } + + /** + * Mark messages as read in a conversation + */ + async markMessagesAsRead(conversationId: string, userId: string) { + const conversation = await this.prisma.conversation.findUnique({ + where: { id: conversationId }, + include: { + agentProfile: { + select: { userId: true }, + }, + }, + }); + + if (!conversation) { + throw new NotFoundException('Conversation not found'); + } + + const isUser = conversation.userId === userId; + const isAgent = conversation.agentProfile.userId === userId; + + if (!isUser && !isAgent) { + throw new ForbiddenException('You are not part of this conversation'); + } + + // Update all unread messages from the other party + await this.prisma.message.updateMany({ + where: { + conversationId, + senderId: { not: userId }, + status: { not: MessageStatus.READ }, + }, + data: { + status: MessageStatus.READ, + readAt: new Date(), + }, + }); + + // Reset unread count for this user + await this.prisma.conversation.update({ + where: { id: conversationId }, + data: isUser ? { userUnreadCount: 0 } : { agentUnreadCount: 0 }, + }); + + return { success: true }; + } + + /** + * Update user's online status + */ + async updateOnlineStatus(userId: string, isOnline: boolean) { + await this.prisma.user.update({ + where: { id: userId }, + data: { + isOnline, + lastSeenAt: isOnline ? undefined : new Date(), + }, + }); + } + + /** + * Mark a message as delivered + */ + async markMessageDelivered(messageId: string) { + return this.prisma.message.update({ + where: { id: messageId }, + data: { + status: MessageStatus.DELIVERED, + deliveredAt: new Date(), + }, + }); + } + + /** + * Get conversation by ID with full details + */ + async getConversationById(conversationId: string, userId: string) { + const conversation = await this.prisma.conversation.findUnique({ + where: { id: conversationId }, + include: { + agentProfile: { + select: { + id: true, + firstName: true, + lastName: true, + avatar: true, + headline: true, + userId: true, + user: { + select: { + isOnline: true, + lastSeenAt: true, + }, + }, + }, + }, + user: { + select: { + id: true, + isOnline: true, + lastSeenAt: true, + userProfile: { + select: { + firstName: true, + lastName: true, + avatar: true, + }, + }, + }, + }, + }, + }); + + if (!conversation) { + throw new NotFoundException('Conversation not found'); + } + + const isUser = conversation.userId === userId; + const isAgent = conversation.agentProfile.userId === userId; + + if (!isUser && !isAgent) { + throw new ForbiddenException('You are not part of this conversation'); + } + + // Transform to include otherParty based on who is requesting + if (isUser) { + // User is viewing - other party is agent + return { + ...conversation, + unreadCount: conversation.userUnreadCount, + otherParty: { + id: conversation.agentProfile.id, + userId: conversation.agentProfile.userId, + name: `${conversation.agentProfile.firstName || ''} ${conversation.agentProfile.lastName || ''}`.trim() || 'Agent', + avatar: conversation.agentProfile.avatar, + headline: conversation.agentProfile.headline, + isOnline: conversation.agentProfile.user?.isOnline || false, + lastSeenAt: conversation.agentProfile.user?.lastSeenAt || null, + }, + }; + } else { + // Agent is viewing - other party is user + return { + ...conversation, + unreadCount: conversation.agentUnreadCount, + otherParty: { + id: conversation.user.id, + name: `${conversation.user.userProfile?.firstName || ''} ${conversation.user.userProfile?.lastName || ''}`.trim() || 'User', + avatar: conversation.user.userProfile?.avatar, + isOnline: conversation.user.isOnline, + lastSeenAt: conversation.user.lastSeenAt, + }, + }; + } + } + + /** + * Get total unread count for a user across all conversations + */ + async getTotalUnreadCount(userId: string, role: UserRole) { + if (role === UserRole.AGENT) { + const agentProfile = await this.prisma.agentProfile.findUnique({ + where: { userId }, + }); + + if (!agentProfile) { + return 0; + } + + const result = await this.prisma.conversation.aggregate({ + where: { agentProfileId: agentProfile.id }, + _sum: { agentUnreadCount: true }, + }); + + return result._sum.agentUnreadCount || 0; + } else { + const result = await this.prisma.conversation.aggregate({ + where: { userId }, + _sum: { userUnreadCount: true }, + }); + + return result._sum.userUnreadCount || 0; + } + } +}