deribit_fix/session/

fix_session.rs

//! FIX session management

use crate::config::gen_id;
use crate::model::message::FixMessage;
use crate::model::position::Position;
use crate::model::request::{NewOrderRequest, OrderSide, OrderType, TimeInForce};
use crate::model::types::MsgType;
use crate::{
    config::DeribitFixConfig,
    connection::Connection,
    error::{DeribitFixError, Result},
    message::{MessageBuilder, PositionReport, RequestForPositions},
};
use base64::prelude::*;
use chrono::Utc;
use rand;
use sha2::{Digest, Sha256};
use std::str::FromStr;
use std::sync::Arc;
use tokio::sync::Mutex;
use tracing::{debug, error, info, trace};

/// FIX session state
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SessionState {
    /// Session is disconnected
    Disconnected,
    /// Logon message sent, waiting for response
    LogonSent,
    /// Session is logged on and active
    LoggedOn,
    /// Logout message sent, waiting for confirmation
    LogoutSent,
}

/// FIX session manager
pub struct Session {
    config: DeribitFixConfig,
    connection: Option<Arc<Mutex<Connection>>>,
    state: SessionState,
    outgoing_seq_num: u32,
    incoming_seq_num: u32,
}

impl Session {
    /// Create a new FIX session
    pub fn new(config: &DeribitFixConfig, connection: Arc<Mutex<Connection>>) -> Result<Self> {
        info!("Creating new FIX session");
        Ok(Self {
            config: config.clone(),
            state: SessionState::Disconnected,
            outgoing_seq_num: 1,
            incoming_seq_num: 1,
            connection: Some(connection),
        })
    }

    /// Set the connection for this session
    pub fn set_connection(&mut self, connection: Arc<Mutex<Connection>>) {
        self.connection = Some(connection);
    }

    /// Get the current session state
    pub fn get_state(&self) -> SessionState {
        self.state
    }

    /// Send a FIX message through the connection
    async fn send_message(&mut self, message: FixMessage) -> Result<()> {
        if let Some(connection) = &self.connection {
            let mut conn_guard = connection.lock().await;
            conn_guard.send_message(&message).await?;
            debug!("Sent FIX message: {}", message.to_string());
        } else {
            return Err(DeribitFixError::Connection(
                "No connection available".to_string(),
            ));
        }
        Ok(())
    }

    /// Perform FIX logon
    pub async fn logon(&mut self) -> Result<()> {
        info!("Performing FIX logon");

        // Generate RawData and password hash according to Deribit FIX spec
        let (raw_data, password_hash) = self.generate_auth_data(&self.config.password)?;

        let mut message_builder = MessageBuilder::new()
            .msg_type(MsgType::Logon)
            .sender_comp_id(self.config.sender_comp_id.clone())
            .target_comp_id(self.config.target_comp_id.clone())
            .msg_seq_num(self.outgoing_seq_num)
            .field(108, self.config.heartbeat_interval.to_string()) // HeartBtInt - Required
            .field(96, raw_data.clone()) // RawData - Required (timestamp.nonce)
            .field(553, self.config.username.clone()) // Username - Required
            .field(554, password_hash); // Password - Required

        // Add RawDataLength if needed (optional but recommended)
        message_builder = message_builder.field(95, raw_data.len().to_string()); // RawDataLength

        // Add optional Deribit-specific tags based on configuration
        if let Some(use_wordsafe_tags) = &self.config.use_wordsafe_tags {
            message_builder =
                message_builder.field(9002, if *use_wordsafe_tags { "Y" } else { "N" }.to_string()); // UseWordsafeTags
        }

        // CancelOnDisconnect - always include based on config
        message_builder = message_builder.field(
            9001,
            if self.config.cancel_on_disconnect {
                "Y"
            } else {
                "N"
            }
            .to_string(),
        ); // CancelOnDisconnect

        if let Some(app_id) = &self.config.app_id {
            message_builder = message_builder.field(9004, app_id.clone()); // DeribitAppId
        }

        if let Some(app_secret) = &self.config.app_secret
            && let Some(raw_data_str) = raw_data
                .split_once('.')
                .map(|(timestamp, nonce)| format!("{}.{}", timestamp, nonce))
            && let Ok(app_sig) = self.calculate_app_signature(&raw_data_str, app_secret)
        {
            message_builder = message_builder.field(9005, app_sig); // DeribitAppSig
        }

        if let Some(deribit_sequential) = &self.config.deribit_sequential {
            message_builder = message_builder.field(
                9007,
                if *deribit_sequential { "Y" } else { "N" }.to_string(),
            ); // DeribitSequential
        }

        if let Some(unsubscribe_exec_reports) = &self.config.unsubscribe_execution_reports {
            message_builder = message_builder.field(
                9009,
                if *unsubscribe_exec_reports { "Y" } else { "N" }.to_string(),
            ); // UnsubscribeExecutionReports
        }

        if let Some(connection_only_exec_reports) = &self.config.connection_only_execution_reports {
            message_builder = message_builder.field(
                9010,
                if *connection_only_exec_reports {
                    "Y"
                } else {
                    "N"
                }
                .to_string(),
            ); // ConnectionOnlyExecutionReports
        }

        if let Some(report_fills_as_exec_reports) = &self.config.report_fills_as_exec_reports {
            message_builder = message_builder.field(
                9015,
                if *report_fills_as_exec_reports {
                    "Y"
                } else {
                    "N"
                }
                .to_string(),
            ); // ReportFillsAsExecReports
        }

        if let Some(display_increment_steps) = &self.config.display_increment_steps {
            message_builder = message_builder.field(
                9018,
                if *display_increment_steps { "Y" } else { "N" }.to_string(),
            ); // DisplayIncrementSteps
        }

        // Add AppID if available
        if let Some(app_id) = &self.config.app_id {
            message_builder = message_builder.field(1128, app_id.clone()); // AppID
        }

        let logon_message = message_builder.build()?;

        // Send the logon message
        self.send_message(logon_message).await?;
        self.state = SessionState::LogonSent;
        self.outgoing_seq_num += 1;

        info!("Logon message sent");
        Ok(())
    }

    /// Perform FIX logout
    pub async fn logout(&mut self) -> Result<()> {
        self.logout_with_options(None, None).await
    }

    /// Perform FIX logout with optional parameters
    pub async fn logout_with_options(
        &mut self,
        text: Option<String>,
        dont_cancel_on_disconnect: Option<bool>,
    ) -> Result<()> {
        info!("Performing FIX logout");

        let mut message_builder = MessageBuilder::new()
            .msg_type(MsgType::Logout)
            .sender_comp_id(self.config.sender_comp_id.clone())
            .target_comp_id(self.config.target_comp_id.clone())
            .msg_seq_num(self.outgoing_seq_num);

        // Add Text field (tag 58) - optional
        let logout_text = text.unwrap_or_else(|| "Normal logout".to_string());
        message_builder = message_builder.field(58, logout_text); // Text

        // Add DontCancelOnDisconnect field (tag 9003) - optional
        if let Some(dont_cancel) = dont_cancel_on_disconnect {
            message_builder =
                message_builder.field(9003, if dont_cancel { "Y" } else { "N" }.to_string()); // DontCancelOnDisconnect
        }

        let logout_message = message_builder.build()?;

        // Send the logout message
        self.send_message(logout_message).await?;
        self.state = SessionState::LogoutSent;
        self.outgoing_seq_num += 1;

        info!("Logout message sent");
        Ok(())
    }

    /// Send a heartbeat message
    pub async fn send_heartbeat(&mut self, test_req_id: Option<String>) -> Result<()> {
        debug!("Sending heartbeat message");

        let mut builder = MessageBuilder::new()
            .msg_type(MsgType::Heartbeat)
            .sender_comp_id(self.config.sender_comp_id.clone())
            .target_comp_id(self.config.target_comp_id.clone())
            .msg_seq_num(self.outgoing_seq_num);

        if let Some(test_req_id) = test_req_id {
            builder = builder.field(112, test_req_id); // TestReqID
        }

        let heartbeat_message = builder.build()?;

        // Send the heartbeat message
        self.send_message(heartbeat_message).await?;
        self.outgoing_seq_num += 1;

        debug!("Heartbeat message sent");
        Ok(())
    }

    /// Send a new order
    pub async fn send_new_order(&mut self, order: NewOrderRequest) -> Result<String> {
        info!("Sending new order: {:?}", order);

        // Use the client order ID if provided, otherwise generate one
        let order_id = order
            .client_order_id
            .clone()
            .unwrap_or_else(|| format!("ORDER_{}", gen_id()));

        // Determine order type
        let ord_type = match order.order_type {
            OrderType::Market => "1",     // Market
            OrderType::Limit => "2",      // Limit
            OrderType::StopLimit => "4",  // Stop Limit
            OrderType::StopMarket => "3", // Stop Market
            _ => "2",                     // Default to Limit for other types
        };

        let mut builder = MessageBuilder::new()
            .msg_type(MsgType::NewOrderSingle)
            .sender_comp_id(self.config.sender_comp_id.clone())
            .target_comp_id(self.config.target_comp_id.clone())
            .msg_seq_num(self.outgoing_seq_num)
            .field(11, order_id.clone()) // ClOrdID
            .field(55, order.instrument_name.clone()) // Symbol
            .field(
                54,
                match order.side {
                    OrderSide::Buy => "1".to_string(),
                    OrderSide::Sell => "2".to_string(),
                },
            ) // Side
            .field(60, Utc::now().format("%Y%m%d-%H:%M:%S%.3f").to_string()) // TransactTime
            .field(38, order.amount.to_string()) // OrderQty
            .field(40, ord_type.to_string()); // OrdType

        // Add price for limit orders
        if order.order_type == OrderType::Limit || order.price.is_some() {
            builder = builder.field(44, order.price.unwrap_or(0.0).to_string()); // Price
        }

        // Add time in force
        let tif = match order.time_in_force {
            TimeInForce::GoodTilCancelled => "1",
            TimeInForce::ImmediateOrCancel => "3",
            TimeInForce::FillOrKill => "4",
            TimeInForce::GoodTilDay => "6",
        };
        builder = builder.field(59, tif.to_string()); // TimeInForce

        // Add execution instructions
        let mut exec_inst = String::new();
        if order.post_only == Some(true) {
            exec_inst.push('6'); // Participate don't initiate
        }
        if order.reduce_only == Some(true) {
            exec_inst.push('E'); // Reduce only
        }
        if !exec_inst.is_empty() {
            builder = builder.field(18, exec_inst); // ExecInst
        }

        // Add label if provided
        if let Some(label) = &order.label {
            builder = builder.field(100010, label.clone()); // Deribit label
        }

        let order_message = builder.build()?;

        // Actually send the message
        self.send_message(order_message).await?;
        self.outgoing_seq_num += 1;

        info!("New order message sent with ID: {}", order_id);
        Ok(order_id)
    }

    /// Cancel an order
    ///
    /// # Arguments
    /// * `order_id` - The order identifier (OrigClOrdID) to cancel
    /// * `symbol` - Optional instrument symbol. Required when canceling by ClOrdID or DeribitLabel,
    ///   but not required when using OrigClOrdID (fastest approach)
    /// * `currency` - Optional currency to speed up search when using ClOrdID or DeribitLabel
    pub async fn cancel_order(&mut self, order_id: String) -> Result<()> {
        self.cancel_order_with_symbol(order_id, None).await
    }

    /// Cancel an order with optional symbol specification
    ///
    /// According to Deribit FIX documentation:
    /// - Canceling by OrigClOrdId is fastest and recommended when possible
    /// - Symbol is required only when OrigClOrdId is absent (canceling by ClOrdID or DeribitLabel)
    /// - Currency can optionally speed up searches by DeribitLabel or ClOrdID
    ///
    /// # Arguments
    /// * `order_id` - The order identifier (OrigClOrdID) to cancel
    /// * `symbol` - Optional instrument symbol (e.g., "BTC-PERPETUAL")
    /// * `currency` - Optional currency to speed up search
    pub async fn cancel_order_with_symbol(
        &mut self,
        order_id: String,
        symbol: Option<String>,
    ) -> Result<()> {
        info!("Cancelling order: {} with symbol: {:?}", order_id, symbol);

        // Generate a proper unique cancel ID using random number instead of timestamp
        let cancel_id = format!("CANCEL_{}", gen_id());

        let mut builder = MessageBuilder::new()
            .msg_type(MsgType::OrderCancelRequest)
            .sender_comp_id(self.config.sender_comp_id.clone())
            .target_comp_id(self.config.target_comp_id.clone())
            .msg_seq_num(self.outgoing_seq_num)
            .field(11, cancel_id) // ClOrdID - Original order identifier assigned by the user
            .field(41, order_id) // OrigClOrdID - Order identifier assigned by Deribit
            .field(60, Utc::now().format("%Y%m%d-%H:%M:%S%.3f").to_string()); // TransactTime

        // Add symbol if provided - required when OrigClOrdId is absent
        if let Some(symbol_value) = symbol {
            builder = builder.field(55, symbol_value); // Symbol
        }

        let cancel_message = builder.build()?;

        // Actually send the cancel message
        self.send_message(cancel_message).await?;
        self.outgoing_seq_num += 1;

        info!("Order cancel message sent");
        Ok(())
    }

    /// Subscribe to market data
    pub async fn subscribe_market_data(&mut self, symbol: String) -> Result<()> {
        info!("Subscribing to market data for: {}", symbol);

        let request_id = format!("MDR_{}", gen_id());

        let market_data_request = MessageBuilder::new()
            .msg_type(MsgType::MarketDataRequest)
            .sender_comp_id(self.config.sender_comp_id.clone())
            .target_comp_id(self.config.target_comp_id.clone())
            .msg_seq_num(self.outgoing_seq_num)
            .field(262, request_id.clone()) // MDReqID
            .field(263, "1".to_string()) // SubscriptionRequestType (1 = Snapshot + Updates)
            .field(264, "0".to_string()) // MarketDepth (0 = Full Book)
            .field(267, "2".to_string()) // NoMDEntryTypes
            .field(269, "0".to_string()) // MDEntryType (0 = Bid)
            .field(269, "1".to_string()) // MDEntryType (1 = Offer)
            .field(146, "1".to_string()) // NoRelatedSym
            .field(55, symbol.clone()) // Symbol
            .build()?;

        // Send the market data request
        self.send_message(market_data_request).await?;
        self.outgoing_seq_num += 1;

        info!(
            "Market data subscription request sent for symbol: {} with ID: {}",
            symbol, request_id
        );
        Ok(())
    }

    /// Request positions asynchronously
    pub async fn request_positions(&mut self) -> Result<Vec<Position>> {
        use std::time::{Duration, Instant};
        use tracing::{debug, info, warn};

        info!("Requesting positions");

        let request_id = format!("POS_{}", gen_id());

        // Create typed position request
        let position_request = RequestForPositions::all_positions(request_id.clone())
            .with_clearing_date(Utc::now().format("%Y%m%d").to_string());

        // Build the FIX message
        let fix_message = position_request.to_fix_message(
            self.config.sender_comp_id.clone(),
            self.config.target_comp_id.clone(),
            self.outgoing_seq_num,
        )?;

        // Send the position request
        self.send_message(fix_message).await?;
        self.outgoing_seq_num += 1;

        info!(
            "Position request sent, awaiting responses for request ID: {}",
            request_id
        );

        // Collect position reports with correlation by PosReqID
        let mut positions = Vec::new();
        let timeout = Duration::from_secs(30); // 30 second timeout
        let start_time = Instant::now();

        loop {
            // Check for timeout
            if start_time.elapsed() > timeout {
                warn!("Position request timed out after {:?}", timeout);
                break;
            }

            // Receive and process messages
            match self.receive_and_process_message().await {
                Ok(Some(message)) => {
                    // Check if this is a PositionReport message
                    if let Some(msg_type_str) = message.get_field(35)
                        && msg_type_str == "AP"
                    {
                        // PositionReport
                        // Check if this position report matches our request ID
                        if let Some(pos_req_id) = message.get_field(710) {
                            if pos_req_id == &request_id {
                                debug!("Received PositionReport for request: {}", request_id);

                                // Check if this is an empty position report (no instrument name)
                                if message.get_field(55).is_none() {
                                    info!(
                                        "Received empty position report - indicates no active positions for this request"
                                    );
                                    debug!(
                                        "Empty PositionReport details: PosReqID={}, PosMaintRptID={:?}",
                                        request_id,
                                        message.get_field(721)
                                    );
                                    // This is an end-of-positions marker indicating no positions, continue waiting
                                    continue;
                                }

                                match PositionReport::try_from_fix_message(&message) {
                                    Ok(position) => {
                                        debug!("Successfully parsed position: {:?}", position);
                                        positions.push(position);
                                    }
                                    Err(e) => {
                                        warn!("Failed to parse PositionReport: {}", e);
                                        debug!("Message fields: {:?}", message);
                                    }
                                }
                            } else {
                                debug!(
                                    "Received PositionReport for different request: {}",
                                    pos_req_id
                                );
                            }
                        }
                    }
                }
                Ok(None) => {
                    // No message received, continue loop
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
                Err(e) => {
                    warn!("Error receiving message: {}", e);
                    // Continue trying to receive more messages
                    tokio::time::sleep(Duration::from_millis(100)).await;
                }
            }

            // For now, we'll break after receiving some positions or after a reasonable time
            // In a real implementation, we might wait for an end-of-transmission signal
            if !positions.is_empty() && start_time.elapsed() > Duration::from_secs(5) {
                debug!(
                    "Received {} positions, stopping collection",
                    positions.len()
                );
                break;
            }
        }

        info!(
            "Position request completed, received {} positions",
            positions.len()
        );
        Ok(positions)
    }

    /// Generate authentication data according to Deribit FIX specification
    /// Returns (raw_data, base64_password_hash)
    pub fn generate_auth_data(&self, access_secret: &str) -> Result<(String, String)> {
        // Generate timestamp (strictly increasing integer in milliseconds)
        let timestamp = Utc::now().timestamp_millis();

        // Generate random nonce (at least 32 bytes as recommended by Deribit)
        let mut nonce_bytes = vec![0u8; 32];
        for byte in nonce_bytes.iter_mut() {
            *byte = rand::random::<u8>();
        }
        let nonce_b64 = BASE64_STANDARD.encode(&nonce_bytes);

        // Create RawData: timestamp.nonce (separated by ASCII period)
        let raw_data = format!("{timestamp}.{nonce_b64}");

        // Calculate password hash: base64(sha256(RawData ++ access_secret))
        let mut auth_data = raw_data.as_bytes().to_vec();
        auth_data.extend_from_slice(access_secret.as_bytes());

        debug!("Auth Data at Timestamp: {}", timestamp);
        trace!("Nonce length: {} bytes", nonce_bytes.len());
        trace!("Nonce (base64): {}", nonce_b64);
        trace!("RawData: {}", raw_data);
        trace!("Access secret: {}", access_secret);
        trace!("Auth data length: {} bytes", auth_data.len());

        let mut hasher = Sha256::new();
        hasher.update(&auth_data);
        let hash_result = hasher.finalize();
        let password_hash = BASE64_STANDARD.encode(hash_result);

        debug!("Password hash: {}", password_hash);

        Ok((raw_data, password_hash))
    }

    /// Calculate application signature for registered apps
    #[allow(dead_code)]
    fn calculate_app_signature(&self, raw_data: &str, app_secret: &str) -> Result<String> {
        let mut hasher = Sha256::new();
        hasher.update(format!("{raw_data}{app_secret}").as_bytes());
        let result = hasher.finalize();
        Ok(BASE64_STANDARD.encode(result))
    }

    /// Get current session state
    pub fn state(&self) -> SessionState {
        self.state
    }

    /// Set session state (for testing)
    pub fn set_state(&mut self, state: SessionState) {
        self.state = state;
    }

    /// Process incoming FIX message
    async fn process_message(&mut self, message: &FixMessage) -> Result<()> {
        debug!("Processing FIX message: {:?}", message);

        // Get message type
        let msg_type_str = message.get_field(35).unwrap_or(&String::new()).clone();

        // Skip messages with empty or missing message type
        if msg_type_str.is_empty() {
            debug!("Skipping message with empty message type");
            return Ok(());
        }

        let msg_type = MsgType::from_str(&msg_type_str).map_err(|_| {
            DeribitFixError::MessageParsing(format!("Unknown message type: {msg_type_str}"))
        })?;

        match msg_type {
            MsgType::Logon => {
                info!("Received logon response");
                self.state = SessionState::LoggedOn;
            }
            MsgType::Logout => {
                info!("Received logout message");
                self.state = SessionState::Disconnected;
            }
            MsgType::Heartbeat => {
                debug!("Received heartbeat");
            }
            MsgType::TestRequest => {
                debug!("Received test request, sending heartbeat response");
                let test_req_id = message.get_field(112);
                self.send_heartbeat(test_req_id.cloned()).await?;
            }
            MsgType::ExecutionReport => {
                debug!("Received ExecutionReport: {:?}", message);
                // ExecutionReport processing - let the client handle the details
            }
            MsgType::PositionReport => {
                debug!("Received PositionReport: {:?}", message);
                // PositionReport processing - let the client handle the details
            }
            MsgType::Reject => {
                error!("Received Reject message: {:?}", message);
                // Reject message processing
            }
            _ => {
                debug!("Received message type: {:?}", msg_type);
            }
        }

        self.incoming_seq_num += 1;
        Ok(())
    }

    /// Receive and process a FIX message from the connection
    pub async fn receive_and_process_message(&mut self) -> Result<Option<FixMessage>> {
        let message = if let Some(connection) = &self.connection {
            let mut conn_guard = connection.lock().await;
            conn_guard.receive_message().await?
        } else {
            None
        };

        if let Some(message) = message {
            self.process_message(&message).await?;
            Ok(Some(message))
        } else {
            Ok(None)
        }
    }
}