From 889b4a136708dbe8d945a4775f309da32ce8adf1 Mon Sep 17 00:00:00 2001 From: Lewis Date: Wed, 20 May 2026 10:11:51 +0300 Subject: [PATCH] refactor(xrpc): newtypes in aggregation tests Lewis: May this revision serve well! --- crates/xrpc/tests/aggregation.rs | 627 +++++++++++++++++++------------ 1 file changed, 386 insertions(+), 241 deletions(-) diff --git a/crates/xrpc/tests/aggregation.rs b/crates/xrpc/tests/aggregation.rs index 571cbe6..9c7f060 100644 --- a/crates/xrpc/tests/aggregation.rs +++ b/crates/xrpc/tests/aggregation.rs @@ -19,6 +19,7 @@ use http::{Request, StatusCode}; use jacquard_common::DefaultStr; 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; @@ -33,6 +34,14 @@ fn at(s: &str) -> AtUri { AtUri::new_owned(s).unwrap() } +fn did(s: &str) -> Did { + Did::new_owned(s).unwrap() +} + +fn rkey(s: &str) -> Rkey { + Rkey::new_owned(s).unwrap() +} + fn nsid(s: &'static str) -> Nsid { Nsid::new_static(s).unwrap() } @@ -92,23 +101,34 @@ impl Harness { } } - fn add_edge(&self, kind: &'static str, subject: &str, source: &str) { + fn add_edge(&self, kind: &Nsid, subject: &AtUri, source: &AtUri) { self.edges.add(Edge { - kind: nsid(kind), - subject: subj(subject), - source: at(source), + kind: kind.clone(), + subject: subj(subject.as_ref()), + source: source.clone(), sort_micros: next_sort_micros(), }); } - async fn mount(&self, did: &str, collection: &str, rkey: &str, value: Value) { - let uri = format!("at://{did}/{collection}/{rkey}"); + async fn mount( + &self, + did: &Did, + collection: &Nsid, + rkey: &Rkey, + value: Value, + ) { + let uri = format!( + "at://{}/{}/{}", + did.as_ref(), + collection.as_ref(), + rkey.as_ref() + ); let body = json!({ "uri": uri, "cid": CID, "value": value }); Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) - .and(query_param("repo", did)) - .and(query_param("collection", collection)) - .and(query_param("rkey", rkey)) + .and(query_param("repo", did.as_ref())) + .and(query_param("collection", collection.as_ref())) + .and(query_param("rkey", rkey.as_ref())) .respond_with(ResponseTemplate::new(200).set_body_json(body)) .mount(&self.server) .await; @@ -154,44 +174,44 @@ async fn json_response(resp: axum::response::Response) -> (StatusCode, Value) { (status, parsed) } -fn issue_body(repo_did: &str, title: &str) -> Value { +fn issue_body(repo_did: &Did, title: &str) -> Value { json!({ "$type": "sh.tangled.repo.issue", - "repo": repo_did, + "repo": repo_did.as_ref(), "title": title, "createdAt": "2026-05-01T00:00:00Z" }) } -fn pull_body(repo_did: &str, title: &str) -> Value { +fn pull_body(repo_did: &Did, title: &str) -> Value { json!({ "$type": "sh.tangled.repo.pull", "title": title, "createdAt": "2026-05-01T00:00:00Z", "rounds": [], "target": { - "repo": repo_did, + "repo": repo_did.as_ref(), "branch": "main" } }) } -fn star_body(subject_did: &str) -> Value { +fn star_body(subject_did: &Did) -> Value { json!({ "$type": "sh.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": { "$type": "sh.tangled.feed.star#repo", - "did": subject_did + "did": subject_did.as_ref() } }) } -fn follow_body(subject_did: &str) -> Value { +fn follow_body(subject_did: &Did) -> Value { json!({ "$type": "sh.tangled.graph.follow", "createdAt": "2026-05-01T00:00:00Z", - "subject": subject_did + "subject": subject_did.as_ref() }) } @@ -234,31 +254,47 @@ async fn count_issues_with_no_edges_returns_zero() { #[tokio::test] async fn list_issues_hydrates_via_slingshot_when_edges_present() { let h = Harness::new().await; - let repo = "did:plc:abalone"; - let subject = format!("at://{repo}"); + let repo = did("did:plc:abalone"); + let subject = at(&format!("at://{}", repo.as_ref())); let owners = [ ("did:plc:nel", "i1", "first"), ("did:plc:olaren", "i2", "second"), ]; stream::iter(owners) - .for_each(|(did, rkey, title)| { + .for_each(|(d, r, title)| { let h = &h; let subject = subject.clone(); + let repo = repo.clone(); async move { + let d_did = did(d); + let rk = rkey(r); h.add_edge( - "sh.tangled.repo.issue", + &nsid("sh.tangled.repo.issue"), &subject, - &format!("at://{did}/sh.tangled.repo.issue/{rkey}"), + &at(&format!( + "at://{}/sh.tangled.repo.issue/{}", + d_did.as_ref(), + rk.as_ref() + )), ); - h.mount(did, "sh.tangled.repo.issue", rkey, issue_body(repo, title)) - .await; + h.mount( + &d_did, + &nsid("sh.tangled.repo.issue"), + &rk, + issue_body(&repo, title), + ) + .await; } }) .await; let app = router(h.state.clone()); let resp = app - .oneshot(list_request("sh.tangled.repo.listIssues", &subject, &[])) + .oneshot(list_request( + "sh.tangled.repo.listIssues", + subject.as_ref(), + &[], + )) .await .unwrap(); let (status, body) = json_response(resp).await; @@ -278,26 +314,30 @@ async fn list_issues_hydrates_via_slingshot_when_edges_present() { #[tokio::test] async fn count_distinct_authors_dedupes_per_author() { let h = Harness::new().await; - let subject = "at://did:plc:abalone"; + let subject = at("at://did:plc:abalone"); h.add_edge( - "sh.tangled.feed.star", - subject, - "at://did:plc:nel/sh.tangled.feed.star/s1", + &nsid("sh.tangled.feed.star"), + &subject, + &at("at://did:plc:nel/sh.tangled.feed.star/s1"), ); h.add_edge( - "sh.tangled.feed.star", - subject, - "at://did:plc:nel/sh.tangled.feed.star/s2", + &nsid("sh.tangled.feed.star"), + &subject, + &at("at://did:plc:nel/sh.tangled.feed.star/s2"), ); h.add_edge( - "sh.tangled.feed.star", - subject, - "at://did:plc:olaren/sh.tangled.feed.star/s3", + &nsid("sh.tangled.feed.star"), + &subject, + &at("at://did:plc:olaren/sh.tangled.feed.star/s3"), ); let app = router(h.state.clone()); let resp = app - .oneshot(list_request("sh.tangled.feed.countStars", subject, &[])) + .oneshot(list_request( + "sh.tangled.feed.countStars", + subject.as_ref(), + &[], + )) .await .unwrap(); let (_, body) = json_response(resp).await; @@ -308,18 +348,21 @@ async fn count_distinct_authors_dedupes_per_author() { #[tokio::test] async fn list_items_stable_across_coverage_promotion() { let h = Harness::new().await; - let subject = "at://did:plc:abalone"; - let did = "did:plc:nel"; + let subject = at("at://did:plc:abalone"); + let nel = did("did:plc:nel"); h.add_edge( - "sh.tangled.feed.star", - subject, - &format!("at://{did}/sh.tangled.feed.star/s1"), + &nsid("sh.tangled.feed.star"), + &subject, + &at(&format!( + "at://{}/sh.tangled.feed.star/s1", + nel.as_ref() + )), ); h.mount( - did, - "sh.tangled.feed.star", - "s1", - star_body("did:plc:abalone"), + &nel, + &nsid("sh.tangled.feed.star"), + &rkey("s1"), + star_body(&did("did:plc:abalone")), ) .await; @@ -327,7 +370,11 @@ async fn list_items_stable_across_coverage_promotion() { h.warming(1, 5); let (_, before) = json_response( app.clone() - .oneshot(list_request("sh.tangled.feed.listStars", subject, &[])) + .oneshot(list_request( + "sh.tangled.feed.listStars", + subject.as_ref(), + &[], + )) .await .unwrap(), ) @@ -336,9 +383,13 @@ async fn list_items_stable_across_coverage_promotion() { h.promote_ready(2, 9); let (_, after) = json_response( - app.oneshot(list_request("sh.tangled.feed.listStars", subject, &[])) - .await - .unwrap(), + app.oneshot(list_request( + "sh.tangled.feed.listStars", + subject.as_ref(), + &[], + )) + .await + .unwrap(), ) .await; assert_eq!( @@ -351,7 +402,8 @@ async fn list_items_stable_across_coverage_promotion() { #[tokio::test] async fn list_paginates_via_cursor() { let h = Harness::new().await; - let subject = "at://did:plc:abalone"; + let subject = at("at://did:plc:abalone"); + let repo = did("did:plc:abalone"); let owners = [ ("did:plc:nel", "i1"), ("did:plc:olaren", "i2"), @@ -360,19 +412,27 @@ async fn list_paginates_via_cursor() { ("did:plc:bailey", "i5"), ]; stream::iter(owners) - .for_each(|(did, rkey)| { + .for_each(|(d, r)| { let h = &h; + let subject = subject.clone(); + let repo = repo.clone(); async move { + let d_did = did(d); + let rk = rkey(r); h.add_edge( - "sh.tangled.repo.issue", - subject, - &format!("at://{did}/sh.tangled.repo.issue/{rkey}"), + &nsid("sh.tangled.repo.issue"), + &subject, + &at(&format!( + "at://{}/sh.tangled.repo.issue/{}", + d_did.as_ref(), + rk.as_ref() + )), ); h.mount( - did, - "sh.tangled.repo.issue", - rkey, - issue_body("did:plc:abalone", &format!("issue-{rkey}")), + &d_did, + &nsid("sh.tangled.repo.issue"), + &rk, + issue_body(&repo, &format!("issue-{}", rk.as_ref())), ) .await; } @@ -384,7 +444,7 @@ async fn list_paginates_via_cursor() { app.clone() .oneshot(list_request( "sh.tangled.repo.listIssues", - subject, + subject.as_ref(), &[("limit", "2")], )) .await @@ -405,7 +465,7 @@ async fn list_paginates_via_cursor() { let (_, page2) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssues", - subject, + subject.as_ref(), &[("limit", "10"), ("cursor", &cursor)], )) .await @@ -431,22 +491,31 @@ async fn list_paginates_via_cursor() { #[tokio::test] async fn pagination_unaffected_by_coverage_promotion() { let h = Harness::new().await; - let subject = "at://did:plc:abalone"; + let subject = at("at://did:plc:abalone"); + let repo = did("did:plc:abalone"); let owners = [("did:plc:nel", "i1"), ("did:plc:olaren", "i2")]; stream::iter(owners) - .for_each(|(did, rkey)| { + .for_each(|(d, r)| { let h = &h; + let subject = subject.clone(); + let repo = repo.clone(); async move { + let d_did = did(d); + let rk = rkey(r); h.add_edge( - "sh.tangled.repo.issue", - subject, - &format!("at://{did}/sh.tangled.repo.issue/{rkey}"), + &nsid("sh.tangled.repo.issue"), + &subject, + &at(&format!( + "at://{}/sh.tangled.repo.issue/{}", + d_did.as_ref(), + rk.as_ref() + )), ); h.mount( - did, - "sh.tangled.repo.issue", - rkey, - issue_body("did:plc:abalone", &format!("issue-{rkey}")), + &d_did, + &nsid("sh.tangled.repo.issue"), + &rk, + issue_body(&repo, &format!("issue-{}", rk.as_ref())), ) .await; } @@ -459,7 +528,7 @@ async fn pagination_unaffected_by_coverage_promotion() { app.clone() .oneshot(list_request( "sh.tangled.repo.listIssues", - subject, + subject.as_ref(), &[("limit", "1")], )) .await @@ -472,7 +541,7 @@ async fn pagination_unaffected_by_coverage_promotion() { let (_, page2) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssues", - subject, + subject.as_ref(), &[("limit", "10"), ("cursor", &cursor)], )) .await @@ -502,42 +571,46 @@ async fn invalid_cursor_returns_400() { #[tokio::test] async fn list_follows_subject_is_followee_did() { let h = Harness::new().await; - let followee = "did:plc:bailey"; - let subject = format!("at://{followee}"); + let followee = did("did:plc:bailey"); + let subject = at(&format!("at://{}", followee.as_ref())); h.add_edge( - "sh.tangled.graph.follow", + &nsid("sh.tangled.graph.follow"), &subject, - "at://did:plc:nel/sh.tangled.graph.follow/f1", + &at("at://did:plc:nel/sh.tangled.graph.follow/f1"), ); h.mount( - "did:plc:nel", - "sh.tangled.graph.follow", - "f1", - follow_body(followee), + &did("did:plc:nel"), + &nsid("sh.tangled.graph.follow"), + &rkey("f1"), + follow_body(&followee), ) .await; let app = router(h.state.clone()); let (status, body) = json_response( - app.oneshot(list_request("sh.tangled.graph.listFollows", &subject, &[])) - .await - .unwrap(), + app.oneshot(list_request( + "sh.tangled.graph.listFollows", + subject.as_ref(), + &[], + )) + .await + .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().unwrap(); assert_eq!(items.len(), 1); - assert_eq!(items[0]["value"]["subject"], followee); + assert_eq!(items[0]["value"]["subject"], followee.as_ref()); } #[tokio::test] async fn upstream_failure_during_hydration_propagates() { let h = Harness::new().await; - let subject = "at://did:plc:abalone"; + let subject = at("at://did:plc:abalone"); h.add_edge( - "sh.tangled.repo.issue", - subject, - "at://did:plc:nel/sh.tangled.repo.issue/missing", + &nsid("sh.tangled.repo.issue"), + &subject, + &at("at://did:plc:nel/sh.tangled.repo.issue/missing"), ); Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) @@ -547,9 +620,13 @@ async fn upstream_failure_during_hydration_propagates() { let app = router(h.state.clone()); let (status, body) = json_response( - app.oneshot(list_request("sh.tangled.repo.listIssues", subject, &[])) - .await - .unwrap(), + app.oneshot(list_request( + "sh.tangled.repo.listIssues", + subject.as_ref(), + &[], + )) + .await + .unwrap(), ) .await; assert_eq!(status, StatusCode::BAD_GATEWAY); @@ -637,16 +714,20 @@ async fn limit_below_min_or_above_max_is_400() { #[tokio::test] async fn count_after_remove_source_returns_zero() { let h = Harness::new().await; - let subject = "at://did:plc:abalone"; - let source = "at://did:plc:nel/sh.tangled.feed.star/s1"; - h.add_edge("sh.tangled.feed.star", subject, source); - h.edges.remove_source(&at(source)); + let subject = at("at://did:plc:abalone"); + let source = at("at://did:plc:nel/sh.tangled.feed.star/s1"); + h.add_edge(&nsid("sh.tangled.feed.star"), &subject, &source); + h.edges.remove_source(&source); let app = router(h.state.clone()); let (_, body) = json_response( - app.oneshot(list_request("sh.tangled.feed.countStars", subject, &[])) - .await - .unwrap(), + app.oneshot(list_request( + "sh.tangled.feed.countStars", + subject.as_ref(), + &[], + )) + .await + .unwrap(), ) .await; assert_eq!(body["count"], json!(0)); @@ -656,21 +737,25 @@ async fn count_after_remove_source_returns_zero() { #[tokio::test] async fn list_issue_comments_hydrates_end_to_end() { let h = Harness::new().await; - let issue_uri = "at://did:plc:abalone/sh.tangled.repo.issue/i1"; - let did = "did:plc:nel"; - let rkey = "c1"; + let issue_uri = at("at://did:plc:abalone/sh.tangled.repo.issue/i1"); + let nel = did("did:plc:nel"); + let rk = rkey("c1"); h.add_edge( - "sh.tangled.repo.issue.comment", - issue_uri, - &format!("at://{did}/sh.tangled.repo.issue.comment/{rkey}"), + &nsid("sh.tangled.repo.issue.comment"), + &issue_uri, + &at(&format!( + "at://{}/sh.tangled.repo.issue.comment/{}", + nel.as_ref(), + rk.as_ref() + )), ); h.mount( - did, - "sh.tangled.repo.issue.comment", - rkey, + &nel, + &nsid("sh.tangled.repo.issue.comment"), + &rk, json!({ "$type": "sh.tangled.repo.issue.comment", - "issue": issue_uri, + "issue": issue_uri.as_ref(), "body": "thoughts", "createdAt": "2026-05-01T00:00:00Z" }), @@ -681,7 +766,7 @@ async fn list_issue_comments_hydrates_end_to_end() { let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.issue.listComments", - issue_uri, + issue_uri.as_ref(), &[], )) .await @@ -692,32 +777,39 @@ async fn list_issue_comments_hydrates_end_to_end() { let items = body["items"].as_array().unwrap(); assert_eq!(items.len(), 1); assert_eq!(items[0]["value"]["body"], json!("thoughts")); - assert_eq!(items[0]["value"]["issue"], json!(issue_uri)); + assert_eq!(items[0]["value"]["issue"], json!(issue_uri.as_ref())); } #[tokio::test] async fn list_item_cid_is_present() { let h = Harness::new().await; - let subject = "at://did:plc:abalone"; - let did = "did:plc:nel"; + let subject = at("at://did:plc:abalone"); + let nel = did("did:plc:nel"); h.add_edge( - "sh.tangled.feed.star", - subject, - &format!("at://{did}/sh.tangled.feed.star/s1"), + &nsid("sh.tangled.feed.star"), + &subject, + &at(&format!( + "at://{}/sh.tangled.feed.star/s1", + nel.as_ref() + )), ); h.mount( - did, - "sh.tangled.feed.star", - "s1", - star_body("did:plc:abalone"), + &nel, + &nsid("sh.tangled.feed.star"), + &rkey("s1"), + star_body(&did("did:plc:abalone")), ) .await; let app = router(h.state.clone()); let (_, body) = json_response( - app.oneshot(list_request("sh.tangled.feed.listStars", subject, &[])) - .await - .unwrap(), + app.oneshot(list_request( + "sh.tangled.feed.listStars", + subject.as_ref(), + &[], + )) + .await + .unwrap(), ) .await; let item = &body["items"][0]; @@ -731,23 +823,23 @@ async fn list_item_cid_is_present() { #[tokio::test] async fn count_issue_comments_subjects_on_issue_uri() { let h = Harness::new().await; - let issue_uri = "at://did:plc:abalone/sh.tangled.repo.issue/i1"; + let issue_uri = at("at://did:plc:abalone/sh.tangled.repo.issue/i1"); h.add_edge( - "sh.tangled.repo.issue.comment", - issue_uri, - "at://did:plc:nel/sh.tangled.repo.issue.comment/c1", + &nsid("sh.tangled.repo.issue.comment"), + &issue_uri, + &at("at://did:plc:nel/sh.tangled.repo.issue.comment/c1"), ); h.add_edge( - "sh.tangled.repo.issue.comment", - issue_uri, - "at://did:plc:olaren/sh.tangled.repo.issue.comment/c2", + &nsid("sh.tangled.repo.issue.comment"), + &issue_uri, + &at("at://did:plc:olaren/sh.tangled.repo.issue.comment/c2"), ); let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.issue.countComments", - issue_uri, + issue_uri.as_ref(), &[], )) .await @@ -762,9 +854,9 @@ async fn count_issue_comments_subjects_on_issue_uri() { #[tokio::test] async fn list_item_404_does_not_become_404_for_subject() { let h = Harness::new().await; - let subject = "at://did:plc:abalone"; - let stale_source = "at://did:plc:nel/sh.tangled.repo.issue/missing"; - h.add_edge("sh.tangled.repo.issue", subject, stale_source); + let subject = at("at://did:plc:abalone"); + let stale_source = at("at://did:plc:nel/sh.tangled.repo.issue/missing"); + h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &stale_source); Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .respond_with(ResponseTemplate::new(404).set_body_json(json!({ @@ -776,9 +868,13 @@ async fn list_item_404_does_not_become_404_for_subject() { let app = router(h.state.clone()); let (status, body) = json_response( - app.oneshot(list_request("sh.tangled.repo.listIssues", subject, &[])) - .await - .unwrap(), + app.oneshot(list_request( + "sh.tangled.repo.listIssues", + subject.as_ref(), + &[], + )) + .await + .unwrap(), ) .await; assert_eq!( @@ -791,7 +887,7 @@ async fn list_item_404_does_not_become_404_for_subject() { body["message"] .as_str() .unwrap_or_default() - .contains(stale_source), + .contains(stale_source.as_ref()), "message should name the missing source uri, got {}", body["message"], ); @@ -800,18 +896,22 @@ async fn list_item_404_does_not_become_404_for_subject() { #[tokio::test] async fn list_item_with_wrong_type_tag_propagates_invalid_record() { let h = Harness::new().await; - let subject = "at://did:plc:abalone"; - let did = "did:plc:nel"; - let rkey = "s1"; - let source = format!("at://{did}/sh.tangled.feed.star/{rkey}"); - h.add_edge("sh.tangled.feed.star", subject, &source); + let subject = at("at://did:plc:abalone"); + let nel = did("did:plc:nel"); + let rk = rkey("s1"); + let source = at(&format!( + "at://{}/sh.tangled.feed.star/{}", + nel.as_ref(), + rk.as_ref() + )); + h.add_edge(&nsid("sh.tangled.feed.star"), &subject, &source); Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) - .and(query_param("repo", did)) + .and(query_param("repo", nel.as_ref())) .and(query_param("collection", "sh.tangled.feed.star")) - .and(query_param("rkey", rkey)) + .and(query_param("rkey", rk.as_ref())) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ - "uri": source, + "uri": source.as_ref(), "cid": CID, "value": { "$type": "sh.tangled.feed.reaction", @@ -824,9 +924,13 @@ async fn list_item_with_wrong_type_tag_propagates_invalid_record() { let app = router(h.state.clone()); let (status, body) = json_response( - app.oneshot(list_request("sh.tangled.feed.listStars", subject, &[])) - .await - .unwrap(), + app.oneshot(list_request( + "sh.tangled.feed.listStars", + subject.as_ref(), + &[], + )) + .await + .unwrap(), ) .await; assert_eq!(status, StatusCode::BAD_GATEWAY); @@ -1054,16 +1158,20 @@ async fn star_endpoints_accept_string_subject_form() { #[tokio::test] async fn list_after_remove_source_returns_empty_items() { let h = Harness::new().await; - let subject = "at://did:plc:abalone"; - let source = "at://did:plc:nel/sh.tangled.feed.star/s1"; - h.add_edge("sh.tangled.feed.star", subject, source); - h.edges.remove_source(&at(source)); + let subject = at("at://did:plc:abalone"); + let source = at("at://did:plc:nel/sh.tangled.feed.star/s1"); + h.add_edge(&nsid("sh.tangled.feed.star"), &subject, &source); + h.edges.remove_source(&source); let app = router(h.state.clone()); let (status, body) = json_response( - app.oneshot(list_request("sh.tangled.feed.listStars", subject, &[])) - .await - .unwrap(), + app.oneshot(list_request( + "sh.tangled.feed.listStars", + subject.as_ref(), + &[], + )) + .await + .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); @@ -1074,66 +1182,78 @@ async fn list_after_remove_source_returns_empty_items() { #[tokio::test] async fn list_pulls_hydrates_via_slingshot_when_edges_present() { let h = Harness::new().await; - let target_did = "did:plc:abalone"; - let subject = format!("at://{target_did}"); - let source_did = "did:plc:nel"; - let rkey = "p1"; + let target_did = did("did:plc:abalone"); + let subject = at(&format!("at://{}", target_did.as_ref())); + let source_did = did("did:plc:nel"); + let rk = rkey("p1"); h.add_edge( - "sh.tangled.repo.pull", + &nsid("sh.tangled.repo.pull"), &subject, - &format!("at://{source_did}/sh.tangled.repo.pull/{rkey}"), + &at(&format!( + "at://{}/sh.tangled.repo.pull/{}", + source_did.as_ref(), + rk.as_ref() + )), ); h.mount( - source_did, - "sh.tangled.repo.pull", - rkey, + &source_did, + &nsid("sh.tangled.repo.pull"), + &rk, json!({ "$type": "sh.tangled.repo.pull", "title": "ship it", "createdAt": "2026-05-01T00:00:00Z", "rounds": [], - "target": {"repo": target_did, "branch": "main"}, + "target": {"repo": target_did.as_ref(), "branch": "main"}, }), ) .await; let app = router(h.state.clone()); let (status, body) = json_response( - app.oneshot(list_request("sh.tangled.repo.listPulls", &subject, &[])) - .await - .unwrap(), + app.oneshot(list_request( + "sh.tangled.repo.listPulls", + subject.as_ref(), + &[], + )) + .await + .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().unwrap(); assert_eq!(items.len(), 1); assert_eq!(items[0]["value"]["title"], json!("ship it")); - assert_eq!(items[0]["value"]["target"]["repo"], json!(target_did)); + assert_eq!(items[0]["value"]["target"]["repo"], json!(target_did.as_ref())); } #[tokio::test] async fn count_pulls_returns_distinct_authors() { let h = Harness::new().await; - let subject = "at://did:plc:abalone"; + let subject = at("at://did:plc:abalone"); h.add_edge( - "sh.tangled.repo.pull", - subject, - "at://did:plc:nel/sh.tangled.repo.pull/p1", + &nsid("sh.tangled.repo.pull"), + &subject, + &at("at://did:plc:nel/sh.tangled.repo.pull/p1"), ); h.add_edge( - "sh.tangled.repo.pull", - subject, - "at://did:plc:olaren/sh.tangled.repo.pull/p2", + &nsid("sh.tangled.repo.pull"), + &subject, + &at("at://did:plc:olaren/sh.tangled.repo.pull/p2"), ); h.add_edge( - "sh.tangled.repo.pull", - subject, - "at://did:plc:nel/sh.tangled.repo.pull/p3", + &nsid("sh.tangled.repo.pull"), + &subject, + &at("at://did:plc:nel/sh.tangled.repo.pull/p3"), ); let app = router(h.state.clone()); let (_, body) = json_response( - app.oneshot(list_request("sh.tangled.repo.countPulls", subject, &[])) - .await - .unwrap(), + app.oneshot(list_request( + "sh.tangled.repo.countPulls", + subject.as_ref(), + &[], + )) + .await + .unwrap(), ) .await; assert_eq!(body["count"], json!(3)); @@ -1143,27 +1263,31 @@ async fn count_pulls_returns_distinct_authors() { #[tokio::test] async fn extractor_to_xrpc_round_trip_for_star() { let h = Harness::new().await; - let subject_did = "did:plc:abalone"; - let source_did = "did:plc:nel"; - let rkey = "s1"; - let source = format!("at://{source_did}/sh.tangled.feed.star/{rkey}"); - let body = star_body(subject_did); + let subject_did = did("did:plc:abalone"); + let source_did = did("did:plc:nel"); + let rk = rkey("s1"); + let source = at(&format!( + "at://{}/sh.tangled.feed.star/{}", + source_did.as_ref(), + rk.as_ref() + )); + let body = star_body(&subject_did); let parsed = bobbin_types::edges::Record::from_json_value(&nsid("sh.tangled.feed.star"), body.clone()) .expect("parse star record"); parsed - .extract_edges(&at(&source)) + .extract_edges(&source) .expect("extract") .into_iter() .for_each(|e| h.edges.add(e)); - h.mount(source_did, "sh.tangled.feed.star", rkey, body) + h.mount(&source_did, &nsid("sh.tangled.feed.star"), &rk, body) .await; let app = router(h.state.clone()); let (status, json) = json_response( app.oneshot(list_request( "sh.tangled.feed.listStars", - &format!("at://{subject_did}"), + &format!("at://{}", subject_did.as_ref()), &[], )) .await @@ -1177,35 +1301,34 @@ async fn extractor_to_xrpc_round_trip_for_star() { ); let items = json["items"].as_array().unwrap(); assert_eq!(items.len(), 1, "expected exactly one star edge"); - assert_eq!(items[0]["value"]["subject"]["did"], json!(subject_did)); + assert_eq!(items[0]["value"]["subject"]["did"], json!(subject_did.as_ref())); } #[tokio::test] async fn list_issues_includes_state_comment_count_and_state_updated_at() { let h = Harness::new().await; - let repo = "did:plc:limpet"; - let subject = format!("at://{repo}"); - let issue_uri_str = "at://did:plc:nel/sh.tangled.repo.issue/i1".to_owned(); - h.add_edge("sh.tangled.repo.issue", &subject, &issue_uri_str); + let repo = did("did:plc:limpet"); + let subject = at(&format!("at://{}", repo.as_ref())); + let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); + h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri); h.mount( - "did:plc:nel", - "sh.tangled.repo.issue", - "i1", - issue_body(repo, "hi"), + &did("did:plc:nel"), + &nsid("sh.tangled.repo.issue"), + &rkey("i1"), + issue_body(&repo, "hi"), ) .await; h.add_edge( - "sh.tangled.repo.issue.comment", - &issue_uri_str, - "at://did:plc:olaren/sh.tangled.repo.issue.comment/c1", + &nsid("sh.tangled.repo.issue.comment"), + &issue_uri, + &at("at://did:plc:olaren/sh.tangled.repo.issue.comment/c1"), ); h.add_edge( - "sh.tangled.repo.issue.comment", - &issue_uri_str, - "at://did:plc:teq/sh.tangled.repo.issue.comment/c2", + &nsid("sh.tangled.repo.issue.comment"), + &issue_uri, + &at("at://did:plc:teq/sh.tangled.repo.issue.comment/c2"), ); - let issue_uri = at(&issue_uri_str); h.state.issue_states.upsert( at("at://did:plc:nel/sh.tangled.repo.issue.state/s1"), issue_uri.clone(), @@ -1221,7 +1344,11 @@ async fn list_issues_includes_state_comment_count_and_state_updated_at() { let app = router(h.state.clone()); let resp = app - .oneshot(list_request("sh.tangled.repo.listIssues", &subject, &[])) + .oneshot(list_request( + "sh.tangled.repo.listIssues", + subject.as_ref(), + &[], + )) .await .unwrap(); let (status, body) = json_response(resp).await; @@ -1241,21 +1368,29 @@ async fn list_issues_includes_state_comment_count_and_state_updated_at() { #[tokio::test] async fn list_issues_omits_state_when_no_state_record() { let h = Harness::new().await; - let repo = "did:plc:limpet"; - let subject = format!("at://{repo}"); - let issue_uri_str = "at://did:plc:nel/sh.tangled.repo.issue/i1".to_owned(); - h.add_edge("sh.tangled.repo.issue", &subject, &issue_uri_str); + let repo = did("did:plc:limpet"); + let subject = at(&format!("at://{}", repo.as_ref())); + let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); + h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri); h.mount( - "did:plc:nel", - "sh.tangled.repo.issue", - "i1", - issue_body(repo, "no state yet"), + &did("did:plc:nel"), + &nsid("sh.tangled.repo.issue"), + &rkey("i1"), + issue_body(&repo, "no state yet"), ) .await; let app = router(h.state.clone()); - let (_status, body) = - json_response(app.oneshot(list_request("sh.tangled.repo.listIssues", &subject, &[])).await.unwrap()).await; + let (_status, body) = json_response( + app.oneshot(list_request( + "sh.tangled.repo.listIssues", + subject.as_ref(), + &[], + )) + .await + .unwrap(), + ) + .await; let item = &body["items"][0]; assert!(item.get("state").is_none(), "state must be absent"); assert!( @@ -1268,8 +1403,8 @@ async fn list_issues_omits_state_when_no_state_record() { #[tokio::test] async fn list_issues_author_filter_restricts_to_matching_did() { let h = Harness::new().await; - let repo = "did:plc:limpet"; - let subject = format!("at://{repo}"); + let repo = did("did:plc:limpet"); + let subject = at(&format!("at://{}", repo.as_ref())); let owners = [ ("did:plc:nel", "n1"), ("did:plc:nel", "n2"), @@ -1277,20 +1412,27 @@ async fn list_issues_author_filter_restricts_to_matching_did() { ("did:plc:olaren", "o2"), ]; stream::iter(owners) - .for_each(|(did, rkey)| { + .for_each(|(d, r)| { let h = &h; let subject = subject.clone(); + let repo = repo.clone(); async move { + let d_did = did(d); + let rk = rkey(r); h.add_edge( - "sh.tangled.repo.issue", + &nsid("sh.tangled.repo.issue"), &subject, - &format!("at://{did}/sh.tangled.repo.issue/{rkey}"), + &at(&format!( + "at://{}/sh.tangled.repo.issue/{}", + d_did.as_ref(), + rk.as_ref() + )), ); h.mount( - did, - "sh.tangled.repo.issue", - rkey, - issue_body(repo, &format!("issue-{rkey}")), + &d_did, + &nsid("sh.tangled.repo.issue"), + &rk, + issue_body(&repo, &format!("issue-{}", rk.as_ref())), ) .await; } @@ -1301,7 +1443,7 @@ async fn list_issues_author_filter_restricts_to_matching_did() { let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssues", - &subject, + subject.as_ref(), &[("author", "did:plc:nel")], )) .await @@ -1336,23 +1478,22 @@ async fn list_issues_invalid_author_returns_400() { #[tokio::test] async fn list_pulls_includes_merged_state_and_comment_count() { let h = Harness::new().await; - let repo = "did:plc:limpet"; - let subject = format!("at://{repo}"); - let pull_uri_str = "at://did:plc:nel/sh.tangled.repo.pull/p1".to_owned(); - h.add_edge("sh.tangled.repo.pull", &subject, &pull_uri_str); + let repo = did("did:plc:limpet"); + let subject = at(&format!("at://{}", repo.as_ref())); + let pull_uri = at("at://did:plc:nel/sh.tangled.repo.pull/p1"); + h.add_edge(&nsid("sh.tangled.repo.pull"), &subject, &pull_uri); h.mount( - "did:plc:nel", - "sh.tangled.repo.pull", - "p1", - pull_body(repo, "fix bug"), + &did("did:plc:nel"), + &nsid("sh.tangled.repo.pull"), + &rkey("p1"), + pull_body(&repo, "fix bug"), ) .await; h.add_edge( - "sh.tangled.repo.pull.comment", - &pull_uri_str, - "at://did:plc:teq/sh.tangled.repo.pull.comment/c1", + &nsid("sh.tangled.repo.pull.comment"), + &pull_uri, + &at("at://did:plc:teq/sh.tangled.repo.pull.comment/c1"), ); - let pull_uri = at(&pull_uri_str); h.state.pull_statuses.upsert( at("at://did:plc:nel/sh.tangled.repo.pull.status/s1"), pull_uri.clone(), @@ -1368,9 +1509,13 @@ async fn list_pulls_includes_merged_state_and_comment_count() { let app = router(h.state.clone()); let (status, body) = json_response( - app.oneshot(list_request("sh.tangled.repo.listPulls", &subject, &[])) - .await - .unwrap(), + app.oneshot(list_request( + "sh.tangled.repo.listPulls", + subject.as_ref(), + &[], + )) + .await + .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); -- 2.51.2