"""Database migrations — the SQL files in db/migrations, applied in name order
and recorded in schema_migrations, so a database knows what it has.

    python -m app.db_migrate status     what's applied, pending, or changed since
    python -m app.db_migrate up         apply everything pending, in order
    python -m app.db_migrate check      exit 1 when anything is pending (a deploy gate)
    python -m app.db_migrate baseline   record every file as applied without running it —
                                        for a database built from db/schema.sql, which
                                        already holds all of them
    python -m app.db_migrate deploy     what a container runs before the API starts: an
                                        empty database gets db/schema.sql and a baseline,
                                        any other one `up` — one deploy at a time
    python -m app.db_migrate rerun      apply every file again, whatever's recorded — CI's
                                        proof that each is safe to re-run over schema.sql

db/schema.sql stays the source of truth for a new database (it's what
app/core/db_provisioning.py applies, followed by a baseline); a migration
file brings an existing one forward. Each file manages its own transaction
(BEGIN … COMMIT) and is written to be safe to re-run. A file that builds
indexes CONCURRENTLY can't run in a transaction or as one multi-statement
query, so it runs statement by statement. The app warns at startup while
anything is pending.
"""

import asyncio
import hashlib
import re
import sys
from dataclasses import dataclass
from pathlib import Path

MIGRATIONS_DIR = Path(__file__).resolve().parents[2] / "db" / "migrations"
SCHEMA_SQL = MIGRATIONS_DIR.parent / "schema.sql"
# Held while deploying, so two instances starting together don't both migrate.
_LOCK_KEY = 0x464C4F57  # "FLOW"

TRACKING_DDL = """
CREATE TABLE IF NOT EXISTS schema_migrations (
  name        text PRIMARY KEY,
  checksum    text NOT NULL,
  applied_at  timestamptz NOT NULL DEFAULT now(),
  how         text NOT NULL DEFAULT 'applied'  -- applied | baseline
)
"""


@dataclass
class Migration:
    name: str
    path: Path
    sql: str

    @property
    def checksum(self) -> str:
        return hashlib.sha256(self.sql.replace("\r\n", "\n").encode()).hexdigest()[:16]

    @property
    def statement_by_statement(self) -> bool:
        return bool(re.search(r"\bCONCURRENTLY\b", self.sql, re.IGNORECASE)) and not re.search(r"^\s*BEGIN\s*;", self.sql, re.IGNORECASE | re.MULTILINE)


def migrations() -> list[Migration]:
    return [Migration(p.name, p, p.read_text(encoding="utf-8")) for p in sorted(MIGRATIONS_DIR.glob("*.sql"))]


def split_statements(sql: str) -> list[str]:
    """Top-level statements of a script: semicolons outside quotes, dollar
    quotes and comments."""
    out, buf, i, n = [], [], 0, len(sql)
    while i < n:
        ch = sql[i]
        if sql.startswith("--", i):
            end = sql.find("\n", i)
            i = n if end == -1 else end
            continue
        if sql.startswith("/*", i):
            end = sql.find("*/", i + 2)
            i = n if end == -1 else end + 2
            continue
        if ch == "'":
            # A quoted string, '' being a quote inside it.
            end = i + 1
            while end < n:
                if sql[end] == "'":
                    if end + 1 < n and sql[end + 1] == "'":
                        end += 2
                        continue
                    break
                end += 1
            buf.append(sql[i : end + 1])
            i = end + 1
            continue
        dollar = re.match(r"\$[A-Za-z_]*\$", sql[i:])
        if dollar:
            tag = dollar.group(0)
            end = sql.find(tag, i + len(tag))
            end = n if end == -1 else end + len(tag)
            buf.append(sql[i:end])
            i = end
            continue
        if ch == ";":
            stmt = "".join(buf).strip()
            if stmt:
                out.append(stmt)
            buf = []
        else:
            buf.append(ch)
        i += 1
    tail = "".join(buf).strip()
    if tail:
        out.append(tail)
    return out


async def _connect():
    import asyncpg
    from sqlalchemy.engine import make_url

    from app.core.config import settings

    url = make_url(settings.database_url)
    return await asyncpg.connect(host=url.host, port=url.port, user=url.username, password=url.password, database=url.database)


async def ensure_tracking(conn) -> None:
    await conn.execute(TRACKING_DDL)


async def applied(conn) -> dict[str, str]:
    await ensure_tracking(conn)
    return {r["name"]: r["checksum"] for r in await conn.fetch("SELECT name, checksum FROM schema_migrations")}


async def pending(conn) -> list[Migration]:
    done = await applied(conn)
    return [m for m in migrations() if m.name not in done]


# They build record table indexes CONCURRENTLY, which Postgres refuses on a
# partitioned table. 2026-11-07-partition-record-tables.sql builds the same
# indexes, so on a partitioned table there's nothing left for these to do. A
# database brought forward with `up` still runs them, before that file.
_BUILT_BY_PARTITIONING = {"2026-10-06-record-query-indexes.sql", "2026-11-05-submissions-tenant-time.sql"}


async def _superseded(conn, m: Migration) -> bool:
    if m.name not in _BUILT_BY_PARTITIONING:
        return False
    return await conn.fetchval("SELECT relkind FROM pg_class WHERE oid = 'master_records'::regclass") == b"p"


async def apply(conn, m: Migration) -> None:
    if await _superseded(conn, m):
        pass
    elif m.statement_by_statement:
        for stmt in split_statements(m.sql):
            await conn.execute(stmt)
    else:
        await conn.execute(m.sql)
    await conn.execute(
        "INSERT INTO schema_migrations (name, checksum) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET checksum = EXCLUDED.checksum, applied_at = now(), how = 'applied'",
        m.name,
        m.checksum,
    )


async def baseline(conn) -> list[str]:
    """Every migration file recorded as applied — for a database built from db/schema.sql."""
    await ensure_tracking(conn)
    names = []
    for m in migrations():
        result = await conn.execute(
            "INSERT INTO schema_migrations (name, checksum, how) VALUES ($1, $2, 'baseline') ON CONFLICT (name) DO NOTHING", m.name, m.checksum
        )
        if result.endswith(" 1"):
            names.append(m.name)
    return names


async def deploy(conn) -> list[str]:
    """An empty database is built from db/schema.sql (and baselined); any
    other is brought forward. Returns what was done, one line each."""
    await conn.execute("SELECT pg_advisory_lock($1)", _LOCK_KEY)
    try:
        if await conn.fetchval("SELECT to_regclass('tenants')") is None:
            await conn.execute(SCHEMA_SQL.read_text(encoding="utf-8"))
            names = await baseline(conn)
            return [f"Built the database from schema.sql ({len(names)} migration(s) recorded as applied)."]
        done = []
        for m in await pending(conn):
            await apply(conn, m)
            done.append(f"Applied {m.name}")
        return done or ["Database is up to date."]
    finally:
        await conn.execute("SELECT pg_advisory_unlock($1)", _LOCK_KEY)


async def main(argv: list[str]) -> int:
    command = argv[1] if len(argv) > 1 else "status"
    conn = await _connect()
    try:
        if command == "deploy":
            for line in await deploy(conn):
                print(line, flush=True)
            return 0
        if command == "baseline":
            names = await baseline(conn)
            print(f"Recorded {len(names)} migration(s) as already applied." if names else "Nothing to record.")
            return 0
        done = await applied(conn)
        todo = [m for m in migrations() if m.name not in done]
        if command == "status":
            for m in migrations():
                state = "pending" if m.name not in done else "CHANGED since applied" if done[m.name] != m.checksum else "applied"
                print(f"  {state:22} {m.name}")
            print(f"{len(todo)} pending.")
            return 0
        if command == "check":
            if todo:
                print("Pending migrations: " + ", ".join(m.name for m in todo))
                return 1
            print("Database is up to date.")
            return 0
        if command == "rerun":
            for m in migrations():
                print(f"Re-running {m.name}…", flush=True)
                await apply(conn, m)
            return 0
        if command == "up":
            for m in todo:
                print(f"Applying {m.name}…", flush=True)
                await apply(conn, m)
            print(f"Applied {len(todo)} migration(s)." if todo else "Nothing to apply.")
            return 0
        print(__doc__)
        return 2
    finally:
        await conn.close()


async def startup_warning(conn) -> str | None:
    """What's wrong, if the database is behind the code — read only (a
    database that has never been tracked isn't changed here)."""
    if await conn.fetchval("SELECT to_regclass('schema_migrations')") is None:
        return "Database migrations aren't tracked yet — run: python -m app.db_migrate status, then baseline (if it's up to date) or up"
    done = {r["name"] for r in await conn.fetch("SELECT name FROM schema_migrations")}
    todo = [m.name for m in migrations() if m.name not in done]
    if todo:
        return f"Database migrations pending: {', '.join(todo)} — run: python -m app.db_migrate up"
    return None


async def warn_if_pending() -> None:
    """At startup: say so when the database is behind the code."""
    import logging

    log = logging.getLogger("flowfabric")
    try:
        conn = await _connect()
    except Exception:  # noqa: BLE001 - the app reports a down database elsewhere
        return
    try:
        message = await startup_warning(conn)
        if message:
            log.warning(message)
    finally:
        await conn.close()


if __name__ == "__main__":
    sys.exit(asyncio.run(main(sys.argv)))