LabsCrypt · Killerjunior · Jul 29, 2026 · Jul 30, 2026 · Jul 30, 2026
diff --git a/backend/README.md b/backend/README.md
@@ -32,6 +32,23 @@ We use Prisma as our ORM to interact with PostgreSQL.
 - Configuration (schema path, migrations path, datasource URL) is defined in `prisma.config.ts`.
 - Run `npx prisma studio` to view the database through a web UI.
 
+## Database Seeding
+
+Run the seed script to populate the database with demo data for local development:
+
+```bash
+npx prisma db seed
+```
+
+**Idempotency guarantee** – The seed script (`prisma/seed.ts`) is safe to run multiple times against the same database. Every record is written with `upsert()` rather than `create()`:
+
+| Model | Unique key used for upsert |
+|---|---|
+| `User` | `publicKey` |
+| `Stream` | `streamId` |
+| `StreamEvent` | `transactionHash` + `eventType` |
+
+Re-running the seed against a populated database will leave existing rows unchanged and will not throw unique-constraint errors or create duplicate rows.
 ### Seeding the database
 
 `prisma/seed.ts` populates the database with demo fixtures for local development. Run it with:

diff --git a/backend/prisma/seed.ts b/backend/prisma/seed.ts
@@ -54,13 +54,22 @@ async function main() {
 
     console.log({ stream1 });
 
-    // Create an example event
-    const event1 = await prisma.streamEvent.create({
-        data: {
+    // Upsert an example event – keyed on the @@unique([transactionHash, eventType])
+    // constraint so re-running the seed never creates duplicate rows.
+    const DEMO_TX_HASH = '6f7e8a9b0c1d2e3f4a5b6c7d8e9f0a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6q7r';
+    const event1 = await prisma.streamEvent.upsert({
+        where: {
+            transactionHash_eventType: {
+                transactionHash: DEMO_TX_HASH,
+                eventType: 'CREATED',
+            },
+        },
+        update: {},
+        create: {
             streamId: stream1.streamId,
             eventType: 'CREATED',
             amount: '1000000000000',
-            transactionHash: '6f7e8a9b0c1d2e3f4a5b6c7d8e9f0a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6q7r',
+            transactionHash: DEMO_TX_HASH,
             ledgerSequence: 123456,
             timestamp: Math.floor(Date.now() / 1000),
             metadata: JSON.stringify({ memo: 'Seed data' }),

diff --git a/backend/src/controllers/stream.controller.ts b/backend/src/controllers/stream.controller.ts
@@ -211,7 +211,7 @@ export const createStream = async (req: Request, res: Response) => {
       },
     });
 
-    return res.status(201).json(stream);
+    return res.status(201).json(JSON.parse(JSON.stringify(stream, (_key, value) => typeof value === "bigint" ? value.toString() : value)));
   } catch (error) {
     if (
       error instanceof RangeError ||

diff --git a/backend/src/lib/indexer-state.ts b/backend/src/lib/indexer-state.ts
@@ -1,4 +1,4 @@
 import { prisma } from './prisma.js';
 import logger from '../logger.js';

 export const INDEXER_STATE_ID = 'singleton';
@@ -7,7 +7,6 @@
   id: string;
   lastLedger: number;
   lastCursor: string | null;
-  createdAt: Date;
   updatedAt: Date;
 }
 

diff --git a/backend/src/middleware/admin-rate-limiter.middleware.ts b/backend/src/middleware/admin-rate-limiter.middleware.ts
@@ -15,7 +15,7 @@ export const adminRateLimiter = rateLimit({
     // Use x-forwarded-for or remote address as key
     const forwarded = req.headers['x-forwarded-for'];
     if (typeof forwarded === 'string') {
-      return forwarded.split(',')[0].trim();
+      return forwarded.split(',')[0]?.trim() ?? 'unknown';
     }
     return req.ip ?? 'unknown';
   },

diff --git a/backend/src/services/soroban-indexer.service.ts b/backend/src/services/soroban-indexer.service.ts
@@ -179,8 +179,13 @@ 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 = BigInt(startTimeRaw ?? timestamp);
-      const timestampBigInt = BigInt(timestamp);
+      const startTime = BigInt(
+        typeof startTimeRaw === 'string' ||
+          typeof startTimeRaw === 'number' ||
+          typeof startTimeRaw === 'bigint'
+          ? startTimeRaw
+          : timestamp,
+      );
 
       if (!sender || !recipient || !tokenAddress || !ratePerSecond || !depositedAmount) return;
 

diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts
@@ -1005,7 +1005,7 @@ export class SorobanEventWorker {
     }
 
     const sender = decodeAddress(body["sender"]);
-    const newEndTime = Number(decodeU64(body["new_end_time"]));
+    const newEndTime = decodeU64(body["new_end_time"]);
     const timestamp = Math.floor(Date.now() / 1000);
 
     await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
@@ -1018,7 +1018,7 @@ export class SorobanEventWorker {
       // Calculate the duration of this pause interval
       let additionalPausedDuration = 0;
       if (currentStream.pausedAt) {
-        additionalPausedDuration = timestamp - currentStream.pausedAt;
+        additionalPausedDuration = timestamp - Number(currentStream.pausedAt);
       }
 
       const newTotalPausedDuration =
@@ -1064,7 +1064,7 @@ export class SorobanEventWorker {
             timestamp,
             metadata: JSON.stringify({
               sender,
-              newEndTime,
+              newEndTime: newEndTime.toString(),
               pausedDuration: additionalPausedDuration,
               totalPausedDuration: newTotalPausedDuration,
             }),
@@ -1080,7 +1080,7 @@ export class SorobanEventWorker {
     sseService.broadcastToStream(String(streamId), "stream.resumed", {
       streamId,
       sender,
-      newEndTime,
+      newEndTime: newEndTime.toString(),
       transactionHash: event.txHash,
       ledger: event.ledger,
       timestamp,

diff --git a/backend/tests/events-wire-format.test.ts b/backend/tests/events-wire-format.test.ts
@@ -91,7 +91,7 @@ describe('event wire format', () => {
 
       expect(decodedKeys).toEqual([...allFields].sort());
 
-      for (const field of HANDLER_READ_FIELDS[eventName]) {
+      for (const field of HANDLER_READ_FIELDS[eventName] ?? []) {
         expect(decoded).toHaveProperty(field);
       }
     },

diff --git a/backend/tests/soroban-event-worker.test.ts b/backend/tests/soroban-event-worker.test.ts
@@ -575,7 +575,7 @@ describe('SorobanEventWorker', () => {
       // Sanity check: depositedAmount/endTime from the (only) applied update
       // match what a single application should produce.
       expect(firstUpdateArgs.data.depositedAmount).toBe('1200');
-      expect(expectedEndTime).toBe(1700000000 + Math.floor(1200 / 10) + 0);
+      expect(expectedEndTime).toBe(1700000000n + BigInt(Math.floor(1200 / 10)));
     });
 
     it('should process admin_transferred events successfully', async () => {

diff --git a/backend/tests/stream.test.ts b/backend/tests/stream.test.ts
@@ -65,8 +65,8 @@ describe('POST /v1/streams', () => {
       ratePerSecond: '100',
       depositedAmount: '86400',
       withdrawnAmount: '0',
-      startTime: 1700000000,
-      lastUpdateTime: 1700000000,
+      startTime: 1700000000n,
+      lastUpdateTime: 1700000000n,
       isPaused: false,
       pausedAt: null,
       totalPausedDuration: 0,
@@ -247,8 +247,8 @@ describe('GET /v1/users/:address/summary', () => {
           ratePerSecond: '10',
           depositedAmount: '100',
           withdrawnAmount: '30',
-          startTime: 1000,
-          lastUpdateTime: 2000,
+          startTime: 1000n,
+          lastUpdateTime: 2000n,
           isPaused: false,
       endTime: null,
       pausedAt: null,
@@ -266,8 +266,8 @@ describe('GET /v1/users/:address/summary', () => {
           ratePerSecond: '20',
           depositedAmount: '200',
           withdrawnAmount: '20',
-          startTime: 1000,
-          lastUpdateTime: 2000,
+          startTime: 1000n,
+          lastUpdateTime: 2000n,
           isPaused: false,
       endTime: null,
       pausedAt: null,
@@ -287,8 +287,8 @@ describe('GET /v1/users/:address/summary', () => {
           ratePerSecond: '10',
           depositedAmount: '1000',
           withdrawnAmount: '100',
-          startTime: 1000,
-          lastUpdateTime: 0,
+          startTime: 1000n,
+          lastUpdateTime: 0n,
           isPaused: false,
       endTime: null,
       pausedAt: null,
@@ -306,8 +306,8 @@ describe('GET /v1/users/:address/summary', () => {
           ratePerSecond: '5',
           depositedAmount: '500',
           withdrawnAmount: '0',
-          startTime: 1000,
-          lastUpdateTime: 0,
+          startTime: 1000n,
+          lastUpdateTime: 0n,
           isPaused: false,
       endTime: null,
       pausedAt: null,
@@ -344,8 +344,8 @@ describe('GET /v1/users/:address/summary', () => {
         ratePerSecond: '1',
         depositedAmount: '100',
         withdrawnAmount: '1',
-        startTime: 1000,
-        lastUpdateTime: 2000,
+        startTime: 1000n,
+        lastUpdateTime: 2000n,
         isPaused: false,
       endTime: null,
       pausedAt: null,

diff --git a/backend/tests/stream.validator.test.ts b/backend/tests/stream.validator.test.ts
@@ -42,7 +42,7 @@ describe('Stream Validator', () => {
     };
     const result = createStreamSchema.safeParse(data);
     expect(result.success).toBe(false);
-    expect(result.error?.issues[0].message).toBe('Rate exceeds maximum allowed value');
+    expect(result.error?.issues[0]?.message).toBe('Rate exceeds maximum allowed value');
   });
 
   it('should accept ratePerSecond at i128 max', () => {

diff --git a/contracts/stream_contract/src/test.rs b/contracts/stream_contract/src/test.rs
@@ -433,8 +433,8 @@ fn test_backdated_start_time_would_immediately_vest_full_amount() {
     // env.ledger().timestamp() — but it demonstrates the risk that would exist
     // if a caller-supplied start_time were ever added.
     let mut stream = client.get_stream(&stream_id).unwrap();
-    stream.start_time = 0;          // backdated far into the past
-    stream.last_update_time = 0;    // sync anchor to match
+    stream.start_time = 0; // backdated far into the past
+    stream.last_update_time = 0; // sync anchor to match
     env.as_contract(&client.address, || {
         env.storage()
             .persistent()
@@ -2709,11 +2709,8 @@ fn test_cancel_state_committed_before_transfers_prevents_double_cancel() {
 fn event_field_names(env: &Env, payload: &soroban_sdk::Val) -> std::vec::Vec<std::string::String> {
     let map = soroban_sdk::Map::<Symbol, soroban_sdk::Val>::try_from_val(env, payload)
         .expect("event data is not a Map");
-    let mut names: std::vec::Vec<std::string::String> = map
-        .keys()
-        .iter()
-        .map(|sym| sym.to_string())
-        .collect();
+    let mut names: std::vec::Vec<std::string::String> =
+        map.keys().iter().map(|sym| sym.to_string()).collect();
     names.sort();
     names
 }

diff --git a/...tend/src/hooks/useIncomingStreams.test.ts → ...end/src/hooks/useIncomingStreams.test.tsx b/...tend/src/hooks/useIncomingStreams.test.ts → ...end/src/hooks/useIncomingStreams.test.tsx
@@ -1,4 +1,5 @@
+/* eslint-disable @typescript-eslint/no-explicit-any */
 import { renderHook, waitFor, act } from "@testing-library/react";
 import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
 import { vi, describe, it, expect, beforeEach, afterEach } from "vitest";
 import React from "react";
@@ -31,6 +32,7 @@
   });
 
   afterEach(() => {
+    vi.useRealTimers();
     queryClient.clear();
   });
 
@@ -89,6 +91,7 @@
       );
 
       const invalidateSpy = vi.spyOn(queryClient, "invalidateQueries");
+      vi.useFakeTimers();
 
       await act(async () => {
         await result.current.mutateAsync({
@@ -102,12 +105,14 @@
         } as any);
       });
 
-      // Wait for pollIndexerForWithdraw to complete and call invalidateQueries
-      await waitFor(() => {
-        expect(invalidateSpy).toHaveBeenCalledWith({
-          queryKey: incomingStreamsQueryKey("pubkey"),
-        });
-      }, { timeout: 10000 });
+      // Advance the retry delays so the fallback invalidation runs immediately.
+      await act(async () => {
+        await vi.advanceTimersByTimeAsync(63_000);
+      });
+
+      expect(invalidateSpy).toHaveBeenCalledWith({
+        queryKey: incomingStreamsQueryKey("pubkey"),
+      });
     });
   });
 });