//! The floor of a WAI room (WAI co-performance §6), through a relay on loopback that holds
//! it: the room's clock publishes a turn schedule that refuses on its events track, and two
//! performers each publish an audio track that one subscriber follows. The object of the
//! performer whose turn it is reaches the subscriber, and the other's does not; once the first
//! hands its turn to the second on its own events track, the second's reaches the subscriber
//! and the first's does not. Each refusal is counted.
//!
//! And the room's audience inputs (§7.3), through a relay that honours a room authority's
//! grants: the clock opens a vote on its events track; a member whose grant carries the
//! audience-input power sends one vote that a subscriber of its events track receives, and a
//! second, past its power, that the relay withholds; a viewer whose grant carries no such power
//! has its vote withheld; each refusal is counted in its session's summary by reason.
//!
//! And the order the relay counts them in (§7.3), its author's chain's: a member's second vote,
//! sent first on a stream of its own, waits for its first, which is forwarded, and is then past
//! the member's power; a second vote whose first never comes is refused once the relay's hold
//! runs out, and the first, sent after, is forwarded.

mod common;

use common::*;
use ed25519_dalek::{Signer, SigningKey};
use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::Parameters;
use joule_moq::TrackNamespace;
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{Authority, FloorConfig, Grant};
use joule_moq_relay::floor::Place;
use joule_moq_relay::wire::{ReqRx, ReqTx};
use joule_moq_relay::RelayServer;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::time::timeout;

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

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

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

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

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

/// A `WPH1` handover (co-performance §6.3).
fn handover(at: u64, to: &SigningKey) -> Vec<u8> {
    let mut b = b"WPH1\x01".to_vec();
    b.extend_from_slice(&at.to_le_bytes());
    b.extend_from_slice(&to.verifying_key().to_bytes());
    b
}

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

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

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

async fn run() {
    let room = [0x61; 32];
    let clock = SigningKey::from_bytes(&[0x62; 32]);
    let (a, b) = (SigningKey::from_bytes(&[0x63; 32]), SigningKey::from_bytes(&[0x64; 32]));
    let server = relay(|c| c.floor = Some(FloorConfig { room_keys: vec![hexkey(&clock)], ..FloorConfig::default() })).await;
    let relay = server.relay().clone();
    let floor = || relay.floor().expect("the relay holds the floor");
    // The room began ten seconds ago; its clock gives the floor to a, then b, ten minutes each.
    let t0 = now_us() - 10_000_000;
    let (c, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let _ce = c.publish(track(room, &clock, "events"), name("0"), 1, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the clock's events");
    let rc = now_us() - t0;
    send(&c, 1, 0, &msf(&meta(&clock, room, 0, rc, &schedule(t0, rc, 600_000_000, &[&a, &b])))).await;
    let place = |who: &SigningKey| floor().place(&track(room, who, "audio"), now_us());
    assert!(until(STEP, || place(&a) == Some(Place::InTurn)).await, "the relay reads the schedule");
    assert_eq!(place(&b), Some(Place::OutOfTurn(1)));

    // Both performers publish audio; one subscriber follows both.
    let (pa, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let (pb, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let _ha = pa.publish(track(room, &a, "audio"), name("0"), 10, Parameters::new(), track_props(0, None)).await.expect("PUBLISH a's audio");
    let _hb = pb.publish(track(room, &b, "audio"), name("0"), 20, Parameters::new(), track_props(0, None)).await.expect("PUBLISH b's audio");
    let (s, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let _sa = s.subscribe(track(room, &a, "audio"), name("0"), Parameters::new()).await.expect("SUBSCRIBE a's audio");
    let _sb = s.subscribe(track(room, &b, "audio"), name("0"), Parameters::new()).await.expect("SUBSCRIBE b's audio");
    send(&pa, 10, 1, b"a in its turn").await;
    send(&pb, 20, 1, b"b out of turn").await;
    // What reaches the subscriber: the object in turn, waited for; then, once the relay has held
    // the other back, anything more for a second.
    let has = |g: &[Got], p: &[u8]| g.iter().any(|x| x.objects.iter().any(|o| o.payload == p));
    let mut got = collect(&s, STEP, |g| has(g, b"a in its turn")).await;
    assert!(until(STEP, || floor().refused() >= 1).await, "the relay holds back the object out of turn");
    got.extend(collect(&s, Duration::from_secs(1), |_| false).await);
    let payloads: Vec<&[u8]> = got.iter().flat_map(|g| g.objects.iter().map(|o| o.payload.as_slice())).collect();
    assert_eq!(payloads, vec![b"a in its turn".as_slice()], "only the object in turn is forwarded");
    assert_eq!(floor().refused(), 1);

    // a hands its turn to b, from now.
    let _ae = pa.publish(track(room, &a, "events"), name("0"), 11, Parameters::new(), track_props(16, None)).await.expect("PUBLISH a's events");
    let at = now_us() - t0;
    send(&pa, 11, 0, &msf(&meta(&a, room, 0, at, &handover(at, &b)))).await;
    assert!(until(STEP, || place(&b) == Some(Place::InTurn)).await, "the relay reads the handover");
    send(&pa, 10, 2, b"a after its handover").await;
    send(&pb, 20, 2, b"b in the turn it was handed").await;
    let mut got = collect(&s, STEP, |g| has(g, b"b in the turn it was handed")).await;
    assert!(until(STEP, || floor().refused() >= 2).await, "the relay holds back a's object after its handover");
    got.extend(collect(&s, Duration::from_secs(1), |_| false).await);
    let payloads: Vec<&[u8]> = got.iter().flat_map(|g| g.objects.iter().map(|o| o.payload.as_slice())).collect();
    assert_eq!(payloads, vec![b"b in the turn it was handed".as_slice()]);
    assert_eq!(floor().refused(), 2);
    for x in [&c, &pa, &pb, &s] {
        x.close(0, "done");
    }
    server.shutdown().await;
}

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

/// A `meta` record with a `prev`, so a chain of several reads as one.
fn meta_after(key: &SigningKey, room: [u8; 32], seq: u64, prev: [u8; 32], t: u64, payload: &[u8]) -> Vec<u8> {
    let mut r = meta(key, room, seq, t, payload);
    r[77..109].copy_from_slice(&prev);
    let body = r.len() - 64;
    let mut m = b"wai:live-event\x01".to_vec();
    m.extend_from_slice(&r[..body]);
    let sig = key.sign(&m).to_bytes();
    r[body..].copy_from_slice(&sig);
    r
}

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

/// A record's digest (live-events §3.2): over all of it but its signature.
fn digest(record: &[u8]) -> [u8; 32] {
    sha256(&record[..record.len() - 64])
}

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

async fn inputs() {
    let room = [0x71; 32];
    let authority = SigningKey::from_bytes(&[0x72; 32]);
    let clock = SigningKey::from_bytes(&[0x73; 32]);
    let (member, viewer, watcher) = (SigningKey::from_bytes(&[0x74; 32]), SigningKey::from_bytes(&[0x75; 32]), SigningKey::from_bytes(&[0x76; 32]));
    let token = b"the clock's token".to_vec();
    let token_sha = hex::encode(sha256(&token));
    let server = relay(|c| {
        c.auth.required = true;
        c.auth.authorities = vec![Authority { key: hexkey(&authority), rooms: vec![] }];
        c.auth.grants = vec![Grant { prefix: vec!["*".into(), hexkey(&clock), "events".into()], token_sha256: token_sha.clone(), publish: true, subscribe: true, proof: false, unlimited: true, seek: false }];
        c.floor = Some(FloorConfig { room_keys: vec![hexkey(&clock)], ..FloorConfig::default() });
    })
    .await;
    let target = format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
    let with = |g: Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g, k.clone())], target: target.clone(), ..ClientOptions::default() };
    let t0 = now_us() - 10_000_000;
    let rc = || now_us() - t0;

    // The clock opens a vote of three options, closing in a minute.
    let (c, _) = connect(&server, Via::WebSocket, ClientOptions { tokens: vec![joule_moq::draft22::Token::UseValue { token_type: 0, value: token.clone() }], ..ClientOptions::default() }).await;
    let _ce = c.publish(track(room, &clock, "events"), name("0"), 1, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the clock's events");
    let mut w = b"WPW1\x01".to_vec();
    w.extend_from_slice(&[1, 0b100, 3, 1]);
    w.extend_from_slice(&(rc() + 60_000_000).to_le_bytes());
    w.extend_from_slice(&[0; 32]);
    w.extend_from_slice(&0u16.to_le_bytes());
    let wr = meta(&clock, room, 0, rc(), &w);
    let wd = digest(&wr);
    send(&c, 1, 0, &msf(&wr)).await;
    let vote = |o: u8| {
        let mut b = b"WPA1\x01".to_vec();
        b.extend_from_slice(&wd);
        b.extend_from_slice(&[3, o]);
        b
    };

    // A watcher follows the member's and the viewer's events tracks.
    let (s, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &watcher, &[]), &watcher)).await;
    // The member may send one input to a window; the viewer none.
    let (pm, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &member, &[("audience_input", "1")]), &member)).await;
    let (pv, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &viewer, &[]), &viewer)).await;
    let _pme = pm.publish(track(room, &member, "events"), name("0"), 2, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the member's events");
    let _pve = pv.publish(track(room, &viewer, "events"), name("0"), 3, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the viewer's events");
    let _sm = s.subscribe(track(room, &member, "events"), name("0"), Parameters::new()).await.expect("SUBSCRIBE the member's events");
    let _sv = s.subscribe(track(room, &viewer, "events"), name("0"), Parameters::new()).await.expect("SUBSCRIBE the viewer's events");
    // The inputs name the window: the floor holds it before the first is sent.
    let f = server.relay().floor().unwrap();
    assert!(until(STEP, || f.holds_window(&room, &wd)).await, "the relay holds the clock's window");
    // The member's first vote, forwarded: the watcher's subscription is live. Each group rides a
    // stream of its own, which the relay reads in a task of its own, so the second is sent once
    // the first has been taken (`the_relay_counts_inputs_in_chain_order` sends them the other way
    // round).
    let first = meta(&member, room, 0, rc(), &vote(1));
    send(&pm, 2, 0, &msf(&first)).await;
    let mut got = collect(&s, STEP, |g| g.iter().any(|x| x.objects.iter().any(|o| o.payload == msf(&first)))).await;
    // Its second, past its power, and the viewer's, with no power: both refused, and nothing
    // more reaches the watcher once the relay has refused them.
    let second = meta_after(&member, room, 1, digest(&first), rc(), &vote(2));
    send(&pm, 2, 1, &msf(&second)).await;
    send(&pv, 3, 0, &msf(&meta(&viewer, room, 0, rc(), &vote(0)))).await;
    assert!(until(STEP, || f.refused_inputs() >= 2).await, "the relay refuses the second vote and the viewer's");
    got.extend(collect(&s, Duration::from_secs(1), |_| false).await);
    let payloads: Vec<&[u8]> = got.iter().flat_map(|g| g.objects.iter().map(|o| o.payload.as_slice())).collect();
    assert_eq!(payloads, vec![msf(&first).as_slice()], "only the member's first vote is forwarded");
    assert_eq!(f.refused_inputs(), 2);
    for x in [&pm, &pv, &s, &c] {
        x.close(0, "done");
    }
    let relay = server.relay().clone();
    assert!(until(STEP, || relay.summaries().len() >= 4).await, "every session's summary");
    let refused: Vec<(String, u64)> = relay.summaries().iter().flat_map(|x| x.summary.tally.publisher.co_refused.clone()).collect();
    let mut refused = refused;
    refused.sort();
    assert_eq!(refused, vec![("input_not_permitted".to_string(), 1), ("input_over_limit".to_string(), 1)]);
    server.shutdown().await;
}

/// A room whose clock has opened a vote of three options, a member whose grant allows one input
/// to a window, and a watcher of the member's events track, through a relay that holds an input
/// ahead of a gap in its author's chain for `hold_ms`.
struct Audience {
    server: RelayServer,
    room: [u8; 32],
    member: SigningKey,
    window: [u8; 32],
    t0: u64,
    clients: [Client; 3],
    _held: Vec<(ReqTx, ReqRx)>,
}

impl Audience {
    async fn open(room: [u8; 32], seed: u8, hold_ms: u64) -> Audience {
        let authority = SigningKey::from_bytes(&[seed; 32]);
        let clock = SigningKey::from_bytes(&[seed + 1; 32]);
        let (member, watcher) = (SigningKey::from_bytes(&[seed + 2; 32]), SigningKey::from_bytes(&[seed + 3; 32]));
        let token = b"the clock's token".to_vec();
        let token_sha = hex::encode(sha256(&token));
        let server = relay(|c| {
            c.auth.required = true;
            c.auth.authorities = vec![Authority { key: hexkey(&authority), rooms: vec![] }];
            c.auth.grants = vec![Grant { prefix: vec!["*".into(), hexkey(&clock), "events".into()], token_sha256: token_sha, publish: true, subscribe: true, proof: false, unlimited: true, seek: false }];
            c.floor = Some(FloorConfig { room_keys: vec![hexkey(&clock)], hold_ms });
        })
        .await;
        let target = format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
        let with = |g: Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g, k.clone())], target: target.clone(), ..ClientOptions::default() };
        let t0 = now_us() - 10_000_000;
        let (c, _) = connect(&server, Via::WebSocket, ClientOptions { tokens: vec![joule_moq::draft22::Token::UseValue { token_type: 0, value: token }], ..ClientOptions::default() }).await;
        let mut held = vec![c.publish(track(room, &clock, "events"), name("0"), 1, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the clock's events")];
        let mut w = b"WPW1\x01".to_vec();
        w.extend_from_slice(&[1, 0b100, 3, 1]);
        w.extend_from_slice(&(now_us() - t0 + 60_000_000).to_le_bytes());
        w.extend_from_slice(&[0; 32]);
        w.extend_from_slice(&0u16.to_le_bytes());
        let wr = meta(&clock, room, 0, now_us() - t0, &w);
        send(&c, 1, 0, &msf(&wr)).await;
        let (s, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &watcher, &[]), &watcher)).await;
        let (pm, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &member, &[("audience_input", "1")]), &member)).await;
        held.push(pm.publish(track(room, &member, "events"), name("0"), 2, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the member's events"));
        let (_, tx, rx) = s.subscribe(track(room, &member, "events"), name("0"), Parameters::new()).await.expect("SUBSCRIBE the member's events");
        held.push((tx, rx));
        let window = digest(&wr);
        let f = server.relay().floor().unwrap();
        assert!(until(STEP, || f.holds_window(&room, &window)).await, "the relay holds the clock's window");
        Audience { server, room, member, window, t0, clients: [c, pm, s], _held: held }
    }

    /// The member's vote for option `o` at seq `seq`, chained after `prev`.
    fn vote(&self, seq: u64, prev: Option<&[u8]>, o: u8) -> Vec<u8> {
        let mut b = b"WPA1\x01".to_vec();
        b.extend_from_slice(&self.window);
        b.extend_from_slice(&[3, o]);
        let t = now_us() - self.t0;
        match prev {
            None => meta(&self.member, self.room, seq, t, &b),
            Some(p) => meta_after(&self.member, self.room, seq, digest(p), t, &b),
        }
    }

    /// What the watcher receives of the member's events: until `want` arrives, then a second more.
    async fn received(&self, want: &[u8]) -> Vec<Vec<u8>> {
        let s = &self.clients[2];
        let mut got = collect(s, STEP, |g| g.iter().any(|x| x.objects.iter().any(|o| o.payload == want))).await;
        got.extend(collect(s, Duration::from_secs(1), |_| false).await);
        got.iter().flat_map(|g| g.objects.iter().map(|o| o.payload.clone())).collect()
    }

    /// Every session closed: the refusals its summaries count, by reason.
    async fn close(self) -> Vec<(String, u64)> {
        for x in &self.clients {
            x.close(0, "done");
        }
        let relay = self.server.relay().clone();
        assert!(until(STEP, || relay.summaries().len() >= 3).await, "every session's summary");
        let mut refused: Vec<(String, u64)> = relay.summaries().iter().flat_map(|x| x.summary.tally.publisher.co_refused.clone()).collect();
        refused.sort();
        self.server.shutdown().await;
        refused
    }
}

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

async fn chain_order() {
    // A hold past any wait of this test, however loaded the host: the relay is to wait for seq 0.
    let a = Audience::open([0x81; 32], 0x82, 60_000).await;
    let f = a.server.relay().floor().unwrap();
    let first = a.vote(0, None, 1);
    let second = a.vote(1, Some(&first), 2);
    // The member's two votes, each a group on a stream of its own, the second opened first: the
    // relay holds seq 1 until seq 0 arrives.
    send(&a.clients[1], 2, 1, &msf(&second)).await;
    let held = until(STEP, || f.waiting(&a.room) == 1).await;
    assert_eq!(f.refused_inputs(), 0);
    send(&a.clients[1], 2, 0, &msf(&first)).await;
    // Seq 0 is counted first and forwarded; seq 1, counted after it, is past the member's power.
    // (A relay that counted them as they came would have forwarded seq 1 and refused seq 0.)
    let got = a.received(&msf(&first)).await;
    assert!(until(STEP, || f.refused_inputs() >= 1).await, "the relay refuses one of them");
    assert_eq!(got, vec![msf(&first)], "seq 0 forwarded, seq 1 withheld");
    assert!(held, "the relay held seq 1 for seq 0");
    assert_eq!((f.refused_inputs(), f.waiting(&a.room)), (1, 0));
    assert_eq!(a.close().await, vec![("input_over_limit".to_string(), 1)]);
}

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

async fn gap_never_fills() {
    let hold = Duration::from_millis(400);
    let a = Audience::open([0x91; 32], 0x92, hold.as_millis() as u64).await;
    let f = a.server.relay().floor().unwrap();
    let first = a.vote(0, None, 1);
    let second = a.vote(1, Some(&first), 2);
    // Seq 1 alone, seq 0 held back by its sender: the relay refuses seq 1 once its hold runs out.
    let sent = Instant::now();
    send(&a.clients[1], 2, 1, &msf(&second)).await;
    assert!(until(STEP, || f.refused_inputs() >= 1).await, "the relay refuses seq 1");
    assert!(sent.elapsed() >= hold, "not before its hold ran out: {:?}", sent.elapsed());
    assert_eq!(f.waiting(&a.room), 0);
    // Seq 0, sent now, is forwarded: no input after it was.
    send(&a.clients[1], 2, 0, &msf(&first)).await;
    let got = a.received(&msf(&first)).await;
    assert_eq!(got, vec![msf(&first)], "seq 1 withheld, seq 0 forwarded");
    assert_eq!(f.refused_inputs(), 1);
    assert_eq!(a.close().await, vec![("input_out_of_order".to_string(), 1)]);
}