//! Monitoring service implementation

use crate::proto::monitoring::{
    monitoring_service_server::MonitoringService, AcknowledgeAlertRequest,
    AcknowledgeAlertResponse, Alert, AlertEvent, AlertEventType, AlertSeverity, AlertStatus,
    AlertType, GetActiveAlertsRequest, GetActiveAlertsResponse, GetEpochHistoryRequest,
    GetEpochHistoryResponse, GetHealthCheckRequest, GetHealthCheckResponse,
    GetLatencyMetricsRequest, GetLatencyMetricsResponse, GetLiveTrainingMetricsRequest,
    GetLiveTrainingMetricsResponse, GetMetricsRequest, GetMetricsResponse,
    GetSystemStatusRequest, GetSystemStatusResponse, GetThroughputMetricsRequest,
    GetThroughputMetricsResponse, HealthCheck, HealthStatus as ProtoHealthStatus, LatencyMetric,
    Metric, MetricType, MetricsEvent, ServiceHealth, ServiceState, ServiceStatus,
    StreamAlertsRequest, StreamMetricsRequest, StreamSystemStatusRequest,
    StreamTrainingMetricsRequest, SubscribeClusterPodsRequest, ClusterPodsResponse,
    SystemHealth, SystemMetrics, SystemStatus, SystemStatusChangeType, SystemStatusEvent,
    ThroughputMetric,
};
use crate::state::{HealthStatus, TradingServiceState};
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::RwLock;
use tonic::{Request, Response, Status};
use tracing::{info, instrument};

/// Monitoring service implementation
#[derive(Debug, Clone)]
pub struct MonitoringServiceImpl {
    state: TradingServiceState,
    /// Tracks when this service instance was started for real uptime reporting
    started_at: Instant,
    /// System info handle for CPU/memory metrics (refreshed on each status query)
    sys: Arc<RwLock<sysinfo::System>>,
    /// Disk info handle (refreshed on each status query)
    disks: Arc<RwLock<sysinfo::Disks>>,
    /// In-memory alert store shared between stream_alerts, acknowledge_alert, and get_active_alerts
    alerts: Arc<RwLock<Vec<Alert>>>,
}

impl MonitoringServiceImpl {
    /// Create new monitoring service
    pub fn new(state: TradingServiceState) -> Self {
        // Construct with initial refresh so the first CPU reading has a baseline.
        let mut sys = sysinfo::System::new();
        sys.refresh_cpu_all();
        sys.refresh_memory();

        Self {
            state,
            started_at: Instant::now(),
            sys: Arc::new(RwLock::new(sys)),
            disks: Arc::new(RwLock::new(sysinfo::Disks::new_with_refreshed_list())),
            alerts: Arc::new(RwLock::new(Vec::new())),
        }
    }
}

/// Collected system-level metrics (CPU, memory, disk).
struct LiveSystemMetrics {
    cpu_usage_percent: f64,
    memory_usage_percent: f64,
    disk_usage_percent: f64,
    /// Network I/O requires sustained measurement; always 0.0 for now.
    network_io_mbps: f64,
}

/// Collect real system metrics from sysinfo handles.
///
/// CPU usage is computed as the global average across all cores since the last
/// refresh. Memory and disk are point-in-time snapshots.  The function is
/// non-blocking apart from acquiring the `RwLock`.
async fn collect_system_metrics(
    sys: &Arc<RwLock<sysinfo::System>>,
    disks: &Arc<RwLock<sysinfo::Disks>>,
) -> LiveSystemMetrics {
    // --- CPU + Memory --------------------------------------------------
    let (cpu_usage, mem_pct) = {
        let mut s = sys.write().await;
        // Refresh CPU counters (delta since last refresh).
        s.refresh_cpu_all();
        s.refresh_memory();

        let cpu = s.global_cpu_usage() as f64; // 0-100 across all cores

        let total = s.total_memory();
        let used = s.used_memory();
        let mem = if total > 0 {
            (used as f64 / total as f64) * 100.0
        } else {
            0.0
        };
        (cpu, mem)
    };

    // --- Disk ----------------------------------------------------------
    let disk_pct = {
        let mut d = disks.write().await;
        d.refresh(true);

        let (total_space, available_space) =
            d.iter().fold((0u64, 0u64), |(ts, avail), disk| {
                (ts + disk.total_space(), avail + disk.available_space())
            });
        if total_space > 0 {
            let used_space = total_space.saturating_sub(available_space);
            (used_space as f64 / total_space as f64) * 100.0
        } else {
            0.0
        }
    };

    LiveSystemMetrics {
        cpu_usage_percent: cpu_usage,
        memory_usage_percent: mem_pct,
        disk_usage_percent: disk_pct,
        // Network I/O measurement requires sustained sampling; leave as 0.0.
        network_io_mbps: 0.0,
    }
}

#[tonic::async_trait]
impl MonitoringService for MonitoringServiceImpl {
    #[instrument(skip_all)]
    async fn get_health_check(
        &self,
        _request: Request<GetHealthCheckRequest>,
    ) -> Result<Response<GetHealthCheckResponse>, Status> {
        // Check health of all components
        let health_status = self.state.get_health_status().await;

        // Convert internal health status to proto health status
        let proto_health_status = match health_status {
            HealthStatus::Healthy => ProtoHealthStatus::Healthy as i32,
            HealthStatus::Degraded => ProtoHealthStatus::Degraded as i32,
            HealthStatus::Unhealthy => ProtoHealthStatus::Unhealthy as i32,
            HealthStatus::Critical => ProtoHealthStatus::Critical as i32,
        };

        let health_checks = vec![HealthCheck {
            check_name: "market_data".to_string(),
            status: proto_health_status,
            message: Some("Market data providers status".to_string()),
            response_time_ms: Some(0.1),
            last_checked: chrono::Utc::now().timestamp(),
            details: std::collections::HashMap::new(),
        }];

        Ok(Response::new(GetHealthCheckResponse {
            health_status: proto_health_status,
            health_checks,
            timestamp: chrono::Utc::now().timestamp(),
        }))
    }

    #[instrument(skip_all)]
    async fn get_metrics(
        &self,
        _request: Request<GetMetricsRequest>,
    ) -> Result<Response<GetMetricsResponse>, Status> {
        // Get basic metrics from config manager
        let cache_total = 100; // Placeholder - config_manager doesn't expose cache stats
        let cache_expired = 0;

        let metrics = vec![
            Metric {
                name: "config_cache_total_entries".to_string(),
                metric_type: MetricType::Gauge as i32,
                value: cache_total as f64,
                unit: "count".to_string(),
                labels: std::collections::HashMap::new(),
                timestamp: chrono::Utc::now().timestamp(),
                statistics: None,
            },
            Metric {
                name: "config_cache_expired_entries".to_string(),
                metric_type: MetricType::Gauge as i32,
                value: cache_expired as f64,
                unit: "count".to_string(),
                labels: std::collections::HashMap::new(),
                timestamp: chrono::Utc::now().timestamp(),
                statistics: None,
            },
            Metric {
                name: "config_cache_hit_ratio".to_string(),
                metric_type: MetricType::Gauge as i32,
                value: if cache_total > 0 {
                    (cache_total - cache_expired) as f64 / cache_total as f64
                } else {
                    1.0
                },
                unit: "ratio".to_string(),
                labels: std::collections::HashMap::new(),
                timestamp: chrono::Utc::now().timestamp(),
                statistics: None,
            },
        ];

        Ok(Response::new(GetMetricsResponse {
            metrics,
            timestamp: chrono::Utc::now().timestamp(),
        }))
    }

    #[instrument(skip_all)]
    async fn get_system_status(
        &self,
        _request: Request<GetSystemStatusRequest>,
    ) -> Result<Response<GetSystemStatusResponse>, Status> {
        let health_status = self.state.get_health_status().await;

        let overall_health = match health_status {
            HealthStatus::Healthy => SystemHealth::Healthy,
            HealthStatus::Degraded => SystemHealth::Degraded,
            HealthStatus::Unhealthy => SystemHealth::Unhealthy,
            HealthStatus::Critical => SystemHealth::Critical,
        };

        // Real uptime measured from service construction (cast to i64 to match proto int64 field)
        let uptime_seconds = self.started_at.elapsed().as_secs() as i64;

        // Check kill switch status and surface any active triggers as critical issues
        let mut critical_issues: Vec<String> = vec![];
        if let Some(kill_switch) = &self.state.kill_switch_system {
            match kill_switch.get_status().await {
                Ok(ks_status) => {
                    if ks_status.is_active {
                        critical_issues
                            .push("Kill switch is ACTIVE — trading halted".to_string());
                    }
                    if ks_status.is_emergency_active {
                        critical_issues.push("Emergency shutdown is ACTIVE".to_string());
                    }
                    if !ks_status.is_healthy {
                        critical_issues.push("Kill switch system is UNHEALTHY".to_string());
                    }
                }
                Err(_) => {
                    critical_issues.push("Kill switch status unavailable".to_string());
                }
            }
        }

        // Collect real system metrics (CPU, memory, disk) from sysinfo.
        let live = collect_system_metrics(&self.sys, &self.disks).await;

        let system_status = SystemStatus {
            overall_health: overall_health as i32,
            healthy_services: 3,
            total_services: 3,
            critical_issues,
            system_uptime_seconds: uptime_seconds,
            system_metrics: Some(SystemMetrics {
                cpu_usage_percent: live.cpu_usage_percent,
                memory_usage_percent: live.memory_usage_percent,
                disk_usage_percent: live.disk_usage_percent,
                network_io_mbps: live.network_io_mbps,
                active_connections: 25,
                total_requests: 1000,
                avg_response_time_ms: 2.5,
                error_rate_percent: 0.1,
            }),
        };

        let service_statuses = vec![ServiceStatus {
            service_name: "trading_service".to_string(),
            health: ServiceHealth::Healthy as i32,
            state: ServiceState::Running as i32,
            version: Some("1.0.0".to_string()),
            error_message: None,
            uptime_seconds,
            last_health_check: chrono::Utc::now().timestamp(),
            metadata: std::collections::HashMap::new(),
            dependencies: vec![],
        }];

        Ok(Response::new(GetSystemStatusResponse {
            overall_status: Some(system_status),
            service_statuses,
            timestamp: chrono::Utc::now().timestamp(),
        }))
    }

    #[instrument(skip_all)]
    async fn get_latency_metrics(
        &self,
        _request: Request<GetLatencyMetricsRequest>,
    ) -> Result<Response<GetLatencyMetricsResponse>, Status> {
        let latency_metrics = vec![LatencyMetric {
            service_name: "trading_service".to_string(),
            operation_name: "submit_order".to_string(),
            avg_latency_ms: 1.5,
            p50_latency_ms: 1.2,
            p95_latency_ms: 3.5,
            p99_latency_ms: 8.0,
            max_latency_ms: 15.0,
            request_count: 1000,
            time_window_start: chrono::Utc::now().timestamp() - 3600,
            time_window_end: chrono::Utc::now().timestamp(),
        }];

        Ok(Response::new(GetLatencyMetricsResponse { latency_metrics }))
    }

    #[instrument(skip_all)]
    async fn get_throughput_metrics(
        &self,
        _request: Request<GetThroughputMetricsRequest>,
    ) -> Result<Response<GetThroughputMetricsResponse>, Status> {
        let throughput_metrics = vec![ThroughputMetric {
            service_name: "trading_service".to_string(),
            operation_name: "submit_order".to_string(),
            requests_per_second: 50.0,
            bytes_per_second: 25000.0,
            total_requests: 180000,
            total_bytes: 90000000,
            time_window_start: chrono::Utc::now().timestamp() - 3600,
            time_window_end: chrono::Utc::now().timestamp(),
        }];

        Ok(Response::new(GetThroughputMetricsResponse {
            throughput_metrics,
        }))
    }

    #[instrument(skip_all)]
    async fn acknowledge_alert(
        &self,
        request: Request<AcknowledgeAlertRequest>,
    ) -> Result<Response<AcknowledgeAlertResponse>, Status> {
        let req = request.into_inner();
        let now = chrono::Utc::now().timestamp();

        let mut store = self.alerts.write().await;
        let found = store
            .iter_mut()
            .find(|a| a.alert_id == req.alert_id);

        match found {
            Some(alert) => {
                alert.acknowledged_at = Some(now);
                alert.acknowledged_by = Some(req.acknowledged_by.clone());
                alert.status = AlertStatus::Acknowledged as i32;
                info!(
                    alert_id = %req.alert_id,
                    acknowledged_by = %req.acknowledged_by,
                    "Alert acknowledged"
                );
                Ok(Response::new(AcknowledgeAlertResponse {
                    success: true,
                    message: format!(
                        "Alert {} acknowledged by {}",
                        req.alert_id, req.acknowledged_by
                    ),
                    timestamp: now,
                }))
            }
            None => Ok(Response::new(AcknowledgeAlertResponse {
                success: false,
                message: format!("Alert {} not found", req.alert_id),
                timestamp: now,
            })),
        }
    }

    #[instrument(skip_all)]
    async fn get_active_alerts(
        &self,
        _request: Request<GetActiveAlertsRequest>,
    ) -> Result<Response<GetActiveAlertsResponse>, Status> {
        let store = self.alerts.read().await;
        let active_alerts: Vec<Alert> = store
            .iter()
            .filter(|a| a.status == AlertStatus::Active as i32)
            .cloned()
            .collect();
        let total_count = active_alerts.len() as i32;
        Ok(Response::new(GetActiveAlertsResponse {
            active_alerts,
            total_count,
        }))
    }

    // Streaming methods (unimplemented for now)
    type StreamSystemStatusStream = std::pin::Pin<
        Box<dyn tokio_stream::Stream<Item = Result<SystemStatusEvent, Status>> + Send>,
    >;

    #[instrument(skip_all)]
    async fn stream_system_status(
        &self,
        request: Request<StreamSystemStatusRequest>,
    ) -> Result<Response<Self::StreamSystemStatusStream>, Status> {
        let req = request.into_inner();
        let update_freq = req.update_frequency_seconds.unwrap_or(5).max(1) as u64;

        info!("StreamSystemStatus started: freq={}s", update_freq);

        let (tx, rx) = tokio::sync::mpsc::channel(128);
        let state = self.state.clone();
        let started_at = self.started_at;
        let sys = Arc::clone(&self.sys);
        let disks = Arc::clone(&self.disks);
        let mut previous_health: Option<i32> = None;

        tokio::spawn(async move {
            let mut interval =
                tokio::time::interval(std::time::Duration::from_secs(update_freq));

            loop {
                interval.tick().await;

                let health_status = state.get_health_status().await;
                let overall_health = match health_status {
                    HealthStatus::Healthy => SystemHealth::Healthy,
                    HealthStatus::Degraded => SystemHealth::Degraded,
                    HealthStatus::Unhealthy => SystemHealth::Unhealthy,
                    HealthStatus::Critical => SystemHealth::Critical,
                };

                let uptime_seconds = started_at.elapsed().as_secs() as i64;

                // Check kill switch for critical issues
                let mut critical_issues: Vec<String> = Vec::new();
                if let Some(kill_switch) = &state.kill_switch_system {
                    if let Ok(ks_status) = kill_switch.get_status().await {
                        if ks_status.is_active {
                            critical_issues
                                .push("Kill switch is ACTIVE — trading halted".to_string());
                        }
                        if ks_status.is_emergency_active {
                            critical_issues.push("Emergency shutdown is ACTIVE".to_string());
                        }
                    }
                }

                // Collect real system metrics
                let live = collect_system_metrics(&sys, &disks).await;

                let system_status = SystemStatus {
                    overall_health: overall_health as i32,
                    healthy_services: 3,
                    total_services: 3,
                    critical_issues,
                    system_uptime_seconds: uptime_seconds,
                    system_metrics: Some(SystemMetrics {
                        cpu_usage_percent: live.cpu_usage_percent,
                        memory_usage_percent: live.memory_usage_percent,
                        disk_usage_percent: live.disk_usage_percent,
                        network_io_mbps: live.network_io_mbps,
                        active_connections: 0,
                        total_requests: 0,
                        avg_response_time_ms: 0.0,
                        error_rate_percent: 0.0,
                    }),
                };

                // Determine change type
                let current_health = overall_health as i32;
                let change_type = match previous_health {
                    Some(prev) if current_health < prev => {
                        SystemStatusChangeType::HealthImproved
                    },
                    Some(prev) if current_health > prev => {
                        SystemStatusChangeType::HealthDegraded
                    },
                    _ => SystemStatusChangeType::Unspecified,
                };
                previous_health = Some(current_health);

                let event = SystemStatusEvent {
                    system_status: Some(system_status),
                    change_type: change_type as i32,
                    timestamp: chrono::Utc::now().timestamp(),
                };

                if tx.send(Ok(event)).await.is_err() {
                    info!("StreamSystemStatus: client disconnected");
                    break;
                }
            }
        });

        Ok(Response::new(Box::pin(
            tokio_stream::wrappers::ReceiverStream::new(rx),
        )))
    }

    type StreamMetricsStream =
        std::pin::Pin<Box<dyn tokio_stream::Stream<Item = Result<MetricsEvent, Status>> + Send>>;

    #[instrument(skip_all)]
    async fn stream_metrics(
        &self,
        request: Request<StreamMetricsRequest>,
    ) -> Result<Response<Self::StreamMetricsStream>, Status> {
        let req = request.into_inner();
        let update_freq = req.update_frequency_seconds.unwrap_or(5).max(1) as u64;

        info!("StreamMetrics started: freq={}s", update_freq);

        let (tx, rx) = tokio::sync::mpsc::channel(128);
        let state = self.state.clone();
        let started_at = self.started_at;
        let sys = Arc::clone(&self.sys);
        let disks = Arc::clone(&self.disks);

        tokio::spawn(async move {
            let mut interval =
                tokio::time::interval(std::time::Duration::from_secs(update_freq));

            loop {
                interval.tick().await;

                let now = chrono::Utc::now().timestamp();
                let uptime_seconds = started_at.elapsed().as_secs() as f64;

                // Collect live system metrics for CPU/memory reporting
                let live = collect_system_metrics(&sys, &disks).await;

                // Read the shared metrics store for trading-specific counters
                let system_metrics = state.metrics.read().await;

                let metrics = vec![
                    Metric {
                        name: "system_uptime_seconds".to_string(),
                        metric_type: MetricType::Gauge as i32,
                        value: uptime_seconds,
                        unit: "seconds".to_string(),
                        labels: std::collections::HashMap::new(),
                        timestamp: now,
                        statistics: None,
                    },
                    Metric {
                        name: "cpu_usage_percent".to_string(),
                        metric_type: MetricType::Gauge as i32,
                        value: live.cpu_usage_percent,
                        unit: "percent".to_string(),
                        labels: std::collections::HashMap::new(),
                        timestamp: now,
                        statistics: None,
                    },
                    Metric {
                        name: "memory_usage_percent".to_string(),
                        metric_type: MetricType::Gauge as i32,
                        value: live.memory_usage_percent,
                        unit: "percent".to_string(),
                        labels: std::collections::HashMap::new(),
                        timestamp: now,
                        statistics: None,
                    },
                    Metric {
                        name: "active_connections".to_string(),
                        metric_type: MetricType::Gauge as i32,
                        value: system_metrics.active_connections as f64,
                        unit: "count".to_string(),
                        labels: std::collections::HashMap::new(),
                        timestamp: now,
                        statistics: None,
                    },
                    Metric {
                        name: "total_requests".to_string(),
                        metric_type: MetricType::Counter as i32,
                        value: system_metrics.total_requests as f64,
                        unit: "count".to_string(),
                        labels: std::collections::HashMap::new(),
                        timestamp: now,
                        statistics: None,
                    },
                    Metric {
                        name: "avg_response_time_ms".to_string(),
                        metric_type: MetricType::Gauge as i32,
                        value: system_metrics.avg_response_time_ms,
                        unit: "milliseconds".to_string(),
                        labels: std::collections::HashMap::new(),
                        timestamp: now,
                        statistics: None,
                    },
                    Metric {
                        name: "error_rate_percent".to_string(),
                        metric_type: MetricType::Gauge as i32,
                        value: system_metrics.error_rate_percent,
                        unit: "percent".to_string(),
                        labels: std::collections::HashMap::new(),
                        timestamp: now,
                        statistics: None,
                    },
                ];

                drop(system_metrics);

                let event = MetricsEvent {
                    metrics,
                    timestamp: now,
                };

                if tx.send(Ok(event)).await.is_err() {
                    info!("StreamMetrics: client disconnected");
                    break;
                }
            }
        });

        Ok(Response::new(Box::pin(
            tokio_stream::wrappers::ReceiverStream::new(rx),
        )))
    }

    type StreamAlertsStream =
        std::pin::Pin<Box<dyn tokio_stream::Stream<Item = Result<AlertEvent, Status>> + Send>>;

    #[instrument(skip_all)]
    async fn stream_alerts(
        &self,
        request: Request<StreamAlertsRequest>,
    ) -> Result<Response<Self::StreamAlertsStream>, Status> {
        let req = request.into_inner();
        let min_severity = req.min_severity.unwrap_or(0);

        info!("StreamAlerts started: min_severity={}", min_severity);

        let (tx, rx) = tokio::sync::mpsc::channel(128);
        let state = self.state.clone();
        let sys = Arc::clone(&self.sys);
        let disks = Arc::clone(&self.disks);
        let alert_store = Arc::clone(&self.alerts);
        let mut alert_counter: u64 = 0;

        tokio::spawn(async move {
            let mut interval =
                tokio::time::interval(std::time::Duration::from_secs(10));

            loop {
                interval.tick().await;

                let now = chrono::Utc::now().timestamp();
                let mut alerts: Vec<AlertEvent> = Vec::new();

                // Check health status for degradation alerts
                let health_status = state.get_health_status().await;
                match health_status {
                    HealthStatus::Critical => {
                        alert_counter += 1;
                        alerts.push(AlertEvent {
                            alert: Some(Alert {
                                alert_id: format!("mon-alert-{}", alert_counter),
                                alert_type: AlertType::HealthCheck as i32,
                                severity: AlertSeverity::Emergency as i32,
                                title: "System health CRITICAL".to_string(),
                                description: "Trading system health is in critical state"
                                    .to_string(),
                                service_name: "trading_service".to_string(),
                                labels: std::collections::HashMap::new(),
                                triggered_at: now,
                                acknowledged_at: None,
                                acknowledged_by: None,
                                resolved_at: None,
                                status: AlertStatus::Active as i32,
                                resolution_note: None,
                            }),
                            event_type: AlertEventType::Triggered as i32,
                            timestamp: now,
                        });
                    },
                    HealthStatus::Unhealthy => {
                        alert_counter += 1;
                        alerts.push(AlertEvent {
                            alert: Some(Alert {
                                alert_id: format!("mon-alert-{}", alert_counter),
                                alert_type: AlertType::HealthCheck as i32,
                                severity: AlertSeverity::Critical as i32,
                                title: "System health UNHEALTHY".to_string(),
                                description: "Trading system health is unhealthy".to_string(),
                                service_name: "trading_service".to_string(),
                                labels: std::collections::HashMap::new(),
                                triggered_at: now,
                                acknowledged_at: None,
                                acknowledged_by: None,
                                resolved_at: None,
                                status: AlertStatus::Active as i32,
                                resolution_note: None,
                            }),
                            event_type: AlertEventType::Triggered as i32,
                            timestamp: now,
                        });
                    },
                    HealthStatus::Degraded => {
                        alert_counter += 1;
                        alerts.push(AlertEvent {
                            alert: Some(Alert {
                                alert_id: format!("mon-alert-{}", alert_counter),
                                alert_type: AlertType::HealthCheck as i32,
                                severity: AlertSeverity::Warning as i32,
                                title: "System health DEGRADED".to_string(),
                                description: "Trading system health is degraded".to_string(),
                                service_name: "trading_service".to_string(),
                                labels: std::collections::HashMap::new(),
                                triggered_at: now,
                                acknowledged_at: None,
                                acknowledged_by: None,
                                resolved_at: None,
                                status: AlertStatus::Active as i32,
                                resolution_note: None,
                            }),
                            event_type: AlertEventType::Triggered as i32,
                            timestamp: now,
                        });
                    },
                    HealthStatus::Healthy => {
                        // No alert for healthy state
                    },
                }

                // Check resource usage thresholds
                let live = collect_system_metrics(&sys, &disks).await;

                if live.cpu_usage_percent > 90.0 {
                    alert_counter += 1;
                    let mut labels = std::collections::HashMap::new();
                    labels.insert(
                        "cpu_usage".to_string(),
                        format!("{:.1}%", live.cpu_usage_percent),
                    );
                    alerts.push(AlertEvent {
                        alert: Some(Alert {
                            alert_id: format!("mon-alert-{}", alert_counter),
                            alert_type: AlertType::ResourceUsage as i32,
                            severity: AlertSeverity::Warning as i32,
                            title: "High CPU usage".to_string(),
                            description: format!(
                                "CPU usage at {:.1}% exceeds 90% threshold",
                                live.cpu_usage_percent
                            ),
                            service_name: "trading_service".to_string(),
                            labels,
                            triggered_at: now,
                            acknowledged_at: None,
                            acknowledged_by: None,
                            resolved_at: None,
                            status: AlertStatus::Active as i32,
                            resolution_note: None,
                        }),
                        event_type: AlertEventType::Triggered as i32,
                        timestamp: now,
                    });
                }

                if live.memory_usage_percent > 90.0 {
                    alert_counter += 1;
                    let mut labels = std::collections::HashMap::new();
                    labels.insert(
                        "memory_usage".to_string(),
                        format!("{:.1}%", live.memory_usage_percent),
                    );
                    alerts.push(AlertEvent {
                        alert: Some(Alert {
                            alert_id: format!("mon-alert-{}", alert_counter),
                            alert_type: AlertType::ResourceUsage as i32,
                            severity: AlertSeverity::Critical as i32,
                            title: "High memory usage".to_string(),
                            description: format!(
                                "Memory usage at {:.1}% exceeds 90% threshold",
                                live.memory_usage_percent
                            ),
                            service_name: "trading_service".to_string(),
                            labels,
                            triggered_at: now,
                            acknowledged_at: None,
                            acknowledged_by: None,
                            resolved_at: None,
                            status: AlertStatus::Active as i32,
                            resolution_note: None,
                        }),
                        event_type: AlertEventType::Triggered as i32,
                        timestamp: now,
                    });
                }

                if live.disk_usage_percent > 95.0 {
                    alert_counter += 1;
                    let mut labels = std::collections::HashMap::new();
                    labels.insert(
                        "disk_usage".to_string(),
                        format!("{:.1}%", live.disk_usage_percent),
                    );
                    alerts.push(AlertEvent {
                        alert: Some(Alert {
                            alert_id: format!("mon-alert-{}", alert_counter),
                            alert_type: AlertType::ResourceUsage as i32,
                            severity: AlertSeverity::Critical as i32,
                            title: "Disk space critically low".to_string(),
                            description: format!(
                                "Disk usage at {:.1}% exceeds 95% threshold",
                                live.disk_usage_percent
                            ),
                            service_name: "trading_service".to_string(),
                            labels,
                            triggered_at: now,
                            acknowledged_at: None,
                            acknowledged_by: None,
                            resolved_at: None,
                            status: AlertStatus::Active as i32,
                            resolution_note: None,
                        }),
                        event_type: AlertEventType::Triggered as i32,
                        timestamp: now,
                    });
                }

                // Persist generated alerts to the shared store, capped at 1000 entries.
                {
                    let mut store = alert_store.write().await;
                    for alert_event in &alerts {
                        if let Some(alert) = &alert_event.alert {
                            store.push(alert.clone());
                        }
                    }
                    const MAX_STORE: usize = 1000;
                    if store.len() > MAX_STORE {
                        let drain_count = store.len() - MAX_STORE;
                        store.drain(..drain_count);
                    }
                }

                // Filter by minimum severity and send
                for alert_event in alerts {
                    let severity = alert_event
                        .alert
                        .as_ref()
                        .map_or(0, |a| a.severity);
                    if severity >= min_severity
                        && tx.send(Ok(alert_event)).await.is_err() {
                            info!("StreamAlerts: client disconnected");
                            return;
                        }
                }
            }
        });

        Ok(Response::new(Box::pin(
            tokio_stream::wrappers::ReceiverStream::new(rx),
        )))
    }

    // ========================================================================
    // Training metrics RPCs (not applicable to trading_service; served by
    // api service which queries Prometheus directly). Return UNIMPLEMENTED.
    // ========================================================================

    type StreamTrainingMetricsStream = std::pin::Pin<
        Box<dyn tokio_stream::Stream<Item = Result<GetLiveTrainingMetricsResponse, Status>> + Send>,
    >;

    async fn get_live_training_metrics(
        &self,
        _request: Request<GetLiveTrainingMetricsRequest>,
    ) -> Result<Response<GetLiveTrainingMetricsResponse>, Status> {
        Err(Status::unimplemented(
            "Training metrics are served by api, not trading_service",
        ))
    }

    async fn stream_training_metrics(
        &self,
        _request: Request<StreamTrainingMetricsRequest>,
    ) -> Result<Response<Self::StreamTrainingMetricsStream>, Status> {
        Err(Status::unimplemented(
            "Training metrics are served by api, not trading_service",
        ))
    }

    async fn get_epoch_history(
        &self,
        _request: Request<GetEpochHistoryRequest>,
    ) -> Result<Response<GetEpochHistoryResponse>, Status> {
        Err(Status::unimplemented(
            "Epoch history is served by api, not trading_service",
        ))
    }

    type SubscribeClusterPodsStream = std::pin::Pin<
        Box<dyn tokio_stream::Stream<Item = Result<ClusterPodsResponse, Status>> + Send>,
    >;

    async fn subscribe_cluster_pods(
        &self,
        _request: Request<SubscribeClusterPodsRequest>,
    ) -> Result<Response<Self::SubscribeClusterPodsStream>, Status> {
        Err(Status::unimplemented(
            "Cluster pods are served by api, not trading_service",
        ))
    }
}
