import { sql } from "kysely";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { type EnqueueOptions, OutboxRelay, type TaskEngine } from "../src/index.ts";
import {
  connectDb,
  dropSchemas,
  migrateBizSchema,
  pgAvailable,
  resetSchemas,
  type TestDb,
  testSchemas,
} from "./helpers.ts";

// 生命周期广播(ADR-006):一条 outbox 事件路由给多个订阅方任务。第一个任务用事件的幂等键,
// 其余用 <键>#<任务名>;失败重投时已入队的由引擎幂等键去重,没入队的补上。

const schemas = testSchemas("outbox_bc");
const hasPg = await pgAvailable();

describe.skipIf(!hasPg)("OutboxRelay 广播给多个订阅方", () => {
  let db: TestDb;

  beforeAll(async () => {
    await resetSchemas(schemas);
    db = connectDb(schemas.biz);
    await migrateBizSchema(db, schemas.biz);
  }, 30_000);
  afterAll(async () => {
    await db?.destroy();
    await dropSchemas(schemas);
  }, 30_000);

  it("每个订阅方各入队一次,幂等键按订阅方区分;中途失败重投时同键去重", async () => {
    await sql`INSERT INTO outbox_event (type, payload, idempotency_key)
      VALUES ('thing.done', '{"id":"t1"}'::jsonb, 'thing.done/t1')`.execute(db);
    const keys = new Map<string, number>();
    let failB = true;
    const engine: Pick<TaskEngine, "enqueue"> = {
      enqueue: async (task: string, _input: unknown, opts: EnqueueOptions) => {
        if (task === "b" && failB) {
          failB = false;
          throw new Error("引擎暂时不可用");
        }
        keys.set(opts.idempotencyKey, (keys.get(opts.idempotencyKey) ?? 0) + 1);
        return { runId: opts.idempotencyKey, deduplicated: false };
      },
    };
    const relay = new OutboxRelay({
      db,
      engine,
      routes: {
        "thing.done": (ev) => [
          { task: "a", input: ev.payload },
          { task: "b", input: ev.payload },
        ],
      },
    });
    const first = await relay.pollOnce();
    expect(first).toEqual({ published: 0, failed: 1 });
    // 退避期过了再投:a 用同一把键再入一次(真实引擎按幂等键返回已有 run),b 补上
    await sql`UPDATE outbox_event SET next_attempt_at = now()`.execute(db);
    expect(await relay.pollOnce()).toEqual({ published: 1, failed: 0 });
    expect([...keys.keys()].sort()).toEqual(["thing.done/t1", "thing.done/t1#b"]);
    expect(keys.get("thing.done/t1")).toBe(2);
    expect(keys.get("thing.done/t1#b")).toBe(1);
    const row = await sql<{ status: string }>`SELECT status FROM outbox_event`.execute(db);
    expect(row.rows).toEqual([{ status: "dispatched" }]);
  });

  it("单个订阅方保持原来的键;返回空数组 / null 直接标记已投递", async () => {
    await sql`DELETE FROM outbox_event`.execute(db);
    await sql`INSERT INTO outbox_event (type, payload, idempotency_key) VALUES
      ('one.sub', '{}'::jsonb, 'k-one'), ('no.sub', '{}'::jsonb, 'k-none')`.execute(db);
    const keys: string[] = [];
    const relay = new OutboxRelay({
      db,
      engine: {
        enqueue: async (_t: string, _i: unknown, o: EnqueueOptions) => {
          keys.push(o.idempotencyKey);
          return { runId: "x", deduplicated: false };
        },
      },
      routes: { "one.sub": () => [{ task: "a", input: {} }], "no.sub": () => [] },
    });
    expect(await relay.pollOnce()).toEqual({ published: 2, failed: 0 });
    expect(keys).toEqual(["k-one"]);
  });
});