diff --git a/src/Defaults/index.ts b/src/Defaults/index.ts index b0a94336..b4a99b05 100644 --- a/src/Defaults/index.ts +++ b/src/Defaults/index.ts @@ -202,17 +202,21 @@ export const DEFAULT_CACHE_MAX_KEYS = { * - BAILEYS_SESSION_CLEANUP_ENABLED: Enable/disable cleanup (default: true) * - BAILEYS_SESSION_CLEANUP_INTERVAL: Cleanup interval in ms (default: 24h) * - BAILEYS_SESSION_CLEANUP_HOUR: Hour to run cleanup (default: 3 = 3am) - * - BAILEYS_SESSION_SECONDARY_INACTIVE_DAYS: Days before cleaning secondary devices (default: 15) + * - BAILEYS_SESSION_SECONDARY_INACTIVE_DAYS: Days before cleaning secondary devices (default: 7) * - BAILEYS_SESSION_PRIMARY_INACTIVE_DAYS: Days before cleaning primary device (default: 30) * - BAILEYS_SESSION_LID_ORPHAN_HOURS: Hours before cleaning orphaned LID sessions (default: 24) + * - BAILEYS_SESSION_CLEANUP_ON_STARTUP: Run cleanup immediately on startup (default: true) + * - BAILEYS_SESSION_AUTO_CLEAN_CORRUPTED: Auto-delete corrupted sessions (Bad MAC) (default: true) */ export const DEFAULT_SESSION_CLEANUP_CONFIG = { enabled: process.env.BAILEYS_SESSION_CLEANUP_ENABLED !== 'false', intervalMs: parseInt(process.env.BAILEYS_SESSION_CLEANUP_INTERVAL || '86400000', 10), // 24h cleanupHour: parseInt(process.env.BAILEYS_SESSION_CLEANUP_HOUR || '3', 10), // 3am - secondaryDeviceInactiveDays: parseInt(process.env.BAILEYS_SESSION_SECONDARY_INACTIVE_DAYS || '15', 10), + secondaryDeviceInactiveDays: parseInt(process.env.BAILEYS_SESSION_SECONDARY_INACTIVE_DAYS || '7', 10), primaryDeviceInactiveDays: parseInt(process.env.BAILEYS_SESSION_PRIMARY_INACTIVE_DAYS || '30', 10), - lidOrphanHours: parseInt(process.env.BAILEYS_SESSION_LID_ORPHAN_HOURS || '24', 10) + lidOrphanHours: parseInt(process.env.BAILEYS_SESSION_LID_ORPHAN_HOURS || '24', 10), + cleanupOnStartup: process.env.BAILEYS_SESSION_CLEANUP_ON_STARTUP !== 'false', + autoCleanCorrupted: process.env.BAILEYS_SESSION_AUTO_CLEAN_CORRUPTED !== 'false' } // Re-export retry constants for backwards compatibility diff --git a/src/Signal/session-cleanup.ts b/src/Signal/session-cleanup.ts index 0e399f1e..e5da98eb 100644 --- a/src/Signal/session-cleanup.ts +++ b/src/Signal/session-cleanup.ts @@ -28,6 +28,8 @@ export interface SessionCleanupConfig { secondaryDeviceInactiveDays: number primaryDeviceInactiveDays: number lidOrphanHours: number + cleanupOnStartup: boolean + autoCleanCorrupted: boolean } /** @@ -71,6 +73,7 @@ export const makeSessionCleanup = ( config: SessionCleanupConfig = DEFAULT_SESSION_CLEANUP_CONFIG ) => { let cleanupInterval: ReturnType | null = null + let initialTimeout: ReturnType | null = null let lastCleanupAt: number = 0 let cleanupRunning: boolean = false @@ -386,11 +389,21 @@ export const makeSessionCleanup = ( cleanupHour: config.cleanupHour, secondaryDeviceInactiveDays: config.secondaryDeviceInactiveDays, primaryDeviceInactiveDays: config.primaryDeviceInactiveDays, - lidOrphanHours: config.lidOrphanHours + lidOrphanHours: config.lidOrphanHours, + cleanupOnStartup: config.cleanupOnStartup, + autoCleanCorrupted: config.autoCleanCorrupted }, '๐Ÿงน Session cleanup scheduler started' ) + // Run immediate cleanup on startup if enabled + if (config.cleanupOnStartup) { + logger.info('๐Ÿš€ Running cleanup on startup...') + runCleanup().catch(err => { + logger.error({ err }, 'Cleanup on startup failed') + }) + } + // Schedule first cleanup at configured hour const msUntilFirst = msUntilNextCleanup() logger.info( @@ -398,7 +411,9 @@ export const makeSessionCleanup = ( 'โฐ First cleanup scheduled' ) - setTimeout(async () => { + initialTimeout = setTimeout(async () => { + initialTimeout = null // Clear reference after execution + // Run first cleanup await runCleanup() @@ -413,6 +428,11 @@ export const makeSessionCleanup = ( * Stop periodic session cleanup */ const stop = () => { + if (initialTimeout) { + clearTimeout(initialTimeout) + initialTimeout = null + } + if (cleanupInterval) { clearInterval(cleanupInterval) cleanupInterval = null diff --git a/src/Utils/decode-wa-message.ts b/src/Utils/decode-wa-message.ts index 5a3b61c9..d08aecfe 100644 --- a/src/Utils/decode-wa-message.ts +++ b/src/Utils/decode-wa-message.ts @@ -1,5 +1,6 @@ import { Boom } from '@hapi/boom' import { proto } from '../../WAProto/index.js' +import { DEFAULT_SESSION_CLEANUP_CONFIG } from '../Defaults' import type { WAMessage, WAMessageKey } from '../Types' import type { SignalRepositoryWithLIDStore } from '../Types/Signal' import { @@ -13,11 +14,13 @@ import { isJidNewsletter, isJidStatusBroadcast, isLidUser, - isPnUser + isPnUser, + jidDecode // transferDevice } from '../WABinary' import { unpadRandomMax16 } from './generics' import type { ILogger } from './logger' +import { retry, type RetryOptions } from './retry-utils' export const getDecryptionJid = async (sender: string, repository: SignalRepositoryWithLIDStore): Promise => { if (isLidUser(sender) || isHostedLidUser(sender)) { @@ -51,12 +54,45 @@ const storeMappingFromEnvelope = async ( export const NO_MESSAGE_FOUND_ERROR_TEXT = 'Message absent from node' export const MISSING_KEYS_ERROR_TEXT = 'Key used already or never filled' +export const BAD_MAC_ERROR_TEXT = 'Bad MAC' // Retry configuration for failed decryption export const DECRYPTION_RETRY_CONFIG = { maxRetries: 3, baseDelayMs: 100, - sessionRecordErrors: ['No session record', 'SessionError: No session record'] + sessionRecordErrors: ['No session record', 'SessionError: No session record'], + corruptedSessionErrors: ['Bad MAC', 'MessageCounterError', MISSING_KEYS_ERROR_TEXT] +} + +/** + * Retry options for decryption operations + * Uses exponential backoff with jitter to handle transient failures + */ +export const DECRYPTION_RETRY_OPTIONS: RetryOptions = { + maxAttempts: 3, + baseDelay: 200, // 200ms base delay + maxDelay: 2000, // 2s max delay + backoffStrategy: 'exponential', + backoffMultiplier: 2, + jitter: 0.2, // 20% jitter + collectMetrics: false, // No Prometheus metrics + operationName: 'message_decryption', + shouldRetry: (error: Error, attempt: number) => { + const errorMsg = error?.message || '' + + // Always retry on session record errors (session might be syncing) + if (DECRYPTION_RETRY_CONFIG.sessionRecordErrors.some(err => errorMsg.includes(err))) { + return attempt < 3 // Retry up to 3 times + } + + // Don't retry on corrupted session errors (need cleanup first) + if (DECRYPTION_RETRY_CONFIG.corruptedSessionErrors.some(err => errorMsg.includes(err))) { + return false + } + + // Retry other transient errors + return attempt < 2 // Retry up to 2 times for unknown errors + } } export const NACK_REASONS = { @@ -72,7 +108,8 @@ export const NACK_REASONS = { UnhandledError: 500, UnsupportedAdminRevoke: 550, UnsupportedLIDGroup: 551, - DBOperationFailed: 552 + DBOperationFailed: 552, + CorruptedSession: 553 } type MessageType = @@ -281,21 +318,44 @@ export const decryptMessageNode = ( try { const e2eType = tag === 'plaintext' ? 'plaintext' : attrs.type + // Wrap decryption in retry logic for transient failures switch (e2eType) { case 'skmsg': - msgBuffer = await repository.decryptGroupMessage({ - group: sender, - authorJid: author, - msg: content - }) + msgBuffer = await retry( + () => repository.decryptGroupMessage({ + group: sender, + authorJid: author, + msg: content + }), + { + ...DECRYPTION_RETRY_OPTIONS, + onRetry: (error, attempt, delay) => { + logger.debug( + { error: error.message, attempt, delay, group: sender, author }, + 'Retrying group message decryption' + ) + } + } + ) break case 'pkmsg': case 'msg': - msgBuffer = await repository.decryptMessage({ - jid: decryptionJid, - type: e2eType, - ciphertext: content - }) + msgBuffer = await retry( + () => repository.decryptMessage({ + jid: decryptionJid, + type: e2eType, + ciphertext: content + }), + { + ...DECRYPTION_RETRY_OPTIONS, + onRetry: (error, attempt, delay) => { + logger.debug( + { error: error.message, attempt, delay, jid: decryptionJid, type: e2eType }, + 'Retrying message decryption' + ) + } + } + ) break case 'plaintext': msgBuffer = content @@ -326,16 +386,44 @@ export const decryptMessageNode = ( fullMessage.message = msg } } catch (err: any) { + const isCorrupted = isCorruptedSessionError(err) + const isSessionRecord = isSessionRecordError(err) + const errorContext = { key: fullMessage.key, err, messageType: tag === 'plaintext' ? 'plaintext' : attrs.type, sender, author, - isSessionRecordError: isSessionRecordError(err) + decryptionJid, + isSessionRecordError: isSessionRecord, + isCorruptedSession: isCorrupted } - logger.error(errorContext, 'failed to decrypt message') + if (isCorrupted) { + logger.error( + errorContext, + 'โš ๏ธ Corrupted session detected - Bad MAC or MessageCounter error.' + ) + + // Automatic cleanup of corrupted session (if enabled) + if (DEFAULT_SESSION_CLEANUP_CONFIG.autoCleanCorrupted) { + try { + await cleanupCorruptedSession(decryptionJid, repository, logger) + logger.info( + { decryptionJid, author }, + 'โœ… Corrupted session cleaned up automatically. New session will be created on next message.' + ) + } catch (cleanupErr) { + logger.error( + { decryptionJid, err: cleanupErr }, + 'Failed to cleanup corrupted session' + ) + } + } + } else { + logger.error(errorContext, 'failed to decrypt message') + } fullMessage.messageStubType = proto.WebMessageInfo.StubType.CIPHERTEXT fullMessage.messageStubParameters = [err.message.toString()] @@ -359,3 +447,55 @@ function isSessionRecordError(error: any): boolean { const errorMessage = error?.message || error?.toString() || '' return DECRYPTION_RETRY_CONFIG.sessionRecordErrors.some(errorPattern => errorMessage.includes(errorPattern)) } + +/** + * Utility function to check if an error indicates a corrupted session + * (Bad MAC, MessageCounterError, Key already used) + */ +export function isCorruptedSessionError(error: any): boolean { + const errorMessage = error?.message || error?.toString() || '' + return DECRYPTION_RETRY_CONFIG.corruptedSessionErrors.some(errorPattern => errorMessage.includes(errorPattern)) +} + +/** + * Clean up corrupted session by deleting all device sessions for a JID + * Signal Protocol will automatically recreate the session on next message + */ +async function cleanupCorruptedSession( + jid: string, + repository: SignalRepositoryWithLIDStore, + logger: ILogger +): Promise { + const { user, device } = jidDecode(jid) || {} + if (!user) { + logger.warn({ jid }, 'Cannot cleanup session - invalid JID') + return + } + + // Build list of JIDs to delete (primary + secondary devices) + const jidsToDelete: string[] = [] + + // Determine if this is a LID or PN + const isLID = jid.endsWith('@lid') + const domain = isLID ? 'lid' : 's.whatsapp.net' + + // Primary device (0) + jidsToDelete.push(`${user}@${domain}`) + + // Secondary devices (1-5 common range for Web/Desktop/etc) + for (let i = 1; i <= 5; i++) { + jidsToDelete.push(`${user}:${i}@${domain}`) + } + + // If specific device was identified and > 5, ensure it's included + if (device !== undefined && device > 5) { + jidsToDelete.push(`${user}:${device}@${domain}`) + } + + await repository.deleteSession(jidsToDelete) + + logger.debug( + { jid, user, device, deletedSessions: jidsToDelete }, + 'Deleted corrupted sessions for user' + ) +}