Merge pull request #679 from A5cend-dev/backend-api-performance-enhancement

#438 [BE-API-084] Backend API Performance Enhancement and Storage - Step 84

Author: soomtochukwu

backend/src/Main.rs

@@ -0,0 +1,212 @@
+// backend/src/main.rs
+//
+// Lance — Freelancer Platform with AI Agent Judge
+// BE-API-083: Async Processing Queue for Dispute File Analysis
+//
+// Bootstraps the Axum HTTP server, SQLx connection pool, tracing infrastructure,
+// and the background worker pool that processes dispute file analysis tasks.
+
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use axum::{middleware, Router};
+use sqlx::postgres::PgPoolOptions;
+use tower_http::{
+    cors::{Any, CorsLayer},
+    request_id::{MakeRequestUuid, PropagateRequestIdLayer, SetRequestIdLayer},
+    timeout::TimeoutLayer,
+    trace::TraceLayer,
+};
+use tracing::info;
+use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
+
+mod db;
+mod error;
+mod models;
+mod queue;
+mod routes;
+mod state;
+
+use queue::worker::spawn_dispute_workers;
+use state::AppState;
+
+/// Application entry point.
+///
+/// Initialisation order:
+/// 1. Tracing subscriber (JSON in production, pretty in dev)
+/// 2. Database pool (with validated pool limits for stability under load)
+/// 3. Async dispute queue + worker pool
+/// 4. Axum router with all middleware layers
+/// 5. TCP listener + graceful shutdown signal
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    // ── 1. Tracing ──────────────────────────────────────────────────────────
+    dotenvy::dotenv().ok();
+
+    let log_format = std::env::var("LOG_FORMAT").unwrap_or_else(|_| "pretty".into());
+
+    let filter = EnvFilter::try_from_default_env()
+        .unwrap_or_else(|_| "backend=debug,tower_http=debug,sqlx=warn".into());
+
+    if log_format == "json" {
+        tracing_subscriber::registry()
+            .with(filter)
+            .with(tracing_subscriber::fmt::layer().json())
+            .init();
+    } else {
+        tracing_subscriber::registry()
+            .with(filter)
+            .with(tracing_subscriber::fmt::layer().pretty())
+            .init();
+    }
+
+    info!(
+        version = env!("CARGO_PKG_VERSION"),
+        "Lance backend starting"
+    );
+
+    // ── 2. Database pool ────────────────────────────────────────────────────
+    let database_url = std::env::var("DATABASE_URL")
+        .expect("DATABASE_URL must be set");
+
+    // Pool tuning: keep max connections bounded so that concurrent load tests
+    // never exhaust the PostgreSQL max_connections limit (acceptance criterion).
+    let max_connections: u32 = std::env::var("DB_MAX_CONNECTIONS")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(20);
+
+    let min_connections: u32 = std::env::var("DB_MIN_CONNECTIONS")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(2);
+
+    let pool = PgPoolOptions::new()
+        .max_connections(max_connections)
+        .min_connections(min_connections)
+        .acquire_timeout(std::time::Duration::from_secs(5))
+        .idle_timeout(std::time::Duration::from_secs(600))
+        .max_lifetime(std::time::Duration::from_secs(1800))
+        .connect(&database_url)
+        .await
+        .expect("Failed to create database pool");
+
+    // Run any pending migrations on startup.
+    sqlx::migrate!("./migrations")
+        .run(&pool)
+        .await
+        .expect("Database migration failed");
+
+    info!(
+        max_connections,
+        min_connections,
+        "Database pool initialised"
+    );
+
+    // ── 3. Async queue + workers ────────────────────────────────────────────
+    let worker_count: usize = std::env::var("DISPUTE_WORKER_COUNT")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(4);
+
+    let queue_capacity: usize = std::env::var("DISPUTE_QUEUE_CAPACITY")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(256);
+
+    let (queue_tx, queue_rx) = async_channel::bounded(queue_capacity);
+
+    // Spawn N background workers that drain the queue concurrently.
+    spawn_dispute_workers(worker_count, queue_rx.clone(), pool.clone());
+
+    info!(
+        worker_count,
+        queue_capacity,
+        "Dispute file analysis queue initialised"
+    );
+
+    // ── 4. Application state ────────────────────────────────────────────────
+    let state = Arc::new(AppState {
+        db: pool,
+        dispute_queue: queue_tx,
+    });
+
+    // ── 5. Router ───────────────────────────────────────────────────────────
+    let app = build_router(state);
+
+    // ── 6. Serve ────────────────────────────────────────────────────────────
+    let host = std::env::var("HOST").unwrap_or_else(|_| "0.0.0.0".into());
+    let port: u16 = std::env::var("PORT")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(8080);
+
+    let addr: SocketAddr = format!("{host}:{port}").parse()?;
+    let listener = tokio::net::TcpListener::bind(addr).await?;
+
+    info!(%addr, "Listening");
+
+    axum::serve(listener, app)
+        .with_graceful_shutdown(shutdown_signal())
+        .await?;
+
+    Ok(())
+}
+
+/// Constructs the full Axum `Router` with all middleware layers attached.
+///
+/// Middleware stack (outermost → innermost):
+///   SetRequestId → PropagateRequestId → TraceLayer → TimeoutLayer → CorsLayer
+fn build_router(state: Arc<AppState>) -> Router {
+    let x_request_id = axum::http::HeaderName::from_static("x-request-id");
+
+    Router::new()
+        .merge(routes::health::router())
+        .merge(routes::disputes::router())
+        .with_state(state)
+        // Emit structured per-request spans that include method, URI, status,
+        // latency, and the propagated x-request-id.
+        .layer(TraceLayer::new_for_http())
+        // Hard request timeout — prevents slow DB queries from starving workers.
+        .layer(TimeoutLayer::new(std::time::Duration::from_secs(30)))
+        // CORS — tighten in production via ALLOWED_ORIGINS env var.
+        .layer(
+            CorsLayer::new()
+                .allow_origin(Any)
+                .allow_methods(Any)
+                .allow_headers(Any),
+        )
+        // Propagate request-id header through response so clients can correlate.
+        .layer(PropagateRequestIdLayer::new(x_request_id.clone()))
+        .layer(SetRequestIdLayer::new(
+            x_request_id,
+            MakeRequestUuid,
+        ))
+}
+
+/// Listens for SIGTERM (Docker/k8s) and Ctrl-C and resolves when either fires.
+async fn shutdown_signal() {
+    use tokio::signal;
+
+    let ctrl_c = async {
+        signal::ctrl_c()
+            .await
+            .expect("failed to install Ctrl+C handler");
+    };
+
+    #[cfg(unix)]
+    let terminate = async {
+        signal::unix::signal(signal::unix::SignalKind::terminate())
+            .expect("failed to install signal handler")
+            .recv()
+            .await;
+    };
+
+    #[cfg(not(unix))]
+    let terminate = std::future::pending::<()>();
+
+    tokio::select! {
+        _ = ctrl_c => { info!("Received Ctrl-C, shutting down") },
+        _ = terminate => { info!("Received SIGTERM, shutting down") },
+    }
+}

backend/src/State.rs

@@ -0,0 +1,120 @@
+/// state.rs
+///
+/// Shared application state injected into every Axum handler via
+/// `axum::extract::State<AppState>`.
+///
+/// This module initialises:
+///   • The SQLx Postgres connection pool (with tuned pool limits).
+///   • The `deadpool_redis` connection pool.
+///   • The `RedisCache` wrapper.
+
+use std::env;
+
+use deadpool_redis::{Config as RedisConfig, Runtime};
+use sqlx::{postgres::PgPoolOptions, PgPool};
+use tracing::{info, instrument};
+
+use crate::cache::RedisCache;
+
+/// Cloneable application state shared across all request handlers.
+#[derive(Clone, Debug)]
+pub struct AppState {
+    /// SQLx Postgres pool.
+    pub db: PgPool,
+    /// Redis cache wrapper.
+    pub cache: RedisCache,
+}
+
+impl AppState {
+    /// Build the `AppState` from environment variables.
+    ///
+    /// Required env vars:
+    /// - `DATABASE_URL`  — Postgres connection string.
+    /// - `REDIS_URL`     — Redis connection string (e.g. `redis://localhost:6379`).
+    ///
+    /// Optional env vars with defaults:
+    /// - `DB_MAX_CONNECTIONS`   (default: 20)
+    /// - `DB_MIN_CONNECTIONS`   (default: 5)
+    /// - `REDIS_POOL_SIZE`      (default: 16)
+    #[instrument(name = "AppState::init")]
+    pub async fn init() -> anyhow::Result<Self> {
+        let database_url = env::var("DATABASE_URL")
+            .expect("DATABASE_URL must be set");
+        let redis_url = env::var("REDIS_URL")
+            .expect("REDIS_URL must be set");
+
+        let db_max: u32 = env::var("DB_MAX_CONNECTIONS")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(20);
+
+        let db_min: u32 = env::var("DB_MIN_CONNECTIONS")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(5);
+
+        let redis_pool_size: usize = env::var("REDIS_POOL_SIZE")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(16);
+
+        // ----------------------------------------------------------------
+        // Postgres pool
+        // ----------------------------------------------------------------
+        info!(db_max, db_min, "Initialising Postgres connection pool");
+
+        let db = PgPoolOptions::new()
+            // Hard ceiling on open connections.
+            .max_connections(db_max)
+            // Keep a warm pool — prevents cold-start latency spikes.
+            .min_connections(db_min)
+            // Fail fast rather than queue requests indefinitely.
+            .acquire_timeout(std::time::Duration::from_secs(5))
+            // Recycle long-lived idle connections to avoid stale socket errors.
+            .idle_timeout(std::time::Duration::from_secs(300))
+            // Validate the connection health before handing it to a handler.
+            .test_before_acquire(true)
+            .connect(&database_url)
+            .await?;
+
+        info!("Postgres pool ready");
+
+        // ----------------------------------------------------------------
+        // Redis pool
+        // ----------------------------------------------------------------
+        info!(redis_pool_size, "Initialising Redis connection pool");
+
+        let redis_cfg = RedisConfig::from_url(&redis_url);
+        let redis_pool = redis_cfg
+            .create_pool(Some(Runtime::Tokio1))
+            .map_err(|e| anyhow::anyhow!("Redis pool creation failed: {e}"))?;
+
+        // Override pool size from config.
+        // deadpool_redis uses a builder; we re-create with explicit size.
+        let redis_pool = deadpool_redis::Config {
+            url: Some(redis_url),
+            pool: Some(deadpool_redis::PoolConfig {
+                max_size: redis_pool_size,
+                ..Default::default()
+            }),
+            ..Default::default()
+        }
+        .create_pool(Some(Runtime::Tokio1))
+        .map_err(|e| anyhow::anyhow!("Redis pool creation failed: {e}"))?;
+
+        // Smoke-test the Redis connection at startup.
+        {
+            let mut conn = redis_pool.get().await
+                .map_err(|e| anyhow::anyhow!("Redis connection test failed: {e}"))?;
+            let pong: String = deadpool_redis::redis::cmd("PING")
+                .query_async(&mut conn)
+                .await
+                .map_err(|e| anyhow::anyhow!("Redis PING failed: {e}"))?;
+            info!(pong, "Redis connection verified");
+        }
+
+        let cache = RedisCache::new(redis_pool);
+
+        Ok(Self { db, cache })
+    }
+}