tsoracle 2.3.2

//
//  ░▀█▀░█▀▀░█▀█░█▀▄░█▀█░█▀▀░█░░░█▀▀
//  ░░█░░▀▀█░█░█░█▀▄░█▀█░█░░░█░░░█▀▀
//  ░░▀░░▀▀▀░▀▀▀░▀░▀░▀░▀░▀▀▀░▀▀▀░▀▀▀
//
//  tsoracle — Distributed Timestamp Oracle
//  https://www.tsoracle.rs
//
//  Copyright (c) 2026 Prisma Risk
//
//  Licensed under the Apache License, Version 2.0 (the "License");
//  you may not use this file except in compliance with the License.
//  You may obtain a copy of the License at
//
//      https://www.apache.org/licenses/LICENSE-2.0
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
//

//! The `tsoracle` CLI: `serve file|openraft|paxos` and `init`.
#![cfg_attr(not(test), warn(clippy::unwrap_used, clippy::expect_used))]

mod cli;

#[cfg(feature = "openraft")]
use std::collections::BTreeMap;

use anyhow::{Context, Result};
use clap::Parser;
use cli::{Cli, Cmd, CommonServeArgs, ServeCmd};
use tracing_subscriber::EnvFilter;
use tsoracle_server::Server;
use tsoracle_standalone::{DriverConfig, Standalone};

#[cfg(feature = "openraft")]
use tsoracle_standalone::admin_proto::{AdminErrorKind, ChangeResponse};

/// Drivers compiled into this build, for the friendly "not included" message.
#[cfg(any(
    not(feature = "file"),
    not(feature = "openraft"),
    not(feature = "paxos")
))]
fn available_drivers() -> &'static [&'static str] {
    &[
        #[cfg(feature = "file")]
        "file",
        #[cfg(feature = "openraft")]
        "openraft",
        #[cfg(feature = "paxos")]
        "paxos",
    ]
}

#[cfg(any(
    not(feature = "file"),
    not(feature = "openraft"),
    not(feature = "paxos")
))]
fn not_compiled_in(driver: &str) -> anyhow::Error {
    anyhow::anyhow!(
        "this build does not include the {driver} driver; rebuild with `--features {driver}`. \
         available drivers: {}",
        available_drivers().join(", ")
    )
}

#[tokio::main]
async fn main() -> Result<()> {
    let cli = Cli::parse();
    match cli.cmd {
        Some(Cmd::Init(args)) => run_init(args.state_dir, args.seed_physical_ms),
        Some(Cmd::Serve(serve)) => dispatch_serve(*serve).await,
        #[cfg(feature = "openraft")]
        Some(Cmd::Admin(cmd)) => dispatch_admin(cmd).await,
        // Bare `tsoracle` defaults to `serve file` when file is compiled in.
        None => {
            #[cfg(feature = "file")]
            {
                let file = cli.serve_file;
                let cfg = DriverConfig::File(tsoracle_standalone::FileConfig {
                    state_dir: file.state_dir,
                });
                return run_serve(file.common, cfg).await;
            }
            #[cfg(not(feature = "file"))]
            {
                anyhow::bail!(
                    "no subcommand given and this build excludes the file driver; \
                     specify `serve <driver>`. available drivers: {}",
                    available_drivers().join(", ")
                );
            }
        }
    }
}

async fn dispatch_serve(serve: ServeCmd) -> Result<()> {
    match serve {
        ServeCmd::File(args) => {
            #[cfg(feature = "file")]
            {
                let cfg = DriverConfig::File(tsoracle_standalone::FileConfig {
                    state_dir: args.state_dir,
                });
                run_serve(args.common, cfg).await
            }
            #[cfg(not(feature = "file"))]
            {
                let _ = args;
                Err(not_compiled_in("file"))
            }
        }
        ServeCmd::Openraft(args) => {
            #[cfg(feature = "openraft")]
            {
                let members = match args.members {
                    Some(s) => Some(parse_members(&s)?),
                    None => None,
                };
                let cfg = DriverConfig::Openraft(tsoracle_standalone::OpenraftConfig {
                    id: args.id,
                    raft_addr: args.raft_addr,
                    raft_dir: args.raft_dir,
                    bootstrap: args.bootstrap,
                    initial_membership: members,
                    tuning: tsoracle_standalone::RaftTuning {
                        heartbeat_ms: args.heartbeat_ms,
                        election_min_ms: args.election_min_ms,
                        election_max_ms: args.election_max_ms,
                    },
                    peer_tls: peer_tls_config(
                        args.peer_tls_cert,
                        args.peer_tls_key,
                        args.peer_tls_ca,
                    )?,
                    admin_listen: args.admin_listen,
                    admin_tls: admin_tls_config(
                        args.admin_tls_cert,
                        args.admin_tls_key,
                        args.admin_tls_ca,
                    )?,
                    allow_insecure_peer: args.allow_insecure_peer,
                });
                run_serve(args.common, cfg).await
            }
            #[cfg(not(feature = "openraft"))]
            {
                let _ = args;
                Err(not_compiled_in("openraft"))
            }
        }
        ServeCmd::Paxos(args) => {
            #[cfg(feature = "paxos")]
            {
                let cfg = DriverConfig::Paxos(tsoracle_standalone::PaxosConfig {
                    node_id: args.node_id,
                    peer_listen: args.peer_listen,
                    peers: tsoracle_standalone::parse_peer_map(&args.peers)
                        .map_err(anyhow::Error::msg)?,
                    tso_peers: tsoracle_standalone::parse_peer_map(&args.tso_peers)
                        .map_err(anyhow::Error::msg)?,
                    data_dir: args.data_dir,
                    tick_interval: args.tick_interval,
                    peer_tls: peer_tls_config(
                        args.peer_tls_cert,
                        args.peer_tls_key,
                        args.peer_tls_ca,
                    )?,
                    allow_insecure_peer: args.allow_insecure_peer,
                });
                run_serve(args.common, cfg).await
            }
            #[cfg(not(feature = "paxos"))]
            {
                let _ = args;
                Err(not_compiled_in("paxos"))
            }
        }
    }
}

#[cfg(feature = "file")]
fn run_init(state_dir: std::path::PathBuf, seed_physical_ms: u64) -> Result<()> {
    tsoracle_standalone::init_file_seeded(&state_dir, seed_physical_ms)
        .with_context(|| format!("init state_dir={}", state_dir.display()))?;
    println!(
        "Initialized {} at seed physical_ms={seed_physical_ms}",
        state_dir.display()
    );
    Ok(())
}

#[cfg(not(feature = "file"))]
fn run_init(_state_dir: std::path::PathBuf, _seed_physical_ms: u64) -> Result<()> {
    Err(not_compiled_in("file"))
}

/// Assemble the client-API server TLS config from flags (server-auth + optional
/// client mTLS). `Ok(None)` when no TLS flags are given (plaintext). Eagerly
/// validated: `from_pem` is lazy and the server applies tls only inside serve()
/// (after bind + "serving on"), so we dry-run the acceptor build here.
fn client_tls_config(
    common: &CommonServeArgs,
) -> anyhow::Result<Option<tonic::transport::ServerTlsConfig>> {
    match (&common.tls_cert, &common.tls_key) {
        (None, None) => {
            if common.tls_client_ca.is_some() {
                anyhow::bail!("--tls-client-ca requires --tls-cert and --tls-key");
            }
            Ok(None)
        }
        (Some(cert), Some(key)) => {
            let cert_pem =
                std::fs::read(cert).with_context(|| format!("read {}", cert.display()))?;
            let key_pem = std::fs::read(key).with_context(|| format!("read {}", key.display()))?;
            let mut tls = tonic::transport::ServerTlsConfig::new()
                .identity(tonic::transport::Identity::from_pem(&cert_pem, &key_pem));
            if let Some(ca) = &common.tls_client_ca {
                let ca_pem = std::fs::read(ca).with_context(|| format!("read {}", ca.display()))?;
                tls = tls.client_ca_root(tonic::transport::Certificate::from_pem(&ca_pem));
            }
            tonic::transport::Server::builder()
                .tls_config(tls.clone())
                .context("invalid client-API TLS configuration")?;
            Ok(Some(tls))
        }
        _ => anyhow::bail!("--tls-cert and --tls-key must be provided together"),
    }
}

#[cfg(feature = "metrics")]
fn install_metrics_exporter(common: &CommonServeArgs) -> Result<()> {
    if common.no_metrics {
        tracing::info!("Prometheus metrics exporter disabled");
        return Ok(());
    }

    metrics_exporter_prometheus::PrometheusBuilder::new()
        .with_http_listener(common.metrics_listen)
        .install()
        .with_context(|| {
            format!(
                "install Prometheus metrics exporter on {}",
                common.metrics_listen
            )
        })?;
    tracing::info!(addr = %common.metrics_listen, "Prometheus metrics exporter listening");
    Ok(())
}

#[cfg(not(feature = "metrics"))]
fn install_metrics_exporter(common: &CommonServeArgs) -> Result<()> {
    if common.no_metrics {
        tracing::debug!("--no-metrics ignored: built without the `metrics` feature");
    }
    Ok(())
}

/// Assemble the peer `PeerTlsConfig` from the all-or-nothing flag trio.
#[cfg(any(feature = "openraft", feature = "paxos"))]
fn peer_tls_config(
    cert: Option<std::path::PathBuf>,
    key: Option<std::path::PathBuf>,
    ca: Option<std::path::PathBuf>,
) -> anyhow::Result<Option<tsoracle_standalone::PeerTlsConfig>> {
    match (cert, key, ca) {
        (None, None, None) => Ok(None),
        (Some(cert), Some(key), Some(ca)) => {
            Ok(Some(tsoracle_standalone::PeerTlsConfig { cert, key, ca }))
        }
        _ => anyhow::bail!(
            "--peer-tls-cert, --peer-tls-key, and --peer-tls-ca must all be set together"
        ),
    }
}

/// Assemble the admin `AdminTlsConfig` from the all-or-nothing flag trio.
#[cfg(feature = "openraft")]
fn admin_tls_config(
    cert: Option<std::path::PathBuf>,
    key: Option<std::path::PathBuf>,
    ca: Option<std::path::PathBuf>,
) -> anyhow::Result<Option<tsoracle_standalone::AdminTlsConfig>> {
    match (cert, key, ca) {
        (None, None, None) => Ok(None),
        (Some(cert), Some(key), Some(ca)) => {
            Ok(Some(tsoracle_standalone::AdminTlsConfig { cert, key, ca }))
        }
        _ => anyhow::bail!(
            "--admin-tls-cert, --admin-tls-key, and --admin-tls-ca must all be set together"
        ),
    }
}

/// Build a `ClientTlsConfig` for `tsoracle admin` from the all-or-nothing flag trio.
#[cfg(feature = "openraft")]
fn admin_client_tls(
    args: &cli::AdminClientTlsArgs,
) -> anyhow::Result<Option<tonic::transport::ClientTlsConfig>> {
    match (
        args.client_tls_cert.as_ref(),
        args.client_tls_key.as_ref(),
        args.client_tls_ca.as_ref(),
    ) {
        (None, None, None) => Ok(None),
        (Some(cert), Some(key), Some(ca)) => {
            let cert_pem =
                std::fs::read(cert).with_context(|| format!("read {}", cert.display()))?;
            let key_pem = std::fs::read(key).with_context(|| format!("read {}", key.display()))?;
            let ca_pem = std::fs::read(ca).with_context(|| format!("read {}", ca.display()))?;
            Ok(Some(
                tonic::transport::ClientTlsConfig::new()
                    .ca_certificate(tonic::transport::Certificate::from_pem(&ca_pem))
                    .identity(tonic::transport::Identity::from_pem(&cert_pem, &key_pem)),
            ))
        }
        _ => anyhow::bail!(
            "--client-tls-cert, --client-tls-key, and --client-tls-ca must all be set together"
        ),
    }
}

/// Dial an admin endpoint, optionally with mTLS material.
#[cfg(feature = "openraft")]
async fn admin_connect(
    endpoint: &str,
    tls: Option<&tonic::transport::ClientTlsConfig>,
) -> anyhow::Result<tonic::transport::Channel> {
    let builder = tonic::transport::Channel::from_shared(endpoint.to_string())
        .with_context(|| format!("invalid endpoint {endpoint}"))?;
    let builder = match tls {
        Some(t) => builder
            .tls_config(t.clone())
            .with_context(|| "apply admin client TLS")?,
        None => builder,
    };
    builder
        .connect()
        .await
        .with_context(|| format!("connect {endpoint}"))
}

/// Resolve when the node should stop serving clients. Two ways in:
///
/// - the OS shutdown signal — graceful: run the driver's pre-shutdown drain
///   (openraft: leadership handoff) before the gRPC server stops accepting;
/// - a transport server died (`fatal` tripped) — fail fast: skip the drain,
///   since the node is already degraded and the priority is a quick exit so
///   an orchestrator can restart it, not a best-effort handoff over a
///   possibly-broken transport.
async fn wait_for_stop(
    shutdown_signal: impl std::future::Future<Output = ()>,
    fatal: tsoracle_standalone::FatalSignal,
    drain: Option<std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>>,
) {
    tokio::select! {
        () = shutdown_signal => {
            if let Some(drain) = drain {
                drain.await;
            }
        }
        component = fatal.tripped() => {
            tracing::error!(component, "transport server died; shutting down to fail fast");
        }
    }
}

async fn run_serve(common: CommonServeArgs, cfg: DriverConfig) -> Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::try_new(&common.log).unwrap_or_else(|_| EnvFilter::new("info")))
        .init();

    install_metrics_exporter(&common)?;

    let mut node: Standalone = tsoracle_standalone::build(cfg)
        .await
        .context("driver bootstrap")?;
    let drain = node.take_drain();
    let tls = client_tls_config(&common)?;
    let mut builder = Server::builder()
        .consensus_driver(node.driver.clone())
        .window_ahead(common.window_ahead)
        .failover_advance(common.failover_advance)
        .lease_ttl_floor(common.lease_ttl_floor)
        .lease_ttl_ceiling(common.lease_ttl_ceiling)
        .heartbeat_interval(common.heartbeat_interval);
    if let Some(tls) = tls {
        builder = builder.tls_config(tls);
    }
    let server = builder.build().context("server build")?;

    let listener = tokio::net::TcpListener::bind(common.listen)
        .await
        .with_context(|| format!("bind {}", common.listen))?;
    let local_addr = listener.local_addr().context("listener.local_addr()")?;
    // Plain-stdout contract: supervisors parse this to discover the OS-picked port.
    println!("serving on {local_addr}");
    tracing::info!(addr = %local_addr, "tsoracle serving");

    // Stop on the OS shutdown signal (graceful: drain first) or on a fatal
    // transport failure (fail fast: a node whose peer/admin server died must
    // not keep running as a zombie).
    let fatal = node.fatal_signal();
    let shutdown = wait_for_stop(tsoracle_server::shutdown_signal(), fatal.clone(), drain);
    let result = server
        .serve_with_listener(listener, shutdown)
        .await
        .context("serve");
    node.shutdown().await;
    if let Some(component) = fatal.check() {
        // Non-zero exit so an orchestrator restarts the node.
        return Err(anyhow::anyhow!(
            "{component} terminated unexpectedly; exiting"
        ));
    }
    result
}

#[cfg(feature = "openraft")]
fn parse_members(input: &str) -> Result<BTreeMap<u64, tsoracle_standalone::MemberAddr>> {
    let mut out = BTreeMap::new();
    for entry in input.split(',') {
        let entry = entry.trim();
        if entry.is_empty() {
            continue;
        }
        let (id, addrs) = entry.split_once('=').with_context(|| {
            format!("bad member {entry:?}, expected id=raft_addr/service_endpoint/admin_endpoint")
        })?;
        let mut parts = addrs.split('/');
        let raft_addr = parts.next().filter(|s| !s.is_empty());
        let service_endpoint = parts.next();
        let admin_endpoint = parts.next();
        let (Some(raft_addr), Some(service_endpoint), Some(admin_endpoint)) =
            (raft_addr, service_endpoint, admin_endpoint)
        else {
            anyhow::bail!(
                "bad member {entry:?}, expected raft_addr/service_endpoint/admin_endpoint"
            );
        };
        out.insert(
            id.trim()
                .parse()
                .with_context(|| format!("bad member id in {entry:?}"))?,
            tsoracle_standalone::MemberAddr {
                raft_addr: raft_addr.trim().to_string(),
                service_endpoint: service_endpoint.trim().to_string(),
                admin_endpoint: admin_endpoint.trim().to_string(),
            },
        );
    }
    Ok(out)
}

#[cfg(feature = "openraft")]
#[cfg_attr(test, derive(Debug, PartialEq, Eq))]
enum ActivationOutcome {
    Success,
    GateRejected(String),
    NotLeader,
    TargetOutOfRange(String),
    Other { kind: String, message: String },
}

#[cfg(feature = "openraft")]
fn classify_activation(resp: &ChangeResponse) -> ActivationOutcome {
    if resp.ok {
        return ActivationOutcome::Success;
    }
    match AdminErrorKind::try_from(resp.error) {
        Ok(AdminErrorKind::MembersBelowTarget) => {
            ActivationOutcome::GateRejected(resp.message.clone())
        }
        Ok(AdminErrorKind::NotLeader) => ActivationOutcome::NotLeader,
        Ok(AdminErrorKind::TargetOutOfRange) => {
            ActivationOutcome::TargetOutOfRange(resp.message.clone())
        }
        // Decoded enum kinds other than the three script-meaningful ones
        // (e.g. MembershipChanged, Unsupported, Driver) — surface the enum
        // debug name. Undecodable integers fall through to the next arm.
        Ok(k) => ActivationOutcome::Other {
            kind: format!("{k:?}"),
            message: resp.message.clone(),
        },
        Err(_) => ActivationOutcome::Other {
            kind: resp.error.to_string(),
            message: resp.message.clone(),
        },
    }
}

#[cfg(feature = "openraft")]
fn role_label(role: i32) -> &'static str {
    use tsoracle_standalone::admin_proto::MemberRole;
    match MemberRole::try_from(role) {
        Ok(MemberRole::Voter) => "voter",
        Ok(MemberRole::Learner) => "learner",
        _ => "unknown",
    }
}

#[cfg(feature = "openraft")]
fn render_capabilities_table(
    report: &tsoracle_standalone::admin_proto::CapabilityReport,
) -> String {
    let mut out = String::from("node  min_readable  max_readable  active_write\n");
    for member in &report.members {
        if member.reachable {
            out.push_str(&format!(
                "{:<4}  {:<12}  {:<12}  {}\n",
                member.id,
                member.min_readable_version,
                member.max_readable_version,
                member.active_write_version
            ));
        } else {
            let detail = if member.unreachable_detail.is_empty() {
                String::new()
            } else {
                format!("    ({})", member.unreachable_detail)
            };
            out.push_str(&format!(
                "{:<4}  {:<12}  {:<12}  {}{}\n",
                member.id, "-", "-", "unreachable", detail
            ));
        }
    }
    out
}

#[cfg(feature = "openraft")]
fn render_members_with_caps(report: &tsoracle_standalone::admin_proto::CapabilityReport) -> String {
    let leader = if report.has_leader {
        report.leader.to_string()
    } else {
        "none".to_string()
    };
    let mut out = format!("leader: {leader}\n");
    for member in &report.members {
        let caps = if member.reachable {
            format!(
                "min_readable={} max_readable={} active_write={}",
                member.min_readable_version,
                member.max_readable_version,
                member.active_write_version
            )
        } else if member.unreachable_detail.is_empty() {
            "capabilities=unreachable".to_string()
        } else {
            format!("capabilities=unreachable ({})", member.unreachable_detail)
        };
        out.push_str(&format!(
            "  id={} role={} raft={} service={} admin={} {}\n",
            member.id,
            role_label(member.role),
            member.raft_addr,
            member.service_endpoint,
            member.admin_endpoint,
            caps
        ));
    }
    out
}

#[cfg(feature = "openraft")]
#[derive(serde::Serialize)]
struct JsonMember {
    id: u64,
    role: String,
    raft_addr: String,
    service_endpoint: String,
    admin_endpoint: String,
    reachable: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    min_readable_version: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    max_readable_version: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    active_write_version: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    unreachable_detail: Option<String>,
}

#[cfg(feature = "openraft")]
#[derive(serde::Serialize)]
struct JsonReport {
    leader: Option<u64>,
    members: Vec<JsonMember>,
}

#[cfg(feature = "openraft")]
fn to_json_report(report: &tsoracle_standalone::admin_proto::CapabilityReport) -> JsonReport {
    let members = report
        .members
        .iter()
        .map(|member| {
            let (min_r, max_r, active_w, detail) = if member.reachable {
                (
                    Some(member.min_readable_version),
                    Some(member.max_readable_version),
                    Some(member.active_write_version),
                    None,
                )
            } else {
                (None, None, None, Some(member.unreachable_detail.clone()))
            };
            JsonMember {
                id: member.id,
                role: role_label(member.role).to_string(),
                raft_addr: member.raft_addr.clone(),
                service_endpoint: member.service_endpoint.clone(),
                admin_endpoint: member.admin_endpoint.clone(),
                reachable: member.reachable,
                min_readable_version: min_r,
                max_readable_version: max_r,
                active_write_version: active_w,
                unreachable_detail: detail,
            }
        })
        .collect();
    JsonReport {
        leader: if report.has_leader {
            Some(report.leader)
        } else {
            None
        },
        members,
    }
}

#[cfg(feature = "openraft")]
async fn call_report_capabilities(
    client: &mut tsoracle_standalone::admin_proto::membership_admin_client::MembershipAdminClient<
        tonic::transport::Channel,
    >,
) -> anyhow::Result<tsoracle_standalone::admin_proto::CapabilityReport> {
    use tsoracle_standalone::admin_proto::ReportCapabilitiesRequest;
    match client
        .report_capabilities(ReportCapabilitiesRequest {})
        .await
    {
        Ok(resp) => Ok(resp.into_inner()),
        Err(status) if status.code() == tonic::Code::Unimplemented => {
            anyhow::bail!("this node's driver does not support format capabilities (openraft only)")
        }
        Err(status) => Err(anyhow::Error::new(status).context("report_capabilities")),
    }
}

#[cfg(feature = "openraft")]
async fn dispatch_admin(cmd: cli::AdminCmd) -> Result<()> {
    use cli::AdminCmd;
    use tsoracle_standalone::admin_proto::membership_admin_client::MembershipAdminClient;
    use tsoracle_standalone::admin_proto::{
        ActivateFormatRequest, AddLearnerRequest, ListMembersRequest, MemberRole, PromoteRequest,
        RemoveNodeRequest,
    };

    // Re-run `op` against the leader if the first call says NOT_LEADER. The
    // leader-hint's scheme is reconstructed from the originally dialed
    // endpoint so a `https://` operator session doesn't get downgraded to
    // plaintext on redirect (and vice versa).
    async fn with_redirect<MakeCall>(
        endpoint: String,
        tls: Option<&tonic::transport::ClientTlsConfig>,
        op: MakeCall,
    ) -> Result<ChangeResponse>
    where
        MakeCall: Fn(
            MembershipAdminClient<tonic::transport::Channel>,
        ) -> std::pin::Pin<
            Box<dyn std::future::Future<Output = Result<ChangeResponse, tonic::Status>> + Send>,
        >,
    {
        let channel = admin_connect(&endpoint, tls).await?;
        let resp = op(MembershipAdminClient::new(channel))
            .await
            .context("admin rpc")?;
        if resp.error == AdminErrorKind::NotLeader as i32 && !resp.leader_admin_endpoint.is_empty()
        {
            let scheme = if endpoint.starts_with("https://") {
                "https"
            } else {
                "http"
            };
            let leader = format!("{scheme}://{}", resp.leader_admin_endpoint);
            let channel = admin_connect(&leader, tls).await?;
            return op(MembershipAdminClient::new(channel))
                .await
                .context("admin rpc (leader)");
        }
        Ok(resp)
    }

    fn report(resp: ChangeResponse) -> Result<()> {
        if resp.ok {
            println!("ok");
            Ok(())
        } else {
            let kind = AdminErrorKind::try_from(resp.error)
                .map(|kind| format!("{kind:?}"))
                .unwrap_or_else(|_| resp.error.to_string());
            anyhow::bail!("admin error ({kind}): {}", resp.message)
        }
    }

    /// Activation-specific reporter: success / gate-rejection / NOT_LEADER /
    /// local-range-rejection each get their own exit code per the issue #484
    /// contract. Anything else exits 1 via the bubbled `anyhow::Error`.
    /// The pure classification lives in `classify_activation` so it can be
    /// unit-tested without forking the process.
    fn report_activation(resp: ChangeResponse) -> Result<()> {
        match classify_activation(&resp) {
            ActivationOutcome::Success => {
                println!("ok");
                Ok(())
            }
            ActivationOutcome::GateRejected(message) => {
                eprintln!("activate-format: gate rejected: {message}");
                std::process::exit(2);
            }
            ActivationOutcome::NotLeader => {
                eprintln!("activate-format: not the leader");
                std::process::exit(3);
            }
            ActivationOutcome::TargetOutOfRange(message) => {
                eprintln!("activate-format: target outside local readable range: {message}");
                std::process::exit(4);
            }
            ActivationOutcome::Other { kind, message } => {
                anyhow::bail!("activate-format error ({kind}): {message}")
            }
        }
    }

    match cmd {
        AdminCmd::Members(args) => {
            let tls = admin_client_tls(&args.tls)?;
            let channel = admin_connect(&args.endpoint, tls.as_ref()).await?;
            let mut client = MembershipAdminClient::new(channel);
            if args.capabilities {
                let report = call_report_capabilities(&mut client).await?;
                print!("{}", render_members_with_caps(&report));
                return Ok(());
            }
            let view = client
                .list_members(ListMembersRequest {})
                .await
                .context("list_members")?
                .into_inner();
            let leader_str: String = if view.has_leader {
                view.leader.to_string()
            } else {
                "none".into()
            };
            println!("leader: {leader_str}");
            for member in view.members {
                let role = MemberRole::try_from(member.role)
                    .map(|role| format!("{role:?}"))
                    .unwrap_or_else(|_| member.role.to_string());
                println!(
                    "  id={} role={role} raft={} service={} admin={}",
                    member.id, member.raft_addr, member.service_endpoint, member.admin_endpoint
                );
            }
            Ok(())
        }
        AdminCmd::AddLearner(args) => {
            let tls = admin_client_tls(&args.tls)?;
            let endpoint = args.endpoint.clone();
            let id = args.id;
            let raft_addr = args.raft_addr.clone();
            let service_endpoint = args.service_endpoint.clone();
            let admin_endpoint = args.admin_endpoint.clone();
            report(
                with_redirect(endpoint, tls.as_ref(), move |mut client| {
                    let request = AddLearnerRequest {
                        id,
                        raft_addr: raft_addr.clone(),
                        service_endpoint: service_endpoint.clone(),
                        admin_endpoint: admin_endpoint.clone(),
                    };
                    Box::pin(
                        async move { client.add_learner(request).await.map(|r| r.into_inner()) },
                    )
                })
                .await?,
            )
        }
        AdminCmd::Promote(args) => {
            let tls = admin_client_tls(&args.tls)?;
            let endpoint = args.endpoint.clone();
            let id = args.id;
            report(
                with_redirect(endpoint, tls.as_ref(), move |mut client| {
                    Box::pin(async move {
                        client
                            .promote(PromoteRequest { id })
                            .await
                            .map(|r| r.into_inner())
                    })
                })
                .await?,
            )
        }
        AdminCmd::Remove(args) => {
            let tls = admin_client_tls(&args.tls)?;
            let endpoint = args.endpoint.clone();
            let id = args.id;
            report(
                with_redirect(endpoint, tls.as_ref(), move |mut client| {
                    Box::pin(async move {
                        client
                            .remove_node(RemoveNodeRequest { id })
                            .await
                            .map(|r| r.into_inner())
                    })
                })
                .await?,
            )
        }
        AdminCmd::ActivateFormat(args) => {
            let tls = admin_client_tls(&args.tls)?;
            let endpoint = args.endpoint.clone();
            let target = args.target;
            report_activation(
                with_redirect(endpoint, tls.as_ref(), move |mut client| {
                    Box::pin(async move {
                        client
                            .activate_format(ActivateFormatRequest {
                                target: target as u32,
                            })
                            .await
                            .map(|r| r.into_inner())
                    })
                })
                .await?,
            )
        }
        AdminCmd::Capabilities(args) => {
            let tls = admin_client_tls(&args.tls)?;
            let channel = admin_connect(&args.endpoint, tls.as_ref()).await?;
            let mut client = MembershipAdminClient::new(channel);
            let report = call_report_capabilities(&mut client).await?;
            if args.json {
                let json = serde_json::to_string_pretty(&to_json_report(&report))
                    .context("serialize capabilities json")?;
                println!("{json}");
            } else {
                print!("{}", render_capabilities_table(&report));
            }
            Ok(())
        }
    }
}

#[cfg(all(test, feature = "openraft"))]
mod capabilities_render_tests {
    use super::{render_capabilities_table, render_members_with_caps, to_json_report};
    use tsoracle_standalone::admin_proto::{CapabilityReport, MemberCapabilities, MemberRole};

    fn reachable(id: u64) -> MemberCapabilities {
        MemberCapabilities {
            id,
            role: MemberRole::Voter as i32,
            raft_addr: format!("h{id}:1"),
            service_endpoint: format!("h{id}:2"),
            admin_endpoint: format!("h{id}:3"),
            reachable: true,
            min_readable_version: 4,
            max_readable_version: 6,
            active_write_version: 4,
            unreachable_detail: String::new(),
        }
    }

    fn unreachable(id: u64) -> MemberCapabilities {
        MemberCapabilities {
            id,
            role: MemberRole::Voter as i32,
            raft_addr: format!("h{id}:1"),
            service_endpoint: format!("h{id}:2"),
            admin_endpoint: format!("h{id}:3"),
            reachable: false,
            min_readable_version: 0,
            max_readable_version: 0,
            active_write_version: 0,
            unreachable_detail: "connection refused".to_string(),
        }
    }

    fn report() -> CapabilityReport {
        CapabilityReport {
            members: vec![reachable(1), unreachable(3)],
            has_leader: true,
            leader: 1,
        }
    }

    #[test]
    fn table_has_header_and_reachable_and_unreachable_rows() {
        let out = render_capabilities_table(&report());
        let mut lines = out.lines();
        assert_eq!(
            lines.next().unwrap(),
            "node  min_readable  max_readable  active_write"
        );
        let reachable_row = lines.next().unwrap();
        assert!(reachable_row.starts_with('1'));
        assert!(reachable_row.contains('4') && reachable_row.contains('6'));
        let unreachable_row = lines.next().unwrap();
        assert!(unreachable_row.starts_with('3'));
        assert!(unreachable_row.contains("unreachable"));
        assert!(unreachable_row.contains("(connection refused)"));
    }

    #[test]
    fn members_view_augments_each_line_with_caps() {
        let out = render_members_with_caps(&report());
        assert!(out.contains("leader: 1"));
        assert!(out.contains("id=1") && out.contains("active_write=4"));
        assert!(
            out.contains("id=3") && out.contains("capabilities=unreachable (connection refused)")
        );
    }

    #[test]
    fn json_omits_version_fields_when_unreachable() {
        let json = serde_json::to_string(&to_json_report(&report())).unwrap();
        assert!(json.contains("\"leader\":1"));
        assert!(json.contains("\"active_write_version\":4"));
        assert!(json.contains("\"unreachable_detail\":\"connection refused\""));
        let unreachable_obj = json.split("\"id\":3").nth(1).unwrap();
        assert!(!unreachable_obj.contains("active_write_version"));
        assert!(unreachable_obj.contains("\"reachable\":false"));
    }
}

#[cfg(test)]
mod wait_for_stop_tests {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicBool, Ordering};

    use tsoracle_standalone::FatalSignal;

    use super::wait_for_stop;

    type DrainFuture = std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>;

    fn drain_probe() -> (Arc<AtomicBool>, Option<DrainFuture>) {
        let drained = Arc::new(AtomicBool::new(false));
        let drain_flag = drained.clone();
        let drain: DrainFuture = Box::pin(async move {
            drain_flag.store(true, Ordering::SeqCst);
        });
        (drained, Some(drain))
    }

    /// A fatal transport failure stops the node without any OS signal, and
    /// skips the graceful drain: the node is already degraded, so the
    /// priority is a fast exit and restart, not a leadership handoff.
    #[tokio::test]
    async fn fatal_trip_stops_without_signal_and_skips_drain() {
        let fatal = FatalSignal::new();
        let (drained, drain) = drain_probe();
        fatal.trip("peer server");
        wait_for_stop(std::future::pending(), fatal, drain).await;
        assert!(
            !drained.load(Ordering::SeqCst),
            "fatal path must skip the graceful drain"
        );
    }

    /// The OS-signal path keeps the existing graceful order: drain runs
    /// before the stop future resolves.
    #[tokio::test]
    async fn signal_path_runs_drain_before_stopping() {
        let fatal = FatalSignal::new();
        let (drained, drain) = drain_probe();
        wait_for_stop(std::future::ready(()), fatal, drain).await;
        assert!(drained.load(Ordering::SeqCst));
    }
}

#[cfg(all(test, feature = "openraft"))]
mod parse_members_tests {
    use super::parse_members;

    #[test]
    fn parses_three_address_members() {
        let map = parse_members("1=10.0.0.1:9/10.0.0.1:8/10.0.0.1:7").unwrap();
        let member = map.get(&1).expect("id 1 present");
        assert_eq!(member.raft_addr, "10.0.0.1:9");
        assert_eq!(member.service_endpoint, "10.0.0.1:8");
        assert_eq!(member.admin_endpoint, "10.0.0.1:7");
    }

    #[test]
    fn rejects_member_missing_admin_endpoint() {
        let err = parse_members("1=10.0.0.1:9/10.0.0.1:8").unwrap_err();
        assert!(
            err.to_string()
                .contains("raft_addr/service_endpoint/admin_endpoint"),
            "got: {err}"
        );
    }
}

#[cfg(all(test, feature = "openraft"))]
mod admin_tls_config_tests {
    use super::admin_tls_config;
    use std::path::PathBuf;

    #[test]
    fn none_returns_none() {
        let result = admin_tls_config(None, None, None).unwrap();
        assert!(result.is_none());
    }

    #[test]
    fn full_trio_returns_some() {
        let cfg = admin_tls_config(
            Some(PathBuf::from("/x/c")),
            Some(PathBuf::from("/x/k")),
            Some(PathBuf::from("/x/a")),
        )
        .unwrap()
        .expect("Some");
        assert_eq!(cfg.cert, PathBuf::from("/x/c"));
        assert_eq!(cfg.key, PathBuf::from("/x/k"));
        assert_eq!(cfg.ca, PathBuf::from("/x/a"));
    }

    #[test]
    fn partial_trio_errors() {
        for (c, k, a) in [
            (Some(PathBuf::from("c")), None, None),
            (None, Some(PathBuf::from("k")), None),
            (None, None, Some(PathBuf::from("a"))),
            (Some(PathBuf::from("c")), Some(PathBuf::from("k")), None),
            (Some(PathBuf::from("c")), None, Some(PathBuf::from("a"))),
            (None, Some(PathBuf::from("k")), Some(PathBuf::from("a"))),
        ] {
            let err = admin_tls_config(c, k, a).unwrap_err();
            assert!(
                err.to_string().contains("must all be set together"),
                "got: {err}"
            );
        }
    }
}

#[cfg(all(test, feature = "openraft"))]
mod admin_client_tls_tests {
    use super::admin_client_tls;
    use crate::cli::AdminClientTlsArgs;
    use std::path::PathBuf;

    fn args(cert: Option<&str>, key: Option<&str>, ca: Option<&str>) -> AdminClientTlsArgs {
        AdminClientTlsArgs {
            client_tls_cert: cert.map(PathBuf::from),
            client_tls_key: key.map(PathBuf::from),
            client_tls_ca: ca.map(PathBuf::from),
        }
    }

    #[test]
    fn none_returns_none() {
        let result = admin_client_tls(&args(None, None, None)).unwrap();
        assert!(result.is_none());
    }

    #[test]
    fn partial_trio_errors() {
        let err = admin_client_tls(&args(Some("c"), None, None)).unwrap_err();
        assert!(
            err.to_string().contains("must all be set together"),
            "got: {err}"
        );
    }

    // No "full trio returns Some" test here — that would require real PEM files
    // on disk because admin_client_tls reads + parses them. The integration test
    // in openraft_membership.rs already proves a real cert produces a working
    // channel.
}

#[cfg(all(test, feature = "openraft"))]
mod activation_exit_code_tests {
    use super::{ActivationOutcome, classify_activation};
    use tsoracle_standalone::admin_proto::{AdminErrorKind, ChangeResponse};

    fn resp(ok: bool, error: AdminErrorKind, message: &str) -> ChangeResponse {
        ChangeResponse {
            ok,
            error: error as i32,
            leader_admin_endpoint: String::new(),
            message: message.into(),
        }
    }

    #[test]
    fn ok_classifies_as_success() {
        let r = resp(true, AdminErrorKind::Unspecified, "");
        assert_eq!(classify_activation(&r), ActivationOutcome::Success);
    }

    #[test]
    fn members_below_target_classifies_as_gate_rejected() {
        let r = resp(false, AdminErrorKind::MembersBelowTarget, "blocked");
        assert!(matches!(
            classify_activation(&r),
            ActivationOutcome::GateRejected(s) if s == "blocked"
        ));
    }

    #[test]
    fn not_leader_classifies_as_not_leader() {
        let r = resp(false, AdminErrorKind::NotLeader, "");
        assert_eq!(classify_activation(&r), ActivationOutcome::NotLeader);
    }

    #[test]
    fn target_out_of_range_classifies_as_target_out_of_range() {
        let r = resp(false, AdminErrorKind::TargetOutOfRange, "range");
        assert!(matches!(
            classify_activation(&r),
            ActivationOutcome::TargetOutOfRange(s) if s == "range"
        ));
    }

    #[test]
    fn driver_error_classifies_as_other() {
        let r = resp(false, AdminErrorKind::Driver, "boom");
        assert!(matches!(
            classify_activation(&r),
            ActivationOutcome::Other { kind, message } if kind == "Driver" && message == "boom"
        ));
    }

    #[test]
    fn membership_changed_classifies_as_other() {
        let r = resp(false, AdminErrorKind::MembershipChanged, "drift");
        // MembershipChanged is a real activation outcome the operator can
        // remediate by re-issuing, but it's NOT a script-meaningful safety
        // rejection (Other → exit 1) — distinct from MembersBelowTarget
        // (exit 2) and TargetOutOfRange (exit 4). Explicit test so a
        // future refactor that quietly conflates them is caught.
        assert!(matches!(
            classify_activation(&r),
            ActivationOutcome::Other { kind, message }
                if kind == "MembershipChanged" && message == "drift"
        ));
    }
}