fix: duplicate workflow reminders caused by scanWorkflowBody (#21767)

* don't scan if there is newer task

* create seperate tasks per workflowStepId

* revert changed constant

* fix comment

* add back missing import

* code clean up

* fix PR feedback

* change function to hasNewerScanTaskForStepId

---------

Co-authored-by: CarinaWolli <wollencarina@gmail.com>
Co-authored-by: Anik Dhabal Babu <81948346+anikdhabal@users.noreply.github.com>

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;
+      }
+    });
+  }
 }

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;
   }
 

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({