Post rewatch errors in a bus and log them

gpunto · gpunto · commit fabc26fca480 · 2025-09-03T10:56:34.000+02:00
diff --git a/stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/client/Create.kt b/stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/client/Create.kt
@@ -26,6 +26,7 @@ import io.getstream.android.core.api.log.StreamLogger
 import io.getstream.android.core.api.log.StreamLoggerProvider
 import io.getstream.android.core.api.model.config.StreamClientSerializationConfig
 import io.getstream.android.core.api.model.config.StreamHttpConfig
+import io.getstream.android.core.api.model.exceptions.StreamClientException
 import io.getstream.android.core.api.model.value.StreamApiKey
 import io.getstream.android.core.api.model.value.StreamHttpClientInfoHeader
 import io.getstream.android.core.api.model.value.StreamUserId
@@ -69,6 +70,7 @@ import kotlinx.coroutines.CoroutineExceptionHandler
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.plus
 import okhttp3.OkHttpClient
 import okhttp3.logging.HttpLoggingInterceptor
@@ -248,12 +250,14 @@ internal fun createFeedsClient(
     val pollsRepository = PollsRepositoryImpl(feedsApi)
 
     val moderation = ModerationImpl(moderationRepository)
+    val errorBus = MutableSharedFlow<StreamClientException>(extraBufferCapacity = 100)
 
     val feedWatchHandler =
         FeedWatchHandler(
             connectionState = client.connectionState,
             feedsRepository = feedsRepository,
             retryProcessor = StreamRetryProcessor(logProvider.taggedLogger("WatchHandler")),
+            errorBus = errorBus,
             scope = clientScope,
         )
 
@@ -281,6 +285,8 @@ internal fun createFeedsClient(
                 maxWeakSubscriptions = Integer.MAX_VALUE,
             ),
         feedWatchHandler = feedWatchHandler,
+        errorBus = errorBus,
+        scope = clientScope,
         logger = logger,
     )
 }
diff --git a/stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/client/FeedsClientImpl.kt b/stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/client/FeedsClientImpl.kt
@@ -19,6 +19,7 @@ import io.getstream.android.core.api.StreamClient
 import io.getstream.android.core.api.log.StreamLogger
 import io.getstream.android.core.api.model.connection.StreamConnectedUser
 import io.getstream.android.core.api.model.connection.StreamConnectionState
+import io.getstream.android.core.api.model.exceptions.StreamClientException
 import io.getstream.android.core.api.model.value.StreamApiKey
 import io.getstream.android.core.api.socket.listeners.StreamClientListener
 import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
@@ -96,11 +97,13 @@ import io.getstream.feeds.android.network.models.DeleteActivitiesRequest
 import io.getstream.feeds.android.network.models.DeleteActivitiesResponse
 import io.getstream.feeds.android.network.models.ListDevicesResponse
 import io.getstream.feeds.android.network.models.WSEvent
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asSharedFlow
+import kotlinx.coroutines.launch
 
 internal class FeedsClientImpl(
     private val coreClient: StreamClient,
@@ -121,6 +124,8 @@ internal class FeedsClientImpl(
     override val moderation: Moderation,
     private val feedWatchHandler: FeedWatchHandler,
     private val logger: StreamLogger,
+    scope: CoroutineScope,
+    errorBus: Flow<StreamClientException>,
 ) : FeedsClient {
 
     override val state: StateFlow<StreamConnectionState>
@@ -149,6 +154,12 @@ internal class FeedsClientImpl(
             }
         }
 
+    init {
+        scope.launch {
+            errorBus.collect { logger.e(it) { "[FeedsClient] Received error from bus" } }
+        }
+    }
+
     override suspend fun connect(): Result<StreamConnectedUser> {
         if (user.type == UserAuthType.ANONYMOUS) {
             logger.e { "[connect] Attempting to connect an anonymous user, returning an error." }
diff --git a/stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/client/reconnect/FeedWatchHandler.kt b/stream-feeds-android-client/src/main/kotlin/io/getstream/feeds/android/client/internal/client/reconnect/FeedWatchHandler.kt
@@ -18,6 +18,7 @@ package io.getstream.feeds.android.client.internal.client.reconnect
 import io.getstream.android.core.api.filter.`in`
 import io.getstream.android.core.api.model.StreamRetryPolicy
 import io.getstream.android.core.api.model.connection.StreamConnectionState
+import io.getstream.android.core.api.model.exceptions.StreamClientException
 import io.getstream.android.core.api.processing.StreamRetryProcessor
 import io.getstream.feeds.android.client.api.model.FeedId
 import io.getstream.feeds.android.client.api.state.query.FeedsFilterField
@@ -26,6 +27,7 @@ import io.getstream.feeds.android.client.internal.repository.FeedsRepository
 import java.util.concurrent.ConcurrentHashMap
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.filterIsInstance
 import kotlinx.coroutines.launch
 
@@ -44,7 +46,8 @@ internal class FeedWatchHandler(
     private val connectionState: Flow<StreamConnectionState>,
     private val feedsRepository: FeedsRepository,
     private val retryProcessor: StreamRetryProcessor,
-    scope: CoroutineScope,
+    private val errorBus: MutableSharedFlow<StreamClientException>,
+    private val scope: CoroutineScope,
 ) {
     private val watched = ConcurrentHashMap<FeedId, Unit>()
 
@@ -65,17 +68,25 @@ internal class FeedWatchHandler(
     }
 
     private suspend fun rewatchAll() {
-        val toRewatch = watched.keys.mapTo(mutableListOf(), FeedId::rawValue)
-        if (toRewatch.isNotEmpty()) {
-            retryProcessor.retry(retryPolicy) {
-                feedsRepository
-                    .queryFeeds(FeedsQuery(FeedsFilterField.feed.`in`(toRewatch)))
-                    .getOrThrow()
-            }
+        val watchedCopy = watched.keys.toSet()
+
+        if (watchedCopy.isNotEmpty()) {
+            val toRewatch = watchedCopy.map(FeedId::rawValue)
+
+            retryProcessor
+                .retry(retryPolicy) {
+                    feedsRepository
+                        .queryFeeds(FeedsQuery(FeedsFilterField.feed.`in`(toRewatch)))
+                        .getOrThrow()
+                }
+                .onFailure { errorBus.emit(StreamFeedRewatchException(watchedCopy, it)) }
         }
     }
 
     companion object {
         private val retryPolicy = StreamRetryPolicy.exponential(maxRetries = 3)
     }
 }
+
+internal class StreamFeedRewatchException(val ids: Set<FeedId>, cause: Throwable? = null) :
+    StreamClientException("Failed to rewatch feeds", cause)
diff --git a/stream-feeds-android-client/src/test/kotlin/io/getstream/feeds/android/client/internal/client/FeedsClientImplTest.kt b/stream-feeds-android-client/src/test/kotlin/io/getstream/feeds/android/client/internal/client/FeedsClientImplTest.kt
@@ -19,6 +19,7 @@ import io.getstream.android.core.api.StreamClient
 import io.getstream.android.core.api.log.StreamLogger
 import io.getstream.android.core.api.model.connection.StreamConnectedUser
 import io.getstream.android.core.api.model.connection.StreamConnectionState
+import io.getstream.android.core.api.model.exceptions.StreamClientException
 import io.getstream.android.core.api.model.value.StreamApiKey
 import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
 import io.getstream.feeds.android.client.api.Moderation
@@ -48,13 +49,19 @@ import io.getstream.feeds.android.network.models.ListDevicesResponse
 import io.mockk.coEvery
 import io.mockk.every
 import io.mockk.mockk
+import io.mockk.verify
 import java.util.Date
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import kotlinx.coroutines.test.runTest
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertTrue
 import org.junit.Test
 
+@OptIn(ExperimentalCoroutinesApi::class)
 internal class FeedsClientImplTest {
     private val coreClient: StreamClient = mockk(relaxed = true)
     private val feedsEventsSubscriptionManager: StreamSubscriptionManager<FeedsEventListener> =
@@ -75,6 +82,8 @@ internal class FeedsClientImplTest {
     private val moderation: Moderation = mockk(relaxed = true)
     private val feedWatchHandler: FeedWatchHandler = mockk(relaxed = true)
     private val logger: StreamLogger = mockk(relaxed = true)
+    private val scope = TestScope(UnconfinedTestDispatcher())
+    private val errorBus = MutableSharedFlow<StreamClientException>()
 
     private val feedsClient: FeedsClientImpl =
         FeedsClientImpl(
@@ -96,8 +105,19 @@ internal class FeedsClientImplTest {
             moderation = moderation,
             feedWatchHandler = feedWatchHandler,
             logger = logger,
+            scope = scope,
+            errorBus = errorBus,
         )
 
+    @Test
+    fun `on error bus event, then log it`() = runTest {
+        val exception = StreamClientException("error!")
+
+        errorBus.emit(exception)
+
+        verify { logger.e(exception, any()) }
+    }
+
     @Test
     fun `connect when user is regular, then subscribes to client and connects successfully`() =
         runTest {
@@ -141,6 +161,8 @@ internal class FeedsClientImplTest {
                 moderation = moderation,
                 feedWatchHandler = feedWatchHandler,
                 logger = logger,
+                scope = scope,
+                errorBus = errorBus,
             )
 
         val result = anonymousClient.connect()
diff --git a/stream-feeds-android-client/src/test/kotlin/io/getstream/feeds/android/client/internal/client/reconnect/FeedWatchHandlerTest.kt b/stream-feeds-android-client/src/test/kotlin/io/getstream/feeds/android/client/internal/client/reconnect/FeedWatchHandlerTest.kt
@@ -15,9 +15,11 @@
  */
 package io.getstream.feeds.android.client.internal.client.reconnect
 
+import app.cash.turbine.test
 import io.getstream.android.core.api.filter.`in`
 import io.getstream.android.core.api.model.StreamRetryPolicy
 import io.getstream.android.core.api.model.connection.StreamConnectionState
+import io.getstream.android.core.api.model.exceptions.StreamClientException
 import io.getstream.android.core.api.processing.StreamRetryProcessor
 import io.getstream.android.core.result.runSafely
 import io.getstream.feeds.android.client.api.model.FeedId
@@ -34,20 +36,29 @@ import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import kotlinx.coroutines.test.runTest
+import org.junit.Assert.assertEquals
 import org.junit.Test
 
 @OptIn(ExperimentalCoroutinesApi::class)
 internal class FeedWatchHandlerTest {
     private val connectionEvents = MutableSharedFlow<StreamConnectionState>()
     private val feedsRepository: FeedsRepository = mockk()
-    private val scope = TestScope(UnconfinedTestDispatcher())
     private val retryProcessor: StreamRetryProcessor = TestRetryProcessor()
+    private val errorBus = MutableSharedFlow<StreamClientException>()
+    private val scope = TestScope(UnconfinedTestDispatcher())
 
-    private val handler = FeedWatchHandler(connectionEvents, feedsRepository, retryProcessor, scope)
+    private val handler =
+        FeedWatchHandler(
+            connectionState = connectionEvents,
+            feedsRepository = feedsRepository,
+            retryProcessor = retryProcessor,
+            errorBus = errorBus,
+            scope = scope,
+        )
 
     @Test
-    fun `on connected event, get feeds that are still being watched`() = runTest {
-        coEvery { feedsRepository.queryFeeds(any()) } returns Result.failure(Exception())
+    fun `on connected event, query feeds that are still being watched`() = runTest {
+        coEvery { feedsRepository.queryFeeds(any()) } returns Result.success(mockk())
         val id1 = FeedId("group", "id1")
         val id2 = FeedId("group", "id2")
         val id3 = FeedId("group", "id3")
@@ -63,6 +74,21 @@ internal class FeedWatchHandlerTest {
         coVerify { feedsRepository.queryFeeds(expectedQuery) }
     }
 
+    @Test
+    fun `on connected event when query fails, post error`() = runTest {
+        coEvery { feedsRepository.queryFeeds(any()) } returns Result.failure(Exception())
+        val id1 = FeedId("group", "id1")
+        val id2 = FeedId("group", "id2")
+        handler.onStartWatching(id1)
+        handler.onStartWatching(id2)
+
+        errorBus.test {
+            connectionEvents.emit(StreamConnectionState.Connected(mockk(), "connection-id"))
+
+            assertEquals(setOf(id1, id2), (awaitItem() as? StreamFeedRewatchException)?.ids)
+        }
+    }
+
     @Test
     fun `on non-connected event, do nothing`() = runTest {
         val id1 = FeedId("group", "id1")
