//! **joule-moq-web**: `joule-moq`'s MOQT codec (`draft-ietf-moq-transport-22`, and `-23`
//! through `joule-moq`'s bridge) and its receipts, for a browser.
//!
//! The page speaks the draft its transport negotiated: [`set_draft`] (`setDraft`, 22 or 23)
//! sets it once the session is open, and every encoder and every decoder made after it
//! speaks that draft on the wire. The page itself is written in -22's model: a -23 session's
//! messages and fetch entries are mapped to and from -22's by `joule_moq::draft23::bridge`,
//! so the JSON a page reads and writes is -22's view whichever draft is on the wire.
//!
//! The crate builds `joule-moq` without its async runtime (`default-features = false`), so
//! what runs in the page is the same codec and the same receipt verifier the relay runs,
//! compiled for `wasm32-unknown-unknown`. The surface is small and byte-oriented: a page
//! hands it the bytes it read from a stream and gets back the messages and objects in
//! them, and asks it for the bytes of what it sends.
//!
//! - Control: the client's SETUP ([`client_setup`]); every request and answer a client
//!   sends, from a JSON description ([`encode_control`]); framed control messages read
//!   from a control or request stream as their bytes arrive ([`MessageStream`]).
//! - Data: subgroup streams written object by object ([`SubgroupWriter`]); subgroup and
//!   fetch streams read as their bytes arrive ([`UniDecoder`]).
//! - Receipts: a publisher signs each group of a media track into its group receipt and
//!   writes the group's receipts-track object and the media object with its `WAI_RECEIPT`
//!   property ([`TrackSigner`]); a subscriber admits each receipts-track object to its
//!   chain and verifies each media object against it, or accounts for an object it did
//!   not receive by inclusion ([`TrackVerifier`]).
//!
//! Signatures are Ed25519, made and verified here by `ed25519-dalek` under the rules
//! `joule-moq`'s verifier holds every receipt to. WebCrypto's Ed25519 is not used: the
//! receipt verifier also needs BLAKE3 and the Merkle walk, which WebCrypto does not have,
//! and one implementation keeps the page's verdicts the relay's.
//!
//! JSON crosses the boundary as strings, in the view `joule-moq`'s vector corpus uses
//! (`draft22::json`): byte strings in lowercase hexadecimal, a namespace as a list of its
//! fields in hexadecimal. Object payloads cross as bytes.

use ed25519_dalek::SigningKey;
use joule_moq::carriage::{object_properties, ObjectExtras, ReceiptBundle, WaiReceiptProperty, SETUP_OPTION_JWP_RECEIPTS};
use joule_moq::draft22::data::{FetchEntry, Limits, ObjectStatus as WireStatus, SubgroupCodec, SubgroupHeader, SubgroupId, SubgroupObject, UniStreamDecoder};
use joule_moq::draft22::kvp::{opt, prop, KeyValuePair, Properties, SetupOptions};
use joule_moq::draft22::params::{GroupOrder, LocationFilter, Parameter, Parameters};
use joule_moq::draft22::{
    json, ControlMessage, Location, NamespaceRequest, Publish, PublishDone, PublishDoneCode, RequestError, RequestErrorCode, RequestOk, RequestUpdate, Setup,
    StreamHeader, SubscribeOk, Token, TrackRequest, UniEvent, WireError,
};
use joule_moq::grant_proof::{self, GrantProof};
use joule_moq::messages::ObjectHeader;
use joule_moq::receipt::{Admission, ChainRefusal, HoldBound, ObjectReceipt, ReceiptBuilder, ReceiptChains, ReceiptVerifier, ReceiptsTrack, VerifiedGroup};
use joule_moq::{GroupId, ObjectId, ObjectStatus, TrackAlias, TrackName, TrackNamespace, JWP_MOQ_EXTENSION_VERSION};
use serde_json::{json, Value};
use std::collections::{BTreeMap, HashMap};
use wasm_bindgen::prelude::*;

/// The ALPN and `WT-Available-Protocols` value of draft-22 (§6.2).
pub const PROTOCOL: &str = "moqt-22";

fn js(e: impl std::fmt::Display) -> JsError {
    JsError::new(&e.to_string())
}

use joule_moq::draft::Draft;
use joule_moq::draft23 as m23;

thread_local! {
    /// The draft this page's session speaks on the wire; -22 until the page says otherwise.
    static DRAFT: std::cell::Cell<Draft> = const { std::cell::Cell::new(Draft::D22) };
}

/// The draft encoders and decoders made from now on speak.
pub fn current_draft() -> Draft {
    DRAFT.with(|d| d.get())
}

/// Set the draft the session speaks: 22 or 23, the drafts a page's transports carry.
pub fn set_draft_number(n: u32) -> Result<(), String> {
    match Draft::from_number(n) {
        Some(d @ (Draft::D22 | Draft::D23)) => {
            DRAFT.with(|c| c.set(d));
            Ok(())
        }
        _ => Err(format!("a page speaks drafts 23 and 22, not {n}")),
    }
}

/// [`set_draft_number`] for a page, once its transport has negotiated the draft.
#[wasm_bindgen(js_name = setDraft)]
pub fn set_draft(n: u32) -> Result<(), JsError> {
    set_draft_number(n).map_err(js)
}

/// The ALPN and `WT-Available-Protocols` value of the draft the page speaks (§6.2).
#[wasm_bindgen]
pub fn protocol() -> String {
    current_draft().alpn_str().into()
}

/// A -22 message as the bytes of the draft the page speaks.
fn wire(m: &ControlMessage) -> Result<Vec<u8>, String> {
    match current_draft() {
        Draft::D23 => {
            let mapped = m23::bridge::control_from22(m, m23::bridge::At::ControlStream).map_err(|u| u.to_string())?;
            let m = mapped.msg.ok_or_else(|| format!("{} has no -23 form", m.name()))?;
            m.encode().map_err(|e| e.to_string())
        }
        _ => m.encode().map_err(|e| e.to_string()),
    }
}

// ---------------------------------------------------------------------------------------
// Control messages
// ---------------------------------------------------------------------------------------

/// A namespace from a JSON array of its fields, each a UTF-8 string.
fn namespace_of(v: &Value) -> Result<TrackNamespace, String> {
    let fields = v.as_array().ok_or("a namespace is an array of strings")?;
    let mut out = Vec::with_capacity(fields.len());
    for f in fields {
        out.push(f.as_str().ok_or("a namespace field is a string")?.to_owned());
    }
    Ok(TrackNamespace::from_strs(out.iter().map(String::as_str)))
}

fn u64_of(v: &Value, k: &str) -> Result<u64, String> {
    v.get(k).and_then(Value::as_u64).ok_or_else(|| format!("`{k}` is an unsigned integer"))
}

fn location_of(v: &Value) -> Result<Location, String> {
    let a = v.as_array().filter(|a| a.len() == 2).ok_or("a location is [group, object]")?;
    match (a[0].as_u64(), a[1].as_u64()) {
        (Some(g), Some(o)) => Ok(Location::new(g, o)),
        _ => Err("a location is [group, object]".into()),
    }
}

fn filter_of(v: &Value) -> Result<LocationFilter, String> {
    let ty = v.get("type").and_then(Value::as_str).ok_or("a location filter names its `type`")?;
    Ok(match ty {
        "none" => LocationFilter::None,
        "next" => LocationFilter::NextObject,
        "relative" => LocationFilter::RelativeStart { start_group: u64_of(v, "start_group")? },
        "absolute" => LocationFilter::AbsoluteStart { start: location_of(&v["start"])? },
        "absolute_group_end" => LocationFilter::AbsoluteStartGroupEnd { start: location_of(&v["start"])?, end_group_delta: u64_of(v, "end_group_delta")? },
        "range" => LocationFilter::AbsoluteRange { start: location_of(&v["start"])?, end_group_delta: u64_of(v, "end_group_delta")?, end_object: u64_of(v, "end_object")? },
        other => return Err(format!("no location filter `{other}`")),
    })
}

fn token_of(hex_value: &str) -> Result<Token, String> {
    Ok(Token::UseValue { token_type: 0, value: hex::decode(hex_value).map_err(|e| format!("a token's value: {e}"))? })
}

/// Message Parameters from a JSON object (§9.20): `subscriber_priority`, `group_order`
/// (1 ascending, 2 descending), `forward`, `location_filter`, `object_delivery_timeout`,
/// `subgroup_delivery_timeout`, `rendezvous_timeout`, `fill_timeout`, `new_group_request`,
/// `authorization_token` (the hexadecimal value of a token of type 0, by value),
/// `sync_support` (`true`: playout timing's Sync Support) and `fill` (FILL_PARAMETERS, the
/// same form).
fn params_of(v: &Value) -> Result<Parameters, String> {
    let mut p = Parameters::new();
    let Some(o) = v.as_object() else {
        return if v.is_null() { Ok(p) } else { Err("parameters are an object".into()) };
    };
    for (k, x) in o {
        let n = || x.as_u64().ok_or_else(|| format!("`{k}` is an unsigned integer"));
        p = p.with(match k.as_str() {
            "subscriber_priority" => Parameter::SubscriberPriority(u8::try_from(n()?).map_err(|_| "a priority is 0 to 255")?),
            "group_order" => Parameter::GroupOrder(GroupOrder::from_u8(u8::try_from(n()?).unwrap_or(0), "group order").map_err(|e| e.to_string())?),
            "forward" => Parameter::Forward(x.as_bool().ok_or("`forward` is a boolean")?),
            "location_filter" => Parameter::LocationFilter(filter_of(x)?),
            "object_delivery_timeout" => Parameter::ObjectDeliveryTimeout(n()?),
            "subgroup_delivery_timeout" => Parameter::SubgroupDeliveryTimeout(n()?),
            "rendezvous_timeout" => Parameter::RendezvousTimeout(n()?),
            "fill_timeout" => Parameter::FillTimeout(n()?),
            "new_group_request" => Parameter::NewGroupRequest(n()?),
            "authorization_token" => Parameter::AuthorizationToken(token_of(x.as_str().ok_or("a token is hexadecimal")?)?),
            "fill" => Parameter::FillParameters(params_of(x)?),
            // Sync Support (draft-nandakumar-moq-tempo-00 §2.4): the relay writes its HopTime
            // on each object. A placeholder, sent only to a relay whose SETUP declared the
            // option (joule_moq::tempo); `true` alone.
            "sync_support" if x.as_bool() == Some(true) => Parameter::SyncSupport,
            other => return Err(format!("no parameter `{other}` here")),
        });
    }
    Ok(p)
}

/// Track Properties from a JSON object: `publisher_priority` (DEFAULT_PUBLISHER_PRIORITY)
/// and `object_delivery_timeout` (OBJECT_DELIVERY_TIMEOUT), in milliseconds.
fn track_props_of(v: &Value) -> Result<Properties, String> {
    let mut p = Properties::new();
    if let Some(x) = v.get("publisher_priority") {
        p.push(KeyValuePair::varint(prop::DEFAULT_PUBLISHER_PRIORITY, x.as_u64().ok_or("`publisher_priority` is an integer")?));
    }
    if let Some(x) = v.get("object_delivery_timeout") {
        p.push(KeyValuePair::varint(prop::OBJECT_DELIVERY_TIMEOUT, x.as_u64().ok_or("`object_delivery_timeout` is an integer")?));
    }
    // Rating claims, each a `WAI_RATING` Track Property (WAI age-assurance §5.7).
    if let Some(x) = v.get("ratings") {
        for c in x.as_array().ok_or("`ratings` is a list of claims")? {
            p.push(joule_moq::rating::property(c.as_str().ok_or("a rating claim is its text")?));
        }
    }
    Ok(p)
}

/// A rating claim (WAI age-assurance §3.1) by the principal whose seed is `seed` about its own
/// track `name` under the namespace `fields` (live-rooms §6.1), from group `from_group` on,
/// rated `min_age` under `scheme` with `value`, issued at the date-time `issued_at`: its text,
/// which a relay honours as the track owner's (§3.6).
#[allow(clippy::too_many_arguments)]
pub fn rating_claim_text(seed: [u8; 32], fields: &[&str], name: &str, from_group: u64, scheme: &str, value: &str, min_age: u64, issued_at: &str) -> Result<String, String> {
    let ns = TrackNamespace::from_strs(fields.iter().copied());
    let track = joule_moq::rating::track_digest(&ns, &TrackName::from_str(name));
    let text = joule_moq::rating::write(&SigningKey::from_bytes(&seed), &joule_moq::rating::Subject::Track { track, from_group }, &[joule_moq::rating::Rating { scheme: scheme.into(), value: value.into(), min_age }], issued_at);
    joule_moq::rating::read(text.as_bytes()).map_err(|r| format!("the claim does not read: {}", r.code()))?;
    Ok(text)
}

/// [`rating_claim_text`] for a page: `fields` a JSON list of the namespace's fields.
#[wasm_bindgen(js_name = ratingClaim)]
#[allow(clippy::too_many_arguments)]
pub fn rating_claim(seed: &[u8], fields: &str, name: &str, from_group: u64, scheme: &str, value: &str, min_age: u64, issued_at: &str) -> Result<String, JsError> {
    let seed: [u8; 32] = seed.try_into().map_err(|_| JsError::new("a seed is 32 bytes"))?;
    let fields: Vec<String> = serde_json::from_str(fields).map_err(js)?;
    let f: Vec<&str> = fields.iter().map(String::as_str).collect();
    rating_claim_text(seed, &f, name, from_group, scheme, value, min_age, issued_at).map_err(js)
}

/// The rating claims a property list carries (age-assurance §5.7): from the `track_properties`
/// of a SUBSCRIBE_OK in this crate's JSON view, the text of each `WAI_RATING`.
#[wasm_bindgen(js_name = ratingsOf)]
pub fn ratings_of(properties_json: &str) -> Result<String, JsError> {
    let v: Value = serde_json::from_str(properties_json).map_err(js)?;
    let out: Vec<String> = v
        .as_array()
        .into_iter()
        .flatten()
        .filter(|p| p["type"].as_u64() == Some(joule_moq::rating::WAI_RATING))
        .filter_map(|p| p["bytes"].as_str().and_then(|h| hex::decode(h).ok()).and_then(|b| String::from_utf8(b).ok()))
        .collect();
    Ok(serde_json::to_string(&out).expect("JSON"))
}

/// The client's SETUP (§9.1): its implementation's name, MAX_REQUEST_UPDATES when not
/// zero, an authorization token of type 0 by value when given, and `joule-moq`'s receipt
/// extension when `receipts` (Setup Option `0x3A46`). Over WebTransport a client sends no
/// PATH or AUTHORITY, and none is sent here.
pub fn setup_bytes(token: Option<&[u8]>, implementation: &str, receipts: bool, max_request_updates: u64) -> Result<Vec<u8>, String> {
    let mut o = SetupOptions::new();
    o.push(KeyValuePair::bytes(opt::MOQT_IMPLEMENTATION, implementation.as_bytes().to_vec()));
    if max_request_updates != 0 {
        o.push(KeyValuePair::varint(opt::MAX_REQUEST_UPDATES, max_request_updates));
    }
    if let Some(t) = token {
        o.push(KeyValuePair::bytes(opt::AUTHORIZATION_TOKEN, Token::UseValue { token_type: 0, value: t.to_vec() }.to_bytes()));
    }
    if receipts {
        o.push(KeyValuePair::varint(SETUP_OPTION_JWP_RECEIPTS, JWP_MOQ_EXTENSION_VERSION as u64));
    }
    wire(&ControlMessage::Setup(Setup { options: o }))
}

/// The bytes a client's control stream begins with: the stream type and its SETUP (§6.3),
/// which [`setup_bytes`] makes; SETUP's type is the control stream's type.
#[wasm_bindgen(js_name = clientSetup)]
pub fn client_setup(token: Option<Vec<u8>>, implementation: &str, receipts: bool, max_request_updates: u64) -> Result<Vec<u8>, JsError> {
    setup_bytes(token.as_deref(), implementation, receipts, max_request_updates).map_err(js)
}

/// The client's SETUP presenting a room grant (WAI live-rooms §10.5.2): the grant's record
/// as an AUTHORIZATION TOKEN (USE_VALUE, type 0) and its proof (`GRANT_PROOF`), signed by
/// the principal's key, from `seed`, over the relay's `challenge` for `target`, the
/// `//authority/path` the client addressed. A grant is public, so the proof is what makes
/// presenting it mean anything; the relay refuses a session that presents one without it.
pub fn setup_with_grant_bytes(record: &[u8], seed: [u8; 32], challenge: [u8; 32], target: &str, implementation: &str, receipts: bool, max_request_updates: u64) -> Result<Vec<u8>, String> {
    let (digest, principal) = grant_proof::grant_digest_and_principal(record).ok_or("not a WRG1 grant record")?;
    let key = SigningKey::from_bytes(&seed);
    if key.verifying_key().to_bytes() != principal {
        return Err("the seed is not the grant's principal's".into());
    }
    let proof = GrantProof::sign(digest, &key, challenge, &format!("{} {target}", grant_proof::VERB_SETUP)).ok_or("a target a proof cannot carry")?;
    setup_presenting(record, &proof.to_bytes(), implementation, receipts, max_request_updates)
}

/// The client's SETUP presenting `record` beside the proof record `proof`, as given.
pub fn setup_presenting(record: &[u8], proof: &[u8], implementation: &str, receipts: bool, max_request_updates: u64) -> Result<Vec<u8>, String> {
    let proof = GrantProof::read(proof).map_err(|r| r.code().to_string())?;
    let mut o = SetupOptions::new();
    o.push(KeyValuePair::bytes(opt::MOQT_IMPLEMENTATION, implementation.as_bytes().to_vec()));
    if max_request_updates != 0 {
        o.push(KeyValuePair::varint(opt::MAX_REQUEST_UPDATES, max_request_updates));
    }
    grant_proof::add_grant(&mut o, record, &proof);
    if receipts {
        o.push(KeyValuePair::varint(SETUP_OPTION_JWP_RECEIPTS, JWP_MOQ_EXTENSION_VERSION as u64));
    }
    wire(&ControlMessage::Setup(Setup { options: o }))
}

/// [`setup_with_grant_bytes`] for a page: the challenge is a `GRANT_CHALLENGE` option
/// (`0x3A47`) read from a SETUP of the relay's on an earlier session, since the relay reads
/// this SETUP before it sends its own.
#[wasm_bindgen(js_name = clientSetupWithGrant)]
pub fn client_setup_with_grant(record: &[u8], seed: &[u8], challenge: &[u8], target: &str, implementation: &str, receipts: bool, max_request_updates: u64) -> Result<Vec<u8>, JsError> {
    let seed: [u8; 32] = seed.try_into().map_err(|_| JsError::new("a seed is 32 bytes"))?;
    let challenge: [u8; 32] = challenge.try_into().map_err(|_| JsError::new("a challenge is 32 bytes"))?;
    setup_with_grant_bytes(record, seed, challenge, target, implementation, receipts, max_request_updates).map_err(js)
}

/// A proof record (WAI live-rooms §10.5.1) for the grant `record`, by the key of `seed`
/// whatever key that is, over `challenge`, for `target`. For a test that a relay refuses a
/// proof by a key that is not the grant's principal (§10.5.4 step 4): a page proving its
/// own grant uses [`client_setup_with_grant`], which signs with the principal's key alone.
#[wasm_bindgen(js_name = grantProofBy)]
pub fn grant_proof_by(record: &[u8], seed: &[u8], challenge: &[u8], target: &str) -> Result<Vec<u8>, JsError> {
    let seed: [u8; 32] = seed.try_into().map_err(|_| JsError::new("a seed is 32 bytes"))?;
    let challenge: [u8; 32] = challenge.try_into().map_err(|_| JsError::new("a challenge is 32 bytes"))?;
    let (digest, _) = grant_proof::grant_digest_and_principal(record).ok_or_else(|| JsError::new("not a WRG1 grant record"))?;
    let p = GrantProof::sign(digest, &SigningKey::from_bytes(&seed), challenge, &format!("{} {target}", grant_proof::VERB_SETUP)).ok_or_else(|| JsError::new("a target a proof cannot carry"))?;
    Ok(p.to_bytes())
}

/// A proof record (WAI live-rooms §10.5.1) for the grant `record` by its principal, the key of
/// `seed`, over `challenge`, for an HTTP request of method `verb` to `target` (`//authority`
/// and the path the page addressed): what the request's `WAI-Room-Proof` carries (§10.5.3). A
/// seed whose key is not the grant's principal makes no proof.
pub fn http_proof_bytes(record: &[u8], seed: [u8; 32], challenge: [u8; 32], verb: &str, target: &str) -> Result<Vec<u8>, String> {
    let (digest, principal) = grant_proof::grant_digest_and_principal(record).ok_or("not a WRG1 grant record")?;
    let key = SigningKey::from_bytes(&seed);
    if key.verifying_key().to_bytes() != principal {
        return Err("the seed is not the grant's principal's".into());
    }
    let p = GrantProof::sign(digest, &key, challenge, &format!("{verb} {target}")).ok_or("a target a proof cannot carry")?;
    Ok(p.to_bytes())
}

/// [`http_proof_bytes`] for a page.
#[wasm_bindgen(js_name = grantProofHttp)]
pub fn grant_proof_http(record: &[u8], seed: &[u8], challenge: &[u8], verb: &str, target: &str) -> Result<Vec<u8>, JsError> {
    let seed: [u8; 32] = seed.try_into().map_err(|_| JsError::new("a seed is 32 bytes"))?;
    let challenge: [u8; 32] = challenge.try_into().map_err(|_| JsError::new("a challenge is 32 bytes"))?;
    http_proof_bytes(record, seed, challenge, verb, target).map_err(js)
}

/// A proof record (WAI live-rooms §10.5.1) naming no grant (its `grant` 32 zero bytes), by the
/// key of `seed`, over `challenge`, for an HTTP request of method `verb` to `target`: what a
/// request proving a key alone carries in `WAI-Room-Proof`, as a conversation's join and a
/// call's inbox take it (WAI live-messaging §10.4).
#[wasm_bindgen(js_name = keyProofHttp)]
pub fn key_proof_http(seed: &[u8], challenge: &[u8], verb: &str, target: &str) -> Result<Vec<u8>, JsError> {
    let seed: [u8; 32] = seed.try_into().map_err(|_| JsError::new("a seed is 32 bytes"))?;
    let challenge: [u8; 32] = challenge.try_into().map_err(|_| JsError::new("a challenge is 32 bytes"))?;
    let p = GrantProof::sign([0; 32], &SigningKey::from_bytes(&seed), challenge, &format!("{verb} {target}")).ok_or_else(|| JsError::new("a target a proof cannot carry"))?;
    Ok(p.to_bytes())
}

/// [`setup_presenting`] for a page.
#[wasm_bindgen(js_name = clientSetupWithProof)]
pub fn client_setup_with_proof(record: &[u8], proof: &[u8], implementation: &str, receipts: bool, max_request_updates: u64) -> Result<Vec<u8>, JsError> {
    setup_presenting(record, proof, implementation, receipts, max_request_updates).map_err(js)
}

/// The Setup Option type of the relay's challenge (WAI live-rooms §10.5.2).
#[wasm_bindgen(js_name = grantChallengeOption)]
pub fn grant_challenge_option() -> u64 {
    grant_proof::SETUP_OPTION_GRANT_CHALLENGE
}

/// The Setup Option type by which a relay says it refuses the session at SETUP, its value
/// the reason's code (WAI live-rooms §10.5.2).
#[wasm_bindgen(js_name = grantRefusedOption)]
pub fn grant_refused_option() -> u64 {
    grant_proof::SETUP_OPTION_GRANT_REFUSED
}

/// A control message a client sends, from its JSON description: `message` is its name in
/// draft-22's Table 5, and the rest are its fields:
///
/// - `SUBSCRIBE`, `FETCH`, `TRACK_STATUS`: `request_id`, `namespace` (an array of UTF-8
///   fields), `name` (UTF-8) and `params` (see [`params_of`]);
/// - `PUBLISH`: those, `track_alias` and `track_properties`;
/// - `PUBLISH_NAMESPACE`, `SUBSCRIBE_NAMESPACE`: `request_id`, `prefix` and `params`;
/// - `REQUEST_OK`: `params` and `track_properties`; `SUBSCRIBE_OK`: `track_alias` too;
/// - `REQUEST_UPDATE`: `request_id`, `params`;
/// - `REQUEST_ERROR`: `code`, `retry_interval`, `reason`;
/// - `PUBLISH_DONE`: `status`, `stream_count`, `reason`.
pub fn control_bytes(desc: &str) -> Result<Vec<u8>, String> {
    let v: Value = serde_json::from_str(desc).map_err(|e| format!("a message description: {e}"))?;
    let name = v.get("message").and_then(Value::as_str).ok_or("a message description names its `message`")?;
    let params = || params_of(v.get("params").unwrap_or(&Value::Null));
    let props = || track_props_of(v.get("track_properties").unwrap_or(&Value::Null));
    let track = || -> Result<TrackRequest, String> {
        Ok(TrackRequest {
            request_id: u64_of(&v, "request_id")?,
            namespace: namespace_of(&v["namespace"])?,
            name: TrackName::from_str(v.get("name").and_then(Value::as_str).ok_or("`name` is a string")?),
            params: params()?,
        })
    };
    let ns_req = || -> Result<NamespaceRequest, String> { Ok(NamespaceRequest { request_id: u64_of(&v, "request_id")?, prefix: namespace_of(&v["prefix"])?, params: params()? }) };
    let reason = || v.get("reason").and_then(Value::as_str).unwrap_or("").as_bytes().to_vec();
    let m = match name {
        "SUBSCRIBE" => ControlMessage::Subscribe(track()?),
        "FETCH" => ControlMessage::Fetch(track()?),
        "TRACK_STATUS" => ControlMessage::TrackStatus(track()?),
        "PUBLISH" => {
            let t = track()?;
            ControlMessage::Publish(Publish { request_id: t.request_id, namespace: t.namespace, name: t.name, track_alias: u64_of(&v, "track_alias")?, params: t.params, track_properties: props()? })
        }
        "PUBLISH_NAMESPACE" => ControlMessage::PublishNamespace(ns_req()?),
        "SUBSCRIBE_NAMESPACE" => ControlMessage::SubscribeNamespace(ns_req()?),
        "REQUEST_OK" => ControlMessage::RequestOk(RequestOk { params: params()?, track_properties: props()? }),
        "SUBSCRIBE_OK" => ControlMessage::SubscribeOk(SubscribeOk { track_alias: u64_of(&v, "track_alias")?, params: params()?, track_properties: props()? }),
        "REQUEST_UPDATE" => ControlMessage::RequestUpdate(RequestUpdate { request_id: u64_of(&v, "request_id")?, params: params()? }),
        "REQUEST_ERROR" => ControlMessage::RequestError(RequestError {
            code: RequestErrorCode(u64_of(&v, "code")?),
            retry_interval: v.get("retry_interval").and_then(Value::as_u64).unwrap_or(0),
            reason: reason(),
            redirect: None,
        }),
        "PUBLISH_DONE" => ControlMessage::PublishDone(PublishDone {
            status: PublishDoneCode(u64_of(&v, "status")?),
            stream_count: v.get("stream_count").and_then(Value::as_u64).unwrap_or(u64::MAX),
            reason: reason(),
        }),
        other => return Err(format!("a client does not send `{other}` here")),
    };
    wire(&m)
}

#[wasm_bindgen(js_name = encodeControl)]
pub fn encode_control(desc: &str) -> Result<Vec<u8>, JsError> {
    control_bytes(desc).map_err(js)
}

/// A refusal as JSON: its kind and the consequence draft-22 gives it.
fn wire_error(e: &WireError) -> String {
    json::error(e).to_string()
}

/// A refusal as JSON with the consequence its draft gives it: at -23, the code -23 gives the
/// fault on `plane` (`joule_moq::draft23::error`).
fn wire_error_at(e: &WireError, draft: Draft, plane: m23::Plane) -> String {
    match draft {
        Draft::D23 => m23::json::error(e, plane).to_string(),
        _ => wire_error(e),
    }
}

/// Framed control messages (§9: Type, a Length, the body: 16 bits at -22, a varint at -23)
/// read from a request stream as its bytes arrive. Each [`MessageStream::next`] gives the
/// next whole message as JSON (`draft22::json::control`, whichever draft is on the wire),
/// or nothing until more bytes come.
#[wasm_bindgen]
pub struct MessageStream {
    buf: Vec<u8>,
    draft: Draft,
    /// The request this stream carries, when the page said ([`MessageStream::for_request`]):
    /// at -23 a FETCH's answer is a REQUEST_OK, which the page reads as FETCH_OK.
    kind: Option<joule_moq::draft22::RequestKind>,
    fetch: Option<Option<LocationFilter>>,
    answered: bool,
}

impl Default for MessageStream {
    fn default() -> Self {
        MessageStream { buf: Vec::new(), draft: current_draft(), kind: None, fetch: None, answered: false }
    }
}

impl MessageStream {
    fn framed_len(b: &[u8]) -> Option<usize> {
        let first = *b.first()?;
        let tlen = joule_moq::draft22::varint::len_from_first(first);
        if b.len() < tlen + 2 {
            return None;
        }
        Some(tlen + 2 + u16::from_be_bytes([b[tlen], b[tlen + 1]]) as usize)
    }

    fn framed_len23(b: &[u8]) -> Option<usize> {
        let (_, tl) = joule_moq::draft22::varint::get(b, "").ok()?;
        let (len, ll) = joule_moq::draft22::varint::get(&b[tl..], "").ok()?;
        tl.checked_add(ll)?.checked_add(usize::try_from(len).ok()?)
    }

    pub fn next_message(&mut self) -> Result<Option<ControlMessage>, WireError> {
        if self.draft == Draft::D23 {
            return self.next_message23();
        }
        match Self::framed_len(&self.buf).filter(|n| *n <= self.buf.len()) {
            Some(n) => {
                let m = ControlMessage::decode_exact(&self.buf[..n])?;
                self.buf.drain(..n);
                Ok(Some(m))
            }
            None => Ok(None),
        }
    }

    fn next_message23(&mut self) -> Result<Option<ControlMessage>, WireError> {
        // A Length past MAX_CONTROL_MESSAGE_SIZE is refused before its body is waited for.
        if let Ok((_, tl)) = joule_moq::draft22::varint::get(&self.buf, "")
            && let Ok((len, _)) = joule_moq::draft22::varint::get(&self.buf[tl..], "")
            && len > m23::control::DEFAULT_MAX_CONTROL_MESSAGE_SIZE
        {
            return Err(WireError::LengthExceedsLimit { what: "control message", len, limit: m23::control::DEFAULT_MAX_CONTROL_MESSAGE_SIZE });
        }
        let Some(n) = Self::framed_len23(&self.buf).filter(|n| *n <= self.buf.len()) else { return Ok(None) };
        let m = m23::ControlMessage::decode_exact(&self.buf[..n])?;
        self.buf.drain(..n);
        let at = match self.kind {
            Some(kind) => m23::bridge::At::Request { kind, answers_update: self.answered && matches!(m, m23::ControlMessage::RequestOk(_) | m23::ControlMessage::RequestError(_)), fetch: self.fetch },
            None => m23::bridge::At::ControlStream,
        };
        if matches!(m, m23::ControlMessage::RequestOk(_) | m23::ControlMessage::RequestError(_) | m23::ControlMessage::SubscribeOk(_)) {
            self.answered = true;
        }
        let mapped = m23::bridge::control_to22(&m, at).map_err(|u| WireError::Unencodable { what: "a -23 message", reason: u.reason })?;
        Ok(Some(mapped.msg))
    }
}

#[wasm_bindgen]
impl MessageStream {
    #[wasm_bindgen(constructor)]
    pub fn new() -> MessageStream {
        MessageStream::default()
    }

    /// The request this stream answers, as the description [`encode_control`] took: at -23
    /// the answer to a FETCH is read as FETCH_OK, its End Location from LARGEST_OBJECT or the
    /// range the FETCH asked for.
    #[wasm_bindgen(js_name = forRequest)]
    pub fn for_request(&mut self, desc: &str) -> Result<(), JsError> {
        use joule_moq::draft22::RequestKind as K;
        let v: Value = serde_json::from_str(desc).map_err(js)?;
        self.kind = match v.get("message").and_then(Value::as_str) {
            Some("SUBSCRIBE") => Some(K::Subscribe),
            Some("FETCH") => Some(K::Fetch),
            Some("TRACK_STATUS") => Some(K::TrackStatus),
            Some("PUBLISH") => Some(K::Publish),
            Some("PUBLISH_NAMESPACE") => Some(K::PublishNamespace),
            Some("SUBSCRIBE_NAMESPACE") => Some(K::SubscribeNamespace),
            _ => None,
        };
        if self.kind == Some(K::Fetch) {
            self.fetch = Some(params_of(v.get("params").unwrap_or(&Value::Null)).map_err(js)?.location_filter());
        }
        Ok(())
    }

    pub fn push(&mut self, bytes: &[u8]) {
        self.buf.extend_from_slice(bytes);
    }

    /// The next whole message as JSON, or `undefined`. (`next` is the page's name for
    /// it; this is not an iterator.)
    #[allow(clippy::should_implement_trait)]
    pub fn next(&mut self) -> Result<Option<String>, JsError> {
        let draft = self.draft;
        self.next_message().map(|m| m.map(|m| json::control(&m).to_string())).map_err(|e| JsError::new(&wire_error_at(&e, draft, m23::Plane::Control)))
    }

    /// Bytes held that are not yet a whole message: a stream that ends with any is cut
    /// inside a message.
    pub fn buffered(&self) -> usize {
        self.buf.len()
    }
}

// ---------------------------------------------------------------------------------------
// Data streams
// ---------------------------------------------------------------------------------------

/// One item read from a unidirectional stream: `kind` is `header`, `object`,
/// `fetch_object`, `end_of_range`, `control` or `padding`; `json` describes it without its
/// payload; [`UniItem::payload`] is the object's payload.
#[wasm_bindgen]
pub struct UniItem {
    kind: String,
    json: String,
    payload: Vec<u8>,
    properties: Option<Properties>,
}

#[wasm_bindgen]
impl UniItem {
    #[wasm_bindgen(getter)]
    pub fn kind(&self) -> String {
        self.kind.clone()
    }
    #[wasm_bindgen(getter)]
    pub fn json(&self) -> String {
        self.json.clone()
    }
    /// The payload, moved out: a second call gives an empty array.
    pub fn payload(&mut self) -> Vec<u8> {
        std::mem::take(&mut self.payload)
    }

    /// An object's property list as the stream carried it, encoded (MOQT §8.3): what a
    /// member opens a sealed object with, its Immutable Properties byte for byte. Empty for
    /// an object with none, and for an item that is not an object.
    pub fn properties(&self) -> Vec<u8> {
        self.properties.as_ref().and_then(|p| p.to_bytes().ok()).unwrap_or_default()
    }
}

fn item_of(e: UniEvent) -> UniItem {
    match e {
        UniEvent::SubgroupObject(o) => UniItem {
            kind: "object".into(),
            json: json!({
                "object_id": o.object_id,
                "status": o.status.code(),
                "properties": json::properties(&o.properties),
            })
            .to_string(),
            payload: o.payload,
            properties: Some(o.properties),
        },
        UniEvent::FetchEntry(FetchEntry::Object(o)) => UniItem {
            kind: "fetch_object".into(),
            json: json!({
                "group_id": o.group_id,
                "object_id": o.object_id,
                "subgroup_id": o.subgroup_id,
                "publisher_priority": o.publisher_priority,
            })
            .to_string(),
            payload: o.payload,
            properties: o.properties,
        },
        other => {
            let v = json::uni_event(&other);
            let kind = match &other {
                UniEvent::Header(_) => "header",
                UniEvent::Control(_) => "control",
                UniEvent::FetchEntry(_) => "end_of_range",
                UniEvent::Padding(_) => "padding",
                _ => "other",
            };
            UniItem { kind: kind.into(), json: v.to_string(), payload: Vec::new(), properties: None }
        }
    }
}

/// The items of one unidirectional stream (a control, subgroup, fetch or padding stream),
/// each given once all of it has arrived: `joule-moq`'s incremental decoder, bounded, of
/// the draft the page speaks; a -23 stream's items are given in -22's model.
#[wasm_bindgen]
pub struct UniDecoder {
    dec: UniInner,
}

enum UniInner {
    D22(UniStreamDecoder),
    D23(Box<m23::UniStreamDecoder>),
}

fn header23(h: m23::StreamHeader) -> StreamHeader {
    match h {
        m23::StreamHeader::Control => StreamHeader::Control,
        m23::StreamHeader::Subgroup(s) => StreamHeader::Subgroup(s),
        m23::StreamHeader::Fetch(f) => StreamHeader::Fetch(f),
        m23::StreamHeader::Padding => StreamHeader::Padding,
    }
}

impl UniDecoder {
    fn draft(&self) -> Draft {
        match self.dec {
            UniInner::D22(_) => Draft::D22,
            UniInner::D23(_) => Draft::D23,
        }
    }

    /// The next item, in -22's model.
    pub fn next_event(&mut self) -> Result<Option<UniEvent>, WireError> {
        match &mut self.dec {
            UniInner::D22(d) => d.next(),
            UniInner::D23(d) => loop {
                let Some(e) = d.next()? else { return Ok(None) };
                return Ok(Some(match e {
                    m23::UniEvent::Header(h) => UniEvent::Header(header23(h)),
                    m23::UniEvent::Control(m) => UniEvent::Control(
                        m23::bridge::control_to22(&m, m23::bridge::At::ControlStream).map_err(|u| WireError::Unencodable { what: "a -23 message", reason: u.reason })?.msg,
                    ),
                    m23::UniEvent::SubgroupObject(o) => UniEvent::SubgroupObject(o),
                    m23::UniEvent::FetchEntry(f) => match m23::bridge::fetch_entry_to22(&f).msg {
                        Some(f) => UniEvent::FetchEntry(f),
                        // End of Group and End of Track: -22 has no entry for them.
                        None => continue,
                    },
                    m23::UniEvent::Padding(n) => UniEvent::Padding(n),
                }));
            },
        }
    }
}

#[wasm_bindgen]
impl UniDecoder {
    /// A decoder that refuses an object payload longer than `max_payload` bytes.
    #[wasm_bindgen(constructor)]
    pub fn new(max_payload: u64) -> UniDecoder {
        let limits = Limits { max_payload, ..Limits::default() };
        let dec = match current_draft() {
            Draft::D23 => UniInner::D23(Box::new(m23::UniStreamDecoder::new(GroupOrder::Ascending, limits))),
            _ => UniInner::D22(UniStreamDecoder::new(GroupOrder::Ascending, limits)),
        };
        UniDecoder { dec }
    }

    /// The Group Order of the FETCH a fetch stream answers: 1 ascending, 2 descending.
    #[wasm_bindgen(js_name = setGroupOrder)]
    pub fn set_group_order(&mut self, order: u8) {
        let o = if order == 2 { GroupOrder::Descending } else { GroupOrder::Ascending };
        match &mut self.dec {
            UniInner::D22(d) => d.set_group_order(o),
            UniInner::D23(d) => d.set_group_order(o),
        }
    }

    pub fn push(&mut self, bytes: &[u8]) {
        match &mut self.dec {
            UniInner::D22(d) => d.push(bytes),
            UniInner::D23(d) => d.push(bytes),
        }
    }

    #[allow(clippy::should_implement_trait)]
    pub fn next(&mut self) -> Result<Option<UniItem>, JsError> {
        let draft = self.draft();
        self.next_event().map(|e| e.map(item_of)).map_err(|e| JsError::new(&wire_error_at(&e, draft, m23::Plane::Stream)))
    }

    /// At the stream's FIN: a stream that ends inside an item is refused.
    pub fn finish(&self) -> Result<(), JsError> {
        match &self.dec {
            UniInner::D22(d) => d.finish(),
            UniInner::D23(d) => d.finish(),
        }
        .map_err(|e| JsError::new(&wire_error_at(&e, self.draft(), m23::Plane::Stream)))
    }
}

/// A subgroup stream (§11.4), written as its objects come: [`SubgroupWriter::header`]
/// once, then [`SubgroupWriter::object`] for each object, the Object IDs increasing.
#[wasm_bindgen]
pub struct SubgroupWriter {
    header: SubgroupHeader,
    codec: SubgroupCodec,
}

#[wasm_bindgen]
impl SubgroupWriter {
    /// Subgroup 0 of group `group_id` of the track whose alias is `track_alias`, at
    /// `priority`; `end_of_group` when the stream's FIN ends the group.
    #[wasm_bindgen(constructor)]
    pub fn new(track_alias: u64, group_id: u64, priority: u8, end_of_group: bool) -> SubgroupWriter {
        let header = SubgroupHeader { track_alias, group_id, subgroup_id: SubgroupId::Zero, publisher_priority: Some(priority), properties: false, end_of_group, first_object: true };
        SubgroupWriter { header, codec: SubgroupCodec::new(header) }
    }

    pub fn header(&self) -> Vec<u8> {
        let mut b = Vec::new();
        self.header.encode(&mut b);
        b
    }

    pub fn object(&mut self, object_id: u64, payload: &[u8]) -> Result<Vec<u8>, JsError> {
        let mut b = Vec::new();
        self.codec.encode_object(&SubgroupObject::normal(object_id, payload.to_vec()), &mut b).map_err(|e| JsError::new(&wire_error(&e)))?;
        Ok(b)
    }
}

// ---------------------------------------------------------------------------------------
// Receipts: the publisher
// ---------------------------------------------------------------------------------------

/// The unmetered marker's signed class (WAI energy-measurement §2.7, tag 0): a page has no
/// meter, and its figure is never a measurement of zero joules.
const UNMETERED: [u8; 1] = [0];

/// One media group signed and written: the receipts-track stream and the media stream.
pub struct Signed {
    pub receipts_stream: Vec<u8>,
    pub media_stream: Vec<u8>,
    pub content_hash: [u8; 32],
    pub link: [u8; 32],
}

/// The publisher's side of one media track and its receipts track (WAI
/// moq-streaming-format §4.1): each group is one object, signed into its own group
/// receipt, whose parent is the group before it, so the track's receipts form one chain.
pub struct Signer {
    key: SigningKey,
    receipt_alias: u64,
    media_alias: u64,
    receipts_alias: u64,
    priority: u8,
    origin: String,
    object_timeout_ms: Option<u64>,
    parent: Option<[u8; 32]>,
    /// Whether each object is one WAI envelope, flagged so (`WAI_ENVELOPE`): a media track's
    /// are; a world track's op batches are not (WAI live-worlds §4.2).
    envelope: bool,
}

impl Signer {
    pub fn new(seed: [u8; 32], receipt_alias: u64, media_alias: u64, receipts_alias: u64, priority: u8, origin: String, object_timeout_ms: Option<u64>) -> Self {
        Signer { key: SigningKey::from_bytes(&seed), receipt_alias, media_alias, receipts_alias, priority, origin, object_timeout_ms, parent: None, envelope: true }
    }

    /// Objects that are not WAI envelopes from here on: no `WAI_ENVELOPE` flag.
    pub fn set_envelope(&mut self, envelope: bool) {
        self.envelope = envelope;
    }

    pub fn public_key(&self) -> [u8; 32] {
        self.key.verifying_key().to_bytes()
    }

    /// Group `group` of the media track: one object, `payload`, its receipt signed under
    /// the publisher's key with the unmetered marker as its figure; the receipts-track
    /// object of the same group; and the media object carrying its `WAI_RECEIPT` property
    /// (moq-streaming-format §5) and the `WAI_ENVELOPE` flag.
    pub fn sign(&mut self, group: u64, payload: &[u8], started: &str, ended: &str) -> Result<Signed, String> {
        self.sign_with(group, payload, None, started, ended)
    }

    /// Group `group` of a sealed media track (WAI confidentiality §4, §8.1): `payload` is
    /// the sealed object's payload, ciphertext and tag, and the receipt is over those bytes;
    /// `properties` are the object's as its seal made them (the Immutable Properties with
    /// its Key ID and envelope flag, the delivery timeout beside them). A sealed object
    /// carries no `WAI_RECEIPT` (§4.3): its receipt rides the receipts track alone.
    pub fn sign_sealed(&mut self, group: u64, payload: &[u8], properties: Properties, started: &str, ended: &str) -> Result<Signed, String> {
        self.sign_with(group, payload, Some(properties), started, ended)
    }

    fn sign_with(&mut self, group: u64, payload: &[u8], sealed: Option<Properties>, started: &str, ended: &str) -> Result<Signed, String> {
        let hash = ObjectReceipt::hash_payload(payload);
        let mut b = ReceiptBuilder::new(TrackAlias(self.receipt_alias), GroupId(group));
        b.add_labelled_object(hash, 0, self.origin.clone(), started, UNMETERED.to_vec()).map_err(|e| e.to_string())?;
        let receipt = ObjectReceipt {
            track_alias: TrackAlias(self.receipt_alias),
            group_id: GroupId(group),
            object_id: ObjectId(0),
            content_hash: hash,
            joules_micro: 0,
            origin: self.origin.clone(),
            merkle_proof: b.proof_for(0).map_err(|e| e.to_string())?,
            energy_class: Some(UNMETERED.to_vec()),
        };
        let g = b.finalize_labelled(&self.key, self.origin.clone(), ended, self.parent).map_err(|e| e.to_string())?;
        let link = g.link_hash();
        let wai = WaiReceiptProperty::from_receipts(&receipt, &g).ok_or("the receipt gives no valid WAI_RECEIPT property")?;
        let bundle = ReceiptBundle { group: g, objects: vec![receipt] };
        let receipts_stream = bundle.encode_stream(self.receipts_alias, group, self.priority).map_err(|e| e.to_string())?;
        let header = ObjectHeader {
            track_alias: TrackAlias(self.media_alias),
            group_id: GroupId(group),
            object_id: ObjectId(0),
            publisher_priority: self.priority,
            status: ObjectStatus::Normal,
            payload_len: payload.len() as u64,
            mrl_layer: 0,
            mrl_total_layers: 0,
            wai_envelope: self.envelope,
        };
        let props = match sealed {
            Some(p) => p,
            None => object_properties(&header, &ObjectExtras { receipt: None, wai_receipt: Some(wai), delivery_timeout_ms: self.object_timeout_ms }).map_err(|e| e.to_string())?,
        };
        let sh = SubgroupHeader { track_alias: self.media_alias, group_id: group, subgroup_id: SubgroupId::Zero, publisher_priority: Some(self.priority), properties: true, end_of_group: true, first_object: true };
        let mut media_stream = Vec::with_capacity(payload.len() + 256);
        sh.encode(&mut media_stream);
        SubgroupCodec::new(sh)
            .encode_object(&SubgroupObject { object_id: 0, properties: props, status: WireStatus::Normal, payload: payload.to_vec() }, &mut media_stream)
            .map_err(|e| e.to_string())?;
        self.parent = Some(link);
        Ok(Signed { receipts_stream, media_stream, content_hash: hash, link })
    }
}

/// A [`Signed`] group handed to the page.
#[wasm_bindgen]
pub struct SignedGroup {
    receipts_stream: Vec<u8>,
    media_stream: Vec<u8>,
    content_hash: String,
}

#[wasm_bindgen]
impl SignedGroup {
    /// The receipts-track stream, moved out. It goes no later than the media stream.
    #[wasm_bindgen(js_name = receiptsStream)]
    pub fn receipts_stream(&mut self) -> Vec<u8> {
        std::mem::take(&mut self.receipts_stream)
    }
    /// The media stream, moved out.
    #[wasm_bindgen(js_name = mediaStream)]
    pub fn media_stream(&mut self) -> Vec<u8> {
        std::mem::take(&mut self.media_stream)
    }
    #[wasm_bindgen(getter, js_name = contentHash)]
    pub fn content_hash(&self) -> String {
        self.content_hash.clone()
    }
}

/// The publisher's signer of one media track, for a page.
#[wasm_bindgen]
pub struct TrackSigner {
    inner: Signer,
}

#[wasm_bindgen]
impl TrackSigner {
    /// `seed`: the publisher's 32-byte Ed25519 seed. `receipt_alias`: the media track's
    /// `wai_receipt_alias`. `media_alias` and `receipts_alias`: the Track Aliases this
    /// session's PUBLISH gave the two tracks. `object_timeout_ms`: an
    /// OBJECT_DELIVERY_TIMEOUT for each media object, 0 for none.
    #[wasm_bindgen(constructor)]
    pub fn new(seed: &[u8], receipt_alias: u64, media_alias: u64, receipts_alias: u64, priority: u8, origin: String, object_timeout_ms: u64) -> Result<TrackSigner, JsError> {
        let seed: [u8; 32] = seed.try_into().map_err(|_| JsError::new("a seed is 32 bytes"))?;
        Ok(TrackSigner { inner: Signer::new(seed, receipt_alias, media_alias, receipts_alias, priority, origin, (object_timeout_ms > 0).then_some(object_timeout_ms)) })
    }

    #[wasm_bindgen(js_name = publicKey)]
    pub fn public_key(&self) -> Vec<u8> {
        self.inner.public_key().to_vec()
    }

    /// `false`: the track's objects are not WAI envelopes (a world track's op batches, WAI
    /// live-worlds §4.2), and carry no `WAI_ENVELOPE` flag.
    #[wasm_bindgen(js_name = setEnvelope)]
    pub fn set_envelope(&mut self, envelope: bool) {
        self.inner.set_envelope(envelope);
    }

    /// Re-alias after a new session (a GOAWAY): the chain continues.
    #[wasm_bindgen(js_name = setAliases)]
    pub fn set_aliases(&mut self, media_alias: u64, receipts_alias: u64) {
        self.inner.media_alias = media_alias;
        self.inner.receipts_alias = receipts_alias;
    }

    pub fn sign(&mut self, group: u64, payload: &[u8], started: &str, ended: &str) -> Result<SignedGroup, JsError> {
        let s = self.inner.sign(group, payload, started, ended).map_err(js)?;
        Ok(SignedGroup { receipts_stream: s.receipts_stream, media_stream: s.media_stream, content_hash: hex::encode(s.content_hash) })
    }

    /// [`Signer::sign_sealed`] for a page: `properties` as a sealed object's
    /// [`e2ee::SealedOut::properties`] gives them.
    #[cfg(feature = "e2ee")]
    #[wasm_bindgen(js_name = signSealed)]
    pub fn sign_sealed(&mut self, group: u64, payload: &[u8], properties: &[u8], started: &str, ended: &str) -> Result<SignedGroup, JsError> {
        let props = Properties::decode(properties, "properties").map_err(|e| JsError::new(&wire_error(&e)))?;
        let s = self.inner.sign_sealed(group, payload, props, started, ended).map_err(js)?;
        Ok(SignedGroup { receipts_stream: s.receipts_stream, media_stream: s.media_stream, content_hash: hex::encode(s.content_hash) })
    }
}

/// The public key of a 32-byte Ed25519 seed.
#[wasm_bindgen(js_name = publicKeyOf)]
pub fn public_key_of(seed: &[u8]) -> Result<Vec<u8>, JsError> {
    let seed: [u8; 32] = seed.try_into().map_err(|_| JsError::new("a seed is 32 bytes"))?;
    Ok(SigningKey::from_bytes(&seed).verifying_key().to_bytes().to_vec())
}

// ---------------------------------------------------------------------------------------
// Receipts: the subscriber
// ---------------------------------------------------------------------------------------

/// What the subscriber holds of one group of the media track once its receipts arrived
/// and verified: the verified root and the object receipts.
struct HeldGroup {
    verified: VerifiedGroup,
    objects: Vec<ObjectReceipt>,
    /// The group receipt's link hash (WAI jwp-receipts §3 step 3): what a frame decoded from
    /// one of its objects names as its `receipt` (WAI browser-sink §7).
    link: [u8; 32],
}

/// The subscriber's side of one media track and its receipts track: each receipts-track
/// object is admitted to the track's chain (WAI jwp-receipts §3 step 3, as a subscriber
/// that joins: the chain is anchored at the first group it receives) after its signature
/// verifies under the publisher's key, and each media object is verified against the
/// receipt of its position.
pub struct Verifier {
    track: ReceiptsTrack,
    verifier: ReceiptVerifier,
    chains: ReceiptChains,
    groups: BTreeMap<u64, HeldGroup>,
    admissions: HashMap<&'static str, u64>,
    refusals: HashMap<String, u64>,
    /// The size of each receipts-track object received, in the binary form, and of the
    /// same receipts in the JSON form this crate's serialisation gives them (what the
    /// receipts track carried before the binary form): a measurement, nothing more.
    sizes: Sizes,
    json_sizes: Sizes,
}

/// A count, the least, the most and the sum of some sizes in bytes.
#[derive(Default)]
struct Sizes {
    n: u64,
    min: u64,
    max: u64,
    sum: u64,
}

impl Sizes {
    fn add(&mut self, b: u64) {
        self.min = if self.n == 0 { b } else { self.min.min(b) };
        self.max = self.max.max(b);
        self.n += 1;
        self.sum += b;
    }

    fn json(&self) -> Value {
        json!({ "objects": self.n, "min": self.min, "max": self.max, "sum": self.sum })
    }
}

/// The length of a bundle in this crate's JSON serialisation of its receipts, an array of
/// the group receipt and the object receipts.
fn json_form_len(b: &ReceiptBundle) -> u64 {
    let mut items = vec![serde_json::to_value(&b.group).unwrap_or(Value::Null)];
    items.extend(b.objects.iter().map(|o| serde_json::to_value(o).unwrap_or(Value::Null)));
    serde_json::to_vec(&items).map_or(0, |v| v.len() as u64)
}

fn admission_name(a: Admission) -> &'static str {
    match a {
        Admission::Linked => "linked",
        Admission::Anchor => "anchor",
        Admission::History => "history",
        Admission::AlreadyAdmitted => "already",
    }
}

fn refusal_name(r: ChainRefusal) -> &'static str {
    match r {
        ChainRefusal::Unverified => "unverified",
        ChainRefusal::MoqtMismatch => "moqt_mismatch",
        ChainRefusal::ParentNotHeld => "parent_not_held",
        ChainRefusal::ChainBreak => "chain_break",
        ChainRefusal::GroupIdNotIncreasing => "group_id_not_increasing",
        ChainRefusal::ParentNotRecovered => "parent_not_recovered",
    }
}

/// The bound this subscriber states on what it holds for missing parents (jwp-receipts
/// §3 step 3, *Holding*).
pub const HOLD: HoldBound = HoldBound { max_groups: 512, max_history: 4096, max_total: 8192, max_age_ms: Some(30_000) };

impl Verifier {
    pub fn new(publisher_key: [u8; 32], receipt_alias: u64) -> Result<Self, String> {
        let key = ed25519_dalek::VerifyingKey::from_bytes(&publisher_key).map_err(|e| format!("the publisher's key: {e}"))?;
        let chains = ReceiptChains::for_moq(HOLD).map_err(|b| format!("a hold bound of {}", b.count))?.with_join();
        Ok(Verifier {
            track: ReceiptsTrack { publisher_key, track_alias: TrackAlias(receipt_alias) },
            verifier: ReceiptVerifier::with_anchors(vec![key]),
            chains,
            groups: BTreeMap::new(),
            admissions: HashMap::new(),
            refusals: HashMap::new(),
            sizes: Sizes::default(),
            json_sizes: Sizes::default(),
        })
    }

    fn refuse(&mut self, why: &str) -> Value {
        *self.refusals.entry(why.to_owned()).or_default() += 1;
        json!({"ok": false, "reason": why})
    }

    /// Group `group` of the receipts track, its payload: the group receipt verified under
    /// the publisher's key (and nothing else), every receipt's coordinates checked against
    /// the MOQT group and Object ID it is for (`moqt_mismatch` otherwise), and the group
    /// admitted to its chain. Groups held behind it that its admission released are
    /// reported in `released`.
    pub fn receipts(&mut self, group: u64, payload: &[u8], now_ms: u64) -> Value {
        let bundle = match ReceiptBundle::from_payload(payload) {
            Ok(b) => b,
            Err(_) => return self.refuse("bundle_malformed"),
        };
        self.sizes.add(payload.len() as u64);
        self.json_sizes.add(json_form_len(&bundle));
        if bundle.objects.iter().enumerate().any(|(i, o)| o.check_moqt(self.track.track_alias, GroupId(group), ObjectId(i as u64)).is_err()) {
            return self.refuse("moqt_mismatch");
        }
        let verified = match self.verifier.verify_group(&bundle.group) {
            Ok(v) => v,
            Err(_) => return self.refuse("unverified"),
        };
        if verified.group_id != GroupId(group) || verified.track_alias != self.track.track_alias {
            return self.refuse("moqt_mismatch");
        }
        for o in &bundle.objects {
            if self.verifier.verify_object(o, &verified).is_err() {
                return self.refuse("object_receipt_invalid");
            }
        }
        let chain = match self.chains.admit_on(&self.track, GroupId(group), &bundle.group, now_ms) {
            Ok(a) => {
                *self.admissions.entry(admission_name(a)).or_default() += 1;
                json!(admission_name(a))
            }
            Err(r) => {
                *self.refusals.entry(format!("chain:{}", refusal_name(r))).or_default() += 1;
                json!(format!("held:{}", refusal_name(r)))
            }
        };
        let released: Vec<Value> = self
            .chains
            .take_events()
            .into_iter()
            .map(|e| {
                let outcome = match e.outcome {
                    Ok(a) => {
                        *self.admissions.entry(admission_name(a)).or_default() += 1;
                        admission_name(a).to_owned()
                    }
                    Err(r) => format!("refused:{}", refusal_name(r)),
                };
                json!({"group": e.group_id, "outcome": outcome})
            })
            .collect();
        let objects = bundle.objects.len();
        let link = bundle.group.link_hash();
        self.groups.insert(group, HeldGroup { verified, objects: bundle.objects, link });
        json!({"ok": true, "chain": chain, "released": released, "objects": objects})
    }

    /// The media object `object` of group `group`, its payload: `verified` when its hash is
    /// the receipt's and the receipt's leaf reaches the signed root; `no_receipt` while the
    /// group's receipts have not arrived; otherwise refused with the reason. A verified
    /// object's result carries its content hash and its group receipt's link hash, which a
    /// frame decoded from it names (WAI browser-sink §7.2).
    pub fn object(&mut self, group: u64, object: u64, payload: &[u8]) -> Value {
        let Some(h) = self.groups.get(&group) else { return json!({"ok": false, "reason": "no_receipt"}) };
        let Some(r) = h.objects.get(object as usize) else { return self.refuse("no_object_receipt") };
        match self.verifier.verify_payload(payload, r, &h.verified) {
            Ok(()) => json!({"ok": true, "content_hash": hex::encode(r.content_hash), "group_receipt": hex::encode(h.link)}),
            Err(_) => self.refuse("payload_mismatch"),
        }
    }

    /// An object that did not arrive, accounted for by inclusion (jwp-receipts §4.4): its
    /// receipt verifies against its group's signed root, without its payload.
    pub fn inclusion(&mut self, group: u64, object: u64) -> Value {
        let Some(h) = self.groups.get(&group) else { return json!({"ok": false, "reason": "no_receipt"}) };
        let Some(r) = h.objects.get(object as usize) else { return json!({"ok": false, "reason": "no_object_receipt"}) };
        match self.verifier.verify_object(r, &h.verified) {
            Ok(()) => json!({"ok": true, "content_hash": hex::encode(r.content_hash)}),
            Err(_) => self.refuse("inclusion_invalid"),
        }
    }

    pub fn has_receipts(&self, group: u64) -> bool {
        self.groups.contains_key(&group)
    }

    /// Forget groups below `group`: what a long-running page keeps is bounded.
    pub fn forget_below(&mut self, group: u64) {
        self.groups = self.groups.split_off(&group);
    }

    pub fn stats(&self) -> Value {
        json!({
            "groups_held": self.groups.len(),
            "admissions": self.admissions,
            "refusals": self.refusals,
            "chain_held": self.chains.held_total(),
            "anchors": self.chains.anchors(&self.track.publisher_key, self.track.track_alias).len(),
            "receipts_bytes": self.sizes.json(),
            "receipts_json_bytes": self.json_sizes.json(),
        })
    }
}

/// The subscriber's verifier of one media track, for a page.
#[wasm_bindgen]
pub struct TrackVerifier {
    inner: Verifier,
}

#[wasm_bindgen]
impl TrackVerifier {
    /// `publisher_key`: the 32-byte key the publisher signs with (the room principal's).
    /// `receipt_alias`: the media track's `wai_receipt_alias`, from its catalog entry.
    #[wasm_bindgen(constructor)]
    pub fn new(publisher_key: &[u8], receipt_alias: u64) -> Result<TrackVerifier, JsError> {
        let k: [u8; 32] = publisher_key.try_into().map_err(|_| JsError::new("a key is 32 bytes"))?;
        Ok(TrackVerifier { inner: Verifier::new(k, receipt_alias).map_err(js)? })
    }

    pub fn receipts(&mut self, group: u64, payload: &[u8], now_ms: u64) -> String {
        self.inner.receipts(group, payload, now_ms).to_string()
    }

    pub fn object(&mut self, group: u64, object: u64, payload: &[u8]) -> String {
        self.inner.object(group, object, payload).to_string()
    }

    pub fn inclusion(&mut self, group: u64, object: u64) -> String {
        self.inner.inclusion(group, object).to_string()
    }

    #[wasm_bindgen(js_name = hasReceipts)]
    pub fn has_receipts(&self, group: u64) -> bool {
        self.inner.has_receipts(group)
    }

    #[wasm_bindgen(js_name = forgetBelow)]
    pub fn forget_below(&mut self, group: u64) {
        self.inner.forget_below(group)
    }

    pub fn stats(&self) -> String {
        self.inner.stats().to_string()
    }
}

/// One track's running receipts, for a page (WAI live-http-egress §3.5): each `meta` event of
/// the track's owner that is a running receipt anchored to the track is held to §3.5's checks
/// and taken into its group's run (`joule_moq::running`), an object is held to what its group's
/// run states of it, and the group's receipts close the run and must agree with it. The event's
/// Location is its anchor's: a page reads a record through `msfRecordRead`, which holds the
/// timeline record's index to the anchor (live-events §8.1).
#[wasm_bindgen]
pub struct RunVerifier {
    room: [u8; 32],
    owner: [u8; 32],
    track: [u8; 32],
    alias: u64,
    runs: std::collections::BTreeMap<u64, joule_moq::running::Run>,
    closed: std::collections::BTreeSet<u64>,
    join: joule_moq::running::Join,
}

/// How many groups' runs a page keeps.
const RUNS_KEPT: usize = 64;

#[wasm_bindgen]
impl RunVerifier {
    /// `room` and `owner` as 64 hexadecimal digits; `namespace` and `name` the track's as its
    /// catalog entry has them (its anchor's digest is over them); `receipt_alias` its
    /// `wai_receipt_alias`.
    #[wasm_bindgen(constructor)]
    pub fn new(room: &str, owner: &str, namespace: &str, name: &str, receipt_alias: u64) -> Result<RunVerifier, JsError> {
        let d = |h: &str| -> Result<[u8; 32], JsError> { hex::decode(h).ok().and_then(|b| b.try_into().ok()).ok_or_else(|| JsError::new("64 hexadecimal digits")) };
        Ok(RunVerifier { room: d(room)?, owner: d(owner)?, track: joule_moq::running::anchor_track(namespace, name), alias: receipt_alias, runs: Default::default(), closed: Default::default(), join: Default::default() })
    }

    /// An event record of the owner's events track: `{"outcome", "group", "object"}`, the
    /// outcome `taken`, `repeat`, `ignored` (its group's receipts came, or its run broke),
    /// `joined` (of a group the page began reading inside, which takes no run, §3.5), one of
    /// §3.5's reasons, or `not_ours` for a record that is not a running receipt anchored to this
    /// track.
    pub fn record(&mut self, record: &[u8]) -> String {
        use joule_moq::running::{check, is_running, Place, Record, Taken, KIND_META};
        let Some(ev) = Record::read(record).filter(|e| e.kind == KIND_META && is_running(&e.payload)) else { return json!({ "outcome": "not_ours" }).to_string() };
        let Some(a) = ev.anchor.filter(|a| a.track == self.track) else { return json!({ "outcome": "not_ours" }).to_string() };
        let place = Place { room: &self.room, owner: &self.owner, track: &self.track, track_alias: self.alias, group: a.group, object: a.object };
        let outcome = match check(record, &place) {
            Err(why) => why,
            Ok(_) if self.closed.contains(&a.group) => "ignored",
            Ok(r) => {
                let joined = !self.runs.contains_key(&a.group) && self.join.inside(a.group, a.object);
                let run = self.runs.entry(a.group).or_default();
                let o = if joined {
                    run.abandon();
                    "joined"
                } else {
                    match run.take(a.object, r) {
                        Ok(Taken::Next) => "taken",
                        Ok(Taken::Repeat) => "repeat",
                        Ok(Taken::Ignored) => "ignored",
                        Err(why) => why,
                    }
                };
                while self.runs.len() > RUNS_KEPT {
                    self.runs.pop_first();
                }
                o
            }
        };
        json!({ "outcome": outcome, "group": a.group, "object": a.object }).to_string()
    }

    /// An object of the track: `{"ok": true}` where its group's run covers it and states its
    /// content hash; otherwise `{"ok": false, "reason"}`, `not_covered` (it waits for its
    /// running receipt or its group's receipts) or `content_hash_mismatch`.
    pub fn object(&self, group: u64, object: u64, payload: &[u8]) -> String {
        match self.runs.get(&group).and_then(|r| r.covers(object)) {
            None => json!({ "ok": false, "reason": "not_covered" }),
            Some(r) if r.content_hash != joule_moq::receipt::ObjectReceipt::hash_payload(payload) => json!({ "ok": false, "reason": "content_hash_mismatch" }),
            Some(_) => json!({ "ok": true }),
        }
        .to_string()
    }

    /// The group's receipts (the receipts track's object of the group), which close its run:
    /// `{"ok": true, "covered"}` where they agree with it (or it has none), `{"ok": false,
    /// "reason": "running_contradicted"}` where they do not, or `receipts_malformed` where they
    /// do not read. Their signature is the track verifier's to check.
    pub fn receipts(&mut self, group: u64, payload: &[u8]) -> String {
        self.closed.insert(group);
        while self.closed.len() > RUNS_KEPT {
            self.closed.pop_first();
        }
        let Ok(b) = joule_moq::carriage::ReceiptBundle::from_payload(payload) else { return json!({ "ok": false, "reason": "receipts_malformed" }).to_string() };
        let covered = self.runs.get(&group).map_or(0, |r| r.len());
        match self.runs.get(&group) {
            Some(run) if !run.agrees(&b.objects) => json!({ "ok": false, "reason": "running_contradicted", "covered": covered }),
            _ => json!({ "ok": true, "covered": covered }),
        }
        .to_string()
    }
}

/// The header of a stream that begins with `bytes`, when they hold it: what a page reads
/// to route a stream before reading its objects.
pub fn header_of(bytes: &[u8]) -> Option<StreamHeader> {
    let mut d = UniStreamDecoder::new(GroupOrder::Ascending, Limits::default());
    d.push(bytes);
    match d.next() {
        Ok(Some(UniEvent::Header(h))) => Some(h),
        _ => None,
    }
}

#[cfg(feature = "e2ee")]
pub mod e2ee;

#[cfg(test)]
mod tests;