refactor: Remove long running task plugin, implement coderabbit suggestions

packages/core/src/service/services/task.service.ts

@@ -1,4 +1,5 @@
 import { Injectable } from '@nestjs/common';
+import CronTime from 'cron-time-generator';
 import { Cron } from 'croner';
 
 import { Instrument } from '../../common/instrument-decorator';
@@ -56,17 +57,15 @@ export class TaskService {
             .getMany();
     }
 
-    private async taskConfigExists(taskId: string): Promise<boolean> {
-        const taskConfig = await this.connection.rawConnection
-            .getRepository(ScheduledTaskRecord)
-            .findOne({ where: { taskId } });
-        return !!taskConfig;
-    }
-
-    private getScheduleIntervalMs(schedule: string): number {
-        const cron = new Cron(schedule);
-        const next1 = cron.nextRun();
-        const next2 = cron.nextRun();
+    private getScheduleIntervalMs(schedule: string | ((cronTime: typeof CronTime) => string)): number {
+        const scheduleString = typeof schedule === 'function' ? schedule(CronTime) : schedule;
+        const cron = new Cron(scheduleString);
+        const nextFn: () => Date | null | undefined =
+            typeof (cron as any).nextRun === 'function'
+                ? (cron as any).nextRun.bind(cron)
+                : (cron as any).next.bind(cron);
+        const next1 = nextFn();
+        const next2 = nextFn();
         if (!next1 || !next2) {
             throw new Error('Could not compute next run times');
         }
@@ -86,14 +85,16 @@ export class TaskService {
     ): Promise<ScheduledTaskRecord[]> {
         const staleTasks: ScheduledTaskRecord[] = [];
         const schedulerTasks = await this.schedulerService.getTaskList();
+        const taskInfoById = new Map<string, { id: string; schedule: string }>(
+            schedulerTasks.map((t: { id: string; schedule: string }) => [t.id, t]),
+        );
 
         for (const task of lockedTasks) {
-            const exists = await this.taskConfigExists(task.taskId);
-            if (!exists || !task.lockedAt) {
+            if (!task.lockedAt) {
                 continue;
             }
 
-            const taskInfo = schedulerTasks.find((t: { id: string }) => t.id === task.taskId);
+            const taskInfo = taskInfoById.get(task.taskId);
             if (!taskInfo) {
                 Logger.verbose(`Task ${task.taskId} not found in scheduler service`, loggerCtx);
                 continue;
@@ -117,12 +118,22 @@ export class TaskService {
     }
 
     private async clearLocks(staleTasks: ScheduledTaskRecord[]): Promise<void> {
-        for (const task of staleTasks) {
-            await this.connection.rawConnection
-                .getRepository(ScheduledTaskRecord)
-                .update({ taskId: task.taskId }, { lockedAt: null });
-
-            Logger.verbose(`Successfully cleaned stale task locks for task "${task.taskId}"`);
-        }
+        await this.connection.withTransaction(async ctx => {
+            const repo = this.connection.getRepository(ctx, ScheduledTaskRecord);
+            for (const task of staleTasks) {
+                const result = await repo.update(
+                    { taskId: task.taskId, lockedAt: task.lockedAt ?? undefined },
+                    { lockedAt: null },
+                );
+                if (result.affected && result.affected > 0) {
+                    Logger.verbose(`Successfully cleaned stale task locks for task "${task.taskId}"`);
+                } else {
+                    Logger.debug(
+                        `Skipped clearing lock for task "${task.taskId}" because the lock has changed since observation`,
+                        loggerCtx,
+                    );
+                }
+            }
+        });
     }
 }

packages/dev-server/example-plugins/long-running-task-plugin/README.md

@@ -1,5 +0,0 @@
-# Assembly Stock Sync Plugin
-
-## Overview
-
-This plugin is a temporary solution to get the `stockOnHand` values for 4 part pick products via boomi. This scheduled task runs every 30 mins and starts a job that updates the value at the default stock location. It will be removed later when Aptean adapts Kea or the data sync import.

packages/dev-server/example-plugins/long-running-task-plugin/index.ts

@@ -1,2 +0,0 @@
-export * from './src/long-running-task';
-

packages/dev-server/example-plugins/long-running-task-plugin/src/index.ts

@@ -1 +0,0 @@
-export * from './long-running-task';

packages/dev-server/example-plugins/long-running-task-plugin/src/long-running-task.ts

@@ -1,17 +0,0 @@
-import { PluginCommonModule, VendurePlugin } from '@vendure/core';
-import { longRunningTaskTrigger } from './scheduled-tasks/long-running-task-trigger';
-
-
-@VendurePlugin({
-  imports: [PluginCommonModule],
-  compatibility: '^3.0.0',
-  configuration: (config) => {
-    config.schedulerOptions.tasks.push(longRunningTaskTrigger);
-    return config;
-  },
-})
-export class LongRunningTaskPlugin {
-  static init(): typeof LongRunningTaskPlugin {
-    return LongRunningTaskPlugin;
-  }
-}

packages/dev-server/example-plugins/long-running-task-plugin/src/scheduled-tasks/long-running-task-trigger.ts

@@ -1,28 +0,0 @@
-import { Logger, ScheduledTask } from '@vendure/core';
-
-
-const loggerCtx = `LongRunningTaskTriggerTask`;
-
-/**
- * A scheduled task that triggers the assembly stock sync job
- * to update assembly stock levels from KEA.
- */
-export const longRunningTaskTrigger = new ScheduledTask({
-  id: 'long-running-task-trigger',
-  description: 'Triggers long running task',
-  schedule: (cron) => cron.every(30).minutes(),
-  async execute({ injector }) {
-
-    Logger.verbose('Starting scheduled long running task', loggerCtx);
-
-    try {
-      await new Promise(resolve => setTimeout(resolve, 5 * 60 * 1000));
-      return { ok: true };
-    } catch (error) {
-      if (error instanceof Error) {
-        Logger.error(`Error triggering long running task: ${error.message}`, loggerCtx);
-      }
-      throw error;
-    }
-  },
-});