feat(socket): integrate circuit breaker with Socket operations

This PR integrates the circuit breaker pattern with Socket operations for improved resilience.

### Changes

- **SocketConfig** - Added circuit breaker configuration options:
  - `enableCircuitBreaker` - Enable/disable circuit breaker (default: true)
  - `queryCircuitBreaker` - Config for query operations
  - `connectionCircuitBreaker` - Config for connection operations
  - `preKeyCircuitBreaker` - Config for pre-key operations

- **Socket Integration** - Circuit breakers protect:
  - `query()` - All WhatsApp queries (most critical)
  - `sendRawMessage()` - Low-level WebSocket sends
  - `uploadPreKeys()` - Pre-key upload operations

- **New Socket Methods**:
  - `circuitBreakers` - Access to circuit breaker instances
  - `getCircuitBreakerStats()` - Get statistics for all breakers
  - `resetCircuitBreakers()` - Reset all breakers to closed state

### Circuit Breaker Features

- Sliding window failure tracking
- Automatic state transitions (closed → open → half-open)
- Configurable thresholds and timeouts
- Prometheus metrics integration
- Event callbacks for monitoring
- Automatic cleanup on connection close

src/Defaults/index.ts

@@ -77,7 +77,9 @@ export const DEFAULT_CONNECTION_CONFIG: SocketConfig = {
 	countryCode: 'US',
 	getMessage: async () => undefined,
 	cachedGroupMetadata: async () => undefined,
-	makeSignalRepository: makeLibSignalRepository
+	makeSignalRepository: makeLibSignalRepository,
+	// Circuit breaker configuration
+	enableCircuitBreaker: true
 }
 
 export const MEDIA_PATH_MAP: { [T in MediaType]?: string } = {

src/Socket/socket.ts

@@ -35,6 +35,12 @@ import {
 	xmppSignedPreKey
 } from '../Utils'
 import { getPlatformId } from '../Utils/browser-utils'
+import {
+	CircuitBreaker,
+	CircuitOpenError,
+	createConnectionCircuitBreaker,
+	createPreKeyCircuitBreaker
+} from '../Utils/circuit-breaker'
 import {
 	assertNodeErrorFree,
 	type BinaryNode,
@@ -71,9 +77,64 @@ export const makeSocket = (config: SocketConfig) => {
 		defaultQueryTimeoutMs,
 		transactionOpts,
 		qrTimeout,
-		makeSignalRepository
+		makeSignalRepository,
+		enableCircuitBreaker = true,
+		queryCircuitBreaker: queryCircuitBreakerConfig,
+		connectionCircuitBreaker: connectionCircuitBreakerConfig,
+		preKeyCircuitBreaker: preKeyCircuitBreakerConfig
 	} = config
 
+	// Initialize circuit breakers if enabled
+	let queryCircuitBreaker: CircuitBreaker | undefined
+	let connectionCircuitBreaker: CircuitBreaker | undefined
+	let preKeyCircuitBreaker: CircuitBreaker | undefined
+
+	if (enableCircuitBreaker) {
+		// Circuit breaker for query operations (most critical)
+		queryCircuitBreaker = createConnectionCircuitBreaker({
+			name: 'socket-query',
+			failureThreshold: 5,
+			failureWindow: 60000,
+			resetTimeout: 30000,
+			successThreshold: 2,
+			timeout: defaultQueryTimeoutMs || 60000,
+			onStateChange: (from, to) => {
+				logger.info({ from, to }, 'Query circuit breaker state changed')
+			},
+			onOpen: () => {
+				logger.warn('Query circuit breaker OPENED - blocking requests')
+			},
+			onClose: () => {
+				logger.info('Query circuit breaker CLOSED - resuming normal operation')
+			},
+			...queryCircuitBreakerConfig
+		})
+
+		// Circuit breaker for connection operations
+		connectionCircuitBreaker = createConnectionCircuitBreaker({
+			name: 'socket-connection',
+			failureThreshold: 3,
+			failureWindow: 30000,
+			resetTimeout: 60000,
+			successThreshold: 1,
+			onStateChange: (from, to) => {
+				logger.info({ from, to }, 'Connection circuit breaker state changed')
+			},
+			...connectionCircuitBreakerConfig
+		})
+
+		// Circuit breaker for pre-key operations
+		preKeyCircuitBreaker = createPreKeyCircuitBreaker({
+			name: 'socket-prekey',
+			onStateChange: (from, to) => {
+				logger.info({ from, to }, 'PreKey circuit breaker state changed')
+			},
+			...preKeyCircuitBreakerConfig
+		})
+
+		logger.info('Circuit breakers initialized for socket operations')
+	}
+
 	const publicWAMBuffer = new BinaryInfo()
 
 	const uqTagId = generateMdTagPrefix()
@@ -110,8 +171,8 @@ export const makeSocket = (config: SocketConfig) => {
 	ws.connect()
 
 	const sendPromise = promisify(ws.send)
-	/** send a raw buffer */
-	const sendRawMessage = async (data: Uint8Array | Buffer) => {
+	/** send a raw buffer (internal implementation) */
+	const sendRawMessageInternal = async (data: Uint8Array | Buffer) => {
 		if (!ws.isOpen) {
 			throw new Boom('Connection Closed', { statusCode: DisconnectReason.connectionClosed })
 		}
@@ -127,6 +188,22 @@ export const makeSocket = (config: SocketConfig) => {
 		})
 	}
 
+	/** send a raw buffer with circuit breaker protection */
+	const sendRawMessage = async (data: Uint8Array | Buffer) => {
+		if (connectionCircuitBreaker) {
+			try {
+				return await connectionCircuitBreaker.execute(() => sendRawMessageInternal(data))
+			} catch (error) {
+				if (error instanceof CircuitOpenError) {
+					logger.warn({ circuitName: error.circuitName }, 'Send blocked by connection circuit breaker')
+				}
+				throw error
+			}
+		}
+
+		return sendRawMessageInternal(data)
+	}
+
 	/** send a binary node */
 	const sendNode = (frame: BinaryNode) => {
 		if (logger.level === 'trace') {
@@ -185,7 +262,7 @@ export const makeSocket = (config: SocketConfig) => {
 	}
 
 	/** send a query, and wait for its response. auto-generates message ID if not provided */
-	const query = async (node: BinaryNode, timeoutMs?: number) => {
+	const queryInternal = async (node: BinaryNode, timeoutMs?: number) => {
 		if (!node.attrs.id) {
 			node.attrs.id = generateMessageTag()
 		}
@@ -206,6 +283,25 @@ export const makeSocket = (config: SocketConfig) => {
 		return result
 	}
 
+	/** send a query with circuit breaker protection */
+	const query = async (node: BinaryNode, timeoutMs?: number) => {
+		// If circuit breaker is enabled, wrap the query
+		if (queryCircuitBreaker) {
+			try {
+				return await queryCircuitBreaker.execute(() => queryInternal(node, timeoutMs))
+			} catch (error) {
+				// If circuit is open, log and rethrow with context
+				if (error instanceof CircuitOpenError) {
+					logger.warn({ circuitName: error.circuitName, state: error.state }, 'Query blocked by circuit breaker')
+				}
+				throw error
+			}
+		}
+
+		// Fallback to direct query if circuit breaker is disabled
+		return queryInternal(node, timeoutMs)
+	}
+
 	// Validate current key-bundle on server; on failure, trigger pre-key upload and rethrow
 	const digestKeyBundle = async (): Promise<void> => {
 		const res = await query({
@@ -460,6 +556,12 @@ export const makeSocket = (config: SocketConfig) => {
 
 	/** generates and uploads a set of pre-keys to the server */
 	const uploadPreKeys = async (count = MIN_PREKEY_COUNT, retryCount = 0) => {
+		// Check if pre-key circuit breaker is open
+		if (preKeyCircuitBreaker?.isOpen()) {
+			logger.warn('PreKey circuit breaker is open, skipping upload')
+			throw new CircuitOpenError('socket-prekey', 'open')
+		}
+
 		// Check minimum interval (except for retries)
 		if (retryCount === 0) {
 			const timeSinceLastUpload = Date.now() - lastUploadTime
@@ -487,14 +589,28 @@ export const makeSocket = (config: SocketConfig) => {
 				return node // Only return node since update is already used
 			}, creds?.me?.id || 'upload-pre-keys')
 
-			// Upload to server (outside transaction, can fail without affecting local keys)
-			try {
+			// Upload to server with circuit breaker protection
+			const uploadToServer = async () => {
 				await query(node)
 				logger.info({ count }, 'uploaded pre-keys successfully')
 				lastUploadTime = Date.now()
+			}
+
+			try {
+				// Use circuit breaker if available
+				if (preKeyCircuitBreaker) {
+					await preKeyCircuitBreaker.execute(uploadToServer)
+				} else {
+					await uploadToServer()
+				}
 			} catch (uploadError) {
 				logger.error({ uploadError: (uploadError as Error).toString(), count }, 'Failed to upload pre-keys to server')
 
+				// Don't retry if circuit breaker is open
+				if (uploadError instanceof CircuitOpenError) {
+					throw uploadError
+				}
+
 				// Exponential backoff retry (max 3 retries)
 				if (retryCount < 3) {
 					const backoffDelay = Math.min(1000 * Math.pow(2, retryCount), 10000)
@@ -617,6 +733,11 @@ export const makeSocket = (config: SocketConfig) => {
 		clearInterval(keepAliveReq)
 		clearTimeout(qrTimer)
 
+		// Clean up circuit breakers
+		queryCircuitBreaker?.destroy()
+		connectionCircuitBreaker?.destroy()
+		preKeyCircuitBreaker?.destroy()
+
 		ws.removeAllListeners('close')
 		ws.removeAllListeners('open')
 		ws.removeAllListeners('message')
@@ -1058,7 +1179,26 @@ export const makeSocket = (config: SocketConfig) => {
 		waitForConnectionUpdate: bindWaitForConnectionUpdate(ev),
 		sendWAMBuffer,
 		executeUSyncQuery,
-		onWhatsApp
+		onWhatsApp,
+		// Circuit breaker utilities
+		circuitBreakers: {
+			query: queryCircuitBreaker,
+			connection: connectionCircuitBreaker,
+			preKey: preKeyCircuitBreaker
+		},
+		/** Get circuit breaker statistics */
+		getCircuitBreakerStats: () => ({
+			query: queryCircuitBreaker?.getStats(),
+			connection: connectionCircuitBreaker?.getStats(),
+			preKey: preKeyCircuitBreaker?.getStats()
+		}),
+		/** Reset all circuit breakers to closed state */
+		resetCircuitBreakers: () => {
+			queryCircuitBreaker?.reset()
+			connectionCircuitBreaker?.reset()
+			preKeyCircuitBreaker?.reset()
+			logger.info('All circuit breakers reset to closed state')
+		}
 	}
 }
 

src/Types/Socket.ts

@@ -2,6 +2,7 @@ import type { Agent } from 'https'
 import type { URL } from 'url'
 import { proto } from '../../WAProto/index.js'
 import type { ILogger } from '../Utils/logger'
+import type { CircuitBreakerOptions } from '../Utils/circuit-breaker'
 import type { AuthenticationState, LIDMapping, SignalAuthState, TransactionCapabilityOptions } from './Auth'
 import type { GroupMetadata } from './GroupMetadata'
 import { type MediaConnInfo, type WAMessageKey } from './Message'
@@ -147,4 +148,21 @@ export type SocketConfig = {
 		logger: ILogger,
 		pnToLIDFunc?: (jids: string[]) => Promise<LIDMapping[] | undefined>
 	) => SignalRepositoryWithLIDStore
+
+	// === Circuit Breaker Configuration ===
+
+	/** Enable circuit breaker protection for socket operations (default: true) */
+	enableCircuitBreaker?: boolean
+
+	/** Circuit breaker configuration for query operations */
+	queryCircuitBreaker?: Partial<CircuitBreakerOptions>
+
+	/** Circuit breaker configuration for connection operations */
+	connectionCircuitBreaker?: Partial<CircuitBreakerOptions>
+
+	/** Circuit breaker configuration for pre-key operations */
+	preKeyCircuitBreaker?: Partial<CircuitBreakerOptions>
+
+	/** Circuit breaker configuration for message operations */
+	messageCircuitBreaker?: Partial<CircuitBreakerOptions>
 }

src/Utils/baileys-event-stream.ts

@@ -321,8 +321,8 @@ export class BaileysEventStream extends EventEmitter {
 		// Find correct position
 		let insertIndex = this.buffer.length
 		for (let i = 0; i < this.buffer.length; i++) {
-			const bufferItem = this.buffer[i]
-			if (bufferItem && PRIORITY_VALUES[bufferItem.priority] > eventPriorityValue) {
+			const bufferEvent = this.buffer[i]
+			if (bufferEvent && PRIORITY_VALUES[bufferEvent.priority] > eventPriorityValue) {
 				insertIndex = i
 				break
 			}

src/Utils/baileys-logger.ts

@@ -333,7 +333,7 @@ export class BaileysLogger implements ILogger {
 
 		// Structured log (skip if level is 'silent')
 		if (level !== 'silent') {
-			const logMethod = this.structuredLogger[level] as ((obj: unknown, msg?: string) => void) | undefined
+			const logMethod = (this.structuredLogger as unknown as Record<string, ((obj: unknown, msg?: string) => void) | undefined>)[level]
 			if (logMethod) {
 				logMethod.call(this.structuredLogger, enrichedObj, msg)
 			}
@@ -435,9 +435,9 @@ export class BaileysLogger implements ILogger {
 			// In production, mask part of the number
 			const parts = jid.split('@')
 			const localPart = parts[0]
-			const domain = parts[1]
-			if (parts.length === 2 && localPart && domain && localPart.length > 4) {
-				return `${localPart.substring(0, 4)}****@${domain}`
+			const domainPart = parts[1]
+			if (parts.length === 2 && localPart && domainPart && localPart.length > 4) {
+				return `${localPart.substring(0, 4)}****@${domainPart}`
 			}
 		}
 		return jid

src/Utils/prometheus-metrics.ts

@@ -612,7 +612,7 @@ export class MetricsRegistry {
 	/**
 	 * Return metrics in Prometheus format
 	 */
-	async metrics(): Promise<string> {
+	async getMetricsOutput(): Promise<string> {
 		const lines: string[] = []
 
 		for (const [name, metric] of this.metricsMap) {
@@ -776,7 +776,7 @@ export const metrics = {
  */
 export function createMetricsHandler(registry: MetricsRegistry = baileysMetrics) {
 	return async (_req: unknown, res: { setHeader: (name: string, value: string) => void; end: (body: string) => void }) => {
-		const metricsOutput = await registry.metrics()
+		const metricsOutput = await registry.getMetricsOutput()
 		res.setHeader('Content-Type', registry.contentType())
 		res.end(metricsOutput)
 	}