feat: add errorLogsService to error-handle middleware to send errors …

…to NOC app

package.json

@@ -14,6 +14,7 @@
   "author": "J.Alberto",
   "license": "ISC",
   "devDependencies": {
+    "@types/amqplib": "^0.10.5",
     "@types/cors": "^2.8.17",
     "@types/express": "^4.17.21",
     "@types/jsonwebtoken": "^9.0.6",
@@ -30,6 +31,8 @@
     "@langchain/openai": "^0.0.14",
     "@supabase/supabase-js": "^2.39.8",
     "@types/morgan": "^1.9.9",
+    "amqp-connection-manager": "^4.1.14",
+    "amqplib": "^0.10.3",
     "cors": "^2.8.5",
     "dotenv": "^16.4.5",
     "env-var": "^7.4.1",

src/config/envs.ts

@@ -13,4 +13,6 @@ export const envs = {
   MONGO_DB_URL: get('MONGO_DB_URL').required().asString(),
 
   JWT_SEED: get('JWT_SEED').required().asString(),
+
+  RABBITMQ_URL: get('RABBITMQ_URL').required().asString(),
 };

src/presentation/chatbot/controller.ts

@@ -21,7 +21,6 @@ export class ChatbotController {
     res.status(200);
 
     for await (const chunk of readable) {
-      console.log({ chunk });
       res.write(chunk);
       await new Promise((resolve) => setTimeout(resolve, 50));
     }

src/presentation/middlewares/error-handler.middleware.ts

@@ -3,15 +3,21 @@ import { NextFunction, Request, Response } from 'express';
 import { CustomAPIError } from '../../domain/errors';
 import { HttpCodes } from '../../config/http-status-codes.adapter';
 import { MulterError } from 'multer';
+import { QueueService } from '../services/queue.service';
+import { envs } from '../../config/envs';
 
 /**
  * Middleware class for handling errors.
  */
 export class ErrorHandlerMiddleware {
+  errorLogsService: QueueService = new QueueService(
+    envs.RABBITMQ_URL,
+    'error-notification'
+  );
   /**
    * Handles errors and sends appropriate responses.
    */
-  static handle(err: Error, _req: Request, res: Response, _next: NextFunction) {
+  handle(err: Error, _req: Request, res: Response, _next: NextFunction) {
     console.log({ err });
 
     let message, statusCode;
@@ -22,6 +28,12 @@ export class ErrorHandlerMiddleware {
       message = 'Unknown error';
     }
 
+    // Handle jwt expired errors
+    if (err?.name && err?.name === 'TokenExpiredError') {
+      statusCode = 401;
+      message = 'JWT token expired';
+    }
+
     // Handle CustomAPIError
     if (err instanceof CustomAPIError) {
       statusCode = err.statusCode;
@@ -34,6 +46,15 @@ export class ErrorHandlerMiddleware {
       message = err.message;
     }
 
+    this.errorLogsService.addMessageToQueue(
+      {
+        message: message,
+        level: statusCode === 500 ? 'high' : 'medium',
+        origin: 'assistant-service',
+      },
+      'error-logs'
+    );
+
     return res.status(statusCode || HttpCodes.INTERNAL_SERVER_ERROR).json({
       name: err?.name || 'Error',
       message: message || err?.message,

src/presentation/server.ts

@@ -35,7 +35,9 @@ export class Server {
     this.app.use(this.routes);
 
     //* Error Handler Middleware
-    this.app.use(ErrorHandlerMiddleware.handle);
+    const errorHandlerMiddleware = new ErrorHandlerMiddleware();
+
+    this.app.use(errorHandlerMiddleware.handle);
 
     this.serverListener = this.app.listen(this.port, async () => {
       try {

src/presentation/services/queue.service.ts

@@ -0,0 +1,53 @@
+import { Channel } from 'amqplib';
+import amqp, { ChannelWrapper } from 'amqp-connection-manager';
+
+/**
+ * ProducerService class for sending messages to RabbitMQ queues.
+ */
+export class QueueService {
+  private channelWrapper: ChannelWrapper;
+
+  /**
+   * Constructs an instance of ProducerService.
+   * @param rabbitmqUrl - URL of the RabbitMQ server.
+   * @param exchange - Name of the exchange to publish messages.
+   */
+  constructor(
+    private readonly rabbitmqUrl: string,
+    private readonly exchange: string
+  ) {
+    // Establishes connection to RabbitMQ server and creates a channel wrapper.
+    const connection = amqp.connect(this.rabbitmqUrl);
+    this.channelWrapper = connection.createChannel({
+      // Ensures the exchange is declared upon channel creation.
+      setup: (channel: Channel) => {
+        return channel.assertExchange(exchange, 'direct', {
+          durable: true,
+        });
+      },
+    });
+    console.log(`${this.exchange} exchange created`);
+  }
+
+  /**
+   * Adds a message to the specified queue in the exchange.
+   * @param payload - Data to be sent in the message.
+   * @param queue - Name of the queue to send the message.
+   */
+  async addMessageToQueue(payload: any, queue: string) {
+    try {
+      console.log({ payload, queue });
+      // Publishes the message to the specified queue in the exchange.
+      await this.channelWrapper.publish(
+        this.exchange,
+        queue,
+        Buffer.from(JSON.stringify(payload)),
+        { persistent: true }
+      );
+
+      console.log('Message sent to queue');
+    } catch (error) {
+      console.log('Error sending message to queue', error);
+    }
+  }
+}