UHD_Media_Tool/apps/server/test/e2e/harness.ts

import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { FakeScript } from "@umt/agent";
import { type DB, migrateToLatest } from "@umt/db";
import {
  createMockQb,
  createMockTmdb,
  createTempTree,
  type MockQb,
  type MockTmdb,
  type TempTree,
  TMDB_FIXTURES,
  type TmdbFixture,
} from "@umt/testkit";
import { Kysely, PostgresDialect, sql } from "kysely";
import pg from "pg";
import { ConfigSchema } from "../../src/config.ts";
import { type Bootstrapped, bootstrap } from "../../src/context.ts";
import { createApp } from "../../src/http/app.ts";
import { EventBus } from "../../src/http/sse.ts";
import { createLogger } from "../../src/logger.ts";
import type { BrowserModuleOptions } from "../../src/modules/browser/service.ts";
import { setDefaultUpstream } from "../../src/modules/pi/store.ts";
import type { ServerModule } from "../../src/modules/types.ts";
import type { WebSearchModuleOptions } from "../../src/modules/websearch/index.ts";
import { type RunningWorker, startWorker } from "../../src/worker/index.ts";
import { withSchemaLock } from "../auth-settings/harness.ts";
import { buildModules } from "./fixtures/modules.ts";

// 端到端夹具:真实 PG(每个文件独立业务 schema + 独立引擎 schema)+ 真实四个模块 + OpenWorkflow 引擎/worker/outbox 中继
// + testkit mock(qB / TMDB / RSS)+ 临时目录(缓存卷、媒体库、Agent 会话)。HTTP 走 createApp(DEV key = admin)。

export const PG_URL = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
export const DEV_KEY = "dev-admin-e2e-key";
export const MASTER_KEY = "e2e-master-key-0123456789";
/** mock TMDB 的 v3 key(32 位十六进制,经设置 metadata.tmdbApiKey 录入) */
export const TMDB_KEY = "0123456789abcdef0123456789abcdef";

export async function pgAvailable(): Promise<boolean> {
  if (process.env.UMT_SKIP_DB_TESTS === "1") return false;
  const c = new pg.Client({ connectionString: PG_URL, connectionTimeoutMillis: 2000 });
  try {
    await c.connect();
    await c.end();
    return true;
  } catch {
    return false;
  }
}

// biome-ignore lint/suspicious/noExplicitAny: 测试断言直接读 JSON
type Loose = any;

export interface E2E {
  name: string;
  biz: string;
  ow: string;
  db: Kysely<DB>;
  booted: Bootstrapped;
  modules: ServerModule[];
  bus: EventBus;
  worker: RunningWorker | null;
  qb: MockQb;
  tmdb: MockTmdb;
  lib: TempTree;
  cache: TempTree;
  sessionDir: string;
  /** Hono 应用(管理员会话用例用 cookie 调用,见 auth-settings/harness.ts 的 Browser) */
  app: ReturnType<typeof createApp>;
  api(method: string, path: string, body?: unknown): Promise<{ status: number; body: Loose }>;
  /** PATCH /settings 并断言成功 */
  settings(patch: Record<string, unknown>): Promise<Loose>;
  startWorker(): Promise<void>;
  close(): Promise<void>;
}

export interface E2EOptions {
  fakeScripts?: FakeScript[];
  /** false:只做 API/引擎客户端(崩溃恢复用例由子进程跑 worker) */
  worker?: boolean;
  tmdbFixtures?: TmdbFixture[];
  /** 网页搜索模块注入(E13) */
  webSearch?: WebSearchModuleOptions;
  /** 浏览器模块注入(E14) */
  browser?: BrowserModuleOptions;
}

export async function waitFor<T>(
  fn: () => Promise<T | null | undefined | false>,
  ms: number,
  what: string,
  every = 200,
): Promise<T> {
  const deadline = Date.now() + ms;
  let last: unknown;
  while (Date.now() < deadline) {
    try {
      const v = await fn();
      if (v) return v as T;
    } catch (err) {
      last = err;
    }
    await new Promise((r) => setTimeout(r, every));
  }
  throw new Error(`等待超时:${what}${last ? `(${(last as Error).message})` : ""}`);
}

export async function createE2E(name: string, opts: E2EOptions = {}): Promise<E2E> {
  const uniq = `${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`;
  const biz = `e2e_${name}_${uniq}`;
  const ow = `${biz}_ow`;
  const admin = new pg.Client({ connectionString: PG_URL });
  await admin.connect();
  await admin.query(`create schema ${biz}`);
  await admin.end();
  const db = new Kysely<DB>({
    dialect: new PostgresDialect({
      pool: new pg.Pool({ connectionString: PG_URL, max: 10, options: `-c search_path=${biz},public` }),
    }),
  });
  await withSchemaLock(() => migrateToLatest(db, { migrationTableSchema: biz }));

  const qb = await createMockQb({ addVisibleAfterPolls: 0 });
  const tmdb = await createMockTmdb({ apiKey: TMDB_KEY, fixtures: opts.tmdbFixtures ?? TMDB_FIXTURES });
  const lib = await createTempTree({}, { sentinel: true, prefix: "umt-e2e-lib-" });
  const cache = await createTempTree({}, { prefix: "umt-e2e-cache-" });
  const sessionDir = await mkdtemp(join(tmpdir(), "umt-e2e-sessions-"));

  const logger = createLogger({ level: process.env.E2E_LOG_LEVEL === "debug" ? "debug" : "silent" });
  const config = ConfigSchema.parse({
    DATABASE_URL: PG_URL,
    SECRET_MASTER_KEY: MASTER_KEY,
    DEV_ADMIN_KEY: DEV_KEY,
    NODE_ENV: "development",
  });
  const bus = new EventBus(5000);
  const modules = buildModules({
    sessionDir,
    ...(opts.fakeScripts ? { fakeScripts: opts.fakeScripts } : {}),
    ...(opts.webSearch ? { webSearch: opts.webSearch } : {}),
    ...(opts.browser ? { browser: opts.browser } : {}),
  });
  const booted = await bootstrap({
    role: "all",
    config,
    logger,
    db,
    events: bus,
    modules,
    engineSchema: ow,
    engineMigrations: true,
  });
  const app = createApp({
    config: { NODE_ENV: "development", DEV_ADMIN_KEY: DEV_KEY },
    logger,
    bus,
    role: "all",
    version: "e2e",
    db,
    modules,
    ctx: booted.ctx,
  });

  let worker: RunningWorker | null = null;
  const e: E2E = {
    name,
    biz,
    ow,
    db,
    booted,
    modules,
    bus,
    worker,
    qb,
    tmdb,
    lib,
    cache,
    sessionDir,
    app,
    async api(method, path, body) {
      const res = await app.request(`/api/v1${path}`, {
        method,
        headers: {
          authorization: `Bearer ${DEV_KEY}`,
          ...(body !== undefined ? { "content-type": "application/json" } : {}),
        },
        ...(body !== undefined ? { body: JSON.stringify(body) } : {}),
      });
      return { status: res.status, body: await res.json() };
    },
    async settings(patch) {
      const r = await e.api("PATCH", "/settings", patch);
      if (r.status !== 200) throw new Error(`PATCH /settings 失败:${JSON.stringify(r.body)}`);
      return r.body.data;
    },
    async startWorker() {
      worker = await startWorker({
        logger,
        heartbeatMs: 60_000,
        bus,
        runtime: {
          ctx: booted.ctx,
          modules,
          setRelay: booted.outbox.setRelay,
          outboxIntervalMs: 100,
          // 并发显式固定,不随 WORKER_CONCURRENCY 等环境变量漂移
          concurrency: 4,
          agentConcurrency: 2,
        },
      });
      e.worker = worker;
    },
    async close() {
      await worker?.stop().catch(() => {});
      await booted.engine.stop().catch(() => {});
      for (const m of [...modules].reverse()) await m.stop?.();
      await db.destroy();
      await qb.close();
      await tmdb.close();
      await lib.cleanup();
      await cache.cleanup();
      await rm(sessionDir, { recursive: true, force: true });
      await withSchemaLock(async () => {
        const c = new pg.Client({ connectionString: PG_URL });
        await c.connect();
        await c.query(`drop schema if exists ${biz} cascade`);
        await c.query(`drop schema if exists ${ow} cascade`);
        await c.end();
      });
    },
  };
  // 通用设置:fake 上游(脚本化假模型,设为默认)、TMDB 走设置里的 secret、结算快一点
  const fakeUpstream = await db
    .insertInto("pi_upstream")
    .values({
      name: "fake",
      provider: "fake",
      auth_kind: "none",
      default_model: "fake-model",
      thinking_level: "medium",
      extra: JSON.stringify({}),
      enabled: true,
      status: "ok",
    })
    .returning("id")
    .executeTakeFirstOrThrow();
  await setDefaultUpstream(booted.ctx, fakeUpstream.id);
  await e.settings({
    metadata: { tmdbApiKey: TMDB_KEY, tmdbBaseUrl: `${tmdb.url}/3` },
    retention: { seedRetentionDays: 0 },
    // 整理最小体积与模块端口的 minVideoBytes 一致(测试文件 ≥ 100 KB)
    organize: { minSizeBytes: 100_000 },
  });
  if (opts.worker !== false) await e.startWorker();
  return e;
}

/** 媒体库根 + qB 实例(保存路径 = 缓存临时目录)+ 站点 */
export async function seedBase(e: E2E, site: { key?: string; baseUrl?: string } = {}) {
  const root = await e.db
    .insertInto("library_root")
    .values({ name: `lib-${e.name}`, media_kind: "mixed", local_path: e.lib.root, management_mode: "managed" })
    .returning("id")
    .executeTakeFirstOrThrow();
  const inst = await e.db
    .insertInto("downloader_instance")
    .values({
      name: `qb-${e.name}`,
      base_url: e.qb.url,
      username: "admin",
      settings: JSON.stringify({ savePath: e.cache.root }),
    })
    .returning("id")
    .executeTakeFirstOrThrow();
  const s = await e.db
    .insertInto("site")
    .values({
      key: site.key ?? "audiences",
      name: "观众",
      adapter: "nexusphp",
      base_url: site.baseUrl ?? "https://audiences.example/",
      rate_limit: JSON.stringify({ minIntervalMs: 0, concurrency: 5 }),
    })
    .returning("id")
    .executeTakeFirstOrThrow();
  return { rootId: root.id, instanceId: inst.id, siteId: s.id };
}

/** 作品 + 订阅 + 1080p Slot + 两个已识别的站点资源(720p / 1080p);不发 release.resolved,只有 Agent 会去规划 */
export async function seedMovieTarget(e: E2E, siteId: string) {
  const entity = await e.db
    .insertInto("media_entity")
    .values({ kind: "movie", title: "沙丘2", original_title: "Dune: Part Two", year: 2024, status: "active" })
    .returning("id")
    .executeTakeFirstOrThrow();
  const sub = await e.db
    .insertInto("subscription")
    .values({ media_entity_id: entity.id, media_kind: "movie", mode: "once" })
    .returning("id")
    .executeTakeFirstOrThrow();
  const slot = await e.db
    .insertInto("acquisition_slot")
    .values({
      media_entity_id: entity.id,
      subscription_id: sub.id,
      name: "1080p",
      hard_conditions: JSON.stringify({ resolutions: ["1080p"] }),
    })
    .returning("id")
    .executeTakeFirstOrThrow();
  const rel = async (title: string, n: number) =>
    (
      await e.db
        .insertInto("source_release")
        .values({
          site_id: siteId,
          site_release_id: String(880000 + n),
          title,
          size_bytes: 300_000,
          media_entity_id: entity.id,
          resolution_status: "resolved",
          raw: JSON.stringify({ _site: { hr: false } }),
        })
        .returning("id")
        .executeTakeFirstOrThrow()
    ).id;
  const r720 = await rel("Dune.Part.Two.2024.720p.WEB-DL.x264-AAA", 1);
  const r1080 = await rel("Dune.Part.Two.2024.1080p.BluRay.x264-GRP", 2);
  return { entityId: entity.id, slotId: slot.id, r720, r1080 };
}

export async function runRow(e: E2E, runId: string) {
  return e.db.selectFrom("agent_run").selectAll().where("id", "=", runId).executeTakeFirstOrThrow();
}

export async function runEvents(e: E2E, runId: string) {
  return e.db.selectFrom("agent_run_event").selectAll().where("run_id", "=", runId).orderBy("seq").execute();
}

/** 等 mock qB 收到某个种子后置为完成并落盘 */
export async function completeDownload(e: E2E, title: string, ms = 60_000) {
  const { torrentFor } = await import("./fixtures/modules.ts");
  const hash = torrentFor(title).hash;
  await waitFor(async () => e.qb.get(hash), ms, `qB 收到 ${title}`);
  await e.qb.complete(hash, { writeFiles: true });
  return hash;
}

export { sql };