#[cfg(feature = "indexer")]
use super::unavailable;
use crate::control::Hydrant;
use axum::extract::FromRequest;
use axum::response::IntoResponse;
use axum::routing::{get, post};
use axum::{Json, Router, extract::State, http::StatusCode};
#[cfg(feature = "indexer")]
use jacquard_common::types::ident::AtIdentifier;
#[cfg(feature = "indexer")]
use jacquard_common::types::string::AtUri;
use serde::Deserialize;
use serde::Serialize;
use smol_str::{SmolStr, ToSmolStr};
use std::fmt::Display;
use std::result::Result;
mod get_host_status;
mod get_latest_commit;
mod get_repo_status;
mod list_hosts;
mod list_repos;
mod request_crawl;
#[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 subscribe_repos;
pub fn router(
#[cfg_attr(not(feature = "indexer"), allow(unused_variables))] record_bodies_available: bool,
) -> Router<Hydrant> {
let r = Router::new()
.route("/xrpc/com.atproto.sync.getHostStatus", get(get_host_status::handle))
.route("/xrpc/com.atproto.sync.listHosts", get(list_hosts::handle))
.route("/xrpc/com.atproto.sync.getLatestCommit", get(get_latest_commit::handle))
.route("/xrpc/com.atproto.sync.getRepoStatus", get(get_repo_status::handle))
.route("/xrpc/com.atproto.sync.listRepos", get(list_repos::handle))
.route(request_crawl::PATH, post(request_crawl::handle));
#[cfg(feature = "indexer")]
let r = r
.route("/xrpc/systems.gaze.hydrant.countRecords", get(count_records::handle))
.route("/xrpc/systems.gaze.hydrant.describeRepo", get(describe_repo::handle))
.route("/xrpc/com.bad-example.identity.resolveMiniDoc", get(resolve_mini_doc::handle))
.route("/xrpc/blue.microcosm.identity.resolveMiniDoc", get(resolve_mini_doc::handle))
.route("/xrpc/com.atproto.repo.describeRepo", get(com_atproto_describe_repo::handle));
#[cfg(feature = "indexer")]
let r = if record_bodies_available {
r.route("/xrpc/com.atproto.repo.getRecord", get(get_record::handle))
.route("/xrpc/blue.microcosm.repo.getRecordByUri", get(get_record_by_uri::handle))
.route("/xrpc/com.bad-example.repo.getUriRecord", get(get_record_by_uri::handle))
.route("/xrpc/com.atproto.repo.listRecords", get(list_records::handle))
.route("/xrpc/com.atproto.sync.getRepo", get(get_repo::handle))
} else {
r
};
#[cfg(feature = "relay")]
let r = r.route(subscribe_repos::PATH, get(subscribe_repos::handle));
r
}
#[derive(Debug, Serialize)]
pub struct XrpcErrorBody {
pub error: &'static str,
#[serde(skip_serializing_if = "Option::is_none")]
pub message: Option<SmolStr>,
}
#[derive(Debug)]
pub struct XrpcErrorResponse {
pub status: StatusCode,
pub error: XrpcErrorBody,
}
impl IntoResponse for XrpcErrorResponse {
fn into_response(self) -> axum::response::Response {
(self.status, Json(self.error)).into_response()
}
}
pub type XrpcResult<T> = Result<T, XrpcErrorResponse>;
const XRPC_PROCEDURE_BODY_LIMIT: usize = 64 * 1024; // 64 KiB
pub struct ExtractXrpc<T>(pub T);
impl<S, T> FromRequest<S> for ExtractXrpc<T>
where
S: Send + Sync,
T: serde::de::DeserializeOwned + Send,
{
type Rejection = XrpcErrorResponse;
async fn from_request(
req: axum::extract::Request,
_state: &S,
) -> Result<Self, Self::Rejection> {
if req.method() == axum::http::Method::POST {
let body = axum::body::to_bytes(req.into_body(), XRPC_PROCEDURE_BODY_LIMIT)
.await
.map_err(payload_too_large)?;
let res: T = serde_json::from_slice(&body).map_err(bad_request)?;
Ok(ExtractXrpc(res))
} else {
let query = req.uri().query().unwrap_or("");
let res: T = serde_html_form::from_str(query).map_err(bad_request)?;
Ok(ExtractXrpc(res))
}
}
}
fn xrpc_error(
status: StatusCode,
error: &'static str,
message: Option<SmolStr>,
) -> XrpcErrorResponse {
XrpcErrorResponse {
status,
error: XrpcErrorBody { error, message },
}
}
fn internal_error(message: impl Display) -> XrpcErrorResponse {
xrpc_error(
StatusCode::INTERNAL_SERVER_ERROR,
"InternalError",
Some(message.to_smolstr()),
)
}
fn bad_request(message: impl Display) -> XrpcErrorResponse {
xrpc_error(
StatusCode::BAD_REQUEST,
"InvalidRequest",
Some(message.to_smolstr()),
)
}
fn payload_too_large(message: impl Display) -> XrpcErrorResponse {
xrpc_error(
StatusCode::PAYLOAD_TOO_LARGE,
"PayloadTooLarge",
Some(message.to_smolstr()),
)
}
fn rate_limited(message: impl Display) -> XrpcErrorResponse {
xrpc_error(
StatusCode::TOO_MANY_REQUESTS,
"RateLimitExceeded",
Some(message.to_smolstr()),
)
}
/// getRecord's lexicon only knows RecordNotFound, so an inactive repo's record is that, with
/// the reason in the message
#[cfg(feature = "indexer")]
fn record_unavailable(reason: &'static str) -> XrpcErrorResponse {
xrpc_error(StatusCode::NOT_FOUND, "RecordNotFound", Some(reason.into()))
}
#[cfg(feature = "indexer")]
fn upstream_error(message: impl Display) -> XrpcErrorResponse {
xrpc_error(
StatusCode::BAD_GATEWAY,
"UpstreamError",
Some(message.to_smolstr()),
)
}
#[cfg(all(test, feature = "indexer"))]
mod tests {
use super::*;
use crate::config::Config;
use crate::db::{self, keys, types::DbRkey};
use crate::types::{RepoState, RepoStatus};
use jacquard_common::types::string::Did;
use miette::IntoDiagnostic;
const DID: &str = "did:plc:ewvi7nxzyoun6zhxrhs64oiz";
const COL: &str = "app.bsky.feed.post";
const RKEY: &str = "3kzbif5moe22m";
/// a hydrant holding one record of a repo whose row says `status`, like a deleted account
/// whose wipe hasn't run yet
async fn one_record(status: RepoStatus) -> miette::Result<(tempfile::TempDir, Hydrant)> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let hydrant = Hydrant::new(Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})
.await?;
let did = Did::new_static(DID).unwrap();
let db = &hydrant.state.db;
let mut repo = RepoState::backfilling();
repo.active = status == RepoStatus::Synced;
repo.status = status;
let body =
serde_ipld_dagcbor::to_vec(&serde_json::json!({"$type": COL, "text": "hi"})).unwrap();
let mut batch = db.inner.batch();
batch.insert(&db.repos, keys::repo_key(&did), db::ser_repo_state(&repo)?);
db.indexer.stage_head(
&mut batch,
keys::record_key(&did, COL, &DbRkey::new(RKEY)),
body,
);
db::set_record_count(&mut batch, db, &did, COL, 1);
batch.commit().into_diagnostic()?;
Ok((tmp, hydrant))
}
async fn query<T: serde::de::DeserializeOwned + Send>(query: &str) -> ExtractXrpc<T> {
let request = axum::extract::Request::builder()
.uri(format!("/xrpc/test?{query}"))
.body(axum::body::Body::empty())
.unwrap();
ExtractXrpc::<T>::from_request(request, &()).await.unwrap()
}
/// what each record-serving endpoint answers for the record, as (endpoint, status, body)
async fn answers(hydrant: &Hydrant) -> Vec<(&'static str, StatusCode, String)> {
fn answer<T: Serialize>(
name: &'static str,
result: XrpcResult<Json<T>>,
) -> (&'static str, StatusCode, String) {
match result {
Ok(Json(out)) => (name, StatusCode::OK, serde_json::to_string(&out).unwrap()),
Err(e) => (name, e.status, serde_json::to_string(&e.error).unwrap()),
}
}
let state = || State(hydrant.clone());
let record = format!("repo={DID}&collection={COL}&rkey={RKEY}");
let uri = format!("at_uri=at://{DID}/{COL}/{RKEY}");
let repo = format!("repo={DID}&collection={COL}");
let count = format!("identifier={DID}&collection={COL}");
vec![
answer(
"getRecord",
get_record::handle(state(), query(&record).await).await,
),
answer(
"getRecordByUri",
get_record_by_uri::handle(state(), query(&uri).await).await,
),
answer(
"listRecords",
list_records::handle(state(), query(&repo).await).await,
),
answer(
"countRecords",
count_records::handle(state(), query(&count).await).await,
),
]
}
#[tokio::test]
async fn an_active_repos_record_is_served() -> miette::Result<()> {
let (_tmp, hydrant) = one_record(RepoStatus::Synced).await?;
for (endpoint, status, body) in answers(&hydrant).await {
assert_eq!(status, StatusCode::OK, "{endpoint}: {body}");
assert!(
body.contains("hi") || body.contains("1"),
"{endpoint}: {body}"
);
}
Ok(())
}
#[tokio::test]
async fn an_inactive_repos_record_is_not_served() -> miette::Result<()> {
for status in [
RepoStatus::Deleted,
RepoStatus::Takendown,
RepoStatus::Suspended,
RepoStatus::Deactivated,
] {
let (_tmp, hydrant) = one_record(status.clone()).await?;
for (endpoint, code, body) in answers(&hydrant).await {
assert_ne!(code, StatusCode::OK, "{status:?} {endpoint}: {body}");
assert!(body.contains("repo is"), "{status:?} {endpoint}: {body}");
}
// the describe endpoints refuse before they resolve anything over the network
let describe = com_atproto_describe_repo::handle(
State(hydrant.clone()),
query(&format!("repo={DID}")).await,
)
.await;
let Err(err) = describe else {
panic!("describeRepo answered");
};
assert!(
serde_json::to_string(&err.error)
.unwrap()
.contains("repo is")
);
let describe = describe_repo::handle(
State(hydrant.clone()),
query(&format!("identifier={DID}")).await,
)
.await;
let Err(err) = describe else {
panic!("describeRepo answered");
};
assert!(
serde_json::to_string(&err.error)
.unwrap()
.contains("repo is")
);
}
Ok(())
}
}