feat: allow sendEvent to update an existing event (#607)

Author: hmacr

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

@@ -3,6 +3,25 @@ import { $transaction, PrismaClientOrTransaction, PrismaErrorSchema, prisma } fr
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { workerQueue } from "~/services/worker.server";
 import { logger } from "../logger.server";
+import { EventRecord, ExternalAccount } from "@trigger.dev/database";
+
+type UpdateEventInput = {
+  tx: PrismaClientOrTransaction;
+  existingEventLog: EventRecord;
+  reqEvent: RawEvent;
+  deliverAt?: Date;
+};
+
+type CreateEventInput = {
+  tx: PrismaClientOrTransaction;
+  event: RawEvent;
+  environment: AuthenticatedEnvironment;
+  deliverAt?: Date;
+  sourceContext?: { id: string; metadata?: any };
+  externalAccount?: ExternalAccount;
+};
+
+const EVENT_UPDATE_THRESHOLD_WINDOW_IN_MSECS = 5 * 1000; // 5 seconds
 
 export class IngestSendEvent {
   #prismaClient: PrismaClientOrTransaction;
@@ -52,34 +71,25 @@ export class IngestSendEvent {
             })
           : undefined;
 
-        // Create a new event in the database
-        const eventLog = await tx.eventRecord.create({
-          data: {
-            organizationId: environment.organizationId,
-            projectId: environment.projectId,
-            environmentId: 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,
-            externalAccountId: externalAccount ? externalAccount.id : undefined,
+        const existingEventLog = await tx.eventRecord.findUnique({
+          where: {
+            eventId_environmentId: {
+              eventId: event.id,
+              environmentId: environment.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}` }
-          );
-        }
+        const eventLog = await (existingEventLog
+          ? this.updateEvent({ tx, existingEventLog, reqEvent: event, deliverAt })
+          : this.createEvent({
+              tx,
+              event,
+              environment,
+              deliverAt,
+              sourceContext,
+              externalAccount,
+            }));
 
         return eventLog;
       });
@@ -95,21 +105,81 @@ export class IngestSendEvent {
         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 });
+      throw error;
+    }
+  }
 
-        return this.#prismaClient.eventRecord.findUniqueOrThrow({
-          where: {
-            eventId_environmentId: {
-              eventId: event.id,
-              environmentId: environment.id,
-            },
-          },
-        });
-      }
+  private async createEvent({
+    tx,
+    event,
+    environment,
+    deliverAt,
+    sourceContext,
+    externalAccount,
+  }: CreateEventInput) {
+    const eventLog = await tx.eventRecord.create({
+      data: {
+        organizationId: environment.organizationId,
+        projectId: environment.projectId,
+        environmentId: 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,
+        externalAccountId: externalAccount ? externalAccount.id : undefined,
+      },
+    });
 
-      throw error;
+    await this.enqueueWorkerEvent(tx, eventLog);
+
+    return eventLog;
+  }
+
+  private async updateEvent({ tx, existingEventLog, reqEvent, deliverAt }: UpdateEventInput) {
+    if (!this.shouldUpdateEvent(existingEventLog)) {
+      logger.debug(`not updating event for event id: ${existingEventLog.eventId}`);
+      return existingEventLog;
+    }
+
+    const updatedEventLog = await tx.eventRecord.update({
+      where: {
+        eventId_environmentId: {
+          eventId: existingEventLog.eventId,
+          environmentId: existingEventLog.environmentId,
+        },
+      },
+      data: {
+        payload: reqEvent.payload ?? existingEventLog.payload,
+        context: reqEvent.context ?? existingEventLog.context,
+        deliverAt: deliverAt ?? new Date(),
+      },
+    });
+
+    await this.enqueueWorkerEvent(tx, updatedEventLog);
+
+    return updatedEventLog;
+  }
+
+  private shouldUpdateEvent(eventLog: EventRecord) {
+    const thresholdTime = new Date(Date.now() + EVENT_UPDATE_THRESHOLD_WINDOW_IN_MSECS);
+
+    return eventLog.deliverAt >= thresholdTime;
+  }
+
+  private async enqueueWorkerEvent(tx: PrismaClientOrTransaction, eventLog: EventRecord) {
+    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}` }
+      );
     }
   }
 }