adam-acs/app/services/exchange_store_service.ts

import { EventEmitter } from 'node:events'
import { getMongoDb } from '#providers/mongodb_provider'
import ExchangeModel from '#models/exchange'
import type { CwmpExchange } from '#models/exchange'
import exchangeStoreConfig from '#config/exchange_store'
import logger from '@adonisjs/core/services/logger'
import { ExchangeMetricsService } from '#services/exchange_metrics_service'

export interface PersistOptions {
  deviceId: string
  direction: 'CPE→ACS' | 'ACS→CPE'
  sessionId: string
  rawXml: string
}

const MAX_RAW_XML_BYTES = 2 * 1024 * 1024 // 2MB

/**
 * Critical CWMP methods that warrant retry on persistence failure.
 * These are RPC calls, responses, and faults that carry important protocol data.
 */
export const CRITICAL_METHODS = new Set([
  'Inform',
  'GetParameterValues',
  'GetParameterValuesResponse',
  'SetParameterValues',
  'SetParameterValuesResponse',
  'AddObject',
  'DeleteObject',
  'Download',
  'Upload',
  'Reboot',
  'FactoryReset',
  'Fault',
])

/**
 * Non-critical CWMP methods that are discarded immediately on failure.
 */
export const NON_CRITICAL_METHODS = new Set(['Empty', 'Unknown'])

export class ExchangeStoreService {
  private static instance: ExchangeStoreService
  private emitter: EventEmitter
  private _activeConnectionCount: number = 0
  /** Per-device queue to serialize persist operations and guarantee ordering */
  private deviceQueues: Map<string, Promise<any>> = new Map()

  private constructor() {
    this.emitter = new EventEmitter()
    this.emitter.setMaxListeners(100)
  }

  static getInstance(): ExchangeStoreService {
    if (!ExchangeStoreService.instance) {
      ExchangeStoreService.instance = new ExchangeStoreService()
    }
    return ExchangeStoreService.instance
  }

  /**
   * Extract CWMP method name from SOAP XML body.
   * Reads the first child element name under the `cwmp:` namespace prefix.
   * Returns "Empty" if the XML is empty/whitespace-only.
   * Returns "Unknown" if no `cwmp:` namespace element is found.
   */
  extractCwmpMethod(xml: string): string {
    if (!xml || xml.trim() === '') {
      return 'Empty'
    }

    // Try to find the cwmp method inside soap-env:Body first (skips cwmp:ID in Header)
    const bodyMatch = xml.match(/<soap-env:Body[^>]*>[\s\S]*?<cwmp:(\w+)[\s/>]/)
    let method: string | null = null
    if (bodyMatch) {
      method = bodyMatch[1]
    } else {
      // Fallback: match any cwmp: element that is NOT "ID"
      const allMatches = xml.matchAll(/<cwmp:(\w+)[\s/>]/g)
      for (const m of allMatches) {
        if (m[1] !== 'ID') {
          method = m[1]
          break
        }
      }
    }

    if (!method) {
      return 'Unknown'
    }

    // For Inform messages, extract event codes to enrich the display
    if (method === 'Inform') {
      const events = this.extractInformEvents(xml)
      if (events.length > 0) {
        return `Inform (${events.join(', ')})`
      }
    }

    return method
  }

  /**
   * Extract event codes from an Inform SOAP message.
   * Looks for <EventCode> elements and returns short labels like "BOOT", "PERIODIC", etc.
   */
  private extractInformEvents(xml: string): string[] {
    const eventRegex = /<EventCode[^>]*>([^<]+)<\/EventCode>/g
    const events: string[] = []
    let match: RegExpExecArray | null

    // Map numeric-only codes to their TR-069 labels
    const numericLabels: Record<string, string> = {
      '0': 'BOOTSTRAP',
      '1': 'BOOT',
      '2': 'PERIODIC',
      '3': 'SCHEDULED',
      '4': 'VALUE CHANGE',
      '5': 'KICKED',
      '6': 'CONNECTION REQUEST',
      '7': 'TRANSFER COMPLETE',
      '8': 'DIAGNOSTICS COMPLETE',
    }

    while ((match = eventRegex.exec(xml)) !== null) {
      const raw = match[1].trim()
      // If numeric only (e.g., "0"), map to label
      if (numericLabels[raw]) {
        events.push(numericLabels[raw])
      } else {
        // Extract just the label part (e.g., "0 BOOTSTRAP" → "BOOTSTRAP")
        const label = raw.replace(/^\d+\s+/, '')
        if (label) {
          events.push(label)
        }
      }
    }

    return events
  }

  /**
   * Persist an exchange with per-device queuing to guarantee strict ordering.
   * Each device's exchanges are serialized — no concurrent persists for the same device.
   * This method is fire-and-forget: it never throws and never blocks CWMP processing.
   * Checks the FEATURE_EXCHANGE_PERSISTENCE flag before persisting.
   */
  async persistExchange(options: PersistOptions): Promise<CwmpExchange | null> {
    if (!exchangeStoreConfig.featureFlags.persistence) {
      return null
    }

    // Queue the operation to serialize per-device
    const prev = this.deviceQueues.get(options.deviceId) || Promise.resolve()
    const current = prev.then(() => this.doPersistExchange(options)).catch(() => null)
    this.deviceQueues.set(options.deviceId, current)
    return current
  }

  /**
   * Internal persist implementation (called within the per-device queue).
   */
  private async doPersistExchange(options: PersistOptions): Promise<CwmpExchange | null> {

    try {
      let rawXml = options.rawXml || ''

      // Truncate rawXml at 2MB with a log warning
      if (Buffer.byteLength(rawXml, 'utf-8') > MAX_RAW_XML_BYTES) {
        logger.warn(
          { deviceId: options.deviceId, originalSize: Buffer.byteLength(rawXml, 'utf-8') },
          '[ExchangeStore] rawXml exceeds 2MB, truncating'
        )
        rawXml = rawXml.slice(0, MAX_RAW_XML_BYTES)
      }

      const cwmpMethod = this.extractCwmpMethod(rawXml)
      const sequenceNumber = await this.getNextSequenceNumber(options.deviceId)
      const timestamp = new Date().toISOString()

      const exchange: CwmpExchange = {
        deviceId: options.deviceId,
        direction: options.direction,
        timestamp,
        sessionId: options.sessionId,
        cwmpMethod,
        rawXml,
        sequenceNumber,
        createdAt: new Date(),
      }

      const isCritical = this.isCriticalMethod(cwmpMethod)

      // Track MongoDB write latency
      const writeStart = performance.now()
      await this.persistWithRetry(exchange, isCritical)
      const writeLatencyMs = performance.now() - writeStart
      ExchangeMetricsService.getInstance().recordMongoWriteLatency(writeLatencyMs)

      // Structured log: persistence success with all required fields
      logger.info(
        {
          deviceId: options.deviceId,
          cwmpMethod,
          direction: options.direction,
          sequenceNumber,
          durationMs: Math.round(writeLatencyMs * 100) / 100,
          operation: 'persist_exchange',
          status: 'success',
          isCritical,
          sessionId: options.sessionId,
        },
        '[ExchangeStore] Exchange persisted successfully'
      )

      // Enforce message cap after successful persist (non-blocking)
      await this.enforceMessageCap(options.deviceId)

      // Emit event for SSE subscribers (attach emit timestamp for SSE push latency tracking)
      ;(exchange as any)._emitTimestamp = performance.now()
      this.emitter.emit(`exchange:${options.deviceId}`, exchange)

      return exchange
    } catch (error) {
      logger.error(
        {
          deviceId: options.deviceId,
          cwmpMethod: this.extractCwmpMethod(options.rawXml || ''),
          direction: options.direction,
          operation: 'persist_exchange',
          status: 'failure',
          error: (error as Error).message,
          sessionId: options.sessionId,
        },
        '[ExchangeStore] Failed to persist exchange'
      )
      return null
    }
  }

  /**
   * Determine if a CWMP method is critical (warrants retry on failure).
   * Critical methods include RPC calls, responses, and faults.
   * Non-critical methods (Empty, Unknown) are discarded immediately on failure.
   */
  isCriticalMethod(cwmpMethod: string): boolean {
    return CRITICAL_METHODS.has(cwmpMethod)
  }

  /**
   * Retry logic for critical message persistence.
   *
   * - Critical messages: up to maxAttempts retries with exponential backoff
   *   (baseDelayMs * 2^attempt: 1s, 2s, 4s by default)
   * - Non-critical messages: discarded immediately on first failure
   * - NEVER throws — logs error and discards after all retries exhausted
   * - Ensures CWMP session is never interrupted regardless of persistence outcome
   */
  async persistWithRetry(doc: CwmpExchange, isCritical: boolean): Promise<void> {
    const { maxAttempts, baseDelayMs } = exchangeStoreConfig.retry

    // First attempt (always)
    try {
      const collection = ExchangeModel.getCollection()
      await collection.insertOne(doc as any)
      return
    } catch (error) {
      if (!isCritical) {
        // Non-critical: log and discard immediately
        logger.warn(
          { deviceId: doc.deviceId, cwmpMethod: doc.cwmpMethod, error: String(error) },
          '[ExchangeStore] Non-critical message persistence failed — discarding'
        )
        return
      }

      logger.warn(
        { deviceId: doc.deviceId, cwmpMethod: doc.cwmpMethod, error: String(error) },
        '[ExchangeStore] Critical message persistence failed — starting retries'
      )
    }

    // Retry loop for critical messages
    for (let attempt = 0; attempt < maxAttempts; attempt++) {
      const delayMs = baseDelayMs * Math.pow(2, attempt)

      await this.sleep(delayMs)

      try {
        const collection = ExchangeModel.getCollection()
        await collection.insertOne(doc as any)
        logger.info(
          { deviceId: doc.deviceId, cwmpMethod: doc.cwmpMethod, attempt: attempt + 1 },
          '[ExchangeStore] Critical message persisted after retry'
        )
        return
      } catch (error) {
        logger.warn(
          {
            deviceId: doc.deviceId,
            cwmpMethod: doc.cwmpMethod,
            attempt: attempt + 1,
            maxAttempts,
            error: String(error),
          },
          '[ExchangeStore] Retry failed'
        )
      }
    }

    // All retries exhausted — log error and discard (NEVER throw)
    logger.error(
      { deviceId: doc.deviceId, cwmpMethod: doc.cwmpMethod, maxAttempts },
      '[ExchangeStore] All retries exhausted for critical message — discarding'
    )
  }

  /**
   * Sleep utility for exponential backoff.
   */
  private sleep(ms: number): Promise<void> {
    return new Promise((resolve) => setTimeout(resolve, ms))
  }

  /**
   * Get next sequence number for a device using atomic increment.
   * Uses the `cwmp_exchange_counters` collection with findOneAndUpdate + upsert.
   */
  async getNextSequenceNumber(deviceId: string): Promise<number> {
    const db = getMongoDb()
    const counters = db.collection('cwmp_exchange_counters')
    const result = await counters.findOneAndUpdate(
      { _id: deviceId as any },
      { $inc: { seq: 1 } },
      { upsert: true, returnDocument: 'after' }
    )
    return result!.seq as number
  }

  /**
   * Subscribe to exchange events for a specific device.
   * The listener will be called with each new CwmpExchange as it's persisted.
   */
  subscribe(deviceId: string, listener: (exchange: CwmpExchange) => void): void {
    this.emitter.on(`exchange:${deviceId}`, listener)
    this._activeConnectionCount++
  }

  /**
   * Unsubscribe from exchange events for a specific device.
   */
  unsubscribe(deviceId: string, listener: (exchange: CwmpExchange) => void): void {
    this.emitter.off(`exchange:${deviceId}`, listener)
    this._activeConnectionCount = Math.max(0, this._activeConnectionCount - 1)
  }

  /**
   * Get the current count of active SSE connections (subscriptions).
   */
  get activeConnectionCount(): number {
    return this._activeConnectionCount
  }

  /**
   * Enforce max messages cap per device.
   * Deletes oldest messages when count exceeds maxMessagesPerDevice.
   * Errors are logged but never thrown (non-blocking).
   */
  async enforceMessageCap(deviceId: string): Promise<void> {
    try {
      const maxCount = exchangeStoreConfig.maxMessagesPerDevice
      const count = await ExchangeModel.countByDevice(deviceId)

      if (count > maxCount) {
        await ExchangeModel.deleteOldestBeyondCap(deviceId, maxCount)
        logger.info(
          { deviceId, count, maxCount, deleted: count - maxCount },
          '[ExchangeStore] Message cap enforced, oldest messages deleted'
        )
      }
    } catch (error) {
      logger.error(
        { deviceId, error },
        '[ExchangeStore] Error enforcing message cap (non-blocking)'
      )
    }
  }
}