test(webapp): avoid repeated replication setup

apps/webapp/test/apiRunListPresenter.readthrough.test.ts

@@ -0,0 +1,142 @@
+import { describe, expect, vi } from "vitest";
+
+// The presenter graph imports `~/db.server` singletons even though the asserted reads use explicit
+// clients. These lazy proxies delegate every access to the real per-test Postgres containers.
+const legacyReplicaHolder = vi.hoisted(() => ({ client: undefined as any }));
+const newClientHolder = vi.hoisted(() => ({ client: undefined as any }));
+const clickhouseHolder = vi.hoisted(() => ({ client: undefined as any }));
+
+vi.mock("~/services/clickhouse/clickhouseFactoryInstance.server", () => ({
+  clickhouseFactory: {
+    getClickhouseForOrganization: async () => {
+      if (!clickhouseHolder.client) {
+        throw new Error("clickhouseHolder.client not set for this test");
+      }
+      return clickhouseHolder.client;
+    },
+  },
+}));
+
+vi.mock("~/db.server", async () => {
+  const { Prisma } = await import("@trigger.dev/database");
+  const lazyProxy = (holder: { client: any }, label: string) =>
+    new Proxy(
+      {},
+      {
+        get(_target, property) {
+          if (!holder.client) {
+            throw new Error(`${label} not set for this test`);
+          }
+          return holder.client[property];
+        },
+      }
+    );
+  const replicaProxy = lazyProxy(legacyReplicaHolder, "legacyReplicaHolder.client");
+  const newProxy = lazyProxy(newClientHolder, "newClientHolder.client");
+
+  return {
+    prisma: replicaProxy,
+    $replica: replicaProxy,
+    runOpsNewPrisma: newProxy,
+    runOpsNewReplica: newProxy,
+    runOpsLegacyPrisma: replicaProxy,
+    runOpsLegacyReplica: replicaProxy,
+    sqlDatabaseSchema: Prisma.sql([`public`]),
+  };
+});
+
+import { createPostgresContainer, replicationContainerTest } from "@internal/testcontainers";
+import { PrismaClient } from "@trigger.dev/database";
+import { setTimeout } from "node:timers/promises";
+import { CURRENT_API_VERSION } from "~/api/versions";
+import { ApiRunListPresenter } from "~/presenters/v3/ApiRunListPresenter.server";
+import { createRun, mirrorParents, seedParents } from "./helpers/apiRunListPresenterTestHelpers";
+import { setupClickhouseReplication } from "./utils/replicationUtils";
+
+vi.setConfig({ testTimeout: 90_000 });
+
+describe("ApiRunListPresenter public /runs routed read-through", () => {
+  replicationContainerTest(
+    "public payload lists run-ops rows served via the routed store (NEW + legacy union)",
+    async ({ clickhouseContainer, redisOptions, postgresContainer, prisma, network }) => {
+      const { clickhouse } = await setupClickhouseReplication({
+        prisma,
+        databaseUrl: postgresContainer.getConnectionUri(),
+        clickhouseUrl: clickhouseContainer.getConnectionUrl(),
+        redisOptions,
+      });
+
+      const { url: newUrl } = await createPostgresContainer(network, {
+        imageTag: "docker.io/postgres:17",
+      });
+      const prismaNew = new PrismaClient({ datasources: { db: { url: newUrl } } });
+      legacyReplicaHolder.client = prisma;
+      clickhouseHolder.client = clickhouse;
+      newClientHolder.client = prismaNew;
+
+      try {
+        const ctx = await seedParents(prisma, "hydrate");
+        await mirrorParents(prismaNew, ctx, "hydrate");
+
+        // PG14 is the logical-replication source, so ClickHouse receives the complete ID set.
+        const legacyOnlyA = await createRun(prisma, ctx, { friendlyId: "run_legacyA" });
+        const legacyOnlyB = await createRun(prisma, ctx, { friendlyId: "run_legacyB" });
+        const migratedA = await createRun(prisma, ctx, { friendlyId: "run_newA" });
+        const migratedB = await createRun(prisma, ctx, { friendlyId: "run_newB" });
+
+        // The routed PG17 rows use distinguishing values that prove NEW hydration won.
+        await createRun(prismaNew, ctx, {
+          friendlyId: "run_newA",
+          taskIdentifier: "my-task-NEW",
+        });
+        await createRun(prismaNew, ctx, {
+          friendlyId: "run_newB",
+          taskIdentifier: "my-task-NEW",
+        });
+        await prismaNew.taskRun.update({
+          where: { friendlyId: "run_newA" },
+          data: { id: migratedA.id },
+        });
+        await prismaNew.taskRun.update({
+          where: { friendlyId: "run_newB" },
+          data: { id: migratedB.id },
+        });
+
+        await setTimeout(1500);
+
+        const presenter = new ApiRunListPresenter(prisma, prisma, {
+          newClient: prismaNew,
+          legacyReplica: prisma,
+          splitEnabled: true,
+        });
+
+        const result = await presenter.call(
+          { id: ctx.projectId },
+          { "page[size]": 10 } as any,
+          CURRENT_API_VERSION,
+          { id: ctx.environmentId, organizationId: ctx.organizationId }
+        );
+
+        const expectedFriendlyIds = [
+          { id: migratedA.id, friendlyId: "run_newA" },
+          { id: migratedB.id, friendlyId: "run_newB" },
+          { id: legacyOnlyA.id, friendlyId: "run_legacyA" },
+          { id: legacyOnlyB.id, friendlyId: "run_legacyB" },
+        ]
+          .sort((a, b) => (a.id < b.id ? 1 : a.id > b.id ? -1 : 0))
+          .map((run) => run.friendlyId);
+        expect(result.data.map((run) => run.id)).toEqual(expectedFriendlyIds);
+
+        const migratedRow = result.data.find((run) => run.id === "run_newA");
+        expect(migratedRow?.taskIdentifier).toBe("my-task-NEW");
+        expect(migratedRow?.taskKind).toBe("STANDARD");
+        expect(result.data.find((run) => run.id === "run_legacyA")?.taskIdentifier).toBe("my-task");
+
+        expect(result.pagination).toHaveProperty("next");
+        expect(result.pagination).toHaveProperty("previous");
+      } finally {
+        await prismaNew.$disconnect();
+      }
+    }
+  );
+});

apps/webapp/test/apiRunListPresenter.test.ts

@@ -1,20 +1,11 @@
 import { describe, expect, vi } from "vitest";
 
-// The presenter graph imports `~/v3/runStore.server` (via RunsRepository) which imports
-// `~/db.server` at load, and the presenter itself reaches `~/db.server`'s `$replica` singleton
-// through `findDisplayableEnvironment` and `getTaskIdentifiers`. Stub the module so those
-// singleton reads resolve. This is the ONLY mock — the DB is NEVER mocked; the proxy delegates
-// to the per-test REAL legacy (PG14) container so the env-lookup + task-identifier reads hit a
-// real database. Everything asserted runs against real containers. Mirrors
-// nextRunListPresenter.readthrough.test.ts.
+// The presenter graph imports `~/db.server` singletons even though these tests pass explicit real
+// clients. The proxies keep those singleton reads on the current warm Postgres fixture.
 const legacyReplicaHolder = vi.hoisted(() => ({ client: undefined as any }));
 const newClientHolder = vi.hoisted(() => ({ client: undefined as any }));
-// `ApiRunListPresenter` resolves its read ClickHouse internally via the `clickhouseFactory`
-// singleton (which imports `~/env.server` and binds to a process-wide default client). Stub the
-// instance module so `getClickhouseForOrganization` returns the per-test container's ClickHouse
-// handle (set by each test before calling). This is a module-resolution shim — the ClickHouse is
-// a REAL testcontainer, never mocked — mirroring the `~/db.server` stub below.
 const clickhouseHolder = vi.hoisted(() => ({ client: undefined as any }));
+
 vi.mock("~/services/clickhouse/clickhouseFactoryInstance.server", () => ({
   clickhouseFactory: {
     getClickhouseForOrganization: async () => {
@@ -25,22 +16,24 @@ vi.mock("~/services/clickhouse/clickhouseFactoryInstance.server", () => ({
     },
   },
 }));
+
 vi.mock("~/db.server", async () => {
   const { Prisma } = await import("@trigger.dev/database");
   const lazyProxy = (holder: { client: any }, label: string) =>
     new Proxy(
       {},
       {
-        get(_t, prop) {
+        get(_target, property) {
           if (!holder.client) {
             throw new Error(`${label} not set for this test`);
           }
-          return holder.client[prop];
+          return holder.client[property];
         },
       }
     );
   const replicaProxy = lazyProxy(legacyReplicaHolder, "legacyReplicaHolder.client");
   const newProxy = lazyProxy(newClientHolder, "newClientHolder.client");
+
   return {
     prisma: replicaProxy,
     $replica: replicaProxy,
@@ -52,347 +45,101 @@ vi.mock("~/db.server", async () => {
   };
 });
 
-import { createPostgresContainer, replicationContainerTest } from "@internal/testcontainers";
-import { PrismaClient } from "@trigger.dev/database";
-import { setTimeout } from "node:timers/promises";
+import { ClickHouse } from "@internal/clickhouse";
+import { containerTest } from "@internal/testcontainers";
 import { CURRENT_API_VERSION } from "~/api/versions";
 import { ApiRunListPresenter } from "~/presenters/v3/ApiRunListPresenter.server";
-import { setupClickhouseReplication } from "./utils/replicationUtils";
+import {
+  addEnvironment,
+  createRun,
+  insertTaskRunV2Rows,
+  seedParents,
+} from "./helpers/apiRunListPresenterTestHelpers";
 
 vi.setConfig({ testTimeout: 90_000 });
 
-type SeedContext = {
-  organizationId: string;
-  projectId: string;
-  environmentId: string;
-  environmentSlug: string;
-};
-
-/**
- * Creates the org/project/env parents on a single prisma client. TaskRun FKs require these to
- * exist on every DB a run lives on, so identical parents (same ids) are seeded on both the
- * legacy (PG14) and new (PG17) databases.
- */
-async function seedParents(
-  prisma: PrismaClient,
-  slug: string,
-  envSlug = `env-${slug}`
-): Promise<SeedContext> {
-  const organization = await prisma.organization.create({
-    data: { title: `org-${slug}`, slug: `org-${slug}` },
+function setupClients(prisma: unknown, clickhouseUrl: string): ClickHouse {
+  const clickhouse = new ClickHouse({
+    url: clickhouseUrl,
+    name: "api-run-list-presenter-test",
+    compression: { request: true },
   });
-  const project = await prisma.project.create({
-    data: {
-      name: `proj-${slug}`,
-      slug: `proj-${slug}`,
-      organizationId: organization.id,
-      externalRef: `proj-${slug}`,
-    },
-  });
-  const runtimeEnvironment = await prisma.runtimeEnvironment.create({
-    data: {
-      slug: envSlug,
-      type: "DEVELOPMENT",
-      projectId: project.id,
-      organizationId: organization.id,
-      apiKey: `tr_dev_${slug}`,
-      pkApiKey: `pk_dev_${slug}`,
-      shortcode: `sc-${slug}`,
-    },
-  });
-
-  return {
-    organizationId: organization.id,
-    projectId: project.id,
-    environmentId: runtimeEnvironment.id,
-    environmentSlug: runtimeEnvironment.slug,
-  };
+  legacyReplicaHolder.client = prisma;
+  newClientHolder.client = prisma;
+  clickhouseHolder.client = clickhouse;
+  return clickhouse;
 }
 
-/** Adds an extra RuntimeEnvironment (control-plane row) to an existing project. */
-async function addEnvironment(
-  prisma: PrismaClient,
-  ctx: SeedContext,
-  slug: string,
-  envSlug: string
-): Promise<string> {
-  const env = await prisma.runtimeEnvironment.create({
-    data: {
-      slug: envSlug,
-      type: "STAGING",
-      projectId: ctx.projectId,
-      organizationId: ctx.organizationId,
-      apiKey: `tr_${envSlug}_${slug}`,
-      pkApiKey: `pk_${envSlug}_${slug}`,
-      shortcode: `sc-${envSlug}-${slug}`,
-    },
-  });
-  return env.id;
-}
-
-/** Mirrors the org/project/env parents onto a second DB with the SAME ids. */
-async function mirrorParents(prisma: PrismaClient, ctx: SeedContext, slug: string): Promise<void> {
-  await prisma.organization.create({
-    data: { id: ctx.organizationId, title: `org-${slug}`, slug: `org-${slug}` },
-  });
-  await prisma.project.create({
-    data: {
-      id: ctx.projectId,
-      name: `proj-${slug}`,
-      slug: `proj-${slug}`,
-      organizationId: ctx.organizationId,
-      externalRef: `proj-${slug}`,
-    },
-  });
-  await prisma.runtimeEnvironment.create({
-    data: {
-      id: ctx.environmentId,
-      slug: ctx.environmentSlug,
-      type: "DEVELOPMENT",
-      projectId: ctx.projectId,
-      organizationId: ctx.organizationId,
-      apiKey: `tr_dev_${slug}_b`,
-      pkApiKey: `pk_dev_${slug}_b`,
-      shortcode: `sc-${slug}-b`,
-    },
-  });
-}
-
-async function createRun(
-  prisma: PrismaClient,
-  ctx: SeedContext,
-  run: {
-    friendlyId: string;
-    taskIdentifier?: string;
-    status?: any;
-    runtimeEnvironmentId?: string;
-  }
-) {
-  return prisma.taskRun.create({
-    data: {
-      friendlyId: run.friendlyId,
-      taskIdentifier: run.taskIdentifier ?? "my-task",
-      status: run.status ?? "PENDING",
-      payload: JSON.stringify({ foo: run.friendlyId }),
-      traceId: run.friendlyId,
-      spanId: run.friendlyId,
-      queue: "test",
-      runTags: [],
-      runtimeEnvironmentId: run.runtimeEnvironmentId ?? ctx.environmentId,
-      projectId: ctx.projectId,
-      organizationId: ctx.organizationId,
-      environmentType: "DEVELOPMENT",
-      engine: "V2",
-    },
-  });
-}
-
-describe("ApiRunListPresenter public /runs list (PG14 legacy + PG17 new)", () => {
-  // Public list serves run-ops rows through the routed store. The
-  // forwarded readThroughDeps thread the dual-DB union into NextRunListPresenter; the public
-  // payload (`{ data, pagination }`) must list the NEW ∪ legacy union, proving the public API
-  // surfaces routed run-ops rows. The migrated/straggler rows (run_newA/run_newB) live on BOTH
-  // DBs with the same id + friendlyId but a DISTINGUISHING taskIdentifier ("my-task-NEW" on PG17),
-  // so a row served from the threaded newClient is identifiable in the public payload.
-  replicationContainerTest(
-    "public payload lists run-ops rows served via the routed store (NEW + legacy union)",
-    async ({ clickhouseContainer, redisOptions, postgresContainer, prisma, network }) => {
-      const { clickhouse } = await setupClickhouseReplication({
-        prisma,
-        databaseUrl: postgresContainer.getConnectionUri(),
-        clickhouseUrl: clickhouseContainer.getConnectionUrl(),
-        redisOptions,
-      });
-
-      const { url: newUrl } = await createPostgresContainer(network, {
-        imageTag: "docker.io/postgres:17",
-      });
-      const prismaNew = new PrismaClient({ datasources: { db: { url: newUrl } } });
-      legacyReplicaHolder.client = prisma;
-      clickhouseHolder.client = clickhouse;
-      // The routed store's default known-migrated probe reads `runOpsNewPrisma` -> PG17.
-      newClientHolder.client = prismaNew;
-
-      try {
-        const ctx = await seedParents(prisma, "hydrate");
-        await mirrorParents(prismaNew, ctx, "hydrate");
-
-        // All four runs land on PG14 (legacy + replication source -> CH gets the full id-set).
-        const legacyOnlyA = await createRun(prisma, ctx, { friendlyId: "run_legacyA" });
-        const legacyOnlyB = await createRun(prisma, ctx, { friendlyId: "run_legacyB" });
-        const migratedA = await createRun(prisma, ctx, { friendlyId: "run_newA" });
-        const migratedB = await createRun(prisma, ctx, { friendlyId: "run_newB" });
-
-        // The two "migrated" runs also live on NEW (authoritative during retention), same ids +
-        // friendlyIds, but a DISTINGUISHING taskIdentifier so a row served from PG17 is
-        // identifiable in the public payload.
-        await createRun(prismaNew, ctx, { friendlyId: "run_newA", taskIdentifier: "my-task-NEW" });
-        await createRun(prismaNew, ctx, { friendlyId: "run_newB", taskIdentifier: "my-task-NEW" });
-        await prismaNew.taskRun.update({
-          where: { friendlyId: "run_newA" },
-          data: { id: migratedA.id },
-        });
-        await prismaNew.taskRun.update({
-          where: { friendlyId: "run_newB" },
-          data: { id: migratedB.id },
-        });
-
-        // Wait for CH replication so the id-set page is non-empty.
-        await setTimeout(1500);
-
-        const presenter = new ApiRunListPresenter(prisma, prisma, {
-          newClient: prismaNew,
-          legacyReplica: prisma,
-          splitEnabled: true,
-        });
-
-        const result = await presenter.call(
-          { id: ctx.projectId },
-          { "page[size]": 10 } as any,
-          CURRENT_API_VERSION,
-          { id: ctx.environmentId, organizationId: ctx.organizationId }
-        );
-
-        // The public payload lists runs by `id` = `run.friendlyId`, id-desc ordered.
-        const expectedFriendlyIds = [
-          { id: migratedA.id, friendlyId: "run_newA" },
-          { id: migratedB.id, friendlyId: "run_newB" },
-          { id: legacyOnlyA.id, friendlyId: "run_legacyA" },
-          { id: legacyOnlyB.id, friendlyId: "run_legacyB" },
-        ]
-          .sort((a, b) => (a.id < b.id ? 1 : a.id > b.id ? -1 : 0))
-          .map((r) => r.friendlyId);
-        expect(result.data.map((r) => r.id)).toEqual(expectedFriendlyIds);
-
-        // The migrated rows must carry the PG17-only taskIdentifier — only possible if the public
-        // path hydrated them through the threaded newClient (PG17). taskKind falls back to STANDARD.
-        const migratedRow = result.data.find((r) => r.id === "run_newA");
-        expect(migratedRow?.taskIdentifier).toBe("my-task-NEW");
-        expect(migratedRow?.taskKind).toBe("STANDARD");
-        // The legacy-only rows surface from PG14, proving the legacyReplica is also exercised.
-        expect(result.data.find((r) => r.id === "run_legacyA")?.taskIdentifier).toBe("my-task");
-
-        // Pagination shape is present.
-        expect(result.pagination).toHaveProperty("next");
-        expect(result.pagination).toHaveProperty("previous");
-      } finally {
-        await prismaNew.$disconnect();
-      }
-    }
-  );
-
-  // Genuinely-empty env returns { data: [], pagination } without error. Exercises the
-  // empty-state probe beneath NextRunListPresenter (no rows on either DB; empty CH page).
-  replicationContainerTest(
+describe("ApiRunListPresenter public /runs list", () => {
+  containerTest(
     "genuinely-empty env returns { data: [], pagination } without error",
-    async ({ clickhouseContainer, redisOptions, postgresContainer, prisma, network }) => {
-      const { clickhouse } = await setupClickhouseReplication({
-        prisma,
-        databaseUrl: postgresContainer.getConnectionUri(),
-        clickhouseUrl: clickhouseContainer.getConnectionUrl(),
-        redisOptions,
+    async ({ clickhouseContainer, prisma }) => {
+      setupClients(prisma, clickhouseContainer.getConnectionUrl());
+      const ctx = await seedParents(prisma, "empty");
+
+      // Keep the split/read-through branch active while both real clients point at the empty DB.
+      const presenter = new ApiRunListPresenter(prisma, prisma, {
+        newClient: prisma,
+        legacyReplica: prisma,
+        splitEnabled: true,
       });
 
-      const { url: newUrl } = await createPostgresContainer(network, {
-        imageTag: "docker.io/postgres:17",
-      });
-      const prismaNew = new PrismaClient({ datasources: { db: { url: newUrl } } });
-      legacyReplicaHolder.client = prisma;
-      clickhouseHolder.client = clickhouse;
+      const result = await presenter.call(
+        { id: ctx.projectId },
+        { "page[size]": 10 } as any,
+        CURRENT_API_VERSION,
+        { id: ctx.environmentId, organizationId: ctx.organizationId }
+      );
 
-      try {
-        const ctx = await seedParents(prisma, "empty");
-        await mirrorParents(prismaNew, ctx, "empty");
-
-        const presenter = new ApiRunListPresenter(prisma, prisma, {
-          newClient: prismaNew,
-          legacyReplica: prisma,
-          splitEnabled: true,
-        });
-
-        const result = await presenter.call(
-          { id: ctx.projectId },
-          { "page[size]": 10 } as any,
-          CURRENT_API_VERSION,
-          { id: ctx.environmentId, organizationId: ctx.organizationId }
-        );
-
-        expect(result.data).toEqual([]);
-        expect(result.pagination).toHaveProperty("next");
-        expect(result.pagination).toHaveProperty("previous");
-      } finally {
-        await prismaNew.$disconnect();
-      }
+      expect(result.data).toEqual([]);
+      expect(result.pagination).toHaveProperty("next");
+      expect(result.pagination).toHaveProperty("previous");
     }
   );
 
-  // Env scoping unchanged: the control-plane runtimeEnvironment.findMany lookup
-  // resolves the requested env via the `_replica` handle (NOT routed), with the 4th `environment`
-  // arg omitted to force that branch. Result is scoped to the requested env only.
-  replicationContainerTest(
+  containerTest(
     "env scoping resolves via the control-plane _replica handle (filter[env], 4th arg omitted)",
-    async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
-      const { clickhouse } = await setupClickhouseReplication({
-        prisma,
-        databaseUrl: postgresContainer.getConnectionUri(),
-        clickhouseUrl: clickhouseContainer.getConnectionUrl(),
-        redisOptions,
-      });
-
-      legacyReplicaHolder.client = prisma;
-      clickhouseHolder.client = clickhouse;
-
+    async ({ clickhouseContainer, prisma }) => {
+      const clickhouse = setupClients(prisma, clickhouseContainer.getConnectionUrl());
       const ctx = await seedParents(prisma, "scoping", "prod");
-      const stagingEnvId = await addEnvironment(prisma, ctx, "scoping", "staging");
+      const stagingEnvironmentId = await addEnvironment(prisma, ctx, "scoping", "staging");
 
-      // Runs in prod only; a run in staging must NOT surface when filter[env]=prod.
-      await createRun(prisma, ctx, { friendlyId: "run_prod1" });
-      await createRun(prisma, ctx, { friendlyId: "run_prod2" });
-      await createRun(prisma, ctx, {
-        friendlyId: "run_staging",
-        runtimeEnvironmentId: stagingEnvId,
-      });
+      // The Postgres rows exercise real hydration; matching ClickHouse rows provide the list IDs.
+      const runs = await Promise.all([
+        createRun(prisma, ctx, { friendlyId: "run_prod1" }),
+        createRun(prisma, ctx, { friendlyId: "run_prod2" }),
+        createRun(prisma, ctx, {
+          friendlyId: "run_staging",
+          runtimeEnvironmentId: stagingEnvironmentId,
+        }),
+      ]);
+      await insertTaskRunV2Rows(clickhouse, runs);
 
-      await setTimeout(1500);
-
-      // Single-handle passthrough; the env lookup runs on `_replica` (= prisma) via findMany.
       const presenter = new ApiRunListPresenter(prisma, prisma);
 
-      // 4th `environment` arg OMITTED -> forces the runtimeEnvironment.findMany branch.
+      // Omitting the fourth argument forces the control-plane runtimeEnvironment.findMany branch.
       const result = await presenter.call(
         { id: ctx.projectId },
         { "page[size]": 10, "filter[env]": ["prod"] } as any,
         CURRENT_API_VERSION
       );
 
-      // Scoped to the resolved prod env only.
-      expect(result.data.map((r) => r.id).sort()).toEqual(["run_prod1", "run_prod2"]);
+      expect(result.data.map((run) => run.id).sort()).toEqual(["run_prod1", "run_prod2"]);
     }
   );
 
-  // Passthrough (single-DB): two-arg-style construction (no readThroughDeps) ->
-  // NextRunListPresenter receives undefined deps -> byte-identical single-DB path. The public
-  // { data, pagination } shape is unchanged.
-  replicationContainerTest(
+  containerTest(
     "single-DB passthrough: no readThroughDeps lists the seeded runs unchanged",
-    async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
-      const { clickhouse } = await setupClickhouseReplication({
-        prisma,
-        databaseUrl: postgresContainer.getConnectionUri(),
-        clickhouseUrl: clickhouseContainer.getConnectionUrl(),
-        redisOptions,
-      });
-
-      legacyReplicaHolder.client = prisma;
-      clickhouseHolder.client = clickhouse;
-
+    async ({ clickhouseContainer, prisma }) => {
+      const clickhouse = setupClients(prisma, clickhouseContainer.getConnectionUrl());
       const ctx = await seedParents(prisma, "passthrough");
-      await createRun(prisma, ctx, { friendlyId: "run_pt1" });
-      await createRun(prisma, ctx, { friendlyId: "run_pt2" });
+      const runs = await Promise.all([
+        createRun(prisma, ctx, { friendlyId: "run_pt1" }),
+        createRun(prisma, ctx, { friendlyId: "run_pt2" }),
+      ]);
+      await insertTaskRunV2Rows(clickhouse, runs);
 
-      await setTimeout(1500);
-
-      // No readThroughDeps -> passthrough, exactly as the routes construct it today.
+      // No readThroughDeps preserves the single-database path used by existing callers.
       const presenter = new ApiRunListPresenter(prisma, prisma);
 
       const result = await presenter.call(
@@ -402,7 +149,7 @@ describe("ApiRunListPresenter public /runs list (PG14 legacy + PG17 new)", () =>
         { id: ctx.environmentId, organizationId: ctx.organizationId }
       );
 
-      expect(result.data.map((r) => r.id).sort()).toEqual(["run_pt1", "run_pt2"]);
+      expect(result.data.map((run) => run.id).sort()).toEqual(["run_pt1", "run_pt2"]);
       expect(result).toHaveProperty("pagination");
       expect(result.pagination).toHaveProperty("next");
       expect(result.pagination).toHaveProperty("previous");

apps/webapp/test/helpers/apiRunListPresenterTestHelpers.ts

@@ -0,0 +1,178 @@
+import type { ClickHouse, TaskRunV2 } from "@internal/clickhouse";
+import type { PrismaClient, TaskRun, TaskRunStatus } from "@trigger.dev/database";
+import { z } from "zod";
+
+export type SeedContext = {
+  organizationId: string;
+  projectId: string;
+  environmentId: string;
+  environmentSlug: string;
+};
+
+/** Creates the org/project/environment parents needed by TaskRun foreign keys. */
+export async function seedParents(
+  prisma: PrismaClient,
+  slug: string,
+  envSlug = `env-${slug}`
+): Promise<SeedContext> {
+  const organization = await prisma.organization.create({
+    data: { title: `org-${slug}`, slug: `org-${slug}` },
+  });
+  const project = await prisma.project.create({
+    data: {
+      name: `proj-${slug}`,
+      slug: `proj-${slug}`,
+      organizationId: organization.id,
+      externalRef: `proj-${slug}`,
+    },
+  });
+  const runtimeEnvironment = await prisma.runtimeEnvironment.create({
+    data: {
+      slug: envSlug,
+      type: "DEVELOPMENT",
+      projectId: project.id,
+      organizationId: organization.id,
+      apiKey: `tr_dev_${slug}`,
+      pkApiKey: `pk_dev_${slug}`,
+      shortcode: `sc-${slug}`,
+    },
+  });
+
+  return {
+    organizationId: organization.id,
+    projectId: project.id,
+    environmentId: runtimeEnvironment.id,
+    environmentSlug: runtimeEnvironment.slug,
+  };
+}
+
+/** Adds another control-plane environment to an existing project. */
+export async function addEnvironment(
+  prisma: PrismaClient,
+  ctx: SeedContext,
+  slug: string,
+  envSlug: string
+): Promise<string> {
+  const environment = await prisma.runtimeEnvironment.create({
+    data: {
+      slug: envSlug,
+      type: "STAGING",
+      projectId: ctx.projectId,
+      organizationId: ctx.organizationId,
+      apiKey: `tr_${envSlug}_${slug}`,
+      pkApiKey: `pk_${envSlug}_${slug}`,
+      shortcode: `sc-${envSlug}-${slug}`,
+    },
+  });
+
+  return environment.id;
+}
+
+/** Mirrors the parents onto another database with the same IDs. */
+export async function mirrorParents(
+  prisma: PrismaClient,
+  ctx: SeedContext,
+  slug: string
+): Promise<void> {
+  await prisma.organization.create({
+    data: { id: ctx.organizationId, title: `org-${slug}`, slug: `org-${slug}` },
+  });
+  await prisma.project.create({
+    data: {
+      id: ctx.projectId,
+      name: `proj-${slug}`,
+      slug: `proj-${slug}`,
+      organizationId: ctx.organizationId,
+      externalRef: `proj-${slug}`,
+    },
+  });
+  await prisma.runtimeEnvironment.create({
+    data: {
+      id: ctx.environmentId,
+      slug: ctx.environmentSlug,
+      type: "DEVELOPMENT",
+      projectId: ctx.projectId,
+      organizationId: ctx.organizationId,
+      apiKey: `tr_dev_${slug}_b`,
+      pkApiKey: `pk_dev_${slug}_b`,
+      shortcode: `sc-${slug}-b`,
+    },
+  });
+}
+
+export async function createRun(
+  prisma: PrismaClient,
+  ctx: SeedContext,
+  run: {
+    friendlyId: string;
+    taskIdentifier?: string;
+    status?: TaskRunStatus;
+    runtimeEnvironmentId?: string;
+  }
+): Promise<TaskRun> {
+  return prisma.taskRun.create({
+    data: {
+      friendlyId: run.friendlyId,
+      taskIdentifier: run.taskIdentifier ?? "my-task",
+      status: run.status ?? "PENDING",
+      payload: JSON.stringify({ foo: run.friendlyId }),
+      traceId: run.friendlyId,
+      spanId: run.friendlyId,
+      queue: "test",
+      runTags: [],
+      runtimeEnvironmentId: run.runtimeEnvironmentId ?? ctx.environmentId,
+      projectId: ctx.projectId,
+      organizationId: ctx.organizationId,
+      environmentType: "DEVELOPMENT",
+      engine: "V2",
+    },
+  });
+}
+
+/** Inserts the ClickHouse list-index rows synchronously, without logical replication. */
+export async function insertTaskRunV2Rows(clickhouse: ClickHouse, runs: TaskRun[]): Promise<void> {
+  const insert = clickhouse.writer.insert({
+    name: "insertApiRunListPresenterTaskRuns",
+    table: "trigger_dev.task_runs_v2",
+    schema: z.any(),
+    settings: { async_insert: 0, enable_json_type: 1, type_json_skip_duplicated_paths: 1 },
+  });
+
+  const rows: TaskRunV2[] = runs.map((run) => ({
+    environment_id: run.runtimeEnvironmentId,
+    organization_id: run.organizationId ?? "",
+    project_id: run.projectId,
+    run_id: run.id,
+    friendly_id: run.friendlyId,
+    updated_at: run.updatedAt.getTime(),
+    created_at: run.createdAt.getTime(),
+    status: run.status,
+    environment_type: run.environmentType ?? "DEVELOPMENT",
+    attempt: run.attemptNumber ?? 1,
+    engine: run.engine,
+    task_identifier: run.taskIdentifier,
+    queue: run.queue,
+    schedule_id: "",
+    batch_id: "",
+    task_version: run.taskVersion ?? "",
+    sdk_version: run.sdkVersion ?? "",
+    cli_version: run.cliVersion ?? "",
+    machine_preset: run.machinePreset ?? "",
+    root_run_id: "",
+    parent_run_id: "",
+    span_id: run.spanId,
+    trace_id: run.traceId,
+    idempotency_key: run.idempotencyKey ?? "",
+    expiration_ttl: run.ttl ?? "",
+    tags: run.runTags,
+    worker_queue: run.workerQueue,
+    region: run.region ?? "",
+    _version: String(run.updatedAt.getTime()),
+    _is_deleted: 0,
+  }));
+
+  const [error] = await insert(rows);
+  if (error) {
+    throw error;
+  }
+}