move v3 over to new methods

Author: ericallam

apps/webapp/app/v3/eventRepository.server.ts

@@ -4,11 +4,9 @@ import { SemanticResourceAttributes } from "@opentelemetry/semantic-conventions"
 import {
   AttemptFailedSpanEvent,
   correctErrorStackTrace,
-  createPacketAttributesAsJson,
   ExceptionEventProperties,
   ExceptionSpanEvent,
   flattenAttributes,
-  IOPacket,
   isExceptionSpanEvent,
   NULL_SENTINEL,
   omit,
@@ -23,14 +21,7 @@ import {
   unflattenAttributes,
 } from "@trigger.dev/core/v3";
 import { parseTraceparent, serializeTraceparent } from "@trigger.dev/core/v3/isomorphic";
-import {
-  Prisma,
-  RuntimeEnvironmentType,
-  TaskEvent,
-  TaskEventKind,
-  TaskEventStatus,
-  TaskRun,
-} from "@trigger.dev/database";
+import { Prisma, TaskEvent, TaskEventKind, TaskEventStatus, TaskRun } from "@trigger.dev/database";
 import { nanoid } from "nanoid";
 import { createHash } from "node:crypto";
 import { EventEmitter } from "node:stream";
@@ -316,66 +307,6 @@ export class EventRepository {
     return await this.#flushBatch(nanoid(), events);
   }
 
-  async completeEvent(
-    storeTable: TaskEventStoreTable,
-    spanId: string,
-    startCreatedAt: Date,
-    endCreatedAt?: Date,
-    options?: UpdateEventOptions
-  ) {
-    const events = await this.queryIncompleteEvents(
-      storeTable,
-      { spanId },
-      startCreatedAt,
-      endCreatedAt
-    );
-
-    if (events.length === 0) {
-      logger.warn("No incomplete events found for spanId", { spanId, options });
-      return;
-    }
-
-    const event = events[0];
-
-    const output = options?.attributes.output
-      ? await createPacketAttributesAsJson(
-          options?.attributes.output,
-          options?.attributes.outputType ?? "application/json"
-        )
-      : undefined;
-
-    logger.debug("Completing event", {
-      spanId,
-      eventId: event.id,
-    });
-
-    const completedEvent = {
-      ...omit(event, "id"),
-      isPartial: false,
-      isError: options?.attributes.isError ?? false,
-      isCancelled: false,
-      status: options?.attributes.isError ? "ERROR" : "OK",
-      links: event.links ?? [],
-      events: event.events ?? (options?.events as any) ?? [],
-      duration: calculateDurationFromStart(event.startTime, options?.endTime),
-      properties: event.properties as Attributes,
-      metadata: event.metadata as Attributes,
-      style: event.style as Attributes,
-      output: output,
-      outputType:
-        options?.attributes.outputType === "application/store" ||
-        options?.attributes.outputType === "text/plain"
-          ? options?.attributes.outputType
-          : "application/json",
-      payload: event.payload as Attributes,
-      payloadType: event.payloadType,
-    } satisfies CreatableEvent;
-
-    await this.insert(completedEvent);
-
-    return completedEvent;
-  }
-
   async completeSuccessfulRunEvent({ run, endTime }: { run: CompleteableTaskRun; endTime?: Date }) {
     const startTime = convertDateToNanoseconds(run.createdAt);
 
@@ -717,75 +648,6 @@ export class EventRepository {
     });
   }
 
-  async #queryEvents(
-    storeTable: TaskEventStoreTable,
-    queryOptions: QueryOptions,
-    startCreatedAt: Date,
-    endCreatedAt?: Date
-  ): Promise<TaskEventRecord[]> {
-    return await this.taskEventStore.findMany(
-      storeTable,
-      queryOptions,
-      startCreatedAt,
-      endCreatedAt
-    );
-  }
-
-  async queryIncompleteEvents(
-    storeTable: TaskEventStoreTable,
-    queryOptions: QueryOptions,
-    startCreatedAt: Date,
-    endCreatedAt?: Date,
-    allowCompleteDuplicate = false
-  ) {
-    // First we will find all the events that match the query options (selecting minimal data).
-    const taskEvents = await this.taskEventStore.findMany(
-      storeTable,
-      queryOptions,
-      startCreatedAt,
-      endCreatedAt,
-      { spanId: true, isPartial: true, isCancelled: true },
-      undefined,
-      { limit: 500 }
-    );
-
-    // Optimize the filtering by pre-processing the data
-    const completeEventSpanIds = new Set<string>();
-    const incompleteEvents: Array<{ spanId: string }> = [];
-
-    // Single pass to categorize events and build lookup structures
-    for (const event of taskEvents) {
-      if (!event.isPartial && !event.isCancelled) {
-        // This is a complete event
-        completeEventSpanIds.add(event.spanId);
-      } else if (event.isPartial && !event.isCancelled) {
-        // This is a potentially incomplete event
-        incompleteEvents.push(event);
-      }
-      // Skip cancelled events as they are not incomplete
-    }
-
-    // Filter incomplete events, excluding those with complete duplicates
-    const filteredTaskEvents = allowCompleteDuplicate
-      ? incompleteEvents
-      : incompleteEvents.filter((event) => !completeEventSpanIds.has(event.spanId));
-
-    if (filteredTaskEvents.length === 0) {
-      return [];
-    }
-
-    return this.#queryEvents(
-      storeTable,
-      {
-        spanId: {
-          in: filteredTaskEvents.map((event) => event.spanId),
-        },
-      },
-      startCreatedAt,
-      endCreatedAt
-    );
-  }
-
   public async getTraceSummary(
     storeTable: TaskEventStoreTable,
     traceId: string,

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

@@ -33,6 +33,7 @@ import { CancelAttemptService } from "./cancelAttempt.server";
 import { CreateCheckpointService } from "./createCheckpoint.server";
 import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
 import { RetryAttemptService } from "./retryAttempt.server";
+import { tryCatch } from "@trigger.dev/core/utils";
 
 type FoundAttempt = Awaited<ReturnType<typeof findAttempt>>;
 
@@ -164,27 +165,20 @@ export class CompleteAttemptService extends BaseService {
       env,
     });
 
-    // Now we need to "complete" the task run event/span
-    await eventRepository.completeEvent(
-      getTaskEventStoreTableForRun(taskRunAttempt.taskRun),
-      taskRunAttempt.taskRun.spanId,
-      taskRunAttempt.taskRun.createdAt,
-      taskRunAttempt.taskRun.completedAt ?? undefined,
-      {
+    const [completeSuccessfulRunEventError] = await tryCatch(
+      eventRepository.completeSuccessfulRunEvent({
+        run: taskRunAttempt.taskRun,
         endTime: new Date(),
-        attributes: {
-          isError: false,
-          output:
-            completion.outputType === "application/store" || completion.outputType === "text/plain"
-              ? completion.output
-              : completion.output
-              ? (safeJsonParse(completion.output) as Attributes)
-              : undefined,
-          outputType: completion.outputType,
-        },
-      }
+      })
     );
 
+    if (completeSuccessfulRunEventError) {
+      logger.error("[CompleteAttemptService] Failed to complete successful run event", {
+        error: completeSuccessfulRunEventError,
+        runId: taskRunAttempt.taskRunId,
+      });
+    }
+
     return "COMPLETED";
   }
 
@@ -322,29 +316,21 @@ export class CompleteAttemptService extends BaseService {
       exitRun(taskRunAttempt.taskRunId);
     }
 
-    // Now we need to "complete" the task run event/span
-    await eventRepository.completeEvent(
-      getTaskEventStoreTableForRun(taskRunAttempt.taskRun),
-      taskRunAttempt.taskRun.spanId,
-      taskRunAttempt.taskRun.createdAt,
-      taskRunAttempt.taskRun.completedAt ?? undefined,
-      {
+    const [completeFailedRunEventError] = await tryCatch(
+      eventRepository.completeFailedRunEvent({
+        run: taskRunAttempt.taskRun,
         endTime: failedAt,
-        attributes: {
-          isError: true,
-        },
-        events: [
-          {
-            name: "exception",
-            time: failedAt,
-            properties: {
-              exception: createExceptionPropertiesFromError(sanitizedError),
-            },
-          },
-        ],
-      }
+        exception: createExceptionPropertiesFromError(sanitizedError),
+      })
     );
 
+    if (completeFailedRunEventError) {
+      logger.error("[CompleteAttemptService] Failed to complete failed run event", {
+        error: completeFailedRunEventError,
+        runId: taskRunAttempt.taskRunId,
+      });
+    }
+
     await this._prisma.taskRun.update({
       where: {
         id: taskRunAttempt.taskRunId,
@@ -385,64 +371,43 @@ export class CompleteAttemptService extends BaseService {
       return "COMPLETED";
     }
 
-    const inProgressEvents = await eventRepository.queryIncompleteEvents(
-      getTaskEventStoreTableForRun(taskRunAttempt.taskRun),
-      {
-        runId: taskRunAttempt.taskRun.friendlyId,
-      },
-      taskRunAttempt.taskRun.createdAt,
-      taskRunAttempt.taskRun.completedAt ?? undefined
-    );
-
     // Handle in-progress events
     switch (status) {
       case "CRASHED": {
-        logger.debug("[CompleteAttemptService] Crashing in-progress events", {
-          inProgressEvents: inProgressEvents.map((event) => event.id),
-        });
-
-        await Promise.all(
-          inProgressEvents.map((event) => {
-            return eventRepository.crashEvent({
-              event,
-              crashedAt: failedAt,
-              exception: createExceptionPropertiesFromError(sanitizedError),
-            });
+        const [createAttemptFailedEventError] = await tryCatch(
+          eventRepository.createAttemptFailedRunEvent({
+            run: taskRunAttempt.taskRun,
+            endTime: failedAt,
+            attemptNumber: taskRunAttempt.number,
+            exception: createExceptionPropertiesFromError(sanitizedError),
           })
         );
 
+        if (createAttemptFailedEventError) {
+          logger.error("[CompleteAttemptService] Failed to create attempt failed run event", {
+            error: createAttemptFailedEventError,
+            runId: taskRunAttempt.taskRunId,
+          });
+        }
+
         break;
       }
       case "SYSTEM_FAILURE": {
-        logger.debug("[CompleteAttemptService] Failing in-progress events", {
-          inProgressEvents: inProgressEvents.map((event) => event.id),
-        });
-
-        await Promise.all(
-          inProgressEvents.map((event) => {
-            return eventRepository.completeEvent(
-              getTaskEventStoreTableForRun(taskRunAttempt.taskRun),
-              event.spanId,
-              taskRunAttempt.taskRun.createdAt,
-              taskRunAttempt.taskRun.completedAt ?? undefined,
-              {
-                endTime: failedAt,
-                attributes: {
-                  isError: true,
-                },
-                events: [
-                  {
-                    name: "exception",
-                    time: failedAt,
-                    properties: {
-                      exception: createExceptionPropertiesFromError(sanitizedError),
-                    },
-                  },
-                ],
-              }
-            );
+        const [createAttemptFailedEventError] = await tryCatch(
+          eventRepository.createAttemptFailedRunEvent({
+            run: taskRunAttempt.taskRun,
+            endTime: failedAt,
+            attemptNumber: taskRunAttempt.number,
+            exception: createExceptionPropertiesFromError(sanitizedError),
           })
         );
+
+        if (createAttemptFailedEventError) {
+          logger.error("[CompleteAttemptService] Failed to create attempt failed run event", {
+            error: createAttemptFailedEventError,
+            runId: taskRunAttempt.taskRunId,
+          });
+        }
       }
     }