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 = setTimeout( () => close('inactivity'), inactivityTimeoutMs ) const absoluteTimer: ReturnType = 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: * data: * * 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 { const queryPromise = ExchangeModel.findByDevice(deviceId, options) const timeoutPromise = new Promise((_, reject) => { setTimeout(() => reject(new Error('QUERY_TIMEOUT')), timeoutMs) }) return Promise.race([queryPromise, timeoutPromise]) } }