tangled.org
Monorepo for Tangled
tangled.org
1use std::sync::Arc;
2
3use axum::body::{Body, to_bytes};
4use bobbin_edge_index::{CoverageWatch, EdgeStore, IssueStateKind, PullStatusKind, StateIndex};
5use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig};
6use bobbin_record_lru::{CacheCapacity, LruRecordStore};
7use bobbin_resolver::RepoIdResolver;
8use bobbin_runtime::{RuntimeHasher, SystemClock};
9use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader};
10use bobbin_slingshot_client::SlingshotClient;
11use bobbin_types::edges::Edge;
12use bobbin_types::ids::SubjectRef;
13use bobbin_xrpc::{AppState, router};
14use http::{Request, StatusCode};
15use jacquard_common::DefaultStr;
16use jacquard_common::types::did::Did;
17use jacquard_common::types::nsid::Nsid;
18use jacquard_common::types::string::AtUri;
19use serde_json::{Value, json};
20use tower::ServiceExt;
21use url::Url;
22use wiremock::matchers::{method, path, query_param};
23use wiremock::{Mock, MockServer, ResponseTemplate};
24
25const CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i";
26
27fn at(s: &str) -> AtUri<DefaultStr> {
28 AtUri::new_owned(s).unwrap()
29}
30
31fn did(s: &str) -> Did<DefaultStr> {
32 Did::new_owned(s).unwrap()
33}
34
35fn nsid(s: &'static str) -> Nsid<DefaultStr> {
36 Nsid::new_static(s).unwrap()
37}
38
39static EDGE_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
40
41fn next_sort_micros() -> u64 {
42 EDGE_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
43}
44
45struct Harness {
46 server: MockServer,
47 edges: Arc<EdgeStore>,
48 state: AppState,
49}
50
51impl Harness {
52 async fn new() -> Self {
53 let server = MockServer::start().await;
54 let edges = Arc::new(EdgeStore::new(RuntimeHasher::default()));
55 let coverage = Arc::new(CoverageWatch::new());
56 let state = AppState::new(
57 Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))),
58 SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(),
59 edges.clone(),
60 Arc::new(StateIndex::<IssueStateKind>::new(RuntimeHasher::default())),
61 Arc::new(StateIndex::<PullStatusKind>::new(RuntimeHasher::default())),
62 coverage.clone(),
63 Arc::new(
64 KnotProxy::new(
65 KnotProxyConfig::default(),
66 KnotHttpConfig::default(),
67 Arc::new(SystemClock::new()),
68 RuntimeHasher::default(),
69 )
70 .unwrap(),
71 ),
72 Arc::new(
73 SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(),
74 ) as Arc<dyn SearchReader>,
75 Arc::new(RepoIdResolver::detached(RuntimeHasher::default())),
76 Arc::new(jacquard_identity::JacquardResolver::default()),
77 );
78 Self {
79 server,
80 edges,
81 state,
82 }
83 }
84
85 fn add_edge(&self, kind: &'static str, subject: SubjectRef, source: &AtUri<DefaultStr>) {
86 self.edges.add(Edge {
87 kind: nsid(kind),
88 subject,
89 source: source.clone(),
90 sort_micros: next_sort_micros(),
91 });
92 }
93
94 async fn mount(&self, did: &Did<DefaultStr>, collection: &str, rkey: &str, value: Value) {
95 let uri = format!("at://{}/{}/{}", did.as_ref(), collection, rkey);
96 let body = json!({ "uri": uri, "cid": CID, "value": value });
97 Mock::given(method("GET"))
98 .and(path("/xrpc/com.atproto.repo.getRecord"))
99 .and(query_param("repo", did.as_ref()))
100 .and(query_param("collection", collection))
101 .and(query_param("rkey", rkey))
102 .respond_with(ResponseTemplate::new(200).set_body_json(body))
103 .mount(&self.server)
104 .await;
105 }
106}
107
108fn enrich_request(body: Value) -> Request<Body> {
109 Request::builder()
110 .method("POST")
111 .uri("/xrpc/sh.tangled.query.enrichResponse")
112 .header("content-type", "application/json")
113 .body(Body::from(serde_json::to_vec(&body).unwrap()))
114 .unwrap()
115}
116
117async fn json_response(resp: axum::response::Response) -> (StatusCode, Value) {
118 let status = resp.status();
119 let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap();
120 let parsed: Value = serde_json::from_slice(&bytes).expect("JSON body");
121 (status, parsed)
122}
123
124fn repo_body(name: &str, repo_did: &Did<DefaultStr>) -> Value {
125 json!({
126 "$type": "sh.tangled.repo",
127 "name": name,
128 "knot": "oyster.cafe",
129 "repoDid": repo_did.as_ref(),
130 "createdAt": "2026-05-01T00:00:00Z"
131 })
132}
133
134fn follow_body(subject: &Did<DefaultStr>) -> Value {
135 json!({
136 "$type": "sh.tangled.graph.follow",
137 "createdAt": "2026-05-01T00:00:00Z",
138 "subject": subject.as_ref()
139 })
140}
141
142/// one repo owned by `owner`, with `stars`/`issues` counts against its repo did
143async fn repo_fixture(h: &Harness, owner: &Did<DefaultStr>, repo_did: &Did<DefaultStr>) {
144 let repo_uri = at(&format!("at://{}/sh.tangled.repo/reef", owner.as_ref()));
145 h.add_edge("sh.tangled.repo", SubjectRef::Did(owner.clone()), &repo_uri);
146 h.mount(
147 owner,
148 "sh.tangled.repo",
149 "reef",
150 repo_body("reef", repo_did),
151 )
152 .await;
153 for (i, stargazer) in ["did:plc:a", "did:plc:b", "did:plc:a"].iter().enumerate() {
154 h.add_edge(
155 "sh.tangled.feed.star",
156 SubjectRef::Did(repo_did.clone()),
157 &at(&format!("at://{stargazer}/sh.tangled.feed.star/s{i}")),
158 );
159 }
160 h.add_edge(
161 "sh.tangled.repo.issue",
162 SubjectRef::Did(repo_did.clone()),
163 &at("at://did:plc:a/sh.tangled.repo.issue/i0"),
164 );
165}
166
167#[tokio::test]
168async fn zero_config_counts_stars_and_issues_for_repo_did() {
169 let h = Harness::new().await;
170 let owner = did("did:plc:nel");
171 let repo_did = did("did:plc:limpet");
172 repo_fixture(&h, &owner, &repo_did).await;
173
174 let app = router(h.state.clone());
175 let (status, body) = json_response(
176 app.oneshot(enrich_request(json!({
177 "xrpc": "sh.tangled.repo.listRepos",
178 "params": { "subject": owner.as_ref() },
179 "enrich": [
180 { "source": "sh.tangled.feed.star:subject", "type": "count" },
181 { "source": "sh.tangled.feed.star:subject", "type": "distinctAuthors" },
182 { "source": "sh.tangled.repo.issue:subject", "type": "count" }
183 ]
184 })))
185 .await
186 .unwrap(),
187 )
188 .await;
189
190 assert_eq!(status, StatusCode::OK, "{body}");
191 assert_eq!(body["output"]["items"].as_array().unwrap().len(), 1);
192 let stats = &body["stats"];
193 assert_eq!(
194 stats["did:plc:limpet"]["sh.tangled.feed.star:subject"]["count"],
195 json!(3)
196 );
197 assert_eq!(
198 stats["did:plc:limpet"]["sh.tangled.feed.star:subject"]["distinctAuthors"],
199 json!(2)
200 );
201 assert_eq!(
202 stats["did:plc:limpet"]["sh.tangled.repo.issue:subject"]["count"],
203 json!(1)
204 );
205 assert!(stats["at://did:plc:nel/sh.tangled.repo/reef"].is_null());
206}
207
208#[tokio::test]
209async fn follow_counts_cover_both_directions() {
210 let h = Harness::new().await;
211 let owner = did("did:plc:nel");
212 // followers, edges pointing at owner
213 for (i, fan) in ["did:plc:a", "did:plc:b"].iter().enumerate() {
214 h.add_edge(
215 "sh.tangled.graph.follow",
216 SubjectRef::Did(owner.clone()),
217 &at(&format!("at://{fan}/sh.tangled.graph.follow/f{i}")),
218 );
219 h.mount(
220 &did(fan),
221 "sh.tangled.graph.follow",
222 &format!("f{i}"),
223 follow_body(&owner),
224 )
225 .await;
226 }
227 // following, via the .by mirror edge since owner is the author here
228 h.add_edge(
229 "sh.tangled.graph.follow.by",
230 SubjectRef::Did(owner.clone()),
231 &at("at://did:plc:nel/sh.tangled.graph.follow/f0"),
232 );
233
234 let app = router(h.state.clone());
235 let (status, body) = json_response(
236 app.oneshot(enrich_request(json!({
237 "xrpc": "sh.tangled.graph.listFollows",
238 "params": { "subject": owner.as_ref() },
239 "enrich": [{ "source": "sh.tangled.graph.follow:subject", "type": "count" }, { "source": "sh.tangled.graph.follow:.repo", "type": "count" }]
240 })))
241 .await
242 .unwrap(),
243 )
244 .await;
245
246 assert_eq!(status, StatusCode::OK, "{body}");
247 let nel = &body["stats"]["did:plc:nel"];
248 assert_eq!(
249 nel["sh.tangled.graph.follow:subject"]["count"],
250 json!(2),
251 "{body}"
252 );
253 assert_eq!(
254 nel["sh.tangled.graph.follow:.repo"]["count"],
255 json!(1),
256 "{body}"
257 );
258}
259
260#[tokio::test]
261async fn sources_scope_which_refs_get_enriched() {
262 let h = Harness::new().await;
263 let owner = did("did:plc:nel");
264 let repo_did = did("did:plc:limpet");
265 repo_fixture(&h, &owner, &repo_did).await;
266
267 let app = router(h.state.clone());
268 let (status, body) = json_response(
269 app.oneshot(enrich_request(json!({
270 "xrpc": "sh.tangled.repo.listRepos",
271 "params": { "subject": owner.as_ref() },
272 "enrich": [{ "source": "sh.tangled.feed.star:subject", "type": "count" }],
273 "sources": ["items[].value.repoDid"]
274 })))
275 .await
276 .unwrap(),
277 )
278 .await;
279 assert_eq!(status, StatusCode::OK, "{body}");
280 let stats = &body["stats"];
281 assert_eq!(
282 stats["did:plc:limpet"]["sh.tangled.feed.star:subject"]["count"],
283 json!(3)
284 );
285 assert_eq!(stats.as_object().unwrap().len(), 1, "{body}");
286
287 // a path matching nothing is empty stats, not an error, since selection is vector-matched
288 let app = router(h.state.clone());
289 let (status, body) = json_response(
290 app.oneshot(enrich_request(json!({
291 "xrpc": "sh.tangled.repo.listRepos",
292 "params": { "subject": owner.as_ref() },
293 "enrich": [{ "source": "sh.tangled.feed.star:subject", "type": "count" }],
294 "sources": ["items[].value.nope"]
295 })))
296 .await
297 .unwrap(),
298 )
299 .await;
300 assert_eq!(status, StatusCode::OK, "{body}");
301 assert_eq!(body["stats"], json!({}));
302}
303
304#[tokio::test]
305async fn inner_record_miss_passes_through_as_404() {
306 let h = Harness::new().await;
307 let app = router(h.state.clone());
308 let (status, body) = json_response(
309 app.oneshot(enrich_request(json!({
310 "xrpc": "sh.tangled.repo.getRepo",
311 "params": { "repo": "at://did:plc:nel/sh.tangled.repo/absent" },
312 "enrich": [{ "source": "sh.tangled.feed.star:subject", "type": "count" }]
313 })))
314 .await
315 .unwrap(),
316 )
317 .await;
318 assert_eq!(status, StatusCode::NOT_FOUND, "{body}");
319 assert_eq!(body["error"], json!("RecordNotFound"));
320}
321
322#[tokio::test]
323async fn rejects_bad_requests() {
324 let h = Harness::new().await;
325 let cases = [
326 json!({ "xrpc": "sh.tangled.nope.nope", "enrich": [] }),
327 json!({
328 "xrpc": "sh.tangled.repo.countRepos",
329 "params": { "subject": "did:plc:nel" },
330 "enrich": [{ "source": "sh.tangled.nope:subject", "type": "count" }]
331 }),
332 json!({
333 "xrpc": "sh.tangled.repo.countRepos",
334 "params": { "subject": "did:plc:nel" },
335 "enrich": [{ "source": "sh.tangled.feed.star:subject.uri", "type": "count" }]
336 }),
337 json!({
338 "xrpc": "sh.tangled.repo.countRepos",
339 "params": { "subject": "did:plc:nel" },
340 "enrich": [{ "source": "sh.tangled.feed.star:.rkey", "type": "count" }]
341 }),
342 json!({
343 "xrpc": "sh.tangled.repo.countRepos",
344 "params": { "subject": "did:plc:nel" },
345 "enrich": [{ "source": "sh.tangled.feed.star:subject", "type": "count" }],
346 "sources": ["items["]
347 }),
348 // no defaults, a source without a colon is rejected
349 json!({
350 "xrpc": "sh.tangled.repo.countRepos",
351 "params": { "subject": "did:plc:nel" },
352 "enrich": [{ "source": "sh.tangled.feed.star", "type": "count" }]
353 }),
354 ];
355 for case in cases {
356 let app = router(h.state.clone());
357 let (status, body) = json_response(app.oneshot(enrich_request(case)).await.unwrap()).await;
358 assert_eq!(status, StatusCode::BAD_REQUEST, "{body}");
359 assert_eq!(body["error"], json!("InvalidRequest"), "{body}");
360 }
361
362 // serde rejects a missing type before the handler sees it
363 // that returns plain-text 422, not our 400 json body
364 let app = router(h.state.clone());
365 let response = app
366 .oneshot(enrich_request(json!({
367 "xrpc": "sh.tangled.repo.countRepos",
368 "params": { "subject": "did:plc:nel" },
369 "enrich": [{ "source": "sh.tangled.feed.star:subject" }]
370 })))
371 .await
372 .unwrap();
373 assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY);
374}
375
376#[tokio::test]
377async fn viewer_aggregation_uses_explicit_viewer_param() {
378 let h = Harness::new().await;
379 let owner = did("did:plc:abc");
380 let repo_did = did("did:plc:limpet");
381 repo_fixture(&h, &owner, &repo_did).await;
382
383 // the viewer already starred this repo, for the checks below
384 let subject = SubjectRef::Did(repo_did.clone());
385 h.state.edges.add(Edge {
386 kind: nsid("sh.tangled.feed.star"),
387 subject,
388 source: at("at://did:plc:nel/sh.tangled.feed.star/r99"),
389 sort_micros: 99,
390 });
391
392 let app = router(h.state.clone());
393
394 // missing viewer param is a 400, viewer descriptors require it
395 let no_viewer = json!({
396 "xrpc": "sh.tangled.repo.listRepos",
397 "params": { "subject": owner.as_ref() },
398 "enrich": [{ "source": "sh.tangled.feed.star:subject", "type": "viewer" }]
399 });
400 let (status, _) = json_response(
401 app.clone()
402 .oneshot(enrich_request(no_viewer))
403 .await
404 .unwrap(),
405 )
406 .await;
407 assert_eq!(status, StatusCode::BAD_REQUEST);
408
409 // viewer who starred it gets their own star uri back
410 let starred_viewer = json!({
411 "xrpc": "sh.tangled.repo.listRepos",
412 "params": { "subject": owner.as_ref() },
413 "enrich": [{ "source": "sh.tangled.feed.star:subject", "type": "viewer" }],
414 "viewer": "did:plc:nel"
415 });
416 let (status, resp) = json_response(
417 app.clone()
418 .oneshot(enrich_request(starred_viewer))
419 .await
420 .unwrap(),
421 )
422 .await;
423 assert_eq!(status, StatusCode::OK);
424 let stats = &resp["stats"][repo_did.as_str()]["sh.tangled.feed.star:subject"];
425 assert_eq!(
426 stats["viewer"],
427 json!("at://did:plc:nel/sh.tangled.feed.star/r99")
428 );
429
430 // a viewer who never starred it gets an explicit null, not absent
431 let other_viewer = json!({
432 "xrpc": "sh.tangled.repo.listRepos",
433 "params": { "subject": owner.as_ref() },
434 "enrich": [{ "source": "sh.tangled.feed.star:subject", "type": "viewer" }],
435 "viewer": "did:plc:someoneelse"
436 });
437 let (status, resp) =
438 json_response(app.oneshot(enrich_request(other_viewer)).await.unwrap()).await;
439 assert_eq!(status, StatusCode::OK);
440 let stats = &resp["stats"][repo_did.as_str()]["sh.tangled.feed.star:subject"];
441 assert_eq!(stats["viewer"], Value::Null);
442}