//! Virtual linear / FAST channel assembly — `wai.video.channel`.
//!
//! A linear channel is a continuous schedule of assets over wall-clock time; the
//! delivery layer turns the slice of that schedule covering a playback window
//! into a manifest. The capstone of the OTT ingest: it composes the other four
//! and turns the one thing a managed streaming operation is really selling —
//! *operational accountability* — into a cryptographic artifact.
//!
//! Two facts a broadcaster today reconstructs from logs after the fact are here
//! signed and recomputable:
//!
//! - **The schedule is a real continuous channel.** [`ChannelReceipt::verify`]
//!   rejects a schedule with a gap or an overlap — a linear channel airs with no
//!   dead air and no double-booking, and the receipt will not sign one that
//!   does.
//! - **The program-as-run is checkable.** `aired` — which entries were on air in
//!   the attested window — is recomputed from the schedule and the window, so a
//!   licensor, an advertiser or a regulator confirms what actually aired against
//!   the signed record rather than a play-out log only the operator holds.
//!
//! Windows chain via `parent_receipt_hash`, so a channel's history is a hash
//! chain of program-as-run windows — the StreamOps audit trail. Each window's
//! manifest is still a `wai.video.manifest`, so the per-class dedup carries
//! through: one generation per window per class, not per viewer. Only the
//! measurement — `joules_micro`, its acquisition class and the class's evidence —
//! is measured-attested: signed, and not re-measurable. Reuses the manifest floor.

use ed25519_dalek::{Signer, SigningKey};

use crate::quantum_ops::GrantRef;
use crate::energy_class::{figure_and_class_cohere, write_optional};
use crate::wai_video_manifest::EnergyProvenance;
use crate::world::receipt::merkle_root;

const DOMAIN_RECEIPT: &[u8] = b"wai:video-channel-receipt\x01";
const DOMAIN_RECEIPT_ID: &[u8] = b"wai:video-channel-receipt-id\x01";
const DOMAIN_ENTRY: &[u8] = b"wai:video-channel-entry\x01";

/// What a scheduled slot carries.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum EntryKind {
    Program,
    Ad,
    Filler,
}

impl EntryKind {
    fn tag(self) -> u8 {
        match self {
            EntryKind::Program => 1,
            EntryKind::Ad => 2,
            EntryKind::Filler => 3,
        }
    }
}

/// One slot in the channel schedule. `start_ms` is an epoch in milliseconds so
/// it shares units with `duration_ms`; the slot occupies `[start_ms, end)`.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ScheduleEntry {
    pub asset_id: String,
    pub start_ms: u64,
    pub duration_ms: u64,
    pub kind: EntryKind,
}

impl ScheduleEntry {
    pub fn end_ms(&self) -> u64 {
        self.start_ms + self.duration_ms
    }
    fn leaf(&self) -> [u8; 32] {
        let mut h = blake3::Hasher::new();
        h.update(DOMAIN_ENTRY);
        h.update(&(self.asset_id.len() as u32).to_be_bytes());
        h.update(self.asset_id.as_bytes());
        h.update(&self.start_ms.to_be_bytes());
        h.update(&self.duration_ms.to_be_bytes());
        h.update(&[self.kind.tag()]);
        *h.finalize().as_bytes()
    }
}

/// Whether a schedule is a valid continuous channel: ordered, gapless and
/// non-overlapping — each entry begins exactly where the previous one ended.
pub fn is_continuous(schedule: &[ScheduleEntry]) -> bool {
    schedule.windows(2).all(|w| w[0].end_ms() == w[1].start_ms)
}

/// The schedule entry on air at `now_ms`, and the offset into it. Deterministic;
/// `None` before the schedule starts or after it ends.
pub fn entry_at(schedule: &[ScheduleEntry], now_ms: u64) -> Option<(usize, u64)> {
    schedule
        .iter()
        .position(|e| now_ms >= e.start_ms && now_ms < e.end_ms())
        .map(|i| (i, now_ms - schedule[i].start_ms))
}

/// The indices on air at any point in the half-open window `[start, end)` —
/// the program-as-run for that window. An entry aired if it overlaps the window.
fn aired_in(schedule: &[ScheduleEntry], window_start_ms: u64, window_end_ms: u64) -> Vec<u64> {
    schedule
        .iter()
        .enumerate()
        .filter(|(_, e)| e.start_ms < window_end_ms && e.end_ms() > window_start_ms)
        .map(|(i, _)| i as u64)
        .collect()
}

/// A signed program-as-run window of a linear channel.
#[derive(Clone, Debug, PartialEq)]
pub struct ChannelReceipt {
    pub channel_id: String,
    /// Merkle commitment over the schedule entries — the program identity.
    pub schedule_root: [u8; 32],
    pub schedule: Vec<ScheduleEntry>,
    pub window_start_ms: u64,
    pub window_end_ms: u64,
    /// Schedule indices on air in the window — recomputed in verify().
    pub aired: Vec<u64>,
    pub work_schedule_ops: u64,
    pub joules_micro: u64,
    pub energy_provenance: Option<EnergyProvenance>,
    pub grant: GrantRef,
    /// The prior window's receipt — the channel history is this hash chain.
    pub parent_receipt_hash: Option<[u8; 32]>,
    pub signer_pubkey: [u8; 32],
    pub signer_id: String,
    pub sig: [u8; 64],
}

impl ChannelReceipt {
    fn signing_payload(&self) -> Vec<u8> {
        let mut o = DOMAIN_RECEIPT.to_vec();
        o.extend_from_slice(&(self.channel_id.len() as u32).to_be_bytes());
        o.extend_from_slice(self.channel_id.as_bytes());
        o.extend_from_slice(&self.schedule_root);
        o.extend_from_slice(&self.window_start_ms.to_be_bytes());
        o.extend_from_slice(&self.window_end_ms.to_be_bytes());
        o.extend_from_slice(&self.work_schedule_ops.to_be_bytes());
        o.extend_from_slice(&self.joules_micro.to_be_bytes());
        write_optional(self.energy_provenance.as_ref(), &mut o);
        o.extend_from_slice(&self.grant.grant_hash);
        o.extend_from_slice(&(self.grant.capability.len() as u32).to_be_bytes());
        o.extend_from_slice(self.grant.capability.as_bytes());
        o.extend_from_slice(&self.grant.joule_ceiling_micro.to_be_bytes());
        match self.grant.funds_ceiling {
            Some(f) => {
                o.push(1);
                o.extend_from_slice(&f.to_be_bytes());
            }
            None => o.push(0),
        }
        match self.parent_receipt_hash {
            Some(h) => {
                o.push(1);
                o.extend_from_slice(&h);
            }
            None => o.push(0),
        }
        o.extend_from_slice(&(self.aired.len() as u64).to_be_bytes());
        for a in &self.aired {
            o.extend_from_slice(&a.to_be_bytes());
        }
        o.extend_from_slice(&(self.schedule.len() as u64).to_be_bytes());
        for e in &self.schedule {
            o.extend_from_slice(&e.leaf());
        }
        o.extend_from_slice(&self.signer_pubkey);
        let id = self.signer_id.as_bytes();
        o.extend_from_slice(&(id.len() as u32).to_be_bytes());
        o.extend_from_slice(id);
        o
    }

    /// Seal a program-as-run window. `aired` is computed here from the schedule
    /// and the window, so it cannot claim a program that did not air.
    #[allow(clippy::too_many_arguments)]
    pub fn seal(
        signer: &SigningKey,
        signer_id: impl Into<String>,
        channel_id: impl Into<String>,
        schedule: Vec<ScheduleEntry>,
        window_start_ms: u64,
        window_end_ms: u64,
        joules_micro: u64,
        energy_provenance: Option<EnergyProvenance>,
        grant: GrantRef,
        parent_receipt_hash: Option<[u8; 32]>,
    ) -> ChannelReceipt {
        let leaves: Vec<[u8; 32]> = schedule.iter().map(ScheduleEntry::leaf).collect();
        let aired = aired_in(&schedule, window_start_ms, window_end_ms);
        let mut r = ChannelReceipt {
            channel_id: channel_id.into(),
            schedule_root: merkle_root(&leaves),
            work_schedule_ops: schedule.len() as u64,
            schedule,
            window_start_ms,
            window_end_ms,
            aired,
            joules_micro,
            energy_provenance,
            grant,
            parent_receipt_hash,
            signer_pubkey: signer.verifying_key().to_bytes(),
            signer_id: signer_id.into(),
            sig: [0u8; 64],
        };
        r.sig = signer.sign(&r.signing_payload()).to_bytes();
        r
    }

    /// Signature + program integrity: the schedule must be a valid continuous
    /// channel (ordered, gapless, non-overlapping), the Merkle root and work
    /// figure must follow from it, the window must be non-empty, `aired` must be
    /// exactly what the schedule and window produce, and the energy figure and
    /// its label stand or fall together.
    pub fn verify(&self) -> bool {
        let leaves: Vec<[u8; 32]> = self.schedule.iter().map(ScheduleEntry::leaf).collect();
        if merkle_root(&leaves) != self.schedule_root {
            return false;
        }
        if self.work_schedule_ops != self.schedule.len() as u64 {
            return false;
        }
        if self.window_start_ms >= self.window_end_ms {
            return false;
        }
        if !is_continuous(&self.schedule) {
            return false;
        }
        if aired_in(&self.schedule, self.window_start_ms, self.window_end_ms) != self.aired {
            return false;
        }
        // The figure and its class stand or fall together, and a hardware
        // class must carry its declared uncertainty (energy_class).
        if !figure_and_class_cohere(self.joules_micro, self.energy_provenance.as_ref()) {
            return false;
        }
        crate::sig::verify_strict(&self.signer_pubkey, &self.signing_payload(), &self.sig).is_ok()
    }

    /// The program-as-run: the schedule entries on air in this window.
    pub fn aired_entries(&self) -> Vec<&ScheduleEntry> {
        self.aired.iter().filter_map(|&i| self.schedule.get(i as usize)).collect()
    }

    /// Microjoules of assembly energy per aired entry in the window, with its
    /// acquisition class. `None` when unmetered or nothing aired.
    pub fn uj_per_aired(&self) -> Option<(f64, EnergyProvenance)> {
        let n = self.aired.len() as u64;
        match (&self.energy_provenance, n) {
            (Some(p), a) if self.joules_micro > 0 && a > 0 => {
                Some((self.joules_micro as f64 / a as f64, p.clone()))
            }
            _ => None,
        }
    }

    pub fn receipt_hash(&self) -> [u8; 32] {
        let mut h = blake3::Hasher::new();
        h.update(DOMAIN_RECEIPT_ID);
        h.update(&self.signing_payload());
        h.update(&self.sig);
        *h.finalize().as_bytes()
    }
}

// ---------------------------------------------------------------------------------------
// The JSON carriage (extensions/revenue-share.md §4): the receipt's fields written out for a
// verifier that is not this crate. The signed bytes are those of `seal` and `signing_payload`,
// unchanged; the carriage only writes them out and reads them back.
// ---------------------------------------------------------------------------------------

/// The members every carriage has (revenue-share §4.1); a class with evidence adds its own.
const CARRIAGE_MEMBERS: [&str; 14] = [
    "channel_id",
    "schedule_root",
    "schedule",
    "window_start_ms",
    "window_end_ms",
    "aired",
    "work_schedule_ops",
    "joules_micro",
    "energy_provenance",
    "grant",
    "parent_receipt_hash",
    "signer_pubkey",
    "signer_id",
    "sig",
];

impl EntryKind {
    /// Its name in the carriage.
    pub fn name(self) -> &'static str {
        match self {
            EntryKind::Program => "program",
            EntryKind::Ad => "ad",
            EntryKind::Filler => "filler",
        }
    }

    pub fn from_name(s: &str) -> Option<EntryKind> {
        match s {
            "program" => Some(EntryKind::Program),
            "ad" => Some(EntryKind::Ad),
            "filler" => Some(EntryKind::Filler),
            _ => None,
        }
    }
}

fn hex(b: &[u8]) -> String {
    b.iter().map(|x| format!("{x:02x}")).collect()
}

/// Lowercase hexadecimal of exactly `N` bytes.
fn unhex<const N: usize>(v: &serde_json::Value) -> Option<[u8; N]> {
    let s = v.as_str()?;
    if s.len() != 2 * N || !s.bytes().all(|c| c.is_ascii_digit() || (b'a'..=b'f').contains(&c)) {
        return None;
    }
    let mut out = [0u8; N];
    for (i, o) in out.iter_mut().enumerate() {
        *o = u8::from_str_radix(&s[2 * i..2 * i + 2], 16).ok()?;
    }
    Some(out)
}

impl ChannelReceipt {
    /// The signing payload, as [`ChannelReceipt::verify`] checks it and revenue-share §4.2
    /// writes it out.
    pub fn payload(&self) -> Vec<u8> {
        self.signing_payload()
    }

    /// The JSON carriage (revenue-share §4.1): every field, in that order. It is the one JSON
    /// form of a channel receipt: a record or tool that carries a receipt as JSON uses this one,
    /// and [`ChannelReceipt::from_json`] reads it back.
    pub fn to_json(&self) -> serde_json::Value {
        use crate::energy_class::EnergyProvenance as P;
        use serde_json::{Map, Value, json};
        let mut m = Map::new();
        m.insert("channel_id".into(), json!(self.channel_id));
        m.insert("schedule_root".into(), json!(hex(&self.schedule_root)));
        let entries: Vec<Value> = self
            .schedule
            .iter()
            .map(|e| json!({"asset_id": e.asset_id, "start_ms": e.start_ms, "duration_ms": e.duration_ms, "kind": e.kind.name()}))
            .collect();
        m.insert("schedule".into(), Value::Array(entries));
        m.insert("window_start_ms".into(), json!(self.window_start_ms));
        m.insert("window_end_ms".into(), json!(self.window_end_ms));
        m.insert("aired".into(), json!(self.aired));
        m.insert("work_schedule_ops".into(), json!(self.work_schedule_ops));
        m.insert("joules_micro".into(), json!(self.joules_micro));
        m.insert("energy_provenance".into(), self.energy_provenance.as_ref().map_or(Value::Null, |p| json!(p.label())));
        match &self.energy_provenance {
            Some(P::OnChipCounter { uncertainty, filtering }) => {
                m.insert("energy_uncertainty".into(), json!({"relative_ppm": uncertainty.relative_ppm, "window_us": uncertainty.window_us}));
                m.insert("energy_counter_filtering".into(), json!(filtering.label()));
            }
            Some(P::CalibratedInstrument { uncertainty, calibration_ref }) => {
                m.insert("energy_uncertainty".into(), json!({"relative_ppm": uncertainty.relative_ppm, "window_us": uncertainty.window_us}));
                m.insert("energy_calibration_ref".into(), json!(calibration_ref));
            }
            _ => {}
        }
        m.insert(
            "grant".into(),
            json!({
                "grant_hash": hex(&self.grant.grant_hash),
                "capability": self.grant.capability,
                "joule_ceiling_micro": self.grant.joule_ceiling_micro,
                "funds_ceiling": self.grant.funds_ceiling,
            }),
        );
        m.insert("parent_receipt_hash".into(), self.parent_receipt_hash.map_or(Value::Null, |h| json!(hex(&h))));
        m.insert("signer_pubkey".into(), json!(hex(&self.signer_pubkey)));
        m.insert("signer_id".into(), json!(self.signer_id));
        m.insert("sig".into(), json!(hex(&self.sig)));
        Value::Object(m)
    }

    /// Read the JSON carriage (revenue-share §4.1): exactly its members, each of its type, and
    /// each entry's end within a u64. A receipt of the legacy class label does not read in it.
    /// Whether it verifies is [`ChannelReceipt::verify`]'s.
    pub fn from_json(v: &serde_json::Value) -> Option<ChannelReceipt> {
        use crate::energy_class::{CounterFiltering, EnergyProvenance as P, Uncertainty};
        use serde_json::Value;
        let o = v.as_object()?;
        let label = match o.get("energy_provenance")? {
            Value::Null => None,
            Value::String(s) => Some(s.as_str()),
            _ => return None,
        };
        let mut want: Vec<&str> = CARRIAGE_MEMBERS.to_vec();
        match label {
            Some("OnChipCounter") => want.extend(["energy_uncertainty", "energy_counter_filtering"]),
            Some("CalibratedInstrument") => want.extend(["energy_uncertainty", "energy_calibration_ref"]),
            _ => {}
        }
        if o.len() != want.len() || !want.iter().all(|k| o.contains_key(*k)) {
            return None;
        }
        let uncertainty = || -> Option<Uncertainty> {
            let u = o.get("energy_uncertainty")?.as_object()?;
            if u.len() != 2 {
                return None;
            }
            Some(Uncertainty { relative_ppm: u32::try_from(u.get("relative_ppm")?.as_u64()?).ok()?, window_us: u.get("window_us")?.as_u64()? })
        };
        let energy_provenance = match label {
            None => None,
            Some("ModelBased") => Some(P::ModelBased),
            Some("Estimator") => Some(P::Estimator),
            Some("OnChipCounter") => Some(P::OnChipCounter {
                uncertainty: uncertainty()?,
                filtering: CounterFiltering::from_label(o.get("energy_counter_filtering")?.as_str()?)?,
            }),
            Some("CalibratedInstrument") => Some(P::CalibratedInstrument {
                uncertainty: uncertainty()?,
                calibration_ref: o.get("energy_calibration_ref")?.as_str()?.to_string(),
            }),
            Some(_) => return None,
        };
        let g = o.get("grant")?.as_object()?;
        if g.len() != 4 {
            return None;
        }
        let grant = GrantRef {
            grant_hash: unhex(g.get("grant_hash")?)?,
            capability: g.get("capability")?.as_str()?.to_string(),
            joule_ceiling_micro: g.get("joule_ceiling_micro")?.as_u64()?,
            funds_ceiling: match g.get("funds_ceiling")? {
                Value::Null => None,
                f => Some(f.as_u64()?),
            },
        };
        let mut schedule = Vec::new();
        for e in o.get("schedule")?.as_array()? {
            let e = e.as_object()?;
            if e.len() != 4 {
                return None;
            }
            let entry = ScheduleEntry {
                asset_id: e.get("asset_id")?.as_str()?.to_string(),
                start_ms: e.get("start_ms")?.as_u64()?,
                duration_ms: e.get("duration_ms")?.as_u64()?,
                kind: EntryKind::from_name(e.get("kind")?.as_str()?)?,
            };
            entry.start_ms.checked_add(entry.duration_ms)?;
            schedule.push(entry);
        }
        let aired = o.get("aired")?.as_array()?.iter().map(Value::as_u64).collect::<Option<Vec<u64>>>()?;
        let u = |k: &str| o.get(k)?.as_u64();
        Some(ChannelReceipt {
            channel_id: o.get("channel_id")?.as_str()?.to_string(),
            schedule_root: unhex(o.get("schedule_root")?)?,
            schedule,
            window_start_ms: u("window_start_ms")?,
            window_end_ms: u("window_end_ms")?,
            aired,
            work_schedule_ops: u("work_schedule_ops")?,
            joules_micro: u("joules_micro")?,
            energy_provenance,
            grant,
            parent_receipt_hash: match o.get("parent_receipt_hash")? {
                Value::Null => None,
                h => Some(unhex(h)?),
            },
            signer_pubkey: unhex(o.get("signer_pubkey")?)?,
            signer_id: o.get("signer_id")?.as_str()?.to_string(),
            sig: unhex(o.get("sig")?)?,
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn key(s: u8) -> SigningKey {
        SigningKey::from_bytes(&[s; 32])
    }

    // A gapless channel: show A [0,10s), ad [10s,12s), show B [12s,22s).
    fn schedule() -> Vec<ScheduleEntry> {
        vec![
            ScheduleEntry { asset_id: "show-A".into(), start_ms: 0, duration_ms: 10_000, kind: EntryKind::Program },
            ScheduleEntry { asset_id: "ad-1".into(), start_ms: 10_000, duration_ms: 2_000, kind: EntryKind::Ad },
            ScheduleEntry { asset_id: "show-B".into(), start_ms: 12_000, duration_ms: 10_000, kind: EntryKind::Program },
        ]
    }

    #[test]
    fn entry_at_and_continuity() {
        let s = schedule();
        assert!(is_continuous(&s));
        assert_eq!(entry_at(&s, 5_000), Some((0, 5_000)));
        assert_eq!(entry_at(&s, 11_000), Some((1, 1_000)));
        assert_eq!(entry_at(&s, 100_000), None);
    }

    #[test]
    fn program_as_run_is_recomputable() {
        // Window [8s, 13s) spans the tail of A, the whole ad, the head of B.
        let g = GrantRef::unbounded("video.channel");
        let r = ChannelReceipt::seal(&key(1), "did:key:chan", "fast-42", schedule(), 8_000, 13_000, 6_000, Some(crate::energy_class::test_on_chip()), g, None);
        assert!(r.verify());
        assert_eq!(r.aired, vec![0, 1, 2], "all three entries touch the window");
        assert_eq!(r.aired_entries().iter().map(|e| e.asset_id.as_str()).collect::<Vec<_>>(), ["show-A", "ad-1", "show-B"]);
    }

    #[test]
    fn a_gap_in_the_schedule_will_not_verify() {
        let mut s = schedule();
        s[2].start_ms = 13_000; // leave 1s of dead air after the ad
        let g = GrantRef::unbounded("video.channel");
        let r = ChannelReceipt::seal(&key(2), "o", "c", s, 0, 22_000, 0, None, g, None);
        assert!(!r.verify(), "a channel with dead air is not a valid linear schedule");
    }

    #[test]
    fn a_misreported_program_as_run_is_rejected() {
        let g = GrantRef::unbounded("video.channel");
        // Window [0, 5s): only show-A aired.
        let mut r = ChannelReceipt::seal(&key(3), "o", "c", schedule(), 0, 5_000, 0, None, g, None);
        assert_eq!(r.aired, vec![0]);
        assert!(r.verify());
        r.aired = vec![0, 1]; // claim the ad aired when it did not
        r.sig = key(3).sign(&r.signing_payload()).to_bytes();
        assert!(!r.verify(), "an aired set the window does not produce must fail");
    }

    #[test]
    fn windows_chain_and_energy_amortises() {
        let g = GrantRef::unbounded("video.channel");
        let w1 = ChannelReceipt::seal(&key(4), "o", "c", schedule(), 0, 10_000, 3_000, Some(EnergyProvenance::ModelBased), g.clone(), None);
        assert!(w1.verify());
        // The next window chains from the first — the channel history is a chain.
        let w2 = ChannelReceipt::seal(&key(4), "o", "c", schedule(), 10_000, 22_000, 6_000, Some(EnergyProvenance::ModelBased), g, Some(w1.receipt_hash()));
        assert!(w2.verify());
        assert_eq!(w2.parent_receipt_hash, Some(w1.receipt_hash()));
        let (uj, _) = w2.uj_per_aired().unwrap();
        assert!((uj - 3000.0).abs() < 1e-9, "6000 uJ / 2 aired (ad-1, show-B) = 3000 each");
    }


    fn hex(b: &[u8]) -> String {
        b.iter().map(|x| format!("{x:02x}")).collect()
    }

    /// A receipt sealed with the legacy `HwShunt` label keeps its signed bytes.
    #[test]
    #[allow(deprecated)] // sealing a legacy HwShunt receipt is the point
    fn legacy_hwshunt_receipt_keeps_its_signed_bytes() {
        let g = GrantRef::unbounded("video.channel");
        let r = ChannelReceipt::seal(&key(1), "did:key:chan", "fast-42", schedule(), 8_000, 13_000, 6_000, Some(EnergyProvenance::HwShunt), g, None);
        assert!(r.verify());
        assert_eq!(hex(&r.receipt_hash()), "4850a8245f9d3e62227a56791e68689dfa7a566e7d18665bf5dc89521a8e2841");
    }

    /// The signature is verified strictly (SPEC §7 *Signatures*): a key of small order, a
    /// non-canonical spelling of one, and an R of small order are each refused, though a
    /// lenient check accepts each.
    #[test]
    fn the_signature_is_verified_strictly() {
        let g = GrantRef::unbounded("video.channel");
        let r = ChannelReceipt::seal(&key(1), "did:key:chan", "fast-42", schedule(), 8_000, 13_000, 6_000, Some(crate::energy_class::test_on_chip()), g, None);
        crate::sig::assert_refuses_forgeries(&r, |r, k, s| (r.signer_pubkey, r.sig) = (k, s), |r| r.signing_payload(), |r| r.verify());
    }

    /// The JSON carriage (revenue-share §4.1) reads back the receipt it wrote, for every class,
    /// so the identity and the verdict a verifier computes from it are the receipt's; the legacy
    /// label, an extra member and an entry whose end passes a u64 do not read.
    #[test]
    fn the_carriage_reads_back_what_it_writes() {
        use crate::energy_class::{CounterFiltering, EnergyProvenance as P, Uncertainty};
        let unc = Uncertainty { relative_ppm: 50_000, window_us: 1_000 };
        let classes = [
            None,
            Some(P::ModelBased),
            Some(P::Estimator),
            Some(P::OnChipCounter { uncertainty: unc, filtering: CounterFiltering::On }),
            Some(P::CalibratedInstrument { uncertainty: unc, calibration_ref: "lab-7".into() }),
        ];
        for (k, class) in classes.into_iter().enumerate() {
            let j = if class.is_some() { 6_000 } else { 0 };
            let mut g = GrantRef::unbounded("video.channel");
            g.funds_ceiling = (k % 2 == 1).then_some(7);
            let parent = (k % 2 == 0).then_some([9u8; 32]);
            let r = ChannelReceipt::seal(&key(5), "did:key:chan", "fast-42", schedule(), 8_000, 13_000, j, class, g, parent);
            assert!(r.verify());
            let back = ChannelReceipt::from_json(&r.to_json()).expect("the carriage reads");
            assert_eq!(back, r);
            assert_eq!(back.receipt_hash(), r.receipt_hash());
        }
        let r = ChannelReceipt::seal(&key(5), "o", "c", schedule(), 0, 5_000, 0, None, GrantRef::unbounded("video.channel"), None);
        let mut extra = r.to_json();
        extra["form"] = serde_json::json!(1);
        assert!(ChannelReceipt::from_json(&extra).is_none());
        let mut legacy = r.to_json();
        legacy["energy_provenance"] = serde_json::json!("HwShunt");
        assert!(ChannelReceipt::from_json(&legacy).is_none());
        let mut past = r.to_json();
        past["schedule"][0]["start_ms"] = serde_json::json!(u64::MAX);
        assert!(ChannelReceipt::from_json(&past).is_none());
    }
}
