Merge pull request #1420 from AppFlowy-IO/refactor/flowy_revision_crate

Refactor: merge multiple revision into one

Author: appflowy

frontend/rust-lib/Cargo.lock

@@ -1,6 +1,6 @@
 use bytes::Bytes;
 use flowy_error::{FlowyError, FlowyResult};
-use flowy_revision::{RevisionCompress, RevisionObjectDeserializer, RevisionObjectSerializer};
+use flowy_revision::{RevisionMergeable, RevisionObjectDeserializer, RevisionObjectSerializer};
 use flowy_sync::entities::revision::Revision;
 use lib_ot::core::{Extension, NodeDataBuilder, NodeOperation, NodeTree, NodeTreeContext, Selection, Transaction};
 use lib_ot::text_delta::DeltaTextOperationBuilder;
@@ -28,9 +28,9 @@ impl Document {
         }
     }
 
-    pub fn md5(&self) -> String {
-        // format!("{:x}", md5::compute(bytes))
-        "".to_owned()
+    pub fn document_md5(&self) -> String {
+        let bytes = self.tree.to_bytes();
+        format!("{:x}", md5::compute(&bytes))
     }
 
     pub fn get_tree(&self) -> &NodeTree {
@@ -96,7 +96,7 @@ impl RevisionObjectSerializer for DocumentRevisionSerde {
 }
 
 pub(crate) struct DocumentRevisionCompress();
-impl RevisionCompress for DocumentRevisionCompress {
+impl RevisionMergeable for DocumentRevisionCompress {
     fn combine_revisions(&self, revisions: Vec<Revision>) -> FlowyResult<Bytes> {
         DocumentRevisionSerde::combine_revisions(revisions)
     }

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

@@ -29,7 +29,9 @@ impl AppFlowyDocumentEditor {
         mut rev_manager: RevisionManager<Arc<ConnectionPool>>,
         cloud_service: Arc<dyn RevisionCloudService>,
     ) -> FlowyResult<Arc<Self>> {
-        let document = rev_manager.load::<DocumentRevisionSerde>(Some(cloud_service)).await?;
+        let document = rev_manager
+            .initialize::<DocumentRevisionSerde>(Some(cloud_service))
+            .await?;
         let rev_manager = Arc::new(rev_manager);
         let command_sender = spawn_edit_queue(user, rev_manager.clone(), document);
         let doc_id = doc_id.to_string();
@@ -81,7 +83,13 @@ fn spawn_edit_queue(
 }
 
 impl DocumentEditor for Arc<AppFlowyDocumentEditor> {
-    fn close(&self) {}
+    #[tracing::instrument(name = "close document editor", level = "trace", skip_all)]
+    fn close(&self) {
+        let rev_manager = self.rev_manager.clone();
+        tokio::spawn(async move {
+            rev_manager.close().await;
+        });
+    }
 
     fn export(&self) -> FutureResult<String, FlowyError> {
         let this = self.clone();

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

@@ -63,7 +63,7 @@ impl DocumentQueue {
             Command::ComposeTransaction { transaction, ret } => {
                 self.document.write().await.apply_transaction(transaction.clone())?;
                 let _ = self
-                    .save_local_operations(transaction, self.document.read().await.md5())
+                    .save_local_operations(transaction, self.document.read().await.document_md5())
                     .await?;
                 let _ = ret.send(Ok(()));
             }
@@ -79,8 +79,7 @@ impl DocumentQueue {
     async fn save_local_operations(&self, transaction: Transaction, md5: String) -> Result<RevId, FlowyError> {
         let bytes = Bytes::from(transaction.to_bytes()?);
         let (base_rev_id, rev_id) = self.rev_manager.next_rev_id_pair();
-        let user_id = self.user.user_id()?;
-        let revision = Revision::new(&self.rev_manager.object_id, base_rev_id, rev_id, bytes, &user_id, md5);
+        let revision = Revision::new(&self.rev_manager.object_id, base_rev_id, rev_id, bytes, md5);
         let _ = self.rev_manager.add_local_revision(&revision).await?;
         Ok(rev_id.into())
     }

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

@@ -5,22 +5,22 @@ use crate::services::rev_sqlite::{SQLiteDeltaDocumentRevisionPersistence, SQLite
 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::{
-    RevisionCloudService, RevisionManager, RevisionPersistence, RevisionWebSocket, SQLiteRevisionSnapshotPersistence,
+    RevisionCloudService, RevisionManager, RevisionPersistence, RevisionPersistenceConfiguration, RevisionWebSocket,
+    SQLiteRevisionSnapshotPersistence,
 };
 use flowy_sync::client_document::initial_delta_document_content;
-use flowy_sync::entities::{
-    document::DocumentIdPB,
-    revision::{md5, RepeatedRevision, Revision},
-    ws_data::ServerRevisionWSData,
-};
+use flowy_sync::entities::{document::DocumentIdPB, revision::Revision, ws_data::ServerRevisionWSData};
+use flowy_sync::util::md5;
 use lib_infra::future::FutureResult;
+use lib_infra::ref_map::{RefCountHashMap, RefCountValue};
 use lib_ws::WSConnectState;
 use std::any::Any;
 use std::{convert::TryInto, sync::Arc};
+use tokio::sync::RwLock;
 
 pub trait DocumentUser: Send + Sync {
     fn user_dir(&self) -> Result<String, FlowyError>;
@@ -78,7 +78,7 @@ impl std::default::Default for DocumentConfig {
 pub struct DocumentManager {
     cloud_service: Arc<dyn DocumentCloudService>,
     rev_web_socket: Arc<dyn RevisionWebSocket>,
-    editor_map: Arc<DocumentEditorMap>,
+    editor_map: Arc<RwLock<RefCountHashMap<RefCountDocumentHandler>>>,
     user: Arc<dyn DocumentUser>,
     persistence: Arc<DocumentPersistence>,
     #[allow(dead_code)]
@@ -96,7 +96,7 @@ impl DocumentManager {
         Self {
             cloud_service,
             rev_web_socket,
-            editor_map: Arc::new(DocumentEditorMap::new()),
+            editor_map: Arc::new(RwLock::new(RefCountHashMap::new())),
             user: document_user,
             persistence: Arc::new(DocumentPersistence::new(database)),
             config,
@@ -126,10 +126,10 @@ impl DocumentManager {
     }
 
     #[tracing::instrument(level = "trace", skip(self, editor_id), fields(editor_id), err)]
-    pub fn close_document_editor<T: AsRef<str>>(&self, editor_id: T) -> Result<(), FlowyError> {
+    pub async fn close_document_editor<T: AsRef<str>>(&self, editor_id: T) -> Result<(), FlowyError> {
         let editor_id = editor_id.as_ref();
         tracing::Span::current().record("editor_id", &editor_id);
-        self.editor_map.remove(editor_id);
+        self.editor_map.write().await.remove(editor_id);
         Ok(())
     }
 
@@ -139,7 +139,7 @@ impl DocumentManager {
         Ok(())
     }
 
-    pub async fn create_document<T: AsRef<str>>(&self, doc_id: T, revisions: RepeatedRevision) -> FlowyResult<()> {
+    pub async fn create_document<T: AsRef<str>>(&self, doc_id: T, revisions: Vec<Revision>) -> FlowyResult<()> {
         let doc_id = doc_id.as_ref().to_owned();
         let db_pool = self.persistence.database.db_pool()?;
         // Maybe we could save the document to disk without creating the RevisionManager
@@ -151,9 +151,9 @@ impl DocumentManager {
     pub async fn receive_ws_data(&self, data: Bytes) {
         let result: Result<ServerRevisionWSData, protobuf::ProtobufError> = data.try_into();
         match result {
-            Ok(data) => match self.editor_map.get(&data.object_id) {
+            Ok(data) => match self.editor_map.read().await.get(&data.object_id) {
                 None => tracing::error!("Can't find any source handler for {:?}-{:?}", data.object_id, data.ty),
-                Some(editor) => match editor.receive_ws_data(data).await {
+                Some(handler) => match handler.0.receive_ws_data(data).await {
                     Ok(_) => {}
                     Err(e) => tracing::error!("{}", e),
                 },
@@ -182,13 +182,13 @@ impl DocumentManager {
     /// returns: Result<Arc<DocumentEditor>, FlowyError>
     ///
     async fn get_document_editor(&self, doc_id: &str) -> FlowyResult<Arc<dyn DocumentEditor>> {
-        match self.editor_map.get(doc_id) {
+        match self.editor_map.read().await.get(doc_id) {
             None => {
                 //
                 tracing::warn!("Should call init_document_editor first");
                 self.init_document_editor(doc_id).await
             }
-            Some(editor) => Ok(editor),
+            Some(handler) => Ok(handler.0.clone()),
         }
     }
 
@@ -218,14 +218,20 @@ impl DocumentManager {
                     DeltaDocumentEditor::new(doc_id, user, rev_manager, self.rev_web_socket.clone(), cloud_service)
                         .await?,
                 );
-                self.editor_map.insert(doc_id, editor.clone());
+                self.editor_map
+                    .write()
+                    .await
+                    .insert(doc_id.to_string(), RefCountDocumentHandler(editor.clone()));
                 Ok(editor)
             }
             DocumentVersionPB::V1 => {
                 let rev_manager = self.make_document_rev_manager(doc_id, pool.clone())?;
                 let editor: Arc<dyn DocumentEditor> =
                     Arc::new(AppFlowyDocumentEditor::new(doc_id, user, rev_manager, cloud_service).await?);
-                self.editor_map.insert(doc_id, editor.clone());
+                self.editor_map
+                    .write()
+                    .await
+                    .insert(doc_id.to_string(), RefCountDocumentHandler(editor.clone()));
                 Ok(editor)
             }
         }
@@ -249,7 +255,8 @@ impl DocumentManager {
     ) -> 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);
+        let configuration = RevisionPersistenceConfiguration::new(100, true);
+        let rev_persistence = RevisionPersistence::new(&user_id, doc_id, disk_cache, configuration);
         // let history_persistence = SQLiteRevisionHistoryPersistence::new(doc_id, pool.clone());
         let snapshot_persistence = SQLiteRevisionSnapshotPersistence::new(doc_id, pool);
         Ok(RevisionManager::new(
@@ -269,7 +276,8 @@ impl DocumentManager {
     ) -> 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);
+        let configuration = RevisionPersistenceConfiguration::new(100, true);
+        let rev_persistence = RevisionPersistence::new(&user_id, doc_id, disk_cache, configuration);
         // let history_persistence = SQLiteRevisionHistoryPersistence::new(doc_id, pool.clone());
         let snapshot_persistence = SQLiteRevisionSnapshotPersistence::new(doc_id, pool);
         Ok(RevisionManager::new(
@@ -294,63 +302,47 @@ impl RevisionCloudService for DocumentRevisionCloudService {
         let params: DocumentIdPB = object_id.to_string().into();
         let server = self.server.clone();
         let token = self.token.clone();
-        let user_id = user_id.to_string();
 
         FutureResult::new(async move {
             match server.fetch_document(&token, params).await? {
                 None => Err(FlowyError::record_not_found().context("Remote doesn't have this document")),
                 Some(payload) => {
                     let bytes = Bytes::from(payload.content.clone());
                     let doc_md5 = md5(&bytes);
-                    let revision = Revision::new(
-                        &payload.doc_id,
-                        payload.base_rev_id,
-                        payload.rev_id,
-                        bytes,
-                        &user_id,
-                        doc_md5,
-                    );
+                    let revision = Revision::new(&payload.doc_id, payload.base_rev_id, payload.rev_id, bytes, doc_md5);
                     Ok(vec![revision])
                 }
             }
         })
     }
 }
 
-pub struct DocumentEditorMap {
-    inner: DashMap<String, Arc<dyn DocumentEditor>>,
-}
+#[derive(Clone)]
+struct RefCountDocumentHandler(Arc<dyn DocumentEditor>);
 
-impl DocumentEditorMap {
-    fn new() -> Self {
-        Self { inner: DashMap::new() }
-    }
-
-    pub(crate) fn insert(&self, editor_id: &str, editor: Arc<dyn DocumentEditor>) {
-        if self.inner.contains_key(editor_id) {
-            log::warn!("Editor:{} already open", editor_id);
-        }
-        self.inner.insert(editor_id.to_string(), editor);
+impl RefCountValue for RefCountDocumentHandler {
+    fn did_remove(&self) {
+        self.0.close();
     }
+}
 
-    pub(crate) fn get(&self, editor_id: &str) -> Option<Arc<dyn DocumentEditor>> {
-        Some(self.inner.get(editor_id)?.clone())
-    }
+impl std::ops::Deref for RefCountDocumentHandler {
+    type Target = Arc<dyn DocumentEditor>;
 
-    pub(crate) fn remove(&self, editor_id: &str) {
-        if let Some(editor) = self.get(editor_id) {
-            editor.close()
-        }
-        self.inner.remove(editor_id);
+    fn deref(&self) -> &Self::Target {
+        &self.0
     }
 }
 
 #[tracing::instrument(level = "trace", skip(web_socket, handlers))]
-fn listen_ws_state_changed(web_socket: Arc<dyn RevisionWebSocket>, handlers: Arc<DocumentEditorMap>) {
+fn listen_ws_state_changed(
+    web_socket: Arc<dyn RevisionWebSocket>,
+    handlers: Arc<RwLock<RefCountHashMap<RefCountDocumentHandler>>>,
+) {
     tokio::spawn(async move {
         let mut notify = web_socket.subscribe_state_changed().await;
         while let Ok(state) = notify.recv().await {
-            handlers.inner.iter().for_each(|handler| {
+            handlers.read().await.values().iter().for_each(|handler| {
                 handler.receive_ws_state(&state);
             })
         }

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

@@ -6,7 +6,7 @@ use bytes::Bytes;
 use flowy_database::ConnectionPool;
 use flowy_error::{internal_error, FlowyResult};
 use flowy_revision::{
-    RevisionCloudService, RevisionCompress, RevisionManager, RevisionObjectDeserializer, RevisionObjectSerializer,
+    RevisionCloudService, RevisionManager, RevisionMergeable, RevisionObjectDeserializer, RevisionObjectSerializer,
     RevisionWebSocket,
 };
 use flowy_sync::entities::ws_data::ServerRevisionWSData;
@@ -45,7 +45,7 @@ impl DeltaDocumentEditor {
         cloud_service: Arc<dyn RevisionCloudService>,
     ) -> FlowyResult<Arc<Self>> {
         let document = rev_manager
-            .load::<DeltaDocumentRevisionSerde>(Some(cloud_service))
+            .initialize::<DeltaDocumentRevisionSerde>(Some(cloud_service))
             .await?;
         let operations = DeltaTextOperations::from_bytes(&document.content)?;
         let rev_manager = Arc::new(rev_manager);
@@ -270,7 +270,7 @@ impl RevisionObjectSerializer for DeltaDocumentRevisionSerde {
 }
 
 pub(crate) struct DeltaDocumentRevisionCompress();
-impl RevisionCompress for DeltaDocumentRevisionCompress {
+impl RevisionMergeable for DeltaDocumentRevisionCompress {
     fn combine_revisions(&self, revisions: Vec<Revision>) -> FlowyResult<Bytes> {
         DeltaDocumentRevisionSerde::combine_revisions(revisions)
     }