//! # Trading Service Event Streaming Module
//!
//! This module provides event streaming capabilities for the trading service,
//! allowing real-time broadcasting of trading events to subscribers.
//!
//! ## Features
//!
//! - Publishing trading events (orders, fills, cancellations)
//! - Risk management event notifications
//! - Market data event streaming
//! - Performance metrics and system events
//! - Event filtering and subscription management

use crate::error::Result;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{broadcast, RwLock};
use tracing::{debug, info};

// Import trading event types
use crate::event_streaming::filters::EventFilter;
use crate::event_streaming::subscriber::TradingEventReceiver;

pub mod events;
pub mod filters;
pub mod publisher;
pub mod subscriber;

// Re-export key types
pub use events::{TradingEvent, TradingEventType};
pub use publisher::EventPublisher;

/// Trading event streaming system
#[derive(Debug, Clone)]
pub struct TradingEventStreamer {
    /// Event publisher for broadcasting events
    pub publisher: EventPublisher,
    /// Subscription manager for handling client subscriptions
    pub subscription_manager: Arc<RwLock<SubscriptionManager>>,
    /// Event buffer for reliable delivery
    pub event_buffer: Arc<RwLock<EventBuffer>>,
    /// Configuration for the streaming system
    pub config: EventStreamingConfig,
}

impl TradingEventStreamer {
    /// Create a new trading event streamer
    pub fn new(config: EventStreamingConfig) -> Self {
        let (sender, _receiver) = broadcast::channel(config.max_subscribers);

        Self {
            publisher: EventPublisher::new(sender),
            subscription_manager: Arc::new(RwLock::new(SubscriptionManager::new())),
            event_buffer: Arc::new(RwLock::new(EventBuffer::new(config.buffer_size))),
            config,
        }
    }

    /// Start the event streaming system
    pub async fn start(&self) -> Result<()> {
        info!("Starting trading event streaming system");

        // Initialize event buffer cleanup task
        let buffer = self.event_buffer.clone();
        let cleanup_interval = self.config.cleanup_interval_secs;

        tokio::spawn(async move {
            let mut interval =
                tokio::time::interval(std::time::Duration::from_secs(cleanup_interval));

            loop {
                interval.tick().await;

                let mut buffer = buffer.write().await;
                buffer.cleanup_expired_events();
            }
        });

        info!("Trading event streaming system started successfully");
        Ok(())
    }

    /// Publish a trading event
    pub async fn publish_event(&self, event: TradingEvent) -> Result<()> {
        // Store event in buffer for replay
        {
            let mut buffer = self.event_buffer.write().await;
            buffer.add_event(event.clone()).await?;
        }

        // Publish event to subscribers
        self.publisher.publish(event.clone()).await?;

        debug!("Published trading event: {:?}", event.event_type());
        Ok(())
    }

    /// Subscribe to trading events with a filter
    pub async fn subscribe(&self, filter: EventFilter) -> Result<TradingEventReceiver> {
        let receiver = self.publisher.subscribe()?;
        let subscription_id = uuid::Uuid::new_v4().to_string();

        {
            let mut manager = self.subscription_manager.write().await;
            manager.add_subscription(subscription_id.clone(), filter.clone());
        }

        Ok(TradingEventReceiver::new(subscription_id, receiver, filter))
    }

    /// Unsubscribe from trading events
    pub async fn unsubscribe(&self, subscription_id: &str) -> Result<()> {
        let mut manager = self.subscription_manager.write().await;
        manager.remove_subscription(subscription_id);

        debug!("Unsubscribed client: {}", subscription_id);
        Ok(())
    }

    /// Get streaming system metrics
    pub async fn get_metrics(&self) -> StreamingMetrics {
        let manager = self.subscription_manager.read().await;
        let buffer = self.event_buffer.read().await;

        StreamingMetrics {
            active_subscriptions: manager.subscription_count(),
            events_published: self.publisher.get_published_count(),
            events_buffered: buffer.len(),
            buffer_memory_usage: buffer.memory_usage(),
            uptime_seconds: self.publisher.get_uptime().as_secs(),
        }
    }

    /// Get historical events from buffer
    pub async fn get_historical_events(
        &self,
        filter: EventFilter,
        limit: Option<usize>,
    ) -> Result<Vec<TradingEvent>> {
        let buffer = self.event_buffer.read().await;
        Ok(buffer.get_filtered_events(filter, limit))
    }

    /// Shutdown the streaming system
    pub async fn shutdown(&self) -> Result<()> {
        info!("Shutting down trading event streaming system");

        // Clear all subscriptions
        {
            let mut manager = self.subscription_manager.write().await;
            manager.clear_all_subscriptions();
        }

        // Clear event buffer
        {
            let mut buffer = self.event_buffer.write().await;
            buffer.clear();
        }

        info!("Trading event streaming system shutdown complete");
        Ok(())
    }
}

/// Configuration for the trading event streaming system
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EventStreamingConfig {
    /// Maximum number of concurrent subscribers
    pub max_subscribers: usize,
    /// Event buffer size for replay capability
    pub buffer_size: usize,
    /// Event cleanup interval in seconds
    pub cleanup_interval_secs: u64,
    /// Maximum event age in seconds before cleanup
    pub max_event_age_secs: u64,
    /// Enable event compression for storage
    pub enable_compression: bool,
    /// Maximum memory usage for event buffer (bytes)
    pub max_buffer_memory: usize,
}

impl Default for EventStreamingConfig {
    fn default() -> Self {
        Self {
            max_subscribers: 1000,
            buffer_size: 10000,
            cleanup_interval_secs: 60,
            max_event_age_secs: 3600, // 1 hour
            enable_compression: true,
            max_buffer_memory: 100 * 1024 * 1024, // 100MB
        }
    }
}

/// Streaming system metrics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamingMetrics {
    pub active_subscriptions: usize,
    pub events_published: u64,
    pub events_buffered: usize,
    pub buffer_memory_usage: usize,
    pub uptime_seconds: u64,
}

/// Event buffer for storing recent events
#[derive(Debug)]
pub struct EventBuffer {
    events: Vec<TimestampedEvent>,
    max_size: usize,
    memory_usage: usize,
}

impl EventBuffer {
    fn new(max_size: usize) -> Self {
        Self {
            events: Vec::with_capacity(max_size),
            max_size,
            memory_usage: 0,
        }
    }

    async fn add_event(&mut self, event: TradingEvent) -> Result<()> {
        let timestamped = TimestampedEvent {
            event,
            stored_at: Utc::now(),
        };

        // Estimate memory usage (rough approximation)
        let event_size =
            std::mem::size_of::<TimestampedEvent>() + timestamped.event.estimated_size();

        // Remove old events if buffer is full
        while self.events.len() >= self.max_size {
            let removed = self.events.remove(0);
            self.memory_usage = self.memory_usage.saturating_sub(
                std::mem::size_of::<TimestampedEvent>() + removed.event.estimated_size(),
            );
        }

        self.events.push(timestamped);
        self.memory_usage += event_size;

        Ok(())
    }

    fn cleanup_expired_events(&mut self) {
        let now = Utc::now();
        let max_age = chrono::TimeDelta::seconds(3600); // 1 hour

        self.events.retain(|event| {
            let age = now - event.stored_at;
            if age <= max_age {
                true
            } else {
                self.memory_usage = self.memory_usage.saturating_sub(
                    std::mem::size_of::<TimestampedEvent>() + event.event.estimated_size(),
                );
                false
            }
        });
    }

    fn get_filtered_events(&self, filter: EventFilter, limit: Option<usize>) -> Vec<TradingEvent> {
        let mut filtered: Vec<TradingEvent> = self
            .events
            .iter()
            .filter(|timestamped| filter.matches(&timestamped.event))
            .map(|timestamped| timestamped.event.clone())
            .collect();

        // Sort by timestamp (newest first)
        filtered.sort_by(|a, b| b.timestamp.cmp(&a.timestamp));

        if let Some(limit) = limit {
            filtered.truncate(limit);
        }

        filtered
    }

    fn len(&self) -> usize {
        self.events.len()
    }

    fn memory_usage(&self) -> usize {
        self.memory_usage
    }

    fn clear(&mut self) {
        self.events.clear();
        self.memory_usage = 0;
    }
}

/// Event with storage timestamp
#[derive(Debug, Clone)]
struct TimestampedEvent {
    event: TradingEvent,
    stored_at: DateTime<Utc>,
}

/// Subscription manager for handling client subscriptions
#[derive(Debug)]
pub struct SubscriptionManager {
    subscriptions: HashMap<String, EventFilter>,
}

impl SubscriptionManager {
    fn new() -> Self {
        Self {
            subscriptions: HashMap::new(),
        }
    }

    fn add_subscription(&mut self, id: String, filter: EventFilter) {
        self.subscriptions.insert(id, filter);
    }

    fn remove_subscription(&mut self, id: &str) {
        self.subscriptions.remove(id);
    }

    fn subscription_count(&self) -> usize {
        self.subscriptions.len()
    }

    fn clear_all_subscriptions(&mut self) {
        self.subscriptions.clear();
    }
}

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

    #[tokio::test]
    async fn test_event_streamer_creation() {
        let config = EventStreamingConfig::default();
        let streamer = TradingEventStreamer::new(config);

        assert!(streamer.start().await.is_ok());
        assert!(streamer.shutdown().await.is_ok());
    }

    #[tokio::test]
    async fn test_event_publishing() {
        let config = EventStreamingConfig::default();
        let streamer = TradingEventStreamer::new(config);

        streamer.start().await.unwrap();

        let event = TradingEvent::new(
            TradingEventType::OrderSubmitted,
            "test_order".to_string(),
            "Test order submission".to_string(),
        );

        assert!(streamer.publish_event(event).await.is_ok());

        let metrics = streamer.get_metrics().await;
        assert_eq!(metrics.events_published, 1);
        assert_eq!(metrics.events_buffered, 1);

        streamer.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn test_subscription_management() {
        let config = EventStreamingConfig::default();
        let streamer = TradingEventStreamer::new(config);

        streamer.start().await.unwrap();

        let filter = EventFilter::for_event_types(vec![TradingEventType::OrderSubmitted]);
        let _subscription = streamer.subscribe(filter).await.unwrap();

        let metrics = streamer.get_metrics().await;
        assert_eq!(metrics.active_subscriptions, 1);

        streamer.shutdown().await.unwrap();
    }

    #[test]
    fn test_event_buffer() {
        let mut buffer = EventBuffer::new(3);

        let event1 = TradingEvent::new(
            TradingEventType::OrderSubmitted,
            "order1".to_string(),
            "First order".to_string(),
        );

        let event2 = TradingEvent::new(
            TradingEventType::OrderFilled,
            "order1".to_string(),
            "Order filled".to_string(),
        );

        tokio::runtime::Runtime::new().unwrap().block_on(async {
            buffer.add_event(event1).await.unwrap();
            buffer.add_event(event2).await.unwrap();

            assert_eq!(buffer.len(), 2);
            assert!(buffer.memory_usage() > 0);
        });
    }

    #[test]
    fn test_subscription_manager() {
        let mut manager = SubscriptionManager::new();

        let filter = EventFilter::for_event_types(vec![TradingEventType::OrderSubmitted]);
        manager.add_subscription("sub1".to_string(), filter);

        assert_eq!(manager.subscription_count(), 1);

        manager.remove_subscription("sub1");
        assert_eq!(manager.subscription_count(), 0);
    }
}
