nornir 0.5.3

//! Iceberg writer/reader for the **airgap-op DAG** (`airgap_events`) — EPIC
//! AIRGAP / AIRGAP8.
//!
//! The durable twin of the lean `skidbladnir` crate's JSONL event sink
//! (`skidbladnir::events`): on an air-gapped target with no warehouse server
//! the crate appends events to `<warehouse_root>/airgap_events.jsonl`; when the
//! fat `nornir` CLI runs against a real warehouse, [`append_airgap_events`]
//! ingests those same rows into the historized Iceberg table so the bring-up DAG
//! is queryable + time-travelable, exactly like `release_events`.
//!
//! Schema: [`super::iceberg_schema::airgap_events`].

use std::path::Path;
use std::sync::Arc;

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

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

const COL_RUN_ID: usize = 0;
const COL_SEQ: usize = 1;
const COL_TS_MICROS: usize = 2;
const COL_PRODUCT: usize = 3;
const COL_OP: usize = 4;
const COL_PHASE: usize = 5;
const COL_STATUS: usize = 6;
const COL_DETAIL: usize = 7;
const COL_DEPENDS_ON: usize = 8;

/// One row of the `airgap_events` table — identical fields to the lean crate's
/// `AirgapEvent`, re-declared here so the warehouse layer needn't depend on the
/// `skidbladnir` crate (the JSONL bytes are the contract).
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AirgapEventRow {
    pub run_id: String,
    pub seq: i64,
    pub ts_micros: i64,
    pub product: String,
    pub op: String,
    pub phase: String,
    pub status: String,
    pub detail: String,
    /// Bring-up producers this op waited on (None = root op).
    pub depends_on: Option<Vec<String>>,
}

/// Read a `skidbladnir` JSONL event file (one JSON object per line) into rows.
/// `depends_on` becomes `None` when empty (matches the optional Iceberg column).
pub fn read_jsonl(path: &Path) -> Result<Vec<AirgapEventRow>> {
    #[derive(serde::Deserialize)]
    struct Raw {
        run_id: String,
        seq: i64,
        ts_micros: i64,
        product: String,
        op: String,
        phase: String,
        status: String,
        #[serde(default)]
        detail: String,
        #[serde(default)]
        depends_on: Vec<String>,
    }
    let text = std::fs::read_to_string(path)?;
    let mut rows = Vec::new();
    for line in text.lines().filter(|l| !l.trim().is_empty()) {
        let r: Raw = serde_json::from_str(line)?;
        rows.push(AirgapEventRow {
            run_id: r.run_id,
            seq: r.seq,
            ts_micros: r.ts_micros,
            product: r.product,
            op: r.op,
            phase: r.phase,
            status: r.status,
            detail: r.detail,
            depends_on: if r.depends_on.is_empty() { None } else { Some(r.depends_on) },
        });
    }
    rows.sort_by_key(|r| (r.run_id.clone(), r.seq));
    Ok(rows)
}

/// Append a batch of `airgap_events` rows (one Iceberg snapshot). Mirrors
/// [`super::release_events::append_release_events`].
pub async fn append_airgap_events(wh: &IcebergWarehouse, rows: &[AirgapEventRow]) -> Result<()> {
    if rows.is_empty() {
        return Ok(());
    }
    let table = wh.catalog().load_table(&wh.table_ident(TABLE_AIRGAP_EVENTS)).await?;
    let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);

    let dep_elem = match arrow_schema.field(COL_DEPENDS_ON).data_type() {
        arrow::datatypes::DataType::List(elem) | arrow::datatypes::DataType::LargeList(elem) => {
            elem.clone()
        }
        other => return Err(anyhow!("depends_on column expected List, got {other:?}")),
    };
    let mut deps_b: ListBuilder<StringBuilder> =
        ListBuilder::new(StringBuilder::new()).with_field(dep_elem);
    for r in rows {
        match &r.depends_on {
            None => deps_b.append(false),
            Some(v) => {
                for d in v {
                    deps_b.values().append_value(d);
                }
                deps_b.append(true);
            }
        }
    }

    let cols: Vec<Arc<dyn Array>> = vec![
        Arc::new(StringArray::from(rows.iter().map(|r| r.run_id.clone()).collect::<Vec<_>>())),
        Arc::new(Int64Array::from(rows.iter().map(|r| r.seq).collect::<Vec<_>>())),
        Arc::new(
            TimestampMicrosecondArray::from(rows.iter().map(|r| r.ts_micros).collect::<Vec<_>>())
                .with_timezone("+00:00"),
        ),
        Arc::new(StringArray::from(rows.iter().map(|r| r.product.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(rows.iter().map(|r| r.op.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(rows.iter().map(|r| r.phase.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(rows.iter().map(|r| r.status.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(rows.iter().map(|r| r.detail.clone()).collect::<Vec<_>>())),
        Arc::new(deps_b.finish()),
    ];
    let batch = RecordBatch::try_new(arrow_schema, cols)?;
    append_batch(wh.catalog(), table, batch).await?;
    Ok(())
}

/// Read the historized `airgap_events` DAG back out of the warehouse Iceberg
/// table — the READ twin of [`append_airgap_events`]. Optionally scoped to one
/// `run_id` (`None` = every recorded run). Rows come back sorted by
/// `(run_id, seq)`, exactly like [`read_jsonl`], so the same
/// [`events_outcome`] renderer shapes either source. Blank/torn Iceberg
/// metadata → 0 rows, never a JSON-EOF crash (load+read).
pub async fn query_airgap_events(
    wh: &IcebergWarehouse,
    run_id: Option<&str>,
) -> Result<Vec<AirgapEventRow>> {
    let batches: Vec<RecordBatch> =
        super::iceberg::load_and_read_all(wh, TABLE_AIRGAP_EVENTS).await?;
    let mut out: Vec<AirgapEventRow> = Vec::new();
    for b in &batches {
        let run = col_str(b, COL_RUN_ID)?;
        let seq = col_i64(b, COL_SEQ)?;
        let ts = col_ts(b, COL_TS_MICROS)?;
        let product = col_str(b, COL_PRODUCT)?;
        let op = col_str(b, COL_OP)?;
        let phase = col_str(b, COL_PHASE)?;
        let status = col_str(b, COL_STATUS)?;
        let detail = col_str(b, COL_DETAIL)?;
        let deps = col_str_list(b, COL_DEPENDS_ON)?;
        for i in 0..b.num_rows() {
            if let Some(want) = run_id {
                if run.value(i) != want {
                    continue;
                }
            }
            out.push(AirgapEventRow {
                run_id: run.value(i).to_string(),
                seq: seq.value(i),
                ts_micros: ts.value(i),
                product: product.value(i).to_string(),
                op: op.value(i).to_string(),
                phase: phase.value(i).to_string(),
                status: status.value(i).to_string(),
                detail: detail.value(i).to_string(),
                // Normalise an empty list to None, matching read_jsonl's contract.
                depends_on: deps[i].as_ref().filter(|d| !d.is_empty()).cloned(),
            });
        }
    }
    out.sort_by(|a, b| (a.run_id.as_str(), a.seq).cmp(&(b.run_id.as_str(), b.seq)));
    Ok(out)
}

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

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

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

fn col_str_list(b: &RecordBatch, idx: usize) -> Result<Vec<Option<Vec<String>>>> {
    let arr = b
        .column(idx)
        .as_any()
        .downcast_ref::<ListArray>()
        .ok_or_else(|| anyhow!("airgap_events col {idx} is not ListArray"))?;
    let mut out = Vec::with_capacity(arr.len());
    for i in 0..arr.len() {
        if arr.is_null(i) {
            out.push(None);
            continue;
        }
        let values = arr.value(i);
        let strs = values
            .as_any()
            .downcast_ref::<StringArray>()
            .ok_or_else(|| anyhow!("airgap_events depends_on element not StringArray"))?;
        let mut v = Vec::with_capacity(strs.len());
        for j in 0..strs.len() {
            if !strs.is_null(j) {
                v.push(strs.value(j).to_string());
            }
        }
        out.push(Some(v));
    }
    Ok(out)
}

// ─── AUT9 CLI command outcomes ───────────────────────────────────────────────
// The airgap READ commands (`events`, `fetch`/verify, `unfold`/plan) rendered as
// the uniform `crate::cli_outcome::CommandOutcome` — the CLI twin of a Facet's
// `state_json`. Each builder takes ALREADY-fetched data (rows / a verify report /
// a rendered plan) and only SHAPES it; it never reads disk or runs a backend
// (CommandOutcome is presentation, never a transport). `ok ⟺ a real, non-empty,
// non-failed result` (RAGNARÖK: empty / tampered ⇒ RED, never silently green).
// See `.nornir/cli-command-contract.md`.

/// `airgap events` — the bring-up DAG read back as a [`CommandOutcome`]. `rows`
/// are already parsed from the JSONL sink (`read_jsonl`); this only shapes +
/// renders them. `ok ⟺ ≥1 event` (an empty/absent DAG is RED).
pub fn events_outcome(
    path: &Path,
    rows: &[AirgapEventRow],
) -> crate::cli_outcome::CommandOutcome {
    use crate::cli_outcome::CommandOutcome;
    if rows.is_empty() {
        return CommandOutcome::fail(
            "airgap events",
            format!("no airgap_events at {} (run `nornir airgap start`)", path.display()),
        );
    }
    let mut human = String::new();
    for ev in rows {
        let deps = ev.depends_on.as_ref().filter(|d| !d.is_empty());
        human.push_str(&format!(
            "  [{}] {:>3} {:<14} {:<8} {:<5} {:<7} {} {}\n",
            ev.run_id,
            ev.seq,
            ev.product,
            ev.op,
            ev.phase,
            ev.status,
            ev.detail,
            match deps {
                Some(d) => format!("← {}", d.join(",")),
                None => String::new(),
            }
        ));
    }
    let data = serde_json::json!({ "events": rows });
    CommandOutcome::ok("airgap events", data, human.trim_end().to_string())
}

/// `airgap fetch` (AIRGAP3 — verify a received bundle) as a [`CommandOutcome`].
/// Takes the already-computed verify fields (from
/// `skidbladnir::fetch::verify_on_receipt`). `ok ⟺ manifest+contents intact`
/// (a tampered/failed bundle is RED, carrying the bad entries in `data`).
pub fn verify_outcome(
    bundle: &Path,
    manifest_root_ok: bool,
    bundle_hash_ok: bool,
    bad_entries: &[String],
) -> crate::cli_outcome::CommandOutcome {
    use crate::cli_outcome::CommandOutcome;
    let ok = manifest_root_ok && bundle_hash_ok && bad_entries.is_empty();
    let data = serde_json::json!({
        "bundle": bundle.display().to_string(),
        "manifest_root_ok": manifest_root_ok,
        "bundle_hash_ok": bundle_hash_ok,
        "bad_entries": bad_entries,
        "ok": ok,
    });
    if ok {
        CommandOutcome::ok(
            "airgap fetch",
            data,
            format!(
                "✓ bundle verified: manifest+contents intact, zero network ({})",
                bundle.display()
            ),
        )
    } else {
        CommandOutcome::fail(
            "airgap fetch",
            format!(
                "✗ bundle verify FAILED for {} (manifest_root_ok={manifest_root_ok}, bundle_hash_ok={bundle_hash_ok}, bad={bad_entries:?})",
                bundle.display()
            ),
        )
    }
}

/// `airgap unfold` (AIRGAP7 — the provisioning PLAN for a target) as a
/// [`CommandOutcome`]. Takes the already-rendered plan `lines` (the backend's
/// qemu argv / HCL / reset body) for `backend`/`target`. `ok ⟺ a non-empty plan`
/// (an unknown backend yields no lines ⇒ RED, handled by the caller's bail).
pub fn unfold_plan_outcome(
    backend: &str,
    target: &str,
    lines: &[String],
) -> crate::cli_outcome::CommandOutcome {
    use crate::cli_outcome::CommandOutcome;
    if lines.is_empty() {
        return CommandOutcome::fail(
            "airgap unfold",
            format!("no provisioning plan for backend `{backend}` target `{target}`"),
        );
    }
    let data = serde_json::json!({
        "backend": backend,
        "target": target,
        "plan": lines,
    });
    CommandOutcome::ok("airgap unfold", data, lines.join("\n"))
}

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

    /// LAW 1 (inject-assert): parse a real JSONL event file (the lean crate's
    /// on-disk format) → assert the rows decode with the right fields and
    /// `depends_on` is `None` for roots / `Some` for edges.
    #[test]
    fn jsonl_decodes_airgap_events() {
        let dir = tempfile::tempdir().unwrap();
        let f = dir.path().join("airgap_events.jsonl");
        std::fs::write(
            &f,
            "{\"run_id\":\"r1\",\"seq\":0,\"ts_micros\":1,\"product\":\"holger\",\"op\":\"install\",\"phase\":\"end\",\"status\":\"ok\",\"detail\":\"\",\"depends_on\":[]}\n\
             {\"run_id\":\"r1\",\"seq\":1,\"ts_micros\":2,\"product\":\"nornir-server\",\"op\":\"start\",\"phase\":\"end\",\"status\":\"ok\",\"detail\":\"healthy\",\"depends_on\":[\"holger\"]}\n",
        )
        .unwrap();

        let rows = read_jsonl(&f).unwrap();
        assert_eq!(rows.len(), 2);
        assert_eq!(rows[0].product, "holger");
        assert_eq!(rows[0].op, "install");
        assert_eq!(rows[0].depends_on, None, "root op has no deps");
        assert_eq!(rows[1].product, "nornir-server");
        assert_eq!(rows[1].depends_on, Some(vec!["holger".to_string()]));
    }

    fn ev(run: &str, seq: i64, product: &str, op: &str, deps: Option<Vec<String>>) -> AirgapEventRow {
        AirgapEventRow {
            run_id: run.into(),
            seq,
            ts_micros: seq + 1,
            product: product.into(),
            op: op.into(),
            phase: "end".into(),
            status: "ok".into(),
            detail: String::new(),
            depends_on: deps,
        }
    }

    /// Round-trip: append a bring-up DAG (roots + dep edges) into the Iceberg
    /// `airgap_events` table, then read it straight back with the READ twin
    /// `query_airgap_events`. Asserts every field survives — including the
    /// optional `depends_on` list (None for roots, Some for edges) — and that
    /// the `run_id` filter scopes correctly. Guards the write-only gap: the
    /// warehouse table now has a reader, not just an ingest writer.
    #[test]
    fn append_query_round_trips_and_scopes_by_run() {
        let dir = tempfile::tempdir().unwrap();
        let wh = IcebergWarehouse::open(dir.path()).unwrap();

        let run_a = vec![
            ev("run-a", 0, "holger", "install", None),
            ev("run-a", 1, "nornir-server", "start", Some(vec!["holger".into()])),
        ];
        let run_b = vec![ev("run-b", 0, "tunnr", "install", None)];
        wh.block_on(append_airgap_events(&wh, &run_a)).unwrap();
        wh.block_on(append_airgap_events(&wh, &run_b)).unwrap();

        // Every run, sorted by (run_id, seq).
        let all = wh.block_on(query_airgap_events(&wh, None)).unwrap();
        assert_eq!(all.len(), 3);
        assert_eq!(all, {
            let mut v = run_a.clone();
            v.extend(run_b.clone());
            v
        });
        // depends_on faithfully preserved: root → None, edge → Some.
        assert_eq!(all[0].depends_on, None);
        assert_eq!(all[1].depends_on, Some(vec!["holger".to_string()]));

        // Scoped to one run.
        let just_a = wh.block_on(query_airgap_events(&wh, Some("run-a"))).unwrap();
        assert_eq!(just_a.len(), 2);
        assert!(just_a.iter().all(|r| r.run_id == "run-a"));

        // The CommandOutcome renderer shapes the warehouse rows the same as JSONL.
        let outcome = events_outcome(std::path::Path::new("/dev/null"), &all);
        assert!(outcome.is_sannr());
        assert_eq!(outcome.data["events"].as_array().unwrap().len(), 3);
    }

    #[test]
    fn events_outcome_empty_is_red_real_is_sannr() {
        let p = std::path::Path::new("/tmp/airgap_events.jsonl");
        // RAGNARÖK: an absent/empty DAG is RED.
        let red = events_outcome(p, &[]);
        assert_eq!(red.command, "airgap events");
        assert!(!red.is_sannr());
        assert!(red.human.contains("no airgap_events"));
        // real rows ⇒ sannr, events surfaced as readable data.
        let rows = vec![
            ev("r1", 0, "holger", "install", None),
            ev("r1", 1, "nornir-server", "start", Some(vec!["holger".into()])),
        ];
        let o = events_outcome(p, &rows);
        assert!(o.is_sannr());
        let arr = o.data["events"].as_array().unwrap();
        assert_eq!(arr.len(), 2);
        assert_eq!(arr[1]["product"], serde_json::json!("nornir-server"));
        assert!(o.human.contains("holger"), "dep edge rendered: {}", o.human);
    }

    #[test]
    fn verify_outcome_intact_is_sannr_tampered_is_red() {
        let b = std::path::Path::new("/srv/lake.tar.zst");
        let ok = verify_outcome(b, true, true, &[]);
        assert_eq!(ok.command, "airgap fetch");
        assert!(ok.is_sannr(), "intact bundle is a true (sannr) outcome");
        assert_eq!(ok.data["ok"], serde_json::json!(true));
        assert_eq!(ok.data["bundle"], serde_json::json!("/srv/lake.tar.zst"));
        // a bad entry ⇒ verify FAILS ⇒ RED.
        let bad = verify_outcome(b, true, false, &["bin/holger".into()]);
        assert!(!bad.is_sannr());
        assert!(bad.human.contains("FAILED"));
        assert!(bad.human.contains("bin/holger"));
    }

    #[test]
    fn unfold_plan_outcome_real_plan_is_sannr_empty_is_red() {
        let lines = vec![
            "qemu-system-x86_64 -accel kvm".to_string(),
            "  -m 4096".to_string(),
        ];
        let o = unfold_plan_outcome("kvm", "njord", &lines);
        assert_eq!(o.command, "airgap unfold");
        assert!(o.is_sannr());
        assert_eq!(o.data["backend"], serde_json::json!("kvm"));
        assert_eq!(o.data["plan"].as_array().unwrap().len(), 2);
        assert!(o.human.contains("qemu-system-x86_64"));
        // no plan ⇒ RED.
        assert!(!unfold_plan_outcome("kvm", "njord", &[]).is_sannr());
    }
}