Refactor/revision compose (#1410)

Author: appflowy

frontend/rust-lib/Cargo.lock

@@ -1,6 +1,6 @@
 #[rustfmt::skip]
 /*
-diesel master support on_conflict on sqlite but not 1.4.7 version. Workaround for this
+diesel master support on_conflict on rev_sqlite but not 1.4.7 version. Workaround for this
 
 match dsl::workspace_table
     .filter(workspace_table::id.eq(table.id.clone()))
@@ -177,20 +177,20 @@ macro_rules! impl_rev_state_map {
             }
         }
 
-        impl std::convert::From<$target> for crate::disk::RevisionState {
+        impl std::convert::From<$target> for RevisionState {
             fn from(s: $target) -> Self {
                 match s {
-                    $target::Sync => crate::disk::RevisionState::Sync,
-                    $target::Ack => crate::disk::RevisionState::Ack,
+                    $target::Sync => RevisionState::Sync,
+                    $target::Ack => RevisionState::Ack,
                 }
             }
         }
 
-        impl std::convert::From<crate::disk::RevisionState> for $target {
-            fn from(s: crate::disk::RevisionState) -> Self {
+        impl std::convert::From<RevisionState> for $target {
+            fn from(s: RevisionState) -> Self {
                 match s {
-                    crate::disk::RevisionState::Sync => $target::Sync,
-                    crate::disk::RevisionState::Ack => $target::Ack,
+                    RevisionState::Sync => $target::Sync,
+                    RevisionState::Ack => $target::Ack,
                 }
             }
         }

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

@@ -4,6 +4,7 @@ use crate::editor::make_transaction_from_revisions;
 use crate::editor::queue::{Command, CommandSender, DocumentQueue};
 use crate::{DocumentEditor, DocumentUser};
 use bytes::Bytes;
+use flowy_database::ConnectionPool;
 use flowy_error::{internal_error, FlowyError, FlowyResult};
 use flowy_revision::{RevisionCloudService, RevisionManager};
 use flowy_sync::entities::ws_data::ServerRevisionWSData;
@@ -18,14 +19,14 @@ pub struct AppFlowyDocumentEditor {
     #[allow(dead_code)]
     doc_id: String,
     command_sender: CommandSender,
-    rev_manager: Arc<RevisionManager>,
+    rev_manager: Arc<RevisionManager<Arc<ConnectionPool>>>,
 }
 
 impl AppFlowyDocumentEditor {
     pub async fn new(
         doc_id: &str,
         user: Arc<dyn DocumentUser>,
-        mut rev_manager: RevisionManager,
+        mut rev_manager: RevisionManager<Arc<ConnectionPool>>,
         cloud_service: Arc<dyn RevisionCloudService>,
     ) -> FlowyResult<Arc<Self>> {
         let document = rev_manager.load::<DocumentRevisionSerde>(Some(cloud_service)).await?;
@@ -70,7 +71,7 @@ impl AppFlowyDocumentEditor {
 
 fn spawn_edit_queue(
     user: Arc<dyn DocumentUser>,
-    rev_manager: Arc<RevisionManager>,
+    rev_manager: Arc<RevisionManager<Arc<ConnectionPool>>>,
     document: Document,
 ) -> CommandSender {
     let (sender, receiver) = mpsc::channel(1000);

frontend/rust-lib/flowy-document/src/editor/editor.rs

@@ -2,13 +2,11 @@
 mod document;
 mod document_serde;
 mod editor;
-mod migration;
 mod queue;
 
 pub use document::*;
 pub use document_serde::*;
 pub use editor::*;
-pub use migration::*;
 
 #[inline]
 pub fn initial_read_me() -> String {

frontend/rust-lib/flowy-document/src/editor/migration/mod.rs

@@ -8,6 +8,7 @@ use flowy_sync::entities::revision::{RevId, Revision};
 use futures::stream::StreamExt;
 use lib_ot::core::Transaction;
 
+use flowy_database::ConnectionPool;
 use std::sync::Arc;
 use tokio::sync::mpsc::{Receiver, Sender};
 use tokio::sync::{oneshot, RwLock};
@@ -17,14 +18,14 @@ pub struct DocumentQueue {
     user: Arc<dyn DocumentUser>,
     document: Arc<RwLock<Document>>,
     #[allow(dead_code)]
-    rev_manager: Arc<RevisionManager>,
+    rev_manager: Arc<RevisionManager<Arc<ConnectionPool>>>,
     receiver: Option<CommandReceiver>,
 }
 
 impl DocumentQueue {
     pub fn new(
         user: Arc<dyn DocumentUser>,
-        rev_manager: Arc<RevisionManager>,
+        rev_manager: Arc<RevisionManager<Arc<ConnectionPool>>>,
         document: Document,
         receiver: CommandReceiver,
     ) -> Self {

frontend/rust-lib/flowy-document/src/editor/mod.rs

@@ -1,13 +1,13 @@
 use crate::editor::{initial_document_content, AppFlowyDocumentEditor, DocumentRevisionCompress};
 use crate::entities::{DocumentVersionPB, EditParams};
 use crate::old_editor::editor::{DeltaDocumentEditor, DeltaDocumentRevisionCompress};
+use crate::services::rev_sqlite::{SQLiteDeltaDocumentRevisionPersistence, SQLiteDocumentRevisionPersistence};
 use crate::services::DocumentPersistence;
 use crate::{errors::FlowyError, DocumentCloudService};
 use bytes::Bytes;
 use dashmap::DashMap;
 use flowy_database::ConnectionPool;
 use flowy_error::FlowyResult;
-use flowy_revision::disk::{SQLiteDeltaDocumentRevisionPersistence, SQLiteDocumentRevisionPersistence};
 use flowy_revision::{
     RevisionCloudService, RevisionManager, RevisionPersistence, RevisionWebSocket, SQLiteRevisionSnapshotPersistence,
 };
@@ -197,7 +197,7 @@ impl DocumentManager {
     /// # Arguments
     ///
     /// * `doc_id`: the id of the document
-    /// * `pool`: sqlite connection pool
+    /// * `pool`: rev_sqlite connection pool
     ///
     /// returns: Result<Arc<DocumentEditor>, FlowyError>
     ///
@@ -231,7 +231,11 @@ impl DocumentManager {
         }
     }
 
-    fn make_rev_manager(&self, doc_id: &str, pool: Arc<ConnectionPool>) -> Result<RevisionManager, FlowyError> {
+    fn make_rev_manager(
+        &self,
+        doc_id: &str,
+        pool: Arc<ConnectionPool>,
+    ) -> Result<RevisionManager<Arc<ConnectionPool>>, FlowyError> {
         match self.config.version {
             DocumentVersionPB::V0 => self.make_delta_document_rev_manager(doc_id, pool),
             DocumentVersionPB::V1 => self.make_document_rev_manager(doc_id, pool),
@@ -242,7 +246,7 @@ impl DocumentManager {
         &self,
         doc_id: &str,
         pool: Arc<ConnectionPool>,
-    ) -> Result<RevisionManager, FlowyError> {
+    ) -> Result<RevisionManager<Arc<ConnectionPool>>, FlowyError> {
         let user_id = self.user.user_id()?;
         let disk_cache = SQLiteDocumentRevisionPersistence::new(&user_id, pool.clone());
         let rev_persistence = RevisionPersistence::new(&user_id, doc_id, disk_cache);
@@ -262,7 +266,7 @@ impl DocumentManager {
         &self,
         doc_id: &str,
         pool: Arc<ConnectionPool>,
-    ) -> Result<RevisionManager, FlowyError> {
+    ) -> Result<RevisionManager<Arc<ConnectionPool>>, FlowyError> {
         let user_id = self.user.user_id()?;
         let disk_cache = SQLiteDeltaDocumentRevisionPersistence::new(&user_id, pool.clone());
         let rev_persistence = RevisionPersistence::new(&user_id, doc_id, disk_cache);

frontend/rust-lib/flowy-document/src/editor/queue.rs

@@ -3,6 +3,7 @@
 use crate::old_editor::queue::{EditDocumentQueue, EditorCommand, EditorCommandSender};
 use crate::{errors::FlowyError, DocumentEditor, DocumentUser};
 use bytes::Bytes;
+use flowy_database::ConnectionPool;
 use flowy_error::{internal_error, FlowyResult};
 use flowy_revision::{
     RevisionCloudService, RevisionCompress, RevisionManager, RevisionObjectDeserializer, RevisionObjectSerializer,
@@ -28,7 +29,7 @@ use tokio::sync::{mpsc, oneshot};
 pub struct DeltaDocumentEditor {
     pub doc_id: String,
     #[allow(dead_code)]
-    rev_manager: Arc<RevisionManager>,
+    rev_manager: Arc<RevisionManager<Arc<ConnectionPool>>>,
     #[cfg(feature = "sync")]
     ws_manager: Arc<flowy_revision::RevisionWebSocketManager>,
     edit_cmd_tx: EditorCommandSender,
@@ -39,7 +40,7 @@ impl DeltaDocumentEditor {
     pub(crate) async fn new(
         doc_id: &str,
         user: Arc<dyn DocumentUser>,
-        mut rev_manager: RevisionManager,
+        mut rev_manager: RevisionManager<Arc<ConnectionPool>>,
         rev_web_socket: Arc<dyn RevisionWebSocket>,
         cloud_service: Arc<dyn RevisionCloudService>,
     ) -> FlowyResult<Arc<Self>> {
@@ -210,7 +211,7 @@ impl std::ops::Drop for DeltaDocumentEditor {
 // The edit queue will exit after the EditorCommandSender was dropped.
 fn spawn_edit_queue(
     user: Arc<dyn DocumentUser>,
-    rev_manager: Arc<RevisionManager>,
+    rev_manager: Arc<RevisionManager<Arc<ConnectionPool>>>,
     delta: DeltaTextOperations,
 ) -> EditorCommandSender {
     let (sender, receiver) = mpsc::channel(1000);
@@ -238,7 +239,7 @@ impl DeltaDocumentEditor {
         Ok(delta)
     }
 
-    pub fn rev_manager(&self) -> Arc<RevisionManager> {
+    pub fn rev_manager(&self) -> Arc<RevisionManager<Arc<ConnectionPool>>> {
         self.rev_manager.clone()
     }
 }

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

@@ -1,6 +1,7 @@
 use crate::old_editor::web_socket::DeltaDocumentResolveOperations;
 use crate::DocumentUser;
 use async_stream::stream;
+use flowy_database::ConnectionPool;
 use flowy_error::FlowyError;
 use flowy_revision::{OperationsMD5, RevisionManager, TransformOperations};
 use flowy_sync::{
@@ -23,14 +24,14 @@ use tokio::sync::{oneshot, RwLock};
 pub(crate) struct EditDocumentQueue {
     document: Arc<RwLock<ClientDocument>>,
     user: Arc<dyn DocumentUser>,
-    rev_manager: Arc<RevisionManager>,
+    rev_manager: Arc<RevisionManager<Arc<ConnectionPool>>>,
     receiver: Option<EditorCommandReceiver>,
 }
 
 impl EditDocumentQueue {
     pub(crate) fn new(
         user: Arc<dyn DocumentUser>,
-        rev_manager: Arc<RevisionManager>,
+        rev_manager: Arc<RevisionManager<Arc<ConnectionPool>>>,
         operations: DeltaTextOperations,
         receiver: EditorCommandReceiver,
     ) -> Self {

frontend/rust-lib/flowy-document/src/old_editor/editor.rs

@@ -1,6 +1,7 @@
 use crate::old_editor::queue::{EditorCommand, EditorCommandSender, TextTransformOperations};
 use crate::TEXT_BLOCK_SYNC_INTERVAL_IN_MILLIS;
 use bytes::Bytes;
+use flowy_database::ConnectionPool;
 use flowy_error::{internal_error, FlowyError, FlowyResult};
 use flowy_revision::*;
 use flowy_sync::entities::revision::Revision;
@@ -41,14 +42,14 @@ impl DeltaDocumentResolveOperations {
     }
 }
 
-pub type DocumentConflictController = ConflictController<DeltaDocumentResolveOperations>;
+pub type DocumentConflictController = ConflictController<DeltaDocumentResolveOperations, Arc<ConnectionPool>>;
 
 #[allow(dead_code)]
 pub(crate) async fn make_document_ws_manager(
     doc_id: String,
     user_id: String,
     edit_cmd_tx: EditorCommandSender,
-    rev_manager: Arc<RevisionManager>,
+    rev_manager: Arc<RevisionManager<Arc<ConnectionPool>>>,
     rev_web_socket: Arc<dyn RevisionWebSocket>,
 ) -> Arc<RevisionWebSocketManager> {
     let ws_data_provider = Arc::new(WSDataProvider::new(&doc_id, Arc::new(rev_manager.clone())));