From 4335e852b92d3b356841c0c8b50cea3c0183904b Mon Sep 17 00:00:00 2001 From: Biniyam Date: Fri, 28 Mar 2025 18:13:12 +0300 Subject: [PATCH 01/16] boiler-plate for the moderator microservice --- src/apps/moderator/main.ts | 27 +++++++++++++++++++ .../moderator/moderator.controller.spec.ts | 20 ++++++++++++++ src/apps/moderator/moderator.controller.ts | 13 +++++++++ src/apps/moderator/moderator.module.ts | 9 +++++++ src/apps/moderator/moderator.service.spec.ts | 18 +++++++++++++ src/apps/moderator/moderator.service.ts | 9 +++++++ 6 files changed, 96 insertions(+) create mode 100644 src/apps/moderator/main.ts create mode 100644 src/apps/moderator/moderator.controller.spec.ts create mode 100644 src/apps/moderator/moderator.controller.ts create mode 100644 src/apps/moderator/moderator.module.ts create mode 100644 src/apps/moderator/moderator.service.spec.ts create mode 100644 src/apps/moderator/moderator.service.ts diff --git a/src/apps/moderator/main.ts b/src/apps/moderator/main.ts new file mode 100644 index 00000000..e124e4e0 --- /dev/null +++ b/src/apps/moderator/main.ts @@ -0,0 +1,27 @@ +import { MicroserviceOptions, Transport } from '@nestjs/microservices'; +import { ModeratorModule } from './moderator.module'; +import { NestFactory } from '@nestjs/core'; + +async function bootstrap() { + const app = await NestFactory.createMicroservice( + ModeratorModule, + { + transport: Transport.RMQ, + options: { + urls: [process.env.RABBITMQ_URL], + queue: 'moderator', + queueOptions: { + durable: true, + messageTtl: 60000, + deadLetterExchange: 'moderator', + deadLetterRoutingKey: 'dead', + }, + noAck: false, + prefetchCount: 1, + }, + }, + ); + await app.listen(); +} + +bootstrap(); diff --git a/src/apps/moderator/moderator.controller.spec.ts b/src/apps/moderator/moderator.controller.spec.ts new file mode 100644 index 00000000..81472b03 --- /dev/null +++ b/src/apps/moderator/moderator.controller.spec.ts @@ -0,0 +1,20 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { ModeratorController } from './moderator.controller'; +import { ModeratorService } from './moderator.service'; + +describe('ModeratorController', () => { + let controller: ModeratorController; + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + controllers: [ModeratorController], + providers: [ModeratorService], + }).compile(); + + controller = module.get(ModeratorController); + }); + + it('should be defined', () => { + expect(controller).toBeDefined(); + }); +}); diff --git a/src/apps/moderator/moderator.controller.ts b/src/apps/moderator/moderator.controller.ts new file mode 100644 index 00000000..4ec4546b --- /dev/null +++ b/src/apps/moderator/moderator.controller.ts @@ -0,0 +1,13 @@ +import { Controller } from '@nestjs/common'; +import { ModeratorService } from './moderator.service'; +import { Ctx, RmqContext, Payload, EventPattern } from '@nestjs/microservices'; + +@Controller() +export class ModeratorController { + constructor(private readonly moderatorService: ModeratorService) {} + + @EventPattern() + async handleMessage(@Payload() data: any, @Ctx() context: RmqContext) { + return await this.moderatorService.handleMessage(data, context); + } +} diff --git a/src/apps/moderator/moderator.module.ts b/src/apps/moderator/moderator.module.ts new file mode 100644 index 00000000..7bbd5c44 --- /dev/null +++ b/src/apps/moderator/moderator.module.ts @@ -0,0 +1,9 @@ +import { Module } from '@nestjs/common'; +import { ModeratorService } from './moderator.service'; +import { ModeratorController } from './moderator.controller'; + +@Module({ + controllers: [ModeratorController], + providers: [ModeratorService], +}) +export class ModeratorModule {} diff --git a/src/apps/moderator/moderator.service.spec.ts b/src/apps/moderator/moderator.service.spec.ts new file mode 100644 index 00000000..f2fc54de --- /dev/null +++ b/src/apps/moderator/moderator.service.spec.ts @@ -0,0 +1,18 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { ModeratorService } from './moderator.service'; + +describe('ModeratorService', () => { + let service: ModeratorService; + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + providers: [ModeratorService], + }).compile(); + + service = module.get(ModeratorService); + }); + + it('should be defined', () => { + expect(service).toBeDefined(); + }); +}); diff --git a/src/apps/moderator/moderator.service.ts b/src/apps/moderator/moderator.service.ts new file mode 100644 index 00000000..b2cf6582 --- /dev/null +++ b/src/apps/moderator/moderator.service.ts @@ -0,0 +1,9 @@ +import { Injectable } from '@nestjs/common'; +import { RmqContext } from '@nestjs/microservices'; + +@Injectable() +export class ModeratorService { + async handleMessage(message: any, context: RmqContext) { + console.log({ message: message, context: context }); + } +} From 34f3825a4c5e27c7b5e4c5e5b3d60c7c34f37c65 Mon Sep 17 00:00:00 2001 From: Biniyam Date: Fri, 28 Mar 2025 18:21:56 +0300 Subject: [PATCH 02/16] send message to the moderator queue if no conversation has been created --- src/modules/message/platform/telegram-message.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/modules/message/platform/telegram-message.ts b/src/modules/message/platform/telegram-message.ts index 34b3c4f6..a49b0f88 100644 --- a/src/modules/message/platform/telegram-message.ts +++ b/src/modules/message/platform/telegram-message.ts @@ -17,7 +17,11 @@ export class TelegramMessageStrategy implements MessageStrategy { const data = JSON.stringify( await this.rabbitmqService.getMessageEchangeData(formattedMessage), ); - this.messageExchangeService.send('message', data); + if (!formattedMessage.conversationId) { + await this.messageExchangeService.send('moderator', data); + } else if (formattedMessage.conversationId) { + await this.messageExchangeService.send('message', data); + } return 'ok'; } From 02e021c51ce0ed55a8d0a1940b47e1512d1b47c5 Mon Sep 17 00:00:00 2001 From: Biniyam Date: Fri, 28 Mar 2025 18:24:46 +0300 Subject: [PATCH 03/16] added an image for the moderator microservice into docker --- docker-compose.yml | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/docker-compose.yml b/docker-compose.yml index a69fd2dd..ac6dabec 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -100,5 +100,21 @@ services: DATABASE_URL: postgres://postgres:password@postgres:5432/mydb?schema=public command: ['node', 'dist/src/apps/consumers/message-consumers/main.js'] + moderator: + build: + context: . + volumes: + - .:/app + platform: linux/amd64 + depends_on: + postgres: + condition: service_healthy + rabbitmq: + condition: service_started + environment: + RABBITMQ_URL: amqp://user:password@rabbitmq:5672 + DATABASE_URL: postgres://postgres:password@postgres:5432/mydb?schema=public + command: ['node', 'dist/src/apps/moderator/main.js'] + volumes: postgres_data: From 1ed2e463dc0ccddf79267cbc9480fdf904d39653 Mon Sep 17 00:00:00 2001 From: Biniyam Date: Fri, 28 Mar 2025 18:43:26 +0300 Subject: [PATCH 04/16] moderator queue creation logic as well as binding the queues --- .../message-exchange-queues.service.ts | 20 ++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/src/common/rabbitmq/message-exchange/message-exchange-queues.service.ts b/src/common/rabbitmq/message-exchange/message-exchange-queues.service.ts index 110b27a4..04ebe767 100644 --- a/src/common/rabbitmq/message-exchange/message-exchange-queues.service.ts +++ b/src/common/rabbitmq/message-exchange/message-exchange-queues.service.ts @@ -4,15 +4,25 @@ import * as amqp from 'amqplib'; @Injectable() export class MessageExchangeQueuesService { private readonly QUEUE_NAMES = ['message', 'database']; - private readonly ROUTING_KEY = 'message'; + private readonly PRE_CONV_QUEUE_NAMES = ['moderator', 'database']; private channel: amqp.Channel; async init(channel: amqp.Channel, exchangeName: string) { this.channel = channel; - await this.createQueue(exchangeName); + await this.createQueue(exchangeName, this.QUEUE_NAMES, 'message'); + await this.createQueue( + exchangeName, + this.PRE_CONV_QUEUE_NAMES, + 'moderator', + ); } - async createQueue(exchangeName: string) { - for (const queueName of this.QUEUE_NAMES) { + + async createQueue( + exchangeName: string, + queueNames: string[], + routingKey: string, + ) { + for (const queueName of queueNames) { try { await this.channel.assertQueue(queueName, { durable: true, @@ -20,7 +30,7 @@ export class MessageExchangeQueuesService { deadLetterExchange: 'message', deadLetterRoutingKey: 'dead', }); - await this.channel.bindQueue(queueName, exchangeName, this.ROUTING_KEY); + await this.channel.bindQueue(queueName, exchangeName, routingKey); } catch (error) { console.error(error); } From 8a1571aa7786a1e3ce0f26c24cd17f2ec2e969b9 Mon Sep 17 00:00:00 2001 From: Biniyam Date: Sat, 29 Mar 2025 11:34:28 +0300 Subject: [PATCH 05/16] debugged inequivalent arg 'x-dead-letter-exchange' for queue 'moderator' --- src/apps/moderator/main.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/apps/moderator/main.ts b/src/apps/moderator/main.ts index e124e4e0..e3ac3f7e 100644 --- a/src/apps/moderator/main.ts +++ b/src/apps/moderator/main.ts @@ -13,7 +13,7 @@ async function bootstrap() { queueOptions: { durable: true, messageTtl: 60000, - deadLetterExchange: 'moderator', + deadLetterExchange: 'message', deadLetterRoutingKey: 'dead', }, noAck: false, From 43b31066cd8f1a8f0b93299124679c650dabda8d Mon Sep 17 00:00:00 2001 From: Biniyam Date: Sat, 29 Mar 2025 11:41:43 +0300 Subject: [PATCH 06/16] returned making the routing key moderator so that to bind moderator queue with the right routing routing key --- src/apps/moderator/main.ts | 2 +- .../message-exchange/message-exchange-queues.service.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/apps/moderator/main.ts b/src/apps/moderator/main.ts index e3ac3f7e..e124e4e0 100644 --- a/src/apps/moderator/main.ts +++ b/src/apps/moderator/main.ts @@ -13,7 +13,7 @@ async function bootstrap() { queueOptions: { durable: true, messageTtl: 60000, - deadLetterExchange: 'message', + deadLetterExchange: 'moderator', deadLetterRoutingKey: 'dead', }, noAck: false, diff --git a/src/common/rabbitmq/message-exchange/message-exchange-queues.service.ts b/src/common/rabbitmq/message-exchange/message-exchange-queues.service.ts index 04ebe767..9d654174 100644 --- a/src/common/rabbitmq/message-exchange/message-exchange-queues.service.ts +++ b/src/common/rabbitmq/message-exchange/message-exchange-queues.service.ts @@ -27,7 +27,7 @@ export class MessageExchangeQueuesService { await this.channel.assertQueue(queueName, { durable: true, messageTtl: 60000, - deadLetterExchange: 'message', + deadLetterExchange: routingKey, deadLetterRoutingKey: 'dead', }); await this.channel.bindQueue(queueName, exchangeName, routingKey); From 5f7fe963c36cf6bed6e6eef2d6118865b4bd9e35 Mon Sep 17 00:00:00 2001 From: Biniyam Date: Sat, 29 Mar 2025 11:51:40 +0300 Subject: [PATCH 07/16] rolled back the routing key to debug error inequivalent arg 'x-dead-letter-exchange' for queue 'database' in vhost '/': received 'moderator' but current is 'message' --- src/apps/moderator/main.ts | 2 +- .../message-exchange/message-exchange-queues.service.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/apps/moderator/main.ts b/src/apps/moderator/main.ts index e124e4e0..e3ac3f7e 100644 --- a/src/apps/moderator/main.ts +++ b/src/apps/moderator/main.ts @@ -13,7 +13,7 @@ async function bootstrap() { queueOptions: { durable: true, messageTtl: 60000, - deadLetterExchange: 'moderator', + deadLetterExchange: 'message', deadLetterRoutingKey: 'dead', }, noAck: false, diff --git a/src/common/rabbitmq/message-exchange/message-exchange-queues.service.ts b/src/common/rabbitmq/message-exchange/message-exchange-queues.service.ts index 9d654174..04ebe767 100644 --- a/src/common/rabbitmq/message-exchange/message-exchange-queues.service.ts +++ b/src/common/rabbitmq/message-exchange/message-exchange-queues.service.ts @@ -27,7 +27,7 @@ export class MessageExchangeQueuesService { await this.channel.assertQueue(queueName, { durable: true, messageTtl: 60000, - deadLetterExchange: routingKey, + deadLetterExchange: 'message', deadLetterRoutingKey: 'dead', }); await this.channel.bindQueue(queueName, exchangeName, routingKey); From b0ae9696a1b19e80856351604bf26a889c59e0f5 Mon Sep 17 00:00:00 2001 From: Biniyam Date: Sat, 29 Mar 2025 18:32:53 +0300 Subject: [PATCH 08/16] configuration to push conversation with the bot to the moderator queue --- src/modules/message/platform/telegram-message.ts | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/src/modules/message/platform/telegram-message.ts b/src/modules/message/platform/telegram-message.ts index a49b0f88..1d90bd4c 100644 --- a/src/modules/message/platform/telegram-message.ts +++ b/src/modules/message/platform/telegram-message.ts @@ -19,8 +19,19 @@ export class TelegramMessageStrategy implements MessageStrategy { ); if (!formattedMessage.conversationId) { await this.messageExchangeService.send('moderator', data); - } else if (formattedMessage.conversationId) { - await this.messageExchangeService.send('message', data); + } else { + const mentor = await this.prisma.mentor.findFirst({ + where: { + Conversation: { + some: { + id: formattedMessage.conversationId, + }, + }, + }, + }); + mentor.isBot + ? await this.messageExchangeService.send('moderator', data) + : await this.messageExchangeService.send('message', data); } return 'ok'; From 6947fdfbfe83defe5ac6032625ca4983b53a1cd5 Mon Sep 17 00:00:00 2001 From: Biniyam Date: Sat, 29 Mar 2025 18:51:48 +0300 Subject: [PATCH 09/16] updated the database consumer so that it delegates new messages to a bot mentor --- .../consumers/database-consumer/database-consumer.service.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/src/apps/consumers/database-consumer/database-consumer.service.ts b/src/apps/consumers/database-consumer/database-consumer.service.ts index cd41d9f5..11ffc4ff 100644 --- a/src/apps/consumers/database-consumer/database-consumer.service.ts +++ b/src/apps/consumers/database-consumer/database-consumer.service.ts @@ -60,6 +60,7 @@ export class DatabaseConsumerService { if (!message.conversationId) { const mentor = await tx.mentor.findFirst({ where: { + isBot: true, Account: { Channel: { some: { From 1f323036139fbd928a1668730f4314b22a8ab483 Mon Sep 17 00:00:00 2001 From: Biniyam Date: Sun, 30 Mar 2025 14:49:23 +0300 Subject: [PATCH 10/16] debugged (only one message was consumed from the moderator queue cause it was not getting and response from the consumer) --- src/apps/moderator/moderator.service.ts | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/src/apps/moderator/moderator.service.ts b/src/apps/moderator/moderator.service.ts index b2cf6582..1c1b9109 100644 --- a/src/apps/moderator/moderator.service.ts +++ b/src/apps/moderator/moderator.service.ts @@ -4,6 +4,14 @@ import { RmqContext } from '@nestjs/microservices'; @Injectable() export class ModeratorService { async handleMessage(message: any, context: RmqContext) { - console.log({ message: message, context: context }); + const channel = context.getChannelRef(); + const orgMsg = context.getMessage(); + try { + console.log({ message: message, context: context }); + channel.ack(orgMsg); + } catch (error) { + console.log(error.message); + channel.nack(orgMsg); + } } } From 45958189055d1eb0316bea7db863746789d06eb6 Mon Sep 17 00:00:00 2001 From: Biniyam Date: Sun, 30 Mar 2025 15:57:15 +0300 Subject: [PATCH 11/16] configured the moderator service to reply to conversations --- package-lock.json | 237 +++++++++++++++++++++++- package.json | 1 + src/apps/moderator/moderator.service.ts | 58 ++++++ 3 files changed, 294 insertions(+), 2 deletions(-) diff --git a/package-lock.json b/package-lock.json index dea1c069..aa651f52 100644 --- a/package-lock.json +++ b/package-lock.json @@ -10,6 +10,7 @@ "license": "UNLICENSED", "dependencies": { "@faker-js/faker": "^9.3.0", + "@langchain/google-genai": "^0.2.1", "@nestjs-modules/mailer": "^2.0.2", "@nestjs/common": "^10.4.8", "@nestjs/config": "^3.3.0", @@ -822,6 +823,13 @@ "dev": true, "license": "MIT" }, + "node_modules/@cfworker/json-schema": { + "version": "4.1.1", + "resolved": "https://registry.npmjs.org/@cfworker/json-schema/-/json-schema-4.1.1.tgz", + "integrity": "sha512-gAmrUZSGtKc3AiBL71iNWxDsyUC5uMaKKGdvzYsBoTW/xi42JQHl7eKV2OYzCUqvc+D2RCcf7EXY2iCyFIk6og==", + "license": "MIT", + "peer": true + }, "node_modules/@colors/colors": { "version": "1.5.0", "resolved": "https://registry.npmjs.org/@colors/colors/-/colors-1.5.0.tgz", @@ -1165,6 +1173,15 @@ "npm": ">=9.0.0" } }, + "node_modules/@google/generative-ai": { + "version": "0.24.0", + "resolved": "https://registry.npmjs.org/@google/generative-ai/-/generative-ai-0.24.0.tgz", + "integrity": "sha512-fnEITCGEB7NdX0BhoYZ/cq/7WPZ1QS5IzJJfC3Tg/OwkvBetMiVJciyaan297OvE4B9Jg1xvo0zIazX/9sGu1Q==", + "license": "Apache-2.0", + "engines": { + "node": ">=18.0.0" + } + }, "node_modules/@graphql-tools/merge": { "version": "9.0.8", "resolved": "https://registry.npmjs.org/@graphql-tools/merge/-/merge-9.0.8.tgz", @@ -1909,6 +1926,72 @@ "@jridgewell/sourcemap-codec": "^1.4.14" } }, + "node_modules/@langchain/core": { + "version": "0.3.43", + "resolved": "https://registry.npmjs.org/@langchain/core/-/core-0.3.43.tgz", + "integrity": "sha512-DwiSUwmZqcuOn7j8SFdeOH1nvaUqG7q8qn3LhobdQYEg5PmjLgd2yLr2KzuT/YWMBfjkOR+Di5K6HEdFmouTxg==", + "license": "MIT", + "peer": true, + "dependencies": { + "@cfworker/json-schema": "^4.0.2", + "ansi-styles": "^5.0.0", + "camelcase": "6", + "decamelize": "1.2.0", + "js-tiktoken": "^1.0.12", + "langsmith": ">=0.2.8 <0.4.0", + "mustache": "^4.2.0", + "p-queue": "^6.6.2", + "p-retry": "4", + "uuid": "^10.0.0", + "zod": "^3.22.4", + "zod-to-json-schema": "^3.22.3" + }, + "engines": { + "node": ">=18" + } + }, + "node_modules/@langchain/core/node_modules/ansi-styles": { + "version": "5.2.0", + "resolved": "https://registry.npmjs.org/ansi-styles/-/ansi-styles-5.2.0.tgz", + "integrity": "sha512-Cxwpt2SfTzTtXcfOlzGEee8O+c+MmUgGrNiBcXnuWxuFJHe6a5Hz7qwhwe5OgaSYI0IJvkLqWX1ASG+cJOkEiA==", + "license": "MIT", + "peer": true, + "engines": { + "node": ">=10" + }, + "funding": { + "url": "https://github.com/chalk/ansi-styles?sponsor=1" + } + }, + "node_modules/@langchain/core/node_modules/camelcase": { + "version": "6.3.0", + "resolved": "https://registry.npmjs.org/camelcase/-/camelcase-6.3.0.tgz", + "integrity": "sha512-Gmy6FhYlCY7uOElZUSbxo2UCDH8owEk996gkbrpsgGtrJLM3J7jGxl9Ic7Qwwj4ivOE5AWZWRMecDdF7hqGjFA==", + "license": "MIT", + "peer": true, + "engines": { + "node": ">=10" + }, + "funding": { + "url": "https://github.com/sponsors/sindresorhus" + } + }, + "node_modules/@langchain/google-genai": { + "version": "0.2.1", + "resolved": "https://registry.npmjs.org/@langchain/google-genai/-/google-genai-0.2.1.tgz", + "integrity": "sha512-30utucZZmL605SSIPY3EQeXYZJS9b5Hlrn+gUuW6TS6bvrONFhl/CqpdvmJ6ho8rXJfHDjETGqr8XcolcRGDGQ==", + "license": "MIT", + "dependencies": { + "@google/generative-ai": "^0.24.0", + "zod-to-json-schema": "^3.22.4" + }, + "engines": { + "node": ">=18" + }, + "peerDependencies": { + "@langchain/core": ">=0.3.17 <0.4.0" + } + }, "node_modules/@ljharb/through": { "version": "2.3.13", "resolved": "https://registry.npmjs.org/@ljharb/through/-/through-2.3.13.tgz", @@ -3133,6 +3216,13 @@ "dev": true, "license": "MIT" }, + "node_modules/@types/retry": { + "version": "0.12.0", + "resolved": "https://registry.npmjs.org/@types/retry/-/retry-0.12.0.tgz", + "integrity": "sha512-wWKOClTTiizcZhXnPY4wikVAwmdYHp8q6DmC+EJUzAMsycb7HB32Kh9RN4+0gExjmPmZSAQjgURXIGATPegAvA==", + "license": "MIT", + "peer": true + }, "node_modules/@types/semver": { "version": "7.5.8", "resolved": "https://registry.npmjs.org/@types/semver/-/semver-7.5.8.tgz", @@ -3193,6 +3283,13 @@ "@types/superagent": "*" } }, + "node_modules/@types/uuid": { + "version": "10.0.0", + "resolved": "https://registry.npmjs.org/@types/uuid/-/uuid-10.0.0.tgz", + "integrity": "sha512-7gqG38EyHgyP1S+7+xomFtL+ZNHcKv6DwNaCZmJmo1vgMugyF3TCnXVg4t1uk89mLNwnLtnY3TpOpCOyp1/xHQ==", + "license": "MIT", + "peer": true + }, "node_modules/@types/validator": { "version": "13.12.2", "resolved": "https://registry.npmjs.org/@types/validator/-/validator-13.12.2.tgz", @@ -5035,6 +5132,16 @@ "integrity": "sha512-9vAdYbHj6x2fLKC4+oPH0kFzY/orMZyG2Aj+kNylHxKGJ/Ed4dpNyAQYwJOdqO4zdM7XpVHmyejQDcQHrnuXbw==", "license": "MIT" }, + "node_modules/console-table-printer": { + "version": "2.12.1", + "resolved": "https://registry.npmjs.org/console-table-printer/-/console-table-printer-2.12.1.tgz", + "integrity": "sha512-wKGOQRRvdnd89pCeH96e2Fn4wkbenSP6LMHfjfyNLMbGuHEFbMqQNuxXqd0oXG9caIOQ1FTvc5Uijp9/4jujnQ==", + "license": "MIT", + "peer": true, + "dependencies": { + "simple-wcswidth": "^1.0.1" + } + }, "node_modules/constantinople": { "version": "4.0.1", "resolved": "https://registry.npmjs.org/constantinople/-/constantinople-4.0.1.tgz", @@ -5251,6 +5358,16 @@ } } }, + "node_modules/decamelize": { + "version": "1.2.0", + "resolved": "https://registry.npmjs.org/decamelize/-/decamelize-1.2.0.tgz", + "integrity": "sha512-z2S+W9X73hAUUki+N+9Za2lBlun89zigOyGrsax+KUQ6wKW4ZoWpEYBkGhQjwAjjDCkWxhY0VKEhk8wzY7F5cA==", + "license": "MIT", + "peer": true, + "engines": { + "node": ">=0.10.0" + } + }, "node_modules/dedent": { "version": "1.5.3", "resolved": "https://registry.npmjs.org/dedent/-/dedent-1.5.3.tgz", @@ -8792,6 +8909,16 @@ "license": "MIT", "optional": true }, + "node_modules/js-tiktoken": { + "version": "1.0.19", + "resolved": "https://registry.npmjs.org/js-tiktoken/-/js-tiktoken-1.0.19.tgz", + "integrity": "sha512-XC63YQeEcS47Y53gg950xiZ4IWmkfMe4p2V9OSaBt26q+p47WHn18izuXzSclCI73B7yGqtfRsT6jcZQI0y08g==", + "license": "MIT", + "peer": true, + "dependencies": { + "base64-js": "^1.5.1" + } + }, "node_modules/js-tokens": { "version": "4.0.0", "resolved": "https://registry.npmjs.org/js-tokens/-/js-tokens-4.0.0.tgz", @@ -8999,6 +9126,30 @@ "node": ">=6" } }, + "node_modules/langsmith": { + "version": "0.3.15", + "resolved": "https://registry.npmjs.org/langsmith/-/langsmith-0.3.15.tgz", + "integrity": "sha512-cv3ebg0Hh0gRbl72cv/uzaZ+KOdfa2mGF1s74vmB2vlNVO/Ap/O9RYaHV+tpR8nwhGZ50R3ILnTOwSwGP+XQxw==", + "license": "MIT", + "peer": true, + "dependencies": { + "@types/uuid": "^10.0.0", + "chalk": "^4.1.2", + "console-table-printer": "^2.12.1", + "p-queue": "^6.6.2", + "p-retry": "4", + "semver": "^7.6.3", + "uuid": "^10.0.0" + }, + "peerDependencies": { + "openai": "*" + }, + "peerDependenciesMeta": { + "openai": { + "optional": true + } + } + }, "node_modules/leac": { "version": "0.6.0", "resolved": "https://registry.npmjs.org/leac/-/leac-0.6.0.tgz", @@ -10031,6 +10182,16 @@ "node": ">= 6.0.0" } }, + "node_modules/mustache": { + "version": "4.2.0", + "resolved": "https://registry.npmjs.org/mustache/-/mustache-4.2.0.tgz", + "integrity": "sha512-71ippSywq5Yb7/tVYyGbkBggbU8H3u5Rz56fH60jGFgr8uHwxs+aSKeqmluIVzM0m0kB7xQjKS6qPfd0b2ZoqQ==", + "license": "MIT", + "peer": true, + "bin": { + "mustache": "bin/mustache" + } + }, "node_modules/mute-stream": { "version": "0.0.8", "resolved": "https://registry.npmjs.org/mute-stream/-/mute-stream-0.0.8.tgz", @@ -10375,7 +10536,6 @@ "resolved": "https://registry.npmjs.org/p-finally/-/p-finally-1.0.0.tgz", "integrity": "sha512-LICb2p9CB7FS+0eR1oqWnHhp0FljGLZCWBE9aix0Uye9W8LTQPwMTYVGWQWIw9RdQiDg4+epXQODwIYJtSJaow==", "license": "MIT", - "optional": true, "engines": { "node": ">=4" } @@ -10412,12 +10572,49 @@ "url": "https://github.com/sponsors/sindresorhus" } }, + "node_modules/p-queue": { + "version": "6.6.2", + "resolved": "https://registry.npmjs.org/p-queue/-/p-queue-6.6.2.tgz", + "integrity": "sha512-RwFpb72c/BhQLEXIZ5K2e+AhgNVmIejGlTgiB9MzZ0e93GRvqZ7uSi0dvRF7/XIXDeNkra2fNHBxTyPDGySpjQ==", + "license": "MIT", + "peer": true, + "dependencies": { + "eventemitter3": "^4.0.4", + "p-timeout": "^3.2.0" + }, + "engines": { + "node": ">=8" + }, + "funding": { + "url": "https://github.com/sponsors/sindresorhus" + } + }, + "node_modules/p-queue/node_modules/eventemitter3": { + "version": "4.0.7", + "resolved": "https://registry.npmjs.org/eventemitter3/-/eventemitter3-4.0.7.tgz", + "integrity": "sha512-8guHBZCwKnFhYdHr2ysuRWErTwhoN2X8XELRlrRwpmfeY2jjuUN4taQMsULKUVo1K4DvZl+0pgfyoysHxvmvEw==", + "license": "MIT", + "peer": true + }, + "node_modules/p-retry": { + "version": "4.6.2", + "resolved": "https://registry.npmjs.org/p-retry/-/p-retry-4.6.2.tgz", + "integrity": "sha512-312Id396EbJdvRONlngUx0NydfrIQ5lsYu0znKVUzVvArzEIt08V1qhtyESbGVd1FGX7UKtiFp5uwKZdM8wIuQ==", + "license": "MIT", + "peer": true, + "dependencies": { + "@types/retry": "0.12.0", + "retry": "^0.13.1" + }, + "engines": { + "node": ">=8" + } + }, "node_modules/p-timeout": { "version": "3.2.0", "resolved": "https://registry.npmjs.org/p-timeout/-/p-timeout-3.2.0.tgz", "integrity": "sha512-rhIwUycgwwKcP9yTOOFK/AKsAopjjCakVqLHePO3CC6Mir1Z99xT+R63jZxAT5lFZLa2inS5h+ZS2GvR99/FBg==", "license": "MIT", - "optional": true, "dependencies": { "p-finally": "^1.0.0" }, @@ -11698,6 +11895,16 @@ "dev": true, "license": "ISC" }, + "node_modules/retry": { + "version": "0.13.1", + "resolved": "https://registry.npmjs.org/retry/-/retry-0.13.1.tgz", + "integrity": "sha512-XQBQ3I8W1Cge0Seh+6gjj03LbmRFWuoszgK9ooCpwYIrhhoO80pfq4cUkU5DkknwfOfFteRwlZ56PYOGYyFWdg==", + "license": "MIT", + "peer": true, + "engines": { + "node": ">= 4" + } + }, "node_modules/reusify": { "version": "1.0.4", "resolved": "https://registry.npmjs.org/reusify/-/reusify-1.0.4.tgz", @@ -12223,6 +12430,13 @@ "url": "https://github.com/sponsors/isaacs" } }, + "node_modules/simple-wcswidth": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/simple-wcswidth/-/simple-wcswidth-1.0.1.tgz", + "integrity": "sha512-xMO/8eNREtaROt7tJvWJqHBDTMFN4eiQ5I4JRMuilwfnFcV5W9u7RUkueNkdw0jPqGMX36iCywelS5yilTuOxg==", + "license": "MIT", + "peer": true + }, "node_modules/sisteransi": { "version": "1.0.5", "resolved": "https://registry.npmjs.org/sisteransi/-/sisteransi-1.0.5.tgz", @@ -14116,6 +14330,25 @@ "dependencies": { "zen-observable": "0.8.15" } + }, + "node_modules/zod": { + "version": "3.24.2", + "resolved": "https://registry.npmjs.org/zod/-/zod-3.24.2.tgz", + "integrity": "sha512-lY7CDW43ECgW9u1TcT3IoXHflywfVqDYze4waEz812jR/bZ8FHDsl7pFQoSZTz5N+2NqRXs8GBwnAwo3ZNxqhQ==", + "license": "MIT", + "peer": true, + "funding": { + "url": "https://github.com/sponsors/colinhacks" + } + }, + "node_modules/zod-to-json-schema": { + "version": "3.24.5", + "resolved": "https://registry.npmjs.org/zod-to-json-schema/-/zod-to-json-schema-3.24.5.tgz", + "integrity": "sha512-/AuWwMP+YqiPbsJx5D6TfgRTc4kTLjsh5SOcd4bLsfUg2RcEXrFMJl1DGgdHy2aCfsIA/cr/1JM0xcB2GZji8g==", + "license": "ISC", + "peerDependencies": { + "zod": "^3.24.1" + } } } } diff --git a/package.json b/package.json index 0872a862..9f28d1e3 100644 --- a/package.json +++ b/package.json @@ -23,6 +23,7 @@ }, "dependencies": { "@faker-js/faker": "^9.3.0", + "@langchain/google-genai": "^0.2.1", "@nestjs-modules/mailer": "^2.0.2", "@nestjs/common": "^10.4.8", "@nestjs/config": "^3.3.0", diff --git a/src/apps/moderator/moderator.service.ts b/src/apps/moderator/moderator.service.ts index 1c1b9109..e7152880 100644 --- a/src/apps/moderator/moderator.service.ts +++ b/src/apps/moderator/moderator.service.ts @@ -1,13 +1,71 @@ import { Injectable } from '@nestjs/common'; import { RmqContext } from '@nestjs/microservices'; +import { ChatGoogleGenerativeAI } from '@langchain/google-genai'; +import { PrismaService } from 'src/modules/prisma/prisma.service'; +import { ChatExchangeService } from '../../common/rabbitmq/chat-exchange/chat-exchange.service'; +import { RabbitmqService } from '../../common/rabbitmq/rabbitmq.service'; +import { Chat } from 'src/types/chat'; @Injectable() export class ModeratorService { + private llm: ChatGoogleGenerativeAI; + public constructor( + private prisma: PrismaService, + private chatExchangeService: ChatExchangeService, + private rabbitmqService: RabbitmqService, + ) { + this.llm = new ChatGoogleGenerativeAI({ + apiKey: process.env.GOOGLE_GENAI_API_KEY, + model: 'gemini-2.0-flash', + }); + } + async handleMessage(message: any, context: RmqContext) { const channel = context.getChannelRef(); const orgMsg = context.getMessage(); try { console.log({ message: message, context: context }); + + const conversationId = + message.metadata.conversationId ?? + ( + await this.prisma.conversation.findFirst({ + where: { address: message.metadata.address }, + }) + )?.id; + + const messageHistory = await this.prisma.message.findMany({ + where: { + Threads: { + some: { + conversationId, + }, + }, + }, + orderBy: { + createdAt: 'asc', + }, + }); + + const messages = messageHistory.map(({ type, body }) => { + return `${type === 'RECEIVED' ? 'mentor' : 'mentee'}: ${body}`; + }); + messages.push( + 'system prompt: You are a human mentor, warm and approachable, guiding mentees with care and insight. Your goal is to understand their struggles as deeply as possible by asking thoughtful questions and making them feel heard. Keep the conversation light and engaging with a touch of humor, helping them relax and open up. Avoid robotic or dry responses—make your words feel natural, like a genuine human conversation. While being concise, ensure your responses hold meaning and value, never cutting out what truly matters. Keep it short, sweet, and impactful and your answers should be in plane text do not format it.', + ); + + const response = await this.llm.invoke(messages); + const chat: Chat = { + type: 'CHAT', + metadata: { + conversationId, + }, + payload: response, + }; + + const formattedData = await this.rabbitmqService.getChatEchangeData(chat); + + this.chatExchangeService.send('chat', formattedData); channel.ack(orgMsg); } catch (error) { console.log(error.message); From 5044db376d37afdba7d6103ad357e52859671055 Mon Sep 17 00:00:00 2001 From: Biniyam Date: Sun, 30 Mar 2025 16:00:42 +0300 Subject: [PATCH 12/16] included all dependencies in the moderator module --- src/apps/moderator/moderator.module.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/apps/moderator/moderator.module.ts b/src/apps/moderator/moderator.module.ts index 7bbd5c44..9173c04d 100644 --- a/src/apps/moderator/moderator.module.ts +++ b/src/apps/moderator/moderator.module.ts @@ -1,8 +1,11 @@ import { Module } from '@nestjs/common'; import { ModeratorService } from './moderator.service'; import { ModeratorController } from './moderator.controller'; +import { RabbitmqModule } from 'src/common/rabbitmq/rabbitmq.module'; +import { PrismaModule } from 'src/modules/prisma/prisma.module'; @Module({ + imports: [PrismaModule, RabbitmqModule], controllers: [ModeratorController], providers: [ModeratorService], }) From 3338b78ed4c0ceff5948a4b09963b46f86278636 Mon Sep 17 00:00:00 2001 From: Biniyam Date: Sun, 30 Mar 2025 16:09:45 +0300 Subject: [PATCH 13/16] debugging (I was trying to access the consumed message as object but was throwing error cause it is stringfied, so placed a parser based on type) --- src/apps/moderator/moderator.service.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/apps/moderator/moderator.service.ts b/src/apps/moderator/moderator.service.ts index e7152880..f149b7fc 100644 --- a/src/apps/moderator/moderator.service.ts +++ b/src/apps/moderator/moderator.service.ts @@ -20,11 +20,11 @@ export class ModeratorService { }); } - async handleMessage(message: any, context: RmqContext) { + async handleMessage(data: any, context: RmqContext) { const channel = context.getChannelRef(); const orgMsg = context.getMessage(); try { - console.log({ message: message, context: context }); + const message = typeof data == 'string' ? JSON.parse(data) : data; const conversationId = message.metadata.conversationId ?? @@ -33,6 +33,7 @@ export class ModeratorService { where: { address: message.metadata.address }, }) )?.id; + console.log(conversationId); const messageHistory = await this.prisma.message.findMany({ where: { From 7a89b61f626c05e292833773ae085c5b69d27f2b Mon Sep 17 00:00:00 2001 From: Biniyam Date: Sun, 30 Mar 2025 16:13:25 +0300 Subject: [PATCH 14/16] debugging (picked only the text part from gemini response since the whole response is non-sense when it is sent) --- src/apps/moderator/moderator.service.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/apps/moderator/moderator.service.ts b/src/apps/moderator/moderator.service.ts index f149b7fc..e0775ed3 100644 --- a/src/apps/moderator/moderator.service.ts +++ b/src/apps/moderator/moderator.service.ts @@ -61,7 +61,7 @@ export class ModeratorService { metadata: { conversationId, }, - payload: response, + payload: response.text, }; const formattedData = await this.rabbitmqService.getChatEchangeData(chat); From 7f052cc6bb54a5e21eed5d8b1dc1b3106f01688d Mon Sep 17 00:00:00 2001 From: Biniyam Date: Sun, 30 Mar 2025 16:14:40 +0300 Subject: [PATCH 15/16] removed a log placed for debugging --- src/apps/moderator/moderator.service.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/src/apps/moderator/moderator.service.ts b/src/apps/moderator/moderator.service.ts index e0775ed3..caf5a120 100644 --- a/src/apps/moderator/moderator.service.ts +++ b/src/apps/moderator/moderator.service.ts @@ -33,7 +33,6 @@ export class ModeratorService { where: { address: message.metadata.address }, }) )?.id; - console.log(conversationId); const messageHistory = await this.prisma.message.findMany({ where: { From 6f3e827adcd8a80f8d8ef76e639feaddf0299caf Mon Sep 17 00:00:00 2001 From: Biniyam Date: Sun, 30 Mar 2025 16:17:27 +0300 Subject: [PATCH 16/16] updated env.example with new variables I've used but not included --- .env.example | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.env.example b/.env.example index 05b73d93..fe1ad9ae 100644 --- a/.env.example +++ b/.env.example @@ -14,3 +14,5 @@ WEB_CALLBACK_URL="" RABBITMQ_URL="" CHAT_PORT="" HOST_URL="" +WEBSOCKET_URL="" +GOOGLE_GENAI_API_KEY=""