feat: health probes, batch lock, chunked worker, trigram search (#823)

* feat: add /healthz /readyz probes with Redis check and distributed batch lock (#721, #722)

* feat: chunked disbursement executor and trigram autocomplete (#723, #725)

Author: ritaifeoluwa

backend/migrations/20260826000001_trigram_index.sql

@@ -0,0 +1,10 @@
+-- #725: Enable pg_trgm extension and add GIN trigram index for
+-- fast prefix/similarity search on usernames.
+
+CREATE EXTENSION IF NOT EXISTS pg_trgm;
+
+-- GIN trigram index on lowercased username for ILIKE 'query%' and
+-- similarity() queries. This makes autocomplete searches scale to
+-- large user tables.
+CREATE INDEX IF NOT EXISTS idx_users_username_trgm
+    ON users USING GIN (LOWER(username) gin_trgm_ops);

backend/src/api/mod.rs

@@ -74,6 +74,7 @@ pub fn user_routes_with_state(state: user::UserState) -> Router {
         )
         .route("/search", get(user::search_users))
         .route("/suggestions", get(user::suggest_usernames))
+        .route("/autocomplete", get(user::autocomplete))
         .route("/friends", get(user::list_friends))
         .route("/friends/request", post(user::send_friend_request))
         .route("/friends/:id/accept", post(user::accept_friend_request))

backend/src/api/user.rs

@@ -407,6 +407,81 @@ pub async fn suggest_usernames(
     .into_response()
 }
 
+/// GET /api/users/autocomplete?q=&limit=
+///
+/// Fast prefix-matching search endpoint backed by a pg_trgm GIN index (#725).
+/// Uses `ILIKE 'query%'` which the trigram index accelerates for large tables.
+/// Falls back to similarity scoring when the prefix index is insufficient.
+pub async fn autocomplete(
+    State(pool): State<sqlx::PgPool>,
+    axum::extract::Query(params): axum::extract::Query<SearchQuery>,
+) -> impl IntoResponse {
+    let term = params.q.trim().to_lowercase();
+    if term.is_empty() {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(serde_json::json!({ "error": "q must not be empty" })),
+        )
+            .into_response();
+    }
+    if term.chars().count() > USERNAME_MAX_LEN {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(serde_json::json!({
+                "error": format!("q must be at most {} characters", USERNAME_MAX_LEN)
+            })),
+        )
+            .into_response();
+    }
+
+    let limit = params.limit.unwrap_or(10).clamp(1, 25);
+    let pattern = format!("{}%", escape_like_pattern(&term));
+
+    // ILIKE 'query%' is served by the trigram GIN index (idx_users_username_trgm)
+    // and by the btree text_pattern_ops index (idx_users_username_lower_pattern).
+    let rows = match sqlx::query(
+        r#"
+        SELECT username, address, avatar_url,
+               similarity(LOWER(username), $1) AS score
+        FROM users
+        WHERE LOWER(username) ILIKE $2 ESCAPE '\'
+        ORDER BY score DESC, LOWER(username) ASC
+        LIMIT $3
+        "#,
+    )
+    .bind(&term)
+    .bind(&pattern)
+    .bind(limit)
+    .fetch_all(&pool)
+    .await
+    {
+        Ok(rows) => rows,
+        Err(e) => {
+            tracing::error!("Autocomplete query failed: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({ "error": "Internal database error" })),
+            )
+                .into_response();
+        }
+    };
+
+    let results: Vec<UserSearchItem> = rows
+        .into_iter()
+        .map(|row| UserSearchItem {
+            username: row.get("username"),
+            address: row.get("address"),
+            avatar_url: row.get("avatar_url"),
+        })
+        .collect();
+
+    Json(serde_json::json!({
+        "query": term,
+        "results": results,
+    }))
+    .into_response()
+}
+
 pub async fn list_friends(State(pool): State<sqlx::PgPool>, auth: AuthUser) -> impl IntoResponse {
     let rows = match sqlx::query(
         r#"

backend/src/main.rs

@@ -9,6 +9,7 @@ use axum::{
     Json, Router,
 };
 use chrono::Utc;
+use redis;
 use serde::Serialize;
 use std::collections::HashMap;
 use std::net::SocketAddr;
@@ -32,6 +33,7 @@ use zaps_backend::services;
 struct HealthState {
     pool: sqlx::PgPool,
     stellar_rpc_url: String,
+    redis_url: Option<String>,
 }
 
 #[derive(Serialize)]
@@ -64,11 +66,20 @@ struct RpcHealth {
     error: Option<String>,
 }
 
+#[derive(Serialize)]
+struct RedisHealth {
+    status: &'static str,
+    latency_ms: u64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    error: Option<String>,
+}
+
 #[derive(Serialize)]
 struct HealthComponents {
     database: DbHealth,
     yield_db: YieldDbHealth,
     soroban_rpc: RpcHealth,
+    redis: RedisHealth,
 }
 
 #[derive(Serialize)]
@@ -227,15 +238,18 @@ async fn main() {
     let bridge_state =
         api::bridge::BridgeState::new(pool.clone(), config.allbridge_api_url.clone());
 
-    // Health check state: pool + Soroban RPC URL for live component probing.
+    // Health check state: pool + Soroban RPC URL + Redis for live component probing.
     let health_state = HealthState {
         pool: pool.clone(),
         stellar_rpc_url: config.stellar_rpc_url.clone(),
+        redis_url: config.redis_url.clone(),
     };
 
     // Setup routes
     let public_routes = Router::new()
         .route("/health", get(health_check))
+        .route("/healthz", get(liveness_probe))
+        .route("/readyz", get(readiness_probe))
         .route("/api/v1/config", get(app_config))
         .with_state(health_state);
 
@@ -390,21 +404,26 @@ async fn app_config() -> Json<AppConfigResponse> {
 // ── /health handler ───────────────────────────────────────────────────────────
 
 async fn health_check(State(state): State<HealthState>) -> impl IntoResponse {
-    // Run all three probes concurrently so latencies don't stack.
-    let (db, yield_db, rpc) = tokio::join!(
+    // Run all probes concurrently so latencies don't stack.
+    let (db, yield_db, rpc, redis) = tokio::join!(
         probe_database(&state.pool),
         probe_yield_db(&state.pool),
         probe_soroban_rpc(&state.stellar_rpc_url),
+        probe_redis(state.redis_url.as_deref()),
     );
 
-    let all_ok = db.status == "ok" && yield_db.status == "ok" && rpc.status == "ok";
+    let all_ok = db.status == "ok"
+        && yield_db.status == "ok"
+        && rpc.status == "ok"
+        && redis.status == "ok";
 
     let body = HealthResponse {
         status: if all_ok { "ok" } else { "degraded" },
         components: HealthComponents {
             database: db,
             yield_db,
             soroban_rpc: rpc,
+            redis,
         },
         checked_at: Utc::now().to_rfc3339(),
     };
@@ -418,6 +437,46 @@ async fn health_check(State(state): State<HealthState>) -> impl IntoResponse {
     (code, Json(body))
 }
 
+/// GET /healthz — lightweight liveness probe for Kubernetes.
+/// Returns 200 if the process is alive; does not check dependencies.
+async fn liveness_probe() -> impl IntoResponse {
+    Json(serde_json::json!({ "status": "ok" }))
+}
+
+/// GET /readyz — Kubernetes readiness probe.
+/// Verifies DB and Redis are reachable; returns HTTP 200 only when both pass.
+async fn readiness_probe(State(state): State<HealthState>) -> impl IntoResponse {
+    let (db, redis) = tokio::join!(
+        probe_database(&state.pool),
+        probe_redis(state.redis_url.as_deref()),
+    );
+
+    let db_ok = db.status == "ok";
+    let redis_ok = redis.status == "ok" || state.redis_url.is_none();
+
+    let status = if db_ok && redis_ok { "ok" } else { "not ready" };
+    let code = if db_ok && redis_ok {
+        StatusCode::OK
+    } else {
+        StatusCode::SERVICE_UNAVAILABLE
+    };
+
+    let mut body = serde_json::json!({
+        "status": status,
+        "db": db_ok,
+        "redis": redis_ok,
+    });
+
+    if let Some(e) = &db.error {
+        body["db_error"] = serde_json::json!(e);
+    }
+    if let Some(e) = &redis.error {
+        body["redis_error"] = serde_json::json!(e);
+    }
+
+    (code, Json(body))
+}
+
 // ── Component probes ──────────────────────────────────────────────────────────
 
 /// Basic Postgres connectivity: a single round-trip to the DB pool.
@@ -537,3 +596,44 @@ async fn probe_soroban_rpc(rpc_url: &str) -> RpcHealth {
         },
     }
 }
+
+/// Redis PING probe. Returns "ok" when Redis responds with PONG.
+async fn probe_redis(redis_url: Option<&str>) -> RedisHealth {
+    let start = Instant::now();
+
+    let Some(url) = redis_url else {
+        return RedisHealth {
+            status: "skipped",
+            latency_ms: 0,
+            error: None,
+        };
+    };
+
+    let result: Result<(), String> = async {
+        let client = redis::Client::open(url).map_err(|e| e.to_string())?;
+        let mut conn = redis::tokio::aio::ConnectionManager::new(client)
+            .await
+            .map_err(|e| e.to_string())?;
+        redis::cmd("PING")
+            .query_async::<String>(&mut conn)
+            .await
+            .map_err(|e| e.to_string())?;
+        Ok(())
+    }
+    .await;
+
+    let latency_ms = start.elapsed().as_millis() as u64;
+
+    match result {
+        Ok(()) => RedisHealth {
+            status: "ok",
+            latency_ms,
+            error: None,
+        },
+        Err(e) => RedisHealth {
+            status: "error",
+            latency_ms,
+            error: Some(e),
+        },
+    }
+}

backend/src/services/disbursement_worker.rs

@@ -50,6 +50,9 @@ const DEFAULT_MAX_ATTEMPTS: i32 = 3;
 const DEFAULT_CLAIM_SIZE: i64 = 100;
 /// A claim older than this is assumed to belong to a dead worker.
 const DEFAULT_LEASE_TIMEOUT_SECS: i64 = 900;
+/// Maximum items per sub-chunk when processing a batch claim.
+/// Prevents gas limit issues on-chain and keeps SDP submissions bounded.
+const CHUNK_SIZE: usize = 50;
 
 pub struct DisbursementWorkerConfig {
     pub poll_interval: Duration,
@@ -193,25 +196,35 @@ async fn process_cycle(
         "Dispatching payout batch"
     );
 
-    // Sequential on purpose: SDP rate-limits per account, and parallel
-    // submissions from one source account contend on the Stellar sequence
-    // number.
-    for recipient in recipients {
-        let attempt = recipient.attempt_count + 1;
-        let outcome = dispatch_one(&recipient, sdp_client).await;
+    // Process recipients in sub-chunks of CHUNK_SIZE to prevent gas limit
+    // issues on-chain and keep SDP submissions bounded. Each chunk is
+    // submitted sequentially; recipients within a chunk are also sequential
+    // (SDP rate-limits per account, and parallel submissions contend on
+    // the Stellar sequence number).
+    for (chunk_idx, chunk) in recipients.chunks(CHUNK_SIZE).enumerate() {
+        tracing::debug!(
+            batch_id = %batch_id,
+            chunk = chunk_idx + 1,
+            chunk_size = chunk.len(),
+            "Processing chunk"
+        );
+
+        for recipient in chunk {
+            let attempt = recipient.attempt_count + 1;
+            let outcome = dispatch_one(recipient, sdp_client).await;
 
         match outcome {
             SdpOutcome::Submitted {
                 payment_id: sdp_payment_id,
                 tx_hash,
             } => {
-                mark_submitted(pool, &recipient, sdp_payment_id.as_deref(), tx_hash.as_deref())
+                mark_submitted(pool, recipient, sdp_payment_id.as_deref(), tx_hash.as_deref())
                     .await?;
                 log_dispatch(pool, batch_id, Some(recipient.id), attempt, "SUBMITTED", None, None)
                     .await?;
             }
             SdpOutcome::Retryable(err) if attempt < config.max_attempts => {
-                mark_retry(pool, &recipient, &err).await?;
+                mark_retry(pool, recipient, &err).await?;
                 log_dispatch(
                     pool,
                     batch_id,
@@ -226,7 +239,7 @@ async fn process_cycle(
             SdpOutcome::Retryable(err) | SdpOutcome::Permanent(err) => {
                 // Either permanently bad, or out of retries. Fail this row only
                 // — one dead recipient must not strand the rest of the batch.
-                mark_failed(pool, &recipient, &err).await?;
+                mark_failed(pool, recipient, &err).await?;
                 log_dispatch(
                     pool,
                     batch_id,
@@ -240,6 +253,7 @@ async fn process_cycle(
             }
         }
     }
+    } // end chunk loop
 
     finalize_batch(pool, batch_id).await?;
     Ok(())