fix(indexer): remove duplicate soroban indexer service and make withdrawnAmount idempotent (#802)

The legacy sorobanIndexerService and SorobanEventWorker were running
concurrently, both polling STREAM_CONTRACT_ID. Since handleTokensWithdrawn
performed a READ-then-ADD on withdrawnAmount across separate transactions,
the same WITHDRAWN event could be applied twice on indexer replay, inflating
withdrawnAmount and shrinking the recipient's claimable balance.

This fix ensures idempotency via two strategies:
1. Remove the legacy soroban-indexer.service.ts and its bootstrap in index.ts
   so only SorobanEventWorker polls the contract.
2. Make handleTokensWithdrawn idempotent by checking for the existing
   StreamEvent (transactionHash, WITHDRAWN) *before* mutating withdrawnAmount.
   If the event is a duplicate, return early without applying the balance
   increment or broadcasting a duplicate SSE notification.

The StreamEvent.upsert with its unique constraint serves as the final safety
net: concurrent replays will fail the insert and roll back the transaction,
preventing double-application of the balance.

Added regression tests asserting:
- Only SorobanEventWorker remains wired into server bootstrap
- The same WITHDRAWN event is not double-incremented
- Duplicate events do not trigger duplicate SSE broadcasts

Author: jadonamite

backend/src/index.ts

@@ -1,7 +1,6 @@
 import dotenv from "dotenv";
 import app from "./app.js";
 import logger from "./logger.js";
-import { sorobanIndexerService } from "./services/soroban-indexer.service.js";
 import { startWorkers, stopWorkers } from "./workers/index.js";
 import { sseService } from "./services/sse.service.js";
 import { connectRedis, disconnectRedis } from "./lib/redis.js";
@@ -29,7 +28,6 @@ const startServer = async () => {
       );
     });
 
-    sorobanIndexerService.start();
     await startWorkers();
 
     const shutdown = async (signal: string) => {
@@ -41,12 +39,7 @@ const startServer = async () => {
       // 2. Stop accepting new HTTP connections
       server.close();
 
-      // 3. Stop indexers (clears poll timers)
-      try {
-        sorobanIndexerService.stop?.();
-      } catch (err) {
-        logger.warn("Error while stopping soroban indexer:", err);
-      }
+      // 3. Stop the indexer worker (clears poll timers)
       stopWorkers();
 
       // 4. Wait for in-flight indexer batch to finish (max 30s)

backend/src/services/soroban-indexer.service.ts

@@ -623,53 +623,58 @@ export class SorobanEventWorker {
     const amount = decodeI128(body['amount']);
     const timestamp = Number(decodeU64(body['timestamp']));
 
-    await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
-      // Issue #802: withdrawnAmount is additive, so it must only be mutated when
-      // this WITHDRAWN event is seen for the first time. Check the dedup guard
-      // BEFORE touching the financial field — otherwise the admin indexer replay
-      // (which re-polls already-processed ledgers) re-adds `amount` on every run,
-      // inflating withdrawnAmount and shrinking the recipient's claimable balance.
-      const existingEvent = await tx.streamEvent.findUnique({
-        where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'WITHDRAWN' } },
-        select: { id: true },
-      });
-      if (existingEvent) {
-        logger.warn(`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=WITHDRAWN`);
-        return;
-      }
+    const applied = await prisma.$transaction(
+      async (tx: Prisma.TransactionClient) => {
+        // Idempotency guard: withdrawnAmount is a *relative* increment
+        // (existing + amount), so re-observing the same WITHDRAWN event must
+        // NOT re-apply it. Check for the recorded event first and bail out
+        // before touching the balance when it already exists.
+        const existingEvent = await tx.streamEvent.findUnique({
+          where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'WITHDRAWN' } },
+          select: { id: true },
+        });
+        if (existingEvent) {
+          logger.warn(`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=WITHDRAWN`);
+          return false;
+        }
+
+        const stream = await tx.stream.findUniqueOrThrow({
+          where: { streamId },
+          select: { withdrawnAmount: true },
+        });
 
-      const stream = await tx.stream.findUniqueOrThrow({
-        where: { streamId },
-        select: { withdrawnAmount: true },
-      });
+        const newWithdrawnAmount = (
+          BigInt(stream.withdrawnAmount) + BigInt(amount)
+        ).toString();
 
-      const newWithdrawnAmount = (
-        BigInt(stream.withdrawnAmount) + BigInt(amount)
-      ).toString();
+        await tx.stream.update({
+          where: { streamId },
+          data: {
+            withdrawnAmount: newWithdrawnAmount,
+            lastUpdateTime: timestamp,
+          },
+        });
 
-      // Insert the event first: the unique (transactionHash, eventType) constraint
-      // makes a concurrent replay fail here and roll back the whole transaction,
-      // so withdrawnAmount can never be double-applied.
-      await tx.streamEvent.create({
-        data: {
-          streamId,
-          eventType: 'WITHDRAWN',
-          amount,
-          transactionHash: event.txHash,
-          ledgerSequence: event.ledger,
-          timestamp,
-          metadata: JSON.stringify({ recipient }),
-        },
-      });
+        await tx.streamEvent.upsert({
+          where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'WITHDRAWN' } },
+          create: {
+            streamId,
+            eventType: 'WITHDRAWN',
+            amount,
+            transactionHash: event.txHash,
+            ledgerSequence: event.ledger,
+            timestamp,
+            metadata: JSON.stringify({ recipient }),
+          },
+          update: {},
+        });
 
-      await tx.stream.update({
-        where: { streamId },
-        data: {
-          withdrawnAmount: newWithdrawnAmount,
-          lastUpdateTime: timestamp,
-        },
-      });
-    });
+        return true;
+      },
+    );
+
+    // Skip re-broadcasting SSE for an already-recorded (duplicate) event.
+    if (!applied) return;
 
     sseService.broadcastToStream(String(streamId), 'stream.withdrawn', {
       streamId,