feat: add event for planning unit geom calculation

api/apps/api/src/migrations/api/1632901740965-AddPlanningUnitCalculationEvents.ts

@@ -0,0 +1,25 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class AddPlanningUnitCalculationEvents1632901740965
+  implements MigrationInterface {
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      INSERT INTO api_event_kinds (id) values
+      ('project.planningUnits.submitted/v1/alpha'),
+      ('project.planningUnits.finished/v1/alpha'),
+      ('project.planningUnits.failed/v1/alpha');
+    `);
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `DELETE FROM api_event_kinds WHERE id = 'project.planningUnits.submitted/v1/alpha';`,
+    );
+    await queryRunner.query(
+      `DELETE FROM api_event_kinds WHERE id = 'project.planningUnits.finished/v1/alpha';`,
+    );
+    await queryRunner.query(
+      `DELETE FROM api_event_kinds WHERE id = 'project.planningUnits.failed/v1/alpha';`,
+    );
+  }
+}

api/apps/api/src/modules/analysis/analysis.module.ts

@@ -1,6 +1,6 @@
 import { Module } from '@nestjs/common';
 import { ApiEventsModule } from '@marxan-api/modules/api-events/api-events.module';
-import { queueName } from '@marxan-jobs/planning-unit-geometry';
+import { updateQueueName } from '@marxan-jobs/planning-unit-geometry';
 import { PlanningUnitsModule } from '@marxan-api/modules/planning-units/planning-units.module';
 import { ScenariosPlanningUnitModule } from '@marxan-api/modules/scenarios-planning-unit/scenarios-planning-unit.module';
 
@@ -20,7 +20,7 @@ import { UpdatePlanningUnitsEventsPort } from './providers/planning-units/update
     ScenariosPlanningUnitModule,
     PlanningUnitsModule,
     QueueModule.register({
-      name: queueName,
+      name: updateQueueName,
     }),
   ],
   providers: [

api/apps/api/src/modules/api-events/api-events.controller.ts

@@ -25,11 +25,7 @@ import {
 
 import { DeleteResult } from 'typeorm';
 
-import {
-  ApiEvent,
-  ApiEventResult,
-  QualifiedEventTopic,
-} from './api-event.api.entity';
+import { ApiEvent, ApiEventResult } from './api-event.api.entity';
 import { ApiEventsService } from './api-events.service';
 import { CreateApiEventDTO } from './dto/create.api-event.dto';
 import { API_EVENT_KINDS } from '@marxan/api-events';
@@ -88,6 +84,6 @@ export class ApiEventsController {
     @Param('kind') kind: API_EVENT_KINDS,
     @Param('topic') topic: string,
   ): Promise<DeleteResult> {
-    return await this.service.purgeAll({ kind, topic } as QualifiedEventTopic);
+    return await this.service.purgeAll({ kind, topic });
   }
 }

api/apps/api/src/modules/api-events/api-events.service.ts

@@ -3,6 +3,7 @@ import { InjectRepository } from '@nestjs/typeorm';
 
 import { DeleteResult, Repository } from 'typeorm';
 import { Either, left, right } from 'fp-ts/lib/Either';
+import { FindOperator } from 'typeorm/find-options/FindOperator';
 
 import {
   ApiEvent,
@@ -24,6 +25,12 @@ import { UpdateApiEventDTO } from './dto/update.api-event.dto';
 import { AppInfoDTO } from '../../dto/info.dto';
 import { AppConfig } from '../../utils/config.utils';
 
+export interface QualifiedEventTopicSearch
+  extends Omit<QualifiedEventTopic, 'kind'> {
+  topic: string;
+  kind: FindOperator<QualifiedEventTopic['kind']> | QualifiedEventTopic['kind'];
+}
+
 export const duplicate = Symbol();
 
 @Injectable()
@@ -57,12 +64,19 @@ export class ApiEventsService extends AppBaseService<
    * return the matching event with latest timestamp.
    */
   public async getLatestEventForTopic(
-    qualifiedTopic: QualifiedEventTopic,
+    qualifiedTopic: QualifiedEventTopicSearch,
   ): Promise<ApiEventByTopicAndKind | undefined> {
-    const result = await this.latestEventByTopicAndKindRepo.findOne({
-      topic: qualifiedTopic.topic,
-      kind: qualifiedTopic.kind,
-    });
+    const result = await this.latestEventByTopicAndKindRepo.findOne(
+      {
+        topic: qualifiedTopic.topic,
+        kind: qualifiedTopic.kind,
+      },
+      {
+        order: {
+          timestamp: 'DESC',
+        },
+      },
+    );
     if (!result) {
       throw new NotFoundException(
         `No events found for topic ${qualifiedTopic.topic} and kind ${qualifiedTopic.kind}.`,
@@ -98,7 +112,7 @@ export class ApiEventsService extends AppBaseService<
    * `QualifiedEventTopic` (i.e. a topic qualified by `kind` and `apiVersion`).
    */
   public async purgeAll(
-    qualifiedTopic?: QualifiedEventTopic,
+    qualifiedTopic?: QualifiedEventTopicSearch,
   ): Promise<DeleteResult> {
     if (!isNil(qualifiedTopic)) {
       this.logger.log(

api/apps/api/src/modules/planning-units/planning-units.module.ts

@@ -1,16 +1,22 @@
 import { Module } from '@nestjs/common';
 import { ProxyService } from '@marxan-api/modules/proxy/proxy.service';
 import { PlanningUnitsController } from './planning-units.controller';
-import { PlanningUnitsService } from './planning-units.service';
+import {
+  PlanningUnitsService,
+  queueEventsProvider,
+  queueProvider,
+} from './planning-units.service';
 import { QueueModule } from '@marxan-api/modules/queue/queue.module';
+import { ApiEventsModule } from '@marxan-api/modules/api-events';
 
 @Module({
-  imports: [
-    QueueModule.register({
-      name: 'planning-units',
-    }),
+  imports: [QueueModule.register(), ApiEventsModule],
+  providers: [
+    PlanningUnitsService,
+    ProxyService,
+    queueProvider,
+    queueEventsProvider,
   ],
-  providers: [PlanningUnitsService, ProxyService],
   exports: [PlanningUnitsService],
   controllers: [PlanningUnitsController],
 })

api/apps/api/src/modules/planning-units/planning-units.service.ts

@@ -1,15 +1,80 @@
-import { Injectable } from '@nestjs/common';
+import { FactoryProvider, Inject, Injectable } from '@nestjs/common';
 
-import { QueueService } from '@marxan-api/modules/queue/queue.service';
-import { CreatePlanningUnitsDTO } from './dto/create.planning-units.dto';
+import { Job, Queue, QueueEvents } from 'bullmq';
+import { ApiEventsService } from '@marxan-api/modules/api-events';
+import { API_EVENT_KINDS } from '@marxan/api-events';
+import {
+  PlanningUnitsJob,
+  createQueueName,
+} from '@marxan-jobs/planning-unit-geometry';
+import { QueueBuilder, QueueEventsBuilder } from '@marxan-api/modules/queue';
+
+export const queueToken = Symbol(`planning unit queue token`);
+export const queueProvider: FactoryProvider<Queue<PlanningUnitsJob, void>> = {
+  provide: queueToken,
+  useFactory: (builder: QueueBuilder) => builder.buildQueue(createQueueName),
+  inject: [QueueBuilder],
+};
+export const queueEventsToken = Symbol(`planning units queue events token`);
+export const queueEventsProvider: FactoryProvider<QueueEvents> = {
+  provide: queueEventsToken,
+  useFactory: (builder: QueueEventsBuilder) =>
+    builder.buildQueueEvents(createQueueName),
+  inject: [QueueEventsBuilder],
+};
+
+type CompletedEvent = { jobId: string; returnvalue: string };
+type FailedEvent = { jobId: string; failedReason: string };
 
 @Injectable()
 export class PlanningUnitsService {
   constructor(
-    private readonly queueService: QueueService<CreatePlanningUnitsDTO>,
-  ) {}
+    @Inject(queueToken)
+    private readonly queue: Queue<PlanningUnitsJob, void>,
+    @Inject(queueEventsToken)
+    private readonly queueEvents: QueueEvents,
+    private readonly events: ApiEventsService,
+  ) {
+    this.queueEvents.on(`completed`, (data, eventId) =>
+      this.onCompleted(data, eventId),
+    );
+    this.queueEvents.on(`failed`, (data, eventId) =>
+      this.onFailed(data, eventId),
+    );
+  }
+
+  public async create(jobDefinition: PlanningUnitsJob): Promise<void> {
+    const job = await this.queue.add('create-regular-pu', jobDefinition);
+    await this.events.createIfNotExists({
+      kind: API_EVENT_KINDS.project__planningUnits__submitted__v1__alpha,
+      topic: jobDefinition.projectId,
+      externalId: job.id,
+    });
+  }
+
+  private async onCompleted(data: CompletedEvent, eventId: string) {
+    const job: Job<PlanningUnitsJob> | undefined = await this.queue.getJob(
+      data.jobId,
+    );
+    if (!job) return;
+    const { projectId } = job.data;
+    await this.events.createIfNotExists({
+      kind: API_EVENT_KINDS.project__planningUnits__finished__v1__alpha,
+      topic: projectId,
+      externalId: eventId,
+    });
+  }
 
-  public async create(creationOptions: CreatePlanningUnitsDTO): Promise<void> {
-    await this.queueService.queue.add('create-regular-pu', creationOptions);
+  private async onFailed(data: FailedEvent, eventId: string) {
+    const job: Job<PlanningUnitsJob> | undefined = await this.queue.getJob(
+      data.jobId,
+    );
+    if (!job) return;
+    const { projectId } = job.data;
+    await this.events.createIfNotExists({
+      kind: API_EVENT_KINDS.project__planningUnits__failed__v1__alpha,
+      topic: projectId,
+      externalId: eventId,
+    });
   }
 }

api/apps/api/src/modules/projects/job-status/project-status.view.api.entity.ts

@@ -58,4 +58,9 @@ const eventToJobStatusMapping: Record<ValuesType<ProjectEvents>, JobStatus> = {
     JobStatus.done,
   [API_EVENT_KINDS.project__protectedAreas__submitted__v1__alpha]:
     JobStatus.running,
+  [API_EVENT_KINDS.project__planningUnits__submitted__v1__alpha]:
+    JobStatus.running,
+  [API_EVENT_KINDS.project__planningUnits__finished__v1__alpha]: JobStatus.done,
+  [API_EVENT_KINDS.project__planningUnits__failed__v1__alpha]:
+    JobStatus.failure,
 };

api/apps/api/src/modules/projects/projects-crud.service.ts

@@ -194,7 +194,12 @@ export class ProjectsCrudService extends AppBaseService<
         'creating planning unit job and assigning project to area',
       );
       await Promise.all([
-        this.planningUnitsService.create(createModel),
+        this.planningUnitsService.create({
+          ...createModel,
+          planningUnitAreakm2: createModel.planningUnitAreakm2,
+          planningUnitGridShape: createModel.planningUnitGridShape,
+          projectId: model.id,
+        }),
         this.planningAreasService.assignProject({
           projectId: model.id,
           planningAreaGeometryId: createModel.planningAreaId,
@@ -221,7 +226,12 @@ export class ProjectsCrudService extends AppBaseService<
         createModel.planningAreaId)
     ) {
       await Promise.all([
-        this.planningUnitsService.create(createModel),
+        this.planningUnitsService.create({
+          ...createModel,
+          planningUnitAreakm2: createModel.planningUnitAreakm2,
+          planningUnitGridShape: createModel.planningUnitGridShape,
+          projectId: model.id,
+        }),
         this.planningAreasService.assignProject({
           projectId: model.id,
           planningAreaGeometryId: createModel.planningAreaId,