Support for org-scoped ClickHouse (#3333)

Added `OrganizationDataStore` which allows orgs to have data stored in
specific separate services.

For now this is just used for ClickHouse. When using ClickHouse we get a
client for the factory and pass in the org id.

Particular care has to be made with two hot-insert paths:
1. RunReplicationService
2. OTLPExporter

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Claude <noreply@anthropic.com>

Author: 3 people

.cursor/mcp.json

@@ -1,3 +1,7 @@
 {
-  "mcpServers": {}
+  "mcpServers": {
+    "linear": {
+      "url": "https://mcp.linear.app/mcp"
+    }
+  }
 }

.server-changes/organization-scoped-clickhouse.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+Organization-scoped ClickHouse routing enables customers with HIPAA and other data security requirements to use dedicated database instances

CLAUDE.md

@@ -69,6 +69,17 @@ containerTest("should use both", async ({ prisma, redisOptions }) => {
 });
 ```
 
+## Code Style
+
+### Imports
+
+**Prefer static imports over dynamic imports.** Only use dynamic `import()` when:
+- Circular dependencies cannot be resolved otherwise
+- Code splitting is genuinely needed for performance
+- The module must be loaded conditionally at runtime
+
+Dynamic imports add unnecessary overhead in hot paths and make code harder to analyze. If you find yourself using `await import()`, ask if a regular `import` statement would work instead.
+
 ## Changesets and Server Changes
 
 When modifying any public package (`packages/*` or `integrations/*`), add a changeset:

apps/webapp/app/entry.server.tsx

@@ -24,44 +24,12 @@ import {
   registerRunEngineEventBusHandlers,
   setupBatchQueueCallbacks,
 } from "./v3/runEngineHandlers.server";
+// Touch the sessions replication singleton at entry so it boots deterministically
+// on webapp startup. The singleton's initializer wires start (gated on
+// `clickhouseFactory.isReady()`) and SIGTERM/SIGINT shutdown — mirrors
+// runsReplicationInstance.
 import { sessionsReplicationInstance } from "./services/sessionsReplicationInstance.server";
-import { signalsEmitter } from "./services/signals.server";
-
-// Start the sessions replication service (subscribes to the logical replication
-// slot, runs leader election, flushes to ClickHouse). Done at entry level so it
-// runs deterministically on webapp boot rather than lazily via a singleton
-// reference elsewhere in the module graph.
-if (sessionsReplicationInstance && env.SESSION_REPLICATION_ENABLED === "1") {
-  // Capture a non-nullable reference so the shutdown closure below
-  // doesn't need to re-null-check (TS narrowing doesn't follow through
-  // an inner function scope).
-  const replicator = sessionsReplicationInstance;
-  replicator
-    .start()
-    .then(() => {
-      console.log("🗃️ Sessions replication service started");
-    })
-    .catch((error) => {
-      console.error("🗃️ Sessions replication service failed to start", {
-        error,
-      });
-    });
-
-  // Wrap the async shutdown in a sync handler that catches rejections —
-  // SIGTERM/SIGINT fire during process teardown, and an unhandled
-  // promise rejection from `_replicationClient.stop()` there would
-  // bubble up past the process exit. Matches the pattern in
-  // dynamicFlushScheduler.server.ts.
-  const shutdownSessionsReplication = () => {
-    replicator.shutdown().catch((error) => {
-      console.error("🗃️ Sessions replication service shutdown error", {
-        error,
-      });
-    });
-  };
-  signalsEmitter.on("SIGTERM", shutdownSessionsReplication);
-  signalsEmitter.on("SIGINT", shutdownSessionsReplication);
-}
+void sessionsReplicationInstance;
 
 const ABORT_DELAY = 30000;
 

apps/webapp/app/env.server.ts

@@ -459,7 +459,10 @@ const EnvironmentSchema = z
     // If specified, you must configure the corresponding provider using OBJECT_STORE_{PROTOCOL}_* env vars.
     // Example: OBJECT_STORE_DEFAULT_PROTOCOL=s3 requires OBJECT_STORE_S3_BASE_URL, OBJECT_STORE_S3_ACCESS_KEY_ID, etc.
     // Enables zero-downtime migration between providers (old data keeps working, new data uses new provider).
-    OBJECT_STORE_DEFAULT_PROTOCOL: z.string().regex(/^[a-z0-9]+$/).optional(),
+    OBJECT_STORE_DEFAULT_PROTOCOL: z
+      .string()
+      .regex(/^[a-z0-9]+$/)
+      .optional(),
 
     ARTIFACTS_OBJECT_STORE_BUCKET: z.string().optional(),
     ARTIFACTS_OBJECT_STORE_BASE_URL: z.string().optional(),
@@ -1489,9 +1492,18 @@ const EnvironmentSchema = z
     EVENTS_CLICKHOUSE_MAX_TRACE_DETAILED_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(5_000),
     EVENTS_CLICKHOUSE_MAX_LIVE_RELOADING_SETTING: z.coerce.number().int().default(2000),
 
+    // Organization data stores registry
+    ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS: z.coerce
+      .number()
+      .int()
+      .default(60 * 1000), // 1 minute
+
     // LLM cost tracking
     LLM_COST_TRACKING_ENABLED: BoolEnv.default(true),
-    LLM_PRICING_RELOAD_INTERVAL_MS: z.coerce.number().int().default(5 * 60 * 1000), // 5 minutes
+    LLM_PRICING_RELOAD_INTERVAL_MS: z.coerce
+      .number()
+      .int()
+      .default(5 * 60 * 1000), // 5 minutes
     LLM_PRICING_RELOAD_CHANNEL: z.string().default("llm-registry:reload"),
     LLM_PRICING_RELOAD_DEBOUNCE_MS: z.coerce.number().int().default(1000),
     // Whether to subscribe this process to the LLM_PRICING_RELOAD_CHANNEL.

apps/webapp/app/presenters/v3/AgentListPresenter.server.ts

@@ -3,10 +3,10 @@ import {
   type RuntimeEnvironmentType,
   type TaskTriggerSource,
 } from "@trigger.dev/database";
-import { ClickHouse } from "@internal/clickhouse";
+import { type ClickHouse } from "@internal/clickhouse";
 import { z } from "zod";
 import { $replica } from "~/db.server";
-import { clickhouseClient } from "~/services/clickhouseInstance.server";
+import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { singleton } from "~/utils/singleton";
 import { findCurrentWorkerFromEnvironment } from "~/v3/models/workerDeployment.server";
 
@@ -24,10 +24,7 @@ export type AgentActiveState = {
 };
 
 export class AgentListPresenter {
-  constructor(
-    private readonly clickhouse: ClickHouse,
-    private readonly _replica: PrismaClientOrTransaction
-  ) {}
+  constructor(private readonly _replica: PrismaClientOrTransaction) {}
 
   public async call({
     organizationId,
@@ -40,6 +37,11 @@ export class AgentListPresenter {
     environmentId: string;
     environmentType: RuntimeEnvironmentType;
   }) {
+    const clickhouse = await clickhouseFactory.getClickhouseForOrganization(
+      organizationId,
+      "standard"
+    );
+
     const currentWorker = await findCurrentWorkerFromEnvironment(
       {
         id: environmentId,
@@ -89,20 +91,21 @@ export class AgentListPresenter {
     }
 
     // All queries are deferred for streaming
-    const activeStates = this.#getActiveStates(environmentId, slugs);
-    const conversationSparklines = this.#getConversationSparklines(environmentId, slugs);
-    const costSparklines = this.#getCostSparklines(environmentId, slugs);
-    const tokenSparklines = this.#getTokenSparklines(environmentId, slugs);
+    const activeStates = this.#getActiveStates(clickhouse, environmentId, slugs);
+    const conversationSparklines = this.#getConversationSparklines(clickhouse, environmentId, slugs);
+    const costSparklines = this.#getCostSparklines(clickhouse, environmentId, slugs);
+    const tokenSparklines = this.#getTokenSparklines(clickhouse, environmentId, slugs);
 
     return { agents, activeStates, conversationSparklines, costSparklines, tokenSparklines };
   }
 
   /** Count runs currently executing vs suspended per agent */
   async #getActiveStates(
+    clickhouse: ClickHouse,
     environmentId: string,
     slugs: string[]
   ): Promise<Record<string, AgentActiveState>> {
-    const queryFn = this.clickhouse.reader.query({
+    const queryFn = clickhouse.reader.query({
       name: "agentActiveStates",
       query: `SELECT
           task_identifier,
@@ -140,10 +143,11 @@ export class AgentListPresenter {
 
   /** 24h hourly sparkline of conversation (run) count per agent */
   async #getConversationSparklines(
+    clickhouse: ClickHouse,
     environmentId: string,
     slugs: string[]
   ): Promise<Record<string, number[]>> {
-    const queryFn = this.clickhouse.reader.query({
+    const queryFn = clickhouse.reader.query({
       name: "agentConversationSparklines",
       query: `SELECT
           task_identifier,
@@ -172,10 +176,11 @@ export class AgentListPresenter {
 
   /** 24h hourly sparkline of LLM cost per agent */
   async #getCostSparklines(
+    clickhouse: ClickHouse,
     environmentId: string,
     slugs: string[]
   ): Promise<Record<string, number[]>> {
-    const queryFn = this.clickhouse.reader.query({
+    const queryFn = clickhouse.reader.query({
       name: "agentCostSparklines",
       query: `SELECT
           task_identifier,
@@ -203,10 +208,11 @@ export class AgentListPresenter {
 
   /** 24h hourly sparkline of total tokens per agent */
   async #getTokenSparklines(
+    clickhouse: ClickHouse,
     environmentId: string,
     slugs: string[]
   ): Promise<Record<string, number[]>> {
-    const queryFn = this.clickhouse.reader.query({
+    const queryFn = clickhouse.reader.query({
       name: "agentTokenSparklines",
       query: `SELECT
           task_identifier,
@@ -284,5 +290,5 @@ export class AgentListPresenter {
 export const agentListPresenter = singleton("agentListPresenter", setupAgentListPresenter);
 
 function setupAgentListPresenter() {
-  return new AgentListPresenter(clickhouseClient, $replica);
+  return new AgentListPresenter($replica);
 }

apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts

@@ -9,7 +9,7 @@ import { type Project, type RuntimeEnvironment, type TaskRunStatus } from "@trig
 import assertNever from "assert-never";
 import { z } from "zod";
 import { API_VERSIONS, RunStatusUnspecifiedApiVersion } from "~/api/versions";
-import { clickhouseClient } from "~/services/clickhouseInstance.server";
+import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { logger } from "~/services/logger.server";
 import { CoercedDate } from "~/utils/zod";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
@@ -269,7 +269,8 @@ export class ApiRunListPresenter extends BasePresenter {
         options.machines = searchParams["filter[machine]"];
       }
 
-      const presenter = new NextRunListPresenter(this._replica, clickhouseClient);
+      const clickhouse = await clickhouseFactory.getClickhouseForOrganization(organizationId, "standard");
+      const presenter = new NextRunListPresenter(this._replica, clickhouse);
 
       logger.debug("Calling RunListPresenter", { options });
 

apps/webapp/app/presenters/v3/CreateBulkActionPresenter.server.ts

@@ -1,6 +1,6 @@
 import { type PrismaClient } from "@trigger.dev/database";
 import { CreateBulkActionSearchParams } from "~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.bulkaction";
-import { clickhouseClient } from "~/services/clickhouseInstance.server";
+import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { RunsRepository } from "~/services/runsRepository/runsRepository.server";
 import { getRunFiltersFromRequest } from "../RunFilters.server";
 import { BasePresenter } from "./basePresenter.server";
@@ -24,8 +24,9 @@ export class CreateBulkActionPresenter extends BasePresenter {
       Object.fromEntries(new URL(request.url).searchParams)
     );
 
+    const clickhouse = await clickhouseFactory.getClickhouseForOrganization(organizationId, "standard");
     const runsRepository = new RunsRepository({
-      clickhouse: clickhouseClient,
+      clickhouse,
       prisma: this._replica as PrismaClient,
     });
 

apps/webapp/app/presenters/v3/RunPresenter.server.ts

@@ -3,11 +3,11 @@ import { createTreeFromFlatItems, flattenTree } from "~/components/primitives/Tr
 import { prisma, type PrismaClient } from "~/db.server";
 import { createTimelineSpanEventsFromSpanEvents } from "~/utils/timelineSpanEvents";
 import { getUsername } from "~/utils/username";
-import { resolveEventRepositoryForStore } from "~/v3/eventRepository/index.server";
 import { SpanSummary } from "~/v3/eventRepository/eventRepository.types";
 import { getTaskEventStoreTableForRun } from "~/v3/taskEventStore.server";
 import { isFinalRunStatus } from "~/v3/taskStatus";
 import { env } from "~/env.server";
+import { getEventRepositoryForStore } from "~/v3/eventRepository/index.server";
 
 type Result = Awaited<ReturnType<RunPresenter["call"]>>;
 export type Run = Result["run"];
@@ -145,10 +145,13 @@ export class RunPresenter {
       };
     }
 
-    const eventRepository = resolveEventRepositoryForStore(run.taskEventStore);
+    const repository = await getEventRepositoryForStore(
+      run.taskEventStore,
+      run.runtimeEnvironment.organizationId
+    );
 
     // get the events
-    let traceSummary = await eventRepository.getTraceSummary(
+    let traceSummary = await repository.getTraceSummary(
       getTaskEventStoreTableForRun(run),
       run.runtimeEnvironment.id,
       run.traceId,
@@ -272,7 +275,7 @@ export class RunPresenter {
         overridesBySpanId: traceSummary.overridesBySpanId,
         linkedRunIdBySpanId,
       },
-      maximumLiveReloadingSetting: eventRepository.maximumLiveReloadingSetting,
+      maximumLiveReloadingSetting: repository.maximumLiveReloadingSetting,
     };
   }
 }

apps/webapp/app/presenters/v3/RunTagListPresenter.server.ts

@@ -1,6 +1,6 @@
 import { RunsRepository } from "~/services/runsRepository/runsRepository.server";
 import { BasePresenter } from "./basePresenter.server";
-import { clickhouseClient } from "~/services/clickhouseInstance.server";
+import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { type PrismaClient } from "@trigger.dev/database";
 import { timeFilters } from "~/components/runs/v3/SharedFilters";
 
@@ -37,8 +37,9 @@ export class RunTagListPresenter extends BasePresenter {
   }: TagListOptions) {
     const hasFilters = Boolean(name?.trim());
 
+    const clickhouse = await clickhouseFactory.getClickhouseForOrganization(organizationId, "standard");
     const runsRepository = new RunsRepository({
-      clickhouse: clickhouseClient,
+      clickhouse,
       prisma: this._replica as PrismaClient,
     });