config user watch repo

app_flowy/lib/workspace/domain/i_user.dart

@@ -9,6 +9,11 @@ export 'package:flowy_sdk/protobuf/flowy-user/errors.pb.dart';
 export 'package:flowy_sdk/protobuf/flowy-user/user_detail.pb.dart';
 export 'package:flowy_sdk/protobuf/flowy-workspace/errors.pb.dart';
 
+typedef UserCreateWorkspaceCallback = void Function(
+    Either<List<Workspace>, WorkspaceError> workspacesOrFailed);
+typedef UserDeleteWorkspaceCallback = void Function(
+    Either<List<Workspace>, WorkspaceError> workspacesOrFailed);
+
 abstract class IUser {
   UserDetail get user;
   Future<Either<UserDetail, UserError>> fetchUserDetail(String userId);
@@ -16,3 +21,11 @@ abstract class IUser {
   Future<Either<Unit, WorkspaceError>> deleteWorkspace(String workspaceId);
   Future<Either<Unit, UserError>> signOut();
 }
+
+abstract class IUserWatch {
+  void startWatching(
+      {UserCreateWorkspaceCallback? createWorkspaceCallback,
+      UserDeleteWorkspaceCallback? deleteWorkspaceCallback});
+
+  Future<void> stopWatching();
+}

app_flowy/lib/workspace/infrastructure/deps_resolver.dart

@@ -55,6 +55,8 @@ class HomeDepsResolver {
     // User
     getIt.registerFactoryParam<IUser, UserDetail, void>(
         (user, _) => IUserImpl(repo: UserRepo(user: user)));
+    getIt.registerFactoryParam<IUserWatch, UserDetail, void>(
+        (user, _) => IUserWatchImpl(repo: UserWatchRepo(user: user)));
 
     //Menu Bloc
     getIt.registerFactoryParam<MenuBloc, UserDetail, void>(

app_flowy/lib/workspace/infrastructure/i_user_impl.dart

@@ -1,5 +1,4 @@
 import 'package:dartz/dartz.dart';
-
 import 'package:app_flowy/workspace/domain/i_user.dart';
 import 'package:app_flowy/workspace/infrastructure/repos/user_repo.dart';
 export 'package:app_flowy/workspace/domain/i_user.dart';
@@ -34,3 +33,23 @@ class IUserImpl extends IUser {
     return repo.fetchWorkspaces();
   }
 }
+
+class IUserWatchImpl extends IUserWatch {
+  UserWatchRepo repo;
+  IUserWatchImpl({
+    required this.repo,
+  });
+  @override
+  void startWatching(
+      {UserCreateWorkspaceCallback? createWorkspaceCallback,
+      UserDeleteWorkspaceCallback? deleteWorkspaceCallback}) {
+    repo.startWatching(
+        createWorkspace: createWorkspaceCallback,
+        deleteWorkspace: deleteWorkspaceCallback);
+  }
+
+  @override
+  Future<void> stopWatching() async {
+    await repo.close();
+  }
+}

app_flowy/lib/workspace/infrastructure/repos/user_repo.dart

@@ -1,9 +1,16 @@
+import 'dart:async';
+
 import 'package:dartz/dartz.dart';
 import 'package:flowy_sdk/dispatch/dispatch.dart';
+import 'package:flowy_sdk/protobuf/flowy-observable/subject.pb.dart';
 import 'package:flowy_sdk/protobuf/flowy-user/errors.pb.dart';
 import 'package:flowy_sdk/protobuf/flowy-user/user_detail.pb.dart';
 import 'package:flowy_sdk/protobuf/flowy-workspace/errors.pb.dart';
+import 'package:flowy_sdk/protobuf/flowy-workspace/observable.pb.dart';
 import 'package:flowy_sdk/protobuf/flowy-workspace/workspace_create.pb.dart';
+import 'package:flowy_sdk/rust_stream.dart';
+
+import 'package:app_flowy/workspace/domain/i_user.dart';
 
 class UserRepo {
   final UserDetail user;
@@ -34,3 +41,66 @@ class UserRepo {
     });
   }
 }
+
+class UserWatchRepo {
+  StreamSubscription<ObservableSubject>? _subscription;
+  UserCreateWorkspaceCallback? _createWorkspace;
+  UserDeleteWorkspaceCallback? _deleteWorkspace;
+  late UserRepo _repo;
+  UserWatchRepo({
+    required UserDetail user,
+  }) {
+    _repo = UserRepo(user: user);
+  }
+
+  void startWatching(
+      {UserCreateWorkspaceCallback? createWorkspace,
+      UserDeleteWorkspaceCallback? deleteWorkspace}) {
+    _createWorkspace = createWorkspace;
+    _deleteWorkspace = deleteWorkspace;
+    _subscription = RustStreamReceiver.listen((observable) {
+      if (observable.subjectId != _repo.user.id) {
+        return;
+      }
+
+      final ty = WorkspaceObservable.valueOf(observable.ty);
+      if (ty != null) {
+        _handleObservableType(ty);
+      }
+    });
+  }
+
+  Future<void> close() async {
+    await _subscription?.cancel();
+  }
+
+  void _handleObservableType(WorkspaceObservable ty) {
+    switch (ty) {
+      case WorkspaceObservable.UserCreateWorkspace:
+        if (_createWorkspace == null) {
+          return;
+        }
+        _repo.fetchWorkspaces().then((result) {
+          result.fold(
+            (workspaces) => _createWorkspace!(left(workspaces)),
+            (error) => _createWorkspace!(right(error)),
+          );
+        });
+        break;
+      case WorkspaceObservable.UserDeleteWorkspace:
+        if (_deleteWorkspace == null) {
+          return;
+        }
+        _repo.fetchWorkspaces().then((result) {
+          result.fold(
+            (workspaces) => _deleteWorkspace!(left(workspaces)),
+            (error) => _deleteWorkspace!(right(error)),
+          );
+        });
+        break;
+
+      default:
+        break;
+    }
+  }
+}