import { type EventEnvelope, EventType } from "@umt/contracts";
import { Hono } from "hono";
import { streamSSE } from "hono/streaming";
import { ApiException } from "./errors";

type Listener = (e: EventEnvelope) => void;

/**
 * 进程内事件总线:序号自增,保留最近 capacity 条供断线补读。
 * 多进程部署时需换成 PG LISTEN/NOTIFY 或 outbox 投影(W2)。
 */
export class EventBus {
  private seq = 0;
  private readonly buffer: EventEnvelope[] = [];
  private readonly listeners = new Set<Listener>();

  constructor(private readonly capacity = 1000) {}

  get latestSeq() {
    return this.seq;
  }

  publish<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;
  }

  /** 取 seq > after 的缓冲事件;gap=true 表示请求的起点已被淘汰,客户端需重新拉快照 */
  since(after: number): { events: EventEnvelope[]; gap: boolean } {
    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), gap };
  }

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

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

/** GET /events?after=<seq>(也接受 Last-Event-ID 头) */
export function sseRoutes({ bus, keepaliveMs = 15_000 }: SseOptions) {
  const app = new Hono();
  app.get("/events", (c) => {
    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 必须是非负整数");

    return streamSSE(c, async (stream) => {
      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 } = bus.since(after);
      let last = after;
      if (gap) {
        await stream.writeSSE({ event: EventType.resync, data: JSON.stringify({ latestSeq: bus.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);

      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();
      }
    });
  });
  return app;
}