feat(history-sync): emit messaging-history.set event on sync completion and fix race condition

Author: alexandrereyes

src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts

@@ -993,12 +993,12 @@ export class BaileysStartupService extends ChannelStartupService {
           chatsRaw.push({ remoteJid: chat.id, instanceId: this.instanceId, name: chat.name });
         }
 
-        this.sendDataWebhook(Events.CHATS_SET, chatsRaw);
-
         if (this.configService.get<Database>('DATABASE').SAVE_DATA.HISTORIC) {
           await this.prismaRepository.chat.createMany({ data: chatsRaw, skipDuplicates: true });
         }
 
+        this.sendDataWebhook(Events.CHATS_SET, chatsRaw);
+
         const messagesRaw: any[] = [];
 
         const messagesRepository: Set<string> = new Set(
@@ -1050,15 +1050,15 @@ export class BaileysStartupService extends ChannelStartupService {
           messagesRaw.push(this.prepareMessage(m));
         }
 
+        if (this.configService.get<Database>('DATABASE').SAVE_DATA.HISTORIC) {
+          await this.prismaRepository.message.createMany({ data: messagesRaw, skipDuplicates: true });
+        }
+
         this.sendDataWebhook(Events.MESSAGES_SET, [...messagesRaw], true, undefined, {
           isLatest,
           progress,
         });
 
-        if (this.configService.get<Database>('DATABASE').SAVE_DATA.HISTORIC) {
-          await this.prismaRepository.message.createMany({ data: messagesRaw, skipDuplicates: true });
-        }
-
         if (
           this.configService.get<Chatwoot>('CHATWOOT').ENABLED &&
           this.localChatwoot?.enabled &&
@@ -1075,6 +1075,14 @@ export class BaileysStartupService extends ChannelStartupService {
           contacts.filter((c) => !!c.notify || !!c.name).map((c) => ({ id: c.id, name: c.name ?? c.notify })),
         );
 
+        if (progress === 100) {
+          this.sendDataWebhook(Events.MESSAGING_HISTORY_SET, {
+            messageCount: messagesRaw.length,
+            chatCount: chatsRaw.length,
+            contactCount: contacts?.length ?? 0,
+          });
+        }
+
         contacts = undefined;
         messages = undefined;
         chats = undefined;

src/api/integrations/event/event.controller.ts

@@ -162,6 +162,7 @@ export class EventController {
     'CALL',
     'TYPEBOT_START',
     'TYPEBOT_CHANGE_STATUS',
+    'MESSAGING_HISTORY_SET',
     'REMOVE_INSTANCE',
     'LOGOUT_INSTANCE',
     'INSTANCE_CREATE',

src/config/env.config.ts

@@ -91,6 +91,7 @@ export type EventsRabbitmq = {
   CALL: boolean;
   TYPEBOT_START: boolean;
   TYPEBOT_CHANGE_STATUS: boolean;
+  MESSAGING_HISTORY_SET: boolean;
 };
 
 export type Rabbitmq = {
@@ -150,6 +151,7 @@ export type Sqs = {
     SEND_MESSAGE: boolean;
     TYPEBOT_CHANGE_STATUS: boolean;
     TYPEBOT_START: boolean;
+    MESSAGING_HISTORY_SET: boolean;
   };
 };
 
@@ -223,6 +225,7 @@ export type EventsWebhook = {
   CALL: boolean;
   TYPEBOT_START: boolean;
   TYPEBOT_CHANGE_STATUS: boolean;
+  MESSAGING_HISTORY_SET: boolean;
   ERRORS: boolean;
   ERRORS_WEBHOOK: string;
 };
@@ -256,6 +259,7 @@ export type EventsPusher = {
   CALL: boolean;
   TYPEBOT_START: boolean;
   TYPEBOT_CHANGE_STATUS: boolean;
+  MESSAGING_HISTORY_SET: boolean;
 };
 
 export type ApiKey = { KEY: string };
@@ -537,6 +541,7 @@ export class ConfigService {
           CALL: process.env?.RABBITMQ_EVENTS_CALL === 'true',
           TYPEBOT_START: process.env?.RABBITMQ_EVENTS_TYPEBOT_START === 'true',
           TYPEBOT_CHANGE_STATUS: process.env?.RABBITMQ_EVENTS_TYPEBOT_CHANGE_STATUS === 'true',
+          MESSAGING_HISTORY_SET: process.env?.RABBITMQ_EVENTS_MESSAGING_HISTORY_SET === 'true',
         },
       },
       NATS: {
@@ -574,6 +579,7 @@ export class ConfigService {
           CALL: process.env?.NATS_EVENTS_CALL === 'true',
           TYPEBOT_START: process.env?.NATS_EVENTS_TYPEBOT_START === 'true',
           TYPEBOT_CHANGE_STATUS: process.env?.NATS_EVENTS_TYPEBOT_CHANGE_STATUS === 'true',
+          MESSAGING_HISTORY_SET: process.env?.NATS_EVENTS_MESSAGING_HISTORY_SET === 'true',
         },
       },
       SQS: {
@@ -614,6 +620,7 @@ export class ConfigService {
           SEND_MESSAGE: process.env?.SQS_GLOBAL_SEND_MESSAGE === 'true',
           TYPEBOT_CHANGE_STATUS: process.env?.SQS_GLOBAL_TYPEBOT_CHANGE_STATUS === 'true',
           TYPEBOT_START: process.env?.SQS_GLOBAL_TYPEBOT_START === 'true',
+          MESSAGING_HISTORY_SET: process.env?.SQS_GLOBAL_MESSAGING_HISTORY_SET === 'true',
         },
       },
       KAFKA: {
@@ -657,6 +664,7 @@ export class ConfigService {
           CALL: process.env?.KAFKA_EVENTS_CALL === 'true',
           TYPEBOT_START: process.env?.KAFKA_EVENTS_TYPEBOT_START === 'true',
           TYPEBOT_CHANGE_STATUS: process.env?.KAFKA_EVENTS_TYPEBOT_CHANGE_STATUS === 'true',
+          MESSAGING_HISTORY_SET: process.env?.KAFKA_EVENTS_MESSAGING_HISTORY_SET === 'true',
         },
         SASL:
           process.env?.KAFKA_SASL_ENABLED === 'true'
@@ -722,6 +730,7 @@ export class ConfigService {
           CALL: process.env?.PUSHER_EVENTS_CALL === 'true',
           TYPEBOT_START: process.env?.PUSHER_EVENTS_TYPEBOT_START === 'true',
           TYPEBOT_CHANGE_STATUS: process.env?.PUSHER_EVENTS_TYPEBOT_CHANGE_STATUS === 'true',
+          MESSAGING_HISTORY_SET: process.env?.PUSHER_EVENTS_MESSAGING_HISTORY_SET === 'true',
         },
       },
       WA_BUSINESS: {
@@ -779,6 +788,7 @@ export class ConfigService {
           CALL: process.env?.WEBHOOK_EVENTS_CALL === 'true',
           TYPEBOT_START: process.env?.WEBHOOK_EVENTS_TYPEBOT_START === 'true',
           TYPEBOT_CHANGE_STATUS: process.env?.WEBHOOK_EVENTS_TYPEBOT_CHANGE_STATUS === 'true',
+          MESSAGING_HISTORY_SET: process.env?.WEBHOOK_EVENTS_MESSAGING_HISTORY_SET === 'true',
           ERRORS: process.env?.WEBHOOK_EVENTS_ERRORS === 'true',
           ERRORS_WEBHOOK: process.env?.WEBHOOK_EVENTS_ERRORS_WEBHOOK || '',
         },

src/validate/instance.schema.ts

@@ -86,6 +86,7 @@ export const instanceSchema: JSONSchema7 = {
           'CALL',
           'TYPEBOT_START',
           'TYPEBOT_CHANGE_STATUS',
+          'MESSAGING_HISTORY_SET',
         ],
       },
     },
@@ -123,6 +124,7 @@ export const instanceSchema: JSONSchema7 = {
           'CALL',
           'TYPEBOT_START',
           'TYPEBOT_CHANGE_STATUS',
+          'MESSAGING_HISTORY_SET',
         ],
       },
     },
@@ -160,6 +162,7 @@ export const instanceSchema: JSONSchema7 = {
           'CALL',
           'TYPEBOT_START',
           'TYPEBOT_CHANGE_STATUS',
+          'MESSAGING_HISTORY_SET',
         ],
       },
     },
@@ -197,6 +200,7 @@ export const instanceSchema: JSONSchema7 = {
           'CALL',
           'TYPEBOT_START',
           'TYPEBOT_CHANGE_STATUS',
+          'MESSAGING_HISTORY_SET',
         ],
       },
     },