From 2d3560dc658fd2ab770c572f9ecf9defcb8c2d3c Mon Sep 17 00:00:00 2001 From: mahyargdz Date: Mon, 9 Jun 2025 11:12:17 +0330 Subject: [PATCH] chore: add search module --- src/app.module.ts | 2 + src/common/enums/message.enum.ts | 54 +- src/core/filters/ws-exception.filter.ts | 11 +- src/modules/chatbot/DTO/chat-response.dto.ts | 53 -- .../chatbot/DTO/session-id.param.dto.ts | 11 + src/modules/chatbot/chatbot.controller.ts | 134 ++--- src/modules/chatbot/chatbot.gateway.ts | 564 +++++++++++------- src/modules/chatbot/chatbot.module.ts | 8 +- .../chatbot/constants/chatbot.constants.ts | 107 ++-- .../chatbot/guards/websocket-auth.guard.ts | 15 +- .../chatbot/interfaces/websocket.interface.ts | 52 ++ .../chatbot/providers/chatbot.service.ts | 53 +- .../providers/websocket-auth.service.ts | 178 ++++++ src/modules/search/DTO/search.dto.ts | 9 + src/modules/search/search.controller.ts | 14 + src/modules/search/search.module.ts | 13 + src/modules/search/services/search.service.ts | 32 + .../utils/providers/extract-token.utils.ts | 31 +- 18 files changed, 910 insertions(+), 431 deletions(-) delete mode 100644 src/modules/chatbot/DTO/chat-response.dto.ts create mode 100644 src/modules/chatbot/DTO/session-id.param.dto.ts create mode 100644 src/modules/chatbot/interfaces/websocket.interface.ts create mode 100644 src/modules/chatbot/providers/websocket-auth.service.ts create mode 100644 src/modules/search/DTO/search.dto.ts create mode 100644 src/modules/search/search.controller.ts create mode 100644 src/modules/search/search.module.ts create mode 100644 src/modules/search/services/search.service.ts diff --git a/src/app.module.ts b/src/app.module.ts index d1962d8..c939022 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -24,6 +24,7 @@ import { IndustriesModule } from "./modules/industries/industries.module"; import { InvoicesModule } from "./modules/invoices/invoices.module"; import { NotificationModule } from "./modules/notifications/notifications.module"; import { PaymentsModule } from "./modules/payments/payments.module"; +import { SearchModule } from "./modules/search/search.module"; import { SlidersModule } from "./modules/sliders/sliders.module"; import { TicketsModule } from "./modules/tickets/tickets.module"; import { UploaderModule } from "./modules/uploader/uploader.module"; @@ -63,6 +64,7 @@ import { UtilsModule } from "./modules/utils/utils.module"; PaymentsModule, SlidersModule, ChatbotModule, + SearchModule, ], }) export class AppModule implements NestModule { diff --git a/src/common/enums/message.enum.ts b/src/common/enums/message.enum.ts index ff483b0..dacfd08 100755 --- a/src/common/enums/message.enum.ts +++ b/src/common/enums/message.enum.ts @@ -123,6 +123,8 @@ export const enum CommonMessage { UPDATED = "با موفقیت آپدیت شد", IS_ACTIVE_SHOULD_BE_1_0 = "وضعیت باید یکی از مقادیر ۰ و ۱ باشد", DATE_MUST_BE_DATE = "تاریخ باید یک تاریخ معتبر به صورت رشته باشد", + SESSION_ID_REQUIRED = "شناسه جلسه چت مورد نیاز است", + SESSION_ID_SHOULD_BE_UUID = "شناسه جلسه چت باید یک UUID باشد", } export const enum CategoryMessage { @@ -664,5 +666,55 @@ export const enum SliderMessage { LINK_MAX_LENGTH = "لینک اسلایدر باید حداکثر ۲۵۵ کاراکتر باشد", ORDER_NUMBER = "ترتیب اسلایدر باید یک عدد باشد", IS_ACTIVE_SHOULD_BE_A_BOOLEAN = "وضعیت فعال بودن اسلایدر باید یک مقدار بولی باشد", - SLIDER_NOT_FOUND = "SLIDER_NOT_FOUND", + SLIDER_NOT_FOUND = "اسلایدر یافت نشد", +} + +export const enum WebSocketMessage { + // Authentication errors + AUTHENTICATION_REQUIRED = "احراز هویت ضروری است", + INVALID_TOKEN = "توکن احراز هویت نامعتبر است", + TOKEN_EXPIRED = "توکن منقضی شده است", + UNAUTHORIZED_ACCESS = "دسترسی غیرمجاز", + USER_NOT_AUTHENTICATED = "کاربر احراز هویت نشده است", + + // Connection errors + CONNECTION_FAILED = "اتصال برقرار نشد", + WEBSOCKET_ERROR = "خطا در ارتباط WebSocket", + CONNECTION_TIMEOUT = "زمان اتصال به پایان رسید", + CONNECTION_REFUSED = "اتصال رد شد", + + // Session errors + SESSION_NOT_FOUND = "جلسه چت یافت نشد", + SESSION_CREATION_FAILED = "ایجاد جلسه چت با شکست مواجه شد", + SESSION_ACCESS_DENIED = "دسترسی به جلسه چت مجاز نیست", + SESSION_EXPIRED = "جلسه چت منقضی شده است", + SESSION_ALREADY_EXISTS = "جلسه چت قبلاً وجود دارد", + + // Message errors + MESSAGE_TOO_LONG = "پیام از حداکثر طول مجاز تجاوز کرده است", + MESSAGE_EMPTY = "پیام نمی‌تواند خالی باشد", + MESSAGE_SEND_FAILED = "ارسال پیام با شکست مواجه شد", + INVALID_MESSAGE_FORMAT = "فرمت پیام نامعتبر است", + + // Rate limiting + RATE_LIMIT_EXCEEDED = "تعداد پیام‌های ارسالی بیش از حد مجاز است", + TOO_MANY_CONNECTIONS = "تعداد اتصالات بیش از حد مجاز است", + + // Service errors + LLM_SERVICE_ERROR = "سرویس هوش مصنوعی موقتاً در دسترس نیست", + SERVICE_UNAVAILABLE = "سرویس موقتاً در دسترس نیست", + INTERNAL_SERVER_ERROR = "خطای داخلی سرور", + + // Validation errors + INVALID_SESSION_ID = "شناسه جلسه نامعتبر است", + INVALID_USER_ID = "شناسه کاربر نامعتبر است", + INVALID_MESSAGE_ID = "شناسه پیام نامعتبر است", + REQUIRED_FIELD_MISSING = "فیلد ضروری موجود نیست", + + // Success messages + AUTHENTICATED = "احراز هویت موفقیت‌آمیز", + SESSION_CREATED = "جلسه چت با موفقیت ایجاد شد", + CHAT_JOINED = "با موفقیت به چت متصل شدید", + MESSAGE_SENT = "پیام ارسال شد", + CONNECTION_ESTABLISHED = "اتصال برقرار شد", } diff --git a/src/core/filters/ws-exception.filter.ts b/src/core/filters/ws-exception.filter.ts index 5cb8c45..52d477a 100644 --- a/src/core/filters/ws-exception.filter.ts +++ b/src/core/filters/ws-exception.filter.ts @@ -2,7 +2,8 @@ import { ArgumentsHost, Catch, ExceptionFilter, Logger, UnauthorizedException } import { WsException } from "@nestjs/websockets"; import { Socket } from "socket.io"; -import { CHATBOT_CONSTANTS } from "../../modules/chatbot/constants/chatbot.constants"; +import { WebSocketMessage } from "../../common/enums/message.enum"; +import { WEBSOCKET_EVENTS } from "../../modules/chatbot/constants/chatbot.constants"; @Catch(WsException) export class WsExceptionFilter implements ExceptionFilter { @@ -17,8 +18,8 @@ export class WsExceptionFilter implements ExceptionFilter { this.logger.error(`Data: ${JSON.stringify(data)}`); if (exception instanceof UnauthorizedException) { - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.UNAUTHORIZED, { - message: CHATBOT_CONSTANTS.ERROR_MESSAGES.INVALID_TOKEN, + client.emit(WEBSOCKET_EVENTS.UNAUTHORIZED, { + message: WebSocketMessage.INVALID_TOKEN, }); client.disconnect(); return; @@ -26,9 +27,9 @@ export class WsExceptionFilter implements ExceptionFilter { const errorResponse = { status: "error", - message: exception.message || CHATBOT_CONSTANTS.ERROR_MESSAGES.WEBSOCKET_ERROR, + message: exception.message || WebSocketMessage.WEBSOCKET_ERROR, }; - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.WEBSOCKET_ERROR, errorResponse); + client.emit(WEBSOCKET_EVENTS.WEBSOCKET_ERROR, errorResponse); } } diff --git a/src/modules/chatbot/DTO/chat-response.dto.ts b/src/modules/chatbot/DTO/chat-response.dto.ts deleted file mode 100644 index c986731..0000000 --- a/src/modules/chatbot/DTO/chat-response.dto.ts +++ /dev/null @@ -1,53 +0,0 @@ -import { ApiProperty } from "@nestjs/swagger"; - -import { MessageStatus, MessageType } from "../entities/chat-message.entity"; -import { ChatSessionStatus } from "../entities/chat-session.entity"; - -export class ChatMessageResponseDto { - @ApiProperty() - id!: string; - - @ApiProperty() - content!: string; - - @ApiProperty({ enum: MessageType }) - type!: MessageType; - - @ApiProperty({ enum: MessageStatus }) - status!: MessageStatus; - - @ApiProperty() - createdAt!: Date; - - @ApiProperty({ required: false }) - responseToId?: string; - - @ApiProperty({ required: false }) - metadata?: Record; - - @ApiProperty() - tokensUsed!: number; -} - -export class ChatSessionResponseDto { - @ApiProperty() - id!: string; - - @ApiProperty() - title!: string; - - @ApiProperty({ enum: ChatSessionStatus }) - status!: ChatSessionStatus; - - @ApiProperty() - createdAt!: Date; - - @ApiProperty({ required: false }) - lastMessageAt?: Date; - - @ApiProperty({ type: [ChatMessageResponseDto] }) - messages!: ChatMessageResponseDto[]; - - @ApiProperty({ required: false }) - context?: Record; -} diff --git a/src/modules/chatbot/DTO/session-id.param.dto.ts b/src/modules/chatbot/DTO/session-id.param.dto.ts new file mode 100644 index 0000000..801e21c --- /dev/null +++ b/src/modules/chatbot/DTO/session-id.param.dto.ts @@ -0,0 +1,11 @@ +import { ApiProperty } from "@nestjs/swagger"; +import { IsNotEmpty, IsUUID } from "class-validator"; + +import { CommonMessage } from "../../../common/enums/message.enum"; + +export class SessionIdParamDto { + @IsNotEmpty({ message: CommonMessage.SESSION_ID_REQUIRED }) + @IsUUID("7", { message: CommonMessage.SESSION_ID_SHOULD_BE_UUID }) + @ApiProperty({ description: "Session id of the entity", example: "8b1e8b1e-8b1e-8b1e-8b1e-8b1e8b1e8b1e" }) + sessionId!: string; +} diff --git a/src/modules/chatbot/chatbot.controller.ts b/src/modules/chatbot/chatbot.controller.ts index ad1a69c..d6624e5 100644 --- a/src/modules/chatbot/chatbot.controller.ts +++ b/src/modules/chatbot/chatbot.controller.ts @@ -1,13 +1,14 @@ -import { Body, Controller, Get, Param, ParseIntPipe, ParseUUIDPipe, Post, Put, Query, Res } from "@nestjs/common"; +import { Body, Controller, Get, Param, Post, Put, Query, Res } from "@nestjs/common"; import { ApiOperation, ApiResponse } from "@nestjs/swagger"; import { FastifyReply } from "fastify"; -import { ChatMessageResponseDto, ChatSessionResponseDto } from "./DTO/chat-response.dto"; import { CreateChatSessionDto } from "./DTO/create-chat-session.dto"; import { SendMessageDto } from "./DTO/send-message.dto"; +import { SessionIdParamDto } from "./DTO/session-id.param.dto"; import { ChatbotService } from "./providers/chatbot.service"; import { AuthGuards } from "../../common/decorators/auth-guard.decorator"; import { UserDec } from "../../common/decorators/user.decorator"; +import { PaginationDto } from "../../common/DTO/pagination.dto"; @Controller("chatbot") @AuthGuards() @@ -16,33 +17,29 @@ export class ChatbotController { @Post("sessions") @ApiOperation({ summary: "Create a new chat session" }) - @ApiResponse({ status: 201, description: "Chat session created successfully", type: ChatSessionResponseDto }) - async createSession(@UserDec("id") userId: string, @Body() createDto: CreateChatSessionDto): Promise { + @ApiResponse({ status: 201, description: "Chat session created successfully" }) + createSession(@UserDec("id") userId: string, @Body() createDto: CreateChatSessionDto) { return this.chatbotService.createChatSession(userId, createDto); } @Get("sessions") @ApiOperation({ summary: "Get user's chat sessions" }) - @ApiResponse({ status: 200, description: "Chat sessions retrieved successfully", type: [ChatSessionResponseDto] }) - async getUserSessions( - @UserDec("id") userId: string, - @Query("limit", new ParseIntPipe({ optional: true })) limit = 10, - ): Promise { - return this.chatbotService.getUserChatSessions(userId, limit); + @ApiResponse({ status: 200, description: "Chat sessions retrieved successfully" }) + getUserSessions(@UserDec("id") userId: string, @Query() queryDto: PaginationDto) { + return this.chatbotService.getUserChatSessions(userId, queryDto.limit); } @Get("sessions/:sessionId") @ApiOperation({ summary: "Get a specific chat session with messages" }) - @ApiResponse({ status: 200, description: "Chat session retrieved successfully", type: ChatSessionResponseDto }) - async getSession(@UserDec("id") userId: string, @Param("sessionId", ParseUUIDPipe) sessionId: string): Promise { - return this.chatbotService.getChatSession(sessionId, userId); + @ApiResponse({ status: 200, description: "Chat session retrieved successfully" }) + getSession(@UserDec("id") userId: string, @Param() param: SessionIdParamDto) { + return this.chatbotService.getChatSession(param.sessionId, userId); } @Post("messages") @ApiOperation({ summary: "Send a message in a chat session (Enhanced with LangChain + Fallback)" }) - @ApiResponse({ status: 201, description: "Message sent successfully with enhanced AI", type: ChatMessageResponseDto }) - async sendMessage(@UserDec("id") userId: string, @Body() sendDto: SendMessageDto): Promise { - // Now uses LangChain by default with automatic Gemini fallback + @ApiResponse({ status: 201, description: "Message sent successfully with enhanced AI" }) + sendMessage(@UserDec("id") userId: string, @Body() sendDto: SendMessageDto) { return this.chatbotService.sendMessage(userId, sendDto); } @@ -85,74 +82,67 @@ export class ChatbotController { } } - @Post("messages/langchain") - @ApiOperation({ summary: "Send a message using LangChain with enhanced RAG capabilities" }) - @ApiResponse({ status: 201, description: "Message sent successfully with LangChain", type: ChatMessageResponseDto }) - async sendMessageWithLangChain(@UserDec("id") userId: string, @Body() sendDto: SendMessageDto): Promise { - return this.chatbotService.sendMessageWithLangChain(userId, sendDto); - } - - @Post("messages/langchain/stream") - @ApiOperation({ summary: "Send a message and get streaming response using LangChain" }) - @ApiResponse({ status: 200, description: "Streaming response initiated with LangChain" }) - async sendMessageStreamWithLangChain(@UserDec("id") userId: string, @Body() sendDto: SendMessageDto, @Res() res: FastifyReply): Promise { - try { - const { userMessage, streamGenerator } = await this.chatbotService.sendMessageStreamWithLangChain(userId, sendDto); - - // Set headers for Server-Sent Events - res.header("Content-Type", "text/event-stream"); - res.header("Cache-Control", "no-cache"); - res.header("Connection", "keep-alive"); - res.header("Access-Control-Allow-Origin", "*"); - - // Send initial user message - res.raw.write(`data: ${JSON.stringify({ type: "user_message", data: userMessage })}\n\n`); - - // Start streaming bot response - res.raw.write(`data: ${JSON.stringify({ type: "bot_response_start", provider: "langchain" })}\n\n`); - - const stream = await streamGenerator(); - for await (const chunk of stream) { - if (chunk) { - res.raw.write(`data: ${JSON.stringify({ type: "bot_response_chunk", data: chunk })}\n\n`); - } - } - - // End streaming - res.raw.write(`data: ${JSON.stringify({ type: "bot_response_end" })}\n\n`); - res.raw.end(); - } catch (error) { - console.error(error); - res.raw.write(`data: ${JSON.stringify({ type: "error", data: "خطا در تولید پاسخ با LangChain. لطفاً دوباره تلاش کنید. ⚠️" })}\n\n`); - res.raw.end(); - } - } - @Post("training/refresh") @ApiOperation({ summary: "Refresh LangChain training data from database" }) @ApiResponse({ status: 200, description: "Training data refreshed successfully" }) - async refreshTrainingData(@UserDec("id") _userId: string): Promise<{ message: string }> { - await this.chatbotService.refreshLangChainData(); - return { message: "داده‌های آموزشی با موفقیت به‌روزرسانی شدند 🔄" }; + refreshTrainingData(@UserDec("id") _userId: string) { + return this.chatbotService.refreshLangChainData(); } @Put("sessions/:sessionId/close") @ApiOperation({ summary: "Close a chat session" }) @ApiResponse({ status: 200, description: "Chat session closed successfully" }) - async closeSession(@UserDec("id") userId: string, @Param("sessionId", ParseUUIDPipe) sessionId: string): Promise<{ message: string }> { - await this.chatbotService.closeChatSession(sessionId, userId); - return { message: "جلسه چت با موفقیت بسته شد ✅" }; + closeSession(@UserDec("id") userId: string, @Param() param: SessionIdParamDto) { + return this.chatbotService.closeChatSession(param.sessionId, userId); } @Put("sessions/:sessionId/messages/read") @ApiOperation({ summary: "Mark messages as read" }) @ApiResponse({ status: 200, description: "Messages marked as read successfully" }) - async markMessagesAsRead( - @UserDec("id") userId: string, - @Param("sessionId", ParseUUIDPipe) sessionId: string, - @Body() body: { messageIds: string[] }, - ): Promise<{ message: string }> { - await this.chatbotService.markMessagesAsRead(sessionId, userId, body.messageIds); - return { message: "پیام‌ها با موفقیت به عنوان خوانده شده علامت‌گذاری شدند ✅" }; + markMessagesAsRead(@UserDec("id") userId: string, @Param() param: SessionIdParamDto, @Body() body: { messageIds: string[] }) { + return this.chatbotService.markMessagesAsRead(param.sessionId, userId, body.messageIds); } + + // @Post("messages/langchain") + // @ApiOperation({ summary: "Send a message using LangChain with enhanced RAG capabilities" }) + // @ApiResponse({ status: 201, description: "Message sent successfully with LangChain", type: ChatMessageResponseDto }) + // sendMessageWithLangChain(@UserDec("id") userId: string, @Body() sendDto: SendMessageDto): Promise { + // return this.chatbotService.sendMessageWithLangChain(userId, sendDto); + // } + + // @Post("messages/langchain/stream") + // @ApiOperation({ summary: "Send a message and get streaming response using LangChain" }) + // @ApiResponse({ status: 200, description: "Streaming response initiated with LangChain" }) + // async sendMessageStreamWithLangChain(@UserDec("id") userId: string, @Body() sendDto: SendMessageDto, @Res() res: FastifyReply): Promise { + // try { + // const { userMessage, streamGenerator } = await this.chatbotService.sendMessageStreamWithLangChain(userId, sendDto); + + // // Set headers for Server-Sent Events + // res.header("Content-Type", "text/event-stream"); + // res.header("Cache-Control", "no-cache"); + // res.header("Connection", "keep-alive"); + // res.header("Access-Control-Allow-Origin", "*"); + + // // Send initial user message + // res.raw.write(`data: ${JSON.stringify({ type: "user_message", data: userMessage })}\n\n`); + + // // Start streaming bot response + // res.raw.write(`data: ${JSON.stringify({ type: "bot_response_start", provider: "langchain" })}\n\n`); + + // const stream = await streamGenerator(); + // for await (const chunk of stream) { + // if (chunk) { + // res.raw.write(`data: ${JSON.stringify({ type: "bot_response_chunk", data: chunk })}\n\n`); + // } + // } + + // // End streaming + // res.raw.write(`data: ${JSON.stringify({ type: "bot_response_end" })}\n\n`); + // res.raw.end(); + // } catch (error) { + // console.error(error); + // res.raw.write(`data: ${JSON.stringify({ type: "error", data: "خطا در تولید پاسخ با LangChain. لطفاً دوباره تلاش کنید. ⚠️" })}\n\n`); + // res.raw.end(); + // } + // } } diff --git a/src/modules/chatbot/chatbot.gateway.ts b/src/modules/chatbot/chatbot.gateway.ts index dfdc74b..1398d2a 100644 --- a/src/modules/chatbot/chatbot.gateway.ts +++ b/src/modules/chatbot/chatbot.gateway.ts @@ -1,5 +1,4 @@ -import { Logger, UseGuards, UsePipes, ValidationPipe } from "@nestjs/common"; -import { JwtService } from "@nestjs/jwt"; +import { Logger, UseFilters, UseGuards, UsePipes, ValidationPipe } from "@nestjs/common"; import { ConnectedSocket, MessageBody, @@ -12,265 +11,261 @@ import { } from "@nestjs/websockets"; import { Server, Socket } from "socket.io"; -import { CHATBOT_CONSTANTS } from "./constants/chatbot.constants"; +import { WEBSOCKET_EVENTS } from "./constants/chatbot.constants"; import { AuthenticateDto, CreateSessionDto, JoinChatDto, LeaveChatDto, SendMessageWebSocketDto, TypingDto } from "./DTO/websocket-events.dto"; import { WebSocketAuthGuard } from "./guards/websocket-auth.guard"; +import { AuthenticatedSocket, ConnectedUserInfo, WebSocketResponse } from "./interfaces/websocket.interface"; import { ChatbotService } from "./providers/chatbot.service"; -import { extractTokenFromClient } from "../utils/providers/extract-token.utils"; +import { WebSocketAuthService } from "./providers/websocket-auth.service"; +import { WebSocketMessage } from "../../common/enums/message.enum"; +import { WsExceptionFilter } from "../../core/filters/ws-exception.filter"; @WebSocketGateway({ cors: { origin: "*", credentials: true }, namespace: "/chat", }) -@UseGuards(WebSocketAuthGuard) -// @UseFilters(WsExceptionFilter) @UsePipes(new ValidationPipe({ transform: true })) +@UseFilters(WsExceptionFilter) export class ChatbotGateway implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect { @WebSocketServer() server: Server; private readonly logger = new Logger(ChatbotGateway.name); - private connectedUsers = new Map(); // socketId -> user info - private userSessions = new Map>(); // userId -> Set of socketIds + + // Connection state management + private readonly connectedUsers = new Map(); + private readonly userSessions = new Map>(); constructor( - private chatbotService: ChatbotService, - private jwtService: JwtService, + private readonly chatbotService: ChatbotService, + private readonly authService: WebSocketAuthService, ) {} - afterInit(server: Server) { - this.logger.log("WebSocket Gateway initialized"); - this.logger.log(`Server instance: ${server ? "exists" : "null"}`); + /** + * Gateway initialization lifecycle hook + */ + afterInit(server: Server): void { + this.logger.log("WebSocket Gateway initialized", { + namespace: "/chat", + cors: true, + serverInstance: !!server, + }); } - // Connection Handlers - Authentication happens here - async handleConnection(@ConnectedSocket() client: Socket) { - this.logger.log("=== New Connection ==="); - this.logger.log(`Client ID: ${client.id}`); - this.logger.log(`Client IP: ${client.handshake.address}`); - this.logger.log("===================="); + /** + * Handles new WebSocket connections with authentication + * Implements fail-fast principle for security + */ + async handleConnection(client: Socket): Promise { + const connectionContext = { + clientId: client.id, + clientIP: client.handshake.address, + userAgent: client.handshake.headers["user-agent"], + timestamp: new Date().toISOString(), + }; + + this.logger.log("New WebSocket connection attempt", connectionContext); try { - // Extract token from handshake - const token = extractTokenFromClient(client); + const authResult = await this.authService.authenticateClient(client); - if (!token) { - this.logger.warn("No token provided in WebSocket connection"); - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.UNAUTHORIZED, { - message: CHATBOT_CONSTANTS.ERROR_MESSAGES.AUTHENTICATION_REQUIRED, - }); - client.disconnect(); + if (!authResult.success) { + this.authService.handleAuthenticationFailure(client, authResult.error!); return; } - // Verify JWT token - const payload = await this.jwtService.verifyAsync(token); + const user = authResult.user!; + this.registerUserConnection(client.id, user.id); + this.authService.emitAuthenticationSuccess(client, user); - // Store user data in client - client.data.user = payload; - - this.connectedUsers.set(client.id, { userId: payload.id }); - - if (!this.userSessions.has(payload?.id)) { - this.userSessions.set(payload?.id, new Set()); - } - this.userSessions.get(payload.id)!.add(client.id); - - this.logger.log(`Client connected successfully: ${client.id} for user: ${payload.id}`); - - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.AUTHENTICATED, { - message: CHATBOT_CONSTANTS.SUCCESS_MESSAGES.AUTHENTICATED, - userId: payload.id, + this.logger.log("WebSocket connection established", { + ...connectionContext, + userId: user.id, + authenticated: true, }); } catch (error) { - this.logger.error("Connection authentication failed:", error); - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.UNAUTHORIZED, { - message: CHATBOT_CONSTANTS.ERROR_MESSAGES.INVALID_TOKEN, + this.logger.error("WebSocket connection error", { + ...connectionContext, + error: error instanceof Error ? error.message : String(error), }); - client.disconnect(); + + this.authService.handleAuthenticationFailure(client, WebSocketMessage.CONNECTION_FAILED); } } - handleDisconnect(@ConnectedSocket() client: Socket) { - this.logger.log("=== Client Disconnected ==="); - this.logger.log(`Client ID: ${client.id}`); - this.logger.log("=========================="); - + /** + * Handles client disconnections with cleanup + */ + handleDisconnect(client: Socket): void { const userInfo = this.connectedUsers.get(client.id); + this.logger.log("WebSocket client disconnected", { + clientId: client.id, + userId: userInfo?.userId, + sessionId: userInfo?.sessionId, + }); + if (userInfo) { - this.logger.log(`Client disconnected: ${client.id} for user: ${userInfo.userId}`); - - // Remove from user sessions - const userSockets = this.userSessions.get(userInfo.userId); - if (userSockets) { - userSockets.delete(client.id); - if (userSockets.size === 0) { - this.userSessions.delete(userInfo.userId); - } - } - - // Leave session room if joined - if (userInfo.sessionId) { - client.leave(`session_${userInfo.sessionId}`); - - // Notify others in session - this.server.to(`session_${userInfo.sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.USER_LEFT, { - userId: userInfo.userId, - sessionId: userInfo.sessionId, - }); - } + this.cleanupUserConnection(client, userInfo); } this.connectedUsers.delete(client.id); } - // Authentication Handler - @SubscribeMessage(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.AUTHENTICATE) - @UseGuards(WebSocketAuthGuard) - async handleAuthenticate(@ConnectedSocket() client: Socket, @MessageBody() _data: AuthenticateDto) { + // ============================================================================ + // MESSAGE HANDLERS + // ============================================================================ + + /** + * Handles explicit authentication requests (for re-authentication) + */ + @SubscribeMessage(WEBSOCKET_EVENTS.AUTHENTICATE) + async handleAuthenticate(@ConnectedSocket() client: AuthenticatedSocket, @MessageBody() _data: AuthenticateDto): Promise { try { - // This is handled in connection, but we can re-authenticate if needed - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.AUTHENTICATED, { - message: CHATBOT_CONSTANTS.SUCCESS_MESSAGES.AUTHENTICATED, - userId: client.data.user?.id, + if (!this.authService.isClientAuthenticated(client)) { + throw new Error(WebSocketMessage.USER_NOT_AUTHENTICATED); + } + + this.emitSuccessResponse(client, WEBSOCKET_EVENTS.AUTHENTICATED, { + message: WebSocketMessage.AUTHENTICATED, + userId: client.data.user.id, }); } catch (error) { - this.logger.error("Authentication failed:", error); - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.UNAUTHORIZED, { - message: CHATBOT_CONSTANTS.ERROR_MESSAGES.INVALID_TOKEN, - }); + this.handleMessageError(client, error, WEBSOCKET_EVENTS.AUTHENTICATE); } } - // Session Management - @SubscribeMessage(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.CREATE_SESSION) + /** + * Handles chat session creation + */ + @SubscribeMessage(WEBSOCKET_EVENTS.CREATE_SESSION) @UseGuards(WebSocketAuthGuard) - async handleCreateSession(@ConnectedSocket() client: Socket, @MessageBody() data: CreateSessionDto) { + async handleCreateSession(@ConnectedSocket() client: AuthenticatedSocket, @MessageBody() data: CreateSessionDto): Promise { try { - const userId = client.data.user?.id; + const userId = this.authService.getUserId(client); if (!userId) { - throw new Error("User not authenticated"); + throw new Error(WebSocketMessage.USER_NOT_AUTHENTICATED); } const session = await this.chatbotService.createChatSession(userId, { title: data.title }); - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.SESSION_CREATED, { - message: CHATBOT_CONSTANTS.SUCCESS_MESSAGES.SESSION_CREATED, - session: session, + this.emitSuccessResponse(client, WEBSOCKET_EVENTS.SESSION_CREATED, { + message: WebSocketMessage.SESSION_CREATED, + session, }); - this.logger.log(`Session created: ${session.id} for user: ${userId}`); + this.logger.log("Chat session created", { + sessionId: session.id, + userId, + title: data.title, + }); } catch (error) { - this.logger.error("Session creation failed:", error); - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.SESSION_ERROR, { - message: CHATBOT_CONSTANTS.ERROR_MESSAGES.SESSION_CREATION_FAILED, + this.handleMessageError(client, error, WEBSOCKET_EVENTS.SESSION_ERROR); + this.emitErrorResponse(client, WEBSOCKET_EVENTS.SESSION_ERROR, { + message: WebSocketMessage.SESSION_CREATION_FAILED, }); } } - @SubscribeMessage(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.JOIN_CHAT) + /** + * Handles joining a chat session + */ + @SubscribeMessage(WEBSOCKET_EVENTS.JOIN_CHAT) @UseGuards(WebSocketAuthGuard) - async handleJoinChat(@ConnectedSocket() client: Socket, @MessageBody() data: JoinChatDto) { + async handleJoinChat(@ConnectedSocket() client: AuthenticatedSocket, @MessageBody() data: JoinChatDto): Promise { try { - const userId = client.data.user?.id; + const userId = this.authService.getUserId(client); if (!userId) { - throw new Error("User not authenticated"); + throw new Error(WebSocketMessage.USER_NOT_AUTHENTICATED); } // Verify session ownership await this.chatbotService.getChatSession(data.sessionId, userId); // Leave previous session if any - const userInfo = this.connectedUsers.get(client.id); - if (userInfo?.sessionId) { - client.leave(`session_${userInfo.sessionId}`); - } + await this.leavePreviousSession(client); - // Join new session room + // Join new session await client.join(`session_${data.sessionId}`); + this.updateUserSession(client.id, userId, data.sessionId); - // Update user info - this.connectedUsers.set(client.id, { - userId, - sessionId: data.sessionId, - }); - - this.logger.log(`User ${userId} joined chat session ${data.sessionId}`); - - // Notify client - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.CHAT_JOINED, { - message: CHATBOT_CONSTANTS.SUCCESS_MESSAGES.CHAT_JOINED, + // Emit success events + this.emitSuccessResponse(client, WEBSOCKET_EVENTS.CHAT_JOINED, { + message: WebSocketMessage.CHAT_JOINED, sessionId: data.sessionId, }); // Notify others in session - client.to(`session_${data.sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.USER_JOINED, { + client.to(`session_${data.sessionId}`).emit(WEBSOCKET_EVENTS.USER_JOINED, { userId, sessionId: data.sessionId, }); + + this.logger.log("User joined chat session", { + userId, + sessionId: data.sessionId, + clientId: client.id, + }); } catch (error) { - this.logger.error("Failed to join chat:", error); - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.ERROR, { - message: CHATBOT_CONSTANTS.ERROR_MESSAGES.SESSION_NOT_FOUND, + this.handleMessageError(client, error, WEBSOCKET_EVENTS.JOIN_CHAT); + this.emitErrorResponse(client, WEBSOCKET_EVENTS.ERROR, { + message: WebSocketMessage.SESSION_NOT_FOUND, }); } } - @SubscribeMessage(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.LEAVE_CHAT) + /** + * Handles leaving a chat session + */ + @SubscribeMessage(WEBSOCKET_EVENTS.LEAVE_CHAT) @UseGuards(WebSocketAuthGuard) - async handleLeaveChat(@ConnectedSocket() client: Socket, @MessageBody() data: LeaveChatDto) { + async handleLeaveChat(@ConnectedSocket() client: AuthenticatedSocket, @MessageBody() data: LeaveChatDto): Promise { try { - const userId = client.data.user?.id; + const userId = this.authService.getUserId(client); if (!userId) { - throw new Error("User not authenticated"); + throw new Error(WebSocketMessage.USER_NOT_AUTHENTICATED); } await client.leave(`session_${data.sessionId}`); + this.clearUserSession(client.id); - // Update user info - const userInfo = this.connectedUsers.get(client.id); - if (userInfo) { - userInfo.sessionId = undefined; - } - - this.logger.log(`User ${userId} left chat session ${data.sessionId}`); - - // Notify client - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.CHAT_LEFT, { + // Emit success events + this.emitSuccessResponse(client, WEBSOCKET_EVENTS.CHAT_LEFT, { sessionId: data.sessionId, }); // Notify others in session - client.to(`session_${data.sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.USER_LEFT, { + client.to(`session_${data.sessionId}`).emit(WEBSOCKET_EVENTS.USER_LEFT, { userId, sessionId: data.sessionId, }); - } catch (error) { - this.logger.error("Failed to leave chat:", error); - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.ERROR, { - message: "خطا در خروج از چت", + + this.logger.log("User left chat session", { + userId, + sessionId: data.sessionId, + clientId: client.id, }); + } catch (error) { + this.handleMessageError(client, error, WEBSOCKET_EVENTS.LEAVE_CHAT); } } - // Message Handling - @SubscribeMessage(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.SEND_MESSAGE) + /** + * Handles sending messages in chat + */ + @SubscribeMessage(WEBSOCKET_EVENTS.SEND_MESSAGE) @UseGuards(WebSocketAuthGuard) - async handleSendMessage(@ConnectedSocket() client: Socket, @MessageBody() data: SendMessageWebSocketDto) { + async handleSendMessage(@ConnectedSocket() client: AuthenticatedSocket, @MessageBody() data: SendMessageWebSocketDto): Promise { try { - const userId = client.data.user?.id; + const userId = this.authService.getUserId(client); if (!userId) { - throw new Error("User not authenticated"); + throw new Error(WebSocketMessage.USER_NOT_AUTHENTICATED); } - // Emit typing indicator to session - this.server.to(`session_${data.sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.TYPING_START, { - type: "bot", - sessionId: data.sessionId, - }); + // Start typing indicator + this.emitTypingIndicator(data.sessionId, "bot", true); - // Send message through enhanced service (LangChain by default with Gemini fallback) + // Send message through service const userMessage = await this.chatbotService.sendMessage(userId, { sessionId: data.sessionId, content: data.content, @@ -278,68 +273,157 @@ export class ChatbotGateway implements OnGatewayInit, OnGatewayConnection, OnGat metadata: data.metadata, }); - // Emit user message confirmation to session - this.server.to(`session_${data.sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.MESSAGE_RECEIVED, { + // Emit user message confirmation + this.server.to(`session_${data.sessionId}`).emit(WEBSOCKET_EVENTS.MESSAGE_RECEIVED, { message: userMessage, sessionId: data.sessionId, - enhanced: true, // Indicates enhanced processing + enhanced: true, }); - this.logger.log(`Enhanced message sent in session ${data.sessionId} by user ${userId}`); + this.logger.log("Message sent", { + userId, + sessionId: data.sessionId, + messageId: userMessage.id, + }); - // Generate bot response asynchronously using enhanced service + // Generate bot response asynchronously this.generateBotResponseAsync(data.sessionId, userId, data.content); } catch (error) { - this.logger.error("Failed to send enhanced message:", error); - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.MESSAGE_ERROR, { - message: CHATBOT_CONSTANTS.ERROR_MESSAGES.LLM_SERVICE_ERROR, + this.handleMessageError(client, error, WEBSOCKET_EVENTS.MESSAGE_ERROR); + this.emitErrorResponse(client, WEBSOCKET_EVENTS.MESSAGE_ERROR, { + message: WebSocketMessage.LLM_SERVICE_ERROR, }); + this.emitTypingIndicator(data.sessionId, "bot", false); + } + } - // Stop typing indicator on error - this.server.to(`session_${data.sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.TYPING_STOP, { - type: "bot", - sessionId: data.sessionId, + /** + * Handles typing start events + */ + @SubscribeMessage(WEBSOCKET_EVENTS.TYPING_START) + @UseGuards(WebSocketAuthGuard) + async handleTypingStart(@ConnectedSocket() client: AuthenticatedSocket, @MessageBody() data: TypingDto): Promise { + const userId = this.authService.getUserId(client); + if (!userId) return; + + client.to(`session_${data.sessionId}`).emit(WEBSOCKET_EVENTS.TYPING_START, { + userId, + type: "user", + sessionId: data.sessionId, + }); + } + + /** + * Handles typing stop events + */ + @SubscribeMessage(WEBSOCKET_EVENTS.TYPING_STOP) + @UseGuards(WebSocketAuthGuard) + async handleTypingStop(@ConnectedSocket() client: AuthenticatedSocket, @MessageBody() data: TypingDto): Promise { + const userId = this.authService.getUserId(client); + if (!userId) return; + + client.to(`session_${data.sessionId}`).emit(WEBSOCKET_EVENTS.TYPING_STOP, { + userId, + type: "user", + sessionId: data.sessionId, + }); + } + + // ============================================================================ + // PRIVATE HELPER METHODS + // ============================================================================ + + /** + * Registers a new user connection + */ + private registerUserConnection(clientId: string, userId: string): void { + this.connectedUsers.set(clientId, { userId }); + + if (!this.userSessions.has(userId)) { + this.userSessions.set(userId, new Set()); + } + this.userSessions.get(userId)!.add(clientId); + } + + /** + * Updates user session information + */ + private updateUserSession(clientId: string, _userId: string, sessionId: string): void { + const userInfo = this.connectedUsers.get(clientId); + if (userInfo) { + userInfo.sessionId = sessionId; + } + } + + /** + * Clears user session information + */ + private clearUserSession(clientId: string): void { + const userInfo = this.connectedUsers.get(clientId); + if (userInfo) { + userInfo.sessionId = undefined; + } + } + + /** + * Cleans up user connection on disconnect + */ + private cleanupUserConnection(client: Socket, userInfo: ConnectedUserInfo): void { + // Remove from user sessions + const userSockets = this.userSessions.get(userInfo.userId); + if (userSockets) { + userSockets.delete(client.id); + if (userSockets.size === 0) { + this.userSessions.delete(userInfo.userId); + } + } + + // Leave session room if joined + if (userInfo.sessionId) { + client.leave(`session_${userInfo.sessionId}`); + + // Notify others in session + this.server.to(`session_${userInfo.sessionId}`).emit(WEBSOCKET_EVENTS.USER_LEFT, { + userId: userInfo.userId, + sessionId: userInfo.sessionId, }); } } - // Typing Indicators - @SubscribeMessage(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.TYPING_START) - @UseGuards(WebSocketAuthGuard) - async handleTypingStart(@ConnectedSocket() client: Socket, @MessageBody() data: TypingDto) { - const userId = client.data.user?.id; - if (!userId) return; + /** + * Handles leaving previous session when joining a new one + */ + private async leavePreviousSession(client: Socket): Promise { + const userInfo = this.connectedUsers.get(client.id); + if (userInfo?.sessionId) { + await client.leave(`session_${userInfo.sessionId}`); + } + } - client.to(`session_${data.sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.TYPING_START, { - userId, - type: "user", - sessionId: data.sessionId, + /** + * Emits typing indicators + */ + private emitTypingIndicator(sessionId: string, type: "user" | "bot", isTyping: boolean): void { + const event = isTyping ? WEBSOCKET_EVENTS.TYPING_START : WEBSOCKET_EVENTS.TYPING_STOP; + + this.server.to(`session_${sessionId}`).emit(event, { + type, + sessionId, }); } - @SubscribeMessage(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.TYPING_STOP) - @UseGuards(WebSocketAuthGuard) - async handleTypingStop(@ConnectedSocket() client: Socket, @MessageBody() data: TypingDto) { - const userId = client.data.user?.id; - if (!userId) return; - - client.to(`session_${data.sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.TYPING_STOP, { - userId, - type: "user", - sessionId: data.sessionId, - }); - } - - // Bot Response Methods - private async generateBotResponseAsync(sessionId: string, userId: string, userMessage: string) { + /** + * Generates bot response asynchronously + */ + private async generateBotResponseAsync(sessionId: string, userId: string, userMessage: string): Promise { try { - // Start bot response - now using enhanced LangChain by default - this.server.to(`session_${sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.BOT_RESPONSE_START, { + // Start bot response + this.server.to(`session_${sessionId}`).emit(WEBSOCKET_EVENTS.BOT_RESPONSE_START, { sessionId, - provider: "enhanced", // Indicates LangChain with fallback + provider: "enhanced", }); - // Get streaming response from enhanced service (LangChain by default with Gemini fallback) + // Get streaming response const { streamGenerator } = await this.chatbotService.sendMessageStream(userId, { sessionId, content: userMessage, @@ -350,7 +434,7 @@ export class ChatbotGateway implements OnGatewayInit, OnGatewayConnection, OnGat // Stream chunks to client for await (const chunk of stream) { if (chunk) { - this.server.to(`session_${sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.BOT_RESPONSE_CHUNK, { + this.server.to(`session_${sessionId}`).emit(WEBSOCKET_EVENTS.BOT_RESPONSE_CHUNK, { data: chunk, sessionId, }); @@ -358,53 +442,97 @@ export class ChatbotGateway implements OnGatewayInit, OnGatewayConnection, OnGat } // End bot response - this.server.to(`session_${sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.BOT_RESPONSE_END, { - sessionId, - }); + this.server.to(`session_${sessionId}`).emit(WEBSOCKET_EVENTS.BOT_RESPONSE_END, { sessionId }); - // Stop typing indicator - this.server.to(`session_${sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.TYPING_STOP, { - type: "bot", - sessionId, - }); + this.emitTypingIndicator(sessionId, "bot", false); - this.logger.log(`Enhanced bot response completed for session ${sessionId}`); + this.logger.log("Bot response completed", { sessionId, userId }); } catch (error) { - this.logger.error("Enhanced bot response generation failed:", error); + this.logger.error("Bot response generation failed", { + sessionId, + userId, + error: error instanceof Error ? error.message : String(error), + }); - this.server.to(`session_${sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.ERROR, { - message: CHATBOT_CONSTANTS.ERROR_MESSAGES.LLM_SERVICE_ERROR, + this.server.to(`session_${sessionId}`).emit(WEBSOCKET_EVENTS.ERROR, { + message: WebSocketMessage.LLM_SERVICE_ERROR, sessionId, }); - // Stop typing indicator on error - this.server.to(`session_${sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.TYPING_STOP, { - type: "bot", - sessionId, - }); + this.emitTypingIndicator(sessionId, "bot", false); } } - // Public method to emit bot responses (called from service) - async emitBotResponse(sessionId: string, message: any) { - this.server.to(`session_${sessionId}`).emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.BOT_RESPONSE, { + /** + * Handles message-level errors with proper logging + */ + private handleMessageError(client: Socket, error: unknown, event: WEBSOCKET_EVENTS): void { + const errorContext = this.authService.createErrorContext(client, event); + + this.logger.error(`WebSocket message error: ${event}`, { + ...errorContext, + error: error instanceof Error ? error.message : String(error), + stack: error instanceof Error ? error.stack : undefined, + }); + } + + /** + * Emits success response with consistent structure + */ + private emitSuccessResponse(client: Socket, event: WEBSOCKET_EVENTS, data: T): void { + const response: WebSocketResponse = { + ...data, + status: "success", + timestamp: new Date().toISOString(), + }; + console.log(data); + + client.emit(event, response); + } + + /** + * Emits error response with consistent structure + */ + private emitErrorResponse(client: Socket, event: WEBSOCKET_EVENTS, error: { message: string }): void { + const response: WebSocketResponse = { + status: "error", + message: error.message, + timestamp: new Date().toISOString(), + }; + + client.emit(event, response); + } + + // ============================================================================ + // PUBLIC UTILITY METHODS + // ============================================================================ + + /** + * Public method to emit bot responses (called from service) + */ + async emitBotResponse(sessionId: string, message: unknown): Promise { + this.server.to(`session_${sessionId}`).emit(WEBSOCKET_EVENTS.BOT_RESPONSE, { message, sessionId, }); } - // Utility method to get connected users in a session + /** + * Gets connected users in a session + */ getSessionUsers(sessionId: string): string[] { const users: string[] = []; - this.connectedUsers.forEach((userInfo, _socketId) => { + this.connectedUsers.forEach((userInfo) => { if (userInfo.sessionId === sessionId) { users.push(userInfo.userId); } }); - return [...new Set(users)]; // Remove duplicates + return [...new Set(users)]; } - // Utility method to check if user is online + /** + * Checks if user is online + */ isUserOnline(userId: string): boolean { return this.userSessions.has(userId) && this.userSessions.get(userId)!.size > 0; } diff --git a/src/modules/chatbot/chatbot.module.ts b/src/modules/chatbot/chatbot.module.ts index c4056c6..43ee37b 100644 --- a/src/modules/chatbot/chatbot.module.ts +++ b/src/modules/chatbot/chatbot.module.ts @@ -10,12 +10,14 @@ import { ChatbotService } from "./providers/chatbot.service"; import { DataContextService } from "./providers/data-context.service"; import { LangChainService } from "./providers/langchain.service"; import { LLMService } from "./providers/llm.service"; +import { WebSocketAuthService } from "./providers/websocket-auth.service"; +import { AuthModule } from "../auth/auth.module"; import { UsersModule } from "../users/users.module"; @Module({ - imports: [MikroOrmModule.forFeature([ChatSession, ChatMessage]), UsersModule], + imports: [MikroOrmModule.forFeature([ChatSession, ChatMessage]), AuthModule, UsersModule], controllers: [ChatbotController], - providers: [ChatbotGateway, ChatbotService, LLMService, LangChainService, DataContextService, WebSocketAuthGuard], - exports: [ChatbotService, LangChainService], + providers: [ChatbotService, ChatbotGateway, LLMService, LangChainService, DataContextService, WebSocketAuthGuard, WebSocketAuthService], + exports: [ChatbotService, ChatbotGateway], }) export class ChatbotModule {} diff --git a/src/modules/chatbot/constants/chatbot.constants.ts b/src/modules/chatbot/constants/chatbot.constants.ts index d441ea2..5eec06c 100644 --- a/src/modules/chatbot/constants/chatbot.constants.ts +++ b/src/modules/chatbot/constants/chatbot.constants.ts @@ -239,45 +239,45 @@ export const CHATBOT_CONSTANTS = { INACTIVE: "غیرفعال", }, - WEBSOCKET_EVENTS: { - // Connection events - CONNECT: "connect", - DISCONNECT: "disconnect", - CONNECTION_ERROR: "connection_failed", + // WEBSOCKET_EVENTS: { + // // Connection events + // CONNECT: "connect", + // DISCONNECT: "disconnect", + // CONNECTION_ERROR: "connection_failed", - // Authentication events - AUTHENTICATE: "authenticate", - AUTHENTICATED: "authenticated", - UNAUTHORIZED: "unauthorized", + // // Authentication events + // AUTHENTICATE: "authenticate", + // AUTHENTICATED: "authenticated", + // UNAUTHORIZED: "unauthorized", - // Session events - CREATE_SESSION: "create_session", - SESSION_CREATED: "session_created", - JOIN_CHAT: "join_chat", - CHAT_JOINED: "chat_joined", - LEAVE_CHAT: "leave_chat", - CHAT_LEFT: "chat_left", + // // Session events + // CREATE_SESSION: "create_session", + // SESSION_CREATED: "session_created", + // JOIN_CHAT: "join_chat", + // CHAT_JOINED: "chat_joined", + // LEAVE_CHAT: "leave_chat", + // CHAT_LEFT: "chat_left", - // Message events - SEND_MESSAGE: "send_message", - MESSAGE_RECEIVED: "message_received", - BOT_RESPONSE_START: "bot_response_start", - BOT_RESPONSE_CHUNK: "bot_response_chunk", - BOT_RESPONSE_END: "bot_response_end", - BOT_RESPONSE: "bot_response", + // // Message events + // SEND_MESSAGE: "send_message", + // MESSAGE_RECEIVED: "message_received", + // BOT_RESPONSE_START: "bot_response_start", + // BOT_RESPONSE_CHUNK: "bot_response_chunk", + // BOT_RESPONSE_END: "bot_response_end", + // BOT_RESPONSE: "bot_response", - // Status events - TYPING_START: "typing_start", - TYPING_STOP: "typing_stop", - USER_JOINED: "user_joined", - USER_LEFT: "user_left", + // // Status events + // TYPING_START: "typing_start", + // TYPING_STOP: "typing_stop", + // USER_JOINED: "user_joined", + // USER_LEFT: "user_left", - // Error events - ERROR: "error", - WEBSOCKET_ERROR: "websocket_error", - SESSION_ERROR: "session_error", - MESSAGE_ERROR: "message_error", - }, + // // Error events + // ERROR: "error", + // WEBSOCKET_ERROR: "websocket_error", + // SESSION_ERROR: "session_error", + // MESSAGE_ERROR: "message_error", + // }, ERROR_MESSAGES: { SESSION_NOT_FOUND: "جلسه چت یافت نشد", @@ -299,3 +299,42 @@ export const CHATBOT_CONSTANTS = { AUTHENTICATED: "احراز هویت موفقیت‌آمیز", }, } as const; + +export const enum WEBSOCKET_EVENTS { + CONNECT = "connect", + DISCONNECT = "disconnect", + CONNECTION_ERROR = "connection_failed", + + // Authentication events + AUTHENTICATE = "authenticate", + AUTHENTICATED = "authenticated", + UNAUTHORIZED = "unauthorized", + + // Session events + CREATE_SESSION = "create_session", + SESSION_CREATED = "session_created", + JOIN_CHAT = "join_chat", + CHAT_JOINED = "chat_joined", + LEAVE_CHAT = "leave_chat", + CHAT_LEFT = "chat_left", + + // Message events + SEND_MESSAGE = "send_message", + MESSAGE_RECEIVED = "message_received", + BOT_RESPONSE_START = "bot_response_start", + BOT_RESPONSE_CHUNK = "bot_response_chunk", + BOT_RESPONSE_END = "bot_response_end", + BOT_RESPONSE = "bot_response", + + // Status events + TYPING_START = "typing_start", + TYPING_STOP = "typing_stop", + USER_JOINED = "user_joined", + USER_LEFT = "user_left", + + // Error events + ERROR = "error", + WEBSOCKET_ERROR = "websocket_error", + SESSION_ERROR = "session_error", + MESSAGE_ERROR = "message_error", +} diff --git a/src/modules/chatbot/guards/websocket-auth.guard.ts b/src/modules/chatbot/guards/websocket-auth.guard.ts index 48df594..6e53905 100644 --- a/src/modules/chatbot/guards/websocket-auth.guard.ts +++ b/src/modules/chatbot/guards/websocket-auth.guard.ts @@ -3,7 +3,8 @@ import { JwtService } from "@nestjs/jwt"; import { WsException } from "@nestjs/websockets"; import { Socket } from "socket.io"; -import { CHATBOT_CONSTANTS } from "../constants/chatbot.constants"; +import { WebSocketMessage } from "../../../common/enums/message.enum"; +import { WEBSOCKET_EVENTS } from "../constants/chatbot.constants"; @Injectable() export class WebSocketAuthGuard implements CanActivate { @@ -21,8 +22,8 @@ export class WebSocketAuthGuard implements CanActivate { if (!token) { this.logger.warn("No token provided in WebSocket connection"); throw new WsException({ - event: CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.UNAUTHORIZED, - message: CHATBOT_CONSTANTS.ERROR_MESSAGES.AUTHENTICATION_REQUIRED, + event: WEBSOCKET_EVENTS.UNAUTHORIZED, + message: WebSocketMessage.AUTHENTICATION_REQUIRED, }); } @@ -36,13 +37,13 @@ export class WebSocketAuthGuard implements CanActivate { } catch (error) { this.logger.error("WebSocket authentication failed:", error); - client.emit(CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.UNAUTHORIZED, { - message: CHATBOT_CONSTANTS.ERROR_MESSAGES.INVALID_TOKEN, + client.emit(WEBSOCKET_EVENTS.UNAUTHORIZED, { + message: WebSocketMessage.INVALID_TOKEN, }); throw new WsException({ - event: CHATBOT_CONSTANTS.WEBSOCKET_EVENTS.UNAUTHORIZED, - message: CHATBOT_CONSTANTS.ERROR_MESSAGES.INVALID_TOKEN, + event: WEBSOCKET_EVENTS.UNAUTHORIZED, + message: WebSocketMessage.INVALID_TOKEN, }); } } diff --git a/src/modules/chatbot/interfaces/websocket.interface.ts b/src/modules/chatbot/interfaces/websocket.interface.ts new file mode 100644 index 0000000..326e0fd --- /dev/null +++ b/src/modules/chatbot/interfaces/websocket.interface.ts @@ -0,0 +1,52 @@ +import { Socket } from "socket.io"; + +import { ITokenPayload } from "../../auth/interfaces/IToken-payload"; + +/** + * Extended Socket interface with authenticated user data + */ +export interface AuthenticatedSocket extends Socket { + data: { + user: ITokenPayload; + sessionId?: string; + }; +} + +/** + * User connection information stored in gateway + */ +export interface ConnectedUserInfo { + userId: string; + sessionId?: string; +} + +/** + * WebSocket error context for logging and debugging + */ +export interface WebSocketErrorContext { + clientId: string; + userId?: string; + sessionId?: string; + event?: string; + data?: any; + timestamp: Date; +} + +/** + * WebSocket event response structure + */ +export interface WebSocketResponse { + status: "success" | "error"; + message?: string; + data?: T; + timestamp?: string; +} + +/** + * Authentication result for WebSocket connections + */ +export interface AuthenticationResult { + success: boolean; + user?: ITokenPayload; + error?: string; +} diff --git a/src/modules/chatbot/providers/chatbot.service.ts b/src/modules/chatbot/providers/chatbot.service.ts index 9e894c0..b9321a2 100644 --- a/src/modules/chatbot/providers/chatbot.service.ts +++ b/src/modules/chatbot/providers/chatbot.service.ts @@ -6,7 +6,6 @@ import { LangChainService } from "./langchain.service"; import { LLMService } from "./llm.service"; import { User } from "../../users/entities/user.entity"; // import { UsersService } from "../../users/services/users.service"; -import { ChatMessageResponseDto, ChatSessionResponseDto } from "../DTO/chat-response.dto"; import { CreateChatSessionDto } from "../DTO/create-chat-session.dto"; import { SendMessageDto } from "../DTO/send-message.dto"; import { ChatMessage, MessageStatus, MessageType } from "../entities/chat-message.entity"; @@ -40,7 +39,7 @@ export class ChatbotService { this.logger.log(`Using ${this.defaultProvider} as default LLM provider`); } - async createChatSession(userId: string, createDto: CreateChatSessionDto): Promise { + async createChatSession(userId: string, createDto: CreateChatSessionDto) { const em = this.em.fork(); const user = await em.findOne(User, { id: userId }); if (!user) { @@ -58,14 +57,14 @@ export class ChatbotService { return this.mapSessionToDto(session); } //************************************ */ - async getUserChatSessions(userId: string, limit = 10): Promise { + async getUserChatSessions(userId: string, limit = 10) { const em = this.em.fork(); const sessions = await this.chatSessionRepo.findByUserWithMessages(userId, limit, em); return sessions.map((session) => this.mapSessionToDto(session)); } //************************************ */ - async getChatSession(sessionId: string, userId: string): Promise { + async getChatSession(sessionId: string, userId: string) { const em = this.em.fork(); const session = await em.findOne(ChatSession, { id: sessionId, user: userId }, { populate: ["messages", "messages.sender"] }); @@ -77,7 +76,7 @@ export class ChatbotService { } //************************************ */ - async sendMessage(userId: string, sendDto: SendMessageDto, provider?: LLMProvider): Promise { + async sendMessage(userId: string, sendDto: SendMessageDto, provider?: LLMProvider) { const selectedProvider = provider || this.defaultProvider; // Use LangChain by default for enhanced responses @@ -96,7 +95,7 @@ export class ChatbotService { //************************************ */ - private async sendMessageWithGemini(userId: string, sendDto: SendMessageDto): Promise { + private async sendMessageWithGemini(userId: string, sendDto: SendMessageDto) { const em = this.em.fork(); const session = await em.findOne(ChatSession, { id: sendDto.sessionId, user: userId }, { populate: ["messages"] }); @@ -135,14 +134,7 @@ export class ChatbotService { //************************************ */ - async sendMessageStream( - userId: string, - sendDto: SendMessageDto, - provider?: LLMProvider, - ): Promise<{ - userMessage: ChatMessageResponseDto; - streamGenerator: () => Promise>; - }> { + async sendMessageStream(userId: string, sendDto: SendMessageDto, provider?: LLMProvider) { const selectedProvider = provider || this.defaultProvider; // Use LangChain by default for enhanced streaming responses @@ -161,13 +153,7 @@ export class ChatbotService { //************************************ */ - private async sendMessageStreamWithGemini( - userId: string, - sendDto: SendMessageDto, - ): Promise<{ - userMessage: ChatMessageResponseDto; - streamGenerator: () => Promise>; - }> { + private async sendMessageStreamWithGemini(userId: string, sendDto: SendMessageDto) { const em = this.em.fork(); const session = await em.findOne(ChatSession, { id: sendDto.sessionId, user: userId }, { populate: ["messages"] }); @@ -283,7 +269,7 @@ export class ChatbotService { //************************************ */ - async closeChatSession(sessionId: string, userId: string): Promise { + async closeChatSession(sessionId: string, userId: string) { const em = this.em.fork(); const session = await em.findOne(ChatSession, { id: sessionId, user: userId }); @@ -293,11 +279,13 @@ export class ChatbotService { session.status = ChatSessionStatus.CLOSED; await em.persistAndFlush(session); + + return { message: "Chat session closed successfully" }; } //************************************ */ - async markMessagesAsRead(sessionId: string, userId: string, messageIds: string[]): Promise { + async markMessagesAsRead(sessionId: string, userId: string, messageIds: string[]) { const em = this.em.fork(); const session = await em.findOne(ChatSession, { id: sessionId, user: userId }); @@ -306,11 +294,13 @@ export class ChatbotService { } await this.chatMessageRepo.markAsRead(messageIds, em); + + return { message: "Messages marked as read successfully" }; } //************************************ */ - private mapSessionToDto(session: ChatSession): ChatSessionResponseDto { + private mapSessionToDto(session: ChatSession) { return { id: session.id, title: session.title, @@ -324,7 +314,7 @@ export class ChatbotService { //************************************ */ - private mapMessageToDto(message: ChatMessage): ChatMessageResponseDto { + private mapMessageToDto(message: ChatMessage) { return { id: message.id, content: message.content, @@ -339,7 +329,7 @@ export class ChatbotService { //************************************ */ - async sendMessageWithLangChain(userId: string, sendDto: SendMessageDto): Promise { + async sendMessageWithLangChain(userId: string, sendDto: SendMessageDto) { const em = this.em.fork(); const session = await em.findOne(ChatSession, { id: sendDto.sessionId, user: userId }, { populate: ["messages"] }); @@ -378,13 +368,7 @@ export class ChatbotService { //************************************ */ - async sendMessageStreamWithLangChain( - userId: string, - sendDto: SendMessageDto, - ): Promise<{ - userMessage: ChatMessageResponseDto; - streamGenerator: () => Promise>; - }> { + async sendMessageStreamWithLangChain(userId: string, sendDto: SendMessageDto) { const em = this.em.fork(); const session = await em.findOne(ChatSession, { id: sendDto.sessionId, user: userId }, { populate: ["messages"] }); @@ -496,10 +480,11 @@ export class ChatbotService { //************************************ */ - async refreshLangChainData(): Promise { + async refreshLangChainData() { try { await this.langChainService.refreshVectorStore(); this.logger.log("LangChain vector store refreshed successfully"); + return { message: "LangChain vector store refreshed successfully" }; } catch (error) { this.logger.error("Failed to refresh LangChain vector store", error); throw new Error("Failed to refresh training data"); diff --git a/src/modules/chatbot/providers/websocket-auth.service.ts b/src/modules/chatbot/providers/websocket-auth.service.ts new file mode 100644 index 0000000..e4847bd --- /dev/null +++ b/src/modules/chatbot/providers/websocket-auth.service.ts @@ -0,0 +1,178 @@ +import { Injectable, Logger } from "@nestjs/common"; +import { JwtService } from "@nestjs/jwt"; +import { Socket } from "socket.io"; + +import { WebSocketMessage } from "../../../common/enums/message.enum"; +import { ITokenPayload } from "../../auth/interfaces/IToken-payload"; +import { extractTokenFromClient } from "../../utils/providers/extract-token.utils"; +import { WEBSOCKET_EVENTS } from "../constants/chatbot.constants"; +import { AuthenticationResult, WebSocketErrorContext } from "../interfaces/websocket.interface"; + +/** + * Service responsible for WebSocket authentication logic + * Follows single responsibility principle and provides reusable authentication methods + */ +@Injectable() +export class WebSocketAuthService { + private readonly logger = new Logger(WebSocketAuthService.name); + + constructor(private readonly jwtService: JwtService) {} + + /** + * Authenticates a WebSocket client and returns the result + * + * @param client - Socket.IO client to authenticate + * @returns Promise with authentication result + */ + async authenticateClient(client: Socket): Promise { + try { + const token = extractTokenFromClient(client); + + if (!token) { + this.logAuthenticationAttempt(client, false, "No token provided"); + return { + success: false, + error: WebSocketMessage.AUTHENTICATION_REQUIRED, + }; + } + + const payload = await this.jwtService.verifyAsync(token); + + if (!payload?.id) { + this.logAuthenticationAttempt(client, false, "Invalid token payload"); + return { + success: false, + error: WebSocketMessage.INVALID_TOKEN, + }; + } + + // Store user data in client + client.data.user = payload; + + this.logAuthenticationAttempt(client, true, undefined, payload.id); + + return { + success: true, + user: payload, + }; + } catch (error) { + const errorMessage = this.getAuthErrorMessage(error); + this.logAuthenticationAttempt(client, false, errorMessage); + + return { + success: false, + error: errorMessage, + }; + } + } + + /** + * Emits authentication failure event to client and disconnects + * + * @param client - Socket.IO client + * @param error - Error message to send + */ + handleAuthenticationFailure(client: Socket, error: string): void { + client.emit(WEBSOCKET_EVENTS.UNAUTHORIZED, { + message: error, + timestamp: new Date().toISOString(), + }); + + // Graceful disconnect with delay to ensure message is received + setTimeout(() => { + client.disconnect(true); + }, 100); + } + + /** + * Emits successful authentication event to client + * + * @param client - Socket.IO client + * @param user - Authenticated user payload + */ + emitAuthenticationSuccess(client: Socket, user: ITokenPayload): void { + client.emit(WEBSOCKET_EVENTS.AUTHENTICATED, { + message: WebSocketMessage.AUTHENTICATED, + userId: user.id, + timestamp: new Date().toISOString(), + }); + } + + /** + * Creates error context object for logging and monitoring + * + * @param client - Socket.IO client + * @param event - Event name (optional) + * @param data - Event data (optional) + * @returns WebSocket error context + */ + createErrorContext(client: Socket, event?: WEBSOCKET_EVENTS, data?: any): WebSocketErrorContext { + return { + clientId: client.id, + userId: client.data?.user?.id, + sessionId: client.data?.sessionId, + event, + data, + timestamp: new Date(), + }; + } + + /** + * Validates if client is authenticated + * + * @param client - Socket.IO client to validate + * @returns True if client has valid user data + */ + isClientAuthenticated(client: Socket): boolean { + return !!client.data?.user?.id; + } + + /** + * Gets user ID from authenticated client + * + * @param client - Socket.IO client + * @returns User ID or undefined if not authenticated + */ + getUserId(client: Socket): string | undefined { + return client.data?.user?.id; + } + + /** + * Logs authentication attempts with structured data + */ + private logAuthenticationAttempt(client: Socket, success: boolean, error?: string, userId?: string): void { + const logData = { + clientId: client.id, + clientIP: client.handshake.address, + success, + userId, + error, + timestamp: new Date().toISOString(), + }; + + if (success) { + this.logger.log(`WebSocket authentication successful`, logData); + } else { + this.logger.warn(`WebSocket authentication failed`, logData); + } + } + + /** + * Maps JWT errors to user-friendly messages + */ + private getAuthErrorMessage(error: any): string { + if (error?.name === "TokenExpiredError") { + return WebSocketMessage.TOKEN_EXPIRED; + } + + if (error?.name === "JsonWebTokenError") { + return WebSocketMessage.INVALID_TOKEN; + } + + if (error?.name === "NotBeforeError") { + return WebSocketMessage.INVALID_TOKEN; + } + + return WebSocketMessage.INVALID_TOKEN; + } +} diff --git a/src/modules/search/DTO/search.dto.ts b/src/modules/search/DTO/search.dto.ts new file mode 100644 index 0000000..8f23e93 --- /dev/null +++ b/src/modules/search/DTO/search.dto.ts @@ -0,0 +1,9 @@ +import { ApiProperty } from "@nestjs/swagger"; +import { IsNotEmpty, IsString } from "class-validator"; + +export class SearchDto { + @IsNotEmpty({ message: "Search query is required" }) + @IsString({ message: "Search query must be a string" }) + @ApiProperty({ description: "The search query", example: "aluminum" }) + q: string; +} diff --git a/src/modules/search/search.controller.ts b/src/modules/search/search.controller.ts new file mode 100644 index 0000000..a69d83f --- /dev/null +++ b/src/modules/search/search.controller.ts @@ -0,0 +1,14 @@ +import { Controller, Get, Query } from "@nestjs/common"; + +import { SearchDto } from "./DTO/search.dto"; +import { SearchService } from "./services/search.service"; + +@Controller("search") +export class SearchController { + constructor(private readonly searchService: SearchService) {} + + @Get() + search(@Query() searchDto: SearchDto) { + return this.searchService.search(searchDto); + } +} diff --git a/src/modules/search/search.module.ts b/src/modules/search/search.module.ts new file mode 100644 index 0000000..82a7c6b --- /dev/null +++ b/src/modules/search/search.module.ts @@ -0,0 +1,13 @@ +import { MikroOrmModule } from "@mikro-orm/nestjs"; +import { Module } from "@nestjs/common"; + +import { SearchController } from "./search.controller"; +import { SearchService } from "./services/search.service"; +import { Company } from "../companies/entities/company.entity"; + +@Module({ + imports: [MikroOrmModule.forFeature([Company])], + controllers: [SearchController], + providers: [SearchService], +}) +export class SearchModule {} diff --git a/src/modules/search/services/search.service.ts b/src/modules/search/services/search.service.ts new file mode 100644 index 0000000..a1be726 --- /dev/null +++ b/src/modules/search/services/search.service.ts @@ -0,0 +1,32 @@ +import { Injectable } from "@nestjs/common"; + +import { CompanyRepository } from "../../companies/repositories/company.repository"; +import { SearchDto } from "../DTO/search.dto"; + +@Injectable() +export class SearchService { + constructor(private readonly companyRepository: CompanyRepository) {} + + search(searchDto: SearchDto) { + return this.companyRepository.find( + { + deletedAt: null, + isActive: true, + $or: [ + { name: { $ilike: `%${searchDto.q}%` } }, + { description: { $ilike: `%${searchDto.q}%` } }, + { chiefExecutiveOfficer: { $ilike: `%${searchDto.q}%` } }, + { identificationNumber: { $ilike: `%${searchDto.q}%` } }, + { address: { $ilike: `%${searchDto.q}%` } }, + ], + }, + { + offset: 0, + limit: 10, + populate: ["industry", "user"], + orderBy: { createdAt: "DESC" }, + fields: ["id", "name", "description", "chiefExecutiveOfficer", "address", "coverImageUrl", "profileImageUrl"], + }, + ); + } +} diff --git a/src/modules/utils/providers/extract-token.utils.ts b/src/modules/utils/providers/extract-token.utils.ts index 0bdf2dd..8467993 100644 --- a/src/modules/utils/providers/extract-token.utils.ts +++ b/src/modules/utils/providers/extract-token.utils.ts @@ -1,6 +1,29 @@ import { Socket } from "socket.io"; -export const extractTokenFromClient = (client: Socket): string | undefined => { - const authToken = client.handshake?.auth?.token || client.handshake?.query?.token; - return authToken; -}; +/** + * Extracts authentication token from WebSocket client handshake + * Supports multiple token sources for flexibility + * + * @param client - Socket.IO client instance + * @returns Authentication token string or undefined if not found + */ +export function extractTokenFromClient(client: Socket): string | undefined { + // Priority order: auth.token > query.token > headers.authorization + const authToken = client.handshake?.auth?.token; + if (authToken) { + return authToken; + } + + const queryToken = client.handshake?.query?.token; + if (queryToken && typeof queryToken === "string") { + return queryToken; + } + + const authHeader = client.handshake?.headers?.authorization; + if (authHeader && typeof authHeader === "string") { + // Remove 'Bearer ' prefix if present + return authHeader.replace(/^Bearer\s+/i, ""); + } + + return undefined; +}