mecha10-core 0.6.5

//! Performance Profiling Helpers
//!
//! Provides tools for measuring and analyzing system performance:
//! - Message latency tracking
//! - Throughput measurement
//! - CPU and memory profiling
//! - Node execution timing
//! - Statistical analysis
//!
//! # Features
//! - Zero-copy metrics collection
//! - Low overhead (< 1% CPU)
//! - Statistical aggregation (min, max, mean, p50, p95, p99)
//! - Time-series data export
//! - Real-time monitoring

use crate::prelude::*;
use crate::{Message, Receiver};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use std::collections::VecDeque;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;

/// Generates a `PerformanceMetrics` method that write-locks `inner` and forwards
/// `(topic, <arg>)` to the named tracker's `record`. `record_latency`/`record_message` used to
/// be hand-written copies of this shape differing only in the tracker field and the argument's
/// name/type; cargo-dupes flagged the two fn bodies as an exact duplicate.
macro_rules! record_forwarder {
    ($(#[$meta:meta])* $name:ident, $tracker:ident, $arg:ident: $ty:ty) => {
        $(#[$meta])*
        pub async fn $name(&self, topic: &str, $arg: $ty) {
            let mut inner = self.inner.write().await;
            inner.$tracker.record(topic, $arg);
        }
    };
}

/// Generates a `PerformanceMetrics` method that read-locks `inner` and forwards to the named
/// tracker's `get_stats`. `get_latency_stats`/`get_throughput_stats`/`get_execution_stats` used
/// to be hand-written copies of this shape differing only in the tracker field and the returned
/// stats type; cargo-dupes flagged the three fn bodies as an exact duplicate.
macro_rules! stats_getter {
    ($(#[$meta:meta])* $name:ident, $tracker:ident, $stats:ty) => {
        $(#[$meta])*
        pub async fn $name(&self, key: &str) -> Option<$stats> {
            let inner = self.inner.read().await;
            inner.$tracker.get_stats(key)
        }
    };
}

/// Performance metrics collector
#[derive(Clone)]
pub struct PerformanceMetrics {
    inner: Arc<RwLock<MetricsInner>>,
}

struct MetricsInner {
    latency_tracker: LatencyTracker,
    throughput_tracker: ThroughputTracker,
    execution_tracker: ExecutionTracker,
    memory_tracker: MemoryTracker,
}

impl PerformanceMetrics {
    /// Create a new performance metrics collector
    pub fn new() -> Self {
        Self {
            inner: Arc::new(RwLock::new(MetricsInner {
                latency_tracker: LatencyTracker::new(),
                throughput_tracker: ThroughputTracker::new(),
                execution_tracker: ExecutionTracker::new(),
                memory_tracker: MemoryTracker::new(),
            })),
        }
    }

    record_forwarder!(
        /// Record message latency
        record_latency, latency_tracker, latency_us: u64
    );
    record_forwarder!(
        /// Record message throughput
        record_message, throughput_tracker, size_bytes: usize
    );

    /// Start timing a code block
    pub async fn start_timer(&self, name: &str) -> TimerGuard {
        TimerGuard {
            name: name.to_string(),
            start: Instant::now(),
            metrics: self.clone(),
        }
    }

    /// Record execution time
    pub async fn record_execution(&self, name: &str, duration: Duration) {
        let mut inner = self.inner.write().await;
        inner.execution_tracker.record(name, duration.as_micros() as u64);
    }

    // duplicate: false positive — matches `HealthChecker::register` (packages/runtime/src/health.rs)
    // on AST shape only; unrelated domains, nothing to share.
    /// Record memory usage
    pub async fn record_memory(&self, allocated_bytes: usize, freed_bytes: usize) {
        let mut inner = self.inner.write().await;
        inner.memory_tracker.record(allocated_bytes, freed_bytes);
    }

    stats_getter!(
        /// Get latency statistics for a topic
        get_latency_stats, latency_tracker, LatencyStats
    );
    stats_getter!(
        /// Get throughput statistics for a topic
        get_throughput_stats, throughput_tracker, ThroughputStats
    );
    stats_getter!(
        /// Get execution statistics for a named block
        get_execution_stats, execution_tracker, ExecutionStats
    );

    /// Get memory statistics
    pub async fn get_memory_stats(&self) -> MemoryStats {
        let inner = self.inner.read().await;
        inner.memory_tracker.get_stats()
    }

    /// Get all metrics as a report
    pub async fn get_report(&self) -> PerformanceReport {
        let inner = self.inner.read().await;

        PerformanceReport {
            latency: inner.latency_tracker.get_all_stats(),
            throughput: inner.throughput_tracker.get_all_stats(),
            execution: inner.execution_tracker.get_all_stats(),
            memory: inner.memory_tracker.get_stats(),
            timestamp: now_micros(),
        }
    }

    /// Reset all metrics
    pub async fn reset(&self) {
        let mut inner = self.inner.write().await;
        inner.latency_tracker = LatencyTracker::new();
        inner.throughput_tracker = ThroughputTracker::new();
        inner.execution_tracker = ExecutionTracker::new();
        inner.memory_tracker = MemoryTracker::new();
    }
}

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

/// Timer guard for automatic timing
pub struct TimerGuard {
    name: String,
    start: Instant,
    metrics: PerformanceMetrics,
}

impl Drop for TimerGuard {
    fn drop(&mut self) {
        let duration = self.start.elapsed();
        let metrics = self.metrics.clone();
        let name = self.name.clone();

        tokio::spawn(async move {
            metrics.record_execution(&name, duration).await;
        });
    }
}

/// Count/min/max/mean/percentile fields shared by [`LatencyStats`] and [`ExecutionStats`] —
/// both are computed the same way from a [`SampleBuffer`], differing only in the key field
/// name (`topic` vs `name`) on the public struct each tracker emits.
struct SampleStats {
    count: u64,
    min_us: u64,
    max_us: u64,
    mean_us: f64,
    p50_us: u64,
    p95_us: u64,
    p99_us: u64,
}

impl SampleStats {
    fn from_buffer(buffer: &SampleBuffer) -> Self {
        Self {
            count: buffer.count(),
            min_us: buffer.min(),
            max_us: buffer.max(),
            mean_us: buffer.mean(),
            p50_us: buffer.percentile(0.50),
            p95_us: buffer.percentile(0.95),
            p99_us: buffer.percentile(0.99),
        }
    }
}

/// Tracker keyed by an arbitrary string label (topic name, timed-block name, ...) that
/// buffers raw samples and hands back [`SampleStats`] per key. Backs both [`LatencyTracker`]
/// and [`ExecutionTracker`], which only differ in the public stats struct they wrap the
/// result in.
struct KeyedSampleTracker {
    samples: std::collections::HashMap<String, SampleBuffer>,
}

impl KeyedSampleTracker {
    fn new() -> Self {
        Self {
            samples: std::collections::HashMap::new(),
        }
    }

    fn record(&mut self, key: &str, value_us: u64) {
        self.samples
            .entry(key.to_string())
            .or_insert_with(SampleBuffer::new)
            .add(value_us);
    }

    fn stats_for(&self, key: &str) -> Option<SampleStats> {
        self.samples.get(key).map(SampleStats::from_buffer)
    }

    fn keys(&self) -> impl Iterator<Item = &String> {
        self.samples.keys()
    }
}

/// Generates a `KeyedSampleTracker`-backed tracker type (`new`/`record`/`get_stats`/
/// `get_all_stats`) that reports into `$stats`, keyed by the field named `$field`.
/// [`LatencyTracker`] and [`ExecutionTracker`] are identical modulo that stats type and field
/// name — cargo-dupes flagged their hand-written `record`/`get_stats`/`get_all_stats` bodies as
/// exact duplicates of each other, so both are generated from here instead.
macro_rules! keyed_sample_tracker {
    ($(#[$meta:meta])* $tracker:ident, $stats:ident, $field:ident) => {
        $(#[$meta])*
        struct $tracker(KeyedSampleTracker);

        impl $tracker {
            fn new() -> Self {
                Self(KeyedSampleTracker::new())
            }

            fn record(&mut self, key: &str, value_us: u64) {
                self.0.record(key, value_us);
            }

            fn get_stats(&self, key: &str) -> Option<$stats> {
                self.0.stats_for(key).map(|s| $stats {
                    $field: key.to_string(),
                    count: s.count,
                    min_us: s.min_us,
                    max_us: s.max_us,
                    mean_us: s.mean_us,
                    p50_us: s.p50_us,
                    p95_us: s.p95_us,
                    p99_us: s.p99_us,
                })
            }

            fn get_all_stats(&self) -> Vec<$stats> {
                self.0.keys().filter_map(|key| self.get_stats(key)).collect()
            }
        }
    };
}

keyed_sample_tracker!(
    /// Latency tracker
    LatencyTracker, LatencyStats, topic
);

/// Throughput tracker
struct ThroughputTracker {
    data: std::collections::HashMap<String, ThroughputData>,
}

#[derive(Debug, Clone)]
struct ThroughputData {
    message_count: u64,
    byte_count: u64,
    start_time: Instant,
}

impl ThroughputTracker {
    fn new() -> Self {
        Self {
            data: std::collections::HashMap::new(),
        }
    }

    fn record(&mut self, topic: &str, size_bytes: usize) {
        let entry = self.data.entry(topic.to_string()).or_insert(ThroughputData {
            message_count: 0,
            byte_count: 0,
            start_time: Instant::now(),
        });

        entry.message_count += 1;
        entry.byte_count += size_bytes as u64;
    }

    fn get_stats(&self, topic: &str) -> Option<ThroughputStats> {
        self.data.get(topic).map(|data| {
            let elapsed = data.start_time.elapsed().as_secs_f64();
            let messages_per_sec = if elapsed > 0.0 {
                data.message_count as f64 / elapsed
            } else {
                0.0
            };
            let bytes_per_sec = if elapsed > 0.0 {
                data.byte_count as f64 / elapsed
            } else {
                0.0
            };

            ThroughputStats {
                topic: topic.to_string(),
                message_count: data.message_count,
                byte_count: data.byte_count,
                duration_sec: elapsed,
                messages_per_sec,
                bytes_per_sec,
                megabytes_per_sec: bytes_per_sec / 1_000_000.0,
            }
        })
    }

    // duplicate: false positive — matches `Supervisor::launch_nodes`
    // (packages/runtime/src/supervisor.rs); see the comment there.
    fn get_all_stats(&self) -> Vec<ThroughputStats> {
        self.data.keys().filter_map(|topic| self.get_stats(topic)).collect()
    }
}

keyed_sample_tracker!(
    /// Execution time tracker
    ExecutionTracker, ExecutionStats, name
);

/// Memory usage tracker
struct MemoryTracker {
    total_allocated: usize,
    total_freed: usize,
    current_usage: usize,
    peak_usage: usize,
}

impl MemoryTracker {
    fn new() -> Self {
        Self {
            total_allocated: 0,
            total_freed: 0,
            current_usage: 0,
            peak_usage: 0,
        }
    }

    fn record(&mut self, allocated: usize, freed: usize) {
        self.total_allocated += allocated;
        self.total_freed += freed;
        self.current_usage = self.current_usage.saturating_add(allocated).saturating_sub(freed);

        if self.current_usage > self.peak_usage {
            self.peak_usage = self.current_usage;
        }
    }

    fn get_stats(&self) -> MemoryStats {
        MemoryStats {
            total_allocated_bytes: self.total_allocated,
            total_freed_bytes: self.total_freed,
            current_usage_bytes: self.current_usage,
            peak_usage_bytes: self.peak_usage,
            current_usage_mb: self.current_usage as f64 / 1_000_000.0,
            peak_usage_mb: self.peak_usage as f64 / 1_000_000.0,
        }
    }
}

/// Sample buffer for statistical calculations
struct SampleBuffer {
    samples: VecDeque<u64>,
    max_samples: usize,
}

impl SampleBuffer {
    fn new() -> Self {
        Self {
            samples: VecDeque::new(),
            max_samples: 10000,
        }
    }

    fn add(&mut self, value: u64) {
        if self.samples.len() >= self.max_samples {
            self.samples.pop_front();
        }
        self.samples.push_back(value);
    }

    fn count(&self) -> u64 {
        self.samples.len() as u64
    }

    // duplicate: false positive — `min`/`max` are opposite `Iterator` reductions over the same
    // samples; the AST shape coincidentally matches but there's no shared logic to extract
    // (the whole body is the one differing token, `.min()` vs `.max()`).
    fn min(&self) -> u64 {
        self.samples.iter().min().copied().unwrap_or(0)
    }

    fn max(&self) -> u64 {
        self.samples.iter().max().copied().unwrap_or(0)
    }

    fn mean(&self) -> f64 {
        if self.samples.is_empty() {
            return 0.0;
        }
        let sum: u64 = self.samples.iter().sum();
        sum as f64 / self.samples.len() as f64
    }

    fn percentile(&self, p: f64) -> u64 {
        if self.samples.is_empty() {
            return 0;
        }

        let mut sorted: Vec<u64> = self.samples.iter().copied().collect();
        sorted.sort_unstable();

        let idx = ((sorted.len() as f64 - 1.0) * p) as usize;
        sorted[idx]
    }
}

/// Latency statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LatencyStats {
    pub topic: String,
    pub count: u64,
    pub min_us: u64,
    pub max_us: u64,
    pub mean_us: f64,
    pub p50_us: u64,
    pub p95_us: u64,
    pub p99_us: u64,
}

impl Message for LatencyStats {}

/// Throughput statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ThroughputStats {
    pub topic: String,
    pub message_count: u64,
    pub byte_count: u64,
    pub duration_sec: f64,
    pub messages_per_sec: f64,
    pub bytes_per_sec: f64,
    pub megabytes_per_sec: f64,
}

impl Message for ThroughputStats {}

/// Execution statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutionStats {
    pub name: String,
    pub count: u64,
    pub min_us: u64,
    pub max_us: u64,
    pub mean_us: f64,
    pub p50_us: u64,
    pub p95_us: u64,
    pub p99_us: u64,
}

impl Message for ExecutionStats {}

/// Memory statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MemoryStats {
    pub total_allocated_bytes: usize,
    pub total_freed_bytes: usize,
    pub current_usage_bytes: usize,
    pub peak_usage_bytes: usize,
    pub current_usage_mb: f64,
    pub peak_usage_mb: f64,
}

impl Message for MemoryStats {}

/// Complete performance report
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PerformanceReport {
    pub latency: Vec<LatencyStats>,
    pub throughput: Vec<ThroughputStats>,
    pub execution: Vec<ExecutionStats>,
    pub memory: MemoryStats,
    pub timestamp: u64,
}

impl Message for PerformanceReport {}

/// Helper to measure message latency
pub fn measure_latency(timestamp_us: u64) -> u64 {
    let now = now_micros();
    now.saturating_sub(timestamp_us)
}

/// Extension trait for Context to add profiling
pub trait ProfilingExt {
    /// Get performance metrics
    fn metrics(&self) -> &PerformanceMetrics;

    /// Publish with latency tracking
    #[allow(async_fn_in_trait)]
    async fn publish_profiled<T: Message + Serialize>(&self, topic: &str, message: &T) -> Result<()>;

    /// Subscribe with latency tracking
    #[allow(async_fn_in_trait)]
    async fn subscribe_profiled<T: Message + DeserializeOwned + Send + 'static>(
        &self,
        topic: &str,
    ) -> Result<Receiver<T>>;
}