refactor: repositories (#16038)

server/src/app.module.ts

@@ -13,7 +13,6 @@ import { IWorker } from 'src/constants';
 import { controllers } from 'src/controllers';
 import { entities } from 'src/entities';
 import { ImmichWorker } from 'src/enum';
-import { IEventRepository } from 'src/interfaces/event.interface';
 import { IJobRepository } from 'src/interfaces/job.interface';
 import { AuthGuard } from 'src/middleware/auth.guard';
 import { ErrorInterceptor } from 'src/middleware/error.interceptor';
@@ -22,9 +21,11 @@ import { GlobalExceptionFilter } from 'src/middleware/global-exception.filter';
 import { LoggingInterceptor } from 'src/middleware/logging.interceptor';
 import { providers, repositories } from 'src/repositories';
 import { ConfigRepository } from 'src/repositories/config.repository';
+import { EventRepository } from 'src/repositories/event.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
 import { teardownTelemetry, TelemetryRepository } from 'src/repositories/telemetry.repository';
 import { services } from 'src/services';
+import { AuthService } from 'src/services/auth.service';
 import { CliService } from 'src/services/cli.service';
 import { DatabaseService } from 'src/services/database.service';
 
@@ -78,9 +79,10 @@ class BaseModule implements OnModuleInit, OnModuleDestroy {
   constructor(
     @Inject(IWorker) private worker: ImmichWorker,
     logger: LoggingRepository,
-    @Inject(IEventRepository) private eventRepository: IEventRepository,
+    private eventRepository: EventRepository,
     @Inject(IJobRepository) private jobRepository: IJobRepository,
     private telemetryRepository: TelemetryRepository,
+    private authService: AuthService,
   ) {
     logger.setAppName(this.worker);
   }
@@ -93,6 +95,14 @@ class BaseModule implements OnModuleInit, OnModuleDestroy {
       this.jobRepository.startWorkers();
     }
 
+    this.eventRepository.setAuthFn(async (client) =>
+      this.authService.authenticate({
+        headers: client.request.headers,
+        queryParams: {},
+        metadata: { adminRoute: false, sharedLinkRoute: false, uri: '/api/socket.io' },
+      }),
+    );
+
     this.eventRepository.setup({ services });
     await this.eventRepository.emit('app.bootstrap');
   }

server/src/decorators.ts

@@ -3,8 +3,8 @@ import { ApiExtension, ApiOperation, ApiProperty, ApiTags } from '@nestjs/swagge
 import _ from 'lodash';
 import { ADDED_IN_PREFIX, DEPRECATED_IN_PREFIX, LIFECYCLE_EXTENSION } from 'src/constants';
 import { ImmichWorker, MetadataKey } from 'src/enum';
-import { EmitEvent } from 'src/interfaces/event.interface';
 import { JobName, QueueName } from 'src/interfaces/job.interface';
+import { EmitEvent } from 'src/repositories/event.repository';
 import { setUnion } from 'src/utils/set';
 
 // PostgreSQL uses a 16-bit integer to indicate the number of bound parameters. This means that the

server/src/enum.ts

@@ -391,3 +391,10 @@ export enum DatabaseExtension {
   VECTOR = 'vector',
   VECTORS = 'vectors',
 }
+
+export enum BootstrapEventPriority {
+  // Database service should be initialized before anything else, most other services need database access
+  DatabaseService = -200,
+  // Initialise config after other bootstrap services, stop other services from using config on bootstrap
+  SystemConfig = 100,
+}

server/src/repositories/event.repository.ts

@@ -10,21 +10,15 @@ import {
 import { ClassConstructor } from 'class-transformer';
 import _ from 'lodash';
 import { Server, Socket } from 'socket.io';
+import { SystemConfig } from 'src/config';
 import { EventConfig } from 'src/decorators';
+import { AssetResponseDto } from 'src/dtos/asset-response.dto';
+import { AuthDto } from 'src/dtos/auth.dto';
+import { ReleaseNotification, ServerVersionResponseDto } from 'src/dtos/server.dto';
 import { ImmichWorker, MetadataKey } from 'src/enum';
-import {
-  ArgsOf,
-  ClientEventMap,
-  EmitEvent,
-  EmitHandler,
-  EventItem,
-  IEventRepository,
-  serverEvents,
-  ServerEvents,
-} from 'src/interfaces/event.interface';
+import { JobItem, QueueName } from 'src/interfaces/job.interface';
 import { ConfigRepository } from 'src/repositories/config.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
-import { AuthService } from 'src/services/auth.service';
 import { handlePromiseError } from 'src/utils/misc';
 
 type EmitHandlers = Partial<{ [T in EmitEvent]: Array<EventItem<T>> }>;
@@ -37,14 +31,99 @@ type Item<T extends EmitEvent> = {
   label: string;
 };
 
+type EventMap = {
+  // app events
+  'app.bootstrap': [];
+  'app.shutdown': [];
+
+  'config.init': [{ newConfig: SystemConfig }];
+  // config events
+  'config.update': [
+    {
+      newConfig: SystemConfig;
+      oldConfig: SystemConfig;
+    },
+  ];
+  'config.validate': [{ newConfig: SystemConfig; oldConfig: SystemConfig }];
+
+  // album events
+  'album.update': [{ id: string; recipientIds: string[] }];
+  'album.invite': [{ id: string; userId: string }];
+
+  // asset events
+  'asset.tag': [{ assetId: string }];
+  'asset.untag': [{ assetId: string }];
+  'asset.hide': [{ assetId: string; userId: string }];
+  'asset.show': [{ assetId: string; userId: string }];
+  'asset.trash': [{ assetId: string; userId: string }];
+  'asset.delete': [{ assetId: string; userId: string }];
+
+  // asset bulk events
+  'assets.trash': [{ assetIds: string[]; userId: string }];
+  'assets.delete': [{ assetIds: string[]; userId: string }];
+  'assets.restore': [{ assetIds: string[]; userId: string }];
+
+  'job.start': [QueueName, JobItem];
+
+  // session events
+  'session.delete': [{ sessionId: string }];
+
+  // stack events
+  'stack.create': [{ stackId: string; userId: string }];
+  'stack.update': [{ stackId: string; userId: string }];
+  'stack.delete': [{ stackId: string; userId: string }];
+
+  // stack bulk events
+  'stacks.delete': [{ stackIds: string[]; userId: string }];
+
+  // user events
+  'user.signup': [{ notify: boolean; id: string; tempPassword?: string }];
+
+  // websocket events
+  'websocket.connect': [{ userId: string }];
+};
+
+export const serverEvents = ['config.update'] as const;
+export type ServerEvents = (typeof serverEvents)[number];
+
+export type EmitEvent = keyof EventMap;
+export type EmitHandler<T extends EmitEvent> = (...args: ArgsOf<T>) => Promise<void> | void;
+export type ArgOf<T extends EmitEvent> = EventMap[T][0];
+export type ArgsOf<T extends EmitEvent> = EventMap[T];
+
+export interface ClientEventMap {
+  on_upload_success: [AssetResponseDto];
+  on_user_delete: [string];
+  on_asset_delete: [string];
+  on_asset_trash: [string[]];
+  on_asset_update: [AssetResponseDto];
+  on_asset_hidden: [string];
+  on_asset_restore: [string[]];
+  on_asset_stack_update: string[];
+  on_person_thumbnail: [string];
+  on_server_version: [ServerVersionResponseDto];
+  on_config_update: [];
+  on_new_release: [ReleaseNotification];
+  on_session_delete: [string];
+}
+
+export type EventItem<T extends EmitEvent> = {
+  event: T;
+  handler: EmitHandler<T>;
+  server: boolean;
+};
+
+export type AuthFn = (client: Socket) => Promise<AuthDto>;
+
 @WebSocketGateway({
   cors: true,
   path: '/api/socket.io',
   transports: ['websocket'],
 })
 @Injectable()
-export class EventRepository implements OnGatewayConnection, OnGatewayDisconnect, OnGatewayInit, IEventRepository {
+export class EventRepository implements OnGatewayConnection, OnGatewayDisconnect, OnGatewayInit {
   private emitHandlers: EmitHandlers = {};
+  private authFn?: AuthFn;
 
   @WebSocketServer()
   private server?: Server;
@@ -122,11 +201,7 @@ export class EventRepository implements OnGatewayConnection, OnGatewayDisconnect
   async handleConnection(client: Socket) {
     try {
       this.logger.log(`Websocket Connect:    ${client.id}`);
-      const auth = await this.moduleRef.get(AuthService).authenticate({
-        headers: client.request.headers,
-        queryParams: {},
-        metadata: { adminRoute: false, sharedLinkRoute: false, uri: '/api/socket.io' },
-      });
+      const auth = await this.authenticate(client);
       await client.join(auth.user.id);
       if (auth.session) {
         await client.join(auth.session.id);
@@ -182,4 +257,16 @@ export class EventRepository implements OnGatewayConnection, OnGatewayDisconnect
     this.logger.debug(`Server event: ${event} (send)`);
     this.server?.serverSideEmit(event, ...args);
   }
+
+  setAuthFn(fn: (client: Socket) => Promise<AuthDto>) {
+    this.authFn = fn;
+  }
+
+  private async authenticate(client: Socket) {
+    if (!this.authFn) {
+      throw new Error('Auth function not set');
+    }
+
+    return this.authFn(client);
+  }
 }