stop sync after close the view

backend/Cargo.lock

@@ -1350,9 +1350,11 @@ dependencies = [
  "futures-util",
  "lib-dispatch",
  "lib-infra",
+ "lib-ot",
  "log",
  "protobuf",
  "serde",
+ "serde_json",
  "thread-id",
  "tokio",
 ]

backend/src/services/doc/crud.rs

@@ -59,7 +59,7 @@ pub async fn update_doc(pool: &PgPool, mut params: UpdateDocParams) -> Result<()
 
     let data = Some(params.take_data());
 
-    tracing::Span::current().record("result", &data.as_ref().unwrap_or(&"".to_owned()).as_str());
+    tracing::Span::current().record("delta", &data.as_ref().unwrap_or(&"".to_owned()).as_str());
 
     let (sql, args) = SqlBuilder::update(DOC_TABLE)
         .add_some_arg("data", data)

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

@@ -93,8 +93,10 @@ impl CoreContext {
                         data: delta.to_json(),
                     };
                     let _ = self.view_controller.apply_doc_delta(doc_delta).await?;
-
                     self.view_controller.set_latest_view(&view);
+
+                    // Close the view after initialize
+                    self.view_controller.close_view(view.id.clone().into()).await?;
                 }
                 let _ = self.view_controller.create_view(view).await?;
             }

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

@@ -42,6 +42,10 @@ impl DocCache {
 
     pub(crate) fn remove(&self, id: &str) {
         let doc_id: DocId = id.into();
+        match self.get(id) {
+            Ok(editor) => editor.stop_sync(),
+            Err(e) => log::error!("{}", e),
+        }
         self.inner.remove(&doc_id);
     }
 }

frontend/rust-lib/flowy-document/src/services/doc/controller.rs

@@ -56,6 +56,7 @@ impl DocController {
     }
 
     pub(crate) fn close(&self, doc_id: &str) -> Result<(), DocError> {
+        log::debug!("Close doc {}", doc_id);
         self.cache.remove(doc_id);
         self.ws_manager.remove_handler(doc_id);
         Ok(())

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

@@ -4,7 +4,7 @@ use crate::{
     services::{
         doc::{
             edit::{EditCommand, EditCommandQueue, OpenDocAction, TransformDeltas},
-            revision::{RevisionDownStream, RevisionManager},
+            revision::{RevisionDownStream, RevisionManager, SteamStopRx, SteamStopTx},
         },
         ws::{DocumentWebSocket, WsDocumentHandler},
     },
@@ -35,19 +35,7 @@ pub struct ClientDocEditor {
     ws_sender: Arc<dyn DocumentWebSocket>,
     user: Arc<dyn DocumentUser>,
     ws_msg_tx: UnboundedSender<WsDocumentData>,
-}
-
-#[cfg(feature = "flowy_unit_test")]
-impl ClientDocEditor {
-    pub async fn doc_json(&self) -> DocResult<String> {
-        let (ret, rx) = oneshot::channel::<DocumentResult<String>>();
-        let msg = EditCommand::ReadDoc { ret };
-        let _ = self.edit_tx.send(msg);
-        let s = rx.await.map_err(internal_error)??;
-        Ok(s)
-    }
-
-    pub fn rev_manager(&self) -> Arc<RevisionManager> { self.rev_manager.clone() }
+    stop_sync_tx: tokio::sync::broadcast::Sender<()>,
 }
 
 impl ClientDocEditor {
@@ -63,6 +51,8 @@ impl ClientDocEditor {
         let doc_id = doc_id.to_string();
         let rev_manager = Arc::new(rev_manager);
         let (ws_msg_tx, ws_msg_rx) = mpsc::unbounded_channel();
+        let (stop_sync_tx, _) = tokio::sync::broadcast::channel(2);
+        let cloned_stop_sync_tx = stop_sync_tx.clone();
         let edit_doc = Arc::new(Self {
             doc_id,
             rev_manager,
@@ -70,10 +60,11 @@ impl ClientDocEditor {
             user,
             ws_msg_tx,
             ws_sender,
+            stop_sync_tx,
         });
         edit_doc.notify_open_doc();
 
-        start_sync(edit_doc.clone(), ws_msg_rx);
+        start_sync(edit_doc.clone(), ws_msg_rx, cloned_stop_sync_tx);
         Ok(edit_doc)
     }
 
@@ -112,7 +103,7 @@ impl ClientDocEditor {
         Ok(())
     }
 
-    pub async fn replace<T: ToString>(&mut self, interval: Interval, data: T) -> Result<(), DocError> {
+    pub async fn replace<T: ToString>(&self, interval: Interval, data: T) -> Result<(), DocError> {
         let (ret, rx) = oneshot::channel::<DocumentResult<RichTextDelta>>();
         let msg = EditCommand::Replace {
             interval,
@@ -191,6 +182,12 @@ impl ClientDocEditor {
         Ok(())
     }
 
+    #[tracing::instrument(level = "debug", skip(self), fields(doc_id))]
+    pub fn stop_sync(&self) {
+        tracing::Span::current().record("doc_id", &self.doc_id.as_str());
+        let _ = self.stop_sync_tx.send(());
+    }
+
     #[tracing::instrument(level = "debug", skip(self))]
     fn notify_open_doc(&self) {
         let rev_id: RevId = self.rev_manager.rev_id().into();
@@ -303,13 +300,38 @@ fn spawn_edit_queue(doc_id: &str, delta: RichTextDelta, _pool: Arc<ConnectionPoo
     sender
 }
 
-fn start_sync(editor: Arc<ClientDocEditor>, ws_msg_rx: mpsc::UnboundedReceiver<WsDocumentData>) {
+fn start_sync(
+    editor: Arc<ClientDocEditor>,
+    ws_msg_rx: mpsc::UnboundedReceiver<WsDocumentData>,
+    stop_sync_tx: SteamStopTx,
+) {
     let rev_manager = editor.rev_manager.clone();
     let ws_sender = editor.ws_sender.clone();
 
-    let up_stream = editor.rev_manager.make_up_stream();
-    let down_stream = RevisionDownStream::new(editor, rev_manager, ws_msg_rx, ws_sender);
+    let up_stream = editor.rev_manager.make_up_stream(stop_sync_tx.subscribe());
+    let down_stream = RevisionDownStream::new(editor, rev_manager, ws_msg_rx, ws_sender, stop_sync_tx.subscribe());
 
     tokio::spawn(up_stream.run());
     tokio::spawn(down_stream.run());
 }
+
+#[cfg(feature = "flowy_unit_test")]
+impl ClientDocEditor {
+    pub async fn doc_json(&self) -> DocResult<String> {
+        let (ret, rx) = oneshot::channel::<DocumentResult<String>>();
+        let msg = EditCommand::ReadDoc { ret };
+        let _ = self.edit_tx.send(msg);
+        let s = rx.await.map_err(internal_error)??;
+        Ok(s)
+    }
+
+    pub async fn doc_delta(&self) -> DocResult<RichTextDelta> {
+        let (ret, rx) = oneshot::channel::<DocumentResult<RichTextDelta>>();
+        let msg = EditCommand::ReadDocDelta { ret };
+        let _ = self.edit_tx.send(msg);
+        let delta = rx.await.map_err(internal_error)??;
+        Ok(delta)
+    }
+
+    pub fn rev_manager(&self) -> Arc<RevisionManager> { self.rev_manager.clone() }
+}

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

@@ -105,6 +105,10 @@ impl EditCommandQueue {
                 let data = self.document.read().await.to_json();
                 let _ = ret.send(Ok(data));
             },
+            EditCommand::ReadDocDelta { ret } => {
+                let delta = self.document.read().await.delta().clone();
+                let _ = ret.send(Ok(delta));
+            },
         }
         Ok(())
     }
@@ -170,6 +174,9 @@ pub(crate) enum EditCommand {
     ReadDoc {
         ret: Ret<String>,
     },
+    ReadDocDelta {
+        ret: Ret<RichTextDelta>,
+    },
 }
 
 pub(crate) struct TransformDeltas {

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

@@ -1,7 +1,7 @@
 use crate::{
     errors::{DocError, DocResult},
     services::{
-        doc::revision::{RevisionCache, RevisionUpStream},
+        doc::revision::{RevisionCache, RevisionUpStream, SteamStopRx},
         ws::DocumentWebSocket,
     },
 };
@@ -88,8 +88,8 @@ impl RevisionManager {
         Ok(revision)
     }
 
-    pub(crate) fn make_up_stream(&self) -> RevisionUpStream {
-        RevisionUpStream::new(self.cache.clone(), self.ws_sender.clone())
+    pub(crate) fn make_up_stream(&self, stop_rx: SteamStopRx) -> RevisionUpStream {
+        RevisionUpStream::new(self.cache.clone(), self.ws_sender.clone(), stop_rx)
     }
 }
 

frontend/rust-lib/flowy-document/src/services/doc/revision/sync.rs

@@ -15,7 +15,7 @@ use futures::stream::StreamExt;
 use lib_ot::revision::{RevId, RevisionRange};
 use std::{convert::TryFrom, sync::Arc};
 use tokio::{
-    sync::mpsc,
+    sync::{broadcast, mpsc},
     task::spawn_blocking,
     time::{interval, Duration},
 };
@@ -25,6 +25,7 @@ pub(crate) struct RevisionDownStream {
     rev_manager: Arc<RevisionManager>,
     receiver: Option<mpsc::UnboundedReceiver<WsDocumentData>>,
     ws_sender: Arc<dyn DocumentWebSocket>,
+    stop_rx: Option<SteamStopRx>,
 }
 
 impl RevisionDownStream {
@@ -33,23 +34,35 @@ impl RevisionDownStream {
         rev_manager: Arc<RevisionManager>,
         receiver: mpsc::UnboundedReceiver<WsDocumentData>,
         ws_sender: Arc<dyn DocumentWebSocket>,
+        stop_rx: SteamStopRx,
     ) -> Self {
         RevisionDownStream {
             editor,
             rev_manager,
             receiver: Some(receiver),
             ws_sender,
+            stop_rx: Some(stop_rx),
         }
     }
 
     pub async fn run(mut self) {
         let mut receiver = self.receiver.take().expect("Only take once");
+        let mut stop_rx = self.stop_rx.take().expect("Only take once");
         let stream = stream! {
             loop {
-                match receiver.recv().await {
-                    Some(msg) => yield msg,
-                    None => break,
-                }
+                // match receiver.recv().await {
+                //     Some(msg) => yield msg,
+                //     None => break,
+                // }
+                tokio::select! {
+                    result = receiver.recv() => {
+                        match result {
+                            Some(msg) => yield msg,
+                            None => break,
+                        }
+                    },
+                    _ = stop_rx.recv() => break,
+                };
             }
         };
         stream
@@ -95,25 +108,43 @@ pub(crate) enum UpStreamMsg {
     Tick,
 }
 
+pub type SteamStopRx = broadcast::Receiver<()>;
+pub type SteamStopTx = broadcast::Sender<()>;
+
 pub(crate) struct RevisionUpStream {
     revisions: Arc<dyn RevisionIterator>,
     ws_sender: Arc<dyn DocumentWebSocket>,
+    stop_rx: Option<SteamStopRx>,
 }
 
 impl RevisionUpStream {
-    pub(crate) fn new(revisions: Arc<dyn RevisionIterator>, ws_sender: Arc<dyn DocumentWebSocket>) -> Self {
-        Self { revisions, ws_sender }
+    pub(crate) fn new(
+        revisions: Arc<dyn RevisionIterator>,
+        ws_sender: Arc<dyn DocumentWebSocket>,
+        stop_rx: SteamStopRx,
+    ) -> Self {
+        Self {
+            revisions,
+            ws_sender,
+            stop_rx: Some(stop_rx),
+        }
     }
 
-    pub async fn run(self) {
+    pub async fn run(mut self) {
         let (tx, mut rx) = mpsc::unbounded_channel();
+        let mut stop_rx = self.stop_rx.take().expect("Only take once");
         tokio::spawn(tick(tx));
         let stream = stream! {
             loop {
-                match rx.recv().await {
-                    Some(msg) => yield msg,
-                    None => break,
-                }
+                tokio::select! {
+                    result = rx.recv() => {
+                        match result {
+                            Some(msg) => yield msg,
+                            None => break,
+                        }
+                    },
+                    _ = stop_rx.recv() => break,
+                };
             }
         };
         stream

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

@@ -19,7 +19,7 @@ fn attributes_bold_added() {
             ]"#,
         ),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -31,7 +31,7 @@ fn attributes_bold_added_and_invert_all() {
         Bold(0, Interval::new(0, 3), false),
         AssertDocJson(0, r#"[{"insert":"123"}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -43,7 +43,7 @@ fn attributes_bold_added_and_invert_partial_suffix() {
         Bold(0, Interval::new(2, 4), false),
         AssertDocJson(0, r#"[{"insert":"12","attributes":{"bold":"true"}},{"insert":"34"}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -57,7 +57,7 @@ fn attributes_bold_added_and_invert_partial_suffix2() {
         Bold(0, Interval::new(2, 4), true),
         AssertDocJson(0, r#"[{"insert":"1234","attributes":{"bold":"true"}}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -85,7 +85,7 @@ fn attributes_bold_added_with_new_line() {
             r#"[{"insert":"123","attributes":{"bold":"true"}},{"insert":"\na\n"},{"insert":"456","attributes":{"bold":"true"}},{"insert":"\n"}]"#,
         ),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -97,7 +97,7 @@ fn attributes_bold_added_and_invert_partial_prefix() {
         Bold(0, Interval::new(0, 2), false),
         AssertDocJson(0, r#"[{"insert":"12"},{"insert":"34","attributes":{"bold":"true"}}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -109,7 +109,7 @@ fn attributes_bold_added_consecutive() {
         Bold(0, Interval::new(1, 2), true),
         AssertDocJson(0, r#"[{"insert":"12","attributes":{"bold":"true"}},{"insert":"34"}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -128,7 +128,7 @@ fn attributes_bold_added_italic() {
             r#"[{"insert":"12345678","attributes":{"bold":"true","italic":"true"}},{"insert":"\n"}]"#,
         ),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -156,7 +156,7 @@ fn attributes_bold_added_italic2() {
         ),
     ];
 
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -193,7 +193,7 @@ fn attributes_bold_added_italic3() {
         ),
     ];
 
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -229,7 +229,7 @@ fn attributes_bold_added_italic_delete() {
         AssertDocJson(0, r#"[{"insert":"67"},{"insert":"89","attributes":{"bold":"true"}}]"#),
     ];
 
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -240,7 +240,7 @@ fn attributes_merge_inserted_text_with_same_attribute() {
         InsertBold(0, "456", Interval::new(3, 6)),
         AssertDocJson(0, r#"[{"insert":"123456","attributes":{"bold":"true"}}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -255,7 +255,7 @@ fn attributes_compose_attr_attributes_with_attr_attributes_test() {
         AssertDocJson(1, r#"[{"insert":"1234567","attributes":{"bold":"true"}}]"#),
     ];
 
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -296,7 +296,7 @@ fn attributes_compose_attr_attributes_with_attr_attributes_test2() {
         ),
     ];
 
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -312,7 +312,7 @@ fn attributes_compose_attr_attributes_with_no_attr_attributes_test() {
         AssertDocJson(0, expected),
         AssertDocJson(1, expected),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -324,7 +324,7 @@ fn attributes_replace_heading() {
         AssertDocJson(0, r#"[{"insert":"3456","attributes":{"bold":"true"}}]"#),
     ];
 
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -336,7 +336,7 @@ fn attributes_replace_trailing() {
         AssertDocJson(0, r#"[{"insert":"12345","attributes":{"bold":"true"}}]"#),
     ];
 
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -350,7 +350,7 @@ fn attributes_replace_middle() {
         AssertDocJson(0, r#"[{"insert":"34","attributes":{"bold":"true"}}]"#),
     ];
 
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -362,7 +362,7 @@ fn attributes_replace_all() {
         AssertDocJson(0, r#"[]"#),
     ];
 
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -374,7 +374,7 @@ fn attributes_replace_with_text() {
         AssertDocJson(0, r#"[{"insert":"ab"},{"insert":"456","attributes":{"bold":"true"}}]"#),
     ];
 
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -390,7 +390,7 @@ fn attributes_header_insert_newline_at_middle() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -415,7 +415,7 @@ fn attributes_header_insert_double_newline_at_middle() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -430,7 +430,7 @@ fn attributes_header_insert_newline_at_trailing() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -446,7 +446,7 @@ fn attributes_header_insert_double_newline_at_trailing() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -460,7 +460,7 @@ fn attributes_link_added() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -479,7 +479,7 @@ fn attributes_link_format_with_bold() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -498,7 +498,7 @@ fn attributes_link_insert_char_at_head() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -513,7 +513,7 @@ fn attributes_link_insert_char_at_middle() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -532,7 +532,7 @@ fn attributes_link_insert_char_at_trailing() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -547,7 +547,7 @@ fn attributes_link_insert_newline_at_middle() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -563,7 +563,7 @@ fn attributes_link_auto_format() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -579,7 +579,7 @@ fn attributes_link_auto_format_exist() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -595,7 +595,7 @@ fn attributes_link_auto_format_exist2() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -606,7 +606,7 @@ fn attributes_bullet_added() {
         AssertDocJson(0, r#"[{"insert":"12"},{"insert":"\n","attributes":{"list":"bullet"}}]"#),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -627,7 +627,7 @@ fn attributes_bullet_added_2() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -644,7 +644,7 @@ fn attributes_bullet_remove_partial() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -660,7 +660,7 @@ fn attributes_bullet_auto_exit() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -700,7 +700,7 @@ fn attributes_preserve_block_when_insert_newline_inside() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -717,7 +717,7 @@ fn attributes_preserve_header_format_on_merge() {
         AssertDocJson(0, r#"[{"insert":"123456"},{"insert":"\n","attributes":{"header":1}}]"#),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -736,7 +736,7 @@ fn attributes_format_emoji() {
             r#"[{"insert":"👋 "},{"insert":"\n","attributes":{"header":1}}]"#,
         ),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -756,7 +756,7 @@ fn attributes_preserve_list_format_on_merge() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -795,5 +795,5 @@ fn delta_compose() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }

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

@@ -267,11 +267,11 @@ impl TestBuilder {
         }
     }
 
-    pub fn run_script<C: CustomDocument>(mut self, script: Vec<TestOp>) {
+    pub fn run_scripts<C: CustomDocument>(mut self, scripts: Vec<TestOp>) {
         self.documents = vec![Document::new::<C>(), Document::new::<C>()];
         self.primes = vec![None, None];
         self.deltas = vec![None, None];
-        for (_i, op) in script.iter().enumerate() {
+        for (_i, op) in scripts.iter().enumerate() {
             self.run_op(op);
         }
     }

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

@@ -13,7 +13,7 @@ fn attributes_insert_text() {
         Insert(0, "456", 3),
         AssertDocJson(0, r#"[{"insert":"123456"}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -23,7 +23,7 @@ fn attributes_insert_text_at_head() {
         Insert(0, "456", 0),
         AssertDocJson(0, r#"[{"insert":"456123"}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -33,7 +33,7 @@ fn attributes_insert_text_at_middle() {
         Insert(0, "456", 1),
         AssertDocJson(0, r#"[{"insert":"145623"}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -523,7 +523,7 @@ fn transform_two_plain_delta_test() {
         AssertDocJson(0, r#"[{"insert":"123456"}]"#),
         AssertDocJson(1, r#"[{"insert":"123456"}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -537,7 +537,7 @@ fn transform_two_plain_delta_test2() {
         AssertDocJson(0, r#"[{"insert":"123456"}]"#),
         AssertDocJson(1, r#"[{"insert":"123456"}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -555,7 +555,7 @@ fn transform_two_non_seq_delta() {
         AssertDocJson(0, r#"[{"insert":"123456"}]"#),
         AssertDocJson(1, r#"[{"insert":"123456789"}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -570,7 +570,7 @@ fn transform_two_conflict_non_seq_delta() {
         AssertDocJson(0, r#"[{"insert":"123456"}]"#),
         AssertDocJson(1, r#"[{"insert":"12378456"}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -597,7 +597,7 @@ fn delta_invert_no_attribute_delta2() {
         Invert(0, 1),
         AssertDocJson(0, r#"[{"insert":"123"}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -610,7 +610,7 @@ fn delta_invert_attribute_delta_with_no_attribute_delta() {
         Invert(0, 1),
         AssertDocJson(0, r#"[{"insert":"123","attributes":{"bold":"true"}}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -645,7 +645,7 @@ fn delta_invert_attribute_delta_with_no_attribute_delta2() {
             ]"#,
         ),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -658,7 +658,7 @@ fn delta_invert_no_attribute_delta_with_attribute_delta() {
         Invert(0, 1),
         AssertDocJson(0, r#"[{"insert":"123"}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -677,7 +677,7 @@ fn delta_invert_no_attribute_delta_with_attribute_delta2() {
         Invert(0, 1),
         AssertDocJson(0, r#"[{"insert":"123"}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -718,7 +718,7 @@ fn delta_invert_attribute_delta_with_attribute_delta() {
             ]"#,
         ),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -731,5 +731,5 @@ fn delta_compose_with_missing_delta() {
         AssertDocJson(0, r#"[{"insert":"1234\n"}]"#),
         AssertStr(1, r#"4\n"#),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }

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

@@ -1,18 +1,13 @@
-use flowy_test::editor::*;
+use flowy_test::editor::{EditorScript::*, *};
 
 #[tokio::test]
 async fn create_doc() {
-    let test = EditorTest::new().await;
-    let editor = test.create_doc().await;
-    let rev_manager = editor.rev_manager();
-    assert_eq!(rev_manager.rev_id(), 0);
-
-    let json = editor.doc_json().await.unwrap();
-    assert_eq!(json, r#"[{"insert":"\n"}]"#);
-
-    editor.insert(0, "123").await.unwrap();
-    assert_eq!(rev_manager.rev_id(), 1);
-
-    editor.insert(0, "456").await.unwrap();
-    assert_eq!(rev_manager.rev_id(), 2);
+    let scripts = vec![
+        InsertText("123", 0),
+        AssertRevId(1),
+        InsertText("456", 3),
+        AssertRevId(2),
+        AssertJson(r#"[{"insert":"123456\n"}]"#),
+    ];
+    EditorTest::new().await.run_scripts(scripts).await;
 }

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

@@ -5,7 +5,7 @@ use lib_ot::core::{Interval, NEW_LINE, WHITESPACE};
 #[test]
 fn history_insert_undo() {
     let ops = vec![Insert(0, "123", 0), Undo(0), AssertDocJson(0, r#"[{"insert":"\n"}]"#)];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -19,7 +19,7 @@ fn history_insert_undo_with_lagging() {
         Undo(0),
         AssertDocJson(0, r#"[{"insert":"\n"}]"#),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -32,7 +32,7 @@ fn history_insert_redo() {
         Redo(0),
         AssertDocJson(0, r#"[{"insert":"123\n"}]"#),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -51,7 +51,7 @@ fn history_insert_redo_with_lagging() {
         Undo(0),
         AssertDocJson(0, r#"[{"insert":"123\n"}]"#),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -62,7 +62,7 @@ fn history_bold_undo() {
         Undo(0),
         AssertDocJson(0, r#"[{"insert":"\n"}]"#),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -74,7 +74,7 @@ fn history_bold_undo_with_lagging() {
         Undo(0),
         AssertDocJson(0, r#"[{"insert":"123\n"}]"#),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -87,7 +87,7 @@ fn history_bold_redo() {
         Redo(0),
         AssertDocJson(0, r#" [{"insert":"123","attributes":{"bold":"true"}},{"insert":"\n"}]"#),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -101,7 +101,7 @@ fn history_bold_redo_with_lagging() {
         Redo(0),
         AssertDocJson(0, r#"[{"insert":"123","attributes":{"bold":"true"}},{"insert":"\n"}]"#),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -114,7 +114,7 @@ fn history_delete_undo() {
         Undo(0),
         AssertDocJson(0, r#"[{"insert":"123"}]"#),
     ];
-    TestBuilder::new().run_script::<PlainDoc>(ops);
+    TestBuilder::new().run_scripts::<PlainDoc>(ops);
 }
 
 #[test]
@@ -133,7 +133,7 @@ fn history_delete_undo_2() {
         Undo(0),
         AssertDocJson(0, r#"[{"insert":"\n"}]"#),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -160,7 +160,7 @@ fn history_delete_undo_with_lagging() {
             "#,
         ),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -174,7 +174,7 @@ fn history_delete_redo() {
         Redo(0),
         AssertDocJson(0, r#"[{"insert":"\n"}]"#),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -193,7 +193,7 @@ fn history_replace_undo() {
         Undo(0),
         AssertDocJson(0, r#"[{"insert":"\n"}]"#),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -214,7 +214,7 @@ fn history_replace_undo_with_lagging() {
         Undo(0),
         AssertDocJson(0, r#"[{"insert":"123","attributes":{"bold":"true"}},{"insert":"\n"}]"#),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -233,7 +233,7 @@ fn history_replace_redo() {
             "#,
         ),
     ];
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -252,7 +252,7 @@ fn history_header_added_undo() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -271,7 +271,7 @@ fn history_link_added_undo() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -290,7 +290,7 @@ fn history_link_auto_format_undo_with_lagging() {
         AssertDocJson(0, r#"[{"insert":"https://appflowy.io\n"}]"#),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -313,7 +313,7 @@ fn history_bullet_undo() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -341,7 +341,7 @@ fn history_bullet_undo_with_lagging() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }
 
 #[test]
@@ -368,5 +368,5 @@ fn history_undo_attribute_on_merge_between_line() {
         ),
     ];
 
-    TestBuilder::new().run_script::<FlowyDoc>(ops);
+    TestBuilder::new().run_scripts::<FlowyDoc>(ops);
 }

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

@@ -15,9 +15,10 @@ lib-infra = { path = "../lib-infra" }
 
 flowy-document-infra = { path = "../../../shared-lib/flowy-document-infra" }
 backend-service = { path = "../../../shared-lib/backend-service" }
-
+lib-ot = { path = "../../../shared-lib/lib-ot" }
 
 serde = { version = "1.0", features = ["derive"] }
+serde_json = {version = "1.0"}
 bincode = { version = "1.3"}
 protobuf = {version = "2.24.1"}
 claim = "0.5.0"

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

@@ -1,32 +1,82 @@
 use crate::{helper::ViewTest, FlowySDKTest};
 use flowy_document::services::doc::edit::ClientDocEditor;
 use flowy_document_infra::entities::doc::DocIdentifier;
+use lib_ot::{core::Interval, revision::RevState, rich_text::RichTextDelta};
 use std::sync::Arc;
-use tokio::time::Interval;
+use tokio::time::{sleep, Duration};
 
 pub struct EditorTest {
     pub sdk: FlowySDKTest,
+    pub editor: Arc<ClientDocEditor>,
 }
 
 impl EditorTest {
     pub async fn new() -> Self {
         let sdk = FlowySDKTest::setup();
         let _ = sdk.init_user().await;
-        Self { sdk }
+        let test = ViewTest::new(&sdk).await;
+        let doc_identifier: DocIdentifier = test.view.id.clone().into();
+        let editor = sdk.flowy_document.open(doc_identifier).await.unwrap();
+        Self { sdk, editor }
     }
 
-    pub async fn create_doc(&self) -> Arc<ClientDocEditor> {
-        let test = ViewTest::new(&self.sdk).await;
-        let doc_identifier: DocIdentifier = test.view.id.clone().into();
-        self.sdk.flowy_document.open(doc_identifier).await.unwrap()
+    pub async fn run_scripts(mut self, scripts: Vec<EditorScript>) {
+        for script in scripts {
+            self.run_script(script).await;
+        }
+
+        sleep(Duration::from_secs(10)).await;
+    }
+
+    async fn run_script(&mut self, script: EditorScript) {
+        let rev_manager = self.editor.rev_manager();
+        let cache = rev_manager.revision_cache();
+        let memory_cache = cache.memory_cache();
+        let disk_cache = cache.dish_cache();
+
+        match script {
+            EditorScript::InsertText(s, offset) => {
+                self.editor.insert(offset, s).await.unwrap();
+            },
+            EditorScript::Delete(interval) => {
+                self.editor.delete(interval).await.unwrap();
+            },
+            EditorScript::Replace(interval, s) => {
+                self.editor.replace(interval, s).await.unwrap();
+            },
+            EditorScript::Undo() => {
+                self.editor.undo().await.unwrap();
+            },
+            EditorScript::Redo() => {
+                self.editor.redo().await.unwrap();
+            },
+            EditorScript::AssertRevisionState(rev_id, state) => {},
+            EditorScript::AssertNextSentRevision(rev_id, state) => {},
+            EditorScript::AssertRevId(rev_id) => {
+                assert_eq!(self.editor.rev_manager().rev_id(), rev_id);
+            },
+            EditorScript::AssertJson(expected) => {
+                let expected_delta: RichTextDelta = serde_json::from_str(expected).unwrap();
+                let delta = self.editor.doc_delta().await.unwrap();
+
+                if expected_delta != delta {
+                    eprintln!("✅ expect: {}", expected,);
+                    eprintln!("❌ receive: {}", delta.to_json());
+                }
+                assert_eq!(expected_delta, delta);
+            },
+        }
     }
 }
 
-pub enum EditAction {
+pub enum EditorScript {
     InsertText(&'static str, usize),
     Delete(Interval),
     Replace(Interval, &'static str),
     Undo(),
     Redo(),
+    AssertRevisionState(i64, RevState),
+    AssertNextSentRevision(i64, RevState),
+    AssertRevId(i64),
     AssertJson(&'static str),
 }