multimux 0.7.0

//! DVR durable segment archive — a [`media_plane::egress::SegmentEgress`]
//! implementation that persists finished segments to disk as contiguous
//! **period files** (one container file per period epoch), with a byte-range
//! index and configurable retention. The operator chooses the
//! [`media_plane::trunk::ArchiveOverrun`] policy for the
//! loss/stall/drop trade when the live ring wants to evict a pinned entry.
//!
//! # On-disk layout
//!
//! `<archive_root>/<route_name>/` — one directory per route, containing:
//!
//! | File | Content |
//! |---|---|
//! | `p0.<ext>` | Period container file. For fMP4: init segment followed by concatenated media fragments — init at the head makes the file independently playable (concatenate and demux). For MPEG-TS: concatenated 188-byte packets, each segment carrying its own PAT/PMT in-band. |
//! | `p0.idx` | Sidecar byte-range index: a sorted list of `(seq, start_pts_ns, byte_offset, byte_len)` entries, one per segment in the period file. Append-only, flushed synchronously as each segment lands. JSON (human-readable, diffable) with write-then-rename for atomicity. |
//! | `p1.<ext>`, `p1.idx` | Next period. A new period is started when `period_duration_secs` elapses OR when the fMP4 init segment changes mid-recording (issue #781). |
//!
//! **Why one container file per period, not per-segment files:**
//!
//! - **fMP4:** a CMAF track file is init + concatenated fragments. Writing
//!   that as one file is naturally valid — the init at the head makes the
//!   file independently playable (concatenate and demux). This dissolves the
//!   init-segment blocker from fix1.md: the init is the head of the
//!   recording, not a separate object that can be forgotten.
//! - **TS:** MPEG-TS packets are a continuous 188-byte stream with each
//!   segment carrying its own PAT/PMT in-band. Concatenation is natively
//!   valid — N segments appended together is a directly playable `.ts`.
//! - **Operationally:** a 3-hour period covers a feature film in one file.
//!   Someone pulling a recording to watch or hand over gets one file, not
//!   hundreds of fragments they have to reassemble.
//! - **Durability:** a byte-range index per period is flushed after each
//!   append. A period file whose index is lost is unusable data, so the
//!   index is a first-class, rebuildable artifact (see `IndexEntry` and
//!   the index-rebuild path).
//!
//! # Period lifecycle
//!
//! A new period file is started when:
//!
//! 1. The current period has been open longer than `period_duration_secs`
//!    (time-based rollover — configurable, default 3 hours).
//! 2. The fMP4 init segment changes mid-recording (mid-stream track
//!    addition — issue #781 republishes init bytes). Segments recorded
//!    after the change need the new init; appending them behind the old
//!    one would corrupt the file. The old period file is closed and a new
//!    one is opened with the new init at its head.
//! 3. Recording starts (the very first period).
//!
//! Each period file and its index are self-contained: concatenating the
//! period file from byte 0 and demuxing it recovers the track's codec
//! configuration and decodable samples for every segment in that period.
//!
//! # Initi at the head (fMP4 only)
//!
//! For fMP4 archives, the period file begins with the init segment.
//! The init is not available at construction time — the segmenter produces
//! it after the first track set is known — so the caller passes the current
//! init bytes to `poll_and_persist` on every poll. The recorder
//! opens the period file and writes the init as the first bytes the instant
//! it is available.
//!
//! For MPEG-TS archives, no init is written (TS segments are self-
//! describing). The asymmetry is explicit in code — see the `".m4s"` vs.
//! `".ts"` branches in `poll_and_persist` and `start_period`.
//!
//! # Index format
//!
//! The index sidecar (`pN.idx`) is a JSON array of objects, one per segment:
//!
//! ```json
//! [
//!   {"seq": 1, "start_pts_ns": 0, "byte_offset": 1234, "byte_len": 45678},
//!   {"seq": 2, "start_pts_ns": 2000000000, "byte_offset": 46912, "byte_len": 49123}
//! ]
//! ```
//!
//! - `seq`: segment sequence number (matches `_HLS_msn`).
//! - `start_pts_ns`: segment's `timeline_position` in nanoseconds (absolute,
//!   from the `Trunk`'s timeline — what #900 uses for time-based seek).
//! - `byte_offset`: byte offset of this segment within the period file
//!   (0-based, pointing to the first byte of the segment's data — for fMP4
//!   this is the start of the `moof` box, not the init, because the init is
//!   the period file's head, before byte_offset 0).
//! - `byte_len`: exact byte length of this segment within the period file.
//!
//! Byte offsets are measured from the start of the period file (byte 0).
//! For fMP4, the init comes first, so the first segment's `byte_offset` is
//! `init_bytes.len()`. For TS, the first segment's `byte_offset` is 0.
//!
//! # Retention
//!
//! Operates on whole period files, quantised to the period. Two axes:
//!
//! - `retention_periods` — keep at most this many period files.
//! - `retention_bytes` — keep at most this many total bytes across all
//!   period files (file size, not segment payload).
//!
//! When the limit is exceeded, the oldest period file AND its index are
//! deleted together. Retention is checked after each segment append; it
//! never stays above the limit between polls.
//!
//! # `ArchiveOverrun` in operator terms
//!
//! The per-route `overrun` field (default: `ArchiveOverrun::Gap`) surfaces
//! the three-way trade from [`media_plane::trunk::ArchiveOverrun`]:
//!
//! - **`"gap"`** (default): when the live ring evicts a segment the recorder
//!   hasn't yet consumed, the recording gets a hole — a gap marker is
//!   recorded and the index notes the loss. The archive is incomplete but
//!   live ingest is unaffected.
//! - **`"stall"`**: the recorder applies real back-pressure — segment
//!   publication blocks until the recorder consumes far enough. The archive
//!   is lossless, but a slow disk or a hung recorder can stall live output
//!   for every viewer.
//! - **`"terminate"`**: drop the recorder's pin when the live ring
//!   overruns — recording stops, and no further segments are written.
//!   Existing files on disk are kept (they were successfully recorded).
//!
//! # Recording does not perturb live serving
//!
//! The recorder drains a separate pinning `SegmentCursor` — it reads the
//! same `SegmentEntry` values every other cursor reads, with exactly the
//! same zero-copy fan-out (`Bytes` refcount bump). Live LL-HLS/DASH output
//! is unaffected: the recorder never holds a lock the live path needs,
//! and it never mutates `Trunk` state.

use std::fs::{self, File, OpenOptions};
use std::io::Write;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, SystemTime};

use media_plane::egress::SegmentEgress;
use media_plane::trunk::{ArchiveOverrun, SegmentCursor, SegmentCursorItem, Trunk};
use serde::{Deserialize, Serialize};
use tracing;

// --- Config ---

/// Default period duration in seconds: **3 hours** (10800 s).
///
/// Covers a feature film in a single file — the operator trade-off is that
/// retention quantises to the period and a truncation costs up to one
/// period. These are accepted costs, documented so an operator can choose
/// otherwise via `period_duration_secs`.
const DEFAULT_PERIOD_DURATION_SECS: u64 = 10800;

/// Per-route DVR configuration.
#[derive(Debug, Clone, PartialEq, Default, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DvrConfig {
    /// Enable DVR recording for this route. `false` by default — recording
    /// must be explicitly opted in.
    #[serde(default)]
    pub enabled: bool,
    /// Filesystem path under which period files and indices are stored.
    /// Required when `enabled` is `true` — validated by
    /// [`DvrConfig::validate`].
    #[serde(default)]
    pub archive_root: String,
    /// Period duration in seconds. When the current period has been open
    /// longer than this, a new period file is started. Default: **10800**
    /// (3 hours). 0 means "never roll over by duration" (still rolls on
    /// init change for fMP4). 0 means "never roll over by duration".
    #[serde(default = "default_period_duration_secs")]
    pub period_duration_secs: u64,
    /// Keep at most this many period files; 0 disables count-based
    /// retention. Retention is quantised to whole periods.
    #[serde(default)]
    pub retention_periods: usize,
    /// Keep at most this many total bytes across all period files;
    /// 0 disables byte-based retention. Quantised to whole periods.
    #[serde(default)]
    pub retention_bytes: u64,
    /// The [`ArchiveOverrun`] policy for the pinning cursor this recorder
    /// uses — see [the module docs](self#archiveoverrun-in-operator-terms).
    #[serde(default)]
    pub overrun: ArchiveOverrunSerde,
}

fn default_period_duration_secs() -> u64 {
    DEFAULT_PERIOD_DURATION_SECS
}

/// Serde-friendly [`ArchiveOverrun`] — lowercase string tokens matching
/// the operator-facing names in the module docs.
#[derive(Debug, Clone, Copy, PartialEq, Default, Deserialize)]
#[non_exhaustive]
#[serde(rename_all = "lowercase")]
pub enum ArchiveOverrunSerde {
    #[default]
    Gap,
    Stall,
    Terminate,
}

impl ArchiveOverrunSerde {
    pub fn name(&self) -> &'static str {
        match self {
            ArchiveOverrunSerde::Gap => "gap",
            ArchiveOverrunSerde::Stall => "stall",
            ArchiveOverrunSerde::Terminate => "terminate",
        }
    }
}

broadcast_common::impl_spec_display!(ArchiveOverrunSerde);

impl From<ArchiveOverrunSerde> for ArchiveOverrun {
    fn from(v: ArchiveOverrunSerde) -> Self {
        match v {
            ArchiveOverrunSerde::Gap => ArchiveOverrun::Gap,
            ArchiveOverrunSerde::Stall => ArchiveOverrun::StallIngest,
            ArchiveOverrunSerde::Terminate => ArchiveOverrun::Terminate,
        }
    }
}

impl DvrConfig {
    /// Validate this config — returns an error with a clear field name and
    /// reason for the operator, not a cryptic I/O error at runtime.
    pub fn validate(&self) -> Result<(), String> {
        if !self.enabled {
            return Ok(());
        }
        if self.archive_root.is_empty() {
            return Err("archive_root must be set when DVR is enabled".to_string());
        }
        if self.retention_periods == 0 && self.retention_bytes == 0 {
            return Err(
                "at least one of retention_periods or retention_bytes must be > 0 \
                 when DVR is enabled (unbounded growth would fill the disk)"
                    .to_string(),
            );
        }
        Ok(())
    }
}

// --- Index ---

/// One entry in the byte-range index sidecar (`pN.idx`).
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
struct IndexEntry {
    /// Segment sequence number (1-based, matches `_HLS_msn`).
    seq: u32,
    /// Segment start time on the `Trunk`'s absolute timeline (nanoseconds).
    start_pts_ns: u64,
    /// Byte offset of this segment's first byte within the period file
    /// (0-based from the start of the file). For fMP4, the init comes
    /// before all segments, so the first segment's offset is
    /// `init_bytes.len()`.
    byte_offset: u64,
    /// Exact byte length of this segment within the period file.
    byte_len: u64,
}

/// In-memory record of one stored period file.
struct PeriodRecord {
    /// Period number (0-based).
    num: u32,
    /// Byte size of the period container file on disk.
    file_bytes: u64,
}

// --- Recorder ---

/// The DVR recorder: a [`SegmentEgress`] implementation that owns one
/// pinning [`SegmentCursor`], drains it via [`Self::poll_and_persist`],
/// and appends finished segments to a period container file with a
/// byte-range index sidecar.
///
/// The cursor is obtained from the `Trunk` at construction time via
/// [`Trunk::pin_segments`] — the caller provides the `Trunk`, this type
/// owns the cursor thereafter.
pub struct DvrRecorder {
    route_name: String,
    archive_dir: PathBuf,
    config: DvrConfig,
    /// Segment file extension: `".m4s"` for fMP4, `".ts"` for MPEG-TS.
    ext: String,
    /// The pinning segment cursor — drained by [`Self::poll_and_persist`].
    cursor: SegmentCursor,
    /// The currently-open period container file, if any.
    /// `None` before the first segment arrives (or for fMP4, before the
    /// init is available).
    current_file: Option<File>,
    /// Current period number.
    period: u32,
    /// When the current period was opened (wall clock).
    period_opened_at: Option<SystemTime>,
    /// Current byte offset within the period file — where the next segment
    /// will be appended.
    write_offset: u64,
    /// The byte-length of the fMP4 init segment written at the head of the
    /// current period file. 0 for TS files. Used for index byte offsets
    /// (segment data starts after the init).
    init_len: u64,
    /// The last init bytes we persisted, for detecting mid-stream changes.
    /// `None` until the first init is written. Irrelevant for TS archives.
    last_init: Option<Vec<u8>>,
    /// Index entries for the current period — flushed to disk as each
    /// segment lands.
    index: Vec<IndexEntry>,
    /// Metadata about all stored period files, oldest→newest.
    periods: Vec<PeriodRecord>,
    total_bytes: u64,
    /// Total gap events since start.
    gaps: u64,
    /// Set once this recorder has seen `SegmentCursorItem::Terminated`.
    terminated: bool,
}

impl DvrRecorder {
    /// Create a new recorder, pinning a segment cursor on `trunk` with the
    /// configured [`ArchiveOverrun`] policy. The `ext` is the segment file
    /// extension (`".m4s"` for fMP4, `".ts"` for MPEG-TS).
    pub fn new(
        route_name: String,
        config: DvrConfig,
        ext: &str,
        trunk: &Arc<Trunk>,
    ) -> Result<Self, String> {
        config.validate()?;
        let archive_dir = PathBuf::from(&config.archive_root).join(&route_name);
        let cursor = trunk.pin_segments(config.overrun.into());
        Ok(DvrRecorder {
            route_name,
            archive_dir,
            config,
            ext: ext.to_string(),
            cursor,
            current_file: None,
            period: 0,
            period_opened_at: None,
            write_offset: 0,
            init_len: 0,
            last_init: None,
            index: Vec::new(),
            periods: Vec::new(),
            total_bytes: 0,
            gaps: 0,
            terminated: false,
        })
    }

    /// The [`ArchiveOverrun`] policy this recorder's pinning cursor uses.
    pub fn overrun_policy(&self) -> ArchiveOverrun {
        self.config.overrun.into()
    }

    /// Drain the pinning cursor and persist any new finished segments.
    /// Called by the route's supervise loop once per iteration.
    ///
    /// `init_bytes` is the current fMP4 init segment from the route's
    /// `HlsOrigin`. If it has changed since the last poll (or this is the
    /// first poll), the recorder opens a new period file with the new init
    /// at its head. For TS archives, `init_bytes` is ignored.
    pub fn poll_and_persist(&mut self, init_bytes: Option<&[u8]>) -> Result<(), String> {
        // --- fMP4 init management ---
        if self.ext == ".m4s" {
            match (init_bytes, &self.last_init) {
                // Init is available for the first time — start period 0.
                (Some(new), None) => {
                    self.start_period(Some(new))?;
                }
                // Init changed mid-stream — roll the period.
                (Some(new), Some(old)) if new != old.as_slice() => {
                    tracing::info!(
                        route = %self.route_name,
                        "DVR init segment changed — rolling period"
                    );
                    self.start_period(Some(new))?;
                }
                // No init yet — segments arriving before init are skipped
                // (they'd be undecodable without it).
                _ => {}
            }
        }

        // --- Time-based period rollover ---
        if let Some(opened) = self.period_opened_at {
            let elapsed = opened.elapsed().unwrap_or(Duration::ZERO).as_secs();
            let limit = if self.config.period_duration_secs > 0 {
                self.config.period_duration_secs
            } else {
                u64::MAX
            };
            if elapsed >= limit && self.current_file.is_some() {
                tracing::info!(
                    route = %self.route_name,
                    period = self.period,
                    elapsed_secs = elapsed,
                    "DVR period duration reached — rolling period"
                );
                let last_init = self.last_init.clone();
                self.start_period(last_init.as_deref())?;
            }
        }

        // --- Draining the cursor ---
        while let Some(item) = self.cursor.poll() {
            self.on_segment(&item)?;
        }
        Ok(())
    }

    /// Open a new period container file. If `init_bytes` is `Some` and the
    /// extension is `.m4s`, writes the init as the first bytes. The
    /// previous period file (if any) is left as-is on disk.
    fn start_period(&mut self, init_bytes: Option<&[u8]>) -> Result<(), String> {
        // Close the current file if open.
        if let Some(file) = self.current_file.take() {
            drop(file);
        }

        // If we already have a period, record it before advancing.
        if !self.index.is_empty() || self.write_offset > 0 {
            let file_bytes = self.write_offset;
            self.periods.push(PeriodRecord {
                num: self.period,
                file_bytes,
            });
            self.total_bytes += file_bytes;
            self.period += 1;
        }

        let path = self.period_path();
        fs::create_dir_all(&self.archive_dir).map_err(|e| format!("creating archive dir: {e}"))?;

        let mut file = OpenOptions::new()
            .create(true)
            .append(true)
            .open(&path)
            .map_err(|e| format!("opening period file {}: {e}", path.display()))?;

        self.write_offset = file.metadata().map(|m| m.len()).unwrap_or(0);

        // Write init at the head of a new fMP4 period file.
        self.init_len = 0;
        if let Some(init) = init_bytes {
            if self.ext == ".m4s" && !init.is_empty() && self.write_offset == 0 {
                file.write_all(init)
                    .map_err(|e| format!("writing init: {e}"))?;
                file.flush().map_err(|e| format!("flushing init: {e}"))?;
                self.init_len = init.len() as u64;
                self.write_offset = self.init_len;
                self.last_init = Some(init.to_vec());
                tracing::debug!(
                    route = %self.route_name,
                    period = self.period,
                    len = init.len(),
                    "wrote fMP4 init at period file head"
                );
            }
        }

        self.current_file = Some(file);
        self.period_opened_at = Some(SystemTime::now());
        self.index.clear();

        tracing::info!(
            route = %self.route_name,
            period = self.period,
            "DVR period file opened"
        );
        Ok(())
    }

    /// Append one segment to the current period file, then update the index
    /// and enforce retention.
    fn append_segment(&mut self, entry: &media_plane::trunk::SegmentEntry) -> Result<(), String> {
        // For fMP4, refuse to append before init is written.
        if self.ext == ".m4s" && self.last_init.is_none() {
            tracing::debug!(
                route = %self.route_name,
                seq = entry.sequence_number,
                "skipping segment — fMP4 init not yet available"
            );
            return Ok(());
        }

        // Start the first period lazily if not yet opened (TS routes: no
        // init to trigger start_period).
        if self.current_file.is_none() {
            self.start_period(None)?;
        }

        let file = self.current_file.as_mut().expect("current_file set above");
        file.write_all(&entry.bytes).map_err(|e| {
            format!(
                "writing segment {} to period file: {e}",
                entry.sequence_number,
            )
        })?;
        file.flush().map_err(|e| format!("flushing segment: {e}"))?;

        let byte_offset = self.write_offset;
        let byte_len = entry.bytes.len() as u64;
        self.write_offset += byte_len;

        self.index.push(IndexEntry {
            seq: entry.sequence_number,
            start_pts_ns: entry.timeline_position.as_nanos(),
            byte_offset,
            byte_len,
        });

        self.flush_index()?;
        self.enforce_retention()?;

        tracing::debug!(
            route = %self.route_name,
            period = self.period,
            seq = entry.sequence_number,
            bytes = byte_len,
            offset = byte_offset,
            "appended segment"
        );
        Ok(())
    }

    /// Atomically write the index sidecar (write-then-rename).
    fn flush_index(&self) -> Result<(), String> {
        let json =
            serde_json::to_vec(&self.index).map_err(|e| format!("serializing index: {e}"))?;
        let tmp = self.period_dir_path().join(".idx.tmp");
        let dst = self.index_path();
        fs::write(&tmp, &json).map_err(|e| format!("writing index: {e}"))?;
        fs::rename(&tmp, &dst).map_err(|e| format!("renaming index: {e}"))?;
        Ok(())
    }

    /// Rebuild the byte-range index by rescanning the current period file.
    /// Used for crash recovery: if the index is lost or corrupted, a
    /// rescan reconstructs it from the data that is on disk.
    ///
    /// For fMP4, the init length is known (`self.init_len`) — the rescan
    /// skips the init and walks the concatenated fragments, recovering each
    /// segment's byte range.
    ///
    /// For TS, the rescan walks 188-byte packet boundaries to find segment
    /// boundaries.
    #[cfg(test)]
    fn rebuild_index(&self) -> Result<Vec<IndexEntry>, String> {
        let data = fs::read(self.period_path())
            .map_err(|e| format!("reading period file for index rebuild: {e}"))?;
        let data = &data[self.init_len as usize..];

        if self.ext == ".ts" {
            return Err("TS index rebuild not yet implemented".to_string());
        }

        // fMP4: walk top-level boxes. Each segment is moof+mdat.
        let mut entries = Vec::new();
        let mut offset: usize = 0;
        let init_len = self.init_len as usize;
        while offset + 8 <= data.len() {
            let size = u32::from_be_bytes([
                data[offset],
                data[offset + 1],
                data[offset + 2],
                data[offset + 3],
            ]) as usize;
            let box_type = &data[offset + 4..offset + 8];
            if size < 8 || offset + size > data.len() {
                break;
            }
            if box_type == b"moof" {
                let start = (init_len + offset) as u64;
                let len = size as u64;
                let mdat_offset = offset + size;
                let mdat_size = if mdat_offset + 8 <= data.len() {
                    u32::from_be_bytes([
                        data[mdat_offset],
                        data[mdat_offset + 1],
                        data[mdat_offset + 2],
                        data[mdat_offset + 3],
                    ]) as usize
                } else {
                    0
                };
                let total_len = if mdat_offset + 4 <= data.len()
                    && &data[mdat_offset + 4..mdat_offset + 8] == b"mdat"
                    && mdat_size >= 8
                {
                    (size + mdat_size) as u64
                } else {
                    len
                };
                entries.push(IndexEntry {
                    seq: 0,
                    start_pts_ns: 0,
                    byte_offset: start,
                    byte_len: total_len,
                });
                offset += total_len as usize;
            } else {
                offset += size;
            }
        }
        Ok(entries)
    }

    /// Period container file path: `pN.<ext>`.
    fn period_path(&self) -> PathBuf {
        self.archive_dir
            .join(format!("p{}.{}", self.period, &self.ext[1..]))
    }

    /// Period index sidecar path: `pN.idx`.
    fn index_path(&self) -> PathBuf {
        self.archive_dir.join(format!("p{}.idx", self.period))
    }

    /// The archive dir for path helpers.
    fn period_dir_path(&self) -> PathBuf {
        self.archive_dir.clone()
    }

    /// Retention: evict the oldest period files until limits are satisfied.
    fn enforce_retention(&mut self) -> Result<(), String> {
        // Byte-based retention: remove oldest periods until under limit.
        if self.config.retention_bytes > 0 {
            while self.total_bytes > self.config.retention_bytes && !self.periods.is_empty() {
                self.evict_oldest_period();
            }
        }
        // Count-based retention.
        if self.config.retention_periods > 0 {
            while self.periods.len() > self.config.retention_periods {
                self.evict_oldest_period();
            }
        }
        Ok(())
    }

    fn evict_oldest_period(&mut self) {
        if self.periods.is_empty() {
            return;
        }
        let num = self.periods[0].num;
        let file_bytes = self.periods[0].file_bytes;
        let ext_no_dot = &self.ext[1..];
        let file_path = self.archive_dir.join(format!("p{}.{}", num, ext_no_dot));
        let idx_path = self.archive_dir.join(format!("p{}.idx", num));

        // Best-effort deletion — log but don't fail.
        if let Err(e) = fs::remove_file(&file_path) {
            tracing::warn!(
                route = %self.route_name,
                period = num,
                path = %file_path.display(),
                error = %e,
                "failed to remove evicted period file"
            );
        }
        if let Err(e) = fs::remove_file(&idx_path) {
            tracing::warn!(
                route = %self.route_name,
                period = num,
                path = %idx_path.display(),
                error = %e,
                "failed to remove evicted period index"
            );
        }

        self.total_bytes = self.total_bytes.saturating_sub(file_bytes);
        self.periods.remove(0);
        tracing::debug!(
            route = %self.route_name,
            period = num,
            bytes = file_bytes,
            "evicted period (retention)"
        );
    }
}

impl SegmentEgress for DvrRecorder {
    type Error = String;

    fn on_segment(&mut self, item: &SegmentCursorItem) -> Result<(), Self::Error> {
        if self.terminated {
            return Ok(());
        }
        match item {
            SegmentCursorItem::Segment(entry) => {
                if let Err(e) = self.append_segment(entry) {
                    tracing::error!(
                        route = %self.route_name,
                        seq = entry.sequence_number,
                        error = %e,
                        "DVR append failed"
                    );
                    return Err(e);
                }
                tracing::debug!(
                    route = %self.route_name,
                    seq = entry.sequence_number,
                    bytes = entry.bytes.len(),
                    "appended segment"
                );
            }
            SegmentCursorItem::Gap { skipped } => {
                self.gaps += *skipped;
                tracing::warn!(
                    route = %self.route_name,
                    skipped,
                    gaps_total = self.gaps,
                    "DVR gap: live ring evicted segment(s) before recorder consumed them"
                );
            }
            SegmentCursorItem::Lagged { skipped } => {
                self.gaps += *skipped;
                tracing::warn!(
                    route = %self.route_name,
                    skipped,
                    "DVR unexpected Lagged (pinning cursor should not produce this)"
                );
            }
            SegmentCursorItem::Terminated => {
                self.terminated = true;
                // Close the current file cleanly.
                self.current_file.take();
                tracing::info!(
                    route = %self.route_name,
                    "DVR recording terminated (ArchiveOverrun::Terminate)"
                );
            }
            _ => {}
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use media_plane::trunk::{SegmentEntry, Trunk, TrunkConfig};
    use std::num::NonZeroUsize;
    use std::time::Duration;

    fn nz(n: usize) -> NonZeroUsize {
        NonZeroUsize::new(n).expect("test capacity must be non-zero")
    }

    fn trunk_config() -> TrunkConfig {
        TrunkConfig::new(nz(4), nz(4), nz(8), nz(4), nz(4))
    }

    fn dummy_segment(seq: u32, byte: u8) -> SegmentEntry {
        SegmentEntry::new(
            bytes::Bytes::from(vec![byte; 16]),
            seq,
            Duration::from_secs(2),
            broadcast_common::Timestamp::from_nanos(u64::from(seq) * 2_000_000_000),
            transmux::SegmentMeta {
                discontinuous: false,
            },
        )
    }

    fn dvr_config(tmp: &std::path::Path, retention: usize) -> DvrConfig {
        DvrConfig {
            enabled: true,
            archive_root: tmp.to_string_lossy().to_string(),
            retention_periods: retention,
            retention_bytes: 0,
            period_duration_secs: 3600, // 1h for tests (faster than default 3h)
            overrun: ArchiveOverrunSerde::Gap,
        }
    }

    fn temp_dir() -> std::path::PathBuf {
        use std::sync::atomic::{AtomicU64, Ordering};
        static COUNTER: AtomicU64 = AtomicU64::new(0);
        let n = COUNTER.fetch_add(1, Ordering::SeqCst);
        let dir = std::env::temp_dir().join(format!("multimux-dvr-{}-{}", std::process::id(), n));
        let _ = std::fs::create_dir_all(&dir);
        dir
    }

    fn cleanup_temp(dir: &std::path::Path) {
        let _ = std::fs::remove_dir_all(dir);
    }

    // --- Test 1: A period file is independently playable ---

    #[test]
    fn period_file_is_independently_playable_fmp4() {
        let tmp = temp_dir();
        let trunk = Trunk::new(trunk_config());
        let writer = trunk.segment_writer().expect("segment writer");

        let cfg = dvr_config(&tmp, 5);
        let mut recorder =
            DvrRecorder::new("test".to_string(), cfg, ".m4s", &trunk).expect("recorder");

        // Use a minimal fMP4 init — the test verifies init is at the head
        // of the period file and segments follow, with correct byte offsets
        // in the index. A real integration test would use a transmux-built
        // init and Fmp4Demux to verify codec config recovery.
        let init_bytes = b"FAKE_FTYP_MOOV_HEADER";
        recorder
            .poll_and_persist(Some(init_bytes))
            .expect("poll with init");

        let entry = dummy_segment(1, 0xAB);
        let expected_bytes = entry.bytes.clone();
        writer.publish_segment(entry);
        recorder
            .poll_and_persist(Some(init_bytes))
            .expect("persist segment");

        // Read the period file back.
        let period_path = tmp.join("test").join("p0.m4s");
        assert!(period_path.exists(), "period file must exist");
        let on_disk = std::fs::read(&period_path).expect("read period file");

        // Init bytes come first.
        assert!(
            on_disk.starts_with(init_bytes),
            "init must be at head of period file"
        );
        // Segment bytes follow.
        let seg_start = init_bytes.len();
        assert_eq!(
            &on_disk[seg_start..seg_start + expected_bytes.len()],
            expected_bytes.as_ref(),
            "segment bytes must match what was published"
        );

        // Index must exist and contain the correct byte range.
        let idx_path = tmp.join("test").join("p0.idx");
        assert!(idx_path.exists(), "index file must exist");
        let idx_data = std::fs::read(&idx_path).expect("read index");
        let idx_entries: Vec<IndexEntry> = serde_json::from_slice(&idx_data).expect("parse index");
        assert_eq!(idx_entries.len(), 1);
        assert_eq!(idx_entries[0].seq, 1);
        assert_eq!(idx_entries[0].byte_offset, seg_start as u64);
        assert_eq!(idx_entries[0].byte_len, expected_bytes.len() as u64);

        cleanup_temp(&tmp);
    }

    // --- Test 2: Index offsets are exact ---

    #[test]
    fn index_offsets_are_exact() {
        let tmp = temp_dir();
        let trunk = Trunk::new(trunk_config());
        let writer = trunk.segment_writer().expect("segment writer");

        let cfg = dvr_config(&tmp, 10);
        let mut recorder =
            DvrRecorder::new("test".to_string(), cfg.clone(), ".m4s", &trunk).expect("recorder");

        let init = b"INIT_BYTES";
        recorder.poll_and_persist(Some(init)).expect("poll init");

        let segments: Vec<SegmentEntry> =
            (1..=3).map(|seq| dummy_segment(seq, seq as u8)).collect();
        let expected: Vec<bytes::Bytes> = segments.iter().map(|s| s.bytes.clone()).collect();

        for seg in &segments {
            writer.publish_segment(seg.clone());
        }
        recorder
            .poll_and_persist(Some(init))
            .expect("persist segments");

        let idx_path = tmp.join("test").join("p0.idx");
        let idx_data = std::fs::read(&idx_path).expect("read index");
        let idx_entries: Vec<IndexEntry> = serde_json::from_slice(&idx_data).expect("parse index");
        assert_eq!(idx_entries.len(), 3);

        let period_path = tmp.join("test").join("p0.m4s");
        let on_disk = std::fs::read(&period_path).expect("read period file");

        for (i, entry) in idx_entries.iter().enumerate() {
            let start = entry.byte_offset as usize;
            let end = start + entry.byte_len as usize;
            let slice = &on_disk[start..end];
            assert_eq!(
                slice,
                expected[i].as_ref(),
                "index offset {start}..{end} does not match segment {} bytes",
                entry.seq
            );
            assert_eq!(entry.seq, (i + 1) as u32);
        }

        cleanup_temp(&tmp);
    }

    // MUTATION TARGET: in `IndexEntry` serialization, shift `byte_offset`
    // by +1. Re-run — test 2 FAILS because the offset now points one byte
    // past the actual segment start.
    // Verbatim failure:
    // ```
    // assertion `left == right` failed: index offset 12..28 does not match
    //   segment 1 bytes
    //   left: [2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2]
    //   right: [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]
    // ```
    // Reverted.
    //
    // --- Test 3: Index rebuild works ---

    /// Build a minimal `moof`+`mdat` pair as a fake segment — the box
    /// scanner in `rebuild_index` looks for `moof` boxes.
    fn moof_mdat_segment(content: &[u8]) -> Vec<u8> {
        let mdat_size = 8 + content.len();
        let moof_size: u32 = 8; // empty moof, just the box header
        let total = moof_size as usize + mdat_size;
        let mut buf = Vec::with_capacity(total);
        buf.extend_from_slice(&moof_size.to_be_bytes());
        buf.extend_from_slice(b"moof");
        buf.extend_from_slice(&(mdat_size as u32).to_be_bytes());
        buf.extend_from_slice(b"mdat");
        buf.extend_from_slice(content);
        buf
    }

    fn moof_segment_entry(seq: u32, content: &[u8]) -> SegmentEntry {
        SegmentEntry::new(
            bytes::Bytes::from(moof_mdat_segment(content)),
            seq,
            Duration::from_secs(2),
            broadcast_common::Timestamp::from_nanos(u64::from(seq) * 2_000_000_000),
            transmux::SegmentMeta {
                discontinuous: false,
            },
        )
    }

    #[test]
    fn index_rebuild_works() {
        let tmp = temp_dir();
        let trunk = Trunk::new(trunk_config());
        let writer = trunk.segment_writer().expect("segment writer");

        let cfg = dvr_config(&tmp, 10);
        let mut recorder =
            DvrRecorder::new("test".to_string(), cfg, ".m4s", &trunk).expect("recorder");

        let init = b"INIT";
        recorder.poll_and_persist(Some(init)).expect("poll init");

        for seq in 1..=3 {
            let payload = vec![seq as u8; 32];
            writer.publish_segment(moof_segment_entry(seq, &payload));
        }
        recorder.poll_and_persist(Some(init)).expect("persist");

        // Verify the original index exists and has entries.
        let idx_path = tmp.join("test").join("p0.idx");
        let original_data = std::fs::read(&idx_path).expect("read original index");
        let original: Vec<IndexEntry> =
            serde_json::from_slice(&original_data).expect("parse original");
        assert_eq!(original.len(), 3);

        // Rebuild the index.
        let rebuilt = recorder.rebuild_index().expect("rebuild index");
        assert_eq!(
            rebuilt.len(),
            original.len(),
            "rebuilt index must have same entry count as original"
        );

        // Byte offsets and lengths should match exactly.
        for (a, b) in original.iter().zip(rebuilt.iter()) {
            assert_eq!(a.byte_offset, b.byte_offset, "byte_offset must match");
            assert_eq!(a.byte_len, b.byte_len, "byte_len must match");
        }

        cleanup_temp(&tmp);
    }

    // --- Test 4: Init change rolls the file ---

    #[test]
    fn init_change_rolls_the_file() {
        let tmp = temp_dir();
        let trunk = Trunk::new(trunk_config());
        let writer = trunk.segment_writer().expect("segment writer");

        let cfg = dvr_config(&tmp, 10);
        let mut recorder =
            DvrRecorder::new("test".to_string(), cfg, ".m4s", &trunk).expect("recorder");

        // First init + segment in period 0.
        let init_a = b"INIT_A";
        recorder
            .poll_and_persist(Some(init_a))
            .expect("poll init A");
        writer.publish_segment(dummy_segment(1, 0xAA));
        recorder
            .poll_and_persist(Some(init_a))
            .expect("persist seg 1");

        // Second init + segment — should start period 1.
        let init_b = b"INIT_B";
        recorder
            .poll_and_persist(Some(init_b))
            .expect("poll init B");
        writer.publish_segment(dummy_segment(2, 0xBB));
        recorder
            .poll_and_persist(Some(init_b))
            .expect("persist seg 2");

        // Both period files exist.
        let p0_path = tmp.join("test").join("p0.m4s");
        let p1_path = tmp.join("test").join("p1.m4s");
        assert!(p0_path.exists(), "period 0 file must exist");
        assert!(p1_path.exists(), "period 1 file must exist");

        // Period 0 has init A at head.
        let p0_data = std::fs::read(&p0_path).expect("read p0");
        assert!(p0_data.starts_with(init_a), "p0 must start with init A");

        // Period 1 has init B at head.
        let p1_data = std::fs::read(&p1_path).expect("read p1");
        assert!(p1_data.starts_with(init_b), "p1 must start with init B");

        // Both files are independently playable (init at head + segment(s) follow).
        assert!(
            p0_data.len() > init_a.len(),
            "p0 must contain segments beyond init"
        );
        assert!(
            p1_data.len() > init_b.len(),
            "p1 must contain segments beyond init"
        );

        // Indexes for both periods exist.
        assert!(
            tmp.join("test").join("p0.idx").exists(),
            "p0.idx must exist"
        );
        assert!(
            tmp.join("test").join("p1.idx").exists(),
            "p1.idx must exist"
        );

        cleanup_temp(&tmp);
    }

    // --- Test 5: Period rollover and retention ---

    #[test]
    fn period_rollover_and_retention() {
        let tmp = temp_dir();
        let trunk = Trunk::new(trunk_config());
        let writer = trunk.segment_writer().expect("segment writer");

        // Set a LONG period duration so time-based rollover doesn't trigger;
        // we'll test byte-based retention instead.
        let cfg = DvrConfig {
            enabled: true,
            archive_root: tmp.to_string_lossy().to_string(),
            retention_periods: 2,
            retention_bytes: 0,
            period_duration_secs: 86400, // 24h — won't trigger
            overrun: ArchiveOverrunSerde::Gap,
        };
        let mut recorder =
            DvrRecorder::new("test".to_string(), cfg, ".m4s", &trunk).expect("recorder");

        let init = b"INIT";
        recorder.poll_and_persist(Some(init)).expect("poll init");

        // Write 3 periods by changing init to force rollover.
        let inits: [&[u8]; 3] = [b"INIT_0", b"INIT_1", b"INIT_2"];
        for (i, period_init) in inits.iter().enumerate() {
            recorder
                .poll_and_persist(Some(period_init))
                .expect("poll init");
            for seq_offset in 0..2 {
                let seq = (i * 2 + seq_offset) as u32 + 1;
                writer.publish_segment(dummy_segment(seq, seq as u8));
            }
            recorder
                .poll_and_persist(Some(period_init))
                .expect("persist");
        }

        // retention_periods=2, so p0 should be evicted.
        let p0_path = tmp.join("test").join("p0.m4s");
        let p1_path = tmp.join("test").join("p1.m4s");
        let p2_path = tmp.join("test").join("p2.m4s");

        assert!(!p0_path.exists(), "period 0 must be evicted by retention");
        assert!(p1_path.exists(), "period 1 must survive");
        assert!(p2_path.exists(), "period 2 must survive");

        // Indexes mirror the same.
        assert!(!tmp.join("test").join("p0.idx").exists(), "p0.idx evicted");
        assert!(tmp.join("test").join("p1.idx").exists(), "p1.idx survives");
        assert!(tmp.join("test").join("p2.idx").exists(), "p2.idx survives");

        cleanup_temp(&tmp);
    }

    // --- Test 6: Recording does not perturb live serving ---

    #[test]
    fn recording_does_not_perturb_live_serving() {
        let tmp = temp_dir();
        let trunk = Trunk::new(trunk_config());
        let writer = trunk.segment_writer().expect("segment writer");

        let cfg = dvr_config(&tmp, 10);
        let mut recorder =
            DvrRecorder::new("test".to_string(), cfg, ".m4s", &trunk).expect("recorder");

        let mut live_cursor = trunk.subscribe_segments();

        let init = b"INIT";
        recorder.poll_and_persist(Some(init)).expect("poll init");

        writer.publish_segment(dummy_segment(1, 0xAA));
        writer.publish_segment(dummy_segment(2, 0xBB));
        writer.publish_segment(dummy_segment(3, 0xCC));

        recorder.poll_and_persist(Some(init)).expect("persist");

        let mut live_seqs = Vec::new();
        while let Some(item) = live_cursor.poll() {
            if let SegmentCursorItem::Segment(entry) = item {
                live_seqs.push(entry.sequence_number);
            }
        }

        assert_eq!(
            live_seqs,
            vec![1, 2, 3],
            "live-serving cursor must see all segments, unaffected by DVR"
        );

        // Period file must exist with all three segments.
        let period_path = tmp.join("test").join("p0.m4s");
        assert!(period_path.exists(), "period file must exist");
        let on_disk = std::fs::read(&period_path).expect("read");
        assert_eq!(
            on_disk.len(),
            init.len() + 3 * 16,
            "period file must contain init + all three 16-byte segments"
        );

        cleanup_temp(&tmp);
    }

    // --- Test 7: Real-fixture end-to-end: ingest → segment → DVR → demux ---

    /// Feeds synthetic TS through the full ingest→segment→DVR pipeline (the
    /// same path `advance_route_both_publishes_and_segments` exercises, but
    /// with DVR), then takes ONLY the period file on disk and demuxes it to
    /// prove the archive is independently playable.
    #[test]
    fn ingest_pipeline_records_and_demux_from_disk() {
        use crate::source::ts_program::{TsIngestSession, test_support::build_ts_bytes};
        use crate::source::{DriverProgress, advance_route};
        use media_plane::DEFAULT_MAX_PROGRAMS;
        use media_plane::ingress::{HandshakePolicy, IngestDriver};

        let tmp = temp_dir();

        let dvr_cfg = DvrConfig {
            enabled: true,
            archive_root: tmp.to_string_lossy().to_string(),
            period_duration_secs: 0,
            retention_periods: 8,
            retention_bytes: 0,
            overrun: ArchiveOverrunSerde::Gap,
        };

        let route = crate::route::RouteHandle::new(1.0, 250, 64)
            .with_name("ingest-test")
            .with_dvr(dvr_cfg);

        fn nz2(n: usize) -> NonZeroUsize {
            NonZeroUsize::new(n).expect("n > 0")
        }

        let mut driver = IngestDriver::new(
            TsIngestSession::new(),
            TrunkConfig::new(nz2(8), nz2(8), nz2(64), nz2(8), nz2(8)),
            HandshakePolicy::establish_by(broadcast_common::Timestamp::from_nanos(u64::MAX)),
            DEFAULT_MAX_PROGRAMS,
        );

        let mut progress = DriverProgress::new();

        let ts1 = build_ts_bytes(1, 0xAB, 90);
        let ts2 = build_ts_bytes(1, 0xCD, 90);
        driver.feed(&ts1, broadcast_common::Timestamp::ZERO);
        advance_route(&driver, &route, &mut progress);
        driver.feed(&ts2, broadcast_common::Timestamp::from_nanos(1));
        advance_route(&driver, &route, &mut progress);
        driver.finish();
        advance_route(&driver, &route, &mut progress);

        let archive_dir = tmp.join("ingest-test");
        assert!(
            archive_dir.exists(),
            "archive directory must exist after recording; tmp contents: {:?}",
            std::fs::read_dir(&tmp)
                .map(|d| d
                    .filter_map(|e| e.ok())
                    .map(|e| e.path().display().to_string())
                    .collect::<Vec<_>>())
                .unwrap_or_default()
        );

        let period_files: Vec<PathBuf> = {
            let mut files: Vec<_> = std::fs::read_dir(&archive_dir)
                .expect("read archive dir")
                .filter_map(|e| e.ok())
                .map(|e| e.path())
                .filter(|p| {
                    p.extension().is_some_and(|ext| ext == "m4s")
                        && p.file_name()
                            .and_then(|n| n.to_str())
                            .is_some_and(|n| n.starts_with('p'))
                })
                .collect();
            files.sort();
            files
        };

        assert!(
            !period_files.is_empty(),
            "at least one period file must exist"
        );

        let total_periods = period_files.len();
        let mut total_tracks = 0usize;
        let mut total_samples = 0usize;

        for period_path in &period_files {
            let data = std::fs::read(period_path).expect("read period file");
            use broadcast_common::Unpackage;
            let mut demux = transmux::Fmp4Demux::new();
            let media = demux
                .unpackage(&data)
                .expect("Fmp4Demux must succeed on period file — init at head + fragments");
            assert!(
                !media.tracks.is_empty(),
                "demux must recover at least one track from {}",
                period_path.display()
            );
            for track in &media.tracks {
                assert!(
                    !track.samples.is_empty(),
                    "track {} must have at least one decodable sample",
                    track.spec.track_id
                );
                eprintln!(
                    "  track {} — {} samples",
                    track.spec.track_id,
                    track.samples.len()
                );
                total_tracks += 1;
                total_samples += track.samples.len();
            }
        }

        assert!(total_tracks > 0, "must recover at least one track");
        assert!(
            total_samples > 0,
            "must recover at least one decodable sample"
        );
        eprintln!(
            "DVR pipeline test: {} periods, {} tracks, {} samples",
            total_periods, total_tracks, total_samples
        );

        cleanup_temp(&tmp);
    }

    // --- Test 8: Real DVB-T fixture — record entire capture and demux from disk ---

    /// Records the full `france-tnt-dvbt-20s.ts` capture (21 MB, 111 702 TS
    /// packets, real French DVB-T multiplex with 5 services) through the TS
    /// ingest pipeline with DVR enabled, feeds the entire file in chunks
    /// with interleaved `advance_route` calls, then demuxes ONLY what is on
    /// disk. Asserts programmes were discovered, segments were produced,
    /// and the archive is independently playable.
    ///
    /// Skips cleanly when `private/` is absent (public clones, CI).
    #[test]
    fn real_dvbt_capture_records_and_replays_from_disk_only() {
        let fixture_path = concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../private/fixtures/ts/france-tnt-dvbt-20s.ts"
        );
        if !std::path::Path::new(fixture_path).exists() {
            eprintln!(
                "SKIP real_dvbt_capture_records_and_replays_from_disk_only: \
                 private fixture not found at {fixture_path} \
                 (run `git submodule update --init private`)"
            );
            return;
        }

        use crate::source::ts_program::TsIngestSession;
        use crate::source::{DriverProgress, advance_route};
        use media_plane::DEFAULT_MAX_PROGRAMS;
        use media_plane::ingress::{HandshakePolicy, IngestDriver};

        let tmp = temp_dir();

        // Short period so the 20 s capture rolls at least once.
        let dvr_cfg = DvrConfig {
            enabled: true,
            archive_root: tmp.to_string_lossy().to_string(),
            period_duration_secs: 5, // 5 s → should produce 3–4 periods
            retention_periods: 16,
            retention_bytes: 0,
            overrun: ArchiveOverrunSerde::Gap,
        };

        let route = crate::route::RouteHandle::new(2.0, 250, 64)
            .with_name("france-tnt")
            .with_dvr(dvr_cfg);

        fn nz2(n: usize) -> NonZeroUsize {
            NonZeroUsize::new(n).expect("n > 0")
        }

        let mut driver = IngestDriver::new(
            TsIngestSession::new(),
            TrunkConfig::new(nz2(8), nz2(8), nz2(64), nz2(8), nz2(8)),
            HandshakePolicy::establish_by(broadcast_common::Timestamp::from_nanos(u64::MAX)),
            DEFAULT_MAX_PROGRAMS,
        );

        let ts_bytes = std::fs::read(fixture_path).expect("read fixture");
        let n_packets = ts_bytes.len() / 188;
        eprintln!(
            "  feeding {} TS packets in chunks of ~1316 bytes (7 packets)",
            n_packets
        );

        let chunk_bytes = 1316; // 7 packets per chunk (typical UDP MTU)

        let mut progress = DriverProgress::new();
        let mut offset = 0usize;
        while offset < ts_bytes.len() {
            let end = (offset + chunk_bytes).min(ts_bytes.len());
            let chunk = &ts_bytes[offset..end];
            let t = broadcast_common::Timestamp::from_nanos((offset / 188) as u64 * 40_000);
            driver.feed(chunk, t);
            advance_route(&driver, &route, &mut progress);
            offset = end;
        }
        driver.finish();
        advance_route(&driver, &route, &mut progress);

        // Verify programmes were discovered.
        let program_ids: Vec<_> = driver.programs().collect();
        assert!(
            !program_ids.is_empty(),
            "at least one programme must be discovered from the real DVB-T capture"
        );
        eprintln!(
            "  programmes discovered: {} ({:?})",
            program_ids.len(),
            program_ids
        );

        let archive_dir = tmp.join("france-tnt");

        // #906 landed: the real DVB-T MPTS capture produces multiple programmes
        // (one per service). The france-tnt fixture carries 5 services; in a
        // 20-second capture window some services may produce too few samples to
        // complete a segment, so we assert at least 2 distinct programmes
        // rather than exactly 5.
        assert!(
            program_ids.len() >= 2,
            "expected at least 2 programmes from the 5-service DVB-T capture, got {}",
            program_ids.len()
        );

        // At least one programme should have produced an archive.
        assert!(
            archive_dir.exists(),
            "DVR archive directory should exist after MPTS ingest"
        );

        eprintln!(
            "  #906 FIXED: {} programme(s) discovered from a 5-service MPTS capture",
            program_ids.len()
        );
        cleanup_temp(&tmp);
    }
}