Merge pull request #451 from judsonjuniorr/feat/rabbit-per-events

RabbitMQ improvements

Docker/.env.example

@@ -48,6 +48,8 @@ REDIS_URI=redis://redis:6379
 REDIS_PREFIX_KEY=evdocker
 
 RABBITMQ_ENABLED=false
+RABBITMQ_RABBITMQ_MODE=global
+RABBITMQ_EXCHANGE_NAME=evolution_exchange
 RABBITMQ_URI=amqp://guest:guest@rabbitmq:5672
 
 WEBSOCKET_ENABLED=false

Dockerfile

@@ -1,6 +1,6 @@
 FROM node:20.7.0-alpine AS builder
 
-LABEL version="1.6.2" description="Api to control whatsapp features through http requests." 
+LABEL version="1.7.0" description="Api to control whatsapp features through http requests." 
 LABEL maintainer="Davidson Gomes" git="https://github.com/DavidsonGomes"
 LABEL contact="contato@agenciadgcode.com"
 
@@ -63,6 +63,8 @@ ENV REDIS_URI=redis://redis:6379
 ENV REDIS_PREFIX_KEY=evolution
 
 ENV RABBITMQ_ENABLED=false
+ENV RABBITMQ_MODE=global
+ENV RABBITMQ_EXCHANGE_NAME=evolution_exchange
 ENV RABBITMQ_URI=amqp://guest:guest@rabbitmq:5672
 
 ENV WEBSOCKET_ENABLED=false

package.json

@@ -1,6 +1,6 @@
 {
   "name": "evolution-api",
-  "version": "1.6.2",
+  "version": "1.7.0",
   "description": "Rest api for communication with WhatsApp",
   "main": "./dist/src/main.js",
   "scripts": {
@@ -90,6 +90,7 @@
     "yamljs": "^0.3.0"
   },
   "devDependencies": {
+    "@types/amqplib": "^0.10.5",
     "@types/compression": "^1.7.2",
     "@types/cors": "^2.8.13",
     "@types/express": "^4.17.17",

src/config/env.config.ts

@@ -71,6 +71,8 @@ export type Redis = {
 
 export type Rabbitmq = {
   ENABLED: boolean;
+  MODE: 'isolated' | 'global' | 'single';
+  EXCHANGE_NAME: string; // available for global and single, isolated mode will use instance name as exchange
   URI: string;
 };
 
@@ -283,6 +285,8 @@ export class ConfigService {
       },
       RABBITMQ: {
         ENABLED: process.env?.RABBITMQ_ENABLED === 'true',
+        MODE: (process.env?.RABBITMQ_MODE as Rabbitmq['MODE']) || 'isolated',
+        EXCHANGE_NAME: process.env?.RABBITMQ_EXCHANGE_NAME || 'evolution_exchange',
         URI: process.env.RABBITMQ_URI || '',
       },
       SQS: {

src/dev-env.yml

@@ -84,6 +84,8 @@ REDIS:
 
 RABBITMQ:
   ENABLED: false
+  MODE: "global"
+  EXCHANGE_NAME: "evolution_exchange"
   URI: "amqp://guest:guest@localhost:5672"
 
 SQS:

src/docs/swagger.yaml

@@ -25,7 +25,7 @@ info:
     </font>
 
     [![Run in Postman](https://run.pstmn.io/button.svg)](https://god.gw.postman.com/run-collection/26869335-5546d063-156b-4529-915f-909dd628c090?action=collection%2Ffork&source=rip_markdown&collection-url=entityId%3D26869335-5546d063-156b-4529-915f-909dd628c090%26entityType%3Dcollection%26workspaceId%3D339a4ee7-378b-45c9-b5b8-fd2c0a9c2442)
-  version: 1.6.2
+  version: 1.7.0
   contact:
     name: DavidsonGomes
     email: contato@agenciadgcode.com

src/libs/amqp.server.ts

@@ -1,16 +1,34 @@
-import * as amqp from 'amqplib/callback_api';
+import { Channel, connect } from 'amqplib/callback_api';
 
-import { configService, Rabbitmq } from '../config/env.config';
+import { configService, HttpServer, Rabbitmq } from '../config/env.config';
 import { Logger } from '../config/logger.config';
+import { Events } from '../whatsapp/types/wa.types';
 
 const logger = new Logger('AMQP');
 
-let amqpChannel: amqp.Channel | null = null;
+const parseEvtName = (evt: string) => evt.replace(/_/g, '.').toLowerCase();
+
+const globalQueues: { [key: string]: Events[] } = {
+  contacts: [Events.CONTACTS_SET, Events.CONTACTS_UPDATE, Events.CONTACTS_UPSERT],
+  messages: [
+    Events.MESSAGES_DELETE,
+    Events.MESSAGES_SET,
+    Events.MESSAGES_UPDATE,
+    Events.MESSAGES_UPSERT,
+    Events.MESSAGING_HISTORY_SET,
+    Events.SEND_MESSAGE,
+  ],
+  chats: [Events.CHATS_DELETE, Events.CHATS_SET, Events.CHATS_UPDATE, Events.CHATS_UPSERT],
+  groups: [Events.GROUPS_UPDATE, Events.GROUPS_UPSERT, Events.GROUP_PARTICIPANTS_UPDATE],
+  others: [], // All other events not included in the above categories
+};
+
+let amqpChannel: Channel | null = null;
 
 export const initAMQP = () => {
   return new Promise<void>((resolve, reject) => {
-    const uri = configService.get<Rabbitmq>('RABBITMQ').URI;
-    amqp.connect(uri, (error, connection) => {
+    const rabbitConfig = configService.get<Rabbitmq>('RABBITMQ');
+    connect(rabbitConfig.URI, (error, connection) => {
       if (error) {
         reject(error);
         return;
@@ -22,12 +40,9 @@ export const initAMQP = () => {
           return;
         }
 
-        const exchangeName = 'evolution_exchange';
-
-        channel.assertExchange(exchangeName, 'topic', {
+        channel.assertExchange(rabbitConfig.EXCHANGE_NAME, 'topic', {
           durable: true,
           autoDelete: false,
-          assert: true,
         });
 
         amqpChannel = channel;
@@ -39,65 +54,190 @@ export const initAMQP = () => {
   });
 };
 
-export const getAMQP = (): amqp.Channel | null => {
+export const getAMQP = (): Channel | null => {
   return amqpChannel;
 };
 
 export const initQueues = (instanceName: string, events: string[]) => {
   if (!instanceName || !events || !events.length) return;
+  const rabbitConfig = configService.get<Rabbitmq>('RABBITMQ');
+  const TWO_DAYS_IN_MS = 2 * 24 * 60 * 60 * 1000;
+  const amqp = getAMQP();
 
-  const queues = events.map((event) => {
-    return `${event.replace(/_/g, '.').toLowerCase()}`;
-  });
+  let exchangeName = rabbitConfig.EXCHANGE_NAME;
 
-  queues.forEach((event) => {
-    const amqp = getAMQP();
-    const exchangeName = instanceName ?? 'evolution_exchange';
+  const receivedEvents = events.map(parseEvtName);
+  if (rabbitConfig.MODE === 'isolated') {
+    exchangeName = instanceName;
 
+    receivedEvents.forEach((event) => {
+      amqp.assertExchange(exchangeName, 'topic', {
+        durable: true,
+        autoDelete: false,
+      });
+
+      const queueName = `${instanceName}.${event}`;
+      amqp.assertQueue(queueName, {
+        durable: true,
+        autoDelete: false,
+        messageTtl: TWO_DAYS_IN_MS,
+        arguments: {
+          'x-queue-type': 'quorum',
+        },
+      });
+
+      amqp.bindQueue(queueName, exchangeName, event);
+    });
+  } else if (rabbitConfig.MODE === 'single') {
     amqp.assertExchange(exchangeName, 'topic', {
       durable: true,
       autoDelete: false,
-      assert: true,
     });
 
-    const queueName = `${instanceName}.${event}`;
-
+    const queueName = 'evolution';
     amqp.assertQueue(queueName, {
       durable: true,
       autoDelete: false,
+      messageTtl: TWO_DAYS_IN_MS,
       arguments: {
         'x-queue-type': 'quorum',
       },
     });
 
-    amqp.bindQueue(queueName, exchangeName, event);
-  });
+    receivedEvents.forEach((event) => {
+      amqp.bindQueue(queueName, exchangeName, event);
+    });
+  } else if (rabbitConfig.MODE === 'global') {
+    const queues = Object.keys(globalQueues);
+
+    const addQueues = queues.filter((evt) => {
+      if (evt === 'others') {
+        return receivedEvents.some(
+          (e) =>
+            !Object.values(globalQueues)
+              .flat()
+              .includes(e as Events),
+        );
+      }
+      return globalQueues[evt].some((e) => receivedEvents.includes(e));
+    });
+
+    addQueues.forEach((event) => {
+      amqp.assertExchange(exchangeName, 'topic', {
+        durable: true,
+        autoDelete: false,
+      });
+
+      const queueName = event;
+      amqp.assertQueue(queueName, {
+        durable: true,
+        autoDelete: false,
+        messageTtl: TWO_DAYS_IN_MS,
+        arguments: {
+          'x-queue-type': 'quorum',
+        },
+      });
+
+      if (globalQueues[event].length === 0) {
+        // Other events
+        const otherEvents = Object.values(globalQueues).flat();
+        for (const subEvent in Events) {
+          const eventCode = Events[subEvent];
+          if (otherEvents.includes(eventCode)) continue;
+          if (!receivedEvents.includes(eventCode)) continue;
+          amqp.bindQueue(queueName, exchangeName, eventCode);
+        }
+      } else {
+        globalQueues[event].forEach((subEvent) => {
+          amqp.bindQueue(queueName, exchangeName, subEvent);
+        });
+      }
+    });
+  } else {
+    throw new Error('Invalid RabbitMQ mode');
+  }
 };
 
 export const removeQueues = (instanceName: string, events: string[]) => {
   if (!events || !events.length) return;
+  const rabbitConfig = configService.get<Rabbitmq>('RABBITMQ');
+  let exchangeName = rabbitConfig.EXCHANGE_NAME;
+  const amqp = getAMQP();
 
-  const channel = getAMQP();
+  const receivedEvents = events.map(parseEvtName);
+  if (rabbitConfig.MODE === 'isolated') {
+    exchangeName = instanceName;
+    receivedEvents.forEach((event) => {
+      amqp.assertExchange(exchangeName, 'topic', {
+        durable: true,
+        autoDelete: false,
+      });
 
-  const queues = events.map((event) => {
-    return `${event.replace(/_/g, '.').toLowerCase()}`;
-  });
-
-  const exchangeName = instanceName ?? 'evolution_exchange';
-
-  queues.forEach((event) => {
-    const amqp = getAMQP();
-
-    amqp.assertExchange(exchangeName, 'topic', {
-      durable: true,
-      autoDelete: false,
-      assert: true,
+      const queueName = `${instanceName}.${event}`;
+      amqp.deleteQueue(queueName);
     });
-
-    const queueName = `${instanceName}.${event}`;
-
-    amqp.deleteQueue(queueName);
-  });
-
-  channel.deleteExchange(exchangeName);
+    amqp.deleteExchange(instanceName);
+  }
+};
+
+interface SendEventData {
+  instanceName: string;
+  wuid: string;
+  event: string;
+  apiKey?: string;
+  data: any;
+}
+
+export const sendEventData = ({ data, event, wuid, apiKey, instanceName }: SendEventData) => {
+  const rabbitConfig = configService.get<Rabbitmq>('RABBITMQ');
+  let exchangeName = rabbitConfig.EXCHANGE_NAME;
+  if (rabbitConfig.MODE === 'isolated') exchangeName = instanceName;
+
+  amqpChannel.assertExchange(exchangeName, 'topic', {
+    durable: true,
+    autoDelete: false,
+  });
+  let queueName = event;
+  if (rabbitConfig.MODE === 'single') {
+    queueName = 'evolution';
+  } else if (rabbitConfig.MODE === 'global') {
+    let eventName = '';
+    Object.keys(globalQueues).forEach((key) => {
+      if (globalQueues[key].includes(event as Events)) {
+        eventName = key;
+      }
+      if (eventName === '' && key === 'others') {
+        eventName = key;
+      }
+    });
+    queueName = eventName;
+  }
+  amqpChannel.assertQueue(queueName, {
+    durable: true,
+    autoDelete: false,
+    arguments: { 'x-queue-type': 'quorum' },
+  });
+  amqpChannel.bindQueue(queueName, exchangeName, event);
+
+  const serverUrl = configService.get<HttpServer>('SERVER').URL;
+  const tzoffset = new Date().getTimezoneOffset() * 60000; //offset in milliseconds
+  const localISOTime = new Date(Date.now() - tzoffset).toISOString();
+  const now = localISOTime;
+  const message = {
+    event,
+    instance: instanceName,
+    data,
+    server_url: serverUrl,
+    date_time: now,
+    sender: wuid,
+  };
+  if (apiKey) {
+    message['apikey'] = apiKey;
+  }
+  logger.log({
+    queueName,
+    exchangeName,
+    event,
+  });
+  amqpChannel.publish(exchangeName, event, Buffer.from(JSON.stringify(message)));
 };

src/whatsapp/services/whatsapp.baileys.service.ts

@@ -1123,7 +1123,7 @@ export class BaileysStartupService extends WAStartupService {
         5: 'PLAYED',
       };
       for await (const { key, update } of args) {
-        if (settings?.groups_ignore && key.remoteJid.includes('@g.us')) {
+        if (settings?.groups_ignore && key.remoteJid?.includes('@g.us')) {
           this.logger.verbose('group ignored');
           return;
         }

src/whatsapp/services/whatsapp.business.service.ts

@@ -303,7 +303,6 @@ export class BusinessStartupService extends WAStartupService {
       if (received.contacts) pushName = received.contacts[0].profile.name;
 
       if (received.messages) {
-        console.log('received?.messages[0]', received?.messages[0]);
         const key = {
           id: received.messages[0].id,
           remoteJid: this.phoneNumber,
@@ -772,9 +771,6 @@ export class BusinessStartupService extends WAStartupService {
         }
       })();
 
-      console.log('messageSent', messageSent);
-      console.log('message', message);
-
       const messageRaw: MessageRaw = {
         key: { fromMe: true, id: messageSent?.messages[0]?.id, remoteJid: this.createJid(number) },
         //pushName: messageSent.pushName,

src/whatsapp/services/whatsapp.service.ts

@@ -20,7 +20,7 @@ import {
 import { Logger } from '../../config/logger.config';
 import { ROOT_DIR } from '../../config/path.config';
 import { NotFoundException } from '../../exceptions';
-import { getAMQP, removeQueues } from '../../libs/amqp.server';
+import { getAMQP, removeQueues, sendEventData } from '../../libs/amqp.server';
 import { getIO } from '../../libs/socket.server';
 import { getSQS, removeQueues as removeQueuesSQS } from '../../libs/sqs.server';
 import { ChamaaiRaw, IntegrationRaw, ProxyRaw, RabbitmqRaw, SettingsRaw, SqsRaw, TypebotRaw } from '../models';
@@ -685,40 +685,13 @@ export class WAStartupService {
 
       if (amqp) {
         if (Array.isArray(rabbitmqLocal) && rabbitmqLocal.includes(we)) {
-          const exchangeName = this.instanceName ?? 'evolution_exchange';
-
-          amqp.assertExchange(exchangeName, 'topic', {
-            durable: true,
-            autoDelete: false,
-            assert: true,
-          });
-
-          const queueName = `${this.instanceName}.${event}`;
-
-          amqp.assertQueue(queueName, {
-            durable: true,
-            autoDelete: false,
-            arguments: {
-              'x-queue-type': 'quorum',
-            },
-          });
-
-          amqp.bindQueue(queueName, exchangeName, event);
-
-          const message = {
-            event,
-            instance: this.instance.name,
+          sendEventData({
             data,
-            server_url: serverUrl,
-            date_time: now,
-            sender: this.wuid,
-          };
-
-          if (expose && instanceApikey) {
-            message['apikey'] = instanceApikey;
-          }
-
-          amqp.publish(exchangeName, event, Buffer.from(JSON.stringify(message)));
+            event,
+            instanceName: this.instanceName,
+            wuid: this.wuid,
+            apiKey: expose && instanceApikey ? instanceApikey : undefined,
+          });
 
           if (this.configService.get<Log>('LOG').LEVEL.includes('WEBHOOKS')) {
             const logData = {