Use the clickhouseFactory directly. WIP on new event repository

Author: matt-aitken

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 });

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,

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,

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,
     });

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(),

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,

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,

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

@@ -32,7 +32,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";
@@ -75,7 +75,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

@@ -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

@@ -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