corium-cli 0.1.39

//! `corium` — launchers and admin commands for the distributed topology:
//! `transactor`, `peer-server`, `db *`, `console`, backup/restore, `gc`, and `log`.

mod authz;
mod console;
mod metrics_http;
mod pg_catalog;
mod sql;
mod tui;

use std::net::SocketAddr;
use std::path::PathBuf;
use std::process::ExitCode;
use std::sync::Arc;
use std::time::Duration;

use clap::{Args, Parser, Subcommand, ValueEnum};
use corium_authz::{AuthzConfig, BreakGlass, SystemDbAuthorizer};
use corium_core::KeywordInterner;
use corium_peer::server::PeerServerConfig;
use corium_peer::{Admin, ConnectConfig, Connection, IndexPolicySettings};
use corium_protocol::auth::{DEFAULT_DEV_TOKEN, client_tls, server_tls};
use corium_protocol::authz::{
    ActionClass, AllowAll, Authorizer, CompositeProvider, Guard, IdentityProvider, Principal,
    StaticTokens,
};
use corium_protocol::codec;
use corium_query::edn::{Edn, read_all};
use corium_store::{DbRoot, FsStore, RootStore};
use corium_transactor::StoreSpec;
use corium_transactor::node::{NodeConfig, TransactorNode};

/// Corium database system command line.
#[derive(Parser)]
#[command(name = "corium", version, about)]
struct Cli {
    /// Log rendering for tracing events.
    #[arg(long, global = true, value_enum, default_value_t = LogFormat::Human)]
    log_format: LogFormat,
    #[command(subcommand)]
    command: Command,
}

#[derive(Clone, Copy, Debug, ValueEnum)]
enum LogFormat {
    Human,
    Json,
}

/// Storage-service backend for a transactor's blobs and roots.
#[derive(Clone, Copy, Debug, Default, ValueEnum)]
enum StoreKind {
    /// In-memory, ephemeral (single process); the whole database is lost on
    /// exit. Useful for demos and tests.
    Mem,
    /// Filesystem under `--data-dir` (blobs, roots, and logs).
    #[default]
    Fs,
    /// `PostgreSQL` for blobs and roots; the log stays on the local filesystem
    /// under `--data-dir`. Requires the `postgres` feature.
    Postgres,
    /// Turso (embeddable `SQLite`) for blobs and roots; the log stays on the
    /// local filesystem under `--data-dir`. Requires the `turso` feature.
    Turso,
    /// S3 (or an S3-compatible service) for blobs and roots; the log stays on
    /// the local filesystem under `--data-dir`. Requires the `s3` feature.
    /// Credentials, region, and endpoint come from the standard AWS
    /// environment (`AWS_ACCESS_KEY_ID`, `AWS_REGION`, `AWS_ENDPOINT_URL`,
    /// etc.).
    S3,
}

/// Client-side connection flags (endpoint, auth, TLS).
#[derive(Args, Clone)]
struct ClientFlags {
    /// Transactor endpoint, e.g. `http://127.0.0.1:4334`. With an HA pair,
    /// pass a comma-separated preference list (active first); connections
    /// fail over across it. Admin commands use the first endpoint.
    #[arg(long, default_value = "http://127.0.0.1:4334")]
    transactor: String,
    /// Bearer token for the transactor. Defaults to the shared development
    /// token (`corium_protocol::auth::DEFAULT_DEV_TOKEN`) so a local database
    /// needs no auth flags; pass `--token ""` to connect anonymously, or set
    /// `CORIUM_TOKEN` to use a different secret everywhere.
    #[arg(long, env = "CORIUM_TOKEN")]
    token: Option<String>,
    /// PEM file with a CA certificate to trust (enables TLS).
    #[arg(long)]
    ca: Option<PathBuf>,
    /// Domain name expected on the server certificate.
    #[arg(long)]
    tls_domain: Option<String>,
    /// Bootstrap from the transactor's storage backend directly, discovered
    /// through its `GetStorageInfo` RPC, instead of replaying the transaction
    /// log from basis zero. Requires network reach to that backend and a build
    /// with the matching storage feature.
    #[arg(long)]
    peer_bootstrap: bool,
}

impl ClientFlags {
    /// Endpoint preference list parsed from the comma-separated flag.
    fn endpoints(&self) -> Vec<String> {
        self.transactor
            .split(',')
            .map(|endpoint| endpoint.trim().to_owned())
            .filter(|endpoint| !endpoint.is_empty())
            .collect()
    }

    /// The bearer token to present: the flag/`CORIUM_TOKEN` value when set, the
    /// shared development token when unset, or `None` when explicitly cleared
    /// with `--token ""` (connect anonymously).
    fn token(&self) -> Option<String> {
        resolve_client_token(self.token.as_deref())
    }

    /// First endpoint (admin commands talk to one transactor).
    fn primary(&self) -> String {
        self.endpoints()
            .into_iter()
            .next()
            .unwrap_or_else(|| self.transactor.clone())
    }

    fn tls(&self) -> Result<Option<tonic::transport::ClientTlsConfig>, String> {
        if self.ca.is_none() && self.tls_domain.is_none() {
            return Ok(None);
        }
        client_tls(self.ca.as_deref(), self.tls_domain.as_deref())
            .map(Some)
            .map_err(|error| format!("cannot load CA certificate: {error}"))
    }

    async fn connect_config(&self, db: impl Into<String>) -> Result<ConnectConfig, String> {
        let db = db.into();
        let mut config = ConnectConfig::with_failover(self.endpoints(), db.clone());
        config.token = self.token();
        config.tls = self.tls()?;
        if !self.peer_bootstrap {
            return Ok(config);
        }
        // Ask the transactor how its storage service is configured, then open
        // it directly so the peer bootstraps from the newest published
        // snapshot. The transactor address is the only thing the client needs.
        let mut admin = Admin::connect(&self.primary(), self.token(), self.tls()?)
            .await
            .map_err(|error| format!("cannot connect to transactor: {error}"))?;
        let info = admin
            .get_storage_info(&db)
            .await
            .map_err(|error| format!("cannot fetch storage info: {error}"))?;
        let storage = info
            .storage
            .ok_or_else(|| "transactor returned no storage backend".to_owned())?;
        let (spec, data_dir) = StoreSpec::from_connection(storage)
            .map_err(|error| format!("cannot use transactor storage backend: {error}"))?;
        let store = corium_transactor::NodeStore::open_existing(&spec, &data_dir)
            .await
            .map_err(|error| format!("cannot open peer storage: {error}"))?;
        Ok(config.with_storage(Arc::new(store)))
    }
}

/// Server-side TLS/auth flags.
///
/// Authentication is a permissive default so a local server is usable with no
/// flags: it recognizes the shared development token (and any client presenting
/// it) but *also* admits anonymous callers. Naming a real secret with
/// `--serve-token`, requiring the default token with `--require-auth`, or
/// configuring an OIDC issuer all switch the server to strict mode, where an
/// unrecognized or absent credential is rejected. Authorization defaults to
/// permit-all ([`AllowAll`]) unless `--authz-db` names a policy database, in
/// which case every request is authorized against the relationship policy it
/// holds.
#[derive(Args, Clone)]
struct ServeFlags {
    /// Require this exact bearer token from clients (strict: rejects anonymous
    /// and every other credential). Overrides the shared development token.
    #[arg(long, env = "CORIUM_SERVE_TOKEN")]
    serve_token: Option<String>,
    /// Require the shared development token (or `--serve-token`) on every
    /// request: reject anonymous callers without changing the accepted secret.
    #[arg(long)]
    require_auth: bool,
    /// Disable authentication entirely: accept every request as anonymous.
    #[arg(long, conflicts_with_all = ["serve_token", "require_auth", "oidc_issuer"])]
    serve_open: bool,
    /// OIDC issuer URL. Tokens signed by this issuer are accepted (in addition
    /// to the static token). Requires a build with the `oidc-discovery` feature
    /// (to fetch the issuer's JWKS) or `--oidc-jwks-file` with the `oidc`
    /// feature. Enabling OIDC switches the server to strict mode.
    #[arg(long)]
    oidc_issuer: Option<String>,
    /// Accepted OIDC audience (`aud`); repeatable. Strongly recommended when an
    /// issuer is set.
    #[arg(long = "oidc-audience")]
    oidc_audiences: Vec<String>,
    /// Read the issuer's JWKS from this file instead of fetching it (offline;
    /// works with the `oidc` feature alone).
    #[arg(long)]
    oidc_jwks_file: Option<PathBuf>,
    /// Authorize every request against the self-hosted relationship policy in
    /// this database (create it with `corium authz init`). Without it,
    /// authorization is permit-all.
    #[arg(long, env = "CORIUM_AUTHZ_DB")]
    authz_db: Option<String>,
    /// Re-read the policy database before deciding write and admin actions, so
    /// a control-plane change is never decided from a stale snapshot. Reads
    /// keep using the pinned snapshot either way.
    #[arg(long, requires = "authz_db")]
    authz_fresh_writes: bool,
    /// Role admitted while the policy database cannot be read at all
    /// (repeatable). Break-glass is for operator recovery: it does not
    /// override a policy that denies, only one that is unavailable.
    #[arg(long = "authz-break-glass-role", requires = "authz_db")]
    authz_break_glass_roles: Vec<String>,
    /// Maximum relation hops one authorization check may walk.
    #[arg(long, default_value_t = 8, requires = "authz_db")]
    authz_max_depth: usize,
    /// PEM certificate chain for TLS.
    #[arg(long, requires = "tls_key")]
    tls_cert: Option<PathBuf>,
    /// PEM private key for TLS.
    #[arg(long, requires = "tls_cert")]
    tls_key: Option<PathBuf>,
}

impl ServeFlags {
    fn tls(&self) -> Result<Option<tonic::transport::ServerTlsConfig>, String> {
        match (&self.tls_cert, &self.tls_key) {
            (Some(cert), Some(key)) => server_tls(cert, key)
                .map(Some)
                .map_err(|error| format!("cannot load TLS identity: {error}")),
            _ => Ok(None),
        }
    }

    /// The authorization database and tuning the `--authz-*` flags select.
    ///
    /// # Errors
    /// Rejects `--authz-db` together with `--serve-open`, which would
    /// authorize requests whose identity was never established.
    fn authz(&self) -> Result<Option<(String, AuthzConfig)>, String> {
        let Some(db) = self.authz_db.clone() else {
            return Ok(None);
        };
        if self.serve_open {
            return Err(
                "--serve-open disables authentication; it cannot be combined with --authz-db"
                    .to_owned(),
            );
        }
        let mut config = AuthzConfig {
            limits: corium_authz::Limits {
                max_depth: self.authz_max_depth,
                ..corium_authz::Limits::default()
            },
            ..AuthzConfig::default()
        };
        if self.authz_fresh_writes {
            config.fresh_for = [ActionClass::Write, ActionClass::Admin]
                .into_iter()
                .collect();
        }
        if !self.authz_break_glass_roles.is_empty() {
            config.break_glass = Some(BreakGlass {
                roles: self.authz_break_glass_roles.iter().cloned().collect(),
                ..BreakGlass::default()
            });
        }
        Ok(Some((db, config)))
    }

    /// Builds the request [`Guard`] this server enforces from the auth flags.
    ///
    /// `authorizer` is the policy the surface wired up from `--authz-db`;
    /// `None` leaves authorization permit-all.
    async fn guard_with(&self, authorizer: Option<Arc<dyn Authorizer>>) -> Result<Guard, String> {
        if self.serve_open {
            debug_assert!(authorizer.is_none(), "ServeFlags::authz rejects the pair");
            return Ok(Guard::disabled());
        }
        // The static-token provider recognizes either an explicit `--serve-token`
        // (strict) or the shared development token (permissive unless
        // `--require-auth` / OIDC forces strict mode).
        let (secret, strict_static) = match &self.serve_token {
            Some(token) => (token.clone(), true),
            None => (DEFAULT_DEV_TOKEN.to_owned(), self.require_auth),
        };
        let statics = StaticTokens::new().with(
            secret,
            Principal::new("static-token", "operator").with_role("admin"),
        );

        let oidc = self.oidc_provider().await?;
        let strict = strict_static || oidc.is_some();
        let provider: Arc<dyn IdentityProvider> = match oidc {
            Some(oidc) => Arc::new(CompositeProvider::new(vec![Arc::new(statics), oidc])),
            None => Arc::new(statics),
        };
        // Authorization is permit-all unless `--authz-db` supplied a
        // relationship policy. Anonymous callers are still admitted in
        // permissive mode: with an authorizer in place they simply arrive as
        // `user:anonymous`, so "public read, authenticated write" stays a
        // policy question rather than a flag.
        let authorizer = authorizer.unwrap_or_else(|| Arc::new(AllowAll));
        Ok(Guard::new(provider, authorizer).allow_anonymous(!strict))
    }

    /// Builds the OIDC identity provider when `--oidc-issuer` is set.
    #[cfg(feature = "oidc")]
    async fn oidc_provider(&self) -> Result<Option<Arc<dyn IdentityProvider>>, String> {
        use corium_protocol::authz::ExternalTokens;
        use corium_protocol::oidc::{OidcConfig, OidcVerifier};

        let Some(issuer) = &self.oidc_issuer else {
            return Ok(None);
        };
        let config = OidcConfig::new(issuer.clone(), self.oidc_audiences.clone());
        let verifier = match &self.oidc_jwks_file {
            Some(path) => {
                let json = std::fs::read_to_string(path)
                    .map_err(|error| format!("cannot read JWKS {}: {error}", path.display()))?;
                OidcVerifier::from_jwks_json(config, &json)
                    .map_err(|error| format!("cannot load JWKS: {error}"))?
            }
            #[cfg(feature = "oidc-discovery")]
            None => OidcVerifier::from_discovery(config)
                .await
                .map_err(|error| format!("OIDC discovery failed: {error}"))?,
            #[cfg(not(feature = "oidc-discovery"))]
            None => {
                return Err("OIDC discovery needs the `oidc-discovery` feature; \
                     pass --oidc-jwks-file instead, or rebuild with it enabled"
                    .to_owned());
            }
        };
        Ok(Some(Arc::new(ExternalTokens::new("oidc", verifier))))
    }

    #[cfg(not(feature = "oidc"))]
    #[allow(clippy::unused_async)]
    async fn oidc_provider(&self) -> Result<Option<Arc<dyn IdentityProvider>>, String> {
        if self.oidc_issuer.is_some() || self.oidc_jwks_file.is_some() {
            return Err(
                "OIDC support is not built in; rebuild corium-cli with --features oidc-discovery"
                    .to_owned(),
            );
        }
        Ok(None)
    }
}

#[derive(Subcommand)]
enum Command {
    /// Run a transactor process over a data directory.
    Transactor {
        /// Storage-service backend for blobs and roots.
        #[arg(long, value_enum, default_value_t = StoreKind::Fs)]
        store: StoreKind,
        /// Turso database path for `--store turso` (defaults to
        /// `{data_dir}/store.db`).
        #[arg(long)]
        turso_path: Option<PathBuf>,
        /// `PostgreSQL` connection string for `--store postgres`.
        #[arg(long)]
        postgres_url: Option<String>,
        /// S3 bucket for `--store s3`.
        #[arg(long)]
        s3_bucket: Option<String>,
        /// S3 key prefix for `--store s3` (defaults to the bucket root).
        #[arg(long, default_value = "")]
        s3_prefix: String,
        /// Data directory (filesystem store, logs). Ignored by `--store mem`.
        #[arg(long)]
        data_dir: PathBuf,
        /// Listen address.
        #[arg(long, default_value = "127.0.0.1:4334")]
        listen: SocketAddr,
        /// Stable owner identity for lease records.
        #[arg(long)]
        owner: Option<String>,
        /// Lease time-to-live in milliseconds.
        #[arg(long, default_value_t = 5_000)]
        lease_ttl_ms: i64,
        /// How long to wait for a held lease before giving up (ms).
        #[arg(long, default_value_t = 15_000)]
        lease_wait_ms: i64,
        /// High-availability mode: stand by (and take over on lease expiry)
        /// when another transactor holds a database's lease, instead of
        /// failing startup; on depose, return to standby instead of exiting.
        #[arg(long)]
        ha: bool,
        /// Client endpoint advertised to peers for lease-holder discovery,
        /// e.g. `http://transactor-a:4334`.
        #[arg(long)]
        advertise: Option<String>,
        /// Interval between background index publications (ms).
        #[arg(long, default_value_t = 5_000)]
        index_interval_ms: u64,
        /// Minimum wait before the next index publication, as a multiple of
        /// the previous publication's duration (0 disables the backoff).
        #[arg(long, default_value_t = 4)]
        index_backoff: u32,
        /// Defer index publication while fewer than this many new datoms are
        /// pending (0 publishes any pending work).
        #[arg(long, default_value_t = 0)]
        index_tail_threshold: u64,
        /// Longest a below-threshold tail defers index publication (ms).
        #[arg(long, default_value_t = 60_000)]
        index_tail_deadline_ms: u64,
        /// Interval between subscription heartbeats (ms).
        #[arg(long, default_value_t = 10_000)]
        heartbeat_ms: u64,
        /// Prometheus HTTP listen address (`/metrics`); disabled when omitted.
        #[arg(long)]
        metrics_listen: Option<SocketAddr>,
        /// Scheduled GC interval (for example `1h`); `off` disables it.
        #[arg(long, default_value = "1h")]
        gc_interval: String,
        /// Retain unreachable blobs for at least this long.
        #[arg(long, default_value = "72h")]
        gc_window: String,
        /// Fuel budget per database-function invocation (execution credits).
        #[arg(long, default_value_t = 1_000_000)]
        db_fn_fuel: u64,
        /// Managed-memory budget per database-function invocation (bytes).
        #[arg(long, default_value_t = 16 * 1024 * 1024)]
        db_fn_memory_bytes: usize,
        #[command(flatten)]
        serve: ServeFlags,
    },
    /// Run a peer server hosting one database for thin clients.
    PeerServer {
        /// Database to host.
        #[arg(long)]
        db: String,
        /// Listen address.
        #[arg(long, default_value = "127.0.0.1:4336")]
        listen: SocketAddr,
        /// Fuel ceiling per query (datoms touched).
        #[arg(long, default_value_t = 10_000_000)]
        max_fuel: u64,
        /// Prometheus HTTP listen address (`/metrics`); disabled when omitted.
        #[arg(long)]
        metrics_listen: Option<SocketAddr>,
        #[command(flatten)]
        client: ClientFlags,
        #[command(flatten)]
        serve: ServeFlags,
    },
    /// Serve the database catalog over the `PostgreSQL` wire protocol
    /// (read-only). Clients pick a database with the startup `database`
    /// parameter or `USE <db>`, and list them with `SHOW DATABASES`.
    PostgresServer {
        /// Restrict the databases clients may reach (repeatable). When
        /// omitted, every database in the transactor's catalog is exposed.
        #[arg(long = "database")]
        databases: Vec<String>,
        /// Listen address.
        #[arg(long, default_value = "127.0.0.1:5432")]
        listen: SocketAddr,
        /// Require this cleartext password from clients (trust when omitted).
        #[arg(long)]
        password: Option<String>,
        #[command(flatten)]
        client: ClientFlags,
    },
    /// Database catalog operations.
    #[command(subcommand)]
    Db(DbCommand),
    /// Self-hosted authorization database: create it, grant and revoke
    /// relationships, and ask what the policy decides.
    #[command(subcommand)]
    Authz(authz::AuthzCommand),
    /// Sweep blobs unreachable from any live database root.
    Gc {
        /// Operate offline over a data directory (transactor must be stopped).
        #[arg(long, conflicts_with = "transactor")]
        data_dir: Option<PathBuf>,
        /// Or ask a running transactor to collect.
        #[arg(long)]
        transactor: Option<String>,
        /// Bearer token for the transactor (defaults to the shared development
        /// token; `--token ""` connects anonymously).
        #[arg(long, env = "CORIUM_TOKEN")]
        token: Option<String>,
        /// PEM file with a CA certificate to trust (enables TLS).
        #[arg(long)]
        ca: Option<PathBuf>,
        /// Domain name expected on the server certificate.
        #[arg(long)]
        tls_domain: Option<String>,
        /// Retain unreachable blobs newer than this window (offline and online).
        #[arg(long, default_value = "72h")]
        window: String,
    },
    /// Create or incrementally refresh a database backup from a live transactor.
    Backup {
        /// Running transactor used to discover the source storage service.
        #[arg(long, default_value = "http://127.0.0.1:4334")]
        transactor: String,
        /// Bearer token for the transactor (defaults to the shared development
        /// token; `--token ""` connects anonymously).
        #[arg(long, env = "CORIUM_TOKEN")]
        token: Option<String>,
        /// PEM file with a CA certificate to trust (enables TLS).
        #[arg(long)]
        ca: Option<PathBuf>,
        /// Domain name expected on the server certificate.
        #[arg(long)]
        tls_domain: Option<String>,
        /// Database name.
        db: String,
        /// Binary backup file; reusing it appends an incremental checkpoint.
        destination: PathBuf,
    },
    /// Restore a backup, optionally under a new database name (clone).
    Restore {
        /// Binary backup file.
        source: PathBuf,
        /// Target transactor data directory (transactor must be stopped).
        #[arg(long)]
        data_dir: PathBuf,
        /// Target database name; may differ from the backed-up source name.
        #[arg(long)]
        as_db: String,
    },
    /// Open an interactive peer-local Datalog console.
    Console {
        /// Database name.
        db: String,
        #[command(flatten)]
        client: ClientFlags,
    },
    /// Open a full-screen terminal dashboard: query workbench, store
    /// metrics, live transaction feed, and schema browser.
    Tui {
        /// Database name.
        db: String,
        /// Metrics refresh interval in milliseconds (minimum 250).
        #[arg(long, default_value_t = 2_000)]
        refresh_ms: u64,
        #[command(flatten)]
        client: ClientFlags,
    },
    /// Open a read-only interactive SQL shell.
    Sql {
        /// Database name.
        db: String,
        /// Execute SQL and exit.
        #[arg(short = 'c', long, conflicts_with = "file")]
        command: Option<String>,
        /// Execute SQL from a file and exit.
        #[arg(short = 'f', long, conflicts_with = "command")]
        file: Option<PathBuf>,
        #[command(flatten)]
        client: ClientFlags,
    },
    /// Print committed transactions from a data directory's log.
    Log {
        /// Data directory (store, logs).
        #[arg(long)]
        data_dir: PathBuf,
        /// Database name.
        #[arg(long)]
        db: String,
        /// First transaction to print (inclusive).
        #[arg(long, default_value_t = 0)]
        from: u64,
        /// Last transaction to print (exclusive; 0 = open-ended).
        #[arg(long, default_value_t = 0)]
        to: u64,
    },
}

#[derive(Subcommand)]
enum DbCommand {
    /// Create a database (optionally with an EDN schema file).
    Create {
        /// Database name.
        name: String,
        /// EDN file containing a vector of attribute maps.
        #[arg(long)]
        schema: Option<PathBuf>,
        #[command(flatten)]
        client: ClientFlags,
    },
    /// Delete a database.
    Delete {
        /// Database name.
        name: String,
        #[command(flatten)]
        client: ClientFlags,
    },
    /// Fork a database: create a new database duplicating an existing one
    /// at a transaction basis (e.g. as a sandbox wound back to a point).
    Fork {
        /// Source database name.
        name: String,
        /// Name for the new fork.
        target: String,
        /// Transaction basis to fork at (defaults to the current basis).
        #[arg(long)]
        as_of: Option<u64>,
        #[command(flatten)]
        client: ClientFlags,
    },
    /// List databases.
    List {
        #[command(flatten)]
        client: ClientFlags,
    },
    /// Connect a peer and print database statistics.
    Stats {
        /// Database name.
        name: String,
        #[command(flatten)]
        client: ClientFlags,
    },
    /// Ask the transactor to publish the database's indexes now.
    RequestIndex {
        /// Database name.
        name: String,
        #[command(flatten)]
        client: ClientFlags,
    },
    /// Read or override the database's index-publication pacing at runtime.
    ///
    /// Omitted flags are left unchanged; with no flags the current policy
    /// is printed. Overrides last until the transactor restarts.
    IndexPolicy {
        /// Database name.
        name: String,
        /// Base interval between index publications (ms).
        #[arg(long)]
        interval_ms: Option<u64>,
        /// Minimum wait before the next publication, as a multiple of the
        /// previous publication's duration (0 disables the backoff).
        #[arg(long)]
        backoff: Option<u32>,
        /// Defer publication while fewer than this many new datoms are
        /// pending (0 publishes any pending work).
        #[arg(long)]
        tail_threshold: Option<u64>,
        /// Longest a below-threshold tail defers publication (ms).
        #[arg(long)]
        tail_deadline_ms: Option<u64>,
        #[command(flatten)]
        client: ClientFlags,
    },
}

#[tokio::main]
async fn main() -> ExitCode {
    // The binary links two rustls crypto backends (`ring` via tonic, and
    // `aws-lc-rs` transitively through the cljrs runtime), so rustls cannot
    // auto-select a process-level provider; pin `ring` explicitly before any
    // TLS setup.
    let _ = rustls::crypto::ring::default_provider().install_default();
    let cli = Cli::parse();
    // The TUI owns the terminal; stray tracing output would corrupt it.
    if !matches!(cli.command, Command::Tui { .. }) {
        init_logging(cli.log_format);
    }
    match run(cli).await {
        Ok(()) => ExitCode::SUCCESS,
        Err(message) => {
            eprintln!("corium: {message}");
            ExitCode::FAILURE
        }
    }
}

#[allow(clippy::too_many_lines)]
async fn run(cli: Cli) -> Result<(), String> {
    match cli.command {
        Command::Transactor {
            store,
            turso_path,
            postgres_url,
            s3_bucket,
            s3_prefix,
            data_dir,
            listen,
            owner,
            lease_ttl_ms,
            lease_wait_ms,
            ha,
            advertise,
            index_interval_ms,
            index_backoff,
            index_tail_threshold,
            index_tail_deadline_ms,
            heartbeat_ms,
            metrics_listen,
            gc_interval,
            gc_window,
            db_fn_fuel,
            db_fn_memory_bytes,
            serve,
        } => {
            let store_spec = store_spec(
                store,
                &data_dir,
                turso_path,
                postgres_url,
                s3_bucket,
                s3_prefix,
            )?;
            let mut config = NodeConfig::new(data_dir);
            config.store = store_spec;
            if let Some(owner) = owner {
                config.owner = owner;
            }
            config.lease_ttl_ms = lease_ttl_ms;
            config.lease_wait_ms = lease_wait_ms;
            config.ha = ha;
            config.advertise = advertise;
            config.index_interval = Duration::from_millis(index_interval_ms);
            config.index_backoff = index_backoff;
            config.index_tail_threshold = index_tail_threshold;
            config.index_tail_deadline = Duration::from_millis(index_tail_deadline_ms);
            config.heartbeat_interval = Duration::from_millis(heartbeat_ms);
            config.gc_interval = if gc_interval == "off" {
                None
            } else {
                Some(parse_duration(&gc_interval)?)
            };
            config.gc_retention = parse_duration(&gc_window)?;
            // The built-in `cljrs-tx` runtime is wired by `NodeConfig::new`
            // when the `cljrs` feature is on; apply the flag budgets here.
            #[cfg(feature = "cljrs")]
            {
                config.tx_fn_expander = Some(Arc::new(corium_transactor::txfn::DbFnExpander::new(
                    corium_transactor::txfn::DbFnBudget {
                        fuel: db_fn_fuel,
                        memory_bytes: db_fn_memory_bytes,
                        ..corium_transactor::txfn::DbFnBudget::default()
                    },
                )));
            }
            #[cfg(not(feature = "cljrs"))]
            let _ = (db_fn_fuel, db_fn_memory_bytes);
            let tls = serve.tls()?;
            let node = TransactorNode::open(config)
                .await
                .map_err(|error| format!("cannot open node: {error}"))?;
            // The node leads the authz database like any other, so the
            // authorizer reads a local snapshot and the node's basis watch is
            // its change signal — no extra connection, and write admission is
            // decided at the serialization point.
            let authorizer = match serve.authz()? {
                Some((authz_db, authz_config)) => Some(
                    start_authorizer(
                        Arc::new(corium_transactor::authz::NodePolicySource::new(
                            Arc::clone(&node),
                            authz_db.clone(),
                        )),
                        authz_config,
                        &authz_db,
                    )
                    .await,
                ),
                None => None,
            };
            let guard = serve.guard_with(authorizer).await?;
            let _metrics = if let Some(address) = metrics_listen {
                let metrics_node = Arc::clone(&node);
                Some(
                    metrics_http::spawn(
                        address,
                        Arc::new(move || metrics_node.metrics().prometheus()),
                    )
                    .await?,
                )
            } else {
                None
            };
            let mut shutdown = node.shutdown_watch();
            tracing::info!(
                %listen,
                databases = ?node.list_dbs(),
                standby = ?node.standby_dbs(),
                "transactor serving"
            );
            eprintln!(
                "corium transactor: serving {:?} (standby for {:?}) on {listen}",
                node.list_dbs(),
                node.standby_dbs()
            );
            let server = corium_transactor::server::serve(
                Arc::clone(&node),
                listen,
                guard,
                tls,
                async move {
                    tokio::select! {
                        _ = tokio::signal::ctrl_c() => {}
                        _ = shutdown.changed() => {}
                    }
                },
            );
            server.await.map_err(|error| error.to_string())?;
            // Graceful stop: expire held leases so a standby takes over
            // immediately instead of waiting out the TTL.
            node.release_leases().await;
            if let Some(reason) = node.shutdown_watch().borrow().clone() {
                return Err(format!("shut down: {reason}"));
            }
            Ok(())
        }
        Command::PeerServer {
            db,
            listen,
            max_fuel,
            metrics_listen,
            client,
            serve,
        } => {
            let tls = serve.tls()?;
            let config = client.connect_config(db).await?;
            let connection = Arc::new(
                Connection::connect(config)
                    .await
                    .map_err(|error| format!("cannot connect to transactor: {error}"))?,
            );
            // A second peer connection keeps the policy database in sync
            // locally, so checks stay in-process: no per-request round trip to
            // the transactor.
            let authorizer = match serve.authz()? {
                Some((authz_db, authz_config)) => {
                    let authz_config_client = client.connect_config(authz_db.clone()).await?;
                    let authz_connection =
                        Arc::new(Connection::connect(authz_config_client).await.map_err(
                            |error| {
                                format!(
                                    "cannot connect to authorization database {authz_db:?}: {error}"
                                )
                            },
                        )?);
                    Some(
                        start_authorizer(
                            Arc::new(corium_peer::authz::ConnectionPolicySource::new(
                                authz_connection,
                            )),
                            authz_config,
                            &authz_db,
                        )
                        .await,
                    )
                }
                None => None,
            };
            let guard = serve.guard_with(authorizer).await?;
            eprintln!(
                "corium peer-server: hosting {:?} on {listen}",
                connection.db_name()
            );
            let service = corium_peer::server::PeerServerSvc::new(
                connection,
                PeerServerConfig {
                    max_fuel,
                    ..PeerServerConfig::default()
                },
            )
            .with_guard(guard);
            let metrics = service.metrics();
            let _metrics = if let Some(address) = metrics_listen {
                Some(metrics_http::spawn(address, Arc::new(move || metrics.prometheus())).await?)
            } else {
                None
            };
            tracing::info!(%listen, "peer server serving");
            corium_peer::server::serve_service(service, listen, tls, async {
                let _ = tokio::signal::ctrl_c().await;
            })
            .await
            .map_err(|error| error.to_string())
        }
        Command::PostgresServer {
            databases,
            listen,
            password,
            client,
        } => {
            let listener = tokio::net::TcpListener::bind(listen)
                .await
                .map_err(|error| format!("cannot bind {listen}: {error}"))?;
            let catalog = Arc::new(pg_catalog::PeerCatalog::new(client, databases));
            let pg_config = corium_pgwire::PgWireConfig {
                password,
                ..corium_pgwire::PgWireConfig::default()
            };
            tracing::info!(%listen, "postgres server serving");
            eprintln!("corium postgres-server: serving the database catalog on {listen}");
            corium_pgwire::serve(listener, catalog, pg_config, async {
                let _ = tokio::signal::ctrl_c().await;
            })
            .await
            .map_err(|error| error.to_string())
        }
        Command::Db(command) => run_db(command).await,
        Command::Authz(command) => authz::run(command).await,
        Command::Gc {
            data_dir,
            transactor,
            token,
            ca,
            tls_domain,
            window,
        } => match (data_dir, transactor) {
            (Some(data_dir), None) => {
                let store = FsStore::open(data_dir.join("store"))
                    .map_err(|error| format!("cannot open store: {error}"))?;
                let mut live = Vec::new();
                for root_name in store
                    .list_roots("db:")
                    .await
                    .map_err(|error| error.to_string())?
                {
                    if let Some(root) = store
                        .get_root(&root_name)
                        .await
                        .map_err(|error| error.to_string())?
                        .as_deref()
                        .and_then(DbRoot::decode)
                    {
                        live.extend(root.roots.into_iter().flatten());
                    }
                }
                let report = corium_store::mark_and_sweep_retained(
                    &store,
                    live,
                    |_, bytes| corium_store::index_blob_children(bytes),
                    parse_duration(&window)?,
                    std::time::SystemTime::now(),
                )
                .await
                .map_err(|error| error.to_string())?;
                println!(
                    "{{:marked {} :swept {} :retained {}}}",
                    report.marked, report.swept, report.retained
                );
                Ok(())
            }
            (None, Some(endpoint)) => {
                let flags = ClientFlags {
                    transactor: endpoint,
                    token,
                    ca,
                    tls_domain,
                    peer_bootstrap: false,
                };
                let mut admin = Admin::connect(&flags.primary(), flags.token(), flags.tls()?)
                    .await
                    .map_err(|error| error.to_string())?;
                let swept = admin
                    .gc_deleted_databases_with_retention(Some(parse_duration(&window)?))
                    .await
                    .map_err(|error| error.to_string())?;
                println!("{{:swept {swept}}}");
                Ok(())
            }
            _ => Err("pass exactly one of --data-dir (offline) or --transactor".into()),
        },
        Command::Backup {
            transactor,
            token,
            ca,
            tls_domain,
            db,
            destination,
        } => {
            let tls = if ca.is_none() && tls_domain.is_none() {
                None
            } else {
                Some(
                    client_tls(ca.as_deref(), tls_domain.as_deref())
                        .map_err(|error| format!("cannot load CA certificate: {error}"))?,
                )
            };
            let mut admin =
                Admin::connect(&transactor, resolve_client_token(token.as_deref()), tls)
                    .await
                    .map_err(|error| error.to_string())?;
            let info = admin
                .get_storage_info(&db)
                .await
                .map_err(|error| error.to_string())?;
            let source = corium_transactor::backup::BackupSource::from_info(info)
                .map_err(|error| error.to_string())?;
            let report = corium_transactor::backup::backup(&source, &db, destination)
                .await
                .map_err(|error| error.to_string())?;
            println!(
                "{{:db {db:?} :backup-format {} :writer-version {:?} :basis-t {} :index-basis-t {} :replayed-transactions {} :copied-blobs {} :reused-blobs {}}}",
                report.backup_format_version,
                report.writer_version,
                report.basis_t,
                report.index_basis_t,
                report.replayed_transactions,
                report.copied_blobs,
                report.reused_blobs
            );
            Ok(())
        }
        Command::Restore {
            source,
            data_dir,
            as_db,
        } => {
            let report = corium_transactor::backup::restore(source, data_dir, &as_db)
                .await
                .map_err(|error| error.to_string())?;
            println!(
                "{{:source-db {:?} :db {:?} :backup-format {} :writer-version {:?} :basis-t {} :copied-blobs {} :reused-blobs {}}}",
                report.source_db,
                report.target_db,
                report.backup_format_version,
                report.writer_version,
                report.basis_t,
                report.copied_blobs,
                report.reused_blobs
            );
            Ok(())
        }
        Command::Console { db, client } => {
            let config = client.connect_config(db).await?;
            let connection = Connection::connect(config)
                .await
                .map_err(|error| format!("cannot connect to transactor: {error}"))?;
            console::run(&connection).await
        }
        Command::Tui {
            db,
            refresh_ms,
            client,
        } => {
            let config = client.connect_config(db).await?;
            let connection = Connection::connect(config)
                .await
                .map_err(|error| format!("cannot connect to transactor: {error}"))?;
            tui::run(
                Arc::new(connection),
                Duration::from_millis(refresh_ms.max(250)),
            )
            .await
        }
        Command::Sql {
            db,
            command,
            file,
            client,
        } => {
            let config = client.connect_config(db).await?;
            let connection = Connection::connect(config)
                .await
                .map_err(|error| format!("cannot connect to transactor: {error}"))?;
            sql::run(&connection, command.as_deref(), file.as_deref()).await
        }
        Command::Log {
            data_dir,
            db,
            from,
            to,
        } => run_log(&data_dir, &db, from, to).await,
    }
}

fn init_logging(format: LogFormat) {
    use tracing_subscriber::EnvFilter;
    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
    match format {
        LogFormat::Human => {
            let _ = tracing_subscriber::fmt().with_env_filter(filter).try_init();
        }
        LogFormat::Json => {
            let _ = tracing_subscriber::fmt()
                .json()
                .with_env_filter(filter)
                .try_init();
        }
    }
}

/// Resolves the `--store` flag and backend connection options into a [`StoreSpec`].
fn store_spec(
    store: StoreKind,
    data_dir: &std::path::Path,
    turso_path: Option<PathBuf>,
    postgres_url: Option<String>,
    s3_bucket: Option<String>,
    s3_prefix: String,
) -> Result<StoreSpec, String> {
    match store {
        StoreKind::Mem => Ok(StoreSpec::Memory),
        StoreKind::Fs => Ok(StoreSpec::Fs),
        StoreKind::Postgres => postgres_spec(postgres_url),
        StoreKind::Turso => turso_spec(data_dir, turso_path),
        StoreKind::S3 => s3_spec(s3_bucket, s3_prefix),
    }
}

#[cfg(feature = "postgres")]
fn postgres_spec(postgres_url: Option<String>) -> Result<StoreSpec, String> {
    let connection_string = postgres_url
        .ok_or_else(|| "--postgres-url is required with --store postgres".to_owned())?;
    Ok(StoreSpec::Postgres { connection_string })
}

#[cfg(not(feature = "postgres"))]
fn postgres_spec(_postgres_url: Option<String>) -> Result<StoreSpec, String> {
    Err(
        "this build lacks the PostgreSQL backend; rebuild corium-cli with --features postgres"
            .into(),
    )
}

#[cfg(feature = "turso")]
fn turso_spec(
    data_dir: &std::path::Path,
    turso_path: Option<PathBuf>,
) -> Result<StoreSpec, String> {
    let path = turso_path.unwrap_or_else(|| data_dir.join("store.db"));
    let path = path
        .to_str()
        .ok_or_else(|| format!("turso path is not valid UTF-8: {}", path.display()))?
        .to_owned();
    Ok(StoreSpec::Turso { path })
}

#[cfg(not(feature = "turso"))]
fn turso_spec(
    _data_dir: &std::path::Path,
    _turso_path: Option<PathBuf>,
) -> Result<StoreSpec, String> {
    Err("this build lacks the Turso backend; rebuild corium-cli with --features turso".into())
}

#[cfg(feature = "s3")]
fn s3_spec(s3_bucket: Option<String>, s3_prefix: String) -> Result<StoreSpec, String> {
    let bucket = s3_bucket.ok_or_else(|| "--s3-bucket is required with --store s3".to_owned())?;
    Ok(StoreSpec::S3 {
        bucket,
        prefix: s3_prefix,
    })
}

#[cfg(not(feature = "s3"))]
fn s3_spec(_s3_bucket: Option<String>, _s3_prefix: String) -> Result<StoreSpec, String> {
    Err("this build lacks the S3 backend; rebuild corium-cli with --features s3".into())
}

fn parse_duration(text: &str) -> Result<Duration, String> {
    let split = text
        .find(|character: char| !character.is_ascii_digit())
        .unwrap_or(text.len());
    let amount: u64 = text[..split]
        .parse()
        .map_err(|_| format!("invalid duration {text:?}"))?;
    let unit = &text[split..];
    let seconds = match unit {
        "ms" => return Ok(Duration::from_millis(amount)),
        "s" | "" => amount,
        "m" => amount.saturating_mul(60),
        "h" => amount.saturating_mul(60 * 60),
        "d" => amount.saturating_mul(24 * 60 * 60),
        _ => {
            return Err(format!(
                "invalid duration unit in {text:?}; use ms, s, m, h, or d"
            ));
        }
    };
    Ok(Duration::from_secs(seconds))
}

/// Builds the self-hosted authorizer over `source` and starts its refresh task.
///
/// Startup does not fail when the policy cannot be compiled yet: the authorizer
/// denies every request until it can (that is the fail-closed contract), and
/// the refresh task keeps trying, so a transactor started before
/// `corium authz init` recovers on its own instead of crash-looping.
async fn start_authorizer(
    source: Arc<dyn corium_authz::PolicySource>,
    config: AuthzConfig,
    authz_db: &str,
) -> Arc<dyn Authorizer> {
    let authorizer = Arc::new(SystemDbAuthorizer::with_config(source, config));
    match authorizer.refresh().await {
        Ok(policy) => {
            tracing::info!(
                authz_db,
                authz_t = policy.basis_t(),
                stats = ?policy.stats(),
                "authorization policy loaded"
            );
            eprintln!(
                "corium: authorizing from {authz_db:?} at authz-t {}",
                policy.basis_t()
            );
        }
        Err(error) => {
            tracing::error!(
                authz_db,
                %error,
                "authorization policy is unavailable; denying every request until it loads"
            );
            eprintln!(
                "corium: cannot load the authorization database {authz_db:?} ({error}); \
                 every request will be denied until it loads — run `corium authz init --db {authz_db}`"
            );
        }
    }
    authorizer.spawn_refresh();
    authorizer
}

async fn admin_client(client: &ClientFlags) -> Result<Admin, String> {
    Admin::connect(&client.primary(), client.token(), client.tls()?)
        .await
        .map_err(|error| error.to_string())
}

/// Resolves a client bearer token: the given value when set, the shared
/// development token ([`DEFAULT_DEV_TOKEN`]) when unset, or `None` when the
/// value is present but empty (`--token ""`), meaning connect anonymously.
fn resolve_client_token(token: Option<&str>) -> Option<String> {
    match token {
        Some("") => None,
        Some(value) => Some(value.to_owned()),
        None => Some(DEFAULT_DEV_TOKEN.to_owned()),
    }
}

async fn run_db(command: DbCommand) -> Result<(), String> {
    match command {
        DbCommand::Create {
            name,
            schema,
            client,
        } => {
            let forms = match schema {
                Some(path) => {
                    let text = std::fs::read_to_string(&path)
                        .map_err(|error| format!("cannot read {}: {error}", path.display()))?;
                    let mut forms =
                        read_all(&text).map_err(|error| format!("bad schema EDN: {error}"))?;
                    // Accept either one vector of maps or bare maps.
                    if forms.len() == 1 && matches!(forms[0], Edn::Vector(_)) {
                        let Edn::Vector(items) = forms.remove(0) else {
                            unreachable!()
                        };
                        items
                    } else {
                        forms
                    }
                }
                None => Vec::new(),
            };
            let mut admin = admin_client(&client).await?;
            let created = admin
                .create_database(&name, &forms)
                .await
                .map_err(|error| error.to_string())?;
            println!("{{:db {name:?} :created {created}}}");
            Ok(())
        }
        DbCommand::Delete { name, client } => {
            let mut admin = admin_client(&client).await?;
            let deleted = admin
                .delete_database(&name)
                .await
                .map_err(|error| error.to_string())?;
            println!("{{:db {name:?} :deleted {deleted}}}");
            Ok(())
        }
        DbCommand::Fork {
            name,
            target,
            as_of,
            client,
        } => {
            let mut admin = admin_client(&client).await?;
            let forked = admin
                .fork_database(&name, &target, as_of)
                .await
                .map_err(|error| error.to_string())?;
            match forked {
                Some(basis_t) => println!(
                    "{{:db {target:?} :forked-from {name:?} :basis-t {basis_t} :created true}}"
                ),
                None => println!("{{:db {target:?} :created false}}"),
            }
            Ok(())
        }
        DbCommand::List { client } => {
            let mut admin = admin_client(&client).await?;
            for db in admin
                .list_databases()
                .await
                .map_err(|error| error.to_string())?
            {
                println!("{db}");
            }
            Ok(())
        }
        DbCommand::Stats { name, client } => run_db_stats(name, &client).await,
        DbCommand::RequestIndex { name, client } => {
            let mut admin = admin_client(&client).await?;
            let index_basis_t = admin
                .request_index(&name)
                .await
                .map_err(|error| error.to_string())?;
            println!("{{:db {name:?} :index-basis-t {index_basis_t}}}");
            Ok(())
        }
        DbCommand::IndexPolicy {
            name,
            interval_ms,
            backoff,
            tail_threshold,
            tail_deadline_ms,
            client,
        } => {
            let update = IndexPolicySettings {
                interval_ms,
                backoff,
                tail_threshold,
                tail_deadline_ms,
            };
            run_db_index_policy(&name, update, &client).await
        }
    }
}

async fn run_db_stats(name: String, client: &ClientFlags) -> Result<(), String> {
    let config = client.connect_config(name).await?;
    let connection = Connection::connect(config)
        .await
        .map_err(|error| error.to_string())?;
    let db = connection.sync().await.map_err(|error| error.to_string())?;
    let stats = db.stats();
    let status_response = connection
        .status()
        .await
        .map_err(|error| error.to_string())?;
    println!(
        "{{:basis-t {} :index-basis-t {} :datoms {} :entities {} :attributes {} :index-lag {} :tx-count {} :tx-failures {} :tx-queue-depth {} :gc-runs {} :gc-swept-blobs {}}}",
        db.basis_t(),
        connection.index_basis_t(),
        stats.datoms,
        stats.entities,
        stats.attributes,
        status_response.index_lag,
        status_response.transaction_count,
        status_response.transaction_failure_count,
        status_response.transaction_queue_depth,
        status_response.gc_runs,
        status_response.gc_swept_blobs,
    );
    Ok(())
}

async fn run_db_index_policy(
    name: &str,
    update: IndexPolicySettings,
    client: &ClientFlags,
) -> Result<(), String> {
    let mut admin = admin_client(client).await?;
    let policy = admin
        .set_index_policy(name, update)
        .await
        .map_err(|error| error.to_string())?;
    println!(
        "{{:db {name:?} :interval-ms {} :backoff {} :tail-threshold {} :tail-deadline-ms {}}}",
        policy.interval_ms.unwrap_or_default(),
        policy.backoff.unwrap_or_default(),
        policy.tail_threshold.unwrap_or_default(),
        policy.tail_deadline_ms.unwrap_or_default(),
    );
    Ok(())
}

async fn run_log(data_dir: &std::path::Path, db: &str, from: u64, to: u64) -> Result<(), String> {
    use corium_log::TransactionLog;
    let log = corium_log::VersionedLog::open_read_only(data_dir.join("logs"), db)
        .map_err(|error| format!("cannot open log: {error}"))?;
    // Naming from the meta root makes keyword values readable.
    let interner = match FsStore::open(data_dir.join("store")) {
        Ok(store) => store
            .get_root(&format!("meta:{db}"))
            .await
            .ok()
            .flatten()
            .and_then(|meta| decode_meta_interner(&meta))
            .unwrap_or_default(),
        Err(_) => KeywordInterner::default(),
    };
    let end = if to == 0 { None } else { Some(to) };
    for record in log
        .tx_range(from, end)
        .map_err(|error| format!("cannot read log: {error}"))?
    {
        println!(
            "{{:t {} :tx-instant {} :datoms [",
            record.t, record.tx_instant
        );
        for datom in &record.datoms {
            let value = format_value(&datom.v, &interner);
            println!(
                "  [{} {} {value} {} {}]",
                datom.e.raw(),
                datom.a.raw(),
                datom.tx.sequence(),
                datom.added
            );
        }
        println!("]}}");
    }
    Ok(())
}

fn decode_meta_interner(meta: &[u8]) -> Option<KeywordInterner> {
    let schema_len = usize::try_from(u32::from_be_bytes(meta.get(..4)?.try_into().ok()?)).ok()?;
    let rest = meta.get(4 + schema_len..)?;
    let naming_len = usize::try_from(u32::from_be_bytes(rest.get(..4)?.try_into().ok()?)).ok()?;
    codec::decode_naming(rest.get(4..4 + naming_len)?).ok()
}

fn format_value(value: &corium_core::Value, interner: &KeywordInterner) -> String {
    use corium_core::Value;
    match value {
        Value::Bool(v) => v.to_string(),
        Value::Long(v) => v.to_string(),
        Value::Double(v) => format!("{}", v.0),
        Value::Instant(ms) => format!("#inst {ms}"),
        Value::Uuid(v) => format!("#uuid \"{v:032x}\""),
        Value::Keyword(id) => interner
            .resolve(*id)
            .map_or_else(|| format!("#kw {id}"), ToString::to_string),
        Value::Str(v) => format!("{v:?}"),
        Value::Bytes(bytes) => format!("#bytes[{}]", bytes.len()),
        Value::Ref(e) => format!("#eid {}", e.raw()),
    }
}