refactor(server): app module (#13193)

server/src/app.module.ts

@@ -4,7 +4,6 @@ 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';
@@ -12,6 +11,7 @@ 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,36 +55,38 @@ const imports = [
   TypeOrmModule.forFeature(entities),
 ];
 
+abstract class BaseModule implements OnModuleInit, OnModuleDestroy {
+  private get worker() {
+    return this.getWorker();
+  }
+
+  constructor(
+    @Inject(ILoggerRepository) logger: ILoggerRepository,
+    @Inject(IEventRepository) private eventRepository: IEventRepository,
+  ) {
+    logger.setAppName(this.worker);
+  }
+
+  abstract getWorker(): ImmichWorker;
+
+  async onModuleInit() {
+    this.eventRepository.setup({ services });
+    await this.eventRepository.emit('app.bootstrap', this.worker);
+  }
+
+  async onModuleDestroy() {
+    await this.eventRepository.emit('app.shutdown', this.worker);
+  }
+}
+
 @Module({
   imports: [...imports, ScheduleModule.forRoot()],
   controllers: [...controllers],
   providers: [...common, ...middleware],
 })
-export class ApiModule implements OnModuleInit, OnModuleDestroy {
+export class ApiModule extends BaseModule {
-  constructor(
+  getWorker() {
-    private moduleRef: ModuleRef,
+    return ImmichWorker.API;
-    @Inject(IEventRepository) private eventRepository: IEventRepository,
-    @Inject(ILoggerRepository) private logger: ILoggerRepository,
-  ) {
-    logger.setAppName('Api');
-  }
-
-  async onModuleInit() {
-    const items = setupEventHandlers(this.moduleRef);
-
-    await this.eventRepository.emit('app.bootstrap', 'api');
-
-    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 onModuleDestroy() {
-    await this.eventRepository.emit('app.shutdown');
   }
 }
 
@@ -93,22 +94,9 @@ export class ApiModule implements OnModuleInit, OnModuleDestroy {
   imports: [...imports],
   providers: [...common, SchedulerRegistry],
 })
-export class MicroservicesModule implements OnModuleInit, OnModuleDestroy {
+export class MicroservicesModule extends BaseModule {
-  constructor(
+  getWorker() {
-    private moduleRef: ModuleRef,
+    return ImmichWorker.MICROSERVICES;
-    @Inject(IEventRepository) private eventRepository: IEventRepository,
-    @Inject(ILoggerRepository) logger: ILoggerRepository,
-  ) {
-    logger.setAppName('Microservices');
-  }
-
-  async onModuleInit() {
-    setupEventHandlers(this.moduleRef);
-    await this.eventRepository.emit('app.bootstrap', 'microservices');
-  }
-
-  async onModuleDestroy() {
-    await this.eventRepository.emit('app.shutdown');
   }
 }
 

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.bootstrap': [ImmichWorker];
-  'app.shutdown': [];
+  '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,5 +1,5 @@
 import { Inject, Injectable } from '@nestjs/common';
-import { ModuleRef } from '@nestjs/core';
+import { ModuleRef, Reflector } from '@nestjs/core';
 import {
   OnGatewayConnection,
   OnGatewayDisconnect,
@@ -7,11 +7,16 @@ import {
   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;
     }
 

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

@@ -1,4 +1,5 @@
 import { SystemConfig } from 'src/config';
+import { ImmichWorker } from 'src/enum';
 import { IAssetRepository, WithoutProperty } from 'src/interfaces/asset.interface';
 import { IJobRepository, JobName, JobStatus } from 'src/interfaces/job.interface';
 import { IMachineLearningRepository } from 'src/interfaces/machine-learning.interface';
@@ -61,7 +62,7 @@ describe(SmartInfoService.name, () => {
 
   describe('onBootstrapEvent', () => {
     it('should return if not microservices', async () => {
-      await sut.onBootstrap('api');
+      await sut.onBootstrap(ImmichWorker.API);
 
       expect(systemMock.get).not.toHaveBeenCalled();
       expect(searchMock.getDimensionSize).not.toHaveBeenCalled();
@@ -76,7 +77,7 @@ describe(SmartInfoService.name, () => {
     it('should return if machine learning is disabled', async () => {
       systemMock.get.mockResolvedValue(systemConfigStub.machineLearningDisabled);
 
-      await sut.onBootstrap('microservices');
+      await sut.onBootstrap(ImmichWorker.MICROSERVICES);
 
       expect(systemMock.get).toHaveBeenCalledTimes(1);
       expect(searchMock.getDimensionSize).not.toHaveBeenCalled();
@@ -91,7 +92,7 @@ describe(SmartInfoService.name, () => {
     it('should return if model and DB dimension size are equal', async () => {
       searchMock.getDimensionSize.mockResolvedValue(512);
 
-      await sut.onBootstrap('microservices');
+      await sut.onBootstrap(ImmichWorker.MICROSERVICES);
 
       expect(systemMock.get).toHaveBeenCalledTimes(1);
       expect(searchMock.getDimensionSize).toHaveBeenCalledTimes(1);
@@ -107,7 +108,7 @@ describe(SmartInfoService.name, () => {
       searchMock.getDimensionSize.mockResolvedValue(768);
       jobMock.getQueueStatus.mockResolvedValue({ isActive: false, isPaused: false });
 
-      await sut.onBootstrap('microservices');
+      await sut.onBootstrap(ImmichWorker.MICROSERVICES);
 
       expect(systemMock.get).toHaveBeenCalledTimes(1);
       expect(searchMock.getDimensionSize).toHaveBeenCalledTimes(1);
@@ -122,7 +123,7 @@ describe(SmartInfoService.name, () => {
       searchMock.getDimensionSize.mockResolvedValue(768);
       jobMock.getQueueStatus.mockResolvedValue({ isActive: false, isPaused: true });
 
-      await sut.onBootstrap('microservices');
+      await sut.onBootstrap(ImmichWorker.MICROSERVICES);
 
       expect(systemMock.get).toHaveBeenCalledTimes(1);
       expect(searchMock.getDimensionSize).toHaveBeenCalledTimes(1);

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

@@ -1,6 +1,7 @@
 import { Injectable } from '@nestjs/common';
 import { SystemConfig } from 'src/config';
 import { OnEvent } from 'src/decorators';
+import { ImmichWorker } from 'src/enum';
 import { WithoutProperty } from 'src/interfaces/asset.interface';
 import { DatabaseLock } from 'src/interfaces/database.interface';
 import { ArgOf } from 'src/interfaces/event.interface';
@@ -21,7 +22,7 @@ import { usePagination } from 'src/utils/pagination';
 export class SmartInfoService extends BaseService {
   @OnEvent({ name: 'app.bootstrap' })
   async onBootstrap(app: ArgOf<'app.bootstrap'>) {
-    if (app !== 'microservices') {
+    if (app !== ImmichWorker.MICROSERVICES) {
       return;
     }
 

server/src/services/storage.service.ts

@@ -6,7 +6,9 @@ import { StorageFolder, SystemMetadataKey } from 'src/enum';
 import { DatabaseLock } from 'src/interfaces/database.interface';
 import { IDeleteFilesJob, JobStatus } from 'src/interfaces/job.interface';
 import { BaseService } from 'src/services/base.service';
-import { ImmichStartupError } from 'src/utils/events';
+
+export class ImmichStartupError extends Error {}
+export const isStartUpError = (error: unknown): error is ImmichStartupError => error instanceof ImmichStartupError;
 
 const docsMessage = `Please see https://immich.app/docs/administration/system-integrity#folder-checks for more information.`;
 

server/src/utils/events.ts

@@ -1,62 +0,0 @@
-import { ModuleRef, Reflector } from '@nestjs/core';
-import _ from 'lodash';
-import { EventConfig } from 'src/decorators';
-import { MetadataKey } from 'src/enum';
-import { EmitEvent, EmitHandler, IEventRepository } from 'src/interfaces/event.interface';
-import { services } from 'src/services';
-
-type Item<T extends EmitEvent> = {
-  event: T;
-  handler: EmitHandler<T>;
-  priority: number;
-  server: boolean;
-  label: string;
-};
-
-export class ImmichStartupError extends Error {}
-export const isStartUpError = (error: unknown): error is ImmichStartupError => error instanceof ImmichStartupError;
-
-export const setupEventHandlers = (moduleRef: ModuleRef) => {
-  const reflector = moduleRef.get(Reflector, { strict: false });
-  const repository = moduleRef.get<IEventRepository>(IEventRepository);
-  const items: Item<EmitEvent>[] = [];
-
-  // discovery
-  for (const Service of services) {
-    const instance = 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);
-  }
-
-  return handlers;
-};

server/src/workers/api.ts

@@ -11,7 +11,7 @@ import { IConfigRepository } from 'src/interfaces/config.interface';
 import { ILoggerRepository } from 'src/interfaces/logger.interface';
 import { WebSocketAdapter } from 'src/middleware/websocket.adapter';
 import { ApiService } from 'src/services/api.service';
-import { isStartUpError } from 'src/utils/events';
+import { isStartUpError } from 'src/services/storage.service';
 import { otelStart } from 'src/utils/instrumentation';
 import { useSwagger } from 'src/utils/misc';
 

server/src/workers/microservices.ts

@@ -5,7 +5,7 @@ import { serverVersion } from 'src/constants';
 import { IConfigRepository } from 'src/interfaces/config.interface';
 import { ILoggerRepository } from 'src/interfaces/logger.interface';
 import { WebSocketAdapter } from 'src/middleware/websocket.adapter';
-import { isStartUpError } from 'src/utils/events';
+import { isStartUpError } from 'src/services/storage.service';
 import { otelStart } from 'src/utils/instrumentation';
 
 export async function bootstrap() {
@@ -15,7 +15,6 @@ export async function bootstrap() {
 
   const app = await NestFactory.create(MicroservicesModule, { bufferLogs: true });
   const logger = await app.resolve(ILoggerRepository);
-  logger.setAppName('Microservices');
   logger.setContext('Bootstrap');
   app.useLogger(logger);
   app.useWebSocketAdapter(new WebSocketAdapter(app));

server/test/repositories/event.repository.mock.ts

@@ -3,6 +3,7 @@ import { Mocked, vitest } from 'vitest';
 
 export const newEventRepositoryMock = (): Mocked<IEventRepository> => {
   return {
+    setup: vitest.fn(),
     on: vitest.fn() as any,
     emit: vitest.fn() as any,
     clientSend: vitest.fn() as any,