feat(server): SyncAssetEditV1 (#26446)

* feat: SyncAssetEditV1

* fix: audit table import

* fix: sql tools table fetch

* fix: medium tests (wip)

* fix: circ dependency

* chore: finalize tests

* chore: codegen/lint

* fix: code review

server/test/medium/specs/services/audit.database.spec.ts

@@ -1,3 +1,5 @@
+import { AssetEditAction } from 'src/dtos/editing.dto';
+import { AssetEditRepository } from 'src/repositories/asset-edit.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
 import { PartnerRepository } from 'src/repositories/partner.repository';
 import { UserRepository } from 'src/repositories/user.repository';
@@ -45,6 +47,27 @@ describe('audit', () => {
     });
   });
 
+  describe('asset_edit_audit', () => {
+    it('should not cascade asset deletes to asset_edit_audit', async () => {
+      const assetEditRepo = ctx.get(AssetEditRepository);
+      const { user } = await ctx.newUser();
+      const { asset } = await ctx.newAsset({ ownerId: user.id });
+
+      await assetEditRepo.replaceAll(asset.id, [
+        {
+          action: AssetEditAction.Crop,
+          parameters: { x: 10, y: 20, width: 100, height: 200 },
+        },
+      ]);
+
+      await ctx.database.deleteFrom('asset').where('id', '=', asset.id).execute();
+
+      await expect(
+        ctx.database.selectFrom('asset_edit_audit').select(['id']).where('assetId', '=', asset.id).execute(),
+      ).resolves.toHaveLength(0);
+    });
+  });
+
   describe('assets_audit', () => {
     it('should not cascade user deletes to assets_audit', async () => {
       const userRepo = ctx.get(UserRepository);

server/test/medium/specs/services/sync.service.spec.ts

@@ -1,9 +1,10 @@
+import { schemaFromCode } from '@immich/sql-tools';
 import { Kysely } from 'kysely';
 import { DateTime } from 'luxon';
 import { AssetMetadataKey, UserMetadataKey } from 'src/enum';
 import { DatabaseRepository } from 'src/repositories/database.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
-import { SyncRepository } from 'src/repositories/sync.repository';
+import { BaseSync, SyncRepository } from 'src/repositories/sync.repository';
 import { DB } from 'src/schema';
 import { SyncService } from 'src/services/sync.service';
 import { newMediumService } from 'test/medium.factory';
@@ -222,5 +223,21 @@ describe(SyncService.name, () => {
       expect(after).toHaveLength(1);
       expect(after[0].id).toBe(keep.id);
     });
+
+    it('should cleanup every table', async () => {
+      const { sut } = setup();
+
+      const auditTables = schemaFromCode()
+        .tables.filter((table) => table.name.endsWith('_audit'))
+        .map(({ name }) => name);
+
+      const auditCleanupSpy = vi.spyOn(BaseSync.prototype as any, 'auditCleanup');
+      await expect(sut.onAuditTableCleanup()).resolves.toBeUndefined();
+
+      expect(auditCleanupSpy).toHaveBeenCalledTimes(auditTables.length);
+      for (const table of auditTables) {
+        expect(auditCleanupSpy, `Audit table ${table} was not cleaned up`).toHaveBeenCalledWith(table, 31);
+      }
+    });
   });
 });

server/test/medium/specs/sync/sync-asset-edit.spec.ts

@@ -0,0 +1,300 @@
+import { Kysely } from 'kysely';
+import { AssetEditAction, MirrorAxis } from 'src/dtos/editing.dto';
+import { SyncEntityType, SyncRequestType } from 'src/enum';
+import { AssetEditRepository } from 'src/repositories/asset-edit.repository';
+import { DB } from 'src/schema';
+import { SyncTestContext } from 'test/medium.factory';
+import { factory } from 'test/small.factory';
+import { getKyselyDB } from 'test/utils';
+
+let defaultDatabase: Kysely<DB>;
+
+const setup = async (db?: Kysely<DB>) => {
+  const ctx = new SyncTestContext(db || defaultDatabase);
+  const { auth, user, session } = await ctx.newSyncAuthUser();
+  return { auth, user, session, ctx };
+};
+
+beforeAll(async () => {
+  defaultDatabase = await getKyselyDB();
+});
+
+describe(SyncRequestType.AssetEditsV1, () => {
+  it('should detect and sync the first asset edit', async () => {
+    const { auth, ctx } = await setup();
+    const { asset } = await ctx.newAsset({ ownerId: auth.user.id });
+    const assetEditRepo = ctx.get(AssetEditRepository);
+
+    await assetEditRepo.replaceAll(asset.id, [
+      {
+        action: AssetEditAction.Crop,
+        parameters: { x: 10, y: 20, width: 100, height: 200 },
+      },
+    ]);
+
+    const response = await ctx.syncStream(auth, [SyncRequestType.AssetEditsV1]);
+    expect(response).toEqual([
+      {
+        ack: expect.any(String),
+        data: {
+          id: expect.any(String),
+          assetId: asset.id,
+          action: AssetEditAction.Crop,
+          parameters: { x: 10, y: 20, width: 100, height: 200 },
+          sequence: 0,
+        },
+        type: SyncEntityType.AssetEditV1,
+      },
+      expect.objectContaining({ type: SyncEntityType.SyncCompleteV1 }),
+    ]);
+
+    await ctx.syncAckAll(auth, response);
+    await ctx.assertSyncIsComplete(auth, [SyncRequestType.AssetEditsV1]);
+  });
+
+  it('should detect and sync multiple asset edits for the same asset', async () => {
+    const { auth, ctx } = await setup();
+    const { asset } = await ctx.newAsset({ ownerId: auth.user.id });
+    const assetEditRepo = ctx.get(AssetEditRepository);
+
+    await assetEditRepo.replaceAll(asset.id, [
+      {
+        action: AssetEditAction.Crop,
+        parameters: { x: 10, y: 20, width: 100, height: 200 },
+      },
+      {
+        action: AssetEditAction.Rotate,
+        parameters: { angle: 90 },
+      },
+      {
+        action: AssetEditAction.Mirror,
+        parameters: { axis: MirrorAxis.Horizontal },
+      },
+    ]);
+
+    const response = await ctx.syncStream(auth, [SyncRequestType.AssetEditsV1]);
+    expect(response).toEqual(
+      expect.arrayContaining([
+        {
+          ack: expect.any(String),
+          data: {
+            id: expect.any(String),
+            assetId: asset.id,
+            action: AssetEditAction.Crop,
+            parameters: { x: 10, y: 20, width: 100, height: 200 },
+            sequence: 0,
+          },
+          type: SyncEntityType.AssetEditV1,
+        },
+        {
+          ack: expect.any(String),
+          data: {
+            id: expect.any(String),
+            assetId: asset.id,
+            action: AssetEditAction.Rotate,
+            parameters: { angle: 90 },
+            sequence: 1,
+          },
+          type: SyncEntityType.AssetEditV1,
+        },
+        {
+          ack: expect.any(String),
+          data: {
+            id: expect.any(String),
+            assetId: asset.id,
+            action: AssetEditAction.Mirror,
+            parameters: { axis: MirrorAxis.Horizontal },
+            sequence: 2,
+          },
+          type: SyncEntityType.AssetEditV1,
+        },
+        expect.objectContaining({ type: SyncEntityType.SyncCompleteV1 }),
+      ]),
+    );
+
+    await ctx.syncAckAll(auth, response);
+    await ctx.assertSyncIsComplete(auth, [SyncRequestType.AssetEditsV1]);
+  });
+
+  it('should detect and sync updated edits', async () => {
+    const { auth, ctx } = await setup();
+    const { asset } = await ctx.newAsset({ ownerId: auth.user.id });
+    const assetEditRepo = ctx.get(AssetEditRepository);
+
+    // Create initial edit
+    const edits = await assetEditRepo.replaceAll(asset.id, [
+      {
+        action: AssetEditAction.Crop,
+        parameters: { x: 10, y: 20, width: 100, height: 200 },
+      },
+    ]);
+
+    const response1 = await ctx.syncStream(auth, [SyncRequestType.AssetEditsV1]);
+    await ctx.syncAckAll(auth, response1);
+    await ctx.assertSyncIsComplete(auth, [SyncRequestType.AssetEditsV1]);
+
+    // update the existing edit
+    await ctx.database
+      .updateTable('asset_edit')
+      .set({
+        parameters: { x: 50, y: 60, width: 150, height: 250 },
+      })
+      .where('id', '=', edits[0].id)
+      .execute();
+
+    const response2 = await ctx.syncStream(auth, [SyncRequestType.AssetEditsV1]);
+    expect(response2).toEqual(
+      expect.arrayContaining([
+        {
+          ack: expect.any(String),
+          data: {
+            id: expect.any(String),
+            assetId: asset.id,
+            action: AssetEditAction.Crop,
+            parameters: { x: 50, y: 60, width: 150, height: 250 },
+            sequence: 0,
+          },
+          type: SyncEntityType.AssetEditV1,
+        },
+        expect.objectContaining({ type: SyncEntityType.SyncCompleteV1 }),
+      ]),
+    );
+
+    await ctx.syncAckAll(auth, response2);
+    await ctx.assertSyncIsComplete(auth, [SyncRequestType.AssetEditsV1]);
+  });
+
+  it('should detect and sync deleted asset edits', async () => {
+    const { auth, ctx } = await setup();
+    const { asset } = await ctx.newAsset({ ownerId: auth.user.id });
+    const assetEditRepo = ctx.get(AssetEditRepository);
+
+    // Create initial edit
+    const edits = await assetEditRepo.replaceAll(asset.id, [
+      {
+        action: AssetEditAction.Crop,
+        parameters: { x: 10, y: 20, width: 100, height: 200 },
+      },
+    ]);
+
+    const response1 = await ctx.syncStream(auth, [SyncRequestType.AssetEditsV1]);
+    await ctx.syncAckAll(auth, response1);
+    await ctx.assertSyncIsComplete(auth, [SyncRequestType.AssetEditsV1]);
+
+    // Delete all edits
+    await assetEditRepo.replaceAll(asset.id, []);
+
+    const response2 = await ctx.syncStream(auth, [SyncRequestType.AssetEditsV1]);
+    expect(response2).toEqual(
+      expect.arrayContaining([
+        {
+          ack: expect.any(String),
+          data: {
+            editId: edits[0].id,
+          },
+          type: SyncEntityType.AssetEditDeleteV1,
+        },
+        expect.objectContaining({ type: SyncEntityType.SyncCompleteV1 }),
+      ]),
+    );
+
+    await ctx.syncAckAll(auth, response2);
+    await ctx.assertSyncIsComplete(auth, [SyncRequestType.AssetEditsV1]);
+  });
+
+  it('should only sync asset edits for own user', async () => {
+    const { auth, ctx } = await setup();
+    const { user: user2 } = await ctx.newUser();
+    const { asset } = await ctx.newAsset({ ownerId: user2.id });
+    const assetEditRepo = ctx.get(AssetEditRepository);
+    const { session } = await ctx.newSession({ userId: user2.id });
+    const auth2 = factory.auth({ session, user: user2 });
+
+    await assetEditRepo.replaceAll(asset.id, [
+      {
+        action: AssetEditAction.Crop,
+        parameters: { x: 10, y: 20, width: 100, height: 200 },
+      },
+    ]);
+
+    // User 2 should see their own edit
+    await expect(ctx.syncStream(auth2, [SyncRequestType.AssetEditsV1])).resolves.toEqual([
+      expect.objectContaining({ type: SyncEntityType.AssetEditV1 }),
+      expect.objectContaining({ type: SyncEntityType.SyncCompleteV1 }),
+    ]);
+
+    // User 1 should not see user 2's edit
+    await ctx.assertSyncIsComplete(auth, [SyncRequestType.AssetEditsV1]);
+  });
+
+  it('should sync edits for multiple assets', async () => {
+    const { auth, ctx } = await setup();
+    const { asset: asset1 } = await ctx.newAsset({ ownerId: auth.user.id });
+    const { asset: asset2 } = await ctx.newAsset({ ownerId: auth.user.id });
+    const assetEditRepo = ctx.get(AssetEditRepository);
+
+    await assetEditRepo.replaceAll(asset1.id, [
+      {
+        action: AssetEditAction.Crop,
+        parameters: { x: 10, y: 20, width: 100, height: 200 },
+      },
+    ]);
+
+    await assetEditRepo.replaceAll(asset2.id, [
+      {
+        action: AssetEditAction.Rotate,
+        parameters: { angle: 270 },
+      },
+    ]);
+
+    const response = await ctx.syncStream(auth, [SyncRequestType.AssetEditsV1]);
+    expect(response).toEqual(
+      expect.arrayContaining([
+        {
+          ack: expect.any(String),
+          data: {
+            id: expect.any(String),
+            assetId: asset1.id,
+            action: AssetEditAction.Crop,
+            parameters: { x: 10, y: 20, width: 100, height: 200 },
+            sequence: 0,
+          },
+          type: SyncEntityType.AssetEditV1,
+        },
+        {
+          ack: expect.any(String),
+          data: {
+            id: expect.any(String),
+            assetId: asset2.id,
+            action: AssetEditAction.Rotate,
+            parameters: { angle: 270 },
+            sequence: 0,
+          },
+          type: SyncEntityType.AssetEditV1,
+        },
+        expect.objectContaining({ type: SyncEntityType.SyncCompleteV1 }),
+      ]),
+    );
+
+    await ctx.syncAckAll(auth, response);
+    await ctx.assertSyncIsComplete(auth, [SyncRequestType.AssetEditsV1]);
+  });
+
+  it('should not sync edits for partner assets', async () => {
+    const { auth, ctx } = await setup();
+    const { user: partner } = await ctx.newUser();
+    await ctx.newPartner({ sharedById: partner.id, sharedWithId: auth.user.id });
+    const { asset } = await ctx.newAsset({ ownerId: partner.id });
+    const assetEditRepo = ctx.get(AssetEditRepository);
+
+    await assetEditRepo.replaceAll(asset.id, [
+      {
+        action: AssetEditAction.Crop,
+        parameters: { x: 10, y: 20, width: 100, height: 200 },
+      },
+    ]);
+
+    // Should not see partner's asset edits in own sync
+    await ctx.assertSyncIsComplete(auth, [SyncRequestType.AssetEditsV1]);
+  });
+});