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