feat: TR-069 Exchange Viewer - structured MongoDB storage with SSE streaming and sequence diagram UI

- MongoDB storage: cwmp_exchanges collection with indexes (deviceId+timestamp, deviceId+sequenceNumber unique, TTL 24h) - ExchangeStoreService: CWMP method extraction, per-device queue for ordering, retry logic, message cap (5000/device) - Exchange API: GET/DELETE/SSE stream endpoints with Last-Event-ID reconnection - CwmpLoggerMiddleware: dual-write (MongoDB + legacy files) with fire-and-forget - Frontend: SequenceViewer (CPE↔ACS lifelines, arrows, auto-scroll, virtual scroll) + InspectorPanel (formatted XML, copy, syntax highlighting) - Tr069Trace page: integrated new viewer with col-6/col-6 layout - DeviceDetail page: collapsible exchange section when device is traced - 14 property-based tests, 40+ unit tests, 5 integration tests - Health check includes index verification - Config: feature flags, SSE timeouts, retry policy
2026-06-03 15:52:44 +02:00
1 parent 3fefbf6c40
commit 0c83a4ccb2
@@ -0,0 +1,381 @@
+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 (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])
+  }
+}