import { type ChildProcess, spawn } from "node:child_process";
import { fileURLToPath } from "node:url";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { createApp } from "../../src/http/app.ts";
import { PgEventBus } from "../../src/http/sse-pg.ts";
import { DEV_KEY } from "../helpers.ts";
import { createHarness, type Harness, PG_URL, pgAvailable } from "./harness.ts";

// SSE 多进程:两个 Deno 子进程(订阅者 / 发布者)通过 sse_event + LISTEN/NOTIFY 互通;
// 再验证 API 进程的 GET /events 能推送另一个进程发布的事件,以及 ?after= 补读与裁剪后的 resync。

const hasPg = await pgAvailable();
const SERVER_DIR = fileURLToPath(new URL("../..", import.meta.url));

function proc(args: string[]) {
  const child = spawn(process.execPath, ["run", "-A", "test/fixtures/sse-bus-proc.ts", ...args], {
    cwd: SERVER_DIR,
    env: { ...process.env, DATABASE_URL: PG_URL },
    stdio: ["ignore", "pipe", "pipe"],
  });
  const lines: string[] = [];
  let buf = "";
  let err = "";
  child.stdout!.on("data", (d: Buffer) => {
    buf += d.toString();
    const parts = buf.split("\n");
    buf = parts.pop() ?? "";
    lines.push(...parts.filter(Boolean));
  });
  child.stderr!.on("data", (d: Buffer) => {
    err += d.toString();
  });
  const exited = new Promise<number>((resolve) => child.on("exit", (code) => resolve(code ?? -1)));
  return { child, lines, exited, stderr: () => err };
}

async function until(cond: () => boolean, ms: number, what: string) {
  const deadline = Date.now() + ms;
  while (!cond()) {
    if (Date.now() > deadline) throw new Error(`等待超时:${what}`);
    await new Promise((r) => setTimeout(r, 50));
  }
}

describe.skipIf(!hasPg)("PgEventBus 多进程(真实 PG)", () => {
  let h: Harness;
  let bus: PgEventBus;
  const channel = `umt_events_t${process.pid}${Math.random().toString(36).slice(2, 6)}`;
  const children: ChildProcess[] = [];

  beforeAll(async () => {
    h = await createHarness();
    bus = await PgEventBus.start({
      db: h.db,
      connectionString: PG_URL,
      logger: h.ctx.logger,
      channel,
      retain: 30,
      pruneEvery: 1000,
    });
  }, 60_000);
  afterAll(async () => {
    for (const c of children) c.kill();
    await bus?.stop();
    await h?.close();
  });

  it("子进程 A 订阅、子进程 B 发布:A 按 seq 顺序收齐;本进程也能补读", async () => {
    const sub = proc(["sub", h.schema, channel, "20"]);
    children.push(sub.child);
    await until(() => sub.lines.includes("READY"), 30_000, "订阅进程就绪");
    const seenHere: number[] = [];
    const off = bus.subscribe((e) => seenHere.push(e.seq));
    const pub = proc(["pub", h.schema, channel, "20"]);
    children.push(pub.child);
    expect(await pub.exited).toBe(0);
    expect(await sub.exited).toBe(0);
    const events = sub.lines.filter((l) => l.startsWith("{")).map((l) => JSON.parse(l));
    expect(events).toHaveLength(20);
    expect(events.map((e) => e.payload.i)).toEqual([...Array(20).keys()]);
    const seqs = events.map((e) => e.seq as number);
    expect([...seqs].sort((a, b) => a - b)).toEqual(seqs);
    expect(new Set(events.map((e) => e.payload.pid)).size).toBe(1);
    await until(() => seenHere.length >= 20, 5_000, "本进程收到");
    off();
    const since = await bus.since(seqs[9]!);
    expect(since.events.map((e) => e.seq)).toEqual(seqs.slice(10));
    expect(since.gap).toBe(false);
    expect(since.latestSeq).toBe(seqs[19]);
  }, 60_000);

  it("API 进程的 GET /events 推送其它进程发布的事件;裁剪后旧游标收到 resync", async () => {
    const app = createApp({
      config: { NODE_ENV: "development", DEV_ADMIN_KEY: DEV_KEY },
      logger: h.ctx.logger,
      bus,
      role: "api",
      version: "t",
    });
    const { latestSeq } = await bus.since(0);
    const ac = new AbortController();
    const res = await app.request(`/api/v1/events?after=${latestSeq}`, {
      headers: { authorization: `Bearer ${DEV_KEY}` },
      signal: ac.signal,
    });
    const reader = res.body!.getReader();
    let text = "";
    const reading = (async () => {
      for (;;) {
        const { value, done } = await reader.read();
        if (done) return;
        text += new TextDecoder().decode(value);
      }
    })();
    const pub = proc(["pub", h.schema, channel, "3"]);
    children.push(pub.child);
    expect(await pub.exited).toBe(0);
    await until(() => (text.match(/event: test\.ping/g) ?? []).length >= 3, 10_000, "SSE 推送");
    ac.abort();
    await reader.cancel().catch(() => {});
    await reading.catch(() => {});

    // 裁剪到只剩 30 条后,after=1 已不可补读 → resync
    const more = proc(["pub", h.schema, channel, "40"]);
    children.push(more.child);
    expect(await more.exited).toBe(0);
    await bus.prune();
    const s = await bus.since(1);
    expect(s.gap).toBe(true);
    expect(s.events.length).toBe(30);
    const ac2 = new AbortController();
    const res2 = await app.request("/api/v1/events?after=1", {
      headers: { authorization: `Bearer ${DEV_KEY}` },
      signal: ac2.signal,
    });
    const r2 = res2.body!.getReader();
    const first = new TextDecoder().decode((await r2.read()).value);
    expect(first).toContain("event: system.resync");
    ac2.abort();
    await r2.cancel().catch(() => {});
  }, 60_000);

  it("发布失败不抛(返回 null)", async () => {
    // BigInt 无法 JSON 序列化 → 发布内部出错
    await expect(bus.publish("x", { n: 1n })).resolves.toBeNull();
  });
});