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

Use the clickhouseFactory directly. WIP on new event repository

Browse Source
This commit is contained in:
Matt Aitken
2026-04-02 18:48:06 +01:00
parent 62ec253eae
commit 5700df8b5c
38 changed files with 314 additions and 410 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts
+2 -2
View File
@@ -9,7 +9,7 @@ import { type Project, type RuntimeEnvironment, type TaskRunStatus } from "@trig
import assertNever from "assert-never";
import { z } from "zod";
import { API_VERSIONS, RunStatusUnspecifiedApiVersion } from "~/api/versions";
import { 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 });
apps/webapp/app/presenters/v3/CreateBulkActionPresenter.server.ts
+2 -2
View File
@@ -1,6 +1,6 @@
import { type PrismaClient } from "@trigger.dev/database";
import { CreateBulkActionSearchParams } from "~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.bulkaction";
import { 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,
apps/webapp/app/presenters/v3/RunTagListPresenter.server.ts
+2 -2
View File
@@ -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,
apps/webapp/app/presenters/v3/TaskListPresenter.server.ts
+2 -2
View File
@@ -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,
});
apps/webapp/app/presenters/v3/UsagePresenter.server.ts
+2 -2
View File
@@ -4,7 +4,7 @@ import { getUsage, getUsageSeries } from "~/services/platform.v3.server";
import { createTimeSeriesData } from "~/utils/graphs";
import { BasePresenter } from "./basePresenter.server";
import { DataPoint, linear } from "regression";
import { 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(),
apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts
+2 -2
View File
@@ -1,7 +1,7 @@
import { ScheduleObject } from "@trigger.dev/core/v3";
import { PrismaClient, prisma } from "~/db.server";
import { displayableEnvironment } from "~/models/runtimeEnvironment.server";
import { 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,
apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts
+2 -2
View File
@@ -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,
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboards.$dashboardKey/route.tsx
+2 -2
View File
@@ -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 }[] = [];
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors.$fingerprint/route.tsx
+7 -1
View File
@@ -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 {
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors._index/route.tsx
+2 -2
View File
@@ -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
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsx
+2 -2
View File
@@ -16,7 +16,7 @@ import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { LogsListPresenter, LogEntry } from "~/presenters/v3/LogsListPresenter.server";
import type { LogLevel } from "~/utils/logUtils";
import { $replica, prisma } from "~/db.server";
import { 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
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.$modelId/route.tsx
+2 -2
View File
@@ -28,7 +28,7 @@ import type { QueryWidgetConfig } from "~/components/metrics/QueryWidget";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { ModelRegistryPresenter } from "~/presenters/v3/ModelRegistryPresenter.server";
import { 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);
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models._index/route.tsx
+2 -2
View File
@@ -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();
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.compare/route.tsx
+2 -2
View File
@@ -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);
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug/route.tsx
+2 -2
View File
@@ -70,7 +70,7 @@ import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { type GenerationRow, PromptPresenter } from "~/presenters/v3/PromptPresenter.server";
import { SpanView } from "~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route";
import { 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<ReturnType<typeof presenter.listGenerations>>["generations"] = [];
let generationsPagination: { next?: string } = {};
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts._index/route.tsx
+2 -2
View File
@@ -22,7 +22,7 @@ import { useProject } from "~/hooks/useProject";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { PromptPresenter } from "~/presenters/v3/PromptPresenter.server";
import { 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);
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam/route.tsx
+2 -2
View File
@@ -92,7 +92,7 @@ import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { NextRunListPresenter } from "~/presenters/v3/NextRunListPresenter.server";
import { RunEnvironmentMismatchError, RunPresenter } from "~/presenters/v3/RunPresenter.server";
import { 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,
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx
+2 -2
View File
@@ -44,7 +44,7 @@ import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { getRunFiltersFromRequest } from "~/presenters/RunFilters.server";
import { NextRunListPresenter } from "~/presenters/v3/NextRunListPresenter.server";
import { 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,
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.test.tasks.$taskParam/route.tsx
+2 -2
View File
@@ -74,7 +74,7 @@ import { Dialog, DialogContent, DialogHeader, DialogTrigger } from "~/components
import { DialogClose, DialogDescription } from "@radix-ui/react-dialog";
import { FormButtons } from "~/components/primitives/FormButtons";
import { $replica } from "~/db.server";
import { 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([
apps/webapp/app/routes/api.v1.prompts.$slug.ts
+3 -3
View File
@@ -2,7 +2,7 @@ import { json } from "@remix-run/server-runtime";
import { z } from "zod";
import { prisma } from "~/db.server";
import { PromptPresenter } from "~/presenters/v3/PromptPresenter.server";
import { 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,
apps/webapp/app/routes/api.v1.prompts.$slug.versions.ts
+2 -2
View File
@@ -2,7 +2,7 @@ import { json } from "@remix-run/server-runtime";
import { z } from "zod";
import { prisma } from "~/db.server";
import { PromptPresenter } from "~/presenters/v3/PromptPresenter.server";
import { 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);
apps/webapp/app/routes/api.v1.prompts._index.ts
+2 -2
View File
@@ -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,
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsx
+2 -2
View File
@@ -1,7 +1,7 @@
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { typedjson } from "remix-typedjson";
import { z } from "zod";
import { 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;
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.ts
+2 -2
View File
@@ -6,7 +6,7 @@ import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { LogsListPresenter, type LogLevel, LogsListOptionsSchema } from "~/presenters/v3/LogsListPresenter.server";
import { $replica } from "~/db.server";
import { 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);
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug.generations.ts
+2 -2
View File
@@ -6,7 +6,7 @@ import { EnvironmentParamSchema } from "~/utils/pathBuilder";
import { parsePeriodToMs } from "~/utils/periods";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { 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,
apps/webapp/app/runEngine/concerns/traceEvents.server.ts
+10 -12
View File
@@ -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<string, unknown>,
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(
{
apps/webapp/app/runEngine/services/triggerFailedTask.server.ts
+19 -20
View File
@@ -74,6 +74,7 @@ export class TriggerFailedTaskService {
try {
const { repository, store } = await getEventRepository(
request.environment.organization.id,
request.environment.organization.featureFlags as Record<string, unknown>,
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,
apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts
+89 -87
View File
@@ -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<boolean> {
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<ClickhouseEventRepository> {
): 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<ClickHouse> {
return clickhouseFactory.getClickhouseForOrganization(organizationId, clientType);
}
export async function getEventRepositoryForOrganization(
organizationId: string
): Promise<ClickhouseEventRepository> {
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<ClickHouse> {
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<ClickHouse> {
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}`);
}
}
}
apps/webapp/app/services/queryService.server.ts
+2 -2
View File
@@ -11,7 +11,7 @@ import type { TableSchema, WhereClauseCondition } from "@internal/tsql";
import { z } from "zod";
import { prisma } from "~/db.server";
import { env } from "~/env.server";
import { 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<TOut extends z.ZodSchema>(
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()),
apps/webapp/app/services/runsReplicationService.server.ts
+3 -3
View File
@@ -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<T = any> {
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,
apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts
+1 -2
View File
@@ -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<void> {
insertMany(events: CreateEventInput[]): void {
this.addToBatch(events.flatMap((event) => this.createEventToTaskEventV1Input(event)));
// Dual-write LLM metrics records for spans with cost enrichment
apps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.ts
-112
View File
@@ -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;
}
apps/webapp/app/v3/eventRepository/eventRepository.server.ts
+1 -1
View File
@@ -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));
}
apps/webapp/app/v3/eventRepository/eventRepository.types.ts
+1 -1
View File
@@ -345,7 +345,7 @@ export type TraceDetailedSummary = {
export interface IEventRepository {
maximumLiveReloadingSetting: number;
// Event insertion methods
insertMany(events: CreateEventInput[]): Promise<void>;
insertMany(events: CreateEventInput[]): void;
insertManyImmediate(events: CreateEventInput[]): Promise<void>;
// Run event completion methods
apps/webapp/app/v3/eventRepository/index.server.ts
+40 -49
View File
@@ -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<string, unknown> | 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<string, unknown> | 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;
apps/webapp/app/v3/otlpExporter.server.ts
+86 -67
View File
@@ -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<MetricsV1Input>;
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<MetricsV1Input>,
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<MetricsV1Input>({
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<ExportTraceServiceResponse> {
@@ -74,9 +94,7 @@ class OTLPExporter {
});
}
async exportMetrics(
request: ExportMetricsServiceRequest
): Promise<ExportMetricsServiceResponse> {
async exportMetrics(request: ExportMetricsServiceRequest): Promise<ExportMetricsServiceResponse> {
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<CreateEventInput>; taskEventStore: string }[]
) {
const eventsGroupedByStore = eventsWithStores.reduce((acc, { events, taskEventStore }) => {
acc[taskEventStore] = acc[taskEventStore] || [];
acc[taskEventStore].push(...events);
return acc;
}, {} as Record<string, Array<CreateEventInput>>);
await waitForLlmPricingReady();
// Group by unique event repositories
const routeCache = new Map<string, { key: string; repository: IEventRepository }>();
const groups = new Map<string, { repository: IEventRepository; events: CreateEventInput[] }>();
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<string, unknown>;
} {
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<MetricsV1Input>({
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
);
}
: 8192,
});
}
apps/webapp/app/v3/services/alerts/errorAlertEvaluator.server.ts
+3 -1
View File
@@ -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";
apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts
+3 -3
View File
@@ -7,7 +7,7 @@ import {
} from "@trigger.dev/database";
import { getRunFiltersFromRequest } from "~/presenters/RunFilters.server";
import { type CreateBulkActionPayload } from "~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.bulkaction";
import { 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,