fix(indexer): remove duplicate soroban indexer service (#801)

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/workers/soroban-event-worker.ts

@@ -623,31 +623,38 @@ export class SorobanEventWorker {
     const amount = decodeI128(body['amount']);
     const timestamp = Number(decodeU64(body['timestamp']));
 
-    await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
-      const stream = await tx.stream.findUniqueOrThrow({
-        where: { streamId },
-        select: { withdrawnAmount: true },
-      });
+    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 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,
-        },
-      });
+        await tx.stream.update({
+          where: { streamId },
+          data: {
+            withdrawnAmount: newWithdrawnAmount,
+            lastUpdateTime: timestamp,
+          },
+        });
 
-      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`);
-      } else {
         await tx.streamEvent.upsert({
           where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'WITHDRAWN' } },
           create: {
@@ -661,8 +668,13 @@ export class SorobanEventWorker {
           },
           update: {},
         });
-      }
-    });
+
+        return true;
+      },
+    );
+
+    // Skip re-broadcasting SSE for an already-recorded (duplicate) event.
+    if (!applied) return;
 
     sseService.broadcastToStream(String(streamId), 'stream.withdrawn', {
       streamId,