From 13f8dc5cbb8c2e1f7ac43dd1c5a393388c1dbdd5 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 10 Jun 2026 12:02:55 +0000 Subject: [PATCH] Add pipeline blocker key ingest wiring Co-Authored-By: Paperclip --- cli/src/__tests__/pipelines.test.ts | 30 ++++ cli/src/commands/pipelines.ts | 3 + server/src/__tests__/pipelines-routes.test.ts | 10 +- .../src/__tests__/pipelines-service.test.ts | 117 ++++++++++++ server/src/routes/pipelines.ts | 1 + server/src/services/pipelines.ts | 170 ++++++++++++++++-- 6 files changed, 314 insertions(+), 17 deletions(-) diff --git a/cli/src/__tests__/pipelines.test.ts b/cli/src/__tests__/pipelines.test.ts index df409fac56d..d9326a463f3 100644 --- a/cli/src/__tests__/pipelines.test.ts +++ b/cli/src/__tests__/pipelines.test.ts @@ -89,6 +89,36 @@ describe("pipeline CLI commands", () => { items: [{ caseId: CASE_ID, decision: "request_changes", reason: "Needs edits", expectedVersion: 2 }], }); }); + + it("passes blockedByCaseKeys rows through ingest-batch", async () => { + const dir = await mkdtemp(join(tmpdir(), "paperclip-pipeline-cli-")); + const file = join(dir, "ingest-batch.json"); + await writeFile(file, JSON.stringify([ + { caseKey: "tweet", title: "Tweet", blockedByCaseKeys: ["image", "post"] }, + { caseKey: "image", title: "Image" }, + { caseKey: "post", title: "Post" }, + ])); + const fetchMock = vi.fn() + .mockResolvedValueOnce(jsonResponse([{ id: "33333333-3333-4333-8333-333333333333", key: "content", name: "Content" }])) + .mockResolvedValueOnce(jsonResponse([])); + vi.stubGlobal("fetch", fetchMock); + + try { + await run(["pipelines", "ingest-batch", "content", "--file", file]); + } finally { + await rm(dir, { recursive: true, force: true }); + } + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(fetchMock.mock.calls[1]?.[0]).toBe("http://localhost:3100/api/pipelines/33333333-3333-4333-8333-333333333333/cases/batch"); + expect(JSON.parse(String(fetchMock.mock.calls[1]?.[1]?.body))).toEqual({ + items: [ + { caseKey: "tweet", title: "Tweet", blockedByCaseKeys: ["image", "post"] }, + { caseKey: "image", title: "Image" }, + { caseKey: "post", title: "Post" }, + ], + }); + }); }); function jsonResponse(body: unknown = { ok: true }, init: ResponseInit = { status: 200 }): Response { diff --git a/cli/src/commands/pipelines.ts b/cli/src/commands/pipelines.ts index 18b605d6329..4a88aece67a 100644 --- a/cli/src/commands/pipelines.ts +++ b/cli/src/commands/pipelines.ts @@ -109,6 +109,7 @@ interface IngestOptions extends PipelineOptions { parentCase?: string; workspaceRefJson?: string; blockedBy?: string; + blockedByKey?: string; } interface IngestBatchOptions extends PipelineOptions { @@ -336,6 +337,7 @@ export function registerPipelineCommands(program: Command): void { .option("--parent-case ", "Parent case ID") .option("--workspace-ref-json ", "Workspace ref JSON object") .option("--blocked-by ", "Comma-separated blocker case IDs") + .option("--blocked-by-key ", "Comma-separated blocker case keys") .action((pipeline: string, opts: IngestOptions) => withPipelineErrors(async () => { const ctx = resolvePipelineContext(opts); const pipelineId = await resolvePipelineId(ctx, pipeline); @@ -662,6 +664,7 @@ async function buildIngestBody(opts: IngestOptions): Promise { if (opts.fieldsJson || opts.fieldsFile) body.fields = await readJsonFromOptions(opts.fieldsJson, opts.fieldsFile); if (opts.workspaceRefJson) body.workspaceRef = parseJson(opts.workspaceRefJson); if (opts.blockedBy) body.blockedByCaseIds = parseCsv(opts.blockedBy); + if (opts.blockedByKey) body.blockedByCaseKeys = parseCsv(opts.blockedByKey); return body; } diff --git a/server/src/__tests__/pipelines-routes.test.ts b/server/src/__tests__/pipelines-routes.test.ts index 0369900fd74..42483a5f1eb 100644 --- a/server/src/__tests__/pipelines-routes.test.ts +++ b/server/src/__tests__/pipelines-routes.test.ts @@ -154,10 +154,16 @@ describeEmbeddedPostgres("pipeline routes", () => { .send({ caseKey: "case-1", title: "Case 1", fields: { channel: "blog" } }) .expect(201); const caseId = ingested.body.case.id; - await http + const batchIngest = await http .post(`/api/pipelines/${pipelineId}/cases/batch`) - .send({ items: [{ caseKey: "case-2", title: "Case 2" }, { caseKey: "case-3", title: "Case 3" }] }) + .send({ items: [{ caseKey: "case-2", title: "Case 2", blockedByCaseKeys: ["case-3"] }, { caseKey: "case-3", title: "Case 3" }] }) .expect(200); + expect(batchIngest.body[0].ok).toBe(true); + const routeBlockers = await db + .select() + .from(pipelineCaseBlockers) + .where(eq(pipelineCaseBlockers.caseId, batchIngest.body[0].case.id)); + expect(routeBlockers.map((row) => row.blockedByCaseId)).toEqual([batchIngest.body[1].case.id]); await http.get(`/api/pipelines/${pipelineId}/cases`).expect(200); await http.get(`/api/cases/${caseId}`).expect(200); await http.patch(`/api/cases/${caseId}`).send({ title: "Case 1 updated", expectedVersion: 1 }).expect(200); diff --git a/server/src/__tests__/pipelines-service.test.ts b/server/src/__tests__/pipelines-service.test.ts index 91129671b6e..629ab12244e 100644 --- a/server/src/__tests__/pipelines-service.test.ts +++ b/server/src/__tests__/pipelines-service.test.ts @@ -478,6 +478,123 @@ describeEmbeddedPostgres("pipelineService", () => { expect(moved.case.version).toBe(2); }); + it("resolves in-batch forward blocker case keys", async () => { + const { company, pipeline } = await seedPipeline(); + + const results = await svc.ingestCases({ + companyId: company.id, + pipelineId: pipeline.id, + items: [ + { caseKey: "tweet", title: "Tweet", blockedByCaseKeys: ["image", "post"] }, + { caseKey: "image", title: "Image" }, + { caseKey: "post", title: "Post" }, + ], + actor: userActor, + }); + + expect(results.map((result) => result.ok)).toEqual([true, true, true]); + const successful = results.filter((result): result is Extract<(typeof results)[number], { ok: true }> => result.ok); + const byKey = new Map(successful + .map((result) => [result.case.caseKey, result.case.id])); + const blockers = await db + .select() + .from(pipelineCaseBlockers) + .where(eq(pipelineCaseBlockers.caseId, byKey.get("tweet")!)); + expect(blockers.map((row) => row.blockedByCaseId).sort()).toEqual([ + byKey.get("image")!, + byKey.get("post")!, + ].sort()); + const events = await svc.listCaseEvents(company.id, byKey.get("tweet")!); + const blockersEvent = events.find((event) => event.type === "blockers_set"); + expect(blockersEvent?.payload).toMatchObject({ + blockedByCaseIds: expect.arrayContaining([byKey.get("image")!, byKey.get("post")!]), + blockedByCaseKeys: ["image", "post"], + }); + }); + + it("resolves blocker case keys against existing cases", async () => { + const { company, pipeline } = await seedPipeline(); + const asset = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "asset", + title: "Asset", + actor: userActor, + }); + + const created = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "tweet", + title: "Tweet", + blockedByCaseKeys: ["asset"], + actor: userActor, + }); + + const blockers = await db + .select() + .from(pipelineCaseBlockers) + .where(eq(pipelineCaseBlockers.caseId, created.case.id)); + expect(blockers.map((row) => row.blockedByCaseId)).toEqual([asset.case.id]); + }); + + it("fails only unresolved blocker-key rows in batch ingest", async () => { + const { company, pipeline } = await seedPipeline(); + + const results = await svc.ingestCases({ + companyId: company.id, + pipelineId: pipeline.id, + items: [ + { caseKey: "ok", title: "OK" }, + { caseKey: "missing", title: "Missing", blockedByCaseKeys: ["does-not-exist"] }, + { caseKey: "after", title: "After" }, + ], + actor: userActor, + }); + + expect(results[0]).toMatchObject({ ok: true }); + expect(results[1]).toMatchObject({ + ok: false, + caseKey: "missing", + error: { + status: 404, + details: { code: "blocker_case_key_not_found", missingCaseKeys: ["does-not-exist"] }, + }, + }); + expect(results[2]).toMatchObject({ ok: true }); + const rows = await db.select().from(pipelineCases).where(eq(pipelineCases.pipelineId, pipeline.id)); + expect(rows.map((row) => row.caseKey).sort()).toEqual(["after", "ok"]); + }); + + it("rejects blocker cycles declared by batch case keys", async () => { + const { company, pipeline } = await seedPipeline(); + + const results = await svc.ingestCases({ + companyId: company.id, + pipelineId: pipeline.id, + items: [ + { caseKey: "a", title: "A", blockedByCaseKeys: ["b"] }, + { caseKey: "b", title: "B", blockedByCaseKeys: ["a"] }, + ], + actor: userActor, + }); + + expect(results).toEqual([ + expect.objectContaining({ + ok: false, + caseKey: "a", + error: expect.objectContaining({ status: 409, details: { code: "blocker_cycle", blockedByCaseKeys: ["b"] } }), + }), + expect.objectContaining({ + ok: false, + caseKey: "b", + error: expect.objectContaining({ status: 409, details: { code: "blocker_cycle", blockedByCaseKeys: ["a"] } }), + }), + ]); + const rows = await db.select().from(pipelineCases).where(eq(pipelineCases.pipelineId, pipeline.id)); + expect(rows).toHaveLength(0); + }); + it("rejects parent and blocker cycles and enforces parent depth", async () => { const { company, pipeline } = await seedPipeline(); const a = await svc.ingestCase({ companyId: company.id, pipelineId: pipeline.id, caseKey: "a", title: "A", actor: userActor }); diff --git a/server/src/routes/pipelines.ts b/server/src/routes/pipelines.ts index a2cb9793ac9..bb640e3c057 100644 --- a/server/src/routes/pipelines.ts +++ b/server/src/routes/pipelines.ts @@ -50,6 +50,7 @@ const ingestCaseSchema = z.object({ parentCaseId: z.string().uuid().nullable().optional(), workspaceRef: jsonObjectSchema.nullable().optional(), blockedByCaseIds: z.array(z.string().uuid()).max(100).optional(), + blockedByCaseKeys: z.array(z.string().max(1_024)).max(100).optional(), }); const createPipelineSchema = z.object({ key: z.string().trim().min(1).max(120), diff --git a/server/src/services/pipelines.ts b/server/src/services/pipelines.ts index 02412b6d698..480d3c2ef07 100644 --- a/server/src/services/pipelines.ts +++ b/server/src/services/pipelines.ts @@ -626,6 +626,43 @@ async function validateBlockerSet( return uniqueBlockerIds; } +async function resolveBlockerCaseKeys( + db: PipelineDb, + input: { companyId: string; pipelineId: string; blockedByCaseKeys: string[] }, +) { + const uniqueKeys = [...new Set(input.blockedByCaseKeys)]; + if (uniqueKeys.length !== input.blockedByCaseKeys.length) { + throw unprocessable("Pipeline blocker key set contains duplicate cases", { code: "validation" }); + } + for (const key of uniqueKeys) assertCaseKey(key); + if (uniqueKeys.length === 0) return new Map(); + + const rows = await db + .select({ id: pipelineCases.id, caseKey: pipelineCases.caseKey }) + .from(pipelineCases) + .where(and( + eq(pipelineCases.companyId, input.companyId), + eq(pipelineCases.pipelineId, input.pipelineId), + inArray(pipelineCases.caseKey, uniqueKeys), + )); + if (rows.length !== uniqueKeys.length) { + throw new HttpError(404, "Pipeline blocker case key not found", { + code: "blocker_case_key_not_found", + missingCaseKeys: uniqueKeys.filter((key) => !rows.some((row) => row.caseKey === key)), + }); + } + return new Map(rows.map((row) => [row.caseKey, row.id])); +} + +function pipelineBatchError(error: unknown, fallbackCode = "unknown") { + const httpError = error as { status?: number; message?: string; details?: unknown }; + return { + status: httpError.status ?? 500, + message: httpError.message ?? "Unknown error", + details: httpError.details ?? { code: fallbackCode }, + }; +} + async function enqueueStageAutomationLedger( db: PipelineDb, input: { @@ -1254,6 +1291,7 @@ export function pipelineService(db: Db, deps: { heartbeat?: IssueAssignmentWakeu stageKey?: string | null; parentCaseId?: string | null; blockedByCaseIds?: string[]; + blockedByCaseKeys?: string[]; actor: PipelineActor; }) { assertJsonSize(input.fields ?? {}, "fields"); @@ -1265,10 +1303,18 @@ export function pipelineService(db: Db, deps: { heartbeat?: IssueAssignmentWakeu const pipeline = await getPipelineOrThrow(tx, input.companyId, input.pipelineId); if (pipeline.archivedAt) throw unprocessable("Pipeline is archived", { code: "pipeline_archived" }); await assertValidParentCase(tx, { companyId: input.companyId, parentCaseId: input.parentCaseId ?? null }); + const blockedByCaseKeyMap = await resolveBlockerCaseKeys(tx, { + companyId: input.companyId, + pipelineId: input.pipelineId, + blockedByCaseKeys: input.blockedByCaseKeys ?? [], + }); const blockedByCaseIds = await validateBlockerSet(tx, { companyId: input.companyId, caseId: "__new_case__", - blockedByCaseIds: input.blockedByCaseIds ?? [], + blockedByCaseIds: [ + ...(input.blockedByCaseIds ?? []), + ...Array.from(blockedByCaseKeyMap.values()), + ], }); const stage = input.stageKey ? await getStageByKeyOrThrow(tx, input.pipelineId, input.stageKey) @@ -1336,7 +1382,10 @@ export function pipelineService(db: Db, deps: { heartbeat?: IssueAssignmentWakeu caseId: inserted.id, type: "blockers_set", actor: input.actor, - payload: { blockedByCaseIds }, + payload: { + blockedByCaseIds, + ...(input.blockedByCaseKeys?.length ? { blockedByCaseKeys: input.blockedByCaseKeys } : {}), + }, }); } return { case: inserted, created: true }; @@ -1354,38 +1403,129 @@ export function pipelineService(db: Db, deps: { heartbeat?: IssueAssignmentWakeu stageKey?: string | null; parentCaseId?: string | null; blockedByCaseIds?: string[]; + blockedByCaseKeys?: string[]; }>; actor: PipelineActor; }) { if (input.items.length > MAX_BATCH_INGEST) { throw unprocessable("Batch ingest supports at most 200 items", { code: "validation" }); } + type BatchIngestResult = + | Awaited> & { ok: true } + | { ok: false; caseKey: string | null; error: Record }; const seen = new Set(); - return Promise.all(input.items.map(async (item) => { + const results = new Array(input.items.length); + const pending = new Set(); + const firstBatchKeyIndexes = new Map(); + for (const [index, item] of input.items.entries()) { const key = item.caseKey ?? null; if (key) { + try { + assertCaseKey(key); + } catch (error) { + results[index] = { ok: false as const, caseKey: key, error: pipelineBatchError(error, "validation") }; + continue; + } if (seen.has(key)) { - return { ok: false as const, caseKey: key, error: { code: "duplicate_batch_key" } }; + results[index] = { ok: false as const, caseKey: key, error: { code: "duplicate_batch_key" } }; + continue; } seen.add(key); + firstBatchKeyIndexes.set(key, index); } + pending.add(index); + } + + const referencedKeys = [...new Set(input.items.flatMap((item) => item.blockedByCaseKeys ?? []))]; + const resolvedCaseIdsByKey = new Map(); + for (const key of referencedKeys) { try { - const result = await service.ingestCase({ - ...item, - companyId: input.companyId, - pipelineId: input.pipelineId, - actor: input.actor, + assertCaseKey(key); + } catch { + // Leave invalid keys unresolved so only rows that reference them fail below. + } + } + const validReferencedKeys = referencedKeys.filter((key) => { + try { + assertCaseKey(key); + return true; + } catch { + return false; + } + }); + if (validReferencedKeys.length > 0) { + const rows = await db + .select({ id: pipelineCases.id, caseKey: pipelineCases.caseKey }) + .from(pipelineCases) + .where(and( + eq(pipelineCases.companyId, input.companyId), + eq(pipelineCases.pipelineId, input.pipelineId), + inArray(pipelineCases.caseKey, validReferencedKeys), + )); + for (const row of rows) resolvedCaseIdsByKey.set(row.caseKey, row.id); + } + + while (pending.size > 0) { + let progressed = false; + for (const index of [...pending]) { + const item = input.items[index]!; + const missingKeys = (item.blockedByCaseKeys ?? []).filter((key) => !resolvedCaseIdsByKey.has(key)); + if (missingKeys.length > 0) continue; + + pending.delete(index); + progressed = true; + const key = item.caseKey ?? null; + try { + const result = await service.ingestCase({ + ...item, + companyId: input.companyId, + pipelineId: input.pipelineId, + actor: input.actor, + }); + if (key) resolvedCaseIdsByKey.set(key, result.case.id); + results[index] = { ok: true as const, ...result }; + } catch (error) { + results[index] = { ok: false as const, caseKey: key, error: pipelineBatchError(error) }; + } + } + if (progressed) continue; + + const stuck = new Set(pending); + for (const index of [...stuck]) { + const item = input.items[index]!; + const key = item.caseKey ?? null; + const missingKeys = (item.blockedByCaseKeys ?? []).filter((blockedByCaseKey) => !resolvedCaseIdsByKey.has(blockedByCaseKey)); + const cyclicKeys = missingKeys.filter((blockedByCaseKey) => { + const blockerIndex = firstBatchKeyIndexes.get(blockedByCaseKey); + return blockerIndex !== undefined && stuck.has(blockerIndex); }); - return { ok: true as const, ...result }; - } catch (error) { - const httpError = error as { status?: number; message?: string; details?: unknown }; - return { + results[index] = { ok: false as const, caseKey: key, - error: { status: httpError.status ?? 500, message: httpError.message ?? "Unknown error", details: httpError.details }, + error: cyclicKeys.length === missingKeys.length + ? { + status: 409, + message: "Pipeline blocker cycle detected", + details: { code: "blocker_cycle", blockedByCaseKeys: missingKeys }, + } + : { + status: 404, + message: "Pipeline blocker case key not found", + details: { + code: "blocker_case_key_not_found", + missingCaseKeys: missingKeys.filter((blockedByCaseKey) => !cyclicKeys.includes(blockedByCaseKey)), + }, + }, }; + pending.delete(index); } - })); + } + + return results.map((result, index) => result ?? { + ok: false as const, + caseKey: input.items[index]?.caseKey ?? null, + error: { status: 500, message: "Unknown error", details: { code: "unknown" } }, + }); }, async patchCaseContent(input: {