use feature flags to determine the run taskEvent store

Author: ericallam

.vscode/settings.json

@@ -7,5 +7,5 @@
     "packages/cli-v3/e2e": true
   },
   "vitest.disableWorkspaceWarning": true,
-  "typescript.experimental.useTsgo": true
+  "typescript.experimental.useTsgo": false
 }

apps/webapp/app/env.server.ts

@@ -1109,6 +1109,19 @@ const EnvironmentSchema = z
     CLICKHOUSE_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
     CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),
 
+    EVENTS_CLICKHOUSE_URL: z
+      .string()
+      .optional()
+      .transform((v) => v ?? process.env.CLICKHOUSE_URL),
+    EVENTS_CLICKHOUSE_KEEP_ALIVE_ENABLED: z.string().default("1"),
+    EVENTS_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
+    EVENTS_CLICKHOUSE_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
+    EVENTS_CLICKHOUSE_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
+    EVENTS_CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),
+    EVENTS_CLICKHOUSE_BATCH_SIZE: z.coerce.number().int().default(1000),
+    EVENTS_CLICKHOUSE_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
+    EVENT_REPOSITORY_DEFAULT_STORE: z.enum(["postgres", "clickhouse"]).default("postgres"),
+
     // Bootstrap
     TRIGGER_BOOTSTRAP_ENABLED: z.string().default("0"),
     TRIGGER_BOOTSTRAP_WORKER_GROUP_NAME: z.string().optional(),

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

@@ -1,21 +1,48 @@
 import { EventRepository } from "~/v3/eventRepository.server";
 import { TracedEventSpan, TraceEventConcern, TriggerTaskRequest } from "../types";
 import { SemanticInternalAttributes } from "@trigger.dev/core/v3/semanticInternalAttributes";
-import { BatchId } from "@trigger.dev/core/v3/isomorphic";
 import { TaskRun } from "@trigger.dev/database";
+import { getTaskEventStore } from "~/v3/taskEventStore.server";
+import { ClickhouseEventRepository } from "~/v3/clickhouseEventRepository.server";
+import { IEventRepository } from "~/v3/eventRepository.types";
+import { FEATURE_FLAG, flags } from "~/v3/featureFlags.server";
+import { env } from "~/env.server";
 
 export class DefaultTraceEventsConcern implements TraceEventConcern {
   private readonly eventRepository: EventRepository;
+  private readonly clickhouseEventRepository: ClickhouseEventRepository;
 
-  constructor(eventRepository: EventRepository) {
+  constructor(
+    eventRepository: EventRepository,
+    clickhouseEventRepository: ClickhouseEventRepository
+  ) {
     this.eventRepository = eventRepository;
+    this.clickhouseEventRepository = clickhouseEventRepository;
+  }
+
+  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() };
   }
 
   async traceRun<T>(
     request: TriggerTaskRequest,
-    callback: (span: TracedEventSpan) => Promise<T>
+    callback: (span: TracedEventSpan, store: string) => Promise<T>
   ): Promise<T> {
-    return await this.eventRepository.traceEvent(
+    const { repository, store } = await this.#getEventRepository(request);
+
+    return await repository.traceEvent(
       request.taskId,
       {
         context: request.options?.traceContext,
@@ -38,14 +65,17 @@ export class DefaultTraceEventsConcern implements TraceEventConcern {
           : undefined,
       },
       async (event, traceContext, traceparent) => {
-        return await callback({
-          traceId: event.traceId,
-          spanId: event.spanId,
-          traceContext,
-          traceparent,
-          setAttribute: (key, value) => event.setAttribute(key as any, value),
-          failWithError: event.failWithError.bind(event),
-        });
+        return await callback(
+          {
+            traceId: event.traceId,
+            spanId: event.spanId,
+            traceContext,
+            traceparent,
+            setAttribute: (key, value) => event.setAttribute(key as any, value),
+            failWithError: event.failWithError.bind(event),
+          },
+          store
+        );
       }
     );
   }
@@ -58,11 +88,12 @@ export class DefaultTraceEventsConcern implements TraceEventConcern {
       incomplete: boolean;
       isError: boolean;
     },
-    callback: (span: TracedEventSpan) => Promise<T>
+    callback: (span: TracedEventSpan, store: string) => Promise<T>
   ): Promise<T> {
     const { existingRun, idempotencyKey, incomplete, isError } = options;
+    const { repository, store } = await this.#getEventRepository(request);
 
-    return await this.eventRepository.traceEvent(
+    return await repository.traceEvent(
       `${request.taskId} (cached)`,
       {
         context: request.options?.traceContext,
@@ -99,14 +130,17 @@ export class DefaultTraceEventsConcern implements TraceEventConcern {
           }
         );
 
-        return await callback({
-          traceId: event.traceId,
-          spanId: event.spanId,
-          traceContext,
-          traceparent,
-          setAttribute: (key, value) => event.setAttribute(key as any, value),
-          failWithError: event.failWithError.bind(event),
-        });
+        return await callback(
+          {
+            traceId: event.traceId,
+            spanId: event.spanId,
+            traceContext,
+            traceparent,
+            setAttribute: (key, value) => event.setAttribute(key as any, value),
+            failWithError: event.failWithError.bind(event),
+          },
+          store
+        );
       }
     );
   }

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

@@ -29,7 +29,6 @@ import type {
   TriggerTaskServiceOptions,
   TriggerTaskServiceResult,
 } from "../../v3/services/triggerTask.server";
-import { getTaskEventStore } from "../../v3/taskEventStore.server";
 import { clampMaxDuration } from "../../v3/utils/maxDuration";
 import { IdempotencyKeyConcern } from "../concerns/idempotencyKeys.server";
 import type {
@@ -267,7 +266,7 @@ export class RunEngineTriggerTaskService {
       const workerQueue = await this.queueConcern.getWorkerQueue(environment, body.options?.region);
 
       try {
-        return await this.traceEventConcern.traceRun(triggerRequest, async (event) => {
+        return await this.traceEventConcern.traceRun(triggerRequest, async (event, store) => {
           const result = await this.runNumberIncrementer.incrementRunNumber(
             triggerRequest,
             async (num) => {
@@ -311,7 +310,7 @@ export class RunEngineTriggerTaskService {
                   delayUntil,
                   queuedAt: delayUntil ? undefined : new Date(),
                   maxAttempts: body.options?.maxAttempts,
-                  taskEventStore: getTaskEventStore(),
+                  taskEventStore: store,
                   ttl,
                   tags,
                   oneTimeUseToken: options.oneTimeUseToken,

apps/webapp/app/runEngine/types.ts

@@ -143,7 +143,7 @@ export type TracedEventSpan = {
 export interface TraceEventConcern {
   traceRun<T>(
     request: TriggerTaskRequest,
-    callback: (span: TracedEventSpan) => Promise<T>
+    callback: (span: TracedEventSpan, store: string) => Promise<T>
   ): Promise<T>;
   traceIdempotentRun<T>(
     request: TriggerTaskRequest,
@@ -153,7 +153,7 @@ export interface TraceEventConcern {
       incomplete: boolean;
       isError: boolean;
     },
-    callback: (span: TracedEventSpan) => Promise<T>
+    callback: (span: TracedEventSpan, store: string) => Promise<T>
   ): Promise<T>;
 }
 

apps/webapp/app/v3/clickhouseEventRepository.server.ts

@@ -197,13 +197,3 @@ export class ClickhouseEventRepository implements IEventRepository {
     throw new Error("ClickhouseEventRepository.generateSpanId not implemented");
   }
 }
-
-/**
- * Factory function to create a ClickhouseEventRepository instance.
- * This can be used as an alternative to the PostgreSQL-based EventRepository.
- */
-export function createClickhouseEventRepository(
-  config: ClickhouseEventRepositoryConfig
-): IEventRepository {
-  return new ClickhouseEventRepository(config);
-}

apps/webapp/app/v3/clickhouseEventRepositoryInstance.server.ts

@@ -0,0 +1,45 @@
+import { ClickHouse } from "@internal/clickhouse";
+import { env } from "~/env.server";
+import { singleton } from "~/utils/singleton";
+import { ClickhouseEventRepository } from "./clickhouseEventRepository.server";
+
+export const clickhouseEventRepository = singleton(
+  "clickhouseEventRepository",
+  initializeClickhouseRepository
+);
+
+function initializeClickhouseRepository() {
+  if (!env.EVENTS_CLICKHOUSE_URL) {
+    throw new Error("EVENTS_CLICKHOUSE_URL is not set");
+  }
+
+  const url = new URL(env.EVENTS_CLICKHOUSE_URL);
+  url.searchParams.delete("secure");
+
+  const safeUrl = new URL(url.toString());
+  safeUrl.password = "redacted";
+
+  console.log("🗃️  Initializing Clickhouse event repository", { url: safeUrl.toString() });
+
+  const clickhouse = new ClickHouse({
+    url: url.toString(),
+    name: "task-events",
+    keepAlive: {
+      enabled: env.EVENTS_CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
+      idleSocketTtl: env.EVENTS_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
+    },
+    logLevel: env.EVENTS_CLICKHOUSE_LOG_LEVEL,
+    compression: {
+      request: true,
+    },
+    maxOpenConnections: env.EVENTS_CLICKHOUSE_MAX_OPEN_CONNECTIONS,
+  });
+
+  const repository = new ClickhouseEventRepository({
+    clickhouse: clickhouse,
+    batchSize: env.EVENTS_CLICKHOUSE_BATCH_SIZE,
+    flushInterval: env.EVENTS_CLICKHOUSE_FLUSH_INTERVAL_MS,
+  });
+
+  return repository;
+}

apps/webapp/app/v3/featureFlags.server.ts

@@ -4,18 +4,21 @@ import { prisma, type PrismaClientOrTransaction } from "~/db.server";
 export const FEATURE_FLAG = {
   defaultWorkerInstanceGroupId: "defaultWorkerInstanceGroupId",
   runsListRepository: "runsListRepository",
+  taskEventRepository: "taskEventRepository",
 } as const;
 
 const FeatureFlagCatalog = {
   [FEATURE_FLAG.defaultWorkerInstanceGroupId]: z.string(),
   [FEATURE_FLAG.runsListRepository]: z.enum(["clickhouse", "postgres"]),
+  [FEATURE_FLAG.taskEventRepository]: z.enum(["clickhouse", "postgres"]),
 };
 
 type FeatureFlagKey = keyof typeof FeatureFlagCatalog;
 
 export type FlagsOptions<T extends FeatureFlagKey> = {
   key: T;
   defaultValue?: z.infer<(typeof FeatureFlagCatalog)[T]>;
+  overrides?: Record<string, unknown>;
 };
 
 export function makeFlags(_prisma: PrismaClientOrTransaction = prisma) {
@@ -34,7 +37,17 @@ export function makeFlags(_prisma: PrismaClientOrTransaction = prisma) {
       },
     });
 
-    const parsed = FeatureFlagCatalog[opts.key].safeParse(value?.value);
+    const flagSchema = FeatureFlagCatalog[opts.key];
+
+    if (opts.overrides?.[opts.key]) {
+      const parsed = flagSchema.safeParse(opts.overrides[opts.key]);
+
+      if (parsed.success) {
+        return parsed.data;
+      }
+    }
+
+    const parsed = flagSchema.safeParse(value?.value);
 
     if (!parsed.success) {
       return opts.defaultValue;

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

@@ -14,6 +14,7 @@ import { tracer } from "../tracer.server";
 import { WithRunEngine } from "./baseService.server";
 import { TriggerTaskServiceV1 } from "./triggerTaskV1.server";
 import { env } from "~/env.server";
+import { clickhouseEventRepository } from "../clickhouseEventRepositoryInstance.server";
 
 export type TriggerTaskServiceOptions = {
   idempotencyKey?: string;
@@ -93,7 +94,10 @@ export class TriggerTaskService extends WithRunEngine {
     body: TriggerTaskRequestBody,
     options: TriggerTaskServiceOptions = {}
   ): Promise<TriggerTaskServiceResult | undefined> {
-    const traceEventConcern = new DefaultTraceEventsConcern(eventRepository);
+    const traceEventConcern = new DefaultTraceEventsConcern(
+      eventRepository,
+      clickhouseEventRepository
+    );
 
     const service = new RunEngineTriggerTaskService({
       prisma: this._prisma,

internal-packages/database/prisma/migrations/20250922145116_add_feature_flags_to_organizations/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "public"."Organization" ADD COLUMN     "featureFlags" JSONB;