//! DBN-based Market Data Generator for E2E Testing
//!
//! This module provides a real market data event generator that reads from DBN files
//! and publishes authentic historical market data for testing purposes.
//!
//! Unlike TestMarketDataGenerator which generates mock data with synthetic prices,
//! this generator uses production-quality DBN (Databento Binary) files to provide
//! realistic market data streaming for API Gateway E2E tests.

use crate::event_streaming::events::{EventSeverity, TradingEvent, TradingEventType};
use crate::event_streaming::publisher::EventPublisher;
use anyhow::{Context, Result};
use chrono::{DateTime, TimeZone, Utc};
use dbn::decode::{DbnDecoder, DecodeRecordRef};
use dbn::{OhlcvMsg, VersionUpgradePolicy};
use serde_json::json;
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use tracing::{debug, info, warn};

/// DBN-based market data generator for E2E testing
///
/// Reads real historical market data from DBN files and publishes it as TradingEvents
/// to simulate live market data streaming through the API Gateway.
///
/// ## Features
///
/// - Real historical market data from DBN files
/// - Configurable playback speed (real-time, fast-forward, burst)
/// - Symbol mapping for multi-asset backtesting
/// - Automatic timestamp conversion
/// - Production-quality OHLCV bars
///
/// ## Usage
///
/// ```rust,no_run
/// use trading_service::dbn_market_data_generator::DbnMarketDataGenerator;
/// use std::collections::HashMap;
///
/// # async fn example() -> anyhow::Result<()> {
/// let mut file_mapping = HashMap::new();
/// file_mapping.insert("ES.FUT".to_string(),
///     "test_data/real/databento/ES.FUT_ohlcv-1m_2024-01-02.dbn".to_string());
///
/// let generator = DbnMarketDataGenerator::new(event_publisher, file_mapping).await?;
///
/// // Publish a burst of real market data
/// generator.publish_burst("ES.FUT", 10).await?;
/// # Ok(())
/// # }
/// ```
pub struct DbnMarketDataGenerator {
    event_publisher: Arc<EventPublisher>,
    file_mapping: HashMap<String, String>,
    running: Arc<tokio::sync::RwLock<bool>>,
}

impl DbnMarketDataGenerator {
    /// Create a new DBN-based market data generator
    ///
    /// # Arguments
    ///
    /// * `event_publisher` - Event publisher for broadcasting market data
    /// * `file_mapping` - Map of symbol to DBN file path
    ///
    /// # Returns
    ///
    /// Configured generator ready to publish real market data
    pub async fn new(
        event_publisher: Arc<EventPublisher>,
        file_mapping: HashMap<String, String>,
    ) -> Result<Self> {
        // Validate all files exist
        for (symbol, path) in &file_mapping {
            if !Path::new(path).exists() {
                warn!("DBN file not found for {}: {}", symbol, path);
            }
        }

        info!(
            "Created DBN market data generator with {} symbols",
            file_mapping.len()
        );

        Ok(Self {
            event_publisher,
            file_mapping,
            running: Arc::new(tokio::sync::RwLock::new(false)),
        })
    }

    /// Convert DBN fixed-point price to f64
    fn dbn_price_to_f64(price: i64) -> f64 {
        price as f64 / 1_000_000_000.0
    }

    /// Load OHLCV bars from a DBN file
    ///
    /// # Arguments
    ///
    /// * `symbol` - Trading symbol
    /// * `limit` - Maximum number of bars to load (None = all)
    ///
    /// # Returns
    ///
    /// Vector of OHLCV bars with timestamps
    async fn load_bars(&self, symbol: &str, limit: Option<usize>) -> Result<Vec<OhlcvBar>> {
        let file_path = self
            .file_mapping
            .get(symbol)
            .ok_or_else(|| anyhow::anyhow!("No DBN file configured for symbol: {}", symbol))?;

        if !Path::new(file_path).exists() {
            return Err(anyhow::anyhow!("DBN file not found: {}", file_path));
        }

        debug!("Loading DBN file: {} for symbol: {}", file_path, symbol);

        // Use official dbn crate decoder
        let mut decoder = DbnDecoder::from_file(file_path).context(format!(
            "Failed to create DBN decoder for file: {}",
            file_path
        ))?;

        let _ = decoder.set_upgrade_policy(VersionUpgradePolicy::UpgradeToV3);

        let mut bars = Vec::new();

        // Decode all records
        while let Some(record_ref) = decoder
            .decode_record_ref()
            .context("Failed to decode DBN record")?
        {
            if let Some(ohlcv) = record_ref.get::<OhlcvMsg>() {
                // Convert timestamp
                let ts_nanos = ohlcv.hd.ts_event as i64;
                let secs = ts_nanos / 1_000_000_000;
                let nanos = (ts_nanos % 1_000_000_000) as u32;
                let timestamp = Utc
                    .timestamp_opt(secs, nanos)
                    .single()
                    .ok_or_else(|| anyhow::anyhow!("Invalid timestamp: {}", ts_nanos))?;

                // Convert prices
                let open = Self::dbn_price_to_f64(ohlcv.open);
                let high = Self::dbn_price_to_f64(ohlcv.high);
                let low = Self::dbn_price_to_f64(ohlcv.low);
                let close = Self::dbn_price_to_f64(ohlcv.close);
                let volume = ohlcv.volume as f64;

                bars.push(OhlcvBar {
                    timestamp,
                    open,
                    high,
                    low,
                    close,
                    volume,
                });

                // Check limit
                if let Some(max) = limit {
                    if bars.len() >= max {
                        break;
                    }
                }
            }
        }

        info!(
            "Loaded {} bars from DBN file for symbol: {}",
            bars.len(),
            symbol
        );

        Ok(bars)
    }

    /// Publish a single market data burst for testing
    ///
    /// Loads bars from the DBN file and publishes them as TradingEvents.
    ///
    /// # Arguments
    ///
    /// * `symbol` - Trading symbol to publish data for
    /// * `count` - Number of bars to publish
    ///
    /// # Returns
    ///
    /// Result indicating success or failure
    pub async fn publish_burst(&self, symbol: &str, count: usize) -> Result<()> {
        info!(
            "Publishing {} real market data events for {} from DBN file",
            count, symbol
        );

        // Load bars from DBN file
        let bars = self.load_bars(symbol, Some(count)).await?;

        if bars.is_empty() {
            return Err(anyhow::anyhow!(
                "No bars loaded from DBN file for symbol: {}",
                symbol
            ));
        }

        info!("Publishing {} bars for {}", bars.len(), symbol);

        // Publish each bar as a TradingEvent
        for (i, bar) in bars.iter().enumerate() {
            let event = TradingEvent {
                id: uuid::Uuid::new_v4().to_string(),
                event_type: TradingEventType::PriceUpdate,
                timestamp: bar.timestamp,
                source: "dbn_market_data_generator".to_string(),
                correlation_id: Some(format!("dbn-{}-{}", symbol, i)),
                severity: EventSeverity::Info,
                payload: json!({
                    "symbol": symbol,
                    "price": bar.close,
                    "volume": bar.volume,
                    "timestamp": bar.timestamp.timestamp(),
                    "sequence": i,
                    "ohlcv": {
                        "open": bar.open,
                        "high": bar.high,
                        "low": bar.low,
                        "close": bar.close,
                        "volume": bar.volume,
                    }
                })
                .to_string(),
                metadata: std::collections::HashMap::new(),
            };

            self.event_publisher
                .publish(event)
                .await
                .map_err(|e| anyhow::anyhow!("Failed to publish event: {}", e))?;
        }

        info!(
            "Successfully published {} market data events for {}",
            bars.len(),
            symbol
        );

        Ok(())
    }

    /// Start generating market data events continuously
    ///
    /// Plays back DBN data in a loop with configurable interval between bars.
    ///
    /// # Arguments
    ///
    /// * `interval_ms` - Milliseconds between each bar publication
    pub async fn start(&self, interval_ms: u64) {
        let mut running = self.running.write().await;
        if *running {
            debug!("DBN market data generator already running");
            return;
        }
        *running = true;
        drop(running);

        info!(
            "Starting DBN market data generator with {}ms interval",
            interval_ms
        );

        let event_publisher = Arc::clone(&self.event_publisher);
        let file_mapping = self.file_mapping.clone();
        let running = Arc::clone(&self.running);

        tokio::spawn(async move {
            let mut _sequence = 0u64; // Prefix with _ to avoid unused warning
            let interval_duration = Duration::from_millis(interval_ms);

            loop {
                // Check if we should stop
                {
                    let is_running = running.read().await;
                    if !*is_running {
                        info!("DBN market data generator stopped");
                        break;
                    }
                }

                // Load and publish bars for all symbols
                for (symbol, file_path) in &file_mapping {
                    if !Path::new(file_path).exists() {
                        continue;
                    }

                    // Create a temporary generator to load bars
                    let temp_gen = Self {
                        event_publisher: Arc::clone(&event_publisher),
                        file_mapping: [(symbol.clone(), file_path.clone())]
                            .iter()
                            .cloned()
                            .collect(),
                        running: Arc::clone(&running),
                    };

                    // Publish one bar at a time
                    if let Err(e) = temp_gen.publish_burst(symbol, 1).await {
                        warn!("Failed to publish bar for {}: {}", symbol, e);
                    }
                }

                _sequence += 1;
                tokio::time::sleep(interval_duration).await;
            }
        });
    }

    /// Stop generating market data events
    pub async fn stop(&self) {
        let mut running = self.running.write().await;
        *running = false;
        info!("Stopping DBN market data generator");
    }

    /// Check if generator is running
    pub async fn is_running(&self) -> bool {
        *self.running.read().await
    }
}

/// OHLCV bar representation
#[derive(Debug, Clone)]
struct OhlcvBar {
    timestamp: DateTime<Utc>,
    open: f64,
    high: f64,
    low: f64,
    close: f64,
    volume: f64,
}

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

    #[tokio::test]
    async fn test_dbn_generator_creation() {
        let (sender, _) = broadcast::channel(1000);
        let publisher = Arc::new(EventPublisher::new(sender));

        let mut file_mapping = HashMap::new();
        file_mapping.insert(
            "ES.FUT".to_string(),
            "test_data/real/databento/ES.FUT_ohlcv-1m_2024-01-02.dbn".to_string(),
        );

        let generator = DbnMarketDataGenerator::new(publisher, file_mapping).await;
        assert!(generator.is_ok());
    }

    #[tokio::test]
    async fn test_publish_burst_real_data() {
        // Get workspace root
        let current_dir = std::env::current_dir().expect("INVARIANT: Current directory should be accessible");
        let workspace_root = current_dir
            .ancestors()
            .find(|p| p.join("Cargo.toml").exists() && p.join("test_data").exists())
            .expect("Could not find workspace root");
        let test_file =
            workspace_root.join("test_data/real/databento/ES.FUT_ohlcv-1m_2024-01-02.dbn");

        if !test_file.exists() {
            eprintln!(
                "Test file not found, skipping test: {}",
                test_file.display()
            );
            return;
        }

        let (sender, mut receiver) = broadcast::channel(1000);
        let publisher = Arc::new(EventPublisher::new(sender));

        let mut file_mapping = HashMap::new();
        file_mapping.insert(
            "ES.FUT".to_string(),
            test_file.to_string_lossy().to_string(),
        );

        let generator = DbnMarketDataGenerator::new(publisher, file_mapping)
            .await
            .unwrap();

        // Publish burst of 5 real bars
        generator.publish_burst("ES.FUT", 5).await.unwrap();

        // Verify we received events
        let mut count = 0;
        while let Ok(event) = receiver.try_recv() {
            assert_eq!(event.event_type, TradingEventType::PriceUpdate);
            assert_eq!(event.source, "dbn_market_data_generator");

            // Verify payload has real OHLCV data
            let payload: serde_json::Value = serde_json::from_str(&event.payload).expect("INVARIANT: Deserialization should succeed for valid JSON");
            assert!(payload["ohlcv"]["open"].as_f64().unwrap() > 0.0);
            assert!(payload["ohlcv"]["high"].as_f64().unwrap() > 0.0);
            assert!(payload["ohlcv"]["low"].as_f64().unwrap() > 0.0);
            assert!(payload["ohlcv"]["close"].as_f64().unwrap() > 0.0);
            assert!(payload["ohlcv"]["volume"].as_f64().unwrap() > 0.0);

            count += 1;
        }
        assert_eq!(count, 5, "Should receive exactly 5 market data events");
    }

    #[tokio::test]
    async fn test_generator_lifecycle() {
        let (sender, _) = broadcast::channel(1000);
        let publisher = Arc::new(EventPublisher::new(sender));

        let mut file_mapping = HashMap::new();
        file_mapping.insert(
            "TEST".to_string(),
            "test_data/real/databento/ES.FUT_ohlcv-1m_2024-01-02.dbn".to_string(),
        );

        let generator = DbnMarketDataGenerator::new(publisher, file_mapping)
            .await
            .unwrap();

        assert!(!generator.is_running().await);

        generator.start(100).await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        assert!(generator.is_running().await);

        generator.stop().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        assert!(!generator.is_running().await);
    }
}
