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/index.ts";
import { createTestRegistry } from "./fixtures/tasks.ts";
import {
  connectDb,
  createSideEffectTable,
  dropSchemas,
  migrateBizSchema,
  pgAvailable,
  resetSchemas,
  setupEngine,
  type TestDb,
  testSchemas,
} from "./helpers.ts";

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 migrateBizSchema(db, schemas.biz);
    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 status = 'pending'`.execute(db);
    expect(pending.rows).toEqual([{ type: "other.type" }]); // 未配置路由的 type 不认领
    const dispatched = await sql<{ n: string }>`
      SELECT count(*) n FROM outbox_event WHERE status = 'dispatched' AND published_at IS NOT NULL AND attempts = 1`.execute(
      db,
    );
    // @umt/db 把 int8 解析为 number(全局 pg 类型解析器),这里统一转成数字比较
    expect(Number(dispatched.rows[0]!.n)).toBe(60);

    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([Number(runs.rows[0]!.n), Number(runs.rows[0]!.keys)]).toEqual([60, 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<{ status: string; attempts: number; backoff: boolean; err: string }>`
      SELECT status, attempts, next_attempt_at > now() backoff, last_error->>'message' err FROM outbox_event
      WHERE published_at IS NULL AND type = 'echo.requested'`.execute(db);
    expect(left.rows).toHaveLength(1);
    expect(left.rows[0]).toMatchObject({ status: "pending", attempts: 1, backoff: true });
    // 退避期内不再认领;到期后再失败达到上限置 dead
    expect(await relay.pollOnce()).toEqual({ published: 0, failed: 0 });
    await sql`UPDATE outbox_event SET next_attempt_at = now() WHERE type = 'echo.requested' AND published_at IS NULL`.execute(
      db,
    );
    const strict = new OutboxRelay({
      db,
      engine,
      routes: { "echo.requested": (ev) => ({ task: "echo", input: ev.payload }) },
      logSink: () => {},
      maxAttempts: 2,
    });
    expect(await strict.pollOnce()).toEqual({ published: 0, failed: 1 });
    const dead = await sql<{
      status: string;
    }>`SELECT status FROM outbox_event WHERE type = 'echo.requested' AND published_at IS NULL`.execute(db);
    expect(dead.rows).toEqual([{ status: "dead" }]);
  });

  it("mirror:投递成功后按顺序镜像到实时总线;只镜像不路由的 type 也认领,失败的不镜像", async () => {
    await sql`DELETE FROM outbox_event`.execute(db);
    await sql`INSERT INTO outbox_event (type, payload, at) VALUES
      ('mirror.only', '{"recordId":"r1"}'::jsonb, now() - interval '3 second'),
      ('echo.requested', '{"n":1001}'::jsonb, now() - interval '2 second'),
      ('echo.requested', '{"n":"bad"}'::jsonb, now() - interval '1 second'),
      ('not.mirrored', '{}'::jsonb, now())`.execute(db);
    const seen: { type: string; payload: unknown }[] = [];
    const relay = new OutboxRelay({
      db,
      engine,
      routes: { "echo.requested": (ev) => ({ task: "echo", input: ev.payload }) },
      logSink: () => {},
      mirror: {
        types: ["mirror.only", "echo.requested"],
        publish: (ev) => {
          seen.push({ type: ev.type, payload: ev.payload });
          if (ev.type === "mirror.only") throw new Error("bus down"); // 推送失败不影响投递
        },
      },
    });
    expect(await relay.pollOnce()).toEqual({ published: 2, failed: 1 });
    expect(seen).toEqual([
      { type: "mirror.only", payload: { recordId: "r1" } },
      { type: "echo.requested", payload: { n: 1001 } },
    ]);
    const status = await sql<{ type: string; status: string }>`
      SELECT type, status FROM outbox_event ORDER BY at`.execute(db);
    expect(status.rows.map((r) => `${r.type}:${r.status}`)).toEqual([
      "mirror.only:dispatched",
      "echo.requested:dispatched",
      "echo.requested:pending",
      "not.mirrored:pending",
    ]);
  });
});