chore: sign in with provider (#3592)

* chore: sign in with provider

* feat: implement oauth flow of appflowy cloud

* chore: rename env

* chore: fix deadlock

* fix: login bugs

* chore: clippyt

* chore: update client api

* chore: fmt

frontend/.vscode/tasks.json

@@ -285,6 +285,14 @@
       "options": {
         "cwd": "${workspaceFolder}/appflowy_flutter"
       }
+    },
+    {
+      "label": "AF: Generate AppFlowyEnv",
+      "type": "shell",
+      "command": "dart run build_runner build --delete-conflicting-outputs",
+      "options": {
+        "cwd": "${workspaceFolder}/appflowy_flutter/packages/appflowy_backend"
+      }
     }
   ]
 }

frontend/appflowy_flutter/dev.env

@@ -5,10 +5,11 @@
 
 # Configuring Cloud Type
 # This configuration file is used to specify the cloud type and the necessary configurations for each cloud type. The available options are:
-# Supabase: Set CLOUD_TYPE to 1
-# AppFlowy Cloud: Set CLOUD_TYPE to 2
+# Local: 0
+# Supabase: 1
+# AppFlowy Cloud: 2
 
-CLOUD_TYPE=1
+CLOUD_TYPE=0
 
 # Supabase Configuration
 # If you're using Supabase (CLOUD_TYPE=1), you need to provide the following configurations:
@@ -18,4 +19,5 @@ SUPABASE_ANON_KEY=replace-with-your-supabase-key
 # AppFlowy Cloud Configuration
 # If you're using AppFlowy Cloud (CLOUD_TYPE=2), you need to provide the following configurations:
 APPFLOWY_CLOUD_BASE_URL=replace-with-your-appflowy-cloud-url
-APPFLOWY_CLOUD_BASE_WS_URL=replace-with-your-appflowy-cloud-ws-url
+APPFLOWY_CLOUD_WS_BASE_URL=replace-with-your-appflowy-cloud-ws-url
+APPFLOWY_CLOUD_GOTRUE_URL=replace-with-your-appflowy-cloud-gotrue-url

frontend/appflowy_flutter/lib/env/env.dart

@@ -35,10 +35,17 @@ abstract class Env {
 
   @EnviedField(
     obfuscate: true,
-    varName: 'APPFLOWY_CLOUD_BASE_WS_URL',
+    varName: 'APPFLOWY_CLOUD_WS_BASE_URL',
     defaultValue: '',
   )
-  static final String afCloudBaseWSUrl = _Env.afCloudBaseWSUrl;
+  static final String afCloudWSBaseUrl = _Env.afCloudWSBaseUrl;
+
+  @EnviedField(
+    obfuscate: true,
+    varName: 'APPFLOWY_CLOUD_GOTRUE_URL',
+    defaultValue: '',
+  )
+  static final String afCloudGoTrueUrl = _Env.afCloudGoTrueUrl;
 
   // Supabase Configuration:
   @EnviedField(
@@ -64,6 +71,24 @@ bool get isCloudEnabled {
   }
 }
 
+bool get isSupabaseEnabled {
+  // Only enable supabase in release and develop mode.
+  if (integrationMode().isRelease || integrationMode().isDevelop) {
+    return currentCloudType() == CloudType.supabase;
+  } else {
+    return false;
+  }
+}
+
+bool get isAppFlowyCloudEnabled {
+  // Only enable appflowy cloud in release and develop mode.
+  if (integrationMode().isRelease || integrationMode().isDevelop) {
+    return currentCloudType() == CloudType.appflowyCloud;
+  } else {
+    return false;
+  }
+}
+
 enum CloudType {
   unknown,
   supabase,
@@ -84,7 +109,7 @@ CloudType currentCloudType() {
   }
 
   if (value == 2) {
-    if (Env.afCloudBaseUrl.isEmpty || Env.afCloudBaseWSUrl.isEmpty) {
+    if (Env.afCloudBaseUrl.isEmpty || Env.afCloudWSBaseUrl.isEmpty) {
       Log.error("AppFlowy cloud is not configured");
       return CloudType.unknown;
     } else {

frontend/appflowy_flutter/lib/startup/tasks/rust_sdk.dart

@@ -38,7 +38,8 @@ AppFlowyEnv getAppFlowyEnv() {
 
   final appflowyCloudConfig = AppFlowyCloudConfiguration(
     base_url: Env.afCloudBaseUrl,
-    base_ws_url: Env.afCloudBaseWSUrl,
+    ws_base_url: Env.afCloudWSBaseUrl,
+    gotrue_url: Env.afCloudGoTrueUrl,
   );
 
   return AppFlowyEnv(

frontend/appflowy_flutter/lib/startup/tasks/supabase_task.dart

@@ -28,7 +28,7 @@ SupbaseRealtimeService? realtimeService;
 class InitSupabaseTask extends LaunchTask {
   @override
   Future<void> initialize(LaunchContext context) async {
-    if (!isCloudEnabled) {
+    if (!isSupabaseEnabled) {
       return;
     }
 

frontend/appflowy_flutter/lib/user/application/auth/af_cloud_auth_service.dart

@@ -1,13 +1,23 @@
 import 'dart:async';
 
+import 'package:app_links/app_links.dart';
 import 'package:appflowy/user/application/auth/backend_auth_service.dart';
 import 'package:appflowy/user/application/auth/auth_service.dart';
 import 'package:appflowy/user/application/user_service.dart';
+import 'package:appflowy_backend/dispatch/dispatch.dart';
+import 'package:appflowy_backend/log.dart';
 import 'package:appflowy_backend/protobuf/flowy-error/errors.pb.dart';
 import 'package:appflowy_backend/protobuf/flowy-user/protobuf.dart';
 import 'package:dartz/dartz.dart';
+import 'package:url_launcher/url_launcher.dart';
+
+import 'auth_error.dart';
+import 'device_id.dart';
 
 class AFCloudAuthService implements AuthService {
+  final _appLinks = AppLinks();
+  StreamSubscription<Uri?>? _deeplinkSubscription;
+
   AFCloudAuthService();
 
   final BackendAuthService _backendAuthService = BackendAuthService(
@@ -38,8 +48,72 @@ class AFCloudAuthService implements AuthService {
     required String platform,
     Map<String, String> params = const {},
   }) async {
-    //
-    throw UnimplementedError();
+    final provider = ProviderTypePBExtension.fromPlatform(platform);
+
+    // Get the oauth url from the backend
+    final result = await UserEventGetOauthURLWithProvider(
+      OauthProviderPB.create()..provider = provider,
+    ).send();
+
+    return result.fold(
+      (data) async {
+        // Open the webview with oauth url
+        final uri = Uri.parse(data.oauthUrl);
+        final isSuccess = await launchUrl(
+          uri,
+          mode: LaunchMode.externalApplication,
+          webOnlyWindowName: '_self',
+        );
+
+        final completer = Completer<Either<FlowyError, UserProfilePB>>();
+        _deeplinkSubscription = _appLinks.uriLinkStream.listen(
+          (Uri? uri) async {
+            await _handleUri(uri, completer);
+          },
+          onError: (Object err, StackTrace stackTrace) {
+            Log.error('onDeepLinkError: ${err.toString()}', stackTrace);
+            _deeplinkSubscription?.cancel();
+            completer.complete(left(AuthError.deeplinkError));
+          },
+        );
+
+        if (!isSuccess) {
+          _deeplinkSubscription?.cancel();
+          completer.complete(left(AuthError.signInWithOauthError));
+        }
+
+        return completer.future;
+      },
+      (r) => left(r),
+    );
+  }
+
+  Future<void> _handleUri(
+    Uri? uri,
+    Completer<Either<FlowyError, UserProfilePB>> completer,
+  ) async {
+    if (uri != null) {
+      if (_isAuthCallbackDeeplink(uri)) {
+        // Sign in with url
+        final deviceId = await getDeviceId();
+        final payload = OauthSignInPB(
+          authType: AuthTypePB.AFCloud,
+          map: {
+            AuthServiceMapKeys.signInURL: uri.toString(),
+            AuthServiceMapKeys.deviceId: deviceId
+          },
+        );
+        final result = await UserEventOauthSignIn(payload)
+            .send()
+            .then((value) => value.swap());
+        _deeplinkSubscription?.cancel();
+        completer.complete(result);
+      }
+    } else {
+      Log.error('onDeepLinkError: Unexpect empty deep link callback');
+      _deeplinkSubscription?.cancel();
+      completer.complete(left(AuthError.emptyDeeplink));
+    }
   }
 
   @override
@@ -67,3 +141,22 @@ class AFCloudAuthService implements AuthService {
     return UserBackendService.getCurrentUserProfile();
   }
 }
+
+extension ProviderTypePBExtension on ProviderTypePB {
+  static ProviderTypePB fromPlatform(String platform) {
+    switch (platform) {
+      case 'github':
+        return ProviderTypePB.Github;
+      case 'google':
+        return ProviderTypePB.Google;
+      case 'discord':
+        return ProviderTypePB.Discord;
+      default:
+        throw UnimplementedError();
+    }
+  }
+}
+
+bool _isAuthCallbackDeeplink(Uri uri) {
+  return (uri.fragment.contains('access_token'));
+}

frontend/appflowy_flutter/lib/user/application/auth/auth_error.dart

@@ -17,4 +17,16 @@ class AuthError {
   static final supabaseGetUserError = FlowyError()
     ..msg = 'unable to get user from supabase  -10004'
     ..code = ErrorCode.UserUnauthorized;
+
+  static final signInWithOauthError = FlowyError()
+    ..msg = 'sign in with oauth error -10003'
+    ..code = ErrorCode.UserUnauthorized;
+
+  static final emptyDeeplink = FlowyError()
+    ..msg = 'Unexpected empty deeplink'
+    ..code = ErrorCode.UnexpectedEmpty;
+
+  static final deeplinkError = FlowyError()
+    ..msg = 'Deeplink error'
+    ..code = ErrorCode.Internal;
 }

frontend/appflowy_flutter/lib/user/application/auth/auth_service.dart

@@ -9,6 +9,7 @@ class AuthServiceMapKeys {
   static const String uuid = 'uuid';
   static const String email = 'email';
   static const String deviceId = 'device_id';
+  static const String signInURL = 'sign_in_url';
 }
 
 /// `AuthService` is an abstract class that defines methods related to user authentication.

frontend/appflowy_flutter/lib/user/application/auth/supabase_auth_service.dart

@@ -169,12 +169,12 @@ class SupabaseAuthService implements AuthService {
   Future<Either<FlowyError, UserProfilePB>> _setupAuth({
     required Map<String, String> map,
   }) async {
-    final payload = OAuthPB(
+    final payload = OauthSignInPB(
       authType: AuthTypePB.Supabase,
       map: map,
     );
 
-    return UserEventOAuth(payload).send().then((value) => value.swap());
+    return UserEventOauthSignIn(payload).send().then((value) => value.swap());
   }
 }
 

frontend/appflowy_flutter/lib/user/application/auth/supabase_mock_auth_service.dart

@@ -56,7 +56,7 @@ class MockAuthService implements AuthService {
       final uuid = response.user!.id;
       final email = response.user!.email!;
 
-      final payload = OAuthPB(
+      final payload = OauthSignInPB(
         authType: AuthTypePB.Supabase,
         map: {
           AuthServiceMapKeys.uuid: uuid,
@@ -65,7 +65,7 @@ class MockAuthService implements AuthService {
         },
       );
 
-      return UserEventOAuth(payload).send().then((value) => value.swap());
+      return UserEventOauthSignIn(payload).send().then((value) => value.swap());
     } on AuthException catch (e) {
       Log.error(e);
       return Left(AuthError.supabaseSignInError);

frontend/appflowy_flutter/packages/appflowy_backend/lib/env_serde.dart

@@ -1,10 +1,9 @@
 import 'package:json_annotation/json_annotation.dart';
 
 // Run `dart run build_runner build` to generate the json serialization If the
-// file `env_serde.i.dart` is existed, delete it first.
+// file `env_serde.g.dart` is existed, delete it first.
 //
-// the file `env_serde.g.dart` will be generated in the same directory. Rename
-// the file to `env_serde.i.dart` because the file is ignored by default.
+// the file `env_serde.g.dart` will be generated in the same directory.
 part 'env_serde.g.dart';
 
 @JsonSerializable()
@@ -45,11 +44,13 @@ class SupabaseConfiguration {
 @JsonSerializable()
 class AppFlowyCloudConfiguration {
   final String base_url;
-  final String base_ws_url;
+  final String ws_base_url;
+  final String gotrue_url;
 
   AppFlowyCloudConfiguration({
     required this.base_url,
-    required this.base_ws_url,
+    required this.ws_base_url,
+    required this.gotrue_url,
   });
 
   factory AppFlowyCloudConfiguration.fromJson(Map<String, dynamic> json) =>

frontend/appflowy_flutter/pubspec.lock

@@ -26,7 +26,7 @@ packages:
     source: hosted
     version: "2.0.7"
   app_links:
-    dependency: "direct overridden"
+    dependency: "direct main"
     description:
       path: "."
       ref: c64ce17

frontend/appflowy_flutter/pubspec.yaml

@@ -115,6 +115,7 @@ dependencies:
   # TODO: Consider implementing custom package
   # to gather notification handling for all platforms
   local_notifier: ^0.1.5
+  app_links: ^3.4.1
 
 dev_dependencies:
   flutter_lints: ^2.0.1

frontend/appflowy_tauri/src-tauri/Cargo.lock

@@ -762,7 +762,7 @@ dependencies = [
 [[package]]
 name = "client-api"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=b0c213#b0c213b5c0b941e2013b0126e200f98201a82f54"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=926da91#926da912eab806e5b325e12103d341cea536aa2b"
 dependencies = [
  "anyhow",
  "bytes",
@@ -775,7 +775,6 @@ dependencies = [
  "gotrue-entity",
  "lib0",
  "mime",
- "opener",
  "parking_lot",
  "reqwest",
  "scraper",
@@ -853,7 +852,7 @@ dependencies = [
 [[package]]
 name = "collab"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -872,7 +871,7 @@ dependencies = [
 [[package]]
 name = "collab-database"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -902,7 +901,7 @@ dependencies = [
 [[package]]
 name = "collab-define"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "anyhow",
  "bytes",
@@ -916,7 +915,7 @@ dependencies = [
 [[package]]
 name = "collab-derive"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -928,7 +927,7 @@ dependencies = [
 [[package]]
 name = "collab-document"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "anyhow",
  "collab",
@@ -948,7 +947,7 @@ dependencies = [
 [[package]]
 name = "collab-folder"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "anyhow",
  "chrono",
@@ -979,16 +978,18 @@ dependencies = [
  "collab-persistence",
  "collab-plugins",
  "futures",
+ "lib-infra",
  "parking_lot",
  "serde",
  "serde_json",
+ "tokio",
  "tracing",
 ]
 
 [[package]]
 name = "collab-persistence"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "async-trait",
  "bincode",
@@ -1009,7 +1010,7 @@ dependencies = [
 [[package]]
 name = "collab-plugins"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -1036,7 +1037,7 @@ dependencies = [
 [[package]]
 name = "collab-user"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "anyhow",
  "collab",
@@ -1289,7 +1290,7 @@ dependencies = [
  "cssparser-macros",
  "dtoa-short",
  "itoa 1.0.6",
- "phf 0.11.2",
+ "phf 0.8.0",
  "smallvec",
 ]
 
@@ -1435,13 +1436,15 @@ checksum = "c2e66c9d817f1720209181c316d28635c050fa304f9c79e47a520882661b7308"
 [[package]]
 name = "database-entity"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=b0c213#b0c213b5c0b941e2013b0126e200f98201a82f54"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=926da91#926da912eab806e5b325e12103d341cea536aa2b"
 dependencies = [
+ "anyhow",
  "chrono",
  "collab-define",
  "serde",
  "serde_json",
  "sqlx",
+ "thiserror",
  "uuid",
  "validator",
 ]
@@ -2772,7 +2775,7 @@ dependencies = [
 [[package]]
 name = "gotrue"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=b0c213#b0c213b5c0b941e2013b0126e200f98201a82f54"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=926da91#926da912eab806e5b325e12103d341cea536aa2b"
 dependencies = [
  "anyhow",
  "futures-util",
@@ -2782,12 +2785,13 @@ dependencies = [
  "serde",
  "serde_json",
  "tokio",
+ "tracing",
 ]
 
 [[package]]
 name = "gotrue-entity"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=b0c213#b0c213b5c0b941e2013b0126e200f98201a82f54"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=926da91#926da912eab806e5b325e12103d341cea536aa2b"
 dependencies = [
  "anyhow",
  "reqwest",
@@ -3220,7 +3224,7 @@ dependencies = [
 [[package]]
 name = "infra"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=b0c213#b0c213b5c0b941e2013b0126e200f98201a82f54"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=926da91#926da912eab806e5b325e12103d341cea536aa2b"
 dependencies = [
  "anyhow",
  "reqwest",
@@ -4261,7 +4265,6 @@ version = "0.11.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ade2d8b8f33c7333b51bcf0428d37e217e9f32192ae4772156f65063b8ce03dc"
 dependencies = [
- "phf_macros 0.11.2",
  "phf_shared 0.11.2",
 ]
 
@@ -4353,19 +4356,6 @@ dependencies = [
  "syn 1.0.109",
 ]
 
-[[package]]
-name = "phf_macros"
-version = "0.11.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3444646e286606587e49f3bcf1679b8cef1dc2c5ecc29ddacaffc305180d464b"
-dependencies = [
- "phf_generator 0.11.2",
- "phf_shared 0.11.2",
- "proc-macro2",
- "quote",
- "syn 2.0.29",
-]
-
 [[package]]
 name = "phf_shared"
 version = "0.8.0"
@@ -5579,9 +5569,10 @@ dependencies = [
 [[package]]
 name = "shared_entity"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=b0c213#b0c213b5c0b941e2013b0126e200f98201a82f54"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=926da91#926da912eab806e5b325e12103d341cea536aa2b"
 dependencies = [
  "anyhow",
+ "database-entity",
  "gotrue-entity",
  "opener",
  "reqwest",
@@ -6494,7 +6485,9 @@ checksum = "212d5dcb2a1ce06d81107c3d0ffa3121fe974b73f068c8282cb1c32328113b6c"
 dependencies = [
  "futures-util",
  "log",
+ "native-tls",
  "tokio",
+ "tokio-native-tls",
  "tungstenite",
 ]
 
@@ -6711,6 +6704,7 @@ dependencies = [
  "http",
  "httparse",
  "log",
+ "native-tls",
  "rand 0.8.5",
  "sha1",
  "thiserror",

frontend/appflowy_tauri/src-tauri/Cargo.toml

@@ -38,7 +38,7 @@ custom-protocol = ["tauri/custom-protocol"]
 # Run the script:
 # scripts/tool/update_client_api_rev.sh  new_rev_id
 # ⚠️⚠️⚠️️
-client-api = { git = "https://github.com/AppFlowy-IO/AppFlowy-Cloud", rev = "b0c213" }
+client-api = { git = "https://github.com/AppFlowy-IO/AppFlowy-Cloud", rev = "926da91" }
 # Please use the following script to update collab.
 # Working directory: frontend
 #
@@ -48,14 +48,14 @@ client-api = { git = "https://github.com/AppFlowy-IO/AppFlowy-Cloud", rev = "b0c
 # To switch to the local path, run:
 # scripts/tool/update_collab_source.sh
 # ⚠️⚠️⚠️️
-collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
-collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
-collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
-collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
-collab-plugins = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
-collab-user = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
-collab-define = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
-collab-persistence = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
+collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
+collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
+collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
+collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
+collab-plugins = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
+collab-user = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
+collab-define = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
+collab-persistence = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
 
 
 

frontend/rust-lib/Cargo.lock

@@ -660,7 +660,7 @@ dependencies = [
 [[package]]
 name = "client-api"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=b0c213#b0c213b5c0b941e2013b0126e200f98201a82f54"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=926da91#926da912eab806e5b325e12103d341cea536aa2b"
 dependencies = [
  "anyhow",
  "bytes",
@@ -673,7 +673,6 @@ dependencies = [
  "gotrue-entity",
  "lib0",
  "mime",
- "opener",
  "parking_lot",
  "reqwest",
  "scraper",
@@ -720,7 +719,7 @@ dependencies = [
 [[package]]
 name = "collab"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -739,7 +738,7 @@ dependencies = [
 [[package]]
 name = "collab-database"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -769,7 +768,7 @@ dependencies = [
 [[package]]
 name = "collab-define"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "anyhow",
  "bytes",
@@ -783,7 +782,7 @@ dependencies = [
 [[package]]
 name = "collab-derive"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -795,7 +794,7 @@ dependencies = [
 [[package]]
 name = "collab-document"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "anyhow",
  "collab",
@@ -815,7 +814,7 @@ dependencies = [
 [[package]]
 name = "collab-folder"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "anyhow",
  "chrono",
@@ -846,16 +845,18 @@ dependencies = [
  "collab-persistence",
  "collab-plugins",
  "futures",
+ "lib-infra",
  "parking_lot",
  "serde",
  "serde_json",
+ "tokio",
  "tracing",
 ]
 
 [[package]]
 name = "collab-persistence"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "async-trait",
  "bincode",
@@ -876,7 +877,7 @@ dependencies = [
 [[package]]
 name = "collab-plugins"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -903,7 +904,7 @@ dependencies = [
 [[package]]
 name = "collab-user"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=86c5e8#86c5e8891af93de2d035a57214894e854d39662a"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=82975da#82975da119a0c9d0e8b70585966bb89ed6c97cd3"
 dependencies = [
  "anyhow",
  "collab",
@@ -1135,7 +1136,7 @@ dependencies = [
  "cssparser-macros",
  "dtoa-short",
  "itoa",
- "phf 0.11.2",
+ "phf 0.8.0",
  "smallvec",
 ]
 
@@ -1262,13 +1263,15 @@ checksum = "c2e66c9d817f1720209181c316d28635c050fa304f9c79e47a520882661b7308"
 [[package]]
 name = "database-entity"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=b0c213#b0c213b5c0b941e2013b0126e200f98201a82f54"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=926da91#926da912eab806e5b325e12103d341cea536aa2b"
 dependencies = [
+ "anyhow",
  "chrono",
  "collab-define",
  "serde",
  "serde_json",
  "sqlx",
+ "thiserror",
  "uuid",
  "validator",
 ]
@@ -2434,7 +2437,7 @@ dependencies = [
 [[package]]
 name = "gotrue"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=b0c213#b0c213b5c0b941e2013b0126e200f98201a82f54"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=926da91#926da912eab806e5b325e12103d341cea536aa2b"
 dependencies = [
  "anyhow",
  "futures-util",
@@ -2444,12 +2447,13 @@ dependencies = [
  "serde",
  "serde_json",
  "tokio",
+ "tracing",
 ]
 
 [[package]]
 name = "gotrue-entity"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=b0c213#b0c213b5c0b941e2013b0126e200f98201a82f54"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=926da91#926da912eab806e5b325e12103d341cea536aa2b"
 dependencies = [
  "anyhow",
  "reqwest",
@@ -2807,7 +2811,7 @@ dependencies = [
 [[package]]
 name = "infra"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=b0c213#b0c213b5c0b941e2013b0126e200f98201a82f54"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=926da91#926da912eab806e5b325e12103d341cea536aa2b"
 dependencies = [
  "anyhow",
  "reqwest",
@@ -3570,7 +3574,7 @@ version = "0.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3dfb61232e34fcb633f43d12c58f83c1df82962dcdfa565a4e866ffc17dafe12"
 dependencies = [
- "phf_macros 0.8.0",
+ "phf_macros",
  "phf_shared 0.8.0",
  "proc-macro-hack",
 ]
@@ -3590,7 +3594,6 @@ version = "0.11.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ade2d8b8f33c7333b51bcf0428d37e217e9f32192ae4772156f65063b8ce03dc"
 dependencies = [
- "phf_macros 0.11.2",
  "phf_shared 0.11.2",
 ]
 
@@ -3658,19 +3661,6 @@ dependencies = [
  "syn 1.0.109",
 ]
 
-[[package]]
-name = "phf_macros"
-version = "0.11.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3444646e286606587e49f3bcf1679b8cef1dc2c5ecc29ddacaffc305180d464b"
-dependencies = [
- "phf_generator 0.11.2",
- "phf_shared 0.11.2",
- "proc-macro2",
- "quote",
- "syn 2.0.31",
-]
-
 [[package]]
 name = "phf_shared"
 version = "0.8.0"
@@ -4821,9 +4811,10 @@ dependencies = [
 [[package]]
 name = "shared_entity"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=b0c213#b0c213b5c0b941e2013b0126e200f98201a82f54"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Cloud?rev=926da91#926da912eab806e5b325e12103d341cea536aa2b"
 dependencies = [
  "anyhow",
+ "database-entity",
  "gotrue-entity",
  "opener",
  "reqwest",
@@ -5432,7 +5423,9 @@ checksum = "212d5dcb2a1ce06d81107c3d0ffa3121fe974b73f068c8282cb1c32328113b6c"
 dependencies = [
  "futures-util",
  "log",
+ "native-tls",
  "tokio",
+ "tokio-native-tls",
  "tungstenite",
 ]
 
@@ -5660,6 +5653,7 @@ dependencies = [
  "http",
  "httparse",
  "log",
+ "native-tls",
  "rand 0.8.5",
  "sha1",
  "thiserror",

frontend/rust-lib/Cargo.toml

@@ -82,7 +82,7 @@ incremental = false
 # Run the script:
 # scripts/tool/update_client_api_rev.sh  new_rev_id
 # ⚠️⚠️⚠️️
-client-api = { git = "https://github.com/AppFlowy-IO/AppFlowy-Cloud", rev = "b0c213" }
+client-api = { git = "https://github.com/AppFlowy-IO/AppFlowy-Cloud", rev = "926da91" }
 # Please use the following script to update collab.
 # Working directory: frontend
 #
@@ -92,11 +92,11 @@ client-api = { git = "https://github.com/AppFlowy-IO/AppFlowy-Cloud", rev = "b0c
 # To switch to the local path, run:
 # scripts/tool/update_collab_source.sh
 # ⚠️⚠️⚠️️
-collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
-collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
-collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
-collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
-collab-plugins = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
-collab-user = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
-collab-define = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
-collab-persistence = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "86c5e8" }
+collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
+collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
+collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
+collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
+collab-plugins = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
+collab-user = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
+collab-define = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }
+collab-persistence = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "82975da" }

frontend/rust-lib/collab-integrate/Cargo.toml

@@ -20,6 +20,8 @@ tracing = "0.1"
 parking_lot = "0.12.1"
 futures = "0.3"
 async-trait = "0.1.73"
+tokio = {version = "1.26", features = ["sync"]}
+lib-infra = { path = "../../../shared-lib/lib-infra" }
 
 [features]
 default = []

frontend/rust-lib/collab-integrate/src/collab_builder.rs

@@ -11,8 +11,10 @@ use collab_plugins::cloud_storage::network_state::{CollabNetworkReachability, Co
 use collab_plugins::local_storage::rocksdb::RocksdbDiskPlugin;
 use collab_plugins::local_storage::CollabPersistenceConfig;
 use collab_plugins::snapshot::{CollabSnapshotPlugin, SnapshotPersistence};
-use futures::executor::block_on;
 use parking_lot::{Mutex, RwLock};
+use tracing::trace;
+
+use lib_infra::future::{to_fut, Fut};
 
 #[derive(Clone, Debug)]
 pub enum CollabSource {
@@ -36,19 +38,17 @@ pub enum CollabPluginContext {
   },
 }
 
-#[async_trait]
 pub trait CollabStorageProvider: Send + Sync + 'static {
   fn storage_source(&self) -> CollabSource;
 
-  async fn get_plugins(
+  fn get_plugins(
     &self,
     context: CollabPluginContext,
-  ) -> Vec<Arc<dyn collab::core::collab_plugin::CollabPlugin>>;
+  ) -> Fut<Vec<Arc<dyn collab::core::collab_plugin::CollabPlugin>>>;
 
   fn is_sync_enabled(&self) -> bool;
 }
 
-#[async_trait]
 impl<T> CollabStorageProvider for Arc<T>
 where
   T: CollabStorageProvider,
@@ -57,8 +57,8 @@ where
     (**self).storage_source()
   }
 
-  async fn get_plugins(&self, context: CollabPluginContext) -> Vec<Arc<dyn CollabPlugin>> {
-    (**self).get_plugins(context).await
+  fn get_plugins(&self, context: CollabPluginContext) -> Fut<Vec<Arc<dyn CollabPlugin>>> {
+    (**self).get_plugins(context)
   }
 
   fn is_sync_enabled(&self) -> bool {
@@ -69,7 +69,7 @@ where
 pub struct AppFlowyCollabBuilder {
   network_reachability: CollabNetworkReachability,
   workspace_id: RwLock<Option<String>>,
-  cloud_storage: RwLock<Arc<dyn CollabStorageProvider>>,
+  cloud_storage: tokio::sync::RwLock<Arc<dyn CollabStorageProvider>>,
   snapshot_persistence: Mutex<Option<Arc<dyn SnapshotPersistence>>>,
   device_id: Mutex<String>,
 }
@@ -79,7 +79,7 @@ impl AppFlowyCollabBuilder {
     Self {
       network_reachability: CollabNetworkReachability::new(),
       workspace_id: Default::default(),
-      cloud_storage: RwLock::new(Arc::new(storage_provider)),
+      cloud_storage: tokio::sync::RwLock::new(Arc::new(storage_provider)),
       snapshot_persistence: Default::default(),
       device_id: Default::default(),
     }
@@ -141,7 +141,7 @@ impl AppFlowyCollabBuilder {
   /// - `raw_data`: The raw data of the collaboration object, defined by the [CollabRawData] type.
   /// - `collab_db`: A weak reference to the [RocksCollabDB].
   ///
-  pub fn build(
+  pub async fn build(
     &self,
     uid: i64,
     object_id: &str,
@@ -149,14 +149,16 @@ impl AppFlowyCollabBuilder {
     raw_data: CollabRawData,
     collab_db: Weak<RocksCollabDB>,
   ) -> Result<Arc<MutexCollab>, Error> {
-    self.build_with_config(
-      uid,
-      object_id,
-      object_type,
-      collab_db,
-      raw_data,
-      &CollabPersistenceConfig::default(),
-    )
+    self
+      .build_with_config(
+        uid,
+        object_id,
+        object_type,
+        collab_db,
+        raw_data,
+        &CollabPersistenceConfig::default(),
+      )
+      .await
   }
 
   /// Creates a new collaboration builder with the custom configuration.
@@ -173,7 +175,7 @@ impl AppFlowyCollabBuilder {
   /// - `raw_data`: The raw data of the collaboration object, defined by the [CollabRawData] type.
   /// - `collab_db`: A weak reference to the [RocksCollabDB].
   ///
-  pub fn build_with_config(
+  pub async fn build_with_config(
     &self,
     uid: i64,
     object_id: &str,
@@ -194,21 +196,26 @@ impl AppFlowyCollabBuilder {
         .build()?,
     );
     {
-      let cloud_storage = self.cloud_storage.read();
-      let cloud_storage_type = cloud_storage.storage_source();
+      let cloud_storage_type = self.cloud_storage.read().await.storage_source();
       let collab_object = self.collab_object(uid, object_id, object_type)?;
       match cloud_storage_type {
         CollabSource::AFCloud => {
           #[cfg(feature = "appflowy_cloud_integrate")]
           {
+            trace!("init appflowy cloud collab plugins");
             let local_collab = Arc::downgrade(&collab);
-            let plugins = block_on(
-              cloud_storage.get_plugins(CollabPluginContext::AppFlowyCloud {
+            let plugins = self
+              .cloud_storage
+              .read()
+              .await
+              .get_plugins(CollabPluginContext::AppFlowyCloud {
                 uid,
                 collab_object: collab_object.clone(),
                 local_collab,
-              }),
-            );
+              })
+              .await;
+
+            trace!("add appflowy cloud collab plugins: {}", plugins.len());
             for plugin in plugins {
               collab.lock().add_plugin(plugin);
             }
@@ -217,14 +224,20 @@ impl AppFlowyCollabBuilder {
         CollabSource::Supabase => {
           #[cfg(feature = "supabase_integrate")]
           {
+            trace!("init supabase collab plugins");
             let local_collab = Arc::downgrade(&collab);
             let local_collab_db = collab_db.clone();
-            let plugins = block_on(cloud_storage.get_plugins(CollabPluginContext::Supabase {
-              uid,
-              collab_object: collab_object.clone(),
-              local_collab,
-              local_collab_db,
-            }));
+            let plugins = self
+              .cloud_storage
+              .read()
+              .await
+              .get_plugins(CollabPluginContext::Supabase {
+                uid,
+                collab_object: collab_object.clone(),
+                local_collab,
+                local_collab_db,
+              })
+              .await;
             for plugin in plugins {
               collab.lock().add_plugin(plugin);
             }
@@ -248,7 +261,7 @@ impl AppFlowyCollabBuilder {
       }
     }
 
-    block_on(collab.async_initialize());
+    collab.lock().initialize();
     Ok(collab)
   }
 }
@@ -261,8 +274,8 @@ impl CollabStorageProvider for DefaultCollabStorageProvider {
     CollabSource::Local
   }
 
-  async fn get_plugins(&self, _context: CollabPluginContext) -> Vec<Arc<dyn CollabPlugin>> {
-    vec![]
+  fn get_plugins(&self, _context: CollabPluginContext) -> Fut<Vec<Arc<dyn CollabPlugin>>> {
+    to_fut(async move { vec![] })
   }
 
   fn is_sync_enabled(&self) -> bool {

frontend/rust-lib/flowy-core/src/integrate/trait_impls.rs

@@ -20,8 +20,7 @@ use flowy_storage::{FileStorageService, StorageObject};
 use flowy_user::event_map::UserCloudServiceProvider;
 use flowy_user_deps::cloud::UserCloudService;
 use flowy_user_deps::entities::AuthType;
-use lib_infra::async_trait::async_trait;
-use lib_infra::future::FutureResult;
+use lib_infra::future::{to_fut, Fut, FutureResult};
 
 use crate::integrate::server::{ServerProvider, ServerType, SERVER_PROVIDER_TYPE_KEY};
 
@@ -258,48 +257,53 @@ impl DocumentCloudService for ServerProvider {
   }
 }
 
-#[async_trait]
 impl CollabStorageProvider for ServerProvider {
   fn storage_source(&self) -> CollabSource {
     self.get_server_type().into()
   }
 
-  async fn get_plugins(&self, context: CollabPluginContext) -> Vec<Arc<dyn CollabPlugin>> {
-    let mut plugins: Vec<Arc<dyn CollabPlugin>> = vec![];
+  fn get_plugins(&self, context: CollabPluginContext) -> Fut<Vec<Arc<dyn CollabPlugin>>> {
     match context {
-      CollabPluginContext::Local => {},
+      CollabPluginContext::Local => to_fut(async move { vec![] }),
       CollabPluginContext::AppFlowyCloud {
         uid: _,
         collab_object,
         local_collab,
       } => {
         if let Ok(server) = self.get_server(&ServerType::AFCloud) {
-          match server.collab_ws_channel(&collab_object.object_id).await {
-            Ok(Some((channel, ws_connect_state))) => {
-              let origin = CollabOrigin::Client(CollabClient::new(
-                collab_object.uid,
-                collab_object.device_id.clone(),
-              ));
-              let sync_object = SyncObject::from(collab_object);
-              let (sink, stream) = (channel.sink(), channel.stream());
-              let sink_config = SinkConfig::new().with_timeout(6);
-              let sync_plugin = SyncPlugin::new(
-                origin,
-                sync_object,
-                local_collab,
-                sink,
-                sink_config,
-                stream,
-                Some(channel),
-                ws_connect_state,
-              );
-              plugins.push(Arc::new(sync_plugin));
-            },
-            Ok(None) => {
-              tracing::error!("🔴Failed to get collab ws channel: channel is none");
-            },
-            Err(err) => tracing::error!("🔴Failed to get collab ws channel: {:?}", err),
-          }
+          to_fut(async move {
+            let mut plugins: Vec<Arc<dyn CollabPlugin>> = vec![];
+            match server.collab_ws_channel(&collab_object.object_id).await {
+              Ok(Some((channel, ws_connect_state))) => {
+                let origin = CollabOrigin::Client(CollabClient::new(
+                  collab_object.uid,
+                  collab_object.device_id.clone(),
+                ));
+                let sync_object = SyncObject::from(collab_object);
+                let (sink, stream) = (channel.sink(), channel.stream());
+                let sink_config = SinkConfig::new().with_timeout(6);
+                let sync_plugin = SyncPlugin::new(
+                  origin,
+                  sync_object,
+                  local_collab,
+                  sink,
+                  sink_config,
+                  stream,
+                  Some(channel),
+                  ws_connect_state,
+                );
+                plugins.push(Arc::new(sync_plugin));
+              },
+              Ok(None) => {
+                tracing::error!("🔴Failed to get collab ws channel: channel is none");
+              },
+              Err(err) => tracing::error!("🔴Failed to get collab ws channel: {:?}", err),
+            }
+
+            plugins
+          })
+        } else {
+          to_fut(async move { vec![] })
         }
       },
       CollabPluginContext::Supabase {
@@ -308,6 +312,7 @@ impl CollabStorageProvider for ServerProvider {
         local_collab,
         local_collab_db,
       } => {
+        let mut plugins: Vec<Arc<dyn CollabPlugin>> = vec![];
         if let Some(remote_collab_storage) = self
           .get_server(&ServerType::Supabase)
           .ok()
@@ -322,9 +327,10 @@ impl CollabStorageProvider for ServerProvider {
             local_collab_db,
           )));
         }
+
+        to_fut(async move { plugins })
       },
     }
-    plugins
   }
 
   fn is_sync_enabled(&self) -> bool {

frontend/rust-lib/flowy-core/src/integrate/user.rs

@@ -1,5 +1,7 @@
 use std::sync::Arc;
 
+use anyhow::Context;
+
 use collab_integrate::collab_builder::AppFlowyCollabBuilder;
 use flowy_database2::DatabaseManager;
 use flowy_document2::manager::DocumentManager;
@@ -130,18 +132,22 @@ impl UserStatusCallback for UserStatusCallbackImpl {
           },
           &user_workspace.id,
         )
-        .await?;
+        .await
+        .context("FolderManager error")?;
+
       database_manager
         .initialize_with_new_user(
           user_profile.uid,
           user_workspace.id.clone(),
           user_workspace.database_views_aggregate_id,
         )
-        .await?;
+        .await
+        .context("DatabaseManager error")?;
 
       document_manager
         .initialize_with_new_user(user_profile.uid, user_workspace.id)
-        .await?;
+        .await
+        .context("DocumentManager error")?;
       Ok(())
     })
   }

frontend/rust-lib/flowy-database2/src/manager.rs

@@ -11,7 +11,9 @@ use collab_database::user::{
 };
 use collab_database::views::{CreateDatabaseParams, CreateViewParams, DatabaseLayout};
 use collab_define::CollabType;
+use futures::executor::block_on;
 use tokio::sync::RwLock;
+use tracing::{instrument, trace};
 
 use collab_integrate::collab_builder::AppFlowyCollabBuilder;
 use collab_integrate::{CollabPersistenceConfig, RocksCollabDB};
@@ -87,7 +89,7 @@ impl DatabaseManager {
 
     // If the workspace database not exist in disk, try to fetch from remote.
     if !self.is_collab_exist(uid, &collab_db, &database_views_aggregate_id) {
-      tracing::trace!("workspace database not exist, try to fetch from remote");
+      trace!("workspace database not exist, try to fetch from remote");
       match self
         .cloud_service
         .get_collab_update(&database_views_aggregate_id, CollabType::WorkspaceDatabase)
@@ -106,7 +108,7 @@ impl DatabaseManager {
     }
 
     // Construct the workspace database.
-    tracing::trace!("open workspace database: {}", &database_views_aggregate_id);
+    trace!("open workspace database: {}", &database_views_aggregate_id);
     let collab = collab_builder.build_collab_with_config(
       uid,
       &database_views_aggregate_id,
@@ -125,6 +127,7 @@ impl DatabaseManager {
     Ok(())
   }
 
+  #[instrument(level = "debug", skip_all, err)]
   pub async fn initialize_with_new_user(
     &self,
     user_id: i64,
@@ -170,7 +173,7 @@ impl DatabaseManager {
   }
 
   pub async fn open_database(&self, database_id: &str) -> FlowyResult<Arc<DatabaseEditor>> {
-    tracing::trace!("create new editor for database {}", database_id);
+    trace!("create new editor for database {}", database_id);
     let mut editors = self.editors.write().await;
 
     let wdb = self.get_workspace_database().await?;
@@ -353,7 +356,7 @@ fn subscribe_block_event(workspace_database: &WorkspaceDatabase) {
       match event {
         BlockEvent::DidFetchRow(row_details) => {
           for row_detail in row_details {
-            tracing::trace!("Did fetch row: {:?}", row_detail.row.id);
+            trace!("Did fetch row: {:?}", row_detail.row.id);
             let row_id = row_detail.row.id.clone();
             let pb = DidFetchRowPB::from(row_detail);
             send_notification(&row_id, DatabaseNotification::DidFetchRow)
@@ -426,16 +429,14 @@ impl DatabaseCollabService for UserDatabaseCollabServiceImpl {
     collab_raw_data: CollabRawData,
     config: &CollabPersistenceConfig,
   ) -> Arc<MutexCollab> {
-    self
-      .collab_builder
-      .build_with_config(
-        uid,
-        object_id,
-        object_type,
-        collab_db,
-        collab_raw_data,
-        config,
-      )
-      .unwrap()
+    block_on(self.collab_builder.build_with_config(
+      uid,
+      object_id,
+      object_type,
+      collab_db,
+      collab_raw_data,
+      config,
+    ))
+    .unwrap()
   }
 }

frontend/rust-lib/flowy-document2/src/manager.rs

@@ -8,6 +8,7 @@ use collab_document::document::Document;
 use collab_document::document_data::default_document_data;
 use collab_document::YrsDocAction;
 use parking_lot::RwLock;
+use tracing::instrument;
 
 use collab_integrate::collab_builder::AppFlowyCollabBuilder;
 use collab_integrate::RocksCollabDB;
@@ -55,6 +56,7 @@ impl DocumentManager {
     Ok(())
   }
 
+  #[instrument(level = "debug", skip_all, err)]
   pub async fn initialize_with_new_user(&self, uid: i64, workspace_id: String) -> FlowyResult<()> {
     self.initialize(uid, workspace_id).await?;
     Ok(())
@@ -175,8 +177,22 @@ impl DocumentManager {
     let db = self.user.collab_db(uid)?;
     let collab = self
       .collab_builder
-      .build(uid, doc_id, CollabType::Document, updates, db)?;
+      .build(uid, doc_id, CollabType::Document, updates, db)
+      .await?;
     Ok(collab)
+
+    // let doc_id = doc_id.to_string();
+    // let (tx, rx) = oneshot::channel();
+    // let collab_builder = self.collab_builder.clone();
+    // tokio::spawn(async move {
+    //   let collab = collab_builder
+    //     .build(uid, &doc_id, CollabType::Document, updates, db)
+    //     .await
+    //     .unwrap();
+    //   let _ = tx.send(collab);
+    // });
+    //
+    // Ok(rx.await.unwrap())
   }
 
   fn is_doc_exist(&self, doc_id: &str) -> FlowyResult<bool> {

frontend/rust-lib/flowy-error/src/code.rs

@@ -253,6 +253,9 @@ pub enum ErrorCode {
 
   #[error("Permission denied")]
   NotEnoughPermissions = 83,
+
+  #[error("Internal server error")]
+  InternalServerError = 84,
 }
 
 impl ErrorCode {

frontend/rust-lib/flowy-error/src/errors.rs

@@ -99,6 +99,7 @@ impl FlowyError {
     ErrorCode::UnexpectedCalendarFieldType
   );
   static_flowy_error!(collab_not_sync, ErrorCode::CollabDataNotSync);
+  static_flowy_error!(server_error, ErrorCode::InternalServerError);
 }
 
 impl std::convert::From<ErrorCode> for FlowyError {

frontend/rust-lib/flowy-error/src/impl_from/cloud.rs

@@ -8,12 +8,12 @@ impl From<AppError> for FlowyError {
       client_api::error::ErrorCode::Ok => ErrorCode::Internal,
       client_api::error::ErrorCode::Unhandled => ErrorCode::Internal,
       client_api::error::ErrorCode::RecordNotFound => ErrorCode::RecordNotFound,
+      client_api::error::ErrorCode::FileNotFound => ErrorCode::RecordNotFound,
       client_api::error::ErrorCode::RecordAlreadyExists => ErrorCode::RecordAlreadyExists,
       client_api::error::ErrorCode::InvalidEmail => ErrorCode::EmailFormatInvalid,
       client_api::error::ErrorCode::InvalidPassword => ErrorCode::PasswordFormatInvalid,
       client_api::error::ErrorCode::OAuthError => ErrorCode::UserUnauthorized,
       client_api::error::ErrorCode::MissingPayload => ErrorCode::MissingPayload,
-      client_api::error::ErrorCode::StorageError => ErrorCode::Internal,
       client_api::error::ErrorCode::OpenError => ErrorCode::Internal,
       client_api::error::ErrorCode::InvalidUrl => ErrorCode::InvalidURL,
       client_api::error::ErrorCode::InvalidRequestParams => ErrorCode::InvalidParams,

frontend/rust-lib/flowy-folder2/src/manager.rs

@@ -12,7 +12,7 @@ use collab_folder::core::{
 use parking_lot::{Mutex, RwLock};
 use tokio_stream::wrappers::WatchStream;
 use tokio_stream::StreamExt;
-use tracing::{event, Level};
+use tracing::{event, info, instrument, Level};
 
 use collab_integrate::collab_builder::AppFlowyCollabBuilder;
 use collab_integrate::{CollabPersistenceConfig, RocksCollabDB, YrsDocAction};
@@ -159,13 +159,17 @@ impl FolderManager {
         } => {
           let is_exist = is_exist_in_local_disk(&self.user, &workspace_id).unwrap_or(false);
           if is_exist {
-            let collab = self.collab_for_folder(uid, &workspace_id, collab_db, vec![])?;
+            let collab = self
+              .collab_for_folder(uid, &workspace_id, collab_db, vec![])
+              .await?;
             Folder::open(collab, Some(folder_notifier))
           } else if create_if_not_exist {
             let folder_data =
               DefaultFolderBuilder::build(uid, workspace_id.to_string(), &self.operation_handlers)
                 .await;
-            let collab = self.collab_for_folder(uid, &workspace_id, collab_db, vec![])?;
+            let collab = self
+              .collab_for_folder(uid, &workspace_id, collab_db, vec![])
+              .await?;
             Folder::create(collab, Some(folder_notifier), Some(folder_data))
           } else {
             return Err(FlowyError::new(
@@ -178,11 +182,15 @@ impl FolderManager {
           if raw_data.is_empty() {
             return Err(workspace_data_not_sync_error(uid, &workspace_id));
           }
-          let collab = self.collab_for_folder(uid, &workspace_id, collab_db, raw_data)?;
+          let collab = self
+            .collab_for_folder(uid, &workspace_id, collab_db, raw_data)
+            .await?;
           Folder::open(collab, Some(folder_notifier))
         },
         FolderInitializeDataSource::FolderData(folder_data) => {
-          let collab = self.collab_for_folder(uid, &workspace_id, collab_db, vec![])?;
+          let collab = self
+            .collab_for_folder(uid, &workspace_id, collab_db, vec![])
+            .await?;
           Folder::create(collab, Some(folder_notifier), Some(folder_data))
         },
       };
@@ -205,21 +213,24 @@ impl FolderManager {
     Ok(())
   }
 
-  fn collab_for_folder(
+  async fn collab_for_folder(
     &self,
     uid: i64,
     workspace_id: &str,
     collab_db: Weak<RocksCollabDB>,
     raw_data: CollabRawData,
   ) -> Result<Arc<MutexCollab>, FlowyError> {
-    let collab = self.collab_builder.build_with_config(
-      uid,
-      workspace_id,
-      CollabType::Folder,
-      collab_db,
-      raw_data,
-      &CollabPersistenceConfig::new().enable_snapshot(true),
-    )?;
+    let collab = self
+      .collab_builder
+      .build_with_config(
+        uid,
+        workspace_id,
+        CollabType::Folder,
+        collab_db,
+        raw_data,
+        &CollabPersistenceConfig::new().enable_snapshot(true),
+      )
+      .await?;
     Ok(collab)
   }
 
@@ -236,7 +247,7 @@ impl FolderManager {
       .get_folder_updates(workspace_id, user_id)
       .await?;
 
-    tracing::info!(
+    info!(
       "Get folder updates via {}, number of updates: {}",
       self.cloud_service.service_name(),
       folder_updates.len()
@@ -254,6 +265,7 @@ impl FolderManager {
 
   /// Initialize the folder for the new user.
   /// Using the [DefaultFolderBuilder] to create the default workspace for the new user.
+  #[instrument(level = "debug", skip_all, err)]
   pub async fn initialize_with_new_user(
     &self,
     user_id: i64,
@@ -263,29 +275,41 @@ impl FolderManager {
     workspace_id: &str,
   ) -> FlowyResult<()> {
     // Create the default workspace if the user is new
-    tracing::info!("initialize_when_sign_up: is_new: {}", is_new);
+    info!("initialize_when_sign_up: is_new: {}", is_new);
     if is_new {
       self.initialize(user_id, workspace_id, data_source).await?;
     } else {
       // The folder updates should not be empty, as the folder data is stored
       // when the user signs up for the first time.
-      let folder_updates = self
+      let result = self
         .cloud_service
         .get_folder_updates(workspace_id, user_id)
-        .await?;
-
-      tracing::info!(
-        "Get folder updates via {}, number of updates: {}",
-        self.cloud_service.service_name(),
-        folder_updates.len()
-      );
-      self
-        .initialize(
-          user_id,
-          workspace_id,
-          FolderInitializeDataSource::Cloud(folder_updates),
-        )
-        .await?;
+        .await
+        .map_err(FlowyError::from);
+
+      match result {
+        Ok(folder_updates) => {
+          info!(
+            "Get folder updates via {}, number of updates: {}",
+            self.cloud_service.service_name(),
+            folder_updates.len()
+          );
+          self
+            .initialize(
+              user_id,
+              workspace_id,
+              FolderInitializeDataSource::Cloud(folder_updates),
+            )
+            .await?;
+        },
+        Err(err) => {
+          if err.is_record_not_found() {
+            self.initialize(user_id, workspace_id, data_source).await?;
+          } else {
+            return Err(err);
+          }
+        },
+      }
     }
     Ok(())
   }

frontend/rust-lib/flowy-server-config/src/af_cloud_config.rs

@@ -2,39 +2,47 @@ use serde::{Deserialize, Serialize};
 
 use flowy_error::{ErrorCode, FlowyError};
 
-pub const AF_CLOUD_BASE_URL: &str = "AF_CLOUD_BASE_URL";
-pub const AF_CLOUD_WS_BASE_URL: &str = "AF_CLOUD_WS_BASE_URL";
-pub const AF_CLOUD_GOTRUE_URL: &str = "AF_GOTRUE_URL";
+pub const APPFLOWY_CLOUD_BASE_URL: &str = "APPFLOWY_CLOUD_BASE_URL";
+pub const APPFLOWY_CLOUD_WS_BASE_URL: &str = "APPFLOWY_CLOUD_WS_BASE_URL";
+pub const APPFLOWY_CLOUD_GOTRUE_URL: &str = "APPFLOWY_CLOUD_GOTRUE_URL";
 
 #[derive(Debug, Serialize, Deserialize, Clone, Default)]
 pub struct AFCloudConfiguration {
   pub base_url: String,
-  pub base_ws_url: String,
+  pub ws_base_url: String,
   pub gotrue_url: String,
 }
 
 impl AFCloudConfiguration {
   pub fn from_env() -> Result<Self, FlowyError> {
-    let base_url = std::env::var(AF_CLOUD_BASE_URL)
-      .map_err(|_| FlowyError::new(ErrorCode::InvalidAuthConfig, "Missing AF_CLOUD_BASE_URL"))?;
-
-    let base_ws_url = std::env::var(AF_CLOUD_WS_BASE_URL)
-      .map_err(|_| FlowyError::new(ErrorCode::InvalidAuthConfig, "Missing AF_CLOUD_WS_BASE_URL"))?;
-
-    let gotrue_url = std::env::var(AF_CLOUD_GOTRUE_URL)
+    let base_url = std::env::var(APPFLOWY_CLOUD_BASE_URL).map_err(|_| {
+      FlowyError::new(
+        ErrorCode::InvalidAuthConfig,
+        "Missing APPFLOWY_CLOUD_BASE_URL",
+      )
+    })?;
+
+    let ws_base_url = std::env::var(APPFLOWY_CLOUD_WS_BASE_URL).map_err(|_| {
+      FlowyError::new(
+        ErrorCode::InvalidAuthConfig,
+        "Missing APPFLOWY_CLOUD_WS_BASE_URL",
+      )
+    })?;
+
+    let gotrue_url = std::env::var(APPFLOWY_CLOUD_GOTRUE_URL)
       .map_err(|_| FlowyError::new(ErrorCode::InvalidAuthConfig, "Missing AF_CLOUD_GOTRUE_URL"))?;
 
     Ok(Self {
       base_url,
-      base_ws_url,
+      ws_base_url,
       gotrue_url,
     })
   }
 
   /// Write the configuration to the environment variables.
   pub fn write_env(&self) {
-    std::env::set_var(AF_CLOUD_BASE_URL, &self.base_url);
-    std::env::set_var(AF_CLOUD_WS_BASE_URL, &self.base_ws_url);
-    std::env::set_var(AF_CLOUD_GOTRUE_URL, &self.gotrue_url);
+    std::env::set_var(APPFLOWY_CLOUD_BASE_URL, &self.base_url);
+    std::env::set_var(APPFLOWY_CLOUD_WS_BASE_URL, &self.ws_base_url);
+    std::env::set_var(APPFLOWY_CLOUD_GOTRUE_URL, &self.gotrue_url);
   }
 }

frontend/rust-lib/flowy-server/src/af_cloud/impls/document.rs

@@ -5,6 +5,7 @@ use collab_define::CollabType;
 use collab_document::document::Document;
 
 use flowy_document_deps::cloud::*;
+use flowy_error::FlowyError;
 use lib_infra::future::FutureResult;
 
 use crate::af_cloud::AFServer;
@@ -23,7 +24,10 @@ where
         object_id: document_id.to_string(),
         collab_type: CollabType::Document,
       };
-      let data = try_get_client?.get_collab(params).await?;
+      let data = try_get_client?
+        .get_collab(params)
+        .await
+        .map_err(FlowyError::from)?;
       Ok(vec![data])
     })
   }
@@ -44,7 +48,10 @@ where
         object_id: document_id.clone(),
         collab_type: CollabType::Document,
       };
-      let updates = vec![try_get_client?.get_collab(params).await?];
+      let updates = vec![try_get_client?
+        .get_collab(params)
+        .await
+        .map_err(FlowyError::from)?];
       let document = Document::from_updates(CollabOrigin::Empty, updates, &document_id, vec![])?;
       Ok(document.get_document_data().ok())
     })

frontend/rust-lib/flowy-server/src/af_cloud/impls/folder.rs

@@ -3,6 +3,7 @@ use client_api::entity::QueryCollabParams;
 use collab::core::origin::CollabOrigin;
 use collab_define::CollabType;
 
+use flowy_error::FlowyError;
 use flowy_folder_deps::cloud::{Folder, FolderCloudService, FolderData, FolderSnapshot, Workspace};
 use lib_infra::future::FutureResult;
 
@@ -26,7 +27,10 @@ where
         object_id: workspace_id.clone(),
         collab_type: CollabType::Folder,
       };
-      let updates = vec![try_get_client?.get_collab(params).await?];
+      let updates = vec![try_get_client?
+        .get_collab(params)
+        .await
+        .map_err(FlowyError::from)?];
       let folder =
         Folder::from_collab_raw_data(CollabOrigin::Empty, updates, &workspace_id, vec![])?;
       Ok(folder.get_folder_data())
@@ -49,8 +53,11 @@ where
         object_id: workspace_id,
         collab_type: CollabType::Folder,
       };
-      let updates = vec![try_get_client?.get_collab(params).await?];
-      Ok(updates)
+      let update = try_get_client?
+        .get_collab(params)
+        .await
+        .map_err(FlowyError::from)?;
+      Ok(vec![update])
     })
   }
 

frontend/rust-lib/flowy-server/src/af_cloud/impls/user.rs

@@ -3,7 +3,9 @@ use std::sync::Arc;
 
 use anyhow::{anyhow, Error};
 use client_api::entity::dto::UserUpdateParams;
-use client_api::entity::{AFUserProfileView, AFWorkspace, AFWorkspaces, InsertCollabParams};
+use client_api::entity::{
+  AFUserProfileView, AFWorkspace, AFWorkspaces, InsertCollabParams, OAuthProvider,
+};
 use collab_define::CollabObject;
 
 use flowy_error::{ErrorCode, FlowyError};
@@ -54,7 +56,7 @@ where
     FutureResult::new(async move { Ok(try_get_client?.sign_out().await?) })
   }
 
-  fn generate_sign_in_callback_url(&self, email: &str) -> FutureResult<String, Error> {
+  fn generate_sign_in_url_with_email(&self, email: &str) -> FutureResult<String, Error> {
     let email = email.to_string();
     let try_get_client = self.server.try_get_client();
     FutureResult::new(async move {
@@ -62,7 +64,19 @@ where
       let admin_email = std::env::var("GOTRUE_ADMIN_EMAIL").unwrap();
       let admin_password = std::env::var("GOTRUE_ADMIN_PASSWORD").unwrap();
       let url = try_get_client?
-        .generate_sign_in_callback_url(&admin_email, &admin_password, &email)
+        .generate_sign_in_url_with_email(&admin_email, &admin_password, &email)
+        .await?;
+      Ok(url)
+    })
+  }
+
+  fn generate_oauth_url_with_provider(&self, provider: &str) -> FutureResult<String, Error> {
+    let provider = OAuthProvider::from(provider);
+    let try_get_client = self.server.try_get_client();
+    FutureResult::new(async move {
+      let provider = provider.ok_or(anyhow!("invalid provider"))?;
+      let url = try_get_client?
+        .generate_oauth_url_with_provider(&provider)
         .await?;
       Ok(url)
     })
@@ -196,7 +210,6 @@ where
     FutureResult::new(async move {
       let client = try_get_client?;
       let params = InsertCollabParams::new(
-        collab_object.uid,
         collab_object.object_id.clone(),
         collab_object.collab_type.clone(),
         data,
@@ -219,7 +232,7 @@ pub async fn user_sign_in_with_url(
   client: Arc<AFCloudClient>,
   params: AFCloudOAuthParams,
 ) -> Result<AuthResponse, FlowyError> {
-  let is_new_user = client.sign_in_url(&params.sign_in_url).await?;
+  let is_new_user = client.sign_in_with_url(&params.sign_in_url).await?;
   let (profile, af_workspaces) = tokio::try_join!(client.profile(), client.workspaces())?;
 
   let latest_workspace = to_user_workspace(

frontend/rust-lib/flowy-server/src/af_cloud/server.rs

@@ -7,7 +7,7 @@ use client_api::ws::{
   BusinessID, WSClient, WSClientConfig, WSConnectStateReceiver, WebSocketChannel,
 };
 use client_api::Client;
-use tokio::sync::RwLock;
+use tracing::info;
 
 use flowy_database_deps::cloud::DatabaseCloudService;
 use flowy_document_deps::cloud::DocumentCloudService;
@@ -33,7 +33,7 @@ pub struct AFCloudServer {
   enable_sync: AtomicBool,
   #[allow(dead_code)]
   device_id: Arc<parking_lot::RwLock<String>>,
-  ws_client: Arc<RwLock<WSClient>>,
+  ws_client: Arc<WSClient>,
 }
 
 impl AFCloudServer {
@@ -42,13 +42,8 @@ impl AFCloudServer {
     enable_sync: bool,
     device_id: Arc<parking_lot::RwLock<String>>,
   ) -> Self {
-    let http_client = reqwest::Client::new();
-    let api_client = client_api::Client::from(
-      http_client,
-      &config.base_url,
-      &config.base_ws_url,
-      &config.gotrue_url,
-    );
+    let api_client =
+      client_api::Client::new(&config.base_url, &config.ws_base_url, &config.gotrue_url);
     let token_state_rx = api_client.subscribe_token_state();
     let enable_sync = AtomicBool::new(enable_sync);
 
@@ -57,7 +52,7 @@ impl AFCloudServer {
       ping_per_secs: 8,
       retry_connect_per_pings: 5,
     });
-    let ws_client = Arc::new(RwLock::new(ws_client));
+    let ws_client = Arc::new(ws_client);
     let api_client = Arc::new(api_client);
 
     spawn_ws_conn(&device_id, token_state_rx, &ws_client, &api_client);
@@ -81,7 +76,7 @@ impl AFCloudServer {
 
 impl AppFlowyServer for AFCloudServer {
   fn set_enable_sync(&self, uid: i64, enable: bool) {
-    tracing::info!("{} cloud sync: {}", uid, enable);
+    info!("{} cloud sync: {}", uid, enable);
     self.enable_sync.store(enable, Ordering::SeqCst);
   }
   fn user_service(&self) -> Arc<dyn UserCloudService> {
@@ -115,14 +110,8 @@ impl AppFlowyServer for AFCloudServer {
         match weak_ws_client.upgrade() {
           None => Ok(None),
           Some(ws_client) => {
-            let channel = ws_client
-              .read()
-              .await
-              .subscribe(BusinessID::CollabId, object_id)
-              .await
-              .ok();
-            let connect_state_recv = ws_client.read().await.subscribe_connect_state().await;
-
+            let channel = ws_client.subscribe(BusinessID::CollabId, object_id).ok();
+            let connect_state_recv = ws_client.subscribe_connect_state();
             Ok(channel.map(|c| (c, connect_state_recv)))
           },
         }
@@ -145,7 +134,7 @@ impl AppFlowyServer for AFCloudServer {
 fn spawn_ws_conn(
   device_id: &Arc<parking_lot::RwLock<String>>,
   mut token_state_rx: TokenStateReceiver,
-  ws_client: &Arc<RwLock<WSClient>>,
+  ws_client: &Arc<WSClient>,
   api_client: &Arc<Client>,
 ) {
   let weak_device_id = Arc::downgrade(device_id);
@@ -154,7 +143,7 @@ fn spawn_ws_conn(
 
   tokio::spawn(async move {
     if let Some(ws_client) = weak_ws_client.upgrade() {
-      let mut state_recv = ws_client.read().await.subscribe_connect_state().await;
+      let mut state_recv = ws_client.subscribe_connect_state();
       while let Ok(state) = state_recv.recv().await {
         if !state.is_timeout() {
           continue;
@@ -166,8 +155,8 @@ fn spawn_ws_conn(
         {
           let device_id = device_id.read().clone();
           if let Ok(ws_addr) = api_client.ws_url(&device_id) {
-            tracing::info!("🟢WebSocket Reconnecting");
-            let _ = ws_client.write().await.connect(ws_addr).await;
+            info!("🟢WebSocket Reconnecting");
+            let _ = ws_client.connect(ws_addr).await;
           }
         }
       }
@@ -179,7 +168,7 @@ fn spawn_ws_conn(
   let weak_api_client = Arc::downgrade(api_client);
   tokio::spawn(async move {
     while let Ok(token_state) = token_state_rx.recv().await {
-      tracing::info!("🟢Token state: {:?}", token_state);
+      info!("🟢Token state: {:?}", token_state);
       match token_state {
         TokenState::Refresh => {
           if let (Some(api_client), Some(ws_client), Some(device_id)) = (
@@ -189,14 +178,14 @@ fn spawn_ws_conn(
           ) {
             let device_id = device_id.read().clone();
             if let Ok(ws_addr) = api_client.ws_url(&device_id) {
-              let _ = ws_client.write().await.connect(ws_addr).await;
+              let _ = ws_client.connect(ws_addr).await;
             }
           }
         },
         TokenState::Invalid => {
           if let Some(ws_client) = weak_ws_client.upgrade() {
-            tracing::info!("🟡WebSocket Disconnecting");
-            ws_client.write().await.disconnect().await;
+            info!("🟡WebSocket Disconnecting");
+            ws_client.disconnect().await;
           }
         },
       }

frontend/rust-lib/flowy-server/src/local_server/impls/user.rs

@@ -76,7 +76,7 @@ impl UserCloudService for LocalServerUserAuthServiceImpl {
     FutureResult::new(async { Ok(()) })
   }
 
-  fn generate_sign_in_callback_url(&self, _email: &str) -> FutureResult<String, Error> {
+  fn generate_sign_in_url_with_email(&self, _email: &str) -> FutureResult<String, Error> {
     FutureResult::new(async {
       Err(anyhow::anyhow!(
         "Can't generate callback url when using offline mode"
@@ -84,6 +84,10 @@ impl UserCloudService for LocalServerUserAuthServiceImpl {
     })
   }
 
+  fn generate_oauth_url_with_provider(&self, _provider: &str) -> FutureResult<String, Error> {
+    FutureResult::new(async { Err(anyhow::anyhow!("Can't oauth url when using offline mode")) })
+  }
+
   fn update_user(
     &self,
     _credential: UserCredentials,

frontend/rust-lib/flowy-server/src/supabase/api/user.rs

@@ -165,7 +165,7 @@ where
     FutureResult::new(async { Ok(()) })
   }
 
-  fn generate_sign_in_callback_url(&self, _email: &str) -> FutureResult<String, Error> {
+  fn generate_sign_in_url_with_email(&self, _email: &str) -> FutureResult<String, Error> {
     FutureResult::new(async {
       Err(anyhow::anyhow!(
         "Can't generate callback url when using supabase"
@@ -173,6 +173,14 @@ where
     })
   }
 
+  fn generate_oauth_url_with_provider(&self, _provider: &str) -> FutureResult<String, Error> {
+    FutureResult::new(async {
+      Err(anyhow::anyhow!(
+        "Can't generate oauth url when using supabase"
+      ))
+    })
+  }
+
   fn update_user(
     &self,
     _credential: UserCredentials,

frontend/rust-lib/flowy-server/tests/af_cloud_test/util.rs

@@ -23,18 +23,13 @@ pub fn af_cloud_server(config: AFCloudConfiguration) -> Arc<AFCloudServer> {
 }
 
 pub async fn generate_sign_in_url(user_email: &str, config: &AFCloudConfiguration) -> String {
-  let http_client = reqwest::Client::new();
-  let api_client = client_api::Client::from(
-    http_client,
-    &config.base_url,
-    &config.base_ws_url,
-    &config.gotrue_url,
-  );
+  let api_client =
+    client_api::Client::new(&config.base_url, &config.ws_base_url, &config.gotrue_url);
 
   let admin_email = std::env::var("GOTRUE_ADMIN_EMAIL").unwrap();
   let admin_password = std::env::var("GOTRUE_ADMIN_PASSWORD").unwrap();
   api_client
-    .generate_sign_in_callback_url(&admin_email, &admin_password, user_email)
+    .generate_sign_in_url_with_email(&admin_email, &admin_password, user_email)
     .await
     .unwrap()
 }

frontend/rust-lib/flowy-test/src/lib.rs

@@ -29,7 +29,7 @@ use flowy_notification::entities::SubscribeObject;
 use flowy_notification::{register_notification_sender, NotificationSender};
 use flowy_server::supabase::define::{USER_DEVICE_ID, USER_EMAIL, USER_SIGN_IN_URL, USER_UUID};
 use flowy_user::entities::{
-  AuthTypePB, OAuthCallbackRequestPB, OAuthCallbackResponsePB, OAuthPB, UpdateCloudConfigPB,
+  AuthTypePB, OauthSignInPB, SignInUrlPB, SignInUrlPayloadPB, UpdateCloudConfigPB,
   UserCloudConfigPB, UserProfilePB,
 };
 use flowy_user::errors::{FlowyError, FlowyResult};
@@ -169,13 +169,13 @@ impl FlowyCoreTest {
 
   pub async fn supabase_party_sign_up(&self) -> UserProfilePB {
     let map = third_party_sign_up_param(Uuid::new_v4().to_string());
-    let payload = OAuthPB {
+    let payload = OauthSignInPB {
       map,
       auth_type: AuthTypePB::Supabase,
     };
 
     EventBuilder::new(self.clone())
-      .event(OAuth)
+      .event(OauthSignIn)
       .payload(payload)
       .async_send()
       .await
@@ -198,28 +198,28 @@ impl FlowyCoreTest {
   }
 
   pub async fn af_cloud_sign_in_with_email(&self, email: &str) -> FlowyResult<UserProfilePB> {
-    let payload = OAuthCallbackRequestPB {
+    let payload = SignInUrlPayloadPB {
       email: email.to_string(),
       auth_type: AuthTypePB::AFCloud,
     };
     let sign_in_url = EventBuilder::new(self.clone())
-      .event(OAuthCallbackURL)
+      .event(GetSignInURL)
       .payload(payload)
       .async_send()
       .await
-      .try_parse::<OAuthCallbackResponsePB>()?
+      .try_parse::<SignInUrlPB>()?
       .sign_in_url;
 
     let mut map = HashMap::new();
     map.insert(USER_SIGN_IN_URL.to_string(), sign_in_url);
     map.insert(USER_DEVICE_ID.to_string(), uuid::Uuid::new_v4().to_string());
-    let payload = OAuthPB {
+    let payload = OauthSignInPB {
       map,
       auth_type: AuthTypePB::AFCloud,
     };
 
     let user_profile = EventBuilder::new(self.clone())
-      .event(OAuth)
+      .event(OauthSignIn)
       .payload(payload)
       .async_send()
       .await
@@ -240,13 +240,13 @@ impl FlowyCoreTest {
       USER_EMAIL.to_string(),
       email.unwrap_or_else(|| format!("{}@appflowy.io", nanoid!(10))),
     );
-    let payload = OAuthPB {
+    let payload = OauthSignInPB {
       map,
       auth_type: AuthTypePB::Supabase,
     };
 
     let user_profile = EventBuilder::new(self.clone())
-      .event(OAuth)
+      .event(OauthSignIn)
       .payload(payload)
       .async_send()
       .await

frontend/rust-lib/flowy-test/tests/user/supabase_test/auth_test.rs

@@ -14,7 +14,7 @@ use flowy_server::supabase::define::{USER_EMAIL, USER_UUID};
 use flowy_test::document::document_event::DocumentEventTest;
 use flowy_test::event_builder::EventBuilder;
 use flowy_test::FlowyCoreTest;
-use flowy_user::entities::{AuthTypePB, OAuthPB, UpdateUserProfilePayloadPB, UserProfilePB};
+use flowy_user::entities::{AuthTypePB, OauthSignInPB, UpdateUserProfilePayloadPB, UserProfilePB};
 use flowy_user::errors::ErrorCode;
 use flowy_user::event_map::UserEvent::*;
 
@@ -30,13 +30,13 @@ async fn third_party_sign_up_test() {
       USER_EMAIL.to_string(),
       format!("{}@appflowy.io", nanoid!(6)),
     );
-    let payload = OAuthPB {
+    let payload = OauthSignInPB {
       map,
       auth_type: AuthTypePB::Supabase,
     };
 
     let response = EventBuilder::new(test.clone())
-      .event(OAuth)
+      .event(OauthSignIn)
       .payload(payload)
       .async_send()
       .await
@@ -72,8 +72,8 @@ async fn third_party_sign_up_with_duplicated_uuid() {
     map.insert(USER_EMAIL.to_string(), email.clone());
 
     let response_1 = EventBuilder::new(test.clone())
-      .event(OAuth)
-      .payload(OAuthPB {
+      .event(OauthSignIn)
+      .payload(OauthSignInPB {
         map: map.clone(),
         auth_type: AuthTypePB::Supabase,
       })
@@ -83,8 +83,8 @@ async fn third_party_sign_up_with_duplicated_uuid() {
     dbg!(&response_1);
 
     let response_2 = EventBuilder::new(test.clone())
-      .event(OAuth)
-      .payload(OAuthPB {
+      .event(OauthSignIn)
+      .payload(OauthSignInPB {
         map: map.clone(),
         auth_type: AuthTypePB::Supabase,
       })

frontend/rust-lib/flowy-test/tests/user/supabase_test/workspace_test.rs

@@ -4,7 +4,7 @@ use flowy_folder2::entities::WorkspaceSettingPB;
 use flowy_folder2::event_map::FolderEvent::GetCurrentWorkspace;
 use flowy_server::supabase::define::{USER_EMAIL, USER_UUID};
 use flowy_test::{event_builder::EventBuilder, FlowyCoreTest};
-use flowy_user::entities::{AuthTypePB, OAuthPB, UserProfilePB};
+use flowy_user::entities::{AuthTypePB, OauthSignInPB, UserProfilePB};
 use flowy_user::event_map::UserEvent::*;
 
 use crate::util::*;
@@ -19,13 +19,13 @@ async fn initial_workspace_test() {
       USER_EMAIL.to_string(),
       format!("{}@gmail.com", uuid::Uuid::new_v4()),
     );
-    let payload = OAuthPB {
+    let payload = OauthSignInPB {
       map,
       auth_type: AuthTypePB::Supabase,
     };
 
     let _ = EventBuilder::new(test.clone())
-      .event(OAuth)
+      .event(OauthSignIn)
       .payload(payload)
       .async_send()
       .await

frontend/rust-lib/flowy-user-deps/src/cloud.rs

@@ -70,8 +70,15 @@ pub trait UserCloudService: Send + Sync + 'static {
   /// Sign out an account
   fn sign_out(&self, token: Option<String>) -> FutureResult<(), Error>;
 
-  /// Generate a sign in callback url for the user with the given email
-  fn generate_sign_in_callback_url(&self, email: &str) -> FutureResult<String, Error>;
+  /// Generate a sign in url for the user with the given email
+  fn generate_sign_in_url_with_email(&self, email: &str) -> FutureResult<String, Error>;
+
+  /// When the user opens the OAuth URL, it redirects to the corresponding provider's OAuth web page.
+  /// After the user is authenticated, the browser will open a deep link to the AppFlowy app (iOS, macOS, etc.),
+  /// which will call [Client::sign_in_with_url] to sign in.
+  ///
+  /// For example, the OAuth URL on Google looks like `https://appflowy.io/authorize?provider=google`.
+  fn generate_oauth_url_with_provider(&self, provider: &str) -> FutureResult<String, Error>;
 
   /// Using the user's token to update the user information
   fn update_user(

frontend/rust-lib/flowy-user/src/entities/auth.rs

@@ -79,7 +79,7 @@ impl TryInto<SignUpParams> for SignUpPayloadPB {
 }
 
 #[derive(ProtoBuf, Default)]
-pub struct OAuthPB {
+pub struct OauthSignInPB {
   /// Use this field to store the third party auth information.
   /// Different auth type has different fields.
   /// Supabase:
@@ -93,7 +93,7 @@ pub struct OAuthPB {
 }
 
 #[derive(ProtoBuf, Default)]
-pub struct OAuthCallbackRequestPB {
+pub struct SignInUrlPayloadPB {
   #[pb(index = 1)]
   pub email: String,
 
@@ -102,11 +102,77 @@ pub struct OAuthCallbackRequestPB {
 }
 
 #[derive(ProtoBuf, Default)]
-pub struct OAuthCallbackResponsePB {
+pub struct SignInUrlPB {
   #[pb(index = 1)]
   pub sign_in_url: String,
 }
 
+#[derive(ProtoBuf, Default)]
+pub struct OauthProviderPB {
+  #[pb(index = 1)]
+  pub provider: ProviderTypePB,
+}
+
+#[derive(ProtoBuf_Enum, Eq, PartialEq, Debug, Clone, Default)]
+pub enum ProviderTypePB {
+  Apple = 0,
+  Azure = 1,
+  Bitbucket = 2,
+  Discord = 3,
+  Facebook = 4,
+  Figma = 5,
+  Github = 6,
+  Gitlab = 7,
+  #[default]
+  Google = 8,
+  Keycloak = 9,
+  Kakao = 10,
+  Linkedin = 11,
+  Notion = 12,
+  Spotify = 13,
+  Slack = 14,
+  Workos = 15,
+  Twitch = 16,
+  Twitter = 17,
+  Email = 18,
+  Phone = 19,
+  Zoom = 20,
+}
+
+impl ProviderTypePB {
+  pub fn as_str(&self) -> &str {
+    match self {
+      ProviderTypePB::Apple => "apple",
+      ProviderTypePB::Azure => "azure",
+      ProviderTypePB::Bitbucket => "bitbucket",
+      ProviderTypePB::Discord => "discord",
+      ProviderTypePB::Facebook => "facebook",
+      ProviderTypePB::Figma => "figma",
+      ProviderTypePB::Github => "github",
+      ProviderTypePB::Gitlab => "gitlab",
+      ProviderTypePB::Google => "google",
+      ProviderTypePB::Keycloak => "keycloak",
+      ProviderTypePB::Kakao => "kakao",
+      ProviderTypePB::Linkedin => "linkedin",
+      ProviderTypePB::Notion => "notion",
+      ProviderTypePB::Spotify => "spotify",
+      ProviderTypePB::Slack => "slack",
+      ProviderTypePB::Workos => "workos",
+      ProviderTypePB::Twitch => "twitch",
+      ProviderTypePB::Twitter => "twitter",
+      ProviderTypePB::Email => "email",
+      ProviderTypePB::Phone => "phone",
+      ProviderTypePB::Zoom => "zoom",
+    }
+  }
+}
+
+#[derive(ProtoBuf, Default)]
+pub struct OauthProviderDataPB {
+  #[pb(index = 1)]
+  pub oauth_url: String,
+}
+
 #[derive(ProtoBuf_Enum, Eq, PartialEq, Debug, Clone)]
 pub enum AuthTypePB {
   Local = 0,

frontend/rust-lib/flowy-user/src/event_handler.rs

@@ -215,11 +215,9 @@ pub async fn get_user_setting(
   data_result_ok(user_setting)
 }
 
-/// Only used for third party auth.
-/// Use [UserEvent::SignIn] or [UserEvent::SignUp] If the [AuthType] is Local or SelfHosted
 #[tracing::instrument(level = "debug", skip(data, manager), err)]
 pub async fn oauth_handler(
-  data: AFPluginData<OAuthPB>,
+  data: AFPluginData<OauthSignInPB>,
   manager: AFPluginState<Weak<UserManager>>,
 ) -> DataResult<UserProfilePB, FlowyError> {
   let manager = upgrade_manager(manager)?;
@@ -230,20 +228,33 @@ pub async fn oauth_handler(
 }
 
 #[tracing::instrument(level = "debug", skip(data, manager), err)]
-pub async fn get_oauth_url_handler(
-  data: AFPluginData<OAuthCallbackRequestPB>,
+pub async fn get_sign_in_url_handler(
+  data: AFPluginData<SignInUrlPayloadPB>,
   manager: AFPluginState<Weak<UserManager>>,
-) -> DataResult<OAuthCallbackResponsePB, FlowyError> {
+) -> DataResult<SignInUrlPB, FlowyError> {
   let manager = upgrade_manager(manager)?;
   let params = data.into_inner();
   let auth_type: AuthType = params.auth_type.into();
   let sign_in_url = manager
-    .generate_sign_in_callback_url(&auth_type, &params.email)
+    .generate_sign_in_url_with_email(&auth_type, &params.email)
     .await?;
-  let resp = OAuthCallbackResponsePB { sign_in_url };
+  let resp = SignInUrlPB { sign_in_url };
   data_result_ok(resp)
 }
 
+#[tracing::instrument(level = "debug", skip_all, err)]
+pub async fn sign_in_with_provider_handler(
+  data: AFPluginData<OauthProviderPB>,
+  manager: AFPluginState<Weak<UserManager>>,
+) -> DataResult<OauthProviderDataPB, FlowyError> {
+  let manager = upgrade_manager(manager)?;
+  tracing::debug!("Sign in with provider: {:?}", data.provider.as_str());
+  let sign_in_url = manager.generate_oauth_url(data.provider.as_str()).await?;
+  data_result_ok(OauthProviderDataPB {
+    oauth_url: sign_in_url,
+  })
+}
+
 #[tracing::instrument(level = "debug", skip_all, err)]
 pub async fn set_encrypt_secret_handler(
   manager: AFPluginState<Weak<UserManager>>,

frontend/rust-lib/flowy-user/src/event_map.rs

@@ -37,8 +37,12 @@ pub fn init(user_session: Weak<UserManager>) -> AFPlugin {
     .event(UserEvent::GetCloudConfig, get_cloud_config_handler)
     .event(UserEvent::SetEncryptionSecret, set_encrypt_secret_handler)
     .event(UserEvent::CheckEncryptionSign, check_encrypt_secret_handler)
-    .event(UserEvent::OAuth, oauth_handler)
-    .event(UserEvent::OAuthCallbackURL, get_oauth_url_handler)
+    .event(UserEvent::OauthSignIn, oauth_handler)
+    .event(UserEvent::GetSignInURL, get_sign_in_url_handler)
+    .event(
+      UserEvent::GetOauthURLWithProvider,
+      sign_in_with_provider_handler,
+    )
     .event(
       UserEvent::GetAllUserWorkspaces,
       get_all_user_workspace_handler,
@@ -230,13 +234,16 @@ pub enum UserEvent {
   #[event(output = "UserSettingPB")]
   GetUserSetting = 9,
 
-  #[event(input = "OAuthPB", output = "UserProfilePB")]
-  OAuth = 10,
+  #[event(input = "OauthSignInPB", output = "UserProfilePB")]
+  OauthSignIn = 10,
 
   /// Get the OAuth callback url
   /// Only use when the [AuthType] is AFCloud
-  #[event(input = "OAuthCallbackRequestPB", output = "OAuthCallbackResponsePB")]
-  OAuthCallbackURL = 11,
+  #[event(input = "SignInUrlPayloadPB", output = "SignInUrlPB")]
+  GetSignInURL = 11,
+
+  #[event(input = "OauthProviderPB", output = "OauthProviderDataPB")]
+  GetOauthURLWithProvider = 12,
 
   #[event(input = "UpdateCloudConfigPB")]
   SetCloudConfig = 13,

frontend/rust-lib/flowy-user/src/manager.rs

@@ -4,6 +4,7 @@ use std::sync::{Arc, Weak};
 use collab_user::core::MutexUserAwareness;
 use serde_json::Value;
 use tokio::sync::{Mutex, RwLock};
+use tracing::{debug, info};
 use uuid::Uuid;
 
 use collab_integrate::collab_builder::AppFlowyCollabBuilder;
@@ -134,7 +135,7 @@ impl UserManager {
           {
             Ok(applied_migrations) => {
               if !applied_migrations.is_empty() {
-                tracing::info!("Did apply migrations: {:?}", applied_migrations);
+                info!("Did apply migrations: {:?}", applied_migrations);
               }
             },
             Err(e) => tracing::error!("User data migration failed: {:?}", e),
@@ -312,16 +313,16 @@ impl UserManager {
       UserAwarenessDataSource::Remote
     };
 
+    debug!("Sign up response: {:?}", response);
     if response.is_new_user {
       if let Some(old_user) = migration_user {
         let new_user = MigrationUser {
           user_profile: user_profile.clone(),
           session: new_session.clone(),
         };
-        tracing::info!(
+        info!(
           "Migrate old user data from {:?} to {:?}",
-          old_user.user_profile.uid,
-          new_user.user_profile.uid
+          old_user.user_profile.uid, new_user.user_profile.uid
         );
         self
           .migrate_local_user_to_cloud(&old_user, &new_user)
@@ -524,7 +525,7 @@ impl UserManager {
   }
 
   pub(crate) fn set_session(&self, session: Option<Session>) -> Result<(), FlowyError> {
-    tracing::debug!("Set current user: {:?}", session);
+    debug!("Set current user: {:?}", session);
     match &session {
       None => {
         self.current_session.write().take();
@@ -543,7 +544,7 @@ impl UserManager {
     Ok(())
   }
 
-  pub(crate) async fn generate_sign_in_callback_url(
+  pub(crate) async fn generate_sign_in_url_with_email(
     &self,
     auth_type: &AuthType,
     email: &str,
@@ -551,7 +552,22 @@ impl UserManager {
     self.update_auth_type(auth_type).await;
 
     let auth_service = self.cloud_services.get_user_service()?;
-    let url = auth_service.generate_sign_in_callback_url(email).await?;
+    let url = auth_service
+      .generate_sign_in_url_with_email(email)
+      .await
+      .map_err(|err| FlowyError::server_error().with_context(err))?;
+    Ok(url)
+  }
+
+  pub(crate) async fn generate_oauth_url(
+    &self,
+    oauth_provider: &str,
+  ) -> Result<String, FlowyError> {
+    self.update_auth_type(&AuthType::AFCloud).await;
+    let auth_service = self.cloud_services.get_user_service()?;
+    let url = auth_service
+      .generate_oauth_url_with_provider(oauth_provider)
+      .await?;
     Ok(url)
   }
 
@@ -588,7 +604,7 @@ impl UserManager {
   async fn handler_user_update(&self, user_update: UserUpdate) -> FlowyResult<()> {
     let session = self.get_session()?;
     if session.user_id == user_update.uid {
-      tracing::debug!("Receive user update: {:?}", user_update);
+      debug!("Receive user update: {:?}", user_update);
       let user_profile = self.get_user_profile(user_update.uid).await?;
 
       if !is_user_encryption_sign_valid(&user_profile, &user_update.encryption_sign) {

frontend/rust-lib/flowy-user/src/services/user_awareness.rs

@@ -1,9 +1,11 @@
 use std::sync::{Arc, Weak};
 
+use anyhow::Context;
 use collab::core::collab::{CollabRawData, MutexCollab};
 use collab_define::reminder::Reminder;
 use collab_define::CollabType;
 use collab_user::core::{MutexUserAwareness, UserAwareness};
+use tracing::{error, trace};
 
 use collab_integrate::RocksCollabDB;
 use flowy_error::{ErrorCode, FlowyError, FlowyResult};
@@ -101,12 +103,8 @@ impl UserManager {
     source: UserAwarenessDataSource,
   ) {
     match self.try_initial_user_awareness(session, source).await {
-      Ok(_) => {
-        tracing::trace!("User awareness initialized");
-      },
-      Err(e) => {
-        tracing::error!("Failed to initialize user awareness: {:?}", e);
-      },
+      Ok(_) => trace!("User awareness initialized"),
+      Err(e) => error!("Failed to initialize user awareness: {:?}", e),
     }
   }
 
@@ -128,11 +126,13 @@ impl UserManager {
     session: &Session,
     source: UserAwarenessDataSource,
   ) -> FlowyResult<()> {
-    tracing::trace!("Initializing user awareness from {:?}", source);
+    trace!("Initializing user awareness from {:?}", source);
     let collab_db = self.get_collab_db(session.user_id)?;
     let user_awareness = match source {
       UserAwarenessDataSource::Local => {
-        let collab = self.collab_for_user_awareness(session, collab_db, vec![])?;
+        let collab = self
+          .collab_for_user_awareness(session, collab_db, vec![])
+          .await?;
         MutexUserAwareness::new(UserAwareness::create(collab, None))
       },
       UserAwarenessDataSource::Remote => {
@@ -141,7 +141,10 @@ impl UserManager {
           .get_user_service()?
           .get_user_awareness_updates(session.user_id)
           .await?;
-        let collab = self.collab_for_user_awareness(session, collab_db, data)?;
+        trace!("Get user awareness collab: {}", data.len());
+        let collab = self
+          .collab_for_user_awareness(session, collab_db, data)
+          .await?;
         MutexUserAwareness::new(UserAwareness::create(collab, None))
       },
     };
@@ -154,7 +157,7 @@ impl UserManager {
   /// This function constructs a collaboration instance based on the given session and raw data,
   /// using a collaboration builder. This instance is specifically geared towards handling
   /// user awareness.
-  fn collab_for_user_awareness(
+  async fn collab_for_user_awareness(
     &self,
     session: &Session,
     collab_db: Weak<RocksCollabDB>,
@@ -164,13 +167,16 @@ impl UserManager {
       ErrorCode::Internal,
       "Unexpected error: collab builder is not available",
     ))?;
-    let collab = collab_builder.build(
-      session.user_id,
-      &session.user_id.to_string(),
-      CollabType::UserAwareness,
-      raw_data,
-      collab_db,
-    )?;
+    let collab = collab_builder
+      .build(
+        session.user_id,
+        &session.user_id.to_string(),
+        CollabType::UserAwareness,
+        raw_data,
+        collab_db,
+      )
+      .await
+      .context("Build collab for user awareness failed")?;
     Ok(collab)
   }
 

frontend/rust-lib/lib-dispatch/src/module/module.rs

@@ -1,17 +1,3 @@
-use crate::{
-  errors::{DispatchError, InternalError},
-  module::{container::AFPluginStateMap, AFPluginState},
-  request::{payload::Payload, AFPluginEventRequest, FromAFPluginRequest},
-  response::{AFPluginEventResponse, AFPluginResponder},
-  service::{
-    factory, AFPluginHandler, AFPluginHandlerService, AFPluginServiceFactory, BoxService,
-    BoxServiceFactory, Service, ServiceRequest, ServiceResponse,
-  },
-};
-use futures_core::future::BoxFuture;
-use futures_core::ready;
-use nanoid::nanoid;
-use pin_project::pin_project;
 use std::sync::Arc;
 use std::{
   collections::HashMap,
@@ -23,6 +9,23 @@ use std::{
   task::{Context, Poll},
 };
 
+use futures_core::future::BoxFuture;
+use futures_core::ready;
+use nanoid::nanoid;
+use pin_project::pin_project;
+
+use crate::service::AFPluginHandler;
+use crate::{
+  errors::{DispatchError, InternalError},
+  module::{container::AFPluginStateMap, AFPluginState},
+  request::{payload::Payload, AFPluginEventRequest, FromAFPluginRequest},
+  response::{AFPluginEventResponse, AFPluginResponder},
+  service::{
+    factory, AFPluginHandlerService, AFPluginServiceFactory, BoxService, BoxServiceFactory,
+    Service, ServiceRequest, ServiceResponse,
+  },
+};
+
 pub type AFPluginMap = Arc<HashMap<AFPluginEvent, Arc<AFPlugin>>>;
 pub(crate) fn as_plugin_map(plugins: Vec<AFPlugin>) -> AFPluginMap {
   let mut plugin_map = HashMap::new();
@@ -93,6 +96,7 @@ impl AFPlugin {
     self
   }
 
+  #[track_caller]
   pub fn event<E, H, T, R>(mut self, event: E, handler: H) -> Self
   where
     H: AFPluginHandler<T, R>,