v3 task trigger works with clickhouse

Author: ericallam

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

@@ -7,6 +7,7 @@ import { ClickhouseEventRepository } from "~/v3/eventRepository/clickhouseEventR
 import { IEventRepository } from "~/v3/eventRepository/eventRepository.types";
 import { FEATURE_FLAG, flags } from "~/v3/featureFlags.server";
 import { env } from "~/env.server";
+import { getEventRepository } from "~/v3/eventRepository/index.server";
 
 export class DefaultTraceEventsConcern implements TraceEventConcern {
   private readonly eventRepository: EventRepository;
@@ -23,17 +24,9 @@ export class DefaultTraceEventsConcern implements TraceEventConcern {
   async #getEventRepository(
     request: TriggerTaskRequest
   ): Promise<{ repository: IEventRepository; store: string }> {
-    const taskEventRepository = await flags({
-      key: FEATURE_FLAG.taskEventRepository,
-      defaultValue: env.EVENT_REPOSITORY_DEFAULT_STORE,
-      overrides: request.environment.organization.featureFlags as Record<string, unknown>,
-    });
-
-    if (taskEventRepository === "clickhouse") {
-      return { repository: this.clickhouseEventRepository, store: "clickhouse" };
-    }
-
-    return { repository: this.eventRepository, store: getTaskEventStore() };
+    return await getEventRepository(
+      request.environment.organization.featureFlags as Record<string, unknown>
+    );
   }
 
   async traceRun<T>(

apps/webapp/app/v3/eventRepository/index.server.ts

@@ -4,6 +4,8 @@ import { clickhouseEventRepository } from "./clickhouseEventRepositoryInstance.s
 import { IEventRepository, TraceEventOptions } from "./eventRepository.types";
 import { $replica } from "~/db.server";
 import { logger } from "~/services/logger.server";
+import { FEATURE_FLAG, flags } from "../featureFlags.server";
+import { getTaskEventStore } from "../taskEventStore.server";
 
 export function resolveEventRepositoryForStore(store: string | undefined): IEventRepository {
   const taskEventStore = store ?? env.EVENT_REPOSITORY_DEFAULT_STORE;
@@ -15,6 +17,22 @@ export function resolveEventRepositoryForStore(store: string | undefined): IEven
   return eventRepository;
 }
 
+export async function getEventRepository(
+  featureFlags: Record<string, unknown> | undefined
+): Promise<{ repository: IEventRepository; store: string }> {
+  const taskEventRepository = await flags({
+    key: FEATURE_FLAG.taskEventRepository,
+    defaultValue: env.EVENT_REPOSITORY_DEFAULT_STORE,
+    overrides: featureFlags,
+  });
+
+  if (taskEventRepository === "clickhouse") {
+    return { repository: clickhouseEventRepository, store: "clickhouse" };
+  }
+
+  return { repository: eventRepository, store: getTaskEventStore() };
+}
+
 export async function recordRunDebugLog(
   runId: string,
   message: string,

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

@@ -1,9 +1,8 @@
 import {
   IOPacket,
   packetRequiresOffloading,
-  SemanticInternalAttributes,
-  taskRunErrorToString,
   taskRunErrorEnhancer,
+  taskRunErrorToString,
   TriggerTaskRequestBody,
 } from "@trigger.dev/core/v3";
 import {
@@ -12,6 +11,7 @@ import {
   stringifyDuration,
 } from "@trigger.dev/core/v3/isomorphic";
 import { Prisma } from "@trigger.dev/database";
+import { z } from "zod";
 import { env } from "~/env.server";
 import { createTag, MAX_TAGS_PER_RUN } from "~/models/taskRunTag.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
@@ -22,7 +22,7 @@ import { parseDelay } from "~/utils/delays";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
 import { handleMetadataPacket } from "~/utils/packets";
 import { marqs } from "~/v3/marqs/index.server";
-import { eventRepository } from "../eventRepository/eventRepository.server";
+import { getEventRepository } from "../eventRepository/index.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
 import { findCurrentWorkerFromEnvironment } from "../models/workerDeployment.server";
 import { guardQueueSizeLimitsForEnv } from "../queueSizeLimits.server";
@@ -33,16 +33,14 @@ import { startActiveSpan } from "../tracer.server";
 import { clampMaxDuration } from "../utils/maxDuration";
 import { BaseService, ServiceValidationError } from "./baseService.server";
 import { EnqueueDelayedRunService } from "./enqueueDelayedRun.server";
+import { enqueueRun } from "./enqueueRun.server";
 import { ExpireEnqueuedRunService } from "./expireEnqueuedRun.server";
 import {
   MAX_ATTEMPTS,
   OutOfEntitlementError,
   TriggerTaskServiceOptions,
   TriggerTaskServiceResult,
 } from "./triggerTask.server";
-import { getTaskEventStore } from "../taskEventStore.server";
-import { enqueueRun } from "./enqueueRun.server";
-import { z } from "zod";
 
 // This is here for backwords compatibility for v3 users
 const QueueOptions = z.object({
@@ -290,8 +288,12 @@ export class TriggerTaskServiceV1 extends BaseService {
           })
         : undefined;
 
+      const { repository, store } = await getEventRepository(
+        environment.organization.featureFlags as Record<string, unknown>
+      );
+
       try {
-        const result = await eventRepository.traceEvent(
+        const result = await repository.traceEvent(
           taskId,
           {
             context: options.traceContext,
@@ -398,7 +400,7 @@ export class TriggerTaskServiceV1 extends BaseService {
                     queuedAt: delayUntil ? undefined : new Date(),
                     queueTimestamp,
                     maxAttempts: body.options?.maxAttempts,
-                    taskEventStore: getTaskEventStore(),
+                    taskEventStore: store,
                     ttl,
                     tags:
                       tagIds.length === 0