feat: migration database (#2009)

* feat: migration database

* ci: fix tauri ci

* feat: migrate database view

* ci: fix ci

Author: appflowy

frontend/appflowy_tauri/package.json

@@ -27,6 +27,7 @@
     "is-hotkey": "^0.2.0",
     "jest": "^29.5.0",
     "nanoid": "^4.0.0",
+    "protoc-gen-ts": "^0.8.5",
     "react": "^18.2.0",
     "react-dom": "^18.2.0",
     "react-error-boundary": "^3.1.4",

frontend/rust-lib/flowy-codegen/src/flowy_toml.rs

@@ -20,11 +20,15 @@ pub struct FlowyConfig {
 }
 
 fn default_proto_output() -> String {
-  "resources/proto".to_owned()
+  let mut path = PathBuf::from("resources");
+  path.push("proto");
+  path.to_str().unwrap().to_owned()
 }
 
 fn default_protobuf_crate() -> String {
-  "src/protobuf".to_owned()
+  let mut path = PathBuf::from("src");
+  path.push("protobuf");
+  path.to_str().unwrap().to_owned()
 }
 
 impl FlowyConfig {

frontend/rust-lib/flowy-codegen/src/protobuf_file/mod.rs

@@ -127,6 +127,15 @@ fn generate_ts_protobuf_files(
   }
   let protoc_bin_path = protoc_bin_path.to_str().unwrap().to_owned();
   paths.iter().for_each(|path| {
+    // if let Err(err) = Command::new(protoc_bin_path.clone())
+    //   .arg(format!("--ts_out={}", output.to_str().unwrap()))
+    //   .arg(format!("--proto_path={}", proto_file_output_path))
+    //   .arg(path)
+    //   .spawn()
+    // {
+    //   panic!("Generate ts pb file failed: {}, {:?}", path, err);
+    // }
+
     let result = cmd_lib::run_cmd! {
         ${protoc_bin_path} --ts_out=${output} --proto_path=${proto_file_output_path} ${path}
     };

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

@@ -60,12 +60,12 @@ impl FolderDepsResolver {
       .await,
     );
 
-    if let (Ok(user_id), Ok(token)) = (user.user_id(), user.token()) {
-      match folder_manager.initialize(&user_id, &token).await {
-        Ok(_) => {},
-        Err(e) => tracing::error!("Initialize folder manager failed: {}", e),
-      }
-    }
+    // if let (Ok(user_id), Ok(token)) = (user.user_id(), user.token()) {
+    //   match folder_manager.initialize(&user_id, &token).await {
+    //     Ok(_) => {},
+    //     Err(e) => tracing::error!("Initialize folder manager failed: {}", e),
+    //   }
+    // }
 
     let receiver = Arc::new(FolderWSMessageReceiverImpl(folder_manager.clone()));
     ws_conn.add_ws_message_receiver(receiver).unwrap();
@@ -339,7 +339,7 @@ impl ViewDataProcessor for DatabaseViewDataProcessor {
   }
 }
 
-fn layout_type_from_view_layout(layout: ViewLayoutTypePB) -> LayoutTypePB {
+pub fn layout_type_from_view_layout(layout: ViewLayoutTypePB) -> LayoutTypePB {
   match layout {
     ViewLayoutTypePB::Grid => LayoutTypePB::Grid,
     ViewLayoutTypePB::Board => LayoutTypePB::Board,

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

@@ -25,21 +25,12 @@ impl DatabaseDepsResolver {
   ) -> Arc<DatabaseManager> {
     let user = Arc::new(GridUserImpl(user_session.clone()));
     let rev_web_socket = Arc::new(GridRevisionWebSocket(ws_conn));
-    let database_manager = Arc::new(DatabaseManager::new(
-      user.clone(),
+    Arc::new(DatabaseManager::new(
+      user,
       rev_web_socket,
       task_scheduler,
       Arc::new(DatabaseDBConnectionImpl(user_session)),
-    ));
-
-    if let (Ok(user_id), Ok(token)) = (user.user_id(), user.token()) {
-      match database_manager.initialize(&user_id, &token).await {
-        Ok(_) => {},
-        Err(e) => tracing::error!("Initialize grid manager failed: {}", e),
-      }
-    }
-
-    database_manager
+    ))
   }
 }
 

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

@@ -6,7 +6,7 @@ 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::entities::{ViewDataFormatPB, ViewLayoutTypePB};
 use flowy_folder::{errors::FlowyError, manager::FolderManager};
 pub use flowy_net::get_client_server_configuration;
 use flowy_net::local_server::LocalServer;
@@ -17,6 +17,7 @@ use flowy_user::services::{UserSession, UserSessionConfig};
 use lib_dispatch::prelude::*;
 use lib_dispatch::runtime::tokio_default_runtime;
 
+use flowy_database::entities::LayoutTypePB;
 use lib_infra::future::{to_fut, Fut};
 use module::make_plugins;
 pub use module::*;
@@ -306,7 +307,36 @@ 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.database_manager.initialize(user_id, token).await?;
+
+    let cloned_folder_manager = self.folder_manager.clone();
+    let get_views_fn = to_fut(async move {
+      cloned_folder_manager
+        .get_current_workspace()
+        .await
+        .map(|workspace| {
+          workspace
+            .apps
+            .items
+            .into_iter()
+            .flat_map(|app| app.belongings.items)
+            .flat_map(|view| match view.layout {
+              ViewLayoutTypePB::Grid | ViewLayoutTypePB::Board | ViewLayoutTypePB::Calendar => {
+                Some((
+                  view.id,
+                  view.name,
+                  layout_type_from_view_layout(view.layout),
+                ))
+              },
+              _ => None,
+            })
+            .collect::<Vec<(String, String, LayoutTypePB)>>()
+        })
+        .unwrap_or_default()
+    });
+    self
+      .database_manager
+      .initialize(user_id, token, get_views_fn)
+      .await?;
     self
       .ws_conn
       .start(token.to_owned(), user_id.to_owned())

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

@@ -7,7 +7,7 @@ use crate::services::database_view::{
   make_database_view_rev_manager, make_database_view_revision_pad, DatabaseViewEditor,
 };
 use crate::services::persistence::block_index::BlockRowIndexer;
-use crate::services::persistence::database_ref::{DatabaseInfo, DatabaseRef, DatabaseRefIndexer};
+use crate::services::persistence::database_ref::{DatabaseInfo, DatabaseRefs, DatabaseViewRef};
 use crate::services::persistence::kv::DatabaseKVPersistence;
 use crate::services::persistence::migration::DatabaseMigration;
 use crate::services::persistence::rev_sqlite::{
@@ -29,6 +29,7 @@ use flowy_revision::{
 use flowy_sqlite::ConnectionPool;
 use flowy_task::TaskDispatcher;
 
+use lib_infra::future::Fut;
 use revision_model::Revision;
 use std::sync::Arc;
 use tokio::sync::RwLock;
@@ -43,7 +44,7 @@ pub struct DatabaseManager {
   editors_by_database_id: RwLock<HashMap<String, Arc<DatabaseEditor>>>,
   database_user: Arc<dyn DatabaseUser>,
   block_indexer: Arc<BlockRowIndexer>,
-  database_ref_indexer: Arc<DatabaseRefIndexer>,
+  database_refs: Arc<DatabaseRefs>,
   #[allow(dead_code)]
   kv_persistence: Arc<DatabaseKVPersistence>,
   task_scheduler: Arc<RwLock<TaskDispatcher>>,
@@ -61,30 +62,30 @@ impl DatabaseManager {
     let editors_by_database_id = RwLock::new(HashMap::new());
     let kv_persistence = Arc::new(DatabaseKVPersistence::new(database_db.clone()));
     let block_indexer = Arc::new(BlockRowIndexer::new(database_db.clone()));
-    let database_ref_indexer = Arc::new(DatabaseRefIndexer::new(database_db.clone()));
-    let migration = DatabaseMigration::new(
-      database_user.clone(),
-      database_db,
-      database_ref_indexer.clone(),
-    );
+    let database_refs = Arc::new(DatabaseRefs::new(database_db));
+    let migration = DatabaseMigration::new(database_user.clone(), database_refs.clone());
     Self {
       editors_by_database_id,
       database_user,
       kv_persistence,
       block_indexer,
-      database_ref_indexer,
+      database_refs,
       task_scheduler,
       migration,
     }
   }
 
-  pub async fn initialize_with_new_user(&self, user_id: &str, _token: &str) -> FlowyResult<()> {
-    self.migration.run(user_id).await?;
+  pub async fn initialize_with_new_user(&self, _user_id: &str, _token: &str) -> FlowyResult<()> {
     Ok(())
   }
 
-  pub async fn initialize(&self, user_id: &str, _token: &str) -> FlowyResult<()> {
-    self.migration.run(user_id).await?;
+  pub async fn initialize(
+    &self,
+    user_id: &str,
+    _token: &str,
+    get_views_fn: Fut<Vec<(String, String, LayoutTypePB)>>,
+  ) -> FlowyResult<()> {
+    self.migration.run(user_id, get_views_fn).await?;
     Ok(())
   }
 
@@ -98,7 +99,7 @@ impl DatabaseManager {
   ) -> FlowyResult<()> {
     let db_pool = self.database_user.db_pool()?;
     let _ = self
-      .database_ref_indexer
+      .database_refs
       .bind(database_id, view_id.as_ref(), true, name);
     let rev_manager = self.make_database_rev_manager(database_id, db_pool)?;
     rev_manager.reset_object(revisions).await?;
@@ -113,7 +114,9 @@ impl DatabaseManager {
     revisions: Vec<Revision>,
   ) -> FlowyResult<()> {
     let view_id = view_id.as_ref();
-    let rev_manager = make_database_view_rev_manager(&self.database_user, view_id).await?;
+    let user_id = self.database_user.user_id()?;
+    let pool = self.database_user.db_pool()?;
+    let rev_manager = make_database_view_rev_manager(&user_id, pool, view_id).await?;
     rev_manager.reset_object(revisions).await?;
     Ok(())
   }
@@ -129,12 +132,13 @@ impl DatabaseManager {
     Ok(())
   }
 
+  #[tracing::instrument(level = "trace", skip_all, err)]
   pub async fn open_database_view<T: AsRef<str>>(
     &self,
     view_id: T,
   ) -> FlowyResult<Arc<DatabaseEditor>> {
     let view_id = view_id.as_ref();
-    let database_info = self.database_ref_indexer.get_database_with_view(view_id)?;
+    let database_info = self.database_refs.get_database_with_view(view_id)?;
     self
       .get_or_create_database_editor(&database_info.database_id, view_id)
       .await
@@ -143,7 +147,7 @@ impl DatabaseManager {
   #[tracing::instrument(level = "debug", skip_all)]
   pub async fn close_database_view<T: AsRef<str>>(&self, view_id: T) -> FlowyResult<()> {
     let view_id = view_id.as_ref();
-    let database_info = self.database_ref_indexer.get_database_with_view(view_id)?;
+    let database_info = self.database_refs.get_database_with_view(view_id)?;
     tracing::Span::current().record("database_id", &database_info.database_id);
 
     // Create a temporary reference database_editor in case of holding the write lock
@@ -172,7 +176,7 @@ impl DatabaseManager {
 
   // #[tracing::instrument(level = "debug", skip(self), err)]
   pub async fn get_database_editor(&self, view_id: &str) -> FlowyResult<Arc<DatabaseEditor>> {
-    let database_info = self.database_ref_indexer.get_database_with_view(view_id)?;
+    let database_info = self.database_refs.get_database_with_view(view_id)?;
     let database_editor = self
       .editors_by_database_id
       .read()
@@ -189,13 +193,14 @@ impl DatabaseManager {
   }
 
   pub async fn get_databases(&self) -> FlowyResult<Vec<DatabaseInfo>> {
-    self.database_ref_indexer.get_all_databases()
+    self.database_refs.get_all_databases()
   }
 
-  pub async fn get_database_ref_views(&self, database_id: &str) -> FlowyResult<Vec<DatabaseRef>> {
-    self
-      .database_ref_indexer
-      .get_ref_views_with_database(database_id)
+  pub async fn get_database_ref_views(
+    &self,
+    database_id: &str,
+  ) -> FlowyResult<Vec<DatabaseViewRef>> {
+    self.database_refs.get_ref_views_with_database(database_id)
   }
 
   async fn get_or_create_database_editor(
@@ -277,7 +282,7 @@ impl DatabaseManager {
       database_pad,
       rev_manager,
       self.block_indexer.clone(),
-      self.database_ref_indexer.clone(),
+      self.database_refs.clone(),
       self.task_scheduler.clone(),
     )
     .await?;
@@ -354,7 +359,7 @@ pub async fn link_existing_database(
     .await?;
 
   let _ = database_manager
-    .database_ref_indexer
+    .database_refs
     .bind(database_id, view_id, false, &name);
   Ok(())
 }
@@ -424,14 +429,14 @@ pub async fn create_new_database(
   Ok(())
 }
 
-impl DatabaseRefIndexerQuery for DatabaseRefIndexer {
-  fn get_ref_views(&self, database_id: &str) -> FlowyResult<Vec<DatabaseRef>> {
+impl DatabaseRefIndexerQuery for DatabaseRefs {
+  fn get_ref_views(&self, database_id: &str) -> FlowyResult<Vec<DatabaseViewRef>> {
     self.get_ref_views_with_database(database_id)
   }
 }
 
-impl DatabaseRefIndexerQuery for Arc<DatabaseRefIndexer> {
-  fn get_ref_views(&self, database_id: &str) -> FlowyResult<Vec<DatabaseRef>> {
+impl DatabaseRefIndexerQuery for Arc<DatabaseRefs> {
+  fn get_ref_views(&self, database_id: &str) -> FlowyResult<Vec<DatabaseViewRef>> {
     (**self).get_ref_views(database_id)
   }
 }

frontend/rust-lib/flowy-database/src/services/database/database_editor.rs

@@ -18,7 +18,7 @@ use crate::services::database_view::{
 };
 use crate::services::filter::FilterType;
 use crate::services::persistence::block_index::BlockRowIndexer;
-use crate::services::persistence::database_ref::DatabaseRef;
+use crate::services::persistence::database_ref::DatabaseViewRef;
 use crate::services::row::{DatabaseBlockRow, DatabaseBlockRowRevision, RowRevisionBuilder};
 use bytes::Bytes;
 use database_model::*;
@@ -42,7 +42,7 @@ use std::sync::Arc;
 use tokio::sync::{broadcast, RwLock};
 
 pub trait DatabaseRefIndexerQuery: Send + Sync + 'static {
-  fn get_ref_views(&self, database_id: &str) -> FlowyResult<Vec<DatabaseRef>>;
+  fn get_ref_views(&self, database_id: &str) -> FlowyResult<Vec<DatabaseViewRef>>;
 }
 
 pub struct DatabaseEditor {
@@ -695,6 +695,7 @@ impl DatabaseEditor {
     Ok(())
   }
 
+  #[tracing::instrument(level = "trace", skip(self), err)]
   pub async fn get_database(&self, view_id: &str) -> FlowyResult<DatabasePB> {
     let pad = self.database_pad.read().await;
     let fields = pad