feat(api): add canceling to processing jobs

Vizzuality · Aug 10, 2021 · efbfc8c · efbfc8c
1 parent 8bc1c32
commit efbfc8c
Show file tree

Hide file tree

Showing 9 changed files with 195 additions and 49 deletions.
diff --git a/api/apps/api/src/modules/api-events/dto/scenario-geofeature-data-v1-alpha.ts b/api/apps/api/src/modules/api-events/dto/scenario-geofeature-data-v1-alpha.ts
@@ -1,10 +1,7 @@
 import { ApiProperty } from '@nestjs/swagger';
 import { IsIn, IsUUID } from 'class-validator';
-import { API_EVENT_KINDS } from '@marxan/api-events';
-import {
-  ScenarioGeofeatureDataV1Alpha,
-  ScenarioGeofeatureEvents,
-} from '../events-data/scenario-geofeature-data-v1-alpha';
+import { API_EVENT_KINDS, ScenarioGeofeatureEvents } from '@marxan/api-events';
+import { ScenarioGeofeatureDataV1Alpha } from '../events-data/scenario-geofeature-data-v1-alpha';
 import { ValuesType } from 'utility-types';
 
 // it's guarded by typing, no mismatch possible

diff --git a/api/apps/api/src/modules/api-events/events-data/scenario-geofeature-data-v1-alpha.ts b/api/apps/api/src/modules/api-events/events-data/scenario-geofeature-data-v1-alpha.ts
@@ -1,14 +1,6 @@
-import { API_EVENT_KINDS } from '@marxan/api-events';
+import { ScenarioGeofeatureEvents } from '@marxan/api-events';
 import { ValuesType } from 'utility-types';
 
-export type ScenarioGeofeatureEvents = Pick<
-  typeof API_EVENT_KINDS,
-  Extract<
-    keyof typeof API_EVENT_KINDS,
-    `scenario__geofeature${`Copy` | `Split` | `Stratification`}${string}`
-  >
->;
-
 export class ScenarioGeofeatureDataV1Alpha {
   kind!: ValuesType<ScenarioGeofeatureEvents>;
   featureId!: string;

diff --git a/api/apps/api/src/modules/geo-features/processing/run.service.ts b/api/apps/api/src/modules/geo-features/processing/run.service.ts
@@ -1,17 +1,26 @@
 import { Inject } from '@nestjs/common';
-import { Queue } from 'bullmq';
+import { Job, Queue } from 'bullmq';
+import { left, right, Either } from 'fp-ts/Either';
 import {
   CopyJobData,
+  FeaturesJobData,
+  FeaturesJobProgress,
   SplitJobData,
   StratificationJobData,
 } from '@marxan/geofeature-calculations';
-import { API_EVENT_KINDS } from '@marxan/api-events';
+import {
+  API_EVENT_KINDS,
+  ScenarioGeofeatureEventValues,
+} from '@marxan/api-events';
+import { isDefined } from '@marxan/utils';
 import { ApiEventsService } from '@marxan-api/modules/api-events/api-events.service';
 import {
   copyQueueToken,
   splitQueueToken,
   stratificationQueueToken,
 } from './queue-providers';
+export const notFound = Symbol('not found');
+export type NotFound = typeof notFound;
 
 export class RunService {
   constructor(
@@ -25,39 +34,49 @@ export class RunService {
   ) {}
 
   async runCopy(data: CopyJobData) {
-    const job = await this.copyQueue.add(`run`, data);
-    const kind =
-      API_EVENT_KINDS.scenario__geofeatureCopy__submitted__v1__alpha1;
-    await this.apiEvents.create({
-      externalId: job.id + kind,
-      kind,
-      topic: data.scenarioId,
-      data: {
-        kind,
-        featureId: data.featureId,
-      },
-    });
+    await this.run(
+      this.copyQueue,
+      data,
+      API_EVENT_KINDS.scenario__geofeatureCopy__submitted__v1__alpha1,
+    );
+  }
+
+  async cancelCopy(data: FeaturesJobData): Promise<Either<NotFound, void>> {
+    return await this.cancel(this.copyQueue, data);
   }
 
   async runSplit(data: SplitJobData) {
-    const job = await this.splitQueue.add(`run`, data);
-    const kind =
-      API_EVENT_KINDS.scenario__geofeatureSplit__submitted__v1__alpha1;
-    await this.apiEvents.create({
-      externalId: job.id + kind,
-      kind,
-      topic: data.scenarioId,
-      data: {
-        kind,
-        featureId: data.featureId,
-      },
-    });
+    await this.run(
+      this.splitQueue,
+      data,
+      API_EVENT_KINDS.scenario__geofeatureSplit__submitted__v1__alpha1,
+    );
+  }
+
+  async cancelSplit(data: FeaturesJobData): Promise<Either<NotFound, void>> {
+    return await this.cancel(this.splitQueue, data);
   }
 
   async runStratification(data: StratificationJobData) {
-    const job = await this.stratificationQueue.add(`run`, data);
-    const kind =
-      API_EVENT_KINDS.scenario__geofeatureStratification__submitted__v1__alpha1;
+    await this.run(
+      this.stratificationQueue,
+      data,
+      API_EVENT_KINDS.scenario__geofeatureStratification__submitted__v1__alpha1,
+    );
+  }
+
+  async cancelStratification(
+    data: FeaturesJobData,
+  ): Promise<Either<NotFound, void>> {
+    return await this.cancel(this.stratificationQueue, data);
+  }
+
+  private async run(
+    queue: Queue<FeaturesJobData>,
+    data: FeaturesJobData,
+    kind: ScenarioGeofeatureEventValues,
+  ) {
+    const job = await queue.add(`run`, data);
     await this.apiEvents.create({
       externalId: job.id + kind,
       kind,
@@ -68,4 +87,33 @@ export class RunService {
       },
     });
   }
+
+  private async cancel(
+    queue: Queue<FeaturesJobData>,
+    data: FeaturesJobData,
+  ): Promise<Either<NotFound, void>> {
+    const activeJobs: Job<FeaturesJobData>[] = await queue.getJobs([
+      'active',
+      'waiting',
+    ]);
+    const job = activeJobs.find(
+      (job) =>
+        job.data.featureId === data.featureId &&
+        job.data.scenarioId === data.scenarioId,
+    );
+    if (!isDefined(job)) return left(notFound);
+
+    if (await job.isWaiting()) await job.remove();
+    if (await job.isActive()) {
+      const cancellingProgress: FeaturesJobProgress = {
+        type: 'canceled',
+        canceled: true,
+        featureId: data.featureId,
+        scenarioId: data.scenarioId,
+      };
+      await job.updateProgress(cancellingProgress);
+    }
+
+    return right(void 0);
+  }
 }
diff --git a/api/apps/geoprocessing/src/modules/features/features.module.ts b/api/apps/geoprocessing/src/modules/features/features.module.ts
@@ -9,10 +9,13 @@ import { WorkerModule } from '@marxan-geoprocessing/modules/worker';
 import {
   copyQueueNameProvider,
   copyWorkerBuilderProvider,
+  copyQueueEventsProvider,
   splitQueueNameProvider,
   splitWorkerBuilderProvider,
+  splitQueueEventsProvider,
   stratificationQueueNameProvider,
   stratificationWorkerBuilderProvider,
+  stratificationQueueEventsProvider,
 } from './processing/worker-builder.providers';
 import { ProcessingWorker } from './processing/processing.worker';
 
@@ -26,10 +29,13 @@ import { ProcessingWorker } from './processing/processing.worker';
     FeatureService,
     copyQueueNameProvider,
     copyWorkerBuilderProvider,
+    copyQueueEventsProvider,
     splitQueueNameProvider,
     splitWorkerBuilderProvider,
+    splitQueueEventsProvider,
     stratificationQueueNameProvider,
     stratificationWorkerBuilderProvider,
+    stratificationQueueEventsProvider,
     ProcessingWorker,
   ],
   controllers: [FeaturesController],

diff --git a/api/apps/geoprocessing/src/modules/features/processing/processing.worker.ts b/api/apps/geoprocessing/src/modules/features/processing/processing.worker.ts
@@ -1,16 +1,21 @@
 import { Inject, Injectable } from '@nestjs/common';
-import { Job, Worker } from 'bullmq';
+import { Job, QueueEvents, Worker } from 'bullmq';
 import {
   CopyJobData,
+  FeaturesJobCancelProgress,
+  FeaturesJobProgress,
   SplitJobData,
   StratificationJobData,
 } from '@marxan/geofeature-calculations';
 import { WorkerBuilder } from '@marxan-geoprocessing/modules/worker';
 import {
+  copyQueueEventsToken,
   copyQueueNameToken,
   copyWorkerBuilderToken,
+  splitQueueEventsToken,
   splitQueueNameToken,
   splitWorkerBuilderToken,
+  stratificationQueueEventsToken,
   stratificationQueueNameToken,
   stratificationWorkerBuilderToken,
 } from './worker-builder.providers';
@@ -31,36 +36,82 @@ export class ProcessingWorker {
     stratificationQueueName: string,
     @Inject(stratificationWorkerBuilderToken)
     stratificationWorkerBuilder: WorkerBuilder,
+    @Inject(copyQueueEventsToken)
+    copyQueueEvents: QueueEvents,
+    @Inject(splitQueueEventsToken)
+    splitQueueEvents: QueueEvents,
+    @Inject(stratificationQueueEventsToken)
+    stratificationQueueEvents: QueueEvents,
   ) {
     this.copyWorker = copyWorkerBuilder.build<CopyJobData, void>(
       copyQueueName,
       {
         process: this.copyProcess.bind(this),
       },
     );
+    copyQueueEvents.on(
+      `progress`,
+      async ({ data }: { data: FeaturesJobProgress }) => {
+        if (this.isCancel(data)) await this.cancelCopy(data);
+      },
+    );
     this.splitWorker = splitWorkerBuilder.build<SplitJobData, void>(
       splitQueueName,
       {
         process: this.splitProcess.bind(this),
       },
     );
+    splitQueueEvents.on(
+      `progress`,
+      async ({ data }: { data: FeaturesJobProgress }) => {
+        if (this.isCancel(data)) await this.cancelSplit(data);
+      },
+    );
     this.stratificationWorker = stratificationWorkerBuilder.build<
       StratificationJobData,
       void
     >(stratificationQueueName, {
       process: this.stratificationProcess.bind(this),
     });
+    stratificationQueueEvents.on(
+      `progress`,
+      async ({ data }: { data: FeaturesJobProgress }) => {
+        if (this.isCancel(data)) await this.cancelStratification(data);
+      },
+    );
+  }
+
+  private async copyProcess(_job: Job<CopyJobData>): Promise<void> {
+    //
   }
 
-  async copyProcess(_job: Job<CopyJobData>): Promise<void> {
+  private async splitProcess(_job: Job<SplitJobData>): Promise<void> {
     //
   }
 
-  async splitProcess(_job: Job<SplitJobData>): Promise<void> {
+  private async stratificationProcess(
+    _job: Job<StratificationJobData>,
+  ): Promise<void> {
     //
   }
 
-  async stratificationProcess(_job: Job<StratificationJobData>): Promise<void> {
+  private async cancelCopy(_data: FeaturesJobCancelProgress): Promise<void> {
     //
   }
+
+  private async cancelSplit(_data: FeaturesJobCancelProgress): Promise<void> {
+    //
+  }
+
+  private async cancelStratification(
+    _data: FeaturesJobCancelProgress,
+  ): Promise<void> {
+    //
+  }
+
+  private isCancel(
+    data: FeaturesJobProgress,
+  ): data is FeaturesJobCancelProgress {
+    return data.type === 'canceled' && data.canceled;
+  }
 }
diff --git a/api/apps/geoprocessing/src/modules/features/processing/worker-builder.providers.ts b/api/apps/geoprocessing/src/modules/features/processing/worker-builder.providers.ts
@@ -4,7 +4,11 @@ import {
   splitQueueName,
   stratificationQueueName,
 } from '@marxan/geofeature-calculations';
-import { WorkerBuilder } from '@marxan-geoprocessing/modules/worker';
+import {
+  QueueEventsBuilder,
+  WorkerBuilder,
+} from '@marxan-geoprocessing/modules/worker';
+import { QueueEvents } from 'bullmq';
 
 export const copyQueueNameToken = Symbol('copy queue token');
 export const copyQueueNameProvider: ValueProvider<string> = {
@@ -18,6 +22,14 @@ export const copyWorkerBuilderProvider: FactoryProvider<WorkerBuilder> = {
   inject: [WorkerBuilder],
 };
 
+export const copyQueueEventsToken = Symbol('copy queue events token');
+export const copyQueueEventsProvider: FactoryProvider<QueueEvents> = {
+  provide: copyQueueEventsToken,
+  useFactory: (queueEventsBuilder: QueueEventsBuilder, queueName: string) =>
+    queueEventsBuilder.buildQueueEvents(queueName),
+  inject: [QueueEventsBuilder, copyQueueNameToken],
+};
+
 export const splitQueueNameToken = Symbol('split queue token');
 export const splitQueueNameProvider: ValueProvider<string> = {
   provide: splitQueueNameToken,
@@ -30,6 +42,14 @@ export const splitWorkerBuilderProvider: FactoryProvider<WorkerBuilder> = {
   inject: [WorkerBuilder],
 };
 
+export const splitQueueEventsToken = Symbol('split queue events token');
+export const splitQueueEventsProvider: FactoryProvider<QueueEvents> = {
+  provide: splitQueueEventsToken,
+  useFactory: (queueEventsBuilder: QueueEventsBuilder, queueName: string) =>
+    queueEventsBuilder.buildQueueEvents(queueName),
+  inject: [QueueEventsBuilder, splitQueueNameToken],
+};
+
 export const stratificationQueueNameToken = Symbol(
   'stratification queue token',
 );
@@ -45,3 +65,13 @@ export const stratificationWorkerBuilderProvider: FactoryProvider<WorkerBuilder>
   useFactory: (builder: WorkerBuilder) => builder,
   inject: [WorkerBuilder],
 };
+
+export const stratificationQueueEventsToken = Symbol(
+  'stratification queue events token',
+);
+export const stratificationQueueEventsProvider: FactoryProvider<QueueEvents> = {
+  provide: stratificationQueueEventsToken,
+  useFactory: (queueEventsBuilder: QueueEventsBuilder, queueName: string) =>
+    queueEventsBuilder.buildQueueEvents(queueName),
+  inject: [QueueEventsBuilder, stratificationQueueNameToken],
+};
diff --git a/api/libs/api-events/src/api-event-kinds.enum.ts b/api/libs/api-events/src/api-event-kinds.enum.ts
@@ -1,3 +1,5 @@
+import { ValuesType } from 'utility-types';
+
 export enum API_EVENT_KINDS {
   user__signedUp__v1alpha1 = 'user.signedUp/v1alpha1',
   user__accountActivationTokenGenerated__v1alpha1 = 'user.accountActivationTokenGenerated/v1alpha1',
@@ -38,3 +40,13 @@ export type ScenarioEvents = Pick<
   typeof API_EVENT_KINDS,
   Extract<keyof typeof API_EVENT_KINDS, `scenario__${string}`>
 >;
+
+export type ScenarioGeoFeatureEventKeys = Extract<
+  keyof typeof API_EVENT_KINDS,
+  `scenario__geofeature${`Copy` | `Split` | `Stratification`}${string}`
+>;
+export type ScenarioGeofeatureEvents = Pick<
+  typeof API_EVENT_KINDS,
+  ScenarioGeoFeatureEventKeys
+>;
+export type ScenarioGeofeatureEventValues = ValuesType<ScenarioGeofeatureEvents>;
diff --git a/api/libs/api-events/src/index.ts b/api/libs/api-events/src/index.ts
@@ -1 +1 @@
-export { API_EVENT_KINDS } from './api-event-kinds.enum';
+export * from './api-event-kinds.enum';