UHD_Media_Tool/apps/server/test/file-manager/harness.ts

import { randomUUID } from "node:crypto";
import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import {
  CSRF_HEADER,
  capabilitySpec,
  DEFAULT_SETTINGS,
  GlobalSettings,
  SESSION_COOKIE,
  type SettingsKey,
} from "@umt/contracts";
import { type DB, migrateToLatest } from "@umt/db";
import { normalizeTitle } from "@umt/domain";
import { Kysely, PostgresDialect, sql } from "kysely";
import pg from "pg";
import { ConfigSchema } from "../../src/config.ts";
import { createOutboxPort } from "../../src/context.ts";
import { createApp } from "../../src/http/app.ts";
import { deriveCsrfToken, hashApiKey, hashSessionToken } from "../../src/http/auth.ts";
import { EventBus } from "../../src/http/sse.ts";
import { createLogger } from "../../src/logger.ts";
import { fakeEngine } from "../../src/modules/acquisition-organize/__tests__/harness.ts";
import { executeRecord, type FrozenPlan } from "../../src/modules/acquisition-organize/services/organize.ts";
import { createCapabilityRegistry } from "../../src/modules/agent/capability-registry.ts";
import { createFileManagerModule, type FmOverrides } from "../../src/modules/file-manager/index.ts";
import type { ServerContext } from "../../src/modules/types.ts";

// 文件管理测试装配:真实 PG(独立 schema)+ 临时目录(库根 / 下载根 / 根外目录)+ 记录入队的假引擎。

export const DB_URL = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
export const DEV_KEY = "dev-admin-fm-test-key";
export const READ_KEY = "fm-organize-read-key";
export const EXEC_KEY = "fm-organize-exec-key";
export const ADMIN_KEY = "fm-admin-scope-key";
/** 传给 api()/raw() 的 key:改用管理员控制台会话 cookie(+CSRF),而不是 Bearer */
export const SESSION = "__admin_session__";
const SESSION_TOKEN = "fm-test-admin-session-token-0123456789";
const KEYS = [
  { key: READ_KEY, scopes: ["organize.read"] },
  { key: EXEC_KEY, scopes: ["organize.execute"] },
  { key: ADMIN_KEY, scopes: ["admin"] },
];

export async function pgAvailable() {
  if (process.env.UMT_SKIP_DB_TESTS === "1") return false;
  const c = new pg.Client({ connectionString: DB_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 FmHarness {
  db: Kysely<DB>;
  engine: ReturnType<typeof fakeEngine>;
  module: ReturnType<typeof createFileManagerModule>;
  events: { type: string; payload: Loose }[];
  base: string;
  lib: string;
  dl: string;
  outside: string;
  libRoot: string;
  dlRoot: string;
  libraryRootId: string;
  instanceId: string;
  api(method: string, path: string, body?: unknown, key?: string): Promise<{ status: number; body: Loose }>;
  /** 原始响应(非 JSON 接口,如 /files/download) */
  raw(method: string, path: string, key?: string): Promise<Response>;
  close(): Promise<void>;
}

export async function createFmHarness(
  name: string,
  overrides: FmOverrides = {},
  /** runTransfers(默认 true):假引擎收到 /files 移动/复制的 verifyAndImport 时就地执行(模拟 worker) */
  opts: { catalog?: boolean; runTransfers?: boolean } = {},
): Promise<FmHarness> {
  const schema = `fm_test_${name}_${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`;
  const admin = new pg.Client({ connectionString: DB_URL });
  await admin.connect();
  await admin.query(`create schema ${schema}`);
  await admin.end();
  const db = new Kysely<DB>({
    dialect: new PostgresDialect({
      pool: new pg.Pool({ connectionString: DB_URL, max: 6, options: `-c search_path=${schema},public` }),
    }),
  });
  await migrateToLatest(db, { migrationTableSchema: schema });

  const base = await mkdtemp(join(tmpdir(), `fm-${name}-`));
  const lib = join(base, "lib");
  const dl = join(base, "dl");
  const outside = join(base, "outside");
  for (const p of [lib, dl, outside]) await mkdir(p, { recursive: true });
  await writeFile(join(lib, ".uhd-root"), "sentinel\n");
  await writeFile(join(outside, "secret.txt"), "top secret\n");

  const logger = createLogger({ level: "silent" });
  const engine = fakeEngine();
  const config = ConfigSchema.parse({
    DATABASE_URL: DB_URL,
    SECRET_MASTER_KEY: "test-master-key-0123456789",
    DEV_ADMIN_KEY: DEV_KEY,
    DATA_DIR: join(base, "data"),
  });
  const bus = new EventBus(50);
  const events: { type: string; payload: Loose }[] = [];
  const busProxy = {
    publish: async <P>(type: string, payload: P) => {
      events.push({ type, payload });
      return bus.publish(type, payload);
    },
    since: bus.since.bind(bus),
    subscribe: bus.subscribe.bind(bus),
  };
  const capabilities = createCapabilityRegistry({ logger });
  if (opts.catalog)
    capabilities.register(capabilitySpec("catalog.lookup"), async (input) => ({
      local: [],
      remote: String(input.query).includes("Remote")
        ? [{ provider: "tmdb", entityType: "movie" as const, externalId: "999", title: "Remote Film", year: 2019 }]
        : [],
    }));
  const ctx: ServerContext = {
    role: "all",
    db,
    config,
    logger,
    events: busProxy,
    engine,
    outbox: createOutboxPort(),
    secrets: { resolve: async () => "x", create: async () => "x", replace: async () => {}, remove: async () => {} },
    settings: {
      async all() {
        const rows = await db.selectFrom("settings").select(["key", "value"]).execute();
        const stored = new Map(rows.map((r) => [r.key, r.value]));
        const out: Record<string, unknown> = {};
        for (const k of Object.keys(DEFAULT_SETTINGS) as SettingsKey[]) {
          const v = stored.get(k);
          out[k] = GlobalSettings.shape[k].parse({ ...DEFAULT_SETTINGS[k], ...((v as object) ?? {}) });
        }
        return out as GlobalSettings;
      },
      async get(k) {
        return (await this.all())[k];
      },
    },
    capabilities,
  };
  const module = createFileManagerModule({
    ...overrides,
    ao: {
      probe: async () => ({ status: "unknown", reason: "ffprobe_unavailable" }),
      ...overrides.ao,
      options: { backgroundLoops: false, minVideoBytes: 10, verifyHash: "sha256", ...(overrides.ao?.options ?? {}) },
    },
    options: { backgroundLoops: false, dataDir: join(base, "data"), transferWaitMs: 10_000, ...overrides.options },
  });
  if (opts.runTransfers !== false) {
    const enqueue = engine.enqueue.bind(engine);
    engine.enqueue = async (task, input, o) => {
      const r = await enqueue(task, input, o);
      const recordId = (input as { recordId?: string }).recordId;
      if (task === "verifyAndImport" && recordId)
        setTimeout(async () => {
          try {
            const rec = await db
              .selectFrom("organization_record")
              .select("plan")
              .where("id", "=", recordId)
              .executeTakeFirst();
            if ((rec?.plan as unknown as FrozenPlan | undefined)?.transfer)
              await executeRecord(module.deps().ao, recordId);
          } catch {
            // 模拟 worker:失败留给测试断言(记录保持 running)
          }
        }, 0);
      return r;
    };
  }
  const app = createApp({
    config: { NODE_ENV: "development", DEV_ADMIN_KEY: DEV_KEY },
    logger,
    bus,
    role: "all",
    version: "test",
    db,
    modules: [module],
    ctx,
    apiKeyStore: {
      async findByHash(hash) {
        const k = KEYS.find((x) => hashApiKey(x.key) === hash);
        return k ? { id: k.key, name: k.key, scopes: k.scopes, revokedAt: null } : null;
      },
      async touch() {},
      async findSession(hash) {
        if (hash !== hashSessionToken(SESSION_TOKEN)) return null;
        const now = new Date();
        return {
          id: "00000000-0000-4000-8000-00000000a001",
          adminUserId: "00000000-0000-4000-8000-00000000a002",
          username: "admin",
          expiresAt: new Date(now.getTime() + 3600_000),
          revokedAt: null,
          lastSeenAt: now,
        };
      },
    },
  });
  const authHeaders = (key: string): Record<string, string> =>
    key === SESSION
      ? { cookie: `${SESSION_COOKIE}=${SESSION_TOKEN}`, [CSRF_HEADER]: deriveCsrfToken(SESSION_TOKEN) }
      : { authorization: `Bearer ${key}` };

  const root = await db
    .insertInto("library_root")
    .values({ name: "电影库", media_kind: "mixed", local_path: lib, management_mode: "managed" })
    .returning("id")
    .executeTakeFirstOrThrow();
  const inst = await db
    .insertInto("downloader_instance")
    .values({ name: "qb", base_url: "http://127.0.0.1:1", settings: JSON.stringify({ savePath: dl }) })
    .returning("id")
    .executeTakeFirstOrThrow();

  return {
    db,
    engine,
    module,
    events,
    base,
    lib,
    dl,
    outside,
    libRoot: `lib:${root.id}`,
    dlRoot: `dl:${inst.id}:0`,
    libraryRootId: root.id,
    instanceId: inst.id,
    async api(method, path, body, key = DEV_KEY) {
      const res = await app.request(`/api/v1${path}`, {
        method,
        headers: {
          ...authHeaders(key),
          ...(body !== undefined ? { "content-type": "application/json" } : {}),
        },
        ...(body !== undefined ? { body: JSON.stringify(body) } : {}),
      });
      return { status: res.status, body: await res.json() };
    },
    raw(method, path, key = DEV_KEY) {
      return Promise.resolve(app.request(`/api/v1${path}`, { method, headers: authHeaders(key) }));
    },
    async close() {
      await db.destroy();
      const c = new pg.Client({ connectionString: DB_URL });
      await c.connect();
      await c.query(`drop schema if exists ${schema} cascade`);
      await c.end();
      await rm(base, { recursive: true, force: true });
    },
  };
}

export const key = () => randomUUID();

export async function seedEntity(db: Kysely<DB>, e: { kind: "movie" | "series"; title: string; year: number }) {
  const id = (
    await db
      .insertInto("media_entity")
      .values({ kind: e.kind, title: e.title, year: e.year, status: "active" })
      .returning("id")
      .executeTakeFirstOrThrow()
  ).id;
  await db
    .insertInto("title_alias")
    .values({ media_entity_id: id, alias: e.title, normalized: normalizeTitle(e.title), source: "user" })
    .execute();
  return id;
}

/** 已登记资产 + 库内位置(relative_path 相对库根) */
export async function seedAsset(
  db: Kysely<DB>,
  h: Pick<FmHarness, "libraryRootId">,
  a: { entityId: string; rel: string; kind?: "video" | "subtitle"; sidecarOf?: string },
) {
  const asset = await db
    .insertInto("media_asset")
    .values({ media_entity_id: a.entityId, kind: a.kind ?? "video", size_bytes: 10 })
    .returning("id")
    .executeTakeFirstOrThrow();
  const loc = await db
    .insertInto("asset_location")
    .values({
      media_asset_id: asset.id,
      location_kind: "library",
      library_root_id: h.libraryRootId,
      relative_path: a.rel,
      status: "available",
      is_primary: !a.sidecarOf,
      sidecar_of_location_id: a.sidecarOf ?? null,
    })
    .returning("id")
    .executeTakeFirstOrThrow();
  return { assetId: asset.id, locationId: loc.id };
}

export async function seedBinding(
  db: Kysely<DB>,
  h: Pick<FmHarness, "libraryRootId">,
  b: { entityId: string; rel: string },
) {
  return (
    await db
      .insertInto("path_binding")
      .values({
        media_entity_id: b.entityId,
        binding_kind: "entity",
        library_root_id: h.libraryRootId,
        relative_path: b.rel,
      })
      .returning("id")
      .executeTakeFirstOrThrow()
  ).id;
}

/** 纳管下载任务(内容 = 下载根下的 name) */
export async function seedDownload(
  db: Kysely<DB>,
  h: Pick<FmHarness, "instanceId" | "dl">,
  t: { name: string; status: "downloading" | "completed" },
) {
  return (
    await db
      .insertInto("download_task")
      .values({
        instance_id: h.instanceId,
        torrent_hash: randomUUID().replace(/-/g, "").padEnd(40, "0").slice(0, 40),
        name: t.name,
        save_path: h.dl,
        content_path: join(h.dl, t.name),
        managed: true,
        status: t.status,
      })
      .returning("id")
      .executeTakeFirstOrThrow()
  ).id;
}

export { sql };