Make ClickHouse required

matt-aitken · matt-aitken · commit 3a4a62216d31 · 2025-07-14T09:44:53.000+01:00
diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts
@@ -912,7 +912,7 @@ const EnvironmentSchema = z.object({
   RUN_REPLICATION_INSERT_MAX_DELAY_MS: z.coerce.number().int().default(2000),
 
   // Clickhouse
-  CLICKHOUSE_URL: z.string().optional(),
+  CLICKHOUSE_URL: z.string(),
   CLICKHOUSE_KEEP_ALIVE_ENABLED: z.string().default("1"),
   CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
   CLICKHOUSE_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
diff --git a/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts b/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts
@@ -213,10 +213,6 @@ export class ApiRunListPresenter extends BasePresenter {
         options.batchId = searchParams["filter[batch]"];
       }
 
-      if (!clickhouseClient) {
-        throw new Error("Clickhouse is not supported yet");
-      }
-
       const presenter = new NextRunListPresenter(this._prisma, clickhouseClient);
 
       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
@@ -24,10 +24,6 @@ export class CreateBulkActionPresenter extends BasePresenter {
       Object.fromEntries(new URL(request.url).searchParams)
     );
 
-    if (!clickhouseClient) {
-      throw new Error("Clickhouse client not found");
-    }
-
     const runsRepository = new RunsRepository({
       clickhouse: clickhouseClient,
       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
@@ -11,7 +11,6 @@ import {
   type CurrentRunningStats,
   type DailyTaskActivity,
   type EnvironmentMetricsRepository,
-  PostgrestEnvironmentMetricsRepository,
 } from "~/services/environmentMetricsRepository.server";
 import { singleton } from "~/utils/singleton";
 import { findCurrentWorkerFromEnvironment } from "~/v3/models/workerDeployment.server";
@@ -110,13 +109,9 @@ export class TaskListPresenter {
 export const taskListPresenter = singleton("taskListPresenter", setupTaskListPresenter);
 
 function setupTaskListPresenter() {
-  const environmentMetricsRepository = clickhouseClient
-    ? new ClickHouseEnvironmentMetricsRepository({
-        clickhouse: clickhouseClient,
-      })
-    : new PostgrestEnvironmentMetricsRepository({
-        prisma: $replica,
-      });
+  const environmentMetricsRepository = new ClickHouseEnvironmentMetricsRepository({
+    clickhouse: clickhouseClient,
+  });
 
   return new TaskListPresenter(environmentMetricsRepository, $replica);
 }
diff --git a/apps/webapp/app/presenters/v3/UsagePresenter.server.ts b/apps/webapp/app/presenters/v3/UsagePresenter.server.ts
@@ -124,60 +124,24 @@ async function getTaskUsageByOrganization(
   endOfMonth: Date,
   replica: PrismaClientOrTransaction
 ) {
-  if (clickhouseClient) {
-    const [queryError, tasks] = await clickhouseClient.taskRuns.getTaskUsageByOrganization({
-      startTime: startOfMonth.getTime(),
-      endTime: endOfMonth.getTime(),
-      organizationId,
-    });
-
-    if (queryError) {
-      throw queryError;
-    }
-
-    return tasks
-      .map((task) => ({
-        taskIdentifier: task.task_identifier,
-        runCount: Number(task.run_count),
-        averageDuration: Number(task.average_duration),
-        averageCost: Number(task.average_cost) + env.CENTS_PER_RUN / 100,
-        totalDuration: Number(task.total_duration),
-        totalCost: Number(task.total_cost) + Number(task.total_base_cost),
-      }))
-      .sort((a, b) => b.totalCost - a.totalCost);
-  } else {
-    return replica.$queryRaw<TaskUsageItem[]>`
-    SELECT
-      tr."taskIdentifier",
-      COUNT(*) AS "runCount",
-      AVG(tr."usageDurationMs") AS "averageDuration",
-      SUM(tr."usageDurationMs") AS "totalDuration",
-      AVG(tr."costInCents") / 100.0 AS "averageCost",
-      SUM(tr."costInCents") / 100.0 AS "totalCost",
-      SUM(tr."baseCostInCents") / 100.0 AS "totalBaseCost"
-    FROM
-        ${sqlDatabaseSchema}."TaskRun" tr
-        JOIN ${sqlDatabaseSchema}."Project" pr ON pr.id = tr."projectId"
-        JOIN ${sqlDatabaseSchema}."Organization" org ON org.id = pr."organizationId"
-        JOIN ${sqlDatabaseSchema}."RuntimeEnvironment" env ON env."id" = tr."runtimeEnvironmentId"
-    WHERE
-        env.type <> 'DEVELOPMENT'
-        AND tr."createdAt" > ${startOfMonth}
-        AND tr."createdAt" < ${endOfMonth}
-        AND org.id = ${organizationId}
-    GROUP BY
-        tr."taskIdentifier";
-  `.then((data) => {
-      return data
-        .map((item) => ({
-          taskIdentifier: item.taskIdentifier,
-          runCount: Number(item.runCount),
-          averageDuration: Number(item.averageDuration),
-          averageCost: Number(item.averageCost) + env.CENTS_PER_RUN / 100,
-          totalDuration: Number(item.totalDuration),
-          totalCost: Number(item.totalCost) + Number(item.totalBaseCost),
-        }))
-        .sort((a, b) => b.totalCost - a.totalCost);
-    });
+  const [queryError, tasks] = await clickhouseClient.taskRuns.getTaskUsageByOrganization({
+    startTime: startOfMonth.getTime(),
+    endTime: endOfMonth.getTime(),
+    organizationId,
+  });
+
+  if (queryError) {
+    throw queryError;
   }
+
+  return tasks
+    .map((task) => ({
+      taskIdentifier: task.task_identifier,
+      runCount: Number(task.run_count),
+      averageDuration: Number(task.average_duration),
+      averageCost: Number(task.average_cost) + env.CENTS_PER_RUN / 100,
+      totalDuration: Number(task.total_duration),
+      totalCost: Number(task.total_cost) + Number(task.total_base_cost),
+    }))
+    .sort((a, b) => b.totalCost - a.totalCost);
 }
diff --git a/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts b/apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts
@@ -77,10 +77,6 @@ export class ViewSchedulePresenter {
       ? nextScheduledTimestamps(schedule.generatorExpression, schedule.timezone, new Date(), 5)
       : [];
 
-    if (!clickhouseClient) {
-      throw new Error("Clickhouse is not supported yet");
-    }
-
     const runPresenter = new NextRunListPresenter(this.#prismaClient, clickhouseClient);
     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
@@ -79,10 +79,6 @@ export class WaitpointPresenter extends BasePresenter {
     const connectedRuns: NextRunListItem[] = [];
 
     if (connectedRunIds.length > 0) {
-      if (!clickhouseClient) {
-        throw new Error("Clickhouse is not supported yet");
-      }
-
       const runPresenter = new NextRunListPresenter(this._prisma, clickhouseClient);
       const { runs } = await runPresenter.call(
         waitpoint.environment.organizationId,
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
@@ -81,10 +81,6 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     throw new Error("Environment not found");
   }
 
-  if (!clickhouseClient) {
-    throw new Error("Clickhouse is not supported yet");
-  }
-
   const filters = await getRunFiltersFromRequest(request);
 
   const presenter = new NextRunListPresenter($replica, clickhouseClient);
diff --git a/apps/webapp/app/services/clickhouseInstance.server.ts b/apps/webapp/app/services/clickhouseInstance.server.ts
@@ -5,11 +5,6 @@ import { singleton } from "~/utils/singleton";
 export const clickhouseClient = singleton("clickhouseClient", initializeClickhouseClient);
 
 function initializeClickhouseClient() {
-  if (!env.CLICKHOUSE_URL) {
-    console.log("🗃️  Clickhouse service not enabled");
-    return;
-  }
-
   const url = new URL(env.CLICKHOUSE_URL);
 
   // Remove secure param
diff --git a/apps/webapp/app/services/environmentMetricsRepository.server.ts b/apps/webapp/app/services/environmentMetricsRepository.server.ts
@@ -1,7 +1,5 @@
-import { ClickHouse } from "@internal/clickhouse";
-import { Logger, LogLevel } from "@trigger.dev/core/logger";
-import type { PrismaClientOrTransaction, TaskRunStatus } from "@trigger.dev/database";
-import { Prisma } from "@trigger.dev/database";
+import { type ClickHouse } from "@internal/clickhouse";
+import type { TaskRunStatus } from "@trigger.dev/database";
 import { QUEUED_STATUSES } from "~/components/runs/v3/TaskRunStatus";
 
 export type DailyTaskActivity = Record<string, ({ day: string } & Record<TaskRunStatus, number>)[]>;
@@ -34,147 +32,6 @@ export interface EnvironmentMetricsRepository {
   }): Promise<AverageDurations>;
 }
 
-export type PostgrestEnvironmentMetricsRepositoryOptions = {
-  prisma: PrismaClientOrTransaction;
-  schema?: string;
-  logger?: Logger;
-  logLevel?: LogLevel;
-};
-
-export class PostgrestEnvironmentMetricsRepository implements EnvironmentMetricsRepository {
-  private readonly logger: Logger;
-  private readonly schema: string;
-
-  constructor(private readonly options: PostgrestEnvironmentMetricsRepositoryOptions) {
-    this.logger =
-      options.logger ??
-      new Logger("PostgrestEnvironmentMetricsRepository", options.logLevel ?? "info");
-    this.schema = options.schema ?? "public";
-  }
-
-  public async getDailyTaskActivity({
-    environmentId,
-    days,
-    tasks,
-  }: {
-    environmentId: string;
-    days: number;
-    tasks: string[];
-  }): Promise<DailyTaskActivity> {
-    if (tasks.length === 0) {
-      return {};
-    }
-
-    const activity = await this.options.prisma.$queryRaw<
-      {
-        taskIdentifier: string;
-        status: TaskRunStatus;
-        day: Date;
-        count: BigInt;
-      }[]
-    >`
-    SELECT
-    tr."taskIdentifier",
-    tr."status",
-    DATE(tr."createdAt") as day,
-    COUNT(*)
-  FROM
-    ${Prisma.sql([this.schema])}."TaskRun" as tr
-  WHERE
-    tr."taskIdentifier" IN (${Prisma.join(tasks)})
-    AND tr."runtimeEnvironmentId" = ${environmentId}
-    AND tr."createdAt" >= (current_date - interval '1 day' * ${days})
-  GROUP BY
-    tr."taskIdentifier",
-    tr."status",
-    day
-  ORDER BY
-    tr."taskIdentifier" ASC,
-    day ASC,
-    tr."status" ASC;`;
-
-    return fillInDailyTaskActivity(activity, days);
-  }
-
-  public async getCurrentRunningStats({
-    environmentId,
-    days,
-    tasks,
-  }: {
-    environmentId: string;
-    days: number;
-    tasks: string[];
-  }): Promise<CurrentRunningStats> {
-    if (tasks.length === 0) {
-      return {};
-    }
-
-    const stats = await this.options.prisma.$queryRaw<
-      {
-        taskIdentifier: string;
-        status: TaskRunStatus;
-        count: BigInt;
-      }[]
-    >`
-    SELECT
-    tr."taskIdentifier",
-    tr.status,
-    COUNT(*)
-  FROM
-    ${Prisma.sql([this.schema])}."TaskRun" as tr
-  WHERE
-    tr."taskIdentifier" IN (${Prisma.join(tasks)})
-    AND tr."runtimeEnvironmentId" = ${environmentId}
-    AND tr."createdAt" >= (current_date - interval '1 day' * ${days})
-    AND tr."status" = ANY(ARRAY[${Prisma.join([
-      ...QUEUED_STATUSES,
-      "EXECUTING",
-    ])}]::\"TaskRunStatus\"[])
-  GROUP BY
-    tr."taskIdentifier",
-    tr.status
-  ORDER BY
-    tr."taskIdentifier" ASC`;
-
-    return fillInCurrentRunningStats(stats, tasks);
-  }
-
-  public async getAverageDurations({
-    environmentId,
-    days,
-    tasks,
-  }: {
-    environmentId: string;
-    days: number;
-    tasks: string[];
-  }): Promise<AverageDurations> {
-    if (tasks.length === 0) {
-      return {};
-    }
-
-    const durations = await this.options.prisma.$queryRaw<
-      {
-        taskIdentifier: string;
-        duration: Number;
-      }[]
-    >`
-    SELECT
-      tr."taskIdentifier",
-      AVG(EXTRACT(EPOCH FROM (tr."updatedAt" - COALESCE(tr."startedAt", tr."lockedAt")))) as duration
-      FROM
-      ${Prisma.sql([this.schema])}."TaskRun" as tr
-    WHERE
-      tr."taskIdentifier" IN (${Prisma.join(tasks)})
-      AND tr."runtimeEnvironmentId" = ${environmentId}
-      AND tr."createdAt" >= (current_date - interval '1 day' * ${days})
-      AND tr."status" IN ('COMPLETED_SUCCESSFULLY', 'COMPLETED_WITH_ERRORS')
-    GROUP BY
-      tr."taskIdentifier";`;
-
-    return Object.fromEntries(durations.map((s) => [s.taskIdentifier, Number(s.duration)]));
-  }
-}
-
 export type ClickHouseEnvironmentMetricsRepositoryOptions = {
   clickhouse: ClickHouse;
 };
diff --git a/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts b/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts
@@ -32,10 +32,6 @@ export class BulkActionService extends BaseService {
   ) {
     const filters = await getFilters(payload, request);
 
-    if (!clickhouseClient) {
-      throw new Error("Clickhouse client not found");
-    }
-
     // Count the runs that will be affected by the bulk action
     const runsRepository = new RunsRepository({
       clickhouse: clickhouseClient,
@@ -144,10 +140,6 @@ export class BulkActionService extends BaseService {
       ...(group.params && typeof group.params === "object" ? group.params : {}),
     });
 
-    if (!clickhouseClient) {
-      throw new Error("Clickhouse client not found");
-    }
-
     const runsRepository = new RunsRepository({
       clickhouse: clickhouseClient,
       prisma: this._replica as PrismaClient,

Original file line number	Diff line number	Diff line change
`@@ -213,10 +213,6 @@ export class ApiRunListPresenter extends BasePresenter {`
`213`	`213`	`options.batchId = searchParams["filter[batch]"];`
`214`	`214`	`}`
`215`	`215`
`216`		`- if (!clickhouseClient) {`
`217`		`- throw new Error("Clickhouse is not supported yet");`
`218`		`- }`
`219`		`-`
`220`	`216`	`const presenter = new NextRunListPresenter(this._prisma, clickhouseClient);`
`221`	`217`
`222`	`218`	`logger.debug("Calling RunListPresenter", { options });`
Original file line number	Diff line number	Diff line change
`@@ -81,10 +81,6 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {`
`81`	`81`	`throw new Error("Environment not found");`
`82`	`82`	`}`
`83`	`83`
`84`		`- if (!clickhouseClient) {`
`85`		`- throw new Error("Clickhouse is not supported yet");`
`86`		`- }`
`87`		`-`
`88`	`84`	`const filters = await getRunFiltersFromRequest(request);`
`89`	`85`
`90`	`86`	`const presenter = new NextRunListPresenter($replica, clickhouseClient);`