Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176use axum::{ Json, Router, extract::{Query, State}, http::StatusCode, routing::{delete, get, post},};use serde::Deserialize;use url::Url;
use crate::control::{FirehoseSourceInfo, Hydrant};
pub fn router() -> Router<Hydrant> { let router = Router::new() .route("/firehose/source", get(get_source)) .route("/firehose/sources", get(list_sources)) .route("/firehose/sources", post(add_source)) .route("/firehose/sources", delete(remove_source)) .route("/firehose/cursors", delete(reset_cursor));
#[cfg(feature = "firehose-diagnostics")] let router = router.route("/firehose/diagnostics", get(get_diagnostics));
router}
#[derive(Debug, Deserialize)]pub struct GetSourceQuery { pub url: Option<String>, pub host: Option<String>,}
#[derive(Debug, Default, Deserialize)]pub struct ListSourcesQuery { pub host: Option<String>, pub url: Option<String>, pub is_pds: Option<bool>, pub running: Option<bool>, pub failing: Option<bool>, pub throttled: Option<bool>,}
pub async fn get_source( State(hydrant): State<Hydrant>, Query(query): Query<GetSourceQuery>,) -> Result<Json<FirehoseSourceInfo>, (StatusCode, String)> { let sources = hydrant.firehose.list_sources().await; let source = match (query.url.as_deref(), query.host.as_deref()) { (Some(url), None) => sources .into_iter() .find(|source| source.url.as_str() == url) .ok_or_else(|| (StatusCode::NOT_FOUND, "source does not exist".to_string()))?, (None, Some(host)) => { let mut matches = sources .into_iter() .filter(|source| source.url.host_str() == Some(host)); let Some(source) = matches.next() else { return Err((StatusCode::NOT_FOUND, "source does not exist".to_string())); }; if matches.next().is_some() { return Err(( StatusCode::CONFLICT, "multiple sources match host; query by url".to_string(), )); } source } (Some(_), Some(_)) => { return Err(( StatusCode::BAD_REQUEST, "query by either url or host, not both".to_string(), )); } (None, None) => { return Err(( StatusCode::BAD_REQUEST, "missing query param: url or host".to_string(), )); } }; Ok(Json(source))}
pub async fn list_sources( State(hydrant): State<Hydrant>, Query(query): Query<ListSourcesQuery>,) -> Json<Vec<FirehoseSourceInfo>> { let mut sources = hydrant.firehose.list_sources().await; sources.retain(|source| { query .host .as_deref() .is_none_or(|host| source.url.host_str() == Some(host)) && query .url .as_deref() .is_none_or(|url| source.url.as_str() == url) && query.is_pds.is_none_or(|is_pds| source.is_pds == is_pds) && query .running .is_none_or(|running| source.running == running) && query .failing .is_none_or(|failing| source.failing == failing) && query .throttled .is_none_or(|throttled| source.throttled == throttled) }); Json(sources)}
#[cfg(feature = "firehose-diagnostics")]pub async fn get_diagnostics( State(hydrant): State<Hydrant>,) -> Json<crate::control::FirehoseDiagnosticsInfo> { Json(hydrant.firehose.diagnostics())}
#[derive(Deserialize)]pub struct AddSourceRequest { pub url: Url, /// true to treat this as a direct PDS connection; enables host authority enforcement. /// defaults to false (aggregating relay). #[serde(default)] pub is_pds: bool,}
pub async fn add_source( State(hydrant): State<Hydrant>, Json(body): Json<AddSourceRequest>,) -> Result<StatusCode, (StatusCode, String)> { hydrant .firehose .add_source(body.url, body.is_pds) .await .map(|_| StatusCode::CREATED) .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))}
#[derive(Deserialize)]pub struct RemoveSourceRequest { pub url: Url,}
pub async fn remove_source( State(hydrant): State<Hydrant>, Json(body): Json<RemoveSourceRequest>,) -> Result<StatusCode, (StatusCode, String)> { hydrant .firehose .remove_source(&body.url) .await .map(|found| { found .then_some(StatusCode::OK) .unwrap_or(StatusCode::NOT_FOUND) }) .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))}
#[derive(Deserialize)]pub struct ResetCursorBody { pub key: String,}
pub async fn reset_cursor( State(hydrant): State<Hydrant>, Json(body): Json<ResetCursorBody>,) -> Result<StatusCode, (StatusCode, String)> { hydrant .firehose .reset_cursor(&body.key) .await .map(|_| StatusCode::OK) .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))}