Feature - Add upload functionality on Web (#231)

* Added file selector

* Extract metadata to upload files to the web

* Added request for uploading

* Generate jpeg/Webp thumbnail for asset uploaded without thumbnail data

* Added generating thumbnail for video and WebSocket broadcast after thumbnail is generated

* Added video length extraction

* Added Uploading Panel

* Added upload progress store and styling the uploaded asset

* Added condition to only show upload panel when there is upload in progress

* Remove asset from the upload list after successfully uploading

* Added WebSocket to listen to upload event on the web

* Added mechanism to check for existing assets before uploading on the web

* Added test workflow

* Update readme

server/apps/immich/src/api-v1/asset/asset.controller.ts

@@ -15,6 +15,7 @@ import {
   Delete,
   Logger,
   Patch,
+  HttpCode,
 } from '@nestjs/common';
 import { JwtAuthGuard } from '../../modules/immich-jwt/guards/jwt-auth.guard';
 import { AssetService } from './asset.service';
@@ -76,6 +77,10 @@ export class AssetController {
             { asset: assetWithThumbnail, fileName: file.originalname, fileSize: file.size, hasThumbnail: true },
             { jobId: savedAsset.id },
           );
+
+          this.wsCommunicateionGateway.server
+            .to(savedAsset.userId)
+            .emit('on_upload_success', JSON.stringify(assetWithThumbnail));
         } else {
           await this.assetUploadedQueue.add(
             'asset-uploaded',
@@ -83,8 +88,6 @@ export class AssetController {
             { jobId: savedAsset.id },
           );
         }
-
-        this.wsCommunicateionGateway.server.to(savedAsset.userId).emit('on_upload_success', JSON.stringify(savedAsset));
       } catch (e) {
         Logger.error(`Error receiving upload file ${e}`);
       }
@@ -171,4 +174,20 @@ export class AssetController {
 
     return result;
   }
+
+  /**
+   * Check duplicated asset before uploading - for Web upload used
+   */
+  @Post('/check')
+  @HttpCode(200)
+  async checkDuplicateAsset(
+    @GetAuthUser() authUser: AuthUserDto,
+    @Body(ValidationPipe) { deviceAssetId }: { deviceAssetId: string },
+  ) {
+    const res = await this.assetService.checkDuplicatedAsset(authUser, deviceAssetId);
+
+    return {
+      isExist: res,
+    };
+  }
 }

server/apps/immich/src/api-v1/asset/asset.module.ts

@@ -24,6 +24,6 @@ import { CommunicationModule } from '../communication/communication.module';
   ],
   controllers: [AssetController],
   providers: [AssetService, BackgroundTaskService],
-  exports: [],
+  exports: [AssetService],
 })
 export class AssetModule {}

server/apps/immich/src/api-v1/asset/asset.service.ts

@@ -1,6 +1,6 @@
 import { BadRequestException, Injectable, InternalServerErrorException, Logger, StreamableFile } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
-import { Repository } from 'typeorm';
+import { IsNull, Not, Repository } from 'typeorm';
 import { AuthUserDto } from '../../decorators/auth-user.decorator';
 import { CreateAssetDto } from './dto/create-asset.dto';
 import { AssetEntity, AssetType } from '@app/database/entities/asset.entity';
@@ -72,6 +72,7 @@ export class AssetService {
       return await this.assetRepository.find({
         where: {
           userId: authUser.id,
+          resizePath: Not(IsNull()),
         },
         relations: ['exifInfo'],
         order: {
@@ -381,4 +382,15 @@ export class AssetService {
       [authUser.id],
     );
   }
+
+  async checkDuplicatedAsset(authUser: AuthUserDto, deviceAssetId: string) {
+    const res = await this.assetRepository.findOne({
+      where: {
+        deviceAssetId,
+        userId: authUser.id,
+      },
+    });
+
+    return res ? true : false;
+  }
 }

server/apps/immich/src/api-v1/communication/communication.gateway.ts

@@ -6,8 +6,9 @@ import { Logger } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
 import { UserEntity } from '@app/database/entities/user.entity';
 import { Repository } from 'typeorm';
+import { query } from 'express';
 
-@WebSocketGateway()
+@WebSocketGateway({ cors: true })
 export class CommunicationGateway implements OnGatewayConnection, OnGatewayDisconnect {
   constructor(
     private immichJwtService: ImmichJwtService,
@@ -21,27 +22,33 @@ export class CommunicationGateway implements OnGatewayConnection, OnGatewayDisco
   handleDisconnect(client: Socket) {
     client.leave(client.nsp.name);
 
-    Logger.log(`Client ${client.id} disconnected`);
+    Logger.log(`Client ${client.id} disconnected from Websocket`, 'WebsocketConnectionEvent');
   }
 
   async handleConnection(client: Socket, ...args: any[]) {
-    Logger.log(`New websocket connection: ${client.id}`, 'NewWebSocketConnection');
-    const accessToken = client.handshake.headers.authorization.split(' ')[1];
-    const res = await this.immichJwtService.validateToken(accessToken);
+    try {
+      Logger.log(`New websocket connection: ${client.id}`, 'WebsocketConnectionEvent');
 
-    if (!res.status) {
-      client.emit('error', 'unauthorized');
-      client.disconnect();
-      return;
+      const accessToken = client.handshake.headers.authorization.split(' ')[1];
+
+      const res = await this.immichJwtService.validateToken(accessToken);
+
+      if (!res.status) {
+        client.emit('error', 'unauthorized');
+        client.disconnect();
+        return;
+      }
+
+      const user = await this.userRepository.findOne({ where: { id: res.userId } });
+      if (!user) {
+        client.emit('error', 'unauthorized');
+        client.disconnect();
+        return;
+      }
+
+      client.join(user.id);
+    } catch (e) {
+      // Logger.error(`Error establish websocket conneciton ${e}`, 'HandleWebscoketConnection');
     }
-
-    const user = await this.userRepository.findOne({ where: { id: res.userId } });
-    if (!user) {
-      client.emit('error', 'unauthorized');
-      client.disconnect();
-      return;
-    }
-
-    client.join(user.id);
   }
 }

server/apps/microservices/src/main.ts

@@ -1,10 +1,13 @@
 import { Logger } from '@nestjs/common';
 import { NestFactory } from '@nestjs/core';
+import { RedisIoAdapter } from '../../immich/src/middlewares/redis-io.adapter.middleware';
 import { MicroservicesModule } from './microservices.module';
 
 async function bootstrap() {
   const app = await NestFactory.create(MicroservicesModule);
 
+  app.useWebSocketAdapter(new RedisIoAdapter(app));
+
   await app.listen(3000, () => {
     if (process.env.NODE_ENV == 'development') {
       Logger.log('Running Immich Microservices in DEVELOPMENT environment', 'ImmichMicroservice');

server/apps/microservices/src/microservices.module.ts

@@ -11,6 +11,9 @@ import { AssetUploadedProcessor } from './processors/asset-uploaded.processor';
 import { ThumbnailGeneratorProcessor } from './processors/thumbnail.processor';
 import { MetadataExtractionProcessor } from './processors/metadata-extraction.processor';
 import { VideoTranscodeProcessor } from './processors/video-transcode.processor';
+import { AssetModule } from '../../immich/src/api-v1/asset/asset.module';
+import { CommunicationGateway } from '../../immich/src/api-v1/communication/communication.gateway';
+import { CommunicationModule } from '../../immich/src/api-v1/communication/communication.module';
 
 @Module({
   imports: [
@@ -56,6 +59,7 @@ import { VideoTranscodeProcessor } from './processors/video-transcode.processor'
         removeOnFail: false,
       },
     }),
+    CommunicationModule,
   ],
   controllers: [],
   providers: [

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

@@ -46,6 +46,7 @@ export class AssetUploadedProcessor {
       await this.metadataExtractionQueue.add('detect-object', { asset }, { jobId: randomUUID() });
     } else {
       // Generate Thumbnail -> Then generate webp, tag image and detect object
+      await this.thumbnailGeneratorQueue.add('generate-jpeg-thumbnail', { asset }, { jobId: randomUUID() });
     }
 
     // Video Conversion
@@ -63,5 +64,10 @@ export class AssetUploadedProcessor {
         { jobId: randomUUID() },
       );
     }
+
+    // Extract video duration if uploaded from the web
+    if (asset.type == AssetType.VIDEO && asset.duration == '0:00:00.000000') {
+      await this.metadataExtractionQueue.add('extract-video-length', { asset }, { jobId: randomUUID() });
+    }
   }
 }

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

@@ -12,6 +12,8 @@ import { Logger } from '@nestjs/common';
 import axios from 'axios';
 import { SmartInfoEntity } from '@app/database/entities/smart-info.entity';
 import { ConfigService } from '@nestjs/config';
+import ffmpeg from 'fluent-ffmpeg';
+// import moment from 'moment';
 
 @Processor('metadata-extraction-queue')
 export class MetadataExtractionProcessor {
@@ -129,4 +131,27 @@ export class MetadataExtractionProcessor {
       Logger.error(`Failed to trigger object detection pipe line ${error.toString()}`);
     }
   }
+
+  @Process({ name: 'extract-video-length', concurrency: 2 })
+  async extractVideoLength(job: Job) {
+    const { asset }: { asset: AssetEntity } = job.data;
+
+    ffmpeg.ffprobe(asset.originalPath, async (err, data) => {
+      if (!err) {
+        if (data.format.duration) {
+          const videoDurationInSecond = parseInt(data.format.duration.toString(), 0);
+
+          const hours = Math.floor(videoDurationInSecond / 3600);
+          const minutes = Math.floor((videoDurationInSecond - hours * 3600) / 60);
+          const seconds = videoDurationInSecond - hours * 3600 - minutes * 60;
+
+          const durationString = `${hours}:${minutes < 10 ? '0' + minutes.toString() : minutes}:${
+            seconds < 10 ? '0' + seconds.toString() : seconds
+          }.000000`;
+
+          await this.assetRepository.update({ id: asset.id }, { duration: durationString });
+        }
+      }
+    });
+  }
 }

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

@@ -1,22 +1,82 @@
-import { Process, Processor } from '@nestjs/bull';
-import { Job } from 'bull';
-import { AssetEntity } from '@app/database/entities/asset.entity';
+import { InjectQueue, Process, Processor } from '@nestjs/bull';
+import { Job, Queue } from 'bull';
+import { AssetEntity, AssetType } from '@app/database/entities/asset.entity';
 import { Repository } from 'typeorm/repository/Repository';
 import { InjectRepository } from '@nestjs/typeorm';
 import sharp from 'sharp';
+import { existsSync, mkdirSync } from 'node:fs';
+import { randomUUID } from 'node:crypto';
+import { CommunicationGateway } from '../../../immich/src/api-v1/communication/communication.gateway';
+import ffmpeg from 'fluent-ffmpeg';
+import { Logger } from '@nestjs/common';
 
 @Processor('thumbnail-generator-queue')
 export class ThumbnailGeneratorProcessor {
   constructor(
     @InjectRepository(AssetEntity)
     private assetRepository: Repository<AssetEntity>,
+
+    @InjectQueue('thumbnail-generator-queue')
+    private thumbnailGeneratorQueue: Queue,
+
+    private wsCommunicateionGateway: CommunicationGateway,
   ) {}
 
   @Process('generate-jpeg-thumbnail')
   async generateJPEGThumbnail(job: Job) {
     const { asset }: { asset: AssetEntity } = job.data;
 
-    console.log(asset);
+    const resizePath = `upload/${asset.userId}/thumb/${asset.deviceId}/`;
+
+    if (!existsSync(resizePath)) {
+      mkdirSync(resizePath, { recursive: true });
+    }
+
+    const temp = asset.originalPath.split('/');
+    const originalFilename = temp[temp.length - 1].split('.')[0];
+    const jpegThumbnailPath = resizePath + originalFilename + '.jpeg';
+
+    if (asset.type == AssetType.IMAGE) {
+      sharp(asset.originalPath)
+        .resize(1440, 2560, { fit: 'inside' })
+        .jpeg()
+        .toFile(jpegThumbnailPath, async (err, info) => {
+          if (!err) {
+            await this.assetRepository.update({ id: asset.id }, { resizePath: jpegThumbnailPath });
+
+            // Update resize path to send to generate webp queue
+            asset.resizePath = jpegThumbnailPath;
+
+            await this.thumbnailGeneratorQueue.add('generate-webp-thumbnail', { asset }, { jobId: randomUUID() });
+            this.wsCommunicateionGateway.server.to(asset.userId).emit('on_upload_success', JSON.stringify(asset));
+          }
+        });
+    }
+
+    if (asset.type == AssetType.VIDEO) {
+      ffmpeg(asset.originalPath)
+        .outputOptions(['-ss 00:00:01.000', '-frames:v 1'])
+        .output(jpegThumbnailPath)
+        .on('start', () => {
+          Logger.log('Start Generating Video Thumbnail', 'generateJPEGThumbnail');
+        })
+        .on('error', (error, b, c) => {
+          Logger.error(`Cannot Generate Video Thumbnail ${error}`, 'generateJPEGThumbnail');
+          // reject();
+        })
+        .on('end', async () => {
+          Logger.log(`Generating Video Thumbnail Success ${asset.id}`, 'generateJPEGThumbnail');
+          await this.assetRepository.update({ id: asset.id }, { resizePath: jpegThumbnailPath });
+
+          // Update resize path to send to generate webp queue
+          asset.resizePath = jpegThumbnailPath;
+
+          await this.thumbnailGeneratorQueue.add('generate-webp-thumbnail', { asset }, { jobId: randomUUID() });
+
+          this.wsCommunicateionGateway.server.to(asset.userId).emit('on_upload_success', JSON.stringify(asset));
+        })
+        .run();
+    }
   }
 
   @Process({ name: 'generate-webp-thumbnail', concurrency: 2 })

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

@@ -42,7 +42,7 @@ export class VideoTranscodeProcessor {
         .outputOptions(['-crf 23', '-preset ultrafast', '-vcodec libx264', '-acodec mp3', '-vf scale=1280:-2'])
         .output(savedEncodedPath)
         .on('start', () => {
-          Logger.log('Start Converting', 'mp4Conversion');
+          Logger.log('Start Converting Video', 'mp4Conversion');
         })
         .on('error', (error, b, c) => {
           Logger.error(`Cannot Convert Video ${error}`, 'mp4Conversion');