"""FrameSink — the binary sending path, at the outbox seam the design left open.

The outbox promised that the destination is a one-line swap. This is the swap: a sink
that implements the same `Sink` protocol the JSON sinks do, so an edge moves to the
binary wire by constructing a different sink and nothing else in the edge moves. What
crosses is only what the lock names — the four NEOf1 objects, packed into frames — and
what `send()` accepts is only what every sink accepts: a snapshot.

The session is received, never negotiated. Which AKE ceremony yields it is the gate's
decision (open pin P3); the lock pins the FIELDS a session must hand over, and
`FrameSession` is exactly those fields plus the device's signer. Establishing it happens
once per edge, elsewhere.

Discipline, all of it from the lock and the founding docs:

  * **A frame is DONE on its `AdmissionReceipt`, never on transport success.** `send()`
    returning True means the seal entered the sending path; the truth about delivery
    lives in `pending` and `admitted`, nowhere else.
  * **Frames chain**: `(epoch, frame_seq, previous_frame_hash)`, and a re-offered frame
    is byte-identical — idempotent on the far side by construction.
  * **No retries, no backoff, no clocks.** The core models the ideal path; a failed
    offer is counted and the frame stays pending. WHEN to offer it again is the organ
    around the core — call `resend_pending()`, on whatever schedule the host owns.
  * **MAC on every frame; an Ed25519 checkpoint signature periodically** — publicly
    verifiable evidence, a few per lane, never per object.
"""

from __future__ import annotations

import hashlib
from dataclasses import dataclass, field
from typing import Callable

from .snapshot import Snapshot
from .wire import (
    FRAME_HEADER_SIZE,
    OBJECTS_PER_FRAME,
    FrameHeader,
    WireError,
    build_frame,
    commitment,
    frame_hash,
)


@dataclass(frozen=True)
class FrameSession:
    """What a session MUST yield, per the lock §3 — the fields, not the ceremony."""

    edge_id: int
    session_id: int
    epoch: int
    policy_profile_id: int
    key: bytes
    signer: Callable[[bytes], bytes] | None = None


@dataclass(frozen=True)
class AdmissionReceipt:
    """The gate's word that one frame is durable: `(frame_seq, admission_seq,
    payload_hash)`, verbatim from FLOW REVIEW 4."""

    frame_seq: int
    admission_seq: int
    payload_hash: bytes


# A transport takes one frame's bytes and returns the gate's receipt, or None when there
# is nothing to say (down, refused, dropped). It is expected not to raise; if it does,
# the failure is counted and the frame stays pending — sending never blocks the business.
Transport = Callable[[bytes], AdmissionReceipt | None]


@dataclass
class FrameSink:
    """The Sink that speaks NEOf1. `send()` is the protocol method the edge calls;
    `emit()` is the object-level entry the attest emitters use."""

    session: FrameSession
    transport: Transport
    objects_per_frame: int = OBJECTS_PER_FRAME
    checkpoint_every: int = 16   # frames 1, 1+N, 1+2N… carry the Ed25519 signature [CHOICE]

    failures: int = 0
    pending: dict[int, bytes] = field(default_factory=dict)     # frame_seq → frame bytes
    admitted: list[AdmissionReceipt] = field(default_factory=list)

    _queue: list[bytes] = field(default_factory=list)
    _object_seq: int = 0          # last global object sequence number assigned
    _frame_seq: int = 0           # last frame sequence number emitted
    _prev_frame_hash: bytes = bytes(32)

    # ── the Sink protocol ────────────────────────────────────────────────────────────

    def send(self, snapshot: Snapshot) -> bool:
        """Queue the period's seal as its 32-byte `snapshot_hash` — the commitment, and
        nothing else (v0.2). True means it entered the sending path; DONE is only ever
        the receipt. Never raises into the caller."""
        try:
            self.emit(bytes.fromhex(snapshot.snapshot_hash()))
        except (WireError, ValueError):
            self.failures += 1     # counted, never propagated — the book is already durable
            return False
        return True

    # ── the object path ──────────────────────────────────────────────────────────────

    def emit(self, packed: bytes) -> None:
        """Queue one 32-byte commitment; a full frame flushes itself. Fail-closed at the
        seam: anything that is not exactly 32 bytes never enters a frame."""
        self._queue.append(commitment(packed))
        if len(self._queue) >= self.objects_per_frame:
            self.flush()

    def flush(self) -> int | None:
        """Pack everything queued (up to one frame's worth) and offer it. Returns the
        frame_seq, or None when there was nothing to send — silence is free."""
        if not self._queue:
            return None
        objects = self._queue[: self.objects_per_frame]
        del self._queue[: len(objects)]

        first = self._object_seq + 1
        self._object_seq += len(objects)
        self._frame_seq += 1
        signed = self.session.signer is not None and (
            (self._frame_seq - 1) % self.checkpoint_every == 0
        )
        payload = b"".join(objects)
        header = FrameHeader(
            edge_id=self.session.edge_id,
            session_id=self.session.session_id,
            epoch=self.session.epoch,
            frame_seq=self._frame_seq,
            first_object_seq=first,
            last_object_seq=self._object_seq,
            object_count=len(objects),
            payload_hash=hashlib.sha256(payload).digest(),
            previous_frame_hash=self._prev_frame_hash,
            policy_profile_id=self.session.policy_profile_id,
            signed=signed,
        )
        frame = build_frame(
            header, payload, key=self.session.key,
            signer=self.session.signer if signed else None,
        )
        self._prev_frame_hash = frame_hash(frame)
        self.pending[self._frame_seq] = frame
        self._offer(frame)
        return self._frame_seq

    def close(self) -> None:
        """Flush the partial frame. There is no goodbye on the wire — the chain simply
        stops, and the next session's epoch starts a new one."""
        while self._queue:
            self.flush()

    # ── receipts ─────────────────────────────────────────────────────────────────────

    def admit(self, receipt: AdmissionReceipt) -> bool:
        """Mark one frame DONE — the only way a frame ever leaves `pending`. The receipt
        must name the pending frame's own payload hash; anything else is counted and
        ignored, because a receipt is a claim and claims are verified."""
        frame = self.pending.get(receipt.frame_seq)
        if frame is None:
            return False
        header = FrameHeader.parse(frame[:FRAME_HEADER_SIZE])
        if receipt.payload_hash != header.payload_hash:
            self.failures += 1
            return False
        del self.pending[receipt.frame_seq]
        self.admitted.append(receipt)
        return True

    def resend_pending(self) -> int:
        """Offer every pending frame once, byte-identical, oldest first. Returns how many
        are STILL pending — the caller owns the schedule, this owns the bytes."""
        for frame_seq in sorted(self.pending):
            self._offer(self.pending[frame_seq])
        return len(self.pending)

    def _offer(self, frame: bytes) -> None:
        try:
            receipt = self.transport(frame)
        except Exception:            # a transport that raises is a transport that is down
            self.failures += 1
            return
        if receipt is None:
            self.failures += 1
            return
        self.admit(receipt)


def ed25519_signer(private_seed: bytes) -> Callable[[bytes], bytes]:
    """A signer from a 32 B seed — the device identity's key, made callable.

    Imports `cryptography` lazily (the `sign` extra): the core stays dependency-free,
    and an edge that never checkpoints never pays for it.
    """
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

    return Ed25519PrivateKey.from_private_bytes(private_seed).sign
