Merge branch 'feat/cancelEvent-function' of https://github.com/Chigala/trigger.dev into Chigala-feat/cancelEvent-function

# Conflicts:
#	apps/webapp/app/platform/zodWorker.server.ts

Author: ericallam

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

@@ -0,0 +1,53 @@
+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";
+
+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 });
+  }
+}

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

@@ -0,0 +1,67 @@
+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({
+          select: {
+            id: true,
+            name: true,
+            createdAt: true,
+            updatedAt: true,
+            environmentId: true,
+            cancelledAt: true,
+            runs: {
+              select: {
+                id: true,
+                status: true,
+                startedAt: true,
+                completedAt: true,
+              },
+            },
+          },
+          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 prisma.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 });
+
+        return updatedEvent;
+      },
+      { timeout: 10000 }
+    );
+  }
+}

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

@@ -6,10 +6,7 @@ import { workerQueue } from "~/services/worker.server";
 export class IngestSendEvent {
   #prismaClient: PrismaClientOrTransaction;
 
-  constructor(
-    prismaClient: PrismaClientOrTransaction = prisma,
-    private deliverEvents = true
-  ) {
+  constructor(prismaClient: PrismaClientOrTransaction = prisma, private deliverEvents = true) {
     this.#prismaClient = prismaClient;
   }
 
@@ -91,7 +88,7 @@ export class IngestSendEvent {
             {
               id: eventLog.id,
             },
-            { runAt: eventLog.deliverAt, tx }
+            { runAt: eventLog.deliverAt, tx, jobKey: `event:${eventLog.id}` }
           );
         }
 

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);

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[]

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,

packages/trigger-sdk/src/io.ts

@@ -221,6 +221,29 @@ export class IO {
     );
   }
 
+  async cancelEvent(key: string | any[], eventId: string) {
+    return await this.runTask(
+      key,
+      {
+        name: "cancelEvent",
+        params: {
+          eventId,
+        },
+        properties: [
+          {
+            label: "id",
+            text: eventId,
+          },
+        ],
+        noop: true,
+      },
+      async (task) => {
+        return await this._triggerClient.cancelEvent(eventId);
+      }
+    );
+  }
+
+
   async updateSource(key: string | any[], options: { key: string } & UpdateTriggerSourceBody) {
     return this.runTask(
       key,

packages/trigger-sdk/src/triggerClient.ts

@@ -551,6 +551,10 @@ export class TriggerClient {
     return this.#client.sendEvent(event, options);
   }
 
+  async cancelEvent(eventId: string) {
+    return this.#client.cancelEvent(eventId);
+  }
+
   async registerSchedule(id: string, key: string, schedule: ScheduleMetadata) {
     return this.#client.registerSchedule(this.id, id, key, schedule);
   }