fix(api-headless-cms-tasks): race conditions while emptying the entries

packages/api-headless-cms-tasks/__tests__/tasks/emptyTrashBinByModel.test.ts

@@ -39,6 +39,43 @@ const listDeletedEntries = async (context: HcmsTasksContext, modelId: string) =>
 jest.setTimeout(720000);
 
 describe("Empty Trash Bin By Model", () => {
+    it("should fail in case of invalid input - missing `modelId`", async () => {
+        const taskDefinition = createEmptyTrashBinByModelTask();
+        const { handler } = useHandler<HcmsTasksContext>({
+            plugins: [taskDefinition, ...createMockModels()]
+        });
+
+        const context = await handler();
+
+        const task = await context.tasks.createTask({
+            name: "Empty Trash Bin By Model",
+            definitionId: taskDefinition.id,
+            input: {}
+        });
+
+        const runner = createRunner({
+            context,
+            task: taskDefinition
+        });
+
+        const result = await runner({
+            webinyTaskId: task.id
+        });
+
+        expect(result).toBeInstanceOf(ResponseErrorResult);
+
+        expect(result).toMatchObject({
+            status: "error",
+            error: {
+                message: `Missing "modelId" in the input.`
+            },
+            webinyTaskId: task.id,
+            webinyTaskDefinitionId: EntriesTask.EmptyTrashBinByModel,
+            tenant: "root",
+            locale: "en-US"
+        });
+    });
+
     it("should fail in case of not existing model", async () => {
         const taskDefinition = createEmptyTrashBinByModelTask();
         const { handler } = useHandler<HcmsTasksContext>({
@@ -109,7 +146,7 @@ describe("Empty Trash Bin By Model", () => {
 
         expect(result).toMatchObject({
             status: "done",
-            message: "Task done: No entries to delete.",
+            message: `Task done: no entries to delete for the "${MODEL_ID}" model.`,
             webinyTaskId: task.id,
             webinyTaskDefinitionId: EntriesTask.EmptyTrashBinByModel,
             tenant: "root",

packages/api-headless-cms-tasks/src/tasks/entries/useCases/EmptyTrashBinByModel/CreateDeleteEntriesTasks.ts

@@ -1,95 +1,103 @@
 import { ITaskResponseResult } from "@webiny/tasks";
 import { CmsEntryListParams } from "@webiny/api-headless-cms/types";
-import { EntriesTask, IDeleteTrashBinEntriesInput, IEmptyTrashBinByModelTaskParams } from "~/types";
+import { TaskCache } from "./TaskCache";
+import { TaskTrigger } from "./TaskTrigger";
+import { IEmptyTrashBinByModelTaskParams } from "~/types";
 
-const DELETE_ENTRIES_IN_BATCH = 50;
-const DELETE_ENTRIES_WAIT_TIME = 5;
+const BATCH_SIZE = 50;
+const WAITING_TIME = 5;
 
 export class CreateDeleteEntriesTasks {
+    private taskCache = new TaskCache();
+    private taskTrigger = new TaskTrigger(this.taskCache);
+
     public async execute(params: IEmptyTrashBinByModelTaskParams): Promise<ITaskResponseResult> {
         const { input, response, isAborted, isCloseToTimeout, context, store } = params;
 
         try {
+            if (!input.modelId) {
+                return response.error(`Missing "modelId" in the input.`);
+            }
+
             const model = await context.cms.getModel(input.modelId);
 
             if (!model) {
                 return response.error(`Model with ${input.modelId} not found!`);
             }
 
-            const totalCount = input.totalCount || 0;
-            let currentBatch = input.currentBatch || 1;
-            let hasMoreEntries = true;
-
-            while (hasMoreEntries) {
-                const listEntriesParams: CmsEntryListParams = {
-                    where: input.where,
-                    after: input.after,
-                    limit: DELETE_ENTRIES_IN_BATCH
-                };
+            const listEntriesParams: CmsEntryListParams = {
+                where: input.where,
+                after: input.after,
+                limit: BATCH_SIZE
+            };
 
-                const [entries, meta] = await context.cms.listDeletedEntries(
-                    model,
-                    listEntriesParams
-                );
+            let currentBatch = input.currentBatch || 1;
 
+            while (true) {
                 if (isAborted()) {
                     return response.aborted();
                 } else if (isCloseToTimeout()) {
+                    await this.taskTrigger.execute(context, store);
                     return response.continue({
                         ...input,
                         ...listEntriesParams,
                         currentBatch
                     });
                 }
 
+                const [entries, meta] = await context.cms.listDeletedEntries(
+                    model,
+                    listEntriesParams
+                );
+
+                // If no entries exist for the provided query, let's return done.
                 if (meta.totalCount === 0) {
-                    if (totalCount > 0) {
-                        return response.continue(
-                            {
-                                ...input,
-                                ...listEntriesParams,
-                                currentBatch,
-                                processing: true
-                            },
-                            {
-                                seconds: DELETE_ENTRIES_WAIT_TIME
-                            }
-                        );
-                    }
-
-                    return response.done("Task done: No entries to delete.");
+                    return response.done(
+                        `Task done: no entries to delete for the "${input.modelId}" model.`
+                    );
+                }
+
+                // If no entries are returned, let's trigger the cached child tasks and continue in `processing` mode.
+                if (entries.length === 0) {
+                    await this.taskTrigger.execute(context, store);
+                    return response.continue(
+                        {
+                            ...input,
+                            ...listEntriesParams,
+                            currentBatch,
+                            totalCount: meta.totalCount,
+                            processing: true
+                        },
+                        { seconds: WAITING_TIME }
+                    );
                 }
 
-                const entryIds = entries.map(entry => entry.entryId);
+                const entryIds = entries.map(entry => entry.id);
 
                 if (entryIds.length > 0) {
-                    await context.tasks.trigger<IDeleteTrashBinEntriesInput>({
-                        definition: EntriesTask.DeleteTrashBinEntries,
-                        name: `Headless CMS - Delete Entries - ${model.name} - #${currentBatch}`,
-                        parent: store.getTask(),
-                        input: {
-                            modelId: input.modelId,
-                            entryIds
-                        }
-                    });
+                    this.taskCache.cacheTask(input.modelId, entryIds);
+                }
+
+                // No more entries paginated, let's trigger the cached child tasks and continue in `processing` mode.
+                if (!meta.hasMoreItems || !meta.cursor) {
+                    await this.taskTrigger.execute(context, store);
+                    return response.continue(
+                        {
+                            ...input,
+                            ...listEntriesParams,
+                            currentBatch,
+                            totalCount: meta.totalCount,
+                            processing: true
+                        },
+                        { seconds: WAITING_TIME }
+                    );
                 }
 
-                hasMoreEntries = meta.hasMoreItems;
-                input.after = meta.cursor;
-                input.totalCount = meta.totalCount;
+                listEntriesParams.after = meta.cursor;
                 currentBatch++;
             }
-
-            return response.continue(
-                {
-                    ...input,
-                    currentBatch
-                },
-                {
-                    seconds: DELETE_ENTRIES_WAIT_TIME
-                }
-            );
         } catch (ex) {
+            console.error("Error while executing CreateDeleteEntriesTasks:", ex);
             return response.error(ex.message ?? "Error while executing CreateDeleteEntriesTasks");
         }
     }

packages/api-headless-cms-tasks/src/tasks/entries/useCases/EmptyTrashBinByModel/ProcessDeleteEntriesTasks.ts

@@ -37,7 +37,7 @@ export class ProcessDeleteEntriesTasks {
             }
 
             return response.done(
-                `Task done: The trash bin has been emptied for the ${input.modelId} model.`
+                `Task done: trash bin for the "${input.modelId}" model has been emptied.`
             );
         } catch (ex) {
             return response.error(ex.message ?? "Error while executing ProcessDeleteEntriesTasks");

packages/api-headless-cms-tasks/src/tasks/entries/useCases/EmptyTrashBinByModel/TaskCache.ts

@@ -0,0 +1,20 @@
+interface TaskCacheItem {
+    modelId: string;
+    entryIds: string[];
+}
+
+export class TaskCache {
+    private taskCache: TaskCacheItem[] = [];
+
+    cacheTask(modelId: string, entryIds: string[]): void {
+        this.taskCache.push({ modelId, entryIds });
+    }
+
+    getTasks(): TaskCacheItem[] {
+        return this.taskCache;
+    }
+
+    clear(): void {
+        this.taskCache = [];
+    }
+}

packages/api-headless-cms-tasks/src/tasks/entries/useCases/EmptyTrashBinByModel/TaskTrigger.ts

@@ -0,0 +1,38 @@
+import { ITaskManagerStore } from "@webiny/tasks";
+import { TaskCache } from "./TaskCache";
+import {
+    EntriesTask,
+    HcmsTasksContext,
+    IDeleteTrashBinEntriesInput,
+    IEmptyTrashBinByModelInput
+} from "~/types";
+
+export class TaskTrigger {
+    constructor(private taskCache: TaskCache) {}
+
+    async execute(context: HcmsTasksContext, store: ITaskManagerStore<IEmptyTrashBinByModelInput>) {
+        const tasks = this.taskCache.getTasks();
+        if (tasks.length === 0) {
+            return;
+        }
+
+        for (const task of tasks) {
+            try {
+                await context.tasks.trigger<IDeleteTrashBinEntriesInput>({
+                    definition: EntriesTask.DeleteTrashBinEntries,
+                    name: `Headless CMS - Delete Entries - ${task.modelId}`,
+                    parent: store.getTask(),
+                    input: {
+                        modelId: task.modelId,
+                        entryIds: task.entryIds
+                    }
+                });
+            } catch (error) {
+                console.error(`Error triggering task for model ${task.modelId}:`, error);
+            }
+        }
+
+        // Clear the cache after processing
+        this.taskCache.clear();
+    }
+}