websocket upload notification - closed #24 (#25)

* Render when a new asset is uploaded from WebSocket notification
* Update Readme

Makefile

@@ -2,4 +2,7 @@ dev:
 	docker-compose -f ./server/docker-compose.yml up
 
 dev-update:
-	docker-compose -f ./server/docker-compose.yml up --build -V
+	docker-compose -f ./server/docker-compose.yml up --build -V 
+
+dev-scale:
+	docker-compose -f ./server/docker-compose.yml up --build -V  --scale immich_server=3 --remove-orphans 

README.md

@@ -28,17 +28,13 @@ This project is under heavy development, there will be continous functions, feat
 
 # Features
 
-[x] Upload assets(videos/images)
-
-[x] View assets
-
-[x] Quick navigation with drag scroll bar
-
-[x] Auto Backup
-
-[x] Support HEIC/HEIF Backup
-
-[x] Extract and display EXIF info
+- Upload assets(videos/images).
+- View assets.
+- Quick navigation with drag scroll bar.
+- Auto Backup.
+- Support HEIC/HEIF Backup.
+- Extract and display EXIF info.
+- Real-time render from multi-device upload event.
 
 # Development
 

mobile/lib/main.dart

@@ -2,9 +2,11 @@ import 'package:flutter/material.dart';
 import 'package:flutter/services.dart';
 import 'package:hive_flutter/hive_flutter.dart';
 import 'package:hooks_riverpod/hooks_riverpod.dart';
+import 'package:immich_mobile/modules/home/providers/asset.provider.dart';
 import 'package:immich_mobile/routing/router.dart';
 import 'package:immich_mobile/shared/providers/app_state.provider.dart';
 import 'package:immich_mobile/shared/providers/backup.provider.dart';
+import 'package:immich_mobile/shared/providers/websocket.provider.dart';
 import 'constants/hive_box.dart';
 import 'package:google_fonts/google_fonts.dart';
 
@@ -36,20 +38,23 @@ class _ImmichAppState extends ConsumerState<ImmichApp> with WidgetsBindingObserv
     switch (state) {
       case AppLifecycleState.resumed:
         debugPrint("[APP STATE] resumed");
-        ref.read(appStateProvider.notifier).state = AppStateEnum.resumed;
-        ref.read(backupProvider.notifier).resumeBackup();
+        ref.watch(appStateProvider.notifier).state = AppStateEnum.resumed;
+        ref.watch(backupProvider.notifier).resumeBackup();
+        ref.watch(websocketProvider.notifier).connect();
+        ref.watch(assetProvider.notifier).getAllAsset();
         break;
       case AppLifecycleState.inactive:
         debugPrint("[APP STATE] inactive");
-        ref.read(appStateProvider.notifier).state = AppStateEnum.inactive;
+        ref.watch(appStateProvider.notifier).state = AppStateEnum.inactive;
+        ref.watch(websocketProvider.notifier).disconnect();
         break;
       case AppLifecycleState.paused:
         debugPrint("[APP STATE] paused");
-        ref.read(appStateProvider.notifier).state = AppStateEnum.paused;
+        ref.watch(appStateProvider.notifier).state = AppStateEnum.paused;
         break;
       case AppLifecycleState.detached:
         debugPrint("[APP STATE] detached");
-        ref.read(appStateProvider.notifier).state = AppStateEnum.detached;
+        ref.watch(appStateProvider.notifier).state = AppStateEnum.detached;
         break;
     }
   }

mobile/lib/modules/home/providers/asset.provider.dart

@@ -10,14 +10,14 @@ import 'package:photo_manager/photo_manager.dart';
 class AssetNotifier extends StateNotifier<List<ImmichAsset>> {
   final AssetService _assetService = AssetService();
   final DeviceInfoService _deviceInfoService = DeviceInfoService();
+  final Ref ref;
 
-  AssetNotifier() : super([]);
+  AssetNotifier(this.ref) : super([]);
 
   getAllAsset() async {
     List<ImmichAsset>? allAssets = await _assetService.getAllAsset();
 
     if (allAssets != null) {
-      allAssets.sortByCompare<DateTime>((e) => DateTime.parse(e.createdAt), (a, b) => b.compareTo(a));
       state = allAssets;
     }
   }
@@ -26,6 +26,10 @@ class AssetNotifier extends StateNotifier<List<ImmichAsset>> {
     state = [];
   }
 
+  onNewAssetUploaded(ImmichAsset newAsset) {
+    state = [...state, newAsset];
+  }
+
   deleteAssets(Set<ImmichAsset> deleteAssets) async {
     var deviceInfo = await _deviceInfoService.getDeviceInfo();
     var deviceId = deviceInfo["deviceId"];
@@ -43,7 +47,6 @@ class AssetNotifier extends StateNotifier<List<ImmichAsset>> {
     }
 
     final List<String> result = await PhotoManager.editor.deleteWithIds(deleteIdList);
-    print(result);
 
     // Delete asset on server
     List<DeleteAssetResponse>? deleteAssetResult = await _assetService.deleteAssets(deleteAssets);
@@ -59,14 +62,13 @@ class AssetNotifier extends StateNotifier<List<ImmichAsset>> {
   }
 }
 
-final currentLocalPageProvider = StateProvider<int>((ref) => 0);
-
 final assetProvider = StateNotifierProvider<AssetNotifier, List<ImmichAsset>>((ref) {
-  return AssetNotifier();
+  return AssetNotifier(ref);
 });
 
 final assetGroupByDateTimeProvider = StateProvider((ref) {
-  var assetGroup = ref.watch(assetProvider);
+  var assets = ref.watch(assetProvider);
 
-  return assetGroup.groupListsBy((element) => DateFormat('y-MM-dd').format(DateTime.parse(element.createdAt)));
+  assets.sortByCompare<DateTime>((e) => DateTime.parse(e.createdAt), (a, b) => b.compareTo(a));
+  return assets.groupListsBy((element) => DateFormat('y-MM-dd').format(DateTime.parse(element.createdAt)));
 });

mobile/lib/modules/home/ui/profile_drawer.dart

@@ -5,6 +5,7 @@ import 'package:immich_mobile/modules/home/providers/asset.provider.dart';
 import 'package:immich_mobile/modules/login/models/authentication_state.model.dart';
 import 'package:immich_mobile/modules/login/providers/authentication.provider.dart';
 import 'package:immich_mobile/shared/providers/backup.provider.dart';
+import 'package:immich_mobile/shared/providers/websocket.provider.dart';
 
 class ProfileDrawer extends ConsumerWidget {
   const ProfileDrawer({Key? key}) : super(key: key);
@@ -60,6 +61,7 @@ class ProfileDrawer extends ConsumerWidget {
               if (res) {
                 ref.watch(backupProvider.notifier).cancelBackup();
                 ref.watch(assetProvider.notifier).clearAllAsset();
+                ref.watch(websocketProvider.notifier).disconnect();
                 AutoRouter.of(context).popUntilRoot();
               }
             },

mobile/lib/modules/home/views/home_page.dart

@@ -11,6 +11,7 @@ import 'package:immich_mobile/modules/home/ui/immich_sliver_appbar.dart';
 import 'package:immich_mobile/modules/home/ui/monthly_title_text.dart';
 import 'package:immich_mobile/modules/home/ui/profile_drawer.dart';
 import 'package:immich_mobile/modules/home/providers/asset.provider.dart';
+import 'package:immich_mobile/shared/providers/websocket.provider.dart';
 import 'package:sliver_tools/sliver_tools.dart';
 
 class HomePage extends HookConsumerWidget {
@@ -25,12 +26,13 @@ class HomePage extends HookConsumerWidget {
     var homePageState = ref.watch(homePageStateProvider);
 
     useEffect(() {
+      ref.read(websocketProvider.notifier).connect();
       ref.read(assetProvider.notifier).getAllAsset();
       return null;
     }, []);
 
     onPopBackFromBackupPage() {
-      ref.read(assetProvider.notifier).getAllAsset();
+      // ref.read(assetProvider.notifier).getAllAsset();
     }
 
     Widget _buildBody() {

mobile/lib/shared/providers/websocket.provider.dart

@@ -0,0 +1,113 @@
+import 'dart:convert';
+
+import 'package:flutter/foundation.dart';
+import 'package:hive/hive.dart';
+import 'package:hooks_riverpod/hooks_riverpod.dart';
+import 'package:immich_mobile/modules/home/providers/asset.provider.dart';
+import 'package:immich_mobile/shared/models/immich_asset.model.dart';
+import 'package:socket_io_client/socket_io_client.dart';
+
+import 'package:immich_mobile/constants/hive_box.dart';
+import 'package:immich_mobile/modules/login/providers/authentication.provider.dart';
+
+class WebscoketState {
+  final Socket? socket;
+  final bool isConnected;
+
+  WebscoketState({
+    this.socket,
+    required this.isConnected,
+  });
+
+  WebscoketState copyWith({
+    Socket? socket,
+    bool? isConnected,
+  }) {
+    return WebscoketState(
+      socket: socket ?? this.socket,
+      isConnected: isConnected ?? this.isConnected,
+    );
+  }
+
+  @override
+  String toString() => 'WebscoketState(socket: $socket, isConnected: $isConnected)';
+
+  @override
+  bool operator ==(Object other) {
+    if (identical(this, other)) return true;
+
+    return other is WebscoketState && other.socket == socket && other.isConnected == isConnected;
+  }
+
+  @override
+  int get hashCode => socket.hashCode ^ isConnected.hashCode;
+}
+
+class WebsocketNotifier extends StateNotifier<WebscoketState> {
+  WebsocketNotifier(this.ref) : super(WebscoketState(socket: null, isConnected: false)) {
+    debugPrint("Init websocket instance");
+  }
+
+  final Ref ref;
+
+  connect() {
+    var authenticationState = ref.watch(authenticationProvider);
+
+    if (authenticationState.isAuthenticated) {
+      var accessToken = Hive.box(userInfoBox).get(accessTokenKey);
+      var endpoint = Hive.box(userInfoBox).get(serverEndpointKey);
+      try {
+        debugPrint("[WEBSOCKET] Attempting to connect to ws");
+        // Configure socket transports must be sepecified
+        Socket socket = io(
+          endpoint,
+          OptionBuilder()
+              .setTransports(['websocket'])
+              .enableReconnection()
+              .enableForceNew()
+              .enableForceNewConnection()
+              .enableAutoConnect()
+              .setExtraHeaders({"Authorization": "Bearer $accessToken"})
+              .build(),
+        );
+
+        socket.onConnect((_) {
+          debugPrint("[WEBSOCKET] Established Websocket Connection");
+          state = WebscoketState(isConnected: true, socket: socket);
+        });
+
+        socket.onDisconnect((_) {
+          debugPrint("[WEBSOCKET] Disconnect to Websocket Connection");
+          state = WebscoketState(isConnected: false, socket: null);
+        });
+
+        socket.on('error', (errorMessage) {
+          debugPrint("Webcoket Error - $errorMessage");
+          state = WebscoketState(isConnected: false, socket: null);
+        });
+
+        socket.on('on_upload_success', (data) {
+          var jsonString = jsonDecode(data.toString());
+          ImmichAsset newAsset = ImmichAsset.fromMap(jsonString);
+          ref.watch(assetProvider.notifier).onNewAssetUploaded(newAsset);
+        });
+      } catch (e) {
+        debugPrint("[WEBSOCKET] Catch Websocket Error - ${e.toString()}");
+      }
+    }
+  }
+
+  disconnect() {
+    debugPrint("[WEBSOCKET] Attempting to disconnect");
+    var socket = state.socket?.disconnect();
+    if (socket != null) {
+      if (socket.disconnected) {
+        state = WebscoketState(isConnected: false, socket: null);
+      }
+    }
+  }
+}
+
+final websocketProvider = StateNotifierProvider<WebsocketNotifier, WebscoketState>((ref) {
+  return WebsocketNotifier(ref);
+});

mobile/pubspec.lock

@@ -742,6 +742,20 @@ packages:
       url: "https://pub.dartlang.org"
     source: hosted
     version: "0.2.5"
+  socket_io_client:
+    dependency: "direct main"
+    description:
+      name: socket_io_client
+      url: "https://pub.dartlang.org"
+    source: hosted
+    version: "2.0.0-beta.4-nullsafety.0"
+  socket_io_common:
+    dependency: transitive
+    description:
+      name: socket_io_common
+      url: "https://pub.dartlang.org"
+    source: hosted
+    version: "2.0.0"
   source_gen:
     dependency: transitive
     description:

mobile/pubspec.yaml

@@ -33,6 +33,7 @@ dependencies:
   sliver_tools: ^0.2.5
   badges: ^2.0.2
   photo_view: ^0.13.0
+  socket_io_client: ^2.0.0-beta.4-nullsafety.0
 
 dev_dependencies:
   flutter_test:

server/docker-compose.yml

@@ -2,16 +2,15 @@ version: '3.8'
 
 
 services:
-  server:
-    container_name: immich_server
+  immich_server:
     image: immich-server-dev:1.0.0
     build:
       context: .
       target: development
       dockerfile: ./Dockerfile
     command: npm run start:dev
-    ports:
-      - "3000:3000"
+    expose: 
+      - "3000"
     volumes:
       - .:/usr/src/app
       - ${UPLOAD_LOCATION}:/usr/src/app/upload
@@ -60,7 +59,7 @@ services:
     networks:
       - immich_network
     depends_on:
-      - server
+      - immich_server
 
 networks:
   immich_network:

server/package.json

@@ -30,7 +30,10 @@
     "@nestjs/passport": "^8.1.0",
     "@nestjs/platform-express": "^8.0.0",
     "@nestjs/platform-fastify": "^8.2.6",
+    "@nestjs/platform-socket.io": "^8.2.6",
     "@nestjs/typeorm": "^8.0.3",
+    "@nestjs/websockets": "^8.2.6",
+    "@socket.io/redis-adapter": "^7.1.0",
     "bcrypt": "^5.0.1",
     "bull": "^4.4.0",
     "class-transformer": "^0.5.1",

server/settings/nginx-conf/nginx.conf

@@ -1,4 +1,15 @@
+
+map $http_upgrade $connection_upgrade {
+  default upgrade;
+  '' close;
+}
+
+# events {
+#   worker_connections 1000;
+# }
+
 server {
+
   client_max_body_size 50000M;
 
   listen 80;
@@ -10,11 +21,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

@@ -0,0 +1,47 @@
+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, OnGatewayDisconnect {
+  constructor(
+    private immichJwtService: ImmichJwtService,
+
+    @InjectRepository(UserEntity)
+    private userRepository: Repository<UserEntity>,
+  ) {}
+
+  @WebSocketServer() server: Server;
+
+  handleDisconnect(client: Socket) {
+    client.leave(client.nsp.name);
+
+    Logger.log(`Client ${client.id} disconnected`);
+  }
+
+  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

@@ -0,0 +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({
+  imports: [TypeOrmModule.forFeature([UserEntity]), ImmichJwtModule, JwtModule.register(jwtConfig)],
+  providers: [CommunicationGateway, CommunicationService, ImmichJwtService],
+  exports: [CommunicationGateway],
+})
+export class CommunicationModule {}

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

@@ -0,0 +1,4 @@
+import { Injectable } from '@nestjs/common';
+
+@Injectable()
+export class CommunicationService {}

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/app.module.ts

@@ -14,6 +14,7 @@ import { BullModule } from '@nestjs/bull';
 import { ImageOptimizeModule } from './modules/image-optimize/image-optimize.module';
 import { ServerInfoModule } from './api-v1/server-info/server-info.module';
 import { BackgroundTaskModule } from './modules/background-task/background-task.module';
+import { CommunicationModule } from './api-v1/communication/communication.module';
 
 @Module({
   imports: [
@@ -40,6 +41,8 @@ import { BackgroundTaskModule } from './modules/background-task/background-task.
     ServerInfoModule,
 
     BackgroundTaskModule,
+
+    CommunicationModule,
   ],
   controllers: [],
   providers: [],

server/src/main.ts

@@ -1,11 +1,15 @@
 import { NestFactory } from '@nestjs/core';
 import { NestExpressApplication } from '@nestjs/platform-express';
 import { AppModule } from './app.module';
+import { RedisIoAdapter } from './middlewares/redis-io.adapter.middleware';
 
 async function bootstrap() {
   const app = await NestFactory.create<NestExpressApplication>(AppModule);
 
   app.set('trust proxy');
+
+  app.useWebSocketAdapter(new RedisIoAdapter(app));
+
   await app.listen(3000);
 }
 bootstrap();

server/src/middlewares/redis-io.adapter.middleware.ts

@@ -0,0 +1,15 @@
+import { IoAdapter } from '@nestjs/platform-socket.io';
+import { RedisClient, createClient } from 'redis';
+import { ServerOptions } from 'socket.io';
+import { createAdapter } from '@socket.io/redis-adapter';
+
+const pubClient = createClient({ url: 'redis://immich_redis:6379' });
+const subClient = pubClient.duplicate();
+
+export class RedisIoAdapter extends IoAdapter {
+  createIOServer(port: number, options?: ServerOptions): any {
+    const server = super.createIOServer(port, options);
+    server.adapter(createAdapter(pubClient, subClient));
+    return server;
+  }
+}

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')
@@ -55,7 +57,12 @@ export class ImageOptimizeProcessor {
               return;
             }
 
-            await this.assetRepository.update(savedAsset, { resizePath: desitnation });
+            const res = await this.assetRepository.update(savedAsset, { resizePath: desitnation });
+            if (res.affected) {
+              this.wsCommunicateionGateway.server
+                .to(savedAsset.userId)
+                .emit('on_upload_success', JSON.stringify(savedAsset));
+            }
           });
       } else {
         sharp(data)
@@ -66,7 +73,12 @@ export class ImageOptimizeProcessor {
               return;
             }
 
-            await this.assetRepository.update(savedAsset, { resizePath: resizePath });
+            const res = await this.assetRepository.update(savedAsset, { resizePath: resizePath });
+            if (res.affected) {
+              this.wsCommunicateionGateway.server
+                .to(savedAsset.userId)
+                .emit('on_upload_success', JSON.stringify(savedAsset));
+            }
           });
       }
     });
@@ -95,7 +107,12 @@ export class ImageOptimizeProcessor {
         filename: `${filename}.png`,
       })
       .on('end', async (a) => {
-        await this.assetRepository.update(savedAsset, { resizePath: `${resizeDir}/${filename}.png` });
+        const res = await this.assetRepository.update(savedAsset, { resizePath: `${resizeDir}/${filename}.png` });
+        if (res.affected) {
+          this.wsCommunicateionGateway.server
+            .to(savedAsset.userId)
+            .emit('on_upload_success', JSON.stringify(savedAsset));
+        }
       });
 
     return 'ok';

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,
+      };
+    }
+  }
 }