diff --git a/backend/src/index.ts b/backend/src/index.ts index f31f9d42..d7182fda 100644 --- a/backend/src/index.ts +++ b/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) => { @@ -42,11 +40,6 @@ const startServer = async () => { server.close(); // 3. Stop indexers (clears poll timers) - try { - sorobanIndexerService.stop?.(); - } catch (err) { - logger.warn("Error while stopping soroban indexer:", err); - } stopWorkers(); // 4. Wait for in-flight indexer batch to finish (max 30s) diff --git a/backend/src/services/indexerService.ts b/backend/src/services/indexerService.ts index 1759d75f..dc145289 100644 --- a/backend/src/services/indexerService.ts +++ b/backend/src/services/indexerService.ts @@ -5,14 +5,9 @@ * indexer. It is the admin/control-plane helper for the source-of-truth * indexer, `SorobanEventWorker` (backend/src/workers/soroban-event-worker.ts). * The functions here only read/reset the shared `IndexerState` cursor row and - * trigger the worker's poll loop. It is intentionally named like the legacy - * indexer below to document that this helper is the "other" indexer entry - * point — see backend/src/services/soroban-indexer.service.ts, which is the - * LEGACY indexer being phased out. See docs/ARCHITECTURE.md for the full - * indexer ownership model. + * trigger the worker's poll loop. * - * NAMING CONVENTION PLAN: once the functional consolidation of the two - * indexers lands (issue #801), this file is expected to be renamed to + * NAMING CONVENTION PLAN: this file is expected to be renamed to * `indexer.service.ts` so every service is kebab-case with a `.service.ts` * suffix. */ diff --git a/backend/src/services/soroban-indexer.service.ts b/backend/src/services/soroban-indexer.service.ts deleted file mode 100644 index 4be8516a..00000000 --- a/backend/src/services/soroban-indexer.service.ts +++ /dev/null @@ -1,281 +0,0 @@ -/** - * LEGACY indexer — being phased out. - * - * The source of truth for event indexing is `SorobanEventWorker` - * (backend/src/workers/soroban-event-worker.ts). That worker handles the full - * event surface (created / topped_up / withdrawn / paused / resumed / - * cancelled / completed / fee_collected / fee_config_updated / - * admin_transferred), uses cursor-based pagination, persists the `IndexerState` - * cursor, and broadcasts SSE updates. - * - * This service is a simpler, second indexer that polls the Soroban RPC on its - * own interval and writes to the same rows as the worker, which means both run - * concurrently and can race on the same Stream / StreamEvent rows (see issue - * #801). It is kept only for backwards compatibility and is being phased out. - * Until the functional consolidation (issue #801) lands, do not extend this - * service with new behavior — mirror any changes in `SorobanEventWorker` - * instead. Once consolidation lands, this file is expected to be removed. - * - * NAMING CONVENTION PLAN: this file already follows the kebab-case `.service.ts` - * convention. The helper file backend/src/services/indexerService.ts (which is - * NOT an indexer) is expected to be renamed to `indexer.service.ts` to match. - */ -import { prisma } from '../lib/prisma.js'; -import logger from '../logger.js'; -import { withRpcRetry, withRpcTimeout } from './sorobanService.js'; - -type JsonRecord = Record; - -interface RpcEvent { - id?: string; - ledger?: number; - ledgerSequence?: number; - txHash?: string; - topic?: unknown[]; - value?: unknown; - contractId?: string; -} - -interface RpcResponse { - result?: { - events?: RpcEvent[]; - }; - error?: { - message?: string; - }; -} - -type IndexedEventType = 'CREATED' | 'CANCELLED' | 'WITHDRAWN' | 'COMPLETED'; - -const RPC_URL = process.env.SOROBAN_RPC_URL ?? 'https://soroban-testnet.stellar.org'; -const POLL_MS = Number(process.env.SOROBAN_INDEXER_POLL_MS ?? 15000); -const START_LEDGER = Number(process.env.SOROBAN_INDEXER_START_LEDGER ?? 0); -const STREAM_CONTRACT_ID = process.env.STREAM_CONTRACT_ID ?? ''; - -export class SorobanIndexerService { - private timer: NodeJS.Timeout | null = null; - private running = false; - private lastLedger = START_LEDGER; - - start() { - if (this.running) return; - this.running = true; - - void this.poll(); - this.timer = setInterval(() => { - void this.poll(); - }, POLL_MS); - - logger.info(`Soroban indexer started (poll=${POLL_MS}ms, startLedger=${this.lastLedger})`); - } - - stop() { - if (this.timer) clearInterval(this.timer); - this.timer = null; - this.running = false; - } - - private async poll() { - if (!STREAM_CONTRACT_ID) return; - - try { - const events = await this.fetchEvents(this.lastLedger + 1); - if (events.length === 0) return; - - let maxLedger = this.lastLedger; - for (const event of events) { - const ledger = Number(event.ledgerSequence ?? event.ledger ?? 0); - if (ledger > maxLedger) maxLedger = ledger; - await this.indexEvent(event, ledger); - } - - this.lastLedger = maxLedger; - } catch (error) { - logger.error('Soroban indexer poll failed', error); - } - } - - private async fetchEvents(startLedger: number): Promise { - const body = { - jsonrpc: '2.0', - id: 1, - method: 'getEvents', - params: { - startLedger, - filters: [{ type: 'contract', contractIds: [STREAM_CONTRACT_ID] }], - pagination: { limit: 100 }, - }, - }; - - const response = await withRpcRetry('getEvents', () => - withRpcTimeout('getEvents', (signal) => - fetch(RPC_URL, { - method: 'POST', - headers: { 'content-type': 'application/json' }, - body: JSON.stringify(body), - signal, - }), - ), - ); - - if (!response.ok) { - throw new Error(`getEvents failed: ${response.status}`); - } - - const payload = (await response.json()) as RpcResponse; - if (payload.error?.message) throw new Error(payload.error.message); - return payload.result?.events ?? []; - } - - private asRecord(value: unknown): JsonRecord | null { - if (!value || typeof value !== 'object' || Array.isArray(value)) return null; - return value as JsonRecord; - } - - private parseEventType(event: RpcEvent): IndexedEventType | null { - const firstTopic = Array.isArray(event.topic) && event.topic.length > 0 - ? String(event.topic[0]).toLowerCase() - : ''; - - if (firstTopic.includes('stream_created')) return 'CREATED'; - if (firstTopic.includes('stream_cancelled')) return 'CANCELLED'; - if (firstTopic.includes('tokens_withdrawn')) return 'WITHDRAWN'; - if (firstTopic.includes('stream_completed')) return 'COMPLETED'; - return null; - } - - private parseStreamId(record: JsonRecord): bigint | null { - const raw = record.stream_id ?? record.streamId; - if (typeof raw === 'bigint' && raw >= 0n) return raw; - if (typeof raw === 'number' && Number.isInteger(raw) && raw >= 0 && Number.isSafeInteger(raw)) { - return BigInt(raw); - } - if (typeof raw === 'string' && /^\d+$/.test(raw.trim())) { - try { - return BigInt(raw.trim()); - } catch { - return null; - } - } - return null; - } - - private readString(record: JsonRecord, ...keys: string[]): string | null { - for (const key of keys) { - const value = record[key]; - if (typeof value === 'string' && value.trim()) return value; - } - return null; - } - - private async ensureUser(publicKey: string) { - await prisma.user.upsert({ - where: { publicKey }, - update: {}, - create: { publicKey }, - }); - } - - private async indexEvent(event: RpcEvent, ledgerSequence: number) { - const eventType = this.parseEventType(event); - if (!eventType) return; - - const value = this.asRecord(event.value); - if (!value) return; - - const streamId = this.parseStreamId(value); - if (!streamId) return; - - const txHash = event.txHash ?? event.id ?? `event-${streamId}-${ledgerSequence}-${eventType}`; - const timestamp = Math.floor(Date.now() / 1000); - - const existing = await prisma.streamEvent.findFirst({ - where: { - streamId, - eventType, - transactionHash: txHash, - ledgerSequence, - }, - select: { id: true }, - }); - if (existing) return; - - if (eventType === 'CREATED') { - const sender = this.readString(value, 'sender'); - const recipient = this.readString(value, 'recipient'); - const tokenAddress = this.readString(value, 'token_address', 'tokenAddress'); - const ratePerSecond = this.readString(value, 'rate_per_second', 'ratePerSecond'); - const depositedAmount = this.readString(value, 'deposited_amount', 'depositedAmount'); - const startTimeStr = this.readString(value, 'start_time', 'startTime') ?? String(timestamp); - const startTime = BigInt(startTimeStr); - - if (!sender || !recipient || !tokenAddress || !ratePerSecond || !depositedAmount) return; - - await this.ensureUser(sender); - await this.ensureUser(recipient); - - await prisma.stream.upsert({ - where: { streamId }, - update: { - sender, - recipient, - tokenAddress, - ratePerSecond, - depositedAmount, - lastUpdateTime: startTime, - isActive: true, - }, - create: { - streamId, - sender, - recipient, - tokenAddress, - ratePerSecond, - depositedAmount, - withdrawnAmount: '0', - startTime, - lastUpdateTime: startTime, - isActive: true, - }, - }); - } else if (eventType === 'CANCELLED') { - await prisma.stream.updateMany({ - where: { streamId }, - data: { isActive: false, lastUpdateTime: BigInt(timestamp) }, - }); - } else if (eventType === 'WITHDRAWN') { - const stream = await prisma.stream.findUnique({ where: { streamId } }); - if (stream) { - const amount = this.readString(value, 'amount') ?? '0'; - const nextWithdrawn = (BigInt(stream.withdrawnAmount) + BigInt(amount)).toString(); - await prisma.stream.update({ - where: { streamId }, - data: { - withdrawnAmount: nextWithdrawn, - lastUpdateTime: BigInt(timestamp), - isActive: BigInt(nextWithdrawn) < BigInt(stream.depositedAmount), - }, - }); - } - } else if (eventType === 'COMPLETED') { - await prisma.stream.updateMany({ - where: { streamId }, - data: { isActive: false, lastUpdateTime: BigInt(timestamp) }, - }); - } - - await prisma.streamEvent.create({ - data: { - streamId, - eventType, - amount: this.readString(value, 'amount'), - transactionHash: txHash, - ledgerSequence, - timestamp: BigInt(timestamp), - metadata: JSON.stringify({ topic: event.topic, value: event.value }), - }, - }); - } -} - -export const sorobanIndexerService = new SorobanIndexerService(); diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index 86b76396..35f07095 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/backend/src/workers/soroban-event-worker.ts @@ -806,22 +806,22 @@ export class SorobanEventWorker { return; } - const stream = await tx.stream.findUniqueOrThrow({ - where: { streamId }, - select: { withdrawnAmount: true }, - }); - - const newWithdrawnAmount = ( - BigInt(stream.withdrawnAmount) + BigInt(amount) - ).toString(); - - await tx.stream.update({ - where: { streamId }, - data: { - withdrawnAmount: newWithdrawnAmount, - lastUpdateTime: timestamp, - }, - }); + // Use an atomic DB-level increment so that even under concurrent + // transactions the withdrawnAmount is never double-counted. Because + // Prisma models withdrawnAmount as String (not Int/BigInt), we use a + // raw UPDATE … SET "withdrawnAmount" = (CAST("withdrawnAmount" AS + // numeric) + $1)::text so the database performs the addition atomically. + // + // The idempotency guard above already prevents re-processing of the + // same event, but this atomic increment provides a safety net at the + // database level. + const amountBigInt = BigInt(amount); + await tx.$executeRaw` + UPDATE "Stream" + SET "withdrawnAmount" = (CAST("withdrawnAmount" AS numeric) + ${amountBigInt})::text, + "lastUpdateTime" = ${BigInt(timestamp)} + WHERE "streamId" = ${streamId} + `; await tx.streamEvent.upsert({ where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'WITHDRAWN' } }, diff --git a/backend/tests/single-indexer.regression.test.ts b/backend/tests/single-indexer.regression.test.ts new file mode 100644 index 00000000..0a4e45da --- /dev/null +++ b/backend/tests/single-indexer.regression.test.ts @@ -0,0 +1,63 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import path from 'path'; +import { fileURLToPath } from 'url'; + +/** + * Regression test for issue #801. + * + * After removing the legacy SorobanIndexerService, only SorobanEventWorker + * must be started during the server boot sequence. This test verifies: + * + * 1. The soroban-indexer.service module no longer exists. + * 2. index.ts does not import or call sorobanIndexerService. + * 3. Only startWorkers (which starts SorobanEventWorker) is called at boot. + */ + +const __dirname = path.dirname(fileURLToPath(import.meta.url)); + +describe('Single indexer regression (#801)', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('soroban-indexer.service.ts should no longer exist', async () => { + const filePath = path.resolve( + __dirname, + '../src/services/soroban-indexer.service.ts', + ); + let exists = false; + try { + await import('fs').then((fs) => { + fs.accessSync(filePath); + exists = true; + }); + } catch { + exists = false; + } + expect(exists).toBe(false); + }); + + it('index.ts should not reference sorobanIndexerService', async () => { + const fs = await import('fs'); + const indexPath = path.resolve(__dirname, '../src/index.ts'); + const content = fs.readFileSync(indexPath, 'utf-8'); + expect(content).not.toContain('sorobanIndexerService'); + expect(content).not.toContain('soroban-indexer.service'); + }); + + it('index.ts should only call startWorkers (not sorobanIndexerService.start)', async () => { + const fs = await import('fs'); + const indexPath = path.resolve(__dirname, '../src/index.ts'); + const content = fs.readFileSync(indexPath, 'utf-8'); + expect(content).toContain('await startWorkers()'); + expect(content).not.toMatch(/sorobanIndexerService\.start\(\)/); + }); + + it('workers/index.ts only starts SorobanEventWorker (no other indexer)', async () => { + const fs = await import('fs'); + const workersIndexPath = path.resolve(__dirname, '../src/workers/index.ts'); + const content = fs.readFileSync(workersIndexPath, 'utf-8'); + expect(content).toContain('sorobanEventWorker.start()'); + expect(content).not.toContain('sorobanIndexerService'); + }); +}); diff --git a/backend/tests/soroban-event-worker.test.ts b/backend/tests/soroban-event-worker.test.ts index 8d7cf386..da9c9131 100644 --- a/backend/tests/soroban-event-worker.test.ts +++ b/backend/tests/soroban-event-worker.test.ts @@ -19,7 +19,7 @@ const mockPrismaObj = vi.hoisted(() => ({ upsert: vi.fn(), create: vi.fn(), }, - $transaction: vi.fn((cb) => cb({ streamEvent: { findUnique: vi.fn(), upsert: vi.fn() }, user: { upsert: vi.fn() }, stream: { upsert: vi.fn(), update: vi.fn() } })), + $transaction: vi.fn((cb) => cb({ streamEvent: { findUnique: vi.fn(), upsert: vi.fn() }, user: { upsert: vi.fn() }, stream: { upsert: vi.fn(), update: vi.fn() }, $executeRaw: vi.fn().mockResolvedValue(1) })), $disconnect: vi.fn(), })); @@ -462,39 +462,48 @@ describe('SorobanEventWorker', () => { // withdrawnAmount starts at '1000'; a single successful withdrawal of // 500 should bring it to '1500' and stay there under replay. + const mockExecuteRaw = vi.fn().mockResolvedValue(1); const mockTx = { stream: { - findUniqueOrThrow: vi.fn().mockResolvedValue({ withdrawnAmount: '1000' }), + findUniqueOrThrow: vi.fn(), update: vi.fn().mockResolvedValue({}), }, streamEvent: { findUnique: vi.fn(), upsert: vi.fn().mockResolvedValue({ id: 'withdraw-event-row' }), }, + $executeRaw: mockExecuteRaw, }; (prisma.$transaction as ReturnType).mockImplementation((cb) => cb(mockTx)); - // First processing: no existing event → withdrawnAmount is updated once. + // First processing: no existing event → atomic increment is executed once. mockTx.streamEvent.findUnique.mockResolvedValueOnce(null); await expect((worker as any).handleTokensWithdrawn(mockEvent, mockEvent.topic![1])).resolves.not.toThrow(); - expect(mockTx.stream.update).toHaveBeenCalledTimes(1); - expect(mockTx.stream.update).toHaveBeenCalledWith({ - where: { streamId: BigInt(streamId) }, - data: { withdrawnAmount: '1500', lastUpdateTime: 1700002000 }, - }); + expect(mockExecuteRaw).toHaveBeenCalledTimes(1); expect(mockTx.streamEvent.upsert).toHaveBeenCalledTimes(1); expect(logger.warn).not.toHaveBeenCalled(); vi.clearAllMocks(); - (prisma.$transaction as ReturnType).mockImplementation((cb) => cb(mockTx)); + const mockExecuteRaw2 = vi.fn().mockResolvedValue(1); + const mockTx2 = { + stream: { + findUniqueOrThrow: vi.fn(), + update: vi.fn().mockResolvedValue({}), + }, + streamEvent: { + findUnique: vi.fn().mockResolvedValue({ id: 'withdraw-event-row' }), + upsert: vi.fn(), + }, + $executeRaw: mockExecuteRaw2, + }; + (prisma.$transaction as ReturnType).mockImplementation((cb) => cb(mockTx2)); // Second processing (replay of same txHash): the event now exists, so // withdrawnAmount must NOT be touched a second time. - mockTx.streamEvent.findUnique.mockResolvedValueOnce({ id: 'withdraw-event-row' }); await expect((worker as any).handleTokensWithdrawn(mockEvent, mockEvent.topic![1])).resolves.not.toThrow(); - expect(mockTx.stream.update).not.toHaveBeenCalled(); - expect(mockTx.streamEvent.upsert).not.toHaveBeenCalled(); + expect(mockExecuteRaw2).not.toHaveBeenCalled(); + expect(mockTx2.streamEvent.upsert).not.toHaveBeenCalled(); expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining('Duplicate StreamEvent skipped')); }); diff --git a/backend/tests/soroban-indexer.test.ts b/backend/tests/soroban-indexer.test.ts deleted file mode 100644 index 645d191e..00000000 --- a/backend/tests/soroban-indexer.test.ts +++ /dev/null @@ -1,82 +0,0 @@ -import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { sorobanIndexerService } from '../src/services/soroban-indexer.service.js'; - -vi.mock('../src/logger.js', () => ({ - default: { - info: vi.fn(), - error: vi.fn(), - warn: vi.fn(), - }, -})); - -// This service only reads/writes via a handful of prisma calls; mocking it -// out keeps these tests independent of whether the Prisma client has been -// generated (e.g. in a checkout without a `prisma generate` step). -vi.mock('../src/lib/prisma.js', () => ({ - prisma: { - streamEvent: { - findFirst: vi.fn(), - create: vi.fn(), - }, - stream: { - upsert: vi.fn(), - updateMany: vi.fn(), - update: vi.fn(), - findUnique: vi.fn(), - }, - user: { - upsert: vi.fn(), - }, - }, -})); - -describe('Soroban Indexer Service', () => { - beforeEach(() => { - vi.clearAllMocks(); - }); - - it('should start and stop the indexer', () => { - sorobanIndexerService.start(); - sorobanIndexerService.stop(); - }); -}); - -describe('Soroban Indexer Service - RPC resilience', () => { - beforeEach(() => { - vi.resetModules(); - vi.clearAllMocks(); - }); - - afterEach(() => { - vi.unstubAllGlobals(); - vi.useRealTimers(); - delete process.env.STREAM_CONTRACT_ID; - delete process.env.SOROBAN_RPC_TIMEOUT_MS; - delete process.env.SOROBAN_RPC_MAX_RETRIES; - }); - - it('bounds a hung getEvents fetch with the configured RPC timeout instead of stalling the poll loop', async () => { - process.env.STREAM_CONTRACT_ID = 'CCONTRACTIDEXAMPLE0000000000000000000000000000000000000'; - process.env.SOROBAN_RPC_TIMEOUT_MS = '1000'; - process.env.SOROBAN_RPC_MAX_RETRIES = '0'; - - vi.stubGlobal( - 'fetch', - vi.fn(() => new Promise(() => {})) // a hung endpoint that never responds - ); - vi.useFakeTimers(); - - const logger = (await import('../src/logger.js')).default; - const { sorobanIndexerService: indexer } = await import('../src/services/soroban-indexer.service.js'); - - indexer.start(); - await vi.advanceTimersByTimeAsync(1000); - - expect(logger.error).toHaveBeenCalledWith( - 'Soroban indexer poll failed', - expect.objectContaining({ name: 'RpcTimeoutError' }) - ); - - indexer.stop(); - }); -}); diff --git a/backend/tests/withdrawn-amount-double-increment.test.ts b/backend/tests/withdrawn-amount-double-increment.test.ts new file mode 100644 index 00000000..b299c560 --- /dev/null +++ b/backend/tests/withdrawn-amount-double-increment.test.ts @@ -0,0 +1,253 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { rpc } from '@stellar/stellar-sdk'; + +/** + * Regression tests for issue #801 — withdrawnAmount double-counting. + * + * These tests prove that: + * 1. The WITHDRAWN handler uses an atomic DB-level increment (not read-then-add). + * 2. Processing the same event twice is a no-op on the second attempt (idempotency). + * 3. Two near-simultaneous withdrawals on the same stream are both applied correctly + * when they carry different txHashes (no lost updates). + */ + +const mockPrismaObj = vi.hoisted(() => ({ + indexerState: { + findUnique: vi.fn(), + create: vi.fn(), + upsert: vi.fn(), + }, + user: { upsert: vi.fn() }, + stream: { + upsert: vi.fn(), + findUniqueOrThrow: vi.fn(), + update: vi.fn(), + }, + streamEvent: { + findUnique: vi.fn(), + upsert: vi.fn(), + }, + $transaction: vi.fn((cb: Function) => + cb({ + stream: { findUniqueOrThrow: vi.fn(), update: vi.fn() }, + streamEvent: { findUnique: vi.fn(), upsert: vi.fn() }, + $executeRaw: vi.fn().mockResolvedValue(1), + }), + ), + $disconnect: vi.fn(), +})); + +vi.mock('../src/lib/prisma.js', () => ({ + default: mockPrismaObj, + prisma: mockPrismaObj, +})); + +vi.mock('../src/services/sse.service.js', () => ({ + sseService: { + broadcastToStream: vi.fn(), + broadcast: vi.fn(), + broadcastToAdmin: vi.fn(), + }, +})); + +vi.mock('../src/logger.js', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + 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 logger from '../src/logger.js'; + +function makeWithdrawnEvent( + txHash: string, + streamId: number, + amount: string, + ledger: number, +): rpc.Api.EventResponse { + return { + id: `event-${txHash}`, + type: 'contract', + ledger, + ledgerClosedAt: '2024-01-01T00:00:00Z', + txHash, + transactionIndex: 0, + operationIndex: 0, + inSuccessfulContractCall: true, + topic: [ + { + switch: () => ({ value: 0 }), + sym: () => 'tokens_withdrawn', + } as any, + { + switch: () => ({ value: 1 }), + u64: () => ({ toString: () => streamId.toString() }), + } as any, + ], + value: { + switch: () => ({ value: 4 }), + map: () => [ + { + key: () => ({ sym: () => 'recipient' }), + val: () => ({ + address: () => ({ + switch: () => ({ value: 0 }), + accountId: () => ({ ed25519: () => Buffer.alloc(32) }), + }), + }), + }, + { + key: () => ({ sym: () => 'amount' }), + val: () => ({ + i128: () => ({ + hi: () => ({ toString: () => '0' }), + lo: () => ({ toString: () => amount }), + }), + }), + }, + { + key: () => ({ sym: () => 'timestamp' }), + val: () => ({ + u64: () => ({ toString: () => '1700002000' }), + }), + }, + ] as any, + } as any, + }; +} + +describe('withdrawnAmount double-increment prevention (#801)', () => { + let worker: SorobanEventWorker; + + beforeEach(() => { + vi.clearAllMocks(); + worker = new SorobanEventWorker(); + }); + + it('uses atomic $executeRaw instead of read-then-add', async () => { + const streamId = 42; + const event = makeWithdrawnEvent('tx-atomic', streamId, '500', 4000); + + const mockExecuteRaw = vi.fn().mockResolvedValue(1); + const mockTx = { + stream: { + findUniqueOrThrow: vi.fn(), + update: vi.fn().mockResolvedValue({}), + }, + streamEvent: { + findUnique: vi.fn().mockResolvedValue(null), + upsert: vi.fn().mockResolvedValue({ id: 'evt-1' }), + }, + $executeRaw: mockExecuteRaw, + }; + + (prisma.$transaction as ReturnType).mockImplementation((cb: Function) => + cb(mockTx), + ); + + await (worker as any).handleTokensWithdrawn(event, event.topic![1]); + + // The update must use an atomic $executeRaw, NOT a pre-read + string concat + expect(mockExecuteRaw).toHaveBeenCalledTimes(1); + // stream.update must NOT have been called for withdrawnAmount (only $executeRaw) + expect(mockTx.stream.update).not.toHaveBeenCalled(); + // stream.findUniqueOrThrow must NOT have been called (no read-then-add) + expect(mockTx.stream.findUniqueOrThrow).not.toHaveBeenCalled(); + }); + + it('processing the same event twice is a no-op on the second attempt', async () => { + const streamId = 55; + const event = makeWithdrawnEvent('tx-idempotent', streamId, '300', 4100); + + // First processing: event does not exist → should execute atomic increment + const mockExecuteRaw1 = vi.fn().mockResolvedValue(1); + (prisma.$transaction as ReturnType).mockImplementation((cb: Function) => + cb({ + stream: { findUniqueOrThrow: vi.fn(), update: vi.fn() }, + streamEvent: { + findUnique: vi.fn().mockResolvedValue(null), + upsert: vi.fn().mockResolvedValue({ id: 'evt-idem' }), + }, + $executeRaw: mockExecuteRaw1, + }), + ); + + await (worker as any).handleTokensWithdrawn(event, event.topic![1]); + expect(mockExecuteRaw1).toHaveBeenCalledTimes(1); + + vi.clearAllMocks(); + + // Second processing: event already exists (duplicate) → should NOT execute atomic increment + const mockExecuteRaw2 = vi.fn().mockResolvedValue(1); + (prisma.$transaction as ReturnType).mockImplementation((cb: Function) => + cb({ + stream: { findUniqueOrThrow: vi.fn(), update: vi.fn() }, + streamEvent: { + findUnique: vi.fn().mockResolvedValue({ id: 'evt-idem' }), + upsert: vi.fn(), + }, + $executeRaw: mockExecuteRaw2, + }), + ); + + await (worker as any).handleTokensWithdrawn(event, event.topic![1]); + + // The second call must NOT execute the atomic increment or upsert + expect(mockExecuteRaw2).not.toHaveBeenCalled(); + expect(logger.warn).toHaveBeenCalledWith( + expect.stringContaining('Duplicate StreamEvent skipped'), + ); + }); + + it('two different withdrawals on the same stream both trigger atomic increments', async () => { + const streamId = 77; + const event1 = makeWithdrawnEvent('tx-first', streamId, '100', 5000); + const event2 = makeWithdrawnEvent('tx-second', streamId, '200', 5001); + + const mockExecuteRaw1 = vi.fn().mockResolvedValue(1); + const mockTx1 = { + stream: { + findUniqueOrThrow: vi.fn(), + update: vi.fn().mockResolvedValue({}), + }, + streamEvent: { + findUnique: vi.fn().mockResolvedValue(null), + upsert: vi.fn().mockResolvedValue({ id: 'evt-1' }), + }, + $executeRaw: mockExecuteRaw1, + }; + + const mockExecuteRaw2 = vi.fn().mockResolvedValue(1); + const mockTx2 = { + stream: { + findUniqueOrThrow: vi.fn(), + update: vi.fn().mockResolvedValue({}), + }, + streamEvent: { + findUnique: vi.fn().mockResolvedValue(null), + upsert: vi.fn().mockResolvedValue({ id: 'evt-2' }), + }, + $executeRaw: mockExecuteRaw2, + }; + + let callCount = 0; + (prisma.$transaction as ReturnType).mockImplementation((cb: Function) => { + const tx = callCount++ === 0 ? mockTx1 : mockTx2; + return cb(tx); + }); + + await (worker as any).handleTokensWithdrawn(event1, event1.topic![1]); + await (worker as any).handleTokensWithdrawn(event2, event2.topic![1]); + + // Both transactions must have executed the atomic increment + expect(mockExecuteRaw1).toHaveBeenCalledTimes(1); + expect(mockExecuteRaw2).toHaveBeenCalledTimes(1); + }); +}); diff --git a/backend/vitest.config.ts b/backend/vitest.config.ts index 6c2d6878..356ff76e 100644 --- a/backend/vitest.config.ts +++ b/backend/vitest.config.ts @@ -26,7 +26,6 @@ export default defineConfig({ 'src/lib/prisma-sandbox.ts', 'src/services/indexer-integration.example.ts', 'src/services/indexerService.ts', - 'src/services/soroban-indexer.service.ts', 'src/services/sorobanService.ts', 'src/workers/soroban-event-worker.ts', ],