refactor(server): notification events (immich-app#10754)

server/src/interfaces/event.interface.ts

@@ -4,17 +4,36 @@ import { ReleaseNotification, ServerVersionResponseDto } from 'src/dtos/server.d
 
 export const IEventRepository = 'IEventRepository';
 
+export type SystemConfigUpdateEvent = { newConfig: SystemConfig; oldConfig: SystemConfig };
+export type AlbumUpdateEvent = {
+  id: string;
+  /** user id */
+  updatedBy: string;
+};
+export type AlbumInviteEvent = { id: string; userId: string };
+export type UserSignupEvent = { notify: boolean; id: string; tempPassword?: string };
+
 type MaybePromise<T> = Promise<T> | T;
+type Handler<T = undefined> = (data: T) => MaybePromise<void>;
 
 const noop = () => {};
 const dummyHandlers = {
-  onBootstrapEvent: noop as (app: 'api' | 'microservices') => MaybePromise<void>,
+  // app events
+  onBootstrapEvent: noop as Handler<'api' | 'microservices'>,
   onShutdownEvent: noop as () => MaybePromise<void>,
-  onConfigUpdateEvent: noop as (update: SystemConfigUpdate) => MaybePromise<void>,
-  onConfigValidateEvent: noop as (update: SystemConfigUpdate) => MaybePromise<void>,
+
+  // config events
+  onConfigUpdateEvent: noop as Handler<SystemConfigUpdateEvent>,
+  onConfigValidateEvent: noop as Handler<SystemConfigUpdateEvent>,
+
+  // album events
+  onAlbumUpdateEvent: noop as Handler<AlbumUpdateEvent>,
+  onAlbumInviteEvent: noop as Handler<AlbumInviteEvent>,
+
+  // user events
+  onUserSignupEvent: noop as Handler<UserSignupEvent>,
 };
 
-export type SystemConfigUpdate = { newConfig: SystemConfig; oldConfig: SystemConfig };
 export type EventHandlers = typeof dummyHandlers;
 export type EmitEvent = keyof EventHandlers;
 export type EmitEventHandler<T extends EmitEvent> = (...args: Parameters<EventHandlers[T]>) => MaybePromise<void>;

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

@@ -5,7 +5,7 @@ import { AlbumUserRole } from 'src/entities/album-user.entity';
 import { IAlbumUserRepository } from 'src/interfaces/album-user.interface';
 import { IAlbumRepository } from 'src/interfaces/album.interface';
 import { IAssetRepository } from 'src/interfaces/asset.interface';
-import { IJobRepository, JobName } from 'src/interfaces/job.interface';
+import { IEventRepository } from 'src/interfaces/event.interface';
 import { IUserRepository } from 'src/interfaces/user.interface';
 import { AlbumService } from 'src/services/album.service';
 import { albumStub } from 'test/fixtures/album.stub';
@@ -15,7 +15,7 @@ import { IAccessRepositoryMock, newAccessRepositoryMock } from 'test/repositorie
 import { newAlbumUserRepositoryMock } from 'test/repositories/album-user.repository.mock';
 import { newAlbumRepositoryMock } from 'test/repositories/album.repository.mock';
 import { newAssetRepositoryMock } from 'test/repositories/asset.repository.mock';
-import { newJobRepositoryMock } from 'test/repositories/job.repository.mock';
+import { newEventRepositoryMock } from 'test/repositories/event.repository.mock';
 import { newUserRepositoryMock } from 'test/repositories/user.repository.mock';
 import { Mocked } from 'vitest';
 
@@ -24,19 +24,19 @@ describe(AlbumService.name, () => {
   let accessMock: IAccessRepositoryMock;
   let albumMock: Mocked<IAlbumRepository>;
   let assetMock: Mocked<IAssetRepository>;
+  let eventMock: Mocked<IEventRepository>;
   let userMock: Mocked<IUserRepository>;
   let albumUserMock: Mocked<IAlbumUserRepository>;
-  let jobMock: Mocked<IJobRepository>;
 
   beforeEach(() => {
     accessMock = newAccessRepositoryMock();
     albumMock = newAlbumRepositoryMock();
     assetMock = newAssetRepositoryMock();
+    eventMock = newEventRepositoryMock();
     userMock = newUserRepositoryMock();
     albumUserMock = newAlbumUserRepositoryMock();
-    jobMock = newJobRepositoryMock();
 
-    sut = new AlbumService(accessMock, albumMock, assetMock, userMock, albumUserMock, jobMock);
+    sut = new AlbumService(accessMock, albumMock, assetMock, eventMock, userMock, albumUserMock);
   });
 
   it('should work', () => {
@@ -381,14 +381,10 @@ describe(AlbumService.name, () => {
         userId: authStub.user2.user.id,
         albumId: albumStub.sharedWithAdmin.id,
       });
-      expect(jobMock.queue.mock.calls).toEqual([
-        [
-          {
-            name: JobName.NOTIFY_ALBUM_INVITE,
-            data: { id: albumStub.sharedWithAdmin.id, recipientId: authStub.user2.user.id },
-          },
-        ],
-      ]);
+      expect(eventMock.emit).toHaveBeenCalledWith('onAlbumInviteEvent', {
+        id: albumStub.sharedWithAdmin.id,
+        userId: userStub.user2.id,
+      });
     });
   });
 
@@ -573,14 +569,10 @@ describe(AlbumService.name, () => {
         albumThumbnailAssetId: 'asset-1',
       });
       expect(albumMock.addAssetIds).toHaveBeenCalledWith('album-123', ['asset-1', 'asset-2', 'asset-3']);
-      expect(jobMock.queue.mock.calls).toEqual([
-        [
-          {
-            name: JobName.NOTIFY_ALBUM_UPDATE,
-            data: { id: 'album-123', senderId: authStub.admin.user.id },
-          },
-        ],
-      ]);
+      expect(eventMock.emit).toHaveBeenCalledWith('onAlbumUpdateEvent', {
+        id: 'album-123',
+        updatedBy: authStub.admin.user.id,
+      });
     });
 
     it('should not set the thumbnail if the album has one already', async () => {
@@ -621,14 +613,10 @@ describe(AlbumService.name, () => {
         albumThumbnailAssetId: 'asset-1',
       });
       expect(albumMock.addAssetIds).toHaveBeenCalledWith('album-123', ['asset-1', 'asset-2', 'asset-3']);
-      expect(jobMock.queue.mock.calls).toEqual([
-        [
-          {
-            name: JobName.NOTIFY_ALBUM_UPDATE,
-            data: { id: 'album-123', senderId: authStub.user1.user.id },
-          },
-        ],
-      ]);
+      expect(eventMock.emit).toHaveBeenCalledWith('onAlbumUpdateEvent', {
+        id: 'album-123',
+        updatedBy: authStub.user1.user.id,
+      });
     });
 
     it('should not allow a shared user with viewer access to add assets', async () => {

server/src/services/album.service.ts

@@ -21,7 +21,7 @@ import { IAccessRepository } from 'src/interfaces/access.interface';
 import { IAlbumUserRepository } from 'src/interfaces/album-user.interface';
 import { AlbumAssetCount, AlbumInfoOptions, IAlbumRepository } from 'src/interfaces/album.interface';
 import { IAssetRepository } from 'src/interfaces/asset.interface';
-import { IJobRepository, JobName } from 'src/interfaces/job.interface';
+import { IEventRepository } from 'src/interfaces/event.interface';
 import { IUserRepository } from 'src/interfaces/user.interface';
 import { addAssets, removeAssets } from 'src/utils/asset.util';
 
@@ -32,9 +32,9 @@ export class AlbumService {
     @Inject(IAccessRepository) private accessRepository: IAccessRepository,
     @Inject(IAlbumRepository) private albumRepository: IAlbumRepository,
     @Inject(IAssetRepository) private assetRepository: IAssetRepository,
+    @Inject(IEventRepository) private eventRepository: IEventRepository,
     @Inject(IUserRepository) private userRepository: IUserRepository,
     @Inject(IAlbumUserRepository) private albumUserRepository: IAlbumUserRepository,
-    @Inject(IJobRepository) private jobRepository: IJobRepository,
   ) {
     this.access = AccessCore.create(accessRepository);
   }
@@ -188,12 +188,9 @@ export class AlbumService {
         updatedAt: new Date(),
         albumThumbnailAssetId: album.albumThumbnailAssetId ?? firstNewAssetId,
       });
-    }
 
-    await this.jobRepository.queue({
-      name: JobName.NOTIFY_ALBUM_UPDATE,
-      data: { id, senderId: auth.user.id },
-    });
+      await this.eventRepository.emit('onAlbumUpdateEvent', { id, updatedBy: auth.user.id });
+    }
 
     return results;
   }
@@ -240,11 +237,7 @@ export class AlbumService {
       }
 
       await this.albumUserRepository.create({ userId: userId, albumId: id, role });
-
-      await this.jobRepository.queue({
-        name: JobName.NOTIFY_ALBUM_INVITE,
-        data: { id: album.id, recipientId: user.id },
-      });
+      await this.eventRepository.emit('onAlbumInviteEvent', { id, userId });
     }
 
     return this.findOrFail(id, { withAssets: true }).then(mapAlbumWithoutAssets);

server/src/services/library.service.ts

@@ -22,7 +22,7 @@ import { LibraryEntity } from 'src/entities/library.entity';
 import { IAssetRepository, WithProperty } from 'src/interfaces/asset.interface';
 import { ICryptoRepository } from 'src/interfaces/crypto.interface';
 import { DatabaseLock, IDatabaseRepository } from 'src/interfaces/database.interface';
-import { OnEvents, SystemConfigUpdate } from 'src/interfaces/event.interface';
+import { OnEvents, SystemConfigUpdateEvent } from 'src/interfaces/event.interface';
 import {
   IBaseJob,
   IEntityJob,
@@ -102,7 +102,7 @@ export class LibraryService implements OnEvents {
     });
   }
 
-  onConfigValidateEvent({ newConfig }: SystemConfigUpdate) {
+  onConfigValidateEvent({ newConfig }: SystemConfigUpdateEvent) {
     const { scan } = newConfig.library;
     if (!validateCronExpression(scan.cronExpression)) {
       throw new Error(`Invalid cron expression ${scan.cronExpression}`);

server/src/services/notification.service.ts

@@ -5,7 +5,13 @@ import { SystemConfigSmtpDto } from 'src/dtos/system-config.dto';
 import { AlbumEntity } from 'src/entities/album.entity';
 import { IAlbumRepository } from 'src/interfaces/album.interface';
 import { IAssetRepository } from 'src/interfaces/asset.interface';
-import { OnEvents, SystemConfigUpdate } from 'src/interfaces/event.interface';
+import {
+  AlbumInviteEvent,
+  AlbumUpdateEvent,
+  OnEvents,
+  SystemConfigUpdateEvent,
+  UserSignupEvent,
+} from 'src/interfaces/event.interface';
 import {
   IEmailJob,
   IJobRepository,
@@ -38,7 +44,7 @@ export class NotificationService implements OnEvents {
     this.configCore = SystemConfigCore.create(systemMetadataRepository, logger);
   }
 
-  async onConfigValidateEvent({ newConfig }: SystemConfigUpdate) {
+  async onConfigValidateEvent({ newConfig }: SystemConfigUpdateEvent) {
     try {
       if (newConfig.notifications.smtp.enabled) {
         await this.notificationRepository.verifySmtp(newConfig.notifications.smtp.transport);
@@ -49,6 +55,20 @@ export class NotificationService implements OnEvents {
     }
   }
 
+  async onUserSignupEvent({ notify, id, tempPassword }: UserSignupEvent) {
+    if (notify) {
+      await this.jobRepository.queue({ name: JobName.NOTIFY_SIGNUP, data: { id, tempPassword } });
+    }
+  }
+
+  async onAlbumUpdateEvent({ id, updatedBy }: AlbumUpdateEvent) {
+    await this.jobRepository.queue({ name: JobName.NOTIFY_ALBUM_UPDATE, data: { id, senderId: updatedBy } });
+  }
+
+  async onAlbumInviteEvent({ id, userId }: AlbumInviteEvent) {
+    await this.jobRepository.queue({ name: JobName.NOTIFY_ALBUM_INVITE, data: { id, recipientId: userId } });
+  }
+
   async sendTestEmail(id: string, dto: SystemConfigSmtpDto) {
     const user = await this.userRepository.get(id, { withDeleted: false });
     if (!user) {

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

@@ -2,7 +2,7 @@ import { Inject, Injectable } from '@nestjs/common';
 import { SystemConfigCore } from 'src/cores/system-config.core';
 import { IAssetRepository, WithoutProperty } from 'src/interfaces/asset.interface';
 import { DatabaseLock, IDatabaseRepository } from 'src/interfaces/database.interface';
-import { OnEvents, SystemConfigUpdate } from 'src/interfaces/event.interface';
+import { OnEvents, SystemConfigUpdateEvent } from 'src/interfaces/event.interface';
 import {
   IBaseJob,
   IEntityJob,
@@ -50,7 +50,7 @@ export class SmartInfoService implements OnEvents {
     await this.jobRepository.resume(QueueName.SMART_SEARCH);
   }
 
-  async onConfigUpdateEvent({ oldConfig, newConfig }: SystemConfigUpdate) {
+  async onConfigUpdateEvent({ oldConfig, newConfig }: SystemConfigUpdateEvent) {
     if (oldConfig.machineLearning.clip.modelName !== newConfig.machineLearning.clip.modelName) {
       await this.repository.init(newConfig.machineLearning.clip.modelName);
     }

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

@@ -21,7 +21,7 @@ import { IAlbumRepository } from 'src/interfaces/album.interface';
 import { IAssetRepository } from 'src/interfaces/asset.interface';
 import { ICryptoRepository } from 'src/interfaces/crypto.interface';
 import { DatabaseLock, IDatabaseRepository } from 'src/interfaces/database.interface';
-import { OnEvents, SystemConfigUpdate } from 'src/interfaces/event.interface';
+import { OnEvents, SystemConfigUpdateEvent } from 'src/interfaces/event.interface';
 import { IEntityJob, JOBS_ASSET_PAGINATION_SIZE, JobStatus } from 'src/interfaces/job.interface';
 import { ILoggerRepository } from 'src/interfaces/logger.interface';
 import { IMoveRepository } from 'src/interfaces/move.interface';
@@ -87,7 +87,7 @@ export class StorageTemplateService implements OnEvents {
     );
   }
 
-  onConfigValidateEvent({ newConfig }: SystemConfigUpdate) {
+  onConfigValidateEvent({ newConfig }: SystemConfigUpdateEvent) {
     try {
       const { compiled } = this.compile(newConfig.storageTemplate.template);
       this.render(compiled, {

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

@@ -20,7 +20,7 @@ import {
   IEventRepository,
   OnEvents,
   ServerEvent,
-  SystemConfigUpdate,
+  SystemConfigUpdateEvent,
 } from 'src/interfaces/event.interface';
 import { ILoggerRepository } from 'src/interfaces/logger.interface';
 import { ISystemMetadataRepository } from 'src/interfaces/system-metadata.interface';
@@ -42,11 +42,7 @@ export class SystemConfigService implements OnEvents {
   @EventHandlerOptions({ priority: -100 })
   async onBootstrapEvent() {
     const config = await this.core.getConfig({ withCache: false });
-    this.config$.next(config);
-  }
-
-  get config$() {
-    return this.core.config$;
+    this.core.config$.next(config);
   }
 
   async getConfig(): Promise<SystemConfigDto> {
@@ -58,7 +54,7 @@ export class SystemConfigService implements OnEvents {
     return mapConfig(defaults);
   }
 
-  onConfigValidateEvent({ newConfig, oldConfig }: SystemConfigUpdate) {
+  onConfigValidateEvent({ newConfig, oldConfig }: SystemConfigUpdateEvent) {
     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.');
     }

server/src/services/user-admin.service.spec.ts

@@ -3,6 +3,7 @@ import { mapUserAdmin } from 'src/dtos/user.dto';
 import { UserStatus } from 'src/entities/user.entity';
 import { IAlbumRepository } from 'src/interfaces/album.interface';
 import { ICryptoRepository } from 'src/interfaces/crypto.interface';
+import { IEventRepository } from 'src/interfaces/event.interface';
 import { IJobRepository, JobName } from 'src/interfaces/job.interface';
 import { ILoggerRepository } from 'src/interfaces/logger.interface';
 import { IUserRepository } from 'src/interfaces/user.interface';
@@ -11,28 +12,30 @@ import { authStub } from 'test/fixtures/auth.stub';
 import { userStub } from 'test/fixtures/user.stub';
 import { newAlbumRepositoryMock } from 'test/repositories/album.repository.mock';
 import { newCryptoRepositoryMock } from 'test/repositories/crypto.repository.mock';
+import { newEventRepositoryMock } from 'test/repositories/event.repository.mock';
 import { newJobRepositoryMock } from 'test/repositories/job.repository.mock';
 import { newLoggerRepositoryMock } from 'test/repositories/logger.repository.mock';
 import { newUserRepositoryMock } from 'test/repositories/user.repository.mock';
 import { Mocked, describe } from 'vitest';
 
 describe(UserAdminService.name, () => {
   let sut: UserAdminService;
-  let userMock: Mocked<IUserRepository>;
-  let cryptoRepositoryMock: Mocked<ICryptoRepository>;
-
   let albumMock: Mocked<IAlbumRepository>;
+  let cryptoMock: Mocked<ICryptoRepository>;
+  let eventMock: Mocked<IEventRepository>;
   let jobMock: Mocked<IJobRepository>;
   let loggerMock: Mocked<ILoggerRepository>;
+  let userMock: Mocked<IUserRepository>;
 
   beforeEach(() => {
     albumMock = newAlbumRepositoryMock();
-    cryptoRepositoryMock = newCryptoRepositoryMock();
+    cryptoMock = newCryptoRepositoryMock();
+    eventMock = newEventRepositoryMock();
     jobMock = newJobRepositoryMock();
     userMock = newUserRepositoryMock();
     loggerMock = newLoggerRepositoryMock();
 
-    sut = new UserAdminService(albumMock, cryptoRepositoryMock, jobMock, userMock, loggerMock);
+    sut = new UserAdminService(albumMock, cryptoMock, eventMock, jobMock, userMock, loggerMock);
 
     userMock.get.mockImplementation((userId) =>
       Promise.resolve([userStub.admin, userStub.user1].find((user) => user.id === userId) ?? null),

server/src/services/user-admin.service.ts

@@ -15,6 +15,7 @@ import { UserMetadataKey } from 'src/entities/user-metadata.entity';
 import { UserStatus } from 'src/entities/user.entity';
 import { IAlbumRepository } from 'src/interfaces/album.interface';
 import { ICryptoRepository } from 'src/interfaces/crypto.interface';
+import { IEventRepository } from 'src/interfaces/event.interface';
 import { IJobRepository, JobName } from 'src/interfaces/job.interface';
 import { ILoggerRepository } from 'src/interfaces/logger.interface';
 import { IUserRepository, UserFindOptions } from 'src/interfaces/user.interface';
@@ -27,6 +28,7 @@ export class UserAdminService {
   constructor(
     @Inject(IAlbumRepository) private albumRepository: IAlbumRepository,
     @Inject(ICryptoRepository) private cryptoRepository: ICryptoRepository,
+    @Inject(IEventRepository) private eventRepository: IEventRepository,
     @Inject(IJobRepository) private jobRepository: IJobRepository,
     @Inject(IUserRepository) private userRepository: IUserRepository,
     @Inject(ILoggerRepository) private logger: ILoggerRepository,
@@ -44,10 +46,12 @@ export class UserAdminService {
     const { notify, ...rest } = dto;
     const user = await this.userCore.createUser(rest);
 
-    const tempPassword = user.shouldChangePassword ? rest.password : undefined;
-    if (notify) {
-      await this.jobRepository.queue({ name: JobName.NOTIFY_SIGNUP, data: { id: user.id, tempPassword } });
-    }
+    await this.eventRepository.emit('onUserSignupEvent', {
+      notify: !!notify,
+      id: user.id,
+      tempPassword: user.shouldChangePassword ? rest.password : undefined,
+    });
+
     return mapUserAdmin(user);
   }