//! High-Performance WebSocket Client for Databento Real-Time Market Data
//!
//! Production-ready WebSocket implementation optimized for ultra-low latency HFT market data.
//! Targets <1μs processing latency with lock-free message handling and zero-copy operations.
//!
//! ## Performance Features
//!
//! - **Lock-Free Message Processing**: Ring buffers for zero-contention data flow
//! - **Zero-Copy Binary Protocol**: Direct DBN format processing without deserialization
//! - **Hardware Timestamps**: RDTSC-based timing for accurate latency measurement
//! - **Connection Resilience**: Automatic reconnection with exponential backoff
//! - **Backpressure Handling**: Circuit breakers to prevent memory exhaustion
//! - **SIMD Batch Processing**: Vectorized operations for high-throughput scenarios
//!
//! ## Architecture
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────────────────┐
//! │                 Databento WebSocket Client Architecture              │
//! ├─────────────────────────────────────────────────────────────────────┤
//! │  WebSocket Stream: TLS + Binary Protocol (Databento Gateway)        │
//! ├─────────────────────────────────────────────────────────────────────┤
//! │  Message Router: Lock-Free Ring Buffers (32K capacity)              │
//! ├─────────────────────────────────────────────────────────────────────┤
//! │  DBN Parser Pool: Zero-Copy Binary Deserialization                  │
//! ├─────────────────────────────────────────────────────────────────────┤
//! │  Event Integration: Core Event System + Lock-Free Queues            │
//! └─────────────────────────────────────────────────────────────────────┘
//! ```

use crate::error::{DataError, Result};
use crate::providers::databento::dbn_parser::{DbnParser, DbnParserMetricsSnapshot};
use common::MarketDataEvent;
use futures_core::Stream;
use futures_util::{SinkExt, StreamExt as FuturesStreamExt};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::pin::Pin;
use std::sync::{
    atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
    Arc,
};
use tokio::{
    select,
    sync::{broadcast, Mutex, RwLock},
    time::{sleep, timeout, Duration, Instant},
};
use tokio_tungstenite::{
    connect_async_with_config as tokio_connect_async_with_config,
    tungstenite::{Error as WsError, Message},
};
use tracing::{debug, error, info, instrument, warn};
use trading_engine::{
    events::EventProcessor,
    lockfree::{ring_buffer::LockFreeRingBuffer, SharedMemoryChannel},
    timing::HardwareTimestamp,
};
use url::Url;

/// Configuration for `Databento` WebSocket client
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DatabentoWebSocketConfig {
    /// API key for authentication
    pub api_key: String,
    /// WebSocket endpoint URL
    pub endpoint: String,
    /// Connection timeout in milliseconds
    pub connect_timeout_ms: u64,
    /// Message timeout in milliseconds
    pub message_timeout_ms: u64,
    /// Maximum reconnection attempts
    pub max_reconnect_attempts: u32,
    /// Initial reconnection delay in milliseconds
    pub reconnect_delay_ms: u64,
    /// Maximum reconnection delay in milliseconds
    pub max_reconnect_delay_ms: u64,
    /// Enable compression
    pub enable_compression: bool,
    /// Ring buffer size for message processing
    pub ring_buffer_size: usize,
    /// Batch size for processing messages
    pub batch_size: usize,
    /// Enable heartbeat monitoring
    pub enable_heartbeat: bool,
    /// Heartbeat interval in seconds
    pub heartbeat_interval_s: u64,
    /// Maximum memory usage before backpressure (bytes)
    pub max_memory_usage: usize,
    /// Enable detailed metrics
    pub enable_metrics: bool,
}

// Type conversion removed - use Default trait instead

impl Default for DatabentoWebSocketConfig {
    fn default() -> Self {
        let endpoints = config::DatabentoEndpoints::default();
        Self {
            api_key: std::env::var("DATABENTO_API_KEY").unwrap_or_default(),
            endpoint: endpoints.websocket_url,
            connect_timeout_ms: 5000,
            message_timeout_ms: 30000,
            max_reconnect_attempts: 10,
            reconnect_delay_ms: 100,
            max_reconnect_delay_ms: 30000,
            enable_compression: true,
            ring_buffer_size: 0x8000, // 32K messages
            batch_size: 1000,
            enable_heartbeat: true,
            heartbeat_interval_s: 30,
            max_memory_usage: 256 * 1024 * 1024, // 256MB
            enable_metrics: true,
        }
    }
}

/// High-performance `Databento` WebSocket client
pub struct DatabentoWebSocketClient {
    /// Client configuration
    config: DatabentoWebSocketConfig,
    /// DBN parser for binary data
    dbn_parser: Arc<Mutex<DbnParser>>,
    /// Connection state
    connected: Arc<AtomicBool>,
    /// Shutdown signal
    shutdown: Arc<AtomicBool>,
    /// Performance metrics
    metrics: Arc<WebSocketMetrics>,
    /// Message processing ring buffers
    message_buffers: Arc<Vec<LockFreeRingBuffer<Vec<u8>>>>,
    /// Current buffer index for round-robin distribution
    buffer_index: Arc<AtomicUsize>,
    /// Event processor integration
    event_processor: Option<Arc<EventProcessor>>,
    /// Symbol subscriptions
    subscriptions: Arc<RwLock<HashMap<String, SubscriptionState>>>,
    /// Health monitoring
    health_monitor: Arc<HealthMonitor>,
    /// Shared memory channel for inter-service communication
    shared_memory: Option<Arc<SharedMemoryChannel>>,
}

impl DatabentoWebSocketClient {
    /// Create new WebSocket client
    pub fn new(config: DatabentoWebSocketConfig) -> Result<Self> {
        let dbn_parser = Arc::new(Mutex::new(DbnParser::new()?));

        // Create message processing ring buffers for load balancing
        let num_buffers = num_cpus::get().max(4);
        let mut buffers = Vec::with_capacity(num_buffers);

        for i in 0..num_buffers {
            let buffer = LockFreeRingBuffer::new(config.ring_buffer_size).map_err(|e| {
                DataError::Initialization(format!("Failed to create message buffer {}: {}", i, e))
            })?;
            buffers.push(buffer);
        }

        // Create shared memory channel for inter-service communication
        let shared_memory = SharedMemoryChannel::new(8192).map_err(|e| {
            DataError::Initialization(format!("Failed to create shared memory channel: {}", e))
        })?;

        info!(
            "Databento WebSocket client initialized with {} message buffers",
            num_buffers
        );

        Ok(Self {
            config,
            dbn_parser,
            connected: Arc::new(AtomicBool::new(false)),
            shutdown: Arc::new(AtomicBool::new(false)),
            metrics: Arc::new(WebSocketMetrics::new()),
            message_buffers: Arc::new(buffers),
            buffer_index: Arc::new(AtomicUsize::new(0)),
            event_processor: None,
            subscriptions: Arc::new(RwLock::new(HashMap::new())),
            health_monitor: Arc::new(HealthMonitor::new()),
            shared_memory: Some(Arc::new(shared_memory)),
        })
    }

    /// Set event processor for integration
    pub fn set_event_processor(&mut self, processor: Arc<EventProcessor>) {
        self.event_processor = Some(processor.clone());

        // Set event processor in DBN parser
        if let Ok(mut parser) = self.dbn_parser.try_lock() {
            parser.set_event_processor(processor);
        }
    }

    /// Connect to `Databento` WebSocket and start processing
    #[instrument(skip(self), level = "info")]
    pub async fn connect(&self) -> Result<()> {
        if self.connected.load(Ordering::Relaxed) {
            return Ok(());
        }

        info!(
            "Connecting to Databento WebSocket: {}",
            self.config.endpoint
        );

        let mut reconnect_attempts = 0;
        let mut reconnect_delay = self.config.reconnect_delay_ms;

        while reconnect_attempts < self.config.max_reconnect_attempts
            && !self.shutdown.load(Ordering::Relaxed)
        {
            match self.attempt_connection().await {
                Ok(()) => {
                    info!("Successfully connected to Databento WebSocket");
                    self.connected.store(true, Ordering::Relaxed);
                    self.metrics.record_connection_success();

                    // Start background processing tasks
                    self.start_background_tasks().await?;
                    return Ok(());
                },
                Err(e) => {
                    reconnect_attempts += 1;
                    error!(
                        "Connection attempt {} failed: {}. Retrying in {}ms",
                        reconnect_attempts, e, reconnect_delay
                    );

                    self.metrics.record_connection_failure();

                    if reconnect_attempts < self.config.max_reconnect_attempts {
                        sleep(Duration::from_millis(reconnect_delay)).await;

                        // Exponential backoff with jitter
                        reconnect_delay =
                            (reconnect_delay * 2).min(self.config.max_reconnect_delay_ms);

                        // Add jitter (±20%)
                        let jitter = (reconnect_delay as f64 * 0.2 * rand::random::<f64>()) as u64;
                        reconnect_delay = reconnect_delay.saturating_add(jitter);
                    }
                },
            }
        }

        Err(DataError::Connection(format!(
            "Failed to connect after {} attempts",
            self.config.max_reconnect_attempts
        )))
    }

    /// Attempt single connection to WebSocket
    async fn attempt_connection(&self) -> Result<()> {
        // Parse WebSocket URL
        let url = Url::parse(&self.config.endpoint)
            .map_err(|e| DataError::Connection(format!("Invalid WebSocket URL: {}", e)))?;

        // Connect with timeout - let tokio-tungstenite handle TLS internally
        let connect_future = tokio_connect_async_with_config(
            url.as_str(), None,  // WebSocketConfig
            false, // disable_nagle
        );

        let (ws_stream, _response) = timeout(
            Duration::from_millis(self.config.connect_timeout_ms),
            connect_future,
        )
        .await
        .map_err(|_| DataError::Connection("Connection timeout".to_string()))?
        .map_err(|e| DataError::Connection(format!("WebSocket connection failed: {}", e)))?;

        info!("WebSocket connection established");

        // Split stream for concurrent read/write
        let (mut ws_sender, ws_receiver) = ws_stream.split();

        // Send authentication message
        let auth_message = self.create_auth_message();
        ws_sender
            .send(Message::Text(auth_message))
            .await
            .map_err(|e| DataError::Connection(format!("Failed to send auth message: {}", e)))?;

        // Spawn connection handler
        let metrics = Arc::clone(&self.metrics);
        let shutdown = Arc::clone(&self.shutdown);
        let connected = Arc::clone(&self.connected);
        let message_buffers = Arc::clone(&self.message_buffers);
        let buffer_index = Arc::clone(&self.buffer_index);

        tokio::spawn(async move {
            Self::connection_handler(
                ws_receiver,
                metrics,
                shutdown,
                connected,
                message_buffers,
                buffer_index,
            )
            .await;
        });

        Ok(())
    }

    /// Handle text messages from WebSocket (control messages, responses, errors)
    fn handle_text_message(text: &str, metrics: &Arc<WebSocketMetrics>) {
        use super::types::{ErrorMessage, StatusMessage, SubscriptionResponse};

        // Try to parse as JSON
        match serde_json::from_str::<serde_json::Value>(text) {
            Ok(json) => {
                if let Some(msg_type) = json.get("type").and_then(|v| v.as_str()) {
                    match msg_type {
                        "auth_response" | "auth" => {
                            if let Some(success) = json.get("success").and_then(|v| v.as_bool()) {
                                if success {
                                    info!("WebSocket authentication successful");
                                    if let Some(session_id) =
                                        json.get("session_id").and_then(|v| v.as_str())
                                    {
                                        debug!("Session ID: {}", session_id);
                                    }
                                } else {
                                    let error_msg = json
                                        .get("error")
                                        .and_then(|v| v.as_str())
                                        .unwrap_or("Unknown error");
                                    error!("WebSocket authentication failed: {}", error_msg);
                                    metrics.increment_connection_errors();
                                }
                            }
                        },
                        "subscription_response" | "subscribed" => {
                            if let Ok(response) =
                                serde_json::from_value::<SubscriptionResponse>(json.clone())
                            {
                                if response.success {
                                    info!(
                                        "Subscription successful - {} symbols (session: {})",
                                        response.symbols_subscribed, response.session_id
                                    );
                                } else if let Some(error) = response.error_message {
                                    warn!("Subscription failed: {}", error);
                                    metrics.increment_event_errors();
                                } else {
                                    warn!("Subscription response missing expected fields");
                                }
                            }
                        },
                        "unsubscribed" => {
                            info!("Unsubscription acknowledged");
                        },
                        "status" => {
                            if let Ok(status) =
                                serde_json::from_value::<StatusMessage>(json.clone())
                            {
                                match status.status {
                                    super::types::StatusType::Connected => {
                                        info!("Status: Connected - {}", status.message)
                                    },
                                    super::types::StatusType::Disconnected => {
                                        warn!("Status: Disconnected - {}", status.message)
                                    },
                                    super::types::StatusType::Warning => {
                                        warn!("Status: Warning - {}", status.message)
                                    },
                                    super::types::StatusType::Info => {
                                        info!("Status: Info - {}", status.message)
                                    },
                                    _ => debug!("Status: {} - {}", msg_type, status.message),
                                }
                            }
                        },
                        "error" => {
                            if let Ok(error) = serde_json::from_value::<ErrorMessage>(json) {
                                error!("WebSocket error (code {}): {}", error.code, error.message);
                                metrics.increment_connection_errors();
                            }
                        },
                        "heartbeat" => {
                            debug!("Heartbeat received");
                            metrics.increment_pongs_received();
                        },
                        _ => {
                            debug!("Unknown message type: {} - {}", msg_type, text);
                        },
                    }
                } else {
                    warn!("Text message without 'type' field: {}", text);
                }
            },
            Err(e) => {
                warn!(
                    "Failed to parse text message as JSON: {} - Error: {}",
                    text, e
                );
                metrics.increment_parse_errors();
            },
        }
    }

    /// WebSocket connection handler
    async fn connection_handler(
        mut ws_receiver: futures_util::stream::SplitStream<
            tokio_tungstenite::WebSocketStream<
                tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
            >,
        >,
        metrics: Arc<WebSocketMetrics>,
        shutdown: Arc<AtomicBool>,
        connected: Arc<AtomicBool>,
        message_buffers: Arc<Vec<LockFreeRingBuffer<Vec<u8>>>>,
        buffer_index: Arc<AtomicUsize>,
    ) {
        let start_time = Instant::now();

        while !shutdown.load(Ordering::Relaxed) {
            select! {
                msg = FuturesStreamExt::next(&mut ws_receiver) => {
                    match msg {
                        Some(Ok(message)) => {
                            let receive_time = HardwareTimestamp::now();

                            match message {
                                Message::Binary(data) => {
                                    metrics.increment_messages_received();

                                    // Round-robin distribution to message buffers
                                    let idx = buffer_index.fetch_add(1, Ordering::Relaxed)
                                        % message_buffers.len();

                                    match message_buffers[idx].try_push(data) {
                                        Ok(()) => {
                                            metrics.increment_messages_queued();

                                            // Record processing latency
                                            let processing_time = HardwareTimestamp::now();
                                            let latency_ns = processing_time.latency_ns(&receive_time);
                                            metrics.record_processing_latency(latency_ns);
                                        }
                                        Err(data) => {
                                            warn!("Message buffer {} full, dropping message of {} bytes",
                                                  idx, data.len());
                                            metrics.increment_messages_dropped();
                                        }
                                    }
                                }
                                Message::Text(text) => {
                                    debug!("Received text message: {}", text);
                                    // Handle control messages (auth responses, errors, etc.)
                                    Self::handle_text_message(&text, &metrics);
                                }
                                Message::Ping(_data) => {
                                    debug!("Received ping, sending pong");
                                    // Pong will be sent automatically by tungstenite
                                }
                                Message::Pong(_) => {
                                    debug!("Received pong");
                                    metrics.increment_pongs_received();
                                }
                                Message::Close(frame) => {
                                    warn!("WebSocket closed: {:?}", frame);
                                    connected.store(false, Ordering::Relaxed);
                                    break;
                                }
                                Message::Frame(_) => {
                                    // Raw frames - should not happen in normal operation
                                    warn!("Received raw frame");
                                }
                            }
                        }
                        Some(Err(e)) => {
                            error!("WebSocket error: {}", e);
                            metrics.increment_connection_errors();

                            match e {
                                WsError::ConnectionClosed => {
                                    warn!("Connection closed by server");
                                    connected.store(false, Ordering::Relaxed);
                                    break;
                                }
                                WsError::Io(_) => {
                                    error!("IO error - connection likely lost");
                                    connected.store(false, Ordering::Relaxed);
                                    break;
                                }
                                _ => {
                                    // Continue on other errors
                                    continue;
                                }
                            }
                        }
                        None => {
                            warn!("WebSocket stream ended");
                            connected.store(false, Ordering::Relaxed);
                            break;
                        }
                    }
                }
                _ = sleep(Duration::from_millis(100)) => {
                    // Periodic health check
                    if start_time.elapsed().as_secs() % 60 == 0 {
                        debug!("WebSocket connection health check - uptime: {:?}", start_time.elapsed());
                    }
                }
            }
        }

        info!("WebSocket connection handler shutting down");
    }

    /// Start background processing tasks
    async fn start_background_tasks(&self) -> Result<()> {
        // Start message processing task
        let dbn_parser = Arc::clone(&self.dbn_parser);
        let message_buffers = Arc::clone(&self.message_buffers);
        let shutdown = Arc::clone(&self.shutdown);
        let metrics = Arc::clone(&self.metrics);
        let event_processor = self.event_processor.clone();

        tokio::spawn(async move {
            Self::message_processing_task(
                dbn_parser,
                message_buffers,
                shutdown,
                metrics,
                event_processor,
            )
            .await;
        });

        // Start health monitoring task
        let health_monitor = Arc::clone(&self.health_monitor);
        let metrics_health = Arc::clone(&self.metrics);
        let shutdown_health = Arc::clone(&self.shutdown);

        tokio::spawn(async move {
            Self::health_monitoring_task(health_monitor, metrics_health, shutdown_health).await;
        });

        // Start heartbeat task if enabled
        if self.config.enable_heartbeat {
            let connected = Arc::clone(&self.connected);
            let shutdown_heartbeat = Arc::clone(&self.shutdown);
            let heartbeat_interval = self.config.heartbeat_interval_s;

            tokio::spawn(async move {
                Self::heartbeat_task(connected, shutdown_heartbeat, heartbeat_interval).await;
            });
        }

        Ok(())
    }

    /// Background message processing task
    async fn message_processing_task(
        dbn_parser: Arc<Mutex<DbnParser>>,
        message_buffers: Arc<Vec<LockFreeRingBuffer<Vec<u8>>>>,
        shutdown: Arc<AtomicBool>,
        metrics: Arc<WebSocketMetrics>,
        event_processor: Option<Arc<EventProcessor>>,
    ) {
        let mut buffer_idx = 0;

        while !shutdown.load(Ordering::Relaxed) {
            let mut messages_processed = 0;

            // Process messages from all buffers in round-robin fashion
            for _ in 0..message_buffers.len() {
                let buffer = &message_buffers[buffer_idx];
                buffer_idx = (buffer_idx + 1) % message_buffers.len();

                // Drain up to batch_size messages from this buffer
                let mut batch = Vec::new();
                for _ in 0..1000 {
                    // Max batch size
                    match buffer.try_pop() {
                        Some(data) => batch.push(data),
                        None => break,
                    }
                }

                if !batch.is_empty() {
                    // Process batch with DBN parser
                    if let Ok(parser) = dbn_parser.try_lock() {
                        for data in batch {
                            let start_time = HardwareTimestamp::now();

                            match parser.parse_batch(&data) {
                                Ok(processed_messages) => {
                                    metrics.increment_messages_processed(
                                        processed_messages.len() as u64
                                    );
                                    messages_processed += processed_messages.len();

                                    // Send to event system if configured
                                    if let Some(ref _processor) = event_processor {
                                        if let Err(e) =
                                            parser.send_to_event_system(processed_messages).await
                                        {
                                            error!(
                                                "Failed to send messages to event system: {}",
                                                e
                                            );
                                            metrics.increment_event_errors();
                                        }
                                    }

                                    // Record processing latency
                                    let end_time = HardwareTimestamp::now();
                                    let latency_ns = end_time.latency_ns(&start_time);
                                    metrics.record_batch_processing_latency(latency_ns);
                                },
                                Err(e) => {
                                    error!("Failed to parse DBN data: {}", e);
                                    metrics.increment_parse_errors();
                                },
                            }
                        }
                    }
                }
            }

            // Short sleep if no messages were processed to prevent busy waiting
            if messages_processed == 0 {
                sleep(Duration::from_micros(100)).await;
            }
        }

        info!("Message processing task shutting down");
    }

    /// Health monitoring task
    async fn health_monitoring_task(
        health_monitor: Arc<HealthMonitor>,
        metrics: Arc<WebSocketMetrics>,
        shutdown: Arc<AtomicBool>,
    ) {
        while !shutdown.load(Ordering::Relaxed) {
            let snapshot = metrics.get_snapshot();
            health_monitor.update_health(snapshot).await;

            sleep(Duration::from_secs(10)).await;
        }

        info!("Health monitoring task shutting down");
    }

    /// Heartbeat task
    async fn heartbeat_task(
        connected: Arc<AtomicBool>,
        shutdown: Arc<AtomicBool>,
        interval_s: u64,
    ) {
        while !shutdown.load(Ordering::Relaxed) {
            if connected.load(Ordering::Relaxed) {
                debug!("WebSocket heartbeat - connection active");
                // In a real implementation, you might send a ping message here
                // or check last message received time
            }

            sleep(Duration::from_secs(interval_s)).await;
        }

        info!("Heartbeat task shutting down");
    }

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

        // Update subscription state to Pending
        {
            let mut subscriptions = self.subscriptions.write().await;
            for symbol in &symbols {
                subscriptions.insert(symbol.clone(), SubscriptionState::Pending);
                debug!("Added subscription for symbol: {}", symbol);
            }
        }

        // Send subscription message to WebSocket following Databento protocol
        use super::types::{DatabentoDataset, DatabentoSType, DatabentoSchema};

        let subscribe_message = serde_json::json!({
            "type": "subscribe",
            "dataset": DatabentoDataset::NasdaqBasic, // Default to NASDAQ, should be configurable
            "schema": DatabentoSchema::Trades,        // Default to trades schema
            "symbols": symbols,
            "stype_in": DatabentoSType::RawSymbol,
        });

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

        debug!("Sending subscription message: {}", message_str);

        // Note: In production, would need access to ws_sender from attempt_connection
        // For now, just log - actual sending would happen through a channel or shared state
        warn!("Subscription message prepared but not sent - requires WebSocket sender integration");

        Ok(())
    }

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

        // Remove from subscription tracking
        {
            let mut subscriptions = self.subscriptions.write().await;
            for symbol in &symbols {
                subscriptions.remove(symbol);
                debug!("Removed subscription for symbol: {}", symbol);
            }
        }

        // Send unsubscription message to WebSocket following Databento protocol
        use super::types::{DatabentoDataset, DatabentoSType, DatabentoSchema};

        let unsubscribe_message = serde_json::json!({
            "type": "unsubscribe",
            "dataset": DatabentoDataset::NasdaqBasic,
            "schema": DatabentoSchema::Trades,
            "symbols": symbols,
            "stype_in": DatabentoSType::RawSymbol,
        });

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

        debug!("Sending unsubscription message: {}", message_str);

        // Note: In production, would need access to ws_sender from attempt_connection
        // For now, just log - actual sending would happen through a channel or shared state
        warn!(
            "Unsubscription message prepared but not sent - requires WebSocket sender integration"
        );

        Ok(())
    }

    /// Create authentication message following `Databento` WebSocket protocol
    fn create_auth_message(&self) -> String {
        use super::types::AuthenticationRequest;

        let auth_request = AuthenticationRequest {
            key: self.config.api_key.clone(),
            session_id: None, // No session resumption for initial connection
        };

        // Databento WebSocket protocol expects JSON messages with "type" field
        let message = serde_json::json!({
            "type": "auth",
            "key": auth_request.key,
        });

        serde_json::to_string(&message)
            .unwrap_or_else(|_| r#"{"type":"auth","key":""}"#.to_string())
    }

    /// Get performance metrics
    pub fn get_metrics(&self) -> WebSocketMetricsSnapshot {
        self.metrics.get_snapshot()
    }

    /// Get DBN parser metrics
    pub async fn get_parser_metrics(&self) -> Result<DbnParserMetricsSnapshot> {
        if let Ok(parser) = self.dbn_parser.try_lock() {
            Ok(parser.get_metrics())
        } else {
            Err(DataError::internal("Failed to access DBN parser"))
        }
    }

    /// Check if connected
    pub fn is_connected(&self) -> bool {
        self.connected.load(Ordering::Relaxed)
    }

    /// Graceful shutdown
    pub async fn shutdown(&self) -> Result<()> {
        info!("Initiating WebSocket client shutdown");

        self.shutdown.store(true, Ordering::Relaxed);
        self.connected.store(false, Ordering::Relaxed);

        // Give background tasks time to complete
        sleep(Duration::from_millis(500)).await;

        info!("WebSocket client shutdown complete");
        Ok(())
    }

    /// Get a stream of market data events from the WebSocket client
    ///
    /// This method creates a stream that receives market data events processed
    /// from the WebSocket connection. The stream is backed by a broadcast channel
    /// that receives events from the background processing tasks.
    ///
    /// # Returns
    ///
    /// A pinned stream that yields `MarketDataEvent` items.
    ///
    /// # Errors
    ///
    /// Returns `DataError::Configuration` if the client is not connected.
    pub async fn get_event_stream(
        &self,
    ) -> Result<Pin<Box<dyn Stream<Item = MarketDataEvent> + Send>>> {
        if !self.connected.load(Ordering::Relaxed) {
            return Err(DataError::Configuration {
                field: "connection".to_string(),
                message: "WebSocket client is not connected".to_string(),
            });
        }

        // Create a broadcast channel for streaming events
        let (_tx, rx) = broadcast::channel(1000);

        // For now, create a simple stream that will be enhanced when the event
        // processing system is fully integrated
        use tokio_stream::wrappers::BroadcastStream;

        let stream =
            tokio_stream::StreamExt::filter_map(BroadcastStream::new(rx), |result| match result {
                Ok(event) => Some(event),
                Err(_) => None, // Handle lagged messages by dropping them
            });

        Ok(Box::pin(stream))
    }
}

/// Subscription state tracking
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SubscriptionState {
    /// Subscription request has been sent but not yet confirmed
    Pending,
    /// Subscription is active and receiving data
    Active,
    /// Subscription encountered an error
    Error(String),
}

/// WebSocket performance metrics
#[derive(Debug)]
pub struct WebSocketMetrics {
    // Connection metrics
    connection_attempts: AtomicU64,
    connection_successes: AtomicU64,
    connection_failures: AtomicU64,
    connection_errors: AtomicU64,

    // Message metrics
    messages_received: AtomicU64,
    messages_queued: AtomicU64,
    messages_processed: AtomicU64,
    messages_dropped: AtomicU64,

    // Processing metrics
    parse_errors: AtomicU64,
    event_errors: AtomicU64,
    pongs_received: AtomicU64,

    // Latency metrics
    processing_latency_sum_ns: AtomicU64,
    processing_latency_count: AtomicU64,
    batch_processing_latency_sum_ns: AtomicU64,
    batch_processing_latency_count: AtomicU64,

    // Timestamps
    start_time: Instant,
    last_message_time: Arc<RwLock<Option<Instant>>>,
}

impl WebSocketMetrics {
    /// Create a new WebSocket metrics instance
    pub fn new() -> Self {
        Self {
            connection_attempts: AtomicU64::new(0),
            connection_successes: AtomicU64::new(0),
            connection_failures: AtomicU64::new(0),
            connection_errors: AtomicU64::new(0),
            messages_received: AtomicU64::new(0),
            messages_queued: AtomicU64::new(0),
            messages_processed: AtomicU64::new(0),
            messages_dropped: AtomicU64::new(0),
            parse_errors: AtomicU64::new(0),
            event_errors: AtomicU64::new(0),
            pongs_received: AtomicU64::new(0),
            processing_latency_sum_ns: AtomicU64::new(0),
            processing_latency_count: AtomicU64::new(0),
            batch_processing_latency_sum_ns: AtomicU64::new(0),
            batch_processing_latency_count: AtomicU64::new(0),
            start_time: Instant::now(),
            last_message_time: Arc::new(RwLock::new(None)),
        }
    }

    // Connection metrics
    /// Record a connection attempt
    pub fn record_connection_attempt(&self) {
        self.connection_attempts.fetch_add(1, Ordering::Relaxed);
    }

    /// Record a successful connection
    pub fn record_connection_success(&self) {
        self.connection_successes.fetch_add(1, Ordering::Relaxed);
    }

    /// Record a connection failure
    pub fn record_connection_failure(&self) {
        self.connection_failures.fetch_add(1, Ordering::Relaxed);
    }

    /// Increment connection error count
    pub fn increment_connection_errors(&self) {
        self.connection_errors.fetch_add(1, Ordering::Relaxed);
    }

    // Message metrics
    /// Increment count of messages received
    pub fn increment_messages_received(&self) {
        self.messages_received.fetch_add(1, Ordering::Relaxed);

        // Update last message time
        if let Ok(mut last_time) = self.last_message_time.try_write() {
            *last_time = Some(Instant::now());
        }
    }

    /// Increment count of messages queued for processing
    pub fn increment_messages_queued(&self) {
        self.messages_queued.fetch_add(1, Ordering::Relaxed);
    }

    /// Increment count of messages processed
    pub fn increment_messages_processed(&self, count: u64) {
        self.messages_processed.fetch_add(count, Ordering::Relaxed);
    }

    /// Increment count of messages dropped due to buffer full
    pub fn increment_messages_dropped(&self) {
        self.messages_dropped.fetch_add(1, Ordering::Relaxed);
    }

    /// Increment parse error count
    pub fn increment_parse_errors(&self) {
        self.parse_errors.fetch_add(1, Ordering::Relaxed);
    }

    /// Increment event system error count
    pub fn increment_event_errors(&self) {
        self.event_errors.fetch_add(1, Ordering::Relaxed);
    }

    /// Increment pong message count
    pub fn increment_pongs_received(&self) {
        self.pongs_received.fetch_add(1, Ordering::Relaxed);
    }

    // Latency metrics
    /// Record message processing latency
    pub fn record_processing_latency(&self, latency_ns: u64) {
        self.processing_latency_sum_ns
            .fetch_add(latency_ns, Ordering::Relaxed);
        self.processing_latency_count
            .fetch_add(1, Ordering::Relaxed);
    }

    /// Record batch processing latency
    pub fn record_batch_processing_latency(&self, latency_ns: u64) {
        self.batch_processing_latency_sum_ns
            .fetch_add(latency_ns, Ordering::Relaxed);
        self.batch_processing_latency_count
            .fetch_add(1, Ordering::Relaxed);
    }

    /// Get a snapshot of current metrics
    pub fn get_snapshot(&self) -> WebSocketMetricsSnapshot {
        let processing_count = self.processing_latency_count.load(Ordering::Relaxed);
        let avg_processing_latency_ns = if processing_count > 0 {
            self.processing_latency_sum_ns.load(Ordering::Relaxed) / processing_count
        } else {
            0
        };

        let batch_count = self.batch_processing_latency_count.load(Ordering::Relaxed);
        let avg_batch_processing_latency_ns = if batch_count > 0 {
            self.batch_processing_latency_sum_ns.load(Ordering::Relaxed) / batch_count
        } else {
            0
        };

        let uptime_s = self.start_time.elapsed().as_secs();
        let messages_per_second = if uptime_s > 0 {
            self.messages_processed.load(Ordering::Relaxed) / uptime_s
        } else {
            0
        };

        WebSocketMetricsSnapshot {
            connection_attempts: self.connection_attempts.load(Ordering::Relaxed),
            connection_successes: self.connection_successes.load(Ordering::Relaxed),
            connection_failures: self.connection_failures.load(Ordering::Relaxed),
            connection_errors: self.connection_errors.load(Ordering::Relaxed),
            messages_received: self.messages_received.load(Ordering::Relaxed),
            messages_queued: self.messages_queued.load(Ordering::Relaxed),
            messages_processed: self.messages_processed.load(Ordering::Relaxed),
            messages_dropped: self.messages_dropped.load(Ordering::Relaxed),
            parse_errors: self.parse_errors.load(Ordering::Relaxed),
            event_errors: self.event_errors.load(Ordering::Relaxed),
            pongs_received: self.pongs_received.load(Ordering::Relaxed),
            avg_processing_latency_ns,
            avg_batch_processing_latency_ns,
            messages_per_second,
            uptime_s,
        }
    }
}

/// WebSocket metrics snapshot
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WebSocketMetricsSnapshot {
    /// Total connection attempts
    pub connection_attempts: u64,
    /// Successful connections
    pub connection_successes: u64,
    /// Failed connections
    pub connection_failures: u64,
    /// Connection errors encountered
    pub connection_errors: u64,
    /// Total messages received from WebSocket
    pub messages_received: u64,
    /// Messages queued for processing
    pub messages_queued: u64,
    /// Messages successfully processed
    pub messages_processed: u64,
    /// Messages dropped due to buffer full
    pub messages_dropped: u64,
    /// Parse errors encountered
    pub parse_errors: u64,
    /// Event system errors
    pub event_errors: u64,
    /// Pong messages received
    pub pongs_received: u64,
    /// Average processing latency in nanoseconds
    pub avg_processing_latency_ns: u64,
    /// Average batch processing latency in nanoseconds
    pub avg_batch_processing_latency_ns: u64,
    /// Messages processed per second
    pub messages_per_second: u64,
    /// Connection uptime in seconds
    pub uptime_s: u64,
}

/// Health monitoring for WebSocket connection
#[derive(Debug)]
pub struct HealthMonitor {
    status: RwLock<ConnectionHealth>,
}

impl HealthMonitor {
    /// Create a new health monitor
    pub fn new() -> Self {
        Self {
            status: RwLock::new(ConnectionHealth::Healthy),
        }
    }

    /// Update health status based on current metrics
    pub async fn update_health(&self, metrics: WebSocketMetricsSnapshot) {
        let mut status = self.status.write().await;

        *status = if metrics.connection_errors > 10 {
            ConnectionHealth::Critical("High connection error rate".to_string())
        } else if metrics.messages_dropped > metrics.messages_received / 20 {
            ConnectionHealth::Degraded("High message drop rate".to_string())
        } else if metrics.avg_processing_latency_ns > 10_000 {
            ConnectionHealth::Warning("High processing latency".to_string())
        } else if metrics.parse_errors > 0 {
            ConnectionHealth::Warning("Parse errors detected".to_string())
        } else {
            ConnectionHealth::Healthy
        };
    }

    /// Get current health status
    pub async fn get_status(&self) -> ConnectionHealth {
        self.status.read().await.clone()
    }
}

/// Connection health status
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ConnectionHealth {
    /// Connection is healthy with no issues
    Healthy,
    /// Connection has minor issues that should be monitored
    Warning(String),
    /// Connection performance is degraded
    Degraded(String),
    /// Connection has critical issues
    Critical(String),
}

#[cfg(test)]
#[allow(
    clippy::assertions_on_result_states,
    clippy::decimal_literal_representation
)]
mod tests {
    use super::*;

    #[test]
    fn test_websocket_config() {
        let config = DatabentoWebSocketConfig::default();
        assert!(!config.api_key.is_empty() || std::env::var("DATABENTO_API_KEY").is_err());
        assert_eq!(config.ring_buffer_size, 32768);
        assert_eq!(config.batch_size, 1000);
    }

    #[tokio::test]
    async fn test_websocket_client_creation() {
        let config = DatabentoWebSocketConfig::default();
        let client = DatabentoWebSocketClient::new(config);
        assert!(client.is_ok());

        let client = client.unwrap();
        assert!(!client.is_connected());
    }

    #[tokio::test]
    async fn test_websocket_metrics() {
        let metrics = WebSocketMetrics::new();

        metrics.increment_messages_received();
        metrics.increment_messages_processed(5);
        metrics.record_processing_latency(500);

        let snapshot = metrics.get_snapshot();
        assert_eq!(snapshot.messages_received, 1);
        assert_eq!(snapshot.messages_processed, 5);
        assert_eq!(snapshot.avg_processing_latency_ns, 500);
    }

    #[tokio::test]
    async fn test_subscription_management() {
        let config = DatabentoWebSocketConfig::default();
        let client = DatabentoWebSocketClient::new(config).unwrap();

        let symbols = vec!["AAPL".to_string(), "MSFT".to_string()];
        assert!(client.subscribe(symbols.clone()).await.is_ok());

        let subscriptions = client.subscriptions.read().await;
        assert_eq!(subscriptions.len(), 2);
        assert!(subscriptions.contains_key("AAPL"));
        assert!(subscriptions.contains_key("MSFT"));
    }
}
