Migu2.0/frontends/apps/simple-platform-vue/src/composables/useProjectionStream.ts

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

/**
 * 与 SimpleLite `/projection/stream`（SSE）通信的轻量封装。
 *
 * 服务器以 `text/event-stream` 推送 JSON 行：
 *   ```
 *   event: car-state
 *   data: { "kind": "car-state", "tick": 123, "payload": { ... } }
 *   ```
 *
 * 离线 / 未启用时返回的 `connected = false`，调用方可继续用 3s 轮询兜底。
 */

export interface StreamEvent {
  kind: string
  tick?: number
  payload?: unknown
}

export interface ProjectionStreamOptions {
  /** SSE URL。默认走 axios baseURL `/api` + `/sl/projection/stream`。 */
  url?: string
  /** 挂载时自动 connect。默认 true。 */
  autoConnect?: boolean
  /** 断线后重连延时（ms）。默认 3000。 */
  reconnectDelayMs?: number
  /**
   * 额外要订阅的命名事件。这些会和 {@link KNOWN_EVENT_KINDS} 合并后注册到 EventSource。
   * 当后端新增 broadcast 事件名时，可不改动 composable，直接通过此参数补充。
   */
  extraEventKinds?: readonly string[]
}

/**
 * SimpleLite 当前后端会广播的全部 SSE 事件名。新增后端事件时务必同步追加，
 * 否则 EventSource 拿不到该事件 —— 这是历史上 alarm / object-* 被静默丢弃的根因。
 */
export const KNOWN_EVENT_KINDS = [
  'snapshot-tick',
  'selection-detail',
  'alarm',
  'object-created',
  'object-deleted',
  'object-patched',
  'object-batch-changed',
  'pick-result',
  'project-saved',
  'project-loaded',
  'car-state',
  'mission-status',
  'monitor-config-updated',
  'workspace-shortcut',
  'viewport-style-updated'
] as const

type Listener = (event: StreamEvent) => void

export function useProjectionStream(opts: ProjectionStreamOptions = {}) {
  const apiBase = (import.meta.env.VITE_API_BASE as string | undefined) ?? '/api'
  const defaultUrl = `${apiBase}/sl/projection/stream`
  const url = opts.url ?? defaultUrl
  const reconnectDelay = opts.reconnectDelayMs ?? 3000
  // 与 api/* 模块统一走 VITE_USE_MOCK 开关（替代旧的 VITE_PROJECTION_MOCK）。
  const isMock = import.meta.env.VITE_USE_MOCK === 'true'

  const connected = ref(false)
  const lastEvent: Ref<StreamEvent | null> = ref(null)
  const listeners = new Set<Listener>()

  let es: EventSource | null = null
  let reconnectTimer: ReturnType<typeof setTimeout> | null = null
  let mockTimer: ReturnType<typeof setInterval> | null = null
  let mockTick = 0
  let stopped = false

  function connect() {
    if (stopped) return

    // Mock 模式：不发 EventSource（无后端可连），改用本地心跳广播
    // `snapshot-tick`，让订阅方的"实时刷新"路径同样可被验证。
    if (isMock) {
      if (mockTimer) return
      connected.value = true
      mockTimer = setInterval(() => {
        mockTick++
        const event: StreamEvent = {
          kind: 'snapshot-tick',
          tick: mockTick,
          payload: { reason: 'mock', tick: mockTick }
        }
        lastEvent.value = event
        listeners.forEach((cb) => {
          try { cb(event) } catch { /* swallow */ }
        })
      }, 5000)
      return
    }

    if (es) return
    try {
      // withCredentials：让浏览器把 httpOnly 鉴权 Cookie (simple.auth.token) 随 SSE 一起带上，
      // 配合后端给 /api/sl/* 反代加的鉴权（EventSource 无法设置 Authorization header，只能靠 Cookie）。
      es = new EventSource(url, { withCredentials: true })
    } catch {
      scheduleReconnect()
      return
    }

    es.onopen = () => {
      connected.value = true
    }

    es.onerror = () => {
      connected.value = false
      closeSocket()
      scheduleReconnect()
    }

    es.onmessage = (msg) => {
      handleEventLine(undefined, msg.data)
    }

    // 命名事件：EventSource 默认只把"无 event: 头"的消息派发给 onmessage，
    // 任何 `event: foo` 都必须通过 addEventListener('foo', ...) 才能收到。
    // 必须覆盖 SimpleLite 后端可能广播的所有事件名 —— 漏一个就等于该事件被丢弃。
    // 当前后端事件源：
    //   - ProjectionStreamModule: snapshot-tick (1Hz 心跳/计数)
    //   - SimpleUI / ReflectionApiController: selection-detail
    //   - AlarmStreamService: alarm
    //   - MapEditApiController: object-created / object-deleted / object-patched
    //                           / object-batch-changed / pick-result
    //                           / project-saved / project-loaded
    //   - 保留 car-state / mission-status 占位以兼容未来事件型推送
    const allKinds = new Set<string>([...KNOWN_EVENT_KINDS, ...(opts.extraEventKinds ?? [])])
    for (const kind of allKinds) {
      es.addEventListener(kind, (msg: MessageEvent) => {
        handleEventLine(kind, msg.data)
      })
    }
  }

  function handleEventLine(kindHint: string | undefined, raw: unknown) {
    if (typeof raw !== 'string' || !raw) return
    try {
      const parsed = JSON.parse(raw)
      const event: StreamEvent =
        typeof parsed === 'object' && parsed !== null && 'kind' in parsed
          ? (parsed as StreamEvent)
          : { kind: kindHint ?? 'message', payload: parsed }
      lastEvent.value = event
      listeners.forEach((cb) => {
        try { cb(event) } catch { /* swallow */ }
      })
    } catch {
      // ignore malformed payload
    }
  }

  function closeSocket() {
    if (es) {
      try { es.close() } catch { /* ignore */ }
      es = null
    }
  }

  function scheduleReconnect() {
    if (stopped) return
    if (reconnectTimer) return
    reconnectTimer = setTimeout(() => {
      reconnectTimer = null
      connect()
    }, reconnectDelay)
  }

  function disconnect() {
    stopped = true
    if (reconnectTimer) {
      clearTimeout(reconnectTimer)
      reconnectTimer = null
    }
    if (mockTimer) {
      clearInterval(mockTimer)
      mockTimer = null
    }
    closeSocket()
    connected.value = false
  }

  function on(cb: Listener) { listeners.add(cb) }
  function off(cb: Listener) { listeners.delete(cb) }

  if (opts.autoConnect !== false) connect()

  onUnmounted(() => {
    disconnect()
    listeners.clear()
  })

  return { connected, lastEvent, connect, disconnect, on, off }
}