//! `GET /metrics` of the room service beside a relay on loopback: it reads strictly as the text
//! format; its counters move with the requests made (grants by role, requests by route, method
//! and status) and its gauges with the room, whatever its kind (a publisher that enters and
//! leaves, a room with co-performance's audience, a watch room's viewer that joins beside a
//! publisher counted once, a remote desktop room's controllers and viewers, a conversation's
//! members; a room that is spent and the next one kept; the clock's session at the relay as it
//! ends); it answers `HEAD` as it does `GET`, without the text; and no room, principal, grant or
//! client address of those requests is in it.

use ed25519_dalek::SigningKey;
use joule_moq::draft22::kvp::{prop, KeyValuePair, Properties};
use joule_moq::draft22::Parameters;
use joule_moq::grant_proof::{self as gp, GrantProof};
use joule_moq::{TrackName, TrackNamespace};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{Authority, DoorRule, Grant};
use joule_moq_relay::transport::mux::MuxConfig;
use joule_moq_relay::transport::websocket;
use joule_moq_relay::{RelayConfig, RelayServer};
use joule_moq_room::keys::Keys;
use joule_moq_room::{http, remote, Config, Service};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::time::timeout;
use wai_metrics::read::{identifier_like, label_values, parse, value, Family};

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

async fn relay(keys: &Keys) -> RelayServer {
    let mut c = RelayConfig::loopback();
    c.quic_listen = None;
    c.auth.required = true;
    c.auth.authorities = vec![Authority { key: hex::encode(keys.authority.verifying_key().to_bytes()), rooms: vec![] }];
    let t = hex::encode(Sha256::digest(keys.clock_token));
    let clock = hex::encode(keys.clock.verifying_key().to_bytes());
    c.auth.grants = vec![
        Grant { prefix: vec!["*".into(), clock, "events".into()], token_sha256: t.clone(), publish: true, subscribe: false, proof: false, unlimited: true, seek: false },
        Grant { prefix: vec![], token_sha256: t, publish: false, subscribe: true, proof: false, unlimited: false, seek: false },
    ];
    c.receipts.door = ["audio", "video"].iter().map(|m| DoorRule { prefix: vec!["*".into(), "*".into(), m.to_string()], trust: vec![], trust_field: Some(1), hold_ms: 2_000 }).collect();
    RelayServer::start(c).await.expect("the relay starts")
}

/// The service beside `server`, `rest` the end of its configuration, `start` starting its clock:
/// listening, its clock's session open.
async fn service(keys: &Keys, server: &RelayServer, rest: &str, start: impl FnOnce(&Arc<Service>)) -> (Arc<Service>, std::net::SocketAddr) {
    let cfg = Config::from_toml(&format!("listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://{}{}\"\n{rest}", server.websocket_addr.unwrap(), server.relay().config().path)).unwrap();
    let svc = Service::new(Keys::from_toml(&keys.to_toml()).unwrap(), &cfg);
    let listener = tokio::net::TcpListener::bind(cfg.listen).await.unwrap();
    let addr = listener.local_addr().unwrap();
    start(&svc);
    tokio::spawn(http::serve(svc.clone(), listener));
    let up = timeout(STEP, async {
        while !svc.connected.load(Ordering::Acquire) {
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    })
    .await;
    assert!(up.is_ok(), "the clock's session opens");
    (svc, addr)
}

/// One HTTP/1.1 request to the service from `client`, through the front that names it, or at the
/// service's own address when `client` is empty: its status, its head and its body.
async fn call(addr: std::net::SocketAddr, method: &str, path: &str, body: Option<&Value>, client: &str) -> (u16, String, String) {
    request(addr, method, path, &[], body, client).await
}

/// [`call`], with `headers` beside the request's own.
async fn request(addr: std::net::SocketAddr, method: &str, path: &str, headers: &[(&str, String)], body: Option<&Value>, client: &str) -> (u16, String, String) {
    let mut s = tokio::net::TcpStream::connect(addr).await.unwrap();
    let b = body.map(|v| v.to_string()).unwrap_or_default();
    let extra: String = headers.iter().map(|(k, v)| format!("{k}: {v}\r\n")).collect();
    let front = if client.is_empty() { String::new() } else { format!("CF-Connecting-IP: {client}\r\n") };
    let req = format!("{method} {path} HTTP/1.1\r\nHost: room.test\r\n{front}{extra}Content-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{b}", b.len());
    s.write_all(req.as_bytes()).await.unwrap();
    let mut out = Vec::new();
    timeout(STEP, s.read_to_end(&mut out)).await.expect("an answer in time").unwrap();
    let text = String::from_utf8(out).unwrap();
    let (head, body) = text.split_once("\r\n\r\n").expect("a response");
    assert!(head.to_ascii_lowercase().contains("cache-control: no-store"), "{head}");
    (head.split(' ').nth(1).unwrap().parse().unwrap(), head.to_ascii_lowercase(), body.to_string())
}

async fn scrape(addr: std::net::SocketAddr) -> (String, Vec<Family>) {
    let (status, head, text) = call(addr, "GET", "/metrics", None, "").await;
    assert_eq!(status, 200, "{text}");
    assert!(head.contains(&format!("content-type: {}", wai_metrics::CONTENT_TYPE)), "{head}");
    let f = parse(&text).unwrap_or_else(|e| panic!("{e}\n{text}"));
    (text, f)
}

/// Scrape until the gauge `name` reads `want`: the service follows a session by what the relay
/// tells it, a moment after the session changes.
async fn settles(addr: std::net::SocketAddr, name: &str, want: f64) {
    let mut last = None;
    let reached = timeout(STEP, async {
        loop {
            let (_, f) = scrape(addr).await;
            last = value(&f, name, &[]);
            if last == Some(want) {
                return;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    })
    .await;
    assert!(reached.is_ok(), "{name} reads {want}, not {last:?}");
}

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

/// `POST /grant` for `k` as `role`, from `client`: the grant's answer.
async fn grant(addr: std::net::SocketAddr, k: &SigningKey, role: &str, client: &str) -> Value {
    let (st, _, g) = call(addr, "POST", "/grant", Some(&json!({"principal": pubhex(k), "role": role})), client).await;
    assert_eq!(st, 200, "{g}");
    serde_json::from_str(&g).unwrap()
}

/// A session at the relay under `grant`, publishing `k`'s media tracks in `room`: the clock
/// follows it in.
async fn enter(server: &RelayServer, room: &str, grant: &[u8], k: &SigningKey) -> (Client, Vec<(joule_moq_relay::wire::ReqTx, joule_moq_relay::wire::ReqRx)>) {
    let url = format!("ws://{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
    let dial = || async { Ok::<joule_moq_relay::transport::SharedConnection, String>(Arc::new(websocket::connect(&url, MuxConfig::default()).await.map_err(|e| e.to_string())?)) };
    let opts = ClientOptions { grants: vec![(grant.to_vec(), k.clone())], target: format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path), ..ClientOptions::default() };
    // The relay reads this SETUP before it sends its own, so the grant is proved over a
    // challenge its SETUP gave: on a second session, where the first held none.
    let c = Client::connect_proving(dial, opts).await.expect("SETUP");
    let mut held = Vec::new();
    let mut alias = 10;
    for m in ["audio", "video"] {
        let ns = TrackNamespace::from_strs([room, pubhex(k).as_str(), m]);
        for name in ["0_receipts", "0"] {
            let mut p = Properties::new();
            p.push(KeyValuePair::varint(prop::DEFAULT_PUBLISHER_PRIORITY, 64));
            held.push(c.publish(ns.clone(), TrackName::from_str(name), alias, Parameters::new(), p).await.expect("PUBLISH"));
            alias += 1;
        }
    }
    (c, held)
}

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

async fn run() {
    let keys = Keys::generate();
    let server = relay(&keys).await;
    let (_svc, addr) = service(&keys, &server, "[room]\nmax_publishers = 2\njoin_wait_s = 30\n[rate]\nclient_burst = 8\nclient_every_s = 600\n", |s| s.start()).await;

    let (_, f) = scrape(addr).await;
    assert_eq!(value(&f, "wai_room_grants_total", &[("role", "viewer")]), Some(0.0));
    assert_eq!(value(&f, "wai_room_grants_total", &[("role", "host")]), None, "this room grants no host");
    assert_eq!(value(&f, "wai_room_relay_connected", &[]), Some(1.0));
    assert_eq!(value(&f, "wai_room_rooms_total", &[]), Some(1.0));
    assert_eq!(value(&f, "wai_room_publishers", &[]), Some(0.0));

    // A publisher's grant and a viewer's, each from its own address; a request that is refused;
    // the room; a path the service does not have, naming a principal; the scrape by another method.
    let (alice, viewer) = (SigningKey::from_bytes(&[0x41; 32]), SigningKey::from_bytes(&[0x42; 32]));
    let a = grant(addr, &alice, "publisher", "203.0.113.41").await;
    let v = grant(addr, &viewer, "viewer", "203.0.113.42").await;
    let (st, _, _) = call(addr, "POST", "/grant", Some(&json!({"principal": "00", "role": "publisher"})), "203.0.113.43").await;
    assert_eq!(st, 400);
    assert_eq!(call(addr, "GET", "/room", None, "203.0.113.41").await.0, 200);
    assert_eq!(call(addr, "GET", &format!("/principal/{}", pubhex(&alice)), None, "203.0.113.41").await.0, 404);
    let (st, head, _) = call(addr, "PUT", "/metrics", None, "").await;
    assert_eq!(st, 405);
    assert!(head.contains("allow: get, head"), "{head}");
    let (st, head, nothing) = call(addr, "HEAD", "/metrics", None, "").await;
    assert_eq!((st, nothing.as_str()), (200, ""));
    assert!(head.contains(&format!("content-type: {}", wai_metrics::CONTENT_TYPE)), "{head}");
    // Through a front, the scrape is a path the service does not have.
    let (st, _, answer) = call(addr, "GET", "/metrics", None, "198.51.100.9").await;
    assert_eq!(st, 404, "{answer}");
    assert!(!answer.contains("wai_room"), "{answer}");
    for (field, given) in [("X-Forwarded-For", "198.51.100.9"), ("X-Forwarded-Host", "room.test")] {
        let (st, _, answer) = request(addr, "GET", "/metrics", &[(field, given.to_string())], None, "").await;
        assert_eq!(st, 404, "{field}: {answer}");
    }

    let (text, f) = scrape(addr).await;
    let req = |route: &str, method: &str, code: &str| value(&f, "wai_room_requests_total", &[("route", route), ("method", method), ("code", code)]);
    assert_eq!(value(&f, "wai_room_grants_total", &[("role", "publisher")]), Some(1.0));
    assert_eq!(value(&f, "wai_room_grants_total", &[("role", "viewer")]), Some(1.0));
    assert_eq!((req("/grant", "POST", "200"), req("/grant", "POST", "400")), (Some(2.0), Some(1.0)));
    assert_eq!((req("/room", "GET", "200"), req("other", "GET", "404")), (Some(1.0), Some(1.0)));
    assert_eq!((req("/metrics", "GET", "200"), req("/metrics", "PUT", "405"), req("/metrics", "GET", "404")), (Some(1.0), Some(1.0), Some(3.0)));
    assert_eq!(req("/metrics", "HEAD", "200"), Some(1.0));
    assert_eq!(value(&f, "wai_room_publishers", &[]), Some(1.0), "granted, and waited for");
    assert_eq!(value(&f, "wai_room_viewer_grants", &[]), Some(1.0));
    assert_eq!(value(&f, "wai_room_present", &[]), Some(0.0), "nobody has entered");

    // The publisher enters, and leaves: the room's gauges follow her.
    let room = a["room"]["room"].as_str().unwrap().to_string();
    let (session, held) = enter(&server, &room, &hex::decode(a["grant"].as_str().unwrap()).unwrap(), &alice).await;
    settles(addr, "wai_room_present", 1.0).await;
    let (text_present, f_present) = scrape(addr).await;
    assert_eq!(value(&f_present, "wai_room_publishers", &[]), Some(1.0), "present, and holding her place");
    drop(held);
    session.close(0, "done");
    settles(addr, "wai_room_present", 0.0).await;
    let (_, f_left) = scrape(addr).await;
    assert_eq!(value(&f_left, "wai_room_publishers", &[]), Some(0.0), "her place is given up");
    assert_eq!(value(&f_left, "wai_room_rooms_total", &[]), Some(1.0));

    // Her room is spent: the next publisher's grant starts another, and it is counted.
    let next = grant(addr, &SigningKey::from_bytes(&[0x4b; 32]), "publisher", "203.0.113.51").await;
    assert_ne!(next["room"]["room"], a["room"]["room"], "another room");
    let (_, f_next) = scrape(addr).await;
    assert_eq!(value(&f_next, "wai_room_rooms_total", &[]), Some(2.0));

    // No party and no thing of one: not a label, and nowhere in the text.
    let ids = [
        pubhex(&alice),
        pubhex(&viewer),
        a["room"]["room"].as_str().unwrap().to_string(),
        a["grant"].as_str().unwrap().to_string(),
        a["digest"].as_str().unwrap().to_string(),
        v["digest"].as_str().unwrap().to_string(),
        hex::encode(keys.authority.verifying_key().to_bytes()),
        hex::encode(keys.clock.verifying_key().to_bytes()),
        "203.0.113.4".to_string(),
        "198.51.100.9".to_string(),
        "room.test".to_string(),
    ];
    for f in [&f, &f_present] {
        for l in label_values(f) {
            assert!(!identifier_like(l), "a label value {l:?}");
        }
    }
    for id in &ids {
        assert!(!text.contains(id.as_str()) && !text_present.contains(id.as_str()), "{id} is in the metrics");
    }

    // The relay stops: the clock's session ends, and the gauge says so.
    server.shutdown().await;
    settles(addr, "wai_room_relay_connected", 0.0).await;
}

/// A room with co-performance: a viewer's grant is an audience member's, holding a place.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_audience_of_a_room_with_co_performance() {
    timeout(Duration::from_secs(120), async {
        let keys = Keys::generate();
        let server = relay(&keys).await;
        let rest = "[room]\nmax_publishers = 3\njoin_wait_s = 30\n[rate]\nclient_burst = 20\n[co]\naudience_input = 2\nlatency_budget_ms = 1500\nlead = \"first\"\nfidelity_every_s = 0\ncommit_after_ms = 800\nlead_ms = 200\n";
        let (_svc, addr) = service(&keys, &server, rest, |s| s.start()).await;
        let (_, f) = scrape(addr).await;
        assert_eq!(value(&f, "wai_room_audience", &[]), Some(0.0));
        grant(addr, &SigningKey::from_bytes(&[0x43; 32]), "viewer", "203.0.113.44").await;
        let (_, f) = scrape(addr).await;
        assert_eq!(value(&f, "wai_room_audience", &[]), Some(1.0), "granted, and waited for");
        assert_eq!(value(&f, "wai_room_viewer_grants", &[]), Some(1.0));
        assert_eq!(value(&f, "wai_room_publishers", &[]), Some(0.0));
        server.shutdown().await;
    })
    .await
    .expect("the test finishes in time");
}

/// A watch room: a viewer joins by a proof and is not among the room's members, a publisher is
/// both among them and followed as a viewer is. Each counts once, and a viewer's leave counts.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_watch_room_counts_the_viewers_who_joined() {
    timeout(Duration::from_secs(120), run_watch()).await.expect("the test finishes in time");
}

async fn run_watch() {
    let keys = Keys::generate();
    let server = relay(&keys).await;
    let dir = std::env::temp_dir().join(format!("joule-moq-room-metrics-watch-{}", std::process::id()));
    std::fs::create_dir_all(&dir).unwrap();
    let file = dir.join("source.wav");
    std::fs::write(&file, (0..4096u32).map(|i| (i % 251) as u8).collect::<Vec<u8>>()).unwrap();
    // The clock's session without the watch loop, which takes a viewer that says nothing for gone.
    let rest = format!("[room]\nwatch_source = \"{}\"\nwatch_duration_ms = 60000\n", file.display());
    let (svc, addr) = service(&keys, &server, &rest, |s| {
        tokio::spawn(s.clone().run_clock());
    })
    .await;

    let (publisher, viewer) = (SigningKey::from_bytes(&[0x44; 32]), SigningKey::from_bytes(&[0x45; 32]));
    let p = grant(addr, &publisher, "publisher", "203.0.113.45").await;
    let v = grant(addr, &viewer, "viewer", "203.0.113.46").await;
    let room = p["room"]["room"].as_str().unwrap().to_string();
    assert_eq!(v["room"]["room"].as_str(), Some(room.as_str()));
    let (_, f) = scrape(addr).await;
    assert_eq!(value(&f, "wai_room_present", &[]), Some(0.0), "nobody has entered");
    assert_eq!(value(&f, "wai_room_viewer_grants", &[]), Some(1.0));

    // The viewer joins, as a page does.
    let g = hex::decode(v["grant"].as_str().unwrap()).unwrap();
    let (_, _, c) = call(addr, "GET", "/watch/challenge", None, "203.0.113.46").await;
    let c: Value = serde_json::from_str(&c).unwrap();
    let c: [u8; 32] = hex::decode(c["challenge"].as_str().unwrap()).unwrap().try_into().unwrap();
    let digest: [u8; 32] = Sha256::digest(&g[..g.len() - 64]).into();
    let proof = wai::live_room::Proof::sign(digest, &viewer, c, "POST //room.test/watch/join").to_bytes();
    let body = json!({"room": room, "grant": hex::encode(&g), "proof": hex::encode(proof)});
    let (st, _, j) = call(addr, "POST", "/watch/join", Some(&body), "203.0.113.46").await;
    assert_eq!(st, 200, "{j}");
    settles(addr, "wai_room_present", 1.0).await;

    // The publisher enters: in the room's members and followed as a viewer is, and counted once.
    let (session, held) = enter(&server, &room, &hex::decode(p["grant"].as_str().unwrap()).unwrap(), &publisher).await;
    settles(addr, "wai_room_publishers", 1.0).await;
    settles(addr, "wai_room_present", 2.0).await;
    // Its join reaches the watch room's own record a moment after the member's state.
    tokio::time::sleep(Duration::from_millis(300)).await;
    let (_, f) = scrape(addr).await;
    assert_eq!(value(&f, "wai_room_present", &[]), Some(2.0));

    // The viewer falls silent, and the clock signs its leave.
    let (generation, gone) = {
        let mut st = svc.state.lock().unwrap();
        let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_micros() as u64;
        let gone = st.epoch.watch.as_mut().expect("a watch room").ended(now + joule_moq_room::watch::SILENT_US + 1_000_000);
        (st.epoch.generation, gone)
    };
    assert_eq!(gone.len(), 1, "the viewer, which joined by a proof");
    assert!(matches!(svc.watch_leave(generation, gone[0].0).await, Ok(Some(_))), "its leave is signed");
    settles(addr, "wai_room_present", 1.0).await;

    drop(held);
    session.close(0, "done");
    settles(addr, "wai_room_present", 0.0).await;
    let _ = std::fs::remove_dir_all(&dir);
    server.shutdown().await;
}

/// A remote desktop room grants by its own roles: its host holds a publisher's place, and its
/// controllers and viewers are its audience and its viewers' grants.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_remote_room_counts_its_controllers_and_viewers() {
    timeout(Duration::from_secs(120), async {
        let keys = Keys::generate();
        let server = relay(&keys).await;
        let rest = "[room]\nmax_present = 8\njoin_wait_s = 30\n[rate]\nclient_burst = 20\nclient_every_s = 1\n[remote]\nwidth = 320\nheight = 192\nfps = 20\nallow = \"all\"\nmax_age_ms = 3000\nlead_ms = 400\n";
        let (_svc, addr) = service(&keys, &server, rest, |s| s.start()).await;
        let (_, f) = scrape(addr).await;
        for role in ["host", "controller", "viewer"] {
            assert_eq!(value(&f, "wai_room_grants_total", &[("role", role)]), Some(0.0), "{role}");
        }
        assert_eq!(value(&f, "wai_room_grants_total", &[("role", "publisher")]), None, "this room grants no publisher");
        let host = SigningKey::from_bytes(&[0x46; 32]);
        let body = json!({"principal": pubhex(&host), "role": "host", "invitation": remote::new_invitation(&keys, 600)});
        let (st, _, g) = call(addr, "POST", "/grant", Some(&body), "203.0.113.47").await;
        assert_eq!(st, 200, "{g}");
        let (_, f) = scrape(addr).await;
        assert_eq!(value(&f, "wai_room_publishers", &[]), Some(1.0), "the host's place");
        assert_eq!(value(&f, "wai_room_viewer_grants", &[]), Some(0.0), "the host's is not a viewer's grant");

        grant(addr, &SigningKey::from_bytes(&[0x47; 32]), "controller", "203.0.113.48").await;
        grant(addr, &SigningKey::from_bytes(&[0x48; 32]), "viewer", "203.0.113.49").await;
        let (_, f) = scrape(addr).await;
        assert_eq!(value(&f, "wai_room_viewer_grants", &[]), Some(2.0));
        assert_eq!(value(&f, "wai_room_audience", &[]), Some(2.0));
        assert_eq!(value(&f, "wai_room_publishers", &[]), Some(1.0));
        for role in ["host", "controller", "viewer"] {
            assert_eq!(value(&f, "wai_room_grants_total", &[("role", role)]), Some(1.0), "{role}");
        }
        server.shutdown().await;
    })
    .await
    .expect("the test finishes in time");
}

/// A conversation's request of `k`, with its grant if it has one, and a proof over a fresh
/// challenge (live-rooms §10.5.3).
async fn conversation(addr: std::net::SocketAddr, k: &SigningKey, grant: Option<&[u8]>, path: &str) -> (u16, Value) {
    let (_, _, c) = call(addr, "GET", "/msg/challenge", None, "203.0.113.50").await;
    let c: Value = serde_json::from_str(&c).unwrap();
    let c: [u8; 32] = hex::decode(c["challenge"].as_str().unwrap()).unwrap().try_into().unwrap();
    let digest = grant.map_or([0; 32], |g| gp::grant_digest_and_principal(g).unwrap().0);
    let proof = GrantProof::sign(digest, k, c, &gp::context("POST", "room.test", path)).unwrap();
    let mut headers = vec![(gp::HEADER_PROOF, gp::base64url(&proof.to_bytes()))];
    if let Some(g) = grant {
        headers.push(("Authorization", format!("Bearer {}", gp::base64url(g))));
    }
    let (st, _, v) = request(addr, "POST", path, &headers, Some(&json!({})), "203.0.113.50").await;
    (st, serde_json::from_str(&v).unwrap_or(Value::Null))
}

/// A conversation keeps its members itself, and they are the principals present: none of them is
/// a publisher or a viewer of a room.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_conversation_counts_its_members() {
    timeout(Duration::from_secs(120), async {
        let keys = Keys::generate();
        let dir = std::env::temp_dir().join(format!("joule-moq-room-metrics-msg-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let cfg = Config::from_toml(&format!("listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://127.0.0.1:1/moq\"\n[messaging]\nstore = \"{}\"\nmax_members = 8\nmax_text = 512\n[rate]\nclient_burst = 64\n", dir.display())).unwrap();
        let svc = Service::new(Keys::from_toml(&keys.to_toml()).unwrap(), &cfg);
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(http::serve(svc.clone(), listener));

        let (_, f) = scrape(addr).await;
        assert_eq!(value(&f, "wai_room_present", &[]), Some(0.0));
        assert_eq!(value(&f, "wai_room_relay_connected", &[]), Some(0.0), "the clock's session was never opened");
        let (alice, bob) = (SigningKey::from_bytes(&[0x49; 32]), SigningKey::from_bytes(&[0x4a; 32]));
        let (st, a) = conversation(addr, &alice, None, "/msg/join").await;
        assert_eq!(st, 200, "{a}");
        let (st, b) = conversation(addr, &bob, None, "/msg/join").await;
        assert_eq!(st, 200, "{b}");
        let (_, f) = scrape(addr).await;
        assert_eq!(value(&f, "wai_room_present", &[]), Some(2.0));
        assert_eq!(value(&f, "wai_room_publishers", &[]), Some(0.0));
        assert_eq!(value(&f, "wai_room_viewer_grants", &[]), Some(0.0));

        let g = hex::decode(b["grant"].as_str().unwrap()).unwrap();
        let (st, v) = conversation(addr, &bob, Some(&g), "/msg/leave").await;
        assert_eq!(st, 200, "{v}");
        let (_, f) = scrape(addr).await;
        assert_eq!(value(&f, "wai_room_present", &[]), Some(1.0), "the one that left is not");
        let _ = std::fs::remove_dir_all(&dir);
    })
    .await
    .expect("the test finishes in time");
}