UHD_Media_Tool/packages/tasks/test/outbox.test.ts

import { sql } from "kysely";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { type EnqueueOptions, OutboxRelay, type TaskEngine } from "../src";
import { createTestRegistry } from "./fixtures/tasks";
import {
  connectDb,
  createOutboxTable,
  createSideEffectTable,
  dropSchemas,
  pgAvailable,
  resetSchemas,
  setupEngine,
  type TestDb,
  testSchemas,
} from "./helpers";

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

describe.skipIf(!hasPg)("OutboxRelay", () => {
  let db: TestDb;
  let engine: Awaited<ReturnType<typeof setupEngine>>;

  beforeAll(async () => {
    await resetSchemas(schemas);
    db = connectDb(schemas.biz);
    await createOutboxTable(db);
    await createSideEffectTable(db);
    engine = await setupEngine(db, schemas, createTestRegistry(db));
  }, 30_000);

  afterAll(async () => {
    await engine?.stop();
    await db?.destroy();
    await dropSchemas(schemas);
  }, 30_000);

  it("两个 relay 并发处理 60 条事件:每条只 enqueue 一次,全部标记已投递", async () => {
    // 业务事务写 outbox
    await db.transaction().execute(async (trx) => {
      for (let i = 0; i < 60; i++) {
        await sql`INSERT INTO outbox_event (type, payload) VALUES ('echo.requested', ${JSON.stringify({ n: i, traceId: `tr-${i}` })}::jsonb)`.execute(
          trx,
        );
      }
      await sql`INSERT INTO outbox_event (type, payload) VALUES ('other.type', '{}'::jsonb)`.execute(trx);
    });

    const calls: string[] = [];
    // 包一层计数,确认没有重复 enqueue(引擎幂等只是兜底)
    const counting: Pick<TaskEngine, "enqueue"> = {
      enqueue: async (task: string, input: unknown, opts: EnqueueOptions) => {
        calls.push(opts.idempotencyKey);
        await new Promise((r) => setTimeout(r, 5));
        return engine.enqueue(task, input, opts);
      },
    };
    const routes = { "echo.requested": (ev: { payload: unknown }) => ({ task: "echo", input: ev.payload }) };
    const relays = [0, 1].map((i) => new OutboxRelay({ db, engine: counting, routes, batchSize: 7 + i }));

    const drain = async (r: OutboxRelay) => {
      const perRelay = { published: 0 };
      for (;;) {
        const res = await r.pollOnce();
        perRelay.published += res.published;
        if (res.published === 0 && res.failed === 0) return perRelay.published;
      }
    };
    const [p0, p1] = await Promise.all(relays.map(drain));

    expect(p0! + p1!).toBe(60);
    expect(p0).toBeGreaterThan(0);
    expect(p1).toBeGreaterThan(0);
    expect(calls).toHaveLength(60);
    expect(new Set(calls).size).toBe(60);

    const pending = await sql<{ type: string }>`SELECT type FROM outbox_event WHERE published_at IS NULL`.execute(db);
    expect(pending.rows).toEqual([{ type: "other.type" }]); // 未配置路由的 type 不认领

    const runs = await sql<{ n: string; keys: string }>`
      SELECT count(*) n, count(DISTINCT idempotency_key) keys FROM tasks_run_meta WHERE task_name = 'echo'`.execute(db);
    expect(runs.rows[0]).toEqual({ n: "60", keys: "60" });
    const traced = await sql<{ trace_id: string }>`
      SELECT m.trace_id FROM tasks_run_meta m
      JOIN outbox_event o ON m.idempotency_key = 'outbox:' || o.id::text
      WHERE o.payload->>'n' = '3'`.execute(db);
    expect(traced.rows).toEqual([{ trace_id: "tr-3" }]);
  }, 60_000);

  it("enqueue 失败的事件保持未投递,下一轮重试", async () => {
    await sql`INSERT INTO outbox_event (type, payload) VALUES ('echo.requested', '{"n":"not-a-number"}'::jsonb)`.execute(
      db,
    );
    const relay = new OutboxRelay({
      db,
      engine,
      routes: { "echo.requested": (ev) => ({ task: "echo", input: ev.payload }) },
      logSink: () => {},
    });
    expect(await relay.pollOnce()).toEqual({ published: 0, failed: 1 });
    const left = await sql<{ n: string }>`
      SELECT count(*) n FROM outbox_event WHERE published_at IS NULL AND type = 'echo.requested'`.execute(db);
    expect(left.rows[0]!.n).toBe("1");
  });
});