refactor(server): events (#13003)

* refactor(server): events

* chore: better type

---------

Co-authored-by: Daniel Dietzler <mail@ddietzler.dev>

server/src/services/database.service.ts

@@ -1,7 +1,7 @@
 import { Inject, Injectable } from '@nestjs/common';
 import { Duration } from 'luxon';
 import semver from 'semver';
-import { OnEmit } from 'src/decorators';
+import { OnEvent } from 'src/decorators';
 import { IConfigRepository } from 'src/interfaces/config.interface';
 import {
   DatabaseExtension,
@@ -74,7 +74,7 @@ export class DatabaseService {
     this.logger.setContext(DatabaseService.name);
   }
 
-  @OnEmit({ event: 'app.bootstrap', priority: -200 })
+  @OnEvent({ name: 'app.bootstrap', priority: -200 })
   async onBootstrap() {
     const version = await this.databaseRepository.getPostgresVersion();
     const current = semver.coerce(version);

server/src/services/job.service.spec.ts

@@ -1,6 +1,5 @@
 import { BadRequestException } from '@nestjs/common';
-import { SystemConfig } from 'src/config';
-import { SystemConfigCore } from 'src/cores/system-config.core';
+import { defaults } from 'src/config';
 import { IAssetRepository } from 'src/interfaces/asset.interface';
 import { IEventRepository } from 'src/interfaces/event.interface';
 import {
@@ -60,6 +59,19 @@ describe(JobService.name, () => {
     expect(sut).toBeDefined();
   });
 
+  describe('onConfigUpdate', () => {
+    it('should update concurrency', () => {
+      sut.onBootstrap('microservices');
+      sut.onConfigUpdate({ oldConfig: defaults, newConfig: defaults });
+
+      expect(jobMock.setConcurrency).toHaveBeenCalledTimes(14);
+      expect(jobMock.setConcurrency).toHaveBeenNthCalledWith(5, QueueName.FACIAL_RECOGNITION, 1);
+      expect(jobMock.setConcurrency).toHaveBeenNthCalledWith(7, QueueName.DUPLICATE_DETECTION, 1);
+      expect(jobMock.setConcurrency).toHaveBeenNthCalledWith(8, QueueName.BACKGROUND_TASK, 5);
+      expect(jobMock.setConcurrency).toHaveBeenNthCalledWith(9, QueueName.STORAGE_TEMPLATE_MIGRATION, 1);
+    });
+  });
+
   describe('handleNightlyJobs', () => {
     it('should run the scheduled jobs', async () => {
       await sut.handleNightlyJobs();
@@ -239,36 +251,6 @@ describe(JobService.name, () => {
       expect(jobMock.addHandler).toHaveBeenCalledTimes(Object.keys(QueueName).length);
     });
 
-    it('should subscribe to config changes', async () => {
-      await sut.init(makeMockHandlers(JobStatus.FAILED));
-
-      SystemConfigCore.create(newSystemMetadataRepositoryMock(false), newLoggerRepositoryMock()).config$.next({
-        job: {
-          [QueueName.BACKGROUND_TASK]: { concurrency: 10 },
-          [QueueName.SMART_SEARCH]: { concurrency: 10 },
-          [QueueName.METADATA_EXTRACTION]: { concurrency: 10 },
-          [QueueName.FACE_DETECTION]: { concurrency: 10 },
-          [QueueName.SEARCH]: { concurrency: 10 },
-          [QueueName.SIDECAR]: { concurrency: 10 },
-          [QueueName.LIBRARY]: { concurrency: 10 },
-          [QueueName.MIGRATION]: { concurrency: 10 },
-          [QueueName.THUMBNAIL_GENERATION]: { concurrency: 10 },
-          [QueueName.VIDEO_CONVERSION]: { concurrency: 10 },
-          [QueueName.NOTIFICATION]: { concurrency: 5 },
-        },
-      } as SystemConfig);
-
-      expect(jobMock.setConcurrency).toHaveBeenCalledWith(QueueName.BACKGROUND_TASK, 10);
-      expect(jobMock.setConcurrency).toHaveBeenCalledWith(QueueName.SMART_SEARCH, 10);
-      expect(jobMock.setConcurrency).toHaveBeenCalledWith(QueueName.METADATA_EXTRACTION, 10);
-      expect(jobMock.setConcurrency).toHaveBeenCalledWith(QueueName.FACE_DETECTION, 10);
-      expect(jobMock.setConcurrency).toHaveBeenCalledWith(QueueName.SIDECAR, 10);
-      expect(jobMock.setConcurrency).toHaveBeenCalledWith(QueueName.LIBRARY, 10);
-      expect(jobMock.setConcurrency).toHaveBeenCalledWith(QueueName.MIGRATION, 10);
-      expect(jobMock.setConcurrency).toHaveBeenCalledWith(QueueName.THUMBNAIL_GENERATION, 10);
-      expect(jobMock.setConcurrency).toHaveBeenCalledWith(QueueName.VIDEO_CONVERSION, 10);
-    });
-
     const tests: Array<{ item: JobItem; jobs: JobName[] }> = [
       {
         item: { name: JobName.SIDECAR_SYNC, data: { id: 'asset-1' } },

server/src/services/job.service.ts

@@ -1,11 +1,12 @@
 import { BadRequestException, Inject, Injectable } from '@nestjs/common';
 import { snakeCase } from 'lodash';
 import { SystemConfigCore } from 'src/cores/system-config.core';
+import { OnEvent } from 'src/decorators';
 import { mapAsset } from 'src/dtos/asset-response.dto';
 import { AllJobStatusResponseDto, JobCommandDto, JobCreateDto, JobStatusDto } from 'src/dtos/job.dto';
 import { AssetType, ManualJobName } from 'src/enum';
 import { IAssetRepository } from 'src/interfaces/asset.interface';
-import { ClientEvent, IEventRepository } from 'src/interfaces/event.interface';
+import { ArgOf, ClientEvent, IEventRepository } from 'src/interfaces/event.interface';
 import {
   ConcurrentQueueName,
   IJobRepository,
@@ -45,6 +46,7 @@ const asJobItem = (dto: JobCreateDto): JobItem => {
 @Injectable()
 export class JobService {
   private configCore: SystemConfigCore;
+  private isMicroservices = false;
 
   constructor(
     @Inject(IAssetRepository) private assetRepository: IAssetRepository,
@@ -59,6 +61,28 @@ export class JobService {
     this.configCore = SystemConfigCore.create(systemMetadataRepository, logger);
   }
 
+  @OnEvent({ name: 'app.bootstrap' })
+  onBootstrap(app: ArgOf<'app.bootstrap'>) {
+    this.isMicroservices = app === 'microservices';
+  }
+
+  @OnEvent({ name: 'config.update', server: true })
+  onConfigUpdate({ newConfig: config, oldConfig }: ArgOf<'config.update'>) {
+    if (!oldConfig || !this.isMicroservices) {
+      return;
+    }
+
+    this.logger.debug(`Updating queue concurrency settings`);
+    for (const queueName of Object.values(QueueName)) {
+      let concurrency = 1;
+      if (this.isConcurrentQueue(queueName)) {
+        concurrency = config.job[queueName].concurrency;
+      }
+      this.logger.debug(`Setting ${queueName} concurrency to ${concurrency}`);
+      this.jobRepository.setConcurrency(queueName, concurrency);
+    }
+  }
+
   async create(dto: JobCreateDto): Promise<void> {
     await this.jobRepository.queue(asJobItem(dto));
   }
@@ -209,18 +233,6 @@ export class JobService {
         }
       });
     }
-
-    this.configCore.config$.subscribe((config) => {
-      this.logger.debug(`Updating queue concurrency settings`);
-      for (const queueName of Object.values(QueueName)) {
-        let concurrency = 1;
-        if (this.isConcurrentQueue(queueName)) {
-          concurrency = config.job[queueName].concurrency;
-        }
-        this.logger.debug(`Setting ${queueName} concurrency to ${concurrency}`);
-        this.jobRepository.setConcurrency(queueName, concurrency);
-      }
-    });
   }
 
   private isConcurrentQueue(name: QueueName): name is ConcurrentQueueName {

server/src/services/library.service.spec.ts

@@ -1,7 +1,6 @@
 import { BadRequestException } from '@nestjs/common';
 import { Stats } from 'node:fs';
-import { SystemConfig } from 'src/config';
-import { SystemConfigCore } from 'src/cores/system-config.core';
+import { defaults, SystemConfig } from 'src/config';
 import { mapLibrary } from 'src/dtos/library.dto';
 import { UserEntity } from 'src/entities/user.entity';
 import { AssetType } from 'src/enum';
@@ -81,22 +80,26 @@ describe(LibraryService.name, () => {
   });
 
   describe('onBootstrapEvent', () => {
-    it('should init cron job and subscribe to config changes', async () => {
+    it('should init cron job and handle config changes', async () => {
       systemMock.get.mockResolvedValue(systemConfigStub.libraryScan);
 
       await sut.onBootstrap();
-      expect(systemMock.get).toHaveBeenCalled();
-      expect(jobMock.addCronJob).toHaveBeenCalled();
 
-      SystemConfigCore.create(newSystemMetadataRepositoryMock(false), newLoggerRepositoryMock()).config$.next({
-        library: {
-          scan: {
-            enabled: true,
-            cronExpression: '0 1 * * *',
+      expect(jobMock.addCronJob).toHaveBeenCalled();
+      expect(systemMock.get).toHaveBeenCalled();
+
+      await sut.onConfigUpdate({
+        oldConfig: defaults,
+        newConfig: {
+          library: {
+            scan: {
+              enabled: true,
+              cronExpression: '0 1 * * *',
+            },
+            watch: { enabled: false },
           },
-          watch: { enabled: true },
-        },
-      } as SystemConfig);
+        } as SystemConfig,
+      });
 
       expect(jobMock.updateCronJob).toHaveBeenCalledWith('libraryScan', '0 1 * * *', true);
     });

server/src/services/library.service.ts

@@ -4,7 +4,7 @@ import path, { basename, parse } from 'node:path';
 import picomatch from 'picomatch';
 import { StorageCore } from 'src/cores/storage.core';
 import { SystemConfigCore } from 'src/cores/system-config.core';
-import { OnEmit } from 'src/decorators';
+import { OnEvent } from 'src/decorators';
 import {
   CreateLibraryDto,
   LibraryResponseDto,
@@ -61,7 +61,7 @@ export class LibraryService {
     this.configCore = SystemConfigCore.create(systemMetadataRepository, this.logger);
   }
 
-  @OnEmit({ event: 'app.bootstrap' })
+  @OnEvent({ name: 'app.bootstrap' })
   async onBootstrap() {
     const config = await this.configCore.getConfig({ withCache: false });
 
@@ -83,19 +83,24 @@ export class LibraryService {
     if (this.watchLibraries) {
       await this.watchAll();
     }
-
-    this.configCore.config$.subscribe(({ library }) => {
-      this.jobRepository.updateCronJob('libraryScan', library.scan.cronExpression, library.scan.enabled);
-
-      if (library.watch.enabled !== this.watchLibraries) {
-        // Watch configuration changed, update accordingly
-        this.watchLibraries = library.watch.enabled;
-        handlePromiseError(this.watchLibraries ? this.watchAll() : this.unwatchAll(), this.logger);
-      }
-    });
   }
 
-  @OnEmit({ event: 'config.validate' })
+  @OnEvent({ name: 'config.update', server: true })
+  async onConfigUpdate({ newConfig: { library }, oldConfig }: ArgOf<'config.update'>) {
+    if (!oldConfig || !this.watchLock) {
+      return;
+    }
+
+    this.jobRepository.updateCronJob('libraryScan', library.scan.cronExpression, library.scan.enabled);
+
+    if (library.watch.enabled !== this.watchLibraries) {
+      // Watch configuration changed, update accordingly
+      this.watchLibraries = library.watch.enabled;
+      await (this.watchLibraries ? this.watchAll() : this.unwatchAll());
+    }
+  }
+
+  @OnEvent({ name: 'config.validate' })
   onConfigValidate({ newConfig }: ArgOf<'config.validate'>) {
     const { scan } = newConfig.library;
     if (!validateCronExpression(scan.cronExpression)) {
@@ -185,7 +190,7 @@ export class LibraryService {
     }
   }
 
-  @OnEmit({ event: 'app.shutdown' })
+  @OnEvent({ name: 'app.shutdown' })
   async onShutdown() {
     await this.unwatchAll();
   }

server/src/services/metadata.service.ts

@@ -8,7 +8,7 @@ import path from 'node:path';
 import { SystemConfig } from 'src/config';
 import { StorageCore } from 'src/cores/storage.core';
 import { SystemConfigCore } from 'src/cores/system-config.core';
-import { OnEmit } from 'src/decorators';
+import { OnEvent } from 'src/decorators';
 import { AssetFaceEntity } from 'src/entities/asset-face.entity';
 import { AssetEntity } from 'src/entities/asset.entity';
 import { ExifEntity } from 'src/entities/exif.entity';
@@ -132,7 +132,7 @@ export class MetadataService {
     );
   }
 
-  @OnEmit({ event: 'app.bootstrap' })
+  @OnEvent({ name: 'app.bootstrap' })
   async onBootstrap(app: ArgOf<'app.bootstrap'>) {
     if (app !== 'microservices') {
       return;
@@ -141,7 +141,12 @@ export class MetadataService {
     await this.init(config);
   }
 
-  @OnEmit({ event: 'config.update' })
+  @OnEvent({ name: 'app.shutdown' })
+  async onShutdown() {
+    await this.repository.teardown();
+  }
+
+  @OnEvent({ name: 'config.update' })
   async onConfigUpdate({ newConfig }: ArgOf<'config.update'>) {
     await this.init(newConfig);
   }
@@ -164,11 +169,6 @@ export class MetadataService {
     }
   }
 
-  @OnEmit({ event: 'app.shutdown' })
-  async onShutdown() {
-    await this.repository.teardown();
-  }
-
   async handleLivePhotoLinking(job: IEntityJob): Promise<JobStatus> {
     const { id } = job;
     const [asset] = await this.assetRepository.getByIds([id], { exifInfo: true });
@@ -333,12 +333,12 @@ export class MetadataService {
     return this.processSidecar(id, false);
   }
 
-  @OnEmit({ event: 'asset.tag' })
+  @OnEvent({ name: 'asset.tag' })
   async handleTagAsset({ assetId }: ArgOf<'asset.tag'>) {
     await this.jobRepository.queue({ name: JobName.SIDECAR_WRITE, data: { id: assetId, tags: true } });
   }
 
-  @OnEmit({ event: 'asset.untag' })
+  @OnEvent({ name: 'asset.untag' })
   async handleUntagAsset({ assetId }: ArgOf<'asset.untag'>) {
     await this.jobRepository.queue({ name: JobName.SIDECAR_WRITE, data: { id: assetId, tags: true } });
   }

server/src/services/microservices.service.ts

@@ -1,5 +1,5 @@
 import { Injectable } from '@nestjs/common';
-import { OnEmit } from 'src/decorators';
+import { OnEvent } from 'src/decorators';
 import { ArgOf } from 'src/interfaces/event.interface';
 import { IDeleteFilesJob, JobName } from 'src/interfaces/job.interface';
 import { AssetService } from 'src/services/asset.service';
@@ -43,7 +43,7 @@ export class MicroservicesService {
     private versionService: VersionService,
   ) {}
 
-  @OnEmit({ event: 'app.bootstrap' })
+  @OnEvent({ name: 'app.bootstrap' })
   async onBootstrap(app: ArgOf<'app.bootstrap'>) {
     if (app !== 'microservices') {
       return;

server/src/services/notification.service.spec.ts

@@ -6,7 +6,7 @@ import { AssetFileEntity } from 'src/entities/asset-files.entity';
 import { AssetFileType, UserMetadataKey } from 'src/enum';
 import { IAlbumRepository } from 'src/interfaces/album.interface';
 import { IAssetRepository } from 'src/interfaces/asset.interface';
-import { IEventRepository } from 'src/interfaces/event.interface';
+import { ClientEvent, IEventRepository } from 'src/interfaces/event.interface';
 import { IJobRepository, JobName, JobStatus } from 'src/interfaces/job.interface';
 import { ILoggerRepository } from 'src/interfaces/logger.interface';
 import { EmailTemplate, INotificationRepository } from 'src/interfaces/notification.interface';
@@ -100,6 +100,15 @@ describe(NotificationService.name, () => {
     expect(sut).toBeDefined();
   });
 
+  describe('onConfigUpdate', () => {
+    it('should emit client and server events', () => {
+      const update = { newConfig: defaults };
+      expect(sut.onConfigUpdate(update)).toBeUndefined();
+      expect(eventMock.clientBroadcast).toHaveBeenCalledWith(ClientEvent.CONFIG_UPDATE, {});
+      expect(eventMock.serverSend).toHaveBeenCalledWith('config.update', update);
+    });
+  });
+
   describe('onConfigValidateEvent', () => {
     it('validates smtp config when enabling smtp', async () => {
       const oldConfig = configs.smtpDisabled;

server/src/services/notification.service.ts

@@ -1,7 +1,7 @@
 import { BadRequestException, Inject, Injectable } from '@nestjs/common';
 import { DEFAULT_EXTERNAL_DOMAIN } from 'src/constants';
 import { SystemConfigCore } from 'src/cores/system-config.core';
-import { OnEmit } from 'src/decorators';
+import { OnEvent } from 'src/decorators';
 import { SystemConfigSmtpDto } from 'src/dtos/system-config.dto';
 import { AlbumEntity } from 'src/entities/album.entity';
 import { IAlbumRepository } from 'src/interfaces/album.interface';
@@ -43,7 +43,13 @@ export class NotificationService {
     this.configCore = SystemConfigCore.create(systemMetadataRepository, logger);
   }
 
-  @OnEmit({ event: 'config.validate', priority: -100 })
+  @OnEvent({ name: 'config.update' })
+  onConfigUpdate({ oldConfig, newConfig }: ArgOf<'config.update'>) {
+    this.eventRepository.clientBroadcast(ClientEvent.CONFIG_UPDATE, {});
+    this.eventRepository.serverSend('config.update', { oldConfig, newConfig });
+  }
+
+  @OnEvent({ name: 'config.validate', priority: -100 })
   async onConfigValidate({ oldConfig, newConfig }: ArgOf<'config.validate'>) {
     try {
       if (
@@ -58,74 +64,74 @@ export class NotificationService {
     }
   }
 
-  @OnEmit({ event: 'asset.hide' })
+  @OnEvent({ name: 'asset.hide' })
   onAssetHide({ assetId, userId }: ArgOf<'asset.hide'>) {
     this.eventRepository.clientSend(ClientEvent.ASSET_HIDDEN, userId, assetId);
   }
 
-  @OnEmit({ event: 'asset.show' })
+  @OnEvent({ name: 'asset.show' })
   async onAssetShow({ assetId }: ArgOf<'asset.show'>) {
     await this.jobRepository.queue({ name: JobName.GENERATE_THUMBNAILS, data: { id: assetId, notify: true } });
   }
 
-  @OnEmit({ event: 'asset.trash' })
+  @OnEvent({ name: 'asset.trash' })
   onAssetTrash({ assetId, userId }: ArgOf<'asset.trash'>) {
     this.eventRepository.clientSend(ClientEvent.ASSET_TRASH, userId, [assetId]);
   }
 
-  @OnEmit({ event: 'asset.delete' })
+  @OnEvent({ name: 'asset.delete' })
   onAssetDelete({ assetId, userId }: ArgOf<'asset.delete'>) {
     this.eventRepository.clientSend(ClientEvent.ASSET_DELETE, userId, assetId);
   }
 
-  @OnEmit({ event: 'assets.trash' })
+  @OnEvent({ name: 'assets.trash' })
   onAssetsTrash({ assetIds, userId }: ArgOf<'assets.trash'>) {
     this.eventRepository.clientSend(ClientEvent.ASSET_TRASH, userId, assetIds);
   }
 
-  @OnEmit({ event: 'assets.restore' })
+  @OnEvent({ name: 'assets.restore' })
   onAssetsRestore({ assetIds, userId }: ArgOf<'assets.restore'>) {
     this.eventRepository.clientSend(ClientEvent.ASSET_RESTORE, userId, assetIds);
   }
 
-  @OnEmit({ event: 'stack.create' })
+  @OnEvent({ name: 'stack.create' })
   onStackCreate({ userId }: ArgOf<'stack.create'>) {
     this.eventRepository.clientSend(ClientEvent.ASSET_STACK_UPDATE, userId, []);
   }
 
-  @OnEmit({ event: 'stack.update' })
+  @OnEvent({ name: 'stack.update' })
   onStackUpdate({ userId }: ArgOf<'stack.update'>) {
     this.eventRepository.clientSend(ClientEvent.ASSET_STACK_UPDATE, userId, []);
   }
 
-  @OnEmit({ event: 'stack.delete' })
+  @OnEvent({ name: 'stack.delete' })
   onStackDelete({ userId }: ArgOf<'stack.delete'>) {
     this.eventRepository.clientSend(ClientEvent.ASSET_STACK_UPDATE, userId, []);
   }
 
-  @OnEmit({ event: 'stacks.delete' })
+  @OnEvent({ name: 'stacks.delete' })
   onStacksDelete({ userId }: ArgOf<'stacks.delete'>) {
     this.eventRepository.clientSend(ClientEvent.ASSET_STACK_UPDATE, userId, []);
   }
 
-  @OnEmit({ event: 'user.signup' })
+  @OnEvent({ name: 'user.signup' })
   async onUserSignup({ notify, id, tempPassword }: ArgOf<'user.signup'>) {
     if (notify) {
       await this.jobRepository.queue({ name: JobName.NOTIFY_SIGNUP, data: { id, tempPassword } });
     }
   }
 
-  @OnEmit({ event: 'album.update' })
+  @OnEvent({ name: 'album.update' })
   async onAlbumUpdate({ id, updatedBy }: ArgOf<'album.update'>) {
     await this.jobRepository.queue({ name: JobName.NOTIFY_ALBUM_UPDATE, data: { id, senderId: updatedBy } });
   }
 
-  @OnEmit({ event: 'album.invite' })
+  @OnEvent({ name: 'album.invite' })
   async onAlbumInvite({ id, userId }: ArgOf<'album.invite'>) {
     await this.jobRepository.queue({ name: JobName.NOTIFY_ALBUM_INVITE, data: { id, recipientId: userId } });
   }
 
-  @OnEmit({ event: 'session.delete' })
+  @OnEvent({ name: 'session.delete' })
   onSessionDelete({ sessionId }: ArgOf<'session.delete'>) {
     // after the response is sent
     setTimeout(() => this.eventRepository.clientSend(ClientEvent.SESSION_DELETE, sessionId, sessionId), 500);

server/src/services/server.service.ts

@@ -3,7 +3,7 @@ import { getBuildMetadata, getServerLicensePublicKey } from 'src/config';
 import { serverVersion } from 'src/constants';
 import { StorageCore } from 'src/cores/storage.core';
 import { SystemConfigCore } from 'src/cores/system-config.core';
-import { OnEmit } from 'src/decorators';
+import { OnEvent } from 'src/decorators';
 import { LicenseKeyDto, LicenseResponseDto } from 'src/dtos/license.dto';
 import {
   ServerAboutResponseDto,
@@ -42,7 +42,7 @@ export class ServerService {
     this.configCore = SystemConfigCore.create(systemMetadataRepository, this.logger);
   }
 
-  @OnEmit({ event: 'app.bootstrap' })
+  @OnEvent({ name: 'app.bootstrap' })
   async onBootstrap(): Promise<void> {
     const featureFlags = await this.getFeatures();
     if (featureFlags.configFile) {

server/src/services/smart-info.service.ts

@@ -1,7 +1,7 @@
 import { Inject, Injectable } from '@nestjs/common';
 import { SystemConfig } from 'src/config';
 import { SystemConfigCore } from 'src/cores/system-config.core';
-import { OnEmit } from 'src/decorators';
+import { OnEvent } from 'src/decorators';
 import { IAssetRepository, WithoutProperty } from 'src/interfaces/asset.interface';
 import { DatabaseLock, IDatabaseRepository } from 'src/interfaces/database.interface';
 import { ArgOf } from 'src/interfaces/event.interface';
@@ -39,7 +39,7 @@ export class SmartInfoService {
     this.configCore = SystemConfigCore.create(systemMetadataRepository, this.logger);
   }
 
-  @OnEmit({ event: 'app.bootstrap' })
+  @OnEvent({ name: 'app.bootstrap' })
   async onBootstrap(app: ArgOf<'app.bootstrap'>) {
     if (app !== 'microservices') {
       return;
@@ -49,7 +49,12 @@ export class SmartInfoService {
     await this.init(config);
   }
 
-  @OnEmit({ event: 'config.validate' })
+  @OnEvent({ name: 'config.update' })
+  async onConfigUpdate({ oldConfig, newConfig }: ArgOf<'config.update'>) {
+    await this.init(newConfig, oldConfig);
+  }
+
+  @OnEvent({ name: 'config.validate' })
   onConfigValidate({ newConfig }: ArgOf<'config.validate'>) {
     try {
       getCLIPModelInfo(newConfig.machineLearning.clip.modelName);
@@ -60,11 +65,6 @@ export class SmartInfoService {
     }
   }
 
-  @OnEmit({ event: 'config.update' })
-  async onConfigUpdate({ oldConfig, newConfig }: ArgOf<'config.update'>) {
-    await this.init(newConfig, oldConfig);
-  }
-
   private async init(newConfig: SystemConfig, oldConfig?: SystemConfig) {
     if (!isSmartSearchEnabled(newConfig.machineLearning)) {
       return;

server/src/services/storage-template.service.spec.ts

@@ -1,6 +1,5 @@
 import { Stats } from 'node:fs';
 import { SystemConfig, defaults } from 'src/config';
-import { SystemConfigCore } from 'src/cores/system-config.core';
 import { AssetEntity } from 'src/entities/asset.entity';
 import { AssetPathType } from 'src/enum';
 import { IAlbumRepository } from 'src/interfaces/album.interface';
@@ -74,7 +73,7 @@ describe(StorageTemplateService.name, () => {
       loggerMock,
     );
 
-    SystemConfigCore.create(systemMock, loggerMock).config$.next(defaults);
+    sut.onConfigUpdate({ newConfig: defaults });
   });
 
   describe('onConfigValidate', () => {
@@ -164,13 +163,15 @@ describe(StorageTemplateService.name, () => {
         originalPath: newMotionPicturePath,
       });
     });
-    it('Should use handlebar if condition for album', async () => {
+
+    it('should use handlebar if condition for album', async () => {
       const asset = assetStub.image;
       const user = userStub.user1;
       const album = albumStub.oneAsset;
       const config = structuredClone(defaults);
       config.storageTemplate.template = '{{y}}/{{#if album}}{{album}}{{else}}other/{{MM}}{{/if}}/{{filename}}';
-      SystemConfigCore.create(systemMock, loggerMock).config$.next(config);
+
+      sut.onConfigUpdate({ oldConfig: defaults, newConfig: config });
 
       userMock.get.mockResolvedValue(user);
       assetMock.getByIds.mockResolvedValueOnce([asset]);
@@ -185,12 +186,13 @@ describe(StorageTemplateService.name, () => {
         pathType: AssetPathType.ORIGINAL,
       });
     });
-    it('Should use handlebar else condition for album', async () => {
+
+    it('should use handlebar else condition for album', async () => {
       const asset = assetStub.image;
       const user = userStub.user1;
       const config = structuredClone(defaults);
       config.storageTemplate.template = '{{y}}/{{#if album}}{{album}}{{else}}other//{{MM}}{{/if}}/{{filename}}';
-      SystemConfigCore.create(systemMock, loggerMock).config$.next(config);
+      sut.onConfigUpdate({ oldConfig: defaults, newConfig: config });
 
       userMock.get.mockResolvedValue(user);
       assetMock.getByIds.mockResolvedValueOnce([asset]);
@@ -205,6 +207,7 @@ describe(StorageTemplateService.name, () => {
         pathType: AssetPathType.ORIGINAL,
       });
     });
+
     it('should migrate previously failed move from original path when it still exists', async () => {
       userMock.get.mockResolvedValue(userStub.user1);
       const previousFailedNewPath = `upload/library/${userStub.user1.id}/2023/Feb/${assetStub.image.id}.jpg`;
@@ -242,6 +245,7 @@ describe(StorageTemplateService.name, () => {
         originalPath: newPath,
       });
     });
+
     it('should migrate previously failed move from previous new path when old path no longer exists, should validate file size still matches before moving', async () => {
       userMock.get.mockResolvedValue(userStub.user1);
       const previousFailedNewPath = `upload/library/${userStub.user1.id}/2023/Feb/${assetStub.image.id}.jpg`;

server/src/services/storage-template.service.ts

@@ -3,7 +3,6 @@ import handlebar from 'handlebars';
 import { DateTime } from 'luxon';
 import path from 'node:path';
 import sanitize from 'sanitize-filename';
-import { SystemConfig } from 'src/config';
 import {
   supportedDayTokens,
   supportedHourTokens,
@@ -15,7 +14,7 @@ import {
 } from 'src/constants';
 import { StorageCore } from 'src/cores/storage.core';
 import { SystemConfigCore } from 'src/cores/system-config.core';
-import { OnEmit } from 'src/decorators';
+import { OnEvent } from 'src/decorators';
 import { AssetEntity } from 'src/entities/asset.entity';
 import { AssetPathType, AssetType, StorageFolder } from 'src/enum';
 import { IAlbumRepository } from 'src/interfaces/album.interface';
@@ -76,7 +75,6 @@ export class StorageTemplateService {
   ) {
     this.logger.setContext(StorageTemplateService.name);
     this.configCore = SystemConfigCore.create(systemMetadataRepository, this.logger);
-    this.configCore.config$.subscribe((config) => this.onConfig(config));
     this.storageCore = StorageCore.create(
       assetRepository,
       cryptoRepository,
@@ -88,7 +86,16 @@ export class StorageTemplateService {
     );
   }
 
-  @OnEmit({ event: 'config.validate' })
+  @OnEvent({ name: 'config.update', server: true })
+  onConfigUpdate({ newConfig }: ArgOf<'config.update'>) {
+    const template = newConfig.storageTemplate.template;
+    if (!this._template || template !== this.template.raw) {
+      this.logger.debug(`Compiling new storage template: ${template}`);
+      this._template = this.compile(template);
+    }
+  }
+
+  @OnEvent({ name: 'config.validate' })
   onConfigValidate({ newConfig }: ArgOf<'config.validate'>) {
     try {
       const { compiled } = this.compile(newConfig.storageTemplate.template);
@@ -282,14 +289,6 @@ export class StorageTemplateService {
     }
   }
 
-  private onConfig(config: SystemConfig) {
-    const template = config.storageTemplate.template;
-    if (!this._template || template !== this.template.raw) {
-      this.logger.debug(`Compiling new storage template: ${template}`);
-      this._template = this.compile(template);
-    }
-  }
-
   private compile(template: string) {
     return {
       raw: template,

server/src/services/storage.service.ts

@@ -1,7 +1,7 @@
 import { Inject, Injectable } from '@nestjs/common';
 import { join } from 'node:path';
 import { StorageCore } from 'src/cores/storage.core';
-import { OnEmit } from 'src/decorators';
+import { OnEvent } from 'src/decorators';
 import { StorageFolder, SystemMetadataKey } from 'src/enum';
 import { DatabaseLock, IDatabaseRepository } from 'src/interfaces/database.interface';
 import { IDeleteFilesJob, JobStatus } from 'src/interfaces/job.interface';
@@ -21,7 +21,7 @@ export class StorageService {
     this.logger.setContext(StorageService.name);
   }
 
-  @OnEmit({ event: 'app.bootstrap' })
+  @OnEvent({ name: 'app.bootstrap' })
   async onBootstrap() {
     await this.databaseRepository.withLock(DatabaseLock.SystemFileMounts, async () => {
       const flags = (await this.systemMetadata.get(SystemMetadataKey.SYSTEM_FLAGS)) || { mountFiles: false };

server/src/services/system-config.service.spec.ts

@@ -6,14 +6,13 @@ import {
   CQMode,
   ImageFormat,
   LogLevel,
-  SystemMetadataKey,
   ToneMapping,
   TranscodeHWAccel,
   TranscodePolicy,
   VideoCodec,
   VideoContainer,
 } from 'src/enum';
-import { IEventRepository, ServerEvent } from 'src/interfaces/event.interface';
+import { IEventRepository } from 'src/interfaces/event.interface';
 import { QueueName } from 'src/interfaces/job.interface';
 import { ILoggerRepository } from 'src/interfaces/logger.interface';
 import { ISystemMetadataRepository } from 'src/interfaces/system-metadata.interface';
@@ -381,14 +380,13 @@ describe(SystemConfigService.name, () => {
   });
 
   describe('updateConfig', () => {
-    it('should update the config and emit client and server events', async () => {
+    it('should update the config and emit an event', async () => {
       systemMock.get.mockResolvedValue(partialConfig);
-
       await expect(sut.updateConfig(updatedConfig)).resolves.toEqual(updatedConfig);
-
-      expect(eventMock.clientBroadcast).toHaveBeenCalled();
-      expect(eventMock.serverSend).toHaveBeenCalledWith(ServerEvent.CONFIG_UPDATE, null);
-      expect(systemMock.set).toHaveBeenCalledWith(SystemMetadataKey.SYSTEM_CONFIG, partialConfig);
+      expect(eventMock.emit).toHaveBeenCalledWith(
+        'config.update',
+        expect.objectContaining({ oldConfig: expect.any(Object), newConfig: updatedConfig }),
+      );
     });
 
     it('should throw an error if a config file is in use', async () => {

server/src/services/system-config.service.ts

@@ -1,7 +1,7 @@
 import { BadRequestException, Inject, Injectable } from '@nestjs/common';
 import { instanceToPlain } from 'class-transformer';
 import _ from 'lodash';
-import { SystemConfig, defaults } from 'src/config';
+import { defaults } from 'src/config';
 import {
   supportedDayTokens,
   supportedHourTokens,
@@ -13,10 +13,10 @@ import {
   supportedYearTokens,
 } from 'src/constants';
 import { SystemConfigCore } from 'src/cores/system-config.core';
-import { OnEmit, OnServerEvent } from 'src/decorators';
+import { OnEvent } from 'src/decorators';
 import { SystemConfigDto, SystemConfigTemplateStorageOptionDto, mapConfig } from 'src/dtos/system-config.dto';
 import { LogLevel } from 'src/enum';
-import { ArgOf, ClientEvent, IEventRepository, ServerEvent } from 'src/interfaces/event.interface';
+import { ArgOf, IEventRepository } from 'src/interfaces/event.interface';
 import { ILoggerRepository } from 'src/interfaces/logger.interface';
 import { ISystemMetadataRepository } from 'src/interfaces/system-metadata.interface';
 import { toPlainObject } from 'src/utils/object';
@@ -32,13 +32,12 @@ export class SystemConfigService {
   ) {
     this.logger.setContext(SystemConfigService.name);
     this.core = SystemConfigCore.create(repository, this.logger);
-    this.core.config$.subscribe((config) => this.setLogLevel(config));
   }
 
-  @OnEmit({ event: 'app.bootstrap', priority: -100 })
+  @OnEvent({ name: 'app.bootstrap', priority: -100 })
   async onBootstrap() {
     const config = await this.core.getConfig({ withCache: false });
-    this.core.config$.next(config);
+    await this.eventRepository.emit('config.update', { newConfig: config });
   }
 
   async getConfig(): Promise<SystemConfigDto> {
@@ -50,7 +49,18 @@ export class SystemConfigService {
     return mapConfig(defaults);
   }
 
-  @OnEmit({ event: 'config.validate' })
+  @OnEvent({ name: 'config.update', server: true })
+  onConfigUpdate({ newConfig: { logging } }: ArgOf<'config.update'>) {
+    const envLevel = this.getEnvLogLevel();
+    const configLevel = logging.enabled ? logging.level : false;
+    const level = envLevel ?? configLevel;
+    this.logger.setLogLevel(level);
+    this.logger.log(`LogLevel=${level} ${envLevel ? '(set via IMMICH_LOG_LEVEL)' : '(set via system config)'}`);
+    // TODO only do this if the event is a socket.io event
+    this.core.invalidateCache();
+  }
+
+  @OnEvent({ name: 'config.validate' })
   onConfigValidate({ newConfig, oldConfig }: ArgOf<'config.validate'>) {
     if (!_.isEqual(instanceToPlain(newConfig.logging), oldConfig.logging) && this.getEnvLogLevel()) {
       throw new Error('Logging cannot be changed while the environment variable IMMICH_LOG_LEVEL is set.');
@@ -73,9 +83,6 @@ export class SystemConfigService {
 
     const newConfig = await this.core.updateConfig(dto);
 
-    // TODO probably move web socket emits to a separate service
-    this.eventRepository.clientBroadcast(ClientEvent.CONFIG_UPDATE, {});
-    this.eventRepository.serverSend(ServerEvent.CONFIG_UPDATE, null);
     await this.eventRepository.emit('config.update', { newConfig, oldConfig });
 
     return mapConfig(newConfig);
@@ -101,19 +108,6 @@ export class SystemConfigService {
     return theme.customCss;
   }
 
-  @OnServerEvent(ServerEvent.CONFIG_UPDATE)
-  async onConfigUpdateEvent() {
-    await this.core.refreshConfig();
-  }
-
-  private setLogLevel({ logging }: SystemConfig) {
-    const envLevel = this.getEnvLogLevel();
-    const configLevel = logging.enabled ? logging.level : false;
-    const level = envLevel ?? configLevel;
-    this.logger.setLogLevel(level);
-    this.logger.log(`LogLevel=${level} ${envLevel ? '(set via IMMICH_LOG_LEVEL)' : '(set via system config)'}`);
-  }
-
   private getEnvLogLevel() {
     return process.env.IMMICH_LOG_LEVEL as LogLevel;
   }

server/src/services/trash.service.ts

@@ -1,5 +1,5 @@
 import { Inject } from '@nestjs/common';
-import { OnEmit } from 'src/decorators';
+import { OnEvent } from 'src/decorators';
 import { BulkIdsDto } from 'src/dtos/asset-ids.response.dto';
 import { AuthDto } from 'src/dtos/auth.dto';
 import { TrashResponseDto } from 'src/dtos/trash.dto';
@@ -54,7 +54,7 @@ export class TrashService {
     return { count };
   }
 
-  @OnEmit({ event: 'assets.delete' })
+  @OnEvent({ name: 'assets.delete' })
   async onAssetsDelete() {
     await this.jobRepository.queue({ name: JobName.QUEUE_TRASH_EMPTY, data: {} });
   }

server/src/services/version.service.ts

@@ -3,11 +3,11 @@ import { DateTime } from 'luxon';
 import semver, { SemVer } from 'semver';
 import { isDev, serverVersion } from 'src/constants';
 import { SystemConfigCore } from 'src/cores/system-config.core';
-import { OnEmit, OnServerEvent } from 'src/decorators';
+import { OnEvent } from 'src/decorators';
 import { ReleaseNotification, ServerVersionResponseDto } from 'src/dtos/server.dto';
 import { VersionCheckMetadata } from 'src/entities/system-metadata.entity';
 import { SystemMetadataKey } from 'src/enum';
-import { ClientEvent, IEventRepository, ServerEvent, ServerEventMap } from 'src/interfaces/event.interface';
+import { ArgOf, ClientEvent, IEventRepository } from 'src/interfaces/event.interface';
 import { IJobRepository, JobName, JobStatus } from 'src/interfaces/job.interface';
 import { ILoggerRepository } from 'src/interfaces/logger.interface';
 import { IServerInfoRepository } from 'src/interfaces/server-info.interface';
@@ -37,7 +37,7 @@ export class VersionService {
     this.configCore = SystemConfigCore.create(systemMetadataRepository, this.logger);
   }
 
-  @OnEmit({ event: 'app.bootstrap' })
+  @OnEvent({ name: 'app.bootstrap' })
   async onBootstrap(): Promise<void> {
     await this.handleVersionCheck();
   }
@@ -90,8 +90,8 @@ export class VersionService {
     return JobStatus.SUCCESS;
   }
 
-  @OnServerEvent(ServerEvent.WEBSOCKET_CONNECT)
-  async onWebsocketConnection({ userId }: ServerEventMap[ServerEvent.WEBSOCKET_CONNECT]) {
+  @OnEvent({ name: 'websocket.connect' })
+  async onWebsocketConnection({ userId }: ArgOf<'websocket.connect'>) {
     this.eventRepository.clientSend(ClientEvent.SERVER_VERSION, userId, serverVersion);
     const metadata = await this.systemMetadataRepository.get(SystemMetadataKey.VERSION_CHECK_STATE);
     if (metadata) {