From 99d8d1c890371701ee65a12f9ef7b99f8cc4c169 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 10 Jun 2026 12:08:12 +0000 Subject: [PATCH] Add pipeline upstream drift notices Co-Authored-By: Paperclip --- .../src/__tests__/pipelines-service.test.ts | 178 ++++++++++++++++++ server/src/services/pipelines.ts | 77 +++++++- 2 files changed, 249 insertions(+), 6 deletions(-) diff --git a/server/src/__tests__/pipelines-service.test.ts b/server/src/__tests__/pipelines-service.test.ts index 629ab12244e..4f89a5614a0 100644 --- a/server/src/__tests__/pipelines-service.test.ts +++ b/server/src/__tests__/pipelines-service.test.ts @@ -124,6 +124,28 @@ describeEmbeddedPostgres("pipelineService", () => { return count ?? 0; } + async function seedLinkedIssue(input: { + companyId: string; + caseId: string; + role: "origin" | "conversation" | "work" | "automation"; + status?: "backlog" | "todo" | "in_progress" | "in_review" | "done" | "blocked" | "cancelled"; + title?: string; + }) { + const [issue] = await db.insert(issues).values({ + companyId: input.companyId, + title: input.title ?? `${input.role} issue`, + status: input.status ?? "todo", + priority: "medium", + }).returning(); + await db.insert(pipelineCaseIssueLinks).values({ + companyId: input.companyId, + caseId: input.caseId, + issueId: issue!.id, + role: input.role, + }); + return issue!; + } + it("seeds default stages and protects non-empty stage deletion", async () => { const { company, pipeline, byKey } = await seedPipeline(); @@ -478,6 +500,162 @@ describeEmbeddedPostgres("pipelineService", () => { expect(moved.case.version).toBe(2); }); + it("posts upstream drift notices to active dependent work issues only", async () => { + const { company, pipeline } = await seedPipeline(); + const upstream = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "draft", + title: "Draft", + actor: userActor, + }); + const workDependent = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "asset-work", + title: "Asset work", + blockedByCaseIds: [upstream.case.id], + actor: userActor, + }); + const conversationDependent = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "asset-conversation", + title: "Asset conversation", + blockedByCaseIds: [upstream.case.id], + actor: userActor, + }); + const workIssue = await seedLinkedIssue({ + companyId: company.id, + caseId: workDependent.case.id, + role: "work", + title: "Asset work issue", + }); + const conversationIssue = await seedLinkedIssue({ + companyId: company.id, + caseId: conversationDependent.case.id, + role: "conversation", + title: "Conversation issue", + }); + + const updated = await svc.patchCaseContent({ + companyId: company.id, + caseId: upstream.case.id, + title: "Draft v2", + expectedVersion: 1, + actor: userActor, + }); + + expect(updated.version).toBe(2); + const workComments = await db.select().from(issueComments).where(eq(issueComments.issueId, workIssue.id)); + expect(workComments).toHaveLength(1); + expect(workComments[0]!.authorType).toBe("system"); + expect(workComments[0]!.body).toBe( + `Upstream case [draft](/PAP/pipelines/${pipeline.id}/cases/${upstream.case.id}) changed (v1→v2).`, + ); + const conversationComments = await db.select().from(issueComments).where(eq(issueComments.issueId, conversationIssue.id)); + expect(conversationComments).toHaveLength(0); + }); + + it("skips upstream drift notices for terminal dependents and dependents without work issues", async () => { + const { company, pipeline } = await seedPipeline(); + const upstream = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "source", + title: "Source", + actor: userActor, + }); + const terminalDependent = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + stageKey: "done", + caseKey: "terminal-dependent", + title: "Terminal dependent", + actor: userActor, + }); + await svc.replaceBlockers({ + companyId: company.id, + caseId: terminalDependent.case.id, + blockedByCaseIds: [upstream.case.id], + actor: userActor, + }); + const noWorkDependent = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "no-work-dependent", + title: "No work dependent", + blockedByCaseIds: [upstream.case.id], + actor: userActor, + }); + const terminalIssue = await seedLinkedIssue({ + companyId: company.id, + caseId: terminalDependent.case.id, + role: "work", + title: "Terminal work issue", + }); + const conversationIssue = await seedLinkedIssue({ + companyId: company.id, + caseId: noWorkDependent.case.id, + role: "conversation", + title: "Non-work issue", + }); + + await svc.patchCaseContent({ + companyId: company.id, + caseId: upstream.case.id, + summary: "Updated source", + expectedVersion: 1, + actor: userActor, + }); + + const terminalComments = await db.select().from(issueComments).where(eq(issueComments.issueId, terminalIssue.id)); + expect(terminalComments).toHaveLength(0); + const conversationComments = await db.select().from(issueComments).where(eq(issueComments.issueId, conversationIssue.id)); + expect(conversationComments).toHaveLength(0); + }); + + it("does not bump versions or notify dependents on no-op content patches", async () => { + const { company, pipeline } = await seedPipeline(); + const upstream = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "noop-source", + title: "No-op source", + fields: { channel: "blog" }, + actor: userActor, + }); + const dependent = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "noop-dependent", + title: "No-op dependent", + blockedByCaseIds: [upstream.case.id], + actor: userActor, + }); + const workIssue = await seedLinkedIssue({ + companyId: company.id, + caseId: dependent.case.id, + role: "work", + title: "No-op work issue", + }); + const beforeEvents = await eventCount(upstream.case.id); + + const patched = await svc.patchCaseContent({ + companyId: company.id, + caseId: upstream.case.id, + title: "No-op source", + fields: { channel: "blog" }, + expectedVersion: 1, + actor: userActor, + }); + + expect(patched.version).toBe(1); + expect(await eventCount(upstream.case.id)).toBe(beforeEvents); + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, workIssue.id)); + expect(comments).toHaveLength(0); + }); + it("resolves in-batch forward blocker case keys", async () => { const { company, pipeline } = await seedPipeline(); diff --git a/server/src/services/pipelines.ts b/server/src/services/pipelines.ts index 480d3c2ef07..77274dc1009 100644 --- a/server/src/services/pipelines.ts +++ b/server/src/services/pipelines.ts @@ -1,4 +1,5 @@ import { randomUUID } from "node:crypto"; +import { isDeepStrictEqual } from "node:util"; import { and, asc, desc, eq, inArray, isNull, ne, or, sql } from "drizzle-orm"; import { alias } from "drizzle-orm/pg-core"; import type { Db } from "@paperclipai/db"; @@ -588,6 +589,53 @@ async function handleBlockersResolved(db: PipelineDb, companyId: string, blocker } } +async function notifyDependentWorkIssuesOfUpstreamContentChange( + db: PipelineDb, + input: { + companyId: string; + upstreamCase: typeof pipelineCases.$inferSelect; + previousVersion: number; + version: number; + }, +) { + const rows = await db + .selectDistinctOn([issues.id], { issueId: issues.id }) + .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), + eq(pipelineCaseIssueLinks.companyId, input.companyId), + 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 upstreamLink = buildCaseDeepLink({ + pipelineId: input.upstreamCase.pipelineId, + caseId: input.upstreamCase.id, + }); + 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({ + companyId: input.companyId, + issueId: row.issueId, + authorType: "system", + body, + }); + await db.update(issues).set({ updatedAt: nowDate() }).where(eq(issues.id, row.issueId)); + } +} + async function validateBlockerSet( db: PipelineDb, input: { companyId: string; caseId: string; blockedByCaseIds: string[] }, @@ -863,14 +911,23 @@ export function pipelineService(db: Db, deps: { heartbeat?: IssueAssignmentWakeu parentCaseId: input.parentCaseId, }); } + const titleChanged = input.title !== undefined && input.title !== current.title; + const summaryChanged = input.summary !== undefined && input.summary !== current.summary; + const fieldsChanged = input.fields !== undefined && !isDeepStrictEqual(input.fields, current.fields); + const parentCaseChanged = input.parentCaseId !== undefined && input.parentCaseId !== current.parentCaseId; + const contentChanged = titleChanged || summaryChanged || fieldsChanged; + if (!contentChanged && !parentCaseChanged) { + return { case: current, event: null }; + } + const patch: Partial = { version: current.version + 1, updatedAt: nowDate(), }; - if (input.title !== undefined) patch.title = input.title; - if (input.summary !== undefined) patch.summary = input.summary; - if (input.fields !== undefined) patch.fields = input.fields; - if (input.parentCaseId !== undefined) patch.parentCaseId = input.parentCaseId; + if (titleChanged) patch.title = input.title; + if (summaryChanged) patch.summary = input.summary; + if (fieldsChanged) patch.fields = input.fields; + if (parentCaseChanged) patch.parentCaseId = input.parentCaseId; const [updated] = await tx .update(pipelineCases) @@ -890,10 +947,10 @@ export function pipelineService(db: Db, deps: { heartbeat?: IssueAssignmentWakeu payload: { previousVersion: current.version, version: updated.version, - parentCaseChanged: input.parentCaseId !== undefined && input.parentCaseId !== current.parentCaseId, + parentCaseChanged, }, }); - if (input.parentCaseId !== undefined && input.parentCaseId !== current.parentCaseId) { + if (parentCaseChanged) { const terminalDelta = isTerminalKind(current.terminalKind) ? 1 : 0; await adjustParentCounts(tx, { parentCaseId: current.parentCaseId, @@ -909,6 +966,14 @@ export function pipelineService(db: Db, deps: { heartbeat?: IssueAssignmentWakeu await handleChildrenTerminal(tx, input.companyId, input.parentCaseId); } } + if (contentChanged) { + await notifyDependentWorkIssuesOfUpstreamContentChange(tx, { + companyId: input.companyId, + upstreamCase: updated, + previousVersion: current.version, + version: updated.version, + }); + } return { case: updated, event }; }