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,
221 pub custom_metrics_count: Option<u32>,
223 pub last_scrape: Option<chrono::DateTime<chrono::Utc>>,
224}
225
226#[derive(Debug, Deserialize)]
228pub struct ConfigUpdateRequest {
229 pub setting: String,
230 pub value: serde_json::Value,
231}
232
233#[derive(Debug, Deserialize)]
241pub struct MaintenanceRequest {
242 pub operation: String,
244 pub target: Option<String>,
246 pub dry_run: Option<bool>,
247}
248
249pub fn routes<T>(
251 queue: Arc<T>,
252 system_state: Arc<RwLock<SystemState>>,
253) -> impl Filter<Extract = impl Reply, Error = warp::Rejection> + Clone
254where
255 T: JobHistory + 'static,
256{
257 let queue_filter = with_shared(queue);
258 let state_filter = with_shared(system_state);
259
260 let info = warp::path("system")
261 .and(warp::path("info"))
262 .and(warp::path::end())
263 .and(warp::get())
264 .and(queue_filter.clone())
265 .and(state_filter.clone())
266 .and_then(system_info_handler);
267
268 let config = warp::path("system")
269 .and(warp::path("config"))
270 .and(warp::path::end())
271 .and(warp::get())
272 .and(state_filter.clone())
273 .and_then(system_config_handler);
274
275 let metrics_info = warp::path("system")
276 .and(warp::path("metrics"))
277 .and(warp::path::end())
278 .and(warp::get())
279 .and(state_filter.clone())
280 .and_then(metrics_info_handler);
281
282 let maintenance = warp::path("system")
283 .and(warp::path("maintenance"))
284 .and(warp::path::end())
285 .and(warp::post())
286 .and(queue_filter.clone())
287 .and(crate::security::json_body())
288 .and_then(maintenance_handler);
289
290 let version = warp::path("version")
291 .and(warp::path::end())
292 .and(warp::get())
293 .and_then(version_handler);
294
295 info.or(config).or(metrics_info).or(maintenance).or(version)
296}
297
298async fn system_info_handler<T>(
300 queue: Arc<T>,
301 system_state: Arc<RwLock<SystemState>>,
302) -> Result<impl Reply, warp::Rejection>
303where
304 T: DatabaseQueue + Send + Sync,
305{
306 let database_healthy = queue.get_all_queue_stats().await.is_ok();
308
309 let build_info = BuildInfo {
310 version: env!("CARGO_PKG_VERSION").to_string(),
311 git_commit: option_env!("GIT_COMMIT").map(|s| s.to_string()),
312 build_date: option_env!("BUILD_DATE").map(|s| s.to_string()),
313 rust_version: get_rust_version(),
314 target_triple: get_target_triple(),
315 };
316
317 let runtime_info = RuntimeInfo {
318 process_id: std::process::id(),
319 memory_usage_bytes: get_memory_usage(),
320 cpu_usage_percent: None, thread_count: None, gc_collections: None, };
324
325 let state = system_state.read().await;
326
327 let database_info = DatabaseInfo {
328 database_type: state.database_type.clone(),
329 connection_url: "***masked***".to_string(),
330 pool_size: state.pool_size,
331 active_connections: None, connection_health: database_healthy,
333 last_migration: None, };
335
336 let features = vec![
337 #[cfg(feature = "postgres")]
338 "postgres".to_string(),
339 #[cfg(feature = "mysql")]
340 "mysql".to_string(),
341 #[cfg(feature = "auth")]
342 "auth".to_string(),
343 ];
344
345 let system_info = SystemInfo {
346 version: env!("CARGO_PKG_VERSION").to_string(),
347 build_info,
348 runtime_info,
349 database_info,
350 features,
351 uptime_seconds: u64::try_from(state.uptime_seconds()).unwrap_or(0),
352 started_at: state.started_at,
353 };
354
355 Ok(json_reply(&ApiResponse::success(system_info)))
356}
357
358async fn system_config_handler(
360 system_state: Arc<RwLock<SystemState>>,
361) -> Result<impl Reply, warp::Rejection> {
362 let state = system_state.read().await;
363 let config = ServerConfig {
364 bind_address: state.config.bind_address.clone(),
365 port: state.config.port,
366 authentication_enabled: state.config.auth.enabled,
367 cors_enabled: state.config.enable_cors,
368 websocket_max_connections: state.config.websocket.max_connections,
369 static_assets_path: state.config.static_dir.to_string_lossy().to_string(),
370 };
371
372 Ok(json_reply(&ApiResponse::success(config)))
373}
374
375async fn metrics_info_handler(
377 _system_state: Arc<RwLock<SystemState>>,
378) -> Result<impl Reply, warp::Rejection> {
379 let metrics_info = MetricsInfo {
380 prometheus_enabled: false,
383 metrics_endpoint: "/metrics".to_string(),
384 custom_metrics_count: get_custom_metrics_count(),
385 last_scrape: get_last_scrape_time().await,
386 };
387
388 Ok(json_reply(&ApiResponse::success(metrics_info)))
389}
390
391async fn maintenance_handler<T>(
393 queue: Arc<T>,
394 request: MaintenanceRequest,
395) -> Result<impl Reply, warp::Rejection>
396where
397 T: JobHistory,
398{
399 let dry_run = request.dry_run.unwrap_or(false);
400
401 match request.operation.as_str() {
402 "cleanup" => {
403 let older_than = chrono::Utc::now() - chrono::Duration::days(7); if dry_run {
405 match queue
406 .count_jobs(None, FinishedKind::Dead, Some(older_than))
407 .await
408 {
409 Ok(count) => {
410 let response = ApiResponse::success(serde_json::json!({
411 "operation": "cleanup",
412 "dry_run": true,
413 "message": format!("Dry run: would clean up {} dead jobs older than 7 days", count),
414 "estimated_deletions": count
415 }));
416 Ok(json_reply(&response))
417 }
418 Err(e) => Ok(error_reply(
419 StatusCode::INTERNAL_SERVER_ERROR,
420 format!("Cleanup dry run failed: {}", e),
421 )),
422 }
423 } else {
424 match queue.purge_dead_jobs(older_than).await {
425 Ok(count) => {
426 let response = ApiResponse::success(serde_json::json!({
427 "operation": "cleanup",
428 "dry_run": false,
429 "message": format!("Cleaned up {} dead jobs", count),
430 "deletions": count
431 }));
432 Ok(json_reply(&response))
433 }
434 Err(e) => Ok(error_reply(
435 StatusCode::INTERNAL_SERVER_ERROR,
436 format!("Cleanup failed: {}", e),
437 )),
438 }
439 }
440 }
441 name => {
442 let Some(operation) = TableMaintenance::parse(name) else {
443 return Ok(error_reply(
444 StatusCode::BAD_REQUEST,
445 format!("Unknown maintenance operation: {}", request.operation),
446 ));
447 };
448 let tables = match maintenance_tables(request.target.as_deref()) {
451 Ok(tables) => tables,
452 Err(message) => return Ok(error_reply(StatusCode::BAD_REQUEST, message)),
453 };
454 match queue.table_maintenance(operation, &tables, dry_run).await {
455 Ok(statements) => {
456 let message = if dry_run {
457 format!("Dry run: would run {} statements", statements.len())
458 } else {
459 format!("Ran {} on {} tables", name, statements.len())
460 };
461 Ok(json_reply(&ApiResponse::success(serde_json::json!({
462 "operation": operation,
463 "dry_run": dry_run,
464 "message": message,
465 "statements": statements,
466 }))))
467 }
468 Err(e) => Ok(error_reply(
469 StatusCode::INTERNAL_SERVER_ERROR,
470 format!("{} failed: {}", name, e),
471 )),
472 }
473 }
474 }
475}
476
477async fn version_handler() -> Result<impl Reply, warp::Rejection> {
479 let version_info = serde_json::json!({
480 "version": env!("CARGO_PKG_VERSION"),
481 "name": env!("CARGO_PKG_NAME"),
482 "description": env!("CARGO_PKG_DESCRIPTION"),
483 "authors": env!("CARGO_PKG_AUTHORS").split(':').collect::<Vec<_>>(),
484 "repository": env!("CARGO_PKG_REPOSITORY"),
485 "license": env!("CARGO_PKG_LICENSE"),
486 "rust_version": get_rust_version(),
487 "build_target": get_target_triple(),
488 });
489
490 Ok(json_reply(&ApiResponse::success(version_info)))
491}
492
493fn get_rust_version() -> String {
495 option_env!("RUSTC_VERSION")
496 .unwrap_or("unknown")
497 .to_string()
498}
499
500fn get_target_triple() -> String {
502 std::env::consts::ARCH.to_string() + "-" + std::env::consts::OS
503}
504
505fn get_memory_usage() -> Option<u64> {
507 #[cfg(target_os = "linux")]
508 {
509 use std::fs;
510 if let Ok(contents) = fs::read_to_string("/proc/self/status") {
511 for line in contents.lines() {
512 if let Some(rest) = line.strip_prefix("VmRSS:")
513 && let Some(kb) = rest
514 .split_whitespace()
515 .next()
516 .and_then(|s| s.parse::<u64>().ok())
517 {
518 return Some(kb.saturating_mul(1024)); }
520 }
521 }
522 }
523
524 None
528}
529
530fn get_custom_metrics_count() -> Option<u32> {
532 None
533}
534
535async fn get_last_scrape_time() -> Option<chrono::DateTime<chrono::Utc>> {
537 None
538}
539
540#[cfg(test)]
541mod tests {
542 use super::*;
543
544 #[test]
546 fn uptime_is_never_negative() {
547 let mut state = SystemState::new(crate::DashboardConfig::default(), "pg".into(), 1);
548 state.started_at = chrono::Utc::now() + chrono::Duration::hours(1);
549 assert_eq!(state.uptime_seconds(), 0);
550 state.started_at = chrono::Utc::now() - chrono::Duration::seconds(90);
551 assert!(state.uptime_seconds() >= 90);
552 }
553
554 #[test]
555 fn test_maintenance_request_deserialization() {
556 let json = r#"{
557 "operation": "cleanup",
558 "target": "completed_jobs",
559 "dry_run": true
560 }"#;
561
562 let request: MaintenanceRequest = serde_json::from_str(json).unwrap();
563 assert_eq!(request.operation, "cleanup");
564 assert_eq!(request.target, Some("completed_jobs".to_string()));
565 assert_eq!(request.dry_run, Some(true));
566 }
567
568 #[test]
569 fn test_build_info_creation() {
570 let build_info = BuildInfo {
571 version: "1.0.0".to_string(),
572 git_commit: Some("abc123".to_string()),
573 build_date: Some("2024-01-01".to_string()),
574 rust_version: "1.70.0".to_string(),
575 target_triple: "x86_64-unknown-linux-gnu".to_string(),
576 };
577
578 let json = serde_json::to_string(&build_info).unwrap();
579 assert!(json.contains("1.0.0"));
580 assert!(json.contains("abc123"));
581 }
582
583 #[test]
584 fn test_unavailable_metrics_are_none_not_zero() {
585 assert_eq!(get_custom_metrics_count(), None);
586 }
587
588 #[tokio::test]
589 async fn test_cleanup_dry_run_reports_database_failure() {
590 let response = maintenance_handler(
591 crate::api::test_support::unreachable_queue(),
592 MaintenanceRequest {
593 operation: "cleanup".to_string(),
594 target: None,
595 dry_run: Some(true),
596 },
597 )
598 .await
599 .unwrap()
600 .into_response();
601 let (status, _) = crate::api::test_support::body_json(response).await;
602 assert_eq!(status, 500);
603 }
604
605 async fn maintenance_status(operation: &str, target: Option<&str>) -> u16 {
606 let response = maintenance_handler(
607 crate::api::test_support::unreachable_queue(),
608 MaintenanceRequest {
609 operation: operation.to_string(),
610 target: target.map(str::to_string),
611 dry_run: Some(false),
612 },
613 )
614 .await
615 .unwrap()
616 .into_response();
617 crate::api::test_support::body_json(response).await.0
618 }
619
620 #[tokio::test]
621 async fn test_table_maintenance_rejects_foreign_tables_before_the_database() {
622 for operation in ["vacuum", "reindex", "optimize"] {
623 assert_eq!(maintenance_status(operation, Some("users")).await, 400);
624 assert_eq!(
626 maintenance_status(operation, Some("hammerwork_jobs")).await,
627 500
628 );
629 }
630 assert_eq!(maintenance_status("defrag", None).await, 400);
631 }
632
633 #[test]
634 fn test_get_rust_version() {
635 let version = get_rust_version();
636 assert!(!version.is_empty());
637 }
638
639 #[test]
640 fn test_get_target_triple() {
641 let target = get_target_triple();
642 assert!(!target.is_empty());
643 assert!(target.contains("-"));
644 }
645}