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