Realtime and task run performance improvements (#2158)

* Add createdAt filter to realtime subscribing with tags

* Filter realtime colums and expose ability to skip some columns

* Add sharding support for electric

* Use unkey cache for the created at filter caching

* Remove 2 unused indexes on TaskRun

* Run list now filters by a single runtime environment

* Remove project ID indexes

* Use clickhouse in task list aggregation queries instead of pg (keep pg for self-hosters)

* WIP clickhouse powered runs list
stuff

* Improve the query to get the latest tasks for the task list presenter

* Update the usage task list to use clickhouse

* Implement next runs list powered by clickhouse

* Add new index for TaskRun for the runs list, by environment ID

* Add runTags gin index

* Handle possibly malicious inputs

* Ignore claude settings

* Better handling not finding an environment on the schedule page

* Use ms since epoch in test, not seconds

* Remove unused function

* Fix test

* Use an env var for the realtime maximum createdAt filter duration (defaults to 1 day)

* Fixed the query builder to correct the group by / order by order

* Make sure runs.list still works

* Create small-birds-arrive.md

Author: ericallam

.changeset/small-birds-arrive.md

@@ -0,0 +1,16 @@
+---
+"@trigger.dev/react-hooks": patch
+---
+
+Added the ability to specify a "createdAt" filter when subscribing to tags in our useRealtime hooks:
+
+```tsx
+// Only subscribe to runs created in the last 10 hours
+useRealtimeRunWithTags("my-tag", { createdAt: "10h" })
+```
+
+You can also now choose to skip subscribing to specific columns by specifying the `skipColumns` option:
+
+```tsx
+useRealtimeRun(run.id, { skipColumns: ["usageDurationMs"] });
+```

.gitignore

@@ -63,3 +63,4 @@ apps/**/public/build
 /packages/core/src/package.json
 /packages/trigger-sdk/src/package.json
 /packages/python/src/package.json
+.claude

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

@@ -539,7 +539,7 @@ function BlankState({ isLoading, filters }: Pick<RunsTableProps, "isLoading" | "
   const environment = useEnvironment();
   if (isLoading) return <TableBlankRow colSpan={15}></TableBlankRow>;
 
-  const { environments, tasks, from, to, ...otherFilters } = filters;
+  const { tasks, from, to, ...otherFilters } = filters;
 
   if (
     filters.tasks.length === 1 &&

apps/webapp/app/env.server.ts

@@ -35,6 +35,9 @@ const EnvironmentSchema = z.object({
   API_ORIGIN: z.string().optional(),
   STREAM_ORIGIN: z.string().optional(),
   ELECTRIC_ORIGIN: z.string().default("http://localhost:3060"),
+  // A comma separated list of electric origins to shard into different electric instances by environmentId
+  // example: "http://localhost:3060,http://localhost:3061,http://localhost:3062"
+  ELECTRIC_ORIGIN_SHARDS: z.string().optional(),
   APP_ENV: z.string().default(process.env.NODE_ENV),
   SERVICE_NAME: z.string().default("trigger.dev webapp"),
   POSTHOG_PROJECT_KEY: z.string().default("phc_LFH7kJiGhdIlnO22hTAKgHpaKhpM8gkzWAFvHmf5vfS"),
@@ -161,6 +164,11 @@ const EnvironmentSchema = z.object({
     .default(process.env.REDIS_TLS_DISABLED ?? "false"),
   REALTIME_STREAMS_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
 
+  REALTIME_MAXIMUM_CREATED_AT_FILTER_AGE_IN_MS: z.coerce
+    .number()
+    .int()
+    .default(24 * 60 * 60 * 1000), // 1 day in milliseconds
+
   PUBSUB_REDIS_HOST: z
     .string()
     .optional()
@@ -738,6 +746,14 @@ const EnvironmentSchema = z.object({
   RUN_REPLICATION_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
   RUN_REPLICATION_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
 
+  // Clickhouse
+  CLICKHOUSE_URL: z.string().optional(),
+  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),
+  CLICKHOUSE_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
+  CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),
+
   // Bootstrap
   TRIGGER_BOOTSTRAP_ENABLED: z.string().default("0"),
   TRIGGER_BOOTSTRAP_WORKER_GROUP_NAME: z.string().optional(),

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

@@ -275,3 +275,36 @@ export function displayableEnvironment(
     userName,
   };
 }
+
+export async function findDisplayableEnvironment(
+  environmentId: string,
+  userId: string | undefined
+) {
+  const environment = await prisma.runtimeEnvironment.findFirst({
+    where: {
+      id: environmentId,
+    },
+    select: {
+      id: true,
+      type: true,
+      slug: true,
+      orgMember: {
+        select: {
+          user: {
+            select: {
+              id: true,
+              name: true,
+              displayName: true,
+            },
+          },
+        },
+      },
+    },
+  });
+
+  if (!environment) {
+    return;
+  }
+
+  return displayableEnvironment(environment, userId);
+}

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

@@ -7,7 +7,7 @@ import { PrismaClientOrTransaction, sqlDatabaseSchema } from "~/db.server";
  * It has indexes for fast performance.
  * It does NOT care about versions, so includes all tasks ever created.
  */
-export function getAllTaskIdentifiers(prisma: PrismaClientOrTransaction, projectId: string) {
+export function getAllTaskIdentifiers(prisma: PrismaClientOrTransaction, environmentId: string) {
   return prisma.$queryRaw<
     {
       slug: string;
@@ -16,6 +16,6 @@ export function getAllTaskIdentifiers(prisma: PrismaClientOrTransaction, project
   >`
     SELECT DISTINCT(slug), "triggerSource"
     FROM ${sqlDatabaseSchema}."BackgroundWorkerTask"
-    WHERE "projectId" = ${projectId}
+    WHERE "runtimeEnvironmentId" = ${environmentId}
     ORDER BY slug ASC;`;
 }

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

@@ -12,6 +12,7 @@ import { CoercedDate } from "~/utils/zod";
 import { ApiRetrieveRunPresenter } from "./ApiRetrieveRunPresenter.server";
 import { type RunListOptions, RunListPresenter } from "./RunListPresenter.server";
 import { BasePresenter } from "./basePresenter.server";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
 
 export const ApiRunListSearchParams = z.object({
   "page[size]": z.coerce.number().int().positive().min(1).max(100).optional(),
@@ -134,9 +135,11 @@ export class ApiRunListPresenter extends BasePresenter {
         options.direction = "backward";
       }
 
+      let environmentId: string | undefined;
+
       // filters
       if (environment) {
-        options.environments = [environment.id];
+        environmentId = environment.id;
       } else {
         if (searchParams["filter[env]"]) {
           const environments = await this._prisma.runtimeEnvironment.findMany({
@@ -148,10 +151,14 @@ export class ApiRunListPresenter extends BasePresenter {
             },
           });
 
-          options.environments = environments.map((env) => env.id);
+          environmentId = environments.at(0)?.id;
         }
       }
 
+      if (!environmentId) {
+        throw new ServiceValidationError("No environment found");
+      }
+
       if (searchParams["filter[status]"]) {
         options.statuses = searchParams["filter[status]"].flatMap((status) =>
           ApiRunListPresenter.apiStatusToRunStatuses(status)
@@ -202,9 +209,9 @@ export class ApiRunListPresenter extends BasePresenter {
 
       logger.debug("Calling RunListPresenter", { options });
 
-      const results = await presenter.call(options);
+      const results = await presenter.call(environmentId, options);
 
-      logger.debug("RunListPresenter results", { results });
+      logger.debug("RunListPresenter results", { runs: results.runs.length });
 
       const data: ListRunResponseItem[] = await Promise.all(
         results.runs.map(async (run) => {