From 11e7c71fd1451b7569eb1251e39e5e5cf8d25eca Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Jan 2026 03:45:52 +0000 Subject: [PATCH] feat(event-buffer): enhance with enterprise-grade features - Add environment variable configuration (BAILEYS_BUFFER_*) - Implement destroy() method for proper resource cleanup - Add force flush capability with flush(force) parameter - Implement buffer overflow detection and prevention - Add adaptive timeout algorithm based on event rate - Replace historyCache.clear() with LRU cache cleanup - Integrate Prometheus metrics for buffer monitoring - Add isDestroyed state protection - Fix createBufferedFunction to prevent orphaned timers - Add comprehensive statistics tracking (getStatistics()) - Add configuration accessor (getConfig()) Enterprise features: - Configurable buffer size limits (maxBufferSize) - Configurable timeout ranges (min/max) - LRU cleanup ratio configuration - Buffer overflow warning threshold - Adaptive timeout based on event rate --- src/Utils/event-buffer.ts | 609 +++++++++++++++++++++++++++++--- src/Utils/prometheus-metrics.ts | 118 +++++++ 2 files changed, 686 insertions(+), 41 deletions(-) diff --git a/src/Utils/event-buffer.ts b/src/Utils/event-buffer.ts index dfcee32f..247f0022 100644 --- a/src/Utils/event-buffer.ts +++ b/src/Utils/event-buffer.ts @@ -16,6 +16,60 @@ import type { ILogger } from './logger' import { updateMessageWithReaction, updateMessageWithReceipt } from './messages' import { isRealMessage, shouldIncrementChatUnread } from './process-message' +// ============================================================================ +// BUFFER CONFIGURATION - Environment Variable Support +// ============================================================================ + +/** + * Buffer configuration loaded from environment variables + * Allows runtime customization without code changes + */ +export interface BufferConfig { + /** Maximum buffer timeout in milliseconds */ + bufferTimeoutMs: number + /** Minimum buffer timeout for adaptive algorithm */ + minBufferTimeoutMs: number + /** Maximum buffer timeout for adaptive algorithm */ + maxBufferTimeoutMs: number + /** Maximum history cache size before cleanup */ + maxHistoryCacheSize: number + /** Maximum events before forcing a flush (overflow protection) */ + maxBufferSize: number + /** Debounce delay for flush after buffered function */ + flushDebounceMs: number + /** Enable adaptive timeout based on load */ + enableAdaptiveTimeout: boolean + /** Enable Prometheus metrics */ + enableMetrics: boolean + /** LRU cleanup percentage (0-1) when cache exceeds max */ + lruCleanupRatio: number + /** Warn threshold for buffer size (percentage of max) */ + bufferWarnThreshold: number +} + +/** + * Load buffer configuration from environment variables + * Uses BAILEYS_BUFFER_* prefix for consistency + */ +export function loadBufferConfig(): BufferConfig { + return { + bufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_TIMEOUT_MS || '30000', 10), + minBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MIN_TIMEOUT_MS || '5000', 10), + maxBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MAX_TIMEOUT_MS || '60000', 10), + maxHistoryCacheSize: parseInt(process.env.BAILEYS_BUFFER_MAX_HISTORY_CACHE || '10000', 10), + maxBufferSize: parseInt(process.env.BAILEYS_BUFFER_MAX_SIZE || '5000', 10), + flushDebounceMs: parseInt(process.env.BAILEYS_BUFFER_FLUSH_DEBOUNCE_MS || '100', 10), + enableAdaptiveTimeout: process.env.BAILEYS_BUFFER_ADAPTIVE_TIMEOUT !== 'false', + enableMetrics: process.env.BAILEYS_BUFFER_METRICS === 'true', + lruCleanupRatio: parseFloat(process.env.BAILEYS_BUFFER_LRU_CLEANUP_RATIO || '0.2'), + bufferWarnThreshold: parseFloat(process.env.BAILEYS_BUFFER_WARN_THRESHOLD || '0.8') + } +} + +// ============================================================================ +// BUFFERABLE EVENTS +// ============================================================================ + const BUFFERABLE_EVENT = [ 'messaging-history.set', 'chats.upsert', @@ -44,95 +98,437 @@ type BaileysEventData = Partial const BUFFERABLE_EVENT_SET = new Set(BUFFERABLE_EVENT) +// ============================================================================ +// BUFFER STATISTICS - For Metrics and Monitoring +// ============================================================================ + +/** + * Statistics about buffer operations for monitoring and debugging + */ +export interface BufferStatistics { + /** Total number of flushes performed */ + totalFlushes: number + /** Total number of forced flushes (overflow/timeout) */ + forcedFlushes: number + /** Total events buffered */ + totalEventsBuffered: number + /** Current buffer size */ + currentBufferSize: number + /** Peak buffer size reached */ + peakBufferSize: number + /** Total buffer overflows prevented */ + overflowsDetected: number + /** Current adaptive timeout */ + currentTimeout: number + /** History cache size */ + historyCacheSize: number + /** LRU cleanups performed */ + lruCleanups: number + /** Average events per flush */ + avgEventsPerFlush: number + /** Buffer creation timestamp */ + createdAt: number + /** Last flush timestamp */ + lastFlushAt: number | null +} + +// ============================================================================ +// EXTENDED EVENT EMITTER TYPE +// ============================================================================ + type BaileysBufferableEventEmitter = BaileysEventEmitter & { /** Use to process events in a batch */ process(handler: (events: BaileysEventData) => void | Promise): () => void /** * starts buffering events, call flush() to release them - * */ + */ buffer(): void /** buffers all events till the promise completes */ // eslint-disable-next-line @typescript-eslint/no-explicit-any createBufferedFunction(work: (...args: A) => Promise): (...args: A) => Promise /** * flushes all buffered events + * @param force - If true, flush even if buffer is empty or not buffering * @returns returns true if the flush actually happened, otherwise false */ - flush(): boolean + flush(force?: boolean): boolean /** is there an ongoing buffer */ isBuffering(): boolean + /** + * Destroy the buffer and clean up all resources + * CRITICAL: Call this when done to prevent memory leaks + */ + destroy(): void + /** Check if buffer has been destroyed */ + isDestroyed(): boolean + /** Get buffer statistics for monitoring */ + getStatistics(): BufferStatistics + /** Get current configuration */ + getConfig(): BufferConfig } +// ============================================================================ +// LRU CACHE ENTRY WITH TIMESTAMP +// ============================================================================ + +interface LRUCacheEntry { + key: string + timestamp: number +} + +/** + * Simple LRU tracking for history cache + * Tracks access times to enable partial cleanup + */ +class LRUTracker { + private entries: Map = new Map() + + add(key: string): void { + this.entries.set(key, Date.now()) + } + + has(key: string): boolean { + return this.entries.has(key) + } + + touch(key: string): void { + if (this.entries.has(key)) { + this.entries.set(key, Date.now()) + } + } + + get size(): number { + return this.entries.size + } + + /** + * Remove oldest entries up to the specified ratio + * @param ratio - Percentage of entries to remove (0-1) + * @returns Array of removed keys + */ + cleanup(ratio: number): string[] { + const removeCount = Math.floor(this.entries.size * ratio) + if (removeCount === 0) return [] + + // Sort by timestamp (oldest first) + const sorted: LRUCacheEntry[] = [] + this.entries.forEach((timestamp, key) => { + sorted.push({ key, timestamp }) + }) + sorted.sort((a, b) => a.timestamp - b.timestamp) + + // Remove oldest entries + const removed: string[] = [] + for (let i = 0; i < removeCount && i < sorted.length; i++) { + const entry = sorted[i] + if (entry) { + this.entries.delete(entry.key) + removed.push(entry.key) + } + } + + return removed + } + + clear(): void { + this.entries.clear() + } +} + +// ============================================================================ +// ADAPTIVE TIMEOUT ALGORITHM +// ============================================================================ + +/** + * Adaptive timeout calculator based on buffer activity + * Adjusts timeout based on event rate to optimize batching + */ +class AdaptiveTimeoutCalculator { + private eventTimestamps: number[] = [] + private readonly windowSize = 100 // Track last 100 events + private readonly minTimeout: number + private readonly maxTimeout: number + private currentTimeout: number + + constructor(minTimeout: number, maxTimeout: number, initialTimeout: number) { + this.minTimeout = minTimeout + this.maxTimeout = maxTimeout + this.currentTimeout = initialTimeout + } + + /** + * Record an event and recalculate optimal timeout + */ + recordEvent(): void { + const now = Date.now() + this.eventTimestamps.push(now) + + // Keep only recent events + if (this.eventTimestamps.length > this.windowSize) { + this.eventTimestamps.shift() + } + + this.recalculateTimeout() + } + + /** + * Get current calculated timeout + */ + getTimeout(): number { + return this.currentTimeout + } + + /** + * Recalculate timeout based on event rate + * High rate = shorter timeout (flush more often) + * Low rate = longer timeout (batch more events) + */ + private recalculateTimeout(): void { + if (this.eventTimestamps.length < 2) { + return // Not enough data + } + + const oldest = this.eventTimestamps[0]! + const newest = this.eventTimestamps[this.eventTimestamps.length - 1]! + const timeSpan = newest - oldest + + if (timeSpan === 0) return + + // Calculate events per second + const eventsPerSecond = (this.eventTimestamps.length / timeSpan) * 1000 + + // Adjust timeout inversely to event rate + // High rate (>100 eps) = min timeout + // Low rate (<1 eps) = max timeout + if (eventsPerSecond > 100) { + this.currentTimeout = this.minTimeout + } else if (eventsPerSecond < 1) { + this.currentTimeout = this.maxTimeout + } else { + // Linear interpolation between min and max + const ratio = Math.log10(eventsPerSecond + 1) / 2 // 0 to 1 scale + this.currentTimeout = this.maxTimeout - (this.maxTimeout - this.minTimeout) * ratio + } + } + + reset(): void { + this.eventTimestamps = [] + this.currentTimeout = (this.minTimeout + this.maxTimeout) / 2 + } +} + +// ============================================================================ +// MAIN EVENT BUFFER FACTORY +// ============================================================================ + /** * The event buffer logically consolidates different events into a single event * making the data processing more efficient. + * + * Enterprise-grade features: + * - Environment variable configuration + * - Proper resource cleanup (destroy method) + * - Buffer overflow detection and prevention + * - Adaptive timeout based on event rate + * - LRU cache cleanup instead of full clear + * - Prometheus metrics integration + * - Force flush capability + * - Comprehensive statistics */ -export const makeEventBuffer = (logger: ILogger): BaileysBufferableEventEmitter => { +export const makeEventBuffer = ( + logger: ILogger, + configOverride?: Partial +): BaileysBufferableEventEmitter => { + // Load configuration + const config: BufferConfig = { + ...loadBufferConfig(), + ...configOverride + } + const ev = new EventEmitter() - const historyCache = new Set() + const historyCache = new LRUTracker() let data = makeBufferData() let isBuffering = false + let destroyed = false let bufferTimeout: NodeJS.Timeout | null = null - let flushPendingTimeout: NodeJS.Timeout | null = null // Add a specific timer for the debounced flush to prevent leak + let flushPendingTimeout: 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 + let currentEventCount = 0 - // take the generic event and fire it as a baileys event + // Statistics tracking + const stats: BufferStatistics = { + totalFlushes: 0, + forcedFlushes: 0, + totalEventsBuffered: 0, + currentBufferSize: 0, + peakBufferSize: 0, + overflowsDetected: 0, + currentTimeout: config.bufferTimeoutMs, + historyCacheSize: 0, + lruCleanups: 0, + avgEventsPerFlush: 0, + createdAt: Date.now(), + lastFlushAt: null + } + + // Adaptive timeout calculator + const adaptiveTimeout = new AdaptiveTimeoutCalculator( + config.minBufferTimeoutMs, + config.maxBufferTimeoutMs, + config.bufferTimeoutMs + ) + + // Metrics integration (lazy loaded to avoid circular deps) + let metricsModule: typeof import('./prometheus-metrics') | null = null + if (config.enableMetrics) { + import('./prometheus-metrics').then(m => { + metricsModule = m + }).catch(() => { + logger.debug('Prometheus metrics not available for event buffer') + }) + } + + // Helper to record metrics + const recordMetrics = (eventType: string, count: number) => { + if (metricsModule) { + metricsModule.recordEventBuffered(eventType, count) + } + } + + const recordFlushMetrics = (eventCount: number, forced: boolean) => { + if (metricsModule) { + metricsModule.recordBufferFlush(eventCount, forced) + } + } + + // Cleanup helper + const clearAllTimers = () => { + if (bufferTimeout) { + clearTimeout(bufferTimeout) + bufferTimeout = null + } + if (flushPendingTimeout) { + clearTimeout(flushPendingTimeout) + flushPendingTimeout = null + } + } + + // Take the generic event and fire it as a baileys event ev.on('event', (map: BaileysEventData) => { for (const event in map) { ev.emit(event, map[event as keyof BaileysEventMap]) } }) + /** + * Check for buffer overflow and handle it + */ + function checkBufferOverflow(): boolean { + if (currentEventCount >= config.maxBufferSize) { + stats.overflowsDetected++ + logger.warn({ + currentSize: currentEventCount, + maxSize: config.maxBufferSize + }, 'Buffer overflow detected, forcing flush') + flush(true) + return true + } + + // Warn if approaching threshold + const threshold = config.maxBufferSize * config.bufferWarnThreshold + if (currentEventCount >= threshold && currentEventCount < config.maxBufferSize) { + logger.debug({ + currentSize: currentEventCount, + threshold, + maxSize: config.maxBufferSize + }, 'Buffer approaching overflow threshold') + } + + return false + } + + /** + * Perform LRU cleanup on history cache + */ + function cleanupHistoryCache(): void { + if (historyCache.size > config.maxHistoryCacheSize) { + const removed = historyCache.cleanup(config.lruCleanupRatio) + stats.lruCleanups++ + logger.debug({ + removed: removed.length, + remaining: historyCache.size, + maxSize: config.maxHistoryCacheSize + }, 'LRU cleanup performed on history cache') + } + } + function buffer() { + if (destroyed) { + logger.warn('Attempted to buffer on destroyed event buffer') + return + } + if (!isBuffering) { logger.debug('Event buffer activated') isBuffering = true bufferCount = 0 + currentEventCount = 0 - if (bufferTimeout) { - clearTimeout(bufferTimeout) - } + clearAllTimers() + + // Use adaptive timeout if enabled + const timeout = config.enableAdaptiveTimeout + ? adaptiveTimeout.getTimeout() + : config.bufferTimeoutMs + stats.currentTimeout = timeout bufferTimeout = setTimeout(() => { if (isBuffering) { - logger.warn('Buffer timeout reached, auto-flushing') - flush() + logger.warn({ timeout }, 'Buffer timeout reached, auto-flushing') + stats.forcedFlushes++ + flush(true) } - }, BUFFER_TIMEOUT_MS) + }, timeout) } // Always increment count when requested bufferCount++ } - function flush() { - if (!isBuffering) { + function flush(force: boolean = false): boolean { + if (destroyed) { + logger.warn('Attempted to flush destroyed event buffer') return false } - logger.debug({ bufferCount }, 'Flushing event buffer') + if (!isBuffering && !force) { + return false + } + + const eventCount = currentEventCount + logger.debug({ bufferCount, eventCount, force }, 'Flushing event buffer') + isBuffering = false bufferCount = 0 + currentEventCount = 0 + stats.totalFlushes++ + stats.lastFlushAt = Date.now() - // Clear timeout - if (bufferTimeout) { - clearTimeout(bufferTimeout) - bufferTimeout = null - } + // Update average events per flush + stats.avgEventsPerFlush = stats.totalFlushes > 0 + ? (stats.avgEventsPerFlush * (stats.totalFlushes - 1) + eventCount) / stats.totalFlushes + : eventCount - if (flushPendingTimeout) { - clearTimeout(flushPendingTimeout) - flushPendingTimeout = null - } + // Clear timeouts + clearAllTimers() - // 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() - } + // Perform LRU cleanup instead of full clear + cleanupHistoryCache() + stats.historyCacheSize = historyCache.size + + // Record metrics + recordFlushMetrics(eventCount, force) const newData = makeBufferData() const chatUpdates = Object.values(data.chatUpdates) @@ -152,13 +548,49 @@ export const makeEventBuffer = (logger: ILogger): BaileysBufferableEventEmitter data = newData - logger.trace({ conditionalChatUpdatesLeft }, 'released buffered events') + logger.trace({ conditionalChatUpdatesLeft, eventCount }, 'released buffered events') + + // Reset adaptive timeout calculator on flush + if (config.enableAdaptiveTimeout) { + adaptiveTimeout.reset() + } return true } + function destroy(): void { + if (destroyed) { + logger.debug('Event buffer already destroyed') + return + } + + logger.debug('Destroying event buffer') + destroyed = true + + // Flush any remaining events + if (isBuffering) { + flush(true) + } + + // Clear all timers + clearAllTimers() + + // Clear all data structures + data = makeBufferData() + historyCache.clear() + + // Remove all event listeners + ev.removeAllListeners() + + logger.debug('Event buffer destroyed successfully') + } + return { process(handler) { + if (destroyed) { + throw new Error('Cannot process on destroyed event buffer') + } + const listener = async (map: BaileysEventData) => { await handler(map) } @@ -169,6 +601,11 @@ export const makeEventBuffer = (logger: ILogger): BaileysBufferableEventEmitter } }, emit(event: BaileysEvent, evData: BaileysEventMap[T]) { + if (destroyed) { + logger.warn({ event }, 'Attempted to emit on destroyed event buffer') + return false + } + // Check if this is a messages.upsert with a different type than what's buffered // If so, flush the buffered messages first to avoid type overshadowing if (event === 'messages.upsert') { @@ -192,6 +629,28 @@ export const makeEventBuffer = (logger: ILogger): BaileysBufferableEventEmitter } if (isBuffering && BUFFERABLE_EVENT_SET.has(event)) { + // Record event for adaptive timeout + if (config.enableAdaptiveTimeout) { + adaptiveTimeout.recordEvent() + } + + // Track statistics + currentEventCount++ + stats.totalEventsBuffered++ + stats.currentBufferSize = currentEventCount + if (currentEventCount > stats.peakBufferSize) { + stats.peakBufferSize = currentEventCount + } + + // Record metrics + recordMetrics(event, 1) + + // Check for overflow + if (checkBufferOverflow()) { + // Buffer was flushed due to overflow, re-buffer this event + buffer() + } + append(data, historyCache, event as BufferableEvent, evData, logger) return true } @@ -199,44 +658,93 @@ export const makeEventBuffer = (logger: ILogger): BaileysBufferableEventEmitter return ev.emit('event', { [event]: evData }) }, isBuffering() { - return isBuffering + return isBuffering && !destroyed + }, + isDestroyed() { + return destroyed }, buffer, flush, + destroy, + getStatistics() { + return { + ...stats, + currentBufferSize: currentEventCount, + historyCacheSize: historyCache.size, + currentTimeout: config.enableAdaptiveTimeout + ? adaptiveTimeout.getTimeout() + : config.bufferTimeoutMs + } + }, + getConfig() { + return { ...config } + }, createBufferedFunction(work) { + // Track the timeout for this specific function call + let functionTimeout: NodeJS.Timeout | null = null + return async (...args) => { + if (destroyed) { + throw new Error('Cannot execute buffered function on destroyed event buffer') + } + buffer() try { const result = await work(...args) + // If this is the only buffer, flush after a small delay if (bufferCount === 1) { - setTimeout(() => { - if (isBuffering && bufferCount === 1) { + // Clear any existing function timeout to prevent orphaned timers + if (functionTimeout) { + clearTimeout(functionTimeout) + } + functionTimeout = setTimeout(() => { + functionTimeout = null + if (isBuffering && bufferCount === 1 && !destroyed) { flush() } - }, 100) // Small delay to allow nested buffers + }, config.flushDebounceMs) } return result } catch (error) { + // Clear timeout on error + if (functionTimeout) { + clearTimeout(functionTimeout) + functionTimeout = null + } throw error } finally { bufferCount = Math.max(0, bufferCount - 1) - if (bufferCount === 0) { - // Only schedule ONE timeout, not 10,000 + if (bufferCount === 0 && !destroyed) { + // Only schedule ONE timeout, not multiple if (!flushPendingTimeout) { - flushPendingTimeout = setTimeout(flush, 100) + flushPendingTimeout = setTimeout(() => { + flushPendingTimeout = null + if (!destroyed) { + flush() + } + }, config.flushDebounceMs) } } } } }, - on: (...args) => ev.on(...args), + on: (...args) => { + if (destroyed) { + throw new Error('Cannot add listener to destroyed event buffer') + } + return ev.on(...args) + }, off: (...args) => ev.off(...args), removeAllListeners: (...args) => ev.removeAllListeners(...args) } } +// ============================================================================ +// BUFFER DATA STRUCTURE +// ============================================================================ + const makeBufferData = (): BufferedEventData => { return { historySets: { @@ -260,9 +768,13 @@ const makeBufferData = (): BufferedEventData => { } } +// ============================================================================ +// EVENT APPEND LOGIC +// ============================================================================ + function append( data: BufferedEventData, - historyCache: Set, + historyCache: LRUTracker, event: E, // eslint-disable-next-line @typescript-eslint/no-explicit-any eventData: any, @@ -282,6 +794,9 @@ function append( historyCache.add(id) absorbingChatUpdate(chat) + } else { + // Touch existing entry for LRU tracking + historyCache.touch(id) } } @@ -295,6 +810,8 @@ function append( if (!historyCache.has(historyContactId) || hasAnyName) { data.historySets.contacts[contact.id] = contact historyCache.add(historyContactId) + } else { + historyCache.touch(historyContactId) } } } @@ -305,6 +822,8 @@ function append( if (!existingMsg && !historyCache.has(key)) { data.historySets.messages[key] = message historyCache.add(key) + } else { + historyCache.touch(key) } } @@ -589,6 +1108,10 @@ function append( } } +// ============================================================================ +// EVENT CONSOLIDATION +// ============================================================================ + function consolidateEvents(data: BufferedEventData) { const map: BaileysEventData = {} @@ -670,6 +1193,10 @@ function consolidateEvents(data: BufferedEventData) { return map } +// ============================================================================ +// UTILITY FUNCTIONS +// ============================================================================ + function concatChats>(a: C, b: Partial) { if ( b.unreadCount === null && // neutralize unread counter diff --git a/src/Utils/prometheus-metrics.ts b/src/Utils/prometheus-metrics.ts index 8cb11703..36cd7dab 100644 --- a/src/Utils/prometheus-metrics.ts +++ b/src/Utils/prometheus-metrics.ts @@ -1468,6 +1468,124 @@ export function trackOperation( } } +// ============================================ +// Event Buffer Metrics Integration +// ============================================ + +// Event buffer metrics (lazy initialized) +let eventBufferMetrics: { + eventsBuffered: Counter | null + bufferFlushes: Counter | null + bufferFlushEvents: Histogram | null + bufferCurrentSize: Gauge | null + bufferPeakSize: Gauge | null + bufferHistoryCacheSize: Gauge | null + bufferOverflows: Counter | null + bufferLruCleanups: Counter | null +} | null = null + +function getEventBufferMetrics() { + if (!eventBufferMetrics) { + eventBufferMetrics = { + eventsBuffered: new Counter( + 'events_buffered_total', + 'Total number of events buffered', + ['event_type'] + ), + bufferFlushes: new Counter( + 'buffer_flushes_total', + 'Total number of buffer flushes', + ['forced'] + ), + bufferFlushEvents: new Histogram( + 'buffer_flush_events', + 'Number of events per buffer flush', + [], + [1, 5, 10, 25, 50, 100, 250, 500, 1000] + ), + bufferCurrentSize: new Gauge( + 'buffer_current_size', + 'Current number of events in buffer' + ), + bufferPeakSize: new Gauge( + 'buffer_peak_size', + 'Peak buffer size reached' + ), + bufferHistoryCacheSize: new Gauge( + 'buffer_history_cache_size', + 'Current size of history cache' + ), + bufferOverflows: new Counter( + 'buffer_overflows_total', + 'Total number of buffer overflows detected' + ), + bufferLruCleanups: new Counter( + 'buffer_lru_cleanups_total', + 'Total number of LRU cache cleanups performed' + ) + } + // Register all metrics + if (eventBufferMetrics.eventsBuffered) baileysMetrics.register(eventBufferMetrics.eventsBuffered) + if (eventBufferMetrics.bufferFlushes) baileysMetrics.register(eventBufferMetrics.bufferFlushes) + if (eventBufferMetrics.bufferFlushEvents) baileysMetrics.register(eventBufferMetrics.bufferFlushEvents) + if (eventBufferMetrics.bufferCurrentSize) baileysMetrics.register(eventBufferMetrics.bufferCurrentSize) + if (eventBufferMetrics.bufferPeakSize) baileysMetrics.register(eventBufferMetrics.bufferPeakSize) + if (eventBufferMetrics.bufferHistoryCacheSize) baileysMetrics.register(eventBufferMetrics.bufferHistoryCacheSize) + if (eventBufferMetrics.bufferOverflows) baileysMetrics.register(eventBufferMetrics.bufferOverflows) + if (eventBufferMetrics.bufferLruCleanups) baileysMetrics.register(eventBufferMetrics.bufferLruCleanups) + } + return eventBufferMetrics +} + +/** + * Record an event being buffered + * Used by event-buffer.ts for metrics integration + */ +export function recordEventBuffered(eventType: string, count: number = 1): void { + try { + const metrics = getEventBufferMetrics() + metrics.eventsBuffered?.inc({ event_type: eventType }, count) + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Record a buffer flush operation + * Used by event-buffer.ts for metrics integration + */ +export function recordBufferFlush(eventCount: number, forced: boolean): void { + try { + const metrics = getEventBufferMetrics() + metrics.bufferFlushes?.inc({ forced: forced ? 'true' : 'false' }) + metrics.bufferFlushEvents?.observe({}, eventCount) + metrics.bufferCurrentSize?.set({}, 0) // Reset after flush + } catch { + // Metrics not initialized, ignore silently + } +} + +/** + * Update buffer statistics gauge + * Used by event-buffer.ts for metrics integration + */ +export function updateBufferStatistics(stats: { + currentSize: number + peakSize: number + historyCacheSize: number + overflowsDetected: number + lruCleanups: number +}): void { + try { + const metrics = getEventBufferMetrics() + metrics.bufferCurrentSize?.set({}, stats.currentSize) + metrics.bufferPeakSize?.set({}, stats.peakSize) + metrics.bufferHistoryCacheSize?.set({}, stats.historyCacheSize) + } catch { + // Metrics not initialized, ignore silently + } +} + // ============================================ // Global Instance // ============================================