"""Append-only PostgreSQL journal and its transactional artifact outbox."""

from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterator, Optional, Sequence

from cl.canonical.hash_profile import content_hash as hash_by_profile
from cl.canonical.jcs import canonical_bytes
from cl.interfaces.storage import StorageBackend, StoredEvent
from cl.model import hashing
from cl.model.veb import DOMAIN_VEB_CONTENT


class StorageConflict(RuntimeError):
    pass


def normalize_db_text(value: Any, field: str, *, optional: bool = False) -> Optional[str]:
    """Normalize PostgreSQL TEXT at the repository boundary.

    psycopg returns ``bytes`` for TEXT when the database encoding is SQL_ASCII
    (for example a macOS ``initdb --no-locale`` test cluster). Domain objects
    remain strict UTF-8 ``str`` values regardless of server encoding.
    """
    if value is None and optional:
        return None
    if isinstance(value, str):
        return value
    if isinstance(value, (bytes, bytearray, memoryview)):
        try:
            return bytes(value).decode("utf-8")
        except UnicodeDecodeError as error:
            raise StorageConflict(f"{field} is not valid UTF-8") from error
    raise StorageConflict(f"{field} has unsupported database type {type(value).__name__}")


@dataclass(frozen=True)
class OutboxEntry:
    outbox_id: int
    journal_offset: int
    content_hash: bytes
    artifact_hash: bytes
    attempt_count: int


def migrate(connection: Any) -> None:
    """Apply the schema AND backfill the anchor_leaf_index serving
    projection from the authoritative batch records, in ONE transaction.
    The backfill is add-only and idempotent — on an upgraded deployment
    it is the one-time pass that makes the whole pre-index anchored
    history servable BEFORE the service takes a single lookup; on an
    up-to-date database it adds nothing. The no-history-scan law binds
    every normal service boot (watermark only) — migration time is
    exactly where the one-time pass belongs."""
    from cl.store.anchor_batches import backfill_leaf_index_rows

    schema = Path(__file__).with_name("schema.sql").read_text(encoding="utf-8")
    try:
        with connection.cursor() as cursor:
            cursor.execute(schema)
            backfill_leaf_index_rows(cursor)
        connection.commit()
    except Exception:
        connection.rollback()
        raise


class PostgresStorage(StorageBackend):
    def __init__(self, connection: Any):
        self._connection = connection

    @classmethod
    def connect(cls, dsn: str) -> "PostgresStorage":
        try:
            import psycopg
        except ImportError as error:
            raise RuntimeError("install cotrugli-ledger[postgres]") from error
        return cls(psycopg.connect(dsn))

    @staticmethod
    def _validate(event: StoredEvent) -> None:
        if len(event.content_hash) != 32 or len(event.artifact_hash) != 32:
            raise ValueError("content_hash and artifact_hash must be 32 bytes")
        if event.body_binding_kind not in (
            "PAC_RO", "INLINE_CANONICAL_BODY", "ENCRYPTED_EXTERNAL_BODY",
        ):
            raise ValueError("unknown body_binding_kind")
        if not event.canonical_bytes or not event.proof_set_bytes:
            raise ValueError("canonical and proof-set bytes must not be empty")
        if not event.verification_profile_ref:
            raise ValueError("verification_profile_ref must not be empty")
        if any(len(link) != 32 for link in event.linked_content_hashes):
            raise ValueError("linked content hashes must be 32 bytes")
        try:
            content = json.loads(event.canonical_bytes.decode("utf-8"))
            if not isinstance(content, dict) or canonical_bytes(content) != event.canonical_bytes:
                raise ValueError("canonical_bytes are not canonical JCS content")
            profile_id = bytes.fromhex(content["hash_profile_id"])
            expected_content = hash_by_profile(profile_id, DOMAIN_VEB_CONTENT, content)
        except (KeyError, TypeError, UnicodeDecodeError, ValueError) as error:
            raise ValueError("invalid canonical VEB content") from error
        if expected_content != event.content_hash:
            raise ValueError("content_hash does not match canonical_bytes")
        try:
            proof_content = json.loads(event.proof_set_bytes.decode("utf-8"))
            if (
                not isinstance(proof_content, dict)
                or canonical_bytes(proof_content) != event.proof_set_bytes
                or proof_content.get("hash_profile_id") != profile_id.hex()
                or not isinstance(proof_content.get("proofs"), list)
            ):
                raise ValueError("invalid canonical proof set")
            proof_hash = hashing.proof_set_hash(profile_id, proof_content["proofs"])
            expected_artifact = hashing.artifact_hash(
                profile_id, event.content_hash, proof_hash, event.verification_profile_ref
            )
        except (TypeError, UnicodeDecodeError, ValueError) as error:
            raise ValueError("invalid canonical proof set") from error
        if expected_artifact != event.artifact_hash:
            raise ValueError("artifact_hash does not match content, proofs and verification profile")
        binding = content.get("body_binding")
        expected_kind = binding.get("kind") if isinstance(binding, dict) else None
        if event.body_binding_kind != "PAC_RO" and expected_kind != event.body_binding_kind:
            raise ValueError("body_binding_kind does not match canonical content")

    @staticmethod
    def _links_json(links: tuple[bytes, ...]) -> str:
        return json.dumps([link.hex() for link in links], separators=(",", ":"))

    @staticmethod
    def _event(row: tuple) -> StoredEvent:
        links = row[7]
        if isinstance(links, str):
            links = json.loads(links)
        return StoredEvent(
            journal_offset=int(row[0]), content_hash=bytes(row[1]), artifact_hash=bytes(row[2]),
            body_binding_kind=normalize_db_text(row[3], "body_binding_kind"),
            canonical_bytes=bytes(row[4]), proof_set_bytes=bytes(row[5]),
            verification_profile_ref=normalize_db_text(row[6], "verification_profile_ref"),
            linked_content_hashes=tuple(bytes.fromhex(value) for value in links),
        )

    @staticmethod
    def _same(existing: StoredEvent, incoming: StoredEvent) -> bool:
        return existing == StoredEvent(
            journal_offset=existing.journal_offset,
            content_hash=incoming.content_hash, artifact_hash=incoming.artifact_hash,
            body_binding_kind=incoming.body_binding_kind, canonical_bytes=incoming.canonical_bytes,
            proof_set_bytes=incoming.proof_set_bytes,
            verification_profile_ref=incoming.verification_profile_ref,
            linked_content_hashes=incoming.linked_content_hashes,
        )

    def _append_event(self, cursor: Any, event: StoredEvent) -> int:
        cursor.execute(
            """
            INSERT INTO canonical_event_journal
                (content_hash, artifact_hash, body_binding_kind, canonical_bytes,
                 proof_set_bytes, verification_profile_ref, linked_content_hashes)
            VALUES (%s, %s, %s, %s, %s, %s, %s::jsonb)
            ON CONFLICT (content_hash) DO NOTHING RETURNING journal_offset
            """,
            (event.content_hash, event.artifact_hash, event.body_binding_kind, event.canonical_bytes,
             event.proof_set_bytes, event.verification_profile_ref,
             self._links_json(event.linked_content_hashes)),
        )
        inserted = cursor.fetchone()
        if inserted is not None:
            return int(inserted[0])
        cursor.execute(
            """
            SELECT journal_offset, content_hash, artifact_hash, body_binding_kind,
                   canonical_bytes, proof_set_bytes, verification_profile_ref, linked_content_hashes
            FROM canonical_event_journal WHERE content_hash = %s
            """, (event.content_hash,),
        )
        row = cursor.fetchone()
        if row is None or not self._same(self._event(row), event):
            raise StorageConflict("content_hash already exists with different immutable event data")
        return int(row[0])

    def append(self, event: StoredEvent) -> int:
        self._validate(event)
        try:
            with self._connection.cursor() as cursor:
                offset = self._append_event(cursor, event)
                cursor.execute(
                    """
                    INSERT INTO artifact_outbox (journal_offset, content_hash, artifact_hash)
                    VALUES (%s, %s, %s) ON CONFLICT (journal_offset) DO NOTHING
                    """, (offset, event.content_hash, event.artifact_hash),
                )
            self._connection.commit()
            return offset
        except Exception:
            self._connection.rollback()
            raise

    def pending_outbox(self, limit: int = 100) -> tuple[OutboxEntry, ...]:
        if limit < 1 or limit > 1000:
            raise ValueError("outbox limit must be between 1 and 1000")
        with self._connection.cursor() as cursor:
            cursor.execute(
                """
                SELECT outbox_id, journal_offset, content_hash, artifact_hash, attempt_count
                FROM artifact_outbox WHERE published_at IS NULL ORDER BY outbox_id ASC LIMIT %s
                """, (limit,),
            )
            rows = cursor.fetchall()
        return tuple(OutboxEntry(int(r[0]), int(r[1]), bytes(r[2]), bytes(r[3]), int(r[4])) for r in rows)

    def mark_outbox_published_many(self, outbox_ids: Sequence[int]) -> None:
        ids = tuple(outbox_ids)
        if not ids or len(set(ids)) != len(ids) or any(type(value) is not int or value < 1 for value in ids):
            raise ValueError("outbox_ids must be distinct positive integers")
        try:
            with self._connection.cursor() as cursor:
                cursor.execute(
                    """
                    UPDATE artifact_outbox
                    SET published_at = COALESCE(published_at, transaction_timestamp()),
                        attempt_count = attempt_count + CASE WHEN published_at IS NULL THEN 1 ELSE 0 END
                    WHERE outbox_id = ANY(%s) RETURNING outbox_id
                    """, (list(ids),),
                )
                if {int(row[0]) for row in cursor.fetchall()} != set(ids):
                    raise KeyError("unknown outbox_id in anchored batch")
            self._connection.commit()
        except Exception:
            self._connection.rollback()
            raise

    def mark_outbox_published(self, outbox_id: int) -> None:
        self.mark_outbox_published_many((outbox_id,))

    def get_by_content_hash(self, content_hash: bytes) -> Optional[StoredEvent]:
        with self._connection.cursor() as cursor:
            cursor.execute(
                """
                SELECT journal_offset, content_hash, artifact_hash, body_binding_kind,
                       canonical_bytes, proof_set_bytes, verification_profile_ref, linked_content_hashes
                FROM canonical_event_journal WHERE content_hash = %s
                """, (content_hash,),
            )
            row = cursor.fetchone()
        return None if row is None else self._event(row)

    def scan_from(self, offset: int, limit: Optional[int] = None) -> Iterator[StoredEvent]:
        # The bound is IN THE SQL (bounded-batch design v0.1): without it this method fetchall()s the
        # whole tail into RAM before the caller sees a single event, so a Python-side cut would bound
        # nothing. `limit=None` keeps the historical full-scan behaviour for replay/outbox callers.
        query = """
                SELECT journal_offset, content_hash, artifact_hash, body_binding_kind,
                       canonical_bytes, proof_set_bytes, verification_profile_ref, linked_content_hashes
                FROM canonical_event_journal WHERE journal_offset >= %s ORDER BY journal_offset ASC
                """
        params: tuple = (offset,)
        if limit is not None:
            if limit < 0:
                raise ValueError("scan_from limit must be non-negative")
            query += " LIMIT %s"
            params = (offset, limit)
        with self._connection.cursor() as cursor:
            cursor.execute(query, params)
            rows = cursor.fetchall()
        return iter(self._event(row) for row in rows)

    def tip_offset(self) -> int:
        with self._connection.cursor() as cursor:
            cursor.execute("SELECT COALESCE(MAX(journal_offset), -1) FROM canonical_event_journal")
            return int(cursor.fetchone()[0])

    def count_from(self, offset: int) -> int:
        # Exact row count, in SQL, no materialisation. `journal_offset` is GENERATED AS IDENTITY and
        # NOT gapless — `ON CONFLICT DO NOTHING` and rolled-back inserts consume sequence values —
        # so offset-range arithmetic over MAX(journal_offset) would count phantom rows.
        with self._connection.cursor() as cursor:
            cursor.execute(
                "SELECT COUNT(*) FROM canonical_event_journal WHERE journal_offset >= %s",
                (offset,),
            )
            return int(cursor.fetchone()[0])
