ndaxrs 0.1.0

// Copyright (C) 2026 ndaxrs Art Morozov
// SPDX-License-Identifier: GPL-3.0-only

//! WebSocket client for the NDAX API.
//!
//! This module provides both async and sync interfaces for connecting to
//! the NDAX WebSocket API.
//!
//! # Architecture
//!
//! ```text
//! User Code
//!     │
//!     ▼
//! NdaxWsAPI (sync, blocking interface)
//!     │ mpsc channel
//!     ▼
//! Worker Thread (tokio runtime)
//!     │
//!     ▼
//! NdaxWsClient (async WebSocket client)
//!     │ tokio-tungstenite
//!     ▼
//! NDAX WebSocket Server
//! ```
//!
//! # Example
//!
//! ```rust,no_run
//! use ndaxrs::ws::{NdaxWsAPI, NdaxWsConfig};
//!
//! fn main() -> ndaxrs::Result<()> {
//!   let config = NdaxWsConfig::builder()
//!         .subscribe_level2(vec![4]) // BTC/CAD instrument
//!         .book_depth(10)
//!         .build();
//!
//!   let api = NdaxWsAPI::new(config)?;
//!
//!   // Get order book data
//!   if let Some(book) = api.get_book(4) {
//!     println!("Best bid: {:?}", book.best_bid());
//!     println!("Best ask: {:?}", book.best_ask());
//!   }
//!
//!   Ok(())
//! }
//! ```

mod config;
mod conn;
mod subscriptions;
mod types;

use std::{
  sync::{atomic::Ordering, Arc},
  thread,
  time::Duration,
};

pub use config::{NdaxWsConfig, NdaxWsConfigBuilder, PrivateConfig};
pub use conn::{NdaxWsClient, WsAPIResults, WsError};
use futures::StreamExt;
use serde::{de::DeserializeOwned, Serialize};
pub use subscriptions::*;
use tokio::{
  runtime,
  sync::{mpsc, oneshot},
  time,
};
pub use types::*;

use crate::{
  messages::{
    AccountPosition,
    CancelOrderRequest,
    GenericResponse,
    GetAccountPositionsRequest,
    GetInstrumentsRequest,
    GetOpenOrdersRequest,
    Instrument,
    NdaxFrame,
    OrderInfo,
    SendOrderRequest,
    SendOrderResponse,
  },
  Error,
  Result,
};

/// Synchronous WebSocket API for NDAX.
///
/// This is a blocking interface that spawns an internal worker thread
/// to manage the async WebSocket connection. It provides a simple API
/// for getting market data without requiring async code.
pub struct NdaxWsAPI {
  /// Handle to the worker thread.
  worker_thread: Option<thread::JoinHandle<()>>,
  /// Sender for commands to the worker thread.
  sender:        mpsc::UnboundedSender<WorkerCommand>,
  /// Shared output from the WebSocket connection.
  output:        Arc<WsAPIResults>,
}

impl NdaxWsAPI {
  /// Create a new WebSocket API instance.
  ///
  /// This will establish a WebSocket connection to NDAX and subscribe to
  /// the configured feeds. The connection runs in a background thread.
  ///
  /// # Arguments
  ///
  /// * `config` - Configuration for the WebSocket connection
  ///
  /// # Errors
  ///
  /// Returns an error if the WebSocket connection cannot be established.
  pub fn new(config: NdaxWsConfig) -> Result<Self> {
    let rt = runtime::Builder::new_current_thread()
      .enable_all()
      .build()
      .map_err(Error::Io)?;

    // Create the client (this handles auth and subscriptions internally)
    let (mut client, mut stream, output) =
      rt.block_on(NdaxWsClient::new(config))?;

    let (sender, mut receiver) = mpsc::unbounded_channel();

    let worker_thread = Some(
            thread::Builder::new()
                .name("ndax-ws-worker".into())
                .spawn(move || {
                    rt.block_on(async move {
                        let mut interval = time::interval(Duration::from_secs(30));

                        loop {
                            tokio::select! {
                                stream_result = stream.next() => {
                                    match stream_result {
                                        Some(result) => {
                                            if let Err(err) = client.update(result) {
                                                log::error!("NDAX WebSocket error: {}", err);
                                                drop(client.close().await);
                                                return;
                                            }
                                        }
                                        None => {
                                            log::warn!("NDAX WebSocket stream closed");
                                            drop(client.close().await);
                                            return;
                                        }
                                    }
                                }
                                cmd = receiver.recv() => {
                                    match cmd {
                                        None | Some(WorkerCommand::Stop) => {
                                            drop(client.close().await);
                                            return;
                                        }
                                        Some(WorkerCommand::Request { name, payload, response_tx }) => {
                                            if let Err(e) = client.request_raw(&name, &payload, response_tx).await {
                                                log::error!("Failed to send request '{}': {}", name, e);
                                            }
                                        }
                                    }
                                }
                                _ = interval.tick() => {
                                    // Heartbeat check could go here
                                }
                            }
                        }
                    })
                })
                .map_err(Error::Io)?,
        );

    Ok(Self {
      worker_thread,
      sender,
      output,
    })
  }

  /// Get the current order book for an instrument.
  ///
  /// # Arguments
  ///
  /// * `instrument_id` - The instrument ID to get the book for
  ///
  /// # Returns
  ///
  /// A copy of the current order book data, or None if not subscribed.
  pub fn get_book(&self, instrument_id: u64) -> Option<BookData> {
    self
      .output
      .books
      .get(&instrument_id)
      .map(|lock| lock.lock().expect("mutex poisoned").clone())
  }

  /// Get the current Level 1 ticker data for an instrument.
  ///
  /// # Arguments
  ///
  /// * `instrument_id` - The instrument ID to get the ticker for
  ///
  /// # Returns
  ///
  /// A copy of the current Level 1 snapshot, or None if not subscribed.
  pub fn get_level1(&self, instrument_id: u64) -> Option<Level1Snapshot> {
    self
      .output
      .level1
      .get(&instrument_id)
      .map(|lock| lock.lock().expect("mutex poisoned").clone())
  }

  /// Get the recent trades for an instrument.
  ///
  /// Returns a clone of the trade history vector, or None if not subscribed.
  pub fn get_trades(&self, instrument_id: u64) -> Option<Vec<TradeData>> {
    self
      .output
      .trades
      .get(&instrument_id)
      .map(|lock| lock.lock().expect("mutex poisoned").clone())
  }

  /// Check if the WebSocket stream is closed.
  ///
  /// If true, this API instance should be dropped and a new one created.
  pub fn is_closed(&self) -> bool {
    self.output.stream_closed.load(Ordering::SeqCst)
  }

  /// Check if the connection is authenticated.
  pub fn is_authenticated(&self) -> bool {
    self.output.authenticated.load(Ordering::SeqCst)
  }

  /// Close the WebSocket connection gracefully.
  ///
  /// This will signal the worker thread to stop and wait for it to finish.
  pub fn close(mut self) {
    let _ = self.sender.send(WorkerCommand::Stop);
    if let Some(thread) = self.worker_thread.take() {
      let _ = thread.join();
    }
  }

  /// Send a request to the NDAX API and wait for a response.
  ///
  /// This is a blocking operation that sends a request and waits for the
  /// response from the server.
  ///
  /// # Arguments
  ///
  /// * `endpoint` - The API endpoint name (e.g., "GetInstruments")
  /// * `payload` - The request payload
  ///
  /// # Type Parameters
  ///
  /// * `T` - The request payload type (must implement Serialize)
  /// * `R` - The response type (must implement DeserializeOwned)
  ///
  /// # Errors
  ///
  /// Returns an error if the request fails or the response cannot be parsed.
  pub fn request<T: Serialize, R: DeserializeOwned>(
    &self,
    endpoint: &str,
    payload: &T,
  ) -> Result<R> {
    let payload_json = serde_json::to_string(payload)?;
    let (response_tx, response_rx) = oneshot::channel();

    self
      .sender
      .send(WorkerCommand::Request {
        name: endpoint.to_string(),
        payload: payload_json,
        response_tx,
      })
      .map_err(|_| Error::ConnectionClosed)?;

    let frame = response_rx
      .blocking_recv()
      .map_err(|_| Error::ConnectionClosed)??;

    frame.parse_payload()
  }

  /// Get the list of available instruments (trading pairs).
  ///
  /// Returns all instruments available on the exchange.
  ///
  /// # Example
  ///
  /// ```rust,no_run
  /// # use ndaxrs::ws::{NdaxWsAPI, NdaxWsConfig};
  /// # fn main() -> ndaxrs::Result<()> {
  /// let api = NdaxWsAPI::new(NdaxWsConfig::builder().build())?;
  /// let instruments = api.get_instruments()?;
  /// for inst in instruments {
  ///   println!("{}: {}", inst.instrument_id, inst.trading_pair());
  /// }
  /// # Ok(())
  /// # }
  /// ```
  pub fn get_instruments(&self) -> Result<Vec<Instrument>> {
    let request = GetInstrumentsRequest {
      oms_id: 1
    };
    self.request("GetInstruments", &request)
  }

  /// Get account positions (balances).
  ///
  /// Returns the balance for each product in the specified account.
  ///
  /// # Arguments
  ///
  /// * `account_id` - The account ID to query
  ///
  /// # Example
  ///
  /// ```rust,no_run
  /// # use ndaxrs::ws::{NdaxWsAPI, NdaxWsConfig};
  /// # fn main() -> ndaxrs::Result<()> {
  /// let api = NdaxWsAPI::new(NdaxWsConfig::builder().build())?;
  /// let positions = api.get_account_positions(12345)?;
  /// for pos in positions {
  ///   println!(
  ///     "{}: {} (hold: {})",
  ///     pos.product_symbol, pos.amount, pos.hold
  ///   );
  /// }
  /// # Ok(())
  /// # }
  /// ```
  pub fn get_account_positions(
    &self,
    account_id: u64,
  ) -> Result<Vec<AccountPosition>> {
    let request = GetAccountPositionsRequest {
      account_id,
      oms_id: 1,
    };
    self.request("GetAccountPositions", &request)
  }

  /// Send an order to the exchange.
  ///
  /// Places a new order on the exchange. The order will be validated
  /// and either accepted or rejected.
  ///
  /// # Arguments
  ///
  /// * `order` - The order request containing all order details
  ///
  /// # Returns
  ///
  /// Returns a response indicating whether the order was accepted
  /// and the assigned order ID if successful.
  ///
  /// # Example
  ///
  /// ```rust,no_run
  /// # use ndaxrs::ws::{NdaxWsAPI, NdaxWsConfig};
  /// # use ndaxrs::messages::orders::create_limit_order;
  /// # use ndaxrs::Side;
  /// # use rust_decimal_macros::dec;
  /// # fn main() -> ndaxrs::Result<()> {
  /// let api = NdaxWsAPI::new(NdaxWsConfig::builder().build())?;
  /// let order = create_limit_order(1, 12345, Side::Buy, dec!(0.001), dec!(1.00));
  /// let response = api.send_order(&order)?;
  /// println!("Order status: {}", response.status);
  /// # Ok(())
  /// # }
  /// ```
  pub fn send_order(
    &self,
    order: &SendOrderRequest,
  ) -> Result<SendOrderResponse> {
    self.request("SendOrder", order)
  }

  /// Cancel an existing order.
  ///
  /// Cancels an open order by order ID or client order ID.
  ///
  /// # Arguments
  ///
  /// * `request` - The cancel request with order identification
  ///
  /// # Example
  ///
  /// ```rust,no_run
  /// # use ndaxrs::ws::{NdaxWsAPI, NdaxWsConfig};
  /// # use ndaxrs::messages::orders::CancelOrderRequest;
  /// # fn main() -> ndaxrs::Result<()> {
  /// let api = NdaxWsAPI::new(NdaxWsConfig::builder().build())?;
  /// let cancel_request = CancelOrderRequest {
  ///   oms_id:          1,
  ///   account_id:      12345,
  ///   order_id:        Some(98765),
  ///   client_order_id: None,
  /// };
  /// let response = api.cancel_order(&cancel_request)?;
  /// # Ok(())
  /// # }
  /// ```
  pub fn cancel_order(
    &self,
    request: &CancelOrderRequest,
  ) -> Result<GenericResponse> {
    self.request("CancelOrder", request)
  }

  /// Get open orders for an account.
  ///
  /// Returns all currently open (working) orders for the account.
  ///
  /// # Arguments
  ///
  /// * `account_id` - The account ID to query
  ///
  /// # Example
  ///
  /// ```rust,no_run
  /// # use ndaxrs::ws::{NdaxWsAPI, NdaxWsConfig};
  /// # fn main() -> ndaxrs::Result<()> {
  /// let api = NdaxWsAPI::new(NdaxWsConfig::builder().build())?;
  /// let orders = api.get_open_orders(12345)?;
  /// for order in orders {
  ///   println!(
  ///     "Order {}: {} @ {}",
  ///     order.order_id, order.quantity, order.price
  ///   );
  /// }
  /// # Ok(())
  /// # }
  /// ```
  pub fn get_open_orders(&self, account_id: u64) -> Result<Vec<OrderInfo>> {
    let request = GetOpenOrdersRequest {
      account_id,
      oms_id: 1,
    };
    self.request("GetOpenOrders", &request)
  }
}

impl Drop for NdaxWsAPI {
  fn drop(&mut self) {
    if let Some(thread) = self.worker_thread.take() {
      let _ = self.sender.send(WorkerCommand::Stop);
      let _ = thread.join();
    }
  }
}

/// Commands sent from the sync API to the worker thread.
enum WorkerCommand {
  /// Stop the worker thread and close the connection.
  Stop,
  /// Send a request and get a response.
  Request {
    name:        String,
    payload:     String,
    response_tx: oneshot::Sender<Result<NdaxFrame>>,
  },
}

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

  #[test]
  fn test_config_builder() {
    let config = NdaxWsConfig::builder()
      .subscribe_level1(vec![1, 2])
      .subscribe_level2(vec![3, 4])
      .book_depth(20)
      .build();

    assert_eq!(config.subscribe_level1, vec![1, 2]);
    assert_eq!(config.subscribe_level2, vec![3, 4]);
    assert_eq!(config.book_depth, 20);
  }

  #[test]
  #[ignore]
  fn test_connect_to_ndax() {
    let config = NdaxWsConfig::builder()
      .subscribe_level2(vec![4])
      .book_depth(10)
      .build();

    let api = NdaxWsAPI::new(config).expect("Failed to connect to NDAX");

    std::thread::sleep(Duration::from_secs(2));

    assert!(!api.is_closed());
    api.close();
  }

  #[test]
  #[ignore]
  fn test_send_receive_frame() {
    // BTC/CAD (instrument 1) is the most liquid market
    let btc_cad_instrument_id = 1;
    let config = NdaxWsConfig::builder()
      .subscribe_level2(vec![btc_cad_instrument_id])
      .book_depth(10)
      .build();

    let api = NdaxWsAPI::new(config).expect("Failed to connect to NDAX");

    // Wait for data with retries - NDAX may take a moment to send the snapshot
    let mut found_data = false;
    let mut book_exists = false;
    for i in 1..=10 {
      std::thread::sleep(Duration::from_secs(1));
      if let Some(book) = api.get_book(btc_cad_instrument_id) {
        book_exists = true;
        if !book.bids.is_empty() || !book.asks.is_empty() {
          println!(
            "Book data received after {}s: {} bids, {} asks",
            i,
            book.bids.len(),
            book.asks.len()
          );
          found_data = true;
          break;
        }
      }
    }

    api.close();

    // The test passes if we got any data, or if the book entry exists
    // (data may be sparse on weekends/holidays)
    if !found_data {
      if book_exists {
        println!(
          "Warning: Book exists but is empty after 10s - market may be \
           inactive"
        );
      } else {
        panic!("Book not found for instrument {}", btc_cad_instrument_id);
      }
    }
  }
}