//! knotless repositories (knotless.dev): git that lives in your atproto account as dev.knotless.* //! records. `create_repo` writes the repo record; `commit` builds a commit without a checkout, //! reading the parent's trees from the repository's pack records, and pushes it as knotless's own //! `repo commit` does: one applyWrites of the pack, the ref's claim and the ref. mod git; mod patch; use std::collections::BTreeMap; use abi::{Call, Did, On, Run, Step, entries}; use data_encoding::{BASE32_NOPAD, BASE64_NOPAD}; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use sha2::{Digest, Sha256}; use git::{BLOB, COMMIT, DIR, Entry, Objects, TREE}; const REPO: &str = "dev.knotless.repo"; const REF: &str = "dev.knotless.ref"; const CLAIM: &str = "dev.knotless.claim"; const PACK: &str = "dev.knotless.pack"; const APPLY: &str = "com.atproto.repo.applyWrites"; const UPLOAD: &str = "com.atproto.repo.uploadBlob"; const GET_BLOB: &str = "com.atproto.sync.getBlob"; const PACK_MIME: &str = "application/x-git-packed-objects"; // A larger pack goes over blobs of at most this many bytes, as knotless's KNOTLESS_CHUNK_SIZE default. const PART_MAX: usize = 4_000_000; // A pack this small rides inline in its record (the lexicon caps `data` at 256 KiB of base64). const INLINE_MAX: usize = 192 * 1024; // Pack records per listRecords page: each can carry 256 KiB of base64, and an event takes 1 MiB. const PAGE: usize = 2; abi::call_ids! { #[derive(Clone, Copy)] enum Id { Repo = "repo", Create = "create", Ref = "ref", Base = "base", Pack = "pack", Packs = "packs", Write = "write", Recheck = "recheck", // Part: uploadBlob of the new pack's part n; Blob: getBlob of part n of the pack being read Part(usize) = "part", Blob(usize) = "blob", } } fn not_found(error: &str) -> bool { error.contains("RecordNotFound") || error.contains("Could not locate record") } fn get(did: &str, collection: &str, rkey: &str) -> Call { Call::new("", "com.atproto.repo.getRecord").params(json!({ "repo": did, "collection": collection, "rkey": rkey })) } fn write(op: &str, collection: &str, rkey: &str, value: Option) -> Value { let mut w = json!({ "$type": format!("com.atproto.repo.applyWrites#{op}"), "collection": collection, "rkey": rkey }); if let Some(mut value) = value { value["$type"] = json!(collection); w["value"] = value; } w } fn valid_name(name: &str) -> Result<(), String> { let ok = !name.is_empty() && name.len() <= 100 && name != "." && name != ".." && name.bytes().all(|b| b.is_ascii_alphanumeric() || b"._~-".contains(&b)); if ok { Ok(()) } else { Err(format!("{name:?} is not a repository name: 1 to 100 of A-Z a-z 0-9 . _ ~ -")) } } fn valid_branch(branch: &str) -> Result<(), String> { let bad = branch.is_empty() || branch.len() > 255 || branch.starts_with(['/', '-', '.']) || branch.ends_with(['/', '.']) || branch.ends_with(".lock") || branch.contains("..") || branch.contains("//") || branch.contains("@{") || branch.chars().any(|c| c.is_control() || " ~^:?*[\\".contains(c)); if bad { Err(format!("{branch:?} is not a branch name")) } else { Ok(()) } } /// knotless's key for a ref: 26 symbols of the lowercase base32 of the name's SHA-256. fn ref_hash(name: &str) -> String { let mut text = BASE32_NOPAD.encode(&Sha256::digest(name.as_bytes())).to_lowercase(); text.truncate(26); text } /// Milliseconds since the epoch of the host's `now` (`2026-11-06T12:34:56.789Z`). fn millis(now: &str) -> Result { let bad = || format!("now is not a UTC datetime: {now}"); let n = |range: std::ops::Range| now.get(range).and_then(|s| s.parse::().ok()).ok_or_else(bad); let (y, m, d) = (n(0..4)?, n(5..7)?, n(8..10)?); let secs = n(11..13)? * 3600 + n(14..16)? * 60 + n(17..19)?; let ms = now.get(19..).filter(|r| r.starts_with('.')).and_then(|r| r[1..].get(..3)).and_then(|s| s.parse::().ok()).unwrap_or(0); // Days from the civil date (Howard Hinnant's algorithm). let y = if m <= 2 { y - 1 } else { y }; let era = y.div_euclid(400); let yoe = y - era * 400; let doy = (153 * (m + if m > 2 { -3 } else { 9 }) + 2) / 5 + d - 1; let days = era * 146097 + yoe * 365 + yoe / 4 - yoe / 100 + doy - 719468; u64::try_from((days * 86400 + secs) * 1000 + ms).map_err(|_| bad()) } /// A TID at `micros`; modules have no randomness, so the clock id comes from `seed`. fn tid(micros: u64, seed: &str) -> String { const S32: &[u8] = b"234567abcdefghijklmnopqrstuvwxyz"; let clock = Sha256::digest(seed.as_bytes()); let mut v = (micros << 10) | (u64::from(clock[0]) << 2 | u64::from(clock[1] & 3)); let mut out = [b'2'; 13]; for slot in out.iter_mut().rev() { *slot = S32[(v & 31) as usize]; v >>= 5; } String::from_utf8(out.to_vec()).unwrap() } fn yes(value: &Option) -> bool { matches!(value, Some(Value::Bool(true))) || matches!(value, Some(Value::String(s)) if s == "true" || s == "on") } // --- create_repo ------------------------------------------------------------------------------- #[derive(Deserialize)] struct NewRepo { did: Did, now: String, name: String, #[serde(default)] description: Option, #[serde(default, rename = "defaultBranch")] default_branch: Option, #[serde(default)] private: Option, } // look for the name -> refuse one that is there, else create the repo record under it. fn create_repo_handler(run: Run) -> Result { let (mut step, on) = Step::::new(run)?; let a = &step.args; let did = a.did.as_str(); let url = format!("knotless://{did}/{}", a.name); match on { On::Start => { valid_name(&a.name)?; if yes(&a.private) { return Err("private repositories live in a space on the PDS (com.atproto.space.*), which this module does not write yet".into()); } if let Some(branch) = a.default_branch.as_deref().filter(|b| !b.is_empty()) { valid_branch(branch)?; } if a.description.as_ref().is_some_and(|d| d.len() > 3000) { return Err("description is longer than 3000 bytes".into()); } step.out.call(Id::Repo, get(did, REPO, &a.name)); } On::Result(Id::Repo, Ok(_)) => return Err(format!("{url} already exists")), On::Result(Id::Repo, Err(e)) if not_found(&e) => { let key = tid(millis(&a.now)? * 1000, &format!("{did}/{}", a.name)); let mut record = json!({ "$type": REPO, "key": key, "name": a.name, "createdAt": a.now, "defaultBranch": a.default_branch.as_deref().filter(|b| !b.is_empty()).unwrap_or("main") }); if let Some(d) = a.description.as_deref().filter(|d| !d.is_empty()) { record["description"] = json!(d); } step.state = json!({ "key": key }); step.out.call(Id::Create, abi::repo::create_record(did, REPO, Some(&a.name), record)); } On::Result(Id::Repo, Err(e)) => return Err(e), On::Result(Id::Create, result) => { let uri = result?["uri"].clone(); let key = step.state["key"].clone(); step.out.finish(json!({ "uri": uri, "url": url, "name": a.name, "key": key })); } _ => {} } step.handle() } // --- commit -------------------------------------------------------------------------------------- #[derive(Deserialize)] struct CommitArgs { did: Did, now: String, name: String, branch: String, message: String, /// The branch a new `branch` starts from: the default branch when absent. #[serde(default)] parent: Option, /// `{name, email}` or "Name "; else the handle (or DID) at `@knotless.invalid`. #[serde(default)] author: Option, #[serde(default)] handle: Option, /// A list of changes, or that list as JSON text (a form field). changes: Value, /// How many times to rebuild on a branch that moved under the write (default 5). #[serde(default)] retries: Option, } #[derive(Deserialize, Clone)] struct Change { path: String, #[serde(default)] content: Option, #[serde(default)] patch: Option, #[serde(default)] delete: bool, #[serde(default)] executable: Option, } /// What the ref read says the commit goes on. #[derive(Default, Serialize, Deserialize, Clone, PartialEq)] struct RefRead { /// The branch's record, when it exists (its generation is what the claim takes). record: Option, /// The parent commit and the pack its ref names. parent: Option, pack: Option, } #[derive(Default, Serialize, Deserialize)] struct CommitState { key: String, default_branch: String, read: RefRead, /// The objects the build needs found so far: hex id -> [kind, base64]. objects: BTreeMap, /// The pack the ref names was read; the walk of pack pages is at `cursor`, `walked` once done. named_read: bool, cursor: Option, walked: bool, /// Blob-backed packs still to read: (rkey, the parts' CIDs), the first one's parts as they arrive. blob_packs: Vec<(String, Vec)>, parts_in: Vec>, /// A pack over blobs being written: the writes, and its parts' blob refs as they upload. upload: Option<(Vec, Vec)>, attempts: u32, error: Option, made: Option, } type CommitStep = Step; fn ref_name(branch: &str) -> String { format!("refs/heads/{branch}") } fn changes(args: &CommitArgs) -> Result, String> { let value = match &args.changes { Value::String(text) => serde_json::from_str(text).map_err(|e| format!("changes is not JSON: {e}"))?, other => other.clone(), }; let changes: Vec = serde_json::from_value(value).map_err(|e| format!("changes must be a list of {{path, content | patch | delete}}: {e}"))?; if changes.is_empty() { return Err("changes is empty".into()); } for c in &changes { let segments: Vec<&str> = c.path.split('/').collect(); if c.path.contains(['\0', '\n']) || segments.iter().any(|s| s.is_empty() || *s == "." || *s == ".." || s.eq_ignore_ascii_case(".git")) { return Err(format!("{:?} is not a path a commit can change", c.path)); } if usize::from(c.content.is_some()) + usize::from(c.patch.is_some()) + usize::from(c.delete) != 1 { return Err(format!("{}: give exactly one of content, patch or delete", c.path)); } } Ok(changes) } fn author(args: &CommitArgs) -> Result<(String, String), String> { let (name, email) = match &args.author { Some(Value::Object(o)) => (o.get("name").and_then(Value::as_str).unwrap_or("").to_string(), o.get("email").and_then(Value::as_str).unwrap_or("").to_string()), Some(Value::String(s)) if !s.trim().is_empty() => { let (n, e) = s.trim().trim_end_matches('>').split_once('<').ok_or("author must be \"Name \"")?; (n.trim().to_string(), e.trim().to_string()) } _ => (String::new(), String::new()), }; let did = args.did.as_str(); let name = if name.is_empty() { args.handle.clone().filter(|h| !h.is_empty()).unwrap_or_else(|| did.to_string()) } else { name }; let email = if email.is_empty() { format!("{did}@knotless.invalid") } else { email }; if format!("{name}{email}").contains(['<', '>', '\n']) { return Err("author name and email may not hold <, > or a newline".into()); } Ok((name, email)) } fn decode(objects: &BTreeMap) -> Result { objects.iter().map(|(id, (kind, b64))| Ok((id.clone(), (*kind, BASE64_NOPAD.decode(b64.trim_end_matches('=').as_bytes()).map_err(|e| e.to_string())?)))).collect() } /// The objects a build needs, reachable from `parent`: its commit, the trees down to each changed /// path, and the blob of each patched file. Returns those found (in `have`) and those not. fn needed(parent: &str, changes: &[Change], have: &Objects) -> Result<(Vec, Vec), String> { let (mut found, mut missing) = (Vec::new(), Vec::new()); let Some((_, commit)) = have.get(parent) else { return Ok((found, vec![parent.to_string()])) }; found.push(parent.to_string()); let paths: Vec<(Vec<&str>, bool)> = changes.iter().map(|c| (c.path.split('/').collect(), c.patch.is_some())).collect(); let mut walk = vec![(git::commit_tree(commit)?, paths)]; while let Some((id, paths)) = walk.pop() { let Some((kind, data)) = have.get(&id) else { missing.push(id); continue; }; found.push(id.clone()); if *kind != TREE { continue; } let entries = git::read_tree(data)?; let mut by: BTreeMap<&str, Vec<(Vec<&str>, bool)>> = BTreeMap::new(); for (segments, blob) in paths { by.entry(segments[0]).or_default().push((segments[1..].to_vec(), blob)); } for (name, rests) in by { let Some(e) = entries.get(name) else { continue }; let deeper: Vec<_> = rests.iter().filter(|(r, _)| !r.is_empty()).cloned().collect(); if e.mode == DIR && !deeper.is_empty() { walk.push((e.id.clone(), deeper)); } else if e.mode != DIR && e.mode != "160000" && rests.iter().any(|(r, blob)| r.is_empty() && *blob) { walk.push((e.id.clone(), Vec::new())); } } } Ok((found, missing)) } /// The tree `changes` (each with its path left, below this tree) make of `tree`; None when empty. fn rewrite(have: &Objects, tree: Option<&str>, changes: Vec<(Vec<&str>, &Change)>, out: &mut Vec<(u8, Vec)>) -> Result, String> { let mut entries = match tree { Some(id) => git::read_tree(&have.get(id).ok_or_else(|| format!("tree {id} is missing"))?.1)?, None => BTreeMap::new(), }; let mut by: BTreeMap<&str, Vec<(Vec<&str>, &Change)>> = BTreeMap::new(); for (segments, change) in changes { by.entry(segments[0]).or_default().push((segments[1..].to_vec(), change)); } for (name, group) in by { let existing = entries.get(name).cloned(); if group.iter().any(|(rest, _)| !rest.is_empty()) { if let Some((_, c)) = group.iter().find(|(rest, _)| rest.is_empty()) { return Err(format!("{} is changed as a file and as a directory", c.path)); } let sub = match &existing { Some(e) if e.mode == DIR => Some(e.id.as_str()), Some(_) => return Err(format!("{name} is a file, not a directory ({})", group[0].1.path)), None => None, }; match rewrite(have, sub, group, out)? { Some(id) => entries.insert(name.to_string(), Entry { mode: DIR.into(), id }), None => entries.remove(name), }; continue; } let [(_, change)] = group[..] else { return Err(format!("{} is changed more than once", group[0].1.path)) }; if existing.as_ref().is_some_and(|e| e.mode == DIR) { return Err(format!("{} is a directory", change.path)); } let text = match (&change.content, &change.patch) { (Some(content), _) => Some(content.clone()), (_, Some(patch)) => { let old = match &existing { Some(e) => { let (_, data) = have.get(&e.id).ok_or_else(|| format!("blob of {} is missing", change.path))?; String::from_utf8(data.clone()).map_err(|_| format!("{} is not text, so it cannot be patched", change.path))? } None => String::new(), }; patch::apply(&old, patch).map_err(|e| format!("{}: {e}", change.path))? } _ => None, }; match text { None => { entries.remove(name); } Some(text) => { let data = text.into_bytes(); let id = git::id(BLOB, &data)?; out.push((BLOB, data)); let mode = match (change.executable, &existing) { (Some(true), _) => "100755".to_string(), (Some(false), _) => "100644".to_string(), (None, Some(e)) => e.mode.clone(), (None, None) => "100644".to_string(), }; entries.insert(name.to_string(), Entry { mode, id }); } } } if entries.is_empty() { return Ok(None); } let data = git::write_tree(&entries)?; let id = git::id(TREE, &data)?; out.push((TREE, data)); Ok(Some(id)) } /// Read the branch's ref, or the parent branch's for a new branch. fn read_ref(step: &mut CommitStep, id: Id, branch: &str) { let rkey = format!("{}.{}", step.state.key, ref_hash(&ref_name(branch))); step.out.call(id, get(step.args.did.as_str(), REF, &rkey)); } /// On a fresh ref read: forget where the pack walk was (the objects found so far still count). fn begin(step: &mut CommitStep, read: RefRead) -> Result<(), String> { let s = &mut step.state; s.read = read; s.named_read = false; s.cursor = None; s.walked = false; s.blob_packs.clear(); s.parts_in.clear(); s.upload = None; advance(step, &Objects::new()) } fn ref_read(record: Option, base: Option<&Value>) -> RefRead { let of = |v: &Value| (v["target"].as_str().map(String::from), v["pack"].as_str().map(String::from)); let (parent, pack) = record.as_ref().or(base).map(of).unwrap_or_default(); RefRead { record, parent, pack } } /// Fold a pack's objects in, keep those the build needs, and build, or read the next pack. fn advance(step: &mut CommitStep, fresh: &Objects) -> Result<(), String> { let changes = changes(&step.args)?; let mut have = decode(&step.state.objects)?; have.extend(fresh.iter().map(|(k, v)| (k.clone(), v.clone()))); let Some(parent) = step.state.read.parent.clone() else { return build(step, &have, &changes) }; let (found, missing) = needed(&parent, &changes, &have)?; for id in found { if let Some((kind, data)) = fresh.get(&id) { step.state.objects.insert(id, (*kind, BASE64_NOPAD.encode(data))); } } if missing.is_empty() { return build(step, &decode(&step.state.objects)?, &changes); } let did = step.args.did.as_str().to_string(); let s = &mut step.state; if let Some((_, cids)) = s.blob_packs.first() { // A blob-backed pack: its parts, read whole (the run's ioBytes must hold them: 16 MiB at most) s.parts_in = vec![None; cids.len()]; for (i, cid) in cids.clone().iter().enumerate() { step.out.call(Id::Blob(i), Call::new("", GET_BLOB).params(json!({ "did": did, "cid": cid }))); } } else if !s.named_read && let Some(pack) = s.read.pack.clone() { s.named_read = true; step.out.call(Id::Pack, get(&did, PACK, &pack)); } else if !s.walked { let cursor = s.cursor.clone().unwrap_or_else(|| format!("{}/", s.key)); step.out.call(Id::Packs, abi::repo::list_records(&did, PACK, PAGE, Some(&cursor))); } else { return Err(format!("the repository has no pack holding {}", missing.join(", "))); } Ok(()) } /// The objects of pack records: inline ones now; those over blobs are queued to read (advance). fn unpack(step: &mut CommitStep, records: &[(String, Value)]) -> Result { let mut objects = Objects::new(); let outside = decode(&step.state.objects)?; for (rkey, value) in records { match value["data"]["$bytes"].as_str() { Some(b64) => { let bytes = BASE64_NOPAD.decode(b64.trim_end_matches('=').as_bytes()).map_err(|e| format!("pack {rkey}: {e}"))?; let mut both = outside.clone(); both.extend(objects.clone()); objects.extend(git::read_pack(&bytes, &both).map_err(|e| format!("pack {rkey}: {e}"))?); } None => { let cids: Vec = value["parts"].as_array().into_iter().flatten().filter_map(|p| p["ref"]["$link"].as_str().map(String::from)).collect(); if cids.is_empty() { return Err(format!("pack {rkey} has neither data nor parts")); } step.state.blob_packs.push((rkey.clone(), cids)); } } } Ok(objects) } fn build(step: &mut CommitStep, have: &Objects, changes: &[Change]) -> Result<(), String> { let a = &step.args; let s = &step.state; let parent = s.read.parent.as_deref(); let parent_tree = match parent { Some(p) => Some(git::commit_tree(&have.get(p).ok_or("the parent commit is missing")?.1)?), None => None, }; let mut objects = Vec::new(); let paths = changes.iter().map(|c| (c.path.split('/').collect(), c)).collect(); let tree = match rewrite(have, parent_tree.as_deref(), paths, &mut objects)? { Some(tree) => tree, None => { objects.push((TREE, Vec::new())); git::id(TREE, &[])? } }; let url = format!("knotless://{}/{}", a.did.as_str(), a.name); if parent_tree.as_deref() == Some(tree.as_str()) { step.out.finish(json!({ "sha": parent, "changed": false, "branch": a.branch, "url": url })); return Ok(()); } let (name, email) = author(a)?; let ms = millis(&a.now)?; let who = format!("{name} <{email}> {} +0000", ms / 1000); let mut message = a.message.clone(); if !message.ends_with('\n') { message.push('\n'); } let text = format!("tree {tree}\n{}author {who}\ncommitter {who}\n\n{message}", parent.map(|p| format!("parent {p}\n")).unwrap_or_default()); let commit = git::id(COMMIT, text.as_bytes())?; objects.push((COMMIT, text.into_bytes())); let mut seen = std::collections::BTreeSet::new(); objects.retain(|(kind, data)| git::id(*kind, data).is_ok_and(|id| !have.contains_key(&id) && seen.insert(id))); let count = objects.len(); let pack = git::write_pack(&objects); let key = &s.key; let pack_rkey = format!("{key}.{}", tid(ms * 1000 + u64::from(s.attempts), &commit)); let mut pack_record = json!({ "repo": key, "tips": [commit], "size": pack.len(), "objects": count, "createdAt": a.now }); // Past INLINE_MAX the pack goes over blobs (`parts`, filled in as they upload), as knotless push does. let parts: Vec<&[u8]> = if pack.len() > INLINE_MAX { pack.chunks(PART_MAX).collect() } else { vec![] }; if parts.is_empty() { pack_record["data"] = json!({ "$bytes": BASE64_NOPAD.encode(&pack) }); } if let Some(p) = parent { pack_record["bases"] = json!([p]); } let name_ref = ref_name(&a.branch); let rkey = format!("{key}.{}", ref_hash(&name_ref)); let mut writes = vec![write("create", PACK, &pack_rkey, Some(pack_record))]; let mut record = json!({ "repo": key, "name": name_ref, "target": commit, "updatedAt": a.now, "pack": pack_rkey }); match &s.read.record { Some(previous) => { let generation = previous["generation"].as_u64().ok_or("the ref record has no generation")?; record["generation"] = json!(generation + 1); record["createdAt"] = previous.get("createdAt").cloned().unwrap_or(json!(a.now)); let claim = json!({ "repo": key, "name": name_ref, "generation": generation, "from": parent, "to": commit, "pack": pack_rkey, "createdAt": a.now }); writes.push(write("create", CLAIM, &format!("{rkey}.{generation}"), Some(claim))); writes.push(write("update", REF, &rkey, Some(record))); } None => { record["generation"] = json!(ms); record["createdAt"] = json!(a.now); writes.push(write("create", REF, &rkey, Some(record))); } } step.state.made = Some(json!({ "sha": commit, "parent": parent, "changed": true, "branch": a.branch, "pack": pack_rkey, "url": url })); if parts.is_empty() { step.out.call(Id::Write, Call::new("", APPLY).input(json!({ "repo": a.did.as_str(), "writes": writes }))); } else { for (i, part) in parts.iter().enumerate() { step.out.call(Id::Part(i), Call::new("", UPLOAD).bytes(BASE64_NOPAD.encode(part), PACK_MIME)); } step.state.upload = Some((writes, vec![Value::Null; parts.len()])); } Ok(()) } // repo record -> the branch's ref (else the parent branch's) -> the packs holding the parent's trees, // the ref's own pack first, then pages newest first -> build -> applyWrites. A write that loses to a // move of the branch rereads the ref and builds again on the new tip, up to `retries` times. fn commit_handler(run: Run) -> Result { let (mut step, on) = CommitStep::new(run)?; let did = step.args.did.as_str().to_string(); match on { On::Start => { valid_name(&step.args.name)?; valid_branch(&step.args.branch)?; if let Some(p) = step.args.parent.as_deref().filter(|p| !p.is_empty()) { valid_branch(p)?; } changes(&step.args)?; author(&step.args)?; millis(&step.args.now)?; step.out.call(Id::Repo, get(&did, REPO, &step.args.name)); } On::Result(Id::Repo, result) => { let repo = result.map_err(|e| if not_found(&e) { format!("knotless://{did}/{} does not exist", step.args.name) } else { e })?; step.state.key = repo["value"]["key"].as_str().ok_or("the repo record has no key")?.to_string(); step.state.default_branch = repo["value"]["defaultBranch"].as_str().unwrap_or("main").to_string(); let branch = step.args.branch.clone(); read_ref(&mut step, Id::Ref, &branch); } On::Result(Id::Ref, Ok(found)) => begin(&mut step, ref_read(Some(found["value"].clone()), None))?, On::Result(Id::Ref, Err(e)) if not_found(&e) => { let base = step.args.parent.clone().filter(|p| !p.is_empty()).unwrap_or_else(|| step.state.default_branch.clone()); read_ref(&mut step, Id::Base, &base); } On::Result(Id::Base, Ok(found)) => begin(&mut step, ref_read(None, Some(&found["value"])))?, // An empty repository, or a parent branch that is not there: the branch's first commit. On::Result(Id::Base, Err(e)) if not_found(&e) => { if step.args.parent.as_deref().is_some_and(|p| !p.is_empty()) { return Err(format!("branch {} does not exist", step.args.parent.as_deref().unwrap_or(""))); } begin(&mut step, RefRead::default())?; } On::Result(Id::Pack, result) => { let rkey = step.state.read.pack.clone().unwrap_or_default(); // A ref may name a pack a compaction folded away since: walk the pages instead. let records = match result { Ok(found) => vec![(rkey, found["value"].clone())], Err(e) if not_found(&e) => Vec::new(), Err(e) => return Err(e), }; let fresh = unpack(&mut step, &records)?; advance(&mut step, &fresh)?; } On::Result(Id::Packs, result) => { let page = result?; let prefix = format!("{}.", step.state.key); let all = page["records"].as_array().cloned().unwrap_or_default(); let mine: Vec<(String, Value)> = all .iter() .filter_map(|r| Some((r["uri"].as_str()?.rsplit('/').next()?.to_string(), r["value"].clone()))) .filter(|(rkey, _)| rkey.starts_with(&prefix)) .collect(); let named = step.state.read.pack.clone(); let mine: Vec<_> = mine.into_iter().filter(|(rkey, _)| Some(rkey) != named.as_ref() || !step.state.named_read).collect(); step.state.walked = all.len() < PAGE || mine.len() < all.len() || page["cursor"].as_str().is_none(); step.state.cursor = all.last().and_then(|r| r["uri"].as_str()).and_then(|u| u.rsplit('/').next()).map(String::from); let fresh = unpack(&mut step, &mine)?; advance(&mut step, &fresh)?; } On::Result(Id::Part(i), result) => { let blob = result?["blob"].clone(); let Some((_, parts)) = step.state.upload.as_mut().filter(|(_, p)| i < p.len()) else { return Err("an upload nobody asked for".into()) }; parts[i] = blob; if parts.iter().all(|p| !p.is_null()) { let (mut writes, parts) = step.state.upload.take().unwrap_or_default(); writes[0]["value"]["parts"] = json!(parts); step.out.call(Id::Write, Call::new("", APPLY).input(json!({ "repo": did, "writes": writes }))); } } On::Result(Id::Blob(i), result) => { let b64 = result?["$bytes"].as_str().ok_or("getBlob answered no bytes")?.to_string(); let s = &mut step.state; let slot = s.parts_in.get_mut(i).ok_or("a blob nobody asked for")?; *slot = Some(b64); if s.parts_in.iter().all(Option::is_some) { let (rkey, _) = s.blob_packs.remove(0); let mut bytes = Vec::new(); for part in std::mem::take(&mut s.parts_in).into_iter().flatten() { bytes.extend(BASE64_NOPAD.decode(part.trim_end_matches('=').as_bytes()).map_err(|e| format!("pack {rkey}: {e}"))?); } let have = decode(&s.objects)?; let fresh = git::read_pack(&bytes, &have).map_err(|e| format!("pack {rkey}: {e}"))?; advance(&mut step, &fresh)?; } } On::Result(Id::Write, Ok(_)) => { let made = step.state.made.take(); step.out.finish(made); } On::Result(Id::Write, Err(e)) => { let retries = match &step.args.retries { Some(Value::Number(n)) => n.as_u64().unwrap_or(5), Some(Value::String(s)) if !s.is_empty() => s.parse().map_err(|_| "retries must be a number")?, _ => 5, }; if step.state.attempts >= retries as u32 { return Err(e); } step.state.error = Some(e); let branch = step.args.branch.clone(); read_ref(&mut step, Id::Recheck, &branch); } On::Result(Id::Recheck, result) => { let now = match result { Ok(found) => Some(found["value"].clone()), Err(e) if not_found(&e) => None, Err(e) => return Err(e), }; // Not moved: the write failed for some other reason. if now == step.state.read.record { return Err(step.state.error.take().unwrap_or_default()); } step.state.attempts += 1; let read = match now { Some(record) => ref_read(Some(record), None), None => RefRead { record: None, ..step.state.read.clone() }, }; begin(&mut step, read)?; } _ => {} } step.handle() } entries!(create_repo => create_repo_handler, commit => commit_handler); #[cfg(test)] mod tests { use super::*; #[test] fn keys_match_knotless() { // A live ref record: at://did:plc:5svqtrhheairglgiiyvutzik/dev.knotless.ref/3mzrf7luhouqg.7eq32bpgrmbxidcfbzlf4dtbop assert_eq!(ref_hash("refs/heads/main"), "7eq32bpgrmbxidcfbzlf4dtbop"); assert_eq!(millis("1970-01-02T00:00:01.5Z").unwrap(), 86_401_000); assert_eq!(millis("2026-11-06T12:34:56.789Z").unwrap(), 1_793_968_496_789); let t = tid(1_794_486_896_789_000, "x"); assert_eq!(t.len(), 13); assert!(t < tid(1_794_486_896_790_000, "x")); } }