foxhunt/services/trading_service/src/event_streaming/subscriber.rs

//! # Event Subscriber Module
//!
//! Provides functionality for subscribing to trading events with filtering
//! and processing capabilities.

use super::events::TradingEvent;
use super::filters::EventFilter;
use crate::error::{Result, TradingServiceError};
use std::time::Duration;
use tokio::sync::broadcast;
use tokio::time::timeout;
use tracing::{debug, error, warn};

/// Trading event receiver with filtering capabilities
#[derive(Debug)]
pub struct TradingEventReceiver {
    /// Unique subscription ID
    pub subscription_id: String,
    /// Broadcast receiver for events
    pub receiver: broadcast::Receiver<TradingEvent>,
    /// Event filter for this subscription
    pub filter: EventFilter,
    /// Statistics for this receiver
    stats: ReceiverStats,
}

impl TradingEventReceiver {
    /// Create a new trading event receiver
    pub fn new(
        subscription_id: String,
        receiver: broadcast::Receiver<TradingEvent>,
        filter: EventFilter,
    ) -> Self {
        Self {
            subscription_id,
            receiver,
            filter,
            stats: ReceiverStats::new(),
        }
    }

    /// Receive the next filtered event
    pub async fn recv(&mut self) -> Option<TradingEvent> {
        loop {
            match self.receiver.recv().await {
                Ok(event) => {
                    self.stats.events_received += 1;

                    if self.filter.matches(&event) {
                        self.stats.events_matched += 1;
                        debug!(
                            "Subscription {} received matching event: {}",
                            self.subscription_id,
                            event.event_type.as_str()
                        );
                        return Some(event);
                    } else {
                        self.stats.events_filtered += 1;
                    }
                },
                Err(broadcast::error::RecvError::Lagged(skipped)) => {
                    self.stats.events_lagged += skipped;
                    warn!(
                        "Subscription {} lagged, skipped {} events",
                        self.subscription_id, skipped
                    );
                    // Continue receiving
                },
                Err(broadcast::error::RecvError::Closed) => {
                    debug!("Subscription {} channel closed", self.subscription_id);
                    return None;
                },
            }
        }
    }

    /// Receive the next event with a timeout
    pub async fn recv_timeout(&mut self, duration: Duration) -> Result<Option<TradingEvent>> {
        match timeout(duration, self.recv()).await {
            Ok(event) => Ok(event),
            Err(_) => Err(TradingServiceError::SubscriptionTimeout {
                message: format!(
                    "Subscription {} timed out after {}ms",
                    self.subscription_id,
                    duration.as_millis()
                ),
            }),
        }
    }

    /// Try to receive an event without blocking
    pub fn try_recv(&mut self) -> Result<Option<TradingEvent>> {
        loop {
            match self.receiver.try_recv() {
                Ok(event) => {
                    self.stats.events_received += 1;

                    if self.filter.matches(&event) {
                        self.stats.events_matched += 1;
                        return Ok(Some(event));
                    } else {
                        self.stats.events_filtered += 1;
                        // Continue to next event
                    }
                },
                Err(broadcast::error::TryRecvError::Lagged(skipped)) => {
                    self.stats.events_lagged += skipped;
                    warn!(
                        "Subscription {} lagged, skipped {} events",
                        self.subscription_id, skipped
                    );
                    // Continue receiving
                },
                Err(broadcast::error::TryRecvError::Empty) => {
                    return Ok(None);
                },
                Err(broadcast::error::TryRecvError::Closed) => {
                    return Err(TradingServiceError::SubscriptionClosed {
                        message: format!("Subscription {} closed", self.subscription_id),
                    });
                },
            }
        }
    }

    /// Update the event filter for this subscription
    pub fn update_filter(&mut self, new_filter: EventFilter) {
        self.filter = new_filter;
        debug!("Updated filter for subscription {}", self.subscription_id);
    }

    /// Get subscription statistics
    pub fn get_stats(&self) -> &ReceiverStats {
        &self.stats
    }

    /// Get subscription ID
    pub fn subscription_id(&self) -> &str {
        &self.subscription_id
    }

    /// Check if the subscription is still active
    pub fn is_active(&self) -> bool {
        !self.receiver.is_closed()
    }
}

/// Statistics for event receivers
#[derive(Debug, Clone)]
pub struct ReceiverStats {
    pub events_received: u64,
    pub events_matched: u64,
    pub events_filtered: u64,
    pub events_lagged: u64,
    pub created_at: chrono::DateTime<chrono::Utc>,
}

impl ReceiverStats {
    fn new() -> Self {
        Self {
            events_received: 0,
            events_matched: 0,
            events_filtered: 0,
            events_lagged: 0,
            created_at: chrono::Utc::now(),
        }
    }

    /// Calculate match rate percentage
    pub fn match_rate(&self) -> f64 {
        if self.events_received > 0 {
            (self.events_matched as f64 / self.events_received as f64) * 100.0
        } else {
            0.0
        }
    }

    /// Calculate filter rate percentage
    pub fn filter_rate(&self) -> f64 {
        if self.events_received > 0 {
            (self.events_filtered as f64 / self.events_received as f64) * 100.0
        } else {
            0.0
        }
    }

    /// Get subscription age
    pub fn age(&self) -> chrono::TimeDelta {
        chrono::Utc::now() - self.created_at
    }
}

/// Event processor for handling received events
#[derive(Debug)]
pub struct EventProcessor<F>
where
    F: Fn(TradingEvent) -> Result<()> + Send + Sync,
{
    receiver: TradingEventReceiver,
    processor_fn: F,
    batch_size: usize,
    process_timeout: Duration,
}

impl<F> EventProcessor<F>
where
    F: Fn(TradingEvent) -> Result<()> + Send + Sync,
{
    /// Create a new event processor
    pub fn new(
        receiver: TradingEventReceiver,
        processor_fn: F,
        batch_size: usize,
        process_timeout: Duration,
    ) -> Self {
        Self {
            receiver,
            processor_fn,
            batch_size,
            process_timeout,
        }
    }

    /// Process events continuously
    pub async fn run(&mut self) -> Result<()> {
        let mut batch = Vec::with_capacity(self.batch_size);

        loop {
            // Try to receive an event with timeout
            match self.receiver.recv_timeout(self.process_timeout).await {
                Ok(Some(event)) => {
                    batch.push(event);

                    // Process batch when full
                    if batch.len() >= self.batch_size {
                        self.process_batch(&mut batch).await?;
                    }
                },
                Ok(None) => {
                    // Channel closed, process remaining events and exit
                    if !batch.is_empty() {
                        self.process_batch(&mut batch).await?;
                    }
                    break;
                },
                Err(TradingServiceError::SubscriptionTimeout { .. }) => {
                    // Timeout occurred, process any pending events
                    if !batch.is_empty() {
                        self.process_batch(&mut batch).await?;
                    }
                    // Continue receiving
                },
                Err(e) => {
                    error!("Error receiving events: {}", e);
                    return Err(e);
                },
            }
        }

        Ok(())
    }

    /// Process a batch of events
    async fn process_batch(&self, batch: &mut Vec<TradingEvent>) -> Result<()> {
        for event in batch.drain(..) {
            if let Err(e) = (self.processor_fn)(event) {
                error!("Error processing event: {}", e);
                // Continue processing remaining events
            }
        }

        Ok(())
    }

    /// Get the underlying receiver
    pub fn receiver(&self) -> &TradingEventReceiver {
        &self.receiver
    }

    /// Get processor statistics
    pub fn get_stats(&self) -> &ReceiverStats {
        self.receiver.get_stats()
    }
}

/// Multi-subscription manager for handling multiple event streams
#[derive(Debug)]
pub struct MultiSubscriptionManager {
    receivers: Vec<TradingEventReceiver>,
    next_index: usize,
}

impl MultiSubscriptionManager {
    /// Create a new multi-subscription manager
    pub fn new() -> Self {
        Self {
            receivers: Vec::new(),
            next_index: 0,
        }
    }

    /// Add a new subscription
    pub fn add_subscription(&mut self, receiver: TradingEventReceiver) {
        self.receivers.push(receiver);
    }

    /// Remove a subscription by ID
    pub fn remove_subscription(&mut self, subscription_id: &str) -> bool {
        if let Some(pos) = self
            .receivers
            .iter()
            .position(|r| r.subscription_id == subscription_id)
        {
            self.receivers.remove(pos);
            // Adjust next_index if necessary
            if self.next_index >= self.receivers.len() && !self.receivers.is_empty() {
                self.next_index = 0;
            }
            true
        } else {
            false
        }
    }

    /// Receive the next event from any subscription (round-robin)
    pub async fn recv_any(&mut self) -> Option<(String, TradingEvent)> {
        if self.receivers.is_empty() {
            return None;
        }

        let start_index = self.next_index;

        loop {
            let receiver = &mut self.receivers[self.next_index];

            // Try to receive without blocking
            match receiver.try_recv() {
                Ok(Some(event)) => {
                    let subscription_id = receiver.subscription_id.clone();
                    self.advance_next_index();
                    return Some((subscription_id, event));
                },
                Ok(None) => {
                    // No event available, try next receiver
                    self.advance_next_index();
                },
                Err(_) => {
                    // Error receiving, remove this subscription
                    let subscription_id = receiver.subscription_id.clone();
                    warn!("Removing failed subscription: {}", subscription_id);
                    self.receivers.remove(self.next_index);

                    if self.receivers.is_empty() {
                        return None;
                    }

                    if self.next_index >= self.receivers.len() {
                        self.next_index = 0;
                    }
                },
            }

            // If we've checked all receivers, wait a bit and try again
            if self.next_index == start_index {
                tokio::time::sleep(Duration::from_millis(1)).await;
            }
        }
    }

    /// Get statistics for all subscriptions
    pub fn get_all_stats(&self) -> Vec<(String, ReceiverStats)> {
        self.receivers
            .iter()
            .map(|r| (r.subscription_id.clone(), r.stats.clone()))
            .collect()
    }

    /// Get number of active subscriptions
    pub fn subscription_count(&self) -> usize {
        self.receivers.len()
    }

    /// Remove all inactive subscriptions
    pub fn cleanup_inactive(&mut self) {
        self.receivers.retain(|r| r.is_active());

        if self.next_index >= self.receivers.len() && !self.receivers.is_empty() {
            self.next_index = 0;
        }
    }

    fn advance_next_index(&mut self) {
        self.next_index = (self.next_index + 1) % self.receivers.len();
    }
}

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

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

    #[tokio::test]
    async fn test_event_receiver() {
        let (sender, receiver) = broadcast::channel(10);
        let filter = EventFilter::for_event_types(vec![TradingEventType::OrderSubmitted]);
        let mut event_receiver =
            TradingEventReceiver::new("test_sub".to_string(), receiver, filter);

        // Send a matching event
        let event = TradingEvent::new(
            TradingEventType::OrderSubmitted,
            "order123".to_string(),
            "test event".to_string(),
        );
        sender.send(event.clone()).unwrap();

        // Receive the event
        let received = event_receiver.recv().await;
        assert!(received.is_some());
        assert_eq!(
            received.unwrap().event_type,
            TradingEventType::OrderSubmitted
        );

        let stats = event_receiver.get_stats();
        assert_eq!(stats.events_received, 1);
        assert_eq!(stats.events_matched, 1);
        assert_eq!(stats.events_filtered, 0);
    }

    #[tokio::test]
    async fn test_event_filtering() {
        let (sender, receiver) = broadcast::channel(10);
        let filter = EventFilter::for_event_types(vec![TradingEventType::OrderSubmitted]);
        let mut event_receiver =
            TradingEventReceiver::new("test_sub".to_string(), receiver, filter);

        // Send a non-matching event
        let event1 = TradingEvent::new(
            TradingEventType::OrderFilled,
            "order123".to_string(),
            "fill event".to_string(),
        );
        sender.send(event1).expect("INVARIANT: Channel should not be closed");

        // Send a matching event
        let event2 = TradingEvent::new(
            TradingEventType::OrderSubmitted,
            "order456".to_string(),
            "submit event".to_string(),
        );
        sender.send(event2.clone()).unwrap();

        // Should receive only the matching event
        let received = event_receiver.recv().await;
        assert!(received.is_some());
        assert_eq!(
            received.unwrap().event_type,
            TradingEventType::OrderSubmitted
        );

        let stats = event_receiver.get_stats();
        assert_eq!(stats.events_received, 2);
        assert_eq!(stats.events_matched, 1);
        assert_eq!(stats.events_filtered, 1);
    }

    #[tokio::test]
    async fn test_multi_subscription_manager() {
        let (sender1, receiver1) = broadcast::channel(10);
        let (sender2, receiver2) = broadcast::channel(10);

        let filter1 = EventFilter::for_event_types(vec![TradingEventType::OrderSubmitted]);
        let filter2 = EventFilter::for_event_types(vec![TradingEventType::OrderFilled]);

        let receiver1 = TradingEventReceiver::new("sub1".to_string(), receiver1, filter1);
        let receiver2 = TradingEventReceiver::new("sub2".to_string(), receiver2, filter2);

        let mut manager = MultiSubscriptionManager::new();
        manager.add_subscription(receiver1);
        manager.add_subscription(receiver2);

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

        // Send events to both channels
        let event1 = TradingEvent::new(
            TradingEventType::OrderSubmitted,
            "order1".to_string(),
            "submit".to_string(),
        );
        sender1.send(event1).expect("INVARIANT: Channel should not be closed");

        let event2 = TradingEvent::new(
            TradingEventType::OrderFilled,
            "order2".to_string(),
            "fill".to_string(),
        );
        sender2.send(event2).expect("INVARIANT: Channel should not be closed");

        // Should be able to receive from both
        let (sub_id, _event) = manager.recv_any().await.unwrap();
        assert!(sub_id == "sub1" || sub_id == "sub2");
    }

    #[test]
    fn test_receiver_stats() {
        let stats = ReceiverStats::new();
        assert_eq!(stats.events_received, 0);
        assert_eq!(stats.match_rate(), 0.0);
        assert_eq!(stats.filter_rate(), 0.0);
    }
}