observe unread events for notifications from rooms which are not muted

domains/store/src/main/kotlin/app/dapk/st/domain/sync/RoomPersistence.kt

@@ -8,6 +8,7 @@ import app.dapk.st.domain.room.MutedStorePersistence
 import app.dapk.st.matrix.common.EventId
 import app.dapk.st.matrix.common.RoomId
 import app.dapk.st.matrix.sync.*
+import com.squareup.sqldelight.Query
 import com.squareup.sqldelight.runtime.coroutines.asFlow
 import com.squareup.sqldelight.runtime.coroutines.mapToList
 import com.squareup.sqldelight.runtime.coroutines.mapToOneNotNull
@@ -56,10 +57,8 @@ internal class RoomPersistence(
         }.distinctUntilChanged()
 
         return database.roomEventQueries.selectRoom(roomId.value)
-            .asFlow()
+            .distinctFlowList()
-            .mapToList()
             .map { it.map { json.decodeFromString(RoomEvent.serializer(), it) } }
-            .distinctUntilChanged()
             .combine(overviewFlow) { events, overview ->
                 RoomState(overview, events)
             }
@@ -91,9 +90,7 @@ internal class RoomPersistence(
 
     override fun observeUnread(): Flow<Map<RoomOverview, List<RoomEvent>>> {
         return database.roomEventQueries.selectAllUnread()
-            .asFlow()
+            .distinctFlowList()
-            .mapToList()
-            .distinctUntilChanged()
             .map {
                 it.groupBy { RoomId(it.room_id) }
                     .mapKeys { overviewPersistence.retrieve(it.key)!! }
@@ -115,6 +112,22 @@ internal class RoomPersistence(
             }
     }
 
+    override fun observeNotMutedUnread(): Flow<Map<RoomOverview, List<RoomEvent>>> {
+        return database.roomEventQueries.selectNotMutedUnread()
+            .distinctFlowList()
+            .map {
+                it.groupBy { RoomId(it.room_id) }
+                    .mapKeys { overviewPersistence.retrieve(it.key)!! }
+                    .mapValues {
+                        it.value.map {
+                            json.decodeFromString(RoomEvent.serializer(), it.blob)
+                        }
+                    }
+            }
+    }
+
+    private fun <T : Any> Query<T>.distinctFlowList() = this.asFlow().mapToList().distinctUntilChanged()
+
     override suspend fun markRead(roomId: RoomId) {
         coroutineDispatchers.withIoContext {
             database.unreadEventQueries.removeRead(room_id = roomId.value)

domains/store/src/main/sqldelight/app/dapk/db/model/RoomEvent.sq

@@ -34,6 +34,16 @@ INNER JOIN dbRoomEvent ON dbUnreadEvent.event_id = dbRoomEvent.event_id
 ORDER BY dbRoomEvent.timestamp_utc DESC
 LIMIT 100;
 
+selectNotMutedUnread:
+SELECT dbRoomEvent.blob, dbRoomEvent.room_id
+FROM dbUnreadEvent
+INNER JOIN dbRoomEvent ON dbUnreadEvent.event_id = dbRoomEvent.event_id
+LEFT OUTER JOIN dbMutedRoom
+    ON dbUnreadEvent.room_id = dbMutedRoom.room_id
+    WHERE dbMutedRoom.room_id IS NULL
+ORDER BY dbRoomEvent.timestamp_utc DESC
+LIMIT 100;
+
 remove:
 DELETE FROM dbRoomEvent
 WHERE room_id = ?;

matrix-chat-engine/src/main/kotlin/app/dapk/st/engine/ObserveUnreadNotificationsUseCaseImpl.kt

@@ -16,7 +16,7 @@ internal typealias ObserveUnreadNotificationsUseCase = () -> Flow<UnreadNotifica
 class ObserveUnreadNotificationsUseCaseImpl(private val roomStore: RoomStore) : ObserveUnreadNotificationsUseCase {
 
     override fun invoke(): Flow<UnreadNotifications> {
-        return roomStore.observeUnread()
+        return roomStore.observeNotMutedUnread()
             .mapWithDiff()
             .avoidShowingPreviousNotificationsOnLaunch()
             .onlyRenderableChanges()

matrix/services/sync/src/main/kotlin/app/dapk/st/matrix/sync/Store.kt

@@ -16,6 +16,7 @@ interface RoomStore : MuteableStore {
     suspend fun markRead(roomId: RoomId)
     fun observeUnread(): Flow<Map<RoomOverview, List<RoomEvent>>>
     fun observeUnreadCountById(): Flow<Map<RoomId, Int>>
+    fun observeNotMutedUnread(): Flow<Map<RoomOverview, List<RoomEvent>>>
     fun observeEvent(eventId: EventId): Flow<EventId>
     suspend fun findEvent(eventId: EventId): RoomEvent?