diff --git a/server/src/queries/session.repository.sql b/server/src/queries/session.repository.sql index f68f2dbe95..824e9f06ac 100644 --- a/server/src/queries/session.repository.sql +++ b/server/src/queries/session.repository.sql @@ -88,6 +88,20 @@ set where "userId" = $2 +-- SessionRepository.requireFullSyncForNonAdmins +update "session" +set + "isPendingSyncReset" = $1 +where + "userId" in ( + select + "user"."id" + from + "user" + where + "user"."isAdmin" = $2 + ) + -- SessionRepository.resetSyncProgress begin update "session" diff --git a/server/src/queries/sync.repository.sql b/server/src/queries/sync.repository.sql index 69f7981bf3..58e7046bbb 100644 --- a/server/src/queries/sync.repository.sql +++ b/server/src/queries/sync.repository.sql @@ -1121,6 +1121,58 @@ where order by "user"."updateId" asc +-- SyncRepository.user.getRelatedUpserts +select + "id", + "name", + "email", + "avatarColor", + "deletedAt", + "updateId", + "profileImagePath", + "profileChangedAt" +from + "user" as "user" +where + "user"."updateId" < $1 + and "user"."updateId" > $2 + and ( + "id" = $3 + or "id" in ( + select + "partner"."sharedWithId" as "id" + from + "partner" + where + "partner"."sharedById" = $4 + ) + or "id" in ( + select + "partner"."sharedById" as "id" + from + "partner" + where + "partner"."sharedWithId" = $5 + ) + or "id" in ( + select + "album_user"."userId" as "id" + from + "album_user" + where + "album_user"."albumId" in ( + select + "album_user"."albumId" as "id" + from + "album_user" + where + "album_user"."userId" = $6 + ) + ) + ) +order by + "user"."updateId" asc + -- SyncRepository.userMetadata.getDeletes select "id", diff --git a/server/src/repositories/session.repository.ts b/server/src/repositories/session.repository.ts index 58a5e721c0..cd5852b8a9 100644 --- a/server/src/repositories/session.repository.ts +++ b/server/src/repositories/session.repository.ts @@ -135,6 +135,15 @@ export class SessionRepository { await this.db.updateTable('session').set({ pinExpiresAt: null }).where('userId', '=', userId).execute(); } + @GenerateSql({ params: [] }) + async requireFullSyncForNonAdmins() { + await this.db + .updateTable('session') + .set({ isPendingSyncReset: true }) + .where((eb) => eb('userId', 'in', eb.selectFrom('user').select('user.id').where('user.isAdmin', '=', false))) + .execute(); + } + @GenerateSql({ params: [DummyValue.UUID] }) async resetSyncProgress(sessionId: string) { await this.db.transaction().execute((tx) => { diff --git a/server/src/repositories/sync.repository.ts b/server/src/repositories/sync.repository.ts index eca8d18e66..99ed57c750 100644 --- a/server/src/repositories/sync.repository.ts +++ b/server/src/repositories/sync.repository.ts @@ -751,6 +751,42 @@ class UserSync extends BaseSync { getUpserts(options: SyncQueryOptions) { return this.upsertQuery('user', options).select(columns.syncUser).stream(); } + + // used when public users are disabled: only self, partners, and album co-members are visible + @GenerateSql({ params: [dummyQueryOptions], stream: true }) + getRelatedUpserts(options: SyncQueryOptions) { + const userId = options.userId; + return this.upsertQuery('user', options) + .select(columns.syncUser) + .where((eb) => + eb.or([ + eb('id', '=', userId), + eb( + 'id', + 'in', + eb.selectFrom('partner').select('partner.sharedWithId as id').where('partner.sharedById', '=', userId), + ), + eb( + 'id', + 'in', + eb.selectFrom('partner').select('partner.sharedById as id').where('partner.sharedWithId', '=', userId), + ), + eb( + 'id', + 'in', + eb + .selectFrom('album_user') + .select('album_user.userId as id') + .where( + 'album_user.albumId', + 'in', + eb.selectFrom('album_user').select('album_user.albumId as id').where('album_user.userId', '=', userId), + ), + ), + ]), + ) + .stream(); + } } class UserMetadataSync extends BaseSync { diff --git a/server/src/services/sync.service.ts b/server/src/services/sync.service.ts index e3842b1503..ef2dcd792a 100644 --- a/server/src/services/sync.service.ts +++ b/server/src/services/sync.service.ts @@ -2,7 +2,7 @@ import { BadRequestException, ForbiddenException, Injectable } from '@nestjs/com import { Insertable } from 'kysely'; import { DateTime, Duration } from 'luxon'; import { Writable } from 'node:stream'; -import { OnJob } from 'src/decorators'; +import { OnEvent, OnJob } from 'src/decorators'; import { AuthDto } from 'src/dtos/auth.dto'; import { SyncAckDeleteDto, @@ -12,7 +12,8 @@ import { SyncItem, SyncStreamDto, } from 'src/dtos/sync.dto'; -import { JobName, QueueName, SyncEntityType, SyncRequestType } from 'src/enum'; +import { ImmichWorker, JobName, QueueName, SyncEntityType, SyncRequestType } from 'src/enum'; +import { ArgOf } from 'src/repositories/event.repository'; import { SyncQueryOptions } from 'src/repositories/sync.repository'; import { SessionSyncCheckpointTable } from 'src/schema/tables/sync-checkpoint.table'; import { BaseService } from 'src/services/base.service'; @@ -86,6 +87,13 @@ const throwSessionRequired = () => { @Injectable() export class SyncService extends BaseService { + @OnEvent({ name: 'ConfigUpdate', workers: [ImmichWorker.Api] }) + async onConfigUpdate({ newConfig, oldConfig }: ArgOf<'ConfigUpdate'>) { + if (oldConfig.server.publicUsers !== newConfig.server.publicUsers) { + await this.sessionRepository.requireFullSyncForNonAdmins(); + } + } + getAcks(auth: AuthDto) { const sessionId = auth.session?.id; if (!sessionId) { @@ -166,7 +174,7 @@ export class SyncService extends BaseService { [SyncRequestType.AlbumAssetsV1]: () => this.syncAlbumAssetsV1(), [SyncRequestType.AuthUsersV1]: () => this.syncAuthUsersV1(options, response, checkpointMap), - [SyncRequestType.UsersV1]: () => this.syncUsersV1(options, response, checkpointMap), + [SyncRequestType.UsersV1]: () => this.syncUsersV1(options, response, checkpointMap, auth), [SyncRequestType.PartnersV1]: () => this.syncPartnersV1(options, response, checkpointMap), [SyncRequestType.AssetsV2]: () => this.syncAssetsV2(options, response, checkpointMap), [SyncRequestType.AssetExifsV1]: () => this.syncAssetExifsV1(options, response, checkpointMap), @@ -246,7 +254,7 @@ export class SyncService extends BaseService { } } - private async syncUsersV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) { + private async syncUsersV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap, auth: AuthDto) { const deleteType = SyncEntityType.UserDeleteV1; const deletes = this.syncRepository.user.getDeletes({ ...options, ack: checkpointMap[deleteType] }); for await (const { id, ...data } of deletes) { @@ -254,7 +262,11 @@ export class SyncService extends BaseService { } const upsertType = SyncEntityType.UserV1; - const upserts = this.syncRepository.user.getUpserts({ ...options, ack: checkpointMap[upsertType] }); + const { server } = await this.getConfig({ withCache: false }); + const canViewAllUsers = auth.user.isAdmin || server.publicUsers; + const upserts = canViewAllUsers + ? this.syncRepository.user.getUpserts({ ...options, ack: checkpointMap[upsertType] }) + : this.syncRepository.user.getRelatedUpserts({ ...options, ack: checkpointMap[upsertType] }); for await (const { updateId, profileImagePath, ...data } of upserts) { send(response, { type: upsertType, ids: [updateId], data: { ...data, hasProfileImage: !!profileImagePath } }); } diff --git a/server/test/medium.factory.ts b/server/test/medium.factory.ts index 860189d9a9..3518c19aaf 100644 --- a/server/test/medium.factory.ts +++ b/server/test/medium.factory.ts @@ -338,7 +338,7 @@ export class SyncTestContext extends MediumTestContext { constructor(database: Kysely) { super(SyncService, { database, - real: [SyncRepository, SyncCheckpointRepository, SessionRepository], + real: [SyncRepository, SyncCheckpointRepository, SessionRepository, ConfigRepository, SystemMetadataRepository], mock: [LoggingRepository], }); } diff --git a/server/test/medium/specs/services/sync.service.spec.ts b/server/test/medium/specs/services/sync.service.spec.ts index c040d584b8..76108e7448 100644 --- a/server/test/medium/specs/services/sync.service.spec.ts +++ b/server/test/medium/specs/services/sync.service.spec.ts @@ -1,9 +1,11 @@ import { schemaFromCode } from '@immich/sql-tools'; import { Kysely } from 'kysely'; import { DateTime } from 'luxon'; +import { SystemConfig } from 'src/config'; import { AssetMetadataKey, UserMetadataKey } from 'src/enum'; import { DatabaseRepository } from 'src/repositories/database.repository'; import { LoggingRepository } from 'src/repositories/logging.repository'; +import { SessionRepository } from 'src/repositories/session.repository'; import { BaseSync, SyncRepository } from 'src/repositories/sync.repository'; import { DB } from 'src/schema'; import { SyncService } from 'src/services/sync.service'; @@ -16,7 +18,7 @@ let defaultDatabase: Kysely; const setup = (db?: Kysely) => { return newMediumService(SyncService, { database: db || defaultDatabase, - real: [DatabaseRepository, SyncRepository], + real: [DatabaseRepository, SyncRepository, SessionRepository], mock: [LoggingRepository], }); }; @@ -33,6 +35,17 @@ const assertTableCount = async (db: Kysely, t: T, count: expect(results).toHaveLength(count); }; +const withPublicUsers = (publicUsers: boolean) => ({ server: { publicUsers } }) as SystemConfig; + +const isPendingSyncReset = async (db: Kysely, sessionId: string) => { + const session = await db + .selectFrom('session') + .select('isPendingSyncReset') + .where('id', '=', sessionId) + .executeTakeFirstOrThrow(); + return session.isPendingSyncReset; +}; + describe(SyncService.name, () => { describe('onAuditTableCleanup', () => { it('should work', async () => { @@ -240,4 +253,56 @@ describe(SyncService.name, () => { } }); }); + + describe('onConfigUpdate', () => { + it('should require a full sync for non-admins when public users is disabled', async () => { + const { sut, ctx } = setup(await getKyselyDB()); + const { user } = await ctx.newUser(); + const { session } = await ctx.newSession({ userId: user.id }); + + await sut.onConfigUpdate({ oldConfig: withPublicUsers(true), newConfig: withPublicUsers(false) }); + + await expect(isPendingSyncReset(ctx.database, session.id)).resolves.toBe(true); + }); + + it('should not require a full sync for admins when public users is disabled', async () => { + const { sut, ctx } = setup(await getKyselyDB()); + const { user } = await ctx.newUser({ isAdmin: true }); + const { session } = await ctx.newSession({ userId: user.id }); + + await sut.onConfigUpdate({ oldConfig: withPublicUsers(true), newConfig: withPublicUsers(false) }); + + await expect(isPendingSyncReset(ctx.database, session.id)).resolves.toBe(false); + }); + + it('should require a full sync for non-admins when public users is enabled', async () => { + const { sut, ctx } = setup(await getKyselyDB()); + const { user } = await ctx.newUser(); + const { session } = await ctx.newSession({ userId: user.id }); + + await sut.onConfigUpdate({ oldConfig: withPublicUsers(false), newConfig: withPublicUsers(true) }); + + await expect(isPendingSyncReset(ctx.database, session.id)).resolves.toBe(true); + }); + + it('should not require a full sync when public users was already disabled', async () => { + const { sut, ctx } = setup(await getKyselyDB()); + const { user } = await ctx.newUser(); + const { session } = await ctx.newSession({ userId: user.id }); + + await sut.onConfigUpdate({ oldConfig: withPublicUsers(false), newConfig: withPublicUsers(false) }); + + await expect(isPendingSyncReset(ctx.database, session.id)).resolves.toBe(false); + }); + + it('should not require a full sync when public users was already enabled', async () => { + const { sut, ctx } = setup(await getKyselyDB()); + const { user } = await ctx.newUser(); + const { session } = await ctx.newSession({ userId: user.id }); + + await sut.onConfigUpdate({ oldConfig: withPublicUsers(true), newConfig: withPublicUsers(true) }); + + await expect(isPendingSyncReset(ctx.database, session.id)).resolves.toBe(false); + }); + }); }); diff --git a/server/test/medium/specs/sync/sync-user.spec.ts b/server/test/medium/specs/sync/sync-user.spec.ts index 7a69e7a411..52a36057b5 100644 --- a/server/test/medium/specs/sync/sync-user.spec.ts +++ b/server/test/medium/specs/sync/sync-user.spec.ts @@ -3,6 +3,7 @@ import { SyncEntityType, SyncRequestType } from 'src/enum'; import { UserRepository } from 'src/repositories/user.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; @@ -13,6 +14,15 @@ const setup = async (db?: Kysely) => { return { auth, user, session, ctx }; }; +const disablePublicUsers = async (ctx: SyncTestContext) => { + const config = await ctx.sut.getConfig({ withCache: false }); + config.server.publicUsers = false; + await ctx.sut.updateConfig(config); +}; + +const getSyncedUserIds = (response: Array<{ type: string; data: any }>) => + response.filter((item) => item.type === SyncEntityType.UserV1).map((item) => item.data.id); + beforeAll(async () => { defaultDatabase = await getKyselyDB(); }); @@ -134,4 +144,68 @@ describe(SyncEntityType.UserV1, () => { expect.objectContaining({ type: SyncEntityType.SyncCompleteV1 }), ]); }); + + describe('public users disabled', () => { + it('should only sync the authenticated user when there are no relationships', async () => { + const { auth, ctx } = await setup(await getKyselyDB()); + await disablePublicUsers(ctx); + + const { user: unrelated } = await ctx.newUser(); + + const response = await ctx.syncStream(auth, [SyncRequestType.UsersV1]); + const ids = getSyncedUserIds(response); + + expect(ids).toEqual([auth.user.id]); + expect(ids).not.toContain(unrelated.id); + }); + + it('should sync partners', async () => { + const { auth, ctx } = await setup(await getKyselyDB()); + await disablePublicUsers(ctx); + + const { user: partner } = await ctx.newUser(); + await ctx.newPartner({ sharedById: auth.user.id, sharedWithId: partner.id }); + const { user: unrelated } = await ctx.newUser(); + + const response = await ctx.syncStream(auth, [SyncRequestType.UsersV1]); + const ids = getSyncedUserIds(response); + + expect(ids).toEqual(expect.arrayContaining([auth.user.id, partner.id])); + expect(ids).not.toContain(unrelated.id); + }); + + it('should sync album co-members', async () => { + const { auth, ctx } = await setup(await getKyselyDB()); + await disablePublicUsers(ctx); + + const { user: coMember } = await ctx.newUser(); + const { album } = await ctx.newAlbum({ ownerId: auth.user.id }); + await ctx.newAlbumUser({ albumId: album.id, userId: coMember.id }); + const { user: unrelated } = await ctx.newUser(); + + const response = await ctx.syncStream(auth, [SyncRequestType.UsersV1]); + const ids = getSyncedUserIds(response); + + expect(ids).toEqual(expect.arrayContaining([auth.user.id, coMember.id])); + expect(ids).not.toContain(unrelated.id); + }); + + it('should still sync all users for an admin', async () => { + const { ctx } = await setup(await getKyselyDB()); + await disablePublicUsers(ctx); + + const { user: admin } = await ctx.newUser({ isAdmin: true }); + const { session: adminSession } = await ctx.newSession({ userId: admin.id }); + const adminAuth = factory.auth({ + session: adminSession, + user: { id: admin.id, name: admin.name, email: admin.email, isAdmin: true }, + }); + const { user: unrelated } = await ctx.newUser(); + + const response = await ctx.syncStream(adminAuth, [SyncRequestType.UsersV1]); + const ids = getSyncedUserIds(response); + + expect(ids).toEqual(expect.arrayContaining([admin.id, unrelated.id])); + }); + }); });