feat: added the ability to cancel Events based on the event ID

Author: Chigala

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

@@ -76,6 +76,10 @@ export type ZodWorkerEnqueueOptions = TaskSpec & {
   tx?: PrismaClientOrTransaction;
 };
 
+export type ZodWorkerDequeueOptions = {
+  tx?: PrismaClientOrTransaction;
+};
+
 export type ZodWorkerOptions<TMessageCatalog extends MessageCatalogSchema> = {
   runnerOptions: RunnerOptions;
   prisma: PrismaClient;
@@ -167,6 +171,18 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
     return job;
   }
 
+  public async dequeue(jobKey: string, option?: ZodWorkerDequeueOptions): Promise<GraphileJob> {
+    if (!this.#runner) {
+      throw new Error("Worker not initialized");
+    }
+    const job = await this.#removeJob(jobKey, option?.tx ?? this.#prisma);
+    logger.debug("dequeued worker task", {
+      job,
+    });
+
+    return job;
+  }
+
   async #addJob(
     identifier: string,
     payload: unknown,
@@ -190,7 +206,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
       spec.queueName || null,
       spec.runAt || null,
       spec.maxAttempts || null,
-      spec.jobKey || null,
+      identifier,
       spec.priority || null,
       spec.jobKeyMode || null,
       spec.flags || null
@@ -209,6 +225,25 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
     return job as GraphileJob;
   }
 
+  async #removeJob(jobKey: string, tx: PrismaClientOrTransaction) {
+    const result = await tx.$queryRawUnsafe(
+      `SELECT * FROM graphile_worker.remove_job(
+          job_key => $1::text
+        )`,
+      jobKey
+    );
+
+    const job = GraphileJobSchema.safeParse(result);
+
+    if (!job.success) {
+      throw new Error(
+        `Failed to remove job from queue, zod parsing error: ${JSON.stringify(job.error)}`
+      );
+    }
+
+    return job.data as GraphileJob;
+  }
+
   #createTaskListFromTasks() {
     const taskList: TaskList = {};
 

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

@@ -0,0 +1,78 @@
+import type { ActionArgs, LoaderArgs } 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";
+
+const ParamsSchema = z.object({
+  eventId: z.string(),
+});
+
+export async function loader({ request, params }: LoaderArgs) {
+  // 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 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 });
+  }
+  // Update the event with cancelledAt
+  let updatedEvent;
+
+  try {
+    updatedEvent = await prisma.eventRecord.update({
+      where: { id: event.id },
+      data: { cancelledAt: new Date() },
+    });
+
+    // Dequeue the event only if the update is successful
+    await workerQueue.dequeue(updatedEvent.id, { tx: prisma });
+
+    return json(updatedEvent);
+  } catch (error) {
+    logger.error("Failed to update event", { error, event });
+
+    return json({ error: "Failed to update event" }, { status: 500 });
+  }
+}

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

@@ -640,6 +640,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/triggerClient.ts

@@ -556,6 +556,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);
   }