chore: generic operation in conflict_resolver

frontend/app_flowy/lib/plugins/doc/application/doc_service.dart

@@ -3,7 +3,7 @@ import 'package:flowy_sdk/dispatch/dispatch.dart';
 
 import 'package:flowy_sdk/protobuf/flowy-folder/view.pb.dart';
 import 'package:flowy_sdk/protobuf/flowy-error/errors.pb.dart';
-import 'package:flowy_sdk/protobuf/flowy-sync/text_block.pb.dart';
+import 'package:flowy_sdk/protobuf/flowy-sync/document.pb.dart';
 import 'package:flowy_sdk/protobuf/flowy-document/entities.pb.dart';
 
 class DocumentService {

frontend/rust-lib/Cargo.lock

@@ -1077,7 +1077,6 @@ dependencies = [
  "flowy-sync",
  "futures-util",
  "lib-infra",
- "lib-ot",
  "lib-ws",
  "serde",
  "serde_json",

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

@@ -1,9 +1,8 @@
-use crate::web_socket::EditorCommandReceiver;
+use crate::web_socket::{DocumentResolveOperations, EditorCommandReceiver};
 use crate::DocumentUser;
 use async_stream::stream;
 use flowy_error::FlowyError;
-use flowy_revision::{OperationsMD5, RevisionManager, TextTransformOperations, TransformOperations};
-
+use flowy_revision::{OperationsMD5, RevisionManager, TransformOperations};
 use flowy_sync::{
     client_document::{history::UndoResult, ClientDocument},
     entities::revision::{RevId, Revision},
@@ -91,21 +90,21 @@ impl EditDocumentQueue {
             EditorCommand::TransformOperations { operations, ret } => {
                 let f = || async {
                     let read_guard = self.document.read().await;
-                    let mut server_prime: Option<TextOperations> = None;
-                    let client_prime: TextOperations;
+                    let mut server_operations: Option<DocumentResolveOperations> = None;
+                    let client_operations: TextOperations;
 
                     if read_guard.is_empty() {
                         // Do nothing
-                        client_prime = operations;
+                        client_operations = operations;
                     } else {
                         let (s_prime, c_prime) = read_guard.get_operations().transform(&operations)?;
-                        client_prime = c_prime;
-                        server_prime = Some(s_prime);
+                        client_operations = c_prime;
+                        server_operations = Some(DocumentResolveOperations(s_prime));
                     }
                     drop(read_guard);
                     Ok::<TextTransformOperations, CollaborateError>(TransformOperations {
-                        client_prime,
-                        server_prime,
+                        client_operations: DocumentResolveOperations(client_operations),
+                        server_operations,
                     })
                 };
                 let _ = ret.send(f().await);
@@ -184,6 +183,8 @@ impl EditDocumentQueue {
     }
 }
 
+pub type TextTransformOperations = TransformOperations<DocumentResolveOperations>;
+
 pub(crate) type Ret<T> = oneshot::Sender<Result<T, CollaborateError>>;
 
 pub(crate) enum EditorCommand {

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

@@ -1,7 +1,9 @@
+use crate::queue::TextTransformOperations;
 use crate::{queue::EditorCommand, TEXT_BLOCK_SYNC_INTERVAL_IN_MILLIS};
 use bytes::Bytes;
-use flowy_error::{internal_error, FlowyError};
+use flowy_error::{internal_error, FlowyError, FlowyResult};
 use flowy_revision::*;
+use flowy_sync::entities::revision::Revision;
 use flowy_sync::{
     entities::{
         revision::RevisionRange,
@@ -10,7 +12,8 @@ use flowy_sync::{
     errors::CollaborateResult,
 };
 use lib_infra::future::{BoxResultFuture, FutureResult};
-use lib_ot::core::AttributeHashMap;
+
+use flowy_sync::util::make_operations_from_revisions;
 use lib_ot::text_delta::TextOperations;
 use lib_ws::WSConnectState;
 use std::{sync::Arc, time::Duration};
@@ -23,6 +26,29 @@ use tokio::sync::{
 pub(crate) type EditorCommandSender = Sender<EditorCommand>;
 pub(crate) type EditorCommandReceiver = Receiver<EditorCommand>;
 
+#[derive(Clone)]
+pub struct DocumentResolveOperations(pub TextOperations);
+
+impl OperationsDeserializer<DocumentResolveOperations> for DocumentResolveOperations {
+    fn deserialize_revisions(revisions: Vec<Revision>) -> FlowyResult<DocumentResolveOperations> {
+        Ok(DocumentResolveOperations(make_operations_from_revisions(revisions)?))
+    }
+}
+
+impl OperationsSerializer for DocumentResolveOperations {
+    fn serialize_operations(&self) -> Bytes {
+        self.0.json_bytes()
+    }
+}
+
+impl DocumentResolveOperations {
+    pub fn into_inner(self) -> TextOperations {
+        self.0
+    }
+}
+
+pub type DocumentConflictController = ConflictController<DocumentResolveOperations>;
+
 #[allow(dead_code)]
 pub(crate) async fn make_document_ws_manager(
     doc_id: String,
@@ -34,7 +60,7 @@ pub(crate) async fn make_document_ws_manager(
     let ws_data_provider = Arc::new(WSDataProvider::new(&doc_id, Arc::new(rev_manager.clone())));
     let resolver = Arc::new(DocumentConflictResolver { edit_cmd_tx });
     let conflict_controller =
-        RichTextConflictController::new(&user_id, resolver, Arc::new(ws_data_provider.clone()), rev_manager);
+        DocumentConflictController::new(&user_id, resolver, Arc::new(ws_data_provider.clone()), rev_manager);
     let ws_data_stream = Arc::new(DocumentRevisionWSDataStream::new(conflict_controller));
     let ws_data_sink = Arc::new(DocumentWSDataSink(ws_data_provider));
     let ping_duration = Duration::from_millis(TEXT_BLOCK_SYNC_INTERVAL_IN_MILLIS);
@@ -65,12 +91,12 @@ fn listen_document_ws_state(_user_id: &str, _doc_id: &str, mut subscriber: broad
 }
 
 pub(crate) struct DocumentRevisionWSDataStream {
-    conflict_controller: Arc<RichTextConflictController>,
+    conflict_controller: Arc<DocumentConflictController>,
 }
 
 impl DocumentRevisionWSDataStream {
     #[allow(dead_code)]
-    pub fn new(conflict_controller: RichTextConflictController) -> Self {
+    pub fn new(conflict_controller: DocumentConflictController) -> Self {
         Self {
             conflict_controller: Arc::new(conflict_controller),
         }
@@ -111,9 +137,10 @@ struct DocumentConflictResolver {
     edit_cmd_tx: EditorCommandSender,
 }
 
-impl ConflictResolver<AttributeHashMap> for DocumentConflictResolver {
-    fn compose_operations(&self, operations: TextOperations) -> BoxResultFuture<OperationsMD5, FlowyError> {
+impl ConflictResolver<DocumentResolveOperations> for DocumentConflictResolver {
+    fn compose_operations(&self, operations: DocumentResolveOperations) -> BoxResultFuture<OperationsMD5, FlowyError> {
         let tx = self.edit_cmd_tx.clone();
+        let operations = operations.into_inner();
         Box::pin(async move {
             let (ret, rx) = oneshot::channel();
             tx.send(EditorCommand::ComposeRemoteOperation {
@@ -131,9 +158,10 @@ impl ConflictResolver<AttributeHashMap> for DocumentConflictResolver {
 
     fn transform_operations(
         &self,
-        operations: TextOperations,
-    ) -> BoxResultFuture<flowy_revision::TextTransformOperations, FlowyError> {
+        operations: DocumentResolveOperations,
+    ) -> BoxResultFuture<TransformOperations<DocumentResolveOperations>, FlowyError> {
         let tx = self.edit_cmd_tx.clone();
+        let operations = operations.into_inner();
         Box::pin(async move {
             let (ret, rx) = oneshot::channel::<CollaborateResult<TextTransformOperations>>();
             tx.send(EditorCommand::TransformOperations { operations, ret })
@@ -146,8 +174,9 @@ impl ConflictResolver<AttributeHashMap> for DocumentConflictResolver {
         })
     }
 
-    fn reset_operations(&self, operations: TextOperations) -> BoxResultFuture<OperationsMD5, FlowyError> {
+    fn reset_operations(&self, operations: DocumentResolveOperations) -> BoxResultFuture<OperationsMD5, FlowyError> {
         let tx = self.edit_cmd_tx.clone();
+        let operations = operations.into_inner();
         Box::pin(async move {
             let (ret, rx) = oneshot::channel();
             let _ = tx

frontend/rust-lib/flowy-folder/src/services/persistence/migration.rs

@@ -3,6 +3,7 @@ use crate::{
     event_map::WorkspaceDatabase,
     services::persistence::{AppTableSql, TrashTableSql, ViewTableSql, WorkspaceTableSql},
 };
+use bytes::Bytes;
 use flowy_database::kv::KV;
 use flowy_error::{FlowyError, FlowyResult};
 use flowy_folder_data_model::revision::{AppRevision, FolderRevision, ViewRevision, WorkspaceRevision};
@@ -11,6 +12,7 @@ use flowy_revision::reset::{RevisionResettable, RevisionStructReset};
 use flowy_sync::client_folder::make_folder_rev_json_str;
 use flowy_sync::entities::revision::Revision;
 use flowy_sync::{client_folder::FolderPad, entities::revision::md5};
+use lib_ot::core::DeltaBuilder;
 use std::sync::Arc;
 
 const V1_MIGRATION: &str = "FOLDER_V1_MIGRATION";
@@ -129,10 +131,11 @@ impl RevisionResettable for FolderRevisionResettable {
         &self.folder_id
     }
 
-    fn target_reset_rev_str(&self, revisions: Vec<Revision>) -> FlowyResult<String> {
+    fn reset_data(&self, revisions: Vec<Revision>) -> FlowyResult<Bytes> {
         let pad = FolderPad::from_revisions(revisions)?;
         let json = pad.to_json()?;
-        Ok(json)
+        let bytes = DeltaBuilder::new().insert(&json).build().json_bytes();
+        Ok(bytes)
     }
 
     fn default_target_rev_str(&self) -> FlowyResult<String> {

frontend/rust-lib/flowy-folder/src/services/web_socket.rs

@@ -1,7 +1,10 @@
 use crate::services::FOLDER_SYNC_INTERVAL_IN_MILLIS;
 use bytes::Bytes;
-use flowy_error::FlowyError;
+use flowy_error::{FlowyError, FlowyResult};
 use flowy_revision::*;
+use flowy_sync::entities::revision::Revision;
+use flowy_sync::server_folder::FolderOperations;
+use flowy_sync::util::make_operations_from_revisions;
 use flowy_sync::{
     client_folder::FolderPad,
     entities::{
@@ -10,10 +13,32 @@ use flowy_sync::{
     },
 };
 use lib_infra::future::{BoxResultFuture, FutureResult};
-use lib_ot::core::{Delta, EmptyAttributes, OperationTransform};
+use lib_ot::core::OperationTransform;
 use parking_lot::RwLock;
 use std::{sync::Arc, time::Duration};
 
+#[derive(Clone)]
+pub struct FolderResolveOperations(pub FolderOperations);
+impl OperationsDeserializer<FolderResolveOperations> for FolderResolveOperations {
+    fn deserialize_revisions(revisions: Vec<Revision>) -> FlowyResult<FolderResolveOperations> {
+        Ok(FolderResolveOperations(make_operations_from_revisions(revisions)?))
+    }
+}
+
+impl OperationsSerializer for FolderResolveOperations {
+    fn serialize_operations(&self) -> Bytes {
+        self.0.json_bytes()
+    }
+}
+
+impl FolderResolveOperations {
+    pub fn into_inner(self) -> FolderOperations {
+        self.0
+    }
+}
+
+pub type FolderConflictController = ConflictController<FolderResolveOperations>;
+
 #[allow(dead_code)]
 pub(crate) async fn make_folder_ws_manager(
     user_id: &str,
@@ -25,7 +50,7 @@ pub(crate) async fn make_folder_ws_manager(
     let ws_data_provider = Arc::new(WSDataProvider::new(folder_id, Arc::new(rev_manager.clone())));
     let resolver = Arc::new(FolderConflictResolver { folder_pad });
     let conflict_controller =
-        ConflictController::<EmptyAttributes>::new(user_id, resolver, Arc::new(ws_data_provider.clone()), rev_manager);
+        FolderConflictController::new(user_id, resolver, Arc::new(ws_data_provider.clone()), rev_manager);
     let ws_data_stream = Arc::new(FolderRevisionWSDataStream::new(conflict_controller));
     let ws_data_sink = Arc::new(FolderWSDataSink(ws_data_provider));
     let ping_duration = Duration::from_millis(FOLDER_SYNC_INTERVAL_IN_MILLIS);
@@ -51,52 +76,57 @@ struct FolderConflictResolver {
     folder_pad: Arc<RwLock<FolderPad>>,
 }
 
-impl ConflictResolver<EmptyAttributes> for FolderConflictResolver {
-    fn compose_operations(&self, delta: Delta) -> BoxResultFuture<OperationsMD5, FlowyError> {
+impl ConflictResolver<FolderResolveOperations> for FolderConflictResolver {
+    fn compose_operations(&self, operations: FolderResolveOperations) -> BoxResultFuture<OperationsMD5, FlowyError> {
+        let operations = operations.into_inner();
         let folder_pad = self.folder_pad.clone();
         Box::pin(async move {
-            let md5 = folder_pad.write().compose_remote_operations(delta)?;
+            let md5 = folder_pad.write().compose_remote_operations(operations)?;
             Ok(md5)
         })
     }
 
-    fn transform_operations(&self, delta: Delta) -> BoxResultFuture<TransformOperations<EmptyAttributes>, FlowyError> {
+    fn transform_operations(
+        &self,
+        operations: FolderResolveOperations,
+    ) -> BoxResultFuture<TransformOperations<FolderResolveOperations>, FlowyError> {
         let folder_pad = self.folder_pad.clone();
+        let operations = operations.into_inner();
         Box::pin(async move {
             let read_guard = folder_pad.read();
-            let mut server_prime: Option<Delta> = None;
-            let client_prime: Delta;
+            let mut server_operations: Option<FolderResolveOperations> = None;
+            let client_operations: FolderResolveOperations;
             if read_guard.is_empty() {
                 // Do nothing
-                client_prime = delta;
+                client_operations = FolderResolveOperations(operations);
             } else {
-                let (s_prime, c_prime) = read_guard.get_operations().transform(&delta)?;
-                client_prime = c_prime;
-                server_prime = Some(s_prime);
+                let (s_prime, c_prime) = read_guard.get_operations().transform(&operations)?;
+                client_operations = FolderResolveOperations(c_prime);
+                server_operations = Some(FolderResolveOperations(s_prime));
             }
             drop(read_guard);
             Ok(TransformOperations {
-                client_prime,
-                server_prime,
+                client_operations,
+                server_operations,
             })
         })
     }
 
-    fn reset_operations(&self, delta: Delta) -> BoxResultFuture<OperationsMD5, FlowyError> {
+    fn reset_operations(&self, operations: FolderResolveOperations) -> BoxResultFuture<OperationsMD5, FlowyError> {
         let folder_pad = self.folder_pad.clone();
         Box::pin(async move {
-            let md5 = folder_pad.write().reset_folder(delta)?;
+            let md5 = folder_pad.write().reset_folder(operations.into_inner())?;
             Ok(md5)
         })
     }
 }
 
 struct FolderRevisionWSDataStream {
-    conflict_controller: Arc<PlainTextConflictController>,
+    conflict_controller: Arc<FolderConflictController>,
 }
 
 impl FolderRevisionWSDataStream {
-    pub fn new(conflict_controller: PlainTextConflictController) -> Self {
+    pub fn new(conflict_controller: FolderConflictController) -> Self {
         Self {
             conflict_controller: Arc::new(conflict_controller),
         }

frontend/rust-lib/flowy-grid/src/services/persistence/migration.rs

@@ -1,5 +1,6 @@
 use crate::manager::GridUser;
 use crate::services::persistence::GridDatabase;
+use bytes::Bytes;
 use flowy_database::kv::KV;
 use flowy_error::FlowyResult;
 use flowy_grid_data_model::revision::GridRevision;
@@ -8,6 +9,7 @@ use flowy_revision::reset::{RevisionResettable, RevisionStructReset};
 use flowy_sync::client_grid::{make_grid_rev_json_str, GridRevisionPad};
 use flowy_sync::entities::revision::Revision;
 use flowy_sync::util::md5;
+use lib_ot::core::DeltaBuilder;
 use std::sync::Arc;
 
 const V1_MIGRATION: &str = "GRID_V1_MIGRATION";
@@ -59,10 +61,11 @@ impl RevisionResettable for GridRevisionResettable {
         &self.grid_id
     }
 
-    fn target_reset_rev_str(&self, revisions: Vec<Revision>) -> FlowyResult<String> {
+    fn reset_data(&self, revisions: Vec<Revision>) -> FlowyResult<Bytes> {
         let pad = GridRevisionPad::from_revisions(revisions)?;
         let json = pad.json_str()?;
-        Ok(json)
+        let bytes = DeltaBuilder::new().insert(&json).build().json_bytes();
+        Ok(bytes)
     }
 
     fn default_target_rev_str(&self) -> FlowyResult<String> {

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

@@ -7,7 +7,6 @@ edition = "2018"
 
 [dependencies]
 flowy-sync = { path = "../../../shared-lib/flowy-sync" }
-lib-ot = { path = "../../../shared-lib/lib-ot" }
 lib-ws = { path = "../../../shared-lib/lib-ws" }
 lib-infra = { path = "../../../shared-lib/lib-infra" }
 flowy-database = { path = "../flowy-database" }
@@ -26,4 +25,4 @@ async-stream = "0.3.2"
 serde_json = {version = "1.0"}
 
 [features]
-flowy_unit_test = ["lib-ot/flowy_unit_test"]
+flowy_unit_test = []

frontend/rust-lib/flowy-revision/src/cache/reset.rs

@@ -1,17 +1,18 @@
 use crate::disk::{RevisionDiskCache, RevisionRecord};
 use crate::{RevisionLoader, RevisionPersistence};
+use bytes::Bytes;
 use flowy_database::kv::KV;
 use flowy_error::{FlowyError, FlowyResult};
 use flowy_sync::entities::revision::Revision;
-use lib_ot::core::DeltaBuilder;
 use serde::{Deserialize, Serialize};
 use std::str::FromStr;
 use std::sync::Arc;
 
 pub trait RevisionResettable {
     fn target_id(&self) -> &str;
+
     // String in json format
-    fn target_reset_rev_str(&self, revisions: Vec<Revision>) -> FlowyResult<String>;
+    fn reset_data(&self, revisions: Vec<Revision>) -> FlowyResult<Bytes>;
 
     // String in json format
     fn default_target_rev_str(&self) -> FlowyResult<String>;
@@ -69,9 +70,8 @@ where
         .load()
         .await?;
 
-        let s = self.target.target_reset_rev_str(revisions)?;
-        let delta_data = DeltaBuilder::new().insert(&s).build().json_bytes();
-        let revision = Revision::initial_revision(&self.user_id, self.target.target_id(), delta_data);
+        let bytes = self.target.reset_data(revisions)?;
+        let revision = Revision::initial_revision(&self.user_id, self.target.target_id(), bytes);
         let record = RevisionRecord::new(revision);
 
         tracing::trace!("Reset {} revision record object", self.target.target_id());

frontend/rust-lib/flowy-revision/src/conflict_resolve.rs

@@ -1,28 +1,39 @@
 use crate::RevisionManager;
 use bytes::Bytes;
 use flowy_error::{FlowyError, FlowyResult};
-use flowy_sync::{
-    entities::{
-        revision::{RepeatedRevision, Revision, RevisionRange},
-        ws_data::ServerRevisionWSDataType,
-    },
-    util::make_operations_from_revisions,
+use flowy_sync::entities::{
+    revision::{RepeatedRevision, Revision, RevisionRange},
+    ws_data::ServerRevisionWSDataType,
 };
 use lib_infra::future::BoxResultFuture;
-use lib_ot::core::{AttributeHashMap, DeltaOperations, EmptyAttributes, OperationAttributes};
 
-use serde::de::DeserializeOwned;
 use std::{convert::TryFrom, sync::Arc};
-
 pub type OperationsMD5 = String;
 
-pub trait ConflictResolver<T>
+pub struct TransformOperations<Operations> {
+    pub client_operations: Operations,
+    pub server_operations: Option<Operations>,
+}
+
+pub trait OperationsDeserializer<T>: Send + Sync {
+    fn deserialize_revisions(revisions: Vec<Revision>) -> FlowyResult<T>;
+}
+
+pub trait OperationsSerializer: Send + Sync {
+    fn serialize_operations(&self) -> Bytes;
+}
+
+pub struct ConflictOperations<T>(T);
+pub trait ConflictResolver<Operations>
 where
-    T: OperationAttributes + Send + Sync,
+    Operations: Send + Sync,
 {
-    fn compose_operations(&self, delta: DeltaOperations<T>) -> BoxResultFuture<OperationsMD5, FlowyError>;
-    fn transform_operations(&self, delta: DeltaOperations<T>) -> BoxResultFuture<TransformOperations<T>, FlowyError>;
-    fn reset_operations(&self, delta: DeltaOperations<T>) -> BoxResultFuture<OperationsMD5, FlowyError>;
+    fn compose_operations(&self, operations: Operations) -> BoxResultFuture<OperationsMD5, FlowyError>;
+    fn transform_operations(
+        &self,
+        operations: Operations,
+    ) -> BoxResultFuture<TransformOperations<Operations>, FlowyError>;
+    fn reset_operations(&self, operations: Operations) -> BoxResultFuture<OperationsMD5, FlowyError>;
 }
 
 pub trait ConflictRevisionSink: Send + Sync + 'static {
@@ -30,26 +41,23 @@ pub trait ConflictRevisionSink: Send + Sync + 'static {
     fn ack(&self, rev_id: String, ty: ServerRevisionWSDataType) -> BoxResultFuture<(), FlowyError>;
 }
 
-pub type RichTextConflictController = ConflictController<AttributeHashMap>;
-pub type PlainTextConflictController = ConflictController<EmptyAttributes>;
-
-pub struct ConflictController<T>
+pub struct ConflictController<Operations>
 where
-    T: OperationAttributes + Send + Sync,
+    Operations: Send + Sync,
 {
     user_id: String,
-    resolver: Arc<dyn ConflictResolver<T> + Send + Sync>,
+    resolver: Arc<dyn ConflictResolver<Operations> + Send + Sync>,
     rev_sink: Arc<dyn ConflictRevisionSink>,
     rev_manager: Arc<RevisionManager>,
 }
 
-impl<T> ConflictController<T>
+impl<Operations> ConflictController<Operations>
 where
-    T: OperationAttributes + Send + Sync + DeserializeOwned + serde::Serialize,
+    Operations: Clone + Send + Sync,
 {
     pub fn new(
         user_id: &str,
-        resolver: Arc<dyn ConflictResolver<T> + Send + Sync>,
+        resolver: Arc<dyn ConflictResolver<Operations> + Send + Sync>,
         rev_sink: Arc<dyn ConflictRevisionSink>,
         rev_manager: Arc<RevisionManager>,
     ) -> Self {
@@ -61,7 +69,12 @@ where
             rev_manager,
         }
     }
+}
 
+impl<Operations> ConflictController<Operations>
+where
+    Operations: OperationsSerializer + OperationsDeserializer<Operations> + Clone + Send + Sync,
+{
     pub async fn receive_bytes(&self, bytes: Bytes) -> FlowyResult<()> {
         let repeated_revision = RepeatedRevision::try_from(bytes)?;
         if repeated_revision.is_empty() {
@@ -103,33 +116,32 @@ where
             }
         }
 
-        let new_delta = make_operations_from_revisions(revisions.clone())?;
-
+        let new_operations = Operations::deserialize_revisions(revisions.clone())?;
         let TransformOperations {
-            client_prime,
-            server_prime,
-        } = self.resolver.transform_operations(new_delta).await?;
+            client_operations,
+            server_operations,
+        } = self.resolver.transform_operations(new_operations).await?;
 
-        match server_prime {
+        match server_operations {
             None => {
                 // The server_prime is None means the client local revisions conflict with the
                 // // server, and it needs to override the client delta.
-                let md5 = self.resolver.reset_operations(client_prime).await?;
+                let md5 = self.resolver.reset_operations(client_operations).await?;
                 let repeated_revision = RepeatedRevision::new(revisions);
                 assert_eq!(repeated_revision.last().unwrap().md5, md5);
                 let _ = self.rev_manager.reset_object(repeated_revision).await?;
                 Ok(None)
             }
-            Some(server_prime) => {
-                let md5 = self.resolver.compose_operations(client_prime.clone()).await?;
+            Some(server_operations) => {
+                let md5 = self.resolver.compose_operations(client_operations.clone()).await?;
                 for revision in &revisions {
                     let _ = self.rev_manager.add_remote_revision(revision).await?;
                 }
                 let (client_revision, server_revision) = make_client_and_server_revision(
                     &self.user_id,
                     &self.rev_manager,
-                    client_prime,
-                    Some(server_prime),
+                    client_operations,
+                    Some(server_operations),
                     md5,
                 );
                 let _ = self.rev_manager.add_remote_revision(&client_revision).await?;
@@ -139,48 +151,26 @@ where
     }
 }
 
-fn make_client_and_server_revision<T>(
+fn make_client_and_server_revision<Operations>(
     user_id: &str,
     rev_manager: &Arc<RevisionManager>,
-    client_delta: DeltaOperations<T>,
-    server_delta: Option<DeltaOperations<T>>,
+    client_operations: Operations,
+    server_operations: Option<Operations>,
     md5: String,
 ) -> (Revision, Option<Revision>)
 where
-    T: OperationAttributes + serde::Serialize,
+    Operations: OperationsSerializer,
 {
     let (base_rev_id, rev_id) = rev_manager.next_rev_id_pair();
-    let client_revision = Revision::new(
-        &rev_manager.object_id,
-        base_rev_id,
-        rev_id,
-        client_delta.json_bytes(),
-        user_id,
-        md5.clone(),
-    );
-
-    match server_delta {
+    let bytes = client_operations.serialize_operations();
+    let client_revision = Revision::new(&rev_manager.object_id, base_rev_id, rev_id, bytes, user_id, md5.clone());
+
+    match server_operations {
         None => (client_revision, None),
-        Some(server_delta) => {
-            let server_revision = Revision::new(
-                &rev_manager.object_id,
-                base_rev_id,
-                rev_id,
-                server_delta.json_bytes(),
-                user_id,
-                md5,
-            );
+        Some(operations) => {
+            let bytes = operations.serialize_operations();
+            let server_revision = Revision::new(&rev_manager.object_id, base_rev_id, rev_id, bytes, user_id, md5);
             (client_revision, Some(server_revision))
         }
     }
 }
-
-pub type TextTransformOperations = TransformOperations<AttributeHashMap>;
-
-pub struct TransformOperations<T>
-where
-    T: OperationAttributes,
-{
-    pub client_prime: DeltaOperations<T>,
-    pub server_prime: Option<DeltaOperations<T>>,
-}

shared-lib/flowy-sync/src/entities/document.rs

@@ -3,7 +3,7 @@ use crate::{
     errors::CollaborateError,
 };
 use flowy_derive::ProtoBuf;
-use lib_ot::{errors::OTError, text_delta::TextOperations};
+use lib_ot::text_delta::TextOperations;
 
 #[derive(ProtoBuf, Default, Debug, Clone)]
 pub struct CreateDocumentParams {

shared-lib/lib-ot/Cargo.toml

@@ -8,8 +8,8 @@ edition = "2018"
 [dependencies]
 bytecount = "0.6.0"
 serde = { version = "1.0", features = ["derive", "rc"] }
+#flowy-revision = { path = "../../frontend/rust-lib/flowy-revision" }
 #protobuf = {version = "2.18.0"}
-#flowy-derive = { path = "../flowy-derive" }
 tokio = { version = "1", features = ["sync"] }
 dashmap = "5"
 md5 = "0.7.0"

shared-lib/lib-ot/src/core/delta/ops_serde.rs

@@ -1,5 +1,6 @@
 use crate::core::delta::operation::OperationAttributes;
 use crate::core::delta::DeltaOperations;
+use serde::de::DeserializeOwned;
 use serde::{
     de::{SeqAccess, Visitor},
     ser::SerializeSeq,