Merge branch 'main' into feat/issues-1031-1030-1029-1028

Author: ogazboiz

backend/README.md

@@ -29,8 +29,27 @@ API_BASE_URL=https://api.staging.flowfi.io
 We use Prisma as our ORM to interact with PostgreSQL.
 
 - Schema is located at `prisma/schema.prisma`.
+- 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.
 
+### Seeding the database
+
+`prisma/seed.ts` populates the database with demo fixtures for local development. Run it with:
+
+```bash
+npm run prisma:seed
+```
+
+(this runs `prisma db seed`, which in turn runs `tsx prisma/seed.ts` as configured under the `prisma.seed` key in `package.json`.)
+
+The script is idempotent (it uses `upsert`/fixed IDs), so it's safe to run multiple times. It creates:
+
+- Two demo users, keyed by fixed Stellar testnet public keys — a sender (`GCM5WPR4DDR24FSAX5LIEM4J7AI3KOWJYANSXEPKYXCSZOTAYXE75AFN`) and a recipient (`GBJCHUKZMTFSLOMNC7P4TS4VJJBTCYL3XKSOLXAUJSD56C4LHND5TWUC`).
+- One demo `Stream` (`streamId: 101`) between those two users, using a fixed demo token address, with a sample rate/deposit amount and `isActive: true`.
+- One demo `StreamEvent` (`eventType: 'CREATED'`) attached to that stream, with sample transaction hash, ledger sequence, and metadata.
+
+These fixtures are intended purely for local development/demo purposes so the frontend has data to render out of the box; they are not used in automated tests.
+
 ## /v1 API
 
 All REST API endpoints are prefixed with `/v1`. Refer to the API Documentation in the root `README.md` and the `docs/` folder for versioning and authentication details.

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/controllers/user.controller.ts

@@ -5,6 +5,54 @@ import { registerUserSchema } from '../validators/user.validator.js';
 import type { AuthenticatedRequest } from '../types/auth.types.js';
 import { DEFAULT_EVENTS_PAGE_SIZE, MAX_EVENTS_PAGE_SIZE } from '../routes/v1/events.routes.js';
 
+/**
+ * Public shape of a Stream, used when embedding streams inside a public
+ * user response. Excludes nothing sensitive today, but is kept explicit
+ * so newly added internal-only fields on the Stream model are not
+ * leaked automatically.
+ */
+const publicStreamSelect = {
+    id: true,
+    streamId: true,
+    sender: true,
+    recipient: true,
+    tokenAddress: true,
+    ratePerSecond: true,
+    depositedAmount: true,
+    withdrawnAmount: true,
+    startTime: true,
+    lastUpdateTime: true,
+    endTime: true,
+    isActive: true,
+    isPaused: true,
+    pausedAt: true,
+    totalPausedDuration: true,
+    createdAt: true,
+    updatedAt: true,
+} as const;
+
+/**
+ * Public shape of a User. Limits the response to fields that are safe to
+ * expose to any caller, so internal-only fields added to the User model
+ * later are excluded by default rather than leaked automatically.
+ */
+const publicUserSelect = {
+    id: true,
+    publicKey: true,
+    createdAt: true,
+    updatedAt: true,
+    sentStreams: {
+        take: 10,
+        orderBy: { createdAt: 'desc' as const },
+        select: publicStreamSelect,
+    },
+    receivedStreams: {
+        take: 10,
+        orderBy: { createdAt: 'desc' as const },
+        select: publicStreamSelect,
+    },
+};
+
 /**
  * Register a new wallet public key
  */
@@ -49,16 +97,7 @@ export const getUser = async (req: Request, res: Response, next: NextFunction) =
 
         const user = await prisma.user.findUnique({
             where: { publicKey },
-            include: {
-                sentStreams: {
-                    take: 10,
-                    orderBy: { createdAt: 'desc' }
-                },
-                receivedStreams: {
-                    take: 10,
-                    orderBy: { createdAt: 'desc' }
-                }
-            }
+            select: publicUserSelect
         });
 
         if (!user) {

backend/src/lib/indexer-state.ts

@@ -1 +1,59 @@
+import { prisma } from './prisma.js';
+import logger from '../logger.js';
+
 export const INDEXER_STATE_ID = 'singleton';
+
+export interface IndexerStateRow {
+  id: string;
+  lastLedger: number;
+  lastCursor: string | null;
+  createdAt: Date;
+  updatedAt: Date;
+}
+
+/**
+ * Ensure the singleton indexer_state row exists.
+ * Uses a catch-and-retry pattern to handle the race condition where two
+ * concurrent callers attempt the first insert simultaneously. If the
+ * unique-constraint violation fires, we treat it as success and re-read.
+ */
+export async function ensureIndexerState(
+  startLedger: number,
+): Promise<IndexerStateRow> {
+  const existing = await prisma.indexerState.findUnique({
+    where: { id: INDEXER_STATE_ID },
+  });
+  if (existing) return existing;
+
+  try {
+    const created = await prisma.indexerState.create({
+      data: {
+        id: INDEXER_STATE_ID,
+        lastLedger: startLedger,
+        lastCursor: null,
+      },
+    });
+    return created;
+  } catch (err: unknown) {
+    // P2002 = Prisma unique-constraint violation (code "P2002")
+    if (
+      err instanceof Error &&
+      'code' in err &&
+      (err as { code: string }).code === 'P2002'
+    ) {
+      logger.warn(
+        '[IndexerState] Concurrent first-insert detected; re-reading existing row.',
+      );
+      const existingAfterRace = await prisma.indexerState.findUnique({
+        where: { id: INDEXER_STATE_ID },
+      });
+      if (!existingAfterRace) {
+        throw new Error(
+          '[IndexerState] Unique-constraint violation but row not found after race.',
+        );
+      }
+      return existingAfterRace;
+    }
+    throw err;
+  }
+}