WIP refactor container and queuing system (#206)

* refactor microservices to machine-learning

* Update tGithub issue template with correct task syntax

* Added microservices container

* Communicate between service based on queue system

* added dependency

* Fixed problem with having to import BullQueue into the individual service

* Added todo

* refactor server into monorepo with microservices

* refactor database and entity to library

* added simple migration

* Move migrations and database config to library

* Migration works in library

* Cosmetic change in logging message

* added user dto

* Fixed issue with testing not able to find the shared library

* Clean up library mapping path

* Added webp generator to microservices

* Update Github Action build latest

* Fixed issue NPM cannot install due to conflict witl Bull Queue

* format project with prettier

* Modified docker-compose file

* Add GH Action for Staging build:

* Fixed GH action job name

* Modified GH Action to only build & push latest when pushing to main

* Added Test 2e2 Github Action

* Added Test 2e2 Github Action

* Implemented microservice to extract exif

* Added cronjob to scan and generate webp thumbnail  at midnight

* Refactor to ireduce hit time to database when running microservices

* Added error handling to asset services that handle read file from disk

* Added video transcoding queue to process one video at a time

* Fixed loading spinner on web while loading covering the info panel

* Add mechanism to show new release announcement to web and mobile app (#209)

* Added changelog page

* Fixed issues based on PR comments

* Fixed issue with video transcoding run on the server

* Change entry point content for backward combatibility when starting up server

* Added announcement box

* Added error handling to failed silently when the app version checking is not able to make the request to GITHUB

* Added new version announcement overlay

* Update message

* Added messages

* Added logic to check and show announcement

* Add method to handle saving new version

* Added button to dimiss the acknowledge message

* Up version for deployment to the app store

server/apps/microservices/src/processors/asset-uploaded.processor.ts

@@ -0,0 +1,67 @@
+import { InjectQueue, OnQueueActive, OnQueueCompleted, OnQueueWaiting, Process, Processor } from '@nestjs/bull';
+import { Job, Queue } from 'bull';
+import { AssetEntity, AssetType } from '@app/database/entities/asset.entity';
+import { InjectRepository } from '@nestjs/typeorm';
+import { Repository } from 'typeorm';
+import { randomUUID } from 'crypto';
+
+@Processor('asset-uploaded-queue')
+export class AssetUploadedProcessor {
+  constructor(
+    @InjectQueue('thumbnail-generator-queue')
+    private thumbnailGeneratorQueue: Queue,
+
+    @InjectQueue('metadata-extraction-queue')
+    private metadataExtractionQueue: Queue,
+
+    @InjectQueue('video-conversion-queue')
+    private videoConversionQueue: Queue,
+
+    @InjectRepository(AssetEntity)
+    private assetRepository: Repository<AssetEntity>,
+  ) {}
+
+  /**
+   * Post processing uploaded asset to perform the following function if missing
+   * 1. Generate JPEG Thumbnail
+   * 2. Generate Webp Thumbnail <-> if JPEG thumbnail exist
+   * 3. EXIF extractor
+   * 4. Reverse Geocoding
+   *
+   * @param job asset-uploaded
+   */
+  @Process('asset-uploaded')
+  async processUploadedVideo(job: Job) {
+    const {
+      asset,
+      fileName,
+      fileSize,
+      hasThumbnail,
+    }: { asset: AssetEntity; fileName: string; fileSize: number; hasThumbnail: boolean } = job.data;
+
+    if (hasThumbnail) {
+      // The jobs below depends on the existence of jpeg thumbnail
+      await this.thumbnailGeneratorQueue.add('generate-webp-thumbnail', { asset }, { jobId: randomUUID() });
+      await this.metadataExtractionQueue.add('tag-image', { asset }, { jobId: randomUUID() });
+      await this.metadataExtractionQueue.add('detect-object', { asset }, { jobId: randomUUID() });
+    } else {
+      // Generate Thumbnail -> Then generate webp, tag image and detect object
+    }
+
+    // Video Conversion
+    if (asset.type == AssetType.VIDEO) {
+      await this.videoConversionQueue.add('mp4-conversion', { asset }, { jobId: randomUUID() });
+    } else {
+      // Extract Metadata/Exif for Images - Currently the library cannot extract EXIF for video yet
+      await this.metadataExtractionQueue.add(
+        'exif-extraction',
+        {
+          asset,
+          fileName,
+          fileSize,
+        },
+        { jobId: randomUUID() },
+      );
+    }
+  }
+}

server/apps/microservices/src/processors/metadata-extraction.processor.ts

@@ -0,0 +1,131 @@
+import { Process, Processor } from '@nestjs/bull';
+import { Job } from 'bull';
+import { AssetEntity } from '@app/database/entities/asset.entity';
+import { Repository } from 'typeorm/repository/Repository';
+import { InjectRepository } from '@nestjs/typeorm';
+import { ExifEntity } from '@app/database/entities/exif.entity';
+import exifr from 'exifr';
+import mapboxGeocoding, { GeocodeService } from '@mapbox/mapbox-sdk/services/geocoding';
+import { MapiResponse } from '@mapbox/mapbox-sdk/lib/classes/mapi-response';
+import { readFile } from 'fs/promises';
+import { Logger } from '@nestjs/common';
+import axios from 'axios';
+import { SmartInfoEntity } from '@app/database/entities/smart-info.entity';
+
+@Processor('metadata-extraction-queue')
+export class MetadataExtractionProcessor {
+  private geocodingClient: GeocodeService;
+
+  constructor(
+    @InjectRepository(AssetEntity)
+    private assetRepository: Repository<AssetEntity>,
+
+    @InjectRepository(ExifEntity)
+    private exifRepository: Repository<ExifEntity>,
+
+    @InjectRepository(SmartInfoEntity)
+    private smartInfoRepository: Repository<SmartInfoEntity>,
+  ) {
+    if (process.env.ENABLE_MAPBOX) {
+      this.geocodingClient = mapboxGeocoding({
+        accessToken: process.env.MAPBOX_KEY,
+      });
+    }
+  }
+
+  @Process('exif-extraction')
+  async extractExifInfo(job: Job) {
+    try {
+      const { asset, fileName, fileSize }: { asset: AssetEntity; fileName: string; fileSize: number } = job.data;
+
+      const fileBuffer = await readFile(asset.originalPath);
+
+      const exifData = await exifr.parse(fileBuffer);
+
+      const newExif = new ExifEntity();
+      newExif.assetId = asset.id;
+      newExif.make = exifData['Make'] || null;
+      newExif.model = exifData['Model'] || null;
+      newExif.imageName = fileName || null;
+      newExif.exifImageHeight = exifData['ExifImageHeight'] || null;
+      newExif.exifImageWidth = exifData['ExifImageWidth'] || null;
+      newExif.fileSizeInByte = fileSize || null;
+      newExif.orientation = exifData['Orientation'] || null;
+      newExif.dateTimeOriginal = exifData['DateTimeOriginal'] || null;
+      newExif.modifyDate = exifData['ModifyDate'] || null;
+      newExif.lensModel = exifData['LensModel'] || null;
+      newExif.fNumber = exifData['FNumber'] || null;
+      newExif.focalLength = exifData['FocalLength'] || null;
+      newExif.iso = exifData['ISO'] || null;
+      newExif.exposureTime = exifData['ExposureTime'] || null;
+      newExif.latitude = exifData['latitude'] || null;
+      newExif.longitude = exifData['longitude'] || null;
+
+      // Reverse GeoCoding
+      if (process.env.ENABLE_MAPBOX && exifData['longitude'] && exifData['latitude']) {
+        const geoCodeInfo: MapiResponse = await this.geocodingClient
+          .reverseGeocode({
+            query: [exifData['longitude'], exifData['latitude']],
+            types: ['country', 'region', 'place'],
+          })
+          .send();
+
+        const res: [] = geoCodeInfo.body['features'];
+
+        const city = res.filter((geoInfo) => geoInfo['place_type'][0] == 'place')[0]['text'];
+        const state = res.filter((geoInfo) => geoInfo['place_type'][0] == 'region')[0]['text'];
+        const country = res.filter((geoInfo) => geoInfo['place_type'][0] == 'country')[0]['text'];
+
+        newExif.city = city || null;
+        newExif.state = state || null;
+        newExif.country = country || null;
+      }
+
+      await this.exifRepository.save(newExif);
+    } catch (e) {
+      Logger.error(`Error extracting EXIF ${e.toString()}`, 'extractExif');
+    }
+  }
+
+  @Process({ name: 'tag-image', concurrency: 2 })
+  async tagImage(job: Job) {
+    const { asset }: { asset: AssetEntity } = job.data;
+
+    const res = await axios.post('http://immich-machine-learning:3001/image-classifier/tag-image', {
+      thumbnailPath: asset.resizePath,
+    });
+
+    if (res.status == 201 && res.data.length > 0) {
+      const smartInfo = new SmartInfoEntity();
+      smartInfo.assetId = asset.id;
+      smartInfo.tags = [...res.data];
+
+      await this.smartInfoRepository.upsert(smartInfo, {
+        conflictPaths: ['assetId'],
+      });
+    }
+  }
+
+  @Process({ name: 'detect-object', concurrency: 2 })
+  async detectObject(job: Job) {
+    try {
+      const { asset }: { asset: AssetEntity } = job.data;
+
+      const res = await axios.post('http://immich-machine-learning:3001/object-detection/detect-object', {
+        thumbnailPath: asset.resizePath,
+      });
+
+      if (res.status == 201 && res.data.length > 0) {
+        const smartInfo = new SmartInfoEntity();
+        smartInfo.assetId = asset.id;
+        smartInfo.objects = [...res.data];
+
+        await this.smartInfoRepository.upsert(smartInfo, {
+          conflictPaths: ['assetId'],
+        });
+      }
+    } catch (error) {
+      Logger.error(`Failed to trigger object detection pipe line ${error.toString()}`);
+    }
+  }
+}

server/apps/microservices/src/processors/thumbnail.processor.ts

@@ -0,0 +1,37 @@
+import { Process, Processor } from '@nestjs/bull';
+import { Job } from 'bull';
+import { AssetEntity } from '@app/database/entities/asset.entity';
+import { Repository } from 'typeorm/repository/Repository';
+import { InjectRepository } from '@nestjs/typeorm';
+import sharp from 'sharp';
+
+@Processor('thumbnail-generator-queue')
+export class ThumbnailGeneratorProcessor {
+  constructor(
+    @InjectRepository(AssetEntity)
+    private assetRepository: Repository<AssetEntity>,
+  ) {}
+
+  @Process('generate-jpeg-thumbnail')
+  async generateJPEGThumbnail(job: Job) {
+    const { asset }: { asset: AssetEntity } = job.data;
+
+    console.log(asset);
+  }
+
+  @Process({ name: 'generate-webp-thumbnail', concurrency: 2 })
+  async generateWepbThumbnail(job: Job) {
+    const { asset }: { asset: AssetEntity } = job.data;
+
+    const webpPath = asset.resizePath.replace('jpeg', 'webp');
+
+    sharp(asset.resizePath)
+      .resize(250)
+      .webp()
+      .toFile(webpPath, (err, info) => {
+        if (!err) {
+          this.assetRepository.update({ id: asset.id }, { webpPath: webpPath });
+        }
+      });
+  }
+}

server/apps/microservices/src/processors/video-transcode.processor.ts

@@ -0,0 +1,59 @@
+import { Process, Processor } from '@nestjs/bull';
+import { Logger } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+import { Job } from 'bull';
+import ffmpeg from 'fluent-ffmpeg';
+import { existsSync, mkdirSync } from 'fs';
+import { Repository } from 'typeorm';
+import { AssetEntity } from '../../../../libs/database/src/entities/asset.entity';
+import { APP_UPLOAD_LOCATION } from '../../../immich/src/constants/upload_location.constant';
+
+@Processor('video-conversion-queue')
+export class VideoTranscodeProcessor {
+  constructor(
+    @InjectRepository(AssetEntity)
+    private assetRepository: Repository<AssetEntity>,
+  ) {}
+
+  @Process({ name: 'mp4-conversion', concurrency: 1 })
+  async mp4Conversion(job: Job) {
+    const { asset }: { asset: AssetEntity } = job.data;
+
+    if (asset.mimeType != 'video/mp4') {
+      const basePath = APP_UPLOAD_LOCATION;
+      const encodedVideoPath = `${basePath}/${asset.userId}/encoded-video`;
+
+      if (!existsSync(encodedVideoPath)) {
+        mkdirSync(encodedVideoPath, { recursive: true });
+      }
+
+      const savedEncodedPath = encodedVideoPath + '/' + asset.id + '.mp4';
+
+      if (asset.encodedVideoPath == '' || !asset.encodedVideoPath) {
+        // Put the processing into its own async function to prevent the job exist right away
+        await this.runFFMPEGPipeLine(asset, savedEncodedPath);
+      }
+    }
+  }
+
+  async runFFMPEGPipeLine(asset: AssetEntity, savedEncodedPath: string): Promise<void> {
+    return new Promise((resolve, reject) => {
+      ffmpeg(asset.originalPath)
+        .outputOptions(['-crf 23', '-preset ultrafast', '-vcodec libx264', '-acodec mp3', '-vf scale=1280:-2'])
+        .output(savedEncodedPath)
+        .on('start', () => {
+          Logger.log('Start Converting', 'mp4Conversion');
+        })
+        .on('error', (error, b, c) => {
+          Logger.error(`Cannot Convert Video ${error}`, 'mp4Conversion');
+          reject();
+        })
+        .on('end', async () => {
+          Logger.log(`Converting Success ${asset.id}`, 'mp4Conversion');
+          await this.assetRepository.update({ id: asset.id }, { encodedVideoPath: savedEncodedPath });
+          resolve();
+        })
+        .run();
+    });
+  }
+}