cadmpeg 0.6.1

// SPDX-License-Identifier: Apache-2.0
//! Fidelity-payload projection and stream extraction for
//! `cadmpeg query fidelity`.
//!
//! The decode sidecar (`<stem>.fidelity.json`) retains source bytes as
//! base64 `retained_records`. The bare view lists them as a table — the
//! extraction address space — and `--stream NAME` reassembles one
//! stream's retained bytes, byte-exactly, into `-o FILE` (or stdout with
//! `--binary-stdout`). The view validates retained record identity, length,
//! and digest before projecting or extracting.

use std::io::Write;
use std::path::{Path, PathBuf};

use crate::application::artifact_store::{FileDestination, OptionalFileDestination};
use anyhow::{bail, Context, Result};
use clap::Args;

use super::{detect, print_json, read_input, Artifact};

/// Input selection for `query fidelity`.
#[derive(Debug, Args)]
#[command(mut_arg("output", |arg| arg.value_name("FILE").requires("stream")),
    mut_arg("force", |arg| arg.requires("output")))]
pub(crate) struct FidelityArgs {
    /// Decode sidecar (`<stem>.fidelity.json`), or `-` for standard input.
    pub(super) file: PathBuf,
    /// Extract the retained bytes of this source stream instead of
    /// printing the table.
    #[arg(long, value_name = "NAME")]
    stream: Option<String>,
    #[command(flatten)]
    output: OptionalFileDestination,
    /// Stream the extracted bytes to stdout even though they are binary.
    #[arg(long, requires = "stream")]
    binary_stdout: bool,
    /// Print the projected table as JSON (record metadata, not the bytes).
    #[arg(long, conflicts_with = "stream")]
    json: bool,
}

impl FidelityArgs {
    /// Resolves the flat clap fields into one fidelity operation.
    pub(super) fn mode(&self) -> Result<FidelityMode<'_>> {
        let Some(stream) = self.stream.as_deref() else {
            return Ok(if self.json {
                FidelityMode::Json
            } else {
                FidelityMode::Table
            });
        };
        let sink = if let Some(file) = self.output.0.as_ref() {
            Sink::File(file)
        } else if self.binary_stdout {
            Sink::Stdout
        } else {
            bail!(
                "retained stream bytes are binary output that a terminal cannot read; \
                 pass `-o FILE` or `--binary-stdout` to stream the bytes anyway"
            );
        };
        Ok(FidelityMode::Extract { stream, sink })
    }
}

/// One complete fidelity query operation.
#[derive(Debug, Clone, Copy)]
pub(in crate::query) enum FidelityMode<'a> {
    Table,
    Json,
    Extract { stream: &'a str, sink: Sink<'a> },
}

/// Destination for extracted retained bytes.
#[derive(Debug, Clone, Copy)]
pub(in crate::query) enum Sink<'a> {
    File(&'a FileDestination),
    Stdout,
}

/// Runs `query fidelity` against one artifact.
pub(super) fn run(file: &Path, mode: FidelityMode<'_>) -> Result<()> {
    let bytes = read_input(file)?;
    match detect(&bytes, file)? {
        Artifact::Sidecar(_) => {}
        Artifact::Cadir(_) => bail!(
            "{} is a CADIR document; the fidelity payload lives in the \
             `<stem>.fidelity.json` decode sidecar `cadmpeg dump` writes \
             next to it",
            file.display()
        ),
        Artifact::Report(_) => bail!(
            "{} is a command report; the fidelity payload lives in the \
             `<stem>.fidelity.json` decode sidecar `cadmpeg dump` writes",
            file.display()
        ),
    }
    let text = std::str::from_utf8(&bytes)
        .with_context(|| format!("reading the decode sidecar {} as UTF-8", file.display()))?;
    let sidecar = cadmpeg_ir::DecodeSidecar::from_json(text)
        .with_context(|| format!("validating the decode sidecar {}", file.display()))?;
    let payload = &sidecar.fidelity;

    match mode {
        FidelityMode::Extract { stream, sink } => extract(file, payload, stream, sink),
        FidelityMode::Json => {
            let records: Vec<serde_json::Value> = payload
                .retained_records()
                .iter()
                .map(|(id, record)| {
                    serde_json::json!({
                        "id": id,
                        "stream": record.stream(),
                        "offset": record.offset(),
                        "byte_len": record.byte_len(),
                        "data_retained": record.data().is_some(),
                    })
                })
                .collect();
            let value = serde_json::json!({
                "annotations": {
                    "streams": payload.annotations.stream_count(),
                    "provenance": payload.annotations.provenance.len(),
                    "exactness": payload.annotations.exactness().len(),
                },
                "retained_records": records,
            });
            print_json("fidelity", value)?;
            Ok(())
        }
        FidelityMode::Table => {
            println!("stream\toffset\tbytes\tdata\tid");
            for (id, record) in payload.retained_records() {
                println!(
                    "{}\t{}\t{}\t{}\t{}",
                    super::cell(record.stream()),
                    record.offset(),
                    record.byte_len(),
                    if record.data().is_some() { "yes" } else { "no" },
                    super::cell(id.as_str()),
                );
            }
            if payload.retained_records().is_empty() {
                eprintln!("(this sidecar retains no source records)");
            }
            eprintln!(
                "annotations: {} streams, {} provenance entries, {} exactness \
                 notes; extract retained bytes with `cadmpeg query fidelity \
                 FILE --stream NAME -o OUT`",
                payload.annotations.stream_count(),
                payload.annotations.provenance.len(),
                payload.annotations.exactness().len(),
            );
            Ok(())
        }
    }
}

/// Reassembles one stream's retained bytes and writes them byte-exactly.
fn extract(
    input: &Path,
    payload: &cadmpeg_ir::SourceFidelity,
    stream: &str,
    sink: Sink<'_>,
) -> Result<()> {
    const SHOWN: usize = 20;
    let selected: Vec<(
        &cadmpeg_ir::ids::UnknownId,
        &cadmpeg_ir::RetainedSourceRecord,
    )> = payload
        .retained_records()
        .iter()
        .filter(|(_, record)| record.stream() == stream)
        .collect();
    if selected.is_empty() {
        if payload.retained_records().is_empty() {
            bail!("this sidecar retains no source records");
        }
        let mut streams: Vec<&str> = payload
            .retained_records()
            .values()
            .map(cadmpeg_ir::RetainedSourceRecord::stream)
            .collect();
        streams.sort_unstable();
        streams.dedup();
        let shown: Vec<&str> = streams.iter().take(SHOWN).copied().collect();
        bail!(
            "no retained record has stream {stream:?}; retained streams: {}{}",
            shown.join(", "),
            if streams.len() > SHOWN { ", …" } else { "" }
        );
    }
    let mut matched = Vec::new();
    let mut missing = Vec::new();
    for (id, record) in selected {
        match record.data() {
            Some(data) => matched.push((id, record, data)),
            None => missing.push(id.as_str()),
        }
    }
    if !missing.is_empty() {
        bail!(
            "stream {stream:?} is retained without bytes (extent and digest \
             only) for: {}",
            missing.join(", ")
        );
    }
    matched.sort_by_key(|(_, record, _)| record.offset());
    let mut assembled: Vec<u8> = Vec::new();
    let mut expected_offset: Option<u64> = None;
    for (_, record, data) in &matched {
        if let Some(expected) = expected_offset {
            if record.offset() != expected {
                let extents: Vec<String> = matched
                    .iter()
                    .map(|(id, record, _)| {
                        format!("{}+{} ({id})", record.offset(), record.byte_len())
                    })
                    .collect();
                bail!(
                    "the retained extents of stream {stream:?} are not \
                     contiguous: {}; extract records individually (their \
                     `data` fields are base64)",
                    extents.join(", ")
                );
            }
        }
        expected_offset = Some(record.end_offset());
        assembled.extend_from_slice(data);
    }

    match sink {
        Sink::File(destination) => {
            destination.write(input, &assembled)?;
            eprintln!(
                "wrote {} bytes from {} record(s) of stream {stream:?} to {}",
                assembled.len(),
                matched.len(),
                destination.path.display()
            );
            Ok(())
        }
        Sink::Stdout => std::io::stdout()
            .write_all(&assembled)
            .context("writing extracted bytes to stdout"),
    }
}