diff --git a/bobbin/crates/bobbin-sim/src/workloads/cold_start_under_live_load.rs b/bobbin/crates/bobbin-sim/src/workloads/cold_start_under_live_load.rs index 2c3f902c0..635f63197 100644 --- a/bobbin/crates/bobbin-sim/src/workloads/cold_start_under_live_load.rs +++ b/bobbin/crates/bobbin-sim/src/workloads/cold_start_under_live_load.rs @@ -8,7 +8,7 @@ use url::Url; use crate::report::{SimOutcome, SimReport}; use crate::workload::{Workload, WorkloadCtx, WorkloadHooks}; -use crate::workloads::util::{AssertNoSlingshot, format_rkey, format_tid}; +use crate::workloads::util::{AssertNoSlingshot, format_tid}; const NAME: &str = "cold-start-under-live-load"; @@ -145,17 +145,15 @@ fn build_frame_log(replay: usize, live: usize) -> Vec<(bool, String)> { "type": "record", "record": { "live": is_live, - "did": owner, + "did": format!("did:plc:repo-{i}"), "rev": format_tid(i), - "collection": "sh.tangled.repo", - "rkey": format_rkey(i), + "collection": "org.tangled.repo.manifest", + "rkey": "self", "action": "create", "record": { - "$type": "sh.tangled.repo", + "$type": "org.tangled.repo.manifest", "createdAt": "2026-05-01T00:00:00Z", - "knot": "oyster.cafe", - "name": format!("repo-{i}"), - "repoDid": format!("did:plc:repo-{i}") + "declaration": { "owner": owner, "slug": format!("repo-{i}") } } } }) diff --git a/bobbin/crates/bobbin-sim/src/workloads/frame_burst.rs b/bobbin/crates/bobbin-sim/src/workloads/frame_burst.rs index c665dd8a1..7fe8d68ac 100644 --- a/bobbin/crates/bobbin-sim/src/workloads/frame_burst.rs +++ b/bobbin/crates/bobbin-sim/src/workloads/frame_burst.rs @@ -11,7 +11,7 @@ use url::Url; use crate::report::{SimOutcome, SimReport}; use crate::workload::{Workload, WorkloadCtx, WorkloadHooks}; -use crate::workloads::util::{AssertNoSlingshot, format_rkey, format_tid}; +use crate::workloads::util::{AssertNoSlingshot, format_tid}; const NAME: &str = "frame-burst"; @@ -135,17 +135,15 @@ fn build_frame_script(count: usize) -> Vec { "type": "record", "record": { "live": false, - "did": authority.as_ref(), + "did": repo_did.as_ref(), "rev": format_tid(i), - "collection": "sh.tangled.repo", - "rkey": format_rkey(i), + "collection": "org.tangled.repo.manifest", + "rkey": "self", "action": "create", "record": { - "$type": "sh.tangled.repo", + "$type": "org.tangled.repo.manifest", "createdAt": "2026-05-01T00:00:00Z", - "knot": "oyster.cafe", - "name": format!("repo-{i}"), - "repoDid": repo_did.as_ref() + "declaration": { "owner": authority.as_ref(), "slug": format!("repo-{i}") } } } }) diff --git a/bobbin/crates/bobbin-sim/src/workloads/hydrant_disconnect_barrage.rs b/bobbin/crates/bobbin-sim/src/workloads/hydrant_disconnect_barrage.rs index 03a0512ba..554fa8402 100644 --- a/bobbin/crates/bobbin-sim/src/workloads/hydrant_disconnect_barrage.rs +++ b/bobbin/crates/bobbin-sim/src/workloads/hydrant_disconnect_barrage.rs @@ -11,7 +11,7 @@ use url::Url; use crate::report::{SimOutcome, SimReport}; use crate::workload::{Workload, WorkloadCtx, WorkloadHooks}; -use crate::workloads::util::{AssertNoSlingshot, format_rkey, format_tid}; +use crate::workloads::util::{AssertNoSlingshot, format_tid}; const NAME: &str = "hydrant-disconnect-barrage"; @@ -159,17 +159,15 @@ fn build_frame_log(count: usize) -> Vec<(u64, String)> { "type": "record", "record": { "live": false, - "did": authority.as_ref(), + "did": repo_did.as_ref(), "rev": format_tid(i), - "collection": "sh.tangled.repo", - "rkey": format_rkey(i), + "collection": "org.tangled.repo.manifest", + "rkey": "self", "action": "create", "record": { - "$type": "sh.tangled.repo", + "$type": "org.tangled.repo.manifest", "createdAt": "2026-05-01T00:00:00Z", - "knot": "oyster.cafe", - "name": format!("repo-{i}"), - "repoDid": repo_did.as_ref() + "declaration": { "owner": authority.as_ref(), "slug": format!("repo-{i}") } } } }) diff --git a/bobbin/crates/bobbin-sim/src/workloads/slingshot_flap.rs b/bobbin/crates/bobbin-sim/src/workloads/slingshot_flap.rs index 2605a90f5..40d110df0 100644 --- a/bobbin/crates/bobbin-sim/src/workloads/slingshot_flap.rs +++ b/bobbin/crates/bobbin-sim/src/workloads/slingshot_flap.rs @@ -221,17 +221,15 @@ fn build_frame_script(cfg: &SlingshotFlapConfig, started_unix: UnixMicros) -> Ve "type": "record", "record": { "live": false, - "did": owner, + "did": repo_did, "rev": format_tid(cfg.cross_did_stars + i), - "collection": REPO_COLLECTION, - "rkey": format_rkey(i), + "collection": "org.tangled.repo.manifest", + "rkey": "self", "action": "create", "record": { - "$type": REPO_COLLECTION, + "$type": "org.tangled.repo.manifest", "createdAt": "2026-05-01T00:00:00Z", - "knot": "oyster.cafe", - "name": format!("repo-{i}"), - "repoDid": repo_did, + "declaration": { "owner": owner, "slug": format_rkey(i) }, } } }) @@ -258,15 +256,13 @@ fn build_frame_script(cfg: &SlingshotFlapConfig, started_unix: UnixMicros) -> Ve "live": true, "did": promoter_did, "rev": live_rev.as_str(), - "collection": REPO_COLLECTION, - "rkey": promoter_rkey, + "collection": "org.tangled.repo.manifest", + "rkey": "self", "action": "create", "record": { - "$type": REPO_COLLECTION, + "$type": "org.tangled.repo.manifest", "createdAt": "2026-05-01T00:00:00Z", - "knot": "oyster.cafe", - "name": promoter_name, - "repoDid": promoter_did, + "declaration": { "owner": promoter_did, "slug": promoter_rkey }, } } }) diff --git a/bobbin/crates/edge-index/benches/profile_activity.rs b/bobbin/crates/edge-index/benches/profile_activity.rs index 49949c81b..3f138c54f 100644 --- a/bobbin/crates/edge-index/benches/profile_activity.rs +++ b/bobbin/crates/edge-index/benches/profile_activity.rs @@ -14,7 +14,7 @@ const GLOBAL_ACTOR_COUNT: usize = 250_000; const PAGE_SIZE: u32 = 1_000; const RECENT_COUNT: usize = 1_000; const KINDS: [(&str, &str); 3] = [ - ("sh.tangled.repo", "sh.tangled.repo"), + ("org.tangled.repo.manifest", "org.tangled.repo.manifest"), ("sh.tangled.repo.issue.by", "sh.tangled.repo.issue"), ("sh.tangled.repo.pull.by", "sh.tangled.repo.pull"), ]; @@ -52,10 +52,13 @@ fn globally_populated_store() -> (EdgeStore, Vec) { for index in 0..GLOBAL_ACTOR_COUNT { let actor = did(&format!("did:plc:actor{index}")); store.add(Edge { - kind: nsid_static("sh.tangled.repo"), + kind: nsid_static("org.tangled.repo.manifest"), subject: SubjectRef::Did(actor.clone()), - source: AtUri::new_owned(format!("at://{}/sh.tangled.repo/repo", actor.as_ref(),)) - .expect("valid at-uri"), + source: AtUri::new_owned(format!( + "at://{}/org.tangled.repo.manifest/self", + actor.as_ref(), + )) + .expect("valid at-uri"), sort_micros: index as u64, }); } diff --git a/bobbin/crates/ingest/benches/json_decode.rs b/bobbin/crates/ingest/benches/json_decode.rs index b651ac01e..ce4cfe0a2 100644 --- a/bobbin/crates/ingest/benches/json_decode.rs +++ b/bobbin/crates/ingest/benches/json_decode.rs @@ -96,23 +96,17 @@ fn repo_text() -> &'static str { "type": "record", "record": { "live": false, - "did": "did:plc:nel", + "did": "did:plc:limpet", "rev": "3lq2zk5wqsh2l", - "collection": "sh.tangled.repo", - "rkey": "3lq2zk5wq0001", + "collection": "org.tangled.repo.manifest", + "rkey": "self", "action": "create", "record": { - "$type": "sh.tangled.repo", + "$type": "org.tangled.repo.manifest", "createdAt": "2026-05-01T00:00:00Z", - "knot": "knot.witchcraft.systems", - "name": "limpet", + "declaration": { "owner": "did:plc:nel", "slug": "limpet" }, "description": "demonstration repository for the bench corpus", - "owner": "did:plc:nel", - "repoDid": "did:plc:limpet", - "labels": [ - "at://did:plc:periwinkle/sh.tangled.label.definition/3lq2zk5wq0010", - "at://did:plc:periwinkle/sh.tangled.label.definition/3lq2zk5wq0011" - ] + "topics": ["bench", "corpus"] } } }"# diff --git a/bobbin/crates/ingest/src/analytics.rs b/bobbin/crates/ingest/src/analytics.rs index 5554db3f2..e405a0bd5 100644 --- a/bobbin/crates/ingest/src/analytics.rs +++ b/bobbin/crates/ingest/src/analytics.rs @@ -190,17 +190,14 @@ fn upsert_event( match record { Record::Repo(repo) => { if let Some(old) = supersedes { - let mut entries = vec![ - prop("repo_at", source.as_ref()), - prop("owner", old.owner.as_ref()), - prop("new_name", repo.name.as_deref().unwrap_or_default()), - ]; - if let Some(old_name) = renamed_from_name(old, records) { - entries.push(prop("old_name", old_name)); - } Some(Mapped { name: "repo_renamed", - properties: props(entries), + properties: props([ + prop("repo_at", source.as_ref()), + prop("owner", old.owner.as_ref()), + prop("new_name", repo.declaration.slug.0.as_str()), + prop("old_name", old.rkey.as_ref()), + ]), }) } else if existed(source) { None @@ -208,7 +205,7 @@ fn upsert_event( Some(Mapped { name: "new_repo", properties: props([ - prop("repo", repo.name.as_deref().unwrap_or_default()), + prop("repo", repo.declaration.slug.0.as_str()), prop("repo_at", source.as_ref()), ]), }) @@ -348,21 +345,6 @@ fn cached_record(source: &AtUri, nsid: &str, records: &dyn RecordStore) -> Optio Some(decoded) } -fn renamed_from_name(old: &RepoIdent, records: &dyn RecordStore) -> Option { - let old_source = - AtUri::from_parts_owned(old.owner.as_ref(), "sh.tangled.repo", old.rkey.as_ref()).ok()?; - let body = records.get(&old_source)?; - let decoded = Record::from_json_bytes( - &Nsid::::new_static("sh.tangled.repo").expect("literal nsid"), - &body.value, - ) - .ok()?; - match decoded { - Record::Repo(repo) => repo.name.as_deref().map(str::to_owned), - _ => None, - } -} - fn rkey_str(source: &AtUri) -> String { source .rkey() @@ -557,42 +539,26 @@ mod tests { } #[test] - fn rename_fires_with_displaced_name_from_cache() { - let store = MapStore::default(); - seed( - &store, - "at://did:plc:dawn/sh.tangled.repo/oldrkeyabcde", - json!({ - "$type": "sh.tangled.repo", "createdAt": "2026-05-01T00:00:00Z", - "name": "old-name", "knot": "knot.example", - }), - ); + fn rename_fires_with_displaced_slug() { let mut op = upsert( - "at://did:plc:dawn/sh.tangled.repo/newrkeyabcde", - "sh.tangled.repo", + "at://did:plc:abalone/org.tangled.repo.manifest/self", + "org.tangled.repo.manifest", json!({ - "$type": "sh.tangled.repo", "createdAt": "2026-05-02T00:00:00Z", - "name": "new-name", "knot": "knot.example", + "$type": "org.tangled.repo.manifest", "createdAt": "2026-05-01T00:00:00Z", + "declaration": {"owner": DAWN, "slug": "new-name"}, }), ); op.supersedes = Some(RepoIdent::new( Did::new_static(DAWN).unwrap(), - "oldrkeyabcde".parse().unwrap(), + "old-name".parse().unwrap(), )); - let Some((name, props)) = mapped(&op, &store) else { + let Some((name, props)) = mapped(&op, &MapStore::default()) else { panic!() }; assert_eq!(name, "repo_renamed"); assert_eq!(prop_str(&props, "old_name"), "old-name"); assert_eq!(prop_str(&props, "new_name"), "new-name"); assert_eq!(prop_str(&props, "owner"), DAWN); - - let store = MapStore::default(); - let Some((name, props)) = mapped(&op, &store) else { - panic!() - }; - assert_eq!(name, "repo_renamed"); - assert!(props.get("old_name").is_none()); } #[test] diff --git a/bobbin/crates/ingest/src/lib.rs b/bobbin/crates/ingest/src/lib.rs index 8f6fa9747..5e709eedb 100644 --- a/bobbin/crates/ingest/src/lib.rs +++ b/bobbin/crates/ingest/src/lib.rs @@ -19,10 +19,10 @@ use bobbin_runtime::{ WsTransport, }; use bobbin_types::edges::{Edge, ExtractError, Record}; -use bobbin_types::ids::{RepoIdent, SubjectRef}; +use bobbin_types::ids::{RepoIdent, SubjectRef, owner_did_from_aturi}; +use bobbin_types::org_tangled::repo::manifest::Manifest; use bobbin_types::record::RecordBody; use bobbin_types::search::{SearchSink, SearchableRecord}; -use bobbin_types::sh_tangled::repo::Repo as RepoRecord; use bytes::Bytes; use futures::StreamExt; use jacquard_common::DefaultStr; @@ -976,7 +976,7 @@ enum PendingOp { }, } -const STALE_RENAME_ALIAS: &str = "a newer sh.tangled.repo record owns this repoDid"; +const STALE_RENAME_ALIAS: &str = "a newer org.tangled.repo.manifest owns this repoDid"; struct Prepared { pending: Pending, @@ -1068,8 +1068,9 @@ async fn claim_pending( PendingOp::Rejected { source, .. } => evict_from_buffer(ctx.buffer, source).await, PendingOp::Delete { source, nsid, .. } => { evict_from_buffer(ctx.buffer, source).await; - if nsid.as_ref() == "sh.tangled.repo" - && let Some(ident) = repo_ident_of(source) + if nsid.as_ref() == "org.tangled.repo.manifest" + && let Some(repo_did) = owner_did_from_aturi(source) + && let Some(ident) = ctx.resolver.lookup_by_repo_did(&repo_did).await { ctx.resolver.forget(&ident.owner, &ident.rkey).await; } @@ -1088,12 +1089,16 @@ async fn claim_pending( async fn claim_repo( source: &AtUri, - repo: &RepoRecord, + repo: &Manifest, ctx: &PipelineCtx<'_, S>, ) -> RepoClaim { - let Some(ident) = repo_ident_of(source) else { + let Some(repo_did) = owner_did_from_aturi(source) else { return RepoClaim::Current { displaced: None }; }; + let ident = RepoIdent::new( + repo.declaration.owner.clone(), + repo.declaration.slug.0.clone(), + ); if let Some(shadow) = ctx.shadow { shadow.note_observed(&ident.owner, &ident.rkey).await; } @@ -1102,18 +1107,11 @@ async fn claim_repo( .observe( ident.owner.clone(), ident.rkey.clone(), - repo.repo_did.clone(), - repo.name.clone(), + Some(repo_did), + None, &repo.created_at, ) .await; - if let RepoClaim::Current { - displaced: Some(prior), - } = &claim - && let Some(prior_uri) = repo_ident_uri(prior) - { - evict_from_buffer(ctx.buffer, &prior_uri).await; - } finalize_drained(ctx, take_observed(ctx.buffer, &ident).await).await; claim } @@ -1506,20 +1504,6 @@ async fn collect_unresolved_deps(edges: &[Edge], resolver: &RepoIdResolver) -> V .await } -async fn remove_superseded_source( - store: &EdgeStore, - records: &dyn RecordStore, - search: &S, - prior: Option<&RepoIdent>, -) { - let Some(prior_uri) = prior.and_then(repo_ident_uri) else { - return; - }; - store.remove_source(&prior_uri); - records.remove(&prior_uri); - search.remove(&prior_uri).await; -} - async fn finalize_drained( ctx: &PipelineCtx<'_, S>, drained: Vec, @@ -1549,7 +1533,6 @@ async fn finalize_drained( ); } let edges = normalize_subjects(edges, ctx.resolver, ctx.coverage, None).await; - remove_superseded_source(ctx.store, ctx.records, ctx.search, supersedes.as_ref()).await; let settled_cid = settle_cid(cid.as_ref()); cache_body(ctx.records, &source, cid, bytes); let outcome = upsert_record_indexes( @@ -1639,9 +1622,8 @@ async fn commit_pending( cid, rev: _, edges, - supersedes, + supersedes: _, } = *pieces; - remove_superseded_source(store, records, search, supersedes.as_ref()).await; let settled_cid = settle_cid(cid.as_ref()); cache_body(records, &source, cid, bytes); let outcome = @@ -1843,14 +1825,6 @@ fn parse_repo_subject_uri(uri: &AtUri) -> Option<(Did, R Some((owner, rkey)) } -fn repo_ident_uri(ident: &RepoIdent) -> Option> { - AtUri::from_parts_owned(ident.owner.as_ref(), "sh.tangled.repo", ident.rkey.as_ref()).ok() -} - -fn repo_ident_of(uri: &AtUri) -> Option { - parse_repo_subject_uri(uri).map(|(owner, rkey)| RepoIdent::new(owner, rkey)) -} - fn build_source_uri(r: &RecordFrame) -> Result, IngestError> { Ok(AtUri::from_parts_owned( r.did.as_ref(), @@ -3505,161 +3479,6 @@ mod tests { }); } - #[tokio::test] - async fn out_of_order_repo_preps_cannot_steal_a_rename() { - let (store, issue_states, pull_statuses, cov, resolver) = fresh(); - let identity = detached_identity(); - let search = NoopSearchSink; - let records = NoopRecordStore; - let ctx = PipelineCtx { - resolver: &resolver, - identity: &identity, - store: &store, - issue_states: &issue_states, - pull_statuses: &pull_statuses, - coverage: &cov, - records: &records, - search: &search, - shadow: None, - buffer: None, - settlements: None, - analytics: None, - }; - let repo_frame = |id: u64, rkey: &str| { - parse_frame(json!({ - "id": id, - "type": "record", - "record": { - "live": false, - "did": "did:plc:bnuy", - "rev": fresh_tid().as_str(), - "collection": "sh.tangled.repo", - "rkey": rkey, - "action": "create", - "record": { - "$type": "sh.tangled.repo", - "createdAt": "2026-05-01T00:00:00Z", - "knot": "oyster.cafe", - "repoDid": "did:plc:abalone" - } - } - })) - }; - // prepped out of order on purpose to replicate, cursor 2 will finish before 1 - let new_pending = prepare_frame(repo_frame(2, "newrename"), &ctx, now()).await; - let old_pending = prepare_frame(repo_frame(1, "oldrename"), &ctx, now()).await; - let old_pending = claim_pending(old_pending, &ctx).await; - let new_pending = claim_pending(new_pending, &ctx).await; - let old_pending = resolve_pending(old_pending, &ctx, None).await; - commit_pending(old_pending, &ctx).await; - let new_pending = resolve_pending(new_pending, &ctx, None).await; - commit_pending(new_pending, &ctx).await; - - let owner = Did::new_owned("did:plc:bnuy").unwrap(); - assert_eq!( - resolver - .lookup_by_repo_did(&Did::new_owned("did:plc:abalone").unwrap()) - .await, - Some(RepoIdent::new(owner.clone(), rkey("newrename"))), - "newrename has the higher cursor so it owns the repoDid. oldrename prepped \ - second, it would win if prep order decided this", - ); - let key = bobbin_types::ids::EdgeKey::new( - Nsid::new_static("sh.tangled.repo").unwrap(), - did_subj("did:plc:bnuy"), - ); - assert_eq!( - store.sources_for(&key), - vec![AtUri::new_owned("at://did:plc:bnuy/sh.tangled.repo/newrename").unwrap(),], - "the retained old alias must be evicted when the renamed record commits", - ); - } - - #[tokio::test] - async fn a_backfilled_rename_alias_cannot_steal_the_repo_did() { - let (store, issue_states, pull_statuses, cov, resolver) = fresh(); - let identity = detached_identity(); - let search = NoopSearchSink; - let records = NoopRecordStore; - let ctx = PipelineCtx { - resolver: &resolver, - identity: &identity, - store: &store, - issue_states: &issue_states, - pull_statuses: &pull_statuses, - coverage: &cov, - records: &records, - search: &search, - shadow: None, - buffer: None, - settlements: None, - analytics: None, - }; - let repo_frame = |id: u64, rkey: &str, created_at: &str| { - parse_frame(json!({ - "id": id, - "type": "record", - "record": { - "live": false, - "did": "did:plc:oppi", - "rev": fresh_tid().as_str(), - "collection": "sh.tangled.repo", - "rkey": rkey, - "action": "create", - "record": { - "$type": "sh.tangled.repo", - "createdAt": created_at, - "knot": "knot1.tangled.sh", - "repoDid": "did:plc:abalone" - } - } - })) - }; - // a backfill replays a repo in rkey order, so the alias trails what replaced it - let mut settled = Vec::new(); - for frame in [ - repo_frame(1, "stinkpot", "2026-07-24T14:33:33Z"), - repo_frame(2, "tortu", "2026-07-20T19:26:39Z"), - ] { - let pending = prepare_frame(frame, &ctx, now()).await; - let pending = claim_pending(pending, &ctx).await; - let pending = resolve_pending(pending, &ctx, None).await; - settled.push(commit_pending(pending, &ctx).await); - } - - let owner = Did::new_owned("did:plc:oppi").unwrap(); - let stinkpot = RepoIdent::new(owner.clone(), rkey("stinkpot")); - assert_eq!( - resolver - .lookup_by_repo_did(&Did::new_owned("did:plc:abalone").unwrap()) - .await, - Some(stinkpot.clone()), - "the newest record owns the repoDid, whichever rkey the replay ends on", - ); - assert_eq!( - resolver.lookup_by_name(&owner, "tortu").await, - Some(stinkpot), - "the old name has to answer with the repo as it stands now", - ); - let key = bobbin_types::ids::EdgeKey::new( - Nsid::new_static("sh.tangled.repo").unwrap(), - did_subj("did:plc:oppi"), - ); - assert_eq!( - store.sources_for(&key), - vec![AtUri::new_owned("at://did:plc:oppi/sh.tangled.repo/stinkpot").unwrap()], - "only the renamed repo belongs in the owner's index", - ); - assert_eq!( - settled.pop().flatten().map(|s| s.outcome), - Some(RecordOutcome::rejected( - Rejection::NotIndexed, - Some(STALE_RENAME_ALIAS) - )), - "the alias is refused on purpose, so a writer polling for it is told", - ); - } - #[tokio::test] async fn unresolvable_repo_subject_drops_edge() { let (store, issue_states, pull_statuses, cov, resolver) = fresh(); @@ -3713,91 +3532,6 @@ mod tests { ); } - #[tokio::test] - async fn repo_without_repo_did_drops_edge() { - let (store, issue_states, pull_statuses, cov, resolver) = fresh(); - let repo: HydrantFrame = parse_frame(json!({ - "id": 1, - "type": "record", - "record": { - "live": false, - "did": "did:plc:nel", - "rev": fresh_tid().as_str(), - "collection": "sh.tangled.repo", - "rkey": "abcabcabcabcz", - "action": "create", - "record": { - "$type": "sh.tangled.repo", - "createdAt": "2026-05-01T00:00:00Z", - "knot": "oyster.cafe", - "name": "abalone" - } - } - })); - handle_frame( - repo, - &store, - &issue_states, - &pull_statuses, - &cov, - &NoopSearchSink, - &NoopRecordStore, - &resolver, - &sys_clock(), - now(), - ) - .await; - - let star: HydrantFrame = parse_frame(json!({ - "id": 2, - "type": "record", - "record": { - "live": false, - "did": "did:plc:olaren", - "rev": fresh_tid().as_str(), - "collection": "sh.tangled.repo.issue", - "rkey": "abcabcabcabcz", - "action": "create", - "record": { - "$type": "sh.tangled.repo.issue", - "createdAt": "2026-05-01T00:00:00Z", - "title": "bug", - "repo": "at://did:plc:nel/sh.tangled.repo/abcabcabcabcz" - } - } - })); - handle_frame( - star, - &store, - &issue_states, - &pull_statuses, - &cov, - &NoopSearchSink, - &NoopRecordStore, - &resolver, - &sys_clock(), - now(), - ) - .await; - - let nsid = Nsid::new_static("sh.tangled.repo.issue").unwrap(); - let uri_keyed = bobbin_types::ids::EdgeKey::new( - nsid.clone(), - uri_subj("at://did:plc:nel/sh.tangled.repo/abcabcabcabcz"), - ); - let owner_keyed = bobbin_types::ids::EdgeKey::new(nsid, did_subj("did:plc:nel")); - assert_eq!( - store.count(&uri_keyed), - 0, - "no canonical DID exists for a repo without repoDID, so the edge must be dropped", - ); - assert_eq!( - store.count(&owner_keyed), - 0, - "the authoring DID is not the canonical repo identity", - ); - } - #[tokio::test] async fn explicit_subject_did_skips_normalization() { let (store, issue_states, pull_statuses, cov, resolver) = fresh(); @@ -3962,28 +3696,10 @@ mod tests { } #[tokio::test] - async fn repo_record_indexes_its_rkey_and_name() { + async fn manifest_indexes_its_owner_and_slug() { let (store, issue_states, pull_statuses, cov, resolver) = fresh(); - let repo: HydrantFrame = parse_frame(json!({ - "id": 1, - "type": "record", - "record": { - "live": false, - "did": "did:plc:nel", - "rev": fresh_tid().as_str(), - "collection": "sh.tangled.repo", - "rkey": "abcabcabcabcz", - "action": "create", - "record": { - "$type": "sh.tangled.repo", - "createdAt": "2026-05-01T00:00:00Z", - "knot": "oyster.cafe", - "name": "abalone" - } - } - })); handle_frame( - repo, + manifest_frame("did:plc:abalone", json!({})), &store, &issue_states, &pull_statuses, @@ -3996,134 +3712,62 @@ mod tests { ) .await; let owner = Did::new_owned("did:plc:nel").unwrap(); + let ident = + bobbin_types::ids::RepoIdent::new(owner.clone(), Rkey::new_owned("abalone").unwrap()); assert_eq!( - resolver.lookup_by_name(&owner, "abcabcabcabcz").await, - Some(bobbin_types::ids::RepoIdent::new( - owner.clone(), - Rkey::new_owned("abcabcabcabcz").unwrap() - )), + resolver.lookup_by_name(&owner, "abalone").await, + Some(ident.clone()) ); assert_eq!( - resolver.lookup_by_name(&owner, "abalone").await, - Some(bobbin_types::ids::RepoIdent::new( - owner.clone(), - Rkey::new_owned("abcabcabcabcz").unwrap() - )), - "the record's cosmetic name resolves to its rkey ident", + resolver + .lookup_by_repo_did(&Did::new_owned("did:plc:abalone").unwrap()) + .await, + Some(ident), ); - } - - #[tokio::test] - async fn fork_record_edges_against_its_source() { - let (store, issue_states, pull_statuses, cov, resolver) = fresh(); - let fork: HydrantFrame = parse_frame(json!({ - "id": 1, - "type": "record", - "record": { - "live": false, - "did": "did:plc:olaren", - "rev": fresh_tid().as_str(), - "collection": "sh.tangled.repo", - "rkey": "abcabcabcabcz", - "action": "create", - "record": { - "$type": "sh.tangled.repo", - "createdAt": "2026-05-01T00:00:00Z", - "knot": "oyster.cafe", - "name": "abalone", - "source": "did:plc:abalone" - } - } - })); - handle_frame( - fork, - &store, - &issue_states, - &pull_statuses, - &cov, - &NoopSearchSink, - &NoopRecordStore, - &resolver, - &sys_clock(), - now(), - ) - .await; let key = bobbin_types::ids::EdgeKey::new( - Nsid::new_static(bobbin_types::edges::REPO_SOURCE_EDGE_KIND).unwrap(), - did_subj("did:plc:abalone"), + Nsid::new_static("org.tangled.repo.manifest").unwrap(), + did_subj("did:plc:nel"), + ); + assert_eq!( + store.sources_for(&key), + vec![AtUri::new_owned("at://did:plc:abalone/org.tangled.repo.manifest/self").unwrap()], + "the repo lists under its declared owner, not the repo account", ); - assert_eq!(store.count(&key), 1); } - fn fork_frame(source: &str) -> HydrantFrame { + fn manifest_frame(repo: &str, extra: serde_json::Value) -> HydrantFrame { + let mut record = json!({ + "$type": "org.tangled.repo.manifest", + "createdAt": "2026-05-01T00:00:00Z", + "declaration": { "owner": "did:plc:nel", "slug": "abalone" } + }); + record + .as_object_mut() + .unwrap() + .extend(extra.as_object().unwrap().clone()); parse_frame(json!({ "id": 1, "type": "record", "record": { "live": false, - "did": "did:plc:olaren", + "did": repo, "rev": fresh_tid().as_str(), - "collection": "sh.tangled.repo", - "rkey": "abcabcabcabcz", + "collection": "org.tangled.repo.manifest", + "rkey": "self", "action": "create", - "record": { - "$type": "sh.tangled.repo", - "createdAt": "2026-05-01T00:00:00Z", - "knot": "oyster.cafe", - "name": "abalone", - "source": source - } + "record": record } })) } - fn fork_edge_key(subject: SubjectRef) -> bobbin_types::ids::EdgeKey { - bobbin_types::ids::EdgeKey::new( - Nsid::new_static(bobbin_types::edges::REPO_SOURCE_EDGE_KIND).unwrap(), - subject, - ) - } - - #[tokio::test] - async fn fork_by_at_uri_edges_against_the_source_did() { - let (store, issue_states, pull_statuses, cov, resolver) = fresh(); - resolver - .observe( - Did::new_owned("did:plc:nel").unwrap(), - Rkey::new_owned("core").unwrap(), - Some(Did::new_owned("did:plc:abalone").unwrap()), - None, - &Datetime::raw_str("2026-05-01T00:00:00Z"), - ) - .await; - handle_frame( - fork_frame("at://did:plc:nel/sh.tangled.repo/core"), - &store, - &issue_states, - &pull_statuses, - &cov, - &NoopSearchSink, - &NoopRecordStore, - &resolver, - &sys_clock(), - now(), - ) - .await; - assert_eq!(store.count(&fork_edge_key(did_subj("did:plc:abalone"))), 1); - assert_eq!( - store.count(&fork_edge_key(uri_subj( - "at://did:plc:nel/sh.tangled.repo/core" - ))), - 0, - "the rkey form would never match a bare-DID query", - ); - } - #[tokio::test] - async fn fork_of_a_repo_without_a_did_has_nothing_to_count_against() { + async fn fork_record_edges_against_its_source() { let (store, issue_states, pull_statuses, cov, resolver) = fresh(); handle_frame( - fork_frame("at://did:plc:nel/sh.tangled.repo/core"), + manifest_frame( + "did:plc:olaren", + json!({ "upstream": { "$type": "org.tangled.repo.defs#sourceTangled", "did": "did:plc:abalone" } }), + ), &store, &issue_states, &pull_statuses, @@ -4135,36 +3779,20 @@ mod tests { now(), ) .await; - assert_eq!( - store.count(&fork_edge_key(uri_subj( - "at://did:plc:nel/sh.tangled.repo/core" - ))), - 0, + let key = bobbin_types::ids::EdgeKey::new( + Nsid::new_static(bobbin_types::edges::REPO_SOURCE_EDGE_KIND).unwrap(), + did_subj("did:plc:abalone"), ); + assert_eq!(store.count(&key), 1); } #[tokio::test] - async fn repo_record_without_a_name_indexes_its_rkey() { + async fn delete_repo_record_evicts_resolver_cache() { let (store, issue_states, pull_statuses, cov, resolver) = fresh(); - let repo: HydrantFrame = parse_frame(json!({ - "id": 1, - "type": "record", - "record": { - "live": false, - "did": "did:plc:nel", - "rev": fresh_tid().as_str(), - "collection": "sh.tangled.repo", - "rkey": "abalone", - "action": "create", - "record": { - "$type": "sh.tangled.repo", - "createdAt": "2026-05-01T00:00:00Z", - "knot": "oyster.cafe" - } - } - })); + let owner = Did::new_owned("did:plc:nel").unwrap(); + let rkey = Rkey::new_owned("abalone").unwrap(); handle_frame( - repo, + manifest_frame("did:plc:abalone", json!({})), &store, &issue_states, &pull_statuses, @@ -4176,45 +3804,19 @@ mod tests { now(), ) .await; - let owner = Did::new_owned("did:plc:nel").unwrap(); - assert_eq!( - resolver.lookup_by_name(&owner, "abalone").await, - Some(bobbin_types::ids::RepoIdent::new( - owner, - Rkey::new_owned("abalone").unwrap() - )), - "repos made before the name field are only reachable by rkey", - ); - } - - #[tokio::test] - async fn delete_repo_record_evicts_resolver_cache() { - let (store, issue_states, pull_statuses, cov, resolver) = fresh(); - let owner = Did::new_owned("did:plc:nel").unwrap(); - let rkey = Rkey::new_owned("abcabcabcabcz").unwrap(); - resolver - .observe( - owner.clone(), - rkey.clone(), - Some(Did::new_owned("did:plc:abalone").unwrap()), - None, - &Datetime::raw_str("2026-05-01T00:00:00Z"), - ) - .await; - resolver.observe_rkey(owner.clone(), rkey.clone()).await; assert!( resolver.cached_resolution(&owner, &rkey).await.is_some(), "observe must seed the cache", ); let delete: HydrantFrame = parse_frame(json!({ - "id": 1, + "id": 2, "type": "record", "record": { "live": false, - "did": owner.as_ref(), + "did": "did:plc:abalone", "rev": fresh_tid().as_str(), - "collection": "sh.tangled.repo", - "rkey": rkey.as_ref(), + "collection": "org.tangled.repo.manifest", + "rkey": "self", "action": "delete" } })); @@ -4233,10 +3835,10 @@ mod tests { .await; assert!( resolver.cached_resolution(&owner, &rkey).await.is_none(), - "deleting the repo record must clear the resolver cache so future observes are not blocked by a stale Authoritative entry", + "deleting the manifest must clear the resolver cache so future observes are not blocked by a stale Authoritative entry", ); assert_eq!( - resolver.lookup_by_name(&owner, "abcabcabcabcz").await, + resolver.lookup_by_name(&owner, "abalone").await, None, "a deleted repo must stop answering on its url", ); @@ -4639,18 +4241,16 @@ mod tests { "type": "record", "record": { "live": true, - "did": "did:plc:olaren", + "did": "did:plc:hydrant", "rev": fresh_tid().as_str(), - "collection": "sh.tangled.repo", - "rkey": "reporkeyabcde", + "collection": "org.tangled.repo.manifest", + "rkey": "self", "action": "create", "cid": VALID_CID, "record": { - "$type": "sh.tangled.repo", + "$type": "org.tangled.repo.manifest", "createdAt": "2026-05-01T00:00:00Z", - "name": "hydrant", - "knot": "knot.example", - "owner": "did:plc:olaren", + "declaration": { "owner": "did:plc:olaren", "slug": "reporkeyabcde" }, } } })); diff --git a/bobbin/crates/resolver/src/legacy_upgrade.rs b/bobbin/crates/resolver/src/legacy_upgrade.rs index cc475839b..bd8fc0d90 100644 --- a/bobbin/crates/resolver/src/legacy_upgrade.rs +++ b/bobbin/crates/resolver/src/legacy_upgrade.rs @@ -3,11 +3,10 @@ use bobbin_types::com_atproto::repo::strong_ref::StrongRef; use bobbin_types::edges::{ExtractError, Record}; use bobbin_types::legacy::{ LEGACY_COMMENT_SENTINEL_CID, LegacyCollaborator, LegacyIssue, LegacyIssueComment, LegacyPull, - LegacyPullComment, LegacyRecord, LegacyRepo, LegacySource, LegacyTarget, + LegacyPullComment, LegacyRecord, LegacySource, LegacyTarget, }; use bobbin_types::sh_tangled::feed::comment::Comment as FeedComment; use bobbin_types::sh_tangled::markup::markdown::Markdown; -use bobbin_types::sh_tangled::repo::Repo; use bobbin_types::sh_tangled::repo::collaborator::Collaborator; use bobbin_types::sh_tangled::repo::issue::Issue; use bobbin_types::sh_tangled::repo::pull::{Pull, Round, Source, Target}; @@ -199,9 +198,8 @@ fn serialize_canon_variant(record: &Record) -> Result, serde Record::Issue(r) => serde_json::to_vec(r), Record::Pull(r) => serde_json::to_vec(r), Record::Collaborator(r) => serde_json::to_vec(r), - Record::Repo(r) => serde_json::to_vec(r), _ => { - unreachable!("upgrade only produces FeedComment/Issue/Pull/Collaborator/Repo") + unreachable!("upgrade only produces FeedComment/Issue/Pull/Collaborator") } } } @@ -215,7 +213,6 @@ pub async fn upgrade(legacy: LegacyRecord, resolver: &RepoIdResolver) -> Option< LegacyRecord::Collaborator(l) => upgrade_collaborator(l, resolver) .await .map(Record::Collaborator), - LegacyRecord::Repo(l) => Some(Record::Repo(upgrade_repo(l))), } } @@ -373,23 +370,6 @@ async fn upgrade_pull( }) } -fn upgrade_repo(l: LegacyRepo) -> Repo { - let _ = l.owner; - Repo { - created_at: l.added_at, - description: l.description, - knot: l.knot, - labels: None, - name: l.name, - repo_did: None, - source: None, - spindle: None, - topics: None, - website: None, - extra_data: l.extra_data, - } -} - async fn upgrade_collaborator( l: LegacyCollaborator, resolver: &RepoIdResolver, @@ -753,24 +733,6 @@ mod tests { } } - #[tokio::test] - async fn upgrade_repo_renames_added_at_and_drops_owner() { - let resolver = RepoIdResolver::detached(RuntimeHasher::default()); - let json = br#"{"$type":"sh.tangled.repo","addedAt":"2025-03-21T10:18:58Z","description":"hi","knot":"knot1.tangled.sh","name":"site","owner":"did:plc:nel"}"#; - let legacy = LegacyRecord::from_json_bytes(&nsid("sh.tangled.repo"), json).expect("decode"); - let canon = upgrade(legacy, &resolver).await.expect("upgrade"); - match canon { - Record::Repo(r) => { - assert_eq!(r.created_at.as_str(), "2025-03-21T10:18:58Z"); - assert_eq!(r.description.as_deref(), Some("hi")); - assert_eq!(r.knot.as_str(), "knot1.tangled.sh"); - assert_eq!(r.name.as_deref(), Some("site")); - assert!(r.repo_did.is_none(), "legacy repos have no repo_did"); - } - other => panic!("expected canon repo, got {other:?}"), - } - } - #[tokio::test] async fn legacy_pull_target_with_empty_repo_did_treats_as_none() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); @@ -793,17 +755,6 @@ mod tests { } } - #[tokio::test] - async fn try_decode_recovers_repo_with_added_at_field() { - let json = br#"{"$type":"sh.tangled.repo","addedAt":"2025-03-21T10:18:58Z","knot":"knot1.tangled.sh","owner":"did:plc:nel"}"#; - let decoded = DecodedRecord::try_decode(&nsid("sh.tangled.repo"), json) - .expect("legacy repo must decode"); - assert!(matches!( - decoded, - DecodedRecord::Legacy(l) if matches!(*l, LegacyRecord::Repo(_)) - )); - } - #[tokio::test] async fn upgrade_pull_source_repo_resolution_is_independent_of_target() { let resolver = RepoIdResolver::detached(RuntimeHasher::default()); diff --git a/bobbin/crates/resolver/src/lib.rs b/bobbin/crates/resolver/src/lib.rs index 29bad13b6..f46690f3c 100644 --- a/bobbin/crates/resolver/src/lib.rs +++ b/bobbin/crates/resolver/src/lib.rs @@ -21,7 +21,6 @@ use tokio::time::Instant; use bobbin_runtime::{Clock, RuntimeHasher}; use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; -use bobbin_types::edges::{ExtractError, Record}; use bobbin_types::ids::{RepoIdent, nsid_static}; use jacquard_common::DefaultStr; use jacquard_common::types::datetime::Datetime; @@ -435,7 +434,7 @@ impl RepoIdResolver { let started = probe.clock.now_instant(); let nsid: Nsid = nsid_static(REPO_COLLECTION); let provisional = match probe.client.get_record(owner, &nsid, rkey).await { - Ok(body) => match repo_did_from_body(&nsid, &body.value) { + Ok(body) => match repo_did_from_body(&body.value) { Ok(Some(did)) => Resolution::Mapped(did), Ok(None) => Resolution::NoRepoDid, Err(e) => { @@ -503,17 +502,14 @@ fn is_garbage_response(err: &SlingshotError) -> bool { ) } -fn repo_did_from_body( - nsid: &Nsid, - body: &[u8], -) -> Result>, ExtractError> { - match DecodedRecord::try_decode(nsid, body)? { - DecodedRecord::Canon(record) => match *record { - Record::Repo(repo) => Ok(repo.repo_did), - _ => Ok(None), - }, - DecodedRecord::Legacy(_) => Ok(None), +// legacy at-uri refs still point at sh.tangled.repo, and only its repoDid matters here +fn repo_did_from_body(body: &[u8]) -> Result>, serde_json::Error> { + #[derive(serde::Deserialize)] + struct RepoDid { + #[serde(rename = "repoDid")] + repo_did: Option>, } + serde_json::from_slice::(body).map(|r| r.repo_did) } #[cfg(test)] @@ -1116,38 +1112,6 @@ mod tests { ); } - #[tokio::test] - async fn slingshot_unparseable_repo_value_caches_as_unresolvable() { - let server = wiremock::MockServer::start().await; - let body = serde_json::json!({ - "uri": "at://did:plc:nel/sh.tangled.repo/abcabcabcabcz", - "cid": "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i", - "value": {"$type": "sh.tangled.repo"} - }); - wiremock::Mock::given(wiremock::matchers::method("GET")) - .and(wiremock::matchers::path("/xrpc/com.atproto.repo.getRecord")) - .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(body)) - .expect(1) - .mount(&server) - .await; - - let client = - SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); - let resolver = - RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default()); - - let owner = did("did:plc:nel"); - let key = rkey("abcabcabcabcz"); - let first = resolver.resolve(&owner, &key).await; - let second = resolver.resolve(&owner, &key).await; - assert_eq!( - first, - Resolution::Unresolvable, - "a repo body that fails lexicon validation must not be conflated with NoRepoDid", - ); - assert_eq!(second, Resolution::Unresolvable); - } - #[tokio::test] async fn slingshot_transport_error_caches_with_short_ttl() { let server = wiremock::MockServer::start().await; diff --git a/bobbin/crates/resolver/src/normalize.rs b/bobbin/crates/resolver/src/normalize.rs index 0104d7f23..e5555bff6 100644 --- a/bobbin/crates/resolver/src/normalize.rs +++ b/bobbin/crates/resolver/src/normalize.rs @@ -93,11 +93,11 @@ use bobbin_types::org_tangled::feed::star::Star; 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::repo::manifest::Manifest; use bobbin_types::sh_tangled::feed::comment::Comment as FeedComment; use bobbin_types::sh_tangled::feed::reaction::Reaction; use bobbin_types::sh_tangled::label::definition::Definition as LabelDefinition; use bobbin_types::sh_tangled::label::op::Op as LabelOp; -use bobbin_types::sh_tangled::repo::Repo; use bobbin_types::sh_tangled::repo::collaborator::Collaborator; use bobbin_types::sh_tangled::repo::issue::Issue; use bobbin_types::sh_tangled::repo::issue::state::State as IssueState; @@ -115,7 +115,7 @@ identity_normalize!( LabelDefinition, LabelOp, SshKey, - Repo, + Manifest, Collaborator, Issue, IssueState, diff --git a/bobbin/crates/types/src/edges.rs b/bobbin/crates/types/src/edges.rs index 0ee2470fa..94b3baab1 100644 --- a/bobbin/crates/types/src/edges.rs +++ b/bobbin/crates/types/src/edges.rs @@ -3,7 +3,7 @@ use alloc::vec::Vec; use jacquard_common::types::did::Did; use jacquard_common::types::nsid::Nsid; -use jacquard_common::types::string::{AtStrError, AtUri, Datetime, UriValue}; +use jacquard_common::types::string::{AtStrError, AtUri, Datetime}; use jacquard_common::types::tid::Tid; use jacquard_common::{BosStr, DefaultStr}; @@ -14,11 +14,11 @@ use crate::org_tangled::feed::subscription::Subscription; 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::repo::manifest::{Manifest, ManifestUpstream}; use crate::sh_tangled::feed::comment::Comment as FeedCommentRecord; use crate::sh_tangled::feed::reaction::Reaction; use crate::sh_tangled::label::definition::Definition as LabelDefinitionRecord; use crate::sh_tangled::label::op::Op as LabelOpRecord; -use crate::sh_tangled::repo::Repo as RepoRecord; use crate::sh_tangled::repo::artifact::Artifact; use crate::sh_tangled::repo::collaborator::Collaborator; use crate::sh_tangled::repo::issue::Issue; @@ -59,7 +59,7 @@ pub enum Record { LabelDefinition(LabelDefinitionRecord), LabelOp(LabelOpRecord), SshKey(SshKey), - Repo(RepoRecord), + Repo(Manifest), Artifact(Artifact), Collaborator(Collaborator), Issue(Issue), @@ -98,7 +98,7 @@ impl Record { "sh.tangled.label.definition" => parse!(LabelDefinition), "sh.tangled.label.op" => parse!(LabelOp), "org.tangled.key.sshKey" => parse!(SshKey), - "sh.tangled.repo" => parse!(Repo), + "org.tangled.repo.manifest" => parse!(Repo), "sh.tangled.repo.artifact" => parse!(Artifact), "sh.tangled.repo.collaborator" => parse!(Collaborator), "sh.tangled.repo.issue" => parse!(Issue), @@ -122,7 +122,7 @@ impl Record { Self::LabelDefinition(_) => "sh.tangled.label.definition", Self::LabelOp(_) => "sh.tangled.label.op", Self::SshKey(_) => "org.tangled.key.sshKey", - Self::Repo(_) => "sh.tangled.repo", + Self::Repo(_) => "org.tangled.repo.manifest", Self::Artifact(_) => "sh.tangled.repo.artifact", Self::Collaborator(_) => "sh.tangled.repo.collaborator", Self::Issue(_) => "sh.tangled.repo.issue", @@ -208,12 +208,16 @@ impl Record { pub const REPO_SOURCE_EDGE_KIND: &str = "sh.tangled.repo.source"; -fn repo_edges(source: &AtUri, record: &RepoRecord) -> Vec { - let mut edges = owner_self_edges("sh.tangled.repo", source); - if let Some(subject) = fork_source_subject(record) { +fn repo_edges(source: &AtUri, record: &Manifest) -> Vec { + let mut edges = one_edge( + "org.tangled.repo.manifest", + SubjectRef::Did(record.declaration.owner.clone()), + source, + ); + if let Some(ManifestUpstream::SourceTangled(upstream)) = &record.upstream { edges.push(Edge { kind: nsid_static(REPO_SOURCE_EDGE_KIND), - subject, + subject: SubjectRef::Did(upstream.did.clone()), source: source.clone(), sort_micros: 0, }); @@ -221,14 +225,6 @@ fn repo_edges(source: &AtUri, record: &RepoRecord) -> Ve edges } -fn fork_source_subject(record: &RepoRecord) -> Option { - match record.source.as_ref()? { - UriValue::Did(did) => Some(SubjectRef::Did(did.clone())), - UriValue::At(uri) => uri_subject_for_record(uri), - _ => None, - } -} - fn repo_subject( uri: &Option>, did: &Option>, @@ -256,7 +252,7 @@ fn one_edge(kind: &'static str, subject: SubjectRef, source: &AtUri) const GLOBAL_KINDS: &[&str] = &[ "org.tangled.feed.star", "org.tangled.graph.follow", - "sh.tangled.repo", + "org.tangled.repo.manifest", ]; /// One `(kind, Global)` edge per allowlisted primary kind, sharing the primary's @@ -554,15 +550,14 @@ mod tests { parsed.extract_edges(&at(source)).expect("extract") } - fn repo_body(source: Option<&str>) -> serde_json::Value { + fn repo_body(upstream: Option) -> serde_json::Value { let mut body = json!({ - "$type": "sh.tangled.repo", + "$type": "org.tangled.repo.manifest", "createdAt": "2026-05-01T00:00:00Z", - "knot": "oyster.cafe", - "name": "abalone" + "declaration": { "owner": "did:plc:nel", "slug": "abalone" } }); - if let Some(source) = source { - body["source"] = json!(source); + if let Some(upstream) = upstream { + body["upstream"] = upstream; } body } @@ -570,54 +565,41 @@ mod tests { #[test] fn plain_repo_only_makes_its_owner_edge() { let edges = extract( - "sh.tangled.repo", - "at://did:plc:nel/sh.tangled.repo/abcabcabcabcz", + "org.tangled.repo.manifest", + "at://did:plc:abalone/org.tangled.repo.manifest/self", repo_body(None), ); assert_eq!(edges.len(), 1); - assert_eq!(edges[0].kind, nsid("sh.tangled.repo")); + assert_eq!(edges[0].kind, nsid("org.tangled.repo.manifest")); assert_eq!(edges[0].subject, did_subj("did:plc:nel")); } #[test] - fn fork_of_a_repo_did_keys_on_the_did() { + fn fork_of_a_tangled_repo_keys_on_the_did() { let edges = extract( - "sh.tangled.repo", - "at://did:plc:nel/sh.tangled.repo/abcabcabcabcz", - repo_body(Some("did:plc:abalone")), + "org.tangled.repo.manifest", + "at://did:plc:abalone/org.tangled.repo.manifest/self", + repo_body(Some( + json!({ "$type": "org.tangled.repo.defs#sourceTangled", "did": "did:plc:clam" }), + )), ); let fork: Vec<&Edge> = edges .iter() .filter(|e| e.kind.as_ref() == REPO_SOURCE_EDGE_KIND) .collect(); assert_eq!(fork.len(), 1); - assert_eq!(fork[0].subject, did_subj("did:plc:abalone")); - } - - #[test] - fn fork_of_a_repo_without_a_did_keys_on_the_record() { - let edges = extract( - "sh.tangled.repo", - "at://did:plc:nel/sh.tangled.repo/abcabcabcabcz", - repo_body(Some("at://did:plc:olaren/sh.tangled.repo/core")), - ); - let fork: Vec<&Edge> = edges - .iter() - .filter(|e| e.kind.as_ref() == REPO_SOURCE_EDGE_KIND) - .collect(); - assert_eq!(fork.len(), 1); - assert_eq!( - fork[0].subject, - uri_subj("at://did:plc:olaren/sh.tangled.repo/core") - ); + assert_eq!(fork[0].subject, did_subj("did:plc:clam")); } #[test] fn fork_of_something_off_network_makes_no_fork_edge() { let edges = extract( - "sh.tangled.repo", - "at://did:plc:nel/sh.tangled.repo/abcabcabcabcz", - repo_body(Some("https://github.com/90-008/awawa")), + "org.tangled.repo.manifest", + "at://did:plc:abalone/org.tangled.repo.manifest/self", + repo_body(Some(json!({ + "$type": "org.tangled.repo.defs#sourceExternal", + "uri": "https://github.com/90-008/awawa" + }))), ); assert!( edges @@ -648,14 +630,9 @@ mod tests { }), ), ( - "sh.tangled.repo", - "at://did:plc:olaren/sh.tangled.repo/abcabcabcabcz", - json!({ - "$type": "sh.tangled.repo", - "createdAt": "2026-05-01T00:00:00Z", - "knot": "knot.example.com", - "name": "myrepo" - }), + "org.tangled.repo.manifest", + "at://did:plc:abalone/org.tangled.repo.manifest/self", + repo_body(None), ), ]; for (collection, source, body) in cases { @@ -1011,29 +988,6 @@ mod tests { assert_eq!(edges[0].subject, uri_subj(pull_uri)); } - #[test] - fn repo_self_edge_keys_on_owner_did() { - let edges = extract( - "sh.tangled.repo", - "at://did:plc:teq/sh.tangled.repo/r1", - json!({ - "$type": "sh.tangled.repo", - "name": "abalone", - "knot": "oyster.cafe", - "createdAt": "2026-05-01T00:00:00Z" - }), - ); - assert_eq!( - edges, - vec![Edge { - kind: nsid("sh.tangled.repo"), - subject: did_subj("did:plc:teq"), - source: at("at://did:plc:teq/sh.tangled.repo/r1"), - sort_micros: 0, - }] - ); - } - #[test] fn ssh_key_string_label_all_self_edge_to_owner() { let cases = [ diff --git a/bobbin/crates/types/src/legacy.rs b/bobbin/crates/types/src/legacy.rs index 464df53db..be192a374 100644 --- a/bobbin/crates/types/src/legacy.rs +++ b/bobbin/crates/types/src/legacy.rs @@ -174,26 +174,6 @@ pub struct LegacyCollaborator { pub extra_data: Option>>, } -#[derive(Debug, Deserialize)] -#[serde( - rename_all = "camelCase", - rename = "sh.tangled.repo", - tag = "$type", - bound(deserialize = "S: Deserialize<'de> + BosStr") -)] -pub struct LegacyRepo { - pub added_at: Datetime, - pub knot: S, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub description: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub name: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub owner: Option>, - #[serde(flatten, default, skip_serializing_if = "Option::is_none")] - pub extra_data: Option>>, -} - #[derive(Debug)] pub enum LegacyRecord { Issue(LegacyIssue), @@ -201,7 +181,6 @@ pub enum LegacyRecord { Pull(Box>), PullComment(LegacyPullComment), Collaborator(LegacyCollaborator), - Repo(LegacyRepo), } impl LegacyRecord { @@ -219,7 +198,6 @@ impl LegacyRecord { "sh.tangled.repo.collaborator" => { Ok(Self::Collaborator(serde_json::from_slice(bytes)?)) } - "sh.tangled.repo" => Ok(Self::Repo(serde_json::from_slice(bytes)?)), other => Err(ExtractError::UnknownCollection(other.into())), } } diff --git a/bobbin/crates/types/src/search.rs b/bobbin/crates/types/src/search.rs index d17616a62..d568b663e 100644 --- a/bobbin/crates/types/src/search.rs +++ b/bobbin/crates/types/src/search.rs @@ -11,9 +11,9 @@ use serde::Serialize; use crate::edges::{ExtractError, Record}; use crate::ids::nsid_static; use crate::org_tangled::actor::profile::Profile; +use crate::org_tangled::repo::manifest::Manifest; use crate::sh_tangled::feed::comment::Comment as FeedCommentRecord; use crate::sh_tangled::label::definition::Definition as LabelDefinitionRecord; -use crate::sh_tangled::repo::Repo as RepoRecord; use crate::sh_tangled::repo::issue::Issue; use crate::sh_tangled::repo::pull::Pull; use crate::sh_tangled::string::TangledString; @@ -45,7 +45,7 @@ impl SearchSink for NoopSearchSink { #[serde(untagged)] pub enum SearchableRecord { Profile(Profile), - Repo(RepoRecord), + Repo(Manifest), Issue(Issue), Pull(Pull), FeedComment(FeedCommentRecord), @@ -89,7 +89,7 @@ impl SearchableRecord { pub fn nsid(&self) -> Nsid { let s: &'static str = match self { Self::Profile(_) => "org.tangled.actor.profile", - Self::Repo(_) => "sh.tangled.repo", + Self::Repo(_) => "org.tangled.repo.manifest", Self::Issue(_) => "sh.tangled.repo.issue", Self::Pull(_) => "sh.tangled.repo.pull", Self::FeedComment(_) => "sh.tangled.feed.comment", @@ -164,7 +164,7 @@ fn profile_doc(source: &AtUri, r: &Profile) -> SearchDoc ) } -fn repo_doc(source: &AtUri, r: &RepoRecord) -> SearchDoc { +fn repo_doc(source: &AtUri, r: &Manifest) -> SearchDoc { let mut parts: Vec = Vec::new(); if let Some(d) = &r.description { parts.push(d.as_str().to_owned()); @@ -172,20 +172,17 @@ fn repo_doc(source: &AtUri, r: &RepoRecord) -> SearchDoc if let Some(topics) = &r.topics { parts.extend(topics.iter().map(|t| t.as_str().to_owned())); } - let rkey = source.rkey(); - let title = r - .name - .as_deref() - .or_else(|| rkey.as_ref().map(|k| k.as_str())) - .unwrap_or(""); - doc( - source, - "sh.tangled.repo", - title, - parts, - Some(r.created_at.timestamp()), - r.repo_did.clone(), - ) + SearchDoc { + author: Some(r.declaration.owner.clone()), + ..doc( + source, + "org.tangled.repo.manifest", + r.declaration.slug.0.as_str(), + parts, + Some(r.created_at.timestamp()), + author_of(source), + ) + } } fn issue_doc(source: &AtUri, r: &Issue) -> SearchDoc { @@ -318,39 +315,29 @@ mod tests { } #[test] - fn repo_indexes_name_description_topics() { + fn repo_indexes_slug_description_topics() { let doc = extract( - &nsid("sh.tangled.repo"), - &at("at://did:plc:teq/sh.tangled.repo/r1"), + &nsid("org.tangled.repo.manifest"), + &at("at://did:plc:scallop/org.tangled.repo.manifest/self"), json!({ - "$type": "sh.tangled.repo", - "name": "scallop", - "knot": "oyster.cafe", + "$type": "org.tangled.repo.manifest", + "declaration": { "owner": "did:plc:teq", "slug": "scallop" }, "description": "shell index for tide pools", "topics": ["intertidal", "molluscs"], "createdAt": "2026-05-01T00:00:00Z" }), ); assert_eq!(doc.title, "scallop"); + assert_eq!(doc.author.as_ref().map(|d| d.as_str()), Some("did:plc:teq")); + assert_eq!( + doc.repo.as_ref().map(|d| d.as_str()), + Some("did:plc:scallop") + ); assert!(doc.body.contains("shell index")); assert!(doc.body.contains("intertidal")); assert!(doc.body.contains("molluscs")); } - #[test] - fn repo_without_name_field_falls_back_to_rkey() { - let doc = extract( - &nsid("sh.tangled.repo"), - &at("at://did:plc:teq/sh.tangled.repo/limpet"), - json!({ - "$type": "sh.tangled.repo", - "knot": "oyster.cafe", - "createdAt": "2026-05-01T00:00:00Z" - }), - ); - assert_eq!(doc.title, "limpet"); - } - #[test] fn non_text_records_have_no_searchable_view() { let parsed = Record::from_json_value( diff --git a/bobbin/crates/xrpc/src/codesearch.rs b/bobbin/crates/xrpc/src/codesearch.rs index 7c8d1083d..b29bad154 100644 --- a/bobbin/crates/xrpc/src/codesearch.rs +++ b/bobbin/crates/xrpc/src/codesearch.rs @@ -214,8 +214,6 @@ mod tests { use bobbin_runtime::{RuntimeHasher, SystemClock}; use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; - use jacquard_common::types::datetime::Datetime; - use jacquard_common::types::recordkey::Rkey; use serde_json::{Value, json}; use std::sync::Arc; use url::Url; @@ -289,24 +287,22 @@ mod tests { } /// Slingshot and zoekt both point at `server`. When `hydratable`, the repo record is - /// mounted and observed so `build_repo_view_basic` resolves. + /// mounted so `build_repo_view_basic` resolves. async fn app(server: &MockServer, hydratable: bool) -> AppState { let base = Url::parse(&server.uri()).unwrap(); if hydratable { Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) - .and(query_param("repo", OWNER)) - .and(query_param("collection", "sh.tangled.repo")) - .and(query_param("rkey", "r1")) + .and(query_param("repo", REPO)) + .and(query_param("collection", "org.tangled.repo.manifest")) + .and(query_param("rkey", "self")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ - "uri": format!("at://{OWNER}/sh.tangled.repo/r1"), + "uri": format!("at://{REPO}/org.tangled.repo.manifest/self"), "cid": CID, "value": { - "$type": "sh.tangled.repo", - "name": "core", - "knot": "oyster.cafe", + "$type": "org.tangled.repo.manifest", + "declaration": { "owner": OWNER, "slug": "core" }, "createdAt": "2026-05-01T00:00:00Z", - "repoDid": REPO, }, }))) .mount(server) @@ -325,18 +321,6 @@ mod tests { Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), Arc::new(crate::default_directory()), ); - if hydratable { - state - .resolver - .observe( - did(OWNER), - Rkey::new_owned("r1").unwrap(), - Some(did(REPO)), - None, - &Datetime::raw_str("2026-05-01T00:00:00Z"), - ) - .await; - } state.with_codesearch(Some(Arc::new(CodeSearch::new(&base).unwrap()))) } diff --git a/bobbin/crates/xrpc/src/enrich.rs b/bobbin/crates/xrpc/src/enrich.rs index 35cae82dc..6d254fda9 100644 --- a/bobbin/crates/xrpc/src/enrich.rs +++ b/bobbin/crates/xrpc/src/enrich.rs @@ -7,8 +7,7 @@ use axum::{ http::{Request, StatusCode}, response::Response, }; -use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static}; -use bobbin_types::sh_tangled::repo::{Repo, RepoRecord}; +use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi}; use futures::stream::{self, StreamExt as _}; use jacquard_common::DefaultStr; use jacquard_common::IntoStatic; @@ -21,10 +20,8 @@ use serde_json::{Map, Value, json}; use tower::ServiceExt; use crate::recordpath::{RecordPath, walk_path}; -use crate::view::{build_repo_view_basic, build_repo_view_basic_from_record}; -use crate::{ - AppState, FETCH_CONCURRENCY, SubjectShape, XrpcError, fetch, mirror_kind, subject_shape, -}; +use crate::view::build_repo_view_basic; +use crate::{AppState, FETCH_CONCURRENCY, SubjectShape, XrpcError, mirror_kind, subject_shape}; pub const TYPE_COUNT: &str = "sh.tangled.query.enrichResponse#count"; pub const TYPE_DISTINCT_AUTHORS: &str = "sh.tangled.query.enrichResponse#distinctAuthors"; @@ -400,15 +397,12 @@ async fn hydrate_repo_view( ) -> Option { let view = match subject { SubjectRef::Did(did) => build_repo_view_basic(state, viewer, did).await.ok()?, - SubjectRef::Uri(uri) => { - let (_, repo) = fetch::>(state, uri) - .await - .ok()?; - let repo_did = repo.repo_did.clone()?; - build_repo_view_basic_from_record(state, viewer, uri, &repo, repo_did) + SubjectRef::Uri(uri) if view_type(uri.collection()?.as_ref()).is_some() => { + build_repo_view_basic(state, viewer, &owner_did_from_aturi(uri)?) .await .ok()? } + SubjectRef::Uri(_) => return None, SubjectRef::Global => return None, }; serde_json::to_value(view).ok() @@ -417,7 +411,7 @@ async fn hydrate_repo_view( /// the only place enrich knows which collections have views fn view_type(collection: &str) -> Option<&'static str> { match collection { - "sh.tangled.repo" => Some(TYPE_REPO_VIEW_BASIC), + "org.tangled.repo.manifest" => Some(TYPE_REPO_VIEW_BASIC), _ => None, } } diff --git a/bobbin/crates/xrpc/src/feed.rs b/bobbin/crates/xrpc/src/feed.rs index 56d5e86ec..1df046da0 100644 --- a/bobbin/crates/xrpc/src/feed.rs +++ b/bobbin/crates/xrpc/src/feed.rs @@ -11,6 +11,7 @@ use bobbin_edge_index::{ use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi}; use bobbin_types::org_tangled::feed::star::{Star, StarRecord, StarSubject}; use bobbin_types::org_tangled::graph::follow::{Follow, FollowRecord}; +use bobbin_types::org_tangled::repo::manifest::{Manifest, ManifestRecord, ManifestUpstream}; use bobbin_types::sh_tangled::actor::{ProfileViewBasic, ProfileViewDetailed, ViewerState}; use bobbin_types::sh_tangled::embed::commit::View as CommitEmbedView; use bobbin_types::sh_tangled::feed::CommentView; @@ -18,9 +19,8 @@ use bobbin_types::sh_tangled::feed::comment::{Comment, CommentRecord}; use bobbin_types::sh_tangled::feed::get_timeline::{ FollowEvent, RepoEvent, StarEvent, TimelineItem, TimelineItemEvent, }; -use bobbin_types::sh_tangled::repo::{Repo, RepoRecord}; use jacquard_axum::service_auth::ExtractOptionalServiceAuth; -use jacquard_common::types::string::{AtUri, Datetime, Did, Handle, UriValue}; +use jacquard_common::types::string::{AtUri, Datetime, Did, Handle}; use jacquard_common::xrpc::XrpcResp; use jacquard_common::{DefaultStr, IntoStatic as _}; @@ -31,7 +31,7 @@ use crate::{ parse_subject, }; -const REPO_NSID: &str = "sh.tangled.repo"; +const REPO_NSID: &str = "org.tangled.repo.manifest"; const STAR_NSID: &str = "org.tangled.feed.star"; const FOLLOW_NSID: &str = "org.tangled.graph.follow"; const STAR_BY_NSID: &str = "org.tangled.feed.star.by"; @@ -254,14 +254,16 @@ async fn hydrate_repo_event( viewer: Option<&Did>, uri: &AtUri, ) -> Result<(TimelineItemEvent, Datetime), XrpcError> { - let (body, repo) = fetch::>(state, uri).await?; - let repo_did = repo.repo_did.clone().ok_or_else(|| { - XrpcError::InvalidRecord("repo has no repo_did; cannot build repoViewBasic".into()) - })?; - let view = build_repo_view_basic_from_record(state, viewer, uri, &repo, repo_did).await?; - let source = match &repo.source { - Some(UriValue::Did(did)) => build_repo_view_basic(state, viewer, did).await.ok(), - Some(UriValue::At(_aturi)) => None, // we drop the legacy support + let (body, repo) = fetch::>(state, uri).await?; + let repo_did = owner_did_from_aturi(uri) + .ok_or_else(|| XrpcError::InvalidParams("repo uri authority must be a did".into()))?; + let view = build_repo_view_basic_from_record(state, viewer, &repo, repo_did).await?; + let source = match &repo.upstream { + Some(ManifestUpstream::SourceTangled(upstream)) => { + build_repo_view_basic(state, viewer, &upstream.did) + .await + .ok() + } _ => None, }; let event_at = repo.created_at.clone(); diff --git a/bobbin/crates/xrpc/src/lib.rs b/bobbin/crates/xrpc/src/lib.rs index df4b5201b..182992007 100644 --- a/bobbin/crates/xrpc/src/lib.rs +++ b/bobbin/crates/xrpc/src/lib.rs @@ -48,6 +48,9 @@ use bobbin_types::org_tangled::feed::subscription::SubscriptionRecord; use bobbin_types::org_tangled::graph::follow::{Follow, FollowRecord}; use bobbin_types::org_tangled::graph::vouch::{Vouch, VouchRecord}; use bobbin_types::org_tangled::key::ssh_key::{SshKey, SshKeyGetRecordOutput, SshKeyRecord}; +use bobbin_types::org_tangled::repo::manifest::{ + Manifest, ManifestGetRecordOutput, ManifestRecord, +}; use bobbin_types::record::RecordBody; use bobbin_types::search::SearchableRecord; use bobbin_types::sh_tangled::feed::comment::{ @@ -68,7 +71,6 @@ use bobbin_types::sh_tangled::repo::pull::status::{ Status as PullStatus, StatusRecord as PullStatusRecord, }; use bobbin_types::sh_tangled::repo::pull::{Pull, PullGetRecordOutput, PullRecord}; -use bobbin_types::sh_tangled::repo::{Repo, RepoGetRecordOutput, RepoRecord}; use bobbin_types::sh_tangled::string::{ TangledString, TangledStringGetRecordOutput, TangledStringRecord, }; @@ -1321,7 +1323,7 @@ edge_kinds! { "sh.tangled.label.definition" => LabelDefinitionRecord, SubjectShape::BareDid; "sh.tangled.label.op" => LabelOpRecord, SubjectShape::OneOfCollections(&["sh.tangled.repo.issue", "sh.tangled.repo.pull"]), mirror LabelOpBy; "org.tangled.key.sshKey" => SshKeyRecord, SubjectShape::BareDid; - "sh.tangled.repo" => RepoRecord, SubjectShape::BareDid; + "org.tangled.repo.manifest" => ManifestRecord, SubjectShape::BareDid; "sh.tangled.repo.artifact" => ArtifactRecord, SubjectShape::BareDid, mirror ArtifactBy; "sh.tangled.repo.collaborator" => CollaboratorRecord, SubjectShape::BareDid, mirror CollaboratorBy; "sh.tangled.repo.issue" => IssueRecord, SubjectShape::BareDid, mirror IssueBy; @@ -1592,55 +1594,42 @@ where async fn get_repo( State(state): State, XrpcQuery(q): XrpcQuery, -) -> Result>>, XrpcError> { - let (body, value) = fetch::>(&state, &q.repo).await?; - Ok(Json(Deduped(RepoGetRecordOutput { - cid: Some(body.cid.clone()), - uri: body.uri.clone(), - value, - }))) +) -> Result>>, XrpcError> { + get_manifest(&state, q.repo).await } async fn get_repo_by_repo_did( State(state): State, XrpcQuery(q): XrpcQuery, -) -> Result>>, XrpcError> { - let ident = state - .resolver - .lookup_by_repo_did(&q.repo_did) - .await - .ok_or(XrpcError::NotFound)?; - let uri = AtUri::::from_parts_owned( - ident.owner.as_str(), - RepoRecord::NSID, - ident.rkey.as_str(), - ) - .expect("Did and Rkey newtypes already validated, at-uri assembly cannot fail"); - let (body, value) = fetch_from_uri::>(&state, uri).await?; - Ok(Json(Deduped(RepoGetRecordOutput { - cid: Some(body.cid.clone()), - uri: body.uri.clone(), - value, - }))) +) -> Result>>, XrpcError> { + get_manifest(&state, view::manifest_uri(&q.repo_did)).await } async fn get_repo_by_name( State(state): State, XrpcQuery(q): XrpcQuery, -) -> Result>>, XrpcError> { +) -> Result>>, XrpcError> { let ident = state .resolver .lookup_by_name(&q.owner, &q.name) .await .ok_or(XrpcError::NotFound)?; - let uri = AtUri::::from_parts_owned( - ident.owner.as_str(), - RepoRecord::NSID, - ident.rkey.as_str(), - ) - .expect("Did and Rkey newtypes already validated, at-uri assembly cannot fail"); - let (body, value) = fetch_from_uri::>(&state, uri).await?; - Ok(Json(Deduped(RepoGetRecordOutput { + let Some(bobbin_resolver::Resolution::Mapped(repo_did)) = state + .resolver + .cached_resolution(&ident.owner, &ident.rkey) + .await + else { + return Err(XrpcError::NotFound); + }; + get_manifest(&state, view::manifest_uri(&repo_did)).await +} + +async fn get_manifest( + state: &AppState, + uri: AtUri, +) -> Result>>, XrpcError> { + let (body, value) = fetch_from_uri::>(state, uri).await?; + Ok(Json(Deduped(ManifestGetRecordOutput { cid: Some(body.cid.clone()), uri: body.uri.clone(), value, @@ -1713,7 +1702,7 @@ async fn get_repos( RawQuery(query): RawQuery, ) -> Result { let uris = collect_repeated(query.as_deref(), BULK_REPOS_KEY); - bulk_fetch::>(&state, uris).await + bulk_fetch::>(&state, uris).await } async fn get_repos_by_repo_dids( @@ -1736,20 +1725,8 @@ async fn get_repos_by_repo_dids( .map_err(|_| XrpcError::InvalidParams(format!("invalid did: {s}"))) }) .collect::, _>>()?; - let mut uris: Vec> = Vec::new(); - for did in &dids { - if let Some(ident) = state.resolver.lookup_by_repo_did(did).await { - uris.push( - AtUri::::from_parts_owned( - ident.owner.as_str(), - RepoRecord::NSID, - ident.rkey.as_str(), - ) - .expect("Did and Rkey newtypes already validated, at-uri assembly cannot fail"), - ); - } - } - bulk_stream::>(&state, uris) + let uris = dids.iter().map(view::manifest_uri).collect(); + bulk_stream::>(&state, uris) } async fn get_profiles( @@ -2561,14 +2538,14 @@ async fn list_repos( State(state): State, XrpcQuery(q): XrpcQuery>, ) -> Result { - list_records::, _>(&state, q).await + list_records::, _>(&state, q).await } async fn count_repos( State(state): State, XrpcQuery(q): XrpcQuery, ) -> Result, XrpcError> { - count_for::(&state, q).map(Json) + count_for::(&state, q).map(Json) } // forks are repo records pointing back at a repo, so they get their own edge diff --git a/bobbin/crates/xrpc/src/profile_activity.rs b/bobbin/crates/xrpc/src/profile_activity.rs index 68c604fbb..45e68f7b2 100644 --- a/bobbin/crates/xrpc/src/profile_activity.rs +++ b/bobbin/crates/xrpc/src/profile_activity.rs @@ -3,13 +3,13 @@ use std::collections::{BTreeMap, HashMap}; use axum::{extract::State, response::Json}; use bobbin_edge_index::{EdgeItem, EdgeStore, PageCursor, PageLimit, SortDir}; use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi}; +use bobbin_types::org_tangled::repo::manifest::{Manifest, ManifestRecord}; use bobbin_types::sh_tangled::actor::get_profile_activity::{ ActivityMonth, GetProfileActivityOutput, IssueActivity, IssueActivityState, PullActivity, PullActivityState, RepoActivity, RepoActivityKind, RepoRef, }; use bobbin_types::sh_tangled::repo::issue::{Issue, IssueRecord}; use bobbin_types::sh_tangled::repo::pull::{Pull, PullRecord}; -use bobbin_types::sh_tangled::repo::{Repo, RepoRecord}; use chrono::{DateTime, TimeZone, Utc}; use futures::{StreamExt, stream}; use jacquard_common::DefaultStr; @@ -18,8 +18,10 @@ use jacquard_common::xrpc::XrpcResp; use serde::Deserialize; use crate::{ - AppState, XrpcError, XrpcQuery, accept_state_source, commit_stats::month_number, fetch, - source_authority_did, view::build_profile_basic, + AppState, XrpcError, XrpcQuery, accept_state_source, + commit_stats::month_number, + fetch, source_authority_did, + view::{build_profile_basic, manifest_uri}, }; const DEFAULT_MONTHS: u32 = 7; @@ -111,7 +113,7 @@ pub(crate) async fn get_profile_activity( fn activity_keys(subject: &Did) -> [EdgeKey; 3] { [ EdgeKey::new( - nsid_static(RepoRecord::NSID), + nsid_static(ManifestRecord::NSID), SubjectRef::Did(subject.clone()), ), EdgeKey::new(nsid_static(ISSUE_BY_NSID), SubjectRef::Did(subject.clone())), @@ -178,9 +180,10 @@ async fn hydrate_activity( let collection = uri.collection(); let result: Result = async { match collection.as_ref().map(|value| value.as_ref()) { - Some(RepoRecord::NSID) => { - let (_, record) = fetch::>(state, &uri).await?; - let kind = if record.source.is_some() { + Some(ManifestRecord::NSID) => { + let (_, record) = + fetch::>(state, &uri).await?; + let kind = if record.upstream.is_some() { RepoActivityKind::Forked } else { RepoActivityKind::Created @@ -346,36 +349,23 @@ async fn repo_ref_by_did( state: &AppState, did: &Did, ) -> Result, XrpcError> { - let ident = state - .resolver - .lookup_by_repo_did(did) - .await - .ok_or(XrpcError::NotFound)?; - let uri = AtUri::from_parts_owned(ident.owner.as_str(), RepoRecord::NSID, ident.rkey.as_str()) - .map_err(|error| XrpcError::InvalidRecord(format!("repo uri assembly: {error}")))?; - let (_, record) = fetch::>(state, &uri).await?; + let uri = manifest_uri(did); + let (_, record) = fetch::>(state, &uri).await?; repo_ref_from_record(state, &uri, &record).await } async fn repo_ref_from_record( state: &AppState, uri: &AtUri, - record: &Repo, + record: &Manifest, ) -> Result, XrpcError> { - let owner_did = owner_did_from_aturi(uri) - .ok_or_else(|| XrpcError::InvalidRecord("repo uri authority must be a did".into()))?; - let owner = build_profile_basic(state, None, &owner_did).await; - let name = record.name.clone().unwrap_or_else(|| { - uri.rkey() - .map(|rkey| DefaultStr::from(rkey.as_ref())) - .unwrap_or_default() - }); + let owner = build_profile_basic(state, None, &record.declaration.owner).await; Ok(RepoRef { uri: uri.clone(), - did: record.repo_did.clone(), + did: owner_did_from_aturi(uri), owner_did: owner.did, owner_handle: owner.handle, - name, + name: DefaultStr::from(record.declaration.slug.0.as_str()), extra_data: None, }) } @@ -439,7 +429,11 @@ mod tests { let actor = did("did:plc:profileactor"); let now = Utc.with_ymd_and_hms(2026, 9, 15, 0, 0, 0).single().unwrap(); let samples = [ - ("sh.tangled.repo", "sh.tangled.repo", "2026-09-01T00:00:00Z"), + ( + ManifestRecord::NSID, + ManifestRecord::NSID, + "2026-09-01T00:00:00Z", + ), ( ISSUE_BY_NSID, "sh.tangled.repo.issue", @@ -481,10 +475,10 @@ mod tests { let store = EdgeStore::new(RuntimeHasher::default()); for index in 0..item_count { store.add(Edge { - kind: nsid_static("sh.tangled.repo"), + kind: nsid_static(ManifestRecord::NSID), subject: SubjectRef::Did(actor.clone()), source: AtUri::new_owned(format!( - "at://did:plc:owner/sh.tangled.repo/item{index}", + "at://did:plc:owner/org.tangled.repo.manifest/item{index}", )) .unwrap(), sort_micros: base + index as u64, diff --git a/bobbin/crates/xrpc/src/view.rs b/bobbin/crates/xrpc/src/view.rs index c7ed04e9d..05814ee0b 100644 --- a/bobbin/crates/xrpc/src/view.rs +++ b/bobbin/crates/xrpc/src/view.rs @@ -1,11 +1,15 @@ use bobbin_edge_index::{EdgeStore, IssueStateKind, PullStatusKind}; use bobbin_types::{ edges::REPO_SOURCE_EDGE_KIND, - ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi}, - org_tangled::{feed::star::StarRecord, graph::follow::FollowRecord}, + ids::{EdgeKey, SubjectRef, nsid_static}, + org_tangled::{ + feed::star::StarRecord, + graph::follow::FollowRecord, + repo::manifest::{Manifest, ManifestRecord}, + }, sh_tangled::{ actor, - repo::{self, Repo, RepoRecord, issue::IssueRecord, pull::PullRecord}, + repo::{self, issue::IssueRecord, pull::PullRecord}, }, }; use jacquard_common::{ @@ -81,42 +85,29 @@ fn build_actor_viewer_state(edges: &EdgeStore, viewer: &Did, subject: &Did) -> a } } +pub(crate) fn manifest_uri(repo_did: &Did) -> AtUri { + AtUri::from_parts_owned(repo_did.as_str(), ManifestRecord::NSID, "self") + .expect("a did and a literal rkey always assemble") +} + pub(crate) async fn build_repo_view_basic( state: &AppState, viewer: Option<&Did>, did: &Did, ) -> Result { - let ident = state - .resolver - .lookup_by_repo_did(did) - .await - .ok_or(XrpcError::NotFound)?; - let uri = AtUri::::from_parts_owned( - ident.owner.as_str(), - RepoRecord::NSID, - ident.rkey.as_str(), - ) - .map_err(|e| XrpcError::InvalidRecord(format!("repo uri assembly: {e}")))?; - let (_, repo) = fetch::>(state, &uri).await?; - build_repo_view_basic_from_record(state, viewer, &uri, &repo, did.clone()).await + let (_, repo) = + fetch::>(state, &manifest_uri(did)).await?; + build_repo_view_basic_from_record(state, viewer, &repo, did.clone()).await } pub(crate) async fn build_repo_view_basic_from_record( state: &AppState, viewer: Option<&Did>, - repo_uri: &AtUri, - repo_record: &Repo, + repo_record: &Manifest, repo_did: Did, ) -> Result { - let owner_did = owner_did_from_aturi(repo_uri) - .ok_or_else(|| XrpcError::InvalidParams("repo uri authority must be a did".into()))?; - let owner = build_profile_basic(state, viewer, &owner_did).await; - let slug = repo_record.name.clone().unwrap_or_else(|| { - repo_uri - .rkey() - .map(|r| DefaultStr::from(r.as_ref())) - .unwrap_or_default() - }); + let owner = build_profile_basic(state, viewer, &repo_record.declaration.owner).await; + let slug = DefaultStr::from(repo_record.declaration.slug.0.as_str()); let subject = SubjectRef::Did(repo_did.clone()); let star_key = EdgeKey::new(nsid_static(StarRecord::NSID), subject.clone()); let star_count = state.edges.count(&star_key) as i64; diff --git a/bobbin/crates/xrpc/tests/aggregation.rs b/bobbin/crates/xrpc/tests/aggregation.rs index a1c94fac0..fc3e5e57b 100644 --- a/bobbin/crates/xrpc/tests/aggregation.rs +++ b/bobbin/crates/xrpc/tests/aggregation.rs @@ -18,7 +18,6 @@ use chrono::Utc; use futures::stream::{self, StreamExt}; use http::{Request, StatusCode}; use jacquard_common::DefaultStr; -use jacquard_common::types::datetime::Datetime; use jacquard_common::types::did::Did; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::recordkey::Rkey; @@ -2221,8 +2220,10 @@ async fn profile_activity_returns_one_hydrated_aggregate() { let actor = did("did:plc:lyna"); let created_repo_did = did("did:plc:freshrepo"); let target_repo_did = did("did:plc:limpet"); - let target_rkey = rkey("target"); - let created_repo_uri = at(&format!("at://{}/sh.tangled.repo/fresh", actor.as_ref(),)); + let created_repo_uri = at(&format!( + "at://{}/org.tangled.repo.manifest/self", + created_repo_did.as_ref(), + )); let issue_uri = at(&format!("at://{}/sh.tangled.repo.issue/i1", actor.as_ref(),)); let pull_uri = at(&format!("at://{}/sh.tangled.repo.pull/p1", actor.as_ref(),)); let event_micros = Utc::now().timestamp_micros() as u64; @@ -2249,7 +2250,7 @@ async fn profile_activity_returns_one_hydrated_aggregate() { ))); for (kind, source) in [ - ("sh.tangled.repo", &created_repo_uri), + ("org.tangled.repo.manifest", &created_repo_uri), ("sh.tangled.repo.issue.by", &issue_uri), ("sh.tangled.repo.pull.by", &pull_uri), ] { @@ -2261,42 +2262,22 @@ async fn profile_activity_returns_one_hydrated_aggregate() { }); } - h.mount( - &actor, - &nsid("sh.tangled.repo"), - &rkey("fresh"), - json!({ - "$type": "sh.tangled.repo", - "name": "fresh", - "knot": "oyster.cafe", - "repoDid": created_repo_did.as_ref(), - "createdAt": Utc::now().to_rfc3339(), - }), - ) - .await; - h.mount( - &actor, - &nsid("sh.tangled.repo"), - &target_rkey, - json!({ - "$type": "sh.tangled.repo", - "name": "core", - "knot": "oyster.cafe", - "repoDid": target_repo_did.as_ref(), - "createdAt": "2025-01-01T00:00:00Z", - }), - ) - .await; - h.state - .resolver - .observe( - actor.clone(), - target_rkey, - Some(target_repo_did.clone()), - Some(DefaultStr::from("core")), - &Datetime::raw_str("2026-05-01T00:00:00Z"), + for (repo_did, slug, created_at) in [ + (&created_repo_did, "fresh", Utc::now().to_rfc3339()), + (&target_repo_did, "core", "2025-01-01T00:00:00Z".to_owned()), + ] { + h.mount( + repo_did, + &nsid("org.tangled.repo.manifest"), + &rkey("self"), + json!({ + "$type": "org.tangled.repo.manifest", + "declaration": { "owner": actor.as_ref(), "slug": slug }, + "createdAt": created_at, + }), ) .await; + } h.mount( &actor, &nsid("sh.tangled.repo.issue"), diff --git a/bobbin/crates/xrpc/tests/bulk.rs b/bobbin/crates/xrpc/tests/bulk.rs index 9c0b2a883..afcaa6015 100644 --- a/bobbin/crates/xrpc/tests/bulk.rs +++ b/bobbin/crates/xrpc/tests/bulk.rs @@ -10,7 +10,6 @@ use bobbin_slingshot_client::SlingshotClient; use bobbin_xrpc::{AppState, router}; use http::{Request, StatusCode}; use jacquard_common::DefaultStr; -use jacquard_common::types::datetime::Datetime; use jacquard_common::types::did::Did; use jacquard_common::types::handle::Handle; use jacquard_common::types::nsid::Nsid; @@ -169,12 +168,11 @@ fn pull_body(target_repo: &Did, title: &str) -> Value { }) } -fn repo_body(name: &str) -> Value { +fn repo_body(slug: &str) -> Value { json!({ - "$type": "sh.tangled.repo", - "name": name, - "knot": "oyster.cafe", - "createdAt": "2026-05-01T00:00:00Z" + "$type": "org.tangled.repo.manifest", + "createdAt": "2026-05-01T00:00:00Z", + "declaration": { "owner": "did:plc:nel", "slug": slug } }) } @@ -189,16 +187,16 @@ fn profile_body(handle: &Handle) -> Value { async fn get_repos_returns_all_resolved_records() { let h = Harness::new().await; h.mount( - &did("did:plc:nel"), - &nsid("sh.tangled.repo"), - &rkey("abalone"), + &did("did:plc:abalone"), + &nsid("org.tangled.repo.manifest"), + &rkey("self"), repo_body("abalone"), ) .await; h.mount( - &did("did:plc:teq"), - &nsid("sh.tangled.repo"), - &rkey("limpet"), + &did("did:plc:limpet"), + &nsid("org.tangled.repo.manifest"), + &rkey("self"), repo_body("limpet"), ) .await; @@ -208,8 +206,8 @@ async fn get_repos_returns_all_resolved_records() { "sh.tangled.repo.getRepos", "repos", &[ - "at://did:plc:nel/sh.tangled.repo/abalone", - "at://did:plc:teq/sh.tangled.repo/limpet", + "at://did:plc:abalone/org.tangled.repo.manifest/self", + "at://did:plc:limpet/org.tangled.repo.manifest/self", ], )) .await @@ -221,7 +219,7 @@ async fn get_repos_returns_all_resolved_records() { assert_eq!(items.len(), 2); let names: Vec<&str> = items .iter() - .map(|v| v["value"]["name"].as_str().unwrap()) + .map(|v| v["value"]["declaration"]["slug"].as_str().unwrap()) .collect(); assert!(names.contains(&"abalone")); assert!(names.contains(&"limpet")); @@ -357,16 +355,16 @@ async fn get_pulls_returns_all_resolved_pulls() { async fn missing_records_are_dropped_silently() { let h = Harness::new().await; h.mount( - &did("did:plc:nel"), - &nsid("sh.tangled.repo"), - &rkey("abalone"), + &did("did:plc:abalone"), + &nsid("org.tangled.repo.manifest"), + &rkey("self"), repo_body("abalone"), ) .await; h.mount_404( - &did("did:plc:teq"), - &nsid("sh.tangled.repo"), - &rkey("ghost"), + &did("did:plc:ghost"), + &nsid("org.tangled.repo.manifest"), + &rkey("self"), ) .await; let app = router(h.state.clone()); @@ -375,8 +373,8 @@ async fn missing_records_are_dropped_silently() { "sh.tangled.repo.getRepos", "repos", &[ - "at://did:plc:nel/sh.tangled.repo/abalone", - "at://did:plc:teq/sh.tangled.repo/ghost", + "at://did:plc:abalone/org.tangled.repo.manifest/self", + "at://did:plc:ghost/org.tangled.repo.manifest/self", ], )) .await @@ -390,24 +388,24 @@ async fn missing_records_are_dropped_silently() { 1, "missing records must be dropped not fail the bulk call" ); - assert_eq!(items[0]["value"]["name"], json!("abalone")); + assert_eq!(items[0]["value"]["declaration"]["slug"], json!("abalone")); } #[tokio::test] async fn transient_failure_drops_only_that_record() { let h = Harness::new().await; h.mount( - &did("did:plc:nel"), - &nsid("sh.tangled.repo"), - &rkey("conch"), + &did("did:plc:conch"), + &nsid("org.tangled.repo.manifest"), + &rkey("self"), repo_body("conch"), ) .await; Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) - .and(query_param("repo", "did:plc:teq")) - .and(query_param("collection", "sh.tangled.repo")) - .and(query_param("rkey", "flaky")) + .and(query_param("repo", "did:plc:flaky")) + .and(query_param("collection", "org.tangled.repo.manifest")) + .and(query_param("rkey", "self")) .respond_with(ResponseTemplate::new(503)) .mount(&h.server) .await; @@ -417,8 +415,8 @@ async fn transient_failure_drops_only_that_record() { "sh.tangled.repo.getRepos", "repos", &[ - "at://did:plc:nel/sh.tangled.repo/conch", - "at://did:plc:teq/sh.tangled.repo/flaky", + "at://did:plc:conch/org.tangled.repo.manifest/self", + "at://did:plc:flaky/org.tangled.repo.manifest/self", ], )) .await @@ -432,16 +430,16 @@ async fn transient_failure_drops_only_that_record() { 1, "transient upstream failure drops that record, it must not fail the bulk call" ); - assert_eq!(items[0]["value"]["name"], json!("conch")); + assert_eq!(items[0]["value"]["declaration"]["slug"], json!("conch")); } #[tokio::test] async fn wrong_collection_uri_fails_bulk_request() { let h = Harness::new().await; h.mount( - &did("did:plc:nel"), - &nsid("sh.tangled.repo"), - &rkey("conch"), + &did("did:plc:conch"), + &nsid("org.tangled.repo.manifest"), + &rkey("self"), repo_body("conch"), ) .await; @@ -451,7 +449,7 @@ async fn wrong_collection_uri_fails_bulk_request() { "sh.tangled.repo.getRepos", "repos", &[ - "at://did:plc:nel/sh.tangled.repo/conch", + "at://did:plc:conch/org.tangled.repo.manifest/self", "at://did:plc:teq/sh.tangled.repo.issue/whelk", ], )) @@ -487,7 +485,7 @@ async fn empty_uri_list_is_rejected() { async fn over_limit_uri_list_is_rejected() { let h = Harness::new().await; let uris: Vec = (0..51) - .map(|i| format!("at://did:plc:nel/sh.tangled.repo/r{i}")) + .map(|i| format!("at://did:plc:r{i}/org.tangled.repo.manifest/self")) .collect(); let refs: Vec<&str> = uris.iter().map(|s| s.as_str()).collect(); let app = router(h.state.clone()); @@ -517,9 +515,9 @@ async fn malformed_uri_in_list_returns_400() { async fn bulk_items_serialize_a_single_type_key() { let h = Harness::new().await; h.mount( - &did("did:plc:teq"), - &nsid("sh.tangled.repo"), - &rkey("abalone"), + &did("did:plc:abalone"), + &nsid("org.tangled.repo.manifest"), + &rkey("self"), repo_body("abalone"), ) .await; @@ -528,7 +526,7 @@ async fn bulk_items_serialize_a_single_type_key() { .oneshot(bulk_request( "sh.tangled.repo.getRepos", "repos", - &["at://did:plc:teq/sh.tangled.repo/abalone"], + &["at://did:plc:abalone/org.tangled.repo.manifest/self"], )) .await .unwrap(); @@ -541,37 +539,17 @@ async fn bulk_items_serialize_a_single_type_key() { #[tokio::test] async fn get_repos_by_repo_dids_returns_resolved_repos() { let h = Harness::new().await; - h.state - .resolver - .observe( - did("did:plc:nel"), - rkey("abalone"), - Some(did("did:plc:limpet")), - None, - &Datetime::raw_str("2026-05-01T00:00:00Z"), - ) - .await; - h.state - .resolver - .observe( - did("did:plc:teq"), - rkey("coral"), - Some(did("did:plc:coral")), - None, - &Datetime::raw_str("2026-05-01T00:00:00Z"), - ) - .await; h.mount( - &did("did:plc:nel"), - &nsid("sh.tangled.repo"), - &rkey("abalone"), + &did("did:plc:limpet"), + &nsid("org.tangled.repo.manifest"), + &rkey("self"), repo_body("abalone"), ) .await; h.mount( - &did("did:plc:teq"), - &nsid("sh.tangled.repo"), - &rkey("coral"), + &did("did:plc:coral"), + &nsid("org.tangled.repo.manifest"), + &rkey("self"), repo_body("coral"), ) .await; @@ -591,29 +569,19 @@ async fn get_repos_by_repo_dids_returns_resolved_repos() { assert_eq!(items.len(), 2, "{body}"); let names: Vec<&str> = items .iter() - .map(|v| v["value"]["name"].as_str().unwrap()) + .map(|v| v["value"]["declaration"]["slug"].as_str().unwrap()) .collect(); assert!(names.contains(&"abalone")); assert!(names.contains(&"coral")); } #[tokio::test] -async fn get_repos_by_repo_dids_skips_unobserved_dids() { +async fn get_repos_by_repo_dids_skips_dids_without_a_manifest() { let h = Harness::new().await; - h.state - .resolver - .observe( - did("did:plc:nel"), - rkey("abalone"), - Some(did("did:plc:limpet")), - None, - &Datetime::raw_str("2026-05-01T00:00:00Z"), - ) - .await; h.mount( - &did("did:plc:nel"), - &nsid("sh.tangled.repo"), - &rkey("abalone"), + &did("did:plc:limpet"), + &nsid("org.tangled.repo.manifest"), + &rkey("self"), repo_body("abalone"), ) .await; @@ -628,11 +596,11 @@ async fn get_repos_by_repo_dids_skips_unobserved_dids() { .unwrap(), ) .await; - // unknown dids are skipped, same as missing bulk records + // dids without a manifest are skipped, same as missing bulk records assert_eq!(status, StatusCode::OK, "{body}"); let items = body["items"].as_array().unwrap(); assert_eq!(items.len(), 1, "{body}"); - assert_eq!(items[0]["value"]["name"], "abalone"); + assert_eq!(items[0]["value"]["declaration"]["slug"], "abalone"); } #[tokio::test] diff --git a/bobbin/crates/xrpc/tests/cold_start.rs b/bobbin/crates/xrpc/tests/cold_start.rs index bcf6f9809..c7a531c03 100644 --- a/bobbin/crates/xrpc/tests/cold_start.rs +++ b/bobbin/crates/xrpc/tests/cold_start.rs @@ -112,13 +112,12 @@ async fn cold_start_serves_all_four_point_lookups() { mount_record( &server, &clam, - &nsid("sh.tangled.repo"), - &rkey("r1"), + &nsid("org.tangled.repo.manifest"), + &rkey("self"), json!({ - "$type": "sh.tangled.repo", - "name": "clam", - "knot": "oyster.cafe", - "createdAt": "2026-05-01T00:00:00Z" + "$type": "org.tangled.repo.manifest", + "createdAt": "2026-05-01T00:00:00Z", + "declaration": { "owner": "did:plc:scallop", "slug": "clam" } }), ) .await; @@ -171,9 +170,9 @@ async fn cold_start_serves_all_four_point_lookups() { ( "sh.tangled.repo.getRepo", "repo", - format!("at://{}/sh.tangled.repo/r1", clam.as_ref()), - "knot", - json!("oyster.cafe"), + format!("at://{}/org.tangled.repo.manifest/self", clam.as_ref()), + "createdAt", + json!("2026-05-01T00:00:00Z"), ), ( "sh.tangled.actor.getProfile", @@ -279,14 +278,13 @@ async fn second_call_is_served_from_lru() { .and(path("/xrpc/com.atproto.repo.getRecord")) .and(query_param("repo", uni.as_ref())) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ - "uri": format!("at://{}/sh.tangled.repo/r1", uni.as_ref()), + "uri": format!("at://{}/org.tangled.repo.manifest/self", uni.as_ref()), "cid": CID, "value": { - "$type": "sh.tangled.repo", - "name": "uni", - "knot": "witchcraft.systems", - "createdAt": "2026-05-01T00:00:00Z" - } + "$type": "org.tangled.repo.manifest", + "createdAt": "2026-05-01T00:00:00Z", + "declaration": { "owner": "did:plc:scallop", "slug": "clam" } + } }))) .expect(1) .mount_as_scoped(&server) @@ -294,7 +292,7 @@ async fn second_call_is_served_from_lru() { let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); - let req_uri = format!("at://{}/sh.tangled.repo/r1", uni.as_ref()); + let req_uri = format!("at://{}/org.tangled.repo.manifest/self", uni.as_ref()); stream::iter(0..3) .for_each(|_| { @@ -338,7 +336,7 @@ async fn handle_authority_is_400() { .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", - "at://witchcraft.systems/sh.tangled.repo/r1", + "at://witchcraft.systems/org.tangled.repo.manifest/self", )) .await .unwrap(); @@ -359,7 +357,7 @@ async fn slingshot_404_propagates_as_404() { .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", - "at://did:plc:clam/sh.tangled.repo/missing", + "at://did:plc:clam/org.tangled.repo.manifest/self", )) .await .unwrap(); @@ -372,8 +370,8 @@ async fn wrong_record_type_is_502() { mount_record( &server, &did("did:plc:clam"), - &nsid("sh.tangled.repo"), - &rkey("r1"), + &nsid("org.tangled.repo.manifest"), + &rkey("self"), json!({ "$type": "org.tangled.graph.follow", "createdAt": "2026-05-01T00:00:00Z" @@ -386,7 +384,7 @@ async fn wrong_record_type_is_502() { .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", - "at://did:plc:clam/sh.tangled.repo/r1", + "at://did:plc:clam/org.tangled.repo.manifest/self", )) .await .unwrap(); @@ -402,7 +400,7 @@ async fn wrong_type_does_not_poison_cache() { .and(path("/xrpc/com.atproto.repo.getRecord")) .and(query_param("repo", "did:plc:clam")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ - "uri": "at://did:plc:clam/sh.tangled.repo/r1", + "uri": "at://did:plc:clam/org.tangled.repo.manifest/self", "cid": CID, "value": { "$type": "org.tangled.graph.follow", @@ -418,7 +416,7 @@ async fn wrong_type_does_not_poison_cache() { xrpc_request( "sh.tangled.repo.getRepo", "repo", - "at://did:plc:clam/sh.tangled.repo/r1", + "at://did:plc:clam/org.tangled.repo.manifest/self", ) }; let first = app.clone().oneshot(req()).await.unwrap(); @@ -469,13 +467,13 @@ async fn upstream_uri_mismatch_routes_to_invalid_record() { .and(path("/xrpc/com.atproto.repo.getRecord")) .and(query_param("repo", "did:plc:clam")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ - "uri": "at://did:plc:limpet/sh.tangled.repo/elsewhere", + "uri": "at://did:plc:limpet/org.tangled.repo.manifest/self", "cid": CID, "value": { - "$type": "sh.tangled.repo", - "knot": "oyster.cafe", - "createdAt": "2026-05-01T00:00:00Z" - } + "$type": "org.tangled.repo.manifest", + "createdAt": "2026-05-01T00:00:00Z", + "declaration": { "owner": "did:plc:scallop", "slug": "clam" } + } }))) .mount(&server) .await; @@ -485,7 +483,7 @@ async fn upstream_uri_mismatch_routes_to_invalid_record() { .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", - "at://did:plc:clam/sh.tangled.repo/r1", + "at://did:plc:clam/org.tangled.repo.manifest/self", )) .await .unwrap(); @@ -500,13 +498,13 @@ async fn upstream_garbage_cid_routes_to_invalid_record() { Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ - "uri": "at://did:plc:clam/sh.tangled.repo/r1", + "uri": "at://did:plc:clam/org.tangled.repo.manifest/self", "cid": "not-a-real-cid", "value": { - "$type": "sh.tangled.repo", - "knot": "oyster.cafe", - "createdAt": "2026-05-01T00:00:00Z" - } + "$type": "org.tangled.repo.manifest", + "createdAt": "2026-05-01T00:00:00Z", + "declaration": { "owner": "did:plc:scallop", "slug": "clam" } + } }))) .mount(&server) .await; @@ -516,7 +514,7 @@ async fn upstream_garbage_cid_routes_to_invalid_record() { .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", - "at://did:plc:clam/sh.tangled.repo/r1", + "at://did:plc:clam/org.tangled.repo.manifest/self", )) .await .unwrap(); @@ -544,7 +542,7 @@ async fn oversize_upstream_body_routes_to_upstream_failed() { .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", - "at://did:plc:clam/sh.tangled.repo/r1", + "at://did:plc:clam/org.tangled.repo.manifest/self", )) .await .unwrap(); @@ -567,7 +565,7 @@ async fn upstream_503_routes_to_upstream_failed() { .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", - "at://did:plc:clam/sh.tangled.repo/r1", + "at://did:plc:clam/org.tangled.repo.manifest/self", )) .await .unwrap(); @@ -577,39 +575,23 @@ async fn upstream_503_routes_to_upstream_failed() { } #[tokio::test] -async fn get_repo_by_repo_did_returns_observed_record() { +async fn get_repo_by_repo_did_returns_the_manifest() { let server = MockServer::start().await; - let owner_did = did("did:plc:scallop"); - let rk = rkey("r1"); let repo_did = did("did:plc:limpet"); mount_record( &server, - &owner_did, - &nsid("sh.tangled.repo"), - &rk, + &repo_did, + &nsid("org.tangled.repo.manifest"), + &rkey("self"), json!({ - "$type": "sh.tangled.repo", - "name": "scallop", - "knot": "oyster.cafe", + "$type": "org.tangled.repo.manifest", "createdAt": "2026-05-01T00:00:00Z", - "repoDid": repo_did.as_ref(), + "declaration": { "owner": "did:plc:scallop", "slug": "scallop" } }), ) .await; - let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; - state - .resolver - .observe( - owner_did.clone(), - rk.clone(), - Some(repo_did.clone()), - None, - &Datetime::raw_str("2026-05-01T00:00:00Z"), - ) - .await; - - let app = router(state); + let app = router(fresh_app(&Url::parse(&server.uri()).unwrap()).await); let resp = app .oneshot(xrpc_request( "sh.tangled.repo.getRepoByRepoDid", @@ -622,31 +604,26 @@ async fn get_repo_by_repo_did_returns_observed_record() { assert_eq!(status, StatusCode::OK); assert_eq!( body["uri"], - format!( - "at://{}/sh.tangled.repo/{}", - owner_did.as_ref(), - rk.as_ref() - ) + format!("at://{}/org.tangled.repo.manifest/self", repo_did.as_ref()) ); - assert_eq!(body["value"]["name"], "scallop"); - assert_eq!(body["value"]["repoDid"], repo_did.as_ref()); + assert_eq!(body["value"]["declaration"]["slug"], "scallop"); } #[tokio::test] async fn get_repo_by_name_returns_observed_record() { let server = MockServer::start().await; let owner_did = did("did:plc:scallop"); + let repo_did = did("did:plc:limpet"); let rk = rkey("core"); mount_record( &server, - &owner_did, - &nsid("sh.tangled.repo"), - &rk, + &repo_did, + &nsid("org.tangled.repo.manifest"), + &rkey("self"), json!({ - "$type": "sh.tangled.repo", - "name": "core", - "knot": "oyster.cafe", + "$type": "org.tangled.repo.manifest", "createdAt": "2026-05-01T00:00:00Z", + "declaration": { "owner": owner_did.as_ref(), "slug": "core" } }), ) .await; @@ -654,7 +631,13 @@ async fn get_repo_by_name_returns_observed_record() { let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; state .resolver - .observe_rkey(owner_did.clone(), rk.clone()) + .observe( + owner_did.clone(), + rk.clone(), + Some(repo_did.clone()), + None, + &Datetime::raw_str("2026-05-01T00:00:00Z"), + ) .await; let app = router(state); @@ -662,7 +645,6 @@ async fn get_repo_by_name_returns_observed_record() { .oneshot(xrpc_request2( "sh.tangled.repo.getRepoByName", ("owner", owner_did.as_ref()), - // the query name is the url rkey; the record's `name` is cosmetic ("name", rk.as_ref()), )) .await @@ -671,13 +653,9 @@ async fn get_repo_by_name_returns_observed_record() { assert_eq!(status, StatusCode::OK); assert_eq!( body["uri"], - format!( - "at://{}/sh.tangled.repo/{}", - owner_did.as_ref(), - rk.as_ref() - ) + format!("at://{}/org.tangled.repo.manifest/self", repo_did.as_ref()) ); - assert_eq!(body["value"]["name"], "core"); + assert_eq!(body["value"]["declaration"]["slug"], "core"); } #[tokio::test] @@ -733,7 +711,7 @@ async fn get_repo_by_name_400_on_invalid_owner() { } #[tokio::test] -async fn get_repo_by_repo_did_404_when_unobserved() { +async fn get_repo_by_repo_did_404_when_manifest_missing() { let server = MockServer::start().await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); @@ -771,13 +749,12 @@ async fn record_values_serialize_a_single_type_key() { mount_record( &server, &did("did:plc:teq"), - &nsid("sh.tangled.repo"), - &rkey("r1"), + &nsid("org.tangled.repo.manifest"), + &rkey("self"), json!({ - "$type": "sh.tangled.repo", - "name": "clam", - "knot": "oyster.cafe", - "createdAt": "2026-05-01T00:00:00Z" + "$type": "org.tangled.repo.manifest", + "createdAt": "2026-05-01T00:00:00Z", + "declaration": { "owner": "did:plc:scallop", "slug": "clam" } }), ) .await; @@ -787,7 +764,7 @@ async fn record_values_serialize_a_single_type_key() { .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", - "at://did:plc:teq/sh.tangled.repo/r1", + "at://did:plc:teq/org.tangled.repo.manifest/self", )) .await .unwrap(); @@ -795,5 +772,8 @@ async fn record_values_serialize_a_single_type_key() { let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap(); let raw = String::from_utf8(bytes.to_vec()).unwrap(); assert_eq!(raw.matches("\"$type\"").count(), 1, "body: {raw}"); - assert!(raw.contains("\"$type\":\"sh.tangled.repo\""), "body: {raw}"); + assert!( + raw.contains("\"$type\":\"org.tangled.repo.manifest\""), + "body: {raw}" + ); } diff --git a/bobbin/crates/xrpc/tests/enrich.rs b/bobbin/crates/xrpc/tests/enrich.rs index c6f7c9ea5..b82e23c15 100644 --- a/bobbin/crates/xrpc/tests/enrich.rs +++ b/bobbin/crates/xrpc/tests/enrich.rs @@ -12,10 +12,8 @@ use bobbin_types::ids::SubjectRef; use bobbin_xrpc::{AppState, router}; use http::{Request, StatusCode}; use jacquard_common::DefaultStr; -use jacquard_common::types::datetime::Datetime; use jacquard_common::types::did::Did; use jacquard_common::types::nsid::Nsid; -use jacquard_common::types::recordkey::Rkey; use jacquard_common::types::string::{AtUri, Handle}; use serde_json::{Value, json}; use tower::ServiceExt; @@ -123,12 +121,10 @@ async fn json_response(resp: axum::response::Response) -> (StatusCode, Value) { (status, parsed) } -fn repo_body(name: &str, repo_did: &Did) -> Value { +fn repo_body(slug: &str, owner: &Did) -> Value { json!({ - "$type": "sh.tangled.repo", - "name": name, - "knot": "oyster.cafe", - "repoDid": repo_did.as_ref(), + "$type": "org.tangled.repo.manifest", + "declaration": { "owner": owner.as_ref(), "slug": slug }, "createdAt": "2026-05-01T00:00:00Z" }) } @@ -142,13 +138,20 @@ fn follow_body() -> Value { /// one repo owned by `owner`, with `stars`/`issues` counts against its repo did async fn repo_fixture(h: &Harness, owner: &Did, repo_did: &Did) { - let repo_uri = at(&format!("at://{}/sh.tangled.repo/reef", owner.as_ref())); - h.add_edge("sh.tangled.repo", SubjectRef::Did(owner.clone()), &repo_uri); + let repo_uri = at(&format!( + "at://{}/org.tangled.repo.manifest/self", + repo_did.as_ref() + )); + h.add_edge( + "org.tangled.repo.manifest", + SubjectRef::Did(owner.clone()), + &repo_uri, + ); h.mount( - owner, - "sh.tangled.repo", - "reef", - repo_body("reef", repo_did), + repo_did, + "org.tangled.repo.manifest", + "self", + repo_body("reef", owner), ) .await; for (i, stargazer) in ["did:plc:a", "did:plc:b", "did:plc:a"].iter().enumerate() { @@ -203,7 +206,7 @@ async fn zero_config_counts_stars_and_issues_for_repo_did() { stats["did:plc:limpet"]["sh.tangled.repo.issue:subject"][COUNT], json!(1) ); - assert!(stats["at://did:plc:nel/sh.tangled.repo/reef"].is_null()); + assert!(stats["at://did:plc:limpet/org.tangled.repo.manifest/self"].is_null()); } #[tokio::test] @@ -335,12 +338,12 @@ async fn targets_scope_each_payload_independently() { { "source": "org.tangled.feed.star:subject", "type": COUNT, - "targets": ["items[].value.repoDid"] + "targets": ["items[].uri"] }, { "source": "sh.tangled.repo.issue:subject", "type": COUNT, - "targets": ["items[].uri"] + "targets": ["items[].value.declaration.owner"] } ] }))) @@ -395,7 +398,7 @@ async fn inner_record_miss_passes_through_as_404() { let (status, body) = json_response( app.oneshot(enrich_request(json!({ "xrpc": "sh.tangled.repo.getRepo", - "params": { "repo": "at://did:plc:nel/sh.tangled.repo/absent" }, + "params": { "repo": "at://did:plc:absent/org.tangled.repo.manifest/self" }, "enrich": [{ "source": "org.tangled.feed.star:subject", "type": COUNT }] }))) .await @@ -440,7 +443,7 @@ async fn rejects_bad_requests() { json!({ "xrpc": "sh.tangled.repo.countRepos", "params": { "subject": "did:plc:nel" }, - "enrich": [{ "source": "sh.tangled.repo:.", "type": COUNT }] + "enrich": [{ "source": "org.tangled.repo.manifest:.", "type": COUNT }] }), json!({ "xrpc": "sh.tangled.repo.countRepos", @@ -619,12 +622,12 @@ async fn minidoc_repo_sources_skip_the_author_index() { .observe(owner.clone(), handle("nel.example.com")); let app = router(h.state.clone()); - // sh.tangled.repo has no author mirror; stats would 400, minidocs must not + // manifests have no author mirror; stats would 400, minidocs must not let (status, body) = json_response( app.oneshot(enrich_request(json!({ "xrpc": "sh.tangled.repo.listRepos", "params": { "subject": owner.as_ref() }, - "enrich": [{ "source": "sh.tangled.repo:.repo", "type": MINIDOC }] + "enrich": [{ "source": "org.tangled.repo.manifest:.repo", "type": MINIDOC }] }))) .await .unwrap(), @@ -633,7 +636,7 @@ async fn minidoc_repo_sources_skip_the_author_index() { assert_eq!(status, StatusCode::OK, "{body}"); assert_eq!( - body["data"]["did:plc:nel"]["sh.tangled.repo:.repo"][MINIDOC]["handle"], + body["data"]["did:plc:nel"]["org.tangled.repo.manifest:.repo"][MINIDOC]["handle"], json!("nel.example.com") ); @@ -642,7 +645,7 @@ async fn minidoc_repo_sources_skip_the_author_index() { app.oneshot(enrich_request(json!({ "xrpc": "sh.tangled.repo.listRepos", "params": { "subject": owner.as_ref() }, - "enrich": [{ "source": "sh.tangled.repo:.repo", "type": COUNT }] + "enrich": [{ "source": "org.tangled.repo.manifest:.repo", "type": COUNT }] }))) .await .unwrap(), @@ -651,20 +654,9 @@ async fn minidoc_repo_sources_skip_the_author_index() { assert_eq!(status, StatusCode::BAD_REQUEST, "{body}"); } -/// repo_fixture plus what views also need, the resolver knowing the repo did -/// and the issue counted as open +/// repo_fixture plus the issue counted as open, which views also need async fn viewable_repo_fixture(h: &Harness, owner: &Did, repo_did: &Did) { repo_fixture(h, owner, repo_did).await; - h.state - .resolver - .observe( - owner.clone(), - Rkey::new_owned("reef").unwrap(), - Some(repo_did.clone()), - Some("reef".into()), - &Datetime::raw_str("2026-05-01T00:00:00Z"), - ) - .await; h.edges.refresh_issue_counts( &h.state.issue_states, &at("at://did:plc:a/sh.tangled.repo.issue/i0"), @@ -684,9 +676,10 @@ async fn hydrates_repo_views_by_repo_did() { "xrpc": "sh.tangled.repo.listRepos", "params": { "subject": owner.as_ref() }, "enrich": [{ - "source": "sh.tangled.repo:.", + "source": "org.tangled.repo.manifest:.", "type": REPO_VIEW, - "targets": ["items[].value.repoDid"] + // the manifest uri's authority is the repo did + "targets": ["items[].uri"] }], "viewer": "did:plc:b" }))) @@ -696,7 +689,7 @@ async fn hydrates_repo_views_by_repo_did() { .await; assert_eq!(status, StatusCode::OK, "{body}"); - let view = &body["data"][repo_did.as_str()]["sh.tangled.repo:."][REPO_VIEW]; + let view = &body["data"][repo_did.as_str()]["org.tangled.repo.manifest:."][REPO_VIEW]; assert_eq!(view["did"], json!(repo_did.as_str()), "{body}"); assert_eq!(view["slug"], json!("reef")); assert_eq!(view["owner"]["did"], json!(owner.as_str())); @@ -722,7 +715,7 @@ async fn hydrates_repo_views_by_record_uri() { "xrpc": "sh.tangled.repo.listRepos", "params": { "subject": owner.as_ref() }, "enrich": [{ - "source": "sh.tangled.repo:.", + "source": "org.tangled.repo.manifest:.", "type": REPO_VIEW, "targets": ["items[].uri"] }] @@ -733,12 +726,12 @@ async fn hydrates_repo_views_by_record_uri() { .await; assert_eq!(status, StatusCode::OK, "{body}"); - let uri = "at://did:plc:nel/sh.tangled.repo/reef"; + let uri = "at://did:plc:limpet/org.tangled.repo.manifest/self"; assert_eq!( - body["data"][uri]["sh.tangled.repo:."][REPO_VIEW]["did"], + body["data"][uri]["org.tangled.repo.manifest:."][REPO_VIEW]["did"], json!(repo_did.as_str()), "{body}" ); - // the owner did rides along as the uri's authority but names no repo + // the owner did is not targeted, so it gets no view assert!(body["data"][owner.as_str()].is_null(), "{body}"); } diff --git a/bobbin/crates/xrpc/tests/trending.rs b/bobbin/crates/xrpc/tests/trending.rs index 040e97019..59d2ef3c8 100644 --- a/bobbin/crates/xrpc/tests/trending.rs +++ b/bobbin/crates/xrpc/tests/trending.rs @@ -13,10 +13,8 @@ use bobbin_types::ids::SubjectRef; use bobbin_xrpc::{AppState, router}; use http::{Request, StatusCode}; use jacquard_common::DefaultStr; -use jacquard_common::types::datetime::Datetime; use jacquard_common::types::did::Did; use jacquard_common::types::nsid::Nsid; -use jacquard_common::types::recordkey::Rkey; use jacquard_common::types::string::AtUri; use serde_json::{Value, json}; use tower::ServiceExt; @@ -38,10 +36,6 @@ fn nsid(value: &'static str) -> Nsid { Nsid::new_static(value).unwrap() } -fn rkey(value: &str) -> Rkey { - Rkey::new_owned(value).unwrap() -} - fn now_micros() -> u64 { let elapsed = SystemTime::now().duration_since(UNIX_EPOCH).unwrap(); elapsed @@ -53,7 +47,6 @@ fn now_micros() -> u64 { struct Harness { server: MockServer, edges: Arc, - resolver: Arc, state: AppState, } @@ -72,13 +65,12 @@ impl Harness { Arc::new( SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(), ) as Arc, - resolver.clone(), + resolver, Arc::new(bobbin_xrpc::default_directory()), ); Self { server, edges, - resolver, state, } } @@ -92,40 +84,18 @@ impl Harness { }); } - async fn mount_repo( - &self, - owner: Did, - record_key: Rkey, - repo_did: Did, - name: &str, - ) { - self.resolver - .observe( - owner.clone(), - record_key.clone(), - Some(repo_did.clone()), - Some(name.into()), - &Datetime::raw_str("2026-05-01T00:00:00Z"), - ) - .await; - let uri = format!( - "at://{}/sh.tangled.repo/{}", - owner.as_ref(), - record_key.as_ref() - ); + async fn mount_repo(&self, owner: Did, repo_did: Did, slug: &str) { Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) - .and(query_param("repo", owner.as_ref())) - .and(query_param("collection", "sh.tangled.repo")) - .and(query_param("rkey", record_key.as_ref())) + .and(query_param("repo", repo_did.as_ref())) + .and(query_param("collection", "org.tangled.repo.manifest")) + .and(query_param("rkey", "self")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ - "uri": uri, + "uri": format!("at://{}/org.tangled.repo.manifest/self", repo_did.as_ref()), "cid": CID, "value": { - "$type": "sh.tangled.repo", - "name": name, - "knot": "knot.example.com", - "repoDid": repo_did.as_ref(), + "$type": "org.tangled.repo.manifest", + "declaration": { "owner": owner.as_ref(), "slug": slug }, "createdAt": "2026-08-01T00:00:00Z" } }))) @@ -188,20 +158,10 @@ async fn get_trending_returns_fixed_viewer_neutral_candidates() { }); harness - .mount_repo( - did("did:plc:owner"), - rkey("compiler"), - compiler.clone(), - "compiler", - ) + .mount_repo(did("did:plc:owner"), compiler.clone(), "compiler") .await; harness - .mount_repo( - did("did:plc:maker"), - rkey("editor"), - editor.clone(), - "editor", - ) + .mount_repo(did("did:plc:maker"), editor.clone(), "editor") .await; let response = router(harness.state) @@ -291,11 +251,11 @@ async fn trending_repos_count_open_issues_open_pulls_and_forks() { harness.add_edge( "sh.tangled.repo.source", reef.clone(), - "at://did:plc:c/sh.tangled.repo/fork", + "at://did:plc:c/org.tangled.repo.manifest/self", now, ); harness - .mount_repo(did("did:plc:owner"), rkey("reef"), reef.clone(), "reef") + .mount_repo(did("did:plc:owner"), reef.clone(), "reef") .await; let response = router(harness.state.clone())