Add pipeline blocker key ingest wiring

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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 {

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 <id>", "Parent case ID")
       .option("--workspace-ref-json <json>", "Workspace ref JSON object")
       .option("--blocked-by <csv>", "Comma-separated blocker case IDs")
+      .option("--blocked-by-key <csv>", "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<JsonObject> {
   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;
 }
 

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);

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 });

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),

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<string, string>();
+
+  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<ReturnType<typeof service.ingestCase>> & { ok: true }
+        | { ok: false; caseKey: string | null; error: Record<string, unknown> };
       const seen = new Set<string>();
-      return Promise.all(input.items.map(async (item) => {
+      const results = new Array<BatchIngestResult | undefined>(input.items.length);
+      const pending = new Set<number>();
+      const firstBatchKeyIndexes = new Map<string, number>();
+      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<string, string>();
+      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: {