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