feat: rabbitMQ

src/app.module.ts

@@ -1,4 +1,9 @@
-import { Module } from '@nestjs/common';
+import {
+  MiddlewareConsumer,
+  Module,
+  NestModule,
+  RequestMethod,
+} from '@nestjs/common';
 import { AppController } from './app.controller';
 import { AppService } from './app.service';
 import * as process from 'process';
@@ -9,6 +14,7 @@ import { OffersModule } from './api/offers/offers.module';
 import { ConfigModule, ConfigService } from '@nestjs/config';
 import configuration from './config/configuration';
 import { HttpModule } from '@nestjs/axios';
+import { LoggerMiddleware } from './middleware/logger.middleware';
 
 @Module({
   imports: [
@@ -32,4 +38,10 @@ import { HttpModule } from '@nestjs/axios';
   controllers: [AppController],
   providers: [AppService],
 })
-export class AppModule {}
+export class AppModule implements NestModule {
+  configure(consumer: MiddlewareConsumer) {
+    consumer
+      .apply(LoggerMiddleware)
+      .forRoutes({ path: '*', method: RequestMethod.ALL });
+  }
+}

src/main.ts

@@ -5,11 +5,13 @@ import { Logger as NestLogger } from '@nestjs/common';
 import { DocumentBuilder, SwaggerModule } from '@nestjs/swagger';
 import 'dotenv/config';
 import * as process from 'process';
+import { setupRabbitMQ } from './rabbitmq/rabbitmq';
 
 async function bootstrap() {
   const app = await NestFactory.create(AppModule, { cors: { origin: '*' } });
   app.setGlobalPrefix('api/v1');
   app.useGlobalPipes(new ValidationPipe());
+  setupRabbitMQ();
 
   const config = new DocumentBuilder()
     .setTitle('Discounts-service api')

src/middleware/logger.middleware.ts

@@ -0,0 +1,47 @@
+import { Injectable, NestMiddleware } from '@nestjs/common';
+import { NextFunction, Request, Response } from 'express';
+import { v4 as uuidd } from 'uuid';
+import * as process from 'process';
+import { getRabbitMQChannel } from '../rabbitmq/rabbitmq';
+
+@Injectable()
+export class LoggerMiddleware implements NestMiddleware {
+  use(req: Request, res: Response, next: NextFunction) {
+    if (next) {
+      next();
+    }
+
+    const correlationIdHeader = 'X-Correlation-Id';
+    if (!req[correlationIdHeader]) {
+      req.headers[correlationIdHeader] = uuidd();
+    }
+
+    const timestamp = new Date().toISOString();
+    const urlString = `${req.protocol}://${req.get('host')}${req.originalUrl}`;
+    const serviceName = '[discounts-service]';
+    const message = `<* ${req.method} ${req.originalUrl} *>`;
+
+    res.on('finish', () => {
+      const { statusCode } = res;
+
+      const log = {
+        timestamp,
+        type: statusCode >= 500 ? 'ERROR' : statusCode >= 400 ? 'WARN' : 'INFO',
+        correlationId: req.headers[correlationIdHeader],
+        urlString,
+        message,
+        serviceName,
+      };
+
+      const rabbitMQChannel = getRabbitMQChannel();
+      if (rabbitMQChannel) {
+        rabbitMQChannel.publish(
+          process.env.RABBITMQ_EXCHANGE,
+          '',
+          Buffer.from(JSON.stringify(log)),
+        );
+        console.log('Log message sent to RabbitMQ:', log);
+      }
+    });
+  }
+}

src/rabbitmq/rabbitmq.ts

@@ -0,0 +1,29 @@
+import * as process from 'process';
+import * as amqp from 'amqplib/callback_api';
+
+const rabbitMQURL = process.env.RABBITMQ_URL;
+const rabbitMQExchange = process.env.EXCHANGE_NAME;
+const rabbitMQQueue = process.env.QUEUE_NAME;
+let rabbitMQChannel: amqp.Channel;
+
+export const setupRabbitMQ = () => {
+  try {
+    amqp.connect(rabbitMQURL, (error0, connection) => {
+      if (error0) throw error0;
+
+      connection.createChannel((error1, channel) => {
+        if (error1) throw error1;
+
+        channel.assertExchange(rabbitMQExchange, 'direct', { durable: true });
+        channel.assertQueue(rabbitMQQueue, { durable: true });
+        channel.bindQueue(rabbitMQQueue, rabbitMQExchange, '');
+        rabbitMQChannel = channel;
+        console.log('Connected to RabbitMQ');
+      });
+    });
+  } catch (e) {
+    console.log(e);
+  }
+};
+
+export const getRabbitMQChannel = () => rabbitMQChannel;