feat(api): add rate limit config CRUD with hot-reload and audit trail

.github/workflows/backend-ci.yml

@@ -92,6 +92,8 @@ jobs:
           PGPASSWORD=utility_secret psql -h localhost -U utility -d utility_test -f src/time_series/compress.sql
           PGPASSWORD=utility_secret psql -h localhost -U utility -d utility_test -f src/soroban/sync.sql
           PGPASSWORD=utility_secret psql -h localhost -U utility -d utility_test -f src/settlement/schema.sql
+          PGPASSWORD=utility_secret psql -h localhost -U utility -d utility_test -f db/rate_limits.sql
+          PGPASSWORD=utility_secret psql -h localhost -U utility -d utility_test -f db/audit_events.sql
         env:
           PGPASSWORD: utility_secret
 

db/rate_limits.sql

@@ -0,0 +1,13 @@
+CREATE TABLE IF NOT EXISTS rate_limit_configs (
+    id BIGSERIAL PRIMARY KEY,
+    scope_type VARCHAR(16) NOT NULL CHECK (scope_type IN ('global', 'service', 'user')),
+    scope_key VARCHAR(255) NOT NULL,
+    max_tokens BIGINT NOT NULL CHECK (max_tokens > 0),
+    refill_rate BIGINT NOT NULL CHECK (refill_rate >= 0),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    UNIQUE (scope_type, scope_key)
+);
+
+CREATE INDEX IF NOT EXISTS idx_rate_limit_configs_scope
+    ON rate_limit_configs (scope_type, scope_key);

src/api/handlers.rs

@@ -773,3 +773,148 @@ pub async fn retry_dlq(
 ) -> Result<Json<RetryDeadLetterResponse>, StatusCode> {
     retry_dlq_entry(pool, id).await
 }
+
+// ---------------------------------------------------------------------------
+// Rate limit configuration handlers
+// ---------------------------------------------------------------------------
+
+#[derive(Deserialize)]
+pub struct RateLimitConfigListQuery {
+    pub scope_type: Option<String>,
+}
+
+#[derive(Deserialize)]
+pub struct RateLimitAuditQuery {
+    pub limit: Option<i64>,
+}
+
+fn map_rate_limit_error(error: crate::api::rate_limit_config::RateLimitConfigError) -> StatusCode {
+    match error {
+        crate::api::rate_limit_config::RateLimitConfigError::InvalidScopeKey
+        | crate::api::rate_limit_config::RateLimitConfigError::InvalidMaxTokens
+        | crate::api::rate_limit_config::RateLimitConfigError::InvalidRefillRate
+        | crate::api::rate_limit_config::RateLimitConfigError::InvalidGlobalScopeKey => {
+            StatusCode::BAD_REQUEST
+        }
+        crate::api::rate_limit_config::RateLimitConfigError::NotFound => StatusCode::NOT_FOUND,
+        crate::api::rate_limit_config::RateLimitConfigError::Conflict => StatusCode::CONFLICT,
+        crate::api::rate_limit_config::RateLimitConfigError::Database(_) => {
+            StatusCode::INTERNAL_SERVER_ERROR
+        }
+    }
+}
+
+pub async fn list_rate_limit_configs(
+    State(pool): State<Pool<Postgres>>,
+    Query(query): Query<RateLimitConfigListQuery>,
+) -> Result<Json<Vec<crate::api::rate_limit_config::RateLimitConfig>>, StatusCode> {
+    let scope_type = query
+        .scope_type
+        .as_deref()
+        .and_then(crate::api::rate_limit_config::RateLimitScopeType::parse);
+    if query.scope_type.is_some() && scope_type.is_none() {
+        return Err(StatusCode::BAD_REQUEST);
+    }
+
+    crate::api::rate_limit_config::list_configs(&pool, scope_type)
+        .await
+        .map(Json)
+        .map_err(|e| {
+            tracing::error!(error = %e, "failed to list rate limit configs");
+            StatusCode::INTERNAL_SERVER_ERROR
+        })
+}
+
+pub async fn get_rate_limit_config(
+    State(pool): State<Pool<Postgres>>,
+    Path(id): Path<i64>,
+) -> Result<Json<crate::api::rate_limit_config::RateLimitConfig>, StatusCode> {
+    crate::api::rate_limit_config::get_config(&pool, id)
+        .await
+        .map_err(|e| {
+            tracing::error!(error = %e, "failed to get rate limit config");
+            StatusCode::INTERNAL_SERVER_ERROR
+        })?
+        .map(Json)
+        .ok_or(StatusCode::NOT_FOUND)
+}
+
+pub async fn create_rate_limit_config(
+    State(state): State<AppState>,
+    Json(body): Json<crate::api::rate_limit_config::CreateRateLimitConfigRequest>,
+) -> Result<(StatusCode, Json<crate::api::rate_limit_config::RateLimitConfig>), StatusCode> {
+    crate::api::rate_limit_config::create_config(
+        &state.pool,
+        &state.rate_limiter,
+        &state.tenant_rate_limiter,
+        &state.service_rate_limiter,
+        body,
+    )
+    .await
+    .map(|config| (StatusCode::CREATED, Json(config)))
+    .map_err(|e| {
+        tracing::error!(error = %e, "failed to create rate limit config");
+        map_rate_limit_error(e)
+    })
+}
+
+pub async fn update_rate_limit_config(
+    State(state): State<AppState>,
+    Path(id): Path<i64>,
+    Json(body): Json<crate::api::rate_limit_config::UpdateRateLimitConfigRequest>,
+) -> Result<Json<crate::api::rate_limit_config::RateLimitConfig>, StatusCode> {
+    crate::api::rate_limit_config::update_config(
+        &state.pool,
+        &state.rate_limiter,
+        &state.tenant_rate_limiter,
+        &state.service_rate_limiter,
+        id,
+        body,
+    )
+    .await
+    .map(Json)
+    .map_err(|e| {
+        tracing::error!(error = %e, "failed to update rate limit config");
+        map_rate_limit_error(e)
+    })
+}
+
+#[derive(Deserialize)]
+pub struct DeleteRateLimitConfigQuery {
+    #[serde(default = "crate::api::rate_limit_config::default_actor")]
+    pub actor: String,
+}
+
+pub async fn delete_rate_limit_config(
+    State(state): State<AppState>,
+    Path(id): Path<i64>,
+    Query(query): Query<DeleteRateLimitConfigQuery>,
+) -> Result<StatusCode, StatusCode> {
+    crate::api::rate_limit_config::delete_config(
+        &state.pool,
+        &state.rate_limiter,
+        &state.tenant_rate_limiter,
+        &state.service_rate_limiter,
+        id,
+        &query.actor,
+    )
+    .await
+    .map(|_| StatusCode::NO_CONTENT)
+    .map_err(|e| {
+        tracing::error!(error = %e, "failed to delete rate limit config");
+        map_rate_limit_error(e)
+    })
+}
+
+pub async fn list_rate_limit_config_audit(
+    State(pool): State<Pool<Postgres>>,
+    Query(query): Query<RateLimitAuditQuery>,
+) -> Result<Json<Vec<crate::api::rate_limit_config::RateLimitAuditEntry>>, StatusCode> {
+    crate::api::rate_limit_config::list_audit_entries(&pool, query.limit.unwrap_or(50))
+        .await
+        .map(Json)
+        .map_err(|e| {
+            tracing::error!(error = %e, "failed to list rate limit audit entries");
+            StatusCode::INTERNAL_SERVER_ERROR
+        })
+}

src/api/middleware.rs

@@ -6,7 +6,7 @@ use axum::{
     response::Response,
 };
 use dashmap::DashMap;
-use parking_lot::Mutex;
+use parking_lot::{Mutex, RwLock};
 use std::collections::VecDeque;
 use std::net::SocketAddr;
 use std::sync::atomic::{AtomicU64, Ordering};
@@ -128,18 +128,31 @@ pub struct FraudContext {
 }
 
 pub struct DynamicRateLimiter {
-    pub(crate) global_bucket: TokenBucket,
+    global_bucket: RwLock<Arc<TokenBucket>>,
+    default_source_limit: RwLock<(u64, u64)>,
     pub(crate) per_source_buckets: DashMap<String, Arc<TokenBucket>>,
     pub(crate) sliding_windows: DashMap<String, Arc<Mutex<SlidingWindow>>>,
     pub(crate) fraud_contexts: DashMap<String, Arc<Mutex<FraudContext>>>,
     pub(crate) rejection_counts: DashMap<String, u64>,
     pub(crate) last_accessed: DashMap<String, Instant>,
 }
 
+const DEFAULT_GLOBAL_MAX_TOKENS: u64 = 10_000;
+const DEFAULT_GLOBAL_REFILL_RATE: u64 = 10_000;
+const DEFAULT_SOURCE_MAX_TOKENS: u64 = 100;
+const DEFAULT_SOURCE_REFILL_RATE: u64 = 100;
+
 impl DynamicRateLimiter {
     pub fn new() -> Arc<Self> {
         let limiter = Arc::new(Self {
-            global_bucket: TokenBucket::new(10000, 10000),
+            global_bucket: RwLock::new(Arc::new(TokenBucket::new(
+                DEFAULT_GLOBAL_MAX_TOKENS,
+                DEFAULT_GLOBAL_REFILL_RATE,
+            ))),
+            default_source_limit: RwLock::new((
+                DEFAULT_SOURCE_MAX_TOKENS,
+                DEFAULT_SOURCE_REFILL_RATE,
+            )),
             per_source_buckets: DashMap::new(),
             sliding_windows: DashMap::new(),
             fraud_contexts: DashMap::new(),
@@ -159,6 +172,20 @@ impl DynamicRateLimiter {
         limiter
     }
 
+    /// Hot-reload the global token bucket without restarting the service.
+    pub fn set_global_limit(&self, max_tokens: u64, refill_rate: u64) {
+        *self.global_bucket.write() = Arc::new(TokenBucket::new(max_tokens, refill_rate));
+    }
+
+    /// Restore the built-in global defaults after a config row is deleted.
+    pub fn reset_global_limit(&self) {
+        self.set_global_limit(DEFAULT_GLOBAL_MAX_TOKENS, DEFAULT_GLOBAL_REFILL_RATE);
+    }
+
+    fn default_source_limit(&self) -> (u64, u64) {
+        *self.default_source_limit.read()
+    }
+
     fn prune_inactive_sources(&self) {
         let now = Instant::now();
         let timeout = Duration::from_secs(300);
@@ -211,21 +238,27 @@ impl DynamicRateLimiter {
         };
 
         // 2. Global rate limit
-        if !self.global_bucket.try_consume(1) {
+        let global_bucket = self.global_bucket.read().clone();
+        if !global_bucket.try_consume(1) {
             self.increment_rejection("global");
             return false;
         }
 
         // 3. Per-source rate limit
-        let limit = if is_flagged { 10 } else { 100 };
+        let (default_max, default_refill) = self.default_source_limit();
+        let limit = if is_flagged {
+            10
+        } else {
+            default_max.max(default_refill)
+        };
 
         let bucket = {
             let b = self
                 .per_source_buckets
                 .get(source_id)
                 .map(|e| e.value().clone());
             if let Some(b) = b {
-                if b.refill_rate == limit || source_id.starts_with("test-large-") {
+                if (b.max_tokens == limit && b.refill_rate == limit) || source_id.starts_with("test-large-") {
                     b
                 } else {
                     let new_b = Arc::new(TokenBucket::new(limit, limit));
@@ -338,12 +371,20 @@ impl DynamicRateLimiter {
     }
 }
 
+pub fn is_rate_limit_admin_path(path: &str) -> bool {
+    path.starts_with("/api/v1/rate-limits/configs")
+}
+
 pub async fn rate_limit_layer(
     State(limiter): State<Arc<DynamicRateLimiter>>,
     connect_info: Option<ConnectInfo<SocketAddr>>,
     req: Request<Body>,
     next: Next,
 ) -> Response {
+    if is_rate_limit_admin_path(req.uri().path()) {
+        return next.run(req).await;
+    }
+
     let source_id = connect_info
         .map(|ConnectInfo(addr)| addr.ip().to_string())
         .unwrap_or_else(|| "unknown".to_string());
@@ -418,18 +459,19 @@ mod tests {
         let source = "test-large-source";
 
         // Use a very high global limit and per-source limit to ensure we can reach 1000
-        limiter.global_bucket.tokens.store(100000, Ordering::SeqCst);
+        limiter.set_global_limit(100_000, 100_000);
         limiter
             .per_source_buckets
             .insert(source.to_string(), Arc::new(TokenBucket::new(2000, 2000)));
 
         for i in 0..1000 {
             let ok = limiter.try_consume(source);
+            let global_tokens = limiter.global_bucket.read().tokens.load(Ordering::SeqCst);
             assert!(
                 ok,
                 "Failed at request {} - Global: {}, Per: {}",
                 i,
-                limiter.global_bucket.tokens.load(Ordering::SeqCst),
+                global_tokens,
                 limiter
                     .per_source_buckets
                     .get(source)
@@ -750,6 +792,10 @@ pub async fn tenant_rate_limit_layer(
     req: Request<Body>,
     next: Next,
 ) -> Response {
+    if is_rate_limit_admin_path(req.uri().path()) {
+        return next.run(req).await;
+    }
+
     let tenant_id = req
         .headers()
         .get("x-tenant-id")
@@ -766,6 +812,67 @@ pub async fn tenant_rate_limit_layer(
     next.run(req).await
 }
 
+/// Per-service tier rate limiter (same token-bucket engine as tenant limiting).
+#[derive(Clone)]
+pub struct ServiceRateLimiter(Arc<TenantRateLimiter>);
+
+impl ServiceRateLimiter {
+    pub fn new(default_max_tokens: u64, default_refill_rate: u64) -> Arc<Self> {
+        Arc::new(Self(TenantRateLimiter::new(
+            default_max_tokens,
+            default_refill_rate,
+        )))
+    }
+
+    pub fn inner(&self) -> &Arc<TenantRateLimiter> {
+        &self.0
+    }
+
+    pub fn set_service_limit(&self, service_id: &str, limit: TenantLimit) {
+        self.0.set_tenant_limit(service_id, limit);
+    }
+
+    pub fn remove_service_override(&self, service_id: &str) {
+        self.0.remove_tenant_override(service_id);
+    }
+
+    pub fn try_consume(&self, service_id: &str, tokens: u64) -> bool {
+        self.0.try_consume(service_id, tokens)
+    }
+}
+
+/// Axum middleware that enforces per-service tier rate limits.
+///
+/// The service tier is identified via the `X-Service-ID` request header. When
+/// the header is absent the request bypasses service-tier limiting.
+pub async fn service_rate_limit_layer(
+    State(limiter): State<Arc<ServiceRateLimiter>>,
+    req: Request<Body>,
+    next: Next,
+) -> Response {
+    if is_rate_limit_admin_path(req.uri().path()) {
+        return next.run(req).await;
+    }
+
+    let Some(service_id) = req
+        .headers()
+        .get("x-service-id")
+        .and_then(|v| v.to_str().ok())
+        .filter(|value| !value.is_empty())
+    else {
+        return next.run(req).await;
+    };
+
+    if !limiter.try_consume(service_id, 1) {
+        warn!(service = %service_id, "service rate limit exceeded");
+        return Response::builder()
+            .status(StatusCode::TOO_MANY_REQUESTS)
+            .body(Body::from("service rate limit exceeded"))
+            .unwrap();
+    }
+    next.run(req).await
+}
+
 pub async fn slo_monitoring_layer(req: Request<Body>, next: Next) -> Response {
     let route = req.uri().path().to_string();
     let started = Instant::now();