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

//! Data Acquisition Service Proxy - Zero-copy gRPC forwarding
//!
//! Forwards to data-acquisition-service:50057 (separate service).
//! All RPCs are unary (no streaming) — simplest proxy.

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

use crate::data_acquisition::data_acquisition_service_client::DataAcquisitionServiceClient;
use crate::data_acquisition::data_acquisition_service_server::DataAcquisitionService;
use crate::data_acquisition::{
    CancelDownloadRequest, CancelDownloadResponse, GetDownloadStatusRequest,
    GetDownloadStatusResponse, HealthCheckRequest, HealthCheckResponse,
    ListDownloadJobsRequest, ListDownloadJobsResponse, ScheduleDownloadRequest,
    ScheduleDownloadResponse, StreamDownloadStatusRequest,
};

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

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

#[tonic::async_trait]
impl DataAcquisitionService for DataAcquisitionProxy {
    #[instrument(skip(self, request), err)]
    async fn schedule_download(
        &self,
        request: Request<ScheduleDownloadRequest>,
    ) -> Result<Response<ScheduleDownloadResponse>, Status> {
        self.client
            .clone()
            .schedule_download(request)
            .await
            .map_err(|e| {
                error!("Backend ScheduleDownload failed: {}", e);
                e
            })
    }

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

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

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

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

    type StreamDownloadStatusStream =
        Pin<Box<dyn Stream<Item = Result<ListDownloadJobsResponse, Status>> + Send>>;

    #[instrument(skip(self, request), err)]
    async fn stream_download_status(
        &self,
        request: Request<StreamDownloadStatusRequest>,
    ) -> Result<Response<Self::StreamDownloadStatusStream>, Status> {
        let req = request.into_inner();
        let interval_secs = if req.interval_seconds == 0 {
            5
        } else {
            req.interval_seconds.clamp(1, 60)
        };
        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.list_download_jobs(Request::new(ListDownloadJobsRequest::default())).await {
                    Ok(resp) => yield Ok(resp.into_inner()),
                    Err(e) => {
                        error!("Backend ListDownloadJobs failed: {}", e);
                        yield Err(e);
                        break;
                    }
                }
            }
        };

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