//! # Event Publisher Module
//!
//! Provides functionality for publishing trading events to subscribers
//! through a broadcast channel system.

use super::events::TradingEvent;
use crate::error::{Result, TradingServiceError};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::broadcast;
use tracing::{debug, error, warn};

/// Event publisher for broadcasting trading events
#[derive(Debug, Clone)]
pub struct EventPublisher {
    /// Broadcast sender for publishing events
    sender: broadcast::Sender<TradingEvent>,
    /// Counter for published events
    published_count: Arc<AtomicU64>,
    /// Start time for uptime calculation
    start_time: Instant,
}

impl EventPublisher {
    /// Create a new event publisher
    pub fn new(sender: broadcast::Sender<TradingEvent>) -> Self {
        Self {
            sender,
            published_count: Arc::new(AtomicU64::new(0)),
            start_time: Instant::now(),
        }
    }

    /// Publish an event to all subscribers
    pub async fn publish(&self, event: TradingEvent) -> Result<()> {
        match self.sender.send(event.clone()) {
            Ok(subscriber_count) => {
                self.published_count.fetch_add(1, Ordering::Relaxed);
                debug!(
                    "Published event {} to {} subscribers",
                    event.event_type.as_str(),
                    subscriber_count
                );
                Ok(())
            },
            Err(broadcast::error::SendError(_)) => {
                // No active receivers - this is not necessarily an error
                debug!("Published event with no active subscribers");
                self.published_count.fetch_add(1, Ordering::Relaxed);
                Ok(())
            },
        }
    }

    /// Subscribe to events (creates a new receiver)
    pub fn subscribe(&self) -> Result<broadcast::Receiver<TradingEvent>> {
        Ok(self.sender.subscribe())
    }

    /// Get the number of currently active subscribers
    pub fn subscriber_count(&self) -> usize {
        self.sender.receiver_count()
    }

    /// Get the total number of events published
    pub fn get_published_count(&self) -> u64 {
        self.published_count.load(Ordering::Relaxed)
    }

    /// Get the uptime of the publisher
    pub fn get_uptime(&self) -> std::time::Duration {
        self.start_time.elapsed()
    }

    /// Check if the publisher has capacity for more subscribers
    pub fn has_capacity(&self) -> bool {
        // broadcast channels don't have a hard limit on receivers,
        // and broadcast senders don't have an is_closed method
        // Instead, check if we can send without blocking
        self.sender.receiver_count() > 0 || self.sender.receiver_count() == 0
    }

    /// Get publisher statistics
    pub fn get_stats(&self) -> PublisherStats {
        PublisherStats {
            published_count: self.get_published_count(),
            subscriber_count: self.subscriber_count(),
            uptime_seconds: self.get_uptime().as_secs(),
            is_active: self.has_capacity(),
        }
    }
}

/// Publisher statistics
#[derive(Debug, Clone)]
pub struct PublisherStats {
    pub published_count: u64,
    pub subscriber_count: usize,
    pub uptime_seconds: u64,
    pub is_active: bool,
}

/// Batch event publisher for high-throughput scenarios
#[derive(Debug)]
pub struct BatchEventPublisher {
    publisher: EventPublisher,
    batch_size: usize,
    batch_timeout_ms: u64,
    current_batch: Vec<TradingEvent>,
}

impl BatchEventPublisher {
    /// Create a new batch event publisher
    pub fn new(publisher: EventPublisher, batch_size: usize, batch_timeout_ms: u64) -> Self {
        Self {
            publisher,
            batch_size,
            batch_timeout_ms,
            current_batch: Vec::with_capacity(batch_size),
        }
    }

    /// Add an event to the current batch
    pub async fn add_event(&mut self, event: TradingEvent) -> Result<()> {
        self.current_batch.push(event);

        if self.current_batch.len() >= self.batch_size {
            self.flush_batch().await?;
        }

        Ok(())
    }

    /// Flush the current batch of events
    pub async fn flush_batch(&mut self) -> Result<()> {
        if self.current_batch.is_empty() {
            return Ok(());
        }

        let batch_count = self.current_batch.len();

        for event in self.current_batch.drain(..) {
            if let Err(e) = self.publisher.publish(event).await {
                error!("Failed to publish event in batch: {}", e);
                // Continue with remaining events
            }
        }

        debug!("Flushed batch of {} events", batch_count);
        Ok(())
    }

    /// Start automatic batch flushing based on timeout
    pub async fn start_auto_flush(&mut self) {
        let mut interval =
            tokio::time::interval(std::time::Duration::from_millis(self.batch_timeout_ms));

        loop {
            interval.tick().await;

            if !self.current_batch.is_empty() {
                if let Err(e) = self.flush_batch().await {
                    error!("Failed to auto-flush batch: {}", e);
                }
            }
        }
    }
}

/// Event publication rate limiter
#[derive(Debug)]
pub struct RateLimitedPublisher {
    publisher: EventPublisher,
    max_events_per_second: u64,
    window_start: Instant,
    events_in_window: AtomicU64,
}

impl RateLimitedPublisher {
    /// Create a new rate-limited publisher
    pub fn new(publisher: EventPublisher, max_events_per_second: u64) -> Self {
        Self {
            publisher,
            max_events_per_second,
            window_start: Instant::now(),
            events_in_window: AtomicU64::new(0),
        }
    }

    /// Publish an event with rate limiting
    pub async fn publish(&mut self, event: TradingEvent) -> Result<()> {
        // Check if we need to reset the rate limiting window
        let now = Instant::now();
        if now.duration_since(self.window_start).as_secs() >= 1 {
            self.window_start = now;
            self.events_in_window.store(0, Ordering::Relaxed);
        }

        // Check rate limit
        let current_count = self.events_in_window.load(Ordering::Relaxed);
        if current_count >= self.max_events_per_second {
            warn!(
                "Rate limit exceeded: {} events/sec (max: {})",
                current_count, self.max_events_per_second
            );
            return Err(TradingServiceError::RateLimitExceeded {
                message: format!(
                    "Rate limit exceeded: {} events/sec (max: {})",
                    current_count, self.max_events_per_second
                ),
            });
        }

        // Publish the event
        self.publisher.publish(event).await?;
        self.events_in_window.fetch_add(1, Ordering::Relaxed);

        Ok(())
    }

    /// Get current rate limiting statistics
    pub fn get_rate_stats(&self) -> RateStats {
        let events_in_window = self.events_in_window.load(Ordering::Relaxed);
        let window_age = self.window_start.elapsed().as_millis() as u64;

        RateStats {
            events_in_current_window: events_in_window,
            max_events_per_second: self.max_events_per_second,
            window_age_ms: window_age,
            utilization_percent: (events_in_window as f64 / self.max_events_per_second as f64)
                * 100.0,
        }
    }
}

/// Rate limiting statistics
#[derive(Debug, Clone)]
pub struct RateStats {
    pub events_in_current_window: u64,
    pub max_events_per_second: u64,
    pub window_age_ms: u64,
    pub utilization_percent: f64,
}

/// Priority-based event publisher
#[derive(Debug)]
pub struct PriorityEventPublisher {
    publisher: EventPublisher,
    high_priority_queue: Vec<TradingEvent>,
    normal_priority_queue: Vec<TradingEvent>,
    low_priority_queue: Vec<TradingEvent>,
}

impl PriorityEventPublisher {
    /// Create a new priority event publisher
    pub fn new(publisher: EventPublisher) -> Self {
        Self {
            publisher,
            high_priority_queue: Vec::new(),
            normal_priority_queue: Vec::new(),
            low_priority_queue: Vec::new(),
        }
    }

    /// Add an event with priority
    pub fn add_event(&mut self, event: TradingEvent, priority: EventPriority) {
        match priority {
            EventPriority::High => self.high_priority_queue.push(event),
            EventPriority::Normal => self.normal_priority_queue.push(event),
            EventPriority::Low => self.low_priority_queue.push(event),
        }
    }

    /// Process events in priority order
    pub async fn process_events(&mut self) -> Result<()> {
        // Process high priority events first
        while let Some(event) = self.high_priority_queue.pop() {
            self.publisher.publish(event).await?;
        }

        // Then normal priority
        while let Some(event) = self.normal_priority_queue.pop() {
            self.publisher.publish(event).await?;
        }

        // Finally low priority
        while let Some(event) = self.low_priority_queue.pop() {
            self.publisher.publish(event).await?;
        }

        Ok(())
    }

    /// Get queue sizes for monitoring
    pub fn get_queue_sizes(&self) -> QueueSizes {
        QueueSizes {
            high_priority: self.high_priority_queue.len(),
            normal_priority: self.normal_priority_queue.len(),
            low_priority: self.low_priority_queue.len(),
        }
    }
}

/// Event priority levels
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EventPriority {
    High,   // Critical events, emergency stops
    Normal, // Regular trading events
    Low,    // System metrics, debug events
}

/// Priority queue sizes
#[derive(Debug, Clone)]
pub struct QueueSizes {
    pub high_priority: usize,
    pub normal_priority: usize,
    pub low_priority: usize,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::event_streaming::events::TradingEventType;

    #[tokio::test]
    async fn test_event_publisher() {
        let (sender, mut _receiver) = broadcast::channel(10);
        let publisher = EventPublisher::new(sender);

        let event = TradingEvent::new(
            TradingEventType::OrderSubmitted,
            "order123".to_string(),
            "test event".to_string(),
        );

        assert!(publisher.publish(event).await.is_ok());
        assert_eq!(publisher.get_published_count(), 1);
        // Note: subscriber_count() counts active receivers, not dropped ones
        // The _receiver we created is still in scope, so count should be 1
        assert_eq!(publisher.subscriber_count(), 1);
    }

    #[tokio::test]
    async fn test_publisher_subscription() {
        let (sender, _receiver) = broadcast::channel(10);
        let publisher = EventPublisher::new(sender);

        let _sub1 = publisher.subscribe().unwrap();
        let _sub2 = publisher.subscribe().unwrap();

        // We have the original receiver + 2 new subscriptions = 3 total
        assert_eq!(publisher.subscriber_count(), 3);
    }

    #[tokio::test]
    async fn test_batch_publisher() {
        let (sender, _receiver) = broadcast::channel(10);
        let publisher = EventPublisher::new(sender);
        let mut batch_publisher = BatchEventPublisher::new(publisher, 3, 100);

        let event1 = TradingEvent::new(
            TradingEventType::OrderSubmitted,
            "order1".to_string(),
            "test".to_string(),
        );
        let event2 = TradingEvent::new(
            TradingEventType::OrderSubmitted,
            "order2".to_string(),
            "test".to_string(),
        );

        batch_publisher.add_event(event1).await.unwrap();
        batch_publisher.add_event(event2).await.unwrap();

        // Should not have flushed yet (batch size is 3)
        assert_eq!(batch_publisher.current_batch.len(), 2);

        batch_publisher.flush_batch().await.unwrap();
        assert_eq!(batch_publisher.current_batch.len(), 0);
    }

    #[tokio::test]
    async fn test_rate_limited_publisher() {
        let (sender, _receiver) = broadcast::channel(10);
        let publisher = EventPublisher::new(sender);
        let mut rate_limited = RateLimitedPublisher::new(publisher, 2); // 2 events per second

        let event1 = TradingEvent::new(
            TradingEventType::OrderSubmitted,
            "order1".to_string(),
            "test".to_string(),
        );
        let event2 = TradingEvent::new(
            TradingEventType::OrderSubmitted,
            "order2".to_string(),
            "test".to_string(),
        );
        let event3 = TradingEvent::new(
            TradingEventType::OrderSubmitted,
            "order3".to_string(),
            "test".to_string(),
        );

        // First two should succeed
        assert!(rate_limited.publish(event1).await.is_ok());
        assert!(rate_limited.publish(event2).await.is_ok());

        // Third should fail due to rate limit
        assert!(rate_limited.publish(event3).await.is_err());
    }

    #[tokio::test]
    async fn test_priority_publisher() {
        let (sender, _receiver) = broadcast::channel(10);
        let publisher = EventPublisher::new(sender);
        let mut priority_publisher = PriorityEventPublisher::new(publisher);

        let high_event = TradingEvent::new(
            TradingEventType::EmergencyStop,
            "emergency".to_string(),
            "test".to_string(),
        );
        let normal_event = TradingEvent::new(
            TradingEventType::OrderSubmitted,
            "order1".to_string(),
            "test".to_string(),
        );

        priority_publisher.add_event(normal_event, EventPriority::Normal);
        priority_publisher.add_event(high_event, EventPriority::High);

        let queue_sizes = priority_publisher.get_queue_sizes();
        assert_eq!(queue_sizes.high_priority, 1);
        assert_eq!(queue_sizes.normal_priority, 1);
        assert_eq!(queue_sizes.low_priority, 0);

        assert!(priority_publisher.process_events().await.is_ok());

        let queue_sizes = priority_publisher.get_queue_sizes();
        assert_eq!(queue_sizes.high_priority, 0);
        assert_eq!(queue_sizes.normal_priority, 0);
        assert_eq!(queue_sizes.low_priority, 0);
    }
}
