Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250use std::str::FromStr;
use crate::control::{Hydrant, RepoInfo};#[cfg(feature = "indexer")]use axum::routing::{delete, post, put};use axum::{ Json, Router, body::Body, extract::{Path, Query, State}, http::{HeaderMap, StatusCode, header}, response::{IntoResponse, Response}, routing::get,};use jacquard_common::types::did::Did;use serde::Deserialize;
pub fn router() -> Router<Hydrant> { #[allow(unused_mut)] let r = Router::new() .route("/repos", get(handle_get_repos)) .route("/repos/{did}", get(handle_get_repo));
#[cfg(feature = "indexer")] let r = r .route("/repos/resync", post(handle_post_resync)) .route("/repos", put(handle_put_repos)) .route("/repos", delete(handle_delete_repos)) .route("/repos/{did}", delete(handle_delete_single_repo));
r}
#[cfg(feature = "indexer")]#[derive(Deserialize, Debug)]pub struct RepoRequest { pub did: String,}
#[derive(Deserialize)]pub struct GetReposParams { pub limit: Option<usize>, pub cursor: Option<String>,}
pub async fn handle_get_repos( State(hydrant): State<Hydrant>, Query(params): Query<GetReposParams>, headers: HeaderMap,) -> Result<Response, (StatusCode, String)> { let limit = params.limit.unwrap_or(100).min(1000); let cursor = params .cursor .map(|c| Did::from_str(&c)) .transpose() .map_err(bad_request)?;
let items = hydrant .state .db .run(move |_db| { hydrant .repos .iter(cursor.as_ref()) .take(limit) .collect::<miette::Result<Vec<_>>>() }) .await .map_err(internal)?;
if prefers_json(&headers) { return Ok(Json(items).into_response()); }
use futures::StreamExt;
let stream = futures::stream::iter(items.into_iter().map(|item| { let json = serde_json::to_string(&item).ok()?; Some(Ok::<_, std::io::Error>(format!("{json}\n"))) })) .filter_map(futures::future::ready);
let body = Body::from_stream(stream);
Ok(([(header::CONTENT_TYPE, "application/x-ndjson")], body).into_response())}
pub async fn handle_get_repo( State(hydrant): State<Hydrant>, Path(did_str): Path<String>,) -> Result<Json<RepoInfo>, (StatusCode, String)> { let did = Did::new_owned(&did_str).map_err(bad_request)?;
let item = hydrant .repos .info(&did) .await .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
item.map(Json) .ok_or_else(|| (StatusCode::NOT_FOUND, "repository not found".to_string()))}
#[cfg(feature = "indexer")]pub async fn handle_put_repos( State(hydrant): State<Hydrant>, headers: HeaderMap, body: Body,) -> Result<Response, (StatusCode, String)> { let items = parse_body(body, &headers).await?;
let dids: Vec<Did> = items .into_iter() .map(|item| Did::new_owned(&item.did).map_err(bad_request)) .collect::<Result<_, _>>()?;
let queued = hydrant .repos .track(dids) .await .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
Ok(did_list_response(queued, &headers))}
#[cfg(feature = "indexer")]pub async fn handle_delete_repos( State(hydrant): State<Hydrant>, headers: HeaderMap, body: Body,) -> Result<Response, (StatusCode, String)> { let items = parse_body(body, &headers).await?;
let dids: Vec<Did> = items .into_iter() .map(|item| Did::new_owned(&item.did).map_err(bad_request)) .collect::<Result<_, _>>()?;
let untracked = hydrant .repos .untrack(dids) .await .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
Ok(did_list_response(untracked, &headers))}
#[cfg(feature = "indexer")]pub async fn handle_delete_single_repo( State(hydrant): State<Hydrant>, Path(did_str): Path<String>, headers: HeaderMap,) -> Result<Response, (StatusCode, String)> { let did = Did::new_owned(&did_str).map_err(bad_request)?;
let untracked = hydrant .repos .untrack([did]) .await .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
Ok(did_list_response(untracked, &headers))}
#[cfg(feature = "indexer")]pub async fn handle_post_resync( State(hydrant): State<Hydrant>, headers: HeaderMap, body: Body,) -> Result<Response, (StatusCode, String)> { let items = parse_body(body, &headers).await?;
let dids: Vec<Did> = items .into_iter() .map(|item| Did::new_owned(&item.did).map_err(bad_request)) .collect::<Result<_, _>>()?;
let queued = hydrant.repos.resync(dids).await.map_err(internal)?;
Ok(did_list_response(queued, &headers))}
fn prefers_json(headers: &HeaderMap) -> bool { let contains_json = |h: axum::http::HeaderName| { headers .get(h) .and_then(|v| v.to_str().ok()) .is_some_and(|v| v.contains("application/json")) }; contains_json(header::ACCEPT) || contains_json(header::CONTENT_TYPE)}
#[cfg(feature = "indexer")]fn did_list_response(dids: Vec<Did>, headers: &HeaderMap) -> Response { if prefers_json(headers) { let body: Vec<String> = dids.into_iter().map(|d| d.to_string()).collect(); Json(body).into_response() } else { let body = dids .iter() .filter_map(|d| serde_json::to_string(&d.as_str()).ok()) .map(|s| format!("{s}\n")) .collect::<String>(); ([(header::CONTENT_TYPE, "application/x-ndjson")], body).into_response() }}
#[cfg(feature = "indexer")]const MAX_BODY_SIZE: usize = 8 * 1024 * 1024; // 8 MiB
#[cfg(feature = "indexer")]async fn parse_body( body: Body, headers: &HeaderMap,) -> Result<Vec<RepoRequest>, (StatusCode, String)> { let content_type = headers .get(header::CONTENT_TYPE) .and_then(|h| h.to_str().ok()) .unwrap_or("") .to_string();
let body_bytes = axum::body::to_bytes(body, MAX_BODY_SIZE) .await .map_err(bad_request)?;
let text = std::str::from_utf8(&body_bytes).map_err(bad_request)?; let trimmed = text.trim();
if content_type.contains("application/json") { serde_json::from_str::<Vec<RepoRequest>>(trimmed) .map_err(|e| bad_request(format!("invalid JSON array: {e}"))) } else { trimmed .lines() .filter(|l| !l.trim().is_empty()) .map(|line| { serde_json::from_str::<RepoRequest>(line) .map_err(|e| bad_request(format!("invalid NDJSON line: {e}"))) }) .collect() }}
fn bad_request<E: std::fmt::Display>(err: E) -> (StatusCode, String) { (StatusCode::BAD_REQUEST, err.to_string())}
fn internal<E: std::fmt::Display>(err: E) -> (StatusCode, String) { (StatusCode::INTERNAL_SERVER_ERROR, err.to_string())}