radicle-lfs-transfer/src/main.rs

mod fetch_worker;
mod kubo;

use anyhow::Context as _;
use serde_json::{Value, json};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::Command;

use fetch_worker::{FetchError, FetchWorker};
use kubo::Kubo;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let kubo = Kubo::new(
        std::env::var("KUBO_API_URL").unwrap_or_else(|_| "http://127.0.0.1:5001".to_string()),
    );

    let stdin = tokio::io::stdin();
    let mut lines = BufReader::new(stdin).lines();
    let mut stdout = tokio::io::stdout();

    // Lazily spawned on the first "download" event, kept alive for the
    // rest of the session -- see `fetch_worker`'s doc comment for why.
    let mut fetch_worker: Option<FetchWorker> = None;

    while let Some(line) = lines.next_line().await? {
        if line.trim().is_empty() {
            continue;
        }
        let msg: Value = serde_json::from_str(&line)?;
        let event = msg.get("event").and_then(Value::as_str).unwrap_or("");

        match event {
            "init" => {
                let resp = match kubo.healthcheck().await {
                    Ok(()) => json!({}),
                    Err(e) => json!({"error": {"code": 1, "message": format!("{e:#}")}}),
                };
                write_line(&mut stdout, &resp).await?;
            }
            "upload" => {
                let resp = handle_upload(&msg).await;
                write_line(&mut stdout, &resp).await?;
            }
            "download" => {
                let resp = handle_download(&msg, &mut fetch_worker).await;
                write_line(&mut stdout, &resp).await?;
            }
            "terminate" => break,
            other => {
                write_line(
                    &mut stdout,
                    &json!({"error": {"code": 2, "message": format!("unknown event: {other}")}}),
                )
                .await?;
            }
        }
    }

    if let Some(worker) = fetch_worker {
        worker.shutdown().await;
    }

    Ok(())
}

async fn handle_upload(msg: &Value) -> Value {
    match upload_inner(msg).await {
        Ok(oid) => json!({"event": "complete", "oid": oid}),
        Err((oid, e)) => json!({
            "event": "complete",
            "oid": oid,
            "error": {"code": 3, "message": format!("{e:#}")},
        }),
    }
}

/// Delegates entirely to `rad lfs store`, which owns add/pin/note-write
/// (and encrypts first, for private repos) on the Radicle side. It's
/// authoritative and idempotent, so there's no separate "already pinned by
/// the pre-commit hook" fast path to maintain here anymore.
async fn upload_inner(msg: &Value) -> Result<String, (String, anyhow::Error)> {
    let oid = field_str(msg, "oid").map_err(|e| (String::new(), e))?;
    let size = field_i64(msg, "size").map_err(|e| (oid.clone(), e))?;
    let path = field_str(msg, "path").map_err(|e| (oid.clone(), e))?;

    let output = Command::new("rad")
        .args(["lfs", "store", "--oid", &oid, "--size", &size.to_string(), "--", &path])
        .output()
        .await
        .map_err(|e| {
            (
                oid.clone(),
                anyhow::Error::from(e).context("failed to spawn `rad lfs store`"),
            )
        })?;

    if !output.status.success() {
        return Err((
            oid,
            anyhow::anyhow!("rad lfs store failed: {}", String::from_utf8_lossy(&output.stderr)),
        ));
    }

    // `rad lfs store` prints only the resulting CID on success. The custom-transfer
    // protocol's "complete" response must echo back the requested oid (not the CID) so
    // git-lfs can match it to the in-flight transfer, but we still sanity-check that a
    // CID actually came back rather than trusting the exit code alone.
    let cid = String::from_utf8_lossy(&output.stdout).trim().to_string();
    if cid.is_empty() {
        return Err((oid, anyhow::anyhow!("rad lfs store exited successfully but printed no CID")));
    }

    Ok(oid)
}

async fn handle_download(msg: &Value, worker: &mut Option<FetchWorker>) -> Value {
    match download_inner(msg, worker).await {
        Ok((oid, path)) => json!({"event": "complete", "oid": oid, "path": path}),
        Err((oid, e)) => json!({
            "event": "complete",
            "oid": oid,
            "error": {"code": 3, "message": format!("{e:#}")},
        }),
    }
}

/// Delegates to the long-lived `rad lfs fetch-batch` worker (see
/// `fetch_worker`), which looks up the note, fetches from IPFS, and
/// decrypts (if needed) on the Radicle side -- kept alive across every
/// download in this session rather than re-spawned per object, so a
/// private repo's passphrase is only prompted for once per session.
async fn download_inner(
    msg: &Value,
    worker: &mut Option<FetchWorker>,
) -> Result<(String, String), (String, anyhow::Error)> {
    let oid = field_str(msg, "oid").map_err(|e| (String::new(), e))?;
    let size = field_i64(msg, "size").map_err(|e| (oid.clone(), e))?;

    // The path we report back must be on the same filesystem as
    // `.git/lfs/objects`, since git-lfs does an atomic `rename()` from
    // there into its object store rather than a copy. The system temp
    // dir (`/tmp`) can be a different filesystem/mount than the repo --
    // on such systems that rename fails outright with "invalid
    // cross-device link" rather than falling back to a copy. Using
    // `.git/lfs/tmp/` (git-lfs's own existing convention for its own
    // temp downloads) guarantees same-filesystem semantics by
    // construction.
    let tmp_dir = git_lfs_tmp_dir().await.map_err(|e| (oid.clone(), e))?;
    let tmp_path = tmp_dir.join(format!("rad-lfs-{oid}"));
    let out = tmp_path.to_string_lossy().to_string();

    if worker.is_none() {
        *worker = Some(
            FetchWorker::spawn()
                .await
                .map_err(|e| (oid.clone(), e.context("failed to start `rad lfs fetch-batch`")))?,
        );
    }

    // One respawn-and-retry if the worker process itself died (broken
    // pipe, unexpected exit) -- a per-object failure (wrong passphrase,
    // no note, decryption failure) is not retried, since a fresh worker
    // wouldn't change that outcome.
    match worker.as_mut().expect("just populated above").fetch(&oid, size, &out).await {
        Ok(()) => Ok((oid, out)),
        Err(FetchError::ObjectFailed(message)) => Err((oid, anyhow::anyhow!(message))),
        Err(FetchError::WorkerDied(err)) => {
            worker.take();
            let mut fresh = FetchWorker::spawn()
                .await
                .map_err(|e| (oid.clone(), e.context("failed to restart `rad lfs fetch-batch`")))?;
            let result = fresh.fetch(&oid, size, &out).await;
            *worker = Some(fresh);
            match result {
                Ok(()) => Ok((oid, out)),
                Err(FetchError::ObjectFailed(message)) => Err((oid, anyhow::anyhow!(message))),
                Err(FetchError::WorkerDied(retry_err)) => Err((
                    oid,
                    retry_err.context(format!("`rad lfs fetch-batch` worker died twice in a row (first: {err:#})")),
                )),
            }
        }
    }
}

/// Resolves (and creates, if needed) `.git/lfs/tmp` for the current
/// repository, regardless of whether the working directory is the repo
/// root or a subdirectory of it.
async fn git_lfs_tmp_dir() -> anyhow::Result<std::path::PathBuf> {
    let output = Command::new("git")
        .args(["rev-parse", "--git-dir"])
        .output()
        .await
        .context("failed to spawn `git rev-parse --git-dir`")?;
    if !output.status.success() {
        anyhow::bail!(
            "git rev-parse --git-dir failed: {}",
            String::from_utf8_lossy(&output.stderr)
        );
    }
    let git_dir = String::from_utf8_lossy(&output.stdout).trim().to_string();
    let tmp_dir = std::path::PathBuf::from(git_dir).join("lfs").join("tmp");
    tokio::fs::create_dir_all(&tmp_dir)
        .await
        .with_context(|| format!("failed to create `{}`", tmp_dir.display()))?;
    Ok(tmp_dir)
}

fn field_str(msg: &Value, key: &str) -> anyhow::Result<String> {
    msg.get(key)
        .and_then(Value::as_str)
        .map(String::from)
        .ok_or_else(|| anyhow::anyhow!("missing/invalid field: {key}"))
}

fn field_i64(msg: &Value, key: &str) -> anyhow::Result<i64> {
    msg.get(key)
        .and_then(Value::as_i64)
        .ok_or_else(|| anyhow::anyhow!("missing/invalid field: {key}"))
}

async fn write_line(stdout: &mut tokio::io::Stdout, value: &Value) -> anyhow::Result<()> {
    let mut s = serde_json::to_string(value)?;
    s.push('\n');
    stdout.write_all(s.as_bytes()).await?;
    stdout.flush().await?;
    Ok(())
}