diff --git a/packages/features/di/webhooks/Webhooks.tokens.ts b/packages/features/di/webhooks/Webhooks.tokens.ts index 7809de84e7..6e0cb344f6 100644 --- a/packages/features/di/webhooks/Webhooks.tokens.ts +++ b/packages/features/di/webhooks/Webhooks.tokens.ts @@ -1,4 +1,7 @@ +import { WEBHOOK_TASKER_DI_TOKENS } from "./tasker/tokens"; + export const WEBHOOK_TOKENS = { + ...WEBHOOK_TASKER_DI_TOKENS, // Core interfaces WEBHOOK_SERVICE: Symbol("IWebhookService"), BOOKING_WEBHOOK_SERVICE: Symbol("IBookingWebhookService"), diff --git a/packages/features/di/webhooks/containers/webhook.ts b/packages/features/di/webhooks/containers/webhook.ts index 267aa3ea39..fdc6f6f675 100644 --- a/packages/features/di/webhooks/containers/webhook.ts +++ b/packages/features/di/webhooks/containers/webhook.ts @@ -21,7 +21,7 @@ import { oooWebhookDataFetcherModule } from "../modules/OOOWebhookDataFetcher.mo import { paymentWebhookDataFetcherModule } from "../modules/PaymentWebhookDataFetcher.module"; import { recordingWebhookDataFetcherModule } from "../modules/RecordingWebhookDataFetcher.module"; import { webhookModule } from "../modules/Webhook.module"; -import { webhookProducerServiceModule } from "../modules/WebhookProducerService.module"; +import { moduleLoader as webhookProducerServiceModuleLoader } from "../modules/WebhookProducerService.module"; import { webhookTaskConsumerModule } from "../modules/WebhookTaskConsumer.module"; import { WEBHOOK_TOKENS } from "../Webhooks.tokens"; @@ -53,7 +53,8 @@ webhookContainer.load(WEBHOOK_TOKENS.RECORDING_DATA_FETCHER, recordingWebhookDat webhookContainer.load(WEBHOOK_TOKENS.OOO_DATA_FETCHER, oooWebhookDataFetcherModule); // Load Producer/Consumer modules -webhookContainer.load(WEBHOOK_TOKENS.WEBHOOK_PRODUCER_SERVICE, webhookProducerServiceModule); +// Use moduleLoader pattern for producer service to load WebhookTasker dependencies +webhookProducerServiceModuleLoader.loadModule(webhookContainer); webhookContainer.load(WEBHOOK_TOKENS.WEBHOOK_TASK_CONSUMER, webhookTaskConsumerModule); export { webhookContainer }; diff --git a/packages/features/di/webhooks/modules/WebhookProducerService.module.ts b/packages/features/di/webhooks/modules/WebhookProducerService.module.ts index 1671d0058c..42743f5ec5 100644 --- a/packages/features/di/webhooks/modules/WebhookProducerService.module.ts +++ b/packages/features/di/webhooks/modules/WebhookProducerService.module.ts @@ -1,18 +1,38 @@ -import { createModule } from "@evyweb/ioctopus"; - +import { bindModuleToClassOnToken, createModule, type ModuleLoader } from "@calcom/features/di/di"; +import { moduleLoader as loggerServiceModule } from "@calcom/features/di/shared/services/logger.service"; import { WebhookTaskerProducerService } from "@calcom/features/webhooks/lib/service/WebhookTaskerProducerService"; -import { SHARED_TOKENS } from "../../shared/shared.tokens"; +import { moduleLoader as webhookTaskerModule } from "../tasker/WebhookTasker.module"; import { WEBHOOK_TOKENS } from "../Webhooks.tokens"; /** * Producer Service Module - * + * * Binds the lightweight WebhookTaskerProducerService. - * Dependencies: Only Tasker and Logger (no heavy deps). + * Dependencies: WebhookTasker and Logger. + * + * The WebhookTasker automatically handles async/sync mode selection: + * - Production: Queues to Trigger.dev for background processing + * - E2E Tests: Executes immediately via WebhookSyncTasker */ -export const webhookProducerServiceModule = createModule(); +const thisModule = createModule(); +const token = WEBHOOK_TOKENS.WEBHOOK_PRODUCER_SERVICE; +const moduleToken = WEBHOOK_TOKENS.WEBHOOK_PRODUCER_SERVICE_MODULE; -webhookProducerServiceModule - .bind(WEBHOOK_TOKENS.WEBHOOK_PRODUCER_SERVICE) - .toClass(WebhookTaskerProducerService, [SHARED_TOKENS.TASKER, SHARED_TOKENS.LOGGER]); +const loadModule = bindModuleToClassOnToken({ + module: thisModule, + moduleToken, + token, + classs: WebhookTaskerProducerService, + depsMap: { + webhookTasker: webhookTaskerModule, + logger: loggerServiceModule, + }, +}); + +export const moduleLoader = { + token, + loadModule, +} satisfies ModuleLoader; + +export const webhookProducerServiceModule = thisModule; diff --git a/packages/features/di/webhooks/tasker/WebhookSyncTasker.module.ts b/packages/features/di/webhooks/tasker/WebhookSyncTasker.module.ts new file mode 100644 index 0000000000..58a303cc34 --- /dev/null +++ b/packages/features/di/webhooks/tasker/WebhookSyncTasker.module.ts @@ -0,0 +1,24 @@ +import { bindModuleToClassOnToken, createModule, type ModuleLoader } from "@calcom/features/di/di"; +import { WebhookSyncTasker } from "@calcom/features/webhooks/lib/tasker/WebhookSyncTasker"; + +import { moduleLoader as webhookTaskConsumerModuleLoader } from "./WebhookTaskConsumer.module"; +import { WEBHOOK_TASKER_DI_TOKENS } from "./tokens"; + +const thisModule = createModule(); +const token = WEBHOOK_TASKER_DI_TOKENS.WEBHOOK_SYNC_TASKER; +const moduleToken = WEBHOOK_TASKER_DI_TOKENS.WEBHOOK_SYNC_TASKER_MODULE; + +const loadModule = bindModuleToClassOnToken({ + module: thisModule, + moduleToken, + token, + classs: WebhookSyncTasker, + depsMap: { + webhookTaskConsumer: webhookTaskConsumerModuleLoader, + }, +}); + +export const moduleLoader = { + token, + loadModule, +} satisfies ModuleLoader; diff --git a/packages/features/di/webhooks/tasker/WebhookTaskConsumer.module.ts b/packages/features/di/webhooks/tasker/WebhookTaskConsumer.module.ts new file mode 100644 index 0000000000..b9ef9c4be1 --- /dev/null +++ b/packages/features/di/webhooks/tasker/WebhookTaskConsumer.module.ts @@ -0,0 +1,52 @@ +import type { Container } from "@evyweb/ioctopus"; + +import type { ModuleLoader } from "@calcom/features/di/di"; +import type { WebhookTaskConsumer } from "@calcom/features/webhooks/lib/service/WebhookTaskConsumer"; + +import { moduleLoader as loggerModuleLoader } from "../../shared/services/logger.service"; +import { moduleLoader as prismaModuleLoader } from "../../modules/Prisma"; +import { taskerServiceModule } from "../../shared/services/tasker.service"; +import { SHARED_TOKENS } from "../../shared/shared.tokens"; +import { bookingWebhookDataFetcherModule } from "../modules/BookingWebhookDataFetcher.module"; +import { formWebhookDataFetcherModule } from "../modules/FormWebhookDataFetcher.module"; +import { oooWebhookDataFetcherModule } from "../modules/OOOWebhookDataFetcher.module"; +import { paymentWebhookDataFetcherModule } from "../modules/PaymentWebhookDataFetcher.module"; +import { recordingWebhookDataFetcherModule } from "../modules/RecordingWebhookDataFetcher.module"; +import { webhookModule } from "../modules/Webhook.module"; +import { webhookTaskConsumerModule } from "../modules/WebhookTaskConsumer.module"; +import { WEBHOOK_TOKENS } from "../Webhooks.tokens"; + +const token = WEBHOOK_TOKENS.WEBHOOK_TASK_CONSUMER; + +const loadModule = (container: Container) => { + loggerModuleLoader.loadModule(container); + prismaModuleLoader.loadModule(container); + container.load(SHARED_TOKENS.TASKER, taskerServiceModule); + + container.load(WEBHOOK_TOKENS.WEBHOOK_EVENT_TYPE_REPOSITORY, webhookModule); + container.load(WEBHOOK_TOKENS.WEBHOOK_USER_REPOSITORY, webhookModule); + container.load(WEBHOOK_TOKENS.WEBHOOK_REPOSITORY, webhookModule); + container.load(WEBHOOK_TOKENS.WEBHOOK_SERVICE, webhookModule); + container.load(WEBHOOK_TOKENS.BOOKING_WEBHOOK_SERVICE, webhookModule); + container.load(WEBHOOK_TOKENS.FORM_WEBHOOK_SERVICE, webhookModule); + container.load(WEBHOOK_TOKENS.RECORDING_WEBHOOK_SERVICE, webhookModule); + container.load(WEBHOOK_TOKENS.OOO_WEBHOOK_SERVICE, webhookModule); + container.load(WEBHOOK_TOKENS.PAYLOAD_BUILDER_FACTORY, webhookModule); + container.load(WEBHOOK_TOKENS.WEBHOOK_NOTIFICATION_HANDLER, webhookModule); + container.load(WEBHOOK_TOKENS.WEBHOOK_NOTIFIER, webhookModule); + + container.load(WEBHOOK_TOKENS.BOOKING_DATA_FETCHER, bookingWebhookDataFetcherModule); + container.load(WEBHOOK_TOKENS.PAYMENT_DATA_FETCHER, paymentWebhookDataFetcherModule); + container.load(WEBHOOK_TOKENS.FORM_DATA_FETCHER, formWebhookDataFetcherModule); + container.load(WEBHOOK_TOKENS.RECORDING_DATA_FETCHER, recordingWebhookDataFetcherModule); + container.load(WEBHOOK_TOKENS.OOO_DATA_FETCHER, oooWebhookDataFetcherModule); + + container.load(WEBHOOK_TOKENS.WEBHOOK_TASK_CONSUMER, webhookTaskConsumerModule); +}; + +export const moduleLoader = { + token, + loadModule, +} satisfies ModuleLoader; + +export type { WebhookTaskConsumer }; diff --git a/packages/features/di/webhooks/tasker/WebhookTasker.module.ts b/packages/features/di/webhooks/tasker/WebhookTasker.module.ts new file mode 100644 index 0000000000..ce2c3c2772 --- /dev/null +++ b/packages/features/di/webhooks/tasker/WebhookTasker.module.ts @@ -0,0 +1,30 @@ +import { bindModuleToClassOnToken, createModule, type ModuleLoader } from "@calcom/features/di/di"; +import { moduleLoader as loggerServiceModule } from "@calcom/features/di/shared/services/logger.service"; +import { WebhookTasker } from "@calcom/features/webhooks/lib/tasker/WebhookTasker"; + +import { moduleLoader as webhookSyncTaskerModule } from "./WebhookSyncTasker.module"; +import { moduleLoader as webhookTriggerTaskerModule } from "./WebhookTriggerTasker.module"; +import { WEBHOOK_TASKER_DI_TOKENS } from "./tokens"; + +const thisModule = createModule(); +const token = WEBHOOK_TASKER_DI_TOKENS.WEBHOOK_TASKER; +const moduleToken = WEBHOOK_TASKER_DI_TOKENS.WEBHOOK_TASKER_MODULE; + +const loadModule = bindModuleToClassOnToken({ + module: thisModule, + moduleToken, + token, + classs: WebhookTasker, + depsMap: { + logger: loggerServiceModule, + asyncTasker: webhookTriggerTaskerModule, + syncTasker: webhookSyncTaskerModule, + }, +}); + +export const moduleLoader = { + token, + loadModule, +} satisfies ModuleLoader; + +export type { WebhookTasker }; diff --git a/packages/features/di/webhooks/tasker/WebhookTriggerTasker.module.ts b/packages/features/di/webhooks/tasker/WebhookTriggerTasker.module.ts new file mode 100644 index 0000000000..be914d122f --- /dev/null +++ b/packages/features/di/webhooks/tasker/WebhookTriggerTasker.module.ts @@ -0,0 +1,24 @@ +import { bindModuleToClassOnToken, createModule, type ModuleLoader } from "@calcom/features/di/di"; +import { moduleLoader as loggerServiceModule } from "@calcom/features/di/shared/services/logger.service"; +import { WebhookTriggerTasker } from "@calcom/features/webhooks/lib/tasker/WebhookTriggerTasker"; + +import { WEBHOOK_TASKER_DI_TOKENS } from "./tokens"; + +const thisModule = createModule(); +const token = WEBHOOK_TASKER_DI_TOKENS.WEBHOOK_TRIGGER_TASKER; +const moduleToken = WEBHOOK_TASKER_DI_TOKENS.WEBHOOK_TRIGGER_TASKER_MODULE; + +const loadModule = bindModuleToClassOnToken({ + module: thisModule, + moduleToken, + token, + classs: WebhookTriggerTasker, + depsMap: { + logger: loggerServiceModule, + }, +}); + +export const moduleLoader = { + token, + loadModule, +} satisfies ModuleLoader; diff --git a/packages/features/di/webhooks/tasker/tokens.ts b/packages/features/di/webhooks/tasker/tokens.ts new file mode 100644 index 0000000000..ecd8a63beb --- /dev/null +++ b/packages/features/di/webhooks/tasker/tokens.ts @@ -0,0 +1,8 @@ +export const WEBHOOK_TASKER_DI_TOKENS = { + WEBHOOK_TASKER: Symbol("WebhookTasker"), + WEBHOOK_TASKER_MODULE: Symbol("WebhookTaskerModule"), + WEBHOOK_SYNC_TASKER: Symbol("WebhookSyncTasker"), + WEBHOOK_SYNC_TASKER_MODULE: Symbol("WebhookSyncTaskerModule"), + WEBHOOK_TRIGGER_TASKER: Symbol("WebhookTriggerTasker"), + WEBHOOK_TRIGGER_TASKER_MODULE: Symbol("WebhookTriggerTaskerModule"), +}; diff --git a/packages/features/webhooks/lib/service/WebhookTaskerProducerService.ts b/packages/features/webhooks/lib/service/WebhookTaskerProducerService.ts index a054bf01d6..e93bc3a171 100644 --- a/packages/features/webhooks/lib/service/WebhookTaskerProducerService.ts +++ b/packages/features/webhooks/lib/service/WebhookTaskerProducerService.ts @@ -1,7 +1,7 @@ import { WebhookTriggerEvents } from "@calcom/prisma/enums"; import { v4 as uuidv4 } from "uuid"; import type { BookingTriggerEvents, PaymentTriggerEvents } from "../factory/versioned/PayloadBuilderFactory"; -import type { ILogger, ITasker } from "../interface/infrastructure"; +import type { ILogger } from "../interface/infrastructure"; import type { IWebhookProducerService, QueueBookingWebhookParams, @@ -10,6 +10,7 @@ import type { QueuePaymentWebhookParams, QueueRecordingWebhookParams, } from "../interface/WebhookProducerService"; +import type { WebhookTasker } from "../tasker/WebhookTasker"; import type { WebhookTaskPayload } from "../types/webhookTask"; /** @@ -20,14 +21,19 @@ import type { WebhookTaskPayload } from "../types/webhookTask"; * This service queues minimal webhook tasks to be processed by WebhookTaskConsumer. * The consumer handles the heavy lifting (DB queries, payload building, HTTP delivery). */ +/** + * Dependencies for WebhookTaskerProducerService + */ +export interface IWebhookTaskerProducerServiceDeps { + webhookTasker: WebhookTasker; + logger: ILogger; +} + export class WebhookTaskerProducerService implements IWebhookProducerService { private readonly log: ILogger; - constructor( - private readonly tasker: ITasker, - logger: ILogger - ) { - this.log = logger.getSubLogger({ prefix: ["[WebhookTaskerProducerService]"] }); + constructor(private readonly deps: IWebhookTaskerProducerServiceDeps) { + this.log = deps.logger.getSubLogger({ prefix: ["[WebhookTaskerProducerService]"] }); } async queueBookingCreatedWebhook(params: QueueBookingWebhookParams): Promise { @@ -207,12 +213,16 @@ export class WebhookTaskerProducerService implements IWebhookProducerService { } /** - * Internal helper to queue task via Tasker + * Internal helper to queue task via WebhookTasker + * + * The WebhookTasker automatically selects the appropriate execution mode: + * - Production: Queues to Trigger.dev for background processing + * - E2E Tests: Executes immediately via WebhookSyncTasker */ private async queueTask(operationId: string, taskPayload: WebhookTaskPayload): Promise { try { - await this.tasker.create("webhookDelivery", taskPayload); - this.log.debug("Webhook delivery task queued", { operationId }); + const result = await this.deps.webhookTasker.deliverWebhook(taskPayload); + this.log.debug("Webhook delivery task queued", { operationId, taskId: result.taskId }); } catch (error) { this.log.error("Failed to queue webhook delivery task", { operationId, diff --git a/packages/features/webhooks/lib/service/__tests__/WebhookTaskerProducerService.test.ts b/packages/features/webhooks/lib/service/__tests__/WebhookTaskerProducerService.test.ts index 36f9f23db4..2f429445aa 100644 --- a/packages/features/webhooks/lib/service/__tests__/WebhookTaskerProducerService.test.ts +++ b/packages/features/webhooks/lib/service/__tests__/WebhookTaskerProducerService.test.ts @@ -1,7 +1,8 @@ import { WebhookTriggerEvents } from "@calcom/prisma/enums"; import { beforeEach, describe, expect, it, vi } from "vitest"; -import type { ITasker } from "../../interface"; + import type { ILogger } from "../../interface/infrastructure"; +import type { WebhookTasker } from "../../tasker/WebhookTasker"; import { WebhookTaskerProducerService } from "../WebhookTaskerProducerService"; /** @@ -11,17 +12,14 @@ import { WebhookTaskerProducerService } from "../WebhookTaskerProducerService"; */ describe("WebhookTaskerProducerService", () => { let producer: WebhookTaskerProducerService; - let mockTasker: ITasker; + let mockWebhookTasker: WebhookTasker; let mockLogger: ILogger; beforeEach(() => { - // Mock Tasker - mockTasker = { - create: vi.fn().mockResolvedValue("task-id-123"), - cleanup: vi.fn().mockResolvedValue(undefined), - cancel: vi.fn().mockResolvedValue("cancelled-task-id"), - cancelWithReference: vi.fn().mockResolvedValue("cancelled-ref-id"), - }; + // Mock WebhookTasker + mockWebhookTasker = { + deliverWebhook: vi.fn().mockResolvedValue({ taskId: "task-id-123" }), + } as unknown as WebhookTasker; // Mock Logger mockLogger = { @@ -32,7 +30,10 @@ describe("WebhookTaskerProducerService", () => { getSubLogger: vi.fn().mockReturnThis(), } as unknown as ILogger; - producer = new WebhookTaskerProducerService(mockTasker, mockLogger); + producer = new WebhookTaskerProducerService({ + webhookTasker: mockWebhookTasker, + logger: mockLogger, + }); }); describe("Constructor & Dependencies", () => { @@ -56,8 +57,7 @@ describe("WebhookTaskerProducerService", () => { userId: 789, }); - expect(mockTasker.create).toHaveBeenCalledWith( - "webhookDelivery", + expect(mockWebhookTasker.deliverWebhook).toHaveBeenCalledWith( expect.objectContaining({ triggerEvent: WebhookTriggerEvents.BOOKING_CREATED, bookingUid: "booking-123", @@ -75,8 +75,8 @@ describe("WebhookTaskerProducerService", () => { bookingUid: "booking-123", }); - const callArgs = vi.mocked(mockTasker.create).mock.calls[0]; - const payload = callArgs[1]; + const callArgs = vi.mocked(mockWebhookTasker.deliverWebhook).mock.calls[0]; + const payload = callArgs[0]; expect(payload).toHaveProperty("operationId"); expect(payload.operationId).toMatch(/^[0-9a-f-]{36}$/); // UUID format @@ -89,8 +89,8 @@ describe("WebhookTaskerProducerService", () => { operationId: "custom-op-id", }); - const callArgs = vi.mocked(mockTasker.create).mock.calls[0]; - const payload = callArgs[1]; + const callArgs = vi.mocked(mockWebhookTasker.deliverWebhook).mock.calls[0]; + const payload = callArgs[0]; expect(payload.operationId).toBe("custom-op-id"); }); @@ -113,6 +113,7 @@ describe("WebhookTaskerProducerService", () => { "Webhook delivery task queued", expect.objectContaining({ operationId: expect.any(String), + taskId: "task-id-123", }) ); }); @@ -124,8 +125,7 @@ describe("WebhookTaskerProducerService", () => { bookingUid: "booking-456", }); - expect(mockTasker.create).toHaveBeenCalledWith( - "webhookDelivery", + expect(mockWebhookTasker.deliverWebhook).toHaveBeenCalledWith( expect.objectContaining({ triggerEvent: WebhookTriggerEvents.BOOKING_CANCELLED, bookingUid: "booking-456", @@ -141,28 +141,28 @@ describe("WebhookTaskerProducerService", () => { metadata: { customField: "value" }, }); - const callArgs = vi.mocked(mockTasker.create).mock.calls[0]; - const payload = callArgs[1]; + const callArgs = vi.mocked(mockWebhookTasker.deliverWebhook).mock.calls[0]; + const payload = callArgs[0]; expect(payload.metadata).toEqual({ customField: "value" }); }); }); describe("Error Handling", () => { - it("should log and rethrow error if Tasker fails", async () => { - const error = new Error("Tasker failed"); - vi.mocked(mockTasker.create).mockRejectedValueOnce(error); + it("should log and rethrow error if WebhookTasker fails", async () => { + const error = new Error("WebhookTasker failed"); + vi.mocked(mockWebhookTasker.deliverWebhook).mockRejectedValueOnce(error); await expect( producer.queueBookingCreatedWebhook({ bookingUid: "booking-123", }) - ).rejects.toThrow("Tasker failed"); + ).rejects.toThrow("WebhookTasker failed"); expect(mockLogger.error).toHaveBeenCalledWith( "Failed to queue webhook delivery task", expect.objectContaining({ - error: "Tasker failed", + error: "WebhookTasker failed", }) ); }); @@ -173,8 +173,7 @@ describe("WebhookTaskerProducerService", () => { await producer.queueBookingCreatedWebhook({ bookingUid: "test-123", }); - expect(mockTasker.create).toHaveBeenCalledWith( - "webhookDelivery", + expect(mockWebhookTasker.deliverWebhook).toHaveBeenCalledWith( expect.objectContaining({ triggerEvent: WebhookTriggerEvents.BOOKING_CREATED }) ); }); @@ -183,8 +182,7 @@ describe("WebhookTaskerProducerService", () => { await producer.queueBookingCancelledWebhook({ bookingUid: "test-123", }); - expect(mockTasker.create).toHaveBeenCalledWith( - "webhookDelivery", + expect(mockWebhookTasker.deliverWebhook).toHaveBeenCalledWith( expect.objectContaining({ triggerEvent: WebhookTriggerEvents.BOOKING_CANCELLED }) ); }); @@ -193,8 +191,7 @@ describe("WebhookTaskerProducerService", () => { await producer.queueBookingRescheduledWebhook({ bookingUid: "test-123", }); - expect(mockTasker.create).toHaveBeenCalledWith( - "webhookDelivery", + expect(mockWebhookTasker.deliverWebhook).toHaveBeenCalledWith( expect.objectContaining({ triggerEvent: WebhookTriggerEvents.BOOKING_RESCHEDULED }) ); }); @@ -203,8 +200,7 @@ describe("WebhookTaskerProducerService", () => { await producer.queueBookingRequestedWebhook({ bookingUid: "test-123", }); - expect(mockTasker.create).toHaveBeenCalledWith( - "webhookDelivery", + expect(mockWebhookTasker.deliverWebhook).toHaveBeenCalledWith( expect.objectContaining({ triggerEvent: WebhookTriggerEvents.BOOKING_REQUESTED }) ); }); @@ -213,8 +209,7 @@ describe("WebhookTaskerProducerService", () => { await producer.queueBookingRejectedWebhook({ bookingUid: "test-123", }); - expect(mockTasker.create).toHaveBeenCalledWith( - "webhookDelivery", + expect(mockWebhookTasker.deliverWebhook).toHaveBeenCalledWith( expect.objectContaining({ triggerEvent: WebhookTriggerEvents.BOOKING_REJECTED }) ); }); @@ -223,8 +218,7 @@ describe("WebhookTaskerProducerService", () => { await producer.queueBookingPaymentInitiatedWebhook({ bookingUid: "test-123", }); - expect(mockTasker.create).toHaveBeenCalledWith( - "webhookDelivery", + expect(mockWebhookTasker.deliverWebhook).toHaveBeenCalledWith( expect.objectContaining({ triggerEvent: WebhookTriggerEvents.BOOKING_PAYMENT_INITIATED }) ); }); @@ -233,8 +227,7 @@ describe("WebhookTaskerProducerService", () => { await producer.queueBookingPaidWebhook({ bookingUid: "test-123", }); - expect(mockTasker.create).toHaveBeenCalledWith( - "webhookDelivery", + expect(mockWebhookTasker.deliverWebhook).toHaveBeenCalledWith( expect.objectContaining({ triggerEvent: WebhookTriggerEvents.BOOKING_PAID }) ); }); @@ -243,8 +236,7 @@ describe("WebhookTaskerProducerService", () => { await producer.queueBookingNoShowUpdatedWebhook({ bookingUid: "test-123", }); - expect(mockTasker.create).toHaveBeenCalledWith( - "webhookDelivery", + expect(mockWebhookTasker.deliverWebhook).toHaveBeenCalledWith( expect.objectContaining({ triggerEvent: WebhookTriggerEvents.BOOKING_NO_SHOW_UPDATED }) ); }); @@ -253,8 +245,7 @@ describe("WebhookTaskerProducerService", () => { await producer.queueFormSubmittedWebhook({ formId: "form-123", }); - expect(mockTasker.create).toHaveBeenCalledWith( - "webhookDelivery", + expect(mockWebhookTasker.deliverWebhook).toHaveBeenCalledWith( expect.objectContaining({ triggerEvent: WebhookTriggerEvents.FORM_SUBMITTED }) ); }); @@ -264,8 +255,7 @@ describe("WebhookTaskerProducerService", () => { recordingId: "rec-123", bookingUid: "booking-123", }); - expect(mockTasker.create).toHaveBeenCalledWith( - "webhookDelivery", + expect(mockWebhookTasker.deliverWebhook).toHaveBeenCalledWith( expect.objectContaining({ triggerEvent: WebhookTriggerEvents.RECORDING_READY }) ); }); @@ -275,8 +265,7 @@ describe("WebhookTaskerProducerService", () => { oooEntryId: 123, userId: 456, }); - expect(mockTasker.create).toHaveBeenCalledWith( - "webhookDelivery", + expect(mockWebhookTasker.deliverWebhook).toHaveBeenCalledWith( expect.objectContaining({ triggerEvent: WebhookTriggerEvents.OOO_CREATED }) ); }); diff --git a/packages/features/webhooks/lib/tasker/WebhookSyncTasker.ts b/packages/features/webhooks/lib/tasker/WebhookSyncTasker.ts new file mode 100644 index 0000000000..9cb0445a5e --- /dev/null +++ b/packages/features/webhooks/lib/tasker/WebhookSyncTasker.ts @@ -0,0 +1,31 @@ +import { nanoid } from "nanoid"; + +import type { WebhookTaskConsumer } from "../service/WebhookTaskConsumer"; +import type { WebhookTaskPayload } from "../types/webhookTask"; +import type { IWebhookTasker, WebhookDeliveryResult } from "./types"; + +/** + * Dependencies for WebhookSyncTasker + */ +export interface IWebhookSyncTaskerDeps { + webhookTaskConsumer: WebhookTaskConsumer; +} + +/** + * Synchronous Webhook Tasker + * + * Executes webhook delivery immediately without queuing. + * Used in E2E tests and development environments where the async tasker + * (InternalTasker + cron) is not available. + * + * This follows the same pattern as MonthlyProrationSyncTasker. + */ +export class WebhookSyncTasker implements IWebhookTasker { + constructor(private readonly deps: IWebhookSyncTaskerDeps) {} + + async deliverWebhook(payload: WebhookTaskPayload): Promise { + const taskId = `sync_${nanoid(10)}`; + await this.deps.webhookTaskConsumer.processWebhookTask(payload, taskId); + return { taskId }; + } +} diff --git a/packages/features/webhooks/lib/tasker/WebhookTasker.test.ts b/packages/features/webhooks/lib/tasker/WebhookTasker.test.ts new file mode 100644 index 0000000000..04b39ce775 --- /dev/null +++ b/packages/features/webhooks/lib/tasker/WebhookTasker.test.ts @@ -0,0 +1,99 @@ +import { describe, expect, it, vi, beforeEach } from "vitest"; + +import type { WebhookTaskConsumer } from "../service/WebhookTaskConsumer"; +import type { WebhookTaskPayload } from "../types/webhookTask"; +import { WebhookSyncTasker } from "./WebhookSyncTasker"; +import { WebhookTriggerTasker } from "./WebhookTriggerTasker"; + +vi.mock("nanoid", () => ({ + nanoid: vi.fn().mockReturnValue("test123456"), +})); + +vi.mock("./trigger/deliver-webhook", () => ({ + deliverWebhook: { + trigger: vi.fn().mockResolvedValue({ id: "trigger-task-id-123" }), + }, +})); + +const createMockWebhookTaskPayload = (): WebhookTaskPayload => ({ + operationId: "test-operation-id", + triggerEvent: "BOOKING_CREATED", + bookingUid: "test-booking-uid", + eventTypeId: 1, + teamId: null, + userId: 1, + timestamp: new Date().toISOString(), +}); + +describe("WebhookSyncTasker", () => { + let mockWebhookTaskConsumer: WebhookTaskConsumer; + let syncTasker: WebhookSyncTasker; + + beforeEach(() => { + vi.clearAllMocks(); + mockWebhookTaskConsumer = { + processWebhookTask: vi.fn().mockResolvedValue(undefined), + } as unknown as WebhookTaskConsumer; + + syncTasker = new WebhookSyncTasker({ + webhookTaskConsumer: mockWebhookTaskConsumer, + }); + }); + + it("should execute webhook delivery immediately via consumer", async () => { + const payload = createMockWebhookTaskPayload(); + + const result = await syncTasker.deliverWebhook(payload); + + expect(mockWebhookTaskConsumer.processWebhookTask).toHaveBeenCalledTimes(1); + expect(mockWebhookTaskConsumer.processWebhookTask).toHaveBeenCalledWith( + payload, + expect.stringMatching(/^sync_/) + ); + expect(result.taskId).toMatch(/^sync_/); + }); + + it("should generate unique task IDs for each delivery", async () => { + const payload = createMockWebhookTaskPayload(); + + const result1 = await syncTasker.deliverWebhook(payload); + const result2 = await syncTasker.deliverWebhook(payload); + + expect(result1.taskId).toBe("sync_test123456"); + expect(result2.taskId).toBe("sync_test123456"); + expect(mockWebhookTaskConsumer.processWebhookTask).toHaveBeenCalledTimes(2); + }); + + it("should propagate errors from consumer", async () => { + const payload = createMockWebhookTaskPayload(); + const error = new Error("Consumer processing failed"); + vi.mocked(mockWebhookTaskConsumer.processWebhookTask).mockRejectedValueOnce(error); + + await expect(syncTasker.deliverWebhook(payload)).rejects.toThrow("Consumer processing failed"); + }); +}); + +describe("WebhookTriggerTasker", () => { + let triggerTasker: WebhookTriggerTasker; + let mockLogger: { info: ReturnType; error: ReturnType }; + + beforeEach(() => { + vi.clearAllMocks(); + mockLogger = { + info: vi.fn(), + error: vi.fn(), + }; + + triggerTasker = new WebhookTriggerTasker({ + logger: mockLogger as never, + }); + }); + + it("should trigger webhook delivery via trigger.dev", async () => { + const payload = createMockWebhookTaskPayload(); + + const result = await triggerTasker.deliverWebhook(payload); + + expect(result.taskId).toBe("trigger-task-id-123"); + }); +}); diff --git a/packages/features/webhooks/lib/tasker/WebhookTasker.ts b/packages/features/webhooks/lib/tasker/WebhookTasker.ts new file mode 100644 index 0000000000..b2f97b40e2 --- /dev/null +++ b/packages/features/webhooks/lib/tasker/WebhookTasker.ts @@ -0,0 +1,44 @@ +import { Tasker } from "@calcom/lib/tasker/Tasker"; +import type { ILogger } from "@calcom/lib/tasker/types"; + +import type { WebhookTaskPayload } from "../types/webhookTask"; +import type { WebhookSyncTasker } from "./WebhookSyncTasker"; +import type { WebhookTriggerTasker } from "./WebhookTriggerTasker"; +import type { IWebhookTasker, WebhookDeliveryResult } from "./types"; + +/** + * Dependencies for WebhookTasker + */ +export interface WebhookTaskerDependencies { + asyncTasker: WebhookTriggerTasker; + syncTasker: WebhookSyncTasker; + logger: ILogger; +} + +/** + * Webhook Tasker with Async/Sync Fallback + * + * This tasker automatically selects the appropriate execution mode: + * - Production (ENABLE_ASYNC_TASKER=true): Uses WebhookTriggerTasker to queue tasks via trigger.dev + * - E2E Tests (ENABLE_ASYNC_TASKER=false): Uses WebhookSyncTasker for immediate execution + * + * The base Tasker class handles the mode selection based on environment variables: + * - ENABLE_ASYNC_TASKER (automatically false in E2E tests) + * - TRIGGER_SECRET_KEY + * - TRIGGER_API_URL + * + * This pattern ensures webhooks are delivered immediately in E2E tests + * without requiring trigger.dev or the cron job that processes queued tasks. + * + * This follows the same pattern as BookingEmailAndSmsTasker and + * PlatformOrganizationBillingTasker. + */ +export class WebhookTasker extends Tasker { + constructor(dependencies: WebhookTaskerDependencies) { + super(dependencies); + } + + async deliverWebhook(payload: WebhookTaskPayload): Promise { + return await this.dispatch("deliverWebhook", payload); + } +} diff --git a/packages/features/webhooks/lib/tasker/WebhookTriggerTasker.ts b/packages/features/webhooks/lib/tasker/WebhookTriggerTasker.ts new file mode 100644 index 0000000000..d918e143e4 --- /dev/null +++ b/packages/features/webhooks/lib/tasker/WebhookTriggerTasker.ts @@ -0,0 +1,24 @@ +import type { ITaskerDependencies } from "@calcom/lib/tasker/types"; + +import type { WebhookTaskPayload } from "../types/webhookTask"; +import type { IWebhookTasker, WebhookDeliveryResult } from "./types"; + +/** + * Trigger.dev Webhook Tasker + * + * Queues webhook delivery tasks to trigger.dev for background processing. + * Used in production environments where ENABLE_ASYNC_TASKER is true and + * trigger.dev is configured. + * + * This follows the same pattern as BookingEmailAndSmsTriggerDevTasker and + * PlatformOrganizationBillingTriggerTasker. + */ +export class WebhookTriggerTasker implements IWebhookTasker { + constructor(public readonly dependencies: ITaskerDependencies) {} + + async deliverWebhook(payload: WebhookTaskPayload): Promise { + const { deliverWebhook } = await import("./trigger/deliver-webhook"); + const handle = await deliverWebhook.trigger(payload); + return { taskId: handle.id }; + } +} diff --git a/packages/features/webhooks/lib/tasker/trigger/config.ts b/packages/features/webhooks/lib/tasker/trigger/config.ts new file mode 100644 index 0000000000..ec68123a24 --- /dev/null +++ b/packages/features/webhooks/lib/tasker/trigger/config.ts @@ -0,0 +1,36 @@ +import type { Queue, schemaTask } from "@trigger.dev/sdk"; +import { queue } from "@trigger.dev/sdk"; + +type WebhookDeliveryTask = Pick[0], "machine" | "retry" | "queue">; + +/** + * Queue configuration for webhook delivery tasks + * + * Webhooks are time-sensitive, so we use a moderate concurrency limit + * to ensure timely delivery while not overwhelming external services. + */ +export const webhookDeliveryQueue: Queue = queue({ + name: "webhook-delivery", + concurrencyLimit: 20, +}); + +/** + * Task configuration for webhook delivery + * + * - machine: small-2x for lightweight HTTP requests + * - retry: 3 attempts with exponential backoff for transient failures + */ +export const webhookDeliveryTaskConfig: WebhookDeliveryTask = { + queue: webhookDeliveryQueue, + machine: "small-2x", + retry: { + maxAttempts: 3, + factor: 1.8, + minTimeoutInMs: 500, + maxTimeoutInMs: 30_000, + randomize: true, + outOfMemory: { + machine: "medium-1x", + }, + }, +}; diff --git a/packages/features/webhooks/lib/tasker/trigger/deliver-webhook.ts b/packages/features/webhooks/lib/tasker/trigger/deliver-webhook.ts new file mode 100644 index 0000000000..94909275de --- /dev/null +++ b/packages/features/webhooks/lib/tasker/trigger/deliver-webhook.ts @@ -0,0 +1,48 @@ +import { ErrorWithCode } from "@calcom/lib/errors"; +import { logger, schemaTask, type TaskWithSchema } from "@trigger.dev/sdk"; + +import type { WebhookTaskPayload } from "../../types/webhookTask"; +import { webhookDeliveryTaskConfig } from "./config"; +import { webhookDeliveryTaskSchema } from "./schema"; + +const WEBHOOK_DELIVERY_JOB_ID = "webhook.deliver" as const; + +/** + * Trigger.dev task for webhook delivery + * + * This task is triggered by WebhookTriggerTasker and processes webhook + * delivery using the WebhookTaskConsumer from the DI container. + * + * The task: + * 1. Imports the DI container getter + * 2. Gets the WebhookTaskConsumer instance + * 3. Calls processWebhookTask with the payload + * + * Errors are logged and re-thrown to enable trigger.dev's retry mechanism. + */ +export const deliverWebhook: TaskWithSchema = + schemaTask({ + id: WEBHOOK_DELIVERY_JOB_ID, + ...webhookDeliveryTaskConfig, + schema: webhookDeliveryTaskSchema, + run: async (payload: WebhookTaskPayload, { ctx }) => { + const { getWebhookTaskConsumer } = await import( + "@calcom/features/di/webhooks/containers/webhook" + ); + + const webhookTaskConsumer = getWebhookTaskConsumer(); + const taskId = ctx.run.id; + + try { + await webhookTaskConsumer.processWebhookTask(payload, taskId); + logger.info("Webhook delivered successfully", { operationId: payload.operationId, taskId }); + } catch (error) { + if (error instanceof Error || error instanceof ErrorWithCode) { + logger.error(error.message, { operationId: payload.operationId, taskId }); + } else { + logger.error("Unknown error in webhook delivery", { error, operationId: payload.operationId, taskId }); + } + throw error; + } + }, + }); diff --git a/packages/features/webhooks/lib/tasker/trigger/schema.ts b/packages/features/webhooks/lib/tasker/trigger/schema.ts new file mode 100644 index 0000000000..bf61c7b943 --- /dev/null +++ b/packages/features/webhooks/lib/tasker/trigger/schema.ts @@ -0,0 +1,8 @@ +/** + * Re-export the webhook task payload schema from the types file + * + * This schema is used by the trigger.dev task to validate the payload. + * We re-export from the canonical source to ensure type consistency. + */ +export { webhookTaskPayloadSchema as webhookDeliveryTaskSchema } from "../../types/webhookTask"; +export type { WebhookTaskPayload as WebhookDeliveryTaskPayload } from "../../types/webhookTask"; diff --git a/packages/features/webhooks/lib/tasker/types.ts b/packages/features/webhooks/lib/tasker/types.ts new file mode 100644 index 0000000000..f79a8782d8 --- /dev/null +++ b/packages/features/webhooks/lib/tasker/types.ts @@ -0,0 +1,20 @@ +import type { WebhookTaskPayload } from "../types/webhookTask"; + +/** + * Result of delivering a webhook task + */ +export type WebhookDeliveryResult = { + taskId: string; +}; + +/** + * Interface for webhook taskers (both sync and trigger.dev implementations) + * + * This interface defines the contract for webhook delivery taskers. + * Implementations include: + * - WebhookSyncTasker: Executes immediately (for E2E tests) + * - WebhookTriggerTasker: Queues to trigger.dev (for production) + */ +export interface IWebhookTasker { + deliverWebhook(payload: WebhookTaskPayload): Promise; +}