nornir 0.5.3

//! 🏢 **Warehouse Deck** — the whole Iceberg warehouse seen as MANY components at
//! once. Unlike the 🗄 Warehouse browser (one table → one grid), this tab hosts
//! a [`facett_warehousedeck::WarehouseDeck`]: a wall of N panes, each binding a
//! warehouse table (or a join) to a view-kind — several **3D graphs** (call graph,
//! dependency graph, …), grids, and charts, all visible simultaneously.
//!
//! nornir OWNS the warehouse + the viz; facett OWNS the rendering. This pane is
//! the glue: it opens the warehouse read-only, reads each table, maps the rows to
//! a facett view-model via [`WarehouseRegistry`], and drives the deck. modgunn's
//! security tables (`sbom_components`, `vuln_findings`, …) live in the same
//! `nornir` namespace and are rendered here exactly like nornir's own.
//!
//! Read-only open coexists with a running `nornir-server` (it falls back to a
//! catalog snapshot when the server holds the redb lock).

use std::path::PathBuf;
use std::sync::Arc;

use arrow::array::{Array, ArrayRef, StringArray};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use eframe::egui;

use facett_warehousedeck::{
    ChartSpec, Facet, GraphLayout, GraphSpec, PaneData, WarehouseDeck, WarehouseRegistry,
};

use crate::warehouse::iceberg::{IcebergWarehouse, TablePreview};
use crate::warehouse::Warehouse;

use super::facett_theme::{Theme, RED};

/// Per-table row caps so a huge table never blows up a 3D layout or a grid.
const GRAPH_ROW_CAP: usize = 1500;
const GRID_ROW_CAP: usize = 500;
const CHART_ROW_CAP: usize = 200;

/// Where the deck reads from — **local** (opens the Iceberg warehouse directly
/// for raw Arrow frames) OR **remote** (goes through the server's
/// `Warehouse.Tables` / `Warehouse.Scan` gRPC, exactly like the 🗄 Warehouse
/// browser). In thin/`NORNIR_SERVER` mode the deck is built from the server's
/// stringified scan rows re-materialised as Utf8 Arrow batches, so the 3D deck
/// works over gRPC AND local, like every other pane.
enum DeckSource {
    Local(PathBuf),
    Remote { endpoint: String, token: String, workspace: String },
}

/// Uniform table access for the deck's `build()` — hides local-vs-remote so the
/// SAME curated pane list drives both. Local returns real typed Arrow frames;
/// remote fetches the server scan (stringified rows) and re-materialises them as
/// a single all-Utf8 batch (columns keep their names, so `edge_graph`'s
/// by-name column lookup and the grid still work; `numeric_chart` parses the
/// string cells).
enum DeckScanner {
    Local(IcebergWarehouse),
    Remote { endpoint: String, token: String, workspace: String },
}

impl DeckScanner {
    fn table_names(&self) -> Vec<String> {
        match self {
            DeckScanner::Local(wh) => wh.table_names().unwrap_or_default(),
            DeckScanner::Remote { endpoint, token, workspace } => {
                super::remote::fetch_tables(endpoint, token, workspace).unwrap_or_default()
            }
        }
    }

    /// Scan `table` (capped) to Arrow frames. Remote rows arrive stringified and
    /// are wrapped in a single Utf8 batch; a scan error yields an empty Vec so a
    /// single bad table never sinks the whole deck.
    fn scan(&self, table: &str, cap: usize) -> Vec<RecordBatch> {
        match self {
            DeckScanner::Local(wh) => wh.scan_limited(table, cap).unwrap_or_default(),
            DeckScanner::Remote { endpoint, token, workspace } => {
                match super::remote::scan_table(endpoint, token, table, cap as u32, workspace) {
                    Ok(p) => preview_to_batches(&p),
                    Err(_) => Vec::new(),
                }
            }
        }
    }
}

/// Re-materialise a stringified [`TablePreview`] (the `Warehouse.Scan` RPC shape)
/// as ONE all-Utf8 [`RecordBatch`], preserving column names. Empty preview →
/// empty Vec.
fn preview_to_batches(p: &TablePreview) -> Vec<RecordBatch> {
    if p.columns.is_empty() {
        return Vec::new();
    }
    let ncols = p.columns.len();
    let mut cols: Vec<Vec<Option<String>>> = vec![Vec::with_capacity(p.rows.len()); ncols];
    for row in &p.rows {
        for (c, cell) in cols.iter_mut().enumerate() {
            cell.push(row.get(c).cloned());
        }
    }
    let fields: Vec<Field> =
        p.columns.iter().map(|n| Field::new(n, DataType::Utf8, true)).collect();
    let arrays: Vec<ArrayRef> = cols
        .into_iter()
        .map(|c| Arc::new(StringArray::from(c)) as ArrayRef)
        .collect();
    match RecordBatch::try_new(Arc::new(Schema::new(fields)), arrays) {
        Ok(b) => vec![b],
        Err(_) => Vec::new(),
    }
}

pub struct WarehouseDeckPane {
    source: DeckSource,
    registry: WarehouseRegistry,
    deck: Option<WarehouseDeck>,
    built: bool,
    err: Option<String>,
    /// Every table name found in the warehouse (for the state dump).
    tables: Vec<String>,
    theme: Theme,
}

impl WarehouseDeckPane {
    /// Build the deck from a local warehouse dir.
    pub fn local(root: PathBuf) -> Self {
        Self::with(DeckSource::Local(root))
    }

    /// Build the deck from a remote server's warehouse over gRPC
    /// (`Warehouse.Tables` / `Warehouse.Scan`), scoped to `workspace` — the same
    /// path the 🗄 Warehouse browser uses in thin mode.
    pub fn remote(endpoint: String, token: String, workspace: String) -> Self {
        Self::with(DeckSource::Remote { endpoint, token, workspace })
    }

    fn with(source: DeckSource) -> Self {
        Self {
            source,
            registry: WarehouseRegistry::default_nordisk(),
            deck: None,
            built: false,
            err: None,
            tables: Vec::new(),
            theme: Theme::default(),
        }
    }

    pub fn set_palette(&mut self, t: Theme) {
        self.theme = t;
    }

    /// **Test seam** (mirrors the other viz panes' `inject_for_test`): set the deck
    /// directly, bypassing warehouse I/O, so a robot test asserts the render +
    /// `state_json` without seeding a real catalog.
    pub fn inject_for_test(&mut self, deck: WarehouseDeck) {
        self.tables = deck.titles().iter().map(|s| s.to_string()).collect();
        self.deck = Some(deck);
        self.err = None;
        self.built = true;
    }

    /// The registry that maps table → view-kind (extend with `register(...)`).
    pub fn registry_mut(&mut self) -> &mut WarehouseRegistry {
        &mut self.registry
    }

    /// 🏢 Warehouse Deck's slice of `state_json` (LAW #6): the composed deck state
    /// (pane count, per-kind counts, each pane's nodes/edges/rows + the per-pane
    /// widget state + the `trace.ran` ledger) plus this pane's error + the number
    /// of warehouse tables discovered.
    pub fn state_json(&self) -> serde_json::Value {
        match &self.deck {
            Some(d) => {
                let mut v = d.state_json();
                if let serde_json::Value::Object(map) = &mut v {
                    map.insert("error".into(), serde_json::json!(self.err));
                    map.insert("tables_in_warehouse".into(), serde_json::json!(self.tables.len()));
                    map.insert("built".into(), serde_json::json!(self.built));
                }
                v
            }
            None => serde_json::json!({
                "built": self.built,
                "error": self.err,
                "pane_count": 0,
                "graph3d_count": 0,
                "grid_count": 0,
                "chart_count": 0,
                "tables_in_warehouse": self.tables.len(),
            }),
        }
    }

    /// Open the warehouse read-only and build the deck. Curated initial slice:
    /// call graph + dependency graph (3D), the bench/test/security grids, and an
    /// event chart — each added only when its table exists with rows. Other tables
    /// fall back to the registry's per-table default.
    fn build(&mut self) {
        self.built = true;
        // Open the table source — local warehouse (raw frames) or the server's
        // Warehouse gRPC (thin mode), so the SAME curated deck works both ways.
        let scanner = match &self.source {
            DeckSource::Local(p) => match IcebergWarehouse::open_read_only(p) {
                Ok(w) => DeckScanner::Local(w),
                Err(e) => {
                    self.err = Some(format!(
                        "open warehouse failed: {e:#}\n(a running nornir-server holds the redb \
                         lock; a snapshot fallback is used when possible)"
                    ));
                    return;
                }
            },
            DeckSource::Remote { endpoint, token, workspace } => DeckScanner::Remote {
                endpoint: endpoint.clone(),
                token: token.clone(),
                workspace: workspace.clone(),
            },
        };
        let tables = scanner.table_names();
        self.tables = tables.clone();
        let has = |t: &str| tables.iter().any(|x| x == t);

        let mut deck = WarehouseDeck::new("🏢 Warehouse Deck").with_cols(2);

        // ── 3D GRAPHS (joins) ────────────────────────────────────────────────
        if has("call_edges") {
            if let Some(g) = edge_graph(&scanner, "call_edges", "caller_path", "callee_ident", GraphLayout::Force) {
                deck.add_for("call graph", "call_edges", PaneData::Graph(g));
            }
        }
        if has("dep_graph_edges") {
            if let Some(g) = edge_graph(&scanner, "dep_graph_edges", "from_repo", "to_repo", GraphLayout::Sphere) {
                deck.add_for("dependency graph", "dep_graph_edges", PaneData::Graph(g));
            }
        }
        if has("scip_call_edges") {
            if let Some(g) = edge_graph(&scanner, "scip_call_edges", "caller_symbol", "callee_symbol", GraphLayout::Force) {
                deck.add_for("resolved call graph", "scip_call_edges", PaneData::Graph(g));
            }
        }

        // ── GRIDS (tabular) ──────────────────────────────────────────────────
        for t in ["bench_runs", "test_outcomes", "vuln_findings", "sbom_components"] {
            if !has(t) {
                continue;
            }
            let batches = scanner.scan(t, GRID_ROW_CAP);
            let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
            if rows > 0 {
                deck.add_for(t, t, PaneData::Grid(batches));
            }
        }

        // ── CHART (event/time-series) ────────────────────────────────────────
        for t in ["git_heat_facts", "bench_telemetry", "release_lineage"] {
            if !has(t) {
                continue;
            }
            if let Some(c) = numeric_chart(&scanner, t) {
                deck.add_for(t, t, PaneData::Chart(c));
                break;
            }
        }

        if deck.pane_count() == 0 {
            self.err = Some(
                "no renderable tables yet — populate the workspace (nornir populate) so \
                 call_edges / dep_graph_edges / bench_runs accrue rows"
                    .into(),
            );
        }
        self.deck = Some(deck);
    }

    pub fn draw(&mut self, ui: &mut egui::Ui) {
        if !self.built {
            self.build();
        }
        ui.horizontal(|ui| {
            ui.heading("🏢 Warehouse Deck");
            ui.weak(format!("· {} tables in warehouse", self.tables.len()));
            if ui.button("↻ rebuild").clicked() {
                self.built = false;
                self.deck = None;
            }
        });
        ui.separator();
        if let Some(err) = &self.err {
            ui.colored_label(RED, err);
        }
        if let Some(deck) = &mut self.deck {
            deck.ui(ui);
        }
    }
}

/// Build a 3D graph from an edge table's `src`/`dst` string columns. Distinct
/// endpoints become nodes (coloured by [`facett_warehousedeck::hash_color`]); each
/// row an edge. `None` if the table is empty / lacks the columns.
fn edge_graph(scanner: &DeckScanner, table: &str, src: &str, dst: &str, layout: GraphLayout) -> Option<GraphSpec> {
    let batches = scanner.scan(table, GRAPH_ROW_CAP);
    let mut idx: std::collections::HashMap<String, usize> = std::collections::HashMap::new();
    let mut nodes: Vec<(String, egui::Color32)> = Vec::new();
    let mut edges: Vec<(usize, usize)> = Vec::new();
    for b in &batches {
        let (Ok(si), Ok(di)) = (b.schema().index_of(src), b.schema().index_of(dst)) else {
            continue;
        };
        let (Some(sc), Some(dc)) = (
            b.column(si).as_any().downcast_ref::<StringArray>(),
            b.column(di).as_any().downcast_ref::<StringArray>(),
        ) else {
            continue;
        };
        for r in 0..b.num_rows() {
            if sc.is_null(r) || dc.is_null(r) {
                continue;
            }
            let s = intern(sc.value(r), &mut nodes, &mut idx);
            let d = intern(dc.value(r), &mut nodes, &mut idx);
            edges.push((s, d));
        }
    }
    if nodes.is_empty() {
        return None;
    }
    Some(GraphSpec::new(nodes, edges).with_layout(layout))
}

/// Intern a node label → stable index, creating + colouring it on first sight.
fn intern(
    label: &str,
    nodes: &mut Vec<(String, egui::Color32)>,
    idx: &mut std::collections::HashMap<String, usize>,
) -> usize {
    if let Some(&i) = idx.get(label) {
        return i;
    }
    let i = nodes.len();
    nodes.push((label.to_string(), facett_warehousedeck::hash_color(label)));
    idx.insert(label.to_string(), i);
    i
}

/// Build a bars chart from a table's first numeric column (value) vs row index.
///
/// Local frames are typed, so the first `Int64/Int32/Float64` column is used.
/// Remote frames arrive all-Utf8 (the server stringifies the scan), so when no
/// typed-numeric column exists we fall back to the first Utf8 column whose cells
/// parse as `f64` — so the chart pane renders over gRPC too.
fn numeric_chart(scanner: &DeckScanner, table: &str) -> Option<ChartSpec> {
    use arrow::array::{Float64Array, Int32Array, Int64Array};
    let batches = scanner.scan(table, CHART_ROW_CAP);
    let first = batches.first()?;

    // Prefer a typed-numeric column (unchanged local behaviour); else the first
    // Utf8 column that parses numerically (remote path).
    let typed = first.schema().fields().iter().enumerate().find_map(|(i, f)| {
        use arrow::datatypes::DataType::*;
        matches!(f.data_type(), Int64 | Int32 | Float64).then(|| (i, f.name().clone()))
    });
    let (col, name) = match typed {
        Some(c) => c,
        None => first.schema().fields().iter().enumerate().find_map(|(i, f)| {
            let parses = first
                .column(i)
                .as_any()
                .downcast_ref::<StringArray>()
                .map(|a| (0..a.len()).any(|r| a.is_valid(r) && a.value(r).parse::<f64>().is_ok()))
                .unwrap_or(false);
            parses.then(|| (i, f.name().clone()))
        })?,
    };

    let mut points: Vec<(f64, f64)> = Vec::new();
    let mut x = 0.0;
    for b in &batches {
        if b.num_columns() <= col {
            continue;
        }
        let column = b.column(col);
        for r in 0..b.num_rows() {
            let y = if let Some(a) = column.as_any().downcast_ref::<Int64Array>() {
                a.is_valid(r).then(|| a.value(r) as f64)
            } else if let Some(a) = column.as_any().downcast_ref::<Int32Array>() {
                a.is_valid(r).then(|| a.value(r) as f64)
            } else if let Some(a) = column.as_any().downcast_ref::<Float64Array>() {
                a.is_valid(r).then(|| a.value(r))
            } else if let Some(a) = column.as_any().downcast_ref::<StringArray>() {
                a.is_valid(r).then(|| a.value(r).parse::<f64>().ok()).flatten()
            } else {
                None
            };
            if let Some(y) = y {
                points.push((x, y));
                x += 1.0;
            }
        }
    }
    if points.is_empty() {
        return None;
    }
    Some(ChartSpec::bars(vec![(name, points)]))
}

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

    /// The RPC→Arrow bridge (N11): a stringified `Warehouse.Scan` preview must
    /// re-materialise as one all-Utf8 batch that preserves column names and
    /// row/cell values, so `edge_graph`'s by-name lookup and the grid render
    /// over gRPC exactly as they do locally.
    #[test]
    fn preview_materialises_to_utf8_batch() {
        let preview = TablePreview {
            columns: vec!["caller_path".into(), "callee_ident".into()],
            rows: vec![
                vec!["a.rs".into(), "foo".into()],
                vec!["b.rs".into(), "bar".into()],
            ],
        };
        let batches = preview_to_batches(&preview);
        assert_eq!(batches.len(), 1, "one batch for a non-empty preview");
        let b = &batches[0];
        assert_eq!(b.num_rows(), 2);
        assert_eq!(b.num_columns(), 2);
        assert_eq!(b.schema().index_of("caller_path").unwrap(), 0);
        assert_eq!(b.schema().index_of("callee_ident").unwrap(), 1);
        let col = b
            .column(0)
            .as_any()
            .downcast_ref::<StringArray>()
            .expect("Utf8 column");
        assert_eq!(col.value(0), "a.rs");
        assert_eq!(col.value(1), "b.rs");
    }

    /// An edge graph built from a remote-style Utf8 batch (via the scan bridge)
    /// interns distinct endpoints into nodes + one edge per row — the deck's
    /// 3D call graph over gRPC.
    #[test]
    fn edge_graph_from_utf8_batch_has_nodes_and_edges() {
        let preview = TablePreview {
            columns: vec!["from_repo".into(), "to_repo".into()],
            rows: vec![
                vec!["nornir".into(), "facett".into()],
                vec!["nornir".into(), "holger".into()],
            ],
        };
        let batches = preview_to_batches(&preview);
        // Reproduce edge_graph's core over the materialised batch.
        let b = &batches[0];
        let si = b.schema().index_of("from_repo").unwrap();
        let di = b.schema().index_of("to_repo").unwrap();
        let sc = b.column(si).as_any().downcast_ref::<StringArray>().unwrap();
        let dc = b.column(di).as_any().downcast_ref::<StringArray>().unwrap();
        let mut nodes: std::collections::HashSet<String> = Default::default();
        let mut edges = 0usize;
        for r in 0..b.num_rows() {
            nodes.insert(sc.value(r).to_string());
            nodes.insert(dc.value(r).to_string());
            edges += 1;
        }
        assert_eq!(edges, 2);
        assert_eq!(nodes.len(), 3, "nornir, facett, holger");
    }

    #[test]
    fn empty_preview_is_no_batches() {
        let preview = TablePreview { columns: vec![], rows: vec![] };
        assert!(preview_to_batches(&preview).is_empty());
    }
}