//! The open sessions stated on the operator's surface, and one request sent from a shell
//! (`src/sessions.rs`, `src/control.rs`, `wai-relay --control`). Each check is held against a
//! fault planted for it:
//!
//! - **`sessions`.** A publisher at -22, a viewer at -23 and a session that holds nothing: the
//!   statement lists the first two under their room with the draft each speaks and when it
//!   opened, the third under `other`, signed by the relay's key. The room reads `sealed` from
//!   its first sealed object on, for every session of it, and a second room in the clear does
//!   not. Planted: a sealed room stated in the clear after signing, and a session left out
//!   after signing, each refused by the signature; a statement signed by another key, caught by
//!   the key it names; and a session that left, which the next statement no longer lists.
//! - **A session's grants and tracks.** A publisher and a viewer, each under a room grant it
//!   proved: each states its grant by digest with the room and role it names, and the
//!   publisher its tracks; the viewer's key is in no field. Planted: a role and a track list
//!   changed after signing, each refused by the signature.
//! - **`wai-relay --control`.** An operator on an edge's own host closes its window with the
//!   program, joins the cut with the core's, and seals with the program; the two relays' link
//!   chains exactly over the window, and the edge's viewer is sent no GOAWAY. Planted: a seal of
//!   a window that is not the next (exit 1, the refusal printed, nothing sealed), an op the
//!   surface does not know (exit 1), two requests on one line (exit 2, nothing sent), and a
//!   socket nobody serves (exit 3). The socket keeps its mode, 0600.

#![cfg(unix)]

mod common;

use common::*;
use ed25519_dalek::SigningKey;
use joule_moq::draft22::data::{ObjectStatus as WireStatus, SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::kvp::{KeyValuePair, Properties};
use joule_moq::draft22::Parameters;
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{UpstreamConfig, UpstreamTransport};
use joule_moq_relay::sessions::SignedSessions;
use joule_moq_relay::window::{link_chain, Cut, SignedWindow};
use joule_moq_relay::RelayServer;
use serde_json::{json, Value};
use std::path::{Path, PathBuf};
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::time::timeout;

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

fn socket_path(what: &str) -> PathBuf {
    std::env::temp_dir().join(format!("wai-relay-gap-{}-{what}.sock", std::process::id()))
}

fn dir(what: &str) -> PathBuf {
    let d = std::env::temp_dir().join(format!("wai-relay-gap-{}-{what}", std::process::id()));
    let _ = std::fs::remove_dir_all(&d);
    std::fs::create_dir_all(&d).unwrap();
    d
}

/// One request on the control socket at `path`, and its answer.
async fn ask(path: &Path, req: Value) -> Value {
    timeout(STEP * 2, async {
        let s = tokio::net::UnixStream::connect(path).await.expect("the control socket");
        let (r, mut w) = s.into_split();
        w.write_all(format!("{req}\n").as_bytes()).await.unwrap();
        let mut line = String::new();
        BufReader::new(r).read_line(&mut line).await.unwrap();
        serde_json::from_str::<Value>(&line).unwrap_or_else(|e| panic!("an answer: {e}: {line:?}"))
    })
    .await
    .expect("an answer in time")
}

async fn sessions(path: &Path, room: Option<&str>) -> SignedSessions {
    let req = match room {
        Some(r) => json!({"op": "sessions", "room": r}),
        None => json!({"op": "sessions"}),
    };
    let a = ask(path, req.clone()).await;
    assert_eq!(a["ok"], true, "{req} answered {a}");
    let s: SignedSessions = serde_json::from_value(a["statement"].clone()).expect("a signed statement of the sessions");
    s.verify().expect("the relay signed it");
    s
}

/// An object whose Immutable Properties carry a secure-objects Key ID (WAI confidentiality
/// §4.3): what a relay knows a sealed room by. Its payload is whatever the member sealed.
fn sealed_properties() -> Properties {
    let mut inner = Properties::new();
    inner.push(KeyValuePair::varint(joule_moq::secure_objects::KEY_ID_PROPERTY, 3 << 32));
    let mut p = Properties::new();
    p.set_immutable(&inner).expect("the properties encode");
    p
}

async fn publish_object(c: &Client, t: &Tracks, g: u64, properties: Properties) {
    let h = SubgroupHeader { track_alias: t.media_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: true, end_of_group: true, first_object: true };
    c.send_subgroup(h, &[SubgroupObject { object_id: 0, properties, status: WireStatus::Normal, payload: payload("s", g, 200) }]).await.expect("the object");
}

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

async fn sessions_run() {
    use joule_moq::draft::Draft;
    let (room, clear) = (hex::encode([0x81; 32]), hex::encode([0x82; 32]));
    let pk = SigningKey::from_bytes(&[0x83; 32]);
    let d = dir("sessions");
    let seed = [0x84u8; 32];
    std::fs::write(d.join("relay.key"), hex::encode(seed)).unwrap();
    let sock = socket_path("sessions");
    let started = joule_moq_relay::relay::now_unix_ms();
    let server = relay(|c| {
        c.control.socket = Some(sock.clone());
        c.summary.key_path = Some(d.join("relay.key"));
        c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
    })
    .await;
    let t = Tracks::new(ns(&[&room, &hexkey(&pk), "video"]), "0", 10, 11, 64);
    let tc = Tracks::new(ns(&[&clear, &hexkey(&pk), "video"]), "0", 20, 21, 64);
    let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _published = (publish_tracks(&p, &t, None).await, publish_tracks(&p, &tc, None).await);
    let (v, _ev) = connect(&server, Via::QuicDraft(Draft::D23), ClientOptions::default()).await;
    let _sub = subscribe_both(&v, &t, Parameters::new()).await;
    let (idle, _ei) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    assert!(until(STEP, || server.relay().stats().sessions_open == 3).await);

    let s = sessions(&sock, None).await;
    assert_eq!(s.statement.relay_key, hex::encode(SigningKey::from_bytes(&seed).verifying_key().to_bytes()), "signed by the relay's own key");
    assert_eq!(s.ids().len(), 3, "{:?}", s.statement);
    let of_room = &s.statement.rooms[&room];
    let drafts: Vec<&str> = of_room.iter().map(|x| x.draft.as_str()).collect();
    assert_eq!(of_room.len(), 2, "the publisher and the viewer hold the room: {of_room:?}");
    assert!(drafts.contains(&"moqt-22") && drafts.contains(&"moqt-23"), "the draft each speaks: {drafts:?}");
    assert!(of_room.iter().all(|x| x.transport == "quic" && x.opened_unix_ms >= started && x.opened_unix_ms <= joule_moq_relay::relay::now_unix_ms()), "when each opened: {of_room:?}");
    assert!(of_room.iter().all(|x| !x.sealed), "no sealed object has passed");
    assert_eq!(s.statement.rooms[&clear].len(), 1, "the publisher alone holds the second room");
    assert_eq!(s.statement.other.len(), 1, "the session that holds nothing");
    assert_eq!((s.statement.other[0].draft.as_str(), s.statement.other[0].transport.as_str()), ("moqt-22", "websocket"));

    // The room's first sealed object passes: every session of the room holds a sealed room's
    // state from then on, and the room in the clear does not.
    publish_object(&p, &t, 0, sealed_properties()).await;
    publish_object(&p, &tc, 0, Properties::new()).await;
    let got = collect(&v, STEP, |x| !x.is_empty()).await;
    assert_eq!(got.len(), 1, "the sealed object is forwarded as any object");
    let s = sessions(&sock, None).await;
    assert!(s.statement.rooms[&room].iter().all(|x| x.sealed) && s.statement.rooms[&room].len() == 2, "{:?}", s.statement.rooms[&room]);
    assert!(s.statement.rooms[&clear].iter().all(|x| !x.sealed), "the room in the clear");
    let one = sessions(&sock, Some(&room)).await;
    assert_eq!(one.statement.room.as_deref(), Some(room.as_str()));
    assert_eq!(one.statement.rooms.keys().collect::<Vec<_>>(), vec![&room], "that room alone");
    assert!(one.statement.other.is_empty());
    let bad = ask(&sock, json!({"op": "sessions", "room": "zz"})).await;
    assert_eq!(bad["ok"], false, "{bad}");

    // Planted, each caught.
    let mut caught = 0;
    let mut cleared = s.clone();
    cleared.statement.rooms.get_mut(&room).unwrap()[0].sealed = false;
    assert!(cleared.verify().expect_err("a sealed room stated in the clear after signing").starts_with("bad_signature"));
    caught += 1;
    let mut fewer = s.clone();
    fewer.statement.rooms.get_mut(&room).unwrap().pop();
    assert!(fewer.verify().expect_err("a session left out after signing").starts_with("bad_signature"));
    caught += 1;
    let stranger = SigningKey::from_bytes(&[0x85; 32]);
    let mut theirs = s.statement.clone();
    theirs.relay_key = hexkey(&stranger);
    theirs.rooms.clear();
    let theirs = SignedSessions::sign(theirs, &stranger);
    theirs.verify().expect("it verifies under the key it names");
    assert_ne!(theirs.statement.relay_key, hex::encode(server.relay().relay_id()), "and that key is not this relay's: an operator holds a statement to the relay it asked");
    caught += 1;
    // A session that left is not listed: a supervisor keeps no clock on it.
    let viewer = s.statement.rooms[&room].iter().find(|x| x.draft == "moqt-23").unwrap().session;
    v.close(0, "done");
    assert!(until(STEP, || server.relay().stats().sessions_open == 2).await);
    let s = sessions(&sock, None).await;
    assert!(!s.ids().contains(&viewer), "the viewer left and is still listed");
    assert_eq!(s.statement.rooms[&room].len(), 1);
    caught += 1;
    eprintln!("planted faults caught (sessions): {caught} of 4");
    assert_eq!(server.relay().control_requests("sessions"), 5, "each counted by its op");
    p.close(0, "done");
    idle.close(0, "done");
    server.shutdown().await;
    let _ = std::fs::remove_dir_all(&d);
}

// ── A session's grants and tracks ──────────────────────────────────────────

/// A grant of `room` for `principal` in the descriptor's role `role`, laid out as live-rooms
/// §4.1 lays it out, with these media bits.
fn room_grant(authority: &SigningKey, room: [u8; 32], principal: &SigningKey, role: u8, publish: u8) -> Vec<u8> {
    use ed25519_dalek::Signer;
    let now = std::time::SystemTime::now().duration_since(std::time::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(&[role, publish, 1, 0, 0, 0]);
    g.extend_from_slice(&[0; 32]);
    g.extend_from_slice(b"WEP1\x01");
    g.extend_from_slice(&[0; 32]);
    g.extend_from_slice(&[0; 28]);
    g.extend_from_slice(&[0xff; 16]);
    g.extend_from_slice(&[0; 3]);
    g.extend_from_slice(&[0, 0, 0, 0]); // no profile, no origin, no attribute
    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
}

/// What the clock of an end-to-end encrypted room writes a principal's `join` and `publish`
/// from (WAI confidentiality §5.4): each session's room grants, by digest with the room and
/// role they state, and the tracks it publishes, in `sessions`. A publisher (role 0) and a
/// viewer (role 1), each under a grant it proved; nothing of either principal beyond the track
/// names a publisher's own tracks carry (live-rooms §6.1).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_sessions_state_their_grants_and_tracks() {
    timeout(Duration::from_secs(120), grants_run()).await.expect("the test finishes in time");
}

async fn grants_run() {
    use joule_moq::grant_proof::GrantHead;
    let authority = SigningKey::from_bytes(&[0x91; 32]);
    let (pk, vk) = (SigningKey::from_bytes(&[0x92; 32]), SigningKey::from_bytes(&[0x93; 32]));
    let room = [0x94u8; 32];
    let room_hex = hex::encode(room);
    let d = dir("grants");
    std::fs::write(d.join("relay.key"), hex::encode([0x95u8; 32])).unwrap();
    let sock = socket_path("grants");
    let server = relay(|c| {
        c.control.socket = Some(sock.clone());
        c.summary.key_path = Some(d.join("relay.key"));
        c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
        c.auth.required = true;
        c.auth.authorities = vec![joule_moq_relay::config::Authority { key: hexkey(&authority), rooms: vec![] }];
    })
    .await;
    let target = format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
    let with = |g: &[u8], k: &SigningKey| ClientOptions { grants: vec![(g.to_vec(), k.clone())], target: target.clone(), ..ClientOptions::default() };
    let (gp, gv) = (room_grant(&authority, room, &pk, 0, 0b010), room_grant(&authority, room, &vk, 1, 0));
    let digest = |g: &[u8]| hex::encode(GrantHead::verified(g, &authority.verifying_key().to_bytes()).expect("the grant reads").digest);
    let t = Tracks::new(ns(&[&room_hex, &hexkey(&pk), "video"]), "0", 10, 11, 64);
    let (p, _) = connect(&server, Via::WebSocket, with(&gp, &pk)).await;
    let _published = publish_tracks(&p, &t, None).await;
    let (v, _) = connect(&server, Via::WebSocket, with(&gv, &vk)).await;
    let _sub = subscribe_both(&v, &t, Parameters::new()).await;
    assert!(until(STEP, || server.relay().stats().sessions_open == 2).await);

    let s = sessions(&sock, Some(&room_hex)).await;
    let of_room = &s.statement.rooms[&room_hex];
    assert_eq!(of_room.len(), 2, "{of_room:?}");
    let publisher = of_room.iter().find(|x| !x.published.is_empty()).expect("the publisher states its tracks");
    let viewer = of_room.iter().find(|x| x.published.is_empty()).expect("the viewer publishes nothing");
    assert_eq!(publisher.grants.len(), 1, "{publisher:?}");
    assert_eq!((publisher.grants[0].digest.clone(), publisher.grants[0].room.clone(), publisher.grants[0].role), (digest(&gp), room_hex.clone(), 0));
    assert_eq!(viewer.grants.len(), 1, "{viewer:?}");
    assert_eq!((viewer.grants[0].digest.clone(), viewer.grants[0].role), (digest(&gv), 1));
    let media = joule_moq::draft22::names::render(&t.ns, Some(&t.media));
    let receipts = joule_moq::draft22::names::render(&t.ns, Some(&t.receipts));
    assert_eq!(publisher.published, { let mut x = vec![media, receipts]; x.sort(); x }, "the media track and its receipts track");
    // Nothing that names the viewer: its key is in no field.
    let text = serde_json::to_string(&s).unwrap();
    assert!(!text.contains(&hexkey(&vk)), "the viewer's principal is stated: {text}");
    // Changed after signing, refused by the signature.
    let mut role = s.clone();
    role.statement.rooms.get_mut(&room_hex).unwrap().iter_mut().find(|x| x.published.is_empty()).unwrap().grants[0].role = 0;
    assert!(role.verify().expect_err("a viewer's role stated as a participant's after signing").starts_with("bad_signature"));
    let mut unpublished = s.clone();
    unpublished.statement.rooms.get_mut(&room_hex).unwrap().iter_mut().for_each(|x| x.published.clear());
    assert!(unpublished.verify().expect_err("a publisher's tracks left out after signing").starts_with("bad_signature"));
    p.close(0, "done");
    v.close(0, "done");
    server.shutdown().await;
    let _ = std::fs::remove_dir_all(&d);
}

// ── One request from a shell ────────────────────────────────────────────────

/// `wai-relay --control <target> '<request>'`: its exit status and what it printed.
fn control(target: &Path, request: &str) -> (i32, String, String) {
    let o = std::process::Command::new(env!("CARGO_BIN_EXE_wai-relay")).arg("--control").arg(target).arg(request).output().expect("wai-relay runs");
    (o.status.code().expect("it exited"), String::from_utf8_lossy(&o.stdout).into_owned(), String::from_utf8_lossy(&o.stderr).into_owned())
}

/// The program, off the test's workers: it blocks until the surface answers.
async fn run_control(target: &Path, request: Value) -> (i32, Value, String) {
    let (target, request) = (target.to_path_buf(), request.to_string());
    let (code, out, err) = timeout(STEP * 2, tokio::task::spawn_blocking(move || control(&target, &request))).await.expect("in time").expect("it ran");
    (code, serde_json::from_str(out.trim()).unwrap_or(Value::Null), err)
}

fn upstream_to(server: &RelayServer, prefixes: &[&[&str]]) -> UpstreamConfig {
    UpstreamConfig {
        name: "core".into(),
        url: format!("ws://{}{}", server.websocket_addr.unwrap(), server.relay().config().path),
        transport: UpstreamTransport::WebSocket,
        prefixes: prefixes.iter().map(|p| p.iter().map(|f| f.to_string()).collect()).collect(),
        ca_path: None,
        cert_sha256: None,
        native_roots: false,
        token_sha256: None,
        token_path: None,
        relay_key: Some(hex::encode(server.relay().relay_id())),
    }
}

fn forwarded(server: &RelayServer) -> u64 {
    server.relay().open_tallies().iter().map(|(_, t)| t.subscriber.objects_forwarded).sum()
}

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

async fn program_run() {
    use std::os::unix::fs::PermissionsExt;
    let room = hex::encode([0x91; 32]);
    let pk = SigningKey::from_bytes(&[0x92; 32]);
    let (sc, se) = (socket_path("core"), socket_path("edge"));
    let quiet = |c: &mut joule_moq_relay::RelayConfig| {
        c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
        c.control.flush_grace_ms = 300;
        c.limits.upstream_reconnect_ms = 100;
    };
    let core = relay(|c| {
        quiet(c);
        c.control.socket = Some(sc.clone());
    })
    .await;
    let up = upstream_to(&core, &[&[&room]]);
    let edge = relay(|c| {
        quiet(c);
        c.control.socket = Some(se.clone());
        c.upstream = vec![up];
    })
    .await;
    assert!(until(STEP, || edge.relay().links().iter().any(|l| l.session().is_some())).await, "the edge's link to the core opens");
    // The link: at -23, the draft both relays prefer, opened on the edge's second dial. Its
    // first SETUP held no challenge of the core's to prove the edge's id over, was refused and
    // given one, and the second proved over it.
    let link = &edge.relay().links()[0];
    assert_eq!(link.session().expect("open").conn.draft(), joule_moq::draft::Draft::D23, "relays peer at -23");
    let dials = link.stats.dials.load(std::sync::atomic::Ordering::Relaxed);
    assert_eq!((dials, link.stats.opened.load(std::sync::atomic::Ordering::Relaxed)), (2, 1), "a challenge on the first dial, the session on the second");
    let stated: joule_moq_relay::sessions::SignedSessions = serde_json::from_value(ask(&sc, json!({"op": "sessions"})).await["statement"].clone()).expect("the core's sessions");
    assert!(stated.statement.other.iter().any(|x| x.draft == "moqt-23" && x.peer.as_deref() == Some(hex::encode(edge.relay().relay_id()).as_str())), "the core states the link, its draft and its peer: {:?}", stated.statement);
    let t = Tracks::new(ns(&[&room, &hexkey(&pk), "video"]), "0", 10, 11, 64);
    let media = joule_moq::draft22::names::render(&t.ns, Some(&t.media));
    let (p, _) = connect(&core, Via::WebSocket, ClientOptions::default()).await;
    let _published = publish_tracks(&p, &t, None).await;
    let (v, _) = connect(&edge, Via::WebSocket, ClientOptions::default()).await;
    let sub = subscribe_both(&v, &t, Parameters::new()).await;
    for g in 0..3 {
        publish_group(&p, &pk, &t, g, &payload("e", g, 300)).await;
    }
    let mut got = collect(&v, STEP, |x| x.len() >= 6).await;
    assert!(until(STEP, || forwarded(&edge) == 6 && forwarded(&core) == 6).await, "every object counted out at both relays");

    // The edge's own operator closes its window there, and the core's closes the core's.
    let (code, ce, err) = run_control(&se, json!({"op": "close"})).await;
    assert_eq!((code, &ce["ok"]), (0, &json!(true)), "{ce} {err}");
    let (code, cc, _) = run_control(&sc, json!({"op": "close"})).await;
    assert_eq!(code, 0, "{cc}");
    let cut_of = |a: &Value| serde_json::from_value::<Cut>(a["cut"].clone()).expect("a cut");
    let joined = Cut::join([&cut_of(&ce), &cut_of(&cc)]);
    assert_eq!(joined.of(&media), Some(3));

    let mut caught = 0;
    // Planted: a seal of a window that is not the next. Refused, exit 1, nothing sealed.
    let (code, a, _) = run_control(&se, json!({"op": "seal", "window": 7, "cut": joined})).await;
    assert_eq!((code, &a["ok"]), (1, &json!(false)), "a refusal exits 1 and is printed: {a}");
    assert!(a["error"].as_str().unwrap_or("").contains("not the closed window"), "{a}");
    assert!(edge.relay().window(0).is_none(), "nothing sealed");
    caught += 1;
    // Planted: an op the surface does not know.
    let (code, a, _) = run_control(&se, json!({"op": "shutdown"})).await;
    assert_eq!((code, &a["ok"]), (1, &json!(false)), "{a}");
    caught += 1;
    // Planted: two requests on one line's place. Nothing is sent.
    let before = edge.relay().control_requests("close");
    let (code, out, err) = control(&se, "{\"op\":\"close\"}\n{\"op\":\"close\"}");
    assert_eq!((code, out.as_str()), (2, ""), "{err}");
    assert_eq!(edge.relay().control_requests("close"), before, "a request the program would not send reached the surface");
    caught += 1;
    // Planted: a socket nobody serves.
    let (code, out, err) = control(&socket_path("nobody"), r#"{"op":"show"}"#);
    assert_eq!((code, out.as_str()), (3, ""), "{err}");
    caught += 1;
    eprintln!("planted faults caught (program): {caught} of 4");

    // Both seal at the join, each by its own operator.
    let (code, a, _) = run_control(&se, json!({"op": "seal", "window": ce["window"], "cut": joined})).await;
    assert_eq!(code, 0, "{a}");
    let (code, b, _) = run_control(&sc, json!({"op": "seal", "window": cc["window"], "cut": joined})).await;
    assert_eq!(code, 0, "{b}");
    assert_eq!(a["cut_sha256"], b["cut_sha256"], "sealed at one cut");
    let statement = |a: Value| serde_json::from_value::<SignedWindow>(a["statement"].clone()).expect("a signed window");
    let (code, w, _) = run_control(&se, json!({"op": "window", "window": ce["window"], "room": room})).await;
    assert_eq!(code, 0);
    let (_, wc, _) = run_control(&sc, json!({"op": "window", "window": cc["window"], "room": room})).await;
    let flows = link_chain(&statement(wc), &statement(w), true).expect("the core and the edge chain over their link in the window");
    assert_eq!((flows[&media].out, flows[&media].inn), (3, 3));

    // The edge rode through the window's end: its viewer was sent no GOAWAY, and receives on.
    for g in 3..5 {
        publish_group(&p, &pk, &t, g, &payload("e", g, 300)).await;
    }
    got.extend(collect(&v, STEP, |x| x.len() >= 4).await);
    assert_eq!(verify(&got, &sub, &pk).0, 5, "every group delivered across the window's end");
    assert!(v.goaway().borrow().is_none(), "the viewer was sent GOAWAY at a window's end");
    assert_eq!(edge.relay().stats().sessions_open, 2, "the viewer and the link, both open");
    for s in [&sc, &se] {
        assert_eq!(std::fs::metadata(s).unwrap().permissions().mode() & 0o777, 0o600, "the socket keeps its mode");
    }
    p.close(0, "done");
    v.close(0, "done");
    edge.shutdown().await;
    core.shutdown().await;
}