knut-bifrost 0.3.0

//! `bifrost falkordb …` — `bifrost watch` (an Iceberg lake streamed into a
//! FalkorDB graph) as a systemd service and as a Kubernetes workload, through
//! skidbladnir: the house's one system installer.
//!
//! Feature `service`.
//!
//! ```text
//! bifrost falkordb unit      <watch flags> [--name N] [--system]   # print the unit `install` writes
//! bifrost falkordb install   <watch flags> [--name N] [--system]   # write + enable + start
//! bifrost falkordb uninstall [--name N] [--system] [--purge]
//! bifrost falkordb start|stop|status [--name N] [--system]
//! bifrost falkordb k8s       <watch flags> --image I [--namespace NS] --out DIR
//! ```
//!
//! `<watch flags>` are exactly `bifrost watch`'s: the lake (`--warehouse`,
//! `--s3-root`, `--rest-uri` with `--catalog`, `--grpc-uri`), `--mapping`,
//! `--falkor`, `--graph`, TLS, interval, metrics.
//!
//! # One spec, two packages
//!
//! [`service_spec`] builds ONE `skidbladnir::service::ServiceSpec`; `install`
//! hands it to skidbladnir's systemd `Lifecycle`, `k8s` to its
//! `KubernetesRenderer` (a StatefulSet, a volume for the cursor, a Secret for
//! the credentials). The command line is rendered from the parsed flags
//! ([`WatchArgs::to_args`]), never copied from argv. The cursor is always the
//! service's own (`<state dir>/cursor.json`), so `--state`, `--once` and
//! `--dry-run` are refused: a service polls forever into a real graph.
//!
//! # Secrets never go on the command line
//!
//! [`SECRET_ENV`] — the S3 keys, the S3 CA as PEM text, the FalkorDB URL when it
//! carries a password (`BIFROST_FALKOR_URL`), the catalog's OAuth2 credential
//! and token — are read from the installer's environment and routed to
//! skidbladnir's secret channel: a 0600 `EnvironmentFile=` for systemd, a
//! Secret for Kubernetes. A `--falkor` URL with credentials in it is refused.
//!
//! # Kubernetes and files
//!
//! skidbladnir renders no ConfigMap, so every file the watch reads — the
//! mapping, PEMs — must already be in the image: `k8s` takes their paths as
//! paths **inside the image** and refuses relative ones. The S3 CA can come as
//! PEM text through `BIFROST_S3_CA_PEM` instead.

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

use clap::{Args, Subcommand};

use crate::catalog_args::{CREDENTIAL_ENV, TOKEN_ENV};
use crate::service_paths::{IMAGE_EXEC, K8S_STATE_DIR, installed_state_dir};
use crate::error::{BifrostError, Result};
use crate::watch_cli::{FALKOR_URL_ENV, WatchArgs};

/// The secrets a watch reads from its environment. Never flags.
pub const SECRET_ENV: [&str; 6] = [
    "MINIO_ACCESS_KEY",
    "MINIO_SECRET_KEY",
    "BIFROST_S3_CA_PEM",
    FALKOR_URL_ENV,
    CREDENTIAL_ENV,
    TOKEN_ENV,
];

/// The default service / workload name.
pub const DEFAULT_NAME: &str = "bifrost-falkordb";
const DESCRIPTION: &str = "bifrost: Iceberg lake into a FalkorDB graph";

#[derive(Subcommand, Debug, Clone)]
pub enum FalkordbCommand {
    /// Print the systemd unit `install` would write.
    Unit(ServiceArgs),
    /// Write, enable and start the systemd unit (skidbladnir).
    Install(ServiceArgs),
    /// Stop, disable and remove the unit. The cursor stays unless `--purge`.
    Uninstall(LifecycleArgs),
    /// Start the installed unit.
    Start(LifecycleArgs),
    /// Stop the installed unit.
    Stop(LifecycleArgs),
    /// Print `active` or `inactive`.
    Status(LifecycleArgs),
    /// Render the Kubernetes package (StatefulSet + Secret + volume) into `--out`.
    K8s(K8sArgs),
}

/// `unit` / `install`: what to watch, and as which service.
#[derive(Args, Debug, Clone, PartialEq, Eq)]
pub struct ServiceArgs {
    #[command(flatten)]
    pub watch: WatchArgs,
    #[command(flatten)]
    pub life: LifecycleArgs,
}

/// Which installed service a lifecycle verb acts on.
#[derive(Args, Debug, Clone, PartialEq, Eq)]
pub struct LifecycleArgs {
    /// Service name (one per lake → graph stream).
    #[arg(long, default_value = DEFAULT_NAME)]
    pub name: String,
    /// System scope (its own `bifrost` user); default is the user scope.
    #[arg(long)]
    pub system: bool,
    /// With `uninstall`: also erase the cursor.
    #[arg(long)]
    pub purge: bool,
}

/// `k8s`: the Kubernetes package.
#[derive(Args, Debug, Clone, PartialEq, Eq)]
pub struct K8sArgs {
    #[command(flatten)]
    pub watch: WatchArgs,
    /// Workload name.
    #[arg(long, default_value = DEFAULT_NAME)]
    pub name: String,
    /// The image carrying `/usr/local/bin/bifrost` (and the mapping file).
    #[arg(long)]
    pub image: String,
    #[arg(long, default_value = "default")]
    pub namespace: String,
    /// Size of the cursor volume.
    #[arg(long, default_value = "1Gi")]
    pub state_size: String,
    /// Directory the manifests are written into (the Secret file is 0600).
    #[arg(long, value_name = "DIR")]
    pub out: PathBuf,
}

fn env(name: &str) -> Option<String> {
    std::env::var(name).ok().filter(|v| !v.is_empty())
}

/// Whether a `redis://user:pass@host` URL carries credentials.
fn has_userinfo(url: &str) -> bool {
    let rest = url.split_once("://").map_or(url, |(_, r)| r);
    rest.split(['/', '?', '#']).next().is_some_and(|authority| authority.contains('@'))
}

/// The watch a service runs: refused flags checked, the cursor its own.
pub fn service_watch(w: &WatchArgs, state: &Path) -> Result<WatchArgs> {
    let refuse = |what: &str| Err(BifrostError::Other(format!("bifrost falkordb: {what}")));
    if w.once {
        return refuse("--once makes no service (it would exit after one round) - use `bifrost watch --once`");
    }
    if w.dry_run {
        return refuse("--dry-run makes no service (it writes no graph) - use `bifrost watch --dry-run`");
    }
    if w.state.is_some() {
        return refuse("--state: an installed service keeps its cursor in its own state dir");
    }
    if let Some(url) = &w.falkor
        && has_userinfo(url)
    {
        return refuse(&format!(
            "--falkor carries a password, and argv is world-readable - set {FALKOR_URL_ENV} instead (it goes to the secret channel)"
        ));
    }
    Ok(WatchArgs { state: Some(state.to_path_buf()), ..w.clone() })
}

/// The ONE spec both packages are rendered from. `exec` is the binary path the
/// target runs; `watch` the already-checked flags (see [`service_watch`]).
/// Secrets are copied from the installer's environment into the secret
/// channel, never into the unit.
pub fn service_spec(name: &str, exec: &str, watch: &WatchArgs, system: bool) -> skidbladnir::service::ServiceSpec {
    use skidbladnir::service::{AgentMode, ServiceSpec};
    use skidbladnir::systemd::{RestartPolicy, Scope};
    let mut spec = ServiceSpec::new(name, DESCRIPTION, exec)
        .scope(if system { Scope::System } else { Scope::User })
        .restart(RestartPolicy::OnFailure)
        .agent(AgentMode::Fat)
        .env("RUST_LOG", "info");
    if let Some(dir) = watch.state.as_deref().and_then(Path::parent) {
        spec = spec.working_dir(dir.to_string_lossy());
    }
    if system {
        spec = spec.user("bifrost");
    }
    for a in watch.to_args() {
        spec = spec.arg(a);
    }
    for k in SECRET_ENV {
        if let Some(v) = env(k) {
            spec = spec.secret_env(k, v);
        }
    }
    spec
}

fn scope(system: bool) -> skidbladnir::systemd::Scope {
    if system { skidbladnir::systemd::Scope::System } else { skidbladnir::systemd::Scope::User }
}

/// The systemd lifecycle for `life.name`, its spec built from `watch` when given.
fn lifecycle(watch: Option<&WatchArgs>, life: &LifecycleArgs) -> Result<skidbladnir::service::Lifecycle> {
    let exec = std::env::current_exe()
        .map(|p| p.to_string_lossy().into_owned())
        .unwrap_or_else(|_| IMAGE_EXEC.into());
    let state_dir = installed_state_dir(&life.name, life.system);
    let mut spec = match watch {
        Some(w) => {
            let cwd = std::env::current_dir()?;
            let w = service_watch(&w.absolutized(&cwd), &state_dir.join("cursor.json"))?;
            service_spec(&life.name, &exec, &w, life.system)
        }
        None => skidbladnir::service::ServiceSpec::new(&life.name, DESCRIPTION, &exec).scope(scope(life.system)),
    };
    // skidbladnir's uninstall erases a declared data root (its law: data dies).
    // The cursor is what lets a reinstall resume without re-ingesting, so it is
    // declared only when the operator asked for `--purge`.
    if life.purge {
        let root = skidbladnir::uninstall::DataRoot::from_env("BIFROST_FALKORDB_DATA", state_dir)
            .map_err(|e| BifrostError::Other(format!("state dir: {e}")))?;
        spec = spec.data_root(&root);
    }
    Ok(skidbladnir::service::Lifecycle::new(spec))
}

fn sk(e: anyhow::Error) -> BifrostError {
    BifrostError::Other(format!("skidbladnir: {e:#}"))
}

/// The files a pod reads must be in the image: absolute paths only.
fn in_image(w: &WatchArgs) -> Result<()> {
    let c = &w.lake.catalog;
    let files = [
        Some(&w.mapping.mapping),
        w.lake.warehouse.as_ref(),
        w.lake.s3_ca.as_ref(),
        w.tls.tls_cert.as_ref(),
        w.tls.tls_key.as_ref(),
        w.tls.tls_ca.as_ref(),
        c.grpc_ca.as_ref(),
        c.grpc_cert.as_ref(),
        c.grpc_key.as_ref(),
        w.metrics_file.as_ref(),
    ];
    for f in files.into_iter().flatten() {
        if !f.is_absolute() {
            return Err(BifrostError::Other(format!(
                "bifrost falkordb k8s: {} is relative - name it as a path inside the image (skidbladnir mounts no ConfigMap)",
                f.display()
            )));
        }
    }
    Ok(())
}

/// The Kubernetes package: the same spec as `install`, run from the image's
/// binary with its cursor on a volume.
pub fn k8s_files(args: &K8sArgs) -> Result<Vec<skidbladnir::service::RenderedFile>> {
    use skidbladnir::service::{ServiceRenderer as _, Storage};
    in_image(&args.watch)?;
    let w = service_watch(&args.watch, &Path::new(K8S_STATE_DIR).join("cursor.json"))?;
    let spec = service_spec(&args.name, IMAGE_EXEC, &w, false)
        .image(&args.image)
        .namespace(&args.namespace)
        .replicas(1)
        .stateful(true)
        .storage(Storage::new(&args.state_size, K8S_STATE_DIR));
    Ok(skidbladnir::k8s::KubernetesRenderer.render(&spec).files)
}

/// Dispatch `bifrost falkordb <verb>`.
pub fn dispatch(cmd: FalkordbCommand) -> Result<()> {
    match cmd {
        FalkordbCommand::Unit(a) => {
            let lc = lifecycle(Some(&a.watch), &a.life)?;
            print!("{}", crate::service_paths::unit_text(&lc, &a.life.name, a.life.system));
            Ok(())
        }
        FalkordbCommand::Install(a) => {
            let lc = lifecycle(Some(&a.watch), &a.life)?;
            crate::service_paths::install(&lc, &a.life.name, a.life.system).map_err(sk)?;
            lc.enable().map_err(sk)?;
            lc.start().map_err(sk)?;
            println!("bifrost falkordb: INSTALLED and started ({})", lc.unit());
            Ok(())
        }
        FalkordbCommand::Uninstall(l) => {
            let r = lifecycle(None, &l)?.uninstall().map_err(sk)?;
            let env = crate::service_paths::remove_env_file(&l.name, l.system)?;
            println!(
                "bifrost falkordb: unit removed = {}, env file removed = {env}, erased {} path(s)",
                r.unit_removed,
                r.erased.len()
            );
            Ok(())
        }
        FalkordbCommand::Start(l) => lifecycle(None, &l)?.start().map_err(sk),
        FalkordbCommand::Stop(l) => lifecycle(None, &l)?.stop().map_err(sk),
        FalkordbCommand::Status(l) => {
            println!("{}", if lifecycle(None, &l)?.status().map_err(sk)? { "active" } else { "inactive" });
            Ok(())
        }
        FalkordbCommand::K8s(a) => {
            let files = k8s_files(&a)?;
            std::fs::create_dir_all(&a.out)?;
            for f in files {
                let file = a.out.join(Path::new(&f.path).file_name().unwrap_or(f.path.as_ref()));
                std::fs::write(&file, &f.contents)?;
                #[cfg(unix)]
                {
                    use std::os::unix::fs::PermissionsExt;
                    std::fs::set_permissions(&file, std::fs::Permissions::from_mode(f.mode))?;
                }
                println!("{}", file.display());
            }
            Ok(())
        }
    }
}

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

    use super::*;

    #[derive(Parser)]
    struct T {
        #[command(subcommand)]
        cmd: FalkordbCommand,
    }

    fn parse(s: &str) -> std::result::Result<FalkordbCommand, clap::Error> {
        T::try_parse_from(std::iter::once("t").chain(s.split_whitespace())).map(|t| t.cmd)
    }

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

    const LAKE: &str = "--rest-uri https://polaris:8181 --catalog polaris --rest-warehouse cat \
        --s3-endpoint https://minio:9000 --s3-bucket lake --mapping /etc/bifrost/m.yml \
        --falkor rediss://falkor:6380 --graph social";

    #[test]
    fn the_unit_runs_watch_with_its_own_cursor_and_parses_back() {
        let Ok(FalkordbCommand::Install(a)) = parse(&format!("install {LAKE} --system")) else { panic!() };
        assert_eq!(a.life.name, DEFAULT_NAME);
        let state = installed_state_dir(&a.life.name, true).join("cursor.json");
        let w = service_watch(&a.watch, &state).unwrap();
        let spec = service_spec(&a.life.name, "/usr/bin/bifrost", &w, true);
        let unit = spec.render_systemd_unit();
        let args = w.to_args();
        assert_eq!(args[0], "watch");
        for x in &args {
            assert!(unit.contains(x.as_str()), "the unit carries {x}");
        }
        assert!(unit.contains("/var/lib/bifrost-falkordb/cursor.json"));
        let back = W::try_parse_from(std::iter::once("t".to_string()).chain(args.into_iter().skip(1))).unwrap().w;
        assert_eq!(back, w, "what the service runs parses back to what was asked for");
    }

    #[test]
    fn once_dry_run_state_and_a_password_in_the_url_are_refused() {
        for extra in ["--once", "--dry-run", "--state /tmp/s.json"] {
            let Ok(FalkordbCommand::Unit(a)) = parse(&format!("unit {LAKE} {extra}")) else { panic!() };
            assert!(service_watch(&a.watch, Path::new("/s/c.json")).is_err(), "{extra} refused");
        }
        let lake = LAKE.replace("rediss://falkor:6380", "redis://:pw@falkor:6379");
        let Ok(FalkordbCommand::Unit(a)) = parse(&format!("unit {lake}")) else { panic!() };
        let e = service_watch(&a.watch, Path::new("/s/c.json")).unwrap_err().to_string();
        assert!(e.contains(FALKOR_URL_ENV) && !e.contains(":pw@"), "{e}");
    }

    #[test]
    fn userinfo_is_found_only_in_the_authority() {
        assert!(has_userinfo("redis://:pw@h:6379"));
        assert!(has_userinfo("rediss://u:p@h"));
        assert!(!has_userinfo("redis://h:6379"));
        assert!(!has_userinfo("redis://h:6379/0?x=a@b"));
    }

    #[test]
    fn systemd_and_k8s_run_the_same_command_and_no_secret_is_on_it() {
        // SAFETY of the test: the env vars are set only in this process.
        unsafe {
            std::env::set_var(CREDENTIAL_ENV, "id:hunter2-not-on-argv");
            std::env::set_var(FALKOR_URL_ENV, "redis://:hunter3@f:6379");
        }
        let Ok(FalkordbCommand::K8s(k)) = parse(&format!("k8s {LAKE} --image reg/bifrost:1 --out /tmp/x")) else {
            panic!()
        };
        let state = Path::new(K8S_STATE_DIR).join("cursor.json");
        let w = service_watch(&k.watch, &state).unwrap();
        let unit = service_spec(&k.name, IMAGE_EXEC, &w, false).render_systemd_unit();
        let files = k8s_files(&k).unwrap();
        let workload = files.iter().find(|f| f.contents.contains("kind: StatefulSet")).expect("a StatefulSet");
        for a in w.to_args() {
            assert!(workload.contents.contains(&a), "k8s args carry {a}");
            assert!(unit.contains(&a), "the unit carries {a}");
        }
        for secret in ["hunter2", "hunter3"] {
            assert!(!unit.contains(secret), "a secret never lands in the 0644 unit");
            assert!(!workload.contents.contains(secret), "a secret never lands in the workload");
        }
        let secret = files.iter().find(|f| f.contents.contains("kind: Secret")).expect("a Secret for the credentials");
        assert_eq!(secret.mode, 0o600);
        assert!(workload.contents.contains(K8S_STATE_DIR), "the cursor lives on the volume");
    }

    #[test]
    fn k8s_refuses_a_file_that_cannot_be_in_the_image() {
        let lake = LAKE.replace("/etc/bifrost/m.yml", "m.yml");
        let Ok(FalkordbCommand::K8s(k)) = parse(&format!("k8s {lake} --image i --out /tmp/x")) else { panic!() };
        assert!(k8s_files(&k).unwrap_err().to_string().contains("inside the image"));
    }

    #[test]
    fn lifecycle_verbs_take_only_a_name() {
        let Ok(FalkordbCommand::Uninstall(l)) = parse("uninstall --name g1 --system --purge") else { panic!() };
        assert_eq!(l, LifecycleArgs { name: "g1".into(), system: true, purge: true });
        assert!(parse("status --mapping m.yml").is_err());
    }
}