use std::path::PathBuf;
use std::sync::Arc;
use axum::{
extract::{rejection::JsonRejection, State},
http::StatusCode,
response::IntoResponse,
routing::post,
Json, Router,
};
use lora_database::{LoraErrorCode, SnapshotAdmin, SnapshotMeta, WalAdmin};
use super::errors::{json_rejection_error, lora_error_response, ErrorResponse};
use super::types::{SnapshotRequest, SnapshotResponse, WalStatusResponse, WalTruncateRequest};
#[derive(Clone)]
pub struct SnapshotAdminConfig {
pub path: PathBuf,
pub admin: Arc<dyn SnapshotAdmin>,
}
#[derive(Clone, Default)]
pub struct AdminConfig {
pub snapshot: Option<SnapshotAdminConfig>,
pub wal: Option<Arc<dyn WalAdmin>>,
}
impl AdminConfig {
pub fn snapshot_only(snapshot_path: PathBuf, admin: Arc<dyn SnapshotAdmin>) -> Self {
Self {
snapshot: Some(SnapshotAdminConfig {
path: snapshot_path,
admin,
}),
wal: None,
}
}
pub fn wal_only(wal: Arc<dyn WalAdmin>) -> Self {
Self {
snapshot: None,
wal: Some(wal),
}
}
pub fn is_empty(&self) -> bool {
self.snapshot.is_none() && self.wal.is_none()
}
}
pub(crate) fn build_admin_router(cfg: AdminConfig) -> Router {
let mut router = Router::new();
if let Some(snap) = cfg.snapshot.clone() {
let snapshot_router: Router = Router::new()
.route("/admin/snapshot/save", post(admin_snapshot_save))
.route("/admin/snapshot/load", post(admin_snapshot_load))
.with_state(snap);
router = router.merge(snapshot_router);
}
if let Some(wal) = cfg.wal.clone() {
let wal_state = WalAdminState {
default_checkpoint_path: cfg.snapshot.as_ref().map(|s| s.path.clone()),
wal,
};
let wal_router: Router = Router::new()
.route("/admin/checkpoint", post(admin_checkpoint))
.route("/admin/wal/status", post(admin_wal_status))
.route("/admin/wal/truncate", post(admin_wal_truncate))
.with_state(wal_state);
router = router.merge(wal_router);
}
router
}
#[derive(Clone)]
struct WalAdminState {
default_checkpoint_path: Option<PathBuf>,
wal: Arc<dyn WalAdmin>,
}
fn resolve_snapshot_path(cfg: &SnapshotAdminConfig, req: Option<&SnapshotRequest>) -> PathBuf {
match req.and_then(|r| r.path.as_deref()) {
Some(p) if !p.trim().is_empty() => PathBuf::from(p),
_ => cfg.path.clone(),
}
}
async fn admin_snapshot_save(
State(cfg): State<SnapshotAdminConfig>,
body: Result<Json<SnapshotRequest>, JsonRejection>,
) -> impl IntoResponse {
let req = match parse_optional_json(body) {
Ok(req) => req,
Err(err) => return lora_error_response(json_rejection_error(err)),
};
let path = resolve_snapshot_path(&cfg, req.as_ref());
match cfg.admin.save_snapshot(&path) {
Ok(meta) => snapshot_response(meta, path),
Err(err) => lora_error_response(err),
}
}
async fn admin_snapshot_load(
State(cfg): State<SnapshotAdminConfig>,
body: Result<Json<SnapshotRequest>, JsonRejection>,
) -> impl IntoResponse {
let req = match parse_optional_json(body) {
Ok(req) => req,
Err(err) => return lora_error_response(json_rejection_error(err)),
};
let path = resolve_snapshot_path(&cfg, req.as_ref());
match cfg.admin.load_snapshot(&path) {
Ok(meta) => snapshot_response(meta, path),
Err(err) => lora_error_response(err),
}
}
fn resolve_checkpoint_path(
state: &WalAdminState,
req: Option<&SnapshotRequest>,
) -> Result<PathBuf, &'static str> {
match req.and_then(|r| r.path.as_deref()) {
Some(p) if !p.trim().is_empty() => Ok(PathBuf::from(p)),
_ => state
.default_checkpoint_path
.clone()
.ok_or("no checkpoint path: pass `path` in the request body or start the server with --snapshot-path"),
}
}
async fn admin_checkpoint(
State(state): State<WalAdminState>,
body: Result<Json<SnapshotRequest>, JsonRejection>,
) -> impl IntoResponse {
let req = match parse_optional_json(body) {
Ok(req) => req,
Err(err) => return lora_error_response(json_rejection_error(err)),
};
let path = match resolve_checkpoint_path(&state, req.as_ref()) {
Ok(p) => p,
Err(msg) => {
return (
StatusCode::BAD_REQUEST,
Json(ErrorResponse::from_parts(LoraErrorCode::Config, msg)),
)
.into_response()
}
};
match state.wal.checkpoint(&path) {
Ok(meta) => snapshot_response(meta, path),
Err(err) => lora_error_response(err),
}
}
fn snapshot_response(meta: SnapshotMeta, path: PathBuf) -> axum::response::Response {
(
StatusCode::OK,
Json(SnapshotResponse {
format_version: meta.format_version,
node_count: meta.node_count as u64,
relationship_count: meta.relationship_count as u64,
wal_lsn: meta.wal_lsn,
path: path.display().to_string(),
}),
)
.into_response()
}
async fn admin_wal_status(State(state): State<WalAdminState>) -> impl IntoResponse {
match state.wal.wal_status() {
Ok(s) => (
StatusCode::OK,
Json(WalStatusResponse {
durable_lsn: s.durable_lsn,
next_lsn: s.next_lsn,
active_segment_id: s.active_segment_id,
oldest_segment_id: s.oldest_segment_id,
bg_failure: s.bg_failure,
}),
)
.into_response(),
Err(err) => lora_error_response(err),
}
}
async fn admin_wal_truncate(
State(state): State<WalAdminState>,
body: Result<Json<WalTruncateRequest>, JsonRejection>,
) -> impl IntoResponse {
let req = match parse_optional_json(body) {
Ok(req) => req,
Err(err) => return lora_error_response(json_rejection_error(err)),
};
let fence = match req.and_then(|r| r.fence_lsn) {
Some(lsn) => lsn,
None => match state.wal.wal_status() {
Ok(s) => s.durable_lsn,
Err(err) => return lora_error_response(err),
},
};
match state.wal.wal_truncate(fence) {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(err) => lora_error_response(err),
}
}
fn parse_optional_json<T>(
body: Result<Json<T>, JsonRejection>,
) -> Result<Option<T>, JsonRejection> {
match body {
Ok(Json(value)) => Ok(Some(value)),
Err(JsonRejection::MissingJsonContentType(_)) => Ok(None),
Err(err) => Err(err),
}
}