feat(core): Implement update of enabled status of tasks

packages/admin-ui/src/lib/core/src/common/generated-types.ts

@@ -3064,6 +3064,7 @@ export type Mutation = {
   updateProvince: Province;
   /** Update an existing Role */
   updateRole: Role;
+  updateScheduledTask: ScheduledTask;
   /** Update an existing Seller */
   updateSeller: Seller;
   /** Update an existing ShippingMethod */
@@ -3924,6 +3925,11 @@ export type MutationUpdateRoleArgs = {
 };
 
 
+export type MutationUpdateScheduledTaskArgs = {
+  input: UpdateScheduledTaskInput;
+};
+
+
 export type MutationUpdateSellerArgs = {
   input: UpdateSellerInput;
 };
@@ -5787,6 +5793,7 @@ export type Sale = Node & StockMovement & {
 
 export type ScheduledTask = {
   __typename?: 'ScheduledTask';
+  description: Scalars['String']['output'];
   id: Scalars['String']['output'];
   isRunning: Scalars['Boolean']['output'];
   lastExecutedAt?: Maybe<Scalars['DateTime']['output']>;
@@ -6762,6 +6769,11 @@ export type UpdateRoleInput = {
   permissions?: InputMaybe<Array<Permission>>;
 };
 
+export type UpdateScheduledTaskInput = {
+  enabled?: InputMaybe<Scalars['Boolean']['input']>;
+  id: Scalars['String']['input'];
+};
+
 export type UpdateSellerInput = {
   customFields?: InputMaybe<Scalars['JSON']['input']>;
   id: Scalars['ID']['input'];

packages/asset-server-plugin/e2e/graphql/generated-e2e-asset-server-plugin-types.ts

@@ -2990,6 +2990,7 @@ export type Mutation = {
     updateProvince: Province;
     /** Update an existing Role */
     updateRole: Role;
+    updateScheduledTask: ScheduledTask;
     /** Update an existing Seller */
     updateSeller: Seller;
     /** Update an existing ShippingMethod */
@@ -3654,6 +3655,10 @@ export type MutationUpdateRoleArgs = {
     input: UpdateRoleInput;
 };
 
+export type MutationUpdateScheduledTaskArgs = {
+    input: UpdateScheduledTaskInput;
+};
+
 export type MutationUpdateSellerArgs = {
     input: UpdateSellerInput;
 };
@@ -5390,6 +5395,7 @@ export type Sale = Node &
     };
 
 export type ScheduledTask = {
+    description: Scalars['String']['output'];
     id: Scalars['String']['output'];
     isRunning: Scalars['Boolean']['output'];
     lastExecutedAt?: Maybe<Scalars['DateTime']['output']>;
@@ -6340,6 +6346,11 @@ export type UpdateRoleInput = {
     permissions?: InputMaybe<Array<Permission>>;
 };
 
+export type UpdateScheduledTaskInput = {
+    enabled?: InputMaybe<Scalars['Boolean']['input']>;
+    id: Scalars['String']['input'];
+};
+
 export type UpdateSellerInput = {
     customFields?: InputMaybe<Scalars['JSON']['input']>;
     id: Scalars['ID']['input'];

packages/common/src/generated-types.ts

@@ -3041,6 +3041,7 @@ export type Mutation = {
   updateProvince: Province;
   /** Update an existing Role */
   updateRole: Role;
+  updateScheduledTask: ScheduledTask;
   /** Update an existing Seller */
   updateSeller: Seller;
   /** Update an existing ShippingMethod */
@@ -3860,6 +3861,11 @@ export type MutationUpdateRoleArgs = {
 };
 
 
+export type MutationUpdateScheduledTaskArgs = {
+  input: UpdateScheduledTaskInput;
+};
+
+
 export type MutationUpdateSellerArgs = {
   input: UpdateSellerInput;
 };
@@ -5709,6 +5715,7 @@ export type Sale = Node & StockMovement & {
 
 export type ScheduledTask = {
   __typename?: 'ScheduledTask';
+  description: Scalars['String']['output'];
   id: Scalars['String']['output'];
   isRunning: Scalars['Boolean']['output'];
   lastExecutedAt?: Maybe<Scalars['DateTime']['output']>;
@@ -6674,6 +6681,11 @@ export type UpdateRoleInput = {
   permissions?: InputMaybe<Array<Permission>>;
 };
 
+export type UpdateScheduledTaskInput = {
+  enabled?: InputMaybe<Scalars['Boolean']['input']>;
+  id: Scalars['String']['input'];
+};
+
 export type UpdateSellerInput = {
   customFields?: InputMaybe<Scalars['JSON']['input']>;
   id: Scalars['ID']['input'];

packages/core/e2e/graphql/generated-e2e-admin-types.ts

@@ -2990,6 +2990,7 @@ export type Mutation = {
     updateProvince: Province;
     /** Update an existing Role */
     updateRole: Role;
+    updateScheduledTask: ScheduledTask;
     /** Update an existing Seller */
     updateSeller: Seller;
     /** Update an existing ShippingMethod */
@@ -3654,6 +3655,10 @@ export type MutationUpdateRoleArgs = {
     input: UpdateRoleInput;
 };
 
+export type MutationUpdateScheduledTaskArgs = {
+    input: UpdateScheduledTaskInput;
+};
+
 export type MutationUpdateSellerArgs = {
     input: UpdateSellerInput;
 };
@@ -5390,6 +5395,7 @@ export type Sale = Node &
     };
 
 export type ScheduledTask = {
+    description: Scalars['String']['output'];
     id: Scalars['String']['output'];
     isRunning: Scalars['Boolean']['output'];
     lastExecutedAt?: Maybe<Scalars['DateTime']['output']>;
@@ -6340,6 +6346,11 @@ export type UpdateRoleInput = {
     permissions?: InputMaybe<Array<Permission>>;
 };
 
+export type UpdateScheduledTaskInput = {
+    enabled?: InputMaybe<Scalars['Boolean']['input']>;
+    id: Scalars['String']['input'];
+};
+
 export type UpdateSellerInput = {
     customFields?: InputMaybe<Scalars['JSON']['input']>;
     id: Scalars['ID']['input'];

packages/core/src/api/resolvers/admin/scheduled-task.resolver.ts

@@ -1,5 +1,5 @@
-import { Query, Resolver } from '@nestjs/graphql';
-import { Permission } from '@vendure/common/lib/generated-types';
+import { Args, Mutation, Query, Resolver } from '@nestjs/graphql';
+import { MutationUpdateScheduledTaskArgs, Permission } from '@vendure/common/lib/generated-types';
 
 import { SchedulerService } from '../../../scheduler/scheduler.service';
 import { Allow } from '../../decorators/allow.decorator';
@@ -13,4 +13,10 @@ export class ScheduledTaskResolver {
     scheduledTasks() {
         return this.schedulerService.getTaskList();
     }
+
+    @Mutation()
+    @Allow(Permission.UpdateSettings, Permission.UpdateSystem)
+    updateScheduledTask(@Args() { input }: MutationUpdateScheduledTaskArgs) {
+        return this.schedulerService.updateTask(input);
+    }
 }

packages/core/src/api/schema/admin-api/scheduled-task.api.graphql

@@ -2,6 +2,15 @@ type Query {
     scheduledTasks: [ScheduledTask!]!
 }
 
+input UpdateScheduledTaskInput {
+    id: String!
+    enabled: Boolean
+}
+
+type Mutation {
+    updateScheduledTask(input: UpdateScheduledTaskInput!): ScheduledTask!
+}
+
 type ScheduledTask {
     id: String!
     description: String!
@@ -11,4 +20,5 @@ type ScheduledTask {
     nextExecutionAt: DateTime
     isRunning: Boolean!
     lastResult: JSON
-}
+    enabled: Boolean!
+} 

packages/core/src/plugin/default-scheduler-plugin/default-scheduler-strategy.ts

@@ -1,7 +1,9 @@
+import { UpdateScheduledTaskInput } from '@vendure/common/lib/generated-types';
 import { Cron } from 'croner';
 import ms from 'ms';
 
 import { Injector } from '../../common';
+import { assertFound } from '../../common/utils';
 import { Logger } from '../../config/logger/vendure-logger';
 import { TransactionalConnection } from '../../connection';
 import { ProcessContext } from '../../process-context';
@@ -11,7 +13,6 @@ import { SchedulerStrategy, TaskReport } from '../../scheduler/scheduler-strateg
 import { DEFAULT_SCHEDULER_PLUGIN_OPTIONS } from './constants';
 import { ScheduledTaskRecord } from './scheduled-task-record.entity';
 import { DefaultSchedulerPluginOptions } from './types';
-
 /**
  * @description
  * The default {@link SchedulerStrategy} implementation that uses the database to
@@ -27,6 +28,7 @@ export class DefaultSchedulerStrategy implements SchedulerStrategy {
     private processContext: ProcessContext;
     private tasks: Map<string, { task: ScheduledTask; isRegistered: boolean }> = new Map();
     private pluginOptions: DefaultSchedulerPluginOptions;
+    private runningTasks: ScheduledTask[] = [];
 
     init(injector: Injector) {
         this.connection = injector.get(TransactionalConnection);
@@ -35,6 +37,15 @@ export class DefaultSchedulerStrategy implements SchedulerStrategy {
         this.injector = injector;
     }
 
+    async destroy() {
+        for (const task of this.runningTasks) {
+            await this.connection.rawConnection
+                .getRepository(ScheduledTaskRecord)
+                .update({ taskId: task.id }, { lockedAt: null });
+            Logger.info(`Released lock for task "${task.id}"`);
+        }
+    }
+
     executeTask(task: ScheduledTask) {
         return async (job: Cron) => {
             if (this.processContext.isServer) {
@@ -48,6 +59,7 @@ export class DefaultSchedulerStrategy implements SchedulerStrategy {
                 .set({ lockedAt: new Date() })
                 .where('taskId = :taskId', { taskId: task.id })
                 .andWhere('lockedAt IS NULL')
+                .andWhere('enabled = TRUE')
                 .execute();
             if (!taskEntity.affected) {
                 return;
@@ -55,6 +67,7 @@ export class DefaultSchedulerStrategy implements SchedulerStrategy {
 
             Logger.verbose(`Executing scheduled task "${task.id}"`);
             try {
+                this.runningTasks.push(task);
                 const timeout = task.options.timeout ?? (this.pluginOptions.defaultTimeout as number);
                 const timeoutMs = typeof timeout === 'number' ? timeout : ms(timeout);
 
@@ -83,6 +96,7 @@ export class DefaultSchedulerStrategy implements SchedulerStrategy {
                     },
                 );
                 Logger.verbose(`Scheduled task "${task.id}" completed successfully`);
+                this.runningTasks = this.runningTasks.filter(t => t !== task);
             } catch (error) {
                 let errorMessage = 'Unknown error';
                 if (error instanceof Error) {
@@ -98,6 +112,7 @@ export class DefaultSchedulerStrategy implements SchedulerStrategy {
                         lastResult: { error: errorMessage } as any,
                     },
                 );
+                this.runningTasks = this.runningTasks.filter(t => t !== task);
             }
         };
     }
@@ -121,6 +136,17 @@ export class DefaultSchedulerStrategy implements SchedulerStrategy {
             .then(task => (task ? this.entityToReport(task) : undefined));
     }
 
+    async updateTask(input: UpdateScheduledTaskInput): Promise<TaskReport> {
+        await this.connection.rawConnection
+            .getRepository(ScheduledTaskRecord)
+            .createQueryBuilder('task')
+            .update()
+            .set({ enabled: input.enabled })
+            .where('taskId = :id', { id: input.id })
+            .execute();
+        return assertFound(this.getTask(input.id));
+    }
+
     private entityToReport(task: ScheduledTaskRecord): TaskReport {
         return {
             id: task.taskId,
@@ -141,6 +167,7 @@ export class DefaultSchedulerStrategy implements SchedulerStrategy {
                 .values({ taskId: task.id })
                 .orIgnore()
                 .execute();
+
             this.tasks.set(task.id, { task, isRegistered: true });
         }
     }

packages/core/src/scheduler/noop-scheduler-strategy.ts

@@ -1,11 +1,25 @@
+import { UpdateScheduledTaskInput } from '@vendure/common/lib/generated-types';
+
 import { Logger } from '../config/logger/vendure-logger';
 
 import { ScheduledTask } from './scheduled-task';
-import { SchedulerStrategy } from './scheduler-strategy';
+import { SchedulerStrategy, TaskReport } from './scheduler-strategy';
 
 export class NoopSchedulerStrategy implements SchedulerStrategy {
+    getTasks(): Promise<TaskReport[]> {
+        return Promise.resolve([]);
+    }
+
+    getTask(id: string): Promise<TaskReport | undefined> {
+        return Promise.resolve(undefined);
+    }
+
     executeTask(task: ScheduledTask) {
         Logger.warn(`No task scheduler is configured! The task ${task.id} will not be executed.`);
         return () => Promise.resolve();
     }
+
+    updateTask(input: UpdateScheduledTaskInput): Promise<TaskReport> {
+        throw new Error(`Not implemented`);
+    }
 }

packages/core/src/scheduler/scheduler-strategy.ts

@@ -1,3 +1,4 @@
+import { UpdateScheduledTaskInput } from '@vendure/common/lib/generated-types';
 import { Cron } from 'croner';
 
 import { InjectableStrategy } from '../common';
@@ -32,4 +33,5 @@ export interface SchedulerStrategy extends InjectableStrategy {
     executeTask(task: ScheduledTask): (job: Cron) => Promise<any> | any;
     getTasks(): Promise<TaskReport[]>;
     getTask(id: string): Promise<TaskReport | undefined>;
+    updateTask(input: UpdateScheduledTaskInput): Promise<TaskReport>;
 }

packages/core/src/scheduler/scheduler.service.ts

@@ -1,4 +1,5 @@
 import { Injectable, OnApplicationBootstrap } from '@nestjs/common';
+import { UpdateScheduledTaskInput } from '@vendure/common/lib/generated-types';
 import CronTime from 'cron-time-generator';
 import { Cron } from 'croner';
 import cronstrue from 'cronstrue';
@@ -8,6 +9,7 @@ import { Logger } from '../config/logger/vendure-logger';
 
 import { NoopSchedulerStrategy } from './noop-scheduler-strategy';
 import { ScheduledTask } from './scheduled-task';
+import { TaskReport } from './scheduler-strategy';
 
 export interface TaskInfo {
     id: string;
@@ -18,6 +20,7 @@ export interface TaskInfo {
     nextExecutionAt: Date | null;
     isRunning: boolean;
     lastResult: any;
+    enabled: boolean;
 }
 
 /**
@@ -45,7 +48,15 @@ export class SchedulerService implements OnApplicationBootstrap {
 
         for (const task of scheduledTasks) {
             const job = this.createCronJob(task);
-            this.jobs.set(task.id, { task, job });
+            const pattern = job.getPattern();
+            if (!pattern) {
+                Logger.warn(`Invalid cron pattern for task ${task.id}`);
+                continue;
+            } else {
+                const schedule = cronstrue.toString(pattern);
+                Logger.info(`Registered scheduled task: ${task.id} - ${schedule}`);
+                this.jobs.set(task.id, { task, job });
+            }
         }
     }
 
@@ -54,28 +65,42 @@ export class SchedulerService implements OnApplicationBootstrap {
      * Returns a list of all the scheduled tasks and their current status.
      */
     getTaskList(): Promise<TaskInfo[]> {
-        return this.configService.schedulerOptions.schedulerStrategy.getTasks().then(taskReports =>
-            taskReports
-                .map(taskReport => {
-                    const job = this.jobs.get(taskReport.id)?.job;
-                    const task = this.jobs.get(taskReport.id)?.task;
-                    if (!job || !task) {
-                        return;
-                    }
-                    const pattern = job.getPattern();
-                    return {
-                        id: taskReport.id,
-                        description: task.options.description ?? '',
-                        schedule: pattern ?? 'unknown',
-                        scheduleDescription: pattern ? cronstrue.toString(pattern) : 'unknown',
-                        lastExecutedAt: taskReport.lastExecutedAt,
-                        nextExecutionAt: job.nextRun(),
-                        isRunning: taskReport.isRunning,
-                        lastResult: taskReport.lastResult,
-                    };
-                })
-                .filter(x => x !== undefined),
-        );
+        return this.configService.schedulerOptions.schedulerStrategy
+            .getTasks()
+            .then(taskReports =>
+                taskReports.map(taskReport => this.createTaskInfo(taskReport)).filter(x => x !== undefined),
+            );
+    }
+
+    updateTask(input: UpdateScheduledTaskInput): Promise<TaskInfo> {
+        return this.configService.schedulerOptions.schedulerStrategy.updateTask(input).then(taskReport => {
+            const taskInfo = this.createTaskInfo(taskReport);
+            if (!taskInfo) {
+                throw new Error(`Task ${input.id} not found`);
+            }
+            return taskInfo;
+        });
+    }
+
+    private createTaskInfo(taskReport: TaskReport): TaskInfo | undefined {
+        const job = this.jobs.get(taskReport.id)?.job;
+        const task = this.jobs.get(taskReport.id)?.task;
+        if (!job || !task) {
+            return;
+        }
+
+        const pattern = job.getPattern();
+        return {
+            id: taskReport.id,
+            description: task.options.description ?? '',
+            schedule: pattern ?? 'unknown',
+            scheduleDescription: pattern ? cronstrue.toString(pattern) : 'unknown',
+            lastExecutedAt: taskReport.lastExecutedAt,
+            nextExecutionAt: job.nextRun(),
+            isRunning: taskReport.isRunning,
+            lastResult: taskReport.lastResult,
+            enabled: taskReport.enabled,
+        };
     }
 
     private createCronJob(task: ScheduledTask) {

packages/elasticsearch-plugin/e2e/graphql/generated-e2e-elasticsearch-plugin-types.ts

@@ -2990,6 +2990,7 @@ export type Mutation = {
     updateProvince: Province;
     /** Update an existing Role */
     updateRole: Role;
+    updateScheduledTask: ScheduledTask;
     /** Update an existing Seller */
     updateSeller: Seller;
     /** Update an existing ShippingMethod */
@@ -3654,6 +3655,10 @@ export type MutationUpdateRoleArgs = {
     input: UpdateRoleInput;
 };
 
+export type MutationUpdateScheduledTaskArgs = {
+    input: UpdateScheduledTaskInput;
+};
+
 export type MutationUpdateSellerArgs = {
     input: UpdateSellerInput;
 };
@@ -5390,6 +5395,7 @@ export type Sale = Node &
     };
 
 export type ScheduledTask = {
+    description: Scalars['String']['output'];
     id: Scalars['String']['output'];
     isRunning: Scalars['Boolean']['output'];
     lastExecutedAt?: Maybe<Scalars['DateTime']['output']>;
@@ -6340,6 +6346,11 @@ export type UpdateRoleInput = {
     permissions?: InputMaybe<Array<Permission>>;
 };
 
+export type UpdateScheduledTaskInput = {
+    enabled?: InputMaybe<Scalars['Boolean']['input']>;
+    id: Scalars['String']['input'];
+};
+
 export type UpdateSellerInput = {
     customFields?: InputMaybe<Scalars['JSON']['input']>;
     id: Scalars['ID']['input'];

packages/payments-plugin/e2e/graphql/generated-admin-types.ts

@@ -3056,6 +3056,7 @@ export type Mutation = {
     updateProvince: Province;
     /** Update an existing Role */
     updateRole: Role;
+    updateScheduledTask: ScheduledTask;
     /** Update an existing Seller */
     updateSeller: Seller;
     /** Update an existing ShippingMethod */
@@ -3724,6 +3725,10 @@ export type MutationUpdateRoleArgs = {
     input: UpdateRoleInput;
 };
 
+export type MutationUpdateScheduledTaskArgs = {
+    input: UpdateScheduledTaskInput;
+};
+
 export type MutationUpdateSellerArgs = {
     input: UpdateSellerInput;
 };
@@ -5465,6 +5470,7 @@ export type Sale = Node &
     };
 
 export type ScheduledTask = {
+    description: Scalars['String']['output'];
     id: Scalars['String']['output'];
     isRunning: Scalars['Boolean']['output'];
     lastExecutedAt?: Maybe<Scalars['DateTime']['output']>;
@@ -6415,6 +6421,11 @@ export type UpdateRoleInput = {
     permissions?: InputMaybe<Array<Permission>>;
 };
 
+export type UpdateScheduledTaskInput = {
+    enabled?: InputMaybe<Scalars['Boolean']['input']>;
+    id: Scalars['String']['input'];
+};
+
 export type UpdateSellerInput = {
     customFields?: InputMaybe<Scalars['JSON']['input']>;
     id: Scalars['ID']['input'];