import { migrateToLatest } from "@umt/db";
import { Kysely, PostgresDialect, sql } from "kysely";
import pg from "pg";
import { 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` };
}

/** search_path 带上 public:db 迁移用到的 pg_trgm 操作符类装在 public */
export function connectDb(schema: string): TestDb {
  const pool = new pg.Pool({ connectionString: PG_URL, max: 6, options: `-c search_path=${schema},public` });
  return new Kysely({ dialect: new PostgresDialect({ pool }) });
}

/** 用 @umt/db 的统一迁移在测试 schema 里建全部业务表(含 tasks_* 与 outbox_event) */
export async function migrateBizSchema(db: TestDb, schema: string) {
  await migrateToLatest(db, { migrationTableSchema: schema });
}

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();
  }
}

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;
}

/** 只连接引擎;业务表由调用方先 migrateBizSchema 建好 */
export async function setupEngine(db: TestDb, schemas: { ow: string }, registry: TaskRegistry) {
  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));
  }
}