import { useInfiniteQuery } from "@tanstack/react-query"; import { LOG_STREAM_EVENT, type LogEntry, type LogPage } from "@umt/contracts"; import { useEffect, useRef, useState } from "react"; import { apiData, buildUrl } from "~/lib/api.ts"; import { qk } from "~/lib/query-keys.ts"; // 日志模块(界面第二轮 V 线):GET /logs 列表(游标"加载更多")与 GET /logs/stream(SSE,同一组筛选)。 export type LogParams = { type?: string; level?: string; from?: string; q?: string; objectType?: string; objectId?: string; system?: string; }; export function useLogs(params: LogParams) { return useInfiniteQuery({ queryKey: qk.list("logs", params), initialPageParam: undefined as string | undefined, queryFn: ({ pageParam, signal }) => apiData("/logs", { params: { ...params, cursor: pageParam, limit: 50 }, signal }), getNextPageParam: (last) => last.nextCursor ?? undefined, }); } export type LogStreamStatus = "off" | "connecting" | "open" | "reconnecting"; type EventSourceLike = Pick & { onerror: ((ev: Event) => unknown) | null }; export type EventSourceFactory = (url: string) => EventSourceLike; /** * 日志流连接(不依赖 React,便于测试):start 连上 /logs/stream(since = 已知最新一条的时间), * 收到的条目交给 onEntry;stop 断开(暂停)。再 start 时从收到过的最新时间续上,服务端先补推这段时间的日志。 * 出错按指数退避重连。 */ export class LogStreamClient { private es: EventSourceLike | null = null; private timer: ReturnType | null = null; private retry = 0; private running = false; since: string | null; constructor( private readonly params: LogParams, private readonly handlers: { onEntry: (e: LogEntry) => void; onStatus: (s: LogStreamStatus) => void }, since: string | null | undefined, private readonly factory: EventSourceFactory = (url) => new EventSource(url, { withCredentials: true }), ) { this.since = since ?? null; } /** 列表加载出更新的条目时推进起点 */ advance(at: string | null | undefined) { this.since = latest(this.since, at) ?? null; } start() { if (this.running) return; this.running = true; this.connect(); } stop() { this.running = false; if (this.timer) clearTimeout(this.timer); this.timer = null; this.es?.close(); this.es = null; this.handlers.onStatus("off"); } private connect() { this.handlers.onStatus(this.retry ? "reconnecting" : "connecting"); const es = this.factory(buildUrl("/logs/stream", { params: { ...this.params, since: this.since ?? undefined } })); this.es = es; es.addEventListener("ready", () => { this.retry = 0; this.handlers.onStatus("open"); }); es.addEventListener(LOG_STREAM_EVENT, (ev) => { let e: LogEntry; try { e = JSON.parse((ev as MessageEvent).data) as LogEntry; } catch { return; } this.advance(e.at); this.handlers.onEntry(e); }); es.onerror = () => { es.close(); if (this.es !== es) return; this.es = null; if (!this.running) return; this.retry += 1; this.handlers.onStatus("reconnecting"); this.timer = setTimeout(() => this.connect(), Math.min(30_000, 1000 * 2 ** this.retry)); }; } } /** * 日志流 hook:enabled 时连,收到的条目按 id 覆盖、新的在前;enabled=false 即暂停。筛选变化时清空重连。 */ export function useLogStream(params: LogParams, opts: { enabled: boolean; since?: string | null }) { const [entries, setEntries] = useState([]); const [status, setStatus] = useState("off"); const key = JSON.stringify(params); const client = useRef(null); const sinceRef = useRef(opts.since); sinceRef.current = opts.since; client.current?.advance(opts.since); // 筛选变了:换一个连接,之前推来的不再适用 useEffect(() => { setEntries([]); const c = new LogStreamClient( JSON.parse(key) as LogParams, { onEntry: (e) => setEntries((prev) => upsertEntries(prev, [e])), onStatus: setStatus }, sinceRef.current, ); client.current = c; return () => { c.stop(); if (client.current === c) client.current = null; }; }, [key]); // key:筛选变化后新建的连接也要按当前开关启动 // biome-ignore lint/correctness/useExhaustiveDependencies: 见上 useEffect(() => { const c = client.current; if (!c || typeof EventSource === "undefined") return; if (opts.enabled) { c.advance(sinceRef.current); c.start(); } else c.stop(); }, [opts.enabled, key]); return { entries, status, clear: () => setEntries([]) }; } function latest(a: string | null | undefined, b: string | null | undefined): string | null | undefined { if (!a) return b; if (!b) return a; return Date.parse(b) > Date.parse(a) ? b : a; } /** 按 id 覆盖合并,时间倒序(同一时间按 id 倒序,与服务端一致) */ export function upsertEntries(base: LogEntry[], incoming: LogEntry[]): LogEntry[] { const map = new Map(base.map((e) => [e.id, e])); for (const e of incoming) map.set(e.id, e); return [...map.values()].sort((a, b) => { const d = Date.parse(b.at) - Date.parse(a.at); return d !== 0 ? d : a.id < b.id ? 1 : a.id > b.id ? -1 : 0; }); }