alice.pds.demo.boltless.dev
This repository has no description
1use std::sync::Arc;
2
3use axum::body::{Body, to_bytes};
4use bobbin_edge_index::{
5 Coverage, CoverageWatch, EdgeStore, HydrantCursor, IssueStateKind, PageToken, PullStatusKind,
6 StateIndex,
7};
8use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig};
9use bobbin_record_lru::{CacheCapacity, LruRecordStore};
10use bobbin_resolver::RepoIdResolver;
11use bobbin_runtime::{RuntimeHasher, SystemClock};
12use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader};
13use bobbin_slingshot_client::SlingshotClient;
14use bobbin_types::edges::Edge;
15use bobbin_types::ids::SubjectRef;
16use bobbin_xrpc::{AppState, router};
17use futures::stream::{self, StreamExt};
18use http::{Request, StatusCode};
19use jacquard_common::DefaultStr;
20use jacquard_common::types::did::Did;
21use jacquard_common::types::nsid::Nsid;
22use jacquard_common::types::recordkey::Rkey;
23use jacquard_common::types::string::AtUri;
24use serde_json::{Value, json};
25use tower::ServiceExt;
26use url::Url;
27use url::form_urlencoded::byte_serialize;
28use wiremock::matchers::{method, path, query_param};
29use wiremock::{Mock, MockServer, ResponseTemplate};
30
31const CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i";
32
33fn at(s: &str) -> AtUri<DefaultStr> {
34 AtUri::new_owned(s).unwrap()
35}
36
37fn did(s: &str) -> Did<DefaultStr> {
38 Did::new_owned(s).unwrap()
39}
40
41fn rkey(s: &str) -> Rkey<DefaultStr> {
42 Rkey::new_owned(s).unwrap()
43}
44
45fn nsid(s: &'static str) -> Nsid<DefaultStr> {
46 Nsid::new_static(s).unwrap()
47}
48
49fn subj(s: &str) -> SubjectRef {
50 Did::<DefaultStr>::new_owned(s)
51 .map(SubjectRef::Did)
52 .unwrap_or_else(|_| SubjectRef::Uri(AtUri::new_owned(s).unwrap()))
53}
54
55struct Harness {
56 server: MockServer,
57 edges: Arc<EdgeStore>,
58 coverage: Arc<CoverageWatch>,
59 state: AppState,
60}
61
62static EDGE_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
63
64fn next_sort_micros() -> u64 {
65 EDGE_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
66}
67
68impl Harness {
69 async fn new() -> Self {
70 let server = MockServer::start().await;
71 let edges = Arc::new(EdgeStore::new(RuntimeHasher::default()));
72 let issue_states = Arc::new(StateIndex::new(RuntimeHasher::default()));
73 let pull_statuses = Arc::new(StateIndex::new(RuntimeHasher::default()));
74 let coverage = Arc::new(CoverageWatch::new());
75 let state = AppState::new(
76 Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))),
77 SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(),
78 edges.clone(),
79 issue_states.clone(),
80 pull_statuses.clone(),
81 coverage.clone(),
82 Arc::new(
83 KnotProxy::new(
84 KnotProxyConfig::default(),
85 KnotHttpConfig::default(),
86 Arc::new(SystemClock::new()),
87 RuntimeHasher::default(),
88 )
89 .unwrap(),
90 ),
91 Arc::new(
92 SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(),
93 ) as Arc<dyn SearchReader>,
94 Arc::new(RepoIdResolver::detached(RuntimeHasher::default())),
95 Arc::new(bobbin_xrpc::default_directory()),
96 );
97 Self {
98 server,
99 edges,
100 coverage,
101 state,
102 }
103 }
104
105 fn add_edge(
106 &self,
107 kind: &Nsid<DefaultStr>,
108 subject: &AtUri<DefaultStr>,
109 source: &AtUri<DefaultStr>,
110 ) {
111 self.edges.add(Edge {
112 kind: kind.clone(),
113 subject: subj(subject.as_ref()),
114 source: source.clone(),
115 sort_micros: next_sort_micros(),
116 });
117 match kind.as_ref() {
118 "sh.tangled.repo.issue" => self
119 .edges
120 .refresh_issue_counts(&self.state.issue_states, source),
121 "sh.tangled.repo.pull" => self
122 .edges
123 .refresh_pull_counts(&self.state.pull_statuses, source),
124 _ => {}
125 }
126 }
127
128 fn upsert_issue_state(
129 &self,
130 source: AtUri<DefaultStr>,
131 issue: AtUri<DefaultStr>,
132 sort_micros: u64,
133 kind: IssueStateKind,
134 ) {
135 self.state
136 .issue_states
137 .upsert(source, issue.clone(), sort_micros, kind);
138 self.edges
139 .refresh_issue_counts(&self.state.issue_states, &issue);
140 }
141
142 fn upsert_pull_status(
143 &self,
144 source: AtUri<DefaultStr>,
145 pull: AtUri<DefaultStr>,
146 sort_micros: u64,
147 kind: PullStatusKind,
148 ) {
149 self.state
150 .pull_statuses
151 .upsert(source, pull.clone(), sort_micros, kind);
152 self.edges
153 .refresh_pull_counts(&self.state.pull_statuses, &pull);
154 }
155
156 async fn mount(
157 &self,
158 did: &Did<DefaultStr>,
159 collection: &Nsid<DefaultStr>,
160 rkey: &Rkey<DefaultStr>,
161 value: Value,
162 ) {
163 let uri = format!(
164 "at://{}/{}/{}",
165 did.as_ref(),
166 collection.as_ref(),
167 rkey.as_ref()
168 );
169 let body = json!({ "uri": uri, "cid": CID, "value": value });
170 Mock::given(method("GET"))
171 .and(path("/xrpc/com.atproto.repo.getRecord"))
172 .and(query_param("repo", did.as_ref()))
173 .and(query_param("collection", collection.as_ref()))
174 .and(query_param("rkey", rkey.as_ref()))
175 .respond_with(ResponseTemplate::new(200).set_body_json(body))
176 .mount(&self.server)
177 .await;
178 }
179
180 fn promote_ready(&self, events: u64, cursor: u64) {
181 self.coverage.update(|_| Coverage::Ready {
182 events_processed: events,
183 last_cursor: HydrantCursor::new(cursor),
184 });
185 }
186
187 fn warming(&self, events: u64, cursor: u64) {
188 self.coverage.update(|_| Coverage::Warming {
189 events_processed: events,
190 last_cursor: HydrantCursor::new(cursor),
191 });
192 }
193}
194
195fn list_request(endpoint: &str, subject: &str, extras: &[(&str, &str)]) -> Request<Body> {
196 let mut qs = format!("subject={subject}");
197 extras.iter().for_each(|(k, v)| {
198 qs.push('&');
199 qs.push_str(k);
200 qs.push('=');
201 qs.push_str(&encode(v));
202 });
203 Request::builder()
204 .uri(format!("/xrpc/{endpoint}?{qs}"))
205 .body(Body::empty())
206 .unwrap()
207}
208
209fn encode(s: &str) -> String {
210 byte_serialize(s.as_bytes()).collect()
211}
212
213async fn json_response(resp: axum::response::Response) -> (StatusCode, Value) {
214 let status = resp.status();
215 let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap();
216 let parsed: Value = serde_json::from_slice(&bytes).expect("JSON body");
217 (status, parsed)
218}
219
220fn issue_body(repo_did: &Did<DefaultStr>, title: &str) -> Value {
221 json!({
222 "$type": "sh.tangled.repo.issue",
223 "repo": repo_did.as_ref(),
224 "title": title,
225 "createdAt": "2026-05-01T00:00:00Z"
226 })
227}
228
229fn pull_body(repo_did: &Did<DefaultStr>, title: &str) -> Value {
230 json!({
231 "$type": "sh.tangled.repo.pull",
232 "title": title,
233 "createdAt": "2026-05-01T00:00:00Z",
234 "rounds": [],
235 "target": {
236 "repo": repo_did.as_ref(),
237 "branch": "main"
238 }
239 })
240}
241
242fn star_body(subject_did: &Did<DefaultStr>) -> Value {
243 json!({
244 "$type": "sh.tangled.feed.star",
245 "createdAt": "2026-05-01T00:00:00Z",
246 "subject": {
247 "$type": "sh.tangled.feed.star#repo",
248 "did": subject_did.as_ref()
249 }
250 })
251}
252
253fn follow_body(subject_did: &Did<DefaultStr>) -> Value {
254 json!({
255 "$type": "sh.tangled.graph.follow",
256 "createdAt": "2026-05-01T00:00:00Z",
257 "subject": subject_did.as_ref()
258 })
259}
260
261#[tokio::test]
262async fn list_issues_with_no_edges_returns_empty_items() {
263 let h = Harness::new().await;
264 let app = router(h.state.clone());
265 let resp = app
266 .oneshot(list_request(
267 "sh.tangled.repo.listIssues",
268 "at://did:plc:abalone",
269 &[],
270 ))
271 .await
272 .unwrap();
273 let (status, body) = json_response(resp).await;
274 assert_eq!(status, StatusCode::OK);
275 assert_eq!(body["items"], json!([]));
276 assert!(body["cursor"].is_null());
277}
278
279#[tokio::test]
280async fn count_issues_with_no_edges_returns_zero() {
281 let h = Harness::new().await;
282 let app = router(h.state.clone());
283 let resp = app
284 .oneshot(list_request(
285 "sh.tangled.repo.countIssues",
286 "at://did:plc:abalone",
287 &[],
288 ))
289 .await
290 .unwrap();
291 let (status, body) = json_response(resp).await;
292 assert_eq!(status, StatusCode::OK);
293 assert_eq!(body["count"], json!(0));
294 assert_eq!(body["distinctAuthors"], json!(0));
295}
296
297#[tokio::test]
298async fn list_issues_hydrates_via_slingshot_when_edges_present() {
299 let h = Harness::new().await;
300 let repo = did("did:plc:abalone");
301 let subject = at(&format!("at://{}", repo.as_ref()));
302 let owners = [
303 ("did:plc:nel", "i1", "first"),
304 ("did:plc:olaren", "i2", "second"),
305 ];
306 stream::iter(owners)
307 .for_each(|(d, r, title)| {
308 let h = &h;
309 let subject = subject.clone();
310 let repo = repo.clone();
311 async move {
312 let d_did = did(d);
313 let rk = rkey(r);
314 h.add_edge(
315 &nsid("sh.tangled.repo.issue"),
316 &subject,
317 &at(&format!(
318 "at://{}/sh.tangled.repo.issue/{}",
319 d_did.as_ref(),
320 rk.as_ref()
321 )),
322 );
323 h.mount(
324 &d_did,
325 &nsid("sh.tangled.repo.issue"),
326 &rk,
327 issue_body(&repo, title),
328 )
329 .await;
330 }
331 })
332 .await;
333
334 let app = router(h.state.clone());
335 let resp = app
336 .oneshot(list_request(
337 "sh.tangled.repo.listIssues",
338 subject.as_ref(),
339 &[],
340 ))
341 .await
342 .unwrap();
343 let (status, body) = json_response(resp).await;
344 assert_eq!(status, StatusCode::OK);
345 let items = body["items"].as_array().expect("items array");
346 assert_eq!(items.len(), 2);
347 let titles: Vec<&str> = items
348 .iter()
349 .map(|v| v["value"]["title"].as_str().unwrap())
350 .collect();
351 assert!(titles.contains(&"first"));
352 assert!(titles.contains(&"second"));
353 assert_eq!(items[0]["cid"], CID);
354 assert!(items[0]["uri"].as_str().unwrap().starts_with("at://"));
355}
356
357#[tokio::test]
358async fn count_distinct_authors_dedupes_per_author() {
359 let h = Harness::new().await;
360 let subject = at("at://did:plc:abalone");
361 h.add_edge(
362 &nsid("sh.tangled.feed.star"),
363 &subject,
364 &at("at://did:plc:nel/sh.tangled.feed.star/s1"),
365 );
366 h.add_edge(
367 &nsid("sh.tangled.feed.star"),
368 &subject,
369 &at("at://did:plc:nel/sh.tangled.feed.star/s2"),
370 );
371 h.add_edge(
372 &nsid("sh.tangled.feed.star"),
373 &subject,
374 &at("at://did:plc:olaren/sh.tangled.feed.star/s3"),
375 );
376
377 let app = router(h.state.clone());
378 let resp = app
379 .oneshot(list_request(
380 "sh.tangled.feed.countStars",
381 subject.as_ref(),
382 &[],
383 ))
384 .await
385 .unwrap();
386 let (_, body) = json_response(resp).await;
387 assert_eq!(body["count"], json!(3));
388 assert_eq!(body["distinctAuthors"], json!(2));
389}
390
391#[tokio::test]
392async fn count_forks_counts_repos_pointing_at_the_source() {
393 let h = Harness::new().await;
394 let subject = at("at://did:plc:abalone");
395 h.add_edge(
396 &nsid("sh.tangled.repo.source"),
397 &subject,
398 &at("at://did:plc:nel/sh.tangled.repo/f1"),
399 );
400 h.add_edge(
401 &nsid("sh.tangled.repo.source"),
402 &subject,
403 &at("at://did:plc:olaren/sh.tangled.repo/f2"),
404 );
405 // the source owner's other repos are not forks of it
406 h.add_edge(
407 &nsid("sh.tangled.repo"),
408 &subject,
409 &at("at://did:plc:nel/sh.tangled.repo/r1"),
410 );
411
412 let app = router(h.state.clone());
413 let resp = app
414 .oneshot(list_request(
415 "sh.tangled.repo.countForks",
416 subject.as_ref(),
417 &[],
418 ))
419 .await
420 .unwrap();
421 let (status, body) = json_response(resp).await;
422 assert_eq!(status, StatusCode::OK);
423 assert_eq!(body["count"], json!(2));
424 assert_eq!(body["distinctAuthors"], json!(2));
425}
426
427#[tokio::test]
428async fn count_forks_of_an_unforked_repo_is_zero() {
429 let h = Harness::new().await;
430 let app = router(h.state.clone());
431 let resp = app
432 .oneshot(list_request(
433 "sh.tangled.repo.countForks",
434 "did:plc:abalone",
435 &[],
436 ))
437 .await
438 .unwrap();
439 let (status, body) = json_response(resp).await;
440 assert_eq!(status, StatusCode::OK);
441 assert_eq!(body["count"], json!(0));
442}
443
444#[tokio::test]
445async fn count_forks_rejects_an_at_uri_subject() {
446 let h = Harness::new().await;
447 let app = router(h.state.clone());
448 let resp = app
449 .oneshot(list_request(
450 "sh.tangled.repo.countForks",
451 "at://did:plc:nel/sh.tangled.repo/core",
452 &[],
453 ))
454 .await
455 .unwrap();
456 let (status, body) = json_response(resp).await;
457 assert_eq!(status, StatusCode::BAD_REQUEST);
458 assert_eq!(body["error"], "InvalidRequest");
459}
460
461#[tokio::test]
462async fn list_items_stable_across_coverage_promotion() {
463 let h = Harness::new().await;
464 let subject = at("at://did:plc:abalone");
465 let nel = did("did:plc:nel");
466 h.add_edge(
467 &nsid("sh.tangled.feed.star"),
468 &subject,
469 &at(&format!("at://{}/sh.tangled.feed.star/s1", nel.as_ref())),
470 );
471 h.mount(
472 &nel,
473 &nsid("sh.tangled.feed.star"),
474 &rkey("s1"),
475 star_body(&did("did:plc:abalone")),
476 )
477 .await;
478
479 let app = router(h.state.clone());
480 h.warming(1, 5);
481 let (_, before) = json_response(
482 app.clone()
483 .oneshot(list_request(
484 "sh.tangled.feed.listStars",
485 subject.as_ref(),
486 &[],
487 ))
488 .await
489 .unwrap(),
490 )
491 .await;
492 assert_eq!(before["items"].as_array().unwrap().len(), 1);
493
494 h.promote_ready(2, 9);
495 let (_, after) = json_response(
496 app.oneshot(list_request(
497 "sh.tangled.feed.listStars",
498 subject.as_ref(),
499 &[],
500 ))
501 .await
502 .unwrap(),
503 )
504 .await;
505 assert_eq!(
506 after["items"].as_array().unwrap().len(),
507 before["items"].as_array().unwrap().len(),
508 );
509 assert_eq!(after["items"], before["items"]);
510}
511
512#[tokio::test]
513async fn list_paginates_via_cursor() {
514 let h = Harness::new().await;
515 let subject = at("at://did:plc:abalone");
516 let repo = did("did:plc:abalone");
517 let owners = [
518 ("did:plc:nel", "i1"),
519 ("did:plc:olaren", "i2"),
520 ("did:plc:teq", "i3"),
521 ("did:plc:lyna", "i4"),
522 ("did:plc:bailey", "i5"),
523 ];
524 stream::iter(owners)
525 .for_each(|(d, r)| {
526 let h = &h;
527 let subject = subject.clone();
528 let repo = repo.clone();
529 async move {
530 let d_did = did(d);
531 let rk = rkey(r);
532 h.add_edge(
533 &nsid("sh.tangled.repo.issue"),
534 &subject,
535 &at(&format!(
536 "at://{}/sh.tangled.repo.issue/{}",
537 d_did.as_ref(),
538 rk.as_ref()
539 )),
540 );
541 h.mount(
542 &d_did,
543 &nsid("sh.tangled.repo.issue"),
544 &rk,
545 issue_body(&repo, &format!("issue-{}", rk.as_ref())),
546 )
547 .await;
548 }
549 })
550 .await;
551
552 let app = router(h.state.clone());
553 let (_, page1) = json_response(
554 app.clone()
555 .oneshot(list_request(
556 "sh.tangled.repo.listIssues",
557 subject.as_ref(),
558 &[("limit", "2")],
559 ))
560 .await
561 .unwrap(),
562 )
563 .await;
564 let page1_items = page1["items"].as_array().unwrap().clone();
565 assert_eq!(page1_items.len(), 2);
566 let cursor = page1["cursor"]
567 .as_str()
568 .expect("first page must yield a cursor")
569 .to_owned();
570 assert!(
571 PageToken::decode_token(&cursor).is_ok(),
572 "cursor must be a TID-shaped token"
573 );
574
575 let (_, page2) = json_response(
576 app.oneshot(list_request(
577 "sh.tangled.repo.listIssues",
578 subject.as_ref(),
579 &[("limit", "10"), ("cursor", &cursor)],
580 ))
581 .await
582 .unwrap(),
583 )
584 .await;
585 let page2_items = page2["items"].as_array().unwrap().clone();
586 assert_eq!(page2_items.len(), 3);
587 assert!(page2["cursor"].is_null(), "tail page must not promise more");
588
589 let union: Vec<&str> = page1_items
590 .iter()
591 .chain(page2_items.iter())
592 .map(|item| item["uri"].as_str().unwrap())
593 .collect();
594 assert_eq!(union.len(), owners.len(), "union covers every owner");
595 let mut sorted = union.clone();
596 sorted.sort();
597 sorted.dedup();
598 assert_eq!(sorted.len(), owners.len(), "no duplicates across pages");
599}
600
601#[tokio::test]
602async fn pagination_unaffected_by_coverage_promotion() {
603 let h = Harness::new().await;
604 let subject = at("at://did:plc:abalone");
605 let repo = did("did:plc:abalone");
606 let owners = [("did:plc:nel", "i1"), ("did:plc:olaren", "i2")];
607 stream::iter(owners)
608 .for_each(|(d, r)| {
609 let h = &h;
610 let subject = subject.clone();
611 let repo = repo.clone();
612 async move {
613 let d_did = did(d);
614 let rk = rkey(r);
615 h.add_edge(
616 &nsid("sh.tangled.repo.issue"),
617 &subject,
618 &at(&format!(
619 "at://{}/sh.tangled.repo.issue/{}",
620 d_did.as_ref(),
621 rk.as_ref()
622 )),
623 );
624 h.mount(
625 &d_did,
626 &nsid("sh.tangled.repo.issue"),
627 &rk,
628 issue_body(&repo, &format!("issue-{}", rk.as_ref())),
629 )
630 .await;
631 }
632 })
633 .await;
634
635 h.warming(1, 5);
636 let app = router(h.state.clone());
637 let (_, page1) = json_response(
638 app.clone()
639 .oneshot(list_request(
640 "sh.tangled.repo.listIssues",
641 subject.as_ref(),
642 &[("limit", "1")],
643 ))
644 .await
645 .unwrap(),
646 )
647 .await;
648 let cursor = page1["cursor"].as_str().unwrap().to_owned();
649
650 h.promote_ready(2, 9);
651 let (_, page2) = json_response(
652 app.oneshot(list_request(
653 "sh.tangled.repo.listIssues",
654 subject.as_ref(),
655 &[("limit", "10"), ("cursor", &cursor)],
656 ))
657 .await
658 .unwrap(),
659 )
660 .await;
661 assert_eq!(page2["items"].as_array().unwrap().len(), 1);
662}
663
664#[tokio::test]
665async fn invalid_cursor_returns_400() {
666 let h = Harness::new().await;
667 let app = router(h.state.clone());
668 let resp = app
669 .oneshot(list_request(
670 "sh.tangled.repo.listIssues",
671 "at://did:plc:abalone",
672 &[("cursor", "not-a-number")],
673 ))
674 .await
675 .unwrap();
676 let (status, body) = json_response(resp).await;
677 assert_eq!(status, StatusCode::BAD_REQUEST);
678 assert_eq!(body["error"], "InvalidRequest");
679}
680
681#[tokio::test]
682async fn list_follows_subject_is_followee_did() {
683 let h = Harness::new().await;
684 let followee = did("did:plc:bailey");
685 let subject = at(&format!("at://{}", followee.as_ref()));
686 h.add_edge(
687 &nsid("sh.tangled.graph.follow"),
688 &subject,
689 &at("at://did:plc:nel/sh.tangled.graph.follow/f1"),
690 );
691 h.mount(
692 &did("did:plc:nel"),
693 &nsid("sh.tangled.graph.follow"),
694 &rkey("f1"),
695 follow_body(&followee),
696 )
697 .await;
698
699 let app = router(h.state.clone());
700 let (status, body) = json_response(
701 app.oneshot(list_request(
702 "sh.tangled.graph.listFollows",
703 subject.as_ref(),
704 &[],
705 ))
706 .await
707 .unwrap(),
708 )
709 .await;
710 assert_eq!(status, StatusCode::OK);
711 let items = body["items"].as_array().unwrap();
712 assert_eq!(items.len(), 1);
713 assert_eq!(items[0]["value"]["subject"], followee.as_ref());
714}
715
716#[tokio::test]
717async fn get_follow_returns_uri_when_present_and_404_otherwise() {
718 let h = Harness::new().await;
719 let followee = did("did:plc:bailey");
720 let subject = at(&format!("at://{}", followee.as_ref()));
721 h.add_edge(
722 &nsid("sh.tangled.graph.follow"),
723 &subject,
724 &at("at://did:plc:nel/sh.tangled.graph.follow/f1"),
725 );
726
727 let app = router(h.state.clone());
728 let (status, body) = json_response(
729 app.clone()
730 .oneshot(list_request(
731 "sh.tangled.graph.getFollow",
732 followee.as_ref(),
733 &[("actor", "did:plc:nel")],
734 ))
735 .await
736 .unwrap(),
737 )
738 .await;
739 assert_eq!(status, StatusCode::OK);
740 assert_eq!(body["uri"], "at://did:plc:nel/sh.tangled.graph.follow/f1");
741
742 // different actor never followed them, so this is 404 not a zero-ish success
743 let (status, _) = json_response(
744 app.oneshot(list_request(
745 "sh.tangled.graph.getFollow",
746 followee.as_ref(),
747 &[("actor", "did:plc:someoneelse")],
748 ))
749 .await
750 .unwrap(),
751 )
752 .await;
753 assert_eq!(status, StatusCode::NOT_FOUND);
754}
755
756#[tokio::test]
757async fn get_star_returns_uri_when_present_and_404_otherwise() {
758 let h = Harness::new().await;
759 let repo_did = did("did:plc:limpet");
760 let subject = at(&format!("at://{}", repo_did.as_ref()));
761 h.add_edge(
762 &nsid("sh.tangled.feed.star"),
763 &subject,
764 &at("at://did:plc:nel/sh.tangled.feed.star/s1"),
765 );
766
767 let app = router(h.state.clone());
768 let (status, body) = json_response(
769 app.clone()
770 .oneshot(list_request(
771 "sh.tangled.feed.getStar",
772 repo_did.as_ref(),
773 &[("actor", "did:plc:nel")],
774 ))
775 .await
776 .unwrap(),
777 )
778 .await;
779 assert_eq!(status, StatusCode::OK);
780 assert_eq!(body["uri"], "at://did:plc:nel/sh.tangled.feed.star/s1");
781
782 let (status, _) = json_response(
783 app.oneshot(list_request(
784 "sh.tangled.feed.getStar",
785 repo_did.as_ref(),
786 &[("actor", "did:plc:someoneelse")],
787 ))
788 .await
789 .unwrap(),
790 )
791 .await;
792 assert_eq!(status, StatusCode::NOT_FOUND);
793}
794
795#[tokio::test]
796async fn upstream_failure_during_hydration_drops_only_that_item() {
797 let h = Harness::new().await;
798 let subject = at("at://did:plc:squid");
799 let kind = nsid("sh.tangled.repo.issue");
800 let repo = did("did:plc:squid");
801 h.add_edge(
802 &kind,
803 &subject,
804 &at("at://did:plc:nel/sh.tangled.repo.issue/ok"),
805 );
806 h.add_edge(
807 &kind,
808 &subject,
809 &at("at://did:plc:teq/sh.tangled.repo.issue/flaky"),
810 );
811 h.mount(
812 &did("did:plc:nel"),
813 &kind,
814 &rkey("ok"),
815 issue_body(&repo, "kelp survey"),
816 )
817 .await;
818 Mock::given(method("GET"))
819 .and(path("/xrpc/com.atproto.repo.getRecord"))
820 .and(query_param("repo", "did:plc:teq"))
821 .and(query_param("collection", "sh.tangled.repo.issue"))
822 .and(query_param("rkey", "flaky"))
823 .respond_with(ResponseTemplate::new(503))
824 .mount(&h.server)
825 .await;
826
827 let app = router(h.state.clone());
828 let (status, body) = json_response(
829 app.oneshot(list_request(
830 "sh.tangled.repo.listIssues",
831 subject.as_ref(),
832 &[],
833 ))
834 .await
835 .unwrap(),
836 )
837 .await;
838 assert_eq!(status, StatusCode::OK);
839 let items = body["items"].as_array().expect("items array");
840 assert_eq!(items.len(), 1, "flaky item dropped, healthy sibling kept");
841 assert_eq!(
842 items[0]["uri"].as_str().unwrap(),
843 "at://did:plc:nel/sh.tangled.repo.issue/ok",
844 );
845}
846
847#[tokio::test]
848async fn transient_failure_keeps_edge_so_count_stays_whole() {
849 let h = Harness::new().await;
850 let subject = at("at://did:plc:squid");
851 let kind = nsid("sh.tangled.repo.issue");
852 let repo = did("did:plc:squid");
853 h.add_edge(
854 &kind,
855 &subject,
856 &at("at://did:plc:nel/sh.tangled.repo.issue/ok"),
857 );
858 h.add_edge(
859 &kind,
860 &subject,
861 &at("at://did:plc:teq/sh.tangled.repo.issue/flaky"),
862 );
863 h.mount(
864 &did("did:plc:nel"),
865 &kind,
866 &rkey("ok"),
867 issue_body(&repo, "kelp survey"),
868 )
869 .await;
870 Mock::given(method("GET"))
871 .and(path("/xrpc/com.atproto.repo.getRecord"))
872 .and(query_param("repo", "did:plc:teq"))
873 .and(query_param("collection", "sh.tangled.repo.issue"))
874 .and(query_param("rkey", "flaky"))
875 .respond_with(ResponseTemplate::new(503))
876 .mount(&h.server)
877 .await;
878
879 let app = router(h.state.clone());
880 let (status, body) = json_response(
881 app.clone()
882 .oneshot(list_request(
883 "sh.tangled.repo.listIssues",
884 subject.as_ref(),
885 &[],
886 ))
887 .await
888 .unwrap(),
889 )
890 .await;
891 assert_eq!(status, StatusCode::OK);
892 assert_eq!(body["items"].as_array().unwrap().len(), 1);
893
894 let (cstatus, cbody) = json_response(
895 app.oneshot(list_request(
896 "sh.tangled.repo.countIssues",
897 subject.as_ref(),
898 &[],
899 ))
900 .await
901 .unwrap(),
902 )
903 .await;
904 assert_eq!(cstatus, StatusCode::OK);
905 assert_eq!(
906 cbody["count"],
907 json!(2),
908 "a transient 503 must not evict the edge, count stays whole",
909 );
910}
911
912#[tokio::test]
913async fn gone_item_is_evicted_so_count_converges_to_list() {
914 let h = Harness::new().await;
915 let subject = at("at://did:plc:squid");
916 let kind = nsid("sh.tangled.repo.issue");
917 let repo = did("did:plc:squid");
918 h.add_edge(
919 &kind,
920 &subject,
921 &at("at://did:plc:nel/sh.tangled.repo.issue/ok"),
922 );
923 h.add_edge(
924 &kind,
925 &subject,
926 &at("at://did:plc:teq/sh.tangled.repo.issue/gone"),
927 );
928 h.mount(
929 &did("did:plc:nel"),
930 &kind,
931 &rkey("ok"),
932 issue_body(&repo, "kelp survey"),
933 )
934 .await;
935 Mock::given(method("GET"))
936 .and(path("/xrpc/com.atproto.repo.getRecord"))
937 .and(query_param("repo", "did:plc:teq"))
938 .and(query_param("collection", "sh.tangled.repo.issue"))
939 .and(query_param("rkey", "gone"))
940 .respond_with(ResponseTemplate::new(404))
941 .mount(&h.server)
942 .await;
943
944 let app = router(h.state.clone());
945 let (status, body) = json_response(
946 app.clone()
947 .oneshot(list_request(
948 "sh.tangled.repo.listIssues",
949 subject.as_ref(),
950 &[],
951 ))
952 .await
953 .unwrap(),
954 )
955 .await;
956 assert_eq!(status, StatusCode::OK);
957 assert_eq!(
958 body["items"].as_array().unwrap().len(),
959 1,
960 "gone item dropped from the page"
961 );
962
963 let (cstatus, cbody) = json_response(
964 app.oneshot(list_request(
965 "sh.tangled.repo.countIssues",
966 subject.as_ref(),
967 &[],
968 ))
969 .await
970 .unwrap(),
971 )
972 .await;
973 assert_eq!(cstatus, StatusCode::OK);
974 assert_eq!(
975 cbody["count"],
976 json!(1),
977 "a definitive 404 must evict the dead edge so count matches the list",
978 );
979}
980
981#[tokio::test]
982async fn handle_authority_subject_is_400() {
983 let h = Harness::new().await;
984 let app = router(h.state.clone());
985 let cases = [
986 "sh.tangled.feed.listStars",
987 "sh.tangled.feed.countStars",
988 "sh.tangled.graph.listFollows",
989 "sh.tangled.graph.countFollows",
990 "sh.tangled.repo.listIssues",
991 "sh.tangled.repo.countIssues",
992 "sh.tangled.repo.listPulls",
993 "sh.tangled.repo.countPulls",
994 "sh.tangled.feed.listComments",
995 "sh.tangled.feed.countComments",
996 ];
997 stream::iter(cases)
998 .for_each(|endpoint| {
999 let app = app.clone();
1000 async move {
1001 let resp = app
1002 .oneshot(list_request(endpoint, "at://oyster.cafe", &[]))
1003 .await
1004 .unwrap();
1005 let (status, body) = json_response(resp).await;
1006 assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint}");
1007 assert_eq!(body["error"], "InvalidRequest", "{endpoint}");
1008 assert!(
1009 body["message"]
1010 .as_str()
1011 .unwrap_or_default()
1012 .contains("did, not a handle"),
1013 "{endpoint}: {}",
1014 body["message"]
1015 );
1016 }
1017 })
1018 .await;
1019}
1020
1021#[tokio::test]
1022async fn empty_subject_is_400() {
1023 let h = Harness::new().await;
1024 let app = router(h.state.clone());
1025 let resp = app
1026 .oneshot(list_request("sh.tangled.repo.listIssues", "", &[]))
1027 .await
1028 .unwrap();
1029 let (status, body) = json_response(resp).await;
1030 assert_eq!(status, StatusCode::BAD_REQUEST);
1031 assert_eq!(body["error"], "InvalidRequest");
1032}
1033
1034#[tokio::test]
1035async fn limit_below_min_or_above_max_is_400() {
1036 let h = Harness::new().await;
1037 let app = router(h.state.clone());
1038 let cases = [("0", "below"), ("1001", "above")];
1039 stream::iter(cases)
1040 .for_each(|(limit, label)| {
1041 let app = app.clone();
1042 async move {
1043 let resp = app
1044 .oneshot(list_request(
1045 "sh.tangled.repo.listIssues",
1046 "at://did:plc:abalone",
1047 &[("limit", limit)],
1048 ))
1049 .await
1050 .unwrap();
1051 let (status, body) = json_response(resp).await;
1052 assert_eq!(status, StatusCode::BAD_REQUEST, "limit {label}");
1053 assert_eq!(body["error"], "InvalidRequest", "limit {label}");
1054 }
1055 })
1056 .await;
1057}
1058
1059#[tokio::test]
1060async fn count_after_remove_source_returns_zero() {
1061 let h = Harness::new().await;
1062 let subject = at("at://did:plc:abalone");
1063 let source = at("at://did:plc:nel/sh.tangled.feed.star/s1");
1064 h.add_edge(&nsid("sh.tangled.feed.star"), &subject, &source);
1065 h.edges.remove_source(&source);
1066
1067 let app = router(h.state.clone());
1068 let (_, body) = json_response(
1069 app.oneshot(list_request(
1070 "sh.tangled.feed.countStars",
1071 subject.as_ref(),
1072 &[],
1073 ))
1074 .await
1075 .unwrap(),
1076 )
1077 .await;
1078 assert_eq!(body["count"], json!(0));
1079 assert_eq!(body["distinctAuthors"], json!(0));
1080}
1081
1082#[tokio::test]
1083async fn list_feed_comments_hydrates_end_to_end() {
1084 let h = Harness::new().await;
1085 let issue_uri = at("at://did:plc:abalone/sh.tangled.repo.issue/i1");
1086 let nel = did("did:plc:nel");
1087 let rk = rkey("c1");
1088 h.add_edge(
1089 &nsid("sh.tangled.feed.comment"),
1090 &issue_uri,
1091 &at(&format!(
1092 "at://{}/sh.tangled.feed.comment/{}",
1093 nel.as_ref(),
1094 rk.as_ref()
1095 )),
1096 );
1097 h.mount(
1098 &nel,
1099 &nsid("sh.tangled.feed.comment"),
1100 &rk,
1101 json!({
1102 "$type": "sh.tangled.feed.comment",
1103 "subject": { "uri": issue_uri.as_ref(), "cid": "bafkqaaa" },
1104 "body": { "$type": "sh.tangled.markup.markdown", "text": "thoughts" },
1105 "createdAt": "2026-05-01T00:00:00Z"
1106 }),
1107 )
1108 .await;
1109
1110 let app = router(h.state.clone());
1111 let (status, body) = json_response(
1112 app.oneshot(list_request(
1113 "sh.tangled.feed.listComments",
1114 issue_uri.as_ref(),
1115 &[],
1116 ))
1117 .await
1118 .unwrap(),
1119 )
1120 .await;
1121 assert_eq!(status, StatusCode::OK);
1122 let items = body["items"].as_array().unwrap();
1123 assert_eq!(items.len(), 1);
1124 assert_eq!(items[0]["value"]["body"]["text"], json!("thoughts"));
1125 assert_eq!(
1126 items[0]["value"]["subject"]["uri"],
1127 json!(issue_uri.as_ref())
1128 );
1129}
1130
1131#[tokio::test]
1132async fn list_item_cid_is_present() {
1133 let h = Harness::new().await;
1134 let subject = at("at://did:plc:abalone");
1135 let nel = did("did:plc:nel");
1136 h.add_edge(
1137 &nsid("sh.tangled.feed.star"),
1138 &subject,
1139 &at(&format!("at://{}/sh.tangled.feed.star/s1", nel.as_ref())),
1140 );
1141 h.mount(
1142 &nel,
1143 &nsid("sh.tangled.feed.star"),
1144 &rkey("s1"),
1145 star_body(&did("did:plc:abalone")),
1146 )
1147 .await;
1148
1149 let app = router(h.state.clone());
1150 let (_, body) = json_response(
1151 app.oneshot(list_request(
1152 "sh.tangled.feed.listStars",
1153 subject.as_ref(),
1154 &[],
1155 ))
1156 .await
1157 .unwrap(),
1158 )
1159 .await;
1160 let item = &body["items"][0];
1161 assert!(
1162 item.as_object().unwrap().contains_key("cid"),
1163 "list items must mirror getRecord output shape and include cid"
1164 );
1165 assert_eq!(item["cid"], json!(CID));
1166}
1167
1168#[tokio::test]
1169async fn count_feed_comments_subjects_on_issue_uri() {
1170 let h = Harness::new().await;
1171 let issue_uri = at("at://did:plc:abalone/sh.tangled.repo.issue/i1");
1172 h.add_edge(
1173 &nsid("sh.tangled.feed.comment"),
1174 &issue_uri,
1175 &at("at://did:plc:nel/sh.tangled.feed.comment/c1"),
1176 );
1177 h.add_edge(
1178 &nsid("sh.tangled.feed.comment"),
1179 &issue_uri,
1180 &at("at://did:plc:olaren/sh.tangled.feed.comment/c2"),
1181 );
1182
1183 let app = router(h.state.clone());
1184 let (status, body) = json_response(
1185 app.oneshot(list_request(
1186 "sh.tangled.feed.countComments",
1187 issue_uri.as_ref(),
1188 &[],
1189 ))
1190 .await
1191 .unwrap(),
1192 )
1193 .await;
1194 assert_eq!(status, StatusCode::OK);
1195 assert_eq!(body["count"], json!(2));
1196 assert_eq!(body["distinctAuthors"], json!(2));
1197}
1198
1199#[tokio::test]
1200async fn list_item_404_dropped_not_404_for_subject() {
1201 let h = Harness::new().await;
1202 let subject = at("at://did:plc:squid");
1203 let kind = nsid("sh.tangled.repo.issue");
1204 let repo = did("did:plc:squid");
1205 h.add_edge(
1206 &kind,
1207 &subject,
1208 &at("at://did:plc:nel/sh.tangled.repo.issue/live"),
1209 );
1210 h.add_edge(
1211 &kind,
1212 &subject,
1213 &at("at://did:plc:teq/sh.tangled.repo.issue/missing"),
1214 );
1215 h.mount(
1216 &did("did:plc:nel"),
1217 &kind,
1218 &rkey("live"),
1219 issue_body(&repo, "kelp survives"),
1220 )
1221 .await;
1222 Mock::given(method("GET"))
1223 .and(path("/xrpc/com.atproto.repo.getRecord"))
1224 .and(query_param("repo", "did:plc:teq"))
1225 .and(query_param("collection", "sh.tangled.repo.issue"))
1226 .and(query_param("rkey", "missing"))
1227 .respond_with(ResponseTemplate::new(404).set_body_json(json!({
1228 "error": "RecordNotFound",
1229 "message": "could not find record"
1230 })))
1231 .mount(&h.server)
1232 .await;
1233
1234 let app = router(h.state.clone());
1235 let (status, body) = json_response(
1236 app.oneshot(list_request(
1237 "sh.tangled.repo.listIssues",
1238 subject.as_ref(),
1239 &[],
1240 ))
1241 .await
1242 .unwrap(),
1243 )
1244 .await;
1245 assert_eq!(
1246 status,
1247 StatusCode::OK,
1248 "a stale-index 404 drops that item, it must not 404 or 502 the subject's list",
1249 );
1250 let items = body["items"].as_array().expect("items array");
1251 assert_eq!(items.len(), 1, "stale 404 item dropped, live sibling kept");
1252 assert_eq!(
1253 items[0]["uri"].as_str().unwrap(),
1254 "at://did:plc:nel/sh.tangled.repo.issue/live",
1255 );
1256}
1257
1258#[tokio::test]
1259async fn list_item_with_wrong_type_tag_dropped() {
1260 let h = Harness::new().await;
1261 let subject = at("at://did:plc:squid");
1262 let kind = nsid("sh.tangled.feed.star");
1263 h.add_edge(
1264 &kind,
1265 &subject,
1266 &at("at://did:plc:nel/sh.tangled.feed.star/good"),
1267 );
1268 h.add_edge(
1269 &kind,
1270 &subject,
1271 &at("at://did:plc:teq/sh.tangled.feed.star/wrong"),
1272 );
1273 h.mount(
1274 &did("did:plc:nel"),
1275 &kind,
1276 &rkey("good"),
1277 star_body(&did("did:plc:squid")),
1278 )
1279 .await;
1280 Mock::given(method("GET"))
1281 .and(path("/xrpc/com.atproto.repo.getRecord"))
1282 .and(query_param("repo", "did:plc:teq"))
1283 .and(query_param("collection", "sh.tangled.feed.star"))
1284 .and(query_param("rkey", "wrong"))
1285 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
1286 "uri": "at://did:plc:teq/sh.tangled.feed.star/wrong",
1287 "cid": CID,
1288 "value": {
1289 "$type": "sh.tangled.feed.reaction",
1290 "createdAt": "2026-05-01T00:00:00Z",
1291 "subject": "at://did:plc:squid"
1292 }
1293 })))
1294 .mount(&h.server)
1295 .await;
1296
1297 let app = router(h.state.clone());
1298 let (status, body) = json_response(
1299 app.oneshot(list_request(
1300 "sh.tangled.feed.listStars",
1301 subject.as_ref(),
1302 &[],
1303 ))
1304 .await
1305 .unwrap(),
1306 )
1307 .await;
1308 assert_eq!(status, StatusCode::OK);
1309 let items = body["items"].as_array().expect("items array");
1310 assert_eq!(items.len(), 1, "wrong-type item dropped, valid star kept");
1311 assert_eq!(
1312 items[0]["uri"].as_str().unwrap(),
1313 "at://did:plc:nel/sh.tangled.feed.star/good",
1314 );
1315}
1316
1317#[tokio::test]
1318async fn list_item_with_mismatched_collection_dropped() {
1319 let h = Harness::new().await;
1320 let subject = at("at://did:plc:squid");
1321 let kind = nsid("sh.tangled.repo.issue");
1322 let repo = did("did:plc:squid");
1323 h.add_edge(
1324 &kind,
1325 &subject,
1326 &at("at://did:plc:nel/sh.tangled.repo.issue/live"),
1327 );
1328 h.add_edge(
1329 &kind,
1330 &subject,
1331 &at("at://did:plc:teq/sh.tangled.feed.star/whelk"),
1332 );
1333 h.mount(
1334 &did("did:plc:nel"),
1335 &kind,
1336 &rkey("live"),
1337 issue_body(&repo, "kelp survives"),
1338 )
1339 .await;
1340
1341 let app = router(h.state.clone());
1342 let (status, body) = json_response(
1343 app.oneshot(list_request(
1344 "sh.tangled.repo.listIssues",
1345 subject.as_ref(),
1346 &[],
1347 ))
1348 .await
1349 .unwrap(),
1350 )
1351 .await;
1352 assert_eq!(
1353 status,
1354 StatusCode::OK,
1355 "a mismatched-collection index edge must not 400 the subject's list",
1356 );
1357 let items = body["items"].as_array().expect("items array");
1358 assert_eq!(
1359 items.len(),
1360 1,
1361 "mismatched-collection edge dropped, live sibling kept"
1362 );
1363 assert_eq!(
1364 items[0]["uri"].as_str().unwrap(),
1365 "at://did:plc:nel/sh.tangled.repo.issue/live",
1366 );
1367}
1368
1369#[tokio::test]
1370async fn bare_did_endpoints_reject_at_uri_subject() {
1371 let h = Harness::new().await;
1372 let app = router(h.state.clone());
1373 let cases = [
1374 "sh.tangled.graph.listFollows",
1375 "sh.tangled.graph.countFollows",
1376 ];
1377 stream::iter(cases)
1378 .for_each(|endpoint| {
1379 let app = app.clone();
1380 async move {
1381 let resp = app
1382 .oneshot(list_request(
1383 endpoint,
1384 "at://did:plc:abalone/sh.tangled.repo/r1",
1385 &[],
1386 ))
1387 .await
1388 .unwrap();
1389 let (status, body) = json_response(resp).await;
1390 assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint}");
1391 assert_eq!(body["error"], "InvalidRequest", "{endpoint}");
1392 assert!(
1393 body["message"]
1394 .as_str()
1395 .unwrap_or_default()
1396 .contains("bare did"),
1397 "{endpoint}: {}",
1398 body["message"],
1399 );
1400 }
1401 })
1402 .await;
1403}
1404
1405#[tokio::test]
1406async fn repo_pointing_endpoints_reject_at_uri_subject() {
1407 let h = Harness::new().await;
1408 let app = router(h.state.clone());
1409 let cases = [
1410 "sh.tangled.repo.listIssues",
1411 "sh.tangled.repo.countIssues",
1412 "sh.tangled.repo.listPulls",
1413 "sh.tangled.repo.countPulls",
1414 "sh.tangled.repo.listArtifacts",
1415 "sh.tangled.repo.countArtifacts",
1416 ];
1417 stream::iter(cases)
1418 .for_each(|endpoint| {
1419 let app = app.clone();
1420 async move {
1421 let resp = app
1422 .oneshot(list_request(
1423 endpoint,
1424 "at://did:plc:abalone/sh.tangled.repo/r1",
1425 &[],
1426 ))
1427 .await
1428 .unwrap();
1429 let (status, body) = json_response(resp).await;
1430 assert_eq!(
1431 status,
1432 StatusCode::BAD_REQUEST,
1433 "{endpoint} must reject rkey-form subjects since rkeys are unstable; clients must send the repoDID",
1434 );
1435 assert!(
1436 body["message"]
1437 .as_str()
1438 .unwrap_or_default()
1439 .contains("bare did"),
1440 "{endpoint}: {}",
1441 body["message"],
1442 );
1443 }
1444 })
1445 .await;
1446}
1447
1448#[tokio::test]
1449async fn repo_pointing_endpoints_accept_bare_did() {
1450 let h = Harness::new().await;
1451 let app = router(h.state.clone());
1452 let cases = [
1453 "sh.tangled.repo.listIssues",
1454 "sh.tangled.repo.countIssues",
1455 "sh.tangled.repo.listPulls",
1456 "sh.tangled.repo.countPulls",
1457 "sh.tangled.repo.listArtifacts",
1458 "sh.tangled.repo.countArtifacts",
1459 ];
1460 stream::iter(cases)
1461 .for_each(|endpoint| {
1462 let app = app.clone();
1463 async move {
1464 let resp = app
1465 .oneshot(list_request(endpoint, "did:plc:abalone", &[]))
1466 .await
1467 .unwrap();
1468 let (status, _body) = json_response(resp).await;
1469 assert_eq!(status, StatusCode::OK, "{endpoint} must accept bare did");
1470 }
1471 })
1472 .await;
1473}
1474
1475#[tokio::test]
1476async fn feed_comment_endpoints_reject_bare_did_or_wrong_collection() {
1477 let h = Harness::new().await;
1478 let app = router(h.state.clone());
1479 let endpoints = [
1480 "sh.tangled.feed.listComments",
1481 "sh.tangled.feed.countComments",
1482 ];
1483 let inputs = [
1484 "at://did:plc:abalone",
1485 "at://did:plc:abalone/sh.tangled.repo/r1",
1486 ];
1487 let cases = endpoints
1488 .iter()
1489 .copied()
1490 .flat_map(|endpoint| inputs.iter().copied().map(move |input| (endpoint, input)));
1491 stream::iter(cases)
1492 .for_each(|(endpoint, input)| {
1493 let app = app.clone();
1494 async move {
1495 let resp = app
1496 .oneshot(list_request(endpoint, input, &[]))
1497 .await
1498 .unwrap();
1499 let (status, body) = json_response(resp).await;
1500 assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint} input={input}");
1501 let msg = body["message"].as_str().unwrap_or_default();
1502 assert!(
1503 msg.contains("sh.tangled.repo.issue")
1504 && msg.contains("sh.tangled.repo.pull")
1505 && msg.contains("sh.tangled.string"),
1506 "{endpoint} input={input}: {msg}",
1507 );
1508 }
1509 })
1510 .await;
1511}
1512
1513#[tokio::test]
1514async fn star_endpoints_reject_unrelated_collection() {
1515 let h = Harness::new().await;
1516 let app = router(h.state.clone());
1517 let endpoints = ["sh.tangled.feed.listStars", "sh.tangled.feed.countStars"];
1518 stream::iter(endpoints)
1519 .for_each(|endpoint| {
1520 let app = app.clone();
1521 async move {
1522 let resp = app
1523 .oneshot(list_request(
1524 endpoint,
1525 "at://did:plc:abalone/sh.tangled.knot/k1",
1526 &[],
1527 ))
1528 .await
1529 .unwrap();
1530 let (status, body) = json_response(resp).await;
1531 assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint}");
1532 let msg = body["message"].as_str().unwrap_or_default();
1533 assert!(msg.contains("sh.tangled.string"), "{endpoint}: {msg}",);
1534 }
1535 })
1536 .await;
1537}
1538
1539#[tokio::test]
1540async fn star_endpoints_reject_repo_uri_subject() {
1541 let h = Harness::new().await;
1542 let app = router(h.state.clone());
1543 let resp = app
1544 .oneshot(list_request(
1545 "sh.tangled.feed.countStars",
1546 "at://did:plc:abalone/sh.tangled.repo/r1",
1547 &[],
1548 ))
1549 .await
1550 .unwrap();
1551 let (status, body) = json_response(resp).await;
1552 assert_eq!(
1553 status,
1554 StatusCode::BAD_REQUEST,
1555 "rkey-form repo URI must be rejected; clients must send the repoDID directly",
1556 );
1557 let msg = body["message"].as_str().unwrap_or_default();
1558 assert!(msg.contains("sh.tangled.string"), "{msg}");
1559}
1560
1561#[tokio::test]
1562async fn star_endpoints_accept_string_subject_form() {
1563 let h = Harness::new().await;
1564 let app = router(h.state.clone());
1565 let resp = app
1566 .oneshot(list_request(
1567 "sh.tangled.feed.countStars",
1568 "at://did:plc:abalone/sh.tangled.string/k1",
1569 &[],
1570 ))
1571 .await
1572 .unwrap();
1573 let (status, body) = json_response(resp).await;
1574 assert_eq!(status, StatusCode::OK);
1575 assert_eq!(body["count"], json!(0));
1576}
1577
1578#[tokio::test]
1579async fn list_after_remove_source_returns_empty_items() {
1580 let h = Harness::new().await;
1581 let subject = at("at://did:plc:abalone");
1582 let source = at("at://did:plc:nel/sh.tangled.feed.star/s1");
1583 h.add_edge(&nsid("sh.tangled.feed.star"), &subject, &source);
1584 h.edges.remove_source(&source);
1585
1586 let app = router(h.state.clone());
1587 let (status, body) = json_response(
1588 app.oneshot(list_request(
1589 "sh.tangled.feed.listStars",
1590 subject.as_ref(),
1591 &[],
1592 ))
1593 .await
1594 .unwrap(),
1595 )
1596 .await;
1597 assert_eq!(status, StatusCode::OK);
1598 assert_eq!(body["items"], json!([]));
1599 assert!(body["cursor"].is_null());
1600}
1601
1602#[tokio::test]
1603async fn list_pulls_hydrates_via_slingshot_when_edges_present() {
1604 let h = Harness::new().await;
1605 let target_did = did("did:plc:abalone");
1606 let subject = at(&format!("at://{}", target_did.as_ref()));
1607 let source_did = did("did:plc:nel");
1608 let rk = rkey("p1");
1609 h.add_edge(
1610 &nsid("sh.tangled.repo.pull"),
1611 &subject,
1612 &at(&format!(
1613 "at://{}/sh.tangled.repo.pull/{}",
1614 source_did.as_ref(),
1615 rk.as_ref()
1616 )),
1617 );
1618 h.mount(
1619 &source_did,
1620 &nsid("sh.tangled.repo.pull"),
1621 &rk,
1622 json!({
1623 "$type": "sh.tangled.repo.pull",
1624 "title": "ship it",
1625 "createdAt": "2026-05-01T00:00:00Z",
1626 "rounds": [],
1627 "target": {"repo": target_did.as_ref(), "branch": "main"},
1628 }),
1629 )
1630 .await;
1631 let app = router(h.state.clone());
1632 let (status, body) = json_response(
1633 app.oneshot(list_request(
1634 "sh.tangled.repo.listPulls",
1635 subject.as_ref(),
1636 &[],
1637 ))
1638 .await
1639 .unwrap(),
1640 )
1641 .await;
1642 assert_eq!(status, StatusCode::OK);
1643 let items = body["items"].as_array().unwrap();
1644 assert_eq!(items.len(), 1);
1645 assert_eq!(items[0]["value"]["title"], json!("ship it"));
1646 assert_eq!(
1647 items[0]["value"]["target"]["repo"],
1648 json!(target_did.as_ref())
1649 );
1650}
1651
1652#[tokio::test]
1653async fn count_pulls_returns_distinct_authors() {
1654 let h = Harness::new().await;
1655 let subject = at("at://did:plc:abalone");
1656 h.add_edge(
1657 &nsid("sh.tangled.repo.pull"),
1658 &subject,
1659 &at("at://did:plc:nel/sh.tangled.repo.pull/p1"),
1660 );
1661 h.add_edge(
1662 &nsid("sh.tangled.repo.pull"),
1663 &subject,
1664 &at("at://did:plc:olaren/sh.tangled.repo.pull/p2"),
1665 );
1666 h.add_edge(
1667 &nsid("sh.tangled.repo.pull"),
1668 &subject,
1669 &at("at://did:plc:nel/sh.tangled.repo.pull/p3"),
1670 );
1671 let app = router(h.state.clone());
1672 let (_, body) = json_response(
1673 app.oneshot(list_request(
1674 "sh.tangled.repo.countPulls",
1675 subject.as_ref(),
1676 &[],
1677 ))
1678 .await
1679 .unwrap(),
1680 )
1681 .await;
1682 assert_eq!(body["count"], json!(3));
1683 assert_eq!(body["distinctAuthors"], json!(2));
1684}
1685
1686#[tokio::test]
1687async fn extractor_to_xrpc_round_trip_for_star() {
1688 let h = Harness::new().await;
1689 let subject_did = did("did:plc:abalone");
1690 let source_did = did("did:plc:nel");
1691 let rk = rkey("s1");
1692 let source = at(&format!(
1693 "at://{}/sh.tangled.feed.star/{}",
1694 source_did.as_ref(),
1695 rk.as_ref()
1696 ));
1697 let body = star_body(&subject_did);
1698 let parsed =
1699 bobbin_types::edges::Record::from_json_value(&nsid("sh.tangled.feed.star"), body.clone())
1700 .expect("parse star record");
1701 parsed
1702 .extract_edges(&source)
1703 .expect("extract")
1704 .into_iter()
1705 .for_each(|e| h.edges.add(e));
1706 h.mount(&source_did, &nsid("sh.tangled.feed.star"), &rk, body)
1707 .await;
1708
1709 let app = router(h.state.clone());
1710 let (status, json) = json_response(
1711 app.oneshot(list_request(
1712 "sh.tangled.feed.listStars",
1713 &format!("at://{}", subject_did.as_ref()),
1714 &[],
1715 ))
1716 .await
1717 .unwrap(),
1718 )
1719 .await;
1720 assert_eq!(
1721 status,
1722 StatusCode::OK,
1723 "extractor key must match handler subject, body was {json}",
1724 );
1725 let items = json["items"].as_array().unwrap();
1726 assert_eq!(items.len(), 1, "expected exactly one star edge");
1727 assert_eq!(
1728 items[0]["value"]["subject"]["did"],
1729 json!(subject_did.as_ref())
1730 );
1731}
1732
1733#[tokio::test]
1734async fn list_issues_includes_state_comment_count_and_state_updated_at() {
1735 let h = Harness::new().await;
1736 let repo = did("did:plc:limpet");
1737 let subject = at(&format!("at://{}", repo.as_ref()));
1738 let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1");
1739 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri);
1740 h.mount(
1741 &did("did:plc:nel"),
1742 &nsid("sh.tangled.repo.issue"),
1743 &rkey("i1"),
1744 issue_body(&repo, "hi"),
1745 )
1746 .await;
1747 h.add_edge(
1748 &nsid("sh.tangled.feed.comment"),
1749 &issue_uri,
1750 &at("at://did:plc:olaren/sh.tangled.feed.comment/c1"),
1751 );
1752 h.add_edge(
1753 &nsid("sh.tangled.feed.comment"),
1754 &issue_uri,
1755 &at("at://did:plc:teq/sh.tangled.feed.comment/c2"),
1756 );
1757
1758 h.upsert_issue_state(
1759 at("at://did:plc:nel/sh.tangled.repo.issue.state/s1"),
1760 issue_uri.clone(),
1761 1_777_593_600_000_000,
1762 IssueStateKind::Open,
1763 );
1764 h.upsert_issue_state(
1765 at("at://did:plc:nel/sh.tangled.repo.issue.state/s2"),
1766 issue_uri.clone(),
1767 1_777_593_700_000_000,
1768 IssueStateKind::Closed,
1769 );
1770
1771 let app = router(h.state.clone());
1772 let resp = app
1773 .oneshot(list_request(
1774 "sh.tangled.repo.listIssues",
1775 subject.as_ref(),
1776 &[],
1777 ))
1778 .await
1779 .unwrap();
1780 let (status, body) = json_response(resp).await;
1781 assert_eq!(status, StatusCode::OK);
1782 let item = &body["items"][0];
1783 assert_eq!(item["state"], json!("closed"));
1784 assert_eq!(item["commentCount"], json!(2));
1785 let updated = item["stateUpdatedAt"]
1786 .as_str()
1787 .expect("stateUpdatedAt must serialize as RFC3339 string");
1788 assert!(
1789 updated.starts_with("2026-"),
1790 "expected 2026 timestamp, got {updated}"
1791 );
1792}
1793
1794#[tokio::test]
1795async fn list_issues_defaults_to_open_when_no_state_record() {
1796 let h = Harness::new().await;
1797 let repo = did("did:plc:limpet");
1798 let subject = at(&format!("at://{}", repo.as_ref()));
1799 let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1");
1800 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri);
1801 h.mount(
1802 &did("did:plc:nel"),
1803 &nsid("sh.tangled.repo.issue"),
1804 &rkey("i1"),
1805 issue_body(&repo, "no state yet"),
1806 )
1807 .await;
1808
1809 let app = router(h.state.clone());
1810 let (_status, body) = json_response(
1811 app.oneshot(list_request(
1812 "sh.tangled.repo.listIssues",
1813 subject.as_ref(),
1814 &[],
1815 ))
1816 .await
1817 .unwrap(),
1818 )
1819 .await;
1820 let item = &body["items"][0];
1821 assert_eq!(
1822 item["state"],
1823 json!("open"),
1824 "absent state record defaults to open"
1825 );
1826 assert!(
1827 item.get("stateUpdatedAt").is_none(),
1828 "stateUpdatedAt must be absent without a state record",
1829 );
1830 assert_eq!(item["commentCount"], json!(0));
1831}
1832
1833#[tokio::test]
1834async fn list_issues_author_filter_restricts_to_matching_did() {
1835 let h = Harness::new().await;
1836 let repo = did("did:plc:limpet");
1837 let subject = at(&format!("at://{}", repo.as_ref()));
1838 let owners = [
1839 ("did:plc:nel", "n1"),
1840 ("did:plc:nel", "n2"),
1841 ("did:plc:olaren", "o1"),
1842 ("did:plc:olaren", "o2"),
1843 ];
1844 stream::iter(owners)
1845 .for_each(|(d, r)| {
1846 let h = &h;
1847 let subject = subject.clone();
1848 let repo = repo.clone();
1849 async move {
1850 let d_did = did(d);
1851 let rk = rkey(r);
1852 h.add_edge(
1853 &nsid("sh.tangled.repo.issue"),
1854 &subject,
1855 &at(&format!(
1856 "at://{}/sh.tangled.repo.issue/{}",
1857 d_did.as_ref(),
1858 rk.as_ref()
1859 )),
1860 );
1861 h.mount(
1862 &d_did,
1863 &nsid("sh.tangled.repo.issue"),
1864 &rk,
1865 issue_body(&repo, &format!("issue-{}", rk.as_ref())),
1866 )
1867 .await;
1868 }
1869 })
1870 .await;
1871
1872 let app = router(h.state.clone());
1873 let (status, body) = json_response(
1874 app.oneshot(list_request(
1875 "sh.tangled.repo.listIssues",
1876 subject.as_ref(),
1877 &[("author", "did:plc:nel")],
1878 ))
1879 .await
1880 .unwrap(),
1881 )
1882 .await;
1883 assert_eq!(status, StatusCode::OK);
1884 let items = body["items"].as_array().expect("items array");
1885 assert_eq!(items.len(), 2, "two issues authored by nel");
1886 let all_nel = items
1887 .iter()
1888 .all(|i| i["uri"].as_str().unwrap().starts_with("at://did:plc:nel/"));
1889 assert!(all_nel, "every returned uri must be authored by nel");
1890}
1891
1892#[tokio::test]
1893async fn list_issues_invalid_author_returns_400() {
1894 let h = Harness::new().await;
1895 let subject = "at://did:plc:limpet".to_owned();
1896 let app = router(h.state.clone());
1897 let resp = app
1898 .oneshot(list_request(
1899 "sh.tangled.repo.listIssues",
1900 &subject,
1901 &[("author", "not-a-did")],
1902 ))
1903 .await
1904 .unwrap();
1905 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
1906}
1907
1908#[tokio::test]
1909async fn list_pulls_includes_merged_state_and_comment_count() {
1910 let h = Harness::new().await;
1911 let repo = did("did:plc:limpet");
1912 let subject = at(&format!("at://{}", repo.as_ref()));
1913 let pull_uri = at("at://did:plc:nel/sh.tangled.repo.pull/p1");
1914 h.add_edge(&nsid("sh.tangled.repo.pull"), &subject, &pull_uri);
1915 h.mount(
1916 &did("did:plc:nel"),
1917 &nsid("sh.tangled.repo.pull"),
1918 &rkey("p1"),
1919 pull_body(&repo, "fix bug"),
1920 )
1921 .await;
1922 h.add_edge(
1923 &nsid("sh.tangled.feed.comment"),
1924 &pull_uri,
1925 &at("at://did:plc:teq/sh.tangled.feed.comment/c1"),
1926 );
1927 h.upsert_pull_status(
1928 at("at://did:plc:nel/sh.tangled.repo.pull.status/s1"),
1929 pull_uri.clone(),
1930 1_777_593_600_000_000,
1931 PullStatusKind::Open,
1932 );
1933 h.upsert_pull_status(
1934 at("at://did:plc:nel/sh.tangled.repo.pull.status/s2"),
1935 pull_uri.clone(),
1936 1_777_593_800_000_000,
1937 PullStatusKind::Merged,
1938 );
1939
1940 let app = router(h.state.clone());
1941 let (status, body) = json_response(
1942 app.oneshot(list_request(
1943 "sh.tangled.repo.listPulls",
1944 subject.as_ref(),
1945 &[],
1946 ))
1947 .await
1948 .unwrap(),
1949 )
1950 .await;
1951 assert_eq!(status, StatusCode::OK);
1952 let item = &body["items"][0];
1953 assert_eq!(item["state"], json!("merged"));
1954 assert_eq!(item["commentCount"], json!(1));
1955}
1956
1957#[tokio::test]
1958async fn list_issues_state_filter_open_includes_records_without_state() {
1959 let h = Harness::new().await;
1960 let repo = did("did:plc:limpet");
1961 let subject = at(&format!("at://{}", repo.as_ref()));
1962 let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1");
1963 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri);
1964 h.mount(
1965 &did("did:plc:nel"),
1966 &nsid("sh.tangled.repo.issue"),
1967 &rkey("i1"),
1968 issue_body(&repo, "fresh"),
1969 )
1970 .await;
1971
1972 let app = router(h.state.clone());
1973 let (status, body) = json_response(
1974 app.oneshot(list_request(
1975 "sh.tangled.repo.listIssues",
1976 subject.as_ref(),
1977 &[("state", "open")],
1978 ))
1979 .await
1980 .unwrap(),
1981 )
1982 .await;
1983 assert_eq!(status, StatusCode::OK);
1984 let items = body["items"].as_array().expect("items array");
1985 assert_eq!(
1986 items.len(),
1987 1,
1988 "absent state record still matches state=open"
1989 );
1990}
1991
1992#[tokio::test]
1993async fn count_issues_splits_open_from_total() {
1994 let h = Harness::new().await;
1995 let repo = did("did:plc:limpet");
1996 let subject = at(&format!("at://{}", repo.as_ref()));
1997 for rk in ["i1", "i2", "i3"] {
1998 h.add_edge(
1999 &nsid("sh.tangled.repo.issue"),
2000 &subject,
2001 &at(&format!("at://did:plc:nel/sh.tangled.repo.issue/{rk}")),
2002 );
2003 }
2004 // the repo owner closing it counts. with no record at all it is open
2005 h.upsert_issue_state(
2006 at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"),
2007 at("at://did:plc:nel/sh.tangled.repo.issue/i1"),
2008 1_777_593_600_000_000,
2009 IssueStateKind::Closed,
2010 );
2011
2012 let app = router(h.state.clone());
2013 let counts = |args: &'static [(&'static str, &'static str)]| {
2014 let app = app.clone();
2015 let subject = subject.clone();
2016 async move {
2017 let (status, body) = json_response(
2018 app.oneshot(list_request(
2019 "sh.tangled.repo.countIssues",
2020 subject.as_ref(),
2021 args,
2022 ))
2023 .await
2024 .unwrap(),
2025 )
2026 .await;
2027 assert_eq!(status, StatusCode::OK);
2028 body["count"].as_u64().expect("count")
2029 }
2030 };
2031
2032 assert_eq!(counts(&[]).await, 3, "no filter is every issue");
2033 assert_eq!(counts(&[("state", "open")]).await, 2);
2034 assert_eq!(counts(&[("state", "closed")]).await, 1);
2035}
2036
2037#[tokio::test]
2038async fn count_issues_open_ignores_third_party_state_source() {
2039 let h = Harness::new().await;
2040 let repo = did("did:plc:limpet");
2041 let subject = at(&format!("at://{}", repo.as_ref()));
2042 let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1");
2043 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri);
2044 h.upsert_issue_state(
2045 at("at://did:plc:nautilus/sh.tangled.repo.issue.state/spoof"),
2046 issue_uri,
2047 1_777_593_800_000_000,
2048 IssueStateKind::Closed,
2049 );
2050
2051 let app = router(h.state.clone());
2052 let (status, body) = json_response(
2053 app.oneshot(list_request(
2054 "sh.tangled.repo.countIssues",
2055 subject.as_ref(),
2056 &[("state", "open")],
2057 ))
2058 .await
2059 .unwrap(),
2060 )
2061 .await;
2062 assert_eq!(status, StatusCode::OK);
2063 assert_eq!(
2064 body["count"],
2065 json!(1),
2066 "a stranger closing an issue must not take it out of the open count",
2067 );
2068}
2069
2070#[tokio::test]
2071async fn filtered_count_follows_state_changes_and_repo_rekeying() {
2072 let h = Harness::new().await;
2073 let old_repo = did("did:plc:limpet");
2074 let new_repo = did("did:plc:scallop");
2075 let old_subject = at(&format!("at://{}", old_repo.as_ref()));
2076 let new_subject = at(&format!("at://{}", new_repo.as_ref()));
2077 let issue = at("at://did:plc:nel/sh.tangled.repo.issue/i1");
2078 let kind = nsid("sh.tangled.repo.issue");
2079 h.edges.upsert_source(
2080 &issue,
2081 vec![Edge {
2082 kind: kind.clone(),
2083 subject: SubjectRef::Did(old_repo),
2084 source: issue.clone(),
2085 sort_micros: next_sort_micros(),
2086 }],
2087 );
2088 h.edges.refresh_issue_counts(&h.state.issue_states, &issue);
2089
2090 let app = router(h.state.clone());
2091 let count_open = |subject: AtUri<DefaultStr>| {
2092 let app = app.clone();
2093 async move {
2094 let (status, body) = json_response(
2095 app.oneshot(list_request(
2096 "sh.tangled.repo.countIssues",
2097 subject.as_ref(),
2098 &[("state", "open")],
2099 ))
2100 .await
2101 .unwrap(),
2102 )
2103 .await;
2104 assert_eq!(status, StatusCode::OK);
2105 body["count"].as_u64().expect("count")
2106 }
2107 };
2108
2109 assert_eq!(count_open(old_subject.clone()).await, 1);
2110 assert_eq!(count_open(old_subject.clone()).await, 1, "repeated read");
2111 assert_eq!(
2112 count_open(new_subject.clone()).await,
2113 0,
2114 "the issue does not belong to the destination yet",
2115 );
2116
2117 h.upsert_issue_state(
2118 at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"),
2119 issue.clone(),
2120 1_777_593_600_000_000,
2121 IssueStateKind::Closed,
2122 );
2123 assert_eq!(
2124 count_open(old_subject.clone()).await,
2125 0,
2126 "state changes are visible without cached classification",
2127 );
2128
2129 h.edges.upsert_source(
2130 &issue,
2131 vec![Edge {
2132 kind,
2133 subject: SubjectRef::Did(new_repo),
2134 source: issue.clone(),
2135 sort_micros: next_sort_micros(),
2136 }],
2137 );
2138 h.edges.refresh_issue_counts(&h.state.issue_states, &issue);
2139 assert_eq!(count_open(old_subject).await, 0);
2140 assert_eq!(
2141 count_open(new_subject).await,
2142 1,
2143 "re-keying reprojects acceptance against the new repo owner",
2144 );
2145}
2146
2147#[tokio::test]
2148async fn count_pulls_splits_by_status() {
2149 let h = Harness::new().await;
2150 let repo = did("did:plc:limpet");
2151 let subject = at(&format!("at://{}", repo.as_ref()));
2152 for rk in ["p1", "p2", "p3"] {
2153 h.add_edge(
2154 &nsid("sh.tangled.repo.pull"),
2155 &subject,
2156 &at(&format!("at://did:plc:nel/sh.tangled.repo.pull/{rk}")),
2157 );
2158 }
2159 h.upsert_pull_status(
2160 at("at://did:plc:limpet/sh.tangled.repo.pull.status/s1"),
2161 at("at://did:plc:nel/sh.tangled.repo.pull/p1"),
2162 1_777_593_600_000_000,
2163 PullStatusKind::Merged,
2164 );
2165 h.upsert_pull_status(
2166 at("at://did:plc:limpet/sh.tangled.repo.pull.status/s2"),
2167 at("at://did:plc:nel/sh.tangled.repo.pull/p2"),
2168 1_777_593_600_000_000,
2169 PullStatusKind::Closed,
2170 );
2171
2172 let app = router(h.state.clone());
2173 let counts = |args: &'static [(&'static str, &'static str)]| {
2174 let app = app.clone();
2175 let subject = subject.clone();
2176 async move {
2177 let (_, body) = json_response(
2178 app.oneshot(list_request(
2179 "sh.tangled.repo.countPulls",
2180 subject.as_ref(),
2181 args,
2182 ))
2183 .await
2184 .unwrap(),
2185 )
2186 .await;
2187 body["count"].as_u64().expect("count")
2188 }
2189 };
2190
2191 assert_eq!(counts(&[]).await, 3);
2192 assert_eq!(counts(&[("status", "open")]).await, 1);
2193 assert_eq!(counts(&[("status", "closed")]).await, 1);
2194 assert_eq!(counts(&[("status", "merged")]).await, 1);
2195}
2196
2197#[tokio::test]
2198async fn count_issues_distinct_authors_follows_the_filter() {
2199 let h = Harness::new().await;
2200 let repo = did("did:plc:limpet");
2201 let subject = at(&format!("at://{}", repo.as_ref()));
2202 h.add_edge(
2203 &nsid("sh.tangled.repo.issue"),
2204 &subject,
2205 &at("at://did:plc:nel/sh.tangled.repo.issue/i1"),
2206 );
2207 h.add_edge(
2208 &nsid("sh.tangled.repo.issue"),
2209 &subject,
2210 &at("at://did:plc:olaren/sh.tangled.repo.issue/i2"),
2211 );
2212 h.upsert_issue_state(
2213 at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"),
2214 at("at://did:plc:olaren/sh.tangled.repo.issue/i2"),
2215 1_777_593_600_000_000,
2216 IssueStateKind::Closed,
2217 );
2218
2219 let app = router(h.state.clone());
2220 let (_, body) = json_response(
2221 app.oneshot(list_request(
2222 "sh.tangled.repo.countIssues",
2223 subject.as_ref(),
2224 &[("state", "open")],
2225 ))
2226 .await
2227 .unwrap(),
2228 )
2229 .await;
2230 assert_eq!(body["count"], json!(1));
2231 assert_eq!(
2232 body["distinctAuthors"],
2233 json!(1),
2234 "authors of filtered-out issues must not be counted",
2235 );
2236
2237 for (args, count, distinct) in [
2238 (&[("author", "did:plc:nel")][..], 1, 1),
2239 (&[("author", "did:plc:nel"), ("state", "open")][..], 1, 1),
2240 (&[("author", "did:plc:olaren"), ("state", "open")][..], 0, 0),
2241 ] {
2242 let (_, body) = json_response(
2243 router(h.state.clone())
2244 .oneshot(list_request(
2245 "sh.tangled.repo.countIssues",
2246 subject.as_ref(),
2247 args,
2248 ))
2249 .await
2250 .unwrap(),
2251 )
2252 .await;
2253 assert_eq!(body["count"], json!(count));
2254 assert_eq!(body["distinctAuthors"], json!(distinct));
2255 }
2256}
2257
2258#[tokio::test]
2259async fn list_issues_state_filter_ignores_third_party_state_source() {
2260 let h = Harness::new().await;
2261 let repo = did("did:plc:limpet");
2262 let subject = at(&format!("at://{}", repo.as_ref()));
2263 let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1");
2264 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri);
2265 h.mount(
2266 &did("did:plc:nel"),
2267 &nsid("sh.tangled.repo.issue"),
2268 &rkey("i1"),
2269 issue_body(&repo, "open issue"),
2270 )
2271 .await;
2272 h.upsert_issue_state(
2273 at("at://did:plc:nautilus/sh.tangled.repo.issue.state/spoof"),
2274 issue_uri.clone(),
2275 1_777_593_800_000_000,
2276 IssueStateKind::Closed,
2277 );
2278
2279 let app = router(h.state.clone());
2280 let (status, body) = json_response(
2281 app.oneshot(list_request(
2282 "sh.tangled.repo.listIssues",
2283 subject.as_ref(),
2284 &[("state", "open")],
2285 ))
2286 .await
2287 .unwrap(),
2288 )
2289 .await;
2290 assert_eq!(status, StatusCode::OK);
2291 let items = body["items"].as_array().expect("items array");
2292 assert_eq!(
2293 items.len(),
2294 1,
2295 "third-party Closed record must not flip filter result for state=open",
2296 );
2297 assert_eq!(items[0]["state"], json!("open"));
2298 assert!(
2299 items[0].get("stateUpdatedAt").is_none(),
2300 "third-party state source must not surface stateUpdatedAt",
2301 );
2302}
2303
2304#[tokio::test]
2305async fn list_pulls_status_filter_ignores_third_party_status_source() {
2306 let h = Harness::new().await;
2307 let repo = did("did:plc:limpet");
2308 let subject = at(&format!("at://{}", repo.as_ref()));
2309 let pull_uri = at("at://did:plc:nel/sh.tangled.repo.pull/p1");
2310 h.add_edge(&nsid("sh.tangled.repo.pull"), &subject, &pull_uri);
2311 h.mount(
2312 &did("did:plc:nel"),
2313 &nsid("sh.tangled.repo.pull"),
2314 &rkey("p1"),
2315 pull_body(&repo, "wip"),
2316 )
2317 .await;
2318 h.upsert_pull_status(
2319 at("at://did:plc:nautilus/sh.tangled.repo.pull.status/spoof"),
2320 pull_uri.clone(),
2321 1_777_593_800_000_000,
2322 PullStatusKind::Merged,
2323 );
2324
2325 let app = router(h.state.clone());
2326 let (status, body) = json_response(
2327 app.oneshot(list_request(
2328 "sh.tangled.repo.listPulls",
2329 subject.as_ref(),
2330 &[("status", "merged")],
2331 ))
2332 .await
2333 .unwrap(),
2334 )
2335 .await;
2336 assert_eq!(status, StatusCode::OK);
2337 let items = body["items"].as_array().expect("items array");
2338 assert_eq!(
2339 items.len(),
2340 0,
2341 "third-party Merged record must not satisfy status=merged"
2342 );
2343}
2344
2345#[tokio::test]
2346async fn list_issues_state_filter_accepts_repo_owner_state_source() {
2347 let h = Harness::new().await;
2348 let repo_owner = did("did:plc:limpet");
2349 let subject = at(&format!("at://{}", repo_owner.as_ref()));
2350 let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1");
2351 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri);
2352 h.mount(
2353 &did("did:plc:nel"),
2354 &nsid("sh.tangled.repo.issue"),
2355 &rkey("i1"),
2356 issue_body(&repo_owner, "owner closed"),
2357 )
2358 .await;
2359 h.upsert_issue_state(
2360 at("at://did:plc:limpet/sh.tangled.repo.issue.state/legit"),
2361 issue_uri.clone(),
2362 1_777_593_800_000_000,
2363 IssueStateKind::Closed,
2364 );
2365
2366 let app = router(h.state.clone());
2367 let (status, body) = json_response(
2368 app.oneshot(list_request(
2369 "sh.tangled.repo.listIssues",
2370 subject.as_ref(),
2371 &[("state", "closed")],
2372 ))
2373 .await
2374 .unwrap(),
2375 )
2376 .await;
2377 assert_eq!(status, StatusCode::OK);
2378 let items = body["items"].as_array().expect("items array");
2379 assert_eq!(
2380 items.len(),
2381 1,
2382 "repo-owner state record must satisfy state=closed"
2383 );
2384 assert_eq!(items[0]["state"], json!("closed"));
2385}
2386
2387#[tokio::test]
2388async fn list_issues_order_asc_returns_oldest_first() {
2389 let h = Harness::new().await;
2390 let repo = did("did:plc:limpet");
2391 let subject = at(&format!("at://{}", repo.as_ref()));
2392 let rkeys = ["a", "b", "c"];
2393 stream::iter(rkeys)
2394 .for_each(|r| {
2395 let h = &h;
2396 let subject = subject.clone();
2397 let repo = repo.clone();
2398 async move {
2399 let rk = rkey(r);
2400 let issue_uri = at(&format!(
2401 "at://did:plc:nel/sh.tangled.repo.issue/{}",
2402 rk.as_ref()
2403 ));
2404 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri);
2405 h.mount(
2406 &did("did:plc:nel"),
2407 &nsid("sh.tangled.repo.issue"),
2408 &rk,
2409 issue_body(&repo, &format!("issue-{}", rk.as_ref())),
2410 )
2411 .await;
2412 }
2413 })
2414 .await;
2415
2416 let app = router(h.state.clone());
2417 let (_, asc) = json_response(
2418 app.clone()
2419 .oneshot(list_request(
2420 "sh.tangled.repo.listIssues",
2421 subject.as_ref(),
2422 &[("order", "asc")],
2423 ))
2424 .await
2425 .unwrap(),
2426 )
2427 .await;
2428 let (_, desc) = json_response(
2429 app.oneshot(list_request(
2430 "sh.tangled.repo.listIssues",
2431 subject.as_ref(),
2432 &[("order", "desc")],
2433 ))
2434 .await
2435 .unwrap(),
2436 )
2437 .await;
2438 let asc_uris: Vec<_> = asc["items"]
2439 .as_array()
2440 .unwrap()
2441 .iter()
2442 .map(|i| i["uri"].as_str().unwrap().to_owned())
2443 .collect();
2444 let desc_uris: Vec<_> = desc["items"]
2445 .as_array()
2446 .unwrap()
2447 .iter()
2448 .map(|i| i["uri"].as_str().unwrap().to_owned())
2449 .collect();
2450 let mut reversed = asc_uris.clone();
2451 reversed.reverse();
2452 assert_eq!(asc_uris.len(), 3);
2453 assert_eq!(desc_uris, reversed, "desc must be exact reverse of asc");
2454}
2455
2456#[tokio::test]
2457async fn list_issues_by_state_filter_narrows_results() {
2458 let h = Harness::new().await;
2459 let author = did("did:plc:nel");
2460 let repo = did("did:plc:limpet");
2461 let open_uri = at("at://did:plc:nel/sh.tangled.repo.issue/open1");
2462 let closed_uri = at("at://did:plc:nel/sh.tangled.repo.issue/closed1");
2463 let author_subject = at(&format!("at://{}", author.as_ref()));
2464 h.edges.add(Edge {
2465 kind: nsid("sh.tangled.repo.issue.by"),
2466 subject: SubjectRef::Did(author.clone()),
2467 source: open_uri.clone(),
2468 sort_micros: next_sort_micros(),
2469 });
2470 h.edges.add(Edge {
2471 kind: nsid("sh.tangled.repo.issue.by"),
2472 subject: SubjectRef::Did(author.clone()),
2473 source: closed_uri.clone(),
2474 sort_micros: next_sort_micros(),
2475 });
2476 h.mount(
2477 &author,
2478 &nsid("sh.tangled.repo.issue"),
2479 &rkey("open1"),
2480 issue_body(&repo, "still open"),
2481 )
2482 .await;
2483 h.mount(
2484 &author,
2485 &nsid("sh.tangled.repo.issue"),
2486 &rkey("closed1"),
2487 issue_body(&repo, "shut"),
2488 )
2489 .await;
2490 h.upsert_issue_state(
2491 at("at://did:plc:nel/sh.tangled.repo.issue.state/s1"),
2492 closed_uri.clone(),
2493 1_777_593_800_000_000,
2494 IssueStateKind::Closed,
2495 );
2496
2497 let app = router(h.state.clone());
2498 let (status, body) = json_response(
2499 app.oneshot(list_request(
2500 "sh.tangled.repo.listIssuesBy",
2501 author_subject.as_ref(),
2502 &[("state", "closed")],
2503 ))
2504 .await
2505 .unwrap(),
2506 )
2507 .await;
2508 assert_eq!(status, StatusCode::OK);
2509 let items = body["items"].as_array().expect("items array");
2510 assert_eq!(
2511 items.len(),
2512 1,
2513 "only the closed issue survives state=closed"
2514 );
2515 assert_eq!(items[0]["uri"], json!(closed_uri.as_ref()));
2516}
2517
2518#[tokio::test]
2519async fn knot_owned_member_is_synthesized_without_slingshot() {
2520 let harness = Harness::new().await;
2521 let knot = bobbin_types::knot_acl::host_to_knot_did("kt.oyster.cafe").unwrap();
2522 let subject = did("did:plc:boltless");
2523 let created = chrono::DateTime::parse_from_rfc3339("2026-06-01T00:00:00Z").unwrap();
2524 let micros = created.timestamp_micros() as u64;
2525 let (source, edges) = bobbin_types::knot_acl::member_upsert(&knot, &subject, micros).unwrap();
2526 harness.edges.upsert_source(&source, edges);
2527 harness.promote_ready(1, 1);
2528
2529 let (status, body) = json_response(
2530 router(harness.state.clone())
2531 .oneshot(list_request(
2532 "sh.tangled.knot.listMembers",
2533 subject.as_ref(),
2534 &[],
2535 ))
2536 .await
2537 .unwrap(),
2538 )
2539 .await;
2540
2541 assert_eq!(status, StatusCode::OK);
2542 let items = body["items"].as_array().expect("items array");
2543 assert_eq!(
2544 items.len(),
2545 1,
2546 "synthesized member must hydrate with no slingshot mock mounted"
2547 );
2548 assert_eq!(items[0]["uri"], json!(source.as_ref()));
2549 assert!(items[0]["cid"].is_null());
2550 assert_eq!(items[0]["value"]["domain"], json!("kt.oyster.cafe"));
2551 assert_eq!(items[0]["value"]["subject"], json!("did:plc:boltless"));
2552 let got = chrono::DateTime::parse_from_rfc3339(
2553 items[0]["value"]["createdAt"]
2554 .as_str()
2555 .expect("createdAt string"),
2556 )
2557 .unwrap();
2558 assert_eq!(got.timestamp_micros(), micros as i64);
2559}
2560
2561#[tokio::test]
2562async fn knot_owned_member_lists_by_knot_did() {
2563 let harness = Harness::new().await;
2564 let knot = bobbin_types::knot_acl::host_to_knot_did("kt.oyster.cafe").unwrap();
2565 let subject = did("did:plc:boltless");
2566 let created = chrono::DateTime::parse_from_rfc3339("2026-06-01T00:00:00Z").unwrap();
2567 let micros = created.timestamp_micros() as u64;
2568 let (source, edges) = bobbin_types::knot_acl::member_upsert(&knot, &subject, micros).unwrap();
2569 harness.edges.upsert_source(&source, edges);
2570 harness.promote_ready(1, 1);
2571
2572 let (status, body) = json_response(
2573 router(harness.state.clone())
2574 .oneshot(list_request(
2575 "sh.tangled.knot.listMembersBy",
2576 knot.as_ref(),
2577 &[],
2578 ))
2579 .await
2580 .unwrap(),
2581 )
2582 .await;
2583
2584 assert_eq!(status, StatusCode::OK);
2585 let items = body["items"].as_array().expect("items array");
2586 assert_eq!(items.len(), 1);
2587 assert_eq!(items[0]["uri"], json!(source.as_ref()));
2588 assert!(items[0]["cid"].is_null());
2589 assert_eq!(items[0]["value"]["domain"], json!("kt.oyster.cafe"));
2590 assert_eq!(items[0]["value"]["subject"], json!("did:plc:boltless"));
2591}
2592
2593#[tokio::test]
2594async fn knot_owned_collaborator_is_synthesized_without_slingshot() {
2595 let harness = Harness::new().await;
2596 let repo = did("did:plc:scallop");
2597 let subject = did("did:plc:olaren");
2598 let created = chrono::DateTime::parse_from_rfc3339("2026-06-03T12:00:00Z").unwrap();
2599 let micros = created.timestamp_micros() as u64;
2600 let (source, edges) =
2601 bobbin_types::knot_acl::collaborator_upsert(&repo, &subject, micros).unwrap();
2602 harness.edges.upsert_source(&source, edges);
2603 harness.promote_ready(1, 1);
2604
2605 let (status, body) = json_response(
2606 router(harness.state.clone())
2607 .oneshot(list_request(
2608 "sh.tangled.repo.listCollaborators",
2609 repo.as_ref(),
2610 &[],
2611 ))
2612 .await
2613 .unwrap(),
2614 )
2615 .await;
2616
2617 assert_eq!(status, StatusCode::OK);
2618 let items = body["items"].as_array().expect("items array");
2619 assert_eq!(items.len(), 1);
2620 assert_eq!(items[0]["uri"], json!(source.as_ref()));
2621 assert!(items[0]["cid"].is_null());
2622 assert_eq!(items[0]["value"]["repo"], json!("did:plc:scallop"));
2623 assert_eq!(items[0]["value"]["subject"], json!("did:plc:olaren"));
2624}
2625
2626#[tokio::test]
2627async fn knot_owned_collaborator_lists_by_subject_did() {
2628 let harness = Harness::new().await;
2629 let repo = did("did:plc:scallop");
2630 let subject = did("did:plc:olaren");
2631 let created = chrono::DateTime::parse_from_rfc3339("2026-06-03T12:00:00Z").unwrap();
2632 let micros = created.timestamp_micros() as u64;
2633 let (source, edges) =
2634 bobbin_types::knot_acl::collaborator_upsert(&repo, &subject, micros).unwrap();
2635 harness.edges.upsert_source(&source, edges);
2636 harness.promote_ready(1, 1);
2637
2638 let (status, body) = json_response(
2639 router(harness.state.clone())
2640 .oneshot(list_request(
2641 "sh.tangled.repo.listCollaboratorsBy",
2642 subject.as_ref(),
2643 &[],
2644 ))
2645 .await
2646 .unwrap(),
2647 )
2648 .await;
2649
2650 assert_eq!(status, StatusCode::OK);
2651 let items = body["items"].as_array().expect("items array");
2652 assert_eq!(items.len(), 1);
2653 assert_eq!(items[0]["uri"], json!(source.as_ref()));
2654 assert!(items[0]["cid"].is_null());
2655 assert_eq!(items[0]["value"]["repo"], json!("did:plc:scallop"));
2656 assert_eq!(items[0]["value"]["subject"], json!("did:plc:olaren"));
2657}