Merge pull request #376 from firstJOASH/Search-Service

Search service

Author: Mystery-CLI

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

@@ -559,8 +559,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 (
@@ -574,10 +574,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'
             )
@@ -591,18 +592,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())
@@ -622,20 +624,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
             )
@@ -646,15 +647,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)
@@ -663,6 +663,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/event_monitor.rs

@@ -347,6 +347,9 @@ impl EventMonitor {
         // Update user analytics for trade lifecycle events
         self.update_user_analytics(event).await;
 
+        // Update user analytics for trade lifecycle events
+        self.update_user_analytics(event).await;
+
         let mut queue = self.job_queue.lock().await;
         if let Err(e) = queue.enqueue(event_job).await {
             error!("Failed to enqueue event job for event {}: {}", event.id, e);
@@ -420,6 +423,70 @@ impl EventMonitor {
         }
     }
 
+    async fn update_user_analytics(&self, event: &Event) {
+        let d = &event.data;
+        let upsert = |address: &str, seller: bool, buyer: bool, amount: i64, completed: bool, disputed: bool, cancelled: bool| {
+            let pool = self.database.pool().clone();
+            let address = address.to_string();
+            tokio::spawn(async move {
+                let _ = sqlx::query(
+                    r#"
+                    INSERT INTO user_analytics
+                        (address, total_trades, trades_as_seller, trades_as_buyer,
+                         total_volume, completed_trades, disputed_trades, cancelled_trades, updated_at)
+                    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW())
+                    ON CONFLICT (address) DO UPDATE SET
+                        total_trades     = user_analytics.total_trades     + EXCLUDED.total_trades,
+                        trades_as_seller = user_analytics.trades_as_seller + EXCLUDED.trades_as_seller,
+                        trades_as_buyer  = user_analytics.trades_as_buyer  + EXCLUDED.trades_as_buyer,
+                        total_volume     = user_analytics.total_volume     + EXCLUDED.total_volume,
+                        completed_trades = user_analytics.completed_trades + EXCLUDED.completed_trades,
+                        disputed_trades  = user_analytics.disputed_trades  + EXCLUDED.disputed_trades,
+                        cancelled_trades = user_analytics.cancelled_trades + EXCLUDED.cancelled_trades,
+                        updated_at       = NOW()
+                    "#,
+                )
+                .bind(&address)
+                .bind(if seller || buyer { 1i32 } else { 0 })
+                .bind(if seller { 1i32 } else { 0 })
+                .bind(if buyer { 1i32 } else { 0 })
+                .bind(amount)
+                .bind(if completed { 1i32 } else { 0 })
+                .bind(if disputed { 1i32 } else { 0 })
+                .bind(if cancelled { 1i32 } else { 0 })
+                .execute(&pool)
+                .await;
+            });
+        };
+
+        match event.event_type.as_str() {
+            "trade_created" => {
+                let seller = d.get("seller").and_then(|v| v.as_str()).unwrap_or_default();
+                let buyer  = d.get("buyer").and_then(|v| v.as_str()).unwrap_or_default();
+                let amount = d.get("amount").and_then(|v| v.as_i64()).unwrap_or(0);
+                upsert(seller, true,  false, amount, false, false, false);
+                upsert(buyer,  false, true,  amount, false, false, false);
+            }
+            "trade_confirmed" => {
+                let seller = d.get("seller").and_then(|v| v.as_str()).unwrap_or_default();
+                let buyer  = d.get("buyer").and_then(|v| v.as_str()).unwrap_or_default();
+                upsert(seller, false, false, 0, true, false, false);
+                upsert(buyer,  false, false, 0, true, false, false);
+            }
+            "trade_disputed" => {
+                let seller = d.get("seller").and_then(|v| v.as_str()).unwrap_or_default();
+                let buyer  = d.get("buyer").and_then(|v| v.as_str()).unwrap_or_default();
+                upsert(seller, false, false, 0, false, true, false);
+                upsert(buyer,  false, false, 0, false, true, false);
+            }
+            "trade_cancelled" => {
+                let seller = d.get("seller").and_then(|v| v.as_str()).unwrap_or_default();
+                upsert(seller, false, false, 0, false, false, true);
+            }
+            _ => {}
+        }
+    }
+
     async fn get_latest_ledger(&self) -> Result<i64, AppError> {
         let url = format!("{}/ledgers?order=desc&limit=1", self.config.horizon_url);
         let response: HorizonResponse<Ledger> = self.client.get(&url).send().await?.json().await?;

indexer/src/handlers.rs

@@ -25,7 +25,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;
 
@@ -320,13 +320,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

@@ -365,6 +365,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

@@ -317,6 +317,7 @@ pub struct GlobalSearchResponse {
     pub trades: Vec<TradeSearchResult>,
     pub users: Vec<DiscoveryResult>,
     pub arbitrators: Vec<DiscoveryResult>,
+    pub profiles: Vec<UserProfile>,
     pub suggestions: Vec<SearchSuggestion>,
 }
 
@@ -630,6 +631,13 @@ pub struct SetPreferenceRequest {
 #[derive(Debug, Clone, Serialize, Deserialize)]
 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>,
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct PushRegistrationRequest {
     pub device_token: String,

indexer/src/user_handlers.rs

@@ -1,4 +1,7 @@
 use axum::{
+    extract::{Path, Query, State},
+    http::StatusCode,
+    response::Json,
     extract::{Path, State},
     http::StatusCode,
     response::Json,
@@ -12,6 +15,7 @@ use crate::error::AppError;
 use crate::handlers::AppState;
 use crate::models::{
     RegisterUserRequest, SetPreferenceRequest, SetVerificationRequest, UpdateProfileRequest,
+    UserAnalyticsRow, UserPreference, UserProfile, UserSearchQuery,
     UserAnalyticsRow, UserPreference, UserProfile,
 };
 
@@ -181,3 +185,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))
+}