Merge pull request #250 from ouchadam/tech/engine-tests

Tech/engine tests

chat-engine/src/testFixtures/kotlin/fixture/Fixtures.kt

@@ -63,4 +63,15 @@ fun anImageMeta(
 fun aRoomState(
     roomOverview: RoomOverview = aRoomOverview(),
     events: List<RoomEvent> = listOf(aRoomMessageEvent()),
-) = RoomState(roomOverview, events)
+) = RoomState(roomOverview, events)
+
+fun aRoomInvite(
+    from: RoomMember = aRoomMember(),
+    roomId: RoomId = aRoomId(),
+    inviteMeta: RoomInvite.InviteMeta = RoomInvite.InviteMeta.DirectMessage,
+) = RoomInvite(from, roomId, inviteMeta)
+
+fun aTypingEvent(
+    roomId: RoomId = aRoomId(),
+    members: List<RoomMember> = listOf(aRoomMember())
+) = Typing(roomId, members)

core/src/main/kotlin/app/dapk/st/core/extensions/FlowExtensions.kt

@@ -20,3 +20,26 @@ suspend fun <T> Flow<T>.firstOrNull(count: Int, predicate: suspend (T) -> Boolea
 
     return result
 }
+
+inline fun <T1, T2, T3, T4, T5, T6, R> combine(
+    flow: Flow<T1>,
+    flow2: Flow<T2>,
+    flow3: Flow<T3>,
+    flow4: Flow<T4>,
+    flow5: Flow<T5>,
+    flow6: Flow<T6>,
+    crossinline transform: suspend (T1, T2, T3, T4, T5, T6) -> R
+): Flow<R> {
+    return kotlinx.coroutines.flow.combine(flow, flow2, flow3, flow4, flow5, flow6) { args: Array<*> ->
+        @Suppress("UNCHECKED_CAST")
+        transform(
+            args[0] as T1,
+            args[1] as T2,
+            args[2] as T3,
+            args[3] as T4,
+            args[4] as T5,
+            args[5] as T6,
+        )
+    }
+}
+

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

@@ -0,0 +1,53 @@
+package app.dapk.st.engine
+
+import app.dapk.st.matrix.common.RoomId
+import app.dapk.st.matrix.common.RoomMember
+import app.dapk.st.matrix.common.UserId
+import app.dapk.st.matrix.common.asString
+import app.dapk.st.matrix.message.MessageService
+import app.dapk.st.matrix.room.RoomService
+
+internal typealias DirectoryMergeWithLocalEchosUseCase = suspend (OverviewState, UserId, Map<RoomId, List<MessageService.LocalEcho>>) -> OverviewState
+
+internal class DirectoryMergeWithLocalEchosUseCaseImpl(
+    private val roomService: RoomService,
+) : DirectoryMergeWithLocalEchosUseCase {
+
+    override suspend fun invoke(overview: OverviewState, selfId: UserId, echos: Map<RoomId, List<MessageService.LocalEcho>>): OverviewState {
+        return when {
+            echos.isEmpty() -> overview
+            else -> overview.map {
+                when (val roomEchos = echos[it.roomId]) {
+                    null -> it
+                    else -> it.mergeWithLocalEchos(
+                        member = roomService.findMember(it.roomId, selfId) ?: RoomMember(
+                            selfId,
+                            null,
+                            avatarUrl = null,
+                        ),
+                        echos = roomEchos,
+                    )
+                }
+            }
+        }
+    }
+
+    private fun RoomOverview.mergeWithLocalEchos(member: RoomMember, echos: List<MessageService.LocalEcho>): RoomOverview {
+        val latestEcho = echos.maxByOrNull { it.timestampUtc }
+        return if (latestEcho != null && latestEcho.timestampUtc > (this.lastMessage?.utcTimestamp ?: 0)) {
+            this.copy(
+                lastMessage = RoomOverview.LastMessage(
+                    content = when (val message = latestEcho.message) {
+                        is MessageService.Message.TextMessage -> message.content.body.asString()
+                        is MessageService.Message.ImageMessage -> "\uD83D\uDCF7"
+                    },
+                    utcTimestamp = latestEcho.timestampUtc,
+                    author = member,
+                )
+            )
+        } else {
+            this
+        }
+    }
+
+}

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

@@ -1,31 +1,35 @@
 package app.dapk.st.engine
 
-import app.dapk.st.matrix.common.*
+import app.dapk.st.core.extensions.combine
+import app.dapk.st.matrix.common.CredentialsStore
 import app.dapk.st.matrix.message.MessageService
-import app.dapk.st.matrix.room.RoomService
 import app.dapk.st.matrix.sync.RoomStore
 import app.dapk.st.matrix.sync.SyncService
 import app.dapk.st.matrix.sync.SyncService.SyncEvent.Typing
-import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flatMapConcat
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.map
 
 internal class DirectoryUseCase(
     private val syncService: SyncService,
     private val messageService: MessageService,
-    private val roomService: RoomService,
     private val credentialsStore: CredentialsStore,
     private val roomStore: RoomStore,
+    private val mergeLocalEchosUseCase: DirectoryMergeWithLocalEchosUseCase,
 ) {
 
     fun state(): Flow<DirectoryState> {
-        return flow { emit(credentialsStore.credentials()!!.userId) }.flatMapMerge { userId ->
+        return flow { emit(credentialsStore.credentials()!!.userId) }.flatMapConcat { userId ->
             combine(
-                overviewDatasource(),
+                syncService.startSyncing(),
+                syncService.overview().map { it.map { it.engine() } },
                 messageService.localEchos(),
                 roomStore.observeUnreadCountById(),
                 syncService.events(),
                 roomStore.observeMuted(),
-            ) { overviewState, localEchos, unread, events, muted ->
+            ) { _, overviewState, localEchos, unread, events, muted ->
-                overviewState.mergeWithLocalEchos(localEchos, userId).map { roomOverview ->
+                mergeLocalEchosUseCase.invoke(overviewState, userId, localEchos).map { roomOverview ->
                     DirectoryItem(
                         overview = roomOverview,
                         unreadCount = UnreadCount(unread[roomOverview.roomId] ?: 0),
@@ -36,50 +40,4 @@ internal class DirectoryUseCase(
             }
         }
     }
-
-    private fun overviewDatasource() = combine(
-        syncService.startSyncing(),
-        syncService.overview().map { it.map { it.engine() } }
-    ) { _, overview -> overview }.filterNotNull()
-
-    private suspend fun OverviewState.mergeWithLocalEchos(localEchos: Map<RoomId, List<MessageService.LocalEcho>>, userId: UserId): OverviewState {
-        return when {
-            localEchos.isEmpty() -> this
-            else -> this.map {
-                when (val roomEchos = localEchos[it.roomId]) {
-                    null -> it
-                    else -> it.mergeWithLocalEchos(
-                        member = roomService.findMember(it.roomId, userId) ?: RoomMember(
-                            userId,
-                            null,
-                            avatarUrl = null,
-                        ),
-                        echos = roomEchos,
-                    )
-                }
-            }
-        }
-    }
-
-    private fun RoomOverview.mergeWithLocalEchos(member: RoomMember, echos: List<MessageService.LocalEcho>): RoomOverview {
-        val latestEcho = echos.maxByOrNull { it.timestampUtc }
-        return if (latestEcho != null && latestEcho.timestampUtc > (this.lastMessage?.utcTimestamp ?: 0)) {
-            this.copy(
-                lastMessage = RoomOverview.LastMessage(
-                    content = when (val message = latestEcho.message) {
-                        is MessageService.Message.TextMessage -> message.content.body.asString()
-                        is MessageService.Message.ImageMessage -> "\uD83D\uDCF7"
-                    },
-                    utcTimestamp = latestEcho.timestampUtc,
-                    author = member,
-                )
-            )
-        } else {
-            this
-        }
-    }
-
 }
-
-
-

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

@@ -2,7 +2,6 @@ package app.dapk.st.engine
 
 import app.dapk.st.matrix.sync.SyncService
 import kotlinx.coroutines.flow.combine
-import kotlinx.coroutines.flow.filterNotNull
 import kotlinx.coroutines.flow.map
 
 class InviteUseCase(
@@ -14,6 +13,6 @@ class InviteUseCase(
     private fun invitesDatasource() = combine(
         syncService.startSyncing(),
         syncService.invites().map { it.map { it.engine() } }
-    ) { _, invites -> invites }.filterNotNull()
+    ) { _, invites -> invites }
 
 }

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

@@ -3,6 +3,7 @@ package app.dapk.st.engine
 import app.dapk.st.core.Base64
 import app.dapk.st.core.BuildMeta
 import app.dapk.st.core.CoroutineDispatchers
+import app.dapk.st.core.JobBag
 import app.dapk.st.core.extensions.ErrorTracker
 import app.dapk.st.matrix.MatrixClient
 import app.dapk.st.matrix.MatrixTaskRunner
@@ -172,14 +173,14 @@ class MatrixEngine internal constructor(
                 DirectoryUseCase(
                     matrix.syncService(),
                     matrix.messageService(),
-                    matrix.roomService(),
                     credentialsStore,
-                    roomStore
+                    roomStore,
+                    DirectoryMergeWithLocalEchosUseCaseImpl(matrix.roomService()),
                 )
             }
             val timelineUseCase = unsafeLazy {
                 val matrix = lazyMatrix.value
-                val mergeWithLocalEchosUseCase = MergeWithLocalEchosUseCaseImpl(LocalEchoMapper(MetaMapper()))
+                val mergeWithLocalEchosUseCase = TimelineMergeWithLocalEchosUseCaseImpl(LocalEchoMapper(MetaMapper()))
                 val timeline = TimelineUseCaseImpl(matrix.syncService(), matrix.messageService(), matrix.roomService(), mergeWithLocalEchosUseCase)
                 ReadMarkingTimeline(roomStore, credentialsStore, timeline, matrix.roomService())
             }
@@ -190,7 +191,16 @@ class MatrixEngine internal constructor(
             }
 
             val mediaDecrypter = unsafeLazy { MatrixMediaDecrypter(base64) }
-            val pushHandler = unsafeLazy { MatrixPushHandler(backgroundScheduler, credentialsStore, lazyMatrix.value.syncService(), roomStore) }
+            val pushHandler = unsafeLazy {
+                MatrixPushHandler(
+                    backgroundScheduler,
+                    credentialsStore,
+                    lazyMatrix.value.syncService(),
+                    roomStore,
+                    coroutineDispatchers,
+                    JobBag(),
+                )
+            }
 
             val invitesUseCase = unsafeLazy { InviteUseCase(lazyMatrix.value.syncService()) }
 

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

@@ -1,6 +1,8 @@
 package app.dapk.st.engine
 
 import app.dapk.st.core.AppLogTag
+import app.dapk.st.core.CoroutineDispatchers
+import app.dapk.st.core.JobBag
 import app.dapk.st.core.log
 import app.dapk.st.matrix.common.CredentialsStore
 import app.dapk.st.matrix.common.EventId
@@ -9,17 +11,20 @@ import app.dapk.st.matrix.common.RoomId
 import app.dapk.st.matrix.message.BackgroundScheduler
 import app.dapk.st.matrix.sync.RoomStore
 import app.dapk.st.matrix.sync.SyncService
-import kotlinx.coroutines.*
+import kotlinx.coroutines.Job
-import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.firstOrNull
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withTimeoutOrNull
 
-private var previousJob: Job? = null
-
-@OptIn(DelicateCoroutinesApi::class)
 class MatrixPushHandler(
     private val backgroundScheduler: BackgroundScheduler,
     private val credentialsStore: CredentialsStore,
     private val syncService: SyncService,
     private val roomStore: RoomStore,
+    private val dispatchers: CoroutineDispatchers,
+    private val jobBag: JobBag,
 ) : PushHandler {
 
     override fun onNewToken(payload: JsonString) {
@@ -35,13 +40,12 @@ class MatrixPushHandler(
 
     override fun onMessageReceived(eventId: EventId?, roomId: RoomId?) {
         log(AppLogTag.PUSH, "push received")
-        previousJob?.cancel()
+        jobBag.replace(MatrixPushHandler::class, dispatchers.global.launch {
-        previousJob = GlobalScope.launch {
             when (credentialsStore.credentials()) {
                 null -> log(AppLogTag.PUSH, "push ignored due to missing api credentials")
                 else -> doSync(roomId, eventId)
             }
-        }
+        })
     }
 
     private suspend fun doSync(roomId: RoomId?, eventId: EventId?) {

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

@@ -4,11 +4,11 @@ import app.dapk.st.matrix.common.EventId
 import app.dapk.st.matrix.common.RoomMember
 import app.dapk.st.matrix.message.MessageService
 
-internal typealias MergeWithLocalEchosUseCase = (RoomState, RoomMember, List<MessageService.LocalEcho>) -> RoomState
+internal typealias TimelineMergeWithLocalEchosUseCase = (RoomState, RoomMember, List<MessageService.LocalEcho>) -> RoomState
 
-internal class MergeWithLocalEchosUseCaseImpl(
+internal class TimelineMergeWithLocalEchosUseCaseImpl(
     private val localEventMapper: LocalEchoMapper,
-) : MergeWithLocalEchosUseCase {
+) : TimelineMergeWithLocalEchosUseCase {
 
     override fun invoke(roomState: RoomState, member: RoomMember, echos: List<MessageService.LocalEcho>): RoomState {
         val echosByEventId = echos.associateBy { it.eventId }

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

@@ -5,9 +5,7 @@ import app.dapk.st.matrix.common.EventId
 import app.dapk.st.matrix.common.RoomId
 import app.dapk.st.matrix.common.UserId
 import app.dapk.st.matrix.room.RoomService
-import app.dapk.st.matrix.sync.RoomEvent
 import app.dapk.st.matrix.sync.RoomStore
-import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.async
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.*
@@ -24,7 +22,7 @@ class ReadMarkingTimeline(
             val credentials = credentialsStore.credentials()!!
             roomStore.markRead(roomId)
             emit(credentials)
-        }.flatMapMerge { credentials ->
+        }.flatMapConcat { credentials ->
             var lastKnownReadEvent: EventId? = null
             observeTimelineUseCase.invoke(roomId, credentials.userId).distinctUntilChanged().onEach { state ->
                 state.latestMessageEventFromOthers(self = credentials.userId)?.let {
@@ -37,8 +35,9 @@ class ReadMarkingTimeline(
         }
     }
 
-    private suspend fun updateRoomReadStateAsync(latestReadEvent: EventId, state: MessengerPageState, isReadReceiptsDisabled: Boolean): Deferred<*> {
+    @Suppress("DeferredResultUnused")
-        return coroutineScope {
+    private suspend fun updateRoomReadStateAsync(latestReadEvent: EventId, state: MessengerPageState, isReadReceiptsDisabled: Boolean) {
+        coroutineScope {
             async {
                 runCatching {
                     roomService.markFullyRead(state.roomState.roomOverview.roomId, latestReadEvent, isPrivate = isReadReceiptsDisabled)
@@ -48,10 +47,9 @@ class ReadMarkingTimeline(
         }
     }
 
-}
+    private fun MessengerPageState.latestMessageEventFromOthers(self: UserId) = this.roomState.events
-
+        .filterIsInstance<RoomEvent.Message>()
-private fun MessengerPageState.latestMessageEventFromOthers(self: UserId) = this.roomState.events
+        .filterNot { it.author.id == self }
-    .filterIsInstance<RoomEvent.Message>()
+        .firstOrNull()
-    .filterNot { it.author.id == self }
+        ?.eventId
-    .firstOrNull()
+}
-    ?.eventId

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

@@ -16,7 +16,7 @@ internal class TimelineUseCaseImpl(
     private val syncService: SyncService,
     private val messageService: MessageService,
     private val roomService: RoomService,
-    private val mergeWithLocalEchosUseCase: MergeWithLocalEchosUseCase
+    private val timelineMergeWithLocalEchosUseCase: TimelineMergeWithLocalEchosUseCase,
 ) : ObserveTimelineUseCase {
 
     override fun invoke(roomId: RoomId, userId: UserId): Flow<MessengerPageState> {
@@ -30,7 +30,7 @@ internal class TimelineUseCaseImpl(
                 roomState = when {
                     localEchos.isEmpty() -> roomState
                     else -> {
-                        mergeWithLocalEchosUseCase.invoke(
+                        timelineMergeWithLocalEchosUseCase.invoke(
                             roomState,
                             roomService.findMember(roomId, userId) ?: userId.toFallbackMember(),
                             localEchos,

matrix-chat-engine/src/test/kotlin/app/dapk/st/engine/DirectoryUseCaseTest.kt

@@ -0,0 +1,115 @@
+package app.dapk.st.engine
+
+import app.dapk.st.matrix.common.RoomId
+import app.dapk.st.matrix.common.RoomMember
+import app.dapk.st.matrix.common.UserId
+import app.dapk.st.matrix.message.MessageService
+import app.dapk.st.matrix.sync.RoomOverview
+import fake.FakeCredentialsStore
+import fake.FakeRoomStore
+import fake.FakeSyncService
+import fixture.aMatrixRoomOverview
+import fixture.aRoomMember
+import fixture.aTypingEvent
+import fixture.aUserCredentials
+import io.mockk.coEvery
+import io.mockk.mockk
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.test.runTest
+import org.amshove.kluent.shouldBeEqualTo
+import org.junit.Test
+import test.delegateReturn
+
+private val A_ROOM_OVERVIEW = aMatrixRoomOverview()
+private const val AN_UNREAD_COUNT = 10
+private const val MUTED_ROOM = true
+private val TYPING_MEMBERS = listOf(aRoomMember())
+
+class DirectoryUseCaseTest {
+
+    private val fakeSyncService = FakeSyncService()
+    private val fakeMessageService = FakeMessageService()
+    private val fakeCredentialsStore = FakeCredentialsStore()
+    private val fakeRoomStore = FakeRoomStore()
+    private val fakeMergeLocalEchosUseCase = FakeDirectoryMergeWithLocalEchosUseCase()
+
+    private val useCase = DirectoryUseCase(
+        fakeSyncService,
+        fakeMessageService,
+        fakeCredentialsStore,
+        fakeRoomStore,
+        fakeMergeLocalEchosUseCase,
+    )
+
+    @Test
+    fun `given empty values, then reads default directory state and maps to engine`() = runTest {
+        givenEmitsDirectoryState(
+            A_ROOM_OVERVIEW,
+            unreadCount = null,
+            isMuted = false,
+        )
+
+        val result = useCase.state().first()
+
+        result shouldBeEqualTo listOf(
+            DirectoryItem(
+                A_ROOM_OVERVIEW.engine(),
+                unreadCount = UnreadCount(0),
+                typing = null,
+                isMuted = false
+            )
+        )
+    }
+
+    @Test
+    fun `given extra state, then reads directory state and maps to engine`() = runTest {
+        givenEmitsDirectoryState(
+            A_ROOM_OVERVIEW,
+            unreadCount = AN_UNREAD_COUNT,
+            isMuted = MUTED_ROOM,
+            typing = TYPING_MEMBERS
+        )
+
+        val result = useCase.state().first()
+
+        result shouldBeEqualTo listOf(
+            DirectoryItem(
+                A_ROOM_OVERVIEW.engine(),
+                unreadCount = UnreadCount(AN_UNREAD_COUNT),
+                typing = aTypingEvent(A_ROOM_OVERVIEW.roomId, TYPING_MEMBERS),
+                isMuted = MUTED_ROOM
+            )
+        )
+    }
+
+    private fun givenEmitsDirectoryState(
+        roomOverview: RoomOverview,
+        unreadCount: Int? = null,
+        isMuted: Boolean = false,
+        typing: List<RoomMember> = emptyList(),
+    ) {
+        val userCredentials = aUserCredentials()
+        fakeCredentialsStore.givenCredentials().returns(userCredentials)
+
+        val matrixOverviewState = listOf(roomOverview)
+
+        fakeSyncService.givenStartsSyncing()
+        fakeSyncService.givenOverview().returns(flowOf(matrixOverviewState))
+        fakeSyncService.givenEvents().returns(flowOf(if (typing.isEmpty()) emptyList() else listOf(aTypingSyncEvent(roomOverview.roomId, typing))))
+
+        fakeMessageService.givenEchos().returns(flowOf(emptyMap()))
+        fakeRoomStore.givenUnreadByCount().returns(flowOf(unreadCount?.let { mapOf(roomOverview.roomId to it) } ?: emptyMap()))
+        fakeRoomStore.givenMuted().returns(flowOf(if (isMuted) setOf(roomOverview.roomId) else emptySet()))
+
+        val mappedOverview = roomOverview.engine()
+        val expectedOverviewState = listOf(mappedOverview)
+        fakeMergeLocalEchosUseCase.givenMergedEchos(expectedOverviewState, userCredentials.userId, emptyMap()).returns(expectedOverviewState)
+    }
+}
+
+class FakeDirectoryMergeWithLocalEchosUseCase : DirectoryMergeWithLocalEchosUseCase by mockk() {
+    fun givenMergedEchos(overviewState: OverviewState, selfId: UserId, echos: Map<RoomId, List<MessageService.LocalEcho>>) = coEvery {
+        this@FakeDirectoryMergeWithLocalEchosUseCase.invoke(overviewState, selfId, echos)
+    }.delegateReturn()
+}

matrix-chat-engine/src/test/kotlin/app/dapk/st/engine/InviteUseCaseTest.kt

@@ -0,0 +1,39 @@
+package app.dapk.st.engine
+
+import app.dapk.st.matrix.common.RoomId
+import app.dapk.st.matrix.common.RoomMember
+import app.dapk.st.matrix.sync.InviteMeta
+import fake.FakeSyncService
+import fixture.aRoomId
+import fixture.aRoomMember
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.test.runTest
+import org.amshove.kluent.shouldBeEqualTo
+import org.junit.Test
+import app.dapk.st.matrix.sync.RoomInvite as MatrixRoomInvite
+
+class InviteUseCaseTest {
+
+    private val fakeSyncService = FakeSyncService()
+    private val useCase = InviteUseCase(fakeSyncService)
+
+    @Test
+    fun `reads invites from sync service and maps to engine`() = runTest {
+        val aMatrixRoomInvite = aMatrixRoomInvite()
+        fakeSyncService.givenStartsSyncing()
+        fakeSyncService.givenInvites().returns(flowOf(listOf(aMatrixRoomInvite)))
+
+        val result = useCase.invites().first()
+
+        result shouldBeEqualTo listOf(aMatrixRoomInvite.engine())
+    }
+
+}
+
+fun aMatrixRoomInvite(
+    from: RoomMember = aRoomMember(),
+    roomId: RoomId = aRoomId(),
+    inviteMeta: InviteMeta = InviteMeta.DirectMessage,
+) = MatrixRoomInvite(from, roomId, inviteMeta)
+

matrix-chat-engine/src/test/kotlin/app/dapk/st/engine/MergeWithLocalEchosUseCaseTest.kt

@@ -17,7 +17,7 @@ private val ANOTHER_ROOM_MESSAGE_EVENT = A_ROOM_MESSAGE_EVENT.copy(eventId = anE
 class MergeWithLocalEchosUseCaseTest {
 
     private val fakeLocalEchoMapper = fake.FakeLocalEventMapper()
-    private val mergeWithLocalEchosUseCase = MergeWithLocalEchosUseCaseImpl(fakeLocalEchoMapper.instance)
+    private val mergeWithLocalEchosUseCase = TimelineMergeWithLocalEchosUseCaseImpl(fakeLocalEchoMapper.instance)
 
     @Test
     fun `given no local echos, when merging text message, then returns original state`() {

matrix-chat-engine/src/test/kotlin/app/dapk/st/engine/ReadMarkingTimelineTest.kt

@@ -0,0 +1,66 @@
+package app.dapk.st.engine
+
+import app.dapk.st.matrix.common.RoomId
+import app.dapk.st.matrix.common.UserId
+import fake.FakeCredentialsStore
+import fake.FakeRoomStore
+import fixture.*
+import io.mockk.every
+import io.mockk.mockk
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.flowOf
+import org.amshove.kluent.shouldBeEqualTo
+import org.junit.Test
+import test.delegateReturn
+import test.runExpectTest
+
+private val A_ROOM_ID = aRoomId()
+private val A_USER_CREDENTIALS = aUserCredentials()
+private val A_ROOM_MESSAGE_FROM_OTHER_USER = aRoomMessageEvent(author = aRoomMember(id = aUserId("another-user")))
+private val A_ROOM_MESSAGE_FROM_SELF = aRoomMessageEvent(author = aRoomMember(id = A_USER_CREDENTIALS.userId))
+private val READ_RECEIPTS_ARE_DISABLED = true
+
+class ReadMarkingTimelineTest {
+
+    private val fakeRoomStore = FakeRoomStore()
+    private val fakeCredentialsStore = FakeCredentialsStore().apply { givenCredentials().returns(A_USER_CREDENTIALS) }
+    private val fakeObserveTimelineUseCase = FakeObserveTimelineUseCase()
+    private val fakeRoomService = FakeRoomService()
+
+    private val readMarkingTimeline = ReadMarkingTimeline(
+        fakeRoomStore,
+        fakeCredentialsStore,
+        fakeObserveTimelineUseCase,
+        fakeRoomService,
+    )
+
+    @Test
+    fun `given a message from self, when fetching, then only marks room as read on initial launch`() = runExpectTest {
+        fakeRoomStore.expectUnit(times = 1) { it.markRead(A_ROOM_ID) }
+        val messengerState = aMessengerState(roomState = aRoomState(events = listOf(A_ROOM_MESSAGE_FROM_SELF)))
+        fakeObserveTimelineUseCase.given(A_ROOM_ID, A_USER_CREDENTIALS.userId).returns(flowOf(messengerState))
+
+        val result = readMarkingTimeline.fetch(A_ROOM_ID, isReadReceiptsDisabled = READ_RECEIPTS_ARE_DISABLED).first()
+
+        result shouldBeEqualTo messengerState
+        verifyExpects()
+    }
+
+    @Test
+    fun `given a message from other user, when fetching, then marks room as read`() = runExpectTest {
+        fakeRoomStore.expectUnit(times = 2) { it.markRead(A_ROOM_ID) }
+        fakeRoomService.expectUnit { it.markFullyRead(A_ROOM_ID, A_ROOM_MESSAGE_FROM_OTHER_USER.eventId, isPrivate = READ_RECEIPTS_ARE_DISABLED) }
+        val messengerState = aMessengerState(roomState = aRoomState(events = listOf(A_ROOM_MESSAGE_FROM_OTHER_USER)))
+        fakeObserveTimelineUseCase.given(A_ROOM_ID, A_USER_CREDENTIALS.userId).returns(flowOf(messengerState))
+
+        val result = readMarkingTimeline.fetch(A_ROOM_ID, isReadReceiptsDisabled = READ_RECEIPTS_ARE_DISABLED).first()
+
+        result shouldBeEqualTo messengerState
+        verifyExpects()
+    }
+
+}
+
+class FakeObserveTimelineUseCase : ObserveTimelineUseCase by mockk() {
+    fun given(roomId: RoomId, userId: UserId) = every { this@FakeObserveTimelineUseCase.invoke(roomId, userId) }.delegateReturn()
+}

matrix-chat-engine/src/test/kotlin/app/dapk/st/engine/SendMessageUseCaseTest.kt

@@ -0,0 +1,139 @@
+package app.dapk.st.engine
+
+import app.dapk.st.matrix.common.RichText
+import app.dapk.st.matrix.message.MessageService
+import app.dapk.st.matrix.message.internal.ImageContentReader
+import fake.FakeLocalIdFactory
+import fixture.aRoomMember
+import fixture.aRoomOverview
+import fixture.anEventId
+import io.mockk.every
+import io.mockk.mockk
+import org.junit.Test
+import test.delegateReturn
+import test.runExpectTest
+import java.time.Clock
+
+private const val AN_IMAGE_URI = ""
+private val AN_IMAGE_META = ImageContentReader.ImageContent(
+    height = 50,
+    width = 100,
+    size = 1000L,
+    fileName = "a file name",
+    mimeType = "image/png"
+)
+private const val A_CURRENT_TIME = 2000L
+private const val A_LOCAL_ID = "a local id"
+private val A_ROOM_OVERVIEW = aRoomOverview(
+    isEncrypted = true
+)
+private val A_REPLY = SendMessage.TextMessage.Reply(
+    aRoomMember(),
+    originalMessage = "",
+    anEventId(),
+    timestampUtc = 7000
+)
+private const val A_TEXT_MESSAGE_CONTENT = "message content"
+
+class SendMessageUseCaseTest {
+
+    private val fakeMessageService = FakeMessageService()
+    private val fakeLocalIdFactory = FakeLocalIdFactory().apply { givenCreate().returns(A_LOCAL_ID) }
+    private val fakeImageContentReader = FakeImageContentReader()
+    private val fakeClock = FakeClock().apply { givenMillis().returns(A_CURRENT_TIME) }
+
+    private val useCase = SendMessageUseCase(
+        fakeMessageService,
+        fakeLocalIdFactory.instance,
+        fakeImageContentReader,
+        fakeClock.instance
+    )
+
+    @Test
+    fun `when sending image message, then schedules message`() = runExpectTest {
+        fakeImageContentReader.givenMeta(AN_IMAGE_URI).returns(AN_IMAGE_META)
+        val expectedImageMessage = createExpectedImageMessage(A_ROOM_OVERVIEW)
+        fakeMessageService.expect { it.scheduleMessage(expectedImageMessage) }
+
+        useCase.send(SendMessage.ImageMessage(uri = AN_IMAGE_URI), A_ROOM_OVERVIEW)
+
+        verifyExpects()
+    }
+
+    @Test
+    fun `when sending text message, then schedules message`() = runExpectTest {
+        val expectedTextMessage = createExpectedTextMessage(A_ROOM_OVERVIEW, A_TEXT_MESSAGE_CONTENT, reply = null)
+        fakeMessageService.expect { it.scheduleMessage(expectedTextMessage) }
+
+        useCase.send(
+            SendMessage.TextMessage(
+                content = A_TEXT_MESSAGE_CONTENT,
+                reply = null,
+            ),
+            A_ROOM_OVERVIEW
+        )
+
+        verifyExpects()
+    }
+
+    @Test
+    fun `given a reply, when sending text message, then schedules message with reply`() = runExpectTest {
+        val expectedTextMessage = createExpectedTextMessage(A_ROOM_OVERVIEW, A_TEXT_MESSAGE_CONTENT, reply = A_REPLY)
+        fakeMessageService.expect { it.scheduleMessage(expectedTextMessage) }
+
+        useCase.send(
+            SendMessage.TextMessage(
+                content = A_TEXT_MESSAGE_CONTENT,
+                reply = A_REPLY,
+            ),
+            A_ROOM_OVERVIEW
+        )
+
+        verifyExpects()
+    }
+
+
+    private fun createExpectedImageMessage(roomOverview: RoomOverview) = MessageService.Message.ImageMessage(
+        MessageService.Message.Content.ImageContent(
+            uri = AN_IMAGE_URI,
+            MessageService.Message.Content.ImageContent.Meta(
+                height = AN_IMAGE_META.height,
+                width = AN_IMAGE_META.width,
+                size = AN_IMAGE_META.size,
+                fileName = AN_IMAGE_META.fileName,
+                mimeType = AN_IMAGE_META.mimeType,
+            )
+        ),
+        roomId = roomOverview.roomId,
+        sendEncrypted = roomOverview.isEncrypted,
+        localId = A_LOCAL_ID,
+        timestampUtc = A_CURRENT_TIME,
+    )
+
+    private fun createExpectedTextMessage(roomOverview: RoomOverview, messageContent: String, reply: SendMessage.TextMessage.Reply?) =
+        MessageService.Message.TextMessage(
+            content = MessageService.Message.Content.TextContent(RichText.of(messageContent)),
+            roomId = roomOverview.roomId,
+            sendEncrypted = roomOverview.isEncrypted,
+            localId = A_LOCAL_ID,
+            timestampUtc = A_CURRENT_TIME,
+            reply = reply?.let {
+                MessageService.Message.TextMessage.Reply(
+                    author = it.author,
+                    originalMessage = RichText.of(it.originalMessage),
+                    replyContent = messageContent,
+                    eventId = it.eventId,
+                    timestampUtc = it.timestampUtc,
+                )
+            }
+        )
+}
+
+class FakeImageContentReader : ImageContentReader by mockk() {
+    fun givenMeta(uri: String) = every { meta(uri) }.delegateReturn()
+}
+
+class FakeClock {
+    val instance = mockk<Clock>()
+    fun givenMillis() = every { instance.millis() }.delegateReturn()
+}

matrix-chat-engine/src/test/kotlin/app/dapk/st/engine/TimelineUseCaseTest.kt

@@ -113,7 +113,7 @@ suspend fun <T> Flow<T>.test(scope: CoroutineScope) = FlowTestObserver(scope, th
     this.collect()
 }
 
-class FakeMergeWithLocalEchosUseCase : MergeWithLocalEchosUseCase by mockk() {
+class FakeMergeWithLocalEchosUseCase : TimelineMergeWithLocalEchosUseCase by mockk() {
     fun givenMerging(roomState: RoomState, roomMember: RoomMember, echos: List<MessageService.LocalEcho>) = every {
         this@FakeMergeWithLocalEchosUseCase.invoke(roomState.engine(), roomMember, echos)
     }.delegateReturn()
@@ -125,9 +125,8 @@ fun aTypingSyncEvent(
 ) = SyncService.SyncEvent.Typing(roomId, members)
 
 class FakeMessageService : MessageService by mockk() {
-
     fun givenEchos(roomId: RoomId) = every { localEchos(roomId) }.delegateReturn()
-
+    fun givenEchos() = every { localEchos() }.delegateReturn()
 }
 
 class FakeRoomService : RoomService by mockk() {
@@ -137,7 +136,7 @@ class FakeRoomService : RoomService by mockk() {
 
 fun aMessengerState(
     self: UserId = aUserId(),
-    roomState: app.dapk.st.engine.RoomState,
+    roomState: app.dapk.st.engine.RoomState = aRoomState(),
     typing: Typing? = null,
     isMuted: Boolean = IS_ROOM_MUTED,
 ) = MessengerPageState(self, roomState, typing, isMuted)

matrix/services/sync/src/testFixtures/kotlin/fake/FakeRoomStore.kt

@@ -10,6 +10,7 @@ import io.mockk.coVerify
 import io.mockk.every
 import io.mockk.mockk
 import kotlinx.coroutines.flow.Flow
+import test.delegateReturn
 
 class FakeRoomStore : RoomStore by mockk() {
 
@@ -34,8 +35,13 @@ class FakeRoomStore : RoomStore by mockk() {
         every { observeUnread() } returns unreadEvents
     }
 
+    fun givenUnreadEvents() = every { observeUnread() }.delegateReturn()
+    fun givenUnreadByCount() = every { observeUnreadCountById() }.delegateReturn()
+
     fun givenNotMutedUnreadEvents(unreadEvents: Flow<Map<RoomOverview, List<RoomEvent>>>) {
         every { observeNotMutedUnread() } returns unreadEvents
     }
 
+    fun givenMuted() = every { observeMuted() }.delegateReturn()
+
 }

matrix/services/sync/src/testFixtures/kotlin/fake/FakeSyncService.kt

@@ -4,17 +4,13 @@ import app.dapk.st.matrix.common.RoomId
 import app.dapk.st.matrix.sync.SyncService
 import io.mockk.every
 import io.mockk.mockk
-import kotlinx.coroutines.flow.emptyFlow
 import kotlinx.coroutines.flow.flowOf
 import test.delegateReturn
 
 class FakeSyncService : SyncService by mockk() {
-    fun givenStartsSyncing() {
+    fun givenStartsSyncing() = every { startSyncing() }.returns(flowOf(Unit))
-        every { startSyncing() }.returns(flowOf(Unit))
-    }
-
     fun givenRoom(roomId: RoomId) = every { room(roomId) }.delegateReturn()
-
+    fun givenEvents(roomId: RoomId? = null) = every { events(roomId) }.delegateReturn()
-    fun givenEvents(roomId: RoomId) = every { events(roomId) }.delegateReturn()
+    fun givenInvites() = every { invites() }.delegateReturn()
-
+    fun givenOverview() = every { overview() }.delegateReturn()
 }