Merge PR #1021: fix(api): resolve compilation errors and complete backend hardening refactor

Author: hman38705

services/api/Cargo.toml

@@ -37,7 +37,7 @@ async-trait = "0.1"
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
-sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio-rustls", "postgres", "chrono", "uuid"] }
+sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio-rustls", "postgres", "chrono", "uuid", "derive"] }
 tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal", "time", "sync"] }
 tokio-util = { version = "0.7", features = ["rt"] }
 tower = { version = "0.5", features = ["util"] }
@@ -60,6 +60,7 @@ base64 = "0.22"
 subtle = "2.5"
 rand = { version = "0.8", features = ["getrandom"] }
 ipnet = "2"
+fastrand = "2.4.1"
 utoipa = { version = "4", features = ["yaml"] }
 
 [features]
@@ -69,8 +70,8 @@ redis-integration = []
 
 [dev-dependencies]
 criterion = { version = "0.5", features = ["html_reports"] }
-testcontainers = "0.23"
-testcontainers-modules = { version = "0.11", features = ["redis"] }
+testcontainers = { version = "0.23" }
+testcontainers-modules = { version = "0.11", features = ["redis", "postgres"] }
 axum = { version = "0.7", features = ["macros"] }
 serde_json = "1"
 

services/api/src/audit.rs

@@ -1,7 +1,6 @@
 use std::net::IpAddr;
 
-pub mod client_ip;
-pub use client_ip::{extract_client_ip, trusted_cidrs_from_env};
+pub use crate::client_ip::{extract_client_ip, trusted_cidrs_from_env};
 
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};

services/api/src/audit_middleware.rs

@@ -1,5 +1,4 @@
-pub mod body_redact;
-pub use body_redact::{body_logging_enabled, redact_sensitive, truncate_body};
+pub use crate::body_redact::{body_logging_enabled, redact_sensitive, truncate_body};
 
 use std::sync::Arc;
 

services/api/src/blockchain.rs

@@ -793,7 +793,7 @@ impl BlockchainClient {
                 total_events = all_events.len(),
                 "fetch_events_since paginated"
             );
-            self.metrics.observe_invalidation("events_pagination_pages", pages);
+            self.metrics.observe_invalidation("events_pagination_pages", pages as usize);
         }
 
         Ok(all_events)

services/api/src/cache/mod.rs

@@ -7,11 +7,8 @@ use std::{
     time::{Duration, Instant},
 };
 
-use redis::redis_module::RedisResult;
-
-
 use anyhow::Context;
-use deadpool_redis::{Config as PoolConfig, Pool, Runtime};
+use deadpool_redis::{Config as PoolConfig, Pool};
 use redis::AsyncCommands;
 use serde::{de::DeserializeOwned, Serialize};
 
@@ -276,8 +273,9 @@ impl RedisCache {
         }
 
         // Deterministically hash the tag so the metadata key is stable.
+        use std::hash::{Hash, Hasher};
         let mut hasher = std::collections::hash_map::DefaultHasher::new();
-        std::hash::Hash::hash(&tag, &mut hasher);
+        tag.cache_keys().join("|").hash(&mut hasher);
         let tag_hash = format!("{:x}", hasher.finish());
 
         let zset_key = self.tag_cfg.tag_key(&tag_hash);
@@ -324,18 +322,19 @@ impl RedisCache {
             "#,
         );
 
-        let mut over_evicted: i64 = 0;
+        let script = std::sync::Arc::new(script);
         self.exec(|mut conn| {
             let zset_key = zset_key.clone();
             let seq_key = seq_key.clone();
             let keys = tag_keys.clone();
+            let script = script.clone();
             async move {
                 let mut argv: Vec<String> = Vec::with_capacity(2 + keys.len());
                 argv.push(tag_ttl_secs.to_string());
                 argv.push(cap.to_string());
                 argv.extend(keys);
 
-                over_evicted = script
+                let _: i64 = script
                     .key(&zset_key)
                     .key(&seq_key)
                     .arg(tag_ttl_secs)
@@ -347,7 +346,6 @@ impl RedisCache {
         })
         .await?;
 
-        // Note: we don't need the evicted count for correctness.
         Ok(())
     }
 
@@ -450,11 +448,14 @@ impl RedisCache {
         T: DeserializeOwned,
     {
         let key = key.to_owned();
-        self.exec(|mut conn| async move {
-            let val: Option<String> = conn.get(&key).await?;
-            match val {
-                Some(raw) => Ok(Some(serde_json::from_str(&raw)?)),
-                None => Ok(None),
+        self.exec(|mut conn| {
+            let key = key.clone();
+            async move {
+                let val: Option<String> = conn.get(&key).await?;
+                match val {
+                    Some(raw) => Ok(Some(serde_json::from_str(&raw)?)),
+                    None => Ok(None),
+                }
             }
         })
         .await
@@ -480,9 +481,12 @@ impl RedisCache {
 
     pub async fn del(&self, key: &str) -> anyhow::Result<()> {
         let key = key.to_owned();
-        self.exec(|mut conn| async move {
-            let _: usize = conn.del(&key).await?;
-            Ok(())
+        self.exec(|mut conn| {
+            let key = key.clone();
+            async move {
+                let _: usize = conn.del(&key).await?;
+                Ok(())
+            }
         })
         .await
     }
@@ -511,23 +515,25 @@ impl RedisCache {
         let pattern = pattern.to_owned();
 
         loop {
-            let pattern_clone = pattern.clone();
             let (next_cursor, batch_deleted) = self
-                .exec(|mut conn| async move {
-                    let (next_cursor, keys): (u64, Vec<String>) = redis::cmd("SCAN")
-                        .arg(cursor)
-                        .arg("MATCH")
-                        .arg(&pattern_clone)
-                        .arg("COUNT")
-                        .arg(100u64)
-                        .query_async(&mut conn)
-                        .await?;
-                    let deleted = if keys.is_empty() {
-                        0
-                    } else {
-                        conn.del(keys).await?
-                    };
-                    Ok((next_cursor, deleted))
+                .exec(|mut conn| {
+                    let pattern_clone = pattern.clone();
+                    async move {
+                        let (next_cursor, keys): (u64, Vec<String>) = redis::cmd("SCAN")
+                            .arg(cursor)
+                            .arg("MATCH")
+                            .arg(&pattern_clone)
+                            .arg("COUNT")
+                            .arg(100u64)
+                            .query_async(&mut conn)
+                            .await?;
+                        let deleted = if keys.is_empty() {
+                            0
+                        } else {
+                            conn.del(keys).await?
+                        };
+                        Ok((next_cursor, deleted))
+                    }
                 })
                 .await?;
 

services/api/src/compression.rs

@@ -1,48 +1,25 @@
-use tower_http::compression::predicate::{NotForContentType, Predicate};
+use axum::http::{header, Extensions, HeaderMap, StatusCode, Version};
 use tower_http::compression::CompressionLayer;
 
-fn should_compress_text_based(content_type: Option<&str>) -> bool {
-    let Some(ct) = content_type else {
-        // If we can't determine content type, avoid wasting CPU.
-        return false;
-    };
+type CompressFn = fn(StatusCode, Version, &HeaderMap, &Extensions) -> bool;
 
-    // Remove common parameters like `charset=utf-8`.
+fn should_compress(
+    _: StatusCode,
+    _: Version,
+    headers: &HeaderMap,
+    _: &Extensions,
+) -> bool {
+    let ct = headers
+        .get(header::CONTENT_TYPE)
+        .and_then(|h| h.to_str().ok())
+        .unwrap_or("");
     let ct = ct.split(';').next().unwrap_or(ct).trim();
-
-    // Only compress text-ish payloads.
-    // Note: application/json is explicitly included.
     ct == "application/json" || ct.starts_with("text/")
 }
 
-pub fn compression_layer() -> CompressionLayer {
-    // Exclude already-compressed/binary formats to avoid CPU waste.
-    // (This primarily protects against cases where `content_type` might be
-    // missing/incorrect while still keeping the middleware safe.)
-    let not_for_binary = NotForContentType::new(vec![
-        "application/zip",
-        "application/gzip",
-        "application/x-gzip",
-        "application/x-zip-compressed",
-        "application/pdf",
-        "image/jpeg",
-        "image/png",
-        "image/webp",
-        "image/gif",
-        "image/svg+xml",
-        "audio/mpeg",
-        "audio/mp4",
-        "video/mp4",
-        "application/octet-stream",
-        "application/x-bzip2",
-        "application/x-7z-compressed",
-    ]);
-
+pub fn compression_layer() -> CompressionLayer<CompressFn> {
     CompressionLayer::new()
         .gzip(true)
         .br(true)
-        // Only apply compression to text-based responses.
-        .compress_when(Predicate::from_fn(should_compress_text_based))
-        .filter(not_for_binary)
+        .compress_when(should_compress as CompressFn)
 }
-

services/api/src/config.rs

@@ -67,7 +67,7 @@ impl CorsConfig {
                     .filter(|s| !s.is_empty())
                     .collect()
             })
-            .unwrap_or_else(|| {
+            .unwrap_or_else(|_| {
                 ["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"]
                     .iter()
                     .map(|s| s.to_string())
@@ -81,7 +81,7 @@ impl CorsConfig {
                     .filter(|s| !s.is_empty())
                     .collect()
             })
-            .unwrap_or_else(|| {
+            .unwrap_or_else(|_| {
                 ["content-type", "authorization"]
                     .iter()
                     .map(|s| s.to_string())

services/api/src/db.rs

@@ -136,7 +136,7 @@ impl Database {
     /// Snapshot pool size/idle into Prometheus gauges.
     /// Call this just before rendering `/metrics` so the values are current.
     pub fn record_pool_metrics(&self) {
-        self.metrics.record_pool_metrics(self.pool.size(), self.pool.num_idle());
+        self.metrics.observe_pool_connections("primary", self.pool.size() as i64, self.pool.num_idle() as i64);
     }
 
     pub async fn new(
@@ -934,7 +934,6 @@ impl Database {
         format!("{:x}", hasher.finalize())
     }
 }
-}
 
 #[cfg(test)]
 mod tests {

services/api/src/email/service.rs

@@ -1,5 +1,4 @@
 use anyhow::{Context, Result};
-use redis::AsyncCommands as _;
 use serde_json::Value;
 use sha2::{Digest, Sha256};
 use std::time::Duration;
@@ -164,7 +163,7 @@ impl EmailService {
         // --- idempotency check ---
         if let (Some(cache), Some(key)) = (&self.cache, idem_key) {
             let redis_key = format!("email:idem:{key}");
-            let mut conn = cache.manager.clone();
+            let mut conn = cache.get_connection().await.context("idempotency Redis connection failed")?;
 
             // Try SET NX — only succeeds for the first send.
             let acquired: Option<String> = redis::cmd("SET")

services/api/src/handlers.rs

@@ -1,4 +1,3 @@
-use crate::content_type::require_json_content_type;
 use std::{
     sync::Arc,
     time::{Duration, Instant},
@@ -752,13 +751,13 @@ pub async fn content(
     let cursor = query.cursor();
     let endpoint = "content";
 
-    let cache_key = keys::api_content(limit);
+    let cache_key = keys::api_content(limit.into());
     let ttl = Duration::from_secs(60 * 60);
 
     let (payload, hit) = state
         .cache
         .get_or_set_json(&cache_key, ttl, || async {
-            let data = state.db.content_cached(limit).await?;
+            let data = state.db.content_cached(limit.into()).await?;
             Ok(data)
         })
         .await
@@ -974,11 +973,11 @@ pub async fn blockchain_user_bets(
 
     let page_data = state
         .blockchain
-        .user_bets_page(&user, page, page_size)
+        .user_bets_page(&user, page, page_size.into())
         .await
         .map_err(into_api_error)?;
 
-    let has_more = (page + 1) * page_size < page_data.total;
+    let has_more = (page + 1) * (page_size as i64) < page_data.total;
     let next_cursor = if has_more {
         Some((page + 1).to_string())
     } else {
@@ -1080,13 +1079,13 @@ pub async fn warm_critical_caches(state: Arc<AppState>) -> anyhow::Result<()> {
 
     let (mut succeeded, mut failed) = (0usize, 0usize);
 
-    warm!("db.statistics",             state.db.statistics_cached().map(|r| r.map(|_| ())),                                                                                      succeeded, failed);
-    warm!("db.featured_markets",       state.db.featured_markets_cached(state.config.featured_limit).map(|r| r.map(|_| ())),                                                     succeeded, failed);
-    warm!("blockchain.health",         state.blockchain.health_check_cached().map(|r| r.map(|_| ())),                                                                             succeeded, failed);
-    warm!("blockchain.platform_stats", state.blockchain.platform_statistics_cached().map(|r| r.map(|_| ())),                                                                     succeeded, failed);
-    warm!("api.statistics",            statistics(State(state.clone())).map(|r| r.map(|_| ()).map_err(|e| anyhow::anyhow!("{e:?}"))),                                             succeeded, failed);
-    warm!("api.featured_markets",      featured_markets(State(state.clone()), Query(PaginationQuery::default())).map(|r| r.map(|_| ()).map_err(|e| anyhow::anyhow!("{e:?}"))),   succeeded, failed);
-    warm!("api.content",               content(State(state.clone()), Query(PaginationQuery::default())).map(|r| r.map(|_| ()).map_err(|e| anyhow::anyhow!("{e:?}"))),             succeeded, failed);
+    warm!("db.statistics",             state.db.statistics_cached(),                                                                                                                succeeded, failed);
+    warm!("db.featured_markets",       state.db.featured_markets_cached(state.config.featured_limit),                                                                               succeeded, failed);
+    warm!("blockchain.health",         state.blockchain.health_check_cached(),                                                                                                       succeeded, failed);
+    warm!("blockchain.platform_stats", state.blockchain.platform_statistics_cached(),                                                                                               succeeded, failed);
+    warm!("api.statistics",            async { statistics(State(state.clone())).await.map(|_| ()).map_err(|e| anyhow::anyhow!("{e:?}")) },                                          succeeded, failed);
+    warm!("api.featured_markets",      async { featured_markets(State(state.clone()), Query(PaginationQuery::default())).await.map(|_| ()).map_err(|e| anyhow::anyhow!("{e:?}")) }, succeeded, failed);
+    warm!("api.content",               async { content(State(state.clone()), Query(PaginationQuery::default())).await.map(|_| ()).map_err(|e| anyhow::anyhow!("{e:?}")) },          succeeded, failed);
 
     tracing::info!(succeeded, failed, total = succeeded + failed, "cache warming complete");
     Ok(())