diff --git a/packages/db/src/migrations/0102_pipeline_upstream_drift_event.sql b/packages/db/src/migrations/0102_pipeline_upstream_drift_event.sql new file mode 100644 index 00000000000..c26d96f6deb --- /dev/null +++ b/packages/db/src/migrations/0102_pipeline_upstream_drift_event.sql @@ -0,0 +1,21 @@ +ALTER TABLE "pipeline_case_events" DROP CONSTRAINT "pipeline_case_events_type_check";--> statement-breakpoint +ALTER TABLE "pipeline_case_events" ADD CONSTRAINT "pipeline_case_events_type_check" CHECK ("pipeline_case_events"."type" in ( + 'ingested', + 'updated', + 'claimed', + 'lease_released', + 'lease_expired', + 'transitioned', + 'transition_suggested', + 'suggestion_resolved', + 'review_decided', + 'conversation_opened', + 'issue_linked', + 'issue_unlinked', + 'automation_executed', + 'automation_failed', + 'blockers_set', + 'blockers_resolved', + 'children_terminal', + 'upstream_drift' + )); diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 73478115f07..f68d43997ad 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -715,6 +715,13 @@ "when": 1781098769536, "tag": "0101_special_bloodstorm", "breakpoints": true + }, + { + "idx": 102, + "version": "7", + "when": 1781145600000, + "tag": "0102_pipeline_upstream_drift_event", + "breakpoints": true } ] -} +} \ No newline at end of file diff --git a/packages/db/src/schema/pipeline_case_events.ts b/packages/db/src/schema/pipeline_case_events.ts index 3ee580635d5..a1480239f0f 100644 --- a/packages/db/src/schema/pipeline_case_events.ts +++ b/packages/db/src/schema/pipeline_case_events.ts @@ -44,7 +44,8 @@ export const pipelineCaseEvents = pgTable( 'automation_failed', 'blockers_set', 'blockers_resolved', - 'children_terminal' + 'children_terminal', + 'upstream_drift' )`, ), actorTypeCheck: check("pipeline_case_events_actor_type_check", sql`${table.actorType} in ('user', 'agent', 'system')`), diff --git a/server/src/__tests__/pipelines-aggregation.test.ts b/server/src/__tests__/pipelines-aggregation.test.ts new file mode 100644 index 00000000000..288e4d95e62 --- /dev/null +++ b/server/src/__tests__/pipelines-aggregation.test.ts @@ -0,0 +1,479 @@ +import { randomUUID } from "node:crypto"; +import express from "express"; +import request from "supertest"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { + activityLog, + agents, + companies, + createDb, + issueComments, + issues, + pipelineAutomationExecutions, + pipelineCaseBlockers, + pipelineCaseEvents, + pipelineCaseIssueLinks, + pipelineCases, + pipelineStages, + pipelineTransitions, + pipelines, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +import { errorHandler } from "../middleware/error-handler.js"; +import { pipelineRoutes } from "../routes/pipelines.js"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe.sequential : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres pipeline aggregation tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describeEmbeddedPostgres("pipeline aggregation routes", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + const noopHeartbeat = { wakeup: async () => null }; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-pipelines-aggregation-"); + db = createDb(tempDb.connectionString); + }, 20_000); + + afterEach(async () => { + await db.delete(activityLog); + await db.delete(pipelineAutomationExecutions); + await db.delete(pipelineCaseBlockers); + await db.delete(pipelineCaseIssueLinks); + await db.delete(pipelineCaseEvents); + await db.delete(pipelineCases); + await db.delete(pipelineTransitions); + await db.delete(pipelineStages); + await db.delete(issueComments); + await db.delete(issues); + await db.delete(pipelines); + await db.delete(agents); + await db.delete(companies); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + function app(actor: Express.Request["actor"]) { + const instance = express(); + instance.use(express.json()); + instance.use((req, _res, next) => { + req.actor = actor; + next(); + }); + instance.use("/api", pipelineRoutes(db, { heartbeat: noopHeartbeat })); + instance.use(errorHandler); + return instance; + } + + async function seedCompany(name = "Aggregation Co") { + const [company] = await db.insert(companies).values({ + name, + issuePrefix: `P${randomUUID().replace(/-/g, "").slice(0, 6).toUpperCase()}`, + }).returning(); + return company!; + } + + async function seedAgent(companyId: string, name = "Pipeline Agent") { + const [agent] = await db.insert(agents).values({ + companyId, + name, + role: "engineer", + adapterType: "codex_local", + }).returning(); + return agent!; + } + + const boardActor: Express.Request["actor"] = { + type: "board", + userId: "board-user", + source: "local_implicit", + isInstanceAdmin: true, + }; + + function agentActor(companyId: string, agentId: string): Express.Request["actor"] { + return { + type: "agent", + agentId, + companyId, + runId: randomUUID(), + source: "agent_key", + }; + } + + const reviewStages = [ + { key: "intake", name: "Intake", kind: "open", position: 100 }, + { key: "drafting", name: "Drafting", kind: "working", position: 200 }, + { + key: "review_human", + name: "Final review", + kind: "review", + position: 300, + config: { approveToStageKey: "done", rejectToStageKey: "cancelled", requestChangesToStageKey: "drafting", requireRejectReason: true, reviewerKind: "human" }, + }, + { + key: "review_any", + name: "Open review", + kind: "review", + position: 400, + config: { approveToStageKey: "done", rejectToStageKey: "cancelled", requireRejectReason: false, reviewerKind: "any" }, + }, + { key: "done", name: "Done", kind: "done", position: 900 }, + { key: "cancelled", name: "Cancelled", kind: "cancelled", position: 1000 }, + ]; + + async function seedPipeline(http: ReturnType, companyId: string, key: string, name: string) { + const created = await http + .post(`/api/companies/${companyId}/pipelines`) + .send({ key, name, stages: reviewStages }) + .expect(201); + return created.body as { id: string; stages: Array<{ id: string; key: string }> }; + } + + it("groups attention into suggestions, reviews, and heads-up with approver filtering", async () => { + const company = await seedCompany(); + const agent = await seedAgent(company.id, "Drafting Agent"); + const board = request(app(boardActor)); + const asAgent = request(app(agentActor(company.id, agent.id))); + + const pipeline = await seedPipeline(board, company.id, "content", "Content production"); + + const suggested = await board + .post(`/api/pipelines/${pipeline.id}/cases`) + .send({ caseKey: "suggested", title: "Launch blog post" }) + .expect(201); + await asAgent + .post(`/api/cases/${suggested.body.case.id}/suggest-transition`) + .send({ toStageKey: "review_human", rationale: "Draft is ready for review", confidence: 0.9 }) + .expect(200); + + const humanReview = await board + .post(`/api/pipelines/${pipeline.id}/cases`) + .send({ caseKey: "human-review", title: "Needs a human decision" }) + .expect(201); + await board + .post(`/api/cases/${humanReview.body.case.id}/transition`) + .send({ toStageKey: "review_human", expectedVersion: 1 }) + .expect(200); + + const anyReview = await board + .post(`/api/pipelines/${pipeline.id}/cases`) + .send({ caseKey: "any-review", title: "Anyone can decide" }) + .expect(201); + await board + .post(`/api/cases/${anyReview.body.case.id}/transition`) + .send({ toStageKey: "review_any", expectedVersion: 1 }) + .expect(200); + + const upstream = await board + .post(`/api/pipelines/${pipeline.id}/cases`) + .send({ caseKey: "upstream", title: "Release plan" }) + .expect(201); + const dependent = await board + .post(`/api/pipelines/${pipeline.id}/cases`) + .send({ caseKey: "dependent", title: "Dependent draft", blockedByCaseIds: [upstream.body.case.id] }) + .expect(201); + const [workIssue] = await db.insert(issues).values({ + companyId: company.id, + title: "Write dependent draft", + status: "in_progress", + priority: "medium", + assigneeAgentId: agent.id, + }).returning(); + await board + .post(`/api/cases/${dependent.body.case.id}/issue-links`) + .send({ issueId: workIssue!.id, role: "work" }) + .expect(201); + await board + .patch(`/api/cases/${upstream.body.case.id}`) + .send({ summary: "Scope changed", expectedVersion: 1 }) + .expect(200); + + const boardFeed = await board.get(`/api/companies/${company.id}/pipelines-attention`).expect(200); + + expect(boardFeed.body.counts).toEqual({ suggestions: 1, reviews: 2, headsUp: 1 }); + const suggestion = boardFeed.body.suggestions[0]; + expect(suggestion.case.id).toBe(suggested.body.case.id); + expect(suggestion.case.pipeline.name).toBe("Content production"); + expect(suggestion.suggestion.toStageKey).toBe("review_human"); + expect(suggestion.suggestion.toStageName).toBe("Final review"); + expect(suggestion.suggestion.rationale).toBe("Draft is ready for review"); + expect(suggestion.suggestion.suggestedBy).toEqual({ agentId: agent.id, agentName: "Drafting Agent" }); + + const reviewCaseIds = boardFeed.body.reviews.map((entry: { case: { id: string } }) => entry.case.id); + expect(reviewCaseIds).toEqual(expect.arrayContaining([humanReview.body.case.id, anyReview.body.case.id])); + const humanEntry = boardFeed.body.reviews.find((entry: { case: { id: string } }) => entry.case.id === humanReview.body.case.id); + expect(humanEntry.review).toMatchObject({ + reviewerKind: "human", + approveToStageKey: "done", + rejectToStageKey: "cancelled", + requestChangesToStageKey: "drafting", + requireRejectReason: true, + expectedVersion: 2, + }); + + const headsUp = boardFeed.body.headsUp[0]; + expect(headsUp.case.id).toBe(dependent.body.case.id); + expect(headsUp.drift.upstream).toMatchObject({ + caseId: upstream.body.case.id, + caseKey: "upstream", + title: "Release plan", + pipelineId: pipeline.id, + pipelineName: "Content production", + }); + expect(headsUp.drift.previousVersion).toBe(1); + expect(headsUp.drift.version).toBe(2); + expect(headsUp.workIssue).toMatchObject({ issueId: workIssue!.id, status: "in_progress" }); + expect(headsUp.activeWork).toMatchObject({ issueId: workIssue!.id, agentId: agent.id, agentName: "Drafting Agent" }); + + // Agent callers only await review stages with reviewerKind "any". + const agentFeed = await asAgent.get(`/api/companies/${company.id}/pipelines-attention`).expect(200); + expect(agentFeed.body.reviews.map((entry: { case: { id: string } }) => entry.case.id)).toEqual([anyReview.body.case.id]); + + // Touching the dependent case resolves the drift notice. + await board + .patch(`/api/cases/${dependent.body.case.id}`) + .send({ title: "Dependent draft v2", expectedVersion: 1 }) + .expect(200); + const resolvedFeed = await board.get(`/api/companies/${company.id}/pipelines-attention`).expect(200); + expect(resolvedFeed.body.headsUp).toEqual([]); + + // Terminal cases drop out of the suggestions group. + await board + .post(`/api/cases/${anyReview.body.case.id}/review`) + .send({ decision: "approve", expectedVersion: 2 }) + .expect(200); + const afterApprove = await board.get(`/api/companies/${company.id}/pipelines-attention`).expect(200); + expect(afterApprove.body.reviews.map((entry: { case: { id: string } }) => entry.case.id)).toEqual([humanReview.body.case.id]); + }); + + it("serves the company-wide learnings feed with joins, filtering, and pagination", async () => { + const company = await seedCompany(); + const board = request(app(boardActor)); + const pipeline = await seedPipeline(board, company.id, "content", "Content production"); + + const first = await board + .post(`/api/pipelines/${pipeline.id}/cases`) + .send({ caseKey: "first", title: "First item" }) + .expect(201); + await board + .post(`/api/cases/${first.body.case.id}/transition`) + .send({ toStageKey: "review_human", expectedVersion: 1 }) + .expect(200); + await board + .post(`/api/cases/${first.body.case.id}/review`) + .send({ decision: "reject", reason: "Wrong audience", expectedVersion: 2 }) + .expect(200); + + const second = await board + .post(`/api/pipelines/${pipeline.id}/cases`) + .send({ caseKey: "second", title: "Second item" }) + .expect(201); + await board + .post(`/api/cases/${second.body.case.id}/transition`) + .send({ toStageKey: "review_human", expectedVersion: 1 }) + .expect(200); + await board + .post(`/api/cases/${second.body.case.id}/review`) + .send({ decision: "request_changes", reason: "Tighten the intro", expectedVersion: 2 }) + .expect(200); + + const feed = await board + .get(`/api/companies/${company.id}/case-events?types=review_decided`) + .expect(200); + expect(feed.body.items).toHaveLength(2); + expect(feed.body.items.every((item: { type: string }) => item.type === "review_decided")).toBe(true); + // Newest first. + expect(feed.body.items[0].case.caseKey).toBe("second"); + expect(feed.body.items[0].payload).toMatchObject({ decision: "request_changes", reason: "Tighten the intro" }); + expect(feed.body.items[0].pipeline).toMatchObject({ id: pipeline.id, name: "Content production" }); + expect(feed.body.items[0].fromStage).toMatchObject({ key: "review_human" }); + expect(feed.body.items[0].toStage).toMatchObject({ key: "drafting" }); + expect(feed.body.items[1].case.caseKey).toBe("first"); + expect(feed.body.items[1].payload).toMatchObject({ decision: "reject", reason: "Wrong audience" }); + + const page = await board + .get(`/api/companies/${company.id}/case-events?types=review_decided,transition_forced&limit=1`) + .expect(200); + expect(page.body.items).toHaveLength(1); + expect(page.body.pagination).toMatchObject({ limit: 1, offset: 0, nextOffset: 1, hasMore: true }); + const nextPage = await board + .get(`/api/companies/${company.id}/case-events?types=review_decided&limit=1&offset=1`) + .expect(200); + expect(nextPage.body.items[0].case.caseKey).toBe("first"); + expect(nextPage.body.pagination.hasMore).toBe(false); + + await board.get(`/api/companies/${company.id}/case-events?types=Bad%20Type!`).expect(400); + await board.get(`/api/companies/${company.id}/case-events?limit=0`).expect(400); + }); + + it("returns the recursive children tree with per-node rollups and a flat summary", async () => { + const company = await seedCompany(); + const board = request(app(boardActor)); + const release = await seedPipeline(board, company.id, "release", "Release train"); + const content = await seedPipeline(board, company.id, "content", "Content production"); + + const parent = await board + .post(`/api/pipelines/${release.id}/cases`) + .send({ caseKey: "launch", title: "Launch v2" }) + .expect(201); + const doneChild = await board + .post(`/api/pipelines/${content.id}/cases`) + .send({ caseKey: "blog", title: "Blog post", parentCaseId: parent.body.case.id }) + .expect(201); + await board + .post(`/api/cases/${doneChild.body.case.id}/transition`) + .send({ toStageKey: "done", expectedVersion: 1 }) + .expect(200); + const openChild = await board + .post(`/api/pipelines/${content.id}/cases`) + .send({ caseKey: "video", title: "Video", parentCaseId: parent.body.case.id }) + .expect(201); + const droppedGrandchild = await board + .post(`/api/pipelines/${content.id}/cases`) + .send({ caseKey: "short", title: "Short clip", parentCaseId: openChild.body.case.id }) + .expect(201); + await board + .post(`/api/cases/${droppedGrandchild.body.case.id}/transition`) + .send({ toStageKey: "cancelled", expectedVersion: 1 }) + .expect(200); + + const tree = await board.get(`/api/cases/${parent.body.case.id}/children`).expect(200); + expect(tree.body.rollup).toEqual({ total: 3, done: 1, dropped: 1, inMotion: 1 }); + expect(tree.body.truncated).toBe(false); + expect(tree.body.totalNodes).toBe(4); + expect(tree.body.case.pipeline).toMatchObject({ id: release.id, name: "Release train" }); + expect(tree.body.childGroups).toHaveLength(1); + const group = tree.body.childGroups[0]; + expect(group.pipeline).toMatchObject({ id: content.id, name: "Content production" }); + expect(group.cases.map((node: { caseKey: string }) => node.caseKey)).toEqual(["blog", "video"]); + const blogNode = group.cases[0]; + expect(blogNode.terminalKind).toBe("done"); + expect(blogNode.stage).toMatchObject({ key: "done", kind: "done" }); + expect(blogNode.rollup).toEqual({ total: 0, done: 0, dropped: 0, inMotion: 0 }); + const videoNode = group.cases[1]; + expect(videoNode.rollup).toEqual({ total: 1, done: 0, dropped: 1, inMotion: 0 }); + expect(videoNode.childGroups[0].cases[0]).toMatchObject({ caseKey: "short", terminalKind: "cancelled" }); + + const detail = await board.get(`/api/cases/${parent.body.case.id}`).expect(200); + expect(detail.body.childrenSummary).toMatchObject({ total: 2, done: 1, dropped: 0, inMotion: 1 }); + + await board.get(`/api/cases/${randomUUID()}/children`).expect(404); + }); + + it("enriches case detail and pipeline case lists with activeWork", async () => { + const company = await seedCompany(); + const agent = await seedAgent(company.id, "Working Agent"); + const board = request(app(boardActor)); + const pipeline = await seedPipeline(board, company.id, "content", "Content production"); + + const idleCase = await board + .post(`/api/pipelines/${pipeline.id}/cases`) + .send({ caseKey: "idle", title: "Idle item" }) + .expect(201); + const workedCase = await board + .post(`/api/pipelines/${pipeline.id}/cases`) + .send({ caseKey: "worked", title: "Worked item" }) + .expect(201); + + const [todoIssue] = await db.insert(issues).values({ + companyId: company.id, + title: "Not started yet", + status: "todo", + priority: "medium", + assigneeAgentId: agent.id, + }).returning(); + const [activeIssue] = await db.insert(issues).values({ + companyId: company.id, + title: "Write the draft", + identifier: "AGG-1", + status: "in_progress", + priority: "medium", + assigneeAgentId: agent.id, + startedAt: new Date("2026-06-10T12:00:00Z"), + }).returning(); + await board + .post(`/api/cases/${workedCase.body.case.id}/issue-links`) + .send({ issueId: todoIssue!.id, role: "work" }) + .expect(201); + await board + .post(`/api/cases/${workedCase.body.case.id}/issue-links`) + .send({ issueId: activeIssue!.id, role: "work" }) + .expect(201); + + const detail = await board.get(`/api/cases/${workedCase.body.case.id}`).expect(200); + expect(detail.body.activeWork).toMatchObject({ + issueId: activeIssue!.id, + issueIdentifier: "AGG-1", + issueTitle: "Write the draft", + agentId: agent.id, + agentName: "Working Agent", + }); + expect(detail.body.activeWork.startedAt).toBeTruthy(); + + const idleDetail = await board.get(`/api/cases/${idleCase.body.case.id}`).expect(200); + expect(idleDetail.body.activeWork).toBeNull(); + + const list = await board.get(`/api/pipelines/${pipeline.id}/cases`).expect(200); + const listedWorked = list.body.find((row: { case: { id: string } }) => row.case.id === workedCase.body.case.id); + const listedIdle = list.body.find((row: { case: { id: string } }) => row.case.id === idleCase.body.case.id); + expect(listedWorked.activeWork).toMatchObject({ issueId: activeIssue!.id, agentName: "Working Agent" }); + expect(listedIdle.activeWork).toBeNull(); + }); + + it("derives pipeline connections from observed cross-pipeline case parentage", async () => { + const company = await seedCompany(); + const board = request(app(boardActor)); + const release = await seedPipeline(board, company.id, "release", "Release train"); + const content = await seedPipeline(board, company.id, "content", "Content production"); + const assets = await seedPipeline(board, company.id, "assets", "Asset production"); + const lonely = await seedPipeline(board, company.id, "lonely", "Unconnected"); + + const releaseCase = await board + .post(`/api/pipelines/${release.id}/cases`) + .send({ caseKey: "launch", title: "Launch" }) + .expect(201); + const contentCase = await board + .post(`/api/pipelines/${content.id}/cases`) + .send({ caseKey: "blog", title: "Blog", parentCaseId: releaseCase.body.case.id }) + .expect(201); + await board + .post(`/api/pipelines/${assets.id}/cases`) + .send({ caseKey: "thumb", title: "Thumbnail", parentCaseId: contentCase.body.case.id }) + .expect(201); + // Same-pipeline parentage must not create a self connection. + await board + .post(`/api/pipelines/${content.id}/cases`) + .send({ caseKey: "blog-child", title: "Blog child", parentCaseId: contentCase.body.case.id }) + .expect(201); + + const listed = await board.get(`/api/companies/${company.id}/pipelines`).expect(200); + const byId = new Map(listed.body.map((row: { id: string }) => [row.id, row])); + expect((byId.get(release.id) as { connections: unknown }).connections).toEqual({ + upstreamPipelineIds: [], + downstreamPipelineIds: [content.id], + }); + expect((byId.get(content.id) as { connections: unknown }).connections).toEqual({ + upstreamPipelineIds: [release.id], + downstreamPipelineIds: [assets.id], + }); + expect((byId.get(assets.id) as { connections: unknown }).connections).toEqual({ + upstreamPipelineIds: [content.id], + downstreamPipelineIds: [], + }); + expect((byId.get(lonely.id) as { connections: unknown }).connections).toEqual({ + upstreamPipelineIds: [], + downstreamPipelineIds: [], + }); + }); +}); diff --git a/server/src/routes/pipelines.ts b/server/src/routes/pipelines.ts index c1b5dfed32c..74f3d03f3b9 100644 --- a/server/src/routes/pipelines.ts +++ b/server/src/routes/pipelines.ts @@ -27,6 +27,20 @@ import { type PipelineStageConfig, type PipelineStageKind, } from "../services/pipelines.js"; +import { + COMPANY_CASE_EVENTS_DEFAULT_LIMIT, + COMPANY_CASE_EVENTS_MAX_LIMIT, + COMPANY_CASE_EVENTS_MAX_TYPES, + getCaseChildrenTree, + getDirectChildrenSummary, + listCompanyCaseEvents, + listPipelineAttention, + loadActiveWorkForCases, + loadPipelineConnections, + PIPELINE_ATTENTION_DEFAULT_LIMIT, + PIPELINE_ATTENTION_MAX_LIMIT, + type AttentionCaller, +} from "../services/pipelines-aggregation.js"; import { assertCompanyAccess } from "./authz.js"; const stageKindSchema = z.enum(["open", "working", "review", "done", "cancelled"]); @@ -184,6 +198,34 @@ function actorForMutation(req: Request): PipelineActor { throw unauthorized(); } +function attentionCallerFor(req: Request): AttentionCaller { + if (req.actor.type === "agent") { + if (!req.actor.agentId) throw unauthorized(); + return { type: "agent", agentId: req.actor.agentId }; + } + if (req.actor.type === "board") { + return { type: "user", userId: req.actor.userId ?? "board" }; + } + throw unauthorized(); +} + +function parseEventTypesQuery(value: unknown): string[] | undefined { + if (value === undefined) return undefined; + const raw = Array.isArray(value) ? value : [value]; + const types = raw + .flatMap((item) => String(item).split(",")) + .map((item) => item.trim()) + .filter((item) => item.length > 0); + if (types.length === 0) return undefined; + if (types.length > COMPANY_CASE_EVENTS_MAX_TYPES) { + throw badRequest(`types accepts at most ${COMPANY_CASE_EVENTS_MAX_TYPES} values`); + } + for (const type of types) { + if (!/^[a-z_]{1,64}$/.test(type)) throw badRequest(`Invalid event type: ${type}`); + } + return [...new Set(types)]; +} + function parseOptionalNonNegativeInteger(value: unknown, name: string) { if (value === undefined) return null; if (Array.isArray(value)) throw badRequest(`${name} must be a single integer`); @@ -287,7 +329,34 @@ export function pipelineRoutes(db: Db, options: Parameters ({ ...row.pipeline, stageCount: row.stageCount, openCaseCount: row.openCaseCount }))); + const connections = await loadPipelineConnections(db, companyId); + res.json(rows.map((row) => ({ + ...row.pipeline, + stageCount: row.stageCount, + openCaseCount: row.openCaseCount, + connections: connections.get(row.pipeline.id) ?? { upstreamPipelineIds: [], downstreamPipelineIds: [] }, + }))); + }); + + router.get("/companies/:companyId/pipelines-attention", async (req, res) => { + const companyId = req.params.companyId as string; + assertPipelineCompanyAccess(req, companyId); + const caller = attentionCallerFor(req); + const requestedLimit = parseOptionalNonNegativeInteger(req.query.limit, "limit"); + if (requestedLimit === 0) throw badRequest("limit must be a positive integer"); + const limit = Math.min(requestedLimit ?? PIPELINE_ATTENTION_DEFAULT_LIMIT, PIPELINE_ATTENTION_MAX_LIMIT); + res.json(await listPipelineAttention(db, { companyId, caller, limit })); + }); + + router.get("/companies/:companyId/case-events", async (req, res) => { + const companyId = req.params.companyId as string; + assertPipelineCompanyAccess(req, companyId); + const types = parseEventTypesQuery(req.query.types); + const requestedLimit = parseOptionalNonNegativeInteger(req.query.limit, "limit"); + if (requestedLimit === 0) throw badRequest("limit must be a positive integer"); + const limit = Math.min(requestedLimit ?? COMPANY_CASE_EVENTS_DEFAULT_LIMIT, COMPANY_CASE_EVENTS_MAX_LIMIT); + const offset = parseOptionalNonNegativeInteger(req.query.offset, "offset") ?? 0; + res.json(await listCompanyCaseEvents(db, { companyId, types, limit, offset })); }); router.post("/companies/:companyId/pipelines", validate(createPipelineSchema), async (req, res) => { @@ -557,7 +626,8 @@ export function pipelineRoutes(db: Db, options: Parameters row.case.id)); + res.json(rows.map((row) => ({ ...row, activeWork: activeWork.get(row.case.id) ?? null }))); }); router.get("/cases/:caseId", async (req, res) => { @@ -792,6 +862,12 @@ export function pipelineRoutes(db: Db, options: Parameters { + const caseId = req.params.caseId as string; + const companyId = await assertCaseAccess(db, req, caseId); + res.json(await getCaseChildrenTree(db, companyId, caseId)); + }); + router.get("/cases/:caseId/rollup", async (req, res) => { const caseId = req.params.caseId as string; const companyId = await assertCaseAccess(db, req, caseId); @@ -846,12 +922,13 @@ async function getCaseDetail(db: Db, companyId: string, caseId: string) { .limit(1) .then((rows) => rows[0] ?? null); if (!row) throw notFound("Pipeline case not found"); - const [allowedNextStages, links, blockers, blocks, children] = await Promise.all([ + const [allowedNextStages, links, blockers, blocks, childrenCounts, activeWorkByCase] = await Promise.all([ db.select().from(pipelineStages).where(eq(pipelineStages.pipelineId, row.case.pipelineId)).orderBy(asc(pipelineStages.position)), db.select().from(pipelineCaseIssueLinks).where(and(eq(pipelineCaseIssueLinks.companyId, companyId), eq(pipelineCaseIssueLinks.caseId, caseId))), db.select().from(pipelineCaseBlockers).where(and(eq(pipelineCaseBlockers.companyId, companyId), eq(pipelineCaseBlockers.caseId, caseId))), db.select().from(pipelineCaseBlockers).where(and(eq(pipelineCaseBlockers.companyId, companyId), eq(pipelineCaseBlockers.blockedByCaseId, caseId))), - db.select().from(pipelineCases).where(and(eq(pipelineCases.companyId, companyId), eq(pipelineCases.parentCaseId, caseId))), + getDirectChildrenSummary(db, companyId, caseId), + loadActiveWorkForCases(db, companyId, [caseId]), ]); return { ...row, @@ -862,8 +939,10 @@ async function getCaseDetail(db: Db, companyId: string, caseId: string) { childrenSummary: { childCount: row.case.childCount, terminalChildCount: row.case.terminalChildCount, - loadedChildren: children.length, + loadedChildren: childrenCounts.total, + ...childrenCounts, }, + activeWork: activeWorkByCase.get(caseId) ?? null, pendingSuggestion: row.case.pendingSuggestion, }; } diff --git a/server/src/services/pipelines-aggregation.ts b/server/src/services/pipelines-aggregation.ts new file mode 100644 index 00000000000..83d590f25f8 --- /dev/null +++ b/server/src/services/pipelines-aggregation.ts @@ -0,0 +1,543 @@ +import { and, asc, desc, eq, inArray, isNotNull, isNull, ne, sql } from "drizzle-orm"; +import { alias } from "drizzle-orm/pg-core"; +import type { Db } from "@paperclipai/db"; +import { + agents, + issues, + pipelineCaseEvents, + pipelineCaseIssueLinks, + pipelineCases, + pipelineStages, + pipelines, +} from "@paperclipai/db"; +import { notFound } from "../errors.js"; + +export const PIPELINE_ATTENTION_DEFAULT_LIMIT = 50; +export const PIPELINE_ATTENTION_MAX_LIMIT = 100; +export const COMPANY_CASE_EVENTS_DEFAULT_LIMIT = 50; +export const COMPANY_CASE_EVENTS_MAX_LIMIT = 100; +export const COMPANY_CASE_EVENTS_MAX_TYPES = 10; +export const CASE_CHILDREN_TREE_MAX_NODES = 1_000; +export const CASE_CHILDREN_TREE_MAX_DEPTH = 10; + +export type AttentionCaller = + | { type: "user"; userId: string } + | { type: "agent"; agentId: string }; + +type CaseRow = typeof pipelineCases.$inferSelect; +type StageRow = typeof pipelineStages.$inferSelect; +type PipelineRow = typeof pipelines.$inferSelect; + +export type ActiveWork = { + issueId: string; + issueIdentifier: string | null; + issueTitle: string; + agentId: string; + agentName: string; + startedAt: Date | null; +}; + +function caseDisplay(row: { case: CaseRow; stage: StageRow; pipeline: PipelineRow }) { + return { + id: row.case.id, + caseKey: row.case.caseKey, + title: row.case.title, + summary: row.case.summary, + version: row.case.version, + terminalKind: row.case.terminalKind, + parentCaseId: row.case.parentCaseId, + updatedAt: row.case.updatedAt, + createdAt: row.case.createdAt, + pipeline: { id: row.pipeline.id, key: row.pipeline.key, name: row.pipeline.name }, + stage: { id: row.stage.id, key: row.stage.key, name: row.stage.name, kind: row.stage.kind }, + }; +} + +// Current review-stage approver semantics: reviewerKind "human" awaits any +// human caller, "any" awaits anyone. B1 (stage requireApproval + named +// approver) extends this predicate when it lands — keep the seam here. +function reviewStageAwaitsCaller(config: Record, caller: AttentionCaller) { + const reviewerKind = typeof config.reviewerKind === "string" ? config.reviewerKind : "human"; + if (reviewerKind === "any") return true; + return caller.type === "user"; +} + +function boundedLimit(limit: number | undefined, fallback: number, max: number) { + return Math.min(max, Math.max(1, Math.floor(limit ?? fallback))); +} + +export async function listPipelineAttention( + db: Db, + input: { companyId: string; caller: AttentionCaller; limit?: number }, +) { + const limit = boundedLimit(input.limit, PIPELINE_ATTENTION_DEFAULT_LIMIT, PIPELINE_ATTENTION_MAX_LIMIT); + + const suggestionAgent = alias(agents, "suggestion_agent"); + const suggestionToStage = alias(pipelineStages, "suggestion_to_stage"); + const suggestionRows = await db + .select({ + case: pipelineCases, + pipeline: pipelines, + stage: pipelineStages, + toStage: suggestionToStage, + suggestingAgent: { id: suggestionAgent.id, name: suggestionAgent.name }, + }) + .from(pipelineCases) + .innerJoin(pipelines, eq(pipelineCases.pipelineId, pipelines.id)) + .innerJoin(pipelineStages, eq(pipelineCases.stageId, pipelineStages.id)) + .leftJoin(suggestionToStage, and( + eq(suggestionToStage.pipelineId, pipelineCases.pipelineId), + eq(suggestionToStage.key, sql`${pipelineCases.pendingSuggestion}->>'toStageKey'`), + )) + .leftJoin(suggestionAgent, sql`${suggestionAgent.id}::text = ${pipelineCases.pendingSuggestion}->>'suggestedByAgentId'`) + .where(and( + eq(pipelineCases.companyId, input.companyId), + eq(pipelines.companyId, input.companyId), + isNull(pipelineCases.terminalKind), + isNotNull(pipelineCases.pendingSuggestion), + )) + .orderBy(desc(pipelineCases.updatedAt)) + .limit(limit); + + const suggestions = suggestionRows.map((row) => { + const suggestion = row.case.pendingSuggestion!; + return { + case: caseDisplay(row), + suggestion: { + id: suggestion.id, + fromStageKey: row.stage.key, + fromStageName: row.stage.name, + toStageKey: suggestion.toStageKey, + toStageName: row.toStage?.name ?? null, + rationale: suggestion.rationale, + confidence: suggestion.confidence ?? null, + createdAt: suggestion.createdAt, + suggestedBy: row.suggestingAgent?.id + ? { agentId: row.suggestingAgent.id, agentName: row.suggestingAgent.name } + : null, + }, + }; + }); + + const reviewRows = await db + .select({ case: pipelineCases, pipeline: pipelines, stage: pipelineStages }) + .from(pipelineCases) + .innerJoin(pipelines, eq(pipelineCases.pipelineId, pipelines.id)) + .innerJoin(pipelineStages, eq(pipelineCases.stageId, pipelineStages.id)) + .where(and( + eq(pipelineCases.companyId, input.companyId), + eq(pipelines.companyId, input.companyId), + eq(pipelineStages.kind, "review"), + isNull(pipelineCases.terminalKind), + )) + .orderBy(asc(pipelineCases.createdAt)) + .limit(PIPELINE_ATTENTION_MAX_LIMIT * 2); + + const reviews = reviewRows + .filter((row) => reviewStageAwaitsCaller((row.stage.config ?? {}) as Record, input.caller)) + .slice(0, limit) + .map((row) => { + const config = (row.stage.config ?? {}) as Record; + return { + case: caseDisplay(row), + review: { + expectedVersion: row.case.version, + approveToStageKey: typeof config.approveToStageKey === "string" ? config.approveToStageKey : null, + rejectToStageKey: typeof config.rejectToStageKey === "string" ? config.rejectToStageKey : null, + requestChangesToStageKey: typeof config.requestChangesToStageKey === "string" ? config.requestChangesToStageKey : null, + requireRejectReason: config.requireRejectReason !== false, + reviewerKind: typeof config.reviewerKind === "string" ? config.reviewerKind : "human", + }, + }; + }); + + const driftRows = await db + .selectDistinctOn([pipelineCaseEvents.caseId], { + event: pipelineCaseEvents, + case: pipelineCases, + pipeline: pipelines, + stage: pipelineStages, + }) + .from(pipelineCaseEvents) + .innerJoin(pipelineCases, eq(pipelineCaseEvents.caseId, pipelineCases.id)) + .innerJoin(pipelines, eq(pipelineCases.pipelineId, pipelines.id)) + .innerJoin(pipelineStages, eq(pipelineCases.stageId, pipelineStages.id)) + .where(and( + eq(pipelineCaseEvents.companyId, input.companyId), + eq(pipelineCaseEvents.type, "upstream_drift"), + eq(pipelineCases.companyId, input.companyId), + isNull(pipelineCases.terminalKind), + sql`${pipelineCaseEvents.createdAt} > ${pipelineCases.updatedAt}`, + )) + .orderBy(asc(pipelineCaseEvents.caseId), desc(pipelineCaseEvents.createdAt)) + .limit(limit); + + const driftCaseIds = driftRows.map((row) => row.case.id); + const upstreamCaseIds = [...new Set(driftRows + .map((row) => (row.event.payload as Record).upstreamCaseId) + .filter((value): value is string => typeof value === "string"))]; + + const [activeWorkByCase, workIssuesByCase, upstreamCases] = await Promise.all([ + loadActiveWorkForCases(db, input.companyId, driftCaseIds), + loadOpenWorkIssuesForCases(db, input.companyId, driftCaseIds), + upstreamCaseIds.length === 0 + ? Promise.resolve([] as Array<{ case: CaseRow; pipeline: PipelineRow }>) + : db + .select({ case: pipelineCases, pipeline: pipelines }) + .from(pipelineCases) + .innerJoin(pipelines, eq(pipelineCases.pipelineId, pipelines.id)) + .where(and(eq(pipelineCases.companyId, input.companyId), inArray(pipelineCases.id, upstreamCaseIds))), + ]); + const upstreamById = new Map(upstreamCases.map((row) => [row.case.id, row])); + + const headsUp = driftRows.map((row) => { + const payload = row.event.payload as Record; + const upstream = typeof payload.upstreamCaseId === "string" ? upstreamById.get(payload.upstreamCaseId) : undefined; + return { + case: caseDisplay(row), + drift: { + eventId: row.event.id, + createdAt: row.event.createdAt, + previousVersion: typeof payload.previousVersion === "number" ? payload.previousVersion : null, + version: typeof payload.version === "number" ? payload.version : null, + upstream: upstream + ? { + caseId: upstream.case.id, + caseKey: upstream.case.caseKey, + title: upstream.case.title, + pipelineId: upstream.pipeline.id, + pipelineName: upstream.pipeline.name, + } + : { + caseId: typeof payload.upstreamCaseId === "string" ? payload.upstreamCaseId : null, + caseKey: typeof payload.upstreamCaseKey === "string" ? payload.upstreamCaseKey : null, + title: null, + pipelineId: typeof payload.upstreamPipelineId === "string" ? payload.upstreamPipelineId : null, + pipelineName: null, + }, + }, + activeWork: activeWorkByCase.get(row.case.id) ?? null, + workIssue: workIssuesByCase.get(row.case.id) ?? null, + }; + }); + + return { + suggestions, + reviews, + headsUp, + counts: { suggestions: suggestions.length, reviews: reviews.length, headsUp: headsUp.length }, + }; +} + +export async function listCompanyCaseEvents( + db: Db, + input: { companyId: string; types?: string[]; limit?: number; offset?: number }, +) { + const limit = boundedLimit(input.limit, COMPANY_CASE_EVENTS_DEFAULT_LIMIT, COMPANY_CASE_EVENTS_MAX_LIMIT); + const offset = Math.max(0, Math.floor(input.offset ?? 0)); + const fromStage = alias(pipelineStages, "from_stage"); + const toStage = alias(pipelineStages, "to_stage"); + const actorAgent = alias(agents, "actor_agent"); + + const rows = await db + .select({ + event: pipelineCaseEvents, + case: { + id: pipelineCases.id, + caseKey: pipelineCases.caseKey, + title: pipelineCases.title, + terminalKind: pipelineCases.terminalKind, + }, + pipeline: { id: pipelines.id, key: pipelines.key, name: pipelines.name }, + fromStage: { id: fromStage.id, key: fromStage.key, name: fromStage.name, kind: fromStage.kind }, + toStage: { id: toStage.id, key: toStage.key, name: toStage.name, kind: toStage.kind }, + actorAgent: { id: actorAgent.id, name: actorAgent.name }, + }) + .from(pipelineCaseEvents) + .innerJoin(pipelineCases, eq(pipelineCaseEvents.caseId, pipelineCases.id)) + .innerJoin(pipelines, eq(pipelineCases.pipelineId, pipelines.id)) + .leftJoin(fromStage, eq(pipelineCaseEvents.fromStageId, fromStage.id)) + .leftJoin(toStage, eq(pipelineCaseEvents.toStageId, toStage.id)) + .leftJoin(actorAgent, eq(pipelineCaseEvents.actorAgentId, actorAgent.id)) + .where(and( + eq(pipelineCaseEvents.companyId, input.companyId), + eq(pipelineCases.companyId, input.companyId), + input.types && input.types.length > 0 ? inArray(pipelineCaseEvents.type, input.types) : undefined, + )) + .orderBy(desc(pipelineCaseEvents.createdAt), desc(pipelineCaseEvents.id)) + .limit(limit + 1) + .offset(offset); + + const hasMore = rows.length > limit; + const items = (hasMore ? rows.slice(0, limit) : rows).map((row) => ({ + ...row.event, + case: row.case, + pipeline: row.pipeline, + fromStage: row.fromStage?.id ? row.fromStage : null, + toStage: row.toStage?.id ? row.toStage : null, + actorAgent: row.actorAgent?.id ? row.actorAgent : null, + })); + + return { + items, + pagination: { limit, offset, nextOffset: hasMore ? offset + limit : null, hasMore }, + }; +} + +export type CaseChildrenRollup = { total: number; done: number; dropped: number; inMotion: number }; + +type SubtreeRow = { + id: string; + parent_case_id: string | null; + pipeline_id: string; + stage_id: string; + case_key: string; + title: string; + terminal_kind: string | null; + created_at: string | Date; + updated_at: string | Date; + depth: number; +}; + +export type CaseChildNode = { + id: string; + caseKey: string; + title: string; + terminalKind: string | null; + createdAt: Date; + updatedAt: Date; + pipeline: { id: string; key: string; name: string }; + stage: { id: string; key: string; name: string; kind: string }; + rollup: CaseChildrenRollup; + childGroups: Array<{ pipeline: { id: string; key: string; name: string }; cases: CaseChildNode[] }>; +}; + +export async function getCaseChildrenTree(db: Db, companyId: string, caseId: string) { + const result = await db.execute(sql` + with recursive subtree as ( + select id, parent_case_id, pipeline_id, stage_id, case_key, title, terminal_kind, created_at, updated_at, 0 as depth + from pipeline_cases + where company_id = ${companyId} and id = ${caseId} + union all + select child.id, child.parent_case_id, child.pipeline_id, child.stage_id, child.case_key, child.title, + child.terminal_kind, child.created_at, child.updated_at, parent.depth + 1 + from pipeline_cases child + join subtree parent on child.parent_case_id = parent.id + where child.company_id = ${companyId} and parent.depth < ${CASE_CHILDREN_TREE_MAX_DEPTH} + ) + select * from subtree limit ${CASE_CHILDREN_TREE_MAX_NODES + 1} + `); + const rows = Array.from(result) as SubtreeRow[]; + if (rows.length === 0) throw notFound("Pipeline case not found"); + const truncated = rows.length > CASE_CHILDREN_TREE_MAX_NODES; + const bounded = truncated ? rows.slice(0, CASE_CHILDREN_TREE_MAX_NODES) : rows; + + const pipelineIds = [...new Set(bounded.map((row) => row.pipeline_id))]; + const stageIds = [...new Set(bounded.map((row) => row.stage_id))]; + const [pipelineRows, stageRows] = await Promise.all([ + db.select({ id: pipelines.id, key: pipelines.key, name: pipelines.name }) + .from(pipelines) + .where(and(eq(pipelines.companyId, companyId), inArray(pipelines.id, pipelineIds))), + db.select({ id: pipelineStages.id, key: pipelineStages.key, name: pipelineStages.name, kind: pipelineStages.kind }) + .from(pipelineStages) + .where(inArray(pipelineStages.id, stageIds)), + ]); + const pipelineById = new Map(pipelineRows.map((row) => [row.id, row])); + const stageById = new Map(stageRows.map((row) => [row.id, row])); + + const nodeById = new Map(); + const childRowsByParent = new Map(); + for (const row of bounded) { + if (row.id !== caseId && row.parent_case_id) { + const list = childRowsByParent.get(row.parent_case_id) ?? []; + list.push(row); + childRowsByParent.set(row.parent_case_id, list); + } + } + + function buildNode(row: SubtreeRow): CaseChildNode { + const childRows = (childRowsByParent.get(row.id) ?? []) + .sort((a, b) => new Date(a.created_at).getTime() - new Date(b.created_at).getTime()); + const children = childRows.map(buildNode); + const rollup: CaseChildrenRollup = { total: 0, done: 0, dropped: 0, inMotion: 0 }; + for (const child of children) { + rollup.total += 1 + child.rollup.total; + rollup.done += (child.terminalKind === "done" ? 1 : 0) + child.rollup.done; + rollup.dropped += (child.terminalKind === "cancelled" ? 1 : 0) + child.rollup.dropped; + rollup.inMotion += (child.terminalKind === null ? 1 : 0) + child.rollup.inMotion; + } + const pipeline = pipelineById.get(row.pipeline_id) ?? { id: row.pipeline_id, key: "", name: "" }; + const groups = new Map(); + for (const child of children) { + const group = groups.get(child.pipeline.id) ?? { pipeline: child.pipeline, cases: [] }; + group.cases.push(child); + groups.set(child.pipeline.id, group); + } + const childGroups = [...groups.values()].sort((a, b) => { + if (a.pipeline.id === row.pipeline_id && b.pipeline.id !== row.pipeline_id) return -1; + if (b.pipeline.id === row.pipeline_id && a.pipeline.id !== row.pipeline_id) return 1; + return a.pipeline.name.localeCompare(b.pipeline.name); + }); + const node: CaseChildNode = { + id: row.id, + caseKey: row.case_key, + title: row.title, + terminalKind: row.terminal_kind, + createdAt: new Date(row.created_at), + updatedAt: new Date(row.updated_at), + pipeline, + stage: stageById.get(row.stage_id) ?? { id: row.stage_id, key: "", name: "", kind: "" }, + rollup, + childGroups, + }; + nodeById.set(row.id, node); + return node; + } + + const rootRow = bounded.find((row) => row.id === caseId); + if (!rootRow) throw notFound("Pipeline case not found"); + const root = buildNode(rootRow); + + return { + case: root, + rollup: root.rollup, + childGroups: root.childGroups, + truncated, + totalNodes: bounded.length, + }; +} + +export async function getDirectChildrenSummary( + db: Db, + companyId: string, + caseId: string, +): Promise { + const [counts] = await db + .select({ + total: sql`count(*)::int`, + done: sql`count(*) filter (where ${pipelineCases.terminalKind} = 'done')::int`, + dropped: sql`count(*) filter (where ${pipelineCases.terminalKind} = 'cancelled')::int`, + inMotion: sql`count(*) filter (where ${pipelineCases.terminalKind} is null)::int`, + }) + .from(pipelineCases) + .where(and(eq(pipelineCases.companyId, companyId), eq(pipelineCases.parentCaseId, caseId))); + return counts ?? { total: 0, done: 0, dropped: 0, inMotion: 0 }; +} + +export async function loadActiveWorkForCases( + db: Db, + companyId: string, + caseIds: string[], +): Promise> { + const map = new Map(caseIds.map((id) => [id, null])); + if (caseIds.length === 0) return map; + const rows = await db + .select({ + caseId: pipelineCaseIssueLinks.caseId, + issueId: issues.id, + issueIdentifier: issues.identifier, + issueTitle: issues.title, + agentId: issues.assigneeAgentId, + agentName: agents.name, + startedAt: issues.startedAt, + issueUpdatedAt: issues.updatedAt, + }) + .from(pipelineCaseIssueLinks) + .innerJoin(issues, eq(pipelineCaseIssueLinks.issueId, issues.id)) + .innerJoin(agents, eq(issues.assigneeAgentId, agents.id)) + .where(and( + eq(pipelineCaseIssueLinks.companyId, companyId), + inArray(pipelineCaseIssueLinks.caseId, caseIds), + inArray(pipelineCaseIssueLinks.role, ["work", "automation"]), + eq(issues.companyId, companyId), + eq(issues.status, "in_progress"), + isNull(issues.hiddenAt), + )) + .orderBy(desc(issues.updatedAt)); + for (const row of rows) { + if (map.get(row.caseId)) continue; + map.set(row.caseId, { + issueId: row.issueId, + issueIdentifier: row.issueIdentifier, + issueTitle: row.issueTitle, + agentId: row.agentId!, + agentName: row.agentName, + startedAt: row.startedAt ?? row.issueUpdatedAt, + }); + } + return map; +} + +async function loadOpenWorkIssuesForCases(db: Db, companyId: string, caseIds: string[]) { + const map = new Map(); + if (caseIds.length === 0) return map; + const rows = await db + .select({ + caseId: pipelineCaseIssueLinks.caseId, + issueId: issues.id, + issueIdentifier: issues.identifier, + title: issues.title, + status: issues.status, + }) + .from(pipelineCaseIssueLinks) + .innerJoin(issues, eq(pipelineCaseIssueLinks.issueId, issues.id)) + .where(and( + eq(pipelineCaseIssueLinks.companyId, companyId), + inArray(pipelineCaseIssueLinks.caseId, caseIds), + eq(pipelineCaseIssueLinks.role, "work"), + eq(issues.companyId, companyId), + ne(issues.status, "done"), + ne(issues.status, "cancelled"), + isNull(issues.hiddenAt), + )) + .orderBy(desc(issues.updatedAt)); + for (const row of rows) { + if (map.has(row.caseId)) continue; + map.set(row.caseId, { + issueId: row.issueId, + issueIdentifier: row.issueIdentifier, + title: row.title, + status: row.status, + }); + } + return map; +} + +export type PipelineConnections = { upstreamPipelineIds: string[]; downstreamPipelineIds: string[] }; + +export async function loadPipelineConnections( + db: Db, + companyId: string, +): Promise> { + const parentCase = alias(pipelineCases, "parent_case"); + const rows = await db + .selectDistinct({ + parentPipelineId: parentCase.pipelineId, + childPipelineId: pipelineCases.pipelineId, + }) + .from(pipelineCases) + .innerJoin(parentCase, eq(pipelineCases.parentCaseId, parentCase.id)) + .where(and( + eq(pipelineCases.companyId, companyId), + eq(parentCase.companyId, companyId), + ne(pipelineCases.pipelineId, parentCase.pipelineId), + )); + const map = new Map(); + const entry = (pipelineId: string) => { + let value = map.get(pipelineId); + if (!value) { + value = { upstreamPipelineIds: [], downstreamPipelineIds: [] }; + map.set(pipelineId, value); + } + return value; + }; + for (const row of rows) { + entry(row.childPipelineId).upstreamPipelineIds.push(row.parentPipelineId); + entry(row.parentPipelineId).downstreamPipelineIds.push(row.childPipelineId); + } + for (const value of map.values()) { + value.upstreamPipelineIds.sort(); + value.downstreamPipelineIds.sort(); + } + return map; +} diff --git a/server/src/services/pipelines.ts b/server/src/services/pipelines.ts index 4b39a25811d..e0116e319d2 100644 --- a/server/src/services/pipelines.ts +++ b/server/src/services/pipelines.ts @@ -617,26 +617,39 @@ async function notifyDependentWorkIssuesOfUpstreamContentChange( version: number; }, ) { - const rows = await db - .selectDistinctOn([issues.id], { issueId: issues.id }) + const dependents = await db + .select({ dependentCase: pipelineCases }) .from(pipelineCaseBlockers) .innerJoin(pipelineCases, eq(pipelineCaseBlockers.caseId, pipelineCases.id)) - .innerJoin(pipelineCaseIssueLinks, eq(pipelineCaseIssueLinks.caseId, pipelineCases.id)) - .innerJoin(issues, eq(pipelineCaseIssueLinks.issueId, issues.id)) .where(and( eq(pipelineCaseBlockers.companyId, input.companyId), eq(pipelineCaseBlockers.blockedByCaseId, input.upstreamCase.id), eq(pipelineCases.companyId, input.companyId), isNull(pipelineCases.terminalKind), + )); + + if (dependents.length === 0) return; + + const dependentCaseIds = dependents.map((row) => row.dependentCase.id); + const linkRows = await db + .select({ caseId: pipelineCaseIssueLinks.caseId, issueId: issues.id }) + .from(pipelineCaseIssueLinks) + .innerJoin(issues, eq(pipelineCaseIssueLinks.issueId, issues.id)) + .where(and( eq(pipelineCaseIssueLinks.companyId, input.companyId), + inArray(pipelineCaseIssueLinks.caseId, dependentCaseIds), eq(pipelineCaseIssueLinks.role, "work"), eq(issues.companyId, input.companyId), ne(issues.status, "done"), ne(issues.status, "cancelled"), isNull(issues.hiddenAt), )); - - if (rows.length === 0) return; + const issueIdsByCase = new Map(); + for (const row of linkRows) { + const list = issueIdsByCase.get(row.caseId) ?? []; + list.push(row.issueId); + issueIdsByCase.set(row.caseId, list); + } const upstreamLink = buildCaseDeepLink({ pipelineId: input.upstreamCase.pipelineId, @@ -644,14 +657,36 @@ async function notifyDependentWorkIssuesOfUpstreamContentChange( }); const body = `Upstream case [${input.upstreamCase.caseKey}](${upstreamLink}) changed (v${input.previousVersion}→v${input.version}).`; - for (const row of rows) { - await db.insert(issueComments).values({ + const notifiedIssueIds = new Set(); + for (const { dependentCase } of dependents) { + const issueIds = issueIdsByCase.get(dependentCase.id) ?? []; + for (const issueId of issueIds) { + if (notifiedIssueIds.has(issueId)) continue; + notifiedIssueIds.add(issueId); + await db.insert(issueComments).values({ + companyId: input.companyId, + issueId, + authorType: "system", + body, + }); + await db.update(issues).set({ updatedAt: nowDate() }).where(eq(issues.id, issueId)); + } + // The drift event intentionally does not bump the dependent case's + // updatedAt: "unresolved drift" is derived as event.createdAt > case.updatedAt. + await writeCaseEvent(db, { companyId: input.companyId, - issueId: row.issueId, - authorType: "system", - body, + caseId: dependentCase.id, + type: "upstream_drift", + actor: { type: "system" }, + payload: { + upstreamCaseId: input.upstreamCase.id, + upstreamCaseKey: input.upstreamCase.caseKey, + upstreamPipelineId: input.upstreamCase.pipelineId, + previousVersion: input.previousVersion, + version: input.version, + notifiedIssueIds: issueIds, + }, }); - await db.update(issues).set({ updatedAt: nowDate() }).where(eq(issues.id, row.issueId)); } }