"""PostgreSQL access for city_briefs_ops.

Mirrors meta-ads-mcp/agent/db.py — psycopg (v3) AsyncConnectionPool with a
checksum-tracked migration runner. Two tables only: `city_briefs` (one row per
show) and `city_briefs_runs` (per-run stage-status rows). This service never
touches any other DB.
"""

from __future__ import annotations

import hashlib
import logging
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any, AsyncIterator

import psycopg
from psycopg import AsyncConnection
from psycopg.types.json import Jsonb
from psycopg_pool import AsyncConnectionPool

from app.config import settings

logger = logging.getLogger(__name__)

MIGRATIONS_DIR = Path(__file__).resolve().parent.parent / "migrations"


def adapt(value: Any) -> Any:
    """Wrap JSON-compatible values so psycopg stores them as jsonb."""
    if isinstance(value, (dict, list)):
        return Jsonb(value)
    return value


def adapt_params(params: tuple[Any, ...] | None) -> tuple[Any, ...] | None:
    if params is None:
        return None
    return tuple(adapt(v) for v in params)


class Database:
    """One pool per database url. Test-friendly via injected url."""

    def __init__(self, url: str | None = None, min_size: int = 1, max_size: int = 6) -> None:
        self._url = url or settings.DATABASE_URL
        self._min_size = min_size
        self._max_size = max_size
        self._pool: AsyncConnectionPool | None = None

    async def open(self) -> None:
        if self._pool is None:
            self._pool = AsyncConnectionPool(
                self._url,
                min_size=self._min_size,
                max_size=self._max_size,
                open=False,
                kwargs={"autocommit": False},
            )
            await self._pool.open(wait=True)

    async def close(self) -> None:
        if self._pool is not None:
            await self._pool.close()
            self._pool = None

    @asynccontextmanager
    async def connection(self) -> AsyncIterator[AsyncConnection]:
        if self._pool is None:
            await self.open()
        assert self._pool is not None
        async with self._pool.connection() as conn:
            yield conn

    async def fetch(self, query: str, params: tuple[Any, ...] | None = None) -> list[dict]:
        async with self.connection() as conn:
            async with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
                await cur.execute(query, adapt_params(params))
                return await cur.fetchall()

    async def fetchone(self, query: str, params: tuple[Any, ...] | None = None) -> dict | None:
        rows = await self.fetch(query, params)
        return rows[0] if rows else None

    async def execute(self, query: str, params: tuple[Any, ...] | None = None) -> None:
        async with self.connection() as conn:
            async with conn.cursor() as cur:
                await cur.execute(query, adapt_params(params))

    async def fetchval(self, query: str, params: tuple[Any, ...] | None = None) -> Any:
        async with self.connection() as conn:
            async with conn.cursor() as cur:
                await cur.execute(query, adapt_params(params))
                row = await cur.fetchone()
                return row[0] if row is not None else None

    @staticmethod
    async def _ensure_migrations_table(conn: AsyncConnection) -> None:
        await conn.execute(
            """
            CREATE TABLE IF NOT EXISTS schema_migrations (
                version    TEXT PRIMARY KEY,
                applied_at TIMESTAMPTZ NOT NULL DEFAULT now(),
                checksum   TEXT NOT NULL DEFAULT ''
            )
            """
        )

    async def migrate(self, migrations_dir: Path | None = None) -> list[str]:
        migrations_dir = migrations_dir or MIGRATIONS_DIR
        applied: list[str] = []
        async with self.connection() as conn:
            await self._ensure_migrations_table(conn)
            async with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
                await cur.execute("SELECT version, checksum FROM schema_migrations")
                seen = {row["version"]: row["checksum"] for row in await cur.fetchall()}

            for path in sorted(migrations_dir.glob("*.sql")):
                version = path.stem
                sql = path.read_text()
                checksum = hashlib.sha256(sql.encode()).hexdigest()
                if version in seen:
                    if seen[version] != checksum:
                        raise RuntimeError(
                            f"Migration {version} is already applied but its "
                            f"checksum does not match the file on disk."
                        )
                    continue
                logger.info("applying migration %s", version)
                async with conn.transaction():
                    await conn.execute(sql)
                    await conn.execute(
                        "INSERT INTO schema_migrations (version, checksum) VALUES (%s, %s)",
                        (version, checksum),
                    )
                applied.append(version)
        return applied


# Shared singleton. Opened by app/main.py lifespan, closed on shutdown.
db = Database()