"""PostgreSQL access (psycopg 3) + a forward-only migration runner.

Standard wiggleverse stack ported to Postgres (SD-0001 D-7): a connection pool and
numbered forward-only `.sql` migrations applied in filename order, recorded in
`schema_migrations`. No ORM. The runner is fail-stop (a failing migration aborts and
records nothing, so it retries next boot) and guarded by a session advisory lock so
two booting processes never double-apply (INV-7).
"""
from __future__ import annotations

from pathlib import Path

import psycopg
from psycopg_pool import ConnectionPool

from . import config

# app/platform/db.py -> parents[2] is backend/, where migrations/ lives.
MIGRATIONS_DIR = Path(__file__).resolve().parents[2] / "migrations"

# Arbitrary fixed key for the migration advisory lock (INV-7 concurrency guard).
_MIGRATION_LOCK_KEY = 0x6563_6F6D_6D31  # "ecomm1"


def open_pool(dsn: str | None = None, *, min_size: int = 1, max_size: int = 5) -> ConnectionPool:
    """Open a small connection pool against the configured database."""
    pool = ConnectionPool(dsn or config.database_url(), min_size=min_size, max_size=max_size, open=False)
    pool.open()
    pool.wait()
    return pool


def _migration_files(migrations_dir: Path) -> list[Path]:
    return sorted(migrations_dir.glob("*.sql"))


def migrate(conn: psycopg.Connection, migrations_dir: Path = MIGRATIONS_DIR) -> list[str]:
    """Apply unapplied migrations in filename order; return the names applied.

    Forward-only and fail-stop: each migration runs with its `schema_migrations`
    insert in one transaction, so a failure records nothing and is retried next time.
    A session advisory lock serializes concurrent migrators (INV-7).
    """
    conn.execute("SELECT pg_advisory_lock(%s)", (_MIGRATION_LOCK_KEY,))
    try:
        conn.execute(
            "CREATE TABLE IF NOT EXISTS schema_migrations ("
            " filename TEXT PRIMARY KEY,"
            " applied_at TIMESTAMPTZ NOT NULL DEFAULT now())"
        )
        conn.commit()
        applied = {
            r[0] for r in conn.execute("SELECT filename FROM schema_migrations").fetchall()
        }
        ran: list[str] = []
        for path in _migration_files(migrations_dir):
            if path.name in applied:
                continue
            try:
                conn.execute(path.read_text())  # whole file; no params -> multi-statement
                conn.execute(
                    "INSERT INTO schema_migrations (filename) VALUES (%s)", (path.name,)
                )
                conn.commit()
            except Exception:
                conn.rollback()
                raise
            ran.append(path.name)
        return ran
    finally:
        conn.execute("SELECT pg_advisory_unlock(%s)", (_MIGRATION_LOCK_KEY,))
        conn.commit()


def pending_migrations(conn: psycopg.Connection, migrations_dir: Path = MIGRATIONS_DIR) -> list[str]:
    """Migration filenames present on disk but not yet recorded as applied.

    Used by /healthz to report migration currency (§6.4). Returns all on-disk
    migrations if the tracking table does not exist yet.
    """
    exists = conn.execute("SELECT to_regclass('public.schema_migrations')").fetchone()[0]
    if exists is None:
        return [p.name for p in _migration_files(migrations_dir)]
    applied = {
        r[0] for r in conn.execute("SELECT filename FROM schema_migrations").fetchall()
    }
    return [p.name for p in _migration_files(migrations_dir) if p.name not in applied]