corium-cli 0.1.46

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

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

use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::sync::Arc;
use std::time::Duration;

#[cfg(feature = "s3")]
use aws_credential_types::{
    Credentials,
    provider::{
        ProvideCredentials, SharedCredentialsProvider, error::CredentialsError,
        future::ProvideCredentials as ProvideCredentialsFuture,
    },
};
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, SegmentCacheConfig};
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};
#[cfg(feature = "s3")]
use corium_store::S3ClientConfig;
use corium_store::{DbRoot, FsStore, RootStore};
use corium_transactor::node::{NodeConfig, TransactorNode};
#[cfg(feature = "s3")]
use corium_transactor::{S3ReadOnlyConfig, S3ReadOnlyCredentials};
use corium_transactor::{StorageInfoConfig, StoreSpec};

/// 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,
}

fn parse_byte_size(value: &str) -> Result<u64, String> {
    let split = value
        .find(|character: char| !character.is_ascii_digit())
        .unwrap_or(value.len());
    let (number, suffix) = value.split_at(split);
    let number = number
        .parse::<u64>()
        .map_err(|_| format!("invalid byte size {value:?}"))?;
    let multiplier = match suffix.to_ascii_lowercase().as_str() {
        "" | "b" => 1,
        "kib" => 1_u64 << 10,
        "mib" => 1_u64 << 20,
        "gib" => 1_u64 << 30,
        "tib" => 1_u64 << 40,
        "kb" => 1_000,
        "mb" => 1_000_000,
        "gb" => 1_000_000_000,
        "tb" => 1_000_000_000_000,
        _ => return Err(format!("unknown byte-size suffix in {value:?}")),
    };
    number
        .checked_mul(multiplier)
        .ok_or_else(|| format!("byte size {value:?} exceeds u64"))
}

/// 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> {
        self.connect_config_cache(db, None).await
    }

    async fn connect_config_cache(
        &self,
        db: impl Into<String>,
        cache: Option<SegmentCacheConfig>,
    ) -> 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}"))?;
        #[cfg(feature = "s3")]
        let mut spec = spec;
        #[cfg(feature = "s3")]
        if let StoreSpec::S3 { client, .. } = &mut spec {
            let initial = sdk_credentials(client)?;
            client.credentials_provider = Some(SharedCredentialsProvider::new(
                StorageInfoCredentialsProvider {
                    client: self.clone(),
                    db: db.clone(),
                    initial: std::sync::Mutex::new(Some(initial)),
                },
            ));
        }
        let store = corium_transactor::NodeStore::open_existing(&spec, &data_dir)
            .await
            .map_err(|error| format!("cannot open peer storage: {error}"))?;
        let storage = Arc::new(store);
        if let Some(cache) = cache {
            config
                .with_storage_cache(storage, &cache)
                .map_err(|error| format!("cannot open segment cache: {error}"))
        } else {
            Ok(config.with_storage(storage))
        }
    }
}

#[cfg(feature = "s3")]
struct StorageInfoCredentialsProvider {
    client: ClientFlags,
    db: String,
    initial: std::sync::Mutex<Option<Credentials>>,
}

#[cfg(feature = "s3")]
impl std::fmt::Debug for StorageInfoCredentialsProvider {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("StorageInfoCredentialsProvider")
            .field("transactor", &self.client.primary())
            .field("db", &self.db)
            .finish_non_exhaustive()
    }
}

#[cfg(feature = "s3")]
impl ProvideCredentials for StorageInfoCredentialsProvider {
    fn provide_credentials<'a>(&'a self) -> ProvideCredentialsFuture<'a>
    where
        Self: 'a,
    {
        if let Some(credentials) = self
            .initial
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .take()
        {
            return ProvideCredentialsFuture::ready(Ok(credentials));
        }
        ProvideCredentialsFuture::new(async move {
            self.fetch()
                .await
                .map_err(|error| CredentialsError::provider_error(std::io::Error::other(error)))
        })
    }
}

#[cfg(feature = "s3")]
impl StorageInfoCredentialsProvider {
    async fn fetch(&self) -> Result<Credentials, String> {
        let mut admin = Admin::connect(
            &self.client.primary(),
            self.client.token(),
            self.client.tls()?,
        )
        .await
        .map_err(|error| format!("cannot connect to transactor: {error}"))?;
        let storage = admin
            .get_storage_info(&self.db)
            .await
            .map_err(|error| format!("cannot refresh storage info: {error}"))?
            .storage
            .ok_or_else(|| "transactor returned no storage backend".to_owned())?;
        let (spec, _) = StoreSpec::from_connection(storage)
            .map_err(|error| format!("cannot use refreshed storage info: {error}"))?;
        let StoreSpec::S3 { client, .. } = spec else {
            return Err(
                "transactor storage backend changed while refreshing S3 credentials".into(),
            );
        };
        sdk_credentials(&client)
    }
}

#[cfg(feature = "s3")]
fn sdk_credentials(config: &S3ClientConfig) -> Result<Credentials, String> {
    let access_key_id = config
        .access_key_id
        .as_deref()
        .ok_or_else(|| "S3 storage info omitted the access key id".to_owned())?;
    let secret_access_key = config
        .secret_access_key
        .as_deref()
        .ok_or_else(|| "S3 storage info omitted the secret access key".to_owned())?;
    Ok(Credentials::new(
        access_key_id,
        secret_access_key,
        config.session_token.clone(),
        config.expires_after,
        "corium-storage-info-refresh",
    ))
}

/// 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)]
#[allow(clippy::large_enum_variant)]
enum Command {
    /// Run a transactor process over a data directory.
    Transactor {
        /// EDN configuration file for storage and read-only discovery
        /// credentials. Explicit CLI flags override file values.
        #[arg(long)]
        config: Option<PathBuf>,
        /// Storage-service backend for blobs and roots.
        #[arg(long, value_enum)]
        store: Option<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>,
        /// Separately provisioned read-only `PostgreSQL` connection string
        /// returned by `GetStorageInfo`.
        #[arg(long, env = "CORIUM_POSTGRES_READ_ONLY_URL")]
        postgres_read_only_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)]
        s3_prefix: Option<String>,
        /// Region advertised to storage-aware S3 clients.
        #[arg(long)]
        s3_region: Option<String>,
        /// Custom endpoint advertised to storage-aware S3-compatible clients.
        #[arg(long)]
        s3_endpoint_url: Option<String>,
        /// Static read-only S3 access-key id returned by `GetStorageInfo`.
        #[arg(
            long,
            env = "CORIUM_S3_READ_ONLY_ACCESS_KEY_ID",
            conflicts_with = "s3_read_only_role_arn"
        )]
        s3_read_only_access_key_id: Option<String>,
        /// Static read-only S3 secret access key returned by `GetStorageInfo`.
        #[arg(
            long,
            env = "CORIUM_S3_READ_ONLY_SECRET_ACCESS_KEY",
            conflicts_with = "s3_read_only_role_arn"
        )]
        s3_read_only_secret_access_key: Option<String>,
        /// Optional session token paired with static read-only S3 keys.
        #[arg(
            long,
            env = "CORIUM_S3_READ_ONLY_SESSION_TOKEN",
            conflicts_with = "s3_read_only_role_arn"
        )]
        s3_read_only_session_token: Option<String>,
        /// IAM role assumed through AWS STS for short-lived read-only S3
        /// credentials returned by each `GetStorageInfo` call.
        #[arg(long, env = "CORIUM_S3_READ_ONLY_ROLE_ARN")]
        s3_read_only_role_arn: Option<String>,
        /// STS session name for `--s3-read-only-role-arn`.
        #[arg(long)]
        s3_read_only_role_session_name: Option<String>,
        /// STS token lifetime in seconds (default 900).
        #[arg(long)]
        s3_read_only_role_duration_seconds: Option<i32>,
        /// Optional external id required by the read-only role's trust policy.
        #[arg(long, env = "CORIUM_S3_READ_ONLY_ROLE_EXTERNAL_ID")]
        s3_read_only_role_external_id: Option<String>,
        /// Data directory (filesystem store, logs). Ignored by `--store mem`.
        #[arg(long)]
        data_dir: Option<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>,
        /// Dedicated directory for the optional local SSD segment cache.
        #[arg(long, requires = "segment_cache_capacity")]
        segment_cache_dir: Option<PathBuf>,
        /// SSD segment-cache capacity (for example `256GiB`).
        #[arg(long, value_parser = parse_byte_size, requires = "segment_cache_dir")]
        segment_cache_capacity: Option<u64>,
        /// Bounded memory front tier (defaults to 64MiB when SSD cache is enabled).
        #[arg(long, value_parser = parse_byte_size, requires = "segment_cache_dir")]
        segment_cache_memory: Option<u64>,
        #[command(flatten)]
        client: ClientFlags,
        #[command(flatten)]
        serve: ServeFlags,
    },
    /// Serve the database catalog over the `PostgreSQL` wire protocol.
    /// 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>,
        /// Enable guarded autocommit INSERT, UPDATE, and DELETE. Without this
        /// flag the server remains read-only.
        #[arg(long)]
        allow_writes: bool,
        #[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 or TOML schema file).
    Create {
        /// Database name.
        name: String,
        /// Schema file: `.toml` for hierarchical TOML, otherwise EDN.
        #[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]
#[allow(clippy::large_futures)]
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 {
            config: config_file,
            store,
            turso_path,
            postgres_url,
            postgres_read_only_url,
            s3_bucket,
            s3_prefix,
            s3_region,
            s3_endpoint_url,
            s3_read_only_access_key_id,
            s3_read_only_secret_access_key,
            s3_read_only_session_token,
            s3_read_only_role_arn,
            s3_read_only_role_session_name,
            s3_read_only_role_duration_seconds,
            s3_read_only_role_external_id,
            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 file = TransactorFileConfig::load(config_file.as_deref())?;
            let store = store.or(file.store).unwrap_or_default();
            let data_dir = data_dir
                .or(file.data_dir)
                .ok_or_else(|| "--data-dir or :data-dir in --config is required".to_owned())?;
            let turso_path = turso_path.or(file.turso_path);
            let postgres_url = postgres_url.or(file.postgres_url);
            let postgres_read_only_url = postgres_read_only_url.or(file.postgres_read_only_url);
            let s3_bucket = s3_bucket.or(file.s3_bucket);
            let s3_prefix = s3_prefix.or(file.s3_prefix).unwrap_or_default();
            let s3_region = s3_region.or(file.s3_region);
            let s3_endpoint_url = s3_endpoint_url.or(file.s3_endpoint_url);
            let s3_read_only_access_key_id =
                s3_read_only_access_key_id.or(file.s3_read_only_access_key_id);
            let s3_read_only_secret_access_key =
                s3_read_only_secret_access_key.or(file.s3_read_only_secret_access_key);
            let s3_read_only_session_token =
                s3_read_only_session_token.or(file.s3_read_only_session_token);
            let s3_read_only_role_arn = s3_read_only_role_arn.or(file.s3_read_only_role_arn);
            let s3_read_only_role_session_name =
                s3_read_only_role_session_name.or(file.s3_read_only_role_session_name);
            let s3_read_only_role_duration_seconds =
                s3_read_only_role_duration_seconds.or(file.s3_read_only_role_duration_seconds);
            let s3_read_only_role_external_id =
                s3_read_only_role_external_id.or(file.s3_read_only_role_external_id);
            let store_spec = store_spec(
                store,
                &data_dir,
                turso_path,
                postgres_url,
                s3_bucket,
                s3_prefix,
                s3_region.clone(),
                s3_endpoint_url.clone(),
            )?;
            let mut config = NodeConfig::new(data_dir);
            config.store = store_spec;
            config.storage_info = storage_info_config(StorageInfoOptions {
                postgres_read_only_url,
                s3_region,
                s3_endpoint_url,
                access_key_id: s3_read_only_access_key_id,
                secret_access_key: s3_read_only_secret_access_key,
                session_token: s3_read_only_session_token,
                role_arn: s3_read_only_role_arn,
                session_name: s3_read_only_role_session_name,
                duration_seconds: s3_read_only_role_duration_seconds,
                external_id: s3_read_only_role_external_id,
            })?;
            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,
            segment_cache_dir,
            segment_cache_capacity,
            segment_cache_memory,
            client,
            serve,
        } => {
            let tls = serve.tls()?;
            let cache = segment_cache_dir.map(|directory| SegmentCacheConfig {
                directory,
                capacity_bytes: segment_cache_capacity.expect("required with cache directory"),
                memory_capacity_bytes: segment_cache_memory.unwrap_or(64 * 1024 * 1024),
            });
            if cache.is_some() && !client.peer_bootstrap {
                return Err("segment cache requires --peer-bootstrap".into());
            }
            let config = client.connect_config_cache(db, cache).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,
            allow_writes,
            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,
                allow_writes,
            ));
            let pg_config = corium_pgwire::PgWireConfig {
                password,
                ..corium_pgwire::PgWireConfig::default()
            };
            tracing::info!(%listen, allow_writes, "postgres server serving");
            let access = if allow_writes {
                "guarded autocommit writes enabled"
            } else {
                "read-only"
            };
            eprintln!(
                "corium postgres-server: serving the database catalog on {listen} ({access})"
            );
            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();
        }
    }
}

#[derive(Default)]
struct TransactorFileConfig {
    store: Option<StoreKind>,
    data_dir: Option<PathBuf>,
    turso_path: Option<PathBuf>,
    postgres_url: Option<String>,
    postgres_read_only_url: Option<String>,
    s3_bucket: Option<String>,
    s3_prefix: Option<String>,
    s3_region: Option<String>,
    s3_endpoint_url: Option<String>,
    s3_read_only_access_key_id: Option<String>,
    s3_read_only_secret_access_key: Option<String>,
    s3_read_only_session_token: Option<String>,
    s3_read_only_role_arn: Option<String>,
    s3_read_only_role_session_name: Option<String>,
    s3_read_only_role_duration_seconds: Option<i32>,
    s3_read_only_role_external_id: Option<String>,
}

impl TransactorFileConfig {
    fn load(path: Option<&std::path::Path>) -> Result<Self, String> {
        let Some(path) = path else {
            return Ok(Self::default());
        };
        let text = std::fs::read_to_string(path)
            .map_err(|error| format!("cannot read config {}: {error}", path.display()))?;
        let form = corium_query::edn::read_one(&text)
            .map_err(|error| format!("bad config EDN in {}: {error}", path.display()))?;
        let Edn::Map(entries) = form else {
            return Err(format!(
                "config {} must contain one EDN map",
                path.display()
            ));
        };
        let mut config = Self::default();
        for (key, value) in entries {
            let Edn::Keyword(key) = key else {
                return Err(format!(
                    "config {} contains a non-keyword key",
                    path.display()
                ));
            };
            if key.namespace.is_some() {
                return Err(format!("unknown config key {key}"));
            }
            match key.name.as_str() {
                "store" => config.store = Some(config_store(&value)?),
                "data-dir" => config.data_dir = Some(PathBuf::from(config_string(&key, value)?)),
                "turso-path" => {
                    config.turso_path = Some(PathBuf::from(config_string(&key, value)?));
                }
                "postgres-url" => config.postgres_url = Some(config_string(&key, value)?),
                "postgres-read-only-url" => {
                    config.postgres_read_only_url = Some(config_string(&key, value)?);
                }
                "s3-bucket" => config.s3_bucket = Some(config_string(&key, value)?),
                "s3-prefix" => config.s3_prefix = Some(config_string(&key, value)?),
                "s3-region" => config.s3_region = Some(config_string(&key, value)?),
                "s3-endpoint-url" => {
                    config.s3_endpoint_url = Some(config_string(&key, value)?);
                }
                "s3-read-only-access-key-id" => {
                    config.s3_read_only_access_key_id = Some(config_string(&key, value)?);
                }
                "s3-read-only-secret-access-key" => {
                    config.s3_read_only_secret_access_key = Some(config_string(&key, value)?);
                }
                "s3-read-only-session-token" => {
                    config.s3_read_only_session_token = Some(config_string(&key, value)?);
                }
                "s3-read-only-role-arn" => {
                    config.s3_read_only_role_arn = Some(config_string(&key, value)?);
                }
                "s3-read-only-role-session-name" => {
                    config.s3_read_only_role_session_name = Some(config_string(&key, value)?);
                }
                "s3-read-only-role-duration-seconds" => {
                    let Edn::Long(seconds) = value else {
                        return Err(format!("config key {key} must be an integer"));
                    };
                    config.s3_read_only_role_duration_seconds = Some(
                        i32::try_from(seconds)
                            .map_err(|_| format!("config key {key} is out of range"))?,
                    );
                }
                "s3-read-only-role-external-id" => {
                    config.s3_read_only_role_external_id = Some(config_string(&key, value)?);
                }
                _ => return Err(format!("unknown config key {key}")),
            }
        }
        Ok(config)
    }
}

fn config_string(key: &corium_core::Keyword, value: Edn) -> Result<String, String> {
    let Edn::Str(value) = value else {
        return Err(format!("config key {key} must be a string"));
    };
    Ok(value)
}

fn config_store(value: &Edn) -> Result<StoreKind, String> {
    let value = match value {
        Edn::Keyword(keyword) if keyword.namespace.is_none() => keyword.name.as_str(),
        Edn::Str(value) => value,
        _ => return Err("config key :store must be a keyword or string".into()),
    };
    match value {
        "mem" => Ok(StoreKind::Mem),
        "fs" => Ok(StoreKind::Fs),
        "postgres" => Ok(StoreKind::Postgres),
        "turso" => Ok(StoreKind::Turso),
        "s3" => Ok(StoreKind::S3),
        _ => Err(format!("unknown config :store value {value:?}")),
    }
}

#[derive(Default)]
struct StorageInfoOptions {
    postgres_read_only_url: Option<String>,
    s3_region: Option<String>,
    s3_endpoint_url: Option<String>,
    access_key_id: Option<String>,
    secret_access_key: Option<String>,
    session_token: Option<String>,
    role_arn: Option<String>,
    session_name: Option<String>,
    duration_seconds: Option<i32>,
    external_id: Option<String>,
}

#[cfg(feature = "s3")]
fn storage_info_config(options: StorageInfoOptions) -> Result<StorageInfoConfig, String> {
    let StorageInfoOptions {
        postgres_read_only_url,
        s3_region,
        s3_endpoint_url,
        access_key_id,
        secret_access_key,
        session_token,
        role_arn,
        session_name,
        duration_seconds,
        external_id,
    } = options;
    let static_present =
        access_key_id.is_some() || secret_access_key.is_some() || session_token.is_some();
    if static_present && role_arn.is_some() {
        return Err("static S3 read-only credentials conflict with --s3-read-only-role-arn".into());
    }
    let credentials = if let Some(role_arn) = role_arn {
        let duration_seconds = duration_seconds.unwrap_or(900);
        if !(900..=43_200).contains(&duration_seconds) {
            return Err("S3 read-only role duration must be between 900 and 43200 seconds".into());
        }
        Some(S3ReadOnlyCredentials::AssumeRole {
            role_arn,
            session_name: session_name.unwrap_or_else(|| "corium-read-only".into()),
            duration_seconds,
            external_id,
        })
    } else if static_present {
        let access_key_id = access_key_id.ok_or_else(|| {
            "S3 read-only access key id and secret access key must be supplied together".to_owned()
        })?;
        let secret_access_key = secret_access_key.ok_or_else(|| {
            "S3 read-only access key id and secret access key must be supplied together".to_owned()
        })?;
        if session_name.is_some() || duration_seconds.is_some() || external_id.is_some() {
            return Err("S3 role options require --s3-read-only-role-arn".into());
        }
        Some(S3ReadOnlyCredentials::Static {
            access_key_id,
            secret_access_key,
            session_token,
        })
    } else {
        if session_name.is_some() || duration_seconds.is_some() || external_id.is_some() {
            return Err("S3 role options require --s3-read-only-role-arn".into());
        }
        None
    };
    Ok(StorageInfoConfig {
        postgres_connection_string: postgres_read_only_url,
        s3: credentials
            .map(|credentials| S3ReadOnlyConfig::new(s3_region, s3_endpoint_url, credentials)),
    })
}

#[cfg(not(feature = "s3"))]
fn storage_info_config(options: StorageInfoOptions) -> Result<StorageInfoConfig, String> {
    let StorageInfoOptions {
        postgres_read_only_url,
        s3_region,
        s3_endpoint_url,
        access_key_id,
        secret_access_key,
        session_token,
        role_arn,
        session_name,
        duration_seconds,
        external_id,
    } = options;
    if [
        s3_region,
        s3_endpoint_url,
        access_key_id,
        secret_access_key,
        session_token,
        role_arn,
        session_name,
        external_id,
    ]
    .iter()
    .any(Option::is_some)
        || duration_seconds.is_some()
    {
        return Err("this build lacks S3 support; rebuild corium-cli with --features s3".into());
    }
    Ok(StorageInfoConfig {
        postgres_connection_string: postgres_read_only_url,
    })
}

/// Resolves the `--store` flag and backend connection options into a [`StoreSpec`].
#[allow(clippy::too_many_arguments)]
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,
    s3_region: Option<String>,
    s3_endpoint_url: Option<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, s3_region, s3_endpoint_url),
    }
}

#[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,
    region: Option<String>,
    endpoint_url: Option<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,
        client: S3ClientConfig {
            region,
            endpoint_url,
            ..S3ClientConfig::default()
        },
    })
}

#[cfg(not(feature = "s3"))]
fn s3_spec(
    _s3_bucket: Option<String>,
    _s3_prefix: String,
    _region: Option<String>,
    _endpoint_url: Option<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) => read_schema_file(&path)?,
                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
        }
    }
}

fn read_schema_file(path: &Path) -> Result<Vec<Edn>, String> {
    let text = std::fs::read_to_string(path)
        .map_err(|error| format!("cannot read {}: {error}", path.display()))?;
    if path
        .extension()
        .and_then(std::ffi::OsStr::to_str)
        .is_some_and(|extension| extension.eq_ignore_ascii_case("toml"))
    {
        return corium_forms::toml_schema::parse_edn(&text)
            .map_err(|error| format!("bad schema TOML: {error}"));
    }

    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!()
        };
        Ok(items)
    } else {
        Ok(forms)
    }
}

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()),
    }
}

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

    #[test]
    fn transactor_edn_config_loads_storage_and_read_only_settings() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("corium.edn");
        std::fs::write(
            &path,
            r#"{
              :store :s3
              :data-dir "/srv/corium"
              :s3-bucket "corium-prod"
              :s3-prefix "tenant/"
              :s3-region "us-west-2"
              :s3-read-only-role-arn "arn:aws:iam::123456789012:role/corium-reader"
              :s3-read-only-role-duration-seconds 1200
            }"#,
        )
        .expect("write config");

        let config = TransactorFileConfig::load(Some(&path)).expect("load config");
        assert!(matches!(config.store, Some(StoreKind::S3)));
        assert_eq!(config.data_dir, Some(PathBuf::from("/srv/corium")));
        assert_eq!(config.s3_bucket.as_deref(), Some("corium-prod"));
        assert_eq!(config.s3_prefix.as_deref(), Some("tenant/"));
        assert_eq!(config.s3_region.as_deref(), Some("us-west-2"));
        assert_eq!(
            config.s3_read_only_role_arn.as_deref(),
            Some("arn:aws:iam::123456789012:role/corium-reader")
        );
        assert_eq!(config.s3_read_only_role_duration_seconds, Some(1200));
    }

    #[test]
    fn transactor_edn_config_rejects_unknown_keys() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("corium.edn");
        std::fs::write(&path, "{:data-dir \"data\" :postgress-url \"typo\"}")
            .expect("write config");
        let error = TransactorFileConfig::load(Some(&path))
            .err()
            .expect("unknown key");
        assert!(error.contains(":postgress-url"));
    }

    #[cfg(feature = "s3")]
    #[test]
    fn s3_read_only_credentials_must_be_complete_and_unambiguous() {
        let incomplete = storage_info_config(StorageInfoOptions {
            access_key_id: Some("ACCESS".into()),
            ..StorageInfoOptions::default()
        })
        .expect_err("missing secret");
        assert!(incomplete.contains("supplied together"));

        let conflicting = storage_info_config(StorageInfoOptions {
            access_key_id: Some("ACCESS".into()),
            secret_access_key: Some("SECRET".into()),
            role_arn: Some("arn:aws:iam::123456789012:role/reader".into()),
            ..StorageInfoOptions::default()
        })
        .expect_err("static and role conflict");
        assert!(conflicting.contains("conflict"));
    }
}

#[cfg(test)]
mod schema_file_tests {
    use corium_core::{Cardinality, Keyword};
    use corium_protocol::schemaform::schema_from_edn;

    use super::read_schema_file;

    #[test]
    fn reads_toml_schema_by_extension() {
        let directory = tempfile::tempdir().expect("tempdir");
        let path = directory.path().join("schema.toml");
        std::fs::write(
            &path,
            r#"
[[entity]]
name = "person"

[entity.attributes]
name = { type = "string", unique = "identity" }
tags = { type = "keyword", many = true }
"#,
        )
        .expect("write schema");

        let forms = read_schema_file(&path).expect("read TOML schema");
        let (schema, idents) = schema_from_edn(&forms).expect("install schema");
        let tags = idents
            .entid(&Keyword::new(Some("person"), "tags"))
            .expect("tags ident");
        assert_eq!(
            schema.get(tags).expect("tags attribute").cardinality,
            Cardinality::Many
        );
    }

    #[test]
    fn continues_to_read_edn_schema() {
        let directory = tempfile::tempdir().expect("tempdir");
        let path = directory.path().join("schema.edn");
        std::fs::write(
            &path,
            "[{:db/ident :person/name :db/valueType :db.type/string}]",
        )
        .expect("write schema");

        let forms = read_schema_file(&path).expect("read EDN schema");
        let (_, idents) = schema_from_edn(&forms).expect("install schema");
        assert!(
            idents
                .entid(&Keyword::new(Some("person"), "name"))
                .is_some()
        );
    }
}