UHD_Media_Tool/apps/web/app/api/logs.ts

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<LogPage>("/logs", { params: { ...params, cursor: pageParam, limit: 50 }, signal }),
    getNextPageParam: (last) => last.nextCursor ?? undefined,
  });
}

export type LogStreamStatus = "off" | "connecting" | "open" | "reconnecting";

type EventSourceLike = Pick<EventSource, "addEventListener" | "close"> & { 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<typeof setTimeout> | 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<string>).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<LogEntry[]>([]);
  const [status, setStatus] = useState<LogStreamStatus>("off");
  const key = JSON.stringify(params);
  const client = useRef<LogStreamClient | null>(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;
  });
}