foxhunt/tli/src/events/event_buffer.rs

//! Event aggregation and buffering with back-pressure handling
//!
//! This module provides memory-efficient event storage with:
//! - Circular buffer with configurable size limits
//! - Back-pressure handling and flow control
//! - Event TTL and automatic cleanup
//! - Memory usage monitoring and alerts
//! - Batch processing and compression
//! - Priority-based event handling

use crate::error::{TliError, TliResult};
use crate::events::{Event, EventFilter, EventSeverity, EventType};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use tokio::sync::{watch, RwLock, Semaphore};
use tokio::time::{interval, Duration, Instant};
use tracing::{debug, error, info, instrument, warn};
use uuid::Uuid;

/// Configuration for event buffer
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EventBufferConfig {
    /// Maximum number of events to store
    pub max_events: usize,
    /// Maximum memory usage in bytes
    pub max_memory_bytes: usize,
    /// Event TTL in seconds (0 = no expiry)
    pub default_ttl_seconds: u64,
    /// Cleanup interval in seconds
    pub cleanup_interval_seconds: u64,
    /// Enable compression for stored events
    pub enable_compression: bool,
    /// Compression threshold in bytes
    pub compression_threshold_bytes: usize,
    /// Enable back-pressure when buffer is full
    pub enable_backpressure: bool,
    /// Back-pressure threshold (percentage of max_events)
    pub backpressure_threshold_percent: f32,
    /// Batch size for processing events
    pub batch_size: usize,
    /// Enable priority queue for critical events
    pub enable_priority_queue: bool,
    /// Memory warning threshold (percentage of max_memory_bytes)
    pub memory_warning_threshold_percent: f32,
}

impl Default for EventBufferConfig {
    fn default() -> Self {
        Self {
            max_events: 100_000,
            max_memory_bytes: 100 * 1024 * 1024, // 100MB
            default_ttl_seconds: 3600,           // 1 hour
            cleanup_interval_seconds: 60,        // 1 minute
            enable_compression: true,
            compression_threshold_bytes: 1024, // 1KB
            enable_backpressure: true,
            backpressure_threshold_percent: 0.8, // 80%
            batch_size: 100,
            enable_priority_queue: true,
            memory_warning_threshold_percent: 0.9, // 90%
        }
    }
}

/// Event buffer metrics for monitoring
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EventBufferMetrics {
    /// Total events currently stored
    pub events_stored: usize,
    /// Memory usage in bytes
    pub memory_usage_bytes: usize,
    /// Events added since start
    pub events_added: u64,
    /// Events removed since start
    pub events_removed: u64,
    /// Events expired since start
    pub events_expired: u64,
    /// Events compressed since start
    pub events_compressed: u64,
    /// Current back-pressure status
    pub backpressure_active: bool,
    /// Number of times back-pressure was triggered
    pub backpressure_count: u64,
    /// Average event size in bytes
    pub average_event_size_bytes: f64,
    /// Events by type
    pub events_by_type: HashMap<String, usize>,
    /// Events by severity
    pub events_by_severity: HashMap<String, usize>,
    /// Last cleanup time
    pub last_cleanup_at: Option<DateTime<Utc>>,
    /// Buffer utilization percentage
    pub utilization_percent: f32,
}

impl Default for EventBufferMetrics {
    fn default() -> Self {
        Self {
            events_stored: 0,
            memory_usage_bytes: 0,
            events_added: 0,
            events_removed: 0,
            events_expired: 0,
            events_compressed: 0,
            backpressure_active: false,
            backpressure_count: 0,
            average_event_size_bytes: 0.0,
            events_by_type: HashMap::new(),
            events_by_severity: HashMap::new(),
            last_cleanup_at: None,
            utilization_percent: 0.0,
        }
    }
}

/// Stored event with metadata
#[derive(Debug, Clone)]
struct StoredEvent {
    /// The event data
    event: Event,
    /// Size in bytes
    size_bytes: usize,
    /// Compressed payload (if compression enabled)
    #[allow(dead_code)]
    compressed_payload: Option<Vec<u8>>,
    /// Insert timestamp
    #[allow(dead_code)]
    inserted_at: Instant,
}

impl StoredEvent {
    fn new(event: Event) -> Self {
        let size_bytes = Self::calculate_size(&event);
        Self {
            event,
            size_bytes,
            compressed_payload: None,
            inserted_at: Instant::now(),
        }
    }

    fn calculate_size(event: &Event) -> usize {
        // Rough estimation of event size in memory
        size_of::<Event>()
            + event.source.len()
            + event.payload.to_string().len()
            + event
                .metadata
                .iter()
                .map(|(k, v)| k.len() + v.len())
                .sum::<usize>()
    }

    #[allow(dead_code)]
    fn compress(&mut self) -> TliResult<()> {
        if self.compressed_payload.is_some() {
            return Ok(()); // Already compressed
        }

        let payload_str = self.event.payload.to_string();
        if payload_str.len() < 1024 {
            return Ok(()); // Too small to compress
        }

        // Simple compression using flate2 (would need to add dependency)
        // For now, just store as-is
        self.compressed_payload = Some(payload_str.into_bytes());
        Ok(())
    }
}

/// Priority level for events
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
enum EventPriority {
    Low = 0,
    Normal = 1,
    High = 2,
    Critical = 3,
}

impl From<EventSeverity> for EventPriority {
    fn from(severity: EventSeverity) -> Self {
        match severity {
            EventSeverity::Info => EventPriority::Low,
            EventSeverity::Warning => EventPriority::Normal,
            EventSeverity::Error => EventPriority::High,
            EventSeverity::Critical => EventPriority::Critical,
        }
    }
}

/// Event buffer that manages memory-efficient event storage
pub struct EventBuffer {
    /// Configuration
    config: EventBufferConfig,
    /// Main event storage (circular buffer)
    events: Arc<RwLock<VecDeque<StoredEvent>>>,
    /// Priority queue for critical events
    priority_events: Arc<RwLock<VecDeque<StoredEvent>>>,
    /// Event index for fast lookups
    event_index: Arc<RwLock<HashMap<Uuid, usize>>>,
    /// Buffer metrics
    metrics: Arc<RwLock<EventBufferMetrics>>,
    /// Back-pressure semaphore
    backpressure_semaphore: Arc<Semaphore>,
    /// Shutdown signal
    shutdown_sender: watch::Sender<bool>,
    shutdown_receiver: watch::Receiver<bool>,
}

impl EventBuffer {
    /// Create a new event buffer
    pub fn new(config: EventBufferConfig) -> Self {
        let backpressure_permits =
            (config.max_events as f32 * config.backpressure_threshold_percent) as usize;
        let backpressure_semaphore = Arc::new(Semaphore::new(backpressure_permits));

        let (shutdown_sender, shutdown_receiver) = watch::channel(false);

        let buffer = Self {
            config,
            events: Arc::new(RwLock::new(VecDeque::new())),
            priority_events: Arc::new(RwLock::new(VecDeque::new())),
            event_index: Arc::new(RwLock::new(HashMap::new())),
            metrics: Arc::new(RwLock::new(EventBufferMetrics::default())),
            backpressure_semaphore,
            shutdown_sender,
            shutdown_receiver,
        };

        // Start cleanup task
        buffer.start_cleanup_task();

        buffer
    }

    /// Add an event to the buffer
    #[instrument(skip(self, event))]
    pub async fn add_event(&self, event: Event) -> TliResult<()> {
        // Check back-pressure
        if self.config.enable_backpressure {
            let permit = self.backpressure_semaphore.try_acquire().map_err(|_| {
                // Update back-pressure metrics
                tokio::spawn({
                    let metrics = self.metrics.clone();
                    async move {
                        let mut m = metrics.write().await;
                        m.backpressure_active = true;
                        m.backpressure_count += 1;
                    }
                });
                TliError::BufferFull("Event buffer back-pressure active".to_string())
            })?;

            // Release permit after processing
            std::mem::forget(permit);
        }

        let stored_event = StoredEvent::new(event.clone());
        let event_id = event.id;
        let priority = EventPriority::from(event.severity.clone());

        // Determine which queue to use
        let use_priority_queue = self.config.enable_priority_queue
            && (priority == EventPriority::Critical || priority == EventPriority::High);

        if use_priority_queue {
            // Add to priority queue
            let mut priority_events = self.priority_events.write().await;
            priority_events.push_back(stored_event);

            // Ensure priority queue doesn't grow too large
            let max_priority_events = self.config.max_events / 10; // 10% of total
            while priority_events.len() > max_priority_events {
                if let Some(removed) = priority_events.pop_front() {
                    self.update_metrics_on_removal(&removed.event).await;
                }
            }
        } else {
            // Add to main buffer
            let mut events = self.events.write().await;
            let mut index = self.event_index.write().await;

            // Check if buffer is full
            if events.len() >= self.config.max_events {
                // Remove oldest event
                if let Some(removed) = events.pop_front() {
                    index.remove(&removed.event.id);
                    self.update_metrics_on_removal(&removed.event).await;
                }
            }

            // Add new event
            let position = events.len();
            events.push_back(stored_event);
            index.insert(event_id, position);
        }

        // Update metrics
        self.update_metrics_on_addition(&event).await;

        // Check memory usage
        self.check_memory_usage().await;

        Ok(())
    }

    /// Get events matching a filter
    pub async fn get_events(&self, filter: &EventFilter, limit: Option<usize>) -> Vec<Event> {
        let mut result = Vec::new();
        let max_results = limit.unwrap_or(1000);

        // Check priority events first
        if self.config.enable_priority_queue {
            let priority_events = self.priority_events.read().await;
            for stored_event in priority_events.iter().rev() {
                // Most recent first
                if result.len() >= max_results {
                    break;
                }

                if !stored_event.event.is_expired() && filter.matches(&stored_event.event) {
                    result.push(stored_event.event.clone());
                }
            }
        }

        // Check main buffer
        if result.len() < max_results {
            let events = self.events.read().await;
            for stored_event in events.iter().rev() {
                // Most recent first
                if result.len() >= max_results {
                    break;
                }

                if !stored_event.event.is_expired() && filter.matches(&stored_event.event) {
                    result.push(stored_event.event.clone());
                }
            }
        }

        result
    }

    /// Get event by ID
    pub async fn get_event_by_id(&self, id: &Uuid) -> Option<Event> {
        // Check priority events first
        if self.config.enable_priority_queue {
            let priority_events = self.priority_events.read().await;
            for stored_event in priority_events.iter() {
                if stored_event.event.id == *id && !stored_event.event.is_expired() {
                    return Some(stored_event.event.clone());
                }
            }
        }

        // Check main buffer
        let index = self.event_index.read().await;
        if let Some(&position) = index.get(id) {
            let events = self.events.read().await;
            if let Some(stored_event) = events.get(position) {
                if !stored_event.event.is_expired() {
                    return Some(stored_event.event.clone());
                }
            }
        }

        None
    }

    /// Get events in a time range
    pub async fn get_events_in_range(
        &self,
        start_time_nanos: i64,
        end_time_nanos: i64,
        limit: Option<usize>,
    ) -> Vec<Event> {
        let filter = EventFilter {
            start_time_nanos: Some(start_time_nanos),
            end_time_nanos: Some(end_time_nanos),
            ..EventFilter::all()
        };

        self.get_events(&filter, limit).await
    }

    /// Get buffer metrics
    pub async fn get_metrics(&self) -> EventBufferMetrics {
        self.metrics.read().await.clone()
    }

    /// Manually trigger cleanup
    pub async fn cleanup(&self) -> TliResult<()> {
        let mut expired_count = 0;
        let mut memory_freed = 0;

        // Clean priority events
        if self.config.enable_priority_queue {
            let mut priority_events = self.priority_events.write().await;
            let original_len = priority_events.len();

            priority_events.retain(|stored_event| {
                let expired = stored_event.event.is_expired();
                if expired {
                    memory_freed += stored_event.size_bytes;
                }
                !expired
            });

            expired_count += original_len - priority_events.len();
        }

        // Clean main buffer
        {
            let mut events = self.events.write().await;
            let mut index = self.event_index.write().await;
            let original_len = events.len();

            let mut retained_events = VecDeque::new();
            let mut new_index = HashMap::new();

            for (_pos, stored_event) in events.drain(..).enumerate() {
                if !stored_event.event.is_expired() {
                    let new_pos = retained_events.len();
                    let event_id = stored_event.event.id;
                    retained_events.push_back(stored_event);
                    new_index.insert(event_id, new_pos);
                } else {
                    memory_freed += stored_event.size_bytes;
                }
            }

            *events = retained_events;
            *index = new_index;

            expired_count += original_len - events.len();
        }

        // Update metrics
        {
            let mut metrics = self.metrics.write().await;
            metrics.events_expired += expired_count as u64;
            metrics.memory_usage_bytes = metrics.memory_usage_bytes.saturating_sub(memory_freed);
            metrics.last_cleanup_at = Some(Utc::now());
            metrics.events_stored = metrics.events_stored.saturating_sub(expired_count);

            // Update utilization
            metrics.utilization_percent =
                (metrics.events_stored as f32 / self.config.max_events as f32) * 100.0;
        }

        if expired_count > 0 {
            debug!(
                "Cleaned up {} expired events, freed {} bytes",
                expired_count, memory_freed
            );
        }

        Ok(())
    }

    /// Clear all events from buffer
    pub async fn clear(&self) -> TliResult<()> {
        {
            let mut events = self.events.write().await;
            let mut priority_events = self.priority_events.write().await;
            let mut index = self.event_index.write().await;

            events.clear();
            priority_events.clear();
            index.clear();
        }

        // Reset metrics
        {
            let mut metrics = self.metrics.write().await;
            *metrics = EventBufferMetrics::default();
        }

        info!("Event buffer cleared");
        Ok(())
    }

    /// Start the cleanup task
    fn start_cleanup_task(&self) {
        let buffer = self.clone();
        tokio::spawn(async move {
            let mut interval =
                interval(Duration::from_secs(buffer.config.cleanup_interval_seconds));
            let mut shutdown = buffer.shutdown_receiver.clone();

            loop {
                tokio::select! {
                    _ = interval.tick() => {
                        if let Err(e) = buffer.cleanup().await {
                            error!("Cleanup task error: {}", e);
                        }
                    }
                    _ = shutdown.changed() => {
                        if *shutdown.borrow() {
                            debug!("Cleanup task shutting down");
                            break;
                        }
                    }
                }
            }
        });
    }

    /// Update metrics when adding an event
    async fn update_metrics_on_addition(&self, event: &Event) {
        let mut metrics = self.metrics.write().await;
        metrics.events_added += 1;
        metrics.events_stored += 1;

        let event_size = StoredEvent::calculate_size(event);
        metrics.memory_usage_bytes += event_size;

        // Update average size
        metrics.average_event_size_bytes = (metrics.average_event_size_bytes
            * (metrics.events_added - 1) as f64
            + event_size as f64)
            / metrics.events_added as f64;

        // Update type counts
        let type_key = event.event_type.as_str().to_string();
        *metrics.events_by_type.entry(type_key).or_insert(0) += 1;

        // Update severity counts
        let severity_key = match event.severity {
            EventSeverity::Info => "info",
            EventSeverity::Warning => "warning",
            EventSeverity::Error => "error",
            EventSeverity::Critical => "critical",
        }
        .to_string();
        *metrics.events_by_severity.entry(severity_key).or_insert(0) += 1;

        // Update utilization
        metrics.utilization_percent =
            (metrics.events_stored as f32 / self.config.max_events as f32) * 100.0;

        // Reset back-pressure if no longer needed
        if metrics.backpressure_active && metrics.utilization_percent < 70.0 {
            metrics.backpressure_active = false;
        }
    }

    /// Update metrics when removing an event
    async fn update_metrics_on_removal(&self, event: &Event) {
        let mut metrics = self.metrics.write().await;
        metrics.events_removed += 1;
        metrics.events_stored = metrics.events_stored.saturating_sub(1);

        let event_size = StoredEvent::calculate_size(event);
        metrics.memory_usage_bytes = metrics.memory_usage_bytes.saturating_sub(event_size);

        // Update type counts
        let type_key = event.event_type.as_str().to_string();
        if let Some(count) = metrics.events_by_type.get_mut(&type_key) {
            *count = count.saturating_sub(1);
        }

        // Update severity counts
        let severity_key = match event.severity {
            EventSeverity::Info => "info",
            EventSeverity::Warning => "warning",
            EventSeverity::Error => "error",
            EventSeverity::Critical => "critical",
        }
        .to_string();
        if let Some(count) = metrics.events_by_severity.get_mut(&severity_key) {
            *count = count.saturating_sub(1);
        }

        // Update utilization
        metrics.utilization_percent =
            (metrics.events_stored as f32 / self.config.max_events as f32) * 100.0;
    }

    /// Check memory usage and trigger warnings
    async fn check_memory_usage(&self) {
        let metrics = self.metrics.read().await;
        let usage_percent =
            (metrics.memory_usage_bytes as f32 / self.config.max_memory_bytes as f32) * 100.0;

        if usage_percent > self.config.memory_warning_threshold_percent * 100.0 {
            warn!(
                "Event buffer memory usage high: {:.1}% ({} bytes)",
                usage_percent, metrics.memory_usage_bytes
            );
        }
    }

    /// Shutdown the buffer
    pub async fn shutdown(&self) -> TliResult<()> {
        info!("Shutting down event buffer");

        if let Err(e) = self.shutdown_sender.send(true) {
            warn!("Failed to send shutdown signal: {}", e);
        }

        // Final cleanup
        self.cleanup().await?;

        info!("Event buffer shutdown complete");
        Ok(())
    }
}

impl Clone for EventBuffer {
    fn clone(&self) -> Self {
        Self {
            config: self.config.clone(),
            events: self.events.clone(),
            priority_events: self.priority_events.clone(),
            event_index: self.event_index.clone(),
            metrics: self.metrics.clone(),
            backpressure_semaphore: self.backpressure_semaphore.clone(),
            shutdown_sender: self.shutdown_sender.clone(),
            shutdown_receiver: self.shutdown_receiver.clone(),
        }
    }
}

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

    #[tokio::test]
    async fn test_event_buffer_basic_operations() {
        let config = EventBufferConfig {
            max_events: 10,
            ..EventBufferConfig::default()
        };
        let buffer = EventBuffer::new(config);

        // Add some events
        for i in 0..5 {
            let event = Event::new(
                EventType::Trading,
                EventSeverity::Info,
                "test".to_string(),
                serde_json::json!({"index": i}),
            );
            buffer.add_event(event).await.unwrap();
        }

        // Check metrics
        let metrics = buffer.get_metrics().await;
        assert_eq!(metrics.events_stored, 5);
        assert_eq!(metrics.events_added, 5);

        // Get all events
        let events = buffer.get_events(&EventFilter::all(), None).await;
        assert_eq!(events.len(), 5);
    }

    #[tokio::test]
    async fn test_event_buffer_overflow() {
        let config = EventBufferConfig {
            max_events: 3,
            ..EventBufferConfig::default()
        };
        let buffer = EventBuffer::new(config);

        // Add more events than the limit
        for i in 0..5 {
            let event = Event::new(
                EventType::Trading,
                EventSeverity::Info,
                "test".to_string(),
                serde_json::json!({"index": i}),
            );
            buffer.add_event(event).await.unwrap();
        }

        // Should only have max_events
        let metrics = buffer.get_metrics().await;
        assert_eq!(metrics.events_stored, 3);
        assert_eq!(metrics.events_added, 5);
        assert_eq!(metrics.events_removed, 2);
    }

    #[tokio::test]
    async fn test_event_filter() {
        let buffer = EventBuffer::new(EventBufferConfig::default());

        // Add events of different types
        let trading_event = Event::new(
            EventType::Trading,
            EventSeverity::Info,
            "test".to_string(),
            serde_json::json!({}),
        );
        let market_event = Event::new(
            EventType::MarketData,
            EventSeverity::Warning,
            "test".to_string(),
            serde_json::json!({}),
        );

        buffer.add_event(trading_event).await.unwrap();
        buffer.add_event(market_event).await.unwrap();

        // Filter by type
        let trading_filter = EventFilter::for_types(vec![EventType::Trading]);
        let trading_events = buffer.get_events(&trading_filter, None).await;
        assert_eq!(trading_events.len(), 1);

        // Filter by severity
        let warning_filter = EventFilter::with_min_severity(EventSeverity::Warning);
        let warning_events = buffer.get_events(&warning_filter, None).await;
        assert_eq!(warning_events.len(), 1);
    }
}