rusty_ems/exchanges/

upbit_websocket_trading.rs

//! Upbit WebSocket Trading Implementation
//!
//! This module provides a high-performance WebSocket trading client for Upbit's private endpoints
//! with JWT authentication, real-time order updates, and asset tracking.
//!
//! # Features
//!
//! - **JWT authentication**: Support for Upbit's JWT-based WebSocket authentication
//! - **Private channels**: Real-time myOrder and myAsset data streams
//! - **Order management**: Integration with REST API for order placement
//! - **Asset tracking**: Real-time balance and position updates
//! - **Connection health monitoring**: Real-time metrics and health status
//! - **Automatic reconnection**: Configurable backoff strategy with exponential delays
//! - **Zero-copy message processing**: Optimized for low-latency trading
//!
//! # Connection Management
//!
//! The client maintains connection health through:
//! - WebSocket ping/pong frames sent every 2 minutes (Upbit recommendation)
//! - Automatic reconnection on connection loss
//! - JWT token refresh before expiration
//! - Configurable timeout handling
//!
//! # Private Channels
//!
//! Supports Upbit's private WebSocket channels:
//! - `myOrder`: Real-time order updates (creation/execution/cancellation)
//! - `myAsset`: Real-time asset/balance updates
//! - Event-driven updates only (no snapshot data)
//!
//! # Usage Example
//!
//! ```rust,no_run
//! use rusty_common::auth::exchanges::upbit::{UpbitAuth, UpbitAuthConfig};
//! use rusty_ems::exchanges::upbit_websocket_trading::UpbitWebSocketTrader;
//! use std::sync::Arc;
//!
//! #[tokio::main]
//! async fn main() -> Result<()> {
//!     let config = UpbitAuthConfig::new("access_key".into(), "secret_key".into());
//!     let auth = Arc::new(UpbitAuth::new(config));
//!
//!     let trader = UpbitWebSocketTrader::new(auth);
//!     let (report_tx, report_rx) = flume::bounded(100);
//!
//!     // Connect and authenticate
//!     trader.connect(report_tx).await?;
//!
//!     Ok(())
//! }
//! ```

use crate::utils::time::timestamp_nanos;
use anyhow::{Result, bail};
use async_trait::async_trait;
use flume::Sender;
use futures::{SinkExt, StreamExt};
use log::{debug, error, info, warn};
use parking_lot::RwLock;
use quanta::Clock;
use rust_decimal::Decimal;
use rusty_common::auth::ExchangeAuthentication;
use rusty_common::auth::exchanges::upbit::UpbitAuth;
use rusty_common::types::Exchange;
use rusty_common::utils::id_generation;
use rusty_common::websocket::{
    Message, WebSocketConfig,
    client::ConnectionState as WsConnectionState,
    connector::{WebSocketConnector, WebSocketSink, WebSocketStream},
    stats::new_shared_stats,
};
use rusty_model::{
    enums::OrderStatus, instruments::InstrumentId, trading_order::Order, venues::Venue,
};
use simd_json;
use simd_json::prelude::{ValueAsArray, ValueAsScalar, ValueObjectAccess};
use smartstring::alias::String as SmartString;
use std::sync::Arc;
use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
use std::time::Duration;
use tokio::sync::RwLock as AsyncRwLock;
use tokio::task::JoinHandle;
use uuid::Uuid;

use crate::execution_engine::ExecutionReport;

/// Upbit private WebSocket API URL
const UPBIT_PRIVATE_WS_URL: &str = "wss://api.upbit.com/websocket/v1/private";

/// Ping interval for keeping WebSocket connection alive (2 minutes as recommended by Upbit)
const PING_INTERVAL_SECONDS: u64 = 120;

/// Pong timeout (10 seconds - if no pong received within this time, consider connection dead)
const PONG_TIMEOUT_SECONDS: u64 = 10;

/// Maximum message size allowed (10MB)
const MAX_MESSAGE_SIZE: usize = 10 * 1024 * 1024;

/// Upbit-specific connection state for proper state machine tracking
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum UpbitConnectionState {
    /// WebSocket connection is not established
    Disconnected = 0,
    /// WebSocket connection is in progress
    Connecting = 1,
    /// WebSocket connection is established but not authenticated
    Connected = 2,
    /// Authentication process is in progress
    Authenticating = 3,
    /// Authentication completed successfully
    Authenticated = 4,
    /// Subscribing to private channels (myOrder, myAsset)
    Subscribing = 5,
    /// Successfully subscribed to all required channels
    Subscribed = 6,
    /// WebSocket disconnection is in progress
    Disconnecting = 7,
}

impl From<u8> for UpbitConnectionState {
    fn from(value: u8) -> Self {
        match value {
            0 => Self::Disconnected,
            1 => Self::Connecting,
            2 => Self::Connected,
            3 => Self::Authenticating,
            4 => Self::Authenticated,
            5 => Self::Subscribing,
            6 => Self::Subscribed,
            7 => Self::Disconnecting,
            _ => Self::Disconnected,
        }
    }
}

impl From<UpbitConnectionState> for WsConnectionState {
    fn from(state: UpbitConnectionState) -> Self {
        match state {
            UpbitConnectionState::Disconnected => Self::Disconnected,
            UpbitConnectionState::Connecting => Self::Connecting,
            UpbitConnectionState::Connected => Self::Connected,
            UpbitConnectionState::Authenticating => Self::Connecting,
            UpbitConnectionState::Authenticated => Self::Connected,
            UpbitConnectionState::Subscribing => Self::Connected,
            UpbitConnectionState::Subscribed => Self::Connected,
            UpbitConnectionState::Disconnecting => Self::Disconnected,
        }
    }
}

/// Connection health information
#[derive(Debug, Clone)]
pub struct ConnectionHealth {
    /// Current connection state
    pub state: UpbitConnectionState,
    /// Whether the WebSocket connection is established
    pub is_connected: bool,
    /// Whether authentication is completed
    pub is_authenticated: bool,
    /// Whether subscribed to private channels
    pub is_subscribed: bool,
    /// Time elapsed since last ping was sent
    pub time_since_last_ping: Option<Duration>,
    /// Time elapsed since last pong was received
    pub time_since_last_pong: Option<Duration>,
    /// Time elapsed since authentication completed
    pub time_since_auth: Option<Duration>,
    /// Time remaining until JWT expires (not applicable for Upbit)
    pub jwt_expires_in: Option<Duration>,
    /// Total number of messages sent
    pub messages_sent: u64,
    /// Total number of messages received
    pub messages_received: u64,
    /// Number of reconnection attempts made
    pub reconnection_attempts: u64,
    /// Number of successful reconnections
    pub successful_reconnections: u64,
    /// Number of failed reconnections
    pub failed_reconnections: u64,
}

/// Connection metrics for monitoring
#[derive(Debug, Default)]
struct ConnectionMetrics {
    /// Number of ping frames sent
    pings_sent: AtomicU64,
    /// Number of pong frames received
    pongs_received: AtomicU64,
    /// Total number of messages sent
    messages_sent: AtomicU64,
    /// Total number of messages received
    messages_received: AtomicU64,
    /// Number of reconnection attempts made
    reconnection_attempts: AtomicU64,
    /// Number of successful reconnections
    successful_reconnections: AtomicU64,
    /// Number of failed reconnections
    failed_reconnections: AtomicU64,
}

/// Task handles for lifecycle management
#[derive(Debug, Default)]
struct TaskHandles {
    /// Handle for the ping task that sends periodic ping frames
    ping_handle: Option<JoinHandle<()>>,
    /// Handle for the message processing task that handles incoming messages
    message_handler: Option<JoinHandle<()>>,
}

/// Upbit WebSocket Trading Client
pub struct UpbitWebSocketTrader {
    /// Authentication handler
    auth: Arc<UpbitAuth>,

    /// WebSocket connection for private data
    ws_sink: Arc<AsyncRwLock<Option<WebSocketSink>>>,
    ws_stream: Arc<AsyncRwLock<Option<WebSocketStream>>>,

    /// Connection state machine
    state: Arc<AtomicU8>,

    /// High-precision clock (using monotonic time)
    clock: Clock,

    /// Last ping sent timestamp (nanoseconds, monotonic)
    last_ping_time: Arc<AtomicU64>,

    /// Last pong received timestamp (nanoseconds, monotonic)
    last_pong_time: Arc<AtomicU64>,

    /// Authentication completed timestamp (nanoseconds, monotonic)
    auth_completed_time: Arc<AtomicU64>,

    /// Task handles for lifecycle management
    task_handles: Arc<AsyncRwLock<TaskHandles>>,

    /// Connection metrics
    metrics: Arc<ConnectionMetrics>,

    /// Reconnection backoff state
    reconnection_backoff_ms: Arc<AtomicU64>,
}

impl Clone for UpbitWebSocketTrader {
    fn clone(&self) -> Self {
        Self {
            auth: self.auth.clone(),
            ws_sink: self.ws_sink.clone(),
            ws_stream: self.ws_stream.clone(),
            state: self.state.clone(),
            clock: self.clock.clone(),
            last_ping_time: self.last_ping_time.clone(),
            last_pong_time: self.last_pong_time.clone(),
            auth_completed_time: self.auth_completed_time.clone(),
            task_handles: self.task_handles.clone(),
            metrics: self.metrics.clone(),
            reconnection_backoff_ms: self.reconnection_backoff_ms.clone(),
        }
    }
}

impl UpbitWebSocketTrader {
    /// Create a new Upbit WebSocket trading client
    #[must_use]
    pub fn new(auth: Arc<UpbitAuth>) -> Self {
        Self {
            auth,
            ws_sink: Arc::new(AsyncRwLock::new(None)),
            ws_stream: Arc::new(AsyncRwLock::new(None)),
            state: Arc::new(AtomicU8::new(UpbitConnectionState::Disconnected as u8)),
            clock: Clock::new(),
            last_ping_time: Arc::new(AtomicU64::new(0)),
            last_pong_time: Arc::new(AtomicU64::new(0)),
            auth_completed_time: Arc::new(AtomicU64::new(0)),
            task_handles: Arc::new(AsyncRwLock::new(TaskHandles::default())),
            metrics: Arc::new(ConnectionMetrics::default()),
            reconnection_backoff_ms: Arc::new(AtomicU64::new(1000)), // Start with 1 second
        }
    }

    /// Get current connection state
    fn get_state(&self) -> UpbitConnectionState {
        UpbitConnectionState::from(self.state.load(Ordering::Acquire))
    }

    /// Set connection state
    fn set_state(&self, new_state: UpbitConnectionState) {
        let old_state = self.get_state();
        self.state.store(new_state as u8, Ordering::Release);

        if old_state != new_state {
            debug!("Connection state transition: {old_state:?} -> {new_state:?}");
        }
    }

    /// Check if WebSocket is connected
    #[must_use]
    pub fn is_connected(&self) -> bool {
        matches!(
            self.get_state(),
            UpbitConnectionState::Connected
                | UpbitConnectionState::Authenticating
                | UpbitConnectionState::Authenticated
                | UpbitConnectionState::Subscribing
                | UpbitConnectionState::Subscribed
        )
    }

    /// Check if WebSocket is authenticated
    #[must_use]
    pub fn is_authenticated(&self) -> bool {
        matches!(
            self.get_state(),
            UpbitConnectionState::Authenticated
                | UpbitConnectionState::Subscribing
                | UpbitConnectionState::Subscribed
        )
    }

    /// Check if subscribed to private channels
    #[must_use]
    pub fn is_subscribed(&self) -> bool {
        self.get_state() == UpbitConnectionState::Subscribed
    }

    /// Get detailed connection health information
    #[must_use]
    pub fn get_connection_health(&self) -> ConnectionHealth {
        let now = self.clock.raw();
        let last_ping = self.last_ping_time.load(Ordering::Acquire);
        let last_pong = self.last_pong_time.load(Ordering::Acquire);
        let auth_time = self.auth_completed_time.load(Ordering::Acquire);

        let time_since_last_ping = if last_ping > 0 {
            Some(Duration::from_nanos(now.saturating_sub(last_ping)))
        } else {
            None
        };

        let time_since_last_pong = if last_pong > 0 {
            Some(Duration::from_nanos(now.saturating_sub(last_pong)))
        } else {
            None
        };

        let time_since_auth = if auth_time > 0 {
            Some(Duration::from_nanos(now.saturating_sub(auth_time)))
        } else {
            None
        };

        ConnectionHealth {
            state: self.get_state(),
            is_connected: self.is_connected(),
            is_authenticated: self.is_authenticated(),
            is_subscribed: self.is_subscribed(),
            time_since_last_ping,
            time_since_last_pong,
            time_since_auth,
            jwt_expires_in: None, // JWT doesn't have explicit expiration in Upbit
            messages_sent: self.metrics.messages_sent.load(Ordering::Relaxed),
            messages_received: self.metrics.messages_received.load(Ordering::Relaxed),
            reconnection_attempts: self.metrics.reconnection_attempts.load(Ordering::Relaxed),
            successful_reconnections: self
                .metrics
                .successful_reconnections
                .load(Ordering::Relaxed),
            failed_reconnections: self.metrics.failed_reconnections.load(Ordering::Relaxed),
        }
    }

    /// Connect to Upbit private WebSocket API
    pub async fn connect(&self, report_tx: Sender<ExecutionReport>) -> Result<()> {
        info!("Connecting to Upbit private WebSocket: {UPBIT_PRIVATE_WS_URL}");

        // Set connecting state
        self.set_state(UpbitConnectionState::Connecting);

        // Generate JWT token for WebSocket authentication
        let jwt = self.auth.generate_websocket_authentication()?;

        // Create WebSocket configuration with JWT in Authorization header
        let config = WebSocketConfig::builder(Exchange::Upbit, UPBIT_PRIVATE_WS_URL.to_string())
            .connect_timeout(Duration::from_secs(10))
            .timeout(Duration::from_secs(30))
            .ping_interval(Duration::from_secs(PING_INTERVAL_SECONDS))
            .pong_timeout(Duration::from_secs(PONG_TIMEOUT_SECONDS))
            .max_frame_size(65536) // 64KB
            .max_message_size(MAX_MESSAGE_SIZE)
            .header("Authorization".to_string(), format!("Bearer {jwt}"))
            .build();

        let stats = new_shared_stats();
        let connection_state = Arc::new(RwLock::new(WsConnectionState::Disconnected));
        let mut connector =
            WebSocketConnector::new(config, stats.clone(), connection_state.clone());

        let (sink, stream) = connector.connect_with_retry(UPBIT_PRIVATE_WS_URL).await?;

        // Store connection references
        *self.ws_sink.write().await = Some(sink);
        *self.ws_stream.write().await = Some(stream);

        // Update state
        self.set_state(UpbitConnectionState::Connected);

        // Subscribe to private channels (authentication is done via headers)
        self.subscribe_to_private_channels().await?;

        // Start background tasks
        self.start_ping_task().await;
        self.start_message_task(report_tx).await;

        info!("Successfully connected and subscribed to Upbit private WebSocket");
        Ok(())
    }

    /// Subscribe to Upbit private channels (myOrder and myAsset)
    async fn subscribe_to_private_channels(&self) -> Result<()> {
        debug!("Subscribing to Upbit private channels");

        self.set_state(UpbitConnectionState::Subscribing);

        // Create subscription for myOrder channel
        let order_subscription = simd_json::json!([
            {
                "ticket": format!("order_{}", Uuid::new_v4())
            },
            {
                "type": "myOrder"
            },
            {
                "format": "DEFAULT" // Can also be "SIMPLE" for abbreviated field names
            }
        ]);

        // Create subscription for myAsset channel
        let asset_subscription = simd_json::json!([
            {
                "ticket": format!("asset_{}", Uuid::new_v4())
            },
            {
                "type": "myAsset"
            },
            {
                "format": "DEFAULT"
            }
        ]);

        // Send order subscription (authentication already done via headers)
        let order_message: SmartString = simd_json::to_string(&order_subscription)?.into();
        self.send_message(Message::Text(order_message)).await?;

        // Send asset subscription (authentication already done via headers)
        let asset_message: SmartString = simd_json::to_string(&asset_subscription)?.into();
        self.send_message(Message::Text(asset_message)).await?;

        // Mark as subscribed
        self.set_state(UpbitConnectionState::Subscribed);
        self.auth_completed_time
            .store(self.clock.raw(), Ordering::Release);

        debug!("Upbit private channels subscription completed");
        Ok(())
    }

    // Removed send_authenticated_message method - authentication is now handled via headers

    /// Send a message through the WebSocket connection
    async fn send_message(&self, message: Message) -> Result<()> {
        if let Some(sink) = self.ws_sink.write().await.as_mut() {
            let frame = message.to_frame_view();
            sink.send(frame).await?;
            self.metrics.messages_sent.fetch_add(1, Ordering::Relaxed);
            Ok(())
        } else {
            bail!("WebSocket sink not available")
        }
    }

    /// Start ping task to keep connection alive
    async fn start_ping_task(&self) {
        let sink = self.ws_sink.clone();
        let last_ping = self.last_ping_time.clone();
        let clock = self.clock.clone();
        let metrics = self.metrics.clone();

        let task = tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(PING_INTERVAL_SECONDS));

            loop {
                interval.tick().await;

                // Send ping frame
                if let Some(sink) = sink.write().await.as_mut() {
                    let ping_message = Message::Ping(Vec::new());
                    let frame = ping_message.to_frame_view();
                    if let Err(e) = sink.send(frame).await {
                        error!("Failed to send ping: {e}");
                        break;
                    }

                    last_ping.store(clock.raw(), Ordering::Release);
                    metrics.pings_sent.fetch_add(1, Ordering::Relaxed);
                } else {
                    break;
                }
            }
        });

        self.task_handles.write().await.ping_handle = Some(task);
    }

    /// Start message processing task
    async fn start_message_task(&self, report_tx: Sender<ExecutionReport>) {
        let stream = self.ws_stream.clone();
        let last_pong_time = self.last_pong_time.clone();
        let clock = self.clock.clone();
        let metrics = self.metrics.clone();

        let task = tokio::spawn(async move {
            loop {
                // Take a lock scope to get the stream
                let mut stream_guard = stream.write().await;
                if let Some(ws_stream) = stream_guard.as_mut() {
                    if let Some(frame) = ws_stream.next().await {
                        drop(stream_guard); // Release the lock before processing

                        let message = Message::from_frame_view(frame);
                        metrics.messages_received.fetch_add(1, Ordering::Relaxed);

                        match message {
                            Message::Text(text) => {
                                debug!("Received text message: {text}");
                                if let Err(e) =
                                    Self::process_message(&text, &report_tx, &clock).await
                                {
                                    error!("Failed to process message: {e}");
                                }
                            }
                            Message::Pong(_) => {
                                last_pong_time.store(clock.raw(), Ordering::Release);
                                metrics.pongs_received.fetch_add(1, Ordering::Relaxed);
                            }
                            Message::Close(_) => {
                                warn!("Received close message from Upbit WebSocket");
                                break;
                            }
                            _ => {
                                debug!("Received other message type: {message:?}");
                            }
                        }
                    } else {
                        break; // Stream ended
                    }
                } else {
                    break; // No stream available
                }
            }
        });

        self.task_handles.write().await.message_handler = Some(task);
    }

    /// Process incoming WebSocket messages
    async fn process_message(
        text: &str,
        report_tx: &Sender<ExecutionReport>,
        clock: &Clock,
    ) -> Result<()> {
        // Parse JSON message
        let json_str = text.to_string();
        let mut json_bytes = json_str.into_bytes();
        let json: simd_json::OwnedValue = simd_json::from_slice(&mut json_bytes)?;

        // Determine message type
        if let Some(msg_type) = json.get("type").and_then(|v| v.as_str()) {
            match msg_type {
                "myOrder" => {
                    Self::handle_order_update(&json, report_tx, clock).await?;
                }
                "myAsset" => {
                    Self::handle_asset_update(&json).await?;
                }
                _ => {
                    debug!("Unknown message type: {msg_type}");
                }
            }
        } else {
            debug!("Message without type field: {text}");
        }

        Ok(())
    }

    /// Handle myOrder updates
    async fn handle_order_update(
        json: &simd_json::OwnedValue,
        report_tx: &Sender<ExecutionReport>,
        clock: &Clock,
    ) -> Result<()> {
        debug!("Processing order update: {json:?}");

        let order_id = json
            .get("uuid")
            .and_then(|v| v.as_str())
            .unwrap_or("unknown")
            .into();

        let market = json
            .get("code")
            .and_then(|v| v.as_str())
            .unwrap_or("UNKNOWN");

        let state = json
            .get("state")
            .and_then(|v| v.as_str())
            .unwrap_or("unknown");

        let status = Self::map_order_status(state);

        let executed_volume = json
            .get("executed_volume")
            .and_then(simd_json::prelude::ValueAsScalar::as_f64)
            .and_then(|f| Decimal::try_from(f).ok())
            .unwrap_or(Decimal::ZERO);

        let remaining_volume = json
            .get("remaining_volume")
            .and_then(simd_json::prelude::ValueAsScalar::as_f64)
            .and_then(|f| Decimal::try_from(f).ok())
            .unwrap_or(Decimal::ZERO);

        let execution_price = json
            .get("price")
            .and_then(simd_json::prelude::ValueAsScalar::as_f64)
            .and_then(|f| Decimal::try_from(f).ok());

        let order_timestamp = json
            .get("order_timestamp")
            .and_then(simd_json::prelude::ValueAsScalar::as_u64)
            .map_or(0, |ts| ts * 1_000_000);

        let trade_timestamp = json
            .get("trade_timestamp")
            .and_then(simd_json::prelude::ValueAsScalar::as_u64)
            .map_or(0, |ts| ts * 1_000_000);

        let report = ExecutionReport {
            id: id_generation::generate_exchange_order_id("upbit"),
            order_id,
            exchange_timestamp: if trade_timestamp > 0 {
                trade_timestamp
            } else if order_timestamp > 0 {
                order_timestamp
            } else {
                // Fallback to system time if both exchange timestamps are unavailable
                timestamp_nanos()
            },
            system_timestamp: timestamp_nanos(),
            instrument_id: InstrumentId::new(market, Venue::Upbit),
            status,
            filled_quantity: executed_volume,
            remaining_quantity: remaining_volume,
            execution_price,
            reject_reason: None,
            exchange_execution_id: json
                .get("trade_uuid")
                .and_then(|v| v.as_str())
                .map(std::convert::Into::into),
            is_final: matches!(
                status,
                OrderStatus::Filled | OrderStatus::Cancelled | OrderStatus::Rejected
            ),
        };

        if let Err(e) = report_tx.try_send(report) {
            error!("Failed to send order update report: {e}");
        }

        Ok(())
    }

    /// Handle myAsset updates
    async fn handle_asset_update(json: &simd_json::OwnedValue) -> Result<()> {
        debug!("Processing asset update: {json:?}");

        // For now, we'll just log asset updates
        // In a full implementation, this could update a portfolio manager
        if let Some(assets) = json.get("assets").and_then(|v| v.as_array()) {
            for asset in assets {
                if let (Some(currency), Some(balance), Some(locked)) = (
                    asset.get("currency").and_then(|v| v.as_str()),
                    asset
                        .get("balance")
                        .and_then(simd_json::prelude::ValueAsScalar::as_f64),
                    asset
                        .get("locked")
                        .and_then(simd_json::prelude::ValueAsScalar::as_f64),
                ) {
                    info!("Asset update - {currency}: balance={balance}, locked={locked}");
                }
            }
        }

        Ok(())
    }

    /// Map Upbit order status to internal `OrderStatus`
    fn map_order_status(status: &str) -> OrderStatus {
        match status {
            "wait" => OrderStatus::New,
            "watch" => OrderStatus::New, // Reserved order waiting
            "trade" => OrderStatus::PartiallyFilled, // Partial execution occurred
            "done" => OrderStatus::Filled,
            "cancel" => OrderStatus::Cancelled,
            _ => OrderStatus::Rejected,
        }
    }

    /// Disconnect from WebSocket
    pub async fn disconnect(&self) -> Result<()> {
        info!("Disconnecting from Upbit WebSocket");

        self.set_state(UpbitConnectionState::Disconnecting);

        // Cancel background tasks
        {
            let mut handles = self.task_handles.write().await;
            if let Some(handle) = handles.ping_handle.take() {
                handle.abort();
            }
            if let Some(handle) = handles.message_handler.take() {
                handle.abort();
            }
        }

        // Close WebSocket connections
        *self.ws_sink.write().await = None;
        *self.ws_stream.write().await = None;

        self.set_state(UpbitConnectionState::Disconnected);

        info!("Disconnected from Upbit WebSocket");
        Ok(())
    }
}

#[async_trait]
impl crate::execution_engine::Exchange for UpbitWebSocketTrader {
    fn venue(&self) -> Venue {
        Venue::Upbit
    }

    async fn place_order(
        &self,
        _order: Order,
        _report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        // Upbit WebSocket doesn't support order placement - orders must be placed via REST API
        // The existing UpbitExchange handles REST order placement
        // This WebSocket client focuses on real-time updates only
        bail!("Order placement via WebSocket not supported by Upbit - use REST API")
    }

    async fn cancel_order(
        &self,
        _order_id: SmartString,
        _report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        // Upbit WebSocket doesn't support order cancellation - must use REST API
        bail!("Order cancellation via WebSocket not supported by Upbit - use REST API")
    }

    async fn modify_order(
        &self,
        _order_id: SmartString,
        _new_price: Option<Decimal>,
        _new_quantity: Option<Decimal>,
        _report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        bail!("Order modification not supported by Upbit")
    }

    async fn cancel_all_orders(
        &self,
        _instrument_id: Option<InstrumentId>,
        _report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        bail!("Cancel all orders via WebSocket not supported by Upbit - use REST API")
    }

    async fn get_order_status(&self, _order_id: &str) -> Result<OrderStatus> {
        bail!("Order status query via WebSocket not supported by Upbit - use REST API")
    }

    async fn connect(&self, report_sender: Sender<ExecutionReport>) -> Result<()> {
        // Now properly implemented with report_sender parameter
        // Call the inherent method directly to avoid recursion
        Self::connect(self, report_sender).await
    }

    async fn disconnect(&self) -> Result<()> {
        self.disconnect().await
    }

    async fn is_connected(&self) -> bool {
        self.is_connected()
    }

    async fn get_instruments(&self) -> Result<smallvec::SmallVec<[InstrumentId; 32]>> {
        bail!("Instrument retrieval via WebSocket not supported by Upbit - use REST API")
    }

    async fn send_fix_message(&self, _message: Vec<u8>) -> Result<()> {
        anyhow::bail!("FIX protocol not supported on Upbit WebSocket")
    }

    async fn receive_fix_message(&self) -> Result<Vec<u8>> {
        anyhow::bail!("FIX protocol not supported on Upbit WebSocket")
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rusty_common::auth::exchanges::upbit::UpbitAuthConfig;
    use rusty_model::enums::OrderSide;
    use simd_json::ValueBuilder;

    fn create_test_auth() -> Arc<UpbitAuth> {
        let config = UpbitAuthConfig::new("test_access_key".into(), "test_secret_key".into());
        Arc::new(UpbitAuth::new(config))
    }

    #[test]
    fn test_upbit_websocket_trader_creation() {
        let auth = create_test_auth();
        let trader = UpbitWebSocketTrader::new(auth);

        assert_eq!(trader.get_state(), UpbitConnectionState::Disconnected);
        assert!(!trader.is_connected());
        assert!(!trader.is_authenticated());
        assert!(!trader.is_subscribed());
    }

    #[test]
    fn test_connection_state_transitions() {
        let auth = create_test_auth();
        let trader = UpbitWebSocketTrader::new(auth);

        // Test state transitions
        trader.set_state(UpbitConnectionState::Connecting);
        assert_eq!(trader.get_state(), UpbitConnectionState::Connecting);

        trader.set_state(UpbitConnectionState::Connected);
        assert_eq!(trader.get_state(), UpbitConnectionState::Connected);
        assert!(trader.is_connected());

        trader.set_state(UpbitConnectionState::Subscribing);
        assert_eq!(trader.get_state(), UpbitConnectionState::Subscribing);
        assert!(trader.is_connected());
        assert!(trader.is_authenticated());

        trader.set_state(UpbitConnectionState::Subscribed);
        assert_eq!(trader.get_state(), UpbitConnectionState::Subscribed);
        assert!(trader.is_subscribed());
    }

    #[test]
    fn test_order_status_mapping() {
        assert_eq!(
            UpbitWebSocketTrader::map_order_status("wait"),
            OrderStatus::New
        );
        assert_eq!(
            UpbitWebSocketTrader::map_order_status("watch"),
            OrderStatus::New
        );
        assert_eq!(
            UpbitWebSocketTrader::map_order_status("trade"),
            OrderStatus::PartiallyFilled
        );
        assert_eq!(
            UpbitWebSocketTrader::map_order_status("done"),
            OrderStatus::Filled
        );
        assert_eq!(
            UpbitWebSocketTrader::map_order_status("cancel"),
            OrderStatus::Cancelled
        );
        assert_eq!(
            UpbitWebSocketTrader::map_order_status("unknown"),
            OrderStatus::Rejected
        );
    }

    #[test]
    fn test_connection_health() {
        let auth = create_test_auth();
        let trader = UpbitWebSocketTrader::new(auth);

        let health = trader.get_connection_health();
        assert_eq!(health.state, UpbitConnectionState::Disconnected);
        assert!(!health.is_connected);
        assert!(!health.is_authenticated);
        assert!(!health.is_subscribed);
        assert!(health.time_since_last_ping.is_none());
        assert_eq!(health.messages_sent, 0);
        assert_eq!(health.reconnection_attempts, 0);
    }

    #[tokio::test]
    async fn test_websocket_only_operations() {
        use crate::execution_engine::Exchange;
        use rusty_model::types::OrderId;

        let auth = create_test_auth();
        let trader = UpbitWebSocketTrader::new(auth);
        let (report_tx, _report_rx) = flume::bounded(100);

        // Test that order operations are not supported via WebSocket
        let order = Order {
            id: OrderId::new(),
            symbol: "KRW-BTC".into(),
            side: OrderSide::Buy,
            order_type: rusty_model::enums::OrderType::Limit,
            quantity: Decimal::from(1),
            price: Some(Decimal::from(50000)),
            stop_price: None,
            exchange_order_id: None,
            venue: Venue::Upbit,
            client_id: "test_client".into(),
            creation_time_ns: 1000000000,
            update_time_ns: 1000000000,
            average_fill_price: None,
            filled_quantity: Decimal::ZERO,
            status: rusty_model::enums::OrderStatus::New,
            time_in_force: rusty_model::enums::TimeInForce::GTC,
            metadata: simd_json::value::owned::Value::null(),
        };

        assert!(trader.place_order(order, report_tx.clone()).await.is_err());
        assert!(
            trader
                .cancel_order("test_id".into(), report_tx)
                .await
                .is_err()
        );
        assert!(trader.get_order_status("test_id").await.is_err());
    }
}