//! Broker Gateway Service Implementation
//!
//! gRPC service for routing orders to brokers via cTrader Open API.
//! Persists orders to PostgreSQL and optionally routes to a live broker.

use futures::Stream;
use sqlx::PgPool;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
use tonic::{Request, Response, Status};
use tracing::{error, info, instrument, warn};

use crate::metrics;
use crate::proto::broker_gateway::*;
use crate::tracing as bg_tracing;

use ctrader_openapi::CTraderClient;

/// Broker Gateway Service state
pub struct BrokerGatewayService {
    db_pool: PgPool,
    session_state: Arc<RwLock<SessionState>>,
    broker_client: Arc<RwLock<Option<CTraderClient>>>,
}

impl BrokerGatewayService {
    pub fn new(db_pool: PgPool, redis_url: &str) -> anyhow::Result<Self> {
        // Validate Redis URL is parseable (connection used for future cache integration)
        let _redis_client = redis::Client::open(redis_url)?;

        Ok(Self {
            db_pool,
            session_state: Arc::new(RwLock::new(SessionState::Active)),
            broker_client: Arc::new(RwLock::new(None)),
        })
    }

    /// Set the cTrader broker client (call after connecting).
    pub async fn set_broker_client(&self, client: CTraderClient) {
        let mut guard = self.broker_client.write().await;
        *guard = Some(client);
        info!("cTrader broker client set successfully");
    }

    /// Generate client order ID (UUID v4)
    fn generate_client_order_id() -> String {
        uuid::Uuid::new_v4().to_string()
    }

    /// Validate order request
    fn validate_order(&self, req: &RouteOrderRequest) -> Result<(), Status> {
        if req.symbol.is_empty() {
            return Err(Status::invalid_argument("Symbol is required"));
        }
        if req.quantity <= 0.0 {
            return Err(Status::invalid_argument("Quantity must be positive"));
        }
        if req.account_id.is_empty() {
            return Err(Status::invalid_argument("Account ID is required"));
        }

        // Validate order type specific fields
        match OrderType::try_from(req.order_type) {
            Ok(OrderType::Limit) => {
                if req.price.is_none() {
                    return Err(Status::invalid_argument(
                        "Price is required for LIMIT orders",
                    ));
                }
            }
            Ok(OrderType::Stop) => {
                if req.stop_price.is_none() {
                    return Err(Status::invalid_argument(
                        "Stop price is required for STOP orders",
                    ));
                }
            }
            Ok(OrderType::StopLimit) => {
                if req.price.is_none() || req.stop_price.is_none() {
                    return Err(Status::invalid_argument(
                        "Both price and stop_price are required for STOP_LIMIT orders",
                    ));
                }
            }
            _ => {}
        }

        Ok(())
    }
}

/// Convert a quantity in lots to cTrader volume (lots * 100,000).
///
/// Returns an error if the result is not finite, negative, or exceeds i64 range.
/// Uses rounding to avoid silent truncation of fractional lots.
fn convert_quantity_to_volume(quantity: f64) -> Result<i64, Status> {
    let volume_f = quantity * 100_000.0;
    if !volume_f.is_finite() || volume_f < 0.0 || volume_f > i64::MAX as f64 {
        return Err(Status::invalid_argument(format!(
            "Volume overflow: quantity {} produces volume {}",
            quantity, volume_f
        )));
    }
    Ok(volume_f.round() as i64)
}

#[tonic::async_trait]
impl broker_gateway_service_server::BrokerGatewayService for BrokerGatewayService {
    #[instrument(skip(self), fields(symbol, side, quantity))]
    async fn route_order(
        &self,
        request: Request<RouteOrderRequest>,
    ) -> Result<Response<RouteOrderResponse>, Status> {
        let start = std::time::Instant::now();
        let req = request.into_inner();

        // Check session state before routing any orders
        {
            let state = *self.session_state.read().await;
            if state != SessionState::Active {
                warn!(
                    session_state = ?state,
                    symbol = %req.symbol,
                    "Order rejected: session is not active"
                );
                return Err(Status::unavailable(format!(
                    "Broker session is not active (state: {:?}). Cannot route orders.",
                    state
                )));
            }
        }

        // Create tracing span
        let side_str = match OrderSide::try_from(req.side) {
            Ok(OrderSide::Buy) => "BUY",
            Ok(OrderSide::Sell) => "SELL",
            _ => "UNKNOWN",
        };
        let order_type_str = match OrderType::try_from(req.order_type) {
            Ok(OrderType::Market) => "MARKET",
            Ok(OrderType::Limit) => "LIMIT",
            Ok(OrderType::Stop) => "STOP",
            Ok(OrderType::StopLimit) => "STOP_LIMIT",
            _ => "UNKNOWN",
        };

        let _span = bg_tracing::span_route_order(
            &req.symbol,
            side_str,
            order_type_str,
            req.quantity,
        );

        info!(
            "RouteOrder called: symbol={}, side={}, quantity={}, order_type={}",
            req.symbol, side_str, req.quantity, order_type_str
        );

        // Validate request
        self.validate_order(&req)?;

        // Generate client order ID
        let client_order_id = Self::generate_client_order_id();
        let submitted_at = chrono::Utc::now();

        // MVP: Save to database with PENDING_SUBMIT status (no actual FIX send)
        let side_str = match OrderSide::try_from(req.side) {
            Ok(OrderSide::Buy) => "BUY",
            Ok(OrderSide::Sell) => "SELL",
            _ => return Err(Status::invalid_argument("Invalid order side")),
        };

        let order_type_str = match OrderType::try_from(req.order_type) {
            Ok(OrderType::Market) => "MARKET",
            Ok(OrderType::Limit) => "LIMIT",
            Ok(OrderType::Stop) => "STOP",
            Ok(OrderType::StopLimit) => "STOP_LIMIT",
            _ => return Err(Status::invalid_argument("Invalid order type")),
        };

        let quantity_decimal = rust_decimal::Decimal::try_from(req.quantity).ok();
        let price_decimal = req.price.and_then(|p| rust_decimal::Decimal::try_from(p).ok());
        let stop_price_decimal = req.stop_price.and_then(|p| rust_decimal::Decimal::try_from(p).ok());
        let metadata_json = serde_json::to_value(&req.metadata).ok();

        sqlx::query(
            r#"
            INSERT INTO broker_orders
            (client_order_id, account_id, symbol, side, order_type, quantity, price, stop_price, status, metadata, submitted_at, created_at, updated_at)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, NOW(), NOW())
            "#,
        )
        .bind(&client_order_id)
        .bind(&req.account_id)
        .bind(&req.symbol)
        .bind(side_str)
        .bind(order_type_str)
        .bind(quantity_decimal)
        .bind(price_decimal)
        .bind(stop_price_decimal)
        .bind("PENDING_SUBMIT")
        .bind(metadata_json)
        .bind(submitted_at)
        .execute(&self.db_pool)
        .await
        .map_err(|e| {
            error!("Failed to insert order into database: {}", e);
            Status::internal(format!("Database error: {}", e))
        })?;

        info!(
            "Order saved to database: client_order_id={}, status=PENDING_SUBMIT",
            client_order_id
        );

        // Record metrics
        metrics::record_order_submitted(&req.symbol, order_type_str, side_str);
        metrics::record_order_latency(order_type_str, start.elapsed().as_secs_f64());

        // Record tracing
        bg_tracing::record_order_submitted(&client_order_id, "PENDING_SUBMIT");
        bg_tracing::record_latency(start);

        // Attempt to route through live broker if available
        {
            let guard = self.broker_client.read().await;
            if let Some(ct) = guard.as_ref() {
                let side = match OrderSide::try_from(req.side) {
                    Ok(OrderSide::Buy) => ctrader_openapi::proto::ProtoOaTradeSide::Buy,
                    Ok(OrderSide::Sell) => ctrader_openapi::proto::ProtoOaTradeSide::Sell,
                    _ => {
                        return Err(Status::invalid_argument("Invalid side"));
                    }
                };
                let ot = match OrderType::try_from(req.order_type) {
                    Ok(OrderType::Market) => ctrader_openapi::proto::ProtoOaOrderType::Market,
                    Ok(OrderType::Limit) => ctrader_openapi::proto::ProtoOaOrderType::Limit,
                    Ok(OrderType::Stop) => ctrader_openapi::proto::ProtoOaOrderType::Stop,
                    Ok(OrderType::StopLimit) => {
                        ctrader_openapi::proto::ProtoOaOrderType::StopLimit
                    }
                    _ => ctrader_openapi::proto::ProtoOaOrderType::Market,
                };

                let volume = convert_quantity_to_volume(req.quantity)?;
                let limit_price = req.price;
                let stop_price_val = req.stop_price;
                let comment = req.metadata.get("comment").cloned();

                match ct
                    .submit_order(
                        &req.symbol,
                        side,
                        volume,
                        ot,
                        limit_price,
                        stop_price_val,
                        None,
                        None,
                        comment,
                    )
                    .await
                {
                    Ok(resp_msg) => {
                        let broker_order_id = resp_msg
                            .client_msg_id
                            .unwrap_or_else(|| format!("ct-{}", resp_msg.payload_type));

                        // Update DB status to SUBMITTED
                        // CRITICAL: Order is already live at broker. If DB update
                        // fails, the order is real but DB shows PENDING_SUBMIT.
                        // Log prominently but do NOT fail the RPC.
                        if let Err(db_err) = sqlx::query(
                            "UPDATE broker_orders SET status = 'SUBMITTED', broker_order_id = $1, updated_at = NOW() WHERE client_order_id = $2",
                        )
                        .bind(&broker_order_id)
                        .bind(&client_order_id)
                        .execute(&self.db_pool)
                        .await
                        {
                            error!(
                                client_order_id = %client_order_id,
                                broker_order_id = %broker_order_id,
                                error = %db_err,
                                "CRITICAL: order submitted to broker but DB status update failed — \
                                 order is LIVE but database shows PENDING_SUBMIT. Manual reconciliation required."
                            );
                        }

                        info!(
                            broker_order_id = %broker_order_id,
                            client_order_id = %client_order_id,
                            "order routed to cTrader"
                        );

                        return Ok(Response::new(RouteOrderResponse {
                            broker_order_id,
                            client_order_id,
                            status: OrderStatus::Submitted as i32,
                            submitted_at: submitted_at.timestamp_nanos_opt().unwrap_or(0),
                            message: "Order submitted to cTrader".to_string(),
                        }));
                    }
                    Err(e) => {
                        warn!(error = %e, "cTrader order submission failed — order remains PENDING_SUBMIT");
                        // Fall through to return PENDING_SUBMIT status
                    }
                }
            }
        }

        // No broker connected or submission failed — return pending status
        Ok(Response::new(RouteOrderResponse {
            broker_order_id: String::new(),
            client_order_id: client_order_id.clone(),
            status: OrderStatus::PendingSubmit as i32,
            submitted_at: submitted_at.timestamp_nanos_opt().unwrap_or(0),
            message: format!(
                "Order queued (no live broker connected). ClOrdID: {}",
                client_order_id
            ),
        }))
    }

    #[instrument(skip(self), fields(client_order_id))]
    async fn cancel_order(
        &self,
        request: Request<CancelOrderRequest>,
    ) -> Result<Response<CancelOrderResponse>, Status> {
        let req = request.into_inner();

        // Create tracing span
        let _span = bg_tracing::span_cancel_order(&req.client_order_id, &req.account_id);

        info!(
            "CancelOrder called: client_order_id={}, account_id={}",
            req.client_order_id, req.account_id
        );

        // Fetch order from database
        let order_row: Option<(String,)> = sqlx::query_as(
            r#"
            SELECT status
            FROM broker_orders
            WHERE client_order_id = $1 AND account_id = $2
            "#,
        )
        .bind(&req.client_order_id)
        .bind(&req.account_id)
        .fetch_optional(&self.db_pool)
        .await
        .map_err(|e| {
            error!("Database error: {}", e);
            Status::internal(format!("Database error: {}", e))
        })?;

        let (status,) = order_row.ok_or_else(|| Status::not_found("Order not found"))?;

        // Check if order is cancellable
        if !matches!(
            status.as_str(),
            "PENDING_SUBMIT" | "SUBMITTED" | "PARTIALLY_FILLED"
        ) {
            return Err(Status::failed_precondition(format!(
                "Order cannot be cancelled (status: {})",
                status
            )));
        }

        // Update status to CANCEL_PENDING
        sqlx::query(
            r#"
            UPDATE broker_orders
            SET status = 'CANCEL_PENDING', updated_at = NOW()
            WHERE client_order_id = $1
            "#,
        )
        .bind(&req.client_order_id)
        .execute(&self.db_pool)
        .await
        .map_err(|e| {
            error!("Failed to update order status: {}", e);
            Status::internal(format!("Database error: {}", e))
        })?;

        // Attempt live broker cancellation if available
        {
            // Try to get the broker_order_id from DB for cTrader cancellation
            let broker_id_row: Option<(Option<String>,)> = sqlx::query_as(
                "SELECT broker_order_id FROM broker_orders WHERE client_order_id = $1",
            )
            .bind(&req.client_order_id)
            .fetch_optional(&self.db_pool)
            .await
            .ok()
            .flatten();

            if let Some((Some(broker_order_id),)) = broker_id_row {
                if let Ok(order_id) = broker_order_id.parse::<i64>() {
                    let guard = self.broker_client.read().await;
                    if let Some(ct) = guard.as_ref() {
                        match ct.cancel_order(order_id).await {
                            Ok(_) => {
                                // CRITICAL: Cancel is confirmed at broker. If DB
                                // update fails, the order is cancelled but DB
                                // still shows CANCEL_PENDING. Log prominently.
                                if let Err(db_err) = sqlx::query(
                                    "UPDATE broker_orders SET status = 'CANCELLED', updated_at = NOW() WHERE client_order_id = $1",
                                )
                                .bind(&req.client_order_id)
                                .execute(&self.db_pool)
                                .await
                                {
                                    error!(
                                        client_order_id = %req.client_order_id,
                                        error = %db_err,
                                        "CRITICAL: cancel confirmed at broker but DB status update failed — \
                                         order is CANCELLED but database shows CANCEL_PENDING. Manual reconciliation required."
                                    );
                                }

                                info!(
                                    client_order_id = %req.client_order_id,
                                    "order cancelled via cTrader"
                                );

                                return Ok(Response::new(CancelOrderResponse {
                                    success: true,
                                    message: "Order cancelled via cTrader".to_string(),
                                    new_status: OrderStatus::Cancelled as i32,
                                }));
                            }
                            Err(e) => {
                                warn!(error = %e, "cTrader cancel failed — order remains CANCEL_PENDING");
                            }
                        }
                    }
                }
            }
        }

        info!(
            "Order cancel request processed: client_order_id={}, new_status=CANCEL_PENDING",
            req.client_order_id
        );

        Ok(Response::new(CancelOrderResponse {
            success: true,
            message: format!(
                "Cancel request queued. Order: {}",
                req.client_order_id
            ),
            new_status: OrderStatus::CancelPending as i32,
        }))
    }

    #[instrument(skip(self), fields(account_id))]
    async fn get_account_state(
        &self,
        request: Request<GetAccountStateRequest>,
    ) -> Result<Response<GetAccountStateResponse>, Status> {
        let req = request.into_inner();
        info!("GetAccountState called: account_id={}", req.account_id);

        // Query live broker if available
        {
            let guard = self.broker_client.read().await;
            if let Some(ct) = guard.as_ref() {
                match ct.get_account_info().await {
                    Ok(info) => {
                        let balance = info.balance as f64 / 100.0; // cents → currency
                        return Ok(Response::new(GetAccountStateResponse {
                            account_id: ct.account_id().to_string(),
                            cash_balance: balance,
                            equity: balance,
                            margin_used: 0.0,
                            margin_available: balance,
                            buying_power: balance,
                            unrealized_pnl: 0.0,
                            realized_pnl: 0.0,
                            last_updated: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
                        }));
                    }
                    Err(e) => {
                        warn!(error = %e, "cTrader account info query failed");
                    }
                }
            }
        }

        warn!(
            account_id = %req.account_id,
            "no live broker connected — cannot query account state"
        );
        Err(Status::failed_precondition(
            "No live broker connected — account state unavailable",
        ))
    }

    #[instrument(skip(self), fields(account_id))]
    async fn get_positions(
        &self,
        request: Request<GetPositionsRequest>,
    ) -> Result<Response<GetPositionsResponse>, Status> {
        let req = request.into_inner();
        info!(
            "GetPositions called: account_id={}, symbol={:?}",
            req.account_id, req.symbol
        );

        // Query live broker if available
        {
            let guard = self.broker_client.read().await;
            if let Some(ct) = guard.as_ref() {
                match ct.get_positions().await {
                    Ok(reconcile) => {
                        let positions: Vec<Position> = reconcile
                            .positions
                            .iter()
                            .map(|p| {
                                let td = &p.trade_data;
                                let qty = td.volume as f64 / 100.0;
                                let price = p.price.unwrap_or(0.0);
                                // Negative quantity for sell-side positions
                                let signed_qty = if td.trade_side == 2 { -qty } else { qty };
                                Position {
                                    symbol: td.symbol_id.to_string(),
                                    quantity: signed_qty,
                                    average_price: price,
                                    market_value: qty * price,
                                    unrealized_pnl: 0.0,
                                }
                            })
                            .collect();

                        return Ok(Response::new(GetPositionsResponse {
                            positions,
                            total_equity: 0.0,
                            total_exposure: 0.0,
                            leverage_ratio: 0.0,
                            timestamp: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
                        }));
                    }
                    Err(e) => {
                        warn!(error = %e, "cTrader position query failed");
                    }
                }
            }
        }

        warn!(
            account_id = %req.account_id,
            symbol = ?req.symbol,
            "no live broker connected — cannot query positions"
        );
        Err(Status::failed_precondition(
            "No live broker connected — position query unavailable",
        ))
    }

    #[instrument(skip(self))]
    async fn get_session_status(
        &self,
        request: Request<GetSessionStatusRequest>,
    ) -> Result<Response<GetSessionStatusResponse>, Status> {
        let req = request.into_inner();
        let session_id = req.session_id.unwrap_or_else(|| "FOXHUNT-CQG".to_string());

        info!("GetSessionStatus called: session_id={}", session_id);

        let state = *self.session_state.read().await;

        // MVP: Return simulated session state
        let response = GetSessionStatusResponse {
            session_id: session_id.clone(),
            state: state as i32,
            sender_seq_num: 1, // Placeholder
            target_seq_num: 1, // Placeholder
            last_heartbeat_sent: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
            last_heartbeat_received: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
            heartbeat_rtt_ms: 0.0,
            connected_at: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
            details: vec![("mvp_mode".to_string(), "true".to_string())]
                .into_iter()
                .collect(),
        };

        Ok(Response::new(response))
    }

    type StreamExecutionsStream =
        tokio_stream::wrappers::ReceiverStream<Result<ExecutionEvent, Status>>;

    #[instrument(skip(self))]
    async fn stream_executions(
        &self,
        request: Request<StreamExecutionsRequest>,
    ) -> Result<Response<Self::StreamExecutionsStream>, Status> {
        let req = request.into_inner();
        info!(
            "StreamExecutions called: account_id={:?}, symbol={:?}",
            req.account_id, req.symbol
        );

        let (tx, rx) = tokio::sync::mpsc::channel(16);

        // Stream live execution events if broker is connected
        {
            let guard = self.broker_client.read().await;
            if let Some(ct) = guard.as_ref() {
                let mut broadcast_rx = ct.subscribe_executions();
                let tx_clone = tx.clone();
                tokio::spawn(async move {
                    loop {
                        match broadcast_rx.recv().await {
                            Ok(msg) => {
                                let event = ExecutionEvent {
                                    execution_id: msg
                                        .client_msg_id
                                        .clone()
                                        .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
                                    broker_order_id: String::new(),
                                    client_order_id: msg
                                        .client_msg_id
                                        .unwrap_or_default(),
                                    symbol: String::new(),
                                    side: OrderSide::Buy as i32,
                                    exec_type: ExecutionType::Trade as i32,
                                    order_status: OrderStatus::Filled as i32,
                                    last_qty: 0.0,
                                    last_price: 0.0,
                                    cum_qty: 0.0,
                                    avg_price: 0.0,
                                    transact_time: chrono::Utc::now()
                                        .timestamp_nanos_opt()
                                        .unwrap_or(0),
                                    text: None,
                                };
                                if tx_clone.send(Ok(event)).await.is_err() {
                                    break;
                                }
                            }
                            Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
                            Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
                                warn!(skipped = n, "execution broadcast lagged");
                            }
                        }
                    }
                });

                // Don't drop tx yet — the spawned task holds tx_clone
                drop(tx);

                return Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new(
                    rx,
                )));
            }
        }

        // No broker connected — close stream immediately
        tokio::spawn(async move {
            warn!("StreamExecutions: no live broker connected");
            drop(tx);
        });

        Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new(
            rx,
        )))
    }

    #[instrument(skip(self))]
    async fn health_check(
        &self,
        _request: Request<HealthCheckRequest>,
    ) -> Result<Response<HealthCheckResponse>, Status> {
        info!("HealthCheck called");

        // Check database connection
        let db_healthy = sqlx::query("SELECT 1")
            .fetch_optional(&self.db_pool)
            .await
            .is_ok();

        let broker_connected = {
            let guard = self.broker_client.read().await;
            guard.is_some()
        };

        let healthy = db_healthy;
        let message = if healthy && broker_connected {
            "Broker Gateway Service is healthy (cTrader connected)".to_string()
        } else if healthy {
            "Broker Gateway Service is healthy (no live broker)".to_string()
        } else {
            "Service degraded: database connection failed".to_string()
        };

        let details = vec![
            ("database".to_string(), db_healthy.to_string()),
            ("broker_connected".to_string(), broker_connected.to_string()),
        ]
        .into_iter()
        .collect();

        Ok(Response::new(HealthCheckResponse {
            healthy,
            message,
            details,
        }))
    }

    type StreamAccountStateStream =
        Pin<Box<dyn Stream<Item = Result<GetAccountStateResponse, Status>> + Send>>;

    async fn stream_account_state(
        &self,
        request: Request<StreamAccountStateRequest>,
    ) -> Result<Response<Self::StreamAccountStateStream>, Status> {
        let req = request.into_inner();
        let interval_secs = if req.interval_seconds == 0 { 3 } else { req.interval_seconds };
        let broker_client = Arc::clone(&self.broker_client);

        let stream = async_stream::stream! {
            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
            loop {
                interval.tick().await;

                // Extract data while holding the read lock, then drop it before
                // yielding so the lock is never held across a suspension point.
                let result = {
                    let guard = broker_client.read().await;
                    match guard.as_ref() {
                        Some(ct) => {
                            let account_id = ct.account_id().to_string();
                            match ct.get_account_info().await {
                                Ok(info) => {
                                    let balance = info.balance as f64 / 100.0;
                                    Ok(GetAccountStateResponse {
                                        account_id,
                                        cash_balance: balance,
                                        equity: balance,
                                        margin_used: 0.0,
                                        margin_available: balance,
                                        buying_power: balance,
                                        unrealized_pnl: 0.0,
                                        realized_pnl: 0.0,
                                        last_updated: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
                                    })
                                }
                                Err(e) => {
                                    warn!(error = %e, "StreamAccountState: broker query failed");
                                    Err(Status::unavailable(format!("Broker query failed: {e}")))
                                }
                            }
                        }
                        None => Err(Status::failed_precondition("No live broker connected")),
                    }
                    // guard dropped here
                };

                match result {
                    Ok(resp) => { yield Ok(resp); }
                    Err(status) => {
                        yield Err(status);
                        break;
                    }
                }
            }
        };

        Ok(Response::new(Box::pin(stream)))
    }

    type StreamSessionStatusStream =
        Pin<Box<dyn Stream<Item = Result<GetSessionStatusResponse, Status>> + Send>>;

    async fn stream_session_status(
        &self,
        request: Request<StreamSessionStatusRequest>,
    ) -> Result<Response<Self::StreamSessionStatusStream>, Status> {
        let req = request.into_inner();
        let interval_secs = if req.interval_seconds == 0 { 5 } else { req.interval_seconds };
        let session_state = Arc::clone(&self.session_state);

        let connected_at = chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0);
        let stream = async_stream::stream! {
            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
            loop {
                interval.tick().await;
                let state = *session_state.read().await;
                yield Ok(GetSessionStatusResponse {
                    session_id: "FOXHUNT-CQG".to_string(),
                    state: state as i32,
                    sender_seq_num: 1,
                    target_seq_num: 1,
                    last_heartbeat_sent: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
                    last_heartbeat_received: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
                    heartbeat_rtt_ms: 0.0,
                    connected_at,
                    details: vec![("mvp_mode".to_string(), "true".to_string())]
                        .into_iter()
                        .collect(),
                });
            }
        };

        Ok(Response::new(Box::pin(stream)))
    }
}

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

    #[test]
    fn test_convert_quantity_normal() {
        // 1.0 lot = 100,000 volume
        assert_eq!(convert_quantity_to_volume(1.0).ok(), Some(100_000));
        // 0.01 lot (micro lot) = 1,000 volume
        assert_eq!(convert_quantity_to_volume(0.01).ok(), Some(1_000));
        // 10.5 lots = 1,050,000 volume
        assert_eq!(convert_quantity_to_volume(10.5).ok(), Some(1_050_000));
    }

    #[test]
    fn test_convert_quantity_fractional_rounding() {
        // 0.011 lots should round to 1,100 (not truncate to 1,099)
        assert_eq!(convert_quantity_to_volume(0.011).ok(), Some(1_100));
        // 0.015 lots = 1,500.0 exactly
        assert_eq!(convert_quantity_to_volume(0.015).ok(), Some(1_500));
    }

    #[test]
    fn test_convert_quantity_overflow() {
        // Huge value should overflow i64
        let result = convert_quantity_to_volume(f64::MAX);
        assert!(result.is_err());
        let err = result.err().map(|s| s.message().to_string());
        assert!(err.as_deref().unwrap_or("").contains("Volume overflow"));
    }

    #[test]
    fn test_convert_quantity_negative() {
        let result = convert_quantity_to_volume(-1.0);
        assert!(result.is_err());
    }

    #[test]
    fn test_convert_quantity_nan() {
        let result = convert_quantity_to_volume(f64::NAN);
        assert!(result.is_err());
    }

    #[test]
    fn test_convert_quantity_infinity() {
        let result = convert_quantity_to_volume(f64::INFINITY);
        assert!(result.is_err());
        let result_neg = convert_quantity_to_volume(f64::NEG_INFINITY);
        assert!(result_neg.is_err());
    }

    #[test]
    fn test_convert_quantity_zero() {
        // Zero is technically valid (non-negative, finite)
        assert_eq!(convert_quantity_to_volume(0.0).ok(), Some(0));
    }
}
