feat(server): run microservices in worker thread (#9426)

feat: start microservices in worker thread and add internal microservices for the server

server/src/interfaces/logger.interface.ts

@@ -3,6 +3,7 @@ import { LogLevel } from 'src/entities/system-config.entity';
 export const ILoggerRepository = 'ILoggerRepository';
 
 export interface ILoggerRepository {
+  setAppName(name: string): void;
   setContext(message: string): void;
   setLogLevel(level: LogLevel): void;
 

server/src/main.ts

@@ -4,8 +4,9 @@ import { json } from 'body-parser';
 import cookieParser from 'cookie-parser';
 import { CommandFactory } from 'nest-commander';
 import { existsSync } from 'node:fs';
+import { Worker } from 'node:worker_threads';
 import sirv from 'sirv';
-import { ApiModule, ImmichAdminModule, MicroservicesModule } from 'src/app.module';
+import { ApiModule, ImmichAdminModule } from 'src/app.module';
 import { WEB_ROOT, envName, excludePaths, isDev, serverVersion } from 'src/constants';
 import { LogLevel } from 'src/entities/system-config.entity';
 import { ILoggerRepository } from 'src/interfaces/logger.interface';
@@ -16,21 +17,6 @@ import { useSwagger } from 'src/utils/misc';
 
 const host = process.env.HOST;
 
-async function bootstrapMicroservices() {
-  otelSDK.start();
-
-  const port = Number(process.env.MICROSERVICES_PORT) || 3002;
-  const app = await NestFactory.create(MicroservicesModule, { bufferLogs: true });
-  const logger = await app.resolve(ILoggerRepository);
-  logger.setContext('ImmichMicroservice');
-  app.useLogger(logger);
-  app.useWebSocketAdapter(new WebSocketAdapter(app));
-
-  await (host ? app.listen(port, host) : app.listen(port));
-
-  logger.log(`Immich Microservices is listening on ${await app.getUrl()} [v${serverVersion}] [${envName}] `);
-}
-
 async function bootstrapApi() {
   otelSDK.start();
 
@@ -38,6 +24,7 @@ async function bootstrapApi() {
   const app = await NestFactory.create<NestExpressApplication>(ApiModule, { bufferLogs: true });
   const logger = await app.resolve(ILoggerRepository);
 
+  logger.setAppName('ImmichServer');
   logger.setContext('ImmichServer');
   app.useLogger(logger);
   app.set('trust proxy', ['loopback', 'linklocal', 'uniquelocal']);
@@ -86,15 +73,28 @@ async function bootstrapImmichAdmin() {
   await CommandFactory.run(ImmichAdminModule);
 }
 
+function bootstrapMicroservicesWorker() {
+  const worker = new Worker('./dist/workers/microservices.js');
+  worker.on('exit', (exitCode) => {
+    if (exitCode !== 0) {
+      console.error(`Microservices worker exited with code ${exitCode}`);
+      process.exit(exitCode);
+    }
+  });
+}
+
 function bootstrap() {
   switch (immichApp) {
     case 'immich': {
       process.title = 'immich_server';
+      if (process.env.INTERNAL_MICROSERVICES === 'true') {
+        bootstrapMicroservicesWorker();
+      }
       return bootstrapApi();
     }
     case 'microservices': {
       process.title = 'immich_microservices';
-      return bootstrapMicroservices();
+      return bootstrapMicroservicesWorker();
     }
     case 'immich-admin': {
       process.title = 'immich_admin_cli';

server/src/repositories/logger.repository.ts

@@ -3,6 +3,7 @@ import { isLogLevelEnabled } from '@nestjs/common/services/utils/is-log-level-en
 import { ClsService } from 'nestjs-cls';
 import { LogLevel } from 'src/entities/system-config.entity';
 import { ILoggerRepository } from 'src/interfaces/logger.interface';
+import { LogColor } from 'src/utils/logger-colors';
 
 const LOG_LEVELS = [LogLevel.VERBOSE, LogLevel.DEBUG, LogLevel.LOG, LogLevel.WARN, LogLevel.ERROR, LogLevel.FATAL];
 
@@ -14,6 +15,12 @@ export class LoggerRepository extends ConsoleLogger implements ILoggerRepository
     super(LoggerRepository.name);
   }
 
+  private static appName?: string = undefined;
+
+  setAppName(name: string): void {
+    LoggerRepository.appName = name;
+  }
+
   isLevelEnabled(level: LogLevel) {
     return isLogLevelEnabled(level, LoggerRepository.logLevels);
   }
@@ -30,6 +37,10 @@ export class LoggerRepository extends ConsoleLogger implements ILoggerRepository
       formattedContext += `[${correlationId}] `;
     }
 
+    if (LoggerRepository.appName) {
+      formattedContext = LogColor.blue(`[${LoggerRepository.appName}] `) + formattedContext;
+    }
+
     return formattedContext;
   }
 }

server/src/utils/logger-colors.ts

@@ -0,0 +1,17 @@
+type ColorTextFn = (text: string) => string;
+
+const isColorAllowed = () => !process.env.NO_COLOR;
+const colorIfAllowed = (colorFn: ColorTextFn) => (text: string) => (isColorAllowed() ? colorFn(text) : text);
+
+export const LogColor = {
+  red: colorIfAllowed((text: string) => `\u001B[31m${text}\u001B[39m`),
+  green: colorIfAllowed((text: string) => `\u001B[32m${text}\u001B[39m`),
+  yellow: colorIfAllowed((text: string) => `\u001B[33m${text}\u001B[39m`),
+  blue: colorIfAllowed((text: string) => `\u001B[34m${text}\u001B[39m`),
+  magentaBright: colorIfAllowed((text: string) => `\u001B[95m${text}\u001B[39m`),
+  cyanBright: colorIfAllowed((text: string) => `\u001B[96m${text}\u001B[39m`),
+};
+
+export const LogStyle = {
+  bold: colorIfAllowed((text: string) => `\u001B[1m${text}\u001B[0m`),
+};

server/src/workers/microservices.ts

@@ -0,0 +1,32 @@
+import { NestFactory } from '@nestjs/core';
+import { isMainThread } from 'node:worker_threads';
+import { MicroservicesModule } from 'src/app.module';
+import { envName, serverVersion } from 'src/constants';
+import { ILoggerRepository } from 'src/interfaces/logger.interface';
+import { WebSocketAdapter } from 'src/middleware/websocket.adapter';
+import { otelSDK } from 'src/utils/instrumentation';
+
+const host = process.env.HOST;
+
+export async function bootstrapMicroservices() {
+  otelSDK.start();
+
+  const port = Number(process.env.MICROSERVICES_PORT) || 3002;
+  const app = await NestFactory.create(MicroservicesModule, { bufferLogs: true });
+  const logger = await app.resolve(ILoggerRepository);
+  logger.setAppName('ImmichMicroservices');
+  logger.setContext('ImmichMicroservices');
+  app.useLogger(logger);
+  app.useWebSocketAdapter(new WebSocketAdapter(app));
+
+  await (host ? app.listen(port, host) : app.listen(port));
+
+  logger.log(`Immich Microservices is listening on ${await app.getUrl()} [v${serverVersion}] [${envName}] `);
+}
+
+if (!isMainThread) {
+  bootstrapMicroservices().catch((error) => {
+    console.error(error);
+    process.exit(1);
+  });
+}

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

@@ -5,6 +5,7 @@ export const newLoggerRepositoryMock = (): Mocked<ILoggerRepository> => {
   return {
     setLogLevel: vitest.fn(),
     setContext: vitest.fn(),
+    setAppName: vitest.fn(),
 
     verbose: vitest.fn(),
     debug: vitest.fn(),