refactor: asset upload

.github/pull_request_template.md

@@ -26,7 +26,6 @@ The `/api/something` endpoint is now `/api/something-else`
 
 ## Checklist:
 
-- [ ] I have carefully read CONTRIBUTING.md
 - [ ] I have performed a self-review of my own code
 - [ ] I have made corresponding changes to the documentation if applicable
 - [ ] I have no unrelated changes in the PR.

.github/workflows/close-llm-pr.yml

@@ -1,38 +0,0 @@
-name: Close LLM-generated PRs
-
-on:
-  pull_request:
-    types: [labeled]
-
-permissions: {}
-
-jobs:
-  comment_and_close:
-    runs-on: ubuntu-latest
-    if: ${{ github.event.label.name == 'llm-generated' }}
-    permissions:
-      pull-requests: write
-    steps:
-      - name: Comment and close
-        env:
-          GH_TOKEN: ${{ github.token }}
-          NODE_ID: ${{ github.event.pull_request.node_id }}
-        run: |
-          gh api graphql \
-            -f prId="$NODE_ID" \
-            -f body="Thank you for your interest in contributing to Immich! Unfortunately this PR looks like it was generated using an LLM. As noted in our CONTRIBUTING.md, we request that you don't use LLMs to generate PRs as those are not a good use of maintainer time." \
-            -f query='
-              mutation CommentAndClosePR($prId: ID!, $body: String!) {
-                addComment(input: {
-                  subjectId: $prId,
-                  body: $body
-                }) {
-                  __typename
-                }
-
-                closePullRequest(input: {
-                  pullRequestId: $prId
-                }) {
-                  __typename
-                }
-              }'

CONTRIBUTING.md

@@ -17,7 +17,7 @@ If you are looking for something to work on, there are discussions and issues wi
 
 ## Use of generative AI
 
-We ask you not to open PRs generated with an LLM. We find that code generated like this tends to need a large amount of back-and-forth, which is a very inefficient use of our time. If we want LLM-generated code, it's much faster for us to use an LLM ourselves than to go through an intermediary via a pull request.
+We generally discourage PRs entirely generated by an LLM. For any part generated by an LLM, please put extra effort into your self-review. By using generative AI without proper self-review, the time you save ends up being more work we need to put in for proper reviews and code cleanup. Please keep that in mind when submitting code by an LLM. Clearly state the use of LLMs/(generative) AI in your pull request as requested by the template.
 
 ## Feature freezes
 

e2e/docker-compose.yml

@@ -10,7 +10,6 @@ services:
   immich-server:
     container_name: immich-e2e-server
     image: immich-server:latest
-    shm_size: 128mb
     build:
       context: ../
       dockerfile: server/Dockerfile
@@ -54,7 +53,6 @@ services:
       POSTGRES_DB: immich
     ports:
       - 5435:5432
-    shm_size: 128mb
     healthcheck:
       test: ['CMD-SHELL', 'pg_isready -U postgres -d immich']
       interval: 1s

pnpm-lock.yaml

@@ -406,12 +406,9 @@ importers:
       '@react-email/render':
         specifier: ^1.1.2
         version: 1.4.0(react-dom@19.2.4(react@19.2.4))(react@19.2.4)
-      '@socket.io/postgres-adapter':
-        specifier: ^0.5.0
-        version: 0.5.0(socket.io-adapter@2.5.6)
-      '@types/pg':
-        specifier: ^8.16.0
-        version: 8.16.0
+      '@socket.io/redis-adapter':
+        specifier: ^8.3.0
+        version: 8.3.0(socket.io-adapter@2.5.6)
       ajv:
         specifier: ^8.17.1
         version: 8.17.1
@@ -565,9 +562,6 @@ importers:
       socket.io:
         specifier: ^4.8.1
         version: 4.8.3
-      socket.io-adapter:
-        specifier: ^2.5.6
-        version: 2.5.6
       tailwindcss-preset-email:
         specifier: ^1.4.0
         version: 1.4.1(tailwindcss@3.4.19(tsx@4.21.0)(yaml@2.8.2))
@@ -3408,10 +3402,6 @@ packages:
   '@microsoft/tsdoc@0.16.0':
     resolution: {integrity: sha512-xgAyonlVVS+q7Vc7qLW0UrJU7rSFcETRWsqdXZtjzRU8dF+6CkozTK4V4y1LwOX7j8r/vHphjDeMeGI4tNGeGA==}
 
-  '@msgpack/msgpack@2.8.0':
-    resolution: {integrity: sha512-h9u4u/jiIRKbq25PM+zymTyW6bhTzELvOoUd+AvYriWOAKpLGnIamaET3pnHYoI5iYphAHBI4ayx0MehR+VVPQ==}
-    engines: {node: '>= 10'}
-
   '@msgpackr-extract/msgpackr-extract-darwin-arm64@3.0.3':
     resolution: {integrity: sha512-QZHtlVgbAdy2zAqNA9Gu1UpIuI8Xvsd1v8ic6B2pZmeFnFcMWiPLfWXh7TVw4eGEZ/C9TH281KwhVoeQUKbyjw==}
     cpu: [arm64]
@@ -4320,9 +4310,9 @@ packages:
   '@socket.io/component-emitter@3.1.2':
     resolution: {integrity: sha512-9BCxFwvbGg/RsZK9tjXd8s4UcwR0MWeFQ1XEKIQVVvAGJyINdrqKMcTRyLoK8Rse1GjzLV9cwjWV1olXRWEXVA==}
 
-  '@socket.io/postgres-adapter@0.5.0':
-    resolution: {integrity: sha512-s1vFsatB4lS429ZbeAi8ju+mZMgtgdSmi9UsZsdcEG++vVtX5z10yDEt4TV8saePscvvGjs6uXvJfMCxz8+M2Q==}
-    engines: {node: '>=12.0.0'}
+  '@socket.io/redis-adapter@8.3.0':
+    resolution: {integrity: sha512-ly0cra+48hDmChxmIpnESKrc94LjRL80TEmZVscuQ/WWkRP81nNj8W8cCGMqbI4L6NCuAaPRSzZF1a9GlAxxnA==}
+    engines: {node: '>=10.0.0'}
     peerDependencies:
       socket.io-adapter: ^2.5.4
 
@@ -9295,6 +9285,9 @@ packages:
   not@0.1.0:
     resolution: {integrity: sha512-5PDmaAsVfnWUgTUbJ3ERwn7u79Z0dYxN9ErxCpVJJqe2RK0PJ3z+iFUxuqjwtlDDegXvtWoxD/3Fzxox7tFGWA==}
 
+  notepack.io@3.0.1:
+    resolution: {integrity: sha512-TKC/8zH5pXIAMVQio2TvVDTtPRX+DJPHDqjRbxogtFiByHyzKmy96RA0JtCQJ+WouyyL4A10xomQzgbUT+1jCg==}
+
   npm-run-path@4.0.1:
     resolution: {integrity: sha512-S48WzZW777zhNIrn7gxOlISNAqi9ZC/uQFnRdbeIHhZhCA6UqpkOT8T1G7BvfdgP4Er8gF4sUbaS0i7QvIfCWw==}
     engines: {node: '>=8'}
@@ -11626,6 +11619,10 @@ packages:
     engines: {node: '>=0.8.0'}
     hasBin: true
 
+  uid2@1.0.0:
+    resolution: {integrity: sha512-+I6aJUv63YAcY9n4mQreLUt0d4lvwkkopDNmpomkAUz0fAkEMV9pRWxN0EjhW1YfRhcuyHg2v3mwddCDW1+LFQ==}
+    engines: {node: '>= 4.0.0'}
+
   uid@2.0.2:
     resolution: {integrity: sha512-u3xV3X7uzvi5b1MncmZo3i2Aw222Zk1keqLA1YkHldREkAhAqi65wuPfe7lHx8H/Wzy+8CE7S7uS3jekIM5s8g==}
     engines: {node: '>=8'}
@@ -15429,8 +15426,6 @@ snapshots:
 
   '@microsoft/tsdoc@0.16.0': {}
 
-  '@msgpack/msgpack@2.8.0': {}
-
   '@msgpackr-extract/msgpackr-extract-darwin-arm64@3.0.3':
     optional: true
 
@@ -16308,15 +16303,13 @@ snapshots:
 
   '@socket.io/component-emitter@3.1.2': {}
 
-  '@socket.io/postgres-adapter@0.5.0(socket.io-adapter@2.5.6)':
+  '@socket.io/redis-adapter@8.3.0(socket.io-adapter@2.5.6)':
     dependencies:
-      '@msgpack/msgpack': 2.8.0
-      '@types/pg': 8.16.0
       debug: 4.3.7
-      pg: 8.18.0
+      notepack.io: 3.0.1
       socket.io-adapter: 2.5.6
+      uid2: 1.0.0
     transitivePeerDependencies:
-      - pg-native
       - supports-color
 
   '@sphinxxxx/color-conversion@2.2.2': {}
@@ -22287,6 +22280,8 @@ snapshots:
 
   not@0.1.0: {}
 
+  notepack.io@3.0.1: {}
+
   npm-run-path@4.0.1:
     dependencies:
       path-key: 3.1.1
@@ -25035,6 +25030,8 @@ snapshots:
   uglify-js@3.19.3:
     optional: true
 
+  uid2@1.0.0: {}
+
   uid@2.0.2:
     dependencies:
       '@lukeed/csprng': 1.1.0

server/package.json

@@ -56,8 +56,7 @@
     "@opentelemetry/semantic-conventions": "^1.34.0",
     "@react-email/components": "^0.5.0",
     "@react-email/render": "^1.1.2",
-    "@socket.io/postgres-adapter": "^0.5.0",
-    "@types/pg": "^8.16.0",
+    "@socket.io/redis-adapter": "^8.3.0",
     "ajv": "^8.17.1",
     "archiver": "^7.0.0",
     "async-lock": "^1.4.0",
@@ -109,7 +108,6 @@
     "sharp": "^0.34.5",
     "sirv": "^3.0.0",
     "socket.io": "^4.8.1",
-    "socket.io-adapter": "^2.5.6",
     "tailwindcss-preset-email": "^1.4.0",
     "thumbhash": "^0.1.1",
     "transformation-matrix": "^3.1.0",

server/src/app.common.ts

@@ -5,9 +5,8 @@ import cookieParser from 'cookie-parser';
 import { existsSync } from 'node:fs';
 import sirv from 'sirv';
 import { excludePaths, serverVersion } from 'src/constants';
-import { SocketIoAdapter } from 'src/enum';
 import { MaintenanceWorkerService } from 'src/maintenance/maintenance-worker.service';
-import { createWebSocketAdapter } from 'src/middleware/websocket.adapter';
+import { WebSocketAdapter } from 'src/middleware/websocket.adapter';
 import { ConfigRepository } from 'src/repositories/config.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
 import { bootstrapTelemetry } from 'src/repositories/telemetry.repository';
@@ -26,7 +25,6 @@ export async function configureExpress(
   {
     permitSwaggerWrite = true,
     ssr,
-    socketIoAdapter,
   }: {
     /**
      * Whether to allow swagger module to write to the specs.json
@@ -38,10 +36,6 @@ export async function configureExpress(
      * Service to use for server-side rendering
      */
     ssr: typeof ApiService | typeof MaintenanceWorkerService;
-    /**
-     * Override the Socket.IO adapter. If not specified, uses the adapter from config.
-     */
-    socketIoAdapter?: SocketIoAdapter;
   },
 ) {
   const configRepository = app.get(ConfigRepository);
@@ -61,7 +55,7 @@ export async function configureExpress(
   }
 
   app.setGlobalPrefix('api', { exclude: excludePaths });
-  app.useWebSocketAdapter(await createWebSocketAdapter(app, socketIoAdapter));
+  app.useWebSocketAdapter(new WebSocketAdapter(app));
 
   useSwagger(app, { write: configRepository.isDev() && permitSwaggerWrite });
 

server/src/app.module.ts

@@ -14,11 +14,12 @@ import { MaintenanceHealthRepository } from 'src/maintenance/maintenance-health.
 import { MaintenanceWebsocketRepository } from 'src/maintenance/maintenance-websocket.repository';
 import { MaintenanceWorkerController } from 'src/maintenance/maintenance-worker.controller';
 import { MaintenanceWorkerService } from 'src/maintenance/maintenance-worker.service';
+import { AssetUploadInterceptor } from 'src/middleware/asset-upload.interceptor';
 import { AuthGuard } from 'src/middleware/auth.guard';
 import { ErrorInterceptor } from 'src/middleware/error.interceptor';
-import { FileUploadInterceptor } from 'src/middleware/file-upload.interceptor';
 import { GlobalExceptionFilter } from 'src/middleware/global-exception.filter';
 import { LoggingInterceptor } from 'src/middleware/logging.interceptor';
+import { UserProfileUploadInterceptor } from 'src/middleware/user-profile-upload.interceptor';
 import { repositories } from 'src/repositories';
 import { AppRepository } from 'src/repositories/app.repository';
 import { ConfigRepository } from 'src/repositories/config.repository';
@@ -46,7 +47,12 @@ const commonMiddleware = [
   { provide: APP_INTERCEPTOR, useClass: ErrorInterceptor },
 ];
 
-const apiMiddleware = [FileUploadInterceptor, ...commonMiddleware, { provide: APP_GUARD, useClass: AuthGuard }];
+const apiMiddleware = [
+  AssetUploadInterceptor,
+  UserProfileUploadInterceptor,
+  ...commonMiddleware,
+  { provide: APP_GUARD, useClass: AuthGuard },
+];
 
 const configRepository = new ConfigRepository();
 const { bull, cls, database, otel } = configRepository.getEnv();

server/src/controllers/asset-media.controller.ts

@@ -1,6 +1,7 @@
 import {
   Body,
   Controller,
+  UploadedFiles as Files,
   Get,
   HttpCode,
   HttpStatus,
@@ -12,7 +13,6 @@ import {
   Query,
   Req,
   Res,
-  UploadedFiles,
   UseInterceptors,
 } from '@nestjs/common';
 import { ApiBody, ApiConsumes, ApiHeader, ApiResponse, ApiTags } from '@nestjs/swagger';
@@ -35,18 +35,17 @@ import {
 } from 'src/dtos/asset-media.dto';
 import { AssetDownloadOriginalDto } from 'src/dtos/asset.dto';
 import { AuthDto } from 'src/dtos/auth.dto';
-import { ApiTag, ImmichHeader, Permission, RouteKey } from 'src/enum';
+import { ApiTag, ImmichHeader, Permission } from 'src/enum';
 import { AssetUploadInterceptor } from 'src/middleware/asset-upload.interceptor';
 import { Auth, Authenticated, FileResponse } from 'src/middleware/auth.guard';
-import { FileUploadInterceptor, getFiles } from 'src/middleware/file-upload.interceptor';
+import { mapUploadedFile, UploadFiles, UploadRequest } from 'src/middleware/upload.interceptor';
 import { LoggingRepository } from 'src/repositories/logging.repository';
 import { AssetMediaService } from 'src/services/asset-media.service';
-import { UploadFiles } from 'src/types';
 import { ImmichFileResponse, sendFile } from 'src/utils/file';
 import { FileNotEmptyValidator, UUIDParamDto } from 'src/validation';
 
 @ApiTags(ApiTag.Assets)
-@Controller(RouteKey.Asset)
+@Controller('assets')
 export class AssetMediaController {
   constructor(
     private logger: LoggingRepository,
@@ -55,7 +54,7 @@ export class AssetMediaController {
 
   @Post()
   @Authenticated({ permission: Permission.AssetUpload, sharedLink: true })
-  @UseInterceptors(AssetUploadInterceptor, FileUploadInterceptor)
+  @UseInterceptors(AssetUploadInterceptor)
   @ApiConsumes('multipart/form-data')
   @ApiHeader({
     name: ImmichHeader.Checksum,
@@ -80,12 +79,21 @@ export class AssetMediaController {
   })
   async uploadAsset(
     @Auth() auth: AuthDto,
-    @UploadedFiles(new ParseFilePipe({ validators: [new FileNotEmptyValidator(['assetData'])] })) files: UploadFiles,
+    @Files(new ParseFilePipe({ validators: [new FileNotEmptyValidator([UploadFieldName.ASSET_DATA])] }))
+    files: UploadFiles,
     @Body() dto: AssetMediaCreateDto,
+    @Req() req: UploadRequest,
     @Res({ passthrough: true }) res: Response,
   ): Promise<AssetMediaResponseDto> {
-    const { file, sidecarFile } = getFiles(files);
-    const responseDto = await this.service.uploadAsset(auth, dto, file, sidecarFile);
+    const file = files[UploadFieldName.ASSET_DATA][0];
+    const sidecarFile = files[UploadFieldName.SIDECAR_DATA]?.[0];
+
+    const responseDto = await this.service.uploadAsset(
+      auth,
+      dto,
+      mapUploadedFile(req, file),
+      sidecarFile ? mapUploadedFile(req, sidecarFile) : undefined,
+    );
 
     if (responseDto.status === AssetMediaStatus.DUPLICATE) {
       res.status(HttpStatus.OK);
@@ -113,7 +121,7 @@ export class AssetMediaController {
   }
 
   @Put(':id/original')
-  @UseInterceptors(FileUploadInterceptor)
+  @UseInterceptors(AssetUploadInterceptor)
   @ApiConsumes('multipart/form-data')
   @ApiResponse({
     status: 200,
@@ -129,13 +137,19 @@ export class AssetMediaController {
   async replaceAsset(
     @Auth() auth: AuthDto,
     @Param() { id }: UUIDParamDto,
-    @UploadedFiles(new ParseFilePipe({ validators: [new FileNotEmptyValidator([UploadFieldName.ASSET_DATA])] }))
+
+    @Files(new ParseFilePipe({ validators: [new FileNotEmptyValidator([UploadFieldName.ASSET_DATA])] }))
     files: UploadFiles,
     @Body() dto: AssetMediaReplaceDto,
+    @Req() req: UploadRequest,
     @Res({ passthrough: true }) res: Response,
   ): Promise<AssetMediaResponseDto> {
-    const { file } = getFiles(files);
-    const responseDto = await this.service.replaceAsset(auth, id, dto, file);
+    const responseDto = await this.service.replaceAsset(
+      auth,
+      id,
+      dto,
+      mapUploadedFile(req, files[UploadFieldName.ASSET_DATA][0]),
+    );
     if (responseDto.status === AssetMediaStatus.DUPLICATE) {
       res.status(HttpStatus.OK);
     }

server/src/controllers/asset.controller.ts

@@ -22,13 +22,13 @@ import {
 import { AuthDto } from 'src/dtos/auth.dto';
 import { AssetEditActionListDto, AssetEditsDto } from 'src/dtos/editing.dto';
 import { AssetOcrResponseDto } from 'src/dtos/ocr.dto';
-import { ApiTag, Permission, RouteKey } from 'src/enum';
+import { ApiTag, Permission } from 'src/enum';
 import { Auth, Authenticated } from 'src/middleware/auth.guard';
 import { AssetService } from 'src/services/asset.service';
 import { UUIDParamDto } from 'src/validation';
 
 @ApiTags(ApiTag.Assets)
-@Controller(RouteKey.Asset)
+@Controller('assets')
 export class AssetController {
   constructor(private service: AssetService) {}
 

server/src/controllers/database-backup.controller.ts

@@ -1,4 +1,15 @@
-import { Body, Controller, Delete, Get, Next, Param, Post, Res, UploadedFile, UseInterceptors } from '@nestjs/common';
+import {
+  Body,
+  Controller,
+  Delete,
+  UploadedFile as File,
+  Get,
+  Next,
+  Param,
+  Post,
+  Res,
+  UseInterceptors,
+} from '@nestjs/common';
 import { FileInterceptor } from '@nestjs/platform-express';
 import { ApiBody, ApiConsumes, ApiTags } from '@nestjs/swagger';
 import { NextFunction, Response } from 'express';
@@ -92,10 +103,7 @@ export class DatabaseBackupController {
     history: new HistoryBuilder().added('v2.5.0').alpha('v2.5.0'),
   })
   @UseInterceptors(FileInterceptor('file'))
-  uploadDatabaseBackup(
-    @UploadedFile()
-    file: Express.Multer.File,
-  ): Promise<void> {
+  uploadDatabaseBackup(@File() file: Express.Multer.File): Promise<void> {
     return this.service.uploadBackup(file);
   }
 }

server/src/controllers/index.ts

@@ -10,7 +10,6 @@ import { DatabaseBackupController } from 'src/controllers/database-backup.contro
 import { DownloadController } from 'src/controllers/download.controller';
 import { DuplicateController } from 'src/controllers/duplicate.controller';
 import { FaceController } from 'src/controllers/face.controller';
-import { InternalController } from 'src/controllers/internal.controller';
 import { JobController } from 'src/controllers/job.controller';
 import { LibraryController } from 'src/controllers/library.controller';
 import { MaintenanceController } from 'src/controllers/maintenance.controller';
@@ -52,7 +51,6 @@ export const controllers = [
   DownloadController,
   DuplicateController,
   FaceController,
-  InternalController,
   JobController,
   LibraryController,
   MaintenanceController,

server/src/controllers/internal.controller.ts

@@ -1,22 +0,0 @@
-import { Body, Controller, NotFoundException, Post, Req } from '@nestjs/common';
-import { ApiExcludeController } from '@nestjs/swagger';
-import { Request } from 'express';
-import { AppRestartEvent, EventRepository } from 'src/repositories/event.repository';
-
-const LOCALHOST_ADDRESSES = new Set(['127.0.0.1', '::1', '::ffff:127.0.0.1']);
-
-@ApiExcludeController()
-@Controller('internal')
-export class InternalController {
-  constructor(private eventRepository: EventRepository) {}
-
-  @Post('restart')
-  async restart(@Req() req: Request, @Body() dto: AppRestartEvent): Promise<void> {
-    const remoteAddress = req.socket.remoteAddress;
-    if (!remoteAddress || !LOCALHOST_ADDRESSES.has(remoteAddress)) {
-      throw new NotFoundException();
-    }
-
-    await this.eventRepository.emit('AppRestart', dto);
-  }
-}

server/src/controllers/user.controller.ts

@@ -2,6 +2,7 @@ import {
   Body,
   Controller,
   Delete,
+  UploadedFile as File,
   Get,
   HttpCode,
   HttpStatus,
@@ -10,7 +11,6 @@ import {
   Post,
   Put,
   Res,
-  UploadedFile,
   UseInterceptors,
 } from '@nestjs/common';
 import { ApiBody, ApiConsumes, ApiTags } from '@nestjs/swagger';
@@ -22,16 +22,17 @@ import { OnboardingDto, OnboardingResponseDto } from 'src/dtos/onboarding.dto';
 import { UserPreferencesResponseDto, UserPreferencesUpdateDto } from 'src/dtos/user-preferences.dto';
 import { CreateProfileImageDto, CreateProfileImageResponseDto } from 'src/dtos/user-profile.dto';
 import { UserAdminResponseDto, UserResponseDto, UserUpdateMeDto } from 'src/dtos/user.dto';
-import { ApiTag, Permission, RouteKey } from 'src/enum';
+import { ApiTag, Permission } from 'src/enum';
 import { Auth, Authenticated, FileResponse } from 'src/middleware/auth.guard';
-import { FileUploadInterceptor } from 'src/middleware/file-upload.interceptor';
+import { UploadedFile } from 'src/middleware/upload.interceptor';
+import { UserProfileUploadInterceptor } from 'src/middleware/user-profile-upload.interceptor';
 import { LoggingRepository } from 'src/repositories/logging.repository';
 import { UserService } from 'src/services/user.service';
 import { sendFile } from 'src/utils/file';
 import { UUIDParamDto } from 'src/validation';
 
 @ApiTags(ApiTag.Users)
-@Controller(RouteKey.User)
+@Controller('users')
 export class UserController {
   constructor(
     private service: UserService,
@@ -177,7 +178,7 @@ export class UserController {
 
   @Post('profile-image')
   @Authenticated({ permission: Permission.UserProfileImageUpdate })
-  @UseInterceptors(FileUploadInterceptor)
+  @UseInterceptors(UserProfileUploadInterceptor)
   @ApiConsumes('multipart/form-data')
   @ApiBody({ description: 'A new avatar for the user', type: CreateProfileImageDto })
   @Endpoint({
@@ -185,11 +186,8 @@ export class UserController {
     description: 'Upload and set a new profile image for the current user.',
     history: new HistoryBuilder().added('v1').beta('v1').stable('v2'),
   })
-  createProfileImage(
-    @Auth() auth: AuthDto,
-    @UploadedFile() fileInfo: Express.Multer.File,
-  ): Promise<CreateProfileImageResponseDto> {
-    return this.service.createProfileImage(auth, fileInfo);
+  createProfileImage(@Auth() auth: AuthDto, @File() file: UploadedFile): Promise<CreateProfileImageResponseDto> {
+    return this.service.createProfileImage(auth, file);
   }
 
   @Delete('profile-image')

server/src/dtos/env.dto.ts

@@ -1,6 +1,6 @@
 import { Transform, Type } from 'class-transformer';
 import { IsEnum, IsInt, IsString, Matches } from 'class-validator';
-import { DatabaseSslMode, ImmichEnvironment, LogFormat, LogLevel, SocketIoAdapter } from 'src/enum';
+import { DatabaseSslMode, ImmichEnvironment, LogFormat, LogLevel } from 'src/enum';
 import { IsIPRange, Optional, ValidateBoolean } from 'src/validation';
 
 export class EnvDto {
@@ -140,11 +140,6 @@ export class EnvDto {
   @Optional()
   IMMICH_WORKERS_EXCLUDE?: string;
 
-  @IsEnum(SocketIoAdapter)
-  @Optional()
-  @Transform(({ value }) => (value ? String(value).toLowerCase().trim() : value))
-  IMMICH_SOCKETIO_ADAPTER?: SocketIoAdapter;
-
   @IsString()
   @Optional()
   DB_DATABASE_NAME?: string;

server/src/enum.ts

@@ -487,11 +487,6 @@ export enum MetadataKey {
   TelemetryEnabled = 'telemetry_enabled',
 }
 
-export enum RouteKey {
-  Asset = 'assets',
-  User = 'users',
-}
-
 export enum CacheControl {
   PrivateWithCache = 'private_with_cache',
   PrivateWithoutCache = 'private_without_cache',
@@ -518,11 +513,6 @@ export enum ImmichTelemetry {
   Job = 'job',
 }
 
-export enum SocketIoAdapter {
-  BroadcastChannel = 'broadcastchannel',
-  Postgres = 'postgres',
-}
-
 export enum ExifOrientation {
   Horizontal = 1,
   MirrorHorizontal = 2,

server/src/main.ts

@@ -1,5 +1,6 @@
 import { Kysely, sql } from 'kysely';
 import { CommandFactory } from 'nest-commander';
+import { ChildProcess, fork } from 'node:child_process';
 import { dirname, join } from 'node:path';
 import { Worker } from 'node:worker_threads';
 import { PostgresError } from 'postgres';
@@ -17,7 +18,7 @@ class Workers {
   /**
    * Currently running workers
    */
-  workers: Partial<Record<ImmichWorker, { kill: () => Promise<void> | void }>> = {};
+  workers: Partial<Record<ImmichWorker, { kill: (signal: NodeJS.Signals) => Promise<void> | void }>> = {};
 
   /**
    * Fail-safe in case anything dies during restart
@@ -100,23 +101,25 @@ class Workers {
     const basePath = dirname(__filename);
     const workerFile = join(basePath, 'workers', `${name}.js`);
 
-    const inspectArg = process.execArgv.find((arg) => arg.startsWith('--inspect'));
-    const workerData: { inspectorPort?: number } = {};
+    let anyWorker: Worker | ChildProcess;
+    let kill: (signal?: NodeJS.Signals) => Promise<void> | void;
 
-    if (inspectArg) {
-      const inspectorPorts: Record<ImmichWorker, number> = {
-        [ImmichWorker.Api]: 9230,
-        [ImmichWorker.Microservices]: 9231,
-        [ImmichWorker.Maintenance]: 9232,
-      };
-      workerData.inspectorPort = inspectorPorts[name];
+    if (name === ImmichWorker.Api) {
+      const worker = fork(workerFile, [], {
+        execArgv: process.execArgv.map((arg) => (arg.startsWith('--inspect') ? '--inspect=0.0.0.0:9231' : arg)),
+      });
+
+      kill = (signal) => void worker.kill(signal);
+      anyWorker = worker;
+    } else {
+      const worker = new Worker(workerFile);
+
+      kill = async () => void (await worker.terminate());
+      anyWorker = worker;
     }
 
-    const worker = new Worker(workerFile, { workerData });
-    const kill = async () => void (await worker.terminate());
-
-    worker.on('error', (error) => this.onError(name, error));
-    worker.on('exit', (exitCode) => this.onExit(name, exitCode));
+    anyWorker.on('error', (error) => this.onError(name, error));
+    anyWorker.on('exit', (exitCode) => this.onExit(name, exitCode));
 
     this.workers[name] = { kill };
   }
@@ -149,8 +152,8 @@ class Workers {
       console.error(`${name} worker exited with code ${exitCode}`);
 
       if (this.workers[ImmichWorker.Api] && name !== ImmichWorker.Api) {
-        console.error('Terminating api worker');
-        void this.workers[ImmichWorker.Api].kill();
+        console.error('Killing api process');
+        void this.workers[ImmichWorker.Api].kill('SIGTERM');
       }
     }
 

server/src/maintenance/maintenance-worker.controller.ts

@@ -2,14 +2,13 @@ import {
   Body,
   Controller,
   Delete,
+  UploadedFile as File,
   Get,
   Next,
-  NotFoundException,
   Param,
   Post,
   Req,
   Res,
-  UploadedFile,
   UseInterceptors,
 } from '@nestjs/common';
 import { FileInterceptor } from '@nestjs/platform-express';
@@ -26,15 +25,12 @@ import { ImmichCookie } from 'src/enum';
 import { MaintenanceRoute } from 'src/maintenance/maintenance-auth.guard';
 import { MaintenanceWorkerService } from 'src/maintenance/maintenance-worker.service';
 import { GetLoginDetails } from 'src/middleware/auth.guard';
-import { AppRestartEvent } from 'src/repositories/event.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
 import { LoginDetails } from 'src/services/auth.service';
 import { sendFile } from 'src/utils/file';
 import { respondWithCookie } from 'src/utils/response';
 import { FilenameParamDto } from 'src/validation';
 
-const LOCALHOST_ADDRESSES = new Set(['127.0.0.1', '::1', '::ffff:127.0.0.1']);
-
 import type { DatabaseBackupController as _DatabaseBackupController } from 'src/controllers/database-backup.controller';
 import type { ServerController as _ServerController } from 'src/controllers/server.controller';
 import { DatabaseBackupDeleteDto, DatabaseBackupListResponseDto } from 'src/dtos/database-backup.dto';
@@ -98,10 +94,7 @@ export class MaintenanceWorkerController {
   @Post('admin/database-backups/upload')
   @MaintenanceRoute()
   @UseInterceptors(FileInterceptor('file'))
-  uploadDatabaseBackup(
-    @UploadedFile()
-    file: Express.Multer.File,
-  ): Promise<void> {
+  uploadDatabaseBackup(@File() file: Express.Multer.File): Promise<void> {
     return this.databaseBackupService.uploadBackup(file);
   }
 
@@ -135,14 +128,4 @@ export class MaintenanceWorkerController {
   setMaintenanceMode(@Body() dto: SetMaintenanceModeDto): void {
     void this.service.setAction(dto);
   }
-
-  @Post('internal/restart')
-  internalRestart(@Req() req: Request, @Body() dto: AppRestartEvent): void {
-    const remoteAddress = req.socket.remoteAddress;
-    if (!remoteAddress || !LOCALHOST_ADDRESSES.has(remoteAddress)) {
-      throw new NotFoundException();
-    }
-
-    this.service.handleInternalRestart(dto);
-  }
 }

server/src/maintenance/maintenance-worker.service.ts

@@ -19,7 +19,6 @@ import { MaintenanceWebsocketRepository } from 'src/maintenance/maintenance-webs
 import { AppRepository } from 'src/repositories/app.repository';
 import { ConfigRepository } from 'src/repositories/config.repository';
 import { DatabaseRepository } from 'src/repositories/database.repository';
-import { AppRestartEvent } from 'src/repositories/event.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
 import { ProcessRepository } from 'src/repositories/process.repository';
 import { StorageRepository } from 'src/repositories/storage.repository';
@@ -291,9 +290,6 @@ export class MaintenanceWorkerService {
 
     const lock = await this.databaseRepository.tryLock(DatabaseLock.MaintenanceOperation);
     if (!lock) {
-      // Another maintenance worker has the lock - poll until maintenance mode ends
-      this.logger.log('Another worker has the maintenance lock, polling for maintenance mode changes...');
-      await this.pollForMaintenanceEnd();
       return;
     }
 
@@ -355,25 +351,4 @@ export class MaintenanceWorkerService {
     this.maintenanceWebsocketRepository.serverSend('AppRestart', state);
     this.appRepository.exitApp();
   }
-
-  handleInternalRestart(state: AppRestartEvent): void {
-    this.maintenanceWebsocketRepository.clientBroadcast('AppRestartV1', state);
-    this.maintenanceWebsocketRepository.serverSend('AppRestart', state);
-    this.appRepository.exitApp();
-  }
-
-  private async pollForMaintenanceEnd(): Promise<void> {
-    const pollIntervalMs = 5000;
-
-    while (true) {
-      await new Promise((resolve) => setTimeout(resolve, pollIntervalMs));
-
-      const state = await this.systemMetadataRepository.get(SystemMetadataKey.MaintenanceMode);
-      if (!state?.isMaintenanceMode) {
-        this.logger.log('Maintenance mode ended, restarting...');
-        this.appRepository.exitApp();
-        return;
-      }
-    }
-  }
 }

server/src/middleware/asset-upload.interceptor.ts

@@ -1,27 +1,32 @@
-import { CallHandler, ExecutionContext, Injectable, NestInterceptor } from '@nestjs/common';
-import { Response } from 'express';
+import { Injectable } from '@nestjs/common';
+import multer from 'multer';
 import { of } from 'rxjs';
-import { AssetMediaResponseDto, AssetMediaStatus } from 'src/dtos/asset-media-response.dto';
+import { UploadFieldName } from 'src/dtos/asset-media.dto';
 import { ImmichHeader } from 'src/enum';
-import { AuthenticatedRequest } from 'src/middleware/auth.guard';
+import { UploadInterceptor } from 'src/middleware/upload.interceptor';
 import { AssetMediaService } from 'src/services/asset-media.service';
 import { fromMaybeArray } from 'src/utils/request';
 
 @Injectable()
-export class AssetUploadInterceptor implements NestInterceptor {
-  constructor(private service: AssetMediaService) {}
-
-  async intercept(context: ExecutionContext, next: CallHandler<any>) {
-    const req = context.switchToHttp().getRequest<AuthenticatedRequest>();
-    const res = context.switchToHttp().getResponse<Response<AssetMediaResponseDto>>();
-
-    const checksum = fromMaybeArray(req.headers[ImmichHeader.Checksum]);
-    const response = await this.service.getUploadAssetIdByChecksum(req.user, checksum);
-    if (response) {
-      res.status(200);
-      return of({ status: AssetMediaStatus.DUPLICATE, id: response.id });
-    }
-
-    return next.handle();
+export class AssetUploadInterceptor extends UploadInterceptor {
+  constructor(service: AssetMediaService) {
+    super({
+      onRequest: async (req, res) => {
+        const checksum = fromMaybeArray(req.headers[ImmichHeader.Checksum]);
+        const response = await service.onBeforeUpload(req.user, checksum);
+        if (response) {
+          res.status(200);
+          return of(response);
+        }
+      },
+      configure: (instance: multer.Multer) =>
+        instance.fields([
+          { name: UploadFieldName.ASSET_DATA, maxCount: 1 },
+          { name: UploadFieldName.SIDECAR_DATA, maxCount: 1 },
+        ]),
+      canUpload: (req, file) => service.canUpload(req.user, file),
+      upload: (req, file) => service.onUpload(req.user, file),
+      remove: (req, file) => service.onUploadRemove(req.user, file),
+    });
   }
 }

server/src/middleware/broadcast-channel.adapter.ts

@@ -1,80 +0,0 @@
-import {
-  ClusterAdapterWithHeartbeat,
-  type ClusterAdapterOptions,
-  type ClusterMessage,
-  type ClusterResponse,
-  type ServerId,
-} from 'socket.io-adapter';
-
-const BC_CHANNEL_NAME = 'immich:socketio';
-
-interface BroadcastChannelPayload {
-  type: 'message' | 'response';
-  sourceUid: string;
-  targetUid?: string;
-  data: unknown;
-}
-
-/**
- * Socket.IO adapter using Node.js BroadcastChannel
- *
- * Relays messages between worker_threads within a single OS process.
- * Zero external dependencies. Does NOT work across containers — use
- * the Postgres adapter for multi-replica deployments.
- */
-class BroadcastChannelAdapter extends ClusterAdapterWithHeartbeat {
-  private readonly channel: BroadcastChannel;
-
-  constructor(nsp: any, opts?: Partial<ClusterAdapterOptions>) {
-    super(nsp, opts ?? {});
-
-    this.channel = new BroadcastChannel(BC_CHANNEL_NAME);
-    this.channel.addEventListener('message', (event: MessageEvent<BroadcastChannelPayload>) => {
-      const msg = event.data;
-      if (msg.sourceUid === this.uid) {
-        return;
-      }
-      if (msg.type === 'message') {
-        this.onMessage(msg.data as ClusterMessage);
-      } else if (msg.type === 'response' && msg.targetUid === this.uid) {
-        this.onResponse(msg.data as ClusterResponse);
-      }
-    });
-
-    this.init();
-  }
-
-  override doPublish(message: ClusterMessage): Promise<string> {
-    this.channel.postMessage({
-      type: 'message',
-      sourceUid: this.uid,
-      data: message,
-    });
-    return Promise.resolve('');
-  }
-
-  override doPublishResponse(requesterUid: ServerId, response: ClusterResponse): Promise<void> {
-    this.channel.postMessage({
-      type: 'response',
-      sourceUid: this.uid,
-      targetUid: requesterUid,
-      data: response,
-    });
-    return Promise.resolve();
-  }
-
-  override close(): void {
-    super.close();
-    this.channel.close();
-  }
-}
-
-export function createBroadcastChannelAdapter(opts?: Partial<ClusterAdapterOptions>) {
-  const options: Partial<ClusterAdapterOptions> = {
-    ...opts,
-  };
-
-  return function (nsp: any) {
-    return new BroadcastChannelAdapter(nsp, options);
-  };
-}

server/src/middleware/file-upload.interceptor.ts

@@ -1,173 +0,0 @@
-import { CallHandler, ExecutionContext, Injectable, NestInterceptor } from '@nestjs/common';
-import { PATH_METADATA } from '@nestjs/common/constants';
-import { Reflector } from '@nestjs/core';
-import { transformException } from '@nestjs/platform-express/multer/multer/multer.utils';
-import { NextFunction, RequestHandler } from 'express';
-import multer, { StorageEngine, diskStorage } from 'multer';
-import { createHash, randomUUID } from 'node:crypto';
-import { Observable } from 'rxjs';
-import { UploadFieldName } from 'src/dtos/asset-media.dto';
-import { RouteKey } from 'src/enum';
-import { AuthRequest } from 'src/middleware/auth.guard';
-import { LoggingRepository } from 'src/repositories/logging.repository';
-import { AssetMediaService } from 'src/services/asset-media.service';
-import { ImmichFile, UploadFile, UploadFiles } from 'src/types';
-import { asUploadRequest, mapToUploadFile } from 'src/utils/asset.util';
-
-export function getFile(files: UploadFiles, property: 'assetData' | 'sidecarData') {
-  const file = files[property]?.[0];
-  return file ? mapToUploadFile(file) : file;
-}
-
-export function getFiles(files: UploadFiles) {
-  return {
-    file: getFile(files, 'assetData') as UploadFile,
-    sidecarFile: getFile(files, 'sidecarData'),
-  };
-}
-
-type DiskStorageCallback = (error: Error | null, result: string) => void;
-
-type ImmichMulterFile = Express.Multer.File & { uuid: string };
-
-interface Callback<T> {
-  (error: Error): void;
-  (error: null, result: T): void;
-}
-
-const callbackify = <T>(target: (...arguments_: any[]) => T, callback: Callback<T>) => {
-  try {
-    return callback(null, target());
-  } catch (error: Error | any) {
-    return callback(error);
-  }
-};
-
-@Injectable()
-export class FileUploadInterceptor implements NestInterceptor {
-  private handlers: {
-    userProfile: RequestHandler;
-    assetUpload: RequestHandler;
-  };
-  private defaultStorage: StorageEngine;
-
-  constructor(
-    private reflect: Reflector,
-    private assetService: AssetMediaService,
-    private logger: LoggingRepository,
-  ) {
-    this.logger.setContext(FileUploadInterceptor.name);
-
-    this.defaultStorage = diskStorage({
-      filename: this.filename.bind(this),
-      destination: this.destination.bind(this),
-    });
-
-    const instance = multer({
-      fileFilter: this.fileFilter.bind(this),
-      storage: {
-        _handleFile: this.handleFile.bind(this),
-        _removeFile: this.removeFile.bind(this),
-      },
-    });
-
-    this.handlers = {
-      userProfile: instance.single(UploadFieldName.PROFILE_DATA),
-      assetUpload: instance.fields([
-        { name: UploadFieldName.ASSET_DATA, maxCount: 1 },
-        { name: UploadFieldName.SIDECAR_DATA, maxCount: 1 },
-      ]),
-    };
-  }
-
-  async intercept(context: ExecutionContext, next: CallHandler<any>): Promise<Observable<any>> {
-    const context_ = context.switchToHttp();
-    const route = this.reflect.get<string>(PATH_METADATA, context.getClass());
-
-    const handler: RequestHandler | null = this.getHandler(route as RouteKey);
-    if (handler) {
-      await new Promise<void>((resolve, reject) => {
-        const next: NextFunction = (error) => (error ? reject(transformException(error)) : resolve());
-        const maybePromise = handler(context_.getRequest(), context_.getResponse(), next);
-        Promise.resolve(maybePromise).catch((error) => reject(error));
-      });
-    } else {
-      this.logger.warn(`Skipping invalid file upload route: ${route}`);
-    }
-
-    return next.handle();
-  }
-
-  private fileFilter(request: AuthRequest, file: Express.Multer.File, callback: multer.FileFilterCallback) {
-    return callbackify(() => this.assetService.canUploadFile(asUploadRequest(request, file)), callback);
-  }
-
-  private filename(request: AuthRequest, file: Express.Multer.File, callback: DiskStorageCallback) {
-    return callbackify(
-      () => this.assetService.getUploadFilename(asUploadRequest(request, file)),
-      callback as Callback<string>,
-    );
-  }
-
-  private destination(request: AuthRequest, file: Express.Multer.File, callback: DiskStorageCallback) {
-    return callbackify(
-      () => this.assetService.getUploadFolder(asUploadRequest(request, file)),
-      callback as Callback<string>,
-    );
-  }
-
-  private handleFile(request: AuthRequest, file: Express.Multer.File, callback: Callback<Partial<ImmichFile>>) {
-    (file as ImmichMulterFile).uuid = randomUUID();
-
-    request.on('error', (error) => {
-      this.logger.warn('Request error while uploading file, cleaning up', error);
-      this.assetService.onUploadError(request, file).catch(this.logger.error);
-    });
-
-    if (!this.isAssetUploadFile(file)) {
-      this.defaultStorage._handleFile(request, file, callback);
-      return;
-    }
-
-    const hash = createHash('sha1');
-    file.stream.on('data', (chunk) => hash.update(chunk));
-    this.defaultStorage._handleFile(request, file, (error, info) => {
-      if (error) {
-        hash.destroy();
-        callback(error);
-      } else {
-        callback(null, { ...info, checksum: hash.digest() });
-      }
-    });
-  }
-
-  private removeFile(request: AuthRequest, file: Express.Multer.File, callback: (error: Error | null) => void) {
-    this.defaultStorage._removeFile(request, file, callback);
-  }
-
-  private isAssetUploadFile(file: Express.Multer.File) {
-    switch (file.fieldname as UploadFieldName) {
-      case UploadFieldName.ASSET_DATA: {
-        return true;
-      }
-    }
-
-    return false;
-  }
-
-  private getHandler(route: RouteKey) {
-    switch (route) {
-      case RouteKey.Asset: {
-        return this.handlers.assetUpload;
-      }
-
-      case RouteKey.User: {
-        return this.handlers.userProfile;
-      }
-
-      default: {
-        return null;
-      }
-    }
-  }
-}

server/src/middleware/upload.interceptor.ts

@@ -0,0 +1,131 @@
+import { CallHandler, ExecutionContext, NestInterceptor, UnauthorizedException } from '@nestjs/common';
+import { transformException } from '@nestjs/platform-express/multer/multer/multer.utils';
+import { NextFunction, RequestHandler, Response } from 'express';
+import multer from 'multer';
+import { Readable } from 'node:stream';
+import { Observable } from 'rxjs';
+import { AuthenticatedRequest } from 'src/middleware/auth.guard';
+import { v4 } from 'uuid';
+
+type Callback<T> = {
+  (error: Error): void;
+  (error: null, result: T): void;
+};
+
+export type UploadFile = {
+  requestId: string;
+  fieldName: string;
+  originalName: string;
+};
+
+export type UploadingFile = UploadFile & {
+  stream: Readable;
+};
+
+export type UploadedFile = UploadFile & { metadata: UploadMetadata };
+
+export type UploadMetadata = {
+  /** folder */
+  folder: string;
+  /** k filename */
+  filename: string;
+  /** full path */
+  path: string;
+  size: number;
+  checksum?: Buffer;
+};
+
+export type UploadFiles = {
+  assetData: Express.Multer.File[];
+  sidecarData: Express.Multer.File[];
+};
+
+export type UploadRequest = AuthenticatedRequest & {
+  requestId: string;
+};
+
+type OnRequest = (req: UploadRequest, res: Response) => Promise<Observable<any> | void>;
+
+const mapUploadFile = (req: UploadRequest, file: Express.Multer.File): UploadFile => {
+  const originalName = req.body?.filename || Buffer.from(file.originalname, 'latin1').toString('utf8');
+  return {
+    requestId: req.requestId,
+    fieldName: file.fieldname,
+    originalName,
+  };
+};
+
+export const mapUploadedFile = (req: UploadRequest, file: Express.Multer.File): UploadedFile => {
+  return { ...mapUploadFile(req, file), metadata: (file as unknown as UploadedFile).metadata };
+};
+
+const handle = <T>(target: () => T | Promise<T>, callback: Callback<T>) => {
+  void Promise.resolve(true)
+    .then(() => target())
+    .then((result) => callback(null, result))
+    .catch((error) => callback(error));
+};
+
+export class UploadInterceptor implements NestInterceptor {
+  private handler: RequestHandler;
+  private onRequest: OnRequest;
+
+  constructor(
+    private options: {
+      /** pre-request hook */
+      onRequest?: OnRequest;
+      configure(instance: multer.Multer): RequestHandler;
+      canUpload(req: UploadRequest, file: UploadFile): boolean;
+      upload(req: UploadRequest, file: UploadingFile): Promise<UploadMetadata>;
+      remove(req: UploadRequest, file: UploadedFile): Promise<void>;
+    },
+  ) {
+    const storage = { _handleFile: this.handleFile.bind(this), _removeFile: this.removeFile.bind(this) };
+    this.handler = options.configure(multer({ fileFilter: this.canUpload.bind(this), storage }));
+    this.onRequest = options.onRequest ?? (() => Promise.resolve());
+  }
+
+  async intercept(context: ExecutionContext, next: CallHandler<any>): Promise<Observable<any>> {
+    const http = context.switchToHttp();
+    const req = http.getRequest<UploadRequest>();
+    const res = http.getResponse<Response>();
+
+    if (!req.user) {
+      throw new UnauthorizedException();
+    }
+
+    req.requestId = v4();
+
+    // hook to preempt the request before file upload
+    const response = await this.onRequest(req, res);
+    if (response) {
+      return response;
+    }
+
+    await new Promise<void>((resolve, reject) => {
+      const next: NextFunction = (error) => (error ? reject(transformException(error)) : resolve());
+      const maybePromise = this.handler(req, res, next);
+      Promise.resolve(maybePromise).catch((error) => reject(error));
+    });
+
+    return next.handle();
+  }
+
+  private canUpload(req: UploadRequest, file: Express.Multer.File, callback: multer.FileFilterCallback) {
+    return handle(() => this.options.canUpload(req, mapUploadFile(req, file)), callback);
+  }
+
+  private handleFile(req: UploadRequest, file: Express.Multer.File, callback: Callback<UploadMetadata>) {
+    return handle<any>(
+      () =>
+        this.options
+          .upload(req, { ...mapUploadFile(req, file), stream: file.stream })
+          .then((metadata) => ({ metadata })),
+      callback,
+    );
+  }
+
+  private removeFile(req: UploadRequest, file: Express.Multer.File, callback: Callback<void>) {
+    return handle(() => this.options.remove(req, mapUploadedFile(req, file)), callback);
+  }
+}

server/src/middleware/user-profile-upload.interceptor.ts

@@ -0,0 +1,17 @@
+import { Injectable } from '@nestjs/common';
+import multer from 'multer';
+import { UploadFieldName } from 'src/dtos/asset-media.dto';
+import { UploadInterceptor } from 'src/middleware/upload.interceptor';
+import { UserService } from 'src/services/user.service';
+
+@Injectable()
+export class UserProfileUploadInterceptor extends UploadInterceptor {
+  constructor(service: UserService) {
+    super({
+      configure: (instance: multer.Multer) => instance.single(UploadFieldName.PROFILE_DATA),
+      canUpload: (req, file) => service.canUpload(req.user, file),
+      upload: (req, file) => service.onUpload(req.user, file),
+      remove: (req, file) => service.onUploadRemove(req.user, file),
+    });
+  }
+}

server/src/middleware/websocket.adapter.ts

@@ -1,103 +1,21 @@
-import { INestApplication, Logger } from '@nestjs/common';
+import { INestApplicationContext } from '@nestjs/common';
 import { IoAdapter } from '@nestjs/platform-socket.io';
-import { Pool, PoolConfig } from 'pg';
-import type { ServerOptions } from 'socket.io';
-import { SocketIoAdapter } from 'src/enum';
-import { createBroadcastChannelAdapter } from 'src/middleware/broadcast-channel.adapter';
+import { createAdapter } from '@socket.io/redis-adapter';
+import { Redis } from 'ioredis';
+import { ServerOptions } from 'socket.io';
 import { ConfigRepository } from 'src/repositories/config.repository';
-import { asPostgresConnectionConfig } from 'src/utils/database';
 
-export type Ssl = 'require' | 'allow' | 'prefer' | 'verify-full' | boolean | object;
-
-export function asPgPoolSsl(ssl?: Ssl): PoolConfig['ssl'] {
-  if (ssl === undefined || ssl === false || ssl === 'allow') {
-    return false;
-  }
-
-  if (ssl === true || ssl === 'prefer' || ssl === 'require') {
-    return { rejectUnauthorized: false };
-  }
-
-  if (ssl === 'verify-full') {
-    return { rejectUnauthorized: true };
-  }
-
-  return ssl;
-}
-
-class BroadcastChannelSocketAdapter extends IoAdapter {
-  private adapterConstructor: ReturnType<typeof createBroadcastChannelAdapter>;
-
-  constructor(app: INestApplication) {
+export class WebSocketAdapter extends IoAdapter {
+  constructor(private app: INestApplicationContext) {
     super(app);
-    this.adapterConstructor = createBroadcastChannelAdapter();
   }
 
   createIOServer(port: number, options?: ServerOptions): any {
+    const { redis } = this.app.get(ConfigRepository).getEnv();
     const server = super.createIOServer(port, options);
-    server.adapter(this.adapterConstructor);
+    const pubClient = new Redis(redis);
+    const subClient = pubClient.duplicate();
+    server.adapter(createAdapter(pubClient, subClient));
     return server;
   }
 }
-
-class PostgresSocketAdapter extends IoAdapter {
-  private adapterConstructor: any;
-
-  constructor(app: INestApplication, adapterConstructor: any) {
-    super(app);
-    this.adapterConstructor = adapterConstructor;
-  }
-
-  createIOServer(port: number, options?: ServerOptions): any {
-    const server = super.createIOServer(port, options);
-    server.adapter(this.adapterConstructor);
-    return server;
-  }
-}
-
-export async function createWebSocketAdapter(
-  app: INestApplication,
-  adapterOverride?: SocketIoAdapter,
-): Promise<IoAdapter> {
-  const logger = new Logger('WebSocketAdapter');
-  const config = new ConfigRepository();
-  const { database, socketIo } = config.getEnv();
-  const adapter = adapterOverride ?? socketIo.adapter;
-
-  switch (adapter) {
-    case SocketIoAdapter.Postgres: {
-      logger.log('Using Postgres Socket.IO adapter');
-      const { createAdapter } = await import('@socket.io/postgres-adapter');
-      const config = asPostgresConnectionConfig(database.config);
-      const pool = new Pool({
-        host: config.host,
-        port: config.port,
-        user: config.username,
-        password: config.password,
-        database: config.database,
-        ssl: asPgPoolSsl(config.ssl),
-        max: 2,
-      });
-
-      await pool.query(`
-        CREATE TABLE IF NOT EXISTS socket_io_attachments (
-            id          bigserial UNIQUE,
-            created_at  timestamptz DEFAULT NOW(),
-            payload     bytea
-        );
-      `);
-
-      pool.on('error', (error) => {
-        logger.error(' Postgres pool error', error);
-      });
-
-      const adapterConstructor = createAdapter(pool);
-      return new PostgresSocketAdapter(app, adapterConstructor);
-    }
-
-    case SocketIoAdapter.BroadcastChannel: {
-      logger.log('Using BroadcastChannel Socket.IO adapter');
-      return new BroadcastChannelSocketAdapter(app);
-    }
-  }
-}

server/src/repositories/app.repository.ts

@@ -1,4 +1,7 @@
 import { Injectable } from '@nestjs/common';
+import { createAdapter } from '@socket.io/redis-adapter';
+import Redis from 'ioredis';
+import { Server as SocketIO } from 'socket.io';
 import { ExitCode } from 'src/enum';
 import { ConfigRepository } from 'src/repositories/config.repository';
 import { AppRestartEvent } from 'src/repositories/event.repository';
@@ -21,17 +24,24 @@ export class AppRepository {
   }
 
   async sendOneShotAppRestart(state: AppRestartEvent): Promise<void> {
-    const { port } = new ConfigRepository().getEnv();
-    const url = `http://127.0.0.1:${port}/api/internal/restart`;
+    const server = new SocketIO();
+    const { redis } = new ConfigRepository().getEnv();
+    const pubClient = new Redis({ ...redis, lazyConnect: true });
+    const subClient = pubClient.duplicate();
 
-    const response = await fetch(url, {
-      method: 'POST',
-      headers: { 'Content-Type': 'application/json' },
-      body: JSON.stringify(state),
+    await Promise.all([pubClient.connect(), subClient.connect()]);
+
+    server.adapter(createAdapter(pubClient, subClient));
+
+    // => corresponds to notification.service.ts#onAppRestart
+    server.emit('AppRestartV1', state, async () => {
+      const responses = await server.serverSideEmitWithAck('AppRestart', state);
+      if (responses.some((response) => response !== 'ok')) {
+        throw new Error("One or more node(s) returned a non-'ok' response to our restart request!");
+      }
+
+      pubClient.disconnect();
+      subClient.disconnect();
     });
-
-    if (!response.ok) {
-      throw new Error(`Failed to trigger app restart: ${response.status} ${response.statusText}`);
-    }
   }
 }

server/src/repositories/config.repository.ts

@@ -20,7 +20,6 @@ import {
   LogFormat,
   LogLevel,
   QueueName,
-  SocketIoAdapter,
 } from 'src/enum';
 import { DatabaseConnectionParams, VectorExtension } from 'src/types';
 import { setDifference } from 'src/utils/set';
@@ -117,10 +116,6 @@ export interface EnvData {
     };
   };
 
-  socketIo: {
-    adapter: SocketIoAdapter;
-  };
-
   noColor: boolean;
   nodeVersion?: string;
 }
@@ -351,10 +346,6 @@ const getEnv = (): EnvData => {
       },
     },
 
-    socketIo: {
-      adapter: dto.IMMICH_SOCKETIO_ADAPTER ?? SocketIoAdapter.Postgres,
-    },
-
     noColor: !!dto.NO_COLOR,
   };
 };

server/src/services/asset-media.service.spec.ts

@@ -1,9 +1,4 @@
-import {
-  BadRequestException,
-  InternalServerErrorException,
-  NotFoundException,
-  UnauthorizedException,
-} from '@nestjs/common';
+import { BadRequestException, InternalServerErrorException, NotFoundException } from '@nestjs/common';
 import { Stats } from 'node:fs';
 import { AssetFile } from 'src/database';
 import { AssetMediaStatus, AssetRejectReason, AssetUploadAction } from 'src/dtos/asset-media-response.dto';
@@ -11,9 +6,8 @@ import { AssetMediaCreateDto, AssetMediaReplaceDto, AssetMediaSize, UploadFieldN
 import { MapAsset } from 'src/dtos/asset-response.dto';
 import { AssetEditAction } from 'src/dtos/editing.dto';
 import { AssetFileType, AssetStatus, AssetType, AssetVisibility, CacheControl, JobName } from 'src/enum';
-import { AuthRequest } from 'src/middleware/auth.guard';
+import { UploadFile } from 'src/middleware/upload.interceptor';
 import { AssetMediaService } from 'src/services/asset-media.service';
-import { UploadBody } from 'src/types';
 import { ASSET_CHECKSUM_CONSTRAINT } from 'src/utils/database';
 import { ImmichFileResponse } from 'src/utils/file';
 import { AssetFileFactory } from 'test/factories/asset-file.factory';
@@ -22,38 +16,17 @@ import { AuthFactory } from 'test/factories/auth.factory';
 import { authStub } from 'test/fixtures/auth.stub';
 import { fileStub } from 'test/fixtures/file.stub';
 import { userStub } from 'test/fixtures/user.stub';
+import { newUuid } from 'test/small.factory';
 import { newTestService, ServiceMocks } from 'test/utils';
 
 const file1 = Buffer.from('d2947b871a706081be194569951b7db246907957', 'hex');
 
-const uploadFile = {
-  nullAuth: {
-    auth: null,
-    body: {},
-    fieldName: UploadFieldName.ASSET_DATA,
-    file: {
-      uuid: 'random-uuid',
-      checksum: Buffer.from('checksum', 'utf8'),
-      originalPath: '/data/library/admin/image.jpeg',
-      originalName: 'image.jpeg',
-      size: 1000,
-    },
-  },
-  filename: (fieldName: UploadFieldName, filename: string, body?: UploadBody) => {
-    return {
-      auth: authStub.admin,
-      body: body || {},
-      fieldName,
-      file: {
-        uuid: 'random-uuid',
-        mimeType: 'image/jpeg',
-        checksum: Buffer.from('checksum', 'utf8'),
-        originalPath: `/data/admin/${filename}`,
-        originalName: filename,
-        size: 1000,
-      },
-    };
-  },
+const create = (fieldName: UploadFieldName, originalName: string): UploadFile => {
+  return {
+    requestId: newUuid(),
+    fieldName,
+    originalName,
+  };
 };
 
 const validImages = [
@@ -208,17 +181,17 @@ describe(AssetMediaService.name, () => {
 
   describe('getUploadAssetIdByChecksum', () => {
     it('should return if checksum is undefined', async () => {
-      await expect(sut.getUploadAssetIdByChecksum(authStub.admin)).resolves.toBe(undefined);
+      await expect(sut.onBeforeUpload(authStub.admin)).resolves.toBe(undefined);
     });
 
     it('should handle a non-existent asset', async () => {
-      await expect(sut.getUploadAssetIdByChecksum(authStub.admin, file1.toString('hex'))).resolves.toBeUndefined();
+      await expect(sut.onBeforeUpload(authStub.admin, file1.toString('hex'))).resolves.toBeUndefined();
       expect(mocks.asset.getUploadAssetIdByChecksum).toHaveBeenCalledWith(authStub.admin.user.id, file1);
     });
 
     it('should find an existing asset', async () => {
       mocks.asset.getUploadAssetIdByChecksum.mockResolvedValue('asset-id');
-      await expect(sut.getUploadAssetIdByChecksum(authStub.admin, file1.toString('hex'))).resolves.toEqual({
+      await expect(sut.onBeforeUpload(authStub.admin, file1.toString('hex'))).resolves.toEqual({
         id: 'asset-id',
         status: AssetMediaStatus.DUPLICATE,
       });
@@ -227,7 +200,7 @@ describe(AssetMediaService.name, () => {
 
     it('should find an existing asset by base64', async () => {
       mocks.asset.getUploadAssetIdByChecksum.mockResolvedValue('asset-id');
-      await expect(sut.getUploadAssetIdByChecksum(authStub.admin, file1.toString('base64'))).resolves.toEqual({
+      await expect(sut.onBeforeUpload(authStub.admin, file1.toString('base64'))).resolves.toEqual({
         id: 'asset-id',
         status: AssetMediaStatus.DUPLICATE,
       });
@@ -236,21 +209,17 @@ describe(AssetMediaService.name, () => {
   });
 
   describe('canUpload', () => {
-    it('should require an authenticated user', () => {
-      expect(() => sut.canUploadFile(uploadFile.nullAuth)).toThrowError(UnauthorizedException);
-    });
-
     for (const { fieldName, valid, invalid } of uploadTests) {
       describe(fieldName, () => {
         for (const filetype of valid) {
           it(`should accept ${filetype}`, () => {
-            expect(sut.canUploadFile(uploadFile.filename(fieldName, `asset${filetype}`))).toEqual(true);
+            expect(sut.canUpload(AuthFactory.create(), create(fieldName, `asset${filetype}`))).toEqual(true);
           });
         }
 
         for (const filetype of invalid) {
           it(`should reject ${filetype}`, () => {
-            expect(() => sut.canUploadFile(uploadFile.filename(fieldName, `asset${filetype}`))).toThrowError(
+            expect(() => sut.canUpload(AuthFactory.create(), create(fieldName, `asset${filetype}`))).toThrowError(
               BadRequestException,
             );
           });
@@ -265,70 +234,22 @@ describe(AssetMediaService.name, () => {
         });
       });
     }
-
-    it('should prefer filename from body over name from path', () => {
-      const pathFilename = 'invalid-file-name';
-      const body = { filename: 'video.mov' };
-      expect(() => sut.canUploadFile(uploadFile.filename(UploadFieldName.ASSET_DATA, pathFilename))).toThrowError(
-        BadRequestException,
-      );
-      expect(sut.canUploadFile(uploadFile.filename(UploadFieldName.ASSET_DATA, pathFilename, body))).toEqual(true);
-    });
-  });
-
-  describe('getUploadFilename', () => {
-    it('should require authentication', () => {
-      expect(() => sut.getUploadFilename(uploadFile.nullAuth)).toThrowError(UnauthorizedException);
-    });
-
-    it('should be the original extension for asset upload', () => {
-      expect(sut.getUploadFilename(uploadFile.filename(UploadFieldName.ASSET_DATA, 'image.jpg'))).toEqual(
-        'random-uuid.jpg',
-      );
-    });
-
-    it('should be the xmp extension for sidecar upload', () => {
-      expect(sut.getUploadFilename(uploadFile.filename(UploadFieldName.SIDECAR_DATA, 'image.html'))).toEqual(
-        'random-uuid.xmp',
-      );
-    });
-
-    it('should be the original extension for profile upload', () => {
-      expect(sut.getUploadFilename(uploadFile.filename(UploadFieldName.PROFILE_DATA, 'image.jpg'))).toEqual(
-        'random-uuid.jpg',
-      );
-    });
-  });
-
-  describe('getUploadFolder', () => {
-    it('should require authentication', () => {
-      expect(() => sut.getUploadFolder(uploadFile.nullAuth)).toThrowError(UnauthorizedException);
-    });
-
-    it('should return profile for profile uploads', () => {
-      expect(sut.getUploadFolder(uploadFile.filename(UploadFieldName.PROFILE_DATA, 'image.jpg'))).toEqual(
-        expect.stringContaining('/data/profile/admin_id'),
-      );
-      expect(mocks.storage.mkdirSync).toHaveBeenCalledWith(expect.stringContaining('/data/profile/admin_id'));
-    });
-
-    it('should return upload for everything else', () => {
-      expect(sut.getUploadFolder(uploadFile.filename(UploadFieldName.ASSET_DATA, 'image.jpg'))).toEqual(
-        expect.stringContaining('/data/upload/admin_id/ra/nd'),
-      );
-      expect(mocks.storage.mkdirSync).toHaveBeenCalledWith(expect.stringContaining('/data/upload/admin_id/ra/nd'));
-    });
   });
 
   describe('uploadAsset', () => {
     it('should throw an error if the quota is exceeded', async () => {
       const file = {
-        uuid: 'random-uuid',
-        originalPath: 'fake_path/asset_1.jpeg',
-        mimeType: 'image/jpeg',
-        checksum: Buffer.from('file hash', 'utf8'),
+        requestId: '1',
+        fieldName: 'assetData',
         originalName: 'asset_1.jpeg',
-        size: 42,
+        metadata: {
+          uuid: 'random-uuid',
+          path: 'fake_path/asset_1.jpeg',
+          folder: 'fake_path',
+          filename: 'asset_1.jpeg',
+          checksum: Buffer.from('file hash', 'utf8'),
+          size: 42,
+        },
       };
 
       mocks.asset.create.mockResolvedValue(assetEntity);
@@ -342,9 +263,9 @@ describe(AssetMediaService.name, () => {
       ).rejects.toBeInstanceOf(BadRequestException);
 
       expect(mocks.asset.create).not.toHaveBeenCalled();
-      expect(mocks.user.updateUsage).not.toHaveBeenCalledWith(authStub.user1.user.id, file.size);
+      expect(mocks.user.updateUsage).not.toHaveBeenCalledWith(authStub.user1.user.id, file.metadata.size);
       expect(mocks.storage.utimes).not.toHaveBeenCalledWith(
-        file.originalPath,
+        file.metadata.path,
         expect.any(Date),
         new Date(createDto.fileModifiedAt),
       );
@@ -1017,31 +938,4 @@ describe(AssetMediaService.name, () => {
       expect(mocks.asset.getByChecksums).toHaveBeenCalledWith(authStub.admin.user.id, [file1, file2]);
     });
   });
-
-  describe('onUploadError', () => {
-    it('should queue a job to delete the uploaded file', async () => {
-      const request = {
-        body: {},
-        user: authStub.user1,
-      } as AuthRequest;
-
-      const file = {
-        fieldname: UploadFieldName.ASSET_DATA,
-        originalname: 'image.jpg',
-        mimetype: 'image/jpeg',
-        buffer: Buffer.from(''),
-        size: 1000,
-        uuid: 'random-uuid',
-        checksum: Buffer.from('checksum', 'utf8'),
-        originalPath: '/data/upload/user-id/ra/nd/random-uuid.jpg',
-      } as unknown as Express.Multer.File;
-
-      await sut.onUploadError(request, file);
-
-      expect(mocks.job.queue).toHaveBeenCalledWith({
-        name: JobName.FileDelete,
-        data: { files: [expect.stringContaining('/data/upload/user-id/ra/nd/random-uuid.jpg')] },
-      });
-    });
-  });
 });

server/src/services/asset-media.service.ts

@@ -1,5 +1,7 @@
 import { BadRequestException, Injectable, InternalServerErrorException, NotFoundException } from '@nestjs/common';
-import { extname } from 'node:path';
+import { createHash } from 'node:crypto';
+import { extname, join } from 'node:path';
+import { pipeline } from 'node:stream/promises';
 import sanitize from 'sanitize-filename';
 import { StorageCore } from 'src/cores/storage.core';
 import { Asset } from 'src/database';
@@ -31,11 +33,10 @@ import {
   Permission,
   StorageFolder,
 } from 'src/enum';
-import { AuthRequest } from 'src/middleware/auth.guard';
+import { UploadedFile, UploadFile, UploadingFile, UploadMetadata } from 'src/middleware/upload.interceptor';
 import { BaseService } from 'src/services/base.service';
-import { UploadFile, UploadRequest } from 'src/types';
 import { requireUploadAccess } from 'src/utils/access';
-import { asUploadRequest, onBeforeLink } from 'src/utils/asset.util';
+import { onBeforeLink } from 'src/utils/asset.util';
 import { isAssetChecksumConstraint } from 'src/utils/database';
 import { getFilenameExtension, getFileNameWithoutExtension, ImmichFileResponse } from 'src/utils/file';
 import { mimeTypes } from 'src/utils/mime-types';
@@ -47,8 +48,8 @@ export interface AssetMediaRedirectResponse {
 
 @Injectable()
 export class AssetMediaService extends BaseService {
-  async getUploadAssetIdByChecksum(auth: AuthDto, checksum?: string): Promise<AssetMediaResponseDto | undefined> {
-    if (!checksum) {
+  async onBeforeUpload(auth: AuthDto, checksum?: string): Promise<AssetMediaResponseDto | undefined> {
+    if (!checksum || !auth) {
       return;
     }
 
@@ -60,78 +61,56 @@ export class AssetMediaService extends BaseService {
     return { id: assetId, status: AssetMediaStatus.DUPLICATE };
   }
 
-  canUploadFile({ auth, fieldName, file, body }: UploadRequest): true {
+  canUpload(auth: AuthDto, file: UploadFile): true {
     requireUploadAccess(auth);
 
-    const filename = body.filename || file.originalName;
-
-    switch (fieldName) {
-      case UploadFieldName.ASSET_DATA: {
-        if (mimeTypes.isAsset(filename)) {
-          return true;
-        }
-        break;
-      }
-
-      case UploadFieldName.SIDECAR_DATA: {
-        if (mimeTypes.isSidecar(filename)) {
-          return true;
-        }
-        break;
-      }
-
-      case UploadFieldName.PROFILE_DATA: {
-        if (mimeTypes.isProfile(filename)) {
-          return true;
-        }
-        break;
-      }
+    if (
+      (file.fieldName === UploadFieldName.ASSET_DATA && mimeTypes.isAsset(file.originalName)) ||
+      (file.fieldName === UploadFieldName.SIDECAR_DATA && mimeTypes.isSidecar(file.originalName))
+    ) {
+      return true;
     }
 
-    this.logger.error(`Unsupported file type ${filename}`);
-    throw new BadRequestException(`Unsupported file type ${filename}`);
+    this.logger.error(`Unsupported file type ${file.originalName}`);
+    throw new BadRequestException(`Unsupported file type ${file.originalName}`);
   }
 
-  getUploadFilename({ auth, fieldName, file, body }: UploadRequest): string {
-    requireUploadAccess(auth);
+  async onUpload(auth: AuthDto, file: UploadingFile): Promise<UploadMetadata> {
+    const stream = file.stream;
+    let checksum: Buffer | undefined;
+    let size = 0;
 
-    const extension = extname(body.filename || file.originalName);
+    const hash = createHash('sha1');
 
-    const lookup = {
-      [UploadFieldName.ASSET_DATA]: extension,
-      [UploadFieldName.SIDECAR_DATA]: '.xmp',
-      [UploadFieldName.PROFILE_DATA]: extension,
-    };
+    stream
+      .on('data', (chunk: Buffer) => {
+        hash.update(chunk);
+        size += chunk.length;
+      })
+      .on('end', () => (checksum = hash.digest()))
+      .on('error', () => hash.destroy());
 
-    return sanitize(`${file.uuid}${lookup[fieldName]}`);
-  }
-
-  getUploadFolder({ auth, fieldName, file }: UploadRequest): string {
-    auth = requireUploadAccess(auth);
-
-    let folder = StorageCore.getNestedFolder(StorageFolder.Upload, auth.user.id, file.uuid);
-    if (fieldName === UploadFieldName.PROFILE_DATA) {
-      folder = StorageCore.getFolderLocation(StorageFolder.Profile, auth.user.id);
-    }
+    const extension = file.fieldName === UploadFieldName.ASSET_DATA ? extname(file.originalName) : '.xmp';
+    const filename = sanitize(`${file.requestId}${extension}`);
+    const folder = StorageCore.getNestedFolder(StorageFolder.Upload, auth.user.id, filename);
+    const path = join(folder, filename);
 
     this.storageRepository.mkdirSync(folder);
 
-    return folder;
+    await pipeline(stream, this.storageRepository.createWriteStream(path));
+
+    return { filename, folder, path, checksum, size };
   }
 
-  async onUploadError(request: AuthRequest, file: Express.Multer.File) {
-    const uploadFilename = this.getUploadFilename(asUploadRequest(request, file));
-    const uploadFolder = this.getUploadFolder(asUploadRequest(request, file));
-    const uploadPath = `${uploadFolder}/${uploadFilename}`;
-
-    await this.jobRepository.queue({ name: JobName.FileDelete, data: { files: [uploadPath] } });
+  async onUploadRemove(auth: AuthDto, file: UploadedFile): Promise<void> {
+    await this.storageRepository.unlink(file.metadata.path);
   }
 
   async uploadAsset(
     auth: AuthDto,
     dto: AssetMediaCreateDto,
-    file: UploadFile,
-    sidecarFile?: UploadFile,
+    file: UploadedFile,
+    sidecarFile?: UploadedFile,
   ): Promise<AssetMediaResponseDto> {
     try {
       await this.requireAccess({
@@ -141,7 +120,7 @@ export class AssetMediaService extends BaseService {
         ids: [auth.user.id],
       });
 
-      this.requireQuota(auth, file.size);
+      this.requireQuota(auth, file.metadata.size);
 
       if (dto.livePhotoVideoId) {
         await onBeforeLink(
@@ -151,7 +130,7 @@ export class AssetMediaService extends BaseService {
       }
       const asset = await this.create(auth.user.id, dto, file, sidecarFile);
 
-      await this.userRepository.updateUsage(auth.user.id, file.size);
+      await this.userRepository.updateUsage(auth.user.id, file.metadata.size);
 
       return { id: asset.id, status: AssetMediaStatus.CREATED };
     } catch (error: any) {
@@ -163,8 +142,8 @@ export class AssetMediaService extends BaseService {
     auth: AuthDto,
     id: string,
     dto: AssetMediaReplaceDto,
-    file: UploadFile,
-    sidecarFile?: UploadFile,
+    file: UploadedFile,
+    sidecarFile?: UploadedFile,
   ): Promise<AssetMediaResponseDto> {
     try {
       await this.requireAccess({ auth, permission: Permission.AssetUpdate, ids: [id] });
@@ -174,9 +153,9 @@ export class AssetMediaService extends BaseService {
         throw new Error('Asset not found');
       }
 
-      this.requireQuota(auth, file.size);
+      this.requireQuota(auth, file.metadata.size);
 
-      await this.replaceFileData(asset.id, dto, file, sidecarFile?.originalPath);
+      await this.replaceFileData(asset.id, dto, file, sidecarFile?.metadata.path);
 
       // Next, create a backup copy of the existing record. The db record has already been updated above,
       // but the local variable holds the original file data paths.
@@ -185,7 +164,7 @@ export class AssetMediaService extends BaseService {
       await this.assetRepository.updateAll([copiedPhoto.id], { deletedAt: new Date(), status: AssetStatus.Trashed });
       await this.eventRepository.emit('AssetTrash', { assetId: copiedPhoto.id, userId: auth.user.id });
 
-      await this.userRepository.updateUsage(auth.user.id, file.size);
+      await this.userRepository.updateUsage(auth.user.id, file.metadata.size);
 
       return { status: AssetMediaStatus.REPLACED, id: copiedPhoto.id };
     } catch (error: any) {
@@ -325,18 +304,18 @@ export class AssetMediaService extends BaseService {
   private async handleUploadError(
     error: any,
     auth: AuthDto,
-    file: UploadFile,
-    sidecarFile?: UploadFile,
+    file: UploadedFile,
+    sidecarFile?: UploadedFile,
   ): Promise<AssetMediaResponseDto> {
     // clean up files
     await this.jobRepository.queue({
       name: JobName.FileDelete,
-      data: { files: [file.originalPath, sidecarFile?.originalPath] },
+      data: { files: [file.metadata.path, sidecarFile?.metadata.path] },
     });
 
     // handle duplicates with a success response
     if (isAssetChecksumConstraint(error)) {
-      const duplicateId = await this.assetRepository.getUploadAssetIdByChecksum(auth.user.id, file.checksum);
+      const duplicateId = await this.assetRepository.getUploadAssetIdByChecksum(auth.user.id, file.metadata.checksum!);
       if (!duplicateId) {
         this.logger.error(`Error locating duplicate for checksum constraint`);
         throw new InternalServerErrorException();
@@ -358,15 +337,15 @@ export class AssetMediaService extends BaseService {
   private async replaceFileData(
     assetId: string,
     dto: AssetMediaReplaceDto,
-    file: UploadFile,
+    file: UploadedFile,
     sidecarPath?: string,
   ): Promise<void> {
     await this.assetRepository.update({
       id: assetId,
 
-      checksum: file.checksum,
-      originalPath: file.originalPath,
-      type: mimeTypes.assetType(file.originalPath),
+      checksum: file.metadata.checksum,
+      originalPath: file.metadata.path,
+      type: mimeTypes.assetType(file.metadata.path),
       originalFileName: file.originalName,
 
       deviceAssetId: dto.deviceAssetId,
@@ -383,9 +362,9 @@ export class AssetMediaService extends BaseService {
       ? this.assetRepository.upsertFile({ assetId, type: AssetFileType.Sidecar, path: sidecarPath })
       : this.assetRepository.deleteFile({ assetId, type: AssetFileType.Sidecar }));
 
-    await this.storageRepository.utimes(file.originalPath, new Date(), new Date(dto.fileModifiedAt));
+    await this.storageRepository.utimes(file.metadata.path, new Date(), new Date(dto.fileModifiedAt));
     await this.assetRepository.upsertExif(
-      { assetId, fileSizeInByte: file.size },
+      { assetId, fileSizeInByte: file.metadata.size },
       { lockedPropertiesBehavior: 'override' },
     );
     await this.jobRepository.queue({
@@ -424,13 +403,13 @@ export class AssetMediaService extends BaseService {
     return created;
   }
 
-  private async create(ownerId: string, dto: AssetMediaCreateDto, file: UploadFile, sidecarFile?: UploadFile) {
+  private async create(ownerId: string, dto: AssetMediaCreateDto, file: UploadedFile, sidecarFile?: UploadedFile) {
     const asset = await this.assetRepository.create({
       ownerId,
       libraryId: null,
 
-      checksum: file.checksum,
-      originalPath: file.originalPath,
+      checksum: file.metadata.checksum!,
+      originalPath: file.metadata.path,
 
       deviceAssetId: dto.deviceAssetId,
       deviceId: dto.deviceId,
@@ -439,7 +418,7 @@ export class AssetMediaService extends BaseService {
       fileModifiedAt: dto.fileModifiedAt,
       localDateTime: dto.fileCreatedAt,
 
-      type: mimeTypes.assetType(file.originalPath),
+      type: mimeTypes.assetType(file.metadata.path),
       isFavorite: dto.isFavorite,
       duration: dto.duration || null,
       visibility: dto.visibility ?? AssetVisibility.Timeline,
@@ -454,14 +433,14 @@ export class AssetMediaService extends BaseService {
     if (sidecarFile) {
       await this.assetRepository.upsertFile({
         assetId: asset.id,
-        path: sidecarFile.originalPath,
+        path: sidecarFile.metadata.path,
         type: AssetFileType.Sidecar,
       });
-      await this.storageRepository.utimes(sidecarFile.originalPath, new Date(), new Date(dto.fileModifiedAt));
+      await this.storageRepository.utimes(sidecarFile.metadata.path, new Date(), new Date(dto.fileModifiedAt));
     }
-    await this.storageRepository.utimes(file.originalPath, new Date(), new Date(dto.fileModifiedAt));
+    await this.storageRepository.utimes(file.metadata.path, new Date(), new Date(dto.fileModifiedAt));
     await this.assetRepository.upsertExif(
-      { assetId: asset.id, fileSizeInByte: file.size },
+      { assetId: asset.id, fileSizeInByte: file.metadata.size },
       { lockedPropertiesBehavior: 'override' },
     );
 

server/src/services/user.service.ts

@@ -1,6 +1,9 @@
 import { BadRequestException, Injectable, NotFoundException } from '@nestjs/common';
 import { Updateable } from 'kysely';
 import { DateTime } from 'luxon';
+import { extname, join } from 'node:path';
+import { pipeline } from 'node:stream/promises';
+import sanitize from 'sanitize-filename';
 import { SALT_ROUNDS } from 'src/constants';
 import { StorageCore } from 'src/cores/storage.core';
 import { OnJob } from 'src/decorators';
@@ -11,15 +14,41 @@ import { UserPreferencesResponseDto, UserPreferencesUpdateDto, mapPreferences }
 import { CreateProfileImageResponseDto } from 'src/dtos/user-profile.dto';
 import { UserAdminResponseDto, UserResponseDto, UserUpdateMeDto, mapUser, mapUserAdmin } from 'src/dtos/user.dto';
 import { CacheControl, JobName, JobStatus, QueueName, StorageFolder, UserMetadataKey } from 'src/enum';
+import { UploadFile, UploadMetadata, UploadedFile, UploadingFile } from 'src/middleware/upload.interceptor';
 import { UserFindOptions } from 'src/repositories/user.repository';
 import { UserTable } from 'src/schema/tables/user.table';
 import { BaseService } from 'src/services/base.service';
 import { JobOf, UserMetadataItem } from 'src/types';
 import { ImmichFileResponse } from 'src/utils/file';
+import { mimeTypes } from 'src/utils/mime-types';
 import { getPreferences, getPreferencesPartial, mergePreferences } from 'src/utils/preferences';
 
 @Injectable()
 export class UserService extends BaseService {
+  canUpload(auth: AuthDto, file: UploadFile) {
+    return mimeTypes.isProfile(file.originalName);
+  }
+
+  async onUpload(auth: AuthDto, file: UploadingFile): Promise<UploadMetadata> {
+    const extension = extname(file.originalName);
+    const filename = sanitize(`${file.requestId}${extension}`);
+    const folder = StorageCore.getNestedFolder(StorageFolder.Profile, auth.user.id, filename);
+    const path = join(folder, filename);
+
+    this.storageRepository.mkdirSync(folder);
+
+    let size = 0;
+    file.stream.on('data', (chunk: Buffer) => (size += chunk.length));
+
+    await pipeline(file.stream, this.storageRepository.createWriteStream(path));
+
+    return { filename, folder, path, size };
+  }
+
+  async onUploadRemove(auth: AuthDto, file: UploadedFile) {
+    await this.storageRepository.unlink(file.metadata.path);
+  }
+
   async search(auth: AuthDto): Promise<UserResponseDto[]> {
     const config = await this.getConfig({ withCache: false });
 
@@ -90,11 +119,11 @@ export class UserService extends BaseService {
     return mapUser(user);
   }
 
-  async createProfileImage(auth: AuthDto, file: Express.Multer.File): Promise<CreateProfileImageResponseDto> {
+  async createProfileImage(auth: AuthDto, file: UploadedFile): Promise<CreateProfileImageResponseDto> {
     const { profileImagePath: oldpath } = await this.findOrFail(auth.user.id, { withDeleted: false });
 
     const user = await this.userRepository.update(auth.user.id, {
-      profileImagePath: file.path,
+      profileImagePath: file.metadata.path,
       profileChangedAt: new Date(),
     });
 

server/src/types.ts

@@ -1,8 +1,6 @@
 import { SystemConfig } from 'src/config';
 import { VECTOR_EXTENSIONS } from 'src/constants';
 import { Asset, AssetFile } from 'src/database';
-import { UploadFieldName } from 'src/dtos/asset-media.dto';
-import { AuthDto } from 'src/dtos/auth.dto';
 import { AssetEditActionItem } from 'src/dtos/editing.dto';
 import { SetMaintenanceModeDto } from 'src/dtos/maintenance.dto';
 import {
@@ -420,37 +418,6 @@ export interface VectorUpdateResult {
   restartRequired: boolean;
 }
 
-export interface ImmichFile extends Express.Multer.File {
-  uuid: string;
-  /** sha1 hash of file */
-  checksum: Buffer;
-}
-
-export interface UploadFile {
-  uuid: string;
-  checksum: Buffer;
-  originalPath: string;
-  originalName: string;
-  size: number;
-}
-
-export interface UploadBody {
-  filename?: string;
-  [key: string]: unknown;
-}
-
-export type UploadRequest = {
-  auth: AuthDto | null;
-  fieldName: UploadFieldName;
-  file: UploadFile;
-  body: UploadBody;
-};
-
-export interface UploadFiles {
-  assetData: ImmichFile[];
-  sidecarData: ImmichFile[];
-}
-
 export interface IBulkAsset {
   getAssetIds: (id: string, assetIds: string[]) => Promise<Set<string>>;
   addAssetIds: (id: string, assetIds: string[]) => Promise<void>;

server/src/utils/asset.util.ts

@@ -2,16 +2,14 @@ import { BadRequestException } from '@nestjs/common';
 import { StorageCore } from 'src/cores/storage.core';
 import { AssetFile, Exif } from 'src/database';
 import { BulkIdErrorReason, BulkIdResponseDto } from 'src/dtos/asset-ids.response.dto';
-import { UploadFieldName } from 'src/dtos/asset-media.dto';
 import { AuthDto } from 'src/dtos/auth.dto';
 import { ExifResponseDto } from 'src/dtos/exif.dto';
 import { AssetFileType, AssetType, AssetVisibility, Permission } from 'src/enum';
-import { AuthRequest } from 'src/middleware/auth.guard';
 import { AccessRepository } from 'src/repositories/access.repository';
 import { AssetRepository } from 'src/repositories/asset.repository';
 import { EventRepository } from 'src/repositories/event.repository';
 import { PartnerRepository } from 'src/repositories/partner.repository';
-import { IBulkAsset, ImmichFile, UploadFile, UploadRequest } from 'src/types';
+import { IBulkAsset } from 'src/types';
 import { checkAccess } from 'src/utils/access';
 
 export const getAssetFile = (files: AssetFile[], type: AssetFileType, { isEdited }: { isEdited: boolean }) => {
@@ -186,25 +184,6 @@ export const onAfterUnlink = async (
   await eventRepository.emit('AssetShow', { assetId: livePhotoVideoId, userId });
 };
 
-export function mapToUploadFile(file: ImmichFile): UploadFile {
-  return {
-    uuid: file.uuid,
-    checksum: file.checksum,
-    originalPath: file.path,
-    originalName: Buffer.from(file.originalname, 'latin1').toString('utf8'),
-    size: file.size,
-  };
-}
-
-export const asUploadRequest = (request: AuthRequest, file: Express.Multer.File): UploadRequest => {
-  return {
-    auth: request.user || null,
-    body: request.body,
-    fieldName: file.fieldname as UploadFieldName,
-    file: mapToUploadFile(file as ImmichFile),
-  };
-};
-
 const isFlipped = (orientation?: string | null) => {
   const value = Number(orientation);
   return value && [5, 6, 7, 8, -90, 90].includes(value);

server/src/utils/maintenance.ts

@@ -1,11 +1,60 @@
+import { createAdapter } from '@socket.io/redis-adapter';
+import Redis from 'ioredis';
 import { SignJWT } from 'jose';
 import { randomBytes } from 'node:crypto';
 import { join } from 'node:path';
+import { Server as SocketIO } from 'socket.io';
 import { StorageCore } from 'src/cores/storage.core';
 import { MaintenanceAuthDto, MaintenanceDetectInstallResponseDto } from 'src/dtos/maintenance.dto';
 import { StorageFolder } from 'src/enum';
+import { ConfigRepository } from 'src/repositories/config.repository';
+import { AppRestartEvent } from 'src/repositories/event.repository';
 import { StorageRepository } from 'src/repositories/storage.repository';
 
+export function sendOneShotAppRestart(state: AppRestartEvent): void {
+  const server = new SocketIO();
+  const { redis } = new ConfigRepository().getEnv();
+  const pubClient = new Redis(redis);
+  const subClient = pubClient.duplicate();
+  server.adapter(createAdapter(pubClient, subClient));
+
+  /**
+   * Keep trying until we manage to stop Immich
+   *
+   * Sometimes there appear to be communication
+   * issues between to the other servers.
+   *
+   * This issue only occurs with this method.
+   */
+  async function tryTerminate() {
+    while (true) {
+      try {
+        const responses = await server.serverSideEmitWithAck('AppRestart', state);
+        if (responses.length > 0) {
+          return;
+        }
+      } catch (error) {
+        console.error(error);
+        console.error('Encountered an error while telling Immich to stop.');
+      }
+
+      console.info(
+        "\nIt doesn't appear that Immich stopped, trying again in a moment.\nIf Immich is already not running, you can ignore this error.",
+      );
+
+      await new Promise((r) => setTimeout(r, 1e3));
+    }
+  }
+
+  // => corresponds to notification.service.ts#onAppRestart
+  server.emit('AppRestartV1', state, () => {
+    void tryTerminate().finally(() => {
+      pubClient.disconnect();
+      subClient.disconnect();
+    });
+  });
+}
+
 export async function createMaintenanceLoginUrl(
   baseUrl: string,
   auth: MaintenanceAuthDto,

server/src/workers/api.ts

@@ -1,21 +1,14 @@
 import { NestFactory } from '@nestjs/core';
 import { NestExpressApplication } from '@nestjs/platform-express';
-import inspector from 'node:inspector';
-import { isMainThread, workerData } from 'node:worker_threads';
 import { configureExpress, configureTelemetry } from 'src/app.common';
 import { ApiModule } from 'src/app.module';
 import { AppRepository } from 'src/repositories/app.repository';
 import { ApiService } from 'src/services/api.service';
 import { isStartUpError } from 'src/utils/misc';
 
-export async function bootstrap() {
+async function bootstrap() {
   process.title = 'immich-api';
 
-  const { inspectorPort } = workerData ?? {};
-  if (inspectorPort) {
-    inspector.open(inspectorPort, '0.0.0.0', false);
-  }
-
   configureTelemetry();
 
   const app = await NestFactory.create<NestExpressApplication>(ApiModule, { bufferLogs: true });
@@ -26,12 +19,10 @@ export async function bootstrap() {
   });
 }
 
-if (!isMainThread || process.send) {
-  bootstrap().catch((error) => {
-    if (!isStartUpError(error)) {
-      console.error(error);
-    }
-
-    process.exit(1);
-  });
-}
+bootstrap().catch((error) => {
+  if (!isStartUpError(error)) {
+    console.error(error);
+  }
+  // eslint-disable-next-line unicorn/no-process-exit
+  process.exit(1);
+});

server/src/workers/maintenance.ts

@@ -1,22 +1,13 @@
 import { NestFactory } from '@nestjs/core';
 import { NestExpressApplication } from '@nestjs/platform-express';
-import inspector from 'node:inspector';
-import { isMainThread, workerData } from 'node:worker_threads';
 import { configureExpress, configureTelemetry } from 'src/app.common';
 import { MaintenanceModule } from 'src/app.module';
-import { SocketIoAdapter } from 'src/enum';
 import { MaintenanceWorkerService } from 'src/maintenance/maintenance-worker.service';
 import { AppRepository } from 'src/repositories/app.repository';
 import { isStartUpError } from 'src/utils/misc';
 
-export async function bootstrap() {
+async function bootstrap() {
   process.title = 'immich-maintenance';
-
-  const { inspectorPort } = workerData ?? {};
-  if (inspectorPort) {
-    inspector.open(inspectorPort, '0.0.0.0', false);
-  }
-
   configureTelemetry();
 
   const app = await NestFactory.create<NestExpressApplication>(MaintenanceModule, { bufferLogs: true });
@@ -25,18 +16,13 @@ export async function bootstrap() {
   void configureExpress(app, {
     permitSwaggerWrite: false,
     ssr: MaintenanceWorkerService,
-    // Use BroadcastChannel instead of Postgres adapter to avoid crash when
-    // pg_terminate_backend() kills all database connections during restore
-    socketIoAdapter: SocketIoAdapter.BroadcastChannel,
   });
 }
 
-if (!isMainThread) {
-  bootstrap().catch((error) => {
-    if (!isStartUpError(error)) {
-      console.error(error);
-    }
-
-    process.exit(1);
-  });
-}
+bootstrap().catch((error) => {
+  if (!isStartUpError(error)) {
+    console.error(error);
+  }
+  // eslint-disable-next-line unicorn/no-process-exit
+  process.exit(1);
+});

server/src/workers/microservices.ts

@@ -1,9 +1,8 @@
 import { NestFactory } from '@nestjs/core';
-import inspector from 'node:inspector';
-import { isMainThread, workerData } from 'node:worker_threads';
+import { isMainThread } from 'node:worker_threads';
 import { MicroservicesModule } from 'src/app.module';
 import { serverVersion } from 'src/constants';
-import { createWebSocketAdapter } from 'src/middleware/websocket.adapter';
+import { WebSocketAdapter } from 'src/middleware/websocket.adapter';
 import { AppRepository } from 'src/repositories/app.repository';
 import { ConfigRepository } from 'src/repositories/config.repository';
 import { LoggingRepository } from 'src/repositories/logging.repository';
@@ -11,11 +10,6 @@ import { bootstrapTelemetry } from 'src/repositories/telemetry.repository';
 import { isStartUpError } from 'src/utils/misc';
 
 export async function bootstrap() {
-  const { inspectorPort } = workerData ?? {};
-  if (inspectorPort) {
-    inspector.open(inspectorPort, '0.0.0.0', false);
-  }
-
   const { telemetry } = new ConfigRepository().getEnv();
   if (telemetry.metrics.size > 0) {
     bootstrapTelemetry(telemetry.microservicesPort);
@@ -30,7 +24,7 @@ export async function bootstrap() {
 
   logger.setContext('Bootstrap');
   app.useLogger(logger);
-  app.useWebSocketAdapter(await createWebSocketAdapter(app));
+  app.useWebSocketAdapter(new WebSocketAdapter(app));
 
   await (host ? app.listen(0, host) : app.listen(0));
 

server/test/medium/specs/middleware/broadcast-channel.adapter.spec.ts

@@ -1,276 +0,0 @@
-import { ClusterMessage, ClusterResponse } from 'socket.io-adapter';
-import { createBroadcastChannelAdapter } from 'src/middleware/broadcast-channel.adapter';
-import { vi } from 'vitest';
-
-const createMockNamespace = () => ({
-  name: '/',
-  sockets: new Map(),
-  adapter: null,
-  server: {
-    encoder: {
-      encode: vi.fn().mockReturnValue([]),
-    },
-    _opts: {},
-    sockets: {
-      sockets: new Map(),
-    },
-  },
-});
-
-describe('BroadcastChannelAdapter', () => {
-  describe('createBroadcastChannelAdapter', () => {
-    it('should return a factory function', () => {
-      const factory = createBroadcastChannelAdapter();
-      expect(typeof factory).toBe('function');
-    });
-
-    it('should create adapter instance when factory is called', () => {
-      const mockNamespace = createMockNamespace();
-      const factory = createBroadcastChannelAdapter();
-      const adapter = factory(mockNamespace);
-
-      expect(adapter).toBeDefined();
-      expect(adapter.doPublish).toBeDefined();
-      expect(adapter.doPublishResponse).toBeDefined();
-
-      adapter.close();
-    });
-  });
-
-  describe('BroadcastChannelAdapter message passing', () => {
-    it('should actually send and receive messages between two adapters', async () => {
-      const factory1 = createBroadcastChannelAdapter();
-      const factory2 = createBroadcastChannelAdapter();
-
-      const namespace1 = createMockNamespace();
-      const namespace2 = createMockNamespace();
-
-      const adapter1 = factory1(namespace1);
-      const adapter2 = factory2(namespace2);
-
-      await new Promise((resolve) => setTimeout(resolve, 100));
-
-      const receivedMessages: ClusterMessage[] = [];
-      const messageReceived = new Promise<void>((resolve) => {
-        const originalOnMessage = adapter2.onMessage.bind(adapter2);
-        adapter2.onMessage = (message: ClusterMessage) => {
-          receivedMessages.push(message);
-          resolve();
-          return originalOnMessage(message);
-        };
-      });
-
-      const testMessage = {
-        type: 2,
-        data: {
-          opts: { rooms: new Set(['room1']) },
-          rooms: ['room1'],
-        },
-        nsp: '/',
-      };
-
-      void adapter1.doPublish(testMessage as any);
-
-      await Promise.race([messageReceived, new Promise((resolve) => setTimeout(resolve, 500))]);
-
-      expect(receivedMessages.length).toBeGreaterThan(0);
-
-      adapter1.close();
-      adapter2.close();
-    });
-
-    it('should send ConfigUpdate-style event and receive it on another adapter', async () => {
-      const factory1 = createBroadcastChannelAdapter();
-      const factory2 = createBroadcastChannelAdapter();
-
-      const namespace1 = createMockNamespace();
-      const namespace2 = createMockNamespace();
-
-      const adapter1 = factory1(namespace1);
-      const adapter2 = factory2(namespace2);
-
-      await new Promise((resolve) => setTimeout(resolve, 100));
-
-      const receivedMessages: ClusterMessage[] = [];
-      const messageReceived = new Promise<void>((resolve) => {
-        const originalOnMessage = adapter2.onMessage.bind(adapter2);
-        adapter2.onMessage = (message: ClusterMessage) => {
-          receivedMessages.push(message);
-          if ((message as any)?.data?.event === 'ConfigUpdate') {
-            resolve();
-          }
-          return originalOnMessage(message);
-        };
-      });
-
-      const configUpdateMessage = {
-        type: 2,
-        data: {
-          event: 'ConfigUpdate',
-          payload: { newConfig: { ffmpeg: { crf: 23 } }, oldConfig: { ffmpeg: { crf: 20 } } },
-          opts: { rooms: new Set() },
-          rooms: [],
-        },
-        nsp: '/',
-      };
-
-      void adapter1.doPublish(configUpdateMessage as any);
-
-      await Promise.race([messageReceived, new Promise((resolve) => setTimeout(resolve, 500))]);
-
-      const configMessages = receivedMessages.filter((m) => (m as any)?.data?.event === 'ConfigUpdate');
-      expect(configMessages.length).toBeGreaterThan(0);
-      expect((configMessages[0] as any).data.payload.newConfig.ffmpeg.crf).toBe(23);
-
-      adapter1.close();
-      adapter2.close();
-    });
-
-    it('should send AppRestart-style event and receive it on another adapter', async () => {
-      const factory1 = createBroadcastChannelAdapter();
-      const factory2 = createBroadcastChannelAdapter();
-
-      const namespace1 = createMockNamespace();
-      const namespace2 = createMockNamespace();
-
-      const adapter1 = factory1(namespace1);
-      const adapter2 = factory2(namespace2);
-
-      await new Promise((resolve) => setTimeout(resolve, 100));
-
-      const receivedMessages: ClusterMessage[] = [];
-      const messageReceived = new Promise<void>((resolve) => {
-        const originalOnMessage = adapter2.onMessage.bind(adapter2);
-        adapter2.onMessage = (message: ClusterMessage) => {
-          receivedMessages.push(message);
-          if ((message as any)?.data?.event === 'AppRestart') {
-            resolve();
-          }
-          return originalOnMessage(message);
-        };
-      });
-
-      const appRestartMessage = {
-        type: 2,
-        data: {
-          event: 'AppRestart',
-          payload: { isMaintenanceMode: true },
-          opts: { rooms: new Set() },
-          rooms: [],
-        },
-        nsp: '/',
-      };
-
-      void adapter1.doPublish(appRestartMessage as any);
-
-      await Promise.race([messageReceived, new Promise((resolve) => setTimeout(resolve, 500))]);
-
-      const restartMessages = receivedMessages.filter((m) => (m as any)?.data?.event === 'AppRestart');
-      expect(restartMessages.length).toBeGreaterThan(0);
-      expect((restartMessages[0] as any).data.payload.isMaintenanceMode).toBe(true);
-
-      adapter1.close();
-      adapter2.close();
-    });
-
-    it('should not receive its own messages (echo prevention)', async () => {
-      const factory = createBroadcastChannelAdapter();
-      const namespace = createMockNamespace();
-      const adapter = factory(namespace);
-
-      await new Promise((resolve) => setTimeout(resolve, 100));
-
-      const receivedOwnMessages: ClusterMessage[] = [];
-      const uniqueMarker = `test-${Date.now()}-${Math.random()}`;
-
-      const originalOnMessage = adapter.onMessage.bind(adapter);
-      adapter.onMessage = (message: ClusterMessage) => {
-        if ((message as any)?.data?.marker === uniqueMarker) {
-          receivedOwnMessages.push(message);
-        }
-        return originalOnMessage(message);
-      };
-
-      const testMessage = {
-        type: 2,
-        data: {
-          marker: uniqueMarker,
-          opts: { rooms: new Set() },
-          rooms: [],
-        },
-        nsp: '/',
-      };
-
-      void adapter.doPublish(testMessage as any);
-
-      await new Promise((resolve) => setTimeout(resolve, 200));
-
-      expect(receivedOwnMessages.length).toBe(0);
-
-      adapter.close();
-    });
-
-    it('should send and receive response messages between adapters', async () => {
-      const factory1 = createBroadcastChannelAdapter();
-      const factory2 = createBroadcastChannelAdapter();
-
-      const namespace1 = createMockNamespace();
-      const namespace2 = createMockNamespace();
-
-      const adapter1 = factory1(namespace1);
-      const adapter2 = factory2(namespace2);
-
-      await new Promise((resolve) => setTimeout(resolve, 100));
-
-      const receivedResponses: ClusterResponse[] = [];
-      const responseReceived = new Promise<void>((resolve) => {
-        const originalOnResponse = adapter1.onResponse.bind(adapter1);
-        adapter1.onResponse = (response: ClusterResponse) => {
-          receivedResponses.push(response);
-          resolve();
-          return originalOnResponse(response);
-        };
-      });
-
-      const responseMessage = {
-        type: 3,
-        data: { result: 'success', count: 42 },
-      };
-
-      void adapter2.doPublishResponse((adapter1 as any).uid, responseMessage as any);
-
-      await Promise.race([responseReceived, new Promise((resolve) => setTimeout(resolve, 500))]);
-
-      expect(receivedResponses.length).toBeGreaterThan(0);
-
-      adapter1.close();
-      adapter2.close();
-    });
-  });
-
-  describe('BroadcastChannelAdapter lifecycle', () => {
-    it('should close cleanly without errors', () => {
-      const factory = createBroadcastChannelAdapter();
-      const namespace = createMockNamespace();
-      const adapter = factory(namespace);
-
-      expect(() => adapter.close()).not.toThrow();
-    });
-
-    it('should handle multiple adapters closing in sequence', () => {
-      const factory1 = createBroadcastChannelAdapter();
-      const factory2 = createBroadcastChannelAdapter();
-      const factory3 = createBroadcastChannelAdapter();
-
-      const adapter1 = factory1(createMockNamespace());
-      const adapter2 = factory2(createMockNamespace());
-      const adapter3 = factory3(createMockNamespace());
-
-      expect(() => {
-        adapter1.close();
-        adapter2.close();
-        adapter3.close();
-      }).not.toThrow();
-    });
-  });
-});

server/test/medium/specs/middleware/websocket-integration.spec.ts

@@ -1,159 +0,0 @@
-import { Server } from 'socket.io';
-import { createBroadcastChannelAdapter } from 'src/middleware/broadcast-channel.adapter';
-import { EventRepository } from 'src/repositories/event.repository';
-import { LoggingRepository } from 'src/repositories/logging.repository';
-import { WebsocketRepository } from 'src/repositories/websocket.repository';
-import { automock } from 'test/utils';
-import { vi } from 'vitest';
-
-describe('WebSocket Integration - serverSend with adapters', () => {
-  describe('BroadcastChannel adapter', () => {
-    it('should broadcast ConfigUpdate event through BroadcastChannel adapter', async () => {
-      const createMockNamespace = () => ({
-        name: '/',
-        sockets: new Map(),
-        adapter: null,
-        server: {
-          encoder: { encode: vi.fn().mockReturnValue([]) },
-          _opts: {},
-          sockets: { sockets: new Map() },
-        },
-      });
-
-      const factory1 = createBroadcastChannelAdapter();
-      const factory2 = createBroadcastChannelAdapter();
-
-      const namespace1 = createMockNamespace();
-      const namespace2 = createMockNamespace();
-
-      const adapter1 = factory1(namespace1);
-      const adapter2 = factory2(namespace2);
-
-      await new Promise((resolve) => setTimeout(resolve, 100));
-
-      const receivedMessages: any[] = [];
-      vi.spyOn(adapter2, 'onMessage').mockImplementation((message: any) => {
-        receivedMessages.push(message);
-      });
-
-      const configUpdatePayload = {
-        type: 5,
-        data: {
-          event: 'ConfigUpdate',
-          args: [{ newConfig: { ffmpeg: { crf: 23 } }, oldConfig: { ffmpeg: { crf: 20 } } }],
-        },
-        nsp: '/',
-      };
-
-      void adapter1.doPublish(configUpdatePayload as any);
-
-      await new Promise((resolve) => setTimeout(resolve, 100));
-
-      const configMessages = receivedMessages.filter((m) => m?.data?.event === 'ConfigUpdate');
-      expect(configMessages.length).toBeGreaterThan(0);
-
-      adapter1.close();
-      adapter2.close();
-    });
-
-    it('should broadcast AppRestart event through BroadcastChannel adapter', async () => {
-      const createMockNamespace = () => ({
-        name: '/',
-        sockets: new Map(),
-        adapter: null,
-        server: {
-          encoder: { encode: vi.fn().mockReturnValue([]) },
-          _opts: {},
-          sockets: { sockets: new Map() },
-        },
-      });
-
-      const factory1 = createBroadcastChannelAdapter();
-      const factory2 = createBroadcastChannelAdapter();
-
-      const namespace1 = createMockNamespace();
-      const namespace2 = createMockNamespace();
-
-      const adapter1 = factory1(namespace1);
-      const adapter2 = factory2(namespace2);
-
-      await new Promise((resolve) => setTimeout(resolve, 100));
-
-      const receivedMessages: any[] = [];
-      vi.spyOn(adapter2, 'onMessage').mockImplementation((message: any) => {
-        receivedMessages.push(message);
-      });
-
-      const appRestartPayload = {
-        type: 5,
-        data: {
-          event: 'AppRestart',
-          args: [{ isMaintenanceMode: true }],
-        },
-        nsp: '/',
-      };
-
-      void adapter1.doPublish(appRestartPayload as any);
-
-      await new Promise((resolve) => setTimeout(resolve, 100));
-
-      const restartMessages = receivedMessages.filter((m) => m?.data?.event === 'AppRestart');
-      expect(restartMessages.length).toBeGreaterThan(0);
-
-      adapter1.close();
-      adapter2.close();
-    });
-  });
-
-  describe('WebsocketRepository with adapter', () => {
-    it('should call serverSideEmit when serverSend is called', () => {
-      const mockServer = {
-        serverSideEmit: vi.fn(),
-        on: vi.fn(),
-      } as unknown as Server;
-
-      const eventRepository = automock(EventRepository, {
-        args: [undefined, undefined, { setContext: () => {} }],
-      });
-      const loggingRepository = automock(LoggingRepository, {
-        args: [undefined, { getEnv: () => ({ noColor: false }) }],
-        strict: false,
-      });
-
-      const websocketRepository = new WebsocketRepository(eventRepository, loggingRepository);
-      (websocketRepository as any).server = mockServer;
-
-      websocketRepository.serverSend('ConfigUpdate', {
-        newConfig: { ffmpeg: { crf: 23 } } as any,
-        oldConfig: { ffmpeg: { crf: 20 } } as any,
-      });
-
-      expect(mockServer.serverSideEmit).toHaveBeenCalledWith('ConfigUpdate', {
-        newConfig: { ffmpeg: { crf: 23 } },
-        oldConfig: { ffmpeg: { crf: 20 } },
-      });
-    });
-
-    it('should call serverSideEmit for AppRestart event', () => {
-      const mockServer = {
-        serverSideEmit: vi.fn(),
-        on: vi.fn(),
-      } as unknown as Server;
-
-      const eventRepository = automock(EventRepository, {
-        args: [undefined, undefined, { setContext: () => {} }],
-      });
-      const loggingRepository = automock(LoggingRepository, {
-        args: [undefined, { getEnv: () => ({ noColor: false }) }],
-        strict: false,
-      });
-
-      const websocketRepository = new WebsocketRepository(eventRepository, loggingRepository);
-      (websocketRepository as any).server = mockServer;
-
-      websocketRepository.serverSend('AppRestart', { isMaintenanceMode: true });
-
-      expect(mockServer.serverSideEmit).toHaveBeenCalledWith('AppRestart', { isMaintenanceMode: true });
-    });
-  });
-});

server/test/medium/specs/middleware/websocket.adapter.spec.ts

@@ -1,70 +0,0 @@
-import { INestApplication } from '@nestjs/common';
-import { IoAdapter } from '@nestjs/platform-socket.io';
-import { SocketIoAdapter } from 'src/enum';
-import { asPgPoolSsl, createWebSocketAdapter } from 'src/middleware/websocket.adapter';
-import { Mocked, vi } from 'vitest';
-
-describe('asPgPoolSsl', () => {
-  it('should return false for undefined ssl', () => {
-    expect(asPgPoolSsl()).toBe(false);
-  });
-
-  it('should return false for ssl = false', () => {
-    expect(asPgPoolSsl(false)).toBe(false);
-  });
-
-  it('should return false for ssl = "allow"', () => {
-    expect(asPgPoolSsl('allow')).toBe(false);
-  });
-
-  it('should return { rejectUnauthorized: false } for ssl = true', () => {
-    expect(asPgPoolSsl(true)).toEqual({ rejectUnauthorized: false });
-  });
-
-  it('should return { rejectUnauthorized: false } for ssl = "prefer"', () => {
-    expect(asPgPoolSsl('prefer')).toEqual({ rejectUnauthorized: false });
-  });
-
-  it('should return { rejectUnauthorized: false } for ssl = "require"', () => {
-    expect(asPgPoolSsl('require')).toEqual({ rejectUnauthorized: false });
-  });
-
-  it('should return { rejectUnauthorized: true } for ssl = "verify-full"', () => {
-    expect(asPgPoolSsl('verify-full')).toEqual({ rejectUnauthorized: true });
-  });
-
-  it('should pass through object ssl config unchanged', () => {
-    const sslConfig = { ca: 'certificate', rejectUnauthorized: true };
-    expect(asPgPoolSsl(sslConfig)).toBe(sslConfig);
-  });
-});
-
-describe('createWebSocketAdapter', () => {
-  let mockApp: Mocked<INestApplication>;
-
-  beforeEach(() => {
-    vi.clearAllMocks();
-
-    mockApp = {
-      getHttpServer: vi.fn().mockReturnValue({}),
-    } as unknown as Mocked<INestApplication>;
-  });
-
-  describe('BroadcastChannel adapter', () => {
-    it('should create BroadcastChannel adapter when configured', async () => {
-      const adapter = await createWebSocketAdapter(mockApp, SocketIoAdapter.BroadcastChannel);
-
-      expect(adapter).toBeDefined();
-      expect(adapter).toBeInstanceOf(IoAdapter);
-    });
-  });
-
-  describe('Postgres adapter', () => {
-    it('should create Postgres adapter when configured', async () => {
-      const adapter = await createWebSocketAdapter(mockApp, SocketIoAdapter.Postgres);
-
-      expect(adapter).toBeDefined();
-      expect(adapter).toBeInstanceOf(IoAdapter);
-    });
-  });
-});

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

@@ -1,4 +1,4 @@
-import { DatabaseExtension, ImmichEnvironment, ImmichWorker, LogFormat, SocketIoAdapter } from 'src/enum';
+import { DatabaseExtension, ImmichEnvironment, ImmichWorker, LogFormat } from 'src/enum';
 import { ConfigRepository, EnvData } from 'src/repositories/config.repository';
 import { RepositoryInterface } from 'src/types';
 import { Mocked, vitest } from 'vitest';
@@ -99,10 +99,6 @@ const envData: EnvData = {
     },
   },
 
-  socketIo: {
-    adapter: SocketIoAdapter.Postgres,
-  },
-
   noColor: false,
 };
 

server/test/utils.ts

@@ -13,7 +13,6 @@ import postgres from 'postgres';
 import { UploadFieldName } from 'src/dtos/asset-media.dto';
 import { AssetUploadInterceptor } from 'src/middleware/asset-upload.interceptor';
 import { AuthGuard } from 'src/middleware/auth.guard';
-import { FileUploadInterceptor } from 'src/middleware/file-upload.interceptor';
 import { AccessRepository } from 'src/repositories/access.repository';
 import { ActivityRepository } from 'src/repositories/activity.repository';
 import { AlbumUserRepository } from 'src/repositories/album-user.repository';
@@ -120,7 +119,7 @@ export const controllerSetup = async (controller: ClassConstructor<unknown>, pro
       ...providers,
     ],
   })
-    .overrideInterceptor(FileUploadInterceptor)
+    .overrideInterceptor(AssetUploadInterceptor)
     .useValue(memoryFileInterceptor)
     .overrideInterceptor(AssetUploadInterceptor)
     .useValue(noopInterceptor)