emit more stuff

Author: ericallam

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

@@ -4,7 +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 { emitRunTagsUpdated } from "~/services/runsDashboardInstance.server";
+import { runsDashboard } from "~/services/runsDashboardInstance.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
@@ -95,13 +95,14 @@ export async function action({ request, params }: ActionFunctionArgs) {
       },
     });
 
-    emitRunTagsUpdated({
+    runsDashboard.emit.runTagsUpdated({
       time: new Date(),
       run: {
         id: taskRun.id,
         tags: taskRun.runTags,
         status: taskRun.status,
         updatedAt: taskRun.updatedAt,
+        createdAt: taskRun.createdAt,
       },
       organization: {
         id: authenticationResult.environment.organizationId,

apps/webapp/app/services/runsDashboardInstance.server.ts

@@ -24,54 +24,54 @@ import { tryCatch } from "@trigger.dev/core/utils";
 
 const runDashboardEventBus: RunDashboardEventBus = new EventEmitter<RunDashboardEvents>();
 
-export function emitRunStatusChanged(event: RunDashboardEventRunStatusChanged) {
+function emitRunStatusChanged(event: RunDashboardEventRunStatusChanged) {
   runDashboardEventBus.emit("runStatusChanged", event);
 }
 
-export function emitRunCreated(time: Date, runId: string) {
+function emitRunCreated(time: Date, runId: string) {
   runDashboardEventBus.emit("runCreated", {
     time,
     runId,
   });
 }
 
-export function emitRunAttemptStarted(event: RunDashboardEventRunAttemptStarted) {
+function emitRunAttemptStarted(event: RunDashboardEventRunAttemptStarted) {
   runDashboardEventBus.emit("runAttemptStarted", event);
 }
 
-export function emitRunFailed(event: RunDashboardEventRunFailed) {
+function emitRunFailed(event: RunDashboardEventRunFailed) {
   runDashboardEventBus.emit("runFailed", event);
 }
 
-export function emitRunSucceeded(event: RunDashboardEventRunSucceeded) {
+function emitRunSucceeded(event: RunDashboardEventRunSucceeded) {
   runDashboardEventBus.emit("runSucceeded", event);
 }
 
-export function emitRunCancelled(event: RunDashboardEventRunCancelled) {
+function emitRunCancelled(event: RunDashboardEventRunCancelled) {
   runDashboardEventBus.emit("runCancelled", event);
 }
 
-export function emitRunRetryScheduled(event: RunDashboardEventRunRetryScheduled) {
+function emitRunRetryScheduled(event: RunDashboardEventRunRetryScheduled) {
   runDashboardEventBus.emit("runRetryScheduled", event);
 }
 
-export function emitRunDelayRescheduled(event: RunDashboardEventRunDelayRescheduled) {
+function emitRunDelayRescheduled(event: RunDashboardEventRunDelayRescheduled) {
   runDashboardEventBus.emit("runDelayRescheduled", event);
 }
 
-export function emitRunLocked(event: RunDashboardEventRunLocked) {
+function emitRunLocked(event: RunDashboardEventRunLocked) {
   runDashboardEventBus.emit("runLocked", event);
 }
 
-export function emitRunExpired(event: RunDashboardEventRunExpired) {
+function emitRunExpired(event: RunDashboardEventRunExpired) {
   runDashboardEventBus.emit("runExpired", event);
 }
 
-export function emitRunTagsUpdated(event: RunDashboardEventRunTagsUpdated) {
+function emitRunTagsUpdated(event: RunDashboardEventRunTagsUpdated) {
   runDashboardEventBus.emit("runTagsUpdated", event);
 }
 
-export function emitRunEnqueuedAfterDelay(event: RunDashboardEventRunEnqueuedAfterDelay) {
+function emitRunEnqueuedAfterDelay(event: RunDashboardEventRunEnqueuedAfterDelay) {
   runDashboardEventBus.emit("runEnqueuedAfterDelay", event);
 }
 
@@ -230,7 +230,23 @@ export const runsDashboard = singleton("runsDashboard", () => {
     runDashboardEventBus.emit("runExpired", event);
   });
 
-  return service;
+  return {
+    service,
+    emit: {
+      runStatusChanged: emitRunStatusChanged,
+      runCreated: emitRunCreated,
+      runAttemptStarted: emitRunAttemptStarted,
+      runFailed: emitRunFailed,
+      runSucceeded: emitRunSucceeded,
+      runCancelled: emitRunCancelled,
+      runRetryScheduled: emitRunRetryScheduled,
+      runDelayRescheduled: emitRunDelayRescheduled,
+      runLocked: emitRunLocked,
+      runExpired: emitRunExpired,
+      runTagsUpdated: emitRunTagsUpdated,
+      runEnqueuedAfterDelay: emitRunEnqueuedAfterDelay,
+    },
+  };
 });
 
 async function runCreated(time: Date, runId: string, service: RunsDashboardService) {

apps/webapp/app/services/runsDashboardService.server.ts

@@ -23,6 +23,7 @@ export class RunsDashboardService {
       attempt: event.run.attemptNumber ?? 1,
       event_time: event.time.getTime(),
       updated_at: event.run.updatedAt.getTime(),
+      created_at: event.run.createdAt.getTime(),
       base_cost_in_cents: event.run.baseCostInCents,
       executed_at: event.run.executedAt ? event.run.executedAt.getTime() : undefined,
       event_name: "attempt_started",
@@ -47,6 +48,7 @@ export class RunsDashboardService {
       status: event.run.status,
       event_time: event.time.getTime(),
       updated_at: event.run.updatedAt.getTime(),
+      created_at: event.run.createdAt.getTime(),
       event_name: "enqueued_after_delay",
     });
 
@@ -69,6 +71,7 @@ export class RunsDashboardService {
       status: event.run.status,
       event_time: event.time.getTime(),
       updated_at: event.run.updatedAt.getTime(),
+      created_at: event.run.createdAt.getTime(),
       delay_until: event.run.delayUntil ? event.run.delayUntil.getTime() : undefined,
       event_name: "delay_rescheduled",
     });
@@ -92,6 +95,7 @@ export class RunsDashboardService {
       status: event.run.status,
       event_time: event.time.getTime(),
       updated_at: event.run.updatedAt.getTime(),
+      created_at: event.run.createdAt.getTime(),
       base_cost_in_cents: event.run.baseCostInCents,
       task_version: event.run.taskVersion ?? undefined,
       sdk_version: event.run.sdkVersion ?? undefined,
@@ -124,6 +128,7 @@ export class RunsDashboardService {
       status: event.run.status,
       event_time: event.time.getTime(),
       updated_at: event.run.updatedAt.getTime(),
+      created_at: event.run.createdAt.getTime(),
       event_name: "status_changed",
     });
 
@@ -146,6 +151,7 @@ export class RunsDashboardService {
       status: event.run.status,
       event_time: event.time.getTime(),
       updated_at: event.run.updatedAt.getTime(),
+      created_at: event.run.createdAt.getTime(),
       expired_at: event.run.expiredAt ? event.run.expiredAt.getTime() : undefined,
       event_name: "run_expired",
     });
@@ -171,6 +177,7 @@ export class RunsDashboardService {
       status: event.run.status,
       event_time: event.time.getTime(),
       updated_at: event.run.updatedAt.getTime(),
+      created_at: event.run.createdAt.getTime(),
       completed_at: event.run.completedAt ? event.run.completedAt.getTime() : undefined,
       usage_duration_ms: event.run.usageDurationMs,
       cost_in_cents: event.run.costInCents,
@@ -198,6 +205,7 @@ export class RunsDashboardService {
       status: event.run.status,
       event_time: event.time.getTime(),
       updated_at: event.run.updatedAt.getTime(),
+      created_at: event.run.createdAt.getTime(),
       completed_at: event.run.completedAt ? event.run.completedAt.getTime() : undefined,
       error: event.run.error,
       attempt: event.run.attemptNumber,
@@ -225,6 +233,7 @@ export class RunsDashboardService {
       status: event.run.status,
       event_time: event.time.getTime(),
       updated_at: event.run.updatedAt.getTime(),
+      created_at: event.run.createdAt.getTime(),
       machine_preset: event.run.nextMachineAfterOOM ?? undefined,
       attempt: event.run.attemptNumber,
       error: event.run.error,
@@ -250,6 +259,7 @@ export class RunsDashboardService {
       status: event.run.status,
       event_time: event.time.getTime(),
       updated_at: event.run.updatedAt.getTime(),
+      created_at: event.run.createdAt.getTime(),
       completed_at: event.run.completedAt ? event.run.completedAt.getTime() : undefined,
       error: event.run.error ? (event.run.error as TaskRunError) : undefined,
       attempt: event.run.attemptNumber,
@@ -275,6 +285,7 @@ export class RunsDashboardService {
       status: event.run.status,
       event_time: event.time.getTime(),
       updated_at: event.run.updatedAt.getTime(),
+      created_at: event.run.createdAt.getTime(),
       tags: event.run.tags,
       event_name: "tags_updated",
     });
@@ -439,6 +450,7 @@ export type RunDashboardEvents = {
         tags: string[];
         status: TaskRunStatus;
         updatedAt: Date;
+        createdAt: Date;
       };
       organization: {
         id: string;

apps/webapp/app/v3/marqs/devQueueConsumer.server.ts

@@ -16,7 +16,7 @@ import { findQueueInEnvironment, sanitizeQueueName } from "~/models/taskQueue.se
 import { RedisClient, createRedisClient } from "~/redis.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
-import { emitRunLocked } from "~/services/runsDashboardInstance.server";
+import { runsDashboard } from "~/services/runsDashboardInstance.server";
 import { marqs } from "~/v3/marqs/index.server";
 import { resolveVariablesForEnvironment } from "../environmentVariables/environmentVariablesRepository.server";
 import { FailedTaskRunService } from "../failedTaskRun.server";
@@ -543,11 +543,12 @@ export class DevQueueConsumer {
           messageId: message.messageId,
         });
 
-        emitRunLocked({
+        runsDashboard.emit.runLocked({
           time: new Date(),
           run: {
             id: lockedTaskRun.id,
             updatedAt: lockedTaskRun.updatedAt,
+            createdAt: lockedTaskRun.createdAt,
             status: lockedTaskRun.status,
             lockedAt,
             lockedById: backgroundTask.id,

apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts

@@ -37,7 +37,7 @@ import { findEnvironmentById } from "~/models/runtimeEnvironment.server";
 import { findQueueInEnvironment, sanitizeQueueName } from "~/models/taskQueue.server";
 import { generateJWTTokenForEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
-import { emitRunLocked, emitRunStatusChanged } from "~/services/runsDashboardInstance.server";
+import { runsDashboard } from "~/services/runsDashboardInstance.server";
 import { singleton } from "~/utils/singleton";
 import { marqs } from "~/v3/marqs/index.server";
 import {
@@ -930,12 +930,13 @@ export class SharedQueueConsumer {
           });
 
           if (lockedTaskRun.organizationId) {
-            emitRunLocked({
+            runsDashboard.emit.runLocked({
               time: new Date(),
               run: {
                 id: lockedTaskRun.id,
                 status: lockedTaskRun.status,
                 updatedAt: lockedTaskRun.updatedAt,
+                createdAt: lockedTaskRun.createdAt,
                 lockedAt,
                 lockedById: backgroundTask.id,
                 lockedToVersionId: worker.id,
@@ -1478,12 +1479,13 @@ export class SharedQueueConsumer {
     });
 
     if (run.organizationId) {
-      emitRunStatusChanged({
+      runsDashboard.emit.runStatusChanged({
         time: new Date(),
         run: {
           id: runId,
           status: "WAITING_FOR_DEPLOY",
           updatedAt: run.updatedAt,
+          createdAt: run.createdAt,
         },
         organization: {
           id: run.organizationId,

apps/webapp/app/v3/services/completeAttempt.server.ts

@@ -21,7 +21,7 @@ import { PrismaClientOrTransaction } from "~/db.server";
 import { env } from "~/env.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
-import { emitRunRetryScheduled } from "~/services/runsDashboardInstance.server";
+import { runsDashboard } from "~/services/runsDashboardInstance.server";
 import { safeJsonParse } from "~/utils/json";
 import { marqs } from "~/v3/marqs/index.server";
 import { createExceptionPropertiesFromError, eventRepository } from "../eventRepository.server";
@@ -617,7 +617,7 @@ export class CompleteAttemptService extends BaseService {
       },
     });
 
-    emitRunRetryScheduled({
+    runsDashboard.emit.runRetryScheduled({
       time: new Date(),
       run: {
         id: taskRunAttempt.taskRunId,
@@ -630,6 +630,7 @@ export class CompleteAttemptService extends BaseService {
         taskIdentifier: taskRunAttempt.taskRun.taskIdentifier,
         baseCostInCents: taskRunAttempt.taskRun.baseCostInCents,
         updatedAt: taskRunAttempt.taskRun.updatedAt,
+        createdAt: taskRunAttempt.taskRun.createdAt,
         error,
       },
       organization: {

apps/webapp/app/v3/services/createTaskRunAttempt.server.ts

@@ -6,7 +6,7 @@ import { findQueueInEnvironment } from "~/models/taskQueue.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { reportInvocationUsage } from "~/services/platform.v3.server";
-import { emitRunAttemptStarted } from "~/services/runsDashboardInstance.server";
+import { runsDashboard } from "~/services/runsDashboardInstance.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
 import { machinePresetFromConfig, machinePresetFromRun } from "../machinePresets.server";
 import { FINAL_RUN_STATUSES } from "../taskStatus";
@@ -182,7 +182,7 @@ export class CreateTaskRunAttemptService extends BaseService {
         });
       }
 
-      emitRunAttemptStarted({
+      runsDashboard.emit.runAttemptStarted({
         time: new Date(),
         run: {
           id: taskRun.id,

apps/webapp/app/v3/services/enqueueDelayedRun.server.ts

@@ -1,6 +1,6 @@
 import { parseNaturalLanguageDuration } from "@trigger.dev/core/v3/isomorphic";
 import { logger } from "~/services/logger.server";
-import { emitRunEnqueuedAfterDelay } from "~/services/runsDashboardInstance.server";
+import { runsDashboard } from "~/services/runsDashboardInstance.server";
 import { workerQueue } from "~/services/worker.server";
 import { commonWorker } from "../commonWorker.server";
 import { BaseService } from "./baseService.server";
@@ -100,13 +100,14 @@ export class EnqueueDelayedRunService extends BaseService {
     }
 
     if (run.organizationId) {
-      emitRunEnqueuedAfterDelay({
+      runsDashboard.emit.runEnqueuedAfterDelay({
         time: new Date(),
         run: {
           id: run.id,
           status: run.status,
           queuedAt: run.queuedAt ?? new Date(),
           updatedAt: run.updatedAt,
+          createdAt: run.createdAt,
         },
         organization: {
           id: run.organizationId,

apps/webapp/app/v3/services/executeTasksWaitingForDeploy.ts

@@ -1,7 +1,7 @@
 import { PrismaClientOrTransaction } from "~/db.server";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
-import { emitRunStatusChanged } from "~/services/runsDashboardInstance.server";
+import { runsDashboard } from "~/services/runsDashboardInstance.server";
 import { workerQueue } from "~/services/worker.server";
 import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
@@ -52,6 +52,8 @@ export class ExecuteTasksWaitingForDeployService extends BaseService {
         taskIdentifier: true,
         concurrencyKey: true,
         queue: true,
+        updatedAt: true,
+        createdAt: true,
       },
       take: maxCount + 1,
     });
@@ -80,12 +82,13 @@ export class ExecuteTasksWaitingForDeployService extends BaseService {
     }
 
     for (const run of runsWaitingForDeploy) {
-      emitRunStatusChanged({
+      runsDashboard.emit.runStatusChanged({
         time: new Date(),
         run: {
           id: run.id,
           status: run.status,
-          updatedAt: new Date(),
+          updatedAt: run.updatedAt,
+          createdAt: run.createdAt,
         },
         organization: {
           id: backgroundWorker.runtimeEnvironment.organizationId,