refactor(server): bull jobs (#2569)

* refactor(server): bull jobs

* chore: add comment

* chore: metadata test coverage

* fix typo

---------

Co-authored-by: Alex Tran <alex.tran1502@gmail.com>

server/apps/microservices/src/main.ts

@@ -1,9 +1,9 @@
+import { getLogLevels, SERVER_VERSION } from '@app/domain';
+import { RedisIoAdapter } from '@app/infra';
 import { Logger } from '@nestjs/common';
 import { NestFactory } from '@nestjs/core';
-import { SERVER_VERSION } from '@app/domain';
-import { getLogLevels } from '@app/domain';
-import { RedisIoAdapter } from '@app/infra';
 import { MicroservicesModule } from './microservices.module';
+import { ProcessorService } from './processor.service';
 import { MetadataExtractionProcessor } from './processors/metadata-extraction.processor';
 
 const logger = new Logger('ImmichMicroservice');
@@ -15,6 +15,8 @@ async function bootstrap() {
 
   const listeningPort = Number(process.env.MICROSERVICES_PORT) || 3002;
 
+  await app.get(ProcessorService).init();
+
   app.useWebSocketAdapter(new RedisIoAdapter(app));
 
   const metadataService = app.get(MetadataExtractionProcessor);

server/apps/microservices/src/microservices.module.ts

@@ -3,17 +3,8 @@ import { InfraModule } from '@app/infra';
 import { ExifEntity } from '@app/infra/entities';
 import { Module } from '@nestjs/common';
 import { TypeOrmModule } from '@nestjs/typeorm';
-import {
-  BackgroundTaskProcessor,
-  ClipEncodingProcessor,
-  FacialRecognitionProcessor,
-  ObjectTaggingProcessor,
-  SearchIndexProcessor,
-  StorageTemplateMigrationProcessor,
-  ThumbnailGeneratorProcessor,
-  VideoTranscodeProcessor,
-} from './processors';
-import { MetadataExtractionProcessor, SidecarProcessor } from './processors/metadata-extraction.processor';
+import { ProcessorService } from './processor.service';
+import { MetadataExtractionProcessor } from './processors/metadata-extraction.processor';
 
 @Module({
   imports: [
@@ -21,17 +12,6 @@ import { MetadataExtractionProcessor, SidecarProcessor } from './processors/meta
     DomainModule.register({ imports: [InfraModule] }),
     TypeOrmModule.forFeature([ExifEntity]),
   ],
-  providers: [
-    ThumbnailGeneratorProcessor,
-    MetadataExtractionProcessor,
-    VideoTranscodeProcessor,
-    ObjectTaggingProcessor,
-    ClipEncodingProcessor,
-    StorageTemplateMigrationProcessor,
-    BackgroundTaskProcessor,
-    SearchIndexProcessor,
-    FacialRecognitionProcessor,
-    SidecarProcessor,
-  ],
+  providers: [MetadataExtractionProcessor, ProcessorService],
 })
 export class MicroservicesModule {}

server/apps/microservices/src/processor.service.ts

@@ -0,0 +1,105 @@
+import {
+  AssetService,
+  FacialRecognitionService,
+  JobName,
+  JOBS_TO_QUEUE,
+  MediaService,
+  MetadataService,
+  PersonService,
+  QueueName,
+  QUEUE_TO_CONCURRENCY,
+  SearchService,
+  SmartInfoService,
+  StorageService,
+  StorageTemplateService,
+  SystemConfigService,
+  UserService,
+} from '@app/domain';
+import { getQueueToken } from '@nestjs/bull';
+import { Injectable } from '@nestjs/common';
+import { ModuleRef } from '@nestjs/core';
+import { Queue } from 'bull';
+import { MetadataExtractionProcessor } from './processors/metadata-extraction.processor';
+
+type JobHandler<T = any> = (data: T) => void | Promise<void>;
+
+@Injectable()
+export class ProcessorService {
+  constructor(
+    private moduleRef: ModuleRef,
+    // TODO refactor to domain
+    private metadataProcessor: MetadataExtractionProcessor,
+
+    private assetService: AssetService,
+    private facialRecognitionService: FacialRecognitionService,
+    private mediaService: MediaService,
+    private metadataService: MetadataService,
+    private personService: PersonService,
+    private searchService: SearchService,
+    private smartInfoService: SmartInfoService,
+    private storageTemplateService: StorageTemplateService,
+    private storageService: StorageService,
+    private systemConfigService: SystemConfigService,
+    private userService: UserService,
+  ) {}
+
+  private handlers: Record<JobName, JobHandler> = {
+    [JobName.ASSET_UPLOADED]: (data) => this.assetService.handleAssetUpload(data),
+    [JobName.DELETE_FILES]: (data) => this.storageService.handleDeleteFiles(data),
+    [JobName.USER_DELETE_CHECK]: () => this.userService.handleUserDeleteCheck(),
+    [JobName.USER_DELETION]: (data) => this.userService.handleUserDelete(data),
+    [JobName.QUEUE_OBJECT_TAGGING]: (data) => this.smartInfoService.handleQueueObjectTagging(data),
+    [JobName.DETECT_OBJECTS]: (data) => this.smartInfoService.handleDetectObjects(data),
+    [JobName.CLASSIFY_IMAGE]: (data) => this.smartInfoService.handleClassifyImage(data),
+    [JobName.QUEUE_ENCODE_CLIP]: (data) => this.smartInfoService.handleQueueEncodeClip(data),
+    [JobName.ENCODE_CLIP]: (data) => this.smartInfoService.handleEncodeClip(data),
+    [JobName.SEARCH_INDEX_ALBUMS]: () => this.searchService.handleIndexAlbums(),
+    [JobName.SEARCH_INDEX_ASSETS]: () => this.searchService.handleIndexAssets(),
+    [JobName.SEARCH_INDEX_FACES]: () => this.searchService.handleIndexFaces(),
+    [JobName.SEARCH_INDEX_ALBUM]: (data) => this.searchService.handleIndexAlbum(data),
+    [JobName.SEARCH_INDEX_ASSET]: (data) => this.searchService.handleIndexAsset(data),
+    [JobName.SEARCH_INDEX_FACE]: (data) => this.searchService.handleIndexFace(data),
+    [JobName.SEARCH_REMOVE_ALBUM]: (data) => this.searchService.handleRemoveAlbum(data),
+    [JobName.SEARCH_REMOVE_ASSET]: (data) => this.searchService.handleRemoveAsset(data),
+    [JobName.SEARCH_REMOVE_FACE]: (data) => this.searchService.handleRemoveFace(data),
+    [JobName.STORAGE_TEMPLATE_MIGRATION]: () => this.storageTemplateService.handleMigration(),
+    [JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE]: (data) => this.storageTemplateService.handleMigrationSingle(data),
+    [JobName.SYSTEM_CONFIG_CHANGE]: () => this.systemConfigService.refreshConfig(),
+    [JobName.QUEUE_GENERATE_THUMBNAILS]: (data) => this.mediaService.handleQueueGenerateThumbnails(data),
+    [JobName.GENERATE_JPEG_THUMBNAIL]: (data) => this.mediaService.handleGenerateJpegThumbnail(data),
+    [JobName.GENERATE_WEBP_THUMBNAIL]: (data) => this.mediaService.handleGenerateWepbThumbnail(data),
+    [JobName.QUEUE_VIDEO_CONVERSION]: (data) => this.mediaService.handleQueueVideoConversion(data),
+    [JobName.VIDEO_CONVERSION]: (data) => this.mediaService.handleVideoConversion(data),
+    [JobName.QUEUE_METADATA_EXTRACTION]: (data) => this.metadataProcessor.handleQueueMetadataExtraction(data),
+    [JobName.EXIF_EXTRACTION]: (data) => this.metadataProcessor.extractExifInfo(data),
+    [JobName.EXTRACT_VIDEO_METADATA]: (data) => this.metadataProcessor.extractVideoMetadata(data),
+    [JobName.QUEUE_RECOGNIZE_FACES]: (data) => this.facialRecognitionService.handleQueueRecognizeFaces(data),
+    [JobName.RECOGNIZE_FACES]: (data) => this.facialRecognitionService.handleRecognizeFaces(data),
+    [JobName.GENERATE_FACE_THUMBNAIL]: (data) => this.facialRecognitionService.handleGenerateFaceThumbnail(data),
+    [JobName.PERSON_CLEANUP]: () => this.personService.handlePersonCleanup(),
+    [JobName.QUEUE_SIDECAR]: (data) => this.metadataService.handleQueueSidecar(data),
+    [JobName.SIDECAR_DISCOVERY]: (data) => this.metadataService.handleSidecarDiscovery(data),
+    [JobName.SIDECAR_SYNC]: (data) => this.metadataService.handleSidecarSync(data),
+  };
+
+  async init() {
+    const queueSeen: Partial<Record<QueueName, boolean>> = {};
+
+    for (const jobName of Object.values(JobName)) {
+      const handler = this.handlers[jobName];
+      const queueName = JOBS_TO_QUEUE[jobName];
+      const queue = this.moduleRef.get<Queue>(getQueueToken(queueName), { strict: false });
+
+      // only set concurrency on the first job for a queue, since concurrency stacks
+      const seen = queueSeen[queueName];
+      const concurrency = seen ? 0 : QUEUE_TO_CONCURRENCY[queueName];
+      queueSeen[queueName] = true;
+
+      await queue.isReady();
+
+      queue.process(jobName, concurrency, async (job): Promise<void> => {
+        await handler(job.data);
+      });
+    }
+  }
+}

server/apps/microservices/src/processors.ts

@@ -1,219 +0,0 @@
-import {
-  AssetService,
-  FacialRecognitionService,
-  IAssetFaceJob,
-  IAssetJob,
-  IBaseJob,
-  IBulkEntityJob,
-  IDeleteFilesJob,
-  IFaceThumbnailJob,
-  IUserDeletionJob,
-  JobName,
-  MediaService,
-  PersonService,
-  QueueName,
-  SearchService,
-  SmartInfoService,
-  StorageService,
-  StorageTemplateService,
-  SystemConfigService,
-  UserService,
-} from '@app/domain';
-import { Process, Processor } from '@nestjs/bull';
-import { Job } from 'bull';
-
-@Processor(QueueName.BACKGROUND_TASK)
-export class BackgroundTaskProcessor {
-  constructor(
-    private assetService: AssetService,
-    private personService: PersonService,
-    private storageService: StorageService,
-    private systemConfigService: SystemConfigService,
-    private userService: UserService,
-  ) {}
-
-  @Process(JobName.ASSET_UPLOADED)
-  async onAssetUpload(job: Job<IAssetJob>) {
-    await this.assetService.handleAssetUpload(job.data);
-  }
-
-  @Process(JobName.DELETE_FILES)
-  async onDeleteFile(job: Job<IDeleteFilesJob>) {
-    await this.storageService.handleDeleteFiles(job.data);
-  }
-
-  @Process(JobName.SYSTEM_CONFIG_CHANGE)
-  async onSystemConfigChange() {
-    await this.systemConfigService.refreshConfig();
-  }
-
-  @Process(JobName.USER_DELETE_CHECK)
-  async onUserDeleteCheck() {
-    await this.userService.handleUserDeleteCheck();
-  }
-
-  @Process(JobName.USER_DELETION)
-  async onUserDelete(job: Job<IUserDeletionJob>) {
-    await this.userService.handleUserDelete(job.data);
-  }
-
-  @Process(JobName.PERSON_CLEANUP)
-  async onPersonCleanup() {
-    await this.personService.handlePersonCleanup();
-  }
-}
-
-@Processor(QueueName.OBJECT_TAGGING)
-export class ObjectTaggingProcessor {
-  constructor(private smartInfoService: SmartInfoService) {}
-
-  @Process({ name: JobName.QUEUE_OBJECT_TAGGING, concurrency: 0 })
-  async onQueueObjectTagging(job: Job<IBaseJob>) {
-    await this.smartInfoService.handleQueueObjectTagging(job.data);
-  }
-
-  @Process({ name: JobName.DETECT_OBJECTS, concurrency: 1 })
-  async onDetectObjects(job: Job<IAssetJob>) {
-    await this.smartInfoService.handleDetectObjects(job.data);
-  }
-
-  @Process({ name: JobName.CLASSIFY_IMAGE, concurrency: 1 })
-  async onClassifyImage(job: Job<IAssetJob>) {
-    await this.smartInfoService.handleClassifyImage(job.data);
-  }
-}
-
-@Processor(QueueName.RECOGNIZE_FACES)
-export class FacialRecognitionProcessor {
-  constructor(private facialRecognitionService: FacialRecognitionService) {}
-
-  @Process({ name: JobName.QUEUE_RECOGNIZE_FACES, concurrency: 0 })
-  async onQueueRecognizeFaces(job: Job<IBaseJob>) {
-    await this.facialRecognitionService.handleQueueRecognizeFaces(job.data);
-  }
-
-  @Process({ name: JobName.RECOGNIZE_FACES, concurrency: 1 })
-  async onRecognizeFaces(job: Job<IAssetJob>) {
-    await this.facialRecognitionService.handleRecognizeFaces(job.data);
-  }
-
-  @Process({ name: JobName.GENERATE_FACE_THUMBNAIL, concurrency: 1 })
-  async onGenerateFaceThumbnail(job: Job<IFaceThumbnailJob>) {
-    await this.facialRecognitionService.handleGenerateFaceThumbnail(job.data);
-  }
-}
-
-@Processor(QueueName.CLIP_ENCODING)
-export class ClipEncodingProcessor {
-  constructor(private smartInfoService: SmartInfoService) {}
-
-  @Process({ name: JobName.QUEUE_ENCODE_CLIP, concurrency: 0 })
-  async onQueueClipEncoding(job: Job<IBaseJob>) {
-    await this.smartInfoService.handleQueueEncodeClip(job.data);
-  }
-
-  @Process({ name: JobName.ENCODE_CLIP, concurrency: 1 })
-  async onEncodeClip(job: Job<IAssetJob>) {
-    await this.smartInfoService.handleEncodeClip(job.data);
-  }
-}
-
-@Processor(QueueName.SEARCH)
-export class SearchIndexProcessor {
-  constructor(private searchService: SearchService) {}
-
-  @Process(JobName.SEARCH_INDEX_ALBUMS)
-  async onIndexAlbums() {
-    await this.searchService.handleIndexAlbums();
-  }
-
-  @Process(JobName.SEARCH_INDEX_ASSETS)
-  async onIndexAssets() {
-    await this.searchService.handleIndexAssets();
-  }
-
-  @Process(JobName.SEARCH_INDEX_FACES)
-  async onIndexFaces() {
-    await this.searchService.handleIndexFaces();
-  }
-
-  @Process(JobName.SEARCH_INDEX_ALBUM)
-  onIndexAlbum(job: Job<IBulkEntityJob>) {
-    this.searchService.handleIndexAlbum(job.data);
-  }
-
-  @Process(JobName.SEARCH_INDEX_ASSET)
-  onIndexAsset(job: Job<IBulkEntityJob>) {
-    this.searchService.handleIndexAsset(job.data);
-  }
-
-  @Process(JobName.SEARCH_INDEX_FACE)
-  async onIndexFace(job: Job<IAssetFaceJob>) {
-    await this.searchService.handleIndexFace(job.data);
-  }
-
-  @Process(JobName.SEARCH_REMOVE_ALBUM)
-  onRemoveAlbum(job: Job<IBulkEntityJob>) {
-    this.searchService.handleRemoveAlbum(job.data);
-  }
-
-  @Process(JobName.SEARCH_REMOVE_ASSET)
-  onRemoveAsset(job: Job<IBulkEntityJob>) {
-    this.searchService.handleRemoveAsset(job.data);
-  }
-
-  @Process(JobName.SEARCH_REMOVE_FACE)
-  onRemoveFace(job: Job<IAssetFaceJob>) {
-    this.searchService.handleRemoveFace(job.data);
-  }
-}
-
-@Processor(QueueName.STORAGE_TEMPLATE_MIGRATION)
-export class StorageTemplateMigrationProcessor {
-  constructor(private storageTemplateService: StorageTemplateService) {}
-
-  @Process({ name: JobName.STORAGE_TEMPLATE_MIGRATION })
-  async onTemplateMigration() {
-    await this.storageTemplateService.handleTemplateMigration();
-  }
-
-  @Process({ name: JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE })
-  async onTemplateMigrationSingle(job: Job<IAssetJob>) {
-    await this.storageTemplateService.handleTemplateMigrationSingle(job.data);
-  }
-}
-
-@Processor(QueueName.THUMBNAIL_GENERATION)
-export class ThumbnailGeneratorProcessor {
-  constructor(private mediaService: MediaService) {}
-
-  @Process({ name: JobName.QUEUE_GENERATE_THUMBNAILS, concurrency: 0 })
-  async onQueueGenerateThumbnails(job: Job<IBaseJob>) {
-    await this.mediaService.handleQueueGenerateThumbnails(job.data);
-  }
-
-  @Process({ name: JobName.GENERATE_JPEG_THUMBNAIL, concurrency: 3 })
-  async onGenerateJpegThumbnail(job: Job<IAssetJob>) {
-    await this.mediaService.handleGenerateJpegThumbnail(job.data);
-  }
-
-  @Process({ name: JobName.GENERATE_WEBP_THUMBNAIL, concurrency: 3 })
-  async onGenerateWepbThumbnail(job: Job<IAssetJob>) {
-    await this.mediaService.handleGenerateWepbThumbnail(job.data);
-  }
-}
-
-@Processor(QueueName.VIDEO_CONVERSION)
-export class VideoTranscodeProcessor {
-  constructor(private mediaService: MediaService) {}
-
-  @Process({ name: JobName.QUEUE_VIDEO_CONVERSION, concurrency: 0 })
-  async onQueueVideoConversion(job: Job<IBaseJob>): Promise<void> {
-    await this.mediaService.handleQueueVideoConversion(job.data);
-  }
-
-  @Process({ name: JobName.VIDEO_CONVERSION, concurrency: 1 })
-  async onVideoConversion(job: Job<IAssetJob>) {
-    await this.mediaService.handleVideoConversion(job.data);
-  }
-}

server/apps/microservices/src/processors/metadata-extraction.processor.ts

@@ -10,15 +10,12 @@ import {
   QueueName,
   usePagination,
   WithoutProperty,
-  WithProperty,
 } from '@app/domain';
 import { AssetEntity, AssetType, ExifEntity } from '@app/infra/entities';
-import { Process, Processor } from '@nestjs/bull';
 import { Inject, Logger } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { InjectRepository } from '@nestjs/typeorm';
 import tz_lookup from '@photostructure/tz-lookup';
-import { Job } from 'bull';
 import { ExifDateTime, exiftool, Tags } from 'exiftool-vendored';
 import ffmpeg, { FfprobeData } from 'fluent-ffmpeg';
 import { Duration } from 'luxon';
@@ -33,7 +30,6 @@ interface ImmichTags extends Tags {
   ContentIdentifier?: string;
 }
 
-@Processor(QueueName.METADATA_EXTRACTION)
 export class MetadataExtractionProcessor {
   private logger = new Logger(MetadataExtractionProcessor.name);
   private assetCore: AssetCore;
@@ -73,10 +69,9 @@ export class MetadataExtractionProcessor {
     }
   }
 
-  @Process(JobName.QUEUE_METADATA_EXTRACTION)
-  async handleQueueMetadataExtraction(job: Job<IBaseJob>) {
+  async handleQueueMetadataExtraction(job: IBaseJob) {
     try {
-      const { force } = job.data;
+      const { force } = job;
       const assetPagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (pagination) => {
         return force
           ? this.assetRepository.getAll(pagination)
@@ -94,9 +89,8 @@ export class MetadataExtractionProcessor {
     }
   }
 
-  @Process(JobName.EXIF_EXTRACTION)
-  async extractExifInfo(job: Job<IAssetJob>) {
-    let asset = job.data.asset;
+  async extractExifInfo(job: IAssetJob) {
+    let asset = job.asset;
 
     try {
       const mediaExifData = await exiftool.read<ImmichTags>(asset.originalPath).catch((error: any) => {
@@ -223,9 +217,8 @@ export class MetadataExtractionProcessor {
     }
   }
 
-  @Process({ name: JobName.EXTRACT_VIDEO_METADATA, concurrency: 2 })
-  async extractVideoMetadata(job: Job<IAssetJob>) {
-    let asset = job.data.asset;
+  async extractVideoMetadata(job: IAssetJob) {
+    let asset = job.asset;
 
     if (!asset.isVisible) {
       return;
@@ -370,83 +363,3 @@ export class MetadataExtractionProcessor {
     return Duration.fromObject({ seconds: videoDurationInSecond }).toFormat('hh:mm:ss.SSS');
   }
 }
-
-@Processor(QueueName.SIDECAR)
-export class SidecarProcessor {
-  private logger = new Logger(SidecarProcessor.name);
-  private assetCore: AssetCore;
-
-  constructor(
-    @Inject(IAssetRepository) private assetRepository: IAssetRepository,
-    @Inject(IJobRepository) private jobRepository: IJobRepository,
-  ) {
-    this.assetCore = new AssetCore(assetRepository, jobRepository);
-  }
-
-  @Process(JobName.QUEUE_SIDECAR)
-  async handleQueueSidecar(job: Job<IBaseJob>) {
-    try {
-      const { force } = job.data;
-      const assetPagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (pagination) => {
-        return force
-          ? this.assetRepository.getWith(pagination, WithProperty.SIDECAR)
-          : this.assetRepository.getWithout(pagination, WithoutProperty.SIDECAR);
-      });
-
-      for await (const assets of assetPagination) {
-        for (const asset of assets) {
-          const name = force ? JobName.SIDECAR_SYNC : JobName.SIDECAR_DISCOVERY;
-          await this.jobRepository.queue({ name, data: { asset } });
-        }
-      }
-    } catch (error: any) {
-      this.logger.error(`Unable to queue sidecar scanning`, error?.stack);
-    }
-  }
-
-  @Process(JobName.SIDECAR_SYNC)
-  async handleSidecarSync(job: Job<IAssetJob>) {
-    const { asset } = job.data;
-    if (!asset.isVisible) {
-      return;
-    }
-
-    try {
-      const name = asset.type === AssetType.VIDEO ? JobName.EXTRACT_VIDEO_METADATA : JobName.EXIF_EXTRACTION;
-      await this.jobRepository.queue({ name, data: { asset } });
-    } catch (error: any) {
-      this.logger.error(`Unable to queue metadata extraction`, error?.stack);
-    }
-  }
-
-  @Process(JobName.SIDECAR_DISCOVERY)
-  async handleSidecarDiscovery(job: Job<IAssetJob>) {
-    let { asset } = job.data;
-    if (!asset.isVisible) {
-      return;
-    }
-
-    if (asset.sidecarPath) {
-      return;
-    }
-
-    try {
-      await fs.promises.access(`${asset.originalPath}.xmp`, fs.constants.W_OK);
-
-      try {
-        asset = await this.assetCore.save({ id: asset.id, sidecarPath: `${asset.originalPath}.xmp` });
-        // TODO: optimize to only queue assets with recent xmp changes
-        const name = asset.type === AssetType.VIDEO ? JobName.EXTRACT_VIDEO_METADATA : JobName.EXIF_EXTRACTION;
-        await this.jobRepository.queue({ name, data: { asset } });
-      } catch (error: any) {
-        this.logger.error(`Unable to sync sidecar`, error?.stack);
-      }
-    } catch (error: any) {
-      if (error.code == 'EACCES') {
-        this.logger.error(`Unable to queue metadata extraction, file is not writable`, error?.stack);
-      }
-
-      return;
-    }
-  }
-}