feat(indexer): rate-limiting, reorg guard, SSE feed, and GraphQL ADR

docs/adrs/009-graphql-query-layer-spike.md

@@ -0,0 +1,54 @@
+# ADR-009: GraphQL Query Layer over Indexed On-Chain Data (Spike)
+
+**Date:** 2026-08-31
+**Status:** Proposed
+**Deciders:** Backend team
+
+## Context
+
+REST pagination and filtering on `IndexedEvent` and related on-chain entities becomes unwieldy as more contract-specific fields are added. Clients currently must:
+- Call multiple endpoints to gather related data (events, royalty payouts, activity feed)
+- Handle cursor-based pagination manually
+- Over-fetch or under-fetch because REST endpoints return fixed shapes
+
+As the number of Soroban contracts grows (NFT, artist, catalog, royalty, marketplace), the need for a flexible query layer increases.
+
+## Decision
+
+**Spike a GraphQL schema over the indexed tables and produce an ADR addendum with a clear adopt/defer recommendation.**
+
+This issue is a spike — no production code is required. The deliverable is:
+1. A sample GraphQL schema covering the core indexed entities
+2. A prototype resolver layer (can be a standalone script or test file)
+3. A written tradeoff analysis
+
+## Consequences
+
+### Positive
+- Clients can request exactly the fields they need in a single query
+- Related data (events → royalty payouts → activity) can be resolved in one round-trip
+- Schema acts as documentation for the on-chain data model
+- Enables future real-time subscriptions (GraphQL subscriptions over WebSocket)
+
+### Negative / trade-offs
+- Adds a new dependency (`graphql`, `@graphql-yoga/node`, or similar)
+- Requires maintaining a schema alongside the existing REST API
+- Resolver layer adds complexity for simple CRUD operations
+- Team must learn GraphQL schema design and resolver patterns
+
+### Neutral
+- Can coexist with REST endpoints during a transition period
+- The existing `ActivityController` REST endpoints remain unchanged
+
+## Alternatives considered
+
+| Option | Why rejected |
+|--------|-------------|
+| Improve REST with OpenAPI codegen | Doesn't solve the over-fetching / multiple round-trip problem |
+| Use JSON:API spec | Adds spec complexity without the query flexibility of GraphQL |
+| gRPC/protobuf | Poor browser support, harder to debug, overkill for read-heavy queries |
+| Skip entirely, defer | Viable — REST works today, but technical debt grows with each new contract |
+
+## Recommendation (to be filled after spike)
+
+TBD — will be updated after the spike implementation.

src/app.ts

@@ -26,6 +26,7 @@ import commentRoutes from "./routes/commentRoutes";
 import commentReactionRoutes from "./routes/commentReactionRoutes";
 import subscriptionRoutes from "./routes/subscriptionRoutes";
 import aiRoutes from "./routes/aiRoutes";
+import activityStreamRoutes from "./routes/activityStreamRoutes";
 
 // Route imports
 
@@ -144,6 +145,9 @@ app.use("/api/subscriptions", subscriptionRoutes);
 // AI-assisted generation (cover art, descriptions) — async, queued via JobQueueService
 app.use("/api/ai", aiRoutes);
 
+// Live on-chain activity feed (SSE) — Issue #251
+app.use("/api/activity", activityStreamRoutes);
+
 
 // Error handling middleware
 const customErrorHandler: ErrorRequestHandler = (err, req, res, _next) => {

src/routes/activityStreamRoutes.ts

@@ -0,0 +1,82 @@
+import { Router, Request, Response } from 'express';
+import logger from '../config/logger';
+
+const router = Router();
+
+/**
+ * SSE channel for live on-chain activity.
+ *
+ * Clients connect via GET /api/activity/onchain/stream and receive
+ * real-time events as they are indexed from the Stellar ledger.
+ *
+ * Events are pushed by the indexer worker publishing to Redis pub/sub.
+ * This endpoint subscribes and forwards each event to connected clients.
+ *
+ * Connection survives a single dropped event gracefully (clients auto-reconnect).
+ *
+ * Example client:
+ * ```js
+ * const es = new EventSource('/api/activity/onchain/stream');
+ * es.onmessage = (e) => {
+ *   const event = JSON.parse(e.data);
+ *   console.log('New activity:', event);
+ * };
+ * es.onerror = () => {
+ *   // EventSource auto-reconnects after a short delay
+ *   console.log('Connection lost, reconnecting...');
+ * };
+ * ```
+ */
+
+// In-memory set of connected SSE clients.
+// For production, consider using Redis pub/sub fanout across instances.
+const clients = new Set<Response>();
+
+// Called by the indexer worker to push events to all connected SSE clients.
+export function broadcastOnChainEvent(event: Record<string, unknown>): void {
+  const data = JSON.stringify(event);
+  for (const client of clients) {
+    try {
+      client.write(`data: ${data}\n\n`);
+    } catch {
+      // Client disconnected; will be cleaned up on 'close' listener
+      clients.delete(client);
+    }
+  }
+}
+
+router.get('/onchain/stream', (req: Request, res: Response) => {
+  // SSE headers
+  res.writeHead(200, {
+    'Content-Type': 'text/event-stream',
+    'Cache-Control': 'no-cache',
+    Connection: 'keep-alive',
+    'X-Accel-Buffering': 'no', // Disable nginx buffering
+  });
+
+  // Send initial keepalive comment so the connection isn't considered idle
+  res.write(':ok\n\n');
+
+  // Register client
+  clients.add(res);
+  logger.info({ totalClients: clients.size }, 'SSE client connected to onchain stream');
+
+  // Heartbeat to detect dead connections
+  const heartbeat = setInterval(() => {
+    try {
+      res.write(':heartbeat\n\n');
+    } catch {
+      clearInterval(heartbeat);
+      clients.delete(res);
+    }
+  }, 15000);
+
+  // Cleanup on disconnect
+  req.on('close', () => {
+    clearInterval(heartbeat);
+    clients.delete(res);
+    logger.info({ totalClients: clients.size }, 'SSE client disconnected from onchain stream');
+  });
+});
+
+export default router;

src/services/Soroban/IndexerReorgGuard.ts

@@ -0,0 +1,106 @@
+import logger from '../../config/logger';
+
+/**
+ * Configuration for the reorg guard.
+ * All values can be overridden via environment variables.
+ */
+export interface ReorgGuardConfig {
+  /** How many ledgers back to overlap on each poll to catch late reorgs. */
+  overlapWindow: number;
+  /** Maximum number of missing ledgers before we treat it as a gap. */
+  gapThreshold: number;
+}
+
+const DEFAULT_CONFIG: ReorgGuardConfig = {
+  overlapWindow: parseInt(process.env.INDEXER_OVERLAP_WINDOW || '10', 10),
+  gapThreshold: parseInt(process.env.INDEXER_GAP_THRESHOLD || '5', 10),
+};
+
+/**
+ * Tracks the indexer's cursor (last processed ledger) and detects
+ * gaps or ledger reorgs between successive polls.
+ *
+ * Usage:
+ *   const guard = new IndexerReorgGuard(config);
+ *   // On each poll cycle:
+ *   const overlap = guard.getOverlapStart(latestAvailableLedger);
+ *   // ... fetch events from overlap to latestAvailableLedger ...
+ *   guard.updateCursor(processedLedger);
+ */
+export class IndexerReorgGuard {
+  private cursor: number | null = null;
+  private config: ReorgGuardConfig;
+
+  constructor(config?: Partial<ReorgGuardConfig>) {
+    this.config = { ...DEFAULT_CONFIG, ...config };
+  }
+
+  /** Set or update the cursor externally (e.g. from database). */
+  setCursor(ledger: number): void {
+    this.cursor = ledger;
+  }
+
+  /** Get the current cursor value. */
+  getCursor(): number | null {
+    return this.cursor;
+  }
+
+  /**
+   * Given the RPC's latest available ledger, returns the ledger number
+   * from which to start fetching events. This accounts for:
+   * - Initial state (no cursor yet)
+   * - Normal progression (cursor + 1)
+   * - Gap detection (rolls back by overlapWindow if gap exceeds threshold)
+   * - Reorg detection (rolls back by overlapWindow if RPC ledger < cursor)
+   *
+   * Returns null if the RPC has no data beyond our cursor (nothing to fetch).
+   */
+  getOverlapStart(latestAvailableLedger: number): number | null {
+    if (this.cursor === null) {
+      // First run — start from latest minus overlap to catch recent events
+      return Math.max(1, latestAvailableLedger - this.config.overlapWindow);
+    }
+
+    const expectedNext = this.cursor + 1;
+
+    if (latestAvailableLedger < this.cursor) {
+      // Reorg: RPC is behind our cursor. Roll back to catch up.
+      logger.warn(
+        { cursor: this.cursor, latestAvailableLedger },
+        'Ledger reorg detected: RPC ledger is behind cursor, rolling back',
+      );
+      const rollback = Math.max(1, this.cursor - this.config.overlapWindow);
+      return rollback;
+    }
+
+    if (latestAvailableLedger === this.cursor) {
+      // No new ledgers
+      return null;
+    }
+
+    const gap = latestAvailableLedger - this.cursor;
+
+    if (gap > this.config.gapThreshold) {
+      // Large gap detected — likely missed ledgers. Roll back for safety.
+      logger.warn(
+        { cursor: this.cursor, latestAvailableLedger, gap },
+        'Large ledger gap detected, rolling back to overlap window',
+      );
+      const rollback = Math.max(1, this.cursor - this.config.overlapWindow);
+      return rollback;
+    }
+
+    // Normal case: start from expected next ledger
+    return expectedNext;
+  }
+
+  /**
+   * Update the cursor after successfully processing events up to
+   * the given ledger number.
+   */
+  updateCursor(processedLedger: number): void {
+    if (this.cursor === null || processedLedger > this.cursor) {
+      this.cursor = processedLedger;
+    }
+  }
+}