varde/varde-ctl/src/main.rs

//! varde-ctl: CLI client for the varde daemon.
//!
//! Doubles as the reference implementation of the socket protocol: every
//! subcommand maps 1:1 onto a [`varde_proto::Request`].

use std::path::PathBuf;

use anyhow::{bail, Context, Result};
use clap::{Parser, Subcommand, ValueEnum};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixStream;
use varde_proto::{Envelope, MaterializeMode, PinPolicy, Request, ResponseData};

#[derive(Parser)]
#[command(name = "varde-ctl", version, about = "Control the varde daemon")]
struct Cli {
    /// Socket path (default: /run/varde/varde.sock as root,
    /// $XDG_RUNTIME_DIR/varde.sock otherwise; env: VARDE_SOCKET)
    #[arg(long, global = true, env = "VARDE_SOCKET")]
    socket: Option<PathBuf>,

    /// Print raw JSON responses instead of human-readable output
    #[arg(long, global = true)]
    json: bool,

    #[command(subcommand)]
    command: Command,
}

#[derive(Subcommand)]
enum Command {
    /// Import a file or directory into the store
    Add {
        /// Path to import (made absolute before sending)
        path: PathBuf,
        /// Recurse into directories
        #[arg(short, long)]
        recursive: bool,
    },
    /// Pin a hash: keep it, fetch it if absent, serve it per policy
    Pin {
        /// BLAKE3 hash (hex)
        hash: String,
        /// Serve to any LAN peer, not just trusted ones
        #[arg(long)]
        open_lan: bool,
    },
    /// Remove a pin (data stays until `gc`)
    Unpin {
        /// BLAKE3 hash (hex)
        hash: String,
    },
    /// Export content from the store to a path
    Materialize {
        /// BLAKE3 hash (hex)
        hash: String,
        /// Destination path (made absolute before sending)
        dest: PathBuf,
        /// Export mode
        #[arg(long, value_enum, default_value_t = CliMode::ReflinkOrCopy)]
        mode: CliMode,
    },
    /// Show daemon status, or status of one hash
    Status {
        /// BLAKE3 hash (hex)
        hash: Option<String>,
    },
    /// List pins
    List,
    /// Ticket operations (out-of-band sharing)
    #[command(subcommand)]
    Ticket(TicketCommand),
    /// Peer trust operations
    #[command(subcommand)]
    Peer(PeerCommand),
    /// Push fully-present content to a trusted peer
    Push {
        /// BLAKE3 hash (hex)
        hash: String,
        /// The receiving peer's iroh NodeId (z-base-32); must be trusted
        /// on both sides
        node_id: String,
    },
    /// Drop all unpinned blobs
    Gc,
    /// Stream daemon events to stdout (one JSON object per line)
    Subscribe,
}

#[derive(Subcommand)]
enum TicketCommand {
    /// Export a ticket for a hash
    Export {
        /// BLAKE3 hash (hex)
        hash: String,
    },
    /// Import a ticket: pin and fetch its content
    Import {
        /// Ticket string from `ticket export`
        ticket: String,
        /// Serve to any LAN peer, not just trusted ones
        #[arg(long)]
        open_lan: bool,
    },
}

#[derive(Subcommand)]
enum PeerCommand {
    /// Trust a peer NodeId
    Trust {
        /// iroh NodeId (z-base-32)
        node_id: String,
    },
    /// Revoke trust in a peer NodeId
    Untrust {
        /// iroh NodeId (z-base-32)
        node_id: String,
    },
    /// List trusted peers
    List,
}

#[derive(Clone, Copy, ValueEnum)]
enum CliMode {
    ReflinkOrCopy,
    Copy,
}

impl From<CliMode> for MaterializeMode {
    fn from(m: CliMode) -> Self {
        match m {
            CliMode::ReflinkOrCopy => MaterializeMode::ReflinkOrCopy,
            CliMode::Copy => MaterializeMode::Copy,
        }
    }
}

fn default_socket() -> PathBuf {
    // Mirrors the daemon's mode detection: root talks to the system
    // daemon, everyone else to their user daemon.
    if is_root() {
        PathBuf::from("/run/varde/varde.sock")
    } else if let Some(dir) = std::env::var_os("XDG_RUNTIME_DIR") {
        PathBuf::from(dir).join("varde.sock")
    } else {
        PathBuf::from("/run/varde/varde.sock")
    }
}

fn is_root() -> bool {
    // SAFETY: geteuid has no preconditions and cannot fail.
    unsafe { geteuid() == 0 }
}

extern "C" {
    fn geteuid() -> u32;
}

fn absolute(path: PathBuf) -> Result<String> {
    let abs =
        std::path::absolute(&path).with_context(|| format!("resolving path {}", path.display()))?;
    Ok(abs.display().to_string())
}

fn to_request(command: Command) -> Result<Request> {
    Ok(match command {
        Command::Add { path, recursive } => Request::Add {
            path: absolute(path)?,
            recursive,
        },
        Command::Pin { hash, open_lan } => Request::Pin {
            hash,
            policy: PinPolicy { open_lan },
        },
        Command::Unpin { hash } => Request::Unpin { hash },
        Command::Materialize { hash, dest, mode } => Request::Materialize {
            hash,
            dest: absolute(dest)?,
            mode: mode.into(),
        },
        Command::Status { hash } => Request::Status { hash },
        Command::List => Request::List {},
        Command::Ticket(TicketCommand::Export { hash }) => Request::TicketExport { hash },
        Command::Ticket(TicketCommand::Import { ticket, open_lan }) => Request::TicketImport {
            ticket,
            pin_policy: PinPolicy { open_lan },
        },
        Command::Peer(PeerCommand::Trust { node_id }) => Request::PeerTrust { node_id },
        Command::Peer(PeerCommand::Untrust { node_id }) => Request::PeerUntrust { node_id },
        Command::Peer(PeerCommand::List) => Request::PeerList {},
        Command::Push { hash, node_id } => Request::Push { hash, node_id },
        Command::Gc => Request::Gc {},
        Command::Subscribe => Request::Subscribe {},
    })
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<()> {
    let cli = Cli::parse();
    let socket = cli.socket.clone().unwrap_or_else(default_socket);
    let subscribe = matches!(cli.command, Command::Subscribe);
    let request = to_request(cli.command)?;

    let stream = UnixStream::connect(&socket)
        .await
        .with_context(|| format!("connecting to daemon at {}", socket.display()))?;
    let (read_half, mut write_half) = stream.into_split();
    let mut lines = BufReader::new(read_half).lines();

    let mut line = serde_json::to_string(&request)?;
    line.push('\n');
    write_half.write_all(line.as_bytes()).await?;

    let reply = lines
        .next_line()
        .await?
        .context("daemon closed the connection without replying")?;
    let envelope: Envelope =
        serde_json::from_str(&reply).context("daemon sent an unparseable reply")?;

    if !envelope.ok {
        let err = envelope.error.context("error reply without error body")?;
        bail!("{:?}: {}", err.code, err.message);
    }

    if cli.json {
        println!("{reply}");
    } else {
        print_human(envelope.data.as_ref());
    }

    if subscribe {
        // Connection is now an event stream; relay lines until it closes.
        while let Some(event) = lines.next_line().await? {
            println!("{event}");
        }
    }
    Ok(())
}

fn print_human(data: Option<&ResponseData>) {
    let Some(data) = data else {
        println!("ok");
        return;
    };
    match data {
        ResponseData::Added { hash, bytes } => println!("added {hash} ({bytes} bytes)"),
        ResponseData::Pinned { hash } => println!("pinned {hash}"),
        ResponseData::Bytes { data } => println!("{data}"),
        ResponseData::Done {} => println!("ok"),
        ResponseData::Materialized { bytes, reflinked } => {
            let how = if *reflinked { "reflinked" } else { "copied" };
            println!("materialized {bytes} bytes ({how})");
        }
        ResponseData::Status(s) => {
            println!("varde-daemon {} (protocol {})", s.version, s.protocol);
            if let Some(node_id) = &s.node_id {
                println!("node id:  {node_id}");
            }
            println!("store:    {}", s.store_path);
            println!("pins:     {}", s.pins);
            println!("peers:    {} connected", s.connected_peers);
        }
        ResponseData::HashStatus(h) => {
            println!("hash:     {}", h.hash);
            let total = h
                .total_bytes
                .map(|t| t.to_string())
                .unwrap_or_else(|| "?".to_string());
            println!("bytes:    {}/{total}", h.have_bytes);
            println!("complete: {}", h.complete);
            println!("pinned:   {}", h.pinned);
            if !h.active_peers.is_empty() {
                println!("active:   {}", h.active_peers.join(", "));
            }
        }
        ResponseData::Pins { pins } => {
            if pins.is_empty() {
                println!("no pins");
            }
            for pin in pins {
                let state = if pin.complete { "complete" } else { "partial" };
                let lan = if pin.policy.open_lan { " open-lan" } else { "" };
                println!("{} {state}{lan}", pin.hash);
            }
        }
        ResponseData::Ticket { ticket } => println!("{ticket}"),
        ResponseData::Peers { peers } => {
            if peers.is_empty() {
                println!("no trusted peers");
            }
            for peer in peers {
                let state = if peer.connected {
                    "connected"
                } else {
                    "offline"
                };
                println!("{} {state}", peer.node_id);
            }
        }
        ResponseData::GcDone { blobs_removed } => {
            println!("gc: removed {blobs_removed} blobs");
        }
        ResponseData::Pushed { hash, bytes } => {
            println!("pushed {hash} ({bytes} bytes)");
        }
    }
}