Realtime and task run performance improvements

.claude/settings.local.json

@@ -0,0 +1,13 @@
+{
+  "permissions": {
+    "allow": [
+      "Bash(grep:*)",
+      "Bash(rg:*)",
+      "Bash(find:*)",
+      "Bash(/Users/eric/.claude/local/node_modules/@anthropic-ai/claude-code/vendor/ripgrep/arm64-darwin/rg \"\\.findFirst|\\.findMany|\\.findUnique|\\.create|\\.update|\\.delete|\\.upsert|\\.count|\\.aggregate\" --type ts -A 3 -B 1)",
+      "Bash(/Users/eric/.claude/local/node_modules/@anthropic-ai/claude-code/vendor/ripgrep/arm64-darwin/rg \"taskRun\\.\" --type ts -B 2 -A 2)",
+      "WebFetch(domain:clickhouse.com)"
+    ],
+    "deny": []
+  }
+}

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"),
@@ -738,6 +741,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,7 +209,7 @@ 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 });