//! The DVR store: a window of a room's tracks, with their receipts, kept on disk and served
//! to FETCH below the live edge (WAI `live-recording.md` §8).
//!
//! **What it keeps.** Every track whose namespace a `[[dvr.window]]` rule matches (the rule
//! with the longest prefix): media tracks, their receipts tracks, events tracks, catalogs
//! and takes alike. A group is stored once the cache holds it whole (`crate::cache`): its
//! objects as they arrived, with every field a FETCH carries (Subgroup ID or datagram,
//! publisher priority, properties, status, payload). Group *g* of a receipts track is the
//! receipts of media group *g* (WAI `moq-streaming-format.md` §4.1), so a viewer that
//! fetches both checks a recorded object exactly as it checks a live one.
//!
//! **The window.** Each rule has an age (`max_age_s`, at most 24 hours) and a byte budget
//! (`max_bytes`) over its tracks together. Past either, the oldest group goes first, whole,
//! and group *g* of a media track goes with group *g* of its receipts track. A group at or
//! below a track's eviction mark is no longer known to the relay: a FETCH that reaches it
//! is told so with an End of Unknown Range (MOQT §3.2.2, §11.4.1.2), since an evicted
//! object existed, and an unmarked gap would say it did not.
//!
//! **On disk.** One directory per track under `tracks/`, named by the first 16 bytes of the
//! SHA-256 of its rendered name (MOQT §8.8) in hexadecimal, holding `TRACK` (its namespace,
//! name and properties, with a SHA-256 check) and its **segments**, `s<n, 20 digits>.wds`:
//! files a group record is appended to as each group is stored, a new one begun every ten
//! seconds or 16 MiB. A segment is a sequence of frames, `len(u32) state(u8) record`, the
//! state 1 for a record that stands and 0 for one a later record replaced; a record is
//!
//! ```text
//! header   "WDG1" 1 track(32) group(u64) arrived_us(u64) count(u32) erased(u32) check(32)
//! object   id(u64) subgroup(u64, 2^64-1 for a datagram) priority(u8) status(u8)
//!          props_len(u32) payload_len(u32) digest(32) check(32) props payload
//! erased   id(u64) digest(32) why(u8: 1 redacted, 2 did not verify, 3 taken down)
//! ```
//!
//! Integers are little-endian. `track` is the SHA-256 of the rendered track; a header's
//! `check` is SHA-256 over the header's bytes before it; an object's `digest` is the content
//! hash its receipt names it by (WAI `jwp-receipts.md` §2.1), and its `check` SHA-256 over
//! its fixed fields before `check`, its properties and its payload. A record that replaces
//! another (a group written again: an object erased, or a subgroup that came late) is
//! appended, and the old frame's state set to 0 and its record's bytes overwritten with
//! zeros. A segment is deleted once no group's standing record is in it.
//!
//! **Durability.** The writer syncs every segment it wrote, and the directories it made
//! files in, at least once a second and before it answers a flush or stops; a shutdown
//! flushes. A crash loses at most the last second's groups. On start the store reads every
//! segment again: a frame a crash cut short, at a segment's end, is cut off and counted
//! (`torn_writes`); a record whose header does not check is dropped; an object whose
//! `check` fails, or, on a track with a receipts track, whose payload does not verify
//! against group *g*'s receipts (the group receipt's signature under the door's trust for
//! the namespace, the object's receipt at its position, its Merkle proof to the signed
//! root), or, on an events track, a record of which does not verify under its principal, is
//! dropped and counted, and its group written again with the object among the erased. It is
//! never served: a FETCH that reaches it is told End of Unknown Range. Each object's digest
//! is checked again as it is read.
//!
//! **Moderation.** The relay reads the `decision` events (WAI `live-events.md` §4.11) on the
//! events tracks it already forwards. A decision is honoured when its record verifies, its
//! author owns the track it rides and is a moderator the relay knows: a room key it is
//! configured with (`dvr.room_keys`, the keys of `auth.authorities`), or the principal of a
//! grant with the moderate power its publishing session proved. The last honoured decision
//! on a target stands (WAI `live-moderation.md` §4.5): deny and redact withhold the target
//! from every DVR read, and from every FETCH the cache answers; redact also erases its bytes
//! from the store (an event record from the object that carries it, the object's other
//! records kept). Allow restores what deny withheld. The honoured decisions are kept in
//! `decisions.wdl` and survive a restart.
//!
//! **A room whose objects are sealed** (WAI `confidentiality.md`: an end-to-end encrypted
//! room) is never kept, since such a room is never recorded and a relay keeps nothing of its
//! tracks beyond delivery (§9). The store knows one by its first sealed object: an object
//! whose properties carry the secure-objects Key ID (§4.3), which a relay reads without a key.
//! A group holding one is never written. From then on the store writes nothing of the room,
//! deletes what it already held (the room's events and receipts tracks, in the clear, can be
//! whole before its first media group is), and drops the room's decisions from
//! `decisions.wdl`; it still honours them for what the cache serves. The room is recorded in
//! `sealed.wdl`, `room(32) check(32)` per room with `check` the SHA-256 of the room, so a
//! restart keeps refusing it.
//!
//! **A room in which a minor publishes** (WAI `age-assurance.md` §8.4) is never kept either.
//! The relay knows one by a minor's publishing grant: a session whose SETUP proved a grant of
//! an authority it trusts whose `publish` is not 0 and whose attribute `minor` is `true`
//! (WAI `live-rooms.md` §4.4). From then on the store treats the room as it treats a sealed
//! one: it writes nothing of it, deletes what it held, keeps none of its decisions, and
//! records it in `sealed.wdl`, which so holds every room the store never keeps.
//!
//! **Takedowns** (WAI `takedown.md` §10.1). A signed takedown (`crate::takedown`) is honoured
//! when its signer is a key the store trusts (`dvr.takedown_keys`, and the keys of
//! `auth.authorities`) and its signature verifies; it rides as an object of the track
//! `[key, takedown]`, `0`, whose key is the signer's, or is given to [`Dvr::takedown`]. The
//! store then forgets every covered object, as it forgets a sealed room, whatever its room: an
//! object whose content hash its `exact` set or its `matched` set names, and every object of a
//! track its `exact` set names (by the track's digest, WAI live-rooms §6.2) from the group it
//! names on. Each is erased from every group that holds it (`why` 3), never stored again (a
//! group that becomes whole later stores it erased), and never served, to a moderator's
//! session either. The takedowns are kept in `takedowns.wdl`, `len(u32) record check(32)` per
//! takedown, and verified again on start, which makes again any erasure a crash left undone.
//! Once an erasure is on disk the store signs a purge with the relay's key and writes it to
//! `purges/<link>.json`, the takedown's link hash in hexadecimal, once.

use crate::auth::pattern_matches;
use crate::cache::CachedObject;
use crate::config::DvrConfig;
use crate::events::{self, Event, REDACT};
use crate::relay::TrackKey;
use bytes::Bytes;
use ed25519_dalek::SigningKey;
use joule_moq::carriage::ReceiptBundle;
use joule_moq::draft22::data::ObjectStatus;
use joule_moq::draft22::kvp::{prop, Properties};
use joule_moq::draft22::names::render;
use joule_moq::draft22::Location;
use joule_moq::receipt::{ObjectReceipt, ReceiptVerifier};
use joule_moq::secure_objects::{self, Refusal};
use joule_moq::{GroupId, ObjectId, TrackName, TrackNamespace};
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::io::Write;
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{mpsc, Arc, Mutex};
use std::time::{Duration, Instant};

const GROUP_MAGIC: &[u8; 4] = b"WDG1";
const TRACK_MAGIC: &[u8; 4] = b"WDT1";
const VERSION: u8 = 1;
const HEADER_LEN: usize = 4 + 1 + 32 + 8 + 8 + 4 + 4 + 32;
const OBJECT_FIXED: usize = 8 + 8 + 1 + 1 + 4 + 4 + 32 + 32;
/// A frame's length and state, before its record.
const FRAME: u64 = 5;
const DATAGRAM: u64 = u64::MAX;
/// How long a segment takes groups, and how large it grows, before another is begun.
const SEGMENT_SPAN_US: u64 = 10_000_000;
const SEGMENT_MAX: u64 = 16 << 20;
/// How often the writer syncs what it wrote.
const SYNC_EVERY: Duration = Duration::from_secs(1);
/// Why an object is among a group's erased.
pub const ERASED_REDACTED: u8 = 1;
pub const ERASED_UNVERIFIED: u8 = 2;
pub const ERASED_TAKEN_DOWN: u8 = 3;
/// The resolution of the store's metering series.
const POINT_US: u64 = 1_000_000;
/// How long the metering series keeps its points.
const SERIES_KEEP_US: u64 = 48 * 3_600 * 1_000_000;

fn sha256(parts: &[&[u8]]) -> [u8; 32] {
    let mut c = ring::digest::Context::new(&ring::digest::SHA256);
    for p in parts {
        c.update(p);
    }
    c.finish().as_ref().try_into().expect("32 bytes")
}

pub fn now_us() -> u64 {
    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_micros() as u64)
}

/// The room a namespace is a track of (WAI live-rooms §6.1): its first field, 64 lowercase
/// hexadecimal digits.
pub fn room_of(ns: &TrackNamespace) -> Option<[u8; 32]> {
    let f = ns.0.first()?;
    if f.len() != 64 || !f.iter().all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(b)) {
        return None;
    }
    hex::decode(f).ok()?.try_into().ok()
}

/// A key a namespace field names, 64 lowercase hexadecimal digits.
fn key_field(ns: &TrackNamespace, i: usize) -> Option<[u8; 32]> {
    let f = ns.0.get(i)?;
    if f.len() != 64 {
        return None;
    }
    hex::decode(f).ok()?.try_into().ok()
}

/// Whether a track is a room's events track (live-events §8.1): `[R, P, events]`, `0`.
pub fn is_events_track(key: &TrackKey) -> bool {
    key.0 .0.len() == 3 && key.0 .0[2] == b"events" && key.1 .0 == b"0" && room_of(&key.0).is_some()
}

fn track_digest(key: &TrackKey) -> [u8; 32] {
    sha256(&[render(&key.0, Some(&key.1)).as_bytes()])
}

/// A track's digest as a takedown names it (WAI live-rooms §6.2): SHA-256 over the MOQT §8.8
/// rendering of its namespace, one zero byte, and its name.
pub fn room_track_digest(key: &TrackKey) -> [u8; 32] {
    sha256(&[render(&key.0, None).as_bytes(), &[0], &key.1 .0])
}

/// Whether a track is one takedowns ride (WAI takedown §10.1): `[K, takedown]`, `0`, with `K`
/// a key in 64 lowercase hexadecimal digits. The key, if so.
pub fn takedown_track_key(key: &TrackKey) -> Option<[u8; 32]> {
    (key.0 .0.len() == 2 && key.0 .0[1] == crate::takedown::TRACK_FIELD && key.1 .0 == b"0").then(|| room_of(&key.0)).flatten()
}

// ── The index ──────────────────────────────────────────────────────────────

#[derive(Clone, Debug)]
struct DObject {
    id: u64,
    subgroup: Option<u64>,
    priority: u8,
    status: ObjectStatus,
    /// Where its properties begin in its segment; its payload follows them.
    at: u64,
    props_len: u32,
    payload_len: u32,
    digest: [u8; 32],
    /// The digests of its event records, on an events track.
    records: Vec<[u8; 32]>,
}

#[derive(Clone, Debug)]
struct DGroup {
    arrived_us: u64,
    /// Payload bytes of its objects.
    bytes: u64,
    objects: Vec<DObject>,
    /// Objects no longer held, by Object ID, with their digests.
    erased: Vec<(u64, [u8; 32], u8)>,
    /// Its standing record: the segment, and the frame's place and record length.
    seg: u64,
    frame_at: u64,
    record_len: u32,
}

#[derive(Debug)]
struct DTrack {
    dir: PathBuf,
    window: usize,
    room: Option<[u8; 32]>,
    events: bool,
    props: Properties,
    groups: BTreeMap<u64, DGroup>,
    /// Groups known not to exist (a Prior Group ID Gap, MOQT §10.8), `lo ..= hi` by `lo`.
    absent: BTreeMap<u64, u64>,
    /// Groups at or below this are no longer known to the store.
    evicted_through: Option<u64>,
    end_of_track: Option<Location>,
    /// Each segment, with the groups whose standing record is in it.
    segs: BTreeMap<u64, usize>,
}

impl DTrack {
    fn largest(&self) -> Option<Location> {
        let (g, grp) = self.groups.iter().next_back()?;
        let o = grp.objects.iter().map(|o| o.id).chain(grp.erased.iter().map(|e| e.0)).max()?;
        Some(Location::new(*g, o))
    }
}

/// The store's metering of one room (WAI live-metering §3.9): what it wrote, what it holds
/// over time, what it dropped. Points are kept each second a change falls in.
#[derive(Clone, Debug, Default)]
struct Series {
    t_last: u64,
    held: u64,
    acc: u128,
    stored_objects: u64,
    stored_bytes: u64,
    dropped: u64,
    points: Vec<(u64, u128, u64, u64, u64, u64)>,
}

impl Series {
    fn change(&mut self, now: u64, held: i128, objects: u64, bytes: u64, dropped: u64) {
        if self.t_last != 0 && now > self.t_last {
            self.acc += self.held as u128 * (now - self.t_last) as u128;
        }
        self.t_last = self.t_last.max(now);
        self.held = (self.held as i128 + held).max(0) as u64;
        self.stored_objects += objects;
        self.stored_bytes += bytes;
        self.dropped += dropped;
        let p = (self.t_last, self.acc, self.held, self.stored_objects, self.stored_bytes, self.dropped);
        match self.points.last_mut() {
            Some(last) if last.0 / POINT_US == p.0 / POINT_US => *last = p,
            _ => self.points.push(p),
        }
        if self.points.first().is_some_and(|f| f.0 + SERIES_KEEP_US < now) {
            self.points.retain(|x| x.0 + SERIES_KEEP_US >= now);
        }
    }

    /// (byte-microseconds held, objects stored, bytes stored, dropped) up to `x`.
    fn at(&self, x: u64) -> (u128, u64, u64, u64) {
        if self.t_last != 0 && x >= self.t_last {
            return (self.acc + self.held as u128 * (x - self.t_last) as u128, self.stored_objects, self.stored_bytes, self.dropped);
        }
        match self.points.partition_point(|p| p.0 <= x) {
            0 => (0, 0, 0, 0),
            i => {
                let p = self.points[i - 1];
                (p.1 + p.2 as u128 * (x - p.0) as u128, p.3, p.4, p.5)
            }
        }
    }
}

/// What the store did for one room over a span (WAI live-metering §3.9).
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct RoomFigures {
    pub objects_stored: u64,
    pub bytes_stored: u64,
    /// Payload bytes held, integrated over the span, in byte-seconds, rounded down.
    pub byte_s: u64,
    /// Objects dropped because they did not verify against their receipts.
    pub dropped: u64,
}

#[derive(Default)]
struct Index {
    tracks: HashMap<TrackKey, DTrack>,
    window_bytes: Vec<u64>,
    /// Where each digest is held: (track, group, Object ID).
    by_digest: HashMap<[u8; 32], Vec<(TrackKey, u64, u64)>>,
    /// The standing decision on each (room, target): its verdict.
    standing: HashMap<([u8; 32], [u8; 32]), u8>,
    decided: HashSet<[u8; 32]>,
    series: HashMap<[u8; 32], Series>,
    /// Rooms whose objects are sealed: nothing of them is kept.
    sealed: HashSet<[u8; 32]>,
    /// The takedowns honoured, by link hash; the objects they cover, by content hash; and
    /// the tracks they cover, by digest, each from a group on.
    takedowns: HashSet<[u8; 32]>,
    taken: HashSet<[u8; 32]>,
    taken_tracks: Vec<([u8; 32], u64)>,
}

impl Index {
    fn withheld(&self, room: Option<[u8; 32]>, digest: &[u8; 32]) -> bool {
        room.is_some_and(|r| self.standing.get(&(r, *digest)).is_some_and(|v| *v != events::ALLOW))
    }

    /// Whether a takedown covers group `group` of `key`, by its track.
    fn track_gone(&self, key: &TrackKey, group: u64) -> bool {
        if self.taken_tracks.is_empty() {
            return false;
        }
        let d = room_track_digest(key);
        self.taken_tracks.iter().any(|(t, from)| *t == d && group >= *from)
    }

    /// Record a takedown honoured. False when it was already.
    fn take(&mut self, t: &crate::takedown::Takedown) -> bool {
        if !self.takedowns.insert(t.link) {
            return false;
        }
        self.taken.extend(t.objects.iter().copied());
        for x in &t.tracks {
            if !self.taken_tracks.contains(x) {
                self.taken_tracks.push(*x);
            }
        }
        true
    }

    fn unindex(&mut self, key: &TrackKey, g: u64, o: &DObject) {
        for d in std::iter::once(&o.digest).chain(o.records.iter()) {
            if let Some(v) = self.by_digest.get_mut(d) {
                v.retain(|x| !(x.0 == *key && x.1 == g && x.2 == o.id));
                if v.is_empty() {
                    self.by_digest.remove(d);
                }
            }
        }
    }

    fn reindex(&mut self, key: &TrackKey, g: u64, o: &DObject) {
        for d in std::iter::once(&o.digest).chain(o.records.iter()) {
            let e = self.by_digest.entry(*d).or_default();
            if !e.iter().any(|x| x.0 == *key && x.1 == g && x.2 == o.id) {
                e.push((key.clone(), g, o.id));
            }
        }
    }
}

/// Counters across the store.
#[derive(Default)]
pub struct DvrStats {
    pub groups_written: AtomicU64,
    pub objects_written: AtomicU64,
    pub groups_evicted: AtomicU64,
    /// Groups that became whole after the window had moved past them (their receipts or
    /// media group already evicted): not stored.
    pub groups_late: AtomicU64,
    /// Objects dropped because their check or their receipt did not verify.
    pub objects_dropped: AtomicU64,
    /// Frames a crash cut short at a segment's end, cut off at start.
    pub torn_writes: AtomicU64,
    /// Objects erased by a redaction.
    pub objects_redacted: AtomicU64,
    pub decisions: AtomicU64,
    /// Segments synced, and the syncs that wrote them out.
    pub syncs: AtomicU64,
    /// Takedowns honoured, objects they erased, and purges signed.
    pub takedowns: AtomicU64,
    pub objects_taken_down: AtomicU64,
    pub purges_signed: AtomicU64,
}

/// A snapshot of [`DvrStats`].
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct DvrCounts {
    pub groups_written: u64,
    pub objects_written: u64,
    pub groups_evicted: u64,
    pub groups_late: u64,
    pub objects_dropped: u64,
    pub torn_writes: u64,
    pub objects_redacted: u64,
    pub decisions: u64,
    pub syncs: u64,
    pub groups_held: u64,
    pub bytes_held: u64,
    pub segments: u64,
    /// Rooms whose objects are sealed, of which the store keeps nothing.
    pub rooms_sealed: u64,
    pub takedowns: u64,
    pub objects_taken_down: u64,
    pub purges_signed: u64,
}

/// One object read from the store, or a place it can no longer serve.
#[derive(Clone, Debug)]
pub enum Slot {
    Object(StoredObject),
    /// No longer held, or withheld: its Object ID.
    Unknown(u64),
}

#[derive(Clone, Debug)]
pub struct StoredObject {
    pub id: u64,
    pub subgroup: Option<u64>,
    pub priority: u8,
    pub status: ObjectStatus,
    pub properties: Properties,
    pub payload: Bytes,
}

/// What the store holds of a track over a span of groups, for a FETCH's plan.
#[derive(Clone, Debug, Default)]
pub struct Snapshot {
    pub held: BTreeSet<u64>,
    pub absent: Vec<(u64, u64)>,
    pub evicted_through: Option<u64>,
    pub largest: Option<Location>,
    pub end_of_track: Option<Location>,
    pub props: Properties,
}

impl Snapshot {
    /// Whether group `g` is known not to exist, and the last group of its run if so.
    pub fn absent_through(&self, g: u64) -> Option<u64> {
        self.absent.iter().filter(|(lo, hi)| *lo <= g && g <= *hi).map(|(_, hi)| *hi).max()
    }
}

enum Job {
    Seal { key: TrackKey, props: Properties, group: u64, arrived_us: u64, objects: Vec<((u64, u64), CachedObject)> },
    Erase { room: [u8; 32], digest: [u8; 32] },
    Log(Vec<u8>),
    Forget([u8; 32]),
    TakeDown { t: crate::takedown::Takedown, rec: Vec<u8> },
    Flush(mpsc::Sender<()>),
}

struct Window {
    prefix: TrackNamespace,
    max_age_us: u64,
    max_bytes: u64,
}

/// The segment a track appends to.
struct OpenSeg {
    seq: u64,
    file: std::fs::File,
    started_us: u64,
    len: u64,
}

/// What the writer alone holds: each track's open segment, and what it has to sync.
#[derive(Default)]
struct Writer {
    open: HashMap<TrackKey, OpenSeg>,
    next: HashMap<TrackKey, u64>,
    dirty: HashSet<PathBuf>,
    dirs: HashSet<PathBuf>,
    last_sync: Option<Instant>,
}

pub struct Dvr {
    root: PathBuf,
    windows: Vec<Window>,
    suffix: Vec<u8>,
    room_keys: Vec<[u8; 32]>,
    /// The keys whose takedowns the store honours.
    takedown_keys: Vec<[u8; 32]>,
    /// The key the store signs its purges with: the relay's.
    purge_key: Option<SigningKey>,
    index: Mutex<Index>,
    tx: Mutex<Option<mpsc::Sender<Job>>>,
    worker: Mutex<Option<std::thread::JoinHandle<()>>>,
    pub stats: DvrStats,
}

fn put_short_field(o: &mut Vec<u8>, b: &[u8]) {
    o.extend_from_slice(&(b.len() as u16).to_le_bytes());
    o.extend_from_slice(b);
}

fn track_dir_name(key: &TrackKey) -> String {
    hex::encode(&track_digest(key)[..16])
}

fn seg_path(dir: &Path, seq: u64) -> PathBuf {
    dir.join(format!("s{seq:020}.wds"))
}

fn seg_seq(path: &Path) -> Option<u64> {
    let n = path.file_name()?.to_str()?;
    n.strip_prefix('s')?.strip_suffix(".wds")?.parse().ok()
}

fn sync_dir(dir: &Path) {
    if let Ok(f) = std::fs::File::open(dir) {
        let _ = f.sync_all();
    }
}

/// Write `bytes` to `path` durably: a temporary file, synced, renamed into place, and the
/// directory synced.
fn write_durably(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
    let tmp = path.with_extension("tmp");
    {
        let mut f = std::fs::File::create(&tmp)?;
        f.write_all(bytes)?;
        f.sync_all()?;
    }
    std::fs::rename(&tmp, path)?;
    if let Some(d) = path.parent() {
        sync_dir(d);
    }
    Ok(())
}

fn status_code(s: ObjectStatus) -> u8 {
    s.code() as u8
}

/// The bytes of a group record (module documentation), and its objects' index, their
/// places counted from the record's start.
fn encode_group(track_digest: &[u8; 32], group: u64, arrived_us: u64, objects: &[(DObject, Bytes, Bytes)], erased: &[(u64, [u8; 32], u8)]) -> (Vec<u8>, Vec<DObject>) {
    let mut o = Vec::new();
    o.extend_from_slice(GROUP_MAGIC);
    o.push(VERSION);
    o.extend_from_slice(track_digest);
    o.extend_from_slice(&group.to_le_bytes());
    o.extend_from_slice(&arrived_us.to_le_bytes());
    o.extend_from_slice(&(objects.len() as u32).to_le_bytes());
    o.extend_from_slice(&(erased.len() as u32).to_le_bytes());
    let check = sha256(&[&o]);
    o.extend_from_slice(&check);
    let mut index = Vec::with_capacity(objects.len());
    for (d, props, payload) in objects {
        let mut fixed = Vec::with_capacity(OBJECT_FIXED);
        fixed.extend_from_slice(&d.id.to_le_bytes());
        fixed.extend_from_slice(&d.subgroup.unwrap_or(DATAGRAM).to_le_bytes());
        fixed.push(d.priority);
        fixed.push(status_code(d.status));
        fixed.extend_from_slice(&(props.len() as u32).to_le_bytes());
        fixed.extend_from_slice(&(payload.len() as u32).to_le_bytes());
        fixed.extend_from_slice(&d.digest);
        let check = sha256(&[&fixed, props, payload]);
        o.extend_from_slice(&fixed);
        o.extend_from_slice(&check);
        let at = o.len() as u64;
        o.extend_from_slice(props);
        o.extend_from_slice(payload);
        index.push(DObject { at, props_len: props.len() as u32, payload_len: payload.len() as u32, ..d.clone() });
    }
    for (id, digest, why) in erased {
        o.extend_from_slice(&id.to_le_bytes());
        o.extend_from_slice(digest);
        o.push(*why);
    }
    (o, index)
}

/// A group record that read: its header facts, each object with its properties and payload
/// and whether its `check` held (places counted from the record's start), and its erased
/// entries.
struct Decoded {
    group: u64,
    arrived_us: u64,
    count: u32,
    objects: Vec<(DObject, Bytes, Bytes, bool)>,
    erased: Vec<(u64, [u8; 32], u8)>,
}

/// Read a group record. `None` when its header does not check; objects past a truncation
/// are left out, and with them the erased entries.
fn decode_group(b: &[u8], track_digest: &[u8; 32]) -> Option<Decoded> {
    if b.len() < HEADER_LEN || &b[..4] != GROUP_MAGIC || b[4] != VERSION || &b[5..37] != track_digest {
        return None;
    }
    if sha256(&[&b[..HEADER_LEN - 32]])[..] != b[HEADER_LEN - 32..HEADER_LEN] {
        return None;
    }
    let u64_at = |i: usize| u64::from_le_bytes(b[i..i + 8].try_into().expect("8"));
    let u32_at = |i: usize| u32::from_le_bytes(b[i..i + 4].try_into().expect("4"));
    let (group, arrived_us, count, erased_n) = (u64_at(37), u64_at(45), u32_at(53), u32_at(57) as usize);
    let mut p = HEADER_LEN;
    let mut objects = Vec::new();
    for _ in 0..count {
        if b.len() < p + OBJECT_FIXED {
            break;
        }
        let f = &b[p..p + OBJECT_FIXED];
        let id = u64::from_le_bytes(f[..8].try_into().expect("8"));
        let sg = u64::from_le_bytes(f[8..16].try_into().expect("8"));
        let priority = f[16];
        let status = ObjectStatus::from_code(f[17] as u64).unwrap_or(ObjectStatus::Normal);
        let props_len = u32::from_le_bytes(f[18..22].try_into().expect("4"));
        let payload_len = u32::from_le_bytes(f[22..26].try_into().expect("4"));
        let digest: [u8; 32] = f[26..58].try_into().expect("32");
        let at = p + OBJECT_FIXED;
        let end = at + props_len as usize + payload_len as usize;
        if b.len() < end {
            break;
        }
        let props = Bytes::copy_from_slice(&b[at..at + props_len as usize]);
        let payload = Bytes::copy_from_slice(&b[at + props_len as usize..end]);
        let ok = sha256(&[&f[..58], &props, &payload])[..] == f[58..90];
        objects.push((DObject { id, subgroup: (sg != DATAGRAM).then_some(sg), priority, status, at: at as u64, props_len, payload_len, digest, records: Vec::new() }, props, payload, ok));
        p = end;
    }
    let mut erased = Vec::new();
    if objects.len() == count as usize {
        for _ in 0..erased_n {
            if b.len() < p + 41 {
                break;
            }
            erased.push((u64::from_le_bytes(b[p..p + 8].try_into().expect("8")), b[p + 8..p + 40].try_into().expect("32"), b[p + 40]));
            p += 41;
        }
    }
    Some(Decoded { group, arrived_us, count, objects, erased })
}

fn encode_track(key: &TrackKey, props: &Properties) -> Vec<u8> {
    let mut o = TRACK_MAGIC.to_vec();
    o.push(VERSION);
    o.push(key.0 .0.len() as u8);
    for f in &key.0 .0 {
        put_short_field(&mut o, f);
    }
    put_short_field(&mut o, &key.1 .0);
    let p = props.to_bytes().unwrap_or_default();
    o.extend_from_slice(&(p.len() as u32).to_le_bytes());
    o.extend_from_slice(&p);
    let check = sha256(&[&o]);
    o.extend_from_slice(&check);
    o
}

fn decode_track(b: &[u8]) -> Option<(TrackKey, Properties)> {
    if b.len() < 4 + 1 + 1 + 32 || &b[..4] != TRACK_MAGIC || b[4] != VERSION {
        return None;
    }
    let (body, check) = b.split_at(b.len() - 32);
    if sha256(&[body])[..] != *check {
        return None;
    }
    let mut p = 6;
    let mut field = || -> Option<Vec<u8>> {
        let n = u16::from_le_bytes(body.get(p..p + 2)?.try_into().ok()?) as usize;
        let v = body.get(p + 2..p + 2 + n)?.to_vec();
        p += 2 + n;
        Some(v)
    };
    let mut fields = Vec::new();
    for _ in 0..body[5] {
        fields.push(field()?);
    }
    let name = field()?;
    let n = u32::from_le_bytes(body.get(p..p + 4)?.try_into().ok()?) as usize;
    let props = Properties::decode(body.get(p + 4..p + 4 + n)?, "dvr track properties").ok()?;
    if p + 4 + n != body.len() {
        return None;
    }
    Some(((TrackNamespace(fields), TrackName(name)), props))
}

/// The Prior Group ID Gap an object's properties carry (MOQT §10.8): the groups before it
/// that do not exist.
/// Whether an object may be sealed (WAI confidentiality §4.3): its properties carry a Key
/// ID, or do not read (Immutable Properties that do not decode could hide one). Only an
/// object whose properties read and carry none is in the clear.
pub fn is_sealed(props: &Properties) -> bool {
    !matches!(secure_objects::key_id_of(props), Err(Refusal::KeyIdMissing))
}

fn prior_groups(props: &Properties, group: u64) -> Option<(u64, u64)> {
    let n = props.find(prop::PRIOR_GROUP_ID_GAP).ok().flatten().and_then(|k| k.as_varint())?;
    (n > 0 && n <= group).then(|| (group - n, group - 1))
}

fn read_at(path: &Path, at: u64, len: usize) -> std::io::Result<Vec<u8>> {
    let f = std::fs::File::open(path)?;
    let mut b = vec![0u8; len];
    f.read_exact_at(&mut b, at)?;
    Ok(b)
}

/// A group's record held in memory to be written again: each object with its properties
/// and payload.
type Rows = Vec<(DObject, Bytes, Bytes)>;

impl Dvr {
    /// Open the store at `cfg.path`: read every track and segment in it, check every object
    /// again, and start its writer. `verifier` gives the receipts verifier of a namespace;
    /// `authorities` are the keys of `auth.authorities`, room keys besides `cfg.room_keys` and
    /// takedown keys besides `cfg.takedown_keys`; `purge_key` signs the store's purges (none
    /// are signed without it).
    pub fn open(cfg: &DvrConfig, suffix: &str, authorities: Vec<[u8; 32]>, purge_key: Option<SigningKey>, verifier: impl Fn(&TrackNamespace) -> ReceiptVerifier) -> Result<Arc<Dvr>, String> {
        cfg.check()?;
        let root = cfg.path.clone();
        std::fs::create_dir_all(root.join("tracks")).map_err(|e| format!("dvr: {}: {e}", root.display()))?;
        let mut room_keys: Vec<[u8; 32]> = cfg.room_keys.iter().filter_map(|k| hex::decode(k).ok()?.try_into().ok()).collect();
        let mut takedown_keys: Vec<[u8; 32]> = cfg.takedown_keys.iter().filter_map(|k| hex::decode(k).ok()?.try_into().ok()).collect();
        takedown_keys.extend(authorities.iter().copied());
        room_keys.extend(authorities);
        let windows = cfg
            .window
            .iter()
            .map(|w| Window { prefix: TrackNamespace::from_strs(w.prefix.iter().map(String::as_str)), max_age_us: w.max_age_s * 1_000_000, max_bytes: w.max_bytes })
            .collect::<Vec<_>>();
        let dvr = Arc::new(Dvr {
            root,
            index: Mutex::new(Index { window_bytes: vec![0; windows.len()], ..Index::default() }),
            windows,
            suffix: suffix.as_bytes().to_vec(),
            room_keys,
            takedown_keys,
            purge_key,
            tx: Mutex::new(None),
            worker: Mutex::new(None),
            stats: DvrStats::default(),
        });
        let mut w = Writer::default();
        dvr.load(&mut w, &verifier)?;
        let (tx, rx) = mpsc::channel();
        *dvr.tx.lock().unwrap() = Some(tx);
        let me = dvr.clone();
        let worker = std::thread::Builder::new().name("joule-moq-relay-dvr".into()).spawn(move || me.run(w, rx)).map_err(|e| e.to_string())?;
        *dvr.worker.lock().unwrap() = Some(worker);
        Ok(dvr)
    }

    /// The window that keeps `ns`: the rule with the longest prefix that matches it.
    pub fn window_of(&self, ns: &TrackNamespace) -> Option<usize> {
        self.windows.iter().enumerate().filter(|(_, w)| pattern_matches(&w.prefix, ns)).max_by_key(|(_, w)| w.prefix.0.len()).map(|(i, _)| i)
    }

    /// Whether a window rule matches `ns`. Moderation reads the decisions of every events
    /// track it matches, a sealed room's among them.
    pub fn keeps(&self, ns: &TrackNamespace) -> bool {
        self.window_of(ns).is_some()
    }

    /// Whether the store writes the tracks of `ns`: a rule matches, and it is no room whose
    /// objects are sealed and no room in which a minor publishes.
    pub fn stores(&self, ns: &TrackNamespace) -> bool {
        self.keeps(ns) && !room_of(ns).is_some_and(|r| self.index.lock().unwrap().sealed.contains(&r))
    }

    /// Room `room`'s objects are sealed: from now the store keeps nothing of it, and the
    /// writer deletes what it held.
    fn seal_room(&self, room: [u8; 32]) {
        if self.index.lock().unwrap().sealed.insert(room) {
            tracing::info!(target: "joule_moq_relay::dvr", room = %hex::encode(room), "a room's objects are sealed: the store keeps nothing of it");
            self.send(Job::Forget(room));
        }
    }

    /// A minor publishes in room `room` (WAI age-assurance §8.4): from now the store keeps
    /// nothing of it, as of a room whose objects are sealed, and the writer deletes what it held.
    pub fn minors_room(&self, room: [u8; 32]) {
        if self.index.lock().unwrap().sealed.insert(room) {
            tracing::info!(target: "joule_moq_relay::dvr", room = %hex::encode(room), "a minor publishes in a room: the store keeps nothing of it");
            self.send(Job::Forget(room));
        }
    }

    fn send(&self, j: Job) {
        if let Some(tx) = self.tx.lock().unwrap().as_ref() {
            let _ = tx.send(j);
        }
    }

    /// Wait until every write asked for so far is on disk, synced.
    pub fn flush(&self) {
        let (tx, rx) = mpsc::channel();
        self.send(Job::Flush(tx));
        let _ = rx.recv_timeout(Duration::from_secs(60));
    }

    /// Flush, then stop the writer. The store is closed; a new one can open its path.
    pub fn close(&self) {
        self.flush();
        self.tx.lock().unwrap().take();
        if let Some(w) = self.worker.lock().unwrap().take() {
            let _ = w.join();
        }
    }

    /// Group `group` of `key` is whole in the cache: store it, unless it is stored with as
    /// many objects already. Cheap: the write is the writer's.
    pub fn seal(&self, key: &TrackKey, props: &Properties, group: u64, arrived_us: u64, objects: Vec<((u64, u64), CachedObject)>) {
        if objects.is_empty() || !self.stores(&key.0) {
            return;
        }
        // A sealed object is never written, and its room is never kept again
        // (confidentiality §9).
        if objects.iter().any(|(_, o)| is_sealed(&o.properties)) {
            if let Some(room) = room_of(&key.0) {
                self.seal_room(room);
            }
            return;
        }
        {
            let ix = self.index.lock().unwrap();
            if let Some(t) = ix.tracks.get(key) {
                if t.evicted_through.is_some_and(|e| group <= e) {
                    self.stats.groups_late.fetch_add(1, Ordering::Relaxed);
                    return;
                }
                if t.groups.get(&group).is_some_and(|g| g.objects.len() + g.erased.len() >= objects.len()) {
                    return;
                }
            }
        }
        self.send(Job::Seal { key: key.clone(), props: props.clone(), group, arrived_us, objects });
    }

    /// What the store holds of `key` from group `lo` to `hi`.
    pub fn snapshot(&self, key: &TrackKey, lo: u64, hi: u64) -> Option<Snapshot> {
        let ix = self.index.lock().unwrap();
        let t = ix.tracks.get(key)?;
        Some(Snapshot {
            held: t.groups.range(lo..=hi).map(|(g, _)| *g).collect(),
            absent: t.absent.iter().filter(|(l, h)| **h >= lo && **l <= hi).map(|(l, h)| (*l, *h)).collect(),
            evicted_through: t.evicted_through,
            largest: t.largest(),
            end_of_track: t.end_of_track,
            props: t.props.clone(),
        })
    }

    /// The Largest Object the store holds of `key`.
    pub fn largest(&self, key: &TrackKey) -> Option<Location> {
        self.index.lock().unwrap().tracks.get(key).and_then(DTrack::largest)
    }

    /// The track's properties as the store recorded them.
    pub fn props(&self, key: &TrackKey) -> Option<Properties> {
        self.index.lock().unwrap().tracks.get(key).map(|t| t.props.clone())
    }

    /// Whether any decision withholds something in `room`.
    pub fn withholds_in(&self, room: Option<[u8; 32]>) -> bool {
        room.is_some_and(|r| self.index.lock().unwrap().standing.iter().any(|((rr, _), v)| *rr == r && *v != events::ALLOW))
    }

    /// Whether the standing decision on `digest` in `room` withholds it.
    pub fn withheld(&self, room: Option<[u8; 32]>, digest: &[u8; 32]) -> bool {
        self.index.lock().unwrap().withheld(room, digest)
    }

    /// The objects of group `group` of `key` from `from` to `to` (Object IDs, inclusive),
    /// read from disk, in Object ID order, with what is no longer held or withheld marked.
    /// An events-track object some of whose records are withheld comes back without them
    /// (`crate::events::withhold`). With `withhold` false (a moderator's session, WAI
    /// live-moderation §6.3) nothing is withheld; what is erased stays unknown. `None` when
    /// the store no longer holds the group.
    pub fn read_group(&self, key: &TrackKey, group: u64, from: u64, to: u64, withhold: bool) -> Option<Vec<Slot>> {
        // A group written again between the index and the read (an erasure) is read again.
        for attempt in 0..2 {
            let (path, place, objs, erased, room, is_events) = {
                let ix = self.index.lock().unwrap();
                let t = ix.tracks.get(key)?;
                let g = t.groups.get(&group)?;
                (seg_path(&t.dir, g.seg), (g.seg, g.frame_at), g.objects.clone(), g.erased.clone(), t.room.filter(|_| withhold), t.events)
            };
            let Ok(f) = std::fs::File::open(&path) else {
                if attempt == 0 {
                    continue;
                }
                return None;
            };
            let mut out: Vec<(u64, Slot)> = erased.iter().filter(|e| e.0 >= from && e.0 <= to).map(|e| (e.0, Slot::Unknown(e.0))).collect();
            let mut moved = false;
            // What a takedown covers is never served, to any session (WAI takedown §10.1).
            let gone = self.index.lock().unwrap().track_gone(key, group);
            for o in objs.iter().filter(|o| o.id >= from && o.id <= to) {
                if gone || self.index.lock().unwrap().taken.contains(&o.digest) {
                    out.push((o.id, Slot::Unknown(o.id)));
                    continue;
                }
                let mut b = vec![0u8; o.props_len as usize + o.payload_len as usize];
                let payload = match f.read_exact_at(&mut b, o.at) {
                    Ok(()) => Bytes::from(b.split_off(o.props_len as usize)),
                    Err(_) => Bytes::new(),
                };
                if ObjectReceipt::hash_payload(&payload) != o.digest {
                    moved = true;
                    out.push((o.id, Slot::Unknown(o.id)));
                    continue;
                }
                let ix = self.index.lock().unwrap();
                if ix.withheld(room, &o.digest) {
                    out.push((o.id, Slot::Unknown(o.id)));
                    continue;
                }
                let payload = if is_events && o.records.iter().any(|d| ix.withheld(room, d)) {
                    match events::withhold(&payload, |d| ix.withheld(room, d)) {
                        Some(p) => Bytes::from(p.into_owned()),
                        None => {
                            out.push((o.id, Slot::Unknown(o.id)));
                            continue;
                        }
                    }
                } else {
                    payload
                };
                drop(ix);
                let properties = if b.is_empty() { Properties::new() } else { Properties::decode(&b, "dvr object properties").unwrap_or_default() };
                out.push((o.id, Slot::Object(StoredObject { id: o.id, subgroup: o.subgroup, priority: o.priority, status: o.status, properties, payload })));
            }
            if moved {
                // Written again since the index was read: read it where it stands now.
                let now = self.index.lock().unwrap().tracks.get(key).and_then(|t| t.groups.get(&group)).map(|g| (g.seg, g.frame_at));
                if attempt == 0 && now.is_some() && now != Some(place) {
                    continue;
                }
                tracing::warn!(target: "joule_moq_relay::dvr", file = %path.display(), group, "a stored object whose bytes no longer hold its digest: not served");
            }
            out.sort_by_key(|(id, _)| *id);
            return Some(out.into_iter().map(|(_, s)| s).collect());
        }
        None
    }

    /// What the store did for `room` between two instants (WAI live-metering §3.9).
    pub fn figures(&self, room: [u8; 32], from_us: u64, to_us: u64) -> RoomFigures {
        let ix = self.index.lock().unwrap();
        let Some(s) = ix.series.get(&room) else { return RoomFigures::default() };
        let (a0, o0, b0, d0) = s.at(from_us);
        let (a1, o1, b1, d1) = s.at(to_us.max(from_us));
        RoomFigures { objects_stored: o1 - o0, bytes_stored: b1 - b0, byte_s: ((a1 - a0) / 1_000_000) as u64, dropped: d1 - d0 }
    }

    pub fn counts(&self) -> DvrCounts {
        let s = &self.stats;
        let ix = self.index.lock().unwrap();
        DvrCounts {
            groups_written: s.groups_written.load(Ordering::Relaxed),
            objects_written: s.objects_written.load(Ordering::Relaxed),
            groups_evicted: s.groups_evicted.load(Ordering::Relaxed),
            groups_late: s.groups_late.load(Ordering::Relaxed),
            objects_dropped: s.objects_dropped.load(Ordering::Relaxed),
            torn_writes: s.torn_writes.load(Ordering::Relaxed),
            objects_redacted: s.objects_redacted.load(Ordering::Relaxed),
            decisions: s.decisions.load(Ordering::Relaxed),
            syncs: s.syncs.load(Ordering::Relaxed),
            groups_held: ix.tracks.values().map(|t| t.groups.len() as u64).sum(),
            bytes_held: ix.window_bytes.iter().sum(),
            segments: ix.tracks.values().map(|t| t.segs.len() as u64).sum(),
            rooms_sealed: ix.sealed.len() as u64,
            takedowns: s.takedowns.load(Ordering::Relaxed),
            objects_taken_down: s.objects_taken_down.load(Ordering::Relaxed),
            purges_signed: s.purges_signed.load(Ordering::Relaxed),
        }
    }

    /// The groups the store holds of `key`, ascending.
    pub fn groups(&self, key: &TrackKey) -> Vec<u64> {
        self.index.lock().unwrap().tracks.get(key).map(|t| t.groups.keys().copied().collect()).unwrap_or_default()
    }

    /// Where group `group` of `key` stands on disk: its segment and the place of its
    /// frame, for tests and tools.
    pub fn group_location(&self, key: &TrackKey, group: u64) -> Option<(PathBuf, u64)> {
        let ix = self.index.lock().unwrap();
        let t = ix.tracks.get(key)?;
        let g = t.groups.get(&group)?;
        Some((seg_path(&t.dir, g.seg), g.frame_at))
    }

    /// The directory of `key`'s files, for tests and tools.
    pub fn track_dir(&self, key: &TrackKey) -> Option<PathBuf> {
        self.index.lock().unwrap().tracks.get(key).map(|t| t.dir.clone())
    }

    // ── Takedowns ───────────────────────────────────────────────────────────

    /// Whether any takedown is in force.
    pub fn takes_down(&self) -> bool {
        !self.index.lock().unwrap().takedowns.is_empty()
    }

    /// Whether a takedown covers the object of content hash `digest`, or group `group` of
    /// `key` by its track.
    pub fn taken_down(&self, key: &TrackKey, group: u64, digest: &[u8; 32]) -> bool {
        let ix = self.index.lock().unwrap();
        ix.taken.contains(digest) || ix.track_gone(key, group)
    }

    /// Honour a signed takedown (WAI takedown §10.1): verify it under the keys the store
    /// trusts, record what it covers, and have the writer erase it, log it and sign the
    /// purge. Its link hash, or why it is not honoured. A takedown honoured before is not
    /// applied again.
    pub fn takedown(&self, rec: &[u8]) -> Result<[u8; 32], String> {
        self.takedown_under(rec, &self.takedown_keys)
    }

    fn takedown_under(&self, rec: &[u8], trusted: &[[u8; 32]]) -> Result<[u8; 32], String> {
        let t = crate::takedown::read(rec, trusted)?;
        if !self.index.lock().unwrap().take(&t) {
            return Ok(t.link);
        }
        self.stats.takedowns.fetch_add(1, Ordering::Relaxed);
        tracing::info!(target: "joule_moq_relay::dvr", takedown = %hex::encode(t.link), objects = t.objects.len(), tracks = t.tracks.len(), "a takedown is honoured");
        let link = t.link;
        self.send(Job::TakeDown { t, rec: rec.to_vec() });
        Ok(link)
    }

    /// One object of a takedown track `[K, takedown]` the relay forwards: a takedown signed by
    /// `K`, honoured when `K` is a key the store trusts.
    pub fn takedown_object(&self, ns: &TrackNamespace, payload: &[u8]) {
        let Some(k) = room_of(ns).filter(|_| ns.0.len() == 2 && ns.0[1] == crate::takedown::TRACK_FIELD) else { return };
        if !self.takedown_keys.contains(&k) {
            return;
        }
        if let Err(e) = self.takedown_under(payload, &[k]) {
            tracing::warn!(target: "joule_moq_relay::dvr", error = %e, "a takedown is not honoured");
        }
    }

    /// Where the purge of the takedown `link` is written.
    pub fn purge_path(&self, link: &[u8; 32]) -> PathBuf {
        self.root.join("purges").join(format!("{}.json", hex::encode(link)))
    }

    /// The writer's part of a takedown: log it, erase what it covers, sync, sign the purge.
    fn apply_takedown(&self, w: &mut Writer, t: &crate::takedown::Takedown, rec: Option<&[u8]>) {
        if let Some(rec) = rec {
            let mut b = (rec.len() as u32).to_le_bytes().to_vec();
            b.extend_from_slice(rec);
            b.extend_from_slice(&sha256(&[rec]));
            let r = std::fs::OpenOptions::new().create(true).append(true).open(self.root.join("takedowns.wdl")).and_then(|mut f| {
                f.write_all(&b)?;
                f.sync_all()
            });
            if let Err(e) = r {
                tracing::error!(target: "joule_moq_relay::dvr", error = %e, "the takedowns log could not be written");
            }
        }
        for d in &t.objects {
            self.erase(w, None, *d, ERASED_TAKEN_DOWN);
        }
        for (track, from) in &t.tracks {
            self.erase_track(w, track, *from);
        }
        self.sync(w);
        self.write_purge(t);
    }

    /// Sign and write the purge of `t`, once.
    fn write_purge(&self, t: &crate::takedown::Takedown) {
        let Some(key) = &self.purge_key else { return };
        let path = self.purge_path(&t.link);
        if path.exists() {
            return;
        }
        let body = crate::takedown::purge(t, now_us() / 1_000_000, key);
        let r = std::fs::create_dir_all(self.root.join("purges")).and_then(|_| write_durably(&path, body.as_bytes()));
        match r {
            Ok(()) => {
                self.stats.purges_signed.fetch_add(1, Ordering::Relaxed);
                tracing::info!(target: "joule_moq_relay::dvr", takedown = %hex::encode(t.link), path = %path.display(), "a purge is signed");
            }
            Err(e) => tracing::error!(target: "joule_moq_relay::dvr", error = %e, "a purge could not be written"),
        }
    }

    /// Erase every object of each group, from `from` on, of every track whose digest is
    /// `track` (WAI live-rooms §6.2).
    fn erase_track(&self, w: &mut Writer, track: &[u8; 32], from: u64) {
        let groups: Vec<(TrackKey, PathBuf, u64)> = {
            let ix = self.index.lock().unwrap();
            ix.tracks
                .iter()
                .filter(|(k, _)| room_track_digest(k) == *track)
                .flat_map(|(k, t)| t.groups.range(from..).map(move |(g, _)| (k.clone(), t.dir.clone(), *g)))
                .collect()
        };
        for (key, dir, g) in groups {
            let Some(grp) = self.index.lock().unwrap().tracks.get(&key).and_then(|t| t.groups.get(&g)).cloned() else { continue };
            if grp.objects.is_empty() {
                continue;
            }
            let mut erased = grp.erased.clone();
            erased.extend(grp.objects.iter().map(|o| (o.id, o.digest, ERASED_TAKEN_DOWN)));
            if self.rewrite(w, &key, &dir, g, grp.arrived_us, &Vec::new(), &erased, (grp.seg, grp.frame_at, grp.record_len)).is_none() {
                continue;
            }
            let room = {
                let mut ix = self.index.lock().unwrap();
                for o in &grp.objects {
                    ix.unindex(&key, g, o);
                }
                let (window, room) = ix.tracks.get(&key).map_or((None, None), |t| (Some(t.window), t.room));
                if let Some(gr) = ix.tracks.get_mut(&key).and_then(|t| t.groups.get_mut(&g)) {
                    gr.bytes = 0;
                }
                if let Some(wi) = window {
                    ix.window_bytes[wi] = ix.window_bytes[wi].saturating_sub(grp.bytes);
                }
                room
            };
            self.series_change(room, now_us(), -(grp.bytes as i128), 0, 0, 0);
            self.stats.objects_taken_down.fetch_add(grp.objects.len() as u64, Ordering::Relaxed);
            tracing::info!(target: "joule_moq_relay::dvr", dir = %dir.display(), group = g, objects = grp.objects.len(), "a takedown erased a group of a track");
        }
    }

    // ── Moderation ──────────────────────────────────────────────────────────

    /// One object of an events track `ns` the relay forwards: each `decision` record of it
    /// is honoured when it verifies, its author owns the track, and the author is a
    /// moderator (`moderators`: the principals of grants with the moderate power the
    /// publishing session proved, in this room).
    pub fn events_object(&self, ns: &TrackNamespace, payload: &[u8], moderators: &[[u8; 32]]) {
        let (Some(room), Some(owner)) = (room_of(ns), key_field(ns, 1)) else { return };
        if !(ns.0.len() == 3 && ns.0[2] == b"events") {
            return;
        }
        let Some(recs) = events::records(payload) else { return };
        for (_, rec) in recs {
            let Some(e) = Event::read(&rec) else { continue };
            if e.decision().is_none() || e.room != room || e.principal != owner {
                continue;
            }
            if self.index.lock().unwrap().decided.contains(&e.digest) {
                continue;
            }
            if !(self.room_keys.contains(&owner) || moderators.contains(&owner)) || !e.verifies(&rec) {
                continue;
            }
            self.apply_decision(&e);
            if !self.index.lock().unwrap().sealed.contains(&room) {
                self.send(Job::Log(rec));
            }
        }
    }

    /// Returns the target to erase, for a redaction.
    fn apply_decision(&self, e: &Event) -> Option<[u8; 32]> {
        let (verdict, target) = e.decision()?;
        let mut ix = self.index.lock().unwrap();
        if !ix.decided.insert(e.digest) {
            return None;
        }
        ix.standing.insert((e.room, target), verdict);
        drop(ix);
        self.stats.decisions.fetch_add(1, Ordering::Relaxed);
        tracing::info!(target: "joule_moq_relay::dvr", room = %hex::encode(e.room), target_digest = %hex::encode(target), verdict, "a decision is honoured");
        if verdict == REDACT {
            self.send(Job::Erase { room: e.room, digest: target });
            return Some(target);
        }
        None
    }

    // ── Segments ────────────────────────────────────────────────────────────

    /// Append `record` to `key`'s open segment (beginning another past its span or size),
    /// returning the segment and the frame's place.
    fn append(&self, w: &mut Writer, key: &TrackKey, dir: &Path, record: &[u8]) -> std::io::Result<(u64, u64)> {
        let now = now_us();
        let stale = w.open.get(key).is_none_or(|s| now.saturating_sub(s.started_us) >= SEGMENT_SPAN_US || s.len >= SEGMENT_MAX);
        if stale {
            if let Some(old) = w.open.remove(key) {
                self.retire(key, old.seq, None);
            }
            let seq = *w.next.get(key).unwrap_or(&0);
            w.next.insert(key.clone(), seq + 1);
            let path = seg_path(dir, seq);
            let file = std::fs::OpenOptions::new().create(true).truncate(true).read(true).write(true).open(&path)?;
            w.dirs.insert(dir.to_path_buf());
            w.open.insert(key.clone(), OpenSeg { seq, file, started_us: now, len: 0 });
            if let Some(t) = self.index.lock().unwrap().tracks.get_mut(key) {
                t.segs.entry(seq).or_insert(0);
            }
        }
        let s = w.open.get_mut(key).expect("opened above");
        let mut frame = Vec::with_capacity(FRAME as usize + record.len());
        frame.extend_from_slice(&(record.len() as u32).to_le_bytes());
        frame.push(1);
        frame.extend_from_slice(record);
        s.file.write_all_at(&frame, s.len)?;
        let at = s.len;
        s.len += frame.len() as u64;
        w.dirty.insert(seg_path(dir, s.seq));
        Ok((s.seq, at))
    }

    /// Mark a record replaced: its frame's state 0 and its bytes zeros.
    fn kill(&self, w: &mut Writer, dir: &Path, seq: u64, frame_at: u64, record_len: u32) {
        let path = seg_path(dir, seq);
        let r = std::fs::OpenOptions::new().write(true).open(&path).and_then(|f| {
            f.write_all_at(&[0], frame_at + 4)?;
            f.write_all_at(&vec![0u8; record_len as usize], frame_at + FRAME)
        });
        if r.is_ok() {
            w.dirty.insert(path);
        }
    }

    /// Delete segment `seq` of `key` when no group's standing record is in it and it is not
    /// the one being appended to (`open`, the writer's).
    fn retire(&self, key: &TrackKey, seq: u64, open: Option<u64>) {
        if open == Some(seq) {
            return;
        }
        let path = {
            let mut ix = self.index.lock().unwrap();
            let Some(t) = ix.tracks.get_mut(key) else { return };
            if t.segs.get(&seq).is_some_and(|n| *n > 0) {
                return;
            }
            t.segs.remove(&seq);
            seg_path(&t.dir, seq)
        };
        let _ = std::fs::remove_file(&path);
    }

    /// Sync every segment written since the last sync, and the directories files were made
    /// in.
    fn sync(&self, w: &mut Writer) {
        for p in w.dirty.drain() {
            if let Ok(f) = std::fs::OpenOptions::new().write(true).open(&p) {
                let _ = f.sync_data();
            }
            self.stats.syncs.fetch_add(1, Ordering::Relaxed);
        }
        for d in w.dirs.drain() {
            sync_dir(&d);
        }
        w.last_sync = Some(Instant::now());
    }

    /// Write group `g` of `key` again from `rows` and `erased`: append the new record, kill
    /// the old, and update the index. Returns the group's new objects.
    #[allow(clippy::too_many_arguments)]
    fn rewrite(&self, w: &mut Writer, key: &TrackKey, dir: &Path, g: u64, arrived_us: u64, rows: &Rows, erased: &[(u64, [u8; 32], u8)], old: (u64, u64, u32)) -> Option<Vec<DObject>> {
        let (bytes, mut index) = encode_group(&track_digest(key), g, arrived_us, rows, erased);
        let (seq, at) = match self.append(w, key, dir, &bytes) {
            Ok(x) => x,
            Err(e) => {
                tracing::error!(target: "joule_moq_relay::dvr", error = %e, "a group could not be written again");
                return None;
            }
        };
        for (o, r) in index.iter_mut().zip(rows) {
            o.at += at + FRAME;
            o.records = r.0.records.clone();
        }
        {
            let mut ix = self.index.lock().unwrap();
            if let Some(t) = ix.tracks.get_mut(key) {
                *t.segs.entry(seq).or_insert(0) += 1;
                if let Some(n) = t.segs.get_mut(&old.0) {
                    *n = n.saturating_sub(1);
                }
                if let Some(gr) = t.groups.get_mut(&g) {
                    gr.objects = index.clone();
                    gr.erased = erased.to_vec();
                    gr.seg = seq;
                    gr.frame_at = at;
                    gr.record_len = bytes.len() as u32;
                }
            }
        }
        self.kill(w, dir, old.0, old.1, old.2);
        let open = w.open.get(key).map(|s| s.seq);
        self.retire(key, old.0, open);
        Some(index)
    }

    // ── Loading ─────────────────────────────────────────────────────────────

    fn load(&self, w: &mut Writer, verifier: &dyn Fn(&TrackNamespace) -> ReceiptVerifier) -> Result<(), String> {
        let now = now_us();
        // The rooms found sealed before. Nothing of them is loaded: their decisions and the
        // tracks a crash left of them are deleted.
        let path = self.root.join("sealed.wdl");
        if let Ok(b) = std::fs::read(&path) {
            let mut good = 0;
            for c in b.as_chunks::<64>().0 {
                if sha256(&[&c[..32]])[..] != c[32..] {
                    break;
                }
                self.index.lock().unwrap().sealed.insert(c[..32].try_into().expect("32"));
                good += 64;
            }
            if good != b.len() {
                let _ = std::fs::OpenOptions::new().write(true).open(&path).and_then(|f| f.set_len(good as u64));
            }
        }
        let sealed = self.index.lock().unwrap().sealed.clone();
        if !sealed.is_empty() {
            self.rewrite_decisions(|room| !sealed.contains(room));
        }
        // The decisions honoured before, each verified again.
        let log = self.root.join("decisions.wdl");
        if let Ok(b) = std::fs::read(&log) {
            let mut p = 0;
            let mut good = 0;
            while b.len() >= p + 4 {
                let n = u32::from_le_bytes(b[p..p + 4].try_into().expect("4")) as usize;
                let Some(rec) = b.get(p + 4..p + 4 + n) else { break };
                let Some(check) = b.get(p + 4 + n..p + 4 + n + 32) else { break };
                if sha256(&[rec])[..] != *check {
                    break;
                }
                if let Some(e) = Event::read(rec).filter(|e| e.verifies(rec)) {
                    // Erasures are made below, once the groups are read.
                    let (verdict, target) = e.decision().unwrap_or((0, [0; 32]));
                    let mut ix = self.index.lock().unwrap();
                    if verdict != 0 && ix.decided.insert(e.digest) {
                        ix.standing.insert((e.room, target), verdict);
                        self.stats.decisions.fetch_add(1, Ordering::Relaxed);
                    }
                }
                p += 4 + n + 32;
                good = p;
            }
            if good != b.len() {
                // A record a crash cut short: the log keeps what precedes it.
                let _ = std::fs::OpenOptions::new().write(true).open(&log).and_then(|f| f.set_len(good as u64));
            }
        }
        // The takedowns honoured before, each verified again (WAI takedown §10.1).
        let mut takedowns: Vec<crate::takedown::Takedown> = Vec::new();
        let tlog = self.root.join("takedowns.wdl");
        if let Ok(b) = std::fs::read(&tlog) {
            let (mut p, mut good) = (0, 0);
            while b.len() >= p + 4 {
                let n = u32::from_le_bytes(b[p..p + 4].try_into().expect("4")) as usize;
                let Some(rec) = b.get(p + 4..p + 4 + n) else { break };
                let Some(check) = b.get(p + 4 + n..p + 4 + n + 32) else { break };
                if sha256(&[rec])[..] != *check {
                    break;
                }
                match crate::takedown::read(rec, &self.takedown_keys) {
                    Ok(t) => {
                        if self.index.lock().unwrap().take(&t) {
                            self.stats.takedowns.fetch_add(1, Ordering::Relaxed);
                            takedowns.push(t);
                        }
                    }
                    Err(e) => tracing::warn!(target: "joule_moq_relay::dvr", error = %e, "a logged takedown no longer verifies: not honoured"),
                }
                p += 4 + n + 32;
                good = p;
            }
            if good != b.len() {
                let _ = std::fs::OpenOptions::new().write(true).open(&tlog).and_then(|f| f.set_len(good as u64));
            }
        }
        let tracks_dir = self.root.join("tracks");
        let mut loaded: Vec<(TrackKey, DTrack, HashMap<u64, Rows>)> = Vec::new();
        // Groups whose standing record no longer says what the index holds: written again.
        let mut changed: HashSet<(TrackKey, u64)> = HashSet::new();
        for d in std::fs::read_dir(&tracks_dir).map_err(|e| e.to_string())?.flatten() {
            let dir = d.path();
            let Some((key, props)) = std::fs::read(dir.join("TRACK")).ok().and_then(|b| decode_track(&b)) else {
                tracing::warn!(target: "joule_moq_relay::dvr", dir = %dir.display(), "a track directory whose TRACK does not read: left alone");
                continue;
            };
            if room_of(&key.0).is_some_and(|r| sealed.contains(&r)) {
                let _ = std::fs::remove_dir_all(&dir);
                continue;
            }
            let Some(window) = self.window_of(&key.0) else { continue };
            let digest = track_digest(&key);
            let mut t = DTrack { dir: dir.clone(), window, room: room_of(&key.0), events: is_events_track(&key), props, groups: BTreeMap::new(), absent: BTreeMap::new(), evicted_through: None, end_of_track: None, segs: BTreeMap::new() };
            let mut segs: Vec<(u64, PathBuf)> = std::fs::read_dir(&dir).map_err(|e| e.to_string())?.flatten().filter_map(|e| seg_seq(&e.path()).map(|s| (s, e.path()))).collect();
            segs.sort();
            // Each group's standing record: the last that reads; the earlier ones killed.
            let mut rows: HashMap<u64, Rows> = HashMap::new();
            let mut replaced: Vec<(u64, u64, u32)> = Vec::new();
            for (seq, path) in &segs {
                let Ok(b) = std::fs::read(path) else { continue };
                t.segs.insert(*seq, 0);
                let mut p = 0usize;
                while p < b.len() {
                    let whole = b.len() >= p + FRAME as usize && {
                        let n = u32::from_le_bytes(b[p..p + 4].try_into().expect("4")) as usize;
                        b.len() >= p + FRAME as usize + n
                    };
                    if !whole {
                        // A frame a crash cut short: cut off.
                        self.stats.torn_writes.fetch_add(1, Ordering::Relaxed);
                        let _ = std::fs::OpenOptions::new().write(true).open(path).and_then(|f| f.set_len(p as u64));
                        break;
                    }
                    let n = u32::from_le_bytes(b[p..p + 4].try_into().expect("4")) as usize;
                    let state = b[p + 4];
                    let rec = &b[p + FRAME as usize..p + FRAME as usize + n];
                    let frame_at = p as u64;
                    p += FRAME as usize + n;
                    if state != 1 {
                        continue;
                    }
                    let Some(dg) = decode_group(rec, &digest) else {
                        self.stats.objects_dropped.fetch_add(1, Ordering::Relaxed);
                        self.series_change(t.room, now, 0, 0, 0, 1);
                        tracing::warn!(target: "joule_moq_relay::dvr", file = %path.display(), at = frame_at, "a group record whose header does not check: dropped");
                        replaced.push((*seq, frame_at, n as u32));
                        continue;
                    };
                    if let Some(prev) = t.groups.remove(&dg.group) {
                        replaced.push((prev.seg, prev.frame_at, prev.record_len));
                        if let Some(c) = t.segs.get_mut(&prev.seg) {
                            *c = c.saturating_sub(1);
                        }
                    }
                    let base = frame_at + FRAME;
                    let short = dg.count as usize - dg.objects.len();
                    let mut grp = DGroup { arrived_us: dg.arrived_us, bytes: 0, objects: Vec::new(), erased: dg.erased, seg: *seq, frame_at, record_len: n as u32 };
                    let mut kept: Rows = Vec::new();
                    if short > 0 {
                        changed.insert((key.clone(), dg.group));
                    }
                    for (mut o, props, payload, ok) in dg.objects {
                        o.at += base;
                        if !ok || ObjectReceipt::hash_payload(&payload) != o.digest {
                            self.dropped(&t, dg.group, &o, &mut grp, "its check does not hold");
                            changed.insert((key.clone(), dg.group));
                            continue;
                        }
                        if t.events {
                            o.records = events::digests(&payload);
                        }
                        kept.push((o, props, payload));
                    }
                    if short > 0 {
                        self.stats.objects_dropped.fetch_add(short as u64, Ordering::Relaxed);
                        self.series_change(t.room, now, 0, 0, 0, short as u64);
                    }
                    for (o, props, payload) in &kept {
                        if let Some(r) = prior_groups(&Properties::decode(props, "dvr").unwrap_or_default(), dg.group) {
                            t.absent.insert(r.0, r.1);
                        }
                        if o.status == ObjectStatus::EndOfTrack {
                            t.end_of_track = Some(Location::new(dg.group, o.id));
                        }
                        grp.bytes += payload.len() as u64;
                        grp.objects.push(o.clone());
                    }
                    *t.segs.entry(*seq).or_insert(0) += 1;
                    t.groups.insert(dg.group, grp);
                    rows.insert(dg.group, kept);
                }
            }
            for (seq, at, len) in replaced {
                self.kill(w, &dir, seq, at, len);
            }
            w.next.insert(key.clone(), segs.last().map_or(0, |(s, _)| s + 1));
            loaded.push((key, t, rows));
        }
        // Every object of a track with a receipts track, against its group's receipts; every
        // record of an events track, under its principal.
        let held: HashSet<TrackKey> = loaded.iter().map(|(k, _, _)| k.clone()).collect();
        let receipts: HashMap<TrackKey, HashMap<u64, Bytes>> = loaded
            .iter()
            .filter(|(k, _, _)| self.media_of(&k.1).is_some())
            .map(|(k, _, rows)| (k.clone(), rows.iter().filter_map(|(g, r)| r.first().map(|(_, _, p)| (*g, p.clone()))).collect()))
            .collect();
        for (k, t, rows) in &mut loaded {
            let rkey = (k.0.clone(), self.receipts_of(&k.1));
            let check_receipts = self.media_of(&k.1).is_none() && held.contains(&rkey);
            if !(check_receipts || t.events) {
                continue;
            }
            let v = verifier(&k.0);
            let groups: Vec<u64> = t.groups.keys().copied().collect();
            for g in groups {
                let bundle = if check_receipts { receipts.get(&rkey).and_then(|m| m.get(&g)).and_then(|b| ReceiptBundle::from_payload(b).ok()) } else { None };
                let vg = bundle.as_ref().and_then(|b| (b.group.group_id == GroupId(g)).then(|| v.verify_group(&b.group).ok()).flatten());
                let Some(r) = rows.get_mut(&g) else { continue };
                let mut bad = Vec::new();
                for (o, _, payload) in r.iter() {
                    let ok = if check_receipts {
                        match (&bundle, &vg) {
                            (Some(b), Some(vg)) => b.objects.iter().find(|x| x.object_id == ObjectId(o.id)).is_some_and(|x| v.verify_payload(payload, x, vg).is_ok()),
                            _ => o.status != ObjectStatus::Normal,
                        }
                    } else {
                        match events::records(payload) {
                            Some(rs) => rs.iter().all(|(_, x)| Event::read(x).is_some_and(|e| e.verifies(x))),
                            None => true,
                        }
                    };
                    if !ok {
                        bad.push(o.clone());
                    }
                }
                let mut grp = t.groups.remove(&g).expect("held");
                if !bad.is_empty() {
                    changed.insert((k.clone(), g));
                }
                for o in &bad {
                    self.dropped(t, g, o, &mut grp, if check_receipts { "it does not verify against its receipt" } else { "a record of it does not verify" });
                    r.retain(|x| x.0.id != o.id);
                }
                t.groups.insert(g, grp);
            }
        }
        // Into the index; then every group whose record no longer says what the index holds
        // (an object dropped, or a redaction the log asks for that a crash left undone) is
        // written again.
        let mut again: Vec<(TrackKey, PathBuf, u64)> = Vec::new();
        {
            let mut ix = self.index.lock().unwrap();
            for (k, mut t, _) in std::mem::take(&mut loaded) {
                for (g, grp) in t.groups.iter_mut() {
                    // What a takedown covers, erased (a crash can leave it held).
                    let gone = ix.track_gone(&k, *g);
                    let taken: Vec<DObject> = grp.objects.iter().filter(|o| gone || ix.taken.contains(&o.digest)).cloned().collect();
                    if !taken.is_empty() {
                        changed.insert((k.clone(), *g));
                    }
                    for o in &taken {
                        grp.objects.retain(|x| x.id != o.id);
                        grp.bytes = grp.bytes.saturating_sub(o.payload_len as u64);
                        grp.erased.push((o.id, o.digest, ERASED_TAKEN_DOWN));
                        self.stats.objects_taken_down.fetch_add(1, Ordering::Relaxed);
                    }
                    let redacted: Vec<DObject> = grp.objects.iter().filter(|o| t.room.is_some_and(|r| ix.standing.get(&(r, o.digest)) == Some(&REDACT))).cloned().collect();
                    if !redacted.is_empty() {
                        changed.insert((k.clone(), *g));
                    }
                    for o in &redacted {
                        grp.objects.retain(|x| x.id != o.id);
                        grp.bytes = grp.bytes.saturating_sub(o.payload_len as u64);
                        grp.erased.push((o.id, o.digest, ERASED_REDACTED));
                        self.stats.objects_redacted.fetch_add(1, Ordering::Relaxed);
                    }
                    ix.window_bytes[t.window] += grp.bytes;
                    if let Some(room) = t.room {
                        ix.series.entry(room).or_default().change(now, grp.bytes as i128, 0, 0, 0);
                    }
                    for o in &grp.objects {
                        ix.reindex(&k, *g, o);
                    }
                    if changed.contains(&(k.clone(), *g)) {
                        again.push((k.clone(), t.dir.clone(), *g));
                    }
                }
                // What is below the first group held is no longer known to the store.
                t.evicted_through = t.groups.keys().next().and_then(|g| g.checked_sub(1));
                ix.tracks.insert(k, t);
            }
        }
        for (k, dir, g) in again {
            self.write_again_if_changed(w, &k, &dir, g);
        }
        // Event records a redaction the log holds targets, erased from the objects that
        // carry them.
        let redactions: Vec<([u8; 32], [u8; 32])> = self.index.lock().unwrap().standing.iter().filter(|(_, v)| **v == REDACT).map(|(k, _)| *k).collect();
        for (room, target) in redactions {
            self.erase(w, Some(room), target, ERASED_REDACTED);
        }
        // The takedowns the log holds: what a crash left of their erasures is made now, and a
        // purge a crash left unsigned is signed.
        for t in &takedowns {
            self.apply_takedown(w, t, None);
        }
        self.sync(w);
        let n: usize = self.index.lock().unwrap().tracks.values().map(|t| t.groups.len()).sum();
        tracing::info!(target: "joule_moq_relay::dvr", path = %self.root.display(), groups = n, dropped = self.stats.objects_dropped.load(Ordering::Relaxed), torn = self.stats.torn_writes.load(Ordering::Relaxed), "the DVR store is open");
        Ok(())
    }

    /// Write group `g` again when its standing record holds other objects than the index
    /// does (some dropped, or erased).
    fn write_again_if_changed(&self, w: &mut Writer, key: &TrackKey, dir: &Path, g: u64) {
        let Some(grp) = self.index.lock().unwrap().tracks.get(key).and_then(|t| t.groups.get(&g)).cloned() else { return };
        let path = seg_path(dir, grp.seg);
        let Ok(rec) = read_at(&path, grp.frame_at + FRAME, grp.record_len as usize) else { return };
        let Some(dg) = decode_group(&rec, &track_digest(key)) else { return };
        if dg.count as usize == grp.objects.len() && dg.erased.len() == grp.erased.len() {
            return;
        }
        let rows: Rows = grp.objects.iter().filter_map(|o| dg.objects.iter().find(|x| x.0.id == o.id).map(|(_, props, payload, _)| (o.clone(), props.clone(), payload.clone()))).collect();
        self.rewrite(w, key, dir, g, grp.arrived_us, &rows, &grp.erased, (grp.seg, grp.frame_at, grp.record_len));
    }

    /// An object that did not verify at load: out of the group, among its erased, counted.
    fn dropped(&self, t: &DTrack, g: u64, o: &DObject, grp: &mut DGroup, why: &str) {
        self.stats.objects_dropped.fetch_add(1, Ordering::Relaxed);
        self.series_change(t.room, now_us(), 0, 0, 0, 1);
        tracing::warn!(target: "joule_moq_relay::dvr", dir = %t.dir.display(), group = g, object = o.id, reason = why, "a stored object dropped: never served");
        if grp.objects.iter().any(|x| x.id == o.id) {
            grp.bytes = grp.bytes.saturating_sub(o.payload_len as u64);
        }
        grp.objects.retain(|x| x.id != o.id);
        if !grp.erased.iter().any(|e| e.0 == o.id) {
            grp.erased.push((o.id, o.digest, ERASED_UNVERIFIED));
        }
    }

    fn series_change(&self, room: Option<[u8; 32]>, now: u64, held: i128, objects: u64, bytes: u64, dropped: u64) {
        if let Some(r) = room {
            self.index.lock().unwrap().series.entry(r).or_default().change(now, held, objects, bytes, dropped);
        }
    }

    fn media_of(&self, name: &TrackName) -> Option<TrackName> {
        (!self.suffix.is_empty() && name.0.len() > self.suffix.len() && name.0.ends_with(&self.suffix)).then(|| TrackName(name.0[..name.0.len() - self.suffix.len()].to_vec()))
    }

    fn receipts_of(&self, name: &TrackName) -> TrackName {
        let mut v = name.0.clone();
        v.extend_from_slice(&self.suffix);
        TrackName(v)
    }

    // ── The writer ──────────────────────────────────────────────────────────

    fn run(self: Arc<Self>, mut w: Writer, rx: mpsc::Receiver<Job>) {
        loop {
            let job = rx.recv_timeout(SYNC_EVERY);
            match job {
                Ok(Job::Seal { key, props, group, arrived_us, objects }) => self.write_seal(&mut w, key, props, group, arrived_us, objects),
                Ok(Job::Erase { room, digest }) => self.erase(&mut w, Some(room), digest, ERASED_REDACTED),
                Ok(Job::TakeDown { t, rec }) => self.apply_takedown(&mut w, &t, Some(&rec)),
                Ok(Job::Log(rec)) => self.log(&rec),
                Ok(Job::Forget(room)) => self.forget(&mut w, room),
                Ok(Job::Flush(done)) => {
                    self.evict_by_age(&mut w, now_us());
                    self.sync(&mut w);
                    let _ = done.send(());
                }
                Err(mpsc::RecvTimeoutError::Timeout) => {}
                Err(mpsc::RecvTimeoutError::Disconnected) => {
                    self.sync(&mut w);
                    return;
                }
            }
            if w.last_sync.is_none_or(|t| t.elapsed() >= SYNC_EVERY) {
                self.evict_by_age(&mut w, now_us());
                self.sync(&mut w);
            }
        }
    }

    fn log(&self, rec: &[u8]) {
        // A room found sealed after the record was queued: its decisions are not kept.
        if Event::read(rec).is_some_and(|e| self.index.lock().unwrap().sealed.contains(&e.room)) {
            return;
        }
        let path = self.root.join("decisions.wdl");
        let mut b = (rec.len() as u32).to_le_bytes().to_vec();
        b.extend_from_slice(rec);
        b.extend_from_slice(&sha256(&[rec]));
        let r = std::fs::OpenOptions::new().create(true).append(true).open(&path).and_then(|mut f| {
            f.write_all(&b)?;
            f.sync_all()
        });
        if let Err(e) = r {
            tracing::error!(target: "joule_moq_relay::dvr", error = %e, "the decisions log could not be written");
        }
    }

    /// Forget room `room`, whose objects are sealed: record it in `sealed.wdl`, delete every
    /// track of it the store holds, and drop its decisions from `decisions.wdl`.
    fn forget(&self, w: &mut Writer, room: [u8; 32]) {
        let mut rec = room.to_vec();
        rec.extend_from_slice(&sha256(&[&room]));
        let r = std::fs::OpenOptions::new().create(true).append(true).open(self.root.join("sealed.wdl")).and_then(|mut f| {
            f.write_all(&rec)?;
            f.sync_all()
        });
        if let Err(e) = r {
            tracing::error!(target: "joule_moq_relay::dvr", error = %e, "a sealed room could not be recorded");
        }
        sync_dir(&self.root);
        let keys: Vec<TrackKey> = self.index.lock().unwrap().tracks.iter().filter(|(_, t)| t.room == Some(room)).map(|(k, _)| k.clone()).collect();
        for k in &keys {
            self.drop_track(w, k);
        }
        self.rewrite_decisions(|r| *r != room);
        tracing::info!(target: "joule_moq_relay::dvr", room = %hex::encode(room), tracks = keys.len(), "a sealed room is forgotten");
    }

    /// Delete track `key` from the store: its groups, its directory, its writer state.
    fn drop_track(&self, w: &mut Writer, key: &TrackKey) {
        let t = {
            let mut ix = self.index.lock().unwrap();
            let Some(t) = ix.tracks.remove(key) else { return };
            let held: u64 = t.groups.values().map(|g| g.bytes).sum();
            ix.window_bytes[t.window] = ix.window_bytes[t.window].saturating_sub(held);
            for (g, grp) in &t.groups {
                for o in &grp.objects {
                    ix.unindex(key, *g, o);
                }
            }
            if let Some(r) = t.room {
                ix.series.entry(r).or_default().change(now_us(), -(held as i128), 0, 0, 0);
            }
            t
        };
        w.open.remove(key);
        w.next.remove(key);
        w.dirty.retain(|p| !p.starts_with(&t.dir));
        if let Err(e) = std::fs::remove_dir_all(&t.dir) {
            tracing::error!(target: "joule_moq_relay::dvr", error = %e, dir = %t.dir.display(), "a sealed room's track could not be deleted");
        }
        w.dirs.insert(self.root.join("tracks"));
    }

    /// Write `decisions.wdl` again with only the records whose room `keep` keeps; a record
    /// that does not read as an event stays. Nothing is written when every record stays.
    fn rewrite_decisions(&self, keep: impl Fn(&[u8; 32]) -> bool) {
        let path = self.root.join("decisions.wdl");
        let Ok(b) = std::fs::read(&path) else { return };
        let (mut out, mut p, mut dropped) = (Vec::with_capacity(b.len()), 0, false);
        while b.len() >= p + 4 {
            let n = u32::from_le_bytes(b[p..p + 4].try_into().expect("4")) as usize;
            let Some(frame) = b.get(p..p + 4 + n + 32) else { break };
            if Event::read(&frame[4..4 + n]).is_some_and(|e| !keep(&e.room)) {
                dropped = true;
            } else {
                out.extend_from_slice(frame);
            }
            p += 4 + n + 32;
        }
        if dropped && let Err(e) = write_durably(&path, &out) {
            tracing::error!(target: "joule_moq_relay::dvr", error = %e, "the decisions log could not be written again");
        }
    }

    fn write_seal(&self, w: &mut Writer, key: TrackKey, props: Properties, group: u64, arrived_us: u64, objects: Vec<((u64, u64), CachedObject)>) {
        let Some(window) = self.window_of(&key.0) else { return };
        let room = room_of(&key.0);
        // A seal queued before its room was found sealed. The room is marked before its
        // forgetting is queued, and this thread takes jobs in order: a seal it takes after the
        // mark is refused here, and one it took before is deleted by the forgetting.
        if room.is_some_and(|r| self.index.lock().unwrap().sealed.contains(&r)) {
            return;
        }
        let events = is_events_track(&key);
        let dir = self.root.join("tracks").join(track_dir_name(&key));
        let known = self.index.lock().unwrap().tracks.contains_key(&key);
        if !known {
            if let Err(e) = std::fs::create_dir_all(&dir).and_then(|_| write_durably(&dir.join("TRACK"), &encode_track(&key, &props))) {
                tracing::error!(target: "joule_moq_relay::dvr", error = %e, "a track could not be stored");
                return;
            }
            w.dirs.insert(self.root.join("tracks"));
            let mut ix = self.index.lock().unwrap();
            ix.tracks.entry(key.clone()).or_insert_with(|| DTrack { dir: dir.clone(), window, room, events, props: props.clone(), groups: BTreeMap::new(), absent: BTreeMap::new(), evicted_through: None, end_of_track: None, segs: BTreeMap::new() });
        }
        // A seal checked against the window when it was queued can reach the writer after
        // the window has moved past its group: it is late here too, and never written below
        // the evicted mark (where it would leave the window two runs).
        if self.index.lock().unwrap().tracks.get(&key).is_some_and(|t| t.evicted_through.is_some_and(|e| group <= e)) {
            self.stats.groups_late.fetch_add(1, Ordering::Relaxed);
            return;
        }
        let mut rows: Rows = Vec::with_capacity(objects.len());
        let mut erased = Vec::new();
        {
            let ix = self.index.lock().unwrap();
            let gone = ix.track_gone(&key, group);
            for ((_, oid), o) in &objects {
                let d = ObjectReceipt::hash_payload(&o.payload);
                // A takedown covers it: it is never stored (WAI takedown §10.1).
                if gone || ix.taken.contains(&d) {
                    erased.push((*oid, d, ERASED_TAKEN_DOWN));
                    self.stats.objects_taken_down.fetch_add(1, Ordering::Relaxed);
                    continue;
                }
                if room.is_some_and(|r| ix.standing.get(&(r, d)) == Some(&REDACT)) {
                    erased.push((*oid, d, ERASED_REDACTED));
                    continue;
                }
                let records = if events { events::digests(&o.payload) } else { Vec::new() };
                let row = DObject { id: *oid, subgroup: o.subgroup, priority: o.publisher_priority, status: o.status, at: 0, props_len: 0, payload_len: 0, digest: d, records };
                rows.push((row, Bytes::from(o.properties.to_bytes().unwrap_or_default()), o.payload.clone()));
            }
        }
        let (bytes, mut index) = encode_group(&track_digest(&key), group, arrived_us, &rows, &erased);
        let (seq, at) = match self.append(w, &key, &dir, &bytes) {
            Ok(x) => x,
            Err(e) => {
                tracing::error!(target: "joule_moq_relay::dvr", error = %e, "a group could not be stored");
                return;
            }
        };
        for o in &mut index {
            o.at += at + FRAME;
        }
        let payload_bytes: u64 = index.iter().map(|o| o.payload_len as u64).sum();
        let mut ix = self.index.lock().unwrap();
        let Some(t) = ix.tracks.get_mut(&key) else { return };
        if t.evicted_through.is_some_and(|e| group <= e) {
            drop(ix);
            self.kill(w, &dir, seq, at, bytes.len() as u32);
            self.stats.groups_late.fetch_add(1, Ordering::Relaxed);
            return;
        }
        for (_, o) in &objects {
            if let Some(r) = prior_groups(&o.properties, group) {
                t.absent.insert(r.0, r.1);
            }
        }
        if let Some(((g, oo), _)) = objects.iter().find(|(_, x)| x.status == ObjectStatus::EndOfTrack) {
            t.end_of_track = Some(Location::new(*g, *oo));
        }
        *t.segs.entry(seq).or_insert(0) += 1;
        let old = t.groups.insert(group, DGroup { arrived_us, bytes: payload_bytes, objects: index.clone(), erased, seg: seq, frame_at: at, record_len: bytes.len() as u32 });
        if let Some(o) = &old
            && let Some(n) = t.segs.get_mut(&o.seg)
        {
            *n = n.saturating_sub(1);
        }
        let (old_bytes, old_n) = old.as_ref().map_or((0, 0), |g| (g.bytes, g.objects.len()));
        ix.window_bytes[window] = (ix.window_bytes[window] + payload_bytes).saturating_sub(old_bytes);
        if let Some(o) = &old {
            for x in &o.objects {
                ix.unindex(&key, group, x);
            }
        }
        for o in &index {
            ix.reindex(&key, group, o);
        }
        let new_objects = index.len().saturating_sub(old_n) as u64;
        if let Some(r) = room {
            ix.series.entry(r).or_default().change(now_us(), payload_bytes as i128 - old_bytes as i128, new_objects, payload_bytes.saturating_sub(old_bytes), 0);
        }
        drop(ix);
        if let Some(o) = old {
            self.kill(w, &dir, o.seg, o.frame_at, o.record_len);
            let open = w.open.get(&key).map(|s| s.seq);
            self.retire(&key, o.seg, open);
        }
        self.stats.groups_written.fetch_add(1, Ordering::Relaxed);
        self.stats.objects_written.fetch_add(new_objects, Ordering::Relaxed);
        self.evict_by_bytes(w, window);
    }

    /// Erase what a redaction targets from every group that holds it: the group is written
    /// again with the object among its erased, and its old record's bytes are overwritten.
    /// An event record is erased from the events-track object that carries it, the object's
    /// other records kept byte for byte; an object left with none is erased whole.
    fn erase(&self, w: &mut Writer, room: Option<[u8; 32]>, digest: [u8; 32], why: u8) {
        let places: Vec<(TrackKey, u64, u64)> = self.index.lock().unwrap().by_digest.get(&digest).cloned().unwrap_or_default();
        for (key, g, oid) in places {
            let (dir, grp, track_room) = {
                let ix = self.index.lock().unwrap();
                let Some(t) = ix.tracks.get(&key).filter(|t| room.is_none_or(|r| t.room == Some(r))) else { continue };
                let Some(grp) = t.groups.get(&g).cloned() else { continue };
                (t.dir.clone(), grp, t.room)
            };
            let Some(o) = grp.objects.iter().find(|o| o.id == oid).cloned() else { continue };
            let Ok(rec) = read_at(&seg_path(&dir, grp.seg), grp.frame_at + FRAME, grp.record_len as usize) else { continue };
            let Some(dg) = decode_group(&rec, &track_digest(&key)) else { continue };
            let mut rows: Rows = grp.objects.iter().filter_map(|x| dg.objects.iter().find(|y| y.0.id == x.id).map(|(_, props, payload, _)| (x.clone(), props.clone(), payload.clone()))).collect();
            let mut erased = grp.erased.clone();
            // The object's new payload, when a record is taken out of it and others stay.
            let replaced: Option<Bytes> = if o.digest == digest {
                None
            } else {
                let Some((_, _, payload)) = rows.iter().find(|x| x.0.id == oid) else { continue };
                events::withhold(payload, |d| *d == digest).map(|p| Bytes::from(p.into_owned()))
            };
            let delta: i128 = match &replaced {
                None => {
                    rows.retain(|x| x.0.id != oid);
                    erased.push((oid, o.digest, why));
                    -(o.payload_len as i128)
                }
                Some(p) => {
                    for x in rows.iter_mut().filter(|x| x.0.id == oid) {
                        x.0.digest = ObjectReceipt::hash_payload(p);
                        x.0.records = events::digests(p);
                        x.2 = p.clone();
                    }
                    p.len() as i128 - o.payload_len as i128
                }
            };
            let Some(index) = self.rewrite(w, &key, &dir, g, grp.arrived_us, &rows, &erased, (grp.seg, grp.frame_at, grp.record_len)) else { continue };
            {
                let mut ix = self.index.lock().unwrap();
                ix.unindex(&key, g, &o);
                if let Some(x) = index.iter().find(|x| x.id == oid) {
                    ix.reindex(&key, g, x);
                }
                if let Some(t) = ix.tracks.get_mut(&key) {
                    let window = t.window;
                    if let Some(gr) = t.groups.get_mut(&g) {
                        gr.bytes = (gr.bytes as i128 + delta).max(0) as u64;
                    }
                    ix.window_bytes[window] = (ix.window_bytes[window] as i128 + delta).max(0) as u64;
                }
                if let Some(r) = track_room {
                    ix.series.entry(r).or_default().change(now_us(), delta, 0, 0, 0);
                }
            }
            if why == ERASED_TAKEN_DOWN {
                self.stats.objects_taken_down.fetch_add(1, Ordering::Relaxed);
            } else {
                self.stats.objects_redacted.fetch_add(1, Ordering::Relaxed);
            }
            tracing::info!(target: "joule_moq_relay::dvr", dir = %dir.display(), group = g, object = oid, whole = replaced.is_none(), why, "an object erased from the store");
        }
    }

    fn evict_by_bytes(&self, w: &mut Writer, window: usize) {
        loop {
            let victim = {
                let ix = self.index.lock().unwrap();
                if ix.window_bytes[window] <= self.windows[window].max_bytes {
                    return;
                }
                oldest(&ix, |t| t.window == window)
            };
            match victim {
                Some((k, g)) => self.evict(w, &k, g),
                None => return,
            }
        }
    }

    fn evict_by_age(&self, w: &mut Writer, now: u64) {
        for (i, win) in self.windows.iter().enumerate() {
            let floor = now.saturating_sub(win.max_age_us);
            loop {
                let victim = {
                    let ix = self.index.lock().unwrap();
                    oldest(&ix, |t| t.window == i).filter(|(k, g)| ix.tracks[k].groups[g].arrived_us < floor)
                };
                match victim {
                    Some((k, g)) => self.evict(w, &k, g),
                    None => break,
                }
            }
        }
    }

    /// Evict group `g` of `key`, and group `g` of its receipts or media track with it. A
    /// segment no group stands in any longer is deleted.
    fn evict(&self, w: &mut Writer, key: &TrackKey, g: u64) {
        let partner = match self.media_of(&key.1) {
            Some(m) => (key.0.clone(), m),
            None => (key.0.clone(), self.receipts_of(&key.1)),
        };
        for k in [key.clone(), partner] {
            // The mark moves to `g`, and every group at or below it goes: none is ever held
            // below a track's evicted mark. Groups are sealed out of order under load and a
            // group's arrival time can be later than a higher group's, so evicting `g` alone
            // could leave a lower group held while higher ones went: a gap in the window.
            let victims: Vec<u64> = {
                let mut ix = self.index.lock().unwrap();
                let Some(t) = ix.tracks.get_mut(&k) else { continue };
                let through = t.evicted_through.map_or(g, |e| e.max(g));
                t.evicted_through = Some(through);
                t.absent.retain(|_, hi| *hi > through);
                t.groups.range(..=through).map(|(gg, _)| *gg).collect()
            };
            for v in victims {
                let seg = {
                    let mut ix = self.index.lock().unwrap();
                    let Some(t) = ix.tracks.get_mut(&k) else { break };
                    let Some(grp) = t.groups.remove(&v) else { continue };
                    if let Some(n) = t.segs.get_mut(&grp.seg) {
                        *n = n.saturating_sub(1);
                    }
                    let (window, room) = (t.window, t.room);
                    ix.window_bytes[window] = ix.window_bytes[window].saturating_sub(grp.bytes);
                    for o in &grp.objects {
                        ix.unindex(&k, v, o);
                    }
                    if let Some(r) = room {
                        ix.series.entry(r).or_default().change(now_us(), -(grp.bytes as i128), 0, 0, 0);
                    }
                    grp.seg
                };
                self.stats.groups_evicted.fetch_add(1, Ordering::Relaxed);
                let open = w.open.get(&k).map(|s| s.seq);
                self.retire(&k, seg, open);
            }
        }
    }
}

/// The oldest group held among the tracks `f` selects: by arrival, then track and group.
fn oldest(ix: &Index, f: impl Fn(&DTrack) -> bool) -> Option<(TrackKey, u64)> {
    // Within a track, the lowest group goes first, so a track's window stays one run of
    // groups: groups finish, and are sealed, out of order under load, and evicting by
    // arrival left gaps below a late group. Across tracks, the track whose lowest group
    // arrived first.
    ix.tracks
        .iter()
        .filter(|(_, t)| f(t))
        .filter_map(|(k, t)| t.groups.iter().next().map(|(g, grp)| (grp.arrived_us, k.clone(), *g)))
        .min_by(|a, b| a.0.cmp(&b.0).then_with(|| a.2.cmp(&b.2)))
        .map(|(_, k, g)| (k, g))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_group_record_reads_back_and_a_flipped_byte_is_found() {
        let td = [7u8; 32];
        let o = |id: u64| DObject { id, subgroup: Some(0), priority: 3, status: ObjectStatus::Normal, at: 0, props_len: 0, payload_len: 0, digest: ObjectReceipt::hash_payload(format!("p{id}").as_bytes()), records: vec![] };
        let rows: Rows = (0..3).map(|i| (o(i), Bytes::new(), Bytes::from(format!("p{i}")))).collect();
        let (mut b, index) = encode_group(&td, 5, 99, &rows, &[(3, [1; 32], ERASED_REDACTED)]);
        let dg = decode_group(&b, &td).unwrap();
        assert_eq!((dg.group, dg.arrived_us, dg.count, dg.erased.len()), (5, 99, 3, 1));
        assert!(dg.objects.iter().all(|x| x.3));
        assert_eq!(index[1].at, dg.objects[1].0.at);
        // A flipped payload byte fails that object's check alone.
        let at = index[1].at as usize;
        b[at] ^= 0x40;
        let dg = decode_group(&b, &td).unwrap();
        assert_eq!(dg.objects.iter().map(|x| x.3).collect::<Vec<_>>(), vec![true, false, true]);
        // A flipped header byte drops the group; so does another track's record.
        b[40] ^= 1;
        assert!(decode_group(&b, &td).is_none());
        let (b2, _) = encode_group(&td, 5, 99, &rows, &[]);
        assert!(decode_group(&b2, &[8; 32]).is_none());
    }

    #[test]
    fn a_track_record_reads_back() {
        let key = (TrackNamespace::from_strs(["r", "p", "video"]), TrackName::from_str("0"));
        let mut p = Properties::new();
        p.push(joule_moq::draft22::kvp::KeyValuePair::varint(prop::DEFAULT_PUBLISHER_PRIORITY, 7));
        let b = encode_track(&key, &p);
        assert_eq!(decode_track(&b), Some((key, p)));
        let mut bad = b.clone();
        bad[7] ^= 1;
        assert!(decode_track(&bad).is_none());
    }

    #[test]
    fn the_series_integrates_what_is_held() {
        let mut s = Series::default();
        s.change(1_000_000, 100, 1, 100, 0);
        s.change(3_000_000, 50, 1, 50, 0);
        // 100 bytes for 2 s, then 150.
        assert_eq!(s.at(3_000_000).0, 100 * 2_000_000);
        assert_eq!(s.at(4_000_000).0, 100 * 2_000_000 + 150 * 1_000_000);
        assert_eq!(s.at(0), (0, 0, 0, 0));
        assert_eq!((s.at(5_000_000).1, s.at(5_000_000).2), (2, 150));
    }

    #[test]
    fn segment_names_read_back() {
        assert_eq!(seg_seq(&seg_path(Path::new("/x"), 42)), Some(42));
        assert_eq!(seg_seq(Path::new("/x/TRACK")), None);
    }
}