foxhunt/services/backtesting_service/src/main.rs

#![warn(missing_docs)]
#![deny(clippy::unwrap_used)]
#![deny(clippy::expect_used)]

//! # Foxhunt Backtesting Service
//!
//! Standalone backtesting service for the Foxhunt HFT trading system.
//! This service provides comprehensive strategy testing and performance analysis capabilities.

use anyhow::{Context, Result};
use rustls::crypto::CryptoProvider;
use std::net::SocketAddr;
use tonic::transport::Server;
use tracing::{error, info, warn};

mod dbn_data_source;
mod dbn_repository;
mod health;
mod performance;
mod repositories;
mod repository_impl;
mod service;
mod storage;
mod strategy_engine;
mod tls_config;

// Import the generated gRPC code
mod foxhunt {
    pub mod tli {
        tonic::include_proto!("foxhunt.tli");
    }
}

use config::schemas::S3Config;
use config::structures::BacktestingDatabaseConfig;
use model_loader::backtesting_cache::{BacktestCacheConfig, BacktestingModelCache};
use repository_impl::create_repositories;
use service::BacktestingServiceImpl;
use std::sync::Arc;
use storage::StorageManager;
use tls_config::BacktestingServiceTlsConfig;

// Import ObjectStoreBackend from external storage crate (not local storage module)
use ::storage::ObjectStoreBackend;

/// Main entry point for the backtesting service
#[tokio::main]
async fn main() -> Result<()> {
    // Wave 77 Agent 3: Initialize crypto provider FIRST before any TLS operations
    // This fixes the "Could not automatically determine the process-level CryptoProvider" panic
    CryptoProvider::install_default(rustls::crypto::ring::default_provider())
        .map_err(|_| anyhow::anyhow!("Failed to install default crypto provider"))?;

    // Initialize observability (JSON logging + OpenTelemetry tracing via OTLP)
    let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT")
        .unwrap_or_else(|_| "http://localhost:4317".to_string());
    if let Err(e) = common::observability::init_observability(
        "backtesting_service",
        &otlp_endpoint,
    ).await {
        eprintln!("Failed to initialize observability: {}", e);
    }

    info!("Starting Foxhunt Backtesting Service");

    // Get database configuration from environment
    let database_url =
        std::env::var("DATABASE_URL").context("DATABASE_URL environment variable is required")?;

    // Wave 67 Agent 2: Optimized database configuration for backtesting workloads
    // Based on Wave 66 Agent 8 analysis - increased statement cache for better performance
    let database_config = BacktestingDatabaseConfig {
        database_url,
        max_connections: Some(10),
        min_connections: Some(2),
        acquire_timeout_ms: Some(5000),
        statement_cache_capacity: Some(500), // Increased from 100 to 500 for better cache hit rate
        enable_logging: Some(false),
    };

    // Configuration loaded from environment
    info!("Configuration loaded from environment variables");

    info!("Backtesting configuration loaded successfully");

    // Initialize storage manager
    let storage_manager = Arc::new(
        StorageManager::new(&database_config)
            .await
            .context("Failed to initialize storage manager")?,
    );

    // Initialize S3 storage for model loading
    let s3_config = S3Config {
        bucket_name: std::env::var("MODEL_S3_BUCKET")
            .unwrap_or_else(|_| "foxhunt-models".to_string()),
        region: std::env::var("AWS_REGION").unwrap_or_else(|_| "us-east-1".to_string()),
        access_key_id: std::env::var("AWS_ACCESS_KEY_ID").ok(),
        secret_access_key: std::env::var("AWS_SECRET_ACCESS_KEY").ok(),
        session_token: std::env::var("AWS_SESSION_TOKEN").ok(),
        endpoint_url: std::env::var("AWS_ENDPOINT_URL").ok(),
        force_path_style: std::env::var("AWS_FORCE_PATH_STYLE")
            .map(|v| v == "true")
            .unwrap_or(false),
        max_retry_attempts: 3,
        timeout: std::time::Duration::from_secs(30),
        use_ssl: true,
    };

    // Create S3 storage backend for models
    let s3_storage = ObjectStoreBackend::new(s3_config, None)
        .await
        .context("Failed to create S3 storage backend")?;

    // Initialize model cache for backtesting with S3 storage
    let cache_config = BacktestCacheConfig {
        cache_dir: std::env::var("MODEL_CACHE_DIR")
            .unwrap_or_else(|_| "/tmp/foxhunt/model_cache".to_string())
            .into(),
        ..Default::default()
    };

    let mut model_cache = BacktestingModelCache::new(s3_storage, cache_config)
        .await
        .context("Failed to create BacktestingModelCache for backtesting")?;

    // Initialize model cache - this will scan for existing models
    model_cache
        .initialize()
        .await
        .context("Failed to initialize ModelCache")?;

    let model_cache = Arc::new(model_cache);
    info!("Backtesting model cache initialized with S3 storage and historical version support");

    // Create repositories with dependency injection
    let repositories = Arc::new(
        create_repositories(storage_manager)
            .await
            .context("Failed to create repositories")?,
    );

    // Initialize the service with repository injection and model cache
    let service = BacktestingServiceImpl::new(repositories, Some(Arc::clone(&model_cache)))
        .await
        .context("Failed to initialize backtesting service")?;

    // Initialize TLS configuration for mTLS
    // Read TLS configuration from environment variables (Docker deployment)
    let tls_enabled = std::env::var("TLS_ENABLED")
        .ok()
        .and_then(|v| v.parse::<bool>().ok())
        .unwrap_or(false);

    info!("TLS Configuration:");
    info!("  TLS Enabled: {}", tls_enabled);

    let tls_config = if tls_enabled {
        let cert_path = std::env::var("TLS_CERT_PATH")
            .context("TLS_CERT_PATH environment variable required when TLS is enabled")?;
        let key_path = std::env::var("TLS_KEY_PATH")
            .context("TLS_KEY_PATH environment variable required when TLS is enabled")?;
        let ca_cert_path = std::env::var("TLS_CA_PATH")
            .context("TLS_CA_PATH environment variable required for mTLS")?;
        let require_client_cert = std::env::var("TLS_REQUIRE_CLIENT_CERT")
            .ok()
            .and_then(|v| v.parse::<bool>().ok())
            .unwrap_or(true); // Default to requiring client certs for security

        info!("  Certificate Path: {}", cert_path);
        info!("  Key Path: {}", key_path);
        info!("  CA Cert Path: {}", ca_cert_path);
        info!("  Require Client Cert: {}", require_client_cert);

        BacktestingServiceTlsConfig::from_files(
            &cert_path,
            &key_path,
            &ca_cert_path,
            require_client_cert,
        )
        .await
        .context("Failed to initialize TLS configuration from files")?
    } else {
        info!("  TLS is disabled - using default configuration");
        info!("  ⚠️  WARNING: Running without TLS in production is NOT recommended");

        // Create a default config that won't be used (server won't apply TLS)
        BacktestingServiceTlsConfig::from_files(
            "/tmp/foxhunt/certs/server-cert.pem",
            "/tmp/foxhunt/certs/server-key.pem",
            "/tmp/foxhunt/certs/ca/ca-cert.pem",
            false,
        )
        .await
        .unwrap_or_else(|e| {
            warn!(
                "Failed to load default TLS config (expected when TLS disabled): {}",
                e
            );
            // Return a minimal config that won't be used
            BacktestingServiceTlsConfig {
                server_identity: tonic::transport::Identity::from_pem(vec![0], vec![0]),
                ca_certificate: tonic::transport::Certificate::from_pem(vec![0]),
                require_client_cert: false,
                protocol_version: tls_config::TlsProtocolVersion::Tls13,
                enable_revocation_check: false,
                crl_url: None,
            }
        })
    };

    // Setup gRPC server
    let grpc_port = std::env::var("GRPC_PORT")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(50053);
    let addr: SocketAddr = format!("0.0.0.0:{}", grpc_port)
        .parse()
        .context("Invalid server address in configuration")?;

    info!("Starting gRPC server on {}", addr);

    // Wave 67 Agent 3: HTTP/2 streaming performance optimizations
    // Apply same optimizations as other services for consistency
    let enable_http2_opts = std::env::var("ENABLE_HTTP2_OPTIMIZATIONS")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(true);

    if enable_http2_opts {
        info!("✅ HTTP/2 optimizations enabled:");
        info!("   - tcp_nodelay: true (-40ms Nagle delay)");
        info!("   - Stream window: 1MB");
        info!("   - Connection window: 10MB");
        info!("   - Adaptive window: true");
        info!("   - Max streams: 10,000 (production scale)");
    } else {
        info!("⚠️  HTTP/2 optimizations disabled via feature flag");
    }

    // Start the server with HTTP/2 optimizations
    let mut server_builder = Server::builder();

    if enable_http2_opts {
        server_builder = server_builder
            .tcp_nodelay(true)  // Critical: eliminates 40ms Nagle delay
            .http2_keepalive_interval(Some(std::time::Duration::from_secs(30)))
            .http2_keepalive_timeout(Some(std::time::Duration::from_secs(10)))
            .initial_stream_window_size(Some(1024 * 1024))  // 1MB
            .initial_connection_window_size(Some(10 * 1024 * 1024))  // 10MB
            .http2_adaptive_window(Some(true))
            .max_concurrent_streams(Some(10_000)); // Increased from 1,024 to 10,000 for production scale
    }

    // Initialize Prometheus metrics
    backtesting_service::simple_metrics::init_metrics();
    let service_start = std::time::Instant::now();

    // Spawn uptime updater
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(std::time::Duration::from_secs(1));
        loop {
            interval.tick().await;
            backtesting_service::simple_metrics::update_uptime(service_start);
        }
    });

    // Start Prometheus metrics HTTP endpoint on port 9093
    tokio::spawn(async {
        use axum::{routing::get, Router};
        use prometheus::{Encoder, TextEncoder};

        async fn metrics_handler() -> Result<String, String> {
            let encoder = TextEncoder::new();
            let metric_families = prometheus::gather();
            let mut buffer = vec![];
            encoder
                .encode(&metric_families, &mut buffer)
                .map_err(|e| format!("Failed to encode metrics: {}", e))?;
            String::from_utf8(buffer)
                .map_err(|e| format!("Failed to convert metrics to UTF-8: {}", e))
        }

        async fn metrics_handler_wrapper() -> String {
            metrics_handler()
                .await
                .unwrap_or_else(|e| format!("Error: {}", e))
        }

        let metrics_app = Router::new().route("/metrics", get(metrics_handler_wrapper));

        let metrics_addr = "0.0.0.0:9093";
        info!(
            "Prometheus metrics endpoint listening on http://{}",
            metrics_addr
        );

        let listener = match tokio::net::TcpListener::bind(metrics_addr).await {
            Ok(l) => l,
            Err(e) => {
                error!("Failed to bind metrics endpoint {}: {}", metrics_addr, e);
                return;
            },
        };

        if let Err(e) = axum::serve(listener, metrics_app).await {
            error!("Metrics server failed: {}", e);
        }
    });

    // Start HTTP health check server on port 8082 (no TLS required)
    let health_port = std::env::var("HEALTH_PORT")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(8082);
    let health_addr: SocketAddr = format!("0.0.0.0:{}", health_port)
        .parse()
        .context("Invalid health server address")?;

    info!("Starting health check server on {}", health_addr);

    let health_app = health::health_router();

    tokio::spawn(async move {
        let health_listener = match tokio::net::TcpListener::bind(health_addr).await {
            Ok(l) => l,
            Err(e) => {
                error!("Failed to bind health server {}: {}", health_addr, e);
                return;
            },
        };

        if let Err(e) = axum::serve(health_listener, health_app).await {
            error!("Health server failed: {}", e);
        }
    });

    // Setup gRPC health reporting service
    let (health_reporter, health_service) = tonic_health::server::health_reporter();

    // Mark backtesting service as SERVING
    health_reporter
        .set_serving::<foxhunt::tli::backtesting_service_server::BacktestingServiceServer<BacktestingServiceImpl>>()
        .await;

    info!("gRPC health service configured - backtesting service marked as SERVING");

    // Build server with optional TLS and HTTP/2 optimizations
    let router = if tls_enabled {
        info!("✅ TLS enabled - configuring mTLS for gRPC server");
        server_builder
            .tls_config(tls_config.to_server_tls_config())?
            .add_service(health_service)  // Add gRPC health service first
            .add_service(
                foxhunt::tli::backtesting_service_server::BacktestingServiceServer::new(service),
            )
    } else {
        info!("⚠️  TLS disabled - running gRPC server without encryption");
        server_builder
            .add_service(health_service)  // Add gRPC health service first
            .add_service(
                foxhunt::tli::backtesting_service_server::BacktestingServiceServer::new(service),
            )
    };

    info!(
        "🚀 Backtesting Service ready - starting gRPC server on {}",
        addr
    );
    router
        .serve_with_shutdown(addr, shutdown_signal())
        .await
        .context("gRPC server failed")?;

    info!("Backtesting Service stopped gracefully");
    Ok(())
}

/// Waits for a CTRL+C signal for graceful shutdown
async fn shutdown_signal() {
    if let Err(e) = tokio::signal::ctrl_c().await {
        tracing::error!("Failed to install CTRL+C handler: {}", e);
    }
    tracing::info!("Shutdown signal received, stopping gracefully...");
}