foxhunt/services/api/src/grpc/trading_direct_proxy.rs

//! Trading Service Direct Proxy - Zero-copy gRPC forwarding for trading.TradingService
//!
//! Forwards requests from fxt CLI (package: trading) directly to trading-service backend.
//! Distinct from TradingServiceProxy which serves the monolithic foxhunt.tli.TradingService.

use std::pin::Pin;
use std::time::Duration;

use futures::Stream;
use tonic::{Request, Response, Status};
use tracing::{error, instrument};

use crate::trading_backend::trading_service_client::TradingServiceClient;
use crate::trading_backend::trading_service_server::TradingService;
use crate::trading_backend::{
    CancelOrderRequest, CancelOrderResponse, ExecutionEvent, GetExecutionHistoryRequest,
    GetExecutionHistoryResponse, GetOrderBookRequest, GetOrderBookResponse,
    GetOrderStatusRequest, GetOrderStatusResponse, GetPositionsRequest, GetPositionsResponse,
    GetPortfolioSummaryRequest, GetPortfolioSummaryResponse, GetRegimeStateRequest,
    GetRegimeStateResponse, GetRegimeTransitionsRequest, GetRegimeTransitionsResponse,
    MarketDataEvent, MlOrderRequest, MlOrderResponse, MlPerformanceRequest,
    MlPerformanceResponse, MlPredictionsRequest, MlPredictionsResponse, OrderEvent,
    PositionEvent, StreamExecutionsRequest, StreamMarketDataRequest, StreamOrderBookRequest,
    StreamOrdersRequest, StreamPortfolioSummaryRequest, StreamPositionsRequest,
    SubmitOrderRequest, SubmitOrderResponse,
};

#[derive(Debug, Clone)]
pub struct TradingDirectProxy {
    client: TradingServiceClient<tonic::transport::Channel>,
}

impl TradingDirectProxy {
    pub fn new(client: TradingServiceClient<tonic::transport::Channel>) -> Self {
        Self { client }
    }
}

#[tonic::async_trait]
impl TradingService for TradingDirectProxy {
    type StreamOrdersStream = Pin<Box<dyn Stream<Item = Result<OrderEvent, Status>> + Send>>;
    type StreamPositionsStream =
        Pin<Box<dyn Stream<Item = Result<PositionEvent, Status>> + Send>>;
    type StreamMarketDataStream =
        Pin<Box<dyn Stream<Item = Result<MarketDataEvent, Status>> + Send>>;
    type StreamExecutionsStream =
        Pin<Box<dyn Stream<Item = Result<ExecutionEvent, Status>> + Send>>;
    type StreamPortfolioSummaryStream =
        Pin<Box<dyn Stream<Item = Result<GetPortfolioSummaryResponse, Status>> + Send>>;
    type StreamOrderBookStream =
        Pin<Box<dyn Stream<Item = Result<GetOrderBookResponse, Status>> + Send>>;

    #[instrument(skip(self, request), err)]
    async fn submit_order(
        &self,
        request: Request<SubmitOrderRequest>,
    ) -> Result<Response<SubmitOrderResponse>, Status> {
        self.client.clone().submit_order(request).await.map_err(|e| {
            error!("Backend SubmitOrder failed: {}", e);
            e
        })
    }

    #[instrument(skip(self, request), err)]
    async fn cancel_order(
        &self,
        request: Request<CancelOrderRequest>,
    ) -> Result<Response<CancelOrderResponse>, Status> {
        self.client.clone().cancel_order(request).await.map_err(|e| {
            error!("Backend CancelOrder failed: {}", e);
            e
        })
    }

    #[instrument(skip(self, request), err)]
    async fn get_order_status(
        &self,
        request: Request<GetOrderStatusRequest>,
    ) -> Result<Response<GetOrderStatusResponse>, Status> {
        self.client.clone().get_order_status(request).await.map_err(|e| {
            error!("Backend GetOrderStatus failed: {}", e);
            e
        })
    }

    #[instrument(skip(self, request), err)]
    async fn stream_orders(
        &self,
        request: Request<StreamOrdersRequest>,
    ) -> Result<Response<Self::StreamOrdersStream>, Status> {
        let stream = self.client.clone().stream_orders(request).await.map_err(|e| {
            error!("Backend StreamOrders failed: {}", e);
            e
        })?;
        Ok(Response::new(Box::pin(stream.into_inner())))
    }

    #[instrument(skip(self, request), err)]
    async fn get_positions(
        &self,
        request: Request<GetPositionsRequest>,
    ) -> Result<Response<GetPositionsResponse>, Status> {
        self.client.clone().get_positions(request).await.map_err(|e| {
            error!("Backend GetPositions failed: {}", e);
            e
        })
    }

    #[instrument(skip(self, request), err)]
    async fn stream_positions(
        &self,
        request: Request<StreamPositionsRequest>,
    ) -> Result<Response<Self::StreamPositionsStream>, Status> {
        let stream = self.client.clone().stream_positions(request).await.map_err(|e| {
            error!("Backend StreamPositions failed: {}", e);
            e
        })?;
        Ok(Response::new(Box::pin(stream.into_inner())))
    }

    #[instrument(skip(self, request), err)]
    async fn get_portfolio_summary(
        &self,
        request: Request<GetPortfolioSummaryRequest>,
    ) -> Result<Response<GetPortfolioSummaryResponse>, Status> {
        self.client
            .clone()
            .get_portfolio_summary(request)
            .await
            .map_err(|e| {
                error!("Backend GetPortfolioSummary failed: {}", e);
                e
            })
    }

    #[instrument(skip(self, request), err)]
    async fn stream_market_data(
        &self,
        request: Request<StreamMarketDataRequest>,
    ) -> Result<Response<Self::StreamMarketDataStream>, Status> {
        let stream = self
            .client
            .clone()
            .stream_market_data(request)
            .await
            .map_err(|e| {
                error!("Backend StreamMarketData failed: {}", e);
                e
            })?;
        Ok(Response::new(Box::pin(stream.into_inner())))
    }

    #[instrument(skip(self, request), err)]
    async fn get_order_book(
        &self,
        request: Request<GetOrderBookRequest>,
    ) -> Result<Response<GetOrderBookResponse>, Status> {
        self.client.clone().get_order_book(request).await.map_err(|e| {
            error!("Backend GetOrderBook failed: {}", e);
            e
        })
    }

    #[instrument(skip(self, request), err)]
    async fn stream_executions(
        &self,
        request: Request<StreamExecutionsRequest>,
    ) -> Result<Response<Self::StreamExecutionsStream>, Status> {
        let stream = self
            .client
            .clone()
            .stream_executions(request)
            .await
            .map_err(|e| {
                error!("Backend StreamExecutions failed: {}", e);
                e
            })?;
        Ok(Response::new(Box::pin(stream.into_inner())))
    }

    #[instrument(skip(self, request), err)]
    async fn get_execution_history(
        &self,
        request: Request<GetExecutionHistoryRequest>,
    ) -> Result<Response<GetExecutionHistoryResponse>, Status> {
        self.client
            .clone()
            .get_execution_history(request)
            .await
            .map_err(|e| {
                error!("Backend GetExecutionHistory failed: {}", e);
                e
            })
    }

    #[instrument(skip(self, request), err)]
    async fn submit_ml_order(
        &self,
        request: Request<MlOrderRequest>,
    ) -> Result<Response<MlOrderResponse>, Status> {
        self.client.clone().submit_ml_order(request).await.map_err(|e| {
            error!("Backend SubmitMLOrder failed: {}", e);
            e
        })
    }

    #[instrument(skip(self, request), err)]
    async fn get_ml_predictions(
        &self,
        request: Request<MlPredictionsRequest>,
    ) -> Result<Response<MlPredictionsResponse>, Status> {
        self.client
            .clone()
            .get_ml_predictions(request)
            .await
            .map_err(|e| {
                error!("Backend GetMLPredictions failed: {}", e);
                e
            })
    }

    #[instrument(skip(self, request), err)]
    async fn get_ml_performance(
        &self,
        request: Request<MlPerformanceRequest>,
    ) -> Result<Response<MlPerformanceResponse>, Status> {
        self.client
            .clone()
            .get_ml_performance(request)
            .await
            .map_err(|e| {
                error!("Backend GetMLPerformance failed: {}", e);
                e
            })
    }

    #[instrument(skip(self, request), err)]
    async fn get_regime_state(
        &self,
        request: Request<GetRegimeStateRequest>,
    ) -> Result<Response<GetRegimeStateResponse>, Status> {
        self.client
            .clone()
            .get_regime_state(request)
            .await
            .map_err(|e| {
                error!("Backend GetRegimeState failed: {}", e);
                e
            })
    }

    #[instrument(skip(self, request), err)]
    async fn get_regime_transitions(
        &self,
        request: Request<GetRegimeTransitionsRequest>,
    ) -> Result<Response<GetRegimeTransitionsResponse>, Status> {
        self.client
            .clone()
            .get_regime_transitions(request)
            .await
            .map_err(|e| {
                error!("Backend GetRegimeTransitions failed: {}", e);
                e
            })
    }

    #[instrument(skip(self, request), err)]
    async fn stream_portfolio_summary(
        &self,
        request: Request<StreamPortfolioSummaryRequest>,
    ) -> Result<Response<Self::StreamPortfolioSummaryStream>, Status> {
        let req = request.into_inner();
        let interval_secs = if req.interval_seconds == 0 {
            3
        } else {
            req.interval_seconds.clamp(1, 60)
        };
        let account_id = req.account_id;
        let mut client = self.client.clone();

        let stream = async_stream::stream! {
            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
            loop {
                interval.tick().await;
                match client.get_portfolio_summary(Request::new(GetPortfolioSummaryRequest {
                    account_id: account_id.clone(),
                })).await {
                    Ok(resp) => yield Ok(resp.into_inner()),
                    Err(e) => {
                        error!("Backend GetPortfolioSummary failed: {}", e);
                        yield Err(e);
                        break;
                    }
                }
            }
        };

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

    #[instrument(skip(self, request), err)]
    async fn stream_order_book(
        &self,
        request: Request<StreamOrderBookRequest>,
    ) -> Result<Response<Self::StreamOrderBookStream>, Status> {
        let req = request.into_inner();
        let interval_secs = if req.interval_seconds == 0 {
            1
        } else {
            req.interval_seconds.clamp(1, 60)
        };
        let symbol = req.symbol;
        let depth = req.depth;
        let mut client = self.client.clone();

        let stream = async_stream::stream! {
            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
            loop {
                interval.tick().await;
                match client.get_order_book(Request::new(GetOrderBookRequest {
                    symbol: symbol.clone(),
                    depth: if depth == 0 { None } else { Some(depth) },
                })).await {
                    Ok(resp) => yield Ok(resp.into_inner()),
                    Err(e) => {
                        error!("Backend GetOrderBook failed: {}", e);
                        yield Err(e);
                        break;
                    }
                }
            }
        };

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