Merge branch 'Chigala-feat/cancelEvent-function'

ericallam · ericallam · commit 0493cebaed90 · 2023-08-21T17:49:48.000+01:00
diff --git a/.changeset/curly-balloons-matter.md b/.changeset/curly-balloons-matter.md
@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Adding the ability to cancel events that were sent with a delayed delivery
diff --git a/apps/webapp/app/db.server.ts b/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;
     }
 
diff --git a/apps/webapp/app/routes/api.v1.events.$eventId.cancel.ts b/apps/webapp/app/routes/api.v1.events.$eventId.cancel.ts
@@ -0,0 +1,51 @@
+import type { ActionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { CancelEventService } from "~/services/events/cancelEvent.server";
+import { logger } from "~/services/logger.server";
+
+const ParamsSchema = z.object({
+  eventId: z.string(),
+});
+
+export async function action({ request, params }: ActionArgs) {
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "POST") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const authenticatedEnv = authenticationResult.environment;
+
+  const parsed = ParamsSchema.safeParse(params);
+
+  if (!parsed.success) {
+    return json({ error: "Invalid or Missing eventId" }, { status: 400 });
+  }
+
+  const { eventId } = parsed.data;
+
+  const service = new CancelEventService();
+  try {
+    const updatedEvent = await service.call(authenticatedEnv, eventId);
+
+    if (!updatedEvent) {
+      return json({ error: "Event not found" }, { status: 404 });
+    }
+
+    return json(updatedEvent);
+  } catch (err) {
+    logger.error("CancelEventService.call() error", {
+      error: err,
+    });
+
+    return json({ error: "Internal Server Error" }, { status: 500 });
+  }
+}
diff --git a/apps/webapp/app/services/events/cancelEvent.server.ts b/apps/webapp/app/services/events/cancelEvent.server.ts
@@ -0,0 +1,47 @@
+import type { EventRecord } from "@trigger.dev/database";
+import { $transaction, PrismaClientOrTransaction, prisma } from "~/db.server";
+import { workerQueue } from "../worker.server";
+import { AuthenticatedEnvironment } from "../apiAuth.server";
+
+export class CancelEventService {
+  #prismaClient: PrismaClientOrTransaction;
+
+  constructor(prismaClient: PrismaClientOrTransaction = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(
+    environment: AuthenticatedEnvironment,
+    eventId: string
+  ): Promise<EventRecord | undefined> {
+    return await $transaction(this.#prismaClient, async (tx) => {
+      const event = await tx.eventRecord.findUnique({
+        where: {
+          eventId_environmentId: {
+            eventId: eventId,
+            environmentId: environment.id,
+          },
+        },
+      });
+
+      if (!event) {
+        return;
+      }
+
+      if (event.cancelledAt) {
+        return event;
+      }
+
+      //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 });
+
+      return updatedEvent;
+    });
+  }
+}
diff --git a/apps/webapp/app/services/events/ingestSendEvent.server.ts b/apps/webapp/app/services/events/ingestSendEvent.server.ts
@@ -2,14 +2,12 @@ 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;
 
-  constructor(
-    prismaClient: PrismaClientOrTransaction = prisma,
-    private deliverEvents = true
-  ) {
+  constructor(prismaClient: PrismaClientOrTransaction = prisma, private deliverEvents = true) {
     this.#prismaClient = prismaClient;
   }
 
@@ -36,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 }
-          );
-        }
+          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: {
diff --git a/examples/job-catalog/src/events.ts b/examples/job-catalog/src/events.ts
@@ -29,4 +29,26 @@ client.defineJob({
   },
 });
 
+client.defineJob({
+  id: "cancel-event-example",
+  name: "Cancel Event Example",
+  version: "1.0.0",
+  trigger: eventTrigger({
+    name: "cancel.event.example",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.sendEvent(
+      "send-event",
+      { name: "Cancellable Event", id: payload.id },
+      {
+        deliverAt: new Date(Date.now() + 1000 * 60 * 60 * 24), // 24 hours from now
+      }
+    );
+
+    await io.wait("wait-1", 60); // 1 minute
+
+    await io.cancelEvent("cancel-event", payload.id);
+  },
+});
+
 createExpressServer(client);
diff --git a/packages/database/prisma/migrations/20230814131639_added_cancelled_at_column_to_the_event_record_table/migration.sql b/packages/database/prisma/migrations/20230814131639_added_cancelled_at_column_to_the_event_record_table/migration.sql
@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "EventRecord" ADD COLUMN     "cancelledAt" TIMESTAMP(3);
diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma
@@ -642,6 +642,7 @@ model EventRecord {
 
   createdAt DateTime @default(now())
   updatedAt DateTime @updatedAt
+  cancelledAt DateTime? 
 
   isTest Boolean  @default(false)
   runs   JobRun[]
diff --git a/packages/trigger-sdk/src/apiClient.ts b/packages/trigger-sdk/src/apiClient.ts
@@ -201,6 +201,22 @@ export class ApiClient {
     });
   }
 
+  async cancelEvent(eventId: string) {
+    const apiKey = await this.#apiKey();
+
+    this.#logger.debug("Cancelling event", {
+      eventId,
+    });
+
+    return await zodfetch(ApiEventLogSchema, `${this.#apiUrl}/api/v1/events/${eventId}/cancel`, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+        Authorization: `Bearer ${apiKey}`,
+      },
+    });
+  }
+
   async updateSource(
     client: string,
     key: string,
diff --git a/packages/trigger-sdk/src/io.ts b/packages/trigger-sdk/src/io.ts
@@ -221,6 +221,32 @@ export class IO {
     );
   }
 
+  /** `io.cancelEvent()` allows you to cancel an event that was previously sent with `io.sendEvent()`. This will prevent any Jobs from running that are listening for that event if the event was sent with a delay
+   * @param key
+   * @param eventId
+   * @returns
+   */
+  async cancelEvent(key: string | any[], eventId: string) {
+    return await this.runTask(
+      key,
+      {
+        name: "cancelEvent",
+        params: {
+          eventId,
+        },
+        properties: [
+          {
+            label: "id",
+            text: eventId,
+          },
+        ],
+      },
+      async (task) => {
+        return await this._triggerClient.cancelEvent(eventId);
+      }
+    );
+  }
+
   async updateSource(key: string | any[], options: { key: string } & UpdateTriggerSourceBody) {
     return this.runTask(
       key,
@@ -432,7 +458,10 @@ export class IO {
 =   * @param onError The callback that will be called when the Task fails. The callback receives the error, the Task and the IO as parameters. If you wish to retry then return an object with a `retryAt` property.
    * @returns A Promise that resolves with the returned value of the callback.
    */
-  async runTask<TResult extends SerializableJson | void = void, TCallbackResult extends unknown = TResult>(
+  async runTask<
+    TResult extends SerializableJson | void = void,
+    TCallbackResult extends unknown = TResult,
+  >(
     key: string | any[],
     options: RunTaskOptions,
     callback: (task: IOTask, io: IO) => Promise<TCallbackResult | TResult>,
@@ -525,7 +554,7 @@ export class IO {
         const result = await callback(task, this);
 
         const output = SerializableJsonSchema.parse(result) as TResult;
-        
+
         this._logger.debug("Completing using output", {
           idempotencyKey,
           task,
diff --git a/packages/trigger-sdk/src/triggerClient.ts b/packages/trigger-sdk/src/triggerClient.ts

-Original file line number
+Diff line change
@@ @@ -0,0 +1,5 @@ @@
 +---
 +"@trigger.dev/sdk": patch
 +---
++
 +Adding the ability to cancel events that were sent with a delayed delivery
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+-- AlterTable`
	`2`	`+ALTER TABLE "EventRecord" ADD COLUMN "cancelledAt" TIMESTAMP(3);`