infra/docker/Dockerfile

@@ -1,5 +0,0 @@
-FROM rabbitmq:3.8-management
-RUN apt-get update && apt-get install -y wget unzip
-RUN wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v3.8.0/rabbitmq_delayed_message_exchange-3.8.0.ez
-RUN mv rabbitmq_delayed_message_exchange-3.8.0.ez $RABBITMQ_HOME/plugins/
-RUN rabbitmq-plugins enable --offline rabbitmq_delayed_message_exchange

infra/docker/docker-compose.yaml

@@ -16,20 +16,25 @@ services:
     expose:
       - 27017
   rabbitmq:
-    image: todorabbit:latest
+    image: heidiks/rabbitmq-delayed-message-exchange:latest
     restart: always
     ports:
       - 5672:5672
       - 15672:15672
     volumes:
-      - rabbitmq_volume:/var/lib/rabbitmq
+      - rabbitmq_vol:/var/lib/rabbitmq
       - ./rabbitmq-init-scripts/init.sh:/docker-entrypoint-initdb.d/init.sh
     environment:
       - RABBITMQ_DEFAULT_USER=${RABBITMQ_DEFAULT_USER}
       - RABBITMQ_DEFAULT_PASS=${RABBITMQ_DEFAULT_PASS}
+    command: |
+      bash -c "
+        rabbitmq-plugins enable rabbitmq_delayed_message_exchange && \
+        rabbitmq-server
+      "
     expose:
       - 5672
       - 15672
 volumes:
   mongodb_vol:
-  rabbitmq_volume:
+  rabbitmq_vol:

notification-service/src/rabbitmq/RabbitMQ.ts

@@ -1,20 +1,26 @@
-import amqp, { ConsumeMessage } from 'amqplib';
+import amqp from 'amqplib';
 import { ITodo } from '../interfaces/ITodo';
 import { MongoDbModel } from '../mongodb/MongoDb';
 
 const env = require('dotenv').config().parsed;
 
+
 export class RabbitMQ {
   channel: amqp.Channel;
   queueName: string;
-  exchangeName: string;
-
   mongoClient: MongoDbModel;
+  exchange: string;
+  queue: string;
+  routingKey: string;
+
   constructor() {
-    this.queueName = env.RABBITMQ_QUEUE_NAME;
-
     this.mongoClient = new MongoDbModel();
+    
+    this.exchange = 'delayed_exchange';
+    this.queue = 'delayed_queue';
+    this.routingKey = 'delayed_routing_key'
 
+    this.queueName = env.RABBITMQ_QUEUE_NAME;
     this.connect().then(() => {
       console.log('RabbitMQ connected');
     }).catch((error) => {
@@ -23,22 +29,20 @@ export class RabbitMQ {
   }
 
   async connect() {
-    try {
+    const connection = await amqp.connect({
-      const connection = await amqp.connect({
+      protocol: 'amqp',
-        protocol: 'amqp',
+      hostname: env.RABBITMQ_HOST,
-        hostname: env.RABBITMQ_HOST,
+      port: parseInt(env.RABBITMQ_PORT),
-        port: parseInt(env.RABBITMQ_PORT),
+      username: env.RABBITMQ_USERNAME,
-        username: env.RABBITMQ_USERNAME,
+      password: env.RABBITMQ_PASSWORD,
-        password: env.RABBITMQ_PASSWORD,
+    });
-      });
+    this.channel = await connection.createChannel();
+    await this.channel.assertExchange(this.exchange, 'x-delayed-message', { durable: true, autoDelete: false, arguments: { 'x-delayed-type': 'direct' } });
+    await this.channel.assertQueue(this.queueName, { durable: true });
+    await this.channel.bindQueue(this.queueName, this.exchange, this.routingKey);
 
-      this.channel = await connection.createChannel();
+
-      await this.channel.assertQueue(this.queueName);
+    await this.channel.assertQueue(this.queueName, { durable: true });
-      console.log('Channel and queue asserted successfully #####');
-    } catch (error) {
-      console.error('Error connecting to RabbitMQ:', error);
-      throw error;
-    }
   }
 
   async create(payload: ITodo, delayTimeForQueue: number) {
@@ -55,43 +59,33 @@ export class RabbitMQ {
   }
 
   async startConsumer() {
-    this.channel.assertQueue(this.queueName);
+    try {
-    console.log("Consuming");
+      await this.channel.prefetch(1);
-    this.channel.consume(this.queueName, (message: ConsumeMessage | null) => {
+      console.log('Consumer started, waiting for messages...');
-      if (message) {
+      this.channel.consume(this.queueName, async (message) => {
-        const todo = JSON.parse(message.content.toString());
+        if (message) {
-        this.channel.ack(message);
+          const { payload, delayTimeForQueue } = JSON.parse(message.content.toString()) as {
-        console.log('Received notification:', todo);
+            payload: ITodo;
-      }
+            delayTimeForQueue: number;
-    });
+          };
+
+          console.log('Received notification:', payload);
+
+          try {
+            await this.mongoClient.updateTodoStatus(payload);
+            console.log('Updated todo status in the DB');
+          } catch {
+            await this.create(payload, delayTimeForQueue);
+            this.channel.ack(message);
+            console.log('Published new message with delay, THE DB IS DOWN!: ', delayTimeForQueue);
+            return;
+          }
+
+          this.channel.ack(message);
+        }
+      });
+    } catch (error) {
+      console.error('Error consuming messages from RabbitMQ:', error);
+    }
   }
-
+}
-
-  // try { 
-  //   console.log('Consumer started, waiting for messages...');
-  //   this.channel.consume(this.queueName, async (message) => {
-  //     if (message) {
-  //       const { payload, delayTimeForQueue } = JSON.parse(message.content.toString()) as {
-  //         payload: ITodo;
-  //         delayTimeForQueue: number;
-  //       };
-
-  //       console.log('Received notification:', payload);
-
-  //       try {
-  //         await this.mongoClient.updateTodoStatus(payload);
-  //         console.log('Updated todo status in the DB');
-  //       } catch {
-  //         await this.create(payload, delayTimeForQueue);
-  //         this.channel.ack(message);
-  //         console.log('Published new message with delay, THE DB IS DOWN!: ', delayTimeForQueue);
-  //         return;
-  //       }
-
-  //       this.channel.ack(message);
-  //     }
-  //   });
-  // } catch (error) {
-  //   console.error('Error consuming messages from RabbitMQ:', error);
-  // }
-}

request.http

@@ -7,9 +7,9 @@ POST http://localhost:3000/todo
 Content-Type: application/json
 
 {
-    "title": "-2!!!!NEW!!!!",
+    "title": "Go to the gymNEW!!!",
     "description": "Go to the gym at 8pm",
-    "due_date": "2023-07-09T19:00:00.891Z"
+    "due_date": "2023-07-08T15:24:00.000Z"
 }
 
 ### update request

todo-service/package.json

@@ -15,14 +15,11 @@
     "aws-sdk": "^2.1413.0",
     "dotenv": "^16.3.1",
     "express": "^4.18.2",
-    "luxon": "^3.3.0",
     "moment": "^2.29.4",
-    "moment-timezone": "^0.5.43",
     "mongoose": "^7.3.1"
   },
   "devDependencies": {
     "@types/amqplib": "^0.10.1",
-    "@types/express": "^4.17.17",
+    "@types/express": "^4.17.17"
-    "@types/luxon": "^3.3.0"
   }
 }

todo-service/src/middleware/createTodoMiddleWare.ts

@@ -1,40 +1,36 @@
-import { Request, Response, NextFunction } from 'express';
 import { ApiError } from '../utils/ApiError';
-import { DateTime } from 'luxon';
+import { Request, Response, NextFunction } from 'express';
-import moment from 'moment-timezone';
+import moment from 'moment';
 
-const createTodoMiddleWare = async (req: Request, res: Response, next: NextFunction) => {
+const createTodoMiddleware = async (req: Request, res: Response, next: NextFunction) => {
   const { title, description, due_date } = req.body;
+
   if (!title || !due_date) {
     const error = new ApiError(`${!title ? 'title' : 'due_date'} is required`, 400, 'Bad Request');
     return next(error);
   }
+
   try {
-    if(new Date(due_date) < new Date()) {
+    const inputDate = new Date(due_date);
-      const error = new ApiError(`due_date must be greater than current date`, 400, 'Bad Request');
+    if (isNaN(inputDate.getTime()) || inputDate < new Date()) {
+      const error = new ApiError('due_date must be greater than current date', 400, 'Bad Request');
       return next(error);
-    } 
+    }
   } catch {
-    const error = new ApiError(`due_date must be ISO `, 400, 'Bad Request');
+    const parsedDate = moment(due_date, 'YYYY-MM-DD HH:mm:ss');
-    return next(error);
+    if (!parsedDate.isValid() || parsedDate < moment()) {
+      const error = new ApiError('due_date must be greater than current date', 400, 'Bad Request');
+      return next(error);
+    }
   }
 
-  
   if (!description) {
     req.body.description = '';
   }
 
-  //check if date is valid, this is valid: 2023-07-08T14:00:00.000Z
-  const date = DateTime.fromISO(due_date);
-  if (!date.isValid) {
-    const error = new ApiError(`due_date must be a valid date Format`, 400, 'Bad Request');
-    return next(error);
-  }
-
-
   next();
 }
 
 export {
-  createTodoMiddleWare
+  createTodoMiddleware
-}
+}

todo-service/src/rabbitmq/RabbitMQ.ts

@@ -1,18 +1,21 @@
-import amqp, { Options, ConsumeMessage, Channel } from 'amqplib';
+import amqp, { Options } from 'amqplib';
 import { ITodo } from '../schemas/todoSchema';
 
 const env = require('dotenv').config().parsed;
 
 export class RabbitMQ {
-  connection: amqp.Connection;
   channel: amqp.Channel;
-  queue: string;
+  queueName: string;
   exchange: string;
-
+  queue: string;
+  routingKey: string;
 
   constructor() {
-    this.queue = env.RABBITMQ_QUEUE_NAME;
+    this.queueName = env.RABBITMQ_QUEUE_NAME;
+
     this.exchange = 'delayed_exchange';
+    this.queue = 'delayed_queue';
+    this.routingKey = 'delayed_routing_key'
 
     this.connect().then(() => {
       console.log('RabbitMQ connected');
@@ -22,26 +25,20 @@ export class RabbitMQ {
   }
 
   async connect() {
+    console.log("IN CONNECT")
     try {
-      this.connection = await amqp.connect({
+      const connection = await amqp.connect({
         protocol: 'amqp',
         hostname: env.RABBITMQ_HOST,
         port: parseInt(env.RABBITMQ_PORT),
         username: env.RABBITMQ_USERNAME,
         password: env.RABBITMQ_PASSWORD,
       });
-
+      this.channel = await connection.createChannel();
-      this.channel = await this.connection.createChannel();
+      await this.channel.assertExchange(this.exchange, 'x-delayed-message', { durable: true, autoDelete: false, arguments: { 'x-delayed-type': 'direct' } });
-      await this.channel.assertQueue(this.queue);
+      await this.channel.assertQueue(this.queueName, { durable: true });
-      await this.channel.assertExchange(this.exchange, 'x-delayed-message', {
+      await this.channel.bindQueue(this.queueName, this.exchange, this.routingKey);
-        durable: true,
+      console.log('Channel and queue asserted successfully');
-        arguments: {
-          'x-delayed-type': 'direct'
-        }
-      });
-      await this.channel.bindQueue(this.queue, this.exchange, this.queue);
-
-      console.log('Channel and queue asserted successfully #####');
     } catch (error) {
       console.error('Error connecting to RabbitMQ:', error);
       throw error;
@@ -49,26 +46,16 @@ export class RabbitMQ {
   }
 
   async create(payload: ITodo) {
-
+    console.log("IN CREATE!")
-    const delayTimeForQueue = this.calculateDelayTimeForQueue(payload);
+    const delayTimeForQueue: number = new Date(payload.due_date).getTime() - Date.now();
-    console.log("The Queue will be delayed for: ", delayTimeForQueue, " ms");
+    const message = JSON.stringify({ payload, delayTimeForQueue: delayTimeForQueue });
-
+    const options: Options.Publish = { persistent: true, headers: { 'x-delay': delayTimeForQueue } };
-    const message = JSON.stringify({ payload });
-    const options: Options.Publish = { headers: { 'x-delay': delayTimeForQueue } };
     try {
-      await this.channel.publish(this.exchange, this.queue, Buffer.from(message), 
+      await this.channel.publish(this.exchange, this.routingKey, Buffer.from(message), options);
-      options
+      console.log('Message sent to the queue');
-      )
-      console.log(`Queue name is: ${this.queue}`)
     } catch (error) {
       console.error('Error sending message to RabbitMQ:', error);
       throw error;
     }
-
-  }
-
-  calculateDelayTimeForQueue(payload: ITodo) {
-    const delayTime = payload.due_date.getTime() - Date.now();
-    return delayTime;
   }
 }

todo-service/src/routes/todoRouter.ts

@@ -1,6 +1,6 @@
 import { Router } from 'express';
 import { TodoController } from '../controllers/todoController';
-import { createTodoMiddleWare } from '../middleware/createTodoMiddleWare';
+import { createTodoMiddleware } from '../middleware/createTodoMiddleWare';
 
 class TodoRouter {
   router: Router;
@@ -15,8 +15,8 @@ class TodoRouter {
   private setRoutes() {
     this.router.get('/', this.todoController.getAll);
     this.router.get('/:id', this.todoController.getOne);
-    this.router.post('/', createTodoMiddleWare, this.todoController.createOne);
+    this.router.post('/', createTodoMiddleware, this.todoController.createOne);
-    this.router.put('/:id', createTodoMiddleWare, this.todoController.updateOne);
+    this.router.put('/:id', createTodoMiddleware, this.todoController.updateOne);
     this.router.delete('/:id', this.todoController.deleteOne);
     this.router.delete('/', this.todoController.removeAll)