UHD_Media_Tool/apps/server/test/bootstrap.test.ts

import { EventType } from "@umt/contracts";
import { type DB, migrateToLatest } from "@umt/db";
import { defineTask } from "@umt/tasks";
import { Kysely, PostgresDialect, sql } from "kysely";
import pg from "pg";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { z } from "zod";
import { ConfigSchema } from "../src/config.ts";
import { type Bootstrapped, bootstrap } from "../src/context.ts";
import { EventBus } from "../src/http/sse.ts";
import type { ServerModule } from "../src/modules/types.ts";
import { type RunningWorker, startWorker } from "../src/worker/index.ts";
import { withSchemaLock } from "./auth-settings/harness.ts";
import { memoryLogger } from "./helpers.ts";

// 真实 PG:独立业务 schema + 独立引擎 schema,验证 bootstrap → worker → outbox → 模块任务 的整条装配。
const url = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
const uniq = `${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`;
const biz = `server_test_${uniq}`;
const ow = `${biz}_ow`;

async function pgAvailable() {
  const c = new pg.Client({ connectionString: url, connectionTimeoutMillis: 2000 });
  try {
    await c.connect();
    await c.end();
    return true;
  } catch {
    return false;
  }
}
const hasPg = process.env.UMT_SKIP_DB_TESTS !== "1" && (await pgAvailable());

describe.skipIf(!hasPg)("bootstrap + worker(真实 PG)", () => {
  let db: Kysely<DB>;
  let booted: Bootstrapped;
  let worker: RunningWorker;
  const seen: string[] = [];

  const probe: ServerModule = {
    name: "probe",
    registerRoutes() {},
    registerTasks(registry, ctx) {
      registry.register(
        defineTask({
          name: "evaluateAcquisition",
          version: 1,
          inputSchema: z.object({ sourceReleaseId: z.string().optional() }),
          handler: async (t) => {
            const s = await ctx.settings.get("cache");
            seen.push(`${t.input.sourceReleaseId}:${s.hardThresholdPercent}`);
            return null;
          },
        }),
      );
    },
    outboxRoutes: () => ({
      [EventType.releaseResolved]: (ev) => ({
        task: "evaluateAcquisition",
        input: { sourceReleaseId: (ev.payload as { sourceReleaseId: string }).sourceReleaseId },
      }),
    }),
  };

  beforeAll(async () => {
    const admin = new pg.Client({ connectionString: url });
    await admin.connect();
    await admin.query(`create schema ${biz}`);
    await admin.end();
    db = new Kysely<DB>({
      dialect: new PostgresDialect({
        pool: new pg.Pool({ connectionString: url, max: 5, options: `-c search_path=${biz},public` }),
      }),
    });
    await withSchemaLock(() => migrateToLatest(db, { migrationTableSchema: biz }));
    await db
      .insertInto("settings")
      .values({ key: "cache", value: JSON.stringify({ hardThresholdPercent: 95, resumePercent: 90 }) })
      .execute();
    const config = ConfigSchema.parse({ DATABASE_URL: url, SECRET_MASTER_KEY: "test-master-key-0123456789" });
    const { logger } = memoryLogger();
    booted = await bootstrap({
      role: "all",
      config,
      logger,
      db,
      events: new EventBus(10),
      modules: [probe],
      engineSchema: ow,
    });
    worker = await startWorker({
      logger,
      heartbeatMs: 60_000,
      runtime: { ctx: booted.ctx, modules: [probe], setRelay: booted.outbox.setRelay, outboxIntervalMs: 100 },
    });
  }, 60_000);

  afterAll(async () => {
    await worker?.stop();
    await booted?.engine.stop();
    await db?.destroy();
    await withSchemaLock(async () => {
      const admin = new pg.Client({ connectionString: url });
      await admin.connect();
      await admin.query(`drop schema if exists ${biz} cascade`);
      await admin.query(`drop schema if exists ${ow} cascade`);
      await admin.end();
    });
  }, 60_000);

  it("settings 读取合并缺省值", async () => {
    const cache = await booted.ctx.settings.get("cache");
    expect(cache).toMatchObject({ hardThresholdPercent: 95, resumePercent: 90, checkIntervalSec: 60 });
    expect((await booted.ctx.settings.all()).automation.enabled).toBe(true);
    await expect(booted.ctx.secrets.resolve("x")).rejects.toThrow(/不存在/);
  });

  it("业务事务内 emit → 中继路由 → worker 执行模块任务;同幂等键只投一次;非法 payload 拒写", async () => {
    const payload = {
      sourceReleaseId: "11111111-1111-4111-8111-111111111111",
      observationId: null,
      mediaEntityId: "22222222-2222-4222-8222-222222222222",
      contentUnitIds: [],
      evidenceLevel: "rule_confirmed" as const,
    };
    await db.transaction().execute(async (trx) => {
      await booted.ctx.outbox.emit(trx, EventType.releaseResolved, payload, { idempotencyKey: "rr:1" });
      await booted.ctx.outbox.emit(trx, EventType.releaseResolved, payload, { idempotencyKey: "rr:1" });
    });
    await expect(
      booted.ctx.outbox.emit(db, EventType.releaseResolved, { ...payload, evidenceLevel: "bogus" } as never),
    ).rejects.toThrow(/payload/);
    expect(booted.ctx.outbox.relay).not.toBeNull();

    const deadline = Date.now() + 30_000;
    while (seen.length === 0 && Date.now() < deadline) await new Promise((r) => setTimeout(r, 200));
    expect(seen).toEqual([`${payload.sourceReleaseId}:95`]);
    const rows = await sql<{ status: string }>`select status from outbox_event`.execute(db);
    expect(rows.rows).toEqual([{ status: "dispatched" }]);
  }, 40_000);
});