From 3ebae01605f0bf5ebf0fa6479e8786057aadd090 Mon Sep 17 00:00:00 2001 From: morteza-mortezai Date: Sat, 29 Nov 2025 11:46:23 +0330 Subject: [PATCH] chatbot : init --- .../chatbot/DTO/create-chat-session.dto.ts | 13 + src/modules/chatbot/DTO/send-message.dto.ts | 22 + .../chatbot/DTO/session-id.param.dto.ts | 11 + .../chatbot/DTO/websocket-events.dto.ts | 62 ++ src/modules/chatbot/chatbot.controller.ts | 148 +++++ src/modules/chatbot/chatbot.gateway.ts | 539 ++++++++++++++++++ src/modules/chatbot/chatbot.module.ts | 23 + .../chatbot/constants/chatbot.constants.ts | 340 +++++++++++ .../chatbot/entities/chat-message.entity.ts | 48 ++ .../chatbot/entities/chat-session.entity.ts | 35 ++ .../chatbot/guards/websocket-auth.guard.ts | 55 ++ .../chatbot/interfaces/chatbot.interface.ts | 36 ++ .../chatbot/interfaces/websocket.interface.ts | 52 ++ .../chatbot/providers/chatbot.service.ts | 495 ++++++++++++++++ .../chatbot/providers/data-context.service.ts | 226 ++++++++ .../chatbot/providers/langchain.service.ts | 389 +++++++++++++ src/modules/chatbot/providers/llm.service.ts | 237 ++++++++ .../providers/websocket-auth.service.ts | 178 ++++++ .../repositories/chat-message.repository.ts | 34 ++ .../repositories/chat-session.repository.ts | 33 ++ 20 files changed, 2976 insertions(+) create mode 100644 src/modules/chatbot/DTO/create-chat-session.dto.ts create mode 100644 src/modules/chatbot/DTO/send-message.dto.ts create mode 100644 src/modules/chatbot/DTO/session-id.param.dto.ts create mode 100644 src/modules/chatbot/DTO/websocket-events.dto.ts create mode 100644 src/modules/chatbot/chatbot.controller.ts create mode 100644 src/modules/chatbot/chatbot.gateway.ts create mode 100644 src/modules/chatbot/chatbot.module.ts create mode 100644 src/modules/chatbot/constants/chatbot.constants.ts create mode 100644 src/modules/chatbot/entities/chat-message.entity.ts create mode 100644 src/modules/chatbot/entities/chat-session.entity.ts create mode 100644 src/modules/chatbot/guards/websocket-auth.guard.ts create mode 100644 src/modules/chatbot/interfaces/chatbot.interface.ts create mode 100644 src/modules/chatbot/interfaces/websocket.interface.ts create mode 100644 src/modules/chatbot/providers/chatbot.service.ts create mode 100644 src/modules/chatbot/providers/data-context.service.ts create mode 100644 src/modules/chatbot/providers/langchain.service.ts create mode 100644 src/modules/chatbot/providers/llm.service.ts create mode 100644 src/modules/chatbot/providers/websocket-auth.service.ts create mode 100644 src/modules/chatbot/repositories/chat-message.repository.ts create mode 100644 src/modules/chatbot/repositories/chat-session.repository.ts diff --git a/src/modules/chatbot/DTO/create-chat-session.dto.ts b/src/modules/chatbot/DTO/create-chat-session.dto.ts new file mode 100644 index 0000000..2753f13 --- /dev/null +++ b/src/modules/chatbot/DTO/create-chat-session.dto.ts @@ -0,0 +1,13 @@ +import { ApiProperty, ApiPropertyOptional } from "@nestjs/swagger"; +import { IsOptional, IsString, MaxLength } from "class-validator"; + +export class CreateChatSessionDto { + @ApiProperty({ description: "Title for the chat session", example: "Help with payment issues" }) + @IsString() + @MaxLength(255) + title!: string; + + @IsOptional() + @ApiPropertyOptional({ description: "Initial context for the chat", example: { user: "John Doe", orderId: "123456" } }) + context?: Record; +} diff --git a/src/modules/chatbot/DTO/send-message.dto.ts b/src/modules/chatbot/DTO/send-message.dto.ts new file mode 100644 index 0000000..68e9317 --- /dev/null +++ b/src/modules/chatbot/DTO/send-message.dto.ts @@ -0,0 +1,22 @@ +import { ApiProperty } from "@nestjs/swagger"; +import { IsOptional, IsString, IsUUID, MaxLength } from "class-validator"; + +export class SendMessageDto { + @ApiProperty({ description: "Message content", example: "How can I cancel my subscription?" }) + @IsString() + @MaxLength(2000) + content!: string; + + @ApiProperty({ description: "Chat session ID" }) + @IsUUID() + sessionId!: string; + + @ApiProperty({ description: "ID of message being replied to", required: false }) + @IsOptional() + @IsUUID() + responseToId?: string; + + @ApiProperty({ description: "Additional metadata for the message", required: false }) + @IsOptional() + metadata?: 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/DTO/websocket-events.dto.ts b/src/modules/chatbot/DTO/websocket-events.dto.ts new file mode 100644 index 0000000..a7223b0 --- /dev/null +++ b/src/modules/chatbot/DTO/websocket-events.dto.ts @@ -0,0 +1,62 @@ +import { IsNotEmpty, IsOptional, IsString, IsUUID, MaxLength } from "class-validator"; + +export class AuthenticateDto { + @IsNotEmpty() + @IsString() + token!: string; +} + +export class CreateSessionDto { + @IsNotEmpty() + @IsString() + @MaxLength(100) + title!: string; +} + +export class JoinChatDto { + @IsNotEmpty() + @IsUUID() + sessionId!: string; + + @IsNotEmpty() + @IsString() + userId!: string; +} + +export class LeaveChatDto { + @IsNotEmpty() + @IsUUID() + sessionId!: string; +} + +export class SendMessageWebSocketDto { + @IsNotEmpty() + @IsUUID() + sessionId!: string; + + @IsNotEmpty() + @IsString() + @MaxLength(4000) + content!: string; + + @IsNotEmpty() + @IsString() + userId!: string; + + @IsOptional() + @IsUUID() + responseToId?: string; + + @IsOptional() + metadata?: Record; +} + +export class TypingDto { + @IsNotEmpty() + @IsUUID() + sessionId!: string; + + @IsNotEmpty() + @IsString() + userId!: string; +} diff --git a/src/modules/chatbot/chatbot.controller.ts b/src/modules/chatbot/chatbot.controller.ts new file mode 100644 index 0000000..d6624e5 --- /dev/null +++ b/src/modules/chatbot/chatbot.controller.ts @@ -0,0 +1,148 @@ +import { Body, Controller, Get, Param, Post, Put, Query, Res } from "@nestjs/common"; +import { ApiOperation, ApiResponse } from "@nestjs/swagger"; +import { FastifyReply } from "fastify"; + +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() +export class ChatbotController { + constructor(private chatbotService: ChatbotService) {} + + @Post("sessions") + @ApiOperation({ summary: "Create a new chat session" }) + @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" }) + 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" }) + 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" }) + sendMessage(@UserDec("id") userId: string, @Body() sendDto: SendMessageDto) { + return this.chatbotService.sendMessage(userId, sendDto); + } + + @Post("messages/stream") + @ApiOperation({ summary: "Send a message and get streaming response (Enhanced with LangChain + Fallback)" }) + @ApiResponse({ status: 200, description: "Enhanced streaming response initiated" }) + async sendMessageStream(@UserDec("id") userId: string, @Body() sendDto: SendMessageDto, @Res() res: FastifyReply): Promise { + try { + // Now uses LangChain by default with automatic Gemini fallback + const { userMessage, streamGenerator } = await this.chatbotService.sendMessageStream(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, enhanced: true })}\n\n`); + + // Start streaming bot response + res.raw.write(`data: ${JSON.stringify({ type: "bot_response_start", provider: "enhanced" })}\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 با خطا مواجه شد و به Gemini بازگشت ⚠️" })}\n\n`, + ); + res.raw.end(); + } + } + + @Post("training/refresh") + @ApiOperation({ summary: "Refresh LangChain training data from database" }) + @ApiResponse({ status: 200, description: "Training data refreshed successfully" }) + 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" }) + 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" }) + 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 new file mode 100644 index 0000000..1398d2a --- /dev/null +++ b/src/modules/chatbot/chatbot.gateway.ts @@ -0,0 +1,539 @@ +import { Logger, UseFilters, UseGuards, UsePipes, ValidationPipe } from "@nestjs/common"; +import { + ConnectedSocket, + MessageBody, + OnGatewayConnection, + OnGatewayDisconnect, + OnGatewayInit, + SubscribeMessage, + WebSocketGateway, + WebSocketServer, +} from "@nestjs/websockets"; +import { Server, Socket } from "socket.io"; + +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 { 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", +}) +@UsePipes(new ValidationPipe({ transform: true })) +@UseFilters(WsExceptionFilter) +export class ChatbotGateway implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect { + @WebSocketServer() + server: Server; + + private readonly logger = new Logger(ChatbotGateway.name); + + // Connection state management + private readonly connectedUsers = new Map(); + private readonly userSessions = new Map>(); + + constructor( + private readonly chatbotService: ChatbotService, + private readonly authService: WebSocketAuthService, + ) {} + + /** + * Gateway initialization lifecycle hook + */ + afterInit(server: Server): void { + this.logger.log("WebSocket Gateway initialized", { + namespace: "/chat", + cors: true, + serverInstance: !!server, + }); + } + + /** + * 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 { + const authResult = await this.authService.authenticateClient(client); + + if (!authResult.success) { + this.authService.handleAuthenticationFailure(client, authResult.error!); + return; + } + + const user = authResult.user!; + this.registerUserConnection(client.id, user.id); + this.authService.emitAuthenticationSuccess(client, user); + + this.logger.log("WebSocket connection established", { + ...connectionContext, + userId: user.id, + authenticated: true, + }); + } catch (error) { + this.logger.error("WebSocket connection error", { + ...connectionContext, + error: error instanceof Error ? error.message : String(error), + }); + + this.authService.handleAuthenticationFailure(client, WebSocketMessage.CONNECTION_FAILED); + } + } + + /** + * 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.cleanupUserConnection(client, userInfo); + } + + this.connectedUsers.delete(client.id); + } + + // ============================================================================ + // MESSAGE HANDLERS + // ============================================================================ + + /** + * Handles explicit authentication requests (for re-authentication) + */ + @SubscribeMessage(WEBSOCKET_EVENTS.AUTHENTICATE) + async handleAuthenticate(@ConnectedSocket() client: AuthenticatedSocket, @MessageBody() _data: AuthenticateDto): Promise { + try { + 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.handleMessageError(client, error, WEBSOCKET_EVENTS.AUTHENTICATE); + } + } + + /** + * Handles chat session creation + */ + @SubscribeMessage(WEBSOCKET_EVENTS.CREATE_SESSION) + @UseGuards(WebSocketAuthGuard) + async handleCreateSession(@ConnectedSocket() client: AuthenticatedSocket, @MessageBody() data: CreateSessionDto): Promise { + try { + const userId = this.authService.getUserId(client); + if (!userId) { + throw new Error(WebSocketMessage.USER_NOT_AUTHENTICATED); + } + + const session = await this.chatbotService.createChatSession(userId, { title: data.title }); + + this.emitSuccessResponse(client, WEBSOCKET_EVENTS.SESSION_CREATED, { + message: WebSocketMessage.SESSION_CREATED, + session, + }); + + this.logger.log("Chat session created", { + sessionId: session.id, + userId, + title: data.title, + }); + } catch (error) { + this.handleMessageError(client, error, WEBSOCKET_EVENTS.SESSION_ERROR); + this.emitErrorResponse(client, WEBSOCKET_EVENTS.SESSION_ERROR, { + message: WebSocketMessage.SESSION_CREATION_FAILED, + }); + } + } + + /** + * Handles joining a chat session + */ + @SubscribeMessage(WEBSOCKET_EVENTS.JOIN_CHAT) + @UseGuards(WebSocketAuthGuard) + async handleJoinChat(@ConnectedSocket() client: AuthenticatedSocket, @MessageBody() data: JoinChatDto): Promise { + try { + const userId = this.authService.getUserId(client); + if (!userId) { + throw new Error(WebSocketMessage.USER_NOT_AUTHENTICATED); + } + + // Verify session ownership + await this.chatbotService.getChatSession(data.sessionId, userId); + + // Leave previous session if any + await this.leavePreviousSession(client); + + // Join new session + await client.join(`session_${data.sessionId}`); + this.updateUserSession(client.id, userId, data.sessionId); + + // 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(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.handleMessageError(client, error, WEBSOCKET_EVENTS.JOIN_CHAT); + this.emitErrorResponse(client, WEBSOCKET_EVENTS.ERROR, { + message: WebSocketMessage.SESSION_NOT_FOUND, + }); + } + } + + /** + * Handles leaving a chat session + */ + @SubscribeMessage(WEBSOCKET_EVENTS.LEAVE_CHAT) + @UseGuards(WebSocketAuthGuard) + async handleLeaveChat(@ConnectedSocket() client: AuthenticatedSocket, @MessageBody() data: LeaveChatDto): Promise { + try { + const userId = this.authService.getUserId(client); + if (!userId) { + throw new Error(WebSocketMessage.USER_NOT_AUTHENTICATED); + } + + await client.leave(`session_${data.sessionId}`); + this.clearUserSession(client.id); + + // Emit success events + this.emitSuccessResponse(client, WEBSOCKET_EVENTS.CHAT_LEFT, { + sessionId: data.sessionId, + }); + + // Notify others in session + client.to(`session_${data.sessionId}`).emit(WEBSOCKET_EVENTS.USER_LEFT, { + userId, + sessionId: data.sessionId, + }); + + this.logger.log("User left chat session", { + userId, + sessionId: data.sessionId, + clientId: client.id, + }); + } catch (error) { + this.handleMessageError(client, error, WEBSOCKET_EVENTS.LEAVE_CHAT); + } + } + + /** + * Handles sending messages in chat + */ + @SubscribeMessage(WEBSOCKET_EVENTS.SEND_MESSAGE) + @UseGuards(WebSocketAuthGuard) + async handleSendMessage(@ConnectedSocket() client: AuthenticatedSocket, @MessageBody() data: SendMessageWebSocketDto): Promise { + try { + const userId = this.authService.getUserId(client); + if (!userId) { + throw new Error(WebSocketMessage.USER_NOT_AUTHENTICATED); + } + + // Start typing indicator + this.emitTypingIndicator(data.sessionId, "bot", true); + + // Send message through service + const userMessage = await this.chatbotService.sendMessage(userId, { + sessionId: data.sessionId, + content: data.content, + responseToId: data.responseToId, + metadata: data.metadata, + }); + + // Emit user message confirmation + this.server.to(`session_${data.sessionId}`).emit(WEBSOCKET_EVENTS.MESSAGE_RECEIVED, { + message: userMessage, + sessionId: data.sessionId, + enhanced: true, + }); + + this.logger.log("Message sent", { + userId, + sessionId: data.sessionId, + messageId: userMessage.id, + }); + + // Generate bot response asynchronously + this.generateBotResponseAsync(data.sessionId, userId, data.content); + } catch (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); + } + } + + /** + * 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, + }); + } + } + + /** + * 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}`); + } + } + + /** + * 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, + }); + } + + /** + * Generates bot response asynchronously + */ + private async generateBotResponseAsync(sessionId: string, userId: string, userMessage: string): Promise { + try { + // Start bot response + this.server.to(`session_${sessionId}`).emit(WEBSOCKET_EVENTS.BOT_RESPONSE_START, { + sessionId, + provider: "enhanced", + }); + + // Get streaming response + const { streamGenerator } = await this.chatbotService.sendMessageStream(userId, { + sessionId, + content: userMessage, + }); + + const stream = await streamGenerator(); + + // Stream chunks to client + for await (const chunk of stream) { + if (chunk) { + this.server.to(`session_${sessionId}`).emit(WEBSOCKET_EVENTS.BOT_RESPONSE_CHUNK, { + data: chunk, + sessionId, + }); + } + } + + // End bot response + this.server.to(`session_${sessionId}`).emit(WEBSOCKET_EVENTS.BOT_RESPONSE_END, { sessionId }); + + this.emitTypingIndicator(sessionId, "bot", false); + + this.logger.log("Bot response completed", { sessionId, userId }); + } catch (error) { + this.logger.error("Bot response generation failed", { + sessionId, + userId, + error: error instanceof Error ? error.message : String(error), + }); + + this.server.to(`session_${sessionId}`).emit(WEBSOCKET_EVENTS.ERROR, { + message: WebSocketMessage.LLM_SERVICE_ERROR, + sessionId, + }); + + this.emitTypingIndicator(sessionId, "bot", false); + } + } + + /** + * 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, + }); + } + + /** + * Gets connected users in a session + */ + getSessionUsers(sessionId: string): string[] { + const users: string[] = []; + this.connectedUsers.forEach((userInfo) => { + if (userInfo.sessionId === sessionId) { + users.push(userInfo.userId); + } + }); + return [...new Set(users)]; + } + + /** + * 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 new file mode 100644 index 0000000..43ee37b --- /dev/null +++ b/src/modules/chatbot/chatbot.module.ts @@ -0,0 +1,23 @@ +import { MikroOrmModule } from "@mikro-orm/nestjs"; +import { Module } from "@nestjs/common"; + +import { ChatbotController } from "./chatbot.controller"; +import { ChatbotGateway } from "./chatbot.gateway"; +import { ChatMessage } from "./entities/chat-message.entity"; +import { ChatSession } from "./entities/chat-session.entity"; +import { WebSocketAuthGuard } from "./guards/websocket-auth.guard"; +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]), AuthModule, UsersModule], + controllers: [ChatbotController], + 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 new file mode 100644 index 0000000..5eec06c --- /dev/null +++ b/src/modules/chatbot/constants/chatbot.constants.ts @@ -0,0 +1,340 @@ +export const CHATBOT_CONSTANTS = { + DEFAULT_MODEL: "gemini-2.0-flash", + DEFAULT_TEMPERATURE: 0.7, + DEFAULT_MAX_TOKENS: 2000, + MAX_CONVERSATION_HISTORY: 20, + + // General system prompt focused on company guidance + SYSTEM_PROMPT: `شما یک دستیار هوشمند AI برای پلتفرم DZone هستید که در راهنمایی کاربران درباره شرکت‌ها و صنایع تخصص دارید. شما باید تمامی پاسخ‌ها را به زبان فارسی ارائه دهید. + +## وظایف اصلی شما: +🏢 **راهنمایی شرکت‌ها**: کمک به کاربران برای یافتن شرکت‌های مناسب و اطلاعات تماس آن‌ها +🏭 **معرفی صنایع**: ارائه اطلاعات کامل درباره صنایع مختلف و شرکت‌های فعال در هر حوزه +💼 **محصولات و خدمات**: راهنمایی درباره محصولات و خدمات ارائه شده توسط شرکت‌ها +🔍 **جستجو و مقایسه**: کمک به انتخاب بهترین شرکت برای نیازهای خاص کاربران +📞 **اطلاعات تماس**: ارائه اطلاعات کامل تماس شامل تلفن، ایمیل و آدرس شرکت‌ها + +## اصول پاسخگویی: +✅ همیشه به زبان فارسی پاسخ دهید +✅ از اطلاعات پایگاه داده شرکت‌ها و صنایع برای پاسخ‌های دقیق استفاده کنید +✅ در صورت سوال درباره شرکت خاص، اطلاعات کامل آن را ارائه دهید +✅ همیشه اطلاعات تماس (تلفن، ایمیل، آدرس) شرکت‌ها را اضافه کنید +✅ اگر چیزی نمی‌دانید، صادقانه اعتراف کنید و راه حل جایگزین پیشنهاد دهید +✅ از ایموجی‌ها برای بهتر کردن تجربه کاربر استفاده کنید +✅ پاسخ‌های خود را ساختاربندی و خوانا ارائه دهید + +## قالب پاسخ برای شرکت‌ها: +📍 **نام شرکت**: [نام کامل] +📞 **تلفن**: [شماره تماس] +📧 **ایمیل**: [آدرس ایمیل] +🏠 **آدرس**: [آدرس کامل] +🏭 **صنعت**: [نوع صنعت] +💼 **محصولات/خدمات**: [فهرست محصولات و خدمات] + +اگر اطلاعات شرکت‌ها یا صنایع مربوطه از پایگاه داده در دسترس است، حتماً از آن استفاده کرده و منبع آن را مشخص کنید.`, + + // New company guidance system prompt + COMPANY_GUIDANCE_SYSTEM_PROMPT: `شما یک دستیار هوشمند برای راهنمایی کاربران درباره شرکت‌ها و صنایع موجود در پلتفرم DZone هستید. + +## ماموریت شما: +- راهنمایی کاربران برای یافتن شرکت‌های مناسب +- ارائه اطلاعات کامل درباره شرکت‌ها، محصولات و خدمات +- معرفی صنایع مختلف و شرکت‌های فعال در هر حوزه +- کمک به انتخاب بهترین شرکت برای نیازهای کاربران + +## اطلاعات شرکت‌ها و صنایع: +{context} + +## تاریخچه مکالمه: +{conversation_history} + +## سوال کاربر: +{question} + +## دستورالعمل پاسخ: +بر اساس اطلاعات شرکت‌ها و صنایع موجود، پاسخی جامع و کاربردی به زبان فارسی ارائه دهید: + +**نکات مهم:** +- اگر کاربر درباره شرکت خاصی سوال می‌پرسد، اطلاعات کامل آن شرکت (نام، آدرس، تلفن، محصولات، خدمات) را ارائه دهید +- اگر سوال درباره صنعت خاصی است، شرکت‌های فعال در آن صنعت را معرفی کنید +- در صورت جستجوی محصول یا خدمت، شرکت‌های مرتبط را پیشنهاد دهید +- همیشه اطلاعات تماس (تلفن، ایمیل، آدرس) شرکت‌ها را ارائه دهید +- از ایموجی‌های مناسب برای بهتر شدن خوانایی استفاده کنید +- پاسخ خود را ساختاربندی و منظم ارائه دهید + +پاسخ شما:`, + + // Company data templates + COMPANY_DATA_TEMPLATES: { + COMPANY_INFO: `نام شرکت: {name} +مدیر عامل: {ceo} +ایمیل: {email} +تلفن: {phone} +شماره ثبت: {registrationNumber} +تاریخ تاسیس: {establishmentDate} +آدرس: {address} +وب‌سایت: {website} +توضیحات: {description} +صنعت فعالیت: {industry} +وضعیت: {status} +کسب و کار والد: {business} + +این شرکت در پلتفرم DZone ثبت شده و آماده ارائه خدمات و محصولات خود به مشتریان می‌باشد.`, + + COMPANY_PRODUCT: `محصول شرکت {companyName}: +نام محصول: {productTitle} +شرکت تولیدکننده: {companyName} +صنعت: {industry} +توضیحات شرکت: {companyDescription} +آدرس شرکت: {companyAddress} +تلفن تماس: {companyPhone} +ایمیل: {companyEmail} + +این محصول توسط شرکت {companyName} در صنعت {industry} تولید و ارائه می‌شود.`, + + COMPANY_SERVICE: `خدمات شرکت {companyName}: +نام خدمت: {serviceTitle} +شرکت ارائه‌دهنده: {companyName} +صنعت: {industry} +توضیحات شرکت: {companyDescription} +آدرس شرکت: {companyAddress} +تلفن تماس: {companyPhone} +ایمیل: {companyEmail} + +این خدمت توسط شرکت {companyName} در صنعت {industry} ارائه می‌شود.`, + + INDUSTRY_INFO: `صنعت: {title} +وضعیت: {status} +تعداد شرکت‌های فعال: {companiesCount} +کسب و کار والد: {business} + +شرکت‌های فعال در این صنعت: +{companiesList} + +این صنعت شامل شرکت‌هایی است که در حوزه {title} فعالیت می‌کنند و خدمات و محصولات مختلفی ارائه می‌دهند.`, + }, + + // Company guidance documents + COMPANY_GUIDANCE_DOCUMENTS: { + COMPANY_GUIDE: `راهنمای شرکت‌ها در پلتفرم DZone: + +🏢 **اطلاعات شرکت‌ها**: +در پلتفرم DZone شرکت‌های مختلفی در صنایع متنوع فعالیت می‌کنند. هر شرکت دارای اطلاعات کاملی شامل: +- نام و مشخصات کامل شرکت +- اطلاعات تماس (تلفن، ایمیل، آدرس) +- صنعت فعالیت و حوزه کاری +- محصولات و خدمات ارائه شده +- وضعیت فعالیت و مجوزهای لازم + +💼 **محصولات و خدمات**: +هر شرکت می‌تواند محصولات و خدمات متنوعی ارائه دهد: +- محصولات: کالاها و تولیدات شرکت +- خدمات: سرویس‌ها و راهکارهای ارائه شده +- اطلاعات تماس برای استعلام و سفارش +- جزئیات کامل درباره هر محصول و خدمت + +🏭 **دسته‌بندی صنایع**: +شرکت‌ها بر اساس صنعت فعالیت دسته‌بندی شده‌اند: +- هر صنعت شامل شرکت‌های مرتبط +- امکان جستجو بر اساس نوع صنعت +- مشاهده تمام شرکت‌های یک صنعت خاص +- اطلاعات کامل درباره حوزه فعالیت`, + + COMPANY_FAQ: `سوالات متداول درباره شرکت‌ها: + +❓ **چگونه اطلاعات یک شرکت را پیدا کنم؟** +می‌توانید با ذکر نام شرکت یا صنعت مورد نظر، اطلاعات کامل شامل آدرس، تلفن و محصولات آن را دریافت کنید. + +❓ **چه شرکت‌هایی در یک صنعت خاص فعالیت می‌کنند؟** +با ذکر نام صنعت، لیست کامل شرکت‌های فعال در آن حوزه را مشاهده خواهید کرد. + +❓ **چگونه با یک شرکت تماس بگیرم؟** +اطلاعات تماس شامل تلفن، ایمیل و آدرس تمام شرکت‌ها در پلتفرم موجود است. + +❓ **چه محصولات و خدماتی ارائه می‌شود؟** +هر شرکت لیست کاملی از محصولات و خدمات خود ارائه داده که قابل مشاهده است. + +❓ **چگونه شرکت مناسب برای نیاز خودم پیدا کنم؟** +با توصیف نیاز خود یا نوع محصول/خدمت مورد نظر، شرکت‌های مرتبط را معرفی خواهم کرد.`, + + SEARCH_GUIDE: `راهنمای جستجو و یافتن شرکت‌ها: + +🔍 **نحوه جستجو**: +- جستجو بر اساس نام شرکت +- جستجو بر اساس نوع صنعت +- جستجو بر اساس محصول یا خدمت +- جستجو بر اساس منطقه جغرافیایی + +📊 **اطلاعات قابل دریافت**: +- اطلاعات کامل شرکت (نام، آدرس، تلفن) +- لیست محصولات و توضیحات +- لیست خدمات ارائه شده +- صنعت و حوزه فعالیت +- وضعیت فعالیت شرکت + +🎯 **توصیه‌های مفید**: +- برای یافتن شرکت مناسب، نیاز خود را واضح بیان کنید +- از کلمات کلیدی مرتبط با صنعت استفاده کنید +- در صورت نیاز، اطلاعات تماس برای پیگیری بیشتر درخواست کنید +- برای مقایسه، چندین شرکت در یک صنعت را بررسی کنید`, + + INDUSTRY_GUIDE: `راهنمای صنایع و حوزه‌های فعالیت: + +🏭 **انواع صنایع موجود**: +در پلتفرم DZone شرکت‌های مختلفی در صنایع متنوع حضور دارند: +- صنایع تولیدی و ساختمانی +- خدمات فناوری اطلاعات +- صنایع غذایی و کشاورزی +- خدمات مالی و بیمه +- حمل و نقل و لجستیک +- آموزش و تربیت +- سلامت و درمان +- و سایر صنایع + +💡 **مزایای دسته‌بندی صنعتی**: +- یافتن آسان شرکت‌های مرتبط +- مقایسه خدمات در یک صنعت +- انتخاب بهترین گزینه برای نیاز شما +- دسترسی به اطلاعات تخصصی هر حوزه + +📈 **نحوه استفاده**: +- ابتدا صنعت مورد نظر خود را مشخص کنید +- سپس شرکت‌های فعال در آن صنعت را بررسی کنید +- محصولات و خدمات مناسب را انتخاب کنید +- برای اطلاعات بیشتر با شرکت تماس بگیرید`, + }, + + // Logging messages for LangChain service + LANGCHAIN_MESSAGES: { + LOADING_DATA: "Loading company and industry data for model training...", + NO_DATA_FOUND: "No data found for model training", + DOCUMENTS_SPLIT: "{totalDocs} documents split into {chunks} chunks", + VECTOR_STORE_READY: "Vector store successfully initialized - ready to guide users about companies and industries", + VECTOR_STORE_ERROR: "Error initializing vector store", + DOCUMENTS_LOADED: "{count} documents loaded from company and industry database", + LOADING_ERROR: "Error loading company and industry documents", + SERVICE_NOT_INITIALIZED: "LangChain service is not initialized", + RESPONSE_ERROR: "Error generating LangChain response", + STREAM_RESPONSE_ERROR: "Error generating LangChain stream response", + REFRESHING_VECTOR_STORE: "Refreshing vector store with new company and industry data...", + }, + + // Source type labels in Farsi + SOURCE_TYPE_LABELS: { + company: "شرکت", + product: "محصول", + service: "خدمت", + industry: "صنعت", + company_guide: "راهنمای شرکت‌ها", + company_faq: "سوالات متداول", + search_guide: "راهنمای جستجو", + industry_guide: "راهنمای صنایع", + }, + + // Default values for missing data + DEFAULT_VALUES: { + NO_WEBSITE: "ندارد", + UNSPECIFIED: "مشخص نشده", + ACTIVE: "فعال", + INACTIVE: "غیرفعال", + }, + + // WEBSOCKET_EVENTS: { + // // Connection 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", + // }, + + ERROR_MESSAGES: { + SESSION_NOT_FOUND: "جلسه چت یافت نشد", + UNAUTHORIZED_ACCESS: "دسترسی غیرمجاز به جلسه چت", + MESSAGE_TOO_LONG: "پیام از حداکثر طول مجاز تجاوز کرده است", + RATE_LIMIT_EXCEEDED: "تعداد پیام‌های ارسالی بیش از حد مجاز است", + LLM_SERVICE_ERROR: "سرویس هوش مصنوعی موقتاً در دسترس نیست", + INVALID_TOKEN: "توکن احراز هویت نامعتبر است", + AUTHENTICATION_REQUIRED: "احراز هویت ضروری است", + SESSION_CREATION_FAILED: "ایجاد جلسه چت با شکست مواجه شد", + CONNECTION_FAILED: "اتصال برقرار نشد", + WEBSOCKET_ERROR: "خطا در ارتباط WebSocket", + }, + + SUCCESS_MESSAGES: { + SESSION_CREATED: "جلسه چت با موفقیت ایجاد شد", + CHAT_JOINED: "با موفقیت به چت متصل شدید", + MESSAGE_SENT: "پیام ارسال شد", + 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/entities/chat-message.entity.ts b/src/modules/chatbot/entities/chat-message.entity.ts new file mode 100644 index 0000000..ac5eb46 --- /dev/null +++ b/src/modules/chatbot/entities/chat-message.entity.ts @@ -0,0 +1,48 @@ +import { Entity, EntityRepositoryType, Enum, ManyToOne, Opt, Property } from "@mikro-orm/core"; + +import { ChatSession } from "./chat-session.entity"; +import { BaseEntity } from "../../../common/entities/base.entity"; +import { User } from "../../users/entities/user.entity"; +import { ChatMessageRepository } from "../repositories/chat-message.repository"; + +export enum MessageType { + USER = "user", + BOT = "bot", + SYSTEM = "system", +} + +export enum MessageStatus { + SENT = "sent", + DELIVERED = "delivered", + READ = "read", + FAILED = "failed", +} + +@Entity({ repository: () => ChatMessageRepository }) +export class ChatMessage extends BaseEntity { + @Property({ type: "text" }) + content!: string; + + @Enum({ items: () => MessageType, nativeEnumName: "message_type" }) + type!: MessageType; + + @Enum({ items: () => MessageStatus, nativeEnumName: "message_status", default: MessageStatus.SENT }) + status: MessageStatus & Opt; + + @ManyToOne(() => ChatSession) + session!: ChatSession; + + @ManyToOne(() => User, { nullable: true }) + sender?: User; + + @Property({ type: "json", nullable: true }) + metadata?: Record; + + @Property({ nullable: true }) + responseToId?: string; + + @Property({ default: 0 }) + tokensUsed: number = 0; + + [EntityRepositoryType]?: ChatMessageRepository; +} diff --git a/src/modules/chatbot/entities/chat-session.entity.ts b/src/modules/chatbot/entities/chat-session.entity.ts new file mode 100644 index 0000000..2973b44 --- /dev/null +++ b/src/modules/chatbot/entities/chat-session.entity.ts @@ -0,0 +1,35 @@ +import { Collection, Entity, EntityRepositoryType, Enum, ManyToOne, OneToMany, Opt, Property } from "@mikro-orm/core"; + +import { ChatMessage } from "./chat-message.entity"; +import { BaseEntity } from "../../../common/entities/base.entity"; +import { User } from "../../users/entities/user.entity"; +import { ChatSessionRepository } from "../repositories/chat-session.repository"; + +export enum ChatSessionStatus { + ACTIVE = "active", + CLOSED = "closed", + ARCHIVED = "archived", +} + +@Entity({ repository: () => ChatSessionRepository }) +export class ChatSession extends BaseEntity { + @Property({ type: "varchar" }) + title!: string; + + @ManyToOne(() => User) + user!: User; + + @Enum({ items: () => ChatSessionStatus, nativeEnumName: "chat_session_status", default: ChatSessionStatus.ACTIVE }) + status: ChatSessionStatus & Opt; + + @OneToMany(() => ChatMessage, (chatMessage) => chatMessage.session) + messages = new Collection(this); + + @Property({ type: "json", nullable: true }) + context?: Record; + + @Property({ nullable: true }) + lastMessageAt?: Date; + + [EntityRepositoryType]?: ChatSessionRepository; +} diff --git a/src/modules/chatbot/guards/websocket-auth.guard.ts b/src/modules/chatbot/guards/websocket-auth.guard.ts new file mode 100644 index 0000000..6e53905 --- /dev/null +++ b/src/modules/chatbot/guards/websocket-auth.guard.ts @@ -0,0 +1,55 @@ +import { CanActivate, ExecutionContext, Injectable, Logger } from "@nestjs/common"; +import { JwtService } from "@nestjs/jwt"; +import { WsException } from "@nestjs/websockets"; +import { Socket } from "socket.io"; + +import { WebSocketMessage } from "../../../common/enums/message.enum"; +import { WEBSOCKET_EVENTS } from "../constants/chatbot.constants"; + +@Injectable() +export class WebSocketAuthGuard implements CanActivate { + private readonly logger = new Logger(WebSocketAuthGuard.name); + + constructor(private jwtService: JwtService) {} + + async canActivate(context: ExecutionContext): Promise { + const client: Socket = context.switchToWs().getClient(); + + try { + // Get token from auth object or handshake query + const token = this.extractTokenFromHeader(client); + + if (!token) { + this.logger.warn("No token provided in WebSocket connection"); + throw new WsException({ + event: WEBSOCKET_EVENTS.UNAUTHORIZED, + message: WebSocketMessage.AUTHENTICATION_REQUIRED, + }); + } + + // Verify JWT token + const payload = await this.jwtService.verifyAsync(token); + + client.data.user = payload; + this.logger.log(`WebSocket authentication successful for user: ${payload.id}`); + + return true; + } catch (error) { + this.logger.error("WebSocket authentication failed:", error); + + client.emit(WEBSOCKET_EVENTS.UNAUTHORIZED, { + message: WebSocketMessage.INVALID_TOKEN, + }); + + throw new WsException({ + event: WEBSOCKET_EVENTS.UNAUTHORIZED, + message: WebSocketMessage.INVALID_TOKEN, + }); + } + } + + private extractTokenFromHeader(client: Socket): string | undefined { + const authToken = client.handshake?.auth?.token || client.handshake?.query?.token; + return authToken; + } +} diff --git a/src/modules/chatbot/interfaces/chatbot.interface.ts b/src/modules/chatbot/interfaces/chatbot.interface.ts new file mode 100644 index 0000000..af11748 --- /dev/null +++ b/src/modules/chatbot/interfaces/chatbot.interface.ts @@ -0,0 +1,36 @@ +export interface IChatbotResponse { + message: string; + confidence: number; + sources?: string[]; + tokensUsed: number; + context?: Record; +} + +export interface IChatContext { + userId: string; + sessionId: string; + conversationHistory: IChatMessage[]; + userPreferences?: Record; + businessContext?: Record; +} + +export interface IChatMessage { + content: string; + type: "user" | "bot" | "system"; + timestamp: Date; + metadata?: Record; +} + +export interface ILLMConfig { + model: string; + temperature: number; + maxTokens: number; + topP: number; + apiKey: string; +} + +export interface IDataSource { + query: (question: string, context?: Record) => Promise; + type: "database" | "vector" | "api"; + name: string; +} 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 new file mode 100644 index 0000000..b9321a2 --- /dev/null +++ b/src/modules/chatbot/providers/chatbot.service.ts @@ -0,0 +1,495 @@ +import { EntityManager } from "@mikro-orm/postgresql"; +import { BadRequestException, Injectable, Logger, NotFoundException } from "@nestjs/common"; +import { ConfigService } from "@nestjs/config"; + +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 { CreateChatSessionDto } from "../DTO/create-chat-session.dto"; +import { SendMessageDto } from "../DTO/send-message.dto"; +import { ChatMessage, MessageStatus, MessageType } from "../entities/chat-message.entity"; +import { ChatSession, ChatSessionStatus } from "../entities/chat-session.entity"; +import { IChatContext } from "../interfaces/chatbot.interface"; +import { ChatMessageRepository } from "../repositories/chat-message.repository"; +import { ChatSessionRepository } from "../repositories/chat-session.repository"; + +export enum LLMProvider { + LANGCHAIN = "langchain", + GEMINI = "gemini", +} + +@Injectable() +export class ChatbotService { + private readonly logger = new Logger(ChatbotService.name); + private defaultProvider: LLMProvider; + + constructor( + private em: EntityManager, + private llmService: LLMService, + private langChainService: LangChainService, + private chatSessionRepo: ChatSessionRepository, + private chatMessageRepo: ChatMessageRepository, + private configService: ConfigService, + // private usersService: UsersService, + ) { + // Configure default provider - LangChain for enhanced responses + this.defaultProvider = this.configService.get("DEFAULT_LLM_PROVIDER") === "gemini" ? LLMProvider.GEMINI : LLMProvider.LANGCHAIN; + + this.logger.log(`Using ${this.defaultProvider} as default LLM provider`); + } + + async createChatSession(userId: string, createDto: CreateChatSessionDto) { + const em = this.em.fork(); + const user = await em.findOne(User, { id: userId }); + if (!user) { + throw new NotFoundException("User not found"); + } + + const session = new ChatSession(); + session.title = createDto.title; + session.user = user; + session.context = createDto.context || {}; + session.lastMessageAt = new Date(); + + await em.persistAndFlush(session); + + return this.mapSessionToDto(session); + } + //************************************ */ + 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) { + const em = this.em.fork(); + const session = await em.findOne(ChatSession, { id: sessionId, user: userId }, { populate: ["messages", "messages.sender"] }); + + if (!session) { + throw new NotFoundException("Chat session not found"); + } + + return this.mapSessionToDto(session); + } + //************************************ */ + + async sendMessage(userId: string, sendDto: SendMessageDto, provider?: LLMProvider) { + const selectedProvider = provider || this.defaultProvider; + + // Use LangChain by default for enhanced responses + if (selectedProvider === LLMProvider.LANGCHAIN) { + try { + return await this.sendMessageWithLangChain(userId, sendDto); + } catch (error) { + this.logger.error("LangChain failed, falling back to Gemini", error); + // Fallback to Gemini service + return await this.sendMessageWithGemini(userId, sendDto); + } + } else { + return await this.sendMessageWithGemini(userId, sendDto); + } + } + + //************************************ */ + + 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"] }); + + if (!session) { + throw new NotFoundException("Chat session not found"); + } + + if (session.status !== ChatSessionStatus.ACTIVE) { + throw new BadRequestException("Cannot send message to inactive session"); + } + + const user = await em.findOne(User, { id: userId }); + if (!user) { + throw new NotFoundException("User not found"); + } + + // Create user message + const userMessage = new ChatMessage(); + userMessage.content = sendDto.content; + userMessage.type = MessageType.USER; + userMessage.session = session; + userMessage.sender = user; + userMessage.responseToId = sendDto.responseToId; + userMessage.metadata = sendDto.metadata; + + await em.persistAndFlush(userMessage); + + // Update session last message time + await this.chatSessionRepo.updateLastMessageTime(session.id, em); + + // Generate bot response asynchronously using original Gemini service + this.generateBotResponse(session, userMessage, userId); + + return this.mapMessageToDto(userMessage); + } + + //************************************ */ + + async sendMessageStream(userId: string, sendDto: SendMessageDto, provider?: LLMProvider) { + const selectedProvider = provider || this.defaultProvider; + + // Use LangChain by default for enhanced streaming responses + if (selectedProvider === LLMProvider.LANGCHAIN) { + try { + return await this.sendMessageStreamWithLangChain(userId, sendDto); + } catch (error) { + this.logger.error("LangChain streaming failed, falling back to Gemini", error); + // Fallback to Gemini service + return await this.sendMessageStreamWithGemini(userId, sendDto); + } + } else { + return await this.sendMessageStreamWithGemini(userId, sendDto); + } + } + + //************************************ */ + + 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"] }); + + if (!session) { + throw new NotFoundException("Chat session not found"); + } + + if (session.status !== ChatSessionStatus.ACTIVE) { + throw new BadRequestException("Cannot send message to inactive session"); + } + + const user = await em.findOne(User, { id: userId }); + if (!user) { + throw new NotFoundException("User not found"); + } + + // Create user message + const userMessage = new ChatMessage(); + userMessage.content = sendDto.content; + userMessage.type = MessageType.USER; + userMessage.session = session; + userMessage.sender = user; + userMessage.responseToId = sendDto.responseToId; + userMessage.metadata = sendDto.metadata; + + await em.persistAndFlush(userMessage); + + // Update session last message time + await this.chatSessionRepo.updateLastMessageTime(session.id, em); + + // Get conversation history for context + const history = await this.chatMessageRepo.getConversationHistory(session.id, 20, em); + + // Build context for LLM + const context: IChatContext = { + userId, + sessionId: session.id, + conversationHistory: history.reverse().map((msg) => ({ + content: msg.content, + type: msg.type as "user" | "bot" | "system", + timestamp: msg.createdAt, + metadata: msg.metadata, + })), + userPreferences: session.context, + }; + + // Return a function that generates the stream using original LLM service + const streamGenerator = () => this.llmService.generateStreamResponse(sendDto.content, context); + + return { + userMessage: this.mapMessageToDto(userMessage), + streamGenerator, + }; + } + + //************************************ */ + + private async generateBotResponse(session: ChatSession, userMessage: ChatMessage, userId: string): Promise { + const em = this.em.fork(); + try { + // Get conversation history + const history = await this.chatMessageRepo.getConversationHistory(session.id, 20, em); + + // Build context for LLM + const context: IChatContext = { + userId, + sessionId: session.id, + conversationHistory: history.reverse().map((msg) => ({ + content: msg.content, + type: msg.type as "user" | "bot" | "system", + timestamp: msg.createdAt, + metadata: msg.metadata, + })), + userPreferences: session.context, + }; + + // Generate response using LLM + const llmResponse = await this.llmService.generateResponse(userMessage.content, context); + + // Create bot message + const botMessage = new ChatMessage(); + botMessage.content = llmResponse.message; + botMessage.type = MessageType.BOT; + botMessage.session = session; + botMessage.responseToId = userMessage.id; + botMessage.tokensUsed = llmResponse.tokensUsed; + botMessage.metadata = { + confidence: llmResponse.confidence, + sources: llmResponse.sources, + llmContext: llmResponse.context, + }; + + await em.persistAndFlush(botMessage); + + // Update session last message time + await this.chatSessionRepo.updateLastMessageTime(session.id, em); + + this.logger.log(`Generated bot response for session ${session.id}`); + } catch (error) { + this.logger.error(`Failed to generate bot response for session ${session.id}`, error); + + // Create error message + const errorMessage = new ChatMessage(); + errorMessage.content = "متأسفم، در حال حاضر مشکلی در پاسخگویی دارم. لطفاً دوباره تلاش کنید. 🙏"; + errorMessage.type = MessageType.BOT; + errorMessage.session = session; + errorMessage.responseToId = userMessage.id; + errorMessage.status = MessageStatus.FAILED; + + await em.persistAndFlush(errorMessage); + } + } + + //************************************ */ + + async closeChatSession(sessionId: string, userId: string) { + const em = this.em.fork(); + const session = await em.findOne(ChatSession, { id: sessionId, user: userId }); + + if (!session) { + throw new NotFoundException("Chat session not found"); + } + + session.status = ChatSessionStatus.CLOSED; + await em.persistAndFlush(session); + + return { message: "Chat session closed successfully" }; + } + + //************************************ */ + + async markMessagesAsRead(sessionId: string, userId: string, messageIds: string[]) { + const em = this.em.fork(); + const session = await em.findOne(ChatSession, { id: sessionId, user: userId }); + + if (!session) { + throw new NotFoundException("Chat session not found"); + } + + await this.chatMessageRepo.markAsRead(messageIds, em); + + return { message: "Messages marked as read successfully" }; + } + + //************************************ */ + + private mapSessionToDto(session: ChatSession) { + return { + id: session.id, + title: session.title, + status: session.status, + createdAt: session.createdAt, + lastMessageAt: session.lastMessageAt, + context: session.context, + messages: session.messages ? session.messages.getItems().map((msg) => this.mapMessageToDto(msg)) : [], + }; + } + + //************************************ */ + + private mapMessageToDto(message: ChatMessage) { + return { + id: message.id, + content: message.content, + type: message.type, + status: message.status, + createdAt: message.createdAt, + responseToId: message.responseToId, + metadata: message.metadata, + tokensUsed: message.tokensUsed, + }; + } + + //************************************ */ + + async sendMessageWithLangChain(userId: string, sendDto: SendMessageDto) { + const em = this.em.fork(); + const session = await em.findOne(ChatSession, { id: sendDto.sessionId, user: userId }, { populate: ["messages"] }); + + if (!session) { + throw new NotFoundException("Chat session not found"); + } + + if (session.status !== ChatSessionStatus.ACTIVE) { + throw new BadRequestException("Cannot send message to inactive session"); + } + + const user = await em.findOne(User, { id: userId }); + if (!user) { + throw new NotFoundException("User not found"); + } + + // Create user message + const userMessage = new ChatMessage(); + userMessage.content = sendDto.content; + userMessage.type = MessageType.USER; + userMessage.session = session; + userMessage.sender = user; + userMessage.responseToId = sendDto.responseToId; + userMessage.metadata = sendDto.metadata; + + await em.persistAndFlush(userMessage); + + // Update session last message time + await this.chatSessionRepo.updateLastMessageTime(session.id, em); + + // Generate bot response using LangChain + this.generateBotResponseWithLangChain(session, userMessage, userId); + + return this.mapMessageToDto(userMessage); + } + + //************************************ */ + + async sendMessageStreamWithLangChain(userId: string, sendDto: SendMessageDto) { + const em = this.em.fork(); + const session = await em.findOne(ChatSession, { id: sendDto.sessionId, user: userId }, { populate: ["messages"] }); + + if (!session) { + throw new NotFoundException("Chat session not found"); + } + + if (session.status !== ChatSessionStatus.ACTIVE) { + throw new BadRequestException("Cannot send message to inactive session"); + } + + const user = await em.findOne(User, { id: userId }); + if (!user) { + throw new NotFoundException("User not found"); + } + + // Create user message + const userMessage = new ChatMessage(); + userMessage.content = sendDto.content; + userMessage.type = MessageType.USER; + userMessage.session = session; + userMessage.sender = user; + userMessage.responseToId = sendDto.responseToId; + userMessage.metadata = sendDto.metadata; + + await em.persistAndFlush(userMessage); + + // Update session last message time + await this.chatSessionRepo.updateLastMessageTime(session.id, em); + + // Get conversation history for context + const history = await this.chatMessageRepo.getConversationHistory(session.id, 20, em); + + // Build context for LangChain + const context: IChatContext = { + userId, + sessionId: session.id, + conversationHistory: history.reverse().map((msg) => ({ + content: msg.content, + type: msg.type as "user" | "bot" | "system", + timestamp: msg.createdAt, + metadata: msg.metadata, + })), + userPreferences: session.context, + }; + + // Return a function that generates the stream using LangChain + const streamGenerator = () => this.langChainService.generateStreamResponse(sendDto.content, context); + + return { + userMessage: this.mapMessageToDto(userMessage), + streamGenerator, + }; + } + + //************************************ */ + + private async generateBotResponseWithLangChain(session: ChatSession, userMessage: ChatMessage, userId: string): Promise { + const em = this.em.fork(); + try { + // Get conversation history + const history = await this.chatMessageRepo.getConversationHistory(session.id, 20, em); + + // Build context for LangChain + const context: IChatContext = { + userId, + sessionId: session.id, + conversationHistory: history.reverse().map((msg) => ({ + content: msg.content, + type: msg.type as "user" | "bot" | "system", + timestamp: msg.createdAt, + metadata: msg.metadata, + })), + userPreferences: session.context, + }; + + // Generate response using LangChain + const langChainResponse = await this.langChainService.generateResponse(userMessage.content, context); + + // Create bot message + const botMessage = new ChatMessage(); + botMessage.content = langChainResponse.message; + botMessage.type = MessageType.BOT; + botMessage.session = session; + botMessage.responseToId = userMessage.id; + botMessage.tokensUsed = langChainResponse.tokensUsed; + botMessage.metadata = { + confidence: langChainResponse.confidence, + sources: langChainResponse.sources, + llmContext: langChainResponse.context, + provider: "langchain", + documentsRetrieved: langChainResponse.context?.documentsRetrieved || 0, + }; + + await em.persistAndFlush(botMessage); + + // Update session last message time + await this.chatSessionRepo.updateLastMessageTime(session.id, em); + + this.logger.log(`Generated LangChain bot response for session ${session.id}`); + } catch (error) { + this.logger.error(`Failed to generate LangChain bot response for session ${session.id}`, error); + + // Fallback to regular LLM service + this.logger.log(`Falling back to regular LLM service for session ${session.id}`); + await this.generateBotResponse(session, userMessage, userId); + } + } + + //************************************ */ + + 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/data-context.service.ts b/src/modules/chatbot/providers/data-context.service.ts new file mode 100644 index 0000000..00c6138 --- /dev/null +++ b/src/modules/chatbot/providers/data-context.service.ts @@ -0,0 +1,226 @@ +import { EntityManager } from "@mikro-orm/core"; +import { Injectable, Logger } from "@nestjs/common"; + +import { Business } from "../../businesses/entities/business.entity"; +import { Company } from "../../companies/entities/company.entity"; +import { Industry } from "../../industries/entities/industry.entity"; +import { Invoice } from "../../invoices/entities/invoice.entity"; +import { InvoiceStatus } from "../../invoices/enums/invoice-status.enum"; +import { Payment } from "../../payments/entities/payment.entity"; +import { PaymentStatus } from "../../payments/enums/payment-status.enum"; +import { Ticket } from "../../tickets/entities/ticket.entity"; +import { TicketStatus } from "../../tickets/enums/ticket-status.enum"; +import { IChatContext } from "../interfaces/chatbot.interface"; + +@Injectable() +export class DataContextService { + private readonly logger = new Logger(DataContextService.name); + + constructor(private em: EntityManager) {} + + async getRelevantContext(message: string, context: IChatContext): Promise<{ data: string[]; sources: string[] }> { + const relevantData: string[] = []; + const sources: string[] = []; + + try { + // Analyze message to determine what data to fetch + const keywords = this.extractKeywords(message); + + // Fetch relevant data based on keywords and user context + await Promise.all([ + this.getBusinessData(keywords, context.userId, relevantData, sources), + this.getInvoiceData(keywords, context.userId, relevantData, sources), + this.getPaymentData(keywords, context.userId, relevantData, sources), + this.getCompanyData(keywords, context.userId, relevantData, sources), + this.getTicketData(keywords, context.userId, relevantData, sources), + this.getIndustryData(keywords, relevantData, sources), + ]); + + return { data: relevantData, sources }; + } catch (error) { + this.logger.error("Failed to get relevant context", error); + return { data: [], sources: [] }; + } + } + + private extractKeywords(message: string): string[] { + const commonWords = ["the", "is", "at", "which", "on", "a", "an", "and", "or", "but", "in", "with", "to", "for", "of", "as", "by"]; + + return message + .toLowerCase() + .replace(/[^\w\s]/g, "") + .split(/\s+/) + .filter((word) => word.length > 2 && !commonWords.includes(word)); + } + + private async getBusinessData(keywords: string[], userId: string, relevantData: string[], sources: string[]): Promise { + if (this.hasBusinessKeywords(keywords)) { + try { + const businesses = await this.em.find(Business, { danakSubscriptionId: userId }, { limit: 5, populate: ["users"] }); + + businesses.forEach((business) => { + relevantData.push( + `Business: ${business.name} - Domain: ${business.domain || "No domain"} - Verified: ${business.isDomainVerified ? "Yes" : "No"}`, + ); + sources.push("Businesses Database"); + }); + } catch (error) { + this.logger.warn("Failed to fetch business data", error); + } + } + } + + private async getInvoiceData(keywords: string[], userId: string, relevantData: string[], sources: string[]): Promise { + if (this.hasInvoiceKeywords(keywords)) { + try { + // Find user's businesses first + const userBusinesses = await this.em.find(Business, { danakSubscriptionId: userId }); + + if (userBusinesses.length > 0) { + const businessIds = userBusinesses.map((b) => b.id); + const invoices = await this.em.find(Invoice, { business: { $in: businessIds } }, { limit: 10, orderBy: { createdAt: "DESC" } }); + + const summary = { + total: invoices.length, + paid: invoices.filter((inv) => inv.status === InvoiceStatus.PAID).length, + pending: invoices.filter((inv) => inv.status === InvoiceStatus.PENDING).length, + totalAmount: invoices.reduce((sum, inv) => sum + Number(inv.totalPrice), 0), + }; + + relevantData.push( + `Invoice Summary: Total invoices: ${summary.total}, Paid: ${summary.paid}, Pending: ${summary.pending}, Total amount: $${summary.totalAmount}`, + ); + sources.push("Invoices Database"); + } + } catch (error) { + this.logger.warn("Failed to fetch invoice data", error); + } + } + } + + private async getPaymentData(keywords: string[], userId: string, relevantData: string[], sources: string[]): Promise { + if (this.hasPaymentKeywords(keywords)) { + try { + // Find user's businesses first + const userBusinesses = await this.em.find(Business, { danakSubscriptionId: userId }); + + if (userBusinesses.length > 0) { + const businessIds = userBusinesses.map((b) => b.id); + const payments = await this.em.find(Payment, { business: { $in: businessIds } }, { limit: 10, orderBy: { createdAt: "DESC" } }); + + const summary = { + total: payments.length, + successful: payments.filter((pay) => pay.status === PaymentStatus.COMPLETED).length, + failed: payments.filter((pay) => pay.status === PaymentStatus.FAILED).length, + totalAmount: payments.reduce((sum, pay) => sum + Number(pay.amount), 0), + }; + + relevantData.push( + `Payment Summary: Total payments: ${summary.total}, Successful: ${summary.successful}, Failed: ${summary.failed}, Total amount: $${summary.totalAmount}`, + ); + sources.push("Payments Database"); + } + } catch (error) { + this.logger.warn("Failed to fetch payment data", error); + } + } + } + + private async getCompanyData(keywords: string[], userId: string, relevantData: string[], sources: string[]): Promise { + if (this.hasCompanyKeywords(keywords)) { + try { + // Find user's businesses first + const userBusinesses = await this.em.find(Business, { danakSubscriptionId: userId }); + + if (userBusinesses.length > 0) { + const businessIds = userBusinesses.map((b) => b.id); + const companies = await this.em.find(Company, { business: { $in: businessIds } }, { limit: 5, populate: ["industry"] }); + + companies.forEach((company) => { + relevantData.push(`Company: ${company.name} - Industry: ${company.industry?.title || "N/A"} - Status: ${company.status}`); + sources.push("Companies Database"); + }); + } + } catch (error) { + this.logger.warn("Failed to fetch company data", error); + } + } + } + + private async getTicketData(keywords: string[], userId: string, relevantData: string[], sources: string[]): Promise { + if (this.hasTicketKeywords(keywords)) { + try { + // Find user's businesses first + const userBusinesses = await this.em.find(Business, { danakSubscriptionId: userId }); + + if (userBusinesses.length > 0) { + const businessIds = userBusinesses.map((b) => b.id); + const tickets = await this.em.find( + Ticket, + { business: { $in: businessIds } }, + { limit: 10, orderBy: { createdAt: "DESC" }, populate: ["category"] }, + ); + + const summary = { + total: tickets.length, + pending: tickets.filter((t) => t.status === TicketStatus.PENDING).length, + answered: tickets.filter((t) => t.status === TicketStatus.ANSWERED).length, + closed: tickets.filter((t) => t.status === TicketStatus.CLOSED).length, + categories: [...new Set(tickets.map((t) => t.category?.title || "Uncategorized"))], + }; + + relevantData.push( + `Support Tickets: Total: ${summary.total}, Pending: ${summary.pending}, Answered: ${summary.answered}, Closed: ${summary.closed}, Categories: ${summary.categories.join(", ")}`, + ); + sources.push("Support Tickets Database"); + } + } catch (error) { + this.logger.warn("Failed to fetch ticket data", error); + } + } + } + + private async getIndustryData(keywords: string[], relevantData: string[], sources: string[]): Promise { + if (this.hasIndustryKeywords(keywords)) { + try { + const industries = await this.em.find(Industry, {}, { limit: 10 }); + + const industryInfo = industries.map((ind) => `${ind.title}: ${ind.isActive ? "Active" : "Inactive"}`); + relevantData.push(`Available Industries: ${industryInfo.join("; ")}`); + sources.push("Industries Database"); + } catch (error) { + this.logger.warn("Failed to fetch industry data", error); + } + } + } + + private hasBusinessKeywords(keywords: string[]): boolean { + const businessTerms = ["business", "company", "startup", "enterprise", "organization", "firm"]; + return keywords.some((keyword) => businessTerms.includes(keyword)); + } + + private hasInvoiceKeywords(keywords: string[]): boolean { + const invoiceTerms = ["invoice", "bill", "billing", "payment", "charge", "fee", "cost", "price"]; + return keywords.some((keyword) => invoiceTerms.includes(keyword)); + } + + private hasPaymentKeywords(keywords: string[]): boolean { + const paymentTerms = ["payment", "pay", "transaction", "money", "refund", "charge", "card", "bank"]; + return keywords.some((keyword) => paymentTerms.includes(keyword)); + } + + private hasCompanyKeywords(keywords: string[]): boolean { + const companyTerms = ["company", "corporation", "business", "organization", "enterprise"]; + return keywords.some((keyword) => companyTerms.includes(keyword)); + } + + private hasTicketKeywords(keywords: string[]): boolean { + const ticketTerms = ["ticket", "support", "help", "issue", "problem", "bug", "error"]; + return keywords.some((keyword) => ticketTerms.includes(keyword)); + } + + private hasIndustryKeywords(keywords: string[]): boolean { + const industryTerms = ["industry", "sector", "field", "domain", "market"]; + return keywords.some((keyword) => industryTerms.includes(keyword)); + } +} diff --git a/src/modules/chatbot/providers/langchain.service.ts b/src/modules/chatbot/providers/langchain.service.ts new file mode 100644 index 0000000..4bbc174 --- /dev/null +++ b/src/modules/chatbot/providers/langchain.service.ts @@ -0,0 +1,389 @@ +import { Document } from "@langchain/core/documents"; +import { StringOutputParser } from "@langchain/core/output_parsers"; +import { PromptTemplate } from "@langchain/core/prompts"; +import { RunnableSequence } from "@langchain/core/runnables"; +import { ChatGoogleGenerativeAI, GoogleGenerativeAIEmbeddings } from "@langchain/google-genai"; +import { EntityManager } from "@mikro-orm/core"; +import { Injectable, Logger, OnModuleInit } from "@nestjs/common"; +import { ConfigService } from "@nestjs/config"; +import { RecursiveCharacterTextSplitter } from "langchain/text_splitter"; +import { MemoryVectorStore } from "langchain/vectorstores/memory"; + +import { Company } from "../../companies/entities/company.entity"; +import { Industry } from "../../industries/entities/industry.entity"; +import { CHATBOT_CONSTANTS } from "../constants/chatbot.constants"; +import { IChatContext, IChatbotResponse } from "../interfaces/chatbot.interface"; + +@Injectable() +export class LangChainService implements OnModuleInit { + private readonly logger = new Logger(LangChainService.name); + private embeddings: GoogleGenerativeAIEmbeddings; + private llm: ChatGoogleGenerativeAI; + private vectorStore: MemoryVectorStore; + private textSplitter: RecursiveCharacterTextSplitter; + private isInitialized = false; + + constructor( + private em: EntityManager, + private configService: ConfigService, + ) { + // Initialize Google Gemini components + this.embeddings = new GoogleGenerativeAIEmbeddings({ + apiKey: this.configService.getOrThrow("GEMINI_API_KEY"), + modelName: "embedding-001", // Google's embedding model + }); + + this.llm = new ChatGoogleGenerativeAI({ + apiKey: this.configService.getOrThrow("GEMINI_API_KEY"), + model: CHATBOT_CONSTANTS.DEFAULT_MODEL, + temperature: CHATBOT_CONSTANTS.DEFAULT_TEMPERATURE, + maxOutputTokens: CHATBOT_CONSTANTS.DEFAULT_MAX_TOKENS, + }); + + this.textSplitter = new RecursiveCharacterTextSplitter({ + chunkSize: 1000, + chunkOverlap: 200, + separators: ["\n\n", "\n", ".", "!", "?", "؟", "!", ".", " ", ""], + }); + } + + async onModuleInit() { + await this.initializeVectorStore(); + } + + private async initializeVectorStore() { + try { + this.logger.log(CHATBOT_CONSTANTS.LANGCHAIN_MESSAGES.LOADING_DATA); + + // Load company and industry data for training + const documents = await this.loadCompanyAndIndustryDocuments(); + + if (documents.length === 0) { + this.logger.warn(CHATBOT_CONSTANTS.LANGCHAIN_MESSAGES.NO_DATA_FOUND); + this.vectorStore = new MemoryVectorStore(this.embeddings); + this.isInitialized = true; + return; + } + + // Split documents into chunks + const splitDocs = await this.textSplitter.splitDocuments(documents); + + this.logger.log( + CHATBOT_CONSTANTS.LANGCHAIN_MESSAGES.DOCUMENTS_SPLIT.replace("{totalDocs}", documents.length.toString()).replace( + "{chunks}", + splitDocs.length.toString(), + ), + ); + + // Create vector store from documents + this.vectorStore = await MemoryVectorStore.fromDocuments(splitDocs, this.embeddings); + + this.isInitialized = true; + this.logger.log(CHATBOT_CONSTANTS.LANGCHAIN_MESSAGES.VECTOR_STORE_READY); + } catch (error) { + this.logger.error(CHATBOT_CONSTANTS.LANGCHAIN_MESSAGES.VECTOR_STORE_ERROR, error); + // Fallback to empty vector store + this.vectorStore = new MemoryVectorStore(this.embeddings); + this.isInitialized = true; + } + } + + private async loadCompanyAndIndustryDocuments(): Promise { + const documents: Document[] = []; + const em = this.em.fork(); + + try { + // Load companies data with their products and services + const companies = await em.find(Company, { isActive: true, deletedAt: null }, { populate: ["industry", "business", "products", "services"] }); + + companies.forEach((company) => { + // Company basic information using template + const companyContent = this.replaceTemplate(CHATBOT_CONSTANTS.COMPANY_DATA_TEMPLATES.COMPANY_INFO, { + name: company.name, + ceo: company.chiefExecutiveOfficer, + email: company.email, + phone: company.phone, + registrationNumber: company.identificationNumber, + establishmentDate: new Date(company.dateOfEstablishment).toLocaleDateString("fa-IR"), + address: company.address, + website: company.websiteUrl || CHATBOT_CONSTANTS.DEFAULT_VALUES.NO_WEBSITE, + description: company.description, + industry: company.industry?.title || CHATBOT_CONSTANTS.DEFAULT_VALUES.UNSPECIFIED, + status: company.status, + business: company.business?.name || CHATBOT_CONSTANTS.DEFAULT_VALUES.UNSPECIFIED, + }); + + documents.push( + new Document({ + pageContent: companyContent, + metadata: { + type: "company", + id: company.id, + name: company.name, + industry: company.industry?.title, + businessId: company.business?.id, + status: company.status, + }, + }), + ); + + // Company products using template + if (company.products && company.products.getItems().length > 0) { + company.products.getItems().forEach((product) => { + const productContent = this.replaceTemplate(CHATBOT_CONSTANTS.COMPANY_DATA_TEMPLATES.COMPANY_PRODUCT, { + companyName: company.name, + productTitle: product.title, + industry: company.industry?.title || CHATBOT_CONSTANTS.DEFAULT_VALUES.UNSPECIFIED, + companyDescription: company.description, + companyAddress: company.address, + companyPhone: company.phone, + companyEmail: company.email, + }); + + documents.push( + new Document({ + pageContent: productContent, + metadata: { + type: "product", + id: product.id, + title: product.title, + companyId: company.id, + companyName: company.name, + industry: company.industry?.title, + }, + }), + ); + }); + } + + // Company services using template + if (company.services && company.services.getItems().length > 0) { + company.services.getItems().forEach((service) => { + const serviceContent = this.replaceTemplate(CHATBOT_CONSTANTS.COMPANY_DATA_TEMPLATES.COMPANY_SERVICE, { + companyName: company.name, + serviceTitle: service.title, + industry: company.industry?.title || CHATBOT_CONSTANTS.DEFAULT_VALUES.UNSPECIFIED, + companyDescription: company.description, + companyAddress: company.address, + companyPhone: company.phone, + companyEmail: company.email, + }); + + documents.push( + new Document({ + pageContent: serviceContent, + metadata: { + type: "service", + id: service.id, + title: service.title, + companyId: company.id, + companyName: company.name, + industry: company.industry?.title, + }, + }), + ); + }); + } + }); + + // Load industries data using template + const industries = await em.find(Industry, { isActive: true, deletedAt: null }, { populate: ["companies", "business"] }); + industries.forEach((industry) => { + const companiesInIndustry = industry.companies?.getItems().filter((c) => c.isActive && !c.deletedAt) || []; + + const industryContent = this.replaceTemplate(CHATBOT_CONSTANTS.COMPANY_DATA_TEMPLATES.INDUSTRY_INFO, { + title: industry.title, + status: industry.isActive ? CHATBOT_CONSTANTS.DEFAULT_VALUES.ACTIVE : CHATBOT_CONSTANTS.DEFAULT_VALUES.INACTIVE, + companiesCount: companiesInIndustry.length.toString(), + business: industry.business?.name || CHATBOT_CONSTANTS.DEFAULT_VALUES.UNSPECIFIED, + companiesList: companiesInIndustry.map((company) => `- ${company.name}`).join("\n"), + }); + + documents.push( + new Document({ + pageContent: industryContent, + metadata: { + type: "industry", + id: industry.id, + title: industry.title, + isActive: industry.isActive, + companiesCount: companiesInIndustry.length, + businessId: industry.business?.id, + }, + }), + ); + }); + + // Add comprehensive Farsi guidance documents from constants + const companyGuidanceDocuments = [ + new Document({ + pageContent: CHATBOT_CONSTANTS.COMPANY_GUIDANCE_DOCUMENTS.COMPANY_GUIDE, + metadata: { + type: "company_guide", + category: "guidance", + language: "فارسی", + }, + }), + new Document({ + pageContent: CHATBOT_CONSTANTS.COMPANY_GUIDANCE_DOCUMENTS.COMPANY_FAQ, + metadata: { + type: "company_faq", + category: "support", + language: "فارسی", + }, + }), + new Document({ + pageContent: CHATBOT_CONSTANTS.COMPANY_GUIDANCE_DOCUMENTS.SEARCH_GUIDE, + metadata: { + type: "search_guide", + category: "tutorial", + language: "فارسی", + }, + }), + new Document({ + pageContent: CHATBOT_CONSTANTS.COMPANY_GUIDANCE_DOCUMENTS.INDUSTRY_GUIDE, + metadata: { + type: "industry_guide", + category: "education", + language: "فارسی", + }, + }), + ]; + + documents.push(...companyGuidanceDocuments); + + this.logger.log(CHATBOT_CONSTANTS.LANGCHAIN_MESSAGES.DOCUMENTS_LOADED.replace("{count}", documents.length.toString())); + return documents; + } catch (error) { + this.logger.error(CHATBOT_CONSTANTS.LANGCHAIN_MESSAGES.LOADING_ERROR, error); + return []; + } + } + + async generateResponse(message: string, context: IChatContext): Promise { + if (!this.isInitialized) { + throw new Error(CHATBOT_CONSTANTS.LANGCHAIN_MESSAGES.SERVICE_NOT_INITIALIZED); + } + + try { + // Retrieve relevant documents based on the user's query + const relevantDocs = await this.vectorStore.similaritySearch(message, 5); + + // Build context from retrieved documents + const retrievedContext = relevantDocs.map((doc, index) => `${index + 1}. ${doc.pageContent}`).join("\n\n"); + + // Build conversation history + const conversationHistory = + context.conversationHistory + ?.slice(-CHATBOT_CONSTANTS.MAX_CONVERSATION_HISTORY / 4) + ?.map((msg) => { + const roleLabel = msg.type === "user" ? "کاربر" : msg.type === "bot" ? "ربات" : "سیستم"; + return `${roleLabel}: ${msg.content}`; + }) + ?.join("\n") || ""; + + // Create the prompt template using constants + const promptTemplate = PromptTemplate.fromTemplate(CHATBOT_CONSTANTS.COMPANY_GUIDANCE_SYSTEM_PROMPT); + + // Create the runnable sequence + const chain = RunnableSequence.from([promptTemplate, this.llm, new StringOutputParser()]); + + // Execute the chain + const response = await chain.invoke({ + context: retrievedContext, + conversation_history: conversationHistory, + question: message, + }); + + // Calculate approximate token usage + const tokensUsed = this.estimateTokens(retrievedContext) + this.estimateTokens(message) + this.estimateTokens(response); + + return { + message: response, + confidence: 0.9, // Higher confidence for company guidance + sources: relevantDocs.map((doc) => { + const type = doc.metadata.type || "شرکت"; + return CHATBOT_CONSTANTS.SOURCE_TYPE_LABELS[type as keyof typeof CHATBOT_CONSTANTS.SOURCE_TYPE_LABELS] || type; + }), + tokensUsed, + context: { + model: `${CHATBOT_CONSTANTS.DEFAULT_MODEL}-company-guide`, + relevantDataFound: relevantDocs.length > 0, + documentsRetrieved: relevantDocs.length, + vectorStoreInitialized: this.isInitialized, + language: "فارسی", + dataTypes: relevantDocs.map((doc) => doc.metadata.type).filter((type, index, arr) => arr.indexOf(type) === index), + focus: "company_and_industry_guidance", + }, + }; + } catch (error) { + this.logger.error(CHATBOT_CONSTANTS.LANGCHAIN_MESSAGES.RESPONSE_ERROR, error); + throw new Error(CHATBOT_CONSTANTS.ERROR_MESSAGES.LLM_SERVICE_ERROR); + } + } + + async generateStreamResponse(message: string, context: IChatContext) { + if (!this.isInitialized) { + throw new Error(CHATBOT_CONSTANTS.LANGCHAIN_MESSAGES.SERVICE_NOT_INITIALIZED); + } + + try { + // Retrieve relevant documents + const relevantDocs = await this.vectorStore.similaritySearch(message, 5); + const retrievedContext = relevantDocs.map((doc, index) => `${index + 1}. ${doc.pageContent}`).join("\n\n"); + + // Build conversation history + const conversationHistory = + context.conversationHistory + ?.slice(-CHATBOT_CONSTANTS.MAX_CONVERSATION_HISTORY / 4) + ?.map((msg) => { + const roleLabel = msg.type === "user" ? "کاربر" : msg.type === "bot" ? "ربات" : "سیستم"; + return `${roleLabel}: ${msg.content}`; + }) + ?.join("\n") || ""; + + // Create prompt template using constants + const promptTemplate = PromptTemplate.fromTemplate(CHATBOT_CONSTANTS.COMPANY_GUIDANCE_SYSTEM_PROMPT); + + // Create streaming chain + const chain = RunnableSequence.from([promptTemplate, this.llm, new StringOutputParser()]); + + // Return async iterable for streaming + const stream = await chain.stream({ + context: retrievedContext, + conversation_history: conversationHistory, + question: message, + }); + + return stream; + } catch (error) { + this.logger.error(CHATBOT_CONSTANTS.LANGCHAIN_MESSAGES.STREAM_RESPONSE_ERROR, error); + throw new Error(CHATBOT_CONSTANTS.ERROR_MESSAGES.LLM_SERVICE_ERROR); + } + } + + async refreshVectorStore(): Promise { + this.logger.log(CHATBOT_CONSTANTS.LANGCHAIN_MESSAGES.REFRESHING_VECTOR_STORE); + await this.initializeVectorStore(); + } + + private estimateTokens(text: string): number { + // Rough estimation: 1 token ≈ 4 characters for most languages + return Math.ceil(text.length / 4); + } + + /** + * Replace placeholders in template with actual values + */ + private replaceTemplate(template: string, values: Record): string { + let result = template; + Object.entries(values).forEach(([key, value]) => { + result = result.replace(new RegExp(`{${key}}`, "g"), value); + }); + return result; + } + + get initialized(): boolean { + return this.isInitialized; + } +} diff --git a/src/modules/chatbot/providers/llm.service.ts b/src/modules/chatbot/providers/llm.service.ts new file mode 100644 index 0000000..4640fa2 --- /dev/null +++ b/src/modules/chatbot/providers/llm.service.ts @@ -0,0 +1,237 @@ +import { GoogleGenerativeAI, HarmBlockThreshold, HarmCategory } from "@google/generative-ai"; +import { Injectable, Logger } from "@nestjs/common"; +import { ConfigService } from "@nestjs/config"; + +import { DataContextService } from "./data-context.service"; +import { CHATBOT_CONSTANTS } from "../constants/chatbot.constants"; +import { IChatContext, IChatbotResponse, ILLMConfig } from "../interfaces/chatbot.interface"; + +@Injectable() +export class LLMService { + private readonly logger = new Logger(LLMService.name); + private genAI: GoogleGenerativeAI; + private config: ILLMConfig; + + constructor( + private configService: ConfigService, + private dataContextService: DataContextService, + ) { + this.config = { + model: this.configService.get("GEMINI_MODEL", CHATBOT_CONSTANTS.DEFAULT_MODEL), + temperature: Number(this.configService.get("GEMINI_TEMPERATURE", CHATBOT_CONSTANTS.DEFAULT_TEMPERATURE)), + maxTokens: Number(this.configService.get("GEMINI_MAX_TOKENS", CHATBOT_CONSTANTS.DEFAULT_MAX_TOKENS)), + topP: Number(this.configService.get("GEMINI_TOP_P", 0.95)), + apiKey: this.configService.getOrThrow("GEMINI_API_KEY"), + }; + + this.genAI = new GoogleGenerativeAI(this.config.apiKey); + } + + async generateResponse(message: string, context: IChatContext): Promise { + try { + // Get relevant data from database + const relevantData = await this.dataContextService.getRelevantContext(message, context); + + // Build system instruction with context + const systemInstruction = this.buildSystemInstruction(relevantData); + + // Get the generative model with system instruction + const model = this.genAI.getGenerativeModel({ + model: this.config.model, + systemInstruction, + generationConfig: { + temperature: this.config.temperature, + maxOutputTokens: this.config.maxTokens, + topP: this.config.topP, + }, + safetySettings: [ + { + category: HarmCategory.HARM_CATEGORY_HARASSMENT, + threshold: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE, + }, + { + category: HarmCategory.HARM_CATEGORY_HATE_SPEECH, + threshold: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE, + }, + { + category: HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT, + threshold: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE, + }, + { + category: HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, + threshold: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE, + }, + ], + }); + + // Build conversation history + const conversationHistory = this.buildConversationHistory(context); + + // Generate response using conversation history or direct message + let result; + if (conversationHistory.length > 0) { + // Use chat session for multi-turn conversation + const chat = model.startChat({ + history: conversationHistory, + }); + result = await chat.sendMessage(message); + } else { + // Single message generation + result = await model.generateContent(message); + } + + const response = await result.response; + const botMessage = response.text() || "متأسفم، نمی‌توانم پاسخی تولید کنم. لطفاً سوال خود را دوباره مطرح کنید. 🤖"; + + // Calculate token usage (approximate) + const tokensUsed = this.estimateTokens(systemInstruction) + this.estimateTokens(message) + this.estimateTokens(botMessage); + + return { + message: botMessage, + confidence: this.calculateConfidence(response), + sources: relevantData.sources, + tokensUsed, + context: { + model: this.config.model, + relevantDataFound: relevantData.data.length > 0, + finishReason: response.candidates?.[0]?.finishReason || "unknown", + safetyRatings: response.candidates?.[0]?.safetyRatings || [], + }, + }; + } catch (error) { + this.logger.error("Failed to generate Gemini response", error); + throw new Error("Failed to generate response from AI service"); + } + } + + async generateStreamResponse(message: string, context: IChatContext): Promise> { + try { + // Get relevant data from database + const relevantData = await this.dataContextService.getRelevantContext(message, context); + + // Build system instruction with context + const systemInstruction = this.buildSystemInstruction(relevantData); + + // Get the generative model + const model = this.genAI.getGenerativeModel({ + model: this.config.model, + systemInstruction, + generationConfig: { + temperature: this.config.temperature, + maxOutputTokens: this.config.maxTokens, + topP: this.config.topP, + }, + }); + + // Build conversation history + const conversationHistory = this.buildConversationHistory(context); + + let streamResult; + if (conversationHistory.length > 0) { + const chat = model.startChat({ + history: conversationHistory, + }); + streamResult = await chat.sendMessageStream(message); + } else { + streamResult = await model.generateContentStream(message); + } + + return this.createAsyncIterableFromStream(streamResult); + } catch (error) { + this.logger.error("Failed to generate streaming Gemini response", error); + throw new Error("Failed to generate streaming response from AI service"); + } + } + + private buildSystemInstruction(relevantData: { data: string[]; sources: string[] }): string { + let instruction = CHATBOT_CONSTANTS.SYSTEM_PROMPT; + + // Add relevant data context + if (relevantData.data.length > 0) { + instruction += "\n\nRelevant information from our database:\n"; + instruction += relevantData.data.map((data, index) => `${index + 1}. ${data}`).join("\n"); + instruction += "\n\nUse this information to provide accurate and specific answers."; + } + + return instruction; + } + + private buildConversationHistory(context: IChatContext): Array<{ role: string; parts: Array<{ text: string }> }> { + if (!context.conversationHistory || context.conversationHistory.length === 0) { + return []; + } + + const history = []; + const recentHistory = context.conversationHistory.slice(-CHATBOT_CONSTANTS.MAX_CONVERSATION_HISTORY).filter((msg) => msg.type !== "system"); + + for (const msg of recentHistory) { + history.push({ + role: msg.type === "user" ? "user" : "model", + parts: [{ text: msg.content }], + }); + } + + return history; + } + + private async *createAsyncIterableFromStream(streamResult: any): AsyncIterable { + try { + // For Google Generative AI, the streamResult itself is the async iterable + for await (const chunk of streamResult.stream) { + const chunkText = chunk.text(); + if (chunkText) { + yield chunkText; + } + } + } catch (error) { + this.logger.error("Error in streaming response", error); + + // If the above doesn't work, try alternative approach for older SDK versions + try { + const response = await streamResult.response; + const text = response.text(); + if (text) { + yield text; + } + } catch (fallbackError) { + this.logger.error("Fallback streaming approach also failed", fallbackError); + throw new Error("Failed to process streaming response"); + } + } + } + + private calculateConfidence(response: any): number { + const text = response.text() || ""; + + // Check safety ratings - lower confidence if blocked + const safetyRatings = response.candidates?.[0]?.safetyRatings || []; + const hasHighRiskRatings = safetyRatings.some((rating: any) => rating.probability === "HIGH" || rating.probability === "MEDIUM"); + + if (hasHighRiskRatings) { + return 0.4; + } + + // Check for uncertainty indicators + if (text.includes("I don't know") || text.includes("I'm not sure") || text.includes("uncertain")) { + return 0.3; + } + + // Check response length and quality + if (text.length < 50) { + return 0.6; + } + + // Check if response uses provided data + if (text.includes("based on") || text.includes("according to")) { + return 0.95; + } + + return 0.8; + } + + private estimateTokens(text: string): number { + // Rough estimation: 1 token ≈ 4 characters for English text + // Gemini uses similar tokenization to other models + return Math.ceil(text.length / 4); + } +} 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/chatbot/repositories/chat-message.repository.ts b/src/modules/chatbot/repositories/chat-message.repository.ts new file mode 100644 index 0000000..2b8f17c --- /dev/null +++ b/src/modules/chatbot/repositories/chat-message.repository.ts @@ -0,0 +1,34 @@ +import { EntityManager, EntityRepository } from "@mikro-orm/postgresql"; + +import { ChatMessage, MessageStatus } from "../entities/chat-message.entity"; + +export class ChatMessageRepository extends EntityRepository { + async findBySessionWithPagination(sessionId: string, offset = 0, limit = 50, em?: EntityManager) { + const repository = em ? em.getRepository(ChatMessage) : this; + return repository.find( + { session: sessionId }, + { + populate: ["sender"], + orderBy: { createdAt: "ASC" }, + offset, + limit, + }, + ); + } + + async getConversationHistory(sessionId: string, limit = 20, em?: EntityManager) { + const repository = em ? em.getRepository(ChatMessage) : this; + return repository.find( + { session: sessionId }, + { + orderBy: { createdAt: "DESC" }, + limit, + }, + ); + } + + async markAsRead(messageIds: string[], em?: EntityManager) { + const repository = em ? em.getRepository(ChatMessage) : this; + return repository.nativeUpdate({ id: { $in: messageIds } }, { status: MessageStatus.READ }); + } +} diff --git a/src/modules/chatbot/repositories/chat-session.repository.ts b/src/modules/chatbot/repositories/chat-session.repository.ts new file mode 100644 index 0000000..aa9081b --- /dev/null +++ b/src/modules/chatbot/repositories/chat-session.repository.ts @@ -0,0 +1,33 @@ +import { EntityManager, EntityRepository } from "@mikro-orm/postgresql"; + +import { ChatSession, ChatSessionStatus } from "../entities/chat-session.entity"; + +export class ChatSessionRepository extends EntityRepository { + async findByUserWithMessages(userId: string, limit = 10, em?: EntityManager) { + const repository = em ? em.getRepository(ChatSession) : this; + return repository.find( + { user: userId }, + { + populate: ["messages"], + orderBy: { lastMessageAt: "DESC" }, + limit, + }, + ); + } + + async findActiveByUser(userId: string, em?: EntityManager) { + const repository = em ? em.getRepository(ChatSession) : this; + return repository.find( + { user: userId, status: ChatSessionStatus.ACTIVE }, + { + populate: ["messages"], + orderBy: { lastMessageAt: "DESC" }, + }, + ); + } + + async updateLastMessageTime(sessionId: string, em?: EntityManager) { + const repository = em ? em.getRepository(ChatSession) : this; + return repository.nativeUpdate({ id: sessionId }, { lastMessageAt: new Date() }); + } +}