clawstor/claw-store/src/main.rs

mod actions;
mod cargo_init;
mod cluster;
mod config;
mod daemon;
mod head_watch;
mod hot;
mod manifest;
mod restore;
mod serve;
mod snapshot;
mod sync;
mod zfs;

use anyhow::{bail, Context, Result};
use clap::{Parser, Subcommand};
use config::Config;
use manifest::{Manifest, Project};
use std::net::SocketAddr;
use std::path::PathBuf;
use zfs::{SystemZfs, ZfsOps};

#[derive(Parser)]
#[command(name = "claw-store", about = "Fleet storage tier manager", version)]
struct Cli {
    #[arg(long, default_value = "/etc/claw-store/config.toml")]
    config: PathBuf,
    #[command(subcommand)]
    cmd: Cmd,
}

#[derive(Subcommand)]
enum Cmd {
    /// Register a project and wire up hot tier (format: org/repo)
    Activate {
        project: String,
        #[arg(long, help = "Clone from this URL if not present in warm tier")]
        clone: Option<String>,
    },
    /// Sync to peer then remove from hot tier (keeps warm clone)
    Deactivate { project: String },
    /// List all repos in warm tier with activation status
    List,
    /// Push commits to origin then notify peer to pull
    Sync { project: String },
    /// Pull latest from origin into warm tier (also invoked by peer)
    Pull { project: String },
    /// Run the background daemon (snapshot cron, GC, replication, sync retry)
    Daemon,
    /// Show tier usage, active projects, recent snapshots
    Status,
    /// Run hot tier GC manually
    Gc,
    /// Take a ZFS snapshot of the warm tier now
    Snapshot,
    /// List available snapshots
    ListSnapshots { project: String },
    /// Restore a project from a snapshot
    Restore { project: String, snapshot: String },
    /// Trigger nightly replication to cold tier now
    Replicate,
    /// Start the HTTP API server for the dashboard
    Serve {
        #[arg(long, default_value = "7700")]
        port: u16,
        #[arg(long)]
        static_dir: Option<PathBuf>,
    },
    /// Mark a project as pinned — survives every GC pass (stale + LRU)
    Pin { project: String },
    /// Unmark a project — it's now a normal GC candidate again
    Unpin { project: String },
    /// Probe a cluster peer's LAN and Tailscale endpoints; print the winning route
    ClusterProbe {
        /// Peer name as listed in `[[cluster.peers]]` in the config
        peer: String,
    },
    /// Bootstrap gossip, wait for convergence, print the peer membership table
    ClusterStatus {
        /// This node's name as advertised to peers.
        #[arg(long)]
        name: String,
        /// How long to wait for convergence before printing.
        #[arg(long, default_value = "3")]
        wait_secs: u64,
    },
    /// Ping a peer over the QUIC RPC layer with an ephemeral fleet CA.
    /// Demonstrates the mTLS handshake end-to-end against a live peer.
    /// The `--rpc-addr` must match the peer's advertised RPC endpoint
    /// (gossip port + 1 by default).
    ///
    /// If `--tls-dir` is set, loads a persistent NodeIdentity from that
    /// directory (produced by `fleet-ca sign`) — the peer must have
    /// signed against the same CA.
    ClusterPing {
        /// This node's name — used as the SAN in the ephemeral leaf cert
        /// when `--tls-dir` is not set.
        #[arg(long)]
        name: String,
        /// Peer's node name — used as the expected server name (SAN) on
        /// the incoming cert.
        #[arg(long)]
        peer: String,
        /// Peer's RPC socket. Typically gossip_port + 1.
        #[arg(long)]
        rpc_addr: SocketAddr,
        /// Payload to send. Echoed back with a "pong:" prefix.
        #[arg(long, default_value = "hello")]
        payload: String,
        /// Optional: dir containing `ca.crt`, `node.crt`, `node.key`
        /// (from `fleet-ca sign --out-dir <dir>`). When absent, falls
        /// back to an ephemeral in-memory CA.
        #[arg(long)]
        tls_dir: Option<PathBuf>,
    },
    /// Generate a fresh fleet root CA and write it to disk. Run once per
    /// cluster, on the primary node. Every other node needs only the
    /// public `ca.crt` — but this dir also gets `ca.key`, which must
    /// stay on the signing host.
    FleetCaInit {
        /// Directory to write `ca.crt` + `ca.key` into.
        #[arg(long)]
        dir: PathBuf,
        /// Common Name embedded in the CA's distinguished name.
        #[arg(long, default_value = "clawstor fleet CA")]
        cn: String,
    },
    /// Sign a leaf certificate for a node under the fleet CA. Outputs
    /// `ca.crt` (public), `node.crt` (public), `node.key` (private,
    /// 0o600) into `--out-dir`. Copy those three files to the target
    /// node and point `[cluster.tls]` at them.
    FleetCaSign {
        /// Directory holding the CA (`ca.crt` + `ca.key`) — the same
        /// dir passed to `fleet-ca init`.
        #[arg(long)]
        ca_dir: PathBuf,
        /// Node name — becomes the SAN + Common Name on the leaf cert.
        #[arg(long)]
        node: String,
        /// Output directory for the three per-node PEM files.
        #[arg(long)]
        out_dir: PathBuf,
    },
    /// Call the peer's PeerStatus RPC and print its local view of the
    /// cluster. Requires that peer be running a daemon with
    /// `[cluster.tls]` configured, and that our `--tls-dir` was signed
    /// by the same CA as the peer's.
    ClusterPeerStatus {
        /// Peer's node name — must match the peer's cert SAN.
        #[arg(long)]
        peer: String,
        /// Peer's RPC socket. Typically gossip_port + 1.
        #[arg(long)]
        rpc_addr: SocketAddr,
        /// Directory holding this node's mTLS material
        /// (`ca.crt` + `node.crt` + `node.key` from `fleet-ca sign`).
        #[arg(long)]
        tls_dir: PathBuf,
    },
}

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(std::env::var("RUST_LOG").unwrap_or_else(|_| "info".into()))
        .init();

    let cli = Cli::parse();

    // Config-independent commands: filesystem-only, run before we touch
    // /etc/claw-store/config.toml. Lets an operator bootstrap the CA on
    // a fresh box that doesn't have a config yet.
    match &cli.cmd {
        Cmd::FleetCaInit { dir, cn } => return cmd_fleet_ca_init(dir, cn),
        Cmd::FleetCaSign {
            ca_dir,
            node,
            out_dir,
        } => return cmd_fleet_ca_sign(ca_dir, node, out_dir),
        _ => {}
    }

    let cfg = Config::load(&cli.config)
        .with_context(|| format!("loading config from {}", cli.config.display()))?;
    let manifest_path = Manifest::default_path();
    let mut manifest = Manifest::load(&manifest_path)?;
    let zfs = SystemZfs;

    match cli.cmd {
        Cmd::Activate { project, clone } =>
            cmd_activate(&cfg, &mut manifest, &manifest_path, &project, clone.as_deref())?,
        Cmd::Deactivate { project } =>
            cmd_deactivate(&cfg, &mut manifest, &manifest_path, &project)?,
        Cmd::List => cmd_list(&cfg, &manifest)?,
        Cmd::Sync { project } => cmd_sync(&cfg, &manifest, &project)?,
        Cmd::Pull { project } => cmd_pull(&cfg, &mut manifest, &manifest_path, &project)?,
        Cmd::Daemon => daemon::run(cfg, manifest).await?,
        Cmd::Status => cmd_status(&cfg, &manifest, &zfs)?,
        Cmd::Gc => cmd_gc(&cfg, &mut manifest)?,
        Cmd::Snapshot => cmd_snapshot(&cfg, &zfs)?,
        Cmd::ListSnapshots { project } => cmd_list_snapshots(&cfg, &zfs, &project)?,
        Cmd::Restore { project, snapshot } => cmd_restore(&cfg, &zfs, &project, &snapshot)?,
        Cmd::Replicate => cmd_replicate(&cfg, &zfs)?,
        Cmd::Serve { port, static_dir } =>
            serve::run_server(cfg, manifest, port, static_dir).await?,
        Cmd::Pin { project } => cmd_set_pin(&manifest_path, &project, true)?,
        Cmd::Unpin { project } => cmd_set_pin(&manifest_path, &project, false)?,
        Cmd::ClusterProbe { peer } => cmd_cluster_probe(&cfg, &peer).await?,
        Cmd::ClusterStatus { name, wait_secs } => cmd_cluster_status(&cfg, &name, wait_secs).await?,
        Cmd::ClusterPing {
            name,
            peer,
            rpc_addr,
            payload,
            tls_dir,
        } => cmd_cluster_ping(&name, &peer, rpc_addr, &payload, tls_dir.as_deref()).await?,
        Cmd::ClusterPeerStatus {
            peer,
            rpc_addr,
            tls_dir,
        } => cmd_cluster_peer_status(&peer, rpc_addr, &tls_dir).await?,
        Cmd::FleetCaInit { .. } | Cmd::FleetCaSign { .. } => {
            // Handled by the config-independent short-circuit above.
            unreachable!("fleet-ca commands short-circuit before config load");
        }
    }
    Ok(())
}

// ── cluster peer-status ───────────────────────────────────────────────────────

async fn cmd_cluster_peer_status(
    peer: &str,
    rpc_addr: SocketAddr,
    tls_dir: &std::path::Path,
) -> Result<()> {
    use cluster::rpc::call_peer_status;
    use cluster::transport::{NodeIdentity, QuicClient};

    let identity = NodeIdentity::from_pem_dir(tls_dir)
        .with_context(|| format!("loading identity from {}", tls_dir.display()))?;
    let client = QuicClient::new("0.0.0.0:0".parse()?, identity)?;
    let conn = client.connect(rpc_addr, peer).await?;
    let status = call_peer_status(&conn).await?;

    println!(
        "peer:  {} (zone: {})",
        status.local_name, status.local_zone
    );
    println!();
    if status.peers.is_empty() {
        println!("  (no peers known)");
    } else {
        println!(
            "  {:<20} {:<14} {:<8} {:<22} {:<20}",
            "NAME", "ZONE", "STATE", "RPC LAN", "HOT USED / MAX"
        );
        println!("  {}", "-".repeat(90));
        for p in &status.peers {
            let state = if p.alive { "alive" } else { "dead" };
            let hot = match (p.hot_used_bytes, p.hot_max_bytes) {
                (Some(u), Some(m)) => format!("{u} / {m}"),
                (Some(u), None) => format!("{u} / -"),
                _ => "-".into(),
            };
            println!(
                "  {:<20} {:<14} {:<8} {:<22} {:<20}",
                p.name,
                p.zone,
                state,
                p.rpc_lan
                    .map(|a| a.to_string())
                    .unwrap_or_else(|| "-".into()),
                hot,
            );
        }
    }

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    Ok(())
}

// ── fleet-ca init / sign ─────────────────────────────────────────────────────

fn cmd_fleet_ca_init(dir: &std::path::Path, cn: &str) -> Result<()> {
    use cluster::transport::FleetCa;
    if dir.join("ca.crt").exists() {
        bail!(
            "refusing to overwrite existing CA at {}: delete it first if you really want a new one",
            dir.display()
        );
    }
    let ca = FleetCa::generate(cn).context("generating fleet CA")?;
    ca.save(dir).context("saving fleet CA to disk")?;
    println!("fleet CA written:");
    println!("  {}", dir.join("ca.crt").display());
    println!("  {}  (chmod 0600; keep on primary only)", dir.join("ca.key").display());
    println!();
    println!("Sign a per-node identity with:");
    println!(
        "  claw-store fleet-ca-sign --ca-dir {} --node <name> --out-dir <dir>",
        dir.display()
    );
    Ok(())
}

fn cmd_fleet_ca_sign(
    ca_dir: &std::path::Path,
    node: &str,
    out_dir: &std::path::Path,
) -> Result<()> {
    use cluster::transport::FleetCa;
    let ca = FleetCa::load(ca_dir).context("loading fleet CA")?;
    ca.sign_leaf_to_pem(node, out_dir)
        .context("signing + writing per-node PEMs")?;
    println!("signed leaf for {node}:");
    println!("  {}", out_dir.join("ca.crt").display());
    println!("  {}", out_dir.join("node.crt").display());
    println!("  {}  (chmod 0600; distribute securely)", out_dir.join("node.key").display());
    println!();
    println!("On {node}, point [cluster.tls] in the config at those three paths.");
    Ok(())
}

// ── cluster ping ─────────────────────────────────────────────────────────────

/// Round-trip a `ping` payload to `peer` over the QUIC RPC transport
/// using an ephemeral fleet CA. Both this node and the peer must have
/// invoked `cluster-ping` with a shared CA — the current implementation
/// generates one CA per invocation via `NodeIdentity::generate_test_pair`,
/// meaning peers can only talk to each other when they share the same
/// invocation (typical dev-loopback usage). Production identity loading
/// from persistent PEM files lands in Phase 1d.
async fn cmd_cluster_ping(
    name: &str,
    peer: &str,
    rpc_addr: SocketAddr,
    payload: &str,
    tls_dir: Option<&std::path::Path>,
) -> Result<()> {
    use cluster::transport::{ping, NodeIdentity, QuicClient};

    // Two identity paths:
    //   1. `--tls-dir` present → load persisted PEM. The peer must have
    //      been signed by the same CA (via `fleet-ca sign`); otherwise
    //      the TLS handshake fails.
    //   2. Absent → ephemeral CA. Only useful when both endpoints run
    //      in the same process (dev/loopback demo).
    let id_self = match tls_dir {
        Some(dir) => NodeIdentity::from_pem_dir(dir)
            .with_context(|| format!("loading node identity from {}", dir.display()))?,
        None => {
            let (id_self, _id_peer) = NodeIdentity::generate_test_pair(name, peer)?;
            id_self
        }
    };

    let client = QuicClient::new("0.0.0.0:0".parse()?, id_self)?;
    let conn = client.connect(rpc_addr, peer).await?;
    let response = ping(&conn, payload.as_bytes()).await?;
    println!("→ sent: {}", payload);
    println!("← recv: {}", String::from_utf8_lossy(&response));
    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    Ok(())
}

// ── cluster status ───────────────────────────────────────────────────────────

async fn cmd_cluster_status(cfg: &Config, name: &str, wait_secs: u64) -> Result<()> {
    let cluster_cfg = cfg
        .cluster
        .as_ref()
        .context("no [cluster] section in config; cannot start gossip")?;
    let gossip = cluster::gossip::ClusterGossip::bootstrap(cluster_cfg, name).await?;
    println!("gossip bootstrapped as {}; waiting {}s for convergence...", name, wait_secs);
    tokio::time::sleep(std::time::Duration::from_secs(wait_secs)).await;

    let mut peers = gossip.peers().await;
    peers.sort_by(|a, b| a.name.cmp(&b.name));

    println!();
    println!(
        "{:<20} {:<14} {:<8} {:<22} {:<22} {:<20}",
        "PEER", "ZONE", "STATE", "RPC LAN", "RPC TAILSCALE", "HOT USED / MAX"
    );
    println!("{}", "-".repeat(110));
    if peers.is_empty() {
        println!("(no peers yet — either alone in the fleet or gossip still converging)");
    } else {
        for p in &peers {
            let state = if p.alive { "alive" } else { "dead" };
            let hot = match (p.hot_used_bytes, p.hot_max_bytes) {
                (Some(u), Some(m)) => format!("{u} / {m}"),
                (Some(u), None) => format!("{u} / -"),
                _ => "-".into(),
            };
            println!(
                "{:<20} {:<14} {:<8} {:<22} {:<22} {:<20}",
                p.name,
                p.zone,
                state,
                p.rpc_lan.map(|a| a.to_string()).unwrap_or_else(|| "-".into()),
                p.rpc_tailscale.map(|a| a.to_string()).unwrap_or_else(|| "-".into()),
                hot,
            );
        }
    }
    println!();
    gossip.shutdown();
    Ok(())
}

// ── cluster probe ────────────────────────────────────────────────────────────

async fn cmd_cluster_probe(cfg: &Config, peer_name: &str) -> Result<()> {
    let cluster_cfg = cfg
        .cluster
        .as_ref()
        .context("no [cluster] section in config; cannot probe peers")?;
    let peer = cluster_cfg
        .peer(peer_name)
        .with_context(|| format!("peer {peer_name:?} not found in cluster.peers"))?;
    let probe = cluster::LanFirstProbe::new();
    let winner = probe.probe(peer).await?;
    println!("peer:    {}", peer.name);
    println!("zone:    {}", peer.zone);
    println!("route:   {}", winner.kind.as_str());
    println!("addr:    {}", winner.addr);
    println!("elapsed: {:?}", winner.elapsed);
    Ok(())
}

// ── pin / unpin ──────────────────────────────────────────────────────────────

fn cmd_set_pin(manifest_path: &std::path::Path, project: &str, pinned: bool) -> Result<()> {
    let m = Manifest::update(manifest_path, |m| {
        match m.get_mut(project) {
            Some(p) => {
                p.pinned = pinned;
                Ok(())
            }
            None => bail!(
                "no such project in manifest: {project}\n\
                 (only activated projects can be pinned; run `claw-store activate {project}` first)"
            ),
        }
    })?;
    let _ = m; // suppress unused — we just want the side effect
    println!(
        "{}: {}",
        if pinned { "pinned" } else { "unpinned" },
        project
    );
    Ok(())
}

// ── activate ─────────────────────────────────────────────────────────────────

fn cmd_activate(
    cfg: &Config,
    manifest: &mut Manifest,
    manifest_path: &std::path::Path,
    project: &str,
    clone_url: Option<&str>,
) -> Result<()> {
    let warm = warm_path(cfg, project);
    let hot_target = hot_path(cfg, project);

    if !warm.exists() {
        match clone_url {
            Some(url) => {
                println!("Cloning {} → {}", url, warm.display());
                let warm_str = warm.to_str()
                    .with_context(|| format!("non-UTF-8 path: {}", warm.display()))?;
                let status = std::process::Command::new("git")
                    .args(["clone", url, warm_str])
                    .status()?;
                anyhow::ensure!(status.success(), "git clone failed");
            }
            None => bail!(
                "warm path does not exist: {}\nUse --clone <url> to clone it first.",
                warm.display()
            ),
        }
    }

    std::fs::create_dir_all(&hot_target)?;
    cargo_init::write_cargo_config(&warm, &hot_target)?;

    // Audit followup — use locked, atomic update so a concurrent
    // dashboard activation (POSTing to /api/activate which shells out
    // back to this same CLI) can't race the writer. `upsert` preserves
    // an existing row's `pinned` flag — re-activating a pinned project
    // doesn't silently unpin it.
    let new_row = Project {
        name: project.to_string(),
        warm_path: warm.clone(),
        hot_target_path: hot_target.clone(),
        last_build: None,
        last_active: None,
        last_sync: None,
        pinned: false,
    };
    Manifest::update(manifest_path, |m| {
        let preserved_pin = m.get(project).map(|p| p.pinned).unwrap_or(false);
        let mut row = new_row.clone();
        row.pinned = preserved_pin;
        m.upsert(row);
        Ok(())
    })?;
    // Keep the in-memory `manifest` in sync with disk so downstream
    // commands in the same process see the new row.
    *manifest = Manifest::load(manifest_path)?;

    println!("activated: {}", project);
    println!("  source  → {}", warm.display());
    println!("  target/ → {}", hot_target.display());
    Ok(())
}

// ── deactivate ────────────────────────────────────────────────────────────────

/// CLI-facing deactivate: thin wrapper around `actions::deactivate_project`
/// that adds println! feedback for the terminal. The daemon calls the
/// shared function directly with tracing::info! instead.
fn cmd_deactivate(
    cfg: &Config,
    manifest: &mut Manifest,
    manifest_path: &std::path::Path,
    project: &str,
) -> Result<()> {
    if cfg.peer.is_some() {
        println!("Syncing {} to peer before deactivate...", project);
    }
    let outcome = actions::deactivate_project(cfg, manifest_path, project)?;
    if let Some(err) = &outcome.sync_error {
        eprintln!("  warning: sync failed ({}), queued for retry", err);
    }
    // Refresh the caller's in-memory manifest to match the on-disk state.
    *manifest = Manifest::load(manifest_path)?;
    let warm = warm_path(cfg, project);
    let freed = if outcome.freed_bytes > 0 {
        format!(", freed {:.1} MB", outcome.freed_mb())
    } else {
        String::new()
    };
    println!("deactivated: {} (warm clone kept at {}{})", project, warm.display(), freed);
    Ok(())
}

// ── list ──────────────────────────────────────────────────────────────────────

fn cmd_list(cfg: &Config, manifest: &Manifest) -> Result<()> {
    let base = &cfg.warm.projects_path;
    println!("{:<45} {:<10} {}", "PROJECT", "STATUS", "LAST ACTIVE");
    println!("{}", "-".repeat(75));

    let mut rows: Vec<(String, bool, String)> = Vec::new();

    if let Ok(top_entries) = std::fs::read_dir(base) {
        for org_entry in top_entries.flatten() {
            let org_path = org_entry.path();
            if !org_path.is_dir() { continue; }
            let org = org_entry.file_name().to_string_lossy().to_string();

            if org_path.join(".git").exists() {
                // Top-level git repo (flat layout)
                let active = manifest.get(&org);
                let last = active
                    .and_then(|p| p.last_active)
                    .map(|t| t.format("%Y-%m-%d %H:%M").to_string())
                    .unwrap_or_else(|| "—".into());
                rows.push((org, active.is_some(), last));
            } else {
                // Org directory — scan for repos inside it
                if let Ok(repos) = std::fs::read_dir(&org_path) {
                    for repo_entry in repos.flatten() {
                        let repo_path = repo_entry.path();
                        if !repo_path.is_dir() || !repo_path.join(".git").exists() { continue; }
                        let repo = repo_entry.file_name().to_string_lossy().to_string();
                        let key = format!("{}/{}", org, repo);
                        let active = manifest.get(&key);
                        let last = active
                            .and_then(|p| p.last_active)
                            .map(|t| t.format("%Y-%m-%d %H:%M").to_string())
                            .unwrap_or_else(|| "—".into());
                        rows.push((key, active.is_some(), last));
                    }
                }
            }
        }
    }

    rows.sort_by(|a, b| a.0.cmp(&b.0));
    rows.dedup_by(|a, b| a.0 == b.0);

    let active_count = rows.iter().filter(|r| r.1).count();
    for (name, is_active, last) in &rows {
        let status = if *is_active { "active" } else { "warm" };
        println!("{:<45} {:<10} {}", name, status, last);
    }
    println!("\n{} repos ({} active, {} warm-only)", rows.len(), active_count, rows.len() - active_count);
    Ok(())
}

// ── sync ─────────────────────────────────────────────────────────────────────

fn cmd_sync(cfg: &Config, manifest: &Manifest, project: &str) -> Result<()> {
    let peer = cfg.peer.as_ref()
        .context("no [peer] configured — add peer.host and peer.user to config.toml")?;

    let warm = manifest.get(project)
        .map(|p| p.warm_path.clone())
        .unwrap_or_else(|| warm_path(cfg, project));

    println!("Pushing {}...", project);
    sync::sync_project(&warm, project, &peer.user, &peer.host)?;
    println!("  pushed to origin");
    println!("  notified {}@{} to pull", peer.user, peer.host);
    Ok(())
}

// ── pull ─────────────────────────────────────────────────────────────────────

fn cmd_pull(
    cfg: &Config,
    manifest: &mut Manifest,
    manifest_path: &std::path::Path,
    project: &str,
) -> Result<()> {
    let warm = manifest.get(project)
        .map(|p| p.warm_path.clone())
        .unwrap_or_else(|| warm_path(cfg, project));

    sync::pull_project(&warm)?;

    // Stamp last_sync on the manifest entry if it exists.
    Manifest::update(manifest_path, |m| {
        if let Some(p) = m.get_mut(project) {
            p.last_sync = Some(chrono::Utc::now());
        }
        Ok(())
    })?;
    *manifest = Manifest::load(manifest_path)?;
    println!("pulled: {}", project);
    Ok(())
}

// ── existing commands ─────────────────────────────────────────────────────────

fn cmd_status(cfg: &Config, manifest: &Manifest, zfs: &SystemZfs) -> Result<()> {
    println!("=== claw-store status — {} ===\n", cfg.node.name);
    println!("HOT  tier: {}", cfg.hot.path.display());
    let used = hot::total_used_gb(manifest).unwrap_or(0.0);
    println!("  used: {:.1} GB / {} GB max\n", used, cfg.hot.max_gb);
    println!("WARM tier: {}", cfg.warm.projects_path.display());
    let snaps = zfs.list_snapshots(&cfg.warm.zfs_dataset).unwrap_or_default();
    println!("  snapshots: {}\n", snaps.len());
    if let Some(cold) = &cfg.cold {
        println!("COLD tier: {}\n", cold.archive_path.display());
    }
    if let Some(peer) = &cfg.peer {
        println!("PEER: {}@{}\n", peer.user, peer.host);
    }
    let pinned_count = manifest.projects.iter().filter(|p| p.pinned).count();
    println!(
        "Active projects ({}{}):",
        manifest.projects.len(),
        if pinned_count > 0 { format!(", {pinned_count} pinned") } else { String::new() },
    );
    for p in &manifest.projects {
        let active = p.last_active
            .map(|t| t.format("%Y-%m-%d %H:%M").to_string())
            .unwrap_or_else(|| "never".into());
        let synced = p.last_sync
            .map(|t| t.format("%Y-%m-%d %H:%M").to_string())
            .unwrap_or_else(|| "never".into());
        let pin_marker = if p.pinned { " 📌" } else { "" };
        println!("  {}{} (active: {}, synced: {})", p.name, pin_marker, active, synced);
    }
    Ok(())
}

fn cmd_gc(cfg: &Config, manifest: &mut Manifest) -> Result<()> {
    let stale = hot::gc_stale_targets(manifest, cfg.hot.stale_hours)?;
    for name in &stale { println!("  evicted (stale): {}", name); }
    let space = hot::gc_by_space(manifest, cfg.hot.max_gb as f64)?;
    for name in &space { println!("  evicted (space): {}", name); }
    if stale.is_empty() && space.is_empty() {
        println!("Nothing to evict.");
        return Ok(());
    }
    // Persist via the locked atomic path — `manifest.save` is now
    // flock+rename internally, so this is safe under concurrent writers.
    manifest.save(&Manifest::default_path())?;
    Ok(())
}

fn cmd_snapshot(cfg: &Config, zfs: &SystemZfs) -> Result<()> {
    let ts = chrono::Utc::now().format("%Y-%m-%d-%H%M").to_string();
    snapshot::run_snapshot_cycle(
        zfs, &cfg.warm.zfs_dataset, &ts,
        cfg.warm.snapshot_retain_hours as usize,
        cfg.warm.snapshot_retain_days as usize,
        cfg.warm.snapshot_retain_weeks as usize,
    )?;
    println!("Snapshot {} taken.", ts);
    Ok(())
}

fn cmd_list_snapshots(cfg: &Config, zfs: &SystemZfs, _project: &str) -> Result<()> {
    let snaps = zfs.list_snapshots(&cfg.warm.zfs_dataset)?;
    if snaps.is_empty() { println!("No snapshots found."); return Ok(()); }
    for s in &snaps { println!("  {}", s); }
    Ok(())
}

fn cmd_restore(cfg: &Config, zfs: &SystemZfs, project: &str, snap: &str) -> Result<()> {
    let full_snap = format!("{}@{}", cfg.warm.zfs_dataset, snap);
    let found = restore::find_snapshot(zfs, &cfg.warm.zfs_dataset, snap)?;
    anyhow::ensure!(found.is_some(), "snapshot '{}' not found", snap);
    let path = restore::restore_project(zfs, project, &full_snap, &cfg.warm.zfs_dataset)?;
    println!("Restored to: {}", path.display());
    println!("Cleanup: zfs destroy {}/{}-restore-{}", cfg.warm.zfs_dataset, project, snap);
    Ok(())
}

fn cmd_replicate(cfg: &Config, zfs: &SystemZfs) -> Result<()> {
    let rep = cfg.replication.as_ref()
        .context("no replication config — this node does not replicate")?;
    let host = rep.send_to_host.as_ref().context("send_to_host not set")?;
    let user = rep.send_to_user.as_ref().context("send_to_user not set")?;
    let dest = rep.cold_dataset_on_peer.as_ref().context("cold_dataset_on_peer not set")?;
    snapshot::replicate_to_cold(zfs, &cfg.warm.zfs_dataset, user, host, dest)?;
    println!("Replication to {}@{} complete.", user, host);
    Ok(())
}

// ── path helpers ──────────────────────────────────────────────────────────────

fn warm_path(cfg: &Config, project: &str) -> PathBuf {
    cfg.warm.projects_path.join(project)
}

fn hot_path(cfg: &Config, project: &str) -> PathBuf {
    cfg.hot.path.join(project)
}