github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Implement organization-scoped ClickHouse instances

Browse Source 
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.
This commit is contained in:
Matt Aitken
2026-03-26 16:22:42 +00:00
parent 8aa1e55588
commit 4e57592242
39 changed files with 959 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.cursor/mcp.json
+5 -1
View File
@@ -1,3 +1,7 @@
{
"mcpServers": {}
"mcpServers": {
"linear": {
"url": "https://mcp.linear.app/mcp"
}
}
}
.server-changes/organization-scoped-clickhouse.md
+6
View File
@@ -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
+3 -2
View File
@@ -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
+3 -2
View File
@@ -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
+3 -2
View File
@@ -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
+12 -13
View File
@@ -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,9 +73,15 @@ 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,
@@ -86,7 +89,7 @@ export class TaskListPresenter {
tasks: slugs,
});
const runningStats = this.environmentMetricsRepository.getCurrentRunningStats({
const runningStats = environmentMetricsRepository.getCurrentRunningStats({
organizationId,
projectId,
environmentId,
@@ -94,7 +97,7 @@ export class TaskListPresenter {
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
+3 -2
View File
@@ -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
+3 -2
View File
@@ -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
+3 -2
View File
@@ -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
+5 -3
View File
@@ -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)
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors._index/route.tsx
+3 -2
View File
@@ -70,7 +70,7 @@ import {
type ErrorOccurrences,
type ErrorsList as ErrorsListData,
} from "~/presenters/v3/ErrorsListPresenter.server";
import { logsClickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import { getCurrentPlan } from "~/services/platform.v3.server";
import { requireUser } from "~/services/session.server";
import { formatNumberCompact } from "~/utils/numberFormatter";
@@ -123,7 +123,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const plan = await getCurrentPlan(project.organizationId);
const retentionLimitDays = plan?.v3Subscription?.plan?.limits.logRetentionDays.number ?? 30;
const presenter = new ErrorsListPresenter($replica, logsClickhouseClient);
const logsClickhouse = await getClickhouseForOrganization(project.organizationId, "logs");
const presenter = new ErrorsListPresenter($replica, logsClickhouse);
const listPromise = presenter
.call(project.organizationId, environment.id, {
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsx
+3 -2
View File
@@ -16,7 +16,7 @@ import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { LogsListPresenter, LogEntry } from "~/presenters/v3/LogsListPresenter.server";
import type { LogLevel } from "~/utils/logUtils";
import { $replica, prisma } from "~/db.server";
import { logsClickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import { NavBar, PageTitle } from "~/components/primitives/PageHeader";
import { PageBody, PageContainer } from "~/components/layout/AppLayout";
import { Suspense, useCallback, useEffect, useMemo, useRef, useState, useTransition } from "react";
@@ -137,7 +137,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const plan = await getCurrentPlan(project.organizationId);
const retentionLimitDays = plan?.v3Subscription?.plan?.limits.logRetentionDays.number ?? 30;
const presenter = new LogsListPresenter($replica, logsClickhouseClient);
const logsClickhouse = await getClickhouseForOrganization(project.organizationId, "logs");
const presenter = new LogsListPresenter($replica, logsClickhouse);
const listPromise = presenter
.call(project.organizationId, environment.id, {
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.$modelId/route.tsx
+3 -2
View File
@@ -28,7 +28,7 @@ import type { QueryWidgetConfig } from "~/components/metrics/QueryWidget";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { ModelRegistryPresenter } from "~/presenters/v3/ModelRegistryPresenter.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import { requireUserId } from "~/services/session.server";
import { useOrganization } from "~/hooks/useOrganizations";
import { useProject } from "~/hooks/useProject";
@@ -68,7 +68,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
throw new Response("Environment not found", { status: 404 });
}
const presenter = new ModelRegistryPresenter(clickhouseClient);
const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard");
const presenter = new ModelRegistryPresenter(clickhouse);
const model = await presenter.getModelDetail(modelId);
if (!model) {
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models._index/route.tsx
+3 -2
View File
@@ -71,7 +71,7 @@ import {
type PopularModel,
ModelRegistryPresenter,
} from "~/presenters/v3/ModelRegistryPresenter.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import { requireUserId } from "~/services/session.server";
import { useEnvironment } from "~/hooks/useEnvironment";
import { useOrganization } from "~/hooks/useOrganizations";
@@ -109,7 +109,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
throw new Response("Environment not found", { status: 404 });
}
const presenter = new ModelRegistryPresenter(clickhouseClient);
const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard");
const presenter = new ModelRegistryPresenter(clickhouse);
const catalog = await presenter.getModelCatalog();
const now = new Date();
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.compare/route.tsx
+3 -2
View File
@@ -20,7 +20,7 @@ import {
type ModelComparisonItem,
ModelRegistryPresenter,
} from "~/presenters/v3/ModelRegistryPresenter.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import { requireUserId } from "~/services/session.server";
import { useOrganization } from "~/hooks/useOrganizations";
import { useProject } from "~/hooks/useProject";
@@ -55,7 +55,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
return typedjson({ comparison: [] as ModelComparisonItem[], models: responseModels });
}
const presenter = new ModelRegistryPresenter(clickhouseClient);
const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard");
const presenter = new ModelRegistryPresenter(clickhouse);
const now = new Date();
const sevenDaysAgo = new Date(now.getTime() - 7 * 24 * 60 * 60 * 1000);
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug/route.tsx
+4 -3
View File
@@ -70,7 +70,7 @@ import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { type GenerationRow, PromptPresenter } from "~/presenters/v3/PromptPresenter.server";
import { SpanView } from "~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import { getResizableSnapshot } from "~/services/resizablePanel.server";
import { requireUserId } from "~/services/session.server";
import { PromptService } from "~/v3/services/promptService.server";
@@ -242,7 +242,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const startTime = fromTime ? new Date(fromTime) : new Date(Date.now() - periodMs);
const endTime = toTime ? new Date(toTime) : new Date();
const presenter = new PromptPresenter(clickhouseClient);
const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard");
const presenter = new PromptPresenter(clickhouse);
let generations: Awaited<ReturnType<typeof presenter.listGenerations>>["generations"] = [];
let generationsPagination: { next?: string } = {};
try {
@@ -273,7 +274,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
// Load distinct filter values and resizable snapshots in parallel
const distinctQuery = (col: string, name: string) =>
clickhouseClient.reader.query({
clickhouse.reader.query({
name,
query: `SELECT DISTINCT ${col} AS val FROM trigger_dev.llm_metrics_v1 WHERE environment_id = {environmentId: String} AND prompt_slug = {promptSlug: String} AND ${col} != '' ORDER BY val`,
params: z.object({ environmentId: z.string(), promptSlug: z.string() }),
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts._index/route.tsx
+3 -2
View File
@@ -22,7 +22,7 @@ import { useProject } from "~/hooks/useProject";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { PromptPresenter } from "~/presenters/v3/PromptPresenter.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import { requireUserId } from "~/services/session.server";
import { docsPath, EnvironmentParamSchema, v3PromptsPath } from "~/utils/pathBuilder";
import { LinkButton } from "~/components/primitives/Buttons";
@@ -46,7 +46,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
throw new Response("Environment not found", { status: 404 });
}
const presenter = new PromptPresenter(clickhouseClient);
const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard");
const presenter = new PromptPresenter(clickhouse);
const prompts = await presenter.listPrompts(project.id, environment.id);
const sparklines = await presenter.getUsageSparklines(
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam/route.tsx
+3 -2
View File
@@ -92,7 +92,7 @@ import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { NextRunListPresenter } from "~/presenters/v3/NextRunListPresenter.server";
import { RunEnvironmentMismatchError, RunPresenter } from "~/presenters/v3/RunPresenter.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import { getImpersonationId } from "~/services/impersonation.server";
import { logger } from "~/services/logger.server";
import { getResizableSnapshot } from "~/services/resizablePanel.server";
@@ -182,7 +182,8 @@ async function getRunsListFromTableState({
return null;
}
const runsListPresenter = new NextRunListPresenter($replica, clickhouseClient);
const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard");
const runsListPresenter = new NextRunListPresenter($replica, clickhouse);
const currentPageResult = await runsListPresenter.call(project.organizationId, environment.id, {
userId,
projectId: project.id,
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx
+3 -2
View File
@@ -44,7 +44,7 @@ import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { getRunFiltersFromRequest } from "~/presenters/RunFilters.server";
import { NextRunListPresenter } from "~/presenters/v3/NextRunListPresenter.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import {
setRootOnlyFilterPreference,
uiPreferencesStorage,
@@ -87,7 +87,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const filters = await getRunFiltersFromRequest(request);
const presenter = new NextRunListPresenter($replica, clickhouseClient);
const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard");
const presenter = new NextRunListPresenter($replica, clickhouse);
const list = presenter.call(project.organizationId, environment.id, {
userId,
projectId: project.id,
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.test.tasks.$taskParam/route.tsx
+3 -2
View File
@@ -74,7 +74,7 @@ import { Dialog, DialogContent, DialogHeader, DialogTrigger } from "~/components
import { DialogClose, DialogDescription } from "@radix-ui/react-dialog";
import { FormButtons } from "~/components/primitives/FormButtons";
import { $replica } from "~/db.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import { RegionsPresenter, type Region } from "~/presenters/v3/RegionsPresenter.server";
import { TestSidebarTabs } from "./TestSidebarTabs";
import { AIPayloadTabContent } from "./AIPayloadTabContent";
@@ -102,7 +102,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
});
}
const presenter = new TestTaskPresenter($replica, clickhouseClient);
const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard");
const presenter = new TestTaskPresenter($replica, clickhouse);
try {
const [result, regionsResult] = await Promise.all([
presenter.call({
apps/webapp/app/routes/api.v1.prompts.$slug.ts
+12 -3
View File
@@ -2,7 +2,7 @@ import { json } from "@remix-run/server-runtime";
import { z } from "zod";
import { prisma } from "~/db.server";
import { PromptPresenter } from "~/presenters/v3/PromptPresenter.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import {
createActionApiRoute,
createLoaderApiRoute,
@@ -33,6 +33,13 @@ export const loader = createLoaderApiRoute(
slug: params.slug,
},
},
include: {
project: {
select: {
organizationId: true,
},
},
},
});
},
authorization: {
@@ -46,7 +53,8 @@ export const loader = createLoaderApiRoute(
return json({ error: "Prompt not found" }, { status: 404 });
}
const presenter = new PromptPresenter(clickhouseClient);
const clickhouse = await getClickhouseForOrganization(prompt.project.organizationId, "standard");
const presenter = new PromptPresenter(clickhouse);
const version = await presenter.resolveVersion(prompt.id, {
version: searchParams.version,
label: searchParams.label,
@@ -117,7 +125,8 @@ const { action } = createActionApiRoute(
return json({ error: "Prompt not found" }, { status: 404 });
}
const presenter = new PromptPresenter(clickhouseClient);
const clickhouse = await getClickhouseForOrganization(authentication.environment.organizationId, "standard");
const presenter = new PromptPresenter(clickhouse);
const version = await presenter.resolveVersion(prompt.id, {
version: body.version,
label: body.label,
apps/webapp/app/routes/api.v1.prompts.$slug.versions.ts
+10 -2
View File
@@ -2,7 +2,7 @@ import { json } from "@remix-run/server-runtime";
import { z } from "zod";
import { prisma } from "~/db.server";
import { PromptPresenter } from "~/presenters/v3/PromptPresenter.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
const ParamsSchema = z.object({
@@ -23,6 +23,13 @@ export const loader = createLoaderApiRoute(
slug: params.slug,
},
},
include: {
project: {
select: {
organizationId: true,
},
},
},
});
},
authorization: {
@@ -36,7 +43,8 @@ export const loader = createLoaderApiRoute(
return json({ error: "Prompt not found" }, { status: 404 });
}
const presenter = new PromptPresenter(clickhouseClient);
const clickhouse = await getClickhouseForOrganization(prompt.project.organizationId, "standard");
const presenter = new PromptPresenter(clickhouse);
const versions = await presenter.listVersions(prompt.id);
return json({
apps/webapp/app/routes/api.v1.prompts._index.ts
+3 -2
View File
@@ -1,6 +1,6 @@
import { json } from "@remix-run/server-runtime";
import { PromptPresenter } from "~/presenters/v3/PromptPresenter.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
export const loader = createLoaderApiRoute(
@@ -15,7 +15,8 @@ export const loader = createLoaderApiRoute(
},
},
async ({ authentication }) => {
const presenter = new PromptPresenter(clickhouseClient);
const clickhouse = await getClickhouseForOrganization(authentication.environment.organizationId, "standard");
const presenter = new PromptPresenter(clickhouse);
const prompts = await presenter.listPrompts(
authentication.environment.projectId,
authentication.environment.id
apps/webapp/app/routes/otel.v1.logs.ts
+3 -2
View File
@@ -4,12 +4,13 @@ import { otlpExporter } from "~/v3/otlpExporter.server";
export async function action({ request }: ActionFunctionArgs) {
try {
const exporter = await otlpExporter;
const contentType = request.headers.get("content-type")?.toLowerCase() ?? "";
if (contentType.startsWith("application/json")) {
const body = await request.json();
const exportResponse = await otlpExporter.exportLogs(body as ExportLogsServiceRequest);
const exportResponse = await exporter.exportLogs(body as ExportLogsServiceRequest);
return json(exportResponse, { status: 200 });
} else if (contentType.startsWith("application/x-protobuf")) {
@@ -17,7 +18,7 @@ export async function action({ request }: ActionFunctionArgs) {
const exportRequest = ExportLogsServiceRequest.decode(new Uint8Array(buffer));
const exportResponse = await otlpExporter.exportLogs(exportRequest);
const exportResponse = await exporter.exportLogs(exportRequest);
return new Response(ExportLogsServiceResponse.encode(exportResponse).finish(), {
status: 200,
apps/webapp/app/routes/otel.v1.metrics.ts
+3 -2
View File
@@ -7,12 +7,13 @@ import { otlpExporter } from "~/v3/otlpExporter.server";
export async function action({ request }: ActionFunctionArgs) {
try {
const exporter = await otlpExporter;
const contentType = request.headers.get("content-type")?.toLowerCase() ?? "";
if (contentType.startsWith("application/json")) {
const body = await request.json();
const exportResponse = await otlpExporter.exportMetrics(
const exportResponse = await exporter.exportMetrics(
body as ExportMetricsServiceRequest
);
@@ -22,7 +23,7 @@ export async function action({ request }: ActionFunctionArgs) {
const exportRequest = ExportMetricsServiceRequest.decode(new Uint8Array(buffer));
const exportResponse = await otlpExporter.exportMetrics(exportRequest);
const exportResponse = await exporter.exportMetrics(exportRequest);
return new Response(ExportMetricsServiceResponse.encode(exportResponse).finish(), {
status: 200,
apps/webapp/app/routes/otel.v1.traces.ts
+3 -2
View File
@@ -4,12 +4,13 @@ import { otlpExporter } from "~/v3/otlpExporter.server";
export async function action({ request }: ActionFunctionArgs) {
try {
const exporter = await otlpExporter;
const contentType = request.headers.get("content-type")?.toLowerCase() ?? "";
if (contentType.startsWith("application/json")) {
const body = await request.json();
const exportResponse = await otlpExporter.exportTraces(body as ExportTraceServiceRequest);
const exportResponse = await exporter.exportTraces(body as ExportTraceServiceRequest);
return json(exportResponse, { status: 200 });
} else if (contentType.startsWith("application/x-protobuf")) {
@@ -17,7 +18,7 @@ export async function action({ request }: ActionFunctionArgs) {
const exportRequest = ExportTraceServiceRequest.decode(new Uint8Array(buffer));
const exportResponse = await otlpExporter.exportTraces(exportRequest);
const exportResponse = await exporter.exportTraces(exportRequest);
return new Response(ExportTraceServiceResponse.encode(exportResponse).finish(), {
status: 200,
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsx
+3 -2
View File
@@ -1,7 +1,7 @@
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { typedjson } from "remix-typedjson";
import { z } from "zod";
import { logsClickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import { requireUserId } from "~/services/session.server";
import { LogDetailPresenter } from "~/presenters/v3/LogDetailPresenter.server";
import { findProjectBySlug } from "~/models/project.server";
@@ -43,7 +43,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const [traceId, spanId, , startTime] = parts;
const presenter = new LogDetailPresenter($replica, logsClickhouseClient);
const logsClickhouse = await getClickhouseForOrganization(project.organizationId, "logs");
const presenter = new LogDetailPresenter($replica, logsClickhouse);
let result;
try {
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.ts
+3 -2
View File
@@ -6,7 +6,7 @@ import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { LogsListPresenter, type LogLevel, LogsListOptionsSchema } from "~/presenters/v3/LogsListPresenter.server";
import { $replica } from "~/db.server";
import { logsClickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import { getCurrentPlan } from "~/services/platform.v3.server";
// Valid log levels for filtering
@@ -69,7 +69,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
retentionLimitDays,
}) as any; // Validated by LogsListOptionsSchema at runtime
const presenter = new LogsListPresenter($replica, logsClickhouseClient);
const logsClickhouse = await getClickhouseForOrganization(project.organizationId, "logs");
const presenter = new LogsListPresenter($replica, logsClickhouse);
const result = await presenter.call(project.organizationId, environment.id, options);
return json({
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug.generations.ts
+3 -2
View File
@@ -6,7 +6,7 @@ import { EnvironmentParamSchema } from "~/utils/pathBuilder";
import { parsePeriodToMs } from "~/utils/periods";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
import {
PromptPresenter,
type GenerationRow,
@@ -59,7 +59,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const operations = url.searchParams.getAll("operations").filter(Boolean);
const providers = url.searchParams.getAll("providers").filter(Boolean);
const presenter = new PromptPresenter(clickhouseClient);
const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard");
const presenter = new PromptPresenter(clickhouse);
const result = await presenter.listGenerations({
environmentId: environment.id,
promptSlug,
apps/webapp/app/services/admin/missingLlmModels.server.ts
+7 -3
View File
@@ -1,4 +1,4 @@
import { adminClickhouseClient } from "~/services/clickhouseInstance.server";
import { getAdminClickhouse } from "~/services/clickhouse/clickhouseFactory.server";
import { llmPricingRegistry } from "~/v3/llmPricingRegistry.server";
export type MissingLlmModel = {
@@ -13,8 +13,10 @@ export async function getMissingLlmModels(opts: {
const lookbackHours = opts.lookbackHours ?? 24;
const since = new Date(Date.now() - lookbackHours * 60 * 60 * 1000);
const adminClickhouse = getAdminClickhouse();
// queryBuilderFast returns a factory function — call it to get the builder
const createBuilder = adminClickhouseClient.reader.queryBuilderFast<{
const createBuilder = adminClickhouse.reader.queryBuilderFast<{
model: string;
system: string;
cnt: string;
@@ -93,7 +95,9 @@ export async function getMissingModelSamples(opts: {
const limit = opts.limit ?? 10;
const since = new Date(Date.now() - lookbackHours * 60 * 60 * 1000);
const createBuilder = adminClickhouseClient.reader.queryBuilderFast<MissingModelSample>({
const adminClickhouse = getAdminClickhouse();
const createBuilder = adminClickhouse.reader.queryBuilderFast<MissingModelSample>({
name: "missingModelSamples",
table: "trigger_dev.task_events_v2",
columns: [
apps/webapp/app/services/clickhouse/clickhouseCredentialsService.server.ts
+109
View File
@@ -0,0 +1,109 @@
import { getSecretStore } from "~/services/secrets/secretStore.server";
import { prisma } from "~/db.server";
import {
ClickhouseConnectionSchema,
getClickhouseSecretKey,
} from "./clickhouseSecretSchemas.server";
import { clearClickhouseCacheForOrganization } from "./clickhouseFactory.server";
export async function setOrganizationClickhouseUrl(
organizationId: string,
clientType: "standard" | "events" | "replication",
url: string
): Promise<void> {
// Validate URL format
const connection = ClickhouseConnectionSchema.parse({ url });
// Store in SecretStore
const secretStore = getSecretStore("DATABASE");
const secretKey = getClickhouseSecretKey(organizationId, clientType);
await secretStore.setSecret(secretKey, connection);
// Update featureFlags to reference the secret
const org = await prisma.organization.findUnique({
where: { id: organizationId },
select: { featureFlags: true },
});
const featureFlags = (org?.featureFlags || {}) as any;
const clickhouseConfig = featureFlags.clickhouse || {};
clickhouseConfig[clientType] = secretKey;
featureFlags.clickhouse = clickhouseConfig;
await prisma.organization.update({
where: { id: organizationId },
data: { featureFlags },
});
// Clear cache
clearClickhouseCacheForOrganization(organizationId);
}
export async function removeOrganizationClickhouseUrl(
organizationId: string,
clientType: "standard" | "events" | "replication"
): Promise<void> {
// Remove from SecretStore
const secretStore = getSecretStore("DATABASE");
const secretKey = getClickhouseSecretKey(organizationId, clientType);
await secretStore.deleteSecret(secretKey);
// Update featureFlags
const org = await prisma.organization.findUnique({
where: { id: organizationId },
select: { featureFlags: true },
});
if (org?.featureFlags) {
const featureFlags = org.featureFlags as any;
if (featureFlags.clickhouse && featureFlags.clickhouse[clientType]) {
delete featureFlags.clickhouse[clientType];
// If no more clickhouse configs, remove the clickhouse key entirely
if (Object.keys(featureFlags.clickhouse).length === 0) {
delete featureFlags.clickhouse;
}
await prisma.organization.update({
where: { id: organizationId },
data: { featureFlags },
});
}
}
// Clear cache
clearClickhouseCacheForOrganization(organizationId);
}
export async function getOrganizationClickhouseUrl(
organizationId: string,
clientType: "standard" | "events" | "replication"
): Promise<string | null> {
const org = await prisma.organization.findUnique({
where: { id: organizationId },
select: { featureFlags: true },
});
if (!org?.featureFlags) {
return null;
}
const clickhouseConfig = (org.featureFlags as any).clickhouse;
if (!clickhouseConfig || typeof clickhouseConfig !== "object") {
return null;
}
const secretKey = clickhouseConfig[clientType];
if (!secretKey || typeof secretKey !== "string") {
return null;
}
const secretStore = getSecretStore("DATABASE");
const connection = await secretStore.getSecret(ClickhouseConnectionSchema, secretKey);
if (!connection) {
return null;
}
return connection.url;
}
apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts
+422
View File
@@ -0,0 +1,422 @@
/**
* ClickHouse Factory - Organization-Scoped ClickHouse Routing
*
* This module provides organization-scoped ClickHouse instance routing to support:
* - HIPAA compliance (dedicated ClickHouse clusters)
* - High-volume customer isolation
* - Geographic data residency requirements
* - Performance tier differentiation
*
* ## Architecture
*
* ### Credential Storage
* - ClickHouse URLs stored encrypted in SecretStore (AES-256-GCM)
* - Organization references secret via `featureFlags.clickhouse` JSON
* - No plaintext credentials in database
*
* ### Caching Strategy
* - **Org configs**: Unkey cache with LRU memory (5min fresh, 10min stale, SWR)
* - **ClickHouse clients**: Cached by hostname hash (multiple orgs share same instance)
* - **Event repositories**: Cached by hostname hash (stateful, must be reused)
* - **Security**: Memory-only cache for org configs (no credentials in Redis)
*
* ## Usage in Presenters
*
* Presenters should fetch org-specific ClickHouse clients in their `call()` method:
*
* ```typescript
* import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server";
*
* export class MyPresenter extends BasePresenter {
* constructor(private options: PresenterOptions = {}) {
* super();
* }
*
* async call({ organizationId, ... }) {
* const clickhouse = await getClickhouseForOrganization(organizationId, "standard");
* // Use clickhouse for queries...
* }
* }
* ```
*
* ## Usage in Services
*
* The replication service and OTLP exporter automatically route data by organization.
* Other services should follow the same pattern when working with ClickHouse.
*
* @module clickhouseFactory
*/
import { ClickHouse } from "@internal/clickhouse";
import { createHash } from "crypto";
import { createCache, DefaultStatefulContext, Namespace } from "@unkey/cache";
import { createLRUMemoryStore } from "@internal/cache";
import { getSecretStore } from "~/services/secrets/secretStore.server";
import { prisma } from "~/db.server";
import {
ClickhouseConnectionSchema,
getClickhouseSecretKey,
} from "./clickhouseSecretSchemas.server";
import { ClickhouseEventRepository } from "~/v3/eventRepository/clickhouseEventRepository.server";
import { env } from "~/env.server";
import { singleton } from "~/utils/singleton";
// Module-level caches for ClickHouse clients and event repositories
const clickhouseClientCache = new Map<string, ClickHouse>();
const eventRepositoryCache = new Map<string, ClickhouseEventRepository>();
// Default ClickHouse clients (not exported - internal use only)
const defaultClickhouseClient = singleton("clickhouseClient", initializeClickhouseClient);
function initializeClickhouseClient() {
const url = new URL(env.CLICKHOUSE_URL);
url.searchParams.delete("secure");
console.log(`🗃️ Clickhouse service enabled to host ${url.host}`);
return new ClickHouse({
url: url.toString(),
name: "clickhouse-instance",
keepAlive: {
enabled: env.CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
idleSocketTtl: env.CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
},
logLevel: env.CLICKHOUSE_LOG_LEVEL,
compression: {
request: true,
},
maxOpenConnections: env.CLICKHOUSE_MAX_OPEN_CONNECTIONS,
});
}
const defaultLogsClickhouseClient = singleton(
"logsClickhouseClient",
initializeLogsClickhouseClient
);
function initializeLogsClickhouseClient() {
if (!env.LOGS_CLICKHOUSE_URL) {
throw new Error("LOGS_CLICKHOUSE_URL is not set");
}
const url = new URL(env.LOGS_CLICKHOUSE_URL);
url.searchParams.delete("secure");
return new ClickHouse({
url: url.toString(),
name: "logs-clickhouse",
keepAlive: {
enabled: env.CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
idleSocketTtl: env.CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
},
logLevel: env.CLICKHOUSE_LOG_LEVEL,
compression: {
request: true,
},
maxOpenConnections: env.CLICKHOUSE_MAX_OPEN_CONNECTIONS,
clickhouseSettings: {
max_memory_usage: env.CLICKHOUSE_LOGS_LIST_MAX_MEMORY_USAGE.toString(),
max_bytes_before_external_sort:
env.CLICKHOUSE_LOGS_LIST_MAX_BYTES_BEFORE_EXTERNAL_SORT.toString(),
max_threads: env.CLICKHOUSE_LOGS_LIST_MAX_THREADS,
...(env.CLICKHOUSE_LOGS_LIST_MAX_ROWS_TO_READ && {
max_rows_to_read: env.CLICKHOUSE_LOGS_LIST_MAX_ROWS_TO_READ.toString(),
}),
...(env.CLICKHOUSE_LOGS_LIST_MAX_EXECUTION_TIME && {
max_execution_time: env.CLICKHOUSE_LOGS_LIST_MAX_EXECUTION_TIME,
}),
},
});
}
const defaultAdminClickhouseClient = singleton(
"adminClickhouseClient",
initializeAdminClickhouseClient
);
function initializeAdminClickhouseClient() {
if (!env.ADMIN_CLICKHOUSE_URL) {
throw new Error("ADMIN_CLICKHOUSE_URL is not set");
}
const url = new URL(env.ADMIN_CLICKHOUSE_URL);
url.searchParams.delete("secure");
return new ClickHouse({
url: url.toString(),
name: "admin-clickhouse",
keepAlive: {
enabled: env.CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
idleSocketTtl: env.CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
},
logLevel: env.CLICKHOUSE_LOG_LEVEL,
compression: {
request: true,
},
maxOpenConnections: env.CLICKHOUSE_MAX_OPEN_CONNECTIONS,
});
}
const defaultQueryClickhouseClient = singleton(
"queryClickhouseClient",
initializeQueryClickhouseClient
);
function initializeQueryClickhouseClient() {
if (!env.QUERY_CLICKHOUSE_URL) {
throw new Error("QUERY_CLICKHOUSE_URL is not set");
}
const url = new URL(env.QUERY_CLICKHOUSE_URL);
url.searchParams.delete("secure");
return new ClickHouse({
url: url.toString(),
name: "query-clickhouse",
keepAlive: {
enabled: env.CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
idleSocketTtl: env.CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
},
logLevel: env.CLICKHOUSE_LOG_LEVEL,
compression: {
request: true,
},
maxOpenConnections: env.CLICKHOUSE_MAX_OPEN_CONNECTIONS,
});
}
// Org config cache with Unkey (memory-only, no Redis for security)
type OrgClickhouseConfig = {
organizationId: string;
hostnameHash: string;
url: string;
clientType: string;
};
const ctx = new DefaultStatefulContext();
const memory = createLRUMemoryStore(1000);
const orgConfigCache = createCache({
orgClickhouse: new Namespace<OrgClickhouseConfig | null>(ctx, {
stores: [memory], // Memory-only, no Redis store for security
fresh: 5 * 60 * 1000, // 5 minutes
stale: 10 * 60 * 1000, // 10 minutes (SWR pattern)
}),
});
function hashHostname(url: string): string {
const parsed = new URL(url);
return createHash("sha256").update(parsed.hostname).digest("hex");
}
async function getOrgClickhouseConfig(
ctx: DefaultStatefulContext,
orgId: string,
clientType: string
): Promise<OrgClickhouseConfig | null> {
const org = await prisma.organization.findUnique({
where: { id: orgId },
select: { featureFlags: true },
});
if (!org?.featureFlags) {
return null;
}
const clickhouseConfig = (org.featureFlags as any).clickhouse;
if (!clickhouseConfig || typeof clickhouseConfig !== "object") {
return null;
}
const secretKey = clickhouseConfig[clientType];
if (!secretKey || typeof secretKey !== "string") {
return null;
}
const secretStore = getSecretStore("DATABASE");
const connection = await secretStore.getSecret(ClickhouseConnectionSchema, secretKey);
if (!connection) {
return null;
}
const hostnameHash = hashHostname(connection.url);
return {
organizationId: orgId,
hostnameHash,
url: connection.url,
clientType,
};
}
export async function getClickhouseForOrganization(
organizationId: string,
clientType: "standard" | "events" | "replication" | "logs" | "query" | "admin"
): Promise<ClickHouse> {
// Try to get org-specific config
const configResult = await orgConfigCache.orgClickhouse.swr(
`org:${organizationId}:ch:${clientType}`,
async () => getOrgClickhouseConfig(ctx, organizationId, clientType)
);
// Handle Result type - check for error or null value
const config = configResult.err ? null : configResult.val;
// If no custom config, return appropriate default client
if (!config) {
switch (clientType) {
case "standard":
case "events":
case "replication":
return defaultClickhouseClient;
case "logs":
return defaultLogsClickhouseClient;
case "query":
return defaultQueryClickhouseClient;
case "admin":
return defaultAdminClickhouseClient;
}
}
// Check if client already exists for this hostname
const cacheKey = `${config.hostnameHash}:${clientType}`;
let client = clickhouseClientCache.get(cacheKey);
if (!client) {
const url = new URL(config.url);
url.searchParams.delete("secure");
client = new ClickHouse({
url: url.toString(),
name: `org-clickhouse-${clientType}`,
keepAlive: {
enabled: env.CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
idleSocketTtl: env.CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
},
logLevel: env.CLICKHOUSE_LOG_LEVEL,
compression: {
request: true,
},
maxOpenConnections: env.CLICKHOUSE_MAX_OPEN_CONNECTIONS,
});
clickhouseClientCache.set(cacheKey, client);
}
return client;
}
export async function getEventRepositoryForOrganization(
organizationId: string
): Promise<ClickhouseEventRepository> {
// Try to get org-specific config
const configResult = await orgConfigCache.orgClickhouse.swr(
`org:${organizationId}:ch:events`,
async () => getOrgClickhouseConfig(ctx, organizationId, "events")
);
// Handle Result type - check for error or null value
const config = configResult.err ? null : configResult.val;
// If no custom config, return default repository (created on demand)
if (!config) {
const defaultKey = "default:events";
let defaultRepo = eventRepositoryCache.get(defaultKey);
if (!defaultRepo) {
// Create default event repository using standard clickhouse client
// This matches the existing pattern in clickhouseEventRepositoryInstance.server.ts
const eventsClickhouse = await getEventsClickhouseClient();
defaultRepo = new ClickhouseEventRepository({
clickhouse: eventsClickhouse,
batchSize: env.EVENTS_CLICKHOUSE_BATCH_SIZE,
flushInterval: env.EVENTS_CLICKHOUSE_FLUSH_INTERVAL_MS,
maximumTraceSummaryViewCount: env.EVENTS_CLICKHOUSE_MAX_TRACE_SUMMARY_VIEW_COUNT,
maximumTraceDetailedSummaryViewCount:
env.EVENTS_CLICKHOUSE_MAX_TRACE_DETAILED_SUMMARY_VIEW_COUNT,
maximumLiveReloadingSetting: env.EVENTS_CLICKHOUSE_MAX_LIVE_RELOADING_SETTING,
insertStrategy: env.EVENTS_CLICKHOUSE_INSERT_STRATEGY,
waitForAsyncInsert: env.EVENTS_CLICKHOUSE_WAIT_FOR_ASYNC_INSERT === "1",
asyncInsertMaxDataSize: env.EVENTS_CLICKHOUSE_ASYNC_INSERT_MAX_DATA_SIZE,
asyncInsertBusyTimeoutMs: env.EVENTS_CLICKHOUSE_ASYNC_INSERT_BUSY_TIMEOUT_MS,
startTimeMaxAgeMs: env.EVENTS_CLICKHOUSE_START_TIME_MAX_AGE_MS,
llmMetricsBatchSize: env.LLM_METRICS_BATCH_SIZE,
llmMetricsFlushInterval: env.LLM_METRICS_FLUSH_INTERVAL_MS,
llmMetricsMaxBatchSize: env.LLM_METRICS_MAX_BATCH_SIZE,
llmMetricsMaxConcurrency: env.LLM_METRICS_MAX_CONCURRENCY,
version: "v2",
});
eventRepositoryCache.set(defaultKey, defaultRepo);
}
return defaultRepo;
}
// Check if repository already exists for this hostname
const cacheKey = `${config.hostnameHash}:events`;
let repository = eventRepositoryCache.get(cacheKey);
if (!repository) {
const client = await getClickhouseForOrganization(organizationId, "events");
repository = new ClickhouseEventRepository({
clickhouse: client,
batchSize: env.EVENTS_CLICKHOUSE_BATCH_SIZE,
flushInterval: env.EVENTS_CLICKHOUSE_FLUSH_INTERVAL_MS,
maximumTraceSummaryViewCount: env.EVENTS_CLICKHOUSE_MAX_TRACE_SUMMARY_VIEW_COUNT,
maximumTraceDetailedSummaryViewCount:
env.EVENTS_CLICKHOUSE_MAX_TRACE_DETAILED_SUMMARY_VIEW_COUNT,
maximumLiveReloadingSetting: env.EVENTS_CLICKHOUSE_MAX_LIVE_RELOADING_SETTING,
insertStrategy: env.EVENTS_CLICKHOUSE_INSERT_STRATEGY,
waitForAsyncInsert: env.EVENTS_CLICKHOUSE_WAIT_FOR_ASYNC_INSERT === "1",
asyncInsertMaxDataSize: env.EVENTS_CLICKHOUSE_ASYNC_INSERT_MAX_DATA_SIZE,
asyncInsertBusyTimeoutMs: env.EVENTS_CLICKHOUSE_ASYNC_INSERT_BUSY_TIMEOUT_MS,
startTimeMaxAgeMs: env.EVENTS_CLICKHOUSE_START_TIME_MAX_AGE_MS,
llmMetricsBatchSize: env.LLM_METRICS_BATCH_SIZE,
llmMetricsFlushInterval: env.LLM_METRICS_FLUSH_INTERVAL_MS,
llmMetricsMaxBatchSize: env.LLM_METRICS_MAX_BATCH_SIZE,
llmMetricsMaxConcurrency: env.LLM_METRICS_MAX_CONCURRENCY,
version: "v2",
});
eventRepositoryCache.set(cacheKey, repository);
}
return repository;
}
// Helper to create the default events ClickHouse client
async function getEventsClickhouseClient(): Promise<ClickHouse> {
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");
return 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: env.EVENTS_CLICKHOUSE_COMPRESSION_REQUEST === "1",
},
maxOpenConnections: env.EVENTS_CLICKHOUSE_MAX_OPEN_CONNECTIONS,
});
}
/**
* Get admin ClickHouse client for cross-organization queries
* This should only be used for admin tools and analytics that need to query across all orgs
*/
export function getAdminClickhouse(): ClickHouse {
return defaultAdminClickhouseClient;
}
// Clear caches when needed (e.g., when org config changes)
export function clearClickhouseCacheForOrganization(organizationId: string): void {
// The Unkey cache will naturally expire based on TTL (5min fresh, 10min stale)
// No explicit removal needed - cache entries will be refreshed on next access
// Note: We don't clear client/repository caches as they're keyed by hostname
// and may be shared by other orgs
}
apps/webapp/app/services/clickhouse/clickhouseFactory.test.ts
+155
View File
@@ -0,0 +1,155 @@
import { describe, it, expect, beforeEach, vi } from "vitest";
import { prisma } from "~/db.server";
import {
getClickhouseForOrganization,
getEventRepositoryForOrganization,
clearClickhouseCacheForOrganization,
} from "./clickhouseFactory.server";
import {
setOrganizationClickhouseUrl,
removeOrganizationClickhouseUrl,
getOrganizationClickhouseUrl,
} from "./clickhouseCredentialsService.server";
describe("ClickHouse Factory", () => {
const testOrgId = "test-org-" + Date.now();
const testClickhouseUrl = "https://test-ch.example.com:8443?user=test&password=secret";
beforeEach(async () => {
// Clean up any existing test data
await prisma.organization.deleteMany({
where: { id: testOrgId },
});
});
it("should return default ClickHouse client when org has no custom config", async () => {
const client = await getClickhouseForOrganization(testOrgId, "standard");
expect(client).toBeDefined();
// Default client should be returned (not null)
expect(client).toBeTruthy();
});
it("should set and retrieve organization ClickHouse URL", async () => {
// First create the test organization
await prisma.organization.create({
data: {
id: testOrgId,
title: "Test Org",
slug: "test-org-" + Date.now(),
},
});
// Set the URL
await setOrganizationClickhouseUrl(testOrgId, "standard", testClickhouseUrl);
// Retrieve it
const retrievedUrl = await getOrganizationClickhouseUrl(testOrgId, "standard");
expect(retrievedUrl).toBe(testClickhouseUrl);
// Verify it's stored in featureFlags
const org = await prisma.organization.findUnique({
where: { id: testOrgId },
select: { featureFlags: true },
});
expect(org?.featureFlags).toBeDefined();
const featureFlags = org?.featureFlags as any;
expect(featureFlags.clickhouse).toBeDefined();
expect(featureFlags.clickhouse.standard).toBeDefined();
// Clean up
await removeOrganizationClickhouseUrl(testOrgId, "standard");
await prisma.organization.delete({ where: { id: testOrgId } });
});
it("should remove organization ClickHouse URL", async () => {
// First create the test organization
await prisma.organization.create({
data: {
id: testOrgId,
title: "Test Org",
slug: "test-org-" + Date.now(),
},
});
// Set and then remove
await setOrganizationClickhouseUrl(testOrgId, "standard", testClickhouseUrl);
await removeOrganizationClickhouseUrl(testOrgId, "standard");
// Verify it's gone
const retrievedUrl = await getOrganizationClickhouseUrl(testOrgId, "standard");
expect(retrievedUrl).toBeNull();
// Clean up
await prisma.organization.delete({ where: { id: testOrgId } });
});
it("should cache ClickHouse clients by hostname", async () => {
// This test verifies that multiple orgs pointing to the same ClickHouse hostname
// share the same client instance (deduplication)
const org1Id = testOrgId + "-1";
const org2Id = testOrgId + "-2";
// Create test organizations
await prisma.organization.createMany({
data: [
{ id: org1Id, title: "Test Org 1", slug: "test-org-1-" + Date.now() },
{ id: org2Id, title: "Test Org 2", slug: "test-org-2-" + Date.now() },
],
});
// Set both orgs to use the same ClickHouse URL
await setOrganizationClickhouseUrl(org1Id, "standard", testClickhouseUrl);
await setOrganizationClickhouseUrl(org2Id, "standard", testClickhouseUrl);
// Get clients for both orgs
const client1 = await getClickhouseForOrganization(org1Id, "standard");
const client2 = await getClickhouseForOrganization(org2Id, "standard");
// Both should be defined
expect(client1).toBeDefined();
expect(client2).toBeDefined();
// They should be the same instance (cached by hostname)
expect(client1).toBe(client2);
// Clean up
await removeOrganizationClickhouseUrl(org1Id, "standard");
await removeOrganizationClickhouseUrl(org2Id, "standard");
await prisma.organization.deleteMany({
where: { id: { in: [org1Id, org2Id] } },
});
});
it("should clear cache when organization config changes", async () => {
// Create test organization
await prisma.organization.create({
data: {
id: testOrgId,
title: "Test Org",
slug: "test-org-" + Date.now(),
},
});
// Set URL
await setOrganizationClickhouseUrl(testOrgId, "standard", testClickhouseUrl);
// Get client to populate cache
const client1 = await getClickhouseForOrganization(testOrgId, "standard");
// Clear cache
clearClickhouseCacheForOrganization(testOrgId);
// Get client again (should hit the database again, not cache)
const client2 = await getClickhouseForOrganization(testOrgId, "standard");
// Both should be defined
expect(client1).toBeDefined();
expect(client2).toBeDefined();
// Clean up
await removeOrganizationClickhouseUrl(testOrgId, "standard");
await prisma.organization.delete({ where: { id: testOrgId } });
});
});
apps/webapp/app/services/clickhouse/clickhouseSecretSchemas.server.ts
+11
View File
@@ -0,0 +1,11 @@
import { z } from "zod";
export const ClickhouseConnectionSchema = z.object({
url: z.string().url(),
});
export type ClickhouseConnection = z.infer<typeof ClickhouseConnectionSchema>;
export function getClickhouseSecretKey(orgId: string, clientType: string): string {
return `org:${orgId}:clickhouse:${clientType}`;
}
apps/webapp/app/services/clickhouseInstance.server.ts
-130
View File
@@ -1,130 +0,0 @@
import { ClickHouse } from "@internal/clickhouse";
import { env } from "~/env.server";
import { singleton } from "~/utils/singleton";
export const clickhouseClient = singleton("clickhouseClient", initializeClickhouseClient);
function initializeClickhouseClient() {
const url = new URL(env.CLICKHOUSE_URL);
// Remove secure param
url.searchParams.delete("secure");
console.log(`🗃️ Clickhouse service enabled to host ${url.host}`);
const clickhouse = new ClickHouse({
url: url.toString(),
name: "clickhouse-instance",
keepAlive: {
enabled: env.CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
idleSocketTtl: env.CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
},
logLevel: env.CLICKHOUSE_LOG_LEVEL,
compression: {
request: true,
},
maxOpenConnections: env.CLICKHOUSE_MAX_OPEN_CONNECTIONS,
});
return clickhouse;
}
export const logsClickhouseClient = singleton(
"logsClickhouseClient",
initializeLogsClickhouseClient
);
function initializeLogsClickhouseClient() {
if (!env.LOGS_CLICKHOUSE_URL) {
throw new Error("LOGS_CLICKHOUSE_URL is not set");
}
const url = new URL(env.LOGS_CLICKHOUSE_URL);
// Remove secure param
url.searchParams.delete("secure");
return new ClickHouse({
url: url.toString(),
name: "logs-clickhouse",
keepAlive: {
enabled: env.CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
idleSocketTtl: env.CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
},
logLevel: env.CLICKHOUSE_LOG_LEVEL,
compression: {
request: true,
},
maxOpenConnections: env.CLICKHOUSE_MAX_OPEN_CONNECTIONS,
clickhouseSettings: {
max_memory_usage: env.CLICKHOUSE_LOGS_LIST_MAX_MEMORY_USAGE.toString(),
max_bytes_before_external_sort:
env.CLICKHOUSE_LOGS_LIST_MAX_BYTES_BEFORE_EXTERNAL_SORT.toString(),
max_threads: env.CLICKHOUSE_LOGS_LIST_MAX_THREADS,
...(env.CLICKHOUSE_LOGS_LIST_MAX_ROWS_TO_READ && {
max_rows_to_read: env.CLICKHOUSE_LOGS_LIST_MAX_ROWS_TO_READ.toString(),
}),
...(env.CLICKHOUSE_LOGS_LIST_MAX_EXECUTION_TIME && {
max_execution_time: env.CLICKHOUSE_LOGS_LIST_MAX_EXECUTION_TIME,
}),
},
});
}
export const adminClickhouseClient = singleton(
"adminClickhouseClient",
initializeAdminClickhouseClient
);
function initializeAdminClickhouseClient() {
if (!env.ADMIN_CLICKHOUSE_URL) {
throw new Error("ADMIN_CLICKHOUSE_URL is not set");
}
const url = new URL(env.ADMIN_CLICKHOUSE_URL);
url.searchParams.delete("secure");
return new ClickHouse({
url: url.toString(),
name: "admin-clickhouse",
keepAlive: {
enabled: env.CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
idleSocketTtl: env.CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
},
logLevel: env.CLICKHOUSE_LOG_LEVEL,
compression: {
request: true,
},
maxOpenConnections: env.CLICKHOUSE_MAX_OPEN_CONNECTIONS,
});
}
export const queryClickhouseClient = singleton(
"queryClickhouseClient",
initializeQueryClickhouseClient
);
function initializeQueryClickhouseClient() {
if (!env.QUERY_CLICKHOUSE_URL) {
throw new Error("QUERY_CLICKHOUSE_URL is not set");
}
const url = new URL(env.QUERY_CLICKHOUSE_URL);
// Remove secure param
url.searchParams.delete("secure");
return new ClickHouse({
url: url.toString(),
name: "query-clickhouse",
keepAlive: {
enabled: env.CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
idleSocketTtl: env.CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
},
logLevel: env.CLICKHOUSE_LOG_LEVEL,
compression: {
request: true,
},
maxOpenConnections: env.CLICKHOUSE_MAX_OPEN_CONNECTIONS,
});
}
apps/webapp/app/services/queryService.server.ts
+3 -2
View File
@@ -11,7 +11,7 @@ import type { TableSchema, WhereClauseCondition } from "@internal/tsql";
import { z } from "zod";
import { prisma } from "~/db.server";
import { env } from "~/env.server";
import { queryClickhouseClient } from "./clickhouseInstance.server";
import { getClickhouseForOrganization } from "./clickhouse/clickhouseFactory.server";
import {
queryConcurrencyLimiter,
DEFAULT_ORG_CONCURRENCY_LIMIT,
@@ -275,7 +275,8 @@ export async function executeQuery<TOut extends z.ZodSchema>(
environment: Object.fromEntries(environments.map((e) => [e.id, e.slug])),
};
const result = await executeTSQL(queryClickhouseClient.reader, {
const queryClickhouse = await getClickhouseForOrganization(organizationId, "query");
const result = await executeTSQL(queryClickhouse.reader, {
...baseOptions,
schema: z.record(z.any()),
tableSchema: querySchemas,
apps/webapp/app/services/runsReplicationService.server.ts
+90 -17
View File
@@ -617,18 +617,65 @@ export class RunsReplicationService {
payloadInserts: payloadInserts.length,
});
// Group task runs by organization for routing to correct ClickHouse instance
const taskRunsByOrg = new Map<string, TaskRunInsertArray[]>();
for (const taskRun of taskRunInserts) {
const orgId = getTaskRunField(taskRun, "organization_id");
const orgRuns = taskRunsByOrg.get(orgId) || [];
orgRuns.push(taskRun);
taskRunsByOrg.set(orgId, orgRuns);
}
// Group payloads by organization (extract from run_id -> task runs mapping)
const payloadsByOrg = new Map<string, PayloadInsertArray[]>();
for (const payload of payloadInserts) {
const runId = getPayloadField(payload, "run_id");
// Find the corresponding task run to get its organization
const taskRun = taskRunInserts.find((tr) => getTaskRunField(tr, "run_id") === runId);
if (taskRun) {
const orgId = getTaskRunField(taskRun, "organization_id");
const orgPayloads = payloadsByOrg.get(orgId) || [];
orgPayloads.push(payload);
payloadsByOrg.set(orgId, orgPayloads);
}
}
// Insert task runs and payloads with retry logic for connection errors
const [taskRunError, taskRunResult] = await this.#insertWithRetry(
(attempt) => this.#insertTaskRunInserts(taskRunInserts, attempt),
"task run inserts",
flushId
// Process each organization's data in parallel
const insertPromises = Array.from(taskRunsByOrg.entries()).map(
async ([orgId, orgTaskRuns]) => {
const orgPayloads = payloadsByOrg.get(orgId) || [];
const [taskRunError, taskRunResult] = await this.#insertWithRetry(
(attempt) => this.#insertTaskRunInserts(orgId, orgTaskRuns, attempt),
"task run inserts",
flushId
);
const [payloadError, payloadResult] = await this.#insertWithRetry(
(attempt) => this.#insertPayloadInserts(orgId, orgPayloads, attempt),
"payload inserts",
flushId
);
return { taskRunError, payloadError, orgId };
}
);
const [payloadError, payloadResult] = await this.#insertWithRetry(
(attempt) => this.#insertPayloadInserts(payloadInserts, attempt),
"payload inserts",
flushId
);
const results = await Promise.all(insertPromises);
// Aggregate errors from all organizations
let taskRunError: Error | null = null;
let payloadError: Error | null = null;
for (const result of results) {
if (result.taskRunError) {
taskRunError = result.taskRunError;
}
if (result.payloadError) {
payloadError = result.payloadError;
}
}
// Log any errors that occurred
if (taskRunError) {
@@ -770,19 +817,32 @@ export class RunsReplicationService {
};
}
async #insertTaskRunInserts(taskRunInserts: TaskRunInsertArray[], attempt: number) {
async #insertTaskRunInserts(
organizationId: string,
taskRunInserts: TaskRunInsertArray[],
attempt: number
) {
return await startSpan(this._tracer, "insertTaskRunsInserts", async (span) => {
const [insertError, insertResult] =
await this.options.clickhouse.taskRuns.insertCompactArrays(taskRunInserts, {
// Get the appropriate ClickHouse client for this organization
const { getClickhouseForOrganization } = await import(
"~/services/clickhouse/clickhouseFactory.server"
);
const clickhouse = await getClickhouseForOrganization(organizationId, "replication");
const [insertError, insertResult] = await clickhouse.taskRuns.insertCompactArrays(
taskRunInserts,
{
params: {
clickhouse_settings: this.#getClickhouseInsertSettings(),
},
});
}
);
if (insertError) {
this.logger.error("Error inserting task run inserts attempt", {
error: insertError,
attempt,
organizationId,
});
recordSpanError(span, insertError);
@@ -793,19 +853,32 @@ export class RunsReplicationService {
});
}
async #insertPayloadInserts(payloadInserts: PayloadInsertArray[], attempt: number) {
async #insertPayloadInserts(
organizationId: string,
payloadInserts: PayloadInsertArray[],
attempt: number
) {
return await startSpan(this._tracer, "insertPayloadInserts", async (span) => {
const [insertError, insertResult] =
await this.options.clickhouse.taskRuns.insertPayloadsCompactArrays(payloadInserts, {
// Get the appropriate ClickHouse client for this organization
const { getClickhouseForOrganization } = await import(
"~/services/clickhouse/clickhouseFactory.server"
);
const clickhouse = await getClickhouseForOrganization(organizationId, "replication");
const [insertError, insertResult] = await clickhouse.taskRuns.insertPayloadsCompactArrays(
payloadInserts,
{
params: {
clickhouse_settings: this.#getClickhouseInsertSettings(),
},
});
}
);
if (insertError) {
this.logger.error("Error inserting payload inserts attempt", {
error: insertError,
attempt,
organizationId,
});
recordSpanError(span, insertError);
apps/webapp/app/v3/otlpExporter.server.ts
+37 -10
View File
@@ -20,7 +20,6 @@ import {
} from "@trigger.dev/otlp-importer";
import type { MetricsV1Input } from "@internal/clickhouse";
import { logger } from "~/services/logger.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import { DynamicFlushScheduler } from "./dynamicFlushScheduler.server";
import { ClickhouseEventRepository } from "./eventRepository/clickhouseEventRepository.server";
import {
@@ -118,21 +117,26 @@ class OTLPExporter {
async #exportEvents(
eventsWithStores: { events: Array<CreateEventInput>; taskEventStore: string }[]
) {
const eventsGroupedByStore = eventsWithStores.reduce((acc, { events, taskEventStore }) => {
acc[taskEventStore] = acc[taskEventStore] || [];
acc[taskEventStore].push(...events);
// Group events by both store and organization for proper routing
const eventsGroupedByStoreAndOrg = eventsWithStores.reduce((acc, { events, taskEventStore }) => {
for (const event of events) {
const orgId = event.organizationId || "default";
const key = `${taskEventStore}:${orgId}`;
acc[key] = acc[key] || { store: taskEventStore, orgId, events: [] };
acc[key].events.push(event);
}
return acc;
}, {} as Record<string, Array<CreateEventInput>>);
}, {} as Record<string, { store: string; orgId: string; events: Array<CreateEventInput> }>);
let eventCount = 0;
for (const [store, events] of Object.entries(eventsGroupedByStore)) {
const eventRepository = this.#getEventRepositoryForStore(store);
for (const { store, orgId, events } of Object.values(eventsGroupedByStoreAndOrg)) {
const eventRepository = await this.#getEventRepositoryForStoreAndOrg(store, orgId);
await waitForLlmPricingReady();
const enrichedEvents = enrichCreatableEvents(events);
this.#logEventsVerbose(enrichedEvents, `exportEvents ${store}`);
this.#logEventsVerbose(enrichedEvents, `exportEvents ${store}:${orgId}`);
eventCount += enrichedEvents.length;
@@ -142,6 +146,19 @@ class OTLPExporter {
return eventCount;
}
async #getEventRepositoryForStoreAndOrg(store: string, orgId: string): Promise<IEventRepository> {
// For ClickHouse stores with a specific org (not "default"), use org-specific repository
if ((store === "clickhouse" || store === "clickhouse_v2") && orgId !== "default") {
const { getEventRepositoryForOrganization } = await import(
"~/services/clickhouse/clickhouseFactory.server"
);
return await getEventRepositoryForOrganization(orgId);
}
// Fall back to default repositories for non-ClickHouse stores or default org
return this.#getEventRepositoryForStore(store);
}
#getEventRepositoryForStore(store: string): IEventRepository {
if (store === "clickhouse") {
return this._clickhouseEventRepository;
@@ -1172,12 +1189,22 @@ function hasUnpairedSurrogateAtEnd(str: string): boolean {
export const otlpExporter = singleton("otlpExporter", initializeOTLPExporter);
function initializeOTLPExporter() {
async function initializeOTLPExporter() {
// Metrics are written globally (not per-org), use standard clickhouse
// We use a dummy org ID since metrics table is global
const { getClickhouseForOrganization } = await import(
"~/services/clickhouse/clickhouseFactory.server"
);
// Use a sentinel org ID for global metrics writes
// In practice, all orgs currently share the same metrics table/instance
const metricsClickhouse = await getClickhouseForOrganization("METRICS_GLOBAL", "standard");
const metricsFlushScheduler = new DynamicFlushScheduler<MetricsV1Input>({
batchSize: env.METRICS_CLICKHOUSE_BATCH_SIZE,
flushInterval: env.METRICS_CLICKHOUSE_FLUSH_INTERVAL_MS,
callback: async (_flushId, batch) => {
await clickhouseClient.metrics.insert(batch);
await metricsClickhouse.metrics.insert(batch);
},
minConcurrency: 1,
maxConcurrency: env.METRICS_CLICKHOUSE_MAX_CONCURRENCY,
apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts
+5 -3
View File
@@ -7,7 +7,7 @@ import {
} from "@trigger.dev/database";
import { getRunFiltersFromRequest } from "~/presenters/RunFilters.server";
import { type CreateBulkActionPayload } 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 {
parseRunListInputOptions,
type RunListInputFilters,
@@ -38,8 +38,9 @@ export class BulkActionService extends BaseService {
const filters = await getFilters(payload, request);
// Count the runs that will be affected by the bulk action
const clickhouse = await getClickhouseForOrganization(organizationId, "standard");
const runsRepository = new RunsRepository({
clickhouse: clickhouseClient,
clickhouse,
prisma: this._replica as PrismaClient,
});
const count = await runsRepository.countRuns({
@@ -147,8 +148,9 @@ export class BulkActionService extends BaseService {
...rawParams,
});
const clickhouse = await getClickhouseForOrganization(group.project.organizationId, "standard");
const runsRepository = new RunsRepository({
clickhouse: clickhouseClient,
clickhouse,
prisma: this._replica as PrismaClient,
});