fix(contracts,backend,frontend): test fee round-to-zero, BigInt timestamps, event write race, and stray label

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;

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())
 

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)),
       },
     });
 

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();
@@ -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}`);

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;

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,15 +225,15 @@ export class SorobanIndexerService {
           where: { streamId },
           data: {
             withdrawnAmount: nextWithdrawn,
-            lastUpdateTime: timestamp,
+            lastUpdateTime: BigInt(timestamp),
             isActive: BigInt(nextWithdrawn) < BigInt(stream.depositedAmount),
           },
         });
       }
     } 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 }),
       },
     });

backend/src/workers/soroban-event-worker.ts

@@ -33,7 +33,7 @@
  * Full value = hi * 2^64 + lo.
  */
 export function decodeI128(val: xdr.ScVal): string {
   const parts = val.i128();
   const hi = BigInt.asIntN(64, BigInt(parts.hi().toString()));
   const lo = BigInt.asUintN(64, BigInt(parts.lo().toString()));
   return ((hi << 64n) | lo).toString();
@@ -377,7 +377,10 @@
             new_fee_rate_bps: newFeeRateBps,
           }),
         },
-        update: {},
+        update: {
+          ledgerSequence: event.ledger,
+          timestamp,
+        },
       });
     });
 
@@ -428,7 +431,10 @@
             transactionHash: event.txHash,
           }),
         },
-        update: {},
+        update: {
+          ledgerSequence: event.ledger,
+          timestamp,
+        },
       });
     });
 
@@ -466,13 +472,13 @@
     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 @@
             timestamp: startTime,
             metadata: JSON.stringify({ tokenAddress, ratePerSecond }),
           },
-          update: {},
+          update: {
+            ledgerSequence: event.ledger,
+            timestamp: startTime,
+          },
         });
       }
     });
@@ -600,9 +609,9 @@
       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 @@
           timestamp,
           metadata: JSON.stringify({ newDepositedAmount }),
         },
-        update: {},
+        update: {
+          ledgerSequence: event.ledger,
+          timestamp,
+        },
       });
     });
 
@@ -693,7 +705,10 @@
           timestamp,
           metadata: JSON.stringify({ recipient }),
         },
-        update: {},
+        update: {
+          ledgerSequence: event.ledger,
+          timestamp,
+        },
       });
     });
 
@@ -762,7 +777,10 @@
             timestamp,
             metadata: JSON.stringify({ amountWithdrawn, refundedAmount }),
           },
-          update: {},
+          update: {
+            ledgerSequence: event.ledger,
+            timestamp,
+          },
         });
       }
     });
@@ -832,7 +850,10 @@
             timestamp,
             metadata: JSON.stringify({ recipient }),
           },
-          update: {},
+          update: {
+            ledgerSequence: event.ledger,
+            timestamp,
+          },
         });
       }
     });
@@ -962,7 +983,10 @@
             timestamp,
             metadata: JSON.stringify({ sender, pausedAt }),
           },
-          update: {},
+          update: {
+            ledgerSequence: event.ledger,
+            timestamp,
+          },
         });
       }
     });
@@ -1002,7 +1026,7 @@
       // Calculate the duration of this pause interval
       let additionalPausedDuration = 0;
       if (currentStream.pausedAt) {
         additionalPausedDuration = timestamp - currentStream.pausedAt;
       }
 
       const newTotalPausedDuration =
@@ -1053,7 +1077,10 @@
               totalPausedDuration: newTotalPausedDuration,
             }),
           },
-          update: {},
+          update: {
+            ledgerSequence: event.ledger,
+            timestamp,
+          },
         });
       }
     });