GetStream · gpunto · Jul 29, 2026 · Jul 30, 2026 · Jul 30, 2026 · Jul 30, 2026
@@ -356,9 +356,11 @@ public final class io/getstream/chat/android/client/api/ChatClientConfig {
 	public fun <init> (ZLjava/util/Set;ZZLio/getstream/chat/android/models/TimeDuration;Lkotlin/jvm/functions/Function0;Lio/getstream/chat/android/client/api/MessageLimitConfig;Z)V
 	public fun <init> (ZLjava/util/Set;ZZLio/getstream/chat/android/models/TimeDuration;Lkotlin/jvm/functions/Function0;Lio/getstream/chat/android/client/api/MessageLimitConfig;ZZ)V
 	public fun <init> (ZLjava/util/Set;ZZLio/getstream/chat/android/models/TimeDuration;Lkotlin/jvm/functions/Function0;Lio/getstream/chat/android/client/api/MessageLimitConfig;ZZLio/getstream/chat/android/client/cache/StreamCacheConfig;)V
-	public synthetic fun <init> (ZLjava/util/Set;ZZLio/getstream/chat/android/models/TimeDuration;Lkotlin/jvm/functions/Function0;Lio/getstream/chat/android/client/api/MessageLimitConfig;ZZLio/getstream/chat/android/client/cache/StreamCacheConfig;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public fun <init> (ZLjava/util/Set;ZZLio/getstream/chat/android/models/TimeDuration;Lkotlin/jvm/functions/Function0;Lio/getstream/chat/android/client/api/MessageLimitConfig;ZZLio/getstream/chat/android/client/cache/StreamCacheConfig;Z)V
+	public synthetic fun <init> (ZLjava/util/Set;ZZLio/getstream/chat/android/models/TimeDuration;Lkotlin/jvm/functions/Function0;Lio/getstream/chat/android/client/api/MessageLimitConfig;ZZLio/getstream/chat/android/client/cache/StreamCacheConfig;ZILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public final fun component1 ()Z
 	public final fun component10 ()Lio/getstream/chat/android/client/cache/StreamCacheConfig;
+	public final fun component11 ()Z
 	public final fun component2 ()Ljava/util/Set;
 	public final fun component3 ()Z
 	public final fun component4 ()Z
@@ -367,8 +369,8 @@ public final class io/getstream/chat/android/client/api/ChatClientConfig {
 	public final fun component7 ()Lio/getstream/chat/android/client/api/MessageLimitConfig;
 	public final fun component8 ()Z
 	public final fun component9 ()Z
-	public final fun copy (ZLjava/util/Set;ZZLio/getstream/chat/android/models/TimeDuration;Lkotlin/jvm/functions/Function0;Lio/getstream/chat/android/client/api/MessageLimitConfig;ZZLio/getstream/chat/android/client/cache/StreamCacheConfig;)Lio/getstream/chat/android/client/api/ChatClientConfig;
-	public static synthetic fun copy$default (Lio/getstream/chat/android/client/api/ChatClientConfig;ZLjava/util/Set;ZZLio/getstream/chat/android/models/TimeDuration;Lkotlin/jvm/functions/Function0;Lio/getstream/chat/android/client/api/MessageLimitConfig;ZZLio/getstream/chat/android/client/cache/StreamCacheConfig;ILjava/lang/Object;)Lio/getstream/chat/android/client/api/ChatClientConfig;
+	public final fun copy (ZLjava/util/Set;ZZLio/getstream/chat/android/models/TimeDuration;Lkotlin/jvm/functions/Function0;Lio/getstream/chat/android/client/api/MessageLimitConfig;ZZLio/getstream/chat/android/client/cache/StreamCacheConfig;Z)Lio/getstream/chat/android/client/api/ChatClientConfig;
+	public static synthetic fun copy$default (Lio/getstream/chat/android/client/api/ChatClientConfig;ZLjava/util/Set;ZZLio/getstream/chat/android/models/TimeDuration;Lkotlin/jvm/functions/Function0;Lio/getstream/chat/android/client/api/MessageLimitConfig;ZZLio/getstream/chat/android/client/cache/StreamCacheConfig;ZILjava/lang/Object;)Lio/getstream/chat/android/client/api/ChatClientConfig;
 	public fun equals (Ljava/lang/Object;)Z
 	public final fun getCacheConfig ()Lio/getstream/chat/android/client/cache/StreamCacheConfig;
 	public final fun getFastEventParsing ()Z
@@ -381,6 +383,7 @@ public final class io/getstream/chat/android/client/api/ChatClientConfig {
 	public final fun getUserPresence ()Z
 	public fun hashCode ()I
 	public final fun isAutomaticSyncOnReconnectEnabled ()Z
+	public final fun isLocalUnreadCountEnabled ()Z
 	public fun toString ()Ljava/lang/String;
 }
 
@@ -967,7 +970,8 @@ public final class io/getstream/chat/android/client/api/state/StateRegistry {
 	public fun <init> (Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/Job;Lkotlin/jvm/functions/Function0;Lkotlinx/coroutines/CoroutineScope;Lio/getstream/chat/android/client/api/MessageLimitConfig;)V
 	public fun <init> (Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/Job;Lkotlin/jvm/functions/Function0;Lkotlinx/coroutines/CoroutineScope;Lio/getstream/chat/android/client/api/MessageLimitConfig;Lkotlinx/coroutines/flow/StateFlow;)V
 	public fun <init> (Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/Job;Lkotlin/jvm/functions/Function0;Lkotlinx/coroutines/CoroutineScope;Lio/getstream/chat/android/client/api/MessageLimitConfig;Lkotlinx/coroutines/flow/StateFlow;Z)V
-	public synthetic fun <init> (Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/Job;Lkotlin/jvm/functions/Function0;Lkotlinx/coroutines/CoroutineScope;Lio/getstream/chat/android/client/api/MessageLimitConfig;Lkotlinx/coroutines/flow/StateFlow;ZILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public fun <init> (Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/Job;Lkotlin/jvm/functions/Function0;Lkotlinx/coroutines/CoroutineScope;Lio/getstream/chat/android/client/api/MessageLimitConfig;Lkotlinx/coroutines/flow/StateFlow;ZZ)V
+	public synthetic fun <init> (Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/flow/StateFlow;Lkotlinx/coroutines/Job;Lkotlin/jvm/functions/Function0;Lkotlinx/coroutines/CoroutineScope;Lio/getstream/chat/android/client/api/MessageLimitConfig;Lkotlinx/coroutines/flow/StateFlow;ZZILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public final fun channel (Ljava/lang/String;Ljava/lang/String;)Lio/getstream/chat/android/client/channel/state/ChannelState;
 	public final fun clear ()V
 	public final fun queryChannels (Lio/getstream/chat/android/models/FilterObject;Lio/getstream/chat/android/models/querysort/QuerySorter;)Lio/getstream/chat/android/client/api/state/QueryChannelsState;

@@ -89,6 +89,11 @@ import io.getstream.chat.android.models.TimeDuration
  *
  * @param cacheConfig Configuration for the SDK's user-configurable on-disk caches. By default all caches are off. See
  * [StreamCacheConfig] for the available cache types and their options.
+ *
+ * @param isLocalUnreadCountEnabled When enabled, the per-channel unread count is tracked on-device for channels where
+ * server-side read events are disabled ([io.getstream.chat.android.models.Config.readEventsEnabled] is `false`, typical
+ * for livestream channels): incoming messages increment a local counter and marking the channel as read resets it
+ * without a network request. Defaults to `false`.
  */
 public data class ChatClientConfig @JvmOverloads constructor(
     public val offlineEnabled: Boolean = true,
@@ -101,6 +106,7 @@ public data class ChatClientConfig @JvmOverloads constructor(
     public val useLegacyChannelLogic: Boolean = false,
     public val fastEventParsing: Boolean = false,
     public val cacheConfig: StreamCacheConfig = StreamCacheConfig(),
+    public val isLocalUnreadCountEnabled: Boolean = false,
 )
 
 /**

@@ -58,6 +58,8 @@ import java.util.concurrent.ConcurrentHashMap
  * @param messageLimitConfig Configuration for message limits.
  * @param mutedUsers The current list of muted users.
  * @param useLegacyChannelState Whether to use the legacy channel state implementation.
+ * @param isLocalUnreadCountEnabled Whether the per-channel unread count is tracked on-device for channels with
+ * server-side read events disabled.
  */
 @Suppress("LongParameterList", "TooManyFunctions")
 public class StateRegistry @JvmOverloads constructor(
@@ -70,6 +72,7 @@ public class StateRegistry @JvmOverloads constructor(
     private val messageLimitConfig: MessageLimitConfig,
     private val mutedUsers: StateFlow<List<Mute>> = MutableStateFlow(emptyList()),
     private val useLegacyChannelState: Boolean = true,
+    private val isLocalUnreadCountEnabled: Boolean = false,
 ) {
 
     private val logger by taggedLogger("Chat:StateRegistry")
@@ -188,6 +191,7 @@ public class StateRegistry @JvmOverloads constructor(
             activeLiveLocations = activeLiveLocations,
             baseMessageLimit = baseMessageLimit,
             now = now,
+            isLocalUnreadCountEnabled = isLocalUnreadCountEnabled,
         )
     }
 
@@ -203,6 +207,7 @@ public class StateRegistry @JvmOverloads constructor(
             mutedUsers = mutedUsers,
             liveLocations = activeLiveLocations,
             messageLimit = baseMessageLimit,
+            isLocalUnreadCountEnabled = isLocalUnreadCountEnabled,
         )
     }
 

@@ -21,10 +21,12 @@ import io.getstream.chat.android.client.extensions.getCreatedAtOrDefault
 import io.getstream.chat.android.client.extensions.internal.NEVER
 import io.getstream.chat.android.client.extensions.syncUnreadCountWithReads
 import io.getstream.chat.android.client.internal.offline.extensions.launchWithMutex
+import io.getstream.chat.android.client.internal.offline.repository.domain.channel.userread.internal.toModel
 import io.getstream.chat.android.client.persistance.repository.ChannelRepository
 import io.getstream.chat.android.client.utils.message.isPinned
 import io.getstream.chat.android.core.utils.date.minOf
 import io.getstream.chat.android.models.Channel
+import io.getstream.chat.android.models.ChannelUserRead
 import io.getstream.chat.android.models.DraftMessage
 import io.getstream.chat.android.models.Member
 import io.getstream.chat.android.models.Message
@@ -45,13 +47,15 @@ internal class DatabaseChannelRepository(
     private val getUser: suspend (userId: String) -> User,
     private val getMessage: suspend (messageId: String) -> Message?,
     private val getDraftMessage: suspend (cid: String) -> DraftMessage?,
+    private val currentUserId: String,
     private val now: () -> Long = { System.currentTimeMillis() },
     cacheSize: Int = 1000,
 ) : ChannelRepository {
 
     private val logger by taggedLogger("Chat:ChannelRepository")
     private val channelCache = LruCache<String, Channel>(cacheSize)
     private val dbMutex = Mutex()
+    private val cacheMutex = Mutex()
 
     override suspend fun insertChannel(channel: Channel) {
         insertChannels(listOf(channel))
@@ -64,23 +68,82 @@ internal class DatabaseChannelRepository(
      */
     override suspend fun insertChannels(channels: Collection<Channel>) {
         if (channels.isEmpty()) return
-        val updatedChannels = channels
-            .map { channelCache[it.cid]?.let { cachedChannel -> it.combine(cachedChannel) } ?: it }
-        val channelToInsert = updatedChannels
-            .filter { channelCache[it.cid] != it }
-            .map { it.toEntity() }
-        cacheChannel(updatedChannels)
+        val storedReadsByCid = channels
+            .filterNot { it.config.readEventsEnabled || channelCache[it.cid] != null }
+            .associate { it.cid to channelDao.select(it.cid)?.reads?.values?.map { read -> read.toModel(getUser) } }
+        val channelsToInsert = cacheMutex.withLock {
+            val updatedChannels = channels
+                .map { channelCache[it.cid]?.let { cachedChannel -> it.combine(cachedChannel) } ?: it }
+                .map {
+                    if (it.config.readEventsEnabled) it else it.preserveLocallyTrackedReads(storedReadsByCid[it.cid])
+                }
+            updatedChannels
+                .filter { channelCache[it.cid] != it }
+                .also { cacheChannel(updatedChannels) }
+        }
         scope.launchWithMutex(dbMutex) {
-            logger.v {
-                "[insertChannels] inserting ${channelToInsert.size} entities on DB, " +
-                    "updated ${updatedChannels.size} on cache"
-            }
-            channelToInsert
+            logger.v { "[insertChannels] inserting ${channelsToInsert.size} entities on DB" }
+            channelsToInsert
                 .takeUnless { it.isEmpty() }
+                // Re-read the cache at write time: DAO writes are not guaranteed to run in launch
+                // order, so the last write must persist the latest merged state, not its snapshot
+                ?.map { channel -> (channelCache[channel.cid] ?: channel).toEntity() }
                 ?.let { channelDao.insertMany(it) }
         }
     }
 
+    /**
+     * Keeps the stored current user read of read-events-disabled channels: it is tracked on-device and
+     * must never be overwritten by server data (a recency check is not enough - server reads carry
+     * lastReceivedEventDate = last_message_at, tying it). Other users' reads are merged by recency.
+     *
+     * [preloadedStoredReads] is the persisted read state fetched before acquiring the lock; the cache
+     * still takes priority when populated.
+     */
+    private fun Channel.preserveLocallyTrackedReads(preloadedStoredReads: List<ChannelUserRead>?): Channel {
+        val storedReads = channelCache[cid]?.read ?: preloadedStoredReads ?: return this
+        if (storedReads.isEmpty()) return this
+        val mergedByUser = read.associateByTo(mutableMapOf(), ChannelUserRead::getUserId)
+        storedReads.forEach { stored ->
+            val incoming = mergedByUser[stored.getUserId()]
+            mergedByUser[stored.getUserId()] = when {
+                incoming == null -> stored
+                stored.getUserId() == currentUserId -> stored.copy(
+                    // Only the user info and the delivered fields are merged from the server,
+                    // mirroring the in-memory merge in the channel state.
+                    user = incoming.user,
+                    lastDeliveredAt = incoming.lastDeliveredAt ?: stored.lastDeliveredAt,
+                    lastDeliveredMessageId = incoming.lastDeliveredMessageId ?: stored.lastDeliveredMessageId,
+                )
+                stored.lastReceivedEventDate.after(incoming.lastReceivedEventDate) -> stored
+                else -> incoming
+            }
+        }
+        return copy(read = mergedByUser.values.toList()).syncUnreadCountWithReads()
+    }
+
+    /**
+     * The cache merge runs under [cacheMutex] and the DAO writes re-read the cache, so the persisted
+     * row converges to the latest merged state regardless of write order.
+     */
+    override suspend fun upsertChannelReads(cid: String, reads: List<ChannelUserRead>) {
+        if (reads.isEmpty()) return
+        // Resolve the stored channel before the lock so no DB read runs while cacheMutex is held.
+        val stored = selectChannel(cid) ?: return
+        val updatedChannel = cacheMutex.withLock {
+            // The cache takes priority in case it was updated between the read above and the lock.
+            val base = channelCache[cid] ?: stored
+            base
+                .copy(read = (reads + base.read).distinctBy(ChannelUserRead::getUserId))
+                .syncUnreadCountWithReads()
+                .also { cacheChannel(it) }
+        }
+        scope.launchWithMutex(dbMutex) {
+            logger.v { "[upsertChannelReads] cid: $cid, reads.size: ${reads.size}" }
+            channelDao.insert((channelCache[cid] ?: updatedChannel).toEntity())
+        }
+    }
+
     private fun cacheChannel(vararg channels: Channel) {
         channels.forEach { channelCache.put(it.cid, it) }
     }

@@ -86,6 +86,7 @@ internal class DatabaseRepositoryFactory(
                 getUser,
                 getMessage,
                 messageRepository::selectDraftMessagesByCid,
+                currentUser.id,
                 now,
             )
                 .also { repository ->

@@ -115,6 +115,7 @@ import io.getstream.chat.android.client.internal.state.plugin.state.global.inter
 import io.getstream.chat.android.client.persistance.repository.RepositoryFacade
 import io.getstream.chat.android.client.setup.state.ClientState
 import io.getstream.chat.android.client.utils.channel.calculateNewLastMessageAt
+import io.getstream.chat.android.client.utils.internal.ChannelId
 import io.getstream.chat.android.client.utils.mergePartially
 import io.getstream.chat.android.client.utils.observable.Disposable
 import io.getstream.chat.android.core.internal.lazy.parameterizedLazy
@@ -165,6 +166,7 @@ internal class EventHandlerSequential(
     private val sideEffect: suspend () -> Unit,
     private val syncedEvents: Flow<List<ChatEvent>>,
     private val bufferConfig: MessageBufferConfig,
+    private val isLocalUnreadCountEnabled: Boolean,
     scope: CoroutineScope,
     private val groupedUnreadChannelsUpdater: GroupedUnreadChannelsUpdater = GroupedUnreadChannelsUpdater(
         stateRegistry = stateRegistry,
@@ -891,6 +893,8 @@ internal class EventHandlerSequential(
         // execute the batch
         batch.execute()
 
+        persistLocallyTrackedReads(events)
+
         // handle delete and truncate events
         for (event in events) {
             when (event) {
@@ -971,6 +975,30 @@ internal class EventHandlerSequential(
         }
     }
 
+    /**
+     * Persists the in-memory reads of the locally tracked channels affected by new-message events, so
+     * the on-device unread count survives a restart. Runs after the batch stored the server data.
+     */
+    private suspend fun persistLocallyTrackedReads(events: List<ChatEvent>) {
+        if (!isLocalUnreadCountEnabled) return
+        events
+            .mapNotNullTo(mutableSetOf()) { event ->
+                when (event) {
+                    is NewMessageEvent -> ChannelId.fromCid(event.cid)
+                    is NotificationMessageNewEvent -> ChannelId.fromCid(event.cid)
+                    else -> null
+                }
+            }
+            .forEach { channelId ->
+                if (!stateRegistry.isActiveChannel(channelId)) return@forEach
+                val channelState = stateRegistry.channel(channelId)
+                if (channelState.channelConfig.value.readEventsEnabled) return@forEach
+                val stateReads = channelState.reads.value
+                if (stateReads.isEmpty()) return@forEach
+                repos.upsertChannelReads(cid = channelId.cid, reads = stateReads)
+            }
+    }
+
     private suspend fun deleteMessagesFromUser(cid: String?, userId: String, hard: Boolean, deletedAt: Date) {
         val messages = if (cid != null) {
             // Delete messages only in the specified channel

@@ -111,6 +111,7 @@ public class StreamStatePluginFactory(
             messageLimitConfig = config.messageLimitConfig,
             mutedUsers = mutableGlobalState.muted,
             useLegacyChannelState = config.useLegacyChannelLogic,
+            isLocalUnreadCountEnabled = config.isLocalUnreadCountEnabled,
         )
 
         val isQueryingFree = MutableStateFlow(true)
@@ -163,6 +164,7 @@ public class StreamStatePluginFactory(
             syncedEvents = syncManager.syncedEvents,
             sideEffect = syncManager::awaitSyncing,
             bufferConfig = config.messageLimitConfig.messageBufferConfig,
+            isLocalUnreadCountEnabled = config.isLocalUnreadCountEnabled,
         )
 
         val stateErrorHandlerFactory = StateErrorHandlerFactory(
@@ -201,6 +203,7 @@ public class StreamStatePluginFactory(
         sideEffect: suspend () -> Unit,
         syncedEvents: Flow<List<ChatEvent>>,
         bufferConfig: MessageBufferConfig,
+        isLocalUnreadCountEnabled: Boolean,
     ): EventHandler {
         return EventHandlerSequential(
             scope = scope,
@@ -215,6 +218,7 @@ public class StreamStatePluginFactory(
             syncedEvents = syncedEvents,
             sideEffect = sideEffect,
             bufferConfig = bufferConfig,
+            isLocalUnreadCountEnabled = isLocalUnreadCountEnabled,
         )
     }
 }