import { EventType } from "@umt/contracts"; import { useEffect, useRef, useState } from "react"; import { buildUrl } from "./api.ts"; // SSE 封装:整个页面共用一条 EventSource(按订阅引用计数开关)。 // - 服务端按事件类型发命名事件(event: ),所以按 type 订阅;"*" 订阅全部已知类型。 // - 记住最后一个 seq;断线后按指数退避重连 `/events?after=` 补读,不漏不重。 // - 收到 system.resync(起点已被裁剪)时,订阅者应整页重取(useSseInvalidate 自动处理)。 // 鉴权走 cookie 会话(EventSource 不能带 Authorization 头)。 export type SseEvent

= { seq: number; type: string; at: string; payload: P }; export type SseHandler = (event: SseEvent) => void; export type SseStatus = "idle" | "connecting" | "open" | "reconnecting"; const KNOWN_TYPES: string[] = Object.values(EventType); const WILDCARD = "*"; type Sub = { types: Set; handler: SseHandler }; class SseClient { private es: EventSource | null = null; private subs = new Set(); private listened = new Set(); private lastSeq: number | null = null; private retry = 0; private timer: ReturnType | null = null; private statusListeners = new Set<(s: SseStatus) => void>(); status: SseStatus = "idle"; subscribe(types: string[], handler: SseHandler): () => void { const sub: Sub = { types: new Set(types), handler }; this.subs.add(sub); if (!this.es && !this.timer) this.connect(); else for (const t of this.expand(sub.types)) this.listen(t); return () => { this.subs.delete(sub); if (this.subs.size === 0) this.close(); }; } onStatus(fn: (s: SseStatus) => void): () => void { this.statusListeners.add(fn); return () => this.statusListeners.delete(fn); } private setStatus(s: SseStatus) { this.status = s; for (const fn of this.statusListeners) fn(s); } private expand(types: Set): string[] { return types.has(WILDCARD) ? [...KNOWN_TYPES, ...[...types].filter((t) => t !== WILDCARD)] : [...types]; } private connect() { if (typeof EventSource === "undefined") return; this.timer = null; this.setStatus(this.retry ? "reconnecting" : "connecting"); const url = buildUrl("/events", { params: { after: this.lastSeq ?? undefined } }); const es = new EventSource(url, { withCredentials: true }); this.es = es; this.listened.clear(); // resync 永远监听 this.listen(EventType.resync); for (const sub of this.subs) for (const t of this.expand(sub.types)) this.listen(t); es.onopen = () => { this.retry = 0; this.setStatus("open"); }; es.onerror = () => { // 自己掌控重连(带 after=seq),不用浏览器默认重连 es.close(); if (this.es !== es) return; this.es = null; if (this.subs.size === 0) return; const delay = Math.min(30_000, 1000 * 2 ** this.retry); this.retry += 1; this.setStatus("reconnecting"); this.timer = setTimeout(() => this.connect(), delay); }; } private listen(type: string) { if (!this.es || this.listened.has(type)) return; this.listened.add(type); this.es.addEventListener(type, (e) => this.dispatch(type, e as MessageEvent)); } private dispatch(type: string, e: MessageEvent) { let event: SseEvent; try { const parsed = JSON.parse(e.data) as Partial; event = typeof parsed.seq === "number" ? (parsed as SseEvent) : { seq: this.lastSeq ?? 0, type, at: new Date().toISOString(), payload: parsed }; } catch { return; } if (typeof event.seq === "number" && type !== EventType.resync) { if (this.lastSeq !== null && event.seq <= this.lastSeq) return; // 补读重叠去重 this.lastSeq = event.seq; } for (const sub of this.subs) { if (type === EventType.resync || sub.types.has(type) || sub.types.has(WILDCARD)) sub.handler(event); } } private close() { if (this.timer) clearTimeout(this.timer); this.timer = null; this.es?.close(); this.es = null; this.retry = 0; this.setStatus("idle"); } } export const sse = new SseClient(); /** * 订阅 SSE 事件。types 为事件类型数组(见 contracts EventType),"*" 表示全部已知类型。 * handler 可以每次渲染都变,不会重复订阅。system.resync 总会送达。 */ export function useSse(types: string[], handler: SseHandler, enabled = true) { const ref = useRef(handler); ref.current = handler; const key = types.join("|"); useEffect(() => { if (!enabled) return; return sse.subscribe(key.split("|"), (e) => ref.current(e)); }, [key, enabled]); } /** 连接状态(给页面角落的"实时"指示用) */ export function useSseStatus(): SseStatus { const [s, setS] = useState(sse.status); useEffect(() => sse.onStatus(setS), []); return s; }