Feat/marxan 1560 legacy project import batch failed saga

api/apps/api/src/modules/legacy-project-import/infra/failed-legacy-project-import-db-cleanup-queue.provider.ts

@@ -0,0 +1,30 @@
+import { QueueBuilder } from '@marxan-api/modules/queue';
+import {
+  FailedLegacyProjectImportDbCleanupJobInput,
+  FailedLegacyProjectImportDbCleanupJobOutput,
+  failedLegacyProjectImportDbCleanupQueueName,
+} from '@marxan/legacy-project-import';
+import { FactoryProvider } from '@nestjs/common';
+import { Queue } from 'bullmq';
+
+export const failedLegacyProjectImportDbCleanupQueueToken = Symbol(
+  'failed legacy project import db cleanup queue token',
+);
+
+export const failedLegacyProjectImportDbCleanupQueueProvider: FactoryProvider<
+  Queue<
+    FailedLegacyProjectImportDbCleanupJobInput,
+    FailedLegacyProjectImportDbCleanupJobOutput
+  >
+> = {
+  provide: failedLegacyProjectImportDbCleanupQueueToken,
+  useFactory: (
+    queueBuilder: QueueBuilder<
+      FailedLegacyProjectImportDbCleanupJobInput,
+      FailedLegacyProjectImportDbCleanupJobOutput
+    >,
+  ) => {
+    return queueBuilder.buildQueue(failedLegacyProjectImportDbCleanupQueueName);
+  },
+  inject: [QueueBuilder],
+};

api/apps/api/src/modules/legacy-project-import/infra/legacy-project-import-batch-failed.saga.ts

@@ -0,0 +1,25 @@
+import { Injectable } from '@nestjs/common';
+import { ICommand, ofType, Saga } from '@nestjs/cqrs';
+import { Observable, of } from 'rxjs';
+import { mergeMap } from 'rxjs/operators';
+import { MarkLegacyProjectImportAsFailed } from '../application/mark-legacy-project-import-as-failed.command';
+import { LegacyProjectImportBatchFailed } from '../domain/events/legacy-project-import-batch-failed.event';
+import { ScheduleDbCleanupForFailedLegacyProjectImport } from './schedule-db-cleanup-for-failed-legacy-project-import.command';
+
+@Injectable()
+export class LegacyProjectImportBatchFailedSaga {
+  @Saga()
+  saga = (events$: Observable<any>): Observable<ICommand> =>
+    events$.pipe(
+      ofType(LegacyProjectImportBatchFailed),
+      mergeMap((event) =>
+        of(
+          new MarkLegacyProjectImportAsFailed(
+            event.projectId,
+            `${event.batchNumber} batch contains failed pieces`,
+          ),
+          new ScheduleDbCleanupForFailedLegacyProjectImport(event.projectId),
+        ),
+      ),
+    );
+}

api/apps/api/src/modules/legacy-project-import/infra/legacy-project-import.infra.module.ts

@@ -2,18 +2,24 @@ import { ApiEventsModule } from '@marxan-api/modules/api-events';
 import { QueueApiEventsModule } from '@marxan-api/modules/queue-api-events';
 import { Logger, Module, Scope } from '@nestjs/common';
 import { CqrsModule } from '@nestjs/cqrs';
+import { failedLegacyProjectImportDbCleanupQueueProvider } from './failed-legacy-project-import-db-cleanup-queue.provider';
+import { LegacyProjectImportBatchFailedSaga } from './legacy-project-import-batch-failed.saga';
 import {
-  importLegacyProjectPieceQueueProvider,
-  importLegacyProjectPiecenQueueEventsProvider,
   importLegacyProjectPieceEventsFactoryProvider,
+  importLegacyProjectPiecenQueueEventsProvider,
+  importLegacyProjectPieceQueueProvider,
 } from './legacy-project-import-queue.provider';
+import { ScheduleDbCleanupForFailedLegacyProjectImportHandler } from './schedule-db-cleanup-for-failed-legacy-project-import.handler';
 
 @Module({
   imports: [ApiEventsModule, QueueApiEventsModule, CqrsModule],
   providers: [
     importLegacyProjectPieceQueueProvider,
     importLegacyProjectPiecenQueueEventsProvider,
     importLegacyProjectPieceEventsFactoryProvider,
+    failedLegacyProjectImportDbCleanupQueueProvider,
+    LegacyProjectImportBatchFailedSaga,
+    ScheduleDbCleanupForFailedLegacyProjectImportHandler,
     {
       provide: Logger,
       useClass: Logger,

api/apps/api/src/modules/legacy-project-import/infra/schedule-db-cleanup-for-failed-legacy-project-import.command.ts

@@ -0,0 +1,8 @@
+import { ResourceId } from '@marxan/cloning/domain';
+import { Command } from '@nestjs-architects/typed-cqrs';
+
+export class ScheduleDbCleanupForFailedLegacyProjectImport extends Command<void> {
+  constructor(public readonly projectId: ResourceId) {
+    super();
+  }
+}

api/apps/api/src/modules/legacy-project-import/infra/schedule-db-cleanup-for-failed-legacy-project-import.handler.ts

@@ -0,0 +1,39 @@
+import { FailedLegacyProjectImportDbCleanupJobInput } from '@marxan/legacy-project-import';
+import { Inject, Logger } from '@nestjs/common';
+import { CommandHandler, IInferredCommandHandler } from '@nestjs/cqrs';
+import { Queue } from 'bullmq';
+import { failedLegacyProjectImportDbCleanupQueueToken } from './failed-legacy-project-import-db-cleanup-queue.provider';
+import { ScheduleDbCleanupForFailedLegacyProjectImport } from './schedule-db-cleanup-for-failed-legacy-project-import.command';
+
+@CommandHandler(ScheduleDbCleanupForFailedLegacyProjectImport)
+export class ScheduleDbCleanupForFailedLegacyProjectImportHandler
+  implements
+    IInferredCommandHandler<ScheduleDbCleanupForFailedLegacyProjectImport> {
+  constructor(
+    @Inject(failedLegacyProjectImportDbCleanupQueueToken)
+    private readonly queue: Queue<FailedLegacyProjectImportDbCleanupJobInput>,
+    private readonly logger: Logger,
+  ) {
+    this.logger.setContext(
+      ScheduleDbCleanupForFailedLegacyProjectImportHandler.name,
+    );
+  }
+
+  async execute({
+    projectId,
+  }: ScheduleDbCleanupForFailedLegacyProjectImport): Promise<void> {
+    const job = await this.queue.add(
+      `failed-legacy-project-import-db-cleanup`,
+      {
+        projectId: projectId.value,
+      },
+    );
+
+    if (!job) {
+      this.logger.error(
+        `failed-legacy-project-import-db-cleanup job couldn't be added to queue`,
+      );
+      return;
+    }
+  }
+}

api/libs/legacy-project-import/src/failed-legacy-project-import-db-cleanup-queue-name.ts

@@ -0,0 +1,2 @@
+export const failedLegacyProjectImportDbCleanupQueueName =
+  'failed-legacy-project-import-db-cleanup';

api/libs/legacy-project-import/src/index.ts

@@ -1,6 +1,13 @@
 export * from './domain';
 export * from './infra';
-export { LegacyProjectImportJobInput } from './job-input';
-export { LegacyProjectImportJobOutput } from './job-output';
+export {
+  LegacyProjectImportJobInput,
+  FailedLegacyProjectImportDbCleanupJobInput,
+} from './job-input';
+export {
+  LegacyProjectImportJobOutput,
+  FailedLegacyProjectImportDbCleanupJobOutput,
+} from './job-output';
 export { LegacyProjectImportPieceProcessor } from './legacy-project-import-piece-processor.port';
 export { legacyProjectImportQueueName } from './legacy-project-import-queue-name';
+export { failedLegacyProjectImportDbCleanupQueueName } from './failed-legacy-project-import-db-cleanup-queue-name';

api/libs/legacy-project-import/src/job-input.ts

@@ -9,3 +9,7 @@ export interface LegacyProjectImportJobInput {
   readonly scenarioId: string;
   readonly piece: LegacyProjectImportPiece;
 }
+
+export interface FailedLegacyProjectImportDbCleanupJobInput {
+  readonly projectId: string;
+}

api/libs/legacy-project-import/src/job-output.ts

@@ -10,3 +10,7 @@ export interface LegacyProjectImportJobOutput {
   readonly piece: LegacyProjectImportPiece;
   readonly warnings?: string[];
 }
+
+export interface FailedLegacyProjectImportDbCleanupJobOutput {
+  readonly projectId: string;
+}