Expose User.changes (#1500)

* Expose User.changes

* Update core

* Update Core to master

* Fix build

* Fix changelog versions

Author: nirinchev

CHANGELOG.md

@@ -1,16 +1,18 @@
 ## vNext (TBD)
 
 ### Enhancements
-* None
+* Added `User.changes` stream that allows subscribers to receive notifications when the User changes - for example when the user's custom data changes or when their authentication state changes. (PR [#1500](https://github.com/realm/realm-dart/pull/1500))
+* Allow the query builder to construct >, >=, <, <= queries for string constants. This is a case sensitive lexicographical comparison. Improved performance of RQL queries on a non-linked string property using: >, >=, <, <=, operators and fixed behaviour that a null string should be evaulated as less than everything, previously nulls were not matched. (Core 13.26.0-13-gd12c3)
 
 ### Fixed
 * Creating an `AppConfiguration` with an empty appId will now throw an exception rather than crashing the app. (Issue [#1487](https://github.com/realm/realm-dart/issues/1487))
+* Uploading the changesets recovered during an automatic client reset recovery may lead to 'Bad server version' errors and a new client reset. (Core 13.26.0-13-gd12c3)
 
 ### Compatibility
 * Realm Studio: 13.0.0 or later.
 
 ### Internal
-* Using Core x.y.z.
+* Using Core 13.26.0-13-gd12c3
 
 ## 1.8.0 (2024-01-29)
 

lib/src/native/realm_bindings.dart

@@ -4019,6 +4019,24 @@ class RealmLibrary {
       _realm_dart_sync_wait_for_completion_callbackPtr.asFunction<
           void Function(ffi.Pointer<ffi.Void>, ffi.Pointer<realm_error_t>)>();
 
+  void realm_dart_user_change_callback(
+    ffi.Pointer<ffi.Void> userdata,
+    int state,
+  ) {
+    return _realm_dart_user_change_callback(
+      userdata,
+      state,
+    );
+  }
+
+  late final _realm_dart_user_change_callbackPtr = _lookup<
+          ffi
+          .NativeFunction<ffi.Void Function(ffi.Pointer<ffi.Void>, ffi.Int32)>>(
+      'realm_dart_user_change_callback');
+  late final _realm_dart_user_change_callback =
+      _realm_dart_user_change_callbackPtr
+          .asFunction<void Function(ffi.Pointer<ffi.Void>, int)>();
+
   void realm_dart_user_completion_callback(
     ffi.Pointer<ffi.Void> userdata,
     ffi.Pointer<realm_user_t> user,
@@ -10872,6 +10890,38 @@ class RealmLibrary {
           realm_timestamp_t Function(
               ffi.Pointer<realm_flx_sync_subscription_t>)>();
 
+  /// @return a notification token object. Dispose it to stop receiving notifications.
+  ffi.Pointer<realm_sync_user_subscription_token_t>
+      realm_sync_user_on_state_change_register_callback(
+    ffi.Pointer<realm_user_t> arg0,
+    realm_sync_on_user_state_changed_t arg1,
+    ffi.Pointer<ffi.Void> userdata,
+    realm_free_userdata_func_t userdata_free,
+  ) {
+    return _realm_sync_user_on_state_change_register_callback(
+      arg0,
+      arg1,
+      userdata,
+      userdata_free,
+    );
+  }
+
+  late final _realm_sync_user_on_state_change_register_callbackPtr = _lookup<
+          ffi.NativeFunction<
+              ffi.Pointer<realm_sync_user_subscription_token_t> Function(
+                  ffi.Pointer<realm_user_t>,
+                  realm_sync_on_user_state_changed_t,
+                  ffi.Pointer<ffi.Void>,
+                  realm_free_userdata_func_t)>>(
+      'realm_sync_user_on_state_change_register_callback');
+  late final _realm_sync_user_on_state_change_register_callback =
+      _realm_sync_user_on_state_change_register_callbackPtr.asFunction<
+          ffi.Pointer<realm_sync_user_subscription_token_t> Function(
+              ffi.Pointer<realm_user_t>,
+              realm_sync_on_user_state_changed_t,
+              ffi.Pointer<ffi.Void>,
+              realm_free_userdata_func_t)>();
+
   /// Update the schema of an open realm.
   ///
   /// This is equivalent to calling `realm_update_schema_advanced(realm, schema, 0,
@@ -11377,6 +11427,11 @@ class _SymbolAddresses {
                   ffi.Pointer<ffi.Void>, ffi.Pointer<realm_error_t>)>>
       get realm_dart_sync_wait_for_completion_callback =>
           _library._realm_dart_sync_wait_for_completion_callbackPtr;
+  ffi.Pointer<
+          ffi
+          .NativeFunction<ffi.Void Function(ffi.Pointer<ffi.Void>, ffi.Int32)>>
+      get realm_dart_user_change_callback =>
+          _library._realm_dart_user_change_callbackPtr;
   ffi.Pointer<
           ffi.NativeFunction<
               ffi.Void Function(ffi.Pointer<ffi.Void>,
@@ -12488,6 +12543,12 @@ typedef realm_sync_on_subscription_state_changed_tFunction = ffi.Void Function(
     ffi.Pointer<ffi.Void> userdata, ffi.Int32 state);
 typedef Dartrealm_sync_on_subscription_state_changed_tFunction = void Function(
     ffi.Pointer<ffi.Void> userdata, int state);
+typedef realm_sync_on_user_state_changed_t = ffi
+    .Pointer<ffi.NativeFunction<realm_sync_on_user_state_changed_tFunction>>;
+typedef realm_sync_on_user_state_changed_tFunction = ffi.Void Function(
+    ffi.Pointer<ffi.Void> userdata, ffi.Int32 s);
+typedef Dartrealm_sync_on_user_state_changed_tFunction = void Function(
+    ffi.Pointer<ffi.Void> userdata, int s);
 
 abstract class realm_sync_progress_direction {
   static const int RLM_SYNC_PROGRESS_DIRECTION_UPLOAD = 0;
@@ -12635,6 +12696,11 @@ typedef Dartrealm_sync_ssl_verify_func_tFunction = bool Function(
     int preverify_ok,
     int depth);
 
+final class realm_sync_user_subscription_token extends ffi.Opaque {}
+
+typedef realm_sync_user_subscription_token_t
+    = realm_sync_user_subscription_token;
+
 /// Callback function invoked by the sync session once it has uploaded or download
 /// all available changesets. See @a realm_sync_session_wait_for_upload and
 /// @a realm_sync_session_wait_for_download.

lib/src/native/realm_core.dart

@@ -1735,6 +1735,12 @@ class _RealmCore {
     }
   }
 
+  static void user_change_callback(Pointer<Void> userdata, int data) {
+    final controller = userdata as UserNotificationsController;
+
+    controller.onUserChanged();
+  }
+
   RealmNotificationTokenHandle subscribeResultsNotifications(RealmResults results, NotificationsController controller) {
     final pointer = _realmLib.invokeGetPointer(() => _realmLib.realm_results_add_notification_callback(
           results.handle._pointer,
@@ -1783,6 +1789,18 @@ class _RealmCore {
     return RealmNotificationTokenHandle._(pointer, map.realm.handle);
   }
 
+  UserNotificationTokenHandle subscribeUserNotifications(UserNotificationsController controller) {
+    final callback = Pointer.fromFunction<Void Function(Handle, Int32)>(user_change_callback);
+    final userdata = _realmLib.realm_dart_userdata_async_new(controller, callback.cast(), scheduler.handle._pointer);
+    final notification_token = _realmLib.realm_sync_user_on_state_change_register_callback(
+      controller.user.handle._pointer,
+      _realmLib.addresses.realm_dart_user_change_callback,
+      userdata.cast(),
+      _realmLib.addresses.realm_dart_userdata_async_free,
+    );
+    return UserNotificationTokenHandle._(notification_token);
+  }
+
   bool getObjectChangesIsDeleted(RealmObjectChangesHandle handle) {
     return _realmLib.realm_object_changes_is_deleted(handle._pointer);
   }
@@ -3184,6 +3202,10 @@ class RealmNotificationTokenHandle extends RootedHandleBase<realm_notification_t
   RealmNotificationTokenHandle._(Pointer<realm_notification_token> pointer, RealmHandle root) : super(root, pointer, 32);
 }
 
+class UserNotificationTokenHandle extends HandleBase<realm_sync_user_subscription_token> {
+  UserNotificationTokenHandle._(Pointer<realm_sync_user_subscription_token> pointer) : super(pointer, 32);
+}
+
 class RealmSyncSessionConnectionStateNotificationTokenHandle extends HandleBase<realm_sync_session_connection_state_notification_token> {
   RealmSyncSessionConnectionStateNotificationTokenHandle._(Pointer<realm_sync_session_connection_state_notification_token> pointer) : super(pointer, 32);
 }

lib/src/realm_class.dart

@@ -148,7 +148,7 @@ export 'session.dart'
         // ignore: deprecated_member_use_from_same_package
         SyncSessionErrorCode;
 export 'subscription.dart' show Subscription, SubscriptionSet, SubscriptionSetState, MutableSubscriptionSet;
-export 'user.dart' show User, UserState, ApiKeyClient, UserIdentity, ApiKey, FunctionsClient;
+export 'user.dart' show User, UserState, ApiKeyClient, UserIdentity, ApiKey, FunctionsClient, UserChanges;
 export 'native/realm_core.dart' show Decimal128;
 
 /// A [Realm] instance represents a `Realm` database.

lib/src/user.dart

@@ -16,12 +16,24 @@
 //
 ////////////////////////////////////////////////////////////////////////////////
 
+import 'dart:async';
 import 'dart:convert';
+import 'dart:ffi';
 
 import 'native/realm_core.dart';
 import 'realm_class.dart';
 import './app.dart';
 
+/// Describes the changes to a [User] instance - for example when the access token is updated or the user state changes.
+/// Right now, this only conveys information that the user has changed, but in the future it will be enhanced by adding
+/// details about the exact properties that have been updated.
+class UserChanges {
+  /// The user that has changed.
+  final User user;
+
+  const UserChanges._(this.user);
+}
+
 /// This class represents a `user` in an [Atlas App Services](https://www.mongodb.com/docs/atlas/app-services/) application.
 /// A user can log in to the server and, if access is granted, it is possible to synchronize the local Realm to MongoDB Atlas.
 /// Moreover, synchronization is halted when the user is logged out. It is possible to persist a user. By retrieving a user, there is no need to log in again.
@@ -161,6 +173,55 @@ class User {
       throw RealmError('User must be logged in to $clarification');
     }
   }
+
+  /// Gets a [Stream] of [UserChanges] that can be used to receive notifications when the user changes.
+  Stream<UserChanges> get changes {
+    final controller = UserNotificationsController(this);
+    return controller.createStream();
+  }
+}
+
+/// @nodoc
+class UserNotificationsController implements Finalizable {
+  UserNotificationTokenHandle? tokenHandle;
+
+  void start() {
+    if (tokenHandle != null) {
+      throw RealmStateError("User notifications subscription already started");
+    }
+
+    tokenHandle = realmCore.subscribeUserNotifications(this);
+  }
+
+  void stop() {
+    // If handle is null or released, no-op
+    if (tokenHandle?.released != false) {
+      return;
+    }
+
+    tokenHandle!.release();
+    tokenHandle = null;
+  }
+
+  User user;
+
+  late final StreamController<UserChanges> streamController;
+
+  UserNotificationsController(this.user);
+
+  Stream<UserChanges> createStream() {
+    streamController = StreamController<UserChanges>(onListen: start, onCancel: stop);
+    return streamController.stream;
+  }
+
+  void onUserChanged() {
+    final changes = UserChanges._(user);
+    streamController.add(changes);
+  }
+
+  void onError(RealmError error) {
+    streamController.addError(error);
+  }
 }
 
 /// The current state of a [User].

src/realm-core

@@ -173,6 +173,15 @@ RLM_API void realm_dart_sync_connection_state_changed_callback(realm_userdata_t
     });
 }
 
+
+RLM_API void realm_dart_user_change_callback(realm_userdata_t userdata, realm_user_state_e state)
+{
+    auto ud = reinterpret_cast<realm_dart_userdata_async_t>(userdata);
+    ud->scheduler->invoke([ud, state]() {
+        (reinterpret_cast<realm_sync_on_user_state_changed_t>(ud->dart_callback))(ud->handle, state);
+    });
+}
+
 RLM_API void realm_dart_sync_on_subscription_state_changed_callback(realm_userdata_t userdata, realm_flx_sync_subscription_set_state_e state)
 {
     auto ud = reinterpret_cast<realm_dart_userdata_async_t>(userdata);
@@ -297,7 +306,7 @@ RLM_API void realm_dart_void_completion_callback(realm_userdata_t userdata, cons
 struct apikey_buf : realm_app_user_apikey
 {
     apikey_buf(const realm_app_user_apikey& apikey_input)
-        : key_buffer(apikey_input.key ? apikey_input.key : ""), 
+        : key_buffer(apikey_input.key ? apikey_input.key : ""),
         name_buffer(apikey_input.name ? apikey_input.name : "")
     {
         id = apikey_input.id;
@@ -341,8 +350,8 @@ std::vector<apikey_buf> realm_apikey_list_copy(const realm_app_user_apikey_t api
             apikey_list + count,
             std::back_inserter(apikey_list_copy),
             [](const realm_app_user_apikey_t& apikey) {
-                return apikey_buf(apikey);
-            }
+            return apikey_buf(apikey);
+        }
         );
     }
     return apikey_list_copy;
@@ -362,24 +371,24 @@ RLM_API void realm_dart_apikey_list_callback(realm_userdata_t userdata, realm_ap
             apikey_list_buf.end(),
             std::back_inserter(apikey_list),
             [](const apikey_buf& apikey) {
-                return realm_app_user_apikey{
-                    apikey.id,
-                    apikey.key_buffer.c_str(),
-                    apikey.name_buffer.c_str(),
-                    apikey.disabled
-                };
-            }
+            return realm_app_user_apikey{
+                apikey.id,
+                apikey.key_buffer.c_str(),
+                apikey.name_buffer.c_str(),
+                apikey.disabled
+            };
+        }
         );
         (reinterpret_cast<realm_return_apikey_list_func_t>(ud->dart_callback))(ud->handle, apikey_list.data(), apikey_list.size(), error.get());
     });
 }
 
 RLM_API void realm_dart_return_string_callback(realm_userdata_t userdata, const char* serialized_ejson_response, const realm_app_error_t* error) {
     auto error_copy = realm_app_error_copy(error);
-    std::string buf{serialized_ejson_response ? serialized_ejson_response : ""};
+    std::string buf{ serialized_ejson_response ? serialized_ejson_response : "" };
 
     auto ud = reinterpret_cast<realm_dart_userdata_async_t>(userdata);
     ud->scheduler->invoke([ud, buf = std::move(buf), error = std::move(error_copy)]() mutable {
         (reinterpret_cast<realm_return_string_func_t>(ud->dart_callback))(ud->handle, buf.data(), error.get());
     });
-}
+}

src/realm_dart_sync.cpp

@@ -37,6 +37,8 @@ RLM_API void realm_dart_sync_connection_state_changed_callback(realm_userdata_t
                                                                realm_sync_connection_state_e old_state,
                                                                realm_sync_connection_state_e new_state);
 
+RLM_API void realm_dart_user_change_callback(realm_userdata_t userdata, realm_user_state_e state);
+
 RLM_API void realm_dart_sync_on_subscription_state_changed_callback(realm_userdata_t userdata, realm_flx_sync_subscription_set_state_e state);
 
 RLM_API bool realm_dart_sync_before_reset_handler_callback(realm_userdata_t userdata, realm_t* realm);

src/realm_dart_sync.h

@@ -16,8 +16,8 @@
 //
 ////////////////////////////////////////////////////////////////////////////////
 
+import 'dart:async';
 import 'dart:isolate';
-import 'dart:math';
 
 import 'package:test/expect.dart' hide throws;
 
@@ -489,4 +489,26 @@ void main() {
             .having((e) => e.statusCode, 'statusCode', 401)
             .having((e) => e.linkToServerLogs, 'linkToServerLogs', contains('logs?co_id='))));
   });
+
+  baasTest('User.logOut raises changes', (appConfig) async {
+    final app = App(appConfig);
+    final user = await getIntegrationUser(app);
+
+    expect(user.state, UserState.loggedIn);
+
+    final completer = Completer<UserChanges>();
+    final subscription = user.changes.listen((event) {
+      completer.complete(event);
+    });
+
+    await user.logOut();
+
+    expect(user.state, UserState.loggedOut);
+
+    final changeEvent = await completer.future.timeout(Duration(seconds: 15));
+    expect(changeEvent.user, user);
+    expect(changeEvent.user.state, UserState.loggedOut);
+
+    await subscription.cancel();
+  });
 }