"""IngestService — the HTTP-free core of the ingest front door.

One call, `ingest_agent_event`, runs the full canonical path:

  token → VerifiedContext (verify_token; unsigned fails closed)
  → [#2 gate] derive authority_decision_verified from a SIGNED AuthorityDecision receipt (never a bare flag)
  → agent-seam bridge_agent_event → canonical VebContentV1 (FULFILLMENT / EXCEPTION / OBSERVATION, fail-closed)
  → decompose the VEB into an IngressEnvelope + transport → pipeline.ingest → append to the journal

Anchoring + the [#5] return path are explicit: `anchor_pending` folds the un-anchored journal tail into one
batch, anchors its root via the injected `AnchorBackend`, and verifies independently; `anchor_proof` returns
the opaque_root + a per-leaf inclusion proof (and, for the self-attested checkpoint, the signed chain entry) so
a relying party re-derives inclusion rather than trusting an echo. Proof lookup is INDEXED and DURABLE: boot
loads only the anchoring watermark (never the batch history); `anchor_proof` resolves the content_hash through
the store's leaf index, fetches THAT batch record, re-derives the Merkle proof from the stored ordered manifest
and verifies it against the stored root before answering — a corrupt or mismatched index fails closed
(PROOF_LOOKUP_CORRUPT), never a false proof.

The bridge→ingest "gap" is bridged here by decomposition: `bridge_agent_event` produces a VEB and `ingest`
re-derives the same VEB from the envelope's content fields (a superset), so `content_hash` is identical and the
Ledger core is untouched (additive-only).
"""

from __future__ import annotations

from dataclasses import dataclass
from types import SimpleNamespace
from typing import Callable, Mapping, Optional

from cl.agent_seam.authority import (
    AuthorityDecisionContentV1,
    AuthorityDecisionReceiptV1,
    verify_authority_receipt,
)
from cl.agent_seam.bridge import ConnectorProfile, bridge_agent_event
from cl.agent_seam.context import TokenVerifier, verify_token
from cl.agent_seam.signer_registry import SignerRegistry
from cl.anchor.backend import AnchorBackend, AnchorError
from cl.anchor.batch import build_batch, verify_anchor
from cl.anchor.merkle import inclusion_proof
from cl.store.anchor_batches import (
    AnchorAttemptLogEntry,
    AnchorBatchRecord,
    InMemoryAnchorBatchStore,
)
from cl.interfaces.identity import IdentityVerifier
from cl.identity.registry import VerifierRegistry
from cl.identity.signatures import Ed25519Verifier
from cl.interfaces.storage import StorageBackend
from cl.ingress.envelope import IngressEnvelope
from cl.model.pacro import PacRoBodyV1, is_pac_ro
from cl.model.veb import VebContentV1
from cl.model.veb import SEAM_GATED
from cl.pipeline import ingest


class IngestServiceError(Exception):
    """A transport-level failure with a stable code (UNAUTHENTICATED / UNKNOWN_CONNECTOR / REJECTED /
    NOT_ANCHORED / ANCHOR_FAILED / PROOF_LOOKUP_CORRUPT)."""

    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code
        self.message = message


@dataclass(frozen=True)
class AuthorityInput:
    """A signed AuthorityDecision presented on the ingest path — the #2 gate's input."""

    content: AuthorityDecisionContentV1
    receipt: AuthorityDecisionReceiptV1
    signer_key_ref: str


@dataclass(frozen=True)
class IngestConfig:
    hash_profile_id: bytes
    profiles: Mapping[str, ConnectorProfile]
    token_verifier: TokenVerifier
    registry: Optional[VerifierRegistry] = None
    authority_verifier: Optional[IdentityVerifier] = None  # #2 gate signer verification (default Ed25519)
    # PR-0: server-side trust anchor for authority receipts. None = legacy caller-supplied key_ref
    # (byte-identical no-op for the merge=deploy rollout); set = signer resolved from the registry,
    # caller's key_ref ignored, and the issued_at/valid_until window enforced via `authority_now`.
    signer_registry: Optional["SignerRegistry"] = None
    authority_now: Optional[Callable[[], str]] = None      # ISO-8601 clock, injected (host wires UTC now)
    # Bounded-batch design v0.1: the hard cap on leaves fetched+materialised+folded per anchor round.
    # An explicit "unbounded" is no longer a legal production state — values < 1 fail fast at
    # construction. The root stays 32B regardless; what this bounds is the pending list, the Merkle
    # build time, and the persisted per-batch leaf manifest.
    max_anchor_leaves: int = 10000


@dataclass(frozen=True)
class IngestReceipt:
    content_hash: str          # hex
    artifact_hash: str         # hex
    event_class: str           # FULFILLMENT | EXCEPTION | OBSERVATION
    is_pac_ro: bool
    decision: str              # routing Decision: VEB | PAC_RO
    assurance: str             # A3 verification level reached
    was_duplicate: bool
    authority_verified: bool


@dataclass(frozen=True)
class AnchorReceipt:
    opaque_root: str           # hex
    anchored_at_height: int
    leaf_count: int
    assurance_level: str
    verified: bool             # independent re-derivation held
    reason: Optional[str] = None


@dataclass(frozen=True)
class InclusionProofView:
    content_hash: str          # hex
    opaque_root: str           # hex
    anchored_at_height: int
    leaf_index: int
    leaf_count: int
    audit_path: tuple          # tuple[ (sibling_hex: str, sibling_is_right: bool) ]
    verifies: bool
    assurance_level: str
    checkpoint_entry: Optional[dict] = None  # signed self-attestation entry, when the backend exposes one


class IngestService:
    def __init__(self, config: IngestConfig, store: StorageBackend,
                 anchor_backend: AnchorBackend,
                 anchor_batch_store=None) -> None:
        self.config = config
        self.store = store
        self.anchor_backend = anchor_backend
        self._authority_verifier = config.authority_verifier or Ed25519Verifier()
        # Phase 4: the anchoring watermark is DERIVED FROM THE DURABLE
        # STORE at boot — a restarted service resumes exactly after the
        # last durably-anchored offset instead of re-anchoring the whole
        # journal (the old in-memory `-1` reset). The in-memory store
        # keeps the demo host behavior identical (-1 on a fresh process).
        # The watermark is the ONLY thing boot reads: proof lookup goes
        # through the store's durable leaf index per call (review MUST 3
        # holds without rehydrating the batch history into memory).
        self.anchor_batches = anchor_batch_store \
            if anchor_batch_store is not None else InMemoryAnchorBatchStore()
        self._last_anchored_offset = self.anchor_batches.watermark()
        self.anchor_failures = 0   # visible drop counter (never silent)
        if config.max_anchor_leaves < 1:
            raise ValueError(
                "max_anchor_leaves must be >= 1 — an unbounded anchor round is not a legal "
                "production state (bounded-batch design v0.1)")

    @property
    def last_anchored_offset(self) -> int:
        """The durable anchoring watermark (read-only; advances only after a durable batch record)."""
        return self._last_anchored_offset

    @property
    def max_anchor_leaves(self) -> int:
        return self.config.max_anchor_leaves

    def anchor_backlog(self) -> int:
        """Un-anchored journal depth right now — the operator-visible back-pressure signal.

        Counted as EXACT rows past the watermark (`count_from`), never as `tip_offset - watermark`:
        journal offsets are not gapless (PostgreSQL identity values are consumed by
        `ON CONFLICT DO NOTHING` and rollbacks), so offset arithmetic can report phantom backlog
        and schedule drain rounds no rows back."""
        return self.store.count_from(self._last_anchored_offset + 1)

    # --- ingest ---------------------------------------------------------------

    def ingest_agent_event(
        self,
        *,
        token: Optional[str],
        event: dict,
        connector_id: str,
        request_id: str,
        received_at: str,
        action_intent_ref: Optional[str] = None,
        authority: Optional[AuthorityInput] = None,
    ) -> IngestReceipt:
        from cl.agent_seam.context import ContextVerificationError

        # Anti-spoof (INGRESS_ASSURANCE_V1 §2.2): the label is assigned by the door, never by the
        # sender — an inbound event that tries to set it is REJECTED, not overwritten.
        if isinstance(event, dict) and "ingress_assurance" in event:
            raise IngestServiceError(
                "INGRESS_ASSURANCE_SPOOF", "ingress_assurance is door-assigned; senders may not set it"
            )
        try:
            ctx = verify_token(token, self.config.token_verifier)
        except ContextVerificationError as e:
            raise IngestServiceError("UNAUTHENTICATED", str(e))

        profile = self.config.profiles.get(connector_id)
        if profile is None:
            raise IngestServiceError("UNKNOWN_CONNECTOR", f"no connector profile for {connector_id!r}")

        # #2 gate — FULFILLMENT authority is DERIVED from a signed receipt, never asserted by the caller.
        authority_verified = False
        if authority is not None:
            authority_verified = verify_authority_receipt(
                authority.receipt,
                authority.content,
                expected_action_intent_hash=action_intent_ref or "",
                verifier=self._authority_verifier,
                signer_key_ref=authority.signer_key_ref,
                signer_registry=self.config.signer_registry,
                now_iso=self.config.authority_now()
                if self.config.authority_now is not None else None,
            )

        veb = bridge_agent_event(
            event,
            profile=profile,
            verified_ctx=ctx,
            hash_profile_id=self.config.hash_profile_id,
            action_intent_ref=action_intent_ref,
            authority_decision_verified=authority_verified,
        )
        env = self._veb_to_envelope(veb, request_id=request_id, received_at=received_at)
        result = ingest(
            env, self.store, self.config.hash_profile_id, self.config.registry,
            ingress_assurance=SEAM_GATED,  # this service sits behind the #2 gate
        )
        if not result.accepted:
            raise IngestServiceError("REJECTED", result.outcome.reason or "routing rejected the event")

        event_class, pac_ro = self._classify(veb)
        return IngestReceipt(
            content_hash=result.outcome.content_hash.hex(),
            artifact_hash=result.stored.artifact_hash.hex(),
            event_class=event_class,
            is_pac_ro=pac_ro,
            decision=result.outcome.decision.value,
            assurance=result.outcome.assurance.name,
            was_duplicate=result.was_duplicate,
            authority_verified=authority_verified,
        )

    def _veb_to_envelope(
        self, veb: VebContentV1, *, request_id: str, received_at: str
    ) -> IngressEnvelope:
        """Decompose the bridge's canonical VEB into a transport envelope. Only content fields cross; transport
        metadata (request id / receive time) is never hashed. `ingest` re-maps these to the SAME VEB."""
        return IngressEnvelope(
            event_type=veb.event_type,
            actor_ref=veb.actor_ref,
            principal_ref=veb.principal_ref,
            hai5_level=veb.hai5_level,
            semantic_time=veb.semantic_time,
            body_binding=veb.body_binding,
            ingress_request_id=request_id,
            ingress_idempotency_key=request_id,
            received_at=received_at,
            schema_version=veb.schema_version,
            evidence_refs=veb.evidence_refs,
            relations=veb.relations,
            credential_ref=veb.credential_ref,
            policy_snapshot_ref=veb.policy_snapshot_ref,
        )

    @staticmethod
    def _classify(veb: VebContentV1) -> tuple:
        if is_pac_ro(veb):
            return PacRoBodyV1.from_body(veb.body_binding.body).event_class, True
        return "OBSERVATION", False

    # --- anchor (#5 return path) ---------------------------------------------

    def anchor_pending(self) -> Optional[AnchorReceipt]:
        """Fold at most `max_anchor_leaves` of the un-anchored journal tail into ONE batch, anchor its
        root, verify independently. Returns None when there is nothing new to anchor. One call = one
        bounded batch = one receipt; a longer tail drains over successive calls (the auto-anchor tick
        runs a bounded number of rounds computed from the backlog it found at tick start)."""
        pending = list(self.store.scan_from(self._last_anchored_offset + 1,
                                            limit=self.config.max_anchor_leaves))
        if not pending:
            return None
        # Review MUST 2: the submission attempt is durable BEFORE the
        # backend runs (write-ahead SUBMITTING row); a failure lands a
        # durable FAILED row carrying the backend's full retry history.
        # A crash or a retry exhaustion can no longer erase the attempt
        # evidence.
        batch = build_batch(pending)
        span = (pending[0].journal_offset, pending[-1].journal_offset)
        self.anchor_batches.record_attempt(AnchorAttemptLogEntry(
            journal_span_start=span[0], journal_span_end=span[1],
            batch_root=batch.root, status="SUBMITTING"))
        try:
            attestation = self.anchor_backend.anchor(batch.root)
        except AnchorError as e:
            failed = {}
            pop_failed = getattr(
                self.anchor_backend, "pop_anchor_details", None)
            if callable(pop_failed):
                failed = pop_failed() or {}
            self.anchor_batches.record_attempt(AnchorAttemptLogEntry(
                journal_span_start=span[0], journal_span_end=span[1],
                batch_root=batch.root, status="FAILED",
                retry_history=tuple(failed.get("retry_history", ())),
                error=str(e)))
            raise IngestServiceError("ANCHOR_FAILED", str(e))
        sample_proof = (batch.inclusion_proof_for(0)
                        if batch.artifact_hashes else None)
        result = SimpleNamespace(
            batch=batch, attestation=attestation,
            verification=verify_anchor(attestation, batch.root,
                                       sample_proof))

        entry_obj = None
        entry_for_root = getattr(self.anchor_backend, "entry_for_root", None)
        if callable(entry_for_root):
            entry = entry_for_root(result.batch.root)
            if entry is not None:
                entry_obj = entry.to_json()

        # Phase 4: durably record the batch (span + membership + ordered
        # leaf manifest + root + attestation + backend submission details)
        # BEFORE advancing any in-memory state — a crash after this line
        # can never cause a re-anchor of this span (at-least-once before,
        # exactly-once after).
        details = {}
        pop_details = getattr(self.anchor_backend, "pop_anchor_details", None)
        if callable(pop_details):
            details = pop_details() or {}
        self.anchor_batches.record(AnchorBatchRecord(
            journal_span_start=span[0],
            journal_span_end=span[1],
            batch_root=result.batch.root,
            content_hashes=tuple(result.batch.content_hashes),
            artifact_hashes=tuple(result.batch.artifact_hashes),
            submission_outcome=str(
                details.get("submission_outcome", "ANCHORED")),
            attempt_status="ANCHORED",
            attestation={
                "opaque_root": result.attestation.opaque_root.hex(),
                "anchored_at_height":
                    result.attestation.anchored_at_height,
                "finality_certificate_hash":
                    result.attestation.finality_certificate_hash.hex(),
                "assurance_level": result.attestation.assurance_level,
                "block_hash": result.attestation.block_hash.hex(),
            },
            gateway_epoch=details.get("gateway_epoch"),
            gateway_sequence=details.get("gateway_sequence"),
            canonical_position=details.get("canonical_position"),
            position_bundle_reference=details.get(
                "position_bundle_reference"),
            retry_history=tuple(details.get("retry_history", ())),
            checkpoint_entry=entry_obj,
        ))

        self._last_anchored_offset = pending[-1].journal_offset
        return AnchorReceipt(
            opaque_root=result.attestation.opaque_root.hex(),
            anchored_at_height=result.attestation.anchored_at_height,
            leaf_count=len(result.batch.artifact_hashes),
            assurance_level=result.attestation.assurance_level,
            verified=result.verification.verified,
            reason=result.verification.reason,
        )

    def anchor_proof(self, content_hash: bytes) -> Optional[InclusionProofView]:
        """The #5 return: opaque_root + a per-leaf inclusion proof for an anchored event (+ the signed
        checkpoint entry, when self-attested). None if the event has not been anchored yet.

        Indexed durable lookup: the store's leaf index resolves content_hash to the earliest legal
        (batch, leaf_index); the batch record is fetched from the durable store and the proof is
        RE-DERIVED from its stored ordered manifest and verified against its stored root before it
        is returned. The index is only a serving projection — any disagreement with the
        authoritative record (dangling batch, wrong leaf, manifest that no longer reproduces the
        root) fails closed with PROOF_LOOKUP_CORRUPT rather than answering with a false proof."""
        ref = self.anchor_batches.lookup_leaf(content_hash)
        if ref is None:
            return None
        rec = self.anchor_batches.record_by_id(ref.batch_id)
        if rec is None:
            raise IngestServiceError(
                "PROOF_LOOKUP_CORRUPT",
                f"leaf index points at batch {ref.batch_id} which has no "
                "durable record")
        if (ref.leaf_index >= len(rec.content_hashes)
                or rec.content_hashes[ref.leaf_index] != content_hash):
            raise IngestServiceError(
                "PROOF_LOOKUP_CORRUPT",
                f"leaf index (batch {ref.batch_id}, leaf {ref.leaf_index}) "
                "disagrees with the batch membership record")
        if len(rec.artifact_hashes) != len(rec.content_hashes):
            raise IngestServiceError(
                "PROOF_LOOKUP_CORRUPT",
                f"batch {ref.batch_id} manifest length disagrees with its "
                "membership record")
        proof = inclusion_proof(list(rec.artifact_hashes), ref.leaf_index)
        if not proof.verifies_to(rec.batch_root):
            raise IngestServiceError(
                "PROOF_LOOKUP_CORRUPT",
                f"re-derived proof for batch {ref.batch_id} does not "
                "verify against the stored batch root")
        return InclusionProofView(
            content_hash=content_hash.hex(),
            opaque_root=rec.batch_root.hex(),
            anchored_at_height=int(
                rec.attestation.get("anchored_at_height", -1)),
            leaf_index=proof.leaf_index,
            leaf_count=proof.leaf_count,
            audit_path=tuple((s.hex(), right)
                             for s, right in proof.audit_path),
            verifies=True,
            assurance_level=str(rec.attestation.get("assurance_level", "")),
            checkpoint_entry=rec.checkpoint_entry,
        )
