foxhunt/services/api_gateway/src/grpc/backtesting_proxy.rs

//! Backtesting Service Proxy - Zero-copy gRPC forwarding
//!
//! This module implements a high-performance proxy for the backtesting service.
//! Design goals:
//! - <10μs routing overhead
//! - Zero-copy forwarding where possible
//! - Connection pooling via tonic::transport::Channel
//! - Circuit breaker for backend failures
//! - Health checking of backend service

use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use tonic::transport::{Certificate, ClientTlsConfig, Identity};
use tonic::{Request, Response, Status};
use tonic_health::pb::health_client::HealthClient;
use tracing::{debug, error, info, warn, instrument};

// Import generated protobuf types from build.rs
use crate::foxhunt::tli::{
    backtesting_service_client::BacktestingServiceClient,
    backtesting_service_server::BacktestingService,
    BacktestProgressEvent,
    GetBacktestResultsRequest,
    GetBacktestResultsResponse,
    GetBacktestStatusRequest,
    GetBacktestStatusResponse,
    ListBacktestsRequest,
    ListBacktestsResponse,
    // Request/Response types
    StartBacktestRequest,
    StartBacktestResponse,
    StopBacktestRequest,
    StopBacktestResponse,
    SubscribeBacktestProgressRequest,
};

/// Health check state for circuit breaker
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HealthState {
    Healthy,
    Degraded,
    Unhealthy,
}

/// Health checker for backend service with circuit breaker
pub struct HealthChecker {
    state: RwLock<HealthState>,
    consecutive_failures: RwLock<u32>,
    last_health_check: RwLock<Instant>,
    failure_threshold: u32,
    health_check_interval: Duration,
}

impl HealthChecker {
    pub fn new(failure_threshold: u32, health_check_interval: Duration) -> Self {
        Self {
            state: RwLock::new(HealthState::Healthy),
            consecutive_failures: RwLock::new(0),
            last_health_check: RwLock::new(Instant::now()),
            failure_threshold,
            health_check_interval,
        }
    }

    /// Record a successful request
    pub async fn record_success(&self) {
        let mut failures = self.consecutive_failures.write().await;
        *failures = 0;

        let mut state = self.state.write().await;
        if *state != HealthState::Healthy {
            info!("Backtesting service backend recovered to healthy state");
            *state = HealthState::Healthy;
        }
    }

    /// Record a failed request
    pub async fn record_failure(&self) {
        let mut failures = self.consecutive_failures.write().await;
        *failures += 1;

        let mut state = self.state.write().await;

        if *failures >= self.failure_threshold && *state != HealthState::Unhealthy {
            error!(
                "Backtesting service backend marked as unhealthy after {} consecutive failures",
                failures
            );
            *state = HealthState::Unhealthy;
        } else if *failures >= self.failure_threshold / 2 && *state == HealthState::Healthy {
            warn!(
                "Backtesting service backend degraded after {} failures",
                failures
            );
            *state = HealthState::Degraded;
        }
    }

    /// Check if backend is healthy enough to accept requests
    pub async fn is_healthy(&self) -> bool {
        let state = self.state.read().await;
        *state != HealthState::Unhealthy
    }

    /// Get current health state
    pub async fn get_state(&self) -> HealthState {
        *self.state.read().await
    }

    /// Perform proactive gRPC health check using tonic-health
    pub async fn perform_health_check(&self, channel: &tonic::transport::Channel) {
        // Check if enough time has passed since last check
        let now = Instant::now();
        let should_check = {
            let last_check = self.last_health_check.read().await;
            now.duration_since(*last_check) >= self.health_check_interval
        };

        if !should_check {
            return;
        }

        // Update last check time
        {
            let mut last_check = self.last_health_check.write().await;
            *last_check = now;
        }

        // Perform gRPC health check
        let mut health_client = HealthClient::new(channel.clone());
        let request = tonic_health::pb::HealthCheckRequest {
            service: "foxhunt.tli.BacktestingService".to_string(),
        };

        match health_client.check(request).await {
            Ok(response) => {
                let status = response.into_inner().status;
                if status == tonic_health::pb::health_check_response::ServingStatus::Serving as i32
                {
                    self.record_success().await;
                    debug!("Backtesting service health check: SERVING");
                } else {
                    self.record_failure().await;
                    warn!(
                        "Backtesting service health check: NOT_SERVING (status: {})",
                        status
                    );
                }
            },
            Err(e) => {
                self.record_failure().await;
                error!("Backtesting service health check failed: {}", e);
            },
        }
    }
}

/// Backtesting service proxy with zero-copy forwarding
pub struct BacktestingServiceProxy {
    /// Client connection to backend service
    /// tonic::transport::Channel is cheap to clone and reuses connections
    client: BacktestingServiceClient<tonic::transport::Channel>,

    /// Health checker with circuit breaker
    health_checker: Arc<HealthChecker>,

    /// Backend service URL for logging
    backend_url: String,

    /// Underlying channel for health checks
    channel: tonic::transport::Channel,
}

impl BacktestingServiceProxy {
    /// Create a new backtesting service proxy with TLS/mTLS support
    ///
    /// # Arguments
    /// * `backend_url` - URL of the backend backtesting service (e.g., "https://localhost:50053")
    /// * `ca_cert_path` - Path to CA certificate for server verification
    /// * `client_cert_path` - Path to client certificate for mTLS authentication
    /// * `client_key_path` - Path to client private key for mTLS authentication
    ///
    /// # Returns
    /// * `Result<Self>` - Proxy instance or connection error
    pub async fn new(
        backend_url: &str,
        ca_cert_path: Option<&str>,
        client_cert_path: Option<&str>,
        client_key_path: Option<&str>,
    ) -> Result<Self, Box<dyn std::error::Error>> {
        info!(
            "Connecting to backtesting service backend at {}",
            backend_url
        );

        // Establish connection to backend service
        // tonic::transport::Channel automatically handles:
        // - Connection pooling
        // - Keep-alive
        // - Load balancing (if multiple endpoints provided)
        let mut endpoint = tonic::transport::Endpoint::from_shared(backend_url.to_string())?
            .connect_timeout(Duration::from_secs(5))
            .timeout(Duration::from_secs(30))
            .tcp_keepalive(Some(Duration::from_secs(60)))
            .http2_keep_alive_interval(Duration::from_secs(30))
            .keep_alive_while_idle(true);

        // Configure TLS if certificate paths provided (HTTPS URLs)
        if backend_url.starts_with("https://") {
            match (ca_cert_path, client_cert_path, client_key_path) {
                (Some(ca_path), Some(cert_path), Some(key_path)) => {
                    info!("Configuring TLS with mTLS (client certificates)");
                    info!("  CA cert: {}", ca_path);
                    info!("  Client cert: {}", cert_path);
                    info!("  Client key: {}", key_path);

                    // Load certificates from files
                    info!("Reading CA certificate...");
                    let ca_pem = tokio::fs::read_to_string(ca_path).await.map_err(|e| {
                        error!("Failed to read CA certificate at {}: {}", ca_path, e);
                        format!("Failed to read CA certificate at {}: {}", ca_path, e)
                    })?;
                    info!("CA certificate loaded ({} bytes)", ca_pem.len());

                    info!("Reading client certificate...");
                    let client_cert_pem =
                        tokio::fs::read_to_string(cert_path).await.map_err(|e| {
                            error!("Failed to read client certificate at {}: {}", cert_path, e);
                            format!("Failed to read client certificate at {}: {}", cert_path, e)
                        })?;
                    info!(
                        "Client certificate loaded ({} bytes)",
                        client_cert_pem.len()
                    );

                    info!("Reading client key...");
                    let client_key_pem =
                        tokio::fs::read_to_string(key_path).await.map_err(|e| {
                            error!("Failed to read client key at {}: {}", key_path, e);
                            format!("Failed to read client key at {}: {}", key_path, e)
                        })?;
                    info!("Client key loaded ({} bytes)", client_key_pem.len());

                    // Create TLS configuration with mTLS (client authentication)
                    // Extract hostname from backend_url for SNI (Server Name Indication)
                    let hostname = if let Some(host) = backend_url.strip_prefix("https://") {
                        host.split(':').next().unwrap_or("backtesting_service")
                    } else {
                        "backtesting_service"
                    };

                    let tls_config = ClientTlsConfig::new()
                        .ca_certificate(Certificate::from_pem(&ca_pem))
                        .identity(Identity::from_pem(&client_cert_pem, &client_key_pem))
                        .domain_name(hostname); // Use actual hostname for SNI

                    info!("TLS SNI hostname: {}", hostname);

                    endpoint = endpoint.tls_config(tls_config).map_err(|e| {
                        error!("Failed to apply TLS configuration: {:?}", e);
                        error!("  TLS config error: {}", e);
                        e
                    })?;
                    info!("TLS configuration with mTLS applied successfully");
                },
                (Some(ca_path), None, None) => {
                    info!("Configuring TLS with server verification only (no client cert)");
                    let ca_pem = tokio::fs::read_to_string(ca_path).await.map_err(|e| {
                        format!("Failed to read CA certificate at {}: {}", ca_path, e)
                    })?;

                    let tls_config = ClientTlsConfig::new()
                        .ca_certificate(Certificate::from_pem(&ca_pem))
                        .domain_name("foxhunt-services"); // Match server cert CN

                    endpoint = endpoint.tls_config(tls_config)?;
                    info!("TLS configuration with server verification applied successfully");
                },
                _ => {
                    warn!(
                        "HTTPS URL provided but certificate paths incomplete - connection may fail"
                    );
                    warn!(
                        "  CA: {:?}, Client cert: {:?}, Client key: {:?}",
                        ca_cert_path, client_cert_path, client_key_path
                    );
                },
            }
        } else {
            info!("Using HTTP (no TLS) for backtesting service connection");
        }

        let channel = endpoint.connect().await.map_err(|e| {
            error!("Failed to connect to backtesting service: {:?}", e);
            error!("  Error details: {}", e);
            e
        })?;

        info!("Successfully established channel connection to backtesting service");

        let client = BacktestingServiceClient::new(channel.clone());

        // Initialize health checker with circuit breaker
        let health_checker = Arc::new(HealthChecker::new(
            5_u32,                   // failure_threshold: 5 consecutive failures
            Duration::from_secs(10), // health_check_interval: 10 seconds
        ));

        info!("Successfully connected to backtesting service backend");

        Ok(Self {
            client,
            health_checker,
            backend_url: backend_url.to_string(),
            channel,
        })
    }

    /// Perform background health check (should be called periodically)
    pub async fn background_health_check(&self) {
        self.health_checker
            .perform_health_check(&self.channel)
            .await;
    }

    /// Check if backend is currently healthy
    pub async fn is_backend_healthy(&self) -> bool {
        self.health_checker.is_healthy().await
    }

    /// Forward request with health checking and latency tracking
    async fn forward_with_health_check<F, Fut, T>(
        &self,
        operation: &str,
        forward_fn: F,
    ) -> Result<T, Status>
    where
        F: FnOnce() -> Fut,
        Fut: std::future::Future<Output = Result<T, Status>>,
    {
        // Check health before forwarding
        if !self.health_checker.is_healthy().await {
            return Err(Status::unavailable(format!(
                "Backtesting service backend is unhealthy ({})",
                self.backend_url
            )));
        }

        // Track latency
        let start = Instant::now();

        // Forward request
        let result = forward_fn().await;

        let elapsed = start.elapsed();

        // Record health outcome
        match &result {
            Ok(_) => {
                self.health_checker.record_success().await;
                debug!(
                    operation = operation,
                    latency_us = elapsed.as_micros(),
                    "Forwarded request to backtesting service"
                );
            },
            Err(status) => {
                self.health_checker.record_failure().await;
                error!(
                    operation = operation,
                    error = %status,
                    latency_us = elapsed.as_micros(),
                    "Failed to forward request to backtesting service"
                );
            },
        }

        result
    }
}

#[tonic::async_trait]
impl BacktestingService for BacktestingServiceProxy {
    // Define streaming response type for subscribe_backtest_progress
    type SubscribeBacktestProgressStream = tonic::codec::Streaming<BacktestProgressEvent>;

    /// Forward start_backtest request to backend
    #[instrument(skip_all)]
    async fn start_backtest(
        &self,
        request: Request<StartBacktestRequest>,
    ) -> Result<Response<StartBacktestResponse>, Status> {
        self.forward_with_health_check("start_backtest", || async {
            // Clone the client (cheap operation, reuses connection)
            let mut client = self.client.clone();

            // Extract inner request and forward
            let inner_request = request.into_inner();

            // Forward to backend - zero additional allocations
            client.start_backtest(inner_request).await
        })
        .await
    }

    /// Forward get_backtest_status request to backend
    #[instrument(skip_all)]
    async fn get_backtest_status(
        &self,
        request: Request<GetBacktestStatusRequest>,
    ) -> Result<Response<GetBacktestStatusResponse>, Status> {
        self.forward_with_health_check("get_backtest_status", || async {
            let mut client = self.client.clone();
            let inner_request = request.into_inner();
            client.get_backtest_status(inner_request).await
        })
        .await
    }

    /// Forward get_backtest_results request to backend
    #[instrument(skip_all)]
    async fn get_backtest_results(
        &self,
        request: Request<GetBacktestResultsRequest>,
    ) -> Result<Response<GetBacktestResultsResponse>, Status> {
        self.forward_with_health_check("get_backtest_results", || async {
            let mut client = self.client.clone();
            let inner_request = request.into_inner();
            client.get_backtest_results(inner_request).await
        })
        .await
    }

    /// Forward list_backtests request to backend
    #[instrument(skip_all)]
    async fn list_backtests(
        &self,
        request: Request<ListBacktestsRequest>,
    ) -> Result<Response<ListBacktestsResponse>, Status> {
        self.forward_with_health_check("list_backtests", || async {
            let mut client = self.client.clone();
            let inner_request = request.into_inner();
            client.list_backtests(inner_request).await
        })
        .await
    }

    /// Forward subscribe_backtest_progress streaming request to backend
    #[instrument(skip_all)]
    async fn subscribe_backtest_progress(
        &self,
        request: Request<SubscribeBacktestProgressRequest>,
    ) -> Result<Response<Self::SubscribeBacktestProgressStream>, Status> {
        self.forward_with_health_check("subscribe_backtest_progress", || async {
            let mut client = self.client.clone();
            let inner_request = request.into_inner();

            // Forward streaming request - the response is already a stream
            let response = client.subscribe_backtest_progress(inner_request).await?;

            // Extract the stream and pass it through
            Ok(Response::new(response.into_inner()))
        })
        .await
    }

    /// Forward stop_backtest request to backend
    #[instrument(skip_all)]
    async fn stop_backtest(
        &self,
        request: Request<StopBacktestRequest>,
    ) -> Result<Response<StopBacktestResponse>, Status> {
        self.forward_with_health_check("stop_backtest", || async {
            let mut client = self.client.clone();
            let inner_request = request.into_inner();
            client.stop_backtest(inner_request).await
        })
        .await
    }
}

// Implement BacktestingService for Arc<BacktestingServiceProxy> to enable Arc wrapping
#[tonic::async_trait]
impl BacktestingService for Arc<BacktestingServiceProxy> {
    type SubscribeBacktestProgressStream = tonic::codec::Streaming<BacktestProgressEvent>;

    #[instrument(skip_all)]
    async fn start_backtest(
        &self,
        request: Request<StartBacktestRequest>,
    ) -> Result<Response<StartBacktestResponse>, Status> {
        self.as_ref().start_backtest(request).await
    }

    #[instrument(skip_all)]
    async fn get_backtest_status(
        &self,
        request: Request<GetBacktestStatusRequest>,
    ) -> Result<Response<GetBacktestStatusResponse>, Status> {
        self.as_ref().get_backtest_status(request).await
    }

    #[instrument(skip_all)]
    async fn get_backtest_results(
        &self,
        request: Request<GetBacktestResultsRequest>,
    ) -> Result<Response<GetBacktestResultsResponse>, Status> {
        self.as_ref().get_backtest_results(request).await
    }

    #[instrument(skip_all)]
    async fn list_backtests(
        &self,
        request: Request<ListBacktestsRequest>,
    ) -> Result<Response<ListBacktestsResponse>, Status> {
        self.as_ref().list_backtests(request).await
    }

    #[instrument(skip_all)]
    async fn subscribe_backtest_progress(
        &self,
        request: Request<SubscribeBacktestProgressRequest>,
    ) -> Result<Response<Self::SubscribeBacktestProgressStream>, Status> {
        self.as_ref().subscribe_backtest_progress(request).await
    }

    #[instrument(skip_all)]
    async fn stop_backtest(
        &self,
        request: Request<StopBacktestRequest>,
    ) -> Result<Response<StopBacktestResponse>, Status> {
        self.as_ref().stop_backtest(request).await
    }
}

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

    #[tokio::test]
    async fn test_health_checker_success() {
        let checker = HealthChecker::new(3, Duration::from_secs(10));

        assert_eq!(checker.get_state().await, HealthState::Healthy);
        assert!(checker.is_healthy().await);

        checker.record_success().await;
        assert_eq!(checker.get_state().await, HealthState::Healthy);
    }

    #[tokio::test]
    async fn test_health_checker_failure() {
        let checker = HealthChecker::new(3, Duration::from_secs(10));

        checker.record_failure().await;
        assert_eq!(checker.get_state().await, HealthState::Degraded);

        checker.record_failure().await;
        assert_eq!(checker.get_state().await, HealthState::Degraded);

        checker.record_failure().await;
        assert_eq!(checker.get_state().await, HealthState::Unhealthy);
        assert!(!checker.is_healthy().await);
    }

    #[tokio::test]
    async fn test_health_checker_recovery() {
        let checker = HealthChecker::new(3, Duration::from_secs(10));

        // Mark as unhealthy
        for _ in 0..3 {
            checker.record_failure().await;
        }
        assert_eq!(checker.get_state().await, HealthState::Unhealthy);

        // Recovery
        checker.record_success().await;
        assert_eq!(checker.get_state().await, HealthState::Healthy);
        assert!(checker.is_healthy().await);
    }
}