feat(decorator): add OnWorkerEvent

Author: madeindjs

README.md

@@ -10,6 +10,12 @@ Why you should prefer Graphile Worker instead of [Bull](https://github.com/nestj
 
 1. You already have a PostgreSQL in your stack (and you don't want to add a Redis server)
 
+## Features
+
+- use a `GraphileWorkerModule` to register Graphile Worker with a `asRootAsync` to pass dynamic parameters
+- provide a `GraphileWorkerService` to add jobs or start runner
+- provide a `OnWorkerEvenet` decorator to add custom behavior on `job:success` for example
+
 ## Installation
 
 ```bash
@@ -121,6 +127,30 @@ async function bootstrap() {
 bootstrap();
 ```
 
+## `OnWorkerEvent` decorator
+
+You need to add `@GraphileWorkerListener` decorator on your class and then set `@OnWorkerEvent(eventName)` on method:
+
+```ts
+import { GraphileWorkerListener, OnWorkerEvent } from '@app/graphile-worker';
+import { Injectable, Logger } from '@nestjs/common';
+import { WorkerEventMap } from 'graphile-worker';
+
+@Injectable()
+@GraphileWorkerListener()
+export class AppService {
+  private readonly logger = new Logger(AppService.name);
+
+  @OnWorkerEvent('job:success')
+  onJobSuccess({ job }: WorkerEventMap['job:success']) {
+    this.logger.debug(`job #${job.id} finished`);
+    // output: [Nest] 1732  - 09/14/2021, 12:42:45 PM   DEBUG [AppService] job #349 finished
+  }
+}
+```
+
+You can find a complete list of available event on [Graphile Worker's documentation](https://github.com/graphile/worker#workerevents).
+
 ## Test
 
 ```bash

libs/graphile-worker/src/config.interface.ts

@@ -1,12 +1,12 @@
 import { FactoryProvider, ModuleMetadata } from '@nestjs/common';
 import { RunnerOptions } from 'graphile-worker';
 
-export type GraphileWorkerConfiguration = RunnerOptions;
+export type RunnerOptionWithoutEvents = Omit<RunnerOptions, 'events'>;
 
 export interface GraphileWorkerConfigurationFactory {
   createSharedConfiguration():
-    | Promise<GraphileWorkerConfiguration>
-    | GraphileWorkerConfiguration;
+    | Promise<RunnerOptionWithoutEvents>
+    | RunnerOptionWithoutEvents;
 }
 
 export interface GraphileWorkerAsyncConfiguration
@@ -26,7 +26,7 @@ export interface GraphileWorkerAsyncConfiguration
    */
   useFactory?: (
     ...args: any[]
-  ) => Promise<GraphileWorkerConfiguration> | GraphileWorkerConfiguration;
+  ) => Promise<RunnerOptionWithoutEvents> | RunnerOptionWithoutEvents;
 
   /**
    * Optional list of providers to be injected into the context of the Factory function.

libs/graphile-worker/src/configuration.service.ts

@@ -0,0 +1,15 @@
+import { Injectable } from '@nestjs/common';
+import { EventEmitter } from 'events';
+import { RunnerOptions } from 'graphile-worker';
+
+@Injectable()
+export class ConfigurationService {
+  constructor(public readonly config: RunnerOptions) {
+    const events = new EventEmitter();
+
+    this.config.events = events;
+  }
+}
+
+// TODO: try to remove this ?
+export const CONFIGURATION_SERVICE_KEY = Symbol.for(ConfigurationService.name);

libs/graphile-worker/src/graphile-worker.module.ts

@@ -1,13 +1,28 @@
 import { DynamicModule, Module, Provider } from '@nestjs/common';
+import { DiscoveryModule } from '@nestjs/core';
 import {
   GraphileWorkerAsyncConfiguration,
-  GraphileWorkerConfiguration,
   GraphileWorkerConfigurationFactory,
+  RunnerOptionWithoutEvents,
 } from './config.interface';
+import {
+  ConfigurationService,
+  CONFIGURATION_SERVICE_KEY,
+} from './configuration.service';
 import { GraphileWorkerService } from './graphile-worker.service';
+import { ListenerExplorerService } from './listener-explorer.service';
+import { MetadataAccessorService } from './metadata-accessor.service';
 
 export const GRAPHILE_WORKER_TOKEN = Symbol.for('NestJsGraphileWorker');
 
+const internalsProviders = [
+  MetadataAccessorService,
+  ListenerExplorerService,
+  GraphileWorkerService,
+];
+
+const internalsModules = [DiscoveryModule];
+
 @Module({
   providers: [GraphileWorkerService],
   exports: [GraphileWorkerService],
@@ -27,17 +42,20 @@ export class GraphileWorkerModule {
    *  }),
    * ```
    */
-  static forRoot(config: GraphileWorkerConfiguration): DynamicModule {
-    const graphileWorkerService: Provider = {
-      provide: GraphileWorkerService,
-      useValue: new GraphileWorkerService(config),
+  static forRoot(config: RunnerOptionWithoutEvents): DynamicModule {
+    const configurationService = new ConfigurationService(config);
+
+    const graphileConfigurationServiceProvider: Provider = {
+      provide: CONFIGURATION_SERVICE_KEY,
+      useValue: configurationService,
     };
 
     return {
       global: true,
+      imports: internalsModules,
       module: GraphileWorkerModule,
-      providers: [graphileWorkerService],
-      exports: [graphileWorkerService],
+      providers: [graphileConfigurationServiceProvider, ...internalsProviders],
+      exports: [GraphileWorkerService],
     };
   }
 
@@ -67,8 +85,8 @@ export class GraphileWorkerModule {
     return {
       global: true,
       module: GraphileWorkerModule,
-      imports: asyncConfig.imports,
-      providers,
+      imports: [...asyncConfig.imports, ...internalsModules],
+      providers: [...providers, ...internalsProviders],
       exports: providers,
     };
   }
@@ -84,23 +102,22 @@ export class GraphileWorkerModule {
   ): Provider {
     if (options.useFactory) {
       return {
-        provide: GraphileWorkerService,
+        provide: CONFIGURATION_SERVICE_KEY,
         useFactory: async (...args: any[]) => {
-          const configuration = await options.useFactory(...args);
-          return new GraphileWorkerService(configuration);
+          const config = await options.useFactory(...args);
+          return new ConfigurationService(config);
         },
         inject: options.inject || [],
       };
     }
 
     return {
-      provide: GraphileWorkerService,
+      provide: CONFIGURATION_SERVICE_KEY,
       useFactory: async (
         optionsFactory: GraphileWorkerConfigurationFactory,
       ) => {
-        const configuration = await optionsFactory.createSharedConfiguration();
-
-        return new GraphileWorkerService(configuration);
+        const config = await optionsFactory.createSharedConfiguration();
+        return new ConfigurationService(config);
       },
       inject: options.inject,
     };

libs/graphile-worker/src/graphile-worker.service.spec.ts

@@ -1,12 +1,35 @@
 import { Test, TestingModule } from '@nestjs/testing';
+import {
+  ConfigurationService,
+  CONFIGURATION_SERVICE_KEY,
+} from './configuration.service';
 import { GraphileWorkerService } from './graphile-worker.service';
+import { ListenerExplorerService } from './listener-explorer.service';
 
-describe('GraphileWorkerService', () => {
+describe(GraphileWorkerService.name, () => {
   let service: GraphileWorkerService;
 
   beforeEach(async () => {
     const module: TestingModule = await Test.createTestingModule({
-      providers: [GraphileWorkerService],
+      providers: [
+        GraphileWorkerService,
+        {
+          provide: ListenerExplorerService,
+          useValue: {
+            listeners: [],
+          },
+        },
+        {
+          useFactory: () => {
+            const configurationService = new ConfigurationService({
+              taskList: { hello: () => {} },
+              connectionString: 'postgres://example:password@postgres/example',
+            });
+            return configurationService;
+          },
+          provide: CONFIGURATION_SERVICE_KEY,
+        },
+      ],
     }).compile();
 
     service = module.get<GraphileWorkerService>(GraphileWorkerService);

libs/graphile-worker/src/graphile-worker.service.ts

@@ -1,4 +1,4 @@
-import { Injectable, Logger } from '@nestjs/common';
+import { Inject, Injectable, Logger } from '@nestjs/common';
 import {
   Job,
   makeWorkerUtils,
@@ -8,13 +8,31 @@ import {
   TaskSpec,
   WorkerUtils,
 } from 'graphile-worker';
+import {
+  ConfigurationService,
+  CONFIGURATION_SERVICE_KEY,
+} from './configuration.service';
+import { ListenerExplorerService } from './listener-explorer.service';
 
 @Injectable()
 export class GraphileWorkerService {
   private readonly logger = new Logger(GraphileWorkerService.name);
   private isMigrationDone: boolean;
+  private readonly options: RunnerOptions;
+
+  constructor(
+    @Inject(CONFIGURATION_SERVICE_KEY)
+    configuration: ConfigurationService,
+    private readonly explorerService: ListenerExplorerService,
+  ) {
+    configuration.config.events.on('job:success', (...args: any[]) => {
+      this.explorerService.listeners
+        .filter(({ event }) => event === 'job:success')
+        .forEach(({ callback }) => callback(...args));
+    });
 
-  constructor(private readonly options: RunnerOptions) {}
+    this.options = configuration.config;
+  }
 
   /**
    * Runs until either stopped by a signal event like `SIGINT` or by calling the `stop()` method on the resolved object.

libs/graphile-worker/src/index.ts

@@ -1,2 +1,3 @@
 export * from './graphile-worker.module';
 export * from './graphile-worker.service';
+export * from './worker-hooks.decorators';

libs/graphile-worker/src/listener-explorer.service.spec.ts

@@ -0,0 +1,49 @@
+import { Injectable } from '@nestjs/common';
+import { DiscoveryModule } from '@nestjs/core';
+import { Test, TestingModule } from '@nestjs/testing';
+import { GraphileWorkerListener, OnWorkerEvent } from '.';
+import { ListenerExplorerService } from './listener-explorer.service';
+import { MetadataAccessorService } from './metadata-accessor.service';
+
+@Injectable()
+@GraphileWorkerListener()
+class TestListenerService {
+  @OnWorkerEvent('job:success')
+  onJobSuccess() {}
+
+  @OnWorkerEvent('job:error')
+  onJobError() {}
+}
+
+describe(ListenerExplorerService.name, () => {
+  let service: ListenerExplorerService;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      imports: [DiscoveryModule],
+      providers: [
+        ListenerExplorerService,
+        MetadataAccessorService,
+        TestListenerService,
+      ],
+    }).compile();
+
+    service = module.get<ListenerExplorerService>(ListenerExplorerService);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+
+  describe('explore', () => {
+    it('should register TestListenerService', () => {
+      service.explore();
+      expect(service.listeners).toHaveLength(2);
+
+      const eventsRegistered = service.listeners.map((l) => l.event);
+
+      expect(eventsRegistered).toContain('job:success');
+      expect(eventsRegistered).toContain('job:error');
+    });
+  });
+});

libs/graphile-worker/src/listener-explorer.service.ts

@@ -0,0 +1,64 @@
+import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
+import { DiscoveryService } from '@nestjs/core';
+import { InstanceWrapper } from '@nestjs/core/injector/instance-wrapper';
+import { MetadataScanner } from '@nestjs/core/metadata-scanner';
+import { MetadataAccessorService } from './metadata-accessor.service';
+import { WorkerEventName } from './worker-hooks.decorators';
+
+/**
+ * This service is responsible to scan all [Graphile worker decorators](./worker-hooks.decorators.ts) and register them.
+ *
+ * Heavily inspired from [`BullExplorer`](https://github.com/nestjs/bull/blob/c230eab1dc26fb743a3428e61043167866b1e377/lib/bull.explorer.ts)
+ */
+@Injectable()
+export class ListenerExplorerService implements OnModuleInit {
+  private readonly logger = new Logger(ListenerExplorerService.name);
+
+  public readonly listeners: { event: WorkerEventName; callback: Function }[] =
+    [];
+
+  constructor(
+    private readonly discoveryService: DiscoveryService,
+    private readonly metadataAccessor: MetadataAccessorService,
+    private readonly metadataScanner: MetadataScanner,
+  ) {}
+
+  onModuleInit() {
+    this.explore();
+  }
+
+  explore() {
+    const providers: InstanceWrapper[] = this.discoveryService
+      .getProviders()
+      .filter((wrapper: InstanceWrapper) =>
+        this.metadataAccessor.isListener(wrapper.metatype),
+      );
+
+    providers.forEach((wrapper: InstanceWrapper) => {
+      const { instance } = wrapper;
+
+      this.metadataScanner.scanFromPrototype(
+        instance,
+        Object.getPrototypeOf(instance),
+        (key: string) => {
+          if (this.metadataAccessor.isWorkerEvent(instance[key])) {
+            const event = this.metadataAccessor.getListenerMetadata(
+              instance[key],
+            );
+
+            this.listeners.push({
+              event,
+              callback: (...args) => instance[key](...args),
+            });
+
+            this.logger.debug(
+              `Register ${event} from ${
+                (instance as Object).constructor.name
+              }.${key}`,
+            );
+          }
+        },
+      );
+    });
+  }
+}