Made the run repository an interface, deferring just to CH for now

Author: matt-aitken

apps/webapp/app/presenters/RunFilters.server.ts

@@ -3,7 +3,7 @@ import {
   TaskRunListSearchFilters,
 } from "~/components/runs/v3/RunFilters";
 import { getRootOnlyFilterPreference } from "~/services/preferences/uiPreferences.server";
-import { type ParsedRunFilters } from "~/services/runsRepository.server";
+import { type ParsedRunFilters } from "~/services/runsRepository/runsRepository.server";
 
 type FiltersFromRequest = ParsedRunFilters & Required<Pick<ParsedRunFilters, "rootOnly">>;
 

apps/webapp/app/presenters/v3/BulkActionPresenter.server.ts

@@ -1,7 +1,7 @@
 import { getUsername } from "~/utils/username";
 import { BasePresenter } from "./basePresenter.server";
 import { type BulkActionMode } from "~/components/BulkActionFilterSummary";
-import { parseRunListInputOptions } from "~/services/runsRepository.server";
+import { parseRunListInputOptions } from "~/services/runsRepository/runsRepository.server";
 import { TaskRunListSearchFilters } from "~/components/runs/v3/RunFilters";
 
 type BulkActionOptions = {

apps/webapp/app/presenters/v3/CreateBulkActionPresenter.server.ts

@@ -1,7 +1,7 @@
 import { type PrismaClient } from "@trigger.dev/database";
 import { CreateBulkActionSearchParams } from "~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.bulkaction";
 import { clickhouseClient } from "~/services/clickhouseInstance.server";
-import { RunsRepository } from "~/services/runsRepository.server";
+import { RunsRepository } from "~/services/runsRepository/runsRepository.server";
 import { getRunFiltersFromRequest } from "../RunFilters.server";
 import { BasePresenter } from "./basePresenter.server";
 

apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts

@@ -9,7 +9,7 @@ import { type Direction } from "~/components/ListPagination";
 import { timeFilters } from "~/components/runs/v3/SharedFilters";
 import { findDisplayableEnvironment } from "~/models/runtimeEnvironment.server";
 import { getAllTaskIdentifiers } from "~/models/task.server";
-import { RunsRepository } from "~/services/runsRepository.server";
+import { RunsRepository } from "~/services/runsRepository/runsRepository.server";
 import { machinePresetFromRun } from "~/v3/machinePresets.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { isCancellableRunStatus, isFinalRunStatus, isPendingRunStatus } from "~/v3/taskStatus";

…pp/app/services/runsRepository.server.ts …itory/clickhouseRunsRepository.server.tsapps/webapp/app/services/runsRepository.server.ts renamed to apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts apps/webapp/app/services/runsRepository.server.ts renamed to apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts

@@ -1,72 +1,16 @@
-import { type ClickHouse, type ClickhouseQueryBuilder } from "@internal/clickhouse";
-import { type Tracer } from "@internal/tracing";
-import { type Logger, type LogLevel } from "@trigger.dev/core/logger";
-import { MachinePresetName } from "@trigger.dev/core/v3";
-import { BulkActionId, RunId } from "@trigger.dev/core/v3/isomorphic";
-import { TaskRunStatus } from "@trigger.dev/database";
-import parseDuration from "parse-duration";
-import { z } from "zod";
 import { timeFilters } from "~/components/runs/v3/SharedFilters";
-import { type PrismaClient } from "~/db.server";
-
-export type RunsRepositoryOptions = {
-  clickhouse: ClickHouse;
-  prisma: PrismaClient;
-  logger?: Logger;
-  logLevel?: LogLevel;
-  tracer?: Tracer;
-};
-
-const RunStatus = z.enum(Object.values(TaskRunStatus) as [TaskRunStatus, ...TaskRunStatus[]]);
-
-const RunListInputOptionsSchema = z.object({
-  organizationId: z.string(),
-  projectId: z.string(),
-  environmentId: z.string(),
-  //filters
-  tasks: z.array(z.string()).optional(),
-  versions: z.array(z.string()).optional(),
-  statuses: z.array(RunStatus).optional(),
-  tags: z.array(z.string()).optional(),
-  scheduleId: z.string().optional(),
-  period: z.string().optional(),
-  from: z.number().optional(),
-  to: z.number().optional(),
-  isTest: z.boolean().optional(),
-  rootOnly: z.boolean().optional(),
-  batchId: z.string().optional(),
-  runId: z.array(z.string()).optional(),
-  bulkId: z.string().optional(),
-  queues: z.array(z.string()).optional(),
-  machines: MachinePresetName.array().optional(),
-});
-
-export type RunListInputOptions = z.infer<typeof RunListInputOptionsSchema>;
-export type RunListInputFilters = Omit<
-  RunListInputOptions,
-  "organizationId" | "projectId" | "environmentId"
->;
-
-export type ParsedRunFilters = RunListInputFilters & {
-  cursor?: string;
-  direction?: "forward" | "backward";
-};
-
-type FilterRunsOptions = Omit<RunListInputOptions, "period"> & {
-  period: number | undefined;
-};
-
-type Pagination = {
-  page: {
-    size: number;
-    cursor?: string;
-    direction?: "forward" | "backward";
-  };
-};
-
-export type ListRunsOptions = RunListInputOptions & Pagination;
-
-export class RunsRepository {
+import {
+  type FilterRunsOptions,
+  type RunListInputOptions,
+  type IRunsRepository,
+  type ListRunsOptions,
+  type RunsRepositoryOptions,
+} from "./runsRepository.server";
+import parseDuration from "parse-duration";
+import { BulkActionId, RunId } from "@trigger.dev/core/v3/isomorphic";
+import { type ClickhouseQueryBuilder } from "@internal/clickhouse";
+
+export class ClickHouseRunsRepository implements IRunsRepository {
   constructor(private readonly options: RunsRepositoryOptions) {}
 
   async listRunIds(options: ListRunsOptions) {
@@ -373,7 +317,3 @@ function applyRunFiltersToQueryBuilder<T>(
     });
   }
 }
-
-export function parseRunListInputOptions(data: any): RunListInputOptions {
-  return RunListInputOptionsSchema.parse(data);
-}

apps/webapp/app/services/runsRepository/runsRepository.server.ts

@@ -0,0 +1,142 @@
+import { type ClickHouse, type ClickhouseQueryBuilder } from "@internal/clickhouse";
+import { type Tracer } from "@internal/tracing";
+import { type Logger, type LogLevel } from "@trigger.dev/core/logger";
+import { MachinePresetName } from "@trigger.dev/core/v3";
+import { BulkActionId, RunId } from "@trigger.dev/core/v3/isomorphic";
+import { Prisma, TaskRunStatus } from "@trigger.dev/database";
+import parseDuration from "parse-duration";
+import { z } from "zod";
+import { timeFilters } from "~/components/runs/v3/SharedFilters";
+import { type PrismaClient } from "~/db.server";
+import { ClickHouseRunsRepository } from "./clickhouseRunsRepository.server";
+
+export type RunsRepositoryOptions = {
+  clickhouse: ClickHouse;
+  prisma: PrismaClient;
+  logger?: Logger;
+  logLevel?: LogLevel;
+  tracer?: Tracer;
+};
+
+const RunStatus = z.enum(Object.values(TaskRunStatus) as [TaskRunStatus, ...TaskRunStatus[]]);
+
+const RunListInputOptionsSchema = z.object({
+  organizationId: z.string(),
+  projectId: z.string(),
+  environmentId: z.string(),
+  //filters
+  tasks: z.array(z.string()).optional(),
+  versions: z.array(z.string()).optional(),
+  statuses: z.array(RunStatus).optional(),
+  tags: z.array(z.string()).optional(),
+  scheduleId: z.string().optional(),
+  period: z.string().optional(),
+  from: z.number().optional(),
+  to: z.number().optional(),
+  isTest: z.boolean().optional(),
+  rootOnly: z.boolean().optional(),
+  batchId: z.string().optional(),
+  runId: z.array(z.string()).optional(),
+  bulkId: z.string().optional(),
+  queues: z.array(z.string()).optional(),
+  machines: MachinePresetName.array().optional(),
+});
+
+export type RunListInputOptions = z.infer<typeof RunListInputOptionsSchema>;
+export type RunListInputFilters = Omit<
+  RunListInputOptions,
+  "organizationId" | "projectId" | "environmentId"
+>;
+
+export type ParsedRunFilters = RunListInputFilters & {
+  cursor?: string;
+  direction?: "forward" | "backward";
+};
+
+export type FilterRunsOptions = Omit<RunListInputOptions, "period"> & {
+  period: number | undefined;
+};
+
+type Pagination = {
+  page: {
+    size: number;
+    cursor?: string;
+    direction?: "forward" | "backward";
+  };
+};
+
+type ListedRun = Prisma.TaskRunGetPayload<{
+  select: {
+    id: true;
+    friendlyId: true;
+    taskIdentifier: true;
+    taskVersion: true;
+    runtimeEnvironmentId: true;
+    status: true;
+    createdAt: true;
+    startedAt: true;
+    lockedAt: true;
+    delayUntil: true;
+    updatedAt: true;
+    completedAt: true;
+    isTest: true;
+    spanId: true;
+    idempotencyKey: true;
+    ttl: true;
+    expiredAt: true;
+    costInCents: true;
+    baseCostInCents: true;
+    usageDurationMs: true;
+    runTags: true;
+    depth: true;
+    rootTaskRunId: true;
+    batchId: true;
+    metadata: true;
+    metadataType: true;
+    machinePreset: true;
+    queue: true;
+  };
+}>;
+
+export type ListRunsOptions = RunListInputOptions & Pagination;
+
+export interface IRunsRepository {
+  listRunIds(options: ListRunsOptions): Promise<string[]>;
+  listRuns(options: ListRunsOptions): Promise<{
+    runs: ListedRun[];
+    pagination: {
+      nextCursor: string | null;
+      previousCursor: string | null;
+    };
+  }>;
+  countRuns(options: RunListInputOptions): Promise<number>;
+}
+
+export class RunsRepository implements IRunsRepository {
+  private readonly clickHouseRunsRepository: ClickHouseRunsRepository;
+  constructor(private readonly options: RunsRepositoryOptions) {
+    this.clickHouseRunsRepository = new ClickHouseRunsRepository(options);
+  }
+
+  listRunIds(options: ListRunsOptions): Promise<string[]> {
+    return this.clickHouseRunsRepository.listRunIds(options);
+  }
+
+  listRuns(options: ListRunsOptions): Promise<{
+    runs: ListedRun[];
+    pagination: {
+      nextCursor: string | null;
+      previousCursor: string | null;
+    };
+  }> {
+    return this.clickHouseRunsRepository.listRuns(options);
+  }
+
+  countRuns(options: RunListInputOptions): Promise<number> {
+    return this.clickHouseRunsRepository.countRuns(options);
+  }
+}
+
+export function parseRunListInputOptions(data: any): RunListInputOptions {
+  return RunListInputOptionsSchema.parse(data);
+}

apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts

@@ -12,7 +12,7 @@ import {
   parseRunListInputOptions,
   type RunListInputFilters,
   RunsRepository,
-} from "~/services/runsRepository.server";
+} from "~/services/runsRepository/runsRepository.server";
 import { BaseService } from "../baseService.server";
 import { commonWorker } from "~/v3/commonWorker.server";
 import { env } from "~/env.server";

apps/webapp/test/runsRepository.test.ts

@@ -1,6 +1,6 @@
 import { containerTest } from "@internal/testcontainers";
 import { setTimeout } from "node:timers/promises";
-import { RunsRepository } from "~/services/runsRepository.server";
+import { RunsRepository } from "~/services/runsRepository/runsRepository.server";
 import { setupClickhouseReplication } from "./utils/replicationUtils";
 
 vi.setConfig({ testTimeout: 60_000 });