Merge 3cc2d61b32 into ffc83eae36

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"

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",

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) => {

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 {

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 } });
     }

server/test/medium.factory.ts

@@ -338,7 +338,7 @@ export class SyncTestContext extends MediumTestContext<typeof SyncService> {
   constructor(database: Kysely<DB>) {
     super(SyncService, {
       database,
-      real: [SyncRepository, SyncCheckpointRepository, SessionRepository],
+      real: [SyncRepository, SyncCheckpointRepository, SessionRepository, ConfigRepository, SystemMetadataRepository],
       mock: [LoggingRepository],
     });
   }

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<DB>;
 const setup = (db?: Kysely<DB>) => {
   return newMediumService(SyncService, {
     database: db || defaultDatabase,
-    real: [DatabaseRepository, SyncRepository],
+    real: [DatabaseRepository, SyncRepository, SessionRepository],
     mock: [LoggingRepository],
   });
 };
@@ -33,6 +35,17 @@ const assertTableCount = async <T extends keyof DB>(db: Kysely<DB>, t: T, count:
   expect(results).toHaveLength(count);
 };
 
+const withPublicUsers = (publicUsers: boolean) => ({ server: { publicUsers } }) as SystemConfig;
+
+const isPendingSyncReset = async (db: Kysely<DB>, 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);
+    });
+  });
 });

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<DB>;
@@ -13,6 +14,15 @@ const setup = async (db?: Kysely<DB>) => {
   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]));
+    });
+  });
 });