adam-acs/app/controllers/exchange_controller.ts

import type { HttpContext } from '@adonisjs/core/http'
import logger from '@adonisjs/core/services/logger'
import ExchangeModel from '#models/exchange'
import type { CwmpExchange } from '#models/exchange'
import { Tr069TraceService } from '#services/tr069_trace_service'
import { ExchangeStoreService } from '#services/exchange_store_service'
import { ExchangeMetricsService } from '#services/exchange_metrics_service'
import exchangeStoreConfig from '#config/exchange_store'

export default class ExchangeController {
  /**
   * GET /api/v1/tr069-trace/:deviceId/exchanges
   *
   * Returns exchanges for a device ordered by timestamp ascending.
   * Includes metadata about trace status and recording window.
   *
   * Query params:
   *  - limit: number of results (default 200, max 1000)
   *  - after: ISO 8601 timestamp for cursor-based pagination (newer messages)
   *  - before: ISO 8601 timestamp for reverse pagination (older messages)
   *  - latest: if "true", returns the most recent `limit` messages (for initial load)
   */
  async index({ params, request, response }: HttpContext) {
    const startTime = performance.now()
    const { deviceId } = params

    // Validate limit parameter
    const limitParam = request.qs().limit
    let limit = 200

    if (limitParam !== undefined && limitParam !== '') {
      const parsed = Number(limitParam)
      if (!Number.isFinite(parsed) || !Number.isInteger(parsed) || parsed < 1) {
        return response.status(400).json({
          error: 'Invalid limit parameter. Must be an integer between 1 and 1000.',
        })
      }
      limit = Math.min(parsed, 1000)
    }

    // Validate after parameter
    const afterParam = request.qs().after as string | undefined
    if (afterParam !== undefined && afterParam !== '') {
      if (!this.isValidIso8601(afterParam)) {
        return response.status(400).json({
          error: 'Invalid after parameter. Must be a valid ISO 8601 timestamp.',
        })
      }
    }

    // Validate before parameter (for loading older messages in batches)
    const beforeParam = request.qs().before as string | undefined
    if (beforeParam !== undefined && beforeParam !== '') {
      if (!this.isValidIso8601(beforeParam)) {
        return response.status(400).json({
          error: 'Invalid before parameter. Must be a valid ISO 8601 timestamp.',
        })
      }
    }

    // Check if the device has ever been traced (has any exchanges)
    const traceService = Tr069TraceService.getInstance()
    const isCurrentlyTraced = traceService.isTraced(deviceId)
    const hasExchanges = await ExchangeModel.hasExchanges(deviceId)

    // Never-traced device: no exchanges exist and not currently traced
    if (!hasExchanges && !isCurrentlyTraced) {
      return response.status(404).json({
        error: `No trace exists for device ${deviceId}`,
      })
    }

    // Fetch exchanges with read timeout (Requirement 7.4)
    const latestParam = request.qs().latest as string | undefined
    const useLatest = latestParam === 'true'

    const timeoutMs = Number(exchangeStoreConfig.readTimeoutMs)
    let exchanges: CwmpExchange[]
    let isTruncated = false

    try {
      exchanges = await this.queryWithTimeout(
        deviceId,
        { limit, after: afterParam, before: beforeParam, latest: useLatest },
        timeoutMs
      )
    } catch (error) {
      if ((error as Error).message === 'QUERY_TIMEOUT') {
        logger.warn(
          { correlationId: request.id(), deviceId, timeoutMs },
          '[ExchangeController] Read timeout exceeded, returning partial results'
        )
        isTruncated = true
        exchanges = []
      } else {
        throw error
      }
    }

    // Build metadata
    const metadata = await ExchangeModel.getMetadata(deviceId)

    // Track API response time
    const durationMs = performance.now() - startTime
    ExchangeMetricsService.getInstance().recordApiResponseTime('GET /exchanges', durationMs)

    const statusCode = 200
    logger.info(
      {
        correlationId: request.id(),
        method: 'GET',
        path: `/api/v1/tr069-trace/${deviceId}/exchanges`,
        statusCode,
        durationMs: Math.round(durationMs * 100) / 100,
        deviceId,
        resultCount: exchanges.length,
        isTruncated,
      },
      '[ExchangeController] API request completed'
    )

    return response.status(statusCode).json({
      exchanges,
      metadata: {
        traceStatus: isCurrentlyTraced ? 'ACTIVE' : 'STOPPED',
        traceStoppedAt: isCurrentlyTraced ? null : metadata.recordedUntil,
        recordedFrom: metadata.recordedFrom,
        recordedUntil: metadata.recordedUntil,
        total: metadata.count,
        ...(isTruncated ? { isTruncated: true } : {}),
      },
    })
  }

  /**
   * DELETE /api/v1/tr069-trace/:deviceId/exchanges
   *
   * Removes all stored exchanges for the device.
   * Returns 204 with no body.
   */
  async destroy({ params, request, response }: HttpContext) {
    const startTime = performance.now()
    const { deviceId } = params
    await ExchangeModel.deleteByDevice(deviceId)

    // Track API response time
    const durationMs = performance.now() - startTime
    ExchangeMetricsService.getInstance().recordApiResponseTime('DELETE /exchanges', durationMs)

    logger.info(
      {
        correlationId: request.id(),
        method: 'DELETE',
        path: `/api/v1/tr069-trace/${deviceId}/exchanges`,
        statusCode: 204,
        durationMs: Math.round(durationMs * 100) / 100,
        deviceId,
      },
      '[ExchangeController] API request completed'
    )

    return response.status(204).send('')
  }

  /**
   * GET /api/v1/tr069-trace/:deviceId/exchanges/stream (SSE)
   *
   * SSE endpoint for real-time exchange streaming.
   * - Sets SSE headers (Content-Type, Cache-Control, Connection, X-Accel-Buffering)
   * - Handles Last-Event-ID for reconnection (resumes from sequenceNumber > lastEventId)
   * - Subscribes to ExchangeStoreService events, pushes new exchanges as SSE events
   * - Implements server-side timeouts: 5min inactivity, 2h absolute
   * - Returns 404 if device is not traced
   * - Returns 503 if SSE feature is disabled
   *
   * Requirements: 6.1, 6.8, 6.9, 6.10, 6.11
   */
  async stream({ params, request, response }: HttpContext) {
    const startTime = performance.now()
    const { deviceId } = params

    // Check SSE feature flag
    if (!exchangeStoreConfig.featureFlags.sseEnabled) {
      return response.status(503).json({ error: 'SSE streaming is currently disabled' })
    }

    // Requirement 6.11: Return 404 if device is not traced
    const traceService = Tr069TraceService.getInstance()
    if (!traceService.isTraced(deviceId)) {
      return response.status(404).json({ error: 'Device is not being traced' })
    }

    const nodeRes = response.response
    const nodeReq = request.request

    // Requirement 6.1: Set SSE headers
    nodeRes.writeHead(200, {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
      'X-Accel-Buffering': 'no',
    })

    // Send retry interval so clients know how long to wait before reconnecting
    nodeRes.write(`retry: ${exchangeStoreConfig.sse.retryIntervalMs}\n\n`)

    // Requirement 6.8: Handle Last-Event-ID for reconnection
    const lastEventId = request.header('Last-Event-ID')
    if (lastEventId) {
      const seqNumber = Number(lastEventId)
      if (!Number.isNaN(seqNumber)) {
        try {
          const missedExchanges = await ExchangeModel.findAfterSequence(deviceId, seqNumber)
          for (const exchange of missedExchanges) {
            this.sendSSEEvent(nodeRes, exchange)
          }
          if (missedExchanges.length > 0) {
            logger.info(
              {
                correlationId: request.id(),
                deviceId,
                event: 'sse_reconnect_replay',
                lastEventId: seqNumber,
                replayedCount: missedExchanges.length,
              },
              '[SSE] Replayed missed exchanges on reconnection'
            )
          }
        } catch (error) {
          logger.error(
            {
              correlationId: request.id(),
              deviceId,
              event: 'sse_reconnect_error',
              lastEventId,
              error: (error as Error).message,
            },
            '[SSE] Error fetching missed exchanges for reconnection'
          )
        }
      }
    }

    // Subscribe to new events from ExchangeStoreService
    const exchangeStore = ExchangeStoreService.getInstance()
    let closed = false

    // Track SSE stream setup time
    const setupDurationMs = performance.now() - startTime
    ExchangeMetricsService.getInstance().recordApiResponseTime(
      'GET /exchanges/stream',
      setupDurationMs
    )

    const listener = (exchange: any) => {
      if (closed) return
      this.sendSSEEvent(nodeRes, exchange)
      resetInactivityTimer()

      // Track SSE push latency (time from persist/emit to client write)
      if (exchange._emitTimestamp) {
        const pushLatencyMs = performance.now() - exchange._emitTimestamp
        ExchangeMetricsService.getInstance().recordSsePushLatency(pushLatencyMs)
      }
    }

    exchangeStore.subscribe(deviceId, listener)

    // Structured log: SSE connection established
    const isReconnection = !!lastEventId
    logger.info(
      {
        correlationId: request.id(),
        deviceId,
        event: isReconnection ? 'sse_reconnect' : 'sse_connect',
        lastEventId: lastEventId || null,
        setupDurationMs: Math.round(setupDurationMs * 100) / 100,
      },
      `[SSE] Connection ${isReconnection ? 'reconnected' : 'established'}`
    )

    // Requirement 6.10: Timeout management
    const inactivityTimeoutMs = exchangeStoreConfig.sse.inactivityTimeoutMs
    const absoluteTimeoutMs = exchangeStoreConfig.sse.absoluteTimeoutMs

    let inactivityTimer: ReturnType<typeof setTimeout> = setTimeout(
      () => close('inactivity'),
      inactivityTimeoutMs
    )
    const absoluteTimer: ReturnType<typeof setTimeout> = setTimeout(
      () => close('absolute'),
      absoluteTimeoutMs
    )

    function resetInactivityTimer() {
      clearTimeout(inactivityTimer)
      inactivityTimer = setTimeout(() => close('inactivity'), inactivityTimeoutMs)
    }

    function close(reason?: string) {
      if (closed) return
      closed = true

      exchangeStore.unsubscribe(deviceId, listener)
      clearTimeout(inactivityTimer)
      clearTimeout(absoluteTimer)

      const connectionDurationMs = performance.now() - startTime
      logger.info(
        {
          correlationId: request.id(),
          deviceId,
          event: 'sse_disconnect',
          reason: reason || 'unknown',
          connectionDurationMs: Math.round(connectionDurationMs * 100) / 100,
        },
        '[SSE] Connection closed'
      )

      try {
        nodeRes.end()
      } catch {
        // Response may already be closed
      }
    }

    // Client disconnect
    nodeReq.on('close', () => close('client_disconnect'))
  }

  /**
   * Send a single SSE event to the client.
   * Format:
   *   id: <sequenceNumber>
   *   data: <JSON-encoded CwmpExchange>
   *
   * Requirement 6.9: Event id equals the exchange's sequenceNumber
   */
  private sendSSEEvent(res: import('node:http').ServerResponse, exchange: any): void {
    const data = JSON.stringify(exchange)
    res.write(`id: ${exchange.sequenceNumber}\ndata: ${data}\n\n`)
  }

  /**
   * Validates that a string is a valid ISO 8601 timestamp.
   * Checks that it parses to a valid Date and has the expected format patterns.
   */
  private isValidIso8601(value: string): boolean {
    const date = new Date(value)
    if (Number.isNaN(date.getTime())) {
      return false
    }

    // Must match common ISO 8601 patterns:
    // YYYY-MM-DD, YYYY-MM-DDTHH:mm:ss, YYYY-MM-DDTHH:mm:ssZ, YYYY-MM-DDTHH:mm:ss.sssZ, etc.
    const iso8601Pattern =
      /^\d{4}-\d{2}-\d{2}(T\d{2}:\d{2}(:\d{2}(\.\d{1,6})?)?(Z|[+-]\d{2}:?\d{2})?)?$/
    return iso8601Pattern.test(value)
  }

  /**
   * Execute a findByDevice query with a timeout.
   * If the query exceeds the timeout, throws an error with message 'QUERY_TIMEOUT'.
   * This supports lazy loading with `before`/`latest` parameters for batch loading.
   *
   * Requirements: 7.4 (partial results on timeout)
   */
  private async queryWithTimeout(
    deviceId: string,
    options: { limit: number; after?: string; before?: string; latest?: boolean },
    timeoutMs: number
  ): Promise<CwmpExchange[]> {
    const queryPromise = ExchangeModel.findByDevice(deviceId, options)

    const timeoutPromise = new Promise<never>((_, reject) => {
      setTimeout(() => reject(new Error('QUERY_TIMEOUT')), timeoutMs)
    })

    return Promise.race([queryPromise, timeoutPromise])
  }
}