refactor: stream for sidecar (#17995)

* refactor: stream for sidecar

* chore: make sql

---------

Co-authored-by: Jason Rasmussen <jason@rasm.me>

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

@@ -470,6 +470,18 @@ from
 where
   "assets"."deletedAt" <= $1
 
+-- AssetJobRepository.streamForSidecar
+select
+  "assets"."id"
+from
+  "assets"
+where
+  (
+    "assets"."sidecarPath" = $1
+    or "assets"."sidecarPath" is null
+  )
+  and "assets"."isVisible" = $2
+
 -- AssetJobRepository.streamForDetectFacesJob
 select
   "assets"."id"

server/src/queries/asset.repository.sql

@@ -238,18 +238,13 @@ select
 from
   "assets"
 where
-  (
-    "assets"."sidecarPath" = $1
-    or "assets"."sidecarPath" is null
-  )
-  and "assets"."isVisible" = $2
-  and "deletedAt" is null
+  "deletedAt" is null
 order by
   "createdAt"
 limit
-  $3
+  $1
 offset
-  $4
+  $2
 
 -- AssetRepository.getTimeBuckets
 with

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

@@ -323,6 +323,18 @@ export class AssetJobRepository {
       .stream();
   }
 
+  @GenerateSql({ params: [], stream: true })
+  streamForSidecar(force?: boolean) {
+    return this.db
+      .selectFrom('assets')
+      .select(['assets.id'])
+      .$if(!force, (qb) =>
+        qb.where((eb) => eb.or([eb('assets.sidecarPath', '=', ''), eb('assets.sidecarPath', 'is', null)])),
+      )
+      .where('assets.isVisible', '=', true)
+      .stream();
+  }
+
   @GenerateSql({ params: [], stream: true })
   streamForDetectFacesJob(force?: boolean) {
     return this.assetsWithPreviews()

server/src/repositories/asset.repository.ts

@@ -48,7 +48,6 @@ export interface LivePhotoSearchOptions {
 export enum WithoutProperty {
   THUMBNAIL = 'thumbnail',
   ENCODED_VIDEO = 'encoded-video',
-  SIDECAR = 'sidecar',
 }
 
 export enum WithProperty {
@@ -542,11 +541,6 @@ export class AssetRepository {
           .where((eb) => eb.or([eb('assets.encodedVideoPath', 'is', null), eb('assets.encodedVideoPath', '=', '')])),
       )
 
-      .$if(property === WithoutProperty.SIDECAR, (qb) =>
-        qb
-          .where((eb) => eb.or([eb('assets.sidecarPath', '=', ''), eb('assets.sidecarPath', 'is', null)]))
-          .where('assets.isVisible', '=', true),
-      )
       .$if(property === WithoutProperty.THUMBNAIL, (qb) =>
         qb
           .innerJoin('asset_job_status as job_status', 'assetId', 'assets.id')

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

@@ -5,7 +5,6 @@ import { constants } from 'node:fs/promises';
 import { defaults } from 'src/config';
 import { MapAsset } from 'src/dtos/asset-response.dto';
 import { AssetType, ExifOrientation, ImmichWorker, JobName, JobStatus, SourceType } from 'src/enum';
-import { WithoutProperty } from 'src/repositories/asset.repository';
 import { ImmichTags } from 'src/repositories/metadata.repository';
 import { MetadataService } from 'src/services/metadata.service';
 import { assetStub } from 'test/fixtures/asset.stub';
@@ -1346,12 +1345,11 @@ describe(MetadataService.name, () => {
 
   describe('handleQueueSidecar', () => {
     it('should queue assets with sidecar files', async () => {
-      mocks.asset.getAll.mockResolvedValue({ items: [assetStub.sidecar], hasNextPage: false });
+      mocks.assetJob.streamForSidecar.mockReturnValue(makeStream([assetStub.image]));
 
       await sut.handleQueueSidecar({ force: true });
+      expect(mocks.assetJob.streamForSidecar).toHaveBeenCalledWith(true);
 
-      expect(mocks.asset.getAll).toHaveBeenCalledWith({ take: 1000, skip: 0 });
-      expect(mocks.asset.getWithout).not.toHaveBeenCalled();
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         {
           name: JobName.SIDECAR_SYNC,
@@ -1361,12 +1359,11 @@ describe(MetadataService.name, () => {
     });
 
     it('should queue assets without sidecar files', async () => {
-      mocks.asset.getWithout.mockResolvedValue({ items: [assetStub.image], hasNextPage: false });
+      mocks.assetJob.streamForSidecar.mockReturnValue(makeStream([assetStub.image]));
 
       await sut.handleQueueSidecar({ force: false });
 
-      expect(mocks.asset.getWithout).toHaveBeenCalledWith({ take: 1000, skip: 0 }, WithoutProperty.SIDECAR);
-      expect(mocks.asset.getAll).not.toHaveBeenCalled();
+      expect(mocks.assetJob.streamForSidecar).toHaveBeenCalledWith(false);
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
         {
           name: JobName.SIDECAR_DISCOVERY,

server/src/services/metadata.service.ts

@@ -22,14 +22,12 @@ import {
   QueueName,
   SourceType,
 } from 'src/enum';
-import { WithoutProperty } from 'src/repositories/asset.repository';
 import { ArgOf } from 'src/repositories/event.repository';
 import { ReverseGeocodeResult } from 'src/repositories/map.repository';
 import { ImmichTags } from 'src/repositories/metadata.repository';
 import { BaseService } from 'src/services/base.service';
-import { JobOf } from 'src/types';
+import { JobItem, JobOf } from 'src/types';
 import { isFaceImportEnabled } from 'src/utils/misc';
-import { usePagination } from 'src/utils/pagination';
 import { upsertTags } from 'src/utils/tag';
 
 /** look for a date from these tags (in order) */
@@ -289,23 +287,23 @@ export class MetadataService extends BaseService {
   }
 
   @OnJob({ name: JobName.QUEUE_SIDECAR, queue: QueueName.SIDECAR })
-  async handleQueueSidecar(job: JobOf<JobName.QUEUE_SIDECAR>): Promise<JobStatus> {
-    const { force } = job;
-    const assetPagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (pagination) => {
-      return force
-        ? this.assetRepository.getAll(pagination)
-        : this.assetRepository.getWithout(pagination, WithoutProperty.SIDECAR);
-    });
+  async handleQueueSidecar({ force }: JobOf<JobName.QUEUE_SIDECAR>): Promise<JobStatus> {
+    let jobs: JobItem[] = [];
+    const queueAll = async () => {
+      await this.jobRepository.queueAll(jobs);
+      jobs = [];
+    };
 
-    for await (const assets of assetPagination) {
-      await this.jobRepository.queueAll(
-        assets.map((asset) => ({
-          name: force ? JobName.SIDECAR_SYNC : JobName.SIDECAR_DISCOVERY,
-          data: { id: asset.id },
-        })),
-      );
+    const assets = this.assetJobRepository.streamForSidecar(force);
+    for await (const asset of assets) {
+      jobs.push({ name: force ? JobName.SIDECAR_SYNC : JobName.SIDECAR_DISCOVERY, data: { id: asset.id } });
+      if (jobs.length >= JOBS_ASSET_PAGINATION_SIZE) {
+        await queueAll();
+      }
     }
 
+    await queueAll();
+
     return JobStatus.SUCCESS;
   }