feat: implement webhook tasker with async/sync fallback (#27378)

* feat: implement webhook tasker with async/sync fallback This PR implements a webhook tasker with async/sync fallback architecture to fix failing E2E tests. The solution follows the existing proration tasker pattern and uses Dependency Injection with @evyweb/ioctopus. Key changes: - Create IWebhookTasker interface for webhook delivery - Implement WebhookSyncTasker for immediate execution (E2E tests) - Implement WebhookAsyncTasker for queued execution (production) - Create main WebhookTasker class extending Tasker<IWebhookTasker> - Add DI modules and tokens for all tasker components - Update WebhookTaskerProducerService to use new WebhookTasker - Add unit tests for sync and async taskers The ENABLE_ASYNC_TASKER flag automatically selects the appropriate mode: - Production: Uses WebhookAsyncTasker to queue tasks - E2E Tests: Uses WebhookSyncTasker for immediate execution This ensures webhooks are delivered immediately in E2E tests without requiring the cron job that processes queued tasks. Co-Authored-By: ali@cal.com <alishahbaz7@gmail.com> * fix: update WebhookTaskerProducerService tests for new interface Update tests to use the new deps-based constructor and mockWebhookTasker.deliverWebhook instead of mockTasker.create Co-Authored-By: ali@cal.com <alishahbaz7@gmail.com> * fix: use moduleLoader pattern for WebhookProducerService in container Co-Authored-By: ali@cal.com <alishahbaz7@gmail.com> * refactor: replace InternalTasker with Trigger.dev for webhook delivery This commit refactors the WebhookTasker to use Trigger.dev instead of InternalTasker, following the pattern established in BookingEmailAndSmsTasker and PlatformOrganizationBillingTasker (PR #26803). Changes: - Replace WebhookAsyncTasker with WebhookTriggerTasker that uses trigger.dev - Create trigger.dev task files (deliver-webhook.ts, config.ts, schema.ts) - Update DI modules to use WebhookTriggerTasker - Remove old InternalTasker-based implementation - Update unit tests for new implementation The WebhookSyncTasker continues to execute webhooks immediately for E2E tests where ENABLE_ASYNC_TASKER is automatically false. Co-Authored-By: ali@cal.com <alishahbaz7@gmail.com> * fix: update comments to reflect Trigger.dev usage instead of InternalTasker Co-Authored-By: unknown <> --------- Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-01-29 15:30:07 +04:00
parent db7547fd6d
commit 2ec9e903ce
18 changed files with 538 additions and 67 deletions
@@ -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"),
@@ -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 };
@@ -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;
@@ -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;
@@ -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 };
@@ -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 };
@@ -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;
@@ -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"),
+};
@@ -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<void> {
@@ -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<void> {
    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,
@@ -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 })
      );
    });
@@ -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<WebhookDeliveryResult> {
+    const taskId = `sync_${nanoid(10)}`;
+    await this.deps.webhookTaskConsumer.processWebhookTask(payload, taskId);
+    return { taskId };
+  }
+}
@@ -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<typeof vi.fn>; error: ReturnType<typeof vi.fn> };
+
+  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");
+  });
+});
@@ -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<IWebhookTasker> {
+  constructor(dependencies: WebhookTaskerDependencies) {
+    super(dependencies);
+  }
+
+  async deliverWebhook(payload: WebhookTaskPayload): Promise<WebhookDeliveryResult> {
+    return await this.dispatch("deliverWebhook", payload);
+  }
+}
@@ -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<WebhookDeliveryResult> {
+    const { deliverWebhook } = await import("./trigger/deliver-webhook");
+    const handle = await deliverWebhook.trigger(payload);
+    return { taskId: handle.id };
+  }
+}
@@ -0,0 +1,36 @@
+import type { Queue, schemaTask } from "@trigger.dev/sdk";
+import { queue } from "@trigger.dev/sdk";
+
+type WebhookDeliveryTask = Pick<Parameters<typeof schemaTask>[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",
+    },
+  },
+};
@@ -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<typeof WEBHOOK_DELIVERY_JOB_ID, typeof webhookDeliveryTaskSchema> =
+  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;
+      }
+    },
+  });
@@ -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";
@@ -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<WebhookDeliveryResult>;
+}