feat(api): harden compression and index audit queries

services/api/database/migrations/022_add_audit_filter_time_indexes.sql

@@ -0,0 +1,6 @@
+-- Support action and resource-type filters combined with timestamp ranges/order.
+CREATE INDEX IF NOT EXISTS idx_audit_log_action_time
+    ON audit_log (action, timestamp DESC);
+
+CREATE INDEX IF NOT EXISTS idx_audit_log_resource_type_time
+    ON audit_log (resource_type, timestamp DESC);

services/api/src/compression.rs

@@ -1,3 +1,7 @@
+//! Response compression is limited to JSON and text payloads. SSE and responses
+//! already carrying Content-Encoding are left untouched; known bodies below 32
+//! bytes are also skipped to avoid spending CPU on negligible payloads.
+
 use axum::http::{header, Extensions, HeaderMap, StatusCode, Version};
 use tower_http::compression::CompressionLayer;
 
@@ -14,7 +18,23 @@ fn should_compress(
         .and_then(|h| h.to_str().ok())
         .unwrap_or("");
     let ct = ct.split(';').next().unwrap_or(ct).trim();
-    ct == "application/json" || ct.starts_with("text/")
+    if headers.contains_key(header::CONTENT_ENCODING)
+        || ct.eq_ignore_ascii_case("text/event-stream")
+    {
+        return false;
+    }
+
+    if headers
+        .get(header::CONTENT_LENGTH)
+        .and_then(|value| value.to_str().ok())
+        .and_then(|value| value.parse::<u64>().ok())
+        .is_some_and(|length| length < 32)
+    {
+        return false;
+    }
+
+    ct.eq_ignore_ascii_case("application/json")
+        || ct.get(..5).is_some_and(|prefix| prefix.eq_ignore_ascii_case("text/"))
 }
 
 pub fn compression_layer() -> CompressionLayer<CompressFn> {
@@ -23,3 +43,72 @@ pub fn compression_layer() -> CompressionLayer<CompressFn> {
         .br(true)
         .compress_when(should_compress as CompressFn)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use axum::{
+        body::Body,
+        http::{Request, Response},
+        routing::get,
+        Router,
+    };
+    use tower::ServiceExt;
+
+    async fn response_for(body: &'static str) -> Response<Body> {
+        let app = Router::new()
+            .route(
+                "/",
+                get(move || async move {
+                    Response::builder()
+                        .header(header::CONTENT_TYPE, "application/json")
+                        .header(header::CONTENT_LENGTH, body.len().to_string())
+                        .body(Body::from(body))
+                        .unwrap()
+                }),
+            )
+            .layer(compression_layer());
+
+        app.oneshot(
+            Request::builder()
+                .header(header::ACCEPT_ENCODING, "gzip")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap()
+    }
+
+    #[tokio::test]
+    async fn compresses_large_json_but_not_small_json() {
+        let large = response_for(r#"{"payload":"this JSON body is large enough to compress"}"#).await;
+        assert_eq!(
+            large.headers().get(header::CONTENT_ENCODING).unwrap(),
+            "gzip"
+        );
+
+        let small = response_for(r#"{"ok":true}"#).await;
+        assert!(!small.headers().contains_key(header::CONTENT_ENCODING));
+    }
+
+    #[test]
+    fn skips_sse_and_already_encoded_responses() {
+        let mut headers = HeaderMap::new();
+        headers.insert(header::CONTENT_TYPE, "text/event-stream".parse().unwrap());
+        assert!(!should_compress(
+            StatusCode::OK,
+            Version::HTTP_11,
+            &headers,
+            &Extensions::new()
+        ));
+
+        headers.insert(header::CONTENT_TYPE, "application/json".parse().unwrap());
+        headers.insert(header::CONTENT_ENCODING, "gzip".parse().unwrap());
+        assert!(!should_compress(
+            StatusCode::OK,
+            Version::HTTP_11,
+            &headers,
+            &Extensions::new()
+        ));
+    }
+}

services/api/src/migrations.rs

@@ -145,6 +145,11 @@ const MIGRATIONS: &[Migration] = &[
         name: "020_add_audit_log_actor_time_index",
         sql: include_str!("../database/migrations/020_add_audit_log_actor_time_index.sql"),
     },
+    Migration {
+        version: "022",
+        name: "022_add_audit_filter_time_indexes",
+        sql: include_str!("../database/migrations/022_add_audit_filter_time_indexes.sql"),
+    },
 ];
 
 // ---------------------------------------------------------------------------

services/api/tests/audit_query_plan_tests.rs

@@ -0,0 +1,54 @@
+mod common;
+
+use common::db_fixture::test_pool;
+
+#[tokio::test]
+async fn audit_filter_query_plans_use_indexes() {
+    let pool = test_pool().await;
+    let mut transaction = pool.begin().await.unwrap();
+
+    sqlx::query("SET LOCAL enable_seqscan = off")
+        .execute(&mut *transaction)
+        .await
+        .unwrap();
+
+    let cases = [
+        (
+            "actor and time range",
+            "SELECT id FROM audit_log WHERE actor = 'plan-test' AND timestamp >= TIMESTAMPTZ '2026-01-01 00:00:00+00' AND timestamp <= TIMESTAMPTZ '2026-02-01 00:00:00+00' ORDER BY timestamp DESC LIMIT 100",
+            "idx_audit_log_actor_time",
+        ),
+        (
+            "action and time range",
+            "SELECT id FROM audit_log WHERE action = 'plan-test' AND timestamp >= TIMESTAMPTZ '2026-01-01 00:00:00+00' AND timestamp <= TIMESTAMPTZ '2026-02-01 00:00:00+00' ORDER BY timestamp DESC LIMIT 100",
+            "idx_audit_log_action_time",
+        ),
+        (
+            "resource type and time range",
+            "SELECT id FROM audit_log WHERE resource_type = 'plan-test' AND timestamp >= TIMESTAMPTZ '2026-01-01 00:00:00+00' AND timestamp <= TIMESTAMPTZ '2026-02-01 00:00:00+00' ORDER BY timestamp DESC LIMIT 100",
+            "idx_audit_log_resource_type_time",
+        ),
+        (
+            "statistics time range",
+            "SELECT COUNT(*) FILTER (WHERE status = 'success'), COUNT(*) FILTER (WHERE status = 'failure') FROM audit_log WHERE timestamp >= TIMESTAMPTZ '2026-01-01 00:00:00+00' AND timestamp <= TIMESTAMPTZ '2026-02-01 00:00:00+00'",
+            "idx_audit_log_timestamp",
+        ),
+    ];
+
+    for (description, query, expected_index) in cases {
+        let plan = sqlx::query_scalar::<_, String>(&format!("EXPLAIN (COSTS OFF) {query}"))
+            .fetch_all(&mut *transaction)
+            .await
+            .unwrap()
+            .join("\n");
+
+        assert!(
+            !plan.contains("Seq Scan"),
+            "{description} unexpectedly uses a sequential scan:\n{plan}"
+        );
+        assert!(
+            plan.contains(expected_index),
+            "{description} did not use {expected_index}:\n{plan}"
+        );
+    }
+}