nornir 0.5.4

//! **RemoteWarehouse** — a [`Warehouse`](super::Warehouse) trait impl backed by a
//! running `nornir-server`'s generic warehouse gRPC (`Warehouse.Tables` /
//! `Warehouse.Scan`, package `nornir.v1`), NOT a local Iceberg tree.
//!
//! This is the second concrete backend behind the durable [`Warehouse`] seam
//! (the first is [`iceberg::IcebergWarehouse`](super::iceberg::IcebergWarehouse)).
//! Its whole reason to exist is to **prove the trait is a real seam**: a thin
//! client that owns no redb catalog and no Parquet, yet satisfies the same trait
//! the ~150 sync call sites drive — so a future `nornir` running purely as a
//! thin client can read the warehouse a server owns without any call site
//! knowing it is not local.
//!
//! ## What is wired (Phase 3 skeleton)
//!
//! The server holds the single-writer redb lock, so a thin client is a
//! **reader**. The read surface that the `Warehouse.Tables`/`Warehouse.Scan`
//! RPCs already expose maps 1:1 onto the trait's *browser* methods:
//!
//! | trait method            | transport                                   |
//! |-------------------------|---------------------------------------------|
//! | [`table_names`]         | `Warehouse.Tables` RPC                      |
//! | [`scan_preview`]        | `Warehouse.Scan` RPC (stringified rows)     |
//! | [`scan_limited`]        | `Warehouse.Scan` (limit) → all-`Utf8` batch |
//! | [`scan_arrow`]          | `Warehouse.Scan` (server default) → batch   |
//!
//! [`table_names`]: super::Warehouse::table_names
//! [`scan_preview`]: super::Warehouse::scan_preview
//! [`scan_limited`]: super::Warehouse::scan_limited
//! [`scan_arrow`]: super::Warehouse::scan_arrow
//!
//! The `Warehouse.Scan` RPC returns *stringified* cells (the display preview the
//! egui browser renders), so the reconstructed [`RecordBatch`] is honestly
//! **all-`Utf8`** — types are lost on the wire. A typed remote read (Arrow
//! Flight, or a per-table BenchRuns/… RPC) is the documented follow-up; until it
//! lands, [`scan_filtered`](super::Warehouse::scan_filtered) (needs real
//! predicate pushdown) and [`query_bench_runs`](super::Warehouse::query_bench_runs)
//! (needs the typed row) return a typed [`RemoteWarehouseError::Unsupported`].
//!
//! ## What is NOT wired (typed errors, by design)
//!
//! Every **write** ([`append_arrow`](super::Warehouse::append_arrow),
//! [`ensure_table`](super::Warehouse::ensure_table),
//! [`ensure_columns`](super::Warehouse::ensure_columns),
//! [`append_bench_run`](super::Warehouse::append_bench_run)) returns
//! [`RemoteWarehouseError::Unsupported`]: a thin client never opens the warehouse
//! to write it — it *submits* rows via the `Telemetry.*` RPCs and the server
//! appends them (the contract in `.nornir/data-submit-rpcs.md`). These are typed
//! errors with a clear TODO, never silent no-ops (L2).

use std::sync::Arc;

use anyhow::Result;
use arrow::array::{ArrayRef, RecordBatch, StringArray};
use arrow::datatypes::{DataType, Field, Schema as ArrowSchema};
use uuid::Uuid;

use super::iceberg::TablePreview;
use super::{BenchFilter, ScanFilter, Warehouse};
use crate::bench::BenchRun;

/// The generated `nornir.v1` prost/tonic bindings (message types always; the
/// `warehouse_client` is present whenever tonic is — the same `server`/`viz`/`mcp`
/// gate that compiles this module).
mod pb {
    tonic::include_proto!("nornir.v1");
}

/// Why a [`RemoteWarehouse`] method is not (yet) serviceable — a **typed** error
/// so callers can match on it instead of string-sniffing an `anyhow` chain.
#[derive(Debug, Clone, thiserror::Error)]
pub enum RemoteWarehouseError {
    /// The method is not wired on the remote backend yet. `todo` names the
    /// concrete follow-up (a new RPC, Arrow Flight, or the write-submit path).
    #[error("RemoteWarehouse::{method} is not wired on the thin client yet — {todo}")]
    Unsupported {
        /// The trait method name (e.g. `"append_arrow"`).
        method: &'static str,
        /// The concrete follow-up that would wire it.
        todo: &'static str,
    },
}

fn unsupported(method: &'static str, todo: &'static str) -> anyhow::Error {
    RemoteWarehouseError::Unsupported { method, todo }.into()
}

/// A thin `Warehouse` client that reads a server-owned warehouse over gRPC.
///
/// Holds a dedicated current-thread tokio runtime so the **sync** trait methods
/// can `block_on` the async RPC without a "runtime within a runtime" panic — the
/// exact shape [`IcebergWarehouse`](super::iceberg::IcebergWarehouse) uses for
/// its own runtime, so the ~150 sync call sites stay sync no matter which backend
/// is behind the trait object.
pub struct RemoteWarehouse {
    endpoint: String,
    token: String,
    workspace: String,
    rt: tokio::runtime::Runtime,
}

impl RemoteWarehouse {
    /// Connect a thin warehouse client to `endpoint` (an `http://host:port`, or a
    /// bare `host:port` that is `http://`-prefixed). `token` is the bearer auth
    /// (empty ⇒ no real auth); `workspace` selects the served workspace (empty ⇒
    /// the server default) via the `nornir-workspace` metadata header. Building
    /// the runtime is the only fallible step — the channel is dialed lazily per
    /// RPC (matching `viz::remote`), so `connect` never blocks on the network.
    pub fn connect(
        endpoint: impl Into<String>,
        token: impl Into<String>,
        workspace: impl Into<String>,
    ) -> Result<Self> {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .map_err(|e| anyhow::anyhow!("build tokio runtime for RemoteWarehouse: {e}"))?;
        Ok(Self { endpoint: normalize_endpoint(&endpoint.into()), token: token.into(), workspace: workspace.into(), rt })
    }

    /// The configured endpoint (normalized, `http://`-prefixed).
    pub fn endpoint(&self) -> &str {
        &self.endpoint
    }

    // ── the two networked reads: Warehouse.Tables / Warehouse.Scan ───────────

    fn tables_rpc(&self) -> Result<pb::WarehouseTables> {
        let (endpoint, token, workspace) = (self.endpoint.clone(), self.token.clone(), self.workspace.clone());
        self.rt.block_on(async move {
            let mut client = connect_client(&endpoint, &token, &workspace).await?;
            Ok(client
                .tables(pb::Empty {})
                .await
                .map_err(|s| anyhow::anyhow!("Warehouse.Tables RPC: {s}"))?
                .into_inner())
        })
    }

    fn scan_rpc(&self, table: &str, limit: u32) -> Result<pb::WarehouseScan> {
        let (endpoint, token, workspace, table) =
            (self.endpoint.clone(), self.token.clone(), self.workspace.clone(), table.to_string());
        self.rt.block_on(async move {
            let mut client = connect_client(&endpoint, &token, &workspace).await?;
            Ok(client
                .scan(pb::WarehouseScanRequest { table, limit })
                .await
                .map_err(|s| anyhow::anyhow!("Warehouse.Scan RPC: {s}"))?
                .into_inner())
        })
    }
}

// ── pure, network-free mappers (unit-testable without a server) ─────────────

/// Map a `Warehouse.Tables` response to the trait's sorted table-name list.
pub fn tables_to_names(t: pb::WarehouseTables) -> Vec<String> {
    let mut names = t.names;
    names.sort();
    names
}

/// Map a `Warehouse.Scan` response (columns + stringified rows) to a
/// [`TablePreview`] — the exact struct `IcebergWarehouse::scan_preview` returns,
/// so the generic warehouse browser is source-agnostic.
pub fn scan_to_preview(s: pb::WarehouseScan) -> TablePreview {
    TablePreview { columns: s.columns, rows: s.rows.into_iter().map(|r| r.cells).collect() }
}

/// Reconstruct an **all-`Utf8`** [`RecordBatch`] from a stringified preview. Types
/// were lost on the `Warehouse.Scan` wire, so every column is `Utf8` (honest —
/// a typed remote read is the Flight follow-up). Empty columns ⇒ no batch.
pub fn preview_to_string_batch(p: &TablePreview) -> Result<Vec<RecordBatch>> {
    if p.columns.is_empty() {
        return Ok(Vec::new());
    }
    let fields: Vec<Field> =
        p.columns.iter().map(|c| Field::new(c, DataType::Utf8, true)).collect();
    let schema = Arc::new(ArrowSchema::new(fields));
    let ncols = p.columns.len();
    let cols: Vec<ArrayRef> = (0..ncols)
        .map(|ci| {
            let vals: Vec<Option<String>> =
                p.rows.iter().map(|r| r.get(ci).cloned()).collect();
            Arc::new(StringArray::from(vals)) as ArrayRef
        })
        .collect();
    Ok(vec![RecordBatch::try_new(schema, cols)?])
}

fn normalize_endpoint(endpoint: &str) -> String {
    if endpoint.starts_with("http") {
        endpoint.to_string()
    } else {
        format!("http://{endpoint}")
    }
}

/// Dial a `WarehouseClient` with the shared bearer + `nornir-workspace` header
/// interceptor (the same connect/auth shape as `viz::remote`).
async fn connect_client(
    endpoint: &str,
    token: &str,
    workspace: &str,
) -> Result<
    pb::warehouse_client::WarehouseClient<
        tonic::service::interceptor::InterceptedService<
            tonic::transport::Channel,
            impl FnMut(tonic::Request<()>) -> Result<tonic::Request<()>, tonic::Status>,
        >,
    >,
> {
    let channel = tonic::transport::Channel::from_shared(endpoint.to_string())
        .map_err(|e| anyhow::anyhow!("invalid server url `{endpoint}`: {e}"))?
        .connect()
        .await
        .map_err(|e| anyhow::anyhow!("connect to nornir-server at {endpoint}: {e}"))?;
    let bearer: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>> =
        (!token.is_empty()).then(|| format!("Bearer {token}").parse().ok()).flatten();
    let ws: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>> =
        (!workspace.is_empty()).then(|| workspace.parse().ok()).flatten();
    Ok(pb::warehouse_client::WarehouseClient::with_interceptor(
        channel,
        move |mut req: tonic::Request<()>| {
            if let Some(b) = &bearer {
                req.metadata_mut().insert("authorization", b.clone());
            }
            if let Some(w) = &ws {
                req.metadata_mut().insert("nornir-workspace", w.clone());
            }
            Ok(req)
        },
    ))
}

#[async_trait::async_trait]
impl Warehouse for RemoteWarehouse {
    // ── generic Arrow core ──────────────────────────────────────────────────

    fn append_arrow(&self, _table: &str, _batch: RecordBatch) -> Result<()> {
        Err(unsupported(
            "append_arrow",
            "the server is the single writer; submit rows via the Telemetry.* RPCs \
             (see .nornir/data-submit-rpcs.md), not the trait write path",
        ))
    }

    fn scan_arrow(&self, table: &str) -> Result<Vec<RecordBatch>> {
        // limit 0 ⇒ the server's default page (~200/500 rows). Honest all-Utf8.
        let scan = self.scan_rpc(table, 0)?;
        preview_to_string_batch(&scan_to_preview(scan))
    }

    fn scan_filtered(
        &self,
        _table: &str,
        _filter: &ScanFilter,
        _columns: &[&str],
    ) -> Result<Vec<RecordBatch>> {
        Err(unsupported(
            "scan_filtered",
            "the Warehouse.Scan RPC has no predicate pushdown; wire an Arrow Flight \
             read (or a filtered-scan RPC) so the server prunes files server-side",
        ))
    }

    fn scan_limited(&self, table: &str, max_rows: usize) -> Result<Vec<RecordBatch>> {
        let limit = u32::try_from(max_rows).unwrap_or(u32::MAX);
        let scan = self.scan_rpc(table, limit)?;
        preview_to_string_batch(&scan_to_preview(scan))
    }

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

    fn ensure_table(
        &self,
        _table: &str,
        _schema: ::iceberg::spec::Schema,
        _partition_cols: &[&str],
    ) -> Result<()> {
        Err(unsupported(
            "ensure_table",
            "table/catalog lifecycle is the server's (single-writer) job; the thin \
             client cannot create tables — the server ensures them on first append",
        ))
    }

    fn ensure_columns(&self, _table: &str, _canonical: &::iceberg::spec::Schema) -> Result<()> {
        Err(unsupported(
            "ensure_columns",
            "schema evolution is a server-side write; not exposed over the thin gRPC",
        ))
    }

    fn table_names(&self) -> Result<Vec<String>> {
        Ok(tables_to_names(self.tables_rpc()?))
    }

    fn scan_preview(&self, table: &str, limit: usize) -> Result<TablePreview> {
        let limit = u32::try_from(limit).unwrap_or(u32::MAX);
        Ok(scan_to_preview(self.scan_rpc(table, limit)?))
    }

    fn describe_columns(&self, _table: &str) -> Result<Vec<super::sql::ColumnInfo>> {
        Err(unsupported(
            "describe_columns",
            "column TYPES are lost on the stringified Warehouse.Scan wire (every \
             cell is Utf8); a typed describe needs an Arrow Flight read or a \
             dedicated Describe RPC",
        ))
    }

    // ── named per-table convenience wrappers ────────────────────────────────

    fn append_bench_run(&self, _repo: &str, _run: &BenchRun) -> Result<Uuid> {
        Err(unsupported(
            "append_bench_run",
            "submit bench runs via Telemetry.SubmitBakeoff/SubmitTestResults; the \
             server owns the write",
        ))
    }

    fn query_bench_runs(&self, _filter: &BenchFilter) -> Result<Vec<BenchRun>> {
        Err(unsupported(
            "query_bench_runs",
            "needs a typed row over the wire (the Warehouse.Scan preview is \
             stringified); add a BenchRuns RPC or an Arrow Flight read",
        ))
    }

    // ── async surface ────────────────────────────────────────────────────────
    // The thin client is a READ-ONLY, server-owned-write backend: every async
    // write is Unsupported (submit via the Telemetry.* RPCs), and the async
    // reads have no pushdown/typed wire yet (Arrow Flight follow-up), matching
    // how the sync `scan_filtered`/`query_bench_runs` are already typed-Unsupported.

    async fn append_arrow_async(&self, _table: &str, _batch: RecordBatch) -> Result<()> {
        Err(unsupported(
            "append_arrow_async",
            "the server is the single writer; submit rows via the Telemetry.* RPCs",
        ))
    }

    async fn scan_arrow_async(&self, _table: &str) -> Result<Vec<RecordBatch>> {
        Err(unsupported(
            "scan_arrow_async",
            "the async release readers need a typed Arrow read; wire an Arrow \
             Flight scan (the sync scan_arrow returns a stringified preview only)",
        ))
    }

    async fn append_bench_run_async(&self, _repo: &str, _run: &BenchRun) -> Result<Uuid> {
        Err(unsupported(
            "append_bench_run_async",
            "submit bench runs via Telemetry.SubmitBakeoff/SubmitTestResults; the \
             server owns the write",
        ))
    }

    async fn query_bench_runs_async(&self, _filter: &BenchFilter) -> Result<Vec<BenchRun>> {
        Err(unsupported(
            "query_bench_runs_async",
            "needs a typed row over the wire; add a BenchRuns RPC or an Arrow Flight read",
        ))
    }

    async fn append_symbol_scan_async(
        &self,
        _scan: &crate::knowledge::symbols::SymbolScan,
    ) -> Result<()> {
        Err(unsupported(
            "append_symbol_scan_async",
            "server-owned write; submit knowledge scans server-side",
        ))
    }

    async fn append_git_heat_scan_async(
        &self,
        _scan: &crate::knowledge::git_heat::GitHeatScan,
    ) -> Result<()> {
        Err(unsupported(
            "append_git_heat_scan_async",
            "server-owned write; submit knowledge scans server-side",
        ))
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
}

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

    fn demo_scan() -> pb::WarehouseScan {
        pb::WarehouseScan {
            columns: vec!["repo".into(), "n".into()],
            rows: vec![
                pb::WarehouseRow { cells: vec!["holger".into(), "3".into()] },
                pb::WarehouseRow { cells: vec!["znippy".into(), "20".into()] },
            ],
        }
    }

    /// RED-when-broken: the `Warehouse.Scan` response maps 1:1 onto a
    /// [`TablePreview`] — every column header and every row cell survives. A
    /// mapper that dropped a column or a row (or transposed cells) fails here.
    #[test]
    fn remote_scan_response_maps_to_preview() {
        let p = scan_to_preview(demo_scan());
        assert_eq!(p.columns, vec!["repo".to_string(), "n".to_string()]);
        assert_eq!(p.rows.len(), 2, "both rows survive");
        assert_eq!(p.rows[0], vec!["holger".to_string(), "3".to_string()]);
        assert_eq!(p.rows[1], vec!["znippy".to_string(), "20".to_string()]);
    }

    /// The stringified preview reconstructs as an all-`Utf8` [`RecordBatch`] with
    /// the right shape (columns → fields, rows → values). RED-when-broken: a
    /// mapper that lost the second column or a cell fails the shape asserts.
    #[test]
    fn remote_preview_reconstructs_utf8_batch() {
        let p = scan_to_preview(demo_scan());
        let batches = preview_to_string_batch(&p).unwrap();
        assert_eq!(batches.len(), 1);
        let b = &batches[0];
        assert_eq!(b.num_columns(), 2, "both columns present");
        assert_eq!(b.num_rows(), 2, "both rows present");
        assert_eq!(b.schema().field(0).name(), "repo");
        assert_eq!(b.schema().field(1).name(), "n");
        assert!(matches!(b.schema().field(0).data_type(), DataType::Utf8));
        let repo = b.column(0).as_any().downcast_ref::<StringArray>().unwrap();
        assert_eq!(repo.value(0), "holger");
        assert_eq!(repo.value(1), "znippy");
        let n = b.column(1).as_any().downcast_ref::<StringArray>().unwrap();
        assert_eq!(n.value(0), "3");
        assert_eq!(n.value(1), "20");
    }

    /// `Warehouse.Tables` names come back sorted (the trait contract).
    #[test]
    fn remote_tables_response_sorts_names() {
        let t = pb::WarehouseTables { names: vec!["z_table".into(), "a_table".into(), "m".into()] };
        assert_eq!(tables_to_names(t), vec!["a_table".to_string(), "m".to_string(), "z_table".to_string()]);
    }

    /// **The seam proof.** A `RemoteWarehouse` is a real, boxable
    /// [`Warehouse`] — it can stand behind the trait object exactly where an
    /// `IcebergWarehouse` does. Its writes are typed
    /// [`RemoteWarehouseError::Unsupported`] errors (NEVER silent no-ops), which
    /// is what a thin reader must be. RED-when-broken: if a write silently
    /// returned `Ok(())`, or the error stopped being typed, this fails.
    #[test]
    fn remote_warehouse_is_a_boxed_warehouse_with_typed_write_errors() {
        let wh: Box<dyn Warehouse> =
            Box::new(RemoteWarehouse::connect("127.0.0.1:9", "", "").unwrap());
        let schema = Arc::new(ArrowSchema::new(vec![Field::new("x", DataType::Utf8, true)]));
        let batch = RecordBatch::new_empty(schema);
        let err = wh.append_arrow("t", batch).unwrap_err();
        let typed = err.downcast_ref::<RemoteWarehouseError>();
        assert!(
            matches!(typed, Some(RemoteWarehouseError::Unsupported { method: "append_arrow", .. })),
            "append_arrow is a TYPED Unsupported error, got: {err:#}"
        );
        // query_bench_runs is likewise a typed error, not a silent empty vec.
        let err = wh.query_bench_runs(&BenchFilter::default()).unwrap_err();
        assert!(
            matches!(
                err.downcast_ref::<RemoteWarehouseError>(),
                Some(RemoteWarehouseError::Unsupported { method: "query_bench_runs", .. })
            ),
            "query_bench_runs is a TYPED Unsupported error, got: {err:#}"
        );
        // describe_columns (typed schema discovery) is likewise a typed error on
        // the thin client — column types are lost on the stringified scan wire.
        let err = wh.describe_columns("bench_runs").unwrap_err();
        assert!(
            matches!(
                err.downcast_ref::<RemoteWarehouseError>(),
                Some(RemoteWarehouseError::Unsupported { method: "describe_columns", .. })
            ),
            "describe_columns is a TYPED Unsupported error, got: {err:#}"
        );
    }
}