diff --git a/package.json b/package.json index 2f4a96e5..92c86f4f 100644 --- a/package.json +++ b/package.json @@ -31,7 +31,7 @@ "gen:protobuf": "sh WAProto/GenerateStatics.sh", "format": "prettier --write \"src/**/*.{ts,js,json,md}\"", "lint": "tsc && eslint src --ext .js,.ts", - "lint:fix": "npm run format && npm run lint --fix", + "lint:fix": "npm run format && npm run lint -- --fix", "prepack": "npm run build", "prepare": "npm run build", "preinstall": "node ./engine-requirements.js", @@ -45,6 +45,7 @@ "async-mutex": "^0.5.0", "axios": "^1.6.0", "libsignal": "git+https://github.com/whiskeysockets/libsignal-node", + "lru-cache": "^11.1.0", "music-metadata": "^11.7.0", "pino": "^9.6", "protobufjs": "^7.2.4", diff --git a/src/Defaults/index.ts b/src/Defaults/index.ts index b1126867..4d2a0a1d 100644 --- a/src/Defaults/index.ts +++ b/src/Defaults/index.ts @@ -59,6 +59,8 @@ export const DEFAULT_CONNECTION_CONFIG: SocketConfig = { linkPreviewImageThumbnailWidth: 192, transactionOpts: { maxCommitRetries: 10, delayBetweenTriesMs: 3000 }, generateHighQualityLinkPreview: false, + enableAutoSessionRecreation: true, + enableRecentMessageCache: true, options: {}, appStateMacVerification: { patch: false, diff --git a/src/Signal/libsignal.ts b/src/Signal/libsignal.ts index a1fd2acf..c601c1be 100644 --- a/src/Signal/libsignal.ts +++ b/src/Signal/libsignal.ts @@ -13,6 +13,7 @@ import { GroupCipher, GroupSessionBuilder, SenderKeyDistributionMessage } from ' import { LIDMappingStore } from './lid-mapping' export function makeLibSignalRepository(auth: SignalAuthState): SignalRepository { + const lidMapping = new LIDMappingStore(auth.keys as SignalKeyStoreWithTransaction) const storage = signalStorage(auth, lidMapping) // Simple operation-level deduplication (5 minutes) @@ -127,6 +128,24 @@ export function makeLibSignalRepository(auth: SignalAuthState): SignalRepository const cipher = new libsignal.SessionBuilder(storage, jidToSignalProtocolAddress(jid)) await cipher.initOutgoing(session) }, + async validateSession(jid: string) { + try { + const addr = jidToSignalProtocolAddress(jid) + const session = await storage.loadSession(addr.toString()) + + if (!session) { + return { exists: false, reason: 'no session' } + } + + if (!session.haveOpenSession()) { + return { exists: false, reason: 'no open session' } + } + + return { exists: true } + } catch (error) { + return { exists: false, reason: 'validation error' } + } + }, jidToSignalProtocolAddress(jid) { return jidToSignalProtocolAddress(jid).toString() }, diff --git a/src/Socket/messages-recv.ts b/src/Socket/messages-recv.ts index 0a2a894b..4f4e6db6 100644 --- a/src/Socket/messages-recv.ts +++ b/src/Socket/messages-recv.ts @@ -60,7 +60,8 @@ import { extractGroupMetadata } from './groups' import { makeMessagesSocket } from './messages-send' export const makeMessagesRecvSocket = (config: SocketConfig) => { - const { logger, retryRequestDelayMs, maxMsgRetryCount, getMessage, shouldIgnoreJid } = config + const { logger, retryRequestDelayMs, maxMsgRetryCount, getMessage, shouldIgnoreJid, enableAutoSessionRecreation } = + config const sock = makeMessagesSocket(config) const { ev, @@ -77,7 +78,8 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { relayMessage, sendReceipt, uploadPreKeys, - sendPeerDataOperationMessage + sendPeerDataOperationMessage, + messageRetryManager } = sock /** this mutex ensures that each retryRequest will wait for the previous one to finish */ @@ -169,23 +171,81 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { const { key: msgKey } = fullMessage const msgId = msgKey.id! - const key = `${msgId}:${msgKey?.participant}` - let retryCount = msgRetryCache.get(key) || 0 - if (retryCount >= maxMsgRetryCount) { - logger.debug({ retryCount, msgId }, 'reached retry limit, clearing') - msgRetryCache.del(key) - return + 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) + return + } + + // Increment retry count using new system + const retryCount = messageRetryManager.incrementRetryCount(msgId) + + // Use the new retry count for the rest of the logic + const key = `${msgId}:${msgKey?.participant}` + msgRetryCache.set(key, retryCount) + } else { + // Fallback to old system + const key = `${msgId}:${msgKey?.participant}` + let retryCount = msgRetryCache.get(key) || 0 + if (retryCount >= maxMsgRetryCount) { + logger.debug({ retryCount, msgId }, 'reached retry limit, clearing') + msgRetryCache.del(key) + return + } + + retryCount += 1 + msgRetryCache.set(key, retryCount) } - retryCount += 1 - msgRetryCache.set(key, retryCount) + const key = `${msgId}:${msgKey?.participant}` + const retryCount = msgRetryCache.get(key) || 1 const { account, signedPreKey, signedIdentityKey: identityKey } = authState.creds + const fromJid = node.attrs.from! - if (retryCount === 1) { - //request a resend via phone - const msgId = await requestPlaceholderResend(msgKey) - logger.debug(`sendRetryRequest: requested placeholder resend for message ${msgId}`) + // Check if we should recreate the session + let shouldRecreateSession = false + let recreateReason = '' + + if (enableAutoSessionRecreation && messageRetryManager) { + try { + // Check if we have a session with this JID + const sessionId = signalRepository.jidToSignalProtocolAddress(fromJid) + const hasSession = await signalRepository.validateSession(fromJid) + const result = messageRetryManager.shouldRecreateSession(fromJid, retryCount, hasSession.exists) + shouldRecreateSession = result.recreate + recreateReason = result.reason + + if (shouldRecreateSession) { + logger.info({ fromJid, retryCount, reason: recreateReason }, 'recreating session for retry') + // Delete existing session to force recreation + await authState.keys.set({ session: { [sessionId]: null } }) + 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 msgId = await requestPlaceholderResend(msgKey) + logger.debug(`sendRetryRequest: requested placeholder resend 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}`) + } } const deviceIdentity = encodeSignedDeviceIdentity(account!, true) @@ -223,7 +283,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { receipt.attrs.participant = node.attrs.participant } - if (retryCount > 1 || forceIncludeKeys) { + if (retryCount > 1 || forceIncludeKeys || shouldRecreateSession) { const { update, preKeys } = await getNextPreKeys(authState, 1) const [keyId] = Object.keys(preKeys) @@ -247,7 +307,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { await sendNode(receipt) - logger.info({ msgAttrs: node.attrs, retryCount }, 'sent retry receipt') + logger.info({ msgAttrs: node.attrs, retryCount, shouldRecreateSession, recreateReason }, 'sent retry receipt') }) } @@ -597,21 +657,77 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { } const sendMessagesAgain = async (key: proto.IMessageKey, ids: string[], retryNode: BinaryNode) => { - // todo: implement a cache to store the last 256 sent messages (copy whatsmeow) - const msgs = await Promise.all(ids.map(id => getMessage({ ...key, id }))) 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 - await assertSessions([participant], true) + + // Check if we should recreate session for this retry + let shouldRecreateSession = false + let recreateReason = '' + + if (enableAutoSessionRecreation && messageRetryManager) { + try { + const sessionId = signalRepository.jidToSignalProtocolAddress(participant) + + const hasSession = await signalRepository.validateSession(participant) + const result = messageRetryManager.shouldRecreateSession(participant, retryCount, hasSession.exists) + shouldRecreateSession = result.recreate + recreateReason = result.reason + + if (shouldRecreateSession) { + logger.info({ participant, retryCount, reason: recreateReason }, 'recreating session for outgoing retry') + await authState.keys.set({ session: { [sessionId]: null } }) + } + } catch (error) { + logger.warn({ error, participant }, 'failed to check session recreation for outgoing retry') + } + } + + await assertSessions([participant], shouldRecreateSession) if (isJidGroup(remoteJid)) { await authState.keys.set({ 'sender-key-memory': { [remoteJid]: null } }) } - logger.debug({ participant, sendToAll }, 'forced new session for retry recp') + logger.debug({ participant, sendToAll, shouldRecreateSession, recreateReason }, 'forced new session for retry recp') for (const [i, msg] of msgs.entries()) { if (!ids[i]) continue @@ -816,6 +932,16 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { ev.emit('chats.phoneNumberShare', { lid: node.attrs.from!, jid: node.attrs.sender_pn }) } + 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' + ) + if (msg.message?.protocolMessage?.lidMigrationMappingSyncMessage?.encodedMappingPayload) { try { const payload = msg.message.protocolMessage.lidMigrationMappingSyncMessage.encodedMappingPayload @@ -1409,6 +1535,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { sendRetryRequest, rejectCall, fetchMessageHistory, - requestPlaceholderResend + requestPlaceholderResend, + messageRetryManager } } diff --git a/src/Socket/messages-send.ts b/src/Socket/messages-send.ts index 5e8e7e1b..be81042a 100644 --- a/src/Socket/messages-send.ts +++ b/src/Socket/messages-send.ts @@ -26,6 +26,7 @@ import { getStatusCodeForMediaRetry, getUrlFromDirectPath, getWAUploadToServer, + MessageRetryManager, normalizeMessageContent, parseAndInjectE2ESessions, unixTimestampSeconds @@ -58,7 +59,9 @@ export const makeMessagesSocket = (config: SocketConfig) => { generateHighQualityLinkPreview, options: axiosOptions, patchMessageBeforeSending, - cachedGroupMetadata + cachedGroupMetadata, + enableRecentMessageCache, + maxMsgRetryCount } = config const sock: NewsletterSocket = makeNewsletterSocket(makeGroupsSocket(config)) const { @@ -81,7 +84,10 @@ export const makeMessagesSocket = (config: SocketConfig) => { useClones: false }) - // Prevent race conditions in Signal session encryption by user + // Initialize message retry manager if enabled + const messageRetryManager = enableRecentMessageCache ? new MessageRetryManager(logger, maxMsgRetryCount) : null + + // Prevent race conditions in Signal session encryption by user const encryptionMutex = makeKeyedMutex() let mediaConn: Promise @@ -389,6 +395,9 @@ export const makeMessagesSocket = (config: SocketConfig) => { } jidsRequiringFetch.push(jid) + if (validation.reason) { + logger.debug({ jid, reason: validation.reason }, 'session validation failed') + } } } @@ -1118,6 +1127,11 @@ export const makeMessagesSocket = (config: SocketConfig) => { logger.debug({ msgId }, `sending message to ${participants.length} devices`) await sendNode(stanza) + + // Add message to retry cache if enabled + if (messageRetryManager && !participant) { + messageRetryManager.addRecentMessage(destinationJid, msgId, message) + } }) return msgId @@ -1215,6 +1229,7 @@ export const makeMessagesSocket = (config: SocketConfig) => { sendPeerDataOperationMessage, createParticipantNodes, getUSyncDevices, + messageRetryManager, updateMediaMessage: async (message: proto.IWebMessageInfo) => { const content = assertMediaContent(message.message) const mediaKey = content.mediaKey! diff --git a/src/Types/Chat.ts b/src/Types/Chat.ts index e951c22c..d0a9a86d 100644 --- a/src/Types/Chat.ts +++ b/src/Types/Chat.ts @@ -73,6 +73,8 @@ export type ChatUpdate = Partial< * undefined if the condition is not yet fulfilled * */ conditional: (bufferedData: BufferedEventData) => boolean | undefined + /** last update time */ + timestamp?: number } > diff --git a/src/Types/Signal.ts b/src/Types/Signal.ts index bd389ff4..f20a866b 100644 --- a/src/Types/Signal.ts +++ b/src/Types/Signal.ts @@ -74,6 +74,7 @@ export type SignalRepository = { ciphertext: Uint8Array }> injectE2ESession(opts: E2ESessionOpts): Promise + validateSession(jid: string): Promise<{ exists: boolean; reason?: string }> jidToSignalProtocolAddress(jid: string): string storeLIDPNMapping(lid: string, pn: string): Promise getLIDMappingStore(): LIDMappingStore diff --git a/src/Types/Socket.ts b/src/Types/Socket.ts index 41ed0c62..03e721f7 100644 --- a/src/Types/Socket.ts +++ b/src/Types/Socket.ts @@ -95,6 +95,12 @@ export type SocketConfig = { * */ generateHighQualityLinkPreview: boolean + /** Enable automatic session recreation for failed messages */ + enableAutoSessionRecreation: boolean + + /** Enable recent message caching for retry handling */ + enableRecentMessageCache: boolean + /** * Returns if a jid should be ignored, * no event for that jid will be triggered. diff --git a/src/Utils/event-buffer.ts b/src/Utils/event-buffer.ts index 0dcead48..cd63640d 100644 --- a/src/Utils/event-buffer.ts +++ b/src/Utils/event-buffer.ts @@ -66,7 +66,6 @@ type BaileysBufferableEventEmitter = BaileysEventEmitter & { /** * The event buffer logically consolidates different events into a single event * making the data processing more efficient. - * @param ev the baileys event emitter */ export const makeEventBuffer = (logger: ILogger): BaileysBufferableEventEmitter => { const ev = new EventEmitter() @@ -74,6 +73,10 @@ export const makeEventBuffer = (logger: ILogger): BaileysBufferableEventEmitter let data = makeBufferData() let isBuffering = false + let bufferTimeout: NodeJS.Timeout | null = null + let bufferCount = 0 + const MAX_HISTORY_CACHE_SIZE = 10000 // Limit the history cache size to prevent memory bloat + const BUFFER_TIMEOUT_MS = 30000 // 30 seconds // take the generic event and fire it as a baileys event ev.on('event', (map: BaileysEventData) => { @@ -86,6 +89,21 @@ export const makeEventBuffer = (logger: ILogger): BaileysBufferableEventEmitter if (!isBuffering) { logger.debug('Event buffer activated') isBuffering = true + bufferCount++ + + // Auto-flush after a timeout to prevent infinite buffering + if (bufferTimeout) { + clearTimeout(bufferTimeout) + } + + bufferTimeout = setTimeout(() => { + if (isBuffering) { + logger.warn('Buffer timeout reached, auto-flushing') + flush() + } + }, BUFFER_TIMEOUT_MS) + } else { + bufferCount++ } } @@ -94,8 +112,21 @@ export const makeEventBuffer = (logger: ILogger): BaileysBufferableEventEmitter return false } - logger.debug('Flushing event buffer') + logger.debug({ bufferCount }, 'Flushing event buffer') isBuffering = false + bufferCount = 0 + + // Clear timeout + if (bufferTimeout) { + clearTimeout(bufferTimeout) + bufferTimeout = null + } + + // Clear history cache if it exceeds the max size + if (historyCache.size > MAX_HISTORY_CACHE_SIZE) { + logger.debug({ cacheSize: historyCache.size }, 'Clearing history cache') + historyCache.clear() + } const newData = makeBufferData() const chatUpdates = Object.values(data.chatUpdates) @@ -148,9 +179,25 @@ export const makeEventBuffer = (logger: ILogger): BaileysBufferableEventEmitter return async (...args) => { buffer() try { - return await work(...args) + const result = await work(...args) + // If this is the only buffer, flush after a small delay + if (bufferCount === 1) { + setTimeout(() => { + if (isBuffering && bufferCount === 1) { + flush() + } + }, 100) // Small delay to allow nested buffers + } + + return result + } catch (error) { + throw error } finally { - // Flushing is now controlled centrally by the state machine. + bufferCount = Math.max(0, bufferCount - 1) + if (bufferCount === 0) { + // Auto-flush when no other buffers are active + setTimeout(flush, 100) + } } } }, diff --git a/src/Utils/index.ts b/src/Utils/index.ts index 156abe09..d8bd0627 100644 --- a/src/Utils/index.ts +++ b/src/Utils/index.ts @@ -15,3 +15,4 @@ export * from './use-multi-file-auth-state' export * from './link-preview' export * from './event-buffer' export * from './process-message' +export * from './message-retry-manager' diff --git a/src/Utils/message-retry-manager.ts b/src/Utils/message-retry-manager.ts new file mode 100644 index 00000000..5de5bde6 --- /dev/null +++ b/src/Utils/message-retry-manager.ts @@ -0,0 +1,205 @@ +import { LRUCache } from 'lru-cache' +import { proto } from '../../WAProto' +import type { ILogger } from './logger' + +/** Number of sent messages to cache in memory for handling retry receipts */ +const RECENT_MESSAGES_SIZE = 512 + +/** Timeout for session recreation - 1 hour */ +const RECREATE_SESSION_TIMEOUT = 60 * 60 * 1000 // 1 hour in milliseconds +const PHONE_REQUEST_DELAY = 3000 +export interface RecentMessageKey { + to: string + id: string +} + +export interface RecentMessage { + message: proto.IMessage + timestamp: number +} + +export interface SessionRecreateHistory { + [jid: string]: number // timestamp +} + +export interface RetryCounter { + [messageId: string]: number +} + +export interface PendingPhoneRequest { + [messageId: string]: NodeJS.Timeout +} + +export interface RetryStatistics { + totalRetries: number + successfulRetries: number + failedRetries: number + mediaRetries: number + sessionRecreations: number + phoneRequests: number +} + +export class MessageRetryManager { + private recentMessagesMap = new LRUCache({ + max: RECENT_MESSAGES_SIZE + }) + private sessionRecreateHistory = new LRUCache({ + ttl: RECREATE_SESSION_TIMEOUT * 2, + ttlAutopurge: true + }) + private retryCounters = new LRUCache({ + ttl: 15 * 60 * 1000, + ttlAutopurge: true, + updateAgeOnGet: true + }) // 15 minutes TTL + private pendingPhoneRequests: PendingPhoneRequest = {} + private readonly maxMsgRetryCount: number = 5 + private statistics: RetryStatistics = { + totalRetries: 0, + successfulRetries: 0, + failedRetries: 0, + mediaRetries: 0, + sessionRecreations: 0, + phoneRequests: 0 + } + + constructor( + private logger: ILogger, + maxMsgRetryCount: number + ) { + this.maxMsgRetryCount = maxMsgRetryCount + } + + /** + * Add a recent message to the cache for retry handling + */ + addRecentMessage(to: string, id: string, message: proto.IMessage): void { + const key: RecentMessageKey = { to, id } + const keyStr = this.keyToString(key) + + // Add new message + this.recentMessagesMap.set(keyStr, { + message, + timestamp: Date.now() + }) + + this.logger.debug(`Added message to retry cache: ${to}/${id}`) + } + + /** + * Get a recent message from the cache + */ + getRecentMessage(to: string, id: string): RecentMessage | undefined { + const key: RecentMessageKey = { to, id } + const keyStr = this.keyToString(key) + return this.recentMessagesMap.get(keyStr) + } + + /** + * Check if a session should be recreated based on retry count and history + */ + shouldRecreateSession(jid: string, retryCount: number, hasSession: boolean): { reason: string; recreate: boolean } { + // If we don't have a session, always recreate + if (!hasSession) { + this.sessionRecreateHistory.set(jid, Date.now()) + this.statistics.sessionRecreations++ + return { + reason: "we don't have a Signal session with them", + recreate: true + } + } + + // Only consider recreation if retry count > 1 + if (retryCount < 2) { + return { reason: '', recreate: false } + } + + const now = Date.now() + const prevTime = this.sessionRecreateHistory.get(jid) + + // If no previous recreation or it's been more than an hour + if (!prevTime || now - prevTime > RECREATE_SESSION_TIMEOUT) { + this.sessionRecreateHistory.set(jid, now) + this.statistics.sessionRecreations++ + return { + reason: 'retry count > 1 and over an hour since last recreation', + recreate: true + } + } + + return { reason: '', recreate: false } + } + + /** + * Increment retry counter for a message + */ + incrementRetryCount(messageId: string): number { + this.retryCounters.set(messageId, (this.retryCounters.get(messageId) || 0) + 1) + this.statistics.totalRetries++ + return this.retryCounters.get(messageId)! + } + + /** + * Get retry count for a message + */ + getRetryCount(messageId: string): number { + return this.retryCounters.get(messageId) || 0 + } + + /** + * Check if message has exceeded maximum retry attempts + */ + hasExceededMaxRetries(messageId: string): boolean { + return this.getRetryCount(messageId) >= this.maxMsgRetryCount + } + + /** + * Mark retry as successful + */ + markRetrySuccess(messageId: string): void { + this.statistics.successfulRetries++ + // Clean up retry counter for successful message + this.retryCounters.delete(messageId) + this.cancelPendingPhoneRequest(messageId) + } + + /** + * Mark retry as failed + */ + markRetryFailed(messageId: string): void { + this.statistics.failedRetries++ + this.retryCounters.delete(messageId) + } + + /** + * Schedule a phone request with delay + */ + schedulePhoneRequest(messageId: string, callback: () => void, delay: number = PHONE_REQUEST_DELAY): void { + // Cancel any existing request for this message + this.cancelPendingPhoneRequest(messageId) + + this.pendingPhoneRequests[messageId] = setTimeout(() => { + delete this.pendingPhoneRequests[messageId] + this.statistics.phoneRequests++ + callback() + }, delay) + + this.logger.debug(`Scheduled phone request for message ${messageId} with ${delay}ms delay`) + } + + /** + * Cancel pending phone request + */ + cancelPendingPhoneRequest(messageId: string): void { + const timeout = this.pendingPhoneRequests[messageId] + if (timeout) { + clearTimeout(timeout) + delete this.pendingPhoneRequests[messageId] + this.logger.debug(`Cancelled pending phone request for message ${messageId}`) + } + } + + private keyToString(key: RecentMessageKey): string { + return `${key.to}:${key.id}` + } +}