Implement organization-scoped ClickHouse instances

The only way to get a ClickHouse client now is through the factory.

Refactored all existing code to use that and pass in an org.

The runReplication and otlpExporter are the hot paths here which need special attention in reviews.

Author: matt-aitken

.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

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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
 import { logger } from "~/services/logger.server";
 import { CoercedDate } from "~/utils/zod";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
@@ -259,7 +259,8 @@ export class ApiRunListPresenter extends BasePresenter {
         options.machines = searchParams["filter[machine]"];
       }
 
-      const presenter = new NextRunListPresenter(this._replica, clickhouseClient);
+      const clickhouse = await 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.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 getClickhouseForOrganization(organizationId, "standard");
     const runsRepository = new RunsRepository({
-      clickhouse: clickhouseClient,
+      clickhouse,
       prisma: this._replica as PrismaClient,
     });
 

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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.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 getClickhouseForOrganization(organizationId, "standard");
     const runsRepository = new RunsRepository({
-      clickhouse: clickhouseClient,
+      clickhouse,
       prisma: this._replica as PrismaClient,
     });
 

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

@@ -4,7 +4,7 @@ import {
   type TaskTriggerSource,
 } from "@trigger.dev/database";
 import { $replica } from "~/db.server";
-import { clickhouseClient } from "~/services/clickhouseInstance.server";
+import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
 import {
   type AverageDurations,
   ClickHouseEnvironmentMetricsRepository,
@@ -25,10 +25,7 @@ export type TaskListItem = {
 export type TaskActivity = DailyTaskActivity[string];
 
 export class TaskListPresenter {
-  constructor(
-    private readonly environmentMetricsRepository: EnvironmentMetricsRepository,
-    private readonly _replica: PrismaClientOrTransaction
-  ) {}
+  constructor(private readonly _replica: PrismaClientOrTransaction) {}
 
   public async call({
     organizationId,
@@ -76,25 +73,31 @@ export class TaskListPresenter {
 
     const slugs = tasks.map((t) => t.slug);
 
+    // Create org-specific environment metrics repository
+    const clickhouse = await getClickhouseForOrganization(organizationId, "standard");
+    const environmentMetricsRepository = new ClickHouseEnvironmentMetricsRepository({
+      clickhouse,
+    });
+
     // IMPORTANT: Don't await these, we want to return the promises
     // so we can defer the loading of the data
-    const activity = this.environmentMetricsRepository.getDailyTaskActivity({
+    const activity = environmentMetricsRepository.getDailyTaskActivity({
       organizationId,
       projectId,
       environmentId,
       days: 6, // This actually means 7 days, because we want to show the current day too
       tasks: slugs,
     });
 
-    const runningStats = this.environmentMetricsRepository.getCurrentRunningStats({
+    const runningStats = environmentMetricsRepository.getCurrentRunningStats({
       organizationId,
       projectId,
       environmentId,
       days: 6,
       tasks: slugs,
     });
 
-    const durations = this.environmentMetricsRepository.getAverageDurations({
+    const durations = environmentMetricsRepository.getAverageDurations({
       organizationId,
       projectId,
       environmentId,
@@ -109,9 +112,5 @@ export class TaskListPresenter {
 export const taskListPresenter = singleton("taskListPresenter", setupTaskListPresenter);
 
 function setupTaskListPresenter() {
-  const environmentMetricsRepository = new ClickHouseEnvironmentMetricsRepository({
-    clickhouse: clickhouseClient,
-  });
-
-  return new TaskListPresenter(environmentMetricsRepository, $replica);
+  return new TaskListPresenter($replica);
 }

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

@@ -4,7 +4,7 @@ import { getUsage, getUsageSeries } from "~/services/platform.v3.server";
 import { createTimeSeriesData } from "~/utils/graphs";
 import { BasePresenter } from "./basePresenter.server";
 import { DataPoint, linear } from "regression";
-import { clickhouseClient } from "~/services/clickhouseInstance.server";
+import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
 
 type Options = {
   organizationId: string;
@@ -124,7 +124,8 @@ async function getTaskUsageByOrganization(
   endOfMonth: Date,
   replica: PrismaClientOrTransaction
 ) {
-  const [queryError, tasks] = await clickhouseClient.taskRuns.getTaskUsageByOrganization({
+  const clickhouse = await getClickhouseForOrganization(organizationId, "standard");
+  const [queryError, tasks] = await clickhouse.taskRuns.getTaskUsageByOrganization({
     startTime: startOfMonth.getTime(),
     endTime: endOfMonth.getTime(),
     organizationId,

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

@@ -1,7 +1,7 @@
 import { ScheduleObject } from "@trigger.dev/core/v3";
 import { PrismaClient, prisma } from "~/db.server";
 import { displayableEnvironment } from "~/models/runtimeEnvironment.server";
-import { clickhouseClient } from "~/services/clickhouseInstance.server";
+import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
 import { nextScheduledTimestamps } from "~/v3/utils/calculateNextSchedule.server";
 import { NextRunListPresenter } from "./NextRunListPresenter.server";
 import { scheduleWhereClause } from "~/models/schedules.server";
@@ -75,7 +75,8 @@ export class ViewSchedulePresenter {
       ? nextScheduledTimestamps(schedule.generatorExpression, schedule.timezone, new Date(), 5)
       : [];
 
-    const runPresenter = new NextRunListPresenter(this.#prismaClient, clickhouseClient);
+    const clickhouse = await getClickhouseForOrganization(schedule.project.organizationId, "standard");
+    const runPresenter = new NextRunListPresenter(this.#prismaClient, clickhouse);
     const { runs } = await runPresenter.call(schedule.project.organizationId, environmentId, {
       projectId: schedule.project.id,
       scheduleId: schedule.id,

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

@@ -1,5 +1,5 @@
 import { isWaitpointOutputTimeout, prettyPrintPacket } from "@trigger.dev/core/v3";
-import { clickhouseClient } from "~/services/clickhouseInstance.server";
+import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
 import { generateHttpCallbackUrl } from "~/services/httpCallback.server";
 import { logger } from "~/services/logger.server";
 import { BasePresenter } from "./basePresenter.server";
@@ -79,7 +79,8 @@ export class WaitpointPresenter extends BasePresenter {
     const connectedRuns: NextRunListItem[] = [];
 
     if (connectedRunIds.length > 0) {
-      const runPresenter = new NextRunListPresenter(this._prisma, clickhouseClient);
+      const clickhouse = await getClickhouseForOrganization(waitpoint.environment.organizationId, "standard");
+      const runPresenter = new NextRunListPresenter(this._prisma, clickhouse);
       const { runs } = await runPresenter.call(
         waitpoint.environment.organizationId,
         environmentId,

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboards.$dashboardKey/route.tsx

@@ -31,7 +31,7 @@ import {
   MetricDashboardPresenter,
 } from "~/presenters/v3/MetricDashboardPresenter.server";
 import { PromptPresenter } from "~/presenters/v3/PromptPresenter.server";
-import { clickhouseClient } from "~/services/clickhouseInstance.server";
+import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
 import { requireUser } from "~/services/session.server";
 import { cn } from "~/utils/cn";
 import { EnvironmentParamSchema } from "~/utils/pathBuilder";
@@ -74,10 +74,12 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
 
   const filters = dashboard.filters ?? ["tasks", "queues"];
 
+  const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard");
+
   // Load distinct models from ClickHouse if the dashboard has a models filter
   let possibleModels: { model: string; system: string }[] = [];
   if (filters.includes("models")) {
-    const queryFn = clickhouseClient.reader.query({
+    const queryFn = clickhouse.reader.query({
       name: "getDistinctModels",
       query: `SELECT response_model, any(gen_ai_system) AS gen_ai_system FROM trigger_dev.llm_metrics_v1 WHERE organization_id = {organizationId: String} AND project_id = {projectId: String} AND environment_id = {environmentId: String} AND response_model != '' GROUP BY response_model ORDER BY response_model`,
       params: z.object({
@@ -97,7 +99,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     }
   }
 
-  const promptPresenter = new PromptPresenter(clickhouseClient);
+  const promptPresenter = new PromptPresenter(clickhouse);
   const [possiblePrompts, possibleOperations, possibleProviders] = await Promise.all([
     filters.includes("prompts")
       ? promptPresenter.getDistinctPromptSlugs(project.organizationId, project.id, environment.id)