//! Lock-free sliding window counters for Tier 0 metrics.
//!
//! Uses `AtomicU64` for concurrent, lock-free recording of job processing
//! statistics. Designed to be called from `TrackedDelivery` on every ack/nack
//! with minimal overhead.

use std::sync::atomic::{AtomicU64, Ordering};

use crate::metrics::tier0::Tier0Snapshot;

/// Thread-safe, lock-free counters for a single queue's processing metrics.
///
/// Shared between [`TrackedDelivery`](crate::TrackedDelivery) (which records
/// events) and [`RabbitMqMetricsCollector`](super::RabbitMqMetricsCollector)
/// (which snapshots and resets on each collection interval).
///
/// All operations use `Relaxed` ordering — we accept minor imprecision in
/// exchange for zero contention on hot paths.
pub struct SlidingWindowCounters {
    processed_count: AtomicU64,
    success_count: AtomicU64,
    failure_count: AtomicU64,
    total_duration_ms: AtomicU64,
}

impl SlidingWindowCounters {
    /// Create a new counter set, all zeroed.
    pub fn new() -> Self {
        Self {
            processed_count: AtomicU64::new(0),
            success_count: AtomicU64::new(0),
            failure_count: AtomicU64::new(0),
            total_duration_ms: AtomicU64::new(0),
        }
    }

    /// Record a processed job.
    ///
    /// Called from `TrackedDelivery::ack()` / `nack()` / `reject()`.
    ///
    /// - `duration_ms`: wall-clock processing time in milliseconds.
    /// - `success`: `true` for ack, `false` for nack/reject without requeue.
    #[inline]
    pub fn record_processed(&self, duration_ms: u64, success: bool) {
        self.processed_count.fetch_add(1, Ordering::Relaxed);
        self.total_duration_ms
            .fetch_add(duration_ms, Ordering::Relaxed);

        if success {
            self.success_count.fetch_add(1, Ordering::Relaxed);
        } else {
            self.failure_count.fetch_add(1, Ordering::Relaxed);
        }
    }

    /// Take a snapshot of current counters and reset them to zero.
    ///
    /// Called by the metrics collector on each collection interval.
    /// Uses `swap` to atomically read-and-reset each counter.
    pub fn snapshot_and_reset(&self) -> Tier0Snapshot {
        Tier0Snapshot {
            processed_count: self.processed_count.swap(0, Ordering::Relaxed),
            success_count: self.success_count.swap(0, Ordering::Relaxed),
            failure_count: self.failure_count.swap(0, Ordering::Relaxed),
            total_duration_ms: self.total_duration_ms.swap(0, Ordering::Relaxed),
        }
    }
}

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

// Safety: AtomicU64 is Send+Sync, and we don't hold any non-Send/Sync state.
// These impls are technically auto-derived, but we state them explicitly for clarity.
unsafe impl Send for SlidingWindowCounters {}
unsafe impl Sync for SlidingWindowCounters {}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn new_counters_are_zero() {
        let c = SlidingWindowCounters::new();
        let snap = c.snapshot_and_reset();
        assert_eq!(snap.processed_count, 0);
        assert_eq!(snap.success_count, 0);
        assert_eq!(snap.failure_count, 0);
        assert_eq!(snap.total_duration_ms, 0);
    }

    #[test]
    fn record_success() {
        let c = SlidingWindowCounters::new();
        c.record_processed(150, true);
        c.record_processed(200, true);

        let snap = c.snapshot_and_reset();
        assert_eq!(snap.processed_count, 2);
        assert_eq!(snap.success_count, 2);
        assert_eq!(snap.failure_count, 0);
        assert_eq!(snap.total_duration_ms, 350);
    }

    #[test]
    fn record_failure() {
        let c = SlidingWindowCounters::new();
        c.record_processed(100, false);

        let snap = c.snapshot_and_reset();
        assert_eq!(snap.processed_count, 1);
        assert_eq!(snap.success_count, 0);
        assert_eq!(snap.failure_count, 1);
        assert_eq!(snap.total_duration_ms, 100);
    }

    #[test]
    fn mixed_success_and_failure() {
        let c = SlidingWindowCounters::new();
        c.record_processed(100, true);
        c.record_processed(200, false);
        c.record_processed(300, true);

        let snap = c.snapshot_and_reset();
        assert_eq!(snap.processed_count, 3);
        assert_eq!(snap.success_count, 2);
        assert_eq!(snap.failure_count, 1);
        assert_eq!(snap.total_duration_ms, 600);
    }

    #[test]
    fn snapshot_resets_counters() {
        let c = SlidingWindowCounters::new();
        c.record_processed(100, true);
        c.record_processed(200, false);

        let snap1 = c.snapshot_and_reset();
        assert_eq!(snap1.processed_count, 2);

        // After reset, counters should be zero.
        let snap2 = c.snapshot_and_reset();
        assert_eq!(snap2.processed_count, 0);
        assert_eq!(snap2.success_count, 0);
        assert_eq!(snap2.failure_count, 0);
        assert_eq!(snap2.total_duration_ms, 0);
    }

    #[test]
    fn concurrent_accuracy() {
        use std::sync::Arc;

        let c = Arc::new(SlidingWindowCounters::new());
        let num_threads = 8;
        let ops_per_thread = 10_000;

        let handles: Vec<_> = (0..num_threads)
            .map(|i| {
                let counters = c.clone();
                std::thread::spawn(move || {
                    for j in 0..ops_per_thread {
                        let success = (i + j) % 3 != 0; // ~2/3 success rate
                        counters.record_processed(10, success);
                    }
                })
            })
            .collect();

        for h in handles {
            h.join().unwrap();
        }

        let snap = c.snapshot_and_reset();
        let total = num_threads * ops_per_thread;

        // Total processed must be exact — each thread does exactly ops_per_thread.
        assert_eq!(snap.processed_count, total as u64);

        // Success + failure must equal total.
        assert_eq!(snap.success_count + snap.failure_count, total as u64);

        // Total duration: each op records 10ms.
        assert_eq!(snap.total_duration_ms, (total * 10) as u64);
    }
}
