//! A remote desktop room (WAI remote-desktop) the room service keeps, against a relay on loopback
//! that requires grants with their proofs and trusts the service's authority:
//!
//! - the host agent reaches the service over `https://` and the relay over `wss://`, through TLS
//!   fronts whose certificate it is given to trust, as a host on another machine would; it
//!   presents an invitation the authority signed and gets the host's grant; the clock signs its
//!   `join` and its `publish` of the screen when its screen track appears, and the host states its
//!   screen, with its sealing key, once it holds that `join`; a second key that presents the
//!   invitation is refused;
//! - two viewers make their keys and get viewers' grants; neither carries the control power;
//! - the host hands control to alice by a grant exchange it asks the service for, its grant and a
//!   proof over the service's challenge (a request without the proof, with another key's, or by a
//!   principal that is not the host with its own, is refused); alice asks for the floor and is
//!   given it, and her click is applied; bob's move is refused `input_not_permitted`;
//! - alice types a string and composes a text, sealed to the host (remote-desktop §5.5): the host
//!   injects them as typed, and neither bob, who follows her events, nor the session the service
//!   serves to a viewer's grant holds any trace of them; her text in the clear is refused
//!   `input_unsealed`;
//! - the host hands control to bob too and gives him the floor: his move is applied, alice's key
//!   refused `input_no_floor`;
//! - floor statements §7.2 refuses change nothing (§7.3): alice, who does not hold the floor,
//!   gives it to herself, and bob's release takes its place after the host took the floor back,
//!   a race on the wire; each is stated, the clock, the host and the verifier compute the same
//!   holder, and neither author is cut;
//! - mallory sends a record stamped before the one it follows: the clock cuts her chain there and
//!   removes her, and the host takes nothing more of her;
//! - the session's records are the room's: read with no grant they are refused 401, with a grant
//!   of another room 403, and a viewer's, a controller's and the host's grant with its proof each
//!   read them;
//! - the session as the clock holds it, and as the host holds it, each verify by `wai-rs`, every
//!   outcome the rules', at the commitment level and at the opened level under the host's
//!   opening key, which opens alice's seals to what she typed.
//!
//! A principal flooding past its own bound (`[remote] max_records_per_principal`): bob, a viewer,
//! sends input statements the host refuses; the clock holds his records up to the bound, cuts him
//! there and removes him (`leave` reason 2), and so does the host; the host and alice, a
//! controller, go on, her input after the cut applied; the session the clock holds and the one
//! the host holds verify at both levels; and the session's total bound is not reached.
//!
//! And a session at the clock's record bound on the whole session (`[remote] max_records`): once a
//! record would pass it, the clock holds none past it and ends the session, every principal's
//! `leave` (reason 2) signed, the host's last; the host stops; the room states why; the session
//! the clock holds is whole, within the bound, and verifies; and the next host starts a new room.

use ed25519_dalek::SigningKey;
use joule_moq_relay::config::{Authority, DoorRule, Grant};
use joule_moq_relay::{RelayConfig, RelayServer};
use joule_moq_remote::agent::{self, Command, Config as HostConfig, Decided, Room};
use joule_moq_remote::capture::Pattern;
use joule_moq_remote::encode::Ipcm;
use joule_moq_remote::inject::Recorder;
use joule_moq_remote::service::RoomService;
use joule_moq_remote::tls::{self, Trust};
use joule_moq_remote::viewer::Viewer;
use joule_moq_room::keys::Keys;
use joule_moq_room::remote::{RemoteRole, RemoteService};
use joule_moq_room::{http, remote, Config, Service};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::sync::atomic::Ordering;
use std::time::Duration;
use tokio::time::timeout;
use wai::remote_desktop::{self as rd, class, floor_action, Action, Fate, Refusal, SealFact, TimedAction};

const STEP: Duration = Duration::from_secs(20);
const W: u32 = 320;
const H: u32 = 192;

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 },
    ];
    // The host's screen held to its own key, as every publisher's media is.
    c.receipts.door = vec![DoorRule { prefix: vec!["*".into(), "*".into(), "screen".into()], trust: vec![], trust_field: Some(1), hold_ms: 2_000 }];
    RelayServer::start(c).await.expect("the relay starts")
}

fn at(dt_us: u32, action: Action) -> TimedAction {
    TimedAction { dt_us, action }
}

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

async fn run() {
    let keys = Keys::generate();
    let server = relay(&keys).await;
    let url = format!("ws://{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
    let cfg = Config::from_toml(&format!(
        "listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"{url}\"\n[room]\nmax_present = 8\njoin_wait_s = 30\n[rate]\nclient_burst = 200\nclient_every_s = 1\n[remote]\nwidth = {W}\nheight = {H}\nfps = 20\nallow = \"none\"\nmax_age_ms = 3000\nlead_ms = 400\n"
    ))
    .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();
    svc.start();
    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 service = RoomService::new(&format!("http://{addr}")).unwrap();
    let room0 = service.room().await.expect("the room");
    assert!(room0.host.is_none(), "no host before one presents an invitation");

    // The host, by an invitation the authority signed, through TLS fronts before the service and
    // the relay: a certificate for 127.0.0.1, which the host is given to trust.
    let id = joule_moq_relay::cert::RelayIdentity::dev_self_signed(&["127.0.0.1".into()]).expect("a certificate");
    let service_tls = tls::front(id.chain.clone(), id.key.clone_key(), addr).await.expect("a TLS front for the service");
    let relay_tls = tls::front(id.chain.clone(), id.key.clone_key(), server.websocket_addr.unwrap()).await.expect("a TLS front for the relay");
    let invitation = remote::new_invitation(&keys, 600);
    let host_key = SigningKey::from_bytes(&[0x41; 32]);
    let pattern = Pattern::new(W, H);
    let recorder = Recorder::new(pattern.drawn.clone());
    let injected = recorder.log.clone();
    let hc = HostConfig {
        relay_url: format!("wss://{relay_tls}{}", server.relay().config().path),
        trust: Trust::only(id.chain.clone()),
        opening_key: None,
        room: Room::Service { url: format!("https://{service_tls}"), key: host_key.clone(), invitation: Some(invitation.clone()) },
        fps: 20,
        max_age_ms: 3_000,
        floor: rd::floor_mode::FLOOR,
        give_on_request: true,
        redactions: vec![],
        clip_max: 1000,
        pointer: true,
        key_every: Duration::from_secs(2),
        run_for: Some(Duration::from_secs(180)),
        entered: None,
    };
    let (cmd, cmds) = tokio::sync::mpsc::unbounded_channel();
    let host = tokio::spawn(agent::run(hc, Box::new(pattern), Box::new(Ipcm::new(W, H).unwrap()), Box::new(recorder), cmds));
    let host_pk = host_key.verifying_key().to_bytes();
    let view = timeout(STEP, async {
        loop {
            if let Ok(v) = service.call("GET", "/room", None, &[]).await
                && v.body["remote"]["screen"].is_object()
            {
                return v.body;
            }
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
    })
    .await
    .expect("the host's screen statement reaches the clock");
    assert_eq!(view["remote"]["host"], hex::encode(host_pk), "the invitation's first key hosts");
    // The invitation is bound to the first key that presented it.
    let other = SigningKey::from_bytes(&[0x42; 32]).verifying_key().to_bytes();
    let again = service.call("POST", "/grant", Some(&serde_json::json!({"principal": hex::encode(other), "role": "host", "invitation": invitation})), &[]).await.unwrap();
    assert_eq!((again.status, again.body["error"].as_str()), (403, Some("invitation_used")), "{:?}", again.body);
    let ctl = service.call("POST", "/grant", Some(&serde_json::json!({"principal": hex::encode(other), "role": "controller"})), &[]).await.unwrap();
    assert_eq!((ctl.status, ctl.body["error"].as_str()), (403, Some("control_not_allowed")), "no one asks for control here");

    // Two viewers, each under a key it makes.
    let mut alice = Viewer::join_service(&service, &url, "viewer").await.expect("alice joins");
    let mut bob = Viewer::join_service(&service, &url, "viewer").await.expect("bob joins");
    for (n, v) in [("alice", &mut alice), ("bob", &mut bob)] {
        assert!(v.pump(STEP, |v| v.joined && v.screen.is_some() && v.frames.len() >= 3).await, "{n}: joined, the screen statement and frames ({} frames, joined {})", v.frames.len(), v.joined);
    }
    let screen = alice.screen.clone().unwrap().1;
    assert_eq!((screen.width as u32, screen.height as u32, screen.max_age_ms), (W, H, 3_000));

    // Handing control over is the host's: its grant with a proof over the service's challenge.
    let (host_grant, _) = {
        let room_now = service.room().await.unwrap();
        let g = room_now.grants.iter().find(|g| joule_moq::grant_proof::grant_digest_and_principal(g).is_some_and(|(_, p)| p == host_pk)).cloned().expect("the host's grant is the room's");
        (g, room_now)
    };
    let body = serde_json::json!({"to": hex::encode(alice.me), "control": 3});
    let bare = service.call("POST", "/remote/control", Some(&body), &[("Authorization", format!("Bearer {}", joule_moq::grant_proof::base64url(&host_grant)))]).await.unwrap();
    assert_eq!(bare.status, 401, "the grant alone, with no proof: {:?}", bare.body);
    let wrong = service.with_proof("POST", "/remote/control", Some(&body), &host_grant, &SigningKey::from_bytes(&[0x43; 32])).await.unwrap();
    assert_eq!(wrong.status, 401, "a proof by another key: {:?}", wrong.body);
    // A grant and its proof by a principal that is not the host: the request is refused.
    let eve = SigningKey::from_bytes(&[0x44; 32]);
    let (eve_grant, _) = service.grant(&eve.verifying_key().to_bytes(), "viewer", None).await.unwrap();
    let not_host = service.with_proof("POST", "/remote/control", Some(&body), &eve_grant, &eve).await.unwrap();
    assert_eq!((not_host.status, not_host.body["error"].as_str()), (403, Some("power")), "{:?}", not_host.body);

    // The session's records are the room's: no grant, 401; a grant of another room, 403; a
    // viewer's grant and its proof, the records.
    for path in ["/remote/room", "/remote/state"] {
        let none = service.call("GET", path, None, &[]).await.unwrap();
        assert_eq!(none.status, 401, "{path} with no grant: {:?}", none.body);
        assert!(none.header(joule_moq::grant_proof::HEADER_CHALLENGE).is_some(), "{path}: a challenge for the next request");
    }
    let other_room = {
        let mut e = remote::epoch(&keys, cfg.settings().unwrap(), Some(&RemoteService::new(cfg.remote.clone().unwrap().settings().unwrap())), 9, joule_moq_remote::capture::now_unix_us() - 5_000_000);
        e.grant_remote(&keys, eve.verifying_key().to_bytes(), RemoteRole::Viewer, joule_moq_remote::capture::now_unix_us() / 1_000_000).unwrap().record
    };
    for path in ["/remote/room", "/remote/state"] {
        let other = service.with_proof("GET", path, None, &other_room, &eve).await.unwrap();
        assert_eq!((other.status, other.body["error"].as_str()), (403, Some("room")), "{path} with another room's grant: {:?}", other.body);
    }
    let as_viewer = service.with_proof("GET", "/remote/room", None, &eve_grant, &eve).await.unwrap();
    assert_eq!(as_viewer.status, 200, "a viewer's grant reads the session: {:?}", as_viewer.body);
    assert!(as_viewer.body["chains"].as_array().is_some_and(|c| !c.is_empty()));
    let wrong_key = service.with_proof("GET", "/remote/state", None, &eve_grant, &SigningKey::from_bytes(&[0x45; 32])).await.unwrap();
    assert_eq!(wrong_key.status, 401, "a viewer's grant with another key's proof");

    cmd.send(Command::Control { to: alice.me, control: class::POINTER | class::KEYS }).unwrap();
    assert!(alice.pump(STEP, |v| !v.exchanged.is_empty()).await, "the clock signs alice's grant exchange");
    let effective = alice.exchanged[0].0;
    while alice.now() < effective + 100_000 {
        alice.pump(Duration::from_millis(100), |_| false).await;
    }
    // Alice asks for the floor; the host gives it on request.
    alice.floor(floor_action::REQUEST, [0; 32]).await.unwrap();
    let me_a = alice.me;
    assert!(alice.pump(STEP, |v| v.holder == Some(me_a)).await, "the floor given to alice");
    let a1 = alice.input(vec![at(6_000, Action::Move { x: 100, y: 50 }), at(3_000, Action::Button { button: 0, down: true, x: 100, y: 50 }), at(0, Action::Button { button: 0, down: false, x: 100, y: 50 })]).await.unwrap();
    assert!(alice.pump(STEP, |v| v.outcomes.get(&a1).is_some_and(|o| o.2.is_some_and(|an| v.frames.contains_key(&an.group)))).await, "alice's click applied and its frame held");
    alice.latency(a1, 0).await.expect("alice's latency record");
    let b1 = bob.input(vec![at(0, Action::Move { x: 10, y: 10 })]).await.unwrap();
    assert!(bob.pump(STEP, |v| v.outcomes.contains_key(&b1)).await, "bob's input answered");
    assert_eq!(bob.outcomes[&b1].1.reason, Refusal::NotPermitted.number(), "a viewer carries no control power");

    // Alice types a string and composes a text: sealed to the host's sealing key, which its
    // screen statement states. Bob follows her events, as any principal of the room may.
    let sealing_key = alice.screen.as_ref().unwrap().1.seal.expect("the host states a sealing key");
    bob.follow(alice.me).await.expect("bob follows alice's events");
    let mut typed = Vec::new();
    for (k, u) in [0x0Bu16, 0x18, 0x11, 0x17, 0x08, 0x15, 0x1F].iter().enumerate() {
        typed.push(at(20_000 - 2_000 * k as u32, Action::Key { usage: *u, down: true, modifiers: 0 }));
        typed.push(at(19_000 - 2_000 * k as u32, Action::Key { usage: *u, down: false, modifiers: 0 }));
    }
    typed.push(at(0, Action::Text("pässwörd-hunter2".into())));
    let a_typed = alice.input(typed.clone()).await.unwrap();
    assert!(alice.pump(STEP, |v| v.outcomes.contains_key(&a_typed)).await, "alice's typing answered");
    assert_eq!(alice.outcomes[&a_typed].1.reason, 0, "applied, the seal opened by the host");
    let a_clear = alice.input_on(alice.screen.as_ref().unwrap().0, vec![at(0, Action::Text("in the clear".into()))]).await.unwrap();
    assert!(alice.pump(STEP, |v| v.outcomes.contains_key(&a_clear)).await);
    assert_eq!(alice.outcomes[&a_clear].1.reason, Refusal::Unsealed.number(), "a text in the clear to a host that states a sealing key");
    let alice_pk = alice.me;
    assert!(
        bob.pump(STEP, |v| v.received.iter().filter_map(|r| wai::live_events::Event::parse(r).ok()).filter(|e| e.principal == alice_pk).any(|e| e.payload.windows(12).any(|w| w == b"in the clear"))).await,
        "bob receives alice's input statements"
    );

    // Control to bob too, and the floor handed to him.
    cmd.send(Command::Control { to: bob.me, control: class::POINTER }).unwrap();
    assert!(bob.pump(STEP, |v| !v.exchanged.is_empty()).await, "the clock signs bob's grant exchange");
    let effective = bob.exchanged[0].0;
    while bob.now() < effective + 100_000 {
        bob.pump(Duration::from_millis(100), |_| false).await;
    }
    cmd.send(Command::GiveFloor(bob.me)).unwrap();
    let me_b = bob.me;
    assert!(bob.pump(STEP, |v| v.holder == Some(me_b)).await, "the floor handed to bob");
    assert!(alice.pump(STEP, |v| v.holder == Some(me_b)).await);
    let b2 = bob.input(vec![at(0, Action::Move { x: 200, y: 120 })]).await.unwrap();
    assert!(bob.pump(STEP, |v| v.outcomes.get(&b2).is_some_and(|o| o.2.is_some())).await, "bob's move applied");
    let a2 = alice.input(vec![at(0, Action::Key { usage: 0x04, down: true, modifiers: 0 })]).await.unwrap();
    assert!(alice.pump(STEP, |v| v.outcomes.contains_key(&a2)).await);
    assert_eq!(alice.outcomes[&a2].1.reason, Refusal::NoFloor.number(), "alice, off the floor");
    // A controller's grant, bob's exchanged one, and its proof read the session's state.
    let bob_grant = service.room().await.unwrap().grants.into_iter().find(|g| joule_moq::grant_proof::grant_digest_and_principal(g).is_some_and(|(d, _)| d == bob.exchanged[0].1)).expect("bob's exchanged grant");
    let as_controller = service.with_proof("GET", "/remote/state", None, &bob_grant, bob.key()).await.unwrap();
    assert_eq!((as_controller.status, as_controller.body["verified"]["outcome"].as_str()), (200, Some("accept")), "{:?}", as_controller.body["verified"]);

    // Floor statements §7.2 refuses change nothing: alice, who does not hold the floor, gives it
    // to herself; the host takes the floor back, and bob's release takes its place after it.
    alice.floor(floor_action::GIVE, alice.me).await.unwrap();
    alice.pump(Duration::from_millis(500), |_| false).await;
    assert!(bob.pump(STEP, |v| v.holder == Some(me_b)).await, "bob keeps the floor");
    cmd.send(Command::TakeFloor).unwrap();
    assert!(bob.pump(STEP, |v| v.holder.is_none()).await, "the host takes the floor back");
    bob.floor(floor_action::RELEASE, [0; 32]).await.unwrap();
    let b3 = bob.input(vec![at(0, Action::Move { x: 210, y: 130 })]).await.unwrap();
    assert!(bob.pump(STEP, |v| v.outcomes.contains_key(&b3)).await);
    assert_eq!(bob.outcomes[&b3].1.reason, Refusal::NoFloor.number(), "bob, whose floor the host took");

    // Mallory sends a record stamped before the one it follows: out of her chain's order.
    let mut mallory = Viewer::join_service(&service, &url, "viewer").await.expect("mallory joins");
    assert!(mallory.pump(STEP, |v| v.joined).await, "mallory joined");
    let chat = |s: &str| wai::live_events::Payload::Chat { text: s.into(), origin: None }.encode();
    // A moment after her join, and then a millisecond before that: present, and out of order.
    mallory.pump(Duration::from_millis(300), |_| false).await;
    let now = mallory.now();
    mallory.send_at(now, wai::live_events::kind::CHAT, chat("in order")).await.unwrap();
    mallory.send_at(now - 1_000, wai::live_events::kind::CHAT, chat("back in time")).await.unwrap();
    assert!(mallory.pump(STEP, |v| v.left.is_some()).await, "the clock removes mallory");
    assert_eq!(mallory.left, Some(wai::live_room::leave::REMOVED));
    let state = timeout(STEP, async {
        loop {
            let s = service.with_proof("GET", "/remote/state", None, &host_grant, &host_key).await.unwrap();
            if s.body["cut"].as_array().is_some_and(|c| !c.is_empty()) {
                return s.body;
            }
            tokio::time::sleep(Duration::from_millis(200)).await;
        }
    })
    .await
    .expect("the clock states the cut");
    let cut = state["cut"].as_array().unwrap();
    assert_eq!(cut.len(), 1, "mallory alone is cut: {cut:?}");
    assert_eq!((cut[0]["principal"].as_str(), cut[0]["reason"].as_str(), cut[0]["seq"].as_u64()), (Some(hex::encode(mallory.me).as_str()), Some("time_reversed"), Some(1)), "her record in order is the room's; the one out of it is not: {cut:?}");

    // Every viewer's last records reach the host and the clock before the host stops.
    for v in [&mut alice, &mut bob, &mut mallory] {
        v.pump(Duration::from_millis(700), |_| false).await;
    }
    cmd.send(Command::Stop).unwrap();
    let report = timeout(Duration::from_secs(60), host).await.expect("the host stops").expect("the task").expect("the session");
    for v in [&alice, &bob, &mallory] {
        assert_eq!(v.refused_frames, 0, "every frame and receipt verified");
        v.close();
    }

    // What the host decided and injected.
    let decided = |d: &[u8; 32]| report.decided.iter().find(|(x, _)| x == d).map(|(_, o)| o.clone());
    assert!(matches!(decided(&a1), Some(Decided::Applied { .. })), "{:?}", decided(&a1));
    assert_eq!(decided(&b1), Some(Decided::Refused(Refusal::NotPermitted)));
    assert!(matches!(decided(&b2), Some(Decided::Applied { .. })));
    assert_eq!(decided(&a2), Some(Decided::Refused(Refusal::NoFloor)));
    let log = injected.lock().unwrap().clone();
    assert_eq!(log.len(), 4 + typed.len(), "alice's three actions, what she typed and bob's move, nothing else");
    assert_eq!(&log[3..3 + typed.len()], &typed.iter().map(|a| a.action.clone()).collect::<Vec<_>>()[..], "the host injected what alice typed, opened");
    assert_eq!(decided(&a_clear), Some(Decided::Refused(Refusal::Unsealed)));
    // Bob's received records hold no trace of it: not the text, not a key as an action in the
    // clear carries it (type 4, usage 0x0B little-endian, down).
    for needle in [&b"hunter2"[..], "pässwörd".as_bytes(), &[rd::action_type::KEY, 0x0B, 0x00, 0x01, 0x00]] {
        assert!(!bob.received.iter().any(|r| r.windows(needle.len()).any(|w| w == needle)), "bob's records hold {needle:?}");
    }
    assert_eq!(report.cut, vec![(mallory.me, "time_reversed")], "the host cut mallory where the clock did, and no one for a floor statement");

    // The session as the clock holds it, verified: the service's own statement of it, and wai-rs
    // over the records it serves.
    tokio::time::sleep(Duration::from_millis(800)).await;
    let st = service.with_proof("GET", "/remote/state", None, &host_grant, &host_key).await.unwrap().body;
    assert_eq!(st["verified"]["outcome"], "accept", "{}", st["verified"]);
    // The refused floor statements, stated; the clock's floor is the verifier's and the host's.
    let refused: Vec<(&str, &str)> = st["floor_refused"].as_array().unwrap().iter().map(|f| (f["action"].as_str().unwrap(), f["reason"].as_str().unwrap())).collect();
    assert_eq!(refused, vec![("give", "floor_not_permitted"), ("release", "floor_not_holder")], "{}", st["floor_refused"]);
    let clock_refused: Vec<(&str, &str)> = st["clock_floor"]["refused"].as_array().unwrap().iter().map(|f| (f["action"].as_str().unwrap(), f["reason"].as_str().unwrap())).collect();
    assert_eq!(clock_refused, refused, "the clock refuses what the verifier refuses");
    let verified_holder = st["floor"].as_array().unwrap().last().map(|c| c["holder"].clone());
    assert_eq!(verified_holder, Some(Value::Null), "after the host took it back, no one holds the floor");
    assert_eq!(st["clock_floor"]["holder"], Value::Null, "the clock's holder is the verifier's");
    assert_eq!(report.holder, None, "and the host's");
    // The session as a viewer reads it from the service: no trace of what alice typed.
    let as_viewer = service.with_proof("GET", "/remote/room", None, &eve_grant, &eve).await.unwrap();
    assert_eq!(as_viewer.status, 200);
    let served = serde_json::to_vec(&as_viewer.body).unwrap();
    for needle in ["hunter2", "pässwörd", &hex::encode("hunter2"), &hex::encode("pässwörd")] {
        assert!(!served.windows(needle.len()).any(|w| w == needle.as_bytes()), "the served session holds {needle}");
    }
    let room = service.session(&host_grant, &host_key).await.unwrap();
    let hexes = |v: &Value| v.as_array().unwrap().iter().map(|x| hex::decode(x.as_str().unwrap()).unwrap()).collect::<Vec<_>>();
    let descriptor = hex::decode(room["descriptor"].as_str().unwrap()).unwrap();
    let grants = hexes(&room["grants"]);
    let chains: Vec<Vec<Vec<u8>>> = room["chains"].as_array().unwrap().iter().map(hexes).collect();
    for needle in [&b"hunter2"[..], "pässwörd".as_bytes()] {
        assert!(!chains.iter().flatten().any(|r| r.windows(needle.len()).any(|w| w == needle)), "the clock's records hold {needle:?}");
    }
    let vi = rd::VerifyInput { descriptor: &descriptor, grants: &grants, head: None, chains: &chains };
    // At the opened level, under the host's opening key: every seal opens to what alice typed.
    let opened = rd::verify_opened(&vi, &[report.opening_key.0]);
    assert_eq!(opened.outcome, rd::Outcome::Accept, "the session the clock holds verifies at the opened level");
    assert_eq!(wai::hpke::public_key(&report.opening_key.0), sealing_key, "the opening key is the sealing key's");
    let fact = opened.inputs.iter().find(|f| f.digest == a_typed).expect("alice's typing");
    let Some(SealFact::Opened(st)) = &fact.seal else { panic!("its seal opened: {:?}", fact.seal) };
    let Some(Ok(rd::Statement::Input(clear))) = rd::read_statement(st) else { panic!("an input statement") };
    assert_eq!(clear.actions, typed, "it opens to what alice typed");
    // A wrong opening key opens nothing: the seal is the host's alone.
    let other = rd::verify_opened(&vi, &[[7; 32]]);
    assert_eq!(other.inputs.iter().find(|f| f.digest == a_typed).unwrap().seal, Some(SealFact::Unopened));
    let run = rd::verify(&vi);
    assert_eq!(run.outcome, rd::Outcome::Accept, "the session the clock holds verifies");
    let fate = |d: &[u8; 32]| run.inputs.iter().find(|f| f.digest == *d).map(|f| f.fate);
    assert!(matches!(fate(&a1), Some(Fate::Applied { .. })));
    assert_eq!(fate(&b1), Some(Fate::Refused(Refusal::NotPermitted)));
    assert!(matches!(fate(&b2), Some(Fate::Applied { .. })));
    assert_eq!(fate(&a2), Some(Fate::Refused(Refusal::NoFloor)));
    let holders: Vec<Option<[u8; 32]>> = run.floor.iter().map(|c| c.holder).collect();
    assert_eq!(holders, vec![Some(alice.me), Some(bob.me), None], "the floor to alice, then bob, then taken back; the refused give and release pass nothing");
    assert_eq!(run.floor_refused.len(), 2);
    assert_eq!(run.latencies.len(), 1);
    assert_eq!(run.totals[&host_pk].applied, 3);
    // The clock's chain records mallory's removal.
    let removed = chains[0].iter().filter_map(|r| wai::live_events::Event::parse(r).ok()).any(|e| {
        matches!(wai::live_events::parse_payload(e.kind, &e.payload, u16::MAX), Ok(wai::live_events::Payload::Lifecycle { subject, detail, .. }) if subject == mallory.me && detail == [wai::live_room::leave::REMOVED])
    });
    assert!(removed, "the clock's leave of mallory, reason 2");
    // And the session as the host holds it.
    let run = rd::verify_opened(&rd::VerifyInput { descriptor: &report.descriptor, grants: &report.grants, head: None, chains: &report.chains }, &[report.opening_key.0]);
    assert_eq!(run.outcome, rd::Outcome::Accept, "the session the host holds verifies, at the opened level");
    assert_eq!(run.totals[&host_pk].applied, 3);
    server.shutdown().await;
}

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

/// The bound: 40 records of the room's principals.
const BOUND: usize = 40;

async fn bound() {
    let keys = Keys::generate();
    let server = relay(&keys).await;
    let url = format!("ws://{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
    let cfg = Config::from_toml(&format!(
        "listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"{url}\"\n[room]\nmax_present = 8\njoin_wait_s = 30\n[rate]\nclient_burst = 200\nclient_every_s = 1\n[remote]\nwidth = {W}\nheight = {H}\nfps = 20\nallow = \"all\"\ncontrol = 3\nmax_age_ms = 3000\nmax_records = {BOUND}\n"
    ))
    .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();
    svc.start();
    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 service = RoomService::new(&format!("http://{addr}")).unwrap();
    let host_key = SigningKey::from_bytes(&[0x51; 32]);
    let pattern = Pattern::new(W, H);
    let hc = HostConfig {
        relay_url: url.clone(),
        trust: Trust::system(),
        opening_key: None,
        room: Room::Service { url: format!("http://{addr}"), key: host_key.clone(), invitation: Some(remote::new_invitation(&keys, 600)) },
        fps: 20,
        max_age_ms: 3_000,
        floor: rd::floor_mode::OPEN,
        give_on_request: false,
        redactions: vec![],
        clip_max: 1000,
        pointer: true,
        key_every: Duration::from_secs(2),
        run_for: Some(Duration::from_secs(180)),
        entered: None,
    };
    let (_cmd, cmds) = tokio::sync::mpsc::unbounded_channel();
    let recorder = Recorder::new(pattern.drawn.clone());
    let host = tokio::spawn(agent::run(hc, Box::new(pattern), Box::new(Ipcm::new(W, H).unwrap()), Box::new(recorder), cmds));
    timeout(STEP, async {
        loop {
            if let Ok(v) = service.call("GET", "/room", None, &[]).await
                && v.body["remote"]["screen"].is_object()
            {
                return;
            }
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
    })
    .await
    .expect("the host's screen statement reaches the clock");
    let mut alice = Viewer::join_service(&service, &url, "controller").await.expect("alice joins as a controller");
    assert!(alice.pump(STEP, |v| v.joined && v.screen.is_some()).await, "alice joined and holds the screen statement");
    // Alice sends input until the clock ends the session: a move and a sealed key each time.
    let mut sent = 0;
    while alice.left.is_none() && sent < 4 * BOUND {
        alice.input(vec![at(0, Action::Move { x: 5, y: 5 }), at(0, Action::Key { usage: 0x04, down: true, modifiers: 0 })]).await.unwrap();
        sent += 1;
        alice.pump(Duration::from_millis(60), |v| v.left.is_some()).await;
    }
    assert!(alice.pump(STEP, |v| v.left.is_some()).await, "the clock ends the session: alice's leave ({sent} inputs sent)");
    assert_eq!(alice.left, Some(wai::live_room::leave::REMOVED), "removed");
    let report = timeout(Duration::from_secs(60), host).await.expect("the host stops").expect("the task").expect("a session the clock ended is a report");
    assert!(report.ended, "the host stopped at its own leave");
    let view = service.call("GET", "/room", None, &[]).await.unwrap().body;
    assert_eq!(view["remote"]["ended"], "max_records", "the room states why: {}", view["remote"]);
    assert!(view["remote"]["records_over"].as_u64().unwrap() >= 1);
    assert!(view["remote"]["records"].as_u64().unwrap() <= BOUND as u64, "the store holds no record past the bound: {}", view["remote"]);
    // Alice's records past the end reach the clock and are not held.
    for _ in 0..5 {
        let _ = alice.input_on(alice.screen.as_ref().unwrap().0, vec![at(0, Action::Move { x: 6, y: 6 })]).await;
    }
    tokio::time::sleep(Duration::from_millis(800)).await;
    let view = service.call("GET", "/room", None, &[]).await.unwrap().body;
    assert!(view["remote"]["records"].as_u64().unwrap() <= BOUND as u64, "still within the bound: {}", view["remote"]);
    // The session the clock holds is whole and verifies: every principal's join to its leave.
    let host_grant = view["grants"].as_array().unwrap().iter().filter_map(|g| hex::decode(g.as_str()?).ok()).find(|g| joule_moq::grant_proof::grant_digest_and_principal(g).is_some_and(|(_, p)| p == host_key.verifying_key().to_bytes())).expect("the host's grant");
    let room = service.session(&host_grant, &host_key).await.unwrap();
    let hexes = |v: &Value| v.as_array().unwrap().iter().map(|x| hex::decode(x.as_str().unwrap()).unwrap()).collect::<Vec<_>>();
    let descriptor = hex::decode(room["descriptor"].as_str().unwrap()).unwrap();
    let grants = hexes(&room["grants"]);
    let chains: Vec<Vec<Vec<u8>>> = room["chains"].as_array().unwrap().iter().map(hexes).collect();
    let held: usize = chains[1..].iter().map(Vec::len).sum();
    assert!(held <= BOUND, "{held} records of the principals, the bound {BOUND}");
    let run = rd::verify_opened(&rd::VerifyInput { descriptor: &descriptor, grants: &grants, head: None, chains: &chains }, &[report.opening_key.0]);
    assert_eq!(run.outcome, rd::Outcome::Accept, "the ended session verifies");
    let leaves = chains[0].iter().filter_map(|r| wai::live_events::Event::parse(r).ok()).filter_map(|e| match wai::live_events::parse_payload(e.kind, &e.payload, u16::MAX) {
        Ok(wai::live_events::Payload::Lifecycle { subject, detail, .. }) if e.kind == wai::live_events::kind::LEAVE => Some((subject, detail)),
        _ => None,
    });
    let leaves: Vec<_> = leaves.collect();
    assert_eq!(leaves.last().map(|l| l.0), Some(host_key.verifying_key().to_bytes()), "the host leaves last");
    assert!(leaves.iter().all(|l| l.1 == [wai::live_room::leave::REMOVED]), "every leave is a removal: {leaves:?}");
    // The next host starts a new room.
    let (_, next) = service.grant(&SigningKey::from_bytes(&[0x52; 32]).verifying_key().to_bytes(), "host", Some(remote::new_invitation(&keys, 600).as_str())).await.expect("a new room for the next host");
    assert_ne!(next.room, view["room"].as_str().map(|r| <[u8; 32]>::try_from(hex::decode(r).unwrap()).unwrap()).unwrap(), "a new room");
    alice.close();
    server.shutdown().await;
}

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

/// Bob's bound, the session's, and how many records bob sends: past his own bound, and far
/// enough past it that without that bound he would fill the session's.
const PER: u64 = 24;
const TOTAL: usize = 120;
const FLOOD: usize = 120;

async fn flood() {
    let keys = Keys::generate();
    let server = relay(&keys).await;
    let url = format!("ws://{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
    let cfg = Config::from_toml(&format!(
        "listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"{url}\"\n[room]\nmax_present = 8\njoin_wait_s = 30\n[rate]\nclient_burst = 200\nclient_every_s = 1\n[remote]\nwidth = {W}\nheight = {H}\nfps = 20\nallow = \"all\"\ncontrol = 3\nmax_age_ms = 3000\nmax_records_per_principal = {PER}\nmax_records = {TOTAL}\n"
    ))
    .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();
    svc.start();
    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 service = RoomService::new(&format!("http://{addr}")).unwrap();
    let host_key = SigningKey::from_bytes(&[0x61; 32]);
    let host_pk = host_key.verifying_key().to_bytes();
    let pattern = Pattern::new(W, H);
    let recorder = Recorder::new(pattern.drawn.clone());
    let injected = recorder.log.clone();
    let hc = HostConfig {
        relay_url: url.clone(),
        trust: Trust::system(),
        opening_key: None,
        room: Room::Service { url: format!("http://{addr}"), key: host_key.clone(), invitation: Some(remote::new_invitation(&keys, 600)) },
        fps: 20,
        max_age_ms: 3_000,
        floor: rd::floor_mode::OPEN,
        give_on_request: false,
        redactions: vec![],
        clip_max: 1000,
        pointer: true,
        key_every: Duration::from_secs(2),
        run_for: Some(Duration::from_secs(180)),
        entered: None,
    };
    let (cmd, cmds) = tokio::sync::mpsc::unbounded_channel();
    let host = tokio::spawn(agent::run(hc, Box::new(pattern), Box::new(Ipcm::new(W, H).unwrap()), Box::new(recorder), cmds));
    let view = timeout(STEP, async {
        loop {
            if let Ok(v) = service.call("GET", "/room", None, &[]).await
                && v.body["remote"]["screen"].is_object()
            {
                return v.body;
            }
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
    })
    .await
    .expect("the host's screen statement reaches the clock");
    assert_eq!(view["remote"]["max_records_per_principal"], PER, "the room states the bound per principal");
    let mut alice = Viewer::join_service(&service, &url, "controller").await.expect("alice joins as a controller");
    let mut bob = Viewer::join_service(&service, &url, "viewer").await.expect("bob joins as a viewer");
    for (n, v) in [("alice", &mut alice), ("bob", &mut bob)] {
        assert!(v.pump(STEP, |v| v.joined && v.screen.is_some()).await, "{n} joined and holds the screen statement");
    }
    // Alice's typing, applied.
    let a1 = alice.input(vec![at(1_000, Action::Move { x: 5, y: 5 }), at(0, Action::Key { usage: 0x04, down: true, modifiers: 0 })]).await.unwrap();
    assert!(alice.pump(STEP, |v| v.outcomes.contains_key(&a1)).await);
    assert_eq!(alice.outcomes[&a1].1.reason, 0, "alice's input applied");
    // Bob floods: input statements the host refuses, within his grant's rate.
    for k in 0..FLOOD {
        if bob.left.is_some() {
            break;
        }
        bob.input(vec![at(0, Action::Move { x: (k % 200) as u16, y: 7 })]).await.unwrap();
        bob.pump(Duration::from_millis(12), |v| v.left.is_some()).await;
    }
    assert!(bob.pump(STEP, |v| v.left.is_some()).await, "the clock removes bob");
    assert_eq!(bob.left, Some(wai::live_room::leave::REMOVED));
    let host_grant = view["grants"]
        .as_array()
        .unwrap()
        .iter()
        .filter_map(|g| hex::decode(g.as_str()?).ok())
        .find(|g| joule_moq::grant_proof::grant_digest_and_principal(g).is_some_and(|(_, p)| p == host_pk))
        .expect("the host's grant");
    let state = service.with_proof("GET", "/remote/state", None, &host_grant, &host_key).await.unwrap().body;
    let cut = state["cut"].as_array().unwrap();
    assert_eq!(cut.len(), 1, "bob alone is cut: {cut:?}; the session ended: {}", state["remote"]["ended"]);
    assert_eq!((cut[0]["principal"].as_str(), cut[0]["reason"].as_str(), cut[0]["seq"].as_u64()), (Some(hex::encode(bob.me).as_str()), Some("max_records_per_principal"), Some(PER)), "at his bound");
    assert_eq!(state["remote"]["ended"], Value::Null, "the session goes on");
    // Alice and the host go on.
    let a2 = alice.input(vec![at(0, Action::Move { x: 40, y: 40 }), at(0, Action::Text("after the cut".into()))]).await.unwrap();
    assert!(alice.pump(STEP, |v| v.outcomes.contains_key(&a2)).await, "the host answers alice after bob's removal");
    assert_eq!(alice.outcomes[&a2].1.reason, 0, "alice's input after the cut applied");
    alice.pump(Duration::from_millis(700), |_| false).await;
    cmd.send(Command::Stop).unwrap();
    let report = timeout(Duration::from_secs(60), host).await.expect("the host stops").expect("the task").expect("the session");
    assert!(!report.ended);
    assert_eq!(report.cut, vec![(bob.me, "max_records_per_principal")], "the host cut bob where the clock did");
    assert_eq!(injected.lock().unwrap().iter().filter(|a| matches!(a, Action::Text(t) if t == "after the cut")).count(), 1);
    tokio::time::sleep(Duration::from_millis(800)).await;
    // The session the clock holds, and the one the host holds, verify at both levels.
    let room = service.session(&host_grant, &host_key).await.unwrap();
    let hexes = |v: &Value| v.as_array().unwrap().iter().map(|x| hex::decode(x.as_str().unwrap()).unwrap()).collect::<Vec<_>>();
    let descriptor = hex::decode(room["descriptor"].as_str().unwrap()).unwrap();
    let grants = hexes(&room["grants"]);
    let chains: Vec<Vec<Vec<u8>>> = room["chains"].as_array().unwrap().iter().map(hexes).collect();
    let of = |p: [u8; 32]| chains.iter().find(|c| c.first().and_then(|r| wai::live_events::Event::parse(r).ok()).is_some_and(|e| e.principal == p)).map_or(0, Vec::len);
    assert_eq!(of(bob.me) as u64, PER, "bob's records up to his bound, none past it");
    let held: usize = chains[1..].iter().map(Vec::len).sum();
    assert!(held < TOTAL, "the session's total is not reached: {held}");
    for (who, descriptor, grants, chains) in [("the clock", &descriptor, &grants, &chains), ("the host", &report.descriptor, &report.grants, &report.chains)] {
        let vi = rd::VerifyInput { descriptor, grants, head: None, chains };
        let run = rd::verify(&vi);
        assert_eq!(run.outcome, rd::Outcome::Accept, "the session {who} holds verifies");
        let opened = rd::verify_opened(&vi, &[report.opening_key.0]);
        assert_eq!(opened.outcome, rd::Outcome::Accept, "and at the opened level");
        let fate = |d: &[u8; 32]| run.inputs.iter().find(|f| f.digest == *d).map(|f| f.fate);
        assert!(matches!(fate(&a2), Some(Fate::Applied { .. })), "{who}: alice's input after the cut");
        assert_eq!(run.totals[&host_pk].refused[Refusal::NotPermitted.number() as usize - 1] as u64, PER, "{who}: every input of bob's it holds refused");
    }
    alice.close();
    bob.close();
    server.shutdown().await;
}