refactor: queue in batches (#30698)

server/src/services/asset.service.ts

@@ -1,7 +1,6 @@
 import { BadRequestException, Injectable } from '@nestjs/common';
 import _ from 'lodash';
 import { DateTime, Duration } from 'luxon';
-import { JOBS_ASSET_PAGINATION_SIZE } from 'src/constants';
 import { AssetFile } from 'src/database';
 import { OnJob } from 'src/decorators';
 import { AssetResponseDto, SanitizedAssetResponseDto, mapAsset } from 'src/dtos/asset-response.dto';
@@ -46,7 +45,7 @@ import {
 } from 'src/utils/asset.util';
 import { updateLockedColumns } from 'src/utils/database';
 import { extractTimeZone } from 'src/utils/date';
-import { findOrFail } from 'src/utils/misc';
+import { batched, findOrFail } from 'src/utils/misc';
 import { transformOcrBoundingBox } from 'src/utils/transform';
 
 @Injectable()
@@ -279,31 +278,12 @@ export class AssetService extends BaseService {
       .minus(Duration.fromObject({ days: trashedDays }))
       .toJSDate();
 
-    let chunk: Array<{ id: string; isOffline: boolean }> = [];
+    for await (const assets of batched(this.assetJobRepository.streamForDeletedJob(trashedBefore))) {
-    const queueChunk = async () => {
-      if (chunk.length === 0) {
-        return;
-      }
-
       await this.jobRepository.queueAll(
-        chunk.map(({ id, isOffline }) => ({
+        assets.map(({ id, isOffline }) => ({ name: JobName.AssetDelete, data: { id, deleteOnDisk: !isOffline } })),
-          name: JobName.AssetDelete,
-          data: { id, deleteOnDisk: !isOffline },
-        })),
       );
-      chunk = [];
-    };
-
-    const assets = this.assetJobRepository.streamForDeletedJob(trashedBefore);
-    for await (const asset of assets) {
-      chunk.push(asset);
-      if (chunk.length >= JOBS_ASSET_PAGINATION_SIZE) {
-        await queueChunk();
-      }
     }
 
-    await queueChunk();
-
     return JobStatus.Success;
   }
 

server/src/services/duplicate.service.ts

@@ -1,5 +1,4 @@
 import { Injectable } from '@nestjs/common';
-import { JOBS_ASSET_PAGINATION_SIZE } from 'src/constants';
 import { OnJob } from 'src/decorators';
 import { BulkIdErrorReason, BulkIdResponseDto, BulkIdsDto } from 'src/dtos/asset-ids.response.dto';
 import { MapAsset, mapAsset } from 'src/dtos/asset-response.dto';
@@ -8,9 +7,9 @@ import { DuplicateResolveDto, DuplicateResolveGroupDto, DuplicateResponseDto } f
 import { AssetStatus, AssetVisibility, JobName, JobStatus, Permission, 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 { JobOf } from 'src/types';
 import { suggestDuplicateKeepAssetIds } from 'src/utils/duplicate';
-import { isDuplicateDetectionEnabled } from 'src/utils/misc';
+import { batched, isDuplicateDetectionEnabled } from 'src/utils/misc';
 
 type ResolveRequest = {
   assetUpdate: {
@@ -307,22 +306,12 @@ export class DuplicateService extends BaseService {
       return JobStatus.Skipped;
     }
 
-    let jobs: JobItem[] = [];
+    for await (const assets of batched(this.assetJobRepository.streamForSearchDuplicates(force))) {
-    const queueAll = async () => {
+      await this.jobRepository.queueAll(
-      await this.jobRepository.queueAll(jobs);
+        assets.map((asset) => ({ name: JobName.AssetDetectDuplicates, data: { id: asset.id } })),
-      jobs = [];
+      );
-    };
-
-    const assets = this.assetJobRepository.streamForSearchDuplicates(force);
-    for await (const asset of assets) {
-      jobs.push({ name: JobName.AssetDetectDuplicates, data: { id: asset.id } });
-      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
-        await queueAll();
-      }
     }
 
-    await queueAll();
-
     return JobStatus.Success;
   }
 

server/src/services/integrity.service.ts

@@ -35,7 +35,7 @@ import {
   IIntegrityUntrackedFilesJob,
 } from 'src/types';
 import { ImmichFileResponse } from 'src/utils/file';
-import { handlePromiseError } from 'src/utils/misc';
+import { batched, handlePromiseError } from 'src/utils/misc';
 
 /**
  * Untracked Files:
@@ -201,7 +201,7 @@ export class IntegrityService extends BaseService {
     const reports = this.integrityRepository.streamIntegrityReportsWithAssetChecksum(IntegrityReport.UntrackedFile);
 
     let total = 0;
-    for await (const batchReports of chunk(reports, JOBS_LIBRARY_PAGINATION_SIZE)) {
+    for await (const batchReports of batched(reports, JOBS_LIBRARY_PAGINATION_SIZE)) {
       await this.jobRepository.queue({
         name: JobName.IntegrityUntrackedFilesRefresh,
         data: {
@@ -338,7 +338,7 @@ export class IntegrityService extends BaseService {
     const reports = this.integrityRepository.streamIntegrityReportsWithAssetChecksum(IntegrityReport.MissingFile);
 
     let total = 0;
-    for await (const batchReports of chunk(reports, JOBS_LIBRARY_PAGINATION_SIZE)) {
+    for await (const batchReports of batched(reports, JOBS_LIBRARY_PAGINATION_SIZE)) {
       await this.jobRepository.queue({
         name: JobName.IntegrityMissingFilesRefresh,
         data: {
@@ -365,7 +365,7 @@ export class IntegrityService extends BaseService {
     const assetPaths = this.integrityRepository.streamAssetPathsForMissingFiles();
 
     let total = 0;
-    for await (const batchPaths of chunk(assetPaths, JOBS_LIBRARY_PAGINATION_SIZE)) {
+    for await (const batchPaths of batched(assetPaths, JOBS_LIBRARY_PAGINATION_SIZE)) {
       await this.jobRepository.queue({
         name: JobName.IntegrityMissingFiles,
         data: {
@@ -450,7 +450,7 @@ export class IntegrityService extends BaseService {
     const reports = this.integrityRepository.streamIntegrityReportsWithAssetChecksum(IntegrityReport.ChecksumFail);
 
     let total = 0;
-    for await (const batchReports of chunk(reports, JOBS_LIBRARY_PAGINATION_SIZE)) {
+    for await (const batchReports of batched(reports, JOBS_LIBRARY_PAGINATION_SIZE)) {
       await this.jobRepository.queue({
         name: JobName.IntegrityChecksumFilesRefresh,
         data: {
@@ -656,7 +656,7 @@ export class IntegrityService extends BaseService {
 
     for (const property of properties) {
       const reports = this.integrityRepository.streamIntegrityReportsByProperty(property, type);
-      for await (const batch of chunk(reports, JOBS_LIBRARY_PAGINATION_SIZE)) {
+      for await (const batch of batched(reports, JOBS_LIBRARY_PAGINATION_SIZE)) {
         await this.jobRepository.queue({
           name: JobName.IntegrityDeleteReports,
           data: {
@@ -705,19 +705,3 @@ export class IntegrityService extends BaseService {
     return JobStatus.Success;
   }
 }
-
-async function* chunk<T>(generator: AsyncIterableIterator<T>, n: number) {
-  let chunk: T[] = [];
-  for await (const item of generator) {
-    chunk.push(item);
-
-    if (chunk.length === n) {
-      yield chunk;
-      chunk = [];
-    }
-  }
-
-  if (chunk.length > 0) {
-    yield chunk;
-  }
-}

server/src/services/library.service.ts

@@ -34,7 +34,7 @@ import { AssetTable } from 'src/schema/tables/asset.table';
 import { BaseService } from 'src/services/base.service';
 import { JobOf } from 'src/types';
 import { mimeTypes } from 'src/utils/mime-types';
-import { findOrFail, handlePromiseError } from 'src/utils/misc';
+import { batched, findOrFail, handlePromiseError } from 'src/utils/misc';
 
 @Injectable()
 export class LibraryService extends BaseService {
@@ -375,35 +375,20 @@ export class LibraryService extends BaseService {
 
     await this.assetRepository.updateByLibraryId(libraryId, { deletedAt: new Date() });
 
-    let isAssetsFound = false;
-    let chunk: string[] = [];
-
-    const queueChunk = async () => {
-      if (chunk.length === 0) {
-        return;
-      }
-
-      isAssetsFound = true;
-      this.logger.debug(`Queueing deletion of ${chunk.length} asset(s) in library ${libraryId}`);
-      await this.jobRepository.queueAll(
-        chunk.map((id) => ({ name: JobName.AssetDelete, data: { id, deleteOnDisk: false } })),
-      );
-      chunk = [];
-    };
-
     this.logger.debug(`Will delete all assets in library ${libraryId}`);
-    const assets = this.libraryRepository.streamAssetIds(libraryId);
+    let hasAssets = false;
-    for await (const asset of assets) {
+    for await (const assets of batched(
-      chunk.push(asset.id);
+      this.libraryRepository.streamAssetIds(libraryId),
-
+      JOBS_LIBRARY_PAGINATION_SIZE,
-      if (chunk.length >= JOBS_LIBRARY_PAGINATION_SIZE) {
+    )) {
-        await queueChunk();
+      this.logger.debug(`Queueing deletion of ${assets.length} asset(s) in library ${libraryId}`);
-      }
+      await this.jobRepository.queueAll(
+        assets.map((asset) => ({ name: JobName.AssetDelete, data: { id: asset.id, deleteOnDisk: false } })),
+      );
+      hasAssets = true;
     }
 
-    await queueChunk();
+    if (!hasAssets) {
-
-    if (!isAssetsFound) {
       this.logger.log(`Deleting library ${libraryId}`);
       await this.libraryRepository.delete(libraryId);
     }
@@ -746,15 +731,12 @@ export class LibraryService extends BaseService {
       return JobStatus.Success;
     }
 
-    let chunk: string[] = [];
+    this.logger.log(`Scanning library ${library.id} for assets missing from disk...`);
+
     let count = 0;
-
+    const existingAssets = this.libraryRepository.streamAssetIds(library.id);
-    const queueChunk = async () => {
+    for await (const assets of batched(existingAssets, JOBS_LIBRARY_PAGINATION_SIZE)) {
-      if (chunk.length === 0) {
+      count += assets.length;
-        return;
-      }
-
-      count += chunk.length;
 
       await this.jobRepository.queue({
         name: JobName.LibrarySyncAssets,
@@ -762,32 +744,19 @@ export class LibraryService extends BaseService {
           libraryId: library.id,
           importPaths: library.importPaths,
           exclusionPatterns: library.exclusionPatterns,
-          assetIds: chunk.map((id) => id),
+          assetIds: assets.map(({ id }) => id),
           progressCounter: count,
           totalAssets: assetCount,
         },
       });
-      chunk = [];
 
       const completePercentage = ((100 * count) / assetCount).toFixed(1);
 
       this.logger.log(
         `Queued check of ${count} of ${assetCount} (${completePercentage} %) existing asset(s) so far in library ${library.id}`,
       );
-    };
-
-    this.logger.log(`Scanning library ${library.id} for assets missing from disk...`);
-    const existingAssets = this.libraryRepository.streamAssetIds(library.id);
-
-    for await (const asset of existingAssets) {
-      chunk.push(asset.id);
-      if (chunk.length === JOBS_LIBRARY_PAGINATION_SIZE) {
-        await queueChunk();
-      }
     }
 
-    await queueChunk();
-
     this.logger.log(`Finished queuing ${count} asset check(s) for library ${library.id}`);
 
     return JobStatus.Success;

server/src/services/media.service.spec.ts

@@ -207,8 +207,7 @@ describe(MediaService.name, () => {
       await sut.handleQueueGenerateThumbnails({ force: false });
 
       expect(mocks.assetJob.streamForThumbnailJob).toHaveBeenCalledWith({ force: false, fullsizeEnabled: false });
-      expect(mocks.job.queueAll).toHaveBeenCalledWith([]);
+      expect(mocks.job.queueAll).toHaveBeenCalledTimes(1);
-
       expect(mocks.person.getAll).toHaveBeenCalledWith({ thumbnailPath: '' });
     });
 
@@ -237,7 +236,10 @@ describe(MediaService.name, () => {
       await sut.handleQueueGenerateThumbnails({ force: false });
 
       expect(mocks.assetJob.streamForThumbnailJob).toHaveBeenCalledWith({ force: false, fullsizeEnabled: false });
-      expect(mocks.job.queueAll).toHaveBeenCalledWith([]);
+      expect(mocks.job.queueAll).toHaveBeenCalledTimes(1);
+      expect(mocks.job.queueAll).toHaveBeenCalledWith([
+        { name: JobName.AssetEditThumbnailGeneration, data: { id: asset.id } },
+      ]);
 
       expect(mocks.person.getAll).toHaveBeenCalledWith({ thumbnailPath: '' });
     });

server/src/services/media.service.ts

@@ -1,6 +1,6 @@
 import { Injectable } from '@nestjs/common';
 import { SystemConfig } from 'src/config';
-import { FACE_THUMBNAIL_SIZE, JOBS_ASSET_PAGINATION_SIZE } from 'src/constants';
+import { FACE_THUMBNAIL_SIZE } from 'src/constants';
 import { ImagePathOptions, StorageCore, ThumbnailPathEntity } from 'src/cores/storage.core';
 import { AssetFile } from 'src/database';
 import { OnEvent, OnJob } from 'src/decorators';
@@ -43,7 +43,7 @@ import { getAssetFile, getDimensions } from 'src/utils/asset.util';
 import { checkFaceVisibility, checkOcrVisibility } from 'src/utils/editor';
 import { BaseConfig, ThumbnailConfig } from 'src/utils/media';
 import { mimeTypes } from 'src/utils/mime-types';
-import { clamp } from 'src/utils/misc';
+import { batched, clamp } from 'src/utils/misc';
 import { getOutputDimensions } from 'src/utils/transform';
 
 interface UpsertFileOptions {
@@ -69,52 +69,42 @@ export class MediaService extends BaseService {
   @OnJob({ name: JobName.AssetGenerateThumbnailsQueueAll, queue: QueueName.ThumbnailGeneration })
   async handleQueueGenerateThumbnails({ force }: JobOf<JobName.AssetGenerateThumbnailsQueueAll>): Promise<JobStatus> {
     const config = await this.getConfig({ withCache: true });
-    let jobs: JobItem[] = [];
-
-    const queueAll = async () => {
-      await this.jobRepository.queueAll(jobs);
-      jobs = [];
-    };
 
     const isFullsizeEnabled = config.image.fullsize.enabled;
-    for await (const asset of this.assetJobRepository.streamForThumbnailJob({
+    for await (const assets of batched(
-      force,
+      this.assetJobRepository.streamForThumbnailJob({ force, fullsizeEnabled: isFullsizeEnabled }),
-      fullsizeEnabled: isFullsizeEnabled,
+    )) {
-    })) {
+      const jobs: JobItem[] = [];
-      if (force || !asset.isEdited) {
+      for (const asset of assets) {
-        jobs.push({ name: JobName.AssetGenerateThumbnails, data: { id: asset.id } });
+        if (force || !asset.isEdited) {
-      }
+          jobs.push({ name: JobName.AssetGenerateThumbnails, data: { id: asset.id } });
-
-      if (asset.isEdited) {
-        jobs.push({ name: JobName.AssetEditThumbnailGeneration, data: { id: asset.id } });
-      }
-
-      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
-        await queueAll();
-      }
-    }
-
-    await queueAll();
-
-    const people = this.personRepository.getAll(force ? undefined : { thumbnailPath: '' });
-
-    for await (const person of people) {
-      if (!person.faceAssetId) {
-        const face = await this.personRepository.getRandomFace(person.id);
-        if (!face) {
-          continue;
         }
 
-        await this.personRepository.update({ id: person.id, faceAssetId: face.id });
+        if (asset.isEdited) {
+          jobs.push({ name: JobName.AssetEditThumbnailGeneration, data: { id: asset.id } });
+        }
       }
 
-      jobs.push({ name: JobName.PersonGenerateThumbnail, data: { id: person.id } });
+      await this.jobRepository.queueAll(jobs);
-      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
-        await queueAll();
-      }
     }
 
-    await queueAll();
+    for await (const people of batched(this.personRepository.getAll(force ? undefined : { thumbnailPath: '' }))) {
+      const jobs: JobItem[] = [];
+      for (const person of people) {
+        if (!person.faceAssetId) {
+          const face = await this.personRepository.getRandomFace(person.id);
+          if (!face) {
+            continue;
+          }
+
+          await this.personRepository.update({ id: person.id, faceAssetId: face.id });
+        }
+
+        jobs.push({ name: JobName.PersonGenerateThumbnail, data: { id: person.id } });
+      }
+
+      await this.jobRepository.queueAll(jobs);
+    }
 
     return JobStatus.Success;
   }
@@ -127,30 +117,18 @@ export class MediaService extends BaseService {
       await this.storageCore.removeEmptyDirs(StorageFolder.EncodedVideo);
     }
 
-    let jobs: JobItem[] = [];
+    for await (const assets of batched(this.assetJobRepository.streamForMigrationJob())) {
-    const assets = this.assetJobRepository.streamForMigrationJob();
+      await this.jobRepository.queueAll(
-    for await (const asset of assets) {
+        assets.map((asset) => ({ name: JobName.AssetFileMigration, data: { id: asset.id } })),
-      jobs.push({ name: JobName.AssetFileMigration, data: { id: asset.id } });
+      );
-      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
-        await this.jobRepository.queueAll(jobs);
-        jobs = [];
-      }
     }
 
-    await this.jobRepository.queueAll(jobs);
+    for await (const people of batched(this.personRepository.getAll())) {
-    jobs = [];
+      await this.jobRepository.queueAll(
-
+        people.map((person) => ({ name: JobName.PersonFileMigration, data: { id: person.id } })),
-    for await (const person of this.personRepository.getAll()) {
+      );
-      jobs.push({ name: JobName.PersonFileMigration, data: { id: person.id } });
-
-      if (jobs.length === JOBS_ASSET_PAGINATION_SIZE) {
-        await this.jobRepository.queueAll(jobs);
-        jobs = [];
-      }
     }
 
-    await this.jobRepository.queueAll(jobs);
-
     return JobStatus.Success;
   }
 
@@ -551,18 +529,12 @@ export class MediaService extends BaseService {
   async handleQueueVideoConversion(job: JobOf<JobName.AssetEncodeVideoQueueAll>): Promise<JobStatus> {
     const { force } = job;
 
-    let queue: { name: JobName.AssetEncodeVideo; data: { id: string } }[] = [];
+    for await (const assets of batched(this.assetJobRepository.streamForVideoConversion(force))) {
-    for await (const asset of this.assetJobRepository.streamForVideoConversion(force)) {
+      await this.jobRepository.queueAll(
-      queue.push({ name: JobName.AssetEncodeVideo, data: { id: asset.id } });
+        assets.map((asset) => ({ name: JobName.AssetEncodeVideo, data: { id: asset.id } })),
-
+      );
-      if (queue.length >= JOBS_ASSET_PAGINATION_SIZE) {
-        await this.jobRepository.queueAll(queue);
-        queue = [];
-      }
     }
 
-    await this.jobRepository.queueAll(queue);
-
     return JobStatus.Success;
   }
 

server/src/services/metadata.service.ts

@@ -6,7 +6,7 @@ import { DateTime, Duration } from 'luxon';
 import { Stats } from 'node:fs';
 import { constants } from 'node:fs/promises';
 import { join, parse } from 'node:path';
-import { JOBS_ASSET_PAGINATION_SIZE } from 'src/constants';
+
 import { StorageCore } from 'src/cores/storage.core';
 import { Asset, AssetFile } from 'src/database';
 import { OnEvent, OnJob } from 'src/decorators';
@@ -30,12 +30,12 @@ import { AssetExifTable } from 'src/schema/tables/asset-exif.table';
 import { AssetFaceTable } from 'src/schema/tables/asset-face.table';
 import { PersonTable } from 'src/schema/tables/person.table';
 import { BaseService } from 'src/services/base.service';
-import { JobItem, JobOf } from 'src/types';
+import { JobOf } from 'src/types';
 import { getAssetFiles } from 'src/utils/asset.util';
 import { isAssetChecksumConstraint } from 'src/utils/database';
 import { mergeTimeZone } from 'src/utils/date';
 import { mimeTypes } from 'src/utils/mime-types';
-import { isFaceImportEnabled } from 'src/utils/misc';
+import { batched, isFaceImportEnabled } from 'src/utils/misc';
 import { upsertTags } from 'src/utils/tag';
 import { Tasks } from 'src/utils/tasks';
 
@@ -218,17 +218,12 @@ export class MetadataService extends BaseService {
   async handleQueueMetadataExtraction(job: JobOf<JobName.AssetExtractMetadataQueueAll>): Promise<JobStatus> {
     const { force } = job;
 
-    let queue: { name: JobName.AssetExtractMetadata; data: { id: string } }[] = [];
+    for await (const assets of batched(this.assetJobRepository.streamForMetadataExtraction(force))) {
-    for await (const asset of this.assetJobRepository.streamForMetadataExtraction(force)) {
+      await this.jobRepository.queueAll(
-      queue.push({ name: JobName.AssetExtractMetadata, data: { id: asset.id } });
+        assets.map((asset) => ({ name: JobName.AssetExtractMetadata, data: { id: asset.id } })),
-
+      );
-      if (queue.length >= JOBS_ASSET_PAGINATION_SIZE) {
-        await this.jobRepository.queueAll(queue);
-        queue = [];
-      }
     }
 
-    await this.jobRepository.queueAll(queue);
     return JobStatus.Success;
   }
 
@@ -417,22 +412,12 @@ export class MetadataService extends BaseService {
 
   @OnJob({ name: JobName.SidecarQueueAll, queue: QueueName.Sidecar })
   async handleQueueSidecar({ force }: JobOf<JobName.SidecarQueueAll>): Promise<JobStatus> {
-    let jobs: JobItem[] = [];
+    for await (const assets of batched(this.assetJobRepository.streamForSidecar(force))) {
-    const queueAll = async () => {
+      await this.jobRepository.queueAll(
-      await this.jobRepository.queueAll(jobs);
+        assets.map((asset) => ({ name: JobName.SidecarCheck, data: { id: asset.id } })),
-      jobs = [];
+      );
-    };
-
-    const assets = this.assetJobRepository.streamForSidecar(force);
-    for await (const asset of assets) {
-      jobs.push({ name: JobName.SidecarCheck, data: { id: asset.id } });
-      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
-        await queueAll();
-      }
     }
 
-    await queueAll();
-
     return JobStatus.Success;
   }
 

server/src/services/ocr.service.ts

@@ -1,12 +1,12 @@
 import { Injectable } from '@nestjs/common';
-import { JOBS_ASSET_PAGINATION_SIZE } from 'src/constants';
+
 import { OnJob } from 'src/decorators';
 import { AssetVisibility, JobName, JobStatus, QueueName } from 'src/enum';
 import { OCR } from 'src/repositories/machine-learning.repository';
 import { BaseService } from 'src/services/base.service';
-import { JobItem, JobOf } from 'src/types';
+import { JobOf } from 'src/types';
 import { tokenizeForSearch } from 'src/utils/database';
-import { isOcrEnabled } from 'src/utils/misc';
+import { batched, isOcrEnabled } from 'src/utils/misc';
 
 @Injectable()
 export class OcrService extends BaseService {
@@ -21,19 +21,10 @@ export class OcrService extends BaseService {
       await this.ocrRepository.deleteAll();
     }
 
-    let jobs: JobItem[] = [];
+    for await (const assets of batched(this.assetJobRepository.streamForOcrJob(force))) {
-    const assets = this.assetJobRepository.streamForOcrJob(force);
+      await this.jobRepository.queueAll(assets.map((asset) => ({ name: JobName.Ocr, data: { id: asset.id } })));
-
-    for await (const asset of assets) {
-      jobs.push({ name: JobName.Ocr, data: { id: asset.id } });
-
-      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
-        await this.jobRepository.queueAll(jobs);
-        jobs = [];
-      }
     }
 
-    await this.jobRepository.queueAll(jobs);
     return JobStatus.Success;
   }
 

server/src/services/person.service.ts

@@ -1,6 +1,5 @@
 import { BadRequestException, Injectable, NotFoundException } from '@nestjs/common';
 import { Insertable, Updateable } from 'kysely';
-import { JOBS_ASSET_PAGINATION_SIZE } from 'src/constants';
 import { Person } from 'src/database';
 import { Chunked, OnJob } from 'src/decorators';
 import { BulkIdErrorReason, BulkIdResponseDto, BulkIdsDto } from 'src/dtos/asset-ids.response.dto';
@@ -43,7 +42,7 @@ import { JobItem, JobOf } from 'src/types';
 import { getDimensions } from 'src/utils/asset.util';
 import { ImmichFileResponse } from 'src/utils/file';
 import { mimeTypes } from 'src/utils/mime-types';
-import { findOrFail, isFacialRecognitionEnabled } from 'src/utils/misc';
+import { batched, findOrFail, isFacialRecognitionEnabled } from 'src/utils/misc';
 import { Point, transformPoints } from 'src/utils/transform';
 
 @Injectable()
@@ -277,19 +276,12 @@ export class PersonService extends BaseService {
       await this.personRepository.vacuum({ reindexVectors: true });
     }
 
-    let jobs: JobItem[] = [];
+    for await (const assets of batched(this.assetJobRepository.streamForDetectFacesJob(force))) {
-    const assets = this.assetJobRepository.streamForDetectFacesJob(force);
+      await this.jobRepository.queueAll(
-    for await (const asset of assets) {
+        assets.map((asset) => ({ name: JobName.AssetDetectFaces, data: { id: asset.id } })),
-      jobs.push({ name: JobName.AssetDetectFaces, data: { id: asset.id } });
+      );
-
-      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
-        await this.jobRepository.queueAll(jobs);
-        jobs = [];
-      }
     }
 
-    await this.jobRepository.queueAll(jobs);
-
     if (force === undefined) {
       await this.jobRepository.queue({ name: JobName.PersonCleanup });
     }
@@ -435,22 +427,16 @@ export class PersonService extends BaseService {
     await this.databaseRepository.prewarm(VectorIndex.Face);
 
     const lastRun = new Date().toISOString();
-    const facePagination = this.personRepository.getAllFaces(
+
+    const faces = this.personRepository.getAllFaces(
       force ? undefined : { personId: null, sourceType: SourceType.MachineLearning },
     );
-
+    for await (const batch of batched(faces)) {
-    let jobs: { name: JobName.FacialRecognition; data: { id: string; deferred: false } }[] = [];
+      await this.jobRepository.queueAll(
-    for await (const face of facePagination) {
+        batch.map((face) => ({ name: JobName.FacialRecognition, data: { id: face.id, deferred: false } })),
-      jobs.push({ name: JobName.FacialRecognition, data: { id: face.id, deferred: false } });
+      );
-
-      if (jobs.length === JOBS_ASSET_PAGINATION_SIZE) {
-        await this.jobRepository.queueAll(jobs);
-        jobs = [];
-      }
     }
 
-    await this.jobRepository.queueAll(jobs);
-
     await this.systemMetadataRepository.set(SystemMetadataKey.FacialRecognitionState, { lastRun });
 
     return JobStatus.Success;

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

@@ -1,12 +1,12 @@
 import { Injectable } from '@nestjs/common';
 import { SystemConfig } from 'src/config';
-import { JOBS_ASSET_PAGINATION_SIZE } from 'src/constants';
+
 import { OnEvent, OnJob } from 'src/decorators';
 import { AssetVisibility, DatabaseLock, ImmichWorker, JobName, JobStatus, QueueName } from 'src/enum';
 import { ArgOf } from 'src/repositories/event.repository';
 import { BaseService } from 'src/services/base.service';
-import { JobItem, JobOf } from 'src/types';
+import { JobOf } from 'src/types';
-import { getCLIPModelInfo, isSmartSearchEnabled } from 'src/utils/misc';
+import { batched, getCLIPModelInfo, isSmartSearchEnabled } from 'src/utils/misc';
 
 @Injectable()
 export class SmartInfoService extends BaseService {
@@ -77,18 +77,10 @@ export class SmartInfoService extends BaseService {
       await this.databaseRepository.setDimensionSize(dimSize);
     }
 
-    let queue: JobItem[] = [];
+    for await (const assets of batched(this.assetJobRepository.streamForEncodeClip(force))) {
-    const assets = this.assetJobRepository.streamForEncodeClip(force);
+      await this.jobRepository.queueAll(assets.map((asset) => ({ name: JobName.SmartSearch, data: { id: asset.id } })));
-    for await (const asset of assets) {
-      queue.push({ name: JobName.SmartSearch, data: { id: asset.id } });
-      if (queue.length >= JOBS_ASSET_PAGINATION_SIZE) {
-        await this.jobRepository.queueAll(queue);
-        queue = [];
-      }
     }
 
-    await this.jobRepository.queueAll(queue);
-
     return JobStatus.Success;
   }
 

server/src/services/trash.service.ts

@@ -1,11 +1,11 @@
 import { Injectable } from '@nestjs/common';
-import { JOBS_ASSET_PAGINATION_SIZE } from 'src/constants';
 import { OnEvent, OnJob } from 'src/decorators';
 import { BulkIdsDto } from 'src/dtos/asset-ids.response.dto';
 import { AuthDto } from 'src/dtos/auth.dto';
 import { TrashResponseDto } from 'src/dtos/trash.dto';
 import { JobName, JobStatus, Permission, QueueName } from 'src/enum';
 import { BaseService } from 'src/services/base.service';
+import { batched } from 'src/utils/misc';
 
 @Injectable()
 export class TrashService extends BaseService {
@@ -47,39 +47,16 @@ export class TrashService extends BaseService {
 
   @OnJob({ name: JobName.AssetEmptyTrash, queue: QueueName.BackgroundTask })
   async handleEmptyTrash() {
-    const assets = this.trashRepository.getDeletedIds();
-
     let count = 0;
-    const batch: string[] = [];
+    for await (const assets of batched(this.trashRepository.getDeletedIds())) {
-    for await (const { id } of assets) {
+      await this.jobRepository.queueAll(
-      batch.push(id);
+        assets.map(({ id }) => ({ name: JobName.AssetDelete, data: { id, deleteOnDisk: true } })),
-
+      );
-      if (batch.length === JOBS_ASSET_PAGINATION_SIZE) {
+      count += assets.length;
-        await this.handleBatch(batch);
-        count += batch.length;
-        batch.length = 0;
-      }
     }
 
-    await this.handleBatch(batch);
-    count += batch.length;
-    batch.length = 0;
-
     this.logger.log(`Queued ${count} asset(s) for deletion from the trash`);
 
     return JobStatus.Success;
   }
-
-  private async handleBatch(ids: string[]) {
-    this.logger.debug(`Queueing ${ids.length} asset(s) for deletion from the trash`);
-    await this.jobRepository.queueAll(
-      ids.map((assetId) => ({
-        name: JobName.AssetDelete,
-        data: {
-          id: assetId,
-          deleteOnDisk: true,
-        },
-      })),
-    );
-  }
 }

server/src/utils/misc.ts

@@ -14,7 +14,7 @@ import path from 'node:path';
 import picomatch from 'picomatch';
 import parse from 'picomatch/lib/parse';
 import { SystemConfig } from 'src/config';
-import { CLIP_MODEL_INFO, endpointTags, serverVersion } from 'src/constants';
+import { CLIP_MODEL_INFO, JOBS_ASSET_PAGINATION_SIZE, endpointTags, serverVersion } from 'src/constants';
 import { extraModels } from 'src/decorators';
 import { ApiCustomExtension, ImmichCookie, ImmichHeader, MetadataKey } from 'src/enum';
 import { LoggingRepository } from 'src/repositories/logging.repository';
@@ -120,6 +120,23 @@ export const findOrFail = async <T>(find: () => Promise<T>, entity: string): Pro
   return value;
 };
 
+export async function* batched<T>(items: AsyncIterable<T>, size = JOBS_ASSET_PAGINATION_SIZE): AsyncGenerator<T[]> {
+  let batch: T[] = [];
+
+  for await (const item of items) {
+    batch.push(item);
+
+    if (batch.length >= size) {
+      yield batch;
+      batch = [];
+    }
+  }
+
+  if (batch.length > 0) {
+    yield batch;
+  }
+}
+
 export interface OpenGraphTags {
   title: string;
   description: string;