diff --git a/src/Defaults/index.ts b/src/Defaults/index.ts index e8d3395c..e5cc38fc 100644 --- a/src/Defaults/index.ts +++ b/src/Defaults/index.ts @@ -106,7 +106,11 @@ export const DEFAULT_CONNECTION_CONFIG: SocketConfig = { // SocketClient: 20 core events (connection, messaging, presence, groups, calls, etc.) // + 20 dynamic listeners (user handlers, plugins) // + 10 buffer slots for high-load scenarios = 50 total - maxSocketClientListeners: 50 + maxSocketClientListeners: 50, + // Unified session telemetry (reduces detection of unofficial clients) + // NOTE: undefined means "check env var first, then default to true" + // This allows BAILEYS_UNIFIED_SESSION_ENABLED env var to have precedence + enableUnifiedSession: undefined } export const MEDIA_PATH_MAP: { [T in MediaType]?: string } = { @@ -189,3 +193,39 @@ export const DEFAULT_CACHE_MAX_KEYS = { // Re-export retry constants for backwards compatibility // Actual definitions are in retry-utils.ts to avoid ESM initialization order issues export { RETRY_BACKOFF_DELAYS, RETRY_JITTER_FACTOR } from '../Utils/retry-utils' + +// ============================================ +// Time Constants +// ============================================ + +/** + * Time constants in milliseconds for various timing calculations. + * Used by unified session, rate limiting, and other time-based features. + * + * @example + * ```typescript + * import { TimeMs } from './Defaults' + * + * // Calculate 3 days in milliseconds + * const threeDays = 3 * TimeMs.Day + * + * // Check if 1 week has passed + * if (Date.now() - lastUpdate > TimeMs.Week) { + * // do something + * } + * ``` + */ +export const TimeMs = { + /** One second in milliseconds (1,000) */ + Second: 1_000, + /** One minute in milliseconds (60,000) */ + Minute: 60_000, + /** One hour in milliseconds (3,600,000) */ + Hour: 3_600_000, + /** One day in milliseconds (86,400,000) */ + Day: 86_400_000, + /** One week in milliseconds (604,800,000) */ + Week: 604_800_000, +} as const + +export type TimeMsKey = keyof typeof TimeMs diff --git a/src/Socket/chats.ts b/src/Socket/chats.ts index 4aede1b1..3c1ab653 100644 --- a/src/Socket/chats.ts +++ b/src/Socket/chats.ts @@ -68,7 +68,7 @@ export const makeChatsSocket = (config: SocketConfig) => { getMessage } = config const sock = makeSocket(config) - const { ev, ws, authState, generateMessageTag, sendNode, query, signalRepository, onUnexpectedError } = sock + const { ev, ws, authState, generateMessageTag, sendNode, query, signalRepository, onUnexpectedError, sendUnifiedSession } = sock let privacySettings: { [_: string]: string } | undefined @@ -674,6 +674,14 @@ export const makeChatsSocket = (config: SocketConfig) => { type } }) + + // Send unified_session telemetry when going online + // This mimics official WhatsApp Web client behavior + if (type === 'available') { + sendUnifiedSession('presence').catch(err => { + logger.debug({ err }, 'Failed to send unified_session on presence available') + }) + } } else { const { server } = jidDecode(toJid)! const isLid = server === 'lid' diff --git a/src/Socket/communities.ts b/src/Socket/communities.ts index 7cfabfd2..0d99e9ea 100644 --- a/src/Socket/communities.ts +++ b/src/Socket/communities.ts @@ -78,11 +78,11 @@ export const makeCommunitiesSocket = (config: SocketConfig) => { } async function parseGroupResult(node: BinaryNode) { - logger.info({ node }, 'parseGroupResult') + logger.debug({ nodeTag: node.tag }, 'parseGroupResult') const groupNode = getBinaryNodeChild(node, 'group') if (groupNode) { try { - logger.info({ groupNode }, 'groupNode') + logger.debug({ groupId: groupNode.attrs?.id }, 'parsing group metadata') const metadata = await sock.groupMetadata(`${groupNode.attrs.id}@g.us`) return metadata ? metadata : Optional.empty() } catch (error) { diff --git a/src/Socket/newsletter.ts b/src/Socket/newsletter.ts index 00c8555c..dac16139 100644 --- a/src/Socket/newsletter.ts +++ b/src/Socket/newsletter.ts @@ -7,21 +7,32 @@ import { makeGroupsSocket } from './groups' import { executeWMexQuery as genericExecuteWMexQuery } from './mex' const parseNewsletterCreateResponse = (response: NewsletterCreateResponse): NewsletterMetadata => { + // Validate response structure before destructuring + if (!response?.id || !response?.thread_metadata) { + throw new Error('Invalid newsletter response: missing id or thread_metadata') + } + const { id, thread_metadata: thread, viewer_metadata: viewer } = response + + // Validate required thread metadata fields + if (!thread.name?.text) { + throw new Error('Invalid newsletter response: missing thread name') + } + return { id: id, owner: undefined, name: thread.name.text, - creation_time: parseInt(thread.creation_time, 10), - description: thread.description.text, - invite: thread.invite, - subscribers: parseInt(thread.subscribers_count, 10), + creation_time: parseInt(thread.creation_time, 10) || 0, + description: thread.description?.text || '', + invite: thread.invite || '', + subscribers: parseInt(thread.subscribers_count, 10) || 0, verification: thread.verification, - picture: { - id: thread.picture.id, - directPath: thread.picture.direct_path - }, - mute_state: viewer.mute + picture: thread.picture ? { + id: thread.picture.id || '', + directPath: thread.picture.direct_path || '' + } : { id: '', directPath: '' }, + mute_state: viewer?.mute } } diff --git a/src/Socket/socket.ts b/src/Socket/socket.ts index 2d3b9f2c..570f6b68 100644 --- a/src/Socket/socket.ts +++ b/src/Socket/socket.ts @@ -60,6 +60,12 @@ import { incrementActiveConnections, decrementActiveConnections } from '../Utils/prometheus-metrics' +import { + createUnifiedSessionManager, + extractServerTime, + shouldEnableUnifiedSession, + type UnifiedSessionManager +} from '../Utils/unified-session' import { BinaryInfo } from '../WAM/BinaryInfo.js' import { USyncQuery, USyncUser } from '../WAUSync/' import { WebSocketClient } from './Client' @@ -87,9 +93,17 @@ export const makeSocket = (config: SocketConfig) => { enableCircuitBreaker = true, queryCircuitBreaker: queryCircuitBreakerConfig, connectionCircuitBreaker: connectionCircuitBreakerConfig, - preKeyCircuitBreaker: preKeyCircuitBreakerConfig + preKeyCircuitBreaker: preKeyCircuitBreakerConfig, + // If enableUnifiedSession is explicitly set (true/false), use it + // Otherwise (undefined), check env var, then default to true + enableUnifiedSession: enableUnifiedSessionConfig } = config + // Resolve enableUnifiedSession: explicit config > env var > default (true) + const enableUnifiedSession = enableUnifiedSessionConfig !== undefined + ? enableUnifiedSessionConfig + : shouldEnableUnifiedSession() + // Initialize circuit breakers if enabled let queryCircuitBreaker: CircuitBreaker | undefined let connectionCircuitBreaker: CircuitBreaker | undefined @@ -141,6 +155,9 @@ export const makeSocket = (config: SocketConfig) => { logger.info('Circuit breakers initialized for socket operations') } + // Unified Session Manager will be initialized after sendNode is defined + let unifiedSessionManager: UnifiedSessionManager | undefined + const publicWAMBuffer = new BinaryInfo() const uqTagId = generateMdTagPrefix() @@ -220,6 +237,28 @@ export const makeSocket = (config: SocketConfig) => { return sendRawMessage(buff) } + // Initialize Unified Session Manager now that sendNode is defined + // (single initialization to avoid duplicating circuit breakers and state) + if (enableUnifiedSession) { + const sendNodeForSession = async (node: BinaryNode): Promise => { + await sendNode(node) + } + unifiedSessionManager = createUnifiedSessionManager({ + enabled: true, + logger, + enableCircuitBreaker, + sendNode: sendNodeForSession + }) + logger.info('Unified session manager initialized') + } + + /** Send unified_session telemetry */ + const sendUnifiedSession = async (trigger: 'login' | 'pairing' | 'presence' | 'manual' = 'manual'): Promise => { + if (unifiedSessionManager) { + await unifiedSessionManager.send(trigger) + } + } + /** * Wait for a message with a certain tag to be received * @param msgId the message tag to await @@ -700,6 +739,13 @@ export const makeSocket = (config: SocketConfig) => { if (!(frame instanceof Uint8Array)) { const msgId = frame.attrs.id + // Update server time offset from any node with timestamp 't' + // This keeps the offset accurate even after long connections + const serverTime = extractServerTime(frame) + if (serverTime) { + unifiedSessionManager?.updateServerTimeOffset(serverTime) + } + if (logger.level === 'trace') { logger.trace({ xml: binaryNodeToString(frame), msg: 'recv xml' }) } @@ -783,6 +829,9 @@ export const makeSocket = (config: SocketConfig) => { connectionCircuitBreaker?.destroy() preKeyCircuitBreaker?.destroy() + // Clean up unified session manager + unifiedSessionManager?.destroy() + ws.removeAllListeners('close') ws.removeAllListeners('open') ws.removeAllListeners('message') @@ -1061,6 +1110,11 @@ export const makeSocket = (config: SocketConfig) => { ev.emit('connection.update', { isNewLogin: true, qr: undefined }) await sendNode(reply) + + // Send unified_session telemetry on successful pairing + sendUnifiedSession('pairing').catch(err => { + logger.debug({ err }, 'Failed to send unified_session on pairing') + }) } catch (error: any) { logger.info({ trace: error.stack }, 'error in pairing') void end(error) @@ -1093,6 +1147,17 @@ export const makeSocket = (config: SocketConfig) => { recordConnectionAttempt('success') incrementActiveConnections() + // Update server time offset from success node + const serverTime = extractServerTime(node) + if (serverTime) { + unifiedSessionManager?.updateServerTimeOffset(serverTime) + } + + // Send unified_session telemetry on successful login + sendUnifiedSession('login').catch(err => { + logger.debug({ err }, 'Failed to send unified_session on login') + }) + if (node.attrs.lid && authState.creds.me?.id) { const myLID = node.attrs.lid process.nextTick(async () => { @@ -1247,6 +1312,15 @@ export const makeSocket = (config: SocketConfig) => { connectionCircuitBreaker?.reset() preKeyCircuitBreaker?.reset() logger.info('All circuit breakers reset to closed state') + }, + // Unified Session Telemetry + /** Send unified_session telemetry manually */ + sendUnifiedSession, + /** Get unified session manager state (for debugging) */ + getUnifiedSessionState: () => unifiedSessionManager?.getState(), + /** Update server time offset (call when receiving server timestamps) */ + updateServerTimeOffset: (serverTime: string | number) => { + unifiedSessionManager?.updateServerTimeOffset(serverTime) } } } diff --git a/src/Types/Socket.ts b/src/Types/Socket.ts index cd4dfb43..227cc62e 100644 --- a/src/Types/Socket.ts +++ b/src/Types/Socket.ts @@ -229,4 +229,27 @@ export type SocketConfig = { * WARNING: Setting to 0 disables limit and allows potential memory leaks! */ maxSocketClientListeners?: number + + // === Unified Session Telemetry === + + /** + * Enable unified_session telemetry to reduce detection of unofficial clients. + * + * When enabled, sends time-based session identifiers that mimic official + * WhatsApp Web client behavior. This may help reduce "Your account may be + * at risk" warnings, though effectiveness is not guaranteed. + * + * Telemetry is sent at specific trigger points: + * - After successful login + * - After successful pairing + * - When sending 'available' presence + * + * Can also be controlled via environment variable: + * BAILEYS_UNIFIED_SESSION_ENABLED=true|false + * + * @default true + * @see https://github.com/tulir/whatsmeow/pull/1057 + * @see https://github.com/WhiskeySockets/Baileys/pull/2294 + */ + enableUnifiedSession?: boolean } diff --git a/src/Utils/health-status.ts b/src/Utils/health-status.ts new file mode 100644 index 00000000..25a92033 --- /dev/null +++ b/src/Utils/health-status.ts @@ -0,0 +1,209 @@ +/** + * Health Status Utilities + * + * Provides health check and status information for monitoring and k8s probes. + * + * @module Utils/health-status + */ + +import { getVersionCacheStatus } from './version-cache.js' +import { globalCircuitRegistry } from './circuit-breaker.js' + +/** + * Circuit breaker health information + */ +export interface CircuitBreakerHealth { + name: string + state: 'closed' | 'open' | 'half-open' + failures: number + successes: number + totalCalls: number +} + +/** + * Cache health information + */ +export interface CacheHealth { + versionCache: { + hasCache: boolean + isExpired: boolean + ageMs: number | null + source: string | null + } +} + +/** + * Overall health status + */ +export interface HealthStatus { + status: 'healthy' | 'degraded' | 'unhealthy' + timestamp: number + uptime: number + version: string + circuitBreakers: CircuitBreakerHealth[] + cache: CacheHealth + checks: { + name: string + status: 'pass' | 'warn' | 'fail' + message?: string + }[] +} + +/** + * Get the current health status of the Baileys instance. + * + * Useful for: + * - Kubernetes liveness/readiness probes + * - Load balancer health checks + * - Monitoring dashboards + * + * @returns HealthStatus object with detailed status information + * + * @example + * ```typescript + * import { getHealthStatus } from '@whiskeysockets/baileys' + * + * // Simple health check endpoint + * app.get('/health', (req, res) => { + * const health = getHealthStatus() + * const statusCode = health.status === 'healthy' ? 200 : + * health.status === 'degraded' ? 200 : 503 + * res.status(statusCode).json(health) + * }) + * + * // Kubernetes probe + * app.get('/healthz', (req, res) => { + * const health = getHealthStatus() + * res.status(health.status !== 'unhealthy' ? 200 : 503).send(health.status) + * }) + * ``` + */ +export function getHealthStatus(): HealthStatus { + const checks: HealthStatus['checks'] = [] + let overallStatus: HealthStatus['status'] = 'healthy' + + // 1. Check version cache + const versionCacheStatus = getVersionCacheStatus() + if (!versionCacheStatus.hasCache) { + checks.push({ + name: 'version_cache', + status: 'warn', + message: 'No version cache available' + }) + if (overallStatus === 'healthy') overallStatus = 'degraded' + } else if (versionCacheStatus.isExpired) { + checks.push({ + name: 'version_cache', + status: 'warn', + message: 'Version cache is expired' + }) + if (overallStatus === 'healthy') overallStatus = 'degraded' + } else { + checks.push({ + name: 'version_cache', + status: 'pass' + }) + } + + // 2. Check circuit breakers + const circuitBreakers: CircuitBreakerHealth[] = [] + let openCircuits = 0 + + for (const [name, breaker] of globalCircuitRegistry.getAll()) { + const stats = breaker.getStats() + const state = breaker.getState() + + circuitBreakers.push({ + name, + state, + failures: stats.totalFailures, + successes: stats.totalSuccesses, + totalCalls: stats.totalCalls + }) + + if (state === 'open') { + openCircuits++ + } + } + + if (openCircuits > 0) { + checks.push({ + name: 'circuit_breakers', + status: openCircuits > 2 ? 'fail' : 'warn', + message: `${openCircuits} circuit breaker(s) are open` + }) + if (openCircuits > 2) { + overallStatus = 'unhealthy' + } else if (overallStatus === 'healthy') { + overallStatus = 'degraded' + } + } else { + checks.push({ + name: 'circuit_breakers', + status: 'pass' + }) + } + + // 3. Check memory usage (warn if > 90%) + const memUsage = process.memoryUsage() + const heapUsedPercent = (memUsage.heapUsed / memUsage.heapTotal) * 100 + + if (heapUsedPercent > 90) { + checks.push({ + name: 'memory', + status: 'warn', + message: `Heap usage is ${heapUsedPercent.toFixed(1)}%` + }) + if (overallStatus === 'healthy') overallStatus = 'degraded' + } else { + checks.push({ + name: 'memory', + status: 'pass' + }) + } + + return { + status: overallStatus, + timestamp: Date.now(), + uptime: process.uptime(), + version: process.env.npm_package_version || 'unknown', + circuitBreakers, + cache: { + versionCache: { + hasCache: versionCacheStatus.hasCache, + isExpired: versionCacheStatus.isExpired, + ageMs: versionCacheStatus.age, + source: versionCacheStatus.source + } + }, + checks + } +} + +/** + * Simple health check - returns true if system is healthy or degraded. + * Use this for basic liveness probes. + * + * @returns true if healthy or degraded, false if unhealthy + */ +export function isHealthy(): boolean { + const status = getHealthStatus() + return status.status !== 'unhealthy' +} + +/** + * Get a simple status string for minimal health endpoints. + * + * @returns 'ok', 'degraded', or 'error' + */ +export function getSimpleHealthStatus(): 'ok' | 'degraded' | 'error' { + const status = getHealthStatus() + switch (status.status) { + case 'healthy': + return 'ok' + case 'degraded': + return 'degraded' + case 'unhealthy': + return 'error' + } +} diff --git a/src/Utils/index.ts b/src/Utils/index.ts index 70bd687f..0b376f3d 100644 --- a/src/Utils/index.ts +++ b/src/Utils/index.ts @@ -36,8 +36,14 @@ export * from './cache-utils' export * from './circuit-breaker' export * from './retry-utils' +// Telemetry and detection mitigation +export * from './unified-session' + // Version management export * from './version-cache' +// Health monitoring +export * from './health-status' + // Event streaming export * from './baileys-event-stream' diff --git a/src/Utils/unified-session.ts b/src/Utils/unified-session.ts new file mode 100644 index 00000000..06948242 --- /dev/null +++ b/src/Utils/unified-session.ts @@ -0,0 +1,414 @@ +/** + * Unified Session Telemetry Implementation + * + * This module implements WhatsApp's unified_session telemetry feature to reduce + * detection of unofficial clients. The implementation is inspired by: + * - whatsmeow PR #1057 (Go implementation) + * - Baileys PR #2294 (TypeScript implementation) + * + * The unified_session is a time-based identifier that mimics the behavior of + * official WhatsApp Web clients, potentially reducing account restriction warnings. + * + * @module Utils/unified-session + * @see https://github.com/tulir/whatsmeow/pull/1057 + * @see https://github.com/WhiskeySockets/Baileys/pull/2294 + */ + +import { metrics } from './prometheus-metrics.js' +import type { ILogger } from './logger.js' +import type { BinaryNode } from '../WABinary/types.js' +import { + CircuitBreaker, + CircuitOpenError, + createConnectionCircuitBreaker +} from './circuit-breaker.js' + +// ============================================ +// Time Constants (Enterprise-grade) +// ============================================ + +// Import TimeMs from Defaults (single source of truth) +import { TimeMs } from '../Defaults/index.js' + +// Re-export for convenience (but not as main export to avoid conflicts) +export { TimeMs } + +/** + * Offset for session ID calculation (3 days in ms). + * This matches the WhatsApp Web official client behavior. + */ +const SESSION_OFFSET_MS = 3 * TimeMs.Day + +// ============================================ +// Types +// ============================================ + +/** + * Unified Session Manager options + */ +export interface UnifiedSessionOptions { + /** Whether unified session telemetry is enabled */ + enabled?: boolean + /** Logger instance for debugging */ + logger?: ILogger + /** Enable circuit breaker protection for send operations */ + enableCircuitBreaker?: boolean + /** Function to send binary nodes to WhatsApp */ + sendNode?: (node: BinaryNode) => Promise +} + +/** + * Unified Session state + */ +export interface UnifiedSessionState { + /** Server time offset in milliseconds (server time - local time) */ + serverTimeOffset: number + /** Last time unified_session was sent (Unix timestamp ms) */ + lastSentTime: number + /** Total number of unified_session messages sent */ + sendCount: number + /** Whether the session manager is initialized */ + isInitialized: boolean +} + +/** + * Trigger types for unified session sending + */ +export type UnifiedSessionTrigger = 'login' | 'pairing' | 'presence' | 'manual' + +// ============================================ +// Core Implementation +// ============================================ + +/** + * Unified Session Manager + * + * Manages the unified_session telemetry feature with: + * - Server time synchronization + * - Rate limiting (prevents spam) + * - Circuit breaker protection + * - Prometheus metrics integration + * - Structured logging + * + * @example + * ```typescript + * const sessionManager = new UnifiedSessionManager({ + * enabled: true, + * logger: myLogger, + * sendNode: (node) => sock.sendNode(node) + * }) + * + * // Update server time offset when receiving server timestamp + * sessionManager.updateServerTimeOffset(serverTimeAttr) + * + * // Send unified_session on login + * await sessionManager.send('login') + * ``` + */ +export class UnifiedSessionManager { + private state: UnifiedSessionState = { + serverTimeOffset: 0, + lastSentTime: 0, + sendCount: 0, + isInitialized: false + } + + private readonly options: Required + private circuitBreaker: CircuitBreaker | null = null + + /** Minimum interval between unified_session sends (1 minute) */ + private static readonly MIN_SEND_INTERVAL_MS = TimeMs.Minute + + constructor(options: UnifiedSessionOptions = {}) { + this.options = { + enabled: options.enabled ?? true, + logger: options.logger ?? console as unknown as ILogger, + enableCircuitBreaker: options.enableCircuitBreaker ?? true, + sendNode: options.sendNode ?? (async () => { + throw new Error('sendNode function not configured') + }) + } + + // Initialize circuit breaker if enabled + if (this.options.enableCircuitBreaker) { + this.circuitBreaker = createConnectionCircuitBreaker({ + name: 'unified-session', + failureThreshold: 3, + failureWindow: 60000, + resetTimeout: 30000, + successThreshold: 1, + timeout: 10000, + onStateChange: (from, to) => { + this.options.logger.debug?.({ from, to }, 'Unified session circuit breaker state changed') + }, + onOpen: () => { + this.options.logger.warn?.('Unified session circuit breaker OPENED') + metrics.circuitBreakerTrips?.inc({ name: 'unified-session' }) + } + }) + } + + this.state.isInitialized = true + this.options.logger.debug?.('UnifiedSessionManager initialized') + } + + /** + * Update the server time offset from a received server timestamp. + * + * WhatsApp includes a 't' attribute in some nodes containing the server's + * Unix timestamp (in seconds). We use this to calculate the offset between + * server time and local time. + * + * @param serverTime - Server timestamp (seconds) from node.attrs.t + */ + updateServerTimeOffset(serverTime: string | number | undefined): void { + if (serverTime === undefined || serverTime === null) { + return + } + + const serverTimeNum = typeof serverTime === 'string' + ? parseInt(serverTime, 10) + : serverTime + + if (isNaN(serverTimeNum) || serverTimeNum <= 0) { + this.options.logger.debug?.( + { serverTime }, + 'Invalid server time received, ignoring' + ) + return + } + + // Server time is in seconds, convert to ms + const serverTimeMs = serverTimeNum * 1000 + const localTimeMs = Date.now() + const newOffset = serverTimeMs - localTimeMs + + // Only update if the offset changed significantly (>1 second) + if (Math.abs(newOffset - this.state.serverTimeOffset) > 1000) { + const oldOffset = this.state.serverTimeOffset + this.state.serverTimeOffset = newOffset + + this.options.logger.debug?.( + { oldOffset, newOffset, serverTime: serverTimeNum }, + 'Server time offset updated' + ) + + // Record metric + metrics.socketEvents?.inc({ event: 'server_time_sync' }) + } + } + + /** + * Calculate the unified session ID. + * + * The algorithm matches WhatsApp Web's official implementation: + * - Takes current time adjusted by server offset + * - Adds 3-day offset + * - Modulo 7 days (one week cycle) + * - Returns as string + * + * @returns The unified session ID as a string + */ + getSessionId(): string { + const adjustedTime = Date.now() + this.state.serverTimeOffset + const sessionId = (adjustedTime + SESSION_OFFSET_MS) % TimeMs.Week + return String(Math.floor(sessionId)) + } + + /** + * Check if enough time has passed since the last send. + * Prevents spamming the server with unified_session messages. + */ + private canSend(): boolean { + const timeSinceLastSend = Date.now() - this.state.lastSentTime + return timeSinceLastSend >= UnifiedSessionManager.MIN_SEND_INTERVAL_MS + } + + /** + * Send the unified_session telemetry to WhatsApp. + * + * This should be called at specific trigger points: + * - After successful login (CB:success) + * - After successful pairing (CB:iq,,pair-success) + * - When sending 'available' presence + * + * @param trigger - What triggered this send (for logging/metrics) + * @returns Promise that resolves when sent, or void if skipped + */ + async send(trigger: UnifiedSessionTrigger = 'manual'): Promise { + // Check if enabled + if (!this.options.enabled) { + this.options.logger.trace?.('Unified session is disabled, skipping') + return + } + + // Rate limiting (except for login which always sends) + if (trigger !== 'login' && !this.canSend()) { + this.options.logger.trace?.( + { trigger, lastSent: this.state.lastSentTime }, + 'Unified session rate limited, skipping' + ) + return + } + + const sessionId = this.getSessionId() + const startTime = Date.now() + + const node: BinaryNode = { + tag: 'ib', + attrs: {}, + content: [{ + tag: 'unified_session', + attrs: { id: sessionId } + }] + } + + const sendOperation = async (): Promise => { + await this.options.sendNode(node) + + // Update state on success + this.state.lastSentTime = Date.now() + this.state.sendCount++ + + const latency = Date.now() - startTime + + this.options.logger.debug?.( + { sessionId, trigger, latency, sendCount: this.state.sendCount }, + 'Unified session telemetry sent' + ) + + // Record metrics + metrics.socketEvents?.inc({ event: 'unified_session_sent' }) + } + + try { + // Execute with circuit breaker if available + if (this.circuitBreaker) { + await this.circuitBreaker.execute(sendOperation) + } else { + await sendOperation() + } + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error) + + if (error instanceof CircuitOpenError) { + this.options.logger.warn?.( + { trigger, circuitState: error.state }, + 'Unified session blocked by circuit breaker' + ) + } else { + this.options.logger.warn?.( + { trigger, error: errorMessage }, + 'Failed to send unified session telemetry' + ) + } + + // Record failure metric + metrics.errors?.inc({ category: 'unified_session', code: 'send_failed' }) + + // Don't rethrow - unified_session is non-critical + } + } + + /** + * Get the current state for debugging/monitoring. + */ + getState(): Readonly { + return { ...this.state } + } + + /** + * Reset the manager state (useful for testing or reconnection). + */ + reset(): void { + this.state = { + serverTimeOffset: 0, + lastSentTime: 0, + sendCount: 0, + isInitialized: true + } + this.circuitBreaker?.reset() + this.options.logger.debug?.('UnifiedSessionManager reset') + } + + /** + * Destroy the manager and clean up resources. + */ + destroy(): void { + this.circuitBreaker?.destroy() + this.state.isInitialized = false + this.options.logger.debug?.('UnifiedSessionManager destroyed') + } +} + +// ============================================ +// Factory Function +// ============================================ + +/** + * Create a new UnifiedSessionManager instance. + * + * @param options - Configuration options + * @returns A new UnifiedSessionManager instance + * + * @example + * ```typescript + * const sessionManager = createUnifiedSessionManager({ + * enabled: config.enableUnifiedSession, + * logger: config.logger, + * sendNode: sock.sendNode + * }) + * ``` + */ +export function createUnifiedSessionManager( + options: UnifiedSessionOptions = {} +): UnifiedSessionManager { + return new UnifiedSessionManager(options) +} + +// ============================================ +// Utility Functions +// ============================================ + +/** + * Extract server time from a binary node's attributes. + * WhatsApp includes 't' attribute in various nodes. + * + * @param node - Binary node with potential time attribute + * @returns Server time in seconds, or undefined if not present + */ +export function extractServerTime(node: BinaryNode): number | undefined { + const timeAttr = node.attrs?.t + if (timeAttr === undefined || timeAttr === null) { + return undefined + } + + const parsed = typeof timeAttr === 'string' + ? parseInt(timeAttr, 10) + : typeof timeAttr === 'number' + ? timeAttr + : undefined + + if (parsed === undefined || isNaN(parsed) || parsed <= 0) { + return undefined + } + + return parsed +} + +/** + * Check if unified_session should be enabled based on environment. + * Can be controlled via environment variable for testing. + * + * @returns true if unified_session should be enabled + */ +export function shouldEnableUnifiedSession(): boolean { + const envValue = process.env.BAILEYS_UNIFIED_SESSION_ENABLED + if (envValue !== undefined) { + return envValue.toLowerCase() === 'true' || envValue === '1' + } + // Default: enabled + return true +} + +export default UnifiedSessionManager diff --git a/src/Utils/version-cache.ts b/src/Utils/version-cache.ts index 9941d890..a5cd9679 100644 --- a/src/Utils/version-cache.ts +++ b/src/Utils/version-cache.ts @@ -86,11 +86,16 @@ async function readCacheFile(filePath: string): Promise { +async function writeCacheFile( + filePath: string, + entry: VersionCacheEntry, + logger?: VersionCacheLogger +): Promise { try { await fs.writeFile(filePath, JSON.stringify(entry, null, 2), 'utf-8') - } catch { - // Ignore write errors - cache is optional + } catch (error) { + // Log write errors for debugging - cache is optional but failures should be visible + logger?.warn({ error, filePath }, 'Failed to write version cache file') } } @@ -124,7 +129,7 @@ async function fetchVersionOnce( memoryCache = entry // Persist to file (async, don't wait) - writeCacheFile(cacheFilePath, entry).catch(() => {}) + writeCacheFile(cacheFilePath, entry, logger).catch(() => {}) logger?.info( { version: entry.version, source: entry.source }, diff --git a/src/__tests__/Utils/unified-session.test.ts b/src/__tests__/Utils/unified-session.test.ts new file mode 100644 index 00000000..68e859d8 --- /dev/null +++ b/src/__tests__/Utils/unified-session.test.ts @@ -0,0 +1,389 @@ +/** + * Unit tests for unified-session.ts + * + * Tests the UnifiedSessionManager implementation that mimics + * official WhatsApp Web client telemetry behavior. + */ + +import { describe, it, expect, beforeEach, afterEach, jest } from '@jest/globals' +import type { BinaryNode } from '../../WABinary/types.js' +import { + UnifiedSessionManager, + createUnifiedSessionManager, + extractServerTime, + shouldEnableUnifiedSession, + type UnifiedSessionOptions +} from '../../Utils/unified-session.js' +import { TimeMs } from '../../Defaults/index.js' + +// Mock the prometheus metrics to avoid side effects +jest.mock('../../Utils/prometheus-metrics.js', () => ({ + metrics: { + socketEvents: { inc: jest.fn() }, + errors: { inc: jest.fn() }, + circuitBreakerTrips: { inc: jest.fn() } + } +})) + +// Mock logger +const createMockLogger = () => ({ + trace: jest.fn(), + debug: jest.fn(), + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + level: 'debug' +}) + +// Type for mock sendNode function +type MockSendNode = jest.Mock<(node: BinaryNode) => Promise> + +describe('UnifiedSessionManager', () => { + let manager: UnifiedSessionManager + let mockLogger: ReturnType + let mockSendNode: MockSendNode + + beforeEach(() => { + mockLogger = createMockLogger() + mockSendNode = jest.fn<(node: BinaryNode) => Promise>().mockResolvedValue(undefined) + + manager = createUnifiedSessionManager({ + enabled: true, + logger: mockLogger as any, + enableCircuitBreaker: false, // Disable for simpler testing + sendNode: mockSendNode + }) + }) + + afterEach(() => { + manager.destroy() + jest.clearAllMocks() + }) + + describe('TimeMs constants', () => { + it('should have correct time values', () => { + expect(TimeMs.Second).toBe(1_000) + expect(TimeMs.Minute).toBe(60_000) + expect(TimeMs.Hour).toBe(3_600_000) + expect(TimeMs.Day).toBe(86_400_000) + expect(TimeMs.Week).toBe(604_800_000) + }) + + it('should have readonly constants (enforced by TypeScript)', () => { + // TypeScript's 'as const' ensures compile-time immutability + // At runtime, we verify the values are what we expect + expect(TimeMs.Second).toBe(1_000) + expect(TimeMs.Week).toBe(604_800_000) + // Verify the object structure is correct + expect(Object.keys(TimeMs)).toEqual(['Second', 'Minute', 'Hour', 'Day', 'Week']) + }) + }) + + describe('initialization', () => { + it('should initialize with default state', () => { + const state = manager.getState() + expect(state.isInitialized).toBe(true) + expect(state.serverTimeOffset).toBe(0) + expect(state.lastSentTime).toBe(0) + expect(state.sendCount).toBe(0) + }) + + it('should create with factory function', () => { + const newManager = createUnifiedSessionManager({ enabled: true }) + expect(newManager.getState().isInitialized).toBe(true) + newManager.destroy() + }) + + it('should respect enabled option', async () => { + const disabledManager = createUnifiedSessionManager({ + enabled: false, + sendNode: mockSendNode + }) + + await disabledManager.send('login') + + expect(mockSendNode).not.toHaveBeenCalled() + disabledManager.destroy() + }) + }) + + describe('getSessionId', () => { + it('should return a string', () => { + const sessionId = manager.getSessionId() + expect(typeof sessionId).toBe('string') + }) + + it('should return a numeric value as string', () => { + const sessionId = manager.getSessionId() + const parsed = parseInt(sessionId, 10) + expect(isNaN(parsed)).toBe(false) + expect(parsed).toBeGreaterThanOrEqual(0) + }) + + it('should return value within week range', () => { + const sessionId = manager.getSessionId() + const value = parseInt(sessionId, 10) + + // Session ID should be modulo 7 days (in ms) + expect(value).toBeLessThan(TimeMs.Week) + }) + + it('should change with server time offset', () => { + const id1 = manager.getSessionId() + + // Update offset by 1 hour + manager.updateServerTimeOffset(Math.floor(Date.now() / 1000) + 3600) + + const id2 = manager.getSessionId() + + // IDs should be different due to offset + expect(id1).not.toBe(id2) + }) + }) + + describe('updateServerTimeOffset', () => { + it('should update offset from string timestamp', () => { + const serverTime = Math.floor(Date.now() / 1000) + 100 + manager.updateServerTimeOffset(serverTime.toString()) + + const state = manager.getState() + // Offset should be approximately 100 seconds (100,000ms) + expect(Math.abs(state.serverTimeOffset - 100_000)).toBeLessThan(1000) + }) + + it('should update offset from number timestamp', () => { + const serverTime = Math.floor(Date.now() / 1000) - 50 + manager.updateServerTimeOffset(serverTime) + + const state = manager.getState() + // Offset should be approximately -50 seconds (-50,000ms) + expect(Math.abs(state.serverTimeOffset + 50_000)).toBeLessThan(1000) + }) + + it('should ignore undefined value', () => { + manager.updateServerTimeOffset(undefined) + expect(manager.getState().serverTimeOffset).toBe(0) + }) + + it('should ignore invalid values', () => { + manager.updateServerTimeOffset('invalid') + expect(manager.getState().serverTimeOffset).toBe(0) + + manager.updateServerTimeOffset(-1) + expect(manager.getState().serverTimeOffset).toBe(0) + + manager.updateServerTimeOffset(0) + expect(manager.getState().serverTimeOffset).toBe(0) + }) + + it('should only update on significant change (>1 second)', () => { + const serverTime = Math.floor(Date.now() / 1000) + 100 + manager.updateServerTimeOffset(serverTime) + const offset1 = manager.getState().serverTimeOffset + + // Small change (< 1 second) should not update + manager.updateServerTimeOffset(serverTime + 0.5) + const offset2 = manager.getState().serverTimeOffset + + expect(offset1).toBe(offset2) + }) + }) + + describe('send', () => { + it('should send unified_session node', async () => { + await manager.send('login') + + expect(mockSendNode).toHaveBeenCalledTimes(1) + + const sentNode = mockSendNode.mock.calls[0]?.[0] as BinaryNode | undefined + expect(sentNode).toBeDefined() + expect(sentNode!.tag).toBe('ib') + expect(Array.isArray(sentNode!.content)).toBe(true) + const content = sentNode!.content as BinaryNode[] + expect(content).toHaveLength(1) + expect(content[0]!.tag).toBe('unified_session') + expect(typeof content[0]!.attrs.id).toBe('string') + }) + + it('should update state after send', async () => { + expect(manager.getState().sendCount).toBe(0) + + await manager.send('login') + + const state = manager.getState() + expect(state.sendCount).toBe(1) + expect(state.lastSentTime).toBeGreaterThan(0) + }) + + it('should rate limit sends (except login)', async () => { + // First send should work + await manager.send('presence') + expect(mockSendNode).toHaveBeenCalledTimes(1) + + // Immediate second send should be skipped (rate limited) + await manager.send('presence') + expect(mockSendNode).toHaveBeenCalledTimes(1) + }) + + it('should not rate limit login trigger', async () => { + await manager.send('login') + expect(mockSendNode).toHaveBeenCalledTimes(1) + + // Login should always send, even immediately after + await manager.send('login') + expect(mockSendNode).toHaveBeenCalledTimes(2) + }) + + it('should not throw on send failure', async () => { + mockSendNode.mockRejectedValue(new Error('Network error')) + + // Should not throw + await expect(manager.send('login')).resolves.toBeUndefined() + + // Reset mock for subsequent tests + mockSendNode.mockResolvedValue(undefined) + }) + + it('should log warning on send failure', async () => { + mockSendNode.mockRejectedValue(new Error('Network error')) + + await manager.send('login') + + expect(mockLogger.warn).toHaveBeenCalled() + + // Reset mock for subsequent tests + mockSendNode.mockResolvedValue(undefined) + }) + }) + + describe('reset', () => { + it('should reset all state', async () => { + // Modify state + manager.updateServerTimeOffset(Math.floor(Date.now() / 1000) + 100) + await manager.send('login') + + const stateBefore = manager.getState() + expect(stateBefore.serverTimeOffset).not.toBe(0) + expect(stateBefore.sendCount).toBe(1) + + // Reset + manager.reset() + + const stateAfter = manager.getState() + expect(stateAfter.serverTimeOffset).toBe(0) + expect(stateAfter.sendCount).toBe(0) + expect(stateAfter.lastSentTime).toBe(0) + expect(stateAfter.isInitialized).toBe(true) + }) + }) + + describe('destroy', () => { + it('should mark as not initialized', () => { + manager.destroy() + expect(manager.getState().isInitialized).toBe(false) + }) + }) +}) + +describe('extractServerTime', () => { + it('should extract time from string attribute', () => { + const node = { tag: 'test', attrs: { t: '1234567890' } } + expect(extractServerTime(node as any)).toBe(1234567890) + }) + + it('should extract time from number attribute', () => { + const node = { tag: 'test', attrs: { t: 1234567890 } } + expect(extractServerTime(node as any)).toBe(1234567890) + }) + + it('should return undefined for missing attribute', () => { + const node = { tag: 'test', attrs: {} } + expect(extractServerTime(node as any)).toBeUndefined() + }) + + it('should return undefined for invalid values', () => { + expect(extractServerTime({ tag: 'test', attrs: { t: 'invalid' } } as any)).toBeUndefined() + expect(extractServerTime({ tag: 'test', attrs: { t: 0 } } as any)).toBeUndefined() + expect(extractServerTime({ tag: 'test', attrs: { t: -1 } } as any)).toBeUndefined() + }) +}) + +describe('shouldEnableUnifiedSession', () => { + const originalEnv = process.env.BAILEYS_UNIFIED_SESSION_ENABLED + + afterEach(() => { + if (originalEnv === undefined) { + delete process.env.BAILEYS_UNIFIED_SESSION_ENABLED + } else { + process.env.BAILEYS_UNIFIED_SESSION_ENABLED = originalEnv + } + }) + + it('should return true by default', () => { + delete process.env.BAILEYS_UNIFIED_SESSION_ENABLED + expect(shouldEnableUnifiedSession()).toBe(true) + }) + + it('should respect environment variable true', () => { + process.env.BAILEYS_UNIFIED_SESSION_ENABLED = 'true' + expect(shouldEnableUnifiedSession()).toBe(true) + + process.env.BAILEYS_UNIFIED_SESSION_ENABLED = '1' + expect(shouldEnableUnifiedSession()).toBe(true) + + process.env.BAILEYS_UNIFIED_SESSION_ENABLED = 'TRUE' + expect(shouldEnableUnifiedSession()).toBe(true) + }) + + it('should respect environment variable false', () => { + process.env.BAILEYS_UNIFIED_SESSION_ENABLED = 'false' + expect(shouldEnableUnifiedSession()).toBe(false) + + process.env.BAILEYS_UNIFIED_SESSION_ENABLED = '0' + expect(shouldEnableUnifiedSession()).toBe(false) + + process.env.BAILEYS_UNIFIED_SESSION_ENABLED = '' + expect(shouldEnableUnifiedSession()).toBe(false) + }) +}) + +describe('UnifiedSessionManager with CircuitBreaker', () => { + let manager: UnifiedSessionManager + let mockSendNode: MockSendNode + + beforeEach(() => { + mockSendNode = jest.fn<(node: BinaryNode) => Promise>().mockResolvedValue(undefined) + + manager = createUnifiedSessionManager({ + enabled: true, + logger: createMockLogger() as any, + enableCircuitBreaker: true, + sendNode: mockSendNode + }) + }) + + afterEach(() => { + manager.destroy() + }) + + it('should work with circuit breaker enabled', async () => { + await manager.send('login') + expect(mockSendNode).toHaveBeenCalledTimes(1) + }) + + it('should handle circuit breaker failures gracefully', async () => { + // Make send fail multiple times to trigger circuit breaker + let callCount = 0 + mockSendNode.mockImplementation(async () => { + callCount++ + if (callCount <= 3) { + throw new Error(`Fail ${callCount}`) + } + }) + + // These should not throw even with failures + await expect(manager.send('login')).resolves.toBeUndefined() + await expect(manager.send('login')).resolves.toBeUndefined() + await expect(manager.send('login')).resolves.toBeUndefined() + }) +})