diff --git a/package.json b/package.json index 0338282..3cc6d1b 100644 --- a/package.json +++ b/package.json @@ -31,7 +31,6 @@ "dependencies": { "@aws-sdk/client-s3": "^3.691.0", "@faker-js/faker": "^7.6.0", - "openai": "^4.0.0", "@nestjs/mapped-types": "^2.0.6", "ansi-colors": "^4.1.3", "axios": "^1.7.7", @@ -40,6 +39,7 @@ "class-transformer": "^0.5.1", "class-validator": "^0.14.1", "compression": "^1.7.5", + "cookie-parser": "^1.4.7", "cors": "^2.8.5", "csv": "^6.3.11", "dayjs": "^1.11.13", @@ -60,6 +60,7 @@ "multer-s3": "^3.0.1", "node-cache": "^5.1.2", "nodemailer": "^6.9.16", + "openai": "^4.0.0", "passport": "^0.7.0", "passport-jwt": "^4.0.1", "reflect-metadata": "^0.2.2", @@ -67,6 +68,7 @@ "socket.io": "^4.8.1", "strip-ansi": "6.0.1", "swagger-ui-express": "^5.0.1", + "ulid": "^3.0.1", "winston": "^3.17.0" }, "devDependencies": { @@ -74,6 +76,7 @@ "@commitlint/config-conventional": "^19.5.0", "@types/bcrypt": "^5.0.2", "@types/compression": "^1.7.5", + "@types/cookie-parser": "^1.4.10", "@types/cors": "^2.8.17", "@types/express": "^4.17.21", "@types/jsonwebtoken": "^9.0.7", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 8be3db9..6382377 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -38,6 +38,9 @@ importers: compression: specifier: ^1.7.5 version: 1.7.5 + cookie-parser: + specifier: ^1.4.7 + version: 1.4.7 cors: specifier: ^2.8.5 version: 2.8.5 @@ -122,6 +125,9 @@ importers: swagger-ui-express: specifier: ^5.0.1 version: 5.0.1(express@4.21.1) + ulid: + specifier: ^3.0.1 + version: 3.0.1 winston: specifier: ^3.17.0 version: 3.17.0 @@ -138,6 +144,9 @@ importers: '@types/compression': specifier: ^1.7.5 version: 1.7.5 + '@types/cookie-parser': + specifier: ^1.4.10 + version: 1.4.10(@types/express@4.17.21) '@types/cors': specifier: ^2.8.17 version: 2.8.17 @@ -1017,6 +1026,11 @@ packages: '@types/conventional-commits-parser@5.0.0': resolution: {integrity: sha512-loB369iXNmAZglwWATL+WRe+CRMmmBPtpolYzIebFaX4YA3x+BEfLqhUAV9WanycKI3TG1IMr5bMJDajDKLlUQ==} + '@types/cookie-parser@1.4.10': + resolution: {integrity: sha512-B4xqkqfZ8Wek+rCOeRxsjMS9OgvzebEzzLYw7NHYuvzb7IdxOkI0ZHGgeEBX4PUM7QGVvNSK60T3OvWj3YfBRg==} + peerDependencies: + '@types/express': '*' + '@types/cookie@0.4.1': resolution: {integrity: sha512-XW/Aa8APYr6jSVVA1y/DEIZX0/GMKLEVekNG727R8cs56ahETkRAy/3DR7+fJyh7oUgGwNQaRfXCun0+KbWY7Q==} @@ -1528,6 +1542,10 @@ packages: engines: {node: '>=16'} hasBin: true + cookie-parser@1.4.7: + resolution: {integrity: sha512-nGUvgXnotP3BsjiLX2ypbQnWoGUPIIfHQNZkkC668ntrzGWEZVW70HDEB1qnNGMicPje6EttlIgzo51YSwNQGw==} + engines: {node: '>= 0.8.0'} + cookie-signature@1.0.6: resolution: {integrity: sha512-QADzlaHc8icV8I7vbaJXJwod9HWYp8uCqf1xa4OfNu1T7JVxQIrUgOWtHdNDtPiywmFbiS12VjotIXLrKM3orQ==} @@ -3505,6 +3523,10 @@ packages: resolution: {integrity: sha512-u3xV3X7uzvi5b1MncmZo3i2Aw222Zk1keqLA1YkHldREkAhAqi65wuPfe7lHx8H/Wzy+8CE7S7uS3jekIM5s8g==} engines: {node: '>=8'} + ulid@3.0.1: + resolution: {integrity: sha512-dPJyqPzx8preQhqq24bBG1YNkvigm87K8kVEHCD+ruZg24t6IFEFv00xMWfxcC4djmFtiTLdFuADn4+DOz6R7Q==} + hasBin: true + unbox-primitive@1.0.2: resolution: {integrity: sha512-61pPlCD9h51VoreyJ0BReideM3MDKMKnh6+V9L08331ipq6Q8OFXZYiqP6n/tbHx4s5I9uRhcye6BrbkizkBDw==} @@ -4987,6 +5009,10 @@ snapshots: dependencies: '@types/node': 20.17.6 + '@types/cookie-parser@1.4.10(@types/express@4.17.21)': + dependencies: + '@types/express': 4.17.21 + '@types/cookie@0.4.1': {} '@types/cors@2.8.17': @@ -5610,6 +5636,11 @@ snapshots: meow: 12.1.1 split2: 4.2.0 + cookie-parser@1.4.7: + dependencies: + cookie: 0.7.2 + cookie-signature: 1.0.6 + cookie-signature@1.0.6: {} cookie@0.7.1: {} @@ -7755,6 +7786,8 @@ snapshots: dependencies: '@lukeed/csprng': 1.1.0 + ulid@3.0.1: {} + unbox-primitive@1.0.2: dependencies: call-bind: 1.0.7 diff --git a/src/app.ts b/src/app.ts index 5ef19ec..0510e7f 100644 --- a/src/app.ts +++ b/src/app.ts @@ -1,4 +1,5 @@ import compression from "compression"; +import cookieParser from "cookie-parser"; import cors from "cors"; import express, { Application } from "express"; import { Container } from "inversify"; @@ -121,6 +122,11 @@ class App { */ app.use(compression()); + /** + * configure cookie parser + */ + app.use(cookieParser()); + /** * configure body parser */ diff --git a/src/modules/chatbot/DTO/create-chat-session.dto.ts b/src/modules/chatbot/DTO/create-chat-session.dto.ts deleted file mode 100644 index 6acfe70..0000000 --- a/src/modules/chatbot/DTO/create-chat-session.dto.ts +++ /dev/null @@ -1,17 +0,0 @@ -import { Expose } from "class-transformer"; -import { IsOptional, IsString, MaxLength } from "class-validator"; - -import { ApiProperty } from "../../../common/decorator/swggerDocs"; - -export class CreateChatSessionDto { - @Expose() - @IsString() - @MaxLength(255) - @ApiProperty({ type: "string", description: "Title for the chat session", example: "Help with payment issues" }) - title!: string; - - @Expose() - @IsOptional() - @ApiProperty({ type: "object", description: "Initial context for the chat", example: { user: "John Doe", orderId: "123456" }, required: false }) - context?: Record; -} diff --git a/src/modules/chatbot/DTO/send-message.dto.ts b/src/modules/chatbot/DTO/send-message.dto.ts index 8ed0894..033a0c8 100644 --- a/src/modules/chatbot/DTO/send-message.dto.ts +++ b/src/modules/chatbot/DTO/send-message.dto.ts @@ -1,9 +1,7 @@ import { Expose } from "class-transformer"; -import { IsOptional, IsString, MaxLength } from "class-validator"; +import { IsNotEmpty, IsOptional, IsString, MaxLength } from "class-validator"; import { ApiProperty } from "../../../common/decorator/swggerDocs"; -import { IsValidId } from "../../../common/decorator/validation.decorator"; -import { ChatSessionModel } from "../models/chat-session.model"; export class SendMessageDto { @Expose() @@ -13,8 +11,9 @@ export class SendMessageDto { content!: string; @Expose() - @IsValidId([ChatSessionModel]) - @ApiProperty({ type: "string", description: "Chat session ID" }) + @IsNotEmpty() + @IsString() + @ApiProperty({ type: "string", description: "Chat session ID (ULID)" }) sessionId!: string; @Expose() diff --git a/src/modules/chatbot/DTO/session-id.param.dto.ts b/src/modules/chatbot/DTO/session-id.param.dto.ts index 754483a..55f29b3 100644 --- a/src/modules/chatbot/DTO/session-id.param.dto.ts +++ b/src/modules/chatbot/DTO/session-id.param.dto.ts @@ -1,14 +1,12 @@ import { Expose } from "class-transformer"; -import { IsNotEmpty } from "class-validator"; +import { IsNotEmpty, IsString } from "class-validator"; import { ApiProperty } from "../../../common/decorator/swggerDocs"; -import { IsValidId } from "../../../common/decorator/validation.decorator"; -import { ChatSessionModel } from "../models/chat-session.model"; export class SessionIdParamDto { @Expose() @IsNotEmpty({ message: "Session ID is required" }) - @IsValidId([ChatSessionModel]) - @ApiProperty({ type: "string", description: "Session id of the entity", example: "66eff8c0c6ad5530b996c432" }) + @IsString() + @ApiProperty({ type: "string", description: "Session id (ULID)", example: "01ARZ3NDEKTSV4RRFFQ69G5FAV" }) sessionId!: string; } diff --git a/src/modules/chatbot/chatbot.controller.ts b/src/modules/chatbot/chatbot.controller.ts index bf3a00c..bb5d509 100644 --- a/src/modules/chatbot/chatbot.controller.ts +++ b/src/modules/chatbot/chatbot.controller.ts @@ -2,18 +2,16 @@ import { Request, Response } from "express"; import { inject } from "inversify"; import { controller, httpGet, httpPost, httpPut, queryParam, request, requestBody, requestParam, response } from "inversify-express-utils"; -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 { getOrCreateChatbotUlid } from "./utils/ulid.util"; import { BaseController } from "../../common/base/controller"; -import { ApiAuth, ApiModel, ApiOperation, ApiParam, ApiResponse, ApiTags } from "../../common/decorator/swggerDocs"; +import { ApiModel, ApiOperation, ApiParam, ApiResponse, ApiTags } from "../../common/decorator/swggerDocs"; import { PaginationDTO } from "../../common/dto/pagination.dto"; import { HttpStatus } from "../../common/enums/httpStatus.enum"; -import { Guard } from "../../core/middlewares/guard.middleware"; import { ValidationMiddleware } from "../../core/middlewares/validator.middleware"; import { IOCTYPES } from "../../IOC/ioc.types"; -import { IUser } from "../user/models/Abstraction/IUser"; @controller("/chatbot") @ApiTags("Chatbot") @@ -22,61 +20,55 @@ class ChatbotController extends BaseController { @ApiOperation("Create a new chat session") @ApiResponse("Chat session created successfully", HttpStatus.Created) - @ApiModel(CreateChatSessionDto) - @ApiAuth() - @httpPost("/sessions", Guard.authUser(), ValidationMiddleware.validateInput(CreateChatSessionDto)) - public async createSession(@request() req: Request, @requestBody() createDto: CreateChatSessionDto) { - const user = req.user as IUser; - const data = await this.chatbotService.createChatSession(user._id.toString(), createDto); + @httpPost("/sessions") + public async createSession(@request() req: Request, @response() res: Response) { + const ulid = getOrCreateChatbotUlid(req, res); + const data = await this.chatbotService.createChatSession(ulid); return this.response({ data }, HttpStatus.Created); } @ApiOperation("Get user's chat sessions") @ApiResponse("Chat sessions retrieved successfully") - @ApiAuth() - @httpGet("/sessions", Guard.authUser(), ValidationMiddleware.validateQuery(PaginationDTO)) - public async getUserSessions(@request() req: Request, @queryParam() queryDto: PaginationDTO) { - const user = req.user as IUser; + @httpGet("/sessions", ValidationMiddleware.validateQuery(PaginationDTO)) + public async getUserSessions(@request() req: Request, @response() res: Response, @queryParam() queryDto: PaginationDTO) { + const ulid = getOrCreateChatbotUlid(req, res); const limit = queryDto.limit || 10; - const data = await this.chatbotService.getUserChatSessions(user._id.toString(), limit); + const data = await this.chatbotService.getUserChatSessions(ulid, limit); return this.response({ data }); } @ApiOperation("Get a specific chat session with messages") @ApiResponse("Chat session retrieved successfully") @ApiParam("sessionId", "Session ID", true) - @ApiAuth() - @httpGet("/sessions/:sessionId", Guard.authUser(), ValidationMiddleware.validateParameter(SessionIdParamDto)) - public async getSession(@request() req: Request, @requestParam() param: SessionIdParamDto) { - const user = req.user as IUser; - const data = await this.chatbotService.getChatSession(param.sessionId, user._id.toString()); + @httpGet("/sessions/:sessionId", ValidationMiddleware.validateParameter(SessionIdParamDto)) + public async getSession(@request() req: Request, @response() res: Response, @requestParam() param: SessionIdParamDto) { + const ulid = getOrCreateChatbotUlid(req, res); + const data = await this.chatbotService.getChatSession(param.sessionId, ulid); return this.response({ data }); } @ApiOperation("Send a message in a chat session") @ApiResponse("Message sent successfully", HttpStatus.Created) @ApiModel(SendMessageDto) - @ApiAuth() - @httpPost("/messages", Guard.authUser(), ValidationMiddleware.validateInput(SendMessageDto)) - public async sendMessage(@request() req: Request, @requestBody() sendDto: SendMessageDto) { - const user = req.user as IUser; - const data = await this.chatbotService.sendMessage(user._id.toString(), sendDto); + @httpPost("/messages", ValidationMiddleware.validateInput(SendMessageDto)) + public async sendMessage(@request() req: Request, @response() res: Response, @requestBody() sendDto: SendMessageDto) { + const ulid = getOrCreateChatbotUlid(req, res); + const data = await this.chatbotService.sendMessage(ulid, sendDto); return this.response({ data }, HttpStatus.Created); } @ApiOperation("Send a message and get streaming response") @ApiResponse("Streaming response initiated") @ApiModel(SendMessageDto) - @ApiAuth() - @httpPost("/messages/stream", Guard.authUser(), ValidationMiddleware.validateInput(SendMessageDto)) + @httpPost("/messages/stream", ValidationMiddleware.validateInput(SendMessageDto)) public async sendMessageStream( @request() req: Request, @requestBody() sendDto: SendMessageDto, @response() res: Response, ): Promise { - const user = req.user as IUser; + const ulid = getOrCreateChatbotUlid(req, res); try { - const { userMessage, streamGenerator } = await this.chatbotService.sendMessageStream(user._id.toString(), sendDto); + const { userMessage, streamGenerator } = await this.chatbotService.sendMessageStream(ulid, sendDto); // Set headers for Server-Sent Events res.setHeader("Content-Type", "text/event-stream"); @@ -90,13 +82,20 @@ class ChatbotController extends BaseController { // Start streaming bot response res.write(`data: ${JSON.stringify({ type: "bot_response_start" })}\n\n`); + let fullResponse = ""; const stream = await streamGenerator(); for await (const chunk of stream) { if (chunk) { + fullResponse += chunk; res.write(`data: ${JSON.stringify({ type: "bot_response_chunk", data: chunk })}\n\n`); } } + // Save bot response to cache after streaming completes + if (fullResponse) { + await this.chatbotService.saveStreamedBotResponse(sendDto.sessionId, userMessage.id, fullResponse, ulid); + } + // End streaming res.write(`data: ${JSON.stringify({ type: "bot_response_end" })}\n\n`); res.end(); @@ -110,26 +109,25 @@ class ChatbotController extends BaseController { @ApiOperation("Close a chat session") @ApiResponse("Chat session closed successfully") @ApiParam("sessionId", "Session ID", true) - @ApiAuth() - @httpPut("/sessions/:sessionId/close", Guard.authUser(), ValidationMiddleware.validateParameter(SessionIdParamDto)) - public async closeSession(@request() req: Request, @requestParam() param: SessionIdParamDto) { - const user = req.user as IUser; - const data = await this.chatbotService.closeChatSession(param.sessionId, user._id.toString()); + @httpPut("/sessions/:sessionId/close", ValidationMiddleware.validateParameter(SessionIdParamDto)) + public async closeSession(@request() req: Request, @response() res: Response, @requestParam() param: SessionIdParamDto) { + const ulid = getOrCreateChatbotUlid(req, res); + const data = await this.chatbotService.closeChatSession(param.sessionId, ulid); return this.response({ data }); } @ApiOperation("Mark messages as read") @ApiResponse("Messages marked as read successfully") @ApiParam("sessionId", "Session ID", true) - @ApiAuth() - @httpPut("/sessions/:sessionId/messages/read", Guard.authUser(), ValidationMiddleware.validateParameter(SessionIdParamDto)) + @httpPut("/sessions/:sessionId/messages/read", ValidationMiddleware.validateParameter(SessionIdParamDto)) public async markMessagesAsRead( @request() req: Request, + @response() res: Response, @requestParam() param: SessionIdParamDto, @requestBody() body: { messageIds: string[] }, ) { - const user = req.user as IUser; - const data = await this.chatbotService.markMessagesAsRead(param.sessionId, user._id.toString(), body.messageIds); + const ulid = getOrCreateChatbotUlid(req, res); + const data = await this.chatbotService.markMessagesAsRead(param.sessionId, ulid, body.messageIds); return this.response({ data }); } } diff --git a/src/modules/chatbot/providers/chatbot.service.ts b/src/modules/chatbot/providers/chatbot.service.ts index f12d506..071c574 100644 --- a/src/modules/chatbot/providers/chatbot.service.ts +++ b/src/modules/chatbot/providers/chatbot.service.ts @@ -1,161 +1,234 @@ import { inject, injectable } from "inversify"; -import { isValidObjectId } from "mongoose"; +import * as ulidLib from "ulid"; import { LLMService } from "./llm.service"; -import { UserModel } from "../../user/models/user.model"; -import { CreateChatSessionDto } from "../DTO/create-chat-session.dto"; -import { SendMessageDto } from "../DTO/send-message.dto"; -import { ChatMessageModel, MessageStatus, MessageType } from "../models/chat-message.model"; -import { ChatSessionModel, ChatSessionStatus } from "../models/chat-session.model"; -import { IChatContext } from "../interfaces/chatbot.interface"; -import { IChatMessage } from "../models/Abstraction/IChatMessage"; -import { IChatSession } from "../models/Abstraction/IChatSession"; -import { ChatMessageRepository } from "../repositories/chat-message.repository"; -import { ChatSessionRepository } from "../repositories/chat-session.repository"; import { BadRequestError, NotFoundError } from "../../../core/app/app.errors"; import { Logger } from "../../../core/logging/logger"; import { IOCTYPES } from "../../../IOC/ioc.types"; +import { CacheService } from "../../../utils/cache.service"; +import { SendMessageDto } from "../DTO/send-message.dto"; +import { IChatContext } from "../interfaces/chatbot.interface"; +import { MessageStatus, MessageType } from "../models/chat-message.model"; +import { ChatSessionStatus } from "../models/chat-session.model"; + +interface CachedSession { + id: string; + ulid: string; + status: ChatSessionStatus; + lastMessageAt?: string; + createdAt: string; +} + +interface CachedMessage { + id: string; + content: string; + type: MessageType; + status: MessageStatus; + sessionId: string; + metadata?: Record; + responseToId?: string; + tokensUsed: number; + createdAt: string; +} @injectable() export class ChatbotService { private readonly logger: Logger; + private readonly CACHE_TTL_SECONDS = 15 * 60; // 15 minutes constructor( @inject(IOCTYPES.ChatbotLLMService) private llmService: LLMService, - @inject(IOCTYPES.ChatbotChatSessionRepository) private chatSessionRepo: ChatSessionRepository, - @inject(IOCTYPES.ChatbotChatMessageRepository) private chatMessageRepo: ChatMessageRepository, + @inject(IOCTYPES.CacheService) private cacheService: CacheService, ) { this.logger = new Logger("ChatbotService"); this.logger.info("Using OpenAI as LLM provider"); } - async createChatSession(userId: string, createDto: CreateChatSessionDto) { - if (!isValidObjectId(userId)) { - throw new BadRequestError("Invalid user ID"); - } + private getSessionCacheKey(ulid: string, sessionId: string): string { + return `chatbot:session:${ulid}:${sessionId}`; + } - const user = await UserModel.findById(userId); - if (!user) { - throw new NotFoundError("User not found"); - } + private getSessionsListCacheKey(ulid: string): string { + return `chatbot:sessions:${ulid}`; + } - const session = await ChatSessionModel.create({ - title: createDto.title, - user: userId, - context: createDto.context || {}, - lastMessageAt: new Date(), + private getMessagesCacheKey(sessionId: string): string { + return `chatbot:messages:${sessionId}`; + } + + async createChatSession(ulid: string) { + const sessionId = ulidLib.ulid(); + const now = new Date().toISOString(); + + const session: CachedSession = { + id: sessionId, + ulid, status: ChatSessionStatus.ACTIVE, - }); + lastMessageAt: now, + createdAt: now, + }; + + // Save session to cache + await this.cacheService.setAsync(this.getSessionCacheKey(ulid, sessionId), JSON.stringify(session), this.CACHE_TTL_SECONDS); + + // Add to sessions list + const sessionsListKey = this.getSessionsListCacheKey(ulid); + const existingSessionsJson = await this.cacheService.getAsync(sessionsListKey); + const sessionsList: string[] = existingSessionsJson ? JSON.parse(existingSessionsJson) : []; + sessionsList.push(sessionId); + await this.cacheService.setAsync(sessionsListKey, JSON.stringify(sessionsList), this.CACHE_TTL_SECONDS); + + // Initialize messages array for this session + await this.cacheService.setAsync(this.getMessagesCacheKey(sessionId), JSON.stringify([]), this.CACHE_TTL_SECONDS); return this.mapSessionToDto(session); } - async getUserChatSessions(userId: string, limit = 10) { - if (!isValidObjectId(userId)) { - throw new BadRequestError("Invalid user ID"); + async getUserChatSessions(ulid: string, limit = 10) { + const sessionsListKey = this.getSessionsListCacheKey(ulid); + const sessionsListJson = await this.cacheService.getAsync(sessionsListKey); + + if (!sessionsListJson) { + return []; + } + + const sessionIds: string[] = JSON.parse(sessionsListJson); + const sessions: CachedSession[] = []; + + // Get sessions in reverse order (newest first) + for (let i = sessionIds.length - 1; i >= 0 && sessions.length < limit; i--) { + const sessionId = sessionIds[i]; + const sessionKey = this.getSessionCacheKey(ulid, sessionId); + const sessionJson = await this.cacheService.getAsync(sessionKey); + + if (sessionJson) { + const session: CachedSession = JSON.parse(sessionJson); + // Get messages for this session + const messagesJson = await this.cacheService.getAsync(this.getMessagesCacheKey(sessionId)); + const messages: CachedMessage[] = messagesJson ? JSON.parse(messagesJson) : []; + sessions.push({ ...session, messages } as any); + } } - const sessions = await this.chatSessionRepo.findByUserWithMessages(userId, limit); return sessions.map((session) => this.mapSessionToDto(session)); } - async getChatSession(sessionId: string, userId: string) { - if (!isValidObjectId(sessionId) || !isValidObjectId(userId)) { - throw new BadRequestError("Invalid session or user ID"); - } + async getChatSession(sessionId: string, ulid: string) { + const sessionKey = this.getSessionCacheKey(ulid, sessionId); + const sessionJson = await this.cacheService.getAsync(sessionKey); - const session = await this.chatSessionRepo.findByUserAndId(sessionId, userId); - if (!session) { + if (!sessionJson) { throw new NotFoundError("Chat session not found"); } - return this.mapSessionToDto(session); + const session: CachedSession = JSON.parse(sessionJson); + + // Get messages for this session + const messagesJson = await this.cacheService.getAsync(this.getMessagesCacheKey(sessionId)); + const messages: CachedMessage[] = messagesJson ? JSON.parse(messagesJson) : []; + + return this.mapSessionToDto({ ...session, messages } as any); } - async sendMessage(userId: string, sendDto: SendMessageDto) { - if (!isValidObjectId(sendDto.sessionId) || !isValidObjectId(userId)) { - throw new BadRequestError("Invalid session or user ID"); - } + async sendMessage(ulid: string, sendDto: SendMessageDto) { + const sessionKey = this.getSessionCacheKey(ulid, sendDto.sessionId); + const sessionJson = await this.cacheService.getAsync(sessionKey); - const session = await ChatSessionModel.findOne({ _id: sendDto.sessionId, user: userId }).populate("messages").lean(); - if (!session) { + if (!sessionJson) { throw new NotFoundError("Chat session not found"); } + const session: CachedSession = JSON.parse(sessionJson); + if (session.status !== ChatSessionStatus.ACTIVE) { throw new BadRequestError("Cannot send message to inactive session"); } - const user = await UserModel.findById(userId); - if (!user) { - throw new NotFoundError("User not found"); - } - // Create user message - const userMessage = await ChatMessageModel.create({ + const messageId = ulidLib.ulid(); + const now = new Date().toISOString(); + const userMessage: CachedMessage = { + id: messageId, content: sendDto.content, type: MessageType.USER, - session: sendDto.sessionId, - sender: userId, + status: MessageStatus.SENT, + sessionId: sendDto.sessionId, responseToId: sendDto.responseToId, metadata: sendDto.metadata, - }); + tokensUsed: 0, + createdAt: now, + }; + + // Save message to cache + const messagesKey = this.getMessagesCacheKey(sendDto.sessionId); + const messagesJson = await this.cacheService.getAsync(messagesKey); + const messages: CachedMessage[] = messagesJson ? JSON.parse(messagesJson) : []; + messages.push(userMessage); + await this.cacheService.setAsync(messagesKey, JSON.stringify(messages), this.CACHE_TTL_SECONDS); // Update session last message time - await this.chatSessionRepo.updateLastMessageTime(sendDto.sessionId); + session.lastMessageAt = now; + await this.cacheService.setAsync(sessionKey, JSON.stringify(session), this.CACHE_TTL_SECONDS); // Generate bot response asynchronously using OpenAI service - this.generateBotResponse(sendDto.sessionId, userMessage._id.toString(), userId); + this.generateBotResponse(sendDto.sessionId, messageId, ulid); return this.mapMessageToDto(userMessage); } - async sendMessageStream(userId: string, sendDto: SendMessageDto) { - if (!isValidObjectId(sendDto.sessionId) || !isValidObjectId(userId)) { - throw new BadRequestError("Invalid session or user ID"); - } + async sendMessageStream(ulid: string, sendDto: SendMessageDto) { + const sessionKey = this.getSessionCacheKey(ulid, sendDto.sessionId); + const sessionJson = await this.cacheService.getAsync(sessionKey); - const session = await ChatSessionModel.findOne({ _id: sendDto.sessionId, user: userId }).lean(); - if (!session) { + if (!sessionJson) { throw new NotFoundError("Chat session not found"); } + const session: CachedSession = JSON.parse(sessionJson); + if (session.status !== ChatSessionStatus.ACTIVE) { throw new BadRequestError("Cannot send message to inactive session"); } - const user = await UserModel.findById(userId); - if (!user) { - throw new NotFoundError("User not found"); - } - // Create user message - const userMessage = await ChatMessageModel.create({ + const messageId = ulidLib.ulid(); + const now = new Date().toISOString(); + const userMessage: CachedMessage = { + id: messageId, content: sendDto.content, type: MessageType.USER, - session: sendDto.sessionId, - sender: userId, + status: MessageStatus.SENT, + sessionId: sendDto.sessionId, responseToId: sendDto.responseToId, metadata: sendDto.metadata, - }); + tokensUsed: 0, + createdAt: now, + }; + + // Save message to cache + const messagesKey = this.getMessagesCacheKey(sendDto.sessionId); + const messagesJson = await this.cacheService.getAsync(messagesKey); + const messages: CachedMessage[] = messagesJson ? JSON.parse(messagesJson) : []; + messages.push(userMessage); + await this.cacheService.setAsync(messagesKey, JSON.stringify(messages), this.CACHE_TTL_SECONDS); // Update session last message time - await this.chatSessionRepo.updateLastMessageTime(sendDto.sessionId); + session.lastMessageAt = now; + await this.cacheService.setAsync(sessionKey, JSON.stringify(session), this.CACHE_TTL_SECONDS); // Get conversation history for context - const history = await this.chatMessageRepo.getConversationHistory(sendDto.sessionId, 20); + const history = messages.slice(-20).map((msg) => ({ + content: msg.content, + type: msg.type as "user" | "bot" | "system", + timestamp: new Date(msg.createdAt), + metadata: msg.metadata, + })); // Build context for LLM const context: IChatContext = { - userId, + userId: ulid, sessionId: sendDto.sessionId, - conversationHistory: history.reverse().map((msg) => ({ - content: msg.content, - type: msg.type as "user" | "bot" | "system", - timestamp: msg.createdAt, - metadata: msg.metadata, - })), - userPreferences: session.context, + conversationHistory: history, }; // Return a function that generates the stream using LLM service @@ -167,42 +240,63 @@ export class ChatbotService { }; } - private async generateBotResponse(sessionId: string, userMessageId: string, userId: string): Promise { + private async generateBotResponse(sessionId: string, userMessageId: string, ulid: string): Promise { try { - // Get conversation history - const history = await this.chatMessageRepo.getConversationHistory(sessionId, 20); - - const session = await ChatSessionModel.findById(sessionId).lean(); - if (!session) { + // Get messages from cache + const messagesKey = this.getMessagesCacheKey(sessionId); + const messagesJson = await this.cacheService.getAsync(messagesKey); + if (!messagesJson) { throw new NotFoundError("Session not found"); } - // Build context for LLM - const context: IChatContext = { - userId, - sessionId, - conversationHistory: history.reverse().map((msg) => ({ - content: msg.content, - type: msg.type as "user" | "bot" | "system", - timestamp: msg.createdAt, - metadata: msg.metadata, - })), - userPreferences: session.context, - }; - - const userMessage = await ChatMessageModel.findById(userMessageId); + const messages: CachedMessage[] = JSON.parse(messagesJson); + const userMessage = messages.find((msg) => msg.id === userMessageId); if (!userMessage) { throw new NotFoundError("User message not found"); } + // Get session + const sessionsListKey = this.getSessionsListCacheKey(ulid); + const sessionsListJson = await this.cacheService.getAsync(sessionsListKey); + if (!sessionsListJson) { + throw new NotFoundError("Session not found"); + } + + const sessionKey = this.getSessionCacheKey(ulid, sessionId); + const sessionJson = await this.cacheService.getAsync(sessionKey); + if (!sessionJson) { + throw new NotFoundError("Session not found"); + } + + const session: CachedSession = JSON.parse(sessionJson); + + // Get conversation history + const history = messages.slice(-20).map((msg) => ({ + content: msg.content, + type: msg.type as "user" | "bot" | "system", + timestamp: new Date(msg.createdAt), + metadata: msg.metadata, + })); + + // Build context for LLM + const context: IChatContext = { + userId: ulid, + sessionId, + conversationHistory: history, + }; + // Generate response using LLM const llmResponse = await this.llmService.generateResponse(userMessage.content, context); // Create bot message - await ChatMessageModel.create({ + const botMessageId = ulidLib.ulid(); + const now = new Date().toISOString(); + const botMessage: CachedMessage = { + id: botMessageId, content: llmResponse.message, type: MessageType.BOT, - session: sessionId, + status: MessageStatus.SENT, + sessionId, responseToId: userMessageId, tokensUsed: llmResponse.tokensUsed, metadata: { @@ -210,83 +304,147 @@ export class ChatbotService { sources: llmResponse.sources, llmContext: llmResponse.context, }, - }); + createdAt: now, + }; + + // Save bot message to cache + messages.push(botMessage); + await this.cacheService.setAsync(messagesKey, JSON.stringify(messages), this.CACHE_TTL_SECONDS); // Update session last message time - await this.chatSessionRepo.updateLastMessageTime(sessionId); + session.lastMessageAt = now; + await this.cacheService.setAsync(sessionKey, JSON.stringify(session), this.CACHE_TTL_SECONDS); this.logger.info(`Generated bot response for session ${sessionId}`); } catch (error) { this.logger.error(`Failed to generate bot response for session ${sessionId}`, error); // Create error message - await ChatMessageModel.create({ + const messagesKey = this.getMessagesCacheKey(sessionId); + const messagesJson = await this.cacheService.getAsync(messagesKey); + const messages: CachedMessage[] = messagesJson ? JSON.parse(messagesJson) : []; + + const errorMessageId = ulidLib.ulid(); + const now = new Date().toISOString(); + const errorMessage: CachedMessage = { + id: errorMessageId, content: "متأسفم، در حال حاضر مشکلی در پاسخگویی دارم. لطفاً دوباره تلاش کنید. 🙏", type: MessageType.BOT, - session: sessionId, - responseToId: userMessageId, status: MessageStatus.FAILED, - }); + sessionId, + responseToId: userMessageId, + tokensUsed: 0, + createdAt: now, + }; + + messages.push(errorMessage); + await this.cacheService.setAsync(messagesKey, JSON.stringify(messages), this.CACHE_TTL_SECONDS); } } - async closeChatSession(sessionId: string, userId: string) { - if (!isValidObjectId(sessionId) || !isValidObjectId(userId)) { - throw new BadRequestError("Invalid session or user ID"); - } + async closeChatSession(sessionId: string, ulid: string) { + const sessionKey = this.getSessionCacheKey(ulid, sessionId); + const sessionJson = await this.cacheService.getAsync(sessionKey); - const session = await ChatSessionModel.findOneAndUpdate( - { _id: sessionId, user: userId }, - { status: ChatSessionStatus.CLOSED }, - { new: true }, - ); - - if (!session) { + if (!sessionJson) { throw new NotFoundError("Chat session not found"); } + const session: CachedSession = JSON.parse(sessionJson); + session.status = ChatSessionStatus.CLOSED; + + await this.cacheService.setAsync(sessionKey, JSON.stringify(session), this.CACHE_TTL_SECONDS); + return { message: "Chat session closed successfully" }; } - async markMessagesAsRead(sessionId: string, userId: string, messageIds: string[]) { - if (!isValidObjectId(sessionId) || !isValidObjectId(userId)) { - throw new BadRequestError("Invalid session or user ID"); - } + async markMessagesAsRead(sessionId: string, ulid: string, messageIds: string[]) { + const sessionKey = this.getSessionCacheKey(ulid, sessionId); + const sessionJson = await this.cacheService.getAsync(sessionKey); - const session = await ChatSessionModel.findOne({ _id: sessionId, user: userId }); - if (!session) { + if (!sessionJson) { throw new NotFoundError("Chat session not found"); } - await this.chatMessageRepo.markAsRead(messageIds); + // Get messages and update their status + const messagesKey = this.getMessagesCacheKey(sessionId); + const messagesJson = await this.cacheService.getAsync(messagesKey); + + if (messagesJson) { + const messages: CachedMessage[] = JSON.parse(messagesJson); + messages.forEach((msg) => { + if (messageIds.includes(msg.id)) { + msg.status = MessageStatus.READ; + } + }); + await this.cacheService.setAsync(messagesKey, JSON.stringify(messages), this.CACHE_TTL_SECONDS); + } return { message: "Messages marked as read successfully" }; } - private mapSessionToDto(session: IChatSession | any) { - const sessionObj = session.toObject ? session.toObject() : session; + async saveStreamedBotResponse(sessionId: string, userMessageId: string, botResponse: string, ulid: string): Promise { + try { + const messagesKey = this.getMessagesCacheKey(sessionId); + const messagesJson = await this.cacheService.getAsync(messagesKey); + if (!messagesJson) { + this.logger.warn(`Cannot save bot response: session ${sessionId} not found in cache`); + return; + } + + const messages: CachedMessage[] = JSON.parse(messagesJson); + const botMessageId = ulidLib.ulid(); + const now = new Date().toISOString(); + + const botMessage: CachedMessage = { + id: botMessageId, + content: botResponse, + type: MessageType.BOT, + status: MessageStatus.SENT, + sessionId, + responseToId: userMessageId, + tokensUsed: 0, // Could estimate tokens if needed + createdAt: now, + }; + + messages.push(botMessage); + await this.cacheService.setAsync(messagesKey, JSON.stringify(messages), this.CACHE_TTL_SECONDS); + + // Update session last message time + const sessionKey = this.getSessionCacheKey(ulid, sessionId); + const sessionJson = await this.cacheService.getAsync(sessionKey); + if (sessionJson) { + const session: CachedSession = JSON.parse(sessionJson); + session.lastMessageAt = now; + await this.cacheService.setAsync(sessionKey, JSON.stringify(session), this.CACHE_TTL_SECONDS); + } + + this.logger.info(`Saved streamed bot response for session ${sessionId}`); + } catch (error) { + this.logger.error(`Failed to save streamed bot response for session ${sessionId}`, error); + } + } + + private mapSessionToDto(session: CachedSession | any) { return { - id: sessionObj._id.toString(), - title: sessionObj.title, - status: sessionObj.status, - createdAt: sessionObj.createdAt, - lastMessageAt: sessionObj.lastMessageAt, - context: sessionObj.context, - messages: sessionObj.messages ? sessionObj.messages.map((msg: any) => this.mapMessageToDto(msg)) : [], + id: session.id, + status: session.status, + createdAt: session.createdAt, + lastMessageAt: session.lastMessageAt, + messages: session.messages ? session.messages.map((msg: CachedMessage) => this.mapMessageToDto(msg)) : [], }; } - private mapMessageToDto(message: IChatMessage | any) { - const messageObj = message.toObject ? message.toObject() : message; + private mapMessageToDto(message: CachedMessage) { return { - id: messageObj._id.toString(), - content: messageObj.content, - type: messageObj.type, - status: messageObj.status, - createdAt: messageObj.createdAt, - responseToId: messageObj.responseToId, - metadata: messageObj.metadata, - tokensUsed: messageObj.tokensUsed, + id: message.id, + content: message.content, + type: message.type, + status: message.status, + createdAt: message.createdAt, + responseToId: message.responseToId, + metadata: message.metadata, + tokensUsed: message.tokensUsed, }; } } diff --git a/src/modules/chatbot/utils/ulid.util.ts b/src/modules/chatbot/utils/ulid.util.ts new file mode 100644 index 0000000..b46a8e3 --- /dev/null +++ b/src/modules/chatbot/utils/ulid.util.ts @@ -0,0 +1,28 @@ +import { Request, Response } from "express"; +import * as ulidLib from "ulid"; + +const CHATBOT_ULID_COOKIE_NAME = "chatbot_session_id"; + +/** + * Get or create ULID for chatbot session from cookies + * @param req Express request object + * @param res Express response object + * @returns ULID string + */ +export function getOrCreateChatbotUlid(req: Request, res: Response): string { + let ulidValue = req.cookies?.[CHATBOT_ULID_COOKIE_NAME]; + + if (!ulidValue) { + ulidValue = ulidLib.ulid(); + // Set cookie with 1 year expiration (or adjust as needed) + res.cookie(CHATBOT_ULID_COOKIE_NAME, ulidValue, { + maxAge: 365 * 24 * 60 * 60 * 1000, // 1 year + httpOnly: true, + secure: process.env.NODE_ENV === "production", + sameSite: "lax", + }); + } + + return ulidValue; +} +