diff --git a/.cursor/mcp.json b/.cursor/mcp.json index da39e4ffa..c4b06a676 100644 --- a/.cursor/mcp.json +++ b/.cursor/mcp.json @@ -1,3 +1,7 @@ { - "mcpServers": {} + "mcpServers": { + "linear": { + "url": "https://mcp.linear.app/mcp" + } + } } diff --git a/.server-changes/organization-scoped-clickhouse.md b/.server-changes/organization-scoped-clickhouse.md new file mode 100644 index 000000000..874b9dc60 --- /dev/null +++ b/.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 diff --git a/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts b/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts index 254ec18d1..b0ba01b9d 100644 --- a/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts +++ b/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 }); diff --git a/apps/webapp/app/presenters/v3/CreateBulkActionPresenter.server.ts b/apps/webapp/app/presenters/v3/CreateBulkActionPresenter.server.ts index acf511f0f..5e8bfc405 100644 --- a/apps/webapp/app/presenters/v3/CreateBulkActionPresenter.server.ts +++ b/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, }); diff --git a/apps/webapp/app/presenters/v3/RunTagListPresenter.server.ts b/apps/webapp/app/presenters/v3/RunTagListPresenter.server.ts index e9de368ec..89b9c8b41 100644 --- a/apps/webapp/app/presenters/v3/RunTagListPresenter.server.ts +++ b/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, }); diff --git a/apps/webapp/app/presenters/v3/TaskListPresenter.server.ts b/apps/webapp/app/presenters/v3/TaskListPresenter.server.ts index f1635f233..a6471c30c 100644 --- a/apps/webapp/app/presenters/v3/TaskListPresenter.server.ts +++ b/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,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); } diff --git a/apps/webapp/app/presenters/v3/UsagePresenter.server.ts b/apps/webapp/app/presenters/v3/UsagePresenter.server.ts index 2fac95617..c4654e870 100644 --- a/apps/webapp/app/presenters/v3/UsagePresenter.server.ts +++ b/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, diff --git a/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts b/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts index f0e955fd0..52ebad96b 100644 --- a/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts +++ b/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, diff --git a/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts b/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts index 9abcdf322..15eaef0d1 100644 --- a/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts +++ b/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, diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboards.$dashboardKey/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboards.$dashboardKey/route.tsx index cd358b7e6..9bbd1c04b 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboards.$dashboardKey/route.tsx +++ b/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) diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors._index/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors._index/route.tsx index e92b5b346..35f813118 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors._index/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors._index/route.tsx @@ -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, { diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsx index af3cc30a2..bccb7125d 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsx @@ -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, { diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.$modelId/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.$modelId/route.tsx index 7a25f996d..4256c64d4 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.$modelId/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.$modelId/route.tsx @@ -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) { diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models._index/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models._index/route.tsx index 7bf257f98..dca4ca948 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models._index/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models._index/route.tsx @@ -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(); diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.compare/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.compare/route.tsx index 661fb2942..879dcf47e 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.compare/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.compare/route.tsx @@ -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); diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug/route.tsx index 5a953c019..f37e8d3fe 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug/route.tsx @@ -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>["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() }), diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts._index/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts._index/route.tsx index 02c7cc444..4e229a48f 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts._index/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts._index/route.tsx @@ -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( diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam/route.tsx index f35376a42..422f5367e 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam/route.tsx @@ -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, diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx index ca7e8b7b0..ba3cd4b83 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx @@ -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, diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.test.tasks.$taskParam/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.test.tasks.$taskParam/route.tsx index ee69419e1..38356c6a2 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.test.tasks.$taskParam/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.test.tasks.$taskParam/route.tsx @@ -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({ diff --git a/apps/webapp/app/routes/api.v1.prompts.$slug.ts b/apps/webapp/app/routes/api.v1.prompts.$slug.ts index 32ea1525c..230ceb127 100644 --- a/apps/webapp/app/routes/api.v1.prompts.$slug.ts +++ b/apps/webapp/app/routes/api.v1.prompts.$slug.ts @@ -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, diff --git a/apps/webapp/app/routes/api.v1.prompts.$slug.versions.ts b/apps/webapp/app/routes/api.v1.prompts.$slug.versions.ts index c40b3e62d..17b88b12c 100644 --- a/apps/webapp/app/routes/api.v1.prompts.$slug.versions.ts +++ b/apps/webapp/app/routes/api.v1.prompts.$slug.versions.ts @@ -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({ diff --git a/apps/webapp/app/routes/api.v1.prompts._index.ts b/apps/webapp/app/routes/api.v1.prompts._index.ts index ccbc0ec38..44f2f86d0 100644 --- a/apps/webapp/app/routes/api.v1.prompts._index.ts +++ b/apps/webapp/app/routes/api.v1.prompts._index.ts @@ -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 diff --git a/apps/webapp/app/routes/otel.v1.logs.ts b/apps/webapp/app/routes/otel.v1.logs.ts index a05ddd24c..1dc7c07c1 100644 --- a/apps/webapp/app/routes/otel.v1.logs.ts +++ b/apps/webapp/app/routes/otel.v1.logs.ts @@ -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, diff --git a/apps/webapp/app/routes/otel.v1.metrics.ts b/apps/webapp/app/routes/otel.v1.metrics.ts index 5529f9310..9a09cb182 100644 --- a/apps/webapp/app/routes/otel.v1.metrics.ts +++ b/apps/webapp/app/routes/otel.v1.metrics.ts @@ -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, diff --git a/apps/webapp/app/routes/otel.v1.traces.ts b/apps/webapp/app/routes/otel.v1.traces.ts index 609b72c04..8e974c7b1 100644 --- a/apps/webapp/app/routes/otel.v1.traces.ts +++ b/apps/webapp/app/routes/otel.v1.traces.ts @@ -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, diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsx index f862ced6b..0e0469bcd 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsx +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsx @@ -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 { diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.ts index 66ddebe4e..d55c74962 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.ts +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.ts @@ -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({ diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug.generations.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug.generations.ts index 77a55ec3f..17a11e058 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug.generations.ts +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug.generations.ts @@ -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, diff --git a/apps/webapp/app/services/admin/missingLlmModels.server.ts b/apps/webapp/app/services/admin/missingLlmModels.server.ts index 7ce6bc2ab..07e6160ee 100644 --- a/apps/webapp/app/services/admin/missingLlmModels.server.ts +++ b/apps/webapp/app/services/admin/missingLlmModels.server.ts @@ -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({ + const adminClickhouse = getAdminClickhouse(); + + const createBuilder = adminClickhouse.reader.queryBuilderFast({ name: "missingModelSamples", table: "trigger_dev.task_events_v2", columns: [ diff --git a/apps/webapp/app/services/clickhouse/clickhouseCredentialsService.server.ts b/apps/webapp/app/services/clickhouse/clickhouseCredentialsService.server.ts new file mode 100644 index 000000000..c2c8c77f7 --- /dev/null +++ b/apps/webapp/app/services/clickhouse/clickhouseCredentialsService.server.ts @@ -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 { + // 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 { + // 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 { + 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; +} diff --git a/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts b/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts new file mode 100644 index 000000000..944988687 --- /dev/null +++ b/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts @@ -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(); +const eventRepositoryCache = new Map(); + +// 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(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 { + 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 { + // 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 { + // 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 { + 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 +} diff --git a/apps/webapp/app/services/clickhouse/clickhouseFactory.test.ts b/apps/webapp/app/services/clickhouse/clickhouseFactory.test.ts new file mode 100644 index 000000000..f0b24b941 --- /dev/null +++ b/apps/webapp/app/services/clickhouse/clickhouseFactory.test.ts @@ -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 } }); + }); +}); diff --git a/apps/webapp/app/services/clickhouse/clickhouseSecretSchemas.server.ts b/apps/webapp/app/services/clickhouse/clickhouseSecretSchemas.server.ts new file mode 100644 index 000000000..016eb717c --- /dev/null +++ b/apps/webapp/app/services/clickhouse/clickhouseSecretSchemas.server.ts @@ -0,0 +1,11 @@ +import { z } from "zod"; + +export const ClickhouseConnectionSchema = z.object({ + url: z.string().url(), +}); + +export type ClickhouseConnection = z.infer; + +export function getClickhouseSecretKey(orgId: string, clientType: string): string { + return `org:${orgId}:clickhouse:${clientType}`; +} diff --git a/apps/webapp/app/services/clickhouseInstance.server.ts b/apps/webapp/app/services/clickhouseInstance.server.ts deleted file mode 100644 index 9c4941671..000000000 --- a/apps/webapp/app/services/clickhouseInstance.server.ts +++ /dev/null @@ -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, - }); -} diff --git a/apps/webapp/app/services/queryService.server.ts b/apps/webapp/app/services/queryService.server.ts index 1f3bdbba1..f24df9eb0 100644 --- a/apps/webapp/app/services/queryService.server.ts +++ b/apps/webapp/app/services/queryService.server.ts @@ -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( 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, diff --git a/apps/webapp/app/services/runsReplicationService.server.ts b/apps/webapp/app/services/runsReplicationService.server.ts index 7930c0548..ca1fba686 100644 --- a/apps/webapp/app/services/runsReplicationService.server.ts +++ b/apps/webapp/app/services/runsReplicationService.server.ts @@ -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(); + 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(); + 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); diff --git a/apps/webapp/app/v3/otlpExporter.server.ts b/apps/webapp/app/v3/otlpExporter.server.ts index 7505693e3..f8b22d7c4 100644 --- a/apps/webapp/app/v3/otlpExporter.server.ts +++ b/apps/webapp/app/v3/otlpExporter.server.ts @@ -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; 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>); + }, {} as Record }>); 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 { + // 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({ 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, diff --git a/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts b/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts index 156b68bff..07a428629 100644 --- a/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts +++ b/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts @@ -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, });