From c4e5d1262aa143dd4f3c2df94ca5537b8db2b3d7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= <55464917+jlucaso1@users.noreply.github.com> Date: Fri, 13 Mar 2026 16:45:49 -0300 Subject: [PATCH] Fix ack handling (#2373) * fix: ack handling to match wa web * fix: prevent duplicate ack on message errors --- src/Socket/messages-recv.ts | 280 +++++------- src/Utils/index.ts | 1 + src/Utils/offline-node-processor.ts | 70 +++ src/Utils/stanza-ack.ts | 45 ++ .../Utils/offline-node-processor.test.ts | 318 +++++++++++++ src/__tests__/Utils/stanza-ack.test.ts | 420 ++++++++++++++++++ 6 files changed, 959 insertions(+), 175 deletions(-) create mode 100644 src/Utils/offline-node-processor.ts create mode 100644 src/Utils/stanza-ack.ts create mode 100644 src/__tests__/Utils/offline-node-processor.test.ts create mode 100644 src/__tests__/Utils/stanza-ack.test.ts diff --git a/src/Socket/messages-recv.ts b/src/Socket/messages-recv.ts index 2c1a5776..9e905277 100644 --- a/src/Socket/messages-recv.ts +++ b/src/Socket/messages-recv.ts @@ -50,6 +50,8 @@ import { xmppSignedPreKey } from '../Utils' import { makeMutex } from '../Utils/make-mutex' +import { makeOfflineNodeProcessor, type MessageType } from '../Utils/offline-node-processor' +import { buildAckStanza } from '../Utils/stanza-ack' import { areJidsSameUser, type BinaryNode, @@ -342,40 +344,9 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { } } - const sendMessageAck = async ({ tag, attrs, content }: BinaryNode, errorCode?: number) => { - const stanza: BinaryNode = { - tag: 'ack', - attrs: { - id: attrs.id!, - to: attrs.from!, - class: tag - } - } - - if (!!errorCode) { - stanza.attrs.error = errorCode.toString() - } - - if (!!attrs.participant) { - stanza.attrs.participant = attrs.participant - } - - if (!!attrs.recipient) { - stanza.attrs.recipient = attrs.recipient - } - - if ( - !!attrs.type && - (tag !== 'message' || getBinaryNodeChild({ tag, attrs, content }, 'unavailable') || errorCode !== 0) - ) { - stanza.attrs.type = attrs.type - } - - if (tag === 'message' && getBinaryNodeChild({ tag, attrs, content }, 'unavailable')) { - stanza.attrs.from = authState.creds.me!.id - } - - logger.debug({ recv: { tag, attrs }, sent: stanza.attrs }, 'sent ack') + const sendMessageAck = async (node: BinaryNode, errorCode?: number) => { + const stanza = buildAckStanza(node, errorCode, authState.creds.me!.id) + logger.debug({ recv: { tag: node.tag, attrs: node.attrs }, sent: stanza.attrs }, 'sent ack') await sendNode(stanza) } @@ -1138,7 +1109,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { }) ]) } finally { - await sendMessageAck(node) + await sendMessageAck(node).catch(ackErr => logger.error({ ackErr }, 'failed to ack receipt')) } } @@ -1175,7 +1146,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { }) ]) } finally { - await sendMessageAck(node) + await sendMessageAck(node).catch(ackErr => logger.error({ ackErr }, 'failed to ack notification')) } } @@ -1194,46 +1165,43 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { return } - const { - fullMessage: msg, - category, - author, - decrypt - } = decryptMessageNode(node, authState.creds.me!.id, authState.creds.me!.lid || '', signalRepository, logger) - - const alt = msg.key.participantAlt || msg.key.remoteJidAlt - // store new mappings we didn't have before - if (!!alt) { - const altServer = jidDecode(alt)?.server - const primaryJid = msg.key.participant || msg.key.remoteJid! - if (altServer === 'lid') { - if (!(await signalRepository.lidMapping.getPNForLID(alt))) { - await signalRepository.lidMapping.storeLIDPNMappings([{ lid: alt, pn: primaryJid }]) - await signalRepository.migrateSession(primaryJid, alt) - } - } else { - await signalRepository.lidMapping.storeLIDPNMappings([{ lid: primaryJid, pn: alt }]) - await signalRepository.migrateSession(alt, primaryJid) - } - } - - if (msg.key?.remoteJid && msg.key?.id && messageRetryManager) { - messageRetryManager.addRecentMessage(msg.key.remoteJid, msg.key.id, msg.message!) - logger.debug( - { - jid: msg.key.remoteJid, - id: msg.key.id - }, - 'Added message to recent cache for retry receipts' - ) - } + let acked = false try { + const { + fullMessage: msg, + category, + author, + decrypt + } = decryptMessageNode(node, authState.creds.me!.id, authState.creds.me!.lid || '', signalRepository, logger) + + const alt = msg.key.participantAlt || msg.key.remoteJidAlt + // store new mappings we didn't have before + if (!!alt) { + const altServer = jidDecode(alt)?.server + const primaryJid = msg.key.participant || msg.key.remoteJid! + if (altServer === 'lid') { + if (!(await signalRepository.lidMapping.getPNForLID(alt))) { + await signalRepository.lidMapping.storeLIDPNMappings([{ lid: alt, pn: primaryJid }]) + await signalRepository.migrateSession(primaryJid, alt) + } + } else { + await signalRepository.lidMapping.storeLIDPNMappings([{ lid: primaryJid, pn: alt }]) + await signalRepository.migrateSession(alt, primaryJid) + } + } + await messageMutex.mutex(async () => { await decrypt() + + if (msg.key?.remoteJid && msg.key?.id && msg.message && messageRetryManager) { + messageRetryManager.addRecentMessage(msg.key.remoteJid, msg.key.id, msg.message) + } + // message failed to decrypt if (msg.messageStubType === proto.WebMessageInfo.StubType.CIPHERTEXT && msg.category !== 'peer') { if (msg?.messageStubParameters?.[0] === MISSING_KEYS_ERROR_TEXT) { + acked = true return sendMessageAck(node, NACK_REASONS.ParsingError) } @@ -1251,12 +1219,14 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { { msgId: msg.key.id, unavailableType }, 'skipping placeholder resend for excluded unavailable type' ) + acked = true return sendMessageAck(node) } const messageAge = unixTimestampSeconds() - toNumber(msg.messageTimestamp) if (messageAge > PLACEHOLDER_MAX_AGE_SECONDS) { logger.debug({ msgId: msg.key.id, messageAge }, 'skipping placeholder resend for old message') + acked = true return sendMessageAck(node) } @@ -1294,6 +1264,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { .catch(err => { logger.warn({ err, msgId: msg.key.id }, 'failed to request placeholder resend for unavailable message') }) + acked = true await sendMessageAck(node) // Don't return — fall through to upsertMessage so the stub is emitted } else { @@ -1305,6 +1276,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { { msgId: msg.key.id, messageAge, remoteJid: msg.key.remoteJid }, 'skipping retry for expired status message' ) + acked = true return sendMessageAck(node) } } @@ -1352,6 +1324,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { } } + acked = true await sendMessageAck(node, NACK_REASONS.UnhandledError) }) } @@ -1379,6 +1352,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { type = 'inactive' } + acked = true await sendReceipt(msg.key.remoteJid!, participant!, [msg.key.id!], type) // send ack for history message @@ -1388,6 +1362,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { await sendReceipt(jid, undefined, [msg.key.id!], 'hist_sync') // TODO: investigate } } else { + acked = true await sendMessageAck(node) logger.debug({ key: msg.key }, 'processed newsletter message without receipts') } @@ -1399,55 +1374,65 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { }) } catch (error) { logger.error({ error, node: binaryNodeToString(node) }, 'error in handling message') + if (!acked) { + await sendMessageAck(node, NACK_REASONS.UnhandledError).catch(ackErr => + logger.error({ ackErr }, 'failed to ack message after error') + ) + } } } const handleCall = async (node: BinaryNode) => { - const { attrs } = node - const [infoChild] = getAllBinaryNodeChildren(node) - const status = getCallStatusFromNode(infoChild!) + try { + const { attrs } = node + const [infoChild] = getAllBinaryNodeChildren(node) - if (!infoChild) { - throw new Boom('Missing call info in call node') + if (!infoChild) { + throw new Boom('Missing call info in call node') + } + + const status = getCallStatusFromNode(infoChild) + + const callId = infoChild.attrs['call-id']! + const from = infoChild.attrs.from! || infoChild.attrs['call-creator']! + + const call: WACallEvent = { + chatId: attrs.from!, + from, + callerPn: infoChild.attrs['caller_pn'], + id: callId, + date: new Date(+attrs.t! * 1000), + offline: !!attrs.offline, + status + } + + if (status === 'offer') { + call.isVideo = !!getBinaryNodeChild(infoChild, 'video') + call.isGroup = infoChild.attrs.type === 'group' || !!infoChild.attrs['group-jid'] + call.groupJid = infoChild.attrs['group-jid'] + await callOfferCache.set(call.id, call) + } + + const existingCall = await callOfferCache.get(call.id) + + // use existing call info to populate this event + if (existingCall) { + call.isVideo = existingCall.isVideo + call.isGroup = existingCall.isGroup + call.callerPn = call.callerPn || existingCall.callerPn + } + + // delete data once call has ended + if (status === 'reject' || status === 'accept' || status === 'timeout' || status === 'terminate') { + await callOfferCache.del(call.id) + } + + ev.emit('call', [call]) + } catch (error) { + logger.error({ error, node: binaryNodeToString(node) }, 'error in handling call') + } finally { + await sendMessageAck(node).catch(ackErr => logger.error({ ackErr }, 'failed to ack call')) } - - const callId = infoChild.attrs['call-id']! - const from = infoChild.attrs.from! || infoChild.attrs['call-creator']! - - const call: WACallEvent = { - chatId: attrs.from!, - from, - callerPn: infoChild.attrs['caller_pn'], - id: callId, - date: new Date(+attrs.t! * 1000), - offline: !!attrs.offline, - status - } - - if (status === 'offer') { - call.isVideo = !!getBinaryNodeChild(infoChild, 'video') - call.isGroup = infoChild.attrs.type === 'group' || !!infoChild.attrs['group-jid'] - call.groupJid = infoChild.attrs['group-jid'] - await callOfferCache.set(call.id, call) - } - - const existingCall = await callOfferCache.get(call.id) - - // use existing call info to populate this event - if (existingCall) { - call.isVideo = existingCall.isVideo - call.isGroup = existingCall.isGroup - call.callerPn = call.callerPn || existingCall.callerPn - } - - // delete data once call has ended - if (status === 'reject' || status === 'accept' || status === 'timeout' || status === 'terminate') { - await callOfferCache.del(call.id) - } - - ev.emit('call', [call]) - - await sendMessageAck(node) } const handleBadAck = async ({ attrs }: BinaryNode) => { @@ -1513,74 +1498,19 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { } } - type MessageType = 'message' | 'call' | 'receipt' | 'notification' - - type OfflineNode = { - type: MessageType - node: BinaryNode - } - - /** Yields control to the event loop to prevent blocking */ - const yieldToEventLoop = (): Promise => { - return new Promise(resolve => setImmediate(resolve)) - } - - const makeOfflineNodeProcessor = () => { - const nodeProcessorMap: Map Promise> = new Map([ + const offlineNodeProcessor = makeOfflineNodeProcessor( + new Map Promise>([ ['message', handleMessage], ['call', handleCall], ['receipt', handleReceipt], ['notification', handleNotification] - ]) - const nodes: OfflineNode[] = [] - let isProcessing = false - - // Number of nodes to process before yielding to event loop - const BATCH_SIZE = 10 - - const enqueue = (type: MessageType, node: BinaryNode) => { - nodes.push({ type, node }) - - if (isProcessing) { - return - } - - isProcessing = true - - const promise = async () => { - let processedInBatch = 0 - - while (nodes.length && ws.isOpen) { - const { type, node } = nodes.shift()! - - const nodeProcessor = nodeProcessorMap.get(type) - - if (!nodeProcessor) { - onUnexpectedError(new Error(`unknown offline node type: ${type}`), 'processing offline node') - continue - } - - await nodeProcessor(node) - processedInBatch++ - - // Yield to event loop after processing a batch - // This prevents blocking the event loop for too long when there are many offline nodes - if (processedInBatch >= BATCH_SIZE) { - processedInBatch = 0 - await yieldToEventLoop() - } - } - - isProcessing = false - } - - promise().catch(error => onUnexpectedError(error, 'processing offline nodes')) + ]), + { + isWsOpen: () => ws.isOpen, + onUnexpectedError, + yieldToEventLoop: () => new Promise(resolve => setImmediate(resolve)) } - - return { enqueue } - } - - const offlineNodeProcessor = makeOfflineNodeProcessor() + ) const processNode = async ( type: MessageType, diff --git a/src/Utils/index.ts b/src/Utils/index.ts index e102ffe5..fa89de91 100644 --- a/src/Utils/index.ts +++ b/src/Utils/index.ts @@ -17,3 +17,4 @@ export * from './process-message' export * from './message-retry-manager' export * from './browser-utils' export * from './identity-change-handler' +export * from './stanza-ack' diff --git a/src/Utils/offline-node-processor.ts b/src/Utils/offline-node-processor.ts new file mode 100644 index 00000000..aa26debb --- /dev/null +++ b/src/Utils/offline-node-processor.ts @@ -0,0 +1,70 @@ +import type { BinaryNode } from '../WABinary' + +export type MessageType = 'message' | 'call' | 'receipt' | 'notification' + +type OfflineNode = { + type: MessageType + node: BinaryNode +} + +export type OfflineNodeProcessorDeps = { + isWsOpen: () => boolean + onUnexpectedError: (error: Error, msg: string) => void + yieldToEventLoop: () => Promise +} + +/** + * Creates a processor for offline stanza nodes that: + * - Queues nodes for sequential processing + * - Yields to the event loop periodically to avoid blocking + * - Catches handler errors to prevent the processing loop from crashing + */ +export function makeOfflineNodeProcessor( + nodeProcessorMap: Map Promise>, + deps: OfflineNodeProcessorDeps, + batchSize = 10 +) { + const nodes: OfflineNode[] = [] + let isProcessing = false + + const enqueue = (type: MessageType, node: BinaryNode) => { + nodes.push({ type, node }) + + if (isProcessing) { + return + } + + isProcessing = true + + const promise = async () => { + let processedInBatch = 0 + + while (nodes.length && deps.isWsOpen()) { + const { type, node } = nodes.shift()! + + const nodeProcessor = nodeProcessorMap.get(type) + + if (!nodeProcessor) { + deps.onUnexpectedError(new Error(`unknown offline node type: ${type}`), 'processing offline node') + continue + } + + await nodeProcessor(node).catch(err => deps.onUnexpectedError(err, `processing offline ${type}`)) + processedInBatch++ + + // Yield to event loop after processing a batch + // This prevents blocking the event loop for too long when there are many offline nodes + if (processedInBatch >= batchSize) { + processedInBatch = 0 + await deps.yieldToEventLoop() + } + } + + isProcessing = false + } + + promise().catch(error => deps.onUnexpectedError(error, 'processing offline nodes')) + } + + return { enqueue } +} diff --git a/src/Utils/stanza-ack.ts b/src/Utils/stanza-ack.ts new file mode 100644 index 00000000..5e2e478f --- /dev/null +++ b/src/Utils/stanza-ack.ts @@ -0,0 +1,45 @@ +import type { BinaryNode } from '../WABinary' + +/** + * Builds an ACK stanza for a received node. + * Pure function -- no I/O, no side effects. + * + * Mirrors WhatsApp Web's ACK construction: + * - WAWebHandleMsgSendAck.sendAck / sendNack + * - WAWebCreateNackFromStanza.createNackFromStanza + */ +export function buildAckStanza(node: BinaryNode, errorCode?: number, meId?: string): BinaryNode { + const { tag, attrs } = node + const stanza: BinaryNode = { + tag: 'ack', + attrs: { + id: attrs.id!, + to: attrs.from!, + class: tag + } + } + + if (errorCode) { + stanza.attrs.error = errorCode.toString() + } + + if (attrs.participant) { + stanza.attrs.participant = attrs.participant + } + + if (attrs.recipient) { + stanza.attrs.recipient = attrs.recipient + } + + // WA Web always includes type when present: `n.type || DROP_ATTR` + if (attrs.type) { + stanza.attrs.type = attrs.type + } + + // WA Web WAWebHandleMsgSendAck.sendAck/sendNack always include `from` for message-class ACKs + if (tag === 'message' && meId) { + stanza.attrs.from = meId + } + + return stanza +} diff --git a/src/__tests__/Utils/offline-node-processor.test.ts b/src/__tests__/Utils/offline-node-processor.test.ts new file mode 100644 index 00000000..58b1d29d --- /dev/null +++ b/src/__tests__/Utils/offline-node-processor.test.ts @@ -0,0 +1,318 @@ +import { jest } from '@jest/globals' +import { makeOfflineNodeProcessor, type MessageType } from '../../Utils/offline-node-processor' +import { type BinaryNode } from '../../WABinary' + +function makeNode(id: string, tag = 'message'): BinaryNode { + return { tag, attrs: { id, from: 'user@s.whatsapp.net', offline: '1' } } +} + +describe('makeOfflineNodeProcessor', () => { + let mockOnUnexpectedError: jest.Mock<(error: Error, msg: string) => void> + let isWsOpen: boolean + let yieldCalls: number + + function createProcessor(handlers: Map Promise>, batchSize = 10) { + return makeOfflineNodeProcessor( + handlers, + { + isWsOpen: () => isWsOpen, + onUnexpectedError: mockOnUnexpectedError, + yieldToEventLoop: async () => { + yieldCalls++ + } + }, + batchSize + ) + } + + beforeEach(() => { + mockOnUnexpectedError = jest.fn() + isWsOpen = true + yieldCalls = 0 + }) + + describe('basic processing', () => { + it('should process a single enqueued node', async () => { + const processed: string[] = [] + const handler = jest.fn<(node: BinaryNode) => Promise>().mockImplementation(async node => { + processed.push(node.attrs.id!) + }) + + const processor = createProcessor(new Map([['message', handler]])) + processor.enqueue('message', makeNode('msg-1')) + + // wait for microtask queue to flush + await new Promise(r => setTimeout(r, 10)) + expect(processed).toEqual(['msg-1']) + expect(handler).toHaveBeenCalledTimes(1) + }) + + it('should process multiple nodes in FIFO order', async () => { + const processed: string[] = [] + const handler = jest.fn<(node: BinaryNode) => Promise>().mockImplementation(async node => { + processed.push(node.attrs.id!) + }) + + const processor = createProcessor(new Map([['message', handler]])) + processor.enqueue('message', makeNode('msg-1')) + processor.enqueue('message', makeNode('msg-2')) + processor.enqueue('message', makeNode('msg-3')) + + await new Promise(r => setTimeout(r, 10)) + expect(processed).toEqual(['msg-1', 'msg-2', 'msg-3']) + }) + + it('should dispatch nodes to correct handler by type', async () => { + const messageIds: string[] = [] + const callIds: string[] = [] + const receiptIds: string[] = [] + const notificationIds: string[] = [] + + const handlers = new Map Promise>([ + [ + 'message', + async n => { + messageIds.push(n.attrs.id!) + } + ], + [ + 'call', + async n => { + callIds.push(n.attrs.id!) + } + ], + [ + 'receipt', + async n => { + receiptIds.push(n.attrs.id!) + } + ], + [ + 'notification', + async n => { + notificationIds.push(n.attrs.id!) + } + ] + ]) + + const processor = createProcessor(handlers) + processor.enqueue('message', makeNode('msg-1')) + processor.enqueue('call', makeNode('call-1', 'call')) + processor.enqueue('receipt', makeNode('rcpt-1', 'receipt')) + processor.enqueue('notification', makeNode('notif-1', 'notification')) + + await new Promise(r => setTimeout(r, 10)) + expect(messageIds).toEqual(['msg-1']) + expect(callIds).toEqual(['call-1']) + expect(receiptIds).toEqual(['rcpt-1']) + expect(notificationIds).toEqual(['notif-1']) + }) + }) + + describe('error resilience', () => { + it('should continue processing after a handler throws', async () => { + const processed: string[] = [] + const handler = jest.fn<(node: BinaryNode) => Promise>().mockImplementation(async node => { + if (node.attrs.id === 'msg-2') { + throw new Error('handler crash') + } + + processed.push(node.attrs.id!) + }) + + const processor = createProcessor(new Map([['message', handler]])) + processor.enqueue('message', makeNode('msg-1')) + processor.enqueue('message', makeNode('msg-2')) + processor.enqueue('message', makeNode('msg-3')) + + await new Promise(r => setTimeout(r, 10)) + // msg-1 and msg-3 should be processed, msg-2 error should be caught + expect(processed).toEqual(['msg-1', 'msg-3']) + expect(mockOnUnexpectedError).toHaveBeenCalledTimes(1) + expect(mockOnUnexpectedError).toHaveBeenCalledWith(expect.any(Error), 'processing offline message') + }) + + it('should continue processing after multiple handler errors', async () => { + const processed: string[] = [] + const handler = jest.fn<(node: BinaryNode) => Promise>().mockImplementation(async node => { + if (node.attrs.id === 'msg-1' || node.attrs.id === 'msg-3') { + throw new Error('crash') + } + + processed.push(node.attrs.id!) + }) + + const processor = createProcessor(new Map([['message', handler]])) + for (let i = 1; i <= 5; i++) { + processor.enqueue('message', makeNode(`msg-${i}`)) + } + + await new Promise(r => setTimeout(r, 10)) + expect(processed).toEqual(['msg-2', 'msg-4', 'msg-5']) + expect(mockOnUnexpectedError).toHaveBeenCalledTimes(2) + }) + + it('should report unknown node type and continue', async () => { + const processed: string[] = [] + const handler = jest.fn<(node: BinaryNode) => Promise>().mockImplementation(async node => { + processed.push(node.attrs.id!) + }) + + const processor = createProcessor(new Map([['message', handler]])) + // Enqueue an unknown type + processor.enqueue('unknown-type' as MessageType, makeNode('unknown-1')) + processor.enqueue('message', makeNode('msg-1')) + + await new Promise(r => setTimeout(r, 10)) + expect(processed).toEqual(['msg-1']) + expect(mockOnUnexpectedError).toHaveBeenCalledWith( + expect.objectContaining({ message: expect.stringContaining('unknown offline node type') }), + 'processing offline node' + ) + }) + }) + + describe('connection awareness', () => { + it('should stop processing when connection closes', async () => { + const processed: string[] = [] + const handler = jest.fn<(node: BinaryNode) => Promise>().mockImplementation(async node => { + processed.push(node.attrs.id!) + if (node.attrs.id === 'msg-2') { + isWsOpen = false + } + }) + + const processor = createProcessor(new Map([['message', handler]])) + for (let i = 1; i <= 5; i++) { + processor.enqueue('message', makeNode(`msg-${i}`)) + } + + await new Promise(r => setTimeout(r, 10)) + // Should stop after msg-2 closes the connection + expect(processed).toEqual(['msg-1', 'msg-2']) + }) + + it('should resume processing when new nodes are enqueued after connection reopens', async () => { + const processed: string[] = [] + const handler = jest.fn<(node: BinaryNode) => Promise>().mockImplementation(async node => { + processed.push(node.attrs.id!) + }) + + // Start with closed connection + isWsOpen = false + const processor = createProcessor(new Map([['message', handler]])) + processor.enqueue('message', makeNode('msg-1')) + + await new Promise(r => setTimeout(r, 10)) + expect(processed).toEqual([]) + + // Reopen connection and enqueue new node + isWsOpen = true + processor.enqueue('message', makeNode('msg-2')) + + await new Promise(r => setTimeout(r, 10)) + // Both nodes should now be processed (msg-1 was still in queue) + expect(processed).toEqual(['msg-1', 'msg-2']) + }) + }) + + describe('batch yielding', () => { + it('should yield to event loop after batchSize nodes', async () => { + const handler = jest.fn<(node: BinaryNode) => Promise>().mockResolvedValue(undefined) + const batchSize = 3 + const processor = createProcessor(new Map([['message', handler]]), batchSize) + + for (let i = 1; i <= 7; i++) { + processor.enqueue('message', makeNode(`msg-${i}`)) + } + + await new Promise(r => setTimeout(r, 10)) + expect(handler).toHaveBeenCalledTimes(7) + // 7 nodes with batchSize 3 => yields after 3rd and 6th = 2 yields + expect(yieldCalls).toBe(2) + }) + + it('should NOT yield for fewer nodes than batchSize', async () => { + const handler = jest.fn<(node: BinaryNode) => Promise>().mockResolvedValue(undefined) + const processor = createProcessor(new Map([['message', handler]]), 10) + + for (let i = 1; i <= 5; i++) { + processor.enqueue('message', makeNode(`msg-${i}`)) + } + + await new Promise(r => setTimeout(r, 10)) + expect(handler).toHaveBeenCalledTimes(5) + expect(yieldCalls).toBe(0) + }) + + it('should yield exactly at batchSize boundary', async () => { + const handler = jest.fn<(node: BinaryNode) => Promise>().mockResolvedValue(undefined) + const processor = createProcessor(new Map([['message', handler]]), 3) + + for (let i = 1; i <= 3; i++) { + processor.enqueue('message', makeNode(`msg-${i}`)) + } + + await new Promise(r => setTimeout(r, 10)) + expect(handler).toHaveBeenCalledTimes(3) + expect(yieldCalls).toBe(1) + }) + }) + + describe('isProcessing guard', () => { + it('should not start a second processing loop for concurrent enqueues', async () => { + let resolveFirst: (() => void) | undefined + const firstPromise = new Promise(r => { + resolveFirst = r + }) + let callCount = 0 + + const handler = jest.fn<(node: BinaryNode) => Promise>().mockImplementation(async () => { + callCount++ + if (callCount === 1) { + // Block on first node to simulate slow processing + await firstPromise + } + }) + + const processor = createProcessor(new Map([['message', handler]])) + processor.enqueue('message', makeNode('msg-1')) + + // Give time for first processing to start + await new Promise(r => setTimeout(r, 5)) + + // Enqueue while first is still processing + processor.enqueue('message', makeNode('msg-2')) + processor.enqueue('message', makeNode('msg-3')) + + // Release the first handler + resolveFirst!() + + await new Promise(r => setTimeout(r, 10)) + expect(handler).toHaveBeenCalledTimes(3) + }) + }) + + describe('mixed error types', () => { + it('should handle both handler errors and unknown types in sequence', async () => { + const processed: string[] = [] + const handler = jest.fn<(node: BinaryNode) => Promise>().mockImplementation(async node => { + if (node.attrs.id === 'msg-2') { + throw new Error('boom') + } + + processed.push(node.attrs.id!) + }) + + const processor = createProcessor(new Map([['message', handler]])) + processor.enqueue('message', makeNode('msg-1')) + processor.enqueue('message', makeNode('msg-2')) // will throw + processor.enqueue('bogus' as MessageType, makeNode('x')) // unknown type + processor.enqueue('message', makeNode('msg-3')) + + await new Promise(r => setTimeout(r, 10)) + expect(processed).toEqual(['msg-1', 'msg-3']) + expect(mockOnUnexpectedError).toHaveBeenCalledTimes(2) + }) + }) +}) diff --git a/src/__tests__/Utils/stanza-ack.test.ts b/src/__tests__/Utils/stanza-ack.test.ts new file mode 100644 index 00000000..fae8cb72 --- /dev/null +++ b/src/__tests__/Utils/stanza-ack.test.ts @@ -0,0 +1,420 @@ +import { buildAckStanza } from '../../Utils/stanza-ack' +import { type BinaryNode } from '../../WABinary' + +describe('buildAckStanza', () => { + describe('basic stanza construction', () => { + it('should build a minimal ACK stanza from a message node', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { + id: 'msg-001', + from: 'user@s.whatsapp.net' + } + } + + const ack = buildAckStanza(node) + expect(ack).toEqual({ + tag: 'ack', + attrs: { + id: 'msg-001', + to: 'user@s.whatsapp.net', + class: 'message' + } + }) + }) + + it('should build ACK for receipt node', () => { + const node: BinaryNode = { + tag: 'receipt', + attrs: { + id: 'rcpt-001', + from: 'user@s.whatsapp.net' + } + } + + const ack = buildAckStanza(node) + expect(ack.attrs.class).toBe('receipt') + }) + + it('should build ACK for notification node', () => { + const node: BinaryNode = { + tag: 'notification', + attrs: { + id: 'notif-001', + from: 'user@s.whatsapp.net' + } + } + + const ack = buildAckStanza(node) + expect(ack.attrs.class).toBe('notification') + }) + + it('should build ACK for call node', () => { + const node: BinaryNode = { + tag: 'call', + attrs: { + id: 'call-001', + from: 'user@s.whatsapp.net' + } + } + + const ack = buildAckStanza(node) + expect(ack.attrs.class).toBe('call') + }) + }) + + describe('error codes (NACK)', () => { + it('should include error attribute when errorCode is provided and non-zero', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { id: 'msg-001', from: 'user@s.whatsapp.net' } + } + + const ack = buildAckStanza(node, 500) + expect(ack.attrs.error).toBe('500') + }) + + it('should include error for all known NACK codes', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { id: 'msg-001', from: 'user@s.whatsapp.net' } + } + + const nackCodes = [487, 488, 491, 495, 496, 500, 552] + for (const code of nackCodes) { + const ack = buildAckStanza(node, code) + expect(ack.attrs.error).toBe(code.toString()) + } + }) + + it('should NOT include error when errorCode is 0', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { id: 'msg-001', from: 'user@s.whatsapp.net' } + } + + const ack = buildAckStanza(node, 0) + expect(ack.attrs.error).toBeUndefined() + }) + + it('should NOT include error when errorCode is undefined', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { id: 'msg-001', from: 'user@s.whatsapp.net' } + } + + const ack = buildAckStanza(node) + expect(ack.attrs.error).toBeUndefined() + }) + }) + + describe('participant and recipient forwarding', () => { + it('should include participant when present in source node', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { + id: 'msg-001', + from: 'group@g.us', + participant: 'sender@s.whatsapp.net' + } + } + + const ack = buildAckStanza(node) + expect(ack.attrs.participant).toBe('sender@s.whatsapp.net') + }) + + it('should NOT include participant when absent', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { id: 'msg-001', from: 'user@s.whatsapp.net' } + } + + const ack = buildAckStanza(node) + expect(ack.attrs.participant).toBeUndefined() + }) + + it('should include recipient when present', () => { + const node: BinaryNode = { + tag: 'receipt', + attrs: { + id: 'rcpt-001', + from: 'user@s.whatsapp.net', + recipient: 'me@s.whatsapp.net' + } + } + + const ack = buildAckStanza(node) + expect(ack.attrs.recipient).toBe('me@s.whatsapp.net') + }) + + it('should NOT include recipient when absent', () => { + const node: BinaryNode = { + tag: 'receipt', + attrs: { id: 'rcpt-001', from: 'user@s.whatsapp.net' } + } + + const ack = buildAckStanza(node) + expect(ack.attrs.recipient).toBeUndefined() + }) + + it('should forward both participant and recipient when both present', () => { + const node: BinaryNode = { + tag: 'receipt', + attrs: { + id: 'rcpt-001', + from: 'group@g.us', + participant: 'sender@s.whatsapp.net', + recipient: 'me@s.whatsapp.net' + } + } + + const ack = buildAckStanza(node) + expect(ack.attrs.participant).toBe('sender@s.whatsapp.net') + expect(ack.attrs.recipient).toBe('me@s.whatsapp.net') + }) + }) + + describe('type attribute handling (WA Web: n.type || DROP_ATTR)', () => { + it('should include type for non-message tags', () => { + const node: BinaryNode = { + tag: 'notification', + attrs: { + id: 'notif-001', + from: 'user@s.whatsapp.net', + type: 'encrypt' + } + } + + const ack = buildAckStanza(node) + expect(ack.attrs.type).toBe('encrypt') + }) + + it('should always include type for message tag when present', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { + id: 'msg-001', + from: 'user@s.whatsapp.net', + type: 'text' + } + } + + // type is always included when present, regardless of errorCode + const ack = buildAckStanza(node, 0) + expect(ack.attrs.type).toBe('text') + }) + + it('should include type for message NACK', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { + id: 'msg-001', + from: 'user@s.whatsapp.net', + type: 'text' + } + } + + const ack = buildAckStanza(node, 500) + expect(ack.attrs.type).toBe('text') + }) + + it('should NOT include type when source node has no type', () => { + const node: BinaryNode = { + tag: 'notification', + attrs: { id: 'notif-001', from: 'user@s.whatsapp.net' } + } + + const ack = buildAckStanza(node) + expect(ack.attrs.type).toBeUndefined() + }) + + it('should include type for receipt tag', () => { + const node: BinaryNode = { + tag: 'receipt', + attrs: { + id: 'rcpt-001', + from: 'user@s.whatsapp.net', + type: 'read' + } + } + + const ack = buildAckStanza(node) + expect(ack.attrs.type).toBe('read') + }) + }) + + describe('from field for message ACKs (WA Web: sendAck/sendNack always include from)', () => { + it('should set from=meId for message ACK when meId provided', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { + id: 'msg-001', + from: 'user@s.whatsapp.net' + } + } + + const ack = buildAckStanza(node, undefined, 'me@s.whatsapp.net') + expect(ack.attrs.from).toBe('me@s.whatsapp.net') + }) + + it('should set from=meId for message NACK when meId provided', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { + id: 'msg-001', + from: 'user@s.whatsapp.net' + } + } + + const ack = buildAckStanza(node, 500, 'me@s.whatsapp.net') + expect(ack.attrs.from).toBe('me@s.whatsapp.net') + }) + + it('should set from=meId for unavailable message with meId', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { + id: 'msg-001', + from: 'user@s.whatsapp.net', + type: 'text' + }, + content: [{ tag: 'unavailable', attrs: {} }] + } + + const ack = buildAckStanza(node, 0, 'me@s.whatsapp.net') + expect(ack.attrs.from).toBe('me@s.whatsapp.net') + }) + + it('should NOT set from for message ACK without meId', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { + id: 'msg-001', + from: 'user@s.whatsapp.net' + } + } + + const ack = buildAckStanza(node) + expect(ack.attrs.from).toBeUndefined() + }) + + it('should NOT set from for non-message tags even with meId', () => { + const node: BinaryNode = { + tag: 'notification', + attrs: { + id: 'notif-001', + from: 'user@s.whatsapp.net', + type: 'encrypt' + } + } + + const ack = buildAckStanza(node, 0, 'me@s.whatsapp.net') + expect(ack.attrs.from).toBeUndefined() + }) + + it('should NOT set from for receipt tag even with meId', () => { + const node: BinaryNode = { + tag: 'receipt', + attrs: { + id: 'rcpt-001', + from: 'user@s.whatsapp.net' + } + } + + const ack = buildAckStanza(node, undefined, 'me@s.whatsapp.net') + expect(ack.attrs.from).toBeUndefined() + }) + + it('should NOT set from for call tag even with meId', () => { + const node: BinaryNode = { + tag: 'call', + attrs: { + id: 'call-001', + from: 'user@s.whatsapp.net' + } + } + + const ack = buildAckStanza(node, undefined, 'me@s.whatsapp.net') + expect(ack.attrs.from).toBeUndefined() + }) + }) + + describe('full stanza construction (combined attributes)', () => { + it('should handle group message NACK with all attributes', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { + id: 'msg-001', + from: 'group@g.us', + participant: 'sender@s.whatsapp.net', + type: 'text' + } + } + + const ack = buildAckStanza(node, 500, 'me@s.whatsapp.net') + expect(ack).toEqual({ + tag: 'ack', + attrs: { + id: 'msg-001', + to: 'group@g.us', + class: 'message', + error: '500', + participant: 'sender@s.whatsapp.net', + type: 'text', + from: 'me@s.whatsapp.net' + } + }) + }) + + it('should handle group message ACK (no error) with from', () => { + const node: BinaryNode = { + tag: 'message', + attrs: { + id: 'msg-001', + from: 'group@g.us', + participant: 'sender@s.whatsapp.net', + type: 'text' + } + } + + const ack = buildAckStanza(node, undefined, 'me@s.whatsapp.net') + expect(ack).toEqual({ + tag: 'ack', + attrs: { + id: 'msg-001', + to: 'group@g.us', + class: 'message', + participant: 'sender@s.whatsapp.net', + type: 'text', + from: 'me@s.whatsapp.net' + } + }) + }) + + it('should handle receipt with participant and recipient (no from)', () => { + const node: BinaryNode = { + tag: 'receipt', + attrs: { + id: 'rcpt-001', + from: 'group@g.us', + participant: 'sender@s.whatsapp.net', + recipient: 'me@s.whatsapp.net', + type: 'read' + } + } + + const ack = buildAckStanza(node) + expect(ack).toEqual({ + tag: 'ack', + attrs: { + id: 'rcpt-001', + to: 'group@g.us', + class: 'receipt', + participant: 'sender@s.whatsapp.net', + recipient: 'me@s.whatsapp.net', + type: 'read' + } + }) + }) + }) +})