UHD_Media_Tool/apps/server/test/sources-identity/harness.ts

import { randomUUID } from "node:crypto";
import { createTmdbClient } from "@umt/adapters";
import { DEFAULT_SETTINGS, type GlobalSettings, type SettingsKey } from "@umt/contracts";
import { type DB, type Kysely, migrateToLatest, sql } from "@umt/db";
import type { TaskEngine } from "@umt/tasks";
import { Kysely as K, PostgresDialect } from "kysely";
import pg from "pg";
import { createOutboxPort } from "../../src/context.ts";
import { createApp } from "../../src/http/app.ts";
import { EventBus } from "../../src/http/sse.ts";
import { createSourcesIdentityModule } from "../../src/modules/sources-identity/index.ts";
import type { RuntimeOverrides } from "../../src/modules/sources-identity/runtime.ts";
import type { SecretStore, ServerContext, SettingsReader } from "../../src/modules/types.ts";
import { withSchemaLock } from "../auth-settings/harness.ts";
import { DEV_KEY, memoryLogger, memoryStore } from "../helpers.ts";

export const READ_KEY = "w2b-read-key";
export const EXEC_KEY = "w2b-exec-key";

// W2-B 集成测试夹具:独立 PG schema + 真实迁移 + fake 引擎/凭证 + 模块运行时。

export const PG_URL = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";

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

/** 内存凭证库(W2-A 的 SecretStore 实现前);记录明文以便断言"明文不入业务表" */
export function fakeSecrets(db: () => Kysely<DB>): SecretStore & { plain: Map<string, string> } {
  const plain = new Map<string, string>();
  return {
    plain,
    async resolve(ref) {
      const v = plain.get(ref);
      if (v === undefined) throw new Error(`unknown secret ${ref}`);
      return v;
    },
    async create({ plaintext, purpose, label }) {
      const id = randomUUID();
      plain.set(id, plaintext);
      // 业务表 secret_ref 有外键:落一行占位(密文不是真的);name 与真实实现一致取 label(secret.name 唯一)
      await db()
        .insertInto("secret")
        .values({ id, name: label ?? `test-${id}`, purpose, ciphertext: Buffer.from("x"), nonce: Buffer.from("x") })
        .execute();
      return id;
    },
    async replace(ref, plaintext) {
      plain.set(ref, plaintext);
    },
    async remove(ref) {
      plain.delete(ref);
    },
  };
}

/** 每次读库的设置读取(不缓存,测试里改 settings 表立即生效) */
export function plainSettings(db: () => Kysely<DB>): SettingsReader {
  const all = async (): Promise<GlobalSettings> => {
    const rows = await db().selectFrom("settings").select(["key", "value"]).execute();
    const out = { ...DEFAULT_SETTINGS } as Record<string, unknown>;
    for (const r of rows) {
      const def = (DEFAULT_SETTINGS as Record<string, object>)[r.key];
      if (def) out[r.key] = { ...def, ...(r.value as object) };
    }
    return out as GlobalSettings;
  };
  return { all, get: async <K extends SettingsKey>(k: K) => (await all())[k] } as SettingsReader;
}

export interface Enqueued {
  task: string;
  input: unknown;
  key: string;
}

export function fakeEngine(): TaskEngine & { enqueued: Enqueued[] } {
  const enqueued: Enqueued[] = [];
  const keys = new Set<string>();
  return {
    enqueued,
    async enqueue(task, input, opts) {
      const dedup = keys.has(opts.idempotencyKey);
      if (!dedup) {
        keys.add(opts.idempotencyKey);
        enqueued.push({ task, input, key: opts.idempotencyKey });
      }
      return { runId: randomUUID(), deduplicated: dedup };
    },
    async signal() {
      return { delivered: 0 };
    },
    async cancel() {
      return false;
    },
    async get() {
      return null;
    },
    async list() {
      return [];
    },
    async startWorker() {},
    async stop() {},
  };
}

export interface Harness {
  db: Kysely<DB>;
  ctx: ServerContext;
  engine: ReturnType<typeof fakeEngine>;
  secrets: ReturnType<typeof fakeSecrets>;
  module: ReturnType<typeof createSourcesIdentityModule>;
  rt: ReturnType<ReturnType<typeof createSourcesIdentityModule>["runtime"]>;
  app: ReturnType<typeof createApp>;
  /** 带 dev admin key 的请求 */
  req(path: string, init?: { method?: string; body?: unknown; key?: string }): Promise<Response>;
  outbox(type?: string): Promise<{ type: string; payload: Record<string, unknown>; idempotency_key: string | null }[]>;
  close(): Promise<void>;
}

export interface HarnessOptions extends RuntimeOverrides {
  tmdbUrl?: string;
}

export async function createHarness(name: string, opts: HarnessOptions = {}): Promise<Harness> {
  const schema = `w2b_${name}_${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`;
  const admin = new K<unknown>({
    dialect: new PostgresDialect({ pool: new pg.Pool({ connectionString: PG_URL, max: 1 }) }),
  });
  await sql.raw(`create schema ${schema}`).execute(admin);
  // 与 @umt/db createDb 相同的 int8 解析(createDb 导入即注册全局解析器)
  await import("@umt/db");
  const db = new K<DB>({
    dialect: new PostgresDialect({
      pool: new pg.Pool({ connectionString: PG_URL, max: 8, options: `-c search_path=${schema},public` }),
    }),
  }) as unknown as Kysely<DB>;
  // 与其他测试进程共用 PG:迁移内省会看到别处正在删除的 schema,用同一把咨询锁排队并重试
  await withSchemaLock(() => migrateToLatest(db, { migrationTableSchema: schema }));

  const { logger } = memoryLogger();
  const engine = fakeEngine();
  const secrets = fakeSecrets(() => db);
  const outboxPort = createOutboxPort();
  const ctx = {
    role: "all",
    db,
    config: { NODE_ENV: "development", DEV_ADMIN_KEY: DEV_KEY } as never,
    logger,
    events: new EventBus(5),
    engine,
    outbox: outboxPort,
    secrets,
    settings: plainSettings(() => db),
    capabilities: {} as never,
  } as ServerContext;
  const { tmdbUrl, ...overrides } = opts;
  const module = createSourcesIdentityModule({
    ...overrides,
    tmdb:
      overrides.tmdb ??
      (() => (tmdbUrl ? createTmdbClient({ apiKey: "test-tmdb-key", baseUrl: `${tmdbUrl}/3`, retries: 0 }) : null)),
  });
  const rt = module.runtime(ctx);
  const app = createApp({
    config: { NODE_ENV: "development", DEV_ADMIN_KEY: DEV_KEY },
    logger,
    bus: new EventBus(5),
    role: "all",
    version: "0.0.0-test",
    modules: [module],
    ctx,
    apiKeyStore: memoryStore({
      [READ_KEY]: { id: "reader", scopes: ["read"] },
      [EXEC_KEY]: { id: "operator", scopes: ["organize.execute"] },
    }),
  });
  return {
    db,
    ctx,
    engine,
    secrets,
    module,
    rt,
    app,
    req(path, init = {}) {
      const headers: Record<string, string> = { authorization: `Bearer ${init.key ?? DEV_KEY}` };
      if (init.body !== undefined) headers["content-type"] = "application/json";
      return Promise.resolve(
        app.request(`/api/v1${path}`, {
          method: init.method ?? (init.body !== undefined ? "POST" : "GET"),
          headers,
          ...(init.body !== undefined ? { body: JSON.stringify(init.body) } : {}),
        }),
      );
    },
    async outbox(type) {
      let q = db.selectFrom("outbox_event").select(["type", "payload", "idempotency_key"]).orderBy("at").orderBy("id");
      if (type) q = q.where("type", "=", type);
      return (await q.execute()) as never;
    },
    async close() {
      await db.destroy();
      await withSchemaLock(() => sql.raw(`drop schema if exists ${schema} cascade`).execute(admin));
      await admin.destroy();
    },
  };
}

/** 登记一个指向 mock 的站点 + 可选账号 */
export async function seedSite(
  h: Harness,
  opts: { key?: string; baseUrl: string; cookie?: string; adapter?: "nexusphp" | "mteam_api" | "generic_rss" },
) {
  const site = await h.db
    .insertInto("site")
    .values({
      key: opts.key ?? "audiences",
      name: opts.key ?? "观众",
      adapter: opts.adapter ?? "nexusphp",
      base_url: opts.baseUrl,
      rate_limit: JSON.stringify({ minIntervalMs: 0, concurrency: 5 }),
      snapshot_enabled: true,
      snapshot_interval_sec: 3600,
    })
    .returningAll()
    .executeTakeFirstOrThrow();
  let accountId: string | null = null;
  if (opts.cookie) {
    const ref = await h.secrets.create({ purpose: "site_cookie", plaintext: opts.cookie });
    const acc = await h.db
      .insertInto("site_account")
      .values({ site_id: site.id, label: "main", auth_kind: "cookie", secret_ref: ref })
      .returning(["id"])
      .executeTakeFirstOrThrow();
    accountId = acc.id;
  }
  return { site, accountId };
}

/** 登记 Feed(URL 存 fake secret;feed_source.url_secret_ref 外键指向 secret 表,这里补一行占位) */
export async function seedFeed(
  h: Harness,
  opts: {
    siteId: string | null;
    url: string;
    name?: string;
    accountId?: string | null;
    purpose?: "movie" | "promotion" | "generic";
  },
) {
  const ref = await h.secrets.create({ purpose: "feed_url", plaintext: opts.url });
  return h.db
    .insertInto("feed_source")
    .values({
      site_id: opts.siteId,
      site_account_id: opts.accountId ?? null,
      name: opts.name ?? "feed",
      purpose: opts.purpose ?? "generic",
      url_secret_ref: ref,
      url_display: opts.url.replace(/passkey=[^&]+/, "passkey=[REDACTED]"),
      poll_interval_sec: 300,
    })
    .returningAll()
    .executeTakeFirstOrThrow();
}

// biome-ignore lint/suspicious/noExplicitAny: 测试断言用
type AnyJson = any;
export const json = async (r: Response) => ({
  status: r.status,
  body: (await r.json()) as { ok: boolean; data: AnyJson; error?: AnyJson },
});