diff --git a/packages/features/tasker/repository.ts b/packages/features/tasker/repository.ts index 121847dfe9..f7e4fd4948 100644 --- a/packages/features/tasker/repository.ts +++ b/packages/features/tasker/repository.ts @@ -3,6 +3,7 @@ import { type Prisma } from "@prisma/client"; import db from "@calcom/prisma"; import { type TaskTypes } from "./tasker"; +import { scanWorkflowBodySchema } from "./tasks/scanWorkflowBody"; const whereSucceeded: Prisma.TaskWhereInput = { succeededAt: { not: null }, @@ -187,4 +188,24 @@ export class Task { // }, // }); } + + static async hasNewerScanTaskForStepId(workflowStepId: number, createdAt: string) { + const tasks = await db.$queryRaw<{ payload: string }[]>` + SELECT "payload" + FROM "Task" + WHERE "type" = 'scanWorkflowBody' + AND "succeededAt" IS NULL + AND (payload::jsonb ->> 'workflowStepId')::int = ${workflowStepId} + `; + + return tasks.some((task) => { + try { + const parsed = scanWorkflowBodySchema.parse(JSON.parse(task.payload)); + if (!parsed.createdAt) return false; + return new Date(parsed.createdAt) > new Date(createdAt); + } catch { + return false; + } + }); + } } diff --git a/packages/features/tasker/tasks/scanWorkflowBody.ts b/packages/features/tasker/tasks/scanWorkflowBody.ts index 334f9f6c52..b63a773aa5 100644 --- a/packages/features/tasker/tasks/scanWorkflowBody.ts +++ b/packages/features/tasker/tasks/scanWorkflowBody.ts @@ -2,6 +2,7 @@ import z from "zod"; import { getTemplateBodyForAction } from "@calcom/features/ee/workflows/lib/actionHelperFunctions"; import compareReminderBodyToTemplate from "@calcom/features/ee/workflows/lib/compareReminderBodyToTemplate"; +import { Task } from "@calcom/features/tasker/repository"; import { lockUser, LockReason } from "@calcom/lib/autoLock"; import logger from "@calcom/lib/logger"; import { getTranslation } from "@calcom/lib/server/i18n"; @@ -11,18 +12,39 @@ import { scheduleWorkflowNotifications } from "@calcom/trpc/server/routers/viewe export const scanWorkflowBodySchema = z.object({ userId: z.number(), - workflowStepIds: z.array(z.number()), + // deprecated: use workflowStepId instead + workflowStepIds: z.array(z.number()).optional(), + workflowStepId: z.number().optional(), + createdAt: z.string().optional(), }); const log = logger.getSubLogger({ prefix: ["[tasker] scanWorkflowBody"] }); export async function scanWorkflowBody(payload: string) { - const { workflowStepIds, userId } = scanWorkflowBodySchema.parse(JSON.parse(payload)); + const { workflowStepIds, userId, createdAt, workflowStepId } = scanWorkflowBodySchema.parse( + JSON.parse(payload) + ); + + const stepIdsToScan: number[] = workflowStepIds ? workflowStepIds : []; + + if (workflowStepId && !workflowStepIds) { + if (createdAt) { + const hasNewerTask = await Task.hasNewerScanTaskForStepId(workflowStepId, createdAt); + + if (hasNewerTask) return; + + stepIdsToScan.push(workflowStepId); + } else { + stepIdsToScan.push(workflowStepId); + } + } + + if (stepIdsToScan.length === 0) return; const workflowSteps = await prisma.workflowStep.findMany({ where: { id: { - in: workflowStepIds, + in: stepIdsToScan, }, }, include: { @@ -116,7 +138,7 @@ export async function scanWorkflowBody(payload: string) { await prisma.workflowStep.updateMany({ where: { id: { - in: workflowStepIds, + in: stepIdsToScan, }, }, data: { @@ -130,7 +152,7 @@ export async function scanWorkflowBody(payload: string) { steps: { some: { id: { - in: workflowStepIds, + in: stepIdsToScan, }, }, }, @@ -142,7 +164,7 @@ export async function scanWorkflowBody(payload: string) { }); if (!workflow) { - log.warn(`Workflow with steps ${workflowStepIds} not found`); + log.warn(`Workflow with steps ${stepIdsToScan} not found`); return; } diff --git a/packages/trpc/server/routers/viewer/workflows/update.handler.ts b/packages/trpc/server/routers/viewer/workflows/update.handler.ts index a186888d58..518c536eb2 100755 --- a/packages/trpc/server/routers/viewer/workflows/update.handler.ts +++ b/packages/trpc/server/routers/viewer/workflows/update.handler.ts @@ -392,7 +392,11 @@ export const updateHandler = async ({ ctx, input }: UpdateOptions) => { }); if (SCANNING_WORKFLOW_STEPS && didBodyChange) { - await tasker.create("scanWorkflowBody", { workflowStepIds: [oldStep.id], userId: ctx.user.id }); + await tasker.create("scanWorkflowBody", { + workflowStepId: oldStep.id, + userId: ctx.user.id, + createdAt: new Date().toISOString(), + }); } else { // schedule notifications for edited steps await scheduleWorkflowNotifications({ @@ -467,11 +471,15 @@ export const updateHandler = async ({ ctx, input }: UpdateOptions) => { ); if (SCANNING_WORKFLOW_STEPS) { - // workflows are scanned then scheduled in the task - await tasker.create("scanWorkflowBody", { - workflowStepIds: createdSteps.map((step) => step.id), - userId: ctx.user.id, - }); + await Promise.all( + createdSteps.map((step) => + tasker.create("scanWorkflowBody", { + workflowStepId: step.id, + userId: ctx.user.id, + createdAt: new Date().toISOString(), + }) + ) + ); } else { // schedule notification for new step await scheduleWorkflowNotifications({