diff --git a/crates/xrpc/tests/cold_start.rs b/crates/xrpc/tests/cold_start.rs new file mode 100644 index 0000000..fe034bf --- /dev/null +++ b/crates/xrpc/tests/cold_start.rs @@ -0,0 +1,472 @@ +use std::sync::Arc; + +use axum::body::{Body, to_bytes}; +use bobbin_record_lru::{CacheCapacity, LruRecordStore}; +use bobbin_slingshot_client::SlingshotClient; +use bobbin_xrpc::{AppState, router}; +use futures::stream::{self, StreamExt}; +use http::{Request, StatusCode}; +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"; + +async fn fresh_app(server_uri: &str) -> AppState { + AppState { + records: Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), + slingshot: SlingshotClient::new(Url::parse(server_uri).unwrap()).unwrap(), + } +} + +async fn mount_record(server: &MockServer, did: &str, collection: &str, rkey: &str, value: Value) { + let uri = format!("at://{did}/{collection}/{rkey}"); + 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)) + .and(query_param("collection", collection)) + .and(query_param("rkey", rkey)) + .respond_with(ResponseTemplate::new(200).set_body_json(body)) + .mount(server) + .await; +} + +fn xrpc_request(endpoint: &str, param: &str, at_uri: &str) -> Request { + let encoded: String = byte_serialize(at_uri.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 did = "did:plc:abalone"; + + mount_record( + &server, + did, + "sh.tangled.repo", + "r1", + json!({ + "$type": "sh.tangled.repo", + "knot": "oyster.cafe", + "createdAt": "2026-05-01T00:00:00Z" + }), + ) + .await; + + mount_record( + &server, + did, + "sh.tangled.actor.profile", + "self", + json!({ + "$type": "sh.tangled.actor.profile", + "bluesky": false, + "description": "abalone shell" + }), + ) + .await; + + mount_record( + &server, + did, + "sh.tangled.repo.issue", + "i1", + json!({ + "$type": "sh.tangled.repo.issue", + "repo": "did:plc:limpet", + "title": "broken", + "createdAt": "2026-05-01T00:00:00Z" + }), + ) + .await; + + mount_record( + &server, + did, + "sh.tangled.repo.pull", + "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(&server.uri()).await; + let app = router(state); + + let cases = [ + ( + "sh.tangled.repo.getRepo", + "repo", + format!("at://{did}/sh.tangled.repo/r1"), + "knot", + json!("oyster.cafe"), + ), + ( + "sh.tangled.actor.getProfile", + "actor", + format!("at://{did}/sh.tangled.actor.profile/self"), + "description", + json!("abalone shell"), + ), + ( + "sh.tangled.repo.getIssue", + "issue", + format!("at://{did}/sh.tangled.repo.issue/i1"), + "title", + json!("broken"), + ), + ( + "sh.tangled.repo.getPull", + "pull", + format!("at://{did}/sh.tangled.repo.pull/p1"), + "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 did = "did:plc:uni"; + let mock = Mock::given(method("GET")) + .and(path("/xrpc/com.atproto.repo.getRecord")) + .and(query_param("repo", did)) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "uri": format!("at://{did}/sh.tangled.repo/r1"), + "cid": CID, + "value": { + "$type": "sh.tangled.repo", + "knot": "nel.pet", + "createdAt": "2026-05-01T00:00:00Z" + } + }))) + .expect(1) + .mount_as_scoped(&server) + .await; + + let state = fresh_app(&server.uri()).await; + let app = router(state); + let req_uri = format!("at://{did}/sh.tangled.repo/r1"); + + 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(&server.uri()).await; + let app = router(state); + let resp = app + .oneshot(xrpc_request( + "sh.tangled.repo.getRepo", + "repo", + "at://did:plc:abalone/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(&server.uri()).await; + let app = router(state); + let resp = app + .oneshot(xrpc_request( + "sh.tangled.repo.getRepo", + "repo", + "at://nel.pet/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(&server.uri()).await; + let app = router(state); + let resp = app + .oneshot(xrpc_request( + "sh.tangled.repo.getRepo", + "repo", + "at://did:plc:abalone/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:plc:abalone", + "sh.tangled.repo", + "r1", + json!({ + "$type": "sh.tangled.knot", + "knot": "oyster.cafe", + "createdAt": "2026-05-01T00:00:00Z" + }), + ) + .await; + let state = fresh_app(&server.uri()).await; + let app = router(state); + let resp = app + .oneshot(xrpc_request( + "sh.tangled.repo.getRepo", + "repo", + "at://did:plc:abalone/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:abalone")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "uri": "at://did:plc:abalone/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(&server.uri()).await; + let app = router(state); + let req = || { + xrpc_request( + "sh.tangled.repo.getRepo", + "repo", + "at://did:plc:abalone/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 missing_uri_param_returns_json_envelope() { + let server = MockServer::start().await; + let state = fresh_app(&server.uri()).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(&server.uri()).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:abalone")) + .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(&server.uri()).await; + let app = router(state); + let resp = app + .oneshot(xrpc_request( + "sh.tangled.repo.getRepo", + "repo", + "at://did:plc:abalone/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:abalone/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(&server.uri()).await; + let app = router(state); + let resp = app + .oneshot(xrpc_request( + "sh.tangled.repo.getRepo", + "repo", + "at://did:plc:abalone/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(&server.uri()).await; + let app = router(state); + let resp = app + .oneshot(xrpc_request( + "sh.tangled.repo.getRepo", + "repo", + "at://did:plc:abalone/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(&server.uri()).await; + let app = router(state); + let resp = app + .oneshot(xrpc_request( + "sh.tangled.repo.getRepo", + "repo", + "at://did:plc:abalone/sh.tangled.repo/r1", + )) + .await + .unwrap(); + let (status, body) = json_response(resp).await; + assert_eq!(status, StatusCode::BAD_GATEWAY); + assert_eq!(body["error"], "UpstreamFailed"); +}