adam-acs/tests/unit/controllers/exchange_controller_stream.spec.ts

import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { EventEmitter } from 'node:events'

// --- Mocks ---

const mockFindByDevice = vi.fn()
const mockHasExchanges = vi.fn()
const mockGetMetadata = vi.fn()
const mockDeleteByDevice = vi.fn()
const mockFindAfterSequence = vi.fn()

vi.mock('#models/exchange', () => ({
  default: {
    findByDevice: (...args: any[]) => mockFindByDevice(...args),
    hasExchanges: (...args: any[]) => mockHasExchanges(...args),
    getMetadata: (...args: any[]) => mockGetMetadata(...args),
    deleteByDevice: (...args: any[]) => mockDeleteByDevice(...args),
    findAfterSequence: (...args: any[]) => mockFindAfterSequence(...args),
  },
}))

const mockIsTraced = vi.fn()

vi.mock('#services/tr069_trace_service', () => ({
  Tr069TraceService: {
    getInstance: () => ({
      isTraced: mockIsTraced,
    }),
  },
}))

const mockEmitter = new EventEmitter()
const mockSubscribe = vi.fn((deviceId: string, listener: Function) => {
  mockEmitter.on(`exchange:${deviceId}`, listener as any)
})
const mockUnsubscribe = vi.fn((deviceId: string, listener: Function) => {
  mockEmitter.off(`exchange:${deviceId}`, listener as any)
})

vi.mock('#services/exchange_store_service', () => ({
  ExchangeStoreService: {
    getInstance: () => ({
      subscribe: mockSubscribe,
      unsubscribe: mockUnsubscribe,
    }),
  },
}))

vi.mock('#config/exchange_store', () => ({
  default: {
    maxMessagesPerDevice: 5000,
    sse: {
      inactivityTimeoutMs: 300_000,
      absoluteTimeoutMs: 7_200_000,
      maxConnectionsPerDevice: 5,
      retryIntervalMs: 3000,
    },
    featureFlags: {
      persistence: true,
      sseEnabled: true,
      apiSource: 'mongodb',
    },
  },
}))

vi.mock('@adonisjs/core/services/logger', () => ({
  default: {
    info: vi.fn(),
    error: vi.fn(),
    warn: vi.fn(),
    debug: vi.fn(),
  },
}))

import ExchangeController from '../../../app/controllers/exchange_controller.js'

// --- Helpers ---

function createStreamContext(options: {
  deviceId: string
  lastEventId?: string
}) {
  const written: string[] = []
  const headers: Record<string, string> = {}
  let headWritten = false
  let statusCode = 0
  let ended = false
  const reqEmitter = new EventEmitter()

  const nodeRes = {
    writeHead: vi.fn((code: number, hdrs: Record<string, string>) => {
      statusCode = code
      Object.assign(headers, hdrs)
      headWritten = true
    }),
    write: vi.fn((chunk: string) => {
      written.push(chunk)
    }),
    end: vi.fn(() => {
      ended = true
    }),
  }

  const responseObj = {
    response: nodeRes,
    status: vi.fn(function (this: any, code: number) {
      statusCode = code
      return this
    }),
    json: vi.fn((data: any) => data),
  }

  const ctx = {
    params: { deviceId: options.deviceId },
    request: {
      header: vi.fn((name: string) => {
        if (name === 'Last-Event-ID') return options.lastEventId
        return undefined
      }),
      id: vi.fn(() => 'test-request-id'),
      request: reqEmitter,
    },
    response: responseObj,
  }

  return {
    ctx: ctx as any,
    nodeRes,
    reqEmitter,
    getWritten: () => written,
    getHeaders: () => headers,
    isHeadWritten: () => headWritten,
    isEnded: () => ended,
    getStatusCode: () => statusCode,
  }
}

describe('ExchangeController.stream()', () => {
  let controller: ExchangeController

  beforeEach(() => {
    vi.clearAllMocks()
    vi.useFakeTimers()
    controller = new ExchangeController()
    mockFindAfterSequence.mockResolvedValue([])
  })

  afterEach(() => {
    vi.useRealTimers()
    mockEmitter.removeAllListeners()
  })

  it('returns 404 if device is not traced', async () => {
    mockIsTraced.mockReturnValue(false)
    const { ctx, getStatusCode } = createStreamContext({ deviceId: 'device-unknown' })

    await controller.stream(ctx)

    expect(ctx.response.status).toHaveBeenCalledWith(404)
    expect(ctx.response.json).toHaveBeenCalledWith({ error: 'Device is not being traced' })
  })

  it('sets correct SSE headers on traced device', async () => {
    mockIsTraced.mockReturnValue(true)
    const { ctx, getHeaders, getStatusCode } = createStreamContext({ deviceId: 'device-1' })

    await controller.stream(ctx)

    expect(getStatusCode()).toBe(200)
    expect(getHeaders()).toEqual({
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
      'X-Accel-Buffering': 'no',
    })
  })

  it('sends retry interval as first SSE message', async () => {
    mockIsTraced.mockReturnValue(true)
    const { ctx, getWritten } = createStreamContext({ deviceId: 'device-1' })

    await controller.stream(ctx)

    expect(getWritten()[0]).toBe('retry: 3000\n\n')
  })

  it('subscribes to ExchangeStoreService events for the device', async () => {
    mockIsTraced.mockReturnValue(true)
    const { ctx } = createStreamContext({ deviceId: 'device-1' })

    await controller.stream(ctx)

    expect(mockSubscribe).toHaveBeenCalledWith('device-1', expect.any(Function))
  })

  it('pushes new exchanges as SSE events with id: sequenceNumber', async () => {
    mockIsTraced.mockReturnValue(true)
    const { ctx, getWritten } = createStreamContext({ deviceId: 'device-1' })

    await controller.stream(ctx)

    const exchange = {
      deviceId: 'device-1',
      direction: 'CPE→ACS',
      timestamp: '2025-01-15T10:30:00.123Z',
      sessionId: '10.0.0.1:54321->192.168.1.1:8080',
      cwmpMethod: 'Inform',
      rawXml: '<soap-env:Envelope />',
      sequenceNumber: 42,
    }

    // Emit an exchange event
    mockEmitter.emit('exchange:device-1', exchange)

    const written = getWritten()
    const sseEvent = written[written.length - 1]
    expect(sseEvent).toBe(`id: 42\ndata: ${JSON.stringify(exchange)}\n\n`)
  })

  it('handles Last-Event-ID for reconnection — sends missed exchanges', async () => {
    mockIsTraced.mockReturnValue(true)

    const missedExchanges = [
      {
        deviceId: 'device-1',
        direction: 'CPE→ACS',
        timestamp: '2025-01-15T10:30:01.000Z',
        sessionId: 'sess-1',
        cwmpMethod: 'Inform',
        rawXml: '<xml/>',
        sequenceNumber: 6,
      },
      {
        deviceId: 'device-1',
        direction: 'ACS→CPE',
        timestamp: '2025-01-15T10:30:02.000Z',
        sessionId: 'sess-1',
        cwmpMethod: 'InformResponse',
        rawXml: '<xml/>',
        sequenceNumber: 7,
      },
    ]
    mockFindAfterSequence.mockResolvedValue(missedExchanges)

    const { ctx, getWritten } = createStreamContext({
      deviceId: 'device-1',
      lastEventId: '5',
    })

    await controller.stream(ctx)

    expect(mockFindAfterSequence).toHaveBeenCalledWith('device-1', 5)

    const written = getWritten()
    // After retry message, should have 2 missed exchanges
    expect(written[1]).toBe(`id: 6\ndata: ${JSON.stringify(missedExchanges[0])}\n\n`)
    expect(written[2]).toBe(`id: 7\ndata: ${JSON.stringify(missedExchanges[1])}\n\n`)
  })

  it('closes connection on client disconnect', async () => {
    mockIsTraced.mockReturnValue(true)
    const { ctx, reqEmitter, isEnded } = createStreamContext({ deviceId: 'device-1' })

    await controller.stream(ctx)

    // Simulate client disconnect
    reqEmitter.emit('close')

    expect(isEnded()).toBe(true)
    expect(mockUnsubscribe).toHaveBeenCalledWith('device-1', expect.any(Function))
  })

  it('closes connection after inactivity timeout (5min)', async () => {
    mockIsTraced.mockReturnValue(true)
    const { ctx, isEnded } = createStreamContext({ deviceId: 'device-1' })

    await controller.stream(ctx)

    expect(isEnded()).toBe(false)

    // Advance time by 5 minutes (inactivity timeout)
    vi.advanceTimersByTime(300_000)

    expect(isEnded()).toBe(true)
    expect(mockUnsubscribe).toHaveBeenCalledWith('device-1', expect.any(Function))
  })

  it('closes connection after absolute timeout (2h)', async () => {
    mockIsTraced.mockReturnValue(true)
    const { ctx, isEnded } = createStreamContext({ deviceId: 'device-1' })

    await controller.stream(ctx)

    // Send an exchange to reset inactivity timer at 4min
    vi.advanceTimersByTime(240_000)
    mockEmitter.emit('exchange:device-1', {
      deviceId: 'device-1',
      direction: 'CPE→ACS',
      timestamp: new Date().toISOString(),
      sessionId: 'sess',
      cwmpMethod: 'Inform',
      rawXml: '<xml/>',
      sequenceNumber: 1,
    })

    expect(isEnded()).toBe(false)

    // Advance to absolute timeout (2h from start)
    vi.advanceTimersByTime(7_200_000 - 240_000)

    expect(isEnded()).toBe(true)
  })

  it('resets inactivity timer when a new exchange is pushed', async () => {
    mockIsTraced.mockReturnValue(true)
    const { ctx, isEnded } = createStreamContext({ deviceId: 'device-1' })

    await controller.stream(ctx)

    // Advance 4 minutes (just under the 5min inactivity timeout)
    vi.advanceTimersByTime(240_000)
    expect(isEnded()).toBe(false)

    // Emit an exchange event → resets inactivity timer
    mockEmitter.emit('exchange:device-1', {
      deviceId: 'device-1',
      direction: 'CPE→ACS',
      timestamp: new Date().toISOString(),
      sessionId: 'sess',
      cwmpMethod: 'Inform',
      rawXml: '<xml/>',
      sequenceNumber: 1,
    })

    // Advance another 4 minutes (still within reset inactivity window)
    vi.advanceTimersByTime(240_000)
    expect(isEnded()).toBe(false)

    // Advance 1 more minute to exceed the 5min inactivity window
    vi.advanceTimersByTime(60_001)
    expect(isEnded()).toBe(true)
  })

  it('does not push events after connection is closed', async () => {
    mockIsTraced.mockReturnValue(true)
    const { ctx, reqEmitter, nodeRes } = createStreamContext({ deviceId: 'device-1' })

    await controller.stream(ctx)

    // Close connection
    reqEmitter.emit('close')

    // Clear the write mock to check no further writes
    nodeRes.write.mockClear()

    // Emit exchange after close
    mockEmitter.emit('exchange:device-1', {
      deviceId: 'device-1',
      direction: 'CPE→ACS',
      timestamp: new Date().toISOString(),
      sessionId: 'sess',
      cwmpMethod: 'Inform',
      rawXml: '<xml/>',
      sequenceNumber: 99,
    })

    expect(nodeRes.write).not.toHaveBeenCalled()
  })
})