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);

  it("outbox → SSE 桥:feed / decision / download / organization 事件提交后镜像到总线,feed.due 不推", async () => {
    const got: { type: string; payload: unknown }[] = [];
    const off = booted.ctx.events.subscribe((e) => {
      if (e.type !== EventType.heartbeat) got.push({ type: e.type, payload: e.payload });
    });
    const decisionId = "33333333-3333-4333-8333-333333333333";
    const feedId = "44444444-4444-4444-8444-444444444444";
    await db.transaction().execute(async (trx) => {
      await booted.ctx.outbox.emit(trx, EventType.feedDue, { feedId, reason: "manual" });
      await booted.ctx.outbox.emit(trx, EventType.feedFetched, {
        feedId,
        newItems: 2,
        updatedItems: 0,
        notModified: false,
      });
      await booted.ctx.outbox.emit(trx, EventType.decisionCreated, {
        decisionId,
        kind: "identity_unresolved",
        severity: "info",
        title: "待识别",
      });
    });
    const deadline = Date.now() + 20_000;
    while (got.length < 2 && Date.now() < deadline) await new Promise((r) => setTimeout(r, 100));
    off();
    // 同一事务内的事件 at 相同,相对顺序不保证
    got.sort((a, b) => a.type.localeCompare(b.type));
    expect(got).toEqual([
      { type: EventType.decisionCreated, payload: expect.objectContaining({ decisionId }) },
      { type: EventType.feedFetched, payload: expect.objectContaining({ feedId, newItems: 2 }) },
    ]);
    // feed.due 没有路由也不镜像,保持待认领
    const due = await sql<{ status: string }>`select status from outbox_event where type = 'feed.due'`.execute(db);
    expect(due.rows).toEqual([{ status: "pending" }]);
  }, 30_000);
});