asterdex-sdk 0.1.3

// Integration tests for WebSocket streams (market data + user data).
//
// Run: cargo test --test aster_websocket -- --include-ignored --test-threads=1
//
// All tests are #[ignore]. They open real WebSocket connections to
// wss://fstream.asterdex-testnet.com.
//
// MUST use --test-threads=1: tests that also call the REST API share EIP-712
// nonce state through the same Credentials instance.

use std::time::Duration;
use tokio::time::timeout;
use asterdex_sdk::{
    AsterDexError, CancelOrderParams, PlaceOrderParams, FuturesClient, UserDataEvent,
    WebSocketClient, WebSocketMessage, WebSocketStream,
};

const WS_BASE_URL_DEFAULT: &str = "wss://fstream.asterdex-testnet.com";

/// Timeout for receiving a single WS message from high-frequency streams.
const MSG_TIMEOUT_SECS: u64 = 15;

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

fn build_ws_client() -> WebSocketClient {
    let _ = dotenvy::dotenv();
    let url = std::env::var("ASTERDEX_WS_BASE_URL")
        .unwrap_or_else(|_| WS_BASE_URL_DEFAULT.to_string());
    WebSocketClient::new(&url).expect("failed to build WebSocketClient")
}

fn build_rest_client() -> FuturesClient {
    let _ = dotenvy::dotenv();
    FuturesClient::from_env().expect("FuturesClient::from_env() failed — check .env")
}

/// Receive messages from a `WebSocketStream`, skipping subscription-confirmation
/// `Unknown` messages, and return the first typed `WebSocketMessage`.
/// Returns `None` on timeout, stream close, or error.
async fn recv_typed(stream: &mut WebSocketStream, timeout_secs: u64) -> Option<WebSocketMessage> {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs);
    loop {
        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
        if remaining.is_zero() {
            eprintln!("recv_typed: timed out after {timeout_secs}s");
            return None;
        }
        match timeout(remaining, stream.recv()).await {
            Ok(Some(Ok(raw))) => {
                match serde_json::from_value::<WebSocketMessage>(raw.clone()) {
                    Ok(WebSocketMessage::Unknown(_)) => {
                        // Subscription confirmation {"result":null,"id":1} or heartbeat — skip
                        continue;
                    }
                    Ok(msg) => return Some(msg),
                    Err(e) => {
                        eprintln!("recv_typed: parse error: {e} — raw: {raw}");
                        continue;
                    }
                }
            }
            Ok(Some(Err(e))) => {
                eprintln!("recv_typed: stream error: {e:?}");
                return None;
            }
            Ok(None) => {
                eprintln!("recv_typed: stream closed");
                return None;
            }
            Err(_elapsed) => {
                eprintln!("recv_typed: timed out after {timeout_secs}s");
                return None;
            }
        }
    }
}

/// Same as `recv_typed` but receives raw `serde_json::Value` (no WebSocketMessage parsing).
async fn recv_raw(
    stream: &mut WebSocketStream,
    timeout_secs: u64,
) -> Option<serde_json::Value> {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs);
    loop {
        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
        if remaining.is_zero() {
            eprintln!("recv_raw: timed out after {timeout_secs}s");
            return None;
        }
        match timeout(remaining, stream.recv()).await {
            Ok(Some(Ok(raw))) => {
                // Skip subscription confirmations
                if raw.get("result").is_some() || raw.get("id").is_some() {
                    continue;
                }
                return Some(raw);
            }
            Ok(Some(Err(e))) => {
                eprintln!("recv_raw: stream error: {e:?}");
                return None;
            }
            Ok(None) => return None,
            Err(_) => return None,
        }
    }
}

// ===========================================================================
// SECTION 1 — Connection
// ===========================================================================

/// Connect to a single stream and verify a raw message arrives within the timeout.
#[tokio::test]
#[ignore]
async fn test_ws_connect_receives_message() {
    let client = build_ws_client();
    let mut stream = client
        .subscribe(vec!["btcusdt@markPrice"])
        .await
        .expect("subscribe to btcusdt@markPrice failed");

    let msg = recv_raw(&mut stream, MSG_TIMEOUT_SECS).await;
    assert!(
        msg.is_some(),
        "no message received from btcusdt@markPrice within {MSG_TIMEOUT_SECS}s"
    );
}

// ===========================================================================
// SECTION 2 — Market Data Streams — typed messages
// ===========================================================================

/// markPrice stream → deserializes as WebSocketMessage::MarkPrice.
///
/// Emits every 3 s — very reliable for testing.
#[tokio::test]
#[ignore]
async fn test_ws_mark_price_stream() {
    let client = build_ws_client();
    let mut stream = client
        .subscribe(vec!["btcusdt@markPrice"])
        .await
        .expect("subscribe to btcusdt@markPrice failed");

    let msg = recv_typed(&mut stream, MSG_TIMEOUT_SECS).await;
    let msg = msg.expect("no message received within timeout from btcusdt@markPrice");

    match msg {
        WebSocketMessage::MarkPrice(event) => {
            assert_eq!(event.symbol, "BTCUSDT");
            let price: f64 = event.mark_price.parse().expect("mark_price not f64");
            assert!(price > 0.0, "mark_price must be positive");
            assert!(event.next_funding_time > 0);
        }
        other => panic!("expected MarkPrice, got: {other:?}"),
    }
}

/// miniTicker stream → deserializes as WebSocketMessage::MiniTicker.
///
/// Emits every 1 s — reliable.
#[tokio::test]
#[ignore]
async fn test_ws_mini_ticker_stream() {
    let client = build_ws_client();
    let mut stream = client
        .subscribe(vec!["btcusdt@miniTicker"])
        .await
        .expect("subscribe to btcusdt@miniTicker failed");

    let msg = recv_typed(&mut stream, MSG_TIMEOUT_SECS).await;
    let msg = msg.expect("no message received within timeout from btcusdt@miniTicker");

    match msg {
        WebSocketMessage::MiniTicker(event) => {
            assert_eq!(event.symbol, "BTCUSDT");
            let close: f64 = event.close_price.parse().expect("close_price not f64");
            assert!(close > 0.0, "close price must be positive");
        }
        other => panic!("expected MiniTicker, got: {other:?}"),
    }
}

/// ticker stream → deserializes as WebSocketMessage::Ticker.
///
/// Emits every 1 s.
#[tokio::test]
#[ignore]
async fn test_ws_ticker_stream() {
    let client = build_ws_client();
    let mut stream = client
        .subscribe(vec!["btcusdt@ticker"])
        .await
        .expect("subscribe to btcusdt@ticker failed");

    let msg = recv_typed(&mut stream, MSG_TIMEOUT_SECS).await;
    let msg = msg.expect("no message received within timeout from btcusdt@ticker");

    match msg {
        WebSocketMessage::Ticker(event) => {
            assert_eq!(event.symbol, "BTCUSDT");
            let last: f64 = event.last_price.parse().expect("last_price not f64");
            assert!(last > 0.0);
            assert!(event.trade_count > 0);
        }
        other => panic!("expected Ticker, got: {other:?}"),
    }
}

/// aggTrade stream → deserializes as WebSocketMessage::AggTrade.
///
/// Event-driven — relies on at least one trade happening on testnet.
/// Test is skipped gracefully on timeout (low-activity testnet).
#[tokio::test]
#[ignore]
async fn test_ws_agg_trade_stream() {
    let client = build_ws_client();
    let mut stream = client
        .subscribe(vec!["btcusdt@aggTrade"])
        .await
        .expect("subscribe to btcusdt@aggTrade failed");

    match recv_typed(&mut stream, MSG_TIMEOUT_SECS).await {
        None => eprintln!("btcusdt@aggTrade: no message within {MSG_TIMEOUT_SECS}s — low-activity testnet, skipping"),
        Some(WebSocketMessage::AggTrade(event)) => {
            assert_eq!(event.symbol, "BTCUSDT");
            let price: f64 = event.price.parse().expect("price not f64");
            assert!(price > 0.0);
            assert!(event.agg_trade_id > 0);
        }
        Some(other) => panic!("expected AggTrade, got: {other:?}"),
    }
}

/// bookTicker stream → deserializes as WebSocketMessage::BookTicker.
///
/// Updates on every order book change.
#[tokio::test]
#[ignore]
async fn test_ws_book_ticker_stream() {
    let client = build_ws_client();
    let mut stream = client
        .subscribe(vec!["btcusdt@bookTicker"])
        .await
        .expect("subscribe to btcusdt@bookTicker failed");

    match recv_typed(&mut stream, MSG_TIMEOUT_SECS).await {
        None => eprintln!("btcusdt@bookTicker: no message within {MSG_TIMEOUT_SECS}s — skipping"),
        Some(WebSocketMessage::BookTicker(event)) => {
            assert_eq!(event.symbol, "BTCUSDT");
            let bid: f64 = event.best_bid_price.parse().expect("bid not f64");
            let ask: f64 = event.best_ask_price.parse().expect("ask not f64");
            assert!(bid > 0.0 && ask > 0.0);
            assert!(bid < ask, "bid {bid} >= ask {ask}");
        }
        Some(other) => panic!("expected BookTicker, got: {other:?}"),
    }
}

/// kline_1m stream → deserializes as WebSocketMessage::Kline.
///
/// Emits on each trade (may be infrequent on low-activity testnet).
#[tokio::test]
#[ignore]
async fn test_ws_kline_stream() {
    let client = build_ws_client();
    let mut stream = client
        .subscribe(vec!["btcusdt@kline_1m"])
        .await
        .expect("subscribe to btcusdt@kline_1m failed");

    match recv_typed(&mut stream, MSG_TIMEOUT_SECS).await {
        None => eprintln!("btcusdt@kline_1m: no message within {MSG_TIMEOUT_SECS}s — skipping"),
        Some(WebSocketMessage::Kline(event)) => {
            assert_eq!(event.symbol, "BTCUSDT");
            assert_eq!(event.kline.interval, "1m");
            let open: f64 = event.kline.open.parse().expect("open not f64");
            assert!(open > 0.0);
        }
        Some(other) => panic!("expected Kline, got: {other:?}"),
    }
}

/// depth5 stream → deserializes as `PartialDepth` (Binance format: no `"e"` field) or
/// `DiffDepth` (AsterDex testnet sends `"e":"depthUpdate"` for this stream).
/// Both variants carry bids and asks — verify the spread is valid.
#[tokio::test]
#[ignore]
async fn test_ws_partial_depth_stream() {
    let client = build_ws_client();
    let mut stream = client
        .subscribe(vec!["btcusdt@depth5"])
        .await
        .expect("subscribe to btcusdt@depth5 failed");

    let (bids, asks) = match recv_typed(&mut stream, MSG_TIMEOUT_SECS).await {
        None => {
            eprintln!("btcusdt@depth5: no message within {MSG_TIMEOUT_SECS}s — skipping");
            return;
        }
        // Binance-style partial depth (no "e" key)
        Some(WebSocketMessage::PartialDepth(event)) => (event.bids, event.asks),
        // AsterDex testnet emits depthUpdate on depth5 stream
        Some(WebSocketMessage::DiffDepth(event)) => (event.bids, event.asks),
        Some(other) => panic!("expected PartialDepth or DiffDepth, got: {other:?}"),
    };

    assert!(!bids.is_empty(), "bids is empty");
    assert!(!asks.is_empty(), "asks is empty");
    let best_bid: f64 = bids[0][0].parse().expect("bid price not f64");
    let best_ask: f64 = asks[0][0].parse().expect("ask price not f64");
    assert!(best_bid > 0.0 && best_ask > 0.0);
    assert!(best_bid < best_ask, "bid {best_bid} >= ask {best_ask}");
}

/// depth stream (differential depth) → deserializes as WebSocketMessage::DiffDepth.
#[tokio::test]
#[ignore]
async fn test_ws_diff_depth_stream() {
    let client = build_ws_client();
    let mut stream = client
        .subscribe(vec!["btcusdt@depth"])
        .await
        .expect("subscribe to btcusdt@depth failed");

    match recv_typed(&mut stream, MSG_TIMEOUT_SECS).await {
        None => eprintln!("btcusdt@depth: no message within {MSG_TIMEOUT_SECS}s — skipping"),
        Some(WebSocketMessage::DiffDepth(event)) => {
            assert_eq!(event.symbol, "BTCUSDT");
            assert!(event.final_update_id >= event.first_update_id);
        }
        Some(other) => panic!("expected DiffDepth, got: {other:?}"),
    }
}

/// !markPrice@arr stream → deserializes as WebSocketMessage::AllMarkPrices.
///
/// Emits all-symbol mark prices as a JSON array every 3 s.
#[tokio::test]
#[ignore]
async fn test_ws_all_mark_prices_stream() {
    let client = build_ws_client();
    let mut stream = client
        .subscribe(vec!["!markPrice@arr"])
        .await
        .expect("subscribe to !markPrice@arr failed");

    let msg = recv_typed(&mut stream, MSG_TIMEOUT_SECS).await;
    let msg = msg.expect("no message from !markPrice@arr within timeout");

    match msg {
        WebSocketMessage::AllMarkPrices(prices) => {
            assert!(!prices.is_empty(), "AllMarkPrices array is empty");
            assert!(
                prices.iter().any(|mp| mp.symbol == "BTCUSDT"),
                "BTCUSDT not in AllMarkPrices"
            );
        }
        other => panic!("expected AllMarkPrices, got: {other:?}"),
    }
}

/// !miniTicker@arr stream → deserializes as WebSocketMessage::AllMiniTickers.
#[tokio::test]
#[ignore]
async fn test_ws_all_mini_tickers_stream() {
    let client = build_ws_client();
    let mut stream = client
        .subscribe(vec!["!miniTicker@arr"])
        .await
        .expect("subscribe to !miniTicker@arr failed");

    let msg = recv_typed(&mut stream, MSG_TIMEOUT_SECS).await;
    let msg = msg.expect("no message from !miniTicker@arr within timeout");

    match msg {
        WebSocketMessage::AllMiniTickers(tickers) => {
            assert!(!tickers.is_empty(), "AllMiniTickers array is empty");
            assert!(
                tickers.iter().any(|t| t.symbol == "BTCUSDT"),
                "BTCUSDT not in AllMiniTickers"
            );
        }
        other => panic!("expected AllMiniTickers, got: {other:?}"),
    }
}

// ===========================================================================
// SECTION 3 — Multi-stream subscription
// ===========================================================================

/// Subscribe to 3 streams simultaneously — verify all 3 deliver messages.
#[tokio::test]
#[ignore]
async fn test_ws_multi_stream_subscription() {
    let client = build_ws_client();
    // Use high-frequency streams for reliability
    let mut stream = client
        .subscribe(vec![
            "btcusdt@markPrice",
            "btcusdt@miniTicker",
            "btcusdt@ticker",
        ])
        .await
        .expect("multi-stream subscribe failed");

    let mut seen_mark_price = false;
    let mut seen_mini_ticker = false;
    let mut seen_ticker = false;

    let deadline = tokio::time::Instant::now() + Duration::from_secs(MSG_TIMEOUT_SECS);

    while !seen_mark_price || !seen_mini_ticker || !seen_ticker {
        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
        if remaining.is_zero() {
            break;
        }
        match timeout(remaining, stream.recv()).await {
            Ok(Some(Ok(raw))) => {
                match serde_json::from_value::<WebSocketMessage>(raw) {
                    Ok(WebSocketMessage::MarkPrice(_)) => seen_mark_price = true,
                    Ok(WebSocketMessage::MiniTicker(_)) => seen_mini_ticker = true,
                    Ok(WebSocketMessage::Ticker(_)) => seen_ticker = true,
                    Ok(WebSocketMessage::Unknown(_)) => {} // skip sub confirmation
                    Ok(_) => {}
                    Err(_) => {}
                }
            }
            _ => break,
        }
    }

    assert!(seen_mark_price, "did not receive MarkPrice in multi-stream");
    assert!(seen_mini_ticker, "did not receive MiniTicker in multi-stream");
    assert!(seen_ticker, "did not receive Ticker in multi-stream");
}

// ===========================================================================
// SECTION 4 — User Data Stream
// ===========================================================================

/// Connect to a user data stream and verify the connection succeeds.
#[tokio::test]
#[ignore]
async fn test_ws_user_data_stream_connects() {
    let rest = build_rest_client();

    // Create a fresh listen key
    let lk_resp = rest
        .create_listen_key()
        .await
        .expect("create_listen_key failed");
    let listen_key = lk_resp.data.listen_key.clone();

    // Connect
    let ws = build_ws_client();
    let mut stream = ws
        .subscribe_user_data(&listen_key)
        .await
        .expect("subscribe_user_data failed");

    // Just verify the connection is alive — try to receive with a short timeout
    // We don't require a message (no account activity yet).
    let _ = timeout(Duration::from_secs(3), stream.recv()).await;

    // Cleanup
    let _ = rest.delete_listen_key(&listen_key).await;
}

/// Full user-data flow: create listen key → connect → place order →
/// receive ORDER_TRADE_UPDATE → cancel order.
#[tokio::test]
#[ignore]
async fn test_ws_user_data_order_update_event() {
    let rest = build_rest_client();

    // Create listen key and connect BEFORE placing the order
    let lk_resp = rest
        .create_listen_key()
        .await
        .expect("create_listen_key failed");
    let listen_key = lk_resp.data.listen_key.clone();

    let ws = build_ws_client();
    let mut user_stream = ws
        .subscribe_user_data(&listen_key)
        .await
        .expect("subscribe_user_data failed");

    // Place a BTCUSDT LIMIT BUY (far below market so it rests in book)
    let order = rest
        .place_order(PlaceOrderParams {
            symbol: "BTCUSDT".to_string(),
            side: "BUY".to_string(),
            order_type: "LIMIT".to_string(),
            quantity: "0.001".to_string(),
            price: Some("10000.0".to_string()),
            time_in_force: Some("GTC".to_string()),
            ..Default::default()
        })
        .await
        .expect("place_order failed");
    let order_id = order.data.order_id;

    // Wait for the ORDER_TRADE_UPDATE event via next_event()
    let event_received = {
        let deadline = tokio::time::Instant::now() + Duration::from_secs(MSG_TIMEOUT_SECS);
        let mut found = false;
        loop {
            let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
            if remaining.is_zero() {
                eprintln!("test_ws_user_data_order_update_event: timeout waiting for event");
                break;
            }
            match timeout(remaining, user_stream.next_event()).await {
                Ok(Some(Ok(UserDataEvent::OrderUpdate(update)))) => {
                    if update.order.order_id == order_id {
                        assert_eq!(update.order.symbol, "BTCUSDT");
                        assert_eq!(update.order.side, "BUY");
                        assert_eq!(update.order.order_status, "NEW");
                        found = true;
                        break;
                    }
                }
                Ok(Some(Ok(UserDataEvent::Unknown(_)))) => continue,
                Ok(Some(Ok(_))) => continue,
                Ok(Some(Err(e))) => {
                    eprintln!("next_event error: {e:?}");
                    break;
                }
                Ok(None) => break,
                Err(_) => break,
            }
        }
        found
    };

    // Cleanup: cancel order and delete listen key
    let _ = rest
        .cancel_order(CancelOrderParams {
            symbol: "BTCUSDT".to_string(),
            order_id: Some(order_id),
            ..Default::default()
        })
        .await;
    let _ = rest.delete_listen_key(&listen_key).await;

    assert!(
        event_received,
        "ORDER_TRADE_UPDATE for order {order_id} was not received within {MSG_TIMEOUT_SECS}s"
    );
}

/// Place + cancel order and expect both NEW and CANCELED ORDER_TRADE_UPDATE events.
#[tokio::test]
#[ignore]
async fn test_ws_user_data_cancel_event() {
    let rest = build_rest_client();

    let lk_resp = rest
        .create_listen_key()
        .await
        .expect("create_listen_key failed");
    let listen_key = lk_resp.data.listen_key.clone();

    let ws = build_ws_client();
    let mut user_stream = ws
        .subscribe_user_data(&listen_key)
        .await
        .expect("subscribe_user_data failed");

    // Place order
    let order = rest
        .place_order(PlaceOrderParams {
            symbol: "BTCUSDT".to_string(),
            side: "BUY".to_string(),
            order_type: "LIMIT".to_string(),
            quantity: "0.001".to_string(),
            price: Some("10000.0".to_string()),
            time_in_force: Some("GTC".to_string()),
            ..Default::default()
        })
        .await
        .expect("place_order failed");
    let order_id = order.data.order_id;

    // Cancel order
    let _ = rest
        .cancel_order(CancelOrderParams {
            symbol: "BTCUSDT".to_string(),
            order_id: Some(order_id),
            ..Default::default()
        })
        .await
        .expect("cancel_order failed");

    // Expect to see CANCELED event within timeout
    let mut saw_canceled = false;
    let deadline = tokio::time::Instant::now() + Duration::from_secs(MSG_TIMEOUT_SECS);

    loop {
        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
        if remaining.is_zero() {
            eprintln!("test_ws_user_data_cancel_event: timeout");
            break;
        }
        match timeout(remaining, user_stream.next_event()).await {
            Ok(Some(Ok(UserDataEvent::OrderUpdate(update)))) => {
                if update.order.order_id == order_id
                    && update.order.order_status == "CANCELED"
                {
                    saw_canceled = true;
                    break;
                }
            }
            Ok(Some(Ok(_))) | Ok(Some(Err(_))) => continue,
            _ => break,
        }
    }

    let _ = rest.delete_listen_key(&listen_key).await;

    assert!(
        saw_canceled,
        "CANCELED ORDER_TRADE_UPDATE was not received within {MSG_TIMEOUT_SECS}s"
    );
}

// ===========================================================================
// SECTION 5 — Error Handling (integration-level)
// ===========================================================================

/// Subscribing with an empty stream list fails before any connection attempt.
#[tokio::test]
#[ignore]
async fn test_ws_empty_streams_returns_error() {
    let client = build_ws_client();
    let result = client.subscribe(vec![]).await;
    assert!(
        matches!(
            result,
            Err(AsterDexError::WebSocketError { ref message }) if message.contains("at least one stream")
        ),
        "expected WebSocketError about empty streams, got: {result:?}"
    );
}

/// Subscribing with 201 streams fails with a validation error (no network call made).
#[tokio::test]
#[ignore]
async fn test_ws_max_streams_exceeded_returns_error() {
    let client = build_ws_client();
    let streams: Vec<&str> = (0..201).map(|_| "btcusdt@aggTrade").collect();
    let result = client.subscribe(streams).await;
    assert!(
        matches!(
            result,
            Err(AsterDexError::WebSocketError { ref message }) if message.contains("max 200 streams")
        ),
        "expected WebSocketError about max streams, got: {result:?}"
    );
}

/// Invalid WebSocket URL returns ConfigError.
#[tokio::test]
#[ignore]
async fn test_ws_invalid_url_returns_config_error() {
    let result = WebSocketClient::new("not a url :::");
    assert!(
        matches!(result, Err(AsterDexError::ConfigError { .. })),
        "expected ConfigError, got: {result:?}"
    );
}

// ===========================================================================
// SECTION 6 — Raw message structure validation
// ===========================================================================

/// Verify that the combined-stream envelope is unwrapped correctly:
/// `{ "stream": "btcusdt@markPrice", "data": {...} }` → the `data` object
/// is forwarded to the channel (not the outer envelope).
#[tokio::test]
#[ignore]
async fn test_ws_envelope_unwrapped() {
    let client = build_ws_client();
    let mut stream = client
        .subscribe(vec!["btcusdt@markPrice"])
        .await
        .expect("subscribe failed");

    let raw = recv_raw(&mut stream, MSG_TIMEOUT_SECS).await;
    let raw = raw.expect("no raw message from btcusdt@markPrice");

    // After unwrapping, the message should be the inner data object.
    // It should have "e" == "markPriceUpdate", not a "stream" wrapper key.
    assert!(
        raw.get("stream").is_none(),
        "stream envelope was NOT unwrapped — outer 'stream' key still present"
    );
    let event_type = raw.get("e").and_then(|e| e.as_str()).unwrap_or("");
    assert_eq!(
        event_type, "markPriceUpdate",
        "expected event type 'markPriceUpdate', got '{event_type}'"
    );
}