feat: support selective queue consumer loading via BACKEND_QUEUE_CONSUMER env var

Author: boris-w

apps/nestjs-backend/src/features/attachments/attachments-crop.module.ts

@@ -1,13 +1,16 @@
 import { Module } from '@nestjs/common';
 import { EventJobModule } from '../../event-emitter/event-job/event-job.module';
-import { conditionalQueueProcessorProviders } from '../../utils/queue';
+import { conditionalQueueProcessorProviders, QueueConsumerType } from '../../utils/queue';
 import { ATTACHMENTS_CROP_QUEUE, AttachmentsCropJob } from './attachments-crop.job';
 import { AttachmentsCropQueueProcessor } from './attachments-crop.processor';
 import { AttachmentsStorageModule } from './attachments-storage.module';
 
 @Module({
   providers: [
-    ...conditionalQueueProcessorProviders(AttachmentsCropQueueProcessor),
+    ...conditionalQueueProcessorProviders({
+      consumer: QueueConsumerType.ImageCrop,
+      providers: [AttachmentsCropQueueProcessor],
+    }),
     AttachmentsCropJob,
   ],
   imports: [EventJobModule.registerQueue(ATTACHMENTS_CROP_QUEUE), AttachmentsStorageModule],

apps/nestjs-backend/src/features/base/base-import-processor/base-import-attachments-csv.module.ts

@@ -1,6 +1,6 @@
 import { Module } from '@nestjs/common';
 import { EventJobModule } from '../../../event-emitter/event-job/event-job.module';
-import { conditionalQueueProcessorProviders } from '../../../utils/queue';
+import { conditionalQueueProcessorProviders, QueueConsumerType } from '../../../utils/queue';
 import { StorageModule } from '../../attachments/plugins/storage.module';
 import {
   BASE_IMPORT_ATTACHMENTS_CSV_QUEUE,
@@ -9,7 +9,10 @@ import {
 import { BaseImportAttachmentsCsvQueueProcessor } from './base-import-attachments-csv.processor';
 @Module({
   providers: [
-    ...conditionalQueueProcessorProviders(BaseImportAttachmentsCsvQueueProcessor),
+    ...conditionalQueueProcessorProviders({
+      consumer: QueueConsumerType.ImportExport,
+      providers: [BaseImportAttachmentsCsvQueueProcessor],
+    }),
     BaseImportAttachmentsCsvJob,
   ],
   imports: [EventJobModule.registerQueue(BASE_IMPORT_ATTACHMENTS_CSV_QUEUE), StorageModule],

apps/nestjs-backend/src/features/base/base-import-processor/base-import-attachments.module.ts

@@ -1,6 +1,6 @@
 import { Module } from '@nestjs/common';
 import { EventJobModule } from '../../../event-emitter/event-job/event-job.module';
-import { conditionalQueueProcessorProviders } from '../../../utils/queue';
+import { conditionalQueueProcessorProviders, QueueConsumerType } from '../../../utils/queue';
 import { StorageModule } from '../../attachments/plugins/storage.module';
 import { BASE_IMPORT_ATTACHMENTS_CSV_QUEUE } from './base-import-attachments-csv.job';
 import { BaseImportAttachmentsCsvModule } from './base-import-attachments-csv.module';
@@ -11,7 +11,10 @@ import {
 import { BaseImportAttachmentsQueueProcessor } from './base-import-attachments.processor';
 @Module({
   providers: [
-    ...conditionalQueueProcessorProviders(BaseImportAttachmentsQueueProcessor),
+    ...conditionalQueueProcessorProviders({
+      consumer: QueueConsumerType.ImportExport,
+      providers: [BaseImportAttachmentsQueueProcessor],
+    }),
     BaseImportAttachmentsJob,
   ],
   imports: [

apps/nestjs-backend/src/features/base/base-import-processor/base-import-csv.module.ts

@@ -1,14 +1,20 @@
 import { Module } from '@nestjs/common';
 import { EventJobModule } from '../../../event-emitter/event-job/event-job.module';
-import { conditionalQueueProcessorProviders } from '../../../utils/queue';
+import { conditionalQueueProcessorProviders, QueueConsumerType } from '../../../utils/queue';
 import { StorageModule } from '../../attachments/plugins/storage.module';
 import { BASE_IMPORT_ATTACHMENTS_CSV_QUEUE } from './base-import-attachments-csv.job';
 import { BASE_IMPORT_CSV_QUEUE, BaseImportCsvJob } from './base-import-csv.job';
 import { BaseImportCsvQueueProcessor } from './base-import-csv.processor';
 import { BaseImportJunctionCsvModule } from './base-import-junction-csv.module';
 
 @Module({
-  providers: [BaseImportCsvJob, ...conditionalQueueProcessorProviders(BaseImportCsvQueueProcessor)],
+  providers: [
+    BaseImportCsvJob,
+    ...conditionalQueueProcessorProviders({
+      consumer: QueueConsumerType.ImportExport,
+      providers: [BaseImportCsvQueueProcessor],
+    }),
+  ],
   imports: [
     EventJobModule.registerQueue(BASE_IMPORT_CSV_QUEUE),
     EventJobModule.registerQueue(BASE_IMPORT_ATTACHMENTS_CSV_QUEUE),

apps/nestjs-backend/src/features/base/base-import-processor/base-import-junction-csv.module.ts

@@ -1,6 +1,6 @@
 import { Module } from '@nestjs/common';
 import { EventJobModule } from '../../../event-emitter/event-job/event-job.module';
-import { conditionalQueueProcessorProviders } from '../../../utils/queue';
+import { conditionalQueueProcessorProviders, QueueConsumerType } from '../../../utils/queue';
 import { StorageModule } from '../../attachments/plugins/storage.module';
 import {
   BASE_IMPORT_JUNCTION_CSV_QUEUE,
@@ -10,7 +10,10 @@ import { BaseImportJunctionCsvQueueProcessor } from './base-import-junction.proc
 
 @Module({
   providers: [
-    ...conditionalQueueProcessorProviders(BaseImportJunctionCsvQueueProcessor),
+    ...conditionalQueueProcessorProviders({
+      consumer: QueueConsumerType.ImportExport,
+      providers: [BaseImportJunctionCsvQueueProcessor],
+    }),
     BaseImportJunctionCsvJob,
   ],
   imports: [EventJobModule.registerQueue(BASE_IMPORT_JUNCTION_CSV_QUEUE), StorageModule],

apps/nestjs-backend/src/features/import/open-api/import-csv-chunk.module.ts

@@ -2,7 +2,7 @@ import { Module } from '@nestjs/common';
 import { EventEmitterModule } from '@nestjs/event-emitter';
 import { EventJobModule } from '../../../event-emitter/event-job/event-job.module';
 import { ShareDbModule } from '../../../share-db/share-db.module';
-import { conditionalQueueProcessorProviders } from '../../../utils/queue';
+import { conditionalQueueProcessorProviders, QueueConsumerType } from '../../../utils/queue';
 import { StorageModule } from '../../attachments/plugins/storage.module';
 import { NotificationModule } from '../../notification/notification.module';
 import { RecordOpenApiModule } from '../../record/open-api/record-open-api.module';
@@ -12,7 +12,10 @@ import { ImportCsvModule } from './import-csv.module';
 
 @Module({
   providers: [
-    ...conditionalQueueProcessorProviders(ImportTableCsvChunkQueueProcessor),
+    ...conditionalQueueProcessorProviders({
+      consumer: QueueConsumerType.ImportExport,
+      providers: [ImportTableCsvChunkQueueProcessor],
+    }),
     ImportTableCsvChunkJob,
   ],
   imports: [

apps/nestjs-backend/src/features/import/open-api/import-csv.module.ts

@@ -2,7 +2,7 @@ import { Module } from '@nestjs/common';
 import { EventEmitterModule } from '@nestjs/event-emitter';
 import { EventJobModule } from '../../../event-emitter/event-job/event-job.module';
 import { ShareDbModule } from '../../../share-db/share-db.module';
-import { conditionalQueueProcessorProviders } from '../../../utils/queue';
+import { conditionalQueueProcessorProviders, QueueConsumerType } from '../../../utils/queue';
 import { StorageModule } from '../../attachments/plugins/storage.module';
 import { NotificationModule } from '../../notification/notification.module';
 import { RecordOpenApiModule } from '../../record/open-api/record-open-api.module';
@@ -11,7 +11,10 @@ import { ImportTableCsvQueueProcessor } from './import-csv.processor';
 
 @Module({
   providers: [
-    ...conditionalQueueProcessorProviders(ImportTableCsvQueueProcessor),
+    ...conditionalQueueProcessorProviders({
+      consumer: QueueConsumerType.ImportExport,
+      providers: [ImportTableCsvQueueProcessor],
+    }),
     ImportTableCsvJob,
   ],
   imports: [

apps/nestjs-backend/src/features/mail-sender/open-api/mail-sender.merge.module.ts

@@ -12,7 +12,12 @@ import { MailSenderMergeProcessor } from './mail-sender.merge.processor';
     EventJobModule.registerQueue(MAIL_SENDER_QUEUE),
     SettingOpenApiModule,
   ],
-  providers: [...conditionalQueueProcessorProviders(MailSenderMergeProcessor), MailSenderMergeJob],
+  providers: [
+    ...conditionalQueueProcessorProviders({
+      providers: [MailSenderMergeProcessor],
+    }),
+    MailSenderMergeJob,
+  ],
   exports: [MailSenderMergeJob],
 })
 export class MailSenderMergeModule {}

apps/nestjs-backend/src/utils/queue.ts

@@ -1,3 +1,49 @@
-export function conditionalQueueProcessorProviders<T>(...providers: T[]): T[] {
-  return process.env.BACKEND_DISABLE_QUEUE_CONSUMER === 'true' ? [] : providers;
+import { Logger } from '@nestjs/common';
+
+export enum QueueConsumerType {
+  Automation = 'automation',
+  ImportExport = 'import-export',
+  ImageCrop = 'image-crop',
+  Default = 'default',
+}
+
+export function conditionalQueueProcessorProviders<T>(
+  ...opts: {
+    consumer?: QueueConsumerType;
+    providers: T[];
+  }[]
+): T[] {
+  if (process.env.BACKEND_DISABLE_QUEUE_CONSUMER === 'true') {
+    return [];
+  }
+  const selectedConsumer = (process.env.BACKEND_QUEUE_CONSUMER?.split(',')?.filter((v) =>
+    Object.values(QueueConsumerType).includes(v as QueueConsumerType)
+  ) || []) as QueueConsumerType[];
+
+  // If selected consumer is provided, return providers for the selected consumer
+  if (selectedConsumer.length > 0) {
+    const providers: T[] = [];
+    for (const opt of opts) {
+      const consumer = opt.consumer || QueueConsumerType.Default;
+      if (selectedConsumer.includes(consumer)) {
+        providers.push(...opt.providers);
+      }
+    }
+    providers.length > 0 &&
+      Logger.log(
+        `Queue Consumer Providers (${selectedConsumer.join(', ')}): ${providers.map((p) => (typeof p === 'function' ? p.name : 'unknown')).join(', ')}`
+      );
+    return providers;
+  }
+
+  // If no selected consumer is provided, return providers for all consumers
+  const providers: T[] = [];
+  for (const opt of opts) {
+    providers.push(...opt.providers);
+  }
+  providers.length > 0 &&
+    Logger.log(
+      `Queue Consumer Providers (ALL): ${providers.map((p) => (typeof p === 'function' ? p.name : 'unknown')).join(', ')}`
+    );
+  return providers;
 }