Fixed an issue with IngestSendEvent not returning an existing event

Author: ericallam

apps/webapp/app/db.server.ts

@@ -30,6 +30,8 @@ export type PrismaTransactionOptions = {
 
   /**  Sets the transaction isolation level. By default this is set to the value currently configured in your database. */
   isolationLevel?: Prisma.TransactionIsolationLevel;
+
+  rethrowPrismaErrors?: boolean;
 };
 
 export async function $transaction<R>(
@@ -53,6 +55,10 @@ export async function $transaction<R>(
         name: error.name,
       });
 
+      if (options?.rethrowPrismaErrors) {
+        throw error;
+      }
+
       return;
     }
 

apps/webapp/app/routes/api.v1.events.$eventId.cancel.ts

@@ -1,11 +1,9 @@
 import type { ActionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
-import { prisma } from "~/db.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
-import { workerQueue } from "~/services/worker.server";
-import { logger } from "~/services/logger.server";
 import { CancelEventService } from "~/services/events/cancelEvent.server";
+import { logger } from "~/services/logger.server";
 
 const ParamsSchema = z.object({
   eventId: z.string(),

apps/webapp/app/services/events/cancelEvent.server.ts

@@ -14,54 +14,34 @@ export class CancelEventService {
     environment: AuthenticatedEnvironment,
     eventId: string
   ): Promise<EventRecord | undefined> {
-    return await $transaction(
-      this.#prismaClient,
-      async (tx) => {
-        const event = await tx.eventRecord.findUnique({
-          select: {
-            id: true,
-            name: true,
-            createdAt: true,
-            updatedAt: true,
-            environmentId: true,
-            cancelledAt: true,
-            runs: {
-              select: {
-                id: true,
-                status: true,
-                startedAt: true,
-                completedAt: true,
-              },
-            },
+    return await $transaction(this.#prismaClient, async (tx) => {
+      const event = await tx.eventRecord.findUnique({
+        where: {
+          eventId_environmentId: {
+            eventId: eventId,
+            environmentId: environment.id,
           },
-          where: {
-            eventId_environmentId: {
-              eventId: eventId,
-              environmentId: environment.id,
-            },
-          },
-        });
+        },
+      });
 
-        if (!event) {
-          return;
-        }
+      if (!event) {
+        return;
+      }
 
-        if (event.cancelledAt) {
-          return event;
-        }
+      if (event.cancelledAt) {
+        return event;
+      }
 
-        //update the cancelledAt column in the eventRecord table
-        const updatedEvent = await prisma.eventRecord.update({
-          where: { id: event.id },
-          data: { cancelledAt: new Date() },
-        });
+      //update the cancelledAt column in the eventRecord table
+      const updatedEvent = await tx.eventRecord.update({
+        where: { id: event.id },
+        data: { cancelledAt: new Date() },
+      });
 
-        // Dequeue the event after the db has been updated
-        await workerQueue.dequeue(`event:${event.id}`, { tx: prisma });
+      // Dequeue the event after the db has been updated
+      await workerQueue.dequeue(`event:${event.id}`, { tx });
 
-        return updatedEvent;
-      },
-      { timeout: 10000 }
-    );
+      return updatedEvent;
+    });
   }
 }

apps/webapp/app/services/events/ingestSendEvent.server.ts

@@ -2,6 +2,7 @@ import type { RawEvent, SendEventOptions } from "@trigger.dev/core";
 import { $transaction, PrismaClientOrTransaction, PrismaErrorSchema, prisma } from "~/db.server";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { workerQueue } from "~/services/worker.server";
+import { logger } from "../logger.server";
 
 export class IngestSendEvent {
   #prismaClient: PrismaClientOrTransaction;
@@ -33,71 +34,87 @@ export class IngestSendEvent {
     try {
       const deliverAt = this.#calculateDeliverAt(options);
 
-      return await $transaction(this.#prismaClient, async (tx) => {
-        const externalAccount = options?.accountId
-          ? await tx.externalAccount.findUniqueOrThrow({
-              where: {
-                environmentId_identifier: {
-                  environmentId: environment.id,
-                  identifier: options.accountId,
+      return await $transaction(
+        this.#prismaClient,
+        async (tx) => {
+          const externalAccount = options?.accountId
+            ? await tx.externalAccount.findUniqueOrThrow({
+                where: {
+                  environmentId_identifier: {
+                    environmentId: environment.id,
+                    identifier: options.accountId,
+                  },
                 },
-              },
-            })
-          : undefined;
+              })
+            : undefined;
 
-        // Create a new event in the database
-        const eventLog = await tx.eventRecord.create({
-          data: {
-            organization: {
-              connect: {
-                id: environment.organizationId,
+          // Create a new event in the database
+          const eventLog = await tx.eventRecord.create({
+            data: {
+              organization: {
+                connect: {
+                  id: environment.organizationId,
+                },
               },
-            },
-            project: {
-              connect: {
-                id: environment.projectId,
+              project: {
+                connect: {
+                  id: environment.projectId,
+                },
               },
-            },
-            environment: {
-              connect: {
-                id: environment.id,
+              environment: {
+                connect: {
+                  id: environment.id,
+                },
               },
+              eventId: event.id,
+              name: event.name,
+              timestamp: event.timestamp ?? new Date(),
+              payload: event.payload ?? {},
+              context: event.context ?? {},
+              source: event.source ?? "trigger.dev",
+              sourceContext,
+              deliverAt: deliverAt,
+              externalAccount: externalAccount
+                ? {
+                    connect: {
+                      id: externalAccount.id,
+                    },
+                  }
+                : {},
             },
-            eventId: event.id,
-            name: event.name,
-            timestamp: event.timestamp ?? new Date(),
-            payload: event.payload ?? {},
-            context: event.context ?? {},
-            source: event.source ?? "trigger.dev",
-            sourceContext,
-            deliverAt: deliverAt,
-            externalAccount: externalAccount
-              ? {
-                  connect: {
-                    id: externalAccount.id,
-                  },
-                }
-              : {},
-          },
-        });
+          });
 
-        if (this.deliverEvents) {
-          // Produce a message to the event bus
-          await workerQueue.enqueue(
-            "deliverEvent",
-            {
-              id: eventLog.id,
-            },
-            { runAt: eventLog.deliverAt, tx, jobKey: `event:${eventLog.id}` }
-          );
-        }
+          if (this.deliverEvents) {
+            // Produce a message to the event bus
+            await workerQueue.enqueue(
+              "deliverEvent",
+              {
+                id: eventLog.id,
+              },
+              { runAt: eventLog.deliverAt, tx, jobKey: `event:${eventLog.id}` }
+            );
+          }
 
-        return eventLog;
-      });
+          return eventLog;
+        },
+        { rethrowPrismaErrors: true }
+      );
     } catch (error) {
       const prismaError = PrismaErrorSchema.safeParse(error);
+
+      if (!prismaError.success) {
+        logger.debug("Error parsing prisma error", {
+          error,
+          parseError: prismaError.error.format(),
+        });
+
+        throw error;
+      }
+
       // If the error is a Prisma unique constraint error, it means that the event already exists
       if (prismaError.success && prismaError.data.code === "P2002") {
+        logger.debug("Event already exists, finding and returning", { event, environment });
+
         return this.#prismaClient.eventRecord.findUniqueOrThrow({
           where: {
             eventId_environmentId: {