add server document

backend/src/context.rs

@@ -9,7 +9,7 @@ use crate::services::document::{
     persistence::DocumentKVPersistence,
     ws_receiver::{make_document_ws_receiver, HttpDocumentCloudPersistence},
 };
-use flowy_collaboration::sync::ServerDocumentManager;
+use flowy_collaboration::server_document::ServerDocumentManager;
 use lib_ws::WSModule;
 use sqlx::PgPool;
 use std::sync::Arc;

backend/src/services/document/persistence.rs

@@ -14,7 +14,7 @@ use flowy_collaboration::{
         ResetDocumentParams,
         Revision as RevisionPB,
     },
-    sync::ServerDocumentManager,
+    server_document::ServerDocumentManager,
     util::make_doc_from_revisions,
 };
 

backend/src/services/document/router.rs

@@ -14,7 +14,7 @@ use flowy_collaboration::{
         DocumentId as DocumentIdPB,
         ResetDocumentParams as ResetDocumentParamsPB,
     },
-    sync::ServerDocumentManager,
+    server_document::ServerDocumentManager,
 };
 use std::sync::Arc;
 

backend/src/services/document/ws_actor.rs

@@ -13,7 +13,7 @@ use flowy_collaboration::{
         DocumentClientWSDataType as DocumentClientWSDataTypePB,
         Revision as RevisionPB,
     },
-    sync::{RevisionUser, ServerDocumentManager, SyncResponse},
+    server_document::{RevisionUser, ServerDocumentManager, SyncResponse},
 };
 use futures::stream::StreamExt;
 use std::sync::Arc;

backend/src/services/document/ws_receiver.rs

@@ -18,7 +18,7 @@ use flowy_collaboration::{
         RepeatedRevision as RepeatedRevisionPB,
         Revision as RevisionPB,
     },
-    sync::{DocumentCloudPersistence, ServerDocumentManager},
+    server_document::{DocumentCloudPersistence, ServerDocumentManager},
     util::repeated_revision_from_repeated_revision_pb,
 };
 use lib_infra::future::BoxResultFuture;

backend/tests/api_test/workspace_test.rs

@@ -2,7 +2,7 @@
 
 use crate::util::helper::{ViewTest, *};
 use flowy_collaboration::{
-    document::{Document, PlainDoc},
+    client_document::{ClientDocument, PlainDoc},
     entities::{
         doc::{CreateDocParams, DocumentId},
         revision::{md5, RepeatedRevision, Revision},
@@ -240,7 +240,7 @@ async fn doc_create() {
     let server = TestUserServer::new().await;
     let doc_id = uuid::Uuid::new_v4().to_string();
     let user_id = "a".to_owned();
-    let mut document = Document::new::<PlainDoc>();
+    let mut document = ClientDocument::new::<PlainDoc>();
     let mut offset = 0;
     for i in 0..1000 {
         let content = i.to_string();

backend/tests/document_test/edit_script.rs

@@ -16,7 +16,7 @@ use parking_lot::RwLock;
 use backend::services::document::persistence::{read_document, reset_document};
 use flowy_collaboration::entities::revision::{RepeatedRevision, Revision};
 use flowy_collaboration::protobuf::{RepeatedRevision as RepeatedRevisionPB, DocumentId as DocumentIdPB};
-use flowy_collaboration::sync::ServerDocumentManager;
+use flowy_collaboration::server_document::ServerDocumentManager;
 use flowy_net::ws::connection::FlowyWebSocketConnect;
 use lib_ot::core::Interval;
 

backend/tests/document_test/edit_test.rs

@@ -1,5 +1,5 @@
 use crate::document_test::edit_script::{DocScript, DocumentTest};
-use flowy_collaboration::document::{Document, NewlineDoc};
+use flowy_collaboration::client_document::{ClientDocument, NewlineDoc};
 use lib_ot::{core::Interval, rich_text::RichTextAttribute};
 
 #[rustfmt::skip]
@@ -75,7 +75,7 @@ async fn delta_sync_while_editing_with_attribute() {
 #[actix_rt::test]
 async fn delta_sync_with_server_push() {
     let test = DocumentTest::new().await;
-    let mut document = Document::new::<NewlineDoc>();
+    let mut document = ClientDocument::new::<NewlineDoc>();
     document.insert(0, "123").unwrap();
     document.insert(3, "456").unwrap();
     let json = document.to_json();
@@ -109,7 +109,7 @@ async fn delta_sync_with_server_push() {
 #[actix_rt::test]
 async fn delta_sync_with_server_push_after_reset_document() {
     let test = DocumentTest::new().await;
-    let mut document = Document::new::<NewlineDoc>();
+    let mut document = ClientDocument::new::<NewlineDoc>();
     document.insert(0, "123").unwrap();
     let json = document.to_json();
 
@@ -148,7 +148,7 @@ async fn delta_sync_with_server_push_after_reset_document() {
 #[actix_rt::test]
 async fn delta_sync_while_local_rev_less_than_server_rev() {
     let test = DocumentTest::new().await;
-    let mut document = Document::new::<NewlineDoc>();
+    let mut document = ClientDocument::new::<NewlineDoc>();
     document.insert(0, "123").unwrap();
     let json = document.to_json();
 
@@ -190,7 +190,7 @@ async fn delta_sync_while_local_rev_less_than_server_rev() {
 #[actix_rt::test]
 async fn delta_sync_while_local_rev_greater_than_server_rev() {
     let test = DocumentTest::new().await;
-    let mut document = Document::new::<NewlineDoc>();
+    let mut document = ClientDocument::new::<NewlineDoc>();
     document.insert(0, "123").unwrap();
     let json = document.to_json();
 

backend/tests/util/helper.rs

@@ -8,7 +8,7 @@ use backend_service::{
     errors::ServerError,
 };
 use flowy_collaboration::{
-    document::default::initial_delta_string,
+    client_document::default::initial_delta_string,
     entities::doc::{CreateDocParams, DocumentId, DocumentInfo},
 };
 use flowy_core_data_model::entities::prelude::*;

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

@@ -1,5 +1,5 @@
 use chrono::Utc;
-use flowy_collaboration::document::default::{initial_delta, initial_read_me};
+use flowy_collaboration::client_document::default::{initial_delta, initial_read_me};
 use flowy_core_data_model::{entities::view::CreateViewParams, user_default};
 use lazy_static::lazy_static;
 use parking_lot::RwLock;

frontend/rust-lib/flowy-core/src/services/view/controller.rs

@@ -138,7 +138,7 @@ impl ViewController {
         })
     }
 
-    #[tracing::instrument(level = "debug", skip(self,params), fields(doc_id = %params.doc_id), err)]
+    #[tracing::instrument(level = "debug", skip(self, params), err)]
     pub(crate) async fn close_view(&self, params: DocumentId) -> Result<(), FlowyError> {
         let _ = self.document_ctx.controller.close_document(&params.doc_id)?;
         Ok(())

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

@@ -155,12 +155,15 @@ impl ClientDocumentEditor {
         Ok(())
     }
 
-    #[tracing::instrument(level = "debug", skip(self))]
     pub fn stop(&self) { self.ws_manager.stop(); }
 
     pub(crate) fn ws_handler(&self) -> Arc<dyn DocumentWSReceiver> { self.ws_manager.clone() }
 }
 
+impl std::ops::Drop for ClientDocumentEditor {
+    fn drop(&mut self) { tracing::trace!("{} ClientDocumentEditor was dropped", self.doc_id) }
+}
+
 // The edit queue will exit after the EditorCommandSender was dropped.
 fn spawn_edit_queue(
     user: Arc<dyn DocumentUser>,

frontend/rust-lib/flowy-document/src/core/edit/queue.rs

@@ -4,7 +4,7 @@ use crate::{
 };
 use async_stream::stream;
 use flowy_collaboration::{
-    document::{history::UndoResult, Document, NewlineDoc},
+    client_document::{history::UndoResult, ClientDocument, NewlineDoc},
     entities::revision::{RepeatedRevision, RevId, Revision},
     errors::CollaborateError,
     util::make_delta_from_revisions,
@@ -21,7 +21,7 @@ use tokio::sync::{oneshot, RwLock};
 // The EditorCommandQueue executes each command that will alter the document in
 // serial.
 pub(crate) struct EditorCommandQueue {
-    document: Arc<RwLock<Document>>,
+    document: Arc<RwLock<ClientDocument>>,
     user: Arc<dyn DocumentUser>,
     rev_manager: Arc<DocumentRevisionManager>,
     receiver: Option<EditorCommandReceiver>,
@@ -34,7 +34,7 @@ impl EditorCommandQueue {
         delta: RichTextDelta,
         receiver: EditorCommandReceiver,
     ) -> Self {
-        let document = Arc::new(RwLock::new(Document::from_delta(delta)));
+        let document = Arc::new(RwLock::new(ClientDocument::from_delta(delta)));
         Self {
             document,
             user,

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

@@ -14,7 +14,7 @@ use flowy_collaboration::{
 use flowy_error::FlowyResult;
 use futures_util::{future, stream, stream::StreamExt};
 use lib_infra::future::FutureResult;
-use lib_ot::{core::Operation, errors::OTError, rich_text::RichTextDelta};
+use lib_ot::{core::Operation, rich_text::RichTextDelta};
 use std::{collections::VecDeque, sync::Arc};
 use tokio::sync::RwLock;
 
@@ -26,20 +26,20 @@ pub struct DocumentRevisionManager {
     pub(crate) doc_id: String,
     user_id: String,
     rev_id_counter: RevIdCounter,
-    cache: Arc<DocumentRevisionCache>,
+    revision_cache: Arc<DocumentRevisionCache>,
-    sync_seq: Arc<RevisionSyncSequence>,
+    revision_sync_seq: Arc<RevisionSyncSequence>,
 }
 
 impl DocumentRevisionManager {
-    pub fn new(user_id: &str, doc_id: &str, cache: Arc<DocumentRevisionCache>) -> Self {
+    pub fn new(user_id: &str, doc_id: &str, revision_cache: Arc<DocumentRevisionCache>) -> Self {
         let rev_id_counter = RevIdCounter::new(0);
-        let sync_seq = Arc::new(RevisionSyncSequence::new());
+        let revision_sync_seq = Arc::new(RevisionSyncSequence::new());
         Self {
             doc_id: doc_id.to_string(),
             user_id: user_id.to_owned(),
             rev_id_counter,
-            cache,
+            revision_cache,
-            sync_seq,
+            revision_sync_seq,
         }
     }
 
@@ -48,7 +48,8 @@ impl DocumentRevisionManager {
             doc_id: self.doc_id.clone(),
             user_id: self.user_id.clone(),
             server,
-            cache: self.cache.clone(),
+            revision_cache: self.revision_cache.clone(),
+            revision_sync_seq: self.revision_sync_seq.clone(),
         }
         .load()
         .await?;
@@ -61,7 +62,7 @@ impl DocumentRevisionManager {
     pub async fn reset_document(&self, revisions: RepeatedRevision) -> FlowyResult<()> {
         let rev_id = pair_rev_id_from_revisions(&revisions).1;
         let _ = self
-            .cache
+            .revision_cache
             .reset_with_revisions(&self.doc_id, revisions.into_inner())
             .await?;
         self.rev_id_counter.set(rev_id);
@@ -73,7 +74,10 @@ impl DocumentRevisionManager {
         if revision.delta_data.is_empty() {
             return Err(FlowyError::internal().context("Delta data should be empty"));
         }
-        let _ = self.cache.add(revision.clone(), RevisionState::Ack, true).await?;
+        let _ = self
+            .revision_cache
+            .add(revision.clone(), RevisionState::Ack, true)
+            .await?;
         self.rev_id_counter.set(revision.rev_id);
         Ok(())
     }
@@ -84,15 +88,18 @@ impl DocumentRevisionManager {
             return Err(FlowyError::internal().context("Delta data should be empty"));
         }
 
-        let record = self.cache.add(revision.clone(), RevisionState::Local, true).await?;
+        let record = self
-        self.sync_seq.add_revision(record).await?;
+            .revision_cache
+            .add(revision.clone(), RevisionState::Local, true)
+            .await?;
+        self.revision_sync_seq.add_revision_record(record).await?;
         Ok(())
     }
 
     #[tracing::instrument(level = "debug", skip(self), err)]
     pub async fn ack_revision(&self, rev_id: i64) -> Result<(), FlowyError> {
-        if self.sync_seq.ack(&rev_id).await.is_ok() {
+        if self.revision_sync_seq.ack(&rev_id).await.is_ok() {
-            self.cache.ack(rev_id).await;
+            self.revision_cache.ack(rev_id).await;
         }
         Ok(())
     }
@@ -109,28 +116,28 @@ impl DocumentRevisionManager {
 
     pub async fn get_revisions_in_range(&self, range: RevisionRange) -> Result<Vec<Revision>, FlowyError> {
         debug_assert!(range.doc_id == self.doc_id);
-        let revisions = self.cache.revisions_in_range(range.clone()).await?;
+        let revisions = self.revision_cache.revisions_in_range(range.clone()).await?;
         Ok(revisions)
     }
 
     pub fn next_sync_revision(&self) -> FutureResult<Option<Revision>, FlowyError> {
-        let sync_seq = self.sync_seq.clone();
+        let revision_sync_seq = self.revision_sync_seq.clone();
-        let cache = self.cache.clone();
+        let revision_cache = self.revision_cache.clone();
         FutureResult::new(async move {
-            match sync_seq.next_sync_revision().await {
+            match revision_sync_seq.next_sync_revision_record().await {
-                None => match sync_seq.next_sync_rev_id().await {
+                None => match revision_sync_seq.next_sync_rev_id().await {
                     None => Ok(None),
-                    Some(rev_id) => Ok(cache.get(rev_id).await.map(|record| record.revision)),
+                    Some(rev_id) => Ok(revision_cache.get(rev_id).await.map(|record| record.revision)),
                 },
                 Some((_, record)) => Ok(Some(record.revision)),
             }
         })
     }
 
-    pub async fn latest_revision(&self) -> Revision { self.cache.latest_revision().await }
+    pub async fn latest_revision(&self) -> Revision { self.revision_cache.latest_revision().await }
 
     pub async fn get_revision(&self, rev_id: i64) -> Option<Revision> {
-        self.cache.get(rev_id).await.map(|record| record.revision)
+        self.revision_cache.get(rev_id).await.map(|record| record.revision)
     }
 }
 
@@ -152,12 +159,17 @@ impl std::default::Default for RevisionSyncSequence {
 impl RevisionSyncSequence {
     fn new() -> Self { RevisionSyncSequence::default() }
 
-    async fn add_revision(&self, record: RevisionRecord) -> Result<(), OTError> {
+    async fn add_revision_record(&self, record: RevisionRecord) -> FlowyResult<()> {
+        if !record.state.is_local() {
+            return Ok(());
+        }
+
         // The last revision's rev_id must be greater than the new one.
         if let Some(rev_id) = self.local_revs.read().await.back() {
             if *rev_id >= record.revision.rev_id {
-                return Err(OTError::revision_id_conflict()
+                return Err(
-                    .context(format!("The new revision's id must be greater than {}", rev_id)));
+                    FlowyError::internal().context(format!("The new revision's id must be greater than {}", rev_id))
+                );
             }
         }
         self.local_revs.write().await.push_back(record.revision.rev_id);
@@ -181,7 +193,7 @@ impl RevisionSyncSequence {
         Ok(())
     }
 
-    async fn next_sync_revision(&self) -> Option<(i64, RevisionRecord)> {
+    async fn next_sync_revision_record(&self) -> Option<(i64, RevisionRecord)> {
         match self.local_revs.read().await.front() {
             None => None,
             Some(rev_id) => self.revs_map.get(rev_id).map(|r| (*r.key(), r.value().clone())),
@@ -195,12 +207,13 @@ struct RevisionLoader {
     doc_id: String,
     user_id: String,
     server: Arc<dyn RevisionServer>,
-    cache: Arc<DocumentRevisionCache>,
+    revision_cache: Arc<DocumentRevisionCache>,
+    revision_sync_seq: Arc<RevisionSyncSequence>,
 }
 
 impl RevisionLoader {
     async fn load(&self) -> Result<Vec<Revision>, FlowyError> {
-        let records = self.cache.batch_get(&self.doc_id)?;
+        let records = self.revision_cache.batch_get(&self.doc_id)?;
         let revisions: Vec<Revision>;
         if records.is_empty() {
             let doc = self.server.fetch_document(&self.doc_id).await?;
@@ -214,16 +227,24 @@ impl RevisionLoader {
                 &self.user_id,
                 doc_md5,
             );
-            let _ = self.cache.add(revision.clone(), RevisionState::Ack, true).await?;
+            let _ = self
+                .revision_cache
+                .add(revision.clone(), RevisionState::Ack, true)
+                .await?;
             revisions = vec![revision];
         } else {
-            // Sync the records if their state is RevisionState::Local.
             stream::iter(records.clone())
                 .filter(|record| future::ready(record.state == RevisionState::Local))
                 .for_each(|record| async move {
-                    match self.cache.add(record.revision, record.state, false).await {
+                    let f = || async {
+                        // Sync the records if their state is RevisionState::Local.
+                        let _ = self.revision_sync_seq.add_revision_record(record.clone()).await?;
+                        let _ = self.revision_cache.add(record.revision, record.state, false).await?;
+                        Ok::<(), FlowyError>(())
+                    };
+                    match f().await {
                         Ok(_) => {},
-                        Err(e) => tracing::error!("{}", e),
+                        Err(e) => tracing::error!("[RevisionLoader]: {}", e),
                     }
                 })
                 .await;
@@ -274,5 +295,5 @@ impl RevisionSyncSequence {
 
 #[cfg(feature = "flowy_unit_test")]
 impl DocumentRevisionManager {
-    pub fn revision_cache(&self) -> Arc<DocumentRevisionCache> { self.cache.clone() }
+    pub fn revision_cache(&self) -> Arc<DocumentRevisionCache> { self.revision_cache.clone() }
 }

frontend/rust-lib/flowy-document/src/core/web_socket/ws_manager.rs

@@ -96,7 +96,7 @@ impl DocumentWebSocketManager {
 
     pub(crate) fn stop(&self) {
         if self.stop_sync_tx.send(()).is_ok() {
-            tracing::debug!("{} stop sync", self.doc_id)
+            tracing::trace!("{} stop sync", self.doc_id)
         }
     }
 }
@@ -134,6 +134,10 @@ pub struct DocumentWSStream {
     stop_rx: Option<SinkStopRx>,
 }
 
+impl std::ops::Drop for DocumentWSStream {
+    fn drop(&mut self) { tracing::trace!("{} DocumentWSStream was dropped", self.doc_id) }
+}
+
 impl DocumentWSStream {
     pub fn new(
         doc_id: &str,
@@ -282,6 +286,10 @@ impl DocumentWSSink {
     }
 }
 
+impl std::ops::Drop for DocumentWSSink {
+    fn drop(&mut self) { tracing::trace!("{} DocumentWSSink was dropped", self.doc_id) }
+}
+
 async fn tick(sender: mpsc::Sender<()>) {
     let mut interval = interval(Duration::from_millis(SYNC_INTERVAL_IN_MILLIS));
     while sender.send(()).await.is_ok() {

frontend/rust-lib/flowy-document/tests/editor/attribute_test.rs

@@ -1,6 +1,6 @@
 #![cfg_attr(rustfmt, rustfmt::skip)]
 use crate::editor::{TestBuilder, TestOp::*};
-use flowy_collaboration::document::{NewlineDoc, PlainDoc};
+use flowy_collaboration::client_document::{NewlineDoc, PlainDoc};
 use lib_ot::core::{Interval, OperationTransformable, NEW_LINE, WHITESPACE, FlowyStr};
 use unicode_segmentation::UnicodeSegmentation;
 use lib_ot::rich_text::RichTextDelta;

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

@@ -5,7 +5,7 @@ mod serde_test;
 mod undo_redo_test;
 
 use derive_more::Display;
-use flowy_collaboration::document::{Document, InitialDocumentText};
+use flowy_collaboration::client_document::{ClientDocument, InitialDocumentText};
 use lib_ot::{
     core::*,
     rich_text::{RichTextAttribute, RichTextAttributes, RichTextDelta},
@@ -82,7 +82,7 @@ pub enum TestOp {
 }
 
 pub struct TestBuilder {
-    documents: Vec<Document>,
+    documents: Vec<ClientDocument>,
     deltas: Vec<Option<RichTextDelta>>,
     primes: Vec<Option<RichTextDelta>>,
 }
@@ -266,7 +266,7 @@ impl TestBuilder {
     }
 
     pub fn run_scripts<C: InitialDocumentText>(mut self, scripts: Vec<TestOp>) {
-        self.documents = vec![Document::new::<C>(), Document::new::<C>()];
+        self.documents = vec![ClientDocument::new::<C>(), ClientDocument::new::<C>()];
         self.primes = vec![None, None];
         self.deltas = vec![None, None];
         for (_i, op) in scripts.iter().enumerate() {

frontend/rust-lib/flowy-document/tests/editor/op_test.rs

@@ -1,6 +1,6 @@
 #![allow(clippy::all)]
 use crate::editor::{Rng, TestBuilder, TestOp::*};
-use flowy_collaboration::document::{NewlineDoc, PlainDoc};
+use flowy_collaboration::client_document::{NewlineDoc, PlainDoc};
 use lib_ot::{
     core::*,
     rich_text::{AttributeBuilder, RichTextAttribute, RichTextAttributes, RichTextDelta},

frontend/rust-lib/flowy-document/tests/editor/serde_test.rs

@@ -1,4 +1,4 @@
-use flowy_collaboration::document::{Document, PlainDoc};
+use flowy_collaboration::client_document::{ClientDocument, PlainDoc};
 use lib_ot::{
     core::*,
     rich_text::{AttributeBuilder, RichTextAttribute, RichTextAttributeValue, RichTextDelta},
@@ -104,10 +104,13 @@ fn delta_serde_null_test() {
 
 #[test]
 fn document_insert_serde_test() {
-    let mut document = Document::new::<PlainDoc>();
+    let mut document = ClientDocument::new::<PlainDoc>();
     document.insert(0, "\n").unwrap();
     document.insert(0, "123").unwrap();
     let json = document.to_json();
     assert_eq!(r#"[{"insert":"123\n"}]"#, json);
-    assert_eq!(r#"[{"insert":"123\n"}]"#, Document::from_json(&json).unwrap().to_json());
+    assert_eq!(
+        r#"[{"insert":"123\n"}]"#,
+        ClientDocument::from_json(&json).unwrap().to_json()
+    );
 }

frontend/rust-lib/flowy-document/tests/editor/undo_redo_test.rs

@@ -1,5 +1,5 @@
 use crate::editor::{TestBuilder, TestOp::*};
-use flowy_collaboration::document::{NewlineDoc, PlainDoc, RECORD_THRESHOLD};
+use flowy_collaboration::client_document::{NewlineDoc, PlainDoc, RECORD_THRESHOLD};
 use lib_ot::core::{Interval, NEW_LINE, WHITESPACE};
 
 #[test]

frontend/rust-lib/flowy-net/src/cloud/document.rs

@@ -4,7 +4,7 @@ use backend_service::{
     response::FlowyResponse,
 };
 use flowy_collaboration::{
-    document::default::initial_delta_string,
+    client_document::default::initial_delta_string,
     entities::doc::{CreateDocParams, DocumentId, DocumentInfo, ResetDocumentParams},
 };
 use flowy_error::FlowyError;

frontend/rust-lib/flowy-net/src/ws/local/local_server.rs

@@ -4,14 +4,14 @@ use flowy_collaboration::{
     entities::ws::{DocumentClientWSData, DocumentClientWSDataType},
     errors::CollaborateError,
     protobuf::DocumentClientWSData as DocumentClientWSDataPB,
-    sync::*,
+    server_document::*,
 };
 use lib_ws::{WSModule, WebSocketRawMessage};
 use std::{convert::TryInto, fmt::Debug, sync::Arc};
 use tokio::sync::{mpsc, mpsc::UnboundedSender};
 
 pub struct LocalDocumentServer {
-    pub doc_manager: Arc<ServerDocumentManager>,
+    doc_manager: Arc<ServerDocumentManager>,
     sender: mpsc::UnboundedSender<WebSocketRawMessage>,
 }
 

frontend/rust-lib/flowy-net/src/ws/local/local_ws.rs

@@ -50,22 +50,21 @@ impl std::default::Default for LocalWebSocket {
 }
 
 impl LocalWebSocket {
-    fn restart_ws_receiver(&self) -> mpsc::Receiver<()> {
+    fn cancel_pre_spawn_client(&self) {
         if let Some(stop_tx) = self.local_server_stop_tx.read().clone() {
             tokio::spawn(async move {
                 let _ = stop_tx.send(()).await;
             });
         }
-        let (stop_tx, stop_rx) = mpsc::channel::<()>(1);
-        *self.local_server_stop_tx.write() = Some(stop_tx);
-        stop_rx
     }
 
     fn spawn_client_ws_receiver(&self, _addr: String) {
         let mut ws_receiver = self.ws_sender.subscribe();
         let local_server = self.local_server.clone();
         let user_id = self.user_id.clone();
-        let mut stop_rx = self.restart_ws_receiver();
+        let _ = self.cancel_pre_spawn_client();
+        let (stop_tx, mut stop_rx) = mpsc::channel::<()>(1);
+        *self.local_server_stop_tx.write() = Some(stop_tx);
         tokio::spawn(async move {
             loop {
                 tokio::select! {

frontend/rust-lib/flowy-net/src/ws/local/persistence.rs

@@ -4,7 +4,7 @@ use flowy_collaboration::{
     entities::doc::DocumentInfo,
     errors::CollaborateError,
     protobuf::{RepeatedRevision as RepeatedRevisionPB, Revision as RevisionPB},
-    sync::*,
+    server_document::*,
     util::{make_doc_from_revisions, repeated_revision_from_repeated_revision_pb},
 };
 use lib_infra::future::BoxResultFuture;

shared-lib/flowy-collaboration/src/client_document/default/mod.rs

@@ -14,7 +14,7 @@ pub fn initial_read_me() -> RichTextDelta {
 
 #[cfg(test)]
 mod tests {
-    use crate::document::default::initial_read_me;
+    use crate::client_document::default::initial_read_me;
 
     #[test]
     fn load_read_me() {

shared-lib/flowy-collaboration/src/client_document/document_pad.rs

@@ -1,5 +1,5 @@
 use crate::{
-    document::{
+    client_document::{
         default::initial_delta,
         history::{History, UndoResult},
         view::{View, RECORD_THRESHOLD},
@@ -26,7 +26,7 @@ impl InitialDocumentText for NewlineDoc {
     fn initial_delta() -> RichTextDelta { initial_delta() }
 }
 
-pub struct Document {
+pub struct ClientDocument {
     delta: RichTextDelta,
     history: History,
     view: View,
@@ -34,11 +34,11 @@ pub struct Document {
     notify: Option<mpsc::UnboundedSender<()>>,
 }
 
-impl Document {
+impl ClientDocument {
     pub fn new<C: InitialDocumentText>() -> Self { Self::from_delta(C::initial_delta()) }
 
     pub fn from_delta(delta: RichTextDelta) -> Self {
-        Document {
+        ClientDocument {
             delta,
             history: History::new(),
             view: View::new(),
@@ -185,7 +185,7 @@ impl Document {
     pub fn is_empty<C: InitialDocumentText>(&self) -> bool { self.delta == C::initial_delta() }
 }
 
-impl Document {
+impl ClientDocument {
     fn invert(&self, delta: &RichTextDelta) -> Result<(RichTextDelta, RichTextDelta), CollaborateError> {
         // c = a.compose(b)
         // d = b.invert(a)

shared-lib/flowy-collaboration/src/client_document/extensions/delete/default_delete.rs

@@ -1,4 +1,4 @@
-use crate::document::DeleteExt;
+use crate::client_document::DeleteExt;
 use lib_ot::{
     core::{DeltaBuilder, Interval},
     rich_text::RichTextDelta,

shared-lib/flowy-collaboration/src/client_document/extensions/delete/preserve_line_format_merge.rs

@@ -1,4 +1,4 @@
-use crate::{document::DeleteExt, util::is_newline};
+use crate::{client_document::DeleteExt, util::is_newline};
 use lib_ot::{
     core::{Attributes, DeltaBuilder, DeltaIter, Interval, Utf16CodeUnitMetric, NEW_LINE},
     rich_text::{plain_attributes, RichTextDelta},

shared-lib/flowy-collaboration/src/client_document/extensions/format/resolve_block_format.rs

@@ -4,7 +4,7 @@ use lib_ot::{
 };
 
 use crate::{
-    document::{extensions::helper::line_break, FormatExt},
+    client_document::{extensions::helper::line_break, FormatExt},
     util::find_newline,
 };
 

shared-lib/flowy-collaboration/src/client_document/extensions/format/resolve_inline_format.rs

@@ -4,7 +4,7 @@ use lib_ot::{
 };
 
 use crate::{
-    document::{extensions::helper::line_break, FormatExt},
+    client_document::{extensions::helper::line_break, FormatExt},
     util::find_newline,
 };
 

shared-lib/flowy-collaboration/src/client_document/extensions/insert/auto_exit_block.rs

@@ -1,4 +1,4 @@
-use crate::{document::InsertExt, util::is_newline};
+use crate::{client_document::InsertExt, util::is_newline};
 use lib_ot::{
     core::{is_empty_line_at_index, DeltaBuilder, DeltaIter},
     rich_text::{attributes_except_header, RichTextAttributeKey, RichTextDelta},

shared-lib/flowy-collaboration/src/client_document/extensions/insert/auto_format.rs

@@ -1,4 +1,4 @@
-use crate::{document::InsertExt, util::is_whitespace};
+use crate::{client_document::InsertExt, util::is_whitespace};
 use lib_ot::{
     core::{count_utf16_code_units, DeltaBuilder, DeltaIter},
     rich_text::{plain_attributes, RichTextAttribute, RichTextAttributes, RichTextDelta},

shared-lib/flowy-collaboration/src/client_document/extensions/insert/default_insert.rs

@@ -1,4 +1,4 @@
-use crate::document::InsertExt;
+use crate::client_document::InsertExt;
 use lib_ot::{
     core::{Attributes, DeltaBuilder, DeltaIter, NEW_LINE},
     rich_text::{RichTextAttributeKey, RichTextAttributes, RichTextDelta},

shared-lib/flowy-collaboration/src/client_document/extensions/insert/mod.rs

@@ -1,4 +1,4 @@
-use crate::document::InsertExt;
+use crate::client_document::InsertExt;
 pub use auto_exit_block::*;
 pub use auto_format::*;
 pub use default_insert::*;

shared-lib/flowy-collaboration/src/client_document/extensions/insert/preserve_block_format.rs

@@ -1,4 +1,4 @@
-use crate::{document::InsertExt, util::is_newline};
+use crate::{client_document::InsertExt, util::is_newline};
 use lib_ot::{
     core::{DeltaBuilder, DeltaIter, NEW_LINE},
     rich_text::{

shared-lib/flowy-collaboration/src/client_document/extensions/insert/preserve_inline_format.rs

@@ -1,5 +1,5 @@
 use crate::{
-    document::InsertExt,
+    client_document::InsertExt,
     util::{contain_newline, is_newline},
 };
 use lib_ot::{

shared-lib/flowy-collaboration/src/client_document/extensions/insert/reset_format_on_new_line.rs

@@ -1,4 +1,4 @@
-use crate::{document::InsertExt, util::is_newline};
+use crate::{client_document::InsertExt, util::is_newline};
 use lib_ot::{
     core::{DeltaBuilder, DeltaIter, Utf16CodeUnitMetric, NEW_LINE},
     rich_text::{RichTextAttributeKey, RichTextAttributes, RichTextDelta},

shared-lib/flowy-collaboration/src/client_document/mod.rs

@@ -1,12 +1,12 @@
 #![allow(clippy::module_inception)]
 
-pub use document::*;
+pub use document_pad::*;
 pub(crate) use extensions::*;
 pub use view::*;
 
 mod data;
 pub mod default;
-mod document;
+mod document_pad;
 mod extensions;
 pub mod history;
 mod view;

shared-lib/flowy-collaboration/src/client_document/view.rs

@@ -1,4 +1,4 @@
-use crate::document::*;
+use crate::client_document::*;
 use lib_ot::{
     core::{trim, Interval},
     errors::{ErrorBuilder, OTError, OTErrorCode},

shared-lib/flowy-collaboration/src/entities/revision.rs

@@ -181,6 +181,15 @@ pub enum RevisionState {
     Ack   = 1,
 }
 
+impl RevisionState {
+    pub fn is_local(&self) -> bool {
+        match self {
+            RevisionState::Local => true,
+            RevisionState::Ack => false,
+        }
+    }
+}
+
 impl AsRef<RevisionState> for RevisionState {
     fn as_ref(&self) -> &RevisionState { &self }
 }

shared-lib/flowy-collaboration/src/lib.rs

@@ -1,8 +1,7 @@
-pub mod document;
+pub mod client_document;
 pub mod entities;
 pub mod errors;
 pub mod protobuf;
-pub mod sync;
+pub mod server_document;
 pub mod util;
-
 pub use lib_ot::rich_text::RichTextDelta;

shared-lib/flowy-collaboration/src/server_document/document_manager.rs

@@ -1,9 +1,8 @@
 use crate::{
-    document::Document,
     entities::{doc::DocumentInfo, ws::DocumentServerWSDataBuilder},
     errors::{internal_error, CollaborateError, CollaborateResult},
     protobuf::{DocumentClientWSData, RepeatedRevision as RepeatedRevisionPB, Revision as RevisionPB},
-    sync::{RevisionSynchronizer, RevisionUser, SyncResponse},
+    server_document::{document_pad::ServerDocument, RevisionSynchronizer, RevisionUser, SyncResponse},
 };
 use async_stream::stream;
 use dashmap::DashMap;
@@ -186,7 +185,7 @@ impl OpenDocHandle {
         let synchronizer = Arc::new(RevisionSynchronizer::new(
             &doc.doc_id,
             doc.rev_id,
-            Document::from_delta(delta),
+            ServerDocument::from_delta(delta),
             persistence,
         ));
 

shared-lib/flowy-collaboration/src/server_document/document_pad.rs

@@ -0,0 +1,39 @@
+use crate::{client_document::InitialDocumentText, errors::CollaborateError};
+use lib_ot::{core::*, rich_text::RichTextDelta};
+
+pub struct ServerDocument {
+    delta: RichTextDelta,
+}
+
+impl ServerDocument {
+    pub fn new<C: InitialDocumentText>() -> Self { Self::from_delta(C::initial_delta()) }
+
+    pub fn from_delta(delta: RichTextDelta) -> Self { ServerDocument { delta } }
+
+    pub fn from_json(json: &str) -> Result<Self, CollaborateError> {
+        let delta = RichTextDelta::from_json(json)?;
+        Ok(Self::from_delta(delta))
+    }
+
+    pub fn to_json(&self) -> String { self.delta.to_json() }
+
+    pub fn to_bytes(&self) -> Vec<u8> { self.delta.clone().to_bytes().to_vec() }
+
+    pub fn to_plain_string(&self) -> String { self.delta.apply("").unwrap() }
+
+    pub fn delta(&self) -> &RichTextDelta { &self.delta }
+
+    pub fn md5(&self) -> String {
+        let bytes = self.to_bytes();
+        format!("{:x}", md5::compute(bytes))
+    }
+
+    pub fn compose_delta(&mut self, delta: RichTextDelta) -> Result<(), CollaborateError> {
+        // tracing::trace!("{} compose {}", &self.delta.to_json(), delta.to_json());
+        let composed_delta = self.delta.compose(&delta)?;
+        self.delta = composed_delta;
+        Ok(())
+    }
+
+    pub fn is_empty<C: InitialDocumentText>(&self) -> bool { self.delta == C::initial_delta() }
+}

shared-lib/flowy-collaboration/src/server_document/mod.rs

@@ -0,0 +1,6 @@
+mod document_manager;
+mod document_pad;
+mod revision_sync;
+
+pub use document_manager::*;
+pub use revision_sync::*;

shared-lib/flowy-collaboration/src/server_document/revision_sync.rs

@@ -1,12 +1,11 @@
 use crate::{
-    document::Document,
     entities::{
         revision::RevisionRange,
         ws::{DocumentServerWSData, DocumentServerWSDataBuilder},
     },
     errors::CollaborateError,
     protobuf::{RepeatedRevision as RepeatedRevisionPB, Revision as RevisionPB},
-    sync::DocumentCloudPersistence,
+    server_document::{document_pad::ServerDocument, DocumentCloudPersistence},
     util::*,
 };
 use lib_ot::{core::OperationTransformable, rich_text::RichTextDelta};
@@ -35,7 +34,7 @@ pub enum SyncResponse {
 pub struct RevisionSynchronizer {
     pub doc_id: String,
     pub rev_id: AtomicI64,
-    document: Arc<RwLock<Document>>,
+    document: Arc<RwLock<ServerDocument>>,
     persistence: Arc<dyn DocumentCloudPersistence>,
 }
 
@@ -43,7 +42,7 @@ impl RevisionSynchronizer {
     pub fn new(
         doc_id: &str,
         rev_id: i64,
-        document: Document,
+        document: ServerDocument,
         persistence: Arc<dyn DocumentCloudPersistence>,
     ) -> RevisionSynchronizer {
         let document = Arc::new(RwLock::new(document));
@@ -100,7 +99,7 @@ impl RevisionSynchronizer {
             },
             Ordering::Equal => {
                 // Do nothing
-                log::warn!("Applied revision rev_id is the same as cur_rev_id");
+                tracing::warn!("Applied revision rev_id is the same as cur_rev_id");
             },
             Ordering::Greater => {
                 // The client document is outdated. Transform the client revision delta and then
@@ -147,7 +146,7 @@ impl RevisionSynchronizer {
         let delta = make_delta_from_revision_pb(revisions)?;
 
         let _ = self.persistence.reset_document(&doc_id, repeated_revision).await?;
-        *self.document.write() = Document::from_delta(delta);
+        *self.document.write() = ServerDocument::from_delta(delta);
         let _ = self.rev_id.fetch_update(SeqCst, SeqCst, |_e| Some(rev_id));
         Ok(())
     }

shared-lib/flowy-collaboration/src/sync/mod.rs

@@ -1,5 +0,0 @@
-mod server;
-mod synchronizer;
-
-pub use server::*;
-pub use synchronizer::*;

shared-lib/flowy-core-data-model/src/entities/view/view_create.rs

@@ -7,7 +7,7 @@ use crate::{
         view::{ViewName, ViewThumbnail},
     },
 };
-use flowy_collaboration::document::default::initial_delta_string;
+use flowy_collaboration::client_document::default::initial_delta_string;
 use flowy_derive::{ProtoBuf, ProtoBuf_Enum};
 use std::convert::TryInto;
 

shared-lib/lib-ot/src/core/operation/operation.rs

@@ -205,12 +205,12 @@ where
     T: Attributes,
 {
     pub fn merge_or_new(&mut self, n: usize, attributes: T) -> Option<Operation<T>> {
-        tracing::trace!(
+        // tracing::trace!(
-            "merge_retain_or_new_op: len: {:?}, l: {} - r: {}",
+        //     "merge_retain_or_new_op: len: {:?}, l: {} - r: {}",
-            n,
+        //     n,
-            self.attributes,
+        //     self.attributes,
-            attributes
+        //     attributes
-        );
+        // );
         if self.attributes == attributes {
             self.n += n;
             None