fix: potential async errors (#1772)

Author: appflowy

frontend/appflowy_tauri/src-tauri/Cargo.lock

@@ -1,6 +1,6 @@
-export * from './classes/flowy-user';
-export * from './classes/flowy-document';
-export * from './classes/flowy-grid';
-export * from './classes/flowy-folder';
-export * from './classes/flowy-net';
-export * from './classes/flowy-error';
+export * from "./classes/flowy-user";
+export * from "./classes/flowy-document";
+export * from "./classes/flowy-database";
+export * from "./classes/flowy-folder";
+export * from "./classes/flowy-net";
+export * from "./classes/flowy-error";

frontend/appflowy_tauri/src/services/backend/index.ts

@@ -10,12 +10,13 @@ lib-dispatch = { path = "../lib-dispatch" }
 lib-log = { path = "../lib-log" }
 flowy-user = { path = "../flowy-user" }
 flowy-net = { path = "../flowy-net" }
-flowy-folder = { path = "../flowy-folder", default-features = false }
-flowy-database = { path = "../flowy-database", default-features = false }
+flowy-folder = { path = "../flowy-folder" }
+flowy-database = { path = "../flowy-database" }
 grid-model = { path = "../../../shared-lib/grid-model" }
+user-model = { path = "../../../shared-lib/user-model" }
 flowy-client-ws = { path = "../../../shared-lib/flowy-client-ws" }
 flowy-sqlite = { path = "../flowy-sqlite", optional = true }
-flowy-document = { path = "../flowy-document", default-features = false }
+flowy-document = { path = "../flowy-document" }
 flowy-revision = { path = "../flowy-revision" }
 flowy-error = { path = "../flowy-error", features = ["adaptor_ws"] }
 flowy-task = { path = "../flowy-task" }
@@ -32,6 +33,7 @@ lib-ws = { path = "../../../shared-lib/lib-ws" }
 lib-infra = { path = "../../../shared-lib/lib-infra" }
 
 [features]
+default = ["rev-sqlite"]
 http_sync = ["flowy-folder/cloud_sync", "flowy-document/cloud_sync"]
 native_sync = ["flowy-folder/cloud_sync", "flowy-document/cloud_sync"]
 use_bunyan = ["lib-log/use_bunyan"]

frontend/rust-lib/Cargo.lock

@@ -1,21 +1,23 @@
 mod deps_resolve;
 pub mod module;
-pub use flowy_net::get_client_server_configuration;
-
 use crate::deps_resolve::*;
-
 use flowy_client_ws::{listen_on_websocket, FlowyWebSocketConnect, NetworkType};
 use flowy_database::manager::DatabaseManager;
 use flowy_document::entities::DocumentVersionPB;
 use flowy_document::{DocumentConfig, DocumentManager};
+use flowy_error::FlowyResult;
 use flowy_folder::entities::ViewDataFormatPB;
 use flowy_folder::{errors::FlowyError, manager::FolderManager};
+pub use flowy_net::get_client_server_configuration;
 use flowy_net::local_server::LocalServer;
 use flowy_net::ClientServerConfiguration;
 use flowy_task::{TaskDispatcher, TaskRunner};
-use flowy_user::services::{notifier::UserStatus, UserSession, UserSessionConfig};
+use flowy_user::event_map::UserStatusCallback;
+use flowy_user::services::{UserSession, UserSessionConfig};
 use lib_dispatch::prelude::*;
 use lib_dispatch::runtime::tokio_default_runtime;
+
+use lib_infra::future::{to_fut, Fut};
 use module::make_plugins;
 pub use module::*;
 use std::time::Duration;
@@ -27,6 +29,7 @@ use std::{
     },
 };
 use tokio::sync::{broadcast, RwLock};
+use user_model::UserProfile;
 
 static INIT_LOG: AtomicBool = AtomicBool::new(false);
 
@@ -83,9 +86,10 @@ fn create_log_filter(level: String, with_crates: Vec<String>) -> String {
     filters.push(format!("flowy_folder={}", level));
     filters.push(format!("flowy_user={}", level));
     filters.push(format!("flowy_document={}", level));
-    filters.push(format!("flowy_grid={}", level));
-    filters.push(format!("flowy_collaboration={}", "info"));
-    filters.push(format!("flowy_notification={}", level));
+    filters.push(format!("flowy_database={}", level));
+    filters.push(format!("flowy_sync={}", "info"));
+    filters.push(format!("flowy_client_sync={}", "info"));
+    filters.push(format!("flowy_notification={}", "info"));
     filters.push(format!("lib_ot={}", level));
     filters.push(format!("lib_ws={}", level));
     filters.push(format!("lib_infra={}", level));
@@ -162,6 +166,21 @@ impl FlowySDK {
             )
         });
 
+        let user_status_listener = UserStatusListener {
+            document_manager: document_manager.clone(),
+            folder_manager: folder_manager.clone(),
+            grid_manager: grid_manager.clone(),
+            ws_conn: ws_conn.clone(),
+            config: config.clone(),
+        };
+        let user_status_callback = UserStatusCallbackImpl {
+            listener: Arc::new(user_status_listener),
+        };
+        let cloned_user_session = user_session.clone();
+        runtime.block_on(async move {
+            cloned_user_session.clone().init(user_status_callback).await;
+        });
+
         let event_dispatcher = Arc::new(AFPluginDispatcher::construct(runtime, || {
             make_plugins(
                 &ws_conn,
@@ -171,16 +190,7 @@ impl FlowySDK {
                 &document_manager,
             )
         }));
-
-        _start_listening(
-            &config,
-            &event_dispatcher,
-            &ws_conn,
-            &user_session,
-            &document_manager,
-            &folder_manager,
-            &grid_manager,
-        );
+        _start_listening(&event_dispatcher, &ws_conn, &folder_manager);
 
         Self {
             config,
@@ -201,36 +211,17 @@ impl FlowySDK {
 }
 
 fn _start_listening(
-    config: &FlowySDKConfig,
     event_dispatcher: &AFPluginDispatcher,
     ws_conn: &Arc<FlowyWebSocketConnect>,
-    user_session: &Arc<UserSession>,
-    document_manager: &Arc<DocumentManager>,
     folder_manager: &Arc<FolderManager>,
-    grid_manager: &Arc<DatabaseManager>,
 ) {
-    let subscribe_user_status = user_session.notifier.subscribe_user_status();
     let subscribe_network_type = ws_conn.subscribe_network_ty();
     let folder_manager = folder_manager.clone();
-    let grid_manager = grid_manager.clone();
-    let cloned_folder_manager = folder_manager.clone();
+    let cloned_folder_manager = folder_manager;
     let ws_conn = ws_conn.clone();
-    let user_session = user_session.clone();
-    let document_manager = document_manager.clone();
-    let config = config.clone();
 
     event_dispatcher.spawn(async move {
-        user_session.init();
         listen_on_websocket(ws_conn.clone());
-        _listen_user_status(
-            config,
-            ws_conn.clone(),
-            subscribe_user_status,
-            document_manager,
-            folder_manager,
-            grid_manager,
-        )
-        .await;
     });
 
     event_dispatcher.spawn(async move {
@@ -253,66 +244,6 @@ fn mk_local_server(
     }
 }
 
-async fn _listen_user_status(
-    config: FlowySDKConfig,
-    ws_conn: Arc<FlowyWebSocketConnect>,
-    mut subscribe: broadcast::Receiver<UserStatus>,
-    document_manager: Arc<DocumentManager>,
-    folder_manager: Arc<FolderManager>,
-    grid_manager: Arc<DatabaseManager>,
-) {
-    while let Ok(status) = subscribe.recv().await {
-        let result = || async {
-            match status {
-                UserStatus::Login { token, user_id } => {
-                    tracing::trace!("User did login");
-                    folder_manager.initialize(&user_id, &token).await?;
-                    document_manager.initialize(&user_id).await?;
-                    grid_manager.initialize(&user_id, &token).await?;
-                    ws_conn.start(token, user_id).await?;
-                }
-                UserStatus::Logout { token: _, user_id } => {
-                    tracing::trace!("User did logout");
-                    folder_manager.clear(&user_id).await;
-                    ws_conn.stop().await;
-                }
-                UserStatus::Expired { token: _, user_id } => {
-                    tracing::trace!("User session has been expired");
-                    folder_manager.clear(&user_id).await;
-                    ws_conn.stop().await;
-                }
-                UserStatus::SignUp { profile, ret } => {
-                    tracing::trace!("User did sign up");
-
-                    let view_data_type = match config.document.version {
-                        DocumentVersionPB::V0 => ViewDataFormatPB::DeltaFormat,
-                        DocumentVersionPB::V1 => ViewDataFormatPB::TreeFormat,
-                    };
-                    folder_manager
-                        .initialize_with_new_user(&profile.id, &profile.token, view_data_type)
-                        .await?;
-                    document_manager
-                        .initialize_with_new_user(&profile.id, &profile.token)
-                        .await?;
-
-                    grid_manager
-                        .initialize_with_new_user(&profile.id, &profile.token)
-                        .await?;
-
-                    ws_conn.start(profile.token.clone(), profile.id.clone()).await?;
-                    let _ = ret.send(());
-                }
-            }
-            Ok::<(), FlowyError>(())
-        };
-
-        match result().await {
-            Ok(_) => {}
-            Err(e) => tracing::error!("{}", e),
-        }
-    }
-}
-
 async fn _listen_network_status(mut subscribe: broadcast::Receiver<NetworkType>, _core: Arc<FolderManager>) {
     while let Ok(_new_type) = subscribe.recv().await {
         // core.network_state_changed(new_type);
@@ -345,3 +276,75 @@ fn mk_user_session(
     let cloud_service = UserDepsResolver::resolve(local_server, server_config);
     Arc::new(UserSession::new(user_config, cloud_service))
 }
+
+struct UserStatusListener {
+    document_manager: Arc<DocumentManager>,
+    folder_manager: Arc<FolderManager>,
+    grid_manager: Arc<DatabaseManager>,
+    ws_conn: Arc<FlowyWebSocketConnect>,
+    config: FlowySDKConfig,
+}
+
+impl UserStatusListener {
+    async fn did_sign_in(&self, token: &str, user_id: &str) -> FlowyResult<()> {
+        self.folder_manager.initialize(user_id, token).await?;
+        self.document_manager.initialize(user_id).await?;
+        self.grid_manager.initialize(user_id, token).await?;
+        self.ws_conn.start(token.to_owned(), user_id.to_owned()).await?;
+        Ok(())
+    }
+
+    async fn did_sign_up(&self, user_profile: &UserProfile) -> FlowyResult<()> {
+        let view_data_type = match self.config.document.version {
+            DocumentVersionPB::V0 => ViewDataFormatPB::DeltaFormat,
+            DocumentVersionPB::V1 => ViewDataFormatPB::TreeFormat,
+        };
+        self.folder_manager
+            .initialize_with_new_user(&user_profile.id, &user_profile.token, view_data_type)
+            .await?;
+        self.document_manager
+            .initialize_with_new_user(&user_profile.id, &user_profile.token)
+            .await?;
+
+        self.grid_manager
+            .initialize_with_new_user(&user_profile.id, &user_profile.token)
+            .await?;
+
+        self.ws_conn
+            .start(user_profile.token.clone(), user_profile.id.clone())
+            .await?;
+        Ok(())
+    }
+
+    async fn did_expired(&self, _token: &str, user_id: &str) -> FlowyResult<()> {
+        self.folder_manager.clear(user_id).await;
+        self.ws_conn.stop().await;
+        Ok(())
+    }
+}
+
+struct UserStatusCallbackImpl {
+    listener: Arc<UserStatusListener>,
+}
+
+impl UserStatusCallback for UserStatusCallbackImpl {
+    fn did_sign_in(&self, token: &str, user_id: &str) -> Fut<FlowyResult<()>> {
+        let listener = self.listener.clone();
+        let token = token.to_owned();
+        let user_id = user_id.to_owned();
+        to_fut(async move { listener.did_sign_in(&token, &user_id).await })
+    }
+
+    fn did_sign_up(&self, user_profile: &UserProfile) -> Fut<FlowyResult<()>> {
+        let listener = self.listener.clone();
+        let user_profile = user_profile.clone();
+        to_fut(async move { listener.did_sign_up(&user_profile).await })
+    }
+
+    fn did_expired(&self, token: &str, user_id: &str) -> Fut<FlowyResult<()>> {
+        let listener = self.listener.clone();
+        let token = token.to_owned();
+        let user_id = user_id.to_owned();
+        to_fut(async move { listener.did_expired(&token, &user_id).await })
+    }
+}

frontend/rust-lib/flowy-core/Cargo.toml

@@ -56,7 +56,7 @@ flowy-database = { path = "", features = ["flowy_unit_test"]}
 flowy-codegen = { path = "../flowy-codegen"}
 
 [features]
-default = []
+default = ["rev-sqlite"]
 rev-sqlite = ["flowy-sqlite"]
 dart = ["flowy-codegen/dart", "flowy-notification/dart"]
 ts = ["flowy-codegen/ts", "flowy-notification/ts"]

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

@@ -58,6 +58,7 @@ flowy-codegen = { path = "../flowy-codegen"}
 
 
 [features]
+default = ["rev-sqlite"]
 sync = []
 cloud_sync = ["sync"]
 rev-sqlite = ["flowy-sqlite"]

frontend/rust-lib/flowy-database/Cargo.toml

@@ -65,7 +65,6 @@ impl DocumentMigration {
         //
 
         KV::set_bool(&key, true);
-        tracing::debug!("Run document v1 migration");
         Ok(())
     }
 }

frontend/rust-lib/flowy-document/Cargo.toml

@@ -48,7 +48,7 @@ flowy-codegen = { path = "../flowy-codegen"}
 
 
 [features]
-default = []
+default = ["rev-sqlite"]
 sync = []
 cloud_sync = ["sync"]
 rev-sqlite = ["flowy-sqlite", "flowy-folder/rev-sqlite"]

frontend/rust-lib/flowy-document/src/services/migration.rs

@@ -209,7 +209,7 @@ impl ViewController {
         let deleted_view = self
             .persistence
             .begin_transaction(|transaction| {
-                let view = transaction.read_view(&view_id)?;
+                let view = transaction.read_view(view_id)?;
                 let views = read_belonging_views_on_local(&view.app_id, self.trash_controller.clone(), &transaction)?;
 
                 let index = views
@@ -223,12 +223,12 @@ impl ViewController {
             })
             .await?;
 
-        send_notification(&view_id, FolderNotification::ViewMoveToTrash)
+        send_notification(view_id, FolderNotification::ViewMoveToTrash)
             .payload(deleted_view)
             .send();
 
-        let processor = self.get_data_processor_from_view_id(&view_id).await?;
-        processor.close_view(&view_id).await?;
+        let processor = self.get_data_processor_from_view_id(view_id).await?;
+        processor.close_view(view_id).await?;
         Ok(())
     }