From dd2878929b891013a643e719ea1f2b714ca11916 Mon Sep 17 00:00:00 2001 From: Samuel Dahunsi Date: Wed, 29 Jul 2026 17:15:55 +0100 Subject: [PATCH 1/2] fix(contracts,backend,frontend): resolve issues #794, #830, #831, and #862 - #794: Verify collect_fee rounds-to-zero branch test in contracts/stream_contract/src/test.rs - #830: Change Stream (startTime, lastUpdateTime, endTime, pausedAt) and StreamEvent (timestamp) to BigInt in schema.prisma and add migration - #831: Fix controller vs indexer event write race using upsert on transactionHash_eventType in withdraw.ts and updating ledgerSequence/timestamp in worker upsert - #862: Verify stray Token: label removal in frontend stream detail page --- .../migration.sql | 8 ++ backend/prisma/schema.prisma | 10 +-- backend/src/routes/v1/streams/withdraw.ts | 13 ++- backend/src/workers/soroban-event-worker.ts | 55 +++++++++--- backend/tests/eventRace.test.ts | 87 +++++++++++++++++++ backend/tests/withdraw.handler.test.ts | 12 ++- 6 files changed, 162 insertions(+), 23 deletions(-) create mode 100644 backend/prisma/migrations/20260729170000_timestamp_bigint/migration.sql create mode 100644 backend/tests/eventRace.test.ts diff --git a/backend/prisma/migrations/20260729170000_timestamp_bigint/migration.sql b/backend/prisma/migrations/20260729170000_timestamp_bigint/migration.sql new file mode 100644 index 00000000..e419f40e --- /dev/null +++ b/backend/prisma/migrations/20260729170000_timestamp_bigint/migration.sql @@ -0,0 +1,8 @@ +-- AlterTable +ALTER TABLE "Stream" ALTER COLUMN "startTime" SET DATA TYPE BIGINT, +ALTER COLUMN "lastUpdateTime" SET DATA TYPE BIGINT, +ALTER COLUMN "endTime" SET DATA TYPE BIGINT, +ALTER COLUMN "pausedAt" SET DATA TYPE BIGINT; + +-- AlterTable +ALTER TABLE "StreamEvent" ALTER COLUMN "timestamp" SET DATA TYPE BIGINT; diff --git a/backend/prisma/schema.prisma b/backend/prisma/schema.prisma index 0b0e001b..664105b7 100644 --- a/backend/prisma/schema.prisma +++ b/backend/prisma/schema.prisma @@ -34,12 +34,12 @@ model Stream { ratePerSecond String // Rate as string to preserve precision (i128) depositedAmount String // Total deposited amount (i128) withdrawnAmount String // Total withdrawn amount (i128) - startTime Int // Unix timestamp when stream started - lastUpdateTime Int // Unix timestamp of last update - endTime Int? // Unix timestamp when stream ends + startTime BigInt // Unix timestamp when stream started + lastUpdateTime BigInt // Unix timestamp of last update + endTime BigInt? // Unix timestamp when stream ends isActive Boolean @default(true) isPaused Boolean @default(false) - pausedAt Int? // Unix timestamp when paused + pausedAt BigInt? // Unix timestamp when paused totalPausedDuration Int @default(0) // Accumulated paused duration in seconds createdAt DateTime @default(now()) updatedAt DateTime @updatedAt @@ -73,7 +73,7 @@ model StreamEvent { amount String? // Amount involved in the event (for top-ups, withdrawals) transactionHash String // Stellar transaction hash ledgerSequence Int // Ledger sequence number - timestamp Int // Unix timestamp + timestamp BigInt // Unix timestamp metadata String? // JSON string for additional event data createdAt DateTime @default(now()) diff --git a/backend/src/routes/v1/streams/withdraw.ts b/backend/src/routes/v1/streams/withdraw.ts index 9abc268c..9b063301 100644 --- a/backend/src/routes/v1/streams/withdraw.ts +++ b/backend/src/routes/v1/streams/withdraw.ts @@ -122,9 +122,15 @@ export const withdrawHandler = async (req: AuthenticatedRequest, res: Response) }, }); - // Create a WITHDRAWN event - await prisma.streamEvent.create({ - data: { + // Create or update a WITHDRAWN event + await prisma.streamEvent.upsert({ + where: { + transactionHash_eventType: { + transactionHash: result.txHash, + eventType: 'WITHDRAWN', + }, + }, + create: { streamId: parsedStreamId, eventType: 'WITHDRAWN', amount: claimable.claimableAmount, @@ -133,6 +139,7 @@ export const withdrawHandler = async (req: AuthenticatedRequest, res: Response) timestamp: now, metadata: JSON.stringify({ withdrawnBy: req.user.publicKey }), }, + update: {}, }); logger.info(`Stream ${parsedStreamId} withdrawn by ${req.user.publicKey}`); diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index c2ac91ab..ade69ae4 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/backend/src/workers/soroban-event-worker.ts @@ -377,7 +377,10 @@ export class SorobanEventWorker { new_fee_rate_bps: newFeeRateBps, }), }, - update: {}, + update: { + ledgerSequence: event.ledger, + timestamp, + }, }); }); @@ -428,7 +431,10 @@ export class SorobanEventWorker { transactionHash: event.txHash, }), }, - update: {}, + update: { + ledgerSequence: event.ledger, + timestamp, + }, }); }); @@ -466,13 +472,13 @@ export class SorobanEventWorker { const tokenAddress = decodeAddress(body["token_address"]); const ratePerSecond = decodeI128(body["rate_per_second"]); const depositedAmount = decodeI128(body["deposited_amount"]); - const startTime = Number(decodeU64(body["start_time"])); + const startTime = BigInt(decodeU64(body["start_time"])); const ratePerSecondBigInt = BigInt(ratePerSecond); const endTime = ratePerSecondBigInt === 0n ? null - : startTime + Number(BigInt(depositedAmount) / ratePerSecondBigInt); + : startTime + BigInt(depositedAmount) / ratePerSecondBigInt; await prisma.$transaction(async (tx: Prisma.TransactionClient) => { await tx.user.upsert({ @@ -542,7 +548,10 @@ export class SorobanEventWorker { timestamp: startTime, metadata: JSON.stringify({ tokenAddress, ratePerSecond }), }, - update: {}, + update: { + ledgerSequence: event.ledger, + timestamp: startTime, + }, }); } }); @@ -600,9 +609,9 @@ export class SorobanEventWorker { const newEndTime = ratePerSecondBigInt === 0n ? null - : stream.startTime + - Number(BigInt(newDepositedAmount) / ratePerSecondBigInt) + - stream.totalPausedDuration; + : BigInt(stream.startTime) + + (BigInt(newDepositedAmount) / ratePerSecondBigInt) + + BigInt(stream.totalPausedDuration); await tx.stream.update({ where: { streamId }, @@ -624,7 +633,10 @@ export class SorobanEventWorker { timestamp, metadata: JSON.stringify({ newDepositedAmount }), }, - update: {}, + update: { + ledgerSequence: event.ledger, + timestamp, + }, }); }); @@ -693,7 +705,10 @@ export class SorobanEventWorker { timestamp, metadata: JSON.stringify({ recipient }), }, - update: {}, + update: { + ledgerSequence: event.ledger, + timestamp, + }, }); }); @@ -762,7 +777,10 @@ export class SorobanEventWorker { timestamp, metadata: JSON.stringify({ amountWithdrawn, refundedAmount }), }, - update: {}, + update: { + ledgerSequence: event.ledger, + timestamp, + }, }); } }); @@ -832,7 +850,10 @@ export class SorobanEventWorker { timestamp, metadata: JSON.stringify({ recipient }), }, - update: {}, + update: { + ledgerSequence: event.ledger, + timestamp, + }, }); } }); @@ -962,7 +983,10 @@ export class SorobanEventWorker { timestamp, metadata: JSON.stringify({ sender, pausedAt }), }, - update: {}, + update: { + ledgerSequence: event.ledger, + timestamp, + }, }); } }); @@ -1053,7 +1077,10 @@ export class SorobanEventWorker { totalPausedDuration: newTotalPausedDuration, }), }, - update: {}, + update: { + ledgerSequence: event.ledger, + timestamp, + }, }); } }); diff --git a/backend/tests/eventRace.test.ts b/backend/tests/eventRace.test.ts new file mode 100644 index 00000000..361f195d --- /dev/null +++ b/backend/tests/eventRace.test.ts @@ -0,0 +1,87 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { withdrawHandler } from '../src/routes/v1/streams/withdraw.js'; +import { prisma } from '../src/lib/prisma.js'; +import { claimableAmountService } from '../src/services/claimable.service.js'; +import { withdraw as sorobanWithdraw } from '../src/services/sorobanService.js'; +import type { Response } from 'express'; +import type { AuthenticatedRequest } from '../src/types/auth.types.js'; + +vi.mock('../src/lib/prisma.js', () => ({ + prisma: { + stream: { + findUnique: vi.fn(), + update: vi.fn(), + }, + streamEvent: { + upsert: vi.fn(), + }, + }, +})); + +vi.mock('../src/services/claimable.service.js', () => ({ + claimableAmountService: { + getClaimableAmount: vi.fn(), + }, +})); + +vi.mock('../src/services/sorobanService.js', () => ({ + withdraw: vi.fn(), +})); + +vi.mock('../src/logger.js', () => ({ + default: { + info: vi.fn(), + error: vi.fn(), + warn: vi.fn(), + }, +})); + +describe('Action Controller vs Worker Event Write Race Guard (Issue #831)', () => { + let req: Partial; + let res: Partial; + + beforeEach(() => { + vi.clearAllMocks(); + req = { + params: { streamId: '100' }, + user: { publicKey: 'GRECIPIENT' } as any, + }; + res = { + status: vi.fn().mockReturnThis(), + json: vi.fn().mockReturnThis(), + }; + }); + + it('withdrawHandler uses upsert on transactionHash_eventType preventing P2002 duplicate crashes when worker processes event first', async () => { + const mockStream = { + streamId: 100, + recipient: 'GRECIPIENT', + withdrawnAmount: '0', + depositedAmount: '1000', + isActive: true, + }; + (prisma.stream.findUnique as any).mockResolvedValue(mockStream); + (claimableAmountService.getClaimableAmount as any).mockReturnValue({ actionable: true, claimableAmount: '500' }); + (sorobanWithdraw as any).mockResolvedValue({ txHash: 'tx_race_123' }); + (prisma.stream.update as any).mockResolvedValue({ ...mockStream, withdrawnAmount: '500' }); + (prisma.streamEvent.upsert as any).mockResolvedValue({ id: 'evt_1' }); + + await withdrawHandler(req as AuthenticatedRequest, res as Response); + + expect(res.status).toHaveBeenCalledWith(200); + expect(prisma.streamEvent.upsert).toHaveBeenCalledWith({ + where: { + transactionHash_eventType: { + transactionHash: 'tx_race_123', + eventType: 'WITHDRAWN', + }, + }, + create: expect.objectContaining({ + streamId: 100, + eventType: 'WITHDRAWN', + transactionHash: 'tx_race_123', + }), + update: {}, + }); + }); +}); diff --git a/backend/tests/withdraw.handler.test.ts b/backend/tests/withdraw.handler.test.ts index 24217d65..91e4943e 100644 --- a/backend/tests/withdraw.handler.test.ts +++ b/backend/tests/withdraw.handler.test.ts @@ -14,6 +14,7 @@ vi.mock('../src/lib/prisma.js', () => ({ }, streamEvent: { create: vi.fn(), + upsert: vi.fn(), }, }, })); @@ -81,6 +82,15 @@ describe('Withdraw Handler', () => { expect(res.status).toHaveBeenCalledWith(200); expect(res.json).toHaveBeenCalledWith(expect.objectContaining({ success: true, txHash: 'tx123' })); - expect(prisma.streamEvent.create).toHaveBeenCalled(); + expect(prisma.streamEvent.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + where: { + transactionHash_eventType: { + transactionHash: 'tx123', + eventType: 'WITHDRAWN', + }, + }, + }) + ); }); }); From cef143bb9e4d60e160026059b61bbf9f57b67a62 Mon Sep 17 00:00:00 2001 From: Samuel Dahunsi Date: Wed, 29 Jul 2026 17:22:11 +0100 Subject: [PATCH 2/2] fix(backend): type timestamp fields as BigInt across controllers and indexers (#830) - Update createStream, topUpStreamHandler, withdrawHandler, and sorobanIndexerService to pass BigInt values for startTime, lastUpdateTime, endTime, and timestamp - Update ClaimableStreamState interface to accept number | bigint for timestamp fields --- backend/src/controllers/stream.controller.ts | 10 +++++----- backend/src/routes/v1/streams/withdraw.ts | 2 +- backend/src/services/claimable.service.ts | 12 ++++++------ backend/src/services/soroban-indexer.service.ts | 17 +++++++++-------- 4 files changed, 21 insertions(+), 20 deletions(-) diff --git a/backend/src/controllers/stream.controller.ts b/backend/src/controllers/stream.controller.ts index 968def20..e02ca46d 100644 --- a/backend/src/controllers/stream.controller.ts +++ b/backend/src/controllers/stream.controller.ts @@ -177,7 +177,7 @@ export const createStream = async (req: Request, res: Response) => { } const endTime = - parsedStartTime + Number(parsedDepositedAmount / parsedRatePerSecond); + BigInt(parsedStartTime) + (parsedDepositedAmount / parsedRatePerSecond); // Issue #809: never let the upsert update branch touch a stream owned by a // different wallet. The caller is already proven to equal `sender` above, so @@ -195,7 +195,7 @@ export const createStream = async (req: Request, res: Response) => { where: { streamId: parsedStreamId }, update: { isActive: true, - lastUpdateTime: Math.floor(Date.now() / 1000), + lastUpdateTime: BigInt(Math.floor(Date.now() / 1000)), }, create: { streamId: parsedStreamId, @@ -205,9 +205,9 @@ export const createStream = async (req: Request, res: Response) => { ratePerSecond, depositedAmount, withdrawnAmount: "0", - startTime: parsedStartTime, + startTime: BigInt(parsedStartTime), endTime, - lastUpdateTime: parsedStartTime, + lastUpdateTime: BigInt(parsedStartTime), }, }); @@ -738,7 +738,7 @@ export const topUpStreamHandler = async (req: Request, res: Response) => { where: { streamId }, data: { depositedAmount: newDeposited, - lastUpdateTime: Math.floor(Date.now() / 1000), + lastUpdateTime: BigInt(Math.floor(Date.now() / 1000)), }, }); diff --git a/backend/src/routes/v1/streams/withdraw.ts b/backend/src/routes/v1/streams/withdraw.ts index 9b063301..9c6e9846 100644 --- a/backend/src/routes/v1/streams/withdraw.ts +++ b/backend/src/routes/v1/streams/withdraw.ts @@ -106,7 +106,7 @@ export const withdrawHandler = async (req: AuthenticatedRequest, res: Response) // Call Soroban service const result = await sorobanWithdraw(parsedStreamId, req.user.publicKey); - const now = Math.floor(Date.now() / 1000); + const now = BigInt(Math.floor(Date.now() / 1000)); const nextWithdrawnAmount = ( BigInt(stream.withdrawnAmount) + BigInt(claimable.claimableAmount) ).toString(); diff --git a/backend/src/services/claimable.service.ts b/backend/src/services/claimable.service.ts index 78b75839..7907fed6 100644 --- a/backend/src/services/claimable.service.ts +++ b/backend/src/services/claimable.service.ts @@ -8,11 +8,11 @@ export interface ClaimableStreamState { ratePerSecond: string; depositedAmount: string; withdrawnAmount: string; - startTime: number; - lastUpdateTime: number; + startTime: number | bigint; + lastUpdateTime: number | bigint; isActive: boolean; isPaused: boolean; - pausedAt: number | null; + pausedAt: number | bigint | null; totalPausedDuration: number; updatedAt?: Date; } @@ -114,14 +114,14 @@ export class ClaimableAmountService { }; } - const anchorTime = BigInt(Math.max(0, stream.lastUpdateTime)); + const anchorTime = BigInt(stream.lastUpdateTime) > 0n ? BigInt(stream.lastUpdateTime) : 0n; const nowTs = BigInt(Math.max(0, calculatedAt)); let elapsed = nowTs > anchorTime ? nowTs - anchorTime : 0n; // Paused duration is handled by the contract updating lastUpdateTime on resume, // but we still account for it if it's currently paused. - if (stream.isPaused && stream.pausedAt !== null) { - const currentPauseStart = BigInt(Math.max(0, stream.pausedAt)); + if (stream.isPaused && stream.pausedAt !== null && stream.pausedAt !== undefined) { + const currentPauseStart = BigInt(stream.pausedAt) > 0n ? BigInt(stream.pausedAt) : 0n; if (nowTs > currentPauseStart) { const currentPauseDuration = nowTs - currentPauseStart; elapsed = elapsed > currentPauseDuration ? elapsed - currentPauseDuration : 0n; diff --git a/backend/src/services/soroban-indexer.service.ts b/backend/src/services/soroban-indexer.service.ts index e051712e..53fd1a4c 100644 --- a/backend/src/services/soroban-indexer.service.ts +++ b/backend/src/services/soroban-indexer.service.ts @@ -179,7 +179,8 @@ export class SorobanIndexerService { const ratePerSecond = this.readString(value, 'rate_per_second', 'ratePerSecond'); const depositedAmount = this.readString(value, 'deposited_amount', 'depositedAmount'); const startTimeRaw = value.start_time ?? value.startTime ?? timestamp; - const startTime = Number(startTimeRaw); + const startTime = BigInt(startTimeRaw ?? timestamp); + const timestampBigInt = BigInt(timestamp); if (!sender || !recipient || !tokenAddress || !ratePerSecond || !depositedAmount) return; @@ -194,7 +195,7 @@ export class SorobanIndexerService { tokenAddress, ratePerSecond, depositedAmount, - lastUpdateTime: Number.isFinite(startTime) ? startTime : timestamp, + lastUpdateTime: startTime, isActive: true, }, create: { @@ -205,15 +206,15 @@ export class SorobanIndexerService { ratePerSecond, depositedAmount, withdrawnAmount: '0', - startTime: Number.isFinite(startTime) ? startTime : timestamp, - lastUpdateTime: Number.isFinite(startTime) ? startTime : timestamp, + startTime, + lastUpdateTime: startTime, isActive: true, }, }); } else if (eventType === 'CANCELLED') { await prisma.stream.updateMany({ where: { streamId }, - data: { isActive: false, lastUpdateTime: timestamp }, + data: { isActive: false, lastUpdateTime: BigInt(timestamp) }, }); } else if (eventType === 'WITHDRAWN') { const stream = await prisma.stream.findUnique({ where: { streamId } }); @@ -224,7 +225,7 @@ export class SorobanIndexerService { where: { streamId }, data: { withdrawnAmount: nextWithdrawn, - lastUpdateTime: timestamp, + lastUpdateTime: BigInt(timestamp), isActive: BigInt(nextWithdrawn) < BigInt(stream.depositedAmount), }, }); @@ -232,7 +233,7 @@ export class SorobanIndexerService { } else if (eventType === 'COMPLETED') { await prisma.stream.updateMany({ where: { streamId }, - data: { isActive: false, lastUpdateTime: timestamp }, + data: { isActive: false, lastUpdateTime: BigInt(timestamp) }, }); } @@ -243,7 +244,7 @@ export class SorobanIndexerService { amount: this.readString(value, 'amount'), transactionHash: txHash, ledgerSequence, - timestamp, + timestamp: BigInt(timestamp), metadata: JSON.stringify({ topic: event.topic, value: event.value }), }, });