From b71d5223dc251a74159b415c3d3abdf1f62f6435 Mon Sep 17 00:00:00 2001 From: dawn <90008@gaze.systems> Date: Wed, 25 Mar 2026 18:24:27 +0000 Subject: [PATCH] [api] refactor xrpcs into modules --- src/api/debug.rs | 6 +++--- src/api/xrpc.rs | 238 ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- src/api/xrpc/count_records.rs | 51 +++++++++++++++++++++++++++++++++++++++++++++++++++ src/api/xrpc/get_record.rs | 33 +++++++++++++++++++++++++++++++++ src/api/xrpc/list_records.rs | 45 +++++++++++++++++++++++++++++++++++++++++++++ src/api/xrpc/mod.rs | 112 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 6 file(s) changed, 244 insertion(s)(+), 241 deletion(s)(-) diff --git a/src/api/debug.rs b/src/api/debug.rs --- a/src/api/debug.rs +++ b/src/api/debug.rs @@ -287,9 +287,9 @@ let state_clone = state.clone(); tokio::task::spawn_blocking(move || { - let _ = ks.remove(b"dummy_tombstone123"); - let _ = state_clone.db.persist(); - let _ = ks.rotate_memtable_and_wait(); + ks.remove(b"dummy_tombstone123")?; + state_clone.db.inner.persist(fjall::PersistMode::Buffer)?; + ks.rotate_memtable_and_wait()?; ks.major_compact() }) .await diff --git a/src/api/xrpc.rs b/src/api/xrpc.rs deleted file mode 100644 --- a/src/api/xrpc.rs +++ /dev/null @@ -1,238 +0,0 @@ -use crate::control::Hydrant; -use axum::extract::FromRequest; -use axum::response::IntoResponse; -use axum::{Json, Router, extract::State, http::StatusCode}; -use jacquard_api::com_atproto::repo::{ - get_record::{GetRecordError, GetRecordOutput, GetRecordRequest}, - list_records::{ListRecordsOutput, ListRecordsRequest, Record as RepoRecord}, -}; -use jacquard_common::types::ident::AtIdentifier; -use jacquard_common::xrpc::{XrpcEndpoint, XrpcMethod}; -use jacquard_common::{IntoStatic, xrpc::XrpcRequest}; -use jacquard_common::{ - types::string::AtUri, - xrpc::{GenericXrpcError, XrpcError}, -}; -use serde::{Deserialize, Serialize}; -use smol_str::ToSmolStr; -use std::fmt::Display; - -pub fn router() -> Router { - Router::new() - .route( - GetRecordRequest::PATH, - axum::routing::get(handle_get_record), - ) - .route( - ListRecordsRequest::PATH, - axum::routing::get(handle_list_records), - ) - .route(CountRecords::PATH, axum::routing::get(handle_count_records)) -} - -#[derive(Debug)] -pub struct XrpcErrorResponse { - pub status: StatusCode, - pub error: XrpcError, -} - -impl IntoResponse for XrpcErrorResponse { - fn into_response(self) -> axum::response::Response { - (self.status, Json(self.error)).into_response() - } -} - -pub type XrpcResult = Result>; - -pub struct ExtractXrpc(pub E::Request<'static>); - -impl FromRequest for ExtractXrpc -where - S: Send + Sync, - E: XrpcEndpoint, - E::Request<'static>: Send, - for<'de> E::Request<'de>: Deserialize<'de> + IntoStatic>, -{ - type Rejection = XrpcErrorResponse; - - async fn from_request( - req: axum::extract::Request, - _state: &S, - ) -> Result { - let nsid = E::Request::<'static>::NSID; - match E::METHOD { - XrpcMethod::Query => { - let query = req.uri().query().unwrap_or(""); - let res: E::Request<'_> = - serde_urlencoded::from_str(query).map_err(|e| bad_request(nsid, e))?; - Ok(ExtractXrpc(res.into_static())) - } - XrpcMethod::Procedure(_) => { - let body = axum::body::to_bytes(req.into_body(), usize::MAX) - .await - .map_err(|e| internal_error(nsid, e))?; - let res: E::Request<'_> = - serde_json::from_slice(&body).map_err(|e| bad_request(nsid, e))?; - Ok(ExtractXrpc(res.into_static())) - } - } - } -} - -fn internal_error( - nsid: &'static str, - message: impl Display, -) -> XrpcErrorResponse { - XrpcErrorResponse { - status: StatusCode::INTERNAL_SERVER_ERROR, - error: XrpcError::Generic(GenericXrpcError { - error: "InternalError".into(), - message: Some(message.to_smolstr()), - nsid, - method: "GET", - http_status: StatusCode::INTERNAL_SERVER_ERROR, - }), - } -} - -fn bad_request( - nsid: &'static str, - message: impl Display, -) -> XrpcErrorResponse { - XrpcErrorResponse { - status: StatusCode::BAD_REQUEST, - error: XrpcError::Generic(GenericXrpcError { - error: "InvalidRequest".into(), - message: Some(message.to_smolstr()), - nsid, - method: "GET", - http_status: StatusCode::BAD_REQUEST, - }), - } -} - -pub async fn handle_get_record( - State(hydrant): State, - ExtractXrpc(req): ExtractXrpc, -) -> Result>, XrpcErrorResponse>> { - let record = hydrant - .repos - .resolve(&req.repo) - .await - .map_err(|e| internal_error(GetRecordRequest::PATH, e))? - .get_record(&req.collection, &req.rkey.0) - .await - .map_err(|e| internal_error(GetRecordRequest::PATH, e))?; - let Some(record) = record else { - return Err(XrpcErrorResponse { - status: StatusCode::NOT_FOUND, - error: XrpcError::Xrpc(GetRecordError::RecordNotFound(None)), - }); - }; - - Ok(Json(GetRecordOutput { - uri: AtUri::from_parts_owned( - record.did.as_str(), - req.collection.as_str(), - req.rkey.0.as_str(), - ) - .unwrap(), - cid: Some(record.cid), - value: record.value, - extra_data: Default::default(), - })) -} - -pub async fn handle_list_records( - State(hydrant): State, - ExtractXrpc(req): ExtractXrpc, -) -> Result>, XrpcErrorResponse> { - let limit = req.limit.unwrap_or(50).min(100) as usize; - let reverse = req.reverse.unwrap_or(false); - let cursor = req.cursor.as_deref(); - - let repo = hydrant - .repos - .resolve(&req.repo) - .await - .map_err(|e| internal_error(GetRecordRequest::PATH, e))?; - let list = repo - .list_records(req.collection.as_str(), limit, reverse, cursor) - .await - .map_err(|e| bad_request(ListRecordsRequest::PATH, e))?; - - let records = list - .records - .into_iter() - .filter_map(|r| { - let uri = AtUri::from_parts_owned( - repo.did.as_str(), - req.collection.as_str(), - r.rkey.as_str(), - ) - .ok()?; - Some(RepoRecord { - uri, - cid: r.cid, - value: r.value, - extra_data: Default::default(), - }) - }) - .collect(); - - Ok(Json(ListRecordsOutput { - records, - cursor: list.cursor.map(|r| r.into()), - extra_data: Default::default(), - })) -} - -#[derive(Serialize, Deserialize, jacquard_derive::IntoStatic)] -pub struct CountRecordsOutput { - pub count: u64, -} - -pub struct CountRecordsResponse; -impl jacquard_common::xrpc::XrpcResp for CountRecordsResponse { - const NSID: &'static str = "systems.gaze.hydrant.countRecords"; - const ENCODING: &'static str = "application/json"; - type Output<'de> = CountRecordsOutput; - type Err<'de> = GenericXrpcError; -} - -#[derive(Serialize, Deserialize, jacquard_derive::IntoStatic)] -pub struct CountRecordsRequestData<'i> { - #[serde(borrow)] - pub identifier: AtIdentifier<'i>, - pub collection: String, -} - -impl<'a> jacquard_common::xrpc::XrpcRequest for CountRecordsRequestData<'a> { - const NSID: &'static str = "systems.gaze.hydrant.countRecords"; - const METHOD: jacquard_common::xrpc::XrpcMethod = jacquard_common::xrpc::XrpcMethod::Query; - type Response = CountRecordsResponse; -} - -pub struct CountRecords; -impl jacquard_common::xrpc::XrpcEndpoint for CountRecords { - const PATH: &'static str = "/xrpc/systems.gaze.hydrant.countRecords"; - const METHOD: jacquard_common::xrpc::XrpcMethod = jacquard_common::xrpc::XrpcMethod::Query; - type Request<'de> = CountRecordsRequestData<'de>; - type Response = CountRecordsResponse; -} - -pub async fn handle_count_records( - State(hydrant): State, - ExtractXrpc(req): ExtractXrpc, -) -> XrpcResult> { - let count = hydrant - .repos - .resolve(&req.identifier) - .await - .map_err(|e| internal_error(GetRecordRequest::PATH, e))? - .count_records(&req.collection) - .await - .map_err(|e| internal_error(CountRecords::PATH, e))?; - - Ok(Json(CountRecordsOutput { count })) -} diff --git a/src/api/xrpc/count_records.rs b/src/api/xrpc/count_records.rs new file mode 100644 --- /dev/null +++ b/src/api/xrpc/count_records.rs @@ -0,0 +1,51 @@ +use super::*; + +#[derive(Serialize, Deserialize, jacquard_derive::IntoStatic)] +pub struct CountRecordsOutput { + pub count: u64, +} + +pub struct CountRecordsResponse; +impl jacquard_common::xrpc::XrpcResp for CountRecordsResponse { + const NSID: &'static str = "systems.gaze.hydrant.countRecords"; + const ENCODING: &'static str = "application/json"; + type Output<'de> = CountRecordsOutput; + type Err<'de> = GenericXrpcError; +} + +#[derive(Serialize, Deserialize, jacquard_derive::IntoStatic)] +pub struct CountRecordsRequestData<'i> { + #[serde(borrow)] + pub identifier: AtIdentifier<'i>, + pub collection: String, +} + +impl<'a> jacquard_common::xrpc::XrpcRequest for CountRecordsRequestData<'a> { + const NSID: &'static str = "systems.gaze.hydrant.countRecords"; + const METHOD: jacquard_common::xrpc::XrpcMethod = jacquard_common::xrpc::XrpcMethod::Query; + type Response = CountRecordsResponse; +} + +pub struct CountRecords; +impl jacquard_common::xrpc::XrpcEndpoint for CountRecords { + const PATH: &'static str = "/xrpc/systems.gaze.hydrant.countRecords"; + const METHOD: jacquard_common::xrpc::XrpcMethod = jacquard_common::xrpc::XrpcMethod::Query; + type Request<'de> = CountRecordsRequestData<'de>; + type Response = CountRecordsResponse; +} + +pub async fn handle( + State(hydrant): State, + ExtractXrpc(req): ExtractXrpc, +) -> XrpcResult> { + let count = hydrant + .repos + .resolve(&req.identifier) + .await + .map_err(|e| internal_error(GetRecordRequest::PATH, e))? + .count_records(&req.collection) + .await + .map_err(|e| internal_error(CountRecords::PATH, e))?; + + Ok(Json(CountRecordsOutput { count })) +} diff --git a/src/api/xrpc/get_record.rs b/src/api/xrpc/get_record.rs new file mode 100644 --- /dev/null +++ b/src/api/xrpc/get_record.rs @@ -0,0 +1,33 @@ +use super::*; + +pub async fn handle( + State(hydrant): State, + ExtractXrpc(req): ExtractXrpc, +) -> Result>, XrpcErrorResponse>> { + let record = hydrant + .repos + .resolve(&req.repo) + .await + .map_err(|e| internal_error(GetRecordRequest::PATH, e))? + .get_record(&req.collection, &req.rkey.0) + .await + .map_err(|e| internal_error(GetRecordRequest::PATH, e))?; + let Some(record) = record else { + return Err(XrpcErrorResponse { + status: StatusCode::NOT_FOUND, + error: XrpcError::Xrpc(GetRecordError::RecordNotFound(None)), + }); + }; + + Ok(Json(GetRecordOutput { + uri: AtUri::from_parts_owned( + record.did.as_str(), + req.collection.as_str(), + req.rkey.0.as_str(), + ) + .unwrap(), + cid: Some(record.cid), + value: record.value, + extra_data: Default::default(), + })) +} diff --git a/src/api/xrpc/list_records.rs b/src/api/xrpc/list_records.rs new file mode 100644 --- /dev/null +++ b/src/api/xrpc/list_records.rs @@ -0,0 +1,45 @@ +use super::*; + +pub async fn handle( + State(hydrant): State, + ExtractXrpc(req): ExtractXrpc, +) -> Result>, XrpcErrorResponse> { + let limit = req.limit.unwrap_or(50).min(100) as usize; + let reverse = req.reverse.unwrap_or(false); + let cursor = req.cursor.as_deref(); + + let repo = hydrant + .repos + .resolve(&req.repo) + .await + .map_err(|e| internal_error(GetRecordRequest::PATH, e))?; + let list = repo + .list_records(req.collection.as_str(), limit, reverse, cursor) + .await + .map_err(|e| bad_request(ListRecordsRequest::PATH, e))?; + + let records = list + .records + .into_iter() + .filter_map(|r| { + let uri = AtUri::from_parts_owned( + repo.did.as_str(), + req.collection.as_str(), + r.rkey.as_str(), + ) + .ok()?; + Some(RepoRecord { + uri, + cid: r.cid, + value: r.value, + extra_data: Default::default(), + }) + }) + .collect(); + + Ok(Json(ListRecordsOutput { + records, + cursor: list.cursor.map(|r| r.into()), + extra_data: Default::default(), + })) +} diff --git a/src/api/xrpc/mod.rs b/src/api/xrpc/mod.rs new file mode 100644 --- /dev/null +++ b/src/api/xrpc/mod.rs @@ -0,0 +1,112 @@ +use crate::api::xrpc::count_records::CountRecords; +use crate::control::Hydrant; +use axum::extract::FromRequest; +use axum::response::IntoResponse; +use axum::routing::get; +use axum::{Json, Router, extract::State, http::StatusCode}; +use jacquard_api::com_atproto::repo::{ + get_record::{GetRecordError, GetRecordOutput, GetRecordRequest}, + list_records::{ListRecordsOutput, ListRecordsRequest, Record as RepoRecord}, +}; +use jacquard_common::types::ident::AtIdentifier; +use jacquard_common::xrpc::{XrpcEndpoint, XrpcMethod}; +use jacquard_common::{IntoStatic, xrpc::XrpcRequest}; +use jacquard_common::{ + types::string::AtUri, + xrpc::{GenericXrpcError, XrpcError}, +}; +use serde::{Deserialize, Serialize}; +use smol_str::ToSmolStr; +use std::fmt::Display; + +mod count_records; +mod get_record; +mod list_records; + +pub fn router() -> Router { + Router::new() + .route(GetRecordRequest::PATH, get(get_record::handle)) + .route(ListRecordsRequest::PATH, get(list_records::handle)) + .route(CountRecords::PATH, get(count_records::handle)) +} + +#[derive(Debug)] +pub struct XrpcErrorResponse { + pub status: StatusCode, + pub error: XrpcError, +} + +impl IntoResponse for XrpcErrorResponse { + fn into_response(self) -> axum::response::Response { + (self.status, Json(self.error)).into_response() + } +} + +pub type XrpcResult = Result>; + +pub struct ExtractXrpc(pub E::Request<'static>); + +impl FromRequest for ExtractXrpc +where + S: Send + Sync, + E: XrpcEndpoint, + E::Request<'static>: Send, + for<'de> E::Request<'de>: Deserialize<'de> + IntoStatic>, +{ + type Rejection = XrpcErrorResponse; + + async fn from_request( + req: axum::extract::Request, + _state: &S, + ) -> Result { + let nsid = E::Request::<'static>::NSID; + match E::METHOD { + XrpcMethod::Query => { + let query = req.uri().query().unwrap_or(""); + let res: E::Request<'_> = + serde_urlencoded::from_str(query).map_err(|e| bad_request(nsid, e))?; + Ok(ExtractXrpc(res.into_static())) + } + XrpcMethod::Procedure(_) => { + let body = axum::body::to_bytes(req.into_body(), usize::MAX) + .await + .map_err(|e| internal_error(nsid, e))?; + let res: E::Request<'_> = + serde_json::from_slice(&body).map_err(|e| bad_request(nsid, e))?; + Ok(ExtractXrpc(res.into_static())) + } + } + } +} + +fn internal_error( + nsid: &'static str, + message: impl Display, +) -> XrpcErrorResponse { + XrpcErrorResponse { + status: StatusCode::INTERNAL_SERVER_ERROR, + error: XrpcError::Generic(GenericXrpcError { + error: "InternalError".into(), + message: Some(message.to_smolstr()), + nsid, + method: "GET", + http_status: StatusCode::INTERNAL_SERVER_ERROR, + }), + } +} + +fn bad_request( + nsid: &'static str, + message: impl Display, +) -> XrpcErrorResponse { + XrpcErrorResponse { + status: StatusCode::BAD_REQUEST, + error: XrpcError::Generic(GenericXrpcError { + error: "InvalidRequest".into(), + message: Some(message.to_smolstr()), + nsid, + method: "GET", + http_status: StatusCode::BAD_REQUEST, + }), + } +} -- tangled.sh