diff --git a/scripts/whatsapp-bridge/bridge.js b/scripts/whatsapp-bridge/bridge.js index 4c65740c0..736715d69 100644 --- a/scripts/whatsapp-bridge/bridge.js +++ b/scripts/whatsapp-bridge/bridge.js @@ -83,6 +83,19 @@ const CHUNK_DELAY_MS = parseInt(process.env.WHATSAPP_CHUNK_DELAY_MS || '300', 10 // fires. Fail fast instead so the gateway can surface a real error and retry. const SEND_TIMEOUT_MS = parseInt(process.env.WHATSAPP_SEND_TIMEOUT_MS || '60000', 10); +// --- Send queue: serialise all sock.sendMessage() calls across concurrent +// HTTP handlers so a single Baileys socket never has overlapping sends. +// Overlapping sends are the root cause of cross-chat contamination +// (#33360) — the WhatsApp protocol-level routing can misdeliver when +// two sendMessage() Promises race on the same socket. --- +let _sendQueue = Promise.resolve(); + +function enqueueSend(fn) { + const task = _sendQueue.then(() => fn(), () => fn()); + _sendQueue = task.catch(() => {}); + return task; +} + function sleep(ms) { return new Promise(resolve => setTimeout(resolve, ms)); } @@ -95,8 +108,10 @@ function sendWithTimeout(chatId, payload, timeoutMs = SEND_TIMEOUT_MS) { timeoutMs, ); }); - return Promise.race([sock.sendMessage(chatId, payload), timeoutPromise]) - .finally(() => clearTimeout(timer)); + return enqueueSend(() => + Promise.race([sock.sendMessage(chatId, payload), timeoutPromise]) + .finally(() => clearTimeout(timer)) + ); } function formatOutgoingMessage(message) { diff --git a/scripts/whatsapp-bridge/bridge.sendqueue.integration.test.mjs b/scripts/whatsapp-bridge/bridge.sendqueue.integration.test.mjs new file mode 100644 index 000000000..ad4d2b890 --- /dev/null +++ b/scripts/whatsapp-bridge/bridge.sendqueue.integration.test.mjs @@ -0,0 +1,29 @@ +/** + * Integration test: verify bridge.js has the send queue wired in. + * This reads the source file and asserts the required patterns exist. + */ +import { strict as assert } from 'node:assert'; +import { readFileSync } from 'node:fs'; +import { resolve, dirname } from 'node:path'; +import { fileURLToPath } from 'node:url'; + +const __dirname = dirname(fileURLToPath(import.meta.url)); +const src = readFileSync(resolve(__dirname, 'bridge.js'), 'utf-8'); + +// 1. Queue primitives exist +assert.ok(src.includes('let _sendQueue = Promise.resolve()'), + 'bridge.js must define _sendQueue'); +assert.ok(src.includes('function enqueueSend(fn)'), + 'bridge.js must define enqueueSend'); + +// 2. sendWithTimeout is wrapped +assert.ok(src.includes('return enqueueSend(() =>'), + 'sendWithTimeout must call enqueueSend'); + +// 3. The queue is before sendWithTimeout in the file +const queueIdx = src.indexOf('let _sendQueue'); +const sendTimeoutIdx = src.indexOf('function sendWithTimeout'); +assert.ok(queueIdx < sendTimeoutIdx, + '_sendQueue must be defined before sendWithTimeout'); + +console.log('✅ bridge.js send queue integration check passed.'); diff --git a/scripts/whatsapp-bridge/bridge.sendqueue.test.mjs b/scripts/whatsapp-bridge/bridge.sendqueue.test.mjs new file mode 100644 index 000000000..4504a745a --- /dev/null +++ b/scripts/whatsapp-bridge/bridge.sendqueue.test.mjs @@ -0,0 +1,112 @@ +/** + * Regression tests for the WhatsApp bridge send queue (#33360). + * + * The bridge must serialise all sock.sendMessage() calls through a + * promise-based queue so that concurrent HTTP /send requests never + * produce overlapping Baileys socket writes. Overlapping writes are + * the confirmed root cause of cross-chat contamination. + * + * These tests exercise the queue itself — they do NOT require a live + * WhatsApp socket. + */ + +import { strict as assert } from 'node:assert'; + +// ------------------------------------------------------------------ +// 1. Unit test for the queue primitives +// ------------------------------------------------------------------ + +/** + * Replicate the queue logic from bridge.js so we can test it in + * isolation without importing the full module (which would trigger + * Baileys / express side effects). + */ +function createSendQueue() { + let _sendQueue = Promise.resolve(); + + function enqueueSend(fn) { + const task = _sendQueue.then(() => fn(), () => fn()); + _sendQueue = task.catch(() => {}); + return task; + } + + return { enqueueSend }; +} + +// -- serial ordering ------------------------------------------------- +{ + const { enqueueSend } = createSendQueue(); + const order = []; + + const a = enqueueSend(async () => { + await new Promise(r => setTimeout(r, 30)); + order.push('a'); + return 'A'; + }); + const b = enqueueSend(async () => { + order.push('b'); + return 'B'; + }); + const c = enqueueSend(async () => { + await new Promise(r => setTimeout(r, 10)); + order.push('c'); + return 'C'; + }); + + const results = await Promise.all([a, b, c]); + assert.deepStrictEqual(results, ['A', 'B', 'C'], 'all tasks resolve'); + assert.deepStrictEqual(order, ['a', 'b', 'c'], 'tasks execute in FIFO order'); + console.log(' ✓ serial ordering'); +} + +// -- error isolation (one rejection does not stall the queue) -------- +{ + const { enqueueSend } = createSendQueue(); + const order = []; + + const bad = enqueueSend(async () => { + order.push('bad'); + throw new Error('boom'); + }); + const good = enqueueSend(async () => { + order.push('good'); + return 'ok'; + }); + + await assert.rejects(() => bad, /boom/, 'bad task rejects'); + const g = await good; + assert.strictEqual(g, 'ok', 'good task still resolves'); + assert.deepStrictEqual(order, ['bad', 'good'], 'good runs after bad'); + console.log(' ✓ error isolation'); +} + +// -- timeout still fires (wrapped inside enqueueSend) ---------------- +{ + const { enqueueSend } = createSendQueue(); + const timedOut = enqueueSend(async () => { + await new Promise((_, reject) => setTimeout(() => reject(new Error('timeout')), 20)); + }); + await assert.rejects(() => timedOut, /timeout/, 'inner timeout propagates'); + console.log(' ✓ timeout propagation'); +} + +// -- concurrent enqueues maintain single-consumer semantics ---------- +{ + const { enqueueSend } = createSendQueue(); + let concurrent = 0; + let maxConcurrent = 0; + + async function tracked() { + concurrent += 1; + if (concurrent > maxConcurrent) maxConcurrent = concurrent; + await new Promise(r => setTimeout(r, 5)); + concurrent -= 1; + } + + await Promise.all(Array.from({ length: 20 }, () => enqueueSend(tracked))); + assert.strictEqual(maxConcurrent, 1, 'never more than one in-flight'); + assert.strictEqual(concurrent, 0, 'all finished'); + console.log(' ✓ single-consumer concurrency'); +} + +console.log('\n✅ All send-queue tests passed.');