notify server when open the doc with local rev_id

app_flowy/packages/flowy_sdk/lib/protobuf/flowy-document/doc.pb.dart

@@ -282,6 +282,81 @@ class DocDelta extends $pb.GeneratedMessage {
   void clearData() => clearField(2);
 }
 
+class NewDocUser extends $pb.GeneratedMessage {
+  static final $pb.BuilderInfo _i = $pb.BuilderInfo(const $core.bool.fromEnvironment('protobuf.omit_message_names') ? '' : 'NewDocUser', createEmptyInstance: create)
+    ..aOS(1, const $core.bool.fromEnvironment('protobuf.omit_field_names') ? '' : 'userId')
+    ..aInt64(2, const $core.bool.fromEnvironment('protobuf.omit_field_names') ? '' : 'revId')
+    ..aOS(3, const $core.bool.fromEnvironment('protobuf.omit_field_names') ? '' : 'docId')
+    ..hasRequiredFields = false
+  ;
+
+  NewDocUser._() : super();
+  factory NewDocUser({
+    $core.String? userId,
+    $fixnum.Int64? revId,
+    $core.String? docId,
+  }) {
+    final _result = create();
+    if (userId != null) {
+      _result.userId = userId;
+    }
+    if (revId != null) {
+      _result.revId = revId;
+    }
+    if (docId != null) {
+      _result.docId = docId;
+    }
+    return _result;
+  }
+  factory NewDocUser.fromBuffer($core.List<$core.int> i, [$pb.ExtensionRegistry r = $pb.ExtensionRegistry.EMPTY]) => create()..mergeFromBuffer(i, r);
+  factory NewDocUser.fromJson($core.String i, [$pb.ExtensionRegistry r = $pb.ExtensionRegistry.EMPTY]) => create()..mergeFromJson(i, r);
+  @$core.Deprecated(
+  'Using this can add significant overhead to your binary. '
+  'Use [GeneratedMessageGenericExtensions.deepCopy] instead. '
+  'Will be removed in next major version')
+  NewDocUser clone() => NewDocUser()..mergeFromMessage(this);
+  @$core.Deprecated(
+  'Using this can add significant overhead to your binary. '
+  'Use [GeneratedMessageGenericExtensions.rebuild] instead. '
+  'Will be removed in next major version')
+  NewDocUser copyWith(void Function(NewDocUser) updates) => super.copyWith((message) => updates(message as NewDocUser)) as NewDocUser; // ignore: deprecated_member_use
+  $pb.BuilderInfo get info_ => _i;
+  @$core.pragma('dart2js:noInline')
+  static NewDocUser create() => NewDocUser._();
+  NewDocUser createEmptyInstance() => create();
+  static $pb.PbList<NewDocUser> createRepeated() => $pb.PbList<NewDocUser>();
+  @$core.pragma('dart2js:noInline')
+  static NewDocUser getDefault() => _defaultInstance ??= $pb.GeneratedMessage.$_defaultFor<NewDocUser>(create);
+  static NewDocUser? _defaultInstance;
+
+  @$pb.TagNumber(1)
+  $core.String get userId => $_getSZ(0);
+  @$pb.TagNumber(1)
+  set userId($core.String v) { $_setString(0, v); }
+  @$pb.TagNumber(1)
+  $core.bool hasUserId() => $_has(0);
+  @$pb.TagNumber(1)
+  void clearUserId() => clearField(1);
+
+  @$pb.TagNumber(2)
+  $fixnum.Int64 get revId => $_getI64(1);
+  @$pb.TagNumber(2)
+  set revId($fixnum.Int64 v) { $_setInt64(1, v); }
+  @$pb.TagNumber(2)
+  $core.bool hasRevId() => $_has(1);
+  @$pb.TagNumber(2)
+  void clearRevId() => clearField(2);
+
+  @$pb.TagNumber(3)
+  $core.String get docId => $_getSZ(2);
+  @$pb.TagNumber(3)
+  set docId($core.String v) { $_setString(2, v); }
+  @$pb.TagNumber(3)
+  $core.bool hasDocId() => $_has(2);
+  @$pb.TagNumber(3)
+  void clearDocId() => clearField(3);
+}
+
 class QueryDocParams extends $pb.GeneratedMessage {
   static final $pb.BuilderInfo _i = $pb.BuilderInfo(const $core.bool.fromEnvironment('protobuf.omit_message_names') ? '' : 'QueryDocParams', createEmptyInstance: create)
     ..aOS(1, const $core.bool.fromEnvironment('protobuf.omit_field_names') ? '' : 'docId')

app_flowy/packages/flowy_sdk/lib/protobuf/flowy-document/doc.pbjson.dart

@@ -54,6 +54,18 @@ const DocDelta$json = const {
 
 /// Descriptor for `DocDelta`. Decode as a `google.protobuf.DescriptorProto`.
 final $typed_data.Uint8List docDeltaDescriptor = $convert.base64Decode('CghEb2NEZWx0YRIVCgZkb2NfaWQYASABKAlSBWRvY0lkEhIKBGRhdGEYAiABKAlSBGRhdGE=');
+@$core.Deprecated('Use newDocUserDescriptor instead')
+const NewDocUser$json = const {
+  '1': 'NewDocUser',
+  '2': const [
+    const {'1': 'user_id', '3': 1, '4': 1, '5': 9, '10': 'userId'},
+    const {'1': 'rev_id', '3': 2, '4': 1, '5': 3, '10': 'revId'},
+    const {'1': 'doc_id', '3': 3, '4': 1, '5': 9, '10': 'docId'},
+  ],
+};
+
+/// Descriptor for `NewDocUser`. Decode as a `google.protobuf.DescriptorProto`.
+final $typed_data.Uint8List newDocUserDescriptor = $convert.base64Decode('CgpOZXdEb2NVc2VyEhcKB3VzZXJfaWQYASABKAlSBnVzZXJJZBIVCgZyZXZfaWQYAiABKANSBXJldklkEhUKBmRvY19pZBgDIAEoCVIFZG9jSWQ=');
 @$core.Deprecated('Use queryDocParamsDescriptor instead')
 const QueryDocParams$json = const {
   '1': 'QueryDocParams',

app_flowy/packages/flowy_sdk/lib/protobuf/flowy-document/ws.pbenum.dart

@@ -14,14 +14,14 @@ class WsDataType extends $pb.ProtobufEnum {
   static const WsDataType PushRev = WsDataType._(1, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'PushRev');
   static const WsDataType PullRev = WsDataType._(2, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'PullRev');
   static const WsDataType Conflict = WsDataType._(3, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'Conflict');
-  static const WsDataType NewConnection = WsDataType._(4, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'NewConnection');
+  static const WsDataType NewDocUser = WsDataType._(4, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'NewDocUser');
 
   static const $core.List<WsDataType> values = <WsDataType> [
     Acked,
     PushRev,
     PullRev,
     Conflict,
-    NewConnection,
+    NewDocUser,
   ];
 
   static final $core.Map<$core.int, WsDataType> _byValue = $pb.ProtobufEnum.initByValue(values);

app_flowy/packages/flowy_sdk/lib/protobuf/flowy-document/ws.pbjson.dart

@@ -16,12 +16,12 @@ const WsDataType$json = const {
     const {'1': 'PushRev', '2': 1},
     const {'1': 'PullRev', '2': 2},
     const {'1': 'Conflict', '2': 3},
-    const {'1': 'NewConnection', '2': 4},
+    const {'1': 'NewDocUser', '2': 4},
   ],
 };
 
 /// Descriptor for `WsDataType`. Decode as a `google.protobuf.EnumDescriptorProto`.
-final $typed_data.Uint8List wsDataTypeDescriptor = $convert.base64Decode('CgpXc0RhdGFUeXBlEgkKBUFja2VkEAASCwoHUHVzaFJldhABEgsKB1B1bGxSZXYQAhIMCghDb25mbGljdBADEhEKDU5ld0Nvbm5lY3Rpb24QBA==');
+final $typed_data.Uint8List wsDataTypeDescriptor = $convert.base64Decode('CgpXc0RhdGFUeXBlEgkKBUFja2VkEAASCwoHUHVzaFJldhABEgsKB1B1bGxSZXYQAhIMCghDb25mbGljdBADEg4KCk5ld0RvY1VzZXIQBA==');
 @$core.Deprecated('Use wsDocumentDataDescriptor instead')
 const WsDocumentData$json = const {
   '1': 'WsDocumentData',

backend/src/service/doc/ws_actor.rs

@@ -6,7 +6,7 @@ use crate::service::{
 use actix_rt::task::spawn_blocking;
 use actix_web::web::Data;
 use async_stream::stream;
-use flowy_document::protobuf::{Revision, WsDataType, WsDocumentData};
+use flowy_document::protobuf::{NewDocUser, Revision, WsDataType, WsDocumentData};
 use flowy_net::errors::{internal_error, Result as DocResult, ServerError};
 use futures::stream::StreamExt;
 use sqlx::PgPool;
@@ -69,27 +69,37 @@ impl DocWsActor {
         .await
         .map_err(internal_error)??;
 
+        let data = document_data.data;
+
         match document_data.ty {
             WsDataType::Acked => Ok(()),
-            WsDataType::PushRev => self.handle_push_rev(user, socket, document_data.data, pool).await,
+            WsDataType::PushRev => self.handle_push_rev(user, socket, data, pool).await,
-            WsDataType::NewConnection => {
+            WsDataType::NewDocUser => self.handle_new_doc_user(socket, data).await,
-                // TODO: send notifications to other users who visited the doc
-                Ok(())
-            },
             WsDataType::PullRev => Ok(()),
             WsDataType::Conflict => Ok(()),
         }
     }
 
+    async fn handle_new_doc_user(&self, socket: Socket, data: Vec<u8>) -> DocResult<()> {
+        let user = spawn_blocking(move || {
+            let user: NewDocUser = parse_from_bytes(&data)?;
+            DocResult::Ok(user)
+        })
+        .await
+        .map_err(internal_error)??;
+
+        unimplemented!()
+    }
+
     async fn handle_push_rev(
         &self,
         user: Arc<WsUser>,
         socket: Socket,
-        revision_data: Vec<u8>,
+        data: Vec<u8>,
         pool: Data<PgPool>,
     ) -> DocResult<()> {
         let revision = spawn_blocking(move || {
-            let revision: Revision = parse_from_bytes(&revision_data)?;
+            let revision: Revision = parse_from_bytes(&data)?;
             let _ = verify_md5(&revision)?;
             DocResult::Ok(revision)
         })

backend/tests/document/edit.rs

@@ -4,6 +4,7 @@ use crate::document::helper::{DocScript, DocumentTest};
 async fn edit_doc_insert_text() {
     let test = DocumentTest::new().await;
     test.run_scripts(vec![
+        DocScript::ConnectWs,
         DocScript::SendText(0, "abc"),
         DocScript::SendText(3, "123"),
         DocScript::SendText(6, "efg"),

rust-lib/flowy-derive/src/derive_cache/derive_cache.rs

@@ -57,6 +57,7 @@ pub fn category_from_str(type_str: &str) -> TypeCategory {
         | "Doc"
         | "UpdateDocParams"
         | "DocDelta"
+        | "NewDocUser"
         | "QueryDocParams"
         | "RevId"
         | "Revision"

rust-lib/flowy-document/src/entities/doc/doc.rs

@@ -51,6 +51,18 @@ pub struct DocDelta {
     pub data: String, // Delta
 }
 
+#[derive(ProtoBuf, Default, Debug, Clone)]
+pub struct NewDocUser {
+    #[pb(index = 1)]
+    pub user_id: String,
+
+    #[pb(index = 2)]
+    pub rev_id: i64,
+
+    #[pb(index = 3)]
+    pub doc_id: String,
+}
+
 #[derive(ProtoBuf, Default, Debug, Clone)]
 pub struct QueryDocParams {
     #[pb(index = 1)]

rust-lib/flowy-document/src/entities/doc/revision.rs

@@ -77,14 +77,14 @@ pub struct Revision {
 
 impl std::fmt::Debug for Revision {
     fn fmt(&self, f: &mut ::std::fmt::Formatter<'_>) -> ::std::fmt::Result {
-        f.write_fmt(format_args!("doc_id {}, ", self.doc_id));
+        let _ = f.write_fmt(format_args!("doc_id {}, ", self.doc_id))?;
-        f.write_fmt(format_args!("rev_id {}, ", self.rev_id));
+        let _ = f.write_fmt(format_args!("rev_id {}, ", self.rev_id))?;
         match Delta::from_bytes(&self.delta_data) {
             Ok(delta) => {
-                f.write_fmt(format_args!("delta {:?}", delta.to_json()));
+                let _ = f.write_fmt(format_args!("delta {:?}", delta.to_json()))?;
             },
             Err(e) => {
-                f.write_fmt(format_args!("delta {:?}", e));
+                let _ = f.write_fmt(format_args!("delta {:?}", e))?;
             },
         }
         Ok(())

rust-lib/flowy-document/src/entities/ws/ws.rs

@@ -1,4 +1,7 @@
-use crate::{entities::doc::Revision, errors::DocError};
+use crate::{
+    entities::doc::{NewDocUser, Revision},
+    errors::DocError,
+};
 use bytes::Bytes;
 use flowy_derive::{ProtoBuf, ProtoBuf_Enum};
 use flowy_ws::{WsMessage, WsModule};
@@ -6,11 +9,11 @@ use std::convert::{TryFrom, TryInto};
 
 #[derive(Debug, Clone, ProtoBuf_Enum, Eq, PartialEq, Hash)]
 pub enum WsDataType {
-    Acked         = 0,
+    Acked      = 0,
-    PushRev       = 1,
+    PushRev    = 1,
-    PullRev       = 2, // data should be Revision
+    PullRev    = 2, // data should be Revision
-    Conflict      = 3,
+    Conflict   = 3,
-    NewConnection = 4,
+    NewDocUser = 4,
 }
 
 impl WsDataType {
@@ -41,13 +44,24 @@ pub struct WsDocumentData {
 
 impl std::convert::From<Revision> for WsDocumentData {
     fn from(revision: Revision) -> Self {
-        let id = revision.doc_id.clone();
+        let doc_id = revision.doc_id.clone();
         let bytes: Bytes = revision.try_into().unwrap();
-        let data = bytes.to_vec();
         Self {
-            doc_id: id,
+            doc_id,
             ty: WsDataType::PushRev,
-            data,
+            data: bytes.to_vec(),
+        }
+    }
+}
+
+impl std::convert::From<NewDocUser> for WsDocumentData {
+    fn from(user: NewDocUser) -> Self {
+        let doc_id = user.doc_id.clone();
+        let bytes: Bytes = user.try_into().unwrap();
+        Self {
+            doc_id,
+            ty: WsDataType::NewDocUser,
+            data: bytes.to_vec(),
         }
     }
 }

rust-lib/flowy-document/src/protobuf/model/doc.rs

@@ -897,6 +897,242 @@ impl ::protobuf::reflect::ProtobufValue for DocDelta {
     }
 }
 
+#[derive(PartialEq,Clone,Default)]
+pub struct NewDocUser {
+    // message fields
+    pub user_id: ::std::string::String,
+    pub rev_id: i64,
+    pub doc_id: ::std::string::String,
+    // special fields
+    pub unknown_fields: ::protobuf::UnknownFields,
+    pub cached_size: ::protobuf::CachedSize,
+}
+
+impl<'a> ::std::default::Default for &'a NewDocUser {
+    fn default() -> &'a NewDocUser {
+        <NewDocUser as ::protobuf::Message>::default_instance()
+    }
+}
+
+impl NewDocUser {
+    pub fn new() -> NewDocUser {
+        ::std::default::Default::default()
+    }
+
+    // string user_id = 1;
+
+
+    pub fn get_user_id(&self) -> &str {
+        &self.user_id
+    }
+    pub fn clear_user_id(&mut self) {
+        self.user_id.clear();
+    }
+
+    // Param is passed by value, moved
+    pub fn set_user_id(&mut self, v: ::std::string::String) {
+        self.user_id = v;
+    }
+
+    // Mutable pointer to the field.
+    // If field is not initialized, it is initialized with default value first.
+    pub fn mut_user_id(&mut self) -> &mut ::std::string::String {
+        &mut self.user_id
+    }
+
+    // Take field
+    pub fn take_user_id(&mut self) -> ::std::string::String {
+        ::std::mem::replace(&mut self.user_id, ::std::string::String::new())
+    }
+
+    // int64 rev_id = 2;
+
+
+    pub fn get_rev_id(&self) -> i64 {
+        self.rev_id
+    }
+    pub fn clear_rev_id(&mut self) {
+        self.rev_id = 0;
+    }
+
+    // Param is passed by value, moved
+    pub fn set_rev_id(&mut self, v: i64) {
+        self.rev_id = v;
+    }
+
+    // string doc_id = 3;
+
+
+    pub fn get_doc_id(&self) -> &str {
+        &self.doc_id
+    }
+    pub fn clear_doc_id(&mut self) {
+        self.doc_id.clear();
+    }
+
+    // Param is passed by value, moved
+    pub fn set_doc_id(&mut self, v: ::std::string::String) {
+        self.doc_id = v;
+    }
+
+    // Mutable pointer to the field.
+    // If field is not initialized, it is initialized with default value first.
+    pub fn mut_doc_id(&mut self) -> &mut ::std::string::String {
+        &mut self.doc_id
+    }
+
+    // Take field
+    pub fn take_doc_id(&mut self) -> ::std::string::String {
+        ::std::mem::replace(&mut self.doc_id, ::std::string::String::new())
+    }
+}
+
+impl ::protobuf::Message for NewDocUser {
+    fn is_initialized(&self) -> bool {
+        true
+    }
+
+    fn merge_from(&mut self, is: &mut ::protobuf::CodedInputStream<'_>) -> ::protobuf::ProtobufResult<()> {
+        while !is.eof()? {
+            let (field_number, wire_type) = is.read_tag_unpack()?;
+            match field_number {
+                1 => {
+                    ::protobuf::rt::read_singular_proto3_string_into(wire_type, is, &mut self.user_id)?;
+                },
+                2 => {
+                    if wire_type != ::protobuf::wire_format::WireTypeVarint {
+                        return ::std::result::Result::Err(::protobuf::rt::unexpected_wire_type(wire_type));
+                    }
+                    let tmp = is.read_int64()?;
+                    self.rev_id = tmp;
+                },
+                3 => {
+                    ::protobuf::rt::read_singular_proto3_string_into(wire_type, is, &mut self.doc_id)?;
+                },
+                _ => {
+                    ::protobuf::rt::read_unknown_or_skip_group(field_number, wire_type, is, self.mut_unknown_fields())?;
+                },
+            };
+        }
+        ::std::result::Result::Ok(())
+    }
+
+    // Compute sizes of nested messages
+    #[allow(unused_variables)]
+    fn compute_size(&self) -> u32 {
+        let mut my_size = 0;
+        if !self.user_id.is_empty() {
+            my_size += ::protobuf::rt::string_size(1, &self.user_id);
+        }
+        if self.rev_id != 0 {
+            my_size += ::protobuf::rt::value_size(2, self.rev_id, ::protobuf::wire_format::WireTypeVarint);
+        }
+        if !self.doc_id.is_empty() {
+            my_size += ::protobuf::rt::string_size(3, &self.doc_id);
+        }
+        my_size += ::protobuf::rt::unknown_fields_size(self.get_unknown_fields());
+        self.cached_size.set(my_size);
+        my_size
+    }
+
+    fn write_to_with_cached_sizes(&self, os: &mut ::protobuf::CodedOutputStream<'_>) -> ::protobuf::ProtobufResult<()> {
+        if !self.user_id.is_empty() {
+            os.write_string(1, &self.user_id)?;
+        }
+        if self.rev_id != 0 {
+            os.write_int64(2, self.rev_id)?;
+        }
+        if !self.doc_id.is_empty() {
+            os.write_string(3, &self.doc_id)?;
+        }
+        os.write_unknown_fields(self.get_unknown_fields())?;
+        ::std::result::Result::Ok(())
+    }
+
+    fn get_cached_size(&self) -> u32 {
+        self.cached_size.get()
+    }
+
+    fn get_unknown_fields(&self) -> &::protobuf::UnknownFields {
+        &self.unknown_fields
+    }
+
+    fn mut_unknown_fields(&mut self) -> &mut ::protobuf::UnknownFields {
+        &mut self.unknown_fields
+    }
+
+    fn as_any(&self) -> &dyn (::std::any::Any) {
+        self as &dyn (::std::any::Any)
+    }
+    fn as_any_mut(&mut self) -> &mut dyn (::std::any::Any) {
+        self as &mut dyn (::std::any::Any)
+    }
+    fn into_any(self: ::std::boxed::Box<Self>) -> ::std::boxed::Box<dyn (::std::any::Any)> {
+        self
+    }
+
+    fn descriptor(&self) -> &'static ::protobuf::reflect::MessageDescriptor {
+        Self::descriptor_static()
+    }
+
+    fn new() -> NewDocUser {
+        NewDocUser::new()
+    }
+
+    fn descriptor_static() -> &'static ::protobuf::reflect::MessageDescriptor {
+        static descriptor: ::protobuf::rt::LazyV2<::protobuf::reflect::MessageDescriptor> = ::protobuf::rt::LazyV2::INIT;
+        descriptor.get(|| {
+            let mut fields = ::std::vec::Vec::new();
+            fields.push(::protobuf::reflect::accessor::make_simple_field_accessor::<_, ::protobuf::types::ProtobufTypeString>(
+                "user_id",
+                |m: &NewDocUser| { &m.user_id },
+                |m: &mut NewDocUser| { &mut m.user_id },
+            ));
+            fields.push(::protobuf::reflect::accessor::make_simple_field_accessor::<_, ::protobuf::types::ProtobufTypeInt64>(
+                "rev_id",
+                |m: &NewDocUser| { &m.rev_id },
+                |m: &mut NewDocUser| { &mut m.rev_id },
+            ));
+            fields.push(::protobuf::reflect::accessor::make_simple_field_accessor::<_, ::protobuf::types::ProtobufTypeString>(
+                "doc_id",
+                |m: &NewDocUser| { &m.doc_id },
+                |m: &mut NewDocUser| { &mut m.doc_id },
+            ));
+            ::protobuf::reflect::MessageDescriptor::new_pb_name::<NewDocUser>(
+                "NewDocUser",
+                fields,
+                file_descriptor_proto()
+            )
+        })
+    }
+
+    fn default_instance() -> &'static NewDocUser {
+        static instance: ::protobuf::rt::LazyV2<NewDocUser> = ::protobuf::rt::LazyV2::INIT;
+        instance.get(NewDocUser::new)
+    }
+}
+
+impl ::protobuf::Clear for NewDocUser {
+    fn clear(&mut self) {
+        self.user_id.clear();
+        self.rev_id = 0;
+        self.doc_id.clear();
+        self.unknown_fields.clear();
+    }
+}
+
+impl ::std::fmt::Debug for NewDocUser {
+    fn fmt(&self, f: &mut ::std::fmt::Formatter<'_>) -> ::std::fmt::Result {
+        ::protobuf::text_format::fmt(self, f)
+    }
+}
+
+impl ::protobuf::reflect::ProtobufValue for NewDocUser {
+    fn as_ref(&self) -> ::protobuf::reflect::ReflectValueRef {
+        ::protobuf::reflect::ReflectValueRef::Message(self)
+    }
+}
+
 #[derive(PartialEq,Clone,Default)]
 pub struct QueryDocParams {
     // message fields
@@ -1064,45 +1300,57 @@ static file_descriptor_proto_data: &'static [u8] = b"\
     UpdateDocParams\x12\x15\n\x06doc_id\x18\x01\x20\x01(\tR\x05docId\x12\x12\
     \n\x04data\x18\x02\x20\x01(\tR\x04data\x12\x15\n\x06rev_id\x18\x03\x20\
     \x01(\x03R\x05revId\"5\n\x08DocDelta\x12\x15\n\x06doc_id\x18\x01\x20\x01\
-    (\tR\x05docId\x12\x12\n\x04data\x18\x02\x20\x01(\tR\x04data\"'\n\x0eQuer\
+    (\tR\x05docId\x12\x12\n\x04data\x18\x02\x20\x01(\tR\x04data\"S\n\nNewDoc\
-    yDocParams\x12\x15\n\x06doc_id\x18\x01\x20\x01(\tR\x05docIdJ\xe7\x05\n\
+    User\x12\x17\n\x07user_id\x18\x01\x20\x01(\tR\x06userId\x12\x15\n\x06rev\
-    \x06\x12\x04\0\0\x16\x01\n\x08\n\x01\x0c\x12\x03\0\0\x12\n\n\n\x02\x04\0\
+    _id\x18\x02\x20\x01(\x03R\x05revId\x12\x15\n\x06doc_id\x18\x03\x20\x01(\
-    \x12\x04\x02\0\x05\x01\n\n\n\x03\x04\0\x01\x12\x03\x02\x08\x17\n\x0b\n\
+    \tR\x05docId\"'\n\x0eQueryDocParams\x12\x15\n\x06doc_id\x18\x01\x20\x01(\
-    \x04\x04\0\x02\0\x12\x03\x03\x04\x12\n\x0c\n\x05\x04\0\x02\0\x05\x12\x03\
+    \tR\x05docIdJ\xa4\x07\n\x06\x12\x04\0\0\x1b\x01\n\x08\n\x01\x0c\x12\x03\
-    \x03\x04\n\n\x0c\n\x05\x04\0\x02\0\x01\x12\x03\x03\x0b\r\n\x0c\n\x05\x04\
+    \0\0\x12\n\n\n\x02\x04\0\x12\x04\x02\0\x05\x01\n\n\n\x03\x04\0\x01\x12\
-    \0\x02\0\x03\x12\x03\x03\x10\x11\n\x0b\n\x04\x04\0\x02\x01\x12\x03\x04\
+    \x03\x02\x08\x17\n\x0b\n\x04\x04\0\x02\0\x12\x03\x03\x04\x12\n\x0c\n\x05\
-    \x04\x14\n\x0c\n\x05\x04\0\x02\x01\x05\x12\x03\x04\x04\n\n\x0c\n\x05\x04\
+    \x04\0\x02\0\x05\x12\x03\x03\x04\n\n\x0c\n\x05\x04\0\x02\0\x01\x12\x03\
-    \0\x02\x01\x01\x12\x03\x04\x0b\x0f\n\x0c\n\x05\x04\0\x02\x01\x03\x12\x03\
+    \x03\x0b\r\n\x0c\n\x05\x04\0\x02\0\x03\x12\x03\x03\x10\x11\n\x0b\n\x04\
-    \x04\x12\x13\n\n\n\x02\x04\x01\x12\x04\x06\0\n\x01\n\n\n\x03\x04\x01\x01\
+    \x04\0\x02\x01\x12\x03\x04\x04\x14\n\x0c\n\x05\x04\0\x02\x01\x05\x12\x03\
-    \x12\x03\x06\x08\x0b\n\x0b\n\x04\x04\x01\x02\0\x12\x03\x07\x04\x12\n\x0c\
+    \x04\x04\n\n\x0c\n\x05\x04\0\x02\x01\x01\x12\x03\x04\x0b\x0f\n\x0c\n\x05\
-    \n\x05\x04\x01\x02\0\x05\x12\x03\x07\x04\n\n\x0c\n\x05\x04\x01\x02\0\x01\
+    \x04\0\x02\x01\x03\x12\x03\x04\x12\x13\n\n\n\x02\x04\x01\x12\x04\x06\0\n\
-    \x12\x03\x07\x0b\r\n\x0c\n\x05\x04\x01\x02\0\x03\x12\x03\x07\x10\x11\n\
+    \x01\n\n\n\x03\x04\x01\x01\x12\x03\x06\x08\x0b\n\x0b\n\x04\x04\x01\x02\0\
-    \x0b\n\x04\x04\x01\x02\x01\x12\x03\x08\x04\x14\n\x0c\n\x05\x04\x01\x02\
+    \x12\x03\x07\x04\x12\n\x0c\n\x05\x04\x01\x02\0\x05\x12\x03\x07\x04\n\n\
-    \x01\x05\x12\x03\x08\x04\n\n\x0c\n\x05\x04\x01\x02\x01\x01\x12\x03\x08\
+    \x0c\n\x05\x04\x01\x02\0\x01\x12\x03\x07\x0b\r\n\x0c\n\x05\x04\x01\x02\0\
-    \x0b\x0f\n\x0c\n\x05\x04\x01\x02\x01\x03\x12\x03\x08\x12\x13\n\x0b\n\x04\
+    \x03\x12\x03\x07\x10\x11\n\x0b\n\x04\x04\x01\x02\x01\x12\x03\x08\x04\x14\
-    \x04\x01\x02\x02\x12\x03\t\x04\x15\n\x0c\n\x05\x04\x01\x02\x02\x05\x12\
+    \n\x0c\n\x05\x04\x01\x02\x01\x05\x12\x03\x08\x04\n\n\x0c\n\x05\x04\x01\
-    \x03\t\x04\t\n\x0c\n\x05\x04\x01\x02\x02\x01\x12\x03\t\n\x10\n\x0c\n\x05\
+    \x02\x01\x01\x12\x03\x08\x0b\x0f\n\x0c\n\x05\x04\x01\x02\x01\x03\x12\x03\
-    \x04\x01\x02\x02\x03\x12\x03\t\x13\x14\n\n\n\x02\x04\x02\x12\x04\x0b\0\
+    \x08\x12\x13\n\x0b\n\x04\x04\x01\x02\x02\x12\x03\t\x04\x15\n\x0c\n\x05\
-    \x0f\x01\n\n\n\x03\x04\x02\x01\x12\x03\x0b\x08\x17\n\x0b\n\x04\x04\x02\
+    \x04\x01\x02\x02\x05\x12\x03\t\x04\t\n\x0c\n\x05\x04\x01\x02\x02\x01\x12\
-    \x02\0\x12\x03\x0c\x04\x16\n\x0c\n\x05\x04\x02\x02\0\x05\x12\x03\x0c\x04\
+    \x03\t\n\x10\n\x0c\n\x05\x04\x01\x02\x02\x03\x12\x03\t\x13\x14\n\n\n\x02\
-    \n\n\x0c\n\x05\x04\x02\x02\0\x01\x12\x03\x0c\x0b\x11\n\x0c\n\x05\x04\x02\
+    \x04\x02\x12\x04\x0b\0\x0f\x01\n\n\n\x03\x04\x02\x01\x12\x03\x0b\x08\x17\
-    \x02\0\x03\x12\x03\x0c\x14\x15\n\x0b\n\x04\x04\x02\x02\x01\x12\x03\r\x04\
+    \n\x0b\n\x04\x04\x02\x02\0\x12\x03\x0c\x04\x16\n\x0c\n\x05\x04\x02\x02\0\
-    \x14\n\x0c\n\x05\x04\x02\x02\x01\x05\x12\x03\r\x04\n\n\x0c\n\x05\x04\x02\
+    \x05\x12\x03\x0c\x04\n\n\x0c\n\x05\x04\x02\x02\0\x01\x12\x03\x0c\x0b\x11\
-    \x02\x01\x01\x12\x03\r\x0b\x0f\n\x0c\n\x05\x04\x02\x02\x01\x03\x12\x03\r\
+    \n\x0c\n\x05\x04\x02\x02\0\x03\x12\x03\x0c\x14\x15\n\x0b\n\x04\x04\x02\
-    \x12\x13\n\x0b\n\x04\x04\x02\x02\x02\x12\x03\x0e\x04\x15\n\x0c\n\x05\x04\
+    \x02\x01\x12\x03\r\x04\x14\n\x0c\n\x05\x04\x02\x02\x01\x05\x12\x03\r\x04\
-    \x02\x02\x02\x05\x12\x03\x0e\x04\t\n\x0c\n\x05\x04\x02\x02\x02\x01\x12\
+    \n\n\x0c\n\x05\x04\x02\x02\x01\x01\x12\x03\r\x0b\x0f\n\x0c\n\x05\x04\x02\
-    \x03\x0e\n\x10\n\x0c\n\x05\x04\x02\x02\x02\x03\x12\x03\x0e\x13\x14\n\n\n\
+    \x02\x01\x03\x12\x03\r\x12\x13\n\x0b\n\x04\x04\x02\x02\x02\x12\x03\x0e\
-    \x02\x04\x03\x12\x04\x10\0\x13\x01\n\n\n\x03\x04\x03\x01\x12\x03\x10\x08\
+    \x04\x15\n\x0c\n\x05\x04\x02\x02\x02\x05\x12\x03\x0e\x04\t\n\x0c\n\x05\
-    \x10\n\x0b\n\x04\x04\x03\x02\0\x12\x03\x11\x04\x16\n\x0c\n\x05\x04\x03\
+    \x04\x02\x02\x02\x01\x12\x03\x0e\n\x10\n\x0c\n\x05\x04\x02\x02\x02\x03\
-    \x02\0\x05\x12\x03\x11\x04\n\n\x0c\n\x05\x04\x03\x02\0\x01\x12\x03\x11\
+    \x12\x03\x0e\x13\x14\n\n\n\x02\x04\x03\x12\x04\x10\0\x13\x01\n\n\n\x03\
-    \x0b\x11\n\x0c\n\x05\x04\x03\x02\0\x03\x12\x03\x11\x14\x15\n\x0b\n\x04\
+    \x04\x03\x01\x12\x03\x10\x08\x10\n\x0b\n\x04\x04\x03\x02\0\x12\x03\x11\
-    \x04\x03\x02\x01\x12\x03\x12\x04\x14\n\x0c\n\x05\x04\x03\x02\x01\x05\x12\
+    \x04\x16\n\x0c\n\x05\x04\x03\x02\0\x05\x12\x03\x11\x04\n\n\x0c\n\x05\x04\
-    \x03\x12\x04\n\n\x0c\n\x05\x04\x03\x02\x01\x01\x12\x03\x12\x0b\x0f\n\x0c\
+    \x03\x02\0\x01\x12\x03\x11\x0b\x11\n\x0c\n\x05\x04\x03\x02\0\x03\x12\x03\
-    \n\x05\x04\x03\x02\x01\x03\x12\x03\x12\x12\x13\n\n\n\x02\x04\x04\x12\x04\
+    \x11\x14\x15\n\x0b\n\x04\x04\x03\x02\x01\x12\x03\x12\x04\x14\n\x0c\n\x05\
-    \x14\0\x16\x01\n\n\n\x03\x04\x04\x01\x12\x03\x14\x08\x16\n\x0b\n\x04\x04\
+    \x04\x03\x02\x01\x05\x12\x03\x12\x04\n\n\x0c\n\x05\x04\x03\x02\x01\x01\
-    \x04\x02\0\x12\x03\x15\x04\x16\n\x0c\n\x05\x04\x04\x02\0\x05\x12\x03\x15\
+    \x12\x03\x12\x0b\x0f\n\x0c\n\x05\x04\x03\x02\x01\x03\x12\x03\x12\x12\x13\
-    \x04\n\n\x0c\n\x05\x04\x04\x02\0\x01\x12\x03\x15\x0b\x11\n\x0c\n\x05\x04\
+    \n\n\n\x02\x04\x04\x12\x04\x14\0\x18\x01\n\n\n\x03\x04\x04\x01\x12\x03\
-    \x04\x02\0\x03\x12\x03\x15\x14\x15b\x06proto3\
+    \x14\x08\x12\n\x0b\n\x04\x04\x04\x02\0\x12\x03\x15\x04\x17\n\x0c\n\x05\
+    \x04\x04\x02\0\x05\x12\x03\x15\x04\n\n\x0c\n\x05\x04\x04\x02\0\x01\x12\
+    \x03\x15\x0b\x12\n\x0c\n\x05\x04\x04\x02\0\x03\x12\x03\x15\x15\x16\n\x0b\
+    \n\x04\x04\x04\x02\x01\x12\x03\x16\x04\x15\n\x0c\n\x05\x04\x04\x02\x01\
+    \x05\x12\x03\x16\x04\t\n\x0c\n\x05\x04\x04\x02\x01\x01\x12\x03\x16\n\x10\
+    \n\x0c\n\x05\x04\x04\x02\x01\x03\x12\x03\x16\x13\x14\n\x0b\n\x04\x04\x04\
+    \x02\x02\x12\x03\x17\x04\x16\n\x0c\n\x05\x04\x04\x02\x02\x05\x12\x03\x17\
+    \x04\n\n\x0c\n\x05\x04\x04\x02\x02\x01\x12\x03\x17\x0b\x11\n\x0c\n\x05\
+    \x04\x04\x02\x02\x03\x12\x03\x17\x14\x15\n\n\n\x02\x04\x05\x12\x04\x19\0\
+    \x1b\x01\n\n\n\x03\x04\x05\x01\x12\x03\x19\x08\x16\n\x0b\n\x04\x04\x05\
+    \x02\0\x12\x03\x1a\x04\x16\n\x0c\n\x05\x04\x05\x02\0\x05\x12\x03\x1a\x04\
+    \n\n\x0c\n\x05\x04\x05\x02\0\x01\x12\x03\x1a\x0b\x11\n\x0c\n\x05\x04\x05\
+    \x02\0\x03\x12\x03\x1a\x14\x15b\x06proto3\
 ";
 
 static file_descriptor_proto_lazy: ::protobuf::rt::LazyV2<::protobuf::descriptor::FileDescriptorProto> = ::protobuf::rt::LazyV2::INIT;

rust-lib/flowy-document/src/protobuf/model/ws.rs

@@ -261,7 +261,7 @@ pub enum WsDataType {
     PushRev = 1,
     PullRev = 2,
     Conflict = 3,
-    NewConnection = 4,
+    NewDocUser = 4,
 }
 
 impl ::protobuf::ProtobufEnum for WsDataType {
@@ -275,7 +275,7 @@ impl ::protobuf::ProtobufEnum for WsDataType {
             1 => ::std::option::Option::Some(WsDataType::PushRev),
             2 => ::std::option::Option::Some(WsDataType::PullRev),
             3 => ::std::option::Option::Some(WsDataType::Conflict),
-            4 => ::std::option::Option::Some(WsDataType::NewConnection),
+            4 => ::std::option::Option::Some(WsDataType::NewDocUser),
             _ => ::std::option::Option::None
         }
     }
@@ -286,7 +286,7 @@ impl ::protobuf::ProtobufEnum for WsDataType {
             WsDataType::PushRev,
             WsDataType::PullRev,
             WsDataType::Conflict,
-            WsDataType::NewConnection,
+            WsDataType::NewDocUser,
         ];
         values
     }
@@ -317,10 +317,10 @@ impl ::protobuf::reflect::ProtobufValue for WsDataType {
 static file_descriptor_proto_data: &'static [u8] = b"\
     \n\x08ws.proto\"X\n\x0eWsDocumentData\x12\x15\n\x06doc_id\x18\x01\x20\
     \x01(\tR\x05docId\x12\x1b\n\x02ty\x18\x02\x20\x01(\x0e2\x0b.WsDataTypeR\
-    \x02ty\x12\x12\n\x04data\x18\x03\x20\x01(\x0cR\x04data*R\n\nWsDataType\
+    \x02ty\x12\x12\n\x04data\x18\x03\x20\x01(\x0cR\x04data*O\n\nWsDataType\
     \x12\t\n\x05Acked\x10\0\x12\x0b\n\x07PushRev\x10\x01\x12\x0b\n\x07PullRe\
-    v\x10\x02\x12\x0c\n\x08Conflict\x10\x03\x12\x11\n\rNewConnection\x10\x04\
+    v\x10\x02\x12\x0c\n\x08Conflict\x10\x03\x12\x0e\n\nNewDocUser\x10\x04J\
-    J\xb4\x03\n\x06\x12\x04\0\0\r\x01\n\x08\n\x01\x0c\x12\x03\0\0\x12\n\n\n\
+    \xb4\x03\n\x06\x12\x04\0\0\r\x01\n\x08\n\x01\x0c\x12\x03\0\0\x12\n\n\n\
     \x02\x04\0\x12\x04\x02\0\x06\x01\n\n\n\x03\x04\0\x01\x12\x03\x02\x08\x16\
     \n\x0b\n\x04\x04\0\x02\0\x12\x03\x03\x04\x16\n\x0c\n\x05\x04\0\x02\0\x05\
     \x12\x03\x03\x04\n\n\x0c\n\x05\x04\0\x02\0\x01\x12\x03\x03\x0b\x11\n\x0c\
@@ -339,8 +339,8 @@ static file_descriptor_proto_data: &'static [u8] = b"\
     \x03\n\x04\x0b\n\x0c\n\x05\x05\0\x02\x02\x02\x12\x03\n\x0e\x0f\n\x0b\n\
     \x04\x05\0\x02\x03\x12\x03\x0b\x04\x11\n\x0c\n\x05\x05\0\x02\x03\x01\x12\
     \x03\x0b\x04\x0c\n\x0c\n\x05\x05\0\x02\x03\x02\x12\x03\x0b\x0f\x10\n\x0b\
-    \n\x04\x05\0\x02\x04\x12\x03\x0c\x04\x16\n\x0c\n\x05\x05\0\x02\x04\x01\
+    \n\x04\x05\0\x02\x04\x12\x03\x0c\x04\x13\n\x0c\n\x05\x05\0\x02\x04\x01\
-    \x12\x03\x0c\x04\x11\n\x0c\n\x05\x05\0\x02\x04\x02\x12\x03\x0c\x14\x15b\
+    \x12\x03\x0c\x04\x0e\n\x0c\n\x05\x05\0\x02\x04\x02\x12\x03\x0c\x11\x12b\
     \x06proto3\
 ";
 

rust-lib/flowy-document/src/protobuf/proto/doc.proto

@@ -18,6 +18,11 @@ message DocDelta {
     string doc_id = 1;
     string data = 2;
 }
+message NewDocUser {
+    string user_id = 1;
+    int64 rev_id = 2;
+    string doc_id = 3;
+}
 message QueryDocParams {
     string doc_id = 1;
 }

rust-lib/flowy-document/src/protobuf/proto/ws.proto

@@ -10,5 +10,5 @@ enum WsDataType {
     PushRev = 1;
     PullRev = 2;
     Conflict = 3;
-    NewConnection = 4;
+    NewDocUser = 4;
 }

rust-lib/flowy-document/src/services/doc/doc_controller.rs

@@ -120,12 +120,13 @@ impl DocController {
         // let doc = self.read_doc(doc_id, pool.clone()).await?;
         let ws_sender = self.ws.read().sender();
         let token = self.user.token()?;
+        let user = self.user.clone();
         let server = Arc::new(RevisionServerImpl {
             token,
             server: self.server.clone(),
         });
 
-        let edit_ctx = Arc::new(ClientEditDoc::new(doc_id, pool, ws_sender, server).await?);
+        let edit_ctx = Arc::new(ClientEditDoc::new(doc_id, pool, ws_sender, server, user).await?);
         self.ws.write().register_handler(doc_id, edit_ctx.clone());
         self.cache.set(edit_ctx.clone());
         Ok(edit_ctx)

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

@@ -62,8 +62,7 @@ impl Document {
         match &self.notify {
             None => {},
             Some(notify) => {
-                let notify = notify.clone();
+                let _ = notify.send(());
-                notify.send(());
             },
         }
     }

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

@@ -4,10 +4,11 @@ use crate::{
         ws::{WsDataType, WsDocumentData},
     },
     errors::{internal_error, DocError, DocResult},
+    module::DocumentUser,
     services::{
         doc::{
             edit::{actor::DocumentEditActor, message::EditMsg},
-            revision::{RevisionManager, RevisionServer},
+            revision::{DocRevision, RevisionCmd, RevisionManager, RevisionServer, RevisionStoreActor},
             UndoResult,
         },
         ws::{WsDocumentHandler, WsDocumentSender},
@@ -32,22 +33,25 @@ impl ClientEditDoc {
     pub(crate) async fn new(
         doc_id: &str,
         pool: Arc<ConnectionPool>,
-        ws_sender: Arc<dyn WsDocumentSender>,
+        ws: Arc<dyn WsDocumentSender>,
         server: Arc<dyn RevisionServer>,
+        user: Arc<dyn DocumentUser>,
     ) -> DocResult<Self> {
-        let (rev_manager, delta) = RevisionManager::new(doc_id, pool.clone(), ws_sender, server).await?;
+        let user_id = user.user_id()?;
-        let rev_manager = Arc::new(rev_manager);
+        let rev_store = spawn_rev_store_actor(doc_id, pool.clone(), server.clone());
-        let (sender, receiver) = mpsc::unbounded_channel::<EditMsg>();
+        let DocRevision { rev_id, delta } = fetch_document(rev_store.clone()).await?;
-        let edit_actor = DocumentEditActor::new(doc_id, delta, pool.clone(), receiver);
-        tokio::spawn(edit_actor.run());
 
-        let edit_context = Self {
+        log::info!("😁 Document delta: {:?}", delta);
-            doc_id: doc_id.to_string(),
+
+        let rev_manager = Arc::new(RevisionManager::new(doc_id, &user_id, rev_id, ws, rev_store));
+        let document = spawn_doc_edit_actor(doc_id, delta, pool.clone());
+        let doc_id = doc_id.to_string();
+        Ok(Self {
+            doc_id,
             rev_manager,
-            document: sender,
+            document,
             pool,
-        };
+        })
-        Ok(edit_context)
     }
 
     pub async fn insert<T: ToString>(&self, index: usize, data: T) -> Result<(), DocError> {
@@ -141,7 +145,7 @@ impl ClientEditDoc {
         let (base_rev_id, rev_id) = self.rev_manager.next_rev_id();
         let delta_data = delta_data.to_vec();
         let revision = Revision::new(base_rev_id, rev_id, delta_data, &self.doc_id, RevType::Local);
-        self.rev_manager.add_revision(revision).await;
+        let _ = self.rev_manager.add_revision(revision).await?;
         Ok(rev_id.into())
     }
 
@@ -180,7 +184,7 @@ impl WsDocumentHandler for ClientEditDoc {
                     let range = RevisionRange::try_from(bytes)?;
                     let _ = rev_manager.send_revisions(range)?;
                 },
-                WsDataType::NewConnection => {},
+                WsDataType::NewDocUser => {},
                 WsDataType::Acked => {
                     let rev_id = RevId::try_from(bytes)?;
                     let _ = rev_manager.ack_rev(rev_id);
@@ -230,3 +234,34 @@ async fn handle_push_rev(
         },
     }
 }
+
+fn spawn_rev_store_actor(
+    doc_id: &str,
+    pool: Arc<ConnectionPool>,
+    server: Arc<dyn RevisionServer>,
+) -> mpsc::Sender<RevisionCmd> {
+    let (sender, receiver) = mpsc::channel::<RevisionCmd>(50);
+    let actor = RevisionStoreActor::new(doc_id, pool, receiver, server);
+    tokio::spawn(actor.run());
+    sender
+}
+
+fn spawn_doc_edit_actor(doc_id: &str, delta: Delta, pool: Arc<ConnectionPool>) -> UnboundedSender<EditMsg> {
+    let (sender, receiver) = mpsc::unbounded_channel::<EditMsg>();
+    let actor = DocumentEditActor::new(&doc_id, delta, pool.clone(), receiver);
+    tokio::spawn(actor.run());
+    sender
+}
+
+async fn fetch_document(sender: mpsc::Sender<RevisionCmd>) -> DocResult<DocRevision> {
+    let (ret, rx) = oneshot::channel();
+    let _ = sender.send(RevisionCmd::DocumentDelta { ret }).await;
+
+    match rx.await {
+        Ok(result) => Ok(result?),
+        Err(e) => {
+            log::error!("fetch_document: {}", e);
+            Err(DocError::internal().context(format!("fetch_document: {}", e)))
+        },
+    }
+}

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

@@ -15,7 +15,7 @@ use tokio::{
     task::{spawn_blocking, JoinHandle},
 };
 
-pub enum StoreCmd {
+pub enum RevisionCmd {
     Revision {
         revision: Revision,
     },
@@ -31,22 +31,22 @@ pub enum StoreCmd {
     },
 }
 
-pub struct RevisionStore {
+pub struct RevisionStoreActor {
     doc_id: String,
     persistence: Arc<Persistence>,
     revs: Arc<DashMap<i64, RevisionOperation>>,
     delay_save: RwLock<Option<JoinHandle<()>>>,
-    receiver: Option<mpsc::Receiver<StoreCmd>>,
+    receiver: Option<mpsc::Receiver<RevisionCmd>>,
     server: Arc<dyn RevisionServer>,
 }
 
-impl RevisionStore {
+impl RevisionStoreActor {
     pub fn new(
         doc_id: &str,
         pool: Arc<ConnectionPool>,
-        receiver: mpsc::Receiver<StoreCmd>,
+        receiver: mpsc::Receiver<RevisionCmd>,
         server: Arc<dyn RevisionServer>,
-    ) -> RevisionStore {
+    ) -> RevisionStoreActor {
         let persistence = Arc::new(Persistence::new(pool));
         let revs = Arc::new(DashMap::new());
         let doc_id = doc_id.to_owned();
@@ -74,19 +74,19 @@ impl RevisionStore {
         stream.for_each(|msg| self.handle_message(msg)).await;
     }
 
-    async fn handle_message(&self, cmd: StoreCmd) {
+    async fn handle_message(&self, cmd: RevisionCmd) {
         match cmd {
-            StoreCmd::Revision { revision } => {
+            RevisionCmd::Revision { revision } => {
                 self.handle_new_revision(revision).await;
             },
-            StoreCmd::AckRevision { rev_id } => {
+            RevisionCmd::AckRevision { rev_id } => {
                 self.handle_revision_acked(rev_id).await;
             },
-            StoreCmd::SendRevisions { range, ret } => {
+            RevisionCmd::SendRevisions { range, ret } => {
                 let result = revs_in_range(&self.doc_id, self.persistence.clone(), range).await;
                 let _ = ret.send(result);
             },
-            StoreCmd::DocumentDelta { ret } => {
+            RevisionCmd::DocumentDelta { ret } => {
                 let delta = fetch_document(&self.doc_id, self.server.clone(), self.persistence.clone()).await;
                 let _ = ret.send(delta);
             },
@@ -224,15 +224,13 @@ async fn revs_in_range(doc_id: &str, persistence: Arc<Persistence>, range: Revis
 
 struct Persistence {
     rev_sql: Arc<RevTableSql>,
-    doc_sql: Arc<DocTableSql>,
     pool: Arc<ConnectionPool>,
 }
 
 impl Persistence {
     fn new(pool: Arc<ConnectionPool>) -> Self {
         let rev_sql = Arc::new(RevTableSql {});
-        let doc_sql = Arc::new(DocTableSql {});
+        Self { rev_sql, pool }
-        Self { rev_sql, doc_sql, pool }
     }
 }
 

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

@@ -2,13 +2,16 @@ use crate::{
     entities::doc::{RevType, Revision, RevisionRange},
     errors::DocError,
     services::{
-        doc::revision::store::{RevisionStore, StoreCmd},
+        doc::revision::actor::{RevisionCmd, RevisionStoreActor},
         util::RevIdCounter,
         ws::WsDocumentSender,
     },
 };
 
-use crate::{entities::doc::RevId, errors::DocResult};
+use crate::{
+    entities::doc::{NewDocUser, RevId},
+    errors::DocResult,
+};
 use flowy_database::ConnectionPool;
 use flowy_infra::future::ResultFuture;
 use flowy_ot::core::Delta;
@@ -27,37 +30,33 @@ pub trait RevisionServer: Send + Sync {
 
 pub struct RevisionManager {
     doc_id: String,
+    user_id: String,
     rev_id_counter: RevIdCounter,
     ws: Arc<dyn WsDocumentSender>,
-    store: mpsc::Sender<StoreCmd>,
+    rev_store: mpsc::Sender<RevisionCmd>,
     pending_revs: RwLock<VecDeque<Revision>>,
 }
 
 impl RevisionManager {
-    pub async fn new(
+    pub fn new(
         doc_id: &str,
-        pool: Arc<ConnectionPool>,
+        user_id: &str,
+        rev_id: RevId,
         ws: Arc<dyn WsDocumentSender>,
-        server: Arc<dyn RevisionServer>,
+        rev_store: mpsc::Sender<RevisionCmd>,
-    ) -> DocResult<(Self, Delta)> {
+    ) -> Self {
-        let (sender, receiver) = mpsc::channel::<StoreCmd>(50);
+        notify_open_doc(&ws, user_id, doc_id, &rev_id);
-        let store = RevisionStore::new(doc_id, pool, receiver, server);
-        tokio::spawn(store.run());
 
-        let DocRevision { rev_id, delta } = fetch_document(sender.clone()).await?;
-        log::info!("😁Document delta: {:?}", delta);
-
-        let doc_id = doc_id.to_string();
         let rev_id_counter = RevIdCounter::new(rev_id.into());
         let pending_revs = RwLock::new(VecDeque::new());
-        let manager = Self {
+        Self {
-            doc_id,
+            doc_id: doc_id.to_string(),
+            user_id: user_id.to_string(),
             rev_id_counter,
             ws,
             pending_revs,
-            store: sender,
+            rev_store,
-        };
+        }
-        Ok((manager, delta))
     }
 
     pub fn push_compose_revision(&self, revision: Revision) { self.pending_revs.write().push_front(revision); }
@@ -66,10 +65,10 @@ impl RevisionManager {
 
     #[tracing::instrument(level = "debug", skip(self))]
     pub async fn add_revision(&self, revision: Revision) -> Result<(), DocError> {
-        let cmd = StoreCmd::Revision {
+        let cmd = RevisionCmd::Revision {
             revision: revision.clone(),
         };
-        let _ = self.store.send(cmd).await;
+        let _ = self.rev_store.send(cmd).await;
 
         match revision.ty {
             RevType::Local => match self.ws.send(revision.into()) {
@@ -85,9 +84,9 @@ impl RevisionManager {
     }
 
     pub fn ack_rev(&self, rev_id: RevId) -> Result<(), DocError> {
-        let sender = self.store.clone();
+        let sender = self.rev_store.clone();
         tokio::spawn(async move {
-            let _ = sender.send(StoreCmd::AckRevision { rev_id }).await;
+            let _ = sender.send(RevisionCmd::AckRevision { rev_id }).await;
         });
         Ok(())
     }
@@ -103,25 +102,25 @@ impl RevisionManager {
     pub fn send_revisions(&self, range: RevisionRange) -> Result<(), DocError> {
         debug_assert!(&range.doc_id == &self.doc_id);
         let (ret, _rx) = oneshot::channel();
-        let sender = self.store.clone();
+        let sender = self.rev_store.clone();
 
         tokio::spawn(async move {
-            let _ = sender.send(StoreCmd::SendRevisions { range, ret }).await;
+            let _ = sender.send(RevisionCmd::SendRevisions { range, ret }).await;
         });
 
         unimplemented!()
     }
 }
 
-async fn fetch_document(sender: mpsc::Sender<StoreCmd>) -> DocResult<DocRevision> {
+fn notify_open_doc(ws: &Arc<dyn WsDocumentSender>, user_id: &str, doc_id: &str, rev_id: &RevId) {
-    let (ret, rx) = oneshot::channel();
+    let new_doc_user = NewDocUser {
-    let _ = sender.send(StoreCmd::DocumentDelta { ret }).await;
+        user_id: user_id.to_string(),
+        rev_id: rev_id.clone().into(),
+        doc_id: doc_id.to_string(),
+    };
 
-    match rx.await {
+    match ws.send(new_doc_user.into()) {
-        Ok(result) => Ok(result?),
+        Ok(_) => {},
-        Err(e) => {
+        Err(e) => log::error!("Send new_doc_user failed: {:?}", e),
-            log::error!("fetch_document: {}", e);
-            Err(DocError::internal().context(format!("fetch_document: {}", e)))
-        },
     }
 }

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

@@ -1,5 +1,6 @@
+mod actor;
 mod manager;
-mod store;
 mod util;
 
+pub use actor::*;
 pub use manager::*;

rust-lib/flowy-infra/src/errors/builder.rs

@@ -1,46 +0,0 @@
-use std::{fmt::Debug, marker::PhantomData};
-
-pub trait Build<C> {
-    fn build(code: C, msg: String) -> Self;
-}
-#[allow(dead_code)]
-pub struct Builder<C, O> {
-    pub code: C,
-    pub msg: Option<String>,
-    phantom: PhantomData<O>,
-}
-
-impl<C, O> Builder<C, O>
-where
-    C: Debug,
-    O: Build<C> + Build<C>,
-{
-    pub fn new(code: C) -> Self {
-        Builder {
-            code,
-            msg: None,
-            phantom: PhantomData,
-        }
-    }
-
-    pub fn msg<M>(mut self, msg: M) -> Self
-    where
-        M: Into<String>,
-    {
-        self.msg = Some(msg.into());
-        self
-    }
-
-    pub fn error<Err>(mut self, err: Err) -> Self
-    where
-        Err: std::fmt::Debug,
-    {
-        self.msg = Some(format!("{:?}", err));
-        self
-    }
-
-    pub fn build(mut self) -> O {
-        let msg = self.msg.take().unwrap_or("".to_owned());
-        O::build(self.code, msg)
-    }
-}

rust-lib/flowy-infra/src/errors/mod.rs

@@ -1,3 +0,0 @@
-pub mod builder;
-
-pub use builder::*;

rust-lib/flowy-infra/src/lib.rs

@@ -4,7 +4,6 @@ extern crate diesel;
 #[macro_use]
 extern crate diesel_derives;
 
-pub mod errors;
 pub mod future;
 pub mod kv;
 mod protobuf;