use 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>,
/// only the repositories the indexer still has work queued for
#[serde(default)]
pub queued: bool,
}
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)?;
#[cfg(not(feature = "indexer"))]
if params.queued {
return Err((
StatusCode::BAD_REQUEST,
"queued needs indexer mode, there's no backfill queue here".to_string(),
));
}
#[cfg(feature = "indexer")]
let queued = params.queued;
let items = hydrant
.state
.db
.run(move |_db| {
#[cfg(feature = "indexer")]
if queued {
return hydrant
.repos
.iter_queued(cursor.as_ref())
.take(limit)
.collect::<miette::Result<Vec<_>>>();
}
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())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn queued_is_off_unless_asked_for() {
let parse = |uri: &str| {
Query::<GetReposParams>::try_from_uri(&uri.parse().unwrap())
.unwrap()
.0
};
assert!(!parse("http://hydrant/repos?limit=5").queued);
assert!(!parse("http://hydrant/repos?queued=false").queued);
assert!(parse("http://hydrant/repos?queued=true").queued);
}
}