Tags listing now uses ClickHouse

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

@@ -58,10 +58,10 @@ 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";
+import { appliedSummary, FilterMenuProvider, TimeFilter, timeFilters } from "./SharedFilters";
 import { AIFilterInput } from "./AIFilterInput";
 import {
   allTaskRunStatuses,
@@ -71,6 +71,7 @@ import {
   TaskRunStatusCombo,
 } from "./TaskRunStatus";
 import { TaskTriggerSourceIcon } from "./TaskTriggerSource";
+import { environment } from "effect/Differ";
 
 export const RunStatus = z.enum(allTaskRunStatuses);
 
@@ -810,8 +811,8 @@ function TagsDropdown({
   searchValue: string;
   onClose?: () => void;
 }) {
-  const project = useProject();
-  const { values, replace } = useSearchParams();
+  const environment = useEnvironment();
+  const { values, value, replace } = useSearchParams();
 
   const handleChange = (values: string[]) => {
     clearSearchValue();
@@ -822,6 +823,12 @@ function TagsDropdown({
     });
   };
 
+  const { period, from, to } = timeFilters({
+    period: value("period"),
+    from: value("from"),
+    to: value("to"),
+  });
+
   const tagValues = values("tags").filter((v) => v !== "");
   const selected = tagValues.length > 0 ? tagValues : undefined;
 
@@ -832,8 +839,17 @@ function TagsDropdown({
     if (searchValue) {
       searchParams.set("name", encodeURIComponent(searchValue));
     }
-    fetcher.load(`/resources/projects/${project.slug}/runs/tags?${searchParams}`);
-  }, [searchValue]);
+    if (period) {
+      searchParams.set("period", period);
+    }
+    if (from) {
+      searchParams.set("from", from.getTime().toString());
+    }
+    if (to) {
+      searchParams.set("to", to.getTime().toString());
+    }
+    fetcher.load(`/resources/environments/${environment.id}/runs/tags?${searchParams}`);
+  }, [searchValue, period, from?.getTime(), to?.getTime()]);
 
   const filtered = useMemo(() => {
     let items: string[] = [];
@@ -845,7 +861,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,16 @@
+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";
+import { timeFilters } from "~/components/runs/v3/SharedFilters";
 
 export type TagListOptions = {
-  userId?: string;
+  organizationId: string;
+  environmentId: string;
   projectId: string;
+  period?: string;
+  from?: Date;
+  to?: Date;
   //filters
   name?: string;
   //pagination
@@ -17,40 +25,39 @@ export type TagListItem = TagList["tags"][number];
 
 export class RunTagListPresenter extends BasePresenter {
   public async call({
-    userId,
+    organizationId,
+    environmentId,
     projectId,
     name,
+    period,
+    from,
+    to,
     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,
+    });
+
+    const tags = await runsRepository.listTags({
+      organizationId,
+      projectId,
+      environmentId,
+      query: name,
+      period,
+      from: from ? from.getTime() : undefined,
+      to: to ? to.getTime() : undefined,
+      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,67 @@
+import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { timeFilters } from "~/components/runs/v3/SharedFilters";
+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 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 { period, from, to } = timeFilters({
+    period: parsedSearchParams.data.period,
+    from: parsedSearchParams.data.from,
+    to: parsedSearchParams.data.to,
+  });
+
+  const presenter = new RunTagListPresenter();
+  const result = await presenter.call({
+    environmentId: environment.id,
+    projectId: environment.projectId,
+    organizationId: environment.organizationId,
+    name: parsedSearchParams.data.name,
+    period,
+    from,
+    to,
+  });
+  return result;
+}

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.ai-filter.tsx

@@ -65,13 +65,16 @@ export async function action({ request, params }: ActionFunctionArgs) {
     query: async (search) => {
       const tagPresenter = new RunTagListPresenter();
       const tags = await tagPresenter.call({
+        organizationId: environment.organizationId,
         projectId: environment.projectId,
+        environmentId: environment.id,
         name: search,
         page: 1,
         pageSize: 50,
+        period: "1y",
       });
       return {
-        tags: tags.tags.map((t) => t.name),
+        tags: tags.tags,
       };
     },
   };

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

@@ -6,8 +6,10 @@ import {
   type ListRunsOptions,
   type RunListInputOptions,
   type RunsRepositoryOptions,
+  type TagListOptions,
   convertRunListInputOptionsToFilterRunsOptions,
 } from "./runsRepository.server";
+import parseDuration from "parse-duration";
 
 export class ClickHouseRunsRepository implements IRunsRepository {
   constructor(private readonly options: RunsRepositoryOptions) {}
@@ -162,6 +164,53 @@ 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 periodMs = options.period ? parseDuration(options.period) ?? undefined : undefined;
+    if (periodMs) {
+      queryBuilder.where("created_at >= fromUnixTimestamp64Milli({period: Int64})", {
+        period: new Date(Date.now() - periodMs).getTime(),
+      });
+    }
+
+    if (options.from) {
+      queryBuilder.where("created_at >= fromUnixTimestamp64Milli({from: Int64})", {
+        from: options.from,
+      });
+    }
+
+    if (options.to) {
+      queryBuilder.where("created_at <= fromUnixTimestamp64Milli({to: Int64})", { to: options.to });
+    }
+
+    const [queryError, result] = await queryBuilder.execute();
+
+    if (queryError) {
+      throw queryError;
+    }
+
+    if (result.length === 0) {
+      return {
+        tags: [],
+      };
+    }
+
+    return {
+      tags: result.flatMap((row) => row.tags),
+    };
+  }
 }
 
 function applyRunFiltersToQueryBuilder<T>(

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

@@ -8,6 +8,7 @@ import {
   type ListedRun,
   type RunListInputOptions,
   type RunsRepositoryOptions,
+  type 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" }