#![deny(clippy::unwrap_used, clippy::expect_used)] //! Foxhunt API Service //! //! Unified gRPC gateway with tonic-web for browser access and 6-layer authentication. //! Optimized for HFT requirements with <10us authentication overhead. use anyhow::Result; use clap::Parser; use std::sync::Arc; use std::time::Duration; use tracing::{error, info}; use common::metrics::GrpcMetricsLayer; use tonic_web::GrpcWebLayer; use tower_http::cors::{AllowHeaders, CorsLayer}; // Import all needed types from the library use api::auth::jwt::JwtConfig; use api::auth::{ AuditLogger, AuthInterceptor, AuthzService, JwtService, RateLimiter, RevocationService, }; #[derive(Parser, Debug)] #[command(name = "api", about = "Foxhunt API Service")] struct Args { /// gRPC server bind address #[arg(long, env = "GATEWAY_BIND_ADDR", default_value = "0.0.0.0:50051")] bind_addr: String, /// JWT secret (or use JWT_SECRET_FILE for production) #[arg(long, env = "JWT_SECRET")] jwt_secret: Option, /// JWT issuer #[arg(long, env = "JWT_ISSUER", default_value = "foxhunt-trading")] jwt_issuer: String, /// JWT audience #[arg(long, env = "JWT_AUDIENCE", default_value = "trading-api")] jwt_audience: String, /// Redis URL for JWT revocation #[arg(long, env = "REDIS_URL", default_value = "redis://localhost:6379")] redis_url: String, /// Rate limit (requests per second per user) #[arg(long, env = "RATE_LIMIT_RPS", default_value = "100")] rate_limit_rps: u32, /// Enable audit logging #[arg(long, env = "ENABLE_AUDIT_LOGGING", default_value = "true")] enable_audit_logging: bool, } #[tokio::main] async fn main() -> Result<()> { // Install ring as the default rustls CryptoProvider (required by kube-rs for in-cluster TLS) if let Err(_already) = rustls::crypto::ring::default_provider().install_default() { // Another provider already installed -- fine } let args = Args::parse(); // 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("api", Some(&otlp_endpoint)) { eprintln!("Failed to initialize observability: {}", e); } info!("Starting Foxhunt API Service"); info!("Bind address: {}", args.bind_addr); info!("JWT issuer: {}", args.jwt_issuer); info!("JWT audience: {}", args.jwt_audience); info!("Redis URL: {}", args.redis_url); info!("Rate limit: {} req/s per user", args.rate_limit_rps); info!("Audit logging: {}", args.enable_audit_logging); // Load JWT configuration (Vault-based with fallback) info!("Loading JWT configuration..."); let jwt_config = match load_jwt_config().await { Ok(config) => { info!("JWT configuration loaded successfully"); config }, Err(e) => { error!("Failed to load JWT configuration: {}", e); return Err(e); }, }; // Initialize authentication components info!("Initializing authentication services..."); let jwt_service = JwtService::new( jwt_config.jwt_secret.clone(), jwt_config.jwt_issuer.clone(), jwt_config.jwt_audience.clone(), ); info!("JWT service initialized with cached decoding key"); let revocation_service = RevocationService::new(&args.redis_url) .await .map_err(|e| anyhow::anyhow!("Failed to connect to Redis for revocation service: {}", e))?; info!("JWT revocation service connected to Redis"); let authz_service = AuthzService::new(); info!("Authorization service initialized with permission cache"); let rate_limiter = RateLimiter::new(args.rate_limit_rps) .map_err(|e| anyhow::anyhow!("Failed to create rate limiter: {}", e))?; info!("Rate limiter initialized ({} req/s)", args.rate_limit_rps); let audit_logger = AuditLogger::new(args.enable_audit_logging); info!("Audit logger initialized"); // Create authentication interceptor let auth_interceptor = AuthInterceptor::new( jwt_service, revocation_service, authz_service, rate_limiter, audit_logger, ); info!("6-layer authentication interceptor ready"); info!("API Service initialization complete"); info!("Ready to accept gRPC requests with <10us auth overhead"); // Initialize backend service proxies info!("Connecting to backend services..."); let trading_backend_url = std::env::var("TRADING_SERVICE_URL") .unwrap_or_else(|_| "http://localhost:50052".to_string()); let backtesting_backend_url = std::env::var("BACKTESTING_SERVICE_URL") .unwrap_or_else(|_| "http://localhost:50053".to_string()); let ml_training_backend_url = std::env::var("ML_TRAINING_SERVICE_URL") .unwrap_or_else(|_| "http://localhost:50054".to_string()); let broker_gw_url = std::env::var("BROKER_GATEWAY_SERVICE_URL") .unwrap_or_else(|_| "http://localhost:50056".to_string()); let data_acq_url = std::env::var("DATA_ACQUISITION_SERVICE_URL") .unwrap_or_else(|_| "http://localhost:50057".to_string()); let ml_service_url = std::env::var("ML_SERVICE_URL") .unwrap_or_else(|_| trading_backend_url.clone()); let trading_agent_url = std::env::var("TRADING_AGENT_SERVICE_URL") .unwrap_or_else(|_| "http://localhost:50055".to_string()); let prometheus_url = std::env::var("PROMETHEUS_URL") .unwrap_or_else(|_| "http://prometheus-stack-kube-prom-prometheus.foxhunt.svc:9090".to_string()); let monitoring_stream_interval: u32 = std::env::var("MONITORING_STREAM_INTERVAL") .ok() .and_then(|v| v.parse().ok()) .unwrap_or(3); // Load TLS certificate paths for backtesting service (mTLS) let backtesting_tls_ca_cert = std::env::var("BACKTESTING_TLS_CA_CERT").ok(); let backtesting_tls_client_cert = std::env::var("BACKTESTING_TLS_CLIENT_CERT").ok(); let backtesting_tls_client_key = std::env::var("BACKTESTING_TLS_CLIENT_KEY").ok(); // Load TLS certificate paths for ML training service (mTLS) let ml_training_tls_ca_cert = std::env::var("ML_TRAINING_TLS_CA_CERT").ok(); let ml_training_tls_client_cert = std::env::var("ML_TRAINING_TLS_CLIENT_CERT").ok(); let ml_training_tls_client_key = std::env::var("ML_TRAINING_TLS_CLIENT_KEY").ok(); // Initialize trading service proxy let mut trading_proxy = api::grpc::TradingServiceProxy::new_lazy(&trading_backend_url) .map_err(|e| anyhow::anyhow!("Failed to create trading service proxy: {}", e))?; info!( "Trading service proxy initialized ({})", trading_backend_url ); // Initialize backtesting service proxy (optional - graceful degradation) info!("Attempting to initialize Backtesting Service proxy..."); info!(" Backend URL: {}", backtesting_backend_url); info!(" CA cert: {:?}", backtesting_tls_ca_cert); info!(" Client cert: {:?}", backtesting_tls_client_cert); info!(" Client key: {:?}", backtesting_tls_client_key); let backtesting_proxy = match api::grpc::BacktestingServiceProxy::new( &backtesting_backend_url, backtesting_tls_ca_cert.as_deref(), backtesting_tls_client_cert.as_deref(), backtesting_tls_client_key.as_deref(), ) .await { Ok(proxy) => { info!( "Backtesting service proxy initialized ({})", backtesting_backend_url ); Some(Arc::new(proxy)) }, Err(e) => { error!("Backtesting service initialization failed!"); error!(" Error type: {:?}", e); error!(" Error message: {}", e); info!("Backtesting service unavailable: {}. API will run without backtesting endpoints.", e); info!(" Backtesting endpoints will return 503 Service Unavailable"); None }, }; // Spawn background health check task for backtesting service if let Some(proxy) = backtesting_proxy.as_ref() { let proxy_clone = Arc::clone(proxy); tokio::spawn(async move { let mut interval = tokio::time::interval(Duration::from_secs(10)); loop { interval.tick().await; proxy_clone.background_health_check().await; } }); info!("Backtesting service health check task started (10s interval)"); } // Initialize ML training service proxy (optional - graceful degradation) info!("Attempting to initialize ML Training Service proxy..."); info!(" Backend URL: {}", ml_training_backend_url); info!(" CA cert: {:?}", ml_training_tls_ca_cert); info!(" Client cert: {:?}", ml_training_tls_client_cert); info!(" Client key: {:?}", ml_training_tls_client_key); let ml_config = api::grpc::MlTrainingBackendConfig { address: ml_training_backend_url.clone(), connect_timeout_ms: 5000, request_timeout_ms: 30000, circuit_breaker_failures: 5, circuit_breaker_reset_secs: 30, tls_ca_cert_path: ml_training_tls_ca_cert.clone(), tls_client_cert_path: ml_training_tls_client_cert.clone(), tls_client_key_path: ml_training_tls_client_key.clone(), }; let ml_training_proxy = match api::grpc::setup_ml_training_proxy(ml_config).await { Ok(proxy) => { info!( "ML training service proxy initialized ({})", ml_training_backend_url ); Some(proxy) }, Err(e) => { error!("ML Training service initialization failed!"); error!(" Error type: {:?}", e); error!(" Error message: {}", e); info!( "ML Training service unavailable: {}. API will run without ML endpoints.", e ); info!(" ML training endpoints will return 503 Service Unavailable"); None }, }; // Initialize monitoring service handler (direct Prometheus scraping, no backend needed) let monitoring_backends = vec![ ("trading-service".to_string(), trading_backend_url.clone()), ("backtesting-service".to_string(), backtesting_backend_url.clone()), ("ml-training-service".to_string(), ml_training_backend_url.clone()), ("broker-gateway".to_string(), broker_gw_url.clone()), ("data-acquisition-service".to_string(), data_acq_url.clone()), ("trading-agent-service".to_string(), trading_agent_url.clone()), ]; let monitoring_handler = api::MonitoringServiceHandler::new( &prometheus_url, monitoring_stream_interval, ) .with_backends(monitoring_backends); info!( "Monitoring service handler initialized (Prometheus={}, interval={}s, backends=6)", prometheus_url, monitoring_stream_interval ); // Initialize direct proxy services (trading, risk, config, ML, broker_gw, data_acq, trading_agent) let trading_channel = tonic::transport::Channel::from_shared(trading_backend_url.clone()) .map_err(|e| anyhow::anyhow!("Invalid TRADING_SERVICE_URL: {e}"))? .connect_lazy(); let trading_direct_proxy = api::TradingDirectProxy::new( api::trading_backend::trading_service_client::TradingServiceClient::new( trading_channel.clone(), ), ); let risk_proxy = api::RiskServiceProxy::new( api::risk::risk_service_client::RiskServiceClient::new(trading_channel.clone()), ); let config_direct_proxy = api::ConfigServiceProxy::new( api::config_backend::config_service_client::ConfigServiceClient::new( trading_channel, ), ); let ml_channel = tonic::transport::Channel::from_shared(ml_service_url.clone()) .map_err(|e| anyhow::anyhow!("Invalid ML_SERVICE_URL: {e}"))? .connect_lazy(); let ml_proxy = api::MlServiceProxy::new( api::ml_inference::ml_service_client::MlServiceClient::new(ml_channel), ); let broker_channel = tonic::transport::Channel::from_shared(broker_gw_url.clone()) .map_err(|e| anyhow::anyhow!("Invalid BROKER_GATEWAY_SERVICE_URL: {e}"))? .connect_lazy(); let broker_proxy = api::BrokerGatewayProxy::new( api::broker_gateway::broker_gateway_service_client::BrokerGatewayServiceClient::new( broker_channel, ), ); let data_channel = tonic::transport::Channel::from_shared(data_acq_url.clone()) .map_err(|e| anyhow::anyhow!("Invalid DATA_ACQUISITION_SERVICE_URL: {e}"))? .connect_lazy(); let data_proxy = api::DataAcquisitionProxy::new( api::data_acquisition::data_acquisition_service_client::DataAcquisitionServiceClient::new( data_channel, ), ); let agent_channel = tonic::transport::Channel::from_shared(trading_agent_url.clone()) .map_err(|e| anyhow::anyhow!("Invalid TRADING_AGENT_SERVICE_URL: {e}"))? .connect_lazy(); let agent_proxy = api::TradingAgentProxy::new( api::trading_agent::trading_agent_service_client::TradingAgentServiceClient::new( agent_channel, ), ); info!("Direct proxy services initialized:"); info!(" - trading.TradingService -> {}", trading_backend_url); info!(" - risk.RiskService -> {}", trading_backend_url); info!(" - config.ConfigService -> {}", trading_backend_url); info!(" - ml.MLService -> {}", ml_service_url); info!(" - broker_gateway.BrokerGatewayService -> {}", broker_gw_url); info!(" - data_acquisition.DataAcquisitionService -> {}", data_acq_url); info!(" - trading_agent.TradingAgentService -> {}", trading_agent_url); // Setup backend health monitoring for SubscribeSystemStatus. // A background task checks gRPC health on each backend and pushes // updates through a watch channel consumed by the streaming RPC. let (health_tx, health_rx) = api::BackendHealthState::new(); trading_proxy.set_health_state(health_rx); { use tonic_health::pb::health_client::HealthClient; // Build persistent lazy channels (reused across checks) let backends: Vec<(&str, tonic::transport::Channel)> = vec![ ("trading-service", tonic::transport::Channel::from_shared(trading_backend_url.clone()) .map_err(|e| anyhow::anyhow!("invalid trading URL: {}", e))?.connect_lazy()), ("ml-training-service", tonic::transport::Channel::from_shared(ml_training_backend_url.clone()) .map_err(|e| anyhow::anyhow!("invalid ml-training URL: {}", e))?.connect_lazy()), ("backtesting-service", tonic::transport::Channel::from_shared(backtesting_backend_url.clone()) .map_err(|e| anyhow::anyhow!("invalid backtesting URL: {}", e))?.connect_lazy()), ("broker-gateway", tonic::transport::Channel::from_shared(broker_gw_url.clone()) .map_err(|e| anyhow::anyhow!("invalid broker-gateway URL: {}", e))?.connect_lazy()), ("data-acquisition-service", tonic::transport::Channel::from_shared(data_acq_url.clone()) .map_err(|e| anyhow::anyhow!("invalid data-acquisition URL: {}", e))?.connect_lazy()), ("trading-agent-service", tonic::transport::Channel::from_shared(trading_agent_url.clone()) .map_err(|e| anyhow::anyhow!("invalid trading-agent URL: {}", e))?.connect_lazy()), ]; let backends: Vec<(String, tonic::transport::Channel)> = backends .into_iter() .map(|(n, c)| (n.to_string(), c)) .collect(); tokio::spawn(async move { let mut interval = tokio::time::interval(Duration::from_secs(5)); loop { interval.tick().await; let mut entries = Vec::with_capacity(backends.len()); for (name, channel) in &backends { let mut hc = HealthClient::new(channel.clone()); let req = tonic_health::pb::HealthCheckRequest { service: String::new(), // overall server health }; let (healthy, msg) = match tokio::time::timeout( Duration::from_millis(2000), hc.check(req), ) .await { Ok(Ok(_)) => (true, "Serving"), // Unimplemented means gRPC server is UP but has no health service Ok(Err(s)) if s.code() == tonic::Code::Unimplemented => { (true, "Serving (no health check)") } Ok(Err(_)) => (false, "Error"), Err(_) => (false, "Unreachable"), }; entries.push(api::ServiceHealthEntry { name: name.clone(), healthy, message: msg.to_string(), }); } let _ = health_tx.send(entries); } }); info!("Backend health monitor started (5s interval)"); } // Initialize configuration manager (requires database) let database_url = std::env::var("DATABASE_URL") .map_err(|_| anyhow::anyhow!("DATABASE_URL environment variable must be set"))?; let db_pool = sqlx::PgPool::connect(&database_url) .await .map_err(|e| anyhow::anyhow!("Failed to connect to database: {}", e))?; info!("Database connection established"); // Create Redis connection for config manager let redis_client = redis::Client::open(args.redis_url.clone()) .map_err(|e| anyhow::anyhow!("Failed to create Redis client: {}", e))?; let redis_conn = redis::aio::ConnectionManager::new(redis_client) .await .map_err(|e| anyhow::anyhow!("Failed to create Redis connection manager: {}", e))?; let mut config_manager = api::ConfigurationManager::new(db_pool.clone(), redis_conn) .await .map_err(|e| anyhow::anyhow!("Failed to create configuration manager: {}", e))?; // Start NOTIFY listener for hot-reload config_manager .start_listening() .await .map_err(|e| anyhow::anyhow!("Failed to start configuration listener: {}", e))?; info!("Configuration manager initialized with hot-reload"); // Load TLS configuration if enabled (Wave H1 Security Enforcement) let tls_enabled = std::env::var("TLS_ENABLED") .unwrap_or_else(|_| "false".to_string()) .parse::() .unwrap_or(false); let tls_config = if tls_enabled { info!("TLS/mTLS enabled - initializing TLS 1.3 configuration"); let cert_path = std::env::var("TLS_CERT_PATH") .unwrap_or_else(|_| "./certs/server-cert.pem".to_string()); let key_path = std::env::var("TLS_KEY_PATH").unwrap_or_else(|_| "./certs/server-key.pem".to_string()); let ca_path = std::env::var("TLS_CA_PATH").unwrap_or_else(|_| "./certs/ca/ca-cert.pem".to_string()); let require_client_cert = std::env::var("TLS_REQUIRE_CLIENT_CERT") .unwrap_or_else(|_| "true".to_string()) .parse::() .unwrap_or(true); let enable_revocation = std::env::var("MTLS_ENABLE_REVOCATION_CHECK") .unwrap_or_else(|_| "false".to_string()) .parse::() .unwrap_or(false); let crl_url = std::env::var("MTLS_CRL_URL").ok(); let tls = api::auth::mtls::ApiGatewayTlsConfig::from_files( &cert_path, &key_path, &ca_path, require_client_cert, enable_revocation, crl_url, ) .await?; info!( "TLS configuration loaded - Protocol: TLS 1.3, mTLS: {}, Revocation: {}", require_client_cert, enable_revocation ); Some(tls) } else { info!("TLS disabled - running in development mode (set TLS_ENABLED=true for production)"); None }; // Build gRPC server with all services use api::auth_proto::auth_service_server::AuthServiceServer; use api::foxhunt::tli::{ backtesting_service_server::BacktestingServiceServer, trading_service_server::TradingServiceServer, }; use api::ml_training::ml_training_service_server::MlTrainingServiceServer; use api::monitoring::monitoring_service_server::MonitoringServiceServer; let addr: std::net::SocketAddr = args .bind_addr .parse() .map_err(|e| anyhow::anyhow!("Failed to parse bind address '{}': {}", args.bind_addr, e))?; info!("Starting gRPC server on {}", addr); // Setup graceful shutdown let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); // Spawn signal handler for graceful shutdown tokio::spawn(async move { match tokio::signal::ctrl_c().await { Ok(()) => { info!("Received shutdown signal, draining connections..."); } Err(e) => { error!("Failed to listen for ctrl-c signal: {}", e); } } let _ = shutdown_tx.send(()); }); // Setup health check service let (health_reporter, health_service) = tonic_health::server::health_reporter(); // Always register trading service (required) health_reporter .set_serving::>() .await; // Conditionally register optional services if backtesting_proxy.is_some() { health_reporter .set_serving::>() .await; } if ml_training_proxy.is_some() { health_reporter .set_serving::>() .await; } // Monitoring handler is always available (direct Prometheus, no backend dependency) health_reporter .set_serving::>() .await; // Initialize and start Prometheus metrics HTTP endpoint on port 9091 let gateway_metrics = api::metrics::GatewayMetrics::new() .map_err(|e| anyhow::anyhow!("Failed to initialize gateway metrics: {}", e))?; // Add service info metric (always present) use prometheus::{register_gauge_with_registry, Opts}; let service_info = register_gauge_with_registry!( Opts::new( "api_service_info", "API service information" ) .const_label("version", env!("CARGO_PKG_VERSION")) .const_label("service", "api"), gateway_metrics.registry().as_ref() ) .map_err(|e| anyhow::anyhow!("Failed to register service info: {}", e))?; service_info.set(1.0); // Add uptime gauge (matches pattern of other services) let uptime_gauge = register_gauge_with_registry!( Opts::new("api_uptime_seconds", "API service uptime in seconds"), gateway_metrics.registry().as_ref() ) .map_err(|e| anyhow::anyhow!("Failed to register uptime gauge: {}", e))?; let start_instant = std::time::Instant::now(); let uptime_clone = uptime_gauge.clone(); tokio::spawn(async move { let mut tick = tokio::time::interval(Duration::from_secs(5)); loop { tick.tick().await; uptime_clone.set(start_instant.elapsed().as_secs_f64()); } }); let metrics_registry = gateway_metrics.registry(); tokio::spawn(async move { // Use combined router for metrics AND health/resilience endpoints let combined_app = api::metrics::combined_router(metrics_registry); let metrics_addr = "0.0.0.0:9091"; info!( "Prometheus metrics endpoint listening on http://{}", metrics_addr ); info!("Health endpoints available:"); info!(" - GET http://{}/health/liveness", metrics_addr); info!(" - GET http://{}/health/readiness", metrics_addr); info!(" - GET http://{}/health/startup", metrics_addr); info!( " - GET http://{}/resilience/circuit-breaker/status", metrics_addr ); info!( " - GET http://{}/resilience/rate-limit/status", metrics_addr ); info!(" - GET http://{}/resilience/timeout/config", metrics_addr); info!(" - GET http://{}/resilience/retry/config", metrics_addr); let listener = match tokio::net::TcpListener::bind(metrics_addr).await { Ok(l) => l, Err(e) => { error!("Failed to bind metrics endpoint on {}: {}", metrics_addr, e); return; } }; if let Err(e) = axum::serve(listener, combined_app).await { error!("Metrics server failed: {}", e); } }); // Configure CORS for gRPC-Web browser access let cors_origins = std::env::var("CORS_ORIGINS") .unwrap_or_else(|_| "http://localhost:5173".to_string()); let origins: Vec = cors_origins .split(',') .filter_map(|s| s.trim().parse().ok()) .collect(); let cors = CorsLayer::new() .allow_origin(origins) .allow_headers(AllowHeaders::any()) .allow_methods([http::Method::POST, http::Method::OPTIONS]) .expose_headers([ http::HeaderName::from_static("grpc-status"), http::HeaderName::from_static("grpc-message"), ]); // Build server with HTTP/2 optimizations and gRPC-Web support let mut server_builder = if let Some(ref tls) = tls_config { tonic::transport::Server::builder() .accept_http1(true) // Required for grpc-web .layer(GrpcMetricsLayer::new("api")) .layer(cors.clone()) .layer(GrpcWebLayer::new()) .tls_config(tls.to_server_tls_config())? .max_concurrent_streams(Some(10_000)) .http2_keepalive_interval(Some(Duration::from_secs(30))) .http2_keepalive_timeout(Some(Duration::from_secs(10))) } else { tonic::transport::Server::builder() .accept_http1(true) // Required for grpc-web .layer(GrpcMetricsLayer::new("api")) .layer(cors.clone()) .layer(GrpcWebLayer::new()) .max_concurrent_streams(Some(10_000)) .http2_keepalive_interval(Some(Duration::from_secs(30))) .http2_keepalive_timeout(Some(Duration::from_secs(10))) } .layer(tower::ServiceBuilder::new().layer(tower::layer::util::Identity::new())); // Placeholder for auth interceptor layer // Add gRPC reflection service (unauthenticated -- for grpcurl/debugging) let reflection_service = tonic_reflection::server::Builder::configure() .register_encoded_file_descriptor_set(api::TLI_FILE_DESCRIPTOR_SET) .register_encoded_file_descriptor_set(api::MONITORING_FILE_DESCRIPTOR_SET) .register_encoded_file_descriptor_set(api::ML_TRAINING_FILE_DESCRIPTOR_SET) .register_encoded_file_descriptor_set(api::TRADING_AGENT_FILE_DESCRIPTOR_SET) .register_encoded_file_descriptor_set(api::TRADING_FILE_DESCRIPTOR_SET) .register_encoded_file_descriptor_set(api::RISK_FILE_DESCRIPTOR_SET) .register_encoded_file_descriptor_set(api::CONFIG_FILE_DESCRIPTOR_SET) .register_encoded_file_descriptor_set(api::BROKER_GATEWAY_FILE_DESCRIPTOR_SET) .register_encoded_file_descriptor_set(api::DATA_ACQUISITION_FILE_DESCRIPTOR_SET) .register_encoded_file_descriptor_set(api::ML_FILE_DESCRIPTOR_SET) .register_encoded_file_descriptor_set(api::AUTH_FILE_DESCRIPTOR_SET) .build_v1() .map_err(|e| anyhow::anyhow!("Failed to build reflection service: {}", e))?; // Create unauthenticated auth service (Login / RefreshToken -- pre-authentication) let auth_grpc_service = api::auth::grpc_login::AuthGrpcService::new(Arc::new(jwt_config)); // Add health + reflection services, then unauthenticated auth service let mut router = server_builder .add_service(health_service) .add_service(reflection_service) .add_service(AuthServiceServer::new(auth_grpc_service)); // Always add trading service (required) with authentication router = router.add_service(TradingServiceServer::with_interceptor( trading_proxy, auth_interceptor.clone(), )); // Track service availability for logging let backtesting_available = backtesting_proxy.is_some(); let ml_training_available = ml_training_proxy.is_some(); // Conditionally add optional services with authentication if let Some(backtesting) = backtesting_proxy.as_ref() { // Clone the entire Arc - tonic services can work with Arc-wrapped implementations router = router.add_service(BacktestingServiceServer::with_interceptor( Arc::clone(backtesting), auth_interceptor.clone(), )); } if let Some(ml_training) = ml_training_proxy { router = router.add_service(MlTrainingServiceServer::with_interceptor( ml_training, auth_interceptor.clone(), )); } // Monitoring handler is always available (direct Prometheus scraping, no backend) router = router.add_service(MonitoringServiceServer::with_interceptor( monitoring_handler, auth_interceptor.clone(), )); // Register all direct proxy services with auth interceptor { use api::trading_backend::trading_service_server::TradingServiceServer as TradingDirectServer; use api::risk::risk_service_server::RiskServiceServer; use api::broker_gateway::broker_gateway_service_server::BrokerGatewayServiceServer; use api::data_acquisition::data_acquisition_service_server::DataAcquisitionServiceServer; use api::ml_inference::ml_service_server::MlServiceServer; use api::config_backend::config_service_server::ConfigServiceServer as ConfigBackendServer; use api::trading_agent::trading_agent_service_server::TradingAgentServiceServer; router = router .add_service(TradingDirectServer::with_interceptor( trading_direct_proxy, auth_interceptor.clone(), )) .add_service(RiskServiceServer::with_interceptor( risk_proxy, auth_interceptor.clone(), )) .add_service(BrokerGatewayServiceServer::with_interceptor( broker_proxy, auth_interceptor.clone(), )) .add_service(DataAcquisitionServiceServer::with_interceptor( data_proxy, auth_interceptor.clone(), )) .add_service(MlServiceServer::with_interceptor( ml_proxy, auth_interceptor.clone(), )) .add_service(ConfigBackendServer::with_interceptor( config_direct_proxy, auth_interceptor.clone(), )) .add_service(TradingAgentServiceServer::with_interceptor( agent_proxy, auth_interceptor.clone(), )); } // Log startup information info!("API Service listening on {}", addr); info!(" - gRPC-Web: enabled (CORS origins: {})", cors_origins); info!(" - Trading Service: {} (REQUIRED)", trading_backend_url); info!( " - Backtesting Service: {} ({})", backtesting_backend_url, if backtesting_available { "AVAILABLE" } else { "UNAVAILABLE" } ); info!( " - ML Training Service: {} ({})", ml_training_backend_url, if ml_training_available { "AVAILABLE" } else { "UNAVAILABLE" } ); info!( " - Monitoring: direct Prometheus scraping ({})", prometheus_url ); info!(" - trading.TradingService -> {}", trading_backend_url); info!(" - risk.RiskService -> {}", trading_backend_url); info!(" - config.ConfigService -> {}", trading_backend_url); info!(" - ml.MLService -> {}", ml_service_url); info!(" - broker_gateway.BrokerGatewayService -> {}", broker_gw_url); info!(" - data_acquisition.DataAcquisitionService -> {}", data_acq_url); info!(" - trading_agent.TradingAgentService -> {}", trading_agent_url); info!(" - Health checks: enabled"); info!(" - Authentication: 6-layer (<10us overhead)"); info!(" - Rate limiting: {} req/s per user", args.rate_limit_rps); // Start server with graceful shutdown let server = router.serve_with_shutdown(addr, async { shutdown_rx.await.ok(); }); // Start server server.await?; info!("API Service shutdown complete"); Ok(()) } /// Load JWT configuration from Vault (production) or environment (development) /// /// Priority: /// 1. Vault (secret/foxhunt/jwt) - Production /// 2. JWT_SECRET_FILE - File-based secret /// 3. JWT_SECRET env var - Development fallback async fn load_jwt_config() -> Result { JwtConfig::new().await }