refactor(server): app module (immich-app#13193)

server/src/app.module.ts

@@ -4,14 +4,14 @@ import { ConfigModule } from '@nestjs/config';
 import { APP_FILTER, APP_GUARD, APP_INTERCEPTOR, APP_PIPE, ModuleRef } from '@nestjs/core';
 import { ScheduleModule, SchedulerRegistry } from '@nestjs/schedule';
 import { TypeOrmModule } from '@nestjs/typeorm';
-import _ from 'lodash';
 import { ClsModule } from 'nestjs-cls';
 import { OpenTelemetryModule } from 'nestjs-otel';
 import { commands } from 'src/commands';
 import { bullConfig, bullQueues, clsConfig, immichAppConfig } from 'src/config';
 import { controllers } from 'src/controllers';
 import { databaseConfig } from 'src/database.config';
 import { entities } from 'src/entities';
+import { ImmichWorker } from 'src/enum';
 import { IEventRepository } from 'src/interfaces/event.interface';
 import { ILoggerRepository } from 'src/interfaces/logger.interface';
 import { AuthGuard } from 'src/middleware/auth.guard';
@@ -22,7 +22,6 @@ import { LoggingInterceptor } from 'src/middleware/logging.interceptor';
 import { repositories } from 'src/repositories';
 import { services } from 'src/services';
 import { DatabaseService } from 'src/services/database.service';
-import { setupEventHandlers } from 'src/utils/events';
 import { otelConfig } from 'src/utils/instrumentation';
 
 const common = [...services, ...repositories];
@@ -56,59 +55,48 @@ const imports = [
   TypeOrmModule.forFeature(entities),
 ];
 
-@Module({
-  imports: [...imports, ScheduleModule.forRoot()],
-  controllers: [...controllers],
-  providers: [...common, ...middleware],
-})
-export class ApiModule implements OnModuleInit, OnModuleDestroy {
+abstract class BaseModule implements OnModuleInit, OnModuleDestroy {
+  private get worker() {
+    return this.getWorker();
+  }
+
   constructor(
-    private moduleRef: ModuleRef,
+    @Inject(ILoggerRepository) logger: ILoggerRepository,
     @Inject(IEventRepository) private eventRepository: IEventRepository,
-    @Inject(ILoggerRepository) private logger: ILoggerRepository,
   ) {
-    logger.setAppName('Api');
+    logger.setAppName(this.worker);
   }
 
-  async onModuleInit() {
-    const items = setupEventHandlers(this.moduleRef);
-
-    await this.eventRepository.emit('app.bootstrap', 'api');
+  abstract getWorker(): ImmichWorker;
 
-    this.logger.setContext('EventLoader');
-    const eventMap = _.groupBy(items, 'event');
-    for (const [event, handlers] of Object.entries(eventMap)) {
-      for (const { priority, label } of handlers) {
-        this.logger.verbose(`Added ${event} {${label}${priority ? '' : ', ' + priority}} event`);
-      }
-    }
+  async onModuleInit() {
+    this.eventRepository.setup({ services });
+    await this.eventRepository.emit('app.bootstrap', this.worker);
   }
 
   async onModuleDestroy() {
-    await this.eventRepository.emit('app.shutdown');
+    await this.eventRepository.emit('app.shutdown', this.worker);
   }
 }
 
 @Module({
-  imports: [...imports],
-  providers: [...common, SchedulerRegistry],
+  imports: [...imports, ScheduleModule.forRoot()],
+  controllers: [...controllers],
+  providers: [...common, ...middleware],
 })
-export class MicroservicesModule implements OnModuleInit, OnModuleDestroy {
-  constructor(
-    private moduleRef: ModuleRef,
-    @Inject(IEventRepository) private eventRepository: IEventRepository,
-    @Inject(ILoggerRepository) logger: ILoggerRepository,
-  ) {
-    logger.setAppName('Microservices');
-  }
-
-  async onModuleInit() {
-    setupEventHandlers(this.moduleRef);
-    await this.eventRepository.emit('app.bootstrap', 'microservices');
+export class ApiModule extends BaseModule {
+  getWorker() {
+    return ImmichWorker.API;
   }
+}
 
-  async onModuleDestroy() {
-    await this.eventRepository.emit('app.shutdown');
+@Module({
+  imports: [...imports],
+  providers: [...common, SchedulerRegistry],
+})
+export class MicroservicesModule extends BaseModule {
+  getWorker() {
+    return ImmichWorker.MICROSERVICES;
   }
 }
 

server/src/interfaces/event.interface.ts

@@ -1,13 +1,15 @@
+import { ClassConstructor } from 'class-transformer';
 import { SystemConfig } from 'src/config';
 import { AssetResponseDto } from 'src/dtos/asset-response.dto';
 import { ReleaseNotification, ServerVersionResponseDto } from 'src/dtos/server.dto';
+import { ImmichWorker } from 'src/enum';
 
 export const IEventRepository = 'IEventRepository';
 
 type EventMap = {
   // app events
-  'app.bootstrap': ['api' | 'microservices'];
-  'app.shutdown': [];
+  'app.bootstrap': [ImmichWorker];
+  'app.shutdown': [ImmichWorker];
 
   // config events
   'config.update': [
@@ -85,6 +87,7 @@ export type EventItem<T extends EmitEvent> = {
 };
 
 export interface IEventRepository {
+  setup(options: { services: ClassConstructor<unknown>[] }): void;
   on<T extends keyof EventMap>(item: EventItem<T>): void;
   emit<T extends keyof EventMap>(event: T, ...args: ArgsOf<T>): Promise<void>;
 

server/src/interfaces/logger.interface.ts

@@ -1,9 +1,9 @@
-import { LogLevel } from 'src/enum';
+import { ImmichWorker, LogLevel } from 'src/enum';
 
 export const ILoggerRepository = 'ILoggerRepository';
 
 export interface ILoggerRepository {
-  setAppName(name: string): void;
+  setAppName(name: ImmichWorker): void;
   setContext(message: string): void;
   setLogLevel(level: LogLevel | false): void;
   isLevelEnabled(level: LogLevel): boolean;

server/src/repositories/event.repository.ts

@@ -1,17 +1,22 @@
 import { Inject, Injectable } from '@nestjs/common';
-import { ModuleRef } from '@nestjs/core';
+import { ModuleRef, Reflector } from '@nestjs/core';
 import {
   OnGatewayConnection,
   OnGatewayDisconnect,
   OnGatewayInit,
   WebSocketGateway,
   WebSocketServer,
 } from '@nestjs/websockets';
+import { ClassConstructor } from 'class-transformer';
+import _ from 'lodash';
 import { Server, Socket } from 'socket.io';
+import { EventConfig } from 'src/decorators';
+import { MetadataKey } from 'src/enum';
 import {
   ArgsOf,
   ClientEventMap,
   EmitEvent,
+  EmitHandler,
   EventItem,
   IEventRepository,
   serverEvents,
@@ -24,6 +29,14 @@ import { handlePromiseError } from 'src/utils/misc';
 
 type EmitHandlers = Partial<{ [T in EmitEvent]: Array<EventItem<T>> }>;
 
+type Item<T extends EmitEvent> = {
+  event: T;
+  handler: EmitHandler<T>;
+  priority: number;
+  server: boolean;
+  label: string;
+};
+
 @Instrumentation()
 @WebSocketGateway({
   cors: true,
@@ -44,6 +57,49 @@ export class EventRepository implements OnGatewayConnection, OnGatewayDisconnect
     this.logger.setContext(EventRepository.name);
   }
 
+  setup({ services }: { services: ClassConstructor<unknown>[] }) {
+    const reflector = this.moduleRef.get(Reflector, { strict: false });
+    const repository = this.moduleRef.get<IEventRepository>(IEventRepository);
+    const items: Item<EmitEvent>[] = [];
+
+    // discovery
+    for (const Service of services) {
+      const instance = this.moduleRef.get<any>(Service);
+      const ctx = Object.getPrototypeOf(instance);
+      for (const property of Object.getOwnPropertyNames(ctx)) {
+        const descriptor = Object.getOwnPropertyDescriptor(ctx, property);
+        if (!descriptor || descriptor.get || descriptor.set) {
+          continue;
+        }
+
+        const handler = instance[property];
+        if (typeof handler !== 'function') {
+          continue;
+        }
+
+        const event = reflector.get<EventConfig>(MetadataKey.EVENT_CONFIG, handler);
+        if (!event) {
+          continue;
+        }
+
+        items.push({
+          event: event.name,
+          priority: event.priority || 0,
+          server: event.server ?? false,
+          handler: handler.bind(instance),
+          label: `${Service.name}.${handler.name}`,
+        });
+      }
+    }
+
+    const handlers = _.orderBy(items, ['priority'], ['asc']);
+
+    // register by priority
+    for (const handler of handlers) {
+      repository.on(handler);
+    }
+  }
+
   afterInit(server: Server) {
     this.logger.log('Initialized websocket server');
 

server/src/repositories/logger.repository.spec.ts

@@ -1,4 +1,5 @@
 import { ClsService } from 'nestjs-cls';
+import { ImmichWorker } from 'src/enum';
 import { IConfigRepository } from 'src/interfaces/config.interface';
 import { LoggerRepository } from 'src/repositories/logger.repository';
 import { mockEnvData, newConfigRepositoryMock } from 'test/repositories/config.repository.mock';
@@ -22,18 +23,18 @@ describe(LoggerRepository.name, () => {
       configMock.getEnv.mockReturnValue(mockEnvData({ noColor: false }));
 
       sut = new LoggerRepository(clsMock, configMock);
-      sut.setAppName('api');
+      sut.setAppName(ImmichWorker.API);
 
-      expect(sut['formatContext']('context')).toBe('\u001B[33m[api:context]\u001B[39m ');
+      expect(sut['formatContext']('context')).toBe('\u001B[33m[Api:context]\u001B[39m ');
     });
 
     it('should not use colors when noColor is true', () => {
       configMock.getEnv.mockReturnValue(mockEnvData({ noColor: true }));
 
       sut = new LoggerRepository(clsMock, configMock);
-      sut.setAppName('api');
+      sut.setAppName(ImmichWorker.API);
 
-      expect(sut['formatContext']('context')).toBe('[api:context] ');
+      expect(sut['formatContext']('context')).toBe('[Api:context] ');
     });
   });
 });

server/src/repositories/logger.repository.ts

@@ -34,7 +34,7 @@ export class LoggerRepository extends ConsoleLogger implements ILoggerRepository
   private static appName?: string = undefined;
 
   setAppName(name: string): void {
-    LoggerRepository.appName = name;
+    LoggerRepository.appName = name.charAt(0).toUpperCase() + name.slice(1);
   }
 
   isLevelEnabled(level: LogLevel) {

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

@@ -1,5 +1,6 @@
 import { BadRequestException } from '@nestjs/common';
 import { defaults } from 'src/config';
+import { ImmichWorker } from 'src/enum';
 import { IAssetRepository } from 'src/interfaces/asset.interface';
 import {
   IJobRepository,
@@ -40,7 +41,7 @@ describe(JobService.name, () => {
 
   describe('onConfigUpdate', () => {
     it('should update concurrency', () => {
-      sut.onBootstrap('microservices');
+      sut.onBootstrap(ImmichWorker.MICROSERVICES);
       sut.onConfigUpdate({ oldConfig: defaults, newConfig: defaults });
 
       expect(jobMock.setConcurrency).toHaveBeenCalledTimes(14);

server/src/services/job.service.ts

@@ -3,7 +3,7 @@ import { snakeCase } from 'lodash';
 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 { AssetType, ImmichWorker, ManualJobName } from 'src/enum';
 import { ArgOf } from 'src/interfaces/event.interface';
 import {
   ConcurrentQueueName,
@@ -43,7 +43,7 @@ export class JobService extends BaseService {
 
   @OnEvent({ name: 'app.bootstrap' })
   onBootstrap(app: ArgOf<'app.bootstrap'>) {
-    this.isMicroservices = app === 'microservices';
+    this.isMicroservices = app === ImmichWorker.MICROSERVICES;
   }
 
   @OnEvent({ name: 'config.update', server: true })

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

@@ -3,7 +3,7 @@ import { randomBytes } from 'node:crypto';
 import { Stats } from 'node:fs';
 import { constants } from 'node:fs/promises';
 import { ExifEntity } from 'src/entities/exif.entity';
-import { AssetType, SourceType } from 'src/enum';
+import { AssetType, ImmichWorker, SourceType } from 'src/enum';
 import { IAlbumRepository } from 'src/interfaces/album.interface';
 import { IAssetRepository, WithoutProperty } from 'src/interfaces/asset.interface';
 import { ICryptoRepository } from 'src/interfaces/crypto.interface';
@@ -73,7 +73,7 @@ describe(MetadataService.name, () => {
 
   describe('onBootstrapEvent', () => {
     it('should pause and resume queue during init', async () => {
-      await sut.onBootstrap('microservices');
+      await sut.onBootstrap(ImmichWorker.MICROSERVICES);
 
       expect(jobMock.pause).toHaveBeenCalledTimes(1);
       expect(mapMock.init).toHaveBeenCalledTimes(1);
@@ -83,7 +83,7 @@ describe(MetadataService.name, () => {
     it('should return if reverse geocoding is disabled', async () => {
       systemMock.get.mockResolvedValue({ reverseGeocoding: { enabled: false } });
 
-      await sut.onBootstrap('microservices');
+      await sut.onBootstrap(ImmichWorker.MICROSERVICES);
 
       expect(jobMock.pause).not.toHaveBeenCalled();
       expect(mapMock.init).not.toHaveBeenCalled();

server/src/services/metadata.service.ts

@@ -12,7 +12,7 @@ import { AssetFaceEntity } from 'src/entities/asset-face.entity';
 import { AssetEntity } from 'src/entities/asset.entity';
 import { ExifEntity } from 'src/entities/exif.entity';
 import { PersonEntity } from 'src/entities/person.entity';
-import { AssetType, SourceType } from 'src/enum';
+import { AssetType, ImmichWorker, SourceType } from 'src/enum';
 import { WithoutProperty } from 'src/interfaces/asset.interface';
 import { DatabaseLock } from 'src/interfaces/database.interface';
 import { ArgOf } from 'src/interfaces/event.interface';
@@ -89,7 +89,7 @@ const validateRange = (value: number | undefined, min: number, max: number): Non
 export class MetadataService extends BaseService {
   @OnEvent({ name: 'app.bootstrap' })
   async onBootstrap(app: ArgOf<'app.bootstrap'>) {
-    if (app !== 'microservices') {
+    if (app !== ImmichWorker.MICROSERVICES) {
       return;
     }
     const config = await this.getConfig({ withCache: false });

server/src/services/microservices.service.ts

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