//! The room service against a relay on loopback that trusts its authority (the relay's
//! `[[auth.authorities]]`) and its clock's token:
//!
//! - a page's key gets a publisher's grant over HTTP; with the grant and a proof the
//!   publisher's session publishes its tracks, and the clock signs `join` and two
//!   `publish` for it on the clock's events track; when the session closes, `leave`;
//! - the room (its descriptor, its grants and the clock's chain as the relay serves it)
//!   verifies as WAI live-rooms §9 verifies a room;
//! - the grant alone, without the proof, opens no session;
//! - publishers are bounded, a viewer is not counted among them, grants are bounded per
//!   client, and a new room starts once the last publisher has left;
//! - a viewer may enter a room nobody has joined: the clock's events track is published with
//!   the grant, so the viewer's subscription is answered, and the clock's next record (the
//!   first publisher's `join`) reaches it;
//! - an end-to-end encrypted room (WAI confidentiality): its record checks against its
//!   descriptor, members key through the service's log as a page does (KeyPackages, the
//!   clock's proposals, a GroupInfo, commits and Welcomes, each checked by the clock before
//!   it is carried), a post the clock refuses is not carried, and a member that leaves is
//!   removed by the next commit.

use ed25519_dalek::SigningKey;
use joule_moq::draft22::data::FetchEntry;
use joule_moq::draft22::kvp::{prop, KeyValuePair, Properties};
use joule_moq::draft22::{Parameters, SessionError, Token};
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, TransportError};
use joule_moq_relay::{RelayConfig, RelayServer};
use joule_moq_room::keys::Keys;
use joule_moq_room::{http, Config, Service};
use serde_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;

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")
}

/// One HTTP/1.1 request to the service: its status and its JSON.
async fn call(addr: std::net::SocketAddr, method: &str, path: &str, body: Option<&Value>, client: &str) -> (u16, Value) {
    let mut s = tokio::net::TcpStream::connect(addr).await.unwrap();
    let b = body.map(|v| v.to_string()).unwrap_or_default();
    let req = format!("{method} {path} HTTP/1.1\r\nHost: room.test\r\nCF-Connecting-IP: {client}\r\nContent-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}");
    let status: u16 = head.split(' ').nth(1).unwrap().parse().unwrap();
    (status, serde_json::from_str(body).unwrap_or(Value::Null))
}

fn key(i: u8) -> SigningKey {
    SigningKey::from_bytes(&[i; 32])
}

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

async fn ask(addr: std::net::SocketAddr, k: &SigningKey, role: &str, client: &str) -> (u16, Value) {
    call(addr, "POST", "/grant", Some(&serde_json::json!({"principal": pubhex(k), "role": role})), client).await
}

async fn session(server: &RelayServer, opts: ClientOptions) -> Client {
    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())?)) };
    // A session that stays to see the relay refuse it: one SETUP, as it is.
    if opts.ignore_refusal {
        return Client::connect(dial().await.expect("a WebSocket session"), opts).await.expect("SETUP");
    }
    // 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.
    Client::connect_proving(dial, opts).await.expect("SETUP")
}

fn with_grant(server: &RelayServer, grant: &[u8], k: &SigningKey) -> ClientOptions {
    ClientOptions { grants: vec![(grant.to_vec(), k.clone())], target: format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path), ..ClientOptions::default() }
}

async fn publish_media(c: &Client, room: &str, k: &SigningKey) -> Vec<(joule_moq_relay::wire::ReqTx, joule_moq_relay::wire::ReqRx)> {
    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;
        }
    }
    held
}

/// The clock's chain as the relay serves its events track: every object of every group,
/// read as live-events §8.1 reads them, each record once, in `seq` order.
async fn chain(c: &Client, room: &str, clock: &str, t0_us: u64) -> Vec<Vec<u8>> {
    let ns = TrackNamespace::from_strs([room, clock, "events"]);
    let (_, entries) = c.fetch(ns, TrackName::from_str("0"), Parameters::new()).await.expect("FETCH the events track");
    let mut by_seq = std::collections::BTreeMap::new();
    for e in entries {
        let FetchEntry::Object(o) = e else { continue };
        let arr: Vec<Value> = serde_json::from_slice(&o.payload).expect("an object is a JSON array");
        for r in arr {
            let rec = wai::live_events::from_msf_record(&r.to_string(), t0_us).expect("an MSF record");
            let ev = wai::live_events::Event::parse(&rec).expect("a WEV1 record");
            by_seq.insert(ev.seq, rec);
        }
    }
    by_seq.into_values().collect()
}

fn kinds(chain: &[Vec<u8>]) -> Vec<String> {
    chain.iter().map(|r| wai::live_events::kind_name(wai::live_events::Event::parse(r).unwrap().kind).unwrap().to_string()).collect()
}

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

async fn run() {
    let keys = Keys::generate();
    let server = relay(&keys).await;
    let cfg = Config::from_toml(&format!(
        "listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://{}{}\"\n[room]\nmax_publishers = 2\njoin_wait_s = 30\n[rate]\nclient_burst = 4\nclient_every_s = 600\n",
        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();
    tokio::spawn(svc.clone().run_clock());
    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");
    let (st, health) = call(addr, "GET", "/health", None, "1.1.1.1").await;
    assert_eq!((st, health["relay"].as_bool()), (200, Some(true)));

    // Alice asks to publish and enters with her grant and its proof.
    let alice = key(0x31);
    let (st, a) = ask(addr, &alice, "publisher", "10.0.0.1").await;
    assert_eq!(st, 200, "{a}");
    let room = a["room"]["room"].as_str().unwrap().to_string();
    let t0 = a["room"]["t0_us"].as_u64().unwrap();
    let ga = hex::decode(a["grant"].as_str().unwrap()).unwrap();
    // The grant alone opens no session.
    let bare = session(&server, ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: ga.clone() }], ignore_refusal: true, ..ClientOptions::default() }).await;
    assert_eq!(joule_moq::grant_proof::refusal_of(&bare.peer_setup.options).as_deref(), Some("pop_missing"), "the relay's SETUP says it refuses");
    match timeout(STEP, bare.conn.closed()).await.expect("closed") {
        TransportError::Closed { code, .. } => assert_eq!(code, SessionError::UNAUTHORIZED.0),
        other => panic!("{other:?}"),
    }
    let ca = session(&server, with_grant(&server, &ga, &alice)).await;
    let held = publish_media(&ca, &room, &alice).await;

    // The clock announces her: join, then publish for each media kind.
    let viewer = key(0x33);
    let (st, v) = ask(addr, &viewer, "viewer", "10.0.0.3").await;
    assert_eq!((st, v["role"].as_str()), (200, Some("viewer")));
    assert_eq!(v["room"]["room"].as_str(), Some(room.as_str()), "a viewer enters the room as it is");
    let cv = session(&server, with_grant(&server, &hex::decode(v["grant"].as_str().unwrap()).unwrap(), &viewer)).await;
    let clock = pubhex(&keys.clock);
    let mut c = Vec::new();
    for _ in 0..100 {
        c = chain(&cv, &room, &clock, t0).await.into_iter().collect();
        if c.len() >= 3 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    assert_eq!(kinds(&c), ["join", "publish", "publish"]);

    // Bob too; a third publisher is refused, a viewer is not.
    let bob = key(0x32);
    let (st, b) = ask(addr, &bob, "publisher", "10.0.0.2").await;
    assert_eq!(st, 200, "{b}");
    let cb = session(&server, with_grant(&server, &hex::decode(b["grant"].as_str().unwrap()).unwrap(), &bob)).await;
    let held_b = publish_media(&cb, &room, &bob).await;
    let (st, full) = ask(addr, &key(0x34), "publisher", "10.0.0.4").await;
    assert_eq!((st, full["error"].as_str()), (409, Some("full")));
    let (st, _) = ask(addr, &key(0x35), "viewer", "10.0.0.4").await;
    assert_eq!(st, 200);

    // Alice leaves; the clock says so.
    drop(held);
    ca.close(0, "done");
    for _ in 0..100 {
        c = chain(&cv, &room, &clock, t0).await;
        if c.len() >= 7 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    assert_eq!(kinds(&c), ["join", "publish", "publish", "join", "publish", "publish", "leave"]);

    // The room verifies: its descriptor, the grants the service lists, the clock's chain.
    let (_, r) = call(addr, "GET", "/room", None, "10.0.0.9").await;
    let descriptor = hex::decode(r["descriptor"].as_str().unwrap()).unwrap();
    let grants: Vec<Vec<u8>> = r["grants"].as_array().unwrap().iter().map(|g| hex::decode(g.as_str().unwrap()).unwrap()).collect();
    assert_eq!(grants.len(), 2, "the publishers' grants");
    let run = wai::live_room::verify_room(&descriptor, &grants, None, &[c.clone()]);
    assert!(matches!(run.outcome, wai::live_room::RoomOutcome::Accept), "{:?}", run.outcome);
    assert_eq!(run.state.present.len(), 1);

    // Grants are bounded per client.
    let mut statuses = Vec::new();
    for i in 0..6 {
        statuses.push(ask(addr, &key(0x40 + i), "viewer", "10.9.9.9").await.0);
    }
    assert_eq!(statuses.iter().filter(|s| **s == 429).count(), 2, "{statuses:?}");

    // Bob leaves too; the next publisher starts a new room.
    drop(held_b);
    cb.close(0, "done");
    for _ in 0..100 {
        if chain(&cv, &room, &clock, t0).await.len() >= 8 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    let (st, n) = ask(addr, &key(0x50), "publisher", "10.0.0.5").await;
    assert_eq!(st, 200, "{n}");
    assert_ne!(n["room"]["room"].as_str(), Some(room.as_str()), "a new room");
    assert_eq!(n["room"]["grants"].as_array().unwrap().len(), 1);
    cv.close(0, "done");
    server.shutdown().await;
}

/// The relay answers a subscription only once its track has a publisher (or refuses it at
/// its rendezvous wait), and the clock signs nothing in a room nobody has joined: the clock's
/// events track has to be published with the grant, or a viewer of an empty room is refused.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_viewer_enters_a_room_nobody_has_joined() {
    timeout(Duration::from_secs(120), run_empty()).await.expect("the test finishes in time");
}

async fn run_empty() {
    let keys = Keys::generate();
    let server = relay(&keys).await;
    let cfg = Config::from_toml(&format!(
        "listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://{}{}\"\n[room]\nmax_publishers = 2\njoin_wait_s = 30\n",
        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();
    tokio::spawn(svc.clone().run_clock());
    tokio::spawn(http::serve(svc.clone(), listener));
    timeout(STEP, async {
        while !svc.connected.load(Ordering::Acquire) {
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("the clock's session opens");

    // A viewer first, in a room nobody has joined: its subscription to the clock's events
    // track is answered.
    let viewer = key(0x51);
    let (st, v) = ask(addr, &viewer, "viewer", "10.0.0.5").await;
    assert_eq!((st, v["role"].as_str()), (200, Some("viewer")), "{v}");
    let room = v["room"]["room"].as_str().unwrap().to_string();
    let cv = session(&server, with_grant(&server, &hex::decode(v["grant"].as_str().unwrap()).unwrap(), &viewer)).await;
    let clock = pubhex(&keys.clock);
    let ns = TrackNamespace::from_strs([room.as_str(), clock.as_str(), "events"]);
    cv.subscribe(ns, TrackName::from_str("0"), Parameters::new()).await.expect("a viewer of an empty room subscribes to the clock's events track");

    // Then a publisher enters; the clock signs her `join`, and it reaches the viewer on that
    // subscription.
    let alice = key(0x52);
    let (st, a) = ask(addr, &alice, "publisher", "10.0.0.6").await;
    assert_eq!(st, 200, "{a}");
    assert_eq!(a["room"]["room"].as_str(), Some(room.as_str()), "the publisher enters the viewer's room");
    let ca = session(&server, with_grant(&server, &hex::decode(a["grant"].as_str().unwrap()).unwrap(), &alice)).await;
    let _held = publish_media(&ca, &room, &alice).await;
    assert!(cv.next_data(STEP).await.is_some(), "the clock's record reaches the viewer on its subscription");
}

// ------------------------------------------------------------------- end-to-end encryption

/// A member as a page is one: it reads the room's log in order and creates or commits
/// through it when the room calls on it.
struct Member {
    m: joule_moq_mls::RoomMember,
    at: usize,
    seen: Vec<joule_moq_mls::Received>,
}

async fn post(addr: std::net::SocketAddr, room: &str, frames: &[Vec<u8>]) -> (u16, Value) {
    let body = serde_json::json!({"room": room, "frames": frames.iter().map(hex::encode).collect::<Vec<_>>()});
    call(addr, "POST", "/mls/post", Some(&body), "10.1.1.1").await
}

/// Read the log to its end and act, until nothing is left to read or do.
async fn sync(addr: std::net::SocketAddr, room: &str, p: &mut Member) {
    loop {
        let (st, v) = call(addr, "GET", &format!("/mls?room={room}&from={}&wait_ms=0", p.at), None, "10.1.1.1").await;
        assert_eq!(st, 200, "{v}");
        let entries = v["entries"].as_array().unwrap().clone();
        for e in &entries {
            assert_eq!(e["seq"].as_u64(), Some(p.at as u64));
            if let Some(h) = e["event"].as_str() {
                let ev = wai::live_events::Event::parse(&hex::decode(h).unwrap()).unwrap();
                ev.verify_signature().unwrap();
                if let Ok(wai::live_events::Payload::Lifecycle { subject, .. }) = wai::live_events::parse_payload(ev.kind, &ev.payload, 1024) {
                    match wai::live_events::kind_name(ev.kind) {
                        Some("join") => p.m.event(&joule_moq_mls::Event::Join { subject, keyed: true }),
                        Some("leave") => p.m.event(&joule_moq_mls::Event::Leave { subject, reason: 1 }),
                        _ => {}
                    }
                }
            } else {
                let f = hex::decode(e["mls"].as_str().unwrap()).unwrap();
                p.seen.push(p.m.receive(&f).expect("every carried message applies"));
            }
            p.at += 1;
        }
        let mut acted = false;
        if p.m.owes_create() {
            let info = p.m.stage_create().unwrap();
            if post(addr, room, &[info]).await.0 != 200 {
                p.m.withdraw();
            }
            acted = true;
        } else if !p.m.staged() && p.m.is_committer() && p.m.clock_pending() {
            let c = p.m.stage_commit(false).unwrap();
            let frames: Vec<Vec<u8>> = std::iter::once(c.commit).chain(c.welcome).collect();
            if post(addr, room, &frames).await.0 != 200 {
                p.m.withdraw();
            }
            acted = true;
        }
        if entries.is_empty() && !acted {
            return;
        }
    }
}

async fn until_epochs(addr: std::net::SocketAddr, room: &str, ms: &mut [&mut Member], want: &[Option<u64>]) {
    for _ in 0..200 {
        for m in ms.iter_mut() {
            sync(addr, room, m).await;
        }
        if ms.iter().map(|m| m.m.epoch()).collect::<Vec<_>>() == want {
            return;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    panic!("epochs {:?}, not {want:?}", ms.iter().map(|m| m.m.epoch()).collect::<Vec<_>>());
}

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

async fn run_e2ee() {
    use joule_moq::secure_objects::room::{Incoming, Outgoing};
    use joule_moq::secure_objects::{Plaintext, Suite};
    let keys = Keys::generate();
    let server = relay(&keys).await;
    let cfg = Config::from_toml(&format!(
        "listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://{}{}\"\n[room]\nmax_publishers = 3\njoin_wait_s = 30\ne2ee = true\n",
        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();
    tokio::spawn(svc.clone().run_clock());
    tokio::spawn(http::serve(svc.clone(), listener));
    timeout(STEP, async {
        while !svc.connected.load(Ordering::Acquire) {
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("the clock's session opens");

    // The room's record checks against its descriptor: its authority, recording mode 0.
    let (_, r) = call(addr, "GET", "/room", None, "10.0.0.9").await;
    let room = r["room"].as_str().unwrap().to_string();
    let room_id: [u8; 32] = hex::decode(&room).unwrap().try_into().unwrap();
    let d = wai::live_room::Descriptor::parse(&hex::decode(r["descriptor"].as_str().unwrap()).unwrap()).unwrap();
    let record = hex::decode(r["e2ee"]["record"].as_str().expect("a confidentiality record")).unwrap();
    let (_, suite) = joule_moq_mls::Record::check(&record, &room_id, &d.authority, d.recording.mode).expect("the record checks");
    assert_eq!(suite, Suite::AesGcm128Sha256);
    let clock = keys.clock.verifying_key().to_bytes();
    let member = |k: &SigningKey| Member { m: joule_moq_mls::RoomMember::new(room_id, clock, k.clone(), suite), at: 0, seen: Vec::new() };

    // Alice and Bob are granted, leave their KeyPackages, and publish.
    let (alice, bob) = (key(0x61), key(0x62));
    let (mut ma, mut mb) = (member(&alice), member(&bob));
    let mut held = Vec::new();
    let mut sessions = Vec::new();
    for (k, m, client) in [(&alice, &mut ma, "10.0.1.1"), (&bob, &mut mb, "10.0.1.2")] {
        let (st, g) = ask(addr, k, "publisher", client).await;
        assert_eq!(st, 200, "{g}");
        let kp = m.m.key_package().unwrap();
        let (st, v) = call(addr, "POST", "/mls/key-package", Some(&serde_json::json!({"room": room, "key_package": hex::encode(kp)})), client).await;
        assert_eq!((st, v["principal"].as_str()), (200, Some(pubhex(k).as_str())));
        let c = session(&server, with_grant(&server, &hex::decode(g["grant"].as_str().unwrap()).unwrap(), k)).await;
        held.push(publish_media(&c, &room, k).await);
        sessions.push(c);
        // Alice, the first present, creates the group through the log before Bob enters.
        if sessions.len() == 1 {
            until_epochs(addr, &room, &mut [m], &[Some(0)]).await;
        }
    }
    // A KeyPackage of a principal the room did not grant is not kept.
    let mut stranger = member(&key(0x69));
    let kp = stranger.m.key_package().unwrap();
    let (st, _) = call(addr, "POST", "/mls/key-package", Some(&serde_json::json!({"room": room, "key_package": hex::encode(kp)})), "10.0.1.9").await;
    assert_eq!(st, 403);

    // Alice creates the group through the log; Bob follows it, and joins by her commit's Welcome.
    until_epochs(addr, &room, &mut [&mut ma, &mut mb], &[Some(1), Some(1)]).await;
    assert!(ma.seen.contains(&joule_moq_mls::Received::Created));
    assert!(mb.seen.contains(&joule_moq_mls::Received::Joined(1)));
    // Posts the clock refuses are not carried: a second GroupInfo, a commit for an epoch
    // already ended, a Welcome alone.
    let (_, log) = call(addr, "GET", &format!("/mls?room={room}&from=0&wait_ms=0"), None, "10.1.1.1").await;
    let mls: Vec<Vec<u8>> = log["entries"].as_array().unwrap().iter().filter_map(|e| e["mls"].as_str()).map(|h| hex::decode(h).unwrap()).collect();
    let info = mls.iter().find(|f| f[0] == joule_moq_mls::group::frame::GROUP_INFO).unwrap().clone();
    let commit = mls.iter().find(|f| f[0] == joule_moq_mls::group::frame::COMMIT).unwrap().clone();
    let welcome = mls.iter().find(|f| f[0] == joule_moq_mls::group::frame::WELCOME).unwrap().clone();
    let end = log["end"].as_u64().unwrap();
    assert_eq!(post(addr, &room, &[info]).await.0, 409);
    assert_eq!(post(addr, &room, &[commit]).await.0, 409);
    assert_eq!(post(addr, &room, &[welcome]).await.0, 400);
    let (_, after) = call(addr, "GET", &format!("/mls?room={room}&from=0&wait_ms=0"), None, "10.1.1.1").await;
    assert_eq!(after["end"].as_u64(), Some(end), "nothing refused was carried");

    // Each opens what the other seals.
    let ns = TrackNamespace::from_strs([room.as_str(), pubhex(&alice).as_str(), "video"]);
    let name = TrackName::from_str("0");
    let pt = Plaintext::with_digest(b"a frame".to_vec());
    let s = ma.m.seal(&Outgoing { ns: &ns, name: &name, group: 1, object: 0, priority: 96, plaintext: &pt, immutable: &[], mutable: &[] }).unwrap();
    let got = mb.m.open(&Incoming { ns: &ns, name: &name, group: 1, object: 0, priority: 96, properties: &s.properties, payload: &s.payload }).unwrap();
    assert_eq!(got.plaintext, pt);

    // Bob leaves: the clock records it and proposes his removal; Alice commits it.
    drop(held.pop());
    sessions.pop().unwrap().close(0, "done");
    until_epochs(addr, &room, &mut [&mut ma, &mut mb], &[Some(2), None]).await;
    assert!(mb.seen.contains(&joule_moq_mls::Received::Removed(2)));

    // A room since replaced, or never there, has no log here.
    let (st, _) = call(addr, "GET", &format!("/mls?room={}&from=0&wait_ms=0", "00".repeat(32)), None, "10.1.1.1").await;
    assert_eq!(st, 410);
    for c in sessions {
        c.close(0, "done");
    }
    server.shutdown().await;
}

/// A raw GET: the status and the body's bytes.
async fn get_bytes(addr: std::net::SocketAddr, path: &str) -> (u16, Vec<u8>) {
    let mut s = tokio::net::TcpStream::connect(addr).await.unwrap();
    s.write_all(format!("GET {path} HTTP/1.1\r\nHost: room.test\r\nConnection: close\r\n\r\n").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 at = out.windows(4).position(|w| w == b"\r\n\r\n").expect("a response") + 4;
    let head = String::from_utf8_lossy(&out[..at]).to_string();
    (head.split(' ').nth(1).unwrap().parse().unwrap(), out[at..].to_vec())
}

/// A watch room (WAI co-watch) against a relay on loopback: its source served and named by its
/// policy; viewers admitted by a proof of their key for the request they make, a proof over
/// another target or presented again refused; their controls and presentation statements carried
/// when co-watch §5 and §7.2 accept them, and one that does not refused; the clock's skew
/// statements of each window, and a viewer's leave once it falls silent; and the room the log
/// carries verifies as co-watch §11 verifies it.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_watch_room_carries_its_viewers_and_states_their_skew() {
    timeout(Duration::from_secs(120), run_watch()).await.expect("the test finishes in time");
}

async fn run_watch() {
    use wai::co_watch::{self as cw, action, state, Control, Presentation};
    let keys = Keys::generate();
    let server = relay(&keys).await;
    let dir = std::env::temp_dir().join(format!("joule-moq-room-watch-{}", std::process::id()));
    std::fs::create_dir_all(&dir).unwrap();
    let file = dir.join("source.wav");
    let bytes: Vec<u8> = (0..4096u32).map(|i| (i % 251) as u8).collect();
    std::fs::write(&file, &bytes).unwrap();
    let cfg = Config::from_toml(&format!(
        "listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://{}{}\"\npublic_prefix = \"/room-api\"\n[room]\nwatch_source = \"{}\"\nwatch_duration_ms = 60000\nwatch_window_ms = 2000\n",
        server.websocket_addr.unwrap(),
        server.relay().config().path,
        file.display()
    ))
    .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();
    tokio::spawn(svc.clone().run_clock());
    tokio::spawn(svc.clone().run_watch());
    tokio::spawn(http::serve(svc.clone(), listener));
    timeout(STEP, async {
        while !svc.connected.load(Ordering::Acquire) {
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("the clock's session opens");

    // The room names its source by the policy; the service serves those bytes.
    let (_, r) = call(addr, "GET", "/room", None, "10.0.1.9").await;
    let policy_bytes = hex::decode(r["watch"]["policy"].as_str().expect("a watch room")).unwrap();
    let policy = cw::WatchPolicy::parse(&policy_bytes).unwrap();
    let (st, served) = get_bytes(addr, "/watch/source").await;
    assert_eq!((st, served.len()), (200, bytes.len()));
    let object = cw::SourceObject { content_hash: Sha256::digest(&served).into(), duration_ms: 60_000 };
    assert_eq!(policy.source, cw::source_digest(&[object]), "the served bytes are the policy's source");

    // Two viewers are granted, each with every action (watch_control 15).
    let (a, b) = (key(0x51), key(0x52));
    let mut grants = Vec::new();
    for (k, ip) in [(&a, "10.0.1.1"), (&b, "10.0.1.2")] {
        let (st, g) = ask(addr, k, "viewer", ip).await;
        assert_eq!(st, 200, "{g}");
        grants.push(hex::decode(g["grant"].as_str().unwrap()).unwrap());
    }
    let room = r["room"].as_str().unwrap().to_string();
    let room_id: [u8; 32] = hex::decode(&room).unwrap().try_into().unwrap();
    let t0 = r["t0_us"].as_u64().unwrap();
    let digest = |g: &[u8]| -> [u8; 32] { Sha256::digest(&g[..g.len() - 64]).into() };
    let join = |k: SigningKey, g: Vec<u8>, target: &'static str| {
        let room = room.clone();
        async move {
        let (_, c) = call(addr, "GET", "/watch/challenge", None, "10.0.1.9").await;
        let c: [u8; 32] = hex::decode(c["challenge"].as_str().unwrap()).unwrap().try_into().unwrap();
        let proof = wai::live_room::Proof::sign(digest(&g), &k, c, &format!("POST {target}")).to_bytes();
        let body = serde_json::json!({"room": room, "grant": hex::encode(&g), "proof": hex::encode(&proof)});
        let first = call(addr, "POST", "/watch/join", Some(&body), "10.0.1.9").await;
        let again = call(addr, "POST", "/watch/join", Some(&body), "10.0.1.9").await;
        (first, again)
        }
    };
    // A proof made for another target is refused.
    let ((st, e), _) = join(a.clone(), grants[0].clone(), "//room.test/watch/join").await;
    assert_eq!((st, e["error"].as_str()), (401, Some("pop_wrong_context")), "the client addressed the service under its prefix");
    let mut joined = Vec::new();
    for (k, g) in [(&a, &grants[0]), (&b, &grants[1])] {
        let ((st, j), (st2, j2)) = join((*k).clone(), g.clone(), "//room.test/room-api/watch/join").await;
        assert_eq!(st, 200, "{j}");
        assert_eq!((st2, j2["error"].as_str()), (401, Some("pop_replayed")), "a proof is good once");
        joined.push(j["t"].as_u64().unwrap());
    }

    // Each viewer's chain, posted as a page posts it.
    struct Page {
        k: SigningKey,
        seq: u64,
        prev: [u8; 32],
    }
    let now_t = || (std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_micros() as u64).saturating_sub(t0);
    let post = |p: &mut Page, t: u64, payload: Vec<u8>| {
        let e = wai::live_events::Event::sign(&p.k, room_id, p.seq, p.prev, t, None, wai::live_events::kind::META, payload);
        let body = serde_json::json!({"room": room, "events": [hex::encode(e.to_bytes())]});
        (e, body)
    };
    let mut pages = [Page { k: a.clone(), seq: 0, prev: [0; 32] }, Page { k: b.clone(), seq: 0, prev: [0; 32] }];
    // A control in the past is refused, and the chain does not move.
    let t = now_t().max(joined[0]);
    let (_, body) = post(&mut pages[0], t, Control { action: action::PLAY, at: t.saturating_sub(1), arg: 0 }.encode());
    let (st, e) = call(addr, "POST", "/watch/post", Some(&body), "10.0.1.1").await;
    assert_eq!((st, e["error"].as_str()), (409, Some("control_not_ahead")));
    // Viewer a plays the source 700 ms ahead.
    let play_at = t + 700_000;
    let (ev, body) = post(&mut pages[0], t, Control { action: action::PLAY, at: play_at, arg: 0 }.encode());
    let (st, e) = call(addr, "POST", "/watch/post", Some(&body), "10.0.1.1").await;
    assert_eq!(st, 200, "{e}");
    pages[0].seq += 1;
    pages[0].prev = ev.digest();
    // Both state their presentation every quarter second for five windows, near the room.
    let end = now_t() + 10_000_000;
    let mut k = 0i64;
    while now_t() < end {
        for (i, p) in pages.iter_mut().enumerate() {
            let x = now_t().max(joined[i]);
            let controlled = x.saturating_sub(play_at).min(60_000_000) * u64::from(x >= play_at);
            let err = [12_000i64, -9_000, 20_000, -14_000][((k + i as i64) % 4) as usize];
            let pos = (controlled as i64 + err).max(0) as u64;
            let s = Presentation::sign(&p.k, room_id, policy.source, x, pos, if x >= play_at { state::PLAYING } else { state::PAUSED }, 1_000_000);
            let (ev, body) = post(p, x, s.encode());
            let (st, e) = call(addr, "POST", "/watch/post", Some(&body), "10.0.1.1").await;
            assert_eq!(st, 200, "{e}");
            p.seq += 1;
            p.prev = ev.digest();
        }
        k += 1;
        tokio::time::sleep(Duration::from_millis(250)).await;
    }
    // The clock states each window that ended; then, silent, both viewers leave.
    let room_now = timeout(Duration::from_secs(30), async {
        loop {
            let (_, v) = call(addr, "GET", "/watch/room", None, "10.0.1.9").await;
            let leaves = v["chains"][0].as_array().unwrap().iter().filter(|h| wai::live_events::Event::parse(&hex::decode(h.as_str().unwrap()).unwrap()).unwrap().kind == wai::live_events::kind::LEAVE).count();
            if leaves == 2 {
                return v;
            }
            tokio::time::sleep(Duration::from_millis(250)).await;
        }
    })
    .await
    .expect("the clock records both leaving");
    let hexes = |v: &Value| -> Vec<Vec<u8>> { v.as_array().unwrap().iter().map(|h| hex::decode(h.as_str().unwrap()).unwrap()).collect() };
    let chains: Vec<Vec<Vec<u8>>> = room_now["chains"].as_array().unwrap().iter().map(hexes).collect();
    let run = cw::verify(&cw::Input {
        descriptor: &hex::decode(room_now["descriptor"].as_str().unwrap()).unwrap(),
        grants: &hexes(&room_now["grants"]),
        head: None,
        chains: &chains,
        policies: &hexes(&room_now["policies"]),
        sources: &[Some(vec![object])],
    });
    assert_eq!(run.outcome, cw::Outcome::Accept);
    assert!(run.skews.len() >= 4, "the clock states every window: {}", run.skews.len());
    let checked: Vec<_> = run.skews.iter().filter_map(|s| s.recomputed.as_ref()).collect();
    assert_eq!(checked.len(), run.skews.len());
    // Every window both were present for whole, after the play settled, is stated in sync.
    let settled = play_at + u64::from(policy.settle_ms) * 1000;
    let full: Vec<_> = run.skews.iter().filter(|s| s.from >= settled && s.to <= end).collect();
    assert!(!full.is_empty(), "a window after the play settled");
    for s in full {
        let (v, p) = s.recomputed.as_ref().unwrap();
        assert!(v.len() == 2 && v.iter().all(|x| x.verdict == cw::verdict::IN_SYNC) && p.iter().all(|x| x.verdict == cw::verdict::IN_SYNC), "{s:?}");
    }
    assert_eq!(room_now["refused"]["control_not_ahead"].as_u64(), Some(1));
    let _ = std::fs::remove_dir_all(&dir);
}

/// A watch room's viewer enters by a proof of `k` over a fresh challenge, for `target`: the
/// answer to its `POST /watch/join`.
async fn watch_join(addr: std::net::SocketAddr, room: &str, k: &SigningKey, grant: &[u8], target: &str) -> (u16, Value) {
    let (_, c) = call(addr, "GET", "/watch/challenge", None, "10.0.1.9").await;
    let c: [u8; 32] = hex::decode(c["challenge"].as_str().unwrap()).unwrap().try_into().unwrap();
    let digest: [u8; 32] = Sha256::digest(&grant[..grant.len() - 64]).into();
    let proof = wai::live_room::Proof::sign(digest, k, c, &format!("POST {target}")).to_bytes();
    let body = serde_json::json!({"room": room, "grant": hex::encode(grant), "proof": hex::encode(&proof)});
    call(addr, "POST", "/watch/join", Some(&body), "10.0.1.9").await
}

/// A watch room's viewer taken for gone (silent past the room's limit) while its `leave` waits
/// behind the clock's other records, as it does while the clock signs a window's skew
/// statements: its post in between is refused `not_present`, and it enters again before the
/// `leave` is signed. The clock then signs that `leave` first and its `join` after, so the
/// room's own `leave` for it finds nothing left to sign, and the event refused is carried
/// signed again. The other order, the `leave` signed before it enters again, signs the `join`
/// alone. Every `join` comes while the viewer is not present (live-rooms §7.2), and the room
/// the log carries verifies (co-watch §11).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_watch_viewer_taken_for_gone_enters_again_after_its_leave() {
    timeout(Duration::from_secs(60), run_watch_again()).await.expect("the test finishes in time");
}

async fn run_watch_again() {
    use wai::co_watch::{self as cw, state, Presentation};
    use wai::live_events::{kind, Event};
    let keys = Keys::generate();
    let server = relay(&keys).await;
    let dir = std::env::temp_dir().join(format!("joule-moq-room-watch-again-{}", std::process::id()));
    std::fs::create_dir_all(&dir).unwrap();
    let file = dir.join("source.wav");
    std::fs::write(&file, vec![7u8; 4096]).unwrap();
    let cfg = Config::from_toml(&format!(
        "listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://{}{}\"\n[room]\nwatch_source = \"{}\"\nwatch_duration_ms = 60000\n",
        server.websocket_addr.unwrap(),
        server.relay().config().path,
        file.display()
    ))
    .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();
    // The clock's session and the service, without the watch loop: the test asks for the
    // viewer's `leave` and has the clock sign it when the case calls for it.
    tokio::spawn(svc.clone().run_clock());
    tokio::spawn(http::serve(svc.clone(), listener));
    timeout(STEP, async {
        while !svc.connected.load(Ordering::Acquire) {
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("the clock's session opens");

    let (_, r) = call(addr, "GET", "/room", None, "10.0.1.9").await;
    let policy = cw::WatchPolicy::parse(&hex::decode(r["watch"]["policy"].as_str().expect("a watch room")).unwrap()).unwrap();
    let room = r["room"].as_str().unwrap().to_string();
    let room_id: [u8; 32] = hex::decode(&room).unwrap().try_into().unwrap();
    let t0 = r["t0_us"].as_u64().unwrap();
    let a = key(0x61);
    let pa = a.verifying_key().to_bytes();
    let (st, g) = ask(addr, &a, "viewer", "10.0.1.1").await;
    assert_eq!(st, 200, "{g}");
    let grant = hex::decode(g["grant"].as_str().unwrap()).unwrap();
    let target = "//room.test/watch/join";
    let unix_us = || std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_micros() as u64;
    let now_t = || unix_us().saturating_sub(t0);
    let enter = || {
        let (room, a, grant) = (&room, &a, &grant);
        async move {
            let (st, j) = watch_join(addr, room, a, grant, target).await;
            assert_eq!((st, j["again"].as_bool()), (200, None), "a join the clock signs: {j}");
            j["t"].as_u64().unwrap()
        }
    };
    // Its presentation statement at `t`, the event at `seq` of its chain after `prev`; posted.
    let statement = |seq: u64, prev: [u8; 32], t: u64| {
        Event::sign(&a, room_id, seq, prev, t, None, kind::META, Presentation::sign(&a, room_id, policy.source, t, 0, state::PAUSED, 1_000_000).encode())
    };
    let post = |e: &Event| {
        let body = serde_json::json!({"room": room, "events": [hex::encode(e.to_bytes())]});
        async move { call(addr, "POST", "/watch/post", Some(&body), "10.0.1.1").await }
    };
    // The room takes the viewer for gone, as it does once it is silent past its limit: its
    // `leave` is asked of the clock, and not yet signed.
    let silent = || {
        let mut st = svc.state.lock().unwrap();
        let generation = st.epoch.generation;
        let ended = st.epoch.watch.as_mut().expect("a watch room").ended(unix_us() + joule_moq_room::watch::SILENT_US + 1_000_000);
        assert_eq!(ended, vec![(pa, wai::live_room::leave::CONNECTION_LOST)], "the viewer is taken for gone");
        generation
    };

    let joined = enter().await;
    let e = statement(0, [0; 32], now_t().max(joined));
    let (st, r) = post(&e).await;
    assert_eq!(st, 200, "{r}");
    let (mut seq, mut prev) = (1, e.digest());

    // Taken for gone, and back before its `leave` is signed.
    let generation = silent();
    let (st, r) = post(&statement(seq, prev, now_t())).await;
    assert_eq!((st, r["error"].as_str()), (409, Some("not_present")), "a viewer taken for gone posts nothing");
    let joined = enter().await;
    assert_eq!(svc.watch_leave(generation, pa).await, Ok(None), "the leave was signed before the join, and is not signed after it");
    let e = statement(seq, prev, now_t().max(joined));
    let (st, r) = post(&e).await;
    assert_eq!(st, 200, "the position refused, signed again after the join: {r}");
    (seq, prev) = (seq + 1, e.digest());

    // Taken for gone, its `leave` signed, then back.
    let generation = silent();
    assert!(matches!(svc.watch_leave(generation, pa).await, Ok(Some(_))), "the room's leave for it is signed");
    assert_eq!(svc.watch_leave(generation, pa).await, Ok(None), "and signed once");
    let (st, r) = post(&statement(seq, prev, now_t())).await;
    assert_eq!((st, r["error"].as_str()), (409, Some("not_present")));
    let joined = enter().await;
    let e = statement(seq, prev, now_t().max(joined));
    let (st, r) = post(&e).await;
    assert_eq!(st, 200, "{r}");

    // The clock's records of the viewer, in order, and the room as a verifier is given it.
    let (_, v) = call(addr, "GET", "/watch/room", None, "10.0.1.9").await;
    let hexes = |v: &Value| -> Vec<Vec<u8>> { v.as_array().unwrap().iter().map(|h| hex::decode(h.as_str().unwrap()).unwrap()).collect() };
    let chains: Vec<Vec<Vec<u8>>> = v["chains"].as_array().unwrap().iter().map(hexes).collect();
    let life: Vec<u8> = chains[0].iter().map(|r| Event::parse(r).unwrap().kind).filter(|k| *k == kind::JOIN || *k == kind::LEAVE).collect();
    assert_eq!(life, [kind::JOIN, kind::LEAVE, kind::JOIN, kind::LEAVE, kind::JOIN], "the clock's lifecycle of the viewer");
    assert_eq!(chains[1].len(), 3, "the viewer's chain, whole");
    let run = cw::verify(&cw::Input {
        descriptor: &hex::decode(v["descriptor"].as_str().unwrap()).unwrap(),
        grants: &hexes(&v["grants"]),
        head: None,
        chains: &chains,
        policies: &hexes(&v["policies"]),
        sources: &[Some(vec![cw::SourceObject { content_hash: Sha256::digest(vec![7u8; 4096]).into(), duration_ms: 60_000 }])],
    });
    assert_eq!(run.outcome, cw::Outcome::Accept);
    assert_eq!(v["refused"]["not_present"].as_u64(), Some(2));
    let _ = std::fs::remove_dir_all(&dir);
}