working addJobs

We've been keen to try this for a while as it means we can remove redis as a
dependency, which makes Immich easier to setup and run.

This replaces bullmq with a bespoke postgres queue. Jobs in the queue are
processed either immediately via triggers and notifications, or eventually if a
notification is missed.

docker/docker-compose.dev.yml

@@ -33,7 +33,6 @@ services:
       - ${UPLOAD_LOCATION}/photos/upload:/usr/src/app/upload/upload
       - /usr/src/app/node_modules
       - /etc/localtime:/etc/localtime:ro
-      - ../flickr30k-images:/flickr30k:ro
     env_file:
       - .env
     environment:
@@ -148,25 +147,25 @@ services:
       -c wal_compression=on
 
   # set IMMICH_TELEMETRY_INCLUDE=all in .env to enable metrics
-  immich-prometheus:
+  # immich-prometheus:
-    container_name: immich_prometheus
+  #   container_name: immich_prometheus
-    ports:
+  #   ports:
-      - 9090:9090
+  #     - 9090:9090
-    image: prom/prometheus
+  #   image: prom/prometheus
-    volumes:
+  #   volumes:
-      - ./prometheus.yml:/etc/prometheus/prometheus.yml
+  #     - ./prometheus.yml:/etc/prometheus/prometheus.yml
-      - prometheus-data:/prometheus
+  #     - prometheus-data:/prometheus
 
   # first login uses admin/admin
   # add data source for http://immich-prometheus:9090 to get started
-  immich-grafana:
+  # immich-grafana:
-    container_name: immich_grafana
+  #   container_name: immich_grafana
-    command: ['./run.sh', '-disable-reporting']
+  #   command: ['./run.sh', '-disable-reporting']
-    ports:
+  #   ports:
-      - 3001:3000
+  #     - 3000:3000
-    image: grafana/grafana:10.3.3-ubuntu
+  #   image: grafana/grafana:10.3.3-ubuntu
-    volumes:
+  #   volumes:
-      - grafana-data:/var/lib/grafana
+  #     - grafana-data:/var/lib/grafana
 
 volumes:
   model-cache:

docker/prometheus.yml

@@ -1,14 +1,12 @@
 global:
-  scrape_interval: 3s
+  scrape_interval:     15s
-  evaluation_interval: 3s
+  evaluation_interval: 15s
 
 scrape_configs:
   - job_name: immich_api
-    scrape_interval: 3s
     static_configs:
-      - targets: ["immich-server:8081"]
+      - targets: ['immich-server:8081']
 
   - job_name: immich_microservices
-    scrape_interval:
     static_configs:
-      - targets: ["immich-server:8082"]
+      - targets: ['immich-server:8082']

server/package-lock.json

@@ -38,7 +38,7 @@
         "fast-glob": "^3.3.2",
         "fluent-ffmpeg": "^2.1.2",
         "geo-tz": "^8.0.0",
-        "graphile-worker": "^0.16.6",
+        "graphile-worker": "^0.17.0-canary.1fcb2a0",
         "handlebars": "^4.7.8",
         "i18n-iso-countries": "^7.6.0",
         "joi": "^17.10.0",
@@ -10085,16 +10085,16 @@
       "license": "MIT"
     },
     "node_modules/graphile-worker": {
-      "version": "0.16.6",
+      "version": "0.17.0-canary.1fcb2a0",
-      "resolved": "https://registry.npmjs.org/graphile-worker/-/graphile-worker-0.16.6.tgz",
+      "resolved": "https://registry.npmjs.org/graphile-worker/-/graphile-worker-0.17.0-canary.1fcb2a0.tgz",
-      "integrity": "sha512-e7gGYDmGqzju2l83MpzX8vNG/lOtVJiSzI3eZpAFubSxh/cxs7sRrRGBGjzBP1kNG0H+c95etPpNRNlH65PYhw==",
+      "integrity": "sha512-eG02GZ0U1eSMBdfHlQg9+jaNXpr9gs1cwqfFeney3BHpEMSvG3jw+7SdQJPVUgF8wnt8dRRfhkbpzaXGSOr+MQ==",
       "license": "MIT",
       "dependencies": {
         "@graphile/logger": "^0.2.0",
         "@types/debug": "^4.1.10",
         "@types/pg": "^8.10.5",
         "cosmiconfig": "^8.3.6",
-        "graphile-config": "^0.0.1-beta.4",
+        "graphile-config": "^0.0.1-beta.14",
         "json5": "^2.2.3",
         "pg": "^8.11.3",
         "tslib": "^2.6.2",
@@ -10104,7 +10104,8 @@
         "graphile-worker": "dist/cli.js"
       },
       "engines": {
-        "node": ">=14.0.0"
+        "node": ">=14.0.0",
+        "yarn": "^1.22.22"
       }
     },
     "node_modules/handlebars": {

server/package.json

@@ -63,7 +63,7 @@
     "fast-glob": "^3.3.2",
     "fluent-ffmpeg": "^2.1.2",
     "geo-tz": "^8.0.0",
-    "graphile-worker": "^0.16.6",
+    "graphile-worker": "^0.17.0-canary.1fcb2a0",
     "handlebars": "^4.7.8",
     "i18n-iso-countries": "^7.6.0",
     "joi": "^17.10.0",

server/src/repositories/database.repository.ts

@@ -255,10 +255,10 @@ export class DatabaseRepository {
       }
     }
 
-    if (error) {
+    // if (error) {
-      this.logger.error(`Kysely migrations failed: ${error}`);
+    //   this.logger.error(`Kysely migrations failed: ${error}`);
-      throw error;
+    //   throw error;
-    }
+    // }
 
     this.logger.debug('Finished running kysely migrations');
   }

server/src/repositories/job.repository.ts

@@ -1,9 +1,8 @@
 import { Injectable } from '@nestjs/common';
 import { ModuleRef, Reflector } from '@nestjs/core';
 import { ClassConstructor } from 'class-transformer';
-import { makeWorkerUtils, run, Runner, TaskSpec, WorkerUtils } from 'graphile-worker';
+import { AddJobsJobSpec, makeWorkerUtils, run, Runner, WorkerUtils } from 'graphile-worker';
 import { Kysely } from 'kysely';
-import { DateTime, Duration } from 'luxon';
 import { InjectKysely } from 'nestjs-kysely';
 import pg, { PoolConfig } from 'pg';
 import { DB } from 'src/db';
@@ -101,8 +100,6 @@ export class JobRepository {
         waiting: ${pool.waitingCount}`);
     }, 5000);
 
-    pool.setMaxListeners(100);
-
     pool.on('connect', (client) => {
       client.setMaxListeners(200);
     });
@@ -131,7 +128,7 @@ export class JobRepository {
       concurrency,
       taskList: {
         [queueName]: async (payload: unknown): Promise<void> => {
-          this.logger.log(`Job ${queueName} started with payload: ${JSON.stringify(payload)}`);
+          // this.logger.log(`Job ${queueName} started with payload: ${JSON.stringify(payload)}`);
           await this.eventRepository.emit('job.start', queueName, payload as JobItem);
         },
       },
@@ -180,21 +177,21 @@ export class JobRepository {
     return (this.handlers[name] as JobMapItem).queueName;
   }
 
-  async run({ name, data }: JobItem): Promise<JobStatus> {
+  run({ name, data }: JobItem): Promise<JobStatus> {
     const item = this.handlers[name as JobName];
     if (!item) {
       this.logger.warn(`Skipping unknown job: "${name}"`);
-      return JobStatus.SKIPPED;
+      return Promise.resolve(JobStatus.SKIPPED);
     }
     return item.handler(data);
   }
 
-  async queue(item: JobItem): Promise<void> {
+  queue(item: JobItem): Promise<unknown> {
-    await this.workerUtils!.addJob(this.getQueueName(item.name), item, this.getJobOptions(item));
+    return this.queueAll([item]);
   }
 
-  async queueAll(items: JobItem[]): Promise<void> {
+  queueAll(items: JobItem[]): Promise<unknown> {
-    await Promise.all(items.map((item) => this.queue(item)));
+    return this.workerUtils!.addJobs(items.map((item) => this.getJobSpec(item)));
   }
 
   // todo: are we actually generating sql
@@ -279,23 +276,28 @@ export class JobRepository {
     return { paused: state?.[queueName]?.paused ?? false };
   }
 
-  private getJobOptions(item: JobItem): TaskSpec | undefined {
+  private getJobSpec(item: JobItem): AddJobsJobSpec {
+    const identifier = (this.handlers[item.name] as JobMapItem).queueName;
     switch (item.name) {
       case JobName.NOTIFY_ALBUM_UPDATE: {
-        let runAt: Date | undefined;
+        return {
-        if (item.data?.delay) {
+          identifier,
-          runAt = DateTime.now().plus(Duration.fromMillis(item.data.delay)).toJSDate();
+          payload: item,
-        }
+          jobKey: item.data.id,
-        return { jobKey: item.data.id, runAt };
+          runAt: item.data?.delay ? new Date(Date.now() + item.data.delay) : undefined,
+        };
       }
       case JobName.STORAGE_TEMPLATE_MIGRATION_SINGLE: {
-        return { jobKey: QueueName.STORAGE_TEMPLATE_MIGRATION };
+        return { identifier, payload: item, jobKey: QueueName.STORAGE_TEMPLATE_MIGRATION };
       }
       case JobName.GENERATE_PERSON_THUMBNAIL: {
-        return { priority: 1 };
+        return { identifier, payload: item, priority: 1 };
       }
       case JobName.QUEUE_FACIAL_RECOGNITION: {
-        return { jobKey: JobName.QUEUE_FACIAL_RECOGNITION };
+        return { identifier, payload: item, jobKey: JobName.QUEUE_FACIAL_RECOGNITION };
+      }
+      default: {
+        return { identifier, payload: item };
       }
     }
   }

server/src/services/duplicate.service.ts

@@ -4,11 +4,10 @@ import { OnJob } from 'src/decorators';
 import { mapAsset } from 'src/dtos/asset-response.dto';
 import { AuthDto } from 'src/dtos/auth.dto';
 import { DuplicateResponseDto } from 'src/dtos/duplicate.dto';
-import { AssetFileType, JobName, JobStatus, QueueName } from 'src/enum';
+import { JobName, JobStatus, QueueName } from 'src/enum';
 import { AssetDuplicateResult } from 'src/repositories/search.repository';
 import { BaseService } from 'src/services/base.service';
 import { JobItem, JobOf } from 'src/types';
-import { getAssetFile } from 'src/utils/asset.util';
 import { isDuplicateDetectionEnabled } from 'src/utils/misc';
 
 @Injectable()
@@ -60,49 +59,6 @@ export class DuplicateService extends BaseService {
       return JobStatus.FAILED;
     }
 
-    if (asset.stackId) {
-      this.logger.debug(`Asset ${id} is part of a stack, skipping`);
-      return JobStatus.SKIPPED;
-    }
-
-    if (!asset.isVisible) {
-      this.logger.debug(`Asset ${id} is not visible, skipping`);
-      return JobStatus.SKIPPED;
-    }
-
-    const previewFile = getAssetFile(asset.files || [], AssetFileType.PREVIEW);
-    if (!previewFile) {
-      this.logger.warn(`Asset ${id} is missing preview image`);
-      return JobStatus.FAILED;
-    }
-
-    if (!asset.embedding) {
-      this.logger.debug(`Asset ${id} is missing embedding`);
-      return JobStatus.FAILED;
-    }
-
-    const duplicateAssets = await this.searchRepository.searchDuplicates({
-      assetId: asset.id,
-      embedding: asset.embedding,
-      maxDistance: machineLearning.duplicateDetection.maxDistance,
-      type: asset.type,
-      userIds: [asset.ownerId],
-    });
-
-    let assetIds = [asset.id];
-    if (duplicateAssets.length > 0) {
-      this.logger.debug(
-        `Found ${duplicateAssets.length} duplicate${duplicateAssets.length === 1 ? '' : 's'} for asset ${asset.id}`,
-      );
-      assetIds = await this.updateDuplicates(asset, duplicateAssets);
-    } else if (asset.duplicateId) {
-      this.logger.debug(`No duplicates found for asset ${asset.id}, removing duplicateId`);
-      await this.assetRepository.update({ id: asset.id, duplicateId: null });
-    }
-
-    const duplicatesDetectedAt = new Date();
-    await this.assetRepository.upsertJobStatus(...assetIds.map((assetId) => ({ assetId, duplicatesDetectedAt })));
-
     return JobStatus.SUCCESS;
   }
 

server/src/services/job.service.ts

@@ -5,7 +5,6 @@ import { OnEvent } from 'src/decorators';
 import { mapAsset } from 'src/dtos/asset-response.dto';
 import { AllJobStatusResponseDto, JobCommandDto, JobCreateDto, JobStatusDto } from 'src/dtos/job.dto';
 import {
-  AssetType,
   BootstrapEventPriority,
   ImmichWorker,
   JobCommand,
@@ -131,7 +130,7 @@ export class JobService extends BaseService {
     return response;
   }
 
-  private async start(name: QueueName, { force }: JobCommandDto): Promise<void> {
+  private async start(name: QueueName, { force }: JobCommandDto): Promise<unknown> {
     const { active } = await this.jobRepository.getJobCounts(name);
     if (active > 0) {
       throw new BadRequestException(`Jobs are already running`);
@@ -306,12 +305,9 @@ export class JobService extends BaseService {
         const jobs: JobItem[] = [
           { name: JobName.SMART_SEARCH, data: item.data },
           { name: JobName.FACE_DETECTION, data: item.data },
+          { name: JobName.VIDEO_CONVERSION, data: item.data },
         ];
 
-        if (asset.type === AssetType.VIDEO) {
-          jobs.push({ name: JobName.VIDEO_CONVERSION, data: item.data });
-        }
-
         await this.jobRepository.queueAll(jobs);
         if (asset.isVisible) {
           this.eventRepository.clientSend('on_upload_success', asset.ownerId, mapAsset(asset));

server/src/services/media.service.ts

@@ -5,24 +5,19 @@ import { Exif } from 'src/database';
 import { OnEvent, OnJob } from 'src/decorators';
 import { SystemConfigFFmpegDto } from 'src/dtos/system-config.dto';
 import {
-  AssetFileType,
   AssetPathType,
-  AssetType,
   AudioCodec,
   Colorspace,
   JobName,
   JobStatus,
-  LogLevel,
   QueueName,
   RawExtractedFormat,
   StorageFolder,
-  TranscodeHWAccel,
   TranscodePolicy,
   TranscodeTarget,
   VideoCodec,
-  VideoContainer,
+  VideoContainer
 } from 'src/enum';
-import { UpsertFileOptions } from 'src/repositories/asset.repository';
 import { BaseService } from 'src/services/base.service';
 import {
   AudioStreamInfo,
@@ -34,7 +29,7 @@ import {
   VideoStreamInfo,
 } from 'src/types';
 import { getAssetFiles } from 'src/utils/asset.util';
-import { BaseConfig, ThumbnailConfig } from 'src/utils/media';
+import { ThumbnailConfig } from 'src/utils/media';
 import { mimeTypes } from 'src/utils/mime-types';
 
 @Injectable()
@@ -49,26 +44,25 @@ export class MediaService extends BaseService {
 
   @OnJob({ name: JobName.QUEUE_GENERATE_THUMBNAILS, queue: QueueName.THUMBNAIL_GENERATION })
   async handleQueueGenerateThumbnails({ force }: JobOf<JobName.QUEUE_GENERATE_THUMBNAILS>): Promise<JobStatus> {
-    let jobs: JobItem[] = [];
+    for (let i = 0; i < 10; i++) {
+      let thumbJobs: JobItem[] = [];
+      for await (const asset of this.assetJobRepository.streamForThumbnailJob(!!force)) {
+        const { previewFile, thumbnailFile } = getAssetFiles(asset.files);
 
-    const queueAll = async () => {
+        if (!previewFile || !thumbnailFile || !asset.thumbhash || force) {
-      await this.jobRepository.queueAll(jobs);
+          thumbJobs.push({ name: JobName.GENERATE_THUMBNAILS, data: { id: asset.id } });
-      jobs = [];
+          continue;
-    };
+        }
 
-    for await (const asset of this.assetJobRepository.streamForThumbnailJob(!!force)) {
+        if (thumbJobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
-      const { previewFile, thumbnailFile } = getAssetFiles(asset.files);
+          await this.jobRepository.queueAll(thumbJobs);
-
+          thumbJobs = [];
-      if (!previewFile || !thumbnailFile || !asset.thumbhash || force) {
+        }
-        jobs.push({ name: JobName.GENERATE_THUMBNAILS, data: { id: asset.id } });
-      }
-
-      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
-        await queueAll();
       }
+      await this.jobRepository.queueAll(thumbJobs);
     }
 
-    await queueAll();
+    const jobs: JobItem[] = [];
 
     const people = this.personRepository.getAll(force ? undefined : { thumbnailPath: '' });
 
@@ -83,12 +77,9 @@ export class MediaService extends BaseService {
       }
 
       jobs.push({ name: JobName.GENERATE_PERSON_THUMBNAIL, data: { id: person.id } });
-      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
-        await queueAll();
-      }
     }
 
-    await queueAll();
+    await this.jobRepository.queueAll(jobs);
 
     return JobStatus.SUCCESS;
   }
@@ -151,75 +142,6 @@ export class MediaService extends BaseService {
       this.logger.warn(`Thumbnail generation failed for asset ${id}: not found`);
       return JobStatus.FAILED;
     }
-
-    if (!asset.isVisible) {
-      this.logger.verbose(`Thumbnail generation skipped for asset ${id}: not visible`);
-      return JobStatus.SKIPPED;
-    }
-
-    let generated: {
-      previewPath: string;
-      thumbnailPath: string;
-      fullsizePath?: string;
-      thumbhash: Buffer;
-    };
-    if (asset.type === AssetType.VIDEO || asset.originalFileName.toLowerCase().endsWith('.gif')) {
-      generated = await this.generateVideoThumbnails(asset);
-    } else if (asset.type === AssetType.IMAGE) {
-      generated = await this.generateImageThumbnails(asset);
-    } else {
-      this.logger.warn(`Skipping thumbnail generation for asset ${id}: ${asset.type} is not an image or video`);
-      return JobStatus.SKIPPED;
-    }
-
-    const { previewFile, thumbnailFile, fullsizeFile } = getAssetFiles(asset.files);
-    const toUpsert: UpsertFileOptions[] = [];
-    if (previewFile?.path !== generated.previewPath) {
-      toUpsert.push({ assetId: asset.id, path: generated.previewPath, type: AssetFileType.PREVIEW });
-    }
-
-    if (thumbnailFile?.path !== generated.thumbnailPath) {
-      toUpsert.push({ assetId: asset.id, path: generated.thumbnailPath, type: AssetFileType.THUMBNAIL });
-    }
-
-    if (generated.fullsizePath && fullsizeFile?.path !== generated.fullsizePath) {
-      toUpsert.push({ assetId: asset.id, path: generated.fullsizePath, type: AssetFileType.FULLSIZE });
-    }
-
-    if (toUpsert.length > 0) {
-      await this.assetRepository.upsertFiles(toUpsert);
-    }
-
-    const pathsToDelete: string[] = [];
-    if (previewFile && previewFile.path !== generated.previewPath) {
-      this.logger.debug(`Deleting old preview for asset ${asset.id}`);
-      pathsToDelete.push(previewFile.path);
-    }
-
-    if (thumbnailFile && thumbnailFile.path !== generated.thumbnailPath) {
-      this.logger.debug(`Deleting old thumbnail for asset ${asset.id}`);
-      pathsToDelete.push(thumbnailFile.path);
-    }
-
-    if (fullsizeFile && fullsizeFile.path !== generated.fullsizePath) {
-      this.logger.debug(`Deleting old fullsize preview image for asset ${asset.id}`);
-      pathsToDelete.push(fullsizeFile.path);
-      if (!generated.fullsizePath) {
-        // did not generate a new fullsize image, delete the existing record
-        await this.assetRepository.deleteFiles([fullsizeFile]);
-      }
-    }
-
-    if (pathsToDelete.length > 0) {
-      await Promise.all(pathsToDelete.map((path) => this.storageRepository.unlink(path)));
-    }
-
-    if (!asset.thumbhash || Buffer.compare(asset.thumbhash, generated.thumbhash) !== 0) {
-      await this.assetRepository.update({ id: asset.id, thumbhash: generated.thumbhash });
-    }
-
-    await this.assetRepository.upsertJobStatus({ assetId: asset.id, previewAt: new Date(), thumbnailAt: new Date() });
-
     return JobStatus.SUCCESS;
   }
 
@@ -364,81 +286,6 @@ export class MediaService extends BaseService {
     if (!asset) {
       return JobStatus.FAILED;
     }
-
-    const input = asset.originalPath;
-    const output = StorageCore.getEncodedVideoPath(asset);
-    this.storageCore.ensureFolders(output);
-
-    const { videoStreams, audioStreams, format } = await this.mediaRepository.probe(input, {
-      countFrames: this.logger.isLevelEnabled(LogLevel.DEBUG), // makes frame count more reliable for progress logs
-    });
-    const videoStream = this.getMainStream(videoStreams);
-    const audioStream = this.getMainStream(audioStreams);
-    if (!videoStream || !format.formatName) {
-      return JobStatus.FAILED;
-    }
-
-    if (!videoStream.height || !videoStream.width) {
-      this.logger.warn(`Skipped transcoding for asset ${asset.id}: no video streams found`);
-      return JobStatus.FAILED;
-    }
-
-    let { ffmpeg } = await this.getConfig({ withCache: true });
-    const target = this.getTranscodeTarget(ffmpeg, videoStream, audioStream);
-    if (target === TranscodeTarget.NONE && !this.isRemuxRequired(ffmpeg, format)) {
-      if (asset.encodedVideoPath) {
-        this.logger.log(`Transcoded video exists for asset ${asset.id}, but is no longer required. Deleting...`);
-        await this.jobRepository.queue({ name: JobName.DELETE_FILES, data: { files: [asset.encodedVideoPath] } });
-        await this.assetRepository.update({ id: asset.id, encodedVideoPath: null });
-      } else {
-        this.logger.verbose(`Asset ${asset.id} does not require transcoding based on current policy, skipping`);
-      }
-
-      return JobStatus.SKIPPED;
-    }
-
-    const command = BaseConfig.create(ffmpeg, this.videoInterfaces).getCommand(target, videoStream, audioStream);
-    if (ffmpeg.accel === TranscodeHWAccel.DISABLED) {
-      this.logger.log(`Transcoding video ${asset.id} without hardware acceleration`);
-    } else {
-      this.logger.log(
-        `Transcoding video ${asset.id} with ${ffmpeg.accel.toUpperCase()}-accelerated encoding and${ffmpeg.accelDecode ? '' : ' software'} decoding`,
-      );
-    }
-
-    try {
-      await this.mediaRepository.transcode(input, output, command);
-    } catch (error: any) {
-      this.logger.error(`Error occurred during transcoding: ${error.message}`);
-      if (ffmpeg.accel === TranscodeHWAccel.DISABLED) {
-        return JobStatus.FAILED;
-      }
-
-      let partialFallbackSuccess = false;
-      if (ffmpeg.accelDecode) {
-        try {
-          this.logger.error(`Retrying with ${ffmpeg.accel.toUpperCase()}-accelerated encoding and software decoding`);
-          ffmpeg = { ...ffmpeg, accelDecode: false };
-          const command = BaseConfig.create(ffmpeg, this.videoInterfaces).getCommand(target, videoStream, audioStream);
-          await this.mediaRepository.transcode(input, output, command);
-          partialFallbackSuccess = true;
-        } catch (error: any) {
-          this.logger.error(`Error occurred during transcoding: ${error.message}`);
-        }
-      }
-
-      if (!partialFallbackSuccess) {
-        this.logger.error(`Retrying with ${ffmpeg.accel.toUpperCase()} acceleration disabled`);
-        ffmpeg = { ...ffmpeg, accel: TranscodeHWAccel.DISABLED };
-        const command = BaseConfig.create(ffmpeg, this.videoInterfaces).getCommand(target, videoStream, audioStream);
-        await this.mediaRepository.transcode(input, output, command);
-      }
-    }
-
-    this.logger.log(`Successfully encoded ${asset.id}`);
-
-    await this.assetRepository.update({ id: asset.id, encodedVideoPath: output });
-
     return JobStatus.SUCCESS;
   }
 

server/src/services/metadata.service.ts

@@ -23,11 +23,9 @@ import {
   SourceType,
 } from 'src/enum';
 import { ArgOf } from 'src/repositories/event.repository';
-import { ReverseGeocodeResult } from 'src/repositories/map.repository';
 import { ImmichTags } from 'src/repositories/metadata.repository';
 import { BaseService } from 'src/services/base.service';
 import { JobItem, JobOf } from 'src/types';
-import { isFaceImportEnabled } from 'src/utils/misc';
 import { upsertTags } from 'src/utils/tag';
 
 /** look for a date from these tags (in order) */
@@ -165,17 +163,20 @@ export class MetadataService extends BaseService {
   async handleQueueMetadataExtraction(job: JobOf<JobName.QUEUE_METADATA_EXTRACTION>): Promise<JobStatus> {
     const { force } = job;
 
-    let queue: { name: JobName.METADATA_EXTRACTION; data: { id: string } }[] = [];
+    for (let i = 0; i < 10; i++) {
-    for await (const asset of this.assetJobRepository.streamForMetadataExtraction(force)) {
+      let queue: { name: JobName.METADATA_EXTRACTION; data: { id: string } }[] = [];
-      queue.push({ name: JobName.METADATA_EXTRACTION, data: { id: asset.id } });
+      for await (const asset of this.assetJobRepository.streamForMetadataExtraction(force)) {
+        queue.push({ name: JobName.METADATA_EXTRACTION, data: { id: asset.id, source: 'upload' } as any });
 
-      if (queue.length >= JOBS_ASSET_PAGINATION_SIZE) {
+        if (queue.length >= JOBS_ASSET_PAGINATION_SIZE) {
-        await this.jobRepository.queueAll(queue);
+          await this.jobRepository.queueAll(queue);
-        queue = [];
+          queue = [];
+        }
       }
+
+      await this.jobRepository.queueAll(queue);
     }
 
-    await this.jobRepository.queueAll(queue);
     return JobStatus.SUCCESS;
   }
 
@@ -189,98 +190,6 @@ export class MetadataService extends BaseService {
     if (!asset) {
       return JobStatus.FAILED;
     }
-
-    const [exifTags, stats] = await Promise.all([
-      this.getExifTags(asset),
-      this.storageRepository.stat(asset.originalPath),
-    ]);
-    this.logger.verbose('Exif Tags', exifTags);
-
-    const dates = this.getDates(asset, exifTags, stats);
-
-    const { width, height } = this.getImageDimensions(exifTags);
-    let geo: ReverseGeocodeResult = { country: null, state: null, city: null },
-      latitude: number | null = null,
-      longitude: number | null = null;
-    if (this.hasGeo(exifTags)) {
-      latitude = exifTags.GPSLatitude;
-      longitude = exifTags.GPSLongitude;
-      if (reverseGeocoding.enabled) {
-        geo = await this.mapRepository.reverseGeocode({ latitude, longitude });
-      }
-    }
-
-    const exifData: Insertable<Exif> = {
-      assetId: asset.id,
-
-      // dates
-      dateTimeOriginal: dates.dateTimeOriginal,
-      modifyDate: stats.mtime,
-      timeZone: dates.timeZone,
-
-      // gps
-      latitude,
-      longitude,
-      country: geo.country,
-      state: geo.state,
-      city: geo.city,
-
-      // image/file
-      fileSizeInByte: stats.size,
-      exifImageHeight: validate(height),
-      exifImageWidth: validate(width),
-      orientation: validate(exifTags.Orientation)?.toString() ?? null,
-      projectionType: exifTags.ProjectionType ? String(exifTags.ProjectionType).toUpperCase() : null,
-      bitsPerSample: this.getBitsPerSample(exifTags),
-      colorspace: exifTags.ColorSpace ?? null,
-
-      // camera
-      make: exifTags.Make ?? exifTags?.Device?.Manufacturer ?? exifTags.AndroidMake ?? null,
-      model: exifTags.Model ?? exifTags?.Device?.ModelName ?? exifTags.AndroidModel ?? null,
-      fps: validate(Number.parseFloat(exifTags.VideoFrameRate!)),
-      iso: validate(exifTags.ISO) as number,
-      exposureTime: exifTags.ExposureTime ?? null,
-      lensModel: getLensModel(exifTags),
-      fNumber: validate(exifTags.FNumber),
-      focalLength: validate(exifTags.FocalLength),
-
-      // comments
-      description: String(exifTags.ImageDescription || exifTags.Description || '').trim(),
-      profileDescription: exifTags.ProfileDescription || null,
-      rating: validateRange(exifTags.Rating, -1, 5),
-
-      // grouping
-      livePhotoCID: (exifTags.ContentIdentifier || exifTags.MediaGroupUUID) ?? null,
-      autoStackId: this.getAutoStackId(exifTags),
-    };
-
-    const promises: Promise<unknown>[] = [
-      this.assetRepository.upsertExif(exifData),
-      this.assetRepository.update({
-        id: asset.id,
-        duration: exifTags.Duration?.toString() ?? null,
-        localDateTime: dates.localDateTime,
-        fileCreatedAt: dates.dateTimeOriginal ?? undefined,
-        fileModifiedAt: stats.mtime,
-      }),
-      this.applyTagList(asset, exifTags),
-    ];
-
-    if (this.isMotionPhoto(asset, exifTags)) {
-      promises.push(this.applyMotionPhotos(asset, exifTags, dates, stats));
-    }
-
-    if (isFaceImportEnabled(metadata) && this.hasTaggedFaces(exifTags)) {
-      promises.push(this.applyTaggedFaces(asset, exifTags));
-    }
-
-    await Promise.all(promises);
-    if (exifData.livePhotoCID) {
-      await this.linkLivePhotos(asset, exifData);
-    }
-
-    await this.assetRepository.upsertJobStatus({ assetId: asset.id, metadataExtractedAt: new Date() });
-
     return JobStatus.SUCCESS;
   }
 
@@ -425,9 +334,9 @@ export class MetadataService extends BaseService {
         typeof tag === 'number'
           ? String(tag)
           : tag
-              .split('|')
+            .split('|')
-              .map((tag) => tag.replaceAll('/', '|'))
+            .map((tag) => tag.replaceAll('/', '|'))
-              .join('/'),
+            .join('/'),
       );
     } else if (exifTags.Keywords) {
       let keywords = exifTags.Keywords;

server/src/services/person.service.ts

@@ -1,9 +1,7 @@
 import { BadRequestException, Injectable, NotFoundException } from '@nestjs/common';
-import { Insertable, Updateable } from 'kysely';
+import { Updateable } from 'kysely';
-import { FACE_THUMBNAIL_SIZE, JOBS_ASSET_PAGINATION_SIZE } from 'src/constants';
+import { JOBS_ASSET_PAGINATION_SIZE } from 'src/constants';
-import { StorageCore } from 'src/cores/storage.core';
 import { Person } from 'src/database';
-import { AssetFaces, FaceSearch } from 'src/db';
 import { Chunked, OnJob } from 'src/decorators';
 import { BulkIdErrorReason, BulkIdResponseDto } from 'src/dtos/asset-ids.response.dto';
 import { AuthDto } from 'src/dtos/auth.dto';
@@ -27,7 +25,6 @@ import {
 import {
   AssetType,
   CacheControl,
-  ImageFormat,
   JobName,
   JobStatus,
   Permission,
@@ -291,79 +288,6 @@ export class PersonService extends BaseService {
     }
 
     const asset = await this.assetJobRepository.getForDetectFacesJob(id);
-    const previewFile = asset?.files[0];
-    if (!asset || asset.files.length !== 1 || !previewFile) {
-      return JobStatus.FAILED;
-    }
-
-    if (!asset.isVisible) {
-      return JobStatus.SKIPPED;
-    }
-
-    const { imageHeight, imageWidth, faces } = await this.machineLearningRepository.detectFaces(
-      machineLearning.urls,
-      previewFile.path,
-      machineLearning.facialRecognition,
-    );
-    this.logger.debug(`${faces.length} faces detected in ${previewFile.path}`);
-
-    const facesToAdd: (Insertable<AssetFaces> & { id: string })[] = [];
-    const embeddings: FaceSearch[] = [];
-    const mlFaceIds = new Set<string>();
-
-    for (const face of asset.faces) {
-      if (face.sourceType === SourceType.MACHINE_LEARNING) {
-        mlFaceIds.add(face.id);
-      }
-    }
-
-    const heightScale = imageHeight / (asset.faces[0]?.imageHeight || 1);
-    const widthScale = imageWidth / (asset.faces[0]?.imageWidth || 1);
-    for (const { boundingBox, embedding } of faces) {
-      const scaledBox = {
-        x1: boundingBox.x1 * widthScale,
-        y1: boundingBox.y1 * heightScale,
-        x2: boundingBox.x2 * widthScale,
-        y2: boundingBox.y2 * heightScale,
-      };
-      const match = asset.faces.find((face) => this.iou(face, scaledBox) > 0.5);
-
-      if (match && !mlFaceIds.delete(match.id)) {
-        embeddings.push({ faceId: match.id, embedding });
-      } else if (!match) {
-        const faceId = this.cryptoRepository.randomUUID();
-        facesToAdd.push({
-          id: faceId,
-          assetId: asset.id,
-          imageHeight,
-          imageWidth,
-          boundingBoxX1: boundingBox.x1,
-          boundingBoxY1: boundingBox.y1,
-          boundingBoxX2: boundingBox.x2,
-          boundingBoxY2: boundingBox.y2,
-        });
-        embeddings.push({ faceId, embedding });
-      }
-    }
-    const faceIdsToRemove = [...mlFaceIds];
-
-    if (facesToAdd.length > 0 || faceIdsToRemove.length > 0 || embeddings.length > 0) {
-      await this.personRepository.refreshFaces(facesToAdd, faceIdsToRemove, embeddings);
-    }
-
-    if (faceIdsToRemove.length > 0) {
-      this.logger.log(`Removed ${faceIdsToRemove.length} faces below detection threshold in asset ${id}`);
-    }
-
-    if (facesToAdd.length > 0) {
-      this.logger.log(`Detected ${facesToAdd.length} new faces in asset ${id}`);
-      const jobs = facesToAdd.map((face) => ({ name: JobName.FACIAL_RECOGNITION, data: { id: face.id } }) as const);
-      await this.jobRepository.queueAll([{ name: JobName.QUEUE_FACIAL_RECOGNITION, data: { force: false } }, ...jobs]);
-    } else if (embeddings.length > 0) {
-      this.logger.log(`Added ${embeddings.length} face embeddings for asset ${id}`);
-    }
-
-    await this.assetRepository.upsertJobStatus({ assetId: asset.id, facesRecognizedAt: new Date() });
 
     return JobStatus.SUCCESS;
   }
@@ -547,26 +471,6 @@ export class PersonService extends BaseService {
       this.logger.error(`Could not generate person thumbnail for ${id}: missing data`);
       return JobStatus.FAILED;
     }
-
-    const { ownerId, x1, y1, x2, y2, oldWidth, oldHeight } = data;
-
-    const { width, height, inputPath } = await this.getInputDimensions(data);
-
-    const thumbnailPath = StorageCore.getPersonThumbnailPath({ id, ownerId });
-    this.storageCore.ensureFolders(thumbnailPath);
-
-    const thumbnailOptions = {
-      colorspace: image.colorspace,
-      format: ImageFormat.JPEG,
-      size: FACE_THUMBNAIL_SIZE,
-      quality: image.thumbnail.quality,
-      crop: this.getCrop({ old: { width: oldWidth, height: oldHeight }, new: { width, height } }, { x1, y1, x2, y2 }),
-      processInvalidImages: process.env.IMMICH_PROCESS_INVALID_IMAGES === 'true',
-    };
-
-    await this.mediaRepository.generateThumbnail(inputPath, thumbnailOptions, thumbnailPath);
-    await this.personRepository.update({ id, thumbnailPath });
-
     return JobStatus.SUCCESS;
   }
 

server/src/services/smart-info.service.ts

@@ -103,30 +103,6 @@ export class SmartInfoService extends BaseService {
     if (!asset || asset.files.length !== 1) {
       return JobStatus.FAILED;
     }
-
-    if (!asset.isVisible) {
-      return JobStatus.SKIPPED;
-    }
-
-    const embedding = await this.machineLearningRepository.encodeImage(
-      machineLearning.urls,
-      asset.files[0].path,
-      machineLearning.clip,
-    );
-
-    if (this.databaseRepository.isBusy(DatabaseLock.CLIPDimSize)) {
-      this.logger.verbose(`Waiting for CLIP dimension size to be updated`);
-      await this.databaseRepository.wait(DatabaseLock.CLIPDimSize);
-    }
-
-    const newConfig = await this.getConfig({ withCache: true });
-    if (machineLearning.clip.modelName !== newConfig.machineLearning.clip.modelName) {
-      // Skip the job if the the model has changed since the embedding was generated.
-      return JobStatus.SKIPPED;
-    }
-
-    await this.searchRepository.upsert(asset.id, embedding);
-
     return JobStatus.SUCCESS;
   }
 }

server/src/services/storage-template.service.ts

@@ -10,7 +10,6 @@ import { AssetPathType, AssetType, DatabaseLock, JobName, JobStatus, QueueName,
 import { ArgOf } from 'src/repositories/event.repository';
 import { BaseService } from 'src/services/base.service';
 import { JobOf, StorageAsset } from 'src/types';
-import { getLivePhotoMotionFilename } from 'src/utils/file';
 
 const storageTokens = {
   secondOptions: ['s', 'ss', 'SSS'],
@@ -128,21 +127,6 @@ export class StorageTemplateService extends BaseService {
     if (!asset) {
       return JobStatus.FAILED;
     }
-
-    const user = await this.userRepository.get(asset.ownerId, {});
-    const storageLabel = user?.storageLabel || null;
-    const filename = asset.originalFileName || asset.id;
-    await this.moveAsset(asset, { storageLabel, filename });
-
-    // move motion part of live photo
-    if (asset.livePhotoVideoId) {
-      const livePhotoVideo = await this.assetJobRepository.getForStorageTemplateJob(asset.livePhotoVideoId);
-      if (!livePhotoVideo) {
-        return JobStatus.FAILED;
-      }
-      const motionFilename = getLivePhotoMotionFilename(filename, livePhotoVideo.originalPath);
-      await this.moveAsset(livePhotoVideo, { storageLabel, filename: motionFilename });
-    }
     return JobStatus.SUCCESS;
   }