guoju0808/ry-vue3/src/utils/sseClient.js

// SSE 客户端封装 — 用 @microsoft/fetch-event-source (可设 Authorization header, 原生 EventSource 不行)
// 设计要点 (沿用 task_plan.md 决策):
//   - 单例连接: App 启动时调 sseStart() 一次, 退出时 sseStop()
//   - 推送只推信号: new_message 事件 = 重拉 /business/message/my, 内容由 DB 提供
//   - 断线重连: fetch-event-source 内置无限重试, 无需自己实现指数退避
//   - 鉴权: Authorization: Bearer <localStorage['ry_token']>, 避免 token 进 URL/nginx log
//   - 后端心跳 ": ping" (SSE 注释帧) 不触发 onmessage, 仅保活防代理 60s 断连

import { fetchEventSource } from '@microsoft/fetch-event-source'

const SSE_PATH = import.meta.env.VITE_APP_BASE_API + '/business/message/stream'

// 监听器表 (允许多组件订阅)
const listeners = { new_message: new Set(), connected: new Set() }

// SSE 连接单例状态 (避免重复订阅)
let ctrl = null        // AbortController, 用于关闭连接
let started = false    // 是否已启动订阅

// ====================== 全局事件总线 ======================
// 设计: 路由切换时 page 组件会 unmount, 直接 sseOn('new_message') 会丢订阅.
// AdminLayout 是唯一永驻的组件, 由它统一把 SSE 事件转发到 'global:new_message' 总线,
// 各 page 订阅总线, 这样 navbar bell 角标永远在, 切页面不影响.
const globalBus = new Set()

function emit(eventName, payload) {
  const set = listeners[eventName]
  if (set) {
    for (const fn of set) {
      try { fn(payload) } catch (e) { console.error(`[SSE] listener error on ${eventName}`, e) }
    }
  }
  // SSE 的 new_message 同时扇出到全局总线, 由 AdminLayout 维护 unreadCount, page 订阅刷新列表
  if (eventName === 'new_message') {
    for (const fn of globalBus) {
      try { fn(payload) } catch (e) { console.error(`[bus] listener error`, e) }
    }
  }
}

export function on(eventName, handler) {
  const set = listeners[eventName]
  if (!set) throw new Error(`[SSE] unknown event: ${eventName}`)
  set.add(handler)
  return () => set.delete(handler)   // 返回 unsubscribe
}

/**
 * 订阅全局总线 (SSE 事件的二级 fan-out, 用于 page 组件, 跨路由不掉)
 * 推荐各业务页面用这个, 不要直接 sseOn('new_message') — 后者绑在 AdminLayout 里更合理
 */
export function onGlobal(handler) {
  globalBus.add(handler)
  return () => globalBus.delete(handler)
}

export function sseStart() {
  if (started) return
  started = true

  const token = localStorage.getItem('ry_token')
  if (!token) {
    console.warn('[SSE] 无 token, 跳过订阅 (用户未登录)')
    started = false
    return
  }

  ctrl = new AbortController()
  fetchEventSource(SSE_PATH, {
    signal: ctrl.signal,
    headers: { Authorization: `Bearer ${token}` },
    openWhenHidden: true,   // 切到后台 tab 也保持, 减少重连风暴
    onopen(response) {
      if (response.ok) {
        console.info('[SSE] connected')
        return
      }
      // 401 等鉴权失败: 立刻停, fetch-event-source 会一直重试会刷屏
      if (response.status === 401 || response.status === 403) {
        console.warn(`[SSE] 鉴权失败 status=${response.status}, 停止重试`)
        ctrl.abort()
        started = false
        return
      }
      throw new Error(`SSE open failed: ${response.status}`)
    },
    onmessage(msg) {
      // msg.event: 'connected' | 'new_message' | 其它
      // msg.data: 服务端 .data() 写入的字符串, 格式 {type:'new_message', unread:N, silent:bool}
      // silent 由后端业务语义决定: false=业务事件触发 (manager 审核/分配/...), true=用户主动操作 (markRead)
      // 前端按 flag 统一决策弹/不弹, 避免每个业务页面重复判断
      if (msg.event === 'new_message') {
        let unread = 0, silent = false
        try {
          const parsed = JSON.parse(msg.data)
          unread = parsed.unread ?? 0
          silent = !!parsed.silent
        } catch {}
        emit('new_message', { unread, silent })
      } else if (msg.event === 'connected') {
        emit('connected', {})
      }
    },
    onclose() {
      console.info('[SSE] closed')
    },
    onerror(err) {
      console.warn('[SSE] error', err)
      // fetch-event-source 内部会自动重试, 这里只需要记日志, **不要 throw** (throw 会中断重试)
    }
  }).catch((e) => {
    if (e?.name !== 'AbortError') console.error('[SSE] fetchEventSource error', e)
    started = false
  })
}

export function sseStop() {
  if (ctrl) { try { ctrl.abort() } catch {} ; ctrl = null }
  started = false
  console.info('[SSE] stopped')
}