UHD_Media_Tool/packages/db/test/schema.test.ts

import { readFileSync } from "node:fs";
import { Kysely, PostgresDialect, sql } from "kysely";
import { NO_MIGRATIONS } from "kysely/migration";
import pg from "pg";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { formatWithBiome, generateTypesSource, TYPES_FILE } from "../scripts/gen-types-lib.ts";
import { createMigrator, type DB, migrateToLatest } from "../src/index.ts";

// 对真实 PG(默认开发库 55433)跑迁移往返;每次测试用独立 schema,互不干扰且不碰 public。
// 无数据库环境可设 UMT_SKIP_DB_TESTS=1 跳过(CI 默认要求数据库可用)。
const url = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
const skip = process.env.UMT_SKIP_DB_TESTS === "1";
const schema = `umt_test_${process.pid}_${Date.now().toString(36)}`;

const EXPECTED_TABLES = [
  // 底座(0001/0002)
  "settings",
  "api_key",
  "audit_event",
  "outbox_event",
  "secret",
  // 实例
  "library_root",
  "downloader_instance",
  "media_server_instance",
  "path_mapping",
  // 内容 / 身份
  "media_entity",
  "content_unit",
  "title_alias",
  "external_identity",
  "identity_revision",
  // 来源
  "site",
  "site_account",
  "feed_source",
  "source_release",
  "observation",
  // 文件
  "media_asset",
  "asset_unit_coverage",
  "asset_location",
  "path_binding",
  // 规则
  "policy_revision",
  "subscription",
  "acquisition_slot",
  "preference_tag",
  "preference_suppression",
  // 执行
  "task",
  "acquisition_plan",
  "download_task",
  "asset_operation",
  // 人工/AI
  "agent_run",
  "artifact",
  "decision",
  "approval",
  "automation_grant",
  // 运维 / 整理
  "reconcile_run",
  "server_mapping",
  "organization_record",
  "organization_file_result",
  // tasks 包自有表(0013)与 agent 事件(0014)
  "tasks_run_meta",
  "tasks_fact",
  "tasks_op_ledger",
  "agent_run_event",
  // W2-B 来源/身份/元数据(0018–0020)
  "site_snapshot",
  "identity_candidate",
  "resolution_attempt",
  "identity_redirect",
  "metadata_field",
  "field_override",
  // META-1 元数据补全记录(0049)
  "metadata_enrichment",
  // W2-D Agent 装配(0021)
  "capability_invocation",
  "agent_run_message",
  // 登录门禁(0015)与 SSE 多进程总线(0016)
  "admin_user",
  "admin_session",
  "login_attempt",
  "sse_event",
  // W4-C 浏览器实体(0023)
  "browser_provider",
  "browser_session",
  "browser_tab",
  "browser_control_lease",
  // W5-A 文件管理回收站(0024)
  "trash_item",
  // W5-C 订阅增强(0025)
  "slot_deferral",
  "slot_unit",
  "subscription_history",
  // W7-A Pi 上游(0027)
  "pi_upstream",
  // W7-C 定时任务注册表(0028)
  "scheduled_job",
  "scheduled_job_run",
  // W9-A 站点用户数据(0031)
  "site_user_data",
  // W9-C 命名模板版本(0032)
  "naming_template_version",
  // W9-D 规则组(0033)
  "rule_group",
  "rule_atom",
  // W9-I 进程心跳(0034)
  "process_heartbeat",
  "process_heartbeat_sample",
  // AG-1 MCP 服务端配置(0039)
  "mcp_server_config",
  // C1-B 恢复码 / Passkey / WebAuthn 挑战(0037)
  "admin_recovery_code",
  "admin_passkey",
  "webauthn_challenge",
  // C2-A 跨进程站点限流(0041)
  "rate_limit_bucket",
  // C2-A PG 短期缓存(0042,UNLOGGED)
  "kv_cache",
].sort();

/** 不走 entityTable() 的表:追加型/引擎自有表,没有 uuid id + created_at/updated_at 三件套 */
const NON_ENTITY_TABLES = [
  "settings",
  "audit_event",
  "outbox_event",
  "tasks_run_meta",
  "tasks_fact",
  "tasks_op_ledger",
  "agent_run_event",
  "login_attempt",
  "sse_event",
  // 稳定 key 作主键 / 只追加的运行历史
  "scheduled_job",
  "scheduled_job_run",
  // 进程心跳:instance_id 主键 / 只追加的采样
  "process_heartbeat",
  "process_heartbeat_sample",
  // 单行配置(id = 1)
  "mcp_server_config",
  // 一次性凭据:只记创建 / 用掉的时间
  "admin_recovery_code",
  "webauthn_challenge",
  // 限流桶:key 主键,只有 updated_at
  "rate_limit_bucket",
  // 缓存:(region, key) 主键
  "kv_cache",
];

describe.skipIf(skip)("schema migrations (real PG)", () => {
  let admin: Kysely<unknown>;
  let db: Kysely<DB>;

  beforeAll(async () => {
    admin = new Kysely({ dialect: new PostgresDialect({ pool: new pg.Pool({ connectionString: url, max: 1 }) }) });
    await sql.raw(`create schema ${schema}`).execute(admin);
    db = new Kysely<DB>({
      dialect: new PostgresDialect({
        pool: new pg.Pool({ connectionString: url, max: 2, options: `-c search_path=${schema},public` }),
      }),
    });
    await migrateToLatest(db, { migrationTableSchema: schema });
  });

  afterAll(async () => {
    await db?.destroy();
    await sql.raw(`drop schema if exists ${schema} cascade`).execute(admin);
    await admin?.destroy();
  });

  const tables = async () =>
    (
      await sql<{ t: string }>`select table_name as t from information_schema.tables
        where table_schema = current_schema() and table_type = 'BASE TABLE' and table_name not like 'kysely_%'
        order by 1`.execute(db)
    ).rows.map((r) => r.t);

  it("建齐任务书 §3.1-2 全部表", async () => {
    expect(await tables()).toEqual(EXPECTED_TABLES);
  });

  it("src/types.ts 与迁移结果一致(无漂移)", async () => {
    const generated = formatWithBiome(await generateTypesSource(db));
    expect(generated).toBe(readFileSync(TYPES_FILE, "utf8"));
  });

  it("新实体表都有 uuid 主键默认值与 created_at/updated_at", async () => {
    const rows = (
      await sql<{ table_name: string; column_name: string; column_default: string | null }>`
        select table_name, column_name, column_default from information_schema.columns
         where table_schema = current_schema() and column_name in ('id', 'created_at', 'updated_at')`.execute(db)
    ).rows;
    for (const t of EXPECTED_TABLES.filter((x) => !NON_ENTITY_TABLES.includes(x))) {
      const cols = rows.filter((r) => r.table_name === t);
      expect(cols.map((c) => c.column_name).sort(), t).toEqual(["created_at", "id", "updated_at"]);
      expect(cols.find((c) => c.column_name === "id")?.column_default, t).toBe("gen_random_uuid()");
    }
  });

  it("updated_at 触发器与 CHECK 状态约束", async () => {
    const e = await db
      .insertInto("media_entity")
      .values({ kind: "movie", title: "沙丘2", year: 2024 })
      .returningAll()
      .executeTakeFirstOrThrow();
    await sql`select pg_sleep(0.01)`.execute(db);
    const u = await db
      .updateTable("media_entity")
      .set({ title: "沙丘:第二部" })
      .where("id", "=", e.id)
      .returningAll()
      .executeTakeFirstOrThrow();
    expect(u.updated_at.getTime()).toBeGreaterThanOrEqual(e.updated_at.getTime());
    await expect(
      sql`insert into acquisition_slot (media_entity_id, name, status) values (${e.id}, 'x', 'bogus')`.execute(db),
    ).rejects.toThrow(/check constraint/);
  });

  it("external_identity:accepted 唯一,候选可并存", async () => {
    const [a, b] = await db
      .insertInto("media_entity")
      .values([
        { kind: "movie", title: "A" },
        { kind: "movie", title: "B" },
      ])
      .returning("id")
      .execute();
    const base = { provider: "tmdb", entity_type: "movie" as const, external_id: "693134" };
    await db
      .insertInto("external_identity")
      .values([
        { ...base, media_entity_id: a!.id, status: "proposed" },
        { ...base, media_entity_id: b!.id, status: "proposed" },
        { ...base, media_entity_id: a!.id, status: "accepted" },
        { ...base, entity_type: "tv", media_entity_id: b!.id, status: "accepted" },
      ])
      .execute();
    await expect(
      db
        .insertInto("external_identity")
        .values({ ...base, media_entity_id: b!.id, status: "accepted" })
        .execute(),
    ).rejects.toThrow(/external_identity_accepted_key/);
  });

  it("source_release / observation / download_task / Slot 预留 唯一约束", async () => {
    const site = await db
      .insertInto("site")
      .values({ key: "audiences", name: "观众", adapter: "nexusphp", base_url: "https://example.invalid" })
      .returning("id")
      .executeTakeFirstOrThrow();
    const rel = await db
      .insertInto("source_release")
      .values({ site_id: site.id, site_release_id: "123", title: "Dune.Part.Two.2024.2160p" })
      .returning("id")
      .executeTakeFirstOrThrow();
    await expect(
      db.insertInto("source_release").values({ site_id: site.id, site_release_id: "123", title: "dup" }).execute(),
    ).rejects.toThrow(/source_release_site_release_key/);

    const feed = await db
      .insertInto("feed_source")
      .values({ site_id: site.id, name: "电影", url_display: "https://example.invalid/rss?passkey=[REDACTED]" })
      .returning("id")
      .executeTakeFirstOrThrow();
    const obs = { feed_id: feed.id, guid: "g1", content_digest: "d1", title: "t" };
    await db.insertInto("observation").values(obs).execute();
    await expect(db.insertInto("observation").values(obs).execute()).rejects.toThrow(/observation_feed_guid_key/);

    const qb = await db
      .insertInto("downloader_instance")
      .values({ name: "qb1", base_url: "http://qb.invalid" })
      .returning("id")
      .executeTakeFirstOrThrow();
    const dt = { instance_id: qb.id, torrent_hash: "a".repeat(40), name: "x" };
    await db.insertInto("download_task").values(dt).execute();
    await expect(db.insertInto("download_task").values(dt).execute()).rejects.toThrow(
      /download_task_instance_hash_key/,
    );

    const ent = await db
      .insertInto("media_entity")
      .values({ kind: "movie", title: "C" })
      .returning("id")
      .executeTakeFirstOrThrow();
    const slot = await db
      .insertInto("acquisition_slot")
      .values({ media_entity_id: ent.id, name: "archival" })
      .returning("id")
      .executeTakeFirstOrThrow();
    await expect(
      db.insertInto("acquisition_slot").values({ media_entity_id: ent.id, name: "archival" }).execute(),
    ).rejects.toThrow(/acquisition_slot_target_key/);
    const plan = (k: string) => ({
      slot_id: slot.id,
      source_release_id: rel.id,
      status: "reserved" as const,
      payload_hash: "h",
      idempotency_key: k,
    });
    await db.insertInto("acquisition_plan").values(plan("k1")).execute();
    // 0025 起剧集 Slot 允许多个单元不相交的在途计划(服务端在 Slot 行锁内按单元校验);幂等键仍唯一
    await db.insertInto("acquisition_plan").values(plan("k2")).execute();
    await expect(db.insertInto("acquisition_plan").values(plan("k2")).execute()).rejects.toThrow(
      /acquisition_plan_idempotency_key/,
    );
  });

  it("path_binding:同作品同季只能有一个 active 绑定", async () => {
    const root = await db
      .insertInto("library_root")
      .values({ name: "movies", local_path: "/media/movies" })
      .returning("id")
      .executeTakeFirstOrThrow();
    const ent = await db
      .insertInto("media_entity")
      .values({ kind: "series", title: "S" })
      .returning("id")
      .executeTakeFirstOrThrow();
    const b = { media_entity_id: ent.id, library_root_id: root.id, binding_kind: "entity" as const };
    await db
      .insertInto("path_binding")
      .values({ ...b, relative_path: "S (2023)" })
      .execute();
    await expect(
      db
        .insertInto("path_binding")
        .values({ ...b, relative_path: "S2 (2023)" })
        .execute(),
    ).rejects.toThrow(/path_binding_active_key/);
    await db
      .insertInto("path_binding")
      .values({ ...b, relative_path: "S2 (2023)", status: "conflict" })
      .execute();
  });

  it("organization_record.change_seq 随更新与文件明细变化递增", async () => {
    const rec = await db
      .insertInto("organization_record")
      .values({ trigger: "auto", actor_type: "system", action: "copy" })
      .returningAll()
      .executeTakeFirstOrThrow();
    const upd = await db
      .updateTable("organization_record")
      .set({ status: "running" })
      .where("id", "=", rec.id)
      .returning("change_seq")
      .executeTakeFirstOrThrow();
    expect(upd.change_seq).toBeGreaterThan(rec.change_seq);
    await db
      .insertInto("organization_file_result")
      .values({ organization_record_id: rec.id, mode: "copy", source_path: "/downloads/a.mkv" })
      .execute();
    const after = await db
      .selectFrom("organization_record")
      .select("change_seq")
      .where("id", "=", rec.id)
      .executeTakeFirstOrThrow();
    expect(after.change_seq).toBeGreaterThan(upd.change_seq);
    expect(typeof after.change_seq).toBe("number");
  });

  it("title_alias 上有 gin trgm 索引,可模糊召回", async () => {
    const ent = await db
      .insertInto("media_entity")
      .values({ kind: "series", title: "最后生还者" })
      .returning("id")
      .executeTakeFirstOrThrow();
    await db
      .insertInto("title_alias")
      .values({ media_entity_id: ent.id, alias: "The Last of Us", normalized: "the last of us", source: "tmdb" })
      .execute();
    const hit = await sql<{ id: string }>`select media_entity_id as id from title_alias
      where normalized % ${"last of us"}`.execute(db);
    expect(hit.rows.map((r) => r.id)).toContain(ent.id);
  });

  it("agent_run_event:(run_id, seq) 唯一,删 run 级联删事件", async () => {
    const run = await db
      .insertInto("agent_run")
      .values({ trigger: "user", goal: "g" })
      .returning("id")
      .executeTakeFirstOrThrow();
    const ev = { run_id: run.id, seq: 0, type: "run_started" as const, payload: JSON.stringify({ goal: "g" }) };
    await db.insertInto("agent_run_event").values(ev).execute();
    await expect(db.insertInto("agent_run_event").values(ev).execute()).rejects.toThrow(/agent_run_event_pkey/);
    await expect(
      sql`insert into agent_run_event (run_id, seq, type) values (${run.id}, 1, 'bogus')`.execute(db),
    ).rejects.toThrow(/check constraint/);
    await db.deleteFrom("agent_run").where("id", "=", run.id).execute();
    const left = await db.selectFrom("agent_run_event").select("seq").where("run_id", "=", run.id).execute();
    expect(left).toEqual([]);
  });

  it("迁移可完整 down 再 up(往返)", async () => {
    const migrator = createMigrator(db, { migrationTableSchema: schema });
    const down = await migrator.migrateTo(NO_MIGRATIONS);
    expect(down.error).toBeUndefined();
    expect(await tables()).toEqual([]);
    const up = await migrator.migrateToLatest();
    expect(up.error).toBeUndefined();
    expect(await tables()).toEqual(EXPECTED_TABLES);
  }, 60_000);

  it("迁移器只内省目标 schema:别处同时建删 schema 不影响(并行测试共用库)", async () => {
    const migrator = createMigrator(db, { migrationTableSchema: schema });
    let stop = false;
    const churn = (async () => {
      for (let i = 0; !stop && i < 200; i++) {
        const tmp = `${schema}_churn_${i}`;
        await sql.raw(`create schema ${tmp}; create table ${tmp}.t (id serial primary key)`).execute(admin);
        await sql.raw(`drop schema ${tmp} cascade`).execute(admin);
      }
    })();
    try {
      for (let i = 0; i < 20; i++) {
        const r = await migrator.migrateToLatest();
        expect(r.error).toBeUndefined();
      }
    } finally {
      stop = true;
      await churn;
    }
  }, 60_000);
});