feat: full-text search service with tsvector indexes, FTS queries, user search, and analytics

Author: firstJOASH

indexer/migrations/20260329000001_fulltext_search.sql

@@ -0,0 +1,67 @@
+-- Full-text search: add tsvector columns and GIN indexes for fast FTS
+
+-- 1. Events: searchable on event_type, seller, buyer, trade_id from JSONB
+ALTER TABLE events
+    ADD COLUMN IF NOT EXISTS search_vec tsvector;
+
+UPDATE events
+SET search_vec = to_tsvector('english',
+    coalesce(event_type, '') || ' ' ||
+    coalesce(data->>'seller', '') || ' ' ||
+    coalesce(data->>'buyer', '') || ' ' ||
+    coalesce(data->>'trade_id', '') || ' ' ||
+    coalesce(data->>'arbitrator', '')
+);
+
+CREATE INDEX IF NOT EXISTS idx_events_search_vec ON events USING GIN (search_vec);
+
+-- Keep search_vec in sync on insert/update
+CREATE OR REPLACE FUNCTION events_search_vec_update() RETURNS trigger AS $$
+BEGIN
+    NEW.search_vec := to_tsvector('english',
+        coalesce(NEW.event_type, '') || ' ' ||
+        coalesce(NEW.data->>'seller', '') || ' ' ||
+        coalesce(NEW.data->>'buyer', '') || ' ' ||
+        coalesce(NEW.data->>'trade_id', '') || ' ' ||
+        coalesce(NEW.data->>'arbitrator', '')
+    );
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+DROP TRIGGER IF EXISTS trg_events_search_vec ON events;
+CREATE TRIGGER trg_events_search_vec
+    BEFORE INSERT OR UPDATE ON events
+    FOR EACH ROW EXECUTE FUNCTION events_search_vec_update();
+
+-- 2. User profiles: searchable on address, username_hash, verification
+ALTER TABLE user_profiles
+    ADD COLUMN IF NOT EXISTS search_vec tsvector;
+
+UPDATE user_profiles
+SET search_vec = to_tsvector('english',
+    coalesce(address, '') || ' ' ||
+    coalesce(username_hash, '') || ' ' ||
+    coalesce(verification, '')
+);
+
+CREATE INDEX IF NOT EXISTS idx_user_profiles_search_vec ON user_profiles USING GIN (search_vec);
+
+CREATE OR REPLACE FUNCTION user_profiles_search_vec_update() RETURNS trigger AS $$
+BEGIN
+    NEW.search_vec := to_tsvector('english',
+        coalesce(NEW.address, '') || ' ' ||
+        coalesce(NEW.username_hash, '') || ' ' ||
+        coalesce(NEW.verification, '')
+    );
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+DROP TRIGGER IF EXISTS trg_user_profiles_search_vec ON user_profiles;
+CREATE TRIGGER trg_user_profiles_search_vec
+    BEFORE INSERT OR UPDATE ON user_profiles
+    FOR EACH ROW EXECUTE FUNCTION user_profiles_search_vec_update();
+
+-- 3. GIN index on events.data for fast JSONB key lookups
+CREATE INDEX IF NOT EXISTS idx_events_data_gin ON events USING GIN (data);

indexer/src/database.rs

@@ -469,8 +469,8 @@ impl Database {
         let limit = query.limit.unwrap_or(25).clamp(1, 100);
         let offset = query.offset.unwrap_or(0).max(0);
         let q = query.q.clone().unwrap_or_default();
-        let q_pattern = format!("%{}%", q);
 
+        // Use full-text search when a query term is provided, fall back to no filter
         let rows = sqlx::query_as::<_, TradeSearchResult>(
             r#"
             WITH latest_trade_events AS (
@@ -484,10 +484,11 @@ impl Database {
             trade_base AS (
                 SELECT
                     (e.data->>'trade_id')::BIGINT AS trade_id,
-                    e.data->>'seller' AS seller,
-                    e.data->>'buyer' AS buyer,
-                    (e.data->>'amount')::BIGINT AS amount,
-                    e.timestamp AS created_at
+                    e.data->>'seller'             AS seller,
+                    e.data->>'buyer'              AS buyer,
+                    (e.data->>'amount')::BIGINT   AS amount,
+                    e.timestamp                   AS created_at,
+                    e.search_vec
                 FROM events e
                 WHERE e.event_type = 'trade_created'
             )
@@ -501,18 +502,19 @@ impl Database {
             FROM trade_base tb
             JOIN latest_trade_events lte ON lte.trade_id = tb.trade_id
             WHERE
-                ($1 = '' OR tb.trade_id::TEXT ILIKE $2 OR tb.seller ILIKE $2 OR tb.buyer ILIKE $2)
-                AND ($3::TEXT IS NULL OR lte.event_type = $3)
-                AND ($4::TEXT IS NULL OR tb.seller = $4)
-                AND ($5::TEXT IS NULL OR tb.buyer = $5)
-                AND ($6::BIGINT IS NULL OR tb.amount >= $6)
-                AND ($7::BIGINT IS NULL OR tb.amount <= $7)
-            ORDER BY tb.created_at DESC
-            LIMIT $8 OFFSET $9
+                ($1 = '' OR tb.search_vec @@ plainto_tsquery('english', $1))
+                AND ($2::TEXT IS NULL OR lte.event_type = $2)
+                AND ($3::TEXT IS NULL OR tb.seller = $3)
+                AND ($4::TEXT IS NULL OR tb.buyer = $4)
+                AND ($5::BIGINT IS NULL OR tb.amount >= $5)
+                AND ($6::BIGINT IS NULL OR tb.amount <= $6)
+            ORDER BY
+                CASE WHEN $1 = '' THEN 0 ELSE ts_rank(tb.search_vec, plainto_tsquery('english', $1)) END DESC,
+                tb.created_at DESC
+            LIMIT $7 OFFSET $8
             "#,
         )
         .bind(q.as_str())
-        .bind(q_pattern.as_str())
         .bind(query.status.as_deref())
         .bind(query.seller.as_deref())
         .bind(query.buyer.as_deref())
@@ -532,20 +534,19 @@ impl Database {
     ) -> Result<Vec<DiscoveryResult>, AppError> {
         let limit = query.limit.unwrap_or(25).clamp(1, 100);
         let q = query.q.clone().unwrap_or_default();
-        let q_pattern = format!("%{}%", q);
 
         let rows = sqlx::query_as::<_, DiscoveryResult>(
             r#"
             WITH entities AS (
-                SELECT data->>'seller' AS address, 'user' AS role, timestamp
+                SELECT data->>'seller' AS address, 'user' AS role, timestamp, search_vec
                 FROM events
                 WHERE event_type = 'trade_created' AND data->>'seller' IS NOT NULL
                 UNION ALL
-                SELECT data->>'buyer' AS address, 'user' AS role, timestamp
+                SELECT data->>'buyer' AS address, 'user' AS role, timestamp, search_vec
                 FROM events
                 WHERE event_type = 'trade_created' AND data->>'buyer' IS NOT NULL
                 UNION ALL
-                SELECT data->>'arbitrator' AS address, 'arbitrator' AS role, timestamp
+                SELECT data->>'arbitrator' AS address, 'arbitrator' AS role, timestamp, search_vec
                 FROM events
                 WHERE event_type = 'arb_reg' AND data->>'arbitrator' IS NOT NULL
             )
@@ -556,15 +557,14 @@ impl Database {
                 MAX(timestamp) AS last_seen
             FROM entities
             WHERE
-                ($1 = '' OR address ILIKE $2)
-                AND ($3::TEXT IS NULL OR role = $3)
+                ($1 = '' OR search_vec @@ plainto_tsquery('english', $1))
+                AND ($2::TEXT IS NULL OR role = $2)
             GROUP BY address, role
             ORDER BY seen_count DESC, last_seen DESC
-            LIMIT $4
+            LIMIT $3
             "#,
         )
         .bind(q.as_str())
-        .bind(q_pattern.as_str())
         .bind(query.role.as_deref())
         .bind(limit)
         .fetch_all(&self.pool)
@@ -573,6 +573,34 @@ impl Database {
         Ok(rows)
     }
 
+    /// Full-text search over registered user profiles.
+    pub async fn search_users(
+        &self,
+        q: &str,
+        limit: i64,
+        offset: i64,
+    ) -> Result<Vec<crate::models::UserProfile>, AppError> {
+        let rows = sqlx::query_as::<_, crate::models::UserProfile>(
+            r#"
+            SELECT *
+            FROM user_profiles
+            WHERE $1 = '' OR search_vec @@ plainto_tsquery('english', $1)
+            ORDER BY
+                CASE WHEN $1 = '' THEN 0
+                     ELSE ts_rank(search_vec, plainto_tsquery('english', $1))
+                END DESC,
+                registered_at DESC
+            LIMIT $2 OFFSET $3
+            "#,
+        )
+        .bind(q)
+        .bind(limit.clamp(1, 100))
+        .bind(offset.max(0))
+        .fetch_all(&self.pool)
+        .await?;
+        Ok(rows)
+    }
+
     pub async fn get_search_suggestions(
         &self,
         prefix: &str,

indexer/src/handlers.rs

@@ -22,7 +22,7 @@ use crate::models::{
     AuditQuery, DiscoveryQuery, Event, EventQuery, EventStats, GlobalSearchQuery,
     GlobalSearchResponse, HistoryQuery, IndexerStatus, NewAuditLog, PagedResponse,
     PaginatedResponse, ReplayRequest, RetentionRequest, RetentionResponse, StatsResponse,
-    SuggestionQuery, TradeSearchQuery, WebSocketMessage,
+    SuggestionQuery, TradeSearchQuery, UserProfile, WebSocketMessage,
 };
 use crate::websocket::WebSocketManager;
 
@@ -278,13 +278,15 @@ pub async fn global_search(
     let users = state.database.discover_entities(&user_query).await?;
     let arbitrators = state.database.discover_entities(&arb_query).await?;
     let suggestions = state.database.get_search_suggestions(&params.q, 10).await?;
+    let profiles = state.database.search_users(&params.q, limit, 0).await?;
 
     state.database.record_search(&params.q, "global").await?;
 
     Ok(Json(GlobalSearchResponse {
         trades,
         users,
         arbitrators,
+        profiles,
         suggestions,
     }))
 }

indexer/src/main.rs

@@ -322,6 +322,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .route("/webhooks/stats", get(get_webhook_stats))
         // Users
         .route("/users", post(user_handlers::register_user))
+        .route("/users/search", get(user_handlers::search_users))
         .route(
             "/users/:address",
             get(user_handlers::get_user).patch(user_handlers::update_user),

indexer/src/models.rs

@@ -283,6 +283,7 @@ pub struct GlobalSearchResponse {
     pub trades: Vec<TradeSearchResult>,
     pub users: Vec<DiscoveryResult>,
     pub arbitrators: Vec<DiscoveryResult>,
+    pub profiles: Vec<UserProfile>,
     pub suggestions: Vec<SearchSuggestion>,
 }
 
@@ -575,3 +576,10 @@ pub struct SetPreferenceRequest {
 pub struct SetVerificationRequest {
     pub status: String,
 }
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct UserSearchQuery {
+    pub q: Option<String>,
+    pub limit: Option<i64>,
+    pub offset: Option<i64>,
+}

indexer/src/user_handlers.rs

@@ -1,9 +1,7 @@
 use axum::{
-    extract::{Path, State},
+    extract::{Path, Query, State},
     http::StatusCode,
     response::Json,
-    routing::{get, patch, post, put},
-    Router,
 };
 use chrono::Utc;
 use serde_json::json;
@@ -12,7 +10,7 @@ use crate::error::AppError;
 use crate::handlers::AppState;
 use crate::models::{
     RegisterUserRequest, SetPreferenceRequest, SetVerificationRequest, UpdateProfileRequest,
-    UserAnalyticsRow, UserPreference, UserProfile,
+    UserAnalyticsRow, UserPreference, UserProfile, UserSearchQuery,
 };
 
 /// POST /users — register a new user profile
@@ -181,3 +179,20 @@ pub async fn set_verification(
 
     Ok(Json(json!({ "address": address, "verification": req.status })))
 }
+
+/// GET /users/search?q=…&limit=25&offset=0 — full-text search over user profiles
+pub async fn search_users(
+    State(state): State<AppState>,
+    Query(params): Query<UserSearchQuery>,
+) -> Result<Json<Vec<UserProfile>>, AppError> {
+    let q = params.q.unwrap_or_default();
+    let limit = params.limit.unwrap_or(25);
+    let offset = params.offset.unwrap_or(0);
+
+    if !q.is_empty() {
+        state.database.record_search(&q, "users").await?;
+    }
+
+    let results = state.database.search_users(&q, limit, offset).await?;
+    Ok(Json(results))
+}