feat(server): add TaskIdentifier registry to replace expensive distinct query

.server-changes/task-identifier-registry.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+Replace the expensive DISTINCT query for task filter dropdowns with a dedicated TaskIdentifier registry table backed by Redis. Environments migrate automatically on their next deploy, with a transparent fallback to the legacy query for unmigrated environments. Also fixes duplicate dropdown entries when a task changes trigger source, and adds active/archived grouping for removed tasks. Moves BackgroundWorkerTask reads in the trigger hot path to the read replica.

apps/webapp/app/components/logs/LogsTaskFilter.tsx

@@ -4,6 +4,8 @@ import { useMemo } from "react";
 import * as Ariakit from "@ariakit/react";
 import {
   ComboBox,
+  SelectGroup,
+  SelectGroupLabel,
   SelectItem,
   SelectList,
   SelectPopover,
@@ -21,6 +23,7 @@ const shortcut = { key: "t" };
 type TaskOption = {
   slug: string;
   triggerSource: TaskTriggerSource;
+  isInLatestDeployment: boolean;
 };
 
 interface LogsTaskFilterProps {
@@ -126,17 +129,42 @@ function TasksDropdown({
       >
         <ComboBox placeholder={"Filter by task..."} value={searchValue} />
         <SelectList>
-          {filtered.map((item, index) => (
-            <SelectItem
-              key={`${item.triggerSource}-${item.slug}`}
-              value={item.slug}
-              icon={
-                <TaskTriggerSourceIcon source={item.triggerSource} className="size-4 flex-none" />
-              }
-            >
-              {item.slug}
-            </SelectItem>
-          ))}
+          {filtered
+            .filter((item) => item.isInLatestDeployment)
+            .map((item) => (
+              <SelectItem
+                key={item.slug}
+                value={item.slug}
+                icon={
+                  <TaskTriggerSourceIcon source={item.triggerSource} className="size-4 flex-none" />
+                }
+              >
+                {item.slug}
+              </SelectItem>
+            ))}
+          {filtered.some((item) => !item.isInLatestDeployment) && (
+            <SelectGroup>
+              <SelectGroupLabel>Archived</SelectGroupLabel>
+              {filtered
+                .filter((item) => !item.isInLatestDeployment)
+                .map((item) => (
+                  <SelectItem
+                    key={item.slug}
+                    value={item.slug}
+                    icon={
+                      <span className="opacity-50">
+                        <TaskTriggerSourceIcon
+                          source={item.triggerSource}
+                          className="size-4 flex-none"
+                        />
+                      </span>
+                    }
+                  >
+                    {item.slug}
+                  </SelectItem>
+                ))}
+            </SelectGroup>
+          )}
         </SelectList>
       </SelectPopover>
     </SelectProvider>

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

@@ -36,6 +36,8 @@ import { Paragraph } from "~/components/primitives/Paragraph";
 import {
   ComboBox,
   SelectButtonItem,
+  SelectGroup,
+  SelectGroupLabel,
   SelectItem,
   SelectList,
   SelectPopover,
@@ -322,7 +324,7 @@ export function getRunFiltersFromSearchParams(
 }
 
 type RunFiltersProps = {
-  possibleTasks: { slug: string; triggerSource: TaskTriggerSource }[];
+  possibleTasks: { slug: string; triggerSource: TaskTriggerSource; isInLatestDeployment: boolean }[];
   bulkActions: {
     id: string;
     type: BulkActionType;
@@ -627,7 +629,7 @@ function TasksDropdown({
   clearSearchValue: () => void;
   searchValue: string;
   onClose?: () => void;
-  possibleTasks: { slug: string; triggerSource: TaskTriggerSource }[];
+  possibleTasks: { slug: string; triggerSource: TaskTriggerSource; isInLatestDeployment: boolean }[];
 }) {
   const { values, replace } = useSearchParams();
 
@@ -658,17 +660,42 @@ function TasksDropdown({
       >
         <ComboBox placeholder={"Filter by task..."} value={searchValue} />
         <SelectList>
-          {filtered.map((item, index) => (
-            <SelectItem
-              key={`${item.triggerSource}-${item.slug}`}
-              value={item.slug}
-              icon={
-                <TaskTriggerSourceIcon source={item.triggerSource} className="size-4 flex-none" />
-              }
-            >
-              <MiddleTruncate text={item.slug} />
-            </SelectItem>
-          ))}
+          {filtered
+            .filter((item) => item.isInLatestDeployment)
+            .map((item) => (
+              <SelectItem
+                key={item.slug}
+                value={item.slug}
+                icon={
+                  <TaskTriggerSourceIcon source={item.triggerSource} className="size-4 flex-none" />
+                }
+              >
+                <MiddleTruncate text={item.slug} />
+              </SelectItem>
+            ))}
+          {filtered.some((item) => !item.isInLatestDeployment) && (
+            <SelectGroup>
+              <SelectGroupLabel>Archived</SelectGroupLabel>
+              {filtered
+                .filter((item) => !item.isInLatestDeployment)
+                .map((item) => (
+                  <SelectItem
+                    key={item.slug}
+                    value={item.slug}
+                    icon={
+                      <span className="opacity-50">
+                        <TaskTriggerSourceIcon
+                          source={item.triggerSource}
+                          className="size-4 flex-none"
+                        />
+                      </span>
+                    }
+                  >
+                    <MiddleTruncate text={item.slug} />
+                  </SelectItem>
+                ))}
+            </SelectGroup>
+          )}
         </SelectList>
       </SelectPopover>
     </SelectProvider>

apps/webapp/app/models/task.server.ts

@@ -1,6 +1,9 @@
 import type { TaskTriggerSource } from "@trigger.dev/database";
 import { PrismaClientOrTransaction, sqlDatabaseSchema } from "~/db.server";
 
+export { getTaskIdentifiers } from "~/services/taskIdentifierRegistry.server";
+export type { TaskIdentifierEntry } from "~/services/taskIdentifierCache.server";
+
 /**
  *
  * @param prisma An efficient query to get all task identifiers for a project.

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

@@ -13,7 +13,7 @@ import { type ErrorGroupStatus, type PrismaClientOrTransaction } from "@trigger.
 import { type Direction } from "~/components/ListPagination";
 import { timeFilterFromTo } from "~/components/runs/v3/SharedFilters";
 import { findDisplayableEnvironment } from "~/models/runtimeEnvironment.server";
-import { getAllTaskIdentifiers } from "~/models/task.server";
+import { getTaskIdentifiers } from "~/models/task.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { BasePresenter } from "~/presenters/v3/basePresenter.server";
 
@@ -170,7 +170,7 @@ export class ErrorsListPresenter extends BasePresenter {
       (search !== undefined && search !== "") ||
       (statuses !== undefined && statuses.length > 0);
 
-    const possibleTasksAsync = getAllTaskIdentifiers(this.replica, environmentId);
+    const possibleTasksAsync = getTaskIdentifiers(environmentId);
 
     // Pre-filter by status: since status lives in Postgres (ErrorGroupState) and the error
     // list comes from ClickHouse, we resolve inclusion/exclusion sets upfront so that

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

@@ -7,7 +7,7 @@ import parseDuration from "parse-duration";
 import { type Direction } from "~/components/ListPagination";
 import { timeFilterFromTo, timeFilters } from "~/components/runs/v3/SharedFilters";
 import { findDisplayableEnvironment } from "~/models/runtimeEnvironment.server";
-import { getAllTaskIdentifiers } from "~/models/task.server";
+import { getTaskIdentifiers } from "~/models/task.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { kindToLevel, type LogLevel, LogLevelSchema } from "~/utils/logUtils";
 import { BasePresenter } from "~/presenters/v3/basePresenter.server";
@@ -176,7 +176,7 @@ export class LogsListPresenter extends BasePresenter {
       (search !== undefined && search !== "") ||
       !time.isDefault;
 
-    const possibleTasksAsync = getAllTaskIdentifiers(this.replica, environmentId);
+    const possibleTasksAsync = getTaskIdentifiers(environmentId);
 
     const bulkActionsAsync = this.replica.bulkActionGroup.findMany({
       select: {
@@ -386,12 +386,7 @@ export class LogsListPresenter extends BasePresenter {
         next: nextCursor,
         previous: undefined, // For now, only support forward pagination
       },
-      possibleTasks: possibleTasks
-        .map((task) => ({
-          slug: task.slug,
-          triggerSource: task.triggerSource,
-        }))
-        .sort((a, b) => a.slug.localeCompare(b.slug)),
+      possibleTasks,
       bulkActions: bulkActions.map((bulkAction) => ({
         id: bulkAction.friendlyId,
         type: bulkAction.type,

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

@@ -8,7 +8,7 @@ import {
 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 { getTaskIdentifiers } from "~/models/task.server";
 import { RunsRepository } from "~/services/runsRepository/runsRepository.server";
 import { machinePresetFromRun } from "~/v3/machinePresets.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
@@ -105,7 +105,7 @@ export class NextRunListPresenter {
       !time.isDefault;
 
     //get all possible tasks
-    const possibleTasksAsync = getAllTaskIdentifiers(this.replica, environmentId);
+    const possibleTasksAsync = getTaskIdentifiers(environmentId);
 
     //get possible bulk actions
     const bulkActionsAsync = this.replica.bulkActionGroup.findMany({
@@ -256,11 +256,7 @@ export class NextRunListPresenter {
         next: pagination.nextCursor ?? undefined,
         previous: pagination.previousCursor ?? undefined,
       },
-      possibleTasks: possibleTasks
-        .map((task) => ({ slug: task.slug, triggerSource: task.triggerSource }))
-        .sort((a, b) => {
-          return a.slug.localeCompare(b.slug);
-        }),
+      possibleTasks,
       bulkActions: bulkActions.map((bulkAction) => ({
         id: bulkAction.friendlyId,
         type: bulkAction.type,

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

@@ -1,6 +1,7 @@
 import { type RuntimeEnvironmentType, type ScheduleType } from "@trigger.dev/database";
 import { type ScheduleListFilters } from "~/components/runs/v3/ScheduleFilters";
 import { displayableEnvironment } from "~/models/runtimeEnvironment.server";
+import { getTaskIdentifiers } from "~/models/task.server";
 import { getLimit } from "~/services/platform.v3.server";
 import { findCurrentWorkerFromEnvironment } from "~/v3/models/workerDeployment.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
@@ -123,14 +124,10 @@ export class ScheduleListPresenter extends BasePresenter {
     }
 
     //get all possible scheduled tasks
-    const possibleTasks = await this._replica.backgroundWorkerTask.findMany({
-      where: {
-        workerId: latestWorker.id,
-        projectId: project.id,
-        runtimeEnvironmentId: environmentId,
-        triggerSource: "SCHEDULED",
-      },
-    });
+    const allIdentifiers = await getTaskIdentifiers(environmentId);
+    const possibleTasks = allIdentifiers
+      .filter((t) => t.triggerSource === "SCHEDULED" && t.isInLatestDeployment)
+      .map((t) => ({ slug: t.slug }));
 
     //do this here to protect against SQL injection
     search = search && search !== "" ? `%${search}%` : undefined;
@@ -285,7 +282,7 @@ export class ScheduleListPresenter extends BasePresenter {
       totalPages: Math.ceil(totalCount / pageSize),
       totalCount: totalCount,
       schedules,
-      possibleTasks: possibleTasks.map((task) => task.slug).sort((a, b) => a.localeCompare(b)),
+      possibleTasks: possibleTasks.map((task) => task.slug),
       hasFilters,
       limits: {
         used: schedulesCount,

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

@@ -17,14 +17,13 @@ import { TitleWidget } from "~/components/metrics/TitleWidget";
 import { CreateDashboardPageButton } from "~/components/navigation/DashboardDialogs";
 import { NavBar, PageAccessories, PageTitle } from "~/components/primitives/PageHeader";
 import { TimeFilter } from "~/components/runs/v3/SharedFilters";
-import { $replica } from "~/db.server";
 import { useEnvironment } from "~/hooks/useEnvironment";
 import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
 import { useSearchParams } from "~/hooks/useSearchParam";
 import { findProjectBySlug } from "~/models/project.server";
 import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
-import { getAllTaskIdentifiers } from "~/models/task.server";
+import { getTaskIdentifiers } from "~/models/task.server";
 import {
   type BuiltInDashboardFilter,
   type LayoutItem,
@@ -70,7 +69,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
       organizationId: project.organizationId,
       key: dashboardKey,
     }),
-    getAllTaskIdentifiers($replica, environment.id),
+    getTaskIdentifiers(environment.id),
   ]);
 
   const filters = dashboard.filters ?? ["tasks", "queues"];
@@ -114,9 +113,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   return typedjson({
     ...dashboard,
     filters,
-    possibleTasks: possibleTasks
-      .map((task) => ({ slug: task.slug, triggerSource: task.triggerSource }))
-      .sort((a, b) => a.slug.localeCompare(b.slug)),
+    possibleTasks,
     possibleModels,
     possiblePrompts,
     possibleOperations,
@@ -201,7 +198,7 @@ export function MetricDashboard({
   /** Which filters to show. Defaults to ["tasks", "queues"]. */
   filters?: BuiltInDashboardFilter[];
   /** Possible tasks for filtering */
-  possibleTasks?: { slug: string; triggerSource: TaskTriggerSource }[];
+  possibleTasks?: { slug: string; triggerSource: TaskTriggerSource; isInLatestDeployment: boolean }[];
   /** Possible models for filtering */
   possibleModels?: ModelOption[];
   /** Possible prompt slugs for filtering */

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

@@ -35,7 +35,7 @@ import { Sheet, SheetContent } from "~/components/primitives/SheetV3";
 import { useToast } from "~/components/primitives/Toast";
 import { SimpleTooltip } from "~/components/primitives/Tooltip";
 import { QueryEditor, type QueryEditorSaveData } from "~/components/query/QueryEditor";
-import { $replica, prisma } from "~/db.server";
+import { prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { useDashboardEditor } from "~/hooks/useDashboardEditor";
 import { useEnvironment } from "~/hooks/useEnvironment";
@@ -44,7 +44,7 @@ import { useProject } from "~/hooks/useProject";
 import { redirectWithSuccessMessage } from "~/models/message.server";
 import { findProjectBySlug } from "~/models/project.server";
 import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
-import { getAllTaskIdentifiers } from "~/models/task.server";
+import { getTaskIdentifiers } from "~/models/task.server";
 import { MetricDashboardPresenter } from "~/presenters/v3/MetricDashboardPresenter.server";
 import { QueryPresenter } from "~/presenters/v3/QueryPresenter.server";
 import { requireUser, requireUserId } from "~/services/session.server";
@@ -93,7 +93,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     queryPresenter.call({
       organizationId: project.organizationId,
     }),
-    getAllTaskIdentifiers($replica, environment.id),
+    getTaskIdentifiers(environment.id),
   ]);
 
   // Admins and impersonating users can use EXPLAIN
@@ -109,9 +109,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     queryHistory: history,
     isAdmin,
     maxRows: env.QUERY_CLICKHOUSE_MAX_RETURNED_ROWS,
-    possibleTasks: possibleTasks
-      .map((task) => ({ slug: task.slug, triggerSource: task.triggerSource }))
-      .sort((a, b) => a.slug.localeCompare(b.slug)),
+    possibleTasks,
     widgetCount,
   });
 };

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

@@ -2,11 +2,10 @@ import { openai } from "@ai-sdk/openai";
 import { type ActionFunctionArgs, json } from "@remix-run/server-runtime";
 import { tryCatch } from "@trigger.dev/core";
 import { z } from "zod";
-import { $replica } from "~/db.server";
 import { env } from "~/env.server";
 import { findProjectBySlug } from "~/models/project.server";
 import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
-import { getAllTaskIdentifiers } from "~/models/task.server";
+import { getTaskIdentifiers } from "~/models/task.server";
 import { QueueListPresenter } from "~/presenters/v3/QueueListPresenter.server";
 import { RunTagListPresenter } from "~/presenters/v3/RunTagListPresenter.server";
 import { VersionListPresenter } from "~/presenters/v3/VersionListPresenter.server";
@@ -126,7 +125,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
 
   const queryTasks: QueryTasks = {
     query: async () => {
-      const tasks = await getAllTaskIdentifiers($replica, environment.id);
+      const tasks = await getTaskIdentifiers(environment.id);
       return {
         tasks,
       };