refactor: queue asset deletes via stream (#16706)

server/src/repositories/asset.repository.ts

@@ -637,6 +637,14 @@ export class AssetRepository {
     return this.storageTemplateAssetQuery().stream() as AsyncIterableIterator<StorageAsset>;
   }
 
+  streamDeletedAssets(trashedBefore: Date) {
+    return this.db
+      .selectFrom('assets')
+      .select(['id', 'isOffline'])
+      .where('assets.deletedAt', '<=', trashedBefore)
+      .stream();
+  }
+
   @GenerateSql(
     ...Object.values(WithProperty).map((property) => ({
       name: property,

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

@@ -11,7 +11,8 @@ import { authStub } from 'test/fixtures/auth.stub';
 import { faceStub } from 'test/fixtures/face.stub';
 import { partnerStub } from 'test/fixtures/partner.stub';
 import { userStub } from 'test/fixtures/user.stub';
-import { newTestService, ServiceMocks } from 'test/utils';
+import { factory } from 'test/small.factory';
+import { makeStream, newTestService, ServiceMocks } from 'test/utils';
 import { vitest } from 'vitest';
 
 const stats: AssetStats = {
@@ -473,28 +474,30 @@ describe(AssetService.name, () => {
     });
 
     it('should immediately queue assets for deletion if trash is disabled', async () => {
-      mocks.asset.getAll.mockResolvedValue({ hasNextPage: false, items: [assetStub.image] });
+      const asset = factory.asset({ isOffline: false });
+
+      mocks.asset.streamDeletedAssets.mockReturnValue(makeStream([asset]));
       mocks.systemMetadata.get.mockResolvedValue({ trash: { enabled: false } });
 
       await expect(sut.handleAssetDeletionCheck()).resolves.toBe(JobStatus.SUCCESS);
 
-      expect(mocks.asset.getAll).toHaveBeenCalledWith(expect.anything(), { trashedBefore: new Date() });
+      expect(mocks.asset.streamDeletedAssets).toHaveBeenCalledWith(new Date());
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
-        { name: JobName.ASSET_DELETION, data: { id: assetStub.image.id, deleteOnDisk: true } },
+        { name: JobName.ASSET_DELETION, data: { id: asset.id, deleteOnDisk: true } },
       ]);
     });
 
     it('should queue assets for deletion after trash duration', async () => {
-      mocks.asset.getAll.mockResolvedValue({ hasNextPage: false, items: [assetStub.image] });
+      const asset = factory.asset({ isOffline: false });
+
+      mocks.asset.streamDeletedAssets.mockReturnValue(makeStream([asset]));
       mocks.systemMetadata.get.mockResolvedValue({ trash: { enabled: true, days: 7 } });
 
       await expect(sut.handleAssetDeletionCheck()).resolves.toBe(JobStatus.SUCCESS);
 
-      expect(mocks.asset.getAll).toHaveBeenCalledWith(expect.anything(), {
+      expect(mocks.asset.streamDeletedAssets).toHaveBeenCalledWith(DateTime.now().minus({ days: 7 }).toJSDate());
-        trashedBefore: DateTime.now().minus({ days: 7 }).toJSDate(),
-      });
       expect(mocks.job.queueAll).toHaveBeenCalledWith([
-        { name: JobName.ASSET_DELETION, data: { id: assetStub.image.id, deleteOnDisk: true } },
+        { name: JobName.ASSET_DELETION, data: { id: asset.id, deleteOnDisk: true } },
       ]);
     });
   });

server/src/services/asset.service.ts

@@ -25,7 +25,6 @@ import { AssetStatus, JobName, JobStatus, Permission, QueueName } from 'src/enum
 import { BaseService } from 'src/services/base.service';
 import { ISidecarWriteJob, JobItem, JobOf } from 'src/types';
 import { getAssetFiles, getMyPartnerIds, onAfterUnlink, onBeforeLink, onBeforeUnlink } from 'src/utils/asset.util';
-import { usePagination } from 'src/utils/pagination';
 
 @Injectable()
 export class AssetService extends BaseService {
@@ -156,22 +155,30 @@ export class AssetService extends BaseService {
     const trashedBefore = DateTime.now()
       .minus(Duration.fromObject({ days: trashedDays }))
       .toJSDate();
-    const assetPagination = usePagination(JOBS_ASSET_PAGINATION_SIZE, (pagination) =>
-      this.assetRepository.getAll(pagination, { trashedBefore }),
-    );
 
-    for await (const assets of assetPagination) {
+    let chunk: Array<{ id: string; isOffline: boolean }> = [];
-      await this.jobRepository.queueAll(
+    const queueChunk = async () => {
-        assets.map((asset) => ({
+      if (chunk.length > 0) {
-          name: JobName.ASSET_DELETION,
+        await this.jobRepository.queueAll(
-          data: {
+          chunk.map(({ id, isOffline }) => ({
-            id: asset.id,
+            name: JobName.ASSET_DELETION,
-            deleteOnDisk: !asset.isOffline,
+            data: { id, deleteOnDisk: !isOffline },
-          },
+          })),
-        })),
+        );
-      );
+        chunk = [];
+      }
+    };
+
+    const assets = this.assetRepository.streamDeletedAssets(trashedBefore);
+    for await (const asset of assets) {
+      chunk.push(asset);
+      if (chunk.length >= JOBS_ASSET_PAGINATION_SIZE) {
+        await queueChunk();
+      }
     }
 
+    await queueChunk();
+
     return JobStatus.SUCCESS;
   }
 

server/test/repositories/asset.repository.mock.ts

@@ -46,5 +46,6 @@ export const newAssetRepositoryMock = (): Mocked<RepositoryInterface<AssetReposi
     updateByLibraryId: vitest.fn(),
     streamStorageTemplateAssets: vitest.fn(),
     getStorageTemplateAsset: vitest.fn(),
+    streamDeletedAssets: vitest.fn(),
   };
 };