Merge branch 'main' into docs/backend-env-example-completeness

Author: ogazboiz

backend/prisma/migrations/20260630120000_add_stream_event_stream_id_timestamp_index/migration.sql

@@ -0,0 +1,8 @@
+-- Replace the unused (streamId, createdAt) composite with (streamId, timestamp),
+-- which matches streamId-scoped event listings that ORDER BY timestamp.
+
+-- CreateIndex
+CREATE INDEX IF NOT EXISTS "StreamEvent_streamId_timestamp_idx" ON "StreamEvent"("streamId", "timestamp");
+
+-- DropIndex
+DROP INDEX IF EXISTS "StreamEvent_streamId_createdAt_idx";

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
@@ -86,5 +86,6 @@ model StreamEvent {
   @@index([timestamp])
   @@index([transactionHash])
   @@index([createdAt])
-  @@index([streamId, createdAt])
+  @@index([streamId, timestamp])
+  @@unique([transactionHash, eventType])
 }

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

@@ -31,7 +31,8 @@ export const subscribe = async (req: Request, res: Response) => {
 
   try {
     const sourceIp = getClientIp(req);
-    const capacity = sseService.checkCapacity(sourceIp);
+    const authUserId = (req as AuthenticatedRequest).user?.publicKey;
+    const capacity = sseService.checkCapacity(sourceIp, authUserId);
     if (!capacity.allowed) {
       if (capacity.retryAfterSeconds) {
         res.setHeader('Retry-After', String(capacity.retryAfterSeconds));
@@ -51,7 +52,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);
@@ -87,7 +88,7 @@ export const subscribe = async (req: Request, res: Response) => {
     const requestId = requestContext.getStore()?.requestId;
     res.write(`data: ${JSON.stringify({ type: 'connected', clientId, requestId })}\n\n`);
 
-    sseService.addClient(clientId, res, subscriptions, sourceIp);
+    sseService.addClient(clientId, res, subscriptions, sourceIp, publicKey);
     return;
   } catch (error: unknown) {
     if (error instanceof z.ZodError) {

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: {