refactor: stream detect faces (#17996)

server/src/queries/asset.job.repository.sql

@@ -469,3 +469,17 @@ from
   "assets"
 where
   "assets"."deletedAt" <= $1
+
+-- AssetJobRepository.streamForDetectFacesJob
+select
+  "assets"."id"
+from
+  "assets"
+  inner join "asset_job_status" as "job_status" on "assetId" = "assets"."id"
+where
+  "assets"."isVisible" = $1
+  and "assets"."deletedAt" is null
+  and "job_status"."previewAt" is not null
+  and "job_status"."facesRecognizedAt" is null
+order by
+  "assets"."createdAt" desc

server/src/repositories/asset-job.repository.ts

@@ -322,4 +322,13 @@ export class AssetJobRepository {
       .where('assets.deletedAt', '<=', trashedBefore)
       .stream();
   }
+
+  @GenerateSql({ params: [], stream: true })
+  streamForDetectFacesJob(force?: boolean) {
+    return this.assetsWithPreviews()
+      .$if(!force, (qb) => qb.where('job_status.facesRecognizedAt', 'is', null))
+      .select(['assets.id'])
+      .orderBy('assets.createdAt', 'desc')
+      .stream();
+  }
 }

server/src/repositories/asset.repository.ts

@@ -48,8 +48,6 @@ export interface LivePhotoSearchOptions {
 export enum WithoutProperty {
   THUMBNAIL = 'thumbnail',
   ENCODED_VIDEO = 'encoded-video',
-  EXIF = 'exif',
-  FACES = 'faces',
   SIDECAR = 'sidecar',
 }
 
@@ -543,19 +541,7 @@ export class AssetRepository {
           .where('assets.type', '=', AssetType.VIDEO)
           .where((eb) => eb.or([eb('assets.encodedVideoPath', 'is', null), eb('assets.encodedVideoPath', '=', '')])),
       )
-      .$if(property === WithoutProperty.EXIF, (qb) =>
-        qb
-          .leftJoin('asset_job_status as job_status', 'assets.id', 'job_status.assetId')
-          .where((eb) => eb.or([eb('job_status.metadataExtractedAt', 'is', null), eb('assetId', 'is', null)]))
-          .where('assets.isVisible', '=', true),
-      )
-      .$if(property === WithoutProperty.FACES, (qb) =>
-        qb
-          .innerJoin('asset_job_status as job_status', 'assetId', 'assets.id')
-          .where('job_status.previewAt', 'is not', null)
-          .where('job_status.facesRecognizedAt', 'is', null)
-          .where('assets.isVisible', '=', true),
-      )
+
       .$if(property === WithoutProperty.SIDECAR, (qb) =>
         qb
           .where((eb) => eb.or([eb('assets.sidecarPath', '=', ''), eb('assets.sidecarPath', 'is', null)]))

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

@@ -2,7 +2,6 @@ import { BadRequestException, NotFoundException } from '@nestjs/common';
 import { BulkIdErrorReason } from 'src/dtos/asset-ids.response.dto';
 import { mapFaces, mapPerson, PersonResponseDto } from 'src/dtos/person.dto';
 import { CacheControl, Colorspace, ImageFormat, JobName, JobStatus, SourceType, SystemMetadataKey } from 'src/enum';
-import { WithoutProperty } from 'src/repositories/asset.repository';
 import { DetectedFaces } from 'src/repositories/machine-learning.repository';
 import { FaceSearchResult } from 'src/repositories/search.repository';
 import { PersonService } from 'src/services/person.service';
@@ -455,14 +454,11 @@ describe(PersonService.name, () => {
     });
 
     it('should queue missing assets', async () => {
-      mocks.asset.getWithout.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
+      mocks.assetJob.streamForDetectFacesJob.mockReturnValue(makeStream([assetStub.image]));
 
       await sut.handleQueueDetectFaces({ force: false });
 
-      expect(mocks.asset.getWithout).toHaveBeenCalledWith({ skip: 0, take: 1000 }, WithoutProperty.FACES);
+      expect(mocks.assetJob.streamForDetectFacesJob).toHaveBeenCalledWith(false);
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         {
           name: JobName.FACE_DETECTION,
@@ -472,10 +468,7 @@ describe(PersonService.name, () => {
     });
 
     it('should queue all assets', async () => {
-      mocks.asset.getAll.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
+      mocks.assetJob.streamForDetectFacesJob.mockReturnValue(makeStream([assetStub.image]));
       mocks.person.getAllWithoutFaces.mockResolvedValue([personStub.withName]);
 
       await sut.handleQueueDetectFaces({ force: true });
@@ -483,7 +476,7 @@ describe(PersonService.name, () => {
       expect(mocks.person.deleteFaces).toHaveBeenCalledWith({ sourceType: SourceType.MACHINE_LEARNING });
       expect(mocks.person.delete).toHaveBeenCalledWith([personStub.withName.id]);
       expect(mocks.storage.unlink).toHaveBeenCalledWith(personStub.withName.thumbnailPath);
-      expect(mocks.asset.getAll).toHaveBeenCalled();
+      expect(mocks.assetJob.streamForDetectFacesJob).toHaveBeenCalledWith(true);
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         {
           name: JobName.FACE_DETECTION,
@@ -493,17 +486,14 @@ describe(PersonService.name, () => {
     });
 
     it('should refresh all assets', async () => {
-      mocks.asset.getAll.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
+      mocks.assetJob.streamForDetectFacesJob.mockReturnValue(makeStream([assetStub.image]));
 
       await sut.handleQueueDetectFaces({ force: undefined });
 
       expect(mocks.person.delete).not.toHaveBeenCalled();
       expect(mocks.person.deleteFaces).not.toHaveBeenCalled();
       expect(mocks.storage.unlink).not.toHaveBeenCalled();
-      expect(mocks.asset.getAll).toHaveBeenCalled();
+      expect(mocks.assetJob.streamForDetectFacesJob).toHaveBeenCalledWith(undefined);
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         {
           name: JobName.FACE_DETECTION,
@@ -516,16 +506,13 @@ describe(PersonService.name, () => {
     it('should delete existing people and faces if forced', async () => {
       mocks.person.getAll.mockReturnValue(makeStream([faceStub.face1.person, personStub.randomPerson]));
       mocks.person.getAllFaces.mockReturnValue(makeStream([faceStub.face1]));
-      mocks.asset.getAll.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
+      mocks.assetJob.streamForDetectFacesJob.mockReturnValue(makeStream([assetStub.image]));
       mocks.person.getAllWithoutFaces.mockResolvedValue([personStub.randomPerson]);
       mocks.person.deleteFaces.mockResolvedValue();
 
       await sut.handleQueueDetectFaces({ force: true });
 
-      expect(mocks.asset.getAll).toHaveBeenCalled();
+      expect(mocks.assetJob.streamForDetectFacesJob).toHaveBeenCalledWith(true);
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         {
           name: JobName.FACE_DETECTION,

server/src/services/person.service.ts

@@ -36,7 +36,6 @@ import {
   SourceType,
   SystemMetadataKey,
 } from 'src/enum';
-import { WithoutProperty } from 'src/repositories/asset.repository';
 import { BoundingBox } from 'src/repositories/machine-learning.repository';
 import { UpdateFacesData } from 'src/repositories/person.repository';
 import { BaseService } from 'src/services/base.service';
@@ -44,7 +43,6 @@ import { CropOptions, ImageDimensions, InputDimensions, JobItem, JobOf } from 's
 import { ImmichFileResponse } from 'src/utils/file';
 import { mimeTypes } from 'src/utils/mime-types';
 import { isFaceImportEnabled, isFacialRecognitionEnabled } from 'src/utils/misc';
-import { usePagination } from 'src/utils/pagination';
 
 @Injectable()
 export class PersonService extends BaseService {
@@ -265,23 +263,19 @@ export class PersonService extends BaseService {
       await this.handlePersonCleanup();
     }
 
-    const assetPagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (pagination) => {
-      return force === false
-        ? this.assetRepository.getWithout(pagination, WithoutProperty.FACES)
-        : this.assetRepository.getAll(pagination, {
-            orderDirection: 'desc',
-            withFaces: true,
-            withArchived: true,
-            isVisible: true,
-          });
-    });
+    let jobs: JobItem[] = [];
+    const assets = this.assetJobRepository.streamForDetectFacesJob(force);
+    for await (const asset of assets) {
+      jobs.push({ name: JobName.FACE_DETECTION, data: { id: asset.id } });
 
-    for await (const assets of assetPagination) {
-      await this.jobRepository.queueAll(
-        assets.map((asset) => ({ name: JobName.FACE_DETECTION, 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.PERSON_CLEANUP });
     }