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 }
117
118 async fn mount(
119 &self,
120 did: &Did<DefaultStr>,
121 collection: &Nsid<DefaultStr>,
122 rkey: &Rkey<DefaultStr>,
123 value: Value,
124 ) {
125 let uri = format!(
126 "at://{}/{}/{}",
127 did.as_ref(),
128 collection.as_ref(),
129 rkey.as_ref()
130 );
131 let body = json!({ "uri": uri, "cid": CID, "value": value });
132 Mock::given(method("GET"))
133 .and(path("/xrpc/com.atproto.repo.getRecord"))
134 .and(query_param("repo", did.as_ref()))
135 .and(query_param("collection", collection.as_ref()))
136 .and(query_param("rkey", rkey.as_ref()))
137 .respond_with(ResponseTemplate::new(200).set_body_json(body))
138 .mount(&self.server)
139 .await;
140 }
141
142 fn promote_ready(&self, events: u64, cursor: u64) {
143 self.coverage.update(|_| Coverage::Ready {
144 events_processed: events,
145 last_cursor: HydrantCursor::new(cursor),
146 });
147 }
148
149 fn warming(&self, events: u64, cursor: u64) {
150 self.coverage.update(|_| Coverage::Warming {
151 events_processed: events,
152 last_cursor: HydrantCursor::new(cursor),
153 });
154 }
155}
156
157fn list_request(endpoint: &str, subject: &str, extras: &[(&str, &str)]) -> Request<Body> {
158 let mut qs = format!("subject={}", encode(subject));
159 extras.iter().for_each(|(k, v)| {
160 qs.push('&');
161 qs.push_str(k);
162 qs.push('=');
163 qs.push_str(&encode(v));
164 });
165 Request::builder()
166 .uri(format!("/xrpc/{endpoint}?{qs}"))
167 .body(Body::empty())
168 .unwrap()
169}
170
171fn encode(s: &str) -> String {
172 byte_serialize(s.as_bytes()).collect()
173}
174
175async fn json_response(resp: axum::response::Response) -> (StatusCode, Value) {
176 let status = resp.status();
177 let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap();
178 let parsed: Value = serde_json::from_slice(&bytes).expect("JSON body");
179 (status, parsed)
180}
181
182fn issue_body(repo_did: &Did<DefaultStr>, title: &str) -> Value {
183 json!({
184 "$type": "sh.tangled.repo.issue",
185 "repo": repo_did.as_ref(),
186 "title": title,
187 "createdAt": "2026-05-01T00:00:00Z"
188 })
189}
190
191fn pull_body(repo_did: &Did<DefaultStr>, title: &str) -> Value {
192 json!({
193 "$type": "sh.tangled.repo.pull",
194 "title": title,
195 "createdAt": "2026-05-01T00:00:00Z",
196 "rounds": [],
197 "target": {
198 "repo": repo_did.as_ref(),
199 "branch": "main"
200 }
201 })
202}
203
204fn star_body(subject_did: &Did<DefaultStr>) -> Value {
205 json!({
206 "$type": "sh.tangled.feed.star",
207 "createdAt": "2026-05-01T00:00:00Z",
208 "subject": {
209 "$type": "sh.tangled.feed.star#repo",
210 "did": subject_did.as_ref()
211 }
212 })
213}
214
215fn follow_body(subject_did: &Did<DefaultStr>) -> Value {
216 json!({
217 "$type": "sh.tangled.graph.follow",
218 "createdAt": "2026-05-01T00:00:00Z",
219 "subject": subject_did.as_ref()
220 })
221}
222
223#[tokio::test]
224async fn list_issues_with_no_edges_returns_empty_items() {
225 let h = Harness::new().await;
226 let app = router(h.state.clone());
227 let resp = app
228 .oneshot(list_request(
229 "sh.tangled.repo.listIssues",
230 "at://did:plc:abalone",
231 &[],
232 ))
233 .await
234 .unwrap();
235 let (status, body) = json_response(resp).await;
236 assert_eq!(status, StatusCode::OK);
237 assert_eq!(body["items"], json!([]));
238 assert!(body["cursor"].is_null());
239}
240
241#[tokio::test]
242async fn count_issues_with_no_edges_returns_zero() {
243 let h = Harness::new().await;
244 let app = router(h.state.clone());
245 let resp = app
246 .oneshot(list_request(
247 "sh.tangled.repo.countIssues",
248 "at://did:plc:abalone",
249 &[],
250 ))
251 .await
252 .unwrap();
253 let (status, body) = json_response(resp).await;
254 assert_eq!(status, StatusCode::OK);
255 assert_eq!(body["count"], json!(0));
256 assert_eq!(body["distinctAuthors"], json!(0));
257}
258
259#[tokio::test]
260async fn list_issues_hydrates_via_slingshot_when_edges_present() {
261 let h = Harness::new().await;
262 let repo = did("did:plc:abalone");
263 let subject = at(&format!("at://{}", repo.as_ref()));
264 let owners = [
265 ("did:plc:nel", "i1", "first"),
266 ("did:plc:olaren", "i2", "second"),
267 ];
268 stream::iter(owners)
269 .for_each(|(d, r, title)| {
270 let h = &h;
271 let subject = subject.clone();
272 let repo = repo.clone();
273 async move {
274 let d_did = did(d);
275 let rk = rkey(r);
276 h.add_edge(
277 &nsid("sh.tangled.repo.issue"),
278 &subject,
279 &at(&format!(
280 "at://{}/sh.tangled.repo.issue/{}",
281 d_did.as_ref(),
282 rk.as_ref()
283 )),
284 );
285 h.mount(
286 &d_did,
287 &nsid("sh.tangled.repo.issue"),
288 &rk,
289 issue_body(&repo, title),
290 )
291 .await;
292 }
293 })
294 .await;
295
296 let app = router(h.state.clone());
297 let resp = app
298 .oneshot(list_request(
299 "sh.tangled.repo.listIssues",
300 subject.as_ref(),
301 &[],
302 ))
303 .await
304 .unwrap();
305 let (status, body) = json_response(resp).await;
306 assert_eq!(status, StatusCode::OK);
307 let items = body["items"].as_array().expect("items array");
308 assert_eq!(items.len(), 2);
309 let titles: Vec<&str> = items
310 .iter()
311 .map(|v| v["value"]["title"].as_str().unwrap())
312 .collect();
313 assert!(titles.contains(&"first"));
314 assert!(titles.contains(&"second"));
315 assert_eq!(items[0]["cid"], CID);
316 assert!(items[0]["uri"].as_str().unwrap().starts_with("at://"));
317}
318
319#[tokio::test]
320async fn count_distinct_authors_dedupes_per_author() {
321 let h = Harness::new().await;
322 let subject = at("at://did:plc:abalone");
323 h.add_edge(
324 &nsid("sh.tangled.feed.star"),
325 &subject,
326 &at("at://did:plc:nel/sh.tangled.feed.star/s1"),
327 );
328 h.add_edge(
329 &nsid("sh.tangled.feed.star"),
330 &subject,
331 &at("at://did:plc:nel/sh.tangled.feed.star/s2"),
332 );
333 h.add_edge(
334 &nsid("sh.tangled.feed.star"),
335 &subject,
336 &at("at://did:plc:olaren/sh.tangled.feed.star/s3"),
337 );
338
339 let app = router(h.state.clone());
340 let resp = app
341 .oneshot(list_request(
342 "sh.tangled.feed.countStars",
343 subject.as_ref(),
344 &[],
345 ))
346 .await
347 .unwrap();
348 let (_, body) = json_response(resp).await;
349 assert_eq!(body["count"], json!(3));
350 assert_eq!(body["distinctAuthors"], json!(2));
351}
352
353#[tokio::test]
354async fn list_items_stable_across_coverage_promotion() {
355 let h = Harness::new().await;
356 let subject = at("at://did:plc:abalone");
357 let nel = did("did:plc:nel");
358 h.add_edge(
359 &nsid("sh.tangled.feed.star"),
360 &subject,
361 &at(&format!("at://{}/sh.tangled.feed.star/s1", nel.as_ref())),
362 );
363 h.mount(
364 &nel,
365 &nsid("sh.tangled.feed.star"),
366 &rkey("s1"),
367 star_body(&did("did:plc:abalone")),
368 )
369 .await;
370
371 let app = router(h.state.clone());
372 h.warming(1, 5);
373 let (_, before) = json_response(
374 app.clone()
375 .oneshot(list_request(
376 "sh.tangled.feed.listStars",
377 subject.as_ref(),
378 &[],
379 ))
380 .await
381 .unwrap(),
382 )
383 .await;
384 assert_eq!(before["items"].as_array().unwrap().len(), 1);
385
386 h.promote_ready(2, 9);
387 let (_, after) = json_response(
388 app.oneshot(list_request(
389 "sh.tangled.feed.listStars",
390 subject.as_ref(),
391 &[],
392 ))
393 .await
394 .unwrap(),
395 )
396 .await;
397 assert_eq!(
398 after["items"].as_array().unwrap().len(),
399 before["items"].as_array().unwrap().len(),
400 );
401 assert_eq!(after["items"], before["items"]);
402}
403
404#[tokio::test]
405async fn list_paginates_via_cursor() {
406 let h = Harness::new().await;
407 let subject = at("at://did:plc:abalone");
408 let repo = did("did:plc:abalone");
409 let owners = [
410 ("did:plc:nel", "i1"),
411 ("did:plc:olaren", "i2"),
412 ("did:plc:teq", "i3"),
413 ("did:plc:lyna", "i4"),
414 ("did:plc:bailey", "i5"),
415 ];
416 stream::iter(owners)
417 .for_each(|(d, r)| {
418 let h = &h;
419 let subject = subject.clone();
420 let repo = repo.clone();
421 async move {
422 let d_did = did(d);
423 let rk = rkey(r);
424 h.add_edge(
425 &nsid("sh.tangled.repo.issue"),
426 &subject,
427 &at(&format!(
428 "at://{}/sh.tangled.repo.issue/{}",
429 d_did.as_ref(),
430 rk.as_ref()
431 )),
432 );
433 h.mount(
434 &d_did,
435 &nsid("sh.tangled.repo.issue"),
436 &rk,
437 issue_body(&repo, &format!("issue-{}", rk.as_ref())),
438 )
439 .await;
440 }
441 })
442 .await;
443
444 let app = router(h.state.clone());
445 let (_, page1) = json_response(
446 app.clone()
447 .oneshot(list_request(
448 "sh.tangled.repo.listIssues",
449 subject.as_ref(),
450 &[("limit", "2")],
451 ))
452 .await
453 .unwrap(),
454 )
455 .await;
456 let page1_items = page1["items"].as_array().unwrap().clone();
457 assert_eq!(page1_items.len(), 2);
458 let cursor = page1["cursor"]
459 .as_str()
460 .expect("first page must yield a cursor")
461 .to_owned();
462 assert!(
463 PageToken::decode_token(&cursor).is_ok(),
464 "cursor must be a TID-shaped token"
465 );
466
467 let (_, page2) = json_response(
468 app.oneshot(list_request(
469 "sh.tangled.repo.listIssues",
470 subject.as_ref(),
471 &[("limit", "10"), ("cursor", &cursor)],
472 ))
473 .await
474 .unwrap(),
475 )
476 .await;
477 let page2_items = page2["items"].as_array().unwrap().clone();
478 assert_eq!(page2_items.len(), 3);
479 assert!(page2["cursor"].is_null(), "tail page must not promise more");
480
481 let union: Vec<&str> = page1_items
482 .iter()
483 .chain(page2_items.iter())
484 .map(|item| item["uri"].as_str().unwrap())
485 .collect();
486 assert_eq!(union.len(), owners.len(), "union covers every owner");
487 let mut sorted = union.clone();
488 sorted.sort();
489 sorted.dedup();
490 assert_eq!(sorted.len(), owners.len(), "no duplicates across pages");
491}
492
493#[tokio::test]
494async fn pagination_unaffected_by_coverage_promotion() {
495 let h = Harness::new().await;
496 let subject = at("at://did:plc:abalone");
497 let repo = did("did:plc:abalone");
498 let owners = [("did:plc:nel", "i1"), ("did:plc:olaren", "i2")];
499 stream::iter(owners)
500 .for_each(|(d, r)| {
501 let h = &h;
502 let subject = subject.clone();
503 let repo = repo.clone();
504 async move {
505 let d_did = did(d);
506 let rk = rkey(r);
507 h.add_edge(
508 &nsid("sh.tangled.repo.issue"),
509 &subject,
510 &at(&format!(
511 "at://{}/sh.tangled.repo.issue/{}",
512 d_did.as_ref(),
513 rk.as_ref()
514 )),
515 );
516 h.mount(
517 &d_did,
518 &nsid("sh.tangled.repo.issue"),
519 &rk,
520 issue_body(&repo, &format!("issue-{}", rk.as_ref())),
521 )
522 .await;
523 }
524 })
525 .await;
526
527 h.warming(1, 5);
528 let app = router(h.state.clone());
529 let (_, page1) = json_response(
530 app.clone()
531 .oneshot(list_request(
532 "sh.tangled.repo.listIssues",
533 subject.as_ref(),
534 &[("limit", "1")],
535 ))
536 .await
537 .unwrap(),
538 )
539 .await;
540 let cursor = page1["cursor"].as_str().unwrap().to_owned();
541
542 h.promote_ready(2, 9);
543 let (_, page2) = json_response(
544 app.oneshot(list_request(
545 "sh.tangled.repo.listIssues",
546 subject.as_ref(),
547 &[("limit", "10"), ("cursor", &cursor)],
548 ))
549 .await
550 .unwrap(),
551 )
552 .await;
553 assert_eq!(page2["items"].as_array().unwrap().len(), 1);
554}
555
556#[tokio::test]
557async fn invalid_cursor_returns_400() {
558 let h = Harness::new().await;
559 let app = router(h.state.clone());
560 let resp = app
561 .oneshot(list_request(
562 "sh.tangled.repo.listIssues",
563 "at://did:plc:abalone",
564 &[("cursor", "not-a-number")],
565 ))
566 .await
567 .unwrap();
568 let (status, body) = json_response(resp).await;
569 assert_eq!(status, StatusCode::BAD_REQUEST);
570 assert_eq!(body["error"], "InvalidRequest");
571}
572
573#[tokio::test]
574async fn list_follows_subject_is_followee_did() {
575 let h = Harness::new().await;
576 let followee = did("did:plc:bailey");
577 let subject = at(&format!("at://{}", followee.as_ref()));
578 h.add_edge(
579 &nsid("sh.tangled.graph.follow"),
580 &subject,
581 &at("at://did:plc:nel/sh.tangled.graph.follow/f1"),
582 );
583 h.mount(
584 &did("did:plc:nel"),
585 &nsid("sh.tangled.graph.follow"),
586 &rkey("f1"),
587 follow_body(&followee),
588 )
589 .await;
590
591 let app = router(h.state.clone());
592 let (status, body) = json_response(
593 app.oneshot(list_request(
594 "sh.tangled.graph.listFollows",
595 subject.as_ref(),
596 &[],
597 ))
598 .await
599 .unwrap(),
600 )
601 .await;
602 assert_eq!(status, StatusCode::OK);
603 let items = body["items"].as_array().unwrap();
604 assert_eq!(items.len(), 1);
605 assert_eq!(items[0]["value"]["subject"], followee.as_ref());
606}
607
608#[tokio::test]
609async fn upstream_failure_during_hydration_drops_only_that_item() {
610 let h = Harness::new().await;
611 let subject = at("at://did:plc:squid");
612 let kind = nsid("sh.tangled.repo.issue");
613 let repo = did("did:plc:squid");
614 h.add_edge(
615 &kind,
616 &subject,
617 &at("at://did:plc:nel/sh.tangled.repo.issue/ok"),
618 );
619 h.add_edge(
620 &kind,
621 &subject,
622 &at("at://did:plc:teq/sh.tangled.repo.issue/flaky"),
623 );
624 h.mount(
625 &did("did:plc:nel"),
626 &kind,
627 &rkey("ok"),
628 issue_body(&repo, "kelp survey"),
629 )
630 .await;
631 Mock::given(method("GET"))
632 .and(path("/xrpc/com.atproto.repo.getRecord"))
633 .and(query_param("repo", "did:plc:teq"))
634 .and(query_param("collection", "sh.tangled.repo.issue"))
635 .and(query_param("rkey", "flaky"))
636 .respond_with(ResponseTemplate::new(503))
637 .mount(&h.server)
638 .await;
639
640 let app = router(h.state.clone());
641 let (status, body) = json_response(
642 app.oneshot(list_request(
643 "sh.tangled.repo.listIssues",
644 subject.as_ref(),
645 &[],
646 ))
647 .await
648 .unwrap(),
649 )
650 .await;
651 assert_eq!(status, StatusCode::OK);
652 let items = body["items"].as_array().expect("items array");
653 assert_eq!(items.len(), 1, "flaky item dropped, healthy sibling kept");
654 assert_eq!(
655 items[0]["uri"].as_str().unwrap(),
656 "at://did:plc:nel/sh.tangled.repo.issue/ok",
657 );
658}
659
660#[tokio::test]
661async fn transient_failure_keeps_edge_so_count_stays_whole() {
662 let h = Harness::new().await;
663 let subject = at("at://did:plc:squid");
664 let kind = nsid("sh.tangled.repo.issue");
665 let repo = did("did:plc:squid");
666 h.add_edge(
667 &kind,
668 &subject,
669 &at("at://did:plc:nel/sh.tangled.repo.issue/ok"),
670 );
671 h.add_edge(
672 &kind,
673 &subject,
674 &at("at://did:plc:teq/sh.tangled.repo.issue/flaky"),
675 );
676 h.mount(
677 &did("did:plc:nel"),
678 &kind,
679 &rkey("ok"),
680 issue_body(&repo, "kelp survey"),
681 )
682 .await;
683 Mock::given(method("GET"))
684 .and(path("/xrpc/com.atproto.repo.getRecord"))
685 .and(query_param("repo", "did:plc:teq"))
686 .and(query_param("collection", "sh.tangled.repo.issue"))
687 .and(query_param("rkey", "flaky"))
688 .respond_with(ResponseTemplate::new(503))
689 .mount(&h.server)
690 .await;
691
692 let app = router(h.state.clone());
693 let (status, body) = json_response(
694 app.clone()
695 .oneshot(list_request(
696 "sh.tangled.repo.listIssues",
697 subject.as_ref(),
698 &[],
699 ))
700 .await
701 .unwrap(),
702 )
703 .await;
704 assert_eq!(status, StatusCode::OK);
705 assert_eq!(body["items"].as_array().unwrap().len(), 1);
706
707 let (cstatus, cbody) = json_response(
708 app.oneshot(list_request(
709 "sh.tangled.repo.countIssues",
710 subject.as_ref(),
711 &[],
712 ))
713 .await
714 .unwrap(),
715 )
716 .await;
717 assert_eq!(cstatus, StatusCode::OK);
718 assert_eq!(
719 cbody["count"],
720 json!(2),
721 "a transient 503 must not evict the edge, count stays whole",
722 );
723}
724
725#[tokio::test]
726async fn gone_item_is_evicted_so_count_converges_to_list() {
727 let h = Harness::new().await;
728 let subject = at("at://did:plc:squid");
729 let kind = nsid("sh.tangled.repo.issue");
730 let repo = did("did:plc:squid");
731 h.add_edge(
732 &kind,
733 &subject,
734 &at("at://did:plc:nel/sh.tangled.repo.issue/ok"),
735 );
736 h.add_edge(
737 &kind,
738 &subject,
739 &at("at://did:plc:teq/sh.tangled.repo.issue/gone"),
740 );
741 h.mount(
742 &did("did:plc:nel"),
743 &kind,
744 &rkey("ok"),
745 issue_body(&repo, "kelp survey"),
746 )
747 .await;
748 Mock::given(method("GET"))
749 .and(path("/xrpc/com.atproto.repo.getRecord"))
750 .and(query_param("repo", "did:plc:teq"))
751 .and(query_param("collection", "sh.tangled.repo.issue"))
752 .and(query_param("rkey", "gone"))
753 .respond_with(ResponseTemplate::new(404))
754 .mount(&h.server)
755 .await;
756
757 let app = router(h.state.clone());
758 let (status, body) = json_response(
759 app.clone()
760 .oneshot(list_request(
761 "sh.tangled.repo.listIssues",
762 subject.as_ref(),
763 &[],
764 ))
765 .await
766 .unwrap(),
767 )
768 .await;
769 assert_eq!(status, StatusCode::OK);
770 assert_eq!(
771 body["items"].as_array().unwrap().len(),
772 1,
773 "gone item dropped from the page"
774 );
775
776 let (cstatus, cbody) = json_response(
777 app.oneshot(list_request(
778 "sh.tangled.repo.countIssues",
779 subject.as_ref(),
780 &[],
781 ))
782 .await
783 .unwrap(),
784 )
785 .await;
786 assert_eq!(cstatus, StatusCode::OK);
787 assert_eq!(
788 cbody["count"],
789 json!(1),
790 "a definitive 404 must evict the dead edge so count matches the list",
791 );
792}
793
794#[tokio::test]
795async fn handle_authority_subject_is_400() {
796 let h = Harness::new().await;
797 let app = router(h.state.clone());
798 let cases = [
799 "sh.tangled.feed.listStars",
800 "sh.tangled.feed.countStars",
801 "sh.tangled.graph.listFollows",
802 "sh.tangled.graph.countFollows",
803 "sh.tangled.repo.listIssues",
804 "sh.tangled.repo.countIssues",
805 "sh.tangled.repo.listPulls",
806 "sh.tangled.repo.countPulls",
807 "sh.tangled.feed.listComments",
808 "sh.tangled.feed.countComments",
809 ];
810 stream::iter(cases)
811 .for_each(|endpoint| {
812 let app = app.clone();
813 async move {
814 let resp = app
815 .oneshot(list_request(endpoint, "at://oyster.cafe", &[]))
816 .await
817 .unwrap();
818 let (status, body) = json_response(resp).await;
819 assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint}");
820 assert_eq!(body["error"], "InvalidRequest", "{endpoint}");
821 assert!(
822 body["message"]
823 .as_str()
824 .unwrap_or_default()
825 .contains("did, not a handle"),
826 "{endpoint}: {}",
827 body["message"]
828 );
829 }
830 })
831 .await;
832}
833
834#[tokio::test]
835async fn empty_subject_is_400() {
836 let h = Harness::new().await;
837 let app = router(h.state.clone());
838 let resp = app
839 .oneshot(list_request("sh.tangled.repo.listIssues", "", &[]))
840 .await
841 .unwrap();
842 let (status, body) = json_response(resp).await;
843 assert_eq!(status, StatusCode::BAD_REQUEST);
844 assert_eq!(body["error"], "InvalidRequest");
845}
846
847#[tokio::test]
848async fn limit_below_min_or_above_max_is_400() {
849 let h = Harness::new().await;
850 let app = router(h.state.clone());
851 let cases = [("0", "below"), ("1001", "above")];
852 stream::iter(cases)
853 .for_each(|(limit, label)| {
854 let app = app.clone();
855 async move {
856 let resp = app
857 .oneshot(list_request(
858 "sh.tangled.repo.listIssues",
859 "at://did:plc:abalone",
860 &[("limit", limit)],
861 ))
862 .await
863 .unwrap();
864 let (status, body) = json_response(resp).await;
865 assert_eq!(status, StatusCode::BAD_REQUEST, "limit {label}");
866 assert_eq!(body["error"], "InvalidRequest", "limit {label}");
867 }
868 })
869 .await;
870}
871
872#[tokio::test]
873async fn count_after_remove_source_returns_zero() {
874 let h = Harness::new().await;
875 let subject = at("at://did:plc:abalone");
876 let source = at("at://did:plc:nel/sh.tangled.feed.star/s1");
877 h.add_edge(&nsid("sh.tangled.feed.star"), &subject, &source);
878 h.edges.remove_source(&source);
879
880 let app = router(h.state.clone());
881 let (_, body) = json_response(
882 app.oneshot(list_request(
883 "sh.tangled.feed.countStars",
884 subject.as_ref(),
885 &[],
886 ))
887 .await
888 .unwrap(),
889 )
890 .await;
891 assert_eq!(body["count"], json!(0));
892 assert_eq!(body["distinctAuthors"], json!(0));
893}
894
895#[tokio::test]
896async fn list_feed_comments_hydrates_end_to_end() {
897 let h = Harness::new().await;
898 let issue_uri = at("at://did:plc:abalone/sh.tangled.repo.issue/i1");
899 let nel = did("did:plc:nel");
900 let rk = rkey("c1");
901 h.add_edge(
902 &nsid("sh.tangled.feed.comment"),
903 &issue_uri,
904 &at(&format!(
905 "at://{}/sh.tangled.feed.comment/{}",
906 nel.as_ref(),
907 rk.as_ref()
908 )),
909 );
910 h.mount(
911 &nel,
912 &nsid("sh.tangled.feed.comment"),
913 &rk,
914 json!({
915 "$type": "sh.tangled.feed.comment",
916 "subject": { "uri": issue_uri.as_ref(), "cid": "bafkqaaa" },
917 "body": { "$type": "sh.tangled.markup.markdown", "text": "thoughts" },
918 "createdAt": "2026-05-01T00:00:00Z"
919 }),
920 )
921 .await;
922
923 let app = router(h.state.clone());
924 let (status, body) = json_response(
925 app.oneshot(list_request(
926 "sh.tangled.feed.listComments",
927 issue_uri.as_ref(),
928 &[],
929 ))
930 .await
931 .unwrap(),
932 )
933 .await;
934 assert_eq!(status, StatusCode::OK);
935 let items = body["items"].as_array().unwrap();
936 assert_eq!(items.len(), 1);
937 assert_eq!(items[0]["value"]["body"]["text"], json!("thoughts"));
938 assert_eq!(
939 items[0]["value"]["subject"]["uri"],
940 json!(issue_uri.as_ref())
941 );
942}
943
944#[tokio::test]
945async fn list_item_cid_is_present() {
946 let h = Harness::new().await;
947 let subject = at("at://did:plc:abalone");
948 let nel = did("did:plc:nel");
949 h.add_edge(
950 &nsid("sh.tangled.feed.star"),
951 &subject,
952 &at(&format!("at://{}/sh.tangled.feed.star/s1", nel.as_ref())),
953 );
954 h.mount(
955 &nel,
956 &nsid("sh.tangled.feed.star"),
957 &rkey("s1"),
958 star_body(&did("did:plc:abalone")),
959 )
960 .await;
961
962 let app = router(h.state.clone());
963 let (_, body) = json_response(
964 app.oneshot(list_request(
965 "sh.tangled.feed.listStars",
966 subject.as_ref(),
967 &[],
968 ))
969 .await
970 .unwrap(),
971 )
972 .await;
973 let item = &body["items"][0];
974 assert!(
975 item.as_object().unwrap().contains_key("cid"),
976 "list items must mirror getRecord output shape and include cid"
977 );
978 assert_eq!(item["cid"], json!(CID));
979}
980
981#[tokio::test]
982async fn count_feed_comments_subjects_on_issue_uri() {
983 let h = Harness::new().await;
984 let issue_uri = at("at://did:plc:abalone/sh.tangled.repo.issue/i1");
985 h.add_edge(
986 &nsid("sh.tangled.feed.comment"),
987 &issue_uri,
988 &at("at://did:plc:nel/sh.tangled.feed.comment/c1"),
989 );
990 h.add_edge(
991 &nsid("sh.tangled.feed.comment"),
992 &issue_uri,
993 &at("at://did:plc:olaren/sh.tangled.feed.comment/c2"),
994 );
995
996 let app = router(h.state.clone());
997 let (status, body) = json_response(
998 app.oneshot(list_request(
999 "sh.tangled.feed.countComments",
1000 issue_uri.as_ref(),
1001 &[],
1002 ))
1003 .await
1004 .unwrap(),
1005 )
1006 .await;
1007 assert_eq!(status, StatusCode::OK);
1008 assert_eq!(body["count"], json!(2));
1009 assert_eq!(body["distinctAuthors"], json!(2));
1010}
1011
1012#[tokio::test]
1013async fn list_item_404_dropped_not_404_for_subject() {
1014 let h = Harness::new().await;
1015 let subject = at("at://did:plc:squid");
1016 let kind = nsid("sh.tangled.repo.issue");
1017 let repo = did("did:plc:squid");
1018 h.add_edge(
1019 &kind,
1020 &subject,
1021 &at("at://did:plc:nel/sh.tangled.repo.issue/live"),
1022 );
1023 h.add_edge(
1024 &kind,
1025 &subject,
1026 &at("at://did:plc:teq/sh.tangled.repo.issue/missing"),
1027 );
1028 h.mount(
1029 &did("did:plc:nel"),
1030 &kind,
1031 &rkey("live"),
1032 issue_body(&repo, "kelp survives"),
1033 )
1034 .await;
1035 Mock::given(method("GET"))
1036 .and(path("/xrpc/com.atproto.repo.getRecord"))
1037 .and(query_param("repo", "did:plc:teq"))
1038 .and(query_param("collection", "sh.tangled.repo.issue"))
1039 .and(query_param("rkey", "missing"))
1040 .respond_with(ResponseTemplate::new(404).set_body_json(json!({
1041 "error": "RecordNotFound",
1042 "message": "could not find record"
1043 })))
1044 .mount(&h.server)
1045 .await;
1046
1047 let app = router(h.state.clone());
1048 let (status, body) = json_response(
1049 app.oneshot(list_request(
1050 "sh.tangled.repo.listIssues",
1051 subject.as_ref(),
1052 &[],
1053 ))
1054 .await
1055 .unwrap(),
1056 )
1057 .await;
1058 assert_eq!(
1059 status,
1060 StatusCode::OK,
1061 "a stale-index 404 drops that item, it must not 404 or 502 the subject's list",
1062 );
1063 let items = body["items"].as_array().expect("items array");
1064 assert_eq!(items.len(), 1, "stale 404 item dropped, live sibling kept");
1065 assert_eq!(
1066 items[0]["uri"].as_str().unwrap(),
1067 "at://did:plc:nel/sh.tangled.repo.issue/live",
1068 );
1069}
1070
1071#[tokio::test]
1072async fn list_item_with_wrong_type_tag_dropped() {
1073 let h = Harness::new().await;
1074 let subject = at("at://did:plc:squid");
1075 let kind = nsid("sh.tangled.feed.star");
1076 h.add_edge(
1077 &kind,
1078 &subject,
1079 &at("at://did:plc:nel/sh.tangled.feed.star/good"),
1080 );
1081 h.add_edge(
1082 &kind,
1083 &subject,
1084 &at("at://did:plc:teq/sh.tangled.feed.star/wrong"),
1085 );
1086 h.mount(
1087 &did("did:plc:nel"),
1088 &kind,
1089 &rkey("good"),
1090 star_body(&did("did:plc:squid")),
1091 )
1092 .await;
1093 Mock::given(method("GET"))
1094 .and(path("/xrpc/com.atproto.repo.getRecord"))
1095 .and(query_param("repo", "did:plc:teq"))
1096 .and(query_param("collection", "sh.tangled.feed.star"))
1097 .and(query_param("rkey", "wrong"))
1098 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
1099 "uri": "at://did:plc:teq/sh.tangled.feed.star/wrong",
1100 "cid": CID,
1101 "value": {
1102 "$type": "sh.tangled.feed.reaction",
1103 "createdAt": "2026-05-01T00:00:00Z",
1104 "subject": "at://did:plc:squid"
1105 }
1106 })))
1107 .mount(&h.server)
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.listStars",
1114 subject.as_ref(),
1115 &[],
1116 ))
1117 .await
1118 .unwrap(),
1119 )
1120 .await;
1121 assert_eq!(status, StatusCode::OK);
1122 let items = body["items"].as_array().expect("items array");
1123 assert_eq!(items.len(), 1, "wrong-type item dropped, valid star kept");
1124 assert_eq!(
1125 items[0]["uri"].as_str().unwrap(),
1126 "at://did:plc:nel/sh.tangled.feed.star/good",
1127 );
1128}
1129
1130#[tokio::test]
1131async fn list_item_with_mismatched_collection_dropped() {
1132 let h = Harness::new().await;
1133 let subject = at("at://did:plc:squid");
1134 let kind = nsid("sh.tangled.repo.issue");
1135 let repo = did("did:plc:squid");
1136 h.add_edge(
1137 &kind,
1138 &subject,
1139 &at("at://did:plc:nel/sh.tangled.repo.issue/live"),
1140 );
1141 h.add_edge(
1142 &kind,
1143 &subject,
1144 &at("at://did:plc:teq/sh.tangled.feed.star/whelk"),
1145 );
1146 h.mount(
1147 &did("did:plc:nel"),
1148 &kind,
1149 &rkey("live"),
1150 issue_body(&repo, "kelp survives"),
1151 )
1152 .await;
1153
1154 let app = router(h.state.clone());
1155 let (status, body) = json_response(
1156 app.oneshot(list_request(
1157 "sh.tangled.repo.listIssues",
1158 subject.as_ref(),
1159 &[],
1160 ))
1161 .await
1162 .unwrap(),
1163 )
1164 .await;
1165 assert_eq!(
1166 status,
1167 StatusCode::OK,
1168 "a mismatched-collection index edge must not 400 the subject's list",
1169 );
1170 let items = body["items"].as_array().expect("items array");
1171 assert_eq!(
1172 items.len(),
1173 1,
1174 "mismatched-collection edge dropped, live sibling kept"
1175 );
1176 assert_eq!(
1177 items[0]["uri"].as_str().unwrap(),
1178 "at://did:plc:nel/sh.tangled.repo.issue/live",
1179 );
1180}
1181
1182#[tokio::test]
1183async fn bare_did_endpoints_reject_at_uri_subject() {
1184 let h = Harness::new().await;
1185 let app = router(h.state.clone());
1186 let cases = [
1187 "sh.tangled.graph.listFollows",
1188 "sh.tangled.graph.countFollows",
1189 ];
1190 stream::iter(cases)
1191 .for_each(|endpoint| {
1192 let app = app.clone();
1193 async move {
1194 let resp = app
1195 .oneshot(list_request(
1196 endpoint,
1197 "at://did:plc:abalone/sh.tangled.repo/r1",
1198 &[],
1199 ))
1200 .await
1201 .unwrap();
1202 let (status, body) = json_response(resp).await;
1203 assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint}");
1204 assert_eq!(body["error"], "InvalidRequest", "{endpoint}");
1205 assert!(
1206 body["message"]
1207 .as_str()
1208 .unwrap_or_default()
1209 .contains("bare did"),
1210 "{endpoint}: {}",
1211 body["message"],
1212 );
1213 }
1214 })
1215 .await;
1216}
1217
1218#[tokio::test]
1219async fn repo_pointing_endpoints_reject_at_uri_subject() {
1220 let h = Harness::new().await;
1221 let app = router(h.state.clone());
1222 let cases = [
1223 "sh.tangled.repo.listIssues",
1224 "sh.tangled.repo.countIssues",
1225 "sh.tangled.repo.listPulls",
1226 "sh.tangled.repo.countPulls",
1227 "sh.tangled.repo.listArtifacts",
1228 "sh.tangled.repo.countArtifacts",
1229 ];
1230 stream::iter(cases)
1231 .for_each(|endpoint| {
1232 let app = app.clone();
1233 async move {
1234 let resp = app
1235 .oneshot(list_request(
1236 endpoint,
1237 "at://did:plc:abalone/sh.tangled.repo/r1",
1238 &[],
1239 ))
1240 .await
1241 .unwrap();
1242 let (status, body) = json_response(resp).await;
1243 assert_eq!(
1244 status,
1245 StatusCode::BAD_REQUEST,
1246 "{endpoint} must reject rkey-form subjects since rkeys are unstable; clients must send the repoDID",
1247 );
1248 assert!(
1249 body["message"]
1250 .as_str()
1251 .unwrap_or_default()
1252 .contains("bare did"),
1253 "{endpoint}: {}",
1254 body["message"],
1255 );
1256 }
1257 })
1258 .await;
1259}
1260
1261#[tokio::test]
1262async fn repo_pointing_endpoints_accept_bare_did() {
1263 let h = Harness::new().await;
1264 let app = router(h.state.clone());
1265 let cases = [
1266 "sh.tangled.repo.listIssues",
1267 "sh.tangled.repo.countIssues",
1268 "sh.tangled.repo.listPulls",
1269 "sh.tangled.repo.countPulls",
1270 "sh.tangled.repo.listArtifacts",
1271 "sh.tangled.repo.countArtifacts",
1272 ];
1273 stream::iter(cases)
1274 .for_each(|endpoint| {
1275 let app = app.clone();
1276 async move {
1277 let resp = app
1278 .oneshot(list_request(endpoint, "did:plc:abalone", &[]))
1279 .await
1280 .unwrap();
1281 let (status, _body) = json_response(resp).await;
1282 assert_eq!(status, StatusCode::OK, "{endpoint} must accept bare did");
1283 }
1284 })
1285 .await;
1286}
1287
1288#[tokio::test]
1289async fn feed_comment_endpoints_reject_bare_did_or_wrong_collection() {
1290 let h = Harness::new().await;
1291 let app = router(h.state.clone());
1292 let endpoints = [
1293 "sh.tangled.feed.listComments",
1294 "sh.tangled.feed.countComments",
1295 ];
1296 let inputs = [
1297 "at://did:plc:abalone",
1298 "at://did:plc:abalone/sh.tangled.repo/r1",
1299 ];
1300 let cases = endpoints
1301 .iter()
1302 .copied()
1303 .flat_map(|endpoint| inputs.iter().copied().map(move |input| (endpoint, input)));
1304 stream::iter(cases)
1305 .for_each(|(endpoint, input)| {
1306 let app = app.clone();
1307 async move {
1308 let resp = app
1309 .oneshot(list_request(endpoint, input, &[]))
1310 .await
1311 .unwrap();
1312 let (status, body) = json_response(resp).await;
1313 assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint} input={input}");
1314 let msg = body["message"].as_str().unwrap_or_default();
1315 assert!(
1316 msg.contains("sh.tangled.repo.issue") && msg.contains("sh.tangled.repo.pull") && msg.contains("sh.tangled.string"),
1317 "{endpoint} input={input}: {msg}",
1318 );
1319 }
1320 })
1321 .await;
1322}
1323
1324#[tokio::test]
1325async fn star_endpoints_reject_unrelated_collection() {
1326 let h = Harness::new().await;
1327 let app = router(h.state.clone());
1328 let endpoints = ["sh.tangled.feed.listStars", "sh.tangled.feed.countStars"];
1329 stream::iter(endpoints)
1330 .for_each(|endpoint| {
1331 let app = app.clone();
1332 async move {
1333 let resp = app
1334 .oneshot(list_request(
1335 endpoint,
1336 "at://did:plc:abalone/sh.tangled.knot/k1",
1337 &[],
1338 ))
1339 .await
1340 .unwrap();
1341 let (status, body) = json_response(resp).await;
1342 assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint}");
1343 let msg = body["message"].as_str().unwrap_or_default();
1344 assert!(msg.contains("sh.tangled.string"), "{endpoint}: {msg}",);
1345 }
1346 })
1347 .await;
1348}
1349
1350#[tokio::test]
1351async fn star_endpoints_reject_repo_uri_subject() {
1352 let h = Harness::new().await;
1353 let app = router(h.state.clone());
1354 let resp = app
1355 .oneshot(list_request(
1356 "sh.tangled.feed.countStars",
1357 "at://did:plc:abalone/sh.tangled.repo/r1",
1358 &[],
1359 ))
1360 .await
1361 .unwrap();
1362 let (status, body) = json_response(resp).await;
1363 assert_eq!(
1364 status,
1365 StatusCode::BAD_REQUEST,
1366 "rkey-form repo URI must be rejected; clients must send the repoDID directly",
1367 );
1368 let msg = body["message"].as_str().unwrap_or_default();
1369 assert!(msg.contains("sh.tangled.string"), "{msg}");
1370}
1371
1372#[tokio::test]
1373async fn star_endpoints_accept_string_subject_form() {
1374 let h = Harness::new().await;
1375 let app = router(h.state.clone());
1376 let resp = app
1377 .oneshot(list_request(
1378 "sh.tangled.feed.countStars",
1379 "at://did:plc:abalone/sh.tangled.string/k1",
1380 &[],
1381 ))
1382 .await
1383 .unwrap();
1384 let (status, body) = json_response(resp).await;
1385 assert_eq!(status, StatusCode::OK);
1386 assert_eq!(body["count"], json!(0));
1387}
1388
1389#[tokio::test]
1390async fn list_after_remove_source_returns_empty_items() {
1391 let h = Harness::new().await;
1392 let subject = at("at://did:plc:abalone");
1393 let source = at("at://did:plc:nel/sh.tangled.feed.star/s1");
1394 h.add_edge(&nsid("sh.tangled.feed.star"), &subject, &source);
1395 h.edges.remove_source(&source);
1396
1397 let app = router(h.state.clone());
1398 let (status, body) = json_response(
1399 app.oneshot(list_request(
1400 "sh.tangled.feed.listStars",
1401 subject.as_ref(),
1402 &[],
1403 ))
1404 .await
1405 .unwrap(),
1406 )
1407 .await;
1408 assert_eq!(status, StatusCode::OK);
1409 assert_eq!(body["items"], json!([]));
1410 assert!(body["cursor"].is_null());
1411}
1412
1413#[tokio::test]
1414async fn list_pulls_hydrates_via_slingshot_when_edges_present() {
1415 let h = Harness::new().await;
1416 let target_did = did("did:plc:abalone");
1417 let subject = at(&format!("at://{}", target_did.as_ref()));
1418 let source_did = did("did:plc:nel");
1419 let rk = rkey("p1");
1420 h.add_edge(
1421 &nsid("sh.tangled.repo.pull"),
1422 &subject,
1423 &at(&format!(
1424 "at://{}/sh.tangled.repo.pull/{}",
1425 source_did.as_ref(),
1426 rk.as_ref()
1427 )),
1428 );
1429 h.mount(
1430 &source_did,
1431 &nsid("sh.tangled.repo.pull"),
1432 &rk,
1433 json!({
1434 "$type": "sh.tangled.repo.pull",
1435 "title": "ship it",
1436 "createdAt": "2026-05-01T00:00:00Z",
1437 "rounds": [],
1438 "target": {"repo": target_did.as_ref(), "branch": "main"},
1439 }),
1440 )
1441 .await;
1442 let app = router(h.state.clone());
1443 let (status, body) = json_response(
1444 app.oneshot(list_request(
1445 "sh.tangled.repo.listPulls",
1446 subject.as_ref(),
1447 &[],
1448 ))
1449 .await
1450 .unwrap(),
1451 )
1452 .await;
1453 assert_eq!(status, StatusCode::OK);
1454 let items = body["items"].as_array().unwrap();
1455 assert_eq!(items.len(), 1);
1456 assert_eq!(items[0]["value"]["title"], json!("ship it"));
1457 assert_eq!(
1458 items[0]["value"]["target"]["repo"],
1459 json!(target_did.as_ref())
1460 );
1461}
1462
1463#[tokio::test]
1464async fn count_pulls_returns_distinct_authors() {
1465 let h = Harness::new().await;
1466 let subject = at("at://did:plc:abalone");
1467 h.add_edge(
1468 &nsid("sh.tangled.repo.pull"),
1469 &subject,
1470 &at("at://did:plc:nel/sh.tangled.repo.pull/p1"),
1471 );
1472 h.add_edge(
1473 &nsid("sh.tangled.repo.pull"),
1474 &subject,
1475 &at("at://did:plc:olaren/sh.tangled.repo.pull/p2"),
1476 );
1477 h.add_edge(
1478 &nsid("sh.tangled.repo.pull"),
1479 &subject,
1480 &at("at://did:plc:nel/sh.tangled.repo.pull/p3"),
1481 );
1482 let app = router(h.state.clone());
1483 let (_, body) = json_response(
1484 app.oneshot(list_request(
1485 "sh.tangled.repo.countPulls",
1486 subject.as_ref(),
1487 &[],
1488 ))
1489 .await
1490 .unwrap(),
1491 )
1492 .await;
1493 assert_eq!(body["count"], json!(3));
1494 assert_eq!(body["distinctAuthors"], json!(2));
1495}
1496
1497#[tokio::test]
1498async fn extractor_to_xrpc_round_trip_for_star() {
1499 let h = Harness::new().await;
1500 let subject_did = did("did:plc:abalone");
1501 let source_did = did("did:plc:nel");
1502 let rk = rkey("s1");
1503 let source = at(&format!(
1504 "at://{}/sh.tangled.feed.star/{}",
1505 source_did.as_ref(),
1506 rk.as_ref()
1507 ));
1508 let body = star_body(&subject_did);
1509 let parsed =
1510 bobbin_types::edges::Record::from_json_value(&nsid("sh.tangled.feed.star"), body.clone())
1511 .expect("parse star record");
1512 parsed
1513 .extract_edges(&source)
1514 .expect("extract")
1515 .into_iter()
1516 .for_each(|e| h.edges.add(e));
1517 h.mount(&source_did, &nsid("sh.tangled.feed.star"), &rk, body)
1518 .await;
1519
1520 let app = router(h.state.clone());
1521 let (status, json) = json_response(
1522 app.oneshot(list_request(
1523 "sh.tangled.feed.listStars",
1524 &format!("at://{}", subject_did.as_ref()),
1525 &[],
1526 ))
1527 .await
1528 .unwrap(),
1529 )
1530 .await;
1531 assert_eq!(
1532 status,
1533 StatusCode::OK,
1534 "extractor key must match handler subject, body was {json}",
1535 );
1536 let items = json["items"].as_array().unwrap();
1537 assert_eq!(items.len(), 1, "expected exactly one star edge");
1538 assert_eq!(
1539 items[0]["value"]["subject"]["did"],
1540 json!(subject_did.as_ref())
1541 );
1542}
1543
1544#[tokio::test]
1545async fn list_issues_includes_state_comment_count_and_state_updated_at() {
1546 let h = Harness::new().await;
1547 let repo = did("did:plc:limpet");
1548 let subject = at(&format!("at://{}", repo.as_ref()));
1549 let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1");
1550 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri);
1551 h.mount(
1552 &did("did:plc:nel"),
1553 &nsid("sh.tangled.repo.issue"),
1554 &rkey("i1"),
1555 issue_body(&repo, "hi"),
1556 )
1557 .await;
1558 h.add_edge(
1559 &nsid("sh.tangled.feed.comment"),
1560 &issue_uri,
1561 &at("at://did:plc:olaren/sh.tangled.feed.comment/c1"),
1562 );
1563 h.add_edge(
1564 &nsid("sh.tangled.feed.comment"),
1565 &issue_uri,
1566 &at("at://did:plc:teq/sh.tangled.feed.comment/c2"),
1567 );
1568
1569 h.state.issue_states.upsert(
1570 at("at://did:plc:nel/sh.tangled.repo.issue.state/s1"),
1571 issue_uri.clone(),
1572 1_777_593_600_000_000,
1573 IssueStateKind::Open,
1574 );
1575 h.state.issue_states.upsert(
1576 at("at://did:plc:nel/sh.tangled.repo.issue.state/s2"),
1577 issue_uri.clone(),
1578 1_777_593_700_000_000,
1579 IssueStateKind::Closed,
1580 );
1581
1582 let app = router(h.state.clone());
1583 let resp = app
1584 .oneshot(list_request(
1585 "sh.tangled.repo.listIssues",
1586 subject.as_ref(),
1587 &[],
1588 ))
1589 .await
1590 .unwrap();
1591 let (status, body) = json_response(resp).await;
1592 assert_eq!(status, StatusCode::OK);
1593 let item = &body["items"][0];
1594 assert_eq!(item["state"], json!("closed"));
1595 assert_eq!(item["commentCount"], json!(2));
1596 let updated = item["stateUpdatedAt"]
1597 .as_str()
1598 .expect("stateUpdatedAt must serialize as RFC3339 string");
1599 assert!(
1600 updated.starts_with("2026-"),
1601 "expected 2026 timestamp, got {updated}"
1602 );
1603}
1604
1605#[tokio::test]
1606async fn list_issues_defaults_to_open_when_no_state_record() {
1607 let h = Harness::new().await;
1608 let repo = did("did:plc:limpet");
1609 let subject = at(&format!("at://{}", repo.as_ref()));
1610 let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1");
1611 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri);
1612 h.mount(
1613 &did("did:plc:nel"),
1614 &nsid("sh.tangled.repo.issue"),
1615 &rkey("i1"),
1616 issue_body(&repo, "no state yet"),
1617 )
1618 .await;
1619
1620 let app = router(h.state.clone());
1621 let (_status, body) = json_response(
1622 app.oneshot(list_request(
1623 "sh.tangled.repo.listIssues",
1624 subject.as_ref(),
1625 &[],
1626 ))
1627 .await
1628 .unwrap(),
1629 )
1630 .await;
1631 let item = &body["items"][0];
1632 assert_eq!(
1633 item["state"],
1634 json!("open"),
1635 "absent state record defaults to open"
1636 );
1637 assert!(
1638 item.get("stateUpdatedAt").is_none(),
1639 "stateUpdatedAt must be absent without a state record",
1640 );
1641 assert_eq!(item["commentCount"], json!(0));
1642}
1643
1644#[tokio::test]
1645async fn list_issues_author_filter_restricts_to_matching_did() {
1646 let h = Harness::new().await;
1647 let repo = did("did:plc:limpet");
1648 let subject = at(&format!("at://{}", repo.as_ref()));
1649 let owners = [
1650 ("did:plc:nel", "n1"),
1651 ("did:plc:nel", "n2"),
1652 ("did:plc:olaren", "o1"),
1653 ("did:plc:olaren", "o2"),
1654 ];
1655 stream::iter(owners)
1656 .for_each(|(d, r)| {
1657 let h = &h;
1658 let subject = subject.clone();
1659 let repo = repo.clone();
1660 async move {
1661 let d_did = did(d);
1662 let rk = rkey(r);
1663 h.add_edge(
1664 &nsid("sh.tangled.repo.issue"),
1665 &subject,
1666 &at(&format!(
1667 "at://{}/sh.tangled.repo.issue/{}",
1668 d_did.as_ref(),
1669 rk.as_ref()
1670 )),
1671 );
1672 h.mount(
1673 &d_did,
1674 &nsid("sh.tangled.repo.issue"),
1675 &rk,
1676 issue_body(&repo, &format!("issue-{}", rk.as_ref())),
1677 )
1678 .await;
1679 }
1680 })
1681 .await;
1682
1683 let app = router(h.state.clone());
1684 let (status, body) = json_response(
1685 app.oneshot(list_request(
1686 "sh.tangled.repo.listIssues",
1687 subject.as_ref(),
1688 &[("author", "did:plc:nel")],
1689 ))
1690 .await
1691 .unwrap(),
1692 )
1693 .await;
1694 assert_eq!(status, StatusCode::OK);
1695 let items = body["items"].as_array().expect("items array");
1696 assert_eq!(items.len(), 2, "two issues authored by nel");
1697 let all_nel = items
1698 .iter()
1699 .all(|i| i["uri"].as_str().unwrap().starts_with("at://did:plc:nel/"));
1700 assert!(all_nel, "every returned uri must be authored by nel");
1701}
1702
1703#[tokio::test]
1704async fn list_issues_invalid_author_returns_400() {
1705 let h = Harness::new().await;
1706 let subject = "at://did:plc:limpet".to_owned();
1707 let app = router(h.state.clone());
1708 let resp = app
1709 .oneshot(list_request(
1710 "sh.tangled.repo.listIssues",
1711 &subject,
1712 &[("author", "not-a-did")],
1713 ))
1714 .await
1715 .unwrap();
1716 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
1717}
1718
1719#[tokio::test]
1720async fn list_pulls_includes_merged_state_and_comment_count() {
1721 let h = Harness::new().await;
1722 let repo = did("did:plc:limpet");
1723 let subject = at(&format!("at://{}", repo.as_ref()));
1724 let pull_uri = at("at://did:plc:nel/sh.tangled.repo.pull/p1");
1725 h.add_edge(&nsid("sh.tangled.repo.pull"), &subject, &pull_uri);
1726 h.mount(
1727 &did("did:plc:nel"),
1728 &nsid("sh.tangled.repo.pull"),
1729 &rkey("p1"),
1730 pull_body(&repo, "fix bug"),
1731 )
1732 .await;
1733 h.add_edge(
1734 &nsid("sh.tangled.feed.comment"),
1735 &pull_uri,
1736 &at("at://did:plc:teq/sh.tangled.feed.comment/c1"),
1737 );
1738 h.state.pull_statuses.upsert(
1739 at("at://did:plc:nel/sh.tangled.repo.pull.status/s1"),
1740 pull_uri.clone(),
1741 1_777_593_600_000_000,
1742 PullStatusKind::Open,
1743 );
1744 h.state.pull_statuses.upsert(
1745 at("at://did:plc:nel/sh.tangled.repo.pull.status/s2"),
1746 pull_uri.clone(),
1747 1_777_593_800_000_000,
1748 PullStatusKind::Merged,
1749 );
1750
1751 let app = router(h.state.clone());
1752 let (status, body) = json_response(
1753 app.oneshot(list_request(
1754 "sh.tangled.repo.listPulls",
1755 subject.as_ref(),
1756 &[],
1757 ))
1758 .await
1759 .unwrap(),
1760 )
1761 .await;
1762 assert_eq!(status, StatusCode::OK);
1763 let item = &body["items"][0];
1764 assert_eq!(item["state"], json!("merged"));
1765 assert_eq!(item["commentCount"], json!(1));
1766}
1767
1768#[tokio::test]
1769async fn list_issues_state_filter_open_includes_records_without_state() {
1770 let h = Harness::new().await;
1771 let repo = did("did:plc:limpet");
1772 let subject = at(&format!("at://{}", repo.as_ref()));
1773 let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1");
1774 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri);
1775 h.mount(
1776 &did("did:plc:nel"),
1777 &nsid("sh.tangled.repo.issue"),
1778 &rkey("i1"),
1779 issue_body(&repo, "fresh"),
1780 )
1781 .await;
1782
1783 let app = router(h.state.clone());
1784 let (status, body) = json_response(
1785 app.oneshot(list_request(
1786 "sh.tangled.repo.listIssues",
1787 subject.as_ref(),
1788 &[("state", "open")],
1789 ))
1790 .await
1791 .unwrap(),
1792 )
1793 .await;
1794 assert_eq!(status, StatusCode::OK);
1795 let items = body["items"].as_array().expect("items array");
1796 assert_eq!(
1797 items.len(),
1798 1,
1799 "absent state record still matches state=open"
1800 );
1801}
1802
1803#[tokio::test]
1804async fn list_issues_state_filter_ignores_third_party_state_source() {
1805 let h = Harness::new().await;
1806 let repo = did("did:plc:limpet");
1807 let subject = at(&format!("at://{}", repo.as_ref()));
1808 let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1");
1809 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri);
1810 h.mount(
1811 &did("did:plc:nel"),
1812 &nsid("sh.tangled.repo.issue"),
1813 &rkey("i1"),
1814 issue_body(&repo, "open issue"),
1815 )
1816 .await;
1817 h.state.issue_states.upsert(
1818 at("at://did:plc:nautilus/sh.tangled.repo.issue.state/spoof"),
1819 issue_uri.clone(),
1820 1_777_593_800_000_000,
1821 IssueStateKind::Closed,
1822 );
1823
1824 let app = router(h.state.clone());
1825 let (status, body) = json_response(
1826 app.oneshot(list_request(
1827 "sh.tangled.repo.listIssues",
1828 subject.as_ref(),
1829 &[("state", "open")],
1830 ))
1831 .await
1832 .unwrap(),
1833 )
1834 .await;
1835 assert_eq!(status, StatusCode::OK);
1836 let items = body["items"].as_array().expect("items array");
1837 assert_eq!(
1838 items.len(),
1839 1,
1840 "third-party Closed record must not flip filter result for state=open",
1841 );
1842 assert_eq!(items[0]["state"], json!("open"));
1843 assert!(
1844 items[0].get("stateUpdatedAt").is_none(),
1845 "third-party state source must not surface stateUpdatedAt",
1846 );
1847}
1848
1849#[tokio::test]
1850async fn list_pulls_status_filter_ignores_third_party_status_source() {
1851 let h = Harness::new().await;
1852 let repo = did("did:plc:limpet");
1853 let subject = at(&format!("at://{}", repo.as_ref()));
1854 let pull_uri = at("at://did:plc:nel/sh.tangled.repo.pull/p1");
1855 h.add_edge(&nsid("sh.tangled.repo.pull"), &subject, &pull_uri);
1856 h.mount(
1857 &did("did:plc:nel"),
1858 &nsid("sh.tangled.repo.pull"),
1859 &rkey("p1"),
1860 pull_body(&repo, "wip"),
1861 )
1862 .await;
1863 h.state.pull_statuses.upsert(
1864 at("at://did:plc:nautilus/sh.tangled.repo.pull.status/spoof"),
1865 pull_uri.clone(),
1866 1_777_593_800_000_000,
1867 PullStatusKind::Merged,
1868 );
1869
1870 let app = router(h.state.clone());
1871 let (status, body) = json_response(
1872 app.oneshot(list_request(
1873 "sh.tangled.repo.listPulls",
1874 subject.as_ref(),
1875 &[("status", "merged")],
1876 ))
1877 .await
1878 .unwrap(),
1879 )
1880 .await;
1881 assert_eq!(status, StatusCode::OK);
1882 let items = body["items"].as_array().expect("items array");
1883 assert_eq!(
1884 items.len(),
1885 0,
1886 "third-party Merged record must not satisfy status=merged"
1887 );
1888}
1889
1890#[tokio::test]
1891async fn list_issues_state_filter_accepts_repo_owner_state_source() {
1892 let h = Harness::new().await;
1893 let repo_owner = did("did:plc:limpet");
1894 let subject = at(&format!("at://{}", repo_owner.as_ref()));
1895 let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1");
1896 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri);
1897 h.mount(
1898 &did("did:plc:nel"),
1899 &nsid("sh.tangled.repo.issue"),
1900 &rkey("i1"),
1901 issue_body(&repo_owner, "owner closed"),
1902 )
1903 .await;
1904 h.state.issue_states.upsert(
1905 at("at://did:plc:limpet/sh.tangled.repo.issue.state/legit"),
1906 issue_uri.clone(),
1907 1_777_593_800_000_000,
1908 IssueStateKind::Closed,
1909 );
1910
1911 let app = router(h.state.clone());
1912 let (status, body) = json_response(
1913 app.oneshot(list_request(
1914 "sh.tangled.repo.listIssues",
1915 subject.as_ref(),
1916 &[("state", "closed")],
1917 ))
1918 .await
1919 .unwrap(),
1920 )
1921 .await;
1922 assert_eq!(status, StatusCode::OK);
1923 let items = body["items"].as_array().expect("items array");
1924 assert_eq!(
1925 items.len(),
1926 1,
1927 "repo-owner state record must satisfy state=closed"
1928 );
1929 assert_eq!(items[0]["state"], json!("closed"));
1930}
1931
1932#[tokio::test]
1933async fn list_issues_order_asc_returns_oldest_first() {
1934 let h = Harness::new().await;
1935 let repo = did("did:plc:limpet");
1936 let subject = at(&format!("at://{}", repo.as_ref()));
1937 let rkeys = ["a", "b", "c"];
1938 stream::iter(rkeys)
1939 .for_each(|r| {
1940 let h = &h;
1941 let subject = subject.clone();
1942 let repo = repo.clone();
1943 async move {
1944 let rk = rkey(r);
1945 let issue_uri = at(&format!(
1946 "at://did:plc:nel/sh.tangled.repo.issue/{}",
1947 rk.as_ref()
1948 ));
1949 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri);
1950 h.mount(
1951 &did("did:plc:nel"),
1952 &nsid("sh.tangled.repo.issue"),
1953 &rk,
1954 issue_body(&repo, &format!("issue-{}", rk.as_ref())),
1955 )
1956 .await;
1957 }
1958 })
1959 .await;
1960
1961 let app = router(h.state.clone());
1962 let (_, asc) = json_response(
1963 app.clone()
1964 .oneshot(list_request(
1965 "sh.tangled.repo.listIssues",
1966 subject.as_ref(),
1967 &[("order", "asc")],
1968 ))
1969 .await
1970 .unwrap(),
1971 )
1972 .await;
1973 let (_, desc) = json_response(
1974 app.oneshot(list_request(
1975 "sh.tangled.repo.listIssues",
1976 subject.as_ref(),
1977 &[("order", "desc")],
1978 ))
1979 .await
1980 .unwrap(),
1981 )
1982 .await;
1983 let asc_uris: Vec<_> = asc["items"]
1984 .as_array()
1985 .unwrap()
1986 .iter()
1987 .map(|i| i["uri"].as_str().unwrap().to_owned())
1988 .collect();
1989 let desc_uris: Vec<_> = desc["items"]
1990 .as_array()
1991 .unwrap()
1992 .iter()
1993 .map(|i| i["uri"].as_str().unwrap().to_owned())
1994 .collect();
1995 let mut reversed = asc_uris.clone();
1996 reversed.reverse();
1997 assert_eq!(asc_uris.len(), 3);
1998 assert_eq!(desc_uris, reversed, "desc must be exact reverse of asc");
1999}
2000
2001#[tokio::test]
2002async fn list_issues_by_state_filter_narrows_results() {
2003 let h = Harness::new().await;
2004 let author = did("did:plc:nel");
2005 let repo = did("did:plc:limpet");
2006 let open_uri = at("at://did:plc:nel/sh.tangled.repo.issue/open1");
2007 let closed_uri = at("at://did:plc:nel/sh.tangled.repo.issue/closed1");
2008 let author_subject = at(&format!("at://{}", author.as_ref()));
2009 h.edges.add(Edge {
2010 kind: nsid("sh.tangled.repo.issue.by"),
2011 subject: SubjectRef::Did(author.clone()),
2012 source: open_uri.clone(),
2013 sort_micros: next_sort_micros(),
2014 });
2015 h.edges.add(Edge {
2016 kind: nsid("sh.tangled.repo.issue.by"),
2017 subject: SubjectRef::Did(author.clone()),
2018 source: closed_uri.clone(),
2019 sort_micros: next_sort_micros(),
2020 });
2021 h.mount(
2022 &author,
2023 &nsid("sh.tangled.repo.issue"),
2024 &rkey("open1"),
2025 issue_body(&repo, "still open"),
2026 )
2027 .await;
2028 h.mount(
2029 &author,
2030 &nsid("sh.tangled.repo.issue"),
2031 &rkey("closed1"),
2032 issue_body(&repo, "shut"),
2033 )
2034 .await;
2035 h.state.issue_states.upsert(
2036 at("at://did:plc:nel/sh.tangled.repo.issue.state/s1"),
2037 closed_uri.clone(),
2038 1_777_593_800_000_000,
2039 IssueStateKind::Closed,
2040 );
2041
2042 let app = router(h.state.clone());
2043 let (status, body) = json_response(
2044 app.oneshot(list_request(
2045 "sh.tangled.repo.listIssuesBy",
2046 author_subject.as_ref(),
2047 &[("state", "closed")],
2048 ))
2049 .await
2050 .unwrap(),
2051 )
2052 .await;
2053 assert_eq!(status, StatusCode::OK);
2054 let items = body["items"].as_array().expect("items array");
2055 assert_eq!(
2056 items.len(),
2057 1,
2058 "only the closed issue survives state=closed"
2059 );
2060 assert_eq!(items[0]["uri"], json!(closed_uri.as_ref()));
2061}