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

//! ML Inference Service Proxy - Zero-copy gRPC forwarding for ml.MLService
//!
//! Forwards to trading-service backend (ML inference lives in same process).

use futures::Stream;
use std::pin::Pin;
use std::time::Duration;
use tonic::{Request, Response, Status};
use tracing::{error, instrument};

use crate::ml_inference::ml_service_client::MlServiceClient;
use crate::ml_inference::ml_service_server::MlService;
use crate::ml_inference::{
    GetAvailableModelsRequest, GetAvailableModelsResponse, GetEnsembleVoteRequest,
    GetEnsembleVoteResponse, GetFeatureImportanceRequest, GetFeatureImportanceResponse,
    GetModelPerformanceRequest, GetModelPerformanceResponse, GetModelStatusRequest,
    GetModelStatusResponse, GetPredictionRequest, GetPredictionResponse, ModelMetricsEvent,
    PredictionEvent, RetrainModelRequest, RetrainModelResponse, SignalStrengthEvent,
    StreamModelMetricsRequest, StreamModelStatusRequest, StreamPredictionsRequest,
    StreamSignalStrengthRequest,
};

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

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

#[tonic::async_trait]
impl MlService for MlServiceProxy {
    type StreamPredictionsStream =
        Pin<Box<dyn Stream<Item = Result<PredictionEvent, Status>> + Send>>;
    type StreamModelMetricsStream =
        Pin<Box<dyn Stream<Item = Result<ModelMetricsEvent, Status>> + Send>>;
    type StreamSignalStrengthStream =
        Pin<Box<dyn Stream<Item = Result<SignalStrengthEvent, Status>> + Send>>;
    type StreamModelStatusStream =
        Pin<Box<dyn Stream<Item = Result<GetModelStatusResponse, Status>> + Send>>;

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

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

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

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

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

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

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

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

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

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

    #[instrument(skip(self, request), err)]
    async fn stream_model_status(
        &self,
        request: Request<StreamModelStatusRequest>,
    ) -> Result<Response<Self::StreamModelStatusStream>, Status> {
        let req = request.into_inner();
        let interval_secs = if req.interval_seconds == 0 {
            5
        } else {
            req.interval_seconds.clamp(1, 60)
        };
        let model_name = req.model_name;
        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_model_status(Request::new(GetModelStatusRequest {
                    model_name: if model_name.is_empty() { None } else { Some(model_name.clone()) },
                })).await {
                    Ok(resp) => yield Ok(resp.into_inner()),
                    Err(e) => {
                        error!("Backend GetModelStatus failed: {}", e);
                        yield Err(e);
                        break;
                    }
                }
            }
        };

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