added io.cancelEvent support and fixed other requested changes

Author: Chigala

apps/webapp/app/platform/zodWorker.server.ts

@@ -205,7 +205,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
       spec.queueName || null,
       spec.runAt || null,
       spec.maxAttempts || null,
-      spec.jobKey,
+      spec.jobKey || null,
       spec.priority || null,
       spec.jobKeyMode || null,
       spec.flags || null

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

@@ -5,6 +5,7 @@ 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(),
@@ -33,39 +34,21 @@ export async function action({ request, params }: ActionArgs) {
 
   const { eventId } = parsed.data;
 
-  const event = await prisma.eventRecord.findFirst({
-    select: {
-      id: true,
-      name: true,
-      createdAt: true,
-      updatedAt: true,
-      environmentId: true,
-      runs: {
-        select: {
-          id: true,
-          status: true,
-          startedAt: true,
-          completedAt: true,
-        },
-      },
-    },
-    where: {
-      id: eventId,
-      environmentId: authenticatedEnv.id,
-    },
-  });
-  if (!event) {
-    return json({ error: "Event not found" }, { status: 404 });
-  }
+  const service = new CancelEventService();
+  try {
+    const updatedEvent = await service.call(authenticatedEnv, eventId);
 
-//update the cancelledAt column in the eventRecord table
-  const updatedEvent = await prisma.eventRecord.update({
-    where: { id: event.id },
-    data: { cancelledAt: new Date() },
-  });
+    if (!updatedEvent) {
+      return json({ error: "Event not found" }, { status: 404 });
+    }
 
-  // Dequeue the event after the db has been updated
-  await workerQueue.dequeue(event.id, { tx: prisma });
+    return json(updatedEvent);
+  } catch (err) {
+    //send the error
+    logger.error("CancelEventService.call() error", {
+      error: err,
+    });
 
-  return json(updatedEvent);
+    return json({ error: "Internal Server Error" }, { status: 500 });
+  }
 }

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

@@ -0,0 +1,60 @@
+import type { EventDispatcher, EventRecord } from "@trigger.dev/database";
+import type { EventFilter } from "@trigger.dev/core";
+import { EventFilterSchema, eventFilterMatches } from "@trigger.dev/core";
+import { $transaction, PrismaClientOrTransaction, prisma } from "~/db.server";
+import { logger } from "~/services/logger.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.findFirst({
+          select: {
+            id: true,
+            name: true,
+            createdAt: true,
+            updatedAt: true,
+            environmentId: true,
+            runs: {
+              select: {
+                id: true,
+                status: true,
+                startedAt: true,
+                completedAt: true,
+              },
+            },
+          },
+          where: {
+            id: eventId,
+            environmentId: environment.id,
+          },
+        });
+
+        if (!event) {
+          return;
+        }
+
+        //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.id, { tx: prisma });
+
+        return updatedEvent;
+      },
+      { timeout: 10000 }
+    );
+  }
+}

packages/trigger-sdk/src/io.ts

@@ -220,6 +220,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,