Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257use crate::control::Hydrant;use axum::extract::FromRequest;use axum::response::IntoResponse;use axum::routing::get;#[cfg(feature = "relay")]use axum::routing::post;use axum::{Json, Router, extract::State, http::StatusCode};use jacquard_api::com_atproto::sync::get_host_status::GetHostStatusRequest;use jacquard_api::com_atproto::sync::get_latest_commit::GetLatestCommitRequest;use jacquard_api::com_atproto::sync::get_repo_status::GetRepoStatusRequest;use jacquard_api::com_atproto::sync::list_hosts::ListHostsRequest;use jacquard_api::com_atproto::sync::list_repos::ListReposRequest;#[cfg(feature = "indexer")]use jacquard_common::types::ident::AtIdentifier;#[cfg(feature = "indexer")]use jacquard_common::types::string::AtUri;use jacquard_common::xrpc::XrpcRequest;use jacquard_common::xrpc::XrpcResp;use jacquard_common::xrpc::{GenericXrpcError, XrpcError};use jacquard_common::xrpc::{XrpcEndpoint, XrpcMethod};#[cfg(feature = "indexer")]use serde::Deserialize;use serde::Serialize;use smol_str::ToSmolStr;use std::fmt::Display;use std::result::Result;#[cfg(feature = "indexer")]use { crate::api::xrpc::count_records::CountRecords, crate::api::xrpc::describe_repo::DescribeRepo, jacquard_api::com_atproto::repo::{ describe_repo::DescribeRepoRequest as AtprotoDescribeRepoRequest, get_record::{GetRecordError, GetRecordOutput, GetRecordRequest}, list_records::{ListRecordsOutput, ListRecordsRequest, Record as RepoRecord}, }, jacquard_api::com_atproto::sync::get_repo::GetRepoRequest,};#[cfg(feature = "relay")]use { jacquard_api::com_atproto::sync::request_crawl::RequestCrawlRequest, jacquard_api::com_atproto::sync::subscribe_repos::SubscribeReposEndpoint, jacquard_common::xrpc::SubscriptionEndpoint,};
mod get_host_status;mod get_latest_commit;mod get_repo_status;mod list_hosts;mod list_repos;
#[cfg(feature = "indexer")]mod com_atproto_describe_repo;#[cfg(feature = "indexer")]mod count_records;#[cfg(feature = "indexer")]mod describe_repo;#[cfg(feature = "indexer")]mod get_record;#[cfg(feature = "indexer")]mod get_record_by_uri;#[cfg(feature = "indexer")]mod get_repo;#[cfg(feature = "indexer")]mod list_records;#[cfg(feature = "indexer")]mod resolve_mini_doc;
#[cfg(feature = "relay")]mod request_crawl;#[cfg(feature = "relay")]mod subscribe_repos;
pub fn router( #[cfg_attr(not(feature = "indexer"), allow(unused_variables))] record_bodies_available: bool,) -> Router<Hydrant> { let r = Router::new() .route(GetHostStatusRequest::PATH, get(get_host_status::handle)) .route(ListHostsRequest::PATH, get(list_hosts::handle)) .route(GetLatestCommitRequest::PATH, get(get_latest_commit::handle)) .route(GetRepoStatusRequest::PATH, get(get_repo_status::handle)) .route(ListReposRequest::PATH, get(list_repos::handle));
#[cfg(feature = "indexer")] let r = r .route(CountRecords::PATH, get(count_records::handle)) .route(DescribeRepo::PATH, get(describe_repo::handle)) .route( resolve_mini_doc::BadExampleResolveMiniDoc::PATH, get(resolve_mini_doc::handle_bad_example), ) .route( resolve_mini_doc::BlueMicrocosmResolveMiniDoc::PATH, get(resolve_mini_doc::handle_blue_microcosm), ) .route( AtprotoDescribeRepoRequest::PATH, get(com_atproto_describe_repo::handle), );
#[cfg(feature = "indexer")] let r = if record_bodies_available { r.route(GetRecordRequest::PATH, get(get_record::handle)) .route( get_record_by_uri::BlueMicrocosmGetRecordByUri::PATH, get(get_record_by_uri::handle_blue_microcosm), ) .route( get_record_by_uri::BadExampleGetUriRecord::PATH, get(get_record_by_uri::handle_bad_example), ) .route(ListRecordsRequest::PATH, get(list_records::handle)) .route(GetRepoRequest::PATH, get(get_repo::handle)) } else { r };
#[cfg(feature = "relay")] let r = r .route(SubscribeReposEndpoint::PATH, get(subscribe_repos::handle)) .route(RequestCrawlRequest::PATH, post(request_crawl::handle));
r}
#[derive(Debug)]pub struct XrpcErrorResponse<E: std::error::Error = GenericXrpcError> { pub status: StatusCode, pub error: XrpcError<E>,}
impl<E: Serialize + std::error::Error> IntoResponse for XrpcErrorResponse<E> { fn into_response(self) -> axum::response::Response { (self.status, Json(self.error)).into_response() }}
pub type XrpcResult<T, E = GenericXrpcError> = Result<T, XrpcErrorResponse<E>>;
const XRPC_PROCEDURE_BODY_LIMIT: usize = 64 * 1024; // 64 KiB
pub struct ExtractXrpc<E: XrpcEndpoint>(pub E::Request<jacquard_common::DefaultStr>);
impl<S, E> FromRequest<S> for ExtractXrpc<E>where S: Send + Sync, E: XrpcEndpoint, E::Request<jacquard_common::DefaultStr>: Send + serde::de::DeserializeOwned,{ type Rejection = XrpcErrorResponse<GenericXrpcError>;
async fn from_request( req: axum::extract::Request, _state: &S, ) -> Result<Self, Self::Rejection> { let nsid = E::Request::<jacquard_common::DefaultStr>::NSID; match E::METHOD { XrpcMethod::Query => { let query = req.uri().query().unwrap_or(""); let res: E::Request<jacquard_common::DefaultStr> = serde_html_form::from_str(query).map_err(|e| bad_request(nsid, e))?; Ok(ExtractXrpc(res)) } XrpcMethod::Procedure(_) => { let body = axum::body::to_bytes(req.into_body(), XRPC_PROCEDURE_BODY_LIMIT) .await .map_err(|e| payload_too_large(nsid, e))?; let res: E::Request<jacquard_common::DefaultStr> = serde_json::from_slice(&body).map_err(|e| bad_request(nsid, e))?; Ok(ExtractXrpc(res)) } } }}
fn internal_error<E: std::error::Error>( nsid: &'static str, message: impl Display,) -> XrpcErrorResponse<E> { 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<E: std::error::Error>( nsid: &'static str, message: impl Display,) -> XrpcErrorResponse<E> { 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, }), }}
fn payload_too_large<E: std::error::Error>( nsid: &'static str, message: impl Display,) -> XrpcErrorResponse<E> { XrpcErrorResponse { status: StatusCode::PAYLOAD_TOO_LARGE, error: XrpcError::Generic(GenericXrpcError { error: "PayloadTooLarge".into(), message: Some(message.to_smolstr()), nsid, method: "POST", http_status: StatusCode::PAYLOAD_TOO_LARGE, }), }}
#[cfg(any(feature = "indexer", feature = "relay"))]fn rate_limited<E: std::error::Error>( nsid: &'static str, method: &'static str, message: impl Display,) -> XrpcErrorResponse<E> { XrpcErrorResponse { status: StatusCode::TOO_MANY_REQUESTS, error: XrpcError::Generic(GenericXrpcError { error: "RateLimitExceeded".into(), message: Some(message.to_smolstr()), nsid, method, http_status: StatusCode::TOO_MANY_REQUESTS, }), }}
#[cfg(feature = "indexer")]fn upstream_error<E: std::error::Error>( nsid: &'static str, message: impl Display,) -> XrpcErrorResponse<E> { XrpcErrorResponse { status: StatusCode::BAD_GATEWAY, error: XrpcError::Generic(GenericXrpcError { error: "UpstreamError".into(), message: Some(message.to_smolstr()), nsid, method: "GET", http_status: StatusCode::BAD_GATEWAY, }), }}