From 5700df8b5c064365f417efbb4ea24387fd48cce1 Mon Sep 17 00:00:00 2001 From: Matt Aitken Date: Thu, 2 Apr 2026 18:48:06 +0100 Subject: [PATCH] Use the clickhouseFactory directly. WIP on new event repository --- .../v3/ApiRunListPresenter.server.ts | 4 +- .../v3/CreateBulkActionPresenter.server.ts | 4 +- .../v3/RunTagListPresenter.server.ts | 4 +- .../presenters/v3/TaskListPresenter.server.ts | 4 +- .../presenters/v3/UsagePresenter.server.ts | 4 +- .../v3/ViewSchedulePresenter.server.ts | 4 +- .../v3/WaitpointPresenter.server.ts | 4 +- .../route.tsx | 4 +- .../route.tsx | 8 +- .../route.tsx | 4 +- .../route.tsx | 4 +- .../route.tsx | 4 +- .../route.tsx | 4 +- .../route.tsx | 4 +- .../route.tsx | 4 +- .../route.tsx | 4 +- .../route.tsx | 4 +- .../route.tsx | 4 +- .../route.tsx | 4 +- .../webapp/app/routes/api.v1.prompts.$slug.ts | 6 +- .../routes/api.v1.prompts.$slug.versions.ts | 4 +- .../app/routes/api.v1.prompts._index.ts | 4 +- ...projectParam.env.$envParam.logs.$logId.tsx | 4 +- ...ojects.$projectParam.env.$envParam.logs.ts | 4 +- ...nvParam.prompts.$promptSlug.generations.ts | 4 +- .../runEngine/concerns/traceEvents.server.ts | 22 +-- .../services/triggerFailedTask.server.ts | 39 ++-- .../clickhouse/clickhouseFactory.server.ts | 176 +++++++++--------- .../app/services/queryService.server.ts | 4 +- .../services/runsReplicationService.server.ts | 6 +- .../clickhouseEventRepository.server.ts | 3 +- ...lickhouseEventRepositoryInstance.server.ts | 112 ----------- .../eventRepository/eventRepository.server.ts | 2 +- .../eventRepository/eventRepository.types.ts | 2 +- .../app/v3/eventRepository/index.server.ts | 89 ++++----- apps/webapp/app/v3/otlpExporter.server.ts | 153 ++++++++------- .../alerts/errorAlertEvaluator.server.ts | 4 +- .../v3/services/bulk/BulkActionV2.server.ts | 6 +- 38 files changed, 314 insertions(+), 410 deletions(-) delete mode 100644 apps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.ts diff --git a/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts b/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts index b0ba01b9d..357d66501 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } 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,7 @@ export class ApiRunListPresenter extends BasePresenter { options.machines = searchParams["filter[machine]"]; } - const clickhouse = await getClickhouseForOrganization(organizationId, "standard"); + const clickhouse = await clickhouseFactory.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 5e8bfc405..eeb5b3d87 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { RunsRepository } from "~/services/runsRepository/runsRepository.server"; import { getRunFiltersFromRequest } from "../RunFilters.server"; import { BasePresenter } from "./basePresenter.server"; @@ -24,7 +24,7 @@ export class CreateBulkActionPresenter extends BasePresenter { Object.fromEntries(new URL(request.url).searchParams) ); - const clickhouse = await getClickhouseForOrganization(organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(organizationId, "standard"); const runsRepository = new RunsRepository({ 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 89b9c8b41..44d6f2a07 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { type PrismaClient } from "@trigger.dev/database"; import { timeFilters } from "~/components/runs/v3/SharedFilters"; @@ -37,7 +37,7 @@ export class RunTagListPresenter extends BasePresenter { }: TagListOptions) { const hasFilters = Boolean(name?.trim()); - const clickhouse = await getClickhouseForOrganization(organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(organizationId, "standard"); const runsRepository = new RunsRepository({ 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 a6471c30c..8a41d0ef6 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { type AverageDurations, ClickHouseEnvironmentMetricsRepository, @@ -74,7 +74,7 @@ export class TaskListPresenter { const slugs = tasks.map((t) => t.slug); // Create org-specific environment metrics repository - const clickhouse = await getClickhouseForOrganization(organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(organizationId, "standard"); const environmentMetricsRepository = new ClickHouseEnvironmentMetricsRepository({ clickhouse, }); diff --git a/apps/webapp/app/presenters/v3/UsagePresenter.server.ts b/apps/webapp/app/presenters/v3/UsagePresenter.server.ts index c4654e870..d312088b6 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; type Options = { organizationId: string; @@ -124,7 +124,7 @@ async function getTaskUsageByOrganization( endOfMonth: Date, replica: PrismaClientOrTransaction ) { - const clickhouse = await getClickhouseForOrganization(organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(organizationId, "standard"); const [queryError, tasks] = await clickhouse.taskRuns.getTaskUsageByOrganization({ startTime: startOfMonth.getTime(), endTime: endOfMonth.getTime(), diff --git a/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts b/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts index 52ebad96b..5341568e6 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } 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,7 @@ export class ViewSchedulePresenter { ? nextScheduledTimestamps(schedule.generatorExpression, schedule.timezone, new Date(), 5) : []; - const clickhouse = await getClickhouseForOrganization(schedule.project.organizationId, "standard"); + const clickhouse = await clickhouseFactory.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, diff --git a/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts b/apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts index 15eaef0d1..dc9bf3d1e 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } 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,7 @@ export class WaitpointPresenter extends BasePresenter { const connectedRuns: NextRunListItem[] = []; if (connectedRunIds.length > 0) { - const clickhouse = await getClickhouseForOrganization(waitpoint.environment.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(waitpoint.environment.organizationId, "standard"); const runPresenter = new NextRunListPresenter(this._prisma, clickhouse); const { runs } = await runPresenter.call( waitpoint.environment.organizationId, 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 9bbd1c04b..f7866565c 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { requireUser } from "~/services/session.server"; import { cn } from "~/utils/cn"; import { EnvironmentParamSchema } from "~/utils/pathBuilder"; @@ -74,7 +74,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { const filters = dashboard.filters ?? ["tasks", "queues"]; - const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "standard"); // Load distinct models from ClickHouse if the dashboard has a models filter let possibleModels: { model: string; system: string }[] = []; diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors.$fingerprint/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors.$fingerprint/route.tsx index f42c73b5e..bdc0d3588 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors.$fingerprint/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors.$fingerprint/route.tsx @@ -26,7 +26,13 @@ import { } from "~/presenters/v3/ErrorGroupPresenter.server"; import { type NextRunList } from "~/presenters/v3/NextRunListPresenter.server"; import { $replica } from "~/db.server"; -import { logsClickhouseClient, clickhouseClient } from "~/services/clickhouseInstance.server"; +import { + getDefaultClickhouseClient, + getDefaultLogsClickhouseClient, +} from "~/services/clickhouse/clickhouseFactory.server"; + +const clickhouseClient = getDefaultClickhouseClient(); +const logsClickhouseClient = getDefaultLogsClickhouseClient(); import { NavBar, PageTitle } from "~/components/primitives/PageHeader"; import { PageBody } from "~/components/layout/AppLayout"; import { 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 35f813118..38a69cd0f 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } 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,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { const plan = await getCurrentPlan(project.organizationId); const retentionLimitDays = plan?.v3Subscription?.plan?.limits.logRetentionDays.number ?? 30; - const logsClickhouse = await getClickhouseForOrganization(project.organizationId, "logs"); + const logsClickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "logs"); const presenter = new ErrorsListPresenter($replica, logsClickhouse); const listPromise = presenter 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 bccb7125d..02401f043 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } 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,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { const plan = await getCurrentPlan(project.organizationId); const retentionLimitDays = plan?.v3Subscription?.plan?.limits.logRetentionDays.number ?? 30; - const logsClickhouse = await getClickhouseForOrganization(project.organizationId, "logs"); + const logsClickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "logs"); const presenter = new LogsListPresenter($replica, logsClickhouse); const listPromise = presenter 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 4256c64d4..813879cf2 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { requireUserId } from "~/services/session.server"; import { useOrganization } from "~/hooks/useOrganizations"; import { useProject } from "~/hooks/useProject"; @@ -68,7 +68,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { throw new Response("Environment not found", { status: 404 }); } - const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "standard"); const presenter = new ModelRegistryPresenter(clickhouse); const model = await presenter.getModelDetail(modelId); 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 dca4ca948..29c2e4f6c 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { requireUserId } from "~/services/session.server"; import { useEnvironment } from "~/hooks/useEnvironment"; import { useOrganization } from "~/hooks/useOrganizations"; @@ -109,7 +109,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { throw new Response("Environment not found", { status: 404 }); } - const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "standard"); const presenter = new ModelRegistryPresenter(clickhouse); const catalog = await presenter.getModelCatalog(); 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 879dcf47e..1306eb919 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { requireUserId } from "~/services/session.server"; import { useOrganization } from "~/hooks/useOrganizations"; import { useProject } from "~/hooks/useProject"; @@ -55,7 +55,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { return typedjson({ comparison: [] as ModelComparisonItem[], models: responseModels }); } - const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard"); + const clickhouse = await clickhouseFactory.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 f37e8d3fe..de43ea46b 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } 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,7 @@ 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 clickhouse = await getClickhouseForOrganization(project.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "standard"); const presenter = new PromptPresenter(clickhouse); let generations: Awaited>["generations"] = []; let generationsPagination: { next?: 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 4e229a48f..b44c5954f 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } 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,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { throw new Response("Environment not found", { status: 404 }); } - const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "standard"); const presenter = new PromptPresenter(clickhouse); const prompts = await presenter.listPrompts(project.id, environment.id); 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 422f5367e..2f4e79377 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } 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,7 @@ async function getRunsListFromTableState({ return null; } - const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "standard"); const runsListPresenter = new NextRunListPresenter($replica, clickhouse); const currentPageResult = await runsListPresenter.call(project.organizationId, environment.id, { userId, 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 ba3cd4b83..28a3ef46a 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { setRootOnlyFilterPreference, uiPreferencesStorage, @@ -87,7 +87,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { const filters = await getRunFiltersFromRequest(request); - const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "standard"); const presenter = new NextRunListPresenter($replica, clickhouse); const list = presenter.call(project.organizationId, environment.id, { userId, 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 38356c6a2..69ce432f4 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } 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,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { }); } - const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "standard"); const presenter = new TestTaskPresenter($replica, clickhouse); try { const [result, regionsResult] = await Promise.all([ diff --git a/apps/webapp/app/routes/api.v1.prompts.$slug.ts b/apps/webapp/app/routes/api.v1.prompts.$slug.ts index 230ceb127..919133d3c 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { createActionApiRoute, createLoaderApiRoute, @@ -53,7 +53,7 @@ export const loader = createLoaderApiRoute( return json({ error: "Prompt not found" }, { status: 404 }); } - const clickhouse = await getClickhouseForOrganization(prompt.project.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(prompt.project.organizationId, "standard"); const presenter = new PromptPresenter(clickhouse); const version = await presenter.resolveVersion(prompt.id, { version: searchParams.version, @@ -125,7 +125,7 @@ const { action } = createActionApiRoute( return json({ error: "Prompt not found" }, { status: 404 }); } - const clickhouse = await getClickhouseForOrganization(authentication.environment.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(authentication.environment.organizationId, "standard"); const presenter = new PromptPresenter(clickhouse); const version = await presenter.resolveVersion(prompt.id, { version: body.version, 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 17b88b12c..8f953d1a4 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server"; const ParamsSchema = z.object({ @@ -43,7 +43,7 @@ export const loader = createLoaderApiRoute( return json({ error: "Prompt not found" }, { status: 404 }); } - const clickhouse = await getClickhouseForOrganization(prompt.project.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(prompt.project.organizationId, "standard"); const presenter = new PromptPresenter(clickhouse); const versions = await presenter.listVersions(prompt.id); diff --git a/apps/webapp/app/routes/api.v1.prompts._index.ts b/apps/webapp/app/routes/api.v1.prompts._index.ts index 44f2f86d0..a6ad065c2 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server"; export const loader = createLoaderApiRoute( @@ -15,7 +15,7 @@ export const loader = createLoaderApiRoute( }, }, async ({ authentication }) => { - const clickhouse = await getClickhouseForOrganization(authentication.environment.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(authentication.environment.organizationId, "standard"); const presenter = new PromptPresenter(clickhouse); const prompts = await presenter.listPrompts( authentication.environment.projectId, 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 0e0469bcd..ae552e96e 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } 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,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { const [traceId, spanId, , startTime] = parts; - const logsClickhouse = await getClickhouseForOrganization(project.organizationId, "logs"); + const logsClickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "logs"); const presenter = new LogDetailPresenter($replica, logsClickhouse); let result; 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 d55c74962..7bb0db0d5 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { getCurrentPlan } from "~/services/platform.v3.server"; // Valid log levels for filtering @@ -69,7 +69,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { retentionLimitDays, }) as any; // Validated by LogsListOptionsSchema at runtime - const logsClickhouse = await getClickhouseForOrganization(project.organizationId, "logs"); + const logsClickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "logs"); const presenter = new LogsListPresenter($replica, logsClickhouse); const result = await presenter.call(project.organizationId, environment.id, options); 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 17a11e058..5188d8ccd 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { PromptPresenter, type GenerationRow, @@ -59,7 +59,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { const operations = url.searchParams.getAll("operations").filter(Boolean); const providers = url.searchParams.getAll("providers").filter(Boolean); - const clickhouse = await getClickhouseForOrganization(project.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "standard"); const presenter = new PromptPresenter(clickhouse); const result = await presenter.listGenerations({ environmentId: environment.id, diff --git a/apps/webapp/app/runEngine/concerns/traceEvents.server.ts b/apps/webapp/app/runEngine/concerns/traceEvents.server.ts index cb2eaa30a..8c9029e2d 100644 --- a/apps/webapp/app/runEngine/concerns/traceEvents.server.ts +++ b/apps/webapp/app/runEngine/concerns/traceEvents.server.ts @@ -10,6 +10,7 @@ export class DefaultTraceEventsConcern implements TraceEventConcern { parentStore: string | undefined ): Promise<{ repository: IEventRepository; store: string }> { return await getEventRepository( + request.environment.organization.id, request.environment.organization.featureFlags as Record, parentStore ); @@ -162,18 +163,15 @@ export class DefaultTraceEventsConcern implements TraceEventConcern { }, async (event, traceContext, traceparent) => { // Log a message about the debounced trigger - await repository.recordEvent( - `Debounced: using existing run with key "${debounceKey}"`, - { - taskSlug: request.taskId, - environment: request.environment, - attributes: { - runId: existingRun.friendlyId, - }, - context: request.options?.traceContext, - parentId: event.spanId, - } - ); + await repository.recordEvent(`Debounced: using existing run with key "${debounceKey}"`, { + taskSlug: request.taskId, + environment: request.environment, + attributes: { + runId: existingRun.friendlyId, + }, + context: request.options?.traceContext, + parentId: event.spanId, + }); return await callback( { diff --git a/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts b/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts index 0b59a523a..5f985b684 100644 --- a/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts +++ b/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts @@ -74,6 +74,7 @@ export class TriggerFailedTaskService { try { const { repository, store } = await getEventRepository( + request.environment.organization.id, request.environment.organization.featureFlags as Record, undefined ); @@ -81,11 +82,11 @@ export class TriggerFailedTaskService { // Resolve parent run for rootTaskRunId and depth (same as triggerTask.server.ts) const parentRun = request.parentRunId ? await this.prisma.taskRun.findFirst({ - where: { - id: RunId.fromFriendlyId(request.parentRunId), - runtimeEnvironmentId: request.environment.id, - }, - }) + where: { + id: RunId.fromFriendlyId(request.parentRunId), + runtimeEnvironmentId: request.environment.id, + }, + }) : undefined; const depth = parentRun ? parentRun.depth + 1 : 0; @@ -116,18 +117,18 @@ export class TriggerFailedTaskService { // resolveQueueProperties requires the worker to be passed when lockToVersion is present. const lockedToBackgroundWorker = bodyOptions?.lockToVersion ? await this.prisma.backgroundWorker.findFirst({ - where: { - projectId: request.environment.projectId, - runtimeEnvironmentId: request.environment.id, - version: bodyOptions.lockToVersion, - }, - select: { - id: true, - version: true, - sdkVersion: true, - cliVersion: true, - }, - }) + where: { + projectId: request.environment.projectId, + runtimeEnvironmentId: request.environment.id, + version: bodyOptions.lockToVersion, + }, + select: { + id: true, + version: true, + sdkVersion: true, + cliVersion: true, + }, + }) : undefined; const resolved = await queueConcern.resolveQueueProperties( @@ -273,9 +274,7 @@ export class TriggerFailedTaskService { }, taskIdentifier: opts.taskId, payload: - typeof opts.payload === "string" - ? opts.payload - : JSON.stringify(opts.payload ?? ""), + typeof opts.payload === "string" ? opts.payload : JSON.stringify(opts.payload ?? ""), payloadType: opts.payloadType ?? "application/json", error: { type: "INTERNAL_ERROR" as const, diff --git a/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts b/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts index f277bfca5..f87121597 100644 --- a/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts +++ b/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts @@ -1,41 +1,3 @@ -/** - * 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 data store overrides live in the `OrganizationDataStore` table - * - The config JSON stores a `secretKey` that references the SecretStore entry - * - No plaintext credentials in database - * - * ### Caching Strategy - * - **Org → data store mapping**: `OrganizationDataStoresRegistry` (in-memory Map, reloaded - * periodically via setInterval) - * - **ClickHouse clients**: cached by hostname hash (multiple orgs share same instance) - * - **Event repositories**: cached by hostname hash (stateful, must be reused) - * - * ## Usage in Presenters - * - * ```typescript - * import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; - * - * export class MyPresenter extends BasePresenter { - * async call({ organizationId, ... }) { - * const clickhouse = await getClickhouseForOrganization(organizationId, "standard"); - * } - * } - * ``` - * - * @module clickhouseFactory - */ - import { ClickHouse } from "@internal/clickhouse"; import { createHash } from "crypto"; import { ClickhouseEventRepository } from "~/v3/eventRepository/clickhouseEventRepository.server"; @@ -43,6 +5,7 @@ import { env } from "~/env.server"; import { singleton } from "~/utils/singleton"; import { organizationDataStoresRegistry } from "~/services/dataStores/organizationDataStoresRegistryInstance.server"; import type { OrganizationDataStoresRegistry } from "~/services/dataStores/organizationDataStoresRegistry.server"; +import { type IEventRepository } from "~/v3/eventRepository/eventRepository.types"; // --------------------------------------------------------------------------- // Default clients (singleton per process) @@ -199,6 +162,13 @@ export class ClickhouseFactory { constructor(private readonly _registry: OrganizationDataStoresRegistry) {} + async isReady(): Promise { + if (!this._registry.isLoaded) { + await this._registry.isReady; + } + return true; + } + async getClickhouseForOrganization( organizationId: string, clientType: ClientType @@ -207,6 +177,10 @@ export class ClickhouseFactory { await this._registry.isReady; } + return this.getClickhouseForOrganizationSync(organizationId, clientType); + } + + getClickhouseForOrganizationSync(organizationId: string, clientType: ClientType): ClickHouse { const dataStore = this._registry.get(organizationId, "CLICKHOUSE"); if (!dataStore) { @@ -237,36 +211,44 @@ export class ClickhouseFactory { } async getEventRepositoryForOrganization( + store: string, organizationId: string - ): Promise { + ): Promise<{ key: string; repository: IEventRepository }> { if (!this._registry.isLoaded) { await this._registry.isReady; } + return this.getEventRepositoryForOrganizationSync(store, organizationId); + } + + getEventRepositoryForOrganizationSync( + store: string, + organizationId: string + ): { key: string; repository: IEventRepository } { const dataStore = this._registry.get(organizationId, "CLICKHOUSE"); if (!dataStore) { - const defaultKey = "default:events"; + const defaultKey = `default:events:${store}`; let defaultRepo = this._eventRepositoryCache.get(defaultKey); if (!defaultRepo) { - const eventsClickhouse = await getEventsClickhouseClient(); - defaultRepo = buildEventRepository(eventsClickhouse); + const eventsClickhouse = getEventsClickhouseClient(); + defaultRepo = buildEventRepository(store, eventsClickhouse); this._eventRepositoryCache.set(defaultKey, defaultRepo); } - return defaultRepo; + return { key: defaultKey, repository: defaultRepo }; } const hostnameHash = hashHostname(dataStore.url); - const cacheKey = `${hostnameHash}:events`; + const cacheKey = `${hostnameHash}:events:${store}`; let repository = this._eventRepositoryCache.get(cacheKey); if (!repository) { - const client = await this.getClickhouseForOrganization(organizationId, "events"); - repository = buildEventRepository(client); + const client = this.getClickhouseForOrganizationSync(organizationId, "events"); + repository = buildEventRepository(store, client); this._eventRepositoryCache.set(cacheKey, repository); } - return repository; + return { key: cacheKey, repository: repository }; } } @@ -274,28 +256,11 @@ export class ClickhouseFactory { // Singleton factory instance // --------------------------------------------------------------------------- -const clickhouseFactory = singleton( +export const clickhouseFactory = singleton( "clickhouseFactory", () => new ClickhouseFactory(organizationDataStoresRegistry) ); -// --------------------------------------------------------------------------- -// Public API (thin wrappers around the singleton) -// --------------------------------------------------------------------------- - -export async function getClickhouseForOrganization( - organizationId: string, - clientType: ClientType -): Promise { - return clickhouseFactory.getClickhouseForOrganization(organizationId, clientType); -} - -export async function getEventRepositoryForOrganization( - organizationId: string -): Promise { - return clickhouseFactory.getEventRepositoryForOrganization(organizationId); -} - /** * Get admin ClickHouse client for cross-organization queries. * Only use for admin tools and analytics that need to query across all orgs. @@ -304,11 +269,19 @@ export function getAdminClickhouse(): ClickHouse { return defaultAdminClickhouseClient; } +export function getDefaultClickhouseClient(): ClickHouse { + return defaultClickhouseClient; +} + +export function getDefaultLogsClickhouseClient(): ClickHouse { + return defaultLogsClickhouseClient; +} + // --------------------------------------------------------------------------- // Private helpers // --------------------------------------------------------------------------- -async function getEventsClickhouseClient(): Promise { +function getEventsClickhouseClient(): ClickHouse { if (!env.EVENTS_CLICKHOUSE_URL) { throw new Error("EVENTS_CLICKHOUSE_URL is not set"); } @@ -324,29 +297,58 @@ async function getEventsClickhouseClient(): Promise { idleSocketTtl: env.EVENTS_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS, }, logLevel: env.EVENTS_CLICKHOUSE_LOG_LEVEL, - compression: { request: env.EVENTS_CLICKHOUSE_COMPRESSION_REQUEST === "1" }, + compression: { + request: env.EVENTS_CLICKHOUSE_COMPRESSION_REQUEST === "1", + }, maxOpenConnections: env.EVENTS_CLICKHOUSE_MAX_OPEN_CONNECTIONS, }); } -function buildEventRepository(clickhouse: ClickHouse): ClickhouseEventRepository { - return new ClickhouseEventRepository({ - clickhouse, - 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", - }); +function buildEventRepository(store: string, clickhouse: ClickHouse): ClickhouseEventRepository { + switch (store) { + case "clickhouse": { + return new ClickhouseEventRepository({ + clickhouse, + 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: "v1", + }); + } + case "clickhouse_v2": { + return new ClickhouseEventRepository({ + clickhouse: clickhouse, + 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, + 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", + }); + } + default: { + throw new Error(`Unknown ClickHouse event repository store: ${store}`); + } + } } diff --git a/apps/webapp/app/services/queryService.server.ts b/apps/webapp/app/services/queryService.server.ts index f24df9eb0..214232ebd 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 { getClickhouseForOrganization } from "./clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "./clickhouse/clickhouseFactory.server"; import { queryConcurrencyLimiter, DEFAULT_ORG_CONCURRENCY_LIMIT, @@ -275,7 +275,7 @@ export async function executeQuery( environment: Object.fromEntries(environments.map((e) => [e.id, e.slug])), }; - const queryClickhouse = await getClickhouseForOrganization(organizationId, "query"); + const queryClickhouse = await clickhouseFactory.getClickhouseForOrganization(organizationId, "query"); const result = await executeTSQL(queryClickhouse.reader, { ...baseOptions, schema: z.record(z.any()), diff --git a/apps/webapp/app/services/runsReplicationService.server.ts b/apps/webapp/app/services/runsReplicationService.server.ts index 78b7e084e..1a9ff69dc 100644 --- a/apps/webapp/app/services/runsReplicationService.server.ts +++ b/apps/webapp/app/services/runsReplicationService.server.ts @@ -29,7 +29,7 @@ import EventEmitter from "node:events"; import pLimit from "p-limit"; import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings"; import { calculateErrorFingerprint } from "~/utils/errorFingerprinting"; -import { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; interface TransactionEvent { tag: "insert" | "update" | "delete"; @@ -854,7 +854,7 @@ export class RunsReplicationService { ) { return await startSpan(this._tracer, "insertTaskRunsInserts", async (span) => { // Get the appropriate ClickHouse client for this organization - const clickhouse = await getClickhouseForOrganization(organizationId, "replication"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(organizationId, "replication"); const [insertError, insertResult] = await clickhouse.taskRuns.insertCompactArrays( taskRunInserts, @@ -887,7 +887,7 @@ export class RunsReplicationService { ) { return await startSpan(this._tracer, "insertPayloadInserts", async (span) => { // Get the appropriate ClickHouse client for this organization - const clickhouse = await getClickhouseForOrganization(organizationId, "replication"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(organizationId, "replication"); const [insertError, insertResult] = await clickhouse.taskRuns.insertPayloadsCompactArrays( payloadInserts, diff --git a/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts b/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts index 27b96ed7d..6c5204ca3 100644 --- a/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts +++ b/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts @@ -236,7 +236,6 @@ export class ClickhouseEventRepository implements IEventRepository { } async #flushLlmMetricsBatch(flushId: string, rows: LlmMetricsV1Input[]) { - const [insertError] = await this._clickhouse.llmMetrics.insert(rows, { params: { clickhouse_settings: this.#getClickhouseInsertSettings(), @@ -310,7 +309,7 @@ export class ClickhouseEventRepository implements IEventRepository { await tracePubSub.publish(events.map((e) => e.trace_id)); } - async insertMany(events: CreateEventInput[]): Promise { + insertMany(events: CreateEventInput[]): void { this.addToBatch(events.flatMap((event) => this.createEventToTaskEventV1Input(event))); // Dual-write LLM metrics records for spans with cost enrichment diff --git a/apps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.ts b/apps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.ts deleted file mode 100644 index d4e28c584..000000000 --- a/apps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.ts +++ /dev/null @@ -1,112 +0,0 @@ -import { ClickHouse } from "@internal/clickhouse"; -import { env } from "~/env.server"; -import { singleton } from "~/utils/singleton"; -import { ClickhouseEventRepository } from "./clickhouseEventRepository.server"; - -export const clickhouseEventRepository = singleton( - "clickhouseEventRepository", - initializeClickhouseRepository -); - -export const clickhouseEventRepositoryV2 = singleton( - "clickhouseEventRepositoryV2", - initializeClickhouseRepositoryV2 -); - -function getClickhouseClient() { - 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, - }); -} - -function initializeClickhouseRepository() { - if (!env.EVENTS_CLICKHOUSE_URL) { - throw new Error("EVENTS_CLICKHOUSE_URL is not set"); - } - - const url = new URL(env.EVENTS_CLICKHOUSE_URL); - url.searchParams.delete("secure"); - - const safeUrl = new URL(url.toString()); - safeUrl.password = "redacted"; - - console.log("🗃️ Initializing Clickhouse event repository (v1)", { url: safeUrl.toString() }); - - const clickhouse = getClickhouseClient(); - - const repository = new ClickhouseEventRepository({ - clickhouse: clickhouse, - 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: "v1", - }); - - return repository; -} - -function initializeClickhouseRepositoryV2() { - if (!env.EVENTS_CLICKHOUSE_URL) { - throw new Error("EVENTS_CLICKHOUSE_URL is not set"); - } - - const url = new URL(env.EVENTS_CLICKHOUSE_URL); - url.searchParams.delete("secure"); - - const safeUrl = new URL(url.toString()); - safeUrl.password = "redacted"; - - console.log("🗃️ Initializing Clickhouse event repository (v2)", { url: safeUrl.toString() }); - - const clickhouse = getClickhouseClient(); - - const repository = new ClickhouseEventRepository({ - clickhouse: clickhouse, - 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, - 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", - }); - - return repository; -} diff --git a/apps/webapp/app/v3/eventRepository/eventRepository.server.ts b/apps/webapp/app/v3/eventRepository/eventRepository.server.ts index de2a19e39..b761bbd90 100644 --- a/apps/webapp/app/v3/eventRepository/eventRepository.server.ts +++ b/apps/webapp/app/v3/eventRepository/eventRepository.server.ts @@ -151,7 +151,7 @@ export class EventRepository implements IEventRepository { await this.#flushBatch(nanoid(), [this.#createableEventToPrismaEvent(event)]); } - async insertMany(events: CreateEventInput[]) { + insertMany(events: CreateEventInput[]) { this._flushScheduler.addToBatch(events.map(this.#createableEventToPrismaEvent)); } diff --git a/apps/webapp/app/v3/eventRepository/eventRepository.types.ts b/apps/webapp/app/v3/eventRepository/eventRepository.types.ts index 0b45e5364..1a7adfd42 100644 --- a/apps/webapp/app/v3/eventRepository/eventRepository.types.ts +++ b/apps/webapp/app/v3/eventRepository/eventRepository.types.ts @@ -345,7 +345,7 @@ export type TraceDetailedSummary = { export interface IEventRepository { maximumLiveReloadingSetting: number; // Event insertion methods - insertMany(events: CreateEventInput[]): Promise; + insertMany(events: CreateEventInput[]): void; insertManyImmediate(events: CreateEventInput[]): Promise; // Run event completion methods diff --git a/apps/webapp/app/v3/eventRepository/index.server.ts b/apps/webapp/app/v3/eventRepository/index.server.ts index 70ea64403..5c9026572 100644 --- a/apps/webapp/app/v3/eventRepository/index.server.ts +++ b/apps/webapp/app/v3/eventRepository/index.server.ts @@ -1,29 +1,12 @@ import { env } from "~/env.server"; import { eventRepository } from "./eventRepository.server"; -import { - clickhouseEventRepository, - clickhouseEventRepositoryV2, -} from "./clickhouseEventRepositoryInstance.server"; -import { IEventRepository, TraceEventOptions } from "./eventRepository.types"; +import { type IEventRepository, type TraceEventOptions } from "./eventRepository.types"; import { prisma } from "~/db.server"; import { logger } from "~/services/logger.server"; import { FEATURE_FLAG } from "../featureFlags"; import { flag } from "../featureFlags.server"; import { getTaskEventStore } from "../taskEventStore.server"; - -export function resolveEventRepositoryForStore(store: string | undefined): IEventRepository { - const taskEventStore = store ?? env.EVENT_REPOSITORY_DEFAULT_STORE; - - if (taskEventStore === "clickhouse_v2") { - return clickhouseEventRepositoryV2; - } - - if (taskEventStore === "clickhouse") { - return clickhouseEventRepository; - } - - return eventRepository; -} +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; export const EVENT_STORE_TYPES = { POSTGRES: "postgres", @@ -58,63 +41,68 @@ export async function getConfiguredEventRepository( (organization.featureFlags as Record | null) ?? undefined ); + const { repository: resolvedRepository } = await clickhouseFactory.getEventRepositoryForOrganization( + taskEventStore, + organizationId + ); + if (taskEventStore === EVENT_STORE_TYPES.CLICKHOUSE_V2) { - return { repository: clickhouseEventRepositoryV2, store: EVENT_STORE_TYPES.CLICKHOUSE_V2 }; + return { repository: resolvedRepository, store: EVENT_STORE_TYPES.CLICKHOUSE_V2 }; } if (taskEventStore === EVENT_STORE_TYPES.CLICKHOUSE) { - return { repository: clickhouseEventRepository, store: EVENT_STORE_TYPES.CLICKHOUSE }; + return { repository: resolvedRepository, store: EVENT_STORE_TYPES.CLICKHOUSE }; } return { repository: eventRepository, store: EVENT_STORE_TYPES.POSTGRES }; } export async function getEventRepository( + organizationId: string, featureFlags: Record | undefined, parentStore: string | undefined ): Promise<{ repository: IEventRepository; store: string }> { - if (typeof parentStore === "string") { - if (parentStore === "clickhouse_v2") { - return { repository: clickhouseEventRepositoryV2, store: "clickhouse_v2" }; - } - if (parentStore === "clickhouse") { - return { repository: clickhouseEventRepository, store: "clickhouse" }; - } else { - return { repository: eventRepository, store: getTaskEventStore() }; - } + const taskEventStore = parentStore ?? (await resolveTaskEventRepositoryFlag(featureFlags)); + const { repository: resolvedRepository } = await clickhouseFactory.getEventRepositoryForOrganization( + taskEventStore, + organizationId + ); + + if (taskEventStore === "clickhouse_v2") { + return { repository: resolvedRepository, store: "clickhouse_v2" }; } - const taskEventRepository = await resolveTaskEventRepositoryFlag(featureFlags); - - if (taskEventRepository === "clickhouse_v2") { - return { repository: clickhouseEventRepositoryV2, store: "clickhouse_v2" }; - } - - if (taskEventRepository === "clickhouse") { - return { repository: clickhouseEventRepository, store: "clickhouse" }; + if (taskEventStore === "clickhouse") { + return { repository: resolvedRepository, store: "clickhouse" }; } return { repository: eventRepository, store: getTaskEventStore() }; } export async function getV3EventRepository( + organizationId: string, parentStore: string | undefined ): Promise<{ repository: IEventRepository; store: string }> { if (typeof parentStore === "string") { - if (parentStore === "clickhouse_v2") { - return { repository: clickhouseEventRepositoryV2, store: "clickhouse_v2" }; - } - if (parentStore === "clickhouse") { - return { repository: clickhouseEventRepository, store: "clickhouse" }; - } else { - return { repository: eventRepository, store: getTaskEventStore() }; - } + const { repository: resolvedRepository } = await clickhouseFactory.getEventRepositoryForOrganization( + parentStore, + organizationId + ); + return { repository: resolvedRepository, store: parentStore }; } if (env.EVENT_REPOSITORY_DEFAULT_STORE === "clickhouse_v2") { - return { repository: clickhouseEventRepositoryV2, store: "clickhouse_v2" }; + const { repository: resolvedRepository } = await clickhouseFactory.getEventRepositoryForOrganization( + "clickhouse_v2", + organizationId + ); + return { repository: resolvedRepository, store: "clickhouse_v2" }; } else if (env.EVENT_REPOSITORY_DEFAULT_STORE === "clickhouse") { - return { repository: clickhouseEventRepository, store: "clickhouse" }; + const { repository: resolvedRepository } = await clickhouseFactory.getEventRepositoryForOrganization( + "clickhouse", + organizationId + ); + return { repository: resolvedRepository, store: "clickhouse" }; } else { return { repository: eventRepository, store: getTaskEventStore() }; } @@ -203,7 +191,10 @@ async function recordRunEvent( }; } - const $eventRepository = resolveEventRepositoryForStore(foundRun.taskEventStore); + const { repository: $eventRepository } = await clickhouseFactory.getEventRepositoryForOrganization( + foundRun.taskEventStore, + foundRun.runtimeEnvironment.organizationId + ); const { attributes, startTime, ...optionsRest } = options; diff --git a/apps/webapp/app/v3/otlpExporter.server.ts b/apps/webapp/app/v3/otlpExporter.server.ts index 7505693e3..90f377948 100644 --- a/apps/webapp/app/v3/otlpExporter.server.ts +++ b/apps/webapp/app/v3/otlpExporter.server.ts @@ -19,16 +19,16 @@ import { Status_StatusCode, } from "@trigger.dev/otlp-importer"; import type { MetricsV1Input } from "@internal/clickhouse"; +import { ClickHouse } 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 { - clickhouseEventRepository, - clickhouseEventRepositoryV2, -} from "./eventRepository/clickhouseEventRepositoryInstance.server"; + clickhouseFactory, + ClickhouseFactory, + getDefaultClickhouseClient, +} from "~/services/clickhouse/clickhouseFactory.server"; +import { DynamicFlushScheduler } from "./dynamicFlushScheduler.server"; + import { generateSpanId } from "./eventRepository/common.server"; -import { EventRepository, eventRepository } from "./eventRepository/eventRepository.server"; import type { CreatableEventKind, CreatableEventStatus, @@ -39,21 +39,41 @@ import { startSpan } from "./tracing.server"; import { enrichCreatableEvents } from "./utils/enrichCreatableEvents.server"; import { waitForLlmPricingReady } from "./llmPricingRegistry.server"; import { env } from "~/env.server"; -import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings"; import { singleton } from "~/utils/singleton"; +type OTLPExporterConfig = { + clickhouseFactory: ClickhouseFactory; + metrics: { + batchSize: number; + flushInterval: number; + maxConcurrency: number; + }; + verbose: boolean; + spanAttributeValueLengthLimit: number; +}; + class OTLPExporter { private _tracer: Tracer; + private readonly _clickhouseFactory: ClickhouseFactory; + private readonly _defaultMetricsFlushScheduler: DynamicFlushScheduler; + private readonly _verbose: boolean; + private readonly _spanAttributeValueLengthLimit: number; - constructor( - private readonly _eventRepository: EventRepository, - private readonly _clickhouseEventRepository: ClickhouseEventRepository, - private readonly _clickhouseEventRepositoryV2: ClickhouseEventRepository, - private readonly _metricsFlushScheduler: DynamicFlushScheduler, - private readonly _verbose: boolean, - private readonly _spanAttributeValueLengthLimit: number - ) { + constructor(config: OTLPExporterConfig) { this._tracer = trace.getTracer("otlp-exporter"); + this._clickhouseFactory = config.clickhouseFactory; + this._verbose = config.verbose; + this._spanAttributeValueLengthLimit = config.spanAttributeValueLengthLimit; + this._defaultMetricsFlushScheduler = new DynamicFlushScheduler({ + batchSize: config.metrics.batchSize, + flushInterval: config.metrics.flushInterval, + callback: async (_flushId, batch) => { + await config.metrics.clickhouse.metrics.insert(batch); + }, + minConcurrency: 1, + maxConcurrency: config.metrics.maxConcurrency, + loadSheddingEnabled: false, + }); } async exportTraces(request: ExportTraceServiceRequest): Promise { @@ -74,9 +94,7 @@ class OTLPExporter { }); } - async exportMetrics( - request: ExportMetricsServiceRequest - ): Promise { + async exportMetrics(request: ExportMetricsServiceRequest): Promise { return await startSpan(this._tracer, "exportMetrics", async (span) => { const rows = this.#filterResourceMetrics(request.resourceMetrics).flatMap( (resourceMetrics) => { @@ -90,7 +108,7 @@ class OTLPExporter { span.setAttribute("metric_row_count", rows.length); if (rows.length > 0) { - this._metricsFlushScheduler.addToBatch(rows); + this._defaultMetricsFlushScheduler.addToBatch(rows); } return ExportMetricsServiceResponse.create(); @@ -118,42 +136,47 @@ class OTLPExporter { async #exportEvents( eventsWithStores: { events: Array; taskEventStore: string }[] ) { - const eventsGroupedByStore = eventsWithStores.reduce((acc, { events, taskEventStore }) => { - acc[taskEventStore] = acc[taskEventStore] || []; - acc[taskEventStore].push(...events); - return acc; - }, {} as Record>); + await waitForLlmPricingReady(); + + // Group by unique event repositories + const routeCache = new Map(); + const groups = new Map(); + for (const { events, taskEventStore } of eventsWithStores) { + for (const event of events) { + const routeKey = `${event.organizationId}\0${taskEventStore}`; + let resolved = routeCache.get(routeKey); + if (!resolved) { + resolved = this._clickhouseFactory.getEventRepositoryForOrganizationSync( + taskEventStore, + event.organizationId + ); + routeCache.set(routeKey, resolved); + } + + let group = groups.get(resolved.key); + if (!group) { + group = { repository: resolved.repository, events: [] }; + groups.set(resolved.key, group); + } + group.events.push(event); + } + } let eventCount = 0; - for (const [store, events] of Object.entries(eventsGroupedByStore)) { - const eventRepository = this.#getEventRepositoryForStore(store); - - await waitForLlmPricingReady(); + for (const [repoKey, { repository, events }] of groups) { const enrichedEvents = enrichCreatableEvents(events); - this.#logEventsVerbose(enrichedEvents, `exportEvents ${store}`); + this.#logEventsVerbose(enrichedEvents, `exportEvents ${repoKey}`); eventCount += enrichedEvents.length; - await eventRepository.insertMany(enrichedEvents); + repository.insertMany(enrichedEvents); } return eventCount; } - #getEventRepositoryForStore(store: string): IEventRepository { - if (store === "clickhouse") { - return this._clickhouseEventRepository; - } - - if (store === "clickhouse_v2") { - return this._clickhouseEventRepositoryV2; - } - - return this._eventRepository; - } - #logEventsVerbose(events: CreateEventInput[], prefix: string) { if (!this._verbose) return; @@ -393,7 +416,10 @@ function convertSpansToCreateableEvents( SemanticInternalAttributes.METADATA ); - const runTags = extractArrayAttribute(span.attributes ?? [], SemanticInternalAttributes.RUN_TAGS); + const runTags = extractArrayAttribute( + span.attributes ?? [], + SemanticInternalAttributes.RUN_TAGS + ); const properties = truncateAttributes( @@ -464,7 +490,10 @@ function floorToTenSecondBucket(timeUnixNano: bigint | number): string { const flooredMs = Math.floor(epochMs / 10_000) * 10_000; const date = new Date(flooredMs); // Format as ClickHouse DateTime: YYYY-MM-DD HH:MM:SS - return date.toISOString().replace("T", " ").replace(/\.\d{3}Z$/, ""); + return date + .toISOString() + .replace("T", " ") + .replace(/\.\d{3}Z$/, ""); } function convertMetricsToClickhouseRows( @@ -584,8 +613,7 @@ function resolveDataPointContext( attributes: Record; } { const runId = - resourceCtx.runId ?? - extractStringAttribute(dpAttributes, SemanticInternalAttributes.RUN_ID); + resourceCtx.runId ?? extractStringAttribute(dpAttributes, SemanticInternalAttributes.RUN_ID); const taskSlug = resourceCtx.taskSlug ?? extractStringAttribute(dpAttributes, SemanticInternalAttributes.TASK_SLUG); @@ -1173,25 +1201,16 @@ function hasUnpairedSurrogateAtEnd(str: string): boolean { export const otlpExporter = singleton("otlpExporter", initializeOTLPExporter); function initializeOTLPExporter() { - 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); + return new OTLPExporter({ + clickhouseFactory, + metrics: { + batchSize: env.METRICS_CLICKHOUSE_BATCH_SIZE, + flushInterval: env.METRICS_CLICKHOUSE_FLUSH_INTERVAL_MS, + maxConcurrency: env.METRICS_CLICKHOUSE_MAX_CONCURRENCY, }, - minConcurrency: 1, - maxConcurrency: env.METRICS_CLICKHOUSE_MAX_CONCURRENCY, - loadSheddingEnabled: false, - }); - - return new OTLPExporter( - eventRepository, - clickhouseEventRepository, - clickhouseEventRepositoryV2, - metricsFlushScheduler, - process.env.OTLP_EXPORTER_VERBOSE === "1", - process.env.SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT + verbose: process.env.OTLP_EXPORTER_VERBOSE === "1", + spanAttributeValueLengthLimit: process.env.SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT ? parseInt(process.env.SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, 10) - : 8192 - ); -} \ No newline at end of file + : 8192, + }); +} diff --git a/apps/webapp/app/v3/services/alerts/errorAlertEvaluator.server.ts b/apps/webapp/app/v3/services/alerts/errorAlertEvaluator.server.ts index e935a8a69..2c24033f7 100644 --- a/apps/webapp/app/v3/services/alerts/errorAlertEvaluator.server.ts +++ b/apps/webapp/app/v3/services/alerts/errorAlertEvaluator.server.ts @@ -7,7 +7,9 @@ import { } from "@trigger.dev/database"; import { $replica, prisma } from "~/db.server"; import { ErrorAlertConfig } from "~/models/projectAlert.server"; -import { clickhouseClient } from "~/services/clickhouseInstance.server"; +import { getDefaultClickhouseClient } from "~/services/clickhouse/clickhouseFactory.server"; + +const clickhouseClient = getDefaultClickhouseClient(); import { logger } from "~/services/logger.server"; import { alertsWorker } from "~/v3/alertsWorker.server"; diff --git a/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts b/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts index 07a428629..cf1f80165 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 { getClickhouseForOrganization } from "~/services/clickhouse/clickhouseFactory.server"; +import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { parseRunListInputOptions, type RunListInputFilters, @@ -38,7 +38,7 @@ 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 clickhouse = await clickhouseFactory.getClickhouseForOrganization(organizationId, "standard"); const runsRepository = new RunsRepository({ clickhouse, prisma: this._replica as PrismaClient, @@ -148,7 +148,7 @@ export class BulkActionService extends BaseService { ...rawParams, }); - const clickhouse = await getClickhouseForOrganization(group.project.organizationId, "standard"); + const clickhouse = await clickhouseFactory.getClickhouseForOrganization(group.project.organizationId, "standard"); const runsRepository = new RunsRepository({ clickhouse, prisma: this._replica as PrismaClient,