feat(indexer): replace polling with Stellar RPC event fetching

backend/.env.example

@@ -35,6 +35,15 @@ JWT_SECRET=your-jwt-secret-here
 # Indexer configuration
 INDEXER_POLL_INTERVAL_MS=10000
 
+# Fallback polling mode (default: false).
+# When false, the indexer uses Stellar RPC getEvents with cursor-based pagination.
+# When true, the indexer falls back to the legacy fixed-interval polling approach.
+INDEXER_FALLBACK_POLLING_ENABLED=false
+
+# Polling interval for fallback mode (milliseconds, minimum 5000, default 10000).
+# Only used when INDEXER_FALLBACK_POLLING_ENABLED=true.
+INDEXER_FALLBACK_POLL_INTERVAL_MS=10000
+
 # Reconciliation job interval in milliseconds (minimum 10000, default 60000 = 1 minute)
 RECONCILIATION_INTERVAL_MS=60000
 

backend/src/config/validateEnv.ts

@@ -55,6 +55,14 @@ const archiveCronIntervalSchema = z
     message: "must be a valid number >= 60000 (minimum 1 minute)",
   });
 
+// Indexer fallback polling interval validation
+const fallbackPollIntervalSchema = z
+  .string()
+  .transform((val: string) => parseInt(val, 10))
+  .refine((val: number) => !isNaN(val) && val >= 1000, {
+    message: "must be a valid number >= 1000 (minimum 1 second)",
+  });
+
 // Admin API key validation
 const adminApiKeySchema = z
   .string()
@@ -84,7 +92,8 @@ const envSchema = z.object({
   INDEXER_POLL_INTERVAL_MS: indexerPollIntervalSchema.optional().default(10000),
   RECONCILIATION_INTERVAL_MS: reconciliationIntervalSchema.optional().default(60000),
   ARCHIVE_CRON_INTERVAL_MS: archiveCronIntervalSchema.optional().default(86400000),
-  WEBHOOK_DEAD_LETTER_PRUNE_INTERVAL_MS: archiveCronIntervalSchema.optional().default(86400000),
+  INDEXER_FALLBACK_POLLING_ENABLED: z.string().optional().default("false"),
+  INDEXER_FALLBACK_POLL_INTERVAL_MS: fallbackPollIntervalSchema.optional().default(10000),
   ALLOWED_ORIGINS: z.string().optional(),
 });
 
@@ -105,7 +114,8 @@ export interface ValidatedConfig {
   indexerPollIntervalMs: number;
   reconciliationIntervalMs: number;
   archiveCronIntervalMs: number;
-  webhookDeadLetterPruneIntervalMs: number;
+  indexerFallbackPollingEnabled: boolean;
+  indexerFallbackPollIntervalMs: number;
   adminApiKey: string | null;
   allowedOrigins: string | undefined;
 }
@@ -270,7 +280,8 @@ export function validateEnv(): ValidatedConfig {
       indexerPollIntervalMs: env.INDEXER_POLL_INTERVAL_MS,
       reconciliationIntervalMs: env.RECONCILIATION_INTERVAL_MS,
       archiveCronIntervalMs: env.ARCHIVE_CRON_INTERVAL_MS,
-      webhookDeadLetterPruneIntervalMs: env.WEBHOOK_DEAD_LETTER_PRUNE_INTERVAL_MS,
+      indexerFallbackPollingEnabled: env.INDEXER_FALLBACK_POLLING_ENABLED,
+      indexerFallbackPollIntervalMs: env.INDEXER_FALLBACK_POLL_INTERVAL_MS,
     },
     "configuration validated",
   );
@@ -292,7 +303,8 @@ export function validateEnv(): ValidatedConfig {
     indexerPollIntervalMs: env.INDEXER_POLL_INTERVAL_MS,
     reconciliationIntervalMs: env.RECONCILIATION_INTERVAL_MS,
     archiveCronIntervalMs: env.ARCHIVE_CRON_INTERVAL_MS,
-    webhookDeadLetterPruneIntervalMs: env.WEBHOOK_DEAD_LETTER_PRUNE_INTERVAL_MS,
+    indexerFallbackPollingEnabled: process.env.INDEXER_FALLBACK_POLLING_ENABLED === "true",
+    indexerFallbackPollIntervalMs: env.INDEXER_FALLBACK_POLL_INTERVAL_MS,
     adminApiKey,
     allowedOrigins: env.ALLOWED_ORIGINS,
   };

backend/src/migrations/0003_add_indexer_cursor.down.sql

@@ -0,0 +1,4 @@
+-- Migration down: 0003_add_indexer_cursor
+-- Purpose : Remove the indexer checkpoint table.
+
+DROP TABLE IF EXISTS indexer_cursor;

backend/src/migrations/0003_add_indexer_cursor.sql

@@ -0,0 +1,12 @@
+-- Migration: 0003_add_indexer_cursor
+-- Purpose : Add an indexer checkpoint table to persist the last
+--           successfully processed ledger sequence. This allows
+--           the indexer to resume from the correct position after
+--           a restart, preventing event loss and duplicate processing.
+-- Safe to run multiple times: uses CREATE TABLE IF NOT EXISTS
+-- and a singleton row pattern (id = 1).
+
+CREATE TABLE IF NOT EXISTS indexer_cursor (
+    id INTEGER PRIMARY KEY CHECK (id = 1),
+    last_ledger_sequence INTEGER NOT NULL
+);

backend/src/migrations/0003_add_indexer_cursor.ts

@@ -0,0 +1,21 @@
+import Database from "better-sqlite3";
+
+export function up(db: Database.Database): void {
+  db.exec(`
+    CREATE TABLE IF NOT EXISTS indexer_cursor (
+      id INTEGER PRIMARY KEY CHECK (id = 1),
+      last_ledger_sequence INTEGER NOT NULL
+    );
+  `);
+}
+
+export function down(db: Database.Database): void {
+  db.exec(`DROP TABLE IF EXISTS indexer_cursor;`);
+}
+
+if (require.main === module) {
+  const dbPath = process.env.DB_PATH ?? "backend/data/streams.db";
+  const db = new Database(dbPath);
+  up(db);
+  db.close();
+}