nornir 0.5.3

//! Warehouse — append-only columnar store backing nornir's bench/symbol
//! history.
//!
//! The canonical backend is [`iceberg::IcebergWarehouse`] (Apache Iceberg via
//! `iceberg-rust` over the single-file **skade-katalog** redb catalogue).
//! Every derived fact is an Iceberg row keyed by git SHA. A `RemoteWarehouse`
//! (Arrow Flight to `nornir-server`, Phase 5) will reuse the same [`Warehouse`]
//! trait surface; the trait is the durable abstraction.

pub mod iceberg;
pub mod iceberg_schema;
/// Ragnar-fast: a static search-tree (STree64 / Eytzinger) index over the
/// commit/version log so as-of + point-in-time lookups are `O(log n)` instead
/// of a linear scan. See [`ragnar::RagnarLog`].
pub mod ragnar;
/// SQL over the warehouse (`nornir warehouse query "SELECT …"`) — DataFusion
/// executing over the Iceberg tables through native `iceberg-datafusion`
/// `TableProvider`s (projection + filter pushdown to Parquet). See [`sql`].
pub mod sql;
/// Warehouse maintenance — small-file compaction + snapshot expiry (`nornir
/// warehouse compact` / `expire`). See [`maintenance`].
pub mod maintenance;
pub mod dep_graph;
pub mod funnel;
pub mod release_events;
/// Iceberg writer/reader for the reversible `release_changes` ledger (task #32) —
/// one row per mutating release step with before-state, reversed by `release undo`.
pub mod release_changes;
/// Iceberg writer/reader for the `clone_events` stream (PLAN #6) — per-member
/// clone/fetch/republish populate outcomes, readable by a thin viz/CLI client.
pub mod clone_events;
pub mod workspace_events;
/// Iceberg writer/reader for the `airgap_events` DAG (EPIC AIRGAP / AIRGAP8) —
/// the durable twin of the lean `skidbladnir` JSONL event sink.
pub mod airgap_events;
pub mod test_results;
// build-provenance ledger + BuildProvenanceSink (task #36).
pub mod build_provenance;
pub mod test_inventory;
// EPIC SAGA (task #25) — record/replay: warehouse `saga_events` writer/reader +
// the `IcebergSagaSink`. Re-exports the pure model from the `nornir-saga` crate.
pub mod saga_events;
pub mod surface_coverage;
// UI snapshot reference store (render-oracle reference images → WebP in the
// warehouse, out of git). See `.nornir/snapshot-to-warehouse.md`.
pub mod ui_snapshots;
pub mod coverage;
pub mod viz_actions;
pub mod agent_model_runs;
pub mod codegen_judge;
pub mod blob_store;
/// Static extractor for the fn → warehouse-table access fact (AUT7 / EPIC ARCH
/// n-002): the syn pass + the accessor → table map.
pub mod access_scan;
/// Iceberg writer/reader for the `warehouse_access_edges` fact table.
pub mod warehouse_access;
/// Generative-LLM abstraction (EPIC #39): ONE `Generator` trait, three real
/// backends (candle / mistralrs / onnx) behind the `generator(spec)` factory,
/// plus `mock` and an off-by-default `ollama` client. Feeds the bake-off.
pub mod generator;

use std::path::Path;

use anyhow::{Context, Result};
use arrow::array::RecordBatch;
use uuid::Uuid;

use crate::bench::BenchRun;
use crate::config::Storage;

// Re-export skade's backend-neutral scan predicate so trait callers can build
// pushdown filters (`ScanFilter::eq("repo", repo)`, `is_in`, `range`) without
// naming any engine type. This is the durable predicate surface a future
// `SkadeWarehouse` / DuckDB backend implements identically; the Iceberg backend
// lowers it to an `iceberg::expr::Predicate` inside skade.
pub use skade::{Scalar, ScanFilter};

/// The durable storage seam for nornir's warehouse (bench/test/release/symbol
/// history, etc.). **Sync** by design — every method blocks internally on the
/// backend's own runtime (`IcebergWarehouse` owns a tokio `Runtime` and calls
/// `block_on`), so the ~150 sync call sites never have to become async.
///
/// Two layers:
///
///  1. A **generic Arrow core** ([`append_arrow`](Warehouse::append_arrow),
///     [`scan_arrow`](Warehouse::scan_arrow),
///     [`scan_filtered`](Warehouse::scan_filtered),
///     [`scan_limited`](Warehouse::scan_limited)) plus table/catalog lifecycle
///     ([`ensure_table`](Warehouse::ensure_table),
///     [`ensure_columns`](Warehouse::ensure_columns),
///     [`table_names`](Warehouse::table_names)). This is what the sibling
///     modules (`dep_graph`, `release_events`, `test_results`, …) need: build a
///     `RecordBatch` from `iceberg_schema::*` and append / scan it by table
///     name, without reaching for the raw catalog. Backends only have to make
///     these honest and every typed method composes over them.
///
///  2. **Named per-table convenience methods** ([`append_bench_run`] /
///     [`query_bench_runs`]) kept as thin wrappers so existing callers are
///     untouched. (P1 keeps just the two that were already on the trait; the
///     rest stay as inherent `IcebergWarehouse` methods until the next phase
///     migrates consumers onto the trait.)
///
/// [`IcebergWarehouse`](iceberg::IcebergWarehouse) is the **default** (and
/// today the only) implementation: it delegates each method to its existing
/// inherent body, which already wraps skade (`skade::append` / `read_*` /
/// `ingest_parallel`). A future `SkadeWarehouse` (reusing a `skade::Warehouse`
/// + `Table` handle) or a DuckDB backend plugs in behind this same trait — see
/// `.nornir/warehouse-trait-skade-migration.md` for the P2/P3/P4 plan.
///
/// `iceberg_schema.rs` remains the per-table shape source of truth; the trait
/// is table-name + `RecordBatch` based and stays schema-agnostic.
pub trait Warehouse: Send + Sync {
    // ── generic Arrow core ──────────────────────────────────────────────────

    /// Append one Arrow [`RecordBatch`] to `table`. The batch must already match
    /// the table's current `iceberg_schema::*` shape — `append_arrow` only recasts
    /// to the stored schema and does NOT evolve it, so a batch carrying columns
    /// the table lacks is an ERROR. Callers that need add-column evolution must
    /// run [`ensure_columns`](Self::ensure_columns) first to migrate the schema,
    /// then append. (No silent forward schema evolution here.)
    fn append_arrow(&self, table: &str, batch: RecordBatch) -> Result<()>;

    /// Full-snapshot scan of `table` → Arrow, column order preserved (so
    /// positional downcasts stay valid).
    fn scan_arrow(&self, table: &str) -> Result<Vec<RecordBatch>>;

    /// Pushdown scan: prune to rows matching `filter` (file / row-group
    /// granularity — keep a residual per-row guard for exactness) and project
    /// `columns` (empty = all columns, order preserved). Returns an empty result
    /// when the table does not exist yet.
    fn scan_filtered(
        &self,
        table: &str,
        filter: &ScanFilter,
        columns: &[&str],
    ) -> Result<Vec<RecordBatch>>;

    /// Limit / early-break scan: stop once `max_rows` rows are in hand instead
    /// of materializing the whole table (`max_rows == 0` → full scan). Returns
    /// *some* `max_rows` rows, not a deterministic top-N (no ordering guarantee).
    fn scan_limited(&self, table: &str, max_rows: usize) -> Result<Vec<RecordBatch>>;

    // ── table / catalog lifecycle ───────────────────────────────────────────

    /// Create `table` if missing, partitioned by `partition_cols` (identity
    /// transform; empty = unpartitioned). No-op when the table already exists.
    fn ensure_table(
        &self,
        table: &str,
        schema: ::iceberg::spec::Schema,
        partition_cols: &[&str],
    ) -> Result<()>;

    /// Evolve `table`'s stored schema to `canonical` (Iceberg add-column
    /// migration) if it is missing any of `canonical`'s top-level columns.
    /// No-op when the table already carries every canonical column.
    fn ensure_columns(&self, table: &str, canonical: &::iceberg::spec::Schema) -> Result<()>;

    /// Every table name in the warehouse namespace, sorted (catalog metadata
    /// only, no data read).
    fn table_names(&self) -> Result<Vec<String>>;

    // ── named per-table convenience wrappers (thin) ─────────────────────────

    fn append_bench_run(&self, repo: &str, run: &BenchRun) -> Result<Uuid>;
    fn query_bench_runs(&self, filter: &BenchFilter) -> Result<Vec<BenchRun>>;
}

#[derive(Debug, Default, Clone)]
pub struct BenchFilter {
    pub repo: Option<String>,
    pub machine: Option<String>,
    pub limit: Option<usize>,
}

impl BenchFilter {
    pub fn for_repo(repo: impl Into<String>) -> Self {
        Self { repo: Some(repo.into()), machine: None, limit: None }
    }
}

/// Open the configured warehouse. Returns a boxed trait object so
/// callers don't bake an impl into their signatures. All local storage
/// kinds resolve to the Iceberg backend; only `remote` is distinct.
pub fn open(storage: &Storage, workspace_root: &Path) -> Result<Box<dyn Warehouse>> {
    match storage.kind.as_str() {
        "" | "local" | "iceberg" => {
            let root = warehouse_root(storage, workspace_root);
            Ok(Box::new(iceberg::IcebergWarehouse::open(&root)?))
        }
        // #4 Phase 5: a REMOTE warehouse is a pure-Rust git remote of the Iceberg
        // tree — materialize it locally (clone/fetch, exactly the workspace
        // monitor's transport) and open the mirror with the SAME IcebergWarehouse
        // code the local kind uses. Mutations land in the local mirror; pushing
        // the mirror back to the remote is the republish half (git push — a
        // follow-up; the read path is complete here).
        "remote" => {
            let root = remote_warehouse_mirror(&storage.remote_url)?;
            Ok(Box::new(iceberg::IcebergWarehouse::open(&root)?))
        }
        other => anyhow::bail!("unknown storage.kind: {other}"),
    }
}

/// Open the configured warehouse for **read-only** use, tolerating a live
/// `nornir-server` that already holds the exclusive redb lock on
/// `catalog.redb`. When the catalog is locked, this opens a copied-aside
/// read-only snapshot (logged with a WARNING) instead of hard-failing, so a
/// dev-side `nornir` CLI read (the `docs_fresh` gate, docs render, etc.) can
/// never be wedged by the running server. Mutating callers must use
/// [`open`] — the snapshot is read-only by construction.
pub fn open_read_only(storage: &Storage, workspace_root: &Path) -> Result<Box<dyn Warehouse>> {
    match storage.kind.as_str() {
        "" | "local" | "iceberg" => {
            let root = warehouse_root(storage, workspace_root);
            Ok(Box::new(iceberg::IcebergWarehouse::open_read_only(&root)?))
        }
        // #4 Phase 5 READ path: clone/fetch the remote git warehouse to the local
        // mirror, then open it READ-ONLY (lock-tolerant copy-aside) — the thin
        // client's `docs_fresh`/query reads flow through the identical Iceberg
        // reader as the local kind, just sourced over the remote transport.
        "remote" => {
            let root = remote_warehouse_mirror(&storage.remote_url)?;
            Ok(Box::new(iceberg::IcebergWarehouse::open_read_only(&root)?))
        }
        other => anyhow::bail!("unknown storage.kind: {other}"),
    }
}

/// Materialize a REMOTE warehouse locally (#4 Phase 5). The remote warehouse is
/// a pure-Rust git remote of the Iceberg tree (HTTPS→SSH-coerced or SSH, the
/// same transport the workspace monitor clones members over). We clone-or-fetch
/// it into a stable, per-URL cache under `<home>/.nornir/remote-warehouses/<key>`
/// and return that path, so every downstream read/write goes through the SAME
/// local [`iceberg::IcebergWarehouse`] the local kind uses — "mirror the local
/// path, just over the remote transport". Idempotent: a second open fetches into
/// the existing checkout instead of re-cloning.
fn remote_warehouse_mirror(remote_url: &str) -> Result<std::path::PathBuf> {
    let url = remote_url.trim();
    if url.is_empty() {
        anyhow::bail!(
            "storage.kind = \"remote\" needs a [storage].remote_url (the warehouse git remote)"
        );
    }
    let dest = remote_mirror_path(url);
    // SSH remotes drive off the deploy key (like the monitor); http(s) is
    // SSH-coerced by clone_or_fetch, local paths pass through unchanged.
    let ssh_key = if crate::gitio::is_ssh_url(url) {
        crate::gitio::nornir_ssh_key_path()
    } else {
        None
    };
    crate::gitio::clone_or_fetch(url, &dest, ssh_key.as_deref())
        .with_context(|| format!("materialize remote warehouse from `{url}`"))?;
    Ok(dest)
}

/// The local mirror path for a remote warehouse URL — pure (no clone). Cache
/// base: `<home>/.nornir/remote-warehouses` by default, overridable via
/// `NORNIR_REMOTE_WAREHOUSE_CACHE` (operator control + hermetic tests).
fn remote_mirror_path(remote_url: &str) -> std::path::PathBuf {
    let url = remote_url.trim();
    let base = std::env::var_os("NORNIR_REMOTE_WAREHOUSE_CACHE")
        .map(std::path::PathBuf::from)
        .unwrap_or_else(|| crate::config::nornir_home().join("remote-warehouses"));
    base.join(remote_cache_key(url))
}

/// **Write-back / REPUBLISH the remote warehouse** (#4 Phase 5 — the push half).
///
/// After a caller mutates a `kind = "remote"` warehouse (its appends land in the
/// local mirror opened by [`open`]), this commits the mutated Iceberg tree
/// (parquet + catalog) and fast-forward-pushes the mirror back to the remote git
/// warehouse via [`crate::gitio::push_to_local_remote`] — the same pure-Rust git
/// transport, no shell-outs. Non-fast-forward (the remote advanced) fails cleanly
/// without clobbering; re-open the warehouse (which fetches) to reconcile first.
///
/// Local/file remotes are fully supported here. Over-the-network remotes return a
/// clear error (gix 0.84 has no send-pack) — the read path still works; the
/// write-back for network remotes is an out-of-band `git push` until send-pack
/// lands upstream.
pub fn republish_remote(storage: &Storage) -> Result<crate::gitio::LocalPushReport> {
    if storage.kind != "remote" {
        anyhow::bail!("republish_remote: storage.kind is `{}`, not `remote`", storage.kind);
    }
    let url = storage.remote_url.trim();
    if url.is_empty() {
        anyhow::bail!("storage.kind = \"remote\" needs a [storage].remote_url");
    }
    let mirror = remote_mirror_path(url);
    if !mirror.join(".git").exists() {
        anyhow::bail!(
            "no local mirror at {} — open the remote warehouse before republishing",
            mirror.display()
        );
    }
    // Commit the mutated warehouse tree (parquet data files + redb catalog) iff
    // anything changed, so we don't mint empty republish commits.
    if crate::gitio::worktree_freshness(&mirror)?.dirty {
        let msg = format!("nornir: warehouse republish {}", chrono::Utc::now().to_rfc3339());
        crate::gitio::commit_all(&mirror, &msg)
            .context("commit warehouse mirror before push")?;
    }
    crate::gitio::push_to_local_remote(&mirror, url)
        .context("push warehouse mirror back to remote")
}

/// A stable, filesystem-safe cache-dir name for a remote warehouse URL: a short
/// SHA-256 prefix (collision-safe across the handful of remotes) plus a sanitized
/// tail of the URL for human legibility (`ab12cd34-nordisk_warehouse_git`).
fn remote_cache_key(url: &str) -> String {
    use sha2::{Digest, Sha256};
    let mut h = Sha256::new();
    h.update(url.as_bytes());
    let digest = h.finalize();
    let mut hex = String::with_capacity(8);
    for b in digest.iter().take(4) {
        use std::fmt::Write as _;
        let _ = write!(hex, "{b:02x}");
    }
    let sanitized: String = url
        .chars()
        .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
        .collect();
    // A bounded, legible tail (last 40 sanitized chars — the repo/owner end of
    // the URL is the informative part); the SHA prefix guarantees uniqueness.
    let trimmed = sanitized.trim_matches('_');
    let start = trimmed.len().saturating_sub(40);
    let tail = &trimmed[start..];
    format!("{hex}-{tail}")
}

/// Resolve the on-disk warehouse root for a local storage config.
///
/// Precedence (must match `config::Loaded::warehouse_root` and the server's
/// resolution in `bin/nornir-server.rs`):
///   1. explicit `[storage].local_path` → `<workspace_root>/<local_path>/warehouse`
///      (the repo-/workspace-local home; the recommended default for a CLI),
///   2. otherwise the home-derived `<home>/.nornir/warehouse` default
///      (`config::warehouse_default_root`), which the live server also defaults
///      to — a collision risk, hence reads route through [`open_read_only`].
fn warehouse_root(storage: &Storage, workspace_root: &Path) -> std::path::PathBuf {
    if storage.local_path.is_empty() {
        crate::config::warehouse_default_root()
    } else {
        workspace_root.join(&storage.local_path).join("warehouse")
    }
}

#[cfg(test)]
mod remote_warehouse_tests {
    use super::*;
    use crate::bench::{BenchResult, BenchRun};
    use crate::config::Storage;

    fn demo_run() -> BenchRun {
        BenchRun {
            date: "2026-07-08".into(),
            timestamp: Some("2026-07-08T00:00:00Z".into()),
            version: "0.1.0".into(),
            machine: "ci".into(),
            cores: 4,
            results: vec![BenchResult {
                name: "demo".into(),
                metrics: {
                    let mut m = serde_json::Map::new();
                    m.insert("x_mbs".into(), serde_json::json!(123.0));
                    m
                },
                ..Default::default()
            }],
            tests: Vec::new(),
        }
    }

    /// Serialize the tests that mutate the process-global
    /// `NORNIR_REMOTE_WAREHOUSE_CACHE` env var (cargo runs tests in parallel
    /// threads). Poison-tolerant.
    fn env_lock() -> std::sync::MutexGuard<'static, ()> {
        static L: std::sync::Mutex<()> = std::sync::Mutex::new(());
        L.lock().unwrap_or_else(|e| e.into_inner())
    }

    /// #4 Phase 5: `kind = "remote"` with no `remote_url` fails with a helpful
    /// message (NOT the old `not yet implemented (Phase 5)` bail).
    #[test]
    fn remote_kind_without_url_bails_helpfully() {
        let st = Storage { kind: "remote".into(), local_path: String::new(), remote_url: String::new() };
        let e = open(&st, Path::new("/tmp")).map(|_| ()).unwrap_err().to_string();
        assert!(e.contains("remote_url"), "names the missing config: {e}");
        assert!(!e.contains("Phase 5"), "the old stub bail is gone: {e}");
    }

    /// The remote READ path end-to-end over the (network-free) local git
    /// transport: a source warehouse committed to a git repo is cloned into the
    /// per-URL mirror cache and opened read-only through the SAME IcebergWarehouse
    /// reader — the persisted bench run reads back.
    #[test]
    fn remote_read_path_clones_git_warehouse_and_reads_it_back() {
        let _g = env_lock();
        let dir = tempfile::tempdir().unwrap();
        let cache = dir.path().join("cache");
        // Hermetic cache base — never touch the real ~/.nornir.
        unsafe { std::env::set_var("NORNIR_REMOTE_WAREHOUSE_CACHE", &cache); }

        // A real source warehouse with one persisted bench run, then git-commit it.
        let src = dir.path().join("src_wh");
        {
            let wh = iceberg::IcebergWarehouse::open(&src).unwrap();
            wh.append_bench_run("demo", &demo_run()).unwrap();
        } // drop → release the redb lock before committing the tree
        crate::gitio::init(&src).unwrap();
        crate::gitio::commit_all(&src, "seed warehouse").unwrap();

        // Open it as a REMOTE warehouse: clone (local transport) + open read-only.
        let url = src.to_string_lossy().to_string();
        let st = Storage { kind: "remote".into(), local_path: String::new(), remote_url: url.clone() };
        let wh = open_read_only(&st, dir.path()).unwrap();

        // The mirror landed under the hermetic cache and carries the cloned tree.
        let key = remote_cache_key(&url);
        assert!(cache.join(&key).join(".git").exists(), "mirror was cloned into the cache");

        // The persisted run reads back through the remote handle.
        let runs = wh.query_bench_runs(&BenchFilter::for_repo("demo")).unwrap();
        assert!(
            runs.iter().any(|r| r.find("demo").is_some()),
            "the bench run persisted in the remote warehouse reads back over the mirror: {runs:?}",
        );

        unsafe { std::env::remove_var("NORNIR_REMOTE_WAREHOUSE_CACHE"); }
    }

    /// #4 Phase 5 WRITE-BACK end-to-end over the local git transport: open a
    /// remote warehouse, append a NEW bench run into the mirror, `republish_remote`
    /// (commit + fast-forward push back to the remote git warehouse), then open a
    /// FRESH clone (a different cache) and verify the pushed-back mutation is
    /// present — proving the remote warehouse is now read+write, not read-only.
    #[test]
    fn remote_write_back_republish_pushes_mutation_visible_to_a_fresh_clone() {
        let _g = env_lock();
        let dir = tempfile::tempdir().unwrap();

        // The remote warehouse: one bench run, committed to a git repo.
        let src = dir.path().join("src_wh");
        {
            let wh = iceberg::IcebergWarehouse::open(&src).unwrap();
            wh.append_bench_run("demo", &demo_run()).unwrap();
        }
        crate::gitio::init(&src).unwrap();
        crate::gitio::commit_all(&src, "seed warehouse").unwrap();

        let url = src.to_string_lossy().to_string();
        let st =
            Storage { kind: "remote".into(), local_path: String::new(), remote_url: url.clone() };

        // OPEN read-write into mirror-1, append a NEW run (repo `demo2`), drop to
        // release the redb lock, then republish (commit + push back to `src`).
        let cache1 = dir.path().join("cache1");
        unsafe { std::env::set_var("NORNIR_REMOTE_WAREHOUSE_CACHE", &cache1); }
        {
            let wh = open(&st, dir.path()).unwrap();
            wh.append_bench_run("demo2", &demo_run()).unwrap();
        }
        let report = republish_remote(&st).unwrap();
        assert!(!report.created, "the remote branch existed → fast-forward, not create");

        // A FRESH clone (new cache) of the (now-advanced) remote sees `demo2`.
        let cache2 = dir.path().join("cache2");
        unsafe { std::env::set_var("NORNIR_REMOTE_WAREHOUSE_CACHE", &cache2); }
        let wh2 = open_read_only(&st, dir.path()).unwrap();
        let demo2 = wh2.query_bench_runs(&BenchFilter::for_repo("demo2")).unwrap();
        assert!(
            !demo2.is_empty(),
            "the pushed-back mutation (repo `demo2`) is present in a fresh remote clone: {demo2:?}",
        );
        // The original run survived the write-back too.
        let demo = wh2.query_bench_runs(&BenchFilter::for_repo("demo")).unwrap();
        assert!(!demo.is_empty(), "the original run is still present after republish");

        unsafe { std::env::remove_var("NORNIR_REMOTE_WAREHOUSE_CACHE"); }
    }
}