knut-bifrost 0.3.0

//! The `bifrost watch` flags, in the library so `bifrost falkordb install`
//! can render the exact command line its service runs from parsed flags — and
//! a test can parse that rendering back.
//!
//! Feature `cli`. The binary uses these structs as they are.

use std::path::{Path, PathBuf};

use crate::{MappingSpec, mapping::SyncMode};

/// The FalkorDB URL when `--falkor` is absent. A URL with a password belongs here
/// (an installed service gets it through its secret channel), not on argv.
pub const FALKOR_URL_ENV: &str = "BIFROST_FALKOR_URL";
/// The FalkorDB URL when neither `--falkor` nor `BIFROST_FALKOR_URL` names one.
pub const DEFAULT_FALKOR: &str = "redis://127.0.0.1:6379";

/// `--tls-cert <pem> --tls-key <pem> [--tls-ca <pem>]` — the client identity for a FalkorDB
/// running `tls-auth-clients yes`. With `tls-auth-clients-user CN` the certificate's CN IS the
/// user: nothing else is sent. `--tls-ca` verifies the server (default: the system roots).
#[derive(clap::Args, Clone, Debug, Default, PartialEq, Eq)]
pub struct TlsArgs {
    /// Client certificate, PEM.
    #[arg(long, value_name = "PEM", requires = "tls_key")]
    pub tls_cert: Option<PathBuf>,
    /// Client private key, PEM.
    #[arg(long, value_name = "PEM", requires = "tls_cert")]
    pub tls_key: Option<PathBuf>,
    /// CA to verify the server with, PEM.
    #[arg(long, value_name = "PEM")]
    pub tls_ca: Option<PathBuf>,
}

impl TlsArgs {
    pub fn identity(&self) -> Result<Option<graphar_falkordb::FalkorTls>, Box<dyn std::error::Error>> {
        match (&self.tls_cert, &self.tls_key) {
            (Some(c), Some(k)) => Ok(Some(graphar_falkordb::FalkorTls::from_files(c, k, self.tls_ca.as_deref())?)),
            _ => Ok(None),
        }
    }
}

/// The mapping, in whichever format its file is.
#[derive(clap::Args, Clone, Debug, PartialEq, Eq)]
pub struct MappingArgs {
    /// The mapping: bifröst's own YAML/RON (per table, a pipeline of steps and a
    /// `mode`), a PuppyGraph `schema.json` (`.json`: `node` + `edge` lists), or
    /// SQL (`.sql`: one `SELECT` per step, `_label`/`_id` → nodes,
    /// `_type`/`_from_label`/`_from`/`_to_label`/`_to` → edges; feature `sql`).
    #[arg(long)]
    pub mapping: PathBuf,
    /// Sync mode of every table, for a mapping that names none (`.json`, `.sql`):
    /// `cdc` applies inserts, updates and deletes; `incremental` only new rows;
    /// `full` re-reads the table on every change. A YAML/RON mapping names its
    /// own per table and ignores this.
    #[arg(long, default_value = "cdc", value_parser = parse_mode)]
    pub mode: SyncMode,
}

impl MappingArgs {
    pub fn load(&self) -> Result<MappingSpec, Box<dyn std::error::Error>> {
        Ok(crate::mapping::load_mapping(&self.mapping, self.mode)?)
    }
}

pub fn parse_mode(s: &str) -> Result<SyncMode, String> {
    serde_yaml::from_str(s).map_err(|_| format!("{s:?}: one of full, incremental, cdc"))
}

/// Where `watch` finds its Iceberg tables. Exactly one of `--warehouse`,
/// `--s3-root`, `--rest-uri`, `--grpc-uri`; every table the mapping names is
/// looked up there.
#[derive(clap::Args, Clone, Debug, PartialEq, Eq)]
#[group(skip)]
pub struct LakeArgs {
    /// skade Iceberg warehouse directory (`<dir>/catalog.redb` + `warehouse/`).
    #[arg(
        long,
        value_name = "DIR",
        conflicts_with_all = ["s3_root", "rest_uri", "grpc_uri"],
        required_unless_present_any = ["s3_root", "rest_uri", "grpc_uri"]
    )]
    pub warehouse: Option<PathBuf>,
    /// Iceberg tables on S3/MinIO, read with NO catalog: a mapping table `ns.name`
    /// is the table root `<root>/ns/name` (a mapping table written as an `s3://…`
    /// URL is used as-is). Its current metadata is `version-hint.text`, else the
    /// newest `*.metadata.json`. Feature `skade-s3`.
    #[arg(long, value_name = "S3_URL", conflicts_with_all = ["rest_uri", "grpc_uri"])]
    pub s3_root: Option<String>,
    /// An Iceberg REST catalog; a mapping table is `ns.name` in it. MinIO AIStor:
    /// `https://<minio>:9000/_iceberg` with `--sigv4 s3tables` (or `--catalog
    /// aistor`). `--catalog` names a vendor preset. Feature `skade-rest`.
    #[arg(long, value_name = "URL", conflicts_with = "grpc_uri")]
    pub rest_uri: Option<String>,
    /// The REST catalog's `warehouse` (AIStor: the warehouse name, no `s3://`).
    #[arg(long, requires = "rest_uri")]
    pub rest_warehouse: Option<String>,
    /// The namespace a bare mapping table name (`Person`, no `ns.`) is in, with
    /// `--rest-uri` or `--grpc-uri`. A name with a namespace keeps its own.
    #[arg(long, value_name = "NS")]
    pub rest_namespace: Option<String>,
    /// Sign every REST call with AWS SigV4 for this service, using the S3 keys
    /// and region (AIStor: `s3tables`).
    #[arg(long, value_name = "SERVICE", requires = "rest_uri")]
    pub sigv4: Option<String>,
    /// S3/MinIO endpoint of the tables' files, e.g. `https://minio:9000`.
    #[arg(long, value_name = "URL")]
    pub s3_endpoint: Option<String>,
    /// S3 region (MinIO ignores it; SigV4 needs one).
    #[arg(long, default_value = "us-east-1")]
    pub s3_region: String,
    /// Bucket of the tables' files. Defaults to the `--s3-root` bucket, then to
    /// `--rest-warehouse`.
    #[arg(long)]
    pub s3_bucket: Option<String>,
    /// CA (PEM file) to verify an `https://` S3/REST endpoint with; else
    /// `BIFROST_S3_CA_PEM` holds the PEM text.
    #[arg(long, value_name = "PEM")]
    pub s3_ca: Option<PathBuf>,
    /// Which catalog `--rest-uri` is (`--catalog`, OAuth2, headers, vended
    /// credentials), and the `--grpc-uri` catalog.
    #[command(flatten)]
    pub catalog: crate::catalog_args::CatalogArgs,
}

impl LakeArgs {
    /// The default ledger: beside a warehouse, else in the working directory.
    #[cfg(feature = "skade")]
    pub fn default_state(&self) -> PathBuf {
        match &self.warehouse {
            Some(dir) => dir.join(".bifrost-state.json"),
            None => PathBuf::from(".bifrost-state.json"),
        }
    }

    /// The lake the flags name. `state_dir` is where a CA from
    /// `BIFROST_S3_CA_PEM` is written.
    #[cfg(feature = "skade")]
    #[cfg_attr(not(feature = "skade-s3"), allow(unused_variables))]
    pub fn lake(&self, state_dir: &std::path::Path) -> Result<crate::delivery::IcebergLake, Box<dyn std::error::Error>> {
        use crate::delivery::IcebergLake;
        self.catalog.check_unused(self.rest_uri.is_some())?;
        if self.rest_namespace.is_some() && self.rest_uri.is_none() && self.catalog.grpc_uri.is_none() {
            return Err("--rest-namespace needs --rest-uri or --grpc-uri".into());
        }
        if let Some(dir) = &self.warehouse {
            return Ok(IcebergLake::Warehouse(dir.clone()));
        }
        #[cfg(feature = "skade-s3")]
        let s3 = |root_bucket: Option<&str>| -> Result<skade::S3Config, Box<dyn std::error::Error>> {
            use crate::delivery::{s3_ca, s3_config};
            let endpoint = self.s3_endpoint.clone().ok_or("--s3-endpoint is required with --s3-root / --rest-uri")?;
            let bucket = self
                .s3_bucket
                .clone()
                .or_else(|| root_bucket.map(str::to_string))
                .or_else(|| self.rest_warehouse.clone())
                .ok_or("--s3-bucket is required (no bucket in --s3-root or --rest-warehouse)")?;
            let ca = s3_ca(self.s3_ca.as_deref(), state_dir)?;
            Ok(s3_config(endpoint, self.s3_region.clone(), bucket, ca)?)
        };
        if let Some(root) = &self.s3_root {
            #[cfg(feature = "skade-s3")]
            {
                let bucket = crate::delivery::bucket_of(root).ok_or("--s3-root must be an s3://bucket/… URL")?;
                return Ok(IcebergLake::S3 { s3: s3(Some(bucket))?, root: root.clone() });
            }
            #[cfg(not(feature = "skade-s3"))]
            return Err(format!("--s3-root {root}: this build has no S3 reader - rebuild with --features skade-s3").into());
        }
        if let Some(uri) = &self.rest_uri {
            #[cfg(feature = "skade-rest")]
            {
                // bifrost reads the lake; the lake's own writer owns every change.
                let catalog = self.catalog.rest(uri, self.rest_warehouse.clone(), self.sigv4.clone(), s3(None)?, true)?;
                return Ok(IcebergLake::Catalog { connector: std::sync::Arc::new(catalog), namespace: self.rest_namespace.clone() });
            }
            #[cfg(not(feature = "skade-rest"))]
            return Err(format!("--rest-uri {uri}: this build has no REST catalog client - rebuild with --features skade-rest").into());
        }
        if let Some(uri) = &self.catalog.grpc_uri {
            #[cfg(feature = "skade-grpc")]
            {
                // The data: through S3 when `--s3-endpoint` names one, else the local filesystem.
                let data = match self.s3_endpoint {
                    Some(_) => Some(s3(None)?),
                    None => None,
                };
                let catalog = self.catalog.grpc(uri, data, true);
                return Ok(IcebergLake::Catalog { connector: std::sync::Arc::new(catalog), namespace: self.rest_namespace.clone() });
            }
            #[cfg(not(feature = "skade-grpc"))]
            return Err(format!("--grpc-uri {uri}: this build has no gRPC catalog client - rebuild with --features skade-grpc").into());
        }
        Err("name the lake: --warehouse DIR, --s3-root s3://…, --rest-uri URL, or --grpc-uri URL".into())
    }
}


/// `bifrost watch`: an Iceberg lake streamed into a FalkorDB graph.
#[derive(clap::Args, Clone, Debug, PartialEq, Eq)]
pub struct WatchArgs {
    #[command(flatten)]
    pub lake: LakeArgs,
    #[command(flatten)]
    pub mapping: MappingArgs,
    /// FalkorDB / Redis URL (ignored with --dry-run). Default: `BIFROST_FALKOR_URL`,
    /// else `redis://127.0.0.1:6379`. A URL carrying a password belongs in
    /// `BIFROST_FALKOR_URL`, not here (argv is world-readable).
    #[arg(long, value_name = "URL")]
    pub falkor: Option<String>,
    /// Mutual TLS to a FalkorDB that takes nothing else (`--falkor rediss://…`).
    #[command(flatten)]
    pub tls: TlsArgs,
    /// Target graph name.
    #[arg(long, default_value = "bifrost")]
    pub graph: String,
    /// Sync-state ledger path (default: <warehouse>/.bifrost-state.json, and
    /// ./.bifrost-state.json for an S3, REST or gRPC lake). The per-table
    /// last-ingested snapshot cursor lives here, so a restart resumes without
    /// re-ingesting.
    #[arg(long)]
    pub state: Option<PathBuf>,
    /// Run exactly one sync round and exit (else poll forever).
    #[arg(long)]
    pub once: bool,
    /// Poll interval in seconds when looping.
    #[arg(long, default_value_t = 15)]
    pub interval: u64,
    /// Cap (seconds) for the failure backoff (default 10× `--interval`).
    #[arg(long)]
    pub max_backoff: Option<u64>,
    /// On shutdown, write cumulative run metrics as Prometheus text here.
    #[arg(long, value_name = "PATH")]
    pub metrics_file: Option<PathBuf>,
    /// Serve cumulative metrics live over HTTP at `/metrics` on this `ip:port`
    /// (requires the `metrics-http` feature).
    #[arg(long, value_name = "ADDR")]
    pub metrics_addr: Option<String>,
    /// Skip the database: count the graph mutation instead of writing it.
    #[arg(long)]
    pub dry_run: bool,
}

fn push(a: &mut Vec<String>, k: &str, v: Option<String>) {
    if let Some(v) = v {
        a.push(format!("--{k}"));
        a.push(v);
    }
}

fn path(p: &Option<PathBuf>) -> Option<String> {
    p.as_ref().map(|p| p.display().to_string())
}

fn abs(cwd: &Path, p: &Option<PathBuf>) -> Option<PathBuf> {
    p.as_ref().map(|p| if p.is_absolute() { p.clone() } else { cwd.join(p) })
}

impl TlsArgs {
    /// The flags that recreate these.
    pub fn to_args(&self) -> Vec<String> {
        let mut a = Vec::new();
        push(&mut a, "tls-cert", path(&self.tls_cert));
        push(&mut a, "tls-key", path(&self.tls_key));
        push(&mut a, "tls-ca", path(&self.tls_ca));
        a
    }
}

impl MappingArgs {
    /// The flags that recreate these.
    pub fn to_args(&self) -> Vec<String> {
        let mode = serde_yaml::to_string(&self.mode).unwrap_or_default();
        vec!["--mapping".into(), self.mapping.display().to_string(), "--mode".into(), mode.trim().to_string()]
    }
}

impl LakeArgs {
    /// The flags that recreate these — no secrets among them.
    pub fn to_args(&self) -> Vec<String> {
        let mut a = Vec::new();
        push(&mut a, "warehouse", path(&self.warehouse));
        push(&mut a, "s3-root", self.s3_root.clone());
        push(&mut a, "rest-uri", self.rest_uri.clone());
        push(&mut a, "rest-warehouse", self.rest_warehouse.clone());
        push(&mut a, "rest-namespace", self.rest_namespace.clone());
        push(&mut a, "sigv4", self.sigv4.clone());
        push(&mut a, "s3-endpoint", self.s3_endpoint.clone());
        push(&mut a, "s3-region", Some(self.s3_region.clone()));
        push(&mut a, "s3-bucket", self.s3_bucket.clone());
        push(&mut a, "s3-ca", path(&self.s3_ca));
        a.extend(self.catalog.to_args());
        a
    }
}

impl WatchArgs {
    /// The FalkorDB URL: `--falkor`, else `BIFROST_FALKOR_URL`, else the local default.
    pub fn falkor_url(&self) -> String {
        self.falkor
            .clone()
            .or_else(|| std::env::var(FALKOR_URL_ENV).ok().filter(|v| !v.is_empty()))
            .unwrap_or_else(|| DEFAULT_FALKOR.to_string())
    }

    /// `watch …` — the command line that recreates these flags.
    pub fn to_args(&self) -> Vec<String> {
        let mut a = vec!["watch".to_string()];
        a.extend(self.lake.to_args());
        a.extend(self.mapping.to_args());
        push(&mut a, "falkor", self.falkor.clone());
        a.extend(self.tls.to_args());
        push(&mut a, "graph", Some(self.graph.clone()));
        push(&mut a, "state", path(&self.state));
        if self.once {
            a.push("--once".into());
        }
        push(&mut a, "interval", Some(self.interval.to_string()));
        push(&mut a, "max-backoff", self.max_backoff.map(|b| b.to_string()));
        push(&mut a, "metrics-file", path(&self.metrics_file));
        push(&mut a, "metrics-addr", self.metrics_addr.clone());
        if self.dry_run {
            a.push("--dry-run".into());
        }
        a
    }

    /// The same flags with every file path made absolute against `cwd`, so a
    /// service started in another directory reads the same files.
    pub fn absolutized(&self, cwd: &Path) -> Self {
        let mut w = self.clone();
        w.lake.warehouse = abs(cwd, &self.lake.warehouse);
        w.lake.s3_ca = abs(cwd, &self.lake.s3_ca);
        w.lake.catalog = self.lake.catalog.absolutized(cwd);
        w.mapping.mapping = abs(cwd, &Some(self.mapping.mapping.clone())).unwrap_or_default();
        w.tls.tls_cert = abs(cwd, &self.tls.tls_cert);
        w.tls.tls_key = abs(cwd, &self.tls.tls_key);
        w.tls.tls_ca = abs(cwd, &self.tls.tls_ca);
        w.state = abs(cwd, &self.state);
        w.metrics_file = abs(cwd, &self.metrics_file);
        w
    }
}

#[cfg(test)]
mod tests {
    use clap::Parser;

    use super::*;

    #[derive(Parser)]
    struct T {
        #[command(flatten)]
        w: WatchArgs,
    }

    fn parse(args: &[String]) -> WatchArgs {
        T::try_parse_from(std::iter::once("t".to_string()).chain(args.iter().cloned())).unwrap().w
    }

    fn split(s: &str) -> Vec<String> {
        s.split_whitespace().map(str::to_string).collect()
    }

    #[test]
    fn the_watch_command_line_round_trips_through_its_rendering() {
        let w = parse(&split(
            "--rest-uri https://polaris:8181 --catalog polaris --rest-warehouse cat --rest-namespace ns \
             --s3-endpoint https://minio:9000 --s3-bucket lake --rest-header X-T=a \
             --mapping m.yml --mode incremental --falkor rediss://f:6380 --tls-cert c.pem --tls-key k.pem \
             --graph g --state /s/c.json --interval 7 --max-backoff 60 --metrics-addr 127.0.0.1:9090",
        ));
        let rendered = w.to_args();
        assert_eq!(rendered[0], "watch");
        assert_eq!(parse(&rendered[1..]), w);
    }

    #[test]
    fn absolutized_anchors_every_file_and_nothing_else() {
        let w = parse(&split("--warehouse wh --mapping m.yml --tls-ca ca.pem --state st.json --grpc-ca g.pem"))
            .absolutized(Path::new("/srv"));
        assert_eq!(w.lake.warehouse.as_deref(), Some(Path::new("/srv/wh")));
        assert_eq!(w.mapping.mapping, Path::new("/srv/m.yml"));
        assert_eq!(w.tls.tls_ca.as_deref(), Some(Path::new("/srv/ca.pem")));
        assert_eq!(w.state.as_deref(), Some(Path::new("/srv/st.json")));
        assert_eq!(w.lake.catalog.grpc_ca.as_deref(), Some(Path::new("/srv/g.pem")));
        let a = parse(&split("--warehouse /abs --mapping /m.yml")).absolutized(Path::new("/srv"));
        assert_eq!(a.lake.warehouse.as_deref(), Some(Path::new("/abs")));
    }

    #[test]
    fn the_flag_wins_over_the_env_and_the_default_is_local() {
        let w = parse(&split("--warehouse /w --mapping /m.yml --falkor redis://x:1"));
        assert_eq!(w.falkor_url(), "redis://x:1");
    }
}