Something went wrong. Try again.
unified development activity
Something went wrong. Try again.
26 kB · 751 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752//! Read-only access to Tangled: the public appview index, repo records on a//! PDS, and Git history from a knot (via the mirror when it has a copy).
use std::collections::BTreeMap;
use anyhow::{Context, Result, bail};use serde_json::Value;use url::Url;
use crate::db::canonical_time;use crate::http::HttpClient;
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]pub struct RepoRecord { /// The repository's DID, or for a record from before repositories had /// one, `<owner DID>/<name>`: the identifier its knot still serves it by /// (see [`is_legacy_repo`]). pub repo_did: String, pub owner_did: String, pub name: String, pub knot: String, pub url: String, pub website: Option<String>, /// A fork record carries the upstream repo's `source`. pub fork: bool, pub created_at: Option<String>, /// The repo record's own description, as its owner wrote it. pub description: Option<String>,}
/// Every repository record search returned, and how many of them were/// legacy records (no `repoDid`) and distinct owners.#[derive(Debug, Clone, Default)]pub struct SearchResults { pub repos: Vec<RepoRecord>, pub legacy: usize, pub owners: std::collections::BTreeSet<String>,}
/// Whether `repo_id` is a legacy `<owner DID>/<name>` identifier. Knots/// address such a repository by owner and name; the mirror doesn't have it.pub fn is_legacy_repo(repo_id: &str) -> bool { repo_id.contains('/')}
#[derive(Debug, Clone)]pub struct BranchHead { pub name: String, pub head_sha: String, pub is_default: bool,}
#[derive(Debug, Clone)]pub struct CommitRecord { pub sha: String, pub change_id: String, pub author_name: String, pub author_email: String, pub authored_at: String, pub committed_at: String, pub message: String,}
#[derive(Debug, Clone)]pub struct CommitPage { pub commits: Vec<CommitRecord>, pub total: Option<u64>,}
#[derive(Debug, Clone)]pub struct DidDescription { pub handle: String, pub operator: Option<String>, pub collections: Vec<String>,}
#[derive(Debug, Clone)]pub struct Tangled { http: HttpClient, api: String, mirror: String,}
impl Tangled { pub fn new(http: HttpClient, api: impl Into<String>, mirror: impl Into<String>) -> Self { Self { http, api: api.into().trim_end_matches('/').to_string(), mirror: mirror.into().trim_end_matches('/').to_string(), } }
/// The whole public repo index. Search is the only public enumeration that /// reaches every knot; `sh.tangled.repo.listRepos` needs an owner. Records /// without a `repoDid` (repositories created before they had one) are kept /// under their legacy `<owner>/<name>` identifier. pub async fn search_repos(&self) -> Result<SearchResults> { let mut out = BTreeMap::<String, RepoRecord>::new(); let mut owners = std::collections::BTreeSet::new(); let mut cursor: Option<String> = None; let mut pages = 0usize;
loop { let mut url = Url::parse(&format!("{}/xrpc/sh.tangled.search.query", self.api))?; { let mut query = url.query_pairs_mut(); query.append_pair("q", "*"); query.append_pair("nsid", "sh.tangled.repo"); query.append_pair("limit", "100"); if let Some(cursor) = &cursor { query.append_pair("cursor", cursor); } }
let body: Value = self .http .get_json(url.as_str(), None) .await .with_context(|| format!("enumerating Tangled repos (page {})", pages + 1))?; let hits = body .get("hits") .and_then(Value::as_array) .cloned() .unwrap_or_default(); if hits.is_empty() { break; }
for hit in hits { if let Some(record) = repo_from_search_hit(&hit) { owners.insert(record.owner_did.clone()); out.entry(record.repo_did.clone()).or_insert(record); } }
cursor = body .get("cursor") .and_then(Value::as_str) .map(ToString::to_string); pages += 1; if cursor.is_none() { break; } if pages >= 1_000 { bail!("Tangled search did not terminate after 1000 pages"); } }
let repos: Vec<RepoRecord> = out.into_values().collect(); Ok(SearchResults { legacy: repos .iter() .filter(|repo| is_legacy_repo(&repo.repo_did)) .count(), repos, owners, }) }
/// Every account a relay has seen write a `sh.tangled.repo` record. Some /// of them search doesn't index; most of the rest have since deleted the /// record or the account, which a listing of their records shows. pub async fn relay_repo_owners(&self, relay: &str) -> Result<Vec<String>> { let mut out = Vec::new(); let mut cursor: Option<String> = None; for page in 1..=1_000 { let mut url = Url::parse(&format!( "{}/xrpc/com.atproto.sync.listReposByCollection", relay.trim_end_matches('/') ))?; { let mut query = url.query_pairs_mut(); query.append_pair("collection", "sh.tangled.repo"); query.append_pair("limit", "2000"); if let Some(cursor) = &cursor { query.append_pair("cursor", cursor); } } let body: Value = self .http .get_json(url.as_str(), None) .await .with_context(|| { format!("listing sh.tangled.repo owners on {relay} (page {page})") })?; let repos = body .get("repos") .and_then(Value::as_array) .cloned() .unwrap_or_default(); out.extend( repos .iter() .filter_map(|repo| repo.get("did").and_then(Value::as_str)) .map(ToString::to_string), ); cursor = body .get("cursor") .and_then(Value::as_str) .map(ToString::to_string); if cursor.is_none() || repos.is_empty() { return Ok(out); } } bail!("{relay} did not finish listing sh.tangled.repo owners after 1000 pages") }
/// Repo records published by one owner. Used to make sure the person's own /// repositories are present even if search is behind. pub async fn repos_for_owner(&self, pds: &str, did: &str) -> Result<Vec<RepoRecord>> { let mut out = Vec::new(); let mut cursor: Option<String> = None; loop { let mut url = Url::parse(&format!( "{}/xrpc/com.atproto.repo.listRecords", pds.trim_end_matches('/') ))?; { let mut query = url.query_pairs_mut(); query.append_pair("repo", did); query.append_pair("collection", "sh.tangled.repo"); query.append_pair("limit", "100"); if let Some(cursor) = &cursor { query.append_pair("cursor", cursor); } } let body: Value = self .http .get_json(url.as_str(), None) .await .with_context(|| format!("listing Tangled repos for {did}"))?; let records = body .get("records") .and_then(Value::as_array) .cloned() .unwrap_or_default(); for record in records { if let Some(value) = record.get("value") { let rkey = record .get("uri") .and_then(Value::as_str) .and_then(|uri| uri.rsplit('/').next()); if let Some(repo) = repo_from_record(did, value, rkey) { out.push(repo); } } } cursor = body .get("cursor") .and_then(Value::as_str) .map(ToString::to_string); if cursor.is_none() { break; } } Ok(out) }
/// Branch heads. The mirror is tried first because it returns 1000 refs a /// page; the knot's own API is the fallback and is capped at 100. pub async fn branches(&self, knot: &str, repo_did: &str) -> Result<Vec<BranchHead>> { if is_legacy_repo(repo_did) { return self .branches_knot(knot, repo_did) .await .with_context(|| format!("listing branches for {repo_did}")); } match self.branches_mirror(repo_did).await { Ok(branches) => Ok(branches), Err(mirror_error) => { tracing::debug!(repo_did, error = %mirror_error, "mirror branch listing failed; asking the knot"); self.branches_knot(knot, repo_did) .await .with_context(|| format!("listing branches for {repo_did}")) } } }
async fn branches_mirror(&self, repo_did: &str) -> Result<Vec<BranchHead>> { let mut out = Vec::new(); let mut offset = 0u64; let limit = 1000u64; loop { let mut url = Url::parse(&format!( "{}/xrpc/sh.tangled.git.temp.listBranches", self.mirror ))?; { let mut query = url.query_pairs_mut(); query.append_pair("repo", repo_did); query.append_pair("limit", &limit.to_string()); query.append_pair("cursor", &offset.to_string()); } let body: Value = self.http.get_json(url.as_str(), None).await?; let branches = body .get("branches") .and_then(Value::as_array) .cloned() .unwrap_or_default(); let count = branches.len(); for branch in branches { if let Some(head) = branch_head(&branch) { out.push(head); } } if count < limit as usize { break; } offset += limit; } Ok(out) }
async fn branches_knot(&self, knot: &str, repo_did: &str) -> Result<Vec<BranchHead>> { let mut out = Vec::new(); let mut offset = 0u64; let limit = 100u64; loop { let mut url = Url::parse(&format!( "https://{}/xrpc/sh.tangled.repo.branches", knot.trim_end_matches('/') ))?; { let mut query = url.query_pairs_mut(); query.append_pair("repo", repo_did); query.append_pair("limit", &limit.to_string()); query.append_pair("cursor", &offset.to_string()); } let body: Value = self.http.get_json(url.as_str(), None).await?; let branches = body .get("branches") .and_then(Value::as_array) .cloned() .unwrap_or_default(); let count = branches.len(); for branch in branches { if let Some(head) = branch_head(&branch) { out.push(head); } } if count < limit as usize { break; } offset += limit; } Ok(out) }
/// A page of commits reachable from `reference`. pub async fn commits( &self, knot: &str, repo_did: &str, reference: &str, offset: u64, limit: u64, ) -> Result<CommitPage> { if is_legacy_repo(repo_did) { return self .commits_knot(knot, repo_did, reference, offset, limit.min(100)) .await .with_context(|| format!("reading commits for {repo_did} at {reference}")); } match self .commits_mirror(repo_did, reference, offset, limit.min(1000)) .await { Ok(page) => Ok(page), Err(mirror_error) => { tracing::debug!(repo_did, reference, error = %mirror_error, "mirror commit log failed; asking the knot"); self.commits_knot(knot, repo_did, reference, offset, limit.min(100)) .await .with_context(|| format!("reading commits for {repo_did} at {reference}")) } } }
async fn commits_mirror( &self, repo_did: &str, reference: &str, offset: u64, limit: u64, ) -> Result<CommitPage> { let mut url = Url::parse(&format!( "{}/xrpc/sh.tangled.git.temp.listCommits", self.mirror ))?; { let mut query = url.query_pairs_mut(); query.append_pair("repo", repo_did); query.append_pair("ref", reference); query.append_pair("limit", &limit.to_string()); query.append_pair("cursor", &offset.to_string()); } let body: Value = self.http.get_json(url.as_str(), None).await?; parse_commit_page(&body) }
async fn commits_knot( &self, knot: &str, repo_did: &str, reference: &str, offset: u64, limit: u64, ) -> Result<CommitPage> { let mut url = Url::parse(&format!( "https://{}/xrpc/sh.tangled.repo.log", knot.trim_end_matches('/') ))?; { let mut query = url.query_pairs_mut(); query.append_pair("repo", repo_did); query.append_pair("ref", reference); query.append_pair("limit", &limit.to_string()); query.append_pair("cursor", &offset.to_string()); } let body: Value = self.http.get_json(url.as_str(), None).await?; parse_commit_page(&body) }
/// Resolve a DID's PDS and operator. `pds` is the endpoint to ask when it /// is already known (a didbot directory listing gives it to us); otherwise /// the DID document is fetched. pub async fn describe_did(&self, did: &str, pds: Option<&str>) -> Result<DidDescription> { let pds = match pds { Some(pds) => pds.trim_end_matches('/').to_string(), None => self.resolve_pds(did).await?, }; let mut url = Url::parse(&format!("{pds}/xrpc/com.atproto.repo.describeRepo"))?; url.query_pairs_mut().append_pair("repo", did); let body: Value = self.http.get_json(url.as_str(), None).await?; let handle = body .get("handle") .and_then(Value::as_str) .unwrap_or(did) .to_string(); let operator = body .get("didDoc") .and_then(|doc| doc.get("operator")) .and_then(Value::as_str) .map(ToString::to_string); let collections = body .get("collections") .and_then(Value::as_array) .map(|values| { values .iter() .filter_map(Value::as_str) .map(ToString::to_string) .collect() }) .unwrap_or_default(); Ok(DidDescription { handle, operator, collections, }) }
pub async fn resolve_pds(&self, did: &str) -> Result<String> { let doc_url = if let Some(rest) = did.strip_prefix("did:plc:") { format!("https://plc.directory/did:plc:{rest}") } else if let Some(rest) = did.strip_prefix("did:web:") { let decoded = rest.replace("%3A", ":").replace("%3a", ":"); let mut parts = decoded.split(':'); let host = parts.next().unwrap_or(""); let path = parts.collect::<Vec<_>>().join("/"); if path.is_empty() { format!("https://{host}/.well-known/did.json") } else { format!("https://{host}/{path}/did.json") } } else { bail!("unsupported DID method in {did}"); };
let doc: Value = self .http .get_json(&doc_url, None) .await .with_context(|| format!("resolving DID document for {did}"))?; let endpoints = doc .get("service") .and_then(Value::as_array) .cloned() .unwrap_or_default(); for service in endpoints { let id = service.get("id").and_then(Value::as_str).unwrap_or(""); let kind = service.get("type").and_then(Value::as_str).unwrap_or(""); if (id.ends_with("atproto_pds") || kind == "AtprotoPersonalDataServer") && let Some(endpoint) = service.get("serviceEndpoint").and_then(Value::as_str) { return Ok(endpoint.trim_end_matches('/').to_string()); } } bail!("DID document for {did} has no atproto PDS service") }}
fn repo_from_search_hit(hit: &Value) -> Option<RepoRecord> { let uri = hit.get("uri")?.as_str()?; let value = hit.get("value")?; let owner = uri.strip_prefix("at://")?.split('/').next()?.to_string(); let rkey = uri.rsplit('/').next()?.to_string(); let name = value .get("name") .and_then(Value::as_str) .filter(|name| !name.is_empty()) .unwrap_or(&rkey) .to_string(); let repo_did = repo_id(&owner, &name, value)?; let knot = value .get("knot") .and_then(Value::as_str) .unwrap_or("knot1.tangled.sh") .to_string(); let website = value .get("website") .and_then(Value::as_str) .map(ToString::to_string); let fork = value.get("source").is_some(); let created_at = value .get("createdAt") .and_then(Value::as_str) .map(ToString::to_string); let description = value .get("description") .and_then(Value::as_str) .map(ToString::to_string); Some(RepoRecord { url: format!("https://tangled.org/{owner}/{name}"), repo_did, owner_did: owner, name, knot, website, fork, created_at, description, })}
/// A record's repository identifier: its `repoDid`, or for a legacy record/// `<owner>/<name>`, which is how its knot serves it. A legacy record needs a/// knot and a name that can be a path segment to be addressable.fn repo_id(owner_did: &str, name: &str, value: &Value) -> Option<String> { if let Some(repo_did) = value.get("repoDid").and_then(Value::as_str) { return Some(repo_did.to_string()); } let usable = !name.is_empty() && !name.contains('/') && value.get("knot").is_some(); usable.then(|| format!("{owner_did}/{name}"))}
fn repo_from_record(owner_did: &str, value: &Value, rkey: Option<&str>) -> Option<RepoRecord> { // Older records omit `name`; the record key is the repository name. let name = value .get("name") .and_then(Value::as_str) .filter(|name| !name.is_empty()) .or(rkey) .unwrap_or("") .to_string(); let repo_did = repo_id(owner_did, &name, value)?; let knot = value .get("knot") .and_then(Value::as_str) .unwrap_or("knot1.tangled.sh") .to_string(); let website = value .get("website") .and_then(Value::as_str) .map(ToString::to_string); let fork = value.get("source").is_some(); let created_at = value .get("createdAt") .and_then(Value::as_str) .map(ToString::to_string); let description = value .get("description") .and_then(Value::as_str) .map(ToString::to_string); Some(RepoRecord { url: format!("https://tangled.org/{owner_did}/{name}"), repo_did, owner_did: owner_did.to_string(), name, knot, website, fork, created_at, description, })}
fn branch_head(branch: &Value) -> Option<BranchHead> { let reference = branch.get("reference")?; Some(BranchHead { name: reference.get("name")?.as_str()?.to_string(), head_sha: reference.get("hash")?.as_str()?.to_string(), is_default: branch .get("is_default") .and_then(Value::as_bool) .unwrap_or(false), })}
fn parse_commit_page(body: &Value) -> Result<CommitPage> { let total = body.get("total").and_then(Value::as_u64); let raw = body .get("commits") .and_then(Value::as_array) .cloned() .unwrap_or_default(); let mut commits = Vec::with_capacity(raw.len()); for commit in raw { let Some(hash) = commit.get("hash").and_then(hash_to_hex) else { continue; }; let author = commit.get("author"); let author_name = author .and_then(|value| value.get("Name")) .and_then(Value::as_str) .unwrap_or("") .to_string(); let author_email = author .and_then(|value| value.get("Email")) .and_then(Value::as_str) .unwrap_or("") .to_string(); let authored_raw = author .and_then(|value| value.get("When")) .and_then(Value::as_str) .unwrap_or(""); let committed_raw = commit .get("committer") .and_then(|value| value.get("When")) .and_then(Value::as_str) .unwrap_or(""); let authored_at = canonical_time(authored_raw).unwrap_or_else(|| authored_raw.to_string()); let committed_at = canonical_time(committed_raw).unwrap_or_else(|| committed_raw.to_string()); let full_message = commit.get("message").and_then(Value::as_str).unwrap_or(""); let change_id = crate::model::change_id_from_message(full_message); let message = full_message .lines() .next() .unwrap_or("") .chars() .take(500) .collect::<String>(); commits.push(CommitRecord { sha: hash, change_id, author_name, author_email, authored_at, committed_at, message, }); } Ok(CommitPage { commits, total })}
fn hash_to_hex(value: &Value) -> Option<String> { if let Some(s) = value.as_str() { return Some(s.to_string()); } let bytes: Vec<u8> = value .as_array()? .iter() .map(|byte| byte.as_u64().map(|n| n as u8)) .collect::<Option<Vec<_>>>()?; Some(hex::encode(bytes))}
#[cfg(test)]mod tests { use super::*; use serde_json::json;
#[test] fn parses_byte_array_hashes() { let value = json!([1, 2, 255]); assert_eq!(hash_to_hex(&value).unwrap(), "0102ff"); }
#[test] fn parses_branch_response() { let value = json!({ "reference": {"name": "main", "hash": "abc"}, "commit": {} }); let head = branch_head(&value).unwrap(); assert_eq!(head.name, "main"); assert_eq!(head.head_sha, "abc"); }
#[test] fn a_record_without_a_repo_did_is_kept_by_owner_and_name() { let hit = json!({ "uri": "at://did:plc:owner/sh.tangled.repo/3abc", "value": {"name": "old", "knot": "knot1.tangled.sh"} }); let repo = repo_from_search_hit(&hit).unwrap(); assert_eq!(repo.repo_did, "did:plc:owner/old"); assert!(is_legacy_repo(&repo.repo_did)); assert_eq!(repo.url, "https://tangled.org/did:plc:owner/old"); let current = json!({ "uri": "at://did:plc:owner/sh.tangled.repo/3abd", "value": {"name": "new", "knot": "knot1.tangled.sh", "repoDid": "did:plc:repo"} }); assert_eq!( repo_from_search_hit(¤t).unwrap().repo_did, "did:plc:repo" ); assert!(!is_legacy_repo("did:plc:repo")); // Without a knot, or with a name that can't be a path segment, a // legacy record can't be addressed. let no_knot = json!({"name": "old"}); assert!(repo_from_record("did:plc:owner", &no_knot, Some("3abc")).is_none()); let bad_name = json!({"name": "a/b", "knot": "knot1.tangled.sh"}); assert!(repo_from_record("did:plc:owner", &bad_name, Some("3abc")).is_none()); // The record key stands in for a missing name. let unnamed = json!({"knot": "knot1.tangled.sh"}); assert_eq!( repo_from_record("did:plc:owner", &unnamed, Some("old")) .unwrap() .repo_did, "did:plc:owner/old" ); }
#[test] fn parses_a_commit_page() { let body = json!({ "total": 1, "commits": [{ "hash": [10, 20], "author": {"Name": "@me", "Email": "did:plc:me", "When": "2026-10-20T12:00:00-04:00"}, "committer": {"When": "2026-10-20T16:00:01Z"}, "message": "subject\n\nbody" }] }); let page = parse_commit_page(&body).unwrap(); assert_eq!(page.commits[0].sha, "0a14"); assert_eq!(page.commits[0].authored_at, "2026-10-20T16:00:00Z"); assert_eq!(page.commits[0].message, "subject"); }}