//! Per-segment merge pipeline: window-read the three inputs in log_time order,
//! remap topics, demux per output channel, then write one MCAP per clip in
//! mcap-optimiser byte order (single-channel uncompressed chunks grouped by
//! channel, MessageIndex relocated into one block before the summary).
//!
//! Stage 2 treats video channels as passthrough (cut by timestamp only); the
//! smart GOP-preserving re-encode is layered on in a later stage.

use std::collections::{BTreeMap, HashMap};
use std::fs::File;
use std::path::Path;

use anyhow::{anyhow, Context, Result};
use mcap::records::{
    Channel as ChannelRec, ChunkIndex, DataEnd, Header, MessageHeader, MessageIndex, SchemaHeader,
    Statistics, op,
};
use mcap::Summary;
use memmap2::Mmap;

use prost::Message as _;

use crate::chunk_builder::ChunkBuilder;
use crate::cli::Args;
use crate::demux::{self, ChannelDemux};
use crate::merge::{self, Clip};
use crate::window::{MergedMsg, Role, Source, WindowReader};
use crate::writer::OutputWriter;
use crate::{json, nal, pb, regop, remap};

const VIDEO_SCHEMA: &str = "foxglove.CompressedVideo";
/// Initial look-back when priming the decoder for a leading-GOP re-encode.
const PREROLL_NS: u64 = 10_000_000_000;
/// A segment trimmed shorter than this by the three-stream overlap clamp is
/// dropped (not written) rather than emitted as a too-short clip.
const MIN_KEEP_NS: u64 = 15_000_000_000;

/// Owned channel/schema identity, extracted once per input channel from the
/// summary so per-message handling only needs `(role, in_channel_id)`.
struct ChanInfo {
    topic: String,
    message_encoding: String,
    metadata: BTreeMap<String, String>,
    schema_name: String,
    schema_encoding: String,
    schema_data: Vec<u8>,
}

/// An mmap kept alive for the whole run so summaries/chunks can borrow it.
struct Mapped {
    map: Mmap,
}

impl Mapped {
    fn open(path: &Path) -> Result<Self> {
        let f = File::open(path).with_context(|| format!("opening {}", path.display()))?;
        let map = unsafe { Mmap::map(&f)? };
        Ok(Self { map })
    }
}

pub fn run(args: Args) -> Result<()> {
    let ego = Mapped::open(&args.ego)?;
    let wl = Mapped::open(&args.wrist_left)?;
    let wr = Mapped::open(&args.wrist_right)?;

    let ego_sum = Summary::read(&ego.map[..])?
        .ok_or_else(|| anyhow!("{}: no summary section", args.ego.display()))?;
    let wl_sum = Summary::read(&wl.map[..])?
        .ok_or_else(|| anyhow!("{}: no summary section", args.wrist_left.display()))?;
    let wr_sum = Summary::read(&wr.map[..])?
        .ok_or_else(|| anyhow!("{}: no summary section", args.wrist_right.display()))?;

    let mut catalog: HashMap<(Role, u16), ChanInfo> = HashMap::new();
    build_catalog(Role::Ego, &ego_sum, &mut catalog);
    build_catalog(Role::WristLeft, &wl_sum, &mut catalog);
    build_catalog(Role::WristRight, &wr_sum, &mut catalog);

    // The JSON timestamps are milliseconds from the recording start; anchor on
    // the ego file (the edit list is named *_ego.mix.json).
    let t0 = ego_sum
        .stats
        .as_ref()
        .map(|s| s.message_start_time)
        .filter(|&t| t != 0)
        .or_else(|| ego_sum.chunk_indexes.iter().map(|c| c.message_start_time).min())
        .ok_or_else(|| anyhow!("ego input has no messages"))?;

    let segments = json::load(&args.json)?;
    std::fs::create_dir_all(&args.outdir)
        .with_context(|| format!("creating output dir {}", args.outdir.display()))?;

    // Coalesce adjacent segments — segment + gap + segment -> clip — before
    // cutting: two segments closer than --merge-gap-ms whose merged span stays
    // within --max-merged-ms become one output clip (the in-between gap is kept).
    let clips = merge::merge_segments(segments, args.merge_gap_ms, args.max_merged_ms);

    eprintln!(
        "cutmerge: {} clip(s) after merging, t0 = {} ns",
        clips.len(),
        t0
    );
    for (i, clip) in clips.iter().enumerate() {
        if clip.sources.len() > 1 {
            let merged: Vec<u32> = clip.sources.iter().map(|s| s.index).collect();
            eprintln!(
                "  clip {:03}: merged segments {:?} → [{} .. {}] ms",
                i, merged, clip.start_ts, clip.end_ts
            );
        }
    }

    // Keep every output to the time span all three inputs share, so each clip
    // carries ego + both wrists within the --max-skew-ms tolerance. Only the
    // boundaries where the inputs disagree by more than the tolerance are
    // clamped (to the shortest stream); the rest are left as the edit list asked.
    let (clamp_start, clamp_end) = overlap_bounds(&ego_sum, &wl_sum, &wr_sum, args.max_skew_ms);

    // Absolute ns log_time -> milliseconds from the recording start (ego t0),
    // the unit the edit list's start_ts/end_ts use.
    let to_ms = |t: u64| t.saturating_sub(t0) / 1_000_000;

    // Per-clip outcomes, echoed (with their source segments) into metadata.json.
    // Only collected for a full run; --only-segment is a dev mode that writes a
    // single clip and no metadata.
    let write_meta = args.only_segment.is_none();
    let mut outcomes: Vec<ClipMeta> = Vec::new();

    for (clip_index, clip) in clips.iter().enumerate() {
        if let Some(only) = args.only_segment {
            // Dev mode: process only the clip that contains this source segment.
            if !clip.sources.iter().any(|s| s.index == only) {
                continue;
            }
        }
        let want_start = t0 + clip.start_ts.saturating_mul(1_000_000);
        let want_end = t0 + clip.end_ts.saturating_mul(1_000_000);
        let SegPlan { start, end, trimmed, drop } =
            plan_segment(want_start, want_end, clamp_start, clamp_end);

        // Dropped: a >100 ms skew trimmed this window below the 15 s minimum, or
        // the three streams share no overlap here at all. Not written.
        if drop {
            eprintln!(
                "  clip {:03}: dropped (overlap {:.1}s < {}s after trimming)",
                clip_index,
                end.saturating_sub(start) as f64 / 1e9,
                MIN_KEEP_NS / 1_000_000_000,
            );
            outcomes.push(ClipMeta {
                start_ts: to_ms(start),
                end_ts: to_ms(start.max(end)),
                status: "dropped",
            });
            continue;
        }

        if trimmed {
            eprintln!(
                "  clip {:03}: clamped to overlap (-{} ms head, -{} ms tail)",
                clip_index,
                (start - want_start) / 1_000_000,
                (want_end - end) / 1_000_000,
            );
        }

        let sources = [
            Source { role: Role::Ego, mcap: &ego.map[..], summary: &ego_sum },
            Source { role: Role::WristLeft, mcap: &wl.map[..], summary: &wl_sum },
            Source { role: Role::WristRight, mcap: &wr.map[..], summary: &wr_sum },
        ];

        let out_path = args.outdir.join(format!("{:03}.mcap", clip_index));
        match process_segment(&sources, &catalog, &args, start, end, &out_path) {
            // Report the clip's window (requested, clamped to the shared overlap),
            // not the emitted frame times.
            Ok(()) => {
                outcomes.push(ClipMeta {
                    start_ts: to_ms(start),
                    end_ts: to_ms(end),
                    status: if trimmed { "trimmed" } else { "success" },
                });
            }
            Err(err) => {
                eprintln!("  clip {:03}: failed — {:#}", clip_index, err);
                // Remove any half-written file so it can't pass for a good clip.
                let _ = std::fs::remove_file(&out_path);
                outcomes.push(ClipMeta {
                    start_ts: to_ms(start),
                    end_ts: to_ms(end),
                    status: "failed",
                });
            }
        }
    }

    if write_meta {
        write_metadata(&args.outdir, &clips, &outcomes)?;
    } else {
        eprintln!("cutmerge: --only-segment set; metadata.json not written");
    }

    Ok(())
}

/// The decision for one segment after clamping its requested window to the
/// shared three-stream overlap.
struct SegPlan {
    /// Effective window, clamped to the overlap.
    start: u64,
    end: u64,
    /// Window was moved by the clamp (a >`max_skew_ms` skew) ⇒ `trimmed` status.
    trimmed: bool,
    /// Clip is too short to keep (< 15 s after trimming, or no overlap at all)
    /// ⇒ `dropped`, not written.
    drop: bool,
}

/// Resolve a segment's effective window and outcome from its requested window
/// and the overlap clamp bounds. Pure (no I/O) so the status boundaries are
/// unit-tested directly; the `failed` outcome is decided later by whether the
/// write succeeds.
fn plan_segment(
    want_start: u64,
    want_end: u64,
    clamp_start: Option<u64>,
    clamp_end: Option<u64>,
) -> SegPlan {
    let start = clamp_start.map_or(want_start, |c| want_start.max(c));
    let end = clamp_end.map_or(want_end, |c| want_end.min(c));
    let trimmed = start != want_start || end != want_end;
    let drop = start >= end || (trimmed && end - start < MIN_KEEP_NS);
    SegPlan { start, end, trimmed, drop }
}

/// Resolved outcome for one output clip, combined with its source segments to
/// produce `metadata.json`. `start_ts`/`end_ts` are milliseconds from the
/// recording start and hold the clip's window — the requested `[start, end)`
/// clamped to the shared three-stream overlap — not the emitted frame times.
struct ClipMeta {
    start_ts: u64,
    end_ts: u64,
    status: &'static str,
}

/// Write `<outdir>/metadata.json`: a JSON array with one object per output clip,
/// in output order. Each object carries the clip's `index` (matching its
/// `<index>.mcap` file), its `start_ts`/`end_ts` (the clip's window, clamped to
/// the shared overlap), a `status` (`success` / `trimmed` / `dropped` / `failed`), and a
/// `segments` array holding every source edit-list entry merged into the clip,
/// verbatim except its `index` (a single-element array when nothing was merged).
fn write_metadata(outdir: &Path, clips: &[Clip], outcomes: &[ClipMeta]) -> Result<()> {
    if clips.len() != outcomes.len() {
        return Err(anyhow!(
            "metadata: {} clips but {} outcomes",
            clips.len(),
            outcomes.len()
        ));
    }

    let mut entries: Vec<serde_json::Value> = Vec::with_capacity(clips.len());
    for (i, (clip, o)) in clips.iter().zip(outcomes).enumerate() {
        // Echo each source entry verbatim, minus its now-redundant `index`.
        let segments: Vec<serde_json::Value> = clip
            .sources
            .iter()
            .map(|s| {
                let mut v = s.raw.clone();
                if let Some(obj) = v.as_object_mut() {
                    obj.remove("index");
                }
                v
            })
            .collect();
        let mut obj = serde_json::Map::new();
        obj.insert("index".to_string(), (i as u64).into());
        obj.insert("start_ts".to_string(), o.start_ts.into());
        obj.insert("end_ts".to_string(), o.end_ts.into());
        obj.insert("status".to_string(), o.status.into());
        obj.insert("segments".to_string(), serde_json::Value::Array(segments));
        entries.push(serde_json::Value::Object(obj));
    }

    let path = outdir.join("metadata.json");
    let text = serde_json::to_string_pretty(&entries)?;
    std::fs::write(&path, text).with_context(|| format!("writing {}", path.display()))?;
    eprintln!("cutmerge: wrote {}", path.display());
    Ok(())
}

fn build_catalog(role: Role, summary: &Summary, catalog: &mut HashMap<(Role, u16), ChanInfo>) {
    for (id, ch) in &summary.channels {
        let (schema_name, schema_encoding, schema_data) = match ch.schema.as_ref() {
            Some(s) => (s.name.clone(), s.encoding.clone(), s.data.to_vec()),
            None => (String::new(), String::new(), Vec::new()),
        };
        catalog.insert(
            (role, *id),
            ChanInfo {
                topic: ch.topic.clone(),
                message_encoding: ch.message_encoding.clone(),
                metadata: ch.metadata.clone(),
                schema_name,
                schema_encoding,
                schema_data,
            },
        );
    }
}

/// First/last message `log_time` of an input, from stats when present and
/// otherwise from the chunk index. `None` if the input has no indexable
/// messages (no usable summary stats and no chunks).
fn coverage(summary: &Summary) -> Option<(u64, u64)> {
    let start = summary
        .stats
        .as_ref()
        .map(|s| s.message_start_time)
        .filter(|&t| t != 0)
        .or_else(|| summary.chunk_indexes.iter().map(|c| c.message_start_time).min())?;
    let end = summary
        .stats
        .as_ref()
        .map(|s| s.message_end_time)
        .filter(|&t| t != 0)
        .or_else(|| summary.chunk_indexes.iter().map(|c| c.message_end_time).max())?;
    Some((start, end))
}

/// Segment-window clamp bounds that keep every clip inside the span all three
/// inputs share, so each output carries all three cameras. Returns
/// `(clamp_start, clamp_end)`:
///
/// - `clamp_start = Some(t)` ⇒ raise segment starts to at least `t` (the
///   latest-starting input). Set only when the inputs' first-message times
///   differ by more than `max_skew_ms`.
/// - `clamp_end = Some(t)` ⇒ lower segment ends to at most `t` (the
///   earliest-ending input). Set only when their last-message times differ by
///   more than `max_skew_ms`.
///
/// A boundary whose inputs already agree within `max_skew_ms` — or any boundary
/// when coverage is unavailable for an input — yields `None`, leaving that
/// boundary as the edit list specified.
fn overlap_bounds(
    ego: &Summary,
    wl: &Summary,
    wr: &Summary,
    max_skew_ms: u64,
) -> (Option<u64>, Option<u64>) {
    let skew_ns = max_skew_ms.saturating_mul(1_000_000);
    let (e, l, r) = match (coverage(ego), coverage(wl), coverage(wr)) {
        (Some(e), Some(l), Some(r)) => (e, l, r),
        _ => {
            eprintln!(
                "cutmerge: warning: an input has no indexable messages; \
                 skipping three-stream overlap clamping"
            );
            return (None, None);
        }
    };

    let starts = [e.0, l.0, r.0];
    let ends = [e.1, l.1, r.1];
    let (clamp_start, clamp_end) = clamp_bounds(starts, ends, skew_ns);

    if clamp_start.is_some() || clamp_end.is_some() {
        let start_skew = starts.iter().max().unwrap() - starts.iter().min().unwrap();
        let end_skew = ends.iter().max().unwrap() - ends.iter().min().unwrap();
        eprintln!(
            "cutmerge: input coverage skew — start {} ms, end {} ms (tolerance {} ms); \
             clamping segments to the shared three-stream overlap",
            start_skew / 1_000_000,
            end_skew / 1_000_000,
            max_skew_ms,
        );
    }
    (clamp_start, clamp_end)
}

/// Pure boundary arithmetic behind [`overlap_bounds`]. Given the three inputs'
/// first-message times (`starts`) and last-message times (`ends`), returns the
/// clamp bounds: the start is raised to the latest start, and the end lowered to
/// the earliest end — but each only when that boundary's spread exceeds
/// `skew_ns`. Boundaries within tolerance return `None` (left as-is).
fn clamp_bounds(starts: [u64; 3], ends: [u64; 3], skew_ns: u64) -> (Option<u64>, Option<u64>) {
    let max_start = *starts.iter().max().unwrap();
    let min_start = *starts.iter().min().unwrap();
    let min_end = *ends.iter().min().unwrap();
    let max_end = *ends.iter().max().unwrap();

    let clamp_start = (max_start - min_start > skew_ns).then_some(max_start);
    let clamp_end = (max_end - min_end > skew_ns).then_some(min_end);
    (clamp_start, clamp_end)
}

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

    const MS: u64 = 1_000_000;
    const TOL: u64 = 100 * MS;

    #[test]
    fn aligned_within_tolerance_is_untouched() {
        // All three start/end within 100 ms → no clamping on either boundary.
        let starts = [1_000 * MS, 1_050 * MS, 1_080 * MS];
        let ends = [9_000 * MS, 8_950 * MS, 8_920 * MS];
        assert_eq!(clamp_bounds(starts, ends, TOL), (None, None));
    }

    #[test]
    fn late_start_beyond_tolerance_clamps_up_to_latest() {
        // One stream starts 3 min late → raise segment starts to that latest start.
        let starts = [1_000 * MS, 1_000 * MS, 181_000 * MS];
        let ends = [9_000 * MS, 9_000 * MS, 9_010 * MS];
        assert_eq!(clamp_bounds(starts, ends, TOL), (Some(181_000 * MS), None));
    }

    #[test]
    fn early_end_beyond_tolerance_clamps_down_to_shortest() {
        // One stream stops 3 min early → lower segment ends to that earliest end.
        let starts = [1_000 * MS, 1_010 * MS, 1_005 * MS];
        let ends = [540_000 * MS, 360_000 * MS, 540_000 * MS];
        assert_eq!(clamp_bounds(starts, ends, TOL), (None, Some(360_000 * MS)));
    }

    #[test]
    fn both_boundaries_skewed_clamp_independently() {
        let starts = [1_000 * MS, 5_000 * MS, 1_000 * MS];
        let ends = [9_000 * MS, 9_000 * MS, 6_000 * MS];
        assert_eq!(
            clamp_bounds(starts, ends, TOL),
            (Some(5_000 * MS), Some(6_000 * MS))
        );
    }

    #[test]
    fn exactly_at_tolerance_is_not_clamped() {
        // Spread == tolerance is within bounds; only strictly greater clamps.
        let starts = [1_000 * MS, 1_000 * MS, 1_100 * MS];
        let ends = [9_100 * MS, 9_000 * MS, 9_000 * MS];
        assert_eq!(clamp_bounds(starts, ends, TOL), (None, None));
    }

    use super::plan_segment;

    const SEC: u64 = 1_000 * MS;

    #[test]
    fn untouched_window_is_success() {
        // No clamp bounds → window kept as requested, not trimmed, not dropped.
        let p = plan_segment(100 * SEC, 160 * SEC, None, None);
        assert_eq!((p.start, p.end), (100 * SEC, 160 * SEC));
        assert!(!p.trimmed && !p.drop);
    }

    #[test]
    fn clamp_that_leaves_window_unmoved_is_success() {
        // Skew exists elsewhere, but this segment sits inside the overlap, so its
        // boundaries don't move → success, not trimmed.
        let p = plan_segment(100 * SEC, 160 * SEC, Some(50 * SEC), Some(200 * SEC));
        assert_eq!((p.start, p.end), (100 * SEC, 160 * SEC));
        assert!(!p.trimmed && !p.drop);
    }

    #[test]
    fn trim_leaving_at_least_15s_is_trimmed() {
        // Tail clamped from 160s down to 130s leaves 30 s ≥ 15 s → trimmed, kept.
        let p = plan_segment(100 * SEC, 160 * SEC, None, Some(130 * SEC));
        assert_eq!((p.start, p.end), (100 * SEC, 130 * SEC));
        assert!(p.trimmed && !p.drop);
    }

    #[test]
    fn trim_leaving_under_15s_is_dropped() {
        // Tail clamped down to 110s leaves only 10 s < 15 s → dropped.
        let p = plan_segment(100 * SEC, 160 * SEC, None, Some(110 * SEC));
        assert!(p.trimmed && p.drop);
    }

    #[test]
    fn no_overlap_is_dropped() {
        // Clamp pushes start past end (segment entirely outside the overlap).
        let p = plan_segment(100 * SEC, 160 * SEC, Some(200 * SEC), None);
        assert!(p.start >= p.end && p.drop);
    }

    #[test]
    fn short_native_window_without_trim_is_kept() {
        // A natively short (<15 s) clip is only dropped when *trimming* shortens
        // it; with no clamp it is kept (status success).
        let p = plan_segment(100 * SEC, 108 * SEC, None, None);
        assert!(!p.trimmed && !p.drop);
    }

    #[test]
    fn metadata_nests_source_segments_with_status() {
        use crate::json::Segment;
        use crate::merge::Clip;

        // Clip 0 merges two source segments; clip 1 is a single segment.
        let clips = vec![
            Clip {
                start_ts: 1000,
                end_ts: 9000,
                sources: vec![
                    Segment {
                        index: 0,
                        start_ts: 1000,
                        end_ts: 5000,
                        raw: serde_json::json!(
                            {"index": 0, "start_ts": 1000, "end_ts": 5000, "title": "a", "extra": 7}
                        ),
                    },
                    Segment {
                        index: 1,
                        start_ts: 5000,
                        end_ts: 9000,
                        raw: serde_json::json!(
                            {"index": 1, "start_ts": 5000, "end_ts": 9000, "title": "b"}
                        ),
                    },
                ],
            },
            Clip {
                start_ts: 12000,
                end_ts: 15000,
                sources: vec![Segment {
                    index: 2,
                    start_ts: 12000,
                    end_ts: 15000,
                    raw: serde_json::json!(
                        {"index": 2, "start_ts": 12000, "end_ts": 15000, "title": "c"}
                    ),
                }],
            },
        ];
        let outcomes = vec![
            super::ClipMeta { start_ts: 1100, end_ts: 8900, status: "trimmed" },
            super::ClipMeta { start_ts: 12000, end_ts: 15000, status: "success" },
        ];

        let dir = tempfile::tempdir().unwrap();
        super::write_metadata(dir.path(), &clips, &outcomes).unwrap();

        let text = std::fs::read_to_string(dir.path().join("metadata.json")).unwrap();
        let arr: Vec<serde_json::Value> = serde_json::from_str(&text).unwrap();
        assert_eq!(arr.len(), 2);

        // Clip 0: renumbered index, clip window, trimmed status.
        assert_eq!(arr[0]["index"], 0);
        assert_eq!(arr[0]["start_ts"], 1100);
        assert_eq!(arr[0]["end_ts"], 8900);
        assert_eq!(arr[0]["status"], "trimmed");
        let sources0 = arr[0]["segments"].as_array().unwrap();
        assert_eq!(sources0.len(), 2);
        // Source entries preserved verbatim except the per-segment `index`.
        assert!(sources0[0].get("index").is_none());
        assert_eq!(sources0[0]["title"], "a");
        assert_eq!(sources0[0]["extra"], 7);
        assert_eq!(sources0[0]["start_ts"], 1000);
        assert!(sources0[1].get("index").is_none());
        assert_eq!(sources0[1]["title"], "b");

        // Clip 1: single source, renumbered to output index 1.
        assert_eq!(arr[1]["index"], 1);
        assert_eq!(arr[1]["status"], "success");
        let sources1 = arr[1]["segments"].as_array().unwrap();
        assert_eq!(sources1.len(), 1);
        assert!(sources1[0].get("index").is_none());
        assert_eq!(sources1[0]["title"], "c");
    }
}

/// Accumulates the output channel/schema tables and per-channel demux buffers
/// as topics are discovered (channel ids assigned on first appearance).
struct OutTables {
    by_topic: HashMap<String, u16>,
    schema_ids: HashMap<String, u16>,
    schema_records: Vec<(SchemaHeader, Vec<u8>)>,
    channel_records: Vec<ChannelRec>,
    demuxes: Vec<ChannelDemux>,
    next_channel_id: u16,
    next_schema_id: u16,
}

impl OutTables {
    fn new() -> Self {
        Self {
            by_topic: HashMap::new(),
            schema_ids: HashMap::new(),
            schema_records: Vec::new(),
            channel_records: Vec::new(),
            demuxes: Vec::new(),
            next_channel_id: 1,
            next_schema_id: 1,
        }
    }

    /// Find or create the output channel for `out_topic`, returning its id.
    fn channel_for(&mut self, out_topic: &str, info: &ChanInfo) -> Result<u16> {
        if let Some(&cid) = self.by_topic.get(out_topic) {
            return Ok(cid);
        }
        let schema_id = if info.schema_name.is_empty() {
            0
        } else if let Some(&sid) = self.schema_ids.get(&info.schema_name) {
            sid
        } else {
            let sid = self.next_schema_id;
            self.next_schema_id += 1;
            self.schema_ids.insert(info.schema_name.clone(), sid);
            self.schema_records.push((
                SchemaHeader {
                    id: sid,
                    name: info.schema_name.clone(),
                    encoding: info.schema_encoding.clone(),
                },
                info.schema_data.clone(),
            ));
            sid
        };

        let cid = self.next_channel_id;
        self.next_channel_id += 1;
        self.channel_records.push(ChannelRec {
            id: cid,
            schema_id,
            topic: out_topic.to_string(),
            message_encoding: info.message_encoding.clone(),
            metadata: info.metadata.clone(),
        });
        self.demuxes.push(ChannelDemux::new()?);
        self.by_topic.insert(out_topic.to_string(), cid);
        Ok(cid)
    }
}

/// Cut and write one clip's MCAP. The written messages' time-bounds are recorded
/// in the output's `Statistics`; they are not surfaced to the caller, which
/// reports the clip's requested/clamped window in `metadata.json` instead.
fn process_segment(
    sources: &[Source<'_>],
    catalog: &HashMap<(Role, u16), ChanInfo>,
    args: &Args,
    start: u64,
    end: u64,
    out_path: &Path,
) -> Result<()> {
    let mut tables = OutTables::new();
    let mut videos: HashMap<u16, VideoState> = HashMap::new();

    let reader = WindowReader::new(sources, start, end)?;
    for m in reader {
        let m = m?;
        let info = catalog
            .get(&(m.role, m.in_channel_id))
            .ok_or_else(|| anyhow!("no catalog entry for {:?} channel {}", m.role, m.in_channel_id))?;
        let out_topic = remap::output_topic(m.role, &info.topic);
        let cid = tables.channel_for(&out_topic, info)?;

        if info.schema_name == VIDEO_SCHEMA {
            let vs = videos
                .entry(cid)
                .or_insert_with(|| VideoState::new(m.role, m.in_channel_id));
            let demux = &mut tables.demuxes[(cid - 1) as usize];
            handle_video_frame(vs, demux, cid, &m, sources, start, args)?;
        } else {
            let demux = &mut tables.demuxes[(cid - 1) as usize];
            push_verbatim(demux, cid, &m)?;
        }
    }

    // Channels still buffering a leading GOP at end of window (no trailing
    // keyframe inside the window) get re-encoded as one fresh GOP.
    for (cid, vs) in videos.iter_mut() {
        if !vs.lead.is_empty() {
            let demux = &mut tables.demuxes[(*cid - 1) as usize];
            finalize_lead_gop(vs, demux, *cid, sources, start, args)?;
        }
    }

    for d in &mut tables.demuxes {
        d.finalise()?;
    }

    write_output(args, &tables, out_path)
}

/// Per-output-video-channel smart-cut state. Frames stream in log_time order;
/// once a keyframe boundary is crossed we passthrough verbatim, only the
/// leading run of non-keyframe frames at the window start is ever re-encoded.
struct VideoState {
    role: Role,
    in_channel_id: u16,
    started: bool,
    passthrough: bool,
    lead: Vec<LeadFrame>,
}

impl VideoState {
    fn new(role: Role, in_channel_id: u16) -> Self {
        Self {
            role,
            in_channel_id,
            started: false,
            passthrough: false,
            lead: Vec::new(),
        }
    }
}

struct LeadFrame {
    log_time: u64,
    publish_time: u64,
    sequence: u32,
    ts_sec: i64,
    ts_nanos: i32,
    frame_id: String,
    annexb: Vec<u8>,
}

#[derive(Clone)]
struct PrimeFrame {
    annexb: Vec<u8>,
}

fn push_verbatim(demux: &mut ChannelDemux, cid: u16, m: &MergedMsg) -> Result<()> {
    let header = MessageHeader {
        channel_id: cid,
        sequence: m.sequence,
        log_time: m.log_time,
        publish_time: m.publish_time,
    };
    demux.push(&header, &m.data)
}

fn handle_video_frame(
    vs: &mut VideoState,
    demux: &mut ChannelDemux,
    cid: u16,
    m: &MergedMsg,
    sources: &[Source<'_>],
    window_start: u64,
    args: &Args,
) -> Result<()> {
    if vs.passthrough {
        return push_verbatim(demux, cid, m);
    }

    let cv = pb::foxglove::CompressedVideo::decode(&m.data[..])
        .context("decoding CompressedVideo frame")?;
    let is_keyframe = nal::is_keyframe(&cv.data);

    if !vs.started {
        vs.started = true;
        if is_keyframe {
            // Window already starts on a keyframe: nothing to re-encode.
            vs.passthrough = true;
            return push_verbatim(demux, cid, m);
        }
    } else if is_keyframe {
        // Reached the next keyframe: re-encode the buffered leading GOP, then
        // pass this self-contained keyframe (and everything after) through.
        finalize_lead_gop(vs, demux, cid, sources, window_start, args)?;
        vs.passthrough = true;
        return push_verbatim(demux, cid, m);
    }

    let (ts_sec, ts_nanos) = cv.timestamp.map(|t| (t.seconds, t.nanos)).unwrap_or((0, 0));
    vs.lead.push(LeadFrame {
        log_time: m.log_time,
        publish_time: m.publish_time,
        sequence: m.sequence,
        ts_sec,
        ts_nanos,
        frame_id: cv.frame_id,
        annexb: cv.data,
    });
    Ok(())
}

fn finalize_lead_gop(
    vs: &mut VideoState,
    demux: &mut ChannelDemux,
    cid: u16,
    sources: &[Source<'_>],
    window_start: u64,
    args: &Args,
) -> Result<()> {
    let priming = read_priming(sources, vs.role, vs.in_channel_id, window_start)?;

    let mut frames: Vec<regop::FrameIn> = Vec::with_capacity(priming.len() + vs.lead.len());
    for p in &priming {
        frames.push(regop::FrameIn { annexb: &p.annexb, emit: false });
    }
    for l in &vs.lead {
        frames.push(regop::FrameIn { annexb: &l.annexb, emit: true });
    }

    let reencoded = regop::reencode_lead_gop(&frames, args.crf, &args.preset).with_context(|| {
        format!(
            "re-encoding {}-frame leading GOP ({} priming) for channel {}",
            vs.lead.len(),
            priming.len(),
            cid
        )
    })?;

    for (l, new_annexb) in vs.lead.iter().zip(reencoded.iter()) {
        let cv = pb::foxglove::CompressedVideo {
            timestamp: Some(prost_types::Timestamp {
                seconds: l.ts_sec,
                nanos: l.ts_nanos,
            }),
            frame_id: l.frame_id.clone(),
            data: new_annexb.clone(),
            format: "h264".to_string(),
        };
        let mut buf = Vec::with_capacity(cv.encoded_len());
        cv.encode(&mut buf)?;
        let header = MessageHeader {
            channel_id: cid,
            sequence: l.sequence,
            log_time: l.log_time,
            publish_time: l.publish_time,
        };
        demux.push(&header, &buf)?;
    }
    vs.lead.clear();
    Ok(())
}

/// Read the channel's frames back from the prior keyframe up to (but not
/// including) the window start, so the decoder can be primed to decode the
/// leading partial GOP. Widens the look-back until a keyframe is found.
fn read_priming(
    sources: &[Source<'_>],
    role: Role,
    in_channel_id: u16,
    window_start: u64,
) -> Result<Vec<PrimeFrame>> {
    let source = sources
        .iter()
        .find(|s| s.role == role)
        .ok_or_else(|| anyhow!("no source for role {:?}", role))?;

    let mut preroll = PREROLL_NS;
    loop {
        let from = window_start.saturating_sub(preroll);
        let frames = read_channel_annexb(source, in_channel_id, from, window_start.saturating_sub(1))?;
        if let Some(kf) = frames.iter().rposition(|f| nal::is_keyframe(&f.annexb)) {
            return Ok(frames[kf..].to_vec());
        }
        if from == 0 {
            // No keyframe back to the file start; best effort with what we have.
            return Ok(frames);
        }
        preroll = preroll.saturating_mul(2);
    }
}

fn read_channel_annexb(
    source: &Source<'_>,
    in_channel_id: u16,
    start: u64,
    end: u64,
) -> Result<Vec<PrimeFrame>> {
    let single = [Source {
        role: source.role,
        mcap: source.mcap,
        summary: source.summary,
    }];
    let reader = WindowReader::new(&single, start, end)?;
    let mut out = Vec::new();
    for m in reader {
        let m = m?;
        if m.in_channel_id != in_channel_id {
            continue;
        }
        let cv = pb::foxglove::CompressedVideo::decode(&m.data[..])
            .context("decoding priming CompressedVideo frame")?;
        out.push(PrimeFrame { annexb: cv.data });
    }
    Ok(out)
}

fn write_output(args: &Args, tables: &OutTables, out_path: &Path) -> Result<()> {
    let mut writer = OutputWriter::new(out_path)?;
    writer.write_magic()?;
    writer.write_typed(
        op::HEADER,
        &Header {
            profile: "ros2".to_string(),
            library: args.library.clone(),
        },
    )?;

    for s in &tables.schema_records {
        writer.write_schema(s)?;
    }
    for c in &tables.channel_records {
        writer.write_typed(op::CHANNEL, c)?;
    }

    let mut chunk_indexes: Vec<ChunkIndex> = Vec::new();
    let mut pending: Vec<MessageIndex> = Vec::new();
    let mut message_count: u64 = 0;
    let mut channel_message_counts: BTreeMap<u16, u64> = BTreeMap::new();
    let mut bounds: Option<(u64, u64)> = None;

    for (i, demux) in tables.demuxes.iter().enumerate() {
        let cid = (i + 1) as u16;
        let mut entries = demux.entries.clone();
        if entries.is_empty() {
            continue;
        }
        entries.sort_by_key(|e| e.log_time);

        let mut reader = demux.reopen_for_read()?;
        let mut cb = ChunkBuilder::new(args.chunk_size);
        for e in &entries {
            let (header, payload) = demux::read_entry(&mut reader, e)?;
            update_bounds(&mut bounds, header.log_time);
            cb.push_message(&header, &payload)?;
            if cb.uncompressed_len() >= args.chunk_size {
                flush_chunk(&mut cb, &mut writer, &mut chunk_indexes, &mut pending)?;
            }
        }
        if !cb.is_empty() {
            flush_chunk(&mut cb, &mut writer, &mut chunk_indexes, &mut pending)?;
        }

        channel_message_counts.insert(cid, entries.len() as u64);
        message_count += entries.len() as u64;
    }

    write_message_index_block(&mut writer, &mut chunk_indexes, &pending)?;

    writer.write_typed(op::DATA_END, &DataEnd { data_section_crc: 0 })?;

    let statistics = Statistics {
        message_count,
        schema_count: tables.schema_records.len() as u16,
        channel_count: tables.channel_records.len() as u32,
        attachment_count: 0,
        metadata_count: 0,
        chunk_count: chunk_indexes.len() as u32,
        message_start_time: bounds.map(|(s, _)| s).unwrap_or(0),
        message_end_time: bounds.map(|(_, e)| e).unwrap_or(0),
        channel_message_counts,
    };

    writer.write_summary_and_footer(
        &tables.schema_records,
        &tables.channel_records,
        &chunk_indexes,
        &[],
        &[],
        &statistics,
    )?;
    writer.finish()?;

    eprintln!(
        "  wrote {}  ({} channels, {} messages, {} chunks)",
        out_path.display(),
        statistics.channel_count,
        statistics.message_count,
        statistics.chunk_count,
    );
    Ok(())
}

fn flush_chunk(
    cb: &mut ChunkBuilder,
    writer: &mut OutputWriter,
    chunk_indexes: &mut Vec<ChunkIndex>,
    pending: &mut Vec<MessageIndex>,
) -> Result<()> {
    let (ci, mi) = cb.finish_into(writer)?;
    chunk_indexes.push(ci);
    pending.push(mi);
    Ok(())
}

/// Emit each MessageIndex in one contiguous block before the summary and
/// back-fill the owning ChunkIndex's `message_index_offsets` / length.
fn write_message_index_block(
    writer: &mut OutputWriter,
    chunk_indexes: &mut [ChunkIndex],
    pending: &[MessageIndex],
) -> Result<()> {
    for (ci, mi) in chunk_indexes.iter_mut().zip(pending.iter()) {
        let group_start = writer.position();
        let offset = writer.position();
        writer.write_typed(op::MESSAGE_INDEX, mi)?;
        let mut offsets = BTreeMap::new();
        offsets.insert(mi.channel_id, offset);
        ci.message_index_offsets = offsets;
        ci.message_index_length = writer.position() - group_start;
    }
    Ok(())
}

fn update_bounds(bounds: &mut Option<(u64, u64)>, log_time: u64) {
    *bounds = Some(match *bounds {
        Some((s, e)) => (s.min(log_time), e.max(log_time)),
        None => (log_time, log_time),
    });
}
