ruicesar/InfiniteAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix ack handling (#2373)

Browse Source 
* fix: ack handling to match wa web

* fix: prevent duplicate ack on message errors
This commit is contained in:
João Lucas
2026-03-13 16:45:49 -03:00
committed by GitHub
parent 6afde71691
commit c4e5d1262a
6 changed files with 959 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Socket/messages-recv.ts
+105 -175
View File
@@ -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<WACallEvent>(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<WACallEvent>(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<void> => {
return new Promise(resolve => setImmediate(resolve))
}
const makeOfflineNodeProcessor = () => {
const nodeProcessorMap: Map<MessageType, (node: BinaryNode) => Promise<void>> = new Map([
const offlineNodeProcessor = makeOfflineNodeProcessor(
new Map<MessageType, (node: BinaryNode) => Promise<void>>([
['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,
src/Utils/index.ts
+1
View File
@@ -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'
src/Utils/offline-node-processor.ts
+70
View File
@@ -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<void>
}
/**
* 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<MessageType, (node: BinaryNode) => Promise<void>>,
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 }
}
src/Utils/stanza-ack.ts
+45
View File
@@ -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
}
src/__tests__/Utils/offline-node-processor.test.ts
+318
View File
@@ -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<MessageType, (node: BinaryNode) => Promise<void>>, 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<void>>().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<void>>().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<MessageType, (node: BinaryNode) => Promise<void>>([
[
'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<void>>().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<void>>().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<void>>().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<void>>().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<void>>().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<void>>().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<void>>().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<void>>().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<void>(r => {
resolveFirst = r
})
let callCount = 0
const handler = jest.fn<(node: BinaryNode) => Promise<void>>().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<void>>().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)
})
})
})
src/__tests__/Utils/stanza-ack.test.ts
+420
View File
@@ -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'
}
})
})
})
})