fix: meeting ended and started webhooks trigger (#14864)

Co-authored-by: CarinaWolli <wollencarina@gmail.com>
Co-authored-by: Keith Williams <keithwillcode@gmail.com>

apps/web/playwright/webhook.e2e.ts

@@ -2,6 +2,7 @@ import type { Page } from "@playwright/test";
 import { expect } from "@playwright/test";
 import { v4 as uuidv4 } from "uuid";
 
+import dayjs from "@calcom/dayjs";
 import prisma from "@calcom/prisma";
 import { BookingStatus } from "@calcom/prisma/client";
 
@@ -543,6 +544,131 @@ test.describe("BOOKING_RESCHEDULED", async () => {
   });
 });
 
+test.describe("MEETING_ENDED, MEETING_STARTED", async () => {
+  test("should create/remove scheduledWebhookTriggers for existing bookings", async ({
+    page,
+    users,
+    bookings,
+  }, _testInfo) => {
+    const user = await users.create();
+    await user.apiLogin();
+    const tomorrow = dayjs().add(1, "day");
+    const [eventType] = user.eventTypes;
+    bookings.create(user.id, user.name, eventType.id);
+    bookings.create(user.id, user.name, eventType.id, { startTime: dayjs().add(2, "day").toDate() });
+
+    //create a new webhook with meeting ended trigger here
+    await page.goto("/settings/developer/webhooks");
+    // --- add webhook
+    await page.click('[data-testid="new_webhook"]');
+
+    await page.fill('[name="subscriberUrl"]', "https://www.example.com");
+
+    await Promise.all([
+      page.click("[type=submit]"),
+      page.waitForURL((url) => url.pathname.endsWith("/settings/developer/webhooks")),
+    ]);
+
+    const scheduledTriggers = await prisma.webhookScheduledTriggers.findMany({
+      where: {
+        webhook: {
+          userId: user.id,
+        },
+      },
+      select: {
+        payload: true,
+        webhook: {
+          select: {
+            userId: true,
+            id: true,
+            subscriberUrl: true,
+          },
+        },
+        startAfter: true,
+      },
+    });
+
+    const existingUserBookings = await prisma.booking.findMany({
+      where: {
+        userId: user.id,
+        startTime: {
+          gt: new Date(),
+        },
+      },
+    });
+
+    const meetingStartedTriggers = scheduledTriggers.filter((trigger) =>
+      trigger.payload.includes("MEETING_STARTED")
+    );
+    const meetingEndedTriggers = scheduledTriggers.filter((trigger) =>
+      trigger.payload.includes("MEETING_ENDED")
+    );
+
+    expect(meetingStartedTriggers.length).toBe(existingUserBookings.length);
+    expect(meetingEndedTriggers.length).toBe(existingUserBookings.length);
+
+    expect(meetingStartedTriggers.map((trigger) => trigger.startAfter)).toEqual(
+      expect.arrayContaining(existingUserBookings.map((booking) => booking.startTime))
+    );
+    expect(meetingEndedTriggers.map((trigger) => trigger.startAfter)).toEqual(
+      expect.arrayContaining(existingUserBookings.map((booking) => booking.endTime))
+    );
+
+    page.reload();
+
+    // edit webhook and remove trigger meeting ended trigger
+    await page.click('[data-testid="webhook-edit-button"]');
+    await page.getByRole("button", { name: "Remove Meeting Ended" }).click();
+
+    await Promise.all([
+      page.click("[type=submit]"),
+      page.waitForURL((url) => url.pathname.endsWith("/settings/developer/webhooks")),
+    ]);
+
+    const scheduledTriggersAfterRemovingTrigger = await prisma.webhookScheduledTriggers.findMany({
+      where: {
+        webhook: {
+          userId: user.id,
+        },
+      },
+    });
+
+    const newMeetingStartedTriggers = scheduledTriggersAfterRemovingTrigger.filter((trigger) =>
+      trigger.payload.includes("MEETING_STARTED")
+    );
+    const newMeetingEndedTriggers = scheduledTriggersAfterRemovingTrigger.filter((trigger) =>
+      trigger.payload.includes("MEETING_ENDED")
+    );
+
+    expect(newMeetingStartedTriggers.length).toBe(existingUserBookings.length);
+    expect(newMeetingEndedTriggers.length).toBe(0);
+
+    // disable webhook
+    await page.click('[data-testid="webhook-switch"]');
+
+    await page.waitForLoadState("networkidle");
+
+    const scheduledTriggersAfterDisabling = await prisma.webhookScheduledTriggers.findMany({
+      where: {
+        webhook: {
+          userId: user.id,
+        },
+      },
+      select: {
+        payload: true,
+        webhook: {
+          select: {
+            userId: true,
+          },
+        },
+        startAfter: true,
+      },
+    });
+
+    expect(scheduledTriggersAfterDisabling.length).toBe(0);
+  });
+});
+
 test.describe("FORM_SUBMITTED", async () => {
   test("on submitting user form, triggers user webhook", async ({ page, users, routingForms }) => {
     const webhookReceiver = createHttpServer();

apps/web/test/utils/bookingScenario/expects.ts

@@ -288,20 +288,16 @@ export function expectWebhookToHaveBeenCalledWith(
 
   expect(parsedBody.triggerEvent).toBe(data.triggerEvent);
 
-  if (parsedBody.payload.metadata?.videoCallUrl) {
-    parsedBody.payload.metadata.videoCallUrl = parsedBody.payload.metadata.videoCallUrl
-      ? parsedBody.payload.metadata.videoCallUrl
-      : parsedBody.payload.metadata.videoCallUrl;
-  }
-
-  if (data.payload) {
-    if (data.payload.metadata !== undefined) {
-      expect(parsedBody.payload.metadata).toEqual(expect.objectContaining(data.payload.metadata));
+  if (parsedBody.payload) {
+    if (data.payload) {
+      if (data.payload.metadata !== undefined) {
+        expect(parsedBody.payload.metadata).toEqual(expect.objectContaining(data.payload.metadata));
+      }
+      if (data.payload.responses !== undefined)
+        expect(parsedBody.payload.responses).toEqual(expect.objectContaining(data.payload.responses));
+      const { responses: _1, metadata: _2, ...remainingPayload } = data.payload;
+      expect(parsedBody.payload).toEqual(expect.objectContaining(remainingPayload));
     }
-    if (data.payload.responses !== undefined)
-      expect(parsedBody.payload.responses).toEqual(expect.objectContaining(data.payload.responses));
-    const { responses: _1, metadata: _2, ...remainingPayload } = data.payload;
-    expect(parsedBody.payload).toEqual(expect.objectContaining(remainingPayload));
   }
 }
 

packages/features/bookings/lib/handleCancelBooking.ts

@@ -12,12 +12,13 @@ import { sendCancelledReminders } from "@calcom/features/ee/workflows/lib/remind
 import { deleteScheduledSMSReminder } from "@calcom/features/ee/workflows/lib/reminders/smsReminderManager";
 import { deleteScheduledWhatsappReminder } from "@calcom/features/ee/workflows/lib/reminders/whatsappReminderManager";
 import getWebhooks from "@calcom/features/webhooks/lib/getWebhooks";
-import { cancelScheduledJobs } from "@calcom/features/webhooks/lib/scheduleTrigger";
+import { deleteWebhookScheduledTriggers } from "@calcom/features/webhooks/lib/scheduleTrigger";
 import sendPayload from "@calcom/features/webhooks/lib/sendOrSchedulePayload";
 import type { EventTypeInfo } from "@calcom/features/webhooks/lib/sendPayload";
 import { isPrismaObjOrUndefined, parseRecurringEvent } from "@calcom/lib";
 import { getTeamIdFromEventType } from "@calcom/lib/getTeamIdFromEventType";
 import { HttpError } from "@calcom/lib/http-error";
+import logger from "@calcom/lib/logger";
 import { getTranslation } from "@calcom/lib/server/i18n";
 import { getTimeFormatStringFromUserTimeFormat } from "@calcom/lib/timeFormat";
 import prisma, { bookingMinimalSelect } from "@calcom/prisma";
@@ -30,6 +31,8 @@ import type { CalendarEvent } from "@calcom/types/Calendar";
 import { getAllCredentials } from "./getAllCredentialsForUsersOnEvent/getAllCredentials";
 import cancelAttendeeSeat from "./handleSeats/cancel/cancelAttendeeSeat";
 
+const log = logger.getSubLogger({ prefix: ["handleCancelBooking"] });
+
 async function getBookingToDelete(id: number | undefined, uid: string | undefined) {
   return await prisma.booking.findUnique({
     where: {
@@ -113,7 +116,6 @@ async function getBookingToDelete(id: number | undefined, uid: string | undefine
       destinationCalendar: true,
       smsReminderNumber: true,
       workflowReminders: true,
-      scheduledJobs: true,
       seatsReferences: true,
       responses: true,
       iCalUID: true,
@@ -324,9 +326,9 @@ async function handler(req: CustomRequest) {
   }
 
   let updatedBookings: {
+    id: number;
     uid: string;
     workflowReminders: WorkflowReminder[];
-    scheduledJobs: string[];
     references: {
       type: string;
       credentialId: number | null;
@@ -362,6 +364,7 @@ async function handler(req: CustomRequest) {
         },
       },
       select: {
+        id: true,
         startTime: true,
         endTime: true,
         references: {
@@ -374,7 +377,6 @@ async function handler(req: CustomRequest) {
         },
         workflowReminders: true,
         uid: true,
-        scheduledJobs: true,
       },
     });
     updatedBookings = updatedBookings.concat(allUpdatedBookings);
@@ -398,6 +400,7 @@ async function handler(req: CustomRequest) {
         iCalSequence: evt.iCalSequence || 100,
       },
       select: {
+        id: true,
         startTime: true,
         endTime: true,
         references: {
@@ -410,7 +413,6 @@ async function handler(req: CustomRequest) {
         },
         workflowReminders: true,
         uid: true,
-        scheduledJobs: true,
       },
     });
     updatedBookings.push(updatedBooking);
@@ -441,24 +443,27 @@ async function handler(req: CustomRequest) {
     },
   });
 
-  // delete scheduled jobs of cancelled bookings
-  // FIXME: async calls into ether
-  updatedBookings.forEach((booking) => {
-    cancelScheduledJobs(booking);
-  });
+  const webhookTriggerPromises = [];
+  const workflowReminderPromises = [];
 
-  //Workflows - cancel all reminders for cancelled bookings
-  // FIXME: async calls into ether
-  updatedBookings.forEach((booking) => {
-    booking.workflowReminders.forEach((reminder) => {
+  for (const booking of updatedBookings) {
+    // delete scheduled webhook triggers of cancelled bookings
+    webhookTriggerPromises.push(deleteWebhookScheduledTriggers({ booking }));
+
+    //Workflows - cancel all reminders for cancelled bookings
+    for (const reminder of booking.workflowReminders) {
       if (reminder.method === WorkflowMethods.EMAIL) {
-        deleteScheduledEmailReminder(reminder.id, reminder.referenceId);
+        workflowReminderPromises.push(deleteScheduledEmailReminder(reminder.id, reminder.referenceId));
       } else if (reminder.method === WorkflowMethods.SMS) {
-        deleteScheduledSMSReminder(reminder.id, reminder.referenceId);
+        workflowReminderPromises.push(deleteScheduledSMSReminder(reminder.id, reminder.referenceId));
       } else if (reminder.method === WorkflowMethods.WHATSAPP) {
-        deleteScheduledWhatsappReminder(reminder.id, reminder.referenceId);
+        workflowReminderPromises.push(deleteScheduledWhatsappReminder(reminder.id, reminder.referenceId));
       }
-    });
+    }
+  }
+
+  await Promise.all([...webhookTriggerPromises, ...workflowReminderPromises]).catch((error) => {
+    log.error("An error occurred when deleting workflow reminders and webhook triggers", error);
   });
 
   const prismaPromises: Promise<unknown>[] = [bookingReferenceDeletes];

packages/features/bookings/lib/handleConfirmation.ts

@@ -113,7 +113,6 @@ export async function handleConfirmation(args: {
     }
   }
   let updatedBookings: {
-    scheduledJobs: string[];
     id: number;
     description: string | null;
     location: string | null;
@@ -201,7 +200,6 @@ export async function handleConfirmation(args: {
           smsReminderNumber: true,
           customInputs: true,
           id: true,
-          scheduledJobs: true,
         },
       })
     );
@@ -256,7 +254,6 @@ export async function handleConfirmation(args: {
         location: true,
         customInputs: true,
         id: true,
-        scheduledJobs: true,
       },
     });
     updatedBookings.push(updatedBooking);
@@ -324,17 +321,35 @@ export async function handleConfirmation(args: {
       teamId: booking.eventType?.teamId,
     });
 
+    const scheduleTriggerPromises: Promise<unknown>[] = [];
+
     subscribersMeetingStarted.forEach((subscriber) => {
       updatedBookings.forEach((booking) => {
-        scheduleTrigger(booking, subscriber.subscriberUrl, subscriber, WebhookTriggerEvents.MEETING_STARTED);
+        scheduleTriggerPromises.push(
+          scheduleTrigger({
+            booking,
+            subscriberUrl: subscriber.subscriberUrl,
+            subscriber,
+            triggerEvent: WebhookTriggerEvents.MEETING_STARTED,
+          })
+        );
       });
     });
     subscribersMeetingEnded.forEach((subscriber) => {
       updatedBookings.forEach((booking) => {
-        scheduleTrigger(booking, subscriber.subscriberUrl, subscriber, WebhookTriggerEvents.MEETING_ENDED);
+        scheduleTriggerPromises.push(
+          scheduleTrigger({
+            booking,
+            subscriberUrl: subscriber.subscriberUrl,
+            subscriber,
+            triggerEvent: WebhookTriggerEvents.MEETING_ENDED,
+          })
+        );
       });
     });
 
+    await Promise.all(scheduleTriggerPromises);
+
     const eventTypeInfo: EventTypeInfo = {
       eventTitle: booking.eventType?.title,
       eventDescription: booking.eventType?.description,

packages/features/bookings/lib/handleNewBooking.ts

@@ -53,7 +53,10 @@ import {
 import { getFullName } from "@calcom/features/form-builder/utils";
 import type { GetSubscriberOptions } from "@calcom/features/webhooks/lib/getWebhooks";
 import getWebhooks from "@calcom/features/webhooks/lib/getWebhooks";
-import { cancelScheduledJobs, scheduleTrigger } from "@calcom/features/webhooks/lib/scheduleTrigger";
+import {
+  deleteWebhookScheduledTriggers,
+  scheduleTrigger,
+} from "@calcom/features/webhooks/lib/scheduleTrigger";
 import {
   isPrismaObjOrUndefined,
   parseBookingLimit,
@@ -2269,38 +2272,49 @@ async function handler(
 
   // We are here so, booking doesn't require payment and booking is also created in DB already, through createBooking call
   if (isConfirmedByDefault) {
-    try {
-      const subscribersMeetingEnded = await getWebhooks(subscriberOptionsMeetingEnded);
-      const subscribersMeetingStarted = await getWebhooks(subscriberOptionsMeetingStarted);
+    const subscribersMeetingEnded = await getWebhooks(subscriberOptionsMeetingEnded);
+    const subscribersMeetingStarted = await getWebhooks(subscriberOptionsMeetingStarted);
 
-      subscribersMeetingEnded.forEach((subscriber) => {
-        if (rescheduleUid && originalRescheduledBooking) {
-          cancelScheduledJobs(originalRescheduledBooking, undefined, true);
-        }
-        if (booking && booking.status === BookingStatus.ACCEPTED) {
-          scheduleTrigger(booking, subscriber.subscriberUrl, subscriber, WebhookTriggerEvents.MEETING_ENDED);
-        }
+    let deleteWebhookScheduledTriggerPromise: Promise<unknown> = Promise.resolve();
+    const scheduleTriggerPromises = [];
+
+    if (rescheduleUid && originalRescheduledBooking) {
+      //delete all scheduled triggers for meeting ended and meeting started of booking
+      deleteWebhookScheduledTriggerPromise = deleteWebhookScheduledTriggers({
+        booking: originalRescheduledBooking,
       });
+    }
 
-      subscribersMeetingStarted.forEach((subscriber) => {
-        if (rescheduleUid && originalRescheduledBooking) {
-          cancelScheduledJobs(originalRescheduledBooking, undefined, true);
-        }
-        if (booking && booking.status === BookingStatus.ACCEPTED) {
-          scheduleTrigger(
+    if (booking && booking.status === BookingStatus.ACCEPTED) {
+      for (const subscriber of subscribersMeetingEnded) {
+        scheduleTriggerPromises.push(
+          scheduleTrigger({
             booking,
-            subscriber.subscriberUrl,
+            subscriberUrl: subscriber.subscriberUrl,
             subscriber,
-            WebhookTriggerEvents.MEETING_STARTED
-          );
-        }
-      });
-    } catch (error) {
+            triggerEvent: WebhookTriggerEvents.MEETING_ENDED,
+          })
+        );
+      }
+
+      for (const subscriber of subscribersMeetingStarted) {
+        scheduleTriggerPromises.push(
+          scheduleTrigger({
+            booking,
+            subscriberUrl: subscriber.subscriberUrl,
+            subscriber,
+            triggerEvent: WebhookTriggerEvents.MEETING_STARTED,
+          })
+        );
+      }
+    }
+
+    await Promise.all([deleteWebhookScheduledTriggerPromise, ...scheduleTriggerPromises]).catch((error) => {
       loggerWithEventDetails.error(
-        "Error while running scheduledJobs for booking",
+        "Error while scheduling or canceling webhook triggers",
         JSON.stringify({ error })
       );
-    }
+    });
 
     // Send Webhook call if hooked to BOOKING_CREATED & BOOKING_RESCHEDULED
     await handleWebhookTrigger({ subscriberOptions, eventTrigger, webhookData });

packages/features/bookings/lib/handleSeats/handleSeats.ts

@@ -60,7 +60,6 @@ const handleSeats = async (newSeatedBookingObject: NewSeatedBookingObject) => {
       status: true,
       smsReminderNumber: true,
       endTime: true,
-      scheduledJobs: true,
     },
   });
 

packages/features/bookings/lib/handleSeats/types.d.ts

@@ -53,7 +53,6 @@ export type SeatedBooking = Prisma.BookingGetPayload<{
     status: true;
     smsReminderNumber: true;
     endTime: true;
-    scheduledJobs: true;
   };
 }>;
 

packages/features/webhooks/components/WebhookListItem.tsx

@@ -103,6 +103,7 @@ export default function WebhookListItem(props: {
         <div className="ml-2 flex items-center space-x-4">
           <Switch
             defaultChecked={webhook.active}
+            data-testid="webhook-switch"
             disabled={!canEditWebhook}
             onCheckedChange={() =>
               toggleWebhook.mutate({
@@ -114,7 +115,11 @@ export default function WebhookListItem(props: {
             }
           />
 
-          <Button className="hidden lg:flex" color="secondary" onClick={props.onEditWebhook}>
+          <Button
+            className="hidden lg:flex"
+            color="secondary"
+            onClick={props.onEditWebhook}
+            data-testid="webhook-edit-button">
             {t("edit")}
           </Button>
 

packages/features/webhooks/lib/cron.ts

@@ -1,12 +1,10 @@
 /* Cron job for scheduled webhook events triggers */
 import type { NextApiRequest, NextApiResponse } from "next";
 
-import dayjs from "@calcom/dayjs";
-import logger from "@calcom/lib/logger";
 import { defaultHandler } from "@calcom/lib/server";
 import prisma from "@calcom/prisma";
 
-import { createWebhookSignature, jsonParse } from "./sendPayload";
+import { handleWebhookScheduledTriggers } from "./handleWebhookScheduledTriggers";
 
 async function handler(req: NextApiRequest, res: NextApiResponse) {
   const apiKey = req.headers.authorization || req.query.apiKey;
@@ -15,111 +13,7 @@ async function handler(req: NextApiRequest, res: NextApiResponse) {
     return;
   }
 
-  // make sure to not trigger webhooks that are too far in the past (in case cron job was down for a while)
-  await prisma.webhookScheduledTriggers.deleteMany({
-    where: {
-      startAfter: {
-        lte: dayjs().subtract(1, "day").toISOString(),
-      },
-    },
-  });
-
-  // get jobs that should be run
-  const jobsToRun = await prisma.webhookScheduledTriggers.findMany({
-    where: {
-      startAfter: {
-        lte: dayjs().toISOString(),
-      },
-    },
-    select: {
-      id: true,
-      jobName: true,
-      payload: true,
-      subscriberUrl: true,
-      webhook: {
-        select: {
-          secret: true,
-        },
-      },
-    },
-  });
-
-  const fetchPromises: Promise<any>[] = [];
-
-  // run jobs
-  for (const job of jobsToRun) {
-    // Fetch the webhook configuration so that we can get the secret.
-    const [appId, subscriberId] = job.jobName.split("_");
-    let webhook = job.webhook;
-
-    if (!webhook) {
-      try {
-        webhook = await prisma.webhook.findUniqueOrThrow({
-          where: { id: subscriberId, appId: appId !== "null" ? appId : null },
-        });
-      } catch {
-        logger.error(`Error finding webhook for subscriberId: ${subscriberId}, appId: ${appId}`);
-      }
-    }
-
-    const headers: Record<string, string> = {
-      "Content-Type":
-        !job.payload || jsonParse(job.payload) ? "application/json" : "application/x-www-form-urlencoded",
-    };
-
-    if (webhook) {
-      headers["X-Cal-Signature-256"] = createWebhookSignature({ secret: webhook.secret, body: job.payload });
-    }
-
-    fetchPromises.push(
-      fetch(job.subscriberUrl, {
-        method: "POST",
-        body: job.payload,
-        headers,
-      }).catch((error) => {
-        console.error(`Webhook trigger for subscriber url ${job.subscriberUrl} failed with error: ${error}`);
-      })
-    );
-
-    const parsedJobPayload = JSON.parse(job.payload) as {
-      id: number; // booking id
-      endTime: string;
-      scheduledJobs: string[];
-      triggerEvent: string;
-    };
-
-    // clean finished job
-    await prisma.webhookScheduledTriggers.delete({
-      where: {
-        id: job.id,
-      },
-    });
-
-    const booking = await prisma.booking.findUnique({
-      where: { id: parsedJobPayload.id },
-      select: { id: true, scheduledJobs: true },
-    });
-    if (!booking) {
-      console.log("Error finding booking in webhook trigger:", parsedJobPayload);
-      return res.json({ ok: false });
-    }
-
-    //remove scheduled job from bookings once triggered
-    const updatedScheduledJobs = booking.scheduledJobs.filter((scheduledJob) => {
-      return scheduledJob !== job.jobName;
-    });
-
-    await prisma.booking.update({
-      where: {
-        id: booking.id,
-      },
-      data: {
-        scheduledJobs: updatedScheduledJobs,
-      },
-    });
-  }
-
-  Promise.allSettled(fetchPromises);
+  await handleWebhookScheduledTriggers(prisma);
 
   res.json({ ok: true });
 }

packages/features/webhooks/lib/handleWebhookScheduledTriggers.ts

@@ -0,0 +1,87 @@
+import dayjs from "@calcom/dayjs";
+import logger from "@calcom/lib/logger";
+import type { PrismaClient } from "@calcom/prisma";
+
+import { createWebhookSignature, jsonParse } from "./sendPayload";
+
+export async function handleWebhookScheduledTriggers(prisma: PrismaClient) {
+  await prisma.webhookScheduledTriggers.deleteMany({
+    where: {
+      startAfter: {
+        lte: dayjs().subtract(1, "day").toDate(),
+      },
+    },
+  });
+  // get jobs that should be run
+  const jobsToRun = await prisma.webhookScheduledTriggers.findMany({
+    where: {
+      startAfter: {
+        lte: dayjs().toDate(),
+      },
+    },
+    select: {
+      id: true,
+      jobName: true,
+      payload: true,
+      subscriberUrl: true,
+      webhook: {
+        select: {
+          secret: true,
+        },
+      },
+    },
+  });
+
+  const fetchPromises: Promise<any>[] = [];
+
+  // run jobs
+  for (const job of jobsToRun) {
+    // Fetch the webhook configuration so that we can get the secret.
+    let webhook = job.webhook;
+
+    // only needed to support old jobs that don't have the webhook relationship yet
+    if (!webhook && job.jobName) {
+      const [appId, subscriberId] = job.jobName.split("_");
+      try {
+        webhook = await prisma.webhook.findUniqueOrThrow({
+          where: { id: subscriberId, appId: appId !== "null" ? appId : null },
+        });
+      } catch {
+        logger.error(`Error finding webhook for subscriberId: ${subscriberId}, appId: ${appId}`);
+      }
+    }
+
+    const headers: Record<string, string> = {
+      "Content-Type":
+        !job.payload || jsonParse(job.payload) ? "application/json" : "application/x-www-form-urlencoded",
+    };
+
+    if (webhook) {
+      headers["X-Cal-Signature-256"] = createWebhookSignature({ secret: webhook.secret, body: job.payload });
+    }
+    fetchPromises.push(
+      fetch(job.subscriberUrl, {
+        method: "POST",
+        body: job.payload,
+        headers,
+      }).catch((error) => {
+        console.error(`Webhook trigger for subscriber url ${job.subscriberUrl} failed with error: ${error}`);
+      })
+    );
+
+    const parsedJobPayload = JSON.parse(job.payload) as {
+      id: number; // booking id
+      endTime: string;
+      triggerEvent: string;
+    };
+
+    // clean finished job
+    await prisma.webhookScheduledTriggers.delete({
+      where: {
+        id: job.id,
+      },
+    });
+  }
+
+  Promise.allSettled(fetchPromises);
+}

packages/features/webhooks/lib/scheduleTrigger.ts

@@ -59,7 +59,7 @@ export async function addSubscription({
       if (teamId) {
         where.eventType = { teamId };
       } else {
-        where.userId = userId;
+        where.eventType = { userId };
       }
       const bookings = await prisma.booking.findMany({
         where: {
@@ -72,15 +72,15 @@ export async function addSubscription({
       });
 
       for (const booking of bookings) {
-        scheduleTrigger(
+        scheduleTrigger({
           booking,
-          createSubscription.subscriberUrl,
-          {
+          subscriberUrl: createSubscription.subscriberUrl,
+          subscriber: {
             id: createSubscription.id,
             appId: createSubscription.appId,
           },
-          triggerEvent
-        );
+          triggerEvent,
+        });
       }
     }
 
@@ -111,58 +111,24 @@ export async function deleteSubscription({
     isTeam: boolean;
   } | null;
 }) {
+  const userId = appApiKey ? appApiKey.userId : account && !account.isTeam ? account.id : null;
+  const teamId = appApiKey ? appApiKey.teamId : account && account.isTeam ? account.id : null;
   try {
-    const webhook = await prisma.webhook.findFirst({
-      where: {
-        id: webhookId,
-      },
-    });
-
-    if (webhook?.eventTriggers.includes(WebhookTriggerEvents.MEETING_ENDED)) {
-      const where: Prisma.BookingWhereInput = {};
-
-      if (appApiKey) {
-        if (appApiKey.teamId) {
-          where.eventType = { teamId: appApiKey.teamId };
-        } else {
-          where.userId = appApiKey.userId;
-        }
-      } else if (account) {
-        if (account.isTeam) {
-          where.eventType = { teamId: account.id };
-        } else {
-          where.userId = account.id;
-        }
-      }
-
-      const bookingsWithScheduledJobs = await prisma.booking.findMany({
-        where: {
-          ...where,
-          scheduledJobs: {
-            isEmpty: false,
-          },
-        },
-      });
-      for (const booking of bookingsWithScheduledJobs) {
-        const updatedScheduledJobs = booking.scheduledJobs.filter(
-          (scheduledJob) => scheduledJob !== `${appId}_${webhook.id}`
-        );
-        await prisma.booking.update({
-          where: {
-            id: booking.id,
-          },
-          data: {
-            scheduledJobs: updatedScheduledJobs,
-          },
-        });
-      }
+    let where: Prisma.WebhookWhereInput = {};
+    if (teamId) {
+      where = { teamId };
+    } else {
+      where = { userId };
     }
 
     const deleteWebhook = await prisma.webhook.delete({
       where: {
+        ...where,
+        appId: appId,
         id: webhookId,
       },
     });
+
     if (!deleteWebhook) {
       throw new Error(`Unable to delete webhook ${webhookId}`);
     }
@@ -188,27 +154,16 @@ export async function listBookings(
     isTeam: boolean;
   } | null
 ) {
+  const userId = appApiKey ? appApiKey.userId : account && !account.isTeam ? account.id : null;
+  const teamId = appApiKey ? appApiKey.teamId : account && account.isTeam ? account.id : null;
   try {
     const where: Prisma.BookingWhereInput = {};
-    if (appApiKey) {
-      if (appApiKey.teamId) {
-        where.eventType = {
-          OR: [{ teamId: appApiKey.teamId }, { parent: { teamId: appApiKey.teamId } }],
-        };
-      } else {
-        where.userId = appApiKey.userId;
-      }
-    } else if (account) {
-      if (!account.isTeam) {
-        where.userId = account.id;
-        where.eventType = {
-          teamId: null,
-        };
-      } else {
-        where.eventType = {
-          OR: [{ teamId: account.id }, { parent: { teamId: account.id } }],
-        };
-      }
+    if (teamId) {
+      where.eventType = {
+        OR: [{ teamId }, { parent: { teamId } }],
+      };
+    } else {
+      where.eventType = { userId };
     }
 
     const bookings = await prisma.booking.findMany({
@@ -285,21 +240,24 @@ export async function listBookings(
   }
 }
 
-export async function scheduleTrigger(
-  booking: { id: number; endTime: Date; startTime: Date; scheduledJobs: string[] },
-  subscriberUrl: string,
-  subscriber: { id: string; appId: string | null },
-  triggerEvent: WebhookTriggerEvents
-) {
+export async function scheduleTrigger({
+  booking,
+  subscriberUrl,
+  subscriber,
+  triggerEvent,
+}: {
+  booking: { id: number; endTime: Date; startTime: Date };
+  subscriberUrl: string;
+  subscriber: { id: string; appId: string | null };
+  triggerEvent: WebhookTriggerEvents;
+}) {
   try {
     const payload = JSON.stringify({ triggerEvent, ...booking });
-    const jobName = `${subscriber.appId}_${subscriber.id}`;
 
-    // add scheduled job to database
-    const createTrigger = prisma.webhookScheduledTriggers.create({
+    await prisma.webhookScheduledTriggers.create({
       data: {
-        jobName,
         payload,
+        appId: subscriber.appId,
         startAfter: triggerEvent === WebhookTriggerEvents.MEETING_ENDED ? booking.endTime : booking.startTime,
         subscriberUrl,
         webhook: {
@@ -307,85 +265,69 @@ export async function scheduleTrigger(
             id: subscriber.id,
           },
         },
-      },
-    });
-
-    //add scheduled job name to booking
-    const updateBooking = prisma.booking.update({
-      where: {
-        id: booking.id,
-      },
-      data: {
-        scheduledJobs: {
-          push: jobName,
+        booking: {
+          connect: {
+            id: booking.id,
+          },
         },
       },
     });
-
-    await prisma.$transaction([createTrigger, updateBooking]);
   } catch (error) {
     console.error("Error cancelling scheduled jobs", error);
   }
 }
 
-export async function cancelScheduledJobs(
-  booking: { uid: string; scheduledJobs?: string[] },
-  appId?: string | null,
-  isReschedule?: boolean,
-  triggerEvent?: WebhookTriggerEvents,
-  webhookId?: string
-) {
-  if (!booking.scheduledJobs) return;
-
-  let scheduledJobs = booking.scheduledJobs || [];
-  const promises = booking.scheduledJobs.map(async (scheduledJob) => {
-    if (appId) {
-      if (scheduledJob.startsWith(appId)) {
-        await prisma.webhookScheduledTriggers.deleteMany({
-          where: {
-            jobName: scheduledJob,
-          },
-        });
-        scheduledJobs = scheduledJobs?.filter((job) => scheduledJob !== job) || [];
-      }
-    } else {
-      //if no specific appId given, delete all scheduled jobs of booking
-      if (triggerEvent) {
-        const shouldContain = `"triggerEvent":"${triggerEvent}"`;
-        await prisma.webhookScheduledTriggers.deleteMany({
-          where: {
-            payload: {
-              contains: shouldContain,
-            },
-            webhookId: webhookId,
-          },
-        });
+export async function deleteWebhookScheduledTriggers({
+  booking,
+  appId,
+  triggerEvent,
+  webhookId,
+  userId,
+  teamId,
+}: {
+  booking?: { id: number; uid: string };
+  appId?: string | null;
+  triggerEvent?: WebhookTriggerEvents;
+  webhookId?: string;
+  userId?: number;
+  teamId?: number;
+}) {
+  try {
+    if (appId && (userId || teamId)) {
+      const where: Prisma.BookingWhereInput = {};
+      if (userId) {
+        where.eventType = { userId };
       } else {
-        await prisma.webhookScheduledTriggers.deleteMany({
-          where: {
-            jobName: scheduledJob,
-          },
-        });
-        scheduledJobs = [];
+        where.eventType = { teamId };
       }
-    }
-
-    if (!isReschedule) {
-      await prisma.booking.update({
+      await prisma.webhookScheduledTriggers.deleteMany({
         where: {
-          uid: booking.uid,
-        },
-        data: {
-          scheduledJobs: scheduledJobs,
+          appId: appId,
+          booking: where,
         },
       });
-    }
-  });
+    } else {
+      if (booking) {
+        await prisma.webhookScheduledTriggers.deleteMany({
+          where: {
+            bookingId: booking.id,
+          },
+        });
+      } else if (webhookId) {
+        const where: Prisma.WebhookScheduledTriggersWhereInput = { webhookId: webhookId };
 
-  try {
-    await Promise.all(promises);
+        if (triggerEvent) {
+          const shouldContain = `"triggerEvent":"${triggerEvent}"`;
+          where.payload = { contains: shouldContain };
+        }
+
+        await prisma.webhookScheduledTriggers.deleteMany({
+          where,
+        });
+      }
+    }
   } catch (error) {
-    console.error("Error cancelling scheduled jobs", error);
+    console.error("Error deleting webhookScheduledTriggers ", error);
   }
 }
 
@@ -442,27 +384,16 @@ export async function updateTriggerForExistingBookings(
 
   if (addedEventTriggers.length > 0) {
     const promise = bookings.map((booking) => {
-      return addedEventTriggers.map((trigger) => {
-        scheduleTrigger(booking, webhook.subscriberUrl, webhook, trigger);
+      return addedEventTriggers.map((triggerEvent) => {
+        scheduleTrigger({ booking, subscriberUrl: webhook.subscriberUrl, subscriber: webhook, triggerEvent });
       });
     });
 
     await Promise.all(promise);
   }
 
-  if (removedEventTriggers.length > 0) {
-    const promise = bookings.map((booking) => {
-      removedEventTriggers.map((trigger) =>
-        cancelScheduledJobs(
-          { uid: booking.uid, scheduledJobs: booking.scheduledJobs },
-          undefined,
-          false,
-          trigger,
-          webhook.id
-        )
-      );
-    });
-
-    await Promise.all(promise);
-  }
+  const promise = removedEventTriggers.map((triggerEvent) =>
+    deleteWebhookScheduledTriggers({ triggerEvent, webhookId: webhook.id })
+  );
+  await Promise.all(promise);
 }

packages/features/webhooks/lib/test/webhooks.test.ts

@@ -0,0 +1,72 @@
+import prismock from "../../../../../tests/libs/__mocks__/prisma";
+
+import { describe, expect, beforeEach } from "vitest";
+
+import dayjs from "@calcom/dayjs";
+import { test } from "@calcom/web/test/fixtures/fixtures";
+import { expectWebhookToHaveBeenCalledWith } from "@calcom/web/test/utils/bookingScenario/expects";
+
+import { handleWebhookScheduledTriggers } from "../handleWebhookScheduledTriggers";
+
+describe("Cron job handler", () => {
+  beforeEach(async () => {
+    await prismock.webhookScheduledTriggers.deleteMany();
+  });
+  test(`should delete old webhook scheduled triggers`, async () => {
+    const now = dayjs();
+    await prismock.webhookScheduledTriggers.createMany({
+      data: [
+        {
+          id: 1,
+          subscriberUrl: "https://example.com",
+          startAfter: now.subtract(2, "day").toDate(),
+          payload: "",
+        },
+        {
+          id: 1,
+          subscriberUrl: "https://example.com",
+          startAfter: now.subtract(1, "day").subtract(1, "hour").toDate(),
+          payload: "",
+        },
+        {
+          id: 2,
+          subscriberUrl: "https://example.com",
+          startAfter: now.add(1, "day").toDate(),
+          payload: "",
+        },
+      ],
+    });
+
+    await handleWebhookScheduledTriggers(prismock);
+
+    const scheduledTriggers = await prismock.webhookScheduledTriggers.findMany();
+    expect(scheduledTriggers.length).toBe(1);
+    expect(scheduledTriggers[0].startAfter).toStrictEqual(now.add(1, "day").toDate());
+  });
+  test(`should trigger if current date is after startAfter`, async () => {
+    const now = dayjs();
+    const payload = `{"triggerEvent":"MEETING_ENDED"}`;
+    await prismock.webhookScheduledTriggers.createMany({
+      data: [
+        {
+          id: 1,
+          subscriberUrl: "https://example.com",
+          startAfter: now.add(5, "minute").toDate(),
+          payload,
+        },
+        {
+          id: 2,
+          subscriberUrl: "https://example.com/test",
+          startAfter: now.subtract(5, "minute").toDate(),
+          payload,
+        },
+      ],
+    });
+    await handleWebhookScheduledTriggers(prismock);
+
+    expectWebhookToHaveBeenCalledWith("https://example.com/test", { triggerEvent: "MEETING_ENDED", payload });
+    expect(() =>
+      expectWebhookToHaveBeenCalledWith("https://example.com", { triggerEvent: "MEETING_ENDED", payload })
+    ).toThrow("Webhook not sent to https://example.com for MEETING_ENDED. All webhooks: []");
+  });
+});

packages/prisma/migrations/20240502213807_add_webhook_scheduled_triggers_to_bookings/migration.sql

@@ -0,0 +1,7 @@
+-- AlterTable
+ALTER TABLE "WebhookScheduledTriggers" ADD COLUMN     "appId" TEXT,
+ADD COLUMN     "bookingId" INTEGER,
+ALTER COLUMN "jobName" DROP NOT NULL;
+
+-- AddForeignKey
+ALTER TABLE "WebhookScheduledTriggers" ADD CONSTRAINT "WebhookScheduledTriggers_bookingId_fkey" FOREIGN KEY ("bookingId") REFERENCES "Booking"("id") ON DELETE CASCADE ON UPDATE CASCADE;

packages/prisma/schema.prisma

@@ -523,17 +523,17 @@ enum BookingStatus {
 }
 
 model Booking {
-  id                    Int                  @id @default(autoincrement())
-  uid                   String               @unique
+  id                    Int                        @id @default(autoincrement())
+  uid                   String                     @unique
   // (optional) UID based on slot start/end time & email against duplicates
-  idempotencyKey        String?              @unique
-  user                  User?                @relation(fields: [userId], references: [id], onDelete: Cascade)
+  idempotencyKey        String?                    @unique
+  user                  User?                      @relation(fields: [userId], references: [id], onDelete: Cascade)
   userId                Int?
   // User's email at the time of booking
   /// @zod.email()
   userPrimaryEmail      String?
   references            BookingReference[]
-  eventType             EventType?           @relation(fields: [eventTypeId], references: [id])
+  eventType             EventType?                 @relation(fields: [eventTypeId], references: [id])
   eventTypeId           Int?
   title                 String
   description           String?
@@ -544,12 +544,12 @@ model Booking {
   endTime               DateTime
   attendees             Attendee[]
   location              String?
-  createdAt             DateTime             @default(now())
+  createdAt             DateTime                   @default(now())
   updatedAt             DateTime?
-  status                BookingStatus        @default(ACCEPTED)
-  paid                  Boolean              @default(false)
+  status                BookingStatus              @default(ACCEPTED)
+  paid                  Boolean                    @default(false)
   payment               Payment[]
-  destinationCalendar   DestinationCalendar? @relation(fields: [destinationCalendarId], references: [id])
+  destinationCalendar   DestinationCalendar?       @relation(fields: [destinationCalendarId], references: [id])
   destinationCalendarId Int?
   cancellationReason    String?
   rejectionReason       String?
@@ -560,17 +560,18 @@ model Booking {
   recurringEventId      String?
   smsReminderNumber     String?
   workflowReminders     WorkflowReminder[]
-  scheduledJobs         String[]
+  scheduledJobs         String[] // scheduledJobs is deprecated, please use scheduledTriggers instead
   seatsReferences       BookingSeat[]
   /// @zod.custom(imports.bookingMetadataSchema)
   metadata              Json?
-  isRecorded            Boolean              @default(false)
-  iCalUID               String?              @default("")
-  iCalSequence          Int                  @default(0)
+  isRecorded            Boolean                    @default(false)
+  iCalUID               String?                    @default("")
+  iCalSequence          Int                        @default(0)
   instantMeetingToken   InstantMeetingToken?
   rating                Int?
   ratingFeedback        String?
   noShowHost            Boolean?
+  scheduledTriggers     WebhookScheduledTriggers[]
 
   @@index([eventTypeId])
   @@index([userId])
@@ -1006,14 +1007,17 @@ model WorkflowReminder {
 
 model WebhookScheduledTriggers {
   id            Int       @id @default(autoincrement())
-  jobName       String
+  jobName       String? // jobName is deprecated, not needed when webhook and booking is set
   subscriberUrl String
   payload       String
   startAfter    DateTime
   retryCount    Int       @default(0)
   createdAt     DateTime? @default(now())
+  appId         String?
   webhookId     String?
   webhook       Webhook?  @relation(fields: [webhookId], references: [id], onDelete: Cascade)
+  bookingId     Int?
+  booking       Booking?  @relation(fields: [bookingId], references: [id], onDelete: Cascade)
 }
 
 enum WorkflowTemplates {

packages/trpc/server/routers/loggedInViewer/connectAndJoin.handler.ts

@@ -135,7 +135,6 @@ export const Handler = async ({ ctx, input }: Options) => {
       id: true,
       uid: true,
       status: true,
-      scheduledJobs: true,
     },
   });
 

packages/trpc/server/routers/loggedInViewer/deleteCredential.handler.ts

@@ -4,7 +4,7 @@ import { getCalendar } from "@calcom/app-store/_utils/getCalendar";
 import { DailyLocationType } from "@calcom/core/location";
 import { sendCancelledEmails } from "@calcom/emails";
 import { getCalEventResponses } from "@calcom/features/bookings/lib/getCalEventResponses";
-import { cancelScheduledJobs } from "@calcom/features/webhooks/lib/scheduleTrigger";
+import { deleteWebhookScheduledTriggers } from "@calcom/features/webhooks/lib/scheduleTrigger";
 import { isPrismaObjOrUndefined, parseRecurringEvent } from "@calcom/lib";
 import { deletePayment } from "@calcom/lib/payment/deletePayment";
 import { getTranslation } from "@calcom/lib/server/i18n";
@@ -318,17 +318,12 @@ export const deleteCredentialHandler = async ({ ctx, input }: DeleteCredentialOp
         appId: "zapier",
       },
     });
-    const bookingsWithScheduledJobs = await prisma.booking.findMany({
-      where: {
-        userId: ctx.user.id,
-        scheduledJobs: {
-          isEmpty: false,
-        },
-      },
+
+    deleteWebhookScheduledTriggers({
+      appId: credential.appId,
+      userId: teamId ? undefined : ctx.user.id,
+      teamId,
     });
-    for (const booking of bookingsWithScheduledJobs) {
-      cancelScheduledJobs(booking, credential.appId);
-    }
   }
 
   // Backwards compatibility. Selected calendars cascade on delete when deleting a credential

packages/trpc/server/routers/viewer/bookings/confirm.handler.ts

@@ -98,7 +98,6 @@ export const confirmHandler = async ({ ctx, input }: ConfirmOptions) => {
       recurringEventId: true,
       status: true,
       smsReminderNumber: true,
-      scheduledJobs: true,
     },
   });
 

packages/trpc/server/routers/viewer/bookings/requestReschedule.handler.ts

@@ -12,7 +12,7 @@ import { deleteScheduledWhatsappReminder } from "@calcom/ee/workflows/lib/remind
 import { sendRequestRescheduleEmail } from "@calcom/emails";
 import { getCalEventResponses } from "@calcom/features/bookings/lib/getCalEventResponses";
 import getWebhooks from "@calcom/features/webhooks/lib/getWebhooks";
-import { cancelScheduledJobs } from "@calcom/features/webhooks/lib/scheduleTrigger";
+import { deleteWebhookScheduledTriggers } from "@calcom/features/webhooks/lib/scheduleTrigger";
 import sendPayload from "@calcom/features/webhooks/lib/sendOrSchedulePayload";
 import { isPrismaObjOrUndefined } from "@calcom/lib";
 import { getBookerBaseUrl } from "@calcom/lib/getBookerUrl/server";
@@ -71,7 +71,6 @@ export const requestRescheduleHandler = async ({ ctx, input }: RequestReschedule
       dynamicGroupSlugRef: true,
       destinationCalendar: true,
       smsReminderNumber: true,
-      scheduledJobs: true,
       workflowReminders: true,
       responses: true,
       iCalUID: true,
@@ -149,19 +148,22 @@ export const requestRescheduleHandler = async ({ ctx, input }: RequestReschedule
   });
 
   // delete scheduled jobs of previous booking
-  // FIXME: async fn off into the ether
-  cancelScheduledJobs(bookingToReschedule);
+  const webhookWorkflowPromises = [];
+  webhookWorkflowPromises.push(deleteWebhookScheduledTriggers({ booking: bookingToReschedule }));
 
   //cancel workflow reminders of previous booking
-  // FIXME: more async fns off into the ether
-  bookingToReschedule.workflowReminders.forEach((reminder) => {
+  for (const reminder of bookingToReschedule.workflowReminders) {
     if (reminder.method === WorkflowMethods.EMAIL) {
-      deleteScheduledEmailReminder(reminder.id, reminder.referenceId);
+      webhookWorkflowPromises.push(deleteScheduledEmailReminder(reminder.id, reminder.referenceId));
     } else if (reminder.method === WorkflowMethods.SMS) {
-      deleteScheduledSMSReminder(reminder.id, reminder.referenceId);
+      webhookWorkflowPromises.push(deleteScheduledSMSReminder(reminder.id, reminder.referenceId));
     } else if (reminder.method === WorkflowMethods.WHATSAPP) {
-      deleteScheduledWhatsappReminder(reminder.id, reminder.referenceId);
+      webhookWorkflowPromises.push(deleteScheduledWhatsappReminder(reminder.id, reminder.referenceId));
     }
+  }
+
+  await Promise.all(webhookWorkflowPromises).catch((error) => {
+    log.error("Error while scheduling or canceling webhook triggers", JSON.stringify({ error }));
   });
 
   const [mainAttendee] = bookingToReschedule.attendees;

packages/trpc/server/routers/viewer/webhook/edit.handler.ts

@@ -1,4 +1,7 @@
-import { updateTriggerForExistingBookings } from "@calcom/features/webhooks/lib/scheduleTrigger";
+import {
+  updateTriggerForExistingBookings,
+  deleteWebhookScheduledTriggers,
+} from "@calcom/features/webhooks/lib/scheduleTrigger";
 import { prisma } from "@calcom/prisma";
 import type { TrpcSessionUser } from "@calcom/trpc/server/trpc";
 
@@ -40,7 +43,12 @@ export const editHandler = async ({ input, ctx }: EditOptions) => {
     data,
   });
 
-  await updateTriggerForExistingBookings(webhook, webhook.eventTriggers, updatedWebhook.eventTriggers);
+  if (data.active) {
+    const activeTriggersBefore = webhook.active ? webhook.eventTriggers : [];
+    await updateTriggerForExistingBookings(webhook, activeTriggersBefore, updatedWebhook.eventTriggers);
+  } else if (!data.active && webhook.active) {
+    await deleteWebhookScheduledTriggers({ webhookId: webhook.id });
+  }
 
   return updatedWebhook;
 };