decdn-cli 0.0.1

//! Integration tests for the loopback admin JSON-RPC surface (`appendix-local-admin-http.md`).
//!
//! Spawns the admin server directly and verifies each RPC method round-
//! trips through real HTTP via jsonrpsee's generated client bindings.
//! Unit tests for individual response shapes live next to the server
//! impl in `admin.rs`; this file owns the wire-level checks and the
//! CLI-side error-classification paths.

#![allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::indexing_slicing,
    clippy::panic
)]

use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::num::NonZeroU64;
use std::sync::Arc;
use std::time::Instant;

/// Test-only helper so call sites stay compact. The `unwrap` is
/// statically sound for any positive integer literal we pass in.
const fn nz(v: u64) -> NonZeroU64 {
    match NonZeroU64::new(v) {
        Some(n) => n,
        None => panic!("test value must be > 0"),
    }
}

use decdn_cache::CacheEngine;
use decdn_cli::commands::node as commands;
use decdn_common::admin::{AdminRpcClient, DrainRequest};
use decdn_common::cli::{
    DrainArgs, EvictArgs, HealthArgs, LanesArgs, PoolsArgs, ReloadArgs, StatusArgs,
};
use decdn_incentive::buyer_pool::{BuyerPoolState, BuyerPoolStore};
use decdn_incentive::{LaneState, MemoryBuyerPoolStore, MemoryPoolStateStore, PoolStateStore};
use decdn_node::admin::{
    self, AdminState, DhtStatusHandles, DrainTrigger, LaneActivityClock, LaneStatusHandles,
};
use decdn_node::dht::routing::NodeId;
use decdn_node::dht::{
    ConfigStakerSet, RecordStore, RecordStoreConfig, RepublishScheduler, StakerSet,
};
use decdn_node::metrics::Metrics;
use decdn_protocol::{ContentHash, Coverage};
use jsonrpsee::RpcModule;
use jsonrpsee::core::ClientError;
use jsonrpsee::core::client::ClientT;
use jsonrpsee::http_client::HttpClientBuilder;
use jsonrpsee::rpc_params;
use jsonrpsee::server::{Server, ServerConfig};
use std::assert_matches;
use tokio::net::TcpListener;
use tokio::sync::oneshot;

async fn bind_loopback() -> anyhow::Result<(TcpListener, SocketAddr)> {
    let listener = TcpListener::bind(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0)).await?;
    let addr = listener.local_addr()?;
    Ok((listener, addr))
}

/// Build a throwaway cache engine for tests that don't exercise cache
/// behavior — `AdminState::new` requires one and most of these tests only
/// check other RPC methods. The returned `TempDir` must outlive the
/// engine; callers bind it with `_tmp` to keep RAII in scope.
async fn test_cache() -> anyhow::Result<(CacheEngine, tempfile::TempDir)> {
    let tmp = tempfile::tempdir()?;
    let cache = CacheEngine::open(tmp.path(), Vec::new(), 1).await?;
    Ok((cache, tmp))
}

/// Build `DhtStatusHandles` seeded with two peers in distinct buckets
/// (0 and 255), two stakers, one provider record, one scheduled republish,
/// and a fixed refresh clock — enough for an end-to-end `admin_v1_status`
/// round-trip over real HTTP. Mirrors the unit-level seed in `admin.rs`.
fn seeded_dht_handles() -> DhtStatusHandles {
    let mut table = decdn_node::dht::RoutingTable::new(NodeId::from_bytes([0u8; 32]));
    let mut high = [0u8; 32];
    high[0] = 0x80; // bucket 255
    let mut low = [0u8; 32];
    low[31] = 0x01; // bucket 0
    table.insert(NodeId::from_bytes(high));
    table.insert(NodeId::from_bytes(low));

    let mut stakers = std::collections::HashSet::new();
    stakers.insert(NodeId::from_bytes([1u8; 32]));
    stakers.insert(NodeId::from_bytes([2u8; 32]));
    let staker_set: Arc<dyn StakerSet> = Arc::new(ConfigStakerSet::new(stakers));

    let mut store = RecordStore::new(RecordStoreConfig::default());
    store.insert_at(
        NodeId::from_bytes([5u8; 32]),
        ContentHash::from_bytes([7u8; 32]),
        Coverage::full(1),
        1_000,
    );

    let republish = Arc::new(RepublishScheduler::new());
    republish.schedule_steady(ContentHash::from_bytes([9u8; 32]));

    DhtStatusHandles {
        routing: Arc::new(std::sync::Mutex::new(table)),
        staker_set,
        record_store: Arc::new(std::sync::Mutex::new(store)),
        republish,
        refresh_clock: Arc::new(std::sync::atomic::AtomicU64::new(1_700_000_000_000_000)),
        refresh_interval: std::time::Duration::from_hours(1),
    }
}

/// Spawn an admin server with the given state, returning its URL and a
/// `(stop_tx, join)` pair. The join handle must be awaited after
/// sending on `stop_tx` so the test doesn't leak a background task.
async fn spawn_admin(
    state: AdminState,
) -> anyhow::Result<(String, oneshot::Sender<()>, tokio::task::JoinHandle<()>)> {
    let (listener, addr) = bind_loopback().await?;
    let (stop_tx, stop_rx) = oneshot::channel::<()>();
    let join = tokio::spawn(async move {
        admin::serve(listener, state, stop_rx).await.ok();
    });
    Ok((format!("http://{addr}"), stop_tx, join))
}

#[tokio::test]
async fn health_returns_hex_node_id_and_uptime() -> anyhow::Result<()> {
    let id = [0xABu8; 32];
    let (cache, _tmp) = test_cache().await?;
    let state = AdminState::new(
        id,
        Instant::now(),
        cache,
        None,
        Arc::new(DrainTrigger::new()),
        Arc::new(Metrics::new()),
    );
    let (url, stop_tx, join) = spawn_admin(state).await?;

    let client = HttpClientBuilder::default().build(&url)?;
    let resp = client.health().await?;
    assert_eq!(resp.node_id, "ab".repeat(32));
    // `uptime_s` is whole-seconds-since-`started_at`; on a fast test
    // host this is almost always 0. Just assert the type-correct
    // round-trip — `health` returning at all proves the wire format.
    assert!(
        resp.uptime_s < 60,
        "uptime_s suspiciously large: {}",
        resp.uptime_s
    );

    let _ = stop_tx.send(());
    join.await?;
    Ok(())
}

/// JSON-RPC `-32601 Method not found` is the equivalent of the old
/// HTTP `404` for an unknown route. Confirm the server returns it for
/// a method name outside the `admin_v1_` namespace.
#[tokio::test]
async fn unknown_method_returns_method_not_found() -> anyhow::Result<()> {
    let (cache, _tmp) = test_cache().await?;
    let state = AdminState::new(
        [0u8; 32],
        Instant::now(),
        cache,
        None,
        Arc::new(DrainTrigger::new()),
        Arc::new(Metrics::new()),
    );
    let (url, stop_tx, join) = spawn_admin(state).await?;

    let client = HttpClientBuilder::default().build(&url)?;
    let result: Result<serde_json::Value, ClientError> =
        client.request("admin_v1_nonexistent", rpc_params![]).await;
    match result {
        Err(ClientError::Call(obj)) => {
            // JSON-RPC 2.0 method-not-found code.
            assert_eq!(obj.code(), -32601, "code was {}: {:?}", obj.code(), obj);
        }
        other => panic!("expected Call(-32601), got: {other:?}"),
    }

    let _ = stop_tx.send(());
    join.await?;
    Ok(())
}

/// `admin_v1_lanes` round-trips through real HTTP via jsonrpsee's
/// generated client (issue #749): a store seeded with two lanes
/// surfaces both, ordered by descending outstanding (no activity
/// recorded), with the configured threshold echoed and eligibility
/// computed against it.
#[tokio::test]
async fn lanes_round_trips_seeded_store() -> anyhow::Result<()> {
    use alloy::primitives::{Address, U256};

    // A lane is keyed by `(pool_id, signer, provider)`; the admin surface
    // reports the pool id as the lane id and the signer as both
    // counterparty and voucher signer (the shared-pool model has no separate
    // delegate at the lane level — see `decdn_node::admin::build_lane_snapshots`).
    let mk = |pool_byte: u8, signer_byte: u8, last_amount: u64| -> LaneState {
        let mut id = [0u8; 32];
        id[31] = pool_byte;
        let mut signer = [0u8; 20];
        signer[19] = signer_byte;
        let provider = Address::repeat_byte(0xEE);
        LaneState::hydrate(
            id.into(),
            Address::from(signer),
            provider,
            U256::from(1_000_000_000u64), // cap — irrelevant to the snapshot
            0,                            // expiry — untracked
            U256::from(last_amount),
            U256::from(last_amount), // bytes_delivered — irrelevant to the snapshot
            None,
            decdn_incentive::LaneChain::NONE,
        )
    };

    let store = Arc::new(MemoryPoolStateStore::new());
    store.record(&mk(1, 0xAA, 2_000_000))?;
    store.record(&mk(2, 0xBB, 100_000))?;

    let (cache, _tmp) = test_cache().await?;
    let state = AdminState::new(
        [0u8; 32],
        Instant::now(),
        cache,
        None,
        Arc::new(DrainTrigger::new()),
        Arc::new(Metrics::new()),
    )
    .with_lanes(LaneStatusHandles {
        pool_store: store as Arc<dyn PoolStateStore>,
        lane_activity: LaneActivityClock::empty(),
        redeem_threshold_micro_usdc: 1_000_000,
    });
    let (url, stop_tx, join) = spawn_admin(state).await?;

    let client = HttpClientBuilder::default().build(&url)?;
    let resp = client.lanes().await?;
    assert_eq!(resp.redeem_threshold_micro_usdc, 1_000_000);
    assert_eq!(resp.lanes.len(), 2);
    // No activity → ordered by descending outstanding.
    let first = resp
        .lanes
        .first()
        .ok_or_else(|| anyhow::anyhow!("missing first lane"))?;
    assert_eq!(first.outstanding_micro_usdc, 2_000_000);
    // The shared-pool model tracks no per-voucher nonce; the admin surface
    // reports 0 unconditionally.
    assert_eq!(first.last_nonce, 0);
    assert!(first.settlement_eligible, "2 USDC >= 1 USDC threshold");
    assert!(first.pool_id.starts_with("0x"));
    assert!(first.counterparty.starts_with("0x"));
    assert_eq!(
        first.voucher_signer, first.counterparty,
        "the shared-pool model reports the lane signer as both fields"
    );
    assert_eq!(first.seconds_since_last_voucher, None);
    let second = resp
        .lanes
        .get(1)
        .ok_or_else(|| anyhow::anyhow!("missing second lane"))?;
    assert_eq!(second.outstanding_micro_usdc, 100_000);
    assert!(!second.settlement_eligible, "0.1 USDC < 1 USDC threshold");

    let _ = stop_tx.send(());
    join.await?;
    Ok(())
}

/// `admin_v1_lanes` on a node with no lane handles wired returns
/// an empty list and a zero threshold over the wire — exercising the
/// `with_lanes`-absent path end-to-end.
#[tokio::test]
async fn lanes_without_handles_returns_empty_over_http() -> anyhow::Result<()> {
    let (cache, _tmp) = test_cache().await?;
    let state = AdminState::new(
        [0u8; 32],
        Instant::now(),
        cache,
        None,
        Arc::new(DrainTrigger::new()),
        Arc::new(Metrics::new()),
    );
    let (url, stop_tx, join) = spawn_admin(state).await?;

    let client = HttpClientBuilder::default().build(&url)?;
    let resp = client.lanes().await?;
    assert!(resp.lanes.is_empty());
    assert_eq!(resp.redeem_threshold_micro_usdc, 0);

    let _ = stop_tx.send(());
    join.await?;
    Ok(())
}

/// `admin_v1_pools` round-trips through real HTTP (#2078): a seeded buyer
/// store surfaces every pool, sorted by id, with its deposit intact.
#[tokio::test]
async fn pools_round_trips_seeded_store() -> anyhow::Result<()> {
    use alloy::primitives::{Address, B256, U256};

    let store = Arc::new(MemoryBuyerPoolStore::new());
    let deployment = decdn_incentive::Deployment {
        chain_id: 421_614,
        payment_pool: Address::repeat_byte(0x9c),
    };
    // Two owners, so both rows survive the store's owner index.
    store.record(&BuyerPoolState::new(
        B256::repeat_byte(0xbb),
        deployment,
        Address::repeat_byte(0x22),
        Address::repeat_byte(0xcd),
        U256::from(9_000_000u64),
    ))?;
    store.record(&BuyerPoolState::new(
        B256::repeat_byte(0xaa),
        deployment,
        Address::repeat_byte(0x11),
        Address::repeat_byte(0xcd),
        U256::from(10_000_000u64),
    ))?;

    let (cache, _tmp) = test_cache().await?;
    let state = AdminState::new(
        [0u8; 32],
        Instant::now(),
        cache,
        None,
        Arc::new(DrainTrigger::new()),
        Arc::new(Metrics::new()),
    )
    .with_buyer_pools(store as Arc<dyn BuyerPoolStore>);
    let (url, stop_tx, join) = spawn_admin(state).await?;

    let client = HttpClientBuilder::default().build(&url)?;
    let resp = client.pools().await?;
    assert_eq!(resp.pools.len(), 2);
    assert!(resp.skipped.is_empty());
    let first = resp
        .pools
        .first()
        .ok_or_else(|| anyhow::anyhow!("missing first pool"))?;
    assert_eq!(first.pool_id, format!("{:#x}", B256::repeat_byte(0xaa)));
    assert_eq!(first.deposit_micro_usdc, 10_000_000);
    assert_eq!(first.chain_id, deployment.chain_id);
    assert!(first.owner.starts_with("0x"));

    let _ = stop_tx.send(());
    join.await?;
    Ok(())
}

/// `admin_v1_pools` on a node with no buyer leg wired is an error over the
/// wire, not an empty list. A node that never pays for pulls and a node that
/// owns no pools demand opposite responses from an operator chasing a
/// stranded deposit, and the CLI must be able to tell them apart (#2078).
#[tokio::test]
async fn pools_without_buyer_leg_is_an_error_over_http() -> anyhow::Result<()> {
    let (cache, _tmp) = test_cache().await?;
    let state = AdminState::new(
        [0u8; 32],
        Instant::now(),
        cache,
        None,
        Arc::new(DrainTrigger::new()),
        Arc::new(Metrics::new()),
    );
    let (url, stop_tx, join) = spawn_admin(state).await?;

    let client = HttpClientBuilder::default().build(&url)?;
    let err = client
        .pools()
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected an unavailable error"))?
        .to_string();
    assert!(
        err.contains("no buyer payment-pool store is attached"),
        "error should name the missing store, not blame config, got: {err}"
    );

    let _ = stop_tx.send(());
    join.await?;
    Ok(())
}

/// CLI `pools` against a dropped listener surfaces the connection-refused
/// hint, same as the other admin subcommands.
#[tokio::test]
async fn cli_pools_surfaces_connection_refused() -> anyhow::Result<()> {
    let (listener, addr) = bind_loopback().await?;
    drop(listener);

    let args = PoolsArgs {
        admin_url: Some(format!("http://{addr}")),
        config: None,
        json: false,
        timeout_ms: 2_000,
    };
    let err = commands::pools(&args, None)
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected connection-refused error"))?
        .to_string();
    assert!(
        err.contains("refused"),
        "error should mention 'refused', got: {err}"
    );
    Ok(())
}

/// CLI `lanes` against a dropped listener surfaces the
/// connection-refused hint, same as the other admin subcommands.
#[tokio::test]
async fn cli_lanes_surfaces_connection_refused() -> anyhow::Result<()> {
    let (listener, addr) = bind_loopback().await?;
    drop(listener);

    let args = LanesArgs {
        admin_url: Some(format!("http://{addr}")),
        config: None,
        json: false,
        timeout_ms: 2_000,
    };
    let err = commands::lanes(&args, None)
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected connection-refused error"))?
        .to_string();
    assert!(
        err.contains("refused"),
        "error should mention 'refused', got: {err}"
    );
    Ok(())
}

/// The CLI's connection-refused branch is the single most operator-
/// visible error path (mistyped `--admin-url`, node not running). Bind
/// a loopback socket, drop it, and call `commands::health` against its
/// address: the kernel will return `ECONNREFUSED` and the CLI should
/// surface that with the "is the node running?" hint.
#[tokio::test]
async fn cli_health_surfaces_connection_refused() -> anyhow::Result<()> {
    let (listener, addr) = bind_loopback().await?;
    drop(listener);

    let args = HealthArgs {
        admin_url: Some(format!("http://{addr}")),
        config: None,
        json: false,
        timeout_ms: 2_000,
    };
    let err = commands::health(&args, None)
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected connection-refused error"))?
        .to_string();
    assert!(
        err.contains("refused"),
        "error should mention 'refused', got: {err}"
    );
    Ok(())
}

#[tokio::test]
async fn cli_health_rejects_zero_timeout() -> anyhow::Result<()> {
    let args = HealthArgs {
        admin_url: Some("http://127.0.0.1:1".to_string()),
        config: None,
        json: false,
        timeout_ms: 0,
    };
    let err = commands::health(&args, None)
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected zero-timeout error"))?
        .to_string();
    assert!(
        err.contains("--timeout-ms"),
        "error should mention the flag, got: {err}"
    );
    Ok(())
}

/// Mirror the connection-refused coverage onto `decdn node evict`. The
/// CLI subcommand routes through the same `classify_client_error`
/// path as `health` and the other admin subcommands, but a regression
/// that swallowed the classification (or printed a misleading "evicted"
/// line on failure) would not be caught by the admin-RPC unit tests alone.
#[tokio::test]
async fn cli_evict_surfaces_connection_refused() -> anyhow::Result<()> {
    let (listener, addr) = bind_loopback().await?;
    drop(listener);

    let args = EvictArgs {
        hash: "0".repeat(64),
        dry_run: false,
        admin_url: Some(format!("http://{addr}")),
        config: None,
        json: false,
        timeout_ms: 2_000,
    };
    let err = commands::evict(&args, None)
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected connection-refused error"))?
        .to_string();
    assert!(
        err.contains("refused"),
        "error should mention 'refused', got: {err}"
    );
    Ok(())
}

#[tokio::test]
async fn cli_evict_rejects_zero_timeout() -> anyhow::Result<()> {
    let args = EvictArgs {
        hash: "0".repeat(64),
        dry_run: false,
        admin_url: Some("http://127.0.0.1:1".to_string()),
        config: None,
        json: false,
        timeout_ms: 0,
    };
    let err = commands::evict(&args, None)
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected zero-timeout error"))?
        .to_string();
    assert!(
        err.contains("--timeout-ms"),
        "error should mention the flag, got: {err}"
    );
    Ok(())
}

/// Same coverage on `decdn node reload`: surface the connection-refused
/// hint when the admin port is dead. The unit tests in `admin.rs` cover
/// the server-side branches; this guards the CLI's
/// `classify_client_error` path so a regression there can't reach
/// operators as a silent hang or unhelpful message.
#[tokio::test]
async fn cli_reload_surfaces_connection_refused() -> anyhow::Result<()> {
    let (listener, addr) = bind_loopback().await?;
    drop(listener);

    let args = ReloadArgs {
        admin_url: Some(format!("http://{addr}")),
        config: None,
        json: false,
        timeout_ms: 2_000,
    };
    let err = commands::reload(&args, None)
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected connection-refused error"))?
        .to_string();
    assert!(
        err.contains("refused"),
        "error should mention 'refused', got: {err}"
    );
    Ok(())
}

#[tokio::test]
async fn cli_reload_rejects_zero_timeout() -> anyhow::Result<()> {
    let args = ReloadArgs {
        admin_url: Some("http://127.0.0.1:1".to_string()),
        config: None,
        json: false,
        timeout_ms: 0,
    };
    let err = commands::reload(&args, None)
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected zero-timeout error"))?
        .to_string();
    assert!(
        err.contains("--timeout-ms"),
        "error should mention the flag, got: {err}"
    );
    Ok(())
}

/// `admin_v1_drain` over a live server fires the trigger and returns
/// `initiated: true`. The trigger itself wakes up the `wait()` call on the
/// other end — here we check just the wire-level response because wiring
/// the trigger into a real runtime is the runtime's integration test
/// territory (that would require starting `run()` in a background task,
/// which is test-infrastructure cost). The unit tests in `admin.rs`
/// prove the trigger fires; this test proves the RPC path reaches it.
#[tokio::test]
async fn admin_v1_drain_returns_initiated_true() -> anyhow::Result<()> {
    let (cache, _tmp) = test_cache().await?;
    let drain_trigger = Arc::new(DrainTrigger::new());
    let state = AdminState::new(
        [0u8; 32],
        Instant::now(),
        cache,
        None,
        Arc::clone(&drain_trigger),
        Arc::new(Metrics::new()),
    );
    let (url, stop_tx, join) = spawn_admin(state).await?;

    let client = HttpClientBuilder::default().build(&url)?;
    let resp = client.drain(DrainRequest { wait_admin: false }).await?;
    assert!(resp.initiated, "expected initiated=true");
    assert!(
        !resp.wait_admin_honored,
        "wait_admin=false must report wait_admin_honored=false"
    );
    // wait_admin=false keeps the SIGTERM-equivalent ordering.
    assert!(
        !drain_trigger.wait_admin(),
        "wait_admin=false must not flip wait_admin on"
    );

    // The RPC handler should have fired the trigger.
    let waited =
        tokio::time::timeout(std::time::Duration::from_millis(100), drain_trigger.wait()).await;
    assert!(waited.is_ok(), "drain RPC did not fire the DrainTrigger");

    let _ = stop_tx.send(());
    join.await?;
    Ok(())
}

/// Mirror the connection-refused coverage onto `decdn node drain` so the
/// CLI path is guarded the same way as `peers`, `health`, `reload`, etc.
#[tokio::test]
async fn cli_drain_surfaces_connection_refused() -> anyhow::Result<()> {
    let (listener, addr) = bind_loopback().await?;
    drop(listener);

    let args = DrainArgs {
        admin_url: Some(format!("http://{addr}")),
        config: None,
        json: false,
        timeout_ms: 2_000,
        wait: false,
        wait_timeout_secs: nz(30),
        wait_poll_ms: nz(250),
    };
    let err = commands::drain(&args, None)
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected connection-refused error"))?
        .to_string();
    assert!(
        err.contains("refused"),
        "error should mention 'refused', got: {err}"
    );
    Ok(())
}

#[tokio::test]
async fn cli_drain_rejects_zero_timeout() -> anyhow::Result<()> {
    let args = DrainArgs {
        admin_url: Some("http://127.0.0.1:1".to_string()),
        config: None,
        json: false,
        timeout_ms: 0,
        wait: false,
        wait_timeout_secs: nz(30),
        wait_poll_ms: nz(250),
    };
    let err = commands::drain(&args, None)
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected zero-timeout error"))?
        .to_string();
    assert!(
        err.contains("--timeout-ms"),
        "error should mention the flag, got: {err}"
    );
    Ok(())
}

/// `admin_v1_drain` takes a required `DrainRequest`: a call with no
/// `params` (`curl`, a Python script) is refused with JSON-RPC
/// `-32602 Invalid params` and must NOT fire the drain. We bypass the
/// generated client (which always sends the request) and call the raw
/// `request` method with `rpc_params![]` to reproduce the
/// parameter-less wire shape.
#[tokio::test]
async fn admin_v1_drain_without_params_is_rejected() -> anyhow::Result<()> {
    let (cache, _tmp) = test_cache().await?;
    let drain_trigger = Arc::new(DrainTrigger::new());
    let state = AdminState::new(
        [0u8; 32],
        Instant::now(),
        cache,
        None,
        Arc::clone(&drain_trigger),
        Arc::new(Metrics::new()),
    );
    let (url, stop_tx, join) = spawn_admin(state).await?;

    let client = HttpClientBuilder::default().build(&url)?;
    let res: Result<decdn_common::admin::DrainResponse, _> =
        client.request("admin_v1_drain", rpc_params![]).await;
    let err = res
        .err()
        .ok_or_else(|| anyhow::anyhow!("a no-params drain must be refused"))?;
    assert_matches!(
        &err,
        ClientError::Call(obj) if obj.code() == -32602,
        "expected -32602 Invalid params, got {err:?}"
    );
    let waited =
        tokio::time::timeout(std::time::Duration::from_millis(100), drain_trigger.wait()).await;
    assert!(waited.is_err(), "a refused drain must not fire the trigger");

    let _ = stop_tx.send(());
    join.await?;
    Ok(())
}

/// `decdn node drain --wait` succeeds when `in_flight_streams` is
/// already 0 — the server's gauge starts at 0 and the polling loop
/// returns on the first `health()` tick.
#[tokio::test]
async fn cli_drain_wait_returns_immediately_when_idle() -> anyhow::Result<()> {
    let (cache, _tmp) = test_cache().await?;
    let drain_trigger = Arc::new(DrainTrigger::new());
    let state = AdminState::new(
        [0u8; 32],
        Instant::now(),
        cache,
        None,
        Arc::clone(&drain_trigger),
        Arc::new(Metrics::new()),
    );
    let (url, stop_tx, join) = spawn_admin(state).await?;

    let args = DrainArgs {
        admin_url: Some(url.clone()),
        config: None,
        json: false,
        timeout_ms: 5_000,
        wait: true,
        wait_timeout_secs: nz(10),
        wait_poll_ms: nz(50),
    };
    commands::drain(&args, None).await?;

    // Server-side: drain RPC fired with wait_admin=true.
    assert!(
        drain_trigger.wait_admin(),
        "wait flag should be set on the server's trigger"
    );

    let _ = stop_tx.send(());
    join.await?;
    Ok(())
}

/// `decdn node drain --wait` fails fast (no polling) when the server
/// responds with `wait_admin_honored: false` — an earlier drain without
/// `wait_admin` won the first-writer-wins trigger, or a regression where
/// the runtime ordering wasn't reapplied. Without this guard, the imminent
/// ECONNREFUSED from the early admin tear-down would be misread as drain
/// completion while in-flight streams keep running — exactly the
/// false-success the reviewer flagged.
///
/// The simulated server is a minimal jsonrpsee `RpcModule` that hand-
/// rolls an `admin_v1_drain` returning `wait_admin_honored: false`. No
/// `admin_v1_health` is registered — the CLI must reject before ever
/// polling, so the absence proves the guard fired.
#[tokio::test]
async fn cli_drain_wait_refuses_unhonored_server() -> anyhow::Result<()> {
    let (listener, addr) = bind_loopback().await?;
    let std_listener = listener.into_std()?;
    let config = ServerConfig::builder().http_only().build();
    let server = Server::builder()
        .set_config(config)
        .build_from_tcp(std_listener)
        .map_err(|e| anyhow::anyhow!("build fake admin: {e}"))?;
    let mut module = RpcModule::new(());
    module.register_async_method("admin_v1_drain", |_params, _ctx, _ext| async move {
        // Server reports it will not keep admin alive through the drain,
        // triggering the CLI's safety guard.
        Ok::<_, jsonrpsee::types::ErrorObjectOwned>(serde_json::json!({
            "initiated": true,
            "wait_admin_honored": false,
        }))
    })?;
    let handle = server.start(module);

    let url = format!("http://{addr}");
    let args = DrainArgs {
        admin_url: Some(url.clone()),
        config: None,
        json: false,
        timeout_ms: 5_000,
        wait: true,
        wait_timeout_secs: nz(30),
        wait_poll_ms: nz(250),
    };
    let err = commands::drain(&args, None)
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected unhonored-server error"))?
        .to_string();
    assert!(
        err.contains("wait_admin_honored=false"),
        "error should mention wait_admin_honored=false, got: {err}"
    );

    handle.stop().ok();
    handle.stopped().await;
    Ok(())
}

/// `decdn node drain --wait` against a server that *does* honor
/// `wait_admin` but then closes admin mid-poll (the real
/// late-stop success terminal): the polling client's `health()`
/// call returns `ECONNREFUSED`, which the CLI treats as drain
/// completion. Without this test the ECONNREFUSED branch of the
/// polling loop is uncovered — a regression in
/// `is_admin_closed_transport_error`'s source-chain walk or in the
/// match-arm ordering would silently turn drain success into `Err`.
#[tokio::test]
async fn cli_drain_wait_treats_econnrefused_as_complete() -> anyhow::Result<()> {
    // Build a fake server that:
    //   1. Answers admin_v1_drain with wait_admin_honored=true.
    //   2. Stops the server as soon as the first admin_v1_health
    //      call lands, so the polling loop's *next* tick gets
    //      ECONNREFUSED — exactly the late-stop sequence the
    //      runtime executes in production.
    use std::sync::OnceLock;

    let (listener, addr) = bind_loopback().await?;
    let std_listener = listener.into_std()?;
    let config = ServerConfig::builder().http_only().build();
    let server = Server::builder()
        .set_config(config)
        .build_from_tcp(std_listener)
        .map_err(|e| anyhow::anyhow!("build fake admin: {e}"))?;
    let mut module = RpcModule::new(());
    module.register_async_method("admin_v1_drain", |_p, _c, _e| async move {
        Ok::<_, jsonrpsee::types::ErrorObjectOwned>(serde_json::json!({
            "initiated": true,
            "wait_admin_honored": true,
        }))
    })?;
    // Set lazily: `Server::start()` is called *after* the module is
    // built, so the closure can't capture the handle directly.
    let handle_holder: Arc<OnceLock<jsonrpsee::server::ServerHandle>> = Arc::new(OnceLock::new());
    let handle_for_health = Arc::clone(&handle_holder);
    let stop_guard: Arc<OnceLock<()>> = Arc::new(OnceLock::new());
    module.register_async_method("admin_v1_health", move |_p, _c, _e| {
        let h = Arc::clone(&handle_for_health);
        let g = Arc::clone(&stop_guard);
        async move {
            // First health() call schedules a stop; subsequent
            // calls (if any race in before the stop lands) no-op
            // via the OnceLock guard.
            if g.set(()).is_ok()
                && let Some(handle) = h.get().cloned()
            {
                tokio::spawn(async move {
                    // Brief sleep so *this* poll's response is fully
                    // serialized before the listener closes — the
                    // CLI must observe `in_flight_streams=1` first,
                    // then the *next* poll hits ECONNREFUSED.
                    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
                    handle.stop().ok();
                });
            }
            Ok::<_, jsonrpsee::types::ErrorObjectOwned>(serde_json::json!({
                "node_id": "00".repeat(32),
                "uptime_s": 1,
                "in_flight_streams": 1,
                // `binding` and `registry_active` carry no `serde(default)`
                // (pre-deployment, so no compatibility shims — see CLAUDE.md),
                // which makes this stub a real check that the DTO and its
                // producers stay in step.
                "binding": "unknown",
                "bound_node_id": null,
                // A node being drained is one that was serving, so it was in
                // the on-chain registry (#1030). The drain path does not read
                // this field; it is here because the DTO requires it.
                "registry_active": true,
            }))
        }
    })?;
    let handle = server.start(module);
    let _ = handle_holder.set(handle.clone());

    let url = format!("http://{addr}");
    let args = DrainArgs {
        admin_url: Some(url.clone()),
        config: None,
        json: false,
        timeout_ms: 5_000,
        wait: true,
        wait_timeout_secs: nz(10),
        // 25ms poll < 50ms server-side stop delay: the first poll
        // completes, the server then closes during the next sleep,
        // and the second poll fires ECONNREFUSED.
        wait_poll_ms: nz(25),
    };
    commands::drain(&args, None).await?;

    handle.stopped().await;
    Ok(())
}

/// `decdn node drain --wait` exits non-zero with a `drain_timeout=true`
/// diagnostic on stderr when the in-flight count stays above zero past
/// the wait budget. Simulates the stuck-stream case by holding a
/// dispatch permit for the duration of the test.
#[tokio::test]
async fn cli_drain_wait_times_out_on_stuck_stream() -> anyhow::Result<()> {
    use decdn_common::config::ResolvedSecurity;

    let (cache, _tmp) = test_cache().await?;
    let drain_trigger = Arc::new(DrainTrigger::new());
    let metrics = Arc::new(Metrics::new());
    let state = AdminState::new(
        [0u8; 32],
        Instant::now(),
        cache,
        None,
        Arc::clone(&drain_trigger),
        Arc::clone(&metrics),
    );
    let (url, stop_tx, join) = spawn_admin(state).await?;

    // Hold one permit for the lifetime of the test — the limiter shares
    // the same metrics handle, so `dispatch_in_flight` reads 1 from the
    // CLI's `health()` polls.
    let limiter = decdn_node::dispatch::ConnectionLimiter::new(
        &ResolvedSecurity {
            max_concurrent_handlers: u32::MAX,
            per_source_rate_per_sec: 1e9,
            per_source_burst: u32::MAX,
            max_tracked_sources: 16,
        },
        Arc::clone(&metrics),
    );
    let _held = limiter
        .acquire_for_test(Some(std::net::IpAddr::V4(Ipv4Addr::LOCALHOST)))
        .ok()
        .ok_or_else(|| anyhow::anyhow!("permit acquire"))?;

    let args = DrainArgs {
        admin_url: Some(url.clone()),
        config: None,
        json: false,
        timeout_ms: 5_000,
        wait: true,
        // Sub-second wait budget so the test completes promptly.
        wait_timeout_secs: nz(1),
        wait_poll_ms: nz(50),
    };
    let err = commands::drain(&args, None)
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected timeout error"))?
        .to_string();
    assert!(
        err.contains("timed out"),
        "error should mention timeout, got: {err}"
    );
    assert!(
        err.contains("1 stream(s) still in flight") || err.contains("1 stream(s)"),
        "error should mention 1 in-flight, got: {err}"
    );

    let _ = stop_tx.send(());
    join.await?;
    Ok(())
}

#[tokio::test]
async fn admin_shutdown_closes_listener() -> anyhow::Result<()> {
    let (cache, _tmp) = test_cache().await?;
    let state = AdminState::new(
        [0u8; 32],
        Instant::now(),
        cache,
        None,
        Arc::new(DrainTrigger::new()),
        Arc::new(Metrics::new()),
    );
    let (url, stop_tx, join) = spawn_admin(state).await?;

    // Baseline: server is up.
    let client = HttpClientBuilder::default().build(&url)?;
    client.health().await?;

    let _ = stop_tx.send(());
    join.await?;

    // After shutdown the port should refuse new connections fairly
    // quickly. A fresh client is needed because jsonrpsee's HttpClient
    // uses a connection pool that might otherwise retry silently.
    // Bound the wait so a stuck test fails instead of hanging.
    let client2 = HttpClientBuilder::default().build(&url)?;
    let res = tokio::time::timeout(std::time::Duration::from_secs(2), client2.health()).await;
    match res {
        Ok(Ok(_)) => panic!("admin server still responding after shutdown"),
        // Either a jsonrpsee error or the outer timeout are acceptable
        // — both mean "no longer serving".
        Ok(Err(_)) | Err(_) => Ok(()),
    }
}

/// `admin_v1_status` round-trips through real HTTP: the seeded DHT health
/// (routing fills, staker count, record-store utilization, republish
/// depth, last-refresh timestamp) must survive serde + the generated
/// client bindings unchanged (issue #741).
#[tokio::test]
async fn status_round_trips_dht_health() -> anyhow::Result<()> {
    let (cache, _tmp) = test_cache().await?;
    let state = AdminState::new(
        [0xABu8; 32],
        Instant::now(),
        cache,
        None,
        Arc::new(DrainTrigger::new()),
        Arc::new(Metrics::new()),
    )
    .with_dht(seeded_dht_handles())
    .with_operator_address(
        "0x52908400098527886e0f7030069857d2e4169ee7"
            .parse()
            .expect("valid address"),
    );
    let (url, stop_tx, join) = spawn_admin(state).await?;

    let client = HttpClientBuilder::default().build(&url)?;
    let resp = client.status().await?;

    assert_eq!(resp.node_id, "ab".repeat(32));
    // The operator address survives the round-trip as an EIP-55 checksum.
    assert_eq!(
        resp.operator_address.as_deref(),
        Some("0x52908400098527886E0F7030069857D2E4169EE7")
    );
    assert_eq!(resp.routing.total_peers, 2);
    assert_eq!(resp.routing.non_empty_buckets, 2);
    let indices: Vec<u16> = resp.routing.buckets.iter().map(|b| b.index).collect();
    assert_eq!(indices, vec![0, 255]);
    assert_eq!(resp.routing.bucket_capacity, 20);
    assert_eq!(resp.routing.refresh_interval_s, 3_600);
    assert_eq!(resp.routing.last_refresh_us, Some(1_700_000_000_000_000));
    assert_eq!(resp.known_stakers, 2);
    assert_eq!(resp.record_store.records, 1);
    assert_eq!(resp.record_store.capacity, 1_000_000);
    assert_eq!(resp.republish.scheduled_records, 1);

    let _ = stop_tx.send(());
    join.await?;
    Ok(())
}

/// `decdn node status` against a dead port surfaces the friendly
/// connection-refused message (not a panic), mirroring the peers path.
#[tokio::test]
async fn cli_status_surfaces_connection_refused() -> anyhow::Result<()> {
    let (listener, addr) = bind_loopback().await?;
    drop(listener);

    let args = StatusArgs {
        admin_url: Some(format!("http://{addr}")),
        config: None,
        json: false,
        timeout_ms: 2_000,
    };
    let err = commands::status(&args, None)
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected connection-refused error"))?
        .to_string();
    assert!(
        err.contains("refused"),
        "error should mention 'refused', got: {err}"
    );
    Ok(())
}

/// `decdn node status --timeout-ms 0` is rejected before the client is
/// built, same guard as the other admin subcommands.
#[tokio::test]
async fn cli_status_rejects_zero_timeout() -> anyhow::Result<()> {
    let args = StatusArgs {
        admin_url: Some("http://127.0.0.1:1".to_string()),
        config: None,
        json: false,
        timeout_ms: 0,
    };
    let err = commands::status(&args, None)
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("expected zero-timeout error"))?
        .to_string();
    assert!(
        err.contains("--timeout-ms"),
        "error should mention the flag, got: {err}"
    );
    Ok(())
}