WIP using ClickHouse for the tags filter list

Author: matt-aitken

apps/webapp/app/components/runs/v3/RunFilters.tsx

@@ -58,7 +58,7 @@ import { useProject } from "~/hooks/useProject";
 import { useSearchParams } from "~/hooks/useSearchParam";
 import { type loader as queuesLoader } from "~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues";
 import { type loader as versionsLoader } from "~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.versions";
-import { type loader as tagsLoader } from "~/routes/resources.projects.$projectParam.runs.tags";
+import { type loader as tagsLoader } from "~/routes/resources.environments.$envId.runs.tags";
 import { Button } from "../../primitives/Buttons";
 import { BulkActionTypeCombo } from "./BulkAction";
 import { appliedSummary, FilterMenuProvider, TimeFilter } from "./SharedFilters";
@@ -71,6 +71,7 @@ import {
   TaskRunStatusCombo,
 } from "./TaskRunStatus";
 import { TaskTriggerSourceIcon } from "./TaskTriggerSource";
+import { environment } from "effect/Differ";
 
 export const RunStatus = z.enum(allTaskRunStatuses);
 
@@ -810,7 +811,7 @@ function TagsDropdown({
   searchValue: string;
   onClose?: () => void;
 }) {
-  const project = useProject();
+  const environment = useEnvironment();
   const { values, replace } = useSearchParams();
 
   const handleChange = (values: string[]) => {
@@ -832,7 +833,7 @@ function TagsDropdown({
     if (searchValue) {
       searchParams.set("name", encodeURIComponent(searchValue));
     }
-    fetcher.load(`/resources/projects/${project.slug}/runs/tags?${searchParams}`);
+    fetcher.load(`/resources/environments/${environment.id}/runs/tags?${searchParams}`);
   }, [searchValue]);
 
   const filtered = useMemo(() => {
@@ -845,7 +846,7 @@ function TagsDropdown({
       return matchSorter(items, searchValue);
     }
 
-    items.push(...fetcher.data.tags.map((t) => t.name));
+    items.push(...fetcher.data.tags);
 
     return matchSorter(Array.from(new Set(items)), searchValue);
   }, [searchValue, fetcher.data]);

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

@@ -1,8 +1,13 @@
+import { RunsRepository } from "~/services/runsRepository/runsRepository.server";
 import { BasePresenter } from "./basePresenter.server";
+import { clickhouseClient } from "~/services/clickhouseInstance.server";
+import { type PrismaClient } from "@trigger.dev/database";
 
 export type TagListOptions = {
-  userId?: string;
+  organizationId: string;
+  environmentId: string;
   projectId: string;
+  createdAfter?: Date;
   //filters
   name?: string;
   //pagination
@@ -17,40 +22,38 @@ export type TagListItem = TagList["tags"][number];
 
 export class RunTagListPresenter extends BasePresenter {
   public async call({
-    userId,
+    organizationId,
+    environmentId,
     projectId,
     name,
+    createdAfter,
     page = 1,
     pageSize = DEFAULT_PAGE_SIZE,
   }: TagListOptions) {
     const hasFilters = Boolean(name?.trim());
 
-    const tags = await this._replica.taskRunTag.findMany({
-      where: {
-        projectId,
-        name: name
-          ? {
-              startsWith: name,
-              mode: "insensitive",
-            }
-          : undefined,
-      },
-      orderBy: {
-        id: "desc",
-      },
-      take: pageSize + 1,
-      skip: (page - 1) * pageSize,
+    const runsRepository = new RunsRepository({
+      clickhouse: clickhouseClient,
+      prisma: this._replica as PrismaClient,
+    });
+
+    // Passed in or past 30 days
+    const createdAt = createdAfter ?? new Date(Date.now() - 30 * 24 * 60 * 60 * 1000);
+
+    const tags = await runsRepository.listTags({
+      organizationId,
+      projectId,
+      environmentId,
+      query: name,
+      createdAfter: createdAt,
+      offset: (page - 1) * pageSize,
+      limit: pageSize + 1,
     });
 
     return {
-      tags: tags
-        .map((tag) => ({
-          id: tag.friendlyId,
-          name: tag.name,
-        }))
-        .slice(0, pageSize),
+      tags: tags.tags,
       currentPage: page,
-      hasMore: tags.length > pageSize,
+      hasMore: tags.tags.length > pageSize,
       hasFilters,
     };
   }

apps/webapp/app/routes/resources.environments.$envId.runs.tags.tsx

@@ -0,0 +1,61 @@
+import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { RunTagListPresenter } from "~/presenters/v3/RunTagListPresenter.server";
+import { requireUserId } from "~/services/session.server";
+
+const Params = z.object({
+  envId: z.string(),
+});
+
+const SearchParams = z.object({
+  name: z.string().optional(),
+  period: z.preprocess((value) => (value === "all" ? undefined : value), z.string().optional()),
+  from: z.coerce.number().optional(),
+  to: z.coerce.number().optional(),
+});
+
+export async function loader({ request, params }: LoaderFunctionArgs) {
+  const userId = await requireUserId(request);
+  const { envId } = Params.parse(params);
+
+  const environment = await $replica.runtimeEnvironment.findFirst({
+    select: {
+      id: true,
+      projectId: true,
+      organizationId: true,
+    },
+    where: { id: envId, organization: { members: { some: { userId } } } },
+  });
+
+  if (!environment) {
+    throw new Response("Not Found", { status: 404 });
+  }
+
+  const search = new URL(request.url).searchParams;
+  const name = search.get("name");
+  const period = search.get("period");
+  const from = search.get("from");
+  const to = search.get("to");
+
+  const parsedSearchParams = SearchParams.safeParse({
+    name: name ? decodeURIComponent(name) : undefined,
+    period: search.get("period") ?? undefined,
+    from: search.get("from") ?? undefined,
+    to: search.get("to") ?? undefined,
+  });
+
+  if (!parsedSearchParams.success) {
+    throw new Response("Invalid search params", { status: 400 });
+  }
+
+  const presenter = new RunTagListPresenter();
+  const result = await presenter.call({
+    environmentId: environment.id,
+    projectId: environment.projectId,
+    organizationId: environment.organizationId,
+    name: parsedSearchParams.data.name,
+    createdAfter: parsedSearchParams.data.createdAfter,
+  });
+  return result;
+}

apps/webapp/app/routes/resources.projects.$projectParam.runs.tags.tsx

@@ -6,6 +6,7 @@ import {
   type ListRunsOptions,
   type RunListInputOptions,
   type RunsRepositoryOptions,
+  TagListOptions,
   convertRunListInputOptionsToFilterRunsOptions,
 } from "./runsRepository.server";
 
@@ -162,6 +163,36 @@ export class ClickHouseRunsRepository implements IRunsRepository {
 
     return result[0].count;
   }
+
+  async listTags(options: TagListOptions) {
+    const queryBuilder = this.options.clickhouse.taskRuns
+      .tagQueryBuilder()
+      .where("organization_id = {organizationId: String}", {
+        organizationId: options.organizationId,
+      })
+      .where("project_id = {projectId: String}", {
+        projectId: options.projectId,
+      })
+      .where("environment_id = {environmentId: String}", {
+        environmentId: options.environmentId,
+      });
+
+    const [queryError, result] = await queryBuilder.execute();
+
+    if (queryError) {
+      throw queryError;
+    }
+
+    if (result.length === 0) {
+      throw new Error("No count rows returned");
+    }
+
+    //todo process them
+
+    return {
+      tags: result[0].tags,
+    };
+  }
 }
 
 function applyRunFiltersToQueryBuilder<T>(

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

@@ -8,6 +8,7 @@ import {
   type ListedRun,
   type RunListInputOptions,
   type RunsRepositoryOptions,
+  TagListOptions,
   convertRunListInputOptionsToFilterRunsOptions,
 } from "./runsRepository.server";
 
@@ -104,6 +105,32 @@ export class PostgresRunsRepository implements IRunsRepository {
     return Number(result[0].count);
   }
 
+  async listTags({ projectId, query, offset, limit }: TagListOptions) {
+    const tags = await this.options.prisma.taskRunTag.findMany({
+      select: {
+        name: true,
+      },
+      where: {
+        projectId,
+        name: query
+          ? {
+              startsWith: query,
+              mode: "insensitive",
+            }
+          : undefined,
+      },
+      orderBy: {
+        id: "desc",
+      },
+      take: limit + 1,
+      skip: offset,
+    });
+
+    return {
+      tags: tags.map((tag) => tag.name),
+    };
+  }
+
   #buildRunIdsQuery(
     filterOptions: FilterRunsOptions,
     page: { size: number; cursor?: string; direction?: "forward" | "backward" }

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

@@ -69,6 +69,11 @@ type Pagination = {
   };
 };
 
+type OffsetPagination = {
+  offset: number;
+  limit: number;
+};
+
 export type ListedRun = Prisma.TaskRunGetPayload<{
   select: {
     id: true;
@@ -104,6 +109,19 @@ export type ListedRun = Prisma.TaskRunGetPayload<{
 
 export type ListRunsOptions = RunListInputOptions & Pagination;
 
+export type TagListOptions = {
+  organizationId: string;
+  projectId: string;
+  environmentId: string;
+  createdAfter: Date;
+  /** Performs a case insensitive contains search on the tag name */
+  query?: string;
+} & OffsetPagination;
+
+export type TagList = {
+  tags: string[];
+};
+
 export interface IRunsRepository {
   name: string;
   listRunIds(options: ListRunsOptions): Promise<string[]>;
@@ -115,6 +133,7 @@ export interface IRunsRepository {
     };
   }>;
   countRuns(options: RunListInputOptions): Promise<number>;
+  listTags(options: TagListOptions): Promise<TagList>;
 }
 
 export class RunsRepository implements IRunsRepository {
@@ -291,6 +310,13 @@ export class RunsRepository implements IRunsRepository {
       }
     );
   }
+
+  async listTags(options: TagListOptions): Promise<TagList> {
+    const repository = await this.#getRepository();
+    return startActiveSpan("runsRepository.listTags", async () => {
+      return await repository.listTags(options);
+    });
+  }
 }
 
 export function parseRunListInputOptions(data: any): RunListInputOptions {

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

@@ -11,6 +11,7 @@ import {
   getAverageDurations,
   getTaskUsageByOrganization,
   getTaskRunsCountQueryBuilder,
+  getTaskRunTagsQueryBuilder,
 } from "./taskRuns.js";
 import { Logger, type LogLevel } from "@trigger.dev/core/logger";
 import type { Agent as HttpAgent } from "http";
@@ -147,6 +148,7 @@ export class ClickHouse {
       insertPayloads: insertRawTaskRunPayloads(this.writer),
       queryBuilder: getTaskRunsQueryBuilder(this.reader),
       countQueryBuilder: getTaskRunsCountQueryBuilder(this.reader),
+      tagQueryBuilder: getTaskRunTagsQueryBuilder(this.reader),
       getTaskActivity: getTaskActivityQueryBuilder(this.reader),
       getCurrentRunningStats: getCurrentRunningStats(this.reader),
       getAverageDurations: getAverageDurations(this.reader),

internal-packages/clickhouse/src/index.ts

@@ -115,6 +115,17 @@ export function getTaskRunsCountQueryBuilder(ch: ClickhouseReader, settings?: Cl
   });
 }
 
+export function getTaskRunTagsQueryBuilder(ch: ClickhouseReader, settings?: ClickHouseSettings) {
+  return ch.queryBuilder({
+    name: "getTaskRunTags",
+    baseQuery: "SELECT DISTINCT tags FROM trigger_dev.task_runs_v2",
+    schema: z.object({
+      tags: z.array(z.string()),
+    }),
+    settings,
+  });
+}
+
 export const TaskActivityQueryResult = z.object({
   task_identifier: z.string(),
   status: z.string(),