//! The DVR store (WAI live-recording §8): a window of a room's tracks with their receipts,
//! on disk, served to FETCH below the live edge.
//!
//! - **The window.** A long publish crosses the window's byte budget: the oldest groups go,
//!   whole, each media group with its receipts group, and the window holds what it says. A
//!   window by age empties as its groups pass it, each group counted once: evicted, or late
//!   when the relay took it whole only after the window had passed it (out of order, as a
//!   loaded host takes streams).
//! - **FETCH inside, across and before the window**, ascending and descending: inside, every
//!   object, each verifying against the fetched receipts; across, an End of Unknown Range
//!   for what was evicted and every object after it, from the store and the cache together;
//!   before, an End of Unknown Range alone.
//! - **A restart** keeps the store: every object is checked again, a corrupted one (a
//!   flipped byte; bytes changed with their checks made again, so only the receipt finds
//!   them) is dropped, counted and never served, and a write a crash cut short is removed.
//! - **Moderation.** A room key's redaction withholds its target from every DVR read and
//!   erases it from disk; a deny withholds until an allow restores it; a record of an events
//!   track is withheld from the object that carries it; a decision by someone who is not a
//!   moderator changes nothing; and the decisions hold across a restart.
//! - **The seek power** (WAI live-rooms §10.6): a viewer whose grant lacks it is refused a
//!   FETCH and a fill below the live edge (`seek_not_granted`) and served one at the live
//!   edge; a viewer whose grant holds it is served the window.
//! - **The Rewind filter** (`draft-duke-moq-subscribe-rewind-02`): the Groups before the
//!   current one come from the store on a stream a subgroup, the current Group from the cache
//!   from its start with its later objects on the same stream, each object verifying against
//!   the rewound receipts. Planted: a rewind past the store's window (fewer Groups, stated), a
//!   rewind on a track with no store (the current Group alone), a rewound object that fails
//!   its receipt (never sent, and its stream reset), a filter above MAX_REWIND (the session
//!   closed), and a rewind without the seek power (the current Group alone).
//! - **Metering**: the relay signs `WMH1` version 2, with the store's figures.
//! - **A sealed room** (WAI confidentiality §9) is never kept: its first sealed object is
//!   never written, what the store held of the room (its events and receipts, in the clear)
//!   is deleted with its decisions, nothing more of it is written, and that holds across a
//!   restart, while a room in the clear beside it keeps its window.
//! - **A room in which a minor publishes** (WAI age-assurance §8.4) is never kept either: once
//!   a session proves a minor's publishing grant of the room, what the store held of it is
//!   deleted, nothing more of it is written, and that holds across a restart, while another
//!   room keeps its window; a minor's grant that publishes nothing changes nothing.
//! - **A signed takedown** (WAI takedown §10.1), riding its service's takedown track, makes the
//!   store forget what it covers, in every room: an object by its content hash, a perceptual
//!   match's candidate, and a track from a group on. None of it is served to a FETCH, from the
//!   store or the cache, and its bytes leave the disk; an identical copy published later on
//!   another track is never stored; the store signs a purge with the relay's key, once; a
//!   takedown by a key the store does not trust, on its own track or on the service's,
//!   changes nothing; and all of it holds across a restart.

mod common;

use common::*;
use ed25519_dalek::{Signer, SigningKey};
use joule_moq::carriage::ReceiptBundle;
use joule_moq::draft22::data::{FetchEntry, ObjectStatus, RangeEnd, SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::{GroupOrder, Location, LocationFilter, Parameter, Parameters};
use joule_moq::receipt::{ObjectReceipt, ReceiptBuilder, ReceiptVerifier};
use joule_moq::{GroupId, ObjectId, TrackAlias, TrackNamespace};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{Authority, DvrConfig, DvrWindow};
use joule_moq_relay::{RelayConfig, RelayServer};
use std::collections::BTreeMap;
use std::path::PathBuf;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::time::timeout;

const OBJECTS: u64 = 3;
const LEN: usize = 1000;
/// The cache's budget: about three groups with their receipts, so that what is older comes
/// from the store.
const CACHE: usize = 16 * 1024;

fn now_s() -> u64 {
    SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()
}

fn dvr_config(path: &std::path::Path, max_age_s: u64, max_bytes: u64, room_keys: Vec<String>) -> DvrConfig {
    DvrConfig { path: path.to_path_buf(), window: vec![DvrWindow { prefix: vec!["*".into()], max_age_s, max_bytes }], room_keys, takedown_keys: vec![] }
}

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

fn object_payload(tag: &str, g: u64, o: u64) -> Vec<u8> {
    let mut v = format!("{tag}:{g}:{o}:").into_bytes();
    v.resize(LEN, (g * 7 + o) as u8);
    v
}

/// The receipts of a group of `payloads`, as group `g` of its receipts track carries them.
fn bundle_of(key: &SigningKey, receipt_alias: u64, g: u64, payloads: &[Vec<u8>]) -> ReceiptBundle {
    let mut b = ReceiptBuilder::new(TrackAlias(receipt_alias), GroupId(g));
    for p in payloads {
        b.add_labelled_object(ObjectReceipt::hash_payload(p), 0, "test-publisher", "2026-10-19T00:00:00Z", vec![0]).unwrap();
    }
    let objects = payloads
        .iter()
        .enumerate()
        .map(|(i, p)| ObjectReceipt {
            track_alias: TrackAlias(receipt_alias),
            group_id: GroupId(g),
            object_id: ObjectId(i as u64),
            content_hash: ObjectReceipt::hash_payload(p),
            joules_micro: 0,
            origin: "test-publisher".into(),
            merkle_proof: b.proof_for(i).unwrap(),
            energy_class: Some(vec![0]),
        })
        .collect();
    let group = b.finalize_labelled(key, "test-publisher", "2026-10-19T00:00:01Z", None).unwrap();
    ReceiptBundle { group, objects }
}

/// Publish group `g` of `t`: its receipts, then its objects on one subgroup stream.
async fn publish_multi(c: &Client, key: &SigningKey, t: &Tracks, g: u64, tag: &str) {
    publish_parts(c, key, t, g, tag, true, true).await;
}

/// Publish group `g` of `t` in part: its receipts (with `receipts`), then its objects on one
/// subgroup stream (with `media`).
async fn publish_parts(c: &Client, key: &SigningKey, t: &Tracks, g: u64, tag: &str, receipts: bool, media: bool) {
    let payloads: Vec<Vec<u8>> = (0..OBJECTS).map(|o| object_payload(tag, g, o)).collect();
    if receipts {
        let bundle = bundle_of(key, t.receipt_alias, g, &payloads);
        let rh = SubgroupHeader { track_alias: t.receipts_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
        c.send_subgroup(rh, &[SubgroupObject::normal(0, bundle.to_payload().unwrap())]).await.expect("the receipts");
    }
    if media {
        let mh = SubgroupHeader { track_alias: t.media_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
        let objects: Vec<SubgroupObject> = payloads.into_iter().enumerate().map(|(o, p)| SubgroupObject::normal(o as u64, p)).collect();
        c.send_subgroup(mh, &objects).await.expect("the objects");
    }
}

fn room_tracks(room: [u8; 32], publisher: &SigningKey) -> Tracks {
    Tracks::new(ns(&[&hex::encode(room), &hexkey(publisher), "video"]), "0", 10, 5, 128)
}

async fn flush(server: &RelayServer) {
    let d = server.relay().dvr().expect("a DVR store").clone();
    tokio::task::spawn_blocking(move || d.flush()).await.unwrap();
}

/// Wait, bounded, until the store has taken `n` groups: written, or found late (whole
/// after the window had moved past them, under load).
async fn written(server: &RelayServer, n: u64) {
    let d = server.relay().dvr().expect("a DVR store").clone();
    assert!(until(STEP, || d.counts().groups_written + d.counts().groups_late >= n).await, "{n} groups written: {:?}", d.counts());
    flush(server).await;
}

fn range(start: (u64, u64), end_group: u64) -> Parameters {
    Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStartGroupEnd { start: Location::new(start.0, start.1), end_group_delta: end_group - start.0 }))
}

/// What a fetch stream carried: its objects by location, and its End of Range entries.
#[derive(Debug, Default)]
struct Fetched {
    order: Vec<(u64, u64)>,
    objects: BTreeMap<(u64, u64), Vec<u8>>,
    unknown: Vec<Location>,
    other_ends: Vec<(RangeEnd, Location)>,
}

fn fetched(entries: Vec<FetchEntry>) -> Fetched {
    let mut f = Fetched::default();
    for e in entries {
        match e {
            FetchEntry::Object(o) => {
                f.order.push((o.group_id, o.object_id));
                f.objects.insert((o.group_id, o.object_id), o.payload);
            }
            FetchEntry::EndOfRange { kind: RangeEnd::Unknown, location } => f.unknown.push(location),
            FetchEntry::EndOfRange { kind, location } => f.other_ends.push((kind, location)),
        }
    }
    f
}

/// FETCH a range of the media track and of its receipts track; every media object fetched
/// verifies against the fetched receipts of its group under the publisher's key. Returns
/// the media fetched.
async fn fetch_verified(c: &Client, t: &Tracks, key: &SigningKey, params: Parameters) -> Fetched {
    let (_, media) = c.fetch(t.ns.clone(), t.media.clone(), params.clone()).await.expect("FETCH the media track");
    let (_, rec) = c.fetch(t.ns.clone(), t.receipts.clone(), params).await.expect("FETCH the receipts track");
    let media = fetched(media);
    let rec = fetched(rec);
    let v = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
    for ((g, o), p) in &media.objects {
        let b = ReceiptBundle::from_payload(rec.objects.get(&(*g, 0)).unwrap_or_else(|| panic!("the receipts of group {g} come with it"))).unwrap();
        let vg = v.verify_group(&b.group).expect("the group receipt verifies");
        let r = b.objects.iter().find(|r| r.object_id == ObjectId(*o)).expect("a receipt for the object");
        v.verify_payload(p, r, &vg).unwrap_or_else(|e| panic!("({g}, {o}) verifies against its receipt: {e}"));
    }
    media
}

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

async fn window_run() {
    const GROUPS: u64 = 40;
    let dir = store_dir("window");
    let room = [0x51; 32];
    let publisher = SigningKey::from_bytes(&[0x52; 32]);
    let t = room_tracks(room, &publisher);
    // About seven groups of media and receipts in the window; the cache holds fewer.
    let max_bytes = 30_000;
    let cfg = dvr_config(&dir, 900, max_bytes, vec![]);
    let server = relay(|c| {
        c.cache.max_bytes = CACHE;
        c.dvr = Some(cfg.clone());
    })
    .await;
    let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let _pub = publish_tracks(&p, &t, None).await;
    for g in 0..GROUPS {
        publish_multi(&p, &publisher, &t, g, "w").await;
    }
    written(&server, 2 * GROUPS).await;
    let d = server.relay().dvr().unwrap().clone();
    let media_key = (t.ns.clone(), t.media.clone());
    let rec_key = (t.ns.clone(), t.receipts.clone());
    let held = d.groups(&media_key);
    let first = *held.first().expect("groups held");
    assert!(first > 0, "the oldest groups were evicted");
    assert_eq!(held, (first..GROUPS).collect::<Vec<_>>(), "the window is the newest groups, no gap");
    assert_eq!(d.groups(&rec_key), held, "each media group goes with its receipts group");
    let counts = d.counts();
    assert!(counts.bytes_held <= max_bytes, "the window holds its budget: {counts:?}");
    assert_eq!(counts.groups_evicted + counts.groups_late, 2 * first, "whole groups, media and receipts");
    let all = GROUPS * (OBJECTS + 1);
    assert!(counts.objects_written <= all && counts.objects_written + counts.groups_late * OBJECTS >= all, "every object, but a group the window had passed: {counts:?}");

    let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    // Inside the window: every object, verified, in the requested order.
    let inside = fetch_verified(&v, &t, &publisher, range((first, 0), first + 2)).await;
    assert_eq!(inside.objects.len() as u64, 3 * OBJECTS);
    assert!(inside.unknown.is_empty() && inside.other_ends.is_empty());
    for ((g, o), p) in &inside.objects {
        assert_eq!(p, &object_payload("w", *g, *o));
    }
    let desc = fetch_verified(&v, &t, &publisher, range((first, 0), first + 2).with(Parameter::GroupOrder(GroupOrder::Descending))).await;
    let groups: Vec<u64> = desc.order.iter().map(|x| x.0).collect();
    assert_eq!(groups, [first + 2, first + 2, first + 2, first + 1, first + 1, first + 1, first, first, first], "descending groups, objects ascending");
    assert!(desc.order.windows(2).all(|w| w[0].0 != w[1].0 || w[0].1 < w[1].1));

    // Across: what was evicted is unknown, and the rest comes, from the store and the cache.
    let across = fetch_verified(&v, &t, &publisher, Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(0, 0) }))).await;
    assert_eq!(across.unknown, vec![Location::new(first - 1, u64::MAX)], "End of Unknown Range through the last evicted group");
    assert!(across.other_ends.is_empty());
    assert_eq!(across.objects.len() as u64, (GROUPS - first) * OBJECTS, "every object the window holds and the cache's");
    assert_eq!(across.order.first(), Some(&(first, 0)));
    assert_eq!(across.order.last(), Some(&(GROUPS - 1, OBJECTS - 1)));
    assert!(server.relay().stats().fetch_served_from_dvr >= 2);

    // Before: nothing but the End of Unknown Range.
    assert!(first > 3);
    let (ok, before) = v.fetch(t.ns.clone(), t.media.clone(), range((0, 0), 2)).await.expect("a FETCH before the window");
    assert_eq!(ok.end_location, Location::new(2, u64::MAX));
    let before = fetched(before);
    assert!(before.objects.is_empty());
    assert_eq!(before.unknown, vec![Location::new(2, u64::MAX)]);

    // The record of the room is version 2, with the store's figures.
    v.close(0, "done");
    p.close(0, "done");
    assert!(until(2 * STEP, || server.relay().summaries().len() >= 2).await, "both sessions ended: {:?}; open: {:?}", server.relay().stats(), server.relay().session_diagnostics());
    let rec = server.relay().hop_record(room, 0);
    let bytes = rec.to_bytes();
    assert_eq!(&bytes[..5], b"WMH1\x02", "version 2");
    rec.verify().expect("the record verifies under the relay's key");
    let back = joule_moq_relay::metering::HopRecord::decode(&bytes, &rec.ledger_bytes()).expect("it reads back");
    assert_eq!(back, rec);
    let dv = rec.dvr.expect("the DVR figures");
    let served: u64 = server.relay().summaries().iter().flat_map(|s| s.summary.tally.subscriber.tracks.values().map(|x| x.dvr_objects)).sum();
    assert_eq!(dv.objects_out, served);
    assert!(dv.objects_out >= 2 * 3 * OBJECTS, "the inside fetches came from the store: {dv:?}");
    let out: u64 = rec.entries.iter().filter(|e| e.dir == 2).map(|e| e.objects).sum();
    let inn: u64 = rec.entries.iter().filter(|e| e.dir == 1).map(|e| e.objects).sum();
    assert!(dv.objects_out <= out && dv.objects_stored <= inn, "{dv:?}, out {out}, in {inn}");
    assert_eq!(dv.objects_stored, counts.objects_written, "every object of the room written in the span");
    assert!(dv.byte_s > 0 && dv.dropped == 0);
    server.shutdown().await;
    let _ = std::fs::remove_dir_all(&dir);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_window_by_age_empties_as_its_groups_pass_it() {
    timeout(Duration::from_secs(60), age_run(false)).await.expect("in time");
}

/// The same window when the relay takes the groups whole out of order, as a loaded host does
/// (each stream is read by a task of its own): group 0 and group 2's receipts first, then,
/// once the window has passed group 2, group 1 and group 2's objects.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_window_by_age_passes_groups_taken_out_of_order() {
    timeout(Duration::from_secs(60), age_run(true)).await.expect("in time");
}

async fn age_run(out_of_order: bool) {
    let dir = store_dir(if out_of_order { "age-order" } else { "age" });
    let room = [0x53; 32];
    let publisher = SigningKey::from_bytes(&[0x54; 32]);
    let t = room_tracks(room, &publisher);
    let cfg = dvr_config(&dir, 2, 1 << 30, vec![]);
    let server = relay(|c| {
        // About two groups: what aged out of the store is out of the cache too.
        c.cache.max_bytes = 8 * 1024;
        c.dvr = Some(cfg.clone());
    })
    .await;
    let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let _pub = publish_tracks(&p, &t, None).await;
    let d = server.relay().dvr().unwrap().clone();
    let (key, rkey) = ((t.ns.clone(), t.media.clone()), (t.ns.clone(), t.receipts.clone()));
    let empty = || d.groups(&key).is_empty() && d.groups(&rkey).is_empty();
    if out_of_order {
        // Group 0, then group 2's receipts, each taken before the next is sent: all three
        // are stored.
        publish_multi(&p, &publisher, &t, 0, "a").await;
        written(&server, 2).await;
        publish_parts(&p, &publisher, &t, 2, "a", true, false).await;
        written(&server, 3).await;
        assert!(until(Duration::from_secs(10), empty).await, "past max_age_s groups 0 and 2 go");
        publish_multi(&p, &publisher, &t, 1, "a").await;
        publish_parts(&p, &publisher, &t, 2, "a", false, true).await;
    } else {
        for g in 0..3 {
            publish_multi(&p, &publisher, &t, g, "a").await;
        }
    }
    written(&server, 6).await;
    assert!(until(Duration::from_secs(10), empty).await, "past max_age_s every group goes");
    // Each group was stored and then evicted, or was late: whole only once the window had
    // passed its number, and never stored. Each group stored is counted as it goes, the
    // receipts group evicted with its media group too.
    let c = d.counts();
    assert_eq!(c.groups_evicted + c.groups_late, 6, "every group evicted or late: {c:?}");
    assert_eq!(c.groups_evicted, c.groups_written, "each group stored is counted as it goes: {c:?}");
    if out_of_order {
        // Group 0 and group 2's receipts were stored; group 1 and group 2's objects came after
        // the window had passed group 2.
        assert_eq!((c.groups_written, c.groups_late), (3, 3), "{c:?}");
    }
    // The next group, fetched at once: what aged out is unknown, and the group is served.
    publish_multi(&p, &publisher, &t, 3, "a").await;
    // The relay has taken group 3 into both tracks before the fetches: taking it may
    // evict a pair from the cache, and the media and receipts FETCHes are two requests,
    // so a pair evicted between them would leave media without receipts.
    let relay = server.relay().clone();
    assert!(
        until(STEP, || relay.largest(&key) == Some(Location::new(3, OBJECTS - 1)) && relay.largest(&rkey) == Some(Location::new(3, 0))).await,
        "the relay took group 3"
    );
    let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let f = fetch_verified(&v, &t, &publisher, range((0, 0), 3)).await;
    // What aged out is unknown through the last group neither the store nor the cache
    // holds; every group after it comes whole.
    assert_eq!(f.unknown.len(), 1);
    let k = f.unknown[0];
    assert!(k.object == u64::MAX && k.group < 3, "{k:?}");
    let want: Vec<(u64, u64)> = (k.group + 1..=3).flat_map(|g| (0..OBJECTS).map(move |o| (g, o))).collect();
    assert_eq!(f.objects.keys().copied().collect::<Vec<_>>(), want);
    server.shutdown().await;
    let _ = std::fs::remove_dir_all(&dir);
}

/// Rewrite one object of the group record whose frame is at `frame_at` in a segment, in
/// place: `f` changes its payload (same length); with `fix`, its digest and check are made
/// again so that only its receipt finds it.
fn tamper(path: &std::path::Path, frame_at: u64, object: u64, fix: bool, f: impl Fn(&mut [u8])) {
    let mut b = std::fs::read(path).unwrap();
    let r = frame_at as usize + 5;
    let count = u32::from_le_bytes(b[r + 53..r + 57].try_into().unwrap()) as u64;
    let mut p = r + 93;
    for i in 0..count {
        let props_len = u32::from_le_bytes(b[p + 18..p + 22].try_into().unwrap()) as usize;
        let payload_len = u32::from_le_bytes(b[p + 22..p + 26].try_into().unwrap()) as usize;
        let at = p + 90 + props_len;
        if i == object {
            f(&mut b[at..at + payload_len]);
            if fix {
                let digest = ObjectReceipt::hash_payload(&b[at..at + payload_len]);
                b[p + 26..p + 58].copy_from_slice(&digest);
                let mut c = ring::digest::Context::new(&ring::digest::SHA256);
                c.update(&b[p..p + 58]);
                c.update(&b[p + 90..at + payload_len]);
                let check = c.finish();
                b[p + 58..p + 90].copy_from_slice(check.as_ref());
            }
            std::fs::write(path, &b).unwrap();
            return;
        }
        p = at + payload_len;
    }
    panic!("no object {object}");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_restart_keeps_the_store_and_drops_what_does_not_verify() {
    timeout(Duration::from_secs(120), async {
        const GROUPS: u64 = 10;
        let dir = store_dir("restart");
        let room = [0x55; 32];
        let publisher = SigningKey::from_bytes(&[0x56; 32]);
        let t = room_tracks(room, &publisher);
        let cfg = dvr_config(&dir, 900, 1 << 30, vec![]);
        let config = |c: &mut RelayConfig| {
            c.cache.max_bytes = CACHE;
            c.dvr = Some(cfg.clone());
        };
        let server = relay(config).await;
        let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let _pub = publish_tracks(&p, &t, None).await;
        for g in 0..GROUPS {
            publish_multi(&p, &publisher, &t, g, "r").await;
        }
        written(&server, 2 * GROUPS).await;
        let key = (t.ns.clone(), t.media.clone());
        let d = server.relay().dvr().unwrap().clone();
        let ((g3, at3), (g5, at5), (g9, _)) = (d.group_location(&key, 3).unwrap(), d.group_location(&key, 5).unwrap(), d.group_location(&key, 9).unwrap());
        p.close(0, "done");
        server.shutdown().await;

        // A flipped byte in (3, 1); (5, 0) rewritten with its checks made again; and a frame a
        // crash cut short at the end of a segment.
        tamper(&g3, at3, 1, false, |p| p[10] ^= 0x20);
        tamper(&g5, at5, 0, true, |p| p.reverse());
        let whole = std::fs::metadata(&g9).unwrap().len();
        {
            use std::io::Write;
            let mut f = std::fs::OpenOptions::new().append(true).open(&g9).unwrap();
            f.write_all(&1000u32.to_le_bytes()).unwrap();
            f.write_all(&[1, b'W', b'D', b'G', b'1']).unwrap();
        }

        let server = relay(config).await;
        let d = server.relay().dvr().unwrap().clone();
        let c = d.counts();
        assert_eq!((c.objects_dropped, c.torn_writes), (2, 1), "{c:?}");
        assert_eq!(std::fs::metadata(&g9).unwrap().len(), whole, "the cut-short frame is cut off");
        assert_eq!(d.groups(&key), (0..GROUPS).collect::<Vec<_>>());
        // No publisher now: the store answers alone.
        let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let f = fetch_verified(&v, &t, &publisher, Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(0, 0) }))).await;
        // An End of Range moves to a later group than the entry before it (MOQT §11.4.1.2),
        // so (3, 1)'s marks group 3 from its start: (3, 0) goes with it.
        assert_eq!(f.unknown, vec![Location::new(3, 1), Location::new(5, 0)], "each in its place as unknown");
        assert!(!f.objects.contains_key(&(3, 1)) && !f.objects.contains_key(&(5, 0)), "never served");
        assert_eq!(f.objects.len() as u64, GROUPS * OBJECTS - 3, "every object but the two dropped and the one before (3, 1)");
        assert!(f.objects.contains_key(&(3, 2)) && f.objects.contains_key(&(5, 1)), "the objects after them follow");
        assert_eq!(server.relay().largest(&key), Some(Location::new(GROUPS - 1, OBJECTS - 1)));
        // The dropped objects are out of their files too: a second restart finds nothing to drop.
        server.shutdown().await;
        let server = relay(config).await;
        assert_eq!(server.relay().dvr().unwrap().counts().objects_dropped, 0);
        let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let f = fetch_verified(&v, &t, &publisher, range((3, 0), 5)).await;
        assert_eq!(f.unknown, vec![Location::new(3, 1), Location::new(5, 0)]);
        server.shutdown().await;
        let _ = std::fs::remove_dir_all(&dir);
    })
    .await
    .expect("in time");
}

/// A record of live-events §3.1, without an anchor, signed by `key`.
fn event(key: &SigningKey, room: [u8; 32], seq: u64, t: u64, kind: u8, 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(kind);
    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 event_digest(rec: &[u8]) -> [u8; 32] {
    ring::digest::digest(&ring::digest::SHA256, &rec[..rec.len() - 64]).as_ref().try_into().unwrap()
}

fn decision(key: &SigningKey, room: [u8; 32], seq: u64, verdict: u8, target: [u8; 32]) -> Vec<u8> {
    let mut p = vec![verdict];
    p.extend_from_slice(&target);
    p.extend_from_slice(&1u16.to_le_bytes());
    event(key, room, seq, 1_000 + seq, 0x0B, &p)
}

fn msf(records: &[&[u8]]) -> Vec<u8> {
    use base64::Engine;
    let e: Vec<String> = records.iter().map(|r| format!("{{\"t\":1,\"data\":{{\"wev1\":\"{}\"}}}}", base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(r))).collect();
    format!("[{}]", e.join(",")).into_bytes()
}

/// A principal's events track: PUBLISH it, and send objects of group 0 one stream each.
struct EventsTrack {
    ns: TrackNamespace,
    alias: u64,
}

impl EventsTrack {
    async fn open(c: &Client, room: [u8; 32], who: &SigningKey, alias: u64) -> (EventsTrack, (joule_moq_relay::wire::ReqTx, joule_moq_relay::wire::ReqRx)) {
        let ns = ns(&[&hex::encode(room), &hexkey(who), "events"]);
        let held = c.publish(ns.clone(), name("0"), alias, Parameters::new(), track_props(16, None)).await.expect("PUBLISH an events track");
        (EventsTrack { ns, alias }, held)
    }

    /// Group `g`'s objects: the join set, then one object a record.
    async fn group(&self, c: &Client, g: u64, records: &[Vec<u8>]) {
        let h = SubgroupHeader { track_alias: self.alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(16), properties: false, end_of_group: true, first_object: true };
        let mut objects = vec![SubgroupObject::normal(0, b"[]".to_vec())];
        for (i, r) in records.iter().enumerate() {
            objects.push(SubgroupObject::normal(i as u64 + 1, msf(&[r])));
        }
        c.send_subgroup(h, &objects).await.expect("the events");
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn decisions_withhold_from_the_store_and_hold_across_a_restart() {
    timeout(Duration::from_secs(120), async {
        const GROUPS: u64 = 6;
        let dir = store_dir("moderation");
        let room = [0x57; 32];
        let publisher = SigningKey::from_bytes(&[0x58; 32]);
        let moderator = SigningKey::from_bytes(&[0x59; 32]);
        let t = room_tracks(room, &publisher);
        let cfg = dvr_config(&dir, 900, 1 << 30, vec![hexkey(&moderator)]);
        let config = |c: &mut RelayConfig| {
            c.cache.max_bytes = CACHE;
            c.dvr = Some(cfg.clone());
        };
        let server = relay(config).await;
        let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let _pub = publish_tracks(&p, &t, None).await;
        for g in 0..GROUPS {
            publish_multi(&p, &publisher, &t, g, "m").await;
        }
        // The publisher's events: two chat messages.
        let (pe, _pe_held) = EventsTrack::open(&p, room, &publisher, 40).await;
        let chat1 = event(&publisher, room, 0, 10, 0x03, b"\x02\x00hi");
        let chat2 = event(&publisher, room, 1, 11, 0x03, b"\x03\x00bye");
        pe.group(&p, 0, &[chat1.clone(), chat2.clone()]).await;
        written(&server, 2 * GROUPS + 1).await;
        let d = server.relay().dvr().unwrap().clone();

        let h = |g: u64, o: u64| ObjectReceipt::hash_payload(&object_payload("m", g, o));
        // The moderator (a room key): redact (2, 1), deny (3, 0), redact the first chat. The
        // publisher, no moderator, redacts (4, 0): nothing comes of it.
        let (m, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let (me, _me_held) = EventsTrack::open(&m, room, &moderator, 50).await;
        me.group(&m, 0, &[decision(&moderator, room, 0, 3, h(2, 1)), decision(&moderator, room, 1, 2, h(3, 0)), decision(&moderator, room, 2, 3, event_digest(&chat1))]).await;
        pe.group(&p, 1, &[decision(&publisher, room, 2, 3, h(4, 0))]).await;
        assert!(until(STEP, || d.counts().decisions >= 3 && d.counts().objects_redacted >= 2).await, "{:?}", d.counts());
        assert_eq!(d.counts().decisions, 3, "the publisher's decision is not honoured");

        let all = || Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(0, 0) }));
        let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let f = fetch_verified(&v, &t, &publisher, all()).await;
        assert_eq!(f.unknown, vec![Location::new(2, 1), Location::new(3, 0)], "withheld, each in its place");
        assert_eq!(f.objects.len() as u64, GROUPS * OBJECTS - 3, "and (2, 0), which an End of Range through (2, 1) covers");
        assert!(f.objects.contains_key(&(2, 2)) && f.objects.contains_key(&(3, 1)));
        assert!(f.objects.contains_key(&(4, 0)), "a decision by no moderator withholds nothing");
        // The redacted object's bytes are gone from every file of its track; the denied one's stay.
        let on_disk = |key: &(TrackNamespace, joule_moq::TrackName), bytes: &[u8]| -> bool {
            let dir = d.track_dir(key).unwrap();
            std::fs::read_dir(dir).unwrap().flatten().any(|e| std::fs::read(e.path()).unwrap().windows(bytes.len()).any(|w| w == bytes))
        };
        let media = (t.ns.clone(), t.media.clone());
        assert!(!on_disk(&media, &object_payload("m", 2, 1)), "redacted: erased from disk");
        assert!(on_disk(&media, &object_payload("m", 3, 0)), "denied: kept");
        // The redacted chat is erased from the events track's files; the other stays.
        let ev_key = (pe.ns.clone(), name("0"));
        assert!(!on_disk(&ev_key, &msf(&[&chat1])), "redacted: erased from disk");
        assert!(on_disk(&ev_key, &msf(&[&chat2])));
        // The publisher's events: the redacted chat is withheld, the other is served as sent.
        let (_, ev) = v.fetch(pe.ns.clone(), name("0"), all()).await.expect("FETCH the events track");
        let ev = fetched(ev);
        assert_eq!(ev.unknown, vec![Location::new(0, 1)], "the object holding the redacted record");
        assert_eq!(ev.objects.get(&(0, 2)), Some(&msf(&[&chat2])));

        // An allow supersedes the deny: (3, 0) is served again.
        me.group(&m, 1, &[decision(&moderator, room, 3, 1, h(3, 0))]).await;
        assert!(until(STEP, || d.counts().decisions >= 4).await);
        let f = fetch_verified(&v, &t, &publisher, all()).await;
        assert_eq!(f.unknown, vec![Location::new(2, 1)]);
        assert!(f.objects.contains_key(&(3, 0)));
        assert_eq!(f.objects.len() as u64, GROUPS * OBJECTS - 2);

        // Across a restart the decisions hold.
        for c in [&p, &m, &v] {
            c.close(0, "done");
        }
        server.shutdown().await;
        let server = relay(config).await;
        let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let f = fetch_verified(&v, &t, &publisher, all()).await;
        assert_eq!(f.unknown, vec![Location::new(2, 1)]);
        let (_, ev) = v.fetch(pe.ns.clone(), name("0"), all()).await.expect("FETCH the events track");
        assert_eq!(fetched(ev).unknown, vec![Location::new(0, 1)]);
        server.shutdown().await;
        let _ = std::fs::remove_dir_all(&dir);
    })
    .await
    .expect("in time");
}

/// Publish group `g` of `t` sealed: its receipts, then objects each carrying a secure-objects
/// Key ID in its Immutable Properties (WAI confidentiality §4.3).
async fn publish_sealed(c: &Client, key: &SigningKey, t: &Tracks, g: u64) {
    use joule_moq::secure_objects::{immutable_properties, object_properties};
    let payloads: Vec<Vec<u8>> = (0..OBJECTS).map(|o| object_payload("sealed", g, o)).collect();
    let bundle = bundle_of(key, t.receipt_alias, g, &payloads);
    let rh = SubgroupHeader { track_alias: t.receipts_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
    c.send_subgroup(rh, &[SubgroupObject::normal(0, bundle.to_payload().unwrap())]).await.expect("the receipts");
    let props = object_properties(&immutable_properties(7, &[]), &[]);
    let mh = 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 };
    let objects: Vec<SubgroupObject> = payloads.into_iter().enumerate().map(|(o, p)| SubgroupObject { object_id: o as u64, properties: props.clone(), status: ObjectStatus::Normal, payload: p }).collect();
    c.send_subgroup(mh, &objects).await.expect("the sealed objects");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_sealed_room_is_never_kept() {
    timeout(Duration::from_secs(120), async {
        let dir = store_dir("sealed");
        let (sealed, clear) = ([0x5a; 32], [0x5b; 32]);
        let publisher = SigningKey::from_bytes(&[0x5c; 32]);
        let moderator = SigningKey::from_bytes(&[0x5d; 32]);
        let ts = room_tracks(sealed, &publisher);
        let tc = Tracks::new(ns(&[&hex::encode(clear), &hexkey(&publisher), "video"]), "0", 20, 6, 128);
        let cfg = dvr_config(&dir, 900, 1 << 30, vec![hexkey(&moderator)]);
        let config = |c: &mut RelayConfig| {
            c.cache.max_bytes = CACHE;
            c.dvr = Some(cfg.clone());
        };
        let server = relay(config).await;
        let d = server.relay().dvr().unwrap().clone();
        let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let _ps = publish_tracks(&p, &ts, None).await;
        let _pc = publish_tracks(&p, &tc, None).await;
        // Before the sealed room's first media group: its events, in the clear, and a
        // moderator's decision in each room.
        let (pe, _pe_held) = EventsTrack::open(&p, sealed, &publisher, 40).await;
        let chat = event(&publisher, sealed, 0, 10, 0x03, b"\x02\x00hi");
        pe.group(&p, 0, &[chat]).await;
        let (m, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let (ms, _ms_held) = EventsTrack::open(&m, sealed, &moderator, 50).await;
        let (mc, _mc_held) = EventsTrack::open(&m, clear, &moderator, 60).await;
        let (ds, dc) = (decision(&moderator, sealed, 0, 2, [1; 32]), decision(&moderator, clear, 0, 2, [2; 32]));
        ms.group(&m, 0, std::slice::from_ref(&ds)).await;
        mc.group(&m, 0, std::slice::from_ref(&dc)).await;
        for g in 0..2 {
            publish_multi(&p, &publisher, &tc, g, "c").await;
        }
        written(&server, 7).await;
        assert!(until(STEP, || d.counts().decisions >= 2).await, "{:?}", d.counts());
        let ev_key = (pe.ns.clone(), name("0"));
        let ev_dir = d.track_dir(&ev_key).expect("the sealed room's events, kept so far");
        assert!(ev_dir.exists() && d.groups(&ev_key) == vec![0]);
        let log = dir.join("decisions.wdl");
        let holds = |path: &std::path::Path, bytes: &[u8]| std::fs::read(path).unwrap_or_default().windows(bytes.len()).any(|w| w == bytes);
        assert!(holds(&log, &ds) && holds(&log, &dc));

        // The first sealed group: never written, and the room is forgotten.
        publish_sealed(&p, &publisher, &ts, 0).await;
        assert!(until(STEP, || d.counts().rooms_sealed == 1).await, "{:?}", d.counts());
        flush(&server).await;
        let room_keys = [(ts.ns.clone(), ts.media.clone()), (ts.ns.clone(), ts.receipts.clone()), ev_key.clone(), (ms.ns.clone(), name("0"))];
        for k in &room_keys {
            assert!(d.groups(k).is_empty() && d.track_dir(k).is_none(), "{k:?} is kept");
        }
        assert!(!ev_dir.exists(), "the sealed room's events are deleted from disk");
        assert!(!holds(&log, &ds), "its decision leaves the log");
        assert!(holds(&log, &dc), "the clear room's stays");
        assert_eq!(std::fs::metadata(dir.join("sealed.wdl")).unwrap().len(), 64);
        let clear_media = (tc.ns.clone(), tc.media.clone());
        assert_eq!(d.groups(&clear_media), vec![0, 1], "the clear room keeps its window");

        // Nothing more of it is written: its events and its next sealed group, then a group
        // of the clear room, which is.
        pe.group(&p, 1, &[event(&publisher, sealed, 1, 11, 0x03, b"\x03\x00bye")]).await;
        publish_sealed(&p, &publisher, &ts, 1).await;
        assert!(until(STEP, || server.relay().largest(&(ts.ns.clone(), ts.media.clone())) == Some(Location::new(1, OBJECTS - 1))).await);
        publish_multi(&p, &publisher, &tc, 2, "c").await;
        assert!(until(STEP, || d.groups(&clear_media).contains(&2)).await);
        flush(&server).await;
        for k in &room_keys {
            assert!(d.groups(k).is_empty(), "{k:?} is written after the room was sealed");
        }

        // Across a restart: the room is still sealed, nothing of it is loaded or written, and
        // only the clear room's decision is honoured again.
        for c in [&p, &m] {
            c.close(0, "done");
        }
        server.shutdown().await;
        let server = relay(config).await;
        let d = server.relay().dvr().unwrap().clone();
        assert_eq!((d.counts().rooms_sealed, d.counts().decisions), (1, 1), "{:?}", d.counts());
        assert_eq!(d.groups(&clear_media), vec![0, 1, 2]);
        let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let (pe, _pe_held) = EventsTrack::open(&p, sealed, &publisher, 40).await;
        pe.group(&p, 2, &[event(&publisher, sealed, 2, 12, 0x03, b"\x02\x00ok")]).await;
        let _pc = publish_tracks(&p, &tc, None).await;
        publish_multi(&p, &publisher, &tc, 3, "c").await;
        assert!(until(STEP, || d.groups(&clear_media).contains(&3)).await);
        flush(&server).await;
        assert!(d.groups(&ev_key).is_empty() && d.track_dir(&ev_key).is_none());
        p.close(0, "done");
        server.shutdown().await;
        let _ = std::fs::remove_dir_all(&dir);
    })
    .await
    .expect("in time");
}

/// Publish group `g` of `t` with these payloads: its receipts, then its objects.
async fn publish_payloads(c: &Client, key: &SigningKey, t: &Tracks, g: u64, payloads: Vec<Vec<u8>>) {
    let bundle = bundle_of(key, t.receipt_alias, g, &payloads);
    let rh = SubgroupHeader { track_alias: t.receipts_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
    c.send_subgroup(rh, &[SubgroupObject::normal(0, bundle.to_payload().unwrap())]).await.expect("the receipts");
    let mh = SubgroupHeader { track_alias: t.media_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
    let objects: Vec<SubgroupObject> = payloads.into_iter().enumerate().map(|(o, p)| SubgroupObject::normal(o as u64, p)).collect();
    c.send_subgroup(mh, &objects).await.expect("the objects");
}

/// A takedown as WAI takedown §5.3 lays it out, signed by `by`, naming `exact` and `matched`.
fn takedown_record(by: &SigningKey, exact: Vec<serde_json::Value>, matched: Vec<serde_json::Value>) -> (Vec<u8>, [u8; 32]) {
    let mut v = serde_json::json!({
        "kind": "wai-takedown", "policy": "00".repeat(32), "notice": null, "clock": null, "clock_start": null, "reply_at": null,
        "targets": [exact[0].clone()], "verdict": "remove", "exact": exact, "matched": matched, "ground": "illegal", "category": 3,
        "automated": {"detection": "no", "decision": "not"}, "explanation": "The item is unlawful.", "facts": "A notice named it.",
        "decided_at": "2026-10-21T08:00:00Z", "signer_pubkey": hex::encode(by.verifying_key().to_bytes()),
    });
    let sig = by.sign(&joule_moq_relay::takedown::signing_bytes(&v).unwrap());
    v["sig"] = serde_json::json!(hex::encode(sig.to_bytes()));
    let link = joule_moq_relay::takedown::link_hash(&v).unwrap();
    (serde_json::to_vec(&v).unwrap(), link)
}

/// Publish `records` on the takedown track of `owner`, `[owner, takedown]`, `0`: one group,
/// one object a record.
async fn publish_takedowns(c: &Client, owner: &SigningKey, alias: u64, records: &[Vec<u8>]) -> (joule_moq_relay::wire::ReqTx, joule_moq_relay::wire::ReqRx) {
    let held = c.publish(ns(&[&hexkey(owner), "takedown"]), name("0"), alias, Parameters::new(), track_props(16, None)).await.expect("PUBLISH a takedown track");
    let h = SubgroupHeader { track_alias: alias, group_id: 0, subgroup_id: SubgroupId::Zero, publisher_priority: Some(16), properties: false, end_of_group: true, first_object: true };
    let objects: Vec<SubgroupObject> = records.iter().enumerate().map(|(i, r)| SubgroupObject::normal(i as u64, r.clone())).collect();
    c.send_subgroup(h, &objects).await.expect("the takedowns");
    held
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_signed_takedown_forgets_what_it_covers_and_holds_across_a_restart() {
    timeout(Duration::from_secs(150), async {
        const GROUPS: u64 = 5;
        let dir = store_dir("takedown");
        let key_dir = store_dir("takedown-key");
        let key_file = key_dir.join("relay.key");
        std::fs::write(&key_file, [0x67u8; 32]).unwrap();
        let (room, room2) = ([0x63; 32], [0x68; 32]);
        let publisher = SigningKey::from_bytes(&[0x64; 32]);
        let service = SigningKey::from_bytes(&[0x65; 32]);
        let stranger = SigningKey::from_bytes(&[0x66; 32]);
        let t = room_tracks(room, &publisher);
        let t2 = Tracks::new(ns(&[&hex::encode(room2), &hexkey(&publisher), "video"]), "0", 30, 7, 128);
        let mut cfg = dvr_config(&dir, 900, 1 << 30, vec![]);
        cfg.takedown_keys = vec![hexkey(&service)];
        let config = |c: &mut RelayConfig| {
            c.cache.max_bytes = CACHE;
            c.dvr = Some(cfg.clone());
            c.summary.key_path = Some(key_file.clone());
        };
        let server = relay(config).await;
        let d = server.relay().dvr().unwrap().clone();
        let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let _pub = publish_tracks(&p, &t, None).await;
        for g in 0..GROUPS {
            publish_multi(&p, &publisher, &t, g, "td").await;
        }
        written(&server, 2 * GROUPS).await;

        let h = |g: u64, o: u64| ObjectReceipt::hash_payload(&object_payload("td", g, o));
        let object = |x: [u8; 32]| serde_json::json!({"type": "object", "content_hash": hex::encode(x)});
        let media = (t.ns.clone(), t.media.clone());
        let track = joule_moq_relay::dvr::room_track_digest(&media);
        let exact = vec![object(h(1, 1)), serde_json::json!({"type": "track", "track": hex::encode(track), "from_group": 3})];
        let matched = vec![serde_json::json!({"match": "aa".repeat(32), "candidate": hex::encode(h(0, 2))})];
        // A stranger's takedown, on its own track and on the service's: neither is honoured.
        let (strange, _) = takedown_record(&stranger, exact.clone(), matched.clone());
        let (s, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let _st = publish_takedowns(&s, &stranger, 70, std::slice::from_ref(&strange)).await;
        // The service's track carries the stranger's record, then the service's own.
        let (rec, link) = takedown_record(&service, exact.clone(), matched.clone());
        let (m, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let _sv = publish_takedowns(&m, &service, 71, &[strange.clone(), rec.clone()]).await;
        let purge_path = d.purge_path(&link);
        assert!(until(STEP, || purge_path.exists()).await, "the purge is written: {:?}", d.counts());
        assert_eq!(d.counts().takedowns, 1, "only the service's takedown is honoured");
        assert_eq!(d.takedown(&strange).map_err(|_| ()), Err(()), "a stranger's record is refused");

        // FETCH: what it covers is withheld, from the store and the cache alike; the rest
        // verifies against its receipts.
        let all = || Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(0, 0) }));
        let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        // A group's End of Unknown Range runs through its last withheld object, and covers the
        // objects before it (as a moderation's does): groups 3 and 4 are withheld whole.
        let unknown = vec![Location::new(0, 2), Location::new(1, 1), Location::new(3, OBJECTS - 1), Location::new(4, OBJECTS - 1)];
        let want: std::collections::BTreeSet<(u64, u64)> = [(1, 2), (2, 0), (2, 1), (2, 2)].into_iter().collect();
        let f = fetch_verified(&v, &t, &publisher, all()).await;
        assert_eq!(f.unknown, unknown);
        assert_eq!(f.objects.keys().copied().collect::<std::collections::BTreeSet<_>>(), want);
        // Its bytes are gone from every file of the track.
        let on_disk = |bytes: &[u8]| -> bool {
            let dir = d.track_dir(&media).unwrap();
            std::fs::read_dir(dir).unwrap().flatten().any(|e| std::fs::read(e.path()).unwrap().windows(bytes.len()).any(|w| w == bytes))
        };
        for (g, o) in [(1, 1), (0, 2), (3, 0), (4, 2)] {
            assert!(!on_disk(&object_payload("td", g, o)), "({g}, {o}) erased from disk");
        }
        assert!(on_disk(&object_payload("td", 2, 0)), "what it does not cover stays");
        // The purge: signed with the relay's key, naming each object and the track.
        let purge = std::fs::read_to_string(&purge_path).unwrap();
        let pv: serde_json::Value = serde_json::from_str(&purge).unwrap();
        let sig: [u8; 64] = hex::decode(pv["sig"].as_str().unwrap()).unwrap().try_into().unwrap();
        let relay_key = server.relay().summary_key().to_bytes();
        assert_eq!(pv["signer_pubkey"], hex::encode(relay_key));
        assert!(joule_moq_relay::takedown::verify_strict(&relay_key, &joule_moq_relay::takedown::signing_bytes(&pv).unwrap(), &sig));
        assert_eq!(pv["takedown"], hex::encode(link));
        assert_eq!(pv["kind"], "wai-takedown-purge");
        let named: Vec<String> = pv["subjects"].as_array().unwrap().iter().map(|x| joule_moq_relay::takedown::jcs(x).unwrap()).collect();
        let mut sorted = named.clone();
        sorted.sort();
        assert_eq!(named, sorted, "ascending");
        assert_eq!(named.len(), 3, "the object, the candidate and the track");
        assert!(named.contains(&joule_moq_relay::takedown::jcs(&object(h(1, 1))).unwrap()) && named.contains(&joule_moq_relay::takedown::jcs(&object(h(0, 2))).unwrap()));

        // Across a restart: the takedown holds, the purge is the same bytes, and an identical
        // copy published on another track is never stored or served.
        for c in [&p, &s, &m, &v] {
            c.close(0, "done");
        }
        server.shutdown().await;
        let server = relay(config).await;
        let d = server.relay().dvr().unwrap().clone();
        assert_eq!((d.counts().takedowns, d.counts().purges_signed), (1, 0), "{:?}", d.counts());
        assert_eq!(std::fs::read_to_string(&purge_path).unwrap(), purge);
        let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let f = fetch_verified(&v, &t, &publisher, all()).await;
        assert_eq!(f.unknown, unknown);
        assert_eq!(f.objects.keys().copied().collect::<std::collections::BTreeSet<_>>(), want);
        let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let _pub2 = publish_tracks(&p, &t2, None).await;
        publish_payloads(&p, &publisher, &t2, 0, vec![object_payload("td", 1, 1), object_payload("copy", 0, 1), object_payload("copy", 0, 2)]).await;
        let media2 = (t2.ns.clone(), t2.media.clone());
        assert!(until(STEP, || d.groups(&media2).contains(&0)).await, "{:?}", d.counts());
        flush(&server).await;
        let f = fetch_verified(&v, &t2, &publisher, all()).await;
        assert_eq!(f.objects.keys().copied().collect::<Vec<_>>(), vec![(0, 1), (0, 2)], "the identical copy is withheld");
        let dir2 = d.track_dir(&media2).unwrap();
        let copy = object_payload("td", 1, 1);
        assert!(!std::fs::read_dir(dir2).unwrap().flatten().any(|e| std::fs::read(e.path()).unwrap().windows(copy.len()).any(|w| w == copy)), "and never stored");
        for c in [&p, &v] {
            c.close(0, "done");
        }
        server.shutdown().await;
        let _ = std::fs::remove_dir_all(&dir);
        let _ = std::fs::remove_dir_all(&key_dir);
    })
    .await
    .expect("in time");
}

/// A grant laid out as live-rooms §4.1 lays it out, with these media bits and powers.
fn grant(authority: &SigningKey, room: [u8; 32], principal: &SigningKey, publish: u8, powers: u8) -> Vec<u8> {
    grant_with(authority, room, principal, publish, powers, &[])
}

/// The same grant with these attributes (live-rooms §4.4), already laid out.
fn grant_with(authority: &SigningKey, room: [u8; 32], principal: &SigningKey, publish: u8, powers: u8, attrs: &[u8]) -> Vec<u8> {
    let now = now_s();
    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(&[0, publish, 1, 0, powers, 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]); // no profile, no origin
    g.extend_from_slice(&(attrs.len() as u16).to_le_bytes());
    g.extend_from_slice(attrs);
    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
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn seeking_back_needs_the_seek_power() {
    timeout(Duration::from_secs(120), async {
        const GROUPS: u64 = 6;
        let dir = store_dir("seek");
        let authority = SigningKey::from_bytes(&[0x5a; 32]);
        let (publisher, plain, seeker) = (SigningKey::from_bytes(&[0x5b; 32]), SigningKey::from_bytes(&[0x5c; 32]), SigningKey::from_bytes(&[0x5d; 32]));
        let room = [0x5e; 32];
        let t = room_tracks(room, &publisher);
        let cfg = dvr_config(&dir, 900, 1 << 30, vec![]);
        let server = relay(|c| {
            c.cache.max_bytes = CACHE;
            c.dvr = Some(cfg.clone());
            c.auth.required = true;
            c.auth.authorities = vec![Authority { key: hexkey(&authority), rooms: vec![] }];
        })
        .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 (p, _) = connect(&server, Via::WebSocket, with(grant(&authority, room, &publisher, 0b010, 0), &publisher)).await;
        let _pub = publish_tracks(&p, &t, None).await;
        for g in 0..GROUPS {
            publish_multi(&p, &publisher, &t, g, "s").await;
        }
        written(&server, 2 * GROUPS).await;
        let (a, _) = connect(&server, Via::WebSocket, with(grant(&authority, room, &plain, 0, 0), &plain)).await;
        let (b, _) = connect(&server, Via::WebSocket, with(grant(&authority, room, &seeker, 0, 1 << 4), &seeker)).await;

        // Without the power: refused below the live edge, a FETCH and a fill alike.
        let refused = a.fetch(t.ns.clone(), t.media.clone(), range((0, 0), GROUPS - 1)).await.expect_err("below the live edge");
        assert!(refused.contains("seek_not_granted") && refused.contains(&joule_moq::draft22::RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
        let fill = Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(1, 0) }));
        let params = Parameters::new().with(Parameter::LocationFilter(LocationFilter::NextObject)).with(Parameter::FillParameters(fill));
        let refused = a.subscribe(t.ns.clone(), t.media.clone(), params).await.expect_err("a fill below the live edge");
        assert!(refused.contains("seek_not_granted"), "{refused}");
        // At the live edge: served.
        let f = fetch_verified(&a, &t, &publisher, range((GROUPS - 1, 0), GROUPS - 1)).await;
        assert_eq!(f.objects.len() as u64, OBJECTS);
        // With it: the window.
        let f = fetch_verified(&b, &t, &publisher, range((0, 0), GROUPS - 1)).await;
        assert_eq!(f.objects.len() as u64, GROUPS * OBJECTS);
        assert!(server.relay().stats().seek_refused >= 2);
        // A Rewind filter: the Groups before the current one are a seek back. Without the
        // power the current Group alone is rewound; with it, the Groups asked for.
        let rewind = || Parameters::new().with(Parameter::LocationFilter(LocationFilter::Rewind { groups: 3 }));
        let (ok, _atx, _arx) = a.subscribe(t.ns.clone(), t.media.clone(), rewind()).await.expect("a Rewind without the power");
        assert_eq!(ok.params.rewind_groups(), Some(0), "the current Group alone");
        let (ok, _btx, _brx) = b.subscribe(t.ns.clone(), t.media.clone(), rewind()).await.expect("a Rewind with the power");
        assert_eq!(ok.params.rewind_groups(), Some(3), "the Groups asked for");
        for c in [&p, &a, &b] {
            c.close(0, "done");
        }
        server.shutdown().await;
        let _ = std::fs::remove_dir_all(&dir);
    })
    .await
    .expect("in time");
}

// ── The Rewind filter ────────────────────────────────────────────────────────

/// A group of `t`'s media track left open upstream: its receipts whole, then its first two
/// objects on a subgroup stream that is not finished. Returns the stream and its codec, for
/// [`finish_group`].
async fn open_group(c: &Client, key: &SigningKey, t: &Tracks, g: u64, tag: &str) -> (joule_moq_relay::transport::BoxSend, joule_moq::draft22::data::SubgroupCodec) {
    publish_parts(c, key, t, g, tag, true, false).await;
    let h = SubgroupHeader { track_alias: t.media_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
    let mut b = Vec::new();
    h.encode(&mut b);
    let mut codec = joule_moq::draft22::data::SubgroupCodec::new(h);
    for o in 0..OBJECTS - 1 {
        codec.encode_object(&SubgroupObject::normal(o, object_payload(tag, g, o)), &mut b).unwrap();
    }
    let mut st = timeout(STEP, c.conn.open_uni()).await.expect("a stream in time").expect("a stream");
    timeout(STEP, st.write_all(&b)).await.expect("written in time").expect("written");
    (st, codec)
}

/// The last object of a group [`open_group`] left open, and its FIN.
async fn finish_group(open: (joule_moq_relay::transport::BoxSend, joule_moq::draft22::data::SubgroupCodec), g: u64, tag: &str) {
    let (mut st, mut codec) = open;
    let mut b = Vec::new();
    codec.encode_object(&SubgroupObject::normal(OBJECTS - 1, object_payload(tag, g, OBJECTS - 1)), &mut b).unwrap();
    timeout(STEP, st.write_all(&b)).await.expect("written in time").expect("written");
    st.finish().expect("a FIN");
}

fn rewind(groups: u64) -> Parameters {
    Parameters::new().with(Parameter::LocationFilter(LocationFilter::Rewind { groups }))
}

/// One stream: its Object IDs, and its reset code if it was reset.
type Stream = (Vec<u64>, Option<u64>);

/// What a subscription's streams carried, by group: each stream's Object IDs and whether it
/// was reset.
fn streams_of(got: &[Got], alias: u64) -> BTreeMap<u64, Vec<Stream>> {
    let mut out: BTreeMap<u64, Vec<Stream>> = BTreeMap::new();
    for g in got.iter().filter(|g| g.header.track_alias == alias) {
        out.entry(g.header.group_id).or_default().push((g.objects.iter().map(|o| o.object_id).collect(), g.reset));
    }
    out
}

fn objects_of(got: &[Got], alias: u64) -> usize {
    got.iter().filter(|g| g.header.track_alias == alias).map(|g| g.objects.len()).sum()
}

async fn closed_with(c: &Client) -> u64 {
    match timeout(STEP, c.conn.closed()).await.expect("the session closes in time") {
        joule_moq_relay::transport::TransportError::Closed { code, .. } => code,
        other => panic!("closed otherwise: {other}"),
    }
}

/// Six groups in the store and a seventh open upstream. A Rewind of three is answered
/// REWIND_GROUPS 3: groups 3, 4 and 5 arrive from the store, each on its own stream, ended;
/// the current group arrives from its first object, and the object its publisher sends after
/// the subscription rides the same stream, which ends when the publisher's does. Every object
/// verifies against the receipts a Rewind of the receipts track brought.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_rewind_is_served_from_the_store_and_the_current_group_from_its_start() {
    timeout(Duration::from_secs(120), async {
        const CURRENT: u64 = 6;
        let dir = store_dir("rewind");
        let room = [0x71; 32];
        let publisher = SigningKey::from_bytes(&[0x72; 32]);
        let t = room_tracks(room, &publisher);
        let cfg = dvr_config(&dir, 900, 1 << 30, vec![]);
        let server = relay(|c| c.dvr = Some(cfg.clone())).await;
        let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let _pub = publish_tracks(&p, &t, None).await;
        for g in 0..CURRENT {
            publish_multi(&p, &publisher, &t, g, "rw").await;
        }
        written(&server, 2 * CURRENT).await;
        let open = open_group(&p, &publisher, &t, CURRENT, "rw").await;
        let media_key = (t.ns.clone(), t.media.clone());
        let rec_key = (t.ns.clone(), t.receipts.clone());
        let r = server.relay().clone();
        assert!(until(STEP, || r.largest(&media_key) == Some(Location::new(CURRENT, OBJECTS - 2)) && r.largest(&rec_key) == Some(Location::new(CURRENT, 0))).await, "the open group reached the relay");

        let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        assert_eq!(joule_moq::rewind::max_rewind(&v.peer_setup.options), Some(8), "the relay's SETUP declares MAX_REWIND");
        let (ok, _mtx, _mrx) = v.subscribe(t.ns.clone(), t.media.clone(), rewind(3)).await.expect("SUBSCRIBE with a Rewind filter");
        assert_eq!((ok.params.rewind_groups(), ok.params.largest_object()), (Some(3), Some(Location::new(CURRENT, OBJECTS - 2))));
        joule_moq::rewind::check_response(Some(3), 3, ok.params.largest_object()).expect("an answer the draft allows");
        let (rok, _rtx, _rrx) = v.subscribe(t.ns.clone(), t.receipts.clone(), rewind(3)).await.expect("SUBSCRIBE the receipts with a Rewind filter");
        assert_eq!(rok.params.rewind_groups(), Some(3));
        // After the subscription: the current group's last object, on the stream left open.
        finish_group(open, CURRENT, "rw").await;
        let (ma, ra) = (ok.track_alias, rok.track_alias);
        let got = collect_concurrent(&v, STEP, Duration::from_millis(400), |g| objects_of(g, ma) as u64 == 4 * OBJECTS && objects_of(g, ra) == 4).await;

        let whole: Vec<Stream> = vec![((0..OBJECTS).collect(), None)];
        let media = streams_of(&got, ma);
        assert_eq!(media.keys().copied().collect::<Vec<_>>(), vec![3, 4, 5, CURRENT], "the three groups before the current one, and it; nothing older");
        for (g, streams) in &media {
            assert_eq!(streams, &whole, "group {g}: one stream, every object in order, ended and not reset");
        }
        // Each object against its receipt, from the receipts the receipts track's Rewind brought.
        let verifier = ReceiptVerifier::with_anchors(vec![publisher.verifying_key()]);
        let bundles: BTreeMap<u64, ReceiptBundle> = got.iter().filter(|g| g.header.track_alias == ra).map(|g| (g.header.group_id, ReceiptBundle::from_payload(&g.objects[0].payload).unwrap())).collect();
        assert_eq!(bundles.keys().copied().collect::<Vec<_>>(), vec![3, 4, 5, CURRENT]);
        for s in got.iter().filter(|g| g.header.track_alias == ma) {
            let b = &bundles[&s.header.group_id];
            let vg = verifier.verify_group(&b.group).expect("the group receipt verifies");
            for o in &s.objects {
                assert_eq!(o.payload, object_payload("rw", s.header.group_id, o.object_id));
                let rcpt = b.objects.iter().find(|r| r.object_id == ObjectId(o.object_id)).expect("a receipt for the object");
                verifier.verify_payload(&o.payload, rcpt, &vg).unwrap_or_else(|e| panic!("({}, {}) against its receipt: {e}", s.header.group_id, o.object_id));
            }
        }
        let stats = server.relay().stats();
        assert_eq!((stats.rewinds, stats.rewound_groups), (2, 6));

        // Planted: a Rewind above the MAX_REWIND the relay declared closes the session.
        let (bad, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let m = joule_moq::draft22::ControlMessage::Subscribe(joule_moq::draft22::TrackRequest { request_id: bad.next_request_id(), namespace: t.ns.clone(), name: t.media.clone(), params: rewind(9) });
        let _held = bad.request(&m).await.expect("the request stream");
        assert_eq!(closed_with(&bad).await, joule_moq::draft22::SessionError::PROTOCOL_VIOLATION.0, "a Rewind above MAX_REWIND");

        for c in [&p, &v] {
            c.close(0, "done");
        }
        server.shutdown().await;
        let _ = std::fs::remove_dir_all(&dir);
    })
    .await
    .expect("in time");
}

/// Planted: a Rewind that reaches past the store's window is answered with the Groups the
/// store still holds, stated, and none older arrives; and a Rewind on a track no store keeps
/// is answered REWIND_GROUPS 0, the current Group alone.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_rewind_past_the_window_and_one_with_no_store() {
    timeout(Duration::from_secs(120), async {
        // Past the window: a budget of a few groups, and twenty published.
        const CURRENT: u64 = 20;
        let dir = store_dir("rewind-window");
        let room = [0x73; 32];
        let publisher = SigningKey::from_bytes(&[0x74; 32]);
        let t = room_tracks(room, &publisher);
        let cfg = dvr_config(&dir, 900, 16_000, vec![]);
        let server = relay(|c| c.dvr = Some(cfg.clone())).await;
        let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let _pub = publish_tracks(&p, &t, None).await;
        for g in 0..CURRENT {
            publish_multi(&p, &publisher, &t, g, "pw").await;
        }
        written(&server, 2 * CURRENT).await;
        let open = open_group(&p, &publisher, &t, CURRENT, "pw").await;
        let media_key = (t.ns.clone(), t.media.clone());
        let r = server.relay().clone();
        assert!(until(STEP, || r.largest(&media_key) == Some(Location::new(CURRENT, OBJECTS - 2))).await);
        let held = server.relay().dvr().unwrap().groups(&media_key);
        let k = held.len() as u64;
        assert!(k > 0 && k < 8 && held == (CURRENT - k..CURRENT).collect::<Vec<_>>(), "the window holds a few of the newest groups: {held:?}");
        let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let (ok, _tx, _rx) = v.subscribe(t.ns.clone(), t.media.clone(), rewind(8)).await.expect("SUBSCRIBE");
        assert_eq!(ok.params.rewind_groups(), Some(k), "what the store holds, not what was asked");
        // The current Group is finished once the stored ones have been read. Finished, it is
        // stored, and a store at its budget drops its oldest Group for it; a Group the store
        // dropped since the count was made is not sent (the README, "The Rewind filter"). The
        // test finished it at once, and on a loaded host the oldest Group was gone before it
        // was read: four Groups of the five counted.
        let stored = std::sync::Arc::new(tokio::sync::Notify::new());
        let finishing = tokio::spawn({
            let stored = stored.clone();
            async move {
                stored.notified().await;
                finish_group(open, CURRENT, "pw").await;
            }
        });
        let got = collect_concurrent(&v, STEP, Duration::from_millis(400), |g| {
            let n = objects_of(g, ok.track_alias) as u64;
            if n >= k * OBJECTS {
                stored.notify_one();
            }
            n == (k + 1) * OBJECTS
        })
        .await;
        finishing.abort();
        assert_eq!(streams_of(&got, ok.track_alias).keys().copied().collect::<Vec<_>>(), (CURRENT - k..=CURRENT).collect::<Vec<_>>(), "nothing from before the window");
        for c in [&p, &v] {
            c.close(0, "done");
        }
        server.shutdown().await;
        let _ = std::fs::remove_dir_all(&dir);

        // No store: the cache's current Group from its start, and nothing before it.
        let server = relay(|_| {}).await;
        let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let _pub = publish_tracks(&p, &t, None).await;
        for g in 0..3 {
            publish_multi(&p, &publisher, &t, g, "ns").await;
        }
        let open = open_group(&p, &publisher, &t, 3, "ns").await;
        let r = server.relay().clone();
        assert!(until(STEP, || r.largest(&media_key) == Some(Location::new(3, OBJECTS - 2))).await);
        let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let (ok, _tx, _rx) = v.subscribe(t.ns.clone(), t.media.clone(), rewind(3)).await.expect("SUBSCRIBE");
        assert_eq!(ok.params.rewind_groups(), Some(0), "no store: no Group before the current one");
        finish_group(open, 3, "ns").await;
        let got = collect_concurrent(&v, STEP, Duration::from_millis(400), |g| objects_of(g, ok.track_alias) as u64 == OBJECTS).await;
        let whole: Vec<Stream> = vec![((0..OBJECTS).collect(), None)];
        assert_eq!(streams_of(&got, ok.track_alias), BTreeMap::from([(3, whole)]), "the current Group from its start, on one stream");
        for c in [&p, &v] {
            c.close(0, "done");
        }
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

/// Planted: a stored object that no longer verifies. One whose bytes changed (4, 1), and one
/// changed with its checks made again so that only its receipt finds it (5, 0), are dropped
/// when the store is read again. A Rewind sends neither: group 4's stream carries its first
/// object and is reset, nothing of group 5 is sent, and group 6 and the current Group arrive
/// whole.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_rewound_object_that_fails_its_receipt_is_never_sent() {
    timeout(Duration::from_secs(120), async {
        const STORED: u64 = 7;
        let dir = store_dir("rewind-receipt");
        let room = [0x75; 32];
        let publisher = SigningKey::from_bytes(&[0x76; 32]);
        let t = room_tracks(room, &publisher);
        let cfg = dvr_config(&dir, 900, 1 << 30, vec![]);
        let config = |c: &mut RelayConfig| c.dvr = Some(cfg.clone());
        let server = relay(config).await;
        let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let _pub = publish_tracks(&p, &t, None).await;
        for g in 0..STORED {
            publish_multi(&p, &publisher, &t, g, "rr").await;
        }
        written(&server, 2 * STORED).await;
        let key = (t.ns.clone(), t.media.clone());
        let d = server.relay().dvr().unwrap().clone();
        let ((f4, at4), (f5, at5)) = (d.group_location(&key, 4).unwrap(), d.group_location(&key, 5).unwrap());
        p.close(0, "done");
        server.shutdown().await;
        tamper(&f4, at4, 1, false, |p| p[10] ^= 0x20);
        tamper(&f5, at5, 0, true, |p| p.reverse());

        let server = relay(config).await;
        assert_eq!(server.relay().dvr().unwrap().counts().objects_dropped, 2);
        let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let _pub = publish_tracks(&p, &t, None).await;
        let open = open_group(&p, &publisher, &t, STORED, "rr").await;
        let r = server.relay().clone();
        assert!(until(STEP, || r.largest(&key) == Some(Location::new(STORED, OBJECTS - 2))).await);
        let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let (ok, _tx, _rx) = v.subscribe(t.ns.clone(), t.media.clone(), rewind(3)).await.expect("SUBSCRIBE");
        assert_eq!(ok.params.rewind_groups(), Some(3), "the store holds the three groups");
        finish_group(open, STORED, "rr").await;
        let got = collect_concurrent(&v, STEP, Duration::from_millis(600), |g| objects_of(g, ok.track_alias) as u64 == 1 + 2 * OBJECTS && streams_of(g, ok.track_alias).len() == 3).await;
        let whole: Vec<Stream> = vec![((0..OBJECTS).collect(), None)];
        let reset = joule_moq::draft22::StreamResetCode::UNKNOWN_OBJECT_STATUS.0;
        assert_eq!(
            streams_of(&got, ok.track_alias),
            BTreeMap::from([(4, vec![(vec![0], Some(reset))]), (6, whole.clone()), (STORED, whole)]),
            "group 4 up to the object that failed, reset; nothing of group 5; the rest whole"
        );
        for s in got.iter().filter(|g| g.header.track_alias == ok.track_alias) {
            for o in &s.objects {
                assert_eq!(o.payload, object_payload("rr", s.header.group_id, o.object_id), "no changed byte was sent");
            }
        }

        // Planted while the relay runs, with no restart to find it: a byte of (6, 1) changed in
        // its file. The store holds each object it serves to the digest its index has, as it
        // reads it, so a second Rewind sends group 6 up to that object and resets its stream.
        let (f6, at6) = server.relay().dvr().unwrap().group_location(&key, 6).unwrap();
        tamper(&f6, at6, 1, false, |p| p[3] ^= 0x01);
        let (w, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let (ok, _wtx, _wrx) = w.subscribe(t.ns.clone(), t.media.clone(), rewind(1)).await.expect("SUBSCRIBE");
        assert_eq!(ok.params.rewind_groups(), Some(1));
        let got = collect_concurrent(&w, STEP, Duration::from_millis(600), |g| streams_of(g, ok.track_alias).len() == 2).await;
        let whole: Vec<Stream> = vec![((0..OBJECTS).collect(), None)];
        assert_eq!(streams_of(&got, ok.track_alias), BTreeMap::from([(6, vec![(vec![0], Some(reset))]), (STORED, whole)]), "group 6 up to the object changed while the relay ran, reset");
        for s in got.iter().filter(|g| g.header.track_alias == ok.track_alias) {
            for o in &s.objects {
                assert_eq!(o.payload, object_payload("rr", s.header.group_id, o.object_id), "no changed byte was sent");
            }
        }
        for c in [&p, &v, &w] {
            c.close(0, "done");
        }
        server.shutdown().await;
        let _ = std::fs::remove_dir_all(&dir);
    })
    .await
    .expect("in time");
}

/// The attribute `minor` = `true` (live-rooms §4.4): a short key and a long value.
fn minor_attr() -> Vec<u8> {
    [&[5u8][..], b"minor", &4u16.to_le_bytes(), b"true"].concat()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_minors_room_is_never_kept() {
    timeout(Duration::from_secs(120), async {
        let dir = store_dir("minors");
        let authority = SigningKey::from_bytes(&[0x6a; 32]);
        let (publisher, other, minor, watcher) = (SigningKey::from_bytes(&[0x6b; 32]), SigningKey::from_bytes(&[0x6c; 32]), SigningKey::from_bytes(&[0x6d; 32]), SigningKey::from_bytes(&[0x6e; 32]));
        let (room, clear) = ([0x6f; 32], [0x70; 32]);
        let t = room_tracks(room, &publisher);
        let tc = room_tracks(clear, &other);
        let cfg = dvr_config(&dir, 900, 1 << 30, vec![]);
        let config = |c: &mut RelayConfig| {
            c.cache.max_bytes = CACHE;
            c.dvr = Some(cfg.clone());
            c.auth.required = true;
            c.auth.authorities = vec![Authority { key: hexkey(&authority), rooms: vec![] }];
        };
        let server = relay(config).await;
        let d = server.relay().dvr().unwrap().clone();
        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 (p, _) = connect(&server, Via::WebSocket, with(grant(&authority, room, &publisher, 0b010, 0), &publisher)).await;
        let (o, _) = connect(&server, Via::WebSocket, with(grant(&authority, clear, &other, 0b010, 0), &other)).await;
        let _pt = publish_tracks(&p, &t, None).await;
        let _pc = publish_tracks(&o, &tc, None).await;
        for g in 0..2 {
            publish_multi(&p, &publisher, &t, g, "m").await;
            publish_multi(&o, &other, &tc, g, "c").await;
        }
        written(&server, 8).await;
        let media = (t.ns.clone(), t.media.clone());
        let clear_media = (tc.ns.clone(), tc.media.clone());
        assert_eq!(d.groups(&media), vec![0, 1], "kept until a minor publishes in the room");
        let media_dir = d.track_dir(&media).expect("the room's track, kept so far");

        // A minor's grant that publishes nothing: nothing changes.
        let (w, _) = connect(&server, Via::WebSocket, with(grant_with(&authority, room, &watcher, 0, 0, &minor_attr()), &watcher)).await;
        publish_multi(&p, &publisher, &t, 2, "m").await;
        assert!(until(STEP, || d.groups(&media).contains(&2)).await, "a minor's grant that publishes nothing keeps the window");
        assert_eq!(d.counts().rooms_sealed, 0);

        // A minor's publishing grant, proved: the room is forgotten and never kept again.
        let (m, _) = connect(&server, Via::WebSocket, with(grant_with(&authority, room, &minor, 0b011, 0, &minor_attr()), &minor)).await;
        assert!(until(STEP, || d.counts().rooms_sealed == 1).await, "{:?}", d.counts());
        flush(&server).await;
        let receipts = (t.ns.clone(), t.receipts.clone());
        for k in [&media, &receipts] {
            assert!(d.groups(k).is_empty() && d.track_dir(k).is_none(), "{k:?} is kept");
        }
        assert!(!media_dir.exists(), "the room's media is deleted from disk");
        assert_eq!(std::fs::metadata(dir.join("sealed.wdl")).unwrap().len(), 64, "the room is recorded among those never kept");
        publish_multi(&p, &publisher, &t, 3, "m").await;
        publish_multi(&o, &other, &tc, 2, "c").await;
        assert!(until(STEP, || server.relay().largest(&media) == Some(Location::new(3, OBJECTS - 1))).await, "the room is still delivered");
        assert!(until(STEP, || d.groups(&clear_media).contains(&2)).await);
        flush(&server).await;
        assert!(d.groups(&media).is_empty(), "nothing more of the room is written");
        assert_eq!(d.groups(&clear_media), vec![0, 1, 2], "the other room keeps its window");

        // Across a restart, the room stays forgotten.
        for c in [&p, &o, &w, &m] {
            c.close(0, "done");
        }
        server.shutdown().await;
        let server = relay(config).await;
        let d = server.relay().dvr().unwrap().clone();
        assert_eq!(d.counts().rooms_sealed, 1, "{:?}", d.counts());
        // The restarted relay listens on another port: its proofs name the new target.
        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 (p, _) = connect(&server, Via::WebSocket, with(grant(&authority, room, &publisher, 0b010, 0), &publisher)).await;
        let _pt = publish_tracks(&p, &t, None).await;
        publish_multi(&p, &publisher, &t, 4, "m").await;
        assert!(until(STEP, || server.relay().largest(&media) == Some(Location::new(4, OBJECTS - 1))).await);
        flush(&server).await;
        assert!(d.groups(&media).is_empty() && d.track_dir(&media).is_none());
        assert_eq!(d.groups(&clear_media), vec![0, 1, 2]);
        p.close(0, "done");
        server.shutdown().await;
        let _ = std::fs::remove_dir_all(&dir);
    })
    .await
    .expect("in time");
}

#[test]
fn a_window_past_a_day_is_refused() {
    let cfg = dvr_config(&PathBuf::from("/nonexistent"), 86_401, 1, vec![]);
    let mut c = RelayConfig::loopback();
    c.dvr = Some(cfg);
    assert!(joule_moq_relay::Relay::new(c).is_err());
}