feat: enable collaboration update synchronization between different devices (#3169)

* feat: bypass realtime event

* chore: use user device id

* chore: send realtime update

* chore: setup realtime recever

* chore: setup realtime recever

* chore: clippy

* chore: update collab rev

* chore: update realtime subscription

* chore: fix test

* chore: fmt

* test: fix flutter test

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

@@ -2,6 +2,7 @@ import 'dart:async';
 import 'dart:io';
 
 import 'package:appflowy/env/env.dart';
+import 'package:appflowy/user/application/supabase_realtime.dart';
 import 'package:appflowy/workspace/application/settings/application_data_storage.dart';
 import 'package:flutter/foundation.dart';
 import 'package:supabase_flutter/supabase_flutter.dart';
@@ -22,6 +23,7 @@ const hiveBoxName = 'appflowy_supabase_authentication';
 
 // Used to store the session of the supabase in case of the user switch the different folder.
 Supabase? supabase;
+SupbaseRealtimeService? realtimeService;
 
 class InitSupabaseTask extends LaunchTask {
   @override
@@ -37,12 +39,14 @@ class InitSupabaseTask extends LaunchTask {
 
     supabase?.dispose();
     supabase = null;
-    supabase = await Supabase.initialize(
+    final initializedSupabase = await Supabase.initialize(
       url: Env.supabaseUrl,
       anonKey: Env.supabaseAnonKey,
       debug: kDebugMode,
       localStorage: const SupabaseLocalStorage(),
     );
+    realtimeService = SupbaseRealtimeService(supabase: initializedSupabase);
+    supabase = initializedSupabase;
   }
 }
 

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

@@ -10,6 +10,7 @@ import 'package:appflowy_backend/protobuf/flowy-user/protobuf.dart'
     show SignInPayloadPB, SignUpPayloadPB, UserProfilePB;
 
 import '../../../generated/locale_keys.g.dart';
+import 'device_id.dart';
 
 class AppFlowyAuthService implements AuthService {
   @override
@@ -22,7 +23,8 @@ class AppFlowyAuthService implements AuthService {
     final request = SignInPayloadPB.create()
       ..email = email
       ..password = password
-      ..authType = authType;
+      ..authType = authType
+      ..deviceId = await getDeviceId();
     final response = UserEventSignIn(request).send();
     return response.then((value) => value.swap());
   }
@@ -39,7 +41,8 @@ class AppFlowyAuthService implements AuthService {
       ..name = name
       ..email = email
       ..password = password
-      ..authType = authType;
+      ..authType = authType
+      ..deviceId = await getDeviceId();
     final response = await UserEventSignUp(request).send().then(
           (value) => value.swap(),
         );

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

@@ -9,6 +9,7 @@ class AuthServiceMapKeys {
   // for supabase auth use only.
   static const String uuid = 'uuid';
   static const String email = 'email';
+  static const String deviceId = 'device_id';
 }
 
 abstract class AuthService {

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

@@ -0,0 +1,37 @@
+import 'dart:io';
+
+import 'package:appflowy/startup/startup.dart';
+import 'package:appflowy_backend/log.dart';
+import 'package:device_info_plus/device_info_plus.dart';
+import 'package:flutter/services.dart';
+
+final DeviceInfoPlugin deviceInfo = DeviceInfoPlugin();
+
+Future<String> getDeviceId() async {
+  if (integrationEnv().isTest) {
+    return "test_device_id";
+  }
+
+  String deviceId = "";
+  try {
+    if (Platform.isAndroid) {
+      final AndroidDeviceInfo androidInfo = await deviceInfo.androidInfo;
+      deviceId = androidInfo.device;
+    } else if (Platform.isIOS) {
+      final IosDeviceInfo iosInfo = await deviceInfo.iosInfo;
+      deviceId = iosInfo.identifierForVendor ?? "";
+    } else if (Platform.isMacOS) {
+      final MacOsDeviceInfo macInfo = await deviceInfo.macOsInfo;
+      deviceId = macInfo.systemGUID ?? "";
+    } else if (Platform.isWindows) {
+      final WindowsDeviceInfo windowsInfo = await deviceInfo.windowsInfo;
+      deviceId = windowsInfo.computerName;
+    } else if (Platform.isLinux) {
+      final LinuxDeviceInfo linuxInfo = await deviceInfo.linuxInfo;
+      deviceId = linuxInfo.machineId ?? "";
+    }
+  } on PlatformException {
+    Log.error('Failed to get platform version');
+  }
+  return deviceId;
+}

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

@@ -4,6 +4,7 @@ import 'package:appflowy/env/env.dart';
 import 'package:appflowy/startup/tasks/prelude.dart';
 import 'package:appflowy/user/application/auth/appflowy_auth_service.dart';
 import 'package:appflowy/user/application/auth/auth_service.dart';
+import 'package:appflowy/user/application/auth/device_id.dart';
 import 'package:appflowy/user/application/user_service.dart';
 import 'package:appflowy_backend/dispatch/dispatch.dart';
 import 'package:appflowy_backend/log.dart';
@@ -112,7 +113,8 @@ class SupabaseAuthService implements AuthService {
         return await setupAuth(
           map: {
             AuthServiceMapKeys.uuid: userId,
-            AuthServiceMapKeys.email: userEmail
+            AuthServiceMapKeys.email: userEmail,
+            AuthServiceMapKeys.deviceId: await getDeviceId()
           },
         );
       },
@@ -161,7 +163,8 @@ class SupabaseAuthService implements AuthService {
         return await setupAuth(
           map: {
             AuthServiceMapKeys.uuid: userId,
-            AuthServiceMapKeys.email: userEmail
+            AuthServiceMapKeys.email: userEmail,
+            AuthServiceMapKeys.deviceId: await getDeviceId()
           },
         );
       },

frontend/appflowy_flutter/lib/user/application/supabase_realtime.dart

@@ -0,0 +1,92 @@
+import 'dart:async';
+import 'dart:convert';
+
+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-user/protobuf.dart';
+import 'package:supabase_flutter/supabase_flutter.dart';
+
+/// A service to manage realtime interactions with Supabase.
+///
+/// `SupbaseRealtimeService` handles subscribing to table changes in Supabase
+/// based on the authentication state of a user. The service is initialized with
+/// a reference to a Supabase instance and sets up the necessary subscriptions
+/// accordingly.
+class SupbaseRealtimeService {
+  final Supabase supabase;
+  RealtimeChannel? channel;
+  StreamSubscription<AuthState>? authStateSubscription;
+
+  SupbaseRealtimeService({required this.supabase}) {
+    _subscribeAuthState();
+  }
+
+  void _subscribeAuthState() {
+    final auth = Supabase.instance.client.auth;
+    authStateSubscription = auth.onAuthStateChange.listen((state) async {
+      switch (state.event) {
+        case AuthChangeEvent.signedIn:
+          _subscribeTablesChanges();
+          break;
+        case AuthChangeEvent.signedOut:
+          channel?.unsubscribe();
+          break;
+        case AuthChangeEvent.tokenRefreshed:
+          _subscribeTablesChanges();
+          break;
+        default:
+          break;
+      }
+    });
+  }
+
+  Future<void> _subscribeTablesChanges() async {
+    final result = await UserBackendService.getCurrentUserProfile();
+    result.fold((l) => null, (userProfile) {
+      Log.info("Start listening to table changes");
+      // https://supabase.com/docs/guides/realtime/postgres-changes
+      final filters = [
+        "document",
+        "folder",
+        "database",
+        "database_row",
+        "w_database",
+      ].map(
+        (name) => ChannelFilter(
+          event: 'INSERT',
+          schema: 'public',
+          table: "af_collab_update_$name",
+          filter: 'uid=eq.${userProfile.id}',
+        ),
+      );
+
+      const ops = RealtimeChannelConfig(ack: true);
+      channel = supabase.client.channel("table-db-changes", opts: ops);
+      for (final filter in filters) {
+        channel?.on(
+          RealtimeListenTypes.postgresChanges,
+          filter,
+          (payload, [ref]) {
+            try {
+              final jsonStr = jsonEncode(payload);
+              Log.info("Realtime payload: $jsonStr");
+              final pb = RealtimePayloadPB.create()..jsonStr = jsonStr;
+              UserEventPushRealtimeEvent(pb).send();
+            } catch (e) {
+              Log.error(e);
+            }
+          },
+        );
+      }
+
+      channel?.subscribe(
+        (status, [err]) {
+          Log.info(
+            "subscribe channel statue: $status, err: $err",
+          );
+        },
+      );
+    });
+  }
+}

frontend/appflowy_tauri/src-tauri/Cargo.toml

@@ -34,13 +34,13 @@ default = ["custom-protocol"]
 custom-protocol = ["tauri/custom-protocol"]
 
 [patch.crates-io]
-collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "ba963f" }
-collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "ba963f" }
-collab-persistence = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "ba963f" }
-collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "ba963f" }
-collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "ba963f" }
-appflowy-integrate = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "ba963f" }
-collab-plugins = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "ba963f" }
+collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3881ba" }
+collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3881ba" }
+collab-persistence = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3881ba" }
+collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3881ba" }
+collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3881ba" }
+appflowy-integrate = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3881ba" }
+collab-plugins = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3881ba" }
 
 #collab = { path = "../../../../AppFlowy-Collab/collab" }
 #collab-folder = { path = "../../../../AppFlowy-Collab/collab-folder" }

frontend/rust-lib/Cargo.lock

@@ -96,7 +96,7 @@ checksum = "9c7d0618f0e0b7e8ff11427422b64564d5fb0be1940354bfe2e0529b18a9d9b8"
 [[package]]
 name = "appflowy-integrate"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=ba963f#ba963fa299d294e5b2cafd940b9eaa8520280b7b"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3881ba#3881bab021229020837ae65df604b9b87d0e8497"
 dependencies = [
  "anyhow",
  "collab",
@@ -187,324 +187,6 @@ version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa"
 
-[[package]]
-name = "aws-config"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bcdcf0d683fe9c23d32cf5b53c9918ea0a500375a9fb20109802552658e576c9"
-dependencies = [
- "aws-credential-types",
- "aws-http",
- "aws-sdk-sso",
- "aws-sdk-sts",
- "aws-smithy-async",
- "aws-smithy-client",
- "aws-smithy-http",
- "aws-smithy-http-tower",
- "aws-smithy-json",
- "aws-smithy-types",
- "aws-types",
- "bytes",
- "fastrand",
- "hex",
- "http",
- "hyper",
- "ring",
- "time 0.3.21",
- "tokio",
- "tower",
- "tracing",
- "zeroize",
-]
-
-[[package]]
-name = "aws-credential-types"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1fcdb2f7acbc076ff5ad05e7864bdb191ca70a6fd07668dc3a1a8bcd051de5ae"
-dependencies = [
- "aws-smithy-async",
- "aws-smithy-types",
- "fastrand",
- "tokio",
- "tracing",
- "zeroize",
-]
-
-[[package]]
-name = "aws-endpoint"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8cce1c41a6cfaa726adee9ebb9a56fcd2bbfd8be49fd8a04c5e20fd968330b04"
-dependencies = [
- "aws-smithy-http",
- "aws-smithy-types",
- "aws-types",
- "http",
- "regex",
- "tracing",
-]
-
-[[package]]
-name = "aws-http"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "aadbc44e7a8f3e71c8b374e03ecd972869eb91dd2bc89ed018954a52ba84bc44"
-dependencies = [
- "aws-credential-types",
- "aws-smithy-http",
- "aws-smithy-types",
- "aws-types",
- "bytes",
- "http",
- "http-body",
- "lazy_static",
- "percent-encoding",
- "pin-project-lite",
- "tracing",
-]
-
-[[package]]
-name = "aws-sdk-dynamodb"
-version = "0.27.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "67fb64867fe098cffee7e34352b01bbfa2beb3aa1b2ff0e0a7bf9ff293557852"
-dependencies = [
- "aws-credential-types",
- "aws-endpoint",
- "aws-http",
- "aws-sig-auth",
- "aws-smithy-async",
- "aws-smithy-client",
- "aws-smithy-http",
- "aws-smithy-http-tower",
- "aws-smithy-json",
- "aws-smithy-types",
- "aws-types",
- "bytes",
- "fastrand",
- "http",
- "regex",
- "tokio-stream",
- "tower",
- "tracing",
-]
-
-[[package]]
-name = "aws-sdk-sso"
-version = "0.28.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c8b812340d86d4a766b2ca73f740dfd47a97c2dff0c06c8517a16d88241957e4"
-dependencies = [
- "aws-credential-types",
- "aws-endpoint",
- "aws-http",
- "aws-sig-auth",
- "aws-smithy-async",
- "aws-smithy-client",
- "aws-smithy-http",
- "aws-smithy-http-tower",
- "aws-smithy-json",
- "aws-smithy-types",
- "aws-types",
- "bytes",
- "http",
- "regex",
- "tokio-stream",
- "tower",
- "tracing",
-]
-
-[[package]]
-name = "aws-sdk-sts"
-version = "0.28.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "265fac131fbfc188e5c3d96652ea90ecc676a934e3174eaaee523c6cec040b3b"
-dependencies = [
- "aws-credential-types",
- "aws-endpoint",
- "aws-http",
- "aws-sig-auth",
- "aws-smithy-async",
- "aws-smithy-client",
- "aws-smithy-http",
- "aws-smithy-http-tower",
- "aws-smithy-json",
- "aws-smithy-query",
- "aws-smithy-types",
- "aws-smithy-xml",
- "aws-types",
- "bytes",
- "http",
- "regex",
- "tower",
- "tracing",
-]
-
-[[package]]
-name = "aws-sig-auth"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3b94acb10af0c879ecd5c7bdf51cda6679a0a4f4643ce630905a77673bfa3c61"
-dependencies = [
- "aws-credential-types",
- "aws-sigv4",
- "aws-smithy-http",
- "aws-types",
- "http",
- "tracing",
-]
-
-[[package]]
-name = "aws-sigv4"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9d2ce6f507be68e968a33485ced670111d1cbad161ddbbab1e313c03d37d8f4c"
-dependencies = [
- "aws-smithy-http",
- "form_urlencoded",
- "hex",
- "hmac",
- "http",
- "once_cell",
- "percent-encoding",
- "regex",
- "sha2",
- "time 0.3.21",
- "tracing",
-]
-
-[[package]]
-name = "aws-smithy-async"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "13bda3996044c202d75b91afeb11a9afae9db9a721c6a7a427410018e286b880"
-dependencies = [
- "futures-util",
- "pin-project-lite",
- "tokio",
- "tokio-stream",
-]
-
-[[package]]
-name = "aws-smithy-client"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0a86aa6e21e86c4252ad6a0e3e74da9617295d8d6e374d552be7d3059c41cedd"
-dependencies = [
- "aws-smithy-async",
- "aws-smithy-http",
- "aws-smithy-http-tower",
- "aws-smithy-types",
- "bytes",
- "fastrand",
- "http",
- "http-body",
- "hyper",
- "hyper-rustls",
- "lazy_static",
- "pin-project-lite",
- "rustls",
- "tokio",
- "tower",
- "tracing",
-]
-
-[[package]]
-name = "aws-smithy-http"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2b3b693869133551f135e1f2c77cb0b8277d9e3e17feaf2213f735857c4f0d28"
-dependencies = [
- "aws-smithy-types",
- "bytes",
- "bytes-utils",
- "futures-core",
- "http",
- "http-body",
- "hyper",
- "once_cell",
- "percent-encoding",
- "pin-project-lite",
- "pin-utils",
- "tokio",
- "tokio-util",
- "tracing",
-]
-
-[[package]]
-name = "aws-smithy-http-tower"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3ae4f6c5798a247fac98a867698197d9ac22643596dc3777f0c76b91917616b9"
-dependencies = [
- "aws-smithy-http",
- "aws-smithy-types",
- "bytes",
- "http",
- "http-body",
- "pin-project-lite",
- "tower",
- "tracing",
-]
-
-[[package]]
-name = "aws-smithy-json"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "23f9f42fbfa96d095194a632fbac19f60077748eba536eb0b9fecc28659807f8"
-dependencies = [
- "aws-smithy-types",
-]
-
-[[package]]
-name = "aws-smithy-query"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "98819eb0b04020a1c791903533b638534ae6c12e2aceda3e6e6fba015608d51d"
-dependencies = [
- "aws-smithy-types",
- "urlencoding",
-]
-
-[[package]]
-name = "aws-smithy-types"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "16a3d0bf4f324f4ef9793b86a1701d9700fbcdbd12a846da45eed104c634c6e8"
-dependencies = [
- "base64-simd",
- "itoa",
- "num-integer",
- "ryu",
- "time 0.3.21",
-]
-
-[[package]]
-name = "aws-smithy-xml"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b1b9d12875731bd07e767be7baad95700c3137b56730ec9ddeedb52a5e5ca63b"
-dependencies = [
- "xmlparser",
-]
-
-[[package]]
-name = "aws-types"
-version = "0.55.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6dd209616cc8d7bfb82f87811a5c655dc97537f592689b18743bddf5dc5c4829"
-dependencies = [
- "aws-credential-types",
- "aws-smithy-async",
- "aws-smithy-client",
- "aws-smithy-http",
- "aws-smithy-types",
- "http",
- "rustc_version",
- "tracing",
-]
-
 [[package]]
 name = "axum"
 version = "0.6.15"
@@ -577,16 +259,6 @@ version = "0.21.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a4a4ddaa51a5bc52a6948f74c06d20aaaddb71924eab79b8c97a8c556e942d6a"
 
-[[package]]
-name = "base64-simd"
-version = "0.8.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "339abbe78e73178762e23bea9dfd08e697eb3f3301cd4be981c0f78ba5859195"
-dependencies = [
- "outref",
- "vsimd",
-]
-
 [[package]]
 name = "base64ct"
 version = "1.6.0"
@@ -751,16 +423,6 @@ dependencies = [
  "serde",
 ]
 
-[[package]]
-name = "bytes-utils"
-version = "0.1.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e47d3a8076e283f3acd27400535992edb3ba4b5bb72f8891ad8fbe7932a7d4b9"
-dependencies = [
- "bytes",
- "either",
-]
-
 [[package]]
 name = "bzip2"
 version = "0.4.4"
@@ -925,7 +587,7 @@ dependencies = [
 [[package]]
 name = "collab"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=ba963f#ba963fa299d294e5b2cafd940b9eaa8520280b7b"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3881ba#3881bab021229020837ae65df604b9b87d0e8497"
 dependencies = [
  "anyhow",
  "bytes",
@@ -943,7 +605,7 @@ dependencies = [
 [[package]]
 name = "collab-client-ws"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=ba963f#ba963fa299d294e5b2cafd940b9eaa8520280b7b"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3881ba#3881bab021229020837ae65df604b9b87d0e8497"
 dependencies = [
  "bytes",
  "collab-sync",
@@ -961,7 +623,7 @@ dependencies = [
 [[package]]
 name = "collab-database"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=ba963f#ba963fa299d294e5b2cafd940b9eaa8520280b7b"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3881ba#3881bab021229020837ae65df604b9b87d0e8497"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -988,7 +650,7 @@ dependencies = [
 [[package]]
 name = "collab-derive"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=ba963f#ba963fa299d294e5b2cafd940b9eaa8520280b7b"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3881ba#3881bab021229020837ae65df604b9b87d0e8497"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1000,7 +662,7 @@ dependencies = [
 [[package]]
 name = "collab-document"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=ba963f#ba963fa299d294e5b2cafd940b9eaa8520280b7b"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3881ba#3881bab021229020837ae65df604b9b87d0e8497"
 dependencies = [
  "anyhow",
  "collab",
@@ -1019,7 +681,7 @@ dependencies = [
 [[package]]
 name = "collab-folder"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=ba963f#ba963fa299d294e5b2cafd940b9eaa8520280b7b"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3881ba#3881bab021229020837ae65df604b9b87d0e8497"
 dependencies = [
  "anyhow",
  "chrono",
@@ -1039,7 +701,7 @@ dependencies = [
 [[package]]
 name = "collab-persistence"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=ba963f#ba963fa299d294e5b2cafd940b9eaa8520280b7b"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3881ba#3881bab021229020837ae65df604b9b87d0e8497"
 dependencies = [
  "bincode",
  "chrono",
@@ -1059,13 +721,10 @@ dependencies = [
 [[package]]
 name = "collab-plugins"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=ba963f#ba963fa299d294e5b2cafd940b9eaa8520280b7b"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3881ba#3881bab021229020837ae65df604b9b87d0e8497"
 dependencies = [
  "anyhow",
  "async-trait",
- "aws-config",
- "aws-credential-types",
- "aws-sdk-dynamodb",
  "collab",
  "collab-client-ws",
  "collab-persistence",
@@ -1082,6 +741,7 @@ dependencies = [
  "tokio-retry",
  "tokio-stream",
  "tracing",
+ "uuid",
  "y-sync",
  "yrs",
 ]
@@ -1089,7 +749,7 @@ dependencies = [
 [[package]]
 name = "collab-sync"
 version = "0.1.0"
-source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=ba963f#ba963fa299d294e5b2cafd940b9eaa8520280b7b"
+source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3881ba#3881bab021229020837ae65df604b9b87d0e8497"
 dependencies = [
  "bytes",
  "collab",
@@ -2459,9 +2119,7 @@ checksum = "1788965e61b367cd03a62950836d5cd41560c3577d90e40e0819373194d1661c"
 dependencies = [
  "http",
  "hyper",
- "log",
  "rustls",
- "rustls-native-certs",
  "tokio",
  "tokio-rustls",
 ]
@@ -3112,12 +2770,6 @@ dependencies = [
  "winapi",
 ]
 
-[[package]]
-name = "outref"
-version = "0.5.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4030760ffd992bef45b0ae3f10ce1aba99e33464c90d14dd7c039884963ddc7a"
-
 [[package]]
 name = "overload"
 version = "0.1.1"
@@ -4071,15 +3723,6 @@ version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2"
 
-[[package]]
-name = "rustc_version"
-version = "0.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bfa0f585226d2e68097d4f95d113b15b83a82e819ab25717ec0590d9584ef366"
-dependencies = [
- "semver",
-]
-
 [[package]]
 name = "rustix"
 version = "0.37.11"
@@ -4106,18 +3749,6 @@ dependencies = [
  "webpki",
 ]
 
-[[package]]
-name = "rustls-native-certs"
-version = "0.6.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a9aace74cb666635c918e9c12bc0d348266037aa8eb599b5cba565709a8dff00"
-dependencies = [
- "openssl-probe",
- "rustls-pemfile",
- "schannel",
- "security-framework",
-]
-
 [[package]]
 name = "rustls-pemfile"
 version = "1.0.2"
@@ -4227,12 +3858,6 @@ dependencies = [
  "libc",
 ]
 
-[[package]]
-name = "semver"
-version = "1.0.18"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b0293b4b29daaf487284529cc2f5675b8e57c61f70167ba415a463651fd6a918"
-
 [[package]]
 name = "serde"
 version = "1.0.175"
@@ -4641,7 +4266,6 @@ checksum = "8f3403384eaacbca9923fa06940178ac13e4edb725486d70e8e15881d0c836cc"
 dependencies = [
  "serde",
  "time-core",
- "time-macros",
 ]
 
 [[package]]
@@ -4650,15 +4274,6 @@ version = "0.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7300fbefb4dadc1af235a9cef3737cea692a9d97e1b9cbcd4ebdae6f8868e6fb"
 
-[[package]]
-name = "time-macros"
-version = "0.2.9"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "372950940a5f07bf38dbe211d7283c9e6d7327df53794992d293e534c733d09b"
-dependencies = [
- "time-core",
-]
-
 [[package]]
 name = "tinyvec"
 version = "1.6.0"
@@ -5161,12 +4776,6 @@ dependencies = [
  "percent-encoding",
 ]
 
-[[package]]
-name = "urlencoding"
-version = "2.1.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"
-
 [[package]]
 name = "utf-8"
 version = "0.7.6"
@@ -5217,12 +4826,6 @@ version = "0.9.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f"
 
-[[package]]
-name = "vsimd"
-version = "0.8.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5c3082ca00d5a5ef149bb8b555a72ae84c9c59f7250f013ac822ac2e49b19c64"
-
 [[package]]
 name = "walkdir"
 version = "2.3.3"
@@ -5563,12 +5166,6 @@ dependencies = [
  "winapi",
 ]
 
-[[package]]
-name = "xmlparser"
-version = "0.13.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4d25c75bf9ea12c4040a97f829154768bbbce366287e2dc044af160cd79a13fd"
-
 [[package]]
 name = "y-sync"
 version = "0.3.1"

frontend/rust-lib/Cargo.toml

@@ -38,12 +38,12 @@ opt-level = 3
 incremental = false
 
 [patch.crates-io]
-collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "ba963f" }
-collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "ba963f" }
-collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "ba963f" }
-collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "ba963f" }
-appflowy-integrate = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "ba963f" }
-collab-plugins = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "ba963f" }
+collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3881ba" }
+collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3881ba" }
+collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3881ba" }
+collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3881ba" }
+appflowy-integrate = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3881ba" }
+collab-plugins = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3881ba" }
 
 #collab = { path = "../AppFlowy-Collab/collab" }
 #collab-folder = { path = "../AppFlowy-Collab/collab-folder" }

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

@@ -3,8 +3,9 @@ use std::fmt::{Display, Formatter};
 use std::sync::{Arc, Weak};
 
 use appflowy_integrate::collab_builder::{CollabStorageProvider, CollabStorageType};
-use appflowy_integrate::{CollabType, RemoteCollabStorage, YrsDocAction};
-use parking_lot::RwLock;
+use appflowy_integrate::{CollabObject, CollabType, RemoteCollabStorage, YrsDocAction};
+use parking_lot::{Mutex, RwLock};
+use serde_json::Value;
 use serde_repr::*;
 
 use flowy_database_deps::cloud::*;
@@ -63,6 +64,7 @@ impl Display for ServerProviderType {
 pub struct AppFlowyServerProvider {
   config: AppFlowyCoreConfig,
   provider_type: RwLock<ServerProviderType>,
+  device_id: Mutex<String>,
   providers: RwLock<HashMap<ServerProviderType, Arc<dyn AppFlowyServer>>>,
   supabase_config: RwLock<Option<SupabaseConfiguration>>,
   store_preferences: Weak<StorePreferences>,
@@ -78,12 +80,17 @@ impl AppFlowyServerProvider {
     Self {
       config,
       provider_type: RwLock::new(provider_type),
+      device_id: Default::default(),
       providers: RwLock::new(HashMap::new()),
       supabase_config: RwLock::new(supabase_config),
       store_preferences,
     }
   }
 
+  pub fn set_sync_device(&self, device_id: &str) {
+    *self.device_id.lock() = device_id.to_string();
+  }
+
   pub fn provider_type(&self) -> ServerProviderType {
     self.provider_type.read().clone()
   }
@@ -127,6 +134,7 @@ impl AppFlowyServerProvider {
         Ok::<Arc<dyn AppFlowyServer>, FlowyError>(Arc::new(SupabaseServer::new(config)))
       },
     }?;
+    server.set_sync_device_id(&self.device_id.lock());
 
     self
       .providers
@@ -134,6 +142,13 @@ impl AppFlowyServerProvider {
       .insert(provider_type.clone(), server.clone());
     Ok(server)
   }
+
+  pub fn handle_realtime_event(&self, json: Value) {
+    let provider_type = self.provider_type.read().clone();
+    if let Some(server) = self.providers.read().get(&provider_type) {
+      server.handle_realtime_event(json);
+    }
+  }
 }
 
 impl UserCloudServiceProvider for AppFlowyServerProvider {
@@ -326,14 +341,18 @@ impl CollabStorageProvider for AppFlowyServerProvider {
     self.provider_type().into()
   }
 
-  fn get_storage(&self, storage_type: &CollabStorageType) -> Option<Arc<dyn RemoteCollabStorage>> {
+  fn get_storage(
+    &self,
+    collab_object: &CollabObject,
+    storage_type: &CollabStorageType,
+  ) -> Option<Arc<dyn RemoteCollabStorage>> {
     match storage_type {
       CollabStorageType::Local => None,
       CollabStorageType::AWS => None,
       CollabStorageType::Supabase => self
         .get_provider(&ServerProviderType::Supabase)
         .ok()
-        .and_then(|provider| provider.collab_storage()),
+        .and_then(|provider| provider.collab_storage(collab_object)),
     }
   }
 

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

@@ -10,6 +10,7 @@ use std::{
 };
 
 use appflowy_integrate::collab_builder::{AppFlowyCollabBuilder, CollabStorageType};
+use serde_json::Value;
 use tokio::sync::RwLock;
 
 use flowy_database2::DatabaseManager;
@@ -206,6 +207,7 @@ impl AppFlowyCore {
       folder_manager: folder_manager.clone(),
       database_manager: database_manager.clone(),
       document_manager: document_manager.clone(),
+      server_provider: server_provider.clone(),
       config: config.clone(),
     };
 
@@ -272,6 +274,7 @@ struct UserStatusCallbackImpl {
   folder_manager: Arc<FolderManager>,
   database_manager: Arc<DatabaseManager>,
   document_manager: Arc<DocumentManager>,
+  server_provider: Arc<AppFlowyServerProvider>,
   #[allow(dead_code)]
   config: AppFlowyCoreConfig,
 }
@@ -279,7 +282,12 @@ struct UserStatusCallbackImpl {
 impl UserStatusCallback for UserStatusCallbackImpl {
   fn auth_type_did_changed(&self, _auth_type: AuthType) {}
 
-  fn did_init(&self, user_id: i64, user_workspace: &UserWorkspace) -> Fut<FlowyResult<()>> {
+  fn did_init(
+    &self,
+    user_id: i64,
+    user_workspace: &UserWorkspace,
+    device_id: &str,
+  ) -> Fut<FlowyResult<()>> {
     let user_id = user_id.to_owned();
     let user_workspace = user_workspace.clone();
     let collab_builder = self.collab_builder.clone();
@@ -287,6 +295,9 @@ impl UserStatusCallback for UserStatusCallbackImpl {
     let database_manager = self.database_manager.clone();
     let document_manager = self.document_manager.clone();
 
+    self.server_provider.set_sync_device(device_id);
+    self.collab_builder.set_sync_device(device_id.to_owned());
+
     to_fut(async move {
       collab_builder.initialize(user_workspace.id.clone());
       folder_manager
@@ -306,7 +317,12 @@ impl UserStatusCallback for UserStatusCallbackImpl {
     })
   }
 
-  fn did_sign_in(&self, user_id: i64, user_workspace: &UserWorkspace) -> Fut<FlowyResult<()>> {
+  fn did_sign_in(
+    &self,
+    user_id: i64,
+    user_workspace: &UserWorkspace,
+    device_id: &str,
+  ) -> Fut<FlowyResult<()>> {
     let user_id = user_id.to_owned();
     let user_workspace = user_workspace.clone();
     let collab_builder = self.collab_builder.clone();
@@ -314,6 +330,9 @@ impl UserStatusCallback for UserStatusCallbackImpl {
     let database_manager = self.database_manager.clone();
     let document_manager = self.document_manager.clone();
 
+    self.server_provider.set_sync_device(device_id);
+    self.collab_builder.set_sync_device(device_id.to_owned());
+
     to_fut(async move {
       collab_builder.initialize(user_workspace.id.clone());
       folder_manager
@@ -338,6 +357,7 @@ impl UserStatusCallback for UserStatusCallbackImpl {
     context: SignUpContext,
     user_profile: &UserProfile,
     user_workspace: &UserWorkspace,
+    device_id: &str,
   ) -> Fut<FlowyResult<()>> {
     let user_profile = user_profile.clone();
     let collab_builder = self.collab_builder.clone();
@@ -345,6 +365,9 @@ impl UserStatusCallback for UserStatusCallbackImpl {
     let database_manager = self.database_manager.clone();
     let user_workspace = user_workspace.clone();
     let document_manager = self.document_manager.clone();
+
+    self.server_provider.set_sync_device(device_id);
+    self.collab_builder.set_sync_device(device_id.to_owned());
     to_fut(async move {
       collab_builder.initialize(user_workspace.id.clone());
       folder_manager
@@ -409,6 +432,10 @@ impl UserStatusCallback for UserStatusCallbackImpl {
   fn did_update_network(&self, reachable: bool) {
     self.collab_builder.update_network(reachable);
   }
+
+  fn receive_realtime_event(&self, json: Value) {
+    self.server_provider.handle_realtime_event(json);
+  }
 }
 
 impl From<ServerProviderType> for CollabStorageType {

frontend/rust-lib/flowy-document2/tests/document/util.rs

@@ -1,7 +1,7 @@
-use anyhow::Error;
 use std::ops::Deref;
 use std::sync::Arc;
 
+use anyhow::Error;
 use appflowy_integrate::collab_builder::{AppFlowyCollabBuilder, DefaultCollabStorageProvider};
 use appflowy_integrate::RocksCollabDB;
 use collab_document::blocks::DocumentData;
@@ -14,7 +14,6 @@ use tracing_subscriber::{fmt::Subscriber, util::SubscriberInitExt, EnvFilter};
 use flowy_document2::document::MutexDocument;
 use flowy_document2::manager::{DocumentManager, DocumentUser};
 use flowy_document_deps::cloud::*;
-
 use lib_infra::future::FutureResult;
 
 pub struct DocumentTest {
@@ -83,6 +82,7 @@ pub fn db() -> Arc<RocksCollabDB> {
 
 pub fn default_collab_builder() -> Arc<AppFlowyCollabBuilder> {
   let builder = AppFlowyCollabBuilder::new(DefaultCollabStorageProvider(), None);
+  builder.set_sync_device(uuid::Uuid::new_v4().to_string());
   Arc::new(builder)
 }
 

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

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
-use collab_plugins::cloud_storage::RemoteCollabStorage;
+use collab_plugins::cloud_storage::{CollabObject, RemoteCollabStorage};
+use serde_json::Value;
 
 use flowy_database_deps::cloud::DatabaseCloudService;
 use flowy_document_deps::cloud::DocumentCloudService;
@@ -16,9 +17,11 @@ pub mod util;
 
 pub trait AppFlowyServer: Send + Sync + 'static {
   fn enable_sync(&self, _enable: bool) {}
+  fn set_sync_device_id(&self, _device_id: &str) {}
   fn user_service(&self) -> Arc<dyn UserService>;
   fn folder_service(&self) -> Arc<dyn FolderCloudService>;
   fn database_service(&self) -> Arc<dyn DatabaseCloudService>;
   fn document_service(&self) -> Arc<dyn DocumentCloudService>;
-  fn collab_storage(&self) -> Option<Arc<dyn RemoteCollabStorage>>;
+  fn collab_storage(&self, collab_object: &CollabObject) -> Option<Arc<dyn RemoteCollabStorage>>;
+  fn handle_realtime_event(&self, _json: Value) {}
 }

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

@@ -1,6 +1,6 @@
-use anyhow::Error;
 use std::sync::Arc;
 
+use anyhow::Error;
 use lazy_static::lazy_static;
 use parking_lot::Mutex;
 
@@ -42,6 +42,7 @@ impl UserService for LocalServerUserAuthServiceImpl {
         is_new: true,
         email: Some(params.email),
         token: None,
+        device_id: params.device_id,
       })
     })
   }
@@ -50,10 +51,7 @@ impl UserService for LocalServerUserAuthServiceImpl {
     let db = self.db.clone();
     FutureResult::new(async move {
       let params: SignInParams = params.unbox_or_error::<SignInParams>()?;
-      let uid = match params.uid {
-        None => ID_GEN.lock().next_id(),
-        Some(uid) => uid,
-      };
+      let uid = ID_GEN.lock().next_id();
 
       let user_workspace = db
         .get_user_workspace(uid)?
@@ -65,6 +63,7 @@ impl UserService for LocalServerUserAuthServiceImpl {
         user_workspaces: vec![user_workspace],
         email: Some(params.email),
         token: None,
+        device_id: params.device_id,
       })
     })
   }

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

@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use collab_plugins::cloud_storage::RemoteCollabStorage;
+use collab_plugins::cloud_storage::{CollabObject, RemoteCollabStorage};
 use parking_lot::RwLock;
 use tokio::sync::mpsc;
 
@@ -68,7 +68,7 @@ impl AppFlowyServer for LocalServer {
     Arc::new(LocalServerDocumentCloudServiceImpl())
   }
 
-  fn collab_storage(&self) -> Option<Arc<dyn RemoteCollabStorage>> {
+  fn collab_storage(&self, _collab_object: &CollabObject) -> Option<Arc<dyn RemoteCollabStorage>> {
     None
   }
 }

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

@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use collab_plugins::cloud_storage::RemoteCollabStorage;
+use collab_plugins::cloud_storage::{CollabObject, RemoteCollabStorage};
 
 use flowy_database_deps::cloud::DatabaseCloudService;
 use flowy_document_deps::cloud::DocumentCloudService;
@@ -41,7 +41,7 @@ impl AppFlowyServer for SelfHostServer {
     Arc::new(SelfHostedDocumentCloudServiceImpl())
   }
 
-  fn collab_storage(&self) -> Option<Arc<dyn RemoteCollabStorage>> {
+  fn collab_storage(&self, _collab_object: &CollabObject) -> Option<Arc<dyn RemoteCollabStorage>> {
     None
   }
 }

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

@@ -8,6 +8,7 @@ use collab_plugins::cloud_storage::{
   CollabObject, MsgId, RemoteCollabSnapshot, RemoteCollabState, RemoteCollabStorage,
   RemoteUpdateReceiver,
 };
+use parking_lot::Mutex;
 use tokio::task::spawn_blocking;
 
 use lib_infra::async_trait::async_trait;
@@ -17,15 +18,23 @@ use crate::supabase::api::request::{
   create_snapshot, get_latest_snapshot_from_server, get_updates_from_server,
   FetchObjectUpdateAction, UpdateItem,
 };
-use crate::supabase::api::util::{ExtendedResponse, InsertParamsBuilder};
+use crate::supabase::api::util::{
+  ExtendedResponse, InsertParamsBuilder, SupabaseBinaryColumnEncoder,
+};
 use crate::supabase::api::{PostgresWrapper, SupabaseServerService};
 use crate::supabase::define::*;
 
-pub struct SupabaseCollabStorageImpl<T>(T);
+pub struct SupabaseCollabStorageImpl<T> {
+  server: T,
+  rx: Mutex<Option<RemoteUpdateReceiver>>,
+}
 
 impl<T> SupabaseCollabStorageImpl<T> {
-  pub fn new(server: T) -> Self {
-    Self(server)
+  pub fn new(server: T, rx: Option<RemoteUpdateReceiver>) -> Self {
+    Self {
+      server,
+      rx: Mutex::new(rx),
+    }
   }
 }
 
@@ -39,21 +48,22 @@ where
   }
 
   async fn get_all_updates(&self, object: &CollabObject) -> Result<Vec<Vec<u8>>, Error> {
-    let postgrest = self.0.try_get_weak_postgrest()?;
-    let action = FetchObjectUpdateAction::new(object.id.clone(), object.ty.clone(), postgrest);
+    let postgrest = self.server.try_get_weak_postgrest()?;
+    let action =
+      FetchObjectUpdateAction::new(object.object_id.clone(), object.ty.clone(), postgrest);
     let updates = action.run().await?;
     Ok(updates)
   }
 
   async fn get_latest_snapshot(&self, object_id: &str) -> Option<RemoteCollabSnapshot> {
-    let postgrest = self.0.try_get_postgrest().ok()?;
+    let postgrest = self.server.try_get_postgrest().ok()?;
     get_latest_snapshot_from_server(object_id, postgrest)
       .await
       .ok()?
   }
 
   async fn get_collab_state(&self, object_id: &str) -> Result<Option<RemoteCollabState>, Error> {
-    let postgrest = self.0.try_get_postgrest()?;
+    let postgrest = self.server.try_get_postgrest()?;
     let json = postgrest
       .from("af_collab_state")
       .select("*")
@@ -92,7 +102,7 @@ where
   }
 
   async fn create_snapshot(&self, object: &CollabObject, snapshot: Vec<u8>) -> Result<i64, Error> {
-    let postgrest = self.0.try_get_postgrest()?;
+    let postgrest = self.server.try_get_postgrest()?;
     create_snapshot(&postgrest, object, snapshot).await
   }
 
@@ -102,7 +112,7 @@ where
     _id: MsgId,
     update: Vec<u8>,
   ) -> Result<(), Error> {
-    if let Some(postgrest) = self.0.get_postgrest() {
+    if let Some(postgrest) = self.server.get_postgrest() {
       let workspace_id = object
         .get_workspace_id()
         .ok_or(anyhow::anyhow!("Invalid workspace id"))?;
@@ -118,12 +128,13 @@ where
     _id: MsgId,
     init_update: Vec<u8>,
   ) -> Result<(), Error> {
-    let postgrest = self.0.try_get_postgrest()?;
+    let postgrest = self.server.try_get_postgrest()?;
     let workspace_id = object
       .get_workspace_id()
       .ok_or(anyhow::anyhow!("Invalid workspace id"))?;
 
-    let update_items = get_updates_from_server(&object.id, &object.ty, postgrest.clone()).await?;
+    let update_items =
+      get_updates_from_server(&object.object_id, &object.ty, postgrest.clone()).await?;
 
     // If the update_items is empty, we can send the init_update directly
     if update_items.is_empty() {
@@ -132,14 +143,12 @@ where
       // 2.Merge the updates into one and then delete the merged updates
       let merge_result = spawn_blocking(move || merge_updates(update_items, init_update)).await??;
       tracing::trace!("Merged updates count: {}", merge_result.merged_keys.len());
-      let override_key = merge_result.merged_keys.last().cloned().unwrap();
 
       let value_size = merge_result.new_update.len() as i32;
       let md5 = md5(&merge_result.new_update);
       let new_update = format!("\\x{}", hex::encode(merge_result.new_update));
       let params = InsertParamsBuilder::new()
-        .insert("oid", object.id.clone())
-        .insert("new_key", override_key)
+        .insert("oid", object.object_id.clone())
         .insert("new_value", new_update)
         .insert("md5", md5)
         .insert("value_size", value_size)
@@ -147,10 +156,11 @@ where
         .insert("uid", object.uid)
         .insert("workspace_id", workspace_id)
         .insert("removed_keys", merge_result.merged_keys)
+        .insert("did", object.get_device_id())
         .build();
 
       postgrest
-        .rpc("flush_collab_updates", params)
+        .rpc("flush_collab_updates_v2", params)
         .execute()
         .await?
         .success()
@@ -159,8 +169,12 @@ where
     Ok(())
   }
 
-  async fn subscribe_remote_updates(&self, _object: &CollabObject) -> Option<RemoteUpdateReceiver> {
-    todo!()
+  fn subscribe_remote_updates(&self, _object: &CollabObject) -> Option<RemoteUpdateReceiver> {
+    let rx = self.rx.lock().take();
+    if rx.is_none() {
+      tracing::warn!("The receiver is already taken");
+    }
+    rx
   }
 }
 
@@ -172,14 +186,15 @@ async fn send_update(
 ) -> Result<(), Error> {
   let value_size = update.len() as i32;
   let md5 = md5(&update);
-  let update = format!("\\x{}", hex::encode(update));
+  let update = SupabaseBinaryColumnEncoder::encode(update);
   let builder = InsertParamsBuilder::new()
-    .insert("oid", object.id.clone())
+    .insert("oid", object.object_id.clone())
     .insert("partition_key", partition_key(&object.ty))
     .insert("value", update)
     .insert("uid", object.uid)
     .insert("md5", md5)
     .insert("workspace_id", workspace_id)
+    .insert("did", object.get_device_id())
     .insert("value_size", value_size);
 
   let params = builder.build();

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

@@ -12,4 +12,4 @@ mod folder;
 mod postgres_server;
 mod request;
 mod user;
-mod util;
+pub mod util;

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

@@ -15,7 +15,9 @@ use tokio_retry::{Action, Condition, RetryIf};
 use flowy_database_deps::cloud::{CollabObjectUpdate, CollabObjectUpdateByOid};
 use lib_infra::util::md5;
 
-use crate::supabase::api::util::{ExtendedResponse, InsertParamsBuilder};
+use crate::supabase::api::util::{
+  ExtendedResponse, InsertParamsBuilder, SupabaseBinaryColumnDecoder,
+};
 use crate::supabase::api::PostgresWrapper;
 use crate::supabase::define::*;
 
@@ -127,7 +129,7 @@ pub async fn create_snapshot(
     .from(AF_COLLAB_SNAPSHOT_TABLE)
     .insert(
       InsertParamsBuilder::new()
-        .insert(AF_COLLAB_SNAPSHOT_OID_COLUMN, object.id.clone())
+        .insert(AF_COLLAB_SNAPSHOT_OID_COLUMN, object.object_id.clone())
         .insert("name", object.ty.to_string())
         .insert(AF_COLLAB_SNAPSHOT_BLOB_COLUMN, snapshot)
         .insert(AF_COLLAB_SNAPSHOT_BLOB_SIZE_COLUMN, value_size)
@@ -168,7 +170,7 @@ pub async fn get_latest_snapshot_from_server(
       let blob = value
         .get("blob")
         .and_then(|blob| blob.as_str())
-        .and_then(decode_hex_string)?;
+        .and_then(SupabaseBinaryColumnDecoder::decode)?;
       let sid = value.get("sid").and_then(|id| id.as_i64())?;
       let created_at = value.get("created_at").and_then(|created_at| {
         created_at
@@ -272,7 +274,7 @@ fn parser_update_from_json(json: &Value) -> Result<UpdateItem, Error> {
   let some_record = json
     .get("value")
     .and_then(|value| value.as_str())
-    .and_then(decode_hex_string);
+    .and_then(SupabaseBinaryColumnDecoder::decode);
 
   let some_key = json.get("key").and_then(|value| value.as_i64());
   if let (Some(value), Some(key)) = (some_record, some_key) {
@@ -301,11 +303,6 @@ pub struct UpdateItem {
   pub value: Vec<u8>,
 }
 
-fn decode_hex_string(s: &str) -> Option<Vec<u8>> {
-  let s = s.strip_prefix("\\x")?;
-  hex::decode(s).ok()
-}
-
 pub struct RetryCondition(Weak<PostgresWrapper>);
 impl Condition<anyhow::Error> for RetryCondition {
   fn should_retry(&mut self, _error: &anyhow::Error) -> bool {

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

@@ -89,6 +89,7 @@ where
         is_new: is_new_user,
         email: Some(user_profile.email),
         token: None,
+        device_id: params.device_id,
       })
     })
   }
@@ -115,6 +116,7 @@ where
         user_workspaces,
         email: None,
         token: None,
+        device_id: params.device_id,
       })
     })
   }

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

@@ -5,15 +5,14 @@ use serde_json::Value;
 use flowy_error::{ErrorCode, FlowyError};
 use lib_infra::future::{to_fut, Fut};
 
+#[derive(Default)]
 pub struct InsertParamsBuilder {
   map: serde_json::Map<String, Value>,
 }
 
 impl InsertParamsBuilder {
   pub fn new() -> Self {
-    Self {
-      map: serde_json::Map::new(),
-    }
+    Self::default()
   }
 
   pub fn insert<T: serde::Serialize>(mut self, key: &str, value: T) -> Self {
@@ -126,3 +125,60 @@ async fn parse_response_as_error(response: Response) -> FlowyError {
     ),
   )
 }
+/// An encoder for binary columns in Supabase.
+///
+/// Provides utilities to encode binary data into a format suitable for Supabase columns.
+pub struct SupabaseBinaryColumnEncoder;
+
+impl SupabaseBinaryColumnEncoder {
+  /// Encodes the given binary data into a Supabase-friendly string representation.
+  ///
+  /// # Parameters
+  /// - `value`: The binary data to encode.
+  ///
+  /// # Returns
+  /// Returns the encoded string in the format: `\\xHEX_ENCODED_STRING`
+  pub fn encode<T: AsRef<[u8]>>(value: T) -> String {
+    format!("\\x{}", hex::encode(value))
+  }
+}
+
+/// A decoder for binary columns in Supabase.
+///
+/// Provides utilities to decode a string from Supabase columns back into binary data.
+pub struct SupabaseBinaryColumnDecoder;
+
+impl SupabaseBinaryColumnDecoder {
+  /// Decodes a Supabase binary column string into binary data.
+  ///
+  /// # Parameters
+  /// - `value`: The string representation from a Supabase binary column.
+  ///
+  /// # Returns
+  /// Returns an `Option` containing the decoded binary data if decoding is successful.
+  /// Otherwise, returns `None`.
+  pub fn decode<T: AsRef<str>>(value: T) -> Option<Vec<u8>> {
+    let s = value.as_ref().strip_prefix("\\x")?;
+    hex::decode(s).ok()
+  }
+}
+
+/// A decoder specifically tailored for realtime event binary columns in Supabase.
+///
+/// Decodes the realtime event binary column data using the standard Supabase binary column decoder.
+pub struct SupabaseRealtimeEventBinaryColumnDecoder;
+
+impl SupabaseRealtimeEventBinaryColumnDecoder {
+  /// Decodes a realtime event binary column string from Supabase into binary data.
+  ///
+  /// # Parameters
+  /// - `value`: The string representation from a Supabase realtime event binary column.
+  ///
+  /// # Returns
+  /// Returns an `Option` containing the decoded binary data if decoding is successful.
+  /// Otherwise, returns `None`.
+  pub fn decode<T: AsRef<str>>(value: T) -> Option<Vec<u8>> {
+    let bytes = SupabaseBinaryColumnDecoder::decode(value)?;
+    hex::decode(bytes).ok()
+  }
+}

frontend/rust-lib/flowy-server/src/supabase/define.rs

@@ -1,4 +1,4 @@
-use collab_plugins::cloud_storage::CollabType;
+pub use collab_plugins::cloud_storage::CollabType;
 
 pub const AF_COLLAB_UPDATE_TABLE: &str = "af_collab_update";
 pub const AF_COLLAB_KEY_COLUMN: &str = "key";

frontend/rust-lib/flowy-server/src/supabase/entities.rs

@@ -1,6 +1,11 @@
-use serde::Deserialize;
+use std::fmt;
+use std::fmt::Display;
+
+use serde::de::{Error, Visitor};
+use serde::{Deserialize, Deserializer};
 use uuid::Uuid;
 
+use crate::supabase::api::util::SupabaseRealtimeEventBinaryColumnDecoder;
 use crate::util::deserialize_null_or_default;
 
 pub enum GetUserProfileParams {
@@ -30,3 +35,63 @@ pub(crate) struct UidResponse {
   #[allow(dead_code)]
   pub uid: i64,
 }
+
+#[derive(Debug, Deserialize)]
+pub struct RealtimeCollabUpdateEvent {
+  pub schema: String,
+  pub table: String,
+  #[serde(rename = "eventType")]
+  pub event_type: String,
+  #[serde(rename = "new")]
+  pub payload: RealtimeCollabUpdate,
+}
+
+impl Display for RealtimeCollabUpdateEvent {
+  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+    write!(
+      f,
+      "schema: {}, table: {}, event_type: {}",
+      self.schema, self.table, self.event_type
+    )
+  }
+}
+
+#[derive(Debug, Deserialize)]
+pub struct RealtimeCollabUpdate {
+  pub oid: String,
+  pub uid: i64,
+  pub key: i64,
+  pub did: String,
+  #[serde(deserialize_with = "deserialize_value")]
+  pub value: Vec<u8>,
+}
+
+pub fn deserialize_value<'de, D>(deserializer: D) -> Result<Vec<u8>, D::Error>
+where
+  D: Deserializer<'de>,
+{
+  struct ValueVisitor();
+
+  impl<'de> Visitor<'de> for ValueVisitor {
+    type Value = Vec<u8>;
+
+    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+      formatter.write_str("Expect NodeBody")
+    }
+
+    fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
+    where
+      E: Error,
+    {
+      Ok(SupabaseRealtimeEventBinaryColumnDecoder::decode(v).unwrap_or_default())
+    }
+
+    fn visit_string<E>(self, v: String) -> Result<Self::Value, E>
+    where
+      E: Error,
+    {
+      Ok(SupabaseRealtimeEventBinaryColumnDecoder::decode(v).unwrap_or_default())
+    }
+  }
+  deserializer.deserialize_any(ValueVisitor())
+}

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

@@ -1,7 +1,9 @@
+use std::collections::HashMap;
 use std::sync::Arc;
 
-use collab_plugins::cloud_storage::RemoteCollabStorage;
-use parking_lot::RwLock;
+use collab_plugins::cloud_storage::{CollabObject, RemoteCollabStorage, RemoteUpdateSender};
+use parking_lot::{Mutex, RwLock};
+use serde_json::Value;
 
 use flowy_database_deps::cloud::DatabaseCloudService;
 use flowy_document_deps::cloud::DocumentCloudService;
@@ -14,6 +16,7 @@ use crate::supabase::api::{
   SupabaseDatabaseServiceImpl, SupabaseDocumentServiceImpl, SupabaseFolderServiceImpl,
   SupabaseServerServiceImpl,
 };
+use crate::supabase::entities::RealtimeCollabUpdateEvent;
 use crate::AppFlowyServer;
 
 /// https://www.pgbouncer.org/features.html
@@ -54,11 +57,14 @@ impl PgPoolMode {
 pub struct SupabaseServer {
   #[allow(dead_code)]
   config: SupabaseConfiguration,
+  device_id: Mutex<String>,
+  update_tx: RwLock<HashMap<String, RemoteUpdateSender>>,
   restful_postgres: Arc<RwLock<Option<Arc<RESTfulPostgresServer>>>>,
 }
 
 impl SupabaseServer {
   pub fn new(config: SupabaseConfiguration) -> Self {
+    let update_tx = RwLock::new(HashMap::new());
     let restful_postgres = if config.enable_sync {
       Some(Arc::new(RESTfulPostgresServer::new(config.clone())))
     } else {
@@ -66,6 +72,8 @@ impl SupabaseServer {
     };
     Self {
       config,
+      device_id: Default::default(),
+      update_tx,
       restful_postgres: Arc::new(RwLock::new(restful_postgres)),
     }
   }
@@ -89,6 +97,10 @@ impl AppFlowyServer for SupabaseServer {
     self.set_enable_sync(enable);
   }
 
+  fn set_sync_device_id(&self, device_id: &str) {
+    *self.device_id.lock() = device_id.to_string();
+  }
+
   fn user_service(&self) -> Arc<dyn UserService> {
     Arc::new(RESTfulSupabaseUserAuthServiceImpl::new(
       SupabaseServerServiceImpl(self.restful_postgres.clone()),
@@ -113,9 +125,32 @@ impl AppFlowyServer for SupabaseServer {
     )))
   }
 
-  fn collab_storage(&self) -> Option<Arc<dyn RemoteCollabStorage>> {
+  fn collab_storage(&self, collab_object: &CollabObject) -> Option<Arc<dyn RemoteCollabStorage>> {
+    let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
+    self
+      .update_tx
+      .write()
+      .insert(collab_object.object_id.clone(), tx);
     Some(Arc::new(SupabaseCollabStorageImpl::new(
       SupabaseServerServiceImpl(self.restful_postgres.clone()),
+      Some(rx),
     )))
   }
+
+  fn handle_realtime_event(&self, json: Value) {
+    match serde_json::from_value::<RealtimeCollabUpdateEvent>(json) {
+      Ok(event) => {
+        if let Some(tx) = self.update_tx.read().get(event.payload.oid.as_str()) {
+          if self.device_id.lock().as_str() != event.payload.did.as_str() {
+            if let Err(e) = tx.send(event.payload.value) {
+              tracing::trace!("send realtime update error: {}", e);
+            }
+          }
+        }
+      },
+      Err(e) => {
+        tracing::error!("parser realtime event error: {}", e);
+      },
+    }
+  }
 }

frontend/rust-lib/flowy-server/tests/supabase_test/database_test.rs

@@ -1,10 +1,12 @@
-use crate::supabase_test::util::{
-  collab_service, database_service, get_supabase_config, sign_up_param, user_auth_service,
-};
 use collab_plugins::cloud_storage::{CollabObject, CollabType};
+use uuid::Uuid;
+
 use flowy_user_deps::entities::SignUpResponse;
 use lib_infra::box_any::BoxAny;
-use uuid::Uuid;
+
+use crate::supabase_test::util::{
+  collab_service, database_service, get_supabase_config, sign_up_param, user_auth_service,
+};
 
 #[tokio::test]
 async fn supabase_create_workspace_test() {
@@ -25,7 +27,7 @@ async fn supabase_create_workspace_test() {
     let row_id = uuid::Uuid::new_v4().to_string();
     row_ids.push(row_id.clone());
     let collab_object = CollabObject {
-      id: row_id,
+      object_id: row_id,
       uid: user.user_id,
       ty: CollabType::DatabaseRow,
       meta: Default::default(),

frontend/rust-lib/flowy-server/tests/supabase_test/folder_test.rs

@@ -41,7 +41,7 @@ async fn supabase_get_folder_test() {
   let user: SignUpResponse = user_service.sign_up(BoxAny::new(params)).await.unwrap();
 
   let collab_object = CollabObject {
-    id: user.latest_workspace.id.clone(),
+    object_id: user.latest_workspace.id.clone(),
     uid: user.user_id,
     ty: CollabType::Folder,
     meta: Default::default(),
@@ -124,7 +124,7 @@ async fn supabase_duplicate_updates_test() {
   let user: SignUpResponse = user_service.sign_up(BoxAny::new(params)).await.unwrap();
 
   let collab_object = CollabObject {
-    id: user.latest_workspace.id.clone(),
+    object_id: user.latest_workspace.id.clone(),
     uid: user.user_id,
     ty: CollabType::Folder,
     meta: Default::default(),
@@ -220,7 +220,7 @@ async fn supabase_diff_state_vec_test() {
   let user: SignUpResponse = user_service.sign_up(BoxAny::new(params)).await.unwrap();
 
   let collab_object = CollabObject {
-    id: user.latest_workspace.id.clone(),
+    object_id: user.latest_workspace.id.clone(),
     uid: user.user_id,
     ty: CollabType::Folder,
     meta: Default::default(),

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

@@ -27,6 +27,7 @@ pub fn collab_service() -> Arc<dyn RemoteCollabStorage> {
   let server = Arc::new(RESTfulPostgresServer::new(config));
   Arc::new(SupabaseCollabStorageImpl::new(
     SupabaseServerServiceImpl::new(server),
+    None,
   ))
 }
 

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

@@ -31,6 +31,7 @@ pub fn sign_up(dispatch: Arc<AFPluginDispatcher>) -> SignUpContext {
     name: "app flowy".to_string(),
     password: password.clone(),
     auth_type: AuthTypePB::Local,
+    device_id: uuid::Uuid::new_v4().to_string(),
   }
   .into_bytes()
   .unwrap();
@@ -58,6 +59,7 @@ pub async fn async_sign_up(
     name: "appflowy".to_string(),
     password: password.clone(),
     auth_type,
+    device_id: uuid::Uuid::new_v4().to_string(),
   }
   .into_bytes()
   .unwrap();

frontend/rust-lib/flowy-test/tests/database/supabase_test/helper.rs

@@ -102,7 +102,7 @@ pub fn assert_database_collab_content(
   expected: JsonValue,
 ) {
   let collab = MutexCollab::new(CollabOrigin::Server, database_id, vec![]);
-  collab.lock().with_transact_mut(|txn| {
+  collab.lock().with_origin_transact_mut(|txn| {
     let update = Update::decode_v1(collab_update).unwrap();
     txn.apply_update(update);
   });

frontend/rust-lib/flowy-test/tests/document/supabase_test/helper.rs

@@ -91,7 +91,7 @@ impl Deref for FlowySupabaseDocumentTest {
 
 pub fn assert_document_data_equal(collab_update: &[u8], doc_id: &str, expected: DocumentData) {
   let collab = MutexCollab::new(CollabOrigin::Server, doc_id, vec![]);
-  collab.lock().with_transact_mut(|txn| {
+  collab.lock().with_origin_transact_mut(|txn| {
     let update = Update::decode_v1(collab_update).unwrap();
     txn.apply_update(update);
   });

frontend/rust-lib/flowy-test/tests/folder/supabase_test/helper.rs

@@ -67,7 +67,7 @@ pub fn assert_folder_collab_content(workspace_id: &str, collab_update: &[u8], ex
   }
 
   let collab = MutexCollab::new(CollabOrigin::Server, workspace_id, vec![]);
-  collab.lock().with_transact_mut(|txn| {
+  collab.lock().with_origin_transact_mut(|txn| {
     let update = Update::decode_v1(collab_update).unwrap();
     txn.apply_update(update);
   });

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

@@ -1,6 +1,6 @@
 use flowy_test::user_event::*;
 use flowy_test::{event_builder::EventBuilder, FlowyCoreTest};
-use flowy_user::entities::{AuthTypePB, SignInPayloadPB, SignUpPayloadPB, UserProfilePB};
+use flowy_user::entities::{AuthTypePB, SignInPayloadPB, SignUpPayloadPB};
 use flowy_user::errors::ErrorCode;
 use flowy_user::event_map::UserEvent::*;
 
@@ -15,6 +15,7 @@ async fn sign_up_with_invalid_email() {
       name: valid_name(),
       password: login_password(),
       auth_type: AuthTypePB::Local,
+      device_id: "".to_string(),
     };
 
     assert_eq!(
@@ -38,6 +39,7 @@ async fn sign_up_with_long_password() {
     name: valid_name(),
     password: "1234".repeat(100).as_str().to_string(),
     auth_type: AuthTypePB::Local,
+    device_id: "".to_string(),
   };
 
   assert_eq!(
@@ -53,29 +55,6 @@ async fn sign_up_with_long_password() {
   );
 }
 
-#[tokio::test]
-async fn sign_in_success() {
-  let test = FlowyCoreTest::new();
-  let _ = EventBuilder::new(test.clone()).event(SignOut).sync_send();
-  let sign_up_context = test.sign_up_as_guest().await;
-
-  let request = SignInPayloadPB {
-    email: sign_up_context.user_profile.email.clone(),
-    password: sign_up_context.password.clone(),
-    name: "".to_string(),
-    auth_type: AuthTypePB::Local,
-    uid: Some(sign_up_context.user_profile.id),
-  };
-
-  let response = EventBuilder::new(test.clone())
-    .event(SignIn)
-    .payload(request)
-    .async_send()
-    .await
-    .parse::<UserProfilePB>();
-  dbg!(&response);
-}
-
 #[tokio::test]
 async fn sign_in_with_invalid_email() {
   for email in invalid_email_test_case() {
@@ -85,7 +64,7 @@ async fn sign_in_with_invalid_email() {
       password: login_password(),
       name: "".to_string(),
       auth_type: AuthTypePB::Local,
-      uid: None,
+      device_id: "".to_string(),
     };
 
     assert_eq!(
@@ -112,7 +91,7 @@ async fn sign_in_with_invalid_password() {
       password,
       name: "".to_string(),
       auth_type: AuthTypePB::Local,
-      uid: None,
+      device_id: "".to_string(),
     };
 
     assert!(EventBuilder::new(sdk)

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

@@ -1,7 +1,7 @@
-use anyhow::Error;
 use std::collections::HashMap;
 use std::str::FromStr;
 
+use anyhow::Error;
 use uuid::Uuid;
 
 use flowy_error::{ErrorCode, FlowyError};
@@ -64,7 +64,12 @@ pub fn third_party_params_from_box_any(any: BoxAny) -> Result<ThirdPartyParams,
   let map: HashMap<String, String> = any.unbox_or_error()?;
   let uuid = uuid_from_map(&map)?;
   let email = map.get("email").cloned().unwrap_or_default();
-  Ok(ThirdPartyParams { uuid, email })
+  let device_id = map.get("device_id").cloned().unwrap_or_default();
+  Ok(ThirdPartyParams {
+    uuid,
+    email,
+    device_id,
+  })
 }
 
 pub fn uuid_from_map(map: &HashMap<String, String>) -> Result<Uuid, Error> {

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

@@ -11,6 +11,7 @@ pub struct SignInResponse {
   pub user_workspaces: Vec<UserWorkspace>,
   pub email: Option<String>,
   pub token: Option<String>,
+  pub device_id: String,
 }
 
 #[derive(Default, Serialize, Deserialize, Debug)]
@@ -19,8 +20,7 @@ pub struct SignInParams {
   pub password: String,
   pub name: String,
   pub auth_type: AuthType,
-  // Currently, the uid only used in local sign in.
-  pub uid: Option<i64>,
+  pub device_id: String,
 }
 
 #[derive(Serialize, Deserialize, Default, Debug)]
@@ -29,6 +29,7 @@ pub struct SignUpParams {
   pub name: String,
   pub password: String,
   pub auth_type: AuthType,
+  pub device_id: String,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
@@ -40,6 +41,7 @@ pub struct SignUpResponse {
   pub is_new: bool,
   pub email: Option<String>,
   pub token: Option<String>,
+  pub device_id: String,
 }
 
 #[derive(Clone, Debug)]
@@ -190,4 +192,5 @@ impl From<i32> for AuthType {
 pub struct ThirdPartyParams {
   pub uuid: Uuid,
   pub email: String,
+  pub device_id: String,
 }

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

@@ -21,9 +21,8 @@ pub struct SignInPayloadPB {
   #[pb(index = 4)]
   pub auth_type: AuthTypePB,
 
-  // Only used in local sign in.
-  #[pb(index = 5, one_of)]
-  pub uid: Option<i64>,
+  #[pb(index = 5)]
+  pub device_id: String,
 }
 
 impl TryInto<SignInParams> for SignInPayloadPB {
@@ -38,7 +37,7 @@ impl TryInto<SignInParams> for SignInPayloadPB {
       password: password.0,
       name: self.name,
       auth_type: self.auth_type.into(),
-      uid: self.uid,
+      device_id: self.device_id,
     })
   }
 }
@@ -56,7 +55,11 @@ pub struct SignUpPayloadPB {
 
   #[pb(index = 4)]
   pub auth_type: AuthTypePB,
+
+  #[pb(index = 5)]
+  pub device_id: String,
 }
+
 impl TryInto<SignUpParams> for SignUpPayloadPB {
   type Error = ErrorCode;
 
@@ -70,6 +73,7 @@ impl TryInto<SignUpParams> for SignUpPayloadPB {
       name: name.0,
       password: password.0,
       auth_type: self.auth_type.into(),
+      device_id: self.device_id,
     })
   }
 }

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

@@ -1,8 +1,10 @@
 pub use auth::*;
+pub use realtime::*;
 pub use user_profile::*;
 pub use user_setting::*;
 
 pub mod auth;
 pub mod parser;
+pub mod realtime;
 mod user_profile;
 mod user_setting;

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

@@ -0,0 +1,7 @@
+use flowy_derive::ProtoBuf;
+
+#[derive(ProtoBuf, Default, Clone)]
+pub struct RealtimePayloadPB {
+  #[pb(index = 1)]
+  pub(crate) json_str: String,
+}

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

@@ -226,6 +226,9 @@ pub struct HistoricalUserPB {
 
   #[pb(index = 4)]
   pub auth_type: AuthTypePB,
+
+  #[pb(index = 5)]
+  pub device_id: String,
 }
 
 impl From<Vec<HistoricalUser>> for RepeatedHistoricalUserPB {
@@ -246,6 +249,7 @@ impl From<HistoricalUser> for HistoricalUserPB {
       user_name: historical_user.user_name,
       last_time: historical_user.sign_in_timestamp,
       auth_type: historical_user.auth_type.into(),
+      device_id: historical_user.device_id,
     }
   }
 }

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

@@ -2,6 +2,8 @@ use std::convert::TryFrom;
 use std::sync::Weak;
 use std::{convert::TryInto, sync::Arc};
 
+use serde_json::Value;
+
 use flowy_error::{FlowyError, FlowyResult};
 use flowy_server_config::supabase_config::SupabaseConfiguration;
 use flowy_sqlite::kv::StorePreferences;
@@ -277,6 +279,25 @@ pub async fn open_historical_users_handler(
 ) -> Result<(), FlowyError> {
   let user = user.into_inner();
   let session = upgrade_session(session)?;
-  session.open_historical_user(user.user_id)?;
+  let auth_type = AuthType::from(user.auth_type);
+  session.open_historical_user(user.user_id, user.device_id, auth_type)?;
+  Ok(())
+}
+
+#[tracing::instrument(level = "debug", skip_all, err)]
+pub async fn push_realtime_event_handler(
+  payload: AFPluginData<RealtimePayloadPB>,
+  session: AFPluginState<Weak<UserSession>>,
+) -> Result<(), FlowyError> {
+  match serde_json::from_str::<Value>(&payload.into_inner().json_str) {
+    Ok(json) => {
+      let session = upgrade_session(session)?;
+      session.receive_realtime_event(json).await;
+    },
+    Err(e) => {
+      tracing::error!("Deserialize RealtimePayload failed: {:?}", e);
+    },
+  }
+
   Ok(())
 }

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

@@ -1,6 +1,7 @@
 use std::sync::{Arc, Weak};
 
 use collab_folder::core::FolderData;
+use serde_json::Value;
 use strum_macros::Display;
 
 use flowy_derive::{Flowy_Event, ProtoBuf_Enum};
@@ -49,6 +50,7 @@ pub fn init(user_session: Weak<UserSession>) -> AFPlugin {
     .event(UserEvent::UpdateNetworkState, update_network_state_handler)
     .event(UserEvent::GetHistoricalUsers, get_historical_users_handler)
     .event(UserEvent::OpenHistoricalUser, open_historical_users_handler)
+    .event(UserEvent::PushRealtimeEvent, push_realtime_event_handler)
 }
 
 pub struct SignUpContext {
@@ -62,23 +64,35 @@ pub struct SignUpContext {
 pub trait UserStatusCallback: Send + Sync + 'static {
   /// When the [AuthType] changed, this method will be called. Currently, the auth type
   /// will be changed when the user sign in or sign up.
-  fn auth_type_did_changed(&self, auth_type: AuthType);
+  fn auth_type_did_changed(&self, _auth_type: AuthType) {}
   /// This will be called after the application launches if the user is already signed in.
   /// If the user is not signed in, this method will not be called
-  fn did_init(&self, user_id: i64, user_workspace: &UserWorkspace) -> Fut<FlowyResult<()>>;
+  fn did_init(
+    &self,
+    user_id: i64,
+    user_workspace: &UserWorkspace,
+    device_id: &str,
+  ) -> Fut<FlowyResult<()>>;
   /// Will be called after the user signed in.
-  fn did_sign_in(&self, user_id: i64, user_workspace: &UserWorkspace) -> Fut<FlowyResult<()>>;
+  fn did_sign_in(
+    &self,
+    user_id: i64,
+    user_workspace: &UserWorkspace,
+    device_id: &str,
+  ) -> Fut<FlowyResult<()>>;
   /// Will be called after the user signed up.
   fn did_sign_up(
     &self,
     context: SignUpContext,
     user_profile: &UserProfile,
     user_workspace: &UserWorkspace,
+    device_id: &str,
   ) -> Fut<FlowyResult<()>>;
 
   fn did_expired(&self, token: &str, user_id: i64) -> Fut<FlowyResult<()>>;
   fn open_workspace(&self, user_id: i64, user_workspace: &UserWorkspace) -> Fut<FlowyResult<()>>;
-  fn did_update_network(&self, reachable: bool);
+  fn did_update_network(&self, _reachable: bool) {}
+  fn receive_realtime_event(&self, _json: Value) {}
 }
 
 /// The user cloud service provider.
@@ -114,13 +128,21 @@ where
 /// Acts as a placeholder [UserStatusCallback] for the user session, but does not perform any function
 pub(crate) struct DefaultUserStatusCallback;
 impl UserStatusCallback for DefaultUserStatusCallback {
-  fn auth_type_did_changed(&self, _auth_type: AuthType) {}
-
-  fn did_init(&self, _user_id: i64, _user_workspace: &UserWorkspace) -> Fut<FlowyResult<()>> {
+  fn did_init(
+    &self,
+    _user_id: i64,
+    _user_workspace: &UserWorkspace,
+    _device_id: &str,
+  ) -> Fut<FlowyResult<()>> {
     to_fut(async { Ok(()) })
   }
 
-  fn did_sign_in(&self, _user_id: i64, _user_workspace: &UserWorkspace) -> Fut<FlowyResult<()>> {
+  fn did_sign_in(
+    &self,
+    _user_id: i64,
+    _user_workspace: &UserWorkspace,
+    _device_id: &str,
+  ) -> Fut<FlowyResult<()>> {
     to_fut(async { Ok(()) })
   }
 
@@ -129,6 +151,7 @@ impl UserStatusCallback for DefaultUserStatusCallback {
     _context: SignUpContext,
     _user_profile: &UserProfile,
     _user_workspace: &UserWorkspace,
+    _device_id: &str,
   ) -> Fut<FlowyResult<()>> {
     to_fut(async { Ok(()) })
   }
@@ -140,8 +163,6 @@ impl UserStatusCallback for DefaultUserStatusCallback {
   fn open_workspace(&self, _user_id: i64, _user_workspace: &UserWorkspace) -> Fut<FlowyResult<()>> {
     to_fut(async { Ok(()) })
   }
-
-  fn did_update_network(&self, _reachable: bool) {}
 }
 
 #[derive(Clone, Copy, PartialEq, Eq, Debug, Display, Hash, ProtoBuf_Enum, Flowy_Event)]
@@ -221,4 +242,9 @@ pub enum UserEvent {
 
   #[event(input = "HistoricalUserPB")]
   OpenHistoricalUser = 26,
+
+  /// Push a realtime event to the user. Currently, the realtime event is only used
+  /// when the auth type is: [AuthType::Supabase].
+  #[event(input = "RealtimePayloadPB")]
+  PushRealtimeEvent = 27,
 }

frontend/rust-lib/flowy-user/src/migrations/historical_document.rs

@@ -23,7 +23,7 @@ impl UserDataMigration for HistoricalEmptyDocumentMigration {
   fn run(&self, session: &Session, collab_db: &Arc<RocksCollabDB>) -> FlowyResult<()> {
     let write_txn = collab_db.write_txn();
     if let Ok(updates) = write_txn.get_all_updates(session.user_id, &session.user_workspace.id) {
-      let origin = CollabOrigin::Client(CollabClient::new(session.user_id, ""));
+      let origin = CollabOrigin::Client(CollabClient::new(session.user_id, "phantom"));
       // Deserialize the folder from the raw data
       let folder =
         Folder::from_collab_raw_data(origin.clone(), updates, &session.user_workspace.id, vec![])?;

frontend/rust-lib/flowy-user/src/migrations/local_user_to_cloud.rs

@@ -6,9 +6,10 @@ use collab::core::origin::{CollabClient, CollabOrigin};
 use collab::preclude::Collab;
 use collab_folder::core::{Folder, FolderData};
 
-use crate::migrations::UserMigrationContext;
 use flowy_error::{ErrorCode, FlowyError, FlowyResult};
 
+use crate::migrations::UserMigrationContext;
+
 /// Migration the collab objects of the old user to new user. Currently, it only happens when
 /// the user is a local user and try to use AppFlowy cloud service.
 pub fn migration_user_to_cloud(
@@ -72,7 +73,7 @@ fn migrate_database_storage<'a, W>(
   W: YrsDocAction<'a>,
   PersistenceError: From<W::Error>,
 {
-  let origin = CollabOrigin::Client(CollabClient::new(old_uid, ""));
+  let origin = CollabOrigin::Client(CollabClient::new(old_uid, "phantom"));
   match Collab::new_with_raw_data(origin, old_object_id, updates, vec![]) {
     Ok(collab) => {
       let txn = collab.transact();
@@ -94,7 +95,7 @@ fn migrate_object<'a, W>(
   W: YrsDocAction<'a>,
   PersistenceError: From<W::Error>,
 {
-  let origin = CollabOrigin::Client(CollabClient::new(old_uid, ""));
+  let origin = CollabOrigin::Client(CollabClient::new(old_uid, "phantom"));
   match Collab::new_with_raw_data(origin, object_id, updates, vec![]) {
     Ok(collab) => {
       let txn = collab.transact();
@@ -112,7 +113,7 @@ fn migrate_folder(
   new_workspace_id: &str,
   updates: CollabRawData,
 ) -> Option<FolderData> {
-  let origin = CollabOrigin::Client(CollabClient::new(old_uid, ""));
+  let origin = CollabOrigin::Client(CollabClient::new(old_uid, "phantom"));
   let old_folder_collab = Collab::new_with_raw_data(origin, old_object_id, updates, vec![]).ok()?;
   let mutex_collab = Arc::new(MutexCollab::from_collab(old_folder_collab));
   let old_folder = Folder::open(mutex_collab, None);

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

@@ -13,6 +13,7 @@ use flowy_user_deps::entities::{SignInResponse, SignUpResponse, UserWorkspace};
 #[derive(Debug, Clone, Serialize)]
 pub struct Session {
   pub user_id: i64,
+  pub device_id: String,
   pub user_workspace: UserWorkspace,
 }
 
@@ -32,6 +33,7 @@ impl<'de> Visitor<'de> for SessionVisitor {
     // For historical reasons, the session used to contain a workspace_id field.
     // This field is no longer used, and is replaced by user_workspace.
     let mut workspace_id = None;
+    let mut device_id = "phantom".to_string();
     let mut user_workspace = None;
 
     while let Some(key) = map.next_key::<String>()? {
@@ -42,6 +44,9 @@ impl<'de> Visitor<'de> for SessionVisitor {
         "workspace_id" => {
           workspace_id = Some(map.next_value()?);
         },
+        "device_id" => {
+          device_id = map.next_value()?;
+        },
         "user_workspace" => {
           user_workspace = Some(map.next_value()?);
         },
@@ -65,6 +70,7 @@ impl<'de> Visitor<'de> for SessionVisitor {
 
     let session = Session {
       user_id,
+      device_id,
       user_workspace: user_workspace.ok_or(serde::de::Error::missing_field("user_workspace"))?,
     };
 
@@ -85,6 +91,7 @@ impl std::convert::From<SignInResponse> for Session {
   fn from(resp: SignInResponse) -> Self {
     Session {
       user_id: resp.user_id,
+      device_id: resp.device_id,
       user_workspace: resp.latest_workspace,
     }
   }
@@ -106,6 +113,7 @@ impl From<&SignUpResponse> for Session {
   fn from(value: &SignUpResponse) -> Self {
     Session {
       user_id: value.user_id,
+      device_id: value.device_id.clone(),
       user_workspace: value.latest_workspace.clone(),
     }
   }
@@ -113,9 +121,10 @@ impl From<&SignUpResponse> for Session {
 
 #[cfg(test)]
 mod tests {
-  use super::*;
   use serde_json::json;
 
+  use super::*;
+
   #[derive(serde::Serialize)]
   struct OldSession {
     user_id: i64,

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

@@ -5,6 +5,7 @@ use std::sync::{Arc, Weak};
 use appflowy_integrate::RocksCollabDB;
 use collab_folder::core::FolderData;
 use serde::{Deserialize, Serialize};
+use serde_json::Value;
 use tokio::sync::RwLock;
 use uuid::Uuid;
 
@@ -107,7 +108,7 @@ impl UserSession {
       }
 
       if let Err(e) = user_status_callback
-        .did_init(session.user_id, &session.user_workspace)
+        .did_init(session.user_id, &session.user_workspace, &session.device_id)
         .await
       {
         tracing::error!("Failed to call did_sign_in callback: {:?}", e);
@@ -161,9 +162,16 @@ impl UserSession {
       .await?;
     let session: Session = response.clone().into();
     let uid = session.user_id;
+    let device_id = session.device_id.clone();
     self.set_current_session(Some(session))?;
 
-    self.log_user(uid, response.name.clone(), &auth_type, self.user_dir(uid));
+    self.log_user(
+      uid,
+      &response.device_id,
+      response.name.clone(),
+      &auth_type,
+      self.user_dir(uid),
+    );
 
     let user_workspace = response.latest_workspace.clone();
     save_user_workspaces(
@@ -182,7 +190,7 @@ impl UserSession {
       .user_status_callback
       .read()
       .await
-      .did_sign_in(user_profile.id, &user_workspace)
+      .did_sign_in(user_profile.id, &user_workspace, &device_id)
       .await
     {
       tracing::error!("Failed to call did_sign_in callback: {:?}", e);
@@ -234,7 +242,13 @@ impl UserSession {
     let new_session = Session::from(&response);
     self.set_current_session(Some(new_session.clone()))?;
     let uid = response.user_id;
-    self.log_user(uid, response.name.clone(), &auth_type, self.user_dir(uid));
+    self.log_user(
+      uid,
+      &response.device_id,
+      response.name.clone(),
+      &auth_type,
+      self.user_dir(uid),
+    );
     save_user_workspaces(
       self.db_pool(uid)?,
       response
@@ -282,6 +296,7 @@ impl UserSession {
         sign_up_context,
         &new_user_profile,
         &new_session.user_workspace,
+        &new_session.device_id,
       )
       .await;
     Ok(new_user_profile)
@@ -579,7 +594,14 @@ impl UserSession {
     Ok(())
   }
 
-  fn log_user(&self, uid: i64, user_name: String, auth_type: &AuthType, storage_path: String) {
+  fn log_user(
+    &self,
+    uid: i64,
+    device_id: &str,
+    user_name: String,
+    auth_type: &AuthType,
+    storage_path: String,
+  ) {
     let mut logger_users = self
       .store_preferences
       .get_object::<HistoricalUsers>(HISTORICAL_USER)
@@ -590,6 +612,7 @@ impl UserSession {
       auth_type: auth_type.clone(),
       sign_in_timestamp: timestamp(),
       storage_path,
+      device_id: device_id.to_string(),
     });
     let _ = self
       .store_preferences
@@ -606,7 +629,12 @@ impl UserSession {
     users
   }
 
-  pub fn open_historical_user(&self, uid: i64) -> FlowyResult<()> {
+  pub fn open_historical_user(
+    &self,
+    uid: i64,
+    device_id: String,
+    auth_type: AuthType,
+  ) -> FlowyResult<()> {
     let conn = self.db_connection(uid)?;
     let row = user_workspace_table::dsl::user_workspace_table
       .filter(user_workspace_table::uid.eq(uid))
@@ -614,13 +642,23 @@ impl UserSession {
     let user_workspace = UserWorkspace::from(row);
     let session = Session {
       user_id: uid,
+      device_id,
       user_workspace,
     };
-    self.cloud_services.set_auth_type(AuthType::Local);
+    debug_assert!(auth_type.is_local());
+    self.cloud_services.set_auth_type(auth_type);
     self.set_current_session(Some(session))?;
     Ok(())
   }
 
+  pub async fn receive_realtime_event(&self, json: Value) {
+    self
+      .user_status_callback
+      .read()
+      .await
+      .receive_realtime_event(json);
+  }
+
   /// Returns the current user session.
   pub fn get_session(&self) -> Result<Session, FlowyError> {
     match self
@@ -718,6 +756,8 @@ pub struct HistoricalUser {
   pub auth_type: AuthType,
   pub sign_in_timestamp: i64,
   pub storage_path: String,
+  #[serde(default)]
+  pub device_id: String,
 }
 
 const DEFAULT_AUTH_TYPE: fn() -> AuthType = || AuthType::Local;

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

@@ -1,7 +1,8 @@
-use chrono::{TimeZone, Utc};
-use flowy_error::FlowyError;
 use std::convert::TryFrom;
 
+use chrono::{TimeZone, Utc};
+
+use flowy_error::FlowyError;
 use flowy_sqlite::schema::user_workspace_table;
 use flowy_user_deps::entities::UserWorkspace;