foxhunt/services/data_acquisition_service/src/main.rs

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

//! Data Acquisition Service - Main entry point
//!
//! Starts the gRPC server for automated Databento data downloads

use clap::Parser;
use data_acquisition_service::proto::data_acquisition_service_server::DataAcquisitionServiceServer;
use data_acquisition_service::service::DataAcquisitionServiceImpl;
use std::net::SocketAddr;
use tonic::transport::Server;
use tracing::info;

use common::metrics::GrpcMetricsLayer;

#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None)]
struct Args {
    /// gRPC server port (50057 to avoid collision with trading_agent_service on 50055)
    #[arg(long, default_value = "50057", env = "DATA_ACQUISITION_PORT")]
    port: u16,

    /// Health check endpoint port
    #[arg(long, default_value = "8095", env = "DATA_ACQUISITION_HEALTH_PORT")]
    health_port: u16,

    /// Enable debug logging (set RUST_LOG env for fine-grained control)
    #[arg(long, default_value = "false")]
    debug: bool,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Parse command-line arguments
    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("data_acquisition", Some(&otlp_endpoint)) {
        eprintln!("Failed to initialize observability: {}", e);
    }

    info!("Starting Data Acquisition Service...");
    info!("gRPC port: {}", args.port);
    info!("Health port: {}", args.health_port);

    // Initialize Prometheus metrics
    data_acquisition_service::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;
            data_acquisition_service::metrics::update_uptime(service_start);
        }
    });

    // Start Prometheus metrics HTTP endpoint on port 9097
    let metrics_port: u16 = std::env::var("METRICS_PORT")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(9097);
    tokio::spawn(async move {
        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![];
            let _ = encoder.encode(&metric_families, &mut buffer);
            String::from_utf8(buffer).unwrap_or_else(|_| String::new())
        }

        let app = Router::new().route("/metrics", get(metrics_handler));
        let addr = format!("0.0.0.0:{metrics_port}");
        tracing::info!("Prometheus metrics endpoint listening on http://{}", addr);

        let listener = match tokio::net::TcpListener::bind(&addr).await {
            Ok(l) => l,
            Err(e) => {
                tracing::error!("Failed to bind metrics endpoint {}: {}", addr, e);
                return;
            }
        };
        if let Err(e) = axum::serve(listener, app).await {
            tracing::error!("Metrics server failed: {}", e);
        }
    });

    // Create service instance
    let service = DataAcquisitionServiceImpl::default();

    // Configure gRPC server address
    let addr: SocketAddr = format!("0.0.0.0:{}", args.port).parse()?;

    // Setup gRPC health check service
    let (health_reporter, health_service) = tonic_health::server::health_reporter();
    health_reporter
        .set_serving::<DataAcquisitionServiceServer<DataAcquisitionServiceImpl>>()
        .await;

    info!("Data Acquisition Service listening on {}", addr);

    // Start gRPC server with graceful shutdown
    Server::builder()
        .layer(GrpcMetricsLayer::new("data-acquisition-service"))
        .add_service(health_service)
        .add_service(DataAcquisitionServiceServer::new(service))
        .serve_with_shutdown(addr, shutdown_signal())
        .await?;

    info!("Data Acquisition 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...");
}