johnandersen777.bsky.social/atlibs-knotless

Switch repository

Loading…

//! 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>) -> 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<u64, String> {
    let bad = || format!("now is not a UTC datetime: {now}");
    let n = |range: std::ops::Range<usize>| now.get(range).and_then(|s| s.parse::<i64>().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::<i64>().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<Value>) -> 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<String>,
    #[serde(default, rename = "defaultBranch")]
    default_branch: Option<String>,
    #[serde(default)]
    private: Option<Value>,
}

// look for the name -> refuse one that is there, else create the repo record under it.
fn create_repo_handler(run: Run<NewRepo>) -> Result<abi::Handle, String> {
    let (mut step, on) = Step::<NewRepo, Value, Id>::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<String>,
    /// `{name, email}` or "Name <email>"; else the handle (or DID) at `<did>@knotless.invalid`.
    #[serde(default)]
    author: Option<Value>,
    #[serde(default)]
    handle: Option<String>,
    /// 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<Value>,
}

#[derive(Deserialize, Clone)]
struct Change {
    path: String,
    #[serde(default)]
    content: Option<String>,
    #[serde(default)]
    patch: Option<String>,
    #[serde(default)]
    delete: bool,
    #[serde(default)]
    executable: Option<bool>,
}

/// 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<Value>,
    /// The parent commit and the pack its ref names.
    parent: Option<String>,
    pack: Option<String>,
}

#[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<String, (u8, String)>,
    /// The pack the ref names was read; the walk of pack pages is at `cursor`, `walked` once done.
    named_read: bool,
    cursor: Option<String>,
    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<String>)>,
    parts_in: Vec<Option<String>>,
    /// A pack over blobs being written: the writes, and its parts' blob refs as they upload.
    upload: Option<(Vec<Value>, Vec<Value>)>,
    attempts: u32,
    error: Option<String>,
    made: Option<Value>,
}

type CommitStep = Step<CommitArgs, CommitState, Id>;

fn ref_name(branch: &str) -> String {
    format!("refs/heads/{branch}")
}

fn changes(args: &CommitArgs) -> Result<Vec<Change>, 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<Change> = 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 <email>\"")?;
            (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<String, (u8, String)>) -> Result<Objects, String> {
    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<String>, Vec<String>), 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<u8>)>) -> Result<Option<String>, 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<Value>, 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<Objects, String> {
    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<String> = 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<CommitArgs>) -> Result<abi::Handle, String> {
    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"));
    }
}