UHD_Media_Tool/apps/server/src/http/sse.ts

import { createRoute, OpenAPIHono } from "@hono/zod-openapi";
import { Api, apiResponse, type EventEnvelope, type EventReplayPage, EventType } from "@umt/contracts";
import type { Context } from "hono";
import { streamSSE } from "hono/streaming";
import { ApiException } from "./errors.ts";

export type BusListener = (e: EventEnvelope) => void;

/** since() 结果:gap=true 表示起点已不可补读(被裁剪),客户端需重新拉快照 */
export interface SinceResult {
  events: EventEnvelope[];
  gap: boolean;
  latestSeq: number;
}

/**
 * SSE 总线端口。模块与 SSE 路由只依赖这个接口。
 * 生产实现是 PgEventBus(sse-pg.ts:sse_event 表 + LISTEN/NOTIFY,多进程可见);
 * EventBus 是进程内实现,只给单元测试与无库场景用。
 */
export interface EventBusPort {
  /**
   * 发布事件并广播给所有进程的订阅者;resolve 为带 seq 的信封。
   * 永不 reject:持久化失败时记日志并返回 null(调用方可以不 await)。
   */
  publish<P>(type: string, payload: P): Promise<EventEnvelope<P> | null>;
  /** 取 seq > after 的事件(按 seq 升序,最多 limit 条) */
  since(after: number, limit?: number): Promise<SinceResult>;
  /** 订阅本进程收到的事件(含其它进程发布的);返回退订函数 */
  subscribe(l: BusListener): () => void;
}

/**
 * 进程内事件总线:序号自增,保留最近 capacity 条供断线补读。仅测试/单进程无库场景使用。
 */
export class EventBus implements EventBusPort {
  private seq = 0;
  private readonly buffer: EventEnvelope[] = [];
  private readonly listeners = new Set<BusListener>();

  constructor(private readonly capacity = 1000) {}

  get latestSeq() {
    return this.seq;
  }

  /** 同步发布(测试便利);publish 是它的异步包装 */
  publishSync<P>(type: string, payload: P): EventEnvelope<P> {
    const e: EventEnvelope<P> = { seq: ++this.seq, type, at: new Date().toISOString(), payload };
    this.buffer.push(e);
    if (this.buffer.length > this.capacity) this.buffer.shift();
    for (const l of this.listeners) l(e);
    return e;
  }

  publish<P>(type: string, payload: P): Promise<EventEnvelope<P> | null> {
    return Promise.resolve(this.publishSync(type, payload));
  }

  sinceSync(after: number, limit = 1000): SinceResult {
    const oldest = this.buffer[0]?.seq ?? this.seq + 1;
    const gap = after > 0 && after < oldest - 1;
    return { events: this.buffer.filter((e) => e.seq > after).slice(0, limit), gap, latestSeq: this.seq };
  }

  since(after: number, limit?: number): Promise<SinceResult> {
    return Promise.resolve(this.sinceSync(after, limit));
  }

  subscribe(l: BusListener): () => void {
    this.listeners.add(l);
    return () => this.listeners.delete(l);
  }
}

const SINCE_PAGE = 500;

/** 本进程当前打开的 SSE 流数(进程与健康页用;不含 JSON 补读请求) */
let openStreams = 0;
export function sseConnectionCount(): number {
  return openStreams;
}

export interface SseOptions {
  bus: EventBusPort;
  keepaliveMs?: number;
}

const REPLAY_DEFAULT_LIMIT = 100;

/** JSON 补读还是 SSE:带 limit,或 Accept 只要 JSON(不含 text/event-stream)时走 JSON */
function wantsJson(accept: string | undefined, limit: string | undefined): boolean {
  if (limit !== undefined) return true;
  const a = (accept ?? "").toLowerCase();
  return a.includes("application/json") && !a.includes("text/event-stream");
}

/**
 * GET /events?after=<seq>(也接受 Last-Event-ID 头):
 * - 控制台(cookie 会话)用 EventSource 走 SSE 流;
 * - API Key 调用者用 `GET /events?after=<seq>&limit=<n>` 取 JSON 页(同一 sse_event 表),按 nextAfter 续读。
 */
export function sseRoutes({ bus, keepaliveMs = 15_000 }: SseOptions) {
  const app = new OpenAPIHono();
  const def = Api.events.list;
  const route = createRoute({
    method: "get",
    path: def.path,
    tags: ["events"],
    summary: def.summary,
    responses: {
      200: {
        description: "JSON 补读页（带 limit 或 Accept: application/json）；否则为 text/event-stream",
        content: {
          "application/json": { schema: apiResponse(def.response) },
          "text/event-stream": { schema: { type: "string" } },
        },
      },
    },
  });
  app.openapi(route, (async (c: Context) => {
    const raw = c.req.query("after") ?? c.req.header("last-event-id") ?? "0";
    const after = Number(raw);
    if (!Number.isInteger(after) || after < 0) throw new ApiException("bad_request", "after 必须是非负整数");

    const limitRaw = c.req.query("limit");
    if (wantsJson(c.req.header("accept"), limitRaw)) {
      const limit = limitRaw === undefined ? REPLAY_DEFAULT_LIMIT : Number(limitRaw);
      if (!Number.isInteger(limit) || limit < 1 || limit > 1000)
        throw new ApiException("bad_request", "limit 必须是 1–1000 的整数");
      const r = await bus.since(after, limit + 1);
      const items = r.events.slice(0, limit);
      const data: EventReplayPage = {
        items,
        nextAfter: items.at(-1)?.seq ?? after,
        hasMore: r.events.length > limit,
        latestSeq: r.latestSeq,
        gap: r.gap,
      };
      return c.json({ ok: true as const, data });
    }

    return streamSSE(c, async (stream) => {
      openStreams += 1;
      let counted = true;
      const release = () => {
        if (counted) openStreams = Math.max(0, openStreams - 1);
        counted = false;
      };
      stream.onAbort(release);
      const queue: EventEnvelope[] = [];
      let wake: (() => void) | null = null;
      // 先订阅再读缓冲,避免两者之间的事件丢失
      const unsubscribe = bus.subscribe((e) => {
        queue.push(e);
        wake?.();
      });
      let closed = false;
      stream.onAbort(() => {
        closed = true;
        wake?.();
      });

      const { events, gap, latestSeq } = await bus.since(after, SINCE_PAGE);
      let last = after;
      if (gap) {
        await stream.writeSSE({ event: EventType.resync, data: JSON.stringify({ latestSeq }) });
      }
      const write = async (e: EventEnvelope) => {
        if (e.seq <= last) return;
        last = e.seq;
        await stream.writeSSE({ id: String(e.seq), event: e.type, data: JSON.stringify(e) });
      };
      for (const e of events) await write(e);
      // 积压超过一页时继续翻页补读(实时事件已在 queue 里排队,write 按 seq 去重)
      let page = events;
      while (page.length >= SINCE_PAGE && !closed) {
        page = (await bus.since(last, SINCE_PAGE)).events;
        for (const e of page) await write(e);
      }

      const timer = setInterval(() => {
        void stream.write(": keepalive\n\n");
      }, keepaliveMs);
      try {
        while (!closed) {
          while (queue.length) {
            const e = queue.shift();
            if (e) await write(e);
          }
          if (closed) break;
          await new Promise<void>((r) => {
            if (queue.length || closed) r();
            else wake = r;
          });
          wake = null;
        }
      } finally {
        clearInterval(timer);
        unsubscribe();
        release();
      }
    });
  }) as never);
  return app;
}