1use super::history::{FinishedKind, JobHistory, TableMaintenance, maintenance_tables};
124use super::{ApiResponse, error_reply, json_reply, with_shared};
125use hammerwork::queue::DatabaseQueue;
126use serde::{Deserialize, Serialize};
127use std::sync::Arc;
128use tokio::sync::RwLock;
129use warp::http::StatusCode;
130use warp::{Filter, Reply};
131
132#[derive(Clone)]
134pub struct SystemState {
135 pub started_at: chrono::DateTime<chrono::Utc>,
137 pub config: crate::DashboardConfig,
139 pub database_type: String,
141 pub pool_size: u32,
143}
144
145impl SystemState {
146 pub fn new(config: crate::DashboardConfig, database_type: String, pool_size: u32) -> Self {
147 Self {
148 started_at: chrono::Utc::now(),
149 config,
150 database_type,
151 pool_size,
152 }
153 }
154
155 pub fn uptime_seconds(&self) -> i64 {
158 (chrono::Utc::now() - self.started_at).num_seconds().max(0)
159 }
160}
161
162#[derive(Debug, Serialize)]
164pub struct SystemInfo {
165 pub version: String,
166 pub build_info: BuildInfo,
167 pub runtime_info: RuntimeInfo,
168 pub database_info: DatabaseInfo,
169 pub features: Vec<String>,
170 pub uptime_seconds: u64,
171 pub started_at: chrono::DateTime<chrono::Utc>,
172}
173
174#[derive(Debug, Serialize)]
176pub struct BuildInfo {
177 pub version: String,
178 pub git_commit: Option<String>,
179 pub build_date: Option<String>,
180 pub rust_version: String,
181 pub target_triple: String,
182}
183
184#[derive(Debug, Serialize)]
186pub struct RuntimeInfo {
187 pub process_id: u32,
188 pub memory_usage_bytes: Option<u64>,
189 pub cpu_usage_percent: Option<f64>,
190 pub thread_count: Option<usize>,
191 pub gc_collections: Option<u64>,
192}
193
194#[derive(Debug, Serialize)]
196pub struct DatabaseInfo {
197 pub database_type: String,
198 pub connection_url: String, pub pool_size: u32,
200 pub active_connections: Option<u32>,
201 pub connection_health: bool,
202 pub last_migration: Option<String>,
203}
204
205#[derive(Debug, Serialize)]
207pub struct ServerConfig {
208 pub bind_address: String,
209 pub port: u16,
210 pub authentication_enabled: bool,
211 pub cors_enabled: bool,
212 pub websocket_max_connections: usize,
213 pub static_assets_path: String,
214}
215
216#[derive(Debug, Serialize)]
218pub struct MetricsInfo {
219 pub prometheus_enabled: bool,
220 pub metrics_endpoint: String,
224 pub custom_metrics_count: Option<u32>,
226 pub last_scrape: Option<chrono::DateTime<chrono::Utc>>,
227}
228
229#[derive(Debug, Deserialize)]
231pub struct ConfigUpdateRequest {
232 pub setting: String,
233 pub value: serde_json::Value,
234}
235
236#[derive(Debug, Deserialize)]
244pub struct MaintenanceRequest {
245 pub operation: String,
247 pub target: Option<String>,
249 pub dry_run: Option<bool>,
250}
251
252pub fn routes<T>(
254 queue: Arc<T>,
255 system_state: Arc<RwLock<SystemState>>,
256) -> impl Filter<Extract = impl Reply, Error = warp::Rejection> + Clone
257where
258 T: JobHistory + 'static,
259{
260 let queue_filter = with_shared(queue);
261 let state_filter = with_shared(system_state);
262
263 let info = warp::path("system")
264 .and(warp::path("info"))
265 .and(warp::path::end())
266 .and(warp::get())
267 .and(queue_filter.clone())
268 .and(state_filter.clone())
269 .and_then(system_info_handler);
270
271 let config = warp::path("system")
272 .and(warp::path("config"))
273 .and(warp::path::end())
274 .and(warp::get())
275 .and(state_filter.clone())
276 .and_then(system_config_handler);
277
278 let metrics_info = warp::path("system")
279 .and(warp::path("metrics"))
280 .and(warp::path::end())
281 .and(warp::get())
282 .and(state_filter.clone())
283 .and_then(metrics_info_handler);
284
285 let maintenance = warp::path("system")
286 .and(warp::path("maintenance"))
287 .and(warp::path::end())
288 .and(warp::post())
289 .and(queue_filter.clone())
290 .and(crate::security::json_body())
291 .and_then(maintenance_handler);
292
293 let version = warp::path("version")
294 .and(warp::path::end())
295 .and(warp::get())
296 .and_then(version_handler);
297
298 info.or(config).or(metrics_info).or(maintenance).or(version)
299}
300
301async fn system_info_handler<T>(
303 queue: Arc<T>,
304 system_state: Arc<RwLock<SystemState>>,
305) -> Result<impl Reply, warp::Rejection>
306where
307 T: DatabaseQueue + Send + Sync,
308{
309 let database_healthy = queue.get_all_queue_stats().await.is_ok();
311
312 let build_info = BuildInfo {
313 version: env!("CARGO_PKG_VERSION").to_string(),
314 git_commit: option_env!("GIT_COMMIT").map(|s| s.to_string()),
315 build_date: option_env!("BUILD_DATE").map(|s| s.to_string()),
316 rust_version: get_rust_version(),
317 target_triple: get_target_triple(),
318 };
319
320 let runtime_info = RuntimeInfo {
321 process_id: std::process::id(),
322 memory_usage_bytes: get_memory_usage(),
323 cpu_usage_percent: None, thread_count: None, gc_collections: None, };
327
328 let state = system_state.read().await;
329
330 let database_info = DatabaseInfo {
331 database_type: state.database_type.clone(),
332 connection_url: "***masked***".to_string(),
333 pool_size: state.pool_size,
334 active_connections: None, connection_health: database_healthy,
336 last_migration: None, };
338
339 let features = vec![
340 #[cfg(feature = "postgres")]
341 "postgres".to_string(),
342 #[cfg(feature = "mysql")]
343 "mysql".to_string(),
344 #[cfg(feature = "auth")]
345 "auth".to_string(),
346 ];
347
348 let system_info = SystemInfo {
349 version: env!("CARGO_PKG_VERSION").to_string(),
350 build_info,
351 runtime_info,
352 database_info,
353 features,
354 uptime_seconds: u64::try_from(state.uptime_seconds()).unwrap_or(0),
355 started_at: state.started_at,
356 };
357
358 Ok(json_reply(&ApiResponse::success(system_info)))
359}
360
361async fn system_config_handler(
363 system_state: Arc<RwLock<SystemState>>,
364) -> Result<impl Reply, warp::Rejection> {
365 let state = system_state.read().await;
366 let config = ServerConfig {
367 bind_address: state.config.bind_address.clone(),
368 port: state.config.port,
369 authentication_enabled: state.config.auth.enabled,
370 cors_enabled: state.config.enable_cors,
371 websocket_max_connections: state.config.websocket.max_connections,
372 static_assets_path: state.config.static_dir.to_string_lossy().to_string(),
373 };
374
375 Ok(json_reply(&ApiResponse::success(config)))
376}
377
378async fn metrics_info_handler(
380 _system_state: Arc<RwLock<SystemState>>,
381) -> Result<impl Reply, warp::Rejection> {
382 let metrics_info = MetricsInfo {
383 prometheus_enabled: false,
386 metrics_endpoint: "/metrics".to_string(),
387 custom_metrics_count: None,
388 last_scrape: None,
389 };
390
391 Ok(json_reply(&ApiResponse::success(metrics_info)))
392}
393
394async fn maintenance_handler<T>(
396 queue: Arc<T>,
397 request: MaintenanceRequest,
398) -> Result<impl Reply, warp::Rejection>
399where
400 T: JobHistory,
401{
402 let dry_run = request.dry_run.unwrap_or(false);
403
404 match request.operation.as_str() {
405 "cleanup" => {
406 let older_than = chrono::Utc::now() - chrono::Duration::days(7); if dry_run {
408 match queue
409 .count_jobs(None, FinishedKind::Dead, Some(older_than))
410 .await
411 {
412 Ok(count) => {
413 let response = ApiResponse::success(serde_json::json!({
414 "operation": "cleanup",
415 "dry_run": true,
416 "message": format!("Dry run: would clean up {} dead jobs older than 7 days", count),
417 "estimated_deletions": count
418 }));
419 Ok(json_reply(&response))
420 }
421 Err(e) => Ok(error_reply(
422 StatusCode::INTERNAL_SERVER_ERROR,
423 format!("Cleanup dry run failed: {}", e),
424 )),
425 }
426 } else {
427 match queue
430 .delete_jobs(None, FinishedKind::Dead, Some(older_than))
431 .await
432 {
433 Ok(count) => {
434 let response = ApiResponse::success(serde_json::json!({
435 "operation": "cleanup",
436 "dry_run": false,
437 "message": format!("Cleaned up {} dead jobs", count),
438 "deletions": count
439 }));
440 Ok(json_reply(&response))
441 }
442 Err(e) => Ok(error_reply(
443 StatusCode::INTERNAL_SERVER_ERROR,
444 format!("Cleanup failed: {}", e),
445 )),
446 }
447 }
448 }
449 name => {
450 let Some(operation) = TableMaintenance::parse(name) else {
451 return Ok(error_reply(
452 StatusCode::BAD_REQUEST,
453 format!("Unknown maintenance operation: {}", request.operation),
454 ));
455 };
456 let tables = match maintenance_tables(request.target.as_deref()) {
459 Ok(tables) => tables,
460 Err(message) => return Ok(error_reply(StatusCode::BAD_REQUEST, message)),
461 };
462 match queue.table_maintenance(operation, &tables, dry_run).await {
463 Ok(statements) => {
464 let message = if dry_run {
465 format!("Dry run: would run {} statements", statements.len())
466 } else {
467 format!("Ran {} on {} tables", name, statements.len())
468 };
469 Ok(json_reply(&ApiResponse::success(serde_json::json!({
470 "operation": operation,
471 "dry_run": dry_run,
472 "message": message,
473 "statements": statements,
474 }))))
475 }
476 Err(e) => Ok(error_reply(
477 StatusCode::INTERNAL_SERVER_ERROR,
478 format!("{} failed: {}", name, e),
479 )),
480 }
481 }
482 }
483}
484
485async fn version_handler() -> Result<impl Reply, warp::Rejection> {
487 let version_info = serde_json::json!({
488 "version": env!("CARGO_PKG_VERSION"),
489 "name": env!("CARGO_PKG_NAME"),
490 "description": env!("CARGO_PKG_DESCRIPTION"),
491 "authors": env!("CARGO_PKG_AUTHORS").split(':').collect::<Vec<_>>(),
492 "repository": env!("CARGO_PKG_REPOSITORY"),
493 "license": env!("CARGO_PKG_LICENSE"),
494 "rust_version": get_rust_version(),
495 "build_target": get_target_triple(),
496 });
497
498 Ok(json_reply(&ApiResponse::success(version_info)))
499}
500
501fn get_rust_version() -> String {
503 option_env!("RUSTC_VERSION")
504 .unwrap_or("unknown")
505 .to_string()
506}
507
508fn get_target_triple() -> String {
510 std::env::consts::ARCH.to_string() + "-" + std::env::consts::OS
511}
512
513fn get_memory_usage() -> Option<u64> {
515 #[cfg(target_os = "linux")]
516 {
517 use std::fs;
518 if let Ok(contents) = fs::read_to_string("/proc/self/status") {
519 for line in contents.lines() {
520 if let Some(rest) = line.strip_prefix("VmRSS:")
521 && let Some(kb) = rest
522 .split_whitespace()
523 .next()
524 .and_then(|s| s.parse::<u64>().ok())
525 {
526 return Some(kb.saturating_mul(1024)); }
528 }
529 }
530 }
531
532 None
536}
537
538#[cfg(test)]
539mod tests {
540 use super::*;
541
542 #[test]
544 fn uptime_is_never_negative() {
545 let mut state = SystemState::new(crate::DashboardConfig::default(), "pg".into(), 1);
546 state.started_at = chrono::Utc::now() + chrono::Duration::hours(1);
547 assert_eq!(state.uptime_seconds(), 0);
548 state.started_at = chrono::Utc::now() - chrono::Duration::seconds(90);
549 assert!(state.uptime_seconds() >= 90);
550 }
551
552 #[test]
553 fn test_maintenance_request_deserialization() {
554 let json = r#"{
555 "operation": "cleanup",
556 "target": "completed_jobs",
557 "dry_run": true
558 }"#;
559
560 let request: MaintenanceRequest = serde_json::from_str(json).unwrap();
561 assert_eq!(request.operation, "cleanup");
562 assert_eq!(request.target, Some("completed_jobs".to_string()));
563 assert_eq!(request.dry_run, Some(true));
564 }
565
566 #[test]
567 fn test_build_info_creation() {
568 let build_info = BuildInfo {
569 version: "1.0.0".to_string(),
570 git_commit: Some("abc123".to_string()),
571 build_date: Some("2024-01-01".to_string()),
572 rust_version: "1.70.0".to_string(),
573 target_triple: "x86_64-unknown-linux-gnu".to_string(),
574 };
575
576 let json = serde_json::to_string(&build_info).unwrap();
577 assert!(json.contains("1.0.0"));
578 assert!(json.contains("abc123"));
579 }
580
581 #[test]
582 fn test_unavailable_metrics_are_none_not_zero() {
583 let json = serde_json::to_value(MetricsInfo {
584 prometheus_enabled: false,
585 metrics_endpoint: "/metrics".to_string(),
586 custom_metrics_count: None,
587 last_scrape: None,
588 })
589 .unwrap();
590 assert!(json["custom_metrics_count"].is_null());
591 assert!(json["last_scrape"].is_null());
592 }
593
594 #[tokio::test]
595 async fn test_cleanup_dry_run_reports_database_failure() {
596 let response = maintenance_handler(
597 crate::api::test_support::unreachable_queue(),
598 MaintenanceRequest {
599 operation: "cleanup".to_string(),
600 target: None,
601 dry_run: Some(true),
602 },
603 )
604 .await
605 .unwrap()
606 .into_response();
607 let (status, _) = crate::api::test_support::body_json(response).await;
608 assert_eq!(status, 500);
609 }
610
611 async fn maintenance_status(operation: &str, target: Option<&str>) -> u16 {
612 let response = maintenance_handler(
613 crate::api::test_support::unreachable_queue(),
614 MaintenanceRequest {
615 operation: operation.to_string(),
616 target: target.map(str::to_string),
617 dry_run: Some(false),
618 },
619 )
620 .await
621 .unwrap()
622 .into_response();
623 crate::api::test_support::body_json(response).await.0
624 }
625
626 #[tokio::test]
627 async fn test_table_maintenance_rejects_foreign_tables_before_the_database() {
628 for operation in ["vacuum", "reindex", "optimize"] {
629 assert_eq!(maintenance_status(operation, Some("users")).await, 400);
630 assert_eq!(
632 maintenance_status(operation, Some("hammerwork_jobs")).await,
633 500
634 );
635 }
636 assert_eq!(maintenance_status("defrag", None).await, 400);
637 }
638
639 #[test]
640 fn test_get_rust_version() {
641 let version = get_rust_version();
642 assert!(!version.is_empty());
643 }
644
645 #[test]
646 fn test_get_target_triple() {
647 let target = get_target_triple();
648 assert!(!target.is_empty());
649 assert!(target.contains("-"));
650 }
651}