foxhunt/services/trading_agent_service/src/main.rs

//! Trading Agent Service - Main Entry Point
//!
//! Portfolio management service with universe selection, asset selection,
//! portfolio allocation, and order generation capabilities.

#![deny(clippy::unwrap_used, clippy::expect_used)]

use anyhow::{Context, Result};
use std::sync::Arc;
use tokio::signal;
use tokio::sync::Mutex;
use tonic::transport::{Certificate, Identity, Server, ServerTlsConfig};
use tracing::{error, info};

use common::metrics::GrpcMetricsLayer;

use common::DatabasePool;
use config::manager::ConfigManager;
use config::DatabaseConfig;

use trading_agent_service::proto::trading_agent::trading_agent_service_server::TradingAgentServiceServer;
use trading_agent_service::service::TradingAgentServiceImpl;

/// Default configuration values
const DEFAULT_GRPC_PORT: u16 = 50055;
const DEFAULT_HEALTH_PORT: u16 = 8083;
const DEFAULT_METRICS_PORT: u16 = 9095;

#[tokio::main]
async fn main() -> Result<()> {
    // 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("trading_agent", Some(&otlp_endpoint)) {
        eprintln!("Failed to initialize observability: {}", e);
    }

    info!("Starting Trading Agent Service...");

    // Create service config
    let service_config = config::ServiceConfig {
        name: "trading_agent_service".to_string(),
        environment: std::env::var("ENVIRONMENT").unwrap_or_else(|_| "production".to_string()),
        version: env!("CARGO_PKG_VERSION").to_string(),
        settings: serde_json::json!({}),
    };
    let _config_manager = Arc::new(ConfigManager::new(service_config));

    info!("ConfigManager initialized");

    // Initialize database connection
    let mut database_config = DatabaseConfig::new();
    database_config.max_connections = 20;
    database_config.min_connections = 5;

    let db_pool_wrapper = DatabasePool::new(database_config.into())
        .await
        .context("Failed to create database pool")?;

    let db_pool = db_pool_wrapper.pool().clone();

    info!("Database connection pool initialized");

    // Initialize regime orchestrator for Wave D adaptive strategies
    let regime_orchestrator = ml::regime::orchestrator::RegimeOrchestrator::new(db_pool.clone())
        .await
        .context("Failed to create RegimeOrchestrator")?;
    let regime_orchestrator = Arc::new(Mutex::new(regime_orchestrator));

    info!("RegimeOrchestrator initialized");

    // Initialize unified service
    let trading_agent_service = TradingAgentServiceImpl::new(db_pool.clone(), regime_orchestrator);

    info!("Trading Agent Service initialized");

    // Create health service
    let (health_reporter, health_service) = tonic_health::server::health_reporter();
    health_reporter
        .set_serving::<TradingAgentServiceServer<TradingAgentServiceImpl>>()
        .await;

    // Load TLS configuration
    let tls_config = load_tls_config()
        .await
        .context("Failed to load TLS configuration")?;

    // Build gRPC server
    let grpc_port = std::env::var("GRPC_PORT")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(DEFAULT_GRPC_PORT);
    let addr = format!("0.0.0.0:{}", grpc_port).parse()?;

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

    // Build server with optional TLS
    let mut server_builder = Server::builder()
        .layer(GrpcMetricsLayer::new("trading-agent-service"));

    if let Some(tls) = tls_config {
        info!("🔒 TLS enabled for Trading Agent Service");
        server_builder = Server::builder()
            .layer(GrpcMetricsLayer::new("trading-agent-service"))
            .tls_config(tls)
            .context("Failed to apply TLS configuration")?;
    } else {
        info!("⚠️  TLS disabled - running in insecure mode");
    }

    let server = server_builder
        .add_service(health_service)
        .add_service(TradingAgentServiceServer::new(trading_agent_service))
        .serve_with_shutdown(addr, shutdown_signal());

    info!("Trading Agent Service listening on {}", addr);

    // Start background tasks
    tokio::select! {
        result = server => {
            if let Err(e) = result {
                error!("gRPC server error: {}", e);
            }
        }
        _ = start_health_endpoint(DEFAULT_HEALTH_PORT) => {
            error!("Health endpoint stopped");
        }
        _ = start_metrics_endpoint(DEFAULT_METRICS_PORT) => {
            error!("Metrics endpoint stopped");
        }
    }

    info!("Trading Agent Service shutdown complete");
    Ok(())
}

/// Start health check endpoint
async fn start_health_endpoint(port: u16) -> Result<()> {
    use hyper::server::conn::http1;
    use hyper::service::service_fn;
    use hyper_util::rt::TokioIo;
    use tokio::net::TcpListener;

    let addr: std::net::SocketAddr = ([0, 0, 0, 0], port).into();
    let listener = TcpListener::bind(addr)
        .await
        .context("Failed to bind health endpoint")?;

    info!("Health endpoint listening on http://{}", addr);

    loop {
        let (stream, _) = match listener.accept().await {
            Ok(conn) => conn,
            Err(e) => {
                error!("Failed to accept connection: {}", e);
                continue;
            },
        };

        tokio::spawn(async move {
            let io = TokioIo::new(stream);
            if let Err(e) = http1::Builder::new()
                .serve_connection(io, service_fn(health_handler))
                .await
            {
                error!("Health server error: {}", e);
            }
        });
    }
}

/// Health check handler
async fn health_handler(
    _: hyper::Request<hyper::body::Incoming>,
) -> Result<hyper::Response<http_body_util::Full<bytes::Bytes>>, std::convert::Infallible> {
    use bytes::Bytes;
    use http_body_util::Full;

    let health_response = serde_json::json!({
        "status": "healthy",
        "service": "trading_agent_service",
        "timestamp": chrono::Utc::now().to_rfc3339(),
        "version": env!("CARGO_PKG_VERSION"),
    });

    // Builder with hardcoded status 200 and valid header cannot fail
    let response = hyper::Response::builder()
        .status(200)
        .header("content-type", "application/json")
        .body(Full::new(Bytes::from(health_response.to_string())))
        .unwrap_or_else(|_| {
            hyper::Response::new(Full::new(Bytes::from(r#"{"status":"healthy"}"#)))
        });

    Ok(response)
}

/// Start Prometheus metrics endpoint
async fn start_metrics_endpoint(port: u16) -> Result<()> {
    use axum::{routing::get, Router};
    use prometheus::{Encoder, TextEncoder};

    async fn metrics_handler() -> String {
        let encoder = TextEncoder::new();
        let metric_families = prometheus::gather();
        let mut buffer = vec![];
        // encode() only fails on IO errors writing to Vec, which cannot fail
        let _ = encoder.encode(&metric_families, &mut buffer);
        String::from_utf8(buffer).unwrap_or_default()
    }

    let app = Router::new().route("/metrics", get(metrics_handler));

    let addr = format!("0.0.0.0:{}", port);
    let listener = tokio::net::TcpListener::bind(&addr).await?;

    info!("Metrics endpoint listening on http://{}", addr);

    axum::serve(listener, app)
        .await
        .context("Metrics server failed")?;

    Ok(())
}

/// Load TLS configuration for the Trading Agent Service
async fn load_tls_config() -> Result<Option<ServerTlsConfig>> {
    // Check if TLS is enabled via environment variable
    let tls_enabled = std::env::var("TLS_ENABLED")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(false);

    if !tls_enabled {
        info!("TLS disabled via TLS_ENABLED=false");
        return Ok(None);
    }

    info!("Loading TLS configuration for Trading Agent Service...");

    // Get certificate paths from environment with sensible defaults
    let cert_path = std::env::var("TLS_CERT_PATH")
        .unwrap_or_else(|_| "/tmp/foxhunt/certs/server.crt".to_string());
    let key_path = std::env::var("TLS_KEY_PATH")
        .unwrap_or_else(|_| "/tmp/foxhunt/certs/server.key".to_string());
    let ca_cert_path =
        std::env::var("TLS_CA_PATH").unwrap_or_else(|_| "/tmp/foxhunt/certs/ca.crt".to_string());

    info!("TLS certificate paths:");
    info!("  Server cert: {}", cert_path);
    info!("  Server key: {}", key_path);
    info!("  CA cert: {}", ca_cert_path);

    // Read server certificate and key
    let cert_pem = tokio::fs::read_to_string(&cert_path)
        .await
        .with_context(|| format!("Failed to read server certificate: {}", cert_path))?;

    let key_pem = tokio::fs::read_to_string(&key_path)
        .await
        .with_context(|| format!("Failed to read server private key: {}", key_path))?;

    // Create server identity
    let server_identity = Identity::from_pem(cert_pem, key_pem);

    // Check if mTLS (mutual TLS) is enabled
    let mtls_enabled = std::env::var("MTLS_ENABLED")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(false);

    let mut tls_config = ServerTlsConfig::new().identity(server_identity);

    if mtls_enabled {
        info!("mTLS enabled - requiring client certificates");

        // Read CA certificate for client verification
        let ca_pem = tokio::fs::read_to_string(&ca_cert_path)
            .await
            .with_context(|| format!("Failed to read CA certificate: {}", ca_cert_path))?;

        let ca_certificate = Certificate::from_pem(ca_pem);
        tls_config = tls_config.client_ca_root(ca_certificate);

        info!("✅ mTLS configured - client certificates will be validated");
    } else {
        info!("mTLS disabled - server-side TLS only");
    }

    info!("✅ TLS configuration loaded successfully");
    Ok(Some(tls_config))
}

/// Handle shutdown signals
async fn shutdown_signal() {
    let ctrl_c = async {
        if let Err(e) = signal::ctrl_c().await {
            error!("Failed to install Ctrl+C handler: {}", e);
        }
    };

    #[cfg(unix)]
    let terminate = async {
        match signal::unix::signal(signal::unix::SignalKind::terminate()) {
            Ok(mut signal_stream) => {
                signal_stream.recv().await;
            },
            Err(e) => {
                error!("Failed to install SIGTERM handler: {}", e);
            },
        }
    };

    #[cfg(not(unix))]
    let terminate = std::future::pending::<()>();

    tokio::select! {
        _ = ctrl_c => {}
        _ = terminate => {}
    }

    info!("Shutdown signal received");
}