feature:Event Indexing Optimization

Author: bitstarkbridge

api/src/contracts.ts

@@ -38,11 +38,18 @@ export function isTrade(value: unknown): value is Trade {
   );
 }
 
+const EVENT_CATEGORIES = new Set<string>([
+  'trade', 'arb', 'fee', 'tmpl', 'sub', 'gov', 'sys', 'ins', 'oracle',
+]);
+
 export function isEvent(value: unknown): value is Event {
   return (
     isRecord(value) &&
     typeof value.id === 'string' &&
     typeof value.type === 'string' &&
+    typeof value.category === 'string' &&
+    EVENT_CATEGORIES.has(value.category) &&
+    typeof value.schemaVersion === 'number' &&
     typeof value.tradeId === 'string' &&
     typeof value.timestamp === 'string' &&
     isRecord(value.data)

api/src/models.ts

@@ -8,10 +8,41 @@ export interface Trade {
   timestamp: string;
 }
 
+/** High-level event categories matching the contract's topic layout. */
+export type EventCategory =
+  | 'trade'
+  | 'arb'
+  | 'fee'
+  | 'tmpl'
+  | 'sub'
+  | 'gov'
+  | 'sys'
+  | 'ins'
+  | 'oracle';
+
 export interface Event {
   id: string;
+  /** Canonical snake_case event name, e.g. "trade_created" */
   type: string;
+  /** High-level category for client-side filtering */
+  category: EventCategory;
+  /** Schema version from the contract payload (v field) */
+  schemaVersion: number;
   tradeId: string;
   timestamp: string;
-  data: Record<string, any>;
+  data: Record<string, unknown>;
+}
+
+/** Query parameters accepted by GET /events */
+export interface EventFilterParams {
+  limit?: number;
+  offset?: number;
+  event_type?: string;
+  category?: EventCategory;
+  trade_id?: string;
+  from_ledger?: number;
+  to_ledger?: number;
+  from_time?: string;
+  to_time?: string;
+  contract_id?: string;
 }

api/src/resources.ts

@@ -1,5 +1,5 @@
 import { ApiClient } from './client';
-import { Event, Trade } from './models';
+import { Event, EventCategory, EventFilterParams, Trade } from './models';
 
 export class TradesApi {
   constructor(private client: ApiClient) {}
@@ -28,16 +28,29 @@ export class TradesApi {
 export class EventsApi {
   constructor(private client: ApiClient) {}
 
-  async getEvents(limit = 100, tradeId?: string): Promise<Event[]> {
-    const params = new URLSearchParams({ limit: limit.toString() });
-    if (tradeId) params.append('tradeId', tradeId);
+  async getEvents(filters: EventFilterParams = {}): Promise<Event[]> {
+    const params = new URLSearchParams();
+    if (filters.limit !== undefined) params.set('limit', filters.limit.toString());
+    if (filters.offset !== undefined) params.set('offset', filters.offset.toString());
+    if (filters.event_type) params.set('event_type', filters.event_type);
+    if (filters.category) params.set('category', filters.category);
+    if (filters.trade_id) params.set('trade_id', filters.trade_id);
+    if (filters.from_ledger !== undefined) params.set('from_ledger', filters.from_ledger.toString());
+    if (filters.to_ledger !== undefined) params.set('to_ledger', filters.to_ledger.toString());
+    if (filters.from_time) params.set('from_time', filters.from_time);
+    if (filters.to_time) params.set('to_time', filters.to_time);
+    if (filters.contract_id) params.set('contract_id', filters.contract_id);
     return this.client.get(`/events?${params}`);
   }
 
   async getEventsByTrade(tradeId: string): Promise<Event[]> {
     return this.client.get(`/events/trade/${tradeId}`);
   }
 
+  async getEventsByCategory(category: EventCategory, limit = 50): Promise<Event[]> {
+    return this.getEvents({ category, limit });
+  }
+
   async getEvent(id: string): Promise<Event> {
     return this.client.get(`/events/${id}`);
   }

contract/src/events.rs

@@ -143,11 +143,11 @@ pub fn emit_trade_created(env: &Env, trade_id: u64, seller: Address, buyer: Addr
 }
 
 pub fn emit_compliance_failed(env: &Env, user: Address, reason: &soroban_sdk::String) {
-    env.events().publish((symbol_short!("compl_fail"),), (user, reason.clone()));
+    env.events().publish((cat_sys(), symbol_short!("compl_fail")), EvComplianceFailed { v: EVENT_VERSION, user, reason: reason.clone() });
 }
 
 pub fn emit_compliance_passed(env: &Env, trade_id: u64, seller: Address, buyer: Address, amount: u64) {
-    env.events().publish((symbol_short!("compl_pass"),), (trade_id, seller, buyer, amount));
+    env.events().publish((cat_sys(), symbol_short!("compl_pass")), EvCompliancePassed { v: EVENT_VERSION, trade_id, seller, buyer, amount });
 }
 
 pub fn emit_trade_funded(env: &Env, trade_id: u64) {
@@ -334,6 +334,19 @@ pub fn emit_oracle_unavailable(env: &Env, base: Address, quote: Address) {
 // Upgrade system events
 // ---------------------------------------------------------------------------
 
+// ---------------------------------------------------------------------------
+// Compliance event payloads
+// ---------------------------------------------------------------------------
+
+#[contracttype] #[derive(Clone, Debug)]
+pub struct EvComplianceFailed  { pub v: u32, pub user: Address, pub reason: String }
+#[contracttype] #[derive(Clone, Debug)]
+pub struct EvCompliancePassed  { pub v: u32, pub trade_id: u64, pub seller: Address, pub buyer: Address, pub amount: u64 }
+
+// ---------------------------------------------------------------------------
+// Upgrade system events
+// ---------------------------------------------------------------------------
+
 #[contracttype] #[derive(Clone, Debug)]
 pub struct EvUpgradeProposed   { pub v: u32, pub proposed_by: Address, pub executable_after: u32, pub description: String }
 #[contracttype] #[derive(Clone, Debug)]

indexer/migrations/20260328000000_event_indexing_optimization.sql

@@ -0,0 +1,71 @@
+-- Event indexing optimization: add category, schema_version columns and supporting indexes.
+-- These columns are derived from the two-topic layout (category, event_name) emitted by the
+-- Soroban contract and enable efficient category-level filtering without JSON scanning.
+
+ALTER TABLE events
+    ADD COLUMN IF NOT EXISTS category       VARCHAR(20)  NOT NULL DEFAULT 'unknown',
+    ADD COLUMN IF NOT EXISTS schema_version SMALLINT     NOT NULL DEFAULT 1;
+
+-- Backfill category from event_type for existing rows
+UPDATE events SET category = CASE
+    WHEN event_type IN (
+        'trade_created', 'trade_funded', 'trade_completed', 'trade_confirmed',
+        'trade_cancelled', 'time_released', 'metadata_updated',
+        'dispute_raised', 'dispute_resolved', 'partial_resolved'
+    ) THEN 'trade'
+    WHEN event_type IN (
+        'arbitrator_registered', 'arbitrator_removed',
+        'arbitrator_rated', 'arbitrator_rep_updated'
+    ) THEN 'arb'
+    WHEN event_type IN (
+        'fee_updated', 'fees_withdrawn', 'fees_distributed', 'custom_fee_set',
+        'tier_upgraded', 'tier_downgraded', 'tier_config_updated'
+    ) THEN 'fee'
+    WHEN event_type IN (
+        'template_created', 'template_updated', 'template_deactivated', 'template_trade'
+    ) THEN 'tmpl'
+    WHEN event_type IN (
+        'subscribed', 'subscription_renewed', 'subscription_cancelled'
+    ) THEN 'sub'
+    WHEN event_type IN (
+        'proposal_created', 'vote_cast', 'proposal_executed', 'delegated'
+    ) THEN 'gov'
+    WHEN event_type IN (
+        'insurance_provider_registered', 'insurance_provider_removed',
+        'insurance_purchased', 'insurance_claimed'
+    ) THEN 'ins'
+    WHEN event_type IN (
+        'oracle_registered', 'oracle_removed',
+        'oracle_price_fetched', 'oracle_unavailable'
+    ) THEN 'oracle'
+    ELSE 'sys'
+END
+WHERE category = 'unknown';
+
+-- Backfill schema_version from the JSON payload where available
+UPDATE events
+SET schema_version = (data->>'v')::SMALLINT
+WHERE data->>'v' IS NOT NULL
+  AND (data->>'v') ~ '^\d+$';
+
+-- Indexes for the new columns
+CREATE INDEX IF NOT EXISTS idx_events_category
+    ON events (category);
+
+CREATE INDEX IF NOT EXISTS idx_events_category_ledger
+    ON events (category, ledger DESC);
+
+CREATE INDEX IF NOT EXISTS idx_events_schema_version
+    ON events (schema_version);
+
+-- Composite index for the most common query pattern: category + event_type + ledger range
+CREATE INDEX IF NOT EXISTS idx_events_category_type_ledger
+    ON events (category, event_type, ledger DESC);
+
+-- Timestamp range index (already exists in initial migration, kept for reference)
+-- CREATE INDEX IF NOT EXISTS idx_events_timestamp ON events (timestamp DESC);
+
+-- Partial index for trade events (highest query volume)
+CREATE INDEX IF NOT EXISTS idx_events_trade_category
+    ON events (ledger DESC, timestamp DESC)
+    WHERE category = 'trade';