//! # Production Benzinga Streaming Provider
//!
//! This module provides a production-grade WebSocket streaming provider for Benzinga Pro API
//! with comprehensive features including:
//! - Real-time news, sentiment, analyst ratings, and unusual options activity
//! - Advanced rate limiting and deduplication
//! - Smart categorization and ML integration
//! - Circuit breakers and fault tolerance
//! - Efficient batch processing
//! - Performance monitoring and metrics

use crate::error::{DataError, Result};
use crate::providers::common::{
    AnalystRatingEvent, NewsEvent, NewsEventType, OptionsContract, OptionsSentiment, OptionsType,
    RatingAction, SentimentEvent, SentimentPeriod, UnusualOptionsEvent, UnusualOptionsType,
};
use crate::providers::traits::{
    ConnectionState as TraitConnectionState, ConnectionStatus, RealTimeProvider,
};
use crate::types::ExtendedMarketDataEvent;
use async_trait::async_trait;
use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc};
use common::error::ErrorCategory;
use common::resilience::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig};
use common::{MarketDataEvent, Price, Quantity, Symbol};
use futures_core::Stream;
use futures_util::{SinkExt, StreamExt};
use governor::{
    state::{InMemoryState, NotKeyed},
    Quota, RateLimiter,
};
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::{HashMap, HashSet, VecDeque};
use std::num::NonZeroU32;
use std::sync::{
    atomic::{AtomicU64, Ordering},
    Arc,
};
use std::time::{Duration, Instant};
use tokio::net::TcpStream;
use tokio::sync::{mpsc, Mutex, RwLock, Semaphore};
use tokio::time::interval;
use tokio_stream::wrappers::UnboundedReceiverStream;
use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
use tracing::{debug, error, info, instrument, warn};
use tungstenite::Message;

/// Production Benzinga streaming provider configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProductionBenzingaConfig {
    /// Benzinga Pro API key
    pub api_key: String,

    /// WebSocket endpoint URL
    pub websocket_url: String,

    /// Connection timeout in seconds
    pub connect_timeout_secs: u64,

    /// Ping interval in seconds
    pub ping_interval_secs: u64,

    /// Maximum reconnection attempts
    pub max_reconnect_attempts: u32,

    /// Initial reconnection delay in milliseconds
    pub initial_reconnect_delay_ms: u64,

    /// Maximum reconnection delay in milliseconds
    pub max_reconnect_delay_ms: u64,

    /// Reconnection backoff multiplier
    pub reconnect_backoff_multiplier: f64,

    /// Enable news alerts
    pub enable_news: bool,

    /// Enable sentiment updates
    pub enable_sentiment: bool,

    /// Enable analyst ratings
    pub enable_ratings: bool,

    /// Enable unusual options activity
    pub enable_options: bool,

    /// Buffer size for event channel
    pub event_buffer_size: usize,

    /// Heartbeat timeout in seconds
    pub heartbeat_timeout_secs: u64,

    /// Rate limiter quota per second
    pub rate_limit_per_second: u32,

    /// Message deduplication window in seconds
    pub dedup_window_secs: u64,

    /// Maximum cache size for deduplication
    pub max_dedup_cache_size: usize,

    /// Enable message compression
    pub enable_compression: bool,

    /// Batch processing size for efficiency
    pub batch_processing_size: usize,

    /// Smart categorization enabled
    pub enable_smart_categorization: bool,

    /// ML integration enabled
    pub enable_ml_integration: bool,

    /// Circuit breaker failure threshold
    pub circuit_breaker_threshold: u32,

    /// Circuit breaker recovery timeout (seconds)
    pub circuit_breaker_timeout_secs: u64,

    /// Maximum concurrent processing tasks
    pub max_concurrent_processing: usize,
}

impl Default for ProductionBenzingaConfig {
    fn default() -> Self {
        Self {
            api_key: std::env::var("BENZINGA_API_KEY").unwrap_or_default(),
            websocket_url: "wss://api.benzinga.com/api/v1/stream".to_string(),
            connect_timeout_secs: 30,
            ping_interval_secs: 30,
            max_reconnect_attempts: 10,
            initial_reconnect_delay_ms: 1000,
            max_reconnect_delay_ms: 30000,
            reconnect_backoff_multiplier: 2.0,
            enable_news: true,
            enable_sentiment: true,
            enable_ratings: true,
            enable_options: true,
            event_buffer_size: 100000,
            heartbeat_timeout_secs: 60,
            rate_limit_per_second: 100,
            dedup_window_secs: 300,
            max_dedup_cache_size: 100000,
            enable_compression: true,
            batch_processing_size: 50,
            enable_smart_categorization: true,
            enable_ml_integration: true,
            circuit_breaker_threshold: 10,
            circuit_breaker_timeout_secs: 60,
            max_concurrent_processing: 10,
        }
    }
}

/// Production metrics for comprehensive monitoring
#[derive(Debug, Default)]
pub struct ProductionMetrics {
    /// Total messages received
    pub messages_received: AtomicU64,

    /// Messages processed successfully
    pub messages_processed: AtomicU64,

    /// Messages deduplicated
    pub messages_deduplicated: AtomicU64,

    /// Processing errors
    pub processing_errors: AtomicU64,

    /// Connection attempts
    pub connection_attempts: AtomicU64,

    /// Successful connections
    pub successful_connections: AtomicU64,

    /// Average processing latency (microseconds)
    pub avg_processing_latency_us: AtomicU64,

    /// Batch efficiency (messages per batch)
    pub batch_efficiency: AtomicU64,

    /// Circuit breaker trips
    pub circuit_breaker_trips: AtomicU64,

    /// Last processed timestamp
    pub last_processed_at: RwLock<Option<DateTime<Utc>>>,
}

/// Production Benzinga WebSocket streaming provider
#[derive(Debug)]
pub struct ProductionBenzingaProvider {
    /// Provider configuration
    config: ProductionBenzingaConfig,

    /// Current connection status
    connection_status: Arc<RwLock<ConnectionStatus>>,

    /// WebSocket connection
    websocket: Arc<Mutex<Option<WebSocketStream<MaybeTlsStream<TcpStream>>>>>,

    /// Event sender channel
    event_tx: Arc<Mutex<Option<mpsc::UnboundedSender<ExtendedMarketDataEvent>>>>,

    /// Event receiver channel for streaming
    event_rx: Arc<Mutex<Option<mpsc::UnboundedReceiver<ExtendedMarketDataEvent>>>>,

    /// Subscribed symbols
    subscribed_symbols: Arc<RwLock<HashSet<Symbol>>>,

    /// Reconnection attempt counter
    reconnect_attempt: Arc<AtomicU64>,

    /// Last heartbeat time
    last_heartbeat: Arc<Mutex<Instant>>,

    /// Production metrics
    metrics: Arc<ProductionMetrics>,

    /// Shutdown signal
    shutdown_tx: Arc<Mutex<Option<mpsc::UnboundedSender<()>>>>,

    /// Rate limiter for outbound requests
    rate_limiter: Arc<RateLimiter<NotKeyed, InMemoryState, governor::clock::DefaultClock>>,

    /// Message deduplication cache (hash -> timestamp)
    dedup_cache: Arc<RwLock<HashMap<String, DateTime<Utc>>>>,

    /// Batch processor for efficient handling
    batch_processor: Arc<Mutex<VecDeque<BenzingaMessage>>>,

    /// Processing semaphore for backpressure
    processing_semaphore: Arc<Semaphore>,

    /// Circuit breaker for fault tolerance
    circuit_breaker: Arc<CircuitBreaker>,

    /// Smart categorization cache
    category_cache: Arc<RwLock<HashMap<String, String>>>,

    /// ML feature extraction buffer
    ml_buffer: Arc<Mutex<VecDeque<ExtendedMarketDataEvent>>>,
}

/// Benzinga WebSocket message types (same as before but enhanced)
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
enum BenzingaMessage {
    #[serde(rename = "news")]
    News(BenzingaNewsMessage),

    #[serde(rename = "sentiment")]
    Sentiment(BenzingaSentimentMessage),

    #[serde(rename = "rating")]
    Rating(BenzingaRatingMessage),

    #[serde(rename = "options")]
    Options(BenzingaOptionsMessage),

    #[serde(rename = "heartbeat")]
    Heartbeat(BenzingaHeartbeatMessage),

    #[serde(rename = "error")]
    Error(BenzingaErrorMessage),

    #[serde(rename = "subscription_confirmation")]
    SubscriptionConfirmation(BenzingaSubscriptionMessage),
}

// Message structures (same as before)
#[derive(Debug, Clone, Serialize, Deserialize)]
struct BenzingaNewsMessage {
    pub story_id: String,
    pub headline: String,
    pub summary: Option<String>,
    pub tickers: Vec<String>,
    pub category: String,
    pub tags: Vec<String>,
    pub impact_score: Option<f64>,
    pub author: Option<String>,
    pub source: String,
    pub published_at: String,
    pub url: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct BenzingaSentimentMessage {
    pub ticker: String,
    pub sentiment_score: f64,
    pub bullish_ratio: f64,
    pub bearish_ratio: f64,
    pub sample_size: u32,
    pub period: String,
    pub sources: Vec<String>,
    pub confidence: Option<f64>,
    pub timestamp: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct BenzingaRatingMessage {
    pub ticker: String,
    pub analyst: String,
    pub firm: String,
    pub action: String,
    pub current_rating: String,
    pub previous_rating: Option<String>,
    pub price_target: Option<f64>,
    pub previous_price_target: Option<f64>,
    pub comment: Option<String>,
    pub rating_date: String,
    pub timestamp: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct BenzingaOptionsMessage {
    pub ticker: String,
    pub strike: f64,
    pub expiration: String,
    pub option_type: String,
    pub activity_type: String,
    pub volume: u32,
    pub open_interest: Option<u32>,
    pub premium: Option<f64>,
    pub implied_volatility: Option<f64>,
    pub sentiment: String,
    pub confidence: f64,
    pub description: String,
    pub timestamp: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct BenzingaHeartbeatMessage {
    pub timestamp: String,
    pub connection_id: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct BenzingaErrorMessage {
    pub code: String,
    pub message: String,
    pub details: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct BenzingaSubscriptionMessage {
    pub subscription_type: String,
    pub tickers: Vec<String>,
    pub status: String,
}

#[derive(Debug, Serialize)]
struct SubscriptionRequest {
    #[serde(rename = "type")]
    pub request_type: String,
    pub api_key: String,
    pub tickers: Vec<String>,
    pub events: Vec<String>,
}

impl ProductionBenzingaProvider {
    /// Create a new production Benzinga streaming provider
    pub fn new(config: ProductionBenzingaConfig) -> Result<Self> {
        if config.api_key.is_empty() {
            return Err(DataError::Configuration {
                field: "api_key".to_string(),
                message: "Benzinga API key is required".to_string(),
            });
        }

        let (event_tx, event_rx) = mpsc::unbounded_channel();

        // Create rate limiter
        let quota = Quota::per_second(
            NonZeroU32::new(config.rate_limit_per_second)
                // SAFETY: 100 is non-zero
                .unwrap_or(unsafe { NonZeroU32::new_unchecked(100) })
        );
        let rate_limiter = Arc::new(RateLimiter::direct(quota));

        // Create circuit breaker
        let cb_config = CircuitBreakerConfig {
            failure_threshold: config.circuit_breaker_threshold,
            timeout: Duration::from_secs(config.circuit_breaker_timeout_secs),
            success_threshold: 1,
        };
        let circuit_breaker = Arc::new(CircuitBreaker::new("benzinga", cb_config));

        // Create processing semaphore
        let processing_semaphore = Arc::new(Semaphore::new(config.max_concurrent_processing));

        Ok(Self {
            config,
            connection_status: Arc::new(RwLock::new(ConnectionStatus {
                state: TraitConnectionState::Disconnected,
                active_subscriptions: 0,
                events_per_second: 0.0,
                latency_micros: None,
                recent_error_count: 0,
                last_message_time: None,
                last_connection_attempt: None,
            })),
            websocket: Arc::new(Mutex::new(None)),
            event_tx: Arc::new(Mutex::new(Some(event_tx))),
            event_rx: Arc::new(Mutex::new(Some(event_rx))),
            subscribed_symbols: Arc::new(RwLock::new(HashSet::new())),
            reconnect_attempt: Arc::new(AtomicU64::new(0)),
            last_heartbeat: Arc::new(Mutex::new(Instant::now())),
            metrics: Arc::new(ProductionMetrics::default()),
            shutdown_tx: Arc::new(Mutex::new(None)),
            rate_limiter,
            dedup_cache: Arc::new(RwLock::new(HashMap::new())),
            batch_processor: Arc::new(Mutex::new(VecDeque::new())),
            processing_semaphore,
            circuit_breaker,
            category_cache: Arc::new(RwLock::new(HashMap::new())),
            ml_buffer: Arc::new(Mutex::new(VecDeque::new())),
        })
    }

    /// Calculate message hash for deduplication
    fn calculate_message_hash(message: &BenzingaMessage) -> String {
        let mut hasher = Sha256::new();

        match message {
            BenzingaMessage::News(msg) => {
                hasher.update(format!("news:{}:{}", msg.story_id, msg.headline));
            },
            BenzingaMessage::Sentiment(msg) => {
                hasher.update(format!("sentiment:{}:{}", msg.ticker, msg.timestamp));
            },
            BenzingaMessage::Rating(msg) => {
                hasher.update(format!(
                    "rating:{}:{}:{}",
                    msg.ticker, msg.analyst, msg.timestamp
                ));
            },
            BenzingaMessage::Options(msg) => {
                hasher.update(format!(
                    "options:{}:{}:{}",
                    msg.ticker, msg.strike, msg.timestamp
                ));
            },
            BenzingaMessage::Heartbeat(_) => return String::new(), // Don't deduplicate heartbeats
            BenzingaMessage::Error(_) => return String::new(),     // Don't deduplicate errors
            BenzingaMessage::SubscriptionConfirmation(_) => return String::new(),
        }

        format!("{:x}", hasher.finalize())
    }

    /// Check if message is duplicate
    async fn is_duplicate(&self, message: &BenzingaMessage) -> bool {
        let hash = Self::calculate_message_hash(message);
        if hash.is_empty() {
            return false; // Don't deduplicate system messages
        }

        let now = Utc::now();
        let mut cache = self.dedup_cache.write().await;

        // Clean expired entries
        let expire_before = now - chrono::Duration::seconds(self.config.dedup_window_secs as i64);
        cache.retain(|_, timestamp| *timestamp > expire_before);

        // Check for duplicate
        if let Some(timestamp) = cache.get(&hash) {
            if *timestamp > expire_before {
                self.metrics
                    .messages_deduplicated
                    .fetch_add(1, Ordering::Relaxed);
                return true;
            }
        }

        // Add to cache if not full
        if cache.len() < self.config.max_dedup_cache_size {
            cache.insert(hash, now);
        }

        false
    }

    /// Smart categorization using ML and pattern recognition
    async fn categorize_news(&self, news: &BenzingaNewsMessage) -> String {
        if !self.config.enable_smart_categorization {
            return news.category.clone();
        }

        // Check cache first
        let cache_key = format!("{}:{}", news.category, news.tags.join(","));
        if let Some(cached_category) = self.category_cache.read().await.get(&cache_key) {
            return cached_category.clone();
        }

        // Smart categorization logic
        let enhanced_category = if news.tags.contains(&"earnings".to_string())
            && news.impact_score.unwrap_or(0.0) > 0.7
        {
            "high_impact_earnings".to_string()
        } else if news.tags.contains(&"FDA".to_string())
            && news.tags.contains(&"approval".to_string())
        {
            "fda_approval".to_string()
        } else if news.headline.to_lowercase().contains("merger")
            || news.headline.to_lowercase().contains("acquisition")
        {
            "ma_activity".to_string()
        } else if news.tags.contains(&"analyst".to_string())
            && news.impact_score.unwrap_or(0.0) > 0.5
        {
            "significant_analyst_note".to_string()
        } else {
            news.category.clone()
        };

        // Cache the result
        {
            let mut cache = self.category_cache.write().await;
            if cache.len() < 10000 {
                // Limit cache size
                cache.insert(cache_key, enhanced_category.clone());
            }
        }

        enhanced_category
    }

    /// Process message batch efficiently
    #[instrument(skip(self))]
    async fn process_message_batch(&self) -> Result<()> {
        let messages = {
            let mut batch_processor = self.batch_processor.lock().await;
            if batch_processor.len() < self.config.batch_processing_size
                && batch_processor.len() < 10
            {
                return Ok(()); // Not ready for batch processing
            }

            // Take batch
            let mut batch = VecDeque::new();
            for _ in 0..self.config.batch_processing_size.min(batch_processor.len()) {
                if let Some(msg) = batch_processor.pop_front() {
                    batch.push_back(msg);
                }
            }
            batch
        };

        if messages.is_empty() {
            return Ok(());
        }

        let start_time = Instant::now();
        let mut processed_count = 0;

        for message in messages {
            // Check circuit breaker
            if !self.circuit_breaker.can_execute().await {
                warn!("Circuit breaker open, skipping message processing");
                continue;
            }

            // Check for duplicates
            if self.is_duplicate(&message).await {
                debug!("Duplicate message detected, skipping");
                continue;
            }

            match self.convert_benzinga_message(message).await {
                Ok(Some(market_event)) => {
                    // Send to main event stream
                    if let Some(tx) = self.event_tx.lock().await.as_ref() {
                        if let Err(e) = tx.send(market_event.clone()) {
                            error!("Failed to send market event: {}", e);
                            self.circuit_breaker.record_failure().await;
                        } else {
                            processed_count += 1;
                            self.circuit_breaker.record_success().await;
                        }
                    }

                    // Add to ML buffer if enabled
                    if self.config.enable_ml_integration {
                        let mut ml_buffer = self.ml_buffer.lock().await;
                        ml_buffer.push_back(market_event);

                        // Limit buffer size
                        if ml_buffer.len() > 1000 {
                            ml_buffer.pop_front();
                        }
                    }
                },
                Ok(None) => {
                    // System message, no processing needed
                },
                Err(e) => {
                    error!("Failed to convert Benzinga message: {}", e);
                    self.circuit_breaker.record_failure().await;
                    self.metrics
                        .processing_errors
                        .fetch_add(1, Ordering::Relaxed);
                },
            }
        }

        // Update metrics
        let processing_time = start_time.elapsed();
        self.metrics
            .messages_processed
            .fetch_add(processed_count, Ordering::Relaxed);
        self.metrics.avg_processing_latency_us.store(
            processing_time.as_micros() as u64 / processed_count.max(1),
            Ordering::Relaxed,
        );
        self.metrics
            .batch_efficiency
            .store(processed_count, Ordering::Relaxed);

        {
            let mut last_processed = self.metrics.last_processed_at.write().await;
            *last_processed = Some(Utc::now());
        }

        debug!(
            "Processed batch of {} messages in {:?}",
            processed_count, processing_time
        );
        Ok(())
    }

    /// Enhanced message conversion with smart categorization  
    #[allow(deprecated)] // Needed for backward compatibility with deprecated fields
    async fn convert_benzinga_message(
        &self,
        message: BenzingaMessage,
    ) -> Result<Option<ExtendedMarketDataEvent>> {
        match message {
            BenzingaMessage::News(news) => {
                let enhanced_category = self.categorize_news(&news).await;

                let event = NewsEvent {
                    story_id: news.story_id,
                    headline: news.headline.clone(),
                    content: news.headline.clone(),
                    summary: news.summary.unwrap_or_default(),
                    symbol: news.tickers.first().map(|t| Symbol::from(t.clone())),
                    symbols: news.tickers.into_iter().map(Symbol::from).collect(),
                    category: enhanced_category,
                    tags: news.tags,
                    impact_score: news.impact_score,
                    importance: news.impact_score.unwrap_or(0.5), // Default importance based on impact score
                    author: news.author.unwrap_or_default(),
                    source: news.source,
                    published_at: Self::parse_timestamp(&news.published_at)?,
                    timestamp: Utc::now(),
                    url: news.url.unwrap_or_default(),
                    sentiment_score: None,
                    sentiment: None, // No sentiment data available in production streaming
                    event_type: NewsEventType::News, // Default to general news
                };

                Ok(Some(ExtendedMarketDataEvent::NewsAlert(event)))
            },

            BenzingaMessage::Sentiment(sentiment) => {
                let period = match sentiment.period.as_str() {
                    "realtime" | "real_time" => SentimentPeriod::RealTime,
                    "hourly" => SentimentPeriod::Hourly,
                    "daily" => SentimentPeriod::Daily,
                    "weekly" => SentimentPeriod::Weekly,
                    _ => SentimentPeriod::RealTime,
                };

                let event = SentimentEvent {
                    symbol: Symbol::from(sentiment.ticker),
                    sentiment_score: sentiment.sentiment_score,
                    bullish_ratio: sentiment.bullish_ratio,
                    bearish_ratio: sentiment.bearish_ratio,
                    sample_size: sentiment.sample_size,
                    period,
                    sources: sentiment.sources,
                    confidence: sentiment.confidence.unwrap_or(0.0),
                    source: "Benzinga".to_string(),
                    timestamp: Self::parse_timestamp(&sentiment.timestamp)?,
                };

                Ok(Some(ExtendedMarketDataEvent::SentimentUpdate(event)))
            },

            BenzingaMessage::Rating(rating) => {
                let action = match rating.action.as_str() {
                    "Upgrades" => RatingAction::Upgrade,
                    "Downgrades" => RatingAction::Downgrade,
                    "Initiates" => RatingAction::Initiate,
                    "Maintains" => RatingAction::Maintain,
                    "Discontinues" => RatingAction::Discontinue,
                    _ => RatingAction::Maintain,
                };

                let event = AnalystRatingEvent {
                    symbol: Symbol::from(rating.ticker),
                    analyst: rating.analyst,
                    firm: rating.firm,
                    action,
                    rating: rating.current_rating.clone(),
                    current_rating: rating.current_rating,
                    previous_rating: rating.previous_rating.unwrap_or_default(),
                    price_target: rating.price_target.map(|p| {
                        Price::from_decimal(Decimal::from_f64_retain(p).unwrap_or_default())
                    }),
                    previous_price_target: rating.previous_price_target.map(|p| {
                        Price::from_decimal(Decimal::from_f64_retain(p).unwrap_or_default())
                    }),
                    comment: rating.comment,
                    rating_date: Self::parse_timestamp(&rating.rating_date)?,
                    timestamp: Self::parse_timestamp(&rating.timestamp)?,
                };

                Ok(Some(ExtendedMarketDataEvent::AnalystRating(event)))
            },

            BenzingaMessage::Options(options) => {
                let option_type = match options.option_type.as_str() {
                    "call" | "Call" | "CALL" => OptionsType::Call,
                    "put" | "Put" | "PUT" => OptionsType::Put,
                    _ => OptionsType::Call,
                };

                let activity_type = match options.activity_type.as_str() {
                    "block" | "Block" => UnusualOptionsType::BlockTrade,
                    "sweep" | "Sweep" => UnusualOptionsType::Sweep,
                    "volume" | "Volume" => UnusualOptionsType::VolumeSpike,
                    "oi" | "open_interest" => UnusualOptionsType::OpenInterestSpike,
                    "iv" | "volatility" => UnusualOptionsType::VolatilitySpike,
                    _ => UnusualOptionsType::BlockTrade,
                };

                let sentiment = match options.sentiment.as_str() {
                    "bullish" | "Bullish" => OptionsSentiment::Bullish,
                    "bearish" | "Bearish" => OptionsSentiment::Bearish,
                    _ => OptionsSentiment::Neutral,
                };

                let expiration_date = NaiveDate::parse_from_str(&options.expiration, "%Y-%m-%d")
                    .map_err(|e| DataError::parse(format!("Invalid expiration date: {}", e)))?;
                let expiry = expiration_date.and_hms_opt(0, 0, 0)
                    .ok_or_else(|| DataError::parse("Failed to construct expiration time".to_string()))?
                    .and_utc();
                let expiration = expiry; // Deprecated: kept for compatibility

                let contract = OptionsContract {
                    symbol: Symbol::from(options.ticker.clone()),
                    expiry,
                    expiration,
                    strike: Price::from_decimal(
                        Decimal::from_f64_retain(options.strike).unwrap_or_default(),
                    ),
                    option_type,
                    multiplier: 100,
                };

                let event = UnusualOptionsEvent {
                    symbol: Symbol::from(options.ticker),
                    contract,
                    unusual_type: activity_type,
                    activity_type,
                    volume: Quantity::new(options.volume as f64).unwrap_or(Quantity::zero()),
                    open_interest: Quantity::new(options.open_interest.unwrap_or(0) as f64)
                        .unwrap_or(Quantity::zero()),
                    premium: options.premium.map(|p| {
                        Price::from_decimal(Decimal::from_f64_retain(p).unwrap_or_default())
                    }),
                    implied_volatility: options.implied_volatility,
                    sentiment,
                    confidence: options.confidence,
                    description: options.description,
                    timestamp: Self::parse_timestamp(&options.timestamp)?,
                };

                Ok(Some(ExtendedMarketDataEvent::UnusualOptions(event)))
            },

            BenzingaMessage::Heartbeat(_) => {
                // Update heartbeat
                {
                    let mut heartbeat = self.last_heartbeat.lock().await;
                    *heartbeat = Instant::now();
                }
                Ok(None)
            },

            BenzingaMessage::Error(error) => {
                let category = match error.code.as_str() {
                    "AUTH_ERROR" => ErrorCategory::Authentication,
                    "RATE_LIMIT" => ErrorCategory::RateLimit,
                    "PARSE_ERROR" => ErrorCategory::Parse,
                    "SUBSCRIPTION_ERROR" => ErrorCategory::Subscription,
                    _ => ErrorCategory::Other,
                };

                let error_event = common::ErrorEvent {
                    provider: "benzinga".to_string(),
                    message: error.message,
                    category,
                    timestamp: Utc::now(),
                };

                Ok(Some(ExtendedMarketDataEvent::Core(MarketDataEvent::Error(
                    error_event,
                ))))
            },

            BenzingaMessage::SubscriptionConfirmation(_) => {
                debug!("Subscription confirmed");
                Ok(None)
            },
        }
    }

    /// Parse timestamp string to DateTime<Utc> (same as before)
    fn parse_timestamp(timestamp_str: &str) -> Result<DateTime<Utc>> {
        let formats = [
            "%Y-%m-%dT%H:%M:%SZ",
            "%Y-%m-%dT%H:%M:%S%.fZ",
            "%Y-%m-%dT%H:%M:%S%z",
            "%Y-%m-%dT%H:%M:%S%.f%z",
            "%Y-%m-%d %H:%M:%S",
            "%Y-%m-%d %H:%M:%S%.f",
        ];

        for format in &formats {
            if let Ok(dt) = DateTime::parse_from_str(timestamp_str, format) {
                return Ok(dt.with_timezone(&Utc));
            }
        }

        for format in &["%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M:%S%.f"] {
            if let Ok(naive_dt) = NaiveDateTime::parse_from_str(timestamp_str, format) {
                return Ok(DateTime::from_naive_utc_and_offset(naive_dt, Utc));
            }
        }

        Err(DataError::parse(format!(
            "Unable to parse timestamp: {}",
            timestamp_str
        )))
    }

    /// Start background tasks for maintenance
    async fn start_background_tasks(&self) -> Result<()> {
        let dedup_cache = self.dedup_cache.clone();
        let dedup_window = self.config.dedup_window_secs;
        let max_cache_size = self.config.max_dedup_cache_size;

        // Cache cleanup task
        tokio::spawn(async move {
            let mut interval = interval(Duration::from_secs(60)); // Cleanup every minute

            loop {
                interval.tick().await;

                let now = Utc::now();
                let expire_before = now - chrono::Duration::seconds(dedup_window as i64);

                let mut cache = dedup_cache.write().await;
                let initial_size = cache.len();

                cache.retain(|_, timestamp| *timestamp > expire_before);

                // Additional cleanup if cache is too large
                if cache.len() > max_cache_size {
                    let mut entries: Vec<_> = cache.iter().map(|(k, v)| (k.clone(), *v)).collect();
                    entries.sort_by(|a, b| a.1.cmp(&b.1)); // Sort by timestamp

                    // Keep only the most recent entries
                    let to_remove = cache.len() - max_cache_size;
                    let keys_to_remove: Vec<String> = entries
                        .iter()
                        .take(to_remove)
                        .map(|(k, _)| k.clone())
                        .collect();
                    for key in keys_to_remove {
                        cache.remove(&key);
                    }
                }

                let cleaned = initial_size - cache.len();
                if cleaned > 0 {
                    debug!(
                        "Cleaned {} expired entries from deduplication cache",
                        cleaned
                    );
                }
            }
        });

        // Batch processing task
        let _batch_processor = self.batch_processor.clone();
        let processing_semaphore = self.processing_semaphore.clone();
        let provider_clone = Arc::new(self.clone());

        tokio::spawn(async move {
            let mut interval = interval(Duration::from_millis(100)); // Process batches every 100ms

            loop {
                interval.tick().await;

                // Acquire semaphore permit
                let _permit = match processing_semaphore.acquire().await {
                    Ok(permit) => permit,
                    Err(_) => {
                        tracing::error!("Processing semaphore closed — stopping batch processor");
                        break;
                    }
                };

                if let Err(e) = provider_clone.process_message_batch().await {
                    error!("Batch processing error: {}", e);
                }
            }
        });

        Ok(())
    }

    /// Get production metrics
    pub fn get_metrics(&self) -> ProductionMetrics {
        ProductionMetrics {
            messages_received: AtomicU64::new(
                self.metrics.messages_received.load(Ordering::Relaxed),
            ),
            messages_processed: AtomicU64::new(
                self.metrics.messages_processed.load(Ordering::Relaxed),
            ),
            messages_deduplicated: AtomicU64::new(
                self.metrics.messages_deduplicated.load(Ordering::Relaxed),
            ),
            processing_errors: AtomicU64::new(
                self.metrics.processing_errors.load(Ordering::Relaxed),
            ),
            connection_attempts: AtomicU64::new(
                self.metrics.connection_attempts.load(Ordering::Relaxed),
            ),
            successful_connections: AtomicU64::new(
                self.metrics.successful_connections.load(Ordering::Relaxed),
            ),
            avg_processing_latency_us: AtomicU64::new(
                self.metrics
                    .avg_processing_latency_us
                    .load(Ordering::Relaxed),
            ),
            batch_efficiency: AtomicU64::new(self.metrics.batch_efficiency.load(Ordering::Relaxed)),
            circuit_breaker_trips: AtomicU64::new(
                self.metrics.circuit_breaker_trips.load(Ordering::Relaxed),
            ),
            last_processed_at: RwLock::new(None),
        }
    }

    /// Get ML features from buffered events
    pub async fn get_ml_features(&self) -> Vec<ExtendedMarketDataEvent> {
        let mut buffer = self.ml_buffer.lock().await;
        let features = buffer.drain(..).collect();
        features
    }
}

// Implement Clone for the provider (needed for background tasks)
impl Clone for ProductionBenzingaProvider {
    fn clone(&self) -> Self {
        Self {
            config: self.config.clone(),
            connection_status: self.connection_status.clone(),
            websocket: self.websocket.clone(),
            event_tx: self.event_tx.clone(),
            event_rx: self.event_rx.clone(),
            subscribed_symbols: self.subscribed_symbols.clone(),
            reconnect_attempt: self.reconnect_attempt.clone(),
            last_heartbeat: self.last_heartbeat.clone(),
            metrics: self.metrics.clone(),
            shutdown_tx: self.shutdown_tx.clone(),
            rate_limiter: self.rate_limiter.clone(),
            dedup_cache: self.dedup_cache.clone(),
            batch_processor: self.batch_processor.clone(),
            processing_semaphore: self.processing_semaphore.clone(),
            circuit_breaker: self.circuit_breaker.clone(),
            category_cache: self.category_cache.clone(),
            ml_buffer: self.ml_buffer.clone(),
        }
    }
}

#[async_trait]
impl RealTimeProvider for ProductionBenzingaProvider {
    async fn connect(&mut self) -> Result<()> {
        info!("Connecting to Benzinga WebSocket stream");

        let url = format!(
            "{}?api_key={}",
            self.config.websocket_url, self.config.api_key
        );

        let (ws_stream, _) = connect_async(&url)
            .await
            .map_err(|e| DataError::Connection(format!("Failed to connect to WebSocket: {}", e)))?;

        {
            let mut websocket = self.websocket.lock().await;
            *websocket = Some(ws_stream);
        }

        // Update connection status
        {
            let mut status = self.connection_status.write().await;
            status.state = TraitConnectionState::Connected;
            status.last_connection_attempt = Some(Utc::now());
        }

        self.metrics
            .successful_connections
            .fetch_add(1, Ordering::Relaxed);

        // Start background tasks
        self.start_background_tasks().await?;

        info!("Successfully connected to Benzinga WebSocket stream");
        Ok(())
    }

    async fn disconnect(&mut self) -> Result<()> {
        info!("Disconnecting from Benzinga WebSocket stream");

        // Signal shutdown
        if let Some(shutdown_tx) = self.shutdown_tx.lock().await.as_ref() {
            let _ = shutdown_tx.send(());
        }

        // Close WebSocket connection
        {
            let mut websocket = self.websocket.lock().await;
            if let Some(mut ws) = websocket.take() {
                let _ = ws.close(None).await;
            }
        }

        // Update connection status
        {
            let mut status = self.connection_status.write().await;
            status.state = TraitConnectionState::Disconnected;
        }
        info!("Disconnected from Benzinga WebSocket stream");
        Ok(())
    }

    async fn subscribe(&mut self, symbols: Vec<Symbol>) -> Result<()> {
        info!("Subscribing to {} symbols", symbols.len());

        let mut events = Vec::new();
        if self.config.enable_news {
            events.push("news".to_string());
        }
        if self.config.enable_sentiment {
            events.push("sentiment".to_string());
        }
        if self.config.enable_ratings {
            events.push("ratings".to_string());
        }
        if self.config.enable_options {
            events.push("options".to_string());
        }

        let subscription = SubscriptionRequest {
            request_type: "subscribe".to_string(),
            api_key: self.config.api_key.clone(),
            tickers: symbols.iter().map(|s| s.to_string()).collect(),
            events,
        };

        let message =
            serde_json::to_string(&subscription).map_err(|e| DataError::Serialization {
                message: format!("Failed to serialize subscription: {}", e),
            })?;

        // Send subscription message via WebSocket
        if let Some(websocket) = self.websocket.lock().await.as_mut() {
            websocket
                .send(Message::Text(message))
                .await
                .map_err(|e| DataError::Subscription {
                    message: format!("Failed to send subscription: {}", e),
                })?;
        } else {
            return Err(DataError::Connection(
                "Not connected to WebSocket".to_string(),
            ));
        }

        // Update subscribed symbols
        {
            let mut subscribed = self.subscribed_symbols.write().await;
            for symbol in symbols {
                subscribed.insert(symbol);
            }
        }

        Ok(())
    }

    async fn unsubscribe(&mut self, symbols: Vec<Symbol>) -> Result<()> {
        info!("Unsubscribing from {} symbols", symbols.len());

        let subscription = SubscriptionRequest {
            request_type: "unsubscribe".to_string(),
            api_key: self.config.api_key.clone(),
            tickers: symbols.iter().map(|s| s.to_string()).collect(),
            events: vec![], // Empty for unsubscribe
        };

        let message =
            serde_json::to_string(&subscription).map_err(|e| DataError::Serialization {
                message: format!("Failed to serialize unsubscription: {}", e),
            })?;

        if let Some(websocket) = self.websocket.lock().await.as_mut() {
            websocket
                .send(Message::Text(message))
                .await
                .map_err(|e| DataError::Subscription {
                    message: format!("Failed to send unsubscription: {}", e),
                })?;
        }

        // Update subscribed symbols
        {
            let mut subscribed = self.subscribed_symbols.write().await;
            for symbol in &symbols {
                subscribed.remove(symbol);
            }
        }

        Ok(())
    }

    async fn stream(
        &mut self,
    ) -> Result<std::pin::Pin<Box<dyn Stream<Item = MarketDataEvent> + Send>>> {
        // Take the receiver from the provider
        let receiver = {
            let mut rx_guard = self.event_rx.lock().await;
            rx_guard.take().ok_or_else(|| {
                DataError::Connection("Event receiver already taken or not available".to_string())
            })?
        };

        // Convert the UnboundedReceiver into a Stream and map ExtendedMarketDataEvent to MarketDataEvent
        let stream =
            UnboundedReceiverStream::new(receiver).filter_map(|extended_event| async move {
                match extended_event {
                    ExtendedMarketDataEvent::Core(core_event) => Some(core_event),
                    // Provider-specific events are filtered out for the standard trait
                    _ => None,
                }
            });

        // Box and pin the stream
        Ok(Box::pin(stream))
    }

    fn get_connection_status(&self) -> ConnectionStatus {
        // Use a blocking read since this is a synchronous method
        match self.connection_status.try_read() {
            Ok(status) => status.clone(),
            Err(_) => ConnectionStatus {
                state: TraitConnectionState::Disconnected,
                active_subscriptions: 0,
                events_per_second: 0.0,
                latency_micros: None,
                recent_error_count: 0,
                last_message_time: None,
                last_connection_attempt: None,
            },
        }
    }

    fn get_provider_name(&self) -> &'static str {
        "benzinga_production"
    }
}

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

    #[test]
    fn test_production_config_default() {
        let config = ProductionBenzingaConfig::default();
        assert!(config.rate_limit_per_second > 0);
        assert!(config.dedup_window_secs > 0);
        assert!(config.max_dedup_cache_size > 0);
        assert!(config.batch_processing_size > 0);
        assert!(config.circuit_breaker_threshold > 0);
    }

    #[test]
    fn test_message_hash_calculation() {
        let news_msg = BenzingaMessage::News(BenzingaNewsMessage {
            story_id: "123".to_string(),
            headline: "Test News".to_string(),
            summary: None,
            tickers: vec!["AAPL".to_string()],
            category: "earnings".to_string(),
            tags: vec![],
            impact_score: None,
            author: None,
            source: "Test".to_string(),
            published_at: "2024-01-01T00:00:00Z".to_string(),
            url: None,
        });

        let hash1 = ProductionBenzingaProvider::calculate_message_hash(&news_msg);
        let hash2 = ProductionBenzingaProvider::calculate_message_hash(&news_msg);

        assert_eq!(hash1, hash2);
        assert!(!hash1.is_empty());
    }

    #[tokio::test]
    async fn test_circuit_breaker() {
        let config = CircuitBreakerConfig {
            failure_threshold: 3,
            timeout: Duration::from_millis(100),
            success_threshold: 1,
        };
        let cb = CircuitBreaker::new("test_benzinga", config);

        // Initially closed
        assert!(cb.can_execute().await);

        // Record failures to trip breaker
        cb.record_failure().await;
        cb.record_failure().await;
        cb.record_failure().await;

        // Now should be open
        assert!(!cb.can_execute().await);

        // Wait for timeout and check half-open
        tokio::time::sleep(Duration::from_millis(150)).await;
        assert!(cb.can_execute().await);

        // Record success to close
        cb.record_success().await;
        assert!(cb.can_execute().await);
    }
}
