test(run-engine): run the waitpoint suite against both coordinators

Each case in the waitpoint suite now runs twice, once against Postgres and once
against the store, with the mint kind and the run id shape following the arm.

Three things had to change for the store arm to assert anything real. A store
RUN waitpoint derives its id from its anchor run, so a literal legacy run id
would mint a legacy waitpoint and the case would prove nothing. Waitpoint reads
go through a helper that reads whichever system holds the waitpoint, since
there is no row for the store to read. And block-edge reads union both systems
rather than switching, because a run can hold one edge in each at once and a
test seeing half of them would report the wrong count.

Author: d-cs

internal-packages/run-engine/src/engine/tests/helpers/engineFactory.ts

@@ -1,4 +1,7 @@
 import type { RedisOptions } from "@internal/redis";
+import type { PrismaClient, Waitpoint } from "@trigger.dev/database";
+import { WaitpointStoreCoordinator } from "../../waitpointCoordinator/storeCoordinator.js";
+import { toPrismaWaitpoint } from "../../waitpointCoordinator/waitpointShape.js";
 import { generateRunOpsId, parseWaitpointId, RunId } from "@trigger.dev/core/v3/isomorphic";
 import { RunEngine } from "../../index.js";
 import type { RunEngineOptions } from "../../types.js";
@@ -59,3 +62,78 @@ export function assertStoreResident(waitpointId: string): void {
     );
   }
 }
+
+type ArmRead = { arm: WaitpointArm; prisma: PrismaClient; redisOptions: RedisOptions };
+
+/**
+ * Read a waitpoint from whichever system holds it.
+ *
+ * A test that reads `prisma.waitpoint` directly is asserting against Postgres, and the
+ * store path writes no row there for RUN, BATCH or DATETIME. Going through here lets one
+ * expectation hold on both arms.
+ */
+export async function readWaitpointForArm(
+  args: ArmRead & { waitpointId: string }
+): Promise<Waitpoint | null> {
+  if (parseWaitpointId(args.waitpointId).format === "legacy") {
+    return args.prisma.waitpoint.findFirst({ where: { id: args.waitpointId } });
+  }
+
+  const store = new WaitpointStoreCoordinator({ redisOptions: args.redisOptions });
+  try {
+    const held = await store.readWaitpoint(args.waitpointId);
+    return held ? toPrismaWaitpoint(held.record, held.status, held.completion) : null;
+  } finally {
+    await store.quit();
+  }
+}
+
+export type ArmBlockEdge = {
+  waitpointId: string;
+  batchIndex: number | null;
+  waitpoint: Waitpoint;
+};
+
+/**
+ * A run's blocking edges, from both systems.
+ *
+ * Always unions the two rather than switching on the arm, because a run can hold one edge
+ * in each at the same time and a test that saw only half would report the wrong count.
+ */
+export async function readRunBlockEdgesForArm(
+  args: ArmRead & { runId: string }
+): Promise<ArmBlockEdge[]> {
+  const legacy = await args.prisma.taskRunWaitpoint.findMany({
+    where: { taskRunId: args.runId },
+    include: { waitpoint: true },
+  });
+
+  const edges: ArmBlockEdge[] = legacy.map((edge) => ({
+    waitpointId: edge.waitpointId,
+    batchIndex: edge.batchIndex,
+    waitpoint: edge.waitpoint,
+  }));
+
+  if (args.arm !== "store") {
+    return edges;
+  }
+
+  const store = new WaitpointStoreCoordinator({ redisOptions: args.redisOptions });
+  try {
+    const state = await store.readBlockState(args.runId);
+    for (const edge of state.edges) {
+      const held = await store.readWaitpoint(edge.waitpointId);
+      if (held) {
+        edges.push({
+          waitpointId: edge.waitpointId,
+          batchIndex: edge.batchIndex ?? null,
+          waitpoint: toPrismaWaitpoint(held.record, held.status, held.completion),
+        });
+      }
+    }
+  } finally {
+    await store.quit();
+  }
+
+  return edges;
+}