nornir 0.5.3

//! Iceberg writer + reader + [`BuildProvenanceSink`] for **build-provenance**
//! (`build_provenance`) — task #36.
//!
//! When `nornir build --variant <v> --attest` captures a
//! [`BuildProvenance`] record, it is written to THIS ledger **first** (the
//! warehouse), then dumped as a znippy archive beside the artifact
//! (`nornir::build::seal`). This module is the warehouse half: one Iceberg
//! snapshot per recorded build, partitioned by `repo` so per-repo history reads
//! prune. It mirrors `warehouse::release_changes` / `warehouse::test_results`.
//!
//! Grain: one row per recorded build (`id` = the record UUID). The full record
//! travels in `record_json`, so a reader reconstructs the complete provenance
//! ([`BuildProvenanceRow::record`]) while the flat columns (variant, digest,
//! outcome, host cores/memory, artifact_count) stay queryable.

use std::sync::Arc;

use anyhow::{anyhow, Result};
use arrow::array::{
    Array, Int32Array, Int64Array, RecordBatch, StringArray, TimestampMicrosecondArray,
};
use iceberg::Catalog;
use iceberg::arrow::schema_to_arrow_schema;

use nornir_build_thing::provenance::Outcome;
use nornir_build_thing::{BuildProvenance, BuildProvenanceSink};

use super::iceberg::{append_batch, IcebergWarehouse, TABLE_BUILD_PROVENANCE};

// Column indices match `iceberg_schema::build_provenance` field order.
const COL_ID: usize = 0;
const COL_REPO: usize = 1;
const COL_VARIANT: usize = 2;
const COL_OUTCOME: usize = 3;
const COL_PROFILE: usize = 4;
const COL_TARGET: usize = 5;
const COL_VCS_REV: usize = 6;
const COL_VCS_BRANCH: usize = 7;
const COL_RUSTC: usize = 8;
const COL_HOST_CORES: usize = 9;
const COL_HOST_MEMORY: usize = 10;
const COL_ARTIFACT_COUNT: usize = 11;
const COL_DIGEST: usize = 12;
const COL_BUILT_AT: usize = 13;
const COL_TS_MICROS: usize = 14;
const COL_RECORD_JSON: usize = 15;

/// A flat build-provenance row read back from the warehouse. The complete record
/// is in `record_json` (reconstruct via [`Self::record`]).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BuildProvenanceRow {
    pub id: String,
    pub repo: String,
    pub variant: String,
    pub outcome: String,
    pub profile: String,
    pub target: String,
    pub vcs_rev: String,
    pub vcs_branch: String,
    pub rustc: String,
    pub host_cores: i32,
    pub host_memory_bytes: i64,
    pub artifact_count: i64,
    pub digest: String,
    pub built_at: String,
    pub ts_micros: i64,
    pub record_json: String,
}

impl BuildProvenanceRow {
    /// Reconstruct the full [`BuildProvenance`] record from `record_json`.
    pub fn record(&self) -> Result<BuildProvenance> {
        BuildProvenance::from_json(&self.record_json)
    }

    /// Project a [`BuildProvenance`] into a warehouse row (`digest` = the
    /// attestation's SHA-256 over the canonical record).
    fn from_record(prov: &BuildProvenance) -> Result<Self> {
        let digest = prov.clone().attest()?.digest_hex();
        Ok(BuildProvenanceRow {
            id: prov.id.clone(),
            repo: prov.repo.clone(),
            variant: prov.variant.name.clone(),
            outcome: match prov.outcome {
                Outcome::Success => "success",
                Outcome::Failed => "failed",
            }
            .to_string(),
            profile: prov.inputs.profile.clone(),
            target: prov.inputs.target.clone().unwrap_or_default(),
            vcs_rev: prov.inputs.vcs_rev.clone(),
            vcs_branch: prov.inputs.vcs_branch.clone(),
            rustc: prov.toolchain.rustc.clone(),
            host_cores: prov.host.cores as i32,
            host_memory_bytes: prov.host.memory_bytes as i64,
            artifact_count: prov.artifacts.len() as i64,
            digest,
            built_at: prov.built_at.clone(),
            ts_micros: chrono::Utc::now().timestamp_micros(),
            record_json: prov.to_json()?,
        })
    }
}

/// Append one build-provenance record (one Iceberg snapshot).
pub async fn append_build_provenance(wh: &IcebergWarehouse, prov: &BuildProvenance) -> Result<()> {
    let row = BuildProvenanceRow::from_record(prov)?;
    let table = wh.catalog().load_table(&wh.table_ident(TABLE_BUILD_PROVENANCE)).await?;
    let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);

    let cols: Vec<Arc<dyn Array>> = vec![
        Arc::new(StringArray::from(vec![row.id.clone()])),
        Arc::new(StringArray::from(vec![row.repo.clone()])),
        Arc::new(StringArray::from(vec![row.variant.clone()])),
        Arc::new(StringArray::from(vec![row.outcome.clone()])),
        Arc::new(StringArray::from(vec![row.profile.clone()])),
        Arc::new(StringArray::from(vec![row.target.clone()])),
        Arc::new(StringArray::from(vec![row.vcs_rev.clone()])),
        Arc::new(StringArray::from(vec![row.vcs_branch.clone()])),
        Arc::new(StringArray::from(vec![row.rustc.clone()])),
        Arc::new(Int32Array::from(vec![row.host_cores])),
        Arc::new(Int64Array::from(vec![row.host_memory_bytes])),
        Arc::new(Int64Array::from(vec![row.artifact_count])),
        Arc::new(StringArray::from(vec![row.digest.clone()])),
        Arc::new(StringArray::from(vec![row.built_at.clone()])),
        Arc::new(
            TimestampMicrosecondArray::from(vec![row.ts_micros]).with_timezone("+00:00"),
        ),
        Arc::new(StringArray::from(vec![row.record_json.clone()])),
    ];
    let batch = RecordBatch::try_new(arrow_schema, cols)?;
    append_batch(wh.catalog(), table, batch).await?;
    Ok(())
}

/// Read build-provenance rows, optionally scoped to `repo`, newest-first
/// (`ts_micros` desc).
pub async fn query_build_provenance(
    wh: &IcebergWarehouse,
    repo: Option<&str>,
) -> Result<Vec<BuildProvenanceRow>> {
    let batches: Vec<RecordBatch> =
        super::iceberg::load_and_read_all(wh, TABLE_BUILD_PROVENANCE).await?;

    let mut out: Vec<BuildProvenanceRow> = Vec::new();
    for b in &batches {
        let id = col_str(b, COL_ID)?;
        let repo_c = col_str(b, COL_REPO)?;
        let variant = col_str(b, COL_VARIANT)?;
        let outcome = col_str(b, COL_OUTCOME)?;
        let profile = col_str(b, COL_PROFILE)?;
        let target = col_str(b, COL_TARGET)?;
        let vcs_rev = col_str(b, COL_VCS_REV)?;
        let vcs_branch = col_str(b, COL_VCS_BRANCH)?;
        let rustc = col_str(b, COL_RUSTC)?;
        let cores = col_i32(b, COL_HOST_CORES)?;
        let mem = col_i64(b, COL_HOST_MEMORY)?;
        let artifacts = col_i64(b, COL_ARTIFACT_COUNT)?;
        let digest = col_str(b, COL_DIGEST)?;
        let built_at = col_str(b, COL_BUILT_AT)?;
        let ts = col_ts(b, COL_TS_MICROS)?;
        let record_json = col_str(b, COL_RECORD_JSON)?;
        for i in 0..b.num_rows() {
            let row = BuildProvenanceRow {
                id: id.value(i).to_string(),
                repo: repo_c.value(i).to_string(),
                variant: variant.value(i).to_string(),
                outcome: outcome.value(i).to_string(),
                profile: profile.value(i).to_string(),
                target: target.value(i).to_string(),
                vcs_rev: vcs_rev.value(i).to_string(),
                vcs_branch: vcs_branch.value(i).to_string(),
                rustc: rustc.value(i).to_string(),
                host_cores: cores.value(i),
                host_memory_bytes: mem.value(i),
                artifact_count: artifacts.value(i),
                digest: digest.value(i).to_string(),
                built_at: built_at.value(i).to_string(),
                ts_micros: ts.value(i),
                record_json: record_json.value(i).to_string(),
            };
            if repo.map(|r| row.repo == r).unwrap_or(true) {
                out.push(row);
            }
        }
    }
    out.sort_by(|a, b| b.ts_micros.cmp(&a.ts_micros).then_with(|| a.id.cmp(&b.id)));
    Ok(out)
}

/// The warehouse is the canonical build-provenance sink: `wh.record(&prov)`
/// writes one ledger row via the shared runtime.
impl BuildProvenanceSink for IcebergWarehouse {
    fn record(&self, prov: &BuildProvenance) -> Result<()> {
        self.block_on(append_build_provenance(self, prov))
    }
}

// ─── column helpers ─────────────────────────────────────────────────────

fn col_str<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a StringArray> {
    b.column(idx)
        .as_any()
        .downcast_ref::<StringArray>()
        .ok_or_else(|| anyhow!("build_provenance col {idx} is not StringArray"))
}

fn col_i32<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a Int32Array> {
    b.column(idx)
        .as_any()
        .downcast_ref::<Int32Array>()
        .ok_or_else(|| anyhow!("build_provenance col {idx} is not Int32Array"))
}

fn col_i64<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a Int64Array> {
    b.column(idx)
        .as_any()
        .downcast_ref::<Int64Array>()
        .ok_or_else(|| anyhow!("build_provenance col {idx} is not Int64Array"))
}

fn col_ts<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a TimestampMicrosecondArray> {
    b.column(idx)
        .as_any()
        .downcast_ref::<TimestampMicrosecondArray>()
        .ok_or_else(|| anyhow!("build_provenance col {idx} is not TimestampMicrosecondArray"))
}

#[cfg(test)]
mod tests {
    use super::*;
    use nornir_build_thing::provenance::{HostInfo, Toolchain};
    use nornir_build_thing::variant::{BuildVariant, Profile};

    fn sample_record(variant: &str) -> BuildProvenance {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::create_dir_all(root.join("target/release")).unwrap();
        let art = root.join("target/release").join(format!("holger-{variant}"));
        std::fs::write(&art, format!("ELF {variant}").as_bytes()).unwrap();
        let v = BuildVariant {
            name: variant.to_string(),
            package: Some(format!("holger-{variant}")),
            bin: Some(format!("holger-{variant}")),
            features: vec![variant.to_string()],
            profile: Profile::Release,
            ..Default::default()
        };
        BuildProvenance::capture(
            &v,
            "holger",
            root,
            Toolchain { rustc: "rustc 1.85.0".into(), host: "x86_64".into(), cargo: "cargo 1.85.0".into() },
            HostInfo { cores: 32, memory_bytes: 128 * 1024 * 1024 * 1024 },
            Outcome::Success,
            &[art],
        )
        .unwrap()
    }

    #[test]
    fn warehouse_sink_round_trips_build_provenance() {
        let dir = tempfile::tempdir().unwrap();
        let wh = IcebergWarehouse::open(dir.path()).unwrap();

        // Record all three holger exes through the SINK (the write-warehouse-first path).
        for v in ["embedded", "minimal", "full"] {
            wh.record(&sample_record(v)).unwrap();
        }

        let rows = wh.block_on(query_build_provenance(&wh, Some("holger"))).unwrap();
        assert_eq!(rows.len(), 3, "three builds recorded");
        let variants: std::collections::BTreeSet<&str> =
            rows.iter().map(|r| r.variant.as_str()).collect();
        assert_eq!(variants, ["embedded", "full", "minimal"].into_iter().collect());

        // Host facts + artifact hashes survive the round-trip via record_json.
        let full = rows.iter().find(|r| r.variant == "full").unwrap();
        assert_eq!(full.host_cores, 32);
        assert_eq!(full.host_memory_bytes, 128 * 1024 * 1024 * 1024);
        assert_eq!(full.artifact_count, 1);
        assert_eq!(full.outcome, "success");
        assert_eq!(full.digest.len(), 64, "sha256 hex digest stored");

        // The full record reconstructs, and its digest matches the flat column.
        let rec = full.record().unwrap();
        assert_eq!(rec.variant.name, "full");
        assert_eq!(rec.artifacts.len(), 1);
        assert_eq!(rec.clone().attest().unwrap().digest_hex(), full.digest);
    }
}