Implemented websocket on the server

server/docker-compose.yml

@@ -12,7 +12,8 @@ services:
     command: npm run start:dev
     ports:
       - "3000:3000"
-      - "3001:3001"
+    expose: 
+      - "3000"
     volumes:
       - .:/usr/src/app
       - ${UPLOAD_LOCATION}:/usr/src/app/upload

server/settings/nginx-conf/nginx.conf

@@ -1,4 +1,13 @@
+
+map $http_upgrade $connection_upgrade {
+  default upgrade;
+  '' close;
+}
+
 server {
+  
+
+
   client_max_body_size 50000M;
 
   listen 80;
@@ -10,11 +19,15 @@ server {
     proxy_buffers 64 4k;
     proxy_force_ranges on;
 
+    proxy_http_version 1.1;
     proxy_set_header Host $host;
     proxy_set_header X-Real-IP $remote_addr;
     proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
     proxy_set_header X-Forwarded-Proto $scheme;
+    proxy_set_header Upgrade $http_upgrade;
+    proxy_set_header Connection "upgrade";
+    proxy_set_header Host $host;
 
     proxy_pass http://immich_server:3000;
   }
-}
+}

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

@@ -1,17 +1,47 @@
-import { OnGatewayConnection, WebSocketGateway, WebSocketServer } from '@nestjs/websockets';
+import { OnGatewayConnection, OnGatewayDisconnect, WebSocketGateway, WebSocketServer } from '@nestjs/websockets';
 import { CommunicationService } from './communication.service';
 import { Socket, Server } from 'socket.io';
+import { ImmichJwtService } from '../../modules/immich-jwt/immich-jwt.service';
+import { Logger } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+import { UserEntity } from '../user/entities/user.entity';
+import { Repository } from 'typeorm';
 
 @WebSocketGateway()
-export class CommunicationGateway implements OnGatewayConnection {
-  constructor(private readonly communicationService: CommunicationService) {}
-  @WebSocketServer() server: Server;
-  handleConnection(client: Socket, ...args: any[]) {
-    console.log('Socket IO - New connetrion ', client.id);
+export class CommunicationGateway implements OnGatewayConnection, OnGatewayDisconnect {
+  constructor(
+    private immichJwtService: ImmichJwtService,
 
-    this.server.emit(
-      'on_connected',
-      `Connection Success - hi ${client.id} - you are connecting to chat room ${client.nsp.name}`,
-    );
+    @InjectRepository(UserEntity)
+    private userRepository: Repository<UserEntity>,
+  ) {}
+
+  @WebSocketServer() server: Server;
+
+  handleDisconnect(client: Socket) {
+    client.leave(client.nsp.name);
+
+    console.log('Client left room ', client.rooms);
+  }
+
+  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);
+
+    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);
   }
 }

server/src/api-v1/communication/communication.module.ts

@@ -1,8 +1,16 @@
 import { Module } from '@nestjs/common';
 import { CommunicationService } from './communication.service';
 import { CommunicationGateway } from './communication.gateway';
+import { ImmichJwtModule } from '../../modules/immich-jwt/immich-jwt.module';
+import { ImmichJwtService } from '../../modules/immich-jwt/immich-jwt.service';
+import { JwtModule } from '@nestjs/jwt';
+import { jwtConfig } from '../../config/jwt.config';
+import { TypeOrmModule } from '@nestjs/typeorm';
+import { UserEntity } from '../user/entities/user.entity';
 
 @Module({
-  providers: [CommunicationGateway, CommunicationService]
+  imports: [TypeOrmModule.forFeature([UserEntity]), ImmichJwtModule, JwtModule.register(jwtConfig)],
+  providers: [CommunicationGateway, CommunicationService, ImmichJwtService],
+  exports: [CommunicationGateway],
 })
 export class CommunicationModule {}

server/src/api-v1/user/user.service.ts

@@ -16,16 +16,7 @@ export class UserService {
     return 'This action adds a new user';
   }
 
-  async findAll() {
-    try {
-      return 'welcome';
-      // return await this.userRepository.find();
-      // return await this.userRepository.query('select * from users');
-    } catch (e) {
-      console.log(e);
-    }
-    // return 'helloworld';
-  }
+  async findAll() {}
 
   findOne(id: number) {
     return `This action returns a #${id} user`;

server/src/modules/image-optimize/image-optimize.module.ts

@@ -5,11 +5,16 @@ import { join } from 'path';
 import { AssetModule } from '../../api-v1/asset/asset.module';
 import { AssetService } from '../../api-v1/asset/asset.service';
 import { AssetEntity } from '../../api-v1/asset/entities/asset.entity';
+import { CommunicationGateway } from '../../api-v1/communication/communication.gateway';
+import { CommunicationModule } from '../../api-v1/communication/communication.module';
+import { UserEntity } from '../../api-v1/user/entities/user.entity';
+import { ImmichJwtModule } from '../immich-jwt/immich-jwt.module';
 import { ImageOptimizeProcessor } from './image-optimize.processor';
 import { AssetOptimizeService } from './image-optimize.service';
 
 @Module({
   imports: [
+    CommunicationModule,
     BullModule.registerQueue({
       name: 'optimize',
       defaultJobOptions: {

server/src/modules/image-optimize/image-optimize.processor.ts

@@ -8,14 +8,16 @@ import { existsSync, mkdirSync, readFile } from 'fs';
 import { ConfigService } from '@nestjs/config';
 import ffmpeg from 'fluent-ffmpeg';
 import { APP_UPLOAD_LOCATION } from '../../constants/upload_location.constant';
+import { WebSocketServer } from '@nestjs/websockets';
+import { Socket, Server as SocketIoServer } from 'socket.io';
+import { CommunicationGateway } from '../../api-v1/communication/communication.gateway';
 
 @Processor('optimize')
 export class ImageOptimizeProcessor {
   constructor(
+    private wsCommunicateionGateway: CommunicationGateway,
     @InjectRepository(AssetEntity)
     private assetRepository: Repository<AssetEntity>,
-
-    private configService: ConfigService,
   ) {}
 
   @Process('resize-image')
@@ -67,6 +69,9 @@ export class ImageOptimizeProcessor {
             }
 
             await this.assetRepository.update(savedAsset, { resizePath: resizePath });
+            this.wsCommunicateionGateway.server
+              .to(savedAsset.userId)
+              .emit('on_upload_success', { assetId: savedAsset.id });
           });
       }
     });

server/src/modules/immich-jwt/immich-jwt.service.ts

@@ -1,6 +1,7 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, Logger } from '@nestjs/common';
 import { JwtService } from '@nestjs/jwt';
 import { JwtPayloadDto } from '../../api-v1/auth/dto/jwt-payload.dto';
+import { jwtSecret } from '../../constants/jwt.constant';
 
 @Injectable()
 export class ImmichJwtService {
@@ -11,4 +12,20 @@ export class ImmichJwtService {
       ...payload,
     });
   }
+
+  public async validateToken(accessToken: string) {
+    try {
+      const payload = await this.jwtService.verify(accessToken, { secret: jwtSecret });
+      return {
+        userId: payload['userId'],
+        status: true,
+      };
+    } catch (e) {
+      Logger.error('Error validating token from websocket request', 'ValidateWebsocketToken');
+      return {
+        userId: null,
+        status: false,
+      };
+    }
+  }
 }