feat: backfill album users (#19234)

mobile/openapi/lib/model/sync_entity_type.dart

@@ -38,6 +38,7 @@ class SyncEntityType {
   static const albumV1 = SyncEntityType._(r'AlbumV1');
   static const albumDeleteV1 = SyncEntityType._(r'AlbumDeleteV1');
   static const albumUserV1 = SyncEntityType._(r'AlbumUserV1');
+  static const albumUserBackfillV1 = SyncEntityType._(r'AlbumUserBackfillV1');
   static const albumUserDeleteV1 = SyncEntityType._(r'AlbumUserDeleteV1');
   static const syncAckV1 = SyncEntityType._(r'SyncAckV1');
 
@@ -58,6 +59,7 @@ class SyncEntityType {
     albumV1,
     albumDeleteV1,
     albumUserV1,
+    albumUserBackfillV1,
     albumUserDeleteV1,
     syncAckV1,
   ];
@@ -113,6 +115,7 @@ class SyncEntityTypeTypeTransformer {
         case r'AlbumV1': return SyncEntityType.albumV1;
         case r'AlbumDeleteV1': return SyncEntityType.albumDeleteV1;
         case r'AlbumUserV1': return SyncEntityType.albumUserV1;
+        case r'AlbumUserBackfillV1': return SyncEntityType.albumUserBackfillV1;
         case r'AlbumUserDeleteV1': return SyncEntityType.albumUserDeleteV1;
         case r'SyncAckV1': return SyncEntityType.syncAckV1;
         default:

open-api/immich-openapi-specs.json

@@ -13706,6 +13706,7 @@
           "AlbumV1",
           "AlbumDeleteV1",
           "AlbumUserV1",
+          "AlbumUserBackfillV1",
           "AlbumUserDeleteV1",
           "SyncAckV1"
         ],

open-api/typescript-sdk/src/fetch-client.ts

@@ -4059,6 +4059,7 @@ export enum SyncEntityType {
     AlbumV1 = "AlbumV1",
     AlbumDeleteV1 = "AlbumDeleteV1",
     AlbumUserV1 = "AlbumUserV1",
+    AlbumUserBackfillV1 = "AlbumUserBackfillV1",
     AlbumUserDeleteV1 = "AlbumUserDeleteV1",
     SyncAckV1 = "SyncAckV1"
 }

server/src/database.ts

@@ -355,6 +355,12 @@ export const columns = {
     'updateId',
     'duration',
   ],
+  syncAlbumUser: [
+    'albums_shared_users_users.albumsId as albumId',
+    'albums_shared_users_users.usersId as userId',
+    'albums_shared_users_users.role',
+    'albums_shared_users_users.updateId',
+  ],
   stack: ['stack.id', 'stack.primaryAssetId', 'ownerId'],
   syncAssetExif: [
     'exif.assetId',

server/src/db.d.ts

@@ -98,8 +98,10 @@ export interface AlbumsSharedUsersUsers {
   albumsId: string;
   role: Generated<AlbumUserRole>;
   usersId: string;
-  updatedAt: Generated<Timestamp>;
+  createId: Generated<string>;
+  createdAt: Generated<Timestamp>;
   updateId: Generated<string>;
+  updatedAt: Generated<Timestamp>;
 }
 
 export interface ApiKeys {

server/src/dtos/sync.dto.ts

@@ -161,6 +161,7 @@ export type SyncItem = {
   [SyncEntityType.AlbumV1]: SyncAlbumV1;
   [SyncEntityType.AlbumDeleteV1]: SyncAlbumDeleteV1;
   [SyncEntityType.AlbumUserV1]: SyncAlbumUserV1;
+  [SyncEntityType.AlbumUserBackfillV1]: SyncAlbumUserV1;
   [SyncEntityType.AlbumUserDeleteV1]: SyncAlbumUserDeleteV1;
   [SyncEntityType.SyncAckV1]: object;
 };

server/src/enum.ts

@@ -603,6 +603,7 @@ export enum SyncEntityType {
   AlbumV1 = 'AlbumV1',
   AlbumDeleteV1 = 'AlbumDeleteV1',
   AlbumUserV1 = 'AlbumUserV1',
+  AlbumUserBackfillV1 = 'AlbumUserBackfillV1',
   AlbumUserDeleteV1 = 'AlbumUserDeleteV1',
 
   SyncAckV1 = 'SyncAckV1',

server/src/queries/sync.repository.sql

@@ -394,6 +394,35 @@ where
 order by
   "id" asc
 
+-- SyncRepository.getAlbumBackfill
+select
+  "albumsId" as "id",
+  "createId"
+from
+  "albums_shared_users_users"
+where
+  "usersId" = $1
+  and "createId" >= $2
+  and "createdAt" < now() - interval '1 millisecond'
+order by
+  "createId" asc
+
+-- SyncRepository.getAlbumUsersBackfill
+select
+  "albums_shared_users_users"."albumsId" as "albumId",
+  "albums_shared_users_users"."usersId" as "userId",
+  "albums_shared_users_users"."role",
+  "albums_shared_users_users"."updateId"
+from
+  "albums_shared_users_users"
+where
+  "albumsId" = $1
+  and "updatedAt" < now() - interval '1 millisecond'
+  and "updateId" < $2
+  and "updateId" >= $3
+order by
+  "updateId" asc
+
 -- SyncRepository.getAlbumUserUpserts
 select
   "albums_shared_users_users"."albumsId" as "albumId",

server/src/repositories/sync.repository.ts

@@ -254,16 +254,36 @@ export class SyncRepository {
       .stream();
   }
 
+  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID] })
+  getAlbumBackfill(userId: string, afterCreateId?: string) {
+    return this.db
+      .selectFrom('albums_shared_users_users')
+      .select(['albumsId as id', 'createId'])
+      .where('usersId', '=', userId)
+      .$if(!!afterCreateId, (qb) => qb.where('createId', '>=', afterCreateId!))
+      .where('createdAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
+      .orderBy('createId', 'asc')
+      .execute();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
+  getAlbumUsersBackfill(albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
+    return this.db
+      .selectFrom('albums_shared_users_users')
+      .select(columns.syncAlbumUser)
+      .where('albumsId', '=', albumId)
+      .where('updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
+      .where('updateId', '<', beforeUpdateId)
+      .$if(!!afterUpdateId, (eb) => eb.where('updateId', '>=', afterUpdateId!))
+      .orderBy('updateId', 'asc')
+      .stream();
+  }
+
   @GenerateSql({ params: [DummyValue.UUID], stream: true })
   getAlbumUserUpserts(userId: string, ack?: SyncAck) {
     return this.db
       .selectFrom('albums_shared_users_users')
-      .select([
-        'albums_shared_users_users.albumsId as albumId',
-        'albums_shared_users_users.usersId as userId',
-        'albums_shared_users_users.role',
-        'albums_shared_users_users.updateId',
-      ])
+      .select(columns.syncAlbumUser)
       .where('albums_shared_users_users.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
       .$if(!!ack, (qb) => qb.where('albums_shared_users_users.updateId', '>', ack!.updateId))
       .orderBy('albums_shared_users_users.updateId', 'asc')

server/src/schema/migrations/1750189909087-AddAlbumUserCreateFields.ts

@@ -0,0 +1,15 @@
+import { Kysely, sql } from 'kysely';
+
+export async function up(db: Kysely<any>): Promise<void> {
+  await sql`ALTER TABLE "albums_shared_users_users" ADD "createId" uuid NOT NULL DEFAULT immich_uuid_v7();`.execute(db);
+  await sql`ALTER TABLE "albums_shared_users_users" ADD "createdAt" timestamp with time zone NOT NULL DEFAULT now();`.execute(db);
+  await sql`CREATE INDEX "IDX_album_users_create_id" ON "albums_shared_users_users" ("createId")`.execute(db);
+  await sql`CREATE INDEX "IDX_partners_create_id" ON "partners" ("createId")`.execute(db);
+}
+
+export async function down(db: Kysely<any>): Promise<void> {
+  await sql`DROP INDEX "IDX_partners_create_id";`.execute(db);
+  await sql`DROP INDEX "IDX_album_users_create_id";`.execute(db);
+  await sql`ALTER TABLE "albums_shared_users_users" DROP COLUMN "createId";`.execute(db);
+  await sql`ALTER TABLE "albums_shared_users_users" DROP COLUMN "createdAt";`.execute(db);
+}

server/src/schema/tables/album-user.table.ts

@@ -1,4 +1,4 @@
-import { UpdatedAtTrigger, UpdateIdColumn } from 'src/decorators';
+import { CreateIdColumn, UpdatedAtTrigger, UpdateIdColumn } from 'src/decorators';
 import { AlbumUserRole } from 'src/enum';
 import { album_user_after_insert, album_users_delete_audit } from 'src/schema/functions';
 import { AlbumTable } from 'src/schema/tables/album.table';
@@ -7,6 +7,7 @@ import {
   AfterDeleteTrigger,
   AfterInsertTrigger,
   Column,
+  CreateDateColumn,
   ForeignKeyColumn,
   Index,
   Table,
@@ -51,6 +52,12 @@ export class AlbumUserTable {
   @Column({ type: 'character varying', default: AlbumUserRole.EDITOR })
   role!: AlbumUserRole;
 
+  @CreateIdColumn({ indexName: 'IDX_album_users_create_id' })
+  createId?: string;
+
+  @CreateDateColumn()
+  createdAt!: Date;
+
   @UpdateIdColumn({ indexName: 'IDX_album_users_update_id' })
   updateId?: string;
 

server/src/schema/tables/partner.table.ts

@@ -27,7 +27,7 @@ export class PartnerTable {
   @CreateDateColumn()
   createdAt!: Date;
 
-  @CreateIdColumn()
+  @CreateIdColumn({ indexName: 'IDX_partners_create_id' })
   createId!: string;
 
   @UpdateDateColumn()

server/src/services/sync.service.ts

@@ -138,14 +138,14 @@ export class SyncService extends BaseService {
           break;
         }
 
-        case SyncRequestType.PartnerAssetsV1: {
-          await this.syncPartnerAssetsV1(response, checkpointMap, auth, sessionId);
-
+        case SyncRequestType.AssetExifsV1: {
+          await this.syncAssetExifsV1(response, checkpointMap, auth);
           break;
         }
 
-        case SyncRequestType.AssetExifsV1: {
-          await this.syncAssetExifsV1(response, checkpointMap, auth);
+        case SyncRequestType.PartnerAssetsV1: {
+          await this.syncPartnerAssetsV1(response, checkpointMap, auth, sessionId);
+
           break;
         }
 
@@ -160,7 +160,7 @@ export class SyncService extends BaseService {
         }
 
         case SyncRequestType.AlbumUsersV1: {
-          await this.syncAlbumUsersV1(response, checkpointMap, auth);
+          await this.syncAlbumUsersV1(response, checkpointMap, auth, sessionId);
           break;
         }
 
@@ -330,18 +330,50 @@ export class SyncService extends BaseService {
     }
   }
 
-  private async syncAlbumUsersV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto) {
-    const deletes = this.syncRepository.getAlbumUserDeletes(
-      auth.user.id,
-      checkpointMap[SyncEntityType.AlbumUserDeleteV1],
-    );
+  private async syncAlbumUsersV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto, sessionId: string) {
+    const backfillType = SyncEntityType.AlbumUserBackfillV1;
+    const upsertType = SyncEntityType.AlbumUserV1;
+    const deleteType = SyncEntityType.AlbumUserDeleteV1;
+
+    const backfillCheckpoint = checkpointMap[backfillType];
+    const upsertCheckpoint = checkpointMap[upsertType];
+
+    const deletes = this.syncRepository.getAlbumUserDeletes(auth.user.id, checkpointMap[deleteType]);
+
     for await (const { id, ...data } of deletes) {
-      send(response, { type: SyncEntityType.AlbumUserDeleteV1, ids: [id], data });
+      send(response, { type: deleteType, ids: [id], data });
     }
 
-    const upserts = this.syncRepository.getAlbumUserUpserts(auth.user.id, checkpointMap[SyncEntityType.AlbumUserV1]);
+    const albums = await this.syncRepository.getAlbumBackfill(auth.user.id, backfillCheckpoint?.updateId);
+
+    if (upsertCheckpoint) {
+      const endId = upsertCheckpoint.updateId;
+
+      for (const album of albums) {
+        if (isEntityBackfillComplete(album, backfillCheckpoint)) {
+          continue;
+        }
+
+        const startId = getStartId(album, backfillCheckpoint);
+        const backfill = this.syncRepository.getAlbumUsersBackfill(album.id, startId, endId);
+
+        for await (const { updateId, ...data } of backfill) {
+          send(response, { type: backfillType, ids: [updateId], data });
+        }
+
+        sendEntityBackfillCompleteAck(response, backfillType, album.id);
+      }
+    } else if (albums.length > 0) {
+      await this.upsertBackfillCheckpoint({
+        type: backfillType,
+        sessionId,
+        createId: albums.at(-1)!.createId,
+      });
+    }
+
+    const upserts = this.syncRepository.getAlbumUserUpserts(auth.user.id, checkpointMap[upsertType]);
     for await (const { updateId, ...data } of upserts) {
-      send(response, { type: SyncEntityType.AlbumUserV1, ids: [updateId], data });
+      send(response, { type: upsertType, ids: [updateId], data });
     }
   }
 

server/test/medium/specs/sync/sync-album-user.spec.ts

@@ -2,7 +2,7 @@ import { Kysely } from 'kysely';
 import { DB } from 'src/db';
 import { AlbumUserRole, SyncEntityType, SyncRequestType } from 'src/enum';
 import { mediumFactory, newSyncAuthUser, newSyncTest } from 'test/medium.factory';
-import { getKyselyDB } from 'test/utils';
+import { getKyselyDB, wait } from 'test/utils';
 
 let defaultDatabase: Kysely<DB>;
 
@@ -265,5 +265,95 @@ describe(SyncRequestType.AlbumUsersV1, () => {
         },
       ]);
     });
+
+    it('should backfill album users when a user shares an album with you', async () => {
+      const { auth, sut, testSync, getRepository } = await setup();
+
+      const albumRepo = getRepository('album');
+      const albumUserRepo = getRepository('albumUser');
+      const userRepo = getRepository('user');
+
+      const user1 = mediumFactory.userInsert();
+      const user2 = mediumFactory.userInsert();
+      await userRepo.create(user1);
+      await userRepo.create(user2);
+
+      const album1 = mediumFactory.albumInsert({ ownerId: user1.id });
+      const album2 = mediumFactory.albumInsert({ ownerId: user1.id });
+      await albumRepo.create(album1, [], []);
+      await albumRepo.create(album2, [], []);
+
+      // backfill album user
+      await albumUserRepo.create({ albumsId: album1.id, usersId: user1.id, role: AlbumUserRole.EDITOR });
+      await wait(2);
+      // initial album user
+      await albumUserRepo.create({ albumsId: album2.id, usersId: auth.user.id, role: AlbumUserRole.EDITOR });
+      await wait(2);
+      // post checkpoint album user
+      await albumUserRepo.create({ albumsId: album1.id, usersId: user2.id, role: AlbumUserRole.EDITOR });
+
+      const response = await testSync(auth, [SyncRequestType.AlbumUsersV1]);
+      expect(response).toHaveLength(1);
+      expect(response).toEqual([
+        {
+          ack: expect.any(String),
+          data: expect.objectContaining({
+            albumId: album2.id,
+            role: AlbumUserRole.EDITOR,
+            userId: auth.user.id,
+          }),
+          type: SyncEntityType.AlbumUserV1,
+        },
+      ]);
+
+      // ack initial user
+      const acks = response.map(({ ack }) => ack);
+      await sut.setAcks(auth, { acks });
+
+      // get access to the backfill album user
+      await albumUserRepo.create({ albumsId: album1.id, usersId: auth.user.id, role: AlbumUserRole.EDITOR });
+
+      // should backfill the album user
+      const backfillResponse = await testSync(auth, [SyncRequestType.AlbumUsersV1]);
+      expect(backfillResponse).toEqual([
+        {
+          ack: expect.any(String),
+          data: expect.objectContaining({
+            albumId: album1.id,
+            role: AlbumUserRole.EDITOR,
+            userId: user1.id,
+          }),
+          type: SyncEntityType.AlbumUserBackfillV1,
+        },
+        {
+          ack: expect.stringContaining(SyncEntityType.AlbumUserBackfillV1),
+          data: {},
+          type: SyncEntityType.SyncAckV1,
+        },
+        {
+          ack: expect.any(String),
+          data: expect.objectContaining({
+            albumId: album1.id,
+            role: AlbumUserRole.EDITOR,
+            userId: user2.id,
+          }),
+          type: SyncEntityType.AlbumUserV1,
+        },
+        {
+          ack: expect.any(String),
+          data: expect.objectContaining({
+            albumId: album1.id,
+            role: AlbumUserRole.EDITOR,
+            userId: auth.user.id,
+          }),
+          type: SyncEntityType.AlbumUserV1,
+        },
+      ]);
+
+      await sut.setAcks(auth, { acks: [backfillResponse[1].ack, backfillResponse.at(-1).ack] });
+
+      const finalResponse = await testSync(auth, [SyncRequestType.AlbumUsersV1]);
+      expect(finalResponse).toEqual([]);
+    });
   });
 });