LabsCrypt · jadonamite · Jul 4, 2026
diff --git a/backend/src/services/soroban-indexer.service.ts b/backend/src/services/soroban-indexer.service.ts
diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts
@@ -820,64 +820,73 @@ export class SorobanEventWorker {
     const amount = decodeI128(body["amount"]);
     const timestamp = Number(decodeU64(body["timestamp"]));
 
-    await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
-      // Check for a duplicate BEFORE mutating any Stream fields so that a
-      // replayed event never double-increments withdrawnAmount.
-      const existingEvent = await tx.streamEvent.findUnique({
-        where: {
-          transactionHash_eventType: {
-            transactionHash: event.txHash,
-            eventType: "WITHDRAWN",
+    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;
-      }
+          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,
-        },
-      });
+        await tx.stream.update({
+          where: { streamId },
+          data: {
+            withdrawnAmount: newWithdrawnAmount,
+            lastUpdateTime: timestamp,
+          },
+        });
 
-      await tx.streamEvent.upsert({
-        where: {
-          transactionHash_eventType: {
-            transactionHash: event.txHash,
+        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 }),
           },
-        },
-        create: {
-          streamId,
-          eventType: "WITHDRAWN",
-          amount,
-          transactionHash: event.txHash,
-          ledgerSequence: event.ledger,
-          timestamp,
-          metadata: JSON.stringify({ recipient }),
-        },
-        update: {
-          ledgerSequence: event.ledger,
-          timestamp,
-        },
-      });
-    });
+          update: {
+            ledgerSequence: event.ledger,
+            timestamp,
+          },
+        });
+
+        return true;
+      },
+    );
+
+    // Skip re-broadcasting SSE for an already-recorded (duplicate) event.
+    if (!applied) return;
 
     sseService.broadcastToStream(String(streamId), "stream.withdrawn", {
       streamId,

diff --git a/backend/tests/single-indexer.regression.test.ts b/backend/tests/single-indexer.regression.test.ts
@@ -0,0 +1,159 @@
+/**
+ * Regression tests for issue #801 — "Two indexers run concurrently".
+ *
+ * Previously both `sorobanIndexerService` (services/soroban-indexer.service.ts)
+ * and `SorobanEventWorker` (workers/soroban-event-worker.ts) polled
+ * STREAM_CONTRACT_ID and wrote Stream/StreamEvent rows. Their WITHDRAWN
+ * handlers each did a READ-then-ADD on `withdrawnAmount` in separate
+ * transactions, so the same event could be applied twice → inflated balance.
+ *
+ * These tests lock in the fix:
+ *   1. Only ONE indexer (the worker) polls the contract — the legacy service
+ *      is deleted and is no longer wired into the server entry-point.
+ *   2. Observing the same WITHDRAWN event twice increments `withdrawnAmount`
+ *      exactly once.
+ */
+import { readFileSync, existsSync } from 'node:fs';
+import { fileURLToPath } from 'node:url';
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+import { rpc } from '@stellar/stellar-sdk';
+
+// ─── Mocks (must be registered before importing the worker) ──────────────────
+
+vi.mock('../src/lib/prisma.js', () => ({
+  default: { indexerState: { upsert: vi.fn() } },
+  prisma: {
+    indexerState: { upsert: vi.fn() },
+    stream: { findUniqueOrThrow: vi.fn(), update: vi.fn() },
+    streamEvent: { findUnique: vi.fn(), upsert: vi.fn() },
+    $transaction: vi.fn(),
+  },
+}));
+
+vi.mock('../src/services/sse.service.js', () => ({
+  sseService: { broadcastToStream: vi.fn() },
+}));
+
+vi.mock('../src/logger.js', () => ({
+  default: { info: vi.fn(), warn: vi.fn(), error: vi.fn() },
+}));
+
+import { SorobanEventWorker } from '../src/workers/soroban-event-worker.js';
+import { prisma } from '../src/lib/prisma.js';
+import { sseService } from '../src/services/sse.service.js';
+import logger from '../src/logger.js';
+
+const srcUrl = (rel: string) => fileURLToPath(new URL(rel, import.meta.url));
+
+describe('#801 single indexer', () => {
+  describe('only one indexer instance polls the contract', () => {
+    it('deletes the legacy soroban-indexer.service module', () => {
+      expect(existsSync(srcUrl('../src/services/soroban-indexer.service.ts'))).toBe(false);
+    });
+
+    it('does not wire the legacy indexer service into the server entry-point', () => {
+      const indexSrc = readFileSync(srcUrl('../src/index.ts'), 'utf8');
+      expect(indexSrc).not.toMatch(/soroban-indexer\.service/);
+      expect(indexSrc).not.toMatch(/sorobanIndexerService/);
+      // The worker remains the single indexer started at boot.
+      expect(indexSrc).toMatch(/startWorkers\(\)/);
+    });
+
+    it('exposes exactly one contract poller — the SorobanEventWorker', () => {
+      // startWorkers() is the only indexer bootstrap; it starts the worker and
+      // nothing else. (See workers/index.ts.)
+      const workersSrc = readFileSync(srcUrl('../src/workers/index.ts'), 'utf8');
+      expect(workersSrc).toMatch(/sorobanEventWorker\.start\(\)/);
+      expect(workersSrc).not.toMatch(/sorobanIndexerService/);
+    });
+  });
+
+  describe('withdrawnAmount is not double-incremented on a repeated WITHDRAWN event', () => {
+    let worker: SorobanEventWorker;
+
+    const mockSym = (name: string) => ({ sym: { toString: () => name } } as any);
+    const mockU64 = (value: number | bigint | string) => ({ u64: { toString: () => String(value) } } as any);
+    const mockI128 = (hi: number | bigint | string, lo: number | bigint | string) => ({ i128: { hi: { toString: () => String(hi) }, lo: { toString: () => String(lo) } } } as any);
+    const mockAccountAddr = () => ({ address: { type: 'scAddressTypeAccount', accountId: { ed25519: { value: Buffer.alloc(32) } } } } as any);
+    const mockMapEntry = (keyName: string, val: any) => ({ key: mockSym(keyName), val } as any);
+
+    const streamId = 7;
+    const buildEvent = (): rpc.Api.EventResponse =>
+      ({
+        id: 'withdraw-event-1',
+        type: 'contract',
+        ledger: 4000,
+        ledgerClosedAt: '2024-01-01T00:00:00Z',
+        txHash: 'withdraw-tx-hash',
+        transactionIndex: 0,
+        operationIndex: 0,
+        inSuccessfulContractCall: true,
+        topic: [
+          mockSym('tokens_withdrawn'),
+          mockU64(streamId),
+        ],
+        value: {
+          map: [
+            mockMapEntry('recipient', mockAccountAddr()),
+            mockMapEntry('amount', mockI128('0', '100')),
+            mockMapEntry('timestamp', mockU64('1700005000')),
+          ] as any,
+        } as any,
+      }) as rpc.Api.EventResponse;
+
+    beforeEach(() => {
+      vi.clearAllMocks();
+      worker = new SorobanEventWorker();
+    });
+
+    it('applies the increment once and skips it when the event is seen again', async () => {
+      // A tiny stateful stand-in for the DB so the second observation genuinely
+      // sees the recorded event (as it would in production).
+      const db = {
+        withdrawnAmount: '0',
+        recordedEvent: null as { id: string } | null,
+      };
+
+      const mockTx = {
+        streamEvent: {
+          findUnique: vi.fn(async () => db.recordedEvent),
+          upsert: vi.fn(async () => {
+            db.recordedEvent = { id: 'withdraw-event-row' };
+            return db.recordedEvent;
+          }),
+        },
+        stream: {
+          findUniqueOrThrow: vi.fn(async () => ({ withdrawnAmount: db.withdrawnAmount })),
+          update: vi.fn(async ({ data }: { data: { withdrawnAmount: string } }) => {
+            db.withdrawnAmount = data.withdrawnAmount;
+            return {};
+          }),
+        },
+      };
+
+      (prisma.$transaction as ReturnType<typeof vi.fn>).mockImplementation(
+        (cb: (tx: typeof mockTx) => unknown) => cb(mockTx),
+      );
+
+      const event = buildEvent();
+
+      // First observation: increment 0 → 100 and record the event.
+      await (worker as any).handleTokensWithdrawn(event, event.topic![1]);
+      expect(db.withdrawnAmount).toBe('100');
+      expect(mockTx.stream.update).toHaveBeenCalledTimes(1);
+      expect(mockTx.streamEvent.upsert).toHaveBeenCalledTimes(1);
+      expect(sseService.broadcastToStream).toHaveBeenCalledTimes(1);
+
+      // Second observation of the SAME event: must be a no-op for the balance.
+      await (worker as any).handleTokensWithdrawn(event, event.topic![1]);
+      expect(db.withdrawnAmount).toBe('100'); // NOT '200'
+      expect(mockTx.stream.update).toHaveBeenCalledTimes(1); // still once
+      expect(mockTx.streamEvent.upsert).toHaveBeenCalledTimes(1); // still once
+      // No duplicate SSE notification for the repeated event.
+      expect(sseService.broadcastToStream).toHaveBeenCalledTimes(1);
+      expect(logger.warn).toHaveBeenCalledWith(
+        expect.stringContaining('Duplicate StreamEvent skipped'),
+      );
+    });
+  });
+});