feat: refator webhook to use queue

api/webhook.ts

@@ -1,8 +1,11 @@
 import express from "express";
 import dotenv from "dotenv";
-import { markChatAsRead, sendChatbotReply } from "../services/whatsapp";
+import {  sendChatbotReply } from "../services/whatsapp";
 import { queryToDify } from "../services/dify";
 import { queryToRasa } from "../services/rasa";
+import { sendToQueue } from "../services/queue";
+import {ParamsDictionary, Request} from "express-serve-static-core";
+import { ParsedQs } from "qs";
 
 dotenv.config();
 
@@ -46,9 +49,7 @@ webhookRoutes.post("/", async (req, res) => {
   }
 
   // aknowledge that the message has been read and be processed
-  await markChatAsRead(message.id);
-
-  let chatbotReply = null;
+  await sendToQueue('markChatAsRead', [message.id]);
   let queryText = "";
 
   switch (message.type) {
@@ -71,21 +72,27 @@ webhookRoutes.post("/", async (req, res) => {
     return;
   }
 
+  await sendToQueue('queryPlatform', [req, queryText, message.from]);
+
+
+  res.sendStatus(200);
+});
+
+export const queryPlatform = async (req:Request<ParamsDictionary, unknown, unknown, ParsedQs, Record<string, unknown>>, queryText: string, from: string)=>{
+  let chatbotReply = null;
+
   if (CONNECTION_PLATFORM === DIFY) {
     chatbotReply = await queryToDify({ req, query: queryText });
   } else if (CONNECTION_PLATFORM === RASA) {
     chatbotReply = await queryToRasa({ req, query: queryText });
   }
   console.log("Chatbot Reply:\n", chatbotReply);
 
-  if (!chatbotReply || !chatbotReply.text) {
-    res.sendStatus(200);
+  if (!chatbotReply || !chatbotReply.text) {    
     return;
   }
+  await sendChatbotReply({ to: from, chatbotReply });
 
-  await sendChatbotReply({ to: message.from, chatbotReply });
-
-  res.sendStatus(200);
-});
+}
 
 export default webhookRoutes;

app.ts

@@ -3,13 +3,20 @@ import morgan from "morgan";
 import dotenv from "dotenv";
 import bodyParser from "body-parser";
 
-import webhookRoutes from "./api/webhook";
+import webhookRoutes, { queryPlatform } from "./api/webhook";
+import { setupQueueHandlers } from "./services/queue";
+import { markChatAsRead } from "./services/whatsapp";
 
 // Load environment variables from .env file
 dotenv.config();
 
 const { NODE_ENV } = process.env;
 
+setupQueueHandlers({
+  'markChatAsRead': markChatAsRead,
+  'queryPlatform': queryPlatform
+})
+
 const app = express();
 app.use(bodyParser.urlencoded({ extended: false }));
 app.use(bodyParser.json());

package.json

@@ -14,6 +14,7 @@
   "author": "",
   "license": "ISC",
   "dependencies": {
+    "amqplib": "^0.10.4",
     "axios": "^1.7.2",
     "body-parser": "^1.20.2",
     "dotenv": "^16.4.5",
@@ -26,6 +27,7 @@
   "devDependencies": {
     "@eslint/js": "^8.56.0",
     "@jest/globals": "^29.7.0",
+    "@types/amqplib": "^0.10.5",
     "@types/express": "^4.17.21",
     "@types/jest": "^29.5.12",
     "@types/morgan": "^1.9.9",

services/queue.ts

@@ -0,0 +1,64 @@
+import amqp from "amqplib";
+
+
+const { RABBIT_MQ_URL } = process.env;
+
+const QUEUE_NAME = "WEBHOOK_QUEUE";
+
+export const sendToQueue = async (methodName: string, params: unknown[])=>{
+   let connection;
+    try {
+
+      connection = await amqp.connect(RABBIT_MQ_URL || "");
+      const channel = await connection.createChannel();
+
+      await channel.assertQueue(QUEUE_NAME, { durable: false });
+      const payload = {
+        methodName, 
+        params
+      };
+      channel.sendToQueue(QUEUE_NAME, Buffer.from(JSON.stringify(payload)));
+      console.log(" [x] Sent '%s'", payload);
+      await channel.close();
+    } catch (err) {
+      console.error(err);
+    } finally {
+      if (connection) await connection.close();
+    }  
+}
+
+export const setupQueueHandlers = async (handlerMap:Record<string, unknown>): Promise<void> => {
+    try {
+      console.info(`Connecting to ${RABBIT_MQ_URL}...`)
+      const connection = await amqp.connect(RABBIT_MQ_URL || "");
+      console.info(`Queue connected to ${RABBIT_MQ_URL}`);
+
+      const channel = await connection.createChannel();
+
+      process.once("SIGINT", async () => {
+        await channel.close();
+        await connection.close();
+      });
+
+      await channel.assertQueue(QUEUE_NAME, { durable: false });
+      await channel.consume(
+        QUEUE_NAME,
+        (message) => {
+          if (message) {
+            const payload = JSON.parse(message.content.toString());
+            if(payload.methodName){
+              const handler: (params:[])=>unknown | unknown = handlerMap[payload.methodName] as ((params:[])=>unknown);
+              if(handler){
+                handler.apply(this, payload.params);
+              }
+            }
+          }
+        },
+        { noAck: true }
+      );
+
+      console.log(" [*] Waiting for messages. To exit press CTRL+C");
+    } catch (err) {
+      console.warn(err);
+    }
+  }