Merge pull request #2321 from Vitordotpy/fix/remotejid-normalization-and-cache-race

fix: normalize remoteJid in message updates and handle race condition in contact cache

Author: DavidsonGomes

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

@@ -1565,8 +1565,15 @@ export class BaileysStartupService extends ChannelStartupService {
 
       for await (const { key, update } of args) {
         const keyAny = key as any;
-        const normalizedRemoteJid = keyAny.remoteJid?.replace(/:.*$/, '');
-        const normalizedParticipant = keyAny.participant?.replace(/:.*$/, '');
+        if (keyAny.remoteJid) {
+          keyAny.remoteJid = keyAny.remoteJid.replace(/:.*$/, '');
+        }
+        if (keyAny.participant) {
+          keyAny.participant = keyAny.participant.replace(/:.*$/, '');
+        }
+
+        const normalizedRemoteJid = keyAny.remoteJid;
+        const normalizedParticipant = keyAny.participant;
 
         if (settings?.groupsIgnore && normalizedRemoteJid?.includes('@g.us')) {
           continue;
@@ -1644,18 +1651,48 @@ export class BaileysStartupService extends ChannelStartupService {
 
             const searchId = originalMessageId || key.id;
 
-            const messages = (await this.prismaRepository.$queryRaw`
-              SELECT * FROM "Message"
-              WHERE "instanceId" = ${this.instanceId}
-              AND "key"->>'id' = ${searchId}
-              LIMIT 1
-            `) as any[];
-            findMessage = messages[0] || null;
+            let retries = 0;
+            const maxRetries = 3;
+            const retryDelay = 500; // 500ms delay to avoid blocking for too long
+
+            while (retries < maxRetries) {
+              const messages = (await this.prismaRepository.$queryRaw`
+                SELECT * FROM "Message"
+                WHERE "instanceId" = ${this.instanceId}
+                AND "key"->>'id' = ${searchId}
+                LIMIT 1
+              `) as any[];
+              findMessage = messages[0] || null;
+
+              if (findMessage?.id) {
+                break;
+              }
+
+              retries++;
+              if (retries < maxRetries) {
+                await delay(retryDelay);
+              }
+            }
 
             if (!findMessage?.id) {
-              this.logger.warn(`Original message not found for update. Skipping. Key: ${JSON.stringify(key)}`);
+              this.logger.verbose(
+                `Original message not found for update after ${maxRetries} retries. Skipping. This is expected for protocol messages or ephemeral events not saved to the database. Key: ${JSON.stringify(key)}`,
+              );
               continue;
             }
+
+            // Sync the incoming key.remoteJid with the stored one.
+            // This mutation is safe and necessary because Baileys events might use LIDs while we store Phone JIDs (or vice versa).
+            // Normalizing ensuring downstream logic uses the identifier that exists in our database.
+            if (findMessage?.key?.remoteJid && key.remoteJid !== findMessage.key.remoteJid) {
+              key.remoteJid = findMessage.key.remoteJid;
+            }
+            if (findMessage?.key?.remoteJid && findMessage.key.remoteJid !== key.remoteJid) {
+              this.logger.verbose(
+                `Updating key.remoteJid from ${key.remoteJid} to ${findMessage.key.remoteJid} based on stored message`,
+              );
+              key.remoteJid = findMessage.key.remoteJid;
+            }
             message.messageId = findMessage.id;
           }
 

src/utils/onWhatsappCache.ts

@@ -1,6 +1,7 @@
 import { prismaRepository } from '@api/server.module';
 import { configService, Database } from '@config/env.config';
 import { Logger } from '@config/logger.config';
+import { Prisma } from '@prisma/client';
 import dayjs from 'dayjs';
 
 const logger = new Logger('OnWhatsappCache');
@@ -164,9 +165,28 @@ export async function saveOnWhatsappCache(data: ISaveOnWhatsappCacheParams[]) {
         logger.verbose(
           `[saveOnWhatsappCache] Register does not exist, creating: remoteJid=${remoteJid}, jidOptions=${dataPayload.jidOptions}, lid=${dataPayload.lid}`,
         );
-        await prismaRepository.isOnWhatsapp.create({
-          data: dataPayload,
-        });
+        try {
+          await prismaRepository.isOnWhatsapp.create({
+            data: dataPayload,
+          });
+        } catch (error: any) {
+          // Check for unique constraint violation (Prisma error code P2002)
+          if (
+            error instanceof Prisma.PrismaClientKnownRequestError &&
+            error.code === 'P2002' &&
+            (error.meta?.target as string[])?.includes('remoteJid')
+          ) {
+            logger.verbose(
+              `[saveOnWhatsappCache] Race condition detected for ${remoteJid}, updating existing record instead.`,
+            );
+            await prismaRepository.isOnWhatsapp.update({
+              where: { remoteJid: remoteJid },
+              data: dataPayload,
+            });
+          } else {
+            throw error;
+          }
+        }
       }
     } catch (e) {
       // Loga o erro mas não para a execução dos outros promises