refactor(webapp): route service-layer TaskRun reads through the run store

Relocate the direct TaskRun reads in webapp services, run-engine concerns,
realtime, mollifier and metadata to the RunStore read methods, preserving
the exact client (writer, replica, or transaction) at each site. The run
hydrator now receives the store by injection. Behavior-preserving.

Author: d-cs

apps/webapp/app/models/runtimeEnvironment.server.ts

@@ -1,6 +1,7 @@
 import type { AuthenticatedEnvironment } from "@internal/run-engine";
 import type { Prisma, PrismaClientOrTransaction, RuntimeEnvironment } from "@trigger.dev/database";
 import { $replica, prisma } from "~/db.server";
+import { runStore } from "~/v3/runStore.server";
 import { logger } from "~/services/logger.server";
 import { getUsername } from "~/utils/username";
 import { sanitizeBranchName } from "@trigger.dev/core/v3/utils/gitBranch";
@@ -251,14 +252,17 @@ export async function findEnvironmentFromRun(
 ): Promise<EnvironmentFromRun | null> {
   // The include (no select) already pulls every taskRun scalar, so runTags/batchId
   // ride along for free — no extra query for the realtime publish to send a full record.
-  const taskRun = await (tx ?? $replica).taskRun.findFirst({
-    where: {
+  const taskRun = await runStore.findRun(
+    {
       id: runId,
     },
-    include: {
-      runtimeEnvironment: { include: authIncludeBase },
+    {
+      include: {
+        runtimeEnvironment: { include: authIncludeBase },
+      },
     },
-  });
+    tx ?? $replica
+  );
   if (!taskRun?.runtimeEnvironment) {
     return null;
   }

apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts

@@ -151,16 +151,19 @@ export class IdempotencyKeyConcern {
     }
 
     const existingRun = idempotencyKey
-      ? await this.prisma.taskRun.findFirst({
-          where: {
+      ? await runStore.findRun(
+          {
             runtimeEnvironmentId: request.environment.id,
             idempotencyKey,
             taskIdentifier: request.taskId,
           },
-          include: {
-            associatedWaitpoint: true,
+          {
+            include: {
+              associatedWaitpoint: true,
+            },
           },
-        })
+          this.prisma
+        )
       : undefined;
 
     // Buffer fallback per the mollifier-idempotency design. PG missed —
@@ -329,14 +332,15 @@ export class IdempotencyKeyConcern {
         // Another concurrent trigger committed first. Re-resolve via the
         // existing checks: writer-side PG findFirst first (defeats
         // replica lag), then buffer fallback for the buffered case.
-        const writerRun = await this.prisma.taskRun.findFirst({
-          where: {
+        const writerRun = await runStore.findRun(
+          {
             runtimeEnvironmentId: request.environment.id,
             idempotencyKey,
             taskIdentifier: request.taskId,
           },
-          include: { associatedWaitpoint: true },
-        });
+          { include: { associatedWaitpoint: true } },
+          this.prisma
+        );
         if (writerRun) {
           return { isCached: true, run: writerRun };
         }

apps/webapp/app/runEngine/services/triggerFailedTask.server.ts

@@ -9,6 +9,7 @@ import { getEventRepository } from "~/v3/eventRepository/index.server";
 import { PerformTaskRunAlertsService } from "~/v3/services/alerts/performTaskRunAlerts.server";
 import { DefaultQueueManager } from "../concerns/queues.server";
 import type { TriggerTaskRequest } from "../types";
+import { runStore } from "~/v3/runStore.server";
 
 export type TriggerFailedTaskRequest = {
   /** The task identifier (e.g. "my-task") */
@@ -82,12 +83,13 @@ export class TriggerFailedTaskService {
 
       // Resolve parent run for rootTaskRunId and depth (same as triggerTask.server.ts)
       const parentRun = request.parentRunId
-        ? await this.prisma.taskRun.findFirst({
-            where: {
+        ? await runStore.findRun(
+            {
               id: RunId.fromFriendlyId(request.parentRunId),
               runtimeEnvironmentId: request.environment.id,
             },
-          })
+            this.prisma
+          )
         : undefined;
 
       const depth = parentRun ? parentRun.depth + 1 : 0;
@@ -275,12 +277,13 @@ export class TriggerFailedTaskService {
       let depth = 0;
 
       if (opts.parentRunId) {
-        const parentRun = await this.prisma.taskRun.findFirst({
-          where: {
+        const parentRun = await runStore.findRun(
+          {
             id: RunId.fromFriendlyId(opts.parentRunId),
             runtimeEnvironmentId: opts.environmentId,
           },
-        });
+          this.prisma
+        );
 
         if (parentRun) {
           parentTaskRunId = parentRun.id;

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -67,6 +67,7 @@ import {
 import { mollifyTrigger } from "~/v3/mollifier/mollifierMollify.server";
 import { type MollifierBuffer } from "@trigger.dev/redis-worker";
 import { QueueSizeLimitExceededError, ServiceValidationError } from "~/v3/services/common.server";
+import { runStore } from "~/v3/runStore.server";
 
 class NoopTriggerRacepointSystem implements TriggerRacepointSystem {
   async waitForRacepoint(options: { racepoint: TriggerRacepoints; id: string }): Promise<void> {
@@ -241,12 +242,13 @@ export class RunEngineTriggerTaskService {
 
           // Get parent run if specified
           const parentRun = body.options?.parentRunId
-            ? await this.prisma.taskRun.findFirst({
-              where: {
+            ? await runStore.findRun(
+              {
                 id: RunId.fromFriendlyId(body.options.parentRunId),
                 runtimeEnvironmentId: environment.id,
               },
-            })
+              this.prisma
+            )
             : undefined;
 
           // Validate parent run

apps/webapp/app/services/metadata/updateMetadata.server.ts

@@ -189,18 +189,21 @@ export class UpdateMetadataService {
       // Fetch current run (+ the realtime membership keys, so a flush can publish)
       const run = yield* _(
         Effect.tryPromise(() =>
-          this._prisma.taskRun.findFirst({
-            where: { id: runId },
-            select: {
-              id: true,
-              metadata: true,
-              metadataType: true,
-              metadataVersion: true,
-              runtimeEnvironmentId: true,
-              runTags: true,
-              batchId: true,
+          this._runStore.findRun(
+            { id: runId },
+            {
+              select: {
+                id: true,
+                metadata: true,
+                metadataType: true,
+                metadataVersion: true,
+                runtimeEnvironmentId: true,
+                runTags: true,
+                batchId: true,
+              },
             },
-          })
+            this._prisma
+          )
         )
       );
 
@@ -332,38 +335,41 @@ export class UpdateMetadataService {
   ) {
     const runIdType = runId.startsWith("run_") ? "friendly" : "internal";
 
-    const taskRun = await this._prisma.taskRun.findFirst({
-      where: environment
+    const taskRun = await this._runStore.findRun(
+      environment
         ? {
             runtimeEnvironmentId: environment.id,
             ...(runIdType === "internal" ? { id: runId } : { friendlyId: runId }),
           }
         : {
             ...(runIdType === "internal" ? { id: runId } : { friendlyId: runId }),
           },
-      select: {
-        id: true,
-        batchId: true,
-        runTags: true,
-        completedAt: true,
-        status: true,
-        metadata: true,
-        metadataType: true,
-        metadataVersion: true,
-        parentTaskRun: {
-          select: {
-            id: true,
-            status: true,
+      {
+        select: {
+          id: true,
+          batchId: true,
+          runTags: true,
+          completedAt: true,
+          status: true,
+          metadata: true,
+          metadataType: true,
+          metadataVersion: true,
+          parentTaskRun: {
+            select: {
+              id: true,
+              status: true,
+            },
           },
-        },
-        rootTaskRun: {
-          select: {
-            id: true,
-            status: true,
+          rootTaskRun: {
+            select: {
+              id: true,
+              status: true,
+            },
           },
         },
       },
-    });
+      this._prisma
+    );
 
     if (!taskRun) {
       return;
@@ -427,10 +433,13 @@ export class UpdateMetadataService {
 
     while (attempts <= MAX_RETRIES) {
       // Fetch the latest run data
-      const run = await this._prisma.taskRun.findFirst({
-        where: { id: runId },
-        select: { metadata: true, metadataType: true, metadataVersion: true },
-      });
+      const run = await this._runStore.findRun(
+        { id: runId },
+        {
+          select: { metadata: true, metadataType: true, metadataVersion: true },
+        },
+        this._prisma
+      );
 
       if (!run) {
         throw new Error(`Run ${runId} not found`);

apps/webapp/app/services/realtime/nativeRealtimeClientInstance.server.ts

@@ -1,5 +1,6 @@
 import { getMeter } from "@internal/tracing";
 import { $replica } from "~/db.server";
+import { runStore } from "~/v3/runStore.server";
 import { env } from "~/env.server";
 import { singleton } from "~/utils/singleton";
 import { getCachedLimit } from "../platform.v3.server";
@@ -122,6 +123,7 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient {
   // One RunHydrator shared by the router and the client, so its single-flight + short-TTL cache covers both.
   const runReader = new RunHydrator({
     replica: $replica,
+    runStore,
     cacheTtlMs: env.REALTIME_BACKEND_NATIVE_RUN_CACHE_TTL_MS,
     maxCacheEntries: env.REALTIME_BACKEND_NATIVE_RUN_CACHE_MAX_ENTRIES,
   });

apps/webapp/app/services/realtime/runReader.server.ts

@@ -1,4 +1,5 @@
-import { type Prisma, type PrismaClient } from "@trigger.dev/database";
+import { type Prisma, type PrismaClient, type PrismaClientOrTransaction } from "@trigger.dev/database";
+import type { RunStore } from "@internal/run-store";
 import { BoundedTtlCache } from "./boundedTtlCache";
 import { RESERVED_COLUMNS, type RealtimeRunRow } from "./electricStreamProtocol.server";
 
@@ -79,6 +80,8 @@ export interface RunListResolver {
 export type RunHydratorOptions = {
   /** A read-replica Prisma client (`$replica`). Always Postgres. */
   replica: Pick<PrismaClient, "taskRun">;
+  /** RunStore the reads are routed through; `replica` is passed as the read client. */
+  runStore: RunStore;
   /** Read-through cache TTL (ms) collapsing duplicate refetches for the same run. Set 0 to disable. Defaults to 250ms. */
   cacheTtlMs?: number;
   /** Hard cap on cache entries before expired entries are swept. */
@@ -139,24 +142,28 @@ export class RunHydrator {
     if (ids.length === 0) {
       return [];
     }
-    const rows = await this.options.replica.taskRun.findMany({
-      where: {
-        runtimeEnvironmentId: environmentId,
-        id: { in: ids },
+    const rows = await this.options.runStore.findRuns(
+      {
+        where: {
+          runtimeEnvironmentId: environmentId,
+          id: { in: ids },
+        },
+        select: buildHydratorSelect(skipColumns),
       },
-      select: buildHydratorSelect(skipColumns),
-    });
+      this.options.replica as PrismaClientOrTransaction
+    );
     return rows as unknown as RealtimeRunRow[];
   }
 
   async #fetch(environmentId: string, runId: string): Promise<RealtimeRunRow | null> {
-    const run = await this.options.replica.taskRun.findFirst({
-      where: {
+    const run = await this.options.runStore.findRun(
+      {
         id: runId,
         runtimeEnvironmentId: environmentId,
       },
-      select: RUN_HYDRATOR_SELECT,
-    });
+      { select: RUN_HYDRATOR_SELECT },
+      this.options.replica as PrismaClientOrTransaction
+    );
 
     return (run ?? null) as RealtimeRunRow | null;
   }