Renato Alcara
3d9d7bafe4
perf: reduce message delivery latency across all connection scenarios (#239)
2169 lines
69 KiB
TypeScript
2169 lines
69 KiB
TypeScript
/* eslint-disable max-depth, @typescript-eslint/no-unused-vars */
|
|
import NodeCache from '@cacheable/node-cache'
|
|
import { Boom } from '@hapi/boom'
|
|
import { randomBytes } from 'crypto'
|
|
import Long from 'long'
|
|
import { proto } from '../../WAProto/index.js'
|
|
import {
|
|
DEFAULT_CACHE_TTLS,
|
|
DEFAULT_SESSION_CLEANUP_CONFIG,
|
|
KEY_BUNDLE_TYPE,
|
|
MIN_PREKEY_COUNT,
|
|
PLACEHOLDER_MAX_AGE_SECONDS,
|
|
STATUS_EXPIRY_SECONDS
|
|
} from '../Defaults'
|
|
import type {
|
|
GroupParticipant,
|
|
MessageReceiptType,
|
|
MessageRelayOptions,
|
|
MessageUserReceipt,
|
|
PlaceholderMessageData,
|
|
SocketConfig,
|
|
WACallEvent,
|
|
WAMessage,
|
|
WAMessageKey,
|
|
WAPatchName
|
|
} from '../Types'
|
|
import { WAMessageStatus, WAMessageStubType } from '../Types'
|
|
import {
|
|
aesDecryptCTR,
|
|
aesEncryptGCM,
|
|
cleanMessage,
|
|
Curve,
|
|
decodeMediaRetryNode,
|
|
decodeMessageNode,
|
|
decryptMessageNode,
|
|
delay,
|
|
derivePairingCodeKey,
|
|
encodeBigEndian,
|
|
encodeSignedDeviceIdentity,
|
|
extractAddressingContext,
|
|
getCallStatusFromNode,
|
|
getHistoryMsg,
|
|
getNextPreKeys,
|
|
getStatusFromReceiptType,
|
|
handleIdentityChange,
|
|
hkdf,
|
|
MISSING_KEYS_ERROR_TEXT,
|
|
NACK_REASONS,
|
|
NO_MESSAGE_FOUND_ERROR_TEXT,
|
|
normalizeMessageJids,
|
|
SERVER_ERROR_CODES,
|
|
toNumber,
|
|
unixTimestampSeconds,
|
|
xmppPreKey,
|
|
xmppSignedPreKey
|
|
} from '../Utils'
|
|
import { logMessageReceived, logTcToken } from '../Utils/baileys-logger'
|
|
import { makeMutex } from '../Utils/make-mutex'
|
|
import {
|
|
metrics,
|
|
recordHistorySyncMessages,
|
|
recordMessageFailure,
|
|
recordMessageReceived,
|
|
recordMessageRetry
|
|
} from '../Utils/prometheus-metrics.js'
|
|
import { isTcTokenExpired, resolveTcTokenJid } from '../Utils/tc-token-utils'
|
|
import {
|
|
areJidsSameUser,
|
|
type BinaryNode,
|
|
binaryNodeToString,
|
|
getAllBinaryNodeChildren,
|
|
getBinaryNodeChild,
|
|
getBinaryNodeChildBuffer,
|
|
getBinaryNodeChildren,
|
|
getBinaryNodeChildString,
|
|
isJidGroup,
|
|
isJidNewsletter,
|
|
isJidStatusBroadcast,
|
|
isLidUser,
|
|
isPnUser,
|
|
jidDecode,
|
|
jidNormalizedUser,
|
|
S_WHATSAPP_NET
|
|
} from '../WABinary'
|
|
import { extractGroupMetadata } from './groups'
|
|
import { makeMessagesSocket } from './messages-send'
|
|
|
|
export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
|
const {
|
|
logger,
|
|
retryRequestDelayMs,
|
|
maxMsgRetryCount,
|
|
getMessage,
|
|
shouldIgnoreJid,
|
|
enableAutoSessionRecreation,
|
|
enableCTWARecovery,
|
|
sessionCleanupConfig
|
|
} = config
|
|
// Use nullish coalescing to handle partial config properly
|
|
const autoCleanCorrupted =
|
|
sessionCleanupConfig?.autoCleanCorrupted ?? DEFAULT_SESSION_CLEANUP_CONFIG.autoCleanCorrupted
|
|
const sock = makeMessagesSocket(config)
|
|
const {
|
|
ev,
|
|
authState,
|
|
ws,
|
|
messageMutex,
|
|
notificationMutex,
|
|
receiptMutex,
|
|
signalRepository,
|
|
sessionActivityTracker,
|
|
query,
|
|
upsertMessage,
|
|
resyncAppState,
|
|
onUnexpectedError,
|
|
assertSessions,
|
|
sendNode,
|
|
relayMessage,
|
|
sendReceipt,
|
|
uploadPreKeys,
|
|
sendPeerDataOperationMessage,
|
|
messageRetryManager,
|
|
getPrivacyTokens
|
|
} = sock
|
|
|
|
const getLIDForPN = signalRepository.lidMapping.getLIDForPN.bind(signalRepository.lidMapping)
|
|
|
|
/** this mutex ensures that each retryRequest will wait for the previous one to finish */
|
|
const retryMutex = makeMutex()
|
|
|
|
const msgRetryCache =
|
|
config.msgRetryCounterCache ||
|
|
new NodeCache<number>({
|
|
stdTTL: DEFAULT_CACHE_TTLS.MSG_RETRY, // 1 hour
|
|
useClones: false
|
|
})
|
|
const callOfferCache =
|
|
config.callOfferCache ||
|
|
new NodeCache<WACallEvent>({
|
|
stdTTL: DEFAULT_CACHE_TTLS.CALL_OFFER, // 5 mins
|
|
useClones: false
|
|
})
|
|
|
|
const placeholderResendCache =
|
|
config.placeholderResendCache ||
|
|
new NodeCache({
|
|
stdTTL: DEFAULT_CACHE_TTLS.MSG_RETRY, // 1 hour
|
|
useClones: false
|
|
})
|
|
|
|
// Debounce identity-change session refreshes per JID to avoid bursts
|
|
const identityAssertDebounce = new NodeCache<boolean>({ stdTTL: 5, useClones: false })
|
|
|
|
let sendActiveReceipts = false
|
|
|
|
// ======= tctoken index tracking for cross-session pruning =======
|
|
const TC_TOKEN_INDEX_KEY = '__index'
|
|
const TC_TOKEN_PRUNE_TS_KEY = '__prune_ts'
|
|
const tcTokenKnownJids = new Set<string>()
|
|
const tcTokenRetriedMsgIds = new Set<string>()
|
|
let tcTokenIndexSaveTimer: ReturnType<typeof setTimeout> | undefined
|
|
let lastTcTokenPruneTs = 0
|
|
|
|
// Load persisted JID index and last prune timestamp on startup
|
|
const tcTokenIndexLoaded = (async () => {
|
|
try {
|
|
const data = await authState.keys.get('tctoken', [TC_TOKEN_INDEX_KEY, TC_TOKEN_PRUNE_TS_KEY])
|
|
const entry = data[TC_TOKEN_INDEX_KEY]
|
|
if (entry?.token) {
|
|
const stored = JSON.parse(Buffer.from(entry.token).toString('utf8'))
|
|
if (Array.isArray(stored)) {
|
|
for (const jid of stored) tcTokenKnownJids.add(jid)
|
|
}
|
|
}
|
|
|
|
const pruneEntry = data[TC_TOKEN_PRUNE_TS_KEY]
|
|
if (pruneEntry?.timestamp) {
|
|
lastTcTokenPruneTs = Number(pruneEntry.timestamp)
|
|
}
|
|
} catch {
|
|
/* first run or corrupt index — start fresh */
|
|
}
|
|
})()
|
|
|
|
/** Debounced save of the tctoken JID index (5s) */
|
|
const scheduleTcTokenIndexSave = () => {
|
|
if (tcTokenIndexSaveTimer) clearTimeout(tcTokenIndexSaveTimer)
|
|
tcTokenIndexSaveTimer = setTimeout(async () => {
|
|
try {
|
|
const arr = Array.from(tcTokenKnownJids)
|
|
await authState.keys.set({
|
|
tctoken: {
|
|
[TC_TOKEN_INDEX_KEY]: {
|
|
token: Buffer.from(JSON.stringify(arr), 'utf8'),
|
|
timestamp: unixTimestampSeconds().toString()
|
|
}
|
|
}
|
|
})
|
|
} catch (err) {
|
|
logger.debug({ err }, 'failed to persist tctoken index')
|
|
}
|
|
}, 5000)
|
|
}
|
|
|
|
/** Delete expired tctokens — runs at most once per 24h when coming online */
|
|
const pruneExpiredTcTokens = async () => {
|
|
await tcTokenIndexLoaded
|
|
const pruneSet: Record<string, null> = {}
|
|
const survivingJids: string[] = []
|
|
|
|
const jidsToCheck = Array.from(tcTokenKnownJids).filter(j => j !== TC_TOKEN_INDEX_KEY)
|
|
if (!jidsToCheck.length) return
|
|
|
|
try {
|
|
const allData = await authState.keys.get('tctoken', jidsToCheck)
|
|
for (const jid of jidsToCheck) {
|
|
const entry = allData[jid]
|
|
if (!entry?.token || isTcTokenExpired(entry.timestamp)) {
|
|
pruneSet[jid] = null
|
|
} else {
|
|
survivingJids.push(jid)
|
|
}
|
|
}
|
|
} catch {
|
|
return // batch read failed — skip this pruning cycle
|
|
}
|
|
|
|
const pruneCount = Object.keys(pruneSet).length
|
|
if (pruneCount > 0) {
|
|
await authState.keys.set({ tctoken: pruneSet })
|
|
tcTokenKnownJids.clear()
|
|
for (const jid of survivingJids) tcTokenKnownJids.add(jid)
|
|
scheduleTcTokenIndexSave()
|
|
logTcToken('prune', { pruned: pruneCount, remaining: survivingJids.length })
|
|
}
|
|
}
|
|
// ======= END tctoken index tracking =======
|
|
|
|
const fetchMessageHistory = async (
|
|
count: number,
|
|
oldestMsgKey: WAMessageKey,
|
|
oldestMsgTimestamp: number | Long
|
|
): Promise<string> => {
|
|
if (!authState.creds.me?.id) {
|
|
throw new Boom('Not authenticated')
|
|
}
|
|
|
|
const pdoMessage: proto.Message.IPeerDataOperationRequestMessage = {
|
|
historySyncOnDemandRequest: {
|
|
chatJid: oldestMsgKey.remoteJid,
|
|
oldestMsgFromMe: oldestMsgKey.fromMe,
|
|
oldestMsgId: oldestMsgKey.id,
|
|
oldestMsgTimestampMs: oldestMsgTimestamp,
|
|
onDemandMsgCount: count
|
|
},
|
|
peerDataOperationRequestType: proto.Message.PeerDataOperationRequestType.HISTORY_SYNC_ON_DEMAND
|
|
}
|
|
|
|
return sendPeerDataOperationMessage(pdoMessage)
|
|
}
|
|
|
|
const requestPlaceholderResend = async (
|
|
messageKey: WAMessageKey,
|
|
msgData?: PlaceholderMessageData
|
|
): Promise<string | undefined> => {
|
|
if (!authState.creds.me?.id) {
|
|
throw new Boom('Not authenticated')
|
|
}
|
|
|
|
// Check if already requested using message ID to prevent duplicate PDO requests
|
|
const alreadyRequested = await placeholderResendCache.get<PlaceholderMessageData | boolean>(messageKey?.id!)
|
|
if (alreadyRequested) {
|
|
logger.debug({ messageKey }, 'already requested resend')
|
|
return
|
|
}
|
|
|
|
// Temporarily mark as requested using message ID to prevent race conditions
|
|
await placeholderResendCache.set(messageKey?.id!, true)
|
|
|
|
await delay(2000)
|
|
|
|
// Check if message was received during delay
|
|
if (!(await placeholderResendCache.get<PlaceholderMessageData | boolean>(messageKey?.id!))) {
|
|
logger.debug({ messageKey }, 'message received while resend requested')
|
|
return 'RESOLVED'
|
|
}
|
|
|
|
const pdoMessage = {
|
|
placeholderMessageResendRequest: [
|
|
{
|
|
messageKey
|
|
}
|
|
],
|
|
peerDataOperationRequestType: proto.Message.PeerDataOperationRequestType.PLACEHOLDER_MESSAGE_RESEND
|
|
}
|
|
|
|
// Send PDO and get stanzaId (PDO request ID)
|
|
const stanzaId = await sendPeerDataOperationMessage(pdoMessage)
|
|
|
|
// CRITICAL FIX: Store metadata using stanzaId (not messageKey.id)
|
|
// The PDO response will use stanzaId to identify which request it's responding to
|
|
if (msgData && stanzaId) {
|
|
await placeholderResendCache.set(stanzaId, msgData)
|
|
logger.debug(
|
|
{ messageKey: messageKey.id, stanzaId },
|
|
'CTWA: Cached metadata using stanzaId for PDO response lookup'
|
|
)
|
|
}
|
|
|
|
// Clean up message ID marker after storing with stanzaId
|
|
await placeholderResendCache.del(messageKey?.id!)
|
|
|
|
// Cleanup timeout: if no response after 8s, assume phone is offline
|
|
setTimeout(async () => {
|
|
if (await placeholderResendCache.get<PlaceholderMessageData | boolean>(stanzaId)) {
|
|
logger.debug({ stanzaId }, 'PDO message without response after 8 seconds. Phone possibly offline')
|
|
await placeholderResendCache.del(stanzaId)
|
|
}
|
|
}, 8_000)
|
|
|
|
return stanzaId
|
|
}
|
|
|
|
// Handles mex newsletter notifications
|
|
const handleMexNewsletterNotification = async (node: BinaryNode) => {
|
|
const mexNode = getBinaryNodeChild(node, 'mex')
|
|
if (!mexNode?.content) {
|
|
logger.warn({ node }, 'Invalid mex newsletter notification')
|
|
return
|
|
}
|
|
|
|
let data: any
|
|
try {
|
|
data = JSON.parse(mexNode.content.toString())
|
|
} catch (error) {
|
|
logger.error({ err: error, node }, 'Failed to parse mex newsletter notification')
|
|
return
|
|
}
|
|
|
|
const operation = data?.operation
|
|
const updates = data?.updates
|
|
|
|
if (!updates || !operation) {
|
|
logger.warn({ data }, 'Invalid mex newsletter notification content')
|
|
return
|
|
}
|
|
|
|
logger.info({ operation, updates }, 'got mex newsletter notification')
|
|
|
|
switch (operation) {
|
|
case 'NotificationNewsletterUpdate':
|
|
for (const update of updates) {
|
|
if (update.jid && update.settings && Object.keys(update.settings).length > 0) {
|
|
ev.emit('newsletter-settings.update', {
|
|
id: update.jid,
|
|
update: update.settings
|
|
})
|
|
}
|
|
}
|
|
|
|
break
|
|
|
|
case 'NotificationNewsletterAdminPromote':
|
|
for (const update of updates) {
|
|
if (update.jid && update.user) {
|
|
ev.emit('newsletter-participants.update', {
|
|
id: update.jid,
|
|
author: node.attrs.from!,
|
|
user: update.user,
|
|
new_role: 'ADMIN',
|
|
action: 'promote'
|
|
})
|
|
}
|
|
}
|
|
|
|
break
|
|
|
|
default:
|
|
logger.info({ operation, data }, 'Unhandled mex newsletter notification')
|
|
break
|
|
}
|
|
}
|
|
|
|
// Handles newsletter notifications
|
|
const handleNewsletterNotification = async (node: BinaryNode) => {
|
|
const from = node.attrs.from!
|
|
const child = getAllBinaryNodeChildren(node)[0]!
|
|
const author = node.attrs.participant!
|
|
|
|
logger.info({ from, child }, 'got newsletter notification')
|
|
|
|
switch (child.tag) {
|
|
case 'reaction':
|
|
const reactionUpdate = {
|
|
id: from,
|
|
server_id: child.attrs.message_id!,
|
|
reaction: {
|
|
code: getBinaryNodeChildString(child, 'reaction'),
|
|
count: 1
|
|
}
|
|
}
|
|
ev.emit('newsletter.reaction', reactionUpdate)
|
|
break
|
|
|
|
case 'view':
|
|
const viewUpdate = {
|
|
id: from,
|
|
server_id: child.attrs.message_id!,
|
|
count: parseInt(child.content?.toString() || '0', 10)
|
|
}
|
|
ev.emit('newsletter.view', viewUpdate)
|
|
break
|
|
|
|
case 'participant':
|
|
const participantUpdate = {
|
|
id: from,
|
|
author,
|
|
user: child.attrs.jid!,
|
|
action: child.attrs.action!,
|
|
new_role: child.attrs.role!
|
|
}
|
|
ev.emit('newsletter-participants.update', participantUpdate)
|
|
break
|
|
|
|
case 'update':
|
|
const settingsNode = getBinaryNodeChild(child, 'settings')
|
|
if (settingsNode) {
|
|
const update: Record<string, any> = {}
|
|
const nameNode = getBinaryNodeChild(settingsNode, 'name')
|
|
if (nameNode?.content) update.name = nameNode.content.toString()
|
|
|
|
const descriptionNode = getBinaryNodeChild(settingsNode, 'description')
|
|
if (descriptionNode?.content) update.description = descriptionNode.content.toString()
|
|
|
|
ev.emit('newsletter-settings.update', {
|
|
id: from,
|
|
update
|
|
})
|
|
}
|
|
|
|
break
|
|
|
|
case 'message':
|
|
const plaintextNode = getBinaryNodeChild(child, 'plaintext')
|
|
if (plaintextNode?.content) {
|
|
try {
|
|
const contentBuf =
|
|
typeof plaintextNode.content === 'string'
|
|
? Buffer.from(plaintextNode.content, 'binary')
|
|
: Buffer.from(plaintextNode.content as Uint8Array)
|
|
const messageProto = proto.Message.decode(contentBuf).toJSON()
|
|
const fullMessage = proto.WebMessageInfo.fromObject({
|
|
key: {
|
|
remoteJid: from,
|
|
id: child.attrs.message_id || child.attrs.server_id,
|
|
fromMe: false // TODO: is this really true though
|
|
},
|
|
message: messageProto,
|
|
messageTimestamp: +child.attrs.t!
|
|
}).toJSON() as WAMessage
|
|
await upsertMessage(fullMessage, 'append')
|
|
logger.info('Processed plaintext newsletter message')
|
|
} catch (error) {
|
|
logger.error({ error }, 'Failed to decode plaintext newsletter message')
|
|
}
|
|
}
|
|
|
|
break
|
|
|
|
default:
|
|
logger.warn({ node }, 'Unknown newsletter notification')
|
|
break
|
|
}
|
|
}
|
|
|
|
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')) {
|
|
const meId = authState.creds.me?.id
|
|
if (!meId) throw new Boom('Not authenticated', { statusCode: 401 })
|
|
stanza.attrs.from = meId
|
|
}
|
|
|
|
logger.debug({ recv: { tag, attrs }, sent: stanza.attrs }, 'sent ack')
|
|
await sendNode(stanza)
|
|
}
|
|
|
|
const rejectCall = async (callId: string, callFrom: string) => {
|
|
const meId = authState.creds.me?.id
|
|
if (!meId) throw new Boom('Not authenticated', { statusCode: 401 })
|
|
const stanza: BinaryNode = {
|
|
tag: 'call',
|
|
attrs: {
|
|
from: meId,
|
|
to: callFrom
|
|
},
|
|
content: [
|
|
{
|
|
tag: 'reject',
|
|
attrs: {
|
|
'call-id': callId,
|
|
'call-creator': callFrom,
|
|
count: '0'
|
|
},
|
|
content: undefined
|
|
}
|
|
]
|
|
}
|
|
await query(stanza)
|
|
}
|
|
|
|
const sendRetryRequest = async (node: BinaryNode, forceIncludeKeys = false) => {
|
|
const { fullMessage } = decodeMessageNode(node, authState.creds.me!.id, authState.creds.me!.lid || '')
|
|
const { key: msgKey } = fullMessage
|
|
const msgId = msgKey.id!
|
|
|
|
if (messageRetryManager) {
|
|
// Check if we've exceeded max retries using the new system
|
|
if (messageRetryManager.hasExceededMaxRetries(msgId)) {
|
|
logger.debug({ msgId }, 'reached retry limit with new retry manager, clearing')
|
|
messageRetryManager.markRetryFailed(msgId)
|
|
recordMessageFailure('retry', 'max_retries_reached')
|
|
return
|
|
}
|
|
|
|
// Increment retry count using new system
|
|
const retryCount = messageRetryManager.incrementRetryCount(msgId)
|
|
recordMessageRetry('retry')
|
|
|
|
// Use the new retry count for the rest of the logic
|
|
const key = `${msgId}:${msgKey?.participant}`
|
|
await msgRetryCache.set(key, retryCount)
|
|
} else {
|
|
// Fallback to old system
|
|
const key = `${msgId}:${msgKey?.participant}`
|
|
let retryCount = (await msgRetryCache.get<number>(key)) || 0
|
|
if (retryCount >= maxMsgRetryCount) {
|
|
logger.debug({ retryCount, msgId }, 'reached retry limit, clearing')
|
|
await msgRetryCache.del(key)
|
|
recordMessageFailure('retry', 'max_retries_reached')
|
|
return
|
|
}
|
|
|
|
retryCount += 1
|
|
await msgRetryCache.set(key, retryCount)
|
|
recordMessageRetry('retry')
|
|
}
|
|
|
|
const key = `${msgId}:${msgKey?.participant}`
|
|
const retryCount = (await msgRetryCache.get<number>(key)) || 1
|
|
|
|
const { account, signedPreKey, signedIdentityKey: identityKey } = authState.creds
|
|
const fromJid = node.attrs.from!
|
|
|
|
// Check if we should recreate the session
|
|
let shouldRecreateSession = false
|
|
let recreateReason = ''
|
|
|
|
if (enableAutoSessionRecreation && messageRetryManager && retryCount > 1) {
|
|
try {
|
|
// Check if we have a session with this JID
|
|
const sessionId = signalRepository.jidToSignalProtocolAddress(fromJid)
|
|
const hasSession = await signalRepository.validateSession(fromJid)
|
|
|
|
// Extract error code from retry node if present (for MAC error detection)
|
|
const retryNode = getBinaryNodeChild(node, 'retry')
|
|
const errorAttr = retryNode?.attrs?.error
|
|
const errorCode = messageRetryManager.parseRetryErrorCode(errorAttr)
|
|
|
|
const result = messageRetryManager.shouldRecreateSession(fromJid, hasSession.exists, errorCode)
|
|
shouldRecreateSession = result.recreate
|
|
recreateReason = result.reason
|
|
|
|
if (shouldRecreateSession) {
|
|
logger.debug({ fromJid, retryCount, reason: recreateReason, errorCode }, 'recreating session for retry')
|
|
// Delete existing session to force recreation
|
|
// CRITICAL: Use same transaction key as encrypt/decrypt operations to prevent race
|
|
// Using meId ensures this delete serializes with sendMessage() and other session operations
|
|
await authState.keys.transaction(async () => {
|
|
await authState.keys.set({ session: { [sessionId]: null } })
|
|
}, authState.creds.me?.id || 'session-operation')
|
|
forceIncludeKeys = true
|
|
}
|
|
} catch (error) {
|
|
logger.warn({ error, fromJid }, 'failed to check session recreation')
|
|
}
|
|
}
|
|
|
|
if (retryCount <= 2) {
|
|
// Use new retry manager for phone requests if available
|
|
if (messageRetryManager) {
|
|
// Schedule phone request with delay (like whatsmeow)
|
|
messageRetryManager.schedulePhoneRequest(msgId, async () => {
|
|
try {
|
|
const requestId = await requestPlaceholderResend(msgKey)
|
|
logger.debug(
|
|
`sendRetryRequest: requested placeholder resend (${requestId}) for message ${msgId} (scheduled)`
|
|
)
|
|
} catch (error) {
|
|
logger.warn({ error, msgId }, 'failed to send scheduled phone request')
|
|
}
|
|
})
|
|
} else {
|
|
// Fallback to immediate request
|
|
const msgId = await requestPlaceholderResend(msgKey)
|
|
logger.debug(`sendRetryRequest: requested placeholder resend for message ${msgId}`)
|
|
}
|
|
}
|
|
|
|
if (!account) throw new Boom('Account not available', { statusCode: 401 })
|
|
const deviceIdentity = encodeSignedDeviceIdentity(account, true)
|
|
await authState.keys.transaction(async () => {
|
|
const receipt: BinaryNode = {
|
|
tag: 'receipt',
|
|
attrs: {
|
|
id: msgId,
|
|
type: 'retry',
|
|
to: node.attrs.from!
|
|
},
|
|
content: [
|
|
{
|
|
tag: 'retry',
|
|
attrs: {
|
|
count: retryCount.toString(),
|
|
id: node.attrs.id!,
|
|
t: node.attrs.t!,
|
|
v: '1',
|
|
// ADD ERROR FIELD
|
|
error: '0'
|
|
}
|
|
},
|
|
{
|
|
tag: 'registration',
|
|
attrs: {},
|
|
content: encodeBigEndian(authState.creds.registrationId)
|
|
}
|
|
]
|
|
}
|
|
|
|
if (node.attrs.recipient) {
|
|
receipt.attrs.recipient = node.attrs.recipient
|
|
}
|
|
|
|
if (node.attrs.participant) {
|
|
receipt.attrs.participant = node.attrs.participant
|
|
}
|
|
|
|
if (retryCount > 1 || forceIncludeKeys || shouldRecreateSession) {
|
|
const { update, preKeys } = await getNextPreKeys(authState, 1)
|
|
|
|
const [keyId] = Object.keys(preKeys)
|
|
const key = preKeys[+keyId!]
|
|
|
|
const content = receipt.content! as BinaryNode[]
|
|
content.push({
|
|
tag: 'keys',
|
|
attrs: {},
|
|
content: [
|
|
{ tag: 'type', attrs: {}, content: Buffer.from(KEY_BUNDLE_TYPE) },
|
|
{ tag: 'identity', attrs: {}, content: identityKey.public },
|
|
xmppPreKey(key!, +keyId!),
|
|
xmppSignedPreKey(signedPreKey),
|
|
{ tag: 'device-identity', attrs: {}, content: deviceIdentity }
|
|
]
|
|
})
|
|
|
|
ev.emit('creds.update', update)
|
|
}
|
|
|
|
await sendNode(receipt)
|
|
|
|
logger.info({ msgAttrs: node.attrs, retryCount }, 'sent retry receipt')
|
|
}, authState?.creds?.me?.id || 'sendRetryRequest')
|
|
}
|
|
|
|
const handleEncryptNotification = async (node: BinaryNode) => {
|
|
const from = node.attrs.from
|
|
if (from === S_WHATSAPP_NET) {
|
|
const countChild = getBinaryNodeChild(node, 'count')
|
|
const count = +countChild!.attrs.value!
|
|
const shouldUploadMorePreKeys = count < MIN_PREKEY_COUNT
|
|
|
|
logger.debug({ count, shouldUploadMorePreKeys }, 'recv pre-key count')
|
|
if (shouldUploadMorePreKeys) {
|
|
await uploadPreKeys()
|
|
}
|
|
} else {
|
|
const result = await handleIdentityChange(node, {
|
|
meId: authState.creds.me?.id,
|
|
meLid: authState.creds.me?.lid,
|
|
validateSession: signalRepository.validateSession,
|
|
assertSessions,
|
|
debounceCache: identityAssertDebounce,
|
|
logger
|
|
})
|
|
|
|
// When a session is refreshed (identity change), re-issue tctoken fire-and-forget
|
|
if (result.action === 'session_refreshed') {
|
|
const normalizedJid = jidNormalizedUser(from)
|
|
resolveTcTokenJid(normalizedJid, getLIDForPN)
|
|
.then(async tcJid => {
|
|
const tcData = await authState.keys.get('tctoken', [tcJid])
|
|
const entry = tcData[tcJid]
|
|
if (entry?.token?.length && !isTcTokenExpired(entry.timestamp)) {
|
|
const senderTs = unixTimestampSeconds()
|
|
logTcToken('reissue', { jid: normalizedJid, reason: 'session_refreshed' })
|
|
getPrivacyTokens([normalizedJid], senderTs).catch(err => {
|
|
logTcToken('reissue_fail', { jid: normalizedJid, error: err?.message })
|
|
})
|
|
}
|
|
})
|
|
.catch(() => {
|
|
/* ignore resolution errors */
|
|
})
|
|
}
|
|
|
|
if (result.action === 'no_identity_node') {
|
|
logger.info({ node }, 'unknown encrypt notification')
|
|
}
|
|
}
|
|
}
|
|
|
|
const handleGroupNotification = (fullNode: BinaryNode, child: BinaryNode, msg: Partial<WAMessage>) => {
|
|
// TODO: Support PN/LID (Here is only LID now)
|
|
|
|
const actingParticipantLid = fullNode.attrs.participant
|
|
const actingParticipantPn = fullNode.attrs.participant_pn
|
|
|
|
const affectedParticipantLid = getBinaryNodeChild(child, 'participant')?.attrs?.jid || actingParticipantLid!
|
|
const affectedParticipantPn = getBinaryNodeChild(child, 'participant')?.attrs?.phone_number || actingParticipantPn!
|
|
|
|
switch (child?.tag) {
|
|
case 'create':
|
|
const metadata = extractGroupMetadata(child)
|
|
|
|
msg.messageStubType = WAMessageStubType.GROUP_CREATE
|
|
msg.messageStubParameters = [metadata.subject]
|
|
msg.key = { participant: metadata.owner, participantAlt: metadata.ownerPn }
|
|
|
|
ev.emit('chats.upsert', [
|
|
{
|
|
id: metadata.id,
|
|
name: metadata.subject,
|
|
conversationTimestamp: metadata.creation
|
|
}
|
|
])
|
|
ev.emit('groups.upsert', [
|
|
{
|
|
...metadata,
|
|
author: actingParticipantLid,
|
|
authorPn: actingParticipantPn
|
|
}
|
|
])
|
|
break
|
|
case 'ephemeral':
|
|
case 'not_ephemeral':
|
|
msg.message = {
|
|
protocolMessage: {
|
|
type: proto.Message.ProtocolMessage.Type.EPHEMERAL_SETTING,
|
|
ephemeralExpiration: +(child.attrs.expiration || 0)
|
|
}
|
|
}
|
|
break
|
|
case 'modify':
|
|
const oldNumber = getBinaryNodeChildren(child, 'participant').map(p => p.attrs.jid!)
|
|
msg.messageStubParameters = oldNumber || []
|
|
msg.messageStubType = WAMessageStubType.GROUP_PARTICIPANT_CHANGE_NUMBER
|
|
break
|
|
case 'promote':
|
|
case 'demote':
|
|
case 'remove':
|
|
case 'add':
|
|
case 'leave':
|
|
const stubType = `GROUP_PARTICIPANT_${child.tag.toUpperCase()}`
|
|
msg.messageStubType = WAMessageStubType[stubType as keyof typeof WAMessageStubType]
|
|
|
|
const participants = getBinaryNodeChildren(child, 'participant').map(({ attrs }) => {
|
|
// TODO: Store LID MAPPINGS
|
|
return {
|
|
id: attrs.jid!,
|
|
phoneNumber: isLidUser(attrs.jid) && isPnUser(attrs.phone_number) ? attrs.phone_number : undefined,
|
|
lid: isPnUser(attrs.jid) && isLidUser(attrs.lid) ? attrs.lid : undefined,
|
|
admin: (attrs.type || null) as GroupParticipant['admin']
|
|
}
|
|
})
|
|
|
|
if (
|
|
participants.length === 1 &&
|
|
// if recv. "remove" message and sender removed themselves
|
|
// mark as left
|
|
(areJidsSameUser(participants[0]!.id, actingParticipantLid) ||
|
|
areJidsSameUser(participants[0]!.id, actingParticipantPn)) &&
|
|
child.tag === 'remove'
|
|
) {
|
|
msg.messageStubType = WAMessageStubType.GROUP_PARTICIPANT_LEAVE
|
|
}
|
|
|
|
msg.messageStubParameters = participants.map(a => JSON.stringify(a))
|
|
break
|
|
case 'subject':
|
|
msg.messageStubType = WAMessageStubType.GROUP_CHANGE_SUBJECT
|
|
msg.messageStubParameters = [child.attrs.subject!]
|
|
break
|
|
case 'description':
|
|
const description = getBinaryNodeChild(child, 'body')?.content?.toString()
|
|
msg.messageStubType = WAMessageStubType.GROUP_CHANGE_DESCRIPTION
|
|
msg.messageStubParameters = description ? [description] : undefined
|
|
break
|
|
case 'announcement':
|
|
case 'not_announcement':
|
|
msg.messageStubType = WAMessageStubType.GROUP_CHANGE_ANNOUNCE
|
|
msg.messageStubParameters = [child.tag === 'announcement' ? 'on' : 'off']
|
|
break
|
|
case 'locked':
|
|
case 'unlocked':
|
|
msg.messageStubType = WAMessageStubType.GROUP_CHANGE_RESTRICT
|
|
msg.messageStubParameters = [child.tag === 'locked' ? 'on' : 'off']
|
|
break
|
|
case 'invite':
|
|
msg.messageStubType = WAMessageStubType.GROUP_CHANGE_INVITE_LINK
|
|
msg.messageStubParameters = [child.attrs.code!]
|
|
break
|
|
case 'member_add_mode':
|
|
const addMode = child.content
|
|
if (addMode) {
|
|
msg.messageStubType = WAMessageStubType.GROUP_MEMBER_ADD_MODE
|
|
msg.messageStubParameters = [addMode.toString()]
|
|
}
|
|
|
|
break
|
|
case 'membership_approval_mode':
|
|
const approvalMode = getBinaryNodeChild(child, 'group_join')
|
|
if (approvalMode) {
|
|
msg.messageStubType = WAMessageStubType.GROUP_MEMBERSHIP_JOIN_APPROVAL_MODE
|
|
msg.messageStubParameters = [approvalMode.attrs.state!]
|
|
}
|
|
|
|
break
|
|
case 'created_membership_requests':
|
|
msg.messageStubType = WAMessageStubType.GROUP_MEMBERSHIP_JOIN_APPROVAL_REQUEST_NON_ADMIN_ADD
|
|
msg.messageStubParameters = [
|
|
JSON.stringify({ lid: affectedParticipantLid, pn: affectedParticipantPn }),
|
|
'created',
|
|
child.attrs.request_method!
|
|
]
|
|
break
|
|
case 'revoked_membership_requests':
|
|
const isDenied = areJidsSameUser(affectedParticipantLid, actingParticipantLid)
|
|
// TODO: LIDMAPPING SUPPORT
|
|
msg.messageStubType = WAMessageStubType.GROUP_MEMBERSHIP_JOIN_APPROVAL_REQUEST_NON_ADMIN_ADD
|
|
msg.messageStubParameters = [
|
|
JSON.stringify({ lid: affectedParticipantLid, pn: affectedParticipantPn }),
|
|
isDenied ? 'revoked' : 'rejected'
|
|
]
|
|
break
|
|
}
|
|
}
|
|
|
|
const processNotification = async (node: BinaryNode) => {
|
|
const result: Partial<WAMessage> = {}
|
|
const [child] = getAllBinaryNodeChildren(node)
|
|
const nodeType = node.attrs.type
|
|
const from = jidNormalizedUser(node.attrs.from)
|
|
|
|
switch (nodeType) {
|
|
case 'newsletter':
|
|
await handleNewsletterNotification(node)
|
|
break
|
|
case 'mex':
|
|
await handleMexNewsletterNotification(node)
|
|
break
|
|
case 'w:gp2':
|
|
// TODO: HANDLE PARTICIPANT_PN
|
|
handleGroupNotification(node, child!, result)
|
|
break
|
|
case 'mediaretry':
|
|
const event = decodeMediaRetryNode(node)
|
|
ev.emit('messages.media-update', [event])
|
|
break
|
|
case 'encrypt':
|
|
await handleEncryptNotification(node)
|
|
break
|
|
case 'devices':
|
|
const devices = getBinaryNodeChildren(child, 'device')
|
|
if (
|
|
areJidsSameUser(child!.attrs.jid, authState.creds.me!.id) ||
|
|
areJidsSameUser(child!.attrs.lid, authState.creds.me!.lid)
|
|
) {
|
|
const deviceData = devices.map(d => ({ id: d.attrs.jid, lid: d.attrs.lid }))
|
|
logger.info({ deviceData }, 'my own devices changed')
|
|
}
|
|
|
|
//TODO: drop a new event, add hashes
|
|
|
|
break
|
|
case 'server_sync':
|
|
const update = getBinaryNodeChild(node, 'collection')
|
|
if (update) {
|
|
const name = update.attrs.name as WAPatchName
|
|
await resyncAppState([name], false)
|
|
}
|
|
|
|
break
|
|
case 'picture':
|
|
const setPicture = getBinaryNodeChild(node, 'set')
|
|
const delPicture = getBinaryNodeChild(node, 'delete')
|
|
|
|
// TODO: WAJIDHASH stuff proper support inhouse
|
|
ev.emit('contacts.update', [
|
|
{
|
|
id: jidNormalizedUser(node?.attrs?.from) || (setPicture || delPicture)?.attrs?.hash || '',
|
|
imgUrl: setPicture ? 'changed' : 'removed'
|
|
}
|
|
])
|
|
|
|
if (isJidGroup(from)) {
|
|
const node = setPicture || delPicture
|
|
result.messageStubType = WAMessageStubType.GROUP_CHANGE_ICON
|
|
|
|
if (setPicture) {
|
|
result.messageStubParameters = [setPicture.attrs.id!]
|
|
}
|
|
|
|
result.participant = node?.attrs.author
|
|
result.key = {
|
|
...(result.key || {}),
|
|
participant: setPicture?.attrs.author
|
|
}
|
|
}
|
|
|
|
break
|
|
case 'account_sync':
|
|
if (child!.tag === 'disappearing_mode') {
|
|
const newDuration = +child!.attrs.duration!
|
|
const timestamp = +child!.attrs.t!
|
|
|
|
logger.info({ newDuration }, 'updated account disappearing mode')
|
|
|
|
ev.emit('creds.update', {
|
|
accountSettings: {
|
|
...authState.creds.accountSettings,
|
|
defaultDisappearingMode: {
|
|
ephemeralExpiration: newDuration,
|
|
ephemeralSettingTimestamp: timestamp
|
|
}
|
|
}
|
|
})
|
|
} else if (child!.tag === 'blocklist') {
|
|
const blocklists = getBinaryNodeChildren(child, 'item')
|
|
|
|
for (const { attrs } of blocklists) {
|
|
const blocklist = [attrs.jid!]
|
|
const type = attrs.action === 'block' ? 'add' : 'remove'
|
|
ev.emit('blocklist.update', { blocklist, type })
|
|
}
|
|
}
|
|
|
|
break
|
|
case 'link_code_companion_reg':
|
|
const linkCodeCompanionReg = getBinaryNodeChild(node, 'link_code_companion_reg')
|
|
const ref = toRequiredBuffer(getBinaryNodeChildBuffer(linkCodeCompanionReg, 'link_code_pairing_ref'))
|
|
const primaryIdentityPublicKey = toRequiredBuffer(
|
|
getBinaryNodeChildBuffer(linkCodeCompanionReg, 'primary_identity_pub')
|
|
)
|
|
const primaryEphemeralPublicKeyWrapped = toRequiredBuffer(
|
|
getBinaryNodeChildBuffer(linkCodeCompanionReg, 'link_code_pairing_wrapped_primary_ephemeral_pub')
|
|
)
|
|
const codePairingPublicKey = await decipherLinkPublicKey(primaryEphemeralPublicKeyWrapped)
|
|
const companionSharedKey = Curve.sharedKey(
|
|
authState.creds.pairingEphemeralKeyPair.private,
|
|
codePairingPublicKey
|
|
)
|
|
const random = randomBytes(32)
|
|
const linkCodeSalt = randomBytes(32)
|
|
const linkCodePairingExpanded = hkdf(companionSharedKey, 32, {
|
|
salt: linkCodeSalt,
|
|
info: 'link_code_pairing_key_bundle_encryption_key'
|
|
})
|
|
const encryptPayload = Buffer.concat([
|
|
Buffer.from(authState.creds.signedIdentityKey.public),
|
|
primaryIdentityPublicKey,
|
|
random
|
|
])
|
|
const encryptIv = randomBytes(12)
|
|
const encrypted = aesEncryptGCM(encryptPayload, linkCodePairingExpanded, encryptIv, Buffer.alloc(0))
|
|
const encryptedPayload = Buffer.concat([linkCodeSalt, encryptIv, encrypted])
|
|
const identitySharedKey = Curve.sharedKey(authState.creds.signedIdentityKey.private, primaryIdentityPublicKey)
|
|
const identityPayload = Buffer.concat([companionSharedKey, identitySharedKey, random])
|
|
authState.creds.advSecretKey = Buffer.from(hkdf(identityPayload, 32, { info: 'adv_secret' })).toString('base64')
|
|
await query({
|
|
tag: 'iq',
|
|
attrs: {
|
|
to: S_WHATSAPP_NET,
|
|
type: 'set',
|
|
id: sock.generateMessageTag(),
|
|
xmlns: 'md'
|
|
},
|
|
content: [
|
|
{
|
|
tag: 'link_code_companion_reg',
|
|
attrs: {
|
|
jid: authState.creds.me!.id,
|
|
stage: 'companion_finish'
|
|
},
|
|
content: [
|
|
{
|
|
tag: 'link_code_pairing_wrapped_key_bundle',
|
|
attrs: {},
|
|
content: encryptedPayload
|
|
},
|
|
{
|
|
tag: 'companion_identity_public',
|
|
attrs: {},
|
|
content: authState.creds.signedIdentityKey.public
|
|
},
|
|
{
|
|
tag: 'link_code_pairing_ref',
|
|
attrs: {},
|
|
content: ref
|
|
}
|
|
]
|
|
}
|
|
]
|
|
})
|
|
authState.creds.registered = true
|
|
ev.emit('creds.update', authState.creds)
|
|
break
|
|
case 'privacy_token':
|
|
await handlePrivacyTokenNotification(node)
|
|
break
|
|
}
|
|
|
|
if (Object.keys(result).length) {
|
|
return result
|
|
}
|
|
}
|
|
|
|
const handlePrivacyTokenNotification = async (node: BinaryNode) => {
|
|
const tokensNode = getBinaryNodeChild(node, 'tokens')
|
|
const from = jidNormalizedUser(node.attrs.from)
|
|
|
|
if (!tokensNode) return
|
|
|
|
const tokenNodes = getBinaryNodeChildren(tokensNode, 'token')
|
|
|
|
for (const tokenNode of tokenNodes) {
|
|
const { attrs, content } = tokenNode
|
|
const type = attrs.type
|
|
const timestamp = attrs.t
|
|
|
|
if (type === 'trusted_contact' && content instanceof Uint8Array) {
|
|
// Resolve to LID for consistent storage key
|
|
const senderLid = attrs.sender_lid ? jidNormalizedUser(attrs.sender_lid) : undefined
|
|
const storageJid = senderLid || (await resolveTcTokenJid(from, getLIDForPN))
|
|
|
|
// Timestamp monotonicity guard — only store if incoming >= existing
|
|
const existingData = await authState.keys.get('tctoken', [storageJid])
|
|
const existing = existingData[storageJid]
|
|
const existingTs = existing?.timestamp ? Number(existing.timestamp) : 0
|
|
const incomingTs = timestamp ? Number(timestamp) : 0
|
|
if (existingTs > 0 && incomingTs > 0 && existingTs > incomingTs) {
|
|
continue
|
|
}
|
|
|
|
// Don't store timestamp-less tokens — they expire immediately and would
|
|
// corrupt a valid existing entry if one is already present
|
|
if (!incomingTs) {
|
|
continue
|
|
}
|
|
|
|
await authState.keys.set({
|
|
tctoken: {
|
|
[storageJid]: {
|
|
...existing,
|
|
token: Buffer.from(content),
|
|
timestamp
|
|
}
|
|
}
|
|
})
|
|
|
|
logTcToken('stored', { jid: storageJid, from })
|
|
|
|
// Track JID for cross-session pruning
|
|
tcTokenKnownJids.add(storageJid)
|
|
scheduleTcTokenIndexSave()
|
|
}
|
|
}
|
|
}
|
|
|
|
async function decipherLinkPublicKey(data: Uint8Array | Buffer) {
|
|
const buffer = toRequiredBuffer(data)
|
|
const salt = buffer.slice(0, 32)
|
|
const secretKey = await derivePairingCodeKey(authState.creds.pairingCode!, salt)
|
|
const iv = buffer.slice(32, 48)
|
|
const payload = buffer.slice(48, 80)
|
|
return aesDecryptCTR(payload, secretKey, iv)
|
|
}
|
|
|
|
function toRequiredBuffer(data: Uint8Array | Buffer | undefined) {
|
|
if (data === undefined) {
|
|
throw new Boom('Invalid buffer', { statusCode: 400 })
|
|
}
|
|
|
|
return data instanceof Buffer ? data : Buffer.from(data)
|
|
}
|
|
|
|
const willSendMessageAgain = async (id: string, participant: string) => {
|
|
const key = `${id}:${participant}`
|
|
const retryCount = (await msgRetryCache.get<number>(key)) || 0
|
|
return retryCount < maxMsgRetryCount
|
|
}
|
|
|
|
const updateSendMessageAgainCount = async (id: string, participant: string) => {
|
|
const key = `${id}:${participant}`
|
|
const newValue = ((await msgRetryCache.get<number>(key)) || 0) + 1
|
|
await msgRetryCache.set(key, newValue)
|
|
}
|
|
|
|
const sendMessagesAgain = async (key: WAMessageKey, ids: string[], retryNode: BinaryNode) => {
|
|
const remoteJid = key.remoteJid!
|
|
const participant = key.participant || remoteJid
|
|
|
|
const retryCount = +retryNode.attrs.count! || 1
|
|
|
|
// Try to get messages from cache first, then fallback to getMessage
|
|
const msgs: (proto.IMessage | undefined)[] = []
|
|
for (const id of ids) {
|
|
let msg: proto.IMessage | undefined
|
|
|
|
// Try to get from retry cache first if enabled
|
|
if (messageRetryManager) {
|
|
const cachedMsg = messageRetryManager.getRecentMessage(remoteJid, id)
|
|
if (cachedMsg) {
|
|
msg = cachedMsg.message
|
|
logger.debug({ jid: remoteJid, id }, 'found message in retry cache')
|
|
|
|
// Mark retry as successful since we found the message
|
|
messageRetryManager.markRetrySuccess(id)
|
|
}
|
|
}
|
|
|
|
// Fallback to getMessage if not found in cache
|
|
if (!msg) {
|
|
msg = await getMessage({ ...key, id })
|
|
if (msg) {
|
|
logger.debug({ jid: remoteJid, id }, 'found message via getMessage')
|
|
// Also mark as successful if found via getMessage
|
|
if (messageRetryManager) {
|
|
messageRetryManager.markRetrySuccess(id)
|
|
}
|
|
}
|
|
}
|
|
|
|
msgs.push(msg)
|
|
}
|
|
|
|
// if it's the primary jid sending the request
|
|
// just re-send the message to everyone
|
|
// prevents the first message decryption failure
|
|
const sendToAll = !jidDecode(participant)?.device
|
|
|
|
// Check if we should recreate session for this retry
|
|
let shouldRecreateSession = false
|
|
let recreateReason = ''
|
|
|
|
if (enableAutoSessionRecreation && messageRetryManager && retryCount > 1) {
|
|
try {
|
|
const sessionId = signalRepository.jidToSignalProtocolAddress(participant)
|
|
|
|
const hasSession = await signalRepository.validateSession(participant)
|
|
|
|
// Extract error code from retry node if present (for MAC error detection)
|
|
const errorAttr = retryNode?.attrs?.error as string | undefined
|
|
const errorCode = messageRetryManager.parseRetryErrorCode(errorAttr)
|
|
|
|
const result = messageRetryManager.shouldRecreateSession(participant, hasSession.exists, errorCode)
|
|
shouldRecreateSession = result.recreate
|
|
recreateReason = result.reason
|
|
|
|
if (shouldRecreateSession) {
|
|
logger.debug(
|
|
{ participant, retryCount, reason: recreateReason, errorCode },
|
|
'recreating session for outgoing retry'
|
|
)
|
|
// CRITICAL: Use same transaction key as encrypt/decrypt operations to prevent race
|
|
// Using meId ensures this delete serializes with sendMessage() and other session operations
|
|
await authState.keys.transaction(async () => {
|
|
await authState.keys.set({ session: { [sessionId]: null } })
|
|
}, authState.creds.me?.id || 'session-operation')
|
|
}
|
|
} catch (error) {
|
|
logger.warn({ error, participant }, 'failed to check session recreation for outgoing retry')
|
|
}
|
|
}
|
|
|
|
await assertSessions([participant], true)
|
|
|
|
if (isJidGroup(remoteJid)) {
|
|
await authState.keys.set({ 'sender-key-memory': { [remoteJid]: null } })
|
|
}
|
|
|
|
logger.debug({ participant, sendToAll, shouldRecreateSession, recreateReason }, 'forced new session for retry recp')
|
|
|
|
for (const [i, msg] of msgs.entries()) {
|
|
if (!ids[i]) continue
|
|
|
|
if (msg && (await willSendMessageAgain(ids[i], participant))) {
|
|
await updateSendMessageAgainCount(ids[i], participant)
|
|
const msgRelayOpts: MessageRelayOptions = { messageId: ids[i] }
|
|
|
|
if (sendToAll) {
|
|
msgRelayOpts.useUserDevicesCache = false
|
|
} else {
|
|
msgRelayOpts.participant = {
|
|
jid: participant,
|
|
count: +retryNode.attrs.count!
|
|
}
|
|
}
|
|
|
|
await relayMessage(key.remoteJid!, msg, msgRelayOpts)
|
|
} else {
|
|
logger.debug({ jid: key.remoteJid, id: ids[i] }, 'recv retry request, but message not available')
|
|
}
|
|
}
|
|
}
|
|
|
|
const handleReceipt = async (node: BinaryNode) => {
|
|
const { attrs, content } = node
|
|
const isLid = attrs.from!.includes('lid')
|
|
const isNodeFromMe = areJidsSameUser(
|
|
attrs.participant || attrs.from,
|
|
isLid ? authState.creds.me?.lid : authState.creds.me?.id
|
|
)
|
|
const remoteJid = !isNodeFromMe || isJidGroup(attrs.from) ? attrs.from : attrs.recipient
|
|
const fromMe = !attrs.recipient || ((attrs.type === 'retry' || attrs.type === 'sender') && isNodeFromMe)
|
|
|
|
const key: proto.IMessageKey = {
|
|
remoteJid,
|
|
id: '',
|
|
fromMe,
|
|
participant: attrs.participant
|
|
}
|
|
|
|
if (shouldIgnoreJid(remoteJid!) && remoteJid !== S_WHATSAPP_NET) {
|
|
logger.trace({ remoteJid }, 'ignoring receipt from jid')
|
|
await sendMessageAck(node)
|
|
return
|
|
}
|
|
|
|
const ids = [attrs.id!]
|
|
if (Array.isArray(content)) {
|
|
const items = getBinaryNodeChildren(content[0], 'item')
|
|
ids.push(...items.map(i => i.attrs.id!))
|
|
}
|
|
|
|
try {
|
|
await Promise.all([
|
|
receiptMutex.mutex(jidNormalizedUser(remoteJid) || 'unknown', async () => {
|
|
const status = getStatusFromReceiptType(attrs.type)
|
|
if (
|
|
typeof status !== 'undefined' &&
|
|
// basically, we only want to know when a message from us has been delivered to/read by the other person
|
|
// or another device of ours has read some messages
|
|
(status >= proto.WebMessageInfo.Status.SERVER_ACK || !isNodeFromMe)
|
|
) {
|
|
if (isJidGroup(remoteJid) || isJidStatusBroadcast(remoteJid!)) {
|
|
if (attrs.participant) {
|
|
const updateKey: keyof MessageUserReceipt =
|
|
status === proto.WebMessageInfo.Status.DELIVERY_ACK ? 'receiptTimestamp' : 'readTimestamp'
|
|
ev.emit(
|
|
'message-receipt.update',
|
|
ids.map(id => ({
|
|
key: { ...key, id },
|
|
receipt: {
|
|
userJid: jidNormalizedUser(attrs.participant),
|
|
[updateKey]: +attrs.t!
|
|
}
|
|
}))
|
|
)
|
|
}
|
|
} else {
|
|
ev.emit(
|
|
'messages.update',
|
|
ids.map(id => ({
|
|
key: { ...key, id },
|
|
update: { status, messageTimestamp: toNumber(+(attrs.t ?? 0)) }
|
|
}))
|
|
)
|
|
}
|
|
}
|
|
|
|
if (attrs.type === 'retry') {
|
|
// correctly set who is asking for the retry
|
|
key.participant = key.participant || attrs.from
|
|
const retryNode = getBinaryNodeChild(node, 'retry')
|
|
if (ids[0] && key.participant && (await willSendMessageAgain(ids[0], key.participant))) {
|
|
if (key.fromMe) {
|
|
try {
|
|
await updateSendMessageAgainCount(ids[0], key.participant)
|
|
logger.debug({ attrs, key }, 'recv retry request')
|
|
await sendMessagesAgain(key, ids, retryNode!)
|
|
} catch (error: unknown) {
|
|
logger.error(
|
|
{ key, ids, trace: error instanceof Error ? error.stack : 'Unknown error' },
|
|
'error in sending message again'
|
|
)
|
|
}
|
|
} else {
|
|
logger.info({ attrs, key }, 'recv retry for not fromMe message')
|
|
}
|
|
} else {
|
|
logger.info({ attrs, key }, 'will not send message again, as sent too many times')
|
|
}
|
|
}
|
|
})
|
|
])
|
|
} finally {
|
|
await sendMessageAck(node)
|
|
}
|
|
}
|
|
|
|
const handleNotification = async (node: BinaryNode) => {
|
|
const remoteJid = node.attrs.from
|
|
if (shouldIgnoreJid(remoteJid!) && remoteJid !== S_WHATSAPP_NET) {
|
|
logger.trace({ remoteJid }, 'ignored notification')
|
|
await sendMessageAck(node)
|
|
return
|
|
}
|
|
|
|
try {
|
|
await Promise.all([
|
|
notificationMutex.mutex(jidNormalizedUser(remoteJid) || 'unknown', async () => {
|
|
const msg = await processNotification(node)
|
|
if (msg) {
|
|
const fromMe = areJidsSameUser(node.attrs.participant || remoteJid, authState.creds.me!.id)
|
|
const { senderAlt: participantAlt, addressingMode } = extractAddressingContext(node)
|
|
msg.key = {
|
|
remoteJid,
|
|
fromMe,
|
|
participant: node.attrs.participant,
|
|
participantAlt,
|
|
addressingMode,
|
|
id: node.attrs.id,
|
|
...(msg.key || {})
|
|
}
|
|
msg.participant ??= node.attrs.participant
|
|
msg.messageTimestamp = +node.attrs.t!
|
|
|
|
const fullMsg = proto.WebMessageInfo.fromObject(msg) as WAMessage
|
|
await upsertMessage(fullMsg, 'append')
|
|
}
|
|
})
|
|
])
|
|
} finally {
|
|
await sendMessageAck(node)
|
|
}
|
|
}
|
|
|
|
const handleMessage = async (node: BinaryNode) => {
|
|
if (shouldIgnoreJid(node.attrs.from!) && node.attrs.from !== S_WHATSAPP_NET) {
|
|
logger.trace({ from: node.attrs.from }, 'ignored message')
|
|
// Send a clean ACK (no error code) so the server considers the
|
|
// message delivered. Using error 500 (UnhandledError) previously
|
|
// caused the server to retry delivery, generating duplicate traffic.
|
|
await sendMessageAck(node)
|
|
return
|
|
}
|
|
|
|
const encNode = getBinaryNodeChild(node, 'enc')
|
|
// TODO: temporary fix for crashes and issues resulting of failed msmsg decryption
|
|
if (encNode?.attrs.type === 'msmsg') {
|
|
logger.debug({ key: node.attrs.key }, 'ignored msmsg')
|
|
await sendMessageAck(node, NACK_REASONS.MissingMessageSecret)
|
|
return
|
|
}
|
|
|
|
const {
|
|
fullMessage: msg,
|
|
category,
|
|
author,
|
|
decrypt
|
|
} = decryptMessageNode(
|
|
node,
|
|
authState.creds.me!.id,
|
|
authState.creds.me!.lid || '',
|
|
signalRepository,
|
|
logger,
|
|
autoCleanCorrupted
|
|
)
|
|
|
|
const alt = msg.key.participantAlt || msg.key.remoteJidAlt
|
|
// Handle LID/PN mappings with hybrid approach:
|
|
// - Store mapping operation runs in background (non-critical for decrypt)
|
|
// - Session migration MUST complete before decrypt() to avoid "No session record" errors
|
|
// This addresses Codex/Copilot review concerns about race conditions with decrypt()
|
|
if (!!alt) {
|
|
const altServer = jidDecode(alt)?.server
|
|
const primaryJid = msg.key.participant || msg.key.remoteJid!
|
|
|
|
if (altServer === 'lid') {
|
|
// Check if mapping already exists to avoid unnecessary storage operations
|
|
const existingMapping = await signalRepository.lidMapping.getPNForLID(alt)
|
|
if (!existingMapping) {
|
|
// Store mapping in background (non-critical, doesn't block decrypt)
|
|
signalRepository.lidMapping
|
|
.storeLIDPNMappings([{ lid: alt, pn: primaryJid }])
|
|
.catch(error => logger.warn({ error, alt, primaryJid }, 'Background LID mapping storage failed'))
|
|
}
|
|
|
|
// CRITICAL: ALWAYS migrate session, even if mapping exists
|
|
// Other code paths (e.g., USync device lookup in messages-send.ts:310-319)
|
|
// may create mappings via storeLIDPNMappings() without calling migrateSession()
|
|
// This leaves sessions under PN format while decrypt() expects LID format
|
|
// Skipping migration based on mapping existence causes "No session record" errors
|
|
await signalRepository.migrateSession(primaryJid, alt)
|
|
} else {
|
|
// Check if reverse mapping exists
|
|
const existingMapping = await signalRepository.lidMapping.getLIDForPN(alt)
|
|
if (!existingMapping) {
|
|
// Store mapping in background (non-critical)
|
|
signalRepository.lidMapping
|
|
.storeLIDPNMappings([{ lid: primaryJid, pn: alt }])
|
|
.catch(error => logger.warn({ error, alt, primaryJid }, 'Background LID mapping storage failed'))
|
|
}
|
|
|
|
// CRITICAL: ALWAYS migrate session, even if mapping exists
|
|
// Same reasoning as above - mapping existence doesn't guarantee session migration
|
|
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'
|
|
)
|
|
}
|
|
|
|
// CRITICAL: Normalize JIDs BEFORE acquiring mutex to ensure messages from the same
|
|
// chat (arriving with different JID formats - LID vs PN) use the SAME mutex key.
|
|
// This prevents parallel processing of messages from the same conversation which
|
|
// would break message ordering guarantees.
|
|
// Addresses Copilot/Codex PR #75 critical review: JID normalization vulnerability
|
|
await normalizeMessageJids(msg, signalRepository, logger)
|
|
|
|
try {
|
|
// Use KeyedMutex with NORMALIZED remoteJid for parallel processing across different chats
|
|
// while maintaining sequential order within the same chat
|
|
// Fallback chain: remoteJid (normalized) > msg.key.id (unique) > 'unknown' (serializes all)
|
|
let mutexKey = msg.key.remoteJid
|
|
if (!mutexKey) {
|
|
logger.warn(
|
|
{ msgId: msg.key.id, fromMe: msg.key.fromMe },
|
|
'Missing remoteJid after normalization, using msg.key.id as fallback'
|
|
)
|
|
mutexKey = msg.key.id || 'unknown'
|
|
}
|
|
|
|
await messageMutex.mutex(mutexKey, async () => {
|
|
await decrypt()
|
|
// message failed to decrypt
|
|
if (msg.messageStubType === proto.WebMessageInfo.StubType.CIPHERTEXT && msg.category !== 'peer') {
|
|
// Handle "Missing keys" - standard decryption failure
|
|
// Return NACK with parsing error to signal the issue
|
|
if (msg?.messageStubParameters?.[0] === MISSING_KEYS_ERROR_TEXT) {
|
|
return sendMessageAck(node, NACK_REASONS.ParsingError)
|
|
}
|
|
|
|
// Handle "Message absent from node" - likely a CTWA (Click-to-WhatsApp) ads message
|
|
// These messages are only encrypted for the primary phone, not linked devices
|
|
// We need to request the message content from the phone via PDO (Peer Data Operation)
|
|
if (msg.messageStubParameters?.[0] === NO_MESSAGE_FOUND_ERROR_TEXT) {
|
|
// Skip unavailable fanout types - these messages will never have content available
|
|
// These are system messages that cannot be decrypted or retrieved
|
|
const messageType = msg.messageStubParameters?.[2]
|
|
if (
|
|
messageType === 'bot_unavailable_fanout' ||
|
|
messageType === 'hosted_unavailable_fanout' ||
|
|
messageType === 'view_once_unavailable_fanout'
|
|
) {
|
|
logger.debug(
|
|
{ msgId: msg.key?.id, messageType },
|
|
'CTWA: Skipping placeholder resend for unavailable fanout type'
|
|
)
|
|
metrics.ctwaRecoveryFailures.inc({ reason: 'unavailable_fanout' })
|
|
return sendMessageAck(node)
|
|
}
|
|
|
|
// Skip old messages - don't request resend for messages older than 7 days
|
|
const messageAge = unixTimestampSeconds() - toNumber(msg.messageTimestamp)
|
|
|
|
if (messageAge > PLACEHOLDER_MAX_AGE_SECONDS) {
|
|
logger.debug(
|
|
{ msgId: msg.key?.id, messageAge, maxAge: PLACEHOLDER_MAX_AGE_SECONDS },
|
|
'CTWA: Skipping placeholder resend for old message'
|
|
)
|
|
metrics.ctwaRecoveryFailures.inc({ reason: 'message_too_old' })
|
|
return sendMessageAck(node)
|
|
}
|
|
|
|
if (enableCTWARecovery && msg.key) {
|
|
const startTime = Date.now()
|
|
const msgId = msg.key.id!
|
|
const msgKey = msg.key
|
|
|
|
// Prepare metadata to preserve original message details
|
|
// The phone may not send all metadata in PDO response (e.g., pushName, participantAlt)
|
|
// Caching these ensures we don't lose critical information like sender name and LID mappings
|
|
const msgData: PlaceholderMessageData = {
|
|
key: { ...msgKey },
|
|
pushName: msg.pushName,
|
|
messageTimestamp: msg.messageTimestamp,
|
|
participant: msg.key.participant,
|
|
participantAlt: msg.key.participantAlt
|
|
}
|
|
|
|
logger.info(
|
|
{ msgId, remoteJid: msgKey.remoteJid, messageAge },
|
|
'CTWA: Message absent from node detected, scheduling placeholder resend from phone'
|
|
)
|
|
|
|
// Use messageRetryManager to schedule the phone request with delay
|
|
// This aligns with the upstream philosophy: centralize phone requests in the manager
|
|
// Benefits: 3s delay (avoids spam), auto-cancellation if message arrives
|
|
if (messageRetryManager) {
|
|
metrics.ctwaRecoveryRequests.inc({ status: 'scheduled' })
|
|
|
|
messageRetryManager.schedulePhoneRequest(msgId, async () => {
|
|
try {
|
|
const requestId = await requestPlaceholderResend(msgKey, msgData)
|
|
if (requestId && requestId !== 'RESOLVED') {
|
|
logger.debug({ msgId, requestId }, 'CTWA: Placeholder resend request sent successfully')
|
|
metrics.ctwaRecoveryRequests.inc({ status: 'sent' })
|
|
// Note: The actual message will be emitted via 'messages.upsert'
|
|
// when the PEER_DATA_OPERATION_REQUEST_RESPONSE_MESSAGE is processed
|
|
// in the PDO response handler in src/Utils/process-message.ts
|
|
} else if (requestId === 'RESOLVED') {
|
|
// Message was received while we were waiting
|
|
logger.debug({ msgId }, 'CTWA: Message received during resend delay')
|
|
metrics.ctwaMessagesRecovered.inc()
|
|
metrics.ctwaRecoveryLatency.observe(Date.now() - startTime)
|
|
} else {
|
|
// Already requested (duplicate request prevented by cache)
|
|
logger.debug({ msgId }, 'CTWA: Resend already requested, skipping duplicate')
|
|
}
|
|
} catch (error) {
|
|
logger.warn({ error, msgId }, 'CTWA: Failed to request placeholder resend')
|
|
metrics.ctwaRecoveryFailures.inc({ reason: 'request_failed' })
|
|
}
|
|
})
|
|
} else {
|
|
// Fallback: direct call if messageRetryManager is not available
|
|
metrics.ctwaRecoveryRequests.inc({ status: 'requested' })
|
|
|
|
try {
|
|
const requestId = await requestPlaceholderResend(msgKey, msgData)
|
|
if (requestId && requestId !== 'RESOLVED') {
|
|
logger.debug({ msgId, requestId }, 'CTWA: Placeholder resend request sent successfully (direct)')
|
|
} else if (requestId === 'RESOLVED') {
|
|
// Message arrived during the internal 2s delay in requestPlaceholderResend
|
|
logger.debug({ msgId }, 'CTWA: Message received before direct resend request completed')
|
|
metrics.ctwaMessagesRecovered.inc()
|
|
metrics.ctwaRecoveryLatency.observe(Date.now() - startTime)
|
|
} else {
|
|
// Already requested (duplicate request prevented by cache)
|
|
logger.debug({ msgId }, 'CTWA: Resend already requested, skipping duplicate (direct)')
|
|
}
|
|
} catch (error) {
|
|
logger.warn({ error, msgId }, 'CTWA: Failed to request placeholder resend')
|
|
metrics.ctwaRecoveryFailures.inc({ reason: 'request_failed' })
|
|
}
|
|
}
|
|
} else {
|
|
logger.debug(
|
|
{ msgId: msg.key?.id, enableCTWARecovery },
|
|
'CTWA recovery disabled or missing key, skipping placeholder resend'
|
|
)
|
|
}
|
|
|
|
return sendMessageAck(node)
|
|
}
|
|
|
|
// Skip retry for expired status messages (>24h old)
|
|
if (isJidStatusBroadcast(msg.key.remoteJid!)) {
|
|
const messageAge = unixTimestampSeconds() - toNumber(msg.messageTimestamp)
|
|
if (messageAge > STATUS_EXPIRY_SECONDS) {
|
|
logger.debug(
|
|
{ msgId: msg.key.id, messageAge, remoteJid: msg.key.remoteJid },
|
|
'skipping retry for expired status message'
|
|
)
|
|
return sendMessageAck(node)
|
|
}
|
|
}
|
|
|
|
const errorMessage = msg?.messageStubParameters?.[0] || ''
|
|
const isPreKeyError = errorMessage.includes('PreKey')
|
|
|
|
logger.debug(`[handleMessage] Attempting retry request for failed decryption`)
|
|
|
|
// Handle both pre-key and normal retries in single mutex
|
|
await retryMutex.mutex(async () => {
|
|
try {
|
|
if (!ws.isOpen) {
|
|
logger.debug({ node }, 'Connection closed, skipping retry')
|
|
return
|
|
}
|
|
|
|
// Handle pre-key errors with upload and delay
|
|
if (isPreKeyError) {
|
|
logger.info({ error: errorMessage }, 'PreKey error detected, uploading and retrying')
|
|
|
|
try {
|
|
logger.debug('Uploading pre-keys for error recovery')
|
|
await uploadPreKeys(5)
|
|
logger.debug('Waiting for server to process new pre-keys')
|
|
await delay(1000)
|
|
} catch (uploadErr) {
|
|
logger.error({ uploadErr }, 'Pre-key upload failed, proceeding with retry anyway')
|
|
}
|
|
}
|
|
|
|
const encNode = getBinaryNodeChild(node, 'enc')
|
|
await sendRetryRequest(node, !encNode)
|
|
if (retryRequestDelayMs) {
|
|
await delay(retryRequestDelayMs)
|
|
}
|
|
} catch (err) {
|
|
logger.error({ err, isPreKeyError }, 'Failed to handle retry, attempting basic retry')
|
|
// Still attempt retry even if pre-key upload failed
|
|
try {
|
|
const encNode = getBinaryNodeChild(node, 'enc')
|
|
await sendRetryRequest(node, !encNode)
|
|
} catch (retryErr) {
|
|
logger.error({ retryErr }, 'Failed to send retry after error handling')
|
|
}
|
|
}
|
|
|
|
await sendMessageAck(node, NACK_REASONS.UnhandledError)
|
|
})
|
|
} else {
|
|
if (messageRetryManager && msg.key.id) {
|
|
messageRetryManager.cancelPendingPhoneRequest(msg.key.id)
|
|
}
|
|
|
|
const isNewsletter = isJidNewsletter(msg.key.remoteJid!)
|
|
if (!isNewsletter) {
|
|
// no type in the receipt => message delivered
|
|
let type: MessageReceiptType = undefined
|
|
let participant = msg.key.participant
|
|
if (category === 'peer') {
|
|
// special peer message
|
|
type = 'peer_msg'
|
|
} else if (msg.key.fromMe) {
|
|
// message was sent by us from a different device
|
|
type = 'sender'
|
|
// need to specially handle this case
|
|
if (isLidUser(msg.key.remoteJid!) || isLidUser(msg.key.remoteJidAlt)) {
|
|
participant = author // TODO: investigate sending receipts to LIDs and not PNs
|
|
}
|
|
} else if (!sendActiveReceipts) {
|
|
type = 'inactive'
|
|
}
|
|
|
|
await sendReceipt(msg.key.remoteJid!, participant!, [msg.key.id!], type)
|
|
|
|
// send ack for history message
|
|
const isAnyHistoryMsg = getHistoryMsg(msg.message!)
|
|
if (isAnyHistoryMsg) {
|
|
const jid = jidNormalizedUser(msg.key.remoteJid!)
|
|
await sendReceipt(jid, undefined, [msg.key.id!], 'hist_sync') // TODO: investigate
|
|
}
|
|
} else {
|
|
await sendMessageAck(node)
|
|
logger.debug({ key: msg.key }, 'processed newsletter message without receipts')
|
|
}
|
|
}
|
|
|
|
// JID normalization moved BEFORE mutex acquisition (line 1273) to prevent race conditions
|
|
// cleanMessage still runs inside mutex to ensure atomic message processing
|
|
cleanMessage(msg, authState.creds.me!.id, authState.creds.me!.lid!)
|
|
|
|
await upsertMessage(msg, node.attrs.offline ? 'append' : 'notify')
|
|
|
|
// Log with [BAILEYS] prefix
|
|
if (msg.key.id && msg.key.remoteJid) {
|
|
logMessageReceived(msg.key.id, msg.key.remoteJid)
|
|
}
|
|
|
|
// Record message received metric
|
|
const msgContent = msg.message
|
|
const msgType = msgContent?.conversation
|
|
? 'text'
|
|
: msgContent?.imageMessage
|
|
? 'image'
|
|
: msgContent?.videoMessage
|
|
? 'video'
|
|
: msgContent?.audioMessage
|
|
? 'audio'
|
|
: msgContent?.documentMessage
|
|
? 'document'
|
|
: msgContent?.stickerMessage
|
|
? 'sticker'
|
|
: msgContent?.reactionMessage
|
|
? 'reaction'
|
|
: 'other'
|
|
recordMessageReceived(msgType)
|
|
|
|
// Track session activity for cleanup
|
|
if (sessionActivityTracker && msg.key.remoteJid) {
|
|
sessionActivityTracker.recordActivity(msg.key.remoteJid)
|
|
|
|
// For groups, also track participant activity
|
|
if (msg.key.participant) {
|
|
sessionActivityTracker.recordActivity(msg.key.participant)
|
|
}
|
|
}
|
|
})
|
|
} catch (error) {
|
|
logger.error({ error, node: binaryNodeToString(node) }, 'error in handling message')
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Sanitizes caller phone number to fix known decoder bugs
|
|
*
|
|
* Brazilian Phone Format:
|
|
* - Mobile: 55 + DD + 9XXXXXXXX (13 digits total)
|
|
* Example: 5515991000000 (55 + 15 + 991000000)
|
|
* - Landline: 55 + DD + XXXXXXXX (12 digits total)
|
|
* Example: 551541410000 (55 + 15 + 41410000)
|
|
*
|
|
* Decoder Bug: Landlines incorrectly get trailing zero
|
|
* Example: 551738025555 → 5517380255550 (should be 12, not 13)
|
|
*
|
|
* Detection Logic:
|
|
* - Extract first digit after DDD (position 4 in string)
|
|
* - If digit is 2-5: It's a LANDLINE
|
|
* → 13 digits is ERROR → Remove trailing zero
|
|
* - If digit is 6-9: It's a MOBILE
|
|
* → 13 digits is CORRECT → Don't touch!
|
|
*
|
|
* @param pn - Raw phone number from caller_pn attribute
|
|
* @returns Sanitized phone number
|
|
*/
|
|
const sanitizeCallerPn = (pn: string | undefined): string | undefined => {
|
|
if (!pn) {
|
|
return undefined
|
|
}
|
|
|
|
// Only process Brazilian numbers (country code 55)
|
|
if (!pn.startsWith('55')) {
|
|
return pn
|
|
}
|
|
|
|
// Check if it's a 13-digit number (potential landline bug)
|
|
if (pn.length === 13) {
|
|
// Extract first digit after DDD (position 4)
|
|
// Format: 55 DD X...
|
|
// 01 23 4...
|
|
const firstDigitAfterDDD = pn.charAt(4)
|
|
|
|
// Landline: first digit is 2-5
|
|
// If 13 digits, it's an error (should be 12) - remove trailing zero
|
|
if (['2', '3', '4', '5'].includes(firstDigitAfterDDD)) {
|
|
// Extra validation: only sanitize if ends with 0 (the bug pattern)
|
|
if (pn.endsWith('0')) {
|
|
const sanitized = pn.slice(0, -1)
|
|
logger.debug(
|
|
{ original: pn, sanitized, firstDigit: firstDigitAfterDDD, type: 'landline' },
|
|
'Call: Sanitized Brazilian landline number (removed trailing zero)'
|
|
)
|
|
return sanitized
|
|
}
|
|
}
|
|
|
|
// Mobile: first digit is 6-9
|
|
// 13 digits is CORRECT for mobile - don't sanitize!
|
|
if (['6', '7', '8', '9'].includes(firstDigitAfterDDD)) {
|
|
logger.trace(
|
|
{ pn, firstDigit: firstDigitAfterDDD, type: 'mobile' },
|
|
'Call: Valid Brazilian mobile number (13 digits correct, not sanitizing)'
|
|
)
|
|
return pn
|
|
}
|
|
}
|
|
|
|
return pn
|
|
}
|
|
|
|
const handleCall = async (node: BinaryNode) => {
|
|
const { attrs } = node
|
|
const [infoChild] = getAllBinaryNodeChildren(node)
|
|
const status = getCallStatusFromNode(infoChild!)
|
|
|
|
if (!infoChild) {
|
|
throw new Boom('Missing call info in call node')
|
|
}
|
|
|
|
const callId = infoChild.attrs['call-id']!
|
|
const from = infoChild.attrs.from! || infoChild.attrs['call-creator']!
|
|
|
|
const call: WACallEvent = {
|
|
chatId: attrs.from!,
|
|
from,
|
|
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']
|
|
// Extract and sanitize caller phone number
|
|
call.callerPn = sanitizeCallerPn(infoChild.attrs['caller_pn'])
|
|
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.groupJid = existingCall.groupJid
|
|
// Preserve callerPn across call state updates
|
|
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) => {
|
|
const key: WAMessageKey = { remoteJid: attrs.from, fromMe: true, id: attrs.id }
|
|
|
|
// WARNING: REFRAIN FROM ENABLING THIS FOR NOW. IT WILL CAUSE A LOOP
|
|
// // current hypothesis is that if pash is sent in the ack
|
|
// // it means -- the message hasn't reached all devices yet
|
|
// // we'll retry sending the message here
|
|
// if(attrs.phash) {
|
|
// logger.info({ attrs }, 'received phash in ack, resending message...')
|
|
// const msg = await getMessage(key)
|
|
// if(msg) {
|
|
// await relayMessage(key.remoteJid!, msg, { messageId: key.id!, useUserDevicesCache: false })
|
|
// } else {
|
|
// logger.warn({ attrs }, 'could not send message again, as it was not found')
|
|
// }
|
|
// }
|
|
|
|
// error in acknowledgement,
|
|
// device could not display the message
|
|
if (attrs.error) {
|
|
if (attrs.error === SERVER_ERROR_CODES.MissingTcToken) {
|
|
const msgId = attrs.id
|
|
const jid = jidNormalizedUser(attrs.from)
|
|
logTcToken('error_463', { jid, msgId })
|
|
|
|
// Single-retry: wait 1.5s for the server's tctoken notification to arrive,
|
|
// then resend. A Set prevents infinite retry loops.
|
|
// Composite key (jid:msgId) ensures retries are isolated per destination.
|
|
const retryKey = `${jid}:${msgId}`
|
|
if (msgId && jid && !tcTokenRetriedMsgIds.has(retryKey)) {
|
|
tcTokenRetriedMsgIds.add(retryKey)
|
|
// Each entry auto-expires after 60s — naturally bounded under normal use
|
|
setTimeout(() => tcTokenRetriedMsgIds.delete(retryKey), 60_000)
|
|
;(async () => {
|
|
try {
|
|
await delay(1500)
|
|
const msg =
|
|
(await getMessage(key)) ??
|
|
// Fallback: ack can arrive <30ms after send, before store persists
|
|
messageRetryManager?.getRecentMessage(jid, msgId)?.message
|
|
if (msg) {
|
|
await relayMessage(jid, msg, { messageId: msgId, useUserDevicesCache: true })
|
|
logTcToken('retry_463_ok', { jid, msgId })
|
|
} else {
|
|
logger.warn({ jid, msgId }, '463 retry: message not found in store')
|
|
ev.emit('messages.update', [
|
|
{ key, update: { status: WAMessageStatus.ERROR, messageStubParameters: ['463'] } }
|
|
])
|
|
}
|
|
} catch (err: any) {
|
|
logger.warn({ jid, msgId, err: err?.message }, '463 retry failed')
|
|
ev.emit('messages.update', [
|
|
{ key, update: { status: WAMessageStatus.ERROR, messageStubParameters: ['463'] } }
|
|
])
|
|
}
|
|
})()
|
|
|
|
return
|
|
}
|
|
} else if (attrs.error === SERVER_ERROR_CODES.SmaxInvalid) {
|
|
logTcToken('error_479', { jid: attrs.from, msgId: attrs.id })
|
|
} else {
|
|
logger.warn({ attrs }, 'received error in ack')
|
|
}
|
|
|
|
ev.emit('messages.update', [
|
|
{
|
|
key,
|
|
update: {
|
|
status: WAMessageStatus.ERROR,
|
|
messageStubParameters: [attrs.error]
|
|
}
|
|
}
|
|
])
|
|
}
|
|
}
|
|
|
|
/// processes a node with the given function
|
|
/// and adds the task to the existing buffer if we're buffering events
|
|
const processNodeWithBuffer = async <T>(
|
|
node: BinaryNode,
|
|
identifier: string,
|
|
exec: (node: BinaryNode, offline: boolean) => Promise<T>
|
|
) => {
|
|
ev.buffer()
|
|
await execTask()
|
|
ev.flush()
|
|
|
|
function execTask() {
|
|
return exec(node, false).catch(err => onUnexpectedError(err, identifier))
|
|
}
|
|
}
|
|
|
|
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([
|
|
['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 = 25
|
|
|
|
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'))
|
|
}
|
|
|
|
return { enqueue }
|
|
}
|
|
|
|
const offlineNodeProcessor = makeOfflineNodeProcessor()
|
|
|
|
const processNode = async (
|
|
type: MessageType,
|
|
node: BinaryNode,
|
|
identifier: string,
|
|
exec: (node: BinaryNode) => Promise<void>
|
|
) => {
|
|
const isOffline = !!node.attrs.offline
|
|
|
|
if (isOffline) {
|
|
offlineNodeProcessor.enqueue(type, node)
|
|
} else {
|
|
await processNodeWithBuffer(node, identifier, exec)
|
|
}
|
|
}
|
|
|
|
// recv a message
|
|
ws.on('CB:message', async (node: BinaryNode) => {
|
|
await processNode('message', node, 'processing message', handleMessage)
|
|
})
|
|
|
|
ws.on('CB:call', async (node: BinaryNode) => {
|
|
await processNode('call', node, 'handling call', handleCall)
|
|
})
|
|
|
|
ws.on('CB:receipt', async node => {
|
|
await processNode('receipt', node, 'handling receipt', handleReceipt)
|
|
})
|
|
|
|
ws.on('CB:notification', async (node: BinaryNode) => {
|
|
await processNode('notification', node, 'handling notification', handleNotification)
|
|
})
|
|
ws.on('CB:ack,class:message', (node: BinaryNode) => {
|
|
handleBadAck(node).catch(error => onUnexpectedError(error, 'handling bad ack'))
|
|
})
|
|
|
|
ev.on('call', async ([call]) => {
|
|
if (!call) {
|
|
return
|
|
}
|
|
|
|
// missed call + group call notification message generation
|
|
if (call.status === 'timeout' || (call.status === 'offer' && call.isGroup)) {
|
|
const msg: WAMessage = {
|
|
key: {
|
|
remoteJid: call.chatId,
|
|
id: call.id,
|
|
fromMe: false
|
|
},
|
|
messageTimestamp: unixTimestampSeconds(call.date)
|
|
}
|
|
if (call.status === 'timeout') {
|
|
if (call.isGroup) {
|
|
msg.messageStubType = call.isVideo
|
|
? WAMessageStubType.CALL_MISSED_GROUP_VIDEO
|
|
: WAMessageStubType.CALL_MISSED_GROUP_VOICE
|
|
} else {
|
|
msg.messageStubType = call.isVideo ? WAMessageStubType.CALL_MISSED_VIDEO : WAMessageStubType.CALL_MISSED_VOICE
|
|
}
|
|
} else {
|
|
msg.message = { call: { callKey: Buffer.from(call.id) } }
|
|
}
|
|
|
|
const protoMsg = proto.WebMessageInfo.fromObject(msg) as WAMessage
|
|
await upsertMessage(protoMsg, call.offline ? 'append' : 'notify')
|
|
}
|
|
})
|
|
|
|
ev.on('connection.update', ({ isOnline, connection }) => {
|
|
// Flush pending tctoken index save on disconnect to avoid writing after close
|
|
if (connection === 'close' && tcTokenIndexSaveTimer) {
|
|
clearTimeout(tcTokenIndexSaveTimer)
|
|
tcTokenIndexSaveTimer = undefined
|
|
// Await index load first — prevents overwriting a more complete persisted index
|
|
// if the connection closes before the initial load finishes.
|
|
tcTokenIndexLoaded
|
|
.then(() => {
|
|
Promise.resolve(
|
|
authState.keys.set({
|
|
tctoken: {
|
|
[TC_TOKEN_INDEX_KEY]: {
|
|
token: Buffer.from(JSON.stringify([...tcTokenKnownJids]), 'utf8'),
|
|
timestamp: unixTimestampSeconds().toString()
|
|
}
|
|
}
|
|
})
|
|
).catch(() => {
|
|
/* non-critical */
|
|
})
|
|
})
|
|
.catch(() => {
|
|
/* non-critical */
|
|
})
|
|
}
|
|
|
|
if (typeof isOnline !== 'undefined') {
|
|
sendActiveReceipts = isOnline
|
|
logger.trace(`sendActiveReceipts set to "${sendActiveReceipts}"`)
|
|
|
|
// Prune expired tctokens when coming online (max once per 24h)
|
|
if (isOnline) {
|
|
const now = Date.now()
|
|
const ONE_DAY_MS = 86400000
|
|
if (now - lastTcTokenPruneTs > ONE_DAY_MS) {
|
|
lastTcTokenPruneTs = now
|
|
// Persist prune timestamp so it survives restarts
|
|
Promise.resolve(
|
|
authState.keys.set({
|
|
tctoken: {
|
|
[TC_TOKEN_PRUNE_TS_KEY]: {
|
|
token: Buffer.alloc(0),
|
|
timestamp: now.toString()
|
|
}
|
|
}
|
|
})
|
|
).catch(() => {
|
|
/* non-critical */
|
|
})
|
|
pruneExpiredTcTokens().catch(err => {
|
|
logger.debug({ err: err?.message }, 'tctoken pruning failed')
|
|
})
|
|
}
|
|
}
|
|
}
|
|
})
|
|
|
|
return {
|
|
...sock,
|
|
sendMessageAck,
|
|
sendRetryRequest,
|
|
rejectCall,
|
|
fetchMessageHistory,
|
|
requestPlaceholderResend,
|
|
messageRetryManager
|
|
}
|
|
}
|