From 81a65ebb0f5761c873aa1c82e2c5a5b3fb365070 Mon Sep 17 00:00:00 2001 From: teq Date: Tue, 14 Jul 2026 14:45:07 -0400 Subject: [PATCH] bobbin/xrpc: serve exactly one $type on record values Signed-off-by: teq --- bobbin/crates/xrpc/src/lib.rs | 45 +++++++++++++++++--------- bobbin/crates/xrpc/tests/bulk.rs | 25 ++++++++++++++ bobbin/crates/xrpc/tests/cold_start.rs | 34 +++++++++++++++++++ 3 files changed, 88 insertions(+), 16 deletions(-) diff --git a/bobbin/crates/xrpc/src/lib.rs b/bobbin/crates/xrpc/src/lib.rs index 370859eb..fa1814f6 100644 --- a/bobbin/crates/xrpc/src/lib.rs +++ b/bobbin/crates/xrpc/src/lib.rs @@ -729,6 +729,19 @@ impl From for CoverageEnvelope { } } +struct Deduped(T); + +impl Serialize for Deduped { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + serde_json::to_value(&self.0) + .map_err(serde::ser::Error::custom)? + .serialize(serializer) + } +} + #[derive(Serialize)] #[serde(rename_all = "camelCase")] struct RecordView { @@ -1316,19 +1329,19 @@ where async fn get_repo( State(state): State, XrpcQuery(q): XrpcQuery, -) -> Result>, XrpcError> { +) -> Result>>, XrpcError> { let (body, value) = fetch::>(&state, &q.repo).await?; - Ok(Json(RepoGetRecordOutput { + Ok(Json(Deduped(RepoGetRecordOutput { cid: Some(body.cid.clone()), uri: body.uri.clone(), value, - })) + }))) } async fn get_repo_by_repo_did( State(state): State, XrpcQuery(q): XrpcQuery, -) -> Result>, XrpcError> { +) -> Result>>, XrpcError> { let ident = state .resolver .lookup_by_repo_did(&q.repo_did) @@ -1341,47 +1354,47 @@ async fn get_repo_by_repo_did( ) .expect("Did and Rkey newtypes already validated, at-uri assembly cannot fail"); let (body, value) = fetch_from_uri::>(&state, uri).await?; - Ok(Json(RepoGetRecordOutput { + Ok(Json(Deduped(RepoGetRecordOutput { cid: Some(body.cid.clone()), uri: body.uri.clone(), value, - })) + }))) } async fn get_profile( State(state): State, XrpcQuery(q): XrpcQuery, -) -> Result>, XrpcError> { +) -> Result>>, XrpcError> { let (body, value) = fetch::>(&state, &q.actor).await?; - Ok(Json(ProfileGetRecordOutput { + Ok(Json(Deduped(ProfileGetRecordOutput { cid: Some(body.cid.clone()), uri: body.uri.clone(), value, - })) + }))) } async fn get_issue( State(state): State, XrpcQuery(q): XrpcQuery, -) -> Result>, XrpcError> { +) -> Result>>, XrpcError> { let (body, value) = fetch::>(&state, &q.issue).await?; - Ok(Json(IssueGetRecordOutput { + Ok(Json(Deduped(IssueGetRecordOutput { cid: Some(body.cid.clone()), uri: body.uri.clone(), value, - })) + }))) } async fn get_pull( State(state): State, XrpcQuery(q): XrpcQuery, -) -> Result>, XrpcError> { +) -> Result>>, XrpcError> { let (body, value) = fetch::>(&state, &q.pull).await?; - Ok(Json(PullGetRecordOutput { + Ok(Json(Deduped(PullGetRecordOutput { cid: Some(body.cid.clone()), uri: body.uri.clone(), value, - })) + }))) } async fn get_repos( @@ -1657,7 +1670,7 @@ where Some((Ok::, Infallible>(head), st)) } PagePhase::Body { first } => match st.items.next().await { - Some(Ok(view)) => match serde_json::to_vec(&view) { + Some(Ok(view)) => match serde_json::to_vec(&Deduped(&view)) { Ok(encoded) => { let mut chunk = Vec::with_capacity(encoded.len() + 1); if !first { diff --git a/bobbin/crates/xrpc/tests/bulk.rs b/bobbin/crates/xrpc/tests/bulk.rs index 5f8de0d0..39a812bb 100644 --- a/bobbin/crates/xrpc/tests/bulk.rs +++ b/bobbin/crates/xrpc/tests/bulk.rs @@ -521,3 +521,28 @@ async fn malformed_uri_in_list_returns_400() { .unwrap(); assert_eq!(resp.status(), StatusCode::BAD_REQUEST); } + +#[tokio::test] +async fn bulk_items_serialize_a_single_type_key() { + let h = Harness::new().await; + h.mount( + &did("did:plc:teq"), + &nsid("sh.tangled.repo"), + &rkey("abalone"), + repo_body("abalone"), + ) + .await; + let app = router(h.state.clone()); + let resp = app + .oneshot(bulk_request( + "sh.tangled.repo.getRepos", + "repos", + &["at://did:plc:teq/sh.tangled.repo/abalone"], + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap(); + let raw = String::from_utf8(bytes.to_vec()).unwrap(); + assert_eq!(raw.matches("\"$type\"").count(), 1, "body: {raw}"); +} diff --git a/bobbin/crates/xrpc/tests/cold_start.rs b/bobbin/crates/xrpc/tests/cold_start.rs index dc97f38d..30bc3bea 100644 --- a/bobbin/crates/xrpc/tests/cold_start.rs +++ b/bobbin/crates/xrpc/tests/cold_start.rs @@ -697,3 +697,37 @@ async fn get_repo_by_repo_did_400_on_invalid_did() { .unwrap(); assert_eq!(resp.status(), StatusCode::BAD_REQUEST); } + +#[tokio::test] +async fn record_values_serialize_a_single_type_key() { + let server = MockServer::start().await; + + mount_record( + &server, + &did("did:plc:teq"), + &nsid("sh.tangled.repo"), + &rkey("r1"), + json!({ + "$type": "sh.tangled.repo", + "name": "clam", + "knot": "oyster.cafe", + "createdAt": "2026-05-01T00:00:00Z" + }), + ) + .await; + + let app = router(fresh_app(&Url::parse(&server.uri()).unwrap()).await); + let resp = app + .oneshot(xrpc_request( + "sh.tangled.repo.getRepo", + "repo", + "at://did:plc:teq/sh.tangled.repo/r1", + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap(); + let raw = String::from_utf8(bytes.to_vec()).unwrap(); + assert_eq!(raw.matches("\"$type\"").count(), 1, "body: {raw}"); + assert!(raw.contains("\"$type\":\"sh.tangled.repo\""), "body: {raw}"); +} -- 2.51.2