Skip to content

Commit 282f54b

Browse files
Folex1275Folex1275
authored
Merge pull request #1179 from Hexstar-labs/fix/n1-query-stream-listing-1147
fix(backend): guard stream listing endpoints against N+1 queries (#1147)
2 parents 7ee3d64 + 686c981 commit 282f54b

7 files changed

Lines changed: 415 additions & 5 deletions

‎backend/src/__jest__/stream-listing-endpoints-query-count.test.ts‎

Lines changed: 168 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,168 @@
1+
import request from "supertest";
2+
import express, { Express } from "express";
3+
4+
jest.mock("../lib/db.js", () => ({
5+
prisma: {
6+
stream: {
7+
findMany: jest.fn(),
8+
count: jest.fn(),
9+
},
10+
eventLog: {
11+
findMany: jest.fn(),
12+
},
13+
},
14+
}));
15+
16+
// sanitize.ts uses `import.meta.url`, which ts-jest can't compile under the
17+
// CommonJS transform this project's jest.config.js forces. Stub it out so
18+
// importing streams.routes.ts (which depends on it) doesn't fail to compile.
19+
jest.mock("../security/sanitize.js", () => ({
20+
sanitizeUnknown: (input: unknown) => input,
21+
}));
22+
23+
import { prisma } from "../lib/db.js";
24+
import streamsRouter from "../api/streams.routes.js";
25+
import v2StreamsRouter from "../api/v2/streams.routes.js";
26+
import v3HistoryRouter from "../api/v3/history.routes.js";
27+
import { getSearch } from "../api/public.js";
28+
29+
const ADDRESS = "G" + "A".repeat(55);
30+
const MAX_ALLOWED_QUERIES = 5;
31+
32+
function buildMockStream(overrides: Partial<Record<string, unknown>> = {}) {
33+
return {
34+
id: `stream_${Math.random().toString(36).slice(2)}`,
35+
streamId: `contract_${Math.random().toString(36).slice(2)}`,
36+
txHash: `tx_${Math.random().toString(36).slice(2)}`,
37+
sender: ADDRESS,
38+
receiver: "GBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB",
39+
tokenAddress: "CUSDC",
40+
amount: "1000000000",
41+
duration: 86400,
42+
status: "ACTIVE",
43+
withdrawn: "0",
44+
legacy: false,
45+
migrated: false,
46+
isPrivate: false,
47+
createdAt: new Date(),
48+
...overrides,
49+
};
50+
}
51+
52+
function totalMockCalls(): number {
53+
return (
54+
(prisma.stream.findMany as jest.Mock).mock.calls.length +
55+
(prisma.stream.count as jest.Mock).mock.calls.length +
56+
(prisma.eventLog.findMany as jest.Mock).mock.calls.length
57+
);
58+
}
59+
60+
describe("Stream listing endpoints stay below the N+1 query budget", () => {
61+
beforeEach(() => {
62+
jest.clearAllMocks();
63+
});
64+
65+
describe.each([50, 1000])("with %d streams", (count) => {
66+
const mockStreams = () =>
67+
Array.from({ length: count }, () => buildMockStream());
68+
69+
it("GET /api/v1/streams/:address", async () => {
70+
const app: Express = express();
71+
app.use(express.json());
72+
app.use("/api/v1", streamsRouter);
73+
74+
(prisma.stream.findMany as jest.Mock).mockResolvedValueOnce(mockStreams());
75+
76+
const res = await request(app).get(`/api/v1/streams/${ADDRESS}`);
77+
78+
expect(res.status).toBe(200);
79+
expect(res.body.count).toBe(count);
80+
expect(totalMockCalls()).toBeLessThan(MAX_ALLOWED_QUERIES);
81+
});
82+
83+
it("GET /api/v1/streams/export/:address", async () => {
84+
const app: Express = express();
85+
app.use(express.json());
86+
app.use("/api/v1", streamsRouter);
87+
88+
const streams = mockStreams();
89+
(prisma.stream.findMany as jest.Mock).mockResolvedValueOnce(streams);
90+
(prisma.eventLog.findMany as jest.Mock).mockResolvedValueOnce([]);
91+
92+
const res = await request(app).get(`/api/v1/streams/export/${ADDRESS}`);
93+
94+
expect(res.status).toBe(200);
95+
expect(totalMockCalls()).toBeLessThan(MAX_ALLOWED_QUERIES);
96+
});
97+
98+
it("GET /api/v2/streams/:address", async () => {
99+
const app: Express = express();
100+
app.use(express.json());
101+
app.use("/api/v2/streams", v2StreamsRouter);
102+
103+
(prisma.stream.findMany as jest.Mock).mockResolvedValueOnce(mockStreams());
104+
105+
const res = await request(app).get(`/api/v2/streams/${ADDRESS}`);
106+
107+
expect(res.status).toBe(200);
108+
expect(res.body.v1).toHaveLength(count);
109+
expect(totalMockCalls()).toBeLessThan(MAX_ALLOWED_QUERIES);
110+
});
111+
112+
it("GET /api/v3/history/:address", async () => {
113+
const app: Express = express();
114+
app.use(express.json());
115+
app.use("/api/v3", v3HistoryRouter);
116+
117+
(prisma.stream.count as jest.Mock).mockResolvedValueOnce(count);
118+
(prisma.stream.findMany as jest.Mock).mockResolvedValueOnce(
119+
mockStreams().slice(0, 50),
120+
);
121+
122+
const res = await request(app).get(`/api/v3/history/${ADDRESS}`);
123+
124+
expect(res.status).toBe(200);
125+
expect(res.body.data.total).toBe(count);
126+
expect(totalMockCalls()).toBeLessThan(MAX_ALLOWED_QUERIES);
127+
});
128+
129+
it("GET /api/v1/search", async () => {
130+
const app: Express = express();
131+
app.use(express.json());
132+
app.get("/api/v1/search", getSearch);
133+
134+
(prisma.stream.findMany as jest.Mock).mockResolvedValueOnce(
135+
mockStreams().slice(0, 20),
136+
);
137+
(prisma.stream.count as jest.Mock).mockResolvedValueOnce(count);
138+
139+
const res = await request(app).get("/api/v1/search");
140+
141+
expect(res.status).toBe(200);
142+
expect(res.body.total).toBe(count);
143+
expect(totalMockCalls()).toBeLessThan(MAX_ALLOWED_QUERIES);
144+
});
145+
});
146+
147+
it("query count for 1000 streams is identical to the query count for 50 streams (no per-row queries)", async () => {
148+
const app: Express = express();
149+
app.use(express.json());
150+
app.use("/api/v1", streamsRouter);
151+
152+
(prisma.stream.findMany as jest.Mock).mockResolvedValueOnce(
153+
Array.from({ length: 50 }, () => buildMockStream()),
154+
);
155+
await request(app).get(`/api/v1/streams/${ADDRESS}`);
156+
const callsAt50 = totalMockCalls();
157+
158+
jest.clearAllMocks();
159+
160+
(prisma.stream.findMany as jest.Mock).mockResolvedValueOnce(
161+
Array.from({ length: 1000 }, () => buildMockStream()),
162+
);
163+
await request(app).get(`/api/v1/streams/${ADDRESS}`);
164+
const callsAt1000 = totalMockCalls();
165+
166+
expect(callsAt1000).toBe(callsAt50);
167+
});
168+
});

‎backend/src/__jest__/stream-listing-load.test.ts‎

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
import request from "supertest";
2+
import express, { Express } from "express";
3+
4+
jest.mock("../lib/db.js", () => ({
5+
prisma: {
6+
stream: {
7+
findMany: jest.fn(),
8+
count: jest.fn(),
9+
},
10+
eventLog: {
11+
findMany: jest.fn(),
12+
},
13+
},
14+
}));
15+
16+
// sanitize.ts uses `import.meta.url`, which ts-jest can't compile under the
17+
// CommonJS transform this project's jest.config.js forces. Stub it out so
18+
// importing streams.routes.ts (which depends on it) doesn't fail to compile.
19+
jest.mock("../security/sanitize.js", () => ({
20+
sanitizeUnknown: (input: unknown) => input,
21+
}));
22+
23+
import { prisma } from "../lib/db.js";
24+
import streamsRouter from "../api/streams.routes.js";
25+
26+
const ADDRESS = "G" + "A".repeat(55);
27+
28+
function buildMockStream(index: number) {
29+
return {
30+
id: `stream_${index}`,
31+
streamId: `contract_${index}`,
32+
txHash: `tx_${index}`,
33+
sender: ADDRESS,
34+
receiver: "GBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB",
35+
tokenAddress: "CUSDC",
36+
amount: "1000000000",
37+
duration: 86400,
38+
status: "ACTIVE",
39+
withdrawn: "0",
40+
legacy: false,
41+
migrated: false,
42+
isPrivate: false,
43+
createdAt: new Date(),
44+
};
45+
}
46+
47+
async function runExportFor(app: Express, streamCount: number): Promise<number> {
48+
jest.clearAllMocks();
49+
(prisma.stream.findMany as jest.Mock).mockResolvedValueOnce(
50+
Array.from({ length: streamCount }, (_, i) => buildMockStream(i)),
51+
);
52+
(prisma.eventLog.findMany as jest.Mock).mockResolvedValueOnce(
53+
Array.from({ length: streamCount }, (_, i) => ({
54+
streamId: `contract_${i}`,
55+
ledgerClosedAt: "2024-01-01T00:00:00Z",
56+
metadata: null,
57+
})),
58+
);
59+
60+
const start = Date.now();
61+
const res = await request(app).get(`/api/v1/streams/export/${ADDRESS}`);
62+
const elapsed = Date.now() - start;
63+
64+
expect(res.status).toBe(200);
65+
return elapsed;
66+
}
67+
68+
describe("Stream listing load test (1000 streams)", () => {
69+
it("handles 1000 streams in a single bounded pass (no per-row DB round-trips)", async () => {
70+
const app: Express = express();
71+
app.use(express.json());
72+
app.use("/api/v1", streamsRouter);
73+
74+
// Warm up the JIT/route stack so timing reflects steady-state behavior.
75+
await runExportFor(app, 50);
76+
77+
const elapsed = await runExportFor(app, 1000);
78+
79+
// With DB I/O mocked out, processing 1000 rows is pure in-memory work.
80+
// A real N+1 (or any O(n^2) pass) would blow well past this budget;
81+
// a single-pass batched implementation finishes in low single-digit ms.
82+
expect(elapsed).toBeLessThan(500);
83+
84+
expect(prisma.stream.findMany).toHaveBeenCalledTimes(1);
85+
expect(prisma.eventLog.findMany).toHaveBeenCalledTimes(1);
86+
});
87+
88+
it("query count for the export endpoint is identical at 50 and 1000 streams (linear scaling)", async () => {
89+
const app: Express = express();
90+
app.use(express.json());
91+
app.use("/api/v1", streamsRouter);
92+
93+
await runExportFor(app, 50);
94+
const callsAt50 =
95+
(prisma.stream.findMany as jest.Mock).mock.calls.length +
96+
(prisma.eventLog.findMany as jest.Mock).mock.calls.length;
97+
98+
await runExportFor(app, 1000);
99+
const callsAt1000 =
100+
(prisma.stream.findMany as jest.Mock).mock.calls.length +
101+
(prisma.eventLog.findMany as jest.Mock).mock.calls.length;
102+
103+
expect(callsAt1000).toBe(callsAt50);
104+
});
105+
});

‎backend/src/__jest__/stream-service-n1-query-count.test.ts‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
import { StreamService } from "../services/stream.service.js";
2+
3+
jest.mock("../lib/db.js", () => ({
4+
prisma: {
5+
stream: {
6+
findMany: jest.fn(),
7+
},
8+
},
9+
}));
10+
11+
import { prisma } from "../lib/db.js";
12+
13+
function buildMockStream(overrides: Partial<Record<string, unknown>> = {}) {
14+
return {
15+
id: `stream_${Math.random().toString(36).slice(2)}`,
16+
streamId: null,
17+
txHash: `tx_${Math.random().toString(36).slice(2)}`,
18+
sender: "GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
19+
receiver: "GBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB",
20+
tokenAddress: "CUSDC",
21+
amount: "1000000000",
22+
duration: 86400,
23+
status: "ACTIVE",
24+
withdrawn: "0",
25+
legacy: false,
26+
migrated: false,
27+
isPrivate: false,
28+
createdAt: new Date(),
29+
...overrides,
30+
};
31+
}
32+
33+
describe("StreamService N+1 query guard", () => {
34+
let service: StreamService;
35+
36+
beforeEach(() => {
37+
service = new StreamService();
38+
jest.clearAllMocks();
39+
});
40+
41+
it.each([50, 1000])(
42+
"issues exactly 1 query for getStreamsForAddress with %d streams",
43+
async (count) => {
44+
const mockStreams = Array.from({ length: count }, () => buildMockStream());
45+
(prisma.stream.findMany as jest.Mock).mockResolvedValueOnce(mockStreams);
46+
47+
const result = await service.getStreamsForAddress(
48+
"GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
49+
);
50+
51+
expect(result).toHaveLength(count);
52+
expect(prisma.stream.findMany).toHaveBeenCalledTimes(1);
53+
},
54+
);
55+
56+
it.each([50, 1000])(
57+
"issues exactly 1 query for getStreamsBatch across many addresses with %d streams",
58+
async (count) => {
59+
const addresses = Array.from(
60+
{ length: 25 },
61+
(_, i) => `GADDR${i.toString().padStart(2, "0")}AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA`.slice(0, 56),
62+
);
63+
const mockStreams = Array.from({ length: count }, (_, i) =>
64+
buildMockStream({
65+
sender: addresses[i % addresses.length],
66+
receiver: addresses[(i + 1) % addresses.length],
67+
}),
68+
);
69+
(prisma.stream.findMany as jest.Mock).mockResolvedValueOnce(mockStreams);
70+
71+
const result = await service.getStreamsBatch(addresses);
72+
73+
expect(prisma.stream.findMany).toHaveBeenCalledTimes(1);
74+
// Every requested address has an entry, even ones with no streams.
75+
expect(result.size).toBe(addresses.length);
76+
},
77+
);
78+
79+
it("does not query the database when getStreamsBatch is called with no addresses", async () => {
80+
const result = await service.getStreamsBatch([]);
81+
82+
expect(result.size).toBe(0);
83+
expect(prisma.stream.findMany).not.toHaveBeenCalled();
84+
});
85+
});

‎backend/src/api/streams.routes.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -341,10 +341,11 @@ router.get(
341341
const verificationData = await streamService.verifyStream(streamId);
342342

343343
if (!verificationData) {
344-
return res.status(404).json({
344+
res.status(404).json({
345345
success: false,
346346
error: "Stream not found or verification failed",
347347
});
348+
return;
348349
}
349350

350351
res.json({

‎backend/src/api/v3/history.routes.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
import { Router, Request, Response } from "express";
22
import { z } from "zod";
33
import { prisma } from "../../lib/db.js";
4-
import asyncHandler from "../utils/asyncHandler.js";
4+
import asyncHandler from "../../utils/asyncHandler.js";
55
import type { Prisma } from "../../generated/client/index.js";
66

77
const router = Router();

‎backend/src/api/v3/safe-vault.routes.ts‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
11
import { Router, Request, Response } from "express";
22
import { z } from "zod";
3-
import { SafeVaultService } from "../services/safe-vault.service.js";
4-
import validateRequest from "../middleware/validateRequest.js";
5-
import asyncHandler from "../utils/asyncHandler.js";
3+
import { SafeVaultService } from "../../services/safe-vault.service.js";
4+
import validateRequest from "../../middleware/validateRequest.js";
5+
import asyncHandler from "../../utils/asyncHandler.js";
66

77
const router = Router();
88
const safeVaultService = new SafeVaultService();

0 commit comments

Comments
 (0)