improve: add detection and enhanced logging for corrupted Signal sess…

improve: add detection and enhanced logging for corrupted Signal sess…

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

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<typeof setInterval> | null = null
+	let initialTimeout: ReturnType<typeof setTimeout> | 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

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<string> => {
 	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({
+								msgBuffer = await retry(
-									group: sender,
+									() => repository.decryptGroupMessage({
-									authorJid: author,
+										group: sender,
-									msg: content
+										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({
+								msgBuffer = await retry(
-									jid: decryptionJid,
+									() => repository.decryptMessage({
-									type: e2eType,
+										jid: decryptionJid,
-									ciphertext: content
+										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<void> {
+	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'
+	)
+}