fix(backend): use BigInt for on-chain streamId (u64-safe)

backend/prisma/migrations/20260727090000_stream_id_bigint/migration.sql

@@ -0,0 +1,12 @@
+-- Convert on-chain stream identifiers from int4 to bigint (Soroban u64).
+-- Drop the FK first so both columns can be widened, then recreate it.
+
+ALTER TABLE "StreamEvent" DROP CONSTRAINT IF EXISTS "StreamEvent_streamId_fkey";
+
+ALTER TABLE "Stream" ALTER COLUMN "streamId" TYPE BIGINT USING ("streamId"::bigint);
+ALTER TABLE "StreamEvent" ALTER COLUMN "streamId" TYPE BIGINT USING ("streamId"::bigint);
+
+ALTER TABLE "StreamEvent"
+  ADD CONSTRAINT "StreamEvent_streamId_fkey"
+  FOREIGN KEY ("streamId") REFERENCES "Stream"("streamId")
+  ON DELETE RESTRICT ON UPDATE CASCADE;

backend/prisma/schema.prisma

@@ -27,7 +27,7 @@ model User {
 // Stream model - mirrors on-chain stream state for fast querying
 model Stream {
   id                String    @id @default(uuid())
-  streamId          Int       @unique // On-chain stream ID
+  streamId          BigInt    @unique // On-chain stream ID (Soroban u64)
   sender            String    // Sender's Stellar public key
   recipient         String    // Recipient's Stellar public key
   tokenAddress      String    // Token contract address
@@ -68,7 +68,7 @@ model IndexerState {
 // StreamEvent model - indexer events for tracking all on-chain stream activities
 model StreamEvent {
   id              String   @id @default(uuid())
-  streamId        Int      // Reference to on-chain stream ID
+  streamId        BigInt   // Reference to on-chain stream ID (Soroban u64)
   eventType       String   // EventType: "CREATED", "TOPPED_UP", "WITHDRAWN", "CANCELLED", "COMPLETED", "PAUSED", "RESUMED"
   amount          String?  // Amount involved in the event (for top-ups, withdrawals)
   transactionHash String   // Stellar transaction hash
@@ -87,12 +87,4 @@ model StreamEvent {
   @@index([transactionHash])
   @@index([createdAt])
   @@index([streamId, createdAt])
-  @@unique([transactionHash, eventType])
-}
-
-// IndexerState model - tracks indexer cursor for resumable event processing
-model IndexerState {
-  id         String   @id @default("singleton")
-  lastLedger Int      @default(0)
-  updatedAt  DateTime @updatedAt
 }

backend/src/app.ts

@@ -16,6 +16,7 @@ import { requestIdMiddleware } from "./middleware/requestId.js";
 import v1Routes from "./routes/v1/index.js";
 
 import healthRoutes from "./routes/health.routes.js";
+import "./lib/stream-id.js";
 
 const app = express();
 const isProduction = process.env.NODE_ENV === "production";

backend/src/controllers/sse.controller.ts

@@ -51,7 +51,7 @@ export const subscribe = async (req: Request, res: Response) => {
       where: { OR: [{ sender: publicKey }, { recipient: publicKey }] },
       select: { streamId: true, sender: true, recipient: true },
     });
-    const ownedIds = new Set(ownedStreams.map((s: { streamId: number }) => String(s.streamId)));
+    const ownedIds = new Set(ownedStreams.map((s: { streamId: bigint }) => String(s.streamId)));
     const allowedUserKeys = new Set<string>([publicKey]);
     for (const stream of ownedStreams) {
       allowedUserKeys.add(stream.sender);

backend/src/controllers/stream.controller.ts

@@ -13,6 +13,7 @@ import {
   resumeStream as sorobanResumeStream,
 } from "../services/sorobanService.js";
 import type { AuthenticatedRequest } from "../types/auth.types.js";
+import { parseStreamId } from "../lib/stream-id.js";
 import {
   DEFAULT_EVENTS_PAGE_SIZE,
   MAX_EVENTS_PAGE_SIZE,
@@ -128,10 +129,10 @@ export const createStream = async (req: Request, res: Response) => {
       });
     }
 
-    const parsedStreamId = Number.parseInt(streamId, 10);
+    const parsedStreamId = parseStreamId(streamId);
     const parsedStartTime = Number.parseInt(startTime, 10);
 
-    if (!Number.isFinite(parsedStreamId)) {
+    if (parsedStreamId === null) {
       return res
         .status(400)
         .json({ error: "Invalid streamId: must be a valid integer" });
@@ -347,9 +348,8 @@ export const getStream = async (req: Request, res: Response) => {
     const streamIdParam = Array.isArray(req.params.streamId)
       ? req.params.streamId[0]
       : req.params.streamId;
-    const parsedStreamId = Number.parseInt(streamIdParam ?? "", 10);
-
-    if (!Number.isFinite(parsedStreamId)) {
+    const parsedStreamId = parseStreamId(streamIdParam);
+    if (parsedStreamId === null) {
       return res.status(400).json({ error: "Invalid streamId parameter" });
     }
 
@@ -398,9 +398,8 @@ export const getStreamEvents = async (req: Request, res: Response) => {
     const streamIdParam = Array.isArray(req.params.streamId)
       ? req.params.streamId[0]
       : req.params.streamId;
-    const parsedStreamId = Number.parseInt(streamIdParam ?? "", 10);
-
-    if (!Number.isFinite(parsedStreamId)) {
+    const parsedStreamId = parseStreamId(streamIdParam);
+    if (parsedStreamId === null) {
       return res.status(400).json({ error: "Invalid streamId parameter" });
     }
 
@@ -489,9 +488,8 @@ export const getStreamClaimableAmount = async (req: Request, res: Response) => {
     const streamIdParam = Array.isArray(req.params.streamId)
       ? req.params.streamId[0]
       : req.params.streamId;
-    const parsedStreamId = Number.parseInt(streamIdParam ?? "", 10);
-
-    if (!Number.isFinite(parsedStreamId)) {
+    const parsedStreamId = parseStreamId(streamIdParam);
+    if (parsedStreamId === null) {
       return res.status(400).json({ error: "Invalid streamId parameter" });
     }
 
@@ -686,13 +684,12 @@ const topUpBodySchema = z.object({
  * Adds tokens to a running stream. Only the stream sender may call this.
  */
 export const topUpStreamHandler = async (req: Request, res: Response) => {
-  const streamId = parseInt(
+  const streamId = parseStreamId(
     Array.isArray(req.params.streamId)
-      ? req.params.streamId[0]!
-      : (req.params.streamId ?? ""),
-    10,
+      ? req.params.streamId[0]
+      : req.params.streamId,
   );
-  if (isNaN(streamId)) {
+  if (streamId === null) {
     return res.status(400).json({ error: "Invalid streamId" });
   }
 
@@ -769,9 +766,8 @@ export const pauseStream = async (req: Request, res: Response) => {
     const streamIdParam = Array.isArray(req.params.streamId)
       ? req.params.streamId[0]
       : req.params.streamId;
-    const parsedStreamId = Number.parseInt(streamIdParam ?? "", 10);
-
-    if (!Number.isFinite(parsedStreamId)) {
+    const parsedStreamId = parseStreamId(streamIdParam);
+    if (parsedStreamId === null) {
       return res.status(400).json({ error: "Invalid streamId parameter" });
     }
 
@@ -861,9 +857,8 @@ export const resumeStream = async (req: Request, res: Response) => {
     const streamIdParam = Array.isArray(req.params.streamId)
       ? req.params.streamId[0]
       : req.params.streamId;
-    const parsedStreamId = Number.parseInt(streamIdParam ?? "", 10);
-
-    if (!Number.isFinite(parsedStreamId)) {
+    const parsedStreamId = parseStreamId(streamIdParam);
+    if (parsedStreamId === null) {
       return res.status(400).json({ error: "Invalid streamId parameter" });
     }
 

backend/src/controllers/stream/cancel.ts

@@ -4,6 +4,7 @@ import logger from '../../logger.js';
 import * as sorobanService from '../../services/sorobanService.js';
 import type { AuthenticatedRequest } from '../../types/auth.types.js';
 import * as streamRepository from '../../repositories/stream.repository.js';
+import { parseStreamId } from '../../lib/stream-id.js';
 
 /**
  * @openapi
@@ -55,7 +56,10 @@ export const cancelStreamHandler = async (req: AuthenticatedRequest, res: Respon
       return res.status(400).json({ error: 'Missing streamId parameter' });
     }
 
-    const parsedStreamId = parseInt(streamId, 10);
+    const parsedStreamId = parseStreamId(streamId);
+    if (parsedStreamId === null) {
+      return res.status(400).json({ error: 'Invalid streamId parameter' });
+    }
 
     // 1. Fetch stream from DB
     const stream = await prisma.stream.findUnique({

backend/src/lib/stream-id.ts

@@ -0,0 +1,53 @@
+/**
+ * Helpers for on-chain stream IDs (Soroban u64).
+ *
+ * DB columns are Prisma BigInt / Postgres bigint so values above int4 max
+ * (2_147_483_647) round-trip without overflow. Prefer bigint in application
+ * code; never use Number()/parseInt for identifiers that may exceed 2^53-1.
+ */
+
+const U64_DECIMAL = /^\d+$/;
+
+/**
+ * Parse a path/query/body streamId into a bigint.
+ * Accepts decimal strings and non-negative integers / bigints.
+ */
+export function parseStreamId(raw: unknown): bigint | null {
+  if (typeof raw === 'bigint') {
+    return raw >= 0n ? raw : null;
+  }
+  if (typeof raw === 'number') {
+    if (!Number.isInteger(raw) || raw < 0 || !Number.isSafeInteger(raw)) {
+      return null;
+    }
+    return BigInt(raw);
+  }
+  if (typeof raw === 'string') {
+    const trimmed = raw.trim();
+    if (!U64_DECIMAL.test(trimmed)) return null;
+    try {
+      return BigInt(trimmed);
+    } catch {
+      return null;
+    }
+  }
+  return null;
+}
+
+/**
+ * JSON-safe encoding: numbers stay numbers within Number.MAX_SAFE_INTEGER
+ * (covers all historical int4 IDs and the >2^31 cases the bug cares about
+ * until 2^53); larger u64 values become decimal strings.
+ */
+export function streamIdToJson(streamId: bigint): number | string {
+  const asNumber = Number(streamId);
+  return Number.isSafeInteger(asNumber) ? asNumber : streamId.toString();
+}
+
+// Ensure Express / SSE JSON.stringify can serialize Prisma BigInt fields.
+const bigIntProto = BigInt.prototype as unknown as { toJSON?: () => number | string };
+if (typeof bigIntProto.toJSON !== 'function') {
+  bigIntProto.toJSON = function bigIntToJSON(this: bigint) {
+    return streamIdToJson(this);
+  };
+}

backend/src/repositories/stream.repository.ts

@@ -3,7 +3,7 @@ import { prisma } from '../lib/prisma.js';
 /**
  * Update the status and active flag of a stream in the database.
  */
-export const updateStatus = async (streamId: number, status: 'ACTIVE' | 'CANCELLED' | 'COMPLETED' | 'PAUSED') => {
+export const updateStatus = async (streamId: bigint, status: 'ACTIVE' | 'CANCELLED' | 'COMPLETED' | 'PAUSED') => {
   return prisma.stream.update({
     where: { streamId },
     data: {

backend/src/routes/v1/streams/withdraw.ts

@@ -4,6 +4,7 @@ import logger from '../../../logger.js';
 import { claimableAmountService } from '../../../services/claimable.service.js';
 import { withdraw as sorobanWithdraw } from '../../../services/sorobanService.js';
 import type { AuthenticatedRequest } from '../../../types/auth.types.js';
+import { parseStreamId } from '../../../lib/stream-id.js';
 
 /**
  * @openapi
@@ -56,9 +57,9 @@ export const withdrawHandler = async (req: AuthenticatedRequest, res: Response)
     const streamIdParam = Array.isArray(req.params.streamId)
       ? req.params.streamId[0]
       : req.params.streamId;
-    const parsedStreamId = Number.parseInt(streamIdParam ?? '', 10);
+    const parsedStreamId = parseStreamId(streamIdParam);
 
-    if (!Number.isFinite(parsedStreamId)) {
+    if (parsedStreamId === null) {
       return res.status(400).json({ error: 'Invalid streamId parameter' });
     }
 

backend/src/services/claimable.service.ts

@@ -4,7 +4,7 @@ const I128_MAX = (1n << 127n) - 1n;
 const I128_MIN = -(1n << 127n);
 
 export interface ClaimableStreamState {
-  streamId: number;
+  streamId: bigint;
   ratePerSecond: string;
   depositedAmount: string;
   withdrawnAmount: string;
@@ -18,7 +18,7 @@ export interface ClaimableStreamState {
 }
 
 export interface ClaimableAmountResult {
-  streamId: number;
+  streamId: bigint;
   claimableAmount: string;
   actionable: boolean;
   calculatedAt: number;

backend/src/services/soroban-indexer.service.ts

@@ -116,12 +116,18 @@ export class SorobanIndexerService {
     return null;
   }
 
-  private parseStreamId(record: JsonRecord): number | null {
+  private parseStreamId(record: JsonRecord): bigint | null {
     const raw = record.stream_id ?? record.streamId;
-    if (typeof raw === 'number' && Number.isInteger(raw)) return raw;
-    if (typeof raw === 'string' && raw.trim()) {
-      const parsed = Number(raw);
-      if (Number.isInteger(parsed)) return parsed;
+    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;
   }

backend/src/services/sorobanService.ts

@@ -51,7 +51,7 @@ export function resetServer(): void {
 }
 
 export interface ChainStream {
-  streamId: number;
+  streamId: bigint;
   sender: string;
   recipient: string;
   tokenAddress: string;
@@ -157,7 +157,7 @@ export async function submitContractCall(method: string, args: xdr.ScVal[], send
   return response.hash;
 }
 
-export async function getStreamFromChain(streamId: number): Promise<ChainStream | null> {
+export async function getStreamFromChain(streamId: bigint): Promise<ChainStream | null> {
   if (!getContractId()) return null;
 
   try {
@@ -189,7 +189,7 @@ export async function getStreamFromChain(streamId: number): Promise<ChainStream
   }
 }
 
-export async function getClaimableFromChain(streamId: number): Promise<string | null> {
+export async function getClaimableFromChain(streamId: bigint): Promise<string | null> {
   if (!getContractId()) return null;
 
   try {
@@ -204,13 +204,13 @@ export async function getClaimableFromChain(streamId: number): Promise<string |
   }
 }
 
-export async function cancelStream(streamId: number, senderSecret: string): Promise<string> {
+export async function cancelStream(streamId: bigint, senderSecret: string): Promise<string> {
   return submitContractCall('cancel_stream', [
     nativeToScVal(streamId, { type: 'u64' }),
   ], senderSecret);
 }
 
-export async function topUpStream(streamId: number, amount: bigint, callerAddress: string): Promise<string> {
+export async function topUpStream(streamId: bigint, amount: bigint, callerAddress: string): Promise<string> {
   const keeperSecret = getKeeperSecret();
   if (!keeperSecret) throw new Error('KEEPER_SECRET_KEY not configured');
   return submitContractCall('top_up_stream', [
@@ -236,7 +236,7 @@ export interface PauseResumeResult {
  */
 export async function pauseStream(
   senderAddress: string,
-  streamId: number
+  streamId: bigint
 ): Promise<PauseResumeResult> {
   if (!getContractId()) {
     throw new Error('Stream contract ID not configured');
@@ -270,7 +270,7 @@ export async function pauseStream(
  */
 export async function resumeStream(
   senderAddress: string,
-  streamId: number
+  streamId: bigint
 ): Promise<PauseResumeResult> {
   if (!getContractId()) {
     throw new Error('Stream contract ID not configured');
@@ -302,7 +302,7 @@ export async function resumeStream(
  * matching the current pause/resume backend pattern.
  */
 export async function withdraw(
-  streamId: number,
+  streamId: bigint,
   recipientAddress: string,
 ): Promise<PauseResumeResult> {
   if (!getContractId()) {