adam-acs/inertia/composables/useExchangeStream.ts

import { ref, onUnmounted, type Ref } from 'vue'

/**
 * Represents a single CWMP exchange message received from the backend.
 */
export interface CwmpExchange {
  deviceId: string
  direction: 'CPE→ACS' | 'ACS→CPE'
  timestamp: string
  sessionId: string
  cwmpMethod: string
  rawXml: string
  sequenceNumber: number
  createdAt?: string
}

export type ConnectionState = 'connected' | 'paused' | 'disconnected'

export interface UseExchangeStreamReturn {
  /** Reactive array of received exchange messages */
  messages: Ref<CwmpExchange[]>
  /** Current connection state */
  connectionState: Ref<ConnectionState>
  /** Whether auto-scroll is enabled (paused when user scrolls up) */
  isAutoScrollEnabled: Ref<boolean>
  /** Connect to the SSE stream for the device */
  connect: () => void
  /** Disconnect from the SSE stream */
  disconnect: () => void
  /** Resume auto-scroll and scroll to the latest message */
  resumeAutoScroll: () => void
  /** Clear all exchanges server-side and reset local state */
  clearMessages: () => Promise<void>
}

const POLLING_INTERVAL_MS = 5000
const MAX_CACHED_MESSAGES = 1000

/**
 * Composable for managing real-time CWMP exchange streaming via SSE,
 * with fallback to REST polling if SSE is unavailable.
 *
 * Requirements: 6.2, 6.3, 6.4, 6.5, 6.6, 6.7, 6.8, 8.4
 */
export function useExchangeStream(deviceId: string): UseExchangeStreamReturn {
  const messages = ref<CwmpExchange[]>([])
  const connectionState = ref<ConnectionState>('disconnected')
  const isAutoScrollEnabled = ref<boolean>(true)

  let eventSource: EventSource | null = null
  let lastSequenceNumber: number | null = null
  let pollingTimer: ReturnType<typeof setInterval> | null = null
  let sseSupported = true
  let reconnectAttempts = 0
  const MAX_RECONNECT_ATTEMPTS = 3

  /**
   * Build the SSE stream URL for the device.
   */
  function getStreamUrl(): string {
    return `/api/v1/tr069-trace/${encodeURIComponent(deviceId)}/exchanges/stream`
  }

  /**
   * Build the exchanges REST API URL for the device.
   */
  function getExchangesUrl(after?: string): string {
    const base = `/api/v1/tr069-trace/${encodeURIComponent(deviceId)}/exchanges`
    if (after) {
      return `${base}?after=${encodeURIComponent(after)}`
    }
    return base
  }

  /**
   * Append a message to the messages array, enforcing the client-side cap.
   */
  function appendMessage(exchange: CwmpExchange) {
    messages.value = [...messages.value, exchange]
    if (messages.value.length > MAX_CACHED_MESSAGES) {
      messages.value = messages.value.slice(-MAX_CACHED_MESSAGES)
    }
    lastSequenceNumber = exchange.sequenceNumber
  }

  /**
   * Connect to the SSE stream.
   * If SSE fails (404, 503, or repeated connection errors), falls back to REST polling.
   *
   * Requirements: 6.2, 6.7, 6.8
   */
  function connect() {
    // Clean up any existing connection
    disconnect()
    reconnectAttempts = 0

    if (sseSupported) {
      connectSSE()
    } else {
      startPolling()
    }
  }

  /**
   * Establish an EventSource connection to the SSE endpoint.
   */
  function connectSSE() {
    const url = getStreamUrl()
    eventSource = new EventSource(url)

    eventSource.onopen = () => {
      connectionState.value = isAutoScrollEnabled.value ? 'connected' : 'paused'
      reconnectAttempts = 0
    }

    eventSource.onmessage = (event: MessageEvent) => {
      try {
        const exchange: CwmpExchange = JSON.parse(event.data)
        appendMessage(exchange)

        // Update connection state based on auto-scroll
        connectionState.value = isAutoScrollEnabled.value ? 'connected' : 'paused'
      } catch {
        // Ignore parse errors for malformed SSE data
      }
    }

    eventSource.onerror = () => {
      reconnectAttempts++

      // EventSource automatically reconnects using the retry interval (3s) set by the server.
      // It also sends the Last-Event-ID header automatically on reconnection (Requirement 6.8).
      // After MAX_RECONNECT_ATTEMPTS consecutive errors, fall back to REST polling.
      if (reconnectAttempts > MAX_RECONNECT_ATTEMPTS) {
        // SSE seems unavailable — fall back to REST polling
        sseSupported = false
        closeEventSource()
        connectionState.value = 'disconnected'
        startPolling()
      } else {
        connectionState.value = 'disconnected'
      }
    }
  }

  /**
   * Close the EventSource connection without triggering a fallback.
   */
  function closeEventSource() {
    if (eventSource) {
      eventSource.close()
      eventSource = null
    }
  }

  /**
   * Disconnect from the SSE stream and stop any polling.
   * Preserves all previously displayed messages (Requirement 6.7).
   */
  function disconnect() {
    closeEventSource()
    stopPolling()
    connectionState.value = 'disconnected'
  }

  /**
   * Start REST polling as a fallback when SSE is unavailable.
   * Polls every 5 seconds for new messages using the `after` timestamp parameter.
   */
  function startPolling() {
    stopPolling()
    connectionState.value = isAutoScrollEnabled.value ? 'connected' : 'paused'

    pollingTimer = setInterval(async () => {
      await pollExchanges()
    }, POLLING_INTERVAL_MS)

    // Do an immediate poll
    void pollExchanges()
  }

  /**
   * Stop REST polling.
   */
  function stopPolling() {
    if (pollingTimer !== null) {
      clearInterval(pollingTimer)
      pollingTimer = null
    }
  }

  /**
   * Fetch new exchanges from the REST API using the `after` parameter
   * based on the last received message timestamp.
   */
  async function pollExchanges() {
    try {
      const lastMessage = messages.value.length > 0 ? messages.value[messages.value.length - 1] : null
      const after = lastMessage?.timestamp
      const url = getExchangesUrl(after)

      const response = await fetch(url)

      if (!response.ok) {
        if (response.status === 404 || response.status === 503) {
          // Device not traced or service unavailable
          connectionState.value = 'disconnected'
          stopPolling()
          return
        }
        return
      }

      const data = await response.json()
      const exchanges: CwmpExchange[] = data.exchanges || []

      for (const exchange of exchanges) {
        // Avoid duplicates by checking sequence number
        if (lastSequenceNumber === null || exchange.sequenceNumber > lastSequenceNumber) {
          appendMessage(exchange)
        }
      }

      connectionState.value = isAutoScrollEnabled.value ? 'connected' : 'paused'
    } catch {
      // Network error during polling — keep trying
      connectionState.value = 'disconnected'
    }
  }

  /**
   * Resume auto-scroll and update connection state.
   * Requirement 6.6: button allows user to resume auto-scroll.
   */
  function resumeAutoScroll() {
    isAutoScrollEnabled.value = true
    if (eventSource || pollingTimer) {
      connectionState.value = 'connected'
    }
  }

  /**
   * Clear all exchanges on the server and reset local state.
   * Calls DELETE API and resets messages array.
   * After clearing, verifies SSE connection is still active and reconnects if needed.
   *
   * Requirement 8.4: verify SSE connection remains active,
   * reset to empty state, continue receiving new messages.
   */
  async function clearMessages(): Promise<void> {
    const url = `/api/v1/tr069-trace/${encodeURIComponent(deviceId)}/exchanges`
    const response = await fetch(url, { method: 'DELETE' })

    if (!response.ok && response.status !== 204) {
      throw new Error(`Failed to clear exchanges: ${response.status} ${response.statusText}`)
    }

    // Reset local state
    messages.value = []
    lastSequenceNumber = null

    // Verify SSE connection is still active; reconnect if needed (Requirement 8.4)
    if (eventSource) {
      if (eventSource.readyState === EventSource.CLOSED) {
        connectSSE()
      }
      // If still open, continue receiving — nothing else needed
    } else if (pollingTimer) {
      // Polling is still running, continue as normal
    } else {
      // No active connection — attempt reconnection
      connect()
    }
  }

  // Cleanup on component unmount (Requirement 6.10)
  onUnmounted(() => {
    disconnect()
  })

  return {
    messages,
    connectionState,
    isAutoScrollEnabled,
    connect,
    disconnect,
    resumeAutoScroll,
    clearMessages,
  }
}