import { readdirSync, readFileSync, existsSync } from "node:fs";
import { dirname, join, resolve } from "node:path";
import { fileURLToPath } from "node:url";
import { pool } from "./pool.js";
import { logger } from "../logger.js";

/** Migrations live in infra/migrations (forward-only SQL files, applied in lexical order). */
export function migrationsDir(): string {
  const here = dirname(fileURLToPath(import.meta.url));
  const candidates = [
    resolve(here, "../../../../infra/migrations"), // apps/api/src/db → repo root
    resolve(here, "../../../infra/migrations"), // apps/api/dist → repo root
    resolve(process.cwd(), "infra/migrations"),
    resolve(process.cwd(), "../../infra/migrations"),
  ];
  const found = candidates.find((c) => existsSync(c));
  if (!found) throw new Error(`migrations directory not found (tried ${candidates.join(", ")})`);
  return found;
}

export async function migrate(): Promise<string[]> {
  const dir = migrationsDir();
  const files = readdirSync(dir)
    .filter((f) => f.endsWith(".sql"))
    .sort();
  const client = await pool.connect();
  const applied: string[] = [];
  try {
    await client.query("create table if not exists schema_migrations (version text primary key, applied_at timestamptz not null default now())");
    await client.query("select pg_advisory_lock(7231)");
    const done = new Set((await client.query<{ version: string }>("select version from schema_migrations")).rows.map((r) => r.version));
    for (const f of files) {
      if (done.has(f)) continue;
      const sql = readFileSync(join(dir, f), "utf8");
      logger.info({ migration: f }, "applying migration");
      await client.query("begin");
      try {
        await client.query(sql);
        await client.query("insert into schema_migrations(version) values ($1)", [f]);
        await client.query("commit");
        applied.push(f);
      } catch (err) {
        await client.query("rollback");
        throw new Error(`migration ${f} failed: ${err instanceof Error ? err.message : String(err)}`);
      }
    }
    await client.query("select pg_advisory_unlock(7231)");
  } finally {
    client.release();
  }
  return applied;
}

/** Ensure the daily partition for `day` (and the next one) exists. Idempotent. */
export async function ensureObservationPartition(dayMs: number): Promise<void> {
  const d = new Date(dayMs);
  d.setUTCHours(0, 0, 0, 0);
  for (const offset of [0, 1]) {
    const start = new Date(d.getTime() + offset * 86_400_000);
    const end = new Date(start.getTime() + 86_400_000);
    const name = `observations_${start.toISOString().slice(0, 10).replace(/-/g, "")}`;
    await pool.query(
      `create table if not exists ${name} partition of observations for values from ('${start.toISOString()}') to ('${end.toISOString()}')`,
    );
  }
}