Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640use std::sync::Arc;
use axum::body::{Body, to_bytes};use bobbin_edge_index::{CoverageWatch, EdgeStore, StateIndex};use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig};use bobbin_record_lru::{CacheCapacity, LruRecordStore};use bobbin_resolver::RepoIdResolver;use bobbin_runtime::{RuntimeHasher, SystemClock};use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader};use bobbin_slingshot_client::SlingshotClient;use bobbin_xrpc::{AppState, router};use futures::stream::{self, StreamExt};use http::{Request, StatusCode};use jacquard_common::DefaultStr;use jacquard_common::types::did::Did;use jacquard_common::types::nsid::Nsid;use jacquard_common::types::recordkey::Rkey;use serde_json::{Value, json};use tower::ServiceExt;use url::Url;use url::form_urlencoded::byte_serialize;use wiremock::matchers::{method, path, query_param};use wiremock::{Mock, MockServer, ResponseTemplate};
const CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i";
fn did(s: &str) -> Did<DefaultStr> { Did::new_owned(s).unwrap()}
fn rkey(s: &str) -> Rkey<DefaultStr> { Rkey::new_owned(s).unwrap()}
fn nsid(s: &'static str) -> Nsid<DefaultStr> { Nsid::new_static(s).unwrap()}
async fn fresh_app(server_uri: &Url) -> AppState { AppState::new( Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), SlingshotClient::with_default_http(server_uri.clone()).unwrap(), Arc::new(EdgeStore::new(RuntimeHasher::default())), Arc::new(StateIndex::new(RuntimeHasher::default())), Arc::new(StateIndex::new(RuntimeHasher::default())), Arc::new(CoverageWatch::new()), Arc::new( KnotProxy::new( KnotProxyConfig::default(), KnotHttpConfig::default(), Arc::new(SystemClock::new()), RuntimeHasher::default(), ) .unwrap(), ), Arc::new(SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap()) as Arc<dyn SearchReader>, Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), )}
async fn mount_record( server: &MockServer, did: &Did<DefaultStr>, collection: &Nsid<DefaultStr>, rkey: &Rkey<DefaultStr>, value: Value,) { let uri = format!( "at://{}/{}/{}", did.as_ref(), collection.as_ref(), rkey.as_ref() ); let body = json!({ "uri": uri, "cid": CID, "value": value }); Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .and(query_param("repo", did.as_ref())) .and(query_param("collection", collection.as_ref())) .and(query_param("rkey", rkey.as_ref())) .respond_with(ResponseTemplate::new(200).set_body_json(body)) .mount(server) .await;}
fn xrpc_request(endpoint: &str, param: &str, value: &str) -> Request<Body> { let encoded: String = byte_serialize(value.as_bytes()).collect(); Request::builder() .uri(format!("/xrpc/{endpoint}?{param}={encoded}")) .body(Body::empty()) .unwrap()}
async fn json_response(resp: axum::response::Response) -> (StatusCode, Value) { let status = resp.status(); let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap(); let parsed: Value = serde_json::from_slice(&bytes).expect("response is JSON"); (status, parsed)}
#[tokio::test]async fn cold_start_serves_all_four_point_lookups() { let server = MockServer::start().await; let clam = did("did:plc:clam");
mount_record( &server, &clam, &nsid("sh.tangled.repo"), &rkey("r1"), json!({ "$type": "sh.tangled.repo", "name": "clam", "knot": "oyster.cafe", "createdAt": "2026-05-01T00:00:00Z" }), ) .await;
mount_record( &server, &clam, &nsid("sh.tangled.actor.profile"), &rkey("self"), json!({ "$type": "sh.tangled.actor.profile", "bluesky": false, "description": "clam shell" }), ) .await;
mount_record( &server, &clam, &nsid("sh.tangled.repo.issue"), &rkey("i1"), json!({ "$type": "sh.tangled.repo.issue", "repo": "did:plc:limpet", "title": "broken", "createdAt": "2026-05-01T00:00:00Z" }), ) .await;
mount_record( &server, &clam, &nsid("sh.tangled.repo.pull"), &rkey("p1"), json!({ "$type": "sh.tangled.repo.pull", "title": "ship", "createdAt": "2026-05-01T00:00:00Z", "rounds": [], "target": {"repo": "did:plc:limpet", "branch": "main"} }), ) .await;
let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state);
let cases = [ ( "sh.tangled.repo.getRepo", "repo", format!("at://{}/sh.tangled.repo/r1", clam.as_ref()), "knot", json!("oyster.cafe"), ), ( "sh.tangled.actor.getProfile", "actor", format!("at://{}/sh.tangled.actor.profile/self", clam.as_ref()), "description", json!("clam shell"), ), ( "sh.tangled.repo.getIssue", "issue", format!("at://{}/sh.tangled.repo.issue/i1", clam.as_ref()), "title", json!("broken"), ), ( "sh.tangled.repo.getPull", "pull", format!("at://{}/sh.tangled.repo.pull/p1", clam.as_ref()), "title", json!("ship"), ), ];
stream::iter(cases) .for_each(|(endpoint, param, at_uri, field, expected)| { let app = app.clone(); async move { let resp = app .oneshot(xrpc_request(endpoint, param, &at_uri)) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::OK, "{endpoint} status"); assert_eq!(body["uri"], at_uri, "{endpoint} uri"); assert_eq!(body["cid"], CID, "{endpoint} cid"); assert_eq!( body["value"][field], expected, "{endpoint} body field {field}" ); } }) .await;}
#[tokio::test]async fn second_call_is_served_from_lru() { let server = MockServer::start().await; let uni = did("did:plc:uni"); let mock = Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .and(query_param("repo", uni.as_ref())) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "uri": format!("at://{}/sh.tangled.repo/r1", uni.as_ref()), "cid": CID, "value": { "$type": "sh.tangled.repo", "name": "uni", "knot": "witchcraft.systems", "createdAt": "2026-05-01T00:00:00Z" } }))) .expect(1) .mount_as_scoped(&server) .await;
let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let req_uri = format!("at://{}/sh.tangled.repo/r1", uni.as_ref());
stream::iter(0..3) .for_each(|_| { let app = app.clone(); let req_uri = req_uri.clone(); async move { let resp = app .oneshot(xrpc_request("sh.tangled.repo.getRepo", "repo", &req_uri)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); } }) .await;
drop(mock);}
#[tokio::test]async fn collection_mismatch_is_400() { let server = MockServer::start().await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let resp = app .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", "at://did:plc:clam/sh.tangled.actor.profile/self", )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::BAD_REQUEST);}
#[tokio::test]async fn handle_authority_is_400() { let server = MockServer::start().await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let resp = app .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", "at://witchcraft.systems/sh.tangled.repo/r1", )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::BAD_REQUEST);}
#[tokio::test]async fn slingshot_404_propagates_as_404() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .respond_with(ResponseTemplate::new(404)) .mount(&server) .await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let resp = app .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", "at://did:plc:clam/sh.tangled.repo/missing", )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::NOT_FOUND);}
#[tokio::test]async fn wrong_record_type_is_502() { let server = MockServer::start().await; mount_record( &server, &did("did:plc:clam"), &nsid("sh.tangled.repo"), &rkey("r1"), json!({ "$type": "sh.tangled.knot", "knot": "oyster.cafe", "createdAt": "2026-05-01T00:00:00Z" }), ) .await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let resp = app .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", "at://did:plc:clam/sh.tangled.repo/r1", )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_GATEWAY); assert_eq!(body["error"], "InvalidRecord");}
#[tokio::test]async fn wrong_type_does_not_poison_cache() { let server = MockServer::start().await; let mock = Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .and(query_param("repo", "did:plc:clam")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "uri": "at://did:plc:clam/sh.tangled.repo/r1", "cid": CID, "value": { "$type": "sh.tangled.knot", "knot": "oyster.cafe", "createdAt": "2026-05-01T00:00:00Z" } }))) .expect(2) .mount_as_scoped(&server) .await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let req = || { xrpc_request( "sh.tangled.repo.getRepo", "repo", "at://did:plc:clam/sh.tangled.repo/r1", ) }; let first = app.clone().oneshot(req()).await.unwrap(); assert_eq!(first.status(), StatusCode::BAD_GATEWAY); let second = app.clone().oneshot(req()).await.unwrap(); assert_eq!(second.status(), StatusCode::BAD_GATEWAY); drop(mock);}
#[tokio::test]async fn profile_with_empty_preferred_handle_is_tolerated() { let server = MockServer::start().await; let nel = did("did:plc:nel"); mount_record( &server, &nel, &nsid("sh.tangled.actor.profile"), &rkey("self"), json!({ "$type": "sh.tangled.actor.profile", "bluesky": true, "preferredHandle": "", "description": "empty handle, valid profile" }), ) .await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let at_uri = format!("at://{}/sh.tangled.actor.profile/self", nel.as_ref()); let resp = app .oneshot(xrpc_request( "sh.tangled.actor.getProfile", "actor", &at_uri, )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::OK, "status: {body}"); assert_eq!(body["uri"], at_uri); assert_eq!(body["value"]["description"], "empty handle, valid profile"); assert!(body["value"]["preferredHandle"].is_null());}
#[tokio::test]async fn missing_uri_param_returns_json_envelope() { let server = MockServer::start().await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let req = Request::builder() .uri("/xrpc/sh.tangled.repo.getRepo") .body(Body::empty()) .unwrap(); let resp = app.oneshot(req).await.unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_REQUEST); assert_eq!(body["error"], "InvalidRequest"); assert!(body["message"].is_string());}
#[tokio::test]async fn malformed_at_uri_returns_400_envelope() { let server = MockServer::start().await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let resp = app .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", "definitely-not-an-at-uri", )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_REQUEST); assert_eq!(body["error"], "InvalidRequest");}
#[tokio::test]async fn upstream_uri_mismatch_routes_to_invalid_record() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .and(query_param("repo", "did:plc:clam")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "uri": "at://did:plc:limpet/sh.tangled.repo/elsewhere", "cid": CID, "value": { "$type": "sh.tangled.repo", "knot": "oyster.cafe", "createdAt": "2026-05-01T00:00:00Z" } }))) .mount(&server) .await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let resp = app .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", "at://did:plc:clam/sh.tangled.repo/r1", )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_GATEWAY); assert_eq!(body["error"], "InvalidRecord");}
#[tokio::test]async fn upstream_garbage_cid_routes_to_invalid_record() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "uri": "at://did:plc:clam/sh.tangled.repo/r1", "cid": "not-a-real-cid", "value": { "$type": "sh.tangled.repo", "knot": "oyster.cafe", "createdAt": "2026-05-01T00:00:00Z" } }))) .mount(&server) .await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let resp = app .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", "at://did:plc:clam/sh.tangled.repo/r1", )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_GATEWAY); assert_eq!(body["error"], "InvalidRecord");}
#[tokio::test]async fn oversize_upstream_body_routes_to_upstream_failed() { let server = MockServer::start().await; let payload = vec![b'x'; 8 * 1024 * 1024]; Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .respond_with( ResponseTemplate::new(200) .insert_header("content-type", "application/json") .set_body_bytes(payload), ) .mount(&server) .await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let resp = app .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", "at://did:plc:clam/sh.tangled.repo/r1", )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_GATEWAY); assert_eq!(body["error"], "UpstreamFailed");}
#[tokio::test]async fn upstream_503_routes_to_upstream_failed() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .respond_with(ResponseTemplate::new(503)) .mount(&server) .await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let resp = app .oneshot(xrpc_request( "sh.tangled.repo.getRepo", "repo", "at://did:plc:clam/sh.tangled.repo/r1", )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_GATEWAY); assert_eq!(body["error"], "UpstreamFailed");}
#[tokio::test]async fn get_repo_by_repo_did_returns_observed_record() { let server = MockServer::start().await; let owner_did = did("did:plc:scallop"); let rk = rkey("r1"); let repo_did = did("did:plc:limpet"); mount_record( &server, &owner_did, &nsid("sh.tangled.repo"), &rk, json!({ "$type": "sh.tangled.repo", "name": "scallop", "knot": "oyster.cafe", "createdAt": "2026-05-01T00:00:00Z", "repoDid": repo_did.as_ref(), }), ) .await;
let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; state .resolver .observe(owner_did.clone(), rk.clone(), Some(repo_did.clone())) .await;
let app = router(state); let resp = app .oneshot(xrpc_request( "sh.tangled.repo.getRepoByRepoDid", "repoDid", repo_did.as_ref(), )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::OK); assert_eq!( body["uri"], format!( "at://{}/sh.tangled.repo/{}", owner_did.as_ref(), rk.as_ref() ) ); assert_eq!(body["value"]["name"], "scallop"); assert_eq!(body["value"]["repoDid"], repo_did.as_ref());}
#[tokio::test]async fn get_repo_by_repo_did_404_when_unobserved() { let server = MockServer::start().await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let resp = app .oneshot(xrpc_request( "sh.tangled.repo.getRepoByRepoDid", "repoDid", "did:plc:whelk", )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::NOT_FOUND);}
#[tokio::test]async fn get_repo_by_repo_did_400_on_invalid_did() { let server = MockServer::start().await; let state = fresh_app(&Url::parse(&server.uri()).unwrap()).await; let app = router(state); let resp = app .oneshot(xrpc_request( "sh.tangled.repo.getRepoByRepoDid", "repoDid", "not-a-did", )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::BAD_REQUEST);}