refactor: smart search queue (#17977)

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

@@ -194,6 +194,24 @@ where
   "asset_files"."assetId" = $1
   and "asset_files"."type" = $2
 
+-- AssetJobRepository.streamForEncodeClip
+select
+  "assets"."id"
+from
+  "assets"
+  inner join "asset_job_status" as "job_status" on "assetId" = "assets"."id"
+where
+  "job_status"."previewAt" is not null
+  and "assets"."isVisible" = $1
+  and not exists (
+    select
+    from
+      "smart_search"
+    where
+      "assetId" = "assets"."id"
+  )
+  and "assets"."deletedAt" is null
+
 -- AssetJobRepository.getForClipEncoding
 select
   "assets"."id",

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

@@ -135,6 +135,23 @@ export class AssetJobRepository {
       .execute();
   }
 
+  @GenerateSql({ params: [], stream: true })
+  streamForEncodeClip(force?: boolean) {
+    return this.db
+      .selectFrom('assets')
+      .select(['assets.id'])
+      .innerJoin('asset_job_status as job_status', 'assetId', 'assets.id')
+      .where('job_status.previewAt', 'is not', null)
+      .where('assets.isVisible', '=', true)
+      .$if(!force, (qb) =>
+        qb.where((eb) =>
+          eb.not((eb) => eb.exists(eb.selectFrom('smart_search').whereRef('assetId', '=', 'assets.id'))),
+        ),
+      )
+      .where('assets.deletedAt', 'is', null)
+      .stream();
+  }
+
   @GenerateSql({ params: [DummyValue.UUID] })
   getForClipEncoding(id: string) {
     return this.db

server/src/repositories/asset.repository.ts

@@ -49,7 +49,6 @@ export enum WithoutProperty {
   THUMBNAIL = 'thumbnail',
   ENCODED_VIDEO = 'encoded-video',
   EXIF = 'exif',
-  SMART_SEARCH = 'smart-search',
   DUPLICATE = 'duplicate',
   FACES = 'faces',
   SIDECAR = 'sidecar',
@@ -571,15 +570,6 @@ export class AssetRepository {
           .where((eb) => eb.or([eb('assets.sidecarPath', '=', ''), eb('assets.sidecarPath', 'is', null)]))
           .where('assets.isVisible', '=', true),
       )
-      .$if(property === WithoutProperty.SMART_SEARCH, (qb) =>
-        qb
-          .innerJoin('asset_job_status as job_status', 'assetId', 'assets.id')
-          .where('job_status.previewAt', 'is not', null)
-          .where('assets.isVisible', '=', true)
-          .where((eb) =>
-            eb.not((eb) => eb.exists(eb.selectFrom('smart_search').whereRef('assetId', '=', 'assets.id'))),
-          ),
-      )
       .$if(property === WithoutProperty.THUMBNAIL, (qb) =>
         qb
           .innerJoin('asset_job_status as job_status', 'assetId', 'assets.id')

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

@@ -1,11 +1,10 @@
 import { SystemConfig } from 'src/config';
 import { ImmichWorker, JobName, JobStatus } from 'src/enum';
-import { WithoutProperty } from 'src/repositories/asset.repository';
 import { SmartInfoService } from 'src/services/smart-info.service';
 import { getCLIPModelInfo } from 'src/utils/misc';
 import { assetStub } from 'test/fixtures/asset.stub';
 import { systemConfigStub } from 'test/fixtures/system-config.stub';
-import { newTestService, ServiceMocks } from 'test/utils';
+import { makeStream, newTestService, ServiceMocks } from 'test/utils';
 
 describe(SmartInfoService.name, () => {
   let sut: SmartInfoService;
@@ -152,38 +151,31 @@ describe(SmartInfoService.name, () => {
 
       await sut.handleQueueEncodeClip({});
 
-      expect(mocks.asset.getAll).not.toHaveBeenCalled();
       expect(mocks.asset.getWithout).not.toHaveBeenCalled();
       expect(mocks.search.setDimensionSize).not.toHaveBeenCalled();
     });
 
     it('should queue the assets without clip embeddings', async () => {
-      mocks.asset.getWithout.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
+      mocks.assetJob.streamForEncodeClip.mockReturnValue(makeStream([assetStub.image]));
 
       await sut.handleQueueEncodeClip({ force: false });
 
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         { name: JobName.SMART_SEARCH, data: { id: assetStub.image.id } },
       ]);
-      expect(mocks.asset.getWithout).toHaveBeenCalledWith({ skip: 0, take: 1000 }, WithoutProperty.SMART_SEARCH);
+      expect(mocks.assetJob.streamForEncodeClip).toHaveBeenCalledWith(false);
       expect(mocks.search.setDimensionSize).not.toHaveBeenCalled();
     });
 
     it('should queue all the assets', async () => {
-      mocks.asset.getAll.mockResolvedValue({
-        items: [assetStub.image],
-        hasNextPage: false,
-      });
+      mocks.assetJob.streamForEncodeClip.mockReturnValue(makeStream([assetStub.image]));
 
       await sut.handleQueueEncodeClip({ force: true });
 
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         { name: JobName.SMART_SEARCH, data: { id: assetStub.image.id } },
       ]);
-      expect(mocks.asset.getAll).toHaveBeenCalled();
+      expect(mocks.assetJob.streamForEncodeClip).toHaveBeenCalledWith(true);
       expect(mocks.search.setDimensionSize).toHaveBeenCalledExactlyOnceWith(512);
     });
   });

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

@@ -3,12 +3,10 @@ import { SystemConfig } from 'src/config';
 import { JOBS_ASSET_PAGINATION_SIZE } from 'src/constants';
 import { OnEvent, OnJob } from 'src/decorators';
 import { DatabaseLock, ImmichWorker, JobName, JobStatus, QueueName } from 'src/enum';
-import { WithoutProperty } from 'src/repositories/asset.repository';
 import { ArgOf } from 'src/repositories/event.repository';
 import { BaseService } from 'src/services/base.service';
-import { JobOf } from 'src/types';
+import { JobItem, JobOf } from 'src/types';
 import { getCLIPModelInfo, isSmartSearchEnabled } from 'src/utils/misc';
-import { usePagination } from 'src/utils/pagination';
 
 @Injectable()
 export class SmartInfoService extends BaseService {
@@ -79,17 +77,17 @@ export class SmartInfoService extends BaseService {
       await this.searchRepository.setDimensionSize(dimSize);
     }
 
-    const assetPagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (pagination) => {
-      return force
-        ? this.assetRepository.getAll(pagination, { isVisible: true })
-        : this.assetRepository.getWithout(pagination, WithoutProperty.SMART_SEARCH);
-    });
-
-    for await (const assets of assetPagination) {
-      await this.jobRepository.queueAll(
-        assets.map((asset) => ({ name: JobName.SMART_SEARCH, data: { id: asset.id } })),
-      );
+    let queue: JobItem[] = [];
+    const assets = this.assetJobRepository.streamForEncodeClip(force);
+    for await (const asset of assets) {
+      queue.push({ name: JobName.SMART_SEARCH, 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;
   }