From 5d9d3141519bc79df115625c377be7398de02e4e Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 9 Jun 2026 23:25:15 +0000 Subject: [PATCH] Harden pipeline case event pagination --- server/src/__tests__/pipelines-routes.test.ts | 92 ++++++++++++++++++- server/src/routes/pipelines.ts | 42 ++++++++- server/src/services/pipelines.ts | 38 +++++++- 3 files changed, 164 insertions(+), 8 deletions(-) diff --git a/server/src/__tests__/pipelines-routes.test.ts b/server/src/__tests__/pipelines-routes.test.ts index b22ef129e3e..d80d2a514d7 100644 --- a/server/src/__tests__/pipelines-routes.test.ts +++ b/server/src/__tests__/pipelines-routes.test.ts @@ -30,6 +30,10 @@ import { import { errorHandler } from "../middleware/error-handler.js"; import { issueRoutes } from "../routes/issues.js"; import { pipelineRoutes } from "../routes/pipelines.js"; +import { + PIPELINE_CASE_EVENTS_MAX_LIMIT, + PIPELINE_CONTEXT_PACK_EVENT_LIMIT, +} from "../services/pipelines.js"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe.sequential : describe.skip; @@ -231,6 +235,90 @@ describeEmbeddedPostgres("pipeline routes", () => { await http.delete(`/api/pipelines/${pipelineId}/stages/${stageId}?moveCasesToStageId=${qaStage.body.id}`).expect(200); }); + it("paginates and caps case event responses", async () => { + const company = await seedCompany(); + const http = request(app(boardActor)); + const pipeline = await http.post(`/api/companies/${company.id}/pipelines`).send({ key: "event-cap", name: "Event cap" }).expect(201); + const created = await http + .post(`/api/pipelines/${pipeline.body.id}/cases`) + .send({ caseKey: "event-cap", title: "Event cap" }) + .expect(201); + const caseId = created.body.case.id as string; + const baseTime = Date.now() + 1_000; + + await db.insert(pipelineCaseEvents).values( + Array.from({ length: PIPELINE_CASE_EVENTS_MAX_LIMIT + 25 }, (_, index) => ({ + companyId: company.id, + caseId, + type: "updated", + actorType: "user", + actorUserId: "board-user", + payload: { index }, + createdAt: new Date(baseTime + index), + updatedAt: new Date(baseTime + index), + })), + ); + + const firstPage = await http + .get(`/api/cases/${caseId}/events?limit=${PIPELINE_CASE_EVENTS_MAX_LIMIT + 50}`) + .expect(200); + expect(firstPage.body.items).toHaveLength(PIPELINE_CASE_EVENTS_MAX_LIMIT); + expect(firstPage.body.pagination).toMatchObject({ + limit: PIPELINE_CASE_EVENTS_MAX_LIMIT, + offset: 0, + nextOffset: PIPELINE_CASE_EVENTS_MAX_LIMIT, + hasMore: true, + order: "asc", + }); + + const secondPage = await http + .get(`/api/cases/${caseId}/events?limit=10&offset=${PIPELINE_CASE_EVENTS_MAX_LIMIT}`) + .expect(200); + expect(secondPage.body.items).toHaveLength(10); + expect(secondPage.body.pagination).toMatchObject({ + limit: 10, + offset: PIPELINE_CASE_EVENTS_MAX_LIMIT, + nextOffset: PIPELINE_CASE_EVENTS_MAX_LIMIT + 10, + hasMore: true, + order: "asc", + }); + }); + + it("returns a bounded context-pack event tail for large histories", async () => { + const company = await seedCompany(); + const http = request(app(boardActor)); + const pipeline = await http.post(`/api/companies/${company.id}/pipelines`).send({ key: "context-tail", name: "Context tail" }).expect(201); + const created = await http + .post(`/api/pipelines/${pipeline.body.id}/cases`) + .send({ caseKey: "context-tail", title: "Context tail" }) + .expect(201); + const caseId = created.body.case.id as string; + const eventCount = PIPELINE_CONTEXT_PACK_EVENT_LIMIT + 12; + const baseTime = Date.now() + 1_000; + + await db.insert(pipelineCaseEvents).values( + Array.from({ length: eventCount }, (_, index) => ({ + companyId: company.id, + caseId, + type: "updated", + actorType: "user", + actorUserId: "board-user", + payload: { index }, + createdAt: new Date(baseTime + index), + updatedAt: new Date(baseTime + index), + })), + ); + + const pack = await http.get(`/api/cases/${caseId}/context-pack`).expect(200); + expect(pack.body.events).toHaveLength(PIPELINE_CONTEXT_PACK_EVENT_LIMIT); + expect(pack.body.events.map((event: { payload: { index: number } }) => event.payload.index)).toEqual( + Array.from( + { length: PIPELINE_CONTEXT_PACK_EVENT_LIMIT }, + (_, index) => eventCount - PIPELINE_CONTEXT_PACK_EVENT_LIMIT + index, + ), + ); + }); + it("returns 404 for cross-company pipeline access", async () => { const company = await seedCompany(); const [pipeline] = await db.insert(pipelines).values({ companyId: company.id, key: "x", name: "X" }).returning(); @@ -344,7 +432,7 @@ describeEmbeddedPostgres("pipeline routes", () => { expect(approvedDetail.body.case.title).toBe("Approved title"); expect(approvedDetail.body.case.fields).toEqual({ channel: "blog" }); const approvedEvents = await http.get(`/api/cases/${approved.body.case.id}/events`).expect(200); - expect(approvedEvents.body.map((event: { type: string }) => event.type)).toEqual([ + expect(approvedEvents.body.items.map((event: { type: string }) => event.type)).toEqual([ "ingested", "transitioned", "updated", @@ -361,7 +449,7 @@ describeEmbeddedPostgres("pipeline routes", () => { const reason = " Keep this exact reason. "; await http.post(`/api/cases/${rejected.body.case.id}/review`).send({ decision: "reject", reason, expectedVersion: 2 }).expect(200); const rejectedEvents = await http.get(`/api/cases/${rejected.body.case.id}/events`).expect(200); - const reviewEvent = rejectedEvents.body.find((event: { type: string }) => event.type === "review_decided"); + const reviewEvent = rejectedEvents.body.items.find((event: { type: string }) => event.type === "review_decided"); expect(reviewEvent.payload.reason).toBe(reason); }); diff --git a/server/src/routes/pipelines.ts b/server/src/routes/pipelines.ts index b933584ea55..271347e7ded 100644 --- a/server/src/routes/pipelines.ts +++ b/server/src/routes/pipelines.ts @@ -17,8 +17,16 @@ import { pipelines, } from "@paperclipai/db"; import { validate } from "../middleware/validate.js"; -import { conflict, forbidden, HttpError, notFound, unauthorized, unprocessable } from "../errors.js"; -import { pipelineService, type PipelineActor, type PipelineStageConfig, type PipelineStageKind } from "../services/pipelines.js"; +import { badRequest, conflict, forbidden, HttpError, notFound, unauthorized, unprocessable } from "../errors.js"; +import { + PIPELINE_CASE_EVENTS_DEFAULT_LIMIT, + PIPELINE_CASE_EVENTS_MAX_LIMIT, + PIPELINE_CONTEXT_PACK_EVENT_LIMIT, + pipelineService, + type PipelineActor, + type PipelineStageConfig, + type PipelineStageKind, +} from "../services/pipelines.js"; import { assertCompanyAccess } from "./authz.js"; const stageKindSchema = z.enum(["open", "working", "review", "done", "cancelled"]); @@ -175,6 +183,26 @@ function actorForMutation(req: Request): PipelineActor { throw unauthorized(); } +function parseOptionalNonNegativeInteger(value: unknown, name: string) { + if (value === undefined) return null; + if (Array.isArray(value)) throw badRequest(`${name} must be a single integer`); + const raw = typeof value === "string" ? value.trim() : String(value); + if (!/^\d+$/.test(raw)) throw badRequest(`${name} must be a non-negative integer`); + const parsed = Number(raw); + if (!Number.isSafeInteger(parsed)) throw badRequest(`${name} is too large`); + return parsed; +} + +function parseCaseEventsQuery(query: Request["query"]) { + const requestedLimit = parseOptionalNonNegativeInteger(query.limit, "limit"); + const offset = parseOptionalNonNegativeInteger(query.offset, "offset") ?? 0; + if (requestedLimit === 0) throw badRequest("limit must be a positive integer"); + return { + limit: Math.min(requestedLimit ?? PIPELINE_CASE_EVENTS_DEFAULT_LIMIT, PIPELINE_CASE_EVENTS_MAX_LIMIT), + offset, + }; +} + async function resolvePipelineCompanyId(db: Db, pipelineId: string) { const row = await db .select({ companyId: pipelines.companyId }) @@ -744,7 +772,8 @@ export function pipelineRoutes(db: Db, options: Parameters { const caseId = req.params.caseId as string; const companyId = await assertCaseAccess(db, req, caseId); - res.json(await svc.listCaseEvents(companyId, caseId)); + const pagination = parseCaseEventsQuery(req.query); + res.json(await svc.listCaseEventsPage(companyId, caseId, pagination)); }); router.get("/cases/:caseId/rollup", async (req, res) => { @@ -757,7 +786,10 @@ export function pipelineRoutes(db: Db, options: Parameters limit; + const items = hasMore ? rows.slice(0, limit) : rows; + return { + items, + pagination: { + limit, + offset, + nextOffset: hasMore ? offset + limit : null, + hasMore, + order, + }, + }; + }, + async listCaseEvents(companyId: string, caseId: string) { await getCaseWithStageOrThrow(db, companyId, caseId); return db