net-cli 0.27.0

//! `net transfer (recv-blob|send-blob|recv-dir|send-dir|ls|status|cancel)`
//! — operator surface over the `net_sdk::transport` blob / directory
//! movement primitives (`TRANSFER_CLI_PLAN.md`).
//!
//! Verb shapes:
//!
//! - `recv-blob` / `recv-dir` — **remote**. Stand up an ephemeral mesh,
//!   handshake with the holder (the same routed-attach path the
//!   `net aggregator` RPC verbs use), install the blob-transfer engine
//!   locally (required even to *fetch* — `fetch_blob` registers a
//!   pending transfer on the caller's engine), then pull the content
//!   and reconstruct it on disk. `recv-blob` writes a single file via a
//!   temp-and-rename; `recv-dir` delegates to `fetch_dir`, whose atomic
//!   sibling-temp-dir reconstruction (commit 636d31e) leaves the target
//!   untouched on failure.
//! - `send-blob` / `send-dir` — **local**. Compute the content-addressed
//!   [`BlobRef`] a peer fetches by, and (with `--store <dir>`) stage the
//!   bytes into an on-disk store a serving node can serve from. There is
//!   no "push" — this is the publish-and-fetch model (`TRANSFER_CLI_PLAN`
//!   "Out of scope: net transfer push"). Without `--store` the verb only
//!   prints the reference (a dry content-address computation).
//! - `ls` / `status` / `cancel` — **remote**. Query a holder's transfer
//!   engine over the `blob.transfers` RPC (remote-attach), reporting that
//!   node's requester-side in-flight transfers (what it's fetching). A
//!   single-shot CLI owns no engine, so there is nothing local to inspect;
//!   the holder exposes the RPC via `transport::serve_blob_transfer_rpc`.
//!   `cancel` drops the holder's pending entry, failing its awaiting fetch.
//!
//! `recv-*` **and** `ls` / `status` / `cancel` require remote-attach flags
//! — `--node-addr`, `--node-pubkey`, `--node-id`, `--psk-hex` — each
//! defaultable from the profile, exactly like the aggregator RPC verbs.

use std::path::PathBuf;
use std::sync::Arc;
use std::time::Instant;

use clap::{Args, Subcommand};
use serde::Serialize;

use crate::commands::aggregator::RemoteAttachArgs;
use crate::context::{resolve_profile, CliContext, RemoteAttach};
use crate::error::{generic, invalid_args, sdk, CliError};
use crate::parsers::parse_u64_flexible;
use crate::prelude::{emit_value, OutputFormat};

use net_sdk::dataforts::{MeshBlobAdapter, Redex};
use net_sdk::transport::{self, BlobRef, Encoding};

#[derive(Subcommand, Debug)]
pub enum TransferCommand {
    /// Receive a single blob from a peer and write it to `--out`.
    RecvBlob(RecvBlobArgs),
    /// Compute a blob's content reference (and optionally stage it to a
    /// local store); peers fetch by the printed reference.
    SendBlob(SendBlobArgs),
    /// Receive a directory atomically. Reconstruction uses the
    /// temp-and-rename pattern (commit 636d31e); on failure the local
    /// target is left unchanged.
    RecvDir(RecvDirArgs),
    /// Publish a directory: emit its manifest + chunk references (and
    /// optionally stage them to a local store).
    SendDir(SendDirArgs),
    /// List a holder's in-flight transfers over the mesh — what that node
    /// is *fetching* (requester side), not what it serves to others.
    Ls(LsArgs),
    /// Show one of a holder's fetches by stream id. Exits 0 with
    /// `found: false` when no such transfer is pending (check the field,
    /// not the exit code).
    Status(StatusArgs),
    /// Cancel one of a holder's in-progress fetches by stream id. Exits 0
    /// with `cancelled: false` when there was nothing to cancel (check the
    /// field, not the exit code).
    Cancel(CancelArgs),
}

#[derive(Args, Debug)]
pub struct RecvBlobArgs {
    /// Holder peer id to fetch from (decimal or `0x`-hex). Defaults to
    /// the remote-attach `--node-id` (the node you handshook with).
    #[arg(long, value_parser = crate::parsers::parse_u64_flexible)]
    pub from: Option<u64>,
    /// Content reference: a 32-byte hash (single-chunk blob) or the full
    /// encoded `BlobRef` hex that `send-blob` prints for chunked content.
    #[arg(long = "blob-ref")]
    pub blob_ref: String,
    /// Destination file. Written atomically via `<out>.partial` + rename.
    #[arg(long)]
    pub out: PathBuf,

    #[arg(long)]
    pub identity: Option<PathBuf>,
    #[arg(long, default_value_t = crate::prelude::DEFAULT_SUPERVISOR_NODE)]
    pub node: u64,
    #[command(flatten)]
    pub attach: RemoteAttachArgs,
}

#[derive(Args, Debug)]
pub struct SendBlobArgs {
    /// Source file. Pass `-` to read from stdin.
    pub path: PathBuf,
    /// Stage the bytes into an on-disk store at this directory so a node
    /// rooted there can serve them. Without it, only the reference is
    /// computed and printed (no bytes persisted).
    #[arg(long)]
    pub store: Option<PathBuf>,
}

#[derive(Args, Debug)]
pub struct RecvDirArgs {
    /// Holder peer id to fetch from (decimal or `0x`-hex). Defaults to
    /// the remote-attach `--node-id`.
    #[arg(long, value_parser = crate::parsers::parse_u64_flexible)]
    pub from: Option<u64>,
    /// Directory manifest reference: a 32-byte hash or the full encoded
    /// `BlobRef` hex that `send-dir` prints.
    #[arg(long = "remote-ref")]
    pub remote_ref: String,
    /// Destination directory. Reconstructed atomically (temp + rename).
    #[arg(long)]
    pub out: PathBuf,
    /// Leaf-file fetch concurrency. `0` uses the SDK default.
    #[arg(long, default_value_t = 0)]
    pub concurrency: usize,

    #[arg(long)]
    pub identity: Option<PathBuf>,
    #[arg(long, default_value_t = crate::prelude::DEFAULT_SUPERVISOR_NODE)]
    pub node: u64,
    #[command(flatten)]
    pub attach: RemoteAttachArgs,
}

#[derive(Args, Debug)]
pub struct SendDirArgs {
    /// Source directory.
    pub path: PathBuf,
    /// Stage the manifest + chunks into an on-disk store at this
    /// directory so a node rooted there can serve them. Without it, only
    /// the manifest reference is computed and printed.
    #[arg(long)]
    pub store: Option<PathBuf>,
}

#[derive(Args, Debug)]
pub struct LsArgs {
    #[arg(long)]
    pub identity: Option<PathBuf>,
    #[arg(long, default_value_t = crate::prelude::DEFAULT_SUPERVISOR_NODE)]
    pub node: u64,
    #[command(flatten)]
    pub attach: RemoteAttachArgs,
}

#[derive(Args, Debug)]
pub struct StatusArgs {
    /// Transfer id (stream id) to inspect (decimal or `0x`-hex).
    pub transfer_id: String,

    #[arg(long)]
    pub identity: Option<PathBuf>,
    #[arg(long, default_value_t = crate::prelude::DEFAULT_SUPERVISOR_NODE)]
    pub node: u64,
    #[command(flatten)]
    pub attach: RemoteAttachArgs,
}

#[derive(Args, Debug)]
pub struct CancelArgs {
    /// Transfer id (stream id) to cancel (decimal or `0x`-hex).
    pub transfer_id: String,

    #[arg(long)]
    pub identity: Option<PathBuf>,
    #[arg(long, default_value_t = crate::prelude::DEFAULT_SUPERVISOR_NODE)]
    pub node: u64,
    #[command(flatten)]
    pub attach: RemoteAttachArgs,
}

pub async fn run(
    cmd: TransferCommand,
    output: Option<OutputFormat>,
    config_path: Option<&std::path::Path>,
    profile_name: &str,
    quiet: bool,
) -> Result<(), CliError> {
    match cmd {
        TransferCommand::RecvBlob(args) => {
            run_recv_blob(args, output, config_path, profile_name, quiet).await
        }
        TransferCommand::SendBlob(args) => run_send_blob(args, output).await,
        TransferCommand::RecvDir(args) => {
            run_recv_dir(args, output, config_path, profile_name, quiet).await
        }
        TransferCommand::SendDir(args) => run_send_dir(args, output).await,
        TransferCommand::Ls(args) => run_ls(args, output, config_path, profile_name).await,
        TransferCommand::Status(args) => run_status(args, output, config_path, profile_name).await,
        TransferCommand::Cancel(args) => run_cancel(args, output, config_path, profile_name).await,
    }
}

// ── recv-blob ───────────────────────────────────────────────────────

async fn run_recv_blob(
    args: RecvBlobArgs,
    output: Option<OutputFormat>,
    config_path: Option<&std::path::Path>,
    profile_name: &str,
    quiet: bool,
) -> Result<(), CliError> {
    let blob_ref = parse_content_ref(&args.blob_ref, "--blob-ref")?;

    let profile = resolve_profile(config_path, profile_name).await?;
    let remote = require_remote_attach(&profile, &args.attach, "recv-blob")?;
    // `--from` overrides the attach target (fetch via a relay you
    // handshook with); otherwise the holder is the node you connected to.
    // (Parsed to `u64` at argv time, so no fallible re-parse here.)
    let attached = remote.node_id;
    let source = args.from.unwrap_or(attached);
    let ctx =
        CliContext::build_with_remote(&profile, args.identity.as_deref(), args.node, false, remote)
            .await?;
    let mesh = ctx.require_mesh()?;

    // Installing the engine is required to fetch, not just to serve:
    // `fetch_blob` registers a pending transfer on the local engine.
    transport::serve_blob_transfer(
        mesh,
        Arc::new(MeshBlobAdapter::new("recv", Arc::new(Redex::new()))),
    );

    // A determinate byte bar when the total size is known up front (a
    // Manifest ref, or a Small ref carrying its size); a spinner otherwise
    // (e.g. a bare-hash `--blob-ref`, whose size is unknown on the wire).
    let label = format!("fetching blob from peer {source}");
    let enabled = progress_enabled(output, quiet);
    let declared = blob_ref.size();
    let progress = if declared > 0 {
        Progress::start_bytes(&label, declared, enabled)
    } else {
        Progress::start(&label, enabled)
    };
    let started = Instant::now();
    // Stream the blob chunk-by-chunk straight to disk rather than
    // assembling the whole payload in RAM first: peak memory is ~one chunk,
    // so the practical size ceiling is free disk, not process memory.
    // `fetch_blob_stream` yields verified chunks in manifest order, so
    // writing them sequentially preserves integrity (no whole-blob rehash).
    use futures::StreamExt as _;
    // `fetch_blob_stream` returns a `!Unpin` stream (an `unfold`), so pin it
    // on the stack before polling with `next()`.
    let mut stream = std::pin::pin!(transport::fetch_blob_stream(mesh, source, &blob_ref));
    let mut writer = AtomicFileWriter::create(&args.out).await?;
    let mut total: u64 = 0;
    while let Some(item) = stream.next().await {
        let chunk = item.map_err(|e| {
            sdk(format!(
                "fetch_blob from peer {source} failed: {e}{}",
                relay_hint(source, attached)
            ))
        })?;
        total += chunk.len() as u64;
        progress.inc(chunk.len() as u64);
        writer.write_chunk(&chunk).await?;
    }
    writer.commit().await?;
    let elapsed = started.elapsed();
    progress.finish();

    let view = RecvBlobView {
        peer: source,
        out: args.out.display().to_string(),
        bytes: total,
        duration_secs: elapsed.as_secs_f64(),
        throughput_mib_s: throughput_mib_s(total, elapsed),
    };
    emit_value(OutputFormat::resolve_oneshot(output), &view)
        .map_err(|e| generic(format!("write recv-blob result: {e}")))?;
    Ok(())
}

// ── send-blob ───────────────────────────────────────────────────────

async fn run_send_blob(args: SendBlobArgs, output: Option<OutputFormat>) -> Result<(), CliError> {
    // Stage destination: a persistent adapter when `--store` is given, else
    // `None` — a dry run that computes the reference without persisting.
    let adapter = match args.store.as_deref() {
        Some(dir) => Some(persistent_adapter(dir, "send-blob").await?),
        None => None,
    };

    // Stream the source chunk-at-a-time into the content-addressed
    // reference (and, with `--store`, into the adapter as each chunk is
    // read) so the whole file is never held in memory. `store_blob_reader`
    // returns the exact `BlobRef` the buffered `chunk_payload` path would —
    // the value `recv-blob` fetches by.
    let blob_ref = if args.path.as_os_str() == "-" {
        transport::store_blob_reader(
            adapter.as_deref(),
            tokio::io::stdin(),
            "mesh://transfer",
            Encoding::Replicated,
        )
        .await
    } else {
        let file = tokio::fs::File::open(&args.path)
            .await
            .map_err(|e| generic(format!("open {}: {e}", args.path.display())))?;
        transport::store_blob_reader(
            adapter.as_deref(),
            file,
            "mesh://transfer",
            Encoding::Replicated,
        )
        .await
    }
    .map_err(|e| sdk(format!("send blob from {}: {e}", args.path.display())))?;

    let (small_hash, chunk_count) = match &blob_ref {
        // Bare chunk hash, present only for a single-chunk blob (so the
        // short form is unambiguous).
        BlobRef::Small { hash, .. } => (Some(hex::encode(hash)), 1u64),
        BlobRef::Manifest { chunks, .. } => (None, chunks.len() as u64),
        // `store_blob_reader` only ever yields Small / Manifest. Surface a
        // Tree as a hard error rather than fabricating `chunks: 0` — an
        // invariant break should fail loudly, not emit wrong output.
        BlobRef::Tree { .. } => {
            return Err(sdk("internal: store_blob_reader yielded a Tree BlobRef \
                 (it only produces Small / Manifest) — please report"));
        }
    };

    let view = SendBlobView {
        // Full-fidelity reference: works for single- and multi-chunk
        // content. `recv-blob --blob-ref <this>` decodes it back.
        blob_ref: hex::encode(blob_ref.encode()),
        hash: small_hash,
        size: blob_ref.size(),
        chunks: chunk_count,
        staged_to: args.store.as_deref().map(|d| d.display().to_string()),
    };
    emit_value(OutputFormat::resolve_oneshot(output), &view)
        .map_err(|e| generic(format!("write send-blob result: {e}")))?;
    Ok(())
}

// ── recv-dir ────────────────────────────────────────────────────────

async fn run_recv_dir(
    args: RecvDirArgs,
    output: Option<OutputFormat>,
    config_path: Option<&std::path::Path>,
    profile_name: &str,
    quiet: bool,
) -> Result<(), CliError> {
    let manifest_ref = parse_content_ref(&args.remote_ref, "--remote-ref")?;

    let profile = resolve_profile(config_path, profile_name).await?;
    let remote = require_remote_attach(&profile, &args.attach, "recv-dir")?;
    // `--from` is parsed to `u64` at argv time; default to the attach target.
    let attached = remote.node_id;
    let source = args.from.unwrap_or(attached);
    let ctx =
        CliContext::build_with_remote(&profile, args.identity.as_deref(), args.node, false, remote)
            .await?;
    let mesh = ctx.require_mesh()?;
    transport::serve_blob_transfer(
        mesh,
        Arc::new(MeshBlobAdapter::new("recv", Arc::new(Redex::new()))),
    );

    let spinner = Progress::start(
        &format!("reconstructing directory from peer {source}"),
        progress_enabled(output, quiet),
    );
    let started = Instant::now();
    // `fetch_dir` handles the atomic sibling-temp-dir reconstruction and
    // rolls back on failure, so the target is unchanged unless this
    // returns Ok. It also maps `concurrency == 0` to its own
    // `DEFAULT_FETCH_CONCURRENCY`, so we pass the operator's value through
    // verbatim rather than pre-resolving it here.
    let stats = transport::fetch_dir(mesh, source, &manifest_ref, &args.out, args.concurrency)
        .await
        .map_err(|e| {
            sdk(format!(
                "fetch_dir from peer {source} failed: {e}{}",
                relay_hint(source, attached)
            ))
        })?;
    let elapsed = started.elapsed();
    spinner.finish();

    let view = RecvDirView {
        peer: source,
        out: args.out.display().to_string(),
        files: stats.files as u64,
        dirs: stats.dirs as u64,
        symlinks: stats.symlinks as u64,
        bytes: stats.bytes,
        duration_secs: elapsed.as_secs_f64(),
        throughput_mib_s: throughput_mib_s(stats.bytes, elapsed),
        atomic: true,
    };
    emit_value(OutputFormat::resolve_oneshot(output), &view)
        .map_err(|e| generic(format!("write recv-dir result: {e}")))?;
    Ok(())
}

// ── send-dir ────────────────────────────────────────────────────────

async fn run_send_dir(args: SendDirArgs, output: Option<OutputFormat>) -> Result<(), CliError> {
    if !args.path.is_dir() {
        return Err(invalid_args(format!(
            "send-dir source `{}` is not a directory",
            args.path.display()
        )));
    }

    // `store_dir` walks the tree, content-addresses every leaf, builds
    // the manifest, and stores everything into the adapter — returning
    // the manifest `BlobRef` a peer fetches by. We need a real adapter
    // for it to store into; persist it iff `--store` was given, else use
    // an ephemeral in-memory one (manifest computed, bytes not retained).
    let (adapter, staged) = match args.store.as_deref() {
        Some(dir) => (
            persistent_adapter(dir, "send-dir").await?,
            Some(dir.display().to_string()),
        ),
        None => (
            Arc::new(MeshBlobAdapter::new("send-dir", Arc::new(Redex::new()))),
            None,
        ),
    };

    let manifest_ref = transport::store_dir(&adapter, &args.path)
        .await
        .map_err(|e| sdk(format!("store_dir `{}`: {e}", args.path.display())))?;

    let view = SendDirView {
        remote_ref: hex::encode(manifest_ref.encode()),
        manifest_size: manifest_ref.size(),
        staged_to: staged,
    };
    emit_value(OutputFormat::resolve_oneshot(output), &view)
        .map_err(|e| generic(format!("write send-dir result: {e}")))?;
    Ok(())
}

// ── ls / status / cancel ────────────────────────────────────────────

/// `ls` / `status` / `cancel` all query the holder's `blob.transfers`
/// engine over the mesh (remote-attach), reporting that node's in-flight
/// **requester-side** transfers (what it is currently fetching). Build the
/// connected client once per verb.
async fn transfer_client(
    attach: &RemoteAttachArgs,
    identity: Option<&std::path::Path>,
    node: u64,
    config_path: Option<&std::path::Path>,
    profile_name: &str,
    verb: &str,
) -> Result<(CliContext, u64), CliError> {
    let profile = resolve_profile(config_path, profile_name).await?;
    let remote = require_remote_attach(&profile, attach, verb)?;
    let target = remote.node_id;
    let ctx = CliContext::build_with_remote(&profile, identity, node, false, remote).await?;
    Ok((ctx, target))
}

async fn run_ls(
    args: LsArgs,
    output: Option<OutputFormat>,
    config_path: Option<&std::path::Path>,
    profile_name: &str,
) -> Result<(), CliError> {
    let (ctx, target) = transfer_client(
        &args.attach,
        args.identity.as_deref(),
        args.node,
        config_path,
        profile_name,
        "ls",
    )
    .await?;
    let client = transport::BlobTransferClient::new(ctx.require_mesh_node()?);
    let transfers = client
        .list(target)
        .await
        .map_err(|e| sdk(format!("blob.transfers list on peer {target} failed: {e}")))?;
    let view = LsView {
        transfer_count: transfers.len() as u64,
        transfers: transfers.iter().map(TransferRow::from).collect(),
    };
    emit_value(OutputFormat::resolve_oneshot(output), &view)
        .map_err(|e| generic(format!("write transfer ls: {e}")))?;
    Ok(())
}

async fn run_status(
    args: StatusArgs,
    output: Option<OutputFormat>,
    config_path: Option<&std::path::Path>,
    profile_name: &str,
) -> Result<(), CliError> {
    let stream_id = parse_u64_flexible(&args.transfer_id)
        .map_err(|e| invalid_args(format!("transfer-id `{}`: {e}", args.transfer_id)))?;
    let (ctx, target) = transfer_client(
        &args.attach,
        args.identity.as_deref(),
        args.node,
        config_path,
        profile_name,
        "status",
    )
    .await?;
    let client = transport::BlobTransferClient::new(ctx.require_mesh_node()?);
    let found = client.get(target, stream_id).await.map_err(|e| {
        sdk(format!(
            "blob.transfers status on peer {target} failed: {e}"
        ))
    })?;
    let view = StatusView {
        transfer_id: stream_id,
        found: found.is_some(),
        transfer: found.as_ref().map(TransferRow::from),
    };
    emit_value(OutputFormat::resolve_oneshot(output), &view)
        .map_err(|e| generic(format!("write transfer status: {e}")))?;
    Ok(())
}

async fn run_cancel(
    args: CancelArgs,
    output: Option<OutputFormat>,
    config_path: Option<&std::path::Path>,
    profile_name: &str,
) -> Result<(), CliError> {
    let stream_id = parse_u64_flexible(&args.transfer_id)
        .map_err(|e| invalid_args(format!("transfer-id `{}`: {e}", args.transfer_id)))?;
    let (ctx, target) = transfer_client(
        &args.attach,
        args.identity.as_deref(),
        args.node,
        config_path,
        profile_name,
        "cancel",
    )
    .await?;
    let client = transport::BlobTransferClient::new(ctx.require_mesh_node()?);
    let cancelled = client.cancel(target, stream_id).await.map_err(|e| {
        sdk(format!(
            "blob.transfers cancel on peer {target} failed: {e}"
        ))
    })?;
    let view = CancelView {
        transfer_id: stream_id,
        cancelled,
    };
    emit_value(OutputFormat::resolve_oneshot(output), &view)
        .map_err(|e| generic(format!("write transfer cancel: {e}")))?;
    Ok(())
}

// ── helpers ─────────────────────────────────────────────────────────

/// Parse a content reference: a 32-byte hex hash (a single-chunk
/// `Small` blob) or the full encoded `BlobRef` hex that `send-*` prints
/// for chunked content / directory manifests.
fn parse_content_ref(s: &str, flag: &str) -> Result<BlobRef, CliError> {
    // Strip at most one `0x` / `0X` prefix (not `trim_start_matches`, which
    // would peel repeated prefixes off a `0x0x…` typo).
    let body = s.trim();
    let trimmed = body
        .strip_prefix("0x")
        .or_else(|| body.strip_prefix("0X"))
        .unwrap_or(body);
    let bytes = hex::decode(trimmed)
        .map_err(|e| invalid_args(format!("{flag} `{s}` is not valid hex: {e}")))?;
    if bytes.len() == 32 {
        let mut hash = [0u8; 32];
        hash.copy_from_slice(&bytes);
        // The Small fetch path keys on the hash and ignores the size, so
        // 0 is a safe placeholder for the size we don't have on the wire.
        return Ok(BlobRef::small(format!("mesh://{trimmed}"), hash, 0));
    }
    match BlobRef::decode(&bytes) {
        Ok(Some(r)) => Ok(r),
        Ok(None) => Err(invalid_args(format!(
            "{flag} `{s}` decoded to an empty BlobRef"
        ))),
        Err(e) => Err(invalid_args(format!(
            "{flag} `{s}` is neither a 32-byte hash nor a valid encoded BlobRef: {e}"
        ))),
    }
}

/// Hint appended to a fetch failure when `--from` named a peer other than
/// the attach node: the failure is then often a missing relay route (the
/// attach node has to be able to reach the holder). Empty when fetching
/// directly from the attach node, where no relay is involved and the hint
/// would just be noise.
fn relay_hint(source: u64, attached: u64) -> String {
    if source == attached {
        String::new()
    } else {
        format!(
            " (fetching from peer {source} via attach node {attached}; \
             ensure {attached} has a route to {source})"
        )
    }
}

/// Resolve remote-attach for a recv verb that requires a holder target.
fn require_remote_attach(
    profile: &crate::config::Profile,
    args: &RemoteAttachArgs,
    verb: &str,
) -> Result<RemoteAttach, CliError> {
    crate::context::require_remote_attach(profile, args, || {
        invalid_args(format!(
            "net transfer {verb} needs a holder target: pass --node-addr <IP:PORT> \
             --node-pubkey <HEX> --node-id <N> --psk-hex <HEX> (each can be defaulted \
             in the profile as `node_addr` / `node_pubkey` / `node_id` / `psk_hex`)."
        ))
    })
}

/// Build a persistent on-disk blob adapter rooted at `dir`. The bytes
/// staged here are durable so a node configured against the same
/// directory can serve them.
async fn persistent_adapter(
    dir: &std::path::Path,
    id: &str,
) -> Result<Arc<MeshBlobAdapter>, CliError> {
    tokio::fs::create_dir_all(dir)
        .await
        .map_err(|e| generic(format!("create store dir {}: {e}", dir.display())))?;
    let redex = Arc::new(Redex::new().with_persistent_dir(dir));
    Ok(Arc::new(
        MeshBlobAdapter::new(id, redex).with_persistent(true),
    ))
}

/// Streaming temp-and-rename atomic writer: open `<out>.partial`, append
/// chunks as they arrive, then flush + rename over `<out>` on
/// [`Self::commit`]. A reader never observes a half-written target; on any
/// failure (the writer is dropped without committing) the `.partial` is
/// left in place for inspection — matching `fetch_dir`'s temp-and-rename
/// semantics and `TRANSFER.md` §5.
struct AtomicFileWriter {
    partial: PathBuf,
    out: PathBuf,
    file: tokio::fs::File,
}

impl AtomicFileWriter {
    /// Create the parent directory (if any) and open `<out>.partial`.
    async fn create(out: &std::path::Path) -> Result<Self, CliError> {
        let partial = partial_path(out);
        if let Some(parent) = out.parent() {
            if !parent.as_os_str().is_empty() {
                tokio::fs::create_dir_all(parent)
                    .await
                    .map_err(|e| generic(format!("create out dir {}: {e}", parent.display())))?;
            }
        }
        let file = tokio::fs::File::create(&partial)
            .await
            .map_err(|e| generic(format!("create {}: {e}", partial.display())))?;
        Ok(Self {
            partial,
            out: out.to_path_buf(),
            file,
        })
    }

    /// Append one chunk to the partial file.
    async fn write_chunk(&mut self, bytes: &[u8]) -> Result<(), CliError> {
        use tokio::io::AsyncWriteExt as _;
        self.file
            .write_all(bytes)
            .await
            .map_err(|e| generic(format!("write {}: {e}", self.partial.display())))
    }

    /// Flush the partial file, close it, and rename it over the
    /// destination. Closing before the rename keeps the commit portable
    /// (Windows refuses to rename a file that is still open).
    async fn commit(mut self) -> Result<(), CliError> {
        use tokio::io::AsyncWriteExt as _;
        self.file
            .flush()
            .await
            .map_err(|e| generic(format!("flush {}: {e}", self.partial.display())))?;
        drop(self.file);
        tokio::fs::rename(&self.partial, &self.out)
            .await
            .map_err(|e| {
                generic(format!(
                    "rename {} -> {}: {e} (partial left in place)",
                    self.partial.display(),
                    self.out.display()
                ))
            })
    }
}

/// `<out>.partial` sibling path for the atomic write.
fn partial_path(out: &std::path::Path) -> PathBuf {
    let mut name = out.file_name().unwrap_or_default().to_os_string();
    name.push(".partial");
    out.with_file_name(name)
}

/// MiB/s over the elapsed duration; `0.0` for a zero-length interval.
fn throughput_mib_s(bytes: u64, elapsed: std::time::Duration) -> f64 {
    let secs = elapsed.as_secs_f64();
    if secs <= 0.0 {
        0.0
    } else {
        (bytes as f64 / (1024.0 * 1024.0)) / secs
    }
}

/// Whether the recv-progress spinner should be drawn: only for a *human*
/// effective output format (`table` / `text`) and not under `--quiet`.
/// `--output json` (or any machine format) and `--quiet` both suppress it
/// so an operator asking for machine-readable or silent operation gets no
/// stderr chatter. The stderr-TTY check is applied separately in
/// [`Progress::start`] (pipes/tests get nothing regardless).
fn progress_enabled(output: Option<OutputFormat>, quiet: bool) -> bool {
    !quiet
        && matches!(
            OutputFormat::resolve_oneshot(output),
            OutputFormat::Table | OutputFormat::Text
        )
}

/// Thin wrapper over an indicatif spinner that no-ops unless `enabled`
/// (see [`progress_enabled`]) AND stderr is a TTY (tests, pipes get
/// nothing) — so the diagnostic never lands on stdout or clutters
/// machine-readable / quiet / non-interactive output.
struct Progress(Option<indicatif::ProgressBar>);

impl Progress {
    fn start(msg: &str, enabled: bool) -> Self {
        use std::io::IsTerminal as _;
        if !enabled || !std::io::stderr().is_terminal() {
            return Self(None);
        }
        let pb = indicatif::ProgressBar::new_spinner();
        pb.set_draw_target(indicatif::ProgressDrawTarget::stderr());
        pb.enable_steady_tick(std::time::Duration::from_millis(120));
        pb.set_message(msg.to_string());
        Self(Some(pb))
    }

    /// Determinate byte-progress bar for a transfer whose total size is
    /// known up front (a `Manifest` ref, or a `Small` ref that carries its
    /// size). Advance it with [`Self::inc`] as each chunk lands. Falls back
    /// to the same no-op-unless-TTY gating as [`Self::start`]; callers use
    /// [`Self::start`] (a spinner) when the total is unknown.
    fn start_bytes(msg: &str, total: u64, enabled: bool) -> Self {
        use std::io::IsTerminal as _;
        if !enabled || !std::io::stderr().is_terminal() {
            return Self(None);
        }
        let pb = indicatif::ProgressBar::new(total);
        pb.set_draw_target(indicatif::ProgressDrawTarget::stderr());
        pb.enable_steady_tick(std::time::Duration::from_millis(120));
        let style = indicatif::ProgressStyle::with_template(
            "{msg} [{bar:30}] {bytes}/{total_bytes} ({bytes_per_sec})",
        )
        .unwrap_or_else(|_| indicatif::ProgressStyle::default_bar())
        .progress_chars("=>-");
        pb.set_style(style);
        pb.set_message(msg.to_string());
        Self(Some(pb))
    }

    /// Advance a determinate bar by `n` bytes. No-op for a spinner or a
    /// disabled (`None`) progress handle.
    fn inc(&self, n: u64) {
        if let Some(pb) = &self.0 {
            pb.inc(n);
        }
    }

    /// Clear the spinner before emitting the success summary. Idempotent,
    /// and the [`Drop`] impl runs the same teardown — so the error path
    /// (an early `?` return before `finish`) still clears the spinner
    /// rather than leaving a stale frame above the error on stderr.
    fn finish(mut self) {
        self.clear();
    }

    fn clear(&mut self) {
        if let Some(pb) = self.0.take() {
            pb.finish_and_clear();
        }
    }
}

impl Drop for Progress {
    fn drop(&mut self) {
        self.clear();
    }
}

// ── output views ────────────────────────────────────────────────────

#[derive(Serialize)]
struct RecvBlobView {
    peer: u64,
    out: String,
    bytes: u64,
    duration_secs: f64,
    throughput_mib_s: f64,
}

#[derive(Serialize)]
struct SendBlobView {
    blob_ref: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    hash: Option<String>,
    size: u64,
    chunks: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    staged_to: Option<String>,
}

#[derive(Serialize)]
struct RecvDirView {
    peer: u64,
    out: String,
    files: u64,
    dirs: u64,
    symlinks: u64,
    bytes: u64,
    duration_secs: f64,
    throughput_mib_s: f64,
    /// Always true on success — `fetch_dir` only returns Ok after the
    /// atomic rename committed.
    atomic: bool,
}

#[derive(Serialize)]
struct SendDirView {
    remote_ref: String,
    manifest_size: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    staged_to: Option<String>,
}

#[derive(Serialize)]
struct LsView {
    transfer_count: u64,
    transfers: Vec<TransferRow>,
}

/// One in-flight (requester-side) transfer, rendered from a
/// [`transport::TransferStatus`]. `kind` is always `recv` — the engine
/// tracks fetches, not serving tasks.
#[derive(Serialize)]
struct TransferRow {
    /// Transfer stream id (the cancel handle).
    transfer_id: u64,
    /// Peer the bytes are being fetched from.
    peer: u64,
    /// Lowercase-hex BLAKE3 content address being fetched.
    hash: String,
    bytes_received: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    total_bytes: Option<u64>,
}

impl From<&transport::TransferStatus> for TransferRow {
    fn from(s: &transport::TransferStatus) -> Self {
        Self {
            transfer_id: s.stream_id,
            peer: s.holder,
            hash: hex::encode(s.expected_hash),
            bytes_received: s.bytes_received,
            total_bytes: s.total_bytes,
        }
    }
}

#[derive(Serialize)]
struct StatusView {
    transfer_id: u64,
    found: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    transfer: Option<TransferRow>,
}

#[derive(Serialize)]
struct CancelView {
    transfer_id: u64,
    cancelled: bool,
}

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

    #[test]
    fn parse_content_ref_accepts_a_bare_32_byte_hash() {
        let hex = "ab".repeat(32); // 64 hex chars = 32 bytes
        let parsed = parse_content_ref(&hex, "--blob-ref").expect("bare hash parses");
        match parsed {
            BlobRef::Small {
                hash, size, uri, ..
            } => {
                assert_eq!(hash, [0xabu8; 32]);
                // Size is unknown on the wire for the bare-hash form; the
                // Small fetch path keys on the hash and ignores it.
                assert_eq!(size, 0);
                assert_eq!(uri, format!("mesh://{hex}"));
            }
            other => panic!("expected Small, got {other:?}"),
        }
    }

    #[test]
    fn parse_content_ref_strips_a_single_0x_prefix() {
        let hex = "cd".repeat(32);
        let lower = parse_content_ref(&format!("0x{hex}"), "--blob-ref").expect("0x prefix");
        let upper = parse_content_ref(&format!("0X{hex}"), "--blob-ref").expect("0X prefix");
        let bare = parse_content_ref(&hex, "--blob-ref").expect("bare");
        // The URI is derived from the prefix-stripped body, so all three
        // forms must resolve to the identical reference.
        assert_eq!(lower, bare);
        assert_eq!(upper, bare);
    }

    #[test]
    fn parse_content_ref_does_not_peel_a_doubled_0x_prefix() {
        // `trim_start_matches` would have stripped both, turning a typo into
        // a silently-different (or accidentally-valid) ref. `strip_prefix`
        // peels one, leaving a non-hex `0x…` body that is rejected.
        let doubled = format!("0x0x{}", "ef".repeat(32));
        assert!(parse_content_ref(&doubled, "--blob-ref").is_err());
    }

    #[test]
    fn parse_content_ref_round_trips_a_full_encoded_ref() {
        // A full encoded BlobRef (longer than 32 bytes) takes the decode
        // path and must reconstruct the original verbatim.
        let original = BlobRef::small("mesh://x".to_string(), [7u8; 32], 123);
        let encoded = hex::encode(original.encode());
        let parsed = parse_content_ref(&encoded, "--remote-ref").expect("full ref decodes");
        assert_eq!(parsed, original);
    }

    #[test]
    fn parse_content_ref_rejects_non_hex_and_wrong_length() {
        assert!(parse_content_ref("not-hex", "--blob-ref").is_err());
        // 16 bytes: valid hex, but neither a 32-byte hash nor a decodable ref.
        assert!(parse_content_ref(&"ab".repeat(16), "--blob-ref").is_err());
    }

    #[test]
    fn relay_hint_only_fires_for_a_relayed_fetch() {
        // Fetching directly from the attach node: no relay, no hint.
        assert_eq!(relay_hint(42, 42), "");
        // `--from` names a different peer: hint names both ends so the
        // operator knows which node needs the route.
        let hint = relay_hint(7, 42);
        assert!(hint.contains('7') && hint.contains("42"), "hint: {hint}");
    }
}