fix(backend): cap unbounded findMany queries in summary and SSE subsc…

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,

backend/tests/sse.controller.test.ts

@@ -122,4 +122,31 @@ describe('SSE Controller', () => {
     expect(subscriptions).toContain('user:GCOUNTER');
     expect(subscriptions).not.toContain('user:GOTHER');
   });
+
+  it('should cap owned streams to MAX_SSE_STREAMS on reconnect (Issue #1246)', async () => {
+    (sseService.isShuttingDown as any).mockReturnValue(false);
+    (sseService.checkCapacity as any).mockReturnValue({ allowed: true });
+    (req as any).user = { publicKey: 'GUSER1' };
+
+    // Simulate a wallet with more streams than the cap
+    const manyStreams = Array.from({ length: 600 }, (_, i) => ({
+      streamId: String(i),
+      sender: 'GUSER1',
+      recipient: `GOTHER${i}`,
+    }));
+
+    // The controller should request take: MAX_SSE_STREAMS (500)
+    const cappedStreams = manyStreams.slice(0, 500);
+    (prisma.stream.findMany as any).mockResolvedValue(cappedStreams);
+
+    await subscribe(req as Request, res as Response);
+
+    // Verify the query was bounded
+    const findManyCall = (prisma.stream.findMany as any).mock.calls[0];
+    expect(findManyCall[0].take).toBe(500);
+
+    // The subscriptions should only contain the capped set + user subscription
+    const subscriptions = (sseService.addClient as any).mock.calls[0][2] as string[];
+    expect(subscriptions.length).toBeLessThanOrEqual(501); // 500 streams + user:GUSER1
+  });
 });

backend/tests/stream.controller.test.ts

@@ -1,5 +1,5 @@
 import { describe, it, expect, vi, beforeEach } from 'vitest';
-import { createStream, listStreams, getStream, getStreamEvents, getStreamClaimableAmount, pauseStream, resumeStream } from '../src/controllers/stream.controller.js';
+import { createStream, listStreams, getStream, getStreamEvents, getStreamClaimableAmount, getUserStreamSummary, pauseStream, resumeStream, MAX_USER_STREAMS } from '../src/controllers/stream.controller.js';
 import { prisma } from '../src/lib/prisma.js';
 import { claimableAmountService } from '../src/services/claimable.service.js';
 import * as sorobanService from '../src/services/sorobanService.js';
@@ -283,6 +283,121 @@ describe("Stream Controller", () => {
     });
   });
 
+  describe("getUserStreamSummary", () => {
+    it("should return 400 when address is missing", async () => {
+      req.params = {} as any;
+
+      await getUserStreamSummary(req as any, res as Response);
+
+      expect(res.status).toHaveBeenCalledWith(400);
+    });
+
+    it("should cap outgoing and incoming streams to MAX_USER_STREAMS (Issue #1246)", async () => {
+      req.params = { address: "GUSER1" };
+
+      // Simulate a wallet with more streams than the cap
+      const manyStreams = Array.from({ length: MAX_USER_STREAMS + 100 }, (_, i) => ({
+        streamId: i,
+        ratePerSecond: "10",
+        depositedAmount: "1000",
+        withdrawnAmount: "500",
+        startTime: BigInt(1000 + i),
+        lastUpdateTime: BigInt(2000 + i),
+        isActive: true,
+        isPaused: false,
+        pausedAt: null,
+        totalPausedDuration: null,
+        updatedAt: new Date(),
+      }));
+
+      // findMany is called twice (outgoing + incoming), each returns at most MAX_USER_STREAMS
+      const cappedStreams = manyStreams.slice(0, MAX_USER_STREAMS);
+      (prisma.stream.findMany as any)
+        .mockResolvedValueOnce(cappedStreams)  // outgoing
+        .mockResolvedValueOnce(cappedStreams); // incoming
+
+      (claimableAmountService.getClaimableAmount as any).mockReturnValue({
+        claimableAmount: "0",
+      });
+
+      await getUserStreamSummary(req as any, res as Response);
+
+      expect(res.status).toHaveBeenCalledWith(200);
+      const body = (res.json as any).mock.calls[0][0];
+      // The response should reflect only the capped result set
+      expect(body.totalStreamsCreated).toBe(MAX_USER_STREAMS);
+      // Both findMany calls should have received take: MAX_USER_STREAMS
+      const findManyCalls = (prisma.stream.findMany as any).mock.calls;
+      for (const call of findManyCalls) {
+        expect(call[0].take).toBe(MAX_USER_STREAMS);
+      }
+    });
+
+    it("should not set truncated flag when under the cap (Issue #1246)", async () => {
+      req.params = { address: "GUSER_NO_TRUNC" };
+
+      const fewStreams = Array.from({ length: 5 }, (_, i) => ({
+        streamId: i,
+        ratePerSecond: "10",
+        depositedAmount: "1000",
+        withdrawnAmount: "500",
+        startTime: BigInt(1000 + i),
+        lastUpdateTime: BigInt(2000 + i),
+        isActive: true,
+        isPaused: false,
+        pausedAt: null,
+        totalPausedDuration: null,
+        updatedAt: new Date(),
+      }));
+
+      (prisma.stream.findMany as any)
+        .mockResolvedValueOnce(fewStreams)
+        .mockResolvedValueOnce(fewStreams);
+
+      (claimableAmountService.getClaimableAmount as any).mockReturnValue({
+        claimableAmount: "0",
+      });
+
+      await getUserStreamSummary(req as any, res as Response);
+
+      const body = (res.json as any).mock.calls[0][0];
+      expect(body.truncated).toBeUndefined();
+    });
+
+    it("should set truncated=true when either direction hits the cap (Issue #1246)", async () => {
+      req.params = { address: "GUSER_TRUNC" };
+
+      const atCap = Array.from({ length: MAX_USER_STREAMS }, (_, i) => ({
+        streamId: i,
+        ratePerSecond: "10",
+        depositedAmount: "1000",
+        withdrawnAmount: "0",
+        startTime: BigInt(1000 + i),
+        lastUpdateTime: BigInt(2000 + i),
+        isActive: true,
+        isPaused: false,
+        pausedAt: null,
+        totalPausedDuration: null,
+        updatedAt: new Date(),
+      }));
+      const empty: any[] = [];
+
+      // Outgoing hits cap, incoming is empty
+      (prisma.stream.findMany as any)
+        .mockResolvedValueOnce(atCap)
+        .mockResolvedValueOnce(empty);
+
+      (claimableAmountService.getClaimableAmount as any).mockReturnValue({
+        claimableAmount: "0",
+      });
+
+      await getUserStreamSummary(req as any, res as Response);
+
+      const body = (res.json as any).mock.calls[0][0];
+      expect(body.truncated).toBe(true);
+    });
+  });
+
   describe("pauseStream", () => {
     it("should pause stream", async () => {
       req.params = { streamId: "123" };