//! Real-time streaming of historical market data
//!
//! Provides time-accurate replay of historical market data for backtesting
//! and simulation. Maintains the original timing relationships between events
//! while allowing speed multipliers for faster simulation.
//!
//! # Features
//!
//! - **Time-accurate replay**: Preserves original inter-event delays
//! - **Speed control**: 1x, 10x, 100x or custom speed multipliers
//! - **High throughput**: Async channel-based streaming
//! - **Backpressure handling**: Bounded channels prevent memory exhaustion
//!
//! # Examples
//!
//! ## Real-time replay (1x speed)
//! ```no_run
//! use data::replay::{ParquetDataLoader, MarketDataStreamer};
//!
//! # async fn example() -> anyhow::Result<()> {
//! let loader = ParquetDataLoader::new("market_data.parquet");
//! let events = loader.load_all().await?;
//!
//! let streamer = MarketDataStreamer::new(events);
//! let mut rx = streamer.stream().await;
//!
//! while let Some(event) = rx.recv().await {
//!     println!("Event at {}: {:?}", event.timestamp_ns, event.event_type);
//! }
//! # Ok(())
//! # }
//! ```
//!
//! ## Fast replay (10x speed)
//! ```no_run
//! use data::replay::{ParquetDataLoader, MarketDataStreamer};
//!
//! # async fn example() -> anyhow::Result<()> {
//! let loader = ParquetDataLoader::new("market_data.parquet");
//! let events = loader.load_all().await?;
//!
//! let streamer = MarketDataStreamer::new(events)
//!     .with_speed(10.0)           // 10x faster
//!     .with_buffer_size(10000);   // Larger buffer for high throughput
//!
//! let mut rx = streamer.stream().await;
//!
//! while let Some(event) = rx.recv().await {
//!     // Process events at 10x speed
//! }
//! # Ok(())
//! # }
//! ```
//!
//! ## Instant replay (no delays)
//! ```no_run
//! use data::replay::{ParquetDataLoader, MarketDataStreamer};
//!
//! # async fn example() -> anyhow::Result<()> {
//! let loader = ParquetDataLoader::new("market_data.parquet");
//! let events = loader.load_all().await?;
//!
//! // Speed = f64::INFINITY means no delays
//! let streamer = MarketDataStreamer::new(events)
//!     .with_speed(f64::INFINITY);
//!
//! let mut rx = streamer.stream().await;
//! // Events streamed as fast as possible
//! # Ok(())
//! # }
//! ```

use std::time::Duration;
use tokio::sync::mpsc;
use trading_engine::types::metrics::ParquetMarketDataEvent;

/// Market data streamer for time-accurate replay
///
/// Streams historical market data events while preserving the original
/// timing relationships. Supports speed multipliers for faster simulation
/// and configurable channel buffer sizes for backpressure control.
#[derive(Debug)]
pub struct MarketDataStreamer {
    events: Vec<ParquetMarketDataEvent>,
    replay_speed: f64,
    buffer_size: usize,
}

impl MarketDataStreamer {
    /// Create new streamer with events
    ///
    /// Default configuration:
    /// - Speed: 1.0 (real-time)
    /// - Buffer size: 1000 events
    ///
    /// # Arguments
    /// * `events` - Vector of market data events to stream (must be sorted by timestamp)
    ///
    /// # Examples
    /// ```
    /// use data::replay::MarketDataStreamer;
    ///
    /// let events = vec![]; // Your events here
    /// let streamer = MarketDataStreamer::new(events);
    /// ```
    pub fn new(events: Vec<ParquetMarketDataEvent>) -> Self {
        Self {
            events,
            replay_speed: 1.0,
            buffer_size: 1000,
        }
    }

    /// Set replay speed multiplier
    ///
    /// - 1.0 = real-time (original timing)
    /// - 2.0 = 2x faster
    /// - 0.5 = half speed (slow motion)
    /// - f64::INFINITY = instant (no delays)
    ///
    /// # Arguments
    /// * `speed` - Speed multiplier (must be > 0.0)
    ///
    /// # Panics
    /// Panics if speed <= 0.0 or is NaN
    ///
    /// # Examples
    /// ```
    /// use data::replay::MarketDataStreamer;
    ///
    /// let events = vec![];
    /// let streamer = MarketDataStreamer::new(events)
    ///     .with_speed(10.0); // 10x faster
    /// ```
    pub fn with_speed(mut self, speed: f64) -> Self {
        assert!(speed > 0.0 && !speed.is_nan(), "Speed must be positive");
        self.replay_speed = speed;
        self
    }

    /// Set channel buffer size
    ///
    /// Controls backpressure behavior:
    /// - Small buffer (100-1000): More memory efficient, may slow producer
    /// - Large buffer (10000+): Higher throughput, more memory usage
    ///
    /// # Arguments
    /// * `size` - Buffer size in events
    ///
    /// # Examples
    /// ```
    /// use data::replay::MarketDataStreamer;
    ///
    /// let events = vec![];
    /// let streamer = MarketDataStreamer::new(events)
    ///     .with_buffer_size(10000); // Large buffer for high throughput
    /// ```
    pub fn with_buffer_size(mut self, size: usize) -> Self {
        self.buffer_size = size;
        self
    }

    /// Stream events with time-accurate delays
    ///
    /// Spawns a background task that sends events through an async channel
    /// with delays calculated from timestamp differences and replay speed.
    ///
    /// # Returns
    /// Receiver side of the event channel
    ///
    /// # Timing Behavior
    /// - First event: Sent immediately
    /// - Subsequent events: Delayed by (timestamp_diff / replay_speed)
    /// - Speed = f64::INFINITY: No delays (instant replay)
    ///
    /// # Backpressure
    /// If receiver is slow, producer will block when channel buffer fills.
    /// This prevents unbounded memory growth.
    ///
    /// # Examples
    /// ```no_run
    /// # use data::replay::MarketDataStreamer;
    /// # async fn example() -> anyhow::Result<()> {
    /// let events = vec![]; // Your events
    /// let streamer = MarketDataStreamer::new(events);
    /// let mut rx = streamer.stream().await;
    ///
    /// while let Some(event) = rx.recv().await {
    ///     println!("Event: {:?}", event);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn stream(self) -> mpsc::Receiver<ParquetMarketDataEvent> {
        let (tx, rx) = mpsc::channel(self.buffer_size);

        tokio::spawn(async move {
            let mut prev_timestamp = None;

            for event in self.events {
                // Calculate delay based on timestamp difference
                if let Some(prev_ts) = prev_timestamp {
                    let delay_ns = event.timestamp_ns.saturating_sub(prev_ts);

                    // Only sleep if replay_speed is finite
                    if self.replay_speed.is_finite() {
                        let delay_secs = (delay_ns as f64 / 1_000_000_000.0) / self.replay_speed;

                        // Only sleep for meaningful delays (> 1 microsecond)
                        if delay_secs > 0.000_001 {
                            tokio::time::sleep(Duration::from_secs_f64(delay_secs)).await;
                        }
                    }
                    // If replay_speed is infinite, no sleep (instant replay)
                }

                prev_timestamp = Some(event.timestamp_ns);

                // Send event to channel
                if tx.send(event).await.is_err() {
                    // Receiver dropped, stop streaming
                    break;
                }
            }

            // Channel will close when tx is dropped here
        });

        rx
    }

    /// Get event count
    ///
    /// Returns the number of events that will be streamed.
    ///
    /// # Examples
    /// ```
    /// use data::replay::MarketDataStreamer;
    ///
    /// let events = vec![]; // Your events
    /// let streamer = MarketDataStreamer::new(events);
    /// println!("Will stream {} events", streamer.event_count());
    /// ```
    pub fn event_count(&self) -> usize {
        self.events.len()
    }

    /// Get time span covered by events
    ///
    /// Returns the time difference between first and last event in nanoseconds.
    /// Returns None if there are fewer than 2 events.
    ///
    /// # Examples
    /// ```
    /// use data::replay::MarketDataStreamer;
    ///
    /// let events = vec![]; // Your events (sorted by timestamp)
    /// let streamer = MarketDataStreamer::new(events);
    ///
    /// if let Some(span_ns) = streamer.time_span_ns() {
    ///     println!("Events span {} seconds", span_ns as f64 / 1e9);
    /// }
    /// ```
    pub fn time_span_ns(&self) -> Option<u64> {
        if self.events.len() < 2 {
            return None;
        }

        let first = self.events.first()?.timestamp_ns;
        let last = self.events.last()?.timestamp_ns;

        Some(last.saturating_sub(first))
    }

    /// Estimate replay duration
    ///
    /// Calculates how long the replay will take at current speed.
    /// Returns None if there are fewer than 2 events or speed is infinite.
    ///
    /// # Returns
    /// Estimated replay duration in seconds
    ///
    /// # Examples
    /// ```
    /// use data::replay::MarketDataStreamer;
    ///
    /// let events = vec![]; // Your events
    /// let streamer = MarketDataStreamer::new(events)
    ///     .with_speed(10.0);
    ///
    /// if let Some(duration_secs) = streamer.estimate_replay_duration_secs() {
    ///     println!("Replay will take ~{:.2} seconds", duration_secs);
    /// }
    /// ```
    pub fn estimate_replay_duration_secs(&self) -> Option<f64> {
        if !self.replay_speed.is_finite() {
            return None; // Infinite speed = instant replay
        }

        let span_ns = self.time_span_ns()?;
        Some((span_ns as f64 / 1_000_000_000.0) / self.replay_speed)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use trading_engine::types::metrics::MarketDataEventType;

    fn create_test_event(timestamp_ns: u64, symbol: &str) -> ParquetMarketDataEvent {
        ParquetMarketDataEvent {
            timestamp_ns,
            symbol: symbol.to_string(),
            venue: "test_venue".to_string(),
            event_type: MarketDataEventType::Trade,
            price: Some(100.0),
            quantity: Some(10.0),
            sequence: 0,
            latency_ns: None,
            open: None,
            high: None,
            low: None,
        }
    }

    #[test]
    fn test_market_data_streamer_creation() {
        let events = vec![];
        let streamer = MarketDataStreamer::new(events);
        assert_eq!(streamer.replay_speed, 1.0);
        assert_eq!(streamer.buffer_size, 1000);
    }

    #[test]
    fn test_with_speed() {
        let events = vec![];
        let streamer = MarketDataStreamer::new(events).with_speed(2.0);
        assert_eq!(streamer.replay_speed, 2.0);
    }

    #[test]
    fn test_with_buffer_size() {
        let events = vec![];
        let streamer = MarketDataStreamer::new(events).with_buffer_size(5000);
        assert_eq!(streamer.buffer_size, 5000);
    }

    #[test]
    #[should_panic(expected = "Speed must be positive")]
    fn test_invalid_speed_zero() {
        let events = vec![];
        let _ = MarketDataStreamer::new(events).with_speed(0.0);
    }

    #[test]
    #[should_panic(expected = "Speed must be positive")]
    fn test_invalid_speed_negative() {
        let events = vec![];
        let _ = MarketDataStreamer::new(events).with_speed(-1.0);
    }

    #[test]
    fn test_event_count() {
        let events = vec![
            create_test_event(1000, "BTC/USD"),
            create_test_event(2000, "BTC/USD"),
            create_test_event(3000, "BTC/USD"),
        ];
        let streamer = MarketDataStreamer::new(events);
        assert_eq!(streamer.event_count(), 3);
    }

    #[test]
    fn test_time_span_ns() {
        let events = vec![
            create_test_event(1000, "BTC/USD"),
            create_test_event(2000, "BTC/USD"),
            create_test_event(5000, "BTC/USD"),
        ];
        let streamer = MarketDataStreamer::new(events);
        assert_eq!(streamer.time_span_ns(), Some(4000));
    }

    #[test]
    fn test_time_span_ns_insufficient_events() {
        let events = vec![create_test_event(1000, "BTC/USD")];
        let streamer = MarketDataStreamer::new(events);
        assert_eq!(streamer.time_span_ns(), None);
    }

    #[test]
    fn test_estimate_replay_duration() {
        let events = vec![
            create_test_event(0, "BTC/USD"),
            create_test_event(10_000_000_000, "BTC/USD"), // 10 seconds later
        ];
        let streamer = MarketDataStreamer::new(events).with_speed(2.0);
        let duration = streamer.estimate_replay_duration_secs();
        assert_eq!(duration, Some(5.0)); // 10 seconds / 2.0 speed = 5 seconds
    }

    #[test]
    fn test_estimate_replay_duration_infinite_speed() {
        let events = vec![
            create_test_event(0, "BTC/USD"),
            create_test_event(10_000_000_000, "BTC/USD"),
        ];
        let streamer = MarketDataStreamer::new(events).with_speed(f64::INFINITY);
        assert_eq!(streamer.estimate_replay_duration_secs(), None);
    }

    #[tokio::test]
    async fn test_stream_basic() {
        let events = vec![
            create_test_event(1000, "BTC/USD"),
            create_test_event(2000, "ETH/USD"),
            create_test_event(3000, "SOL/USD"),
        ];

        let streamer = MarketDataStreamer::new(events).with_speed(f64::INFINITY); // Instant replay

        let mut rx = streamer.stream().await;

        let mut received = vec![];
        while let Some(event) = rx.recv().await {
            received.push(event);
        }

        assert_eq!(received.len(), 3);
        assert_eq!(received[0].symbol, "BTC/USD");
        assert_eq!(received[1].symbol, "ETH/USD");
        assert_eq!(received[2].symbol, "SOL/USD");
    }

    #[tokio::test]
    async fn test_stream_respects_timing() {
        let events = vec![
            create_test_event(0, "BTC/USD"),
            create_test_event(100_000_000, "BTC/USD"), // 100ms later
        ];

        let streamer = MarketDataStreamer::new(events).with_speed(1.0);

        let start = tokio::time::Instant::now();
        let mut rx = streamer.stream().await;

        let mut count = 0;
        while let Some(_event) = rx.recv().await {
            count += 1;
        }

        let elapsed = start.elapsed();
        assert_eq!(count, 2);
        // Should take at least 90ms (allowing some slack for timing)
        assert!(elapsed.as_millis() >= 90);
    }

    #[tokio::test]
    async fn test_stream_early_drop() {
        let events = vec![
            create_test_event(1000, "BTC/USD"),
            create_test_event(2000, "ETH/USD"),
            create_test_event(3000, "SOL/USD"),
        ];

        let streamer = MarketDataStreamer::new(events).with_speed(f64::INFINITY);

        let mut rx = streamer.stream().await;

        // Receive only first event, then drop receiver
        let first = rx.recv().await;
        assert!(first.is_some());
        assert_eq!(first.unwrap().symbol, "BTC/USD");

        drop(rx); // Producer should detect and stop
                  // Test passes if no panic/hang occurs
    }
}
