From dc8be81334be602556907984d0006d982156a0e7 Mon Sep 17 00:00:00 2001 From: Hariom Balhara Date: Fri, 28 Feb 2025 19:29:54 +0530 Subject: [PATCH] feat: Task createCRMEvent can retry at cutomizable fixed interval (#19571) --- .../app-store/salesforce/lib/CrmService.ts | 9 +- packages/core/crmManager/errors.ts | 5 + .../core/crmManager/tasker/crmScheduler.ts | 7 +- packages/features/tasker/internal-tasker.ts | 17 +- packages/features/tasker/repository.ts | 19 +- .../crm/__tests__/createCRMEvent.test.ts | 227 +++++++++++++++++- .../tasker/tasks/crm/createCRMEvent.ts | 81 +++++-- packages/features/tasker/tasks/index.ts | 8 + .../migration.sql | 2 + packages/prisma/schema.prisma | 1 + 10 files changed, 351 insertions(+), 25 deletions(-) create mode 100644 packages/core/crmManager/errors.ts create mode 100644 packages/prisma/migrations/20250226133600_add_last_failed_attempt_to_task/migration.sql diff --git a/packages/app-store/salesforce/lib/CrmService.ts b/packages/app-store/salesforce/lib/CrmService.ts index 90210ac2ae..688d8d89dd 100644 --- a/packages/app-store/salesforce/lib/CrmService.ts +++ b/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) => { diff --git a/packages/core/crmManager/errors.ts b/packages/core/crmManager/errors.ts new file mode 100644 index 0000000000..58acefcab2 --- /dev/null +++ b/packages/core/crmManager/errors.ts @@ -0,0 +1,5 @@ +export class RetryableError extends Error { + constructor(message: string) { + super(message); + } +} diff --git a/packages/core/crmManager/tasker/crmScheduler.ts b/packages/core/crmManager/tasker/crmScheduler.ts index c37e23e6af..7bf0ed80d3 100644 --- a/packages/core/crmManager/tasker/crmScheduler.ts +++ b/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 } + ); } } diff --git a/packages/features/tasker/internal-tasker.ts b/packages/features/tasker/internal-tasker.ts index 7cdc5572b5..1eb497d28f 100644 --- a/packages/features/tasker/internal-tasker.ts +++ b/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); diff --git a/packages/features/tasker/repository.ts b/packages/features/tasker/repository.ts index 7f42ab0e2d..673ffc7961 100644 --- a/packages/features/tasker/repository.ts +++ b/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, + }), }, }); } diff --git a/packages/features/tasker/tasks/crm/__tests__/createCRMEvent.test.ts b/packages/features/tasker/tasks/crm/__tests__/createCRMEvent.test.ts index cf8f20d171..d4516f32a7 100644 --- a/packages/features/tasker/tasks/crm/__tests__/createCRMEvent.test.ts +++ b/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: [], + }); + }); }); diff --git a/packages/features/tasker/tasks/crm/createCRMEvent.ts b/packages/features/tasker/tasks/crm/createCRMEvent.ts index 2c3eed966e..48e07b0670 100644 --- a/packages/features/tasker/tasks/crm/createCRMEvent.ts +++ b/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; + 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 { + // 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 { const calendarEvent = await buildCalendarEvent(bookingUid); const bookingReferencesToCreate: Prisma.BookingReferenceUncheckedCreateInput[] = []; + const existingBookingReferences = await prisma.bookingReference.findMany({ + where: { + bookingId: booking.id, + }, + }); + const errorPerApp: Record = {}; // 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 { } 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 { 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 { }); } } 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; } } diff --git a/packages/features/tasker/tasks/index.ts b/packages/features/tasker/tasks/index.ts index 1c9af9fd6d..41c384dc16 100644 --- a/packages/features/tasker/tasks/index.ts +++ b/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 Promise> = { createCRMEvent: () => import("./crm/createCRMEvent").then((module) => module.createCRMEvent), }; +export const tasksConfig = { + createCRMEvent: { + minRetryIntervalMins: IS_PRODUCTION ? 10 : 1, + maxAttempts: 10, + }, +}; export default tasks; diff --git a/packages/prisma/migrations/20250226133600_add_last_failed_attempt_to_task/migration.sql b/packages/prisma/migrations/20250226133600_add_last_failed_attempt_to_task/migration.sql new file mode 100644 index 0000000000..373ac3ea06 --- /dev/null +++ b/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); diff --git a/packages/prisma/schema.prisma b/packages/prisma/schema.prisma index 81c88d8759..d265308e15 100644 --- a/packages/prisma/schema.prisma +++ b/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 {