feat: sync Soroban contract events to database closes #179

backend/.env.example

@@ -15,3 +15,17 @@ SANDBOX_MODE_ENABLED=true
 # Optional: Use a separate database for sandbox
 # If not set, it will use {DATABASE_URL}_sandbox
 SANDBOX_DATABASE_URL="postgresql://user:password@localhost:5432/flowfi_sandbox?schema=public"
+
+# ─── Soroban Event Indexer ────────────────────────────────────────────────────
+# Soroban RPC endpoint (testnet default shown)
+SOROBAN_RPC_URL=https://soroban-testnet.stellar.org
+
+# Deployed FlowFi stream contract address (C...)
+# Leave empty to disable the event indexer
+STREAM_CONTRACT_ID=
+
+# How often the indexer polls for new events (milliseconds, default: 5000)
+INDEXER_POLL_INTERVAL_MS=5000
+
+# Ledger sequence to start indexing from on first run (0 = latest)
+INDEXER_START_LEDGER=0

backend/package.json

@@ -23,6 +23,7 @@
   "license": "ISC",
   "dependencies": {
     "@prisma/adapter-pg": "^7.4.1",
+    "@stellar/stellar-sdk": "^13.1.0",
     "cors": "^2.8.6",
     "dotenv": "^17.3.1",
     "express": "^5.2.1",

backend/prisma/migrations/20260225000000_add_indexer_state/migration.sql

@@ -0,0 +1,9 @@
+-- CreateTable
+CREATE TABLE "IndexerState" (
+    "id" TEXT NOT NULL DEFAULT 'singleton',
+    "lastLedger" INTEGER NOT NULL DEFAULT 0,
+    "lastCursor" TEXT,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "IndexerState_pkey" PRIMARY KEY ("id")
+);

backend/prisma/schema.prisma

@@ -51,6 +51,14 @@ model Stream {
   @@index([isActive])
 }
 
+// IndexerState model - tracks the last processed ledger/cursor for the Soroban event worker
+model IndexerState {
+  id          String   @id @default("singleton")
+  lastLedger  Int      @default(0)
+  lastCursor  String?  // Paging token of the last processed event (for cursor-based pagination)
+  updatedAt   DateTime @updatedAt
+}
+
 // StreamEvent model - indexer events for tracking all on-chain stream activities
 model StreamEvent {
   id              String   @id @default(uuid())

backend/src/index.ts

@@ -1,6 +1,7 @@
 import dotenv from 'dotenv';
 import app from './app.js';
 import logger from './logger.js';
+import { startWorkers, stopWorkers } from './workers/index.js';
 
 dotenv.config();
 
@@ -13,10 +14,26 @@ const startServer = async () => {
         logger.info('Database connection established successfully');
 
         const port = process.env.PORT || 3001;
-        app.listen(port, () => {
+        const server = app.listen(port, () => {
             logger.info(`Server started on port ${port}`);
             logger.info(`API Documentation available at http://localhost:${port}/api-docs`);
         });
+
+        // Start background workers after the HTTP server is up.
+        await startWorkers();
+
+        // Graceful shutdown: stop workers before closing the HTTP server.
+        const shutdown = (signal: string) => {
+            logger.info(`Received ${signal}. Shutting down gracefully...`);
+            stopWorkers();
+            server.close(() => {
+                logger.info('HTTP server closed.');
+                process.exit(0);
+            });
+        };
+
+        process.on('SIGTERM', () => shutdown('SIGTERM'));
+        process.on('SIGINT', () => shutdown('SIGINT'));
     } catch (error) {
         logger.error('Failed to start server due to database connection error:', error);
         process.exit(1);

backend/src/workers/index.ts

@@ -0,0 +1,18 @@
+/**
+ * Workers registry
+ *
+ * Exports a single `startWorkers` function that is called from the main server
+ * entry-point after the database connection is confirmed healthy.
+ */
+
+import { sorobanEventWorker } from './soroban-event-worker.js';
+import logger from '../logger.js';
+
+export async function startWorkers(): Promise<void> {
+  logger.info('[Workers] Starting background workers...');
+  await sorobanEventWorker.start();
+}
+
+export function stopWorkers(): void {
+  sorobanEventWorker.stop();
+}