// server 测试共用的 PG 装配:测试库地址、可用性探测、schema 咨询锁、每文件独立 schema 的建 / 迁 / 删。
// 各模块 harness 只管自己的上下文与种子数据,建库这几步统一走这里。
import { type DB, migrateToLatest } from "@umt/db";
import { Kysely, PostgresDialect } from "kysely";
import pg from "pg";

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

/** 测试库连得上才跑 PG 用例;UMT_SKIP_DB_TESTS=1 显式跳过(deno task test:all 连不上库时直接失败) */
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;
  }
}

/**
 * Kysely 迁移器会反射库里所有 schema 的表;别的测试文件同时 drop schema 时反射会报"schema 不存在"。
 * 用 PG 咨询锁把"迁移 / 删 schema"串行化(跨进程有效),另加重试兜底。
 */
const SCHEMA_LOCK = 0x756d7431;
export async function withSchemaLock<T>(fn: () => Promise<T>): Promise<T> {
  const c = new pg.Client({ connectionString: PG_URL });
  await c.connect();
  try {
    await c.query("select pg_advisory_lock($1)", [SCHEMA_LOCK]);
    for (let attempt = 1; ; attempt++) {
      try {
        return await fn();
      } catch (err) {
        if (attempt >= 4 || !/does not exist/.test(String((err as Error)?.message))) throw err;
        await new Promise((r) => setTimeout(r, 200 * attempt));
      }
    }
  } finally {
    await c.query("select pg_advisory_unlock($1)", [SCHEMA_LOCK]).catch(() => {});
    await c.end();
  }
}

/** 用一次性连接执行一条管理语句(建 / 删 schema) */
export async function adminQuery(q: string): Promise<void> {
  const c = new pg.Client({ connectionString: PG_URL });
  await c.connect();
  try {
    await c.query(q);
  } finally {
    await c.end();
  }
}

/** 进程 + 随机后缀,避免并行文件撞名 */
export const uniqueSchema = (prefix: string) =>
  `${prefix}_${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`;

export async function dropSchemas(...schemas: string[]): Promise<void> {
  await withSchemaLock(async () => {
    for (const s of schemas) await adminQuery(`drop schema if exists ${s} cascade`);
  });
}

export interface TestSchema {
  schema: string;
  db: Kysely<DB>;
  /** 关连接池并删掉 schema(连同 extra,例如引擎 schema) */
  drop(...extra: string[]): Promise<void>;
}

/** 建独立 schema → 连接池(search_path 指向它)→ 加锁迁移到最新 */
export async function createTestSchema(schema: string, opts: { max?: number } = {}): Promise<TestSchema> {
  await adminQuery(`create schema ${schema}`);
  const db = new Kysely<DB>({
    dialect: new PostgresDialect({
      pool: new pg.Pool({ connectionString: PG_URL, max: opts.max ?? 6, options: `-c search_path=${schema},public` }),
    }),
  });
  await withSchemaLock(() => migrateToLatest(db, { migrationTableSchema: schema }));
  return {
    schema,
    db,
    async drop(...extra) {
      await db.destroy();
      await dropSchemas(schema, ...extra);
    },
  };
}