config user observable

app_flowy/packages/flowy_sdk/lib/protobuf/flowy-user/errors.pbenum.dart

@@ -34,8 +34,10 @@ class ErrorCode extends $pb.ProtobufEnum {
   static const ErrorCode UserNameIsEmpty = ErrorCode._(42, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'UserNameIsEmpty');
   static const ErrorCode UserWorkspaceInvalid = ErrorCode._(50, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'UserWorkspaceInvalid');
   static const ErrorCode UserIdInvalid = ErrorCode._(51, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'UserIdInvalid');
-  static const ErrorCode CreateDefaultWorkspaceFailed = ErrorCode._(52, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'CreateDefaultWorkspaceFailed');
-  static const ErrorCode DefaultWorkspaceAlreadyExist = ErrorCode._(53, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'DefaultWorkspaceAlreadyExist');
+  static const ErrorCode UserTokenInvalid = ErrorCode._(54, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'UserTokenInvalid');
+  static const ErrorCode UserNotExist = ErrorCode._(55, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'UserNotExist');
+  static const ErrorCode CreateDefaultWorkspaceFailed = ErrorCode._(60, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'CreateDefaultWorkspaceFailed');
+  static const ErrorCode DefaultWorkspaceAlreadyExist = ErrorCode._(61, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'DefaultWorkspaceAlreadyExist');
   static const ErrorCode ServerError = ErrorCode._(100, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'ServerError');
 
   static const $core.List<ErrorCode> values = <ErrorCode> [
@@ -63,6 +65,8 @@ class ErrorCode extends $pb.ProtobufEnum {
     UserNameIsEmpty,
     UserWorkspaceInvalid,
     UserIdInvalid,
+    UserTokenInvalid,
+    UserNotExist,
     CreateDefaultWorkspaceFailed,
     DefaultWorkspaceAlreadyExist,
     ServerError,

app_flowy/packages/flowy_sdk/lib/protobuf/flowy-user/errors.pbjson.dart

@@ -36,14 +36,16 @@ const ErrorCode$json = const {
     const {'1': 'UserNameIsEmpty', '2': 42},
     const {'1': 'UserWorkspaceInvalid', '2': 50},
     const {'1': 'UserIdInvalid', '2': 51},
-    const {'1': 'CreateDefaultWorkspaceFailed', '2': 52},
-    const {'1': 'DefaultWorkspaceAlreadyExist', '2': 53},
+    const {'1': 'UserTokenInvalid', '2': 54},
+    const {'1': 'UserNotExist', '2': 55},
+    const {'1': 'CreateDefaultWorkspaceFailed', '2': 60},
+    const {'1': 'DefaultWorkspaceAlreadyExist', '2': 61},
     const {'1': 'ServerError', '2': 100},
   ],
 };
 
 /// Descriptor for `ErrorCode`. Decode as a `google.protobuf.EnumDescriptorProto`.
-final $typed_data.Uint8List errorCodeDescriptor = $convert.base64Decode('CglFcnJvckNvZGUSCwoHVW5rbm93bhAAEhoKFlVzZXJEYXRhYmFzZUluaXRGYWlsZWQQARIcChhBY3F1aXJlV3JpdGVMb2NrZWRGYWlsZWQQAhIbChdBY3F1aXJlUmVhZExvY2tlZEZhaWxlZBADEhsKF1VzZXJEYXRhYmFzZURpZE5vdE1hdGNoEAQSHQoZVXNlckRhdGFiYXNlSW50ZXJuYWxFcnJvchAFEhQKEFNxbEludGVybmFsRXJyb3IQBhIYChREYXRhYmFzZUNvbm5lY3RFcnJvchAHEhMKD1VzZXJOb3RMb2dpbllldBAKEhcKE1JlYWRDdXJyZW50SWRGYWlsZWQQCxIYChRXcml0ZUN1cnJlbnRJZEZhaWxlZBAMEhAKDEVtYWlsSXNFbXB0eRAUEhYKEkVtYWlsRm9ybWF0SW52YWxpZBAVEhYKEkVtYWlsQWxyZWFkeUV4aXN0cxAWEhMKD1Bhc3N3b3JkSXNFbXB0eRAeEhMKD1Bhc3N3b3JkVG9vTG9uZxAfEiQKIFBhc3N3b3JkQ29udGFpbnNGb3JiaWRDaGFyYWN0ZXJzECASGQoVUGFzc3dvcmRGb3JtYXRJbnZhbGlkECESFAoQUGFzc3dvcmROb3RNYXRjaBAiEhMKD1VzZXJOYW1lVG9vTG9uZxAoEicKI1VzZXJOYW1lQ29udGFpbnNGb3JiaWRkZW5DaGFyYWN0ZXJzECkSEwoPVXNlck5hbWVJc0VtcHR5ECoSGAoUVXNlcldvcmtzcGFjZUludmFsaWQQMhIRCg1Vc2VySWRJbnZhbGlkEDMSIAocQ3JlYXRlRGVmYXVsdFdvcmtzcGFjZUZhaWxlZBA0EiAKHERlZmF1bHRXb3Jrc3BhY2VBbHJlYWR5RXhpc3QQNRIPCgtTZXJ2ZXJFcnJvchBk');
+final $typed_data.Uint8List errorCodeDescriptor = $convert.base64Decode('CglFcnJvckNvZGUSCwoHVW5rbm93bhAAEhoKFlVzZXJEYXRhYmFzZUluaXRGYWlsZWQQARIcChhBY3F1aXJlV3JpdGVMb2NrZWRGYWlsZWQQAhIbChdBY3F1aXJlUmVhZExvY2tlZEZhaWxlZBADEhsKF1VzZXJEYXRhYmFzZURpZE5vdE1hdGNoEAQSHQoZVXNlckRhdGFiYXNlSW50ZXJuYWxFcnJvchAFEhQKEFNxbEludGVybmFsRXJyb3IQBhIYChREYXRhYmFzZUNvbm5lY3RFcnJvchAHEhMKD1VzZXJOb3RMb2dpbllldBAKEhcKE1JlYWRDdXJyZW50SWRGYWlsZWQQCxIYChRXcml0ZUN1cnJlbnRJZEZhaWxlZBAMEhAKDEVtYWlsSXNFbXB0eRAUEhYKEkVtYWlsRm9ybWF0SW52YWxpZBAVEhYKEkVtYWlsQWxyZWFkeUV4aXN0cxAWEhMKD1Bhc3N3b3JkSXNFbXB0eRAeEhMKD1Bhc3N3b3JkVG9vTG9uZxAfEiQKIFBhc3N3b3JkQ29udGFpbnNGb3JiaWRDaGFyYWN0ZXJzECASGQoVUGFzc3dvcmRGb3JtYXRJbnZhbGlkECESFAoQUGFzc3dvcmROb3RNYXRjaBAiEhMKD1VzZXJOYW1lVG9vTG9uZxAoEicKI1VzZXJOYW1lQ29udGFpbnNGb3JiaWRkZW5DaGFyYWN0ZXJzECkSEwoPVXNlck5hbWVJc0VtcHR5ECoSGAoUVXNlcldvcmtzcGFjZUludmFsaWQQMhIRCg1Vc2VySWRJbnZhbGlkEDMSFAoQVXNlclRva2VuSW52YWxpZBA2EhAKDFVzZXJOb3RFeGlzdBA3EiAKHENyZWF0ZURlZmF1bHRXb3Jrc3BhY2VGYWlsZWQQPBIgChxEZWZhdWx0V29ya3NwYWNlQWxyZWFkeUV4aXN0ED0SDwoLU2VydmVyRXJyb3IQZA==');
 @$core.Deprecated('Use userErrorDescriptor instead')
 const UserError$json = const {
   '1': 'UserError',

app_flowy/packages/flowy_sdk/lib/protobuf/flowy-user/observable.pb.dart

@@ -0,0 +1,11 @@
+///
+//  Generated code. Do not modify.
+//  source: observable.proto
+//
+// @dart = 2.12
+// ignore_for_file: annotate_overrides,camel_case_types,unnecessary_const,non_constant_identifier_names,library_prefixes,unused_import,unused_shown_name,return_of_invalid_type,unnecessary_this,prefer_final_fields
+
+import 'dart:core' as $core;
+
+export 'observable.pbenum.dart';
+

app_flowy/packages/flowy_sdk/lib/protobuf/flowy-user/observable.pbenum.dart

@@ -0,0 +1,28 @@
+///
+//  Generated code. Do not modify.
+//  source: observable.proto
+//
+// @dart = 2.12
+// ignore_for_file: annotate_overrides,camel_case_types,unnecessary_const,non_constant_identifier_names,library_prefixes,unused_import,unused_shown_name,return_of_invalid_type,unnecessary_this,prefer_final_fields
+
+// ignore_for_file: UNDEFINED_SHOWN_NAME
+import 'dart:core' as $core;
+import 'package:protobuf/protobuf.dart' as $pb;
+
+class UserObservable extends $pb.ProtobufEnum {
+  static const UserObservable Unknown = UserObservable._(0, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'Unknown');
+  static const UserObservable UserAuthChanged = UserObservable._(1, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'UserAuthChanged');
+  static const UserObservable UserProfileUpdated = UserObservable._(2, const $core.bool.fromEnvironment('protobuf.omit_enum_names') ? '' : 'UserProfileUpdated');
+
+  static const $core.List<UserObservable> values = <UserObservable> [
+    Unknown,
+    UserAuthChanged,
+    UserProfileUpdated,
+  ];
+
+  static final $core.Map<$core.int, UserObservable> _byValue = $pb.ProtobufEnum.initByValue(values);
+  static UserObservable? valueOf($core.int value) => _byValue[value];
+
+  const UserObservable._($core.int v, $core.String n) : super(v, n);
+}
+

app_flowy/packages/flowy_sdk/lib/protobuf/flowy-user/observable.pbjson.dart

@@ -0,0 +1,22 @@
+///
+//  Generated code. Do not modify.
+//  source: observable.proto
+//
+// @dart = 2.12
+// ignore_for_file: annotate_overrides,camel_case_types,unnecessary_const,non_constant_identifier_names,library_prefixes,unused_import,unused_shown_name,return_of_invalid_type,unnecessary_this,prefer_final_fields,deprecated_member_use_from_same_package
+
+import 'dart:core' as $core;
+import 'dart:convert' as $convert;
+import 'dart:typed_data' as $typed_data;
+@$core.Deprecated('Use userObservableDescriptor instead')
+const UserObservable$json = const {
+  '1': 'UserObservable',
+  '2': const [
+    const {'1': 'Unknown', '2': 0},
+    const {'1': 'UserAuthChanged', '2': 1},
+    const {'1': 'UserProfileUpdated', '2': 2},
+  ],
+};
+
+/// Descriptor for `UserObservable`. Decode as a `google.protobuf.EnumDescriptorProto`.
+final $typed_data.Uint8List userObservableDescriptor = $convert.base64Decode('Cg5Vc2VyT2JzZXJ2YWJsZRILCgdVbmtub3duEAASEwoPVXNlckF1dGhDaGFuZ2VkEAESFgoSVXNlclByb2ZpbGVVcGRhdGVkEAI=');

app_flowy/packages/flowy_sdk/lib/protobuf/flowy-user/observable.pbserver.dart

@@ -0,0 +1,9 @@
+///
+//  Generated code. Do not modify.
+//  source: observable.proto
+//
+// @dart = 2.12
+// ignore_for_file: annotate_overrides,camel_case_types,unnecessary_const,non_constant_identifier_names,library_prefixes,unused_import,unused_shown_name,return_of_invalid_type,unnecessary_this,prefer_final_fields,deprecated_member_use_from_same_package
+
+export 'observable.pb.dart';
+

app_flowy/packages/flowy_sdk/lib/protobuf/flowy-user/protobuf.dart

@@ -1,4 +1,5 @@
 // Auto-generated, do not edit 
+export './observable.pb.dart';
 export './user_table.pb.dart';
 export './errors.pb.dart';
 export './user_profile.pb.dart';

backend/src/workspace_service/app/app.rs

@@ -2,7 +2,7 @@ use flowy_net::{errors::ServerError, response::FlowyResponse};
 
 use crate::{
     entities::workspace::AppTable,
-    sqlx_ext::{map_sqlx_error, DBTransaction, SqlBuilder},
+    sqlx_ext::{map_sqlx_error, SqlBuilder},
     user_service::LoggedUser,
     workspace_service::{
         app::{check_app_id, make_app_from_table, Builder},
@@ -18,7 +18,7 @@ use flowy_workspace::{
         app::parser::{AppDesc, AppName},
         workspace::parser::WorkspaceId,
     },
-    protobuf::{App, CreateAppParams, QueryAppParams, RepeatedApp, RepeatedView, UpdateAppParams},
+    protobuf::{CreateAppParams, QueryAppParams, RepeatedApp, RepeatedView, UpdateAppParams},
 };
 use protobuf::Message;
 use sqlx::{postgres::PgArguments, PgPool, Postgres};
@@ -176,29 +176,3 @@ pub(crate) async fn delete_app(pool: &PgPool, app_id: &str) -> Result<FlowyRespo
 
     Ok(FlowyResponse::success())
 }
-
-// transaction must be commit from caller
-pub(crate) async fn read_apps_belong_to_workspace<'c>(
-    transaction: &mut DBTransaction<'_>,
-    workspace_id: &str,
-) -> Result<RepeatedApp, ServerError> {
-    let workspace_id = WorkspaceId::parse(workspace_id.to_owned()).map_err(invalid_params)?;
-    let (sql, args) = SqlBuilder::select("app_table")
-        .add_field("*")
-        .and_where_eq("workspace_id", workspace_id.0)
-        .build()?;
-
-    let tables = sqlx::query_as_with::<Postgres, AppTable, PgArguments>(&sql, args)
-        .fetch_all(transaction)
-        .await
-        .map_err(map_sqlx_error)?;
-
-    let apps = tables
-        .into_iter()
-        .map(|table| make_app_from_table(table, RepeatedView::default()))
-        .collect::<Vec<App>>();
-
-    let mut repeated_app = RepeatedApp::default();
-    repeated_app.set_items(apps.into());
-    Ok(repeated_app)
-}

backend/src/workspace_service/workspace/workspace.rs

@@ -1,9 +1,5 @@
 use super::builder::Builder;
-use crate::{
-    entities::workspace::WorkspaceTable,
-    sqlx_ext::*,
-    workspace_service::app::app::read_apps_belong_to_workspace,
-};
+use crate::{entities::workspace::WorkspaceTable, sqlx_ext::*};
 use anyhow::Context;
 
 use flowy_net::{
@@ -12,12 +8,24 @@ use flowy_net::{
 };
 
 use crate::{
+    entities::workspace::AppTable,
     user_service::LoggedUser,
-    workspace_service::workspace::{check_workspace_id, make_workspace_from_table},
+    workspace_service::{
+        app::make_app_from_table,
+        view::read_views_belong_to_id,
+        workspace::{check_workspace_id, make_workspace_from_table},
+    },
 };
 use flowy_workspace::{
-    entities::workspace::parser::{WorkspaceDesc, WorkspaceName},
-    protobuf::{CreateWorkspaceParams, RepeatedApp, RepeatedWorkspace, UpdateWorkspaceParams},
+    entities::workspace::parser::{WorkspaceDesc, WorkspaceId, WorkspaceName},
+    protobuf::{
+        App,
+        CreateWorkspaceParams,
+        RepeatedApp,
+        RepeatedView,
+        RepeatedWorkspace,
+        UpdateWorkspaceParams,
+    },
 };
 use sqlx::{postgres::PgArguments, PgPool, Postgres};
 
@@ -157,11 +165,18 @@ pub async fn read_workspaces(
 
     let mut repeated_workspace = RepeatedWorkspace::default();
     let mut workspaces = vec![];
+    // Opti: combine the query
     for table in tables {
-        let apps = read_apps_belong_to_workspace(&mut transaction, &table.id.to_string())
+        let mut apps = read_apps_belong_to_workspace(&mut transaction, &table.id.to_string())
             .await
             .context("Get workspace app")
             .unwrap_or(RepeatedApp::default());
+
+        for app in &mut apps.items {
+            let views = read_views_belong_to_id(&mut transaction, &app.id).await?;
+            app.mut_belongings().set_items(views.into());
+        }
+
         let workspace = make_workspace_from_table(table, Some(apps));
         workspaces.push(workspace);
     }
@@ -171,6 +186,32 @@ pub async fn read_workspaces(
         .context("Failed to commit SQL transaction to read workspace.")?;
 
     repeated_workspace.set_items(workspaces.into());
-
     FlowyResponse::success().pb(repeated_workspace)
 }
+
+// transaction must be commit from caller
+async fn read_apps_belong_to_workspace<'c>(
+    transaction: &mut DBTransaction<'_>,
+    workspace_id: &str,
+) -> Result<RepeatedApp, ServerError> {
+    let transaction = transaction;
+    let workspace_id = WorkspaceId::parse(workspace_id.to_owned()).map_err(invalid_params)?;
+    let (sql, args) = SqlBuilder::select("app_table")
+        .add_field("*")
+        .and_where_eq("workspace_id", workspace_id.0)
+        .build()?;
+
+    let tables = sqlx::query_as_with::<Postgres, AppTable, PgArguments>(&sql, args)
+        .fetch_all(transaction)
+        .await
+        .map_err(map_sqlx_error)?;
+
+    let apps = tables
+        .into_iter()
+        .map(|table| make_app_from_table(table, RepeatedView::default()))
+        .collect::<Vec<App>>();
+
+    let mut repeated_app = RepeatedApp::default();
+    repeated_app.set_items(apps.into());
+    Ok(repeated_app)
+}

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

@@ -80,6 +80,7 @@ pub fn category_from_str(type_str: &str) -> TypeCategory {
         | "FFIStatusCode"
         | "UserStatus"
         | "UserEvent"
+        | "UserObservable"
         => TypeCategory::Enum,
 
         "Option" => TypeCategory::Opt,

rust-lib/flowy-observable/Cargo.toml

@@ -13,6 +13,7 @@ log = "0.4.14"
 bytes = { version = "1.0" }
 
 flowy-derive = {path = "../flowy-derive"}
+flowy-dispatch = {path = "../flowy-dispatch"}
 
 [features]
 dart = []

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

@@ -1,11 +1,79 @@
+use bytes::Bytes;
+
 pub mod dart;
 pub mod entities;
 mod protobuf;
 
-#[cfg(test)]
-mod tests {
-    #[test]
-    fn it_works() {
-        assert_eq!(2 + 2, 4);
+use crate::{dart::RustStreamSender, entities::ObservableSubject};
+use flowy_dispatch::prelude::ToBytes;
+
+pub struct ObservableBuilder {
+    id: String,
+    payload: Option<Bytes>,
+    error: Option<Bytes>,
+    ty: i32,
+    category: String,
+}
+
+impl ObservableBuilder {
+    pub fn new<T: Into<i32>>(id: &str, ty: T, category: &str) -> Self {
+        Self {
+            id: id.to_owned(),
+            ty: ty.into(),
+            payload: None,
+            error: None,
+            category: category.to_owned(),
+        }
+    }
+
+    pub fn payload<T>(mut self, payload: T) -> Self
+    where
+        T: ToBytes,
+    {
+        match payload.into_bytes() {
+            Ok(bytes) => self.payload = Some(bytes),
+            Err(e) => {
+                log::error!("Set observable payload failed: {:?}", e);
+            },
+        }
+
+        self
+    }
+
+    pub fn error<T>(mut self, error: T) -> Self
+    where
+        T: ToBytes,
+    {
+        match error.into_bytes() {
+            Ok(bytes) => self.error = Some(bytes),
+            Err(e) => {
+                log::error!("Set observable error failed: {:?}", e);
+            },
+        }
+        self
+    }
+
+    pub fn build(self) {
+        let payload = match self.payload {
+            None => None,
+            Some(bytes) => Some(bytes.to_vec()),
+        };
+
+        let error = match self.error {
+            None => None,
+            Some(bytes) => Some(bytes.to_vec()),
+        };
+
+        let subject = ObservableSubject {
+            category: self.category,
+            ty: self.ty,
+            id: self.id,
+            payload,
+            error,
+        };
+        match RustStreamSender::post(subject) {
+            Ok(_) => {},
+            Err(error) => log::error!("Send observable subject failed: {}", error),
+        }
     }
 }

rust-lib/flowy-user/Cargo.toml

@@ -14,6 +14,7 @@ flowy-database = { path = "../flowy-database" }
 flowy-sqlite = { path = "../flowy-sqlite" }
 flowy-infra = { path = "../flowy-infra" }
 flowy-net = { path = "../flowy-net" }
+flowy-observable = { path = "../flowy-observable" }
 
 tracing = { version = "0.1", features = ["log"] }
 bytes = "1.0"

rust-lib/flowy-user/Flowy.toml

@@ -1,3 +1,3 @@
 
-proto_crates = ["src/entities", "src/event.rs", "src/errors.rs"]
+proto_crates = ["src/entities", "src/event.rs", "src/errors.rs", "src/observable"]
 event_files = ["src/event.rs"]

rust-lib/flowy-user/src/errors.rs

@@ -76,11 +76,16 @@ pub enum ErrorCode {
     UserWorkspaceInvalid = 50,
     #[display(fmt = "User id is invalid")]
     UserIdInvalid        = 51,
+    #[display(fmt = "User token is invalid")]
+    UserTokenInvalid     = 54,
+    #[display(fmt = "User not exist")]
+    UserNotExist         = 55,
+
     #[display(fmt = "Create user default workspace failed")]
-    CreateDefaultWorkspaceFailed = 52,
+    CreateDefaultWorkspaceFailed = 60,
 
     #[display(fmt = "User default workspace already exists")]
-    DefaultWorkspaceAlreadyExist = 53,
+    DefaultWorkspaceAlreadyExist = 61,
 
     #[display(fmt = "Server error")]
     ServerError          = 100,
@@ -109,45 +114,37 @@ impl std::convert::From<::r2d2::Error> for UserError {
 // use diesel::result::{Error, DatabaseErrorKind};
 // use flowy_sqlite::ErrorKind;
 impl std::convert::From<flowy_sqlite::Error> for UserError {
-    fn from(error: flowy_sqlite::Error) -> Self {
-        // match error.kind() {
-        //     ErrorKind::Msg(_) => {},
-        //     ErrorKind::R2D2(_) => {},
-        //     ErrorKind::Migrations(_) => {},
-        //     ErrorKind::Diesel(diesel_err) => match diesel_err {
-        //         Error::InvalidCString(_) => {},
-        //         Error::DatabaseError(kind, _) => {
-        //             match kind {
-        //                 DatabaseErrorKind::UniqueViolation => {
-        //
-        //                 }
-        //                 _ => {}
-        //             }
-        //         },
-        //         Error::NotFound => {},
-        //         Error::QueryBuilderError(_) => {},
-        //         Error::DeserializationError(_) => {},
-        //         Error::SerializationError(_) => {},
-        //         Error::RollbackTransaction => {},
-        //         Error::AlreadyInTransaction => {},
-        //         Error::__Nonexhaustive => {},
-        //     },
-        //     ErrorKind::Connection(_) => {},
-        //     ErrorKind::Io(_) => {},
-        //     ErrorKind::UnknownMigrationExists(_) => {},
-        //     ErrorKind::__Nonexhaustive { .. } => {},
-        // }
-
-        ErrorBuilder::new(ErrorCode::SqlInternalError).error(error).build()
-    }
+    fn from(error: flowy_sqlite::Error) -> Self { ErrorBuilder::new(ErrorCode::SqlInternalError).error(error).build() }
 }
 
 impl std::convert::From<flowy_net::errors::ServerError> for UserError {
     fn from(error: flowy_net::errors::ServerError) -> Self {
-        match error.code {
-            flowy_net::errors::ErrorCode::PasswordNotMatch => ErrorBuilder::new(ErrorCode::PasswordNotMatch).error(error.msg).build(),
-            _ => ErrorBuilder::new(ErrorCode::ServerError).error(error.msg).build(),
-        }
+        let code = server_error_to_user_error(error.code);
+        ErrorBuilder::new(code).error(error.msg).build()
+    }
+}
+
+use flowy_net::errors::ErrorCode as ServerErrorCode;
+fn server_error_to_user_error(code: ServerErrorCode) -> ErrorCode {
+    match code {
+        ServerErrorCode::InvalidToken => ErrorCode::UserTokenInvalid,
+        ServerErrorCode::Unauthorized => ErrorCode::UserTokenInvalid,
+        ServerErrorCode::PayloadOverflow => ErrorCode::ServerError,
+        ServerErrorCode::PayloadSerdeFail => ErrorCode::ServerError,
+        ServerErrorCode::PayloadUnexpectedNone => ErrorCode::ServerError,
+        ServerErrorCode::ParamsInvalid => ErrorCode::ServerError,
+        ServerErrorCode::ProtobufError => ErrorCode::ServerError,
+        ServerErrorCode::SerdeError => ErrorCode::ServerError,
+        ServerErrorCode::EmailAlreadyExists => ErrorCode::ServerError,
+        ServerErrorCode::PasswordNotMatch => ErrorCode::PasswordNotMatch,
+        ServerErrorCode::ConnectRefused => ErrorCode::ServerError,
+        ServerErrorCode::ConnectTimeout => ErrorCode::ServerError,
+        ServerErrorCode::ConnectClose => ErrorCode::ServerError,
+        ServerErrorCode::ConnectCancel => ErrorCode::ServerError,
+        ServerErrorCode::SqlError => ErrorCode::ServerError,
+        ServerErrorCode::RecordNotFound => ErrorCode::UserNotExist,
+        ServerErrorCode::HttpError => ErrorCode::ServerError,
+        ServerErrorCode::InternalError => ErrorCode::ServerError,
     }
 }
 

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

@@ -5,6 +5,7 @@ pub mod entities;
 pub mod errors;
 pub mod event;
 pub mod module;
+mod observable;
 pub mod protobuf;
 pub mod services;
 

rust-lib/flowy-user/src/observable/mod.rs

@@ -0,0 +1,2 @@
+mod observable;
+pub use observable::*;

rust-lib/flowy-user/src/observable/observable.rs

@@ -0,0 +1,23 @@
+use bytes::Bytes;
+use flowy_derive::ProtoBuf_Enum;
+use flowy_dispatch::prelude::ToBytes;
+use flowy_observable::{dart::RustStreamSender, entities::ObservableSubject, ObservableBuilder};
+
+const OBSERVABLE_CATEGORY: &'static str = "User";
+
+#[derive(ProtoBuf_Enum, Debug)]
+pub(crate) enum UserObservable {
+    Unknown            = 0,
+    UserAuthChanged    = 1,
+    UserProfileUpdated = 2,
+}
+
+impl std::default::Default for UserObservable {
+    fn default() -> Self { UserObservable::Unknown }
+}
+
+impl std::convert::Into<i32> for UserObservable {
+    fn into(self) -> i32 { self as i32 }
+}
+
+pub(crate) fn observable(id: &str, ty: UserObservable) -> ObservableBuilder { ObservableBuilder::new(id, ty, OBSERVABLE_CATEGORY) }

rust-lib/flowy-user/src/protobuf/model/errors.rs

@@ -239,8 +239,10 @@ pub enum ErrorCode {
     UserNameIsEmpty = 42,
     UserWorkspaceInvalid = 50,
     UserIdInvalid = 51,
-    CreateDefaultWorkspaceFailed = 52,
-    DefaultWorkspaceAlreadyExist = 53,
+    UserTokenInvalid = 54,
+    UserNotExist = 55,
+    CreateDefaultWorkspaceFailed = 60,
+    DefaultWorkspaceAlreadyExist = 61,
     ServerError = 100,
 }
 
@@ -275,8 +277,10 @@ impl ::protobuf::ProtobufEnum for ErrorCode {
             42 => ::std::option::Option::Some(ErrorCode::UserNameIsEmpty),
             50 => ::std::option::Option::Some(ErrorCode::UserWorkspaceInvalid),
             51 => ::std::option::Option::Some(ErrorCode::UserIdInvalid),
-            52 => ::std::option::Option::Some(ErrorCode::CreateDefaultWorkspaceFailed),
-            53 => ::std::option::Option::Some(ErrorCode::DefaultWorkspaceAlreadyExist),
+            54 => ::std::option::Option::Some(ErrorCode::UserTokenInvalid),
+            55 => ::std::option::Option::Some(ErrorCode::UserNotExist),
+            60 => ::std::option::Option::Some(ErrorCode::CreateDefaultWorkspaceFailed),
+            61 => ::std::option::Option::Some(ErrorCode::DefaultWorkspaceAlreadyExist),
             100 => ::std::option::Option::Some(ErrorCode::ServerError),
             _ => ::std::option::Option::None
         }
@@ -308,6 +312,8 @@ impl ::protobuf::ProtobufEnum for ErrorCode {
             ErrorCode::UserNameIsEmpty,
             ErrorCode::UserWorkspaceInvalid,
             ErrorCode::UserIdInvalid,
+            ErrorCode::UserTokenInvalid,
+            ErrorCode::UserNotExist,
             ErrorCode::CreateDefaultWorkspaceFailed,
             ErrorCode::DefaultWorkspaceAlreadyExist,
             ErrorCode::ServerError,
@@ -341,7 +347,7 @@ impl ::protobuf::reflect::ProtobufValue for ErrorCode {
 static file_descriptor_proto_data: &'static [u8] = b"\
     \n\x0cerrors.proto\"=\n\tUserError\x12\x1e\n\x04code\x18\x01\x20\x01(\
     \x0e2\n.ErrorCodeR\x04code\x12\x10\n\x03msg\x18\x02\x20\x01(\tR\x03msg*\
-    \xbb\x05\n\tErrorCode\x12\x0b\n\x07Unknown\x10\0\x12\x1a\n\x16UserDataba\
+    \xe3\x05\n\tErrorCode\x12\x0b\n\x07Unknown\x10\0\x12\x1a\n\x16UserDataba\
     seInitFailed\x10\x01\x12\x1c\n\x18AcquireWriteLockedFailed\x10\x02\x12\
     \x1b\n\x17AcquireReadLockedFailed\x10\x03\x12\x1b\n\x17UserDatabaseDidNo\
     tMatch\x10\x04\x12\x1d\n\x19UserDatabaseInternalError\x10\x05\x12\x14\n\
@@ -354,50 +360,51 @@ static file_descriptor_proto_data: &'static [u8] = b"\
     \x12\x19\n\x15PasswordFormatInvalid\x10!\x12\x14\n\x10PasswordNotMatch\
     \x10\"\x12\x13\n\x0fUserNameTooLong\x10(\x12'\n#UserNameContainsForbidde\
     nCharacters\x10)\x12\x13\n\x0fUserNameIsEmpty\x10*\x12\x18\n\x14UserWork\
-    spaceInvalid\x102\x12\x11\n\rUserIdInvalid\x103\x12\x20\n\x1cCreateDefau\
-    ltWorkspaceFailed\x104\x12\x20\n\x1cDefaultWorkspaceAlreadyExist\x105\
-    \x12\x0f\n\x0bServerError\x10dJ\x83\n\n\x06\x12\x04\0\0\"\x01\n\x08\n\
-    \x01\x0c\x12\x03\0\0\x12\n\n\n\x02\x04\0\x12\x04\x02\0\x05\x01\n\n\n\x03\
-    \x04\0\x01\x12\x03\x02\x08\x11\n\x0b\n\x04\x04\0\x02\0\x12\x03\x03\x04\
-    \x17\n\x0c\n\x05\x04\0\x02\0\x06\x12\x03\x03\x04\r\n\x0c\n\x05\x04\0\x02\
-    \0\x01\x12\x03\x03\x0e\x12\n\x0c\n\x05\x04\0\x02\0\x03\x12\x03\x03\x15\
-    \x16\n\x0b\n\x04\x04\0\x02\x01\x12\x03\x04\x04\x13\n\x0c\n\x05\x04\0\x02\
-    \x01\x05\x12\x03\x04\x04\n\n\x0c\n\x05\x04\0\x02\x01\x01\x12\x03\x04\x0b\
-    \x0e\n\x0c\n\x05\x04\0\x02\x01\x03\x12\x03\x04\x11\x12\n\n\n\x02\x05\0\
-    \x12\x04\x06\0\"\x01\n\n\n\x03\x05\0\x01\x12\x03\x06\x05\x0e\n\x0b\n\x04\
-    \x05\0\x02\0\x12\x03\x07\x04\x10\n\x0c\n\x05\x05\0\x02\0\x01\x12\x03\x07\
-    \x04\x0b\n\x0c\n\x05\x05\0\x02\0\x02\x12\x03\x07\x0e\x0f\n\x0b\n\x04\x05\
-    \0\x02\x01\x12\x03\x08\x04\x1f\n\x0c\n\x05\x05\0\x02\x01\x01\x12\x03\x08\
-    \x04\x1a\n\x0c\n\x05\x05\0\x02\x01\x02\x12\x03\x08\x1d\x1e\n\x0b\n\x04\
-    \x05\0\x02\x02\x12\x03\t\x04!\n\x0c\n\x05\x05\0\x02\x02\x01\x12\x03\t\
-    \x04\x1c\n\x0c\n\x05\x05\0\x02\x02\x02\x12\x03\t\x1f\x20\n\x0b\n\x04\x05\
-    \0\x02\x03\x12\x03\n\x04\x20\n\x0c\n\x05\x05\0\x02\x03\x01\x12\x03\n\x04\
-    \x1b\n\x0c\n\x05\x05\0\x02\x03\x02\x12\x03\n\x1e\x1f\n\x0b\n\x04\x05\0\
-    \x02\x04\x12\x03\x0b\x04\x20\n\x0c\n\x05\x05\0\x02\x04\x01\x12\x03\x0b\
-    \x04\x1b\n\x0c\n\x05\x05\0\x02\x04\x02\x12\x03\x0b\x1e\x1f\n\x0b\n\x04\
-    \x05\0\x02\x05\x12\x03\x0c\x04\"\n\x0c\n\x05\x05\0\x02\x05\x01\x12\x03\
-    \x0c\x04\x1d\n\x0c\n\x05\x05\0\x02\x05\x02\x12\x03\x0c\x20!\n\x0b\n\x04\
-    \x05\0\x02\x06\x12\x03\r\x04\x19\n\x0c\n\x05\x05\0\x02\x06\x01\x12\x03\r\
-    \x04\x14\n\x0c\n\x05\x05\0\x02\x06\x02\x12\x03\r\x17\x18\n\x0b\n\x04\x05\
-    \0\x02\x07\x12\x03\x0e\x04\x1d\n\x0c\n\x05\x05\0\x02\x07\x01\x12\x03\x0e\
-    \x04\x18\n\x0c\n\x05\x05\0\x02\x07\x02\x12\x03\x0e\x1b\x1c\n\x0b\n\x04\
-    \x05\0\x02\x08\x12\x03\x0f\x04\x19\n\x0c\n\x05\x05\0\x02\x08\x01\x12\x03\
-    \x0f\x04\x13\n\x0c\n\x05\x05\0\x02\x08\x02\x12\x03\x0f\x16\x18\n\x0b\n\
-    \x04\x05\0\x02\t\x12\x03\x10\x04\x1d\n\x0c\n\x05\x05\0\x02\t\x01\x12\x03\
-    \x10\x04\x17\n\x0c\n\x05\x05\0\x02\t\x02\x12\x03\x10\x1a\x1c\n\x0b\n\x04\
-    \x05\0\x02\n\x12\x03\x11\x04\x1e\n\x0c\n\x05\x05\0\x02\n\x01\x12\x03\x11\
-    \x04\x18\n\x0c\n\x05\x05\0\x02\n\x02\x12\x03\x11\x1b\x1d\n\x0b\n\x04\x05\
-    \0\x02\x0b\x12\x03\x12\x04\x16\n\x0c\n\x05\x05\0\x02\x0b\x01\x12\x03\x12\
-    \x04\x10\n\x0c\n\x05\x05\0\x02\x0b\x02\x12\x03\x12\x13\x15\n\x0b\n\x04\
-    \x05\0\x02\x0c\x12\x03\x13\x04\x1c\n\x0c\n\x05\x05\0\x02\x0c\x01\x12\x03\
-    \x13\x04\x16\n\x0c\n\x05\x05\0\x02\x0c\x02\x12\x03\x13\x19\x1b\n\x0b\n\
-    \x04\x05\0\x02\r\x12\x03\x14\x04\x1c\n\x0c\n\x05\x05\0\x02\r\x01\x12\x03\
-    \x14\x04\x16\n\x0c\n\x05\x05\0\x02\r\x02\x12\x03\x14\x19\x1b\n\x0b\n\x04\
-    \x05\0\x02\x0e\x12\x03\x15\x04\x19\n\x0c\n\x05\x05\0\x02\x0e\x01\x12\x03\
-    \x15\x04\x13\n\x0c\n\x05\x05\0\x02\x0e\x02\x12\x03\x15\x16\x18\n\x0b\n\
-    \x04\x05\0\x02\x0f\x12\x03\x16\x04\x19\n\x0c\n\x05\x05\0\x02\x0f\x01\x12\
-    \x03\x16\x04\x13\n\x0c\n\x05\x05\0\x02\x0f\x02\x12\x03\x16\x16\x18\n\x0b\
-    \n\x04\x05\0\x02\x10\x12\x03\x17\x04*\n\x0c\n\x05\x05\0\x02\x10\x01\x12\
+    spaceInvalid\x102\x12\x11\n\rUserIdInvalid\x103\x12\x14\n\x10UserTokenIn\
+    valid\x106\x12\x10\n\x0cUserNotExist\x107\x12\x20\n\x1cCreateDefaultWork\
+    spaceFailed\x10<\x12\x20\n\x1cDefaultWorkspaceAlreadyExist\x10=\x12\x0f\
+    \n\x0bServerError\x10dJ\xd5\n\n\x06\x12\x04\0\0$\x01\n\x08\n\x01\x0c\x12\
+    \x03\0\0\x12\n\n\n\x02\x04\0\x12\x04\x02\0\x05\x01\n\n\n\x03\x04\0\x01\
+    \x12\x03\x02\x08\x11\n\x0b\n\x04\x04\0\x02\0\x12\x03\x03\x04\x17\n\x0c\n\
+    \x05\x04\0\x02\0\x06\x12\x03\x03\x04\r\n\x0c\n\x05\x04\0\x02\0\x01\x12\
+    \x03\x03\x0e\x12\n\x0c\n\x05\x04\0\x02\0\x03\x12\x03\x03\x15\x16\n\x0b\n\
+    \x04\x04\0\x02\x01\x12\x03\x04\x04\x13\n\x0c\n\x05\x04\0\x02\x01\x05\x12\
+    \x03\x04\x04\n\n\x0c\n\x05\x04\0\x02\x01\x01\x12\x03\x04\x0b\x0e\n\x0c\n\
+    \x05\x04\0\x02\x01\x03\x12\x03\x04\x11\x12\n\n\n\x02\x05\0\x12\x04\x06\0\
+    $\x01\n\n\n\x03\x05\0\x01\x12\x03\x06\x05\x0e\n\x0b\n\x04\x05\0\x02\0\
+    \x12\x03\x07\x04\x10\n\x0c\n\x05\x05\0\x02\0\x01\x12\x03\x07\x04\x0b\n\
+    \x0c\n\x05\x05\0\x02\0\x02\x12\x03\x07\x0e\x0f\n\x0b\n\x04\x05\0\x02\x01\
+    \x12\x03\x08\x04\x1f\n\x0c\n\x05\x05\0\x02\x01\x01\x12\x03\x08\x04\x1a\n\
+    \x0c\n\x05\x05\0\x02\x01\x02\x12\x03\x08\x1d\x1e\n\x0b\n\x04\x05\0\x02\
+    \x02\x12\x03\t\x04!\n\x0c\n\x05\x05\0\x02\x02\x01\x12\x03\t\x04\x1c\n\
+    \x0c\n\x05\x05\0\x02\x02\x02\x12\x03\t\x1f\x20\n\x0b\n\x04\x05\0\x02\x03\
+    \x12\x03\n\x04\x20\n\x0c\n\x05\x05\0\x02\x03\x01\x12\x03\n\x04\x1b\n\x0c\
+    \n\x05\x05\0\x02\x03\x02\x12\x03\n\x1e\x1f\n\x0b\n\x04\x05\0\x02\x04\x12\
+    \x03\x0b\x04\x20\n\x0c\n\x05\x05\0\x02\x04\x01\x12\x03\x0b\x04\x1b\n\x0c\
+    \n\x05\x05\0\x02\x04\x02\x12\x03\x0b\x1e\x1f\n\x0b\n\x04\x05\0\x02\x05\
+    \x12\x03\x0c\x04\"\n\x0c\n\x05\x05\0\x02\x05\x01\x12\x03\x0c\x04\x1d\n\
+    \x0c\n\x05\x05\0\x02\x05\x02\x12\x03\x0c\x20!\n\x0b\n\x04\x05\0\x02\x06\
+    \x12\x03\r\x04\x19\n\x0c\n\x05\x05\0\x02\x06\x01\x12\x03\r\x04\x14\n\x0c\
+    \n\x05\x05\0\x02\x06\x02\x12\x03\r\x17\x18\n\x0b\n\x04\x05\0\x02\x07\x12\
+    \x03\x0e\x04\x1d\n\x0c\n\x05\x05\0\x02\x07\x01\x12\x03\x0e\x04\x18\n\x0c\
+    \n\x05\x05\0\x02\x07\x02\x12\x03\x0e\x1b\x1c\n\x0b\n\x04\x05\0\x02\x08\
+    \x12\x03\x0f\x04\x19\n\x0c\n\x05\x05\0\x02\x08\x01\x12\x03\x0f\x04\x13\n\
+    \x0c\n\x05\x05\0\x02\x08\x02\x12\x03\x0f\x16\x18\n\x0b\n\x04\x05\0\x02\t\
+    \x12\x03\x10\x04\x1d\n\x0c\n\x05\x05\0\x02\t\x01\x12\x03\x10\x04\x17\n\
+    \x0c\n\x05\x05\0\x02\t\x02\x12\x03\x10\x1a\x1c\n\x0b\n\x04\x05\0\x02\n\
+    \x12\x03\x11\x04\x1e\n\x0c\n\x05\x05\0\x02\n\x01\x12\x03\x11\x04\x18\n\
+    \x0c\n\x05\x05\0\x02\n\x02\x12\x03\x11\x1b\x1d\n\x0b\n\x04\x05\0\x02\x0b\
+    \x12\x03\x12\x04\x16\n\x0c\n\x05\x05\0\x02\x0b\x01\x12\x03\x12\x04\x10\n\
+    \x0c\n\x05\x05\0\x02\x0b\x02\x12\x03\x12\x13\x15\n\x0b\n\x04\x05\0\x02\
+    \x0c\x12\x03\x13\x04\x1c\n\x0c\n\x05\x05\0\x02\x0c\x01\x12\x03\x13\x04\
+    \x16\n\x0c\n\x05\x05\0\x02\x0c\x02\x12\x03\x13\x19\x1b\n\x0b\n\x04\x05\0\
+    \x02\r\x12\x03\x14\x04\x1c\n\x0c\n\x05\x05\0\x02\r\x01\x12\x03\x14\x04\
+    \x16\n\x0c\n\x05\x05\0\x02\r\x02\x12\x03\x14\x19\x1b\n\x0b\n\x04\x05\0\
+    \x02\x0e\x12\x03\x15\x04\x19\n\x0c\n\x05\x05\0\x02\x0e\x01\x12\x03\x15\
+    \x04\x13\n\x0c\n\x05\x05\0\x02\x0e\x02\x12\x03\x15\x16\x18\n\x0b\n\x04\
+    \x05\0\x02\x0f\x12\x03\x16\x04\x19\n\x0c\n\x05\x05\0\x02\x0f\x01\x12\x03\
+    \x16\x04\x13\n\x0c\n\x05\x05\0\x02\x0f\x02\x12\x03\x16\x16\x18\n\x0b\n\
+    \x04\x05\0\x02\x10\x12\x03\x17\x04*\n\x0c\n\x05\x05\0\x02\x10\x01\x12\
     \x03\x17\x04$\n\x0c\n\x05\x05\0\x02\x10\x02\x12\x03\x17')\n\x0b\n\x04\
     \x05\0\x02\x11\x12\x03\x18\x04\x1f\n\x0c\n\x05\x05\0\x02\x11\x01\x12\x03\
     \x18\x04\x19\n\x0c\n\x05\x05\0\x02\x11\x02\x12\x03\x18\x1c\x1e\n\x0b\n\
@@ -413,13 +420,16 @@ static file_descriptor_proto_data: &'static [u8] = b"\
     \x12\x03\x1d\x04\x18\n\x0c\n\x05\x05\0\x02\x16\x02\x12\x03\x1d\x1b\x1d\n\
     \x0b\n\x04\x05\0\x02\x17\x12\x03\x1e\x04\x17\n\x0c\n\x05\x05\0\x02\x17\
     \x01\x12\x03\x1e\x04\x11\n\x0c\n\x05\x05\0\x02\x17\x02\x12\x03\x1e\x14\
-    \x16\n\x0b\n\x04\x05\0\x02\x18\x12\x03\x1f\x04&\n\x0c\n\x05\x05\0\x02\
-    \x18\x01\x12\x03\x1f\x04\x20\n\x0c\n\x05\x05\0\x02\x18\x02\x12\x03\x1f#%\
-    \n\x0b\n\x04\x05\0\x02\x19\x12\x03\x20\x04&\n\x0c\n\x05\x05\0\x02\x19\
-    \x01\x12\x03\x20\x04\x20\n\x0c\n\x05\x05\0\x02\x19\x02\x12\x03\x20#%\n\
-    \x0b\n\x04\x05\0\x02\x1a\x12\x03!\x04\x16\n\x0c\n\x05\x05\0\x02\x1a\x01\
-    \x12\x03!\x04\x0f\n\x0c\n\x05\x05\0\x02\x1a\x02\x12\x03!\x12\x15b\x06pro\
-    to3\
+    \x16\n\x0b\n\x04\x05\0\x02\x18\x12\x03\x1f\x04\x1a\n\x0c\n\x05\x05\0\x02\
+    \x18\x01\x12\x03\x1f\x04\x14\n\x0c\n\x05\x05\0\x02\x18\x02\x12\x03\x1f\
+    \x17\x19\n\x0b\n\x04\x05\0\x02\x19\x12\x03\x20\x04\x16\n\x0c\n\x05\x05\0\
+    \x02\x19\x01\x12\x03\x20\x04\x10\n\x0c\n\x05\x05\0\x02\x19\x02\x12\x03\
+    \x20\x13\x15\n\x0b\n\x04\x05\0\x02\x1a\x12\x03!\x04&\n\x0c\n\x05\x05\0\
+    \x02\x1a\x01\x12\x03!\x04\x20\n\x0c\n\x05\x05\0\x02\x1a\x02\x12\x03!#%\n\
+    \x0b\n\x04\x05\0\x02\x1b\x12\x03\"\x04&\n\x0c\n\x05\x05\0\x02\x1b\x01\
+    \x12\x03\"\x04\x20\n\x0c\n\x05\x05\0\x02\x1b\x02\x12\x03\"#%\n\x0b\n\x04\
+    \x05\0\x02\x1c\x12\x03#\x04\x16\n\x0c\n\x05\x05\0\x02\x1c\x01\x12\x03#\
+    \x04\x0f\n\x0c\n\x05\x05\0\x02\x1c\x02\x12\x03#\x12\x15b\x06proto3\
 ";
 
 static file_descriptor_proto_lazy: ::protobuf::rt::LazyV2<::protobuf::descriptor::FileDescriptorProto> = ::protobuf::rt::LazyV2::INIT;

rust-lib/flowy-user/src/protobuf/model/mod.rs

@@ -1,5 +1,8 @@
 // Auto-generated, do not edit 
 
+mod observable; 
+pub use observable::*; 
+
 mod user_table; 
 pub use user_table::*; 
 

rust-lib/flowy-user/src/protobuf/model/observable.rs

@@ -0,0 +1,103 @@
+// This file is generated by rust-protobuf 2.22.1. Do not edit
+// @generated
+
+// https://github.com/rust-lang/rust-clippy/issues/702
+#![allow(unknown_lints)]
+#![allow(clippy::all)]
+
+#![allow(unused_attributes)]
+#![cfg_attr(rustfmt, rustfmt::skip)]
+
+#![allow(box_pointers)]
+#![allow(dead_code)]
+#![allow(missing_docs)]
+#![allow(non_camel_case_types)]
+#![allow(non_snake_case)]
+#![allow(non_upper_case_globals)]
+#![allow(trivial_casts)]
+#![allow(unused_imports)]
+#![allow(unused_results)]
+//! Generated file from `observable.proto`
+
+/// Generated files are compatible only with the same version
+/// of protobuf runtime.
+// const _PROTOBUF_VERSION_CHECK: () = ::protobuf::VERSION_2_22_1;
+
+#[derive(Clone,PartialEq,Eq,Debug,Hash)]
+pub enum UserObservable {
+    Unknown = 0,
+    UserAuthChanged = 1,
+    UserProfileUpdated = 2,
+}
+
+impl ::protobuf::ProtobufEnum for UserObservable {
+    fn value(&self) -> i32 {
+        *self as i32
+    }
+
+    fn from_i32(value: i32) -> ::std::option::Option<UserObservable> {
+        match value {
+            0 => ::std::option::Option::Some(UserObservable::Unknown),
+            1 => ::std::option::Option::Some(UserObservable::UserAuthChanged),
+            2 => ::std::option::Option::Some(UserObservable::UserProfileUpdated),
+            _ => ::std::option::Option::None
+        }
+    }
+
+    fn values() -> &'static [Self] {
+        static values: &'static [UserObservable] = &[
+            UserObservable::Unknown,
+            UserObservable::UserAuthChanged,
+            UserObservable::UserProfileUpdated,
+        ];
+        values
+    }
+
+    fn enum_descriptor_static() -> &'static ::protobuf::reflect::EnumDescriptor {
+        static descriptor: ::protobuf::rt::LazyV2<::protobuf::reflect::EnumDescriptor> = ::protobuf::rt::LazyV2::INIT;
+        descriptor.get(|| {
+            ::protobuf::reflect::EnumDescriptor::new_pb_name::<UserObservable>("UserObservable", file_descriptor_proto())
+        })
+    }
+}
+
+impl ::std::marker::Copy for UserObservable {
+}
+
+impl ::std::default::Default for UserObservable {
+    fn default() -> Self {
+        UserObservable::Unknown
+    }
+}
+
+impl ::protobuf::reflect::ProtobufValue for UserObservable {
+    fn as_ref(&self) -> ::protobuf::reflect::ReflectValueRef {
+        ::protobuf::reflect::ReflectValueRef::Enum(::protobuf::ProtobufEnum::descriptor(self))
+    }
+}
+
+static file_descriptor_proto_data: &'static [u8] = b"\
+    \n\x10observable.proto*J\n\x0eUserObservable\x12\x0b\n\x07Unknown\x10\0\
+    \x12\x13\n\x0fUserAuthChanged\x10\x01\x12\x16\n\x12UserProfileUpdated\
+    \x10\x02J\xa5\x01\n\x06\x12\x04\0\0\x05\x01\n\x08\n\x01\x0c\x12\x03\0\0\
+    \x12\n\n\n\x02\x05\0\x12\x04\x01\0\x05\x01\n\n\n\x03\x05\0\x01\x12\x03\
+    \x01\x05\x13\n\x0b\n\x04\x05\0\x02\0\x12\x03\x02\x04\x10\n\x0c\n\x05\x05\
+    \0\x02\0\x01\x12\x03\x02\x04\x0b\n\x0c\n\x05\x05\0\x02\0\x02\x12\x03\x02\
+    \x0e\x0f\n\x0b\n\x04\x05\0\x02\x01\x12\x03\x03\x04\x18\n\x0c\n\x05\x05\0\
+    \x02\x01\x01\x12\x03\x03\x04\x13\n\x0c\n\x05\x05\0\x02\x01\x02\x12\x03\
+    \x03\x16\x17\n\x0b\n\x04\x05\0\x02\x02\x12\x03\x04\x04\x1b\n\x0c\n\x05\
+    \x05\0\x02\x02\x01\x12\x03\x04\x04\x16\n\x0c\n\x05\x05\0\x02\x02\x02\x12\
+    \x03\x04\x19\x1ab\x06proto3\
+";
+
+static file_descriptor_proto_lazy: ::protobuf::rt::LazyV2<::protobuf::descriptor::FileDescriptorProto> = ::protobuf::rt::LazyV2::INIT;
+
+fn parse_descriptor_proto() -> ::protobuf::descriptor::FileDescriptorProto {
+    ::protobuf::Message::parse_from_bytes(file_descriptor_proto_data).unwrap()
+}
+
+pub fn file_descriptor_proto() -> &'static ::protobuf::descriptor::FileDescriptorProto {
+    file_descriptor_proto_lazy.get(|| {
+        parse_descriptor_proto()
+    })
+}

rust-lib/flowy-user/src/protobuf/proto/errors.proto

@@ -29,7 +29,9 @@ enum ErrorCode {
     UserNameIsEmpty = 42;
     UserWorkspaceInvalid = 50;
     UserIdInvalid = 51;
-    CreateDefaultWorkspaceFailed = 52;
-    DefaultWorkspaceAlreadyExist = 53;
+    UserTokenInvalid = 54;
+    UserNotExist = 55;
+    CreateDefaultWorkspaceFailed = 60;
+    DefaultWorkspaceAlreadyExist = 61;
     ServerError = 100;
 }

rust-lib/flowy-user/src/protobuf/proto/observable.proto

@@ -0,0 +1,6 @@
+syntax = "proto3";
+enum UserObservable {
+    Unknown = 0;
+    UserAuthChanged = 1;
+    UserProfileUpdated = 2;
+}

rust-lib/flowy-user/src/services/user/user_session.rs

@@ -5,6 +5,7 @@ use crate::{
     sql_tables::{UserTable, UserTableChangeset},
 };
 
+use crate::{observable::*, services::server::Server};
 use flowy_database::{
     query_dsl::*,
     schema::{user_table, user_table::dsl},
@@ -12,8 +13,6 @@ use flowy_database::{
     ExpressionMethods,
     UserDatabaseConnection,
 };
-
-use crate::services::server::Server;
 use flowy_infra::kv::KV;
 use flowy_sqlite::ConnectionPool;
 use parking_lot::RwLock;
@@ -118,13 +117,12 @@ impl UserSession {
     }
 
     pub async fn user_profile(&self) -> Result<UserProfile, UserError> {
-        let session = self.get_session()?;
-        let token = session.token;
+        let (user_id, token) = self.get_session()?.into_part();
         let user = dsl::user_table
-            .filter(user_table::id.eq(&session.user_id))
+            .filter(user_table::id.eq(&user_id))
             .first::<UserTable>(&*(self.db_conn()?))?;
 
-        let _ = self.read_user_profile_on_server(&token).await?;
+        let _ = self.read_user_profile_on_server(&token, &user_id).await?;
         Ok(UserProfile::from(user))
     }
 
@@ -139,18 +137,18 @@ impl UserSession {
 }
 
 impl UserSession {
-    async fn read_user_profile_on_server(&self, token: &str) -> Result<(), UserError> {
+    async fn read_user_profile_on_server(&self, token: &str, user_id: &str) -> Result<(), UserError> {
         let server = self.server.clone();
         let token = token.to_owned();
-        let _ = tokio::spawn(async move {
+        let user_id = user_id.to_owned();
+        tokio::spawn(async move {
             match server.get_user(&token).await {
                 Ok(profile) => {
-                    //
-                    log::info!("{:?}", profile);
+                    observable(&user_id, UserObservable::UserProfileUpdated).payload(profile).build();
                 },
                 Err(e) => {
-                    //
-                    log::info!("{:?}", e);
+                    log::error!("{:?}", e);
+                    observable(&user_id, UserObservable::UserProfileUpdated).error(e).build();
                 },
             }
         });
@@ -193,7 +191,7 @@ impl UserSession {
     }
 
     fn set_session(&self, session: Option<Session>) -> Result<(), UserError> {
-        log::debug!("Update user session: {:?}", session);
+        log::debug!("Set user session: {:?}", session);
         match &session {
             None => KV::remove(SESSION_CACHE_KEY).map_err(|e| UserError::new(ErrorCode::SqlInternalError, &e))?,
             Some(session) => KV::set_str(SESSION_CACHE_KEY, session.clone().into()),
@@ -256,6 +254,8 @@ impl Session {
             email: email.to_owned(),
         }
     }
+
+    pub fn into_part(mut self) -> (String, String) { (self.user_id, self.token) }
 }
 
 impl std::convert::From<String> for Session {

rust-lib/flowy-workspace/src/observable/mod.rs

@@ -1,3 +1,3 @@
-mod observable;
+pub(crate) mod observable;
 
 pub(crate) use observable::*;

rust-lib/flowy-workspace/src/observable/observable.rs

@@ -1,7 +1,7 @@
 use bytes::Bytes;
 use flowy_derive::ProtoBuf_Enum;
 use flowy_dispatch::prelude::ToBytes;
-use flowy_observable::{dart::RustStreamSender, entities::ObservableSubject};
+use flowy_observable::{dart::RustStreamSender, entities::ObservableSubject, ObservableBuilder};
 
 const OBSERVABLE_CATEGORY: &'static str = "Workspace";
 
@@ -28,73 +28,8 @@ impl std::default::Default for WorkspaceObservable {
     fn default() -> Self { WorkspaceObservable::Unknown }
 }
 
-pub(crate) struct ObservableBuilder {
-    id: String,
-    payload: Option<Bytes>,
-    error: Option<Bytes>,
-    ty: WorkspaceObservable,
+impl std::convert::Into<i32> for WorkspaceObservable {
+    fn into(self) -> i32 { self as i32 }
 }
 
-impl ObservableBuilder {
-    pub(crate) fn new(id: &str, ty: WorkspaceObservable) -> Self {
-        Self {
-            id: id.to_owned(),
-            ty,
-            payload: None,
-            error: None,
-        }
-    }
-
-    pub(crate) fn payload<T>(mut self, payload: T) -> Self
-    where
-        T: ToBytes,
-    {
-        match payload.into_bytes() {
-            Ok(bytes) => self.payload = Some(bytes),
-            Err(e) => {
-                log::error!("Set observable payload failed: {:?}", e);
-            },
-        }
-
-        self
-    }
-
-    pub(crate) fn error<T>(mut self, error: T) -> Self
-    where
-        T: ToBytes,
-    {
-        match error.into_bytes() {
-            Ok(bytes) => self.error = Some(bytes),
-            Err(e) => {
-                log::error!("Set observable error failed: {:?}", e);
-            },
-        }
-        self
-    }
-
-    pub(crate) fn build(self) {
-        log::trace!("Workspace observable id: {}, ty: {:?}", self.id, self.ty);
-
-        let payload = match self.payload {
-            None => None,
-            Some(bytes) => Some(bytes.to_vec()),
-        };
-
-        let error = match self.error {
-            None => None,
-            Some(bytes) => Some(bytes.to_vec()),
-        };
-
-        let subject = ObservableSubject {
-            category: OBSERVABLE_CATEGORY.to_string(),
-            ty: self.ty as i32,
-            id: self.id,
-            payload,
-            error,
-        };
-        match RustStreamSender::post(subject) {
-            Ok(_) => {},
-            Err(error) => log::error!("Send observable subject failed: {}", error),
-        }
-    }
-}
+pub(crate) fn observable(id: &str, ty: WorkspaceObservable) -> ObservableBuilder { ObservableBuilder::new(id, ty, OBSERVABLE_CATEGORY) }

rust-lib/flowy-workspace/src/services/app_controller.rs

@@ -37,7 +37,7 @@ impl AppController {
 
         // Opti: transaction
         let apps = self.read_local_apps(&app.workspace_id, &*conn)?;
-        ObservableBuilder::new(&app.workspace_id, WorkspaceObservable::WorkspaceCreateApp)
+        observable(&app.workspace_id, WorkspaceObservable::WorkspaceCreateApp)
             .payload(apps)
             .build();
         Ok(app)
@@ -58,7 +58,7 @@ impl AppController {
         let app = self.sql.delete_app(app_id, &*conn)?;
         // Opti: transaction
         let apps = self.read_local_apps(&app.workspace_id, &*conn)?;
-        ObservableBuilder::new(&app.workspace_id, WorkspaceObservable::WorkspaceDeleteApp)
+        observable(&app.workspace_id, WorkspaceObservable::WorkspaceDeleteApp)
             .payload(apps)
             .build();
         Ok(())
@@ -78,9 +78,7 @@ impl AppController {
         let conn = self.database.db_connection()?;
         let _ = self.sql.update_app(changeset, &*conn)?;
         let app: App = self.sql.read_app(&app_id, false, &*conn)?.into();
-        ObservableBuilder::new(&app_id, WorkspaceObservable::AppUpdated)
-            .payload(app)
-            .build();
+        observable(&app_id, WorkspaceObservable::AppUpdated).payload(app).build();
         Ok(())
     }
 }

rust-lib/flowy-workspace/src/services/view_controller.rs

@@ -2,7 +2,7 @@ use crate::{
     entities::view::{CreateViewParams, UpdateViewParams, View},
     errors::WorkspaceError,
     module::WorkspaceDatabase,
-    observable::WorkspaceObservable,
+    observable::observable,
     services::{helper::spawn, server::Server},
     sql_tables::view::{ViewTable, ViewTableChangeset, ViewTableSql},
 };
@@ -10,7 +10,7 @@ use crate::{
 use crate::{
     entities::view::{DeleteViewParams, QueryViewParams, RepeatedView},
     module::WorkspaceUser,
-    observable::ObservableBuilder,
+    observable::WorkspaceObservable,
 };
 use flowy_database::SqliteConnection;
 use std::sync::Arc;
@@ -41,7 +41,7 @@ impl ViewController {
         (conn).immediate_transaction::<_, WorkspaceError, _>(|| {
             let _ = self.sql.create_view(view_table, conn)?;
             let repeated_view = self.read_local_views_belong_to(&view.belong_to_id, conn)?;
-            ObservableBuilder::new(&view.belong_to_id, WorkspaceObservable::AppCreateView)
+            observable(&view.belong_to_id, WorkspaceObservable::AppCreateView)
                 .payload(repeated_view)
                 .build();
             Ok(())
@@ -64,7 +64,7 @@ impl ViewController {
         (conn).immediate_transaction::<_, WorkspaceError, _>(|| {
             let view_table = self.sql.delete_view(view_id, conn)?;
             let repeated_view = self.read_local_views_belong_to(&view_table.belong_to_id, conn)?;
-            ObservableBuilder::new(&view_table.belong_to_id, WorkspaceObservable::AppDeleteView)
+            observable(&view_table.belong_to_id, WorkspaceObservable::AppDeleteView)
                 .payload(repeated_view)
                 .build();
             Ok(())
@@ -92,9 +92,7 @@ impl ViewController {
         (conn).immediate_transaction::<_, WorkspaceError, _>(|| {
             let _ = self.sql.update_view(changeset, conn)?;
             let view: View = self.sql.read_view(&view_id, None, conn)?.into();
-            ObservableBuilder::new(&view_id, WorkspaceObservable::ViewUpdated)
-                .payload(view)
-                .build();
+            observable(&view_id, WorkspaceObservable::ViewUpdated).payload(view).build();
             Ok(())
         })?;
 

rust-lib/flowy-workspace/src/services/workspace_controller.rs

@@ -2,21 +2,29 @@ use crate::{
     entities::{app::App, workspace::*},
     errors::*,
     module::{WorkspaceDatabase, WorkspaceUser},
-    observable::WorkspaceObservable,
+    observable::observable,
     services::{helper::spawn, server::Server, AppController},
     sql_tables::workspace::{WorkspaceTable, WorkspaceTableChangeset, WorkspaceTableSql},
 };
 
 use flowy_infra::kv::KV;
 
-use crate::{entities::app::RepeatedApp, observable::ObservableBuilder};
+use crate::{
+    entities::app::RepeatedApp,
+    observable::WorkspaceObservable,
+    sql_tables::{
+        app::{AppTable, AppTableSql},
+        view::{ViewTable, ViewTableSql},
+    },
+};
 use flowy_database::SqliteConnection;
 use std::sync::Arc;
 
 pub(crate) struct WorkspaceController {
     pub user: Arc<dyn WorkspaceUser>,
     pub workspace_sql: Arc<WorkspaceTableSql>,
-    // pub app_sql: Arc<AppTableSql>,
+    pub app_sql: Arc<AppTableSql>,
+    pub view_sql: Arc<ViewTableSql>,
     pub database: Arc<dyn WorkspaceDatabase>,
     pub app_controller: Arc<AppController>,
     server: Server,
@@ -29,10 +37,14 @@ impl WorkspaceController {
         app_controller: Arc<AppController>,
         server: Server,
     ) -> Self {
-        let sql = Arc::new(WorkspaceTableSql {});
+        let workspace_sql = Arc::new(WorkspaceTableSql {});
+        let app_sql = Arc::new(AppTableSql {});
+        let view_sql = Arc::new(ViewTableSql {});
         Self {
             user,
-            workspace_sql: sql,
+            workspace_sql,
+            app_sql,
+            view_sql,
             database,
             app_controller,
             server,
@@ -58,7 +70,7 @@ impl WorkspaceController {
         (conn).immediate_transaction::<_, WorkspaceError, _>(|| {
             self.workspace_sql.create_workspace(workspace_table, conn)?;
             let repeated_workspace = self.read_local_workspaces(None, &user_id, conn)?;
-            ObservableBuilder::new(&user_id, WorkspaceObservable::UserCreateWorkspace)
+            observable(&user_id, WorkspaceObservable::UserCreateWorkspace)
                 .payload(repeated_workspace)
                 .build();
 
@@ -78,7 +90,7 @@ impl WorkspaceController {
             let _ = self.workspace_sql.update_workspace(changeset, conn)?;
             let user_id = self.user.user_id()?;
             let workspace = self.read_local_workspace(workspace_id.clone(), &user_id, conn)?;
-            ObservableBuilder::new(&workspace_id, WorkspaceObservable::WorkspaceUpdated)
+            observable(&workspace_id, WorkspaceObservable::WorkspaceUpdated)
                 .payload(workspace)
                 .build();
 
@@ -95,7 +107,7 @@ impl WorkspaceController {
         (conn).immediate_transaction::<_, WorkspaceError, _>(|| {
             let _ = self.workspace_sql.delete_workspace(workspace_id, conn)?;
             let repeated_workspace = self.read_local_workspaces(None, &user_id, conn)?;
-            ObservableBuilder::new(&user_id, WorkspaceObservable::UserDeleteWorkspace)
+            observable(&user_id, WorkspaceObservable::UserDeleteWorkspace)
                 .payload(repeated_workspace)
                 .build();
 
@@ -244,40 +256,35 @@ impl WorkspaceController {
     #[tracing::instrument(skip(self), err)]
     async fn read_workspaces_on_server(&self, user_id: String, params: QueryWorkspaceParams) -> Result<(), WorkspaceError> {
         let (token, server) = self.token_with_server()?;
-        let sql = self.workspace_sql.clone();
+        let workspace_sql = self.workspace_sql.clone();
+        let app_sql = self.app_sql.clone();
+        let view_sql = self.view_sql.clone();
         let conn = self.database.db_connection()?;
         spawn(async move {
             // Opti: retry?
             let workspaces = server.read_workspace(&token, params).await?;
             let _ = (&*conn).immediate_transaction::<_, WorkspaceError, _>(|| {
+                log::debug!("Save {} workspace", workspaces.len());
                 for workspace in &workspaces.items {
                     let mut m_workspace = workspace.clone();
                     let apps = m_workspace.apps.take_items();
                     let workspace_table = WorkspaceTable::new(m_workspace, &user_id);
-                    log::debug!("Save workspace");
-                    let _ = sql.create_workspace(workspace_table, &*conn)?;
-                    log::debug!("Save apps");
 
+                    let _ = workspace_sql.create_workspace(workspace_table, &*conn)?;
+                    log::debug!("Save {} apps", apps.len());
                     for mut app in apps {
                         let views = app.belongings.take_items();
-                        // let _ = sql.create_apps(vec![app], &*conn)?;
-
-                        // pub(crate) fn create_apps(&self, apps: Vec<App>, conn: &SqliteConnection) ->
-                        // Result<(), WorkspaceError> {     for app in apps {
-                        //         let _ = self.app_sql.create_app_with(AppTable::new(app), conn)?;
-                        //     }
-                        //     Ok(())
-                        // }
-
-                        log::debug!("Save views");
-                        for _view in views {
-                            //
+                        app_sql.create_app(AppTable::new(app), &*conn);
+
+                        log::debug!("Save {} views", views.len());
+                        for view in views {
+                            view_sql.create_view(ViewTable::new(view), &*conn);
                         }
                     }
                 }
                 Ok(())
             })?;
-            ObservableBuilder::new(&user_id, WorkspaceObservable::WorkspaceListUpdated)
+            observable(&user_id, WorkspaceObservable::WorkspaceListUpdated)
                 .payload(workspaces)
                 .build();
             Result::<(), WorkspaceError>::Ok(())