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

Author: Vitordotpy

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,38 @@ 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;
+
+            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(2000);
+              }
+            }
 
             if (!findMessage?.id) {
               this.logger.warn(`Original message not found for update. Skipping. Key: ${JSON.stringify(key)}`);
               continue;
             }
+            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

@@ -164,9 +164,24 @@ 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.code === 'P2002' && error.meta?.target?.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