nexus/frontend/src/services/live.ts

import { apiFetch } from './api'

export type LiveEventName = 'snapshot' | 'update' | 'heartbeat'
export type LiveMode = 'live' | 'polling'

export interface LiveCursorDto {
  sequence: number
  timestamp: string
  mode: LiveMode
}

export interface LiveUpdateEnvelope {
  type: string
  timestamp: string
  payload: unknown
  sequence: number
  channel: string
}

export interface DashboardLiveEventDto {
  envelope: LiveUpdateEnvelope
  cursor: LiveCursorDto
}

export interface OpenDashboardLiveStreamResult {
  closed: Promise<void>
}

export async function openDashboardLiveStream(
  onEvent: (event: LiveEventName, data: unknown) => void,
  options: { forUser?: string; notificationLimit?: number; afterSequence?: number | null; signal?: AbortSignal } = {},
): Promise<OpenDashboardLiveStreamResult> {
  const params = new URLSearchParams({
    forUser: options.forUser ?? 'bao',
    notificationLimit: String(options.notificationLimit ?? 50),
  })
  if (typeof options.afterSequence === 'number' && Number.isFinite(options.afterSequence) && options.afterSequence > 0) {
    params.set('afterSequence', String(options.afterSequence))
  }

  const response = await apiFetch(`/api/dashboard/live?${params}`, {
    method: 'GET',
    headers: { Accept: 'text/event-stream', 'Cache-Control': 'no-cache' },
    signal: options.signal,
  })

  if (!response.ok || !response.body) {
    throw new Error(`Live stream unavailable: HTTP ${response.status}`)
  }

  const reader = response.body.getReader()
  const decoder = new TextDecoder()
  let buffer = ''

  const flushBlock = (block: string) => {
    const lines = block.split('\n')
    let eventName: LiveEventName = 'update'
    const dataLines: string[] = []

    for (const rawLine of lines) {
      const line = rawLine.trimEnd()
      if (line.startsWith('event:')) eventName = line.slice(6).trim() as LiveEventName
      if (line.startsWith('data:')) dataLines.push(line.slice(5).trim())
    }

    if (!dataLines.length) return
    try {
      onEvent(eventName, JSON.parse(dataLines.join('\n')))
    } catch (error) {
      console.warn('[live] failed to parse event payload', error)
    }
  }

  const closed = (async () => {
    while (true) {
      const { value, done } = await reader.read()
      if (done) break

      buffer += decoder.decode(value, { stream: true })
      const parts = buffer.split('\n\n')
      buffer = parts.pop() ?? ''
      for (const part of parts) {
        if (part.trim()) flushBlock(part)
      }
    }

    if (buffer.trim()) {
      flushBlock(buffer)
      buffer = ''
    }
  })()

  return { closed }
}