//! What the gateway's end-to-end tests share: a relay and a gateway on loopback, a room
//! and its grants, a WebRTC client on str0m (the WHIP publisher and the WHEP player), an
//! HTTP client, synthetic frames, and a relay subscriber that verifies every receipt.
//!
//! Every address is loopback and every wait is bounded.

#![allow(dead_code)]

use base64::Engine;
use bytes::Bytes;
use ed25519_dalek::SigningKey;
use http_body_util::{BodyExt, Full};
use joule_moq::carriage::ReceiptBundle;
use joule_moq::draft22::data::FetchEntry;
use joule_moq::draft22::{ControlMessage, Parameters, StreamHeader, UniEvent};
use joule_moq::receipt::ReceiptVerifier;
use joule_moq::{GroupId, ObjectId, TrackName, TrackNamespace};
use joule_moq_ingest::catalog;
use joule_moq_ingest::frame::{self, Codec};
use joule_moq_ingest::room::{build, Descriptor, Grant};
use joule_moq_ingest::{sdp, GatewayConfig, GatewayServer};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::transport::quic;
use joule_moq_relay::wire::{ReqRx, ReqTx};
use joule_moq_relay::{RelayConfig, RelayServer};
use std::collections::{BTreeMap, HashMap};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use str0m::change::{SdpAnswer, SdpPendingOffer};
use str0m::media::{Direction, Frequency, MediaKind, MediaTime, Mid};
use str0m::net::{Protocol, Receive};
use str0m::{Candidate, Event, IceConnectionState, Input, Output, Rtc, RtcConfig};
use tokio::net::UdpSocket;
use tokio::sync::{mpsc, oneshot};
use tokio::time::timeout;

/// The bound on any one wait.
pub const STEP: Duration = Duration::from_secs(20);

pub fn now_us() -> u64 {
    SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as u64
}

// ── The room, the relay and the gateway ────────────────────────────────────────────

pub struct World {
    pub relay: RelayServer,
    pub gateway: GatewayServer,
    pub authority: SigningKey,
    pub descriptor: Descriptor,
    pub room: [u8; 32],
    pub publisher: SigningKey,
    pub gateway_key: SigningKey,
}

impl World {
    /// A relay whose door checks every receipt of the room's tracks under the gateway's
    /// key, and a gateway that publishes to it, with `f` adjusting the gateway's
    /// configuration.
    pub async fn start(f: impl FnOnce(&mut GatewayConfig)) -> World {
        let authority = SigningKey::from_bytes(&[0x51; 32]);
        let publisher = SigningKey::from_bytes(&[0x52; 32]);
        let gateway_key = SigningKey::from_bytes(&[0x53; 32]);
        let descriptor = build::room(&authority, now_us() - 60_000_000);
        let room = descriptor.room_id();
        let gk = hex::encode(gateway_key.verifying_key().to_bytes());
        let mut rc = RelayConfig::loopback();
        rc.receipts.door = vec![joule_moq_relay::config::DoorRule { prefix: vec![hex::encode(room)], trust: vec![gk], trust_field: None, hold_ms: 10_000 }];
        let relay = RelayServer::start(rc).await.expect("the relay starts");
        let mut gc = GatewayConfig::loopback(relay.quic_addr.expect("a UDP endpoint"));
        gc.relay.trust_der = vec![relay.identity.chain[0].to_vec()];
        gc.signing.key_seed = Some(gateway_key.to_bytes());
        gc.rooms = vec![joule_moq_ingest::config::RoomConfig { descriptor_path: None, revoked: Vec::new(), descriptor: Some(descriptor.to_bytes()) }];
        f(&mut gc);
        let gateway = GatewayServer::start(gc).await.expect("the gateway starts");
        World { relay, gateway, authority, descriptor, room, publisher, gateway_key }
    }

    /// A grant for `who` under role `role` (0 publisher, 1 viewer, 2 observer), good now.
    pub fn grant(&self, who: &SigningKey, role: u8) -> Grant {
        let s = now_us() / 1_000_000;
        build::grant(&self.descriptor, &self.authority, &who.verifying_key().to_bytes(), role, s - 60, s + 3600)
    }

    pub fn token(g: &Grant) -> String {
        base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(g.to_bytes())
    }

    /// `who`'s grant under `role`, with its key, for requests that prove possession.
    pub fn auth(&self, who: &SigningKey, role: u8) -> Auth {
        Auth::new(self.grant(who, role), who)
    }

    pub fn base(&self) -> String {
        format!("http://{}", self.gateway.addr)
    }

    pub fn owner(&self) -> String {
        hex::encode(self.publisher.verifying_key().to_bytes())
    }

    pub fn whep_url(&self) -> String {
        format!("{}/whep/{}/{}", self.base(), hex::encode(self.room), self.owner())
    }

    pub async fn stop(self) {
        self.gateway.shutdown().await;
        self.relay.shutdown().await;
    }
}

// ── HTTP ───────────────────────────────────────────────────────────────────────────

#[derive(Debug)]
pub struct Resp {
    pub status: u16,
    pub headers: HashMap<String, String>,
    pub body: Vec<u8>,
}

impl Resp {
    pub fn header(&self, k: &str) -> Option<&str> {
        self.headers.get(&k.to_ascii_lowercase()).map(String::as_str)
    }
    pub fn text(&self) -> String {
        String::from_utf8_lossy(&self.body).into_owned()
    }
}

/// One HTTP/1.1 request on a connection of its own.
pub async fn http(method: &str, url: &str, headers: &[(&str, &str)], body: &[u8]) -> Resp {
    let rest = url.strip_prefix("http://").expect("an http URL");
    let (host, path) = rest.split_at(rest.find('/').unwrap_or(rest.len()));
    let path = if path.is_empty() { "/" } else { path };
    let stream = timeout(STEP, tokio::net::TcpStream::connect(host)).await.expect("connect in time").expect("connect");
    let (mut sender, conn) = hyper::client::conn::http1::handshake(hyper_util::rt::TokioIo::new(stream)).await.expect("handshake");
    tokio::spawn(conn);
    let mut b = hyper::Request::builder().method(method).uri(path).header("host", host);
    for (k, v) in headers {
        b = b.header(*k, *v);
    }
    let req = b.body(Full::new(Bytes::copy_from_slice(body))).unwrap();
    let resp = timeout(STEP, sender.send_request(req)).await.expect("a response in time").expect("a response");
    let status = resp.status().as_u16();
    let headers = resp.headers().iter().map(|(k, v)| (k.as_str().to_ascii_lowercase(), v.to_str().unwrap_or("").to_string())).collect();
    let body = timeout(STEP, resp.into_body().collect()).await.expect("a body in time").expect("a body").to_bytes().to_vec();
    Resp { status, headers, body }
}

pub fn bearer(token: &str) -> String {
    format!("Bearer {token}")
}

/// A principal's grant and key, and the last challenge the gateway gave it: what a client
/// presents over HTTP (live-rooms §10.5.3).
pub struct Auth {
    pub grant: Grant,
    pub key: SigningKey,
    challenge: Mutex<Option<[u8; 32]>>,
}

/// `METHOD //authority/path` of a URL, without its query (live-rooms §10.5.1).
pub fn context_of(method: &str, url: &str) -> String {
    let rest = url.strip_prefix("http://").expect("an http URL");
    let (host, path) = rest.split_at(rest.find('/').unwrap_or(rest.len()));
    joule_moq::grant_proof::context(method, host, if path.is_empty() { "/" } else { path })
}

impl Auth {
    pub fn new(grant: Grant, key: &SigningKey) -> Auth {
        Auth { grant, key: key.clone(), challenge: Mutex::new(None) }
    }

    pub fn token(&self) -> String {
        World::token(&self.grant)
    }

    pub fn challenge(&self) -> Option<[u8; 32]> {
        *self.challenge.lock().unwrap()
    }

    /// A proof by `key` over `challenge` for `method` at `url`, in base64url.
    pub fn proof(&self, key: &SigningKey, challenge: [u8; 32], method: &str, url: &str) -> String {
        let p = joule_moq::grant_proof::GrantProof::sign(self.grant.digest(), key, challenge, &context_of(method, url)).expect("a proof");
        joule_moq::grant_proof::base64url(&p.to_bytes())
    }

    fn keep(&self, r: &Resp) {
        if let Some(c) = r.header("wai-room-challenge").and_then(joule_moq::grant_proof::from_base64url).and_then(|b| <[u8; 32]>::try_from(b).ok()) {
            *self.challenge.lock().unwrap() = Some(c);
        }
    }

    /// The request as given: the grant as the Bearer token and `proof`, if any. The
    /// response's challenge is kept.
    pub async fn send_raw(&self, method: &str, url: &str, headers: &[(&str, &str)], body: &[u8], proof: Option<String>) -> Resp {
        let auth = bearer(&self.token());
        let mut h: Vec<(&str, &str)> = headers.to_vec();
        h.push(("authorization", &auth));
        if let Some(p) = &proof {
            h.push(("wai-room-proof", p));
        }
        let r = http(method, url, &h, body).await;
        self.keep(&r);
        r
    }

    /// The request with a proof over the last challenge; and, where it is answered `401`
    /// with a fresh challenge, once more over that one, as a client does that has none yet.
    pub async fn send(&self, method: &str, url: &str, headers: &[(&str, &str)], body: &[u8]) -> Resp {
        let proof = self.challenge().map(|c| self.proof(&self.key, c, method, url));
        let r = self.send_raw(method, url, headers, body, proof).await;
        if r.status != 401 || r.header("wai-room-challenge").is_none() || !r.text().contains("pop_") {
            return r;
        }
        let c = self.challenge().expect("a challenge");
        let proof = self.proof(&self.key, c, method, url);
        self.send_raw(method, url, headers, body, Some(proof)).await
    }
}

// ── A WebRTC client on str0m ───────────────────────────────────────────────────────

#[derive(Debug, Clone)]
pub enum PeerEvent {
    Connected,
    Ice(IceConnectionState),
    Media { kind: MediaKind, data: Vec<u8>, rtp: u64 },
}

enum PeerCmd {
    Write { kind: MediaKind, codec: Codec, data: Vec<u8>, ticks: u64 },
    /// Start an ICE restart: the new offer, whose ICE attributes go in the PATCH.
    Restart { reply: oneshot::Sender<String> },
    /// Finish it with the answer fragment the gateway returned.
    Restarted { frag: String, reply: oneshot::Sender<Result<(), String>> },
    Close,
}

/// A WebRTC endpoint driven on its own task, on a loopback UDP socket.
pub struct Peer {
    cmd: mpsc::UnboundedSender<PeerCmd>,
    pub events: Arc<Mutex<Vec<PeerEvent>>>,
    pub offer: String,
    pub answer: String,
    pub location: String,
    pub etag: String,
}

fn client_config(codecs: &[Codec]) -> RtcConfig {
    let mut c = RtcConfig::new().clear_codecs().set_crypto_provider(Arc::new(str0m::crypto::from_feature_flags()));
    for codec in codecs {
        c = match codec {
            Codec::Avc => c.enable_h264(true),
            Codec::Vp8 => c.enable_vp8(true),
            Codec::Vp9 => c.enable_vp9(true),
            Codec::Av1 => c.enable_av1(true),
            Codec::Opus => c.enable_opus(true, false),
        };
    }
    c
}

/// An offer for one audio and one video section in `dir`, on a new endpoint; the
/// endpoint's run loop starts once its answer is applied.
pub struct Offering {
    rtc: Rtc,
    socket: UdpSocket,
    pending: SdpPendingOffer,
    /// The sections this end added: str0m raises no MediaAdded for its own.
    mids: HashMap<MediaKind, Mid>,
    pub offer: String,
}

pub async fn offer(codecs: &[Codec], dir: Direction, audio: bool, video: bool) -> Offering {
    let socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
    let local = socket.local_addr().unwrap();
    let mut rtc = client_config(codecs).build(Instant::now());
    rtc.add_local_candidate(Candidate::host(local, "udp").unwrap());
    let mut api = rtc.sdp_api();
    let mut mids = HashMap::new();
    if audio {
        mids.insert(MediaKind::Audio, api.add_media(MediaKind::Audio, dir, Some("stream".into()), Some("audio".into()), None));
    }
    if video {
        mids.insert(MediaKind::Video, api.add_media(MediaKind::Video, dir, Some("stream".into()), Some("video".into()), None));
    }
    let (o, pending) = api.apply().expect("an offer");
    Offering { rtc, socket, pending, mids, offer: o.to_sdp_string() }
}

impl Offering {
    /// Apply the gateway's answer and start the endpoint.
    pub fn start(mut self, resp: &Resp) -> Peer {
        let answer = resp.text();
        let a = SdpAnswer::from_sdp_string(&answer).expect("the answer parses");
        self.rtc.sdp_api().accept_answer(self.pending, a).expect("the answer applies");
        let (tx, rx) = mpsc::unbounded_channel();
        let events = Arc::new(Mutex::new(Vec::new()));
        tokio::spawn(run_peer(self.rtc, self.socket, rx, events.clone(), answer.clone(), self.mids));
        let location = resp.header("location").expect("a Location").to_string();
        let etag = resp.header("etag").expect("an ETag").trim_matches('"').to_string();
        Peer { cmd: tx, events, offer: self.offer, answer, location, etag }
    }
}

async fn run_peer(mut rtc: Rtc, socket: UdpSocket, mut rx: mpsc::UnboundedReceiver<PeerCmd>, events: Arc<Mutex<Vec<PeerEvent>>>, mut answer: String, mut mids: HashMap<MediaKind, Mid>) {
    let local = socket.local_addr().unwrap();
    let mut buf = vec![0u8; 2048];
    let mut restart: Option<SdpPendingOffer> = None;
    loop {
        let next = loop {
            match rtc.poll_output() {
                Ok(Output::Timeout(t)) => break t,
                Ok(Output::Transmit(t)) => {
                    let _ = socket.try_send_to(&t.contents, t.destination);
                }
                Ok(Output::Event(e)) => {
                    let mut ev = events.lock().unwrap();
                    match e {
                        Event::Connected => ev.push(PeerEvent::Connected),
                        Event::IceConnectionStateChange(s) => ev.push(PeerEvent::Ice(s)),
                        Event::MediaAdded(m) => {
                            mids.insert(m.kind, m.mid);
                        }
                        Event::MediaData(d) => {
                            let kind = if d.params.spec().codec.is_audio() { MediaKind::Audio } else { MediaKind::Video };
                            ev.push(PeerEvent::Media { kind, data: d.data.to_vec(), rtp: d.time.numer() });
                        }
                        _ => {}
                    }
                }
                Err(_) => return,
            }
        };
        let sleep = tokio::time::sleep_until(tokio::time::Instant::from_std(next.max(Instant::now())));
        tokio::select! {
            r = socket.recv_from(&mut buf) => {
                if let Ok((n, from)) = r
                    && let Ok(r) = Receive::new(Protocol::Udp, from, local, &buf[..n])
                {
                    let _ = rtc.handle_input(Input::Receive(Instant::now(), r));
                }
            }
            _ = sleep => {
                let _ = rtc.handle_input(Input::Timeout(Instant::now()));
            }
            c = rx.recv() => match c {
                None | Some(PeerCmd::Close) => {
                    rtc.disconnect();
                    let _ = rtc.poll_output();
                    return;
                }
                Some(PeerCmd::Write { kind, codec, data, ticks }) => {
                    let Some(&mid) = mids.get(&kind) else { continue };
                    let Some(pt) = rtc.writer(mid).and_then(|w| w.payload_params().find(|p| p.spec().codec == codec.to_str0m()).map(|p| p.pt())) else { continue };
                    let freq = if kind == MediaKind::Audio { Frequency::FORTY_EIGHT_KHZ } else { Frequency::NINETY_KHZ };
                    if let Some(w) = rtc.writer(mid) {
                        let _ = w.write(pt, Instant::now(), MediaTime::new(ticks, freq), data);
                    }
                }
                Some(PeerCmd::Restart { reply }) => {
                    let mut api = rtc.sdp_api();
                    api.ice_restart(true);
                    let (o, pending) = api.apply().expect("an ICE restart offer");
                    restart = Some(pending);
                    let _ = reply.send(o.to_sdp_string());
                }
                Some(PeerCmd::Restarted { frag, reply }) => {
                    let r = (|| {
                        let f = sdp::parse_frag(&frag).ok_or("the fragment reads")?;
                        let (u, p) = (f.ufrag.ok_or("an ice-ufrag")?, f.pwd.ok_or("an ice-pwd")?);
                        let a = sdp::restart_answer(&answer, &u, &p, &f.candidates);
                        let parsed = SdpAnswer::from_sdp_string(&a).map_err(|e| e.to_string())?;
                        let pending = restart.take().ok_or("a restart pending")?;
                        rtc.sdp_api().accept_answer(pending, parsed).map_err(|e| e.to_string())?;
                        answer = a;
                        Ok::<(), String>(())
                    })();
                    let _ = reply.send(r.map_err(|e| e.to_string()));
                }
            }
        }
    }
}

impl Peer {
    pub fn write(&self, kind: MediaKind, codec: Codec, data: Vec<u8>, ticks: u64) {
        let _ = self.cmd.send(PeerCmd::Write { kind, codec, data, ticks });
    }

    pub async fn restart_offer(&self) -> String {
        let (tx, rx) = oneshot::channel();
        self.cmd.send(PeerCmd::Restart { reply: tx }).unwrap();
        timeout(STEP, rx).await.expect("in time").expect("an offer")
    }

    pub async fn restarted(&self, frag: String) -> Result<(), String> {
        let (tx, rx) = oneshot::channel();
        self.cmd.send(PeerCmd::Restarted { frag, reply: tx }).unwrap();
        timeout(STEP, rx).await.expect("in time").expect("a result")
    }

    pub fn close(&self) {
        let _ = self.cmd.send(PeerCmd::Close);
    }

    pub fn count(&self, f: impl Fn(&PeerEvent) -> bool) -> usize {
        self.events.lock().unwrap().iter().filter(|e| f(e)).count()
    }

    pub async fn connected(&self) {
        assert!(until(STEP, || self.count(|e| matches!(e, PeerEvent::Connected)) > 0).await, "the WebRTC session connects");
    }

    pub fn media(&self, kind: MediaKind) -> Vec<Vec<u8>> {
        self.events.lock().unwrap().iter().filter_map(|e| if let PeerEvent::Media { kind: k, data, .. } = e { (*k == kind).then(|| data.clone()) } else { None }).collect()
    }
}

/// Wait, bounded, until `f` holds.
pub async fn until(limit: Duration, mut f: impl FnMut() -> bool) -> bool {
    let deadline = Instant::now() + limit;
    while !f() {
        if Instant::now() >= deadline {
            return false;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    true
}

/// POST a WHIP offer with `token` and start the publisher.
pub async fn publish(w: &World, auth: &Auth) -> Peer {
    publish_with(w, auth, &[Codec::Avc, Codec::Opus], true, true).await
}

/// As [`publish`], offering `codecs`, and the sections asked for.
pub async fn publish_with(w: &World, auth: &Auth, codecs: &[Codec], audio: bool, video: bool) -> Peer {
    let o = offer(codecs, Direction::SendOnly, audio, video).await;
    let resp = auth.send("POST", &format!("{}/whip", w.base()), &[("content-type", "application/sdp")], o.offer.as_bytes()).await;
    assert_eq!(resp.status, 201, "WHIP POST: {}", resp.text());
    assert_eq!(resp.header("content-type"), Some("application/sdp"));
    assert!(resp.header("wai-room-challenge").is_some(), "a challenge for the session's next request");
    let p = o.start(&resp);
    p.connected().await;
    p
}

// ── Synthetic media ────────────────────────────────────────────────────────────────

/// A body of `n` bytes, none zero, so no start code appears inside a NAL unit.
fn body(seed: u64, n: usize) -> Vec<u8> {
    (0..n).map(|j| ((seed.wrapping_mul(31).wrapping_add(j as u64 * 7)) % 255 + 1) as u8).collect()
}

/// H.264 access unit `i`: an IDR with its SPS and PPS when `key`, a non-IDR slice
/// otherwise, each NAL unit after a four-byte start code. Large enough to need FU-A.
pub fn avc(i: u64, key: bool) -> Vec<u8> {
    let mut f = Vec::new();
    if key {
        f.extend_from_slice(&[0, 0, 0, 1, 0x67, 0x42, 0xe0, 0x1f, 0x8c, 0x8d, 0x40]);
        f.extend_from_slice(&[0, 0, 0, 1, 0x68, 0xce, 0x3c, 0x80]);
        f.extend_from_slice(&[0, 0, 0, 1, 0x65]);
        f.extend_from_slice(&body(i, 3000));
    } else {
        f.extend_from_slice(&[0, 0, 0, 1, 0x41]);
        f.extend_from_slice(&body(i, 700 + (i % 5) as usize * 100));
    }
    f
}

/// Opus packet `i`: a fullband CELT 20 ms stereo TOC, then bytes.
pub fn opus(i: u64) -> Vec<u8> {
    let mut f = vec![0xfc];
    f.extend_from_slice(&body(i + 7_000, 80 + (i % 7) as usize));
    f
}

/// What a publisher sent, in order.
#[derive(Default, Clone)]
pub struct Sent {
    pub video: Vec<Vec<u8>>,
    pub audio: Vec<Vec<u8>>,
}

/// Send `seconds` of media in real time: 30 video frames a second, a keyframe every
/// `gop`, and 50 audio packets a second. Continues from what `sent` holds.
pub async fn send(p: &Peer, sent: &mut Sent, seconds: u64, gop: u64) {
    let start = Instant::now();
    let v0 = sent.video.len() as u64;
    let a0 = sent.audio.len() as u64;
    let (nv, na) = (30 * seconds, 50 * seconds);
    let (mut v, mut a) = (0, 0);
    while v < nv || a < na {
        let tv = Duration::from_micros(v * 33_333);
        let ta = Duration::from_micros(a * 20_000);
        let next = tv.min(ta);
        tokio::time::sleep_until(tokio::time::Instant::from_std(start + next)).await;
        if v < nv && tv <= ta {
            let i = v0 + v;
            let f = avc(i, i.is_multiple_of(gop));
            p.write(MediaKind::Video, Codec::Avc, f.clone(), 1_000 + i * 3000);
            sent.video.push(f);
            v += 1;
        } else {
            let i = a0 + a;
            let f = opus(i);
            p.write(MediaKind::Audio, Codec::Opus, f.clone(), 5_000 + i * 960);
            sent.audio.push(f);
            a += 1;
        }
    }
}

// ── A subscriber at the relay that verifies every receipt ─────────────────────────

pub async fn relay_client(w: &World) -> (Client, quinn::Endpoint) {
    let ep = quic::client_endpoint("127.0.0.1:0".parse().unwrap(), vec![w.relay.identity.chain[0].clone()], vec![joule_moq::draft22::ALPN.to_vec()], None).unwrap();
    let conn = timeout(STEP, quic::connect(&ep, w.relay.quic_addr.unwrap(), "localhost")).await.expect("in time").expect("a QUIC session");
    let c = Client::connect(Arc::new(conn), ClientOptions::default()).await.expect("SETUP");
    (c, ep)
}

/// One subscribed media track and its receipts track, as the subscriber reads them.
pub struct Watched {
    pub kind: u8,
    pub media_alias: u64,
    pub receipts_alias: u64,
    /// The request streams, held so the subscriptions stand.
    pub held: Vec<ReqTx>,
    pub receipts_rx: ReqRx,
}

/// What a subscriber read: each object by (Track Alias, group, object).
pub type Objects = Arc<Mutex<BTreeMap<(u64, u64, u64), Vec<u8>>>>;

pub struct Watcher {
    pub client: Arc<Client>,
    pub _ep: quinn::Endpoint,
    pub tracks: Vec<Watched>,
    pub objects: Objects,
    pub ended: Arc<Mutex<Vec<u64>>>,
}

/// Subscribe at the relay to the owner's audio and video tracks and their receipts
/// tracks, from their start, reading every stream concurrently.
pub async fn watch(w: &World) -> Watcher {
    watch_kinds(w, &[1, 2]).await
}

/// As [`watch`], for the media kinds `kinds` (1 audio, 2 video).
pub async fn watch_kinds(w: &World, kinds: &[u8]) -> Watcher {
    let (c, ep) = relay_client(w).await;
    let c = Arc::new(c);
    let owner = w.owner();
    let mut tracks = Vec::new();
    let ended = Arc::new(Mutex::new(Vec::new()));
    for &kind in kinds {
        let ns = catalog::media_namespace(&w.room, &owner, kind);
        let (m, mtx, mrx) = c.subscribe(ns.clone(), TrackName::from_str("0"), Parameters::new()).await.expect("SUBSCRIBE the media track");
        let (r, rtx, rrx) = c.subscribe(ns.clone(), catalog::receipts_name("0"), Parameters::new()).await.expect("SUBSCRIBE the receipts track");
        // PUBLISH_DONE on the media track: the track ended.
        let ended = ended.clone();
        let alias = m.track_alias;
        let mut mrx = mrx;
        tokio::spawn(async move {
            while let Ok(Some(msg)) = mrx.next().await {
                if let ControlMessage::PublishDone(_) = msg {
                    ended.lock().unwrap().push(alias);
                    return;
                }
            }
        });
        tracks.push(Watched { kind, media_alias: m.track_alias, receipts_alias: r.track_alias, held: vec![mtx, rtx], receipts_rx: rrx });
    }
    let objects: Objects = Arc::default();
    let (c2, o2) = (c.clone(), objects.clone());
    tokio::spawn(async move {
        loop {
            let Some(mut r) = c2.next_stream(Duration::from_secs(1)).await else {
                if Arc::strong_count(&o2) == 1 {
                    return;
                }
                continue;
            };
            let o2 = o2.clone();
            tokio::spawn(async move {
                let Ok(Ok(h)) = timeout(STEP, r.start()).await else { return };
                while let Ok(Ok(Some(e))) = timeout(STEP, r.next()).await {
                    match (&h, e) {
                        (StreamHeader::Subgroup(sh), UniEvent::SubgroupObject(o)) if !o.payload.is_empty() => {
                            o2.lock().unwrap().insert((sh.track_alias, sh.group_id, o.object_id), o.payload);
                        }
                        (StreamHeader::Fetch(_), UniEvent::FetchEntry(FetchEntry::Object(_))) => {}
                        _ => {}
                    }
                }
            });
        }
    });
    Watcher { client: c, _ep: ep, tracks, objects, ended }
}

/// Every media object of `t` verified against its receipts under `key`: returns the
/// frames, in group and object order, and the number of groups whose receipts verified.
pub fn verified(wt: &Watcher, t: &Watched, key: &SigningKey, codec: Codec) -> (Vec<Vec<u8>>, usize) {
    let v = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
    let objs = wt.objects.lock().unwrap();
    let mut bundles = BTreeMap::new();
    for ((alias, g, _), p) in objs.iter() {
        if *alias == t.receipts_alias {
            let b = ReceiptBundle::from_payload(p).expect("a receipts bundle");
            let vg = v.verify_group(&b.group).expect("the group receipt verifies under the gateway's key");
            assert_eq!(b.group.group_id, GroupId(*g), "a receipt names its MOQT group");
            bundles.insert(*g, (b, vg));
        }
    }
    let mut frames = Vec::new();
    let mut prev: Option<(u64, [u8; 32])> = None;
    for (g, (b, _)) in &bundles {
        if let Some((pg, link)) = prev
            && pg + 1 == *g
        {
            assert_eq!(b.group.parent_receipt_hash, Some(link), "group {g} chains to group {pg}");
        }
        prev = Some((*g, b.group.link_hash()));
    }
    for ((alias, g, o), p) in objs.iter() {
        if *alias != t.media_alias {
            continue;
        }
        let (b, vg) = bundles.get(g).unwrap_or_else(|| panic!("the receipts of group {g} arrived"));
        let r = &b.objects[*o as usize];
        r.check_moqt(b.group.track_alias, GroupId(*g), ObjectId(*o)).expect("the receipt names its object");
        v.verify_payload(p, r, vg).expect("every object verifies against its signed root");
        let opened = frame::open(p).expect("an object is one WAI1 envelope");
        assert_eq!(opened.codec, codec);
        frames.push(frame::unframe(codec, &opened.payload).expect("one frame in its framing"));
    }
    // Every object every receipt names arrived.
    let named: usize = bundles.values().map(|(b, _)| b.objects.len()).sum();
    assert_eq!(named, frames.len(), "every receipted object arrived");
    (frames, bundles.len())
}

pub fn ns_of(w: &World, kind: u8) -> TrackNamespace {
    catalog::media_namespace(&w.room, &w.owner(), kind)
}

/// Wait, bounded, until the subscriber has read nothing new for `quiet`.
pub async fn settle(wt: &Watcher, quiet: Duration) {
    let deadline = Instant::now() + STEP;
    let mut last = (wt.objects.lock().unwrap().len(), Instant::now());
    while Instant::now() < deadline {
        tokio::time::sleep(Duration::from_millis(50)).await;
        let n = wt.objects.lock().unwrap().len();
        if n != last.0 {
            last = (n, Instant::now());
        } else if last.1.elapsed() >= quiet {
            return;
        }
    }
}

/// Wait, bounded, until the gateway's WHIP session has taken `n` frames: everything sent
/// has arrived before the session ends.
pub async fn received(w: &World, p: &Peer, n: usize) -> bool {
    let id = p.location.rsplit('/').next().unwrap().to_string();
    until(STEP, || w.gateway.gateway.session_frames(false, &id).is_some_and(|f| f as usize >= n)).await
}

/// `got` is a run of `sent`, from where its first element is: that index.
pub fn run_of(sent: &[Vec<u8>], got: &[Vec<u8>]) -> usize {
    assert!(!got.is_empty(), "something arrived");
    let j = sent.iter().position(|f| *f == got[0]).expect("the first frame received is one that was sent");
    assert!(j + got.len() <= sent.len(), "more frames received than were sent after the first: {} from {} of {}", got.len(), j, sent.len());
    same(got, &sent[j..j + got.len()], "the frames received are the frames sent, in order, byte for byte, from the first received");
    j
}

/// The payloads of track `t`'s media objects, in group and object order, as envelopes.
pub fn envelopes(wt: &Watcher, t: &Watched) -> Vec<frame::Opened> {
    wt.objects.lock().unwrap().iter().filter(|(k, _)| k.0 == t.media_alias).map(|(_, p)| frame::open(p).expect("an envelope")).collect()
}

/// `got` equals `want`, frame for frame; on failure, says where they part, not every byte.
pub fn same(got: &[Vec<u8>], want: &[Vec<u8>], what: &str) {
    if got == want {
        return;
    }
    let at = got.iter().zip(want).position(|(a, b)| a != b);
    panic!("{what}: {} received, {} sent, first difference at {:?}", got.len(), want.len(), at);
}

/// DELETE a session's resource with `token`.
pub async fn delete(w: &World, location: &str, auth: &Auth) -> Resp {
    auth.send("DELETE", &format!("{}{}", w.base(), location), &[], b"").await
}