diff --git a/server/src/services/asset.service.ts b/server/src/services/asset.service.ts index 72b5ad997b..15385df0e0 100644 --- a/server/src/services/asset.service.ts +++ b/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 }> = []; - const queueChunk = async () => { - if (chunk.length === 0) { - return; - } - + for await (const assets of batched(this.assetJobRepository.streamForDeletedJob(trashedBefore))) { await this.jobRepository.queueAll( - chunk.map(({ id, isOffline }) => ({ - name: JobName.AssetDelete, - data: { id, deleteOnDisk: !isOffline }, - })), + assets.map(({ id, 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; } diff --git a/server/src/services/duplicate.service.ts b/server/src/services/duplicate.service.ts index 8cb04e29ba..08c7630f91 100644 --- a/server/src/services/duplicate.service.ts +++ b/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[] = []; - const queueAll = async () => { - await this.jobRepository.queueAll(jobs); - 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(); - } + for await (const assets of batched(this.assetJobRepository.streamForSearchDuplicates(force))) { + await this.jobRepository.queueAll( + assets.map((asset) => ({ name: JobName.AssetDetectDuplicates, data: { id: asset.id } })), + ); } - await queueAll(); - return JobStatus.Success; } diff --git a/server/src/services/integrity.service.ts b/server/src/services/integrity.service.ts index a7a194b25f..6c7a505812 100644 --- a/server/src/services/integrity.service.ts +++ b/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(generator: AsyncIterableIterator, 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; - } -} diff --git a/server/src/services/library.service.ts b/server/src/services/library.service.ts index 56e5fa6be8..9e5e0cf4d1 100644 --- a/server/src/services/library.service.ts +++ b/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); - for await (const asset of assets) { - chunk.push(asset.id); - - if (chunk.length >= JOBS_LIBRARY_PAGINATION_SIZE) { - await queueChunk(); - } + let hasAssets = false; + for await (const assets of batched( + this.libraryRepository.streamAssetIds(libraryId), + JOBS_LIBRARY_PAGINATION_SIZE, + )) { + 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 (!isAssetsFound) { + if (!hasAssets) { 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 queueChunk = async () => { - if (chunk.length === 0) { - return; - } - - count += chunk.length; + const existingAssets = this.libraryRepository.streamAssetIds(library.id); + for await (const assets of batched(existingAssets, JOBS_LIBRARY_PAGINATION_SIZE)) { + count += assets.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; diff --git a/server/src/services/media.service.spec.ts b/server/src/services/media.service.spec.ts index 290bf947a1..d64e0088eb 100644 --- a/server/src/services/media.service.spec.ts +++ b/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: '' }); }); diff --git a/server/src/services/media.service.ts b/server/src/services/media.service.ts index 956636dd51..19b1688e88 100644 --- a/server/src/services/media.service.ts +++ b/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): Promise { 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({ - force, - fullsizeEnabled: isFullsizeEnabled, - })) { - 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; + for await (const assets of batched( + this.assetJobRepository.streamForThumbnailJob({ force, fullsizeEnabled: isFullsizeEnabled }), + )) { + const jobs: JobItem[] = []; + for (const asset of assets) { + if (force || !asset.isEdited) { + jobs.push({ name: JobName.AssetGenerateThumbnails, data: { id: asset.id } }); } - 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 } }); - if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) { - await queueAll(); - } + await this.jobRepository.queueAll(jobs); } - 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[] = []; - const assets = this.assetJobRepository.streamForMigrationJob(); - for await (const asset of assets) { - jobs.push({ name: JobName.AssetFileMigration, data: { id: asset.id } }); - if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) { - await this.jobRepository.queueAll(jobs); - jobs = []; - } + for await (const assets of batched(this.assetJobRepository.streamForMigrationJob())) { + await this.jobRepository.queueAll( + assets.map((asset) => ({ name: JobName.AssetFileMigration, data: { id: asset.id } })), + ); } - await this.jobRepository.queueAll(jobs); - jobs = []; - - 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 = []; - } + for await (const people of batched(this.personRepository.getAll())) { + await this.jobRepository.queueAll( + people.map((person) => ({ name: JobName.PersonFileMigration, data: { id: person.id } })), + ); } - await this.jobRepository.queueAll(jobs); - return JobStatus.Success; } @@ -551,18 +529,12 @@ export class MediaService extends BaseService { async handleQueueVideoConversion(job: JobOf): Promise { const { force } = job; - let queue: { name: JobName.AssetEncodeVideo; data: { id: string } }[] = []; - for await (const asset of this.assetJobRepository.streamForVideoConversion(force)) { - queue.push({ name: JobName.AssetEncodeVideo, data: { id: asset.id } }); - - if (queue.length >= JOBS_ASSET_PAGINATION_SIZE) { - await this.jobRepository.queueAll(queue); - queue = []; - } + for await (const assets of batched(this.assetJobRepository.streamForVideoConversion(force))) { + await this.jobRepository.queueAll( + assets.map((asset) => ({ name: JobName.AssetEncodeVideo, data: { id: asset.id } })), + ); } - await this.jobRepository.queueAll(queue); - return JobStatus.Success; } diff --git a/server/src/services/metadata.service.ts b/server/src/services/metadata.service.ts index 37dd92e27d..53573549eb 100644 --- a/server/src/services/metadata.service.ts +++ b/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): Promise { const { force } = job; - let queue: { name: JobName.AssetExtractMetadata; data: { id: string } }[] = []; - for await (const asset of this.assetJobRepository.streamForMetadataExtraction(force)) { - queue.push({ name: JobName.AssetExtractMetadata, data: { id: asset.id } }); - - if (queue.length >= JOBS_ASSET_PAGINATION_SIZE) { - await this.jobRepository.queueAll(queue); - queue = []; - } + for await (const assets of batched(this.assetJobRepository.streamForMetadataExtraction(force))) { + await this.jobRepository.queueAll( + assets.map((asset) => ({ name: JobName.AssetExtractMetadata, data: { id: asset.id } })), + ); } - 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): Promise { - let jobs: JobItem[] = []; - const queueAll = async () => { - await this.jobRepository.queueAll(jobs); - 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(); - } + for await (const assets of batched(this.assetJobRepository.streamForSidecar(force))) { + await this.jobRepository.queueAll( + assets.map((asset) => ({ name: JobName.SidecarCheck, data: { id: asset.id } })), + ); } - await queueAll(); - return JobStatus.Success; } diff --git a/server/src/services/ocr.service.ts b/server/src/services/ocr.service.ts index d92d399dba..8e32a44376 100644 --- a/server/src/services/ocr.service.ts +++ b/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[] = []; - const assets = this.assetJobRepository.streamForOcrJob(force); - - 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 = []; - } + for await (const assets of batched(this.assetJobRepository.streamForOcrJob(force))) { + await this.jobRepository.queueAll(assets.map((asset) => ({ name: JobName.Ocr, data: { id: asset.id } }))); } - await this.jobRepository.queueAll(jobs); return JobStatus.Success; } diff --git a/server/src/services/person.service.ts b/server/src/services/person.service.ts index ff381ce736..7c08c36e17 100644 --- a/server/src/services/person.service.ts +++ b/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[] = []; - const assets = this.assetJobRepository.streamForDetectFacesJob(force); - for await (const asset of assets) { - jobs.push({ name: JobName.AssetDetectFaces, data: { id: asset.id } }); - - if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) { - await this.jobRepository.queueAll(jobs); - jobs = []; - } + for await (const assets of batched(this.assetJobRepository.streamForDetectFacesJob(force))) { + await this.jobRepository.queueAll( + assets.map((asset) => ({ name: JobName.AssetDetectFaces, data: { id: asset.id } })), + ); } - 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 }, ); - - let jobs: { name: JobName.FacialRecognition; data: { id: string; deferred: false } }[] = []; - for await (const face of facePagination) { - jobs.push({ name: JobName.FacialRecognition, data: { id: face.id, deferred: false } }); - - if (jobs.length === JOBS_ASSET_PAGINATION_SIZE) { - await this.jobRepository.queueAll(jobs); - jobs = []; - } + for await (const batch of batched(faces)) { + await this.jobRepository.queueAll( + batch.map((face) => ({ name: JobName.FacialRecognition, data: { id: face.id, deferred: false } })), + ); } - await this.jobRepository.queueAll(jobs); - await this.systemMetadataRepository.set(SystemMetadataKey.FacialRecognitionState, { lastRun }); return JobStatus.Success; diff --git a/server/src/services/smart-info.service.ts b/server/src/services/smart-info.service.ts index 93b8944099..19a1774492 100644 --- a/server/src/services/smart-info.service.ts +++ b/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 { getCLIPModelInfo, isSmartSearchEnabled } from 'src/utils/misc'; +import { JobOf } from 'src/types'; +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[] = []; - const assets = this.assetJobRepository.streamForEncodeClip(force); - 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 = []; - } + for await (const assets of batched(this.assetJobRepository.streamForEncodeClip(force))) { + await this.jobRepository.queueAll(assets.map((asset) => ({ name: JobName.SmartSearch, data: { id: asset.id } }))); } - await this.jobRepository.queueAll(queue); - return JobStatus.Success; } diff --git a/server/src/services/trash.service.ts b/server/src/services/trash.service.ts index f1d368111e..368e16beb5 100644 --- a/server/src/services/trash.service.ts +++ b/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 { id } of assets) { - batch.push(id); - - if (batch.length === JOBS_ASSET_PAGINATION_SIZE) { - await this.handleBatch(batch); - count += batch.length; - batch.length = 0; - } + for await (const assets of batched(this.trashRepository.getDeletedIds())) { + await this.jobRepository.queueAll( + assets.map(({ id }) => ({ name: JobName.AssetDelete, data: { id, deleteOnDisk: true } })), + ); + count += assets.length; } - 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, - }, - })), - ); - } } diff --git a/server/src/utils/misc.ts b/server/src/utils/misc.ts index 4db4140c1b..a514007987 100644 --- a/server/src/utils/misc.ts +++ b/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 (find: () => Promise, entity: string): Pro return value; }; +export async function* batched(items: AsyncIterable, size = JOBS_ASSET_PAGINATION_SIZE): AsyncGenerator { + 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;