hammerwork_web/api/

stats.rs

//! Statistics and monitoring API endpoints.
//!
//! This module provides comprehensive monitoring and analytics endpoints for tracking
//! system health, performance metrics, and operational insights across all job queues.
//!
//! # API Endpoints
//!
//! - `GET /api/stats/overview` - System overview with key metrics
//! - `GET /api/stats/detailed` - Detailed statistics with historical trends
//! - `GET /api/stats/trends` - Hourly/daily trend analysis
//! - `GET /api/stats/health` - System health check and alerts
//!
//! # Examples
//!
//! ## System Overview
//!
//! ```rust
//! use hammerwork_web::api::stats::{SystemOverview, SystemHealth, SystemAlert};
//! use chrono::Utc;
//!
//! let overview = SystemOverview {
//!     total_queues: 5,
//!     total_jobs: 10000,
//!     pending_jobs: 50,
//!     running_jobs: 10,
//!     completed_jobs: 9800,
//!     failed_jobs: 125,
//!     dead_jobs: 15,
//!     overall_throughput: 150.5,
//!     overall_error_rate: 0.0125,
//!     avg_processing_time_ms: 250.0,
//!     system_health: SystemHealth {
//!         status: "healthy".to_string(),
//!         database_healthy: true,
//!         high_error_rate: false,
//!         queue_backlog: false,
//!         slow_processing: false,
//!         alerts: vec![],
//!     },
//!     uptime_seconds: 86400,
//!     last_updated: Utc::now(),
//! };
//!
//! assert_eq!(overview.total_queues, 5);
//! assert_eq!(overview.overall_error_rate, 0.0125);
//! assert_eq!(overview.system_health.status, "healthy");
//! ```
//!
//! ## Statistics Queries
//!
//! ```rust
//! use hammerwork_web::api::stats::{StatsQuery, TimeRange};
//! use chrono::{Utc, Duration};
//!
//! let time_range = TimeRange {
//!     start: Utc::now() - Duration::hours(24),
//!     end: Utc::now(),
//! };
//!
//! let query = StatsQuery {
//!     time_range: Some(time_range),
//!     queues: Some(vec!["email".to_string(), "notifications".to_string()]),
//!     granularity: Some("hour".to_string()),
//!     hours: None,
//! };
//!
//! assert!(query.time_range.is_some());
//! assert_eq!(query.queues.as_ref().unwrap().len(), 2);
//! assert_eq!(query.granularity, Some("hour".to_string()));
//! ```
//!
//! ## System Alerts
//!
//! ```rust
//! use hammerwork_web::api::stats::SystemAlert;
//! use chrono::Utc;
//!
//! let alert = SystemAlert {
//!     severity: "warning".to_string(),
//!     message: "Queue backlog detected".to_string(),
//!     queue: Some("image_processing".to_string()),
//!     metric: Some("pending_count".to_string()),
//!     value: Some(1500.0),
//!     threshold: Some(1000.0),
//!     timestamp: Utc::now(),
//! };
//!
//! assert_eq!(alert.severity, "warning");
//! assert_eq!(alert.queue, Some("image_processing".to_string()));
//! assert_eq!(alert.value, Some(1500.0));
//! ```
//!
//! ## Performance Metrics
//!
//! ```rust
//! use hammerwork_web::api::stats::PerformanceMetrics;
//!
//! let metrics = PerformanceMetrics {
//!     database_response_time_ms: Some(5.2),
//!     average_queue_depth: 15.5,
//!     jobs_per_second: 8.3,
//!     memory_usage_mb: Some(512.0),
//!     cpu_usage_percent: Some(45.2),
//!     active_workers: None, // not tracked: there is no worker registry
//!     worker_utilization: None,
//! };
//!
//! assert_eq!(metrics.database_response_time_ms, Some(5.2));
//! assert_eq!(metrics.active_workers, None);
//! ```

use super::history::{ErrorGroup, JobHistory, MAX_BUCKETS, hour_floor};
use super::{ApiResponse, error_reply, json_reply, with_shared};
use hammerwork::queue::DatabaseQueue;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use warp::http::StatusCode;
use warp::{Filter, Reply};

/// System overview statistics
#[derive(Debug, Serialize)]
pub struct SystemOverview {
    pub total_queues: u32,
    pub total_jobs: u64,
    pub pending_jobs: u64,
    pub running_jobs: u64,
    pub completed_jobs: u64,
    pub failed_jobs: u64,
    pub dead_jobs: u64,
    pub overall_throughput: f64,
    pub overall_error_rate: f64,
    pub avg_processing_time_ms: f64,
    pub system_health: SystemHealth,
    pub uptime_seconds: u64,
    pub last_updated: chrono::DateTime<chrono::Utc>,
}

/// System health status
#[derive(Debug, Serialize)]
pub struct SystemHealth {
    pub status: String, // "healthy", "degraded", "critical"
    pub database_healthy: bool,
    pub high_error_rate: bool,
    pub queue_backlog: bool,
    pub slow_processing: bool,
    pub alerts: Vec<SystemAlert>,
}

/// System alert
#[derive(Debug, Serialize)]
pub struct SystemAlert {
    pub severity: String, // "info", "warning", "error", "critical"
    pub message: String,
    pub queue: Option<String>,
    pub metric: Option<String>,
    pub value: Option<f64>,
    pub threshold: Option<f64>,
    pub timestamp: chrono::DateTime<chrono::Utc>,
}

/// Detailed statistics for monitoring
#[derive(Debug, Serialize)]
pub struct DetailedStats {
    pub overview: SystemOverview,
    pub queue_stats: Vec<QueueStats>,
    pub hourly_trends: Vec<HourlyTrend>,
    pub error_patterns: Vec<ErrorPattern>,
    pub performance_metrics: PerformanceMetrics,
}

/// Queue statistics
#[derive(Debug, Serialize)]
pub struct QueueStats {
    pub name: String,
    pub pending: u64,
    pub running: u64,
    pub completed_total: u64,
    pub failed_total: u64,
    pub dead_total: u64,
    pub throughput_per_minute: f64,
    pub avg_processing_time_ms: f64,
    pub error_rate: f64,
    pub oldest_pending_age_seconds: Option<u64>,
    pub priority_distribution: HashMap<String, f32>,
}

/// Hourly trend data
#[derive(Debug, Serialize)]
pub struct HourlyTrend {
    pub hour: chrono::DateTime<chrono::Utc>,
    pub completed: u64,
    pub failed: u64,
    pub throughput: f64,
    /// Mean run time of jobs completed in the hour; `null` when none completed.
    pub avg_processing_time_ms: Option<f64>,
    pub error_rate: f64,
}

/// Error pattern analysis
#[derive(Debug, Serialize)]
pub struct ErrorPattern {
    pub error_type: String,
    pub count: u64,
    pub percentage: f64,
    pub sample_message: String,
    pub first_seen: chrono::DateTime<chrono::Utc>,
    pub last_seen: chrono::DateTime<chrono::Utc>,
    pub affected_queues: Vec<String>,
}

/// Performance metrics
#[derive(Debug, Serialize)]
pub struct PerformanceMetrics {
    /// Measured time to fetch the queue statistics for this response.
    pub database_response_time_ms: Option<f64>,
    pub average_queue_depth: f64,
    pub jobs_per_second: f64,
    pub memory_usage_mb: Option<f64>,
    pub cpu_usage_percent: Option<f64>,
    /// Always `null`: Hammerwork has no worker registry, so worker counts are unknown.
    pub active_workers: Option<u32>,
    /// Always `null`, see `active_workers`.
    pub worker_utilization: Option<f64>,
}

/// Time range for statistics queries
#[derive(Debug, Deserialize)]
pub struct TimeRange {
    pub start: chrono::DateTime<chrono::Utc>,
    pub end: chrono::DateTime<chrono::Utc>,
}

/// Statistics query parameters.
///
/// In a query string, `time_range` is `START,END` (two RFC 3339 timestamps, as in
/// `time_range=2024-01-01T00:00:00Z,2024-01-02T00:00:00Z`) and `queues` is a
/// comma-separated list (`queues=email,reports`). A JSON body form (an object with
/// `start` and `end`, and an array of names) is accepted as well.
#[derive(Debug, Deserialize)]
pub struct StatsQuery {
    #[serde(default, deserialize_with = "de_time_range")]
    pub time_range: Option<TimeRange>,
    /// Only these queues are reported; all queues when absent or empty.
    #[serde(default, deserialize_with = "de_queues")]
    pub queues: Option<Vec<String>>,
    /// Size of the trend buckets: `hour` (the default), `day` (UTC days) or `week`
    /// (UTC weeks starting on Monday).
    pub granularity: Option<String>,
    /// The last N hours (alternative to `time_range`, which a query string cannot carry).
    pub hours: Option<u32>,
}

/// Deserializes `queues` from a comma-separated string or a list of names.
fn de_queues<'de, D>(deserializer: D) -> Result<Option<Vec<String>>, D::Error>
where
    D: serde::Deserializer<'de>,
{
    #[derive(Deserialize)]
    #[serde(untagged)]
    enum Raw {
        List(Vec<String>),
        Csv(String),
    }
    Ok(Option::<Raw>::deserialize(deserializer)?
        .map(|raw| match raw {
            Raw::List(names) => names,
            Raw::Csv(csv) => csv.split(',').map(str::to_string).collect(),
        })
        .map(|names| {
            names
                .into_iter()
                .map(|name| name.trim().to_string())
                .filter(|name| !name.is_empty())
                .collect()
        }))
}

/// Deserializes `time_range` from `START,END` or an object with `start` and `end`.
fn de_time_range<'de, D>(deserializer: D) -> Result<Option<TimeRange>, D::Error>
where
    D: serde::Deserializer<'de>,
{
    use serde::de::Error;

    #[derive(Deserialize)]
    #[serde(untagged)]
    enum Raw {
        Range(TimeRange),
        Pair(String),
    }
    match Option::<Raw>::deserialize(deserializer)? {
        None => Ok(None),
        Some(Raw::Range(range)) => Ok(Some(range)),
        Some(Raw::Pair(pair)) => {
            let parse = |text: &str| {
                chrono::DateTime::parse_from_rfc3339(text.trim())
                    .map(|time| time.with_timezone(&chrono::Utc))
                    .map_err(|e| D::Error::custom(format!("invalid time_range `{text}`: {e}")))
            };
            let (start, end) = pair.split_once(',').ok_or_else(|| {
                D::Error::custom("time_range must be `START,END` (RFC 3339 timestamps)")
            })?;
            Ok(Some(TimeRange {
                start: parse(start)?,
                end: parse(end)?,
            }))
        }
    }
}

/// The trend bucket size of a request.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Granularity {
    Hour,
    Day,
    Week,
}

impl Granularity {
    fn parse(name: Option<&str>) -> Result<Self, String> {
        match name.map(|name| name.trim().to_lowercase()).as_deref() {
            None | Some("") | Some("hour") => Ok(Self::Hour),
            Some("day") => Ok(Self::Day),
            Some("week") => Ok(Self::Week),
            Some(other) => Err(format!(
                "Invalid granularity '{other}'. Valid options: hour, day, week"
            )),
        }
    }

    /// The start of the bucket containing `hour`.
    fn bucket_start(self, hour: chrono::DateTime<chrono::Utc>) -> chrono::DateTime<chrono::Utc> {
        use chrono::{Datelike, TimeZone};
        let day = hour.date_naive();
        let day = match self {
            Self::Hour => return hour,
            Self::Day => day,
            Self::Week => {
                day - chrono::Duration::days(i64::from(day.weekday().num_days_from_monday()))
            }
        };
        chrono::Utc.from_utc_datetime(&day.and_hms_opt(0, 0, 0).unwrap_or_default())
    }
}

/// Merges hourly trends (oldest first) into `granularity` buckets. Throughput is jobs per
/// second over the hours merged; the mean run time is weighted by the completed jobs.
fn roll_up(trends: Vec<HourlyTrend>, granularity: Granularity) -> Vec<HourlyTrend> {
    if granularity == Granularity::Hour {
        return trends;
    }
    struct Acc {
        hour: chrono::DateTime<chrono::Utc>,
        hours: u64,
        completed: u64,
        failed: u64,
        processing_ms: f64,
    }
    let mut buckets: Vec<Acc> = Vec::new();
    for trend in trends {
        let start = granularity.bucket_start(trend.hour);
        if buckets.last().is_none_or(|bucket| bucket.hour != start) {
            buckets.push(Acc {
                hour: start,
                hours: 0,
                completed: 0,
                failed: 0,
                processing_ms: 0.0,
            });
        }
        if let Some(bucket) = buckets.last_mut() {
            bucket.hours += 1;
            bucket.completed += trend.completed;
            bucket.failed += trend.failed;
            bucket.processing_ms +=
                trend.avg_processing_time_ms.unwrap_or(0.0) * trend.completed as f64;
        }
    }
    buckets
        .into_iter()
        .map(|bucket| {
            let total = bucket.completed + bucket.failed;
            HourlyTrend {
                hour: bucket.hour,
                completed: bucket.completed,
                failed: bucket.failed,
                throughput: total as f64 / (bucket.hours * 3600) as f64,
                avg_processing_time_ms: (bucket.completed > 0)
                    .then(|| bucket.processing_ms / bucket.completed as f64),
                error_rate: if total > 0 {
                    bucket.failed as f64 / total as f64
                } else {
                    0.0
                },
            }
        })
        .collect()
}

/// Whether `queue_name` is selected by the request's `queues` filter.
fn queue_selected(queues: &Option<Vec<String>>, queue_name: &str) -> bool {
    match queues {
        Some(names) if !names.is_empty() => names.iter().any(|name| name == queue_name),
        _ => true,
    }
}

/// Create statistics routes
pub fn routes<T>(
    queue: Arc<T>,
    system_state: Arc<tokio::sync::RwLock<crate::api::system::SystemState>>,
) -> impl Filter<Extract = impl Reply, Error = warp::Rejection> + Clone
where
    T: JobHistory + 'static,
{
    let queue_filter = with_shared(queue);
    let state_filter = with_shared(system_state);

    let overview = warp::path("stats")
        .and(warp::path("overview"))
        .and(warp::path::end())
        .and(warp::get())
        .and(queue_filter.clone())
        .and(state_filter.clone())
        .and_then(overview_handler);

    let detailed = warp::path("stats")
        .and(warp::path("detailed"))
        .and(warp::path::end())
        .and(warp::get())
        .and(queue_filter.clone())
        .and(state_filter.clone())
        .and(warp::query::<StatsQuery>())
        .and_then(detailed_stats_handler);

    let trends = warp::path("stats")
        .and(warp::path("trends"))
        .and(warp::path::end())
        .and(warp::get())
        .and(queue_filter.clone())
        .and(warp::query::<StatsQuery>())
        .and_then(trends_handler);

    let health = warp::path("stats")
        .and(warp::path("health"))
        .and(warp::path::end())
        .and(warp::get())
        .and(queue_filter)
        .and_then(health_handler);

    overview.or(detailed).or(trends).or(health)
}

/// Handler for system overview statistics
async fn overview_handler<T>(
    queue: Arc<T>,
    system_state: Arc<tokio::sync::RwLock<crate::api::system::SystemState>>,
) -> Result<impl Reply, warp::Rejection>
where
    T: DatabaseQueue + Send + Sync,
{
    match queue.get_all_queue_stats().await {
        Ok(all_stats) => {
            let uptime_seconds =
                u64::try_from(system_state.read().await.uptime_seconds()).unwrap_or(0);
            let overview = generate_overview_from_stats(&all_stats, uptime_seconds);
            Ok(json_reply(&ApiResponse::success(overview)))
        }
        Err(e) => Ok(error_reply(
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("Failed to get statistics: {}", e),
        )),
    }
}

/// Handler for detailed statistics
async fn detailed_stats_handler<T>(
    queue: Arc<T>,
    system_state: Arc<tokio::sync::RwLock<crate::api::system::SystemState>>,
    query: StatsQuery,
) -> Result<impl Reply, warp::Rejection>
where
    T: JobHistory,
{
    let now = chrono::Utc::now();
    let (start, end) = match resolve_range(&query, now) {
        Ok(range) => range,
        Err(message) => return Ok(error_reply(StatusCode::BAD_REQUEST, message)),
    };
    let granularity = match Granularity::parse(query.granularity.as_deref()) {
        Ok(granularity) => granularity,
        Err(message) => return Ok(error_reply(StatusCode::BAD_REQUEST, message)),
    };

    let db_started = std::time::Instant::now();
    match queue.get_all_queue_stats().await {
        Ok(mut all_stats) => {
            all_stats.retain(|stats| queue_selected(&query.queues, &stats.queue_name));
            let db_ms = db_started.elapsed().as_secs_f64() * 1000.0;
            // Convert hammerwork stats to our API format
            let mut queue_stats: Vec<QueueStats> = Vec::new();
            for stats in all_stats.iter() {
                // Calculate oldest pending age seconds
                let oldest_pending_age_seconds = try_api!(
                    calculate_oldest_pending_age(&queue, &stats.queue_name).await,
                    "Failed to get oldest pending job age"
                );

                // Get priority distribution from priority stats
                let priority_distribution = try_api!(
                    get_priority_distribution(&queue, &stats.queue_name).await,
                    "Failed to get priority distribution"
                );

                queue_stats.push(QueueStats {
                    name: stats.queue_name.clone(),
                    pending: stats.pending_count,
                    running: stats.running_count,
                    completed_total: stats.completed_count,
                    failed_total: super::failed_job_count(stats),
                    dead_total: stats.dead_count,
                    throughput_per_minute: stats.statistics.throughput_per_minute,
                    avg_processing_time_ms: stats.statistics.avg_processing_time_ms,
                    error_rate: stats.statistics.error_rate,
                    oldest_pending_age_seconds,
                    priority_distribution,
                });
            }

            let hourly_trends = try_api!(
                compute_trends(&*queue, start, end, &query.queues, granularity).await,
                "Failed to compute hourly trends"
            );
            let error_patterns = try_api!(
                generate_error_patterns(&*queue, start, &query.queues).await,
                "Failed to compute error patterns"
            );
            let performance_metrics = calculate_performance_metrics(&all_stats, db_ms);

            let uptime_seconds =
                u64::try_from(system_state.read().await.uptime_seconds()).unwrap_or(0);
            let overview = generate_overview_from_stats(&all_stats, uptime_seconds);

            let detailed = DetailedStats {
                overview,
                queue_stats,
                hourly_trends,
                error_patterns,
                performance_metrics,
            };

            Ok(json_reply(&ApiResponse::success(detailed)))
        }
        Err(e) => Ok(error_reply(
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("Failed to get detailed statistics: {}", e),
        )),
    }
}

/// The `[start, end)` window of a stats request: `time_range` when given (validated),
/// otherwise the last 24 hours (23 whole hours plus the current one).
fn resolve_range(
    query: &StatsQuery,
    now: chrono::DateTime<chrono::Utc>,
) -> Result<(chrono::DateTime<chrono::Utc>, chrono::DateTime<chrono::Utc>), String> {
    if let Some(hours) = query.hours {
        if hours == 0 || i64::from(hours) > MAX_BUCKETS {
            return Err(format!("hours must be between 1 and {}", MAX_BUCKETS));
        }
        return Ok((
            hour_floor(now) - chrono::Duration::hours(i64::from(hours) - 1),
            now,
        ));
    }
    match &query.time_range {
        Some(range) => {
            if range.start >= range.end {
                Err("time_range.start must be before time_range.end".to_string())
            } else if range.end - range.start > chrono::Duration::hours(MAX_BUCKETS) {
                Err(format!("time_range may span at most {} hours", MAX_BUCKETS))
            } else {
                Ok((range.start, range.end))
            }
        }
        None => Ok((hour_floor(now) - chrono::Duration::hours(23), now)),
    }
}

/// Handler for trend analysis: hourly completed/failed counts from the database.
async fn trends_handler<T>(queue: Arc<T>, query: StatsQuery) -> Result<impl Reply, warp::Rejection>
where
    T: JobHistory,
{
    let (start, end) = match resolve_range(&query, chrono::Utc::now()) {
        Ok(range) => range,
        Err(message) => return Ok(error_reply(StatusCode::BAD_REQUEST, message)),
    };
    let granularity = match Granularity::parse(query.granularity.as_deref()) {
        Ok(granularity) => granularity,
        Err(message) => return Ok(error_reply(StatusCode::BAD_REQUEST, message)),
    };
    let trends = try_api!(
        compute_trends(&*queue, start, end, &query.queues, granularity).await,
        "Failed to compute hourly trends"
    );
    Ok(json_reply(&ApiResponse::success(trends)))
}

/// Handler for system health check
async fn health_handler<T>(queue: Arc<T>) -> Result<impl Reply, warp::Rejection>
where
    T: DatabaseQueue + Send + Sync,
{
    match queue.get_all_queue_stats().await {
        Ok(all_stats) => {
            let health = assess_system_health(&all_stats);
            Ok(json_reply(&ApiResponse::success(health)))
        }
        Err(e) => {
            let health = SystemHealth {
                status: "critical".to_string(),
                database_healthy: false,
                high_error_rate: false,
                queue_backlog: false,
                slow_processing: false,
                alerts: vec![SystemAlert {
                    severity: "critical".to_string(),
                    message: format!("Database connection failed: {}", e),
                    queue: None,
                    metric: Some("database_connectivity".to_string()),
                    value: None,
                    threshold: None,
                    timestamp: chrono::Utc::now(),
                }],
            };
            Ok(json_reply(&ApiResponse::success(health)))
        }
    }
}

/// Assess overall system health based on queue statistics
fn assess_system_health(stats: &[hammerwork::stats::QueueStats]) -> SystemHealth {
    let mut alerts = Vec::new();
    let mut high_error_rate = false;
    let mut queue_backlog = false;
    let mut slow_processing = false;

    for stat in stats {
        // Check error rate
        if stat.statistics.error_rate > 0.1 {
            // > 10% error rate
            high_error_rate = true;
            alerts.push(SystemAlert {
                severity: "warning".to_string(),
                message: format!("High error rate in queue '{}'", stat.queue_name),
                queue: Some(stat.queue_name.clone()),
                metric: Some("error_rate".to_string()),
                value: Some(stat.statistics.error_rate),
                threshold: Some(0.1),
                timestamp: chrono::Utc::now(),
            });
        }

        // Check queue backlog
        if stat.pending_count > 1000 {
            queue_backlog = true;
            alerts.push(SystemAlert {
                severity: "warning".to_string(),
                message: format!("Large backlog in queue '{}'", stat.queue_name),
                queue: Some(stat.queue_name.clone()),
                metric: Some("pending_count".to_string()),
                value: Some(stat.pending_count as f64),
                threshold: Some(1000.0),
                timestamp: chrono::Utc::now(),
            });
        }

        // Check processing time
        if stat.statistics.avg_processing_time_ms > 30000.0 {
            // > 30 seconds
            slow_processing = true;
            alerts.push(SystemAlert {
                severity: "info".to_string(),
                message: format!("Slow processing in queue '{}'", stat.queue_name),
                queue: Some(stat.queue_name.clone()),
                metric: Some("avg_processing_time_ms".to_string()),
                value: Some(stat.statistics.avg_processing_time_ms),
                threshold: Some(30000.0),
                timestamp: chrono::Utc::now(),
            });
        }
    }

    let status = if alerts.iter().any(|a| a.severity == "critical") {
        "critical"
    } else if alerts.iter().any(|a| a.severity == "warning") {
        "degraded"
    } else {
        "healthy"
    };

    SystemHealth {
        status: status.to_string(),
        database_healthy: true, // If we got here, DB is accessible
        high_error_rate,
        queue_backlog,
        slow_processing,
        alerts,
    }
}

/// Generate system overview from queue statistics
fn generate_overview_from_stats(
    stats: &[hammerwork::stats::QueueStats],
    uptime_seconds: u64,
) -> SystemOverview {
    let mut total_pending = 0;
    let mut total_running = 0;
    let mut total_completed = 0;
    let mut total_failed = 0;
    let mut total_dead = 0;
    let mut total_throughput = 0.0;
    let mut total_processing_time = 0.0;
    let queue_count = stats.len();

    for stat in stats {
        total_pending += stat.pending_count;
        total_running += stat.running_count;
        total_completed += stat.completed_count;
        total_failed += super::failed_job_count(stat);
        total_dead += stat.dead_count;
        total_throughput += stat.statistics.throughput_per_minute;
        total_processing_time += stat.statistics.avg_processing_time_ms;
    }

    let avg_processing_time = if queue_count > 0 {
        total_processing_time / queue_count as f64
    } else {
        0.0
    };

    let total_jobs = total_pending + total_running + total_completed + total_failed;
    let overall_error_rate = if total_jobs > 0 {
        total_failed as f64 / total_jobs as f64
    } else {
        0.0
    };

    let health = assess_system_health(stats);

    SystemOverview {
        total_queues: queue_count as u32,
        total_jobs,
        pending_jobs: total_pending,
        running_jobs: total_running,
        completed_jobs: total_completed,
        failed_jobs: total_failed,
        dead_jobs: total_dead,
        overall_throughput: total_throughput,
        overall_error_rate,
        avg_processing_time_ms: avg_processing_time,
        system_health: health,
        uptime_seconds,
        last_updated: chrono::Utc::now(),
    }
}

/// Calculate the oldest pending job age in seconds for a queue
async fn calculate_oldest_pending_age<T>(
    queue: &Arc<T>,
    queue_name: &str,
) -> hammerwork::Result<Option<u64>>
where
    T: DatabaseQueue + Send + Sync,
{
    // Get ready jobs (pending jobs) and find the oldest
    let jobs = queue.get_ready_jobs(queue_name, 100).await?;
    let now = chrono::Utc::now();
    Ok(jobs
        .iter()
        .filter(|job| matches!(job.status, hammerwork::job::JobStatus::Pending))
        .map(|job| age_seconds(now, job.created_at))
        .max())
}

/// Age in whole seconds between `created_at` and `now`, clamped to zero when
/// `created_at` is in the future (clock skew) instead of wrapping to ~1.8e19.
fn age_seconds(
    now: chrono::DateTime<chrono::Utc>,
    created_at: chrono::DateTime<chrono::Utc>,
) -> u64 {
    u64::try_from((now - created_at).num_seconds()).unwrap_or(0)
}

/// Get priority distribution from priority stats for a queue
async fn get_priority_distribution<T>(
    queue: &Arc<T>,
    queue_name: &str,
) -> hammerwork::Result<HashMap<String, f32>>
where
    T: DatabaseQueue + Send + Sync,
{
    let priority_stats = queue.get_priority_stats(queue_name).await?;
    Ok(priority_stats
        .priority_distribution
        .into_iter()
        .map(|(priority, percentage)| (priority.to_string(), percentage))
        .collect())
}

/// Hourly trends for `[start, end)` computed from the database (zero-filled).
async fn compute_trends<T>(
    queue: &T,
    start: chrono::DateTime<chrono::Utc>,
    end: chrono::DateTime<chrono::Utc>,
    queues: &Option<Vec<String>>,
    granularity: Granularity,
) -> hammerwork::Result<Vec<HourlyTrend>>
where
    T: JobHistory,
{
    let buckets = match queues {
        Some(names) if !names.is_empty() => {
            // The selected queues' activity, summed hour by hour.
            let mut merged: Vec<super::history::HourBucket> = Vec::new();
            for name in names {
                let queue_buckets = queue.hourly_activity(Some(name), start, end).await?;
                if merged.is_empty() {
                    merged = queue_buckets;
                    continue;
                }
                for (total, bucket) in merged.iter_mut().zip(queue_buckets) {
                    let done = total.completed + bucket.completed;
                    total.avg_processing_time_ms = if done == 0 {
                        None
                    } else {
                        Some(
                            (total.avg_processing_time_ms.unwrap_or(0.0) * total.completed as f64
                                + bucket.avg_processing_time_ms.unwrap_or(0.0)
                                    * bucket.completed as f64)
                                / done as f64,
                        )
                    };
                    total.completed = done;
                    total.failed += bucket.failed;
                }
            }
            merged
        }
        _ => queue.hourly_activity(None, start, end).await?,
    };
    Ok(roll_up(
        buckets.into_iter().map(trend_from_bucket).collect(),
        granularity,
    ))
}

fn trend_from_bucket(bucket: super::history::HourBucket) -> HourlyTrend {
    let total = bucket.completed + bucket.failed;
    HourlyTrend {
        hour: bucket.hour,
        completed: bucket.completed,
        failed: bucket.failed,
        throughput: total as f64 / 3600.0,
        avg_processing_time_ms: bucket.avg_processing_time_ms,
        error_rate: if total > 0 {
            bucket.failed as f64 / total as f64
        } else {
            0.0
        },
    }
}

/// Error patterns of failures since `since`, grouped by error type.
///
/// Built from the 500 most frequent distinct error messages, so counts and percentages
/// are exact unless the failure set has more distinct messages than that.
async fn generate_error_patterns<T>(
    queue: &T,
    since: chrono::DateTime<chrono::Utc>,
    queues: &Option<Vec<String>>,
) -> hammerwork::Result<Vec<ErrorPattern>>
where
    T: JobHistory,
{
    let mut groups = queue.error_groups(since, 500).await?;
    groups.retain(|group| queue_selected(queues, &group.queue_name));
    Ok(group_error_patterns(groups))
}

/// Folds per-(queue, message) groups into one pattern per error type with real
/// first/last-seen times and the set of affected queues.
fn group_error_patterns(groups: Vec<ErrorGroup>) -> Vec<ErrorPattern> {
    use std::collections::{BTreeMap, BTreeSet};

    struct Agg {
        count: u64,
        sample: (u64, String),
        first_seen: chrono::DateTime<chrono::Utc>,
        last_seen: chrono::DateTime<chrono::Utc>,
        queues: BTreeSet<String>,
    }

    let total: u64 = groups.iter().map(|g| g.count).sum();
    let mut by_type: BTreeMap<String, Agg> = BTreeMap::new();
    for group in groups {
        let error_type = extract_error_type(&group.message);
        let agg = by_type.entry(error_type).or_insert_with(|| Agg {
            count: 0,
            sample: (0, group.message.clone()),
            first_seen: group.first_seen,
            last_seen: group.last_seen,
            queues: BTreeSet::new(),
        });
        agg.count += group.count;
        if group.count > agg.sample.0 {
            agg.sample = (group.count, group.message.clone());
        }
        agg.first_seen = agg.first_seen.min(group.first_seen);
        agg.last_seen = agg.last_seen.max(group.last_seen);
        agg.queues.insert(group.queue_name);
    }

    let mut patterns: Vec<ErrorPattern> = by_type
        .into_iter()
        .map(|(error_type, agg)| ErrorPattern {
            error_type,
            count: agg.count,
            percentage: if total > 0 {
                agg.count as f64 / total as f64 * 100.0
            } else {
                0.0
            },
            sample_message: agg.sample.1,
            first_seen: agg.first_seen,
            last_seen: agg.last_seen,
            affected_queues: agg.queues.into_iter().collect(),
        })
        .collect();
    patterns.sort_by(|a, b| {
        b.count
            .cmp(&a.count)
            .then_with(|| a.error_type.cmp(&b.error_type))
    });
    patterns
}

/// Calculate performance metrics from queue statistics.
///
/// `database_response_time_ms` is the measured time of the statistics query. Worker
/// counts and utilization are `None`: running jobs are not workers, and there is no
/// worker registry to count them from.
fn calculate_performance_metrics(
    all_stats: &[hammerwork::stats::QueueStats],
    database_response_time_ms: f64,
) -> PerformanceMetrics {
    let total_throughput = all_stats
        .iter()
        .map(|s| s.statistics.throughput_per_minute)
        .sum::<f64>();

    let average_queue_depth = if !all_stats.is_empty() {
        all_stats
            .iter()
            .map(|s| s.pending_count as f64)
            .sum::<f64>()
            / all_stats.len() as f64
    } else {
        0.0
    };

    PerformanceMetrics {
        database_response_time_ms: Some(database_response_time_ms),
        average_queue_depth,
        jobs_per_second: total_throughput / 60.0, // Convert from per minute to per second
        memory_usage_mb: None,
        cpu_usage_percent: None,
        active_workers: None,
        worker_utilization: None,
    }
}

/// Extract error type from error message for grouping
fn extract_error_type(error_msg: &str) -> String {
    // Simple error classification logic
    if error_msg.contains("timeout") || error_msg.contains("Timeout") {
        "Timeout Error".to_string()
    } else if error_msg.contains("connection") || error_msg.contains("Connection") {
        "Connection Error".to_string()
    } else if error_msg.contains("parse")
        || error_msg.contains("Parse")
        || error_msg.contains("invalid")
    {
        "Parse Error".to_string()
    } else if error_msg.contains("permission")
        || error_msg.contains("Permission")
        || error_msg.contains("forbidden")
    {
        "Permission Error".to_string()
    } else if error_msg.contains("not found") || error_msg.contains("Not Found") {
        "Not Found Error".to_string()
    } else {
        // Use first word of error message as type
        error_msg
            .split_whitespace()
            .next()
            .map(|s| format!("{} Error", s))
            .unwrap_or_else(|| "Unknown Error".to_string())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::api::test_support::{body_json, unreachable_queue};
    use chrono::TimeZone;

    #[test]
    fn test_overview_counts_failed_dead_and_timed_out_jobs_as_failed() {
        let mut stats = hammerwork::stats::QueueStats {
            queue_name: "q".to_string(),
            pending_count: 1,
            running_count: 0,
            dead_count: 2,
            timed_out_count: 3,
            completed_count: 4,
            statistics: Default::default(),
        };
        // Jobs in status `Failed` (a failed dependency, a fail-fast batch, `fail_job`)
        stats.statistics.failed = 5;
        let overview = generate_overview_from_stats(&[stats], 0);
        assert_eq!(overview.failed_jobs, 5 + 2 + 3);
        assert_eq!(overview.dead_jobs, 2);
        assert_eq!(overview.total_jobs, 1 + 4 + 10);
        assert!((overview.overall_error_rate - 10.0 / 15.0).abs() < 1e-9);
    }

    fn state() -> Arc<tokio::sync::RwLock<crate::api::system::SystemState>> {
        Arc::new(tokio::sync::RwLock::new(
            crate::api::system::SystemState::new(
                crate::DashboardConfig::default(),
                "PostgreSQL".to_string(),
                1,
            ),
        ))
    }

    fn empty_query() -> StatsQuery {
        serde_json::from_value(serde_json::json!({})).unwrap()
    }

    #[tokio::test]
    async fn test_detailed_stats_returns_500_when_database_is_down() {
        let response = detailed_stats_handler(unreachable_queue(), state(), empty_query())
            .await
            .unwrap()
            .into_response();
        let (status, body) = body_json(response).await;
        assert_eq!(status, 500);
        assert_eq!(body["success"], false);
    }

    #[tokio::test]
    async fn test_trends_returns_500_when_database_is_down() {
        let response = trends_handler(unreachable_queue(), empty_query())
            .await
            .unwrap()
            .into_response();
        let (status, body) = body_json(response).await;
        assert_eq!(status, 500);
        assert_eq!(body["success"], false);
    }

    #[tokio::test]
    async fn test_trends_rejects_invalid_time_range() {
        let inverted: StatsQuery = serde_json::from_value(serde_json::json!({
            "time_range": {"start": "2024-01-02T00:00:00Z", "end": "2024-01-01T00:00:00Z"}
        }))
        .unwrap();
        let response = trends_handler(unreachable_queue(), inverted)
            .await
            .unwrap()
            .into_response();
        assert_eq!(body_json(response).await.0, 400);

        let too_long: StatsQuery = serde_json::from_value(serde_json::json!({
            "time_range": {"start": "2020-01-01T00:00:00Z", "end": "2024-01-01T00:00:00Z"}
        }))
        .unwrap();
        let response = trends_handler(unreachable_queue(), too_long)
            .await
            .unwrap()
            .into_response();
        assert_eq!(body_json(response).await.0, 400);
    }

    #[test]
    fn test_default_range_is_24_hourly_buckets() {
        let now = chrono::Utc::now();
        let (start, end) = resolve_range(&empty_query(), now).unwrap();
        assert_eq!(end, now);
        assert_eq!(crate::api::history::hour_range(start, end).len(), 24);
    }

    #[test]
    fn test_trend_from_bucket_computes_rates_without_inventing_values() {
        let hour = hour_floor(chrono::Utc::now());
        let busy = trend_from_bucket(crate::api::history::HourBucket {
            hour,
            completed: 3,
            failed: 1,
            avg_processing_time_ms: Some(40.0),
        });
        assert_eq!(busy.error_rate, 0.25);
        assert_eq!(busy.throughput, 4.0 / 3600.0);
        let idle = trend_from_bucket(crate::api::history::HourBucket {
            hour,
            completed: 0,
            failed: 0,
            avg_processing_time_ms: None,
        });
        assert_eq!(idle.error_rate, 0.0);
        assert_eq!(idle.avg_processing_time_ms, None);
    }

    #[test]
    fn test_group_error_patterns_uses_real_times_and_queues() {
        let now = chrono::Utc::now();
        let t = |h| now - chrono::Duration::hours(h);
        let group = |queue: &str, msg: &str, count, first, last| ErrorGroup {
            queue_name: queue.to_string(),
            message: msg.to_string(),
            count,
            first_seen: t(first),
            last_seen: t(last),
        };
        let patterns = group_error_patterns(vec![
            group("emails", "connection refused by host a", 6, 10, 2),
            group("reports", "connection reset", 2, 8, 1),
            group("emails", "request timeout", 2, 5, 3),
        ]);
        assert_eq!(patterns.len(), 2);
        let conn = &patterns[0];
        assert_eq!(conn.error_type, "Connection Error");
        assert_eq!(conn.count, 8);
        assert_eq!(conn.percentage, 80.0);
        assert_eq!(conn.sample_message, "connection refused by host a");
        assert_eq!(conn.affected_queues, vec!["emails", "reports"]);
        assert_eq!(conn.first_seen, t(10));
        assert_eq!(conn.last_seen, t(1));
        assert_eq!(patterns[1].affected_queues, vec!["emails"]);
        assert!(group_error_patterns(Vec::new()).is_empty());
    }

    #[test]
    fn test_performance_metrics_report_unknowns_as_none() {
        let metrics = calculate_performance_metrics(&[], 3.5);
        assert_eq!(metrics.database_response_time_ms, Some(3.5));
        assert_eq!(metrics.active_workers, None);
        assert_eq!(metrics.worker_utilization, None);
        assert_eq!(metrics.memory_usage_mb, None);
    }

    #[test]
    fn test_age_seconds_clamps_future_timestamps_to_zero() {
        let now = chrono::Utc::now();
        assert_eq!(age_seconds(now, now + chrono::Duration::seconds(30)), 0);
        assert_eq!(age_seconds(now, now - chrono::Duration::seconds(30)), 30);
    }

    #[test]
    fn test_stats_query_deserialization() {
        let json = r#"{
            "time_range": {
                "start": "2024-01-01T00:00:00Z",
                "end": "2024-01-02T00:00:00Z"
            },
            "queues": ["email", "data-processing"],
            "granularity": "hour"
        }"#;

        let query: StatsQuery = serde_json::from_str(json).unwrap();
        assert!(query.time_range.is_some());
        assert_eq!(query.queues.as_ref().unwrap().len(), 2);
        assert_eq!(query.granularity, Some("hour".to_string()));
    }

    /// #103: `time_range` and `queues` decode from a real query string.
    #[tokio::test]
    async fn test_stats_query_decodes_from_a_query_string() {
        let decode = |query: &str| {
            let path = format!("/stats?{query}");
            async move {
                warp::test::request()
                    .path(&path)
                    .filter(&warp::query::<StatsQuery>())
                    .await
            }
        };
        let query = decode(
            "queues=email,%20reports,&granularity=day\
             &time_range=2024-01-01T00:00:00Z,2024-01-02T00:00:00Z",
        )
        .await
        .unwrap();
        assert_eq!(query.queues.unwrap(), vec!["email", "reports"]);
        assert_eq!(query.granularity.as_deref(), Some("day"));
        let range = query.time_range.unwrap();
        assert_eq!(range.start.to_rfc3339(), "2024-01-01T00:00:00+00:00");
        assert_eq!(range.end - range.start, chrono::Duration::days(1));

        let bare = decode("hours=6").await.unwrap();
        assert!(bare.time_range.is_none() && bare.queues.is_none());
        assert!(decode("time_range=yesterday").await.is_err());
        assert!(decode("time_range=2024-01-01T00:00:00Z").await.is_err());
    }

    #[tokio::test]
    async fn test_stats_routes_accept_query_string_filters() {
        let routes = routes(Arc::new(hammerwork::queue::test::TestQueue::new()), state());
        let status = |path: String| {
            let routes = routes.clone();
            async move {
                warp::test::request()
                    .path(&path)
                    .reply(&routes)
                    .await
                    .status()
                    .as_u16()
            }
        };
        let range = "time_range=2024-01-01T00:00:00Z,2024-01-02T00:00:00Z";
        assert_eq!(
            status(format!("/stats/trends?{range}&queues=a,b&granularity=week")).await,
            200
        );
        assert_eq!(
            status("/stats/trends?granularity=month".to_string()).await,
            400
        );
        assert_eq!(
            status("/stats/trends?time_range=2024-01-02T00:00:00Z,2024-01-01T00:00:00Z".into())
                .await,
            400
        );
    }

    #[test]
    fn test_roll_up_merges_hours_into_days_and_weeks() {
        let at = |day: u32, hour: u32| {
            chrono::Utc
                .with_ymd_and_hms(2024, 1, day, hour, 0, 0)
                .unwrap()
        };
        let trend = |hour, completed, failed, avg| HourlyTrend {
            hour,
            completed,
            failed,
            throughput: 0.0,
            avg_processing_time_ms: avg,
            error_rate: 0.0,
        };
        // Sunday the 7th and Monday the 8th (UTC), three hours in all.
        let hours = || {
            vec![
                trend(at(7, 22), 1, 0, Some(100.0)),
                trend(at(7, 23), 3, 1, Some(200.0)),
                trend(at(8, 0), 0, 0, None),
            ]
        };

        assert_eq!(roll_up(hours(), Granularity::Hour).len(), 3);

        let days = roll_up(hours(), Granularity::Day);
        assert_eq!(days.len(), 2);
        assert_eq!(days[0].hour, at(7, 0));
        assert_eq!((days[0].completed, days[0].failed), (4, 1));
        assert!((days[0].avg_processing_time_ms.unwrap() - 175.0).abs() < 1e-9);
        assert!((days[0].throughput - 5.0 / 7200.0).abs() < 1e-12);
        assert!((days[0].error_rate - 0.2).abs() < 1e-12);
        assert_eq!(days[1].hour, at(8, 0));
        assert_eq!(days[1].avg_processing_time_ms, None);

        let weeks = roll_up(hours(), Granularity::Week);
        assert_eq!(
            weeks.len(),
            2,
            "the Sunday and the Monday are in different weeks"
        );
        assert_eq!(weeks[0].hour, at(1, 0), "Monday the 1st");
        assert_eq!(weeks[1].hour, at(8, 0));
    }

    #[test]
    fn test_granularity_and_queue_filters() {
        assert_eq!(Granularity::parse(None), Ok(Granularity::Hour));
        assert_eq!(Granularity::parse(Some(" DAY ")), Ok(Granularity::Day));
        assert!(Granularity::parse(Some("month")).is_err());
        let all: Option<Vec<String>> = None;
        let some = Some(vec!["a".to_string()]);
        assert!(queue_selected(&all, "x") && queue_selected(&Some(vec![]), "x"));
        assert!(queue_selected(&some, "a") && !queue_selected(&some, "b"));
    }

    #[test]
    fn test_system_alert_serialization() {
        let alert = SystemAlert {
            severity: "warning".to_string(),
            message: "High error rate detected".to_string(),
            queue: Some("email".to_string()),
            metric: Some("error_rate".to_string()),
            value: Some(0.15),
            threshold: Some(0.1),
            timestamp: chrono::Utc::now(),
        };

        let json = serde_json::to_string(&alert).unwrap();
        assert!(json.contains("warning"));
        assert!(json.contains("High error rate"));
    }
}