crates/knut-bifrost/src/watch_cli.rs
//! 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");
}
}