//! Memory Safety and Circuit Breaker Infrastructure
//!
//! Provides bounded collections, circuit breakers, and memory monitoring
//! to prevent memory exhaustion and runaway allocations in the HFT system.

use anyhow::Result;
use serde::{Deserialize, Serialize};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{mpsc, RwLock};
use tracing::{error, info, warn};

/// Memory thresholds and circuit breaker configuration
#[derive(Debug, Clone)]
/// MemoryConfig
/// 
/// Auto-generated documentation placeholder - enhance with specifics
pub struct MemoryConfig {
    /// Maximum heap usage before circuit breaker trips (bytes)
    pub max_heap_bytes: usize,
    /// Warning threshold (80% of max)
    pub warning_threshold_bytes: usize,
    /// Critical threshold (90% of max)
    pub critical_threshold_bytes: usize,
    /// Channel capacity for orders
    pub order_channel_capacity: usize,
    /// Channel capacity for market data
    pub market_data_channel_capacity: usize,
    /// Channel capacity for risk events
    pub risk_channel_capacity: usize,
    /// Channel capacity for audit logs
    pub audit_channel_capacity: usize,
    /// Maximum collection size
    pub max_collection_size: usize,
    /// Memory check interval
    pub memory_check_interval: Duration,
}

impl Default for MemoryConfig {
    fn default() -> Self {
        let max_heap_gb = std::env::var("MAX_HEAP_GB")
            .unwrap_or_else(|_| "8".to_owned())
            .parse::<usize>()
            .unwrap_or(8);

        let max_heap_bytes = max_heap_gb * 1024 * 1024 * 1024;

        Self {
            max_heap_bytes,
            warning_threshold_bytes: (max_heap_bytes as f64 * 0.8) as usize,
            critical_threshold_bytes: (max_heap_bytes as f64 * 0.9) as usize,
            order_channel_capacity: 10_000,
            market_data_channel_capacity: 100_000,
            risk_channel_capacity: 50_000,
            audit_channel_capacity: 25_000,
            max_collection_size: 1_000_000,
            memory_check_interval: Duration::from_secs(1),
        }
    }
}

/// Circuit breaker for memory protection
#[derive(Debug)]
/// MemoryCircuitBreaker
/// 
/// Auto-generated documentation placeholder - enhance with specifics
pub struct MemoryCircuitBreaker {
    config: MemoryConfig,
    current_usage: Arc<AtomicUsize>,
    is_open: Arc<AtomicBool>,
    last_trip_time: Arc<RwLock<Option<Instant>>>,
    allocation_counter: Arc<AtomicUsize>,
    trip_counter: Arc<AtomicUsize>,
}

impl MemoryCircuitBreaker {
    #[must_use]
    pub fn new(config: MemoryConfig) -> Self {
        Self {
            config,
            current_usage: Arc::new(AtomicUsize::new(0)),
            is_open: Arc::new(AtomicBool::new(false)),
            last_trip_time: Arc::new(RwLock::new(None)),
            allocation_counter: Arc::new(AtomicUsize::new(0)),
            trip_counter: Arc::new(AtomicUsize::new(0)),
        }
    }

    /// Check if allocation is allowed
    pub async fn check_allocation(&self, size: usize) -> Result<()> {
        let current = self.current_usage.load(Ordering::Relaxed);

        // Check if circuit is already open
        if self.is_open.load(Ordering::Relaxed) {
            // Check if we can recover (simple time-based recovery)
            if let Some(trip_time) = *self.last_trip_time.read().await {
                if trip_time.elapsed() > Duration::from_secs(30) {
                    self.reset_circuit().await;
                    info!("Memory circuit breaker reset after recovery period");
                } else {
                    return Err(anyhow::anyhow!("Memory circuit breaker is open"));
                }
            }
        }

        // Check if new allocation would exceed threshold
        if current + size > self.config.critical_threshold_bytes {
            self.trip_circuit().await;
            return Err(anyhow::anyhow!(
                "Memory allocation would exceed threshold: {} + {} > {}",
                current,
                size,
                self.config.critical_threshold_bytes
            ));
        }

        // Update usage counter
        self.current_usage.fetch_add(size, Ordering::Relaxed);
        self.allocation_counter.fetch_add(1, Ordering::Relaxed);

        // Log warning if approaching threshold
        let new_usage = current + size;
        if new_usage > self.config.warning_threshold_bytes {
            warn!(
                "Memory usage approaching threshold: {} / {} bytes ({:.1}%)",
                new_usage,
                self.config.max_heap_bytes,
                (new_usage as f64 / self.config.max_heap_bytes as f64) * 100.0
            );
        }

        Ok(())
    }

    /// Release memory allocation
    pub fn release_allocation(&self, size: usize) {
        self.current_usage.fetch_sub(size, Ordering::Relaxed);
    }

    /// Trip the circuit breaker
    async fn trip_circuit(&self) {
        self.is_open.store(true, Ordering::SeqCst);
        *self.last_trip_time.write().await = Some(Instant::now());
        self.trip_counter.fetch_add(1, Ordering::Relaxed);

        error!(
            "Memory circuit breaker TRIPPED! Usage: {} bytes, Threshold: {} bytes",
            self.current_usage.load(Ordering::Relaxed),
            self.config.critical_threshold_bytes
        );
    }

    /// Reset the circuit breaker
    async fn reset_circuit(&self) {
        self.is_open.store(false, Ordering::SeqCst);
        *self.last_trip_time.write().await = None;

        info!(
            "Memory circuit breaker RESET. Current usage: {} bytes",
            self.current_usage.load(Ordering::Relaxed)
        );
    }

    /// Get current memory statistics
    #[must_use]
    pub fn get_memory_stats(&self) -> MemoryStats {
        let current_usage = self.current_usage.load(Ordering::Relaxed);
        let usage_percentage = (current_usage as f64 / self.config.max_heap_bytes as f64) * 100.0;

        MemoryStats {
            current_usage_bytes: current_usage,
            max_heap_bytes: self.config.max_heap_bytes,
            usage_percentage,
            is_circuit_open: self.is_open.load(Ordering::Relaxed),
            allocation_count: self.allocation_counter.load(Ordering::Relaxed),
            trip_count: self.trip_counter.load(Ordering::Relaxed),
            warning_threshold_bytes: self.config.warning_threshold_bytes,
            critical_threshold_bytes: self.config.critical_threshold_bytes,
        }
    }
}

/// Memory usage statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
/// MemoryStats
/// 
/// Auto-generated documentation placeholder - enhance with specifics
pub struct MemoryStats {
    /// Current Usage Bytes
    pub current_usage_bytes: usize,
    /// Max Heap Bytes
    pub max_heap_bytes: usize,
    /// Usage Percentage
    pub usage_percentage: f64,
    /// Is Circuit Open
    pub is_circuit_open: bool,
    /// Allocation Count
    pub allocation_count: usize,
    /// Trip Count
    pub trip_count: usize,
    /// Warning Threshold Bytes
    pub warning_threshold_bytes: usize,
    /// Critical Threshold Bytes
    pub critical_threshold_bytes: usize,
}

/// Bounded channel with automatic backpressure handling
#[derive(Debug)]
/// BoundedChannel
/// 
/// Auto-generated documentation placeholder - enhance with specifics
pub struct BoundedChannel<T> {
    sender: mpsc::Sender<T>,
    receiver: Option<mpsc::Receiver<T>>,
    capacity: usize,
    overflow_strategy: OverflowStrategy,
    dropped_count: Arc<AtomicUsize>,
    sent_count: Arc<AtomicUsize>,
    circuit_breaker: Arc<MemoryCircuitBreaker>,
}

#[derive(Debug, Clone)]
/// OverflowStrategy
/// 
/// Auto-generated documentation placeholder - enhance with specifics
pub enum OverflowStrategy {
    // DropOldest variant
    DropOldest,
    // DropNewest variant
    DropNewest,
    // RejectNew variant
    RejectNew,
    // Block variant
    Block,
}

impl<T> BoundedChannel<T> {
    #[must_use]
    pub fn new(
        capacity: usize,
        strategy: OverflowStrategy,
        circuit_breaker: Arc<MemoryCircuitBreaker>,
    ) -> Self {
        let (sender, receiver) = mpsc::channel(capacity);

        Self {
            sender,
            receiver: Some(receiver),
            capacity,
            overflow_strategy: strategy,
            dropped_count: Arc::new(AtomicUsize::new(0)),
            sent_count: Arc::new(AtomicUsize::new(0)),
            circuit_breaker,
        }
    }

    /// Send message with overflow handling
    pub async fn send(&self, msg: T) -> Result<()> {
        // Check circuit breaker first
        let msg_size = size_of::<T>();
        self.circuit_breaker.check_allocation(msg_size).await?;

        match self.sender.try_send(msg) {
            Ok(()) => {
                self.sent_count.fetch_add(1, Ordering::Relaxed);
                Ok(())
            }
            Err(mpsc::error::TrySendError::Full(msg)) => {
                match self.overflow_strategy {
                    OverflowStrategy::DropNewest => {
                        self.dropped_count.fetch_add(1, Ordering::Relaxed);
                        self.circuit_breaker.release_allocation(msg_size);
                        warn!("Channel full, dropping newest message");
                        Ok(())
                    }
                    OverflowStrategy::RejectNew => {
                        self.dropped_count.fetch_add(1, Ordering::Relaxed);
                        self.circuit_breaker.release_allocation(msg_size);
                        Err(anyhow::anyhow!("Channel full, rejecting new message"))
                    }
                    OverflowStrategy::Block => {
                        // Use blocking send
                        self.sender
                            .send(msg)
                            .await
                            .map_err(|e| anyhow::anyhow!("Failed to send message: {e}"))?;
                        self.sent_count.fetch_add(1, Ordering::Relaxed);
                        Ok(())
                    }
                    OverflowStrategy::DropOldest => {
                        // For this we'd need a different channel implementation
                        // For now, just drop newest
                        self.dropped_count.fetch_add(1, Ordering::Relaxed);
                        self.circuit_breaker.release_allocation(msg_size);
                        warn!("Channel full, dropping message (drop oldest not implemented)");
                        Ok(())
                    }
                }
            }
            Err(mpsc::error::TrySendError::Closed(_)) => {
                self.circuit_breaker.release_allocation(msg_size);
                Err(anyhow::anyhow!("Channel closed"))
            }
        }
    }

    /// Get the receiver (can only be called once)
    pub fn receiver(&mut self) -> Option<mpsc::Receiver<T>> {
        self.receiver.take()
    }

    /// Get channel statistics
    #[must_use]
    pub fn get_stats(&self) -> ChannelStats {
        ChannelStats {
            capacity: self.capacity,
            sent_count: self.sent_count.load(Ordering::Relaxed),
            dropped_count: self.dropped_count.load(Ordering::Relaxed),
            strategy: self.overflow_strategy.clone(),
        }
    }
}

/// Channel statistics
#[derive(Debug, Clone)]
/// ChannelStats
/// 
/// Auto-generated documentation placeholder - enhance with specifics
pub struct ChannelStats {
    /// Capacity
    pub capacity: usize,
    /// Sent Count
    pub sent_count: usize,
    /// Dropped Count
    pub dropped_count: usize,
    /// Strategy
    pub strategy: OverflowStrategy,
}

/// Bounded collection wrapper
#[derive(Debug)]
/// BoundedVec
/// 
/// Auto-generated documentation placeholder - enhance with specifics
pub struct BoundedVec<T> {
    inner: Vec<T>,
    max_size: usize,
    overflow_strategy: OverflowStrategy,
    dropped_count: Arc<AtomicUsize>,
    circuit_breaker: Arc<MemoryCircuitBreaker>,
}

impl<T> BoundedVec<T> {
    #[must_use]
    pub fn new(
        max_size: usize,
        strategy: OverflowStrategy,
        circuit_breaker: Arc<MemoryCircuitBreaker>,
    ) -> Self {
        Self {
            inner: Vec::with_capacity(std::cmp::min(max_size, 1024)), // Start with reasonable capacity
            max_size,
            overflow_strategy: strategy,
            dropped_count: Arc::new(AtomicUsize::new(0)),
            circuit_breaker,
        }
    }

    pub async fn push(&mut self, item: T) -> Result<()> {
        let item_size = size_of::<T>();
        self.circuit_breaker.check_allocation(item_size).await?;

        if self.inner.len() >= self.max_size {
            match self.overflow_strategy {
                OverflowStrategy::DropOldest => {
                    if !self.inner.is_empty() {
                        self.inner.remove(0);
                        self.circuit_breaker.release_allocation(item_size);
                    }
                    self.inner.push(item);
                }
                OverflowStrategy::DropNewest => {
                    self.dropped_count.fetch_add(1, Ordering::Relaxed);
                    self.circuit_breaker.release_allocation(item_size);
                    return Ok(());
                }
                OverflowStrategy::RejectNew => {
                    self.dropped_count.fetch_add(1, Ordering::Relaxed);
                    self.circuit_breaker.release_allocation(item_size);
                    return Err(anyhow::anyhow!("Collection at maximum capacity"));
                }
                OverflowStrategy::Block => {
                    // Can't really block in a sync context, so reject
                    self.dropped_count.fetch_add(1, Ordering::Relaxed);
                    self.circuit_breaker.release_allocation(item_size);
                    return Err(anyhow::anyhow!("Collection at maximum capacity"));
                }
            }
        } else {
            self.inner.push(item);
        }

        Ok(())
    }

    #[must_use]
    pub fn len(&self) -> usize {
        self.inner.len()
    }

    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.inner.is_empty()
    }

    #[must_use]
    pub fn get(&self, index: usize) -> Option<&T> {
        self.inner.get(index)
    }

    pub fn iter(&self) -> std::slice::Iter<'_, T> {
        self.inner.iter()
    }

    pub fn clear(&mut self) {
        let freed_size = self.inner.len() * size_of::<T>();
        self.inner.clear();
        self.circuit_breaker.release_allocation(freed_size);
    }
}

/// Global memory monitor instance
pub static MEMORY_MONITOR: once_cell::sync::Lazy<Arc<MemoryCircuitBreaker>> =
    once_cell::sync::Lazy::new(|| Arc::new(MemoryCircuitBreaker::new(MemoryConfig::default())));

/// Create bounded channel with global circuit breaker
pub fn create_bounded_channel<T>(capacity: usize, strategy: OverflowStrategy) -> BoundedChannel<T> {
    BoundedChannel::new(capacity, strategy, MEMORY_MONITOR.clone())
}

/// Create bounded vector with global circuit breaker
pub fn create_bounded_vec<T>(max_size: usize, strategy: OverflowStrategy) -> BoundedVec<T> {
    BoundedVec::new(max_size, strategy, MEMORY_MONITOR.clone())
}

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

    #[tokio::test]
    async fn test_memory_circuit_breaker() {
        let config = MemoryConfig {
            max_heap_bytes: 1000,
            warning_threshold_bytes: 800,
            critical_threshold_bytes: 900,
            ..Default::default()
        };

        let breaker = MemoryCircuitBreaker::new(config);

        // Normal allocation should work
        assert!(breaker.check_allocation(100).await.is_ok());

        // Large allocation should trip circuit
        assert!(breaker.check_allocation(950).await.is_err());

        // Circuit should be open
        assert!(breaker.is_open.load(Ordering::Relaxed));
    }

    #[tokio::test]
    async fn test_bounded_channel() {
        let config = MemoryConfig::default();
        let breaker = Arc::new(MemoryCircuitBreaker::new(config));

        let mut channel = BoundedChannel::new(2, OverflowStrategy::DropNewest, breaker);
        let mut rx = channel.receiver().unwrap();

        // Should be able to send up to capacity
        assert!(channel.send(1).await.is_ok());
        assert!(channel.send(2).await.is_ok());

        // Next send should succeed but drop (due to DropNewest strategy)
        assert!(channel.send(3).await.is_ok());

        // Should have dropped one message
        assert_eq!(channel.dropped_count.load(Ordering::Relaxed), 1);
    }
}
