diff --git a/backend/src/controllers/sse.controller.ts b/backend/src/controllers/sse.controller.ts index 8ca22fd6..ddc6e3cb 100644 --- a/backend/src/controllers/sse.controller.ts +++ b/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))); diff --git a/backend/src/controllers/stream.controller.ts b/backend/src/controllers/stream.controller.ts index 394c11ad..0ba2a383 100644 --- a/backend/src/controllers/stream.controller.ts +++ b/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,7 +668,11 @@ 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, @@ -659,7 +680,8 @@ export const getUserStreamSummary = async ( currentClaimable: claimableInTotal.toString(), activeOutgoingCount, activeIncomingCount, - }; + ...(truncated ? { truncated: true } : {}), + } satisfies UserStreamSummary & { truncated?: boolean }; userSummaryCache.set(cacheKey, { value: summary, diff --git a/backend/tests/sse.controller.test.ts b/backend/tests/sse.controller.test.ts index 3d31bab4..a65006b7 100644 --- a/backend/tests/sse.controller.test.ts +++ b/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 + }); }); diff --git a/backend/tests/stream.controller.test.ts b/backend/tests/stream.controller.test.ts index e53c9999..4471655d 100644 --- a/backend/tests/stream.controller.test.ts +++ b/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" };