feat: Task createCRMEvent can retry at cutomizable fixed interval (#19571)

packages/app-store/salesforce/lib/CrmService.ts

@@ -4,6 +4,7 @@ import { RRule } from "rrule";
 import { z } from "zod";
 
 import type { FormResponse } from "@calcom/app-store/routing-forms/types/types";
+import { RetryableError } from "@calcom/core/crmManager/errors";
 import { getLocation } from "@calcom/lib/CalEventParser";
 import { WEBAPP_URL } from "@calcom/lib/constants";
 import { checkIfFreeEmailDomain } from "@calcom/lib/freeEmailDomainCheck/checkIfFreeEmailDomain";
@@ -27,6 +28,12 @@ import {
 } from "./enums";
 import { getSalesforceAppKeys } from "./getSalesforceAppKeys";
 
+class SFObjectToUpdateNotFoundError extends RetryableError {
+  constructor(message: string) {
+    super(message);
+  }
+}
+
 type ExtendedTokenResponse = TokenResponse & {
   instance_url: string;
 };
@@ -204,7 +211,7 @@ export default class SalesforceCRMService implements CRM {
 
     if (!firstContact?.id) {
       log.error("No contacts found for event", { contacts });
-      throw new Error("No contacts found for event");
+      throw new SFObjectToUpdateNotFoundError("No contacts found for event");
     }
 
     const eventWhoIds = contacts.reduce((contactIds, contact) => {

packages/core/crmManager/errors.ts

@@ -0,0 +1,5 @@
+export class RetryableError extends Error {
+  constructor(message: string) {
+    super(message);
+  }
+}

packages/core/crmManager/tasker/crmScheduler.ts

@@ -1,8 +1,13 @@
 import tasker from "@calcom/features/tasker";
+import { tasksConfig } from "@calcom/features/tasker/tasks";
 
 class CRMScheduler {
   static async createEvent({ bookingUid }: { bookingUid: string }) {
-    return tasker.create("createCRMEvent", { bookingUid });
+    return tasker.create(
+      "createCRMEvent",
+      { bookingUid },
+      { maxAttempts: tasksConfig.createCRMEvent.maxAttempts ?? 5 }
+    );
   }
 }
 

packages/features/tasker/internal-tasker.ts

@@ -1,6 +1,6 @@
 import { Task } from "./repository";
 import { type TaskerCreate, type Tasker } from "./tasker";
-import tasksMap from "./tasks";
+import tasksMap, { tasksConfig } from "./tasks";
 
 /**
  * This is the default internal Tasker that uses the Task repository to create tasks.
@@ -18,18 +18,25 @@ export class InternalTasker implements Tasker {
     console.info(`Processing ${tasks.length} tasks`, tasks);
 
     const tasksPromises = tasks.map(async (task) => {
-      console.info(`Processing task ${task.id}`, task);
+      console.info(
+        `Processing task ${task.id}, attempt:${task.attempts} maxAttempts:${task.maxAttempts} lastFailedAttempt:${task.lastFailedAttemptAt}`,
+        task
+      );
       const taskHandlerGetter = tasksMap[task.type as keyof typeof tasksMap];
       if (!taskHandlerGetter) throw new Error(`Task handler not found for type ${task.type}`);
-
+      const taskConfig = tasksConfig[task.type as keyof typeof tasksConfig];
       const taskHandler = await taskHandlerGetter();
       return taskHandler(task.payload)
         .then(async () => {
           await Task.succeed(task.id);
         })
         .catch(async (error) => {
-          console.info(`Error processing task ${task.id}: ${error}`);
-          await Task.retry(task.id, error instanceof Error ? error.message : "Unknown error");
+          console.info(`Retrying task ${task.id}: ${error}`);
+          await Task.retry({
+            taskId: task.id,
+            lastError: error instanceof Error ? error.message : "Unknown error",
+            minRetryIntervalMins: taskConfig?.minRetryIntervalMins ?? null,
+          });
         });
     });
     const settled = await Promise.allSettled(tasksPromises);

packages/features/tasker/repository.ts

@@ -104,7 +104,20 @@ export class Task {
     });
   }
 
-  static async retry(taskId: string, lastError?: string) {
+  static async retry({
+    taskId,
+    lastError,
+    minRetryIntervalMins,
+  }: {
+    taskId: string;
+    lastError?: string;
+    minRetryIntervalMins?: number | null;
+  }) {
+    const failedAttemptTime = new Date();
+    const updatedScheduledAt = minRetryIntervalMins
+      ? new Date(failedAttemptTime.getTime() + 1000 * 60 * minRetryIntervalMins)
+      : undefined;
+
     return db.task.update({
       where: {
         id: taskId,
@@ -112,6 +125,10 @@ export class Task {
       data: {
         attempts: { increment: 1 },
         lastError,
+        lastFailedAttemptAt: failedAttemptTime,
+        ...(updatedScheduledAt && {
+          scheduledAt: updatedScheduledAt,
+        }),
       },
     });
   }

packages/features/tasker/tasks/crm/__tests__/createCRMEvent.test.ts

@@ -2,6 +2,7 @@ import prismaMock from "../../../../../../tests/libs/__mocks__/prismaMock";
 
 import { describe, expect, it, beforeEach, vi } from "vitest";
 
+import { RetryableError } from "@calcom/core/crmManager/errors";
 import { BookingStatus } from "@calcom/prisma/enums";
 
 import { createCRMEvent } from "../createCRMEvent";
@@ -87,6 +88,7 @@ describe("createCRMEvent", () => {
     prismaMock.booking.findUnique.mockReset();
     prismaMock.credential.findUnique.mockReset();
     prismaMock.bookingReference.createMany.mockReset();
+    prismaMock.bookingReference.findMany.mockReset();
   });
 
   it("should successfully create a Salesforce CRM event", async () => {
@@ -130,7 +132,7 @@ describe("createCRMEvent", () => {
     prismaMock.booking.findUnique.mockResolvedValueOnce(mockBooking);
     prismaMock.credential.findUnique.mockResolvedValueOnce(mockCredential);
     prismaMock.bookingReference.createMany.mockResolvedValueOnce({ count: 1 });
-
+    prismaMock.bookingReference.findMany.mockResolvedValueOnce([]);
     const payload = JSON.stringify({
       bookingUid: "booking-123",
     });
@@ -208,7 +210,50 @@ describe("createCRMEvent", () => {
     });
   });
 
-  it("should handle CRM creation error gracefully", async () => {
+  it("should not throw if there is 'Credential not found' error", async () => {
+    const mockBooking: Booking = {
+      id: 1,
+      uid: "booking-123",
+      status: BookingStatus.ACCEPTED,
+      title: "Test Booking",
+      startTime: new Date(),
+      endTime: new Date(),
+      user: {
+        id: 1,
+        email: "test@example.com",
+        name: "Test User",
+      },
+      eventType: {
+        metadata: {
+          apps: {
+            salesforce: {
+              enabled: true,
+              credentialId: 1,
+              appCategories: ["crm"],
+            },
+          },
+        },
+      },
+    };
+
+    prismaMock.booking.findUnique.mockResolvedValue(mockBooking);
+    prismaMock.credential.findUnique.mockResolvedValue(null);
+    prismaMock.bookingReference.findMany.mockResolvedValueOnce([]);
+
+    mockCreateEvent.mockRejectedValue(new Error("Salesforce API error"));
+
+    const payload = JSON.stringify({
+      bookingUid: "booking-123",
+    });
+
+    await createCRMEvent(payload);
+
+    expect(prismaMock.bookingReference.createMany).toHaveBeenCalledWith({
+      data: [],
+    });
+  });
+
+  it("should throw(and thus retry) if there is RetryableError from `createEvent`", async () => {
     const mockBooking: Booking = {
       id: 1,
       uid: "booking-123",
@@ -242,7 +287,55 @@ describe("createCRMEvent", () => {
     };
 
     prismaMock.booking.findUnique.mockResolvedValue(mockBooking);
-    prismaMock.credential.findFirst.mockResolvedValue(mockCredential);
+    prismaMock.credential.findUnique.mockResolvedValue(mockCredential);
+    prismaMock.bookingReference.findMany.mockResolvedValueOnce([]);
+
+    mockCreateEvent.mockRejectedValue(new RetryableError("Salesforce API Retryable error"));
+
+    const payload = JSON.stringify({
+      bookingUid: "booking-123",
+    });
+
+    await expect(createCRMEvent(payload)).rejects.toThrow("Salesforce API Retryable error");
+  });
+
+  it("should not throw(and thus not retry) if there is non-RetryableError from `createEvent`", async () => {
+    const mockBooking: Booking = {
+      id: 1,
+      uid: "booking-123",
+      status: BookingStatus.ACCEPTED,
+      title: "Test Booking",
+      startTime: new Date(),
+      endTime: new Date(),
+      user: {
+        id: 1,
+        email: "test@example.com",
+        name: "Test User",
+      },
+      eventType: {
+        metadata: {
+          apps: {
+            salesforce: {
+              enabled: true,
+              credentialId: 1,
+              appCategories: ["crm"],
+            },
+          },
+        },
+      },
+    };
+
+    const mockCredential: CRMCredential = {
+      id: 1,
+      type: "salesforce_crm",
+      key: { key1: "value1" },
+      userId: 1,
+    };
+
+    prismaMock.booking.findUnique.mockResolvedValue(mockBooking);
+    prismaMock.credential.findUnique.mockResolvedValue(mockCredential);
+    prismaMock.bookingReference.findMany.mockResolvedValueOnce([]);
+
     mockCreateEvent.mockRejectedValue(new Error("Salesforce API error"));
 
     const payload = JSON.stringify({
@@ -255,4 +348,132 @@ describe("createCRMEvent", () => {
       data: [],
     });
   });
+
+  it("should process second app evenif first app throws error in processing", async () => {
+    const mockBooking: Booking = {
+      id: 1,
+      uid: "booking-123",
+      status: BookingStatus.ACCEPTED,
+      title: "Test Booking",
+      startTime: new Date(),
+      endTime: new Date(),
+      user: {
+        id: 1,
+        email: "test@example.com",
+        name: "Test User",
+      },
+      eventType: {
+        metadata: {
+          apps: {
+            salesforce: {
+              enabled: true,
+              credentialId: 1,
+              appCategories: ["crm"],
+            },
+            hubspot: {
+              enabled: true,
+              credentialId: 2,
+              appCategories: ["crm"],
+            },
+          },
+        },
+      },
+    };
+
+    const mockSalesforceCredential: CRMCredential = {
+      id: 1,
+      type: "salesforce_crm",
+      key: { key1: "value1" },
+      userId: 1,
+    };
+
+    const mockHubspotCredential: CRMCredential = {
+      id: 2,
+      type: "hubspot",
+      key: { key1: "value1" },
+      userId: 1,
+    };
+
+    prismaMock.booking.findUnique.mockResolvedValue(mockBooking);
+
+    // First call returns Salesforce credential and second call returns Hubspot credential
+    prismaMock.credential.findUnique
+      .mockResolvedValueOnce(mockSalesforceCredential)
+      .mockResolvedValueOnce(mockHubspotCredential);
+
+    prismaMock.bookingReference.findMany.mockResolvedValueOnce([]);
+
+    // Throw error for first app and resolve for second app
+    mockCreateEvent
+      .mockRejectedValueOnce(new Error("Salesforce API error"))
+      .mockResolvedValueOnce({ id: "hubspot-event-123" });
+
+    const payload = JSON.stringify({
+      bookingUid: "booking-123",
+    });
+
+    await createCRMEvent(payload);
+
+    expect(prismaMock.bookingReference.createMany).toHaveBeenCalledWith({
+      data: [
+        {
+          type: "hubspot",
+          uid: "hubspot-event-123",
+          meetingId: "hubspot-event-123",
+          credentialId: 2,
+          bookingId: 1,
+        },
+      ],
+    });
+  });
+
+  it("should skip a credential from creatingEvent if bookingReference already exists(Helpful in case of retry)", async () => {
+    const mockBooking: Booking = {
+      id: 1,
+      uid: "booking-123",
+      status: BookingStatus.ACCEPTED,
+      title: "Test Booking",
+      startTime: new Date(),
+      endTime: new Date(),
+      user: {
+        id: 1,
+        email: "test@example.com",
+        name: "Test User",
+      },
+      eventType: {
+        metadata: {
+          apps: {
+            salesforce: {
+              enabled: true,
+              credentialId: 1,
+              appCategories: ["crm"],
+            },
+          },
+        },
+      },
+    };
+
+    const mockSalesforceCredential: CRMCredential = {
+      id: 1,
+      type: "salesforce_crm",
+      key: { key1: "value1" },
+      userId: 1,
+    };
+
+    prismaMock.booking.findUnique.mockResolvedValue(mockBooking);
+    prismaMock.credential.findUnique.mockResolvedValueOnce(mockSalesforceCredential);
+    prismaMock.bookingReference.findMany.mockResolvedValueOnce([
+      { id: 1, type: "salesforce_crm", uid: "sf-event-123", credentialId: 1, bookingId: 1 },
+    ]);
+
+    const payload = JSON.stringify({
+      bookingUid: "booking-123",
+    });
+
+    await createCRMEvent(payload);
+
+    expect(prismaMock.bookingReference.createMany).toHaveBeenCalledWith({
+      data: [],
+    });
+  });
 });

packages/features/tasker/tasks/crm/createCRMEvent.ts

@@ -1,7 +1,9 @@
 import type { Prisma } from "@prisma/client";
 
 import { appDataSchemas } from "@calcom/app-store/apps.schemas.generated";
+import { RetryableError } from "@calcom/core/crmManager/errors";
 import logger from "@calcom/lib/logger";
+import { safeStringify } from "@calcom/lib/safeStringify";
 import prisma from "@calcom/prisma";
 import { BookingStatus } from "@calcom/prisma/enums";
 import { EventTypeMetaDataSchema } from "@calcom/prisma/zod-utils";
@@ -10,8 +12,35 @@ import buildCalendarEvent from "./lib/buildCalendarEvent";
 import { createCRMEventSchema } from "./schema";
 
 const log = logger.getSubLogger({ prefix: [`[[tasker] createCRMEvent`] });
+type AppSlug = string;
+type UnknownError = unknown;
+
+const handleErrors = ({
+  errorPerApp,
+  payload,
+}: {
+  errorPerApp: Record<AppSlug, UnknownError>;
+  payload: string;
+}) => {
+  const errorMsgs = Object.entries(errorPerApp).map(([appSlug, error]) => {
+    if (error instanceof Error) {
+      return `(app: ${appSlug}) ${error.message}`;
+    }
+    return `(app: ${appSlug}) ${error}`;
+  });
+
+  if (errorMsgs.length > 0) {
+    const hasAnyRetryableErrors = Object.values(errorPerApp).some((error) => error instanceof RetryableError);
+    if (hasAnyRetryableErrors) {
+      // Intentional rethrow to trigger retry
+      throw new Error(errorMsgs.join("\n"));
+    }
+    log.error(`[Will not retry] Error creating CRM event for payload ${payload}: ${errorMsgs.join("\n")}`);
+  }
+};
 
 export async function createCRMEvent(payload: string): Promise<void> {
+  // All errors thrown from this try catch will be cause a retry
   try {
     const parsedPayload = createCRMEventSchema.safeParse(JSON.parse(payload));
 
@@ -75,13 +104,20 @@ export async function createCRMEvent(payload: string): Promise<void> {
     const calendarEvent = await buildCalendarEvent(bookingUid);
 
     const bookingReferencesToCreate: Prisma.BookingReferenceUncheckedCreateInput[] = [];
+    const existingBookingReferences = await prisma.bookingReference.findMany({
+      where: {
+        bookingId: booking.id,
+      },
+    });
 
+    const errorPerApp: Record<AppSlug, UnknownError> = {};
     // Find enabled CRM apps for the event type
     for (const appSlug of Object.keys(eventTypeAppMetadata)) {
+      // Try Catch per app to ensure all apps are tried even if any of them throws an error
+      // If we want to retry for an error from this try catch, then that error must be thrown as a RetryableError
       try {
         const appData = eventTypeAppMetadata[appSlug as keyof typeof eventTypeAppMetadata];
         const appDataSchema = appDataSchemas[appSlug as keyof typeof appDataSchemas];
-
         if (!appData || !appDataSchema) {
           throw new Error(`Could not find appData or appDataSchema for ${appSlug}`);
         }
@@ -94,14 +130,17 @@ export async function createCRMEvent(payload: string): Promise<void> {
         }
 
         const app = appParse.data;
+        const hasCrmCategory =
+          app.appCategories && app.appCategories.some((category: string) => category === "crm");
 
-        if (
-          !app.appCategories ||
-          !app.appCategories.some((category: string) => category === "crm") ||
-          !app.enabled ||
-          !app.credentialId
-        )
+        if (!app.enabled || !app.credentialId || !hasCrmCategory) {
+          log.info(`Skipping CRM app ${appSlug}`, {
+            enabled: app.enabled,
+            credentialId: app.credentialId,
+            hasCrmCategory,
+          });
           continue;
+        }
 
         const crmCredential = await prisma.credential.findUnique({
           where: {
@@ -120,13 +159,23 @@ export async function createCRMEvent(payload: string): Promise<void> {
           throw new Error(`Credential not found for credentialId: ${app.credentialId}`);
         }
 
+        const existingBookingReferenceForTheCredential = existingBookingReferences.find(
+          (reference) => reference.credentialId === crmCredential.id
+        );
+
+        if (existingBookingReferenceForTheCredential) {
+          log.info(`Skipping CRM app ${appSlug} as booking reference already exists`, {
+            credentialId: crmCredential.id,
+            bookingReferenceId: existingBookingReferenceForTheCredential.id,
+          });
+          continue;
+        }
+
         const CrmManager = (await import("@calcom/core/crmManager/crmManager")).default;
 
         const crm = new CrmManager(crmCredential, app);
 
-        const results = await crm.createEvent(calendarEvent).catch((error) => {
-          log.error(`Error creating crm event for credentialId ${app.credentialId}`, error);
-        });
+        const results = await crm.createEvent(calendarEvent);
 
         if (results) {
           bookingReferencesToCreate.push({
@@ -138,17 +187,21 @@ export async function createCRMEvent(payload: string): Promise<void> {
           });
         }
       } catch (error) {
-        log.error(`Error processing CRM app ${appSlug} for booking ${bookingUid}:`, error);
-        // Continue with next app even if one fails
-        continue;
+        errorPerApp[appSlug] = error;
       }
     }
 
     await prisma.bookingReference.createMany({
       data: bookingReferencesToCreate,
     });
+
+    handleErrors({ errorPerApp, payload });
   } catch (error) {
-    log.error(`Error in createCRMEvent for payload: ${payload}:`, error);
+    const errorMsg = `Error creating crm event: error: ${safeStringify(error)} Data: ${safeStringify({
+      payload,
+    })}`;
+    log.error(`[Will retry] ${errorMsg}`);
+    // Intentional rethrow to trigger retry
     throw error;
   }
 }

packages/features/tasker/tasks/index.ts

@@ -1,3 +1,5 @@
+import { IS_PRODUCTION } from "@calcom/lib/constants";
+
 import type { TaskHandler, TaskTypes } from "../tasker";
 
 /**
@@ -22,4 +24,10 @@ const tasks: Record<TaskTypes, () => Promise<TaskHandler>> = {
   createCRMEvent: () => import("./crm/createCRMEvent").then((module) => module.createCRMEvent),
 };
 
+export const tasksConfig = {
+  createCRMEvent: {
+    minRetryIntervalMins: IS_PRODUCTION ? 10 : 1,
+    maxAttempts: 10,
+  },
+};
 export default tasks;

packages/prisma/migrations/20250226133600_add_last_failed_attempt_to_task/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "Task" ADD COLUMN     "lastFailedAttemptAt" TIMESTAMP(3);

packages/prisma/schema.prisma

@@ -1622,6 +1622,7 @@ model Task {
   // The maximum number of times the task can be attempted
   maxAttempts Int       @default(3)
   lastError   String?
+  lastFailedAttemptAt DateTime?
 }
 
 enum SMSLockState {