//! The room's ads (`[ads]`, `src/ads.rs`) through the relay, on loopback.
//!
//! A room whose own source cues two breaks. Two sessions subscribe to their composed video
//! through the relay, each by its session key: one aggregated to a region an SCTE 224 document
//! blacks out over the second break, one to a region it does not. Each session's composed
//! Groups are held to the stitcher's group receipts, its creative records to the room's
//! receipts, its catalog to the decider's key, and the room's records are verified whole: the
//! room and its ad policy (live-ads §9), each session's creative records (§13.6) and composed
//! track (§14.4), and each audience decision against its document (audience-decision §10.7).
//!
//! Every wait is bounded, every listener binds port 0, and everything stops with its test.

use ed25519_dalek::SigningKey;
use joule_moq::carriage::ReceiptBundle;
use joule_moq::draft22::{Parameters, StreamHeader, UniEvent};
use joule_moq::receipt::ReceiptVerifier;
use joule_moq::{TrackName, TrackNamespace};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::transport::mux::MuxConfig;
use joule_moq_relay::transport::{websocket, SharedConnection};
use joule_moq_relay::wire::{ReqRx, ReqTx};
use joule_moq_room::ads::{self, Audit};
use joule_moq_room::dev::{DevRoom, Fixture};
use joule_moq_room::epoch::Role;
use joule_moq_room::source::Source;
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::time::timeout;
use wai::audience_decision::{scte224, Blackout, Verdict};
use wai::live_ads::creative;

const STEP: Duration = Duration::from_secs(20);

fn pack(p: &str) -> PathBuf {
    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../../../wai/pack-conformance").join(p)
}

/// The `[source]` and `[ads]` of the room: breaks at the source's video Groups 3 and 9, four
/// Groups each; two fixture creatives of two Groups each, their sound cut from the Opus rung of
/// a wai-pack package where their video's Groups end; the room's programme's rights in the
/// SCTE 224 document `rights`, read when each break is decided.
fn sections(rights: &Path, traffic: &str) -> String {
    format!(
        "[source]\npackage = {:?}\n[ads]\nfunds_ceiling = 10\ncurrency = \"EUR\"\nslots = 2\nfirst_break = 3\nevery = 6\nbreaks = 2\nbreak_groups = 4\n\
         asset_base = \"https://ads.example/ads/list/\"\ncreative_base = \"https://ads.example/ads/creative/\"\nprogramme = \"room\"\nscte224 = [{:?}]\nclaim_grace_ms = 1000\n{traffic}\
         [[ads.creative]]\npackage = {:?}\nad_id = \"fixture-a\"\ngroups = 2\nprice_micros = 300000\naudio_package = {sound:?}\n\
         [[ads.creative]]\npackage = {:?}\nfrom_group = 2\nad_id = \"fixture-b\"\ngroups = 2\nprice_micros = 200000\naudio_package = {sound:?}\n",
        pack("multi/multi-opus-01/package"),
        rights,
        pack("packs/video-cmaf-01/package"),
        pack("packs/video-cmaf-01/package"),
        sound = pack("multi/multi-opus-01/package"),
    )
}

pub fn now_us() -> u64 {
    joule_moq_room::now_us()
}

/// `GET` or `POST` over a connection of its own: the status and the body.
async fn http(addr: SocketAddr, method: &str, target: &str, body: &[u8], ua: Option<&str>) -> (u16, Vec<u8>) {
    let mut s = timeout(Duration::from_secs(5), tokio::net::TcpStream::connect(addr)).await.expect("in time").expect("a connection");
    let ua = ua.map(|u| format!("User-Agent: {u}\r\n")).unwrap_or_default();
    s.write_all(format!("{method} {target} HTTP/1.1\r\nHost: {addr}\r\n{ua}Content-Length: {}\r\nConnection: close\r\n\r\n", body.len()).as_bytes()).await.expect("a request");
    s.write_all(body).await.expect("a body");
    let mut buf = Vec::new();
    timeout(STEP, s.read_to_end(&mut buf)).await.expect("an answer in time").expect("a read");
    let end = buf.windows(4).position(|w| w == b"\r\n\r\n").expect("a head");
    let status = String::from_utf8_lossy(&buf[..end]).split(' ').nth(1).and_then(|c| c.parse().ok()).expect("a status");
    let head = String::from_utf8_lossy(&buf[..end]).to_ascii_lowercase();
    let body = buf[end + 4..].to_vec();
    // A chunked body, as hyper writes one without a length.
    let body = if head.contains("transfer-encoding: chunked") { dechunk(&body) } else { body };
    (status, body)
}

fn dechunk(mut b: &[u8]) -> Vec<u8> {
    let mut out = Vec::new();
    loop {
        let Some(i) = b.windows(2).position(|w| w == b"\r\n") else { return out };
        let n = usize::from_str_radix(std::str::from_utf8(&b[..i]).expect("a size").trim(), 16).expect("hex");
        if n == 0 {
            return out;
        }
        out.extend_from_slice(&b[i + 2..i + 2 + n]);
        b = &b[i + 2 + n + 2..];
    }
}

async fn json_of(addr: SocketAddr, target: &str) -> Value {
    let (status, body) = http(addr, "GET", target, b"", None).await;
    assert_eq!(status, 200, "{target}: {}", String::from_utf8_lossy(&body));
    serde_json::from_slice(&body).expect("JSON")
}

/// A session on the relay under a viewer's grant the service issues, as `POST /grant` would,
/// its WebSocket upgrade carrying `user_agent`, as a browser's does.
async fn session_as(dev: &DevRoom, user_agent: Option<&'static str>) -> Client {
    let key = SigningKey::from_bytes(&{ use rand_core_06::RngCore as _; let mut secret = [0u8; 32]; rand_core_06::OsRng.fill_bytes(&mut secret); secret });
    let issued = dev.svc.state.lock().unwrap().epoch.grant(&dev.svc.keys, key.verifying_key().to_bytes(), Role::Viewer, joule_moq_room::now_s(), None).expect("a viewer's grant");
    let url = dev.relay_url.clone();
    let dial = || {
        let url = url.clone();
        async move {
            let c = timeout(STEP, websocket::connect_as(&url, MuxConfig::default(), joule_moq::draft::Draft::D22, user_agent)).await.map_err(|_| "no session".to_string())?.map_err(|e| e.to_string())?;
            Ok::<SharedConnection, String>(Arc::new(c))
        }
    };
    let opts = ClientOptions { grants: vec![(issued.record, key)], target: joule_moq_room::proof_target(&dev.relay_url), ..ClientOptions::default() };
    Client::connect_proving(dial, opts).await.expect("a viewer's session")
}

/// What a session received of its stitched tracks, by track: each Group's Objects in order.
#[derive(Default, Debug)]
pub struct Received {
    pub video: BTreeMap<u64, BTreeMap<u64, Vec<u8>>>,
    pub receipts: BTreeMap<u64, ReceiptBundle>,
    /// The composed sound and its receipts.
    pub sound: BTreeMap<u64, BTreeMap<u64, Vec<u8>>>,
    pub sound_receipts: BTreeMap<u64, ReceiptBundle>,
    pub records: Vec<String>,
    pub catalog: Option<Vec<u8>>,
}

/// A player: its session key, its session on the relay and its subscriptions to its stitched
/// tracks.
pub struct Player {
    pub key: SigningKey,
    pub client: Arc<Client>,
    held: Vec<(ReqTx, ReqRx)>,
    aliases: BTreeMap<u64, &'static str>,
}

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

/// A player of the room: it states its region to the service, then subscribes to its catalog,
/// its creative records, its composed video's receipts and its composed video.
async fn player(dev: &DevRoom, prefix: &[String], region: Option<&str>) -> Player {
    player_as(dev, prefix, region, None).await
}

/// [`player`], its session on the relay opened with `user_agent`.
async fn player_as(dev: &DevRoom, prefix: &[String], region: Option<&str>, user_agent: Option<&'static str>) -> Player {
    let key = SigningKey::from_bytes(&{ use rand_core_06::RngCore as _; let mut secret = [0u8; 32]; rand_core_06::OsRng.fill_bytes(&mut secret); secret });
    let pk = hex::encode(key.verifying_key().to_bytes());
    let (st, body) = http(dev.http, "POST", "/ads/session", json!({"player": pk, "region": region}).to_string().as_bytes(), None).await;
    assert_eq!(st, 200, "{}", String::from_utf8_lossy(&body));
    let client = Arc::new(session_as(dev, user_agent).await);
    let mut ns: Vec<&str> = prefix.iter().map(String::as_str).collect();
    ns.push(&pk);
    let ns = TrackNamespace::from_strs(ns);
    let mut held = Vec::new();
    let mut aliases = BTreeMap::new();
    for (name, what) in [("catalog", "catalog"), ("ads", "ads"), ("receipts", "receipts"), (pk.as_str(), "video")] {
        let (ok, tx, rx) = timeout(STEP, client.subscribe(ns.clone(), TrackName::from_str(name), Parameters::new())).await.expect("an answer in time").unwrap_or_else(|e| panic!("SUBSCRIBE {name}: {e}"));
        aliases.insert(ok.track_alias, what);
        held.push((tx, rx));
    }
    // The composed sound, where the room composes one: `[room, stitcher, audio, stitched, D, K]`.
    if dev.svc.ads.as_ref().is_some_and(|a| a.sound()) {
        let mut sound: Vec<&str> = prefix.iter().map(String::as_str).collect();
        sound[2] = "audio";
        sound.push(&pk);
        let sns = TrackNamespace::from_strs(sound);
        for (name, what) in [("receipts", "sound_receipts"), (pk.as_str(), "sound")] {
            let (ok, tx, rx) = timeout(STEP, client.subscribe(sns.clone(), TrackName::from_str(name), Parameters::new())).await.expect("an answer in time").unwrap_or_else(|e| panic!("SUBSCRIBE sound {name}: {e}"));
            aliases.insert(ok.track_alias, what);
            held.push((tx, rx));
        }
    }
    Player { key, client, held, aliases }
}

/// Read what `p` is sent until `until`.
pub async fn receive(p: &Player, until: Instant) -> Received {
    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<(&'static str, u64, u64, Vec<u8>)>();
    let readers = {
        let (client, aliases) = (p.client.clone(), p.aliases.clone());
        tokio::spawn(async move {
            while Instant::now() < until {
                let Some(mut r) = client.next_data(Duration::from_millis(200)).await else { continue };
                let tx = tx.clone();
                let aliases = aliases.clone();
                tokio::spawn(async move {
                    let Ok(StreamHeader::Subgroup(h)) = r.start().await else { return };
                    let Some(what) = aliases.get(&h.track_alias).copied() else { return };
                    while let Ok(Ok(Some(e))) = timeout(Duration::from_secs(30), r.next()).await {
                        if let UniEvent::SubgroupObject(o) = e {
                            let _ = tx.send((what, h.group_id, o.object_id, o.payload));
                        }
                    }
                });
            }
        })
    };
    let mut got = Received::default();
    loop {
        let left = until.saturating_duration_since(Instant::now());
        match timeout(left, rx.recv()).await {
            Ok(Some((what, g, o, payload))) => match what {
                "video" => {
                    got.video.entry(g).or_default().insert(o, payload);
                }
                "receipts" => {
                    got.receipts.insert(g, ReceiptBundle::from_payload(&payload).expect("a receipt bundle"));
                }
                "sound" => {
                    got.sound.entry(g).or_default().insert(o, payload);
                }
                "sound_receipts" => {
                    got.sound_receipts.insert(g, ReceiptBundle::from_payload(&payload).expect("a receipt bundle"));
                }
                "ads" => {
                    let v: Value = serde_json::from_slice(&payload).expect("an array of records");
                    got.records = v.as_array().expect("an array").iter().map(|r| r.to_string()).collect();
                }
                _ => got.catalog = Some(payload),
            },
            _ => break,
        }
    }
    readers.abort();
    got
}

/// The room's records, every fill sealed, and their audit with `origins`.
async fn audited(dev: &DevRoom, origins: &[Vec<u8>]) -> (Value, Audit) {
    let rec = json_of(dev.http, "/ads/records?seal=all").await;
    let a = ads::audit(&rec, origins).expect("the records read");
    (rec, a)
}

/// A creative record's bytes, as the relay carried it, written again in its canonical form: the
/// form the room's records keep.
fn canonical(record: &str) -> String {
    let v: Value = serde_json::from_str(record).expect("a record");
    creative::read(v.to_string().as_bytes()).expect("a record that reads");
    v.to_string()
}

/// The room, its source's key, and the stitched namespace's prefix, once the source publishes.
async fn publishing(dev: &DevRoom) -> ([u8; 32], [u8; 32], Vec<String>) {
    let src = dev.svc.source.clone().expect("a source");
    let ads = dev.svc.ads.clone().expect("ads");
    let until = Instant::now() + STEP;
    loop {
        if src.publishing().is_some()
            && let Some(r) = ads.room.lock().unwrap().as_ref()
        {
            let stitcher = Source::key(&dev.svc.keys.clock).verifying_key().to_bytes();
            return (r.room, stitcher, ads::stitched_namespace(&r.room, &stitcher, &r.traffic_digest));
        }
        assert!(Instant::now() < until, "the source did not start publishing");
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
}

/// Wait until the schedule holds `n` breaks: each break's digest, its room-clock `t` and its
/// duration.
async fn breaks(dev: &DevRoom, n: usize, within: Duration) -> Vec<([u8; 32], u64, u64)> {
    let until = Instant::now() + within;
    loop {
        let b: Vec<([u8; 32], u64, u64)> = dev.svc.ads.as_ref().unwrap().room.lock().unwrap().as_ref().map(|r| r.schedule.breaks.iter().filter_map(|b| Some((b.digest, b.t, b.duration_ms?))).collect()).unwrap_or_default();
        if b.len() >= n {
            return b;
        }
        assert!(Instant::now() < until, "{} breaks of {n} in time", b.len());
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
}

/// Each Group a session received held to the stitcher's group receipt for it: the receipt's
/// signature under the stitcher's key and every Object's payload in its place.
fn verify_composed(got: &Received, stitcher: &[u8; 32]) -> usize {
    verify_groups(&got.video, &got.receipts, stitcher)
}

/// [`verify_composed`] over any composed track and its receipts.
fn verify_groups(track: &BTreeMap<u64, BTreeMap<u64, Vec<u8>>>, receipts: &BTreeMap<u64, ReceiptBundle>, stitcher: &[u8; 32]) -> usize {
    let v = ReceiptVerifier::with_anchors(vec![ed25519_dalek::VerifyingKey::from_bytes(stitcher).expect("a key")]);
    let mut n = 0;
    for (g, objects) in track {
        let Some(bundle) = receipts.get(g) else { continue };
        let group = v.verify_group(&bundle.group).unwrap_or_else(|e| panic!("Group {g}: its receipt: {e}"));
        assert_eq!(bundle.objects.len(), objects.len(), "Group {g}: a receipt for each Object");
        for ((o, payload), r) in objects.iter().zip(&bundle.objects) {
            assert_eq!(r.object_id.0, *o);
            v.verify_payload(payload, r, &group).unwrap_or_else(|e| panic!("Group {g} Object {o}: {e}"));
        }
        n += 1;
    }
    n
}

pub fn sha(b: &[u8]) -> [u8; 32] {
    Sha256::digest(b).into()
}

/// One placed break and one refused by the region rule, end to end through the relay.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_room_stitches_its_breaks_for_each_session() {
    let dir = std::env::temp_dir().join(format!("wai-room-ads-{}", std::process::id()));
    std::fs::create_dir_all(&dir).expect("a directory");
    let rights = dir.join("rights.xml");
    let dev = DevRoom::start_with(&Fixture { extra: sections(&rights, ""), ..Fixture::default() }).await.expect("the room and its relay");
    let (room, stitcher, prefix) = publishing(&dev).await;
    // Two sessions: one in a region the rights black out over the second break, one in another.
    let blacked = player(&dev, &prefix, Some("US-CA")).await;
    let shown = player(&dev, &prefix, Some("FR")).await;
    // The rights arrive once the first break is cued: a blackout in US-CA from after the first
    // break's end, over the second's opening.
    let first = breaks(&dev, 1, STEP).await[0];
    let t0 = dev.svc.ads.as_ref().unwrap().room.lock().unwrap().as_ref().unwrap().t0_us;
    let from = t0 + first.1 + first.2 * 1000 + 1_000_000;
    let doc = scte224::export("room", &[scte224::Span { start: from, end: from + 600_000_000, blackouts: vec![Blackout::new("bo-ca", &["US-CA"], None)] }]).expect("a document");
    std::fs::write(&rights, doc).expect("the rights");
    // Both breaks, and the second's end, and a Group past it.
    let all = breaks(&dev, 2, Duration::from_secs(40)).await;
    let end_us = t0 + all[1].1 + all[1].2 * 1000;
    let until = Instant::now() + Duration::from_micros(end_us.saturating_sub(now_us())) + Duration::from_secs(3);
    let (a, b) = tokio::join!(receive(&blacked, until), receive(&shown, until));
    let (rec, audit) = audited(&dev, &[]).await;
    assert_eq!(audit.outcome, "accept", "{audit:?}");
    // Each session's composed Groups are the stitcher's, receipt by receipt; its sound's too.
    assert!(verify_composed(&a, &stitcher) >= 8 && verify_composed(&b, &stitcher) >= 8, "the composed Groups are held to their receipts");
    assert!(verify_groups(&a.sound, &a.sound_receipts, &stitcher) >= 8 && verify_groups(&b.sound, &b.sound_receipts, &stitcher) >= 8, "the composed sound's Groups are held to their receipts");
    for (p, got) in [(&blacked, &a), (&shown, &b)] {
        let k = hex::encode(p.pk());
        // What it received is what the records say the stitcher published.
        let published: BTreeMap<u64, Vec<String>> = rec["sessions"][&k]["composed"].as_array().expect("composed").iter().map(|g| (g["id"].as_u64().unwrap(), g["objects"].as_array().unwrap().iter().map(|o| o.as_str().unwrap().to_string()).collect())).collect();
        for (g, objects) in &got.video {
            if let Some(want) = published.get(g) {
                let have: Vec<String> = objects.values().map(|b| hex::encode(sha(b))).collect();
                assert_eq!(&have, want, "{k}: Group {g} as published");
            }
        }
        assert_eq!(audit.composed[&k].split(':').next(), Some("ok"), "{k}: its composed track: {}", audit.composed[&k]);
        assert_eq!(audit.composed_sound[&k], audit.composed[&k], "{k}: its composed sound, at the same placements");
        // Where the composed video Group carries a creative's Group, the sound Group of its ID
        // carries that creative's sound Group; elsewhere the programme's sound.
        let ids = |key: &str| -> BTreeMap<u64, Vec<String>> { rec["sessions"][&k][key].as_array().unwrap().iter().map(|g| (g["id"].as_u64().unwrap(), g["objects"].as_array().unwrap().iter().map(|o| o.as_str().unwrap().to_string()).collect())).collect() };
        let (video, sound) = (ids("composed"), ids("composed_sound"));
        // The sound is a track of its own, subscribed after the video, and begins at the
        // programme's next sound Group from then: on a loaded host that is a Group after the
        // video's first, and where that first Group was a break's, its sound was never
        // composed. Every Group from the sound's first on is held.
        let begun = sound.keys().next().copied().unwrap_or(u64::MAX);
        let mut carried = 0;
        for c in rec["creatives"].as_array().unwrap() {
            let groups = |key: &str| -> Vec<Vec<String>> { c[key].as_array().unwrap().iter().map(|g| g.as_array().unwrap().iter().map(|o| o.as_str().unwrap().to_string()).collect()).collect() };
            let (vg, sg) = (groups("groups"), groups("sound"));
            for (id, objects) in &video {
                if *id < begun {
                    continue;
                }
                if let Some(n) = vg.iter().position(|g| g == objects) {
                    assert_eq!(sound.get(id), Some(&sg[n]), "{k}: Group {id} carries the creative's video Group {n} and its sound");
                    carried += 1;
                }
            }
        }
        assert!(carried >= 2, "{k}: the creatives' Groups were carried, picture and sound");
        assert_eq!(audit.records[&k].split(':').next(), Some("ok"), "{k}: its records: {}", audit.records[&k]);
        // Its records as the relay carried them are the room's.
        let kept: Vec<String> = rec["sessions"][&k]["records"].as_array().unwrap().iter().map(|r| r.as_str().unwrap().to_string()).collect();
        assert_eq!(got.records.iter().map(|r| canonical(r)).collect::<Vec<_>>(), kept.iter().map(|r| canonical(r)).collect::<Vec<_>>(), "{k}: its creative records");
        // Its catalog names the decider's key for its records.
        let cat: Value = serde_json::from_slice(got.catalog.as_ref().expect("a catalog")).expect("JSON");
        let kid = hex::encode(ads::decider_key(&dev.svc.keys.clock).verifying_key().to_bytes());
        assert_eq!(cat["tracks"][2]["wai_creative_signing"]["keys"][0]["kid"].as_str(), Some(kid.as_str()));
        assert_eq!(cat["tracks"][0]["name"].as_str(), Some(k.as_str()));
        assert_eq!((cat["tracks"][3]["role"].as_str(), cat["tracks"][3]["wai_receipt_alias"].as_u64()), (Some("audio"), Some(ads::COMPOSED_SOUND_RECEIPT_ALIAS)), "the catalog names the composed sound");
    }
    // The first break is filled for both; the second for the session the rights do not black out.
    let session_of = |p: &Player| hex::encode(wai::live_ads::presentation::session_of(&room, &p.pk()));
    let fills = |p: &Player| rec["receipts"].as_array().unwrap().iter().filter(|r| r["session"].as_str() == Some(session_of(p).as_str())).count();
    assert_eq!((fills(&blacked), fills(&shown)), (1, 2), "the second break is refused by the region rule for one session alone");
    assert_eq!(audit.records[&hex::encode(blacked.pk())], "ok: 1 bound");
    assert_eq!(audit.records[&hex::encode(shown.pk())], "ok: 2 bound");
    // The decisions: each held to its document; withheld in US-CA, shown in FR.
    let mut d: Vec<(Verdict, bool)> = audit.decisions.iter().map(|(_, v, f)| (*v, *f)).collect();
    d.sort_by_key(|(v, _)| v.name());
    assert_eq!(d, vec![(Verdict::Show, true), (Verdict::Withhold, false)], "{:?}", audit.decisions);
    let _ = std::fs::remove_dir_all(&dir);
    drop((blacked.held, shown.held));
}

// ── The edges ───────────────────────────────────────────────────────────────

/// The seed files and keys of the two edges: the relay's and the HTTP egress gateway's.
pub struct Edges {
    pub relay_key: PathBuf,
    pub relay: [u8; 32],
    pub egress: SigningKey,
}

pub fn edges(dir: &Path) -> Edges {
    let relay_seed = [0x21u8; 32];
    let relay_key = dir.join("edge.key");
    std::fs::write(&relay_key, hex::encode(relay_seed)).expect("the edge's key");
    Edges { relay_key, relay: SigningKey::from_bytes(&relay_seed).verifying_key().to_bytes(), egress: SigningKey::from_bytes(&[0x22; 32]) }
}

/// `[ads.traffic]`: a cap of two sessions a device key, the two edges, an address range list of a
/// documentation range and a token list of a crawler that does not exist.
pub fn traffic(dir: &Path, e: &Edges) -> String {
    use wai::live_ads::traffic::{crawlers_bytes, ranges_bytes, Range};
    let ranges = dir.join("ranges.wal");
    std::fs::write(&ranges, ranges_bytes(&[Range { family: 4, prefix: 24, address: vec![192, 0, 2, 0] }])).expect("ranges");
    let crawlers = dir.join("crawlers.wul");
    std::fs::write(&crawlers, crawlers_bytes(&[b"ExampleCrawler".to_vec()])).expect("crawlers");
    format!(
        "[ads.traffic]\nsession_cap = 2\nedges = [{:?}, {:?}]\nranges = {:?}\nranges_source = \"a documentation range, 2026-11-11\"\ncrawlers = {:?}\ncrawlers_source = \"a crawler that does not exist, 2026-11-11\"\n",
        hex::encode(e.relay),
        hex::encode(e.egress.verifying_key().to_bytes()),
        ranges,
        crawlers
    )
}

/// The HTTP egress gateway of the room's source, writing its breaks, as the edge of asset-list
/// requests in front of the room service's decider.
pub async fn egress(dev: &DevRoom, room: [u8; 32], owner: [u8; 32], e: &Edges) -> joule_moq_http::Gateway {
    use joule_moq_http::{AdsEdge, Breaks, Config as GatewayConfig, Gateway};
    use wai::live_egress::Params;
    let src = dev.svc.source.clone().expect("a source");
    let longest = src.tracks.iter().map(|t| t.longest_group_ms()).max().unwrap_or(1000);
    let r = dev.svc.ads.as_ref().unwrap().room.lock().unwrap().as_ref().map(|r| (r.policy.clone(), r.t0_us, r.traffic_digest)).expect("a room with ads");
    let viewer = {
        let key = SigningKey::from_bytes(&{ use rand_core_06::RngCore as _; let mut secret = [0u8; 32]; rand_core_06::OsRng.fill_bytes(&mut secret); secret });
        let issued = dev.svc.state.lock().unwrap().epoch.grant(&dev.svc.keys, key.verifying_key().to_bytes(), Role::Viewer, joule_moq_room::now_s(), None).expect("a grant");
        (issued.record, key)
    };
    let cfg = GatewayConfig {
        relay_url: dev.relay_url.clone(),
        room,
        owner: hex::encode(owner),
        publisher: None,
        listen: "127.0.0.1:0".parse().unwrap(),
        params: Params { target_duration_s: longest.div_ceil(1000).max(1), part_target_ms: 250, window: 6, latency_target_ms: 1500 },
        grant: Some(viewer),
        record_trace: false,
        catalog_wait: Duration::from_secs(10),
        breaks: Some(Breaks { policy: r.0, authority: dev.svc.keys.authority.verifying_key().to_bytes(), t0_unix_us: r.1 }),
    };
    let until = Instant::now() + STEP;
    let g = loop {
        match Gateway::start(cfg.clone()).await {
            Ok(g) => break g,
            Err(why) => {
                assert!(Instant::now() < until, "the gateway: {why}");
                tokio::time::sleep(Duration::from_millis(100)).await;
            }
        }
    };
    g.serve_ads(AdsEdge { key: e.egress.clone(), policy: r.2, decider: format!("http://{}/ads/list/", dev.http) }).expect("the edge");
    g
}

/// A viewer over HTTP asks for a break's asset list through `at` (the gateway, or the decider
/// itself), presenting its session key: the asset list.
pub async fn asset_list(at: SocketAddr, path: &str, cue: &[u8; 32], player: &[u8; 32], ua: &str) -> Value {
    let id = wai::live_ads::break_id(cue);
    let (st, body) = http(at, "GET", &format!("{path}{id}.json?_HLS_interstitial_id={id}&player={}", hex::encode(player)), b"", Some(ua)).await;
    assert_eq!(st, 200, "the asset list: {}", String::from_utf8_lossy(&body));
    serde_json::from_slice(&body).expect("an HLS asset list")
}

/// Each edge writes the origin of each session it terminated, as it saw the connection, under
/// the room's traffic policy: the relay for each subscription to a composed track, the gateway
/// for each asset-list request; each reads and verifies as wai-rs reads origins.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_edges_write_each_sessions_origin() {
    use wai::live_ads::traffic::Origin;
    let dir = std::env::temp_dir().join(format!("wai-room-ads-edges-{}", std::process::id()));
    std::fs::create_dir_all(&dir).expect("a directory");
    let e = edges(&dir);
    let extra = sections(&dir.join("rights.xml"), &traffic(&dir, &e));
    let dev = DevRoom::start_with(&Fixture { extra, edge_key: Some(e.relay_key.clone()), ..Fixture::default() }).await.expect("the room and its relay");
    let (room, stitcher, prefix) = publishing(&dev).await;
    let digest = dev.svc.ads.as_ref().unwrap().room.lock().unwrap().as_ref().unwrap().traffic_digest;
    let a = player_as(&dev, &prefix, Some("FR"), Some("ExamplePlayer/1.0 (tv)")).await;
    let b = player(&dev, &prefix, None).await;
    let gw = egress(&dev, room, stitcher, &e).await;
    // The first break, and an HTTP viewer's asset list for it through the gateway.
    let first = breaks(&dev, 1, STEP).await[0];
    let c = SigningKey::from_bytes(&{ use rand_core_06::RngCore as _; let mut secret = [0u8; 32]; rand_core_06::OsRng.fill_bytes(&mut secret); secret });
    let list = asset_list(gw.addr, "/ads/", &first.0, &c.verifying_key().to_bytes(), "ExampleHttpPlayer/3.0").await;
    let creatives: Vec<String> = dev.svc.ads.as_ref().unwrap().creatives.iter().map(|x| hex::encode(x.hash)).collect();
    let assets = list["ASSETS"].as_array().expect("ASSETS");
    assert!(!assets.is_empty(), "the break is filled: {list}");
    for x in assets {
        let uri = x["URI"].as_str().expect("a URI");
        assert!(creatives.iter().any(|c| uri == format!("https://ads.example/ads/creative/{c}.m3u8")), "{uri}: a creative of the pod");
    }
    // The relay's origins: one for each session's composed track, as the relay saw it.
    let until = Instant::now() + STEP;
    let relay_origins = loop {
        let o = dev.relay.relay().edge().expect("the relay is an edge").origins(0, 100);
        if o.len() >= 2 || Instant::now() > until {
            break o;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    };
    assert_eq!(relay_origins.len(), 2, "one origin a session");
    let read = |b: &[u8]| {
        let o = Origin::parse(b).expect("an origin that reads");
        o.check_signature().expect("signed by its edge");
        o
    };
    for (p, ua) in [(&a, b"ExamplePlayer/1.0 (tv)".to_vec()), (&b, Vec::new())] {
        let o = relay_origins.iter().map(|(_, b)| read(b)).find(|o| o.player == p.pk()).expect("the session's origin");
        assert_eq!((o.policy, o.edge, o.family, o.address.clone(), o.user_agent), (digest, e.relay, 4, vec![127, 0, 0, 1], ua));
    }
    // The gateway's: the HTTP viewer's, with its user agent.
    let g = gw.ads_origins();
    assert_eq!(g.len(), 1);
    let o = read(&g[0]);
    assert_eq!((o.player, o.policy, o.edge, o.address.clone(), o.user_agent), (c.verifying_key().to_bytes(), digest, e.egress.verifying_key().to_bytes(), vec![127, 0, 0, 1], b"ExampleHttpPlayer/3.0".to_vec()));
    // A request that presents no session key is passed on to no one.
    let (st, _) = http(gw.addr, "GET", &format!("/ads/{}.json", wai::live_ads::break_id(&first.0)), b"", None).await;
    assert_eq!(st, 400);
    assert_eq!(gw.ads_origins().len(), 1);
    // A player's binding of its session, held to the room's policy and both signatures.
    let device = SigningKey::from_bytes(&[0x33; 32]);
    let binding = wai::live_ads::stitch::bind(&digest, &device, &a.key, None);
    let (st, body) = http(dev.http, "POST", "/ads/binding", &binding.to_bytes(), None).await;
    assert_eq!(st, 200, "{}", String::from_utf8_lossy(&body));
    let other = wai::live_ads::stitch::bind(&[9; 32], &device, &a.key, None);
    assert_eq!(http(dev.http, "POST", "/ads/binding", &other.to_bytes(), None).await.0, 400, "a binding under another policy");
    let _ = std::fs::remove_dir_all(&dir);
    drop((a.held, b.held));
}

// ── Impressions decided ─────────────────────────────────────────────────────

/// The presentation claim (ad-presentation §5) of a player that presented the whole of slot
/// `slot`'s creative, in view and audible, at normal speed: one span, every Object touched.
/// `frames` sets the claim's frame count, so two claims of one presentation can differ.
fn claim(room: &[u8; 32], cue: &[u8; 32], key: &SigningKey, slot: u8, creative: &wai::live_ads::presentation::Creative, frames: u32) -> Vec<u8> {
    use wai::live_ads::presentation::{self as pr, Claim, Span};
    let player = key.verifying_key().to_bytes();
    let d = u32::try_from(creative.duration_ms()).expect("a creative of u32 ms");
    let spans = vec![Span { start_ms: 0, length_ms: d, visible_bp: pr::ALL_PIXELS, flags: pr::span_flag::AUDIBLE }];
    Claim {
        room: *room,
        cue: *cue,
        session: pr::session_of(room, &player),
        slot,
        creative_hash: creative.digest(),
        player,
        geometry: pr::geometry::AD,
        frames,
        objects: creative.touched(&spans),
        spans,
        attestation: vec![],
        sig: [0; 64],
    }
    .sign(key)
    .to_bytes()
}

/// Each impression a session's creative record states (live-ads §13.5): its break, its slot and
/// its creative's hash.
fn impressions_of(record: &str, run_t0: u64, breaks: &[([u8; 32], u64, u64)]) -> Vec<([u8; 32], u8, [u8; 32])> {
    let rec = creative::read(record.as_bytes()).expect("a record");
    let creative::Index::Wallclock(t) = rec.index else { panic!("a record indexed by wallclock") };
    let cue = breaks.iter().find(|b| (run_t0 + b.1) / 1000 <= t && t < (run_t0 + b.1) / 1000 + b.2).expect("the record's break").0;
    let pod = &rec.pods.as_ref().expect("a pod")[0];
    pod.slots.iter().enumerate().map(|(i, s)| (cue, i as u8, s.identifiers.iter().find(|(k, _)| k == creative::SCHEME_CREATIVE).and_then(|(_, v)| hex::decode(v).ok()?.try_into().ok()).expect("a creative hash"))).collect()
}

/// A composed Group held to its receipt: every Object in its place under the receipt the
/// stitcher signed for the Group, and no Object the receipt does not hold.
fn held_to_receipt(objects: &BTreeMap<u64, Vec<u8>>, bundle: &ReceiptBundle, stitcher: &[u8; 32]) -> Result<(), String> {
    let v = ReceiptVerifier::with_anchors(vec![ed25519_dalek::VerifyingKey::from_bytes(stitcher).expect("a key")]);
    let group = v.verify_group(&bundle.group).map_err(|e| e.to_string())?;
    for (o, payload) in objects {
        let r = bundle.objects.iter().find(|r| r.object_id.0 == *o).ok_or_else(|| format!("Object {o} has no receipt"))?;
        v.verify_payload(payload, r, &group).map_err(|e| e.to_string())?;
    }
    Ok(())
}

/// The room's ads end to end: two breaks, one placed and presented, one refused by the region
/// rule; sessions over MoQ through the relay and over HTTP through the egress gateway, each with
/// its origin from the edge it came through and its device binding; claims signed by each
/// session key; and the room's records verified, every impression decided valid, invalid or
/// undecided by the invalid-traffic rules, the totals recomputing. Planted: a creative record for
/// a break that does not exist, a cue signed by another key, a stitched Object without its
/// receipt, claims from a session no edge saw (undecided, withheld), and one session presenting
/// an impression twice (invalid, excluded).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn impressions_are_decided_by_the_rules_end_to_end() {
    use wai::live_ads::presentation as pr;
    let dir = std::env::temp_dir().join(format!("wai-room-ads-e2e-{}", std::process::id()));
    std::fs::create_dir_all(&dir).expect("a directory");
    let e = edges(&dir);
    let rights = dir.join("rights.xml");
    let dev = DevRoom::start_with(&Fixture { extra: sections(&rights, &traffic(&dir, &e)), edge_key: Some(e.relay_key.clone()), ..Fixture::default() }).await.expect("the room and its relay");
    let (room, stitcher, prefix) = publishing(&dev).await;
    let ads = dev.svc.ads.clone().unwrap();
    let (digest, t0, generation) = ads.room.lock().unwrap().as_ref().map(|r| (r.traffic_digest, r.t0_us, r.generation)).unwrap();
    // A and B play their composed tracks through the relay; C and D are viewers over HTTP.
    let a = player_as(&dev, &prefix, Some("US-CA"), Some("ExamplePlayer/1.0 (tv)")).await;
    let b = player_as(&dev, &prefix, Some("FR"), Some("ExamplePlayer/1.0 (tv)")).await;
    let (c, d) = (SigningKey::from_bytes(&{ use rand_core_06::RngCore as _; let mut secret = [0u8; 32]; rand_core_06::OsRng.fill_bytes(&mut secret); secret }), SigningKey::from_bytes(&{ use rand_core_06::RngCore as _; let mut secret = [0u8; 32]; rand_core_06::OsRng.fill_bytes(&mut secret); secret }));
    for k in [&c, &d] {
        let body = json!({"player": hex::encode(k.verifying_key().to_bytes()), "region": "FR"}).to_string();
        assert_eq!(http(dev.http, "POST", "/ads/session", body.as_bytes(), None).await.0, 200);
    }
    let gw = egress(&dev, room, stitcher, &e).await;
    // Device bindings: A and B on one device, C and D on another; two sessions each, the cap.
    let (dev1, dev2) = (SigningKey::from_bytes(&[0x31; 32]), SigningKey::from_bytes(&[0x32; 32]));
    for (device, k) in [(&dev1, &a.key), (&dev1, &b.key), (&dev2, &c), (&dev2, &d)] {
        let binding = wai::live_ads::stitch::bind(&digest, device, k, None);
        assert_eq!(http(dev.http, "POST", "/ads/binding", &binding.to_bytes(), None).await.0, 200);
    }
    // A and B claim each impression their creative records state as soon as it is filled, as a
    // player claims what it presents: B presents the first break's first slot twice.
    let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let claimer = {
        let (http_addr, ak, bk) = (dev.http, a.key.clone(), b.key.clone());
        let ads = ads.clone();
        let stop = stop.clone();
        let creatives: BTreeMap<String, pr::Creative> = ads.creatives.iter().map(|x| (hex::encode(x.hash), x.presentation.clone())).collect();
        tokio::spawn(async move {
            // By session and record: two sessions given one fill hold one record, byte for byte.
            let mut claimed: Vec<([u8; 32], String)> = Vec::new();
            while !stop.load(std::sync::atomic::Ordering::Acquire) {
                let sched: Vec<([u8; 32], u64, u64)> = ads.room.lock().unwrap().as_ref().map(|r| r.schedule.breaks.iter().filter_map(|b| Some((b.digest, b.t, b.duration_ms?))).collect()).unwrap_or_default();
                let first_cue = sched.first().map(|b| b.0);
                for k in [&ak, &bk] {
                    // The session's records as the stitcher sent them on its `ads` track (the
                    // first test holds the two to be the same).
                    let records: Vec<String> = ads.room.lock().unwrap().as_ref().and_then(|r| r.sessions.get(&k.verifying_key().to_bytes()).map(|s| s.records.clone())).unwrap_or_default();
                    let pk = k.verifying_key().to_bytes();
                    for r in records {
                        if claimed.contains(&(pk, r.clone())) {
                            continue;
                        }
                        for (cue, slot, hash) in impressions_of(&r, t0, &sched) {
                            let cr = &creatives[&hex::encode(hash)];
                            assert_eq!(http(http_addr, "POST", "/ads/claim", &claim(&room, &cue, k, slot, cr, pr::NOT_EXPOSED), None).await.0, 200);
                            if k.verifying_key() == bk.verifying_key() && Some(cue) == first_cue && slot == 0 {
                                assert_eq!(http(http_addr, "POST", "/ads/claim", &claim(&room, &cue, k, slot, cr, 60), None).await.0, 200, "the same impression again");
                            }
                        }
                        claimed.push((pk, r));
                    }
                }
                tokio::time::sleep(Duration::from_millis(200)).await;
            }
        })
    };
    // The rights: a blackout in US-CA over the second break's opening, from after the first's end.
    let first = breaks(&dev, 1, STEP).await[0];
    let from = t0 + first.1 + first.2 * 1000 + 1_000_000;
    let doc = scte224::export("room", &[scte224::Span { start: from, end: from + 600_000_000, blackouts: vec![Blackout::new("bo-ca", &["US-CA"], None)] }]).expect("a document");
    std::fs::write(&rights, doc).expect("the rights");
    // Planted: a cue signed by another key, on the source's chain as the stitcher reads it. It is
    // refused, and places no break.
    let forged = {
        let src = dev.svc.source.as_ref().unwrap().chain.lock().unwrap().clone();
        let last = wai::live_events::Event::parse(src.last().expect("the source's chain")).expect("an event");
        let mut ev = wai::live_events::Event::sign(&SigningKey::from_bytes(&[0x66; 32]), room, last.seq + 1, last.digest(), last.t + 1, None, wai::live_events::kind::CUE, wai::live_events::Payload::Cue(wai::live_events::Cue { cue_type: wai::live_events::cue_type::BREAK, cue_id: 99, duration_ms: 4_000, scte: None }).encode());
        ev.principal = stitcher;
        ev.to_bytes()
    };
    assert_eq!(ads.offer_event(generation, &forged), Err("bad_signature".to_string()), "a cue signed by another key");
    // The HTTP viewers ask for the first break's asset list: C through the gateway, D of the
    // decider itself, where no edge sees it.
    let list_c = asset_list(gw.addr, "/ads/", &first.0, &c.verifying_key().to_bytes(), "ExampleHttpPlayer/3.0").await;
    let list_d = asset_list(dev.http, "/ads/list/", &first.0, &d.verifying_key().to_bytes(), "ExampleHttpPlayer/3.0").await;
    let creatives: BTreeMap<String, pr::Creative> = ads.creatives.iter().map(|x| (hex::encode(x.hash), x.presentation.clone())).collect();
    let of_list = |l: &Value| -> Vec<pr::Creative> { l["ASSETS"].as_array().unwrap().iter().map(|x| creatives[x["URI"].as_str().unwrap().rsplit('/').next().unwrap().trim_end_matches(".m3u8")].clone()).collect() };
    for (k, l) in [(&c, &list_c), (&d, &list_d)] {
        for (slot, cr) in of_list(l).iter().enumerate() {
            assert_eq!(http(dev.http, "POST", "/ads/claim", &claim(&room, &first.0, k, slot as u8, cr, pr::NOT_EXPOSED), None).await.0, 200);
        }
    }
    // A and B play through both breaks.
    let all = breaks(&dev, 2, Duration::from_secs(40)).await;
    let end_us = t0 + all[1].1 + all[1].2 * 1000;
    let until = Instant::now() + Duration::from_micros(end_us.saturating_sub(now_us())) + Duration::from_secs(2);
    let (got_a, got_b) = tokio::join!(receive(&a, until), receive(&b, until));
    stop.store(true, std::sync::atomic::Ordering::Release);
    claimer.await.expect("the claims");
    // The fills are sealed once their claims have had their time.
    tokio::time::sleep(Duration::from_millis(1500)).await;
    let mut origins: Vec<Vec<u8>> = dev.relay.relay().edge().expect("an edge").origins(0, 1000).into_iter().map(|(_, o)| o).collect();
    origins.extend(gw.ads_origins());
    let (rec, audit) = audited(&dev, &origins).await;
    assert_eq!(audit.outcome, "accept", "{audit:?}");

    // Every composed Group A and B received is held to the stitcher's receipt for it.
    for got in [&got_a, &got_b] {
        for (g, objects) in &got.video {
            if let Some(bundle) = got.receipts.get(g) {
                held_to_receipt(objects, bundle, &stitcher).unwrap_or_else(|e| panic!("Group {g}: {e}"));
            }
        }
        assert_eq!(audit.composed[&hex::encode(if std::ptr::eq(got, &got_a) { a.pk() } else { b.pk() })].split(':').next(), Some("ok"), "{:?}", audit.composed);
    }
    // Planted: a stitched Object with no receipt of the stitcher's for it.
    let (g, objects) = got_a.video.iter().find(|(g, _)| got_a.receipts.contains_key(g)).expect("a Group with its receipt");
    let mut planted = objects.clone();
    planted.insert(objects.len() as u64, objects.values().next().unwrap().clone());
    assert_eq!(held_to_receipt(&planted, &got_a.receipts[g], &stitcher), Err(format!("Object {} has no receipt", objects.len())));
    // Planted: a creative record for a break the room does not have, signed by the decider.
    let decider = ads::decider_key(&dev.svc.keys.clock);
    let bogus = creative::write(creative::Index::Wallclock((t0 + all[0].1) / 1000 - 60_000), &creative::Pod { duration_ms: 1_000, slots: vec![creative::Slot { start_ms: 0, duration_ms: 1_000, identifiers: vec![] }] }, &decider);
    let mut planted_rec = rec.clone();
    let ka = hex::encode(a.pk());
    planted_rec["sessions"][&ka]["records"].as_array_mut().unwrap().insert(0, json!(bogus));
    let planted_audit = ads::audit(&planted_rec, &origins).expect("the records read");
    assert_eq!(planted_audit.records[&ka], "refused: record 0: creative_break_unknown");
    // Planted: the forged cue, given to a verifier on the source's chain, refuses the room.
    let mut forged_rec = rec.clone();
    forged_rec["chains"][1].as_array_mut().unwrap().push(json!(hex::encode(&forged)));
    let forged_audit = ads::audit(&forged_rec, &origins).expect("the records read");
    assert!(forged_audit.outcome.starts_with("refuse: bad_signature"), "{}", forged_audit.outcome);

    // The region rule: A is filled for the first break alone; B for both.
    let sess = |k: &SigningKey| hex::encode(pr::session_of(&room, &k.verifying_key().to_bytes()));
    let receipts = rec["receipts"].as_array().unwrap();
    let fills = |k: &SigningKey| receipts.iter().filter(|r| r["session"].as_str() == Some(sess(k).as_str())).count();
    assert_eq!((fills(&a.key), fills(&b.key), fills(&c), fills(&d)), (1, 2, 1, 1));
    assert!(audit.decisions.iter().any(|(_, v, f)| *v == Verdict::Withhold && !f) && audit.decisions.iter().any(|(_, v, f)| *v == Verdict::Show && *f));
    // Each impression decided: the session presenting twice invalid, the one no edge saw
    // undecided, the rest valid; and the totals recompute.
    let receipt_of = |i: usize| receipts[i]["session"].as_str().unwrap().to_string();
    let first_cue = hex::encode(first.0);
    for imp in &audit.impressions {
        let s = receipt_of(imp.receipt);
        let cue = receipts[imp.receipt]["cue"].as_str().unwrap();
        let want = if s == sess(&d) {
            ("undecided", vec!["origin_absent".to_string()])
        } else if s == sess(&b.key) && cue == first_cue && imp.slot == 0 {
            // Two claims of one impression, of different digests and the same media time.
            ("invalid", vec!["ivt_presented_twice".to_string(), "ivt_span_excess".to_string()])
        } else {
            ("valid", vec![])
        };
        if !imp.counts {
            let claims: Vec<String> = rec["claims"].as_array().unwrap().iter().map(|c| {
                let c = pr::Claim::parse(&hex::decode(c.as_str().unwrap()).unwrap()).unwrap();
                format!("{}/{}/{}", &hex::encode(c.session)[..6], &hex::encode(c.cue)[..6], c.slot)
            }).collect();
            let rs: Vec<String> = receipts.iter().map(|r| format!("{}/{}/{}", &r["session"].as_str().unwrap()[..6], &r["cue"].as_str().unwrap()[..6], r["impressions"].as_array().unwrap().len())).collect();
            panic!("every impression was presented and claimed: {imp:?}, session {s}, A {}, B {}, C {}, D {}, cue {cue}; claims {claims:?}; receipts {rs:?}", sess(&a.key), sess(&b.key), sess(&c), sess(&d));
        }
        assert_eq!((imp.verdict.as_str(), imp.codes.clone()), (want.0, want.1), "{imp:?}");
    }
    let t = &audit.tally;
    assert_eq!(t.valid + t.invalid + t.undecided, t.impressions);
    assert_eq!((t.invalid, t.excluded), (1, 1), "{t:?}");
    assert_eq!(t.undecided, receipts[receipts.iter().position(|r| r["session"].as_str() == Some(sess(&d).as_str())).unwrap()]["impressions"].as_array().unwrap().len() as u64);
    assert_eq!(audit.presented_micros, audit.valid_micros + audit.excluded_micros + audit.withheld_micros, "the presented spend is the valid, the excluded and the withheld");
    assert!(audit.valid_micros > 0 && audit.excluded_micros > 0 && audit.withheld_micros > 0, "{audit:?}");
    let _ = std::fs::remove_dir_all(&dir);
    drop((a.held, b.held));
}

// ── An injector's cues ──────────────────────────────────────────────────────

/// A session on the relay as `key`, under `grant`.
async fn session_of(dev: &DevRoom, grant: Vec<u8>, key: SigningKey) -> Client {
    let url = dev.relay_url.clone();
    let dial = || {
        let url = url.clone();
        async move {
            let c = timeout(STEP, websocket::connect_as(&url, MuxConfig::default(), joule_moq::draft::Draft::D22, None)).await.map_err(|_| "no session".to_string())?.map_err(|e| e.to_string())?;
            Ok::<SharedConnection, String>(Arc::new(c))
        }
    };
    let opts = ClientOptions { grants: vec![(grant, key)], target: joule_moq_room::proof_target(&dev.relay_url), ..ClientOptions::default() };
    Client::connect_proving(dial, opts).await.expect("a session")
}

/// An SCTE 104 `multiple_operation_message()` (SCTE 104 §8.2.3) of one `splice_request_data()`:
/// a `spliceStart_normal` of event `id`, `pre_roll_ms` ahead, `tenths` tenths of a second long,
/// with auto-return, timed by the arrival of its last byte (`time_type` 0).
fn splice_start(number: u8, id: u32, pre_roll_ms: u16, tenths: u16) -> Vec<u8> {
    let mut d = vec![1u8];
    d.extend_from_slice(&id.to_be_bytes());
    d.extend_from_slice(&[0, 0]);
    d.extend_from_slice(&pre_roll_ms.to_be_bytes());
    d.extend_from_slice(&tenths.to_be_bytes());
    d.extend_from_slice(&[1, 1, 1, 0]);
    let mut m = vec![0xff, 0xff, 0, 0, 0, 0, number, 0, 1, 0, 0, 1];
    m.extend_from_slice(&joule_moq_ingest::scte::scte104::op::SPLICE.to_be_bytes());
    m.extend_from_slice(&(d.len() as u16).to_be_bytes());
    m.extend_from_slice(&d);
    let n = m.len() as u16;
    m[2..4].copy_from_slice(&n.to_be_bytes());
    m
}

/// The records an injector's events track has sent, the aliases of its subscribers, and the last
/// Group ID it used.
type Sent = Arc<tokio::sync::Mutex<(Vec<String>, Vec<u64>, u64)>>;

/// One Group of an events track to the subscriber of `alias`: `records`, timeline records, as
/// its one Object.
async fn send_records(client: &Client, alias: u64, group: u64, records: &[String]) {
    use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject};
    let h = SubgroupHeader { track_alias: alias, group_id: group, subgroup_id: SubgroupId::Zero, publisher_priority: Some(32), properties: false, end_of_group: true, first_object: true };
    let _ = client.send_subgroup(h, &[SubgroupObject::normal(0, format!("[{}]", records.join(",")).into_bytes())]).await;
}

/// An injector's events track at the relay, published from its start (cue-injection §7.4): each
/// subscriber is answered and given every record so far, and each push sends every record again,
/// in a new Group, to every subscriber.
struct InjectorTrack {
    client: Arc<Client>,
    sent: Sent,
    answering: tokio::task::JoinHandle<()>,
    _ns: (ReqTx, ReqRx),
}

impl InjectorTrack {
    /// Publish `key`'s events track in `room` on a session under `grant`, and wait until the clock
    /// records the injector joining, which it does once the track appears, with no media.
    async fn start(dev: &DevRoom, room: [u8; 32], grant: Vec<u8>, key: &SigningKey) -> InjectorTrack {
        use joule_moq::draft22::{ControlMessage, SubscribeOk};
        let principal = key.verifying_key().to_bytes();
        let client = Arc::new(session_of(dev, grant, key.clone()).await);
        let ns = TrackNamespace::from_strs([hex::encode(room).as_str(), hex::encode(principal).as_str(), "events"]);
        let _ns = client.publish_namespace(ns, Parameters::new()).await.expect("the injector's namespace");
        let sent: Sent = Arc::new(tokio::sync::Mutex::new((Vec::new(), Vec::new(), 0)));
        let answering = {
            let (client, sent) = (client.clone(), sent.clone());
            tokio::spawn(async move {
                let mut held = Vec::new();
                let mut alias = 7000u64;
                loop {
                    let Some((tx, rx, first)) = client.next_request(Duration::from_millis(500)).await else { continue };
                    let ControlMessage::Subscribe(_) = first else { continue };
                    alias += 1;
                    if tx.send(&ControlMessage::SubscribeOk(SubscribeOk { track_alias: alias, params: Parameters::new(), track_properties: joule_moq::draft22::kvp::Properties::new() })).await.is_err() {
                        continue;
                    }
                    held.push((tx, rx));
                    let mut s = sent.lock().await;
                    s.1.push(alias);
                    s.2 += 1;
                    if !s.0.is_empty() {
                        send_records(&client, alias, s.2, &s.0).await;
                    }
                }
            })
        };
        let until = Instant::now() + STEP;
        loop {
            let present = dev.svc.state.lock().unwrap().epoch.members.get(&principal).map(|m| m.state);
            if present == Some(joule_moq_room::epoch::MemberState::Present) {
                break;
            }
            assert!(Instant::now() < until, "the clock never recorded the injector joining: {present:?}");
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
        InjectorTrack { client, sent, answering, _ns }
    }

    /// Send `records` after every record sent before, in a new Group to every subscriber.
    async fn push(&self, records: impl IntoIterator<Item = String>) {
        let mut st = self.sent.lock().await;
        st.0.extend(records);
        st.2 += 1;
        for a in st.1.clone() {
            send_records(&self.client, a, st.2, &st.0).await;
        }
    }
}

/// An injector's cues in a room with ads (cue-injection §7, live-ads §14.3): the injector takes
/// its grant from the room service and publishes events alone, so the clock records its `join`
/// when its events track appears; its automation system's splice instant falls inside a
/// programme Group, so the stitcher places the break from the next Group, the offset stated in
/// the session's receipt; and the room's records, the injector's chain among them, verify whole.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_injectors_break_is_placed_from_the_next_group() {
    use joule_moq_ingest::scte::{scte104, timeline_record, Engine};
    let injector = SigningKey::from_bytes(&[0x44; 32]);
    let ik = hex::encode(injector.verifying_key().to_bytes());
    // The source cues no break of its own here (its first would be its 100 000th Group).
    let extra = format!(
        "[source]\npackage = {:?}\n[ads]\nfunds_ceiling = 10\ncurrency = \"EUR\"\nslots = 2\nfirst_break = 100000\nbreak_groups = 4\n\
         asset_base = \"https://ads.example/ads/list/\"\ncreative_base = \"https://ads.example/ads/creative/\"\nclaim_grace_ms = 1000\ninjectors = [{ik:?}]\n\
         [[ads.creative]]\npackage = {:?}\nad_id = \"fixture-a\"\ngroups = 2\nprice_micros = 300000\n\
         [[ads.creative]]\npackage = {:?}\nfrom_group = 2\nad_id = \"fixture-b\"\ngroups = 2\nprice_micros = 200000\n",
        pack("multi/multi-opus-01/package"),
        pack("packs/video-cmaf-01/package"),
        pack("packs/video-cmaf-01/package"),
    );
    let dev = DevRoom::start_with(&Fixture { extra, ..Fixture::default() }).await.expect("the room and its relay");
    let (room, stitcher, prefix) = publishing(&dev).await;
    let ads = dev.svc.ads.clone().unwrap();
    let (t0, generation) = ads.room.lock().unwrap().as_ref().map(|r| (r.t0_us, r.generation)).unwrap();
    // Another key is no injector of this room.
    let (st, _) = http(dev.http, "POST", "/grant", json!({"principal": hex::encode(SigningKey::from_bytes(&[0x45; 32]).verifying_key().to_bytes()), "role": "injector"}).to_string().as_bytes(), None).await;
    assert_eq!(st, 403, "a key [ads] injectors does not name");
    let (st, body) = http(dev.http, "POST", "/grant", json!({"principal": ik, "role": "injector"}).to_string().as_bytes(), None).await;
    assert_eq!(st, 200, "{}", String::from_utf8_lossy(&body));
    let grant = hex::decode(serde_json::from_slice::<Value>(&body).unwrap()["grant"].as_str().unwrap()).unwrap();
    // The injector publishes its events track from its start (cue-injection §7.4), and the clock
    // records it joining once the track appears, with no media.
    let track = InjectorTrack::start(&dev, room, grant, &injector).await;
    // A session of the room, subscribed before the break.
    let p = player(&dev, &prefix, Some("FR")).await;
    // The programme's Groups so far, and their length: the splice instant 300 ms before a Group
    // boundary at least 5 s ahead, so that the automation system's pre-roll is SCTE 104's least
    // (§12.3) or more.
    let until = Instant::now() + STEP;
    let (last, every) = loop {
        let g: Vec<u64> = ads.room.lock().unwrap().as_ref().unwrap().programme.iter().map(|g| g.start_ms).collect();
        if g.len() >= 3 {
            break (g[g.len() - 1], g[g.len() - 1] - g[g.len() - 2]);
        }
        assert!(Instant::now() < until, "the programme's Groups");
        tokio::time::sleep(Duration::from_millis(100)).await;
    };
    let now_ms = (now_us() - t0) / 1000;
    let boundary = last + every * (now_ms + 5_000 - last).div_ceil(every);
    let splice_ms = boundary - 300;
    let params = scte104::Params { t0_us: t0, leap_s: 18, fps_num: 30, fps_den: 1 };
    let received = now_us() - t0;
    let pre_roll = u16::try_from(splice_ms - received / 1000).expect("a pre-roll in a u16");
    assert!(pre_roll >= scte104::MIN_PREROLL_MS, "a pre-roll of {pre_roll} ms");
    let msg = splice_start(1, 77, pre_roll, 80);
    let tr = scte104::translate(&msg, received, &params).expect("the message translates");
    let mut engine = Engine::new(room, injector.clone(), params, 256);
    assert_eq!(engine.accept(&msg, &tr, received).result, scte104::result::SUCCESS, "the request is taken");
    // The injector signs each cue when its splice instant comes (cue-injection §7.2), after its
    // injection record: the out at the splice, the derived in 8 s on.
    let mut cues = 0;
    let until = Instant::now() + Duration::from_secs(40);
    while cues < 2 {
        assert!(Instant::now() < until, "the injector's cues in time");
        let (signed, _) = engine.due(now_us() - t0);
        for s in signed {
            track.push([timeline_record(t0, s.t, &s.injection), timeline_record(t0, s.t, &s.cue)]).await;
            cues += 1;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    // The break, as the stitcher's schedule holds it: the injector's, opening at the splice.
    let all = breaks(&dev, 1, STEP).await;
    let (cue, t, duration) = all[0];
    assert_eq!((t / 1000, duration), (splice_ms, 8_000), "the injector's break, opening at its splice instant");
    // The session plays through the break.
    let end_us = t0 + t + duration * 1000;
    let until = Instant::now() + Duration::from_micros(end_us.saturating_sub(now_us())) + Duration::from_secs(3);
    let got = receive(&p, until).await;
    let (rec, audit) = audited(&dev, &[]).await;
    track.answering.abort();
    assert_eq!(audit.outcome, "accept", "the room's records, the injector's chain and its join among them: {audit:?}");
    assert_eq!(rec["chains"].as_array().unwrap().len(), 3, "the clock's, the source's and the injector's");
    // The session's fill: decided from the next Group, the offset stated.
    let session = hex::encode(wai::live_ads::presentation::session_of(&room, &p.pk()));
    let receipt = rec["receipts"].as_array().unwrap().iter().find(|r| r["session"].as_str() == Some(session.as_str()) && r["cue"].as_str() == Some(hex::encode(cue).as_str())).expect("the session's receipt for the break").clone();
    let programme: Vec<(u64, u64)> = rec["programme"].as_array().unwrap().iter().map(|g| (g["id"].as_u64().unwrap(), g["start_ms"].as_u64().unwrap())).collect();
    let (next_id, next_ms) = *programme.iter().find(|(_, s)| *s >= splice_ms).expect("the next Group");
    // The programme's Groups need not all be of one length: the splice falls inside a Group,
    // wherever the next boundary is.
    assert!(next_ms > splice_ms && next_ms - splice_ms < 2 * every, "the splice at {splice_ms} ms falls inside a Group; the next begins at {next_ms} ms (predicted {boundary} ms)");
    assert_eq!(receipt["start_offset_ms"].as_u64(), Some(next_ms - splice_ms), "the offset from the splice to the next Group: {receipt}");
    let k = hex::encode(p.pk());
    assert_eq!(audit.composed[&k], "ok: 1 placements, 1 stitched", "{:?}", audit.composed);
    // The Group at the boundary carries the fill's first creative's first Group, received whole and
    // held to the stitcher's receipt; the Group before it is the programme's.
    let first = &receipt["impressions"][0]["creative_hash"];
    let c = ads.creatives.iter().find(|c| Some(hex::encode(c.hash).as_str()) == first.as_str()).expect("a creative of the fill");
    let carried: Vec<[u8; 32]> = got.video.get(&next_id).expect("the boundary's Group").values().map(|b| sha(b)).collect();
    assert_eq!(carried, c.listing().groups[0], "the creative's first Group at the next boundary");
    assert!(got.receipts.contains_key(&next_id) && held_to_receipt(&got.video[&next_id], &got.receipts[&next_id], &stitcher).is_ok());
    let before: Vec<String> = rec["sessions"][&k]["composed"].as_array().unwrap().iter().find(|g| g["id"].as_u64() == Some(next_id - 1)).expect("the Group before")["objects"].as_array().unwrap().iter().map(|o| o.as_str().unwrap().to_string()).collect();
    let programme_before: Vec<String> = rec["programme"].as_array().unwrap().iter().find(|g| g["id"].as_u64() == Some(next_id - 1)).unwrap()["objects"].as_array().unwrap().iter().map(|o| o.as_str().unwrap().to_string()).collect();
    assert_eq!(before, programme_before, "the Group the splice falls in is the programme's");
    // Planted: an injector's record read on the source's track, and a record of a principal no
    // configuration names as a cue's author, are each refused, and place nothing.
    let injected = rec["chains"][2].as_array().unwrap()[0].as_str().map(|h| hex::decode(h).unwrap()).unwrap();
    assert_eq!(ads.offer_event_of(generation, stitcher, &injected), Err("wrong_principal".to_string()));
    assert_eq!(ads.offer_event_of(generation, [0x46; 32], &injected), Err("not_a_cue_author".to_string()));
    drop(p.held);
}

/// `records` in the order first read, each repeat left out: an events track sends its records
/// again (a Group's join set, or here every record with each Group), and a reader holds a repeat
/// as it read it (live-rooms §9.2 step 2).
fn distinct(records: &[Vec<u8>]) -> Vec<Vec<u8>> {
    let mut seen = std::collections::HashSet::new();
    records.iter().filter(|r| seen.insert(r.to_vec())).cloned().collect()
}

/// An injector's captions in a room with ads (essence-contribution §10.6, cue-injection §7.4): a
/// gateway's injector signs a contribution's captions in the chain that carries its cues, each
/// revision followed by its annotation and its window (timed-text §4.2, §4.7). The grant the room
/// service gives it permits exactly the kinds the texts have an injector sign, `meta`, `cue` and
/// `caption`, within its role and the room's ceiling (live-rooms §4.3); a caption's partial and
/// its final reach the stitcher's reader through the relay, which holds them in the injector's
/// chain; and the room's records verify whole (live-rooms §9.2), with that chain as the reader
/// held it and as the injector signed it. Under a grant of `meta` and `cue` alone the room
/// refused the caption `kind_not_permitted`, and so did the stitcher's reader; and with the
/// gateway's track indexing every record `"t"`, the reader refused each of the captions'
/// records `binding_mismatch` (live-events §8.1 indexes an anchored event `"l"`) and took none.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_injectors_captions_are_admitted_by_its_grant() {
    use joule_moq_ingest::scte::{scte104, timeline_record, Engine};
    use joule_moq_ingest::st2110::captions;
    use wai::live_events::{anchor_track, kind};
    let injector = SigningKey::from_bytes(&[0x47; 32]);
    let principal = injector.verifying_key().to_bytes();
    let ik = hex::encode(principal);
    let extra = format!(
        "[source]\npackage = {:?}\n[ads]\nfunds_ceiling = 10\ncurrency = \"EUR\"\nslots = 2\nfirst_break = 100000\nbreak_groups = 4\n\
         asset_base = \"https://ads.example/ads/list/\"\ncreative_base = \"https://ads.example/ads/creative/\"\nclaim_grace_ms = 1000\ninjectors = [{ik:?}]\n\
         [[ads.creative]]\npackage = {:?}\nad_id = \"fixture-a\"\ngroups = 2\nprice_micros = 300000\n",
        pack("multi/multi-opus-01/package"),
        pack("packs/video-cmaf-01/package"),
    );
    let dev = DevRoom::start_with(&Fixture { extra, ..Fixture::default() }).await.expect("the room and its relay");
    let (room, stitcher, _) = publishing(&dev).await;
    let ads = dev.svc.ads.clone().unwrap();
    let t0 = ads.room.lock().unwrap().as_ref().map(|r| r.t0_us).unwrap();
    let (st, body) = http(dev.http, "POST", "/grant", json!({"principal": ik, "role": "injector"}).to_string().as_bytes(), None).await;
    assert_eq!(st, 200, "{}", String::from_utf8_lossy(&body));
    let grant = hex::decode(serde_json::from_slice::<Value>(&body).unwrap()["grant"].as_str().unwrap()).unwrap();
    let track = InjectorTrack::start(&dev, room, grant.clone(), &injector).await;
    // Each caption is anchored to a programme video object, as a contribution's is to the video
    // object of its frame (essence-contribution §10.6).
    let until = Instant::now() + STEP;
    let group = loop {
        if let Some(g) = ads.room.lock().unwrap().as_ref().and_then(|r| r.programme.first().map(|g| g.id)) {
            break g;
        }
        assert!(Instant::now() < until, "the programme's first Group");
        tokio::time::sleep(Duration::from_millis(100)).await;
    };
    let anchor = Some((anchor_track(&format!("{}-{}-video", hex::encode(room), hex::encode(stitcher)), "0"), group, 0));
    // The injector's chain, signed as the gateway's injector signs a caption (`Injector::caption`)
    // and each record indexed as its events track writes it (`timeline_record`): a partial when
    // the caption is shown, its final when it leaves, each with its annotation and its window,
    // CTA-608 data channel 1.
    let mut engine = Engine::new(room, injector.clone(), scte104::Params { t0_us: t0, leap_s: 18, fps_num: 30, fps_den: 1 }, 256);
    let window = captions::window(1, 1, 0);
    let mut chain: Vec<Vec<u8>> = Vec::new();
    let mut partial = [0u8; 32];
    for final_ in [false, true] {
        let caption = captions::caption_payload(u32::from(final_), 1, None, "en", "HELLO");
        let annotation = captions::annotation(partial, final_, 5, 0, &[]);
        let (signed, t) = engine.sign_now(now_us() - t0, &[(anchor, kind::CAPTION, &caption), (anchor, kind::META, &annotation), (anchor, kind::META, &window)]);
        partial = signed[0].1;
        track.push(signed.iter().map(|(b, _)| timeline_record(t0, t, b))).await;
        chain.extend(signed.into_iter().map(|(b, _)| b));
        tokio::time::sleep(Duration::from_millis(300)).await;
    }
    // The stitcher's reader takes each record of the injector's track as the relay carries it,
    // each Group's repeats of the records before it among them.
    let until = Instant::now() + STEP;
    let (held, refused) = loop {
        let (held, refused) = {
            let g = ads.room.lock().unwrap();
            let r = g.as_ref().unwrap();
            let refused: Vec<(Vec<u8>, String)> = r.refused.iter().filter(|(b, _)| wai::live_events::Event::parse(b).is_ok_and(|e| e.principal == principal)).cloned().collect();
            (distinct(r.injected.get(&principal).map_or(&[][..], Vec::as_slice)), refused)
        };
        let taken = held.len() + distinct(&refused.iter().map(|(b, _)| b.clone()).collect::<Vec<_>>()).len();
        if taken >= chain.len() {
            break (held, refused.into_iter().map(|(_, why)| why).collect::<Vec<_>>());
        }
        assert!(Instant::now() < until, "the stitcher's reader took {taken} of the injector's {} records", chain.len());
        tokio::time::sleep(Duration::from_millis(100)).await;
    };
    let (rec, audit) = audited(&dev, &[]).await;
    track.answering.abort();
    // The room holds the injector's chain as it signed it: the clock's chain, the source's and the
    // injector's, each record under its principal's grant (live-rooms §9.2 step 4).
    let mut signed = rec.clone();
    let chains = signed["chains"].as_array_mut().unwrap();
    chains.truncate(2);
    chains.push(json!(chain.iter().map(hex::encode).collect::<Vec<_>>()));
    let as_signed = ads::audit(&signed, &[]).expect("the records read");
    assert_eq!(as_signed.outcome, "accept", "the room's records with the injector's chain as it signed it, its captions among them");
    assert!(refused.is_empty(), "the stitcher's reader refused the injector's records: {refused:?}");
    assert!(held == chain, "the stitcher's reader holds the injector's chain as it was signed: {} distinct records of {}", held.len(), chain.len());
    assert_eq!(audit.outcome, "accept", "the room's records as the service states them: {audit:?}");
    let stated = distinct(&rec["chains"][2].as_array().unwrap().iter().map(|h| hex::decode(h.as_str().unwrap()).unwrap()).collect::<Vec<_>>());
    assert!(stated == chain, "the injector's chain, its captions among them, in the room's records: {} distinct records of {}", stated.len(), chain.len());
    // The grant permits exactly the kinds an injector signs: its injection records and its cues
    // (cue-injection §7), and a contribution's captions with their annotations and windows
    // (essence-contribution §10.6). It is within its role and the room's ceiling (§4.3).
    let descriptor = wai::live_room::Descriptor::parse(&hex::decode(rec["descriptor"].as_str().unwrap()).unwrap()).expect("the descriptor");
    let checked = wai::live_room::check_grants(&descriptor, &room, &[grant]).map_err(|(_, r)| r.code()).expect("the injector's grant");
    let mut want = vec![kind::META, kind::CUE, kind::CAPTION];
    want.sort();
    let kinds: Vec<u8> = (0..=u8::MAX).filter(|k| checked[0].grant.policy.permits(*k)).collect();
    assert_eq!(kinds, want, "the injector's kinds: meta, cue and caption, and no other");
}

/// A contribution's captions in a room without ads (WAI D108): the service grants the role
/// injector to a key `[room] injectors` names, the same grant as in a room with ads (`meta`, `cue`
/// and `caption`, no media), within the participant's role, which carries those kinds, and the
/// room's ceiling (live-rooms §4.3); any other key is refused `not_an_injector`. The clock records
/// the injector present once its events track appears, and the room verifies whole (live-rooms
/// §9) with a caption's partial and its final in the injector's chain, each with its annotation
/// and its window, anchored to a video object of the gateway's contribution and signed as the
/// gateway's injector signs them (`Engine::sign_now`). The service before granted no injector in
/// a room without ads (`403`); with the participant's role left as it was, the grant is not
/// within it.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_injector_is_granted_in_a_room_without_ads() {
    use joule_moq_ingest::scte::{scte104, Engine};
    use joule_moq_ingest::st2110::captions;
    use wai::live_events::{anchor_track, kind};
    let injector = SigningKey::from_bytes(&[0x49; 32]);
    let principal = injector.verifying_key().to_bytes();
    let ik = hex::encode(principal);
    let extra = format!("[room]\nmax_publishers = 4\njoin_wait_s = 30\ninjectors = [{ik:?}]\n");
    let dev = DevRoom::start_with(&Fixture { extra, ..Fixture::default() }).await.expect("the room and its relay");
    assert!(dev.svc.ads.is_none() && dev.svc.source.is_none(), "a room without ads or a source");
    // A key the list does not name is refused.
    let other = hex::encode(SigningKey::from_bytes(&[0x4a; 32]).verifying_key().to_bytes());
    let (st, body) = http(dev.http, "POST", "/grant", json!({"principal": other, "role": "injector"}).to_string().as_bytes(), None).await;
    assert_eq!((st, serde_json::from_slice::<Value>(&body).unwrap()["error"].as_str()), (403, Some("not_an_injector")));
    let (st, body) = http(dev.http, "POST", "/grant", json!({"principal": ik, "role": "injector"}).to_string().as_bytes(), None).await;
    assert_eq!(st, 200, "{}", String::from_utf8_lossy(&body));
    let answer: Value = serde_json::from_slice(&body).unwrap();
    assert_eq!(answer["role"], "injector");
    let grant = hex::decode(answer["grant"].as_str().unwrap()).unwrap();
    let (room, t0, generation) = {
        let s = dev.svc.state.lock().unwrap();
        (s.epoch.id, s.epoch.t0_us, s.epoch.generation)
    };
    let track = InjectorTrack::start(&dev, room, grant.clone(), &injector).await;
    // The gateway's contribution, as a publisher of the room: its captions are anchored to its
    // frames' video objects (essence-contribution §10.6).
    let publisher = hex::encode(SigningKey::from_bytes(&[0x4b; 32]).verifying_key().to_bytes());
    let anchor = Some((anchor_track(&format!("{}-{publisher}-video", hex::encode(room)), "0"), 3, 0));
    let mut engine = Engine::new(room, injector.clone(), scte104::Params { t0_us: t0, leap_s: 18, fps_num: 30, fps_den: 1 }, 256);
    let window = captions::window(1, 1, 0);
    let mut chain: Vec<Vec<u8>> = Vec::new();
    let mut partial = [0u8; 32];
    for final_ in [false, true] {
        let caption = captions::caption_payload(u32::from(final_), 1, None, "en", "HELLO");
        let annotation = captions::annotation(partial, final_, 5, 0, &[]);
        let (signed, _) = engine.sign_now(now_us() - t0, &[(anchor, kind::CAPTION, &caption), (anchor, kind::META, &annotation), (anchor, kind::META, &window)]);
        partial = signed[0].1;
        chain.extend(signed.into_iter().map(|(b, _)| b));
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    // The room as `GET /room` states it, with the clock's chain and the injector's.
    let view = json_of(dev.http, "/room").await;
    let descriptor = hex::decode(view["descriptor"].as_str().unwrap()).unwrap();
    let grants: Vec<Vec<u8>> = view["grants"].as_array().unwrap().iter().map(|g| hex::decode(g.as_str().unwrap()).unwrap()).collect();
    assert!(grants.contains(&grant), "the injector's grant is listed with the room's");
    let d = wai::live_room::Descriptor::parse(&descriptor).expect("the descriptor");
    let checked = wai::live_room::check_grants(&d, &room, std::slice::from_ref(&grant)).map_err(|(_, r)| r.code()).expect("the injector's grant, within its role and the room's ceiling");
    let mut want = vec![kind::META, kind::CUE, kind::CAPTION];
    want.sort();
    let kinds: Vec<u8> = (0..=u8::MAX).filter(|k| checked[0].grant.policy.permits(*k)).collect();
    assert_eq!(kinds, want, "the injector's kinds: meta, cue and caption, and no other");
    assert_eq!(checked[0].grant.caps.publish, 0, "no media");
    // While its events track is up: the clock signs its `join` just after it holds it present,
    // and its `leave` when the track ends.
    let until = Instant::now() + STEP;
    let run = loop {
        let clock = dev.svc.clock_records(generation).await;
        let run = wai::live_room::verify_room(&descriptor, &grants, None, &[clock, chain.clone()]);
        assert!(matches!(run.outcome, wai::live_room::RoomOutcome::Accept), "the room with the injector's captions: {:?}", run.outcome);
        if run.state.present.contains_key(&principal) {
            break run;
        }
        assert!(Instant::now() < until, "the clock never recorded the injector present: {:?}", run.state.present.keys().map(hex::encode).collect::<Vec<_>>());
        tokio::time::sleep(Duration::from_millis(100)).await;
    };
    assert_eq!(run.state.present.get(&principal), Some(&checked[0].grant.digest()), "present under its injector's grant");
    track.answering.abort();
    dev.stop().await;
}