chore(geoprocessing): protected-areas: geo extractor: emit api events

Vizzuality · May 27, 2021 · 0aeab28 · 0aeab28
1 parent 02c0978
commit 0aeab28
Show file tree

Hide file tree

Showing 20 changed files with 314 additions and 40 deletions.
diff --git a/api/src/migrations/api/1621963684141-ProjectSpecificWdpaEvents.ts b/api/src/migrations/api/1621963684141-ProjectSpecificWdpaEvents.ts
@@ -0,0 +1,18 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class CostSurfaceEvents1621963684141 implements MigrationInterface {
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      INSERT INTO api_event_kinds (id) values
+      ('project.protectedAreas.submitted/v1/alpha'),
+      ('project.protectedAreas.finished/v1/alpha'),
+      ('project.protectedAreas.failed/v1/alpha');
+    `);
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `DELETE FROM api_event_kinds WHERE id like 'project.protectedAreas.%';`,
+    );
+  }
+}
diff --git a/api/src/modules/api-events/api-event.api.entity.ts b/api/src/modules/api-events/api-event.api.entity.ts
@@ -32,6 +32,9 @@ export enum API_EVENT_KINDS {
   scenario__costSurface__shapeConversionFailed__v1_alpha1 = 'scenario.costSurface.shapeConversionFailed/v1alpha1',
   scenario__costSurface__costUpdateFailed__v1_alpha1 = 'scenario.costSurface.costUpdateFailed/v1alpha1',
   scenario__costSurface__finished__v1_alpha1 = 'scenario.costSurface.finished/v1alpha1',
+  project__protectedAreas__submitted__v1__alpha = 'project.protectedAreas.submitted/v1/alpha',
+  project__protectedAreas__finished__v1__alpha = 'project.protectedAreas.finished/v1/alpha',
+  project__protectedAreas__failed__v1__alpha = 'project.protectedAreas.failed/v1/alpha',
 }
 
 /**

diff --git a/api/src/modules/projects/protected-areas/__mocks__/api-service.fake.ts b/api/src/modules/projects/protected-areas/__mocks__/api-service.fake.ts
@@ -0,0 +1,11 @@
+import { CreateApiEventDTO } from '../../../api-events/dto/create.api-event.dto';
+import { AppInfoDTO } from '../../../../dto/info.dto';
+import { ApiEvent } from '../../../api-events/api-event.api.entity';
+
+export class ApiServiceFake {
+  mock = jest.fn();
+
+  create(createModel: CreateApiEventDTO, info?: AppInfoDTO): Promise<ApiEvent> {
+    return this.mock(createModel, info);
+  }
+}
diff --git a/api/src/modules/projects/protected-areas/protected-areas.facade.spec.ts b/api/src/modules/projects/protected-areas/protected-areas.facade.spec.ts
@@ -8,10 +8,13 @@ import { Queue } from 'bullmq';
 
 import { QueueService } from '../../queue/queue.service';
 import { FakeLogger } from '../../../utils/__mocks__/fake-logger';
+import { ApiServiceFake } from './__mocks__/api-service.fake';
+import { ApiEventsService } from '../../api-events/api-events.service';
 
 let sut: ProtectedAreasFacade;
 let logger: FakeLogger;
 let addJobMock: jest.SpyInstance;
+let apiEvents: ApiServiceFake;
 
 const projectId = 'project-id';
 const file: Express.Multer.File = {
@@ -35,11 +38,16 @@ beforeEach(async () => {
         provide: Logger,
         useClass: FakeLogger,
       },
+      {
+        provide: ApiEventsService,
+        useClass: ApiServiceFake,
+      },
     ],
   }).compile();
 
   sut = sandbox.get(ProtectedAreasFacade);
   logger = sandbox.get(Logger);
+  apiEvents = sandbox.get(ApiEventsService);
 });
 
 describe(`when job submits successfully`, () => {
@@ -68,6 +76,20 @@ describe(`when job submits successfully`, () => {
       ]
     `);
   });
+
+  it(`should emit 'submitted' event`, () => {
+    expect(apiEvents.mock.mock.calls).toMatchInlineSnapshot(`
+      Array [
+        Array [
+          Object {
+            "kind": "project.protectedAreas.submitted/v1/alpha",
+            "topic": "project-id",
+          },
+          undefined,
+        ],
+      ]
+    `);
+  });
 });
 
 describe(`when job submission fails`, () => {
@@ -87,7 +109,32 @@ describe(`when job submission fails`, () => {
     expect(logger.error.mock.calls[0]).toMatchInlineSnapshot(`
       Array [
         "Failed submitting job to queue for project-id",
-        [Error: Oups],
+        "Error: Oups",
+      ]
+    `);
+  });
+
+  it(`emits both submitted&failed events`, () => {
+    expect(apiEvents.mock.mock.calls).toMatchInlineSnapshot(`
+      Array [
+        Array [
+          Object {
+            "kind": "project.protectedAreas.submitted/v1/alpha",
+            "topic": "project-id",
+          },
+          undefined,
+        ],
+        Array [
+          Object {
+            "data": Object {
+              "error": "Failed submission",
+              "message": "Error: Oups",
+            },
+            "kind": "project.protectedAreas.failed/v1/alpha",
+            "topic": "project-id",
+          },
+          undefined,
+        ],
       ]
     `);
   });

diff --git a/api/src/modules/projects/protected-areas/protected-areas.facade.ts b/api/src/modules/projects/protected-areas/protected-areas.facade.ts
@@ -1,6 +1,8 @@
 import { Injectable, Logger } from '@nestjs/common';
 import { Express } from 'express';
 import { QueueService } from '../../queue/queue.service';
+import { ApiEventsService } from '../../api-events/api-events.service';
+import { API_EVENT_KINDS } from '../../api-events/api-event.api.entity';
 
 export interface ProtectedAreasJobInput {
   projectId: string;
@@ -11,6 +13,7 @@ export interface ProtectedAreasJobInput {
 export class ProtectedAreasFacade {
   constructor(
     private readonly queueService: QueueService<ProtectedAreasJobInput>,
+    private readonly apiEvents: ApiEventsService,
     private readonly logger: Logger = new Logger(ProtectedAreasFacade.name),
   ) {}
 
@@ -21,14 +24,36 @@ export class ProtectedAreasFacade {
         file,
       })
       .then(() => {
-        // ok
+        return this.apiEvents.create({
+          kind: API_EVENT_KINDS.project__protectedAreas__submitted__v1__alpha,
+          topic: projectId,
+        });
       })
-      .catch((error) => {
-        this.logger.error(
-          `Failed submitting job to queue for ${projectId}`,
-          error,
-        );
-        throw error; // failed submission
+      .catch(async (error) => {
+        await this.markAsFailedSubmission(projectId, error);
+        throw error;
       });
   }
+
+  private markAsFailedSubmission = async (
+    projectId: string,
+    error: unknown,
+  ) => {
+    this.logger.error(
+      `Failed submitting job to queue for ${projectId}`,
+      String(error),
+    );
+    await this.apiEvents.create({
+      kind: API_EVENT_KINDS.project__protectedAreas__submitted__v1__alpha,
+      topic: projectId,
+    });
+    await this.apiEvents.create({
+      kind: API_EVENT_KINDS.project__protectedAreas__failed__v1__alpha,
+      topic: projectId,
+      data: {
+        error: `Failed submission`,
+        message: String(error),
+      },
+    });
+  };
 }
diff --git a/api/src/modules/projects/protected-areas/protected-areas.module.ts b/api/src/modules/projects/protected-areas/protected-areas.module.ts
@@ -2,9 +2,11 @@ import { Logger, Module } from '@nestjs/common';
 import { ProtectedAreasFacade } from './protected-areas.facade';
 import { QueueModule } from '../../queue/queue.module';
 import { queueName } from './queue-name';
+import { ApiEventsModule } from '../../api-events/api-events.module';
 
 @Module({
   imports: [
+    ApiEventsModule,
     QueueModule.register({
       name: queueName,
     }),

diff --git a/geoprocessing/package.json b/geoprocessing/package.json
@@ -37,6 +37,7 @@
   "dependencies": {
     "@nestjs/common": "^7.6.5",
     "@nestjs/core": "^7.5.1",
+    "@nestjs/cqrs": "7.0.1",
     "@nestjs/platform-express": "^7.5.1",
     "@nestjs/swagger": "^4.8.0",
     "@nestjs/typeorm": "^7.1.5",

diff --git a/geoprocessing/src/modules/api-events/api-events-subscriber.spec.ts b/geoprocessing/src/modules/api-events/api-events-subscriber.spec.ts
@@ -0,0 +1,63 @@
+import { ApiEventsSubscriber } from './api-events-subscriber';
+import { CqrsModule, EventBus } from '@nestjs/cqrs';
+import { Test } from '@nestjs/testing';
+import { ApiEventsService } from './api-events.service';
+import { API_EVENT_KINDS } from './events.enum';
+import { ApiEvent } from './api.event';
+
+let eventForwarder: FakeForwarder;
+let eventBus: EventBus;
+
+beforeEach(async () => {
+  const sandbox = await Test.createTestingModule({
+    imports: [CqrsModule],
+    providers: [
+      {
+        provide: ApiEventsService,
+        useClass: FakeForwarder,
+      },
+      ApiEventsSubscriber,
+    ],
+  })
+    .compile()
+    .then((app) => app.init());
+
+  eventBus = sandbox.get(EventBus);
+  eventForwarder = sandbox.get(ApiEventsService);
+});
+
+describe(`when emitting event`, () => {
+  beforeEach(async () => {
+    await eventBus.publish(
+      new ApiEvent(
+        'resource',
+        API_EVENT_KINDS.project__protectedAreas__failed__v1__alpha,
+        { 1: 'one' },
+      ),
+    );
+  });
+
+  it(`should forward it`, () => {
+    expect(eventForwarder.mock.mock.calls[0]).toMatchInlineSnapshot(`
+      Array [
+        "resource",
+        "project.protectedAreas.failed/v1/alpha",
+        Object {
+          "1": "one",
+        },
+      ]
+    `);
+  });
+});
+
+class FakeForwarder {
+  mock = jest.fn();
+
+  async create<T>(
+    resourceId: string,
+    kind: API_EVENT_KINDS,
+    data?: T,
+  ): Promise<void> {
+    return this.mock(resourceId, kind, data);
+  }
+}
diff --git a/geoprocessing/src/modules/api-events/api-events-subscriber.ts b/geoprocessing/src/modules/api-events/api-events-subscriber.ts
@@ -0,0 +1,39 @@
+import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common';
+import { EventBus } from '@nestjs/cqrs';
+import { Subscription } from 'rxjs';
+import { ApiEventsService } from './api-events.service';
+import { ApiEvent } from './api.event';
+
+@Injectable()
+export class ApiEventsSubscriber implements OnModuleDestroy {
+  #sub: Subscription;
+  private readonly logger = new Logger(ApiEventsSubscriber.name);
+
+  constructor(
+    private readonly eventBus: EventBus,
+    private readonly eventForwarder: ApiEventsService, // "repository"
+  ) {
+    this.#sub = eventBus.subscribe({
+      next: async (value) => {
+        if (value instanceof ApiEvent) {
+          try {
+            await this.eventForwarder.create(
+              value.resourceId,
+              value.kind,
+              value.data,
+            );
+          } catch (error) {
+            this.logger.error(
+              `Unable to emit ApiEvent ${value.resourceId}+${value.kind}`,
+              error,
+            );
+          }
+        }
+      },
+    });
+  }
+
+  onModuleDestroy() {
+    this.#sub.unsubscribe();
+  }
+}
diff --git a/geoprocessing/src/modules/api-events/api-events.module.ts b/geoprocessing/src/modules/api-events/api-events.module.ts
@@ -1,9 +1,11 @@
 import { HttpModule, Module } from '@nestjs/common';
+import { CqrsModule } from '@nestjs/cqrs';
+import { ApiEventsSubscriber } from './api-events-subscriber';
 import { ApiEventsService } from './api-events.service';
 
 @Module({
-  imports: [HttpModule],
-  providers: [ApiEventsService],
+  imports: [HttpModule, CqrsModule],
+  providers: [ApiEventsService, ApiEventsSubscriber],
   exports: [ApiEventsService],
 })
 export class ApiEventsModule {}
diff --git a/geoprocessing/src/modules/api-events/api-events.service.ts b/geoprocessing/src/modules/api-events/api-events.service.ts
@@ -16,23 +16,24 @@ export class ApiEventsService {
     this.#apiUrl = AppConfig.get<string>('api.url')!;
   }
 
-  async create<T>(
+  async create<T extends Record<string, unknown>>(
     resourceId: string,
     kind: API_EVENT_KINDS,
     data?: T,
   ): Promise<void> {
-    // TODO what if it failed? (currently validateStatus "swallows" the error)
     await this.http
       .post(
-        this.#apiUrl + `/v1/api-events`,
+        this.#apiUrl + `/api/v1/api-events`,
         {
           kind,
           topic: resourceId,
-          data,
+          data: JSON.stringify(data),
         },
         {
           headers: {
             Accept: 'application/json',
+            'Content-Type': 'application/json',
+            'x-api-key': this.#secret,
           },
           validateStatus: () => true,
         },

diff --git a/geoprocessing/src/modules/api-events/api.event.ts b/geoprocessing/src/modules/api-events/api.event.ts
@@ -0,0 +1,10 @@
+import { IEvent } from '@nestjs/cqrs';
+import { API_EVENT_KINDS } from './events.enum';
+
+export class ApiEvent<T extends Record<string, unknown>> implements IEvent {
+  constructor(
+    public readonly resourceId: string,
+    public readonly kind: API_EVENT_KINDS,
+    public readonly data?: T,
+  ) {}
+}
diff --git a/geoprocessing/src/modules/api-events/events.enum.ts b/geoprocessing/src/modules/api-events/events.enum.ts
@@ -10,4 +10,12 @@ export enum API_EVENT_KINDS {
   user__passwordResetTokenGenerated__v1alpha1 = 'user.passwordResetTokenGenerated/v1alpha1',
   user__passwordResetSucceeded__v1alpha1 = 'user.passwordResetSucceeded/v1alpha1',
   user__passwordResetFailed__v1alpha1 = 'user.passwordResetFailed/v1alpha1',
+  scenario__costSurface__submitted__v1_alpha1 = 'scenario.costSurface.submitted/v1alpha1',
+  scenario__costSurface__shapeConverted__v1_alpha1 = 'scenario.costSurface.shapeConverted/v1alpha1',
+  scenario__costSurface__shapeConversionFailed__v1_alpha1 = 'scenario.costSurface.shapeConversionFailed/v1alpha1',
+  scenario__costSurface__costUpdateFailed__v1_alpha1 = 'scenario.costSurface.costUpdateFailed/v1alpha1',
+  scenario__costSurface__finished__v1_alpha1 = 'scenario.costSurface.finished/v1alpha1',
+  project__protectedAreas__submitted__v1__alpha = 'project.protectedAreas.submitted/v1/alpha',
+  project__protectedAreas__finished__v1__alpha = 'project.protectedAreas.finished/v1/alpha',
+  project__protectedAreas__failed__v1__alpha = 'project.protectedAreas.failed/v1/alpha',
 }
diff --git a/geoprocessing/src/modules/api-events/index.ts b/geoprocessing/src/modules/api-events/index.ts
@@ -0,0 +1,2 @@
+export { ApiEventsModule } from './api-events.module';
+export { ApiEvent } from './api.event';