UHD_Media_Tool/packages/tasks/test/helpers.ts

import { Kysely, PostgresDialect, sql } from "kysely";
import pg from "pg";
import { migrateTasksTables, OpenWorkflowEngine, silentSink, type TaskRegistry } from "../src/index.ts";

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

/** 探测开发库;不可达时集成测试跳过(CI 机器应提供 55433) */
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 {
    console.warn(`[tasks] PostgreSQL ${PG_URL} 不可达,跳过集成测试`);
    return false;
  }
}

// biome-ignore lint/suspicious/noExplicitAny: 测试直接用原生 SQL
export type TestDb = Kysely<any>;

/**
 * 每个测试文件用独立 schema:业务表 <name>,引擎表 <name>_ow。
 * 名字带 pid + 随机后缀:同一个 PG 上并发跑的多份测试(多个终端/CI 任务/多个 agent)互不踩踏;
 * 调用方在 afterAll 里 dropSchemas 清理。
 */
export function testSchemas(suffix: string) {
  const uniq = `${process.pid.toString(36)}${Math.random().toString(36).slice(2, 8)}`;
  const biz = `tasks_test_${suffix}_${uniq}`;
  return { biz, ow: `${biz}_ow` };
}

export function connectDb(schema: string): TestDb {
  const pool = new pg.Pool({ connectionString: PG_URL, max: 6, options: `-c search_path=${schema}` });
  return new Kysely({ dialect: new PostgresDialect({ pool }) });
}

export async function resetSchemas(schemas: { biz: string; ow: string }) {
  const admin = connectDb("public");
  try {
    await sql.raw(`DROP SCHEMA IF EXISTS ${schemas.biz} CASCADE`).execute(admin);
    await sql.raw(`DROP SCHEMA IF EXISTS ${schemas.ow} CASCADE`).execute(admin);
    await sql.raw(`CREATE SCHEMA ${schemas.biz}`).execute(admin);
  } finally {
    await admin.destroy();
  }
}

export async function dropSchemas(schemas: { biz: string; ow: string }) {
  const admin = connectDb("public");
  try {
    await sql.raw(`DROP SCHEMA IF EXISTS ${schemas.biz} CASCADE`).execute(admin);
    await sql.raw(`DROP SCHEMA IF EXISTS ${schemas.ow} CASCADE`).execute(admin);
  } finally {
    await admin.destroy();
  }
}

/** 与 packages/db 0001_init + 0002_foundation 的 outbox_event 同构 */
export async function createOutboxTable(db: TestDb) {
  await sql`CREATE TABLE outbox_event (
    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
    at timestamptz NOT NULL DEFAULT now(),
    type text NOT NULL,
    payload jsonb NOT NULL DEFAULT '{}'::jsonb,
    published_at timestamptz,
    idempotency_key text UNIQUE,
    status text NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'dispatched', 'failed', 'dead')),
    attempts integer NOT NULL DEFAULT 0,
    next_attempt_at timestamptz NOT NULL DEFAULT now(),
    last_error jsonb)`.execute(db);
}

export async function createSideEffectTable(db: TestDb) {
  await sql`CREATE TABLE IF NOT EXISTS side_effect (
    id bigserial PRIMARY KEY, run_id text NOT NULL, step text NOT NULL, note text, pid int,
    at timestamptz NOT NULL DEFAULT now())`.execute(db);
}

export async function sideEffects(db: TestDb, runId: string) {
  const r = await sql<{ step: string; note: string | null; pid: number }>`
    SELECT step, note, pid FROM side_effect WHERE run_id = ${runId} ORDER BY id`.execute(db);
  return r.rows;
}

export async function setupEngine(db: TestDb, schemas: { ow: string }, registry: TaskRegistry) {
  await migrateTasksTables(db);
  return OpenWorkflowEngine.connect({
    databaseUrl: PG_URL,
    db,
    registry,
    schema: schemas.ow,
    logSink: silentSink,
    cancelPollMs: 200,
  });
}

export async function waitFor<T>(
  fn: () => Promise<T | null | undefined | false>,
  timeoutMs: number,
  label = "condition",
) {
  const deadline = Date.now() + timeoutMs;
  for (;;) {
    const v = await fn();
    if (v) return v;
    if (Date.now() > deadline) throw new Error(`timeout waiting for ${label}`);
    await new Promise((r) => setTimeout(r, 200));
  }
}