alopex_server/http/
admin_api.rs

1use std::path::Path;
2use std::sync::Arc;
3
4use alopex_cluster::{
5    bootstrap_cluster_control, ClusterBootstrapConfig, ClusterBootstrapMode,
6    ClusterBootstrapOutcome, ClusterMode, ClusterStatusSnapshot, UpgradeOperation,
7};
8use alopex_core::kv::any::AnyKV;
9use axum::extract::{Extension, Path as AxumPath};
10use axum::http::StatusCode;
11use axum::response::{IntoResponse, Response};
12use axum::Json;
13use serde::{Deserialize, Serialize};
14use uuid::Uuid;
15
16use crate::auth::AuthMode;
17use crate::http::{error_response, RequestContext};
18use crate::metrics::ClusterMetricsSurface;
19use crate::ops::backup::{copy_dir_filtered, export_snapshot, BackupHandle};
20use crate::ops::restore::{RestoreHandle, RestoreSource};
21use crate::ops::state::{OperationState, RestoreMetadata};
22use crate::ops::status::StatusReporter;
23use crate::ops::status::StatusView;
24use crate::server::ServerState;
25
26#[derive(Serialize)]
27struct AdminCapabilitiesResponse {
28    scope: &'static str,
29    allowed_actions: Vec<&'static str>,
30    unsupported_actions: Vec<&'static str>,
31}
32
33#[derive(Serialize)]
34struct AdminStatusResponse {
35    version: Option<String>,
36    uptime_secs: Option<u64>,
37    connections: Option<u64>,
38    queries_per_second: Option<f64>,
39    cluster: ClusterStatusSnapshot,
40    cluster_control: ClusterControlAvailability,
41    #[serde(flatten)]
42    status: StatusView,
43}
44
45#[derive(Serialize)]
46struct AdminMetricsResponse {
47    qps: Option<f64>,
48    avg_latency_ms: Option<f64>,
49    p99_latency_ms: Option<f64>,
50    memory_usage_mb: Option<u64>,
51    active_connections: Option<u64>,
52    cluster: ClusterStatusSnapshot,
53    cluster_metrics: ClusterMetricsSurface,
54}
55
56#[derive(Serialize)]
57struct AdminHealthResponse {
58    status: &'static str,
59    message: &'static str,
60    degraded: bool,
61    cluster: ClusterStatusSnapshot,
62}
63
64#[derive(Serialize)]
65struct AdminClusterOperationResponse {
66    action: &'static str,
67    cluster: ClusterStatusSnapshot,
68}
69
70/// Whether the running process can safely accept multi-node metadata control.
71/// A missing prerequisite is a normal, machine-readable state rather than an
72/// implicit in-memory fallback.
73#[derive(Debug, Clone, Serialize)]
74pub struct ClusterControlAvailability {
75    pub available: bool,
76    pub mode: ClusterMode,
77    pub reason: &'static str,
78    pub missing_prerequisites: Vec<alopex_cluster::ClusterCapabilityPrerequisite>,
79}
80
81#[derive(Serialize)]
82struct AdminClusterMetadataResponse {
83    cluster: ClusterStatusSnapshot,
84    control: ClusterControlAvailability,
85    metadata_state_version: Option<u64>,
86    schema_rollout: Option<serde_json::Value>,
87    upgrade: Option<UpgradeOperation>,
88}
89
90/// Explicitly typed cluster management operations.  These are management API
91/// verbs, not SQL DDL statements and never carry a user SQL string.
92#[derive(Debug, Clone, Copy, Deserialize, Serialize)]
93#[serde(rename_all = "snake_case")]
94pub enum AdminClusterManagementOperation {
95    MetadataShow,
96    MembersList,
97    MembersReplace,
98    RangesList,
99    RangesRegister,
100    RangesUpdate,
101    RangesRetire,
102    PlacementGet,
103    PlacementSet,
104    PlacementReplace,
105    ReadPolicyGet,
106    ReadPolicySet,
107    SchemaOwnerGet,
108    SchemaOwnerSet,
109    SchemaRolloutStart,
110    SchemaRolloutStatus,
111    RecoveryStatus,
112    RecoveryRestore,
113    UpgradeStatus,
114    UpgradeStart,
115}
116
117impl AdminClusterManagementOperation {
118    fn is_mutation(self) -> bool {
119        !matches!(
120            self,
121            Self::MetadataShow
122                | Self::MembersList
123                | Self::RangesList
124                | Self::PlacementGet
125                | Self::ReadPolicyGet
126                | Self::SchemaOwnerGet
127                | Self::SchemaRolloutStatus
128                | Self::RecoveryStatus
129                | Self::UpgradeStatus
130        )
131    }
132}
133
134#[derive(Debug, Deserialize)]
135pub struct AdminClusterManagementRequest {
136    pub request_id: String,
137    pub operation: AdminClusterManagementOperation,
138    #[serde(default)]
139    pub expected_version: Option<u64>,
140    /// The public target is opaque at this adapter boundary; the later typed
141    /// consensus adapter validates it against immutable metadata.
142    #[serde(default)]
143    pub target: Option<serde_json::Value>,
144    #[serde(default)]
145    pub confirmed: bool,
146}
147
148#[derive(Serialize)]
149struct AdminClusterManagementResponse {
150    operation_id: String,
151    operation: AdminClusterManagementOperation,
152    outcome_class: &'static str,
153    reason: &'static str,
154    state_version: Option<u64>,
155    control: ClusterControlAvailability,
156    actor: Option<String>,
157}
158
159#[derive(Deserialize)]
160pub struct AdminLifecycleRequest {
161    action: String,
162}
163
164#[derive(Deserialize)]
165pub struct AdminRestoreRequest {
166    #[serde(default)]
167    source: Option<String>,
168}
169
170#[derive(Serialize)]
171struct AdminLifecycleResponse {
172    status: &'static str,
173    message: String,
174}
175
176#[derive(Serialize)]
177struct AdminExportResponse {
178    status: &'static str,
179    location: String,
180}
181
182#[derive(Serialize)]
183struct AdminBackupResponse {
184    handle: String,
185    location: String,
186    state: OperationState,
187}
188
189#[derive(Serialize)]
190struct AdminRestoreResponse {
191    handle: String,
192    state: OperationState,
193    metadata: Option<RestoreMetadata>,
194}
195
196pub async fn capabilities(Extension(state): Extension<Arc<ServerState>>) -> impl IntoResponse {
197    let (scope, allowed_actions, unsupported_actions) = capabilities_for_auth(&state.auth);
198    Json(AdminCapabilitiesResponse {
199        scope,
200        allowed_actions,
201        unsupported_actions,
202    })
203}
204
205pub async fn status(
206    Extension(state): Extension<Arc<ServerState>>,
207    Extension(ctx): Extension<RequestContext>,
208) -> Response {
209    let uptime = state.start_time.elapsed().as_secs();
210    let reporter = StatusReporter::new(state.lifecycle_state.clone(), state.recovery_info.clone());
211    let status = reporter.status_view();
212    let cluster = match state.cluster_status_snapshot() {
213        Ok(snapshot) => snapshot,
214        Err(err) => return error_response(err, &ctx),
215    };
216    let cluster_control = match cluster_control_availability(&cluster) {
217        Ok(control) => control,
218        Err(err) => return error_response(err, &ctx),
219    };
220    state.metrics.record_cluster_status(&cluster);
221    Json(AdminStatusResponse {
222        version: Some(env!("CARGO_PKG_VERSION").to_string()),
223        uptime_secs: Some(uptime),
224        connections: None,
225        queries_per_second: None,
226        cluster,
227        cluster_control,
228        status,
229    })
230    .into_response()
231}
232
233pub async fn cluster_metadata(
234    Extension(state): Extension<Arc<ServerState>>,
235    Extension(ctx): Extension<RequestContext>,
236) -> Response {
237    let cluster = match state.cluster_status_snapshot() {
238        Ok(snapshot) => snapshot,
239        Err(err) => return error_response(err, &ctx),
240    };
241    let control = match cluster_control_availability(&cluster) {
242        Ok(control) => control,
243        Err(err) => return error_response(err, &ctx),
244    };
245    let upgrade = state.upgrade_coordinator.status().ok();
246    Json(AdminClusterMetadataResponse {
247        cluster,
248        control,
249        metadata_state_version: None,
250        schema_rollout: None,
251        upgrade,
252    })
253    .into_response()
254}
255
256pub async fn cluster_management(
257    Extension(state): Extension<Arc<ServerState>>,
258    Extension(ctx): Extension<RequestContext>,
259    Json(request): Json<AdminClusterManagementRequest>,
260) -> Response {
261    let cluster = match state.cluster_status_snapshot() {
262        Ok(snapshot) => snapshot,
263        Err(err) => return error_response(err, &ctx),
264    };
265    let control = match cluster_control_availability(&cluster) {
266        Ok(control) => control,
267        Err(err) => return error_response(err, &ctx),
268    };
269    let (outcome_class, reason) = if request.operation.is_mutation() && !request.confirmed {
270        ("terminal_failure", "confirmation_required")
271    } else if !control.available {
272        ("terminal_failure", "cluster_capability_unavailable")
273    } else {
274        // This branch becomes reachable only after a compatible Chirps
275        // consensus adapter is installed. Keeping it classified as pending
276        // prevents a route registration from falsely claiming a commit.
277        ("pending", "metadata_consensus_adapter_not_attached")
278    };
279    Json(AdminClusterManagementResponse {
280        operation_id: request.request_id,
281        operation: request.operation,
282        outcome_class,
283        reason,
284        state_version: None,
285        control,
286        actor: ctx.actor,
287    })
288    .into_response()
289}
290
291pub async fn metrics(
292    Extension(state): Extension<Arc<ServerState>>,
293    Extension(ctx): Extension<RequestContext>,
294) -> Response {
295    let cluster = match state.cluster_status_snapshot() {
296        Ok(snapshot) => snapshot,
297        Err(err) => return error_response(err, &ctx),
298    };
299    state.metrics.record_cluster_status(&cluster);
300    Json(AdminMetricsResponse {
301        qps: None,
302        avg_latency_ms: None,
303        p99_latency_ms: None,
304        memory_usage_mb: None,
305        active_connections: None,
306        cluster_metrics: ClusterMetricsSurface::from(&cluster),
307        cluster,
308    })
309    .into_response()
310}
311
312pub async fn health(
313    Extension(state): Extension<Arc<ServerState>>,
314    Extension(ctx): Extension<RequestContext>,
315) -> Response {
316    let cluster = match state.cluster_status_snapshot() {
317        Ok(snapshot) => snapshot,
318        Err(err) => return error_response(err, &ctx),
319    };
320    state.metrics.record_cluster_status(&cluster);
321    let (status, message) = if cluster.degraded {
322        ("degraded", "cluster status degraded")
323    } else {
324        ("ok", "ready")
325    };
326    Json(AdminHealthResponse {
327        status,
328        message,
329        degraded: cluster.degraded,
330        cluster,
331    })
332    .into_response()
333}
334
335pub async fn cluster_join(
336    Extension(state): Extension<Arc<ServerState>>,
337    Extension(ctx): Extension<RequestContext>,
338) -> Response {
339    cluster_operation_response(&state, &ctx, "join")
340}
341
342pub async fn cluster_leave(
343    Extension(state): Extension<Arc<ServerState>>,
344    Extension(ctx): Extension<RequestContext>,
345) -> Response {
346    cluster_operation_response(&state, &ctx, "leave")
347}
348
349/// Applies the same bootstrap gate used by the Rust control plane before an
350/// HTTP route advertises metadata mutation support.
351pub fn cluster_control_availability(
352    cluster: &ClusterStatusSnapshot,
353) -> crate::error::Result<ClusterControlAvailability> {
354    if cluster.mode == ClusterMode::SingleNode {
355        return Ok(ClusterControlAvailability {
356            available: false,
357            mode: cluster.mode,
358            reason: "single_node_mode",
359            missing_prerequisites: Vec::new(),
360        });
361    }
362    let outcome = bootstrap_cluster_control(&ClusterBootstrapConfig::compiled_chirps(
363        ClusterBootstrapMode::ClusterAware,
364    ));
365    match outcome {
366        ClusterBootstrapOutcome::ReadyForClusterControl => Ok(ClusterControlAvailability {
367            available: true,
368            mode: cluster.mode,
369            reason: "ready",
370            missing_prerequisites: Vec::new(),
371        }),
372        ClusterBootstrapOutcome::CapabilityUnavailable {
373            missing_prerequisites,
374        } => Ok(ClusterControlAvailability {
375            available: false,
376            mode: cluster.mode,
377            reason: "cluster_capability_unavailable",
378            missing_prerequisites,
379        }),
380        ClusterBootstrapOutcome::SingleNode => unreachable!("cluster-aware input was supplied"),
381    }
382}
383
384pub async fn compaction(
385    Extension(_state): Extension<Arc<ServerState>>,
386    Extension(ctx): Extension<RequestContext>,
387) -> Response {
388    error_response(
389        crate::error::ServerError::NotImplemented(
390            "manual compaction is not available for the server's LSM storage engine".into(),
391        ),
392        &ctx,
393    )
394}
395
396pub async fn start_backup(
397    Extension(state): Extension<Arc<ServerState>>,
398    Extension(ctx): Extension<RequestContext>,
399) -> Response {
400    match state.backup_coordinator.start_backup().await {
401        Ok(handle) => match backup_response(&state, &handle) {
402            Ok(response) => Json(response).into_response(),
403            Err(err) => error_response(err, &ctx),
404        },
405        Err(err) => error_response(err, &ctx),
406    }
407}
408
409pub async fn export(
410    Extension(state): Extension<Arc<ServerState>>,
411    Extension(ctx): Extension<RequestContext>,
412) -> Response {
413    let export_state = state.clone();
414    let result = tokio::task::spawn_blocking(move || perform_export(export_state.as_ref()))
415        .await
416        .map_err(|err| crate::error::ServerError::Internal(err.to_string()))
417        .and_then(|res| res);
418
419    match result {
420        Ok(location) => Json(AdminExportResponse {
421            status: "OK",
422            location,
423        })
424        .into_response(),
425        Err(err) => error_response(err, &ctx),
426    }
427}
428
429pub async fn backup_status(
430    AxumPath(id): AxumPath<String>,
431    Extension(state): Extension<Arc<ServerState>>,
432    Extension(ctx): Extension<RequestContext>,
433) -> Response {
434    let handle = match parse_backup_handle(&id) {
435        Ok(handle) => handle,
436        Err(err) => return error_response(err, &ctx),
437    };
438    match backup_response(&state, &handle) {
439        Ok(response) => Json(response).into_response(),
440        Err(err) => error_response(err, &ctx),
441    }
442}
443
444pub async fn start_restore(
445    Extension(state): Extension<Arc<ServerState>>,
446    Extension(ctx): Extension<RequestContext>,
447    Json(request): Json<AdminRestoreRequest>,
448) -> Response {
449    let source_path = match request.source {
450        Some(source) => source.into(),
451        None => match crate::ops::restore::resolve_default_source(&state.config.data_dir) {
452            Ok(path) => path,
453            Err(crate::error::ServerError::NotFound(_)) => {
454                match state.backup_coordinator.latest_location() {
455                    Some(path) => path,
456                    None => {
457                        let data_dir = state.config.data_dir.clone();
458                        let archive_result = tokio::task::spawn_blocking(move || {
459                            perform_lifecycle_action("archive", Path::new(&data_dir))
460                        })
461                        .await
462                        .map_err(|err| crate::error::ServerError::Internal(err.to_string()))
463                        .and_then(|res| res.map_err(crate::error::ServerError::BadRequest));
464                        if let Err(err) = archive_result {
465                            return error_response(err, &ctx);
466                        }
467                        match crate::ops::restore::resolve_default_source(&state.config.data_dir) {
468                            Ok(path) => path,
469                            Err(err) => return error_response(err, &ctx),
470                        }
471                    }
472                }
473            }
474            Err(err) => return error_response(err, &ctx),
475        },
476    };
477    let source = RestoreSource { path: source_path };
478    match state.restore_coordinator.start_restore(source).await {
479        Ok(handle) => match restore_response(&state, &handle) {
480            Ok(response) => Json(response).into_response(),
481            Err(err) => error_response(err, &ctx),
482        },
483        Err(err) => error_response(err, &ctx),
484    }
485}
486
487pub async fn restore_status(
488    AxumPath(id): AxumPath<String>,
489    Extension(state): Extension<Arc<ServerState>>,
490    Extension(ctx): Extension<RequestContext>,
491) -> Response {
492    let handle = match parse_restore_handle(&id) {
493        Ok(handle) => handle,
494        Err(err) => return error_response(err, &ctx),
495    };
496    match restore_response(&state, &handle) {
497        Ok(response) => Json(response).into_response(),
498        Err(err) => error_response(err, &ctx),
499    }
500}
501
502pub async fn lifecycle(
503    Extension(state): Extension<Arc<ServerState>>,
504    Json(request): Json<AdminLifecycleRequest>,
505) -> impl IntoResponse {
506    let data_dir = state.config.data_dir.clone();
507    let action = request.action;
508    let result = tokio::task::spawn_blocking(move || {
509        perform_lifecycle_action(action.as_str(), Path::new(&data_dir))
510    })
511    .await
512    .map_err(|err| err.to_string())
513    .and_then(|res| res.map_err(|err| err.to_string()));
514
515    match result {
516        Ok(message) => (
517            StatusCode::OK,
518            Json(AdminLifecycleResponse {
519                status: "OK",
520                message,
521            }),
522        )
523            .into_response(),
524        Err(err) => (
525            StatusCode::BAD_REQUEST,
526            Json(AdminLifecycleResponse {
527                status: "Error",
528                message: err,
529            }),
530        )
531            .into_response(),
532    }
533}
534
535fn parse_backup_handle(id: &str) -> crate::error::Result<BackupHandle> {
536    let id = Uuid::parse_str(id)
537        .map_err(|_| crate::error::ServerError::BadRequest("invalid backup handle".into()))?;
538    Ok(BackupHandle { id })
539}
540
541fn parse_restore_handle(id: &str) -> crate::error::Result<RestoreHandle> {
542    let id = Uuid::parse_str(id)
543        .map_err(|_| crate::error::ServerError::BadRequest("invalid restore handle".into()))?;
544    Ok(RestoreHandle { id })
545}
546
547fn backup_response(
548    state: &ServerState,
549    handle: &BackupHandle,
550) -> crate::error::Result<AdminBackupResponse> {
551    let location = state.backup_coordinator.location(handle)?;
552    let status = state.backup_coordinator.status(handle)?;
553    Ok(AdminBackupResponse {
554        handle: handle.id.to_string(),
555        location: location.display().to_string(),
556        state: status,
557    })
558}
559
560fn restore_response(
561    state: &ServerState,
562    handle: &RestoreHandle,
563) -> crate::error::Result<AdminRestoreResponse> {
564    let status = state.restore_coordinator.status(handle)?;
565    let metadata = state.restore_coordinator.metadata(handle)?;
566    Ok(AdminRestoreResponse {
567        handle: handle.id.to_string(),
568        state: status,
569        metadata,
570    })
571}
572
573fn capabilities_for_auth(
574    auth: &crate::auth::AuthMiddleware,
575) -> (&'static str, Vec<&'static str>, Vec<&'static str>) {
576    match auth.mode() {
577        AuthMode::None => ("full", Vec::new(), unsupported_actions()),
578        AuthMode::Dev { .. } => ("restricted", all_actions(), unsupported_actions()),
579    }
580}
581
582fn unsupported_actions() -> Vec<&'static str> {
583    vec!["compaction"]
584}
585
586fn all_actions() -> Vec<&'static str> {
587    vec![
588        "read", "create", "update", "delete", "archive", "restore", "backup", "export", "join",
589        "leave",
590    ]
591}
592
593fn cluster_operation_response(
594    state: &Arc<ServerState>,
595    ctx: &RequestContext,
596    action: &'static str,
597) -> Response {
598    let cluster = match action {
599        "join" => state.cluster_join(),
600        "leave" => state.cluster_leave(),
601        _ => unreachable!("cluster membership action is fixed by route"),
602    };
603    let cluster = match cluster {
604        Ok(snapshot) => snapshot,
605        Err(err) => return error_response(err, ctx),
606    };
607    state.metrics.record_cluster_status(&cluster);
608    Json(AdminClusterOperationResponse { action, cluster }).into_response()
609}
610
611fn perform_lifecycle_action(action: &str, data_dir: &Path) -> Result<String, String> {
612    if !data_dir.exists() {
613        return Err(format!(
614            "Data directory does not exist: {}",
615            data_dir.display()
616        ));
617    }
618    if !data_dir.is_dir() {
619        return Err(format!(
620            "Data directory is not a directory: {}",
621            data_dir.display()
622        ));
623    }
624
625    let lifecycle_root = data_dir.join(".lifecycle");
626    std::fs::create_dir_all(&lifecycle_root).map_err(|err| err.to_string())?;
627
628    match action {
629        "archive" => {
630            let dest = lifecycle_root.join("archive").join(timestamp_dir());
631            copy_data_dir(data_dir, &dest)?;
632            write_latest_marker(&lifecycle_root.join("archive"), &dest)?;
633            Ok(format!("Archived data to {}", dest.display()))
634        }
635        "export" => {
636            let dest = lifecycle_root.join("export").join(timestamp_dir());
637            copy_data_dir(data_dir, &dest)?;
638            write_latest_marker(&lifecycle_root.join("export"), &dest)?;
639            Ok(format!("Exported data to {}", dest.display()))
640        }
641        _ => Err("Unknown lifecycle action.".to_string()),
642    }
643}
644
645fn perform_export(state: &ServerState) -> crate::error::Result<String> {
646    match state.store.as_ref() {
647        AnyKV::Lsm(kv) => {
648            let _ = kv.checkpoint()?;
649        }
650        _ => {
651            return Err(crate::error::ServerError::BadRequest(
652                "checkpoint unsupported for current storage engine".to_string(),
653            ));
654        }
655    }
656    let data_dir = state.config.data_dir.as_path();
657    let lifecycle_root = data_dir.join(".lifecycle");
658    std::fs::create_dir_all(&lifecycle_root)?;
659    let dest = lifecycle_root.join("export").join(timestamp_dir());
660    std::fs::create_dir_all(&dest)?;
661    export_snapshot(data_dir, &dest)?;
662    write_latest_marker(&lifecycle_root.join("export"), &dest)
663        .map_err(crate::error::ServerError::Internal)?;
664    Ok(dest.display().to_string())
665}
666
667fn timestamp_dir() -> String {
668    let seconds = std::time::SystemTime::now()
669        .duration_since(std::time::UNIX_EPOCH)
670        .unwrap_or_default()
671        .as_secs();
672    format!("ts-{seconds}")
673}
674
675fn copy_data_dir(src: &Path, dest: &Path) -> Result<(), String> {
676    std::fs::create_dir_all(dest).map_err(|err| err.to_string())?;
677    copy_dir_filtered(src, dest).map_err(|err| err.to_string())
678}
679
680fn write_latest_marker(root: &Path, dest: &Path) -> Result<(), String> {
681    let marker = root.join("latest");
682    std::fs::create_dir_all(root).map_err(|err| err.to_string())?;
683    std::fs::write(&marker, dest.to_string_lossy().as_bytes()).map_err(|err| err.to_string())?;
684    Ok(())
685}
686
687#[cfg(test)]
688mod tests {
689    use super::*;
690    use alopex_cluster::{ClusterManager, ClusterManagerConfig};
691
692    #[test]
693    fn single_node_metadata_route_never_advertises_multi_node_control() {
694        let manager = ClusterManager::new(ClusterManagerConfig::single_node()).unwrap();
695        let availability = cluster_control_availability(&manager.status_snapshot()).unwrap();
696
697        assert!(!availability.available);
698        assert_eq!(availability.reason, "single_node_mode");
699        assert!(availability.missing_prerequisites.is_empty());
700    }
701
702    #[test]
703    fn only_read_operations_skip_explicit_mutation_confirmation() {
704        assert!(!AdminClusterManagementOperation::MetadataShow.is_mutation());
705        assert!(!AdminClusterManagementOperation::UpgradeStatus.is_mutation());
706        assert!(AdminClusterManagementOperation::RangesRegister.is_mutation());
707        assert!(AdminClusterManagementOperation::SchemaRolloutStart.is_mutation());
708    }
709}
alopex_server/http/admin_api.rs

alopex_server/http/
admin_api.rs
