Observe session database to be able to detect new user and removed user.

libraries/push/impl/src/main/kotlin/io/element/android/libraries/push/impl/userpushstore/UserPushStoreFactory.kt

@@ -17,16 +17,43 @@
 package io.element.android.libraries.push.impl.userpushstore
 
 import android.content.Context
+import io.element.android.libraries.di.AppScope
 import io.element.android.libraries.di.ApplicationContext
+import io.element.android.libraries.di.SingleIn
+import io.element.android.libraries.sessionstorage.api.observer.SessionListener
+import io.element.android.libraries.sessionstorage.api.observer.SessionObserver
 import javax.inject.Inject
 
+@SingleIn(AppScope::class)
 class UserPushStoreFactory @Inject constructor(
     @ApplicationContext private val context: Context,
-) {
+    private val sessionObserver: SessionObserver,
+) : SessionListener {
+    init {
+        observeSessions()
+    }
+
+    // We can have only one class accessing a single data store, so keep a cache of them.
+    private val cache = mutableMapOf<String, UserPushStore>()
     fun create(userId: String): UserPushStore {
-        return UserPushStoreDataStore(
+        return cache.getOrPut(userId) {
-            context = context,
+            UserPushStoreDataStore(
-            userId = userId
+                context = context,
-        )
+                userId = userId
+            )
+        }
+    }
+
+    private fun observeSessions() {
+        sessionObserver.addListener(this)
+    }
+
+    override suspend fun onSessionCreated(userId: String) {
+        // Nothing to do
+    }
+
+    override suspend fun onSessionDeleted(userId: String) {
+        // Delete the store
+        create(userId).reset()
     }
 }

libraries/session-storage/api/src/main/kotlin/io/element/android/libraries/sessionstorage/api/SessionStore.kt

@@ -17,9 +17,11 @@
 package io.element.android.libraries.sessionstorage.api
 
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.map
 
 interface SessionStore {
     fun isLoggedIn(): Flow<Boolean>
+    fun sessionsFlow(): Flow<List<SessionData>>
     suspend fun storeData(sessionData: SessionData)
     suspend fun getSession(sessionId: String): SessionData?
     suspend fun getAllSessions(): List<SessionData>
@@ -30,3 +32,7 @@ interface SessionStore {
 fun List<SessionData>.toUserList(): List<String> {
     return map { it.userId }
 }
+
+fun Flow<List<SessionData>>.toUserListFlow(): Flow<List<String>> {
+    return map { it.toUserList() }
+}

libraries/session-storage/api/src/main/kotlin/io/element/android/libraries/sessionstorage/api/observer/SessionListener.kt

@@ -0,0 +1,22 @@
+/*
+ * Copyright (c) 2023 New Vector Ltd
+ *
+ * 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.element.android.libraries.sessionstorage.api.observer
+
+interface SessionListener {
+    suspend fun onSessionCreated(userId: String)
+    suspend fun onSessionDeleted(userId: String)
+}

libraries/session-storage/api/src/main/kotlin/io/element/android/libraries/sessionstorage/api/observer/SessionObserver.kt

@@ -0,0 +1,22 @@
+/*
+ * Copyright (c) 2023 New Vector Ltd
+ *
+ * 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.element.android.libraries.sessionstorage.api.observer
+
+interface SessionObserver {
+    fun addListener(listener: SessionListener)
+    fun removeListener(listener: SessionListener)
+}

libraries/session-storage/impl-memory/src/main/kotlin/io/element/android/libraries/sessionstorage/impl/memory/InMemorySessionStore.kt

@@ -30,6 +30,10 @@ class InMemorySessionStore : SessionStore {
         return sessionDataFlow.map { it != null }
     }
 
+    override fun sessionsFlow(): Flow<List<SessionData>> {
+        return sessionDataFlow.map { listOfNotNull(it) }
+    }
+
     override suspend fun storeData(sessionData: SessionData) {
         sessionDataFlow.value = sessionData
     }

libraries/session-storage/impl/src/main/kotlin/io/element/android/libraries/sessionstorage/impl/DatabaseSessionStore.kt

@@ -18,6 +18,7 @@ package io.element.android.libraries.sessionstorage.impl
 
 import com.squareup.anvil.annotations.ContributesBinding
 import com.squareup.sqldelight.runtime.coroutines.asFlow
+import com.squareup.sqldelight.runtime.coroutines.mapToList
 import com.squareup.sqldelight.runtime.coroutines.mapToOneOrNull
 import io.element.android.libraries.di.AppScope
 import io.element.android.libraries.di.SingleIn
@@ -25,6 +26,7 @@ import io.element.android.libraries.sessionstorage.api.SessionData
 import io.element.android.libraries.sessionstorage.api.SessionStore
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.map
+import timber.log.Timber
 import javax.inject.Inject
 
 @SingleIn(AppScope::class)
@@ -34,7 +36,10 @@ class DatabaseSessionStore @Inject constructor(
 ) : SessionStore {
 
     override fun isLoggedIn(): Flow<Boolean> {
-        return database.sessionDataQueries.selectFirst().asFlow().mapToOneOrNull().map { it != null }
+        return database.sessionDataQueries.selectFirst()
+            .asFlow()
+            .mapToOneOrNull()
+            .map { it != null }
     }
 
     override suspend fun storeData(sessionData: SessionData) {
@@ -59,6 +64,14 @@ class DatabaseSessionStore @Inject constructor(
             .map { it.toApiModel() }
     }
 
+    override fun sessionsFlow(): Flow<List<SessionData>> {
+        Timber.w("Observing session list!")
+        return database.sessionDataQueries.selectAll()
+            .asFlow()
+            .mapToList()
+            .map { it.map { sessionData -> sessionData.toApiModel() } }
+    }
+
     override suspend fun removeSession(sessionId: String) {
         database.sessionDataQueries.removeSession(sessionId)
     }

libraries/session-storage/impl/src/main/kotlin/io/element/android/libraries/sessionstorage/impl/observer/DefaultSessionObserver.kt

@@ -0,0 +1,91 @@
+/*
+ * Copyright (c) 2023 New Vector Ltd
+ *
+ * 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.element.android.libraries.sessionstorage.impl.observer
+
+import com.squareup.anvil.annotations.ContributesBinding
+import io.element.android.libraries.core.coroutine.CoroutineDispatchers
+import io.element.android.libraries.di.AppScope
+import io.element.android.libraries.di.SingleIn
+import io.element.android.libraries.sessionstorage.api.SessionStore
+import io.element.android.libraries.sessionstorage.api.observer.SessionListener
+import io.element.android.libraries.sessionstorage.api.observer.SessionObserver
+import io.element.android.libraries.sessionstorage.api.toUserListFlow
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
+import java.util.concurrent.CopyOnWriteArraySet
+import javax.inject.Inject
+
+@SingleIn(AppScope::class)
+@ContributesBinding(AppScope::class)
+class DefaultSessionObserver @Inject constructor(
+    private val sessionStore: SessionStore,
+    private val coroutineScope: CoroutineScope,
+    private val dispatchers: CoroutineDispatchers,
+) : SessionObserver {
+    // Keep only the userId
+    private var currentUsers: Set<String>? = null
+
+    init {
+        observeDatabase()
+    }
+
+    private val listeners = CopyOnWriteArraySet<SessionListener>()
+    override fun addListener(listener: SessionListener) {
+        listeners.add(listener)
+    }
+
+    override fun removeListener(listener: SessionListener) {
+        listeners.remove(listener)
+    }
+
+    private fun observeDatabase() {
+        coroutineScope.launch {
+            withContext(dispatchers.io) {
+                sessionStore.sessionsFlow()
+                    .toUserListFlow()
+                    .map { it.toSet() }
+                    .onEach { newUserSet ->
+                        val currentUserSet = currentUsers
+                        if (currentUserSet != null) {
+                            // Compute diff
+                            // Removed user
+                            val removedUsers = currentUserSet - newUserSet
+                            removedUsers.forEach { removedUser ->
+                                listeners.onEach { listener ->
+                                    listener.onSessionDeleted(removedUser)
+                                }
+                            }
+                            // Added user
+                            val addedUsers = newUserSet - currentUserSet
+                            addedUsers.forEach { addedUser ->
+                                listeners.onEach { listener ->
+                                    listener.onSessionDeleted(addedUser)
+                                }
+                            }
+                        }
+
+                        currentUsers = newUserSet
+                    }
+                    .collect()
+            }
+        }
+    }
+}