diff --git a/mobile/openapi/lib/model/sync_entity_type.dart b/mobile/openapi/lib/model/sync_entity_type.dart index f1366a2bfc..654ff45d6f 100644 --- a/mobile/openapi/lib/model/sync_entity_type.dart +++ b/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: diff --git a/open-api/immich-openapi-specs.json b/open-api/immich-openapi-specs.json index 1c240eb775..f942015fb3 100644 --- a/open-api/immich-openapi-specs.json +++ b/open-api/immich-openapi-specs.json @@ -13706,6 +13706,7 @@ "AlbumV1", "AlbumDeleteV1", "AlbumUserV1", + "AlbumUserBackfillV1", "AlbumUserDeleteV1", "SyncAckV1" ], diff --git a/open-api/typescript-sdk/src/fetch-client.ts b/open-api/typescript-sdk/src/fetch-client.ts index 49c97b79e9..0d57652df9 100644 --- a/open-api/typescript-sdk/src/fetch-client.ts +++ b/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" } diff --git a/server/src/database.ts b/server/src/database.ts index 79c550dd52..1cddda1ee6 100644 --- a/server/src/database.ts +++ b/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', diff --git a/server/src/db.d.ts b/server/src/db.d.ts index 5aa8a8c4dc..7a4c319d0b 100644 --- a/server/src/db.d.ts +++ b/server/src/db.d.ts @@ -98,8 +98,10 @@ export interface AlbumsSharedUsersUsers { albumsId: string; role: Generated; usersId: string; - updatedAt: Generated; + createId: Generated; + createdAt: Generated; updateId: Generated; + updatedAt: Generated; } export interface ApiKeys { diff --git a/server/src/dtos/sync.dto.ts b/server/src/dtos/sync.dto.ts index dbd58cde53..91c93fef66 100644 --- a/server/src/dtos/sync.dto.ts +++ b/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; }; diff --git a/server/src/enum.ts b/server/src/enum.ts index 4353e43ad1..4f3fd9a521 100644 --- a/server/src/enum.ts +++ b/server/src/enum.ts @@ -603,6 +603,7 @@ export enum SyncEntityType { AlbumV1 = 'AlbumV1', AlbumDeleteV1 = 'AlbumDeleteV1', AlbumUserV1 = 'AlbumUserV1', + AlbumUserBackfillV1 = 'AlbumUserBackfillV1', AlbumUserDeleteV1 = 'AlbumUserDeleteV1', SyncAckV1 = 'SyncAckV1', diff --git a/server/src/queries/sync.repository.sql b/server/src/queries/sync.repository.sql index 8e52754467..4c4747e5da 100644 --- a/server/src/queries/sync.repository.sql +++ b/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", diff --git a/server/src/repositories/sync.repository.ts b/server/src/repositories/sync.repository.ts index 0f2d382fe0..dee419f334 100644 --- a/server/src/repositories/sync.repository.ts +++ b/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("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("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("now() - interval '1 millisecond'")) .$if(!!ack, (qb) => qb.where('albums_shared_users_users.updateId', '>', ack!.updateId)) .orderBy('albums_shared_users_users.updateId', 'asc') diff --git a/server/src/schema/migrations/1750189909087-AddAlbumUserCreateFields.ts b/server/src/schema/migrations/1750189909087-AddAlbumUserCreateFields.ts new file mode 100644 index 0000000000..0ad59f9e82 --- /dev/null +++ b/server/src/schema/migrations/1750189909087-AddAlbumUserCreateFields.ts @@ -0,0 +1,15 @@ +import { Kysely, sql } from 'kysely'; + +export async function up(db: Kysely): Promise { + 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): Promise { + 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); +} diff --git a/server/src/schema/tables/album-user.table.ts b/server/src/schema/tables/album-user.table.ts index 276efd126a..15300f7c6a 100644 --- a/server/src/schema/tables/album-user.table.ts +++ b/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; diff --git a/server/src/schema/tables/partner.table.ts b/server/src/schema/tables/partner.table.ts index 6b83c6ba4c..8cec2ee58a 100644 --- a/server/src/schema/tables/partner.table.ts +++ b/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() diff --git a/server/src/services/sync.service.ts b/server/src/services/sync.service.ts index 46937a570f..9021aa57e9 100644 --- a/server/src/services/sync.service.ts +++ b/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 }); } } diff --git a/server/test/medium/specs/sync/sync-album-user.spec.ts b/server/test/medium/specs/sync/sync-album-user.spec.ts index 4967df5264..305bead275 100644 --- a/server/test/medium/specs/sync/sync-album-user.spec.ts +++ b/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; @@ -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([]); + }); }); });