//! Broker Gateway Service Proxy - Zero-copy gRPC forwarding for broker_gateway.BrokerGatewayService
//!
//! Forwards to broker-gateway:50056 (separate service).

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

use crate::broker_gateway::broker_gateway_service_client::BrokerGatewayServiceClient;
use crate::broker_gateway::broker_gateway_service_server::BrokerGatewayService;
use crate::broker_gateway::{
    CancelOrderRequest, CancelOrderResponse, ExecutionEvent, GetAccountStateRequest,
    GetAccountStateResponse, GetPositionsRequest, GetPositionsResponse,
    GetSessionStatusRequest, GetSessionStatusResponse, HealthCheckRequest, HealthCheckResponse,
    RouteOrderRequest, RouteOrderResponse, StreamAccountStateRequest, StreamExecutionsRequest,
    StreamSessionStatusRequest,
};

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

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

#[tonic::async_trait]
impl BrokerGatewayService for BrokerGatewayProxy {
    type StreamExecutionsStream =
        Pin<Box<dyn Stream<Item = Result<ExecutionEvent, Status>> + Send>>;

    #[instrument(skip(self, request), err)]
    async fn route_order(
        &self,
        request: Request<RouteOrderRequest>,
    ) -> Result<Response<RouteOrderResponse>, Status> {
        self.client.clone().route_order(request).await.map_err(|e| {
            error!("Backend RouteOrder 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_account_state(
        &self,
        request: Request<GetAccountStateRequest>,
    ) -> Result<Response<GetAccountStateResponse>, Status> {
        self.client
            .clone()
            .get_account_state(request)
            .await
            .map_err(|e| {
                error!("Backend GetAccountState failed: {}", e);
                e
            })
    }

    #[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 get_session_status(
        &self,
        request: Request<GetSessionStatusRequest>,
    ) -> Result<Response<GetSessionStatusResponse>, Status> {
        self.client
            .clone()
            .get_session_status(request)
            .await
            .map_err(|e| {
                error!("Backend GetSessionStatus 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 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 StreamAccountStateStream =
        Pin<Box<dyn Stream<Item = Result<GetAccountStateResponse, Status>> + Send>>;

    #[instrument(skip(self, request), err)]
    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.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.get_account_state(Request::new(GetAccountStateRequest {
                    account_id: String::new(),
                })).await {
                    Ok(resp) => yield Ok(resp.into_inner()),
                    Err(e) => {
                        error!("Backend GetAccountState failed: {}", e);
                        yield Err(e);
                        break;
                    }
                }
            }
        };

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

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

    #[instrument(skip(self, request), err)]
    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.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.get_session_status(Request::new(GetSessionStatusRequest::default())).await {
                    Ok(resp) => yield Ok(resp.into_inner()),
                    Err(e) => {
                        error!("Backend GetSessionStatus failed: {}", e);
                        yield Err(e);
                        break;
                    }
                }
            }
        };

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