Something went wrong. Try again.
⚘ use your pds as a git remote if you want to ⚘
Something went wrong. Try again.
11 kB · 332 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333//! Fetch/clone flow: download bundle chain from PDS and apply to local repo.//!//! Reads the remote state record, downloads bundles as blobs,//! reassembles chunked bundles, applies them via `git bundle unbundle`,//! and updates local refs to match the remote state.
use std::path::Path;
use crate::bundle::apply_bundle;use crate::chunk::reassemble_chunks;use crate::pds_client::PdsClient;use crate::types::{BundleEntry, COLLECTION, RepoState};
/// Result of a fetch or clone operation.#[derive(Debug)]pub enum FetchResult { /// bundles were downloaded and applied Applied { /// number of bundles applied bundles_applied: usize, /// total bytes downloaded bytes_downloaded: u64, }, /// local refs already match remote — nothing to do AlreadyUpToDate,}
/// Clones a repository from PDS into a local git repo.////// The repo at `repo_path` should already be initialized (bare or working)./// Downloads all bundles in the chain from oldest to newest, applies each,/// then updates local refs and checks out the default branch (if working tree).pub async fn clone_repo( client: &PdsClient, did: &str, repo_name: &str, repo_path: &Path,) -> Result<FetchResult, String> { // read remote state let state = read_remote_state(client, did, repo_name).await?;
if state.bundles.is_empty() { return Err("remote state has no bundles".to_string()); }
// download and apply all bundles in order (oldest first) let (applied, downloaded) = download_and_apply(client, did, repo_path, &state.bundles).await?;
// update local refs to match remote state update_local_refs(repo_path, &state).await?;
// checkout default branch if this is a working tree checkout_default_branch(repo_path, &state).await?;
Ok(FetchResult::Applied { bundles_applied: applied, bytes_downloaded: downloaded, })}
/// Fetches new commits from PDS into an existing local repo.////// Compares local refs against the remote state and downloads only/// bundles that contain new commits. Skips bundles whose tips are/// already present in the local repo.pub async fn fetch_repo( client: &PdsClient, did: &str, repo_name: &str, repo_path: &Path,) -> Result<FetchResult, String> { // read remote state let state = read_remote_state(client, did, repo_name).await?;
// find which bundles we still need let new_bundles = find_new_bundles(repo_path, &state.bundles).await?;
if new_bundles.is_empty() { return Ok(FetchResult::AlreadyUpToDate); }
// download and apply only the new bundles let (applied, downloaded) = download_and_apply(client, did, repo_path, &new_bundles).await?;
// update local refs to match remote state update_local_refs(repo_path, &state).await?;
// update the working tree if not a bare repo update_working_tree(repo_path).await?;
Ok(FetchResult::Applied { bundles_applied: applied, bytes_downloaded: downloaded, })}
/// Reads and parses the remote state record from PDS.pub async fn read_remote_state( client: &PdsClient, did: &str, repo_name: &str,) -> Result<RepoState, String> { read_remote_state_opt(client, did, repo_name) .await? .ok_or_else(|| format!("no state record found for {}", repo_name))}
/// Reads the remote state record, or `None` if the repo has no record yet.////// Only a genuinely absent record is `None`; an unreachable PDS or an/// unparseable record is an error. Callers that treat "nothing there" as an/// empty repo need that distinction — reporting a network failure as an empty/// repo tells git the backup is empty when it is merely unreachable.pub async fn read_remote_state_opt( client: &PdsClient, did: &str, repo_name: &str,) -> Result<Option<RepoState>, String> { let Some(record) = client.get_record(did, COLLECTION, repo_name).await? else { return Ok(None); };
serde_json::from_value(record.value) .map(Some) .map_err(|e| format!("failed to parse remote state: {}", e))}
/// Downloads and applies a slice of bundle entries to the local repo.////// Returns (bundles_applied, total_bytes_downloaded).pub async fn download_and_apply( client: &PdsClient, did: &str, repo_path: &Path, bundles: &[BundleEntry],) -> Result<(usize, u64), String> { let mut total_bytes: u64 = 0;
for (i, entry) in bundles.iter().enumerate() { tracing::info!( "downloading bundle {}/{} ({} part(s))", i + 1, bundles.len(), entry.parts.len() );
// download all parts of this bundle let bundle_data = download_bundle_parts(client, did, entry).await?; total_bytes += bundle_data.len() as u64;
// apply the bundle apply_bundle(repo_path, &bundle_data).await?; }
Ok((bundles.len(), total_bytes))}
/// Downloads and reassembles the parts of a single bundle entry.////// Most bundles have a single part. Chunked bundles (>40MB) have/// multiple parts that are concatenated in order.async fn download_bundle_parts( client: &PdsClient, did: &str, entry: &BundleEntry,) -> Result<Vec<u8>, String> { if entry.parts.len() == 1 { // common case: single-part bundle return client.get_blob(did, entry.parts[0].cid()).await; }
// multi-part: download each chunk and reassemble let mut parts = Vec::with_capacity(entry.parts.len()); for part in &entry.parts { let data = client.get_blob(did, part.cid()).await?; parts.push(data); }
Ok(reassemble_chunks(&parts))}
/// Determines which bundles the local repo doesn't have yet.////// Walks the bundle chain from oldest to newest and skips entries/// whose tip commits are already present in the local repo.pub async fn find_new_bundles<'a>( repo_path: &Path, bundles: &'a [BundleEntry],) -> Result<Vec<BundleEntry>, String> { let mut new_bundles = Vec::new();
for entry in bundles { // a bundle is "already applied" if all its tips exist locally let all_tips_present = if entry.tips.is_empty() { false } else { let mut all_present = true; for tip in &entry.tips { if !commit_exists(repo_path, tip).await? { all_present = false; break; } } all_present };
if !all_tips_present { new_bundles.push(entry.clone()); } }
Ok(new_bundles)}
/// Checks if a commit SHA exists in the local repository.async fn commit_exists(repo_path: &Path, sha: &str) -> Result<bool, String> { let output = tokio::process::Command::new("git") .args(["cat-file", "-t", sha]) .current_dir(repo_path) .output() .await .map_err(|e| format!("failed to run git cat-file: {}", e))?;
Ok(output.status.success())}
/// Updates local refs to match the remote state record.////// Uses `git update-ref` for each ref in the state.async fn update_local_refs(repo_path: &Path, state: &RepoState) -> Result<(), String> { for git_ref in &state.refs { let output = tokio::process::Command::new("git") .args(["update-ref", &git_ref.name, &git_ref.sha]) .current_dir(repo_path) .output() .await .map_err(|e| format!("failed to run git update-ref: {}", e))?;
if !output.status.success() { let stderr = String::from_utf8_lossy(&output.stderr); return Err(format!( "git update-ref {} {} failed: {}", git_ref.name, git_ref.sha, stderr.trim() )); } }
Ok(())}
/// Checks out the default branch if the repo has a working tree.////// Picks the first ref that looks like a main branch (main, master),/// or falls back to the first ref in the state.async fn checkout_default_branch(repo_path: &Path, state: &RepoState) -> Result<(), String> { // skip checkout for bare repos if is_bare_repo(repo_path).await? { return Ok(()); }
if state.refs.is_empty() { return Ok(()); }
// find the best default branch let default_ref = state .refs .iter() .find(|r| r.name == "refs/heads/main") .or_else(|| state.refs.iter().find(|r| r.name == "refs/heads/master")) .unwrap_or(&state.refs[0]);
// extract branch name from full ref let branch = default_ref .name .strip_prefix("refs/heads/") .unwrap_or(&default_ref.name);
let output = tokio::process::Command::new("git") .args(["checkout", branch]) .current_dir(repo_path) .output() .await .map_err(|e| format!("failed to run git checkout: {}", e))?;
if !output.status.success() { let stderr = String::from_utf8_lossy(&output.stderr); return Err(format!("git checkout {} failed: {}", branch, stderr.trim())); }
Ok(())}
/// Updates the working tree to match the current branch HEAD.////// After `update_local_refs` moves branch pointers forward, the working/// tree is stale. This resets it to match. Skips bare repos.async fn update_working_tree(repo_path: &Path) -> Result<(), String> { if is_bare_repo(repo_path).await? { return Ok(()); }
// reset the working tree to match the updated branch tip let output = tokio::process::Command::new("git") .args(["reset", "--hard", "HEAD"]) .current_dir(repo_path) .output() .await .map_err(|e| format!("failed to run git reset: {}", e))?;
if !output.status.success() { let stderr = String::from_utf8_lossy(&output.stderr); return Err(format!("git reset --hard HEAD failed: {}", stderr.trim())); }
Ok(())}
/// Returns true if the repo at the given path is a bare repository.async fn is_bare_repo(repo_path: &Path) -> Result<bool, String> { let output = tokio::process::Command::new("git") .args(["rev-parse", "--is-bare-repository"]) .current_dir(repo_path) .output() .await .map_err(|e| format!("failed to run git rev-parse --is-bare-repository: {}", e))?;
let stdout = String::from_utf8_lossy(&output.stdout); Ok(stdout.trim() == "true")}