feat(server): optimize partial facial recognition

server/src/domain/job/job.service.spec.ts

@@ -61,6 +61,7 @@ describe(JobService.name, () => {
         { name: JobName.QUEUE_GENERATE_THUMBNAILS, data: { force: false } },
         { name: JobName.CLEAN_OLD_AUDIT_LOGS },
         { name: JobName.USER_SYNC_USAGE },
+        { name: JobName.QUEUE_FACIAL_RECOGNITION, data: { force: false } },
       ]);
     });
   });
@@ -318,7 +319,7 @@ describe(JobService.name, () => {
       },
       {
         item: { name: JobName.FACE_DETECTION, data: { id: 'asset-1' } },
-        jobs: [JobName.QUEUE_FACIAL_RECOGNITION],
+        jobs: [],
       },
       {
         item: { name: JobName.FACIAL_RECOGNITION, data: { id: 'asset-1' } },

server/src/domain/job/job.service.ts

@@ -174,6 +174,7 @@ export class JobService {
       { name: JobName.QUEUE_GENERATE_THUMBNAILS, data: { force: false } },
       { name: JobName.CLEAN_OLD_AUDIT_LOGS },
       { name: JobName.USER_SYNC_USAGE },
+      { name: JobName.QUEUE_FACIAL_RECOGNITION, data: { force: false } },
     ]);
   }
 
@@ -255,11 +256,6 @@ export class JobService {
         }
         break;
       }
-
-      case JobName.FACE_DETECTION: {
-        await this.jobRepository.queue({ name: JobName.QUEUE_FACIAL_RECOGNITION, data: item.data });
-        break;
-      }
     }
   }
 }

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

@@ -607,14 +607,23 @@ describe(PersonService.name, () => {
 
   describe('handleQueueRecognizeFaces', () => {
     it('should return if machine learning is disabled', async () => {
+      jobMock.getJobCounts.mockResolvedValue({ active: 1, waiting: 0, paused: 0, completed: 0, failed: 0, delayed: 0 });
       configMock.load.mockResolvedValue([{ key: SystemConfigKey.MACHINE_LEARNING_ENABLED, value: false }]);
 
       await expect(sut.handleQueueRecognizeFaces({})).resolves.toBe(true);
-      expect(jobMock.queue).not.toHaveBeenCalled();
+      expect(jobMock.queueAll).not.toHaveBeenCalled();
       expect(configMock.load).toHaveBeenCalled();
     });
 
+    it('should return if recognition jobs are already queued', async () => {
+      jobMock.getJobCounts.mockResolvedValue({ active: 1, waiting: 1, paused: 0, completed: 0, failed: 0, delayed: 0 });
+
+      await expect(sut.handleQueueRecognizeFaces({})).resolves.toBe(true);
+      expect(jobMock.queueAll).not.toHaveBeenCalled();
+    });
+
     it('should queue missing assets', async () => {
+      jobMock.getJobCounts.mockResolvedValue({ active: 1, waiting: 0, paused: 0, completed: 0, failed: 0, delayed: 0 });
       personMock.getAllFaces.mockResolvedValue({
         items: [faceStub.face1],
         hasNextPage: false,
@@ -632,6 +641,7 @@ describe(PersonService.name, () => {
     });
 
     it('should queue all assets', async () => {
+      jobMock.getJobCounts.mockResolvedValue({ active: 1, waiting: 0, paused: 0, completed: 0, failed: 0, delayed: 0 });
       personMock.getAll.mockResolvedValue({
         items: [],
         hasNextPage: false,
@@ -653,6 +663,7 @@ describe(PersonService.name, () => {
     });
 
     it('should delete existing people and faces if forced', async () => {
+      jobMock.getJobCounts.mockResolvedValue({ active: 1, waiting: 0, paused: 0, completed: 0, failed: 0, delayed: 0 });
       personMock.getAll.mockResolvedValue({
         items: [faceStub.face1.person],
         hasNextPage: false,
@@ -738,13 +749,12 @@ describe(PersonService.name, () => {
       expect(assetMock.upsertJobStatus.mock.calls[0][0].facesRecognizedAt?.getTime()).toBeGreaterThan(start);
     });
 
-    it('should create a face with no person', async () => {
+    it('should create a face with no person and queue recognition job', async () => {
+      personMock.createFace.mockResolvedValue(faceStub.face1.id);
       machineLearningMock.detectFaces.mockResolvedValue([detectFaceMock]);
       smartInfoMock.searchFaces.mockResolvedValue([{ face: faceStub.face1, distance: 0.7 }]);
       assetMock.getByIds.mockResolvedValue([assetStub.image]);
-      await sut.handleDetectFaces({ id: assetStub.image.id });
-
-      expect(personMock.createFace).toHaveBeenCalledWith({
+      const face = {
         assetId: 'asset-id',
         embedding: [1, 2, 3, 4],
         boundingBoxX1: 100,
@@ -753,7 +763,14 @@ describe(PersonService.name, () => {
         boundingBoxY2: 200,
         imageHeight: 500,
         imageWidth: 400,
-      });
+      };
+
+      await sut.handleDetectFaces({ id: assetStub.image.id });
+
+      expect(personMock.createFace).toHaveBeenCalledWith(face);
+      expect(jobMock.queueAll).toHaveBeenCalledWith([
+        { name: JobName.FACIAL_RECOGNITION, data: { id: faceStub.face1.id } },
+      ]);
       expect(personMock.reassignFace).not.toHaveBeenCalled();
       expect(personMock.reassignFaces).not.toHaveBeenCalled();
     });

server/src/domain/person/person.service.ts

@@ -332,19 +332,26 @@ export class PersonService {
     this.logger.debug(`${faces.length} faces detected in ${asset.resizePath}`);
     this.logger.verbose(faces.map((face) => ({ ...face, embedding: `vector(${face.embedding.length})` })));
 
-    for (const face of faces) {
-      const mappedFace = {
-        assetId: asset.id,
-        embedding: face.embedding,
-        imageHeight: face.imageHeight,
-        imageWidth: face.imageWidth,
-        boundingBoxX1: face.boundingBox.x1,
-        boundingBoxX2: face.boundingBox.x2,
-        boundingBoxY1: face.boundingBox.y1,
-        boundingBoxY2: face.boundingBox.y2,
-      };
+    if (faces.length) {
+      await this.jobRepository.queue({ name: JobName.QUEUE_FACIAL_RECOGNITION, data: { force: false } });
+
+      const createFaces = faces.map((face) => {
+        const mappedFace = {
+          assetId: asset.id,
+          embedding: face.embedding,
+          imageHeight: face.imageHeight,
+          imageWidth: face.imageWidth,
+          boundingBoxX1: face.boundingBox.x1,
+          boundingBoxX2: face.boundingBox.x2,
+          boundingBoxY1: face.boundingBox.y1,
+          boundingBoxY2: face.boundingBox.y2,
+        };
+
+        return this.repository.createFace(mappedFace);
+      });
 
-      await this.repository.createFace(mappedFace);
+      const faceIds = await Promise.all(createFaces);
+      await this.jobRepository.queueAll(faceIds.map((id) => ({ name: JobName.FACIAL_RECOGNITION, data: { id } })));
     }
 
     await this.assetRepository.upsertJobStatus({
@@ -362,9 +369,15 @@ export class PersonService {
     }
 
     await this.jobRepository.waitForQueueCompletion(QueueName.THUMBNAIL_GENERATION, QueueName.FACE_DETECTION);
+    const { waiting } = await this.jobRepository.getJobCounts(QueueName.FACIAL_RECOGNITION);
 
     if (force) {
       await this.deleteAllPeople();
+    } else if (waiting) {
+      this.logger.debug(
+        `Skipping facial recognition queueing because ${waiting} job${waiting > 1 ? 's are' : ' is'} already queued`,
+      );
+      return true;
     }
 
     const facePagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (pagination) =>

server/src/domain/repositories/person.repository.ts

@@ -38,7 +38,7 @@ export interface IPersonRepository {
   getAssets(personId: string): Promise<AssetEntity[]>;
 
   create(entity: Partial<PersonEntity>): Promise<PersonEntity>;
-  createFace(entity: Partial<AssetFaceEntity>): Promise<void>;
+  createFace(entity: Partial<AssetFaceEntity>): Promise<string>;
   delete(entities: PersonEntity[]): Promise<void>;
   deleteAll(): Promise<void>;
   deleteAllFaces(): Promise<void>;

server/src/infra/repositories/person.repository.ts

@@ -215,11 +215,12 @@ export class PersonRepository implements IPersonRepository {
     return this.personRepository.save(entity);
   }
 
-  async createFace(entity: AssetFaceEntity): Promise<void> {
+  async createFace(entity: AssetFaceEntity): Promise<string> {
     if (!entity.embedding) {
       throw new Error('Embedding is required to create a face');
     }
-    await this.assetFaceRepository.insert({ ...entity, embedding: () => asVector(entity.embedding, true) });
+    const res = await this.assetFaceRepository.insert({ ...entity, embedding: () => asVector(entity.embedding, true) });
+    return res.identifiers[0].id;
   }
 
   async update(entity: Partial<PersonEntity>): Promise<PersonEntity> {