Add support for authentication flows

CHANGELOG.md

@@ -10,6 +10,7 @@
 * Default log level has been changed from `LogLevel.WARN` to `LogLevel.INFO`.
 * Avoid tracking unreferenced realm versions through the garbage collector. (Issue [#1234](https://github.com/realm/realm-kotlin/issues/1234))
 * [Sync] All tokens, passwords and custom function arguments are now obfuscated by default, even if `LogLevel` is set to DEBUG, TRACE or ALL. (Issue [#410](https://github.com/realm/realm-kotlin/issues/410))
+* [Sync] Add support for `App.authenticationChangeAsFlow()` which make it possible to listen to authentication changes like "LoggedIn", "LoggedOut" and "Removed" across all users of the app. (Issue [#749](https://github.com/realm/realm-kotlin/issues/749)).
 * [Sync] `@PersistedName` is now also supported on model classes. (Issue [#1138](https://github.com/realm/realm-kotlin/issues/1138))
 
 ### Fixed

packages/library-sync/src/commonMain/kotlin/io/realm/kotlin/mongodb/App.kt

@@ -23,6 +23,7 @@ import io.realm.kotlin.mongodb.exceptions.AuthException
 import io.realm.kotlin.mongodb.exceptions.InvalidCredentialsException
 import io.realm.kotlin.mongodb.internal.AppConfigurationImpl
 import io.realm.kotlin.mongodb.internal.AppImpl
+import kotlinx.coroutines.flow.Flow
 
 /**
  * An **App** is the main client-side entry point for interacting with an **Atlas App Services
@@ -101,6 +102,14 @@ public interface App {
      */
     public suspend fun login(credentials: Credentials): User
 
+    /**
+     * Create a [Flow] of [AuthenticationChange]-events to receive notifications of updates to all
+     * app user authentication states: login, logout and removal.
+     *
+     * @return a [Flow] of authentication events for users associated with this app.
+     */
+    public fun authenticationChangeAsFlow(): Flow<AuthenticationChange>
+
     /**
      * Close the app instance and release all underlying resources.
      *

packages/library-sync/src/commonMain/kotlin/io/realm/kotlin/mongodb/AuthenticationChange.kt

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2023 Realm Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.realm.kotlin.mongodb
+
+import kotlinx.coroutines.flow.Flow
+
+/**
+ * This sealed class describe the possible events that can be observed on the [Flow] created by
+ * calling [App.authenticationChangeAsFlow].
+ *
+ * The specific states are represented by these subclasses [LoggedIn], [LoggedOut] and
+ * [Removed].
+ *
+ * Changes can thus be consumed this way:
+ *
+ * ```
+ * app.authenticationChangeAsFlow().collect { change: AuthenticationChange ->
+ *       when(change) {
+ *          is LoggedIn -> handleLogin(change.user)
+ *          is LoggedOut -> handleLogOut(change.user)
+ *          is Removed -> handleRemove(change.user)
+ *       }
+ *   }
+ * ```
+ */
+public sealed interface AuthenticationChange {
+    /**
+     * A reference to the [User] this event happened to.
+     *
+     * *Warning:* This is the live user object, so the [User.state] might have diverged from the
+     * event it is associated with, i.e. if a users logs out and back in while the event is
+     * propagating, the state of the user might be [User.State.LOGGED_IN], even though it was
+     * reported as a [LoggedOut] event.
+     */
+    public val user: User
+}
+
+/**
+ * Event emitted when a user logs into the app.
+ */
+public interface LoggedIn : AuthenticationChange
+
+/**
+ * Event emitted when a user is logged out.
+ */
+public interface LoggedOut : AuthenticationChange
+
+/**
+ * Event emitted when a user is removed, which also logs them out.
+ */
+public interface Removed : AuthenticationChange

packages/library-sync/src/commonMain/kotlin/io/realm/kotlin/mongodb/User.kt

@@ -128,6 +128,8 @@ public interface User {
      *
      * @throws io.realm.kotlin.mongodb.exceptions.ServiceException if a failure occurred when
      * communicating with App Services. See [AppException] for details.
+     * @throws IllegalStateException if a consumer listening to [App.authenticationChangeAsFlow]
+     * is too slow consuming events.
      */
     // FIXME add references to allUsers and remove when ready
     //     * All other users will be marked as [User.State.LOGGED_OUT]
@@ -147,6 +149,8 @@ public interface User {
      * @throws IllegalStateException if the user was already removed.
      * @throws io.realm.kotlin.mongodb.exceptions.ServiceException if a failure occurred when
      * communicating with App Services. See [AppException] for details.
+     * @throws IllegalStateException if a consumer listening to [App.authenticationChangeAsFlow]
+     * is too slow consuming events.
      */
     // TODO Document how this method behave if offline
     public suspend fun remove(): User
@@ -163,6 +167,8 @@ public interface User {
      * @throws IllegalStateException if the user was already removed or not logged in.
      * @throws io.realm.kotlin.mongodb.exceptions.ServiceException if a failure occurred when
      * communicating with App Services. See [AppException] for details.
+     * @throws IllegalStateException if a consumer listening to [App.authenticationChangeAsFlow]
+     * is too slow consuming events.
      */
     public suspend fun delete()
 

packages/library-sync/src/commonMain/kotlin/io/realm/kotlin/mongodb/internal/AppImpl.kt

@@ -25,10 +25,14 @@ import io.realm.kotlin.internal.interop.sync.NetworkTransport
 import io.realm.kotlin.internal.util.Validation
 import io.realm.kotlin.internal.util.use
 import io.realm.kotlin.mongodb.App
+import io.realm.kotlin.mongodb.AuthenticationChange
 import io.realm.kotlin.mongodb.Credentials
 import io.realm.kotlin.mongodb.User
 import io.realm.kotlin.mongodb.auth.EmailPasswordAuth
+import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
 
 // TODO Public due to being a transitive dependency to UserImpl
 public class AppImpl(
@@ -38,6 +42,17 @@ public class AppImpl(
     internal val nativePointer: RealmAppPointer
     private val networkTransport: NetworkTransport
 
+    // Allow some delay between events being reported and them being consumed.
+    // When the (somewhat arbitrary) limit is hit, we will throw an exception, since we assume the
+    // consumer is doing something wrong. This is also needed because we don't
+    // want to block user events like logout, delete and remove.
+    @Suppress("MagicNumber")
+    private val authenticationChangeFlow = MutableSharedFlow<AuthenticationChange>(
+        replay = 0,
+        extraBufferCapacity = 8,
+        onBufferOverflow = BufferOverflow.SUSPEND
+    )
+
     init {
         val appResources: Pair<NetworkTransport, NativePointer<RealmAppT>> = configuration.createNativeApp()
         networkTransport = appResources.first
@@ -73,8 +88,29 @@ public class AppImpl(
                 }
             )
             return channel.receive()
-                .getOrThrow()
+                .getOrThrow().also { user: User ->
+                    reportAuthenticationChange(user, User.State.LOGGED_IN)
+                }
+        }
+    }
+
+    internal fun reportAuthenticationChange(user: User, change: User.State) {
+        val event: AuthenticationChange = when (change) {
+            User.State.LOGGED_OUT -> LoggedOutImpl(user)
+            User.State.LOGGED_IN -> LoggedInImpl(user)
+            User.State.REMOVED -> RemovedImpl(user)
         }
+        if (!authenticationChangeFlow.tryEmit(event)) {
+            throw IllegalStateException(
+                "It wasn't possible to emit authentication changes " +
+                    "because a consuming flow was blocked. Increase dispatcher processing resources " +
+                    "or buffer `App.authenticationChangeAsFlow()` with buffer(...)."
+            )
+        }
+    }
+
+    override fun authenticationChangeAsFlow(): Flow<AuthenticationChange> {
+        return authenticationChangeFlow
     }
 
     override fun close() {

packages/library-sync/src/commonMain/kotlin/io/realm/kotlin/mongodb/internal/AuthenticationChangeImpl.kt

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2023 Realm Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.realm.kotlin.mongodb.internal
+
+import io.realm.kotlin.mongodb.LoggedIn
+import io.realm.kotlin.mongodb.LoggedOut
+import io.realm.kotlin.mongodb.Removed
+import io.realm.kotlin.mongodb.User
+
+internal class LoggedInImpl(override val user: User) : LoggedIn
+internal class LoggedOutImpl(override val user: User) : LoggedOut
+internal class RemovedImpl(override val user: User) : Removed

packages/library-sync/src/commonMain/kotlin/io/realm/kotlin/mongodb/internal/UserImpl.kt

@@ -93,30 +93,48 @@ public class UserImpl(
         }
 
     override suspend fun logOut() {
-        Channel<Result<Unit>>(1).use { channel ->
+        Channel<Result<User.State?>>(1).use { channel ->
+            val reportLoggedOut = loggedIn
             RealmInterop.realm_app_log_out(
                 app.nativePointer,
                 nativePointer,
-                channelResultCallback<Unit, Unit>(channel) {
-                    // No-op
+                channelResultCallback<Unit, User.State?>(channel) {
+                    if (reportLoggedOut) {
+                        User.State.LOGGED_OUT
+                    } else {
+                        null
+                    }
                 }
             )
             return@use channel.receive()
-                .getOrThrow()
+                .getOrThrow().also { state: User.State? ->
+                    if (state != null) {
+                        app.reportAuthenticationChange(this, state)
+                    }
+                }
         }
     }
 
     override suspend fun remove(): User {
-        Channel<Result<Unit>>(1).use { channel ->
+        Channel<Result<User.State?>>(1).use { channel ->
+            val reportRemoved = loggedIn
             RealmInterop.realm_app_remove_user(
                 app.nativePointer,
                 nativePointer,
-                channelResultCallback<Unit, Unit>(channel) {
-                    // No-op
+                channelResultCallback<Unit, User.State?>(channel) {
+                    if (reportRemoved) {
+                        User.State.REMOVED
+                    } else {
+                        null
+                    }
                 }
             )
             return@use channel.receive()
-                .getOrThrow()
+                .getOrThrow().also { state: User.State? ->
+                    if (state != null) {
+                        app.reportAuthenticationChange(this, state)
+                    }
+                }
         }
         return this
     }
@@ -125,16 +143,18 @@ public class UserImpl(
         if (state != User.State.LOGGED_IN) {
             throw IllegalStateException("User must be logged in, in order to be deleted.")
         }
-        Channel<Result<Unit>>(1).use { channel ->
+        Channel<Result<User.State>>(1).use { channel ->
             RealmInterop.realm_app_delete_user(
                 app.nativePointer,
                 nativePointer,
-                channelResultCallback<Unit, Unit>(channel) {
-                    // No-op
+                channelResultCallback<Unit, User.State>(channel) {
+                    User.State.REMOVED
                 }
             )
             return@use channel.receive()
-                .getOrThrow()
+                .getOrThrow().also { state: User.State ->
+                    app.reportAuthenticationChange(this, state)
+                }
         }
     }