//! The room's own source over MOQT (`[source]`, `src/source.rs`), through the relay and the HTTP
//! egress gateway (`joule-moq-http`), on loopback. The relay holds the source's video and sound
//! at its door as `wai dev`'s does (`[[receipts.door]]`, `trust_field = 1`), and the door reads
//! the source's running receipts: every object passes on its group's run, or, of a group's last,
//! on its group's receipts where they came first; none waits for the receipts for want of a
//! running receipt, and none is refused.
//!
//! - **Stamps through the relay.** A viewer of the room, with a grant the service issues,
//!   subscribes to the source's video and sound with Sync Support and holds every object it
//!   receives to its stamps: CaptureTimestamp and PlayoutDelay read inside Immutable Properties
//!   under the configured types; the delay the catalog's, within its bound; the instant after
//!   the room began; the capture less the object's own media time (its `tfdt` and durations)
//!   one constant across both tracks, so the source states one capture clock; a HopTime of the
//!   relay's clock, between the capture and the arrival; and the object here before its
//!   instant. Every group read is whole, its objects in order from 0.
//! - **Running receipts through the gateway.** The gateway, a viewer too, follows the source's
//!   video and sound by their hinted parts; each part served while its group is open is checked
//!   against the run served beside it (each record signed by the source, continuing the one
//!   before, its last stating the part's bytes), and every part, once its group's receipts
//!   have come, against them. Nothing is withheld or refused, no running receipt either (the
//!   gateway reads the source's events track in its order, the fill that joins it before the
//!   subscription's objects), and the gateway's trace checks.
//! - **Real time** (`a_thirty_frame_room_is_served_in_real_time`): thirty seconds of 30 fps
//!   video and sound through the door, every part served by the gateway within [`BOUND_MS`] of
//!   its capture stamp and none withheld or missing. With it, how long the door held each
//!   object, and how long after it the group's receipts came: what a door that reads no running
//!   receipt holds it for. The measurement is refused on a host whose load is above a quarter
//!   of its processors, and says so; a figure is this host's on this day.
//!
//! Every wait is bounded, every listener binds port 0, and everything stops with its test.

use ed25519_dalek::SigningKey;
use joule_moq::draft22::{Parameter, Parameters, StreamHeader, UniEvent};
use joule_moq::{TrackName, TrackNamespace};
use joule_moq_http::{Config as GatewayConfig, Gateway};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::transport::mux::MuxConfig;
use joule_moq_relay::transport::{websocket, SharedConnection};
use joule_moq_room::dev::{DevRoom, Fixture};
use joule_moq_room::epoch::Role;
use joule_moq_room::source::{chunk_span, Source};
use joule_moq_room::tempo::Timing;
use serde_json::Value;
use std::collections::{BTreeMap, BTreeSet};
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::time::timeout;
use wai::live_egress::{check_trace, running, Params};
use wai::live_events::anchor_track;

const STEP: Duration = Duration::from_secs(20);
const PLAYOUT_MS: u32 = 1000;
const MAX_DELAY_MS: u32 = 2000;
/// The bound of the real-time test: a part is served this soon after its capture stamp.
const BOUND_MS: u64 = 500;

fn package() -> PathBuf {
    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../../../wai/pack-conformance/multi/multi-opus-01/package")
}

fn now_ms() -> u64 {
    joule_moq_room::now_us() / 1000
}

async fn start() -> DevRoom {
    let extra = format!("[source]\npackage = {:?}\nplayout_delay_ms = {PLAYOUT_MS}\nmax_delay_ms = {MAX_DELAY_MS}\n", package());
    DevRoom::start_with(&Fixture { extra, doors: true, ..Fixture::default() }).await.expect("the room and its relay")
}

/// A track's group's holds at the door: each object, what it passed on, and whether it waited.
type ByGroup<'a> = BTreeMap<(&'a str, u64), Vec<(u64, &'static str, bool)>>;

/// The door passed every object of the source on its group's run or its group's receipts, and
/// none waited for the receipts for want of a running receipt: nothing was refused, no running
/// receipt was refused and no run contradicted, and each object that waited and passed on its
/// group's receipts is one whose running receipt the door read, taken into the run or read once
/// the receipts had closed it. Such an object is one of a group's last: the receipts are sent right
/// after the group's last object, at a transport priority above the running receipts', and once
/// they have come nothing more of the group is held on its run (WAI live-http-egress §3.2). So an
/// object whose running receipt had not reached the door when they came, or whose wait the door
/// took up again only after they came (a task woken late on a loaded host: the door reads the
/// group's receipts first), passes on them; with a final object of one frame (33 ms) after it, a
/// group's last but one meets them too.
fn door_passed_on_runs(dev: &DevRoom) -> Vec<joule_moq_relay::door::Hold> {
    let s = dev.relay.relay().stats();
    assert!(s.door_admitted_on_run > 0 && s.door_running_taken >= s.door_admitted_on_run, "{s:?}");
    assert_eq!((s.refused_at_door, s.groups_dropped_at_door, s.door_running_refused, s.door_runs_contradicted), (0, 0, 0, 0), "{s:?}");
    let door = dev.relay.relay().door();
    let holds: Vec<_> = door.holds().into_iter().filter(|h| h.track.ends_with("-video--0") || h.track.ends_with("-audio--0")).collect();
    assert!(!holds.is_empty(), "the door held the source's objects");
    let read: BTreeSet<(String, u64, u64)> = door.runs_read().into_iter().filter(|r| r.outcome == "taken" || r.outcome == "ignored").map(|r| (r.track, r.group, r.object)).collect();
    let waited: Vec<_> = holds
        .iter()
        .filter(|h| h.by != "run" && (h.by != "receipts" || (h.waited && !read.contains(&(h.track.clone(), h.group, h.object)))))
        .map(|h| (h.track.rsplit('-').nth(2).unwrap_or(""), h.group, h.object, h.by, h.held))
        .collect();
    assert!(waited.is_empty(), "{} of {} objects waited for their group's receipts with no running receipt read, or were refused: {waited:?}", waited.len(), holds.len());
    // So in each group the objects passed on its run come first and those passed on its
    // receipts after them, and a group whose receipts an object waited for passed objects on its
    // run before: runs that came too late for every object would leave none.
    let mut by_group: ByGroup = BTreeMap::new();
    for h in &holds {
        by_group.entry((h.track.as_str(), h.group)).or_default().push((h.object, h.by, h.waited));
    }
    for (k, mut v) in by_group {
        v.sort_unstable();
        let first = v.iter().position(|x| x.1 != "run").unwrap_or(v.len());
        assert!(v[first..].iter().all(|x| x.1 == "receipts"), "{k:?}: an object passed on its run after one passed on its group's receipts: {v:?}");
        assert!(first > 0 || !v.iter().any(|x| x.2), "{k:?}: every object waited for its group's receipts: {v:?}");
    }
    holds
}

/// A viewer's grant for a fresh key, as `POST /grant` would issue it.
fn viewer(dev: &DevRoom) -> (Vec<u8>, SigningKey) {
    let key = SigningKey::from_bytes(&{ use rand_core_06::RngCore as _; let mut secret = [0u8; 32]; rand_core_06::OsRng.fill_bytes(&mut secret); secret });
    let issued = dev.svc.state.lock().unwrap().epoch.grant(&dev.svc.keys, key.verifying_key().to_bytes(), Role::Viewer, joule_moq_room::now_s(), None).expect("a viewer's grant");
    (issued.record, key)
}

async fn session(dev: &DevRoom, grant: (Vec<u8>, SigningKey)) -> Client {
    let url = dev.relay_url.clone();
    let dial = || {
        let url = url.clone();
        async move {
            let c = timeout(STEP, websocket::connect(&url, MuxConfig::default())).await.map_err(|_| "no session".to_string())?.map_err(|e| e.to_string())?;
            Ok::<SharedConnection, String>(Arc::new(c))
        }
    };
    let opts = ClientOptions { grants: vec![grant], target: joule_moq_room::proof_target(&dev.relay_url), ..ClientOptions::default() };
    Client::connect_proving(dial, opts).await.expect("a viewer's session")
}

/// The room, the source's key and the generation it publishes in, once it publishes.
async fn publishing(dev: &DevRoom) -> ([u8; 32], [u8; 32], u64) {
    let src = dev.svc.source.clone().expect("a source");
    let until = Instant::now() + STEP;
    loop {
        if let Some(g) = src.publishing() {
            let room = dev.svc.state.lock().unwrap().epoch.id;
            return (room, Source::key(&dev.svc.keys.clock).verifying_key().to_bytes(), g);
        }
        assert!(Instant::now() < until, "the source did not start publishing");
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
}

/// `GET` over a connection of its own: the status, the body and when its last byte came.
async fn get(addr: SocketAddr, target: &str) -> (u16, Vec<u8>, u64) {
    let mut s = timeout(Duration::from_secs(5), tokio::net::TcpStream::connect(addr)).await.expect("in time").expect("a connection");
    s.write_all(format!("GET /{target} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n\r\n").as_bytes()).await.expect("a request");
    let mut buf = Vec::new();
    let mut chunk = vec![0u8; 64 << 10];
    loop {
        if let Some(end) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
            let text = String::from_utf8_lossy(&buf[..end]).to_lowercase();
            let status = text.split(' ').nth(1).and_then(|c| c.parse().ok()).expect("a status");
            if let Some(n) = text.lines().find_map(|l| l.strip_prefix("content-length: ")).and_then(|n| n.trim().parse::<usize>().ok())
                && buf.len() >= end + 4 + n
            {
                return (status, buf[end + 4..end + 4 + n].to_vec(), now_ms());
            }
        }
        let n = timeout(Duration::from_secs(30), s.read(&mut chunk)).await.unwrap_or_else(|_| panic!("{target}: no answer in time")).expect("a read");
        if n == 0 {
            let end = buf.windows(4).position(|w| w == b"\r\n\r\n").expect("a head");
            let status = String::from_utf8_lossy(&buf[..end]).split(' ').nth(1).and_then(|c| c.parse().ok()).expect("a status");
            return (status, buf[end + 4..].to_vec(), now_ms());
        }
        buf.extend_from_slice(&chunk[..n]);
    }
}

async fn until_ok(addr: SocketAddr, target: &str) -> String {
    let until = Instant::now() + STEP;
    loop {
        let (status, body, _) = get(addr, target).await;
        if status == 200 {
            return String::from_utf8(body).expect("text");
        }
        assert!(Instant::now() < until, "{target}: still {status}");
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
}

/// A media playlist's live edge: the open segment's number, its parts listed, the hinted part.
fn edge(playlist: &str) -> (u64, u64, Option<String>) {
    let (mut msn, mut parts, mut hint) = (0, 0, None);
    let attr = |l: &str| l.split("URI=\"").nth(1).and_then(|r| r.split('"').next()).map(str::to_string);
    for l in playlist.lines() {
        if let Some(n) = l.strip_prefix("#EXT-X-MEDIA-SEQUENCE:") {
            msn = n.parse().expect("a number");
        } else if l.starts_with("#EXT-X-PART:") {
            parts += 1;
        } else if l.starts_with("#EXT-X-PRELOAD-HINT:") {
            hint = attr(l);
        } else if !l.starts_with('#') && !l.is_empty() {
            msn += 1;
            parts = 0;
        }
    }
    (msn, parts, hint)
}

/// The parts a media playlist lists, in order.
fn listed_parts(playlist: &str) -> Vec<String> {
    playlist.lines().filter(|l| l.starts_with("#EXT-X-PART:")).filter_map(|l| l.split("URI=\"").nth(1).and_then(|r| r.split('"').next()).map(|u| u.trim_start_matches("../").to_string())).collect()
}

fn try_group_object(uri: &str) -> Option<(u64, u64)> {
    let (g, o) = uri.strip_prefix("p/").and_then(|r| r.strip_suffix(".m4s")).and_then(|r| r.split_once('.'))?;
    Some((g.parse().ok()?, o.parse().ok()?))
}

fn group_object(uri: &str) -> (u64, u64) {
    try_group_object(uri).expect("a part URI")
}

struct Room {
    room: [u8; 32],
    owner: [u8; 32],
}

impl Room {
    fn place(&self, track: &str) -> (u64, [u8; 32]) {
        let kind = track.split('/').next().expect("a kind");
        (if kind == "video" { 100 } else { 101 }, anchor_track(&format!("{}-{}-{kind}", hex::encode(self.room), hex::encode(self.owner)), "0"))
    }

    /// A part checked against the run served beside it; `false` where it has none (it was held
    /// on its group's receipts).
    async fn verify_running(&self, addr: SocketAddr, track: &str, uri: &str, bytes: &[u8]) -> bool {
        let (g, o) = group_object(uri);
        let (status, body, _) = get(addr, &format!("{track}/{uri}.running")).await;
        if status == 404 {
            return false;
        }
        assert_eq!(status, 200, "{track}/{uri}: its run");
        let v: Value = serde_json::from_slice(&body).expect("JSON");
        let records = v["records"].as_array().expect("records");
        assert_eq!(records.len() as u64, o + 1, "{uri}: one record for each position to its own");
        let (alias, anchor) = self.place(track);
        let mut leaves = Vec::new();
        let mut last = None;
        for (i, r) in records.iter().enumerate() {
            let record = hex::decode(r.as_str().expect("hex")).expect("hex");
            let place = running::Place { room: &self.room, owner: &self.owner, track: &anchor, track_alias: alias, group: g, object: i as u64 };
            let r = running::check(&record, &place).unwrap_or_else(|why| panic!("{uri}: record {i}: {why}"));
            assert_eq!(r.prior, wai::jwp_delivery::root(&leaves), "{uri}: record {i} continues the run");
            leaves.push(r.leaf);
            last = Some(r);
        }
        assert_eq!(last.expect("a record").content_hash, wai::http_binding::content_hash(bytes), "{uri}: the bytes are what the source stated");
        true
    }

    /// A part checked against its receipt index and group receipt, under the source's key.
    async fn verify_receipts(&self, addr: SocketAddr, track: &str, uri: &str, bytes: &[u8]) {
        use wai::http_binding::receipts::{read_group_file, ReceiptIndex};
        let (status, body, _) = get(addr, &format!("{track}/{uri}.receipts")).await;
        assert_eq!(status, 200, "{track}/{uri}: its receipt index");
        let index = ReceiptIndex::from_bytes(&body).expect("an index");
        let item = index.delivery[0].as_ref().expect("a delivery item");
        let (status, file, _) = get(addr, &format!("{track}/p/rcpt/{}.json", hex::encode(item.property.group_receipt))).await;
        assert_eq!(status, 200, "{uri}: its group receipt");
        let group = read_group_file(&item.property.group_receipt, &file).expect("a group receipt under its name");
        assert_eq!(group.signer_pubkey, self.owner);
        item.object_receipt.accept(&group, bytes).unwrap_or_else(|e| panic!("{uri}: {e}"));
    }
}

async fn gateway(dev: &DevRoom, room: &Room) -> Gateway {
    let src = dev.svc.source.clone().expect("a source");
    let longest = src.tracks.iter().map(|t| t.longest_group_ms()).max().unwrap_or(1000);
    let params = Params { target_duration_s: longest.div_ceil(1000).max(1), part_target_ms: 250, window: 6, latency_target_ms: 1500 };
    let cfg = GatewayConfig {
        relay_url: dev.relay_url.clone(),
        room: room.room,
        owner: hex::encode(room.owner),
        publisher: None,
        listen: "127.0.0.1:0".parse().unwrap(),
        params,
        grant: Some(viewer(dev)),
        record_trace: true,
        catalog_wait: Duration::from_secs(10),
        breaks: None,
    };
    let until = Instant::now() + STEP;
    loop {
        match Gateway::start(cfg.clone()).await {
            Ok(g) => return g,
            Err(e) => {
                assert!(Instant::now() < until, "the gateway: {e}");
                tokio::time::sleep(Duration::from_millis(100)).await;
            }
        }
    }
}

/// One object as a viewer received it.
struct Got {
    track: &'static str,
    group: u64,
    object: u64,
    timing: Timing,
    span: (u64, u64),
    arrived_ms: u64,
}

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

async fn stamped_and_running() {
    let dev = start().await;
    dev.relay.relay().door().record_holds();
    let (room, owner, _) = publishing(&dev).await;
    let src = dev.svc.source.clone().expect("a source");
    let types = src.cfg.types();

    // GET /room names the source.
    let (status, body, _) = get(dev.http, "room").await;
    assert_eq!(status, 200);
    let v: Value = serde_json::from_slice(&body).expect("JSON");
    assert_eq!(v["source"]["principal"].as_str(), Some(hex::encode(owner).as_str()));
    // The room's grants, the source's among them, each within its role, as a page checks them
    // before it enters (live-rooms §4.3).
    let descriptor = wai::live_room::Descriptor::parse(&hex::decode(v["descriptor"].as_str().expect("a descriptor")).expect("hex")).expect("the descriptor reads");
    let grants: Vec<Vec<u8>> = v["grants"].as_array().expect("grants").iter().map(|g| hex::decode(g.as_str().expect("hex")).expect("hex")).collect();
    let held = wai::live_room::check_grants(&descriptor, &descriptor.room_id(), &grants).unwrap_or_else(|(i, r)| panic!("grant {i}: {}", r.code()));
    assert!(held.iter().any(|g| g.grant.principal == owner), "the source's grant is among the room's");
    // And the types its stamps are written under, which a page reads them by.
    assert_eq!((v["source"]["types"]["playout_delay"].as_u64(), v["source"]["types"]["capture_timestamp"].as_u64(), v["source"]["types"]["hop_time"].as_u64()), (Some(types.playout_delay), Some(types.capture_timestamp), Some(types.hop_time)));

    // A viewer subscribes to both tracks with Sync Support.
    let c = Arc::new(session(&dev, viewer(&dev)).await);
    let mut aliases = BTreeMap::new();
    let mut subscriptions = Vec::new();
    for kind in ["video", "audio"] {
        let ns = TrackNamespace::from_strs([hex::encode(room).as_str(), hex::encode(owner).as_str(), kind]);
        let until = Instant::now() + STEP;
        let ok = loop {
            match c.subscribe(ns.clone(), TrackName::from_str("0"), Parameters::new().with(Parameter::SyncSupport)).await {
                Ok(s) => break s,
                Err(e) => {
                    assert!(Instant::now() < until, "SUBSCRIBE {kind}: {e}");
                    tokio::time::sleep(Duration::from_millis(100)).await;
                }
            }
        };
        aliases.insert(ok.0.track_alias, if kind == "video" { "video" } else { "audio" });
        subscriptions.push(ok);
    }
    let got: Arc<Mutex<Vec<Got>>> = Arc::default();
    let opened: Arc<Mutex<BTreeSet<(&'static str, u64)>>> = Arc::default();
    let reader = {
        let (c, got, opened, aliases) = (c.clone(), got.clone(), opened.clone(), aliases.clone());
        tokio::spawn(async move {
            while let Some(mut r) = c.next_stream(Duration::from_secs(30)).await {
                let (got, opened, aliases) = (got.clone(), opened.clone(), aliases.clone());
                tokio::spawn(async move {
                    let Ok(StreamHeader::Subgroup(h)) = r.start().await else { return };
                    let Some(track) = aliases.get(&h.track_alias).copied() else { return };
                    opened.lock().unwrap().insert((track, h.group_id));
                    while let Ok(Some(UniEvent::SubgroupObject(o))) = r.next().await {
                        let timing = Timing::read_as(&o.properties, &types).unwrap_or_else(|e| panic!("{track} {}/{}: {e}", h.group_id, o.object_id));
                        let span = chunk_span(&o.payload).expect("a CMAF chunk");
                        got.lock().unwrap().push(Got { track, group: h.group_id, object: o.object_id, timing, span, arrived_ms: now_ms() });
                    }
                });
            }
        })
    };

    // The gateway, meanwhile, follows the video by its hinted parts.
    let r = Room { room, owner };
    let gw = gateway(&dev, &r).await;
    let addr = gw.addr;
    let (mut served, mut on_runs) = (0, 0);
    for track in ["video/0", "audio/0"] {
        let mut playlist = until_ok(addr, &format!("{track}/media.m3u8")).await;
        for _ in 0..16 {
            let (msn, parts, hint) = edge(&playlist);
            let hint = hint.expect("a live playlist hints its next part");
            let reload = format!("{track}/media.m3u8?_HLS_msn={msn}&_HLS_part={parts}");
            let part = format!("{track}/{hint}");
            let ((status, body, _), (rs, next, _)) = tokio::join!(get(addr, &part), get(addr, &reload));
            assert_eq!(rs, 200, "{reload}");
            playlist = String::from_utf8(next).expect("text");
            if status == 200 {
                served += 1;
                if r.verify_running(addr, track, &hint, &body).await {
                    on_runs += 1;
                } else {
                    r.verify_receipts(addr, track, &hint, &body).await;
                }
            } else {
                assert_eq!(status, 404, "{hint}");
            }
        }
    }
    assert!(served >= 24 && on_runs * 10 >= served * 8, "{served} parts served, {on_runs} of them held on their runs");

    // Every part of every closed segment against its group's receipts.
    let mut closed = 0;
    for track in ["video/0", "audio/0"] {
        let playlist = until_ok(addr, &format!("{track}/media.m3u8")).await;
        for s in playlist.lines().filter(|l| l.starts_with("s/")) {
            let (status, list, _) = get(addr, &format!("{track}/{s}.parts")).await;
            assert_eq!(status, 200, "{track}/{s}.parts");
            let list: Value = serde_json::from_slice(&list).expect("JSON");
            for p in list["parts"].as_array().expect("parts") {
                let uri = p["uri"].as_str().expect("a URI").trim_start_matches("../");
                let (status, bytes, _) = get(addr, &format!("{track}/{uri}")).await;
                assert_eq!(status, 200, "{track}/{uri}");
                r.verify_receipts(addr, track, uri, &bytes).await;
                closed += 1;
            }
        }
    }
    assert!(closed >= 10, "{closed} parts of closed segments verified against their receipts");
    assert!(gw.withheld().is_empty(), "{:?}", gw.withheld());
    assert!(gw.running_refused().is_empty(), "{:?}", gw.running_refused());
    let trace = gw.trace().expect("a trace");
    let checked = check_trace(&trace).unwrap_or_else(|why| panic!("the gateway's own trace: {why}"));
    assert!(checked.withheld == 0 && checked.running_refused == 0, "{checked:?}");

    // What the viewer received, held to its stamps.
    reader.abort();
    let got = std::mem::take(&mut *got.lock().unwrap());
    let opened = opened.lock().unwrap().clone();
    assert!(got.len() >= 40, "{} objects received", got.len());
    let t0_ms = dev.svc.state.lock().unwrap().epoch.t0_us / 1000;
    let scale = |track: &str| src.tracks.iter().find(|t| t.kind == track).expect("a track").timescale as u64;
    let mut clock = None;
    let mut by_group: BTreeMap<(&str, u64), Vec<u64>> = BTreeMap::new();
    for o in &got {
        let at = format!("{} {}/{}", o.track, o.group, o.object);
        let (capture, delay, hop) = (o.timing.capture_ms.expect("a CaptureTimestamp"), o.timing.delay_ms.expect("a PlayoutDelay"), o.timing.hop_ms.expect("a HopTime"));
        assert_eq!(delay, PLAYOUT_MS, "{at}: the delay the catalog states");
        assert!(delay <= MAX_DELAY_MS && capture + delay as u64 >= t0_ms, "{at}");
        let media_end = (o.span.0 + o.span.1) * 1000 / scale(o.track);
        let c = capture - media_end;
        assert_eq!(*clock.get_or_insert(c), c, "{at}: one capture clock across the source");
        assert!(hop + 2 >= capture && hop <= o.arrived_ms, "{at}: a HopTime of the relay's clock, {capture} <= {hop} <= {}", o.arrived_ms);
        assert!(o.arrived_ms <= capture + delay as u64, "{at}: arrived {} ms after its instant", o.arrived_ms - capture - delay as u64);
        by_group.entry((o.track, o.group)).or_default().push(o.object);
    }
    // Each group whose stream was read from its start, after the first of each track, is whole
    // and in order (the last may still be open). A track's first group may have been open when
    // the viewer's SUBSCRIBE landed, and is sent from the object then: in order from there.
    let firsts: BTreeMap<&str, u64> = by_group.keys().fold(BTreeMap::new(), |mut m, (t, g)| {
        m.entry(*t).or_insert(*g);
        m
    });
    let mut whole = 0;
    for ((track, g), mut objects) in by_group {
        objects.sort_unstable();
        assert!(opened.contains(&(track, g)));
        let from = if firsts.get(track) == Some(&g) { objects[0] } else { 0 };
        assert_eq!(objects, (from..from + objects.len() as u64).collect::<Vec<_>>(), "{track} {g}: its objects in order from {from}");
        whole += 1;
    }
    assert!(whole >= 4, "{whole} groups read");
    door_passed_on_runs(&dev);

    drop(gw);
    drop(subscriptions);
    c.close(0, "done");
    dev.stop().await;
}

/// The host's one-minute load for each processor, where the system states it.
fn load_per_core() -> Option<f64> {
    let one: f64 = match std::fs::read_to_string("/proc/loadavg") {
        Ok(s) => s.split_whitespace().next()?.parse().ok()?,
        Err(_) => {
            let out = std::process::Command::new("sysctl").args(["-n", "vm.loadavg"]).output().ok()?;
            String::from_utf8_lossy(&out.stdout).trim().trim_start_matches('{').split_whitespace().next()?.parse().ok()?
        }
    };
    Some(one / std::thread::available_parallelism().ok()?.get() as f64)
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_thirty_frame_room_is_served_in_real_time() {
    match load_per_core() {
        Some(l) if l <= 0.25 => {}
        other => {
            println!("not measured: the host's load for each processor is {other:?}, above a quarter; a real-time bound is measured on an idle host only");
            return;
        }
    }
    timeout(Duration::from_secs(120), real_time()).await.expect("in time");
}

async fn real_time() {
    let dev = start().await;
    dev.relay.relay().door().record_holds();
    let (room, owner, _) = publishing(&dev).await;
    let src = dev.svc.source.clone().expect("a source");
    assert_eq!(src.tracks[0].framerate, 30);
    let r = Room { room, owner };
    let gw = gateway(&dev, &r).await;
    let addr = gw.addr;
    let began = Instant::now();
    let mut tasks = Vec::new();
    for track in ["video/0", "audio/0"] {
        tasks.push(tokio::spawn(async move {
            // Each part this client was served live, with when its last byte came.
            let mut served: Vec<((u64, u64), u64)> = Vec::new();
            let mut playlist = until_ok(addr, &format!("{track}/media.m3u8")).await;
            while began.elapsed() < Duration::from_secs(30) {
                // As a client does: a part the playlist lists that it has not fetched yet (one
                // whose hint it never had, at a group's start) it fetches now.
                if let Some(&(last, _)) = served.last() {
                    for uri in listed_parts(&playlist) {
                        let Some(at) = try_group_object(&uri) else { continue };
                        if at > last && served.iter().all(|(x, _)| *x != at) {
                            let (status, _, done) = get(addr, &format!("{track}/{uri}")).await;
                            assert_eq!(status, 200, "{track}/{uri}: listed");
                            served.push((at, done));
                        }
                    }
                }
                let (msn, parts, hint) = edge(&playlist);
                let hint = hint.expect("a hinted part");
                let reload = format!("{track}/media.m3u8?_HLS_msn={msn}&_HLS_part={parts}");
                let part = format!("{track}/{hint}");
                let ((status, _, done), (rs, next, _)) = tokio::join!(get(addr, &part), get(addr, &reload));
                assert_eq!(rs, 200, "{reload}");
                playlist = String::from_utf8(next).expect("text");
                if status == 200 && served.last().is_none_or(|(x, _)| *x < group_object(&hint)) {
                    served.push((group_object(&hint), done));
                }
            }
            (track, served)
        }));
    }
    let mut late = Vec::new();
    for t in tasks {
        let (track, served) = t.await.expect("a follower");
        assert!(served.len() >= 250, "{track}: {} parts served in thirty seconds", served.len());
        let sent = src.sent.lock().unwrap().clone();
        // Every part from the first served to the last, in the order the source sent them.
        let mut order: Vec<(u64, u64)> = sent.keys().filter(|(t, _, _)| t == track).map(|(_, g, o)| (*g, *o)).collect();
        order.sort_unstable();
        let (first, last) = (served[0].0, served[served.len() - 1].0);
        let owed: Vec<(u64, u64)> = order.into_iter().filter(|x| *x >= first && *x <= last).collect();
        let got: Vec<(u64, u64)> = served.iter().map(|x| x.0).collect();
        assert_eq!(got, owed, "{track}: every part, none missing");
        for ((g, o), done) in served {
            let (capture, _) = sent[&(track.to_string(), g, o)];
            late.push(done.saturating_sub(capture));
            assert!(done <= capture + BOUND_MS, "{track} {g}/{o}: served {} ms after its capture stamp", done - capture);
        }
    }
    assert!(gw.withheld().is_empty(), "{:?}", gw.withheld());
    late.sort_unstable();
    println!(
        "measured on this host, this run, over loopback, capture stamp to served: n={} median={} ms p95={} ms max={} ms (bound {BOUND_MS} ms)",
        late.len(),
        late[late.len() / 2],
        late[late.len() * 95 / 100],
        late[late.len() - 1]
    );
    // The door's hold of each object, against when its group's receipts came after it.
    let holds = door_passed_on_runs(&dev);
    let mut held: Vec<u128> = holds.iter().map(|h| h.held.as_micros()).collect();
    let mut without: Vec<u128> = holds.iter().filter_map(|h| h.receipts_after).map(|d| d.as_micros()).collect();
    held.sort_unstable();
    without.sort_unstable();
    assert!(without.len() * 10 >= holds.len() * 9, "{} of {} objects' group receipts came", without.len(), holds.len());
    let q = |v: &[u128], p: usize| v[(v.len() * p / 100).min(v.len() - 1)];
    println!(
        "measured on this host, this run, the door's hold of an object (arrival at the door to passed): on its run n={} median={} us p95={} us max={} us; until its group's receipts (a door that reads no running receipt) n={} median={} ms p95={} ms max={} ms",
        held.len(),
        q(&held, 50),
        q(&held, 95),
        held[held.len() - 1],
        without.len(),
        q(&without, 50) / 1000,
        q(&without, 95) / 1000,
        without[without.len() - 1] / 1000
    );
    drop(gw);
    dev.stop().await;
}