From ae5e98417e87da4dae21479a07b786a96d8a4e25 Mon Sep 17 00:00:00 2001 From: oppiliappan Date: Thu, 29 Oct 2026 12:12:07 +0000 Subject: [PATCH] bobbin: index org.tangled.repo.artifact and attach it to listTags Signed-off-by: oppiliappan --- bobbin/crates/resolver/src/normalize.rs | 2 + bobbin/crates/types/src/edges.rs | 8 ++ bobbin/crates/types/src/search.rs | 1 + bobbin/crates/xrpc/src/lib.rs | 109 ++++++++++++++++++++++-- bobbin/crates/xrpc/tests/extended.rs | 24 +++--- 5 files changed, 125 insertions(+), 19 deletions(-) diff --git a/bobbin/crates/resolver/src/normalize.rs b/bobbin/crates/resolver/src/normalize.rs index e0d2efcc7..ba662f6f3 100644 --- a/bobbin/crates/resolver/src/normalize.rs +++ b/bobbin/crates/resolver/src/normalize.rs @@ -98,6 +98,7 @@ use bobbin_types::org_tangled::graph::follow::Follow; use bobbin_types::org_tangled::graph::vouch::Vouch; use bobbin_types::org_tangled::key::ssh_key::SshKey; use bobbin_types::org_tangled::label::definition::Definition as LabelDefinition; +use bobbin_types::org_tangled::repo::artifact::Artifact as RepoArtifact; use bobbin_types::org_tangled::repo::declaration::Declaration; use bobbin_types::org_tangled::repo::manifest::Manifest; use bobbin_types::org_tangled::review::patch::Patch; @@ -120,4 +121,5 @@ identity_normalize!( Ticket, Patch, SnippetString, + RepoArtifact, ); diff --git a/bobbin/crates/types/src/edges.rs b/bobbin/crates/types/src/edges.rs index c951e9e14..77111c45f 100644 --- a/bobbin/crates/types/src/edges.rs +++ b/bobbin/crates/types/src/edges.rs @@ -22,6 +22,7 @@ use crate::org_tangled::graph::follow::Follow; use crate::org_tangled::graph::vouch::Vouch; use crate::org_tangled::key::ssh_key::SshKey; use crate::org_tangled::label::definition::Definition as LabelDefinitionRecord; +use crate::org_tangled::repo::artifact::Artifact as RepoArtifact; use crate::org_tangled::repo::declaration::Declaration; use crate::org_tangled::repo::manifest::{Manifest, ManifestSource}; use crate::org_tangled::review::patch::Patch; @@ -62,6 +63,7 @@ pub enum Record { Declaration(Declaration), GitRefs(Refs), Artifact(Artifact), + RepoArtifact(RepoArtifact), Collaborator(Collaborator), Ticket(Ticket), Patch(Patch), @@ -137,6 +139,7 @@ impl Record { "org.tangled.git.refs" => parse!(GitRefs), "org.tangled.graph.collaborator" => parse!(Collaborator), "sh.tangled.repo.artifact" => parse!(Artifact), + "org.tangled.repo.artifact" => parse!(RepoArtifact), "org.tangled.track.ticket" => parse!(Ticket), "org.tangled.review.patch" => parse!(Patch), "org.tangled.snippet.string" => parse!(SnippetString), @@ -160,6 +163,7 @@ impl Record { Self::GitRefs(_) => "org.tangled.git.refs", Self::Collaborator(_) => "org.tangled.graph.collaborator", Self::Artifact(_) => "sh.tangled.repo.artifact", + Self::RepoArtifact(_) => "org.tangled.repo.artifact", Self::Ticket(_) => "org.tangled.track.ticket", Self::Patch(_) => "org.tangled.review.patch", Self::SnippetString(_) => "org.tangled.snippet.string", @@ -209,6 +213,7 @@ impl Record { Self::Repo(r) => Some(&r.created_at), Self::Declaration(r) => Some(&r.created_at), Self::Artifact(r) => Some(&r.created_at), + Self::RepoArtifact(r) => Some(&r.created_at), Self::Collaborator(r) => Some(&r.created_at), Self::Ticket(r) => Some(&r.created_at), Self::Patch(r) => Some(&r.created_at), @@ -221,6 +226,7 @@ impl Record { editor_in(match self { Self::FeedComment(r) => r.extra_data.as_ref(), Self::Reaction(r) => r.extra_data.as_ref(), + Self::RepoArtifact(r) => r.extra_data.as_ref(), _ => None, }) } @@ -233,6 +239,7 @@ impl Record { Self::Reaction(r) => reaction_edges(source, r), Self::Follow(r) => follow_edges(source, r), Self::Artifact(r) => artifact_edges(source, r), + Self::RepoArtifact(_) => Ok(owner_self_edges("org.tangled.repo.artifact", source)), Self::Collaborator(_) => Ok(collaborator_edges(source)), // owned by the repo, so the repo is the authority Self::Ticket(r) => Ok(owner_self_edges(ticket_kind(r), source)), @@ -321,6 +328,7 @@ const MIRROR_KINDS: &[(&str, &str)] = &[ ("org.tangled.graph.follow", "org.tangled.graph.follow.by"), ("org.tangled.graph.vouch", "org.tangled.graph.vouch.by"), ("sh.tangled.repo.artifact", "sh.tangled.repo.artifact.by"), + ("org.tangled.repo.artifact", "org.tangled.repo.artifact.by"), ( "org.tangled.graph.collaborator", "org.tangled.graph.collaborator.by", diff --git a/bobbin/crates/types/src/search.rs b/bobbin/crates/types/src/search.rs index aa1b59ee5..1f75e4ea8 100644 --- a/bobbin/crates/types/src/search.rs +++ b/bobbin/crates/types/src/search.rs @@ -98,6 +98,7 @@ impl SearchableRecord { | Record::Vouch(_) | Record::SshKey(_) | Record::Artifact(_) + | Record::RepoArtifact(_) | Record::Collaborator(_) | Record::Declaration(_) | Record::GitRefs(_) diff --git a/bobbin/crates/xrpc/src/lib.rs b/bobbin/crates/xrpc/src/lib.rs index 4b23ecc18..4f4e2c2cc 100644 --- a/bobbin/crates/xrpc/src/lib.rs +++ b/bobbin/crates/xrpc/src/lib.rs @@ -59,9 +59,9 @@ use bobbin_types::org_tangled::review::patch::{Patch, PatchRecord}; use bobbin_types::org_tangled::snippet::string::{ SnippetString, SnippetStringGetRecordOutput, SnippetStringRecord, }; +use bobbin_types::org_tangled::repo::artifact::{Artifact, ArtifactRecord}; use bobbin_types::org_tangled::track::ticket::{Ticket, TicketGetRecordOutput, TicketRecord}; use bobbin_types::record::RecordBody; -use bobbin_types::sh_tangled::repo::artifact::{Artifact, ArtifactRecord}; use futures::Stream; use futures::stream::{self, StreamExt, TryStreamExt}; @@ -701,7 +701,6 @@ const MIRROR_PROXIED_NSIDS: &[&str] = &[ "org.tangled.temp.git.listBranches", "org.tangled.temp.git.listCommits", "org.tangled.temp.git.listCommitsByRef", - "org.tangled.temp.git.listTags", // TODO: remove legacy endpoint support "sh.tangled.git.temp.formatPatch", "sh.tangled.git.temp.listLanguages", @@ -735,6 +734,7 @@ type ProxyParams = Vec<(String, String)>; fn mirror_routes() -> Router { let reads = register_proxied(Router::new(), MIRROR_PROXIED_NSIDS, proxy_mirror_handler); + let reads = register_proxied(reads, &[LIST_TAGS], list_tags_with_artifacts); register_mirror_writes(reads) } @@ -1255,7 +1255,7 @@ edge_kinds! { "org.tangled.label.definition" => LabelDefinitionRecord, SubjectShape::BareDid; "org.tangled.key.sshKey" => SshKeyRecord, SubjectShape::BareDid; "org.tangled.repo.manifest" => ManifestRecord, SubjectShape::BareDid; - "sh.tangled.repo.artifact" => ArtifactRecord, SubjectShape::BareDid, mirror ArtifactBy; + "org.tangled.repo.artifact" => ArtifactRecord, SubjectShape::BareDid, mirror ArtifactBy; "org.tangled.graph.collaborator" => CollaboratorRecord, SubjectShape::BareDid, mirror CollaboratorBy; "org.tangled.snippet.string" => SnippetStringRecord, SubjectShape::BareDid; "org.tangled.feed.subscription" => SubscriptionRecord, SubjectShape::BareDidOrAnyAtUri; @@ -3675,22 +3675,119 @@ async fn proxy_mirror_handler( params: ProxyParams, nsid: Nsid, ) -> Result { + forward_to_mirror(&state, &headers, socket, ¶ms, &nsid) + .await + .map(upstream_to_axum) +} + +async fn forward_to_mirror( + state: &AppState, + headers: &HeaderMap, + socket: SocketPeer, + params: &ProxyParams, + nsid: &Nsid, +) -> Result { let mirror = state .mirror .as_ref() .ok_or_else(|| XrpcError::UpstreamUnavailable("mirror.url is unset".into()))?; - let allowed = filter_request_headers(&headers, socket, &state.client_address); + let allowed = filter_request_headers(headers, socket, &state.client_address); let forward: Vec<(&str, &str)> = params .iter() .map(|(k, v)| (k.as_str(), v.as_str())) .collect(); mirror - .forward_raw(&nsid, &forward, allowed) + .forward_raw(nsid, &forward, allowed) .await - .map(upstream_to_axum) .map_err(map_proxy_error) } +const LIST_TAGS: &str = "org.tangled.temp.git.listTags"; +const MAX_TAGS_BYTES: usize = 4 * 1024 * 1024; + +async fn list_tags_with_artifacts( + state: AppState, + headers: HeaderMap, + socket: SocketPeer, + params: ProxyParams, + nsid: Nsid, +) -> Result { + let repo = params + .iter() + .find(|(k, _)| k == "repo") + .and_then(|(_, v)| Did::::new_owned(v).ok()); + let upstream = forward_to_mirror(&state, &headers, socket, ¶ms, &nsid).await?; + let Some(repo) = repo.filter(|_| upstream.status().is_success()) else { + return Ok(upstream_to_axum(upstream)); + }; + let mut body: serde_json::Value = read_upstream_json(upstream, MAX_TAGS_BYTES).await?; + let mut artifacts = artifacts_by_oid(&state, repo).await; + if let Some(tags) = body.get_mut("tags").and_then(serde_json::Value::as_array_mut) { + for tag in tags { + let Some(views) = tag + .get("oid") + .and_then(serde_json::Value::as_str) + .and_then(|oid| artifacts.remove(oid)) + else { + continue; + }; + if let (Some(tag), Ok(views)) = (tag.as_object_mut(), serde_json::to_value(views)) { + tag.insert("artifacts".into(), views); + } + } + } + Ok(Json(body).into_response()) +} + +async fn artifacts_by_oid( + state: &AppState, + repo: Did, +) -> std::collections::HashMap>>> { + let kind = ::NSID; + let key = EdgeKey::new(nsid_static(kind), SubjectRef::Did(repo)); + let limit = PageLimit::new(PageLimit::MAX).expect("the maximum is a valid limit"); + let page = state + .edges + .list(&key, PageCursor::Start, limit, SortDir::Desc); + hydrate_record_stream::>( + state, + nsid_static(kind), + page.items, + HitProvenance::Indexed, + ) + .filter_map(|view| async move { view.ok() }) + .fold( + std::collections::HashMap::>>>::new(), + |mut by_oid, view: RecordView>| async move { + by_oid + .entry(AsRef::::as_ref(&view.value.subject.oid).to_owned()) + .or_default() + .push(view); + by_oid + }, + ) + .await +} + +async fn read_upstream_json( + response: ProxyResponse, + limit: usize, +) -> Result { + let mut body = response.into_body_stream(); + let mut bytes = Vec::new(); + while let Some(chunk) = body.next().await { + let chunk = chunk.map_err(|e| XrpcError::UpstreamUnavailable(format!("mirror: {e}")))?; + if bytes.len().saturating_add(chunk.len()) > limit { + return Err(XrpcError::UpstreamUnavailable(format!( + "mirror response is over {limit} bytes" + ))); + } + bytes.extend_from_slice(&chunk); + } + serde_json::from_slice(&bytes) + .map_err(|e| XrpcError::UpstreamUnavailable(format!("mirror: {e}"))) +} + const UNMARK_LIAR_ACCOUNT: &str = "org.tangled.foster.unmarkLiarAccount"; // hydrant checks the service auth: bobbin resolves dids through hydrant, which diff --git a/bobbin/crates/xrpc/tests/extended.rs b/bobbin/crates/xrpc/tests/extended.rs index 1df7bb1ca..77e59ecd7 100644 --- a/bobbin/crates/xrpc/tests/extended.rs +++ b/bobbin/crates/xrpc/tests/extended.rs @@ -24,7 +24,6 @@ use wiremock::matchers::{method, path, query_param}; use wiremock::{Mock, MockServer, ResponseTemplate}; const CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i"; -const TAG_BYTES: &str = "AAAAAAAAAAAAAAAAAAAAAAAAAAA="; const ARTIFACT_LINK: &str = "bafkreigh2akiscaildc7gnvtklbsfhdgwz72eolmpckbqr5ej26byp3uli"; fn at(s: &str) -> AtUri { @@ -155,13 +154,12 @@ fn label_definition_body(name: &str) -> Value { }) } -fn artifact_body(repo_did: &Did, name: &str) -> Value { +fn artifact_body(oid: &str, name: &str) -> Value { json!({ - "$type": "sh.tangled.repo.artifact", + "$type": "org.tangled.repo.artifact", "createdAt": "2026-05-01T00:00:00Z", "name": name, - "repoDid": repo_did.as_ref(), - "tag": {"$bytes": TAG_BYTES}, + "subject": {"$type": "org.tangled.git.oid", "oid": oid}, "artifact": { "$type": "blob", "ref": {"$link": ARTIFACT_LINK}, @@ -267,22 +265,22 @@ async fn list_artifacts_keys_on_repo_did() { let h = Harness::new().await; let repo_did = did("did:plc:abalone"); let subject = at(&format!("at://{}", repo_did.as_ref())); - let owner = did("did:plc:nel"); let rk = rkey("a1"); + let oid = "0123456789abcdef0123456789abcdef01234567"; h.add_edge( - &nsid("sh.tangled.repo.artifact"), + &nsid("org.tangled.repo.artifact"), &subject, &at(&format!( - "at://{}/sh.tangled.repo.artifact/{}", - owner.as_ref(), + "at://{}/org.tangled.repo.artifact/{}", + repo_did.as_ref(), rk.as_ref() )), ); h.mount( - &owner, - &nsid("sh.tangled.repo.artifact"), + &repo_did, + &nsid("org.tangled.repo.artifact"), &rk, - artifact_body(&repo_did, "out.bin"), + artifact_body(oid, "out.bin"), ) .await; @@ -301,7 +299,7 @@ async fn list_artifacts_keys_on_repo_did() { let items = body["items"].as_array().unwrap(); assert_eq!(items.len(), 1); assert_eq!(items[0]["value"]["name"], json!("out.bin")); - assert_eq!(items[0]["value"]["repoDid"], json!(repo_did.as_ref())); + assert_eq!(items[0]["value"]["subject"]["oid"], json!(oid)); } #[tokio::test] -- knotmirror 0.0.1