"""Durable anchored-batch records + the boot-time anchoring watermark.

Phase-4 law (NEOva FINAL BUILD PLAN v1.0, Phase 4 — anchor durability):
every anchored batch durably records its journal span, batch membership,
ordered leaf manifest, batch root, gateway sequence + epoch, submission
outcome, canonical position, attestation, position-bundle reference and
attempt/retry history — and the anchored watermark is DERIVED FROM THE
DURABLE STORE at boot. This kills the in-memory `-1` reset that made a
restarted service re-anchor the entire journal.

Store discipline mirrors the rest of `cl.store`: append-only, canonical
ASCII JSON payloads (SQL_ASCII-safe), fail-closed conflicts. The in-memory
implementation exists so the demo host and unit tests share one code path
with the durable deploy — the SERVICE logic never branches on backend.

Indexed durable proof lookup: alongside every batch record the store
writes `content_hash -> (batch_id, leaf_index)` rows, atomically with the
batch (no batch without index, no index without batch). The index is a
REBUILDABLE SERVING PROJECTION — the batch record and its ordered leaf
manifest remain the authority. Two authority-driven maintenance ops, the
SAME law on both backends:

  * `backfill_leaf_index()` — ADD-ONLY: inserts every missing row
    derivable from the batch records, never touches an existing row.
    This is the UPGRADE path and it is wired into `migrate()` (same
    transaction as the schema), so a pre-index deployment comes up with
    its whole anchored history servable — the one-time history pass
    happens at migration time, never on the normal boot path (boot reads
    only the watermark).
  * `rebuild_leaf_index()` — CONTROLLED REPAIR: atomically replaces the
    entire projection with one regenerated from the authority alone
    (removes wrong rows, not just fills gaps). An explicit admin
    operation, never called by the service.

Lookup order (batch_id, then leaf_index) makes the earliest legal anchor
deterministic when the same content_hash appears more than once.
"""

from __future__ import annotations

import json
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple


class AnchorBatchConflict(RuntimeError):
    """A record that would rewrite or overlap already-durable history."""


@dataclass(frozen=True)
class AnchorLeafRef:
    """One serving-projection row: where a content_hash sits in the
    anchored history. `batch_id` is the store's stable batch identity
    (Postgres identity column; 1-based position in memory)."""

    batch_id: int
    leaf_index: int


@dataclass(frozen=True)
class AnchorBatchRecord:
    """One terminally-anchored batch, with every Phase-4 required field."""

    journal_span_start: int              # first journal_offset in the batch
    journal_span_end: int                # last journal_offset in the batch
    batch_root: bytes                    # RFC 6962 batch root (32B)
    content_hashes: Tuple[bytes, ...]    # batch membership, journal order
    artifact_hashes: Tuple[bytes, ...]   # ordered leaf manifest (fold order)
    submission_outcome: str              # ACCEPTED / CANONICAL_EXISTS / CHECKPOINTED / ...
    attempt_status: str                  # terminal status (ANCHORED)
    attestation: dict                    # canonical attestation payload (hex fields)
    gateway_epoch: Optional[int] = None      # era the ACCEPTED envelope signed
    gateway_sequence: Optional[int] = None   # sequence of the accepted envelope
    canonical_position: Optional[int] = None
    position_bundle_reference: Optional[dict] = None  # {"era": E, "segment": S}
    retry_history: Tuple[dict, ...] = field(default_factory=tuple)
    checkpoint_entry: Optional[dict] = None  # signed self-attestation entry

    def __post_init__(self):
        if self.journal_span_end < self.journal_span_start:
            raise ValueError("journal span end precedes start")
        if len(self.batch_root) != 32:
            raise ValueError("batch_root must be 32 bytes")

    def payload(self) -> dict:
        """Canonical, ASCII-safe JSON-able form (bytes as lowercase hex)."""
        return {
            "record_type": "ANCHORED_BATCH_V1",
            "journal_span_start": self.journal_span_start,
            "journal_span_end": self.journal_span_end,
            "batch_root": self.batch_root.hex(),
            "content_hashes": [h.hex() for h in self.content_hashes],
            "artifact_hashes": [h.hex() for h in self.artifact_hashes],
            "submission_outcome": self.submission_outcome,
            "attempt_status": self.attempt_status,
            "attestation": self.attestation,
            "gateway_epoch": self.gateway_epoch,
            "gateway_sequence": self.gateway_sequence,
            "canonical_position": self.canonical_position,
            "position_bundle_reference": self.position_bundle_reference,
            "retry_history": list(self.retry_history),
            "checkpoint_entry": self.checkpoint_entry,
        }

    @classmethod
    def from_payload(cls, value: dict) -> "AnchorBatchRecord":
        if value.get("record_type") != "ANCHORED_BATCH_V1":
            raise ValueError("unknown anchored-batch record type")
        return cls(
            journal_span_start=int(value["journal_span_start"]),
            journal_span_end=int(value["journal_span_end"]),
            batch_root=bytes.fromhex(value["batch_root"]),
            content_hashes=tuple(bytes.fromhex(h)
                                 for h in value["content_hashes"]),
            artifact_hashes=tuple(bytes.fromhex(h)
                                  for h in value["artifact_hashes"]),
            submission_outcome=str(value["submission_outcome"]),
            attempt_status=str(value["attempt_status"]),
            attestation=dict(value["attestation"]),
            gateway_epoch=(None if value.get("gateway_epoch") is None
                           else int(value["gateway_epoch"])),
            gateway_sequence=(None if value.get("gateway_sequence") is None
                              else int(value["gateway_sequence"])),
            canonical_position=(None if value.get("canonical_position") is None
                                else int(value["canonical_position"])),
            position_bundle_reference=value.get("position_bundle_reference"),
            retry_history=tuple(value.get("retry_history", ())),
            checkpoint_entry=value.get("checkpoint_entry"),
        )


@dataclass(frozen=True)
class AnchorAttemptLogEntry:
    """One durable submission attempt (review MUST 2): SUBMITTING is the
    write-ahead row before the backend is called; FAILED carries the full
    retry history and the error. Terminal successes live in
    `anchored_batches`; this log preserves everything that did NOT reach
    the terminal record — a crash or exhaustion can no longer erase the
    attempt evidence."""

    journal_span_start: int
    journal_span_end: int
    batch_root: bytes
    status: str                              # "SUBMITTING" | "FAILED"
    retry_history: Tuple[dict, ...] = field(default_factory=tuple)
    error: Optional[str] = None

    def __post_init__(self):
        if self.status not in ("SUBMITTING", "FAILED"):
            raise ValueError("attempt status must be SUBMITTING or FAILED")
        if len(self.batch_root) != 32:
            raise ValueError("batch_root must be 32 bytes")

    def payload(self) -> dict:
        return {
            "record_type": "ANCHOR_ATTEMPT_V1",
            "journal_span_start": self.journal_span_start,
            "journal_span_end": self.journal_span_end,
            "batch_root": self.batch_root.hex(),
            "status": self.status,
            "retry_history": list(self.retry_history),
            "error": self.error,
        }

    @classmethod
    def from_payload(cls, value: dict) -> "AnchorAttemptLogEntry":
        if value.get("record_type") != "ANCHOR_ATTEMPT_V1":
            raise ValueError("unknown attempt-log record type")
        return cls(
            journal_span_start=int(value["journal_span_start"]),
            journal_span_end=int(value["journal_span_end"]),
            batch_root=bytes.fromhex(value["batch_root"]),
            status=str(value["status"]),
            retry_history=tuple(value.get("retry_history", ())),
            error=value.get("error"),
        )


def _canonical_text(payload: dict) -> str:
    # sort_keys + ensure_ascii: byte-stable and SQL_ASCII-safe
    return json.dumps(payload, sort_keys=True, separators=(",", ":"))


def backfill_leaf_index_rows(cursor: Any) -> int:
    """ADD-ONLY backfill of `anchor_leaf_index` from the authoritative
    batch records, inside the CALLER's transaction (no commit here).
    Existing rows are never touched (ON CONFLICT DO NOTHING). Called by
    `migrate()` so the one-time history pass rides the migration
    transaction — an upgraded deployment answers for its whole anchored
    history before the service takes a single lookup. Returns the number
    of rows added."""
    cursor.execute(
        "SELECT batch_id, payload_text FROM anchored_batches "
        "ORDER BY batch_id")
    rows = cursor.fetchall()
    added = 0
    for batch_id, payload_text in rows:
        rec = AnchorBatchRecord.from_payload(json.loads(payload_text))
        for leaf_index, content_hash in enumerate(rec.content_hashes):
            cursor.execute(
                "INSERT INTO anchor_leaf_index "
                "(content_hash, batch_id, leaf_index) "
                "VALUES (%s, %s, %s) ON CONFLICT DO NOTHING",
                (content_hash, int(batch_id), leaf_index))
            added += int(getattr(cursor, "rowcount", 0) or 0)
    return added


class InMemoryAnchorBatchStore:
    """Demo/unit-test store. Same contract, no durability across processes."""

    def __init__(self) -> None:
        self._records: List[AnchorBatchRecord] = []
        self._attempts: List[AnchorAttemptLogEntry] = []
        # serving projection: content_hash -> [(batch_id, leaf_index), ...]
        self._leaf_index: Dict[bytes, List[AnchorLeafRef]] = {}

    def watermark(self) -> int:
        """Highest durably-anchored journal offset; -1 when none."""
        if not self._records:
            return -1
        return max(r.journal_span_end for r in self._records)

    def record(self, rec: AnchorBatchRecord) -> None:
        if rec.journal_span_start <= self.watermark():
            raise AnchorBatchConflict(
                "anchored-batch span overlaps durable history "
                f"(start {rec.journal_span_start} <= watermark "
                f"{self.watermark()})")
        batch_id = len(self._records) + 1
        # build the projection rows BEFORE mutating anything so a failure
        # here leaves neither batch nor index (mirrors the PG transaction)
        rows = [(h, AnchorLeafRef(batch_id, i))
                for i, h in enumerate(rec.content_hashes)]
        self._records.append(rec)
        for content_hash, ref in rows:
            self._leaf_index.setdefault(content_hash, []).append(ref)

    def by_root(self, batch_root: bytes) -> List[AnchorBatchRecord]:
        return [r for r in self._records if r.batch_root == batch_root]

    def all_records(self) -> List[AnchorBatchRecord]:
        return list(self._records)

    def count(self) -> int:
        return len(self._records)

    def lookup_leaf(self, content_hash: bytes) -> Optional[AnchorLeafRef]:
        """Earliest legal anchor for a content_hash: the row with the
        lowest (batch_id, leaf_index). None when never anchored."""
        refs = self._leaf_index.get(content_hash)
        if not refs:
            return None
        return min(refs, key=lambda r: (r.batch_id, r.leaf_index))

    def record_by_id(self, batch_id: int) -> Optional[AnchorBatchRecord]:
        if 1 <= batch_id <= len(self._records):
            return self._records[batch_id - 1]
        return None

    def _authority_rows(self):
        for pos, rec in enumerate(self._records, start=1):
            for i, h in enumerate(rec.content_hashes):
                yield h, AnchorLeafRef(pos, i)

    def backfill_leaf_index(self) -> int:
        """ADD-ONLY: insert every projection row derivable from the
        authority that is not already present; never touch an existing
        row (same law as the Postgres ON CONFLICT DO NOTHING backfill).
        Returns the number of rows added."""
        added = 0
        for h, ref in self._authority_rows():
            refs = self._leaf_index.setdefault(h, [])
            if ref not in refs:
                refs.append(ref)
                added += 1
        return added

    def rebuild_leaf_index(self) -> int:
        """CONTROLLED REPAIR: atomically replace the whole projection
        with one regenerated from the authority alone — wrong rows are
        removed, not merely supplemented. Returns the row count of the
        rebuilt projection."""
        rebuilt: Dict[bytes, List[AnchorLeafRef]] = {}
        rows = 0
        for h, ref in self._authority_rows():
            rebuilt.setdefault(h, []).append(ref)
            rows += 1
        self._leaf_index = rebuilt
        return rows

    def record_attempt(self, entry: AnchorAttemptLogEntry) -> None:
        self._attempts.append(entry)

    def attempt_log(self) -> List[AnchorAttemptLogEntry]:
        return list(self._attempts)


class PostgresAnchorBatchStore:
    """Durable store over the `anchored_batches` table (append-only,
    mutation rejected by trigger — see schema.sql)."""

    def __init__(self, connection: Any) -> None:
        self._connection = connection

    def watermark(self) -> int:
        with self._connection.cursor() as cursor:
            cursor.execute(
                "SELECT COALESCE(MAX(journal_span_end), -1) "
                "FROM anchored_batches")
            row = cursor.fetchone()
        return int(row[0])

    def record(self, rec: AnchorBatchRecord) -> None:
        if rec.journal_span_start <= self.watermark():
            raise AnchorBatchConflict(
                "anchored-batch span overlaps durable history "
                f"(start {rec.journal_span_start})")
        payload_text = _canonical_text(rec.payload())
        try:
            with self._connection.cursor() as cursor:
                cursor.execute(
                    "INSERT INTO anchored_batches "
                    "(journal_span_start, journal_span_end, batch_root, "
                    " payload_text) VALUES (%s, %s, %s, %s) "
                    "RETURNING batch_id",
                    (rec.journal_span_start, rec.journal_span_end,
                     rec.batch_root, payload_text))
                batch_id = int(cursor.fetchone()[0])
                # serving-projection rows ride the SAME transaction as the
                # batch: one commit covers both, so a failure anywhere in
                # between leaves neither (no batch without index, no index
                # without batch).
                for leaf_index, content_hash in enumerate(rec.content_hashes):
                    cursor.execute(
                        "INSERT INTO anchor_leaf_index "
                        "(content_hash, batch_id, leaf_index) "
                        "VALUES (%s, %s, %s)",
                        (content_hash, batch_id, leaf_index))
            self._connection.commit()
        except Exception as error:  # noqa: BLE001
            self._connection.rollback()
            # the DATABASE is the overlap authority (gist exclusion) — the
            # watermark pre-check above is only a fast-path message; a
            # racing instance loses here, atomically.
            if type(error).__name__ in ("ExclusionViolation",
                                        "UniqueViolation",
                                        "IntegrityError"):
                raise AnchorBatchConflict(
                    "anchored-batch span overlaps durable history "
                    "(database exclusion constraint)") from error
            raise

    def by_root(self, batch_root: bytes) -> List[AnchorBatchRecord]:
        with self._connection.cursor() as cursor:
            cursor.execute(
                "SELECT payload_text FROM anchored_batches "
                "WHERE batch_root = %s ORDER BY batch_id",
                (batch_root,))
            rows = cursor.fetchall()
        return [AnchorBatchRecord.from_payload(json.loads(r[0]))
                for r in rows]

    def all_records(self) -> List[AnchorBatchRecord]:
        with self._connection.cursor() as cursor:
            cursor.execute(
                "SELECT payload_text FROM anchored_batches ORDER BY batch_id")
            rows = cursor.fetchall()
        return [AnchorBatchRecord.from_payload(json.loads(r[0]))
                for r in rows]

    def count(self) -> int:
        with self._connection.cursor() as cursor:
            cursor.execute("SELECT COUNT(*) FROM anchored_batches")
            row = cursor.fetchone()
        return int(row[0])

    def lookup_leaf(self, content_hash: bytes) -> Optional[AnchorLeafRef]:
        """Earliest legal anchor for a content_hash: the row with the
        lowest (batch_id, leaf_index). None when never anchored."""
        with self._connection.cursor() as cursor:
            cursor.execute(
                "SELECT batch_id, leaf_index FROM anchor_leaf_index "
                "WHERE content_hash = %s "
                "ORDER BY batch_id, leaf_index LIMIT 1",
                (content_hash,))
            row = cursor.fetchone()
        if row is None:
            return None
        return AnchorLeafRef(batch_id=int(row[0]), leaf_index=int(row[1]))

    def record_by_id(self, batch_id: int) -> Optional[AnchorBatchRecord]:
        with self._connection.cursor() as cursor:
            cursor.execute(
                "SELECT payload_text FROM anchored_batches "
                "WHERE batch_id = %s",
                (batch_id,))
            row = cursor.fetchone()
        if row is None:
            return None
        return AnchorBatchRecord.from_payload(json.loads(row[0]))

    def backfill_leaf_index(self) -> int:
        """ADD-ONLY: complete the projection with rows derivable from the
        authority, never touching existing rows. Idempotent. The normal
        upgrade path runs this INSIDE `migrate()` (same transaction as
        the schema) — this standalone form exists for parity with the
        in-memory backend and for operational re-runs. Returns the number
        of rows added."""
        try:
            with self._connection.cursor() as cursor:
                added = backfill_leaf_index_rows(cursor)
            self._connection.commit()
        except Exception:
            self._connection.rollback()
            raise
        return added

    def rebuild_leaf_index(self) -> int:
        """CONTROLLED REPAIR: atomically replace the ENTIRE projection
        with one regenerated from the authority alone — wrong rows are
        removed, not merely supplemented (same law as the in-memory
        rebuild). One transaction: TRUNCATE + re-derive; a failure rolls
        the old projection back intact. TRUNCATE is the explicit,
        visible admin path past the append-only row trigger — the
        trigger exists to stop SILENT row edits of the projection, not a
        wholesale, transactional regeneration from the authority (the
        index is never the authority). The service NEVER calls this.
        Returns the row count of the rebuilt projection."""
        try:
            with self._connection.cursor() as cursor:
                cursor.execute("TRUNCATE anchor_leaf_index")
                rows = backfill_leaf_index_rows(cursor)
            self._connection.commit()
        except Exception:
            self._connection.rollback()
            raise
        return rows

    def record_attempt(self, entry: AnchorAttemptLogEntry) -> None:
        payload_text = _canonical_text(entry.payload())
        with self._connection.cursor() as cursor:
            cursor.execute(
                "INSERT INTO anchor_attempt_log "
                "(journal_span_start, journal_span_end, batch_root, "
                " status, payload_text) VALUES (%s, %s, %s, %s, %s)",
                (entry.journal_span_start, entry.journal_span_end,
                 entry.batch_root, entry.status, payload_text))
        self._connection.commit()

    def attempt_log(self) -> List[AnchorAttemptLogEntry]:
        with self._connection.cursor() as cursor:
            cursor.execute(
                "SELECT payload_text FROM anchor_attempt_log "
                "ORDER BY attempt_id")
            rows = cursor.fetchall()
        return [AnchorAttemptLogEntry.from_payload(json.loads(r[0]))
                for r in rows]
