From 9a275c1bca705ea2992d8458d0616b6a274bf756 Mon Sep 17 00:00:00 2001 From: Lewis Date: Mon, 4 May 2026 23:04:12 +0300 Subject: [PATCH] feat(xrpc): sh.tangled.search.query route Lewis: May this revision serve well! --- crates/xrpc/Cargo.toml | 1 + crates/xrpc/src/lib.rs | 142 ++++++++++++++++++++++++++++--- crates/xrpc/tests/aggregation.rs | 2 + crates/xrpc/tests/cold_start.rs | 2 + crates/xrpc/tests/knot_proxy.rs | 2 + 5 files changed, 137 insertions(+), 12 deletions(-) diff --git a/crates/xrpc/Cargo.toml b/crates/xrpc/Cargo.toml index 2f30625..aa11299 100644 --- a/crates/xrpc/Cargo.toml +++ b/crates/xrpc/Cargo.toml @@ -9,6 +9,7 @@ rust-version.workspace = true bobbin-types = { workspace = true } bobbin-edge-index = { workspace = true } bobbin-record-lru = { workspace = true } +bobbin-search = { workspace = true } bobbin-slingshot-client = { workspace = true } bobbin-knot-proxy = { workspace = true } jacquard-common = { workspace = true } diff --git a/crates/xrpc/src/lib.rs b/crates/xrpc/src/lib.rs index ac78e74..b0c8bcc 100644 --- a/crates/xrpc/src/lib.rs +++ b/crates/xrpc/src/lib.rs @@ -22,9 +22,11 @@ use bobbin_edge_index::{ }; use bobbin_knot_proxy::{KnotHost, KnotProxy, KnotProxyError, ProxyResponse, RepoSlug}; use bobbin_record_lru::RecordStore; +use bobbin_search::{SearchCursor, SearchError, SearchHit, SearchIndex, SearchOffset}; use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; use bobbin_types::ids::{EdgeKey, nsid_static}; use bobbin_types::record::RecordBody; +use bobbin_types::search::SearchableRecord; use bobbin_types::sh_tangled::actor::profile::{Profile, ProfileGetRecordOutput, ProfileRecord}; use bobbin_types::sh_tangled::feed::star::{Star, StarRecord}; use bobbin_types::sh_tangled::graph::follow::{Follow, FollowRecord}; @@ -37,6 +39,7 @@ use bobbin_types::sh_tangled::repo::{Repo, RepoGetRecordOutput, RepoRecord}; use futures::stream::{self, StreamExt, TryStreamExt}; use jacquard_common::types::did::Did; use jacquard_common::types::ident::AtIdentifier; +use jacquard_common::types::nsid::Nsid; use jacquard_common::types::string::{AtUri, Cid}; use jacquard_common::xrpc::XrpcResp; use jacquard_common::{DefaultStr, IntoStatic}; @@ -53,6 +56,7 @@ pub struct AppState { pub edges: Arc, pub coverage: Arc, pub knots: Arc, + pub search: Arc, } impl AppState { @@ -62,6 +66,7 @@ impl AppState { edges: Arc, coverage: Arc, knots: Arc, + search: Arc, ) -> Self { Self { records, @@ -69,6 +74,7 @@ impl AppState { edges, coverage, knots, + search, } } } @@ -95,6 +101,7 @@ pub fn router(state: AppState) -> Router { "/xrpc/sh.tangled.repo.issue.countComments", get(count_issue_comments), ) + .route("/xrpc/sh.tangled.search.query", get(search_query)) .merge(knot_proxied_routes()) .with_state(state) } @@ -182,14 +189,14 @@ impl RawAtUriParam { } #[derive(Clone, Copy, Debug, Eq, PartialEq)] -pub struct ExpectedNsid(&'static str); +pub struct ExpectedNsid<'a>(&'a str); -impl ExpectedNsid { - pub const fn new(nsid: &'static str) -> Self { +impl<'a> ExpectedNsid<'a> { + pub const fn new(nsid: &'a str) -> Self { Self(nsid) } - pub const fn as_str(self) -> &'static str { + pub const fn as_str(self) -> &'a str { self.0 } } @@ -226,6 +233,14 @@ struct CountQuery { subject: RawAtUriParam, } +#[derive(Debug, Deserialize)] +struct SearchQueryParams { + q: String, + nsid: Option, + cursor: Option, + limit: Option, +} + pub struct XrpcQuery(pub T); impl FromRequestParts for XrpcQuery @@ -255,6 +270,8 @@ pub enum XrpcError { UpstreamGone(String), #[error("invalid record: {0}")] InvalidRecord(String), + #[error("internal: {0}")] + Internal(String), } #[derive(Serialize)] @@ -271,6 +288,7 @@ impl IntoResponse for XrpcError { Self::UpstreamUnavailable(_) => (StatusCode::BAD_GATEWAY, "UpstreamFailed"), Self::UpstreamGone(_) => (StatusCode::BAD_GATEWAY, "UpstreamGone"), Self::InvalidRecord(_) => (StatusCode::BAD_GATEWAY, "InvalidRecord"), + Self::Internal(_) => (StatusCode::INTERNAL_SERVER_ERROR, "InternalError"), }; let body = ErrorBody { error, @@ -322,6 +340,24 @@ struct CountResponse { distinct_authors: u64, } +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct SearchHitView { + uri: AtUri, + cid: Option>, + nsid: Nsid, + score: f32, + value: SearchableRecord, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct SearchResponse { + coverage: CoverageEnvelope, + hits: Vec, + cursor: Option, +} + fn map_slingshot(err: SlingshotError) -> XrpcError { use SlingshotError as E; match err { @@ -419,9 +455,24 @@ fn parse_limit(raw: Option) -> Result { .map_err(|e| XrpcError::InvalidParams(format!("limit: {e}"))) } +async fn resolve_for_view( + state: &AppState, + expected_nsid: &str, + uri: AtUri, +) -> Result, XrpcError> { + let raw = uri.as_ref().to_owned(); + resolve(state, ExpectedNsid::new(expected_nsid), uri) + .await + .map(|(body, _did)| body) + .map_err(|e| match e { + XrpcError::NotFound => XrpcError::UpstreamGone(raw), + other => other, + }) +} + async fn resolve( state: &AppState, - expected: ExpectedNsid, + expected: ExpectedNsid<'_>, uri: AtUri, ) -> Result<(Arc, Did), XrpcError> { let collection = uri @@ -466,7 +517,7 @@ struct TypeTag<'a> { ty: &'a str, } -fn verify_type_tag(body: &RecordBody, expected: ExpectedNsid) -> Result<(), XrpcError> { +fn verify_type_tag(body: &RecordBody, expected: ExpectedNsid<'_>) -> Result<(), XrpcError> { let tag: TypeTag<'_> = serde_json::from_slice(&body.value) .map_err(|e| XrpcError::InvalidRecord(format!("$type peek: {e}")))?; if tag.ty != expected.as_str() { @@ -555,12 +606,7 @@ where let EdgePage { items, next } = state.edges.list(&key, cursor, limit); let items = stream::iter(items) .map(|uri| async move { - let (body, _did) = resolve(state, ExpectedNsid::new(R::NSID), uri.clone()) - .await - .map_err(|e| match e { - XrpcError::NotFound => XrpcError::UpstreamGone(uri.as_ref().to_owned()), - other => other, - })?; + let body = resolve_for_view(state, R::NSID, uri).await?; let value: V = serde_json::from_slice(&body.value) .map_err(|e| XrpcError::InvalidRecord(e.to_string()))?; Ok::<_, XrpcError>(RecordView { @@ -665,6 +711,78 @@ async fn count_issue_comments( count_for::(&state, q).map(Json) } +async fn search_query( + State(state): State, + XrpcQuery(q): XrpcQuery, +) -> Result, XrpcError> { + if q.q.trim().is_empty() { + return Err(XrpcError::InvalidParams("q must not be empty".into())); + } + let cursor = SearchCursor::from_token(q.cursor.as_deref()) + .map_err(|e| XrpcError::InvalidParams(format!("cursor: {e}")))?; + let limit = parse_limit(q.limit)?; + let nsid_filter = q + .nsid + .as_deref() + .map(|s| { + Nsid::::new_owned(s) + .map_err(|e| XrpcError::InvalidParams(format!("nsid: {e}"))) + }) + .transpose()?; + let coverage = state.coverage.snapshot(); + let page = state + .search + .search(&q.q, nsid_filter.as_ref(), cursor, limit.get()) + .await + .map_err(map_search_err)?; + let state_ref = &state; + let hits: Vec = stream::iter(page.hits) + .map(|hit| hydrate_search_hit(state_ref, hit)) + .buffered(FETCH_CONCURRENCY) + .try_filter_map(|opt| async move { Ok(opt) }) + .try_collect() + .await?; + Ok(Json(SearchResponse { + coverage: coverage.into(), + hits, + cursor: page.next.map(SearchOffset::encode_token), + })) +} + +async fn hydrate_search_hit( + state: &AppState, + hit: SearchHit, +) -> Result, XrpcError> { + let SearchHit { uri, nsid, score } = hit; + let body = match resolve_for_view(state, nsid.as_ref(), uri).await { + Ok(b) => b, + Err(XrpcError::UpstreamGone(_)) => return Ok(None), + Err(other) => return Err(other), + }; + let value = SearchableRecord::from_json_bytes(nsid.as_ref(), &body.value) + .map_err(|e| XrpcError::InvalidRecord(e.to_string()))?; + Ok(Some(SearchHitView { + uri: body.uri.clone(), + cid: Some(body.cid.clone()), + nsid, + score, + value, + })) +} + +fn map_search_err(err: SearchError) -> XrpcError { + use SearchError as E; + match err { + E::Query(e) => XrpcError::InvalidParams(format!("query: {e}")), + e @ (E::Tantivy(_) + | E::InvalidUri(_) + | E::InvalidNsid(_) + | E::MissingField(_) + | E::ThreadSpawn(_) + | E::Cancelled(_)) => XrpcError::Internal(format!("search: {e}")), + } +} + fn map_proxy_error(err: KnotProxyError) -> XrpcError { match err { KnotProxyError::CircuitOpen => { diff --git a/crates/xrpc/tests/aggregation.rs b/crates/xrpc/tests/aggregation.rs index 610fefe..b811e2a 100644 --- a/crates/xrpc/tests/aggregation.rs +++ b/crates/xrpc/tests/aggregation.rs @@ -4,6 +4,7 @@ use axum::body::{Body, to_bytes}; use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor, SourceId}; use bobbin_knot_proxy::{KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; +use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex}; use bobbin_slingshot_client::SlingshotClient; use bobbin_types::edges::Edge; use bobbin_xrpc::{AppState, router}; @@ -47,6 +48,7 @@ impl Harness { edges.clone(), coverage.clone(), Arc::new(KnotProxy::new(KnotProxyConfig::default()).unwrap()), + Arc::new(SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES).unwrap()), ); Self { server, diff --git a/crates/xrpc/tests/cold_start.rs b/crates/xrpc/tests/cold_start.rs index 4474267..ca3b266 100644 --- a/crates/xrpc/tests/cold_start.rs +++ b/crates/xrpc/tests/cold_start.rs @@ -4,6 +4,7 @@ use axum::body::{Body, to_bytes}; use bobbin_edge_index::{CoverageWatch, EdgeStore}; use bobbin_knot_proxy::{KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; +use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex}; use bobbin_slingshot_client::SlingshotClient; use bobbin_xrpc::{AppState, router}; use futures::stream::{self, StreamExt}; @@ -24,6 +25,7 @@ async fn fresh_app(server_uri: &str) -> AppState { Arc::new(EdgeStore::new()), Arc::new(CoverageWatch::new()), Arc::new(KnotProxy::new(KnotProxyConfig::default()).unwrap()), + Arc::new(SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES).unwrap()), ) } diff --git a/crates/xrpc/tests/knot_proxy.rs b/crates/xrpc/tests/knot_proxy.rs index 4a679a2..b083cec 100644 --- a/crates/xrpc/tests/knot_proxy.rs +++ b/crates/xrpc/tests/knot_proxy.rs @@ -5,6 +5,7 @@ use axum::body::{Body, to_bytes}; use bobbin_edge_index::{CoverageWatch, EdgeStore}; use bobbin_knot_proxy::{FailureThreshold, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; +use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex}; use bobbin_slingshot_client::SlingshotClient; use bobbin_xrpc::{AppState, router}; use http::{Request, StatusCode}; @@ -48,6 +49,7 @@ impl Harness { Arc::new(EdgeStore::new()), Arc::new(CoverageWatch::new()), Arc::new(KnotProxy::new(config).unwrap()), + Arc::new(SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES).unwrap()), ); Self { slingshot: slingshot_server, -- 2.51.2