zc2 0.0.29

//! Statistics tracking for the broker's `/stats` endpoint.

use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::RwLock;
use std::time::Instant;

use chrono::{DateTime, Local};
use serde::{Deserialize, Serialize};

/// Maximum number of transactions to keep in history
const MAX_TRANSACTIONS: usize = 100;

/// Maximum number of data points for time-series metrics
const MAX_DATAPOINTS: usize = 60;

/// A recorded task offer (broadcast to peers) for the task pile view
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskOfferRecord {
    /// Task/request ID
    pub task_id: String,
    /// When the offer was broadcast
    pub timestamp: DateTime<Local>,
    /// User being charged
    pub requester_user_id: String,
    /// Max price (zkcr/hr) offered
    pub max_price_per_hour: f64,
    /// Originating broker node name
    pub source_broker: String,
}

/// A recorded transaction for display
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TransactionRecord {
    /// Transaction number
    pub tx_num: u64,
    /// Timestamp
    pub timestamp: DateTime<Local>,
    /// User ID
    pub user_id: String,
    /// Action type (EXECUTE, CREDIT, etc.)
    pub action: String,
    /// Cost in credits
    pub cost: f64,
    /// Balance after transaction
    pub balance: f64,
    /// Target worker (if applicable)
    pub worker: Option<String>,
    /// Duration in milliseconds
    pub duration_ms: f64,
    /// Status (OK, FAIL, PENDING)
    pub status: TransactionStatus,
    /// Agreed price per hour (zkcr/hr) from the worker that executed
    #[serde(default)]
    pub price_per_hour: Option<f64>,
    /// Owner ID of the worker / executor (broker owner for P2P)
    #[serde(default)]
    pub owner_id: Option<String>,
    /// Worker process ID (from X-Zakuro-Pid if present)
    #[serde(default)]
    pub worker_pid: Option<String>,
    /// Worker IP (from X-Zakuro-IP or worker URI)
    #[serde(default)]
    pub worker_ip: Option<String>,
    /// Request/task ID (links to task_offers for execution trace)
    #[serde(default)]
    pub request_id: Option<String>,
}

#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
pub enum TransactionStatus {
    Ok,
    Fail,
    Pending,
}

impl TransactionStatus {
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Ok => "OK",
            Self::Fail => "FAIL",
            Self::Pending => "PENDING",
        }
    }
}

/// Time-series data point
#[derive(Debug, Clone, Copy)]
pub struct DataPoint {
    pub timestamp: Instant,
    pub value: f64,
}

/// Aggregated broker statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BrokerMetrics {
    /// Total requests processed
    pub total_requests: u64,
    /// Successful requests
    pub successful_requests: u64,
    /// Failed requests
    pub failed_requests: u64,
    /// Total credits spent
    pub total_credits_spent: f64,
    /// Requests per second (rolling average)
    pub requests_per_sec: f64,
    /// Average latency (ms)
    pub avg_latency_ms: f64,
    /// P95 latency (ms)
    pub p95_latency_ms: f64,
    /// Active workers
    pub active_workers: usize,
    /// Total workers
    pub total_workers: usize,
    /// Uptime in seconds
    pub uptime_secs: u64,
    /// Local mode (free execution)
    pub local_mode: bool,
    /// Ledger connected
    pub ledger_connected: bool,
    /// This node's WireGuard IP (if connected)
    pub wireguard_ip: Option<String>,
    /// Whether WireGuard is connected
    pub wireguard_connected: bool,
}

impl Default for BrokerMetrics {
    fn default() -> Self {
        Self {
            total_requests: 0,
            successful_requests: 0,
            failed_requests: 0,
            total_credits_spent: 0.0,
            requests_per_sec: 0.0,
            avg_latency_ms: 0.0,
            p95_latency_ms: 0.0,
            active_workers: 0,
            total_workers: 0,
            uptime_secs: 0,
            local_mode: false,
            ledger_connected: false,
            wireguard_ip: None,
            wireguard_connected: false,
        }
    }
}

/// Max task offers to keep in the pile
const MAX_TASK_OFFERS: usize = 50;

/// Thread-safe statistics collector
pub struct StatsCollector {
    /// Start time
    start_time: Instant,
    /// Recent transactions (ring buffer)
    transactions: RwLock<VecDeque<TransactionRecord>>,
    /// Recent task offers (broadcast pile)
    task_offers: RwLock<VecDeque<TaskOfferRecord>>,
    /// Latency measurements (for percentiles)
    latencies: RwLock<VecDeque<f64>>,
    /// Requests per second history
    rps_history: RwLock<VecDeque<DataPoint>>,
    /// Credits spent history
    credits_history: RwLock<VecDeque<DataPoint>>,
    /// Counters
    total_requests: AtomicU64,
    successful_requests: AtomicU64,
    failed_requests: AtomicU64,
    /// Total credits spent (stored as u64 with 6 decimal precision)
    total_credits_spent: AtomicU64,
    /// Request counter for RPS calculation
    requests_this_second: AtomicU64,
    /// Last RPS calculation time
    last_rps_calc: RwLock<Instant>,
}

impl StatsCollector {
    /// Create a new stats collector
    pub fn new() -> Self {
        Self {
            start_time: Instant::now(),
            transactions: RwLock::new(VecDeque::with_capacity(MAX_TRANSACTIONS)),
            task_offers: RwLock::new(VecDeque::with_capacity(MAX_TASK_OFFERS)),
            latencies: RwLock::new(VecDeque::with_capacity(MAX_DATAPOINTS * 10)),
            rps_history: RwLock::new(VecDeque::with_capacity(MAX_DATAPOINTS)),
            credits_history: RwLock::new(VecDeque::with_capacity(MAX_DATAPOINTS)),
            total_requests: AtomicU64::new(0),
            successful_requests: AtomicU64::new(0),
            failed_requests: AtomicU64::new(0),
            total_credits_spent: AtomicU64::new(0),
            requests_this_second: AtomicU64::new(0),
            last_rps_calc: RwLock::new(Instant::now()),
        }
    }

    /// Record a transaction
    pub fn record_transaction(&self, record: TransactionRecord) {
        // Update counters
        self.total_requests.fetch_add(1, Ordering::Relaxed);
        self.requests_this_second.fetch_add(1, Ordering::Relaxed);

        match record.status {
            TransactionStatus::Ok => {
                self.successful_requests.fetch_add(1, Ordering::Relaxed);
            }
            TransactionStatus::Fail => {
                self.failed_requests.fetch_add(1, Ordering::Relaxed);
            }
            TransactionStatus::Pending => {}
        }

        // Display-only series below use `try_write`: the hot path must never
        // block on these locks. The exact metrics (counts, credits) are the
        // atomics above; the latency/transaction/credits *history* feeds the
        // `/stats` only, so under contention dropping a sample is
        // correct — it keeps the request path lock-free at high RPS (the
        // blocking `write()` here was a primary throughput bottleneck).

        // Record latency (sampled under contention)
        if record.duration_ms > 0.0 {
            if let Ok(mut latencies) = self.latencies.try_write() {
                latencies.push_back(record.duration_ms);
                while latencies.len() > MAX_DATAPOINTS * 10 {
                    latencies.pop_front();
                }
            }
        }

        // Record credits spent
        if record.cost > 0.0 {
            let cost_micro = (record.cost * 1_000_000.0) as u64;
            self.total_credits_spent
                .fetch_add(cost_micro, Ordering::Relaxed);

            if let Ok(mut history) = self.credits_history.try_write() {
                history.push_back(DataPoint {
                    timestamp: Instant::now(),
                    value: record.cost,
                });
                while history.len() > MAX_DATAPOINTS {
                    history.pop_front();
                }
            }
        }

        // Add to transaction history (sampled under contention)
        if let Ok(mut txs) = self.transactions.try_write() {
            txs.push_back(record);
            while txs.len() > MAX_TRANSACTIONS {
                txs.pop_front();
            }
        }
    }

    /// Record a task offer (broadcast to peers) for the task pile view
    pub fn record_task_offer(&self, record: TaskOfferRecord) {
        if let Ok(mut offers) = self.task_offers.try_write() {
            offers.push_back(record);
            while offers.len() > MAX_TASK_OFFERS {
                offers.pop_front();
            }
        }
    }

    /// Recent task offers (newest last)
    pub fn recent_task_offers(&self, limit: usize) -> Vec<TaskOfferRecord> {
        if let Ok(offers) = self.task_offers.read() {
            offers.iter().rev().take(limit).cloned().collect()
        } else {
            Vec::new()
        }
    }

    /// Calculate and record requests per second
    pub fn tick_rps(&self) {
        let now = Instant::now();
        let elapsed = {
            let last = self.last_rps_calc.read().unwrap();
            now.duration_since(*last).as_secs_f64()
        };

        if elapsed >= 1.0 {
            let requests = self.requests_this_second.swap(0, Ordering::Relaxed);
            let rps = requests as f64 / elapsed;

            if let Ok(mut history) = self.rps_history.write() {
                history.push_back(DataPoint {
                    timestamp: now,
                    value: rps,
                });
                while history.len() > MAX_DATAPOINTS {
                    history.pop_front();
                }
            }

            *self.last_rps_calc.write().unwrap() = now;
        }
    }

    /// Get recent transactions
    pub fn recent_transactions(&self, limit: usize) -> Vec<TransactionRecord> {
        if let Ok(txs) = self.transactions.read() {
            txs.iter().rev().take(limit).cloned().collect()
        } else {
            Vec::new()
        }
    }

    /// Get RPS history for sparkline
    pub fn rps_history(&self) -> Vec<f64> {
        if let Ok(history) = self.rps_history.read() {
            history.iter().map(|dp| dp.value).collect()
        } else {
            Vec::new()
        }
    }

    /// Get current metrics
    pub fn metrics(
        &self,
        active_workers: usize,
        total_workers: usize,
        local_mode: bool,
        ledger_connected: bool,
        wireguard_ip: Option<String>,
    ) -> BrokerMetrics {
        let total = self.total_requests.load(Ordering::Relaxed);
        let successful = self.successful_requests.load(Ordering::Relaxed);
        let failed = self.failed_requests.load(Ordering::Relaxed);
        let credits_micro = self.total_credits_spent.load(Ordering::Relaxed);

        // Calculate average latency
        let (avg_latency, p95_latency) = if let Ok(latencies) = self.latencies.read() {
            if latencies.is_empty() {
                (0.0, 0.0)
            } else {
                let sum: f64 = latencies.iter().sum();
                let avg = sum / latencies.len() as f64;

                // P95 calculation
                let mut sorted: Vec<f64> = latencies.iter().cloned().collect();
                sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
                let p95_idx = (sorted.len() as f64 * 0.95) as usize;
                let p95 = sorted
                    .get(p95_idx.min(sorted.len() - 1))
                    .cloned()
                    .unwrap_or(0.0);

                (avg, p95)
            }
        } else {
            (0.0, 0.0)
        };

        // Calculate RPS from history
        let rps = if let Ok(history) = self.rps_history.read() {
            if history.is_empty() {
                0.0
            } else {
                history.iter().rev().take(5).map(|dp| dp.value).sum::<f64>()
                    / 5.0f64.min(history.len() as f64)
            }
        } else {
            0.0
        };

        BrokerMetrics {
            total_requests: total,
            successful_requests: successful,
            failed_requests: failed,
            total_credits_spent: credits_micro as f64 / 1_000_000.0,
            requests_per_sec: rps,
            avg_latency_ms: avg_latency,
            p95_latency_ms: p95_latency,
            active_workers,
            total_workers,
            uptime_secs: self.start_time.elapsed().as_secs(),
            local_mode,
            ledger_connected,
            wireguard_connected: wireguard_ip.is_some(),
            wireguard_ip,
        }
    }
}

impl Default for StatsCollector {
    fn default() -> Self {
        Self::new()
    }
}

/// Format duration for display
pub fn format_uptime(secs: u64) -> String {
    if secs < 60 {
        format!("{}s", secs)
    } else if secs < 3600 {
        format!("{}m {}s", secs / 60, secs % 60)
    } else if secs < 86400 {
        format!("{}h {}m", secs / 3600, (secs % 3600) / 60)
    } else {
        format!("{}d {}h", secs / 86400, (secs % 86400) / 3600)
    }
}

/// Format credits for display
pub fn format_credits(amount: f64) -> String {
    if amount >= 0.01 {
        format!("{:.4}", amount)
    } else {
        format!("{:.6}", amount)
    }
}

/// Worker info for stats response
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkerStats {
    pub id: String,
    pub name: String,
    pub uri: String,
    pub status: String,
    pub cpus_available: f64,
    pub memory_available_gib: f64,
    pub gpus_available: u32,
    pub price_per_hour: f64,
    pub active_requests: u32,
    pub avg_latency_ms: f64,
}

/// Complete `/stats` response
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StatsResponse {
    /// Host and port of the broker
    pub host: String,
    pub port: u16,
    /// Recent transactions
    pub transactions: Vec<TransactionRecord>,
    /// Recent task offers (broadcast pile)
    #[serde(default)]
    pub task_offers: Vec<TaskOfferRecord>,
    /// Worker list
    pub workers: Vec<WorkerStats>,
    /// Aggregated metrics
    pub metrics: BrokerMetrics,
    /// RPS history for sparkline
    pub rps_history: Vec<f64>,
    /// This node's WireGuard IP (if connected)
    pub wireguard_ip: Option<String>,
    /// Whether WireGuard is connected
    pub wireguard_connected: bool,
}