Merge pull request #5 from ricash-org/Aissata

Add RabbitMQ

Author: Yaishaa

Docker-compose.yml

@@ -10,6 +10,7 @@ services:
     env_file: .env
     depends_on:
       - postgres
+      - rabbitmq
     networks:
       - notification-net
 
@@ -27,6 +28,18 @@ services:
       - pgdata:/var/lib/postgresql/data
     networks:
       - notification-net
+  
+  rabbitmq:
+    image: rabbitmq:3-management
+    container_name: rabbitmq
+    ports:
+      - "5672:5672"
+      - "15672:15672"
+    environment:
+      RABBITMQ_DEFAULT_USER: ricash
+      RABBITMQ_DEFAULT_PASS: ricash123
+      
+
 
 networks:
   notification-net:

Dockerfile

@@ -31,5 +31,7 @@ COPY --from=builder /app/dist ./dist
 # Exposer le port
 EXPOSE 3000
 
+
 # Lancer la version compilée
 CMD ["node", "dist/server.js"]
+

package-lock.json

@@ -2,17 +2,18 @@
   "name": "notification-service",
   "version": "1.0.0",
   "description": "",
-  "main": "dist/server.js",
+  "main": "src/server.ts",
   "scripts": {
     "dev": "ts-node-dev src/server.ts",
-    "start": "node dist/server.js",
+    "start": "node src/server.ts",
     "build": "tsc"
   },
   "keywords": [],
   "author": "",
   "license": "ISC",
   "type": "commonjs",
   "dependencies": {
+    "amqplib": "^0.10.9",
     "body-parser": "^2.2.0",
     "dotenv": "^17.2.3",
     "express": "^5.1.0",
@@ -23,6 +24,7 @@
     "typeorm": "^0.3.27"
   },
   "devDependencies": {
+    "@types/amqplib": "^0.10.8",
     "@types/express": "^5.0.5",
     "@types/jsonwebtoken": "^9.0.10",
     "@types/node": "^24.10.0",

package.json

@@ -0,0 +1,95 @@
+// import amqp from "amqplib";
+
+// let channel: amqp.Channel;
+
+// export const connectRabbitMQ = async () => {
+//   try {
+//     const connection = await amqp.connect("amqp://admin:admin@rabbitmq:5672");
+//     channel = await connection.createChannel();
+//     console.log("RabbitMQ connecté");
+//   } catch (error) {
+//     console.error("Erreur RabbitMQ:", error);
+//   }
+// };
+
+// export const getChannel = () => channel;
+
+// src/messaging/rabbitmq.ts
+/**
+ * Connexion RabbitMQ partagée pour otp-service
+ * - garde la même connexion/channel pendant tout le runtime
+ * - gère reconnect automatique en cas de fermeture/erreur
+ *
+ * Variables d'environnement :
+ * - RABBITMQ_URL (ex: amqp://admin:admin@rabbitmq:5672)
+ */
+
+// src/messaging/rabbitmq.ts
+import * as amqp from "amqplib";
+import type { Connection, Channel } from "amqplib";
+
+let connection: amqp.Connection | null = null;
+let channel: Channel | null = null;
+
+// Nom des files
+export const QUEUE_MAIN = "notifications.main";
+export const QUEUE_RETRY = "notifications.retry";
+export const QUEUE_DLQ = "notifications.dlq";
+
+// Fonction de connexion + reconnexion automatique
+export async function ensureChannel() {
+  try {
+    console.log("Tentative de connexion à RabbitMQ...");
+
+   let connection = await amqp.connect(process.env.RABBITMQ_URL!);
+
+    // garder une référence locale non-nulle pour satisfaire TypeScript
+    const conn = connection!;
+
+    conn.on("error", (err) => {
+      console.error("⚠️ Erreur RabbitMQ :", err);
+     // connection = null;
+    });
+
+    conn.on("close", () => {
+      console.error("Connexion RabbitMQ fermée. Reconnexion...");
+      //connection = null;
+      setTimeout(ensureChannel, 3000);
+    });
+
+    channel = await conn.createChannel();
+    const ch = channel!;
+
+    console.log(" Connecté à RabbitMQ !");
+
+    // ---- Déclaration des files ----
+
+    // File principale
+    await ch.assertQueue(QUEUE_MAIN, { durable: true });
+
+    // File de retry avec TTL de 5s
+    await ch.assertQueue(QUEUE_RETRY, {
+      durable: true,
+      arguments: {
+        "x-message-ttl": 5000,
+        "x-dead-letter-exchange": "",
+        "x-dead-letter-routing-key": QUEUE_MAIN,
+      },
+    });
+
+    // Dead-Letter Queue
+    await ch.assertQueue(QUEUE_DLQ, { durable: true });
+
+    return ch;
+  } catch (err) {
+    console.error("Erreur de connexion à RabbitMQ :", err);
+    setTimeout(ensureChannel, 3000);
+  }
+}
+
+export function getRabbitChannel(): Channel {
+  if (!channel) {
+    throw new Error("RabbitMQ non connecté !");
+  }
+  return channel;
+}

src/config/rabbitmq.ts

@@ -1,5 +1,6 @@
 import { Request, Response } from "express";
 import { NotificationService } from "../services/notificationService";
+import { publishNotification } from "../messaging/publisher";
 
 const service = new NotificationService();
 
@@ -16,3 +17,11 @@ export const getNotifications = async (req: Request, res: Response) => {
   const list = await service.getAll();
   res.json(list);
 };
+
+
+export async function testRabbitMQ(req: Request, res: Response) {
+  const { queueName, message } = req.body;
+  await publishNotification(queueName, message);
+  res.json({ success: true });
+}
+

src/controllers/notificationController.ts

@@ -1,11 +1,12 @@
 import { Router } from "express";
 import { generateOtp, verifyOtp } from "./controllers/optController";
-import { envoyerNotification } from "./controllers/notificationController";
+import { envoyerNotification, testRabbitMQ } from "./controllers/notificationController";
 
 const router = Router();
 
 //  Notifications
 router.post("/notifications/envoyer", envoyerNotification);
+router.post("/rabbitmq", testRabbitMQ);
 
 //  OTP
 router.post("/otp/generate", generateOtp);

src/index.ts

@@ -0,0 +1,77 @@
+
+// import { getChannel } from "../config/rabbitmq";
+// import { NotificationService } from "../services/notificationService";
+
+// export const startConsumer = async () => {
+//   const channel = getChannel();
+//   const queue = "notifications_queue";
+//   const exchange = "notifications_exchange";
+
+//   await channel.assertExchange(exchange, "direct", { durable: true });
+//   await channel.assertQueue(queue, { durable: true });
+//   await channel.bindQueue(queue, exchange, "send-notification");
+
+//   console.log("📥 En attente de messages...");
+
+//   channel.consume(queue, async (msg) => {
+//     if (msg) {
+//       const data = JSON.parse(msg.content.toString());
+//       console.log("📩 Message reçu :", data);
+
+//       const service = new NotificationService();
+//       await service.envoyerNotification(data);
+
+//       channel.ack(msg);
+//     }
+//   });
+// };
+
+
+
+import { getRabbitChannel, QUEUE_MAIN, QUEUE_RETRY, QUEUE_DLQ } from "../config/rabbitmq";
+import { NotificationService } from "../services/notificationService";
+
+const notifService = new NotificationService();
+
+export async function startConsumer() {
+  const channel = getRabbitChannel();
+
+  console.log("Consumer prêt. En attente de messages...");
+
+  channel.consume(QUEUE_MAIN, async (msg) => {
+    if (!msg) return;
+
+    const content = JSON.parse(msg.content.toString());
+    console.log("Message reçu :", content);
+
+    try {
+      // Envoi de la vraie notification
+      await notifService.envoyerNotification(content);
+
+      channel.ack(msg);
+      console.log("Notification envoyée avec succès !");
+    } catch (error) {
+      console.error("Erreur lors du traitement :", error);
+
+      const retryCount = Number(msg.properties.headers?.["x-retries"] ?? 0);
+
+      if (retryCount < 3) {
+        console.log(`Retry ${retryCount + 1}/3...`);
+
+        channel.sendToQueue(QUEUE_RETRY, msg.content, {
+          headers: { "x-retries": retryCount + 1 },
+        });
+      } else {
+        console.log("Trop d'erreurs → envoi en DLQ");
+
+        const errorMessage = error instanceof Error ? error.message : String(error);
+
+        channel.sendToQueue(QUEUE_DLQ, msg.content, {
+          headers: { "x-final-error": errorMessage },
+        });
+      }
+
+      channel.ack(msg);
+    }
+  });
+}