foxhunt/data/src/providers/databento_streaming.rs

//! # Databento Streaming Market Data Provider
//!
//! High-performance WebSocket client for Databento market data streaming.
//! Provides real-time market data with microsecond timestamps and full order book depth.

use crate::error::{DataError, Result};
use crate::providers::{MarketDataProvider, MarketStatus, ProviderHealthStatus};
use crate::types::TimeRange;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use common::MarketDataEvent;
use serde::{Deserialize, Serialize};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use tokio::sync::broadcast;
use tokio_tungstenite::{connect_async, tungstenite::Message};
use tracing::{debug, error, info, warn};
// MarketDataEvent is already imported from common::types
use common::OrderBookEvent;
use common::Price;
use common::Quantity;
use common::QuoteEvent;
use common::Symbol;
use common::TradeEvent;
use rust_decimal::Decimal;
use url::Url;

/// Databento WebSocket client for real-time market data
#[derive(Debug)]
pub struct DatabentoStreamingProvider {
    /// WebSocket endpoint
    endpoint: String,
    /// API key for authentication
    api_key: String,
    /// Connection status
    connected: Arc<AtomicBool>,
    /// Event sender for market data
    _event_sender: broadcast::Sender<MarketDataEvent>,
    /// Health metrics
    messages_received: Arc<AtomicU64>,
    last_message_time: Arc<AtomicU64>,
    error_count: Arc<AtomicU64>,
    /// Provider name
    name: String,
}

impl DatabentoStreamingProvider {
    /// Create new Databento streaming provider
    pub fn new(api_key: String) -> Result<Self> {
        let (_event_sender, _) = broadcast::channel(10000);

        Ok(Self {
            endpoint: "wss://gateway.databento.com/v2".to_string(),
            api_key,
            connected: Arc::new(AtomicBool::new(false)),
            _event_sender,
            messages_received: Arc::new(AtomicU64::new(0)),
            last_message_time: Arc::new(AtomicU64::new(0)),
            error_count: Arc::new(AtomicU64::new(0)),
            name: "databento".to_string(),
        })
    }

    /// Get market data event receiver for core integration
    pub fn subscribe_market_events(&self) -> broadcast::Receiver<MarketDataEvent> {
        self._event_sender.subscribe()
    }

    /// Handle incoming WebSocket message
    async fn handle_message(&self, message: Message) -> Result<()> {
        match message {
            Message::Text(text) => {
                self.process_text_message(&text).await?;
            },
            Message::Binary(data) => {
                self.process_binary_message(&data).await?;
            },
            Message::Ping(_) => {
                debug!("Received ping from Databento");
                // Pong will be sent automatically by tungstenite
            },
            Message::Pong(_) => {
                debug!("Received pong from Databento");
            },
            Message::Close(frame) => {
                warn!("Databento connection closed: {:?}", frame);
                self.connected.store(false, Ordering::Relaxed);
            },
            _ => {
                warn!("Received unexpected message type from Databento");
            },
        }
        Ok(())
    }

    /// Process text message from Databento
    async fn process_text_message(&self, text: &str) -> Result<()> {
        match serde_json::from_str::<DatabentoMessage>(text) {
            Ok(msg) => {
                self.process_databento_message(msg).await?;
                self.messages_received.fetch_add(1, Ordering::Relaxed);
                self.last_message_time
                    .store(Utc::now().timestamp_millis() as u64, Ordering::Relaxed);
            },
            Err(e) => {
                error!("Failed to parse Databento message: {}", e);
                self.error_count.fetch_add(1, Ordering::Relaxed);
            },
        }
        Ok(())
    }

    /// Process binary message from Databento (optimized format)
    async fn process_binary_message(&self, _data: &[u8]) -> Result<()> {
        // Binary message processing would go here for high-frequency data
        // This would use Databento's binary protocol for maximum performance
        debug!("Received binary message from Databento (not yet implemented)");
        Ok(())
    }

    /// Process parsed Databento message
    async fn process_databento_message(&self, message: DatabentoMessage) -> Result<()> {
        match message {
            DatabentoMessage::Trade(trade) => {
                let event = MarketDataEvent::Trade(TradeEvent {
                    symbol: trade.symbol,
                    timestamp: trade.timestamp,
                    price: Decimal::from(trade.price),
                    size: Decimal::from(trade.size),
                    trade_id: trade.trade_id,
                    exchange: trade.exchange,
                    conditions: vec![], // Add missing field
                    sequence: 0,
                });
                let _ = self._event_sender.send(event);
            },
            DatabentoMessage::Quote(quote) => {
                let event = MarketDataEvent::Quote(QuoteEvent {
                    symbol: quote.symbol,
                    timestamp: quote.timestamp,
                    bid: quote.bid.map(Decimal::from),
                    bid_size: quote.bid_size.map(Decimal::from),
                    ask: quote.ask.map(Decimal::from),
                    ask_size: quote.ask_size.map(Decimal::from),
                    exchange: quote.exchange,
                    bid_exchange: None, // Add missing fields
                    ask_exchange: None,
                    conditions: vec![],
                    sequence: 0,
                });
                let _ = self._event_sender.send(event);
            },
            DatabentoMessage::OrderBook(book) => {
                let event = MarketDataEvent::OrderBook(OrderBookEvent {
                    symbol: book.symbol,
                    timestamp: book.timestamp,
                    bids: book.bids,
                    asks: book.asks,
                });
                let _ = self._event_sender.send(event);
            },
            DatabentoMessage::Status(status) => {
                info!("Databento status update: {:?}", status);
            },
            DatabentoMessage::Error(error) => {
                error!("Databento error: {:?}", error);
                self.error_count.fetch_add(1, Ordering::Relaxed);
            },
        }
        Ok(())
    }

    /// Send subscription message
    async fn send_subscription(&self, symbols: Vec<Symbol>) -> Result<()> {
        let subscription = DatabentoSubscription {
            action: "subscribe".to_string(),
            symbols,
            data_types: vec![
                "trades".to_string(),
                "quotes".to_string(),
                "orderbook".to_string(),
            ],
            schema: "ohlcv-1s".to_string(),
        };

        let message =
            serde_json::to_string(&subscription).map_err(|e| DataError::Serialization {
                message: e.to_string(),
            })?;

        debug!("Sending Databento subscription: {}", message);
        // WebSocket sending would be handled by the connection loop
        Ok(())
    }
}

#[async_trait]
impl MarketDataProvider for DatabentoStreamingProvider {
    async fn connect(&mut self) -> Result<()> {
        if self.connected.load(Ordering::Relaxed) {
            return Ok(());
        }

        info!("Connecting to Databento at {}", self.endpoint);

        let url = Url::parse(&self.endpoint)
            .map_err(|e| DataError::Connection(format!("Invalid Databento URL: {}", e)))?;

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

        info!("Connected to Databento successfully");
        self.connected.store(true, Ordering::Relaxed);

        // Spawn connection handler
        let _connected = Arc::clone(&self.connected);
        let _provider = self.clone();

        tokio::spawn(async move {
            // Connection handling logic would go here
            // This would handle the WebSocket stream and process incoming messages
        });

        Ok(())
    }

    async fn disconnect(&mut self) -> Result<()> {
        if !self.connected.load(Ordering::Relaxed) {
            return Ok(());
        }

        info!("Disconnecting from Databento");
        self.connected.store(false, Ordering::Relaxed);
        Ok(())
    }

    async fn subscribe(&mut self, symbols: Vec<String>) -> Result<()> {
        if !self.connected.load(Ordering::Relaxed) {
            return Err(DataError::Connection(
                "Not connected to Databento".to_string(),
            ));
        }

        info!("Subscribing to {} symbols on Databento", symbols.len());
        // Convert Vec<String> to Vec<Symbol> for internal processing
        let symbol_structs: Vec<Symbol> = symbols
            .into_iter()
            .map(|s| Symbol::from(s.as_str()))
            .collect();
        self.send_subscription(symbol_structs).await?;
        Ok(())
    }

    async fn unsubscribe(&mut self, symbols: Vec<String>) -> Result<()> {
        if !self.connected.load(Ordering::Relaxed) {
            return Ok(());
        }

        info!("Unsubscribing from {} symbols on Databento", symbols.len());

        // Convert Vec<String> to Vec<Symbol> for internal processing
        let symbol_structs: Vec<Symbol> = symbols
            .into_iter()
            .map(|s| Symbol::from(s.as_str()))
            .collect();
        let unsubscription = DatabentoSubscription {
            action: "unsubscribe".to_string(),
            symbols: symbol_structs,
            data_types: vec![],
            schema: "".to_string(),
        };

        let _message =
            serde_json::to_string(&unsubscription).map_err(|e| DataError::Serialization {
                message: e.to_string(),
            })?;

        Ok(())
    }

    async fn get_historical_data(
        &self,
        _symbol: &str,
        _timeframe: &str,
        _range: TimeRange,
    ) -> Result<Vec<MarketDataEvent>> {
        // Historical data retrieval would be implemented here
        // Using Databento's historical data API
        warn!("Historical data retrieval not yet implemented for Databento streaming");
        Ok(vec![])
    }

    async fn get_market_status(&self) -> Result<MarketStatus> {
        // Market status would be retrieved from Databento API
        Ok(MarketStatus {
            is_open: true, // This would be determined from Databento API
            next_open: None,
            next_close: None,
            timezone: "America/New_York".to_string(),
            extended_hours: true,
        })
    }

    fn get_health_status(&self) -> ProviderHealthStatus {
        let now = Utc::now().timestamp_millis() as u64;
        let last_message = self.last_message_time.load(Ordering::Relaxed);
        let messages_received = self.messages_received.load(Ordering::Relaxed);

        // Calculate messages per second over the last minute
        let messages_per_second = if last_message > 0 && now > last_message {
            let seconds_since_last = (now - last_message) / 1000;
            if seconds_since_last > 0 {
                messages_received as f64 / seconds_since_last as f64
            } else {
                0.0
            }
        } else {
            0.0
        };

        ProviderHealthStatus {
            connected: self.connected.load(Ordering::Relaxed),
            last_connected: if self.connected.load(Ordering::Relaxed) {
                Some(Utc::now())
            } else {
                None
            },
            active_subscriptions: 0, // This would track actual subscriptions
            messages_per_second,
            latency_micros: None, // This would be calculated from ping/pong
            error_count: self.error_count.load(Ordering::Relaxed) as u32,
        }
    }

    fn get_name(&self) -> &str {
        &self.name
    }
}

impl Clone for DatabentoStreamingProvider {
    fn clone(&self) -> Self {
        let (_event_sender, _) = broadcast::channel(10000);
        Self {
            endpoint: self.endpoint.clone(),
            api_key: self.api_key.clone(),
            connected: Arc::clone(&self.connected),
            _event_sender,
            messages_received: Arc::clone(&self.messages_received),
            last_message_time: Arc::clone(&self.last_message_time),
            error_count: Arc::clone(&self.error_count),
            name: self.name.clone(),
        }
    }
}

/// Databento message types
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum DatabentoMessage {
    #[serde(rename = "trade")]
    Trade(DatabentoTrade),
    #[serde(rename = "quote")]
    Quote(DatabentoQuote),
    #[serde(rename = "orderbook")]
    OrderBook(DatabentoOrderBook),
    #[serde(rename = "status")]
    Status(DatabentoStatus),
    #[serde(rename = "error")]
    Error(DatabentoError),
}

/// Databento trade message
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DatabentoTrade {
    pub symbol: String,
    pub timestamp: DateTime<Utc>,
    pub price: Price,
    pub size: Quantity,
    pub trade_id: Option<String>,
    pub exchange: Option<String>,
    pub conditions: Option<Vec<String>>,
}

/// Databento quote message
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DatabentoQuote {
    pub symbol: String,
    pub timestamp: DateTime<Utc>,
    pub bid: Option<Price>,
    pub bid_size: Option<Quantity>,
    pub ask: Option<Price>,
    pub ask_size: Option<Quantity>,
    pub exchange: Option<String>,
}

/// Databento order book message
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DatabentoOrderBook {
    pub symbol: String,
    pub timestamp: DateTime<Utc>,
    pub bids: Vec<(Price, Quantity)>,
    pub asks: Vec<(Price, Quantity)>,
    pub sequence: Option<u64>,
}

/// Databento status message
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DatabentoStatus {
    pub message: String,
    pub timestamp: DateTime<Utc>,
    pub level: String,
}

/// Databento error message
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DatabentoError {
    pub error_message: String,
    pub code: Option<i32>,
    pub timestamp: DateTime<Utc>,
}

/// Databento subscription request
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DatabentoSubscription {
    pub action: String,
    pub symbols: Vec<Symbol>,
    pub data_types: Vec<String>,
    pub schema: String,
}

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

    #[test]
    fn test_databento_streaming_provider_creation() {
        let provider = DatabentoStreamingProvider::new("test-api-key".to_string());
        assert!(provider.is_ok());

        let provider = provider.unwrap();
        assert_eq!(provider.get_name(), "databento");
        assert!(!provider.connected.load(Ordering::Relaxed));
    }

    #[test]
    fn test_databento_message_serialization() {
        let trade = DatabentoTrade {
            symbol: "SPY".to_string(),
            timestamp: Utc::now(),
            price: Price::from_f64(425.50).unwrap(),
            size: Quantity::new(100.0).unwrap(),
            trade_id: Some("12345".to_string()),
            exchange: Some("NYSE".to_string()),
            conditions: None,
        };

        let message = DatabentoMessage::Trade(trade);
        let json = serde_json::to_string(&message);
        assert!(json.is_ok());
    }
}