Consolidate queue configs

.github/workflows/build.yml

@@ -47,6 +47,7 @@ jobs:
     runs-on: ubuntu-latest
     needs: determine-service-matrix
     strategy:
+      fail-fast: false
       matrix:
         service: ${{ fromJson(needs.determine-service-matrix.outputs.services) }}
         exclude:

services/account/apps/api/src/api.module.ts

@@ -1,7 +1,6 @@
 import '@frequency-chain/api-augment';
 import { Module } from '@nestjs/common';
 import { EventEmitter2, EventEmitterModule } from '@nestjs/event-emitter';
-import { BullModule } from '@nestjs/bullmq';
 import { ScheduleModule } from '@nestjs/schedule';
 import { RedisModule } from '@liaoliaots/nestjs-redis';
 import { BullBoardModule } from '@bull-board/nestjs';
@@ -10,20 +9,17 @@ import { ExpressAdapter } from '@bull-board/express';
 import { ConfigModule } from '#lib/config/config.module';
 import { ConfigService } from '#lib/config/config.service';
 import { BlockchainModule } from '#lib/blockchain/blockchain.module';
-import { QueueConstants } from '#lib/utils/queues';
 import { redisEventsToEventEmitter } from '#lib/utils/redis';
 import { EnqueueService } from '#lib/services/enqueue-request.service';
-import { Redis } from 'ioredis';
-import { AccountsControllerV1 } from './controllers/v1/accounts-v1.controller';
-import { DelegationControllerV1 } from './controllers/v1/delegation-v1.controller';
-import { HandlesControllerV1 } from './controllers/v1/handles-v1.controller';
-import { KeysControllerV1 } from './controllers/v1/keys-v1.controller';
-import { AccountsService } from './services/accounts.service';
-import { ApiService } from './services/api.service';
-import { DelegationService } from './services/delegation.service';
-import { HandlesService } from './services/handles.service';
-import { KeysService } from './services/keys.service';
-import { HealthController } from './controllers/health.controller';
+import { QueueModule, QueueConstants } from '#lib/queues';
+import {
+  AccountsControllerV1,
+  DelegationControllerV1,
+  HandlesControllerV1,
+  KeysControllerV1,
+  HealthController,
+} from './controllers';
+import { ApiService, AccountsService, HandlesService, DelegationService, KeysService } from './services';
 
 @Module({
   imports: [
@@ -67,21 +63,7 @@ import { HealthController } from './controllers/health.controller';
       },
       true, // isGlobal
     ),
-    BullModule.forRootAsync({
-      imports: [ConfigModule],
-      useFactory: (configService: ConfigService) => ({
-        connection: new Redis(configService.redisUrl.toString()),
-      }),
-      inject: [ConfigService],
-    }),
-    BullModule.registerQueue({
-      name: QueueConstants.TRANSACTION_PUBLISH_QUEUE,
-      defaultJobOptions: {
-        removeOnComplete: 20,
-        removeOnFail: false,
-        attempts: 1,
-      },
-    }),
+    QueueModule.forRoot(),
     // Bullboard UI
     BullBoardModule.forRoot({
       route: '/queues',

services/account/apps/api/src/controllers/index.ts

@@ -0,0 +1,2 @@
+export * from './health.controller';
+export * from './v1';

services/account/apps/api/src/controllers/v1/index.ts

@@ -0,0 +1,4 @@
+export * from './accounts-v1.controller';
+export * from './delegation-v1.controller';
+export * from './handles-v1.controller';
+export * from './keys-v1.controller';

services/account/apps/api/src/services/index.ts

@@ -0,0 +1,5 @@
+export * from './accounts.service';
+export * from './api.service';
+export * from './delegation.service';
+export * from './handles.service';
+export * from './keys.service';

services/account/apps/worker/src/transaction_notifier/notifier.module.ts

@@ -1,24 +1,10 @@
-import { BullModule } from '@nestjs/bullmq';
 import { Module } from '@nestjs/common';
 import { BlockchainModule } from '#lib/blockchain/blockchain.module';
-import { QueueConstants } from '#lib/utils/queues';
-import { ConfigModule } from '#lib/config/config.module';
 import { EnqueueService } from '#lib/services/enqueue-request.service';
 import { TxnNotifierService } from './notifier.service';
 
 @Module({
-  imports: [
-    BlockchainModule,
-    ConfigModule,
-    BullModule.registerQueue({
-      name: QueueConstants.TRANSACTION_PUBLISH_QUEUE,
-      defaultJobOptions: {
-        removeOnComplete: 20,
-        removeOnFail: false,
-        attempts: 3,
-      },
-    }),
-  ],
+  imports: [BlockchainModule],
   providers: [EnqueueService, TxnNotifierService],
   exports: [EnqueueService, TxnNotifierService],
 })

services/account/apps/worker/src/transaction_notifier/notifier.service.ts

@@ -4,10 +4,8 @@ import { Injectable, Logger, OnApplicationBootstrap, OnApplicationShutdown } fro
 import Redis from 'ioredis';
 import { MILLISECONDS_PER_SECOND } from 'time-constants';
 import axios from 'axios';
-import { BlockchainConstants } from '#lib/blockchain/blockchain-constants';
 import { BlockchainService } from '#lib/blockchain/blockchain.service';
 import { TransactionType } from '#lib/types/enums';
-import { QueueConstants } from '#lib/utils/queues';
 import { SECONDS_PER_BLOCK, TxWebhookRsp, RedisUtils } from 'libs/common/src';
 import { createWebhookRsp } from '#worker/transaction_notifier/notifier.service.helper.createWebhookRsp';
 import { BlockchainScannerService, NullScanError } from '#lib/utils/blockchain-scanner.service';
@@ -17,6 +15,7 @@ import { HexString } from '@polkadot/util/types';
 import { ITxStatus } from '#lib/interfaces/tx-status.interface';
 import { FrameSystemEventRecord } from '@polkadot/types/lookup';
 import { ConfigService } from '#lib/config/config.service';
+import { QueueConstants } from '#lib/queues';
 
 @Injectable()
 export class TxnNotifierService

services/account/apps/worker/src/transaction_publisher/publisher.module.ts

@@ -1,25 +1,12 @@
 import '@frequency-chain/api-augment';
 
-import { BullModule } from '@nestjs/bullmq';
 import { Module } from '@nestjs/common';
 import { BlockchainModule } from '#lib/blockchain/blockchain.module';
 import { NonceService } from '#lib/services/nonce.service';
-import { QueueConstants } from '#lib/utils/queues';
-import { ConfigModule } from '#lib/config/config.module';
 import { TransactionPublisherService } from './publisher.service';
 
 @Module({
-  imports: [
-    BlockchainModule,
-    ConfigModule,
-    BullModule.registerQueue({
-      name: QueueConstants.TRANSACTION_PUBLISH_QUEUE,
-      defaultJobOptions: {
-        removeOnComplete: 20,
-        removeOnFail: false,
-      },
-    }),
-  ],
+  imports: [BlockchainModule],
   providers: [TransactionPublisherService, NonceService],
   exports: [TransactionPublisherService],
 })

services/account/apps/worker/src/transaction_publisher/publisher.service.ts

@@ -3,7 +3,6 @@ import { InjectQueue, Processor } from '@nestjs/bullmq';
 import { Injectable, OnApplicationShutdown } from '@nestjs/common';
 import { DelayedError, Job, Queue } from 'bullmq';
 import Redis from 'ioredis';
-import { Hash } from '@polkadot/types/interfaces';
 import { KeyringPair } from '@polkadot/keyring/types';
 import { SubmittableExtrinsic } from '@polkadot/api-base/types';
 import { Codec, ISubmittableResult } from '@polkadot/types/types';
@@ -13,12 +12,11 @@ import { BlockchainService } from '#lib/blockchain/blockchain.service';
 import { createKeys } from '#lib/blockchain/create-keys';
 import { NonceService } from '#lib/services/nonce.service';
 import { TransactionType } from '#lib/types/enums';
-import { QueueConstants } from '#lib/utils/queues';
+import { QueueConstants } from '#lib/queues';
 import { BaseConsumer } from '#worker/BaseConsumer';
 import { RedisUtils, TransactionData } from 'libs/common/src';
 import { ConfigService } from '#lib/config/config.service';
 import { ITxStatus } from '#lib/interfaces/tx-status.interface';
-import { IsEvent } from '@polkadot/types/metadata/decorate/types';
 import { HexString } from '@polkadot/util/types';
 
 export const SECONDS_PER_BLOCK = 12;
@@ -34,7 +32,7 @@ export class TransactionPublisherService extends BaseConsumer implements OnAppli
     await this.checkCapacity();
   }
 
-  public async onApplicationShutdown(signal?: string | undefined): Promise<void> {
+  public async onApplicationShutdown(_signal?: string | undefined): Promise<void> {
     try {
       this.schedulerRegistry.deleteTimeout(CAPACITY_EPOCH_TIMEOUT_NAME);
     } catch (err) {

services/account/apps/worker/src/worker.module.ts

@@ -1,17 +1,15 @@
 import '@frequency-chain/api-augment';
 import { Module } from '@nestjs/common';
-import { BullModule } from '@nestjs/bullmq';
 import { ScheduleModule } from '@nestjs/schedule';
 import { EventEmitter2, EventEmitterModule } from '@nestjs/event-emitter';
 import { RedisModule } from '@liaoliaots/nestjs-redis';
-import { Redis } from 'ioredis';
 import { BlockchainModule } from '#lib/blockchain/blockchain.module';
 import { NonceService } from '#lib/services/nonce.service';
 import { ProviderWebhookService } from '#lib/services/provider-webhook.service';
-import { QueueConstants } from '#lib/utils/queues';
 import { ConfigModule } from '#lib/config/config.module';
 import { ConfigService } from '#lib/config/config.service';
 import { redisEventsToEventEmitter } from '#lib/utils/redis';
+import { QueueModule } from '#lib/queues';
 import { TxnNotifierModule } from './transaction_notifier/notifier.module';
 import { TransactionPublisherModule } from './transaction_publisher/publisher.module';
 
@@ -56,21 +54,7 @@ import { TransactionPublisherModule } from './transaction_publisher/publisher.mo
       },
       true, // isGlobal
     ),
-    BullModule.forRootAsync({
-      imports: [ConfigModule],
-      useFactory: (configService: ConfigService) => ({
-        connection: new Redis(configService.redisUrl.toString(), { maxRetriesPerRequest: null }),
-      }),
-      inject: [ConfigService],
-    }),
-    BullModule.registerQueue({
-      name: QueueConstants.TRANSACTION_PUBLISH_QUEUE,
-      defaultJobOptions: {
-        removeOnComplete: 20,
-        removeOnFail: false,
-        attempts: 1,
-      },
-    }),
+    QueueModule.forRoot({ maxRetriesPerRequest: null }),
     ScheduleModule.forRoot(),
     BlockchainModule,
     TransactionPublisherModule,

services/account/libs/common/src/blockchain/blockchain.module.ts

@@ -1,14 +1,10 @@
-/*
-https://docs.nestjs.com/modules
-*/
 import '@frequency-chain/api-augment';
 
 import { Module } from '@nestjs/common';
-import { ConfigModule } from '#lib/config/config.module';
 import { BlockchainService } from './blockchain.service';
 
 @Module({
-  imports: [ConfigModule],
+  imports: [],
   controllers: [],
   providers: [BlockchainService],
   exports: [BlockchainService],

services/account/libs/common/src/config/config.module.ts

@@ -1,8 +1,9 @@
 import { ConfigModule as NestConfigModule } from '@nestjs/config';
-import { Module } from '@nestjs/common';
+import { Global, Module } from '@nestjs/common';
 import { ConfigService } from './config.service';
 import { configModuleOptions } from './env.config';
 
+@Global()
 @Module({
   imports: [NestConfigModule.forRoot(configModuleOptions)],
   controllers: [],

services/account/libs/common/src/config/index.ts

@@ -0,0 +1,2 @@
+export * from './config.module';
+export * from './config.service';

services/account/libs/common/src/index.ts

@@ -6,7 +6,7 @@ export * from './utils/base-consumer';
 export * from './types/enums';
 export * from './types/dtos/transaction.request.dto';
 export * from './services/nonce.service';
-export * from './utils/queues';
+export * from './queues/queue.constants';
 export * from './services/provider-webhook.service';
 export * from './services/enqueue-request.service';
 export * from './constants';

services/account/libs/common/src/queues/index.ts

@@ -0,0 +1,2 @@
+export * as QueueConstants from './queue.constants';
+export * from './queues.module';

services/account/libs/common/src/queues/queue.constants.ts

@@ -0,0 +1,20 @@
+/**
+ * Name of the queue that publishes account transactions to Frequency blockchain
+ */
+export const TRANSACTION_PUBLISH_QUEUE = 'transactionPublish';
+
+/**
+ * Prefix for Redis keys that store webhook endpoints
+ */
+export const REDIS_WATCHER_PREFIX = 'account-service-watcher';
+
+/**
+ * Debouncer cache key
+ */
+export const DEBOUNCER_CACHE_KEY = 'account-service-debouncer';
+
+/**
+ * Last processed msaId key for Redis
+ * @type {string}
+ */
+export const LAST_PROCESSED_DSNP_ID_KEY: string = 'lastProcessedMsaId';

services/account/libs/common/src/queues/queues.module.ts

@@ -0,0 +1,38 @@
+import { ConfigModule } from '#lib/config/config.module';
+import { ConfigService } from '#lib/config/config.service';
+import { BullModule } from '@nestjs/bullmq';
+import { DynamicModule, Global, Module } from '@nestjs/common';
+import Redis, { RedisOptions } from 'ioredis';
+import * as QueueConstants from './queue.constants';
+
+@Global()
+@Module({
+  imports: [
+    BullModule.registerQueue({
+      name: QueueConstants.TRANSACTION_PUBLISH_QUEUE,
+      defaultJobOptions: {
+        removeOnComplete: 20,
+        removeOnFail: false,
+        attempts: 1,
+      },
+    }),
+  ],
+})
+export class QueueModule {
+  static forRoot(redisOptions?: RedisOptions): DynamicModule {
+    return {
+      module: QueueModule,
+      global: true,
+      imports: [
+        BullModule.forRootAsync({
+          imports: [ConfigModule],
+          useFactory: (configService: ConfigService) => ({
+            connection: new Redis(configService.redisUrl.toString(), redisOptions || {}),
+          }),
+          inject: [ConfigService],
+        }),
+      ],
+      exports: [BullModule],
+    };
+  }
+}

services/account/libs/common/src/services/enqueue-request.service.ts

@@ -3,10 +3,9 @@ import { InjectQueue } from '@nestjs/bullmq';
 import { Injectable, Logger } from '@nestjs/common';
 import { Queue } from 'bullmq';
 import { createHash } from 'crypto';
-import { ConfigService } from '../config/config.service';
-import { QueueConstants } from '../utils/queues';
-import { TransactionData } from '../types/dtos/transaction.request.dto';
-import { TransactionResponse } from '../types/dtos/transaction.response.dto';
+import { QueueConstants } from '#lib/queues';
+import { ConfigService } from '#lib/config/config.service';
+import { TransactionResponse, TransactionData } from '#lib/types/dtos';
 
 @Injectable()
 export class EnqueueService {

services/account/libs/common/src/types/dtos/index.ts

@@ -0,0 +1,10 @@
+export * from './accounts.response.dto';
+export * from './delegation.response.dto';
+export * from './handles.request.dto';
+export * from './keys.request.dto';
+export * from './keys.response.dto';
+export * from './transaction.request.dto';
+export * from './transaction.response.dto';
+export * from './wallet.login.config.response.dto';
+export * from './wallet.login.request.dto';
+export * from './wallet.login.response.dto';

services/account/package.json

@@ -154,6 +154,7 @@
       "^#lib/blockchain(|/.*)$": "<rootDir>/libs/common/src/blockchain/$1",
       "^#lib/config(|/.*)$": "<rootDir>/libs/common/src/config/$1",
       "^#lib/interfaces(|/.*)$": "<rootDir>/libs/common/src/interfaces/$1",
+      "^#lib/queues(|/.*)$": "<rootDir>/libs/common/src/queues/$1",
       "^#lib/services(|/.*)$": "<rootDir>/libs/common/src/services/$1",
       "^#lib/types(|/.*)$": "<rootDir>/libs/common/src/types/$1",
       "^#lib/utils(|/.*)$": "<rootDir>/libs/common/src/utils/$1"