feat(backend): implement Soroban event indexer with cursor tracking

backend/.env.example

@@ -27,3 +27,19 @@ HORIZON_TIMEOUT=30000
 HORIZON_CACHE_TTL=30000
 
 LOG_LEVEL=debug
+
+# On-chain event indexer (pnpm backend:indexer / pnpm backend:backfill)
+SOROBAN_RPC_URL=https://soroban-testnet.stellar.org
+CONTRACT_ID=C...
+# Cold-start ledger, and the fallback used when the stored cursor is found to be
+# outside the RPC's retention window. Pick a ledger no older than your RPC's
+# retention (commonly ~7 days of history on public RPC providers).
+INDEXER_START_LEDGER=1
+# Delay between polls once the worker has caught up with the chain tip, in ms.
+INDEXER_POLL_INTERVAL_MS=5000
+# Max events requested per getEvents call.
+INDEXER_PAGE_LIMIT=100
+# How long SIGTERM/SIGINT wait for an in-flight batch to finish before forcing exit, in ms.
+INDEXER_SHUTDOWN_GRACE_MS=30000
+# Ceiling for the exponential backoff applied after a transient RPC/DB error, in ms.
+INDEXER_MAX_RETRY_DELAY_MS=30000

backend/package.json

@@ -14,7 +14,9 @@
     "type-check": "tsc --noEmit",
     "test": "NODE_ENV=test node --test $(find dist -name '*.test.js' | tr '\n' ' ')",
     "db:migrate": "node dist/db/migrate.js",
-    "db:migrate:down": "node dist/db/migrate.js down"
+    "db:migrate:down": "node dist/db/migrate.js down",
+    "indexer": "node dist/workers/indexer.js",
+    "backfill": "node dist/workers/backfill.js"
   },
   "dependencies": {
     "@stellar/stellar-sdk": "^16.0.1",

backend/src/config/indexer.ts

@@ -0,0 +1,73 @@
+/**
+ * Environment configuration for the on-chain event indexer worker and the
+ * backfill CLI. Read lazily via indexerConfig() (not at module load) so
+ * tests can set process.env per-test before calling it.
+ */
+
+const DEFAULT_POLL_INTERVAL_MS = 5_000;
+const DEFAULT_PAGE_LIMIT = 100;
+const DEFAULT_SHUTDOWN_GRACE_MS = 30_000;
+const DEFAULT_MAX_RETRY_DELAY_MS = 30_000;
+
+function requiredString(name: string): string {
+  const raw = process.env[name];
+  if (!raw) {
+    throw new Error(`${name} must be set`);
+  }
+  return raw;
+}
+
+function positiveInteger(name: string, fallback: number): number {
+  const raw = process.env[name];
+  if (raw === undefined) return fallback;
+  const value = Number(raw);
+  if (!Number.isInteger(value) || value <= 0) {
+    throw new Error(`${name} must be a positive integer`);
+  }
+  return value;
+}
+
+function requiredNonNegativeInteger(name: string): number {
+  const raw = process.env[name];
+  if (raw === undefined) {
+    throw new Error(`${name} must be set`);
+  }
+  const value = Number(raw);
+  if (!Number.isInteger(value) || value < 0) {
+    throw new Error(`${name} must be a non-negative integer`);
+  }
+  return value;
+}
+
+export interface IndexerConfig {
+  /** Soroban RPC endpoint polled for contract events. */
+  rpcUrl: string;
+  /** Contract id (C...) whose "autoshare" events are indexed. */
+  contractId: string;
+  /**
+   * Ledger to start indexing from on a cold start (no stored cursor), and the
+   * ledger the cursor falls back to when it is found to be outside the RPC's
+   * retention window.
+   */
+  startLedger: number;
+  /** Delay between successive getEvents polls once the worker is caught up. */
+  pollIntervalMs: number;
+  /** Max events requested per getEvents call. */
+  pageLimit: number;
+  /** How long SIGTERM waits for an in-flight batch to finish before forcing exit. */
+  shutdownGraceMs: number;
+  /** Ceiling for the exponential backoff applied after a transient RPC/DB error. */
+  maxRetryDelayMs: number;
+}
+
+export function indexerConfig(): IndexerConfig {
+  return {
+    rpcUrl: requiredString('SOROBAN_RPC_URL'),
+    contractId: requiredString('CONTRACT_ID'),
+    startLedger: requiredNonNegativeInteger('INDEXER_START_LEDGER'),
+    pollIntervalMs: positiveInteger('INDEXER_POLL_INTERVAL_MS', DEFAULT_POLL_INTERVAL_MS),
+    pageLimit: positiveInteger('INDEXER_PAGE_LIMIT', DEFAULT_PAGE_LIMIT),
+    shutdownGraceMs: positiveInteger('INDEXER_SHUTDOWN_GRACE_MS', DEFAULT_SHUTDOWN_GRACE_MS),
+    maxRetryDelayMs: positiveInteger('INDEXER_MAX_RETRY_DELAY_MS', DEFAULT_MAX_RETRY_DELAY_MS),
+  };
+}

backend/src/db/__tests__/indexerCursorMigration.test.ts

@@ -0,0 +1,162 @@
+/**
+ * Migration 005 (indexer_cursor) smoke + round-trip tests, run against the
+ * project's real migration files via migrationHelper.ts rather than an
+ * inlined copy of the SQL.
+ *
+ * Requires a live PostgreSQL instance pointed to by DATABASE_URL. Skipped
+ * automatically when it isn't set, matching migrations.test.ts.
+ */
+
+import { describe, it, before, after } from 'node:test';
+import assert from 'node:assert/strict';
+import { Pool } from 'pg';
+import type { QueryResultRow } from 'pg';
+import { applyMigrations, listMigrationFiles } from './migrationHelper.js';
+
+async function tableExists(pool: Pool, tableName: string): Promise<boolean> {
+  interface Row extends QueryResultRow {
+    exists: boolean;
+  }
+  const res = await pool.query<Row>(
+    `SELECT EXISTS (
+       SELECT 1 FROM information_schema.tables
+       WHERE table_schema = 'public' AND table_name = $1
+     ) AS exists`,
+    [tableName]
+  );
+  return res.rows[0]?.exists ?? false;
+}
+
+async function indexExists(pool: Pool, indexName: string): Promise<boolean> {
+  interface Row extends QueryResultRow {
+    exists: boolean;
+  }
+  const res = await pool.query<Row>(
+    `SELECT EXISTS (
+       SELECT 1 FROM pg_indexes
+       WHERE schemaname = 'public' AND indexname = $1
+     ) AS exists`,
+    [indexName]
+  );
+  return res.rows[0]?.exists ?? false;
+}
+
+async function columnInfo(
+  pool: Pool,
+  table: string,
+  column: string
+): Promise<{ is_nullable: string; data_type: string } | null> {
+  interface Row extends QueryResultRow {
+    is_nullable: string;
+    data_type: string;
+  }
+  const res = await pool.query<Row>(
+    `SELECT is_nullable, data_type FROM information_schema.columns
+     WHERE table_schema = 'public' AND table_name = $1 AND column_name = $2`,
+    [table, column]
+  );
+  return res.rows[0] ?? null;
+}
+
+const skipReason: string | false = process.env.DATABASE_URL ? false : 'DATABASE_URL not configured';
+
+describe('Migration 005 — indexer_cursor', { skip: skipReason }, () => {
+  let pool!: Pool;
+
+  before(async () => {
+    pool = new Pool({ connectionString: process.env.DATABASE_URL });
+    await applyMigrations(pool, 'down');
+  });
+
+  after(async () => {
+    await applyMigrations(pool, 'down');
+    await pool.end();
+  });
+
+  it('numbers exactly one migration pair as 005, following the existing 001-004 sequence', () => {
+    const up = listMigrationFiles('up');
+    const has005 = up.some((f) => f.startsWith('005_'));
+    assert.ok(has005, `expected a 005_*_up.sql migration, found: ${up.join(', ')}`);
+    assert.strictEqual(up.filter((f) => f.startsWith('005_')).length, 1);
+  });
+
+  it('applies cleanly on top of 001-004 — indexer_cursor table and index exist', async () => {
+    await applyMigrations(pool, 'up');
+
+    assert.ok(await tableExists(pool, 'indexer_cursor'), 'indexer_cursor table missing');
+    assert.ok(await indexExists(pool, 'idx_indexer_cursor_updated_at'), 'idx_indexer_cursor_updated_at missing');
+  });
+
+  it('contract_id is the primary key (one row per contract)', async () => {
+    await pool.query(
+      `INSERT INTO indexer_cursor (contract_id, last_ledger, paging_token) VALUES ('CONTRACT_A', 100, 'tok-1')`
+    );
+
+    await assert.rejects(
+      () =>
+        pool.query(
+          `INSERT INTO indexer_cursor (contract_id, last_ledger, paging_token) VALUES ('CONTRACT_A', 200, 'tok-2')`
+        ),
+      /duplicate key|unique/i
+    );
+
+    await pool.query(`DELETE FROM indexer_cursor WHERE contract_id = 'CONTRACT_A'`);
+  });
+
+  it('last_ledger is NOT NULL and paging_token is nullable (cold-start rows have no token yet)', async () => {
+    const lastLedgerCol = await columnInfo(pool, 'indexer_cursor', 'last_ledger');
+    const pagingTokenCol = await columnInfo(pool, 'indexer_cursor', 'paging_token');
+
+    assert.strictEqual(lastLedgerCol?.is_nullable, 'NO');
+    assert.strictEqual(pagingTokenCol?.is_nullable, 'YES');
+
+    await assert.rejects(
+      () => pool.query(`INSERT INTO indexer_cursor (contract_id, paging_token) VALUES ('CONTRACT_B', 'tok')`),
+      /null value|not-null/i
+    );
+  });
+
+  it('supports multiple rows, one per contract id', async () => {
+    await pool.query(`INSERT INTO indexer_cursor (contract_id, last_ledger) VALUES ('CONTRACT_C', 10)`);
+    await pool.query(`INSERT INTO indexer_cursor (contract_id, last_ledger) VALUES ('CONTRACT_D', 20)`);
+
+    const res = await pool.query('SELECT contract_id, last_ledger FROM indexer_cursor ORDER BY contract_id');
+    const rows = res.rows.map((r: { contract_id: string; last_ledger: string }) => ({
+      contract_id: r.contract_id,
+      last_ledger: Number(r.last_ledger),
+    }));
+    assert.deepStrictEqual(rows, [
+      { contract_id: 'CONTRACT_C', last_ledger: 10 },
+      { contract_id: 'CONTRACT_D', last_ledger: 20 },
+    ]);
+  });
+
+  it('down migration drops indexer_cursor and its index, restoring prior schema exactly', async () => {
+    await applyMigrations(pool, 'down');
+
+    assert.ok(!(await tableExists(pool, 'indexer_cursor')), 'indexer_cursor still exists after down');
+    assert.ok(
+      !(await indexExists(pool, 'idx_indexer_cursor_updated_at')),
+      'idx_indexer_cursor_updated_at still exists after down'
+    );
+
+    // The tables from earlier migrations must be untouched by 005's down migration.
+    assert.ok(await tableExists(pool, 'transactions'), 'transactions table was dropped by 005 down');
+    assert.ok(await tableExists(pool, 'groups'), 'groups table was dropped by 005 down');
+    assert.ok(await tableExists(pool, 'idempotency_keys'), 'idempotency_keys table was dropped by 005 down');
+  });
+
+  it('is re-appliable — up, down, up again leaves the same schema with no leftover state', async () => {
+    await applyMigrations(pool, 'up');
+    assert.ok(await tableExists(pool, 'indexer_cursor'));
+
+    await applyMigrations(pool, 'down');
+    assert.ok(!(await tableExists(pool, 'indexer_cursor')));
+
+    await applyMigrations(pool, 'up');
+    assert.ok(await tableExists(pool, 'indexer_cursor'));
+
+    const res = await pool.query('SELECT COUNT(*)::int AS count FROM indexer_cursor');
+    assert.strictEqual(res.rows[0].count, 0, 'expected a clean table with no rows after a fresh up');
+  });
+});

backend/src/db/__tests__/migrationHelper.ts

@@ -0,0 +1,34 @@
+/**
+ * Applies the project's real migration files (src/db/migrations/*.sql)
+ * against a given pool — used by tests that need a fully migrated schema
+ * (indexer_cursor plus the tables it references) rather than a hand-inlined
+ * copy of the SQL. Not a test file itself.
+ */
+
+import { readFileSync, readdirSync } from 'node:fs';
+import { dirname, join, resolve } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import type { Pool } from 'pg';
+
+const __dirname = dirname(fileURLToPath(import.meta.url));
+// dist/db/__tests__/migrationHelper.js -> backend root -> src/db/migrations
+const migrationsDir = resolve(__dirname, '../../../src/db/migrations');
+
+function migrationFiles(direction: 'up' | 'down'): string[] {
+  const suffix = `_${direction}.sql`;
+  const files = readdirSync(migrationsDir)
+    .filter((f) => f.endsWith(suffix))
+    .sort();
+  return direction === 'down' ? files.reverse() : files;
+}
+
+export async function applyMigrations(pool: Pool, direction: 'up' | 'down'): Promise<void> {
+  for (const file of migrationFiles(direction)) {
+    const sql = readFileSync(join(migrationsDir, file), 'utf8');
+    await pool.query(sql);
+  }
+}
+
+export function listMigrationFiles(direction: 'up' | 'down'): string[] {
+  return migrationFiles(direction);
+}

backend/src/db/migrations/005_create_indexer_cursor_down.sql

@@ -0,0 +1,6 @@
+-- =============================================================================
+-- Down Migration: 005_create_indexer_cursor
+-- =============================================================================
+
+DROP INDEX IF EXISTS idx_indexer_cursor_updated_at;
+DROP TABLE IF EXISTS indexer_cursor;

backend/src/db/migrations/005_create_indexer_cursor_up.sql

@@ -0,0 +1,15 @@
+-- =============================================================================
+-- Migration: 005_create_indexer_cursor
+-- Tracks per-contract on-chain event indexing progress (last processed ledger
+-- and RPC paging token) so the indexer worker and backfill CLI can resume
+-- without losing or duplicating work across restarts.
+-- =============================================================================
+
+CREATE TABLE IF NOT EXISTS indexer_cursor (
+  contract_id  VARCHAR(64)  PRIMARY KEY,
+  last_ledger  BIGINT       NOT NULL,
+  paging_token TEXT,
+  updated_at   TIMESTAMPTZ  NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_indexer_cursor_updated_at ON indexer_cursor (updated_at);

backend/src/index.ts

@@ -16,6 +16,7 @@ import { idempotency } from './middleware/idempotency.js';
 import { apiLimiter } from './middleware/rateLimiter.js';
 import adminRouter from './routes/admin.js';
 import { reconciliationService } from './services/reconcile.js';
+import { IndexerHealthService } from './services/indexerHealth.js';
 
 dotenv.config();
 validateAuthEnvironment();
@@ -27,6 +28,11 @@ if (process.env.NODE_ENV === 'production' && !process.env.CORS_ORIGIN) {
 const app: Express = express();
 const port = process.env.PORT || 3001;
 const startTime = Date.now();
+const indexerHealthService = new IndexerHealthService(
+  process.env.CONTRACT_ID || '',
+  undefined,
+  process.env.SOROBAN_RPC_URL || ''
+);
 
 app.use(helmet());
 app.use(
@@ -42,14 +48,16 @@ app.get('/', (_req: Request, res: Response) => {
   res.json({ message: 'Welcome to PaymeshStellar Backend API' });
 });
 
-app.get('/health', (_req: Request, res: Response) => {
+app.get('/health', asyncHandler(async (_req: Request, res: Response) => {
+  const indexer = await indexerHealthService.getHealth();
   res.json({
     status: 'ok',
     uptime: Math.floor((Date.now() - startTime) / 1000),
     version: process.env.npm_package_version ?? '0.1.0',
     reconciliation: reconciliationService.getHealth(),
+    indexer,
   });
-});
+}));
 
 if (process.env.NODE_ENV === 'test') {
   app.get(