//! Relays that cascade (`crate::cascade`): an edge relay that dials a home relay, and chains
//! and loops of them, every relay in this process on ephemeral loopback ports.
//!
//! - Two relays, over each transport the edge can dial (native QUIC and WebTransport with a
//!   pinned certificate, and the WebSocket binding): a publisher on each, a subscriber of the
//!   other's track on each, every object verified against its publisher's receipts at the
//!   subscriber and by the door of every relay it crossed; namespaces published both ways and
//!   discovered from either relay; each relay's summary of the link names the other.
//! - Three relays in a chain, with a room's tracks both ways: the path a namespace's
//!   announcement took, and a per-hop record from every hop, whose entries chain hop to hop.
//! - Receipts at every door: a tampered object a relay without a door forwards is refused at
//!   the next relay's door.
//! - A loop of three relays: an announcement never comes back to its origin, and a request for
//!   a track nobody publishes is refused `relay_loop` at once.
//! - Failover: a link to an upstream that is down is passed over for the next; with none up, a
//!   SUBSCRIBE is refused `TIMEOUT` `no_upstream`.
//! - GOAWAY: an upstream's GOAWAY moves the edge's subscriptions to the URI it names, and its
//!   subscriber sees no GOAWAY and loses no group.
//! - The DVR window of the home relay, fetched from the edge: served with the seek power, and
//!   refused `seek_not_granted` at the edge without it.
//! - Co-performance across relays: a member's votes counted in its chain order at the relay it
//!   publishes to, and a watcher at the other relay given exactly what one beside it is; and the
//!   floor held once, at the performer's relay.
//! - SUBSCRIBE_TRACKS across relays: the edge takes a subscriber's request to its link, and the
//!   home relay's tracks under what the link routes come back as PUBLISHes the edge offers on,
//!   one SUBSCRIBE_TRACKS on the link for every request of the edge that asks the same, and
//!   none once the last of them is cancelled.

mod common;

use common::*;
use ed25519_dalek::{Signer, SigningKey};
use joule_moq::carriage::{object_properties, ObjectExtras, ReceiptBundle};
use joule_moq::draft22::data::{FetchEntry, ObjectStatus as WireStatus, SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::{ControlMessage, Location, LocationFilter, NamespaceRequest, Parameter, Parameters, Publish, RequestOk, Token};
use joule_moq::messages::ObjectHeader;
use joule_moq::receipt::ReceiptVerifier;
use joule_moq::{GroupId, ObjectId, ObjectStatus, TrackAlias, TrackNamespace};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{Authority, DoorRule, DvrConfig, DvrWindow, FloorConfig, Grant, UpstreamConfig, UpstreamTransport};
use joule_moq_relay::wire::ReqRx;
use joule_moq_relay::{RelayConfig, RelayServer};
use std::collections::BTreeSet;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::time::timeout;

fn hexkey(k: &SigningKey) -> String {
    hex::encode(k.verifying_key().to_bytes())
}

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

fn sha256(b: &[u8]) -> [u8; 32] {
    ring::digest::digest(&ring::digest::SHA256, b).as_ref().try_into().unwrap()
}

/// A loopback port nothing listens on once this returns.
fn free_port() -> u16 {
    std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port()
}

fn upstream(url: String, transport: UpstreamTransport, prefixes: &[&[&str]]) -> UpstreamConfig {
    UpstreamConfig {
        name: url.clone(),
        url,
        transport,
        prefixes: prefixes.iter().map(|p| p.iter().map(|f| f.to_string()).collect()).collect(),
        ca_path: None,
        cert_sha256: None,
        native_roots: false,
        token_sha256: None,
        token_path: None,
        relay_key: None,
    }
}

/// An upstream entry for `server`, dialed over `via`: native QUIC and WebTransport trust its
/// certificate by its pinned SHA-256.
fn upstream_to(server: &RelayServer, via: Via, prefixes: &[&[&str]]) -> UpstreamConfig {
    let path = server.relay().config().path.clone();
    let pin = Some(hex::encode(server.identity.leaf_sha256));
    let mut u = match via {
        Via::Quic => upstream(format!("moqt://127.0.0.1:{}{path}", server.quic_addr.unwrap().port()), UpstreamTransport::Quic, prefixes),
        Via::WebTransport => upstream(format!("https://127.0.0.1:{}{path}", server.quic_addr.unwrap().port()), UpstreamTransport::WebTransport, prefixes),
        _ => upstream(format!("ws://{}{path}", server.websocket_addr.unwrap()), UpstreamTransport::WebSocket, prefixes),
    };
    if matches!(via, Via::Quic | Via::WebTransport) {
        u.cert_sha256 = pin;
    }
    u.relay_key = Some(hex::encode(server.relay().relay_id()));
    u
}

/// A token in a file of its own, for an upstream entry: (its SHA-256, the file).
fn token_file(tag: &str, token: &[u8]) -> (String, std::path::PathBuf) {
    let p = std::env::temp_dir().join(format!("joule-moq-relay-cascade-{}-{tag}.token", std::process::id()));
    std::fs::write(&p, [token, b"\n"].concat()).unwrap();
    (hex::encode(sha256(token)), p)
}

fn quick(c: &mut RelayConfig) {
    c.limits.goaway_timeout_ms = 500;
    c.limits.upstream_reconnect_ms = 100;
}

/// Wait until `server` has `n` upstream sessions open.
async fn linked(server: &RelayServer, n: usize) {
    assert!(until(STEP, || server.relay().links().iter().filter(|l| l.session().is_some()).count() >= n).await, "the upstream links open");
}

/// The namespaces a SUBSCRIBE_NAMESPACE's stream names, as NAMESPACE and NAMESPACE_DONE leave
/// them, once it holds `want` of them.
async fn namespaces_seen(rx: &mut ReqRx, want: usize) -> BTreeSet<TrackNamespace> {
    let mut seen = BTreeSet::new();
    while seen.len() < want {
        match timeout(STEP, rx.next()).await.expect("NAMESPACE in time").expect("a message") {
            Some(ControlMessage::Namespace(s)) => {
                seen.insert(s);
            }
            Some(ControlMessage::NamespaceDone(s)) => {
                seen.remove(&s);
            }
            other => panic!("expected NAMESPACE, got {other:?}"),
        }
    }
    seen
}

fn door(keys: &[&SigningKey]) -> Vec<DoorRule> {
    vec![DoorRule { prefix: vec!["live".into()], trust: keys.iter().map(|k| hexkey(k)).collect(), trust_field: None, hold_ms: 5_000 }]
}

// ── Two relays, over each transport ────────────────────────────────────────

async fn two_relays(via: Via) {
    const G: u64 = 8;
    let key_a = SigningKey::from_bytes(&[0xc1; 32]);
    let key_b = SigningKey::from_bytes(&[0xc2; 32]);
    let home = relay(|c| {
        quick(c);
        c.receipts.door = door(&[&key_a, &key_b]);
    })
    .await;
    let edge = relay(|c| {
        quick(c);
        c.receipts.door = door(&[&key_a, &key_b]);
        c.upstream = vec![upstream_to(&home, via, &[&["live"]])];
    })
    .await;
    linked(&edge, 1).await;
    let (home_id, edge_id) = (home.relay().relay_id(), edge.relay().relay_id());
    assert_eq!(edge.relay().links()[0].session().unwrap().peer.unwrap().id, home_id, "the edge's link names the home relay by its id");

    // A publisher on each: live/a on the home relay by PUBLISH, live/b on the edge by
    // PUBLISH_NAMESPACE; live/c, a namespace alone, on the home relay.
    let ta = Tracks::new(ns(&["live", "a"]), "cam", 10, 8, 64);
    let tb = Tracks::new(ns(&["live", "b"]), "cam", 20, 9, 96);
    let (pa, _ea) = connect(&home, Via::WebSocket, ClientOptions::default()).await;
    let (pb, _eb) = connect(&edge, Via::WebSocket, ClientOptions::default()).await;
    let (pc, _ec) = connect(&home, Via::WebSocket, ClientOptions::default()).await;
    let _pub_a = publish_tracks(&pa, &ta, None).await;
    let _ns_b = pb.publish_namespace(tb.ns.clone(), Parameters::new()).await.expect("PUBLISH_NAMESPACE on the edge");
    let _ns_c = pc.publish_namespace(ns(&["live", "c"]), Parameters::new()).await.expect("PUBLISH_NAMESPACE on the home relay");
    // The edge publishes live/b on its link: the home relay holds it, its path the edge alone.
    assert!(
        until(STEP, || home.relay().namespaces().iter().any(|(n, _, p)| *n == tb.ns && *p == vec![edge_id])).await,
        "the edge's namespace reaches the home relay, its path the edge"
    );

    // Discovery from either relay: live/b and live/c, wherever each is published.
    let (dh, _) = connect(&home, Via::WebSocket, ClientOptions::default()).await;
    let (de, _) = connect(&edge, Via::WebSocket, ClientOptions::default()).await;
    let (_dht, mut dhr) = dh.subscribe_namespace(ns(&["live"])).await.expect("SUBSCRIBE_NAMESPACE on the home relay");
    let (_det, mut der) = de.subscribe_namespace(ns(&["live"])).await.expect("SUBSCRIBE_NAMESPACE on the edge");
    let want: BTreeSet<TrackNamespace> = [ns(&["b"]), ns(&["c"])].into_iter().collect();
    assert_eq!(namespaces_seen(&mut dhr, 2).await, want, "the home relay names the edge's namespace and its own");
    assert_eq!(namespaces_seen(&mut der, 2).await, want, "the edge names its own namespace and the home relay's");

    // A subscriber of the other's track on each relay.
    let (sb, _e1) = connect(&edge, via, ClientOptions::default()).await;
    let (sa, _e2) = connect(&home, via, ClientOptions::default()).await;
    let vb = subscribe_both(&sb, &ta, Parameters::new()).await;
    let (held_b, va) = tokio::join!(answer_subscribes(&pb, &tb), subscribe_both(&sa, &tb, Parameters::new()));
    let pub_a = async {
        for g in 0..G {
            publish_group(&pa, &key_a, &ta, g, &payload("a", g, 700)).await;
        }
    };
    let pub_b = async {
        for g in 0..G {
            publish_group(&pb, &key_b, &tb, g, &payload("b", g, 900)).await;
        }
    };
    let ((), (), got_b, got_a) = tokio::join!(pub_a, pub_b, collect(&sb, STEP * 2, |g| g.len() as u64 >= 2 * G), collect(&sa, STEP * 2, |g| g.len() as u64 >= 2 * G));
    let (ob, bb, _, _) = verify(&got_b, &vb, &key_a);
    let (oa, ba, _, _) = verify(&got_a, &va, &key_b);
    assert_eq!((ob, bb, oa, ba), (G as usize, G as usize, G as usize, G as usize), "every object of each publisher, verified, at the other relay");
    assert_eq!(home.relay().stats().refused_at_door + edge.relay().stats().refused_at_door, 0);
    assert_eq!(edge.relay().stats().forwarded_upstream, 2, "the media track and its receipts track, each forwarded upstream once");
    drop(held_b);

    for c in [&pa, &pb, &pc, &dh, &de, &sb, &sa] {
        c.close(0, "done");
    }
    let (er, hr) = (edge.relay().clone(), home.relay().clone());
    edge.shutdown().await;
    home.shutdown().await;
    // A session leaves the relay's table before its summary is signed: wait for every one.
    assert!(until(STEP, || er.summaries().len() >= 4 && hr.summaries().len() >= 5).await, "every session's summary at each relay");
    // Every hop's door verified the objects it forwarded: the publishers' signed figures, at
    // the subscriber's relay, for the other relay's media objects.
    // (The link's own summary at each end counts what that relay forwarded over it, verified at
    // its door, too: home's publisher's objects at the home relay, the edge's at the edge.)
    let signed = |r: &joule_moq_relay::Relay, link: bool| {
        r.summaries().iter().filter(|s| s.summary.tally.peer.is_some() == link && s.summary.tally.subscriber.objects_forwarded == 2 * G).map(|s| s.summary.tally.publisher_signed.get("unmetered").map_or(0, |f| f.objects)).collect::<Vec<_>>()
    };
    assert_eq!((signed(&er, false), signed(&hr, false)), (vec![G], vec![G]), "each relay's door verified every media object its subscriber was given");
    assert_eq!((signed(&er, true), signed(&hr, true)), (vec![G], vec![G]), "and every one it sent the other relay");
    // Each relay's summary of the link names the other relay.
    assert!(er.summaries().iter().any(|s| s.summary.tally.peer.as_deref() == Some(hex::encode(home_id).as_str())));
    assert!(hr.summaries().iter().any(|s| s.summary.tally.peer.as_deref() == Some(hex::encode(edge_id).as_str())));
    for s in er.summaries().iter().chain(hr.summaries().iter()) {
        s.verify().expect("every summary verifies under its relay's key");
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn two_relays_over_a_websocket_upstream() {
    timeout(Duration::from_secs(120), two_relays(Via::WebSocket)).await.expect("in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn two_relays_over_a_quic_upstream() {
    timeout(Duration::from_secs(120), two_relays(Via::Quic)).await.expect("in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn two_relays_over_a_webtransport_upstream() {
    timeout(Duration::from_secs(120), two_relays(Via::WebTransport)).await.expect("in time");
}

// ── Three relays in a chain, and their per-hop records ─────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn three_relays_in_a_chain() {
    timeout(Duration::from_secs(120), chain()).await.expect("in time");
}

async fn chain() {
    const G: u64 = 6;
    let t0 = now_us() - 1_000_000;
    let room = [0x5a; 32];
    let rh = hex::encode(room);
    let (kp, kq) = (SigningKey::from_bytes(&[0xd1; 32]), SigningKey::from_bytes(&[0xd2; 32]));
    let tp = Tracks::new(ns(&[&rh, &hexkey(&kp), "audio"]), "0", 30, 31, 32);
    let tq = Tracks::new(ns(&[&rh, &hexkey(&kq), "audio"]), "0", 40, 41, 32);
    let door = |c: &mut RelayConfig| c.receipts.door = vec![DoorRule { prefix: vec![rh.clone()], trust: vec![hexkey(&kp), hexkey(&kq)], trust_field: None, hold_ms: 5_000 }];
    // C dials B, B dials A, each routing the room.
    let a = relay(|c| {
        quick(c);
        door(c);
    })
    .await;
    let b = relay(|c| {
        quick(c);
        door(c);
        c.upstream = vec![upstream_to(&a, Via::WebSocket, &[&[&rh]])];
    })
    .await;
    let cc = relay(|c| {
        quick(c);
        door(c);
        c.upstream = vec![upstream_to(&b, Via::WebSocket, &[&[&rh]])];
    })
    .await;
    linked(&b, 1).await;
    linked(&cc, 1).await;
    let (ida, idb, idc) = (a.relay().relay_id(), b.relay().relay_id(), cc.relay().relay_id());

    // P publishes on A by PUBLISH; Q on C by PUBLISH_NAMESPACE, which reaches A through B.
    let (pp, _) = connect(&a, Via::WebSocket, ClientOptions::default()).await;
    let (pq, _) = connect(&cc, Via::WebSocket, ClientOptions::default()).await;
    let _pub_p = publish_tracks(&pp, &tp, None).await;
    let _ns_q = pq.publish_namespace(tq.ns.clone(), Parameters::new()).await.expect("PUBLISH_NAMESPACE on C");
    assert!(until(STEP, || a.relay().namespaces().iter().any(|(n, _, p)| *n == tq.ns && *p == vec![idc, idb])).await, "Q's namespace reaches A, its path C then B");

    // A viewer of P on C (C → B → A by route), a viewer of Q on A (A → B → C by announcement).
    let (vc, _) = connect(&cc, Via::WebSocket, ClientOptions::default()).await;
    let (va, _) = connect(&a, Via::WebSocket, ClientOptions::default()).await;
    let sp = subscribe_both(&vc, &tp, Parameters::new()).await;
    let (held_q, sq) = tokio::join!(answer_subscribes(&pq, &tq), subscribe_both(&va, &tq, Parameters::new()));
    let pub_p = async {
        for g in 0..G {
            publish_group(&pp, &kp, &tp, g, &payload("p", g, 500)).await;
        }
    };
    let pub_q = async {
        for g in 0..G {
            publish_group(&pq, &kq, &tq, g, &payload("q", g, 600)).await;
        }
    };
    let ((), (), got_c, got_a) = tokio::join!(pub_p, pub_q, collect(&vc, STEP * 2, |g| g.len() as u64 >= 2 * G), collect(&va, STEP * 2, |g| g.len() as u64 >= 2 * G));
    assert_eq!(verify(&got_c, &sp, &kp).0, G as usize, "P's objects, verified, at C");
    assert_eq!(verify(&got_a, &sq, &kq).0, G as usize, "Q's objects, verified, at A");
    drop(held_q);
    for c in [&pp, &pq, &vc, &va] {
        c.close(0, "done");
    }
    let (ra, rb, rc) = (a.relay().clone(), b.relay().clone(), cc.relay().clone());
    cc.shutdown().await;
    b.shutdown().await;
    a.shutdown().await;
    // A session leaves the relay's table before its summary is signed: wait for every one (A:
    // P, its viewer, the link from B; B: its two links; C: Q, its viewer, the link to B).
    assert!(until(STEP, || ra.summaries().len() >= 3 && rb.summaries().len() >= 2 && rc.summaries().len() >= 3).await, "every session's summary at each relay");

    // A per-hop record from every hop, each signed by its own relay; P's media track and Q's
    // chain hop to hop: what one hop sent the next, the next received.
    let render = |t: &Tracks| joule_moq::draft22::names::render(&t.ns, Some(&t.media));
    let digest = |t: &Tracks| match joule_moq_relay::metering::classify(&render(t)).expect("a track of the room").1 {
        joule_moq_relay::metering::Kind::Media { digest, .. } => digest,
        other => panic!("not a media track: {other:?}"),
    };
    let (dp, dq) = (digest(&tp), digest(&tq));
    let recs: Vec<_> = [&ra, &rb, &rc].iter().map(|r| r.hop_record(room, t0)).collect();
    for (rec, id) in recs.iter().zip([ida, idb, idc]) {
        rec.verify().expect("the record verifies under its hop's key");
        assert_eq!(rec.hop, id, "each hop signs its own record");
        // Each hop's own figure: the unmetered marker on a host with no counter, or its own
        // counter's advance (OnChipCounter) where the relay reads one, as on macOS.
        match rec.class_tag {
            0 => assert_eq!(rec.joules, 0, "the unmetered marker is 0 with class tag 0"),
            4 => {}
            t => panic!("class tag {t}: the relay signs the unmetered marker or OnChipCounter"),
        }
    }
    let objects = |rec: &joule_moq_relay::metering::HopRecord, track: [u8; 32], dir: u8| rec.entries.iter().find(|e| e.kind == 1 && e.track == track && e.dir == dir).map_or(0, |e| e.objects);
    for (track, name) in [(dp, "P"), (dq, "Q")] {
        let flow: Vec<(u64, u64)> = recs.iter().map(|r| (objects(r, track, 1), objects(r, track, 2))).collect();
        assert_eq!(flow, vec![(G, G); 3], "{name}'s media objects in and out at A, B and C");
    }
    // Only the viewers are subscriber sessions: a hop link is counted at neither end.
    assert_eq!(recs.iter().map(|r| r.sessions).collect::<Vec<_>>(), vec![1, 0, 1], "one subscriber session at A, none at B, one at C");
    assert!(recs[1].ledger.is_empty(), "B served no principal: its ledger is empty");
    assert_eq!(recs[0].ledger.iter().map(|l| l.sessions).sum::<u32>(), 2, "A's ledger: P's session and the viewer's, not the link");
}

// ── Receipts at every door ─────────────────────────────────────────────────

/// Group `g` of `t` signed over `signed`, its object carrying `sent`.
async fn publish_group_as(c: &Client, key: &SigningKey, t: &Tracks, g: u64, signed: &[u8], sent: &[u8]) {
    let (receipt, bundle) = signed_group(key, t.receipt_alias, g, signed);
    let rh = SubgroupHeader { track_alias: t.receipts_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
    c.send_subgroup(rh, &[SubgroupObject::normal(0, bundle.to_payload().unwrap())]).await.unwrap();
    let header = ObjectHeader { track_alias: TrackAlias(t.media_alias), group_id: GroupId(g), object_id: ObjectId(0), publisher_priority: t.priority, status: ObjectStatus::Normal, payload_len: sent.len() as u64, mrl_layer: 0, mrl_total_layers: 0, wai_envelope: false };
    let props = object_properties(&header, &ObjectExtras { receipt: Some(receipt), wai_receipt: None, delivery_timeout_ms: None }).unwrap();
    let mh = SubgroupHeader { track_alias: t.media_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: true, end_of_group: true, first_object: true };
    c.send_subgroup(mh, &[SubgroupObject { object_id: 0, properties: props, status: WireStatus::Normal, payload: sent.to_vec() }]).await.unwrap();
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn receipts_are_checked_at_every_door() {
    timeout(Duration::from_secs(120), every_door()).await.expect("in time");
}

async fn every_door() {
    const G: u64 = 6;
    const BAD: u64 = 3;
    let key = SigningKey::from_bytes(&[0xe1; 32]);
    // The home relay keeps no door; the edge does.
    let home = relay(quick).await;
    let edge = relay(|c| {
        quick(c);
        c.receipts.door = door(&[&key]);
        c.upstream = vec![upstream_to(&home, Via::WebSocket, &[&["live"]])];
    })
    .await;
    linked(&edge, 1).await;
    let t = Tracks::new(ns(&["live", "t"]), "cam", 10, 8, 64);
    let (p, _) = connect(&home, Via::WebSocket, ClientOptions::default()).await;
    let _pubs = publish_tracks(&p, &t, None).await;
    let (vh, _) = connect(&home, Via::WebSocket, ClientOptions::default()).await;
    let (ve, _) = connect(&edge, Via::WebSocket, ClientOptions::default()).await;
    let sh = subscribe_both(&vh, &t, Parameters::new()).await;
    let se = subscribe_both(&ve, &t, Parameters::new()).await;
    for g in 0..G {
        let good = payload("t", g, 800);
        let mut sent = good.clone();
        if g == BAD {
            // Changed after it was signed: its receipt names the bytes it no longer holds.
            sent[20] ^= 0x01;
        }
        publish_group_as(&p, &key, &t, g, &good, &sent).await;
    }
    let got_h = collect(&vh, STEP, |g| g.len() as u64 >= 2 * G).await;
    let got_e = collect(&ve, STEP, |g| g.iter().filter(|x| x.header.track_alias == se.receipts_alias).count() as u64 >= G && g.iter().filter(|x| x.header.track_alias == se.media_alias).count() as u64 >= G - 1).await;
    // The home relay, with no door, forwarded the tampered object: its own subscriber has it,
    // and it does not verify.
    let bad_at_home = got_h.iter().find(|x| x.header.track_alias == sh.media_alias && x.header.group_id == BAD).expect("the home relay forwarded group 3");
    let mut changed = payload("t", BAD, 800);
    changed[20] ^= 0x01;
    assert_eq!(bad_at_home.objects[0].payload, changed, "the home relay forwarded the changed bytes");
    assert_eq!(home.relay().stats().refused_at_door, 0, "the home relay checked nothing");
    // The edge checked it again at its door and refused it, though its upstream forwarded it.
    let media_e: Vec<u64> = got_e.iter().filter(|x| x.header.track_alias == se.media_alias && x.reset.is_none() && !x.objects.is_empty()).map(|x| x.header.group_id).collect();
    assert!(!media_e.contains(&BAD), "the edge forwarded no tampered object: {media_e:?}");
    assert_eq!(media_e.len() as u64, G - 1, "every other group reached the edge's subscriber");
    // Each group's door verdict is reached in that group's own task, and nothing the
    // subscriber was given waits for group BAD's: every other group can reach it while BAD is
    // still on its way to the door. So the refusal is waited for, then counted.
    assert!(until(STEP, || edge.relay().stats().refused_at_door >= 1).await, "the edge's door refused the tampered object");
    assert_eq!(edge.relay().stats().refused_at_door, 1, "the edge's door refused the tampered object, and nothing else");
    let verifier = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
    for x in got_e.iter().filter(|x| x.header.track_alias == se.receipts_alias) {
        verifier.verify_group(&ReceiptBundle::from_payload(&x.objects[0].payload).unwrap().group).expect("every receipt verifies");
    }
    for c in [&p, &vh, &ve] {
        c.close(0, "done");
    }
    let er = edge.relay().clone();
    edge.shutdown().await;
    home.shutdown().await;
    assert!(until(STEP, || er.summaries().len() >= 2).await, "the edge's viewer's and its link's summaries");
    // The refusal is counted against the link it arrived on: the home relay's session.
    let link = er.summaries().into_iter().find(|s| s.summary.tally.peer.is_some()).expect("the link's summary");
    assert_eq!(link.summary.tally.publisher.refused_at_door, 1);
    // A refusal counted is not a group withheld, and the subscriber was read only until it
    // held every other group: a door that counted late and then forwarded would pass both.
    // What the edge gave its viewer, counted once the session ended, is every group but BAD.
    let viewer = er.summaries().into_iter().find(|s| s.summary.tally.peer.is_none()).expect("the viewer's summary");
    let m = viewer.summary.tally.subscriber.tracks.get(&joule_moq::draft22::names::render(&t.ns, Some(&t.media))).cloned().unwrap_or_default();
    assert_eq!(m.objects_forwarded + m.undelivered_at_close, G - 1, "the edge gave its viewer no tampered object");
}

// ── A loop ─────────────────────────────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_loop_of_relays() {
    timeout(Duration::from_secs(120), relay_loop()).await.expect("in time");
}

async fn relay_loop() {
    // A dials B, B dials C, C dials A, each routing everything under live: a loop.
    // Each relay dials the next, so the three ports are chosen before any of them listens, and
    // another test can bind one in between: a start refused for a port in use is tried again
    // on three new ports, at most five times. Anything else a start refuses fails the test.
    let (ra, rb, rc) = {
        let mut attempt = 0;
        'start: loop {
            attempt += 1;
            let ports = [free_port(), free_port(), free_port()];
            let mut started = Vec::new();
            for i in 0..3 {
                let mut c = RelayConfig::loopback();
                quick(&mut c);
                c.websocket_listen = Some(format!("127.0.0.1:{}", ports[i]).parse().unwrap());
                c.upstream = vec![upstream(format!("ws://127.0.0.1:{}/moq", ports[(i + 1) % 3]), UpstreamTransport::WebSocket, &[&["live"]])];
                match RelayServer::start(c).await {
                    Ok(r) => started.push(r),
                    Err(e) if attempt < 5 && e.contains("Address already in use") => {
                        for r in started {
                            r.shutdown().await;
                        }
                        continue 'start;
                    }
                    Err(e) => panic!("the relay starts: {e}"),
                }
            }
            let mut it = started.into_iter();
            break (it.next().unwrap(), it.next().unwrap(), it.next().unwrap());
        }
    };
    for r in [&ra, &rb, &rc] {
        linked(r, 1).await;
    }
    let (ida, idb) = (ra.relay().relay_id(), rb.relay().relay_id());
    let la = ns(&["live", "a"]);
    let t = Tracks::new(la.clone(), "cam", 10, 8, 64);
    let (p, _) = connect(&ra, Via::WebSocket, ClientOptions::default()).await;
    let _ns = p.publish_namespace(la.clone(), Parameters::new()).await.expect("PUBLISH_NAMESPACE on A");
    // B holds it through A, C through B; it never comes back to A.
    assert!(until(STEP, || rb.relay().namespaces().iter().any(|(n, _, path)| *n == la && *path == vec![ida])).await, "B holds A's namespace");
    assert!(until(STEP, || rc.relay().namespaces().iter().any(|(n, _, path)| *n == la && *path == vec![ida, idb])).await, "C holds it through B");
    tokio::time::sleep(Duration::from_millis(500)).await;
    let at_a: Vec<_> = ra.relay().namespaces().into_iter().filter(|(n, _, _)| *n == la).collect();
    assert_eq!(at_a.len(), 1, "A holds its own namespace once: it never came back around the loop");
    assert!(at_a[0].2.is_empty());
    let loops: u64 = [&ra, &rb, &rc].iter().map(|r| r.relay().stats().relay_loops).sum();
    assert_eq!(loops, 0, "no relay was sent a request its path refuses: none was forwarded toward a relay on its path");
    assert_eq!(rc.relay().links()[0].stats.announced.load(std::sync::atomic::Ordering::Relaxed), 0, "C published nothing back to A");

    // A viewer on C reaches A's publisher through B.
    let (v, _) = connect(&rc, Via::WebSocket, ClientOptions::default()).await;
    let (held, s) = tokio::join!(answer_subscribes(&p, &t), subscribe_both(&v, &t, Parameters::new()));
    let key = SigningKey::from_bytes(&[0xa7; 32]);
    for g in 0..3 {
        publish_group(&p, &key, &t, g, &payload("l", g, 300)).await;
    }
    let got = collect(&v, STEP, |g| g.len() >= 6).await;
    assert_eq!(verify(&got, &s, &key).0, 3, "every object, through the loop's relays, verified");
    drop(held);

    // A track nobody publishes, from B: forwarded to C, then to A, whose only link leads back to
    // B, on the path. Refused at once with relay_loop, not held until a timeout.
    let (w, _) = connect(&rb, Via::WebSocket, ClientOptions::default()).await;
    let asked = std::time::Instant::now();
    let refused = w.subscribe(ns(&["live", "nobody"]), name("x"), Parameters::new()).await.map(|_| ()).expect_err("refused");
    assert!(refused.contains("relay_loop"), "{refused}");
    assert!(asked.elapsed() < Duration::from_secs(4), "refused at once: {:?}", asked.elapsed());
    // A refused it, the relay whose only link leads back to B: B was never sent a request that
    // names it, so no relay refused one where it arrived.
    let loops: Vec<u64> = [&ra, &rb, &rc].iter().map(|r| r.relay().stats().relay_loops).collect();
    assert_eq!(loops, vec![1, 0, 0], "the loop is refused at the relay that would close it, before it is sent");
    for c in [&p, &v, &w] {
        c.close(0, "done");
    }
    for r in [ra, rb, rc] {
        r.shutdown().await;
    }
}

// ── Failover ───────────────────────────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_upstream_that_is_down_is_passed_over() {
    timeout(Duration::from_secs(120), failover()).await.expect("in time");
}

async fn failover() {
    let up = relay(quick).await;
    let dead = format!("ws://127.0.0.1:{}/moq", free_port());
    let edge = relay(|c| {
        quick(c);
        c.upstream = vec![upstream(dead.clone(), UpstreamTransport::WebSocket, &[&["live"]]), upstream_to(&up, Via::WebSocket, &[&["live"]])];
    })
    .await;
    let lonely = relay(|c| {
        quick(c);
        c.upstream = vec![upstream(dead.clone(), UpstreamTransport::WebSocket, &[&["live"]])];
    })
    .await;
    linked(&edge, 1).await;
    assert!(edge.relay().links()[0].session().is_none() && edge.relay().links()[1].session().is_some(), "the first upstream is down, the second up");
    let t = Tracks::new(ns(&["live", "f"]), "cam", 10, 8, 64);
    let (p, _) = connect(&up, Via::WebSocket, ClientOptions::default()).await;
    let _pubs = publish_tracks(&p, &t, None).await;
    let (v, _) = connect(&edge, Via::WebSocket, ClientOptions::default()).await;
    let s = subscribe_both(&v, &t, Parameters::new()).await;
    let key = SigningKey::from_bytes(&[0xf1; 32]);
    for g in 0..3 {
        publish_group(&p, &key, &t, g, &payload("f", g, 300)).await;
    }
    let got = collect(&v, STEP, |g| g.len() >= 6).await;
    assert_eq!(verify(&got, &s, &key).0, 3, "served through the second upstream");
    // No upstream up: a typed refusal, at once.
    let (w, _) = connect(&lonely, Via::WebSocket, ClientOptions::default()).await;
    let refused = w.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.map(|_| ()).expect_err("refused");
    assert!(refused.contains("REQUEST_ERROR TIMEOUT") && refused.contains("no_upstream"), "{refused}");
    assert_eq!(lonely.relay().stats().no_upstream, 1);
    // With a rendezvous, the request waits for an upstream within it, then is refused the same.
    let mut rz = Parameters::new();
    rz.push(Parameter::RendezvousTimeout(300));
    let refused = w.subscribe(t.ns.clone(), t.media.clone(), rz).await.map(|_| ()).expect_err("refused");
    assert!(refused.contains("no_upstream"), "{refused}");
    for c in [&p, &v, &w] {
        c.close(0, "done");
    }
    for r in [edge, lonely, up] {
        r.shutdown().await;
    }
}

// ── GOAWAY ─────────────────────────────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_upstream_goaway_moves_the_subscriptions() {
    timeout(Duration::from_secs(120), goaway()).await.expect("in time");
}

async fn goaway() {
    let key = SigningKey::from_bytes(&[0x9a; 32]);
    let t = Tracks::new(ns(&["live", "g"]), "cam", 10, 8, 64);
    let h2 = relay(quick).await;
    let to_h2 = format!("ws://{}{}", h2.websocket_addr.unwrap(), h2.relay().config().path);
    let h1 = relay(|c| {
        quick(c);
        c.limits.goaway_timeout_ms = 3_000;
        c.goaway_uri = to_h2.clone();
    })
    .await;
    let edge = relay(|c| {
        quick(c);
        c.upstream = vec![upstream(format!("ws://{}{}", h1.websocket_addr.unwrap(), h1.relay().config().path), UpstreamTransport::WebSocket, &[&["live"]])];
    })
    .await;
    linked(&edge, 1).await;
    // The same track published to both, as a publisher with a standby does.
    let (p1, _) = connect(&h1, Via::WebSocket, ClientOptions::default()).await;
    let (p2, _) = connect(&h2, Via::WebSocket, ClientOptions::default()).await;
    let _a = publish_tracks(&p1, &t, None).await;
    let _b = publish_tracks(&p2, &t, None).await;
    let (v, _) = connect(&edge, Via::WebSocket, ClientOptions::default()).await;
    let s = subscribe_both(&v, &t, Parameters::new()).await;
    for g in 0..4 {
        let pl = payload("g", g, 400);
        publish_group(&p1, &key, &t, g, &pl).await;
        publish_group(&p2, &key, &t, g, &pl).await;
    }
    let mut got = collect(&v, STEP, |g| g.len() >= 8).await;
    // H1 goes away, its GOAWAY naming H2: the edge moves its subscriptions there.
    let stopping = tokio::spawn(h1.shutdown());
    let link = edge.relay().links()[0].clone();
    assert!(until(STEP, || link.stats.moved.load(std::sync::atomic::Ordering::Relaxed) >= 2).await, "both subscriptions moved to H2");
    assert!(until(STEP, || link.session().is_some_and(|c| c.peer.unwrap().id == h2.relay().relay_id())).await, "the link's session is to H2");
    for g in 4..8 {
        publish_group(&p2, &key, &t, g, &payload("g", g, 400)).await;
    }
    got.extend(collect(&v, STEP, |g| g.len() >= 8).await);
    let (objects, groups, _, _) = verify(&got, &s, &key);
    assert_eq!((objects, groups), (8, 8), "every group, before and after the move, once");
    assert!(v.goaway().borrow().is_none(), "the upstream's GOAWAY is not passed on to the edge's subscriber");
    let _ = stopping.await;
    for c in [&p1, &p2, &v] {
        c.close(0, "done");
    }
    edge.shutdown().await;
    h2.shutdown().await;
}

// ── The DVR window across relays ───────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_home_dvr_window_is_fetched_from_the_edge() {
    timeout(Duration::from_secs(120), dvr()).await.expect("in time");
}

fn fetched_objects(entries: &[FetchEntry]) -> Vec<(u64, Vec<u8>)> {
    entries
        .iter()
        .filter_map(|e| match e {
            FetchEntry::Object(o) => Some((o.group_id, o.payload.clone())),
            _ => None,
        })
        .collect()
}

async fn dvr() {
    const G: u64 = 8;
    let key = SigningKey::from_bytes(&[0x3d; 32]);
    // The home relay's store in memory where the host has it (`store_dir`): the wait below for
    // its window to hold the track is for the store, not for a disk busy with other jobs.
    let dir = store_dir("cascade-dvr");
    let (pub_tok, peer_tok, seek_tok, live_tok) = (b"publisher".to_vec(), b"edge relay".to_vec(), b"viewer who seeks".to_vec(), b"viewer at the edge".to_vec());
    let (peer_sha, peer_file) = token_file("dvr-peer", &peer_tok);
    let rule = |t: &[u8], prefix: &[&str], publish: bool, subscribe: bool, seek: bool| Grant { prefix: prefix.iter().map(|s| s.to_string()).collect(), token_sha256: hex::encode(sha256(t)), publish, subscribe, proof: false, unlimited: true, seek };
    let home = relay(|c| {
        quick(c);
        c.auth.required = true;
        c.auth.grants = vec![rule(&pub_tok, &["live"], true, false, false), rule(&peer_tok, &[], true, true, true)];
        c.dvr = Some(DvrConfig { path: dir.clone(), window: vec![DvrWindow { prefix: vec!["live".into()], max_age_s: 600, max_bytes: 1 << 30 }], room_keys: vec![], takedown_keys: vec![] });
    })
    .await;
    let edge = relay(|c| {
        quick(c);
        c.auth.required = true;
        c.auth.grants = vec![rule(&seek_tok, &["live"], false, true, true), rule(&live_tok, &["live"], false, true, false)];
        let mut u = upstream_to(&home, Via::WebSocket, &[&["live"]]);
        u.token_sha256 = Some(peer_sha.clone());
        u.token_path = Some(peer_file.clone());
        c.upstream = vec![u];
    })
    .await;
    linked(&edge, 1).await;
    let with = |t: &[u8]| ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: t.to_vec() }], ..ClientOptions::default() };
    let t = Tracks::new(ns(&["live", "d"]), "cam", 10, 8, 64);
    let (p, _) = connect(&home, Via::WebSocket, with(&pub_tok)).await;
    let _pubs = publish_tracks(&p, &t, None).await;
    for g in 0..G {
        publish_group(&p, &key, &t, g, &payload("d", g, 600)).await;
    }
    assert!(until(STEP, || home.relay().dvr().unwrap().largest(&(t.ns.clone(), t.media.clone())).is_some_and(|l| l.group == G - 1)).await, "the home relay's window holds the track");
    let from = |g: u64| Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(g, 0) }));
    // A viewer with the seek power at the edge: the whole window, from the home relay, and every
    // object verifies against the receipts fetched the same way.
    let (vs, _) = connect(&edge, Via::WebSocket, with(&seek_tok)).await;
    let (_, media) = vs.fetch(t.ns.clone(), t.media.clone(), from(0)).await.expect("the window, fetched from the edge");
    let (_, receipts) = vs.fetch(t.ns.clone(), t.receipts.clone(), from(0)).await.expect("its receipts");
    let media = fetched_objects(&media);
    assert_eq!(media.iter().map(|(g, _)| *g).collect::<Vec<_>>(), (0..G).collect::<Vec<_>>());
    let verifier = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
    for ((g, body), (rg, r)) in media.iter().zip(fetched_objects(&receipts)) {
        assert_eq!(*g, rg);
        let b = ReceiptBundle::from_payload(&r).unwrap();
        let vg = verifier.verify_group(&b.group).unwrap();
        verifier.verify_payload(body, &b.objects[0], &vg).expect("a fetched object verifies");
    }
    // A viewer without it, at an edge that knows nothing of the track yet: refused there, below
    // the live edge it asked the home relay for, and served at the live edge.
    let (vl, _) = connect(&edge, Via::WebSocket, with(&live_tok)).await;
    let refused = vl.fetch(t.ns.clone(), t.media.clone(), from(0)).await.map(|_| ()).expect_err("refused");
    assert!(refused.contains("seek_not_granted"), "{refused}");
    assert_eq!(edge.relay().stats().seek_refused, 1);
    let (_, last) = vl.fetch(t.ns.clone(), t.media.clone(), from(G - 1)).await.expect("the live edge");
    assert_eq!(fetched_objects(&last).len(), 1);
    // A SUBSCRIBE whose fill starts at group 0, at an edge that learns the live edge only from
    // its upstream's SUBSCRIBE_OK: refused there once it does.
    let fill = Parameters::new().with(Parameter::FillParameters(Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(0, 0) }))));
    let refused = vl.subscribe(t.ns.clone(), t.media.clone(), fill).await.map(|_| ()).expect_err("a fill from group 0, refused");
    assert!(refused.contains("seek_not_granted"), "{refused}");
    assert_eq!(edge.relay().stats().seek_refused, 2);
    for c in [&p, &vs, &vl] {
        c.close(0, "done");
    }
    edge.shutdown().await;
    home.shutdown().await;
    let _ = std::fs::remove_dir_all(&dir);
    let _ = std::fs::remove_file(&peer_file);
}

// ── Co-performance across relays ───────────────────────────────────────────

/// A `meta` record of live-events §3.1, signed by `key`, after `prev` when given.
fn meta(key: &SigningKey, room: [u8; 32], seq: u64, prev: Option<&[u8]>, t: u64, payload: &[u8]) -> Vec<u8> {
    let mut b = b"WEV1\x01".to_vec();
    b.extend_from_slice(&room);
    b.extend_from_slice(&key.verifying_key().to_bytes());
    b.extend_from_slice(&seq.to_le_bytes());
    b.extend_from_slice(&prev.map_or([0; 32], |p| sha256(&p[..p.len() - 64])));
    b.extend_from_slice(&t.to_le_bytes());
    b.push(0);
    b.push(0x01);
    b.extend_from_slice(&(payload.len() as u16).to_le_bytes());
    b.extend_from_slice(payload);
    let mut m = b"wai:live-event\x01".to_vec();
    m.extend_from_slice(&b);
    b.extend_from_slice(&key.sign(&m).to_bytes());
    b
}

fn msf(record: &[u8]) -> Vec<u8> {
    use base64::Engine;
    format!("[{{\"t\":1,\"data\":{{\"wev1\":\"{}\"}}}}]", base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(record)).into_bytes()
}

/// A viewer's grant of `room` for `principal`, as live-rooms §4.1 lays it out, with `attrs`.
fn viewer_grant(authority: &SigningKey, room: [u8; 32], principal: &SigningKey, attrs: &[(&str, &str)]) -> Vec<u8> {
    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
    let mut g = b"WRG1\x01".to_vec();
    g.extend_from_slice(&room);
    g.extend_from_slice(&principal.verifying_key().to_bytes());
    g.extend_from_slice(&[0; 40]);
    g.extend_from_slice(&(now - 60).to_le_bytes());
    g.extend_from_slice(&(now + 600).to_le_bytes());
    g.extend_from_slice(&[1, 0, 1, 0, 0x10, 0]);
    g.extend_from_slice(&[0; 32]);
    g.extend_from_slice(b"WEP1\x01");
    g.extend_from_slice(&[0x0b; 32]);
    g.extend_from_slice(&1024u16.to_le_bytes());
    g.extend_from_slice(&500u16.to_le_bytes());
    g.extend_from_slice(&50u32.to_le_bytes());
    g.extend_from_slice(&1_000_000u64.to_le_bytes());
    g.extend_from_slice(&1u32.to_le_bytes());
    g.extend_from_slice(&1_000_000u64.to_le_bytes());
    g.extend_from_slice(&[0xff; 16]);
    g.extend_from_slice(&[0; 3]);
    g.extend_from_slice(&[0, 0]);
    let mut a = Vec::new();
    for (k, v) in attrs {
        a.push(k.len() as u8);
        a.extend_from_slice(k.as_bytes());
        a.extend_from_slice(&(v.len() as u16).to_le_bytes());
        a.extend_from_slice(v.as_bytes());
    }
    g.extend_from_slice(&(a.len() as u16).to_le_bytes());
    g.extend_from_slice(&a);
    let mut m = joule_moq::grant_proof::GRANT_DOMAIN.to_vec();
    m.extend_from_slice(&g);
    g.extend_from_slice(&authority.sign(&m).to_bytes());
    g
}

/// One object of group `g`, on a stream of its own.
async fn send(c: &Client, alias: u64, g: u64, payload: &[u8]) {
    let h = SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(16), properties: false, end_of_group: true, first_object: true };
    c.send_subgroup(h, &[SubgroupObject::normal(0, payload.to_vec())]).await.expect("an object");
}

fn room_track(room: [u8; 32], who: &SigningKey, field: &str) -> TrackNamespace {
    ns(&[&hex::encode(room), &hexkey(who), field])
}

/// What `c` receives until `want` arrives, then a second more: the payloads.
async fn payloads_until(c: &Client, want: &[u8]) -> Vec<Vec<u8>> {
    let mut got = collect(c, STEP, |g| g.iter().any(|x| x.objects.iter().any(|o| o.payload == want))).await;
    got.extend(collect(c, Duration::from_secs(1), |_| false).await);
    got.iter().flat_map(|g| g.objects.iter().map(|o| o.payload.clone())).collect()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn co_performance_votes_are_counted_once_in_chain_order() {
    timeout(Duration::from_secs(120), votes()).await.expect("in time");
}

async fn votes() {
    let room = [0xb1; 32];
    let authority = SigningKey::from_bytes(&[0xb2; 32]);
    let clock = SigningKey::from_bytes(&[0xb3; 32]);
    let (member, watcher_h, watcher_e) = (SigningKey::from_bytes(&[0xb4; 32]), SigningKey::from_bytes(&[0xb5; 32]), SigningKey::from_bytes(&[0xb6; 32]));
    let (clock_tok, peer_tok) = (b"the clock's token".to_vec(), b"the edge relay".to_vec());
    let (peer_sha, peer_file) = token_file("votes-peer", &peer_tok);
    let floor = FloorConfig { room_keys: vec![hexkey(&clock)], hold_ms: 60_000 };
    // The clock and a watcher on the home relay; the member and another watcher on the edge.
    let home = relay(|c| {
        quick(c);
        c.auth.required = true;
        c.auth.authorities = vec![Authority { key: hexkey(&authority), rooms: vec![] }];
        c.auth.grants = vec![
            Grant { prefix: vec!["*".into(), hexkey(&clock), "events".into()], token_sha256: hex::encode(sha256(&clock_tok)), publish: true, subscribe: true, proof: false, unlimited: true, seek: false },
            Grant { prefix: vec![], token_sha256: peer_sha.clone(), publish: true, subscribe: true, proof: false, unlimited: true, seek: true },
        ];
        c.floor = Some(floor.clone());
    })
    .await;
    let edge = relay(|c| {
        quick(c);
        c.auth.required = true;
        c.auth.authorities = vec![Authority { key: hexkey(&authority), rooms: vec![] }];
        c.floor = Some(floor.clone());
        let mut u = upstream_to(&home, Via::WebSocket, &[&["*"]]);
        u.token_sha256 = Some(peer_sha.clone());
        u.token_path = Some(peer_file.clone());
        c.upstream = vec![u];
    })
    .await;
    linked(&edge, 1).await;
    let t0 = now_us() - 10_000_000;
    let target = |s: &RelayServer| format!("//{}{}", s.websocket_addr.unwrap(), s.relay().config().path);
    let with = |s: &RelayServer, g: Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g, k.clone())], target: target(s), ..ClientOptions::default() };

    let (c, _) = connect(&home, Via::WebSocket, ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: clock_tok.clone() }], ..ClientOptions::default() }).await;
    let _ce = c.publish(room_track(room, &clock, "events"), name("0"), 1, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the clock's events");
    // The edge's watcher follows the clock's events first, so the edge reads the window as it
    // passes, over its link.
    let (we, _) = connect(&edge, Via::WebSocket, with(&edge, viewer_grant(&authority, room, &watcher_e, &[]), &watcher_e)).await;
    let _wec = we.subscribe(room_track(room, &clock, "events"), name("0"), Parameters::new()).await.expect("the edge's watcher, of the clock's events");
    let mut w = b"WPW1\x01".to_vec();
    w.extend_from_slice(&[1, 0b100, 3, 1]);
    w.extend_from_slice(&(now_us() - t0 + 60_000_000).to_le_bytes());
    w.extend_from_slice(&[0; 32]);
    w.extend_from_slice(&0u16.to_le_bytes());
    let wr = meta(&clock, room, 0, None, now_us() - t0, &w);
    let window = sha256(&wr[..wr.len() - 64]);
    send(&c, 1, 0, &msf(&wr)).await;
    let (fh, fe) = (home.relay().floor().unwrap(), edge.relay().floor().unwrap());
    assert!(until(STEP, || fh.holds_window(&room, &window) && fe.holds_window(&room, &window)).await, "both relays hold the clock's window");

    // The member publishes its events on the edge; a watcher on each relay follows them.
    let (pm, _) = connect(&edge, Via::WebSocket, with(&edge, viewer_grant(&authority, room, &member, &[("audience_input", "1")]), &member)).await;
    let _pme = pm.publish(room_track(room, &member, "events"), name("0"), 2, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the member's events");
    let events = room_track(room, &member, "events");
    assert!(until(STEP, || home.relay().namespaces().iter().any(|(n, _, _)| *n == events)).await, "the edge publishes the member's namespace upstream");
    let (wh, _) = connect(&home, Via::WebSocket, with(&home, viewer_grant(&authority, room, &watcher_h, &[]), &watcher_h)).await;
    let _whm = wh.subscribe(room_track(room, &member, "events"), name("0"), Parameters::new()).await.expect("the home watcher, of the member's events, through the edge");
    let _wem = we.subscribe(room_track(room, &member, "events"), name("0"), Parameters::new()).await.expect("the edge watcher, of the member's events");
    let vote = |seq: u64, prev: Option<&[u8]>, o: u8| {
        let mut b = b"WPA1\x01".to_vec();
        b.extend_from_slice(&window);
        b.extend_from_slice(&[3, o]);
        meta(&member, room, seq, prev, now_us() - t0, &b)
    };
    let first = vote(0, None, 1);
    let second = vote(1, Some(&first), 2);
    // Seq 1 first, on a stream of its own: the edge holds it for seq 0, counts seq 0, and then
    // refuses seq 1, past the member's power.
    send(&pm, 2, 1, &msf(&second)).await;
    assert!(until(STEP, || fe.waiting(&room) == 1).await, "the edge holds seq 1 for seq 0");
    send(&pm, 2, 0, &msf(&first)).await;
    let want = msf(&first);
    let (got_h, got_e) = tokio::join!(payloads_until(&wh, &want), payloads_until(&we, &want));
    let only_first = vec![msf(&first)];
    assert_eq!(got_e.iter().filter(|p| **p != msf(&wr)).cloned().collect::<Vec<_>>(), only_first, "the edge's watcher: seq 0 alone");
    assert_eq!(got_h, only_first, "the home watcher is given exactly what the edge's is");
    // Seq 1 is refused in its own object's task, woken once seq 0 is counted: nothing the
    // watchers were given waits for that, so the refusal is waited for, then counted.
    assert!(until(STEP, || fe.refused_inputs() >= 1).await, "the edge refuses seq 1");
    assert_eq!((fe.refused_inputs(), fh.refused_inputs()), (1, 0), "decided once, at the edge, where the member publishes");
    for x in [&c, &we, &pm, &wh] {
        x.close(0, "done");
    }
    let er = edge.relay().clone();
    edge.shutdown().await;
    home.shutdown().await;
    let _ = std::fs::remove_file(&peer_file);
    // The watchers were read for a second past seq 0, so a refusal counted and then forwarded
    // later than that would pass them. What the edge gave its watcher, counted once its
    // sessions ended, is seq 0 alone.
    assert!(until(STEP, || er.summaries().len() >= 3).await, "the edge's sessions' summaries");
    let member_events = joule_moq::draft22::names::render(&room_track(room, &member, "events"), Some(&name("0")));
    let given: u64 = er.summaries().iter().filter(|s| s.summary.tally.peer.is_none()).filter_map(|s| s.summary.tally.subscriber.tracks.get(&member_events).cloned()).map(|t| t.objects_forwarded + t.undelivered_at_close).sum();
    assert_eq!(given, 1, "the edge gave its watcher seq 0 alone");
}

/// A `WPT1` schedule (co-performance §6.1) under refuse: from `from`, one segment of two turns
/// of `len` µs, held by `holders` in turn.
fn schedule(t0: u64, from: u64, len: u32, holders: &[&SigningKey]) -> Vec<u8> {
    let mut b = b"WPT1\x01".to_vec();
    b.extend_from_slice(&t0.to_le_bytes());
    b.extend_from_slice(&from.to_le_bytes());
    b.push(1);
    b.extend_from_slice(&0u32.to_le_bytes());
    b.push(0);
    b.push(1);
    b.extend_from_slice(&from.to_le_bytes());
    b.extend_from_slice(&len.to_le_bytes());
    b.extend_from_slice(&2u16.to_le_bytes());
    b.push(0);
    b.push(holders.len() as u8);
    for h in holders {
        b.extend_from_slice(&h.verifying_key().to_bytes());
    }
    b
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_floor_is_held_once_where_the_performer_publishes() {
    timeout(Duration::from_secs(120), floor_once()).await.expect("in time");
}

async fn floor_once() {
    let room = [0xc7; 32];
    let clock = SigningKey::from_bytes(&[0xc8; 32]);
    let (x, y) = (SigningKey::from_bytes(&[0xc9; 32]), SigningKey::from_bytes(&[0xca; 32]));
    // The home relay holds the floor; the edge, where X publishes, holds none.
    let home = relay(|c| {
        quick(c);
        c.floor = Some(FloorConfig { room_keys: vec![hexkey(&clock)], ..FloorConfig::default() });
    })
    .await;
    let edge = relay(|c| {
        quick(c);
        c.upstream = vec![upstream_to(&home, Via::WebSocket, &[&["*"]])];
    })
    .await;
    linked(&edge, 1).await;
    // The clock's schedule on the home relay gives Y the turn now and for a minute.
    let t0 = now_us();
    let (c, _) = connect(&home, Via::WebSocket, ClientOptions::default()).await;
    let _ce = c.publish(room_track(room, &clock, "events"), name("0"), 1, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the clock's events");
    send(&c, 1, 0, &msf(&meta(&clock, room, 0, None, 0, &schedule(t0, 0, 60_000_000, &[&y, &x])))).await;
    let fh = home.relay().floor().unwrap();
    let audio = room_track(room, &x, "audio");
    assert!(until(STEP, || fh.place(&audio, now_us()).is_some()).await, "the home relay holds the schedule");
    assert_eq!(fh.place(&audio, now_us()), Some(joule_moq_relay::floor::Place::OutOfTurn(1)), "X is out of turn at the home relay");
    // X publishes on the edge, a viewer on the home relay follows: what the edge forwarded is
    // forwarded, not held to the floor a second time by its arrival a link later.
    let (px, _) = connect(&edge, Via::WebSocket, ClientOptions::default()).await;
    let _pa = px.publish(audio.clone(), name("0"), 5, Parameters::new(), track_props(16, None)).await.expect("PUBLISH X's audio on the edge");
    let (v, _) = connect(&home, Via::WebSocket, ClientOptions::default()).await;
    assert!(until(STEP, || home.relay().namespaces().iter().any(|(n, _, _)| *n == audio)).await, "the edge publishes X's namespace upstream");
    let _sv = v.subscribe(audio.clone(), name("0"), Parameters::new()).await.expect("the viewer, of X's audio, through the edge");
    for g in 0..3 {
        send(&px, 5, g, format!("x{g}").as_bytes()).await;
    }
    let got = collect(&v, STEP, |g| g.len() >= 3).await;
    assert_eq!(got.len(), 3, "every object the edge forwarded reaches the home relay's viewer");
    assert_eq!(fh.refused(), 0, "the home relay refused none of them");
    for k in [&c, &px, &v] {
        k.close(0, "done");
    }
    edge.shutdown().await;
    home.shutdown().await;
}

// ── A relay path, where it arrives ─────────────────────────────────────────

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_relay_path_is_checked_where_it_arrives() {
    timeout(Duration::from_secs(60), arrives()).await.expect("in time");
}

/// A client that declares a relay id, and proves it unless `prove` is false.
async fn as_relay(server: &RelayServer, key: &SigningKey, prove: bool) -> Result<Client, String> {
    as_relay_at(server, key, prove, joule_moq::draft::Draft::D22).await
}

/// [`as_relay`], the session at `draft` over the WebSocket binding.
async fn as_relay_at(server: &RelayServer, key: &SigningKey, prove: bool, draft: joule_moq::draft::Draft) -> Result<Client, String> {
    let url = format!("ws://{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
    let dial = || async {
        let conn = joule_moq_relay::transport::websocket::connect_draft(&url, joule_moq_relay::transport::mux::MuxConfig::default(), draft).await.map_err(|e| e.to_string())?;
        Ok::<_, String>(joule_moq_relay::transport::bridge::wrap(std::sync::Arc::new(conn), draft, joule_moq::draft22::Perspective::Client))
    };
    let opts = ClientOptions {
        extra_setup: vec![joule_moq::draft22::kvp::KeyValuePair::bytes(joule_moq_relay::cascade::RELAY_ID_OPTION, key.verifying_key().to_bytes().to_vec())],
        room_keys: if prove { vec![key.clone()] } else { Vec::new() },
        target: format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path),
        ..ClientOptions::default()
    };
    // The relay reads this SETUP before it sends its own: the id is proved over a challenge an
    // earlier SETUP of its gave, so the first session is given one and the second proves.
    Client::connect_proving(dial, opts).await
}

async fn closed_with(c: &Client) -> u64 {
    match timeout(STEP, c.conn.closed()).await.expect("the session closes in time") {
        joule_moq_relay::transport::TransportError::Closed { code, .. } => code,
        other => panic!("closed otherwise: {other}"),
    }
}

async fn arrives() {
    let server = relay(quick).await;
    let me = server.relay().relay_id();
    let peer = SigningKey::from_bytes(&[0x4e; 32]);
    let pid = peer.verifying_key().to_bytes();
    let path = |p: Vec<[u8; 32]>| Parameters::new().with(Parameter::RelayPath(p));
    // A relay whose id is proved: a path naming the relay that receives it is refused
    // relay_loop, one past the bound relay_hops_exceeded, and one whose last relay is not the
    // peer closes the session.
    let c = as_relay(&server, &peer, true).await.expect("a peer's SETUP");
    let r = c.subscribe(ns(&["live", "x"]), name("y"), path(vec![me, pid])).await.map(|_| ()).expect_err("refused");
    assert!(r.contains("DOES_NOT_EXIST") && r.contains("relay_loop"), "{r}");
    let r = c.subscribe(ns(&["live", "x"]), name("y"), path([[7u8; 32]; 8].into_iter().chain([pid]).collect())).await.map(|_| ()).expect_err("refused");
    assert!(r.contains("relay_hops_exceeded"), "{r}");
    let r = c.publish_namespace(ns(&["live", "x"]), path(vec![me, pid])).await.map(|_| ()).expect_err("refused");
    assert!(r.contains("UNINTERESTED") && r.contains("relay_loop"), "{r}");
    assert_eq!(server.relay().stats().relay_loops, 3);
    let _ = c.subscribe(ns(&["live", "x"]), name("y"), path(vec![[9; 32]])).await;
    assert_eq!(closed_with(&c).await, joule_moq::draft22::SessionError::PROTOCOL_VIOLATION.0, "a path whose last relay is not the peer");
    // A relay id declared without its proof is refused at SETUP, the reason stated.
    let refused = as_relay(&server, &peer, false).await.err().expect("a relay id without its proof");
    assert!(refused.contains("relay_id_unproved"), "{refused}");
    // RELAY_PATH from a session that negotiated none closes it PROTOCOL_VIOLATION (§9.20).
    let (plain, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let _ = plain.subscribe(ns(&["live", "x"]), name("y"), path(vec![pid])).await;
    assert_eq!(closed_with(&plain).await, joule_moq::draft22::SessionError::PROTOCOL_VIOLATION.0, "RELAY_PATH not negotiated");

    // The same at -23, where relays peer too: the path is negotiated by RELAY_ID, read by the
    // -23 codec and bridged, a loop refused and the hop limit held; and from a session that
    // negotiated none it is an unknown parameter, INVALID_PARAMETER (-23 §9.19).
    use joule_moq::draft::Draft;
    let c = as_relay_at(&server, &peer, true, Draft::D23).await.expect("a peer's SETUP at -23");
    assert_eq!(c.conn.draft(), Draft::D23);
    let loops = server.relay().stats().relay_loops;
    let r = c.subscribe(ns(&["live", "x"]), name("y"), path(vec![me, pid])).await.map(|_| ()).expect_err("refused");
    assert!(r.contains("DOES_NOT_EXIST") && r.contains("relay_loop"), "{r}");
    let r = c.subscribe(ns(&["live", "x"]), name("y"), path([[7u8; 32]; 8].into_iter().chain([pid]).collect())).await.map(|_| ()).expect_err("refused");
    assert!(r.contains("relay_hops_exceeded"), "{r}");
    let r = c.publish_namespace(ns(&["live", "x"]), path(vec![me, pid])).await.map(|_| ()).expect_err("refused");
    assert!(r.contains("UNINTERESTED") && r.contains("relay_loop"), "{r}");
    assert_eq!(server.relay().stats().relay_loops, loops + 3, "each refused for its path, at -23");
    assert!(c.conn.notes().iter().all(|n| !n.contains("RELAY_PATH")), "the bridge carried the path, and dropped none: {:?}", c.conn.notes());
    let _ = c.subscribe(ns(&["live", "x"]), name("y"), path(vec![[9; 32]])).await;
    assert_eq!(closed_with(&c).await, joule_moq::draft23::codes::SessionError::INVALID_PARAMETER.0, "a path whose last relay is not the peer, at -23");
    let (plain, _) = connect(&server, Via::WsDraft(Draft::D23), ClientOptions::default()).await;
    let _ = plain.subscribe(ns(&["live", "x"]), name("y"), path(vec![pid])).await;
    assert_eq!(closed_with(&plain).await, joule_moq::draft23::codes::SessionError::INVALID_PARAMETER.0, "RELAY_PATH not negotiated, at -23");
    // At -18 and -16 there is no form for it: a relay id declared there is left unread, the
    // session an ordinary one, and its proof alone makes it no peer.
    let declares = ClientOptions { extra_setup: vec![joule_moq::draft22::kvp::KeyValuePair::bytes(joule_moq_relay::cascade::RELAY_ID_OPTION, pid.to_vec())], ..ClientOptions::default() };
    let (old, _ep) = connect(&server, Via::QuicDraft(Draft::D18), declares).await;
    let r = old.subscribe(ns(&["live", "x"]), name("y"), path(vec![pid])).await.map(|_| ()).expect_err("a path at -18");
    assert!(old.conn.notes().iter().any(|n| n.contains("RELAY_PATH")) || r.contains("NOT_SUPPORTED"), "refused by name at -18: {r}; {:?}", old.conn.notes());
    old.close(0, "done");
    server.shutdown().await;
}

// ── SUBSCRIBE_TRACKS across relays ─────────────────────────────────────────

/// SUBSCRIBE_TRACKS for `prefix` on `c`, its answer, and its stream.
async fn tracks_request(c: &Client, prefix: TrackNamespace) -> (ControlMessage, joule_moq_relay::wire::ReqTx, ReqRx) {
    let id = c.next_request_id();
    let (tx, mut rx) = c.request(&ControlMessage::SubscribeTracks(NamespaceRequest { request_id: id, prefix, params: Parameters::new() })).await.expect("the request goes");
    let answer = Client::response(&mut rx).await.expect("an answer");
    (answer, tx, rx)
}

/// The next PUBLISH the relay opens to `c`, within `wait`, accepted (PUBLISH_OK); `None` if
/// none comes.
async fn publish_offered(c: &Client, wait: Duration) -> Option<(Publish, joule_moq_relay::wire::ReqTx, ReqRx)> {
    let (tx, rx, first) = c.next_request(wait).await?;
    let ControlMessage::Publish(p) = first else { panic!("the relay opened {} where a PUBLISH was expected", first.name()) };
    tx.send(&ControlMessage::RequestOk(RequestOk { params: Parameters::new(), track_properties: Default::default() })).await.expect("PUBLISH_OK goes");
    Some((p, tx, rx))
}

/// One object of group `g`, on the track published under `alias`.
async fn one_object(c: &Client, alias: u64, g: u64, payload: &[u8]) {
    let h = SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(128), properties: false, end_of_group: false, first_object: true };
    c.send_subgroup(h, &[SubgroupObject::normal(0, payload.to_vec())]).await.expect("the object goes");
}

/// Whether an object of the subscription `alias` carrying `payload` reaches `c`.
async fn reaches(c: &Client, alias: u64, payload: &[u8]) -> bool {
    let carries = |g: &Got| g.header.track_alias == alias && g.objects.iter().any(|o| o.payload == payload);
    collect(c, STEP, |got| got.iter().any(carries)).await.iter().any(carries)
}

/// SUBSCRIBE_TRACKS across relays (-22 §7.6, -23 §3.6): a subscriber of the edge asks for
/// every track; the edge takes the request to its upstream link, and the tracks the home relay
/// holds under what the link routes, now and later, come back as PUBLISHes the edge accepts and
/// offers on, their objects arriving on the edge's PUBLISH. A track the link does not route is
/// refused UNINTERESTED at the edge and offered to no one. Two requests of the edge that ask for
/// the same thing of the link are one SUBSCRIBE_TRACKS on it; the last to go cancels it, and the
/// home relay offers the edge nothing more.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn subscribe_tracks_reaches_an_upstream_relay() {
    timeout(Duration::from_secs(120), tracks_across()).await.expect("in time");
}

async fn tracks_across() {
    let home = relay(quick).await;
    // The link routes live/<any>: a prefix with a wildcard field, so the edge asks the home
    // relay for the request's own prefix and holds what comes back to the link's.
    let edge = relay(|c| {
        quick(c);
        c.upstream = vec![upstream_to(&home, Via::Quic, &[&["live", "*"]])];
    })
    .await;
    linked(&edge, 1).await;
    let (hr, er) = (home.relay().clone(), edge.relay().clone());
    let (pa, _ea) = connect(&home, Via::Quic, ClientOptions::default()).await;
    let (_a, _arx) = pa.publish(ns(&["live", "a"]), name("cam"), 71, Parameters::new(), Default::default()).await.expect("PUBLISH live/a on the home relay");
    let (_x, _xrx) = pa.publish(ns(&["other", "x"]), name("cam"), 72, Parameters::new(), Default::default()).await.expect("PUBLISH other/x on the home relay");

    // A subscriber of the edge asks for every track (a prefix of no field).
    let (se, _es) = connect(&edge, Via::Quic, ClientOptions::default()).await;
    let (answer, se_tx, _se_rx) = tracks_request(&se, ns(&[])).await;
    assert!(matches!(answer, ControlMessage::RequestOk(_)), "{answer:?}");
    let (first, _ftx, _frx) = publish_offered(&se, STEP).await.expect("a PUBLISH, at the edge, of the home relay's track");
    assert_eq!((first.namespace.clone(), first.name.clone()), (ns(&["live", "a"]), name("cam")));
    one_object(&pa, 71, 0, b"home-a-0").await;
    assert!(reaches(&se, first.track_alias, b"home-a-0").await, "an object of the home relay's track reaches the edge's subscriber");

    // A track the home relay gains later, under the link's prefix: offered too.
    let (_b, _brx) = pa.publish(ns(&["live", "b"]), name("cam"), 73, Parameters::new(), Default::default()).await.expect("PUBLISH live/b on the home relay");
    let (second, _stx, _srx) = publish_offered(&se, STEP).await.expect("a PUBLISH of the track published later");
    assert_eq!((second.namespace.clone(), second.name.clone()), (ns(&["live", "b"]), name("cam")));
    one_object(&pa, 73, 0, b"home-b-0").await;
    assert!(reaches(&se, second.track_alias, b"home-b-0").await, "an object of the later track reaches the edge's subscriber");
    // Nothing the link does not route.
    assert!(publish_offered(&se, Duration::from_millis(800)).await.is_none(), "other/x, which the link does not route, is not offered at the edge");
    assert_eq!(er.largest(&(ns(&["other", "x"]), name("cam"))), None, "nor held there");

    // A second request of the edge for the same prefix: the one SUBSCRIBE_TRACKS on the link.
    let (se2, _es2) = connect(&edge, Via::Quic, ClientOptions::default()).await;
    let (answer, se2_tx, _se2_rx) = tracks_request(&se2, ns(&[])).await;
    assert!(matches!(answer, ControlMessage::RequestOk(_)), "{answer:?}");
    assert_eq!(er.tracks_up(), vec![(0, ns(&[]), 2)], "one SUBSCRIBE_TRACKS on the link, held by both requests");
    assert_eq!(hr.subscribe_tracks_held().len(), 1, "the home relay holds one SUBSCRIBE_TRACKS of the edge's");

    // Both requests cancelled: the edge cancels its own on the link, and a track the home relay
    // gains after that does not reach the edge.
    for tx in [&se_tx, &se2_tx] {
        tx.reset(joule_moq::draft22::StreamResetCode::CANCELLED.0).await;
    }
    assert!(until(STEP, || er.tracks_up().is_empty()).await, "the edge lets its SUBSCRIBE_TRACKS go: {:?}", er.tracks_up());
    assert!(until(STEP, || hr.subscribe_tracks_held().is_empty()).await, "the home relay's copy of it is cancelled: {:?}", hr.subscribe_tracks_held());
    let (_c, _crx) = pa.publish(ns(&["live", "c"]), name("cam"), 74, Parameters::new(), Default::default()).await.expect("PUBLISH live/c on the home relay");
    one_object(&pa, 74, 0, b"home-c-0").await;
    assert!(until(STEP, || hr.largest(&(ns(&["live", "c"]), name("cam"))).is_some()).await, "the home relay holds live/c");
    tokio::time::sleep(Duration::from_millis(500)).await;
    assert_eq!(er.largest(&(ns(&["live", "c"]), name("cam"))), None, "a track published after the cancellation does not reach the edge");
    // The subscriptions made before stand (-23 §3.6): the edge still holds live/b.
    one_object(&pa, 73, 1, b"home-b-1").await;
    assert!(reaches(&se, second.track_alias, b"home-b-1").await, "the subscription the cancelled request made stands");
    for c in [&pa, &se, &se2] {
        c.close(0, "done");
    }
    edge.shutdown().await;
    home.shutdown().await;
}