"""The durable RECORD BOOK — closed records + witness attestations on disk.

Gap-to-done #2: until now the chain of confirmed range records, hour/day/
quarter/annual tier records and witness signatures lived only in RAM — a
restart lost the book and broke the prev-hash chain. This journal makes
the proof chain durable: append-only, binary framed (length + CRC, the
same discipline as every other NEOva2 store), group-commit fsync, replayed
at boot to restore the exact chain state.

Entries (payload = MAGIC 'NVRB' + version + type + body):
    CONFIRMED   a witnessed ClosedRecord + its BASE and WaW attestations
    TIER        one closing TierRecord (HOUR / DAY / QUARTER / ANNUAL)
    DIVERGENCE  a preserved witness divergence (never deleted)

Signature fields are length-prefixed so a future PQ signature profile
(ML-DSA) fits without a format break.

Boot policy: an INCOMPLETE trailing frame (crash mid-append, never
acknowledged) is truncated with a notice; a CRC-valid-length frame that
fails validation mid-file is real corruption and the book REFUSES to open
(fail closed) — the operator restores from the replica, nothing is
silently skipped.
"""

from __future__ import annotations

import os
import struct
import zlib
from pathlib import Path

from .closing import TierRecord
from .witness import Attestation, ClosedRecord, Divergence

MAGIC = b"NVRB"
VERSION = 1
LEN = struct.Struct(">I")
CRC = struct.Struct(">I")
_HDR = struct.Struct(">4sBB")            # magic, version, entry type
_REC = struct.Struct(">QQ")              # range_start, count
_TIER = struct.Struct(">BQQQQQ")         # tier code, period, first, last,
                                         # object_count, n_children
_U8 = struct.Struct(">B")
_U16 = struct.Struct(">H")

T_CONFIRMED, T_TIER, T_DIVERGENCE = 1, 2, 3
_TIER_CODES = {"HOUR": 0, "DAY": 1, "QUARTER": 2, "ANNUAL": 3}
_TIER_NAMES = {v: k for k, v in _TIER_CODES.items()}
_KIND_CODES = {"BASE": 0, "WAW": 1}
_KIND_NAMES = {v: k for k, v in _KIND_CODES.items()}


class RecordBookCorrupt(Exception):
    """A complete frame failed validation mid-book — fail closed."""


def _pack_att(a: Attestation) -> bytes:
    return (_U8.pack(len(a.witness_id)) + a.witness_id
            + _U8.pack(_KIND_CODES[a.kind])
            + _U16.pack(len(a.signature)) + a.signature)


def _unpack_att(body: bytes, off: int):
    (wl,) = _U8.unpack_from(body, off); off += 1
    wid = body[off:off + wl]; off += wl
    (kc,) = _U8.unpack_from(body, off); off += 1
    (sl,) = _U16.unpack_from(body, off); off += 2
    sig = body[off:off + sl]; off += sl
    return Attestation(wid, _KIND_NAMES[kc], sig), off


class RecordBook:
    def __init__(self, path: str | Path, mirror=None):
        """`mirror`: optional callback(frame_bytes) fired on every append —
        the brain uses it to fan the book to the archive holders (R2: the
        book must live on as many nodes as the archives, progressively;
        the replica is byte-identical and holds no readable business data,
        only opaque commitment-derived records)."""
        self.path = Path(path)
        self.mirror = mirror
        valid_end = self._scan_valid_end()
        if self.path.exists() and self.path.stat().st_size > valid_end:
            print(f"recordbook: truncating {self.path.stat().st_size - valid_end} "
                  f"bytes of incomplete tail (crash mid-append, never acked)",
                  flush=True)
            os.truncate(self.path, valid_end)
        self._f = open(self.path, "ab", buffering=0)
        self._pending = False

    # ---------------------------------------------------------- durability
    def _scan_valid_end(self) -> int:
        """Byte offset of the end of the last COMPLETE valid frame. An
        incomplete tail is truncatable; an invalid complete frame raises."""
        if not self.path.exists():
            return 0
        data = self.path.read_bytes()
        off, n = 0, len(data)
        while off < n:
            if off + LEN.size > n:
                return off                       # incomplete length prefix
            (plen,) = LEN.unpack_from(data, off)
            end = off + LEN.size + plen + CRC.size
            if end > n:
                return off                       # incomplete frame (tail)
            if plen < _HDR.size:
                raise RecordBookCorrupt(f"{self.path}: bad frame at {off}")
            (crc,) = CRC.unpack_from(data, end - CRC.size)
            if zlib.crc32(data[off:end - CRC.size]) & 0xFFFFFFFF != crc:
                raise RecordBookCorrupt(f"{self.path}: CRC fail at {off}")
            magic, ver, etype = _HDR.unpack_from(data, off + LEN.size)
            if magic != MAGIC or ver != VERSION or etype not in (
                    T_CONFIRMED, T_TIER, T_DIVERGENCE):
                raise RecordBookCorrupt(f"{self.path}: bad entry at {off}")
            off = end
        return off

    def _append(self, etype: int, body: bytes, flush: bool):
        payload = _HDR.pack(MAGIC, VERSION, etype) + body
        frame = LEN.pack(len(payload)) + payload
        frame += CRC.pack(zlib.crc32(frame) & 0xFFFFFFFF)
        self._f.write(frame)
        self._pending = True
        if self.mirror is not None:
            self.mirror(frame)                   # fan to the replica nodes
        if flush:
            self.commit()

    def commit(self):
        """Group-commit: one fsync makes every buffered entry durable."""
        if self._pending:
            self._f.flush()
            os.fsync(self._f.fileno())
            self._pending = False

    # ------------------------------------------------------------- writing
    def append_confirmed(self, rec: ClosedRecord, base: Attestation,
                         waw: Attestation, flush: bool = True):
        body = (_REC.pack(rec.range_start, rec.count) + rec.object_root
                + rec.prev_record_hash + _pack_att(base) + _pack_att(waw))
        self._append(T_CONFIRMED, body, flush)

    def append_tier(self, t: TierRecord, flush: bool = True):
        body = (_TIER.pack(_TIER_CODES[t.tier], t.period_id,
                           t.first_position, t.last_position,
                           t.object_count, t.n_children)
                + t.root + t.prev_hash)
        self._append(T_TIER, body, flush)

    def append_divergence(self, d: Divergence, flush: bool = True):
        body = (d.record_hash + d.claimed_root + d.reconstructed_root
                + _U8.pack(len(d.by_witness)) + d.by_witness)
        self._append(T_DIVERGENCE, body, flush)

    # ------------------------------------------------------------- reading
    def replay(self):
        """Yield every entry in append order:
        ("CONFIRMED", ClosedRecord, base, waw) · ("TIER", TierRecord) ·
        ("DIVERGENCE", Divergence). Safe to call on a live book (reads the
        file, not the write handle)."""
        for entry, _end in self.replay_from(self.path, 0):
            yield entry

    @staticmethod
    def parse_entry(data: bytes, off: int = 0):
        """Parse ONE complete entry at `off`. Returns (entry, end_offset) or
        None when the bytes at `off` do not hold a complete frame (an
        unacked tail after a crash) — the caller never sees a partial entry."""
        n = len(data)
        if off + LEN.size > n:
            return None
        (plen,) = LEN.unpack_from(data, off)
        end = off + LEN.size + plen + CRC.size
        if end > n:
            return None
        (crc_expected,) = CRC.unpack_from(data, end - CRC.size)
        if zlib.crc32(data[off:end - CRC.size]) & 0xFFFFFFFF != crc_expected:
            raise RecordBookCorrupt(f"bad CRC in entry at offset {off}")
        body = data[off + LEN.size + _HDR.size:end - CRC.size]
        _, _, etype = _HDR.unpack_from(data, off + LEN.size)
        if etype == T_CONFIRMED:
            start, count = _REC.unpack_from(body, 0)
            p = _REC.size
            root = body[p:p + 32]; p += 32
            prev = body[p:p + 32]; p += 32
            base, p = _unpack_att(body, p)
            waw, p = _unpack_att(body, p)
            return ("CONFIRMED", ClosedRecord(start, count, root, prev),
                    base, waw), end
        if etype == T_TIER:
            tc, period, first, last, cnt, nch = _TIER.unpack_from(body, 0)
            p = _TIER.size
            root = body[p:p + 32]; prev = body[p + 32:p + 64]
            return ("TIER", TierRecord(_TIER_NAMES[tc], period, first,
                                       last, cnt, nch, root, prev)), end
        if etype == T_DIVERGENCE:
            rh, cl, rc = body[0:32], body[32:64], body[64:96]
            (wl,) = _U8.unpack_from(body, 96)
            return ("DIVERGENCE", Divergence(rh, cl, rc,
                                             body[97:97 + wl])), end
        return None

    @staticmethod
    def replay_from(path, offset: int = 0):
        """Yield (entry, end_offset) for every COMPLETE entry from byte
        `offset` on. The book is append-only, so a reader that remembers the
        last end_offset replays only the tail next time — a proof plane must
        never re-read a whole year of closes to learn about one new seal."""
        path = Path(path)
        if not path.exists():
            return
        with open(path, "rb") as f:
            f.seek(offset)
            data = f.read()
        off, n = 0, len(data)
        while off < n:
            got = RecordBook.parse_entry(data, off)
            if got is None:
                break                            # unacked tail (fresh crash)
            entry, end = got
            yield entry, offset + end
            off = end

    @staticmethod
    def frame_bytes_at(path, offset: int) -> bytes:
        """Raw bytes of ONE frame at `offset` (length prefix .. CRC), or b""
        when the file is shorter than that. Used to recognise the same
        book under a new inode: an atomically flipped snapshot copy."""
        try:
            with open(path, "rb") as f:
                f.seek(offset)
                length_raw = f.read(LEN.size)
                if len(length_raw) != LEN.size:
                    return b""
                (plen,) = LEN.unpack(length_raw)
                rest = f.read(plen + CRC.size)
                if len(rest) != plen + CRC.size:
                    return b""
                return length_raw + rest
        except FileNotFoundError:
            return b""

    @staticmethod
    def read_entry_at(path, offset: int):
        """Re-read ONE entry by its byte offset (from replay_from). Lets a
        reader keep offsets in RAM instead of records."""
        with open(path, "rb") as f:
            f.seek(offset)
            length_raw = f.read(LEN.size)
            (plen,) = LEN.unpack(length_raw)
            data = length_raw + f.read(plen + CRC.size)
        got = RecordBook.parse_entry(data, 0)
        if got is None:
            raise RecordBookCorrupt(f"incomplete entry at offset {offset}")
        return got[0]

    def chain_state(self) -> dict:
        """Boot helper: the state the closing pipeline needs to CONTINUE
        the chain — last range-record hash + closed-upto position, the
        confirmed records not yet folded into an hour, the hour/day/quarter
        records not yet folded into their parent tier, and per-tier
        prev-hash / last-period. Closes are strictly ordered, so a parent
        tier record folds everything pending beneath it."""
        prev_range = b"\x00" * 32
        closed_upto = 0
        pending: list = []                       # confirmed, not hour-folded
        hours: list = []                         # hour recs, not day-folded
        days: list = []                          # day recs, not quarter-folded
        quarters: list = []                      # quarter recs, not year-folded
        tiers = {name: {"prev": b"\x00" * 32, "last_period": -1}
                 for name in _TIER_CODES}
        for entry in self.replay():
            if entry[0] == "CONFIRMED":
                rec = entry[1]
                prev_range = rec.record_hash()
                closed_upto = max(closed_upto, rec.range_start + rec.count)
                pending.append((rec, entry[2], entry[3]))
            elif entry[0] == "TIER":
                t = entry[1]
                tiers[t.tier] = {"prev": t.record_hash(),
                                 "last_period": t.period_id}
                if t.tier == "HOUR":
                    hours.append(t)
                    if t.n_children:
                        pending = [p for p in pending
                                   if p[0].range_start > t.last_position]
                elif t.tier == "DAY":
                    days.append(t)
                    hours = []
                elif t.tier == "QUARTER":
                    quarters.append(t)
                    days = []
                elif t.tier == "ANNUAL":
                    quarters = []
        return {"prev_range": prev_range, "closed_upto": closed_upto,
                "pending_confirmed": pending, "tiers": tiers,
                "pending_hours": hours, "pending_days": days,
                "pending_quarters": quarters}

    def close(self):
        self.commit()
        self._f.close()
