From fd64b0f46823c89c4a99d1947ae253cafd3fcc2c Mon Sep 17 00:00:00 2001 From: Carina Wollendorfer <30310907+CarinaWolli@users.noreply.github.com> Date: Fri, 10 May 2024 15:49:16 -0400 Subject: [PATCH] fix: meeting ended and started webhooks trigger (#14864) Co-authored-by: CarinaWolli Co-authored-by: Keith Williams --- apps/web/playwright/webhook.e2e.ts | 126 +++++++++ .../web/test/utils/bookingScenario/expects.ts | 22 +- .../bookings/lib/handleCancelBooking.ts | 41 +-- .../bookings/lib/handleConfirmation.ts | 25 +- .../features/bookings/lib/handleNewBooking.ts | 64 +++-- .../bookings/lib/handleSeats/handleSeats.ts | 1 - .../bookings/lib/handleSeats/types.d.ts | 1 - .../webhooks/components/WebhookListItem.tsx | 7 +- packages/features/webhooks/lib/cron.ts | 110 +------- .../lib/handleWebhookScheduledTriggers.ts | 87 ++++++ .../features/webhooks/lib/scheduleTrigger.ts | 251 +++++++----------- .../webhooks/lib/test/webhooks.test.ts | 72 +++++ .../migration.sql | 7 + packages/prisma/schema.prisma | 32 ++- .../loggedInViewer/connectAndJoin.handler.ts | 1 - .../deleteCredential.handler.ts | 17 +- .../viewer/bookings/confirm.handler.ts | 1 - .../bookings/requestReschedule.handler.ts | 20 +- .../routers/viewer/webhook/edit.handler.ts | 12 +- 19 files changed, 527 insertions(+), 370 deletions(-) create mode 100644 packages/features/webhooks/lib/handleWebhookScheduledTriggers.ts create mode 100644 packages/features/webhooks/lib/test/webhooks.test.ts create mode 100644 packages/prisma/migrations/20240502213807_add_webhook_scheduled_triggers_to_bookings/migration.sql diff --git a/apps/web/playwright/webhook.e2e.ts b/apps/web/playwright/webhook.e2e.ts index 7987556a3b..f02556a706 100644 --- a/apps/web/playwright/webhook.e2e.ts +++ b/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(); diff --git a/apps/web/test/utils/bookingScenario/expects.ts b/apps/web/test/utils/bookingScenario/expects.ts index 3600a4adca..17af1c6d1a 100644 --- a/apps/web/test/utils/bookingScenario/expects.ts +++ b/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)); } } diff --git a/packages/features/bookings/lib/handleCancelBooking.ts b/packages/features/bookings/lib/handleCancelBooking.ts index c7ca758fc5..badb3241e0 100644 --- a/packages/features/bookings/lib/handleCancelBooking.ts +++ b/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[] = [bookingReferenceDeletes]; diff --git a/packages/features/bookings/lib/handleConfirmation.ts b/packages/features/bookings/lib/handleConfirmation.ts index fb85e1b319..380f74b293 100644 --- a/packages/features/bookings/lib/handleConfirmation.ts +++ b/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[] = []; + 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, diff --git a/packages/features/bookings/lib/handleNewBooking.ts b/packages/features/bookings/lib/handleNewBooking.ts index 9abe3bb644..c0c96b7c84 100644 --- a/packages/features/bookings/lib/handleNewBooking.ts +++ b/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 = 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 }); diff --git a/packages/features/bookings/lib/handleSeats/handleSeats.ts b/packages/features/bookings/lib/handleSeats/handleSeats.ts index 81d8e62fd2..0ab74d9c42 100644 --- a/packages/features/bookings/lib/handleSeats/handleSeats.ts +++ b/packages/features/bookings/lib/handleSeats/handleSeats.ts @@ -60,7 +60,6 @@ const handleSeats = async (newSeatedBookingObject: NewSeatedBookingObject) => { status: true, smsReminderNumber: true, endTime: true, - scheduledJobs: true, }, }); diff --git a/packages/features/bookings/lib/handleSeats/types.d.ts b/packages/features/bookings/lib/handleSeats/types.d.ts index b26ca2982c..28b5aaca0e 100644 --- a/packages/features/bookings/lib/handleSeats/types.d.ts +++ b/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; }; }>; diff --git a/packages/features/webhooks/components/WebhookListItem.tsx b/packages/features/webhooks/components/WebhookListItem.tsx index e9b5103220..6371899a1f 100644 --- a/packages/features/webhooks/components/WebhookListItem.tsx +++ b/packages/features/webhooks/components/WebhookListItem.tsx @@ -103,6 +103,7 @@ export default function WebhookListItem(props: {
toggleWebhook.mutate({ @@ -114,7 +115,11 @@ export default function WebhookListItem(props: { } /> - diff --git a/packages/features/webhooks/lib/cron.ts b/packages/features/webhooks/lib/cron.ts index 667e2fd21f..b099612a07 100644 --- a/packages/features/webhooks/lib/cron.ts +++ b/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[] = []; - - // 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 = { - "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 }); } diff --git a/packages/features/webhooks/lib/handleWebhookScheduledTriggers.ts b/packages/features/webhooks/lib/handleWebhookScheduledTriggers.ts new file mode 100644 index 0000000000..cc483d4861 --- /dev/null +++ b/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[] = []; + + // 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 = { + "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); +} diff --git a/packages/features/webhooks/lib/scheduleTrigger.ts b/packages/features/webhooks/lib/scheduleTrigger.ts index cb393d032f..8054e14000 100644 --- a/packages/features/webhooks/lib/scheduleTrigger.ts +++ b/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); } diff --git a/packages/features/webhooks/lib/test/webhooks.test.ts b/packages/features/webhooks/lib/test/webhooks.test.ts new file mode 100644 index 0000000000..f01f58b3c8 --- /dev/null +++ b/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: []"); + }); +}); diff --git a/packages/prisma/migrations/20240502213807_add_webhook_scheduled_triggers_to_bookings/migration.sql b/packages/prisma/migrations/20240502213807_add_webhook_scheduled_triggers_to_bookings/migration.sql new file mode 100644 index 0000000000..d5daed908d --- /dev/null +++ b/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; diff --git a/packages/prisma/schema.prisma b/packages/prisma/schema.prisma index b409550b21..e2ab07595a 100644 --- a/packages/prisma/schema.prisma +++ b/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 { diff --git a/packages/trpc/server/routers/loggedInViewer/connectAndJoin.handler.ts b/packages/trpc/server/routers/loggedInViewer/connectAndJoin.handler.ts index 32fc2c70f9..41b94ecca2 100644 --- a/packages/trpc/server/routers/loggedInViewer/connectAndJoin.handler.ts +++ b/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, }, }); diff --git a/packages/trpc/server/routers/loggedInViewer/deleteCredential.handler.ts b/packages/trpc/server/routers/loggedInViewer/deleteCredential.handler.ts index f3b6351aa8..88dbfdb389 100644 --- a/packages/trpc/server/routers/loggedInViewer/deleteCredential.handler.ts +++ b/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 diff --git a/packages/trpc/server/routers/viewer/bookings/confirm.handler.ts b/packages/trpc/server/routers/viewer/bookings/confirm.handler.ts index 4e35c5f157..65a29612b1 100644 --- a/packages/trpc/server/routers/viewer/bookings/confirm.handler.ts +++ b/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, }, }); diff --git a/packages/trpc/server/routers/viewer/bookings/requestReschedule.handler.ts b/packages/trpc/server/routers/viewer/bookings/requestReschedule.handler.ts index 2de95d325d..f22ebc0fde 100644 --- a/packages/trpc/server/routers/viewer/bookings/requestReschedule.handler.ts +++ b/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; diff --git a/packages/trpc/server/routers/viewer/webhook/edit.handler.ts b/packages/trpc/server/routers/viewer/webhook/edit.handler.ts index 3d5fbcc1cc..8e68f6ce8e 100644 --- a/packages/trpc/server/routers/viewer/webhook/edit.handler.ts +++ b/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; };