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

Merge branch 'master' into feat/call-signaling-upstream

Browse Source
This commit is contained in:
Renato Alcara
2026-03-13 20:53:22 -03:00
committed by GitHub
parent 83a166f1cc c4e5d1262a
commit 49241af14f
8 changed files with 974 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Socket/messages-recv.ts
+106 -133
View File
@@ -52,6 +52,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,
@@ -344,40 +346,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)
}
@@ -1509,7 +1480,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
})
])
} finally {
await sendMessageAck(node)
await sendMessageAck(node).catch(ackErr => logger.error({ ackErr }, 'failed to ack receipt'))
}
}
@@ -1546,7 +1517,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
})
])
} finally {
await sendMessageAck(node)
await sendMessageAck(node).catch(ackErr => logger.error({ ackErr }, 'failed to ack notification'))
}
}
@@ -1565,46 +1536,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)
}
@@ -1622,12 +1590,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)
}
@@ -1665,6 +1635,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 {
@@ -1676,6 +1647,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)
}
}
@@ -1723,6 +1695,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
}
}
acked = true
await sendMessageAck(node, NACK_REASONS.UnhandledError)
})
}
@@ -1750,6 +1723,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
@@ -1759,6 +1733,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')
}
@@ -1770,6 +1745,11 @@ 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')
)
}
}
}
@@ -1910,8 +1890,56 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
ev.emit('call', [call])
}
try {
const { attrs } = node
const [infoChild] = getAllBinaryNodeChildren(node)
await sendMessageAck(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 handleBadAck = async ({ attrs }: BinaryNode) => {
@@ -1977,74 +2005,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,