Merge branch 'main' into fix/wallet-connect-freighter-error-tests

Author: Seunfunmi-319509

.github/workflows/ci.yml

@@ -192,8 +192,16 @@ jobs:
         run: cargo test
         working-directory: contracts
 
+      - name: Cache cargo-tarpaulin
+        id: tarpaulin-cache
+        uses: actions/cache@v4
+        with:
+          path: ~/.cargo/bin/cargo-tarpaulin
+          key: cargo-tarpaulin-${{ runner.os }}-0.37.2
+
       - name: Install cargo-tarpaulin
-        run: cargo install cargo-tarpaulin --locked
+        if: steps.tarpaulin-cache.outputs.cache-hit != 'true'
+        run: cargo install cargo-tarpaulin@0.37.2 --locked
 
       - name: Run Contract Coverage
         run: cargo tarpaulin --workspace --out Xml --output-dir coverage --fail-under 70

.github/workflows/security.yml

@@ -8,7 +8,7 @@ on:
   push:
     branches: [ main, develop ]
   pull_request:
-    branches: [ main ]
+    branches: [ main, develop ]
   schedule:
     - cron: '0 2 * * 0'
 
@@ -41,6 +41,24 @@ jobs:
     - name: Check for known vulnerabilities in backend (production dependencies)
       run: npm audit --workspace=backend --omit=dev --audit-level=critical
 
+    - name: Setup Rust toolchain for contract audit
+      uses: dtolnay/rust-toolchain@stable
+
+    - name: Cache cargo-audit
+      id: cargo-audit-cache
+      uses: actions/cache@v4
+      with:
+        path: ~/.cargo/bin/cargo-audit
+        key: cargo-audit-${{ runner.os }}
+
+    - name: Install cargo-audit
+      if: steps.cargo-audit-cache.outputs.cache-hit != 'true'
+      run: cargo install cargo-audit --locked
+
+    - name: Check for known vulnerabilities in smart contracts (cargo audit)
+      run: cargo audit
+      working-directory: contracts
+
     - name: Verify security setup
       run: npm run verify-security
 
@@ -55,19 +73,40 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        language: [ 'javascript', 'typescript' ]
+        language: [ 'javascript', 'typescript', 'rust' ]
 
     steps:
     - name: Checkout repository
       uses: actions/checkout@v4
 
+    - name: Setup Rust toolchain
+      if: matrix.language == 'rust'
+      uses: dtolnay/rust-toolchain@stable
+      with:
+        toolchain: stable
+        targets: wasm32-unknown-unknown
+        components: clippy
+
+    - name: Rust Cache
+      if: matrix.language == 'rust'
+      uses: Swatinem/rust-cache@v2
+      with:
+        workspace: "contracts -> target"
+
     - name: Initialize CodeQL
       uses: github/codeql-action/init@v3
       with:
         languages: ${{ matrix.language }}
 
+    - name: Build Rust contracts for CodeQL
+      if: matrix.language == 'rust'
+      run: cargo check --workspace --all-targets
+      working-directory: contracts
+
     - name: Autobuild
+      if: matrix.language != 'rust'
       uses: github/codeql-action/autobuild@v3
 
     - name: Perform CodeQL Analysis
       uses: github/codeql-action/analyze@v3
+

backend/.env.example

@@ -88,6 +88,11 @@ REDIS_URL=
 # Time in milliseconds between periodic sweeps to prune expired memory cache entries (default: 60000)
 MEMORY_CACHE_SWEEP_MS=60000
 
+# Maximum number of entries kept in the in-memory cache. When exceeded, the
+# least-recently-used entries are evicted immediately so memory stays bounded
+# even between sweeps (default: 10000)
+MEMORY_CACHE_MAX_ITEMS=10000
+
 # Cache time-to-live (TTL) for claimable amount calculations in milliseconds (default: 5000)
 CLAIMABLE_CACHE_TTL_MS=5000
 

backend/docs/SSE_ARCHITECTURE.md

@@ -3,27 +3,27 @@
 ## System Flow
 
 ```
-┌─────────────────┐
-│  Blockchain     │
-│  Indexer        │
-│  (Stellar)      │
-└────────┬────────┘
-         │ Events
-         ▼
+┌─────────────────────────────────────────────────────────┐
+│              Stellar Blockchain / Soroban                │
+│  - On-chain stream contract executions & ledger events   │
+└────────────────────────────┬────────────────────────────┘
+                             │ On-Chain Events (via Soroban RPC poll)
+                             ▼
 ┌─────────────────────────────────────────────────────────┐
 │                    Backend Server                        │
 │                                                          │
 │  ┌──────────────────────────────────────────────────┐  │
-│  │  Stream Controller                                │  │
-│  │  - Creates/updates streams                        │  │
-│  │  - Calls sseService.broadcast()                   │  │
+│  │  Soroban Event Worker (Indexer)                  │  │
+│  │  - Polls Soroban RPC for confirmed events        │  │
+│  │  - Persists stream state & events to Database    │  │
+│  │  - Calls sseService.broadcastToStream/Admin()     │  │
 │  └──────────────┬───────────────────────────────────┘  │
-│                 │                                        │
+│                 │ Dispatch events (indexer-driven)       │
 │                 ▼                                        │
 │  ┌──────────────────────────────────────────────────┐  │
 │  │  SSE Service                                      │  │
 │  │  - Manages client connections                     │  │
-│  │  - Filters by subscription                        │  │
+│  │  - Filters by subscription (stream/user/admin)    │  │
 │  │  - Broadcasts to matching clients                 │  │
 │  └──────────────┬───────────────────────────────────┘  │
 │                 │                                        │
@@ -40,6 +40,8 @@
     └─────────────────────────────────────┘
 ```
 
+> **Note on Indexer-Driven Event Origin**: SSE broadcast events originate asynchronously from the background indexer worker (`SorobanEventWorker`) only after transaction confirmation on the Stellar ledger, not synchronously from HTTP API controllers (`stream.controller.ts`, etc.). When a user submits an action (create, pause, withdraw, top-up, cancel), API controllers do not broadcast SSE events directly; events are dispatched once the Soroban event is polled and confirmed on-chain. Additionally, background workers like `StreamRunwayWorker` may dispatch computed alerts (e.g. `STREAM_LOW_BALANCE`).
+
 ## Connection Flow
 
 ```
@@ -72,7 +74,7 @@ Client                          Server
   │  [Auto Reconnect - 1s]        │
   │                               │
   │  GET /events/subscribe        │
-  ├──────────────────────────────>│
+  │  ├──────────────────────────────>│
   │                               │
   │  200 OK                       │
   │<──────────────────────────────┤
@@ -122,30 +124,37 @@ Client                          Server
                     └─────────────────┘
 
 Flow:
-1. Backend 1 receives stream creation
+1. Backend 1 (SorobanEventWorker) indexes confirmed on-chain event
 2. Backend 1 publishes to Redis: "stream-events"
-3. All backends (1, 2, 3) receive message
+3. All backends (1, 2, 3) receive message via Redis subscriber
 4. Each backend broadcasts to its connected clients
 5. Total: 33 clients receive the event
 ```
 
 ## Event Broadcasting Logic
 
 ```typescript
-// Broadcast to specific stream
+// Broadcast to specific stream (called by SorobanEventWorker / StreamRunwayWorker)
 sseService.broadcastToStream("123", "stream.created", data)
   ↓
   Filter clients: subscription includes "123" or "*"
   ↓
   Send to matching clients
 
-// Broadcast to user
-sseService.broadcastToUser("GABC...", "stream.created", data)
+// Broadcast to user (called by StreamRunwayWorker)
+sseService.broadcastToUser("GABC...", "STREAM_LOW_BALANCE", data)
   ↓
   Filter clients: subscription includes "user:GABC..." or "*"
   ↓
   Send to matching clients
 
+// Broadcast to admin (called by SorobanEventWorker for admin/fee events)
+sseService.broadcastToAdmin("stream.fee_config_updated", data)
+  ↓
+  Filter clients: admin subscribers ("admin" or "*")
+  ↓
+  Send to matching clients
+
 // Broadcast to all
 sseService.broadcast("stream.created", data)
   ↓

backend/docs/SSE_IMPLEMENTATION.md

@@ -193,7 +193,7 @@ import Redis from 'ioredis';
 const redis = new Redis(process.env.REDIS_URL);
 const subscriber = new Redis(process.env.REDIS_URL);
 
-// Publisher (in stream controller)
+// Publisher (in Soroban event worker / indexer)
 redis.publish('stream-events', JSON.stringify({
   event: 'stream.created',
   data: mockStream,

backend/package.json

@@ -6,6 +6,7 @@
   "main": "index.js",
   "scripts": {
     "prebuild": "prisma generate",
+    "pretest": "prisma generate",
     "test": "vitest run",
     "test:unit": "vitest run --exclude='tests/integration/**'",
     "test:integration": "vitest run tests/integration",

backend/prisma/migrations/20260830000000_remove_redundant_streamid_index/migration.sql

@@ -0,0 +1,2 @@
+-- DropIndex
+DROP INDEX IF EXISTS "Stream_streamId_idx";

backend/prisma/schema.prisma

@@ -52,7 +52,6 @@ model Stream {
   @@index([sender])
   @@index([recipient])
   @@index([tokenAddress])
-  @@index([streamId])
   @@index([isActive])
   @@index([isPaused])
 }

backend/src/controllers/sse.controller.ts

@@ -11,6 +11,14 @@ const subscribeSchema = z.object({
   all: z.boolean().optional().default(false),
 });
 
+/**
+ * Issue #1246: hard cap on the number of streams fetched per SSE session.
+ * Prevents unbounded DB queries when a wallet has thousands of streams.
+ * Users who exceed this cap still receive events for the most recent streams;
+ * the frontend can fall back to polling or pagination for the remainder.
+ */
+const MAX_SSE_STREAMS = 500;
+
 
 function getClientIp(req: Request): string {
   const forwarded = req.headers['x-forwarded-for'];
@@ -48,8 +56,13 @@ export const subscribe = async (req: Request, res: Response) => {
     // Consistent with GET /v1/events/ (which requires requireAuth and is scoped to user's address),
     // SSE subscriptions are also restricted to streams owned by the authenticated user.
     // Scope: only streams where the authenticated user is sender or recipient
+    // Issue #1246: cap the query to prevent unbounded fetches on reconnect.
+    // Ordered by startTime desc so the most recent streams are always included
+    // when the cap is reached.
     const ownedStreams = await prisma.stream.findMany({
       where: { OR: [{ sender: publicKey }, { recipient: publicKey }] },
+      orderBy: { startTime: "desc" },
+      take: MAX_SSE_STREAMS,
       select: { streamId: true, sender: true, recipient: true },
     });
     const ownedIds = new Set(ownedStreams.map((s: { streamId: bigint }) => String(s.streamId)));

backend/src/controllers/stream.controller.ts

@@ -22,6 +22,15 @@ import {
 const DEFAULT_STREAM_PAGE_SIZE = 20;
 const MAX_STREAM_PAGE_SIZE = 100;
 
+/**
+ * Hard cap on the number of streams fetched per user in the summary endpoint.
+ * Prevents unbounded DB queries when a wallet has thousands of streams.
+ * Users who exceed this cap receive a truncated summary (counts and totals
+ * reflect only the most recent streams) plus a `truncated` flag so the
+ * frontend can offer a pagination or export fallback.
+ */
+export const MAX_USER_STREAMS = 500;
+
 interface UserStreamSummary {
   address: string;
   totalStreamsCreated: number;
@@ -590,9 +599,15 @@ export const getUserStreamSummary = async (
 
     pruneUserSummaryCache(nowMs);
 
+    // Issue #1246: cap the number of streams fetched per direction to prevent
+    // unbounded DB queries.  Power users with more than MAX_USER_STREAMS
+    // streams receive a truncated summary (the `truncated` flag lets the
+    // frontend offer a pagination/export fallback).
     const [outgoingStreams, incomingStreams] = await Promise.all([
       prisma.stream.findMany({
         where: { sender: address },
+        orderBy: { startTime: "desc" },
+        take: MAX_USER_STREAMS,
         select: {
           streamId: true,
           ratePerSecond: true,
@@ -609,6 +624,8 @@ export const getUserStreamSummary = async (
       }),
       prisma.stream.findMany({
         where: { recipient: address },
+        orderBy: { startTime: "desc" },
+        take: MAX_USER_STREAMS,
         select: {
           streamId: true,
           ratePerSecond: true,
@@ -651,15 +668,20 @@ export const getUserStreamSummary = async (
       (stream: any) => stream.isActive,
     ).length;
 
-    const summary: UserStreamSummary = {
+    const truncated =
+      outgoingStreams.length >= MAX_USER_STREAMS ||
+      incomingStreams.length >= MAX_USER_STREAMS;
+
+    const summary = {
       address,
       totalStreamsCreated,
       totalStreamedOut,
       totalStreamedIn,
       currentClaimable: claimableInTotal.toString(),
       activeOutgoingCount,
       activeIncomingCount,
-    };
+      ...(truncated ? { truncated: true } : {}),
+    } satisfies UserStreamSummary & { truncated?: boolean };
 
     userSummaryCache.set(cacheKey, {
       value: summary,
@@ -730,19 +752,24 @@ export const topUpStreamHandler = async (req: Request, res: Response) => {
 
     const txHash = await topUpStream(streamId, amount, callerAddress);
 
-    const newDeposited = (BigInt(stream.depositedAmount) + amount).toString();
-    await prisma.stream.update({
-      where: { streamId },
-      data: {
-        depositedAmount: newDeposited,
-        lastUpdateTime: BigInt(Math.floor(Date.now() / 1000)),
-      },
-    });
+    // Use raw SQL atomic increment to prevent concurrent top-ups from
+    // overwriting each other's updates (Issue #1217 — read-compute-write race).
+    // Prisma's built-in { increment } is unavailable on String-typed columns,
+    // so we perform SET deposited_amount = deposited_amount + $1::bigint
+    // directly in a single SQL statement.
+    const now = BigInt(Math.floor(Date.now() / 1000));
+    await prisma.$executeRawUnsafe(
+      `UPDATE "Stream" SET "depositedAmount" = ("depositedAmount"::bigint + $1::bigint)::text, "lastUpdateTime" = $2 WHERE "streamId" = $3`,
+      amount.toString(),
+      now,
+      streamId,
+    );
+    const updatedStream = await prisma.stream.findUnique({ where: { streamId } });
 
     logger.info(`[topUp] stream=${streamId} amount=${amount} txHash=${txHash}`);
     return res
       .status(200)
-      .json({ streamId, txHash, depositedAmount: newDeposited });
+      .json({ streamId, txHash, depositedAmount: updatedStream!.depositedAmount });
   } catch (error: any) {
     logger.error(`[topUp] stream=${streamId} error:`, error);
     return res.status(400).json({ error: 'Failed to top up stream on chain', message: error.message ?? 'Unknown error' });