move over the remaining v3 places that used the postgresql event repository

Author: ericallam

apps/webapp/app/v3/services/cancelAttempt.server.ts

@@ -1,11 +1,9 @@
 import { $transaction, type PrismaClientOrTransaction, prisma } from "~/db.server";
 import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
-import { eventRepository } from "../eventRepository/eventRepository.server";
 import { isCancellableRunStatus } from "../taskStatus";
 import { BaseService } from "./baseService.server";
 import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
-import { getTaskEventStoreTableForRun } from "../taskEventStore.server";
 
 export class CancelAttemptService extends BaseService {
   public async call(

apps/webapp/app/v3/services/cancelTaskRun.server.ts

@@ -1,8 +1,5 @@
 import { RunEngineVersion, type TaskRun } from "@trigger.dev/database";
-import { logger } from "~/services/logger.server";
-import { eventRepository } from "../eventRepository/eventRepository.server";
 import { engine } from "../runEngine.server";
-import { getTaskEventStoreTableForRun } from "../taskEventStore.server";
 import { BaseService } from "./baseService.server";
 import { CancelTaskRunServiceV1 } from "./cancelTaskRunV1.server";
 

apps/webapp/app/v3/services/cancelTaskRunV1.server.ts

@@ -1,7 +1,6 @@
 import { type Prisma } from "@trigger.dev/database";
 import assertNever from "assert-never";
 import { logger } from "~/services/logger.server";
-import { eventRepository } from "../eventRepository/eventRepository.server";
 import { socketIo } from "../handleSocketIo.server";
 import { devPubSub } from "../marqs/devPubSub.server";
 import { CANCELLABLE_ATTEMPT_STATUSES, isCancellableRunStatus } from "../taskStatus";
@@ -11,6 +10,7 @@ import { CancelTaskAttemptDependenciesService } from "./cancelTaskAttemptDepende
 import { CancelableTaskRun } from "./cancelTaskRun.server";
 import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
 import { tryCatch } from "@trigger.dev/core/utils";
+import { resolveEventRepositoryForStore } from "../eventRepository/index.server";
 
 type ExtendedTaskRun = Prisma.TaskRunGetPayload<{
   include: {
@@ -101,6 +101,8 @@ export class CancelTaskRunServiceV1 extends BaseService {
       },
     });
 
+    const eventRepository = resolveEventRepositoryForStore(cancelledTaskRun.taskEventStore);
+
     const [cancelRunEventError] = await tryCatch(
       eventRepository.cancelRunEvent({
         reason: opts.reason,

apps/webapp/app/v3/services/completeAttempt.server.ts

@@ -22,7 +22,6 @@ import { env } from "~/env.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { marqs } from "~/v3/marqs/index.server";
-import { eventRepository } from "../eventRepository/eventRepository.server";
 import { FailedTaskRunRetryHelper } from "../failedTaskRun.server";
 import { socketIo } from "../handleSocketIo.server";
 import { createExceptionPropertiesFromError } from "../eventRepository/common.server";
@@ -32,6 +31,7 @@ import { CancelAttemptService } from "./cancelAttempt.server";
 import { CreateCheckpointService } from "./createCheckpoint.server";
 import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
 import { RetryAttemptService } from "./retryAttempt.server";
+import { resolveEventRepositoryForStore } from "../eventRepository/index.server";
 
 type FoundAttempt = Awaited<ReturnType<typeof findAttempt>>;
 
@@ -163,6 +163,8 @@ export class CompleteAttemptService extends BaseService {
       env,
     });
 
+    const eventRepository = resolveEventRepositoryForStore(taskRunAttempt.taskRun.taskEventStore);
+
     const [completeSuccessfulRunEventError] = await tryCatch(
       eventRepository.completeSuccessfulRunEvent({
         run: taskRunAttempt.taskRun,
@@ -314,6 +316,8 @@ export class CompleteAttemptService extends BaseService {
       exitRun(taskRunAttempt.taskRunId);
     }
 
+    const eventRepository = resolveEventRepositoryForStore(taskRunAttempt.taskRun.taskEventStore);
+
     const [completeFailedRunEventError] = await tryCatch(
       eventRepository.completeFailedRunEvent({
         run: taskRunAttempt.taskRun,
@@ -534,6 +538,8 @@ export class CompleteAttemptService extends BaseService {
   }) {
     const retryAt = new Date(executionRetry.timestamp);
 
+    const eventRepository = resolveEventRepositoryForStore(taskRunAttempt.taskRun.taskEventStore);
+
     // Retry the task run
     await eventRepository.recordEvent(
       `Retry #${execution.attempt.number} delay${oomMachine ? " after OOM" : ""}`,

apps/webapp/app/v3/services/crashTaskRun.server.ts

@@ -1,14 +1,13 @@
+import { tryCatch } from "@trigger.dev/core/utils";
+import { sanitizeError, TaskRunErrorCodes, TaskRunInternalError } from "@trigger.dev/core/v3";
 import { TaskRun, TaskRunAttempt } from "@trigger.dev/database";
-import { eventRepository } from "../eventRepository/eventRepository.server";
-import { BaseService } from "./baseService.server";
-import { logger } from "~/services/logger.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+import { FailedTaskRunRetryHelper } from "../failedTaskRun.server";
 import { CRASHABLE_ATTEMPT_STATUSES, isCrashableRunStatus } from "../taskStatus";
-import { sanitizeError, TaskRunErrorCodes, TaskRunInternalError } from "@trigger.dev/core/v3";
+import { BaseService } from "./baseService.server";
 import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
-import { FailedTaskRunRetryHelper } from "../failedTaskRun.server";
-import { getTaskEventStoreTableForRun } from "../taskEventStore.server";
-import { tryCatch } from "@trigger.dev/core/utils";
+import { resolveEventRepositoryForStore } from "../eventRepository/index.server";
 
 export type CrashTaskRunServiceOptions = {
   reason?: string;
@@ -121,6 +120,8 @@ export class CrashTaskRunService extends BaseService {
       },
     });
 
+    const eventRepository = resolveEventRepositoryForStore(crashedTaskRun.taskEventStore);
+
     const [createAttemptFailedEventError] = await tryCatch(
       eventRepository.completeFailedRunEvent({
         run: crashedTaskRun,

apps/webapp/app/v3/services/expireEnqueuedRun.server.ts

@@ -1,11 +1,10 @@
 import { PrismaClientOrTransaction } from "~/db.server";
 import { logger } from "~/services/logger.server";
 import { commonWorker } from "../commonWorker.server";
-import { eventRepository } from "../eventRepository/eventRepository.server";
 import { BaseService } from "./baseService.server";
 import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
-import { getTaskEventStoreTableForRun } from "../taskEventStore.server";
 import { tryCatch } from "@trigger.dev/core/utils";
+import { resolveEventRepositoryForStore } from "../eventRepository/index.server";
 
 export class ExpireEnqueuedRunService extends BaseService {
   public static async ack(runId: string, tx?: PrismaClientOrTransaction) {
@@ -79,6 +78,8 @@ export class ExpireEnqueuedRunService extends BaseService {
       },
     });
 
+    const eventRepository = resolveEventRepositoryForStore(run.taskEventStore);
+
     if (run.ttl) {
       const [completeExpiredRunEventError] = await tryCatch(
         eventRepository.completeExpiredRunEvent({