feat(n8n): preservar eventos individuais antes do debounce

.github/workflows/check_code_quality.yml

@@ -26,7 +26,7 @@ jobs:
           --health-interval 5s --health-timeout 5s --health-retries 10
     env:
       DATABASE_PROVIDER: postgresql
-      DATABASE_CONNECTION_URI: postgresql://evolution_test:local_ci_only@127.0.0.1:5432/evolution_test
+      DATABASE_CONNECTION_URI: postgresql://evolution_test:local_ci_only@127.0.0.1:5432/evolution_test?connection_limit=64
       XMACNA_INBOX_INTEGRATION: '1'
     steps:
       - uses: actions/checkout@v5

README.md

@@ -69,6 +69,16 @@ Evolution API integrates natively with many platforms:
 
 ---
 
+## Xmacna n8n message events
+
+The n8n webhook includes `individualMessages`, captured before debounce. Each item preserves its provider `keyId`, `remoteJid`, `participant`, `fromMe`, message type and content. `messageTimestamp` is the original epoch in seconds; `providerTimestamp` is its safe, positive millisecond conversion. Missing or invalid values remain absent and must be rejected by a persistence consumer, never replaced with the current time.
+
+Each audio uses its own source for transcription. `transcriptionStatus: unavailable` preserves the original event when transcription fails; a persistence consumer must handle that state explicitly. Raw audio keys and base64 are excluded from this array. Existing aggregate fields and the live `ownerLid` field remain available. A batch with an incomplete event list fails before the webhook POST.
+
+This prepares individual event transport; it does not enable PostgreSQL writes or change the existing in-memory debounce crash window. Verification: `npm run test:xmacna-inbox` includes multi-message, audio and actual n8n service boundary cases.
+
+The PostgreSQL integration fixture uses an explicit `connection_limit=64` and runs test files sequentially. The 840 simultaneous claims within the stress case remain unchanged; this avoids a host-dependent default pool exhausting transaction admission while keeping the default transaction timeouts.
+
 ## Quick Start
 
 ### Prerequisites

package.json

@@ -14,8 +14,8 @@
     "dev:server": "tsx watch ./src/main.ts",
     "test": "tsx watch ./test/all.test.ts",
     "test:xmacna-compat": "node --test test/xmacna-baileys-compat.test.cjs",
-    "test:xmacna-inbox": "tsx --test test/xmacna-s3-storage.test.ts test/xmacna-inbound-inbox.test.ts test/xmacna-inbound-sink-dispatch.test.ts test/xmacna-inbound-service-contracts.test.ts test/xmacna-baileys-message-processor.test.ts test/xmacna-chatbot-debounce.test.ts test/xmacna-outgoing-lid.test.ts test/xmacna-inbound-identity.test.ts test/xmacna-inbound-media-upload.test.ts test/xmacna-inbound-media-guard.test.ts test/xmacna-get-conversation-message.test.ts test/xmacna-chatwoot-group-participant.test.ts",
-    "test:xmacna-inbox:postgres": "tsx --test test/xmacna-inbound-inbox.postgres.test.ts test/xmacna-inbound-identity.postgres.test.ts",
+    "test:xmacna-inbox": "tsx --test test/xmacna-s3-storage.test.ts test/xmacna-inbound-inbox.test.ts test/xmacna-inbound-sink-dispatch.test.ts test/xmacna-inbound-service-contracts.test.ts test/xmacna-baileys-message-processor.test.ts test/xmacna-chatbot-debounce.test.ts test/xmacna-outgoing-lid.test.ts test/xmacna-inbound-identity.test.ts test/xmacna-inbound-media-upload.test.ts test/xmacna-inbound-media-guard.test.ts test/xmacna-get-conversation-message.test.ts test/xmacna-chatwoot-group-participant.test.ts test/xmacna-n8n-individual-messages.test.ts test/xmacna-n8n-individual-service.test.ts",
+    "test:xmacna-inbox:postgres": "tsx --test --test-concurrency=1 test/xmacna-inbound-inbox.postgres.test.ts test/xmacna-inbound-identity.postgres.test.ts",
     "inbound-inbox:maintenance": "tsx src/cli/inboundInboxMaintenance.ts",
     "lint": "eslint --fix --ext .ts src",
     "lint:check": "eslint --ext .ts src",
@@ -160,6 +160,7 @@
     "@typescript-eslint/parser": "^8.44.0",
     "commitizen": "^4.3.1",
     "cz-conventional-changelog": "^3.3.0",
+    "esbuild": "0.27.7",
     "eslint": "^8.45.0",
     "eslint-config-prettier": "^10.1.8",
     "eslint-plugin-import": "^2.31.0",

src/api/integrations/chatbot/base-chatbot.controller.ts

@@ -17,6 +17,7 @@ import {
   ChatbotDebounceStore,
   normalizeMessageTimestamp,
 } from './chatbotDebounce';
+import { captureIndividualChatbotMessage } from './n8n/n8nIndividualMessages';
 
 /**
  * XMACNA_DEBOUNCE_DETACHED_518: os metadados do buffer viajam numa copia rasa da
@@ -971,6 +972,7 @@ export abstract class BaseChatbotController<BotType = any, BotData extends BaseC
             );
           },
           normalizeMessageTimestamp(msg?.messageTimestamp),
+          this.integrationName === 'N8n' ? captureIndividualChatbotMessage(msg, content) : undefined,
         );
       } else {
         await this.processBot(

src/api/integrations/chatbot/chatbot.controller.ts

@@ -21,6 +21,7 @@ import {
   processChatbotDebounce,
 } from './chatbotDebounce';
 import { runBestEffortChatbots } from './chatbotDispatchPolicy';
+import type { IndividualChatbotMessage } from './n8n/n8nIndividualMessages';
 
 export type EmitData = {
   instance: InstanceDto;
@@ -143,9 +144,11 @@ export class ChatbotController {
     debounceTime: number,
     callback: ChatbotDebounceCallback,
     messageTimestamp?: unknown,
+    individualMessage?: IndividualChatbotMessage,
   ): ChatbotDebounceAcceptance {
     return processChatbotDebounce(userMessageDebounce, content, debounceKey, debounceTime, callback, {
       messageTimestamp,
+      individualMessage,
       onMerged: (merged) => this.logger.log('message debounced: ' + merged),
       onFlushed: (flushed) => this.logger.log('Debounce complete. Processing message: ' + flushed),
       onFlushError: (error) =>

src/api/integrations/chatbot/chatbotDebounce.ts

@@ -25,13 +25,17 @@
  * nota, nao como padrao.
  */
 
+import type { IndividualChatbotMessage } from './n8n/n8nIndividualMessages';
+
 export type ChatbotDebounceMetadata = {
   /** Quantas mensagens do lead foram coalescidas neste buffer. */
   messageCount: number;
   /** `messageTimestamp` (epoch em segundos) da primeira mensagem do buffer. */
   firstMessageTimestamp?: number;
   /** `messageTimestamp` (epoch em segundos) da ultima mensagem do buffer. */
   lastMessageTimestamp?: number;
+  /** Provider events captured before coalescing; n8n only. */
+  individualMessages?: IndividualChatbotMessage[];
 };
 
 export type ChatbotDebounceCallback = (content: string, metadata: ChatbotDebounceMetadata) => Promise<void>;
@@ -51,6 +55,7 @@ export type ChatbotDebounceStore = Record<string, ChatbotDebounceEntry>;
 export type ChatbotDebounceOptions = {
   /** `messageTimestamp` normalizado da mensagem que esta sendo aceita. */
   messageTimestamp?: unknown;
+  individualMessage?: IndividualChatbotMessage;
   onMerged?: (content: string) => void;
   onFlushed?: (content: string) => void;
   /** Unico lugar onde uma falha do flush destacado fica visivel. */
@@ -112,6 +117,9 @@ export function processChatbotDebounce(
     existing.message += `\n${content}`;
     existing.callback = callback;
     existing.metadata.messageCount += 1;
+    if (options.individualMessage) {
+      (existing.metadata.individualMessages ??= []).push(structuredClone(options.individualMessage));
+    }
     if (timestamp !== undefined) {
       if (existing.metadata.firstMessageTimestamp === undefined) {
         existing.metadata.firstMessageTimestamp = timestamp;
@@ -139,6 +147,7 @@ export function processChatbotDebounce(
         messageCount: 1,
         firstMessageTimestamp: timestamp,
         lastMessageTimestamp: timestamp,
+        ...(options.individualMessage ? { individualMessages: [structuredClone(options.individualMessage)] } : {}),
       },
       flushed,
       resolveFlushed,
@@ -154,7 +163,7 @@ export function processChatbotDebounce(
     if (store[debounceKey] !== scheduled) return;
     delete store[debounceKey];
     options.onFlushed?.(scheduled.message);
-    const snapshot: ChatbotDebounceMetadata = { ...scheduled.metadata };
+    const snapshot = structuredClone(scheduled.metadata);
     void Promise.resolve()
       .then(() => scheduled.callback(scheduled.message, snapshot))
       .then(
@@ -166,5 +175,5 @@ export function processChatbotDebounce(
       );
   }, debounceTime * 1000);
 
-  return { merged, metadata: { ...scheduled.metadata }, flushed: scheduled.flushed };
+  return { merged, metadata: structuredClone(scheduled.metadata), flushed: scheduled.flushed };
 }

src/api/integrations/chatbot/n8n/n8nIndividualMessages.ts

@@ -0,0 +1,96 @@
+import { normalizeMessageTimestamp } from '../chatbotDebounce';
+
+/** Internal envelope: audio source is retained only in memory, never sent to n8n. */
+export type IndividualChatbotMessage = {
+  keyId?: string;
+  remoteJid?: string;
+  participant?: string;
+  fromMe?: boolean;
+  messageTimestamp?: number;
+  messageType: string;
+  chatInput: string;
+  transcription?: string;
+  audioSource?: { key: Record<string, unknown>; message: unknown };
+};
+
+export type N8nIndividualMessage = Omit<IndividualChatbotMessage, 'audioSource' | 'transcription'> & {
+  providerTimestamp?: number;
+  transcriptionStatus?: 'available' | 'unavailable';
+};
+
+export function captureIndividualChatbotMessage(msg: any, content: string): IndividualChatbotMessage {
+  const message = msg?.message;
+  const body = message?.ephemeralMessage?.message ?? message;
+  const inner = body?.viewOnceMessageV2?.message ?? body?.viewOnceMessage?.message ?? body;
+  const type = inner?.audioMessage
+    ? 'audio'
+    : inner?.imageMessage
+      ? 'image'
+      : inner?.videoMessage
+        ? 'video'
+        : inner?.documentMessage || inner?.documentWithCaptionMessage
+          ? 'document'
+          : inner?.contactMessage || inner?.contactsArrayMessage
+            ? 'contact'
+            : inner?.locationMessage || inner?.liveLocationMessage
+              ? 'location'
+              : 'text';
+  const key = {
+    id: msg?.key?.id,
+    remoteJid: msg?.key?.remoteJid,
+    participant: msg?.key?.participant,
+    fromMe: msg?.key?.fromMe,
+  };
+  const transcription =
+    typeof message?.speechToText === 'string' && message.speechToText.trim() ? message.speechToText : undefined;
+  return {
+    keyId: key.id,
+    remoteJid: key.remoteJid,
+    participant: key.participant,
+    fromMe: key.fromMe,
+    messageTimestamp: normalizeMessageTimestamp(msg?.messageTimestamp),
+    messageType: type,
+    chatInput: content,
+    ...(type === 'audio' && transcription ? { transcription } : {}),
+    ...(type === 'audio' && !transcription ? { audioSource: structuredClone({ key, message }) } : {}),
+  };
+}
+
+/** Explicit projection prevents raw media keys/base64 from leaking into the webhook. */
+export async function prepareN8nIndividualMessages(
+  messages: IndividualChatbotMessage[],
+  transcribe: (source: IndividualChatbotMessage['audioSource']) => Promise<string | null>,
+): Promise<N8nIndividualMessage[]> {
+  const result: N8nIndividualMessage[] = [];
+  for (const item of messages) {
+    let text = item.chatInput;
+    let transcriptionStatus: 'available' | 'unavailable' | undefined;
+    if (item.messageType === 'audio') {
+      transcriptionStatus = item.transcription ? 'available' : 'unavailable';
+      try {
+        if (!item.transcription) {
+          const transcription = item.audioSource ? await transcribe(item.audioSource) : null;
+          if (typeof transcription === 'string' && transcription.trim()) {
+            text = `[audio] ${transcription}`;
+            transcriptionStatus = 'available';
+          }
+        }
+      } catch {
+        // Preserve this event and its original content; never borrow another audio's text.
+      }
+    }
+    const milliseconds = item.messageTimestamp === undefined ? undefined : item.messageTimestamp * 1000;
+    result.push({
+      keyId: item.keyId,
+      remoteJid: item.remoteJid,
+      participant: item.participant,
+      fromMe: item.fromMe,
+      messageTimestamp: item.messageTimestamp,
+      providerTimestamp: Number.isSafeInteger(milliseconds) && milliseconds > 0 ? milliseconds : undefined,
+      messageType: item.messageType,
+      chatInput: text,
+      ...(transcriptionStatus ? { transcriptionStatus } : {}),
+    });
+  }
+  return result;
+}

src/api/integrations/chatbot/n8n/services/n8n.service.ts

@@ -7,6 +7,7 @@ import axios from 'axios';
 import { BaseChatbotService } from '../../base-chatbot.service';
 import { ChatbotDebounceMetadata, normalizeMessageTimestamp } from '../../chatbotDebounce';
 import { OpenaiService } from '../../openai/services/openai.service';
+import { captureIndividualChatbotMessage, prepareN8nIndividualMessages } from '../n8nIndividualMessages';
 
 export class N8nService extends BaseChatbotService<N8n, N8nSetting> {
   private openaiService: OpenaiService;
@@ -56,6 +57,18 @@ export class N8nService extends BaseChatbotService<N8n, N8nSetting> {
       // adivinhar. Sem debounce, os tres campos descrevem uma mensagem so.
       const debounceMetadata = msg?.debounceMetadata as ChatbotDebounceMetadata | undefined;
       const lastMessageTimestamp = normalizeMessageTimestamp(msg?.messageTimestamp);
+      const captured = debounceMetadata?.individualMessages ?? [captureIndividualChatbotMessage(msg, content)];
+      if (captured.length !== (debounceMetadata?.messageCount ?? 1)) {
+        throw new Error('Incomplete individual n8n events; refusing to represent a batch as one event');
+      }
+      const individualMessages = await prepareN8nIndividualMessages(captured, (source) =>
+        this.openaiService.speechToText(source, instance),
+      );
+      const missingTranscriptions = individualMessages.filter(
+        (item) => item.transcriptionStatus === 'unavailable',
+      ).length;
+      if (missingTranscriptions)
+        this.logger.warn(`[N8n] Audio transcription unavailable for ${missingTranscriptions} individual event(s)`);
       const payload: any = {
         chatInput: content,
         sessionId: session.sessionId,
@@ -70,19 +83,13 @@ export class N8nService extends BaseChatbotService<N8n, N8nSetting> {
         instanceName: instance.instanceName,
         serverUrl: this.configService.get<HttpServer>('SERVER').URL,
         apiKey: instance.token,
+        ownerLid: instance.client?.user?.lid ? String(instance.client.user.lid).replace(/[@:].*$/, '') : undefined,
+        individualMessages,
       };
 
       // Handle audio messages
-      if (this.isAudioMessage(content) && msg) {
-        try {
-          this.logger.debug(`[N8n] Downloading audio for Whisper transcription`);
-          const transcription = await this.openaiService.speechToText(msg, instance);
-          if (transcription) {
-            payload.chatInput = `[audio] ${transcription}`;
-          }
-        } catch (err) {
-          this.logger.error(`[N8n] Failed to transcribe audio: ${err}`);
-        }
+      if (captured.some((item) => item.messageType === 'audio')) {
+        payload.chatInput = individualMessages.map((item) => item.chatInput).join('\n');
       }
 
       const headers: Record<string, string> = {};

test/xmacna-n8n-individual-messages.test.ts

@@ -0,0 +1,89 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+
+import { processChatbotDebounce, ChatbotDebounceStore } from '../src/api/integrations/chatbot/chatbotDebounce';
+import {
+  captureIndividualChatbotMessage,
+  prepareN8nIndividualMessages,
+} from '../src/api/integrations/chatbot/n8n/n8nIndividualMessages';
+
+const event = (id: string, time: unknown, message: any, fromMe = false) => ({
+  key: { id, remoteJid: 'lead@lid', participant: 'sender@lid', fromMe }, messageTimestamp: time, message,
+});
+
+test('debounce delivers every provider ID, content and timestamp in arrival order, with isolated snapshots', async () => {
+  const store: ChatbotDebounceStore = {};
+  let payload: any;
+  const deliver = async (_: string, metadata: any) => {
+    payload = await prepareN8nIndividualMessages(metadata.individualMessages, async () => { throw Error('not audio'); });
+  };
+  const firstEvent = captureIndividualChatbotMessage(event('first', '1792788000', { conversation: 'one' }), 'one');
+  const first = processChatbotDebounce(store, 'one', 'key', 0.01, deliver, {
+    messageTimestamp: firstEvent.messageTimestamp, individualMessage: firstEvent,
+  });
+  firstEvent.chatInput = 'caller mutation';
+  first.metadata.individualMessages![0].chatInput = 'receipt mutation';
+  const secondEvent = captureIndividualChatbotMessage(event('second', { toNumber: () => 1792788007 }, { conversation: 'two' }), 'two');
+  const second = processChatbotDebounce(store, 'two', 'key', 0.01, deliver, {
+    messageTimestamp: secondEvent.messageTimestamp, individualMessage: secondEvent,
+  });
+  assert.equal(first.metadata.messageCount, 1);
+  assert.equal(first.metadata.individualMessages!.length, 1);
+  await second.flushed;
+  assert.deepEqual(payload.map((m: any) => [m.keyId, m.chatInput, m.providerTimestamp, m.fromMe, m.remoteJid, m.participant]), [
+    ['first', 'one', 1792788000000, false, 'lead@lid', 'sender@lid'],
+    ['second', 'two', 1792788007000, false, 'lead@lid', 'sender@lid'],
+  ]);
+});
+
+test('two audios retain their own source, transcription and time around a text, without raw media keys in the payload', async () => {
+  const first = event('audio-a', 1792788000, { audioMessage: { mediaKey: new Uint8Array([1, 2]) }, mediaUrl: 'https://local.invalid/a.ogg' });
+  const second = event('audio-b', 1792788003, { audioMessage: {}, base64: 'private-audio-base64' });
+  const captured = [captureIndividualChatbotMessage(first, 'audioMessage|a'),
+    captureIndividualChatbotMessage(event('text', 1792788002, { conversation: 'middle' }), 'middle'),
+    captureIndividualChatbotMessage(second, 'audioMessage|b')];
+  first.message.mediaUrl = 'caller mutation';
+  const ids: unknown[] = [];
+  const result = await prepareN8nIndividualMessages(captured, async (source) => {
+    ids.push(source!.key.id);
+    if (source!.key.id === 'audio-a') assert.equal((source!.message as any).mediaUrl, 'https://local.invalid/a.ogg');
+    return source!.key.id === 'audio-a' ? 'first audio' : 'second audio';
+  });
+  assert.deepEqual(ids, ['audio-a', 'audio-b']);
+  assert.deepEqual(result.map(m => [m.keyId, m.chatInput, m.transcriptionStatus]), [
+    ['audio-a', '[audio] first audio', 'available'], ['text', 'middle', undefined],
+    ['audio-b', '[audio] second audio', 'available'],
+  ]);
+  assert.doesNotMatch(JSON.stringify(result), /mediaKey|audioSource|private-audio-base64/);
+});
+
+test('audio failure retains its own event and marks it unavailable without copying the next transcription', async () => {
+  const result = await prepareN8nIndividualMessages(['bad', 'good'].map((id, n) =>
+    captureIndividualChatbotMessage(event(id, 1792788000 + n, { audioMessage: {} }), `audioMessage|${id}`)),
+  async source => { if (source!.key.id === 'bad') throw Error('Whisper unavailable'); return 'actual text'; });
+  assert.equal(result[0].keyId, 'bad');
+  assert.equal(result[0].chatInput, 'audioMessage|bad');
+  assert.equal(result[0].transcriptionStatus, 'unavailable');
+  assert.equal(result[1].chatInput, '[audio] actual text');
+});
+
+test('already transcribed audio does not request transcription again', async () => {
+  const result = await prepareN8nIndividualMessages([
+    captureIndividualChatbotMessage(event('audio', 1792788000, { audioMessage: {}, speechToText: 'existing text' }), 'existing text'),
+  ], async () => { throw Error('must not call'); });
+  assert.equal(result[0].chatInput, 'existing text');
+  assert.equal(result[0].transcriptionStatus, 'available');
+});
+
+test('missing, invalid or unsafe provider timestamps never become the current time', async () => {
+  for (const time of [undefined, 'bad', Infinity, -1, 0, Number.MAX_SAFE_INTEGER]) {
+    const result = await prepareN8nIndividualMessages([
+      captureIndividualChatbotMessage(event('id', time, { conversation: 'text' }, true), 'text'),
+    ], async () => null);
+    assert.equal(result[0].providerTimestamp, undefined);
+    assert.equal(result[0].fromMe, true);
+  }
+  const noIdentity = await prepareN8nIndividualMessages([captureIndividualChatbotMessage({}, 'text')], async () => null);
+  assert.equal(noIdentity[0].keyId, undefined);
+  assert.equal(noIdentity[0].providerTimestamp, undefined);
+});