Restructure the run events to be more sparse

Author: ericallam

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.next.runs._index/route.tsx

@@ -0,0 +1,10 @@
+import { Outlet } from "@remix-run/react";
+import { PageContainer } from "~/components/layout/AppLayout";
+
+export default function Page() {
+  return (
+    <PageContainer>
+      <Outlet />
+    </PageContainer>
+  );
+}

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

@@ -4,8 +4,7 @@ import { z } from "zod";
 import { prisma } from "~/db.server";
 import { createTag, getTagsForRunId, MAX_TAGS_PER_RUN } from "~/models/taskRunTag.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
-import { emitRunStatusUpdate } from "~/services/runsDashboardInstance.server";
-import { generateFriendlyId } from "~/v3/friendlyIdentifiers";
+import { emitRunTagsUpdated } from "~/services/runsDashboardInstance.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
@@ -96,7 +95,24 @@ export async function action({ request, params }: ActionFunctionArgs) {
       },
     });
 
-    emitRunStatusUpdate(taskRun.id);
+    emitRunTagsUpdated({
+      time: new Date(),
+      run: {
+        id: taskRun.id,
+        tags: taskRun.runTags,
+        status: taskRun.status,
+        updatedAt: taskRun.updatedAt,
+      },
+      organization: {
+        id: authenticationResult.environment.organizationId,
+      },
+      project: {
+        id: authenticationResult.environment.projectId,
+      },
+      environment: {
+        id: authenticationResult.environment.id,
+      },
+    });
 
     return json({ message: `Successfully set ${newTags.length} new tags.` }, { status: 200 });
   } catch (error) {

apps/webapp/app/routes/api.v1.runs.$runId.tags.ts

@@ -7,7 +7,6 @@ import {
   TriggerTaskRequestBody,
 } from "@trigger.dev/core/v3";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
-import type { TaskRunStatusUpdateEnvironment } from "~/services/runsDashboardInstance.server";
 
 export type TriggerTaskServiceOptions = {
   idempotencyKey?: string;

apps/webapp/app/runEngine/types.ts

@@ -6,80 +6,263 @@ import { engine } from "~/v3/runEngine.server";
 import { logger } from "./logger.server";
 import {
   RunDashboardEventBus,
+  RunDashboardEventRunAttemptStarted,
+  RunDashboardEventRunCancelled,
+  RunDashboardEventRunDelayRescheduled,
+  RunDashboardEventRunEnqueuedAfterDelay,
+  RunDashboardEventRunExpired,
+  RunDashboardEventRunFailed,
+  RunDashboardEventRunLocked,
+  RunDashboardEventRunRetryScheduled,
+  RunDashboardEventRunStatusChanged,
+  RunDashboardEventRunSucceeded,
+  RunDashboardEventRunTagsUpdated,
   RunDashboardEvents,
   RunsDashboardService,
 } from "./runsDashboardService.server";
 import { tryCatch } from "@trigger.dev/core/utils";
 
 const runDashboardEventBus: RunDashboardEventBus = new EventEmitter<RunDashboardEvents>();
 
-export function emitRunStatusUpdate(runId: string) {
-  runDashboardEventBus.emit("runStatusUpdate", {
-    time: new Date(),
+export function emitRunStatusChanged(event: RunDashboardEventRunStatusChanged) {
+  runDashboardEventBus.emit("runStatusChanged", event);
+}
+
+export function emitRunCreated(time: Date, runId: string) {
+  runDashboardEventBus.emit("runCreated", {
+    time,
     runId,
   });
 }
 
+export function emitRunAttemptStarted(event: RunDashboardEventRunAttemptStarted) {
+  runDashboardEventBus.emit("runAttemptStarted", event);
+}
+
+export function emitRunFailed(event: RunDashboardEventRunFailed) {
+  runDashboardEventBus.emit("runFailed", event);
+}
+
+export function emitRunSucceeded(event: RunDashboardEventRunSucceeded) {
+  runDashboardEventBus.emit("runSucceeded", event);
+}
+
+export function emitRunCancelled(event: RunDashboardEventRunCancelled) {
+  runDashboardEventBus.emit("runCancelled", event);
+}
+
+export function emitRunRetryScheduled(event: RunDashboardEventRunRetryScheduled) {
+  runDashboardEventBus.emit("runRetryScheduled", event);
+}
+
+export function emitRunDelayRescheduled(event: RunDashboardEventRunDelayRescheduled) {
+  runDashboardEventBus.emit("runDelayRescheduled", event);
+}
+
+export function emitRunLocked(event: RunDashboardEventRunLocked) {
+  runDashboardEventBus.emit("runLocked", event);
+}
+
+export function emitRunExpired(event: RunDashboardEventRunExpired) {
+  runDashboardEventBus.emit("runExpired", event);
+}
+
+export function emitRunTagsUpdated(event: RunDashboardEventRunTagsUpdated) {
+  runDashboardEventBus.emit("runTagsUpdated", event);
+}
+
+export function emitRunEnqueuedAfterDelay(event: RunDashboardEventRunEnqueuedAfterDelay) {
+  runDashboardEventBus.emit("runEnqueuedAfterDelay", event);
+}
+
 export const runsDashboard = singleton("runsDashboard", () => {
   const clickhouse = ClickHouse.fromEnv();
 
   const service = new RunsDashboardService(clickhouse);
 
-  runDashboardEventBus.on("runStatusUpdate", async (event) => {
-    const [upsertError] = await tryCatch(upsertRun(event.time, event.runId, service));
+  runDashboardEventBus.on("runCreated", async (event) => {
+    const [runCreatedError] = await tryCatch(runCreated(event.time, event.runId, service));
 
-    if (upsertError) {
-      logger.error("RunDashboard: runStatusUpdate: upsertRun error", {
+    if (runCreatedError) {
+      logger.error("RunDashboard: runCreated: runCreated error", {
         runId: event.runId,
-        error: upsertError,
+        error: runCreatedError,
       });
     }
   });
 
-  engine.eventBus.on("runStatusChanged", async (event) => {
-    const [upsertError] = await tryCatch(upsertRun(event.time, event.runId, service));
+  runDashboardEventBus.on("runAttemptStarted", async (event) => {
+    const [runAttemptStartedError] = await tryCatch(service.runAttemptStarted(event));
 
-    if (upsertError) {
-      logger.error("RunDashboard: runStatusChanged: upsertRun error", {
-        runId: event.runId,
-        error: upsertError,
+    if (runAttemptStartedError) {
+      logger.error("RunDashboard: runAttemptStarted: runAttemptStarted error", {
+        runId: event.run.id,
+        error: runAttemptStartedError,
       });
     }
   });
 
+  runDashboardEventBus.on("runStatusChanged", async (event) => {
+    const [runStatusChangedError] = await tryCatch(service.runStatusChanged(event));
+
+    if (runStatusChangedError) {
+      logger.error("RunDashboard: runStatusChanged: runStatusChanged error", {
+        runId: event.run.id,
+        error: runStatusChangedError,
+      });
+    }
+  });
+
+  runDashboardEventBus.on("runFailed", async (event) => {
+    const [runFailedError] = await tryCatch(service.runFailed(event));
+
+    if (runFailedError) {
+      logger.error("RunDashboard: runFailed: runFailed error", {
+        runId: event.run.id,
+        error: runFailedError,
+      });
+    }
+  });
+
+  runDashboardEventBus.on("runSucceeded", async (event) => {
+    const [runSucceededError] = await tryCatch(service.runSucceeded(event));
+
+    if (runSucceededError) {
+      logger.error("RunDashboard: runSucceeded: runSucceeded error", {
+        runId: event.run.id,
+        error: runSucceededError,
+      });
+    }
+  });
+
+  runDashboardEventBus.on("runCancelled", async (event) => {
+    const [runCancelledError] = await tryCatch(service.runCancelled(event));
+
+    if (runCancelledError) {
+      logger.error("RunDashboard: runCancelled: runCancelled error", {
+        runId: event.run.id,
+        error: runCancelledError,
+      });
+    }
+  });
+
+  runDashboardEventBus.on("runRetryScheduled", async (event) => {
+    const [runRetryScheduledError] = await tryCatch(service.runRetryScheduled(event));
+
+    if (runRetryScheduledError) {
+      logger.error("RunDashboard: runRetryScheduled: runRetryScheduled error", {
+        runId: event.run.id,
+        error: runRetryScheduledError,
+      });
+    }
+  });
+
+  runDashboardEventBus.on("runDelayRescheduled", async (event) => {
+    const [runDelayRescheduledError] = await tryCatch(service.runDelayRescheduled(event));
+
+    if (runDelayRescheduledError) {
+      logger.error("RunDashboard: runDelayRescheduled: runDelayRescheduled error", {
+        runId: event.run.id,
+        error: runDelayRescheduledError,
+      });
+    }
+  });
+
+  runDashboardEventBus.on("runLocked", async (event) => {
+    const [runLockedError] = await tryCatch(service.runLocked(event));
+
+    if (runLockedError) {
+      logger.error("RunDashboard: runLocked: runLocked error", {
+        runId: event.run.id,
+        error: runLockedError,
+      });
+    }
+  });
+
+  runDashboardEventBus.on("runExpired", async (event) => {
+    const [runExpiredError] = await tryCatch(service.runExpired(event));
+
+    if (runExpiredError) {
+      logger.error("RunDashboard: runExpired: runExpired error", {
+        runId: event.run.id,
+        error: runExpiredError,
+      });
+    }
+  });
+
+  engine.eventBus.on("runCreated", async (event) => {
+    runDashboardEventBus.emit("runCreated", event);
+  });
+
+  engine.eventBus.on("runAttemptStarted", async (event) => {
+    runDashboardEventBus.emit("runAttemptStarted", event);
+  });
+
+  engine.eventBus.on("runStatusChanged", async (event) => {
+    runDashboardEventBus.emit("runStatusChanged", event);
+  });
+
+  engine.eventBus.on("runFailed", async (event) => {
+    runDashboardEventBus.emit("runFailed", event);
+  });
+
+  engine.eventBus.on("runSucceeded", async (event) => {
+    runDashboardEventBus.emit("runSucceeded", event);
+  });
+
+  engine.eventBus.on("runCancelled", async (event) => {
+    runDashboardEventBus.emit("runCancelled", event);
+  });
+
+  engine.eventBus.on("runRetryScheduled", async (event) => {
+    runDashboardEventBus.emit("runRetryScheduled", event);
+  });
+
+  engine.eventBus.on("runDelayRescheduled", async (event) => {
+    runDashboardEventBus.emit("runDelayRescheduled", event);
+  });
+
+  engine.eventBus.on("runLocked", async (event) => {
+    runDashboardEventBus.emit("runLocked", event);
+  });
+
+  engine.eventBus.on("runExpired", async (event) => {
+    runDashboardEventBus.emit("runExpired", event);
+  });
+
   return service;
 });
 
-async function upsertRun(time: Date, runId: string, service: RunsDashboardService) {
+async function runCreated(time: Date, runId: string, service: RunsDashboardService) {
   const run = await prisma.taskRun.findFirst({
     where: {
       id: runId,
     },
   });
 
   if (!run) {
-    logger.error("RunDashboard: upsertRun: run not found", {
+    logger.error("RunDashboard: runCreated: run not found", {
       runId,
     });
 
     return;
   }
 
   if (!run.environmentType) {
-    logger.error("RunDashboard: upsertRun: run environment type not found", {
+    logger.error("RunDashboard: runCreated: run environment type not found", {
       runId,
     });
 
     return;
   }
 
   if (!run.organizationId) {
-    logger.error("RunDashboard: upsertRun: run organization id not found", {
+    logger.error("RunDashboard: runCreated: run organization id not found", {
       runId,
     });
 
     return;
   }
 
-  await service.upsertRun(time, run, run.environmentType, run.organizationId);
+  await service.runCreated(time, run, run.environmentType, run.organizationId);
 }