casper_devnet/

cli.rs

use crate::assets::{self, AssetsLayout, SetupOptions};
use crate::process::{self, ProcessHandle, RunningProcess, StartPlan};
use crate::state::{ProcessKind, ProcessRecord, ProcessStatus, STATE_FILE_NAME, State};
use anyhow::{Result, anyhow};
use backoff::ExponentialBackoff;
use backoff::backoff::Backoff;
use casper_types::U512;
use casper_types::contract_messages::MessagePayload;
use casper_types::execution::ExecutionResult;
use clap::{Args, Parser, Subcommand};
use directories::BaseDirs;
use futures::StreamExt;
use nix::errno::Errno;
use nix::sys::signal::kill;
use nix::unistd::Pid;
use serde::Deserialize;
use spinners::{Spinner, Spinners};
use std::collections::{HashMap, HashSet};
use std::os::unix::process::ExitStatusExt;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;
use tokio::fs as tokio_fs;
use tokio::sync::Mutex;
use tokio::sync::mpsc::{UnboundedSender, unbounded_channel};
use veles_casper_rust_sdk::sse::event::SseEvent;
use veles_casper_rust_sdk::sse::{self, config::ListenerConfig};

/// CLI entrypoint for the devnet launcher.
#[derive(Parser)]
#[command(name = "nctl")]
#[command(
    about = "casper-devnet launcher for local Casper Network development networks",
    long_about = None
)]
pub struct Cli {
    #[command(subcommand)]
    command: Command,
}

/// Top-level CLI subcommands.
#[derive(Subcommand)]
enum Command {
    /// Setup assets (if needed) and start the devnet.
    Start(StartArgs),
    /// Manage assets bundles.
    Assets(AssetsArgs),
    /// Check whether a network has observed a block yet.
    IsReady(IsReadyArgs),
}

/// Arguments for `nctl start`.
#[derive(Args, Clone)]
struct StartArgs {
    /// Network name used in assets paths and configs.
    #[arg(long, default_value = "casper-dev")]
    network_name: String,

    /// Override the base path for network runtime assets.
    #[arg(long, value_name = "PATH")]
    net_path: Option<PathBuf>,

    /// Protocol version to use from the assets store (e.g. 2.1.1).
    #[arg(long)]
    protocol_version: Option<String>,

    /// Number of nodes to create and start.
    #[arg(long = "node-count", aliases = ["nodes", "validators"], default_value_t = 4)]
    node_count: u32,

    /// Number of user accounts to generate (defaults to node count).
    #[arg(long)]
    users: Option<u32>,

    /// Genesis activation delay in seconds.
    #[arg(long, default_value_t = 3)]
    delay: u64,

    /// Log level for child processes (passed as `RUST_LOG`).
    #[arg(long = "log-level", default_value = "info")]
    log_level: String,

    /// Log format for node config files.
    #[arg(long, default_value = "json")]
    node_log_format: String,

    /// Create assets and exit without starting processes.
    #[arg(long)]
    setup_only: bool,

    /// Rebuild assets even if they already exist.
    #[arg(long)]
    force_setup: bool,

    /// Deterministic seed for devnet key generation.
    #[arg(long, default_value = "default")]
    seed: Arc<str>,
}

/// Asset management arguments.
#[derive(Args)]
struct AssetsArgs {
    #[command(subcommand)]
    command: AssetsCommand,
}

/// Asset management subcommands.
#[derive(Subcommand)]
enum AssetsCommand {
    /// Extract a local assets bundle into the assets store.
    Add(AssetsAddArgs),
    /// Download assets bundles from the upstream release.
    Pull(AssetsPullArgs),
    /// List available protocol versions in the assets store.
    List,
}

/// Arguments for `nctl assets add`.
#[derive(Args, Clone)]
struct AssetsAddArgs {
    /// Path to a local assets bundle (.tar.gz).
    #[arg(value_name = "PATH")]
    path: PathBuf,
}

/// Arguments for `nctl assets pull`.
#[derive(Args, Clone)]
struct AssetsPullArgs {
    /// Target triple to select from release assets.
    #[arg(long)]
    target: Option<String>,

    /// Re-download and replace any existing assets.
    #[arg(long)]
    force: bool,
}

/// Arguments for `nctl is-ready`.
#[derive(Args, Clone)]
struct IsReadyArgs {
    /// Network name used in assets paths and configs.
    #[arg(long, default_value = "casper-dev")]
    network_name: String,

    /// Override the base path for network runtime assets.
    #[arg(long, value_name = "PATH")]
    net_path: Option<PathBuf>,
}

/// Parses CLI and runs the selected subcommand.
pub async fn run() -> Result<()> {
    let cli = Cli::parse();
    match cli.command {
        Command::Start(args) => run_start(args).await,
        Command::Assets(args) => run_assets(args).await,
        Command::IsReady(args) => run_is_ready(args).await,
    }
}

async fn run_start(args: StartArgs) -> Result<()> {
    let assets_root = match &args.net_path {
        Some(path) => path.clone(),
        None => assets::default_assets_root()?,
    };
    let layout = AssetsLayout::new(assets_root, args.network_name.clone());
    let assets_path = shorten_home_path(&layout.net_dir().display().to_string());
    println!("assets path: {}", assets_path);
    let assets_exist = layout.exists().await;
    if !args.setup_only && !args.force_setup && assets_exist {
        println!("resuming network operations on {}", layout.network_name());
    }
    let protocol_version = resolve_protocol_version(&args.protocol_version).await?;

    if args.setup_only {
        return run_setup_only(&layout, &args, &protocol_version).await;
    }

    if args.force_setup {
        assets::teardown(&layout).await?;
        assets::setup_local(&layout, &setup_options(&args, &protocol_version)).await?;
    } else if !assets_exist {
        assets::setup_local(&layout, &setup_options(&args, &protocol_version)).await?;
    }

    if !layout.exists().await {
        return Err(anyhow!(
            "assets missing under {}; run with --setup-only to create them",
            shorten_home_path(&layout.net_dir().display().to_string())
        ));
    }

    if !args.force_setup && assets_exist {
        let restored = assets::ensure_consensus_keys(&layout, Arc::clone(&args.seed)).await?;
        if restored > 0 {
            println!("recreated consensus keys for {} node(s)", restored);
        }
    }

    let rust_log = args.log_level.clone();

    let plan = StartPlan { rust_log };

    let state_path = layout.net_dir().join(STATE_FILE_NAME);
    let state = Arc::new(Mutex::new(State::new(state_path).await?));
    let started = {
        let mut state = state.lock().await;
        process::start(&layout, &plan, &mut state).await?
    };

    print_pids(&started);
    print_start_banner(&layout, &started).await;
    print_derived_accounts_summary(&layout).await;

    let node_ids = unique_node_ids(&started);
    let details = format_network_details(&layout, &started).await;
    let health = Arc::new(Mutex::new(SseHealth::new(node_ids.clone(), details)));
    start_sse_spinner(&health).await;
    spawn_sse_listeners(layout.clone(), &node_ids, health, Arc::clone(&state)).await;

    let (event_tx, mut event_rx) = unbounded_channel();
    spawn_ctrlc_listener(event_tx.clone());
    spawn_exit_watchers(started, event_tx);

    if let Some(event) = event_rx.recv().await {
        match event {
            RunEvent::CtrlC => {
                let mut state = state.lock().await;
                process::stop(&mut state).await?;
            }
            RunEvent::ProcessExit {
                id,
                pid,
                code,
                signal,
            } => {
                let mut state = state.lock().await;
                update_exited_process(&mut state, &id, code, signal).await?;
                log_exit(&id, pid, code, signal);
                process::stop(&mut state).await?;
            }
        }
    }

    Ok(())
}

async fn run_setup_only(
    layout: &AssetsLayout,
    args: &StartArgs,
    protocol_version: &str,
) -> Result<()> {
    if args.force_setup {
        assets::teardown(layout).await?;
        assets::setup_local(layout, &setup_options(args, protocol_version)).await?;
        print_derived_accounts_summary(layout).await;
        return Ok(());
    }

    if layout.exists().await {
        println!(
            "assets already exist at {}; use --force-setup to rebuild",
            shorten_home_path(&layout.net_dir().display().to_string())
        );
        print_derived_accounts_summary(layout).await;
        return Ok(());
    }

    assets::setup_local(layout, &setup_options(args, protocol_version)).await?;
    print_derived_accounts_summary(layout).await;
    Ok(())
}

fn record_pid(record: &ProcessRecord) -> Option<u32> {
    if let Some(handle) = &record.pid_handle {
        let pid = handle.load(Ordering::SeqCst);
        if pid != 0 {
            return Some(pid);
        }
    }
    record.pid
}

fn setup_options(args: &StartArgs, protocol_version: &str) -> SetupOptions {
    SetupOptions {
        nodes: args.node_count,
        users: args.users,
        delay_seconds: args.delay,
        network_name: args.network_name.clone(),
        protocol_version: protocol_version.to_string(),
        node_log_format: args.node_log_format.clone(),
        seed: Arc::clone(&args.seed),
    }
}

fn print_pids(records: &[RunningProcess]) {
    for record in records {
        if let Some(pid) = record_pid(&record.record) {
            println!(
                "{} pid={} ({:?})",
                record.record.id, pid, record.record.kind
            );
        }
    }
}

async fn format_network_details(layout: &AssetsLayout, processes: &[RunningProcess]) -> String {
    let symlink_root = layout.net_dir();
    let mut node_pids: HashMap<u32, u32> = HashMap::new();
    let mut sidecar_pids: HashMap<u32, u32> = HashMap::new();
    let mut process_logs: HashMap<u32, Vec<(ProcessKind, u32)>> = HashMap::new();

    for process in processes {
        if let Some(pid) = record_pid(&process.record) {
            match process.record.kind {
                ProcessKind::Node => {
                    node_pids.insert(process.record.node_id, pid);
                }
                ProcessKind::Sidecar => {
                    sidecar_pids.insert(process.record.node_id, pid);
                }
            }
            process_logs
                .entry(process.record.node_id)
                .or_default()
                .push((process.record.kind.clone(), pid));
        }
    }

    let node_ids = unique_node_ids(processes);

    let mut lines = Vec::new();
    lines.push("network details".to_string());
    for node_id in node_ids {
        let node_pid = node_pids
            .get(&node_id)
            .map(|pid| pid.to_string())
            .unwrap_or_else(|| "-".to_string());
        let sidecar_pid = sidecar_pids
            .get(&node_id)
            .map(|pid| pid.to_string())
            .unwrap_or_else(|| "-".to_string());
        lines.push(format!("  node-{}", node_id));
        lines.push(format!(
            "    pids: node={} sidecar={}",
            node_pid, sidecar_pid
        ));
        if let Some(entries) = process_logs.get(&node_id) {
            let mut entries = entries.clone();
            entries.sort_by_key(|entry| process_kind_label(&entry.0).to_string());
            lines.push("    logs".to_string());
            for (kind, pid) in entries {
                let (stdout_link, stderr_link) = log_symlink_paths(&symlink_root, &kind, node_id);
                lines.push(format!(
                    "      {} pid={} stdout={} stderr={}",
                    process_kind_label(&kind),
                    pid,
                    stdout_link,
                    stderr_link
                ));
            }
        }
        lines.push("    endpoints".to_string());
        lines.push(format!("      rest:   {}", assets::rest_endpoint(node_id)));
        lines.push(format!("      sse:    {}", assets::sse_endpoint(node_id)));
        lines.push(format!("      rpc:    {}", assets::rpc_endpoint(node_id)));
        lines.push(format!("      binary: {}", assets::binary_address(node_id)));
        lines.push(format!(
            "      gossip: {}",
            assets::network_address(node_id)
        ));
    }

    lines.join("\n")
}

fn process_kind_label(kind: &ProcessKind) -> &'static str {
    match kind {
        ProcessKind::Node => "node",
        ProcessKind::Sidecar => "sidecar",
    }
}

fn shorten_home_path(path: &str) -> String {
    let path = Path::new(path);
    let Some(base_dirs) = BaseDirs::new() else {
        return path.display().to_string();
    };
    let home = base_dirs.home_dir();
    match path.strip_prefix(home) {
        Ok(stripped) => {
            if stripped.as_os_str().is_empty() {
                return "~".to_string();
            }
            let mut shorthand = PathBuf::from("~");
            shorthand.push(stripped);
            shorthand.display().to_string()
        }
        Err(_) => path.display().to_string(),
    }
}

fn log_symlink_paths(symlink_root: &Path, kind: &ProcessKind, node_id: u32) -> (String, String) {
    let base = match kind {
        ProcessKind::Node => format!("node-{}", node_id),
        ProcessKind::Sidecar => format!("sidecar-{}", node_id),
    };
    let stdout_link = symlink_root.join(format!("{}.stdout", base));
    let stderr_link = symlink_root.join(format!("{}.stderr", base));
    (
        shorten_home_path(&stdout_link.display().to_string()),
        shorten_home_path(&stderr_link.display().to_string()),
    )
}

async fn print_derived_accounts_summary(layout: &AssetsLayout) {
    if let Some(summary) = assets::derived_accounts_summary(layout).await {
        if let Some(parsed) = parse_derived_accounts_csv(&summary) {
            println!("derived accounts");
            if !parsed.validators.is_empty() {
                println!("  validators");
                print_account_group(&parsed.validators);
            }
            if !parsed.users.is_empty() {
                println!("  users");
                print_account_group(&parsed.users);
            }
            if !parsed.other.is_empty() {
                println!("  other");
                print_account_group(&parsed.other);
            }
        } else {
            println!("derived accounts");
            for line in summary.lines() {
                println!("  {}", line);
            }
        }
    }
}

struct DerivedAccountRow {
    name: String,
    key_type: String,
    derivation: String,
    path: String,
    account_hash: String,
    balance: String,
}

struct DerivedAccountsParsed {
    validators: Vec<DerivedAccountRow>,
    users: Vec<DerivedAccountRow>,
    other: Vec<DerivedAccountRow>,
}

fn parse_derived_accounts_csv(summary: &str) -> Option<DerivedAccountsParsed> {
    let mut lines = summary.lines();
    let header = lines.next()?.trim();
    if header != "kind,name,key_type,derivation,path,account_hash,balance" {
        return None;
    }

    let mut parsed = DerivedAccountsParsed {
        validators: Vec::new(),
        users: Vec::new(),
        other: Vec::new(),
    };

    for line in lines {
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        let mut parts = line.splitn(7, ',');
        let kind = parts.next()?.to_string();
        let name = parts.next()?.to_string();
        let key_type = parts.next()?.to_string();
        let derivation = parts.next()?.to_string();
        let path = parts.next()?.to_string();
        let account_hash = parts.next()?.to_string();
        let balance = parts.next()?.to_string();
        let row = DerivedAccountRow {
            name,
            key_type,
            derivation,
            path,
            account_hash,
            balance,
        };
        match kind.as_str() {
            "validator" => parsed.validators.push(row),
            "user" => parsed.users.push(row),
            _ => parsed.other.push(row),
        }
    }

    Some(parsed)
}

fn print_account_group(rows: &[DerivedAccountRow]) {
    for row in rows {
        println!("    {}:", row.name);
        println!("      key_type: {}", row.key_type);
        println!("      derivation: {}", row.derivation);
        println!("      path: {}", row.path);
        println!("      account_hash: {}", row.account_hash);
        println!("      balance: {}", row.balance);
    }
}

fn unique_node_ids(processes: &[RunningProcess]) -> Vec<u32> {
    let mut nodes = HashSet::new();
    for process in processes {
        nodes.insert(process.record.node_id);
    }
    let mut ids: Vec<u32> = nodes.into_iter().collect();
    ids.sort_unstable();
    ids
}

enum RunEvent {
    CtrlC,
    ProcessExit {
        id: String,
        pid: Option<u32>,
        code: Option<i32>,
        signal: Option<i32>,
    },
}

fn spawn_ctrlc_listener(tx: UnboundedSender<RunEvent>) {
    tokio::spawn(async move {
        if tokio::signal::ctrl_c().await.is_ok() {
            let _ = tx.send(RunEvent::CtrlC);
        }
    });
}

fn spawn_exit_watchers(processes: Vec<RunningProcess>, tx: UnboundedSender<RunEvent>) {
    for running in processes {
        let tx = tx.clone();
        tokio::spawn(async move {
            let id = running.record.id.clone();
            match running.handle {
                ProcessHandle::Child(mut child) => {
                    if let Ok(status) = child.wait().await {
                        let pid = record_pid(&running.record).or_else(|| child.id());
                        let code = status.code();
                        let signal = status.signal();
                        let _ = tx.send(RunEvent::ProcessExit {
                            id: id.clone(),
                            pid,
                            code,
                            signal,
                        });
                    }
                }
                ProcessHandle::Task(handle) => {
                    let status = handle.await;
                    let pid = record_pid(&running.record);
                    let (code, signal) = match status {
                        Ok(Ok(())) => (Some(0), None),
                        Ok(Err(_)) => (None, None),
                        Err(_) => (None, None),
                    };
                    let _ = tx.send(RunEvent::ProcessExit {
                        id: id.clone(),
                        pid,
                        code,
                        signal,
                    });
                }
            }
        });
    }
}

const SSE_WAIT_MESSAGE: &str = "Waiting for SSE connection...";
const BLOCK_WAIT_MESSAGE: &str = "Waiting for new blocks...";

struct SseHealth {
    expected_nodes: HashSet<u32>,
    versions: HashMap<u32, String>,
    announced: bool,
    block_seen: bool,
    sse_spinner: Option<Spinner>,
    block_spinner: Option<Spinner>,
    details: String,
}

impl SseHealth {
    fn new(node_ids: Vec<u32>, details: String) -> Self {
        Self {
            expected_nodes: node_ids.into_iter().collect(),
            versions: HashMap::new(),
            announced: false,
            block_seen: false,
            sse_spinner: None,
            block_spinner: None,
            details,
        }
    }
}

async fn should_log_primary(node_id: u32, health: &Arc<Mutex<SseHealth>>) -> bool {
    if node_id != 1 {
        return false;
    }
    let state = health.lock().await;
    state.announced
}

fn start_spinner(message: &str) -> Spinner {
    Spinner::new(Spinners::Dots, message.to_string())
}

async fn start_sse_spinner(health: &Arc<Mutex<SseHealth>>) {
    let mut state = health.lock().await;
    if state.sse_spinner.is_none() {
        state.sse_spinner = Some(start_spinner(SSE_WAIT_MESSAGE));
    }
}

async fn spawn_sse_listeners(
    layout: AssetsLayout,
    node_ids: &[u32],
    health: Arc<Mutex<SseHealth>>,
    state: Arc<Mutex<State>>,
) {
    for node_id in node_ids {
        let node_id = *node_id;
        let endpoint = assets::sse_endpoint(node_id);
        let layout = layout.clone();
        let health = Arc::clone(&health);
        let state = Arc::clone(&state);
        tokio::spawn(async move {
            run_sse_listener(node_id, endpoint, health, state, layout).await;
        });
    }
}

async fn run_sse_listener(
    node_id: u32,
    endpoint: String,
    health: Arc<Mutex<SseHealth>>,
    state: Arc<Mutex<State>>,
    layout: AssetsLayout,
) {
    let mut backoff = ExponentialBackoff::default();

    loop {
        let config = match ListenerConfig::builder()
            .with_endpoint(endpoint.clone())
            .build()
        {
            Ok(config) => config,
            Err(_) => {
                if !sleep_backoff(&mut backoff).await {
                    return;
                }
                continue;
            }
        };

        let stream = match sse::listener(config).await {
            Ok(stream) => {
                backoff.reset();
                stream
            }
            Err(_) => {
                if !sleep_backoff(&mut backoff).await {
                    return;
                }
                continue;
            }
        };

        futures::pin_mut!(stream);
        let mut stream_failed = false;
        while let Some(event) = stream.next().await {
            match event {
                Ok(sse_event) => match sse_event {
                    SseEvent::ApiVersion(version) => {
                        record_api_version(node_id, version.to_string(), &health).await;
                    }
                    SseEvent::BlockAdded { block_hash, block } => {
                        if node_id == 1
                            && let Err(err) = record_last_block_height(&state, block.height()).await
                        {
                            eprintln!("warning: failed to record last block height: {}", err);
                        }
                        if should_log_primary(node_id, &health).await {
                            mark_block_seen(&health, &layout).await;
                            let prefix = timestamp_prefix();
                            println!(
                                "{} Block {} added (height={} era={})",
                                prefix,
                                block_hash,
                                block.height(),
                                block.era_id().value()
                            );
                        }
                    }
                    SseEvent::TransactionAccepted(transaction) => {
                        if node_id == 1 {
                            let prefix = timestamp_prefix();
                            println!("{} Transaction {} accepted", prefix, transaction.hash());
                        }
                    }
                    SseEvent::TransactionProcessed {
                        transaction_hash,
                        execution_result,
                        messages,
                        ..
                    } => {
                        if node_id == 1 {
                            let tx_hash = transaction_hash.to_string();
                            let prefix = timestamp_prefix();
                            log_transaction_processed(
                                &prefix,
                                &tx_hash,
                                &execution_result,
                                &messages,
                            );
                        }
                    }
                    _ => {}
                },
                Err(_) => {
                    stream_failed = true;
                    break;
                }
            }
        }

        if stream_failed && !sleep_backoff(&mut backoff).await {
            return;
        }
    }
}

async fn record_api_version(node_id: u32, version: String, health: &Arc<Mutex<SseHealth>>) {
    let (summary, details, sse_spinner) = {
        let mut state = health.lock().await;
        if !state.expected_nodes.contains(&node_id) {
            return;
        }
        state.versions.insert(node_id, version);
        if state.announced || state.versions.len() != state.expected_nodes.len() {
            return;
        }

        let summary = version_summary(&state.versions);
        let details = state.details.clone();
        let sse_spinner = state.sse_spinner.take();
        if state.block_spinner.is_none() {
            state.block_spinner = Some(start_spinner(BLOCK_WAIT_MESSAGE));
        }
        state.announced = true;
        state.block_seen = false;
        (summary, details, sse_spinner)
    };

    if let Some(mut spinner) = sse_spinner {
        spinner.stop_with_message("SSE connection established.".to_string());
    }
    println!("Network is healthy ({})", summary);
    println!("{}", details);
}

async fn mark_block_seen(health: &Arc<Mutex<SseHealth>>, layout: &AssetsLayout) {
    let (block_spinner, node_ids) = {
        let mut state = health.lock().await;
        if state.block_seen {
            return;
        }
        state.block_seen = true;
        (
            state.block_spinner.take(),
            state.expected_nodes.iter().copied().collect::<Vec<_>>(),
        )
    };

    if let Some(mut spinner) = block_spinner {
        spinner.stop_with_message(BLOCK_WAIT_MESSAGE.to_string());
    }

    match assets::remove_consensus_keys(layout, &node_ids).await {
        Ok(removed) => {
            if removed > 0 {
                println!("Consensus secret keys removed from disk.");
            }
        }
        Err(err) => {
            eprintln!("warning: failed to remove consensus secret keys: {}", err);
        }
    }
}

async fn record_last_block_height(state: &Arc<Mutex<State>>, height: u64) -> Result<()> {
    let mut state = state.lock().await;
    if state.last_block_height == Some(height) {
        return Ok(());
    }
    state.last_block_height = Some(height);
    state.touch().await?;
    Ok(())
}

fn version_summary(versions: &HashMap<u32, String>) -> String {
    let mut unique: Vec<String> = versions.values().cloned().collect();
    unique.sort();
    unique.dedup();
    if unique.len() == 1 {
        format!("version {}", unique[0])
    } else {
        format!("versions {}", unique.join(", "))
    }
}

async fn sleep_backoff(backoff: &mut ExponentialBackoff) -> bool {
    if let Some(delay) = backoff.next_backoff() {
        tokio::time::sleep(delay).await;
        return true;
    }
    false
}

fn log_transaction_processed(
    prefix: &str,
    transaction_hash: &str,
    execution_result: &ExecutionResult,
    messages: &[casper_types::contract_messages::Message],
) {
    let consumed = execution_result.consumed();
    let consumed_cspr = format_cspr_u512(&consumed);
    if let Some(error) = execution_result.error_message() {
        println!(
            "{} Transaction {} processed failed ({}) gas={} gas_cspr={}",
            prefix, transaction_hash, error, consumed, consumed_cspr
        );
    } else {
        println!(
            "{} Transaction {} processed succeeded gas={} gas_cspr={}",
            prefix, transaction_hash, consumed, consumed_cspr
        );
    }

    for message in messages {
        let entity = message.entity_addr().to_formatted_string();
        let topic = message.topic_name();
        let payload = format_message_payload(message.payload());
        println!("{} 📨 {} {}: {}", prefix, entity, topic, payload);
    }
}

fn timestamp_prefix() -> String {
    time::OffsetDateTime::now_utc()
        .format(&time::format_description::well_known::Rfc3339)
        .unwrap_or_else(|_| "unknown-time".to_string())
}

fn format_message_payload(payload: &MessagePayload) -> String {
    match payload {
        MessagePayload::Bytes(bytes) => format!("0x{}", encode_hex(bytes.as_ref())),
        MessagePayload::String(value) => format!("{:?}", value),
    }
}

fn encode_hex(bytes: &[u8]) -> String {
    let mut out = String::with_capacity(bytes.len() * 2);
    for byte in bytes {
        use std::fmt::Write;
        let _ = write!(&mut out, "{:02x}", byte);
    }
    out
}

fn format_cspr_u512(motes: &U512) -> String {
    let motes_str = motes.to_string();
    let digits = motes_str.len();
    if digits <= 9 {
        let frac = format!("{:0>9}", motes_str);
        let frac = frac.trim_end_matches('0');
        if frac.is_empty() {
            return "0".to_string();
        }
        return format!("0.{}", frac);
    }

    let split = digits - 9;
    let (whole, frac) = motes_str.split_at(split);
    let frac = frac.trim_end_matches('0');
    if frac.is_empty() {
        return whole.to_string();
    }
    format!("{}.{}", whole, frac)
}

async fn update_exited_process(
    state: &mut State,
    id: &str,
    code: Option<i32>,
    signal: Option<i32>,
) -> Result<()> {
    for record in &mut state.processes {
        if record.id == id {
            record.last_status = ProcessStatus::Exited;
            record.exit_code = code;
            record.exit_signal = signal;
            record.stopped_at = Some(time::OffsetDateTime::now_utc());
            break;
        }
    }
    state.touch().await?;
    Ok(())
}

fn log_exit(id: &str, pid: Option<u32>, code: Option<i32>, signal: Option<i32>) {
    if let Some(pid) = pid {
        if let Some(signal) = signal {
            println!(
                "process {} (pid {}) exited due to signal {}",
                id, pid, signal
            );
        } else if let Some(code) = code {
            println!("process {} (pid {}) exited with code {}", id, pid, code);
        } else {
            println!("process {} (pid {}) exited", id, pid);
        }
    } else if let Some(signal) = signal {
        println!("process {} exited due to signal {}", id, signal);
    } else if let Some(code) = code {
        println!("process {} exited with code {}", id, code);
    } else {
        println!("process {} exited", id);
    }
}

async fn print_start_banner(layout: &AssetsLayout, processes: &[RunningProcess]) {
    let total_nodes = layout.count_nodes().await.unwrap_or(0);
    let target = format!("all nodes ({})", total_nodes);
    let sidecars = processes
        .iter()
        .filter(|proc| matches!(proc.record.kind, crate::state::ProcessKind::Sidecar))
        .count();
    println!(
        "started {} process(es) for {} (sidecars: {})",
        processes.len(),
        target,
        sidecars
    );
}

fn looks_like_url(path: &Path) -> bool {
    let value = path.to_string_lossy();
    value.starts_with("http://") || value.starts_with("https://")
}

async fn is_dir(path: &Path) -> bool {
    tokio_fs::metadata(path)
        .await
        .map(|meta| meta.is_dir())
        .unwrap_or(false)
}

async fn run_assets(args: AssetsArgs) -> Result<()> {
    match args.command {
        AssetsCommand::Add(add) => run_assets_add(add).await,
        AssetsCommand::Pull(pull) => run_assets_pull(pull).await,
        AssetsCommand::List => run_assets_list().await,
    }
}

async fn run_is_ready(args: IsReadyArgs) -> Result<()> {
    let assets_root = match &args.net_path {
        Some(path) => path.clone(),
        None => assets::default_assets_root()?,
    };
    let layout = AssetsLayout::new(assets_root, args.network_name);
    let argv0 = std::env::args()
        .next()
        .unwrap_or_else(|| "casper-devnet".to_string());
    let setup_cmd = format!("{} start --setup-only", argv0);
    let net_dir = layout.net_dir();
    if !is_dir(&net_dir).await {
        return Err(anyhow!(
            "assets for {} not found; run `{}`",
            layout.network_name(),
            setup_cmd
        ));
    }

    let state_path = net_dir.join(STATE_FILE_NAME);
    let contents = match tokio_fs::read_to_string(&state_path).await {
        Ok(contents) => contents,
        Err(_) => return Err(anyhow!("network is not ready yet")),
    };
    let state =
        match tokio::task::spawn_blocking(move || serde_json::from_str::<State>(&contents)).await {
            Ok(Ok(state)) => state,
            _ => return Err(anyhow!("network is not ready yet")),
        };

    ensure_processes_running(&state)?;

    if state.last_block_height.is_none() {
        return Err(anyhow!("network is not ready yet"));
    }

    ensure_rest_ready(&state).await?;
    Ok(())
}

async fn run_assets_add(args: AssetsAddArgs) -> Result<()> {
    if looks_like_url(&args.path) {
        return Err(anyhow!(
            "assets URL is not supported yet; provide a local .tar.gz path"
        ));
    }
    assets::install_assets_bundle(&args.path).await?;
    println!(
        "assets installed into {}",
        assets::assets_bundle_root()?.display()
    );
    Ok(())
}

async fn run_assets_pull(args: AssetsPullArgs) -> Result<()> {
    assets::pull_assets_bundles(args.target.as_deref(), args.force).await?;
    Ok(())
}

async fn run_assets_list() -> Result<()> {
    let mut versions = assets::list_bundle_versions().await?;
    if versions.is_empty() {
        return Err(anyhow!("no assets bundles found"));
    }
    versions.sort_by(|a, b| b.cmp(a));
    for version in versions {
        println!("{}", version);
    }
    Ok(())
}

async fn resolve_protocol_version(candidate: &Option<String>) -> Result<String> {
    if let Some(raw) = candidate {
        let version = assets::parse_protocol_version(raw)?;
        if !assets::has_bundle_version(&version).await? {
            let argv0 = std::env::args()
                .next()
                .unwrap_or_else(|| "casper-devnet".to_string());
            let pull_cmd = format!("{} assets pull", argv0);
            let add_cmd = format!("{} assets add <path-to-assets.tar.gz>", argv0);
            return Err(anyhow!(
                "assets for version {} not found; run `{}` or `{}`",
                version,
                pull_cmd,
                add_cmd
            ));
        }
        return Ok(version.to_string());
    }
    let versions = assets::list_bundle_versions().await?;
    if versions.is_empty() {
        let argv0 = std::env::args()
            .next()
            .unwrap_or_else(|| "casper-devnet".to_string());
        let pull_cmd = format!("{} assets pull", argv0);
        let add_cmd = format!("{} assets add <path-to-assets.tar.gz>", argv0);
        return Err(anyhow!(
            "no assets found; run `{}` or `{}`",
            pull_cmd,
            add_cmd
        ));
    }
    let version = versions
        .into_iter()
        .max()
        .expect("non-empty assets versions");
    Ok(version.to_string())
}

fn ensure_processes_running(state: &State) -> Result<()> {
    if state.processes.is_empty() {
        return Err(anyhow!("network is not ready yet"));
    }
    for process in &state.processes {
        if !matches!(process.last_status, ProcessStatus::Running) {
            return Err(anyhow!("network is not ready yet"));
        }
        let pid = match process.pid {
            Some(pid) => pid,
            None => return Err(anyhow!("network is not ready yet")),
        };
        if !is_pid_running(pid) {
            return Err(anyhow!("network is not ready yet"));
        }
    }
    Ok(())
}

fn is_pid_running(pid: u32) -> bool {
    let pid = Pid::from_raw(pid as i32);
    match kill(pid, None) {
        Ok(()) => true,
        Err(Errno::ESRCH) => false,
        Err(_) => true,
    }
}

async fn ensure_rest_ready(state: &State) -> Result<()> {
    let node_ids: HashSet<u32> = state
        .processes
        .iter()
        .filter_map(|process| {
            if matches!(process.kind, ProcessKind::Node) {
                Some(process.node_id)
            } else {
                None
            }
        })
        .collect();
    if node_ids.is_empty() {
        return Err(anyhow!("network is not ready yet"));
    }

    let client = reqwest::Client::builder()
        .no_proxy()
        .timeout(Duration::from_secs(2))
        .build()?;

    for node_id in node_ids {
        let url = format!("{}/status", assets::rest_endpoint(node_id));
        let response = match client.get(&url).send().await {
            Ok(response) => response,
            Err(_) => return Err(anyhow!("network is not ready yet")),
        };
        if response.status() != reqwest::StatusCode::OK {
            return Err(anyhow!("network is not ready yet"));
        }
        let status = match response.json::<NodeStatus>().await {
            Ok(status) => status,
            Err(_) => return Err(anyhow!("network is not ready yet")),
        };
        if !status
            .reactor_state
            .as_deref()
            .map(is_ready_reactor_state)
            .unwrap_or(false)
        {
            return Err(anyhow!("network is not ready yet"));
        }
    }
    Ok(())
}

fn is_ready_reactor_state(state: &str) -> bool {
    state == "Validate"
}

#[derive(Deserialize)]
#[allow(dead_code)]
#[serde(rename_all = "snake_case")]
struct NodeStatus {
    peers: Option<Vec<NodePeer>>,
    api_version: Option<String>,
    build_version: Option<String>,
    chainspec_name: Option<String>,
    starting_state_root_hash: Option<String>,
    last_added_block_info: Option<serde_json::Value>,
    our_public_signing_key: Option<String>,
    round_length: Option<serde_json::Value>,
    next_upgrade: Option<serde_json::Value>,
    uptime: Option<String>,
    reactor_state: Option<String>,
    last_progress: Option<String>,
    available_block_range: Option<BlockRange>,
    block_sync: Option<BlockSync>,
    latest_switch_block_hash: Option<serde_json::Value>,
}

#[derive(Deserialize)]
#[allow(dead_code)]
#[serde(rename_all = "snake_case")]
struct NodePeer {
    node_id: Option<String>,
    address: Option<String>,
}

#[derive(Deserialize)]
#[allow(dead_code)]
#[serde(rename_all = "snake_case")]
struct BlockRange {
    low: Option<u64>,
    high: Option<u64>,
}

#[derive(Deserialize)]
#[allow(dead_code)]
#[serde(rename_all = "snake_case")]
struct BlockSync {
    historical: Option<serde_json::Value>,
    forward: Option<serde_json::Value>,
}

#[cfg(test)]
mod tests {
    use super::{encode_hex, format_cspr_u512, format_message_payload, shorten_home_path};
    use casper_types::U512;
    use casper_types::contract_messages::MessagePayload;
    use directories::BaseDirs;

    #[test]
    fn format_cspr_u512_handles_whole_and_fractional() {
        assert_eq!(format_cspr_u512(&U512::zero()), "0");
        assert_eq!(format_cspr_u512(&U512::from(1u64)), "0.000000001");
        assert_eq!(format_cspr_u512(&U512::from(1_000_000_000u64)), "1");
        assert_eq!(
            format_cspr_u512(&U512::from(1_000_000_001u64)),
            "1.000000001"
        );
        assert_eq!(
            format_cspr_u512(&U512::from_dec_str("123000000000").unwrap()),
            "123"
        );
        assert_eq!(
            format_cspr_u512(&U512::from_dec_str("123000000456").unwrap()),
            "123.000000456"
        );
    }

    #[test]
    fn format_message_payload_renders_string_with_quotes() {
        let payload = MessagePayload::String("hello".to_string());
        assert_eq!(format_message_payload(&payload), "\"hello\"");
    }

    #[test]
    fn encode_hex_renders_lowercase() {
        assert_eq!(encode_hex(&[0x00, 0xAB, 0x0f]), "00ab0f");
    }

    #[test]
    fn shorten_home_path_replaces_home_prefix() {
        let Some(base_dirs) = BaseDirs::new() else {
            return;
        };
        let home = base_dirs.home_dir();
        let shortened = shorten_home_path(&home.to_string_lossy());
        assert_eq!(shortened, "~");

        let nested = home.join("devnet/logs/stdout.log");
        let shortened_nested = shorten_home_path(&nested.to_string_lossy());
        assert!(shortened_nested.starts_with("~"));
        assert!(shortened_nested.contains("devnet"));
    }

    #[test]
    fn shorten_home_path_keeps_relative_paths() {
        let input = "relative/path";
        assert_eq!(shorten_home_path(input), input);
    }
}