diff --git a/crates/search/src/lib.rs b/crates/search/src/lib.rs index e5240c5..1be42b6 100644 --- a/crates/search/src/lib.rs +++ b/crates/search/src/lib.rs @@ -389,7 +389,13 @@ impl SearchSink for SearchIndex { } async fn remove(&self, uri: &AtUri) { - if self.inner.tx.send(WriteOp::Remove(uri.clone())).await.is_err() { + if self + .inner + .tx + .send(WriteOp::Remove(uri.clone())) + .await + .is_err() + { warn!("search writer channel closed; remove dropped"); } } diff --git a/crates/slingshot-client/src/lib.rs b/crates/slingshot-client/src/lib.rs index 2c90097..ba39279 100644 --- a/crates/slingshot-client/src/lib.rs +++ b/crates/slingshot-client/src/lib.rs @@ -7,6 +7,7 @@ use cid::Cid as IpldCid; use futures::TryStreamExt; use jacquard_common::BosStr; use jacquard_common::types::did::Did; +use jacquard_common::types::ident::AtIdentifier; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::recordkey::Rkey; use jacquard_common::types::string::{AtStrError, AtUri}; @@ -20,6 +21,7 @@ const USER_AGENT: &str = concat!("bobbin/", env!("CARGO_PKG_VERSION")); const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); const GET_RECORD_PATH: &str = "xrpc/com.atproto.repo.getRecord"; +const RESOLVE_MINI_DOC_PATH: &str = "xrpc/com.bad-example.identity.resolveMiniDoc"; pub const MAX_BODY_BYTES: u64 = 4 * 1024 * 1024; #[derive(Clone, Debug)] @@ -67,6 +69,28 @@ impl SlingshotClient { Ok(Self { http, base }) } + pub async fn resolve_mini_doc( + &self, + identifier: &AtIdentifier, + ) -> Result + where + S: BosStr, + { + let mut url = self.base.join(RESOLVE_MINI_DOC_PATH).expect( + "base url is hierarchical and RESOLVE_MINI_DOC_PATH is a literal relative path", + ); + url.query_pairs_mut() + .clear() + .append_pair("identifier", identifier.as_str()); + + let resp = self.http.get(url).send().await?; + match resp.status() { + StatusCode::OK => read_bounded(resp).await, + StatusCode::NOT_FOUND => Err(SlingshotError::NotFound), + other => Err(SlingshotError::Upstream(other)), + } + } + pub async fn get_record( &self, repo: &Did, diff --git a/crates/types/src/edges.rs b/crates/types/src/edges.rs index a2bf5a9..45b25e5 100644 --- a/crates/types/src/edges.rs +++ b/crates/types/src/edges.rs @@ -201,34 +201,13 @@ fn did_subject>( ) -> Result>, AtStrError> { match (did.as_ref(), uri.as_ref()) { (Some(d), _) => did_to_aturi(d).map(Some), - (None, Some(u)) => uri_authority_did_aturi(u), - (None, None) => Ok(None), + (None, Some(u)) if uri_authority_is_did(u) => aturi_to_owned(u).map(Some), + _ => Ok(None), } } -fn star_subject>( - uri: &Option>, - did: &Option>, -) -> Result>, AtStrError> { - match (did.as_ref(), uri.as_ref()) { - (Some(d), _) => did_to_aturi(d).map(Some), - (None, Some(u)) if is_string_at_uri(u) => aturi_to_owned(u).map(Some), - (None, Some(u)) => uri_authority_did_aturi(u), - (None, None) => Ok(None), - } -} - -fn uri_authority_did_aturi>( - uri: &AtUri, -) -> Result>, AtStrError> { - crate::ids::did_from_aturi(uri.as_ref()) - .map(|d| did_to_aturi(&d)) - .transpose() -} - -fn is_string_at_uri>(uri: &AtUri) -> bool { - uri.collection() - .is_some_and(|c| c.as_ref() == "sh.tangled.string") +fn uri_authority_is_did>(uri: &AtUri) -> bool { + crate::ids::did_from_aturi(uri.as_ref()).is_some() } fn one_edge( @@ -247,7 +226,7 @@ fn star_edges( source: &AtUri, record: &Star, ) -> Result, ExtractError> { - let Some(subject) = star_subject(&record.subject, &record.subject_did)? else { + let Some(subject) = did_subject(&record.subject, &record.subject_did)? else { return Ok(Vec::new()); }; Ok(one_edge("sh.tangled.feed.star", subject, source)) @@ -500,7 +479,7 @@ mod tests { } #[test] - fn star_repo_subject_uri_collapses_to_authority_did() { + fn star_repo_subject_uri_is_preserved_for_normalization() { let edges = extract( "sh.tangled.feed.star", "at://did:plc:olaren/sh.tangled.feed.star/abcabcabcabcz", @@ -511,7 +490,11 @@ mod tests { }), ); assert_eq!(edges.len(), 1); - assert_eq!(edges[0].subject, at("at://did:plc:abalone")); + assert_eq!( + edges[0].subject, + at("at://did:plc:abalone/sh.tangled.repo/r1"), + "extract leaves repo URIs intact for the ingest-time normalization pass to resolve via slingshot or observed repoDID", + ); } #[test] @@ -686,14 +669,17 @@ mod tests { } #[test] - fn artifact_uri_only_collapses_to_authority_did() { + fn artifact_uri_only_is_preserved_for_normalization() { let edges = extract( "sh.tangled.repo.artifact", "at://did:plc:nel/sh.tangled.repo.artifact/abcabcabcabcz", artifact_body(Some("at://did:plc:abalone/sh.tangled.repo/r1"), None), ); assert_eq!(edges.len(), 1); - assert_eq!(edges[0].subject, at("at://did:plc:abalone")); + assert_eq!( + edges[0].subject, + at("at://did:plc:abalone/sh.tangled.repo/r1"), + ); } #[test] @@ -707,7 +693,7 @@ mod tests { } #[test] - fn issue_uri_only_collapses_to_authority_did() { + fn issue_uri_only_is_preserved_for_normalization() { let edges = extract( "sh.tangled.repo.issue", "at://did:plc:nel/sh.tangled.repo.issue/abcabcabcabcz", @@ -719,11 +705,14 @@ mod tests { }), ); assert_eq!(edges.len(), 1); - assert_eq!(edges[0].subject, at("at://did:plc:abalone")); + assert_eq!( + edges[0].subject, + at("at://did:plc:abalone/sh.tangled.repo/r1"), + ); } #[test] - fn pull_uri_only_collapses_to_authority_did() { + fn pull_uri_only_is_preserved_for_normalization() { let edges = extract( "sh.tangled.repo.pull", "at://did:plc:nel/sh.tangled.repo.pull/abcabcabcabcz", @@ -739,7 +728,10 @@ mod tests { }), ); assert_eq!(edges.len(), 1); - assert_eq!(edges[0].subject, at("at://did:plc:abalone")); + assert_eq!( + edges[0].subject, + at("at://did:plc:abalone/sh.tangled.repo/r1"), + ); } #[test] diff --git a/crates/xrpc/Cargo.toml b/crates/xrpc/Cargo.toml index aa11299..a77e4c0 100644 --- a/crates/xrpc/Cargo.toml +++ b/crates/xrpc/Cargo.toml @@ -20,6 +20,8 @@ http = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } +tower-http = { workspace = true } +tracing = { workspace = true } [dev-dependencies] http = { workspace = true } diff --git a/crates/xrpc/src/lib.rs b/crates/xrpc/src/lib.rs index d973479..b337b5e 100644 --- a/crates/xrpc/src/lib.rs +++ b/crates/xrpc/src/lib.rs @@ -61,6 +61,8 @@ use jacquard_common::xrpc::XrpcResp; use jacquard_common::{DefaultStr, IntoStatic}; use serde::{Deserialize, Serialize}; use thiserror::Error; +use tower_http::trace::{DefaultMakeSpan, DefaultOnFailure, DefaultOnResponse, TraceLayer}; +use tracing::Level; const DEFAULT_LIMIT: u32 = 50; const FETCH_CONCURRENCY: usize = 8; @@ -146,7 +148,10 @@ pub fn router(state: AppState) -> Router { .route("/xrpc/sh.tangled.repo.listArtifacts", get(list_artifacts)) .route("/xrpc/sh.tangled.repo.countArtifacts", get(count_artifacts)) .route("/xrpc/sh.tangled.knot.listMembers", get(list_knot_members)) - .route("/xrpc/sh.tangled.knot.countMembers", get(count_knot_members)) + .route( + "/xrpc/sh.tangled.knot.countMembers", + get(count_knot_members), + ) .route( "/xrpc/sh.tangled.spindle.listMembers", get(list_spindle_members), @@ -158,7 +163,18 @@ pub fn router(state: AppState) -> Router { .route("/xrpc/sh.tangled.string.listStrings", get(list_strings)) .route("/xrpc/sh.tangled.string.countStrings", get(count_strings)) .route("/xrpc/sh.tangled.search.query", get(search_query)) + .route( + "/xrpc/com.bad-example.identity.resolveMiniDoc", + get(resolve_mini_doc), + ) .merge(knot_proxied_routes()) + .layer( + TraceLayer::new_for_http() + .make_span_with(DefaultMakeSpan::new().level(Level::INFO)) + .on_request(()) + .on_response(DefaultOnResponse::new().level(Level::INFO)) + .on_failure(DefaultOnFailure::new().level(Level::WARN)), + ) .with_state(state) } @@ -437,7 +453,7 @@ fn parse_uri(raw: &str) -> Result, XrpcError> { pub enum SubjectShape { BareDid, Collection(&'static str), - BareDidOrCollection(&'static str), + BareDidOrOneOfCollections(&'static [&'static str]), OneOfCollections(&'static [&'static str]), } @@ -446,16 +462,17 @@ pub trait HasSubject { } impl HasSubject for StarRecord { - const SHAPE: SubjectShape = SubjectShape::BareDidOrCollection("sh.tangled.string"); + const SHAPE: SubjectShape = + SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.string", "sh.tangled.repo"]); } impl HasSubject for FollowRecord { const SHAPE: SubjectShape = SubjectShape::BareDid; } impl HasSubject for IssueRecord { - const SHAPE: SubjectShape = SubjectShape::BareDid; + const SHAPE: SubjectShape = SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.repo"]); } impl HasSubject for PullRecord { - const SHAPE: SubjectShape = SubjectShape::BareDid; + const SHAPE: SubjectShape = SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.repo"]); } impl HasSubject for IssueCommentRecord { const SHAPE: SubjectShape = SubjectShape::Collection("sh.tangled.repo.issue"); @@ -474,7 +491,7 @@ impl HasSubject for PipelineStatusRecord { const SHAPE: SubjectShape = SubjectShape::Collection("sh.tangled.pipeline"); } impl HasSubject for ArtifactRecord { - const SHAPE: SubjectShape = SubjectShape::BareDid; + const SHAPE: SubjectShape = SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.repo"]); } impl HasSubject for KnotMemberRecord { const SHAPE: SubjectShape = SubjectShape::BareDid; @@ -506,14 +523,15 @@ fn parse_subject(raw: &RawAtUriParam, shape: SubjectShape) -> Result Err(XrpcError::InvalidParams(format!( "subject must be at:///{expected}/" ))), - (SubjectShape::BareDidOrCollection(_), None) => Ok(uri), - (SubjectShape::BareDidOrCollection(expected), Some(c)) if c == expected => { - require_rkey(&uri, expected)?; + (SubjectShape::BareDidOrOneOfCollections(_), None) => Ok(uri), + (SubjectShape::BareDidOrOneOfCollections(allowed), Some(c)) if allowed.contains(&c) => { + require_rkey(&uri, c)?; Ok(uri) } - (SubjectShape::BareDidOrCollection(expected), Some(c)) => { + (SubjectShape::BareDidOrOneOfCollections(allowed), Some(c)) => { Err(XrpcError::InvalidParams(format!( - "subject must be at:// or at:///{expected}/; got collection {c}" + "subject must be at:// or at://// with nsid in [{}]; got collection {c}", + allowed.join(", "), ))) } (SubjectShape::OneOfCollections(allowed), Some(c)) if allowed.contains(&c) => { @@ -929,6 +947,25 @@ async fn count_strings( count_for::(&state, q).map(Json) } +#[derive(Deserialize)] +struct ResolveMiniDocParams { + identifier: String, +} + +async fn resolve_mini_doc( + State(state): State, + XrpcQuery(q): XrpcQuery, +) -> Result { + let identifier = AtIdentifier::::new_owned(q.identifier.trim()) + .map_err(|e| XrpcError::InvalidParams(format!("identifier: {e}")))?; + let body = state + .slingshot + .resolve_mini_doc(&identifier) + .await + .map_err(map_slingshot)?; + Ok((StatusCode::OK, [(CONTENT_TYPE, "application/json")], body).into_response()) +} + async fn search_query( State(state): State, XrpcQuery(q): XrpcQuery, diff --git a/crates/xrpc/tests/aggregation.rs b/crates/xrpc/tests/aggregation.rs index b811e2a..d957a47 100644 --- a/crates/xrpc/tests/aggregation.rs +++ b/crates/xrpc/tests/aggregation.rs @@ -813,10 +813,46 @@ async fn bare_did_endpoints_reject_subject_with_path() { let cases = [ "sh.tangled.graph.listFollows", "sh.tangled.graph.countFollows", + ]; + stream::iter(cases) + .for_each(|endpoint| { + let app = app.clone(); + async move { + let resp = app + .oneshot(list_request( + endpoint, + "at://did:plc:abalone/sh.tangled.repo/r1", + &[], + )) + .await + .unwrap(); + let (status, body) = json_response(resp).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint}"); + assert_eq!(body["error"], "InvalidRequest", "{endpoint}"); + assert!( + body["message"] + .as_str() + .unwrap_or_default() + .contains("bare did at-uri"), + "{endpoint}: {}", + body["message"], + ); + } + }) + .await; +} + +#[tokio::test] +async fn repo_pointing_endpoints_accept_repo_uri_subject() { + let h = Harness::new().await; + let app = router(h.state.clone()); + let cases = [ "sh.tangled.repo.listIssues", "sh.tangled.repo.countIssues", "sh.tangled.repo.listPulls", "sh.tangled.repo.countPulls", + "sh.tangled.repo.listArtifacts", + "sh.tangled.repo.countArtifacts", ]; stream::iter(cases) .for_each(|endpoint| { @@ -830,14 +866,45 @@ async fn bare_did_endpoints_reject_subject_with_path() { )) .await .unwrap(); + let (status, _body) = json_response(resp).await; + assert_eq!( + status, + StatusCode::OK, + "{endpoint} must accept a sh.tangled.repo path subject so NoRepoDid repos stay queryable", + ); + } + }) + .await; +} + +#[tokio::test] +async fn repo_pointing_endpoints_reject_unrelated_collection() { + let h = Harness::new().await; + let app = router(h.state.clone()); + let cases = [ + "sh.tangled.repo.listIssues", + "sh.tangled.repo.listPulls", + "sh.tangled.repo.listArtifacts", + ]; + stream::iter(cases) + .for_each(|endpoint| { + let app = app.clone(); + async move { + let resp = app + .oneshot(list_request( + endpoint, + "at://did:plc:abalone/sh.tangled.knot/k1", + &[], + )) + .await + .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint}"); - assert_eq!(body["error"], "InvalidRequest", "{endpoint}"); assert!( body["message"] .as_str() .unwrap_or_default() - .contains("bare did at-uri"), + .contains("sh.tangled.repo"), "{endpoint}: {}", body["message"], ); @@ -897,26 +964,40 @@ async fn star_endpoints_reject_unrelated_collection() { let resp = app .oneshot(list_request( endpoint, - "at://did:plc:abalone/sh.tangled.repo/r1", + "at://did:plc:abalone/sh.tangled.knot/k1", &[], )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint}"); + let msg = body["message"].as_str().unwrap_or_default(); assert!( - body["message"] - .as_str() - .unwrap_or_default() - .contains("at:// or at:///sh.tangled.string/"), - "{endpoint}: {}", - body["message"], + msg.contains("sh.tangled.string") && msg.contains("sh.tangled.repo"), + "{endpoint}: {msg}", ); } }) .await; } +#[tokio::test] +async fn star_endpoints_accept_repo_subject_form() { + let h = Harness::new().await; + let app = router(h.state.clone()); + let resp = app + .oneshot(list_request( + "sh.tangled.feed.countStars", + "at://did:plc:abalone/sh.tangled.repo/r1", + &[], + )) + .await + .unwrap(); + let (status, body) = json_response(resp).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(body["count"], 0); +} + #[tokio::test] async fn star_endpoints_accept_string_subject_form() { let h = Harness::new().await; diff --git a/crates/xrpc/tests/extended.rs b/crates/xrpc/tests/extended.rs index a91264b..14c93cc 100644 --- a/crates/xrpc/tests/extended.rs +++ b/crates/xrpc/tests/extended.rs @@ -72,13 +72,11 @@ impl Harness { .and(query_param("repo", did)) .and(query_param("collection", collection)) .and(query_param("rkey", rkey)) - .respond_with( - ResponseTemplate::new(200).set_body_json(json!({ - "uri": uri, - "cid": CID, - "value": value, - })), - ) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "uri": uri, + "cid": CID, + "value": value, + }))) .mount(&self.server) .await; } @@ -248,7 +246,10 @@ async fn list_label_definitions_keys_on_owner_did() { let items = body["items"].as_array().unwrap(); assert_eq!(items.len(), 1); assert_eq!(items[0]["value"]["name"], json!("bug")); - assert_eq!(items[0]["value"]["scope"][0], json!("sh.tangled.repo.issue")); + assert_eq!( + items[0]["value"]["scope"][0], + json!("sh.tangled.repo.issue") + ); } #[tokio::test] @@ -529,13 +530,9 @@ async fn list_artifacts_keys_on_repo_did() { let app = router(h.state.clone()); let (status, body) = json_response( - app.oneshot(list_request( - "sh.tangled.repo.listArtifacts", - &subject, - &[], - )) - .await - .unwrap(), + app.oneshot(list_request("sh.tangled.repo.listArtifacts", &subject, &[])) + .await + .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); @@ -567,13 +564,9 @@ async fn list_knot_members_keys_on_subject_did() { let app = router(h.state.clone()); let (status, body) = json_response( - app.oneshot(list_request( - "sh.tangled.knot.listMembers", - &subject, - &[], - )) - .await - .unwrap(), + app.oneshot(list_request("sh.tangled.knot.listMembers", &subject, &[])) + .await + .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); @@ -642,13 +635,9 @@ async fn list_strings_keys_on_owner_did() { let app = router(h.state.clone()); let (status, body) = json_response( - app.oneshot(list_request( - "sh.tangled.string.listStrings", - &subject, - &[], - )) - .await - .unwrap(), + app.oneshot(list_request("sh.tangled.string.listStrings", &subject, &[])) + .await + .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); @@ -702,7 +691,8 @@ async fn extractor_to_xrpc_round_trip_for_pipeline() { .expect("extract") .into_iter() .for_each(|e| h.edges.add(e)); - h.mount(spindle_did, "sh.tangled.pipeline", rkey, body).await; + h.mount(spindle_did, "sh.tangled.pipeline", rkey, body) + .await; let app = router(h.state.clone()); let (status, json) = json_response( @@ -740,7 +730,8 @@ async fn list_pipelines_falls_back_to_owner_did_for_legacy_records() { .expect("extract") .into_iter() .for_each(|e| h.edges.add(e)); - h.mount(spindle_did, "sh.tangled.pipeline", rkey, body).await; + h.mount(spindle_did, "sh.tangled.pipeline", rkey, body) + .await; let app = router(h.state.clone()); let (status, json) = json_response( diff --git a/crates/xrpc/tests/search.rs b/crates/xrpc/tests/search.rs index 5916f50..95b78f7 100644 --- a/crates/xrpc/tests/search.rs +++ b/crates/xrpc/tests/search.rs @@ -340,10 +340,7 @@ async fn pagination_round_trips_via_cursor() { async fn empty_q_returns_400() { let h = Harness::new().await; let app = router(h.state.clone()); - let resp = app - .oneshot(search_request(&[("q", " ")])) - .await - .unwrap(); + let resp = app.oneshot(search_request(&[("q", " ")])).await.unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_REQUEST); assert_eq!(body["error"], json!("InvalidRequest")); @@ -393,10 +390,7 @@ async fn tombstoned_hit_silently_dropped_from_results() { h.index_issue("did:plc:teq", "i2", "kelp survives", "still here") .await; let app = router(h.state.clone()); - let resp = app - .oneshot(search_request(&[("q", "kelp")])) - .await - .unwrap(); + let resp = app.oneshot(search_request(&[("q", "kelp")])).await.unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::OK); let hits = body["hits"].as_array().expect("hits array");