diff --git a/stream-chat-android-client/api/stream-chat-android-client.api b/stream-chat-android-client/api/stream-chat-android-client.api index ea7365557b0..03a802ed7f5 100644 --- a/stream-chat-android-client/api/stream-chat-android-client.api +++ b/stream-chat-android-client/api/stream-chat-android-client.api @@ -356,9 +356,11 @@ public final class io/getstream/chat/android/client/api/ChatClientConfig { public fun (ZLjava/util/Set;ZZLio/getstream/chat/android/models/TimeDuration;Lkotlin/jvm/functions/Function0;Lio/getstream/chat/android/client/api/MessageLimitConfig;Z)V public fun (ZLjava/util/Set;ZZLio/getstream/chat/android/models/TimeDuration;Lkotlin/jvm/functions/Function0;Lio/getstream/chat/android/client/api/MessageLimitConfig;ZZ)V public fun (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 (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 (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 (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 (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 (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 (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 (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 (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 (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; diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/api/ChatClientConfig.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/api/ChatClientConfig.kt index 1855a6a8ade..7cfefaf64fc 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/api/ChatClientConfig.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/api/ChatClientConfig.kt @@ -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, ) /** diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/api/state/StateRegistry.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/api/state/StateRegistry.kt index 7fa5eec382d..b7a1e43d4fa 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/api/state/StateRegistry.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/api/state/StateRegistry.kt @@ -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> = 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, ) } diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/offline/repository/domain/channel/internal/DatabaseChannelRepository.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/offline/repository/domain/channel/internal/DatabaseChannelRepository.kt index 8f44ed39fc9..df6af9305bc 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/offline/repository/domain/channel/internal/DatabaseChannelRepository.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/offline/repository/domain/channel/internal/DatabaseChannelRepository.kt @@ -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,6 +47,7 @@ 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 { @@ -52,6 +55,7 @@ internal class DatabaseChannelRepository( private val logger by taggedLogger("Chat:ChannelRepository") private val channelCache = LruCache(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) { 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?): 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) { + 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) } } diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/offline/repository/factory/internal/DatabaseRepositoryFactory.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/offline/repository/factory/internal/DatabaseRepositoryFactory.kt index 05fea7c78ab..d409354f368 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/offline/repository/factory/internal/DatabaseRepositoryFactory.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/offline/repository/factory/internal/DatabaseRepositoryFactory.kt @@ -86,6 +86,7 @@ internal class DatabaseRepositoryFactory( getUser, getMessage, messageRepository::selectDraftMessagesByCid, + currentUser.id, now, ) .also { repository -> diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/event/handler/internal/EventHandlerSequential.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/event/handler/internal/EventHandlerSequential.kt index 6eb32d267f9..9ce0c700a91 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/event/handler/internal/EventHandlerSequential.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/event/handler/internal/EventHandlerSequential.kt @@ -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>, 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) { + 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 diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/factory/StreamStatePluginFactory.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/factory/StreamStatePluginFactory.kt index 214dd5bd3f2..5d3b313af46 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/factory/StreamStatePluginFactory.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/factory/StreamStatePluginFactory.kt @@ -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>, bufferConfig: MessageBufferConfig, + isLocalUnreadCountEnabled: Boolean, ): EventHandler { return EventHandlerSequential( scope = scope, @@ -215,6 +218,7 @@ public class StreamStatePluginFactory( syncedEvents = syncedEvents, sideEffect = sideEffect, bufferConfig = bufferConfig, + isLocalUnreadCountEnabled = isLocalUnreadCountEnabled, ) } } diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/listener/internal/ChannelMarkReadListenerState.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/listener/internal/ChannelMarkReadListenerState.kt index 14e94b4c357..7f22bac084b 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/listener/internal/ChannelMarkReadListenerState.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/listener/internal/ChannelMarkReadListenerState.kt @@ -17,6 +17,7 @@ package io.getstream.chat.android.client.internal.state.plugin.listener.internal import io.getstream.chat.android.client.internal.state.plugin.logic.internal.LogicRegistry +import io.getstream.chat.android.client.internal.state.plugin.state.channel.internal.MarkReadResult import io.getstream.chat.android.client.plugin.listeners.ChannelMarkReadListener import io.getstream.result.Error import io.getstream.result.Result @@ -41,11 +42,16 @@ internal class ChannelMarkReadListenerState(private val logic: LogicRegistry) : * @return [Result] with information if channel should be marked as read. */ override suspend fun onChannelMarkReadPrecondition(channelType: String, channelId: String): Result { - val shouldMarkRead = logic.channel(channelType, channelId).markRead() - return if (shouldMarkRead) { - Result.Success(Unit) - } else { - Result.Failure(Error.GenericError("Can not mark channel as read with channel id: $channelId")) + return when (logic.channel(channelType, channelId).markRead()) { + MarkReadResult.RemoteRequired -> Result.Success(Unit) + MarkReadResult.HandledLocally -> { + // No server read event follows a local mark-read; refresh the channel-list queries directly. + val cid = "$channelType:$channelId" + logic.getActiveQueryChannelsLogic().forEach { it.refreshChannelState(cid) } + Result.Failure(Error.GenericError("Channel $channelId marked as read locally")) + } + MarkReadResult.NotNeeded -> + Result.Failure(Error.GenericError("Can not mark channel as read with channel id: $channelId")) } } } diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/ChannelLogic.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/ChannelLogic.kt index 94a33d3f10e..7a35cd26c17 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/ChannelLogic.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/ChannelLogic.kt @@ -19,6 +19,7 @@ package io.getstream.chat.android.client.internal.state.plugin.logic.channel.int import io.getstream.chat.android.client.api.models.QueryChannelRequest import io.getstream.chat.android.client.channel.ChannelMessagesUpdateLogic import io.getstream.chat.android.client.events.ChatEvent +import io.getstream.chat.android.client.internal.state.plugin.state.channel.internal.MarkReadResult import io.getstream.chat.android.models.Channel import io.getstream.chat.android.models.Member import io.getstream.chat.android.models.Message @@ -170,11 +171,12 @@ internal interface ChannelLogic { fun setRepliedMessage(message: Message?) /** - * Marks channel as read locally. + * Marks the channel as read for the current user. * - * @return The flag to determine if the channel was marked as read locally. + * @return A [MarkReadResult] describing whether the channel should be marked as read remotely, + * was marked as read on-device, or does not need to be marked as read. */ - fun markRead(): Boolean + fun markRead(): MarkReadResult /** * Checks if typing events are enabled for the channel. diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/ChannelLogicImpl.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/ChannelLogicImpl.kt index 43ad7ab4e58..a4a281f97c4 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/ChannelLogicImpl.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/ChannelLogicImpl.kt @@ -30,6 +30,7 @@ import io.getstream.chat.android.client.extensions.internal.NEVER import io.getstream.chat.android.client.internal.state.model.querychannels.pagination.internal.QueryChannelPaginationRequest import io.getstream.chat.android.client.internal.state.model.querychannels.pagination.internal.toAnyChannelPaginationRequest import io.getstream.chat.android.client.internal.state.plugin.state.channel.internal.ChannelStateImpl +import io.getstream.chat.android.client.internal.state.plugin.state.channel.internal.MarkReadResult import io.getstream.chat.android.client.internal.state.plugin.state.global.internal.MutableGlobalState import io.getstream.chat.android.client.persistance.repository.RepositoryFacade import io.getstream.chat.android.models.Channel @@ -41,6 +42,7 @@ import io.getstream.chat.android.models.toChannelData import io.getstream.result.Result import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.launch import kotlinx.coroutines.withContext import java.util.Date @@ -134,10 +136,11 @@ internal class ChannelLogicImpl( state.setMemberCount(channel.memberCount) state.upsertMembers(channel.members) state.upsertWatchers(channel.watchers, channel.watcherCount) + // The config must be set before the reads: the read merge checks it to protect + // locally tracked reads. + state.setChannelConfig(channel.config) // Update reads state.updateReads(channel.read) - // Update config - state.setChannelConfig(channel.config) // Update messages if (limit > 0) { updateMessages(query, channel) @@ -244,8 +247,21 @@ internal class ChannelLogicImpl( state.setRepliedMessage(message) } - override fun markRead(): Boolean { - return state.markRead() + override fun markRead(): MarkReadResult { + val result = state.markRead() + if (result == MarkReadResult.HandledLocally) { + persistCurrentReads() + } + return result + } + + /** Persists the in-memory reads so the locally tracked unread count survives a restart. */ + private fun persistCurrentReads() { + val reads = state.reads.value + if (reads.isEmpty()) return + coroutineScope.launch { + repository.upsertChannelReads(cid = state.cid, reads = reads) + } } override fun typingEventsEnabled(): Boolean { @@ -291,10 +307,11 @@ internal class ChannelLogicImpl( state.setMemberCount(channel.memberCount) state.upsertMembers(channel.members) state.upsertWatchers(channel.watchers, channel.watcherCount) + // The config must be set before the reads: the read merge checks it to protect + // locally tracked reads. + state.setChannelConfig(channel.config) // Update reads state.updateReads(channel.read) - // Update channel config - state.setChannelConfig(channel.config) // Set pending messages state.setPendingMessages(channel.pendingMessages.map(PendingMessage::message)) // Update messages based on the relationship between the incoming page and existing state. diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelEventHandlerLegacyImpl.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelEventHandlerLegacyImpl.kt index c24cb8f62f4..391deefdd4d 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelEventHandlerLegacyImpl.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelEventHandlerLegacyImpl.kt @@ -117,9 +117,9 @@ internal class ChannelEventHandlerLegacyImpl( // Preserve createdLocallyAt only for messages created by current user, to ensure they are // sorted properly val preserveCreatedLocallyAt = event.message.user.id == getCurrentUserId() - upsertMessage(event.message, preserveCreatedLocallyAt) - // Update channel read state + // Must run before the upsert: messages already in the state do not update the unread count. stateLogic.updateCurrentUserRead(event.createdAt, event.message) + upsertMessage(event.message, preserveCreatedLocallyAt) // Update hidden state if the message is not shadowed if (!event.message.shadowed) { stateLogic.setHidden(false) @@ -131,11 +131,11 @@ internal class ChannelEventHandlerLegacyImpl( } is NotificationMessageNewEvent -> { + // Must run before the upsert: messages already in the state do not update the unread count. + stateLogic.updateCurrentUserRead(event.createdAt, event.message) if (!mutableState.insideSearch.value) { upsertMessage(event.message) } - // Update channel read state - stateLogic.updateCurrentUserRead(event.createdAt, event.message) // Update hidden state if the message is not shadowed if (!event.message.shadowed) { stateLogic.setHidden(false) diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelLogicLegacyImpl.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelLogicLegacyImpl.kt index f6fcef82e29..83147f64996 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelLogicLegacyImpl.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelLogicLegacyImpl.kt @@ -30,6 +30,7 @@ import io.getstream.chat.android.client.internal.state.model.querychannels.pagin import io.getstream.chat.android.client.internal.state.model.querychannels.pagination.internal.toAnyChannelPaginationRequest import io.getstream.chat.android.client.internal.state.plugin.logic.channel.internal.ChannelLogic import io.getstream.chat.android.client.internal.state.plugin.state.channel.internal.ChannelStateLegacyImpl +import io.getstream.chat.android.client.internal.state.plugin.state.channel.internal.MarkReadResult import io.getstream.chat.android.client.persistance.repository.RepositoryFacade import io.getstream.chat.android.client.query.pagination.AnyChannelPaginationRequest import io.getstream.chat.android.models.Channel @@ -199,8 +200,18 @@ internal class ChannelLogicLegacyImpl( stateLogic.setRepliedMessage(message) } - override fun markRead(): Boolean { - return stateLogic.markRead() + override fun markRead(): MarkReadResult { + val result = stateLogic.markRead() + if (result == MarkReadResult.HandledLocally) { + // Persist the reset read so the on-device unread count survives a restart. + val reads = mutableState.reads.value + if (reads.isNotEmpty()) { + coroutineScope.launch { + repos.upsertChannelReads(cid = mutableState.cid, reads = reads) + } + } + } + return result } override fun typingEventsEnabled(): Boolean { diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelStateLogic.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelStateLogic.kt index ffe0aaf7b0b..2fe0fe8ffb2 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelStateLogic.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelStateLogic.kt @@ -30,6 +30,7 @@ import io.getstream.chat.android.client.internal.state.message.attachments.inter import io.getstream.chat.android.client.internal.state.plugin.logic.channel.internal.SearchLogic import io.getstream.chat.android.client.internal.state.plugin.logic.channel.internal.TypingEventPruner import io.getstream.chat.android.client.internal.state.plugin.state.channel.internal.ChannelStateLegacyImpl +import io.getstream.chat.android.client.internal.state.plugin.state.channel.internal.MarkReadResult import io.getstream.chat.android.client.internal.state.plugin.state.global.internal.MutableGlobalState import io.getstream.chat.android.client.setup.state.ClientState import io.getstream.chat.android.client.utils.channel.calculateNewLastMessageAt @@ -81,6 +82,9 @@ internal class ChannelStateLogic( private val logger by taggedLogger(TAG) private val processedMessageIds = LruCache(CACHE_SIZE) + private val isReadTrackedLocally: Boolean + get() = mutableState.isLocalUnreadCountEnabled && !mutableState.channelConfig.value.readEventsEnabled + /** * Used to prune stale active typing events when the sender * of these events was unable to send a stop typing event. @@ -226,6 +230,22 @@ internal class ChannelStateLogic( localRead: ChannelUserRead, serverRead: ChannelUserRead, ): ChannelUserRead { + if (isReadTrackedLocally) { + // Locally tracked reads must never be overwritten by server data. A recency check is not + // enough: server reads carry lastReceivedEventDate = last_message_at, tying the local value. + if (localRead.unreadMessages != serverRead.unreadMessages) { + logger.d { + "[updateReads] preserving locally tracked read, " + + "local.unreadMessages=${localRead.unreadMessages}, " + + "server.unreadMessages=${serverRead.unreadMessages}" + } + } + return localRead.copy( + user = serverRead.user, + lastDeliveredAt = serverRead.lastDeliveredAt ?: localRead.lastDeliveredAt, + lastDeliveredMessageId = serverRead.lastDeliveredMessageId ?: localRead.lastDeliveredMessageId, + ) + } return if (localRead.lastReceivedEventDate.after(serverRead.lastReceivedEventDate)) { // Local state is more recent, preserve it but merge other fields from server logger.d { @@ -598,11 +618,11 @@ internal class ChannelStateLogic( } /** - * Marks channel as read locally. + * Marks the channel as read for the current user. * - * @return The flag to determine if the channel was marked as read locally. + * @return A [MarkReadResult] describing how the mark-read request was handled. */ - fun markRead(): Boolean { + fun markRead(): MarkReadResult { return mutableState.markChannelAsRead() } @@ -646,6 +666,10 @@ internal class ChannelStateLogic( mutableState.setMembersCount(channel.memberCount) + // The config must be set before the reads: the read merge checks it to protect + // locally tracked reads. + mutableState.setChannelConfig(channel.config) + updateReads(channel.read) // there are some edge cases here, this code adds to the members, watchers and messages @@ -673,8 +697,6 @@ internal class ChannelStateLogic( } } - mutableState.setChannelConfig(channel.config) - mutableState.setLoadingOlderMessages(false) mutableState.setLoadingNewerMessages(false) } @@ -891,6 +913,12 @@ internal class ChannelStateLogic( if (isProcessed) { return } + // Skip messages already in the channel state - e.g. a sync-replayed event after a restart, + // when the in-memory processed cache is empty + if (mutableState.getMessageById(message.id) != null) { + processedMessageIds.put(message.id, true) + return + } // Skip update if the channel is muted val isMuted = mutableState.muted.value if (isMuted) { @@ -932,15 +960,34 @@ internal class ChannelStateLogic( return } // Update the unread count - currentRead?.let { - updateRead( - it.copy( - lastReceivedEventDate = eventReceivedDate, - unreadMessages = it.unreadMessages.inc(), - ), + incrementUnreadCount(currentRead, eventReceivedDate) + processedMessageIds.put(message.id, true) + } + + /** The incremented read is upserted directly: the merge in [updateReads] is for server data. */ + private fun incrementUnreadCount(currentRead: ChannelUserRead?, eventReceivedDate: Date) { + val updatedRead = if (currentRead != null) { + currentRead.copy( + lastReceivedEventDate = eventReceivedDate, + unreadMessages = currentRead.unreadMessages.inc(), ) + } else if (isReadTrackedLocally) { + clientState.user.value?.let { user -> + ChannelUserRead( + user = user, + lastReceivedEventDate = eventReceivedDate, + unreadMessages = 1, + // lastRead/lastReadMessageId are unset: there is no server read to anchor them to. + lastRead = Date(0), + lastReadMessageId = null, + ) + } + } else { + null + } + updatedRead?.let { newRead -> + mutableState.upsertReads(listOf(newRead)) } - processedMessageIds.put(message.id, true) } private fun Message.storePoll() { diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/querychannels/internal/QueryChannelsStateLogic.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/querychannels/internal/QueryChannelsStateLogic.kt index d267696d417..3f90cba223c 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/querychannels/internal/QueryChannelsStateLogic.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/logic/querychannels/internal/QueryChannelsStateLogic.kt @@ -187,6 +187,12 @@ internal class QueryChannelsStateLogic( ) } }.forEach { it.await() } + // For read-events-disabled channels the locally tracked read is authoritative over the raw + // payload set above, so re-derive those channels from the merged per-channel state. + validated + .mapNotNull { (id, channel) -> id.cid.takeUnless { channel.config.readEventsEnabled } } + .takeIf { it.isNotEmpty() } + ?.let(::refreshChannels) } private fun Channel.joinMessages(existingChannel: Channel?): Channel = diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateImpl.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateImpl.kt index bf8f608eef6..dc12b04ed92 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateImpl.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateImpl.kt @@ -90,6 +90,7 @@ internal class ChannelStateImpl( private val messageLimit: Int?, val paginationManager: MessagesPaginationManager = MessagesPaginationManagerImpl(), private val attachmentUrlValidator: AttachmentUrlValidator = AttachmentUrlValidator(), + private val isLocalUnreadCountEnabled: Boolean = false, ) : ChannelState { override val cid: String = "$channelType:$channelId" @@ -1070,83 +1071,130 @@ internal class ChannelStateImpl( // Skip update if the message was already processed val isProcessed = processedMessageIds[message.id] == true if (isProcessed) { + logUnreadCountSkip(message, "already processed") + return + } + // Skip messages already in the channel state - e.g. a sync-replayed event after a restart, + // when the in-memory processed cache is empty + if (getMessageById(message.id) != null) { + logUnreadCountSkip(message, "already in state") + processedMessageIds.put(message.id, true) return } // Skip update if the channel is muted val isMuted = muted.value if (isMuted) { + logUnreadCountSkip(message, "channel muted") processedMessageIds.put(message.id, true) return } // Skip update for thread replies not shown in channel val isThreadReplyNotInChannel = message.parentId != null && !message.showInChannel if (isThreadReplyNotInChannel) { + logUnreadCountSkip(message, "thread reply not in channel") processedMessageIds.put(message.id, true) return } // Skip update for messages from current user val isFromCurrentUser = message.user.id == currentUser.value?.id if (isFromCurrentUser) { + logUnreadCountSkip(message, "own message") processedMessageIds.put(message.id, true) return } // Skip update for messages from muted users val isFromMutedUser = mutedUsers.value.any { it.target?.id == message.user.id } if (isFromMutedUser) { + logUnreadCountSkip(message, "author muted") processedMessageIds.put(message.id, true) return } // Skip update for messages from shadow banned users if (message.shadowed) { + logUnreadCountSkip(message, "shadowed") processedMessageIds.put(message.id, true) return } // Skip update for silent messages if (message.silent) { + logUnreadCountSkip(message, "silent") processedMessageIds.put(message.id, true) return } // Skip update if the event is outdated val currentRead = read.value if (currentRead != null && currentRead.lastReceivedEventDate.after(eventReceivedDate)) { + logUnreadCountSkip( + message, + "outdated (read: ${currentRead.lastReceivedEventDate}, event: $eventReceivedDate)", + ) processedMessageIds.put(message.id, true) return } // Update the unread count - currentRead?.let { - updateRead( - it.copy( - lastReceivedEventDate = eventReceivedDate, - unreadMessages = it.unreadMessages.inc(), - ), + incrementUnreadCount(currentRead, eventReceivedDate) + processedMessageIds.put(message.id, true) + } + + private fun logUnreadCountSkip(message: Message, reason: String) { + logger.v { "[updateCurrentUserRead] cid: $cid, skipping ${message.id}: $reason" } + } + + private val isReadTrackedLocally: Boolean + get() = isLocalUnreadCountEnabled && !channelConfig.value.readEventsEnabled + + /** The incremented read is upserted directly: the merge in [updateReads] is for server data. */ + private fun incrementUnreadCount(currentRead: ChannelUserRead?, eventReceivedDate: Date) { + val updatedRead = if (currentRead != null) { + currentRead.copy( + lastReceivedEventDate = eventReceivedDate, + unreadMessages = currentRead.unreadMessages.inc(), ) + } else if (isReadTrackedLocally) { + currentUser.value?.let { user -> + ChannelUserRead( + user = user, + lastReceivedEventDate = eventReceivedDate, + unreadMessages = 1, + // lastRead/lastReadMessageId are unset: there is no server read to anchor them to. + lastRead = Date(0), + lastReadMessageId = null, + ) + } + } else { + null + } + updatedRead?.let { newRead -> + logger.v { "[incrementUnreadCount] cid: $cid, unreadMessages: ${newRead.unreadMessages}" } + _reads.update { current -> + current + (newRead.getUserId() to newRead) + } } - processedMessageIds.put(message.id, true) } /** - * Marks the channel as read for the current user if the following conditions are met: - * 1. Read events are enabled in the channel configuration. - * 2. There are messages in the channel. - * 3. The last message in the channel is different from the last read message for the current user. - * - * @return `true` if the channel was marked as read, `false` otherwise. + * Marks the channel as read for the current user and returns how the request was handled: + * remotely, on-device (read events disabled + local tracking enabled), or not at all. */ - fun markRead(): Boolean { + fun markRead(): MarkReadResult { if (!channelConfig.value.readEventsEnabled) { - // Ignore request, `read_events = false` - return false + if (!isLocalUnreadCountEnabled) { + // Ignore request, `read_events = false` + return MarkReadResult.NotNeeded + } + markReadLocally() + return MarkReadResult.HandledLocally } val lastMessage = _messages.value.lastOrNull() if (lastMessage == null) { // No messages in the channel, nothing to mark as read - return true + return MarkReadResult.RemoteRequired } val currentUserRead = read.value if (currentUserRead == null) { // No read state for current user, but we can still mark the channel as read, and create a new read state // later when we receive the updated read state from the backend - return true + return MarkReadResult.RemoteRequired } return if (lastMessage.id != currentUserRead.lastReadMessageId) { // The last message is different from the last read message, we can mark the channel as read @@ -1158,10 +1206,30 @@ internal class ChannelStateImpl( _reads.update { current -> current + (updatedRead.getUserId() to updatedRead) } - true + MarkReadResult.RemoteRequired } else { // Already marked up to the latest message - false + MarkReadResult.NotNeeded + } + } + + /** + * Resets the current user's unread count on-device, advancing the read state (including + * [ChannelUserRead.lastReadMessageId], which the remote path leaves to the server). + */ + private fun markReadLocally() { + val currentUserRead = read.value ?: return + val lastMessage = _messages.value.lastOrNull() + val readDate = lastMessage?.getCreatedAtOrDefault(Date()) ?: Date() + val updatedRead = currentUserRead.copy( + lastReceivedEventDate = maxOf(currentUserRead.lastReceivedEventDate, readDate), + lastRead = readDate, + lastReadMessageId = lastMessage?.id ?: currentUserRead.lastReadMessageId, + unreadMessages = 0, + ) + logger.v { "[markReadLocally] cid: $cid, lastRead: ${updatedRead.lastRead}" } + _reads.update { current -> + current + (updatedRead.getUserId() to updatedRead) } } @@ -1614,6 +1682,22 @@ internal class ChannelStateImpl( localRead: ChannelUserRead, serverRead: ChannelUserRead, ): ChannelUserRead { + if (isReadTrackedLocally) { + // Locally tracked reads must never be overwritten by server data. A recency check is not + // enough: server reads carry lastReceivedEventDate = last_message_at, tying the local value. + if (localRead.unreadMessages != serverRead.unreadMessages) { + logger.d { + "[updateReads] preserving locally tracked read, " + + "local.unreadMessages=${localRead.unreadMessages}, " + + "server.unreadMessages=${serverRead.unreadMessages}" + } + } + return localRead.copy( + user = serverRead.user, + lastDeliveredAt = serverRead.lastDeliveredAt ?: localRead.lastDeliveredAt, + lastDeliveredMessageId = serverRead.lastDeliveredMessageId ?: localRead.lastDeliveredMessageId, + ) + } return if (localRead.lastReceivedEventDate.after(serverRead.lastReceivedEventDate)) { // Local state is more recent, preserve it but merge other fields from server logger.d { diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateLegacyImpl.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateLegacyImpl.kt index 09dcb908279..0d795df2e64 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateLegacyImpl.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateLegacyImpl.kt @@ -69,6 +69,7 @@ internal class ChannelStateLegacyImpl( latestUsers: StateFlow>, activeLiveLocations: StateFlow>, val baseMessageLimit: Int?, + val isLocalUnreadCountEnabled: Boolean = false, private val now: () -> Long, ) : ChannelState { @@ -591,39 +592,56 @@ internal class ChannelStateLegacyImpl( } /** - * Marks channel as read locally if different conditions are met: - * 1. Channel has read events enabled - * 2. Channel has messages not marked as read yet - * 3. Current user is set - * - * @return The flag to determine if the channel was marked as read locally. + * Marks the channel as read for the current user and returns how the request was handled: + * remotely, on-device (read events disabled + local tracking enabled), or not at all. */ - fun markChannelAsRead(): Boolean = messages.value - .takeIf { channelConfig.value.readEventsEnabled } - ?.lastOrNull() - ?.let { lastMessage -> - when (val currentUserRead = read.value) { - null -> true - else -> - // Allow marking as read if: - // 1. The last read message ID differs from the last message, OR - // 2. There are unread messages (even if lastReadMessageId matches, server state may differ) - currentUserRead - .takeIf { it.lastReadMessageId != lastMessage.id || it.unreadMessages > 0 } - ?.let { - upsertReads( - listOf( - it.copy( - lastReceivedEventDate = lastMessage.getCreatedAtOrDefault(Date()), - lastRead = lastMessage.getCreatedAtOrDefault(Date()), - unreadMessages = 0, - ), - ), - ) - true - } + fun markChannelAsRead(): MarkReadResult { + if (!channelConfig.value.readEventsEnabled) { + if (!isLocalUnreadCountEnabled) { + return MarkReadResult.NotNeeded } - } ?: false + markReadLocally() + return MarkReadResult.HandledLocally + } + val lastMessage = messages.value.lastOrNull() ?: return MarkReadResult.NotNeeded + val currentUserRead = read.value ?: return MarkReadResult.RemoteRequired + // Mark as read if the last read message differs from the last message, or there are unread + // messages (server state may differ even when the ids match). + return if (currentUserRead.lastReadMessageId != lastMessage.id || currentUserRead.unreadMessages > 0) { + upsertReads( + listOf( + currentUserRead.copy( + lastReceivedEventDate = lastMessage.getCreatedAtOrDefault(Date()), + lastRead = lastMessage.getCreatedAtOrDefault(Date()), + unreadMessages = 0, + ), + ), + ) + MarkReadResult.RemoteRequired + } else { + MarkReadResult.NotNeeded + } + } + + /** + * Resets the current user's unread count on-device, advancing the read state (including + * [ChannelUserRead.lastReadMessageId], which the remote path leaves to the server). + */ + private fun markReadLocally() { + val currentUserRead = read.value ?: return + val lastMessage = messages.value.lastOrNull() + val readDate = lastMessage?.getCreatedAtOrDefault(Date()) ?: Date() + upsertReads( + listOf( + currentUserRead.copy( + lastReceivedEventDate = maxOf(currentUserRead.lastReceivedEventDate, readDate), + lastRead = readDate, + lastReadMessageId = lastMessage?.id ?: currentUserRead.lastReadMessageId, + unreadMessages = 0, + ), + ), + ) + } fun removeMessagesBefore(date: Date) { logger.d { "[removeMessagesBefore] date: $date" } diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/MarkReadResult.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/MarkReadResult.kt new file mode 100644 index 00000000000..076e3121c50 --- /dev/null +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/MarkReadResult.kt @@ -0,0 +1,32 @@ +/* + * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved. + * + * Licensed under the Stream License; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://github.com/GetStream/stream-chat-android/blob/main/LICENSE + * + * 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.getstream.chat.android.client.internal.state.plugin.state.channel.internal + +/** + * Outcome of a request to mark a channel as read. + */ +internal sealed interface MarkReadResult { + + /** The channel needs to be marked as read with a remote request. */ + data object RemoteRequired : MarkReadResult + + /** The channel was marked as read on-device; no remote request is needed. */ + data object HandledLocally : MarkReadResult + + /** The channel does not need to be marked as read. */ + data object NotNeeded : MarkReadResult +} diff --git a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/persistance/repository/ChannelRepository.kt b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/persistance/repository/ChannelRepository.kt index 425a539eceb..32beba60106 100644 --- a/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/persistance/repository/ChannelRepository.kt +++ b/stream-chat-android-client/src/main/java/io/getstream/chat/android/client/persistance/repository/ChannelRepository.kt @@ -18,6 +18,7 @@ package io.getstream.chat.android.client.persistance.repository import io.getstream.chat.android.core.internal.InternalStreamChatApi import io.getstream.chat.android.models.Channel +import io.getstream.chat.android.models.ChannelUserRead import io.getstream.chat.android.models.Member import io.getstream.chat.android.models.Message import java.util.Date @@ -155,6 +156,15 @@ public interface ChannelRepository { */ public suspend fun updateLastMessageForChannel(cid: String, lastMessage: Message) + /** + * Upserts the given [reads] into the channel stored with the given [cid], replacing the stored + * reads of the same users. The default no-op must be overridden for the locally tracked unread + * count to survive app restarts. + */ + public suspend fun upsertChannelReads(cid: String, reads: List) { + // no-op by default + } + /** * Evict a [Channel] from the repository. * diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/offline/repository/domain/channel/internal/ChannelRepositoryImplTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/offline/repository/domain/channel/internal/ChannelRepositoryImplTest.kt index 03fbd46d7af..121d80404c8 100644 --- a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/offline/repository/domain/channel/internal/ChannelRepositoryImplTest.kt +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/offline/repository/domain/channel/internal/ChannelRepositoryImplTest.kt @@ -19,16 +19,23 @@ package io.getstream.chat.android.client.internal.offline.repository.domain.chan import io.getstream.chat.android.client.ChatClient import io.getstream.chat.android.client.MockChatClientBuilder import io.getstream.chat.android.randomChannel +import io.getstream.chat.android.randomChannelUserRead +import io.getstream.chat.android.randomConfig import io.getstream.chat.android.randomDraftMessageOrNull import io.getstream.chat.android.randomMessage import io.getstream.chat.android.randomUser import io.getstream.chat.android.test.TestCoroutineRule import kotlinx.coroutines.test.runTest import org.junit.Rule +import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.Arguments +import org.junit.jupiter.params.provider.MethodSource import org.mockito.Mockito import org.mockito.kotlin.any +import org.mockito.kotlin.argumentCaptor import org.mockito.kotlin.doReturn import org.mockito.kotlin.mock import org.mockito.kotlin.never @@ -54,6 +61,7 @@ internal class ChannelRepositoryImplTest { { randomUser() }, { randomMessage() }, { randomDraftMessageOrNull() }, + "current-user", ) @BeforeEach @@ -76,4 +84,154 @@ internal class ChannelRepositoryImplTest { verify(channelDao, never()).insert(any()) } + + @ParameterizedTest(name = "{0}") + @MethodSource("currentUserReadMergeInput") + fun `insertChannels merges the incoming current user read according to the channel config`( + testName: String, + readEventsEnabled: Boolean, + storedEventDate: Date, + incomingEventDate: Date, + expectedUnreadMessages: Int, + ) = runTest { + val repo = repositoryWithUserEchoingIds() + val cid = "messaging:local" + val stored = randomChannel( + id = "local", + type = "messaging", + config = randomConfig(readEventsEnabled = readEventsEnabled), + read = listOf( + randomChannelUserRead(user = currentUser, unreadMessages = 5, lastReceivedEventDate = storedEventDate), + ), + ) + whenever(channelDao.select(cid)) doReturn stored.toEntity() + // The server sends a read with unreadMessages = 0 + val serverPayload = stored.copy( + read = listOf( + randomChannelUserRead(user = currentUser, unreadMessages = 0, lastReceivedEventDate = incomingEventDate), + ), + ) + + repo.insertChannels(listOf(serverPayload)) + + assertEquals(expectedUnreadMessages, persistedUnreadCount(cid)) + } + + @Test + fun `insertChannels lets a newer server read win for other users on a read-events-disabled channel`() = runTest { + val repo = repositoryWithUserEchoingIds() + val cid = "messaging:local" + val otherUser = randomUser(id = "other-user") + val stored = randomChannel( + id = "local", + type = "messaging", + config = randomConfig(readEventsEnabled = false), + read = listOf( + randomChannelUserRead(user = otherUser, unreadMessages = 5, lastReceivedEventDate = Date(1000)), + ), + ) + whenever(channelDao.select(cid)) doReturn stored.toEntity() + val serverPayload = randomChannel( + id = "local", + type = "messaging", + config = randomConfig(readEventsEnabled = false), + read = listOf( + randomChannelUserRead(user = otherUser, unreadMessages = 0, lastReceivedEventDate = Date(2000)), + ), + ) + + repo.insertChannels(listOf(serverPayload)) + + val captor = argumentCaptor>() + verify(channelDao).insertMany(captor.capture()) + val persisted = captor.allValues.flatten().firstOrNull { it.cid == cid } + assertEquals(0, persisted?.reads?.get(otherUser.id)?.unreadMessages) + } + + @Test + fun `upsertChannelReads replaces the stored current user read and keeps other users reads`() = runTest { + val repo = repositoryWithUserEchoingIds() + val cid = "messaging:local" + val otherUser = randomUser(id = "other-user") + val stored = randomChannel( + id = "local", + type = "messaging", + config = randomConfig(readEventsEnabled = false), + read = listOf( + randomChannelUserRead(user = currentUser, unreadMessages = 5, lastReceivedEventDate = Date(2000)), + randomChannelUserRead(user = otherUser, unreadMessages = 7, lastReceivedEventDate = Date(2000)), + ), + ) + whenever(channelDao.select(cid)) doReturn stored.toEntity() + + // The local mark-read resets the count; the write is applied verbatim, no recency merge. + val resetRead = randomChannelUserRead( + user = currentUser, + unreadMessages = 0, + lastReceivedEventDate = Date(2000), + ) + repo.upsertChannelReads(cid, listOf(resetRead)) + + val captor = argumentCaptor() + verify(channelDao).insert(captor.capture()) + val persisted = captor.firstValue + assertEquals(0, persisted.reads[currentUser.id]?.unreadMessages) + assertEquals(7, persisted.reads[otherUser.id]?.unreadMessages) + } + + @Test + fun `upsertChannelReads does nothing when the channel is not stored`() = runTest { + val repo = repositoryWithUserEchoingIds() + val cid = "messaging:missing" + whenever(channelDao.select(cid)) doReturn null + + repo.upsertChannelReads(cid, listOf(randomChannelUserRead(user = currentUser))) + + verify(channelDao, never()).insert(any()) + } + + @Test + fun `upsertChannelReads does nothing when the reads are empty`() = runTest { + val repo = repositoryWithUserEchoingIds() + + repo.upsertChannelReads("messaging:local", emptyList()) + + verify(channelDao, never()).select(any()) + verify(channelDao, never()).insert(any()) + } + + private val currentUser = randomUser(id = "current-user") + + private fun repositoryWithUserEchoingIds(): DatabaseChannelRepository = + DatabaseChannelRepository( + testCoroutines.scope, + channelDao, + { userId -> randomUser(id = userId) }, + { randomMessage() }, + { randomDraftMessageOrNull() }, + currentUser.id, + ) + + /** Returns the current user's unread count from the entity written to the DAO for [cid]. */ + private suspend fun persistedUnreadCount(cid: String): Int? { + val captor = argumentCaptor>() + verify(channelDao).insertMany(captor.capture()) + return captor.allValues.flatten() + .firstOrNull { it.cid == cid } + ?.reads?.get(currentUser.id)?.unreadMessages + } + + companion object { + + @JvmStatic + fun currentUserReadMergeInput() = listOf( + // (test name, readEventsEnabled, storedEventDate, incomingEventDate, expectedUnreadMessages) + // Read events disabled: the read is tracked locally and the stored value always wins + Arguments.of("read events disabled, older incoming read", false, Date(2000), Date(1000), 5), + Arguments.of("read events disabled, tying incoming read", false, Date(2000), Date(2000), 5), + Arguments.of("read events disabled, newer incoming read", false, Date(1000), Date(2000), 5), + // Read events enabled: the incoming read wins + Arguments.of("read events enabled, older incoming read", true, Date(2000), Date(1000), 0), + ) + } } diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/event/TotalUnreadCountTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/event/TotalUnreadCountTest.kt index 395046f8430..18a7f4d3550 100644 --- a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/event/TotalUnreadCountTest.kt +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/event/TotalUnreadCountTest.kt @@ -146,6 +146,7 @@ internal class TotalUnreadCountTest { sideEffect = sideEffect, syncedEvents = syncedEvents, bufferConfig = MessageBufferConfig(), + isLocalUnreadCountEnabled = false, ) fun givenMockedRepositories(): Fixture { diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/event/handler/internal/EventHandlerSequentialTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/event/handler/internal/EventHandlerSequentialTest.kt index 7b4a883481d..7eab720c4c5 100644 --- a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/event/handler/internal/EventHandlerSequentialTest.kt +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/event/handler/internal/EventHandlerSequentialTest.kt @@ -84,6 +84,7 @@ import org.mockito.kotlin.argumentCaptor import org.mockito.kotlin.atLeast import org.mockito.kotlin.doReturn import org.mockito.kotlin.mock +import org.mockito.kotlin.never import org.mockito.kotlin.stub import org.mockito.kotlin.verify import org.mockito.kotlin.whenever @@ -420,6 +421,86 @@ internal class EventHandlerSequentialTest { } } + @Test + fun `When local unread tracking is enabled, a new message event persists the locally tracked reads`() = runTest { + val channelType = "livestream" + val channelId = "local-unread" + val cid = "$channelType:$channelId" + val repos: RepositoryFacade = mock() + repos.stub { + onBlocking { selectChannel(cid) } doReturn + randomChannel(id = channelId, type = channelType, ownCapabilities = emptySet()) + onBlocking { selectChannels(any()) } doReturn emptyList() + onBlocking { selectMessages(any()) } doReturn emptyList() + onBlocking { selectThreads(any()) } doReturn emptyList() + } + val localReads = listOf( + io.getstream.chat.android.randomChannelUserRead(user = currentUser, unreadMessages = 3), + ) + val handler = Fixture() + .withRepositoryFacade(repos) + .withLocalUnreadCountEnabled() + .withLocallyTrackedChannel(channelType, channelId, localReads) + .get(this) + + handler.handleEvents(randomNewMessageEvent(cid = cid, channelType = channelType, channelId = channelId)) + + verify(repos).upsertChannelReads(cid, localReads) + } + + @Test + fun `When local unread tracking is disabled, a new message event does not persist reads`() = runTest { + val channelType = "livestream" + val channelId = "local-unread" + val cid = "$channelType:$channelId" + val repos: RepositoryFacade = mock() + repos.stub { + onBlocking { selectChannel(cid) } doReturn + randomChannel(id = channelId, type = channelType, ownCapabilities = emptySet()) + } + val localReads = listOf( + io.getstream.chat.android.randomChannelUserRead(user = currentUser, unreadMessages = 3), + ) + val handler = Fixture() + .withRepositoryFacade(repos) + .withLocallyTrackedChannel(channelType, channelId, localReads) + .get(this) + + handler.handleEvents(randomNewMessageEvent(cid = cid, channelType = channelType, channelId = channelId)) + + verify(repos, never()).upsertChannelReads(any(), any()) + } + + @Test + fun `When local unread tracking is enabled, a notification new message event persists the locally tracked reads`() = + runTest { + val channelType = "livestream" + val channelId = "local-unread" + val cid = "$channelType:$channelId" + val repos: RepositoryFacade = mock() + repos.stub { + onBlocking { selectChannel(cid) } doReturn + randomChannel(id = channelId, type = channelType, ownCapabilities = emptySet()) + onBlocking { selectChannels(any()) } doReturn emptyList() + onBlocking { selectMessages(any()) } doReturn emptyList() + onBlocking { selectThreads(any()) } doReturn emptyList() + } + val localReads = listOf( + io.getstream.chat.android.randomChannelUserRead(user = currentUser, unreadMessages = 3), + ) + val handler = Fixture() + .withRepositoryFacade(repos) + .withLocalUnreadCountEnabled() + .withLocallyTrackedChannel(channelType, channelId, localReads) + .get(this) + + handler.handleEvents( + randomNotificationMessageNewEvent(cid = cid, channelType = channelType, channelId = channelId), + ) + + verify(repos).upsertChannelReads(cid, localReads) + } + @Test fun `When buffer overflows with DROP_OLDEST, the oldest queued NewMessageEvent is dropped`() = runTest { val fixture = Fixture() @@ -602,6 +683,11 @@ internal class EventHandlerSequentialTest { private val sideEffect: suspend () -> Unit = { sideEffectGate.await() } private val syncedEvents: Flow> = emptyFlow() private var bufferConfig: MessageBufferConfig = MessageBufferConfig() + private var isLocalUnreadCountEnabled: Boolean = false + + fun withLocalUnreadCountEnabled() = apply { + isLocalUnreadCountEnabled = true + } fun withReadEventsCapability(cid: String) = apply { repos.stub { @@ -677,6 +763,27 @@ internal class EventHandlerSequentialTest { whenever(stateRegistry.channel(channelType, channelId)) doReturn channelMutableState } + /** + * Stubs [stateRegistry] so the channel identified by [channelType] / [channelId] is + * active, has read events disabled, and carries the given locally tracked [reads]. + */ + fun withLocallyTrackedChannel( + channelType: String, + channelId: String, + reads: List, + ) = apply { + val channelState: io.getstream.chat.android.client.channel.state.ChannelState = + mock { + on { it.channelConfig } doReturn + MutableStateFlow(io.getstream.chat.android.models.Config(readEventsEnabled = false)) + on { it.reads } doReturn MutableStateFlow(reads) + } + val cid = io.getstream.chat.android.client.utils.internal.ChannelId + .fromTypeAndId(channelType, channelId)!! + whenever(stateRegistry.isActiveChannel(cid)) doReturn true + whenever(stateRegistry.channel(cid)) doReturn channelState + } + fun get(scope: CoroutineScope) = EventHandlerSequential( currentUserId = currentUser.id, subscribeForEvents = subscribeForEvents, @@ -688,6 +795,7 @@ internal class EventHandlerSequentialTest { sideEffect = sideEffect, syncedEvents = syncedEvents, bufferConfig = bufferConfig, + isLocalUnreadCountEnabled = isLocalUnreadCountEnabled, scope = scope, ) } diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/event/handler/internal/EventHandlerSequentialUserMessagesDeletedTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/event/handler/internal/EventHandlerSequentialUserMessagesDeletedTest.kt index 5ab23f068a3..9e19d652d7f 100644 --- a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/event/handler/internal/EventHandlerSequentialUserMessagesDeletedTest.kt +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/event/handler/internal/EventHandlerSequentialUserMessagesDeletedTest.kt @@ -347,6 +347,7 @@ internal class EventHandlerSequentialUserMessagesDeletedTest { sideEffect = sideEffect, syncedEvents = syncedEvents, bufferConfig = MessageBufferConfig(), + isLocalUnreadCountEnabled = false, scope = scope, ) } diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/listener/internal/ChannelMarkReadListenerStateTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/listener/internal/ChannelMarkReadListenerStateTest.kt new file mode 100644 index 00000000000..33485f201a9 --- /dev/null +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/listener/internal/ChannelMarkReadListenerStateTest.kt @@ -0,0 +1,88 @@ +/* + * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved. + * + * Licensed under the Stream License; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://github.com/GetStream/stream-chat-android/blob/main/LICENSE + * + * 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.getstream.chat.android.client.internal.state.plugin.listener.internal + +import io.getstream.chat.android.client.internal.state.plugin.logic.channel.internal.ChannelLogic +import io.getstream.chat.android.client.internal.state.plugin.logic.internal.LogicRegistry +import io.getstream.chat.android.client.internal.state.plugin.logic.querychannels.internal.QueryChannelsLogic +import io.getstream.chat.android.client.internal.state.plugin.state.channel.internal.MarkReadResult +import io.getstream.result.Result +import kotlinx.coroutines.test.runTest +import org.junit.jupiter.api.Assertions.assertInstanceOf +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.Arguments +import org.junit.jupiter.params.provider.MethodSource +import org.mockito.kotlin.doReturn +import org.mockito.kotlin.mock +import org.mockito.kotlin.never +import org.mockito.kotlin.times +import org.mockito.kotlin.verify + +internal class ChannelMarkReadListenerStateTest { + + private val channelType = "messaging" + private val channelId = "123" + + @ParameterizedTest(name = "{0}") + @MethodSource("markReadResults") + fun `onChannelMarkReadPrecondition maps the mark-read result to a precondition and a channel-list refresh`( + testName: String, + markReadResult: MarkReadResult, + succeeds: Boolean, + refreshes: Boolean, + ) = runTest { + // A remote mark-read succeeds (the network request follows) and relies on the server read + // event for the refresh; a local one fails (no request) and must refresh the channel-list + // queries itself; a not-needed one fails and refreshes nothing. + val queryChannelsLogic: QueryChannelsLogic = mock() + val listener = listenerWith(markReadResult, listOf(queryChannelsLogic)) + // when + val result = listener.onChannelMarkReadPrecondition(channelType, channelId) + // then + assertInstanceOf( + if (succeeds) Result.Success::class.java else Result.Failure::class.java, + result, + ) + verify(queryChannelsLogic, if (refreshes) times(1) else never()) + .refreshChannelState("$channelType:$channelId") + } + + private fun listenerWith( + markReadResult: MarkReadResult, + activeQueryChannelsLogic: List = emptyList(), + ): ChannelMarkReadListenerState { + val channelLogic: ChannelLogic = mock { + on { it.markRead() } doReturn markReadResult + } + val logic: LogicRegistry = mock { + on { it.channel(channelType, channelId) } doReturn channelLogic + on { it.getActiveQueryChannelsLogic() } doReturn activeQueryChannelsLogic + } + return ChannelMarkReadListenerState(logic) + } + + companion object { + + @JvmStatic + fun markReadResults() = listOf( + // (test name, markReadResult, succeeds, refreshes) + Arguments.of("remote required", MarkReadResult.RemoteRequired, true, false), + Arguments.of("handled locally", MarkReadResult.HandledLocally, false, true), + Arguments.of("not needed", MarkReadResult.NotNeeded, false, false), + ) + } +} diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/ChannelLogicImplTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/ChannelLogicImplTest.kt index 723ff69d823..1e4bb41f065 100644 --- a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/ChannelLogicImplTest.kt +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/ChannelLogicImplTest.kt @@ -20,6 +20,7 @@ import io.getstream.chat.android.client.api.models.Pagination import io.getstream.chat.android.client.api.models.QueryChannelRequest import io.getstream.chat.android.client.channel.ChannelMessagesUpdateLogic import io.getstream.chat.android.client.internal.state.plugin.state.channel.internal.ChannelStateImpl +import io.getstream.chat.android.client.internal.state.plugin.state.channel.internal.MarkReadResult import io.getstream.chat.android.client.internal.state.plugin.state.channel.internal.MessagesPaginationManager import io.getstream.chat.android.client.internal.state.plugin.state.global.internal.MutableGlobalState import io.getstream.chat.android.client.persistance.repository.RepositoryFacade @@ -50,6 +51,7 @@ import org.junit.jupiter.api.Test import org.junit.jupiter.api.extension.RegisterExtension import org.mockito.kotlin.any import org.mockito.kotlin.atLeastOnce +import org.mockito.kotlin.inOrder import org.mockito.kotlin.mock import org.mockito.kotlin.never import org.mockito.kotlin.verify @@ -887,22 +889,56 @@ internal class ChannelLogicImplTest { @Test fun `markRead should delegate to stateImpl and return result`() { // Given - whenever(stateImpl.markRead()).thenReturn(true) + whenever(stateImpl.markRead()).thenReturn(MarkReadResult.RemoteRequired) // When val result = sut.markRead() // Then - assertTrue(result) + assertEquals(MarkReadResult.RemoteRequired, result) verify(stateImpl).markRead() } @Test - fun `markRead should return false when stateImpl returns false`() { + fun `markRead should return the result from stateImpl`() { // Given - whenever(stateImpl.markRead()).thenReturn(false) + whenever(stateImpl.markRead()).thenReturn(MarkReadResult.NotNeeded) // When val result = sut.markRead() // Then - assertFalse(result) + assertEquals(MarkReadResult.NotNeeded, result) + } + + @Test + fun `markRead persists the reset read when handled locally`() = runTest { + // Given + val reads = listOf(randomChannelUserRead()) + whenever(stateImpl.markRead()).thenReturn(MarkReadResult.HandledLocally) + whenever(stateImpl.cid).thenReturn(cid) + whenever(stateImpl.reads).thenReturn(MutableStateFlow(reads)) + // When + sut.markRead() + // Then + verify(repository).upsertChannelReads(cid, reads) + } + + @Test + fun `markRead does not persist when not handled locally`() = runTest { + // Given + whenever(stateImpl.markRead()).thenReturn(MarkReadResult.RemoteRequired) + // When + sut.markRead() + // Then + verify(repository, never()).upsertChannelReads(any(), any()) + } + + @Test + fun `markRead does not persist when the state has no reads`() = runTest { + // Given + whenever(stateImpl.markRead()).thenReturn(MarkReadResult.HandledLocally) + whenever(stateImpl.reads).thenReturn(MutableStateFlow(emptyList())) + // When + sut.markRead() + // Then + verify(repository, never()).upsertChannelReads(any(), any()) } // endregion @@ -1038,6 +1074,29 @@ internal class ChannelLogicImplTest { verify(stateImpl).updateChannelData(any<(ChannelData?) -> ChannelData?>()) } + @Test + fun `should set the channel config before updating reads`() = runTest { + // The read merge checks the config to protect locally tracked reads, so the config + // must be up to date before the reads are merged. + val channel = randomChannel( + id = "123", + type = "messaging", + messages = emptyList(), + members = emptyList(), + watchers = emptyList(), + read = listOf(randomChannelUserRead()), + memberCount = 5, + watcherCount = 0, + ) + // When + sut.updateDataForChannel(channel = channel, messageLimit = 0) + // Then + inOrder(stateImpl) { + verify(stateImpl).setChannelConfig(channel.config) + verify(stateImpl).updateReads(channel.read) + } + } + @Test fun `should update member count`() = runTest { // Given diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelStateLogicTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelStateLogicTest.kt index 595140e3bca..dd708571763 100644 --- a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelStateLogicTest.kt +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/logic/channel/internal/legacy/ChannelStateLogicTest.kt @@ -66,6 +66,7 @@ import org.mockito.kotlin.any import org.mockito.kotlin.doAnswer import org.mockito.kotlin.doReturn import org.mockito.kotlin.eq +import org.mockito.kotlin.inOrder import org.mockito.kotlin.mock import org.mockito.kotlin.never import org.mockito.kotlin.spy @@ -766,6 +767,37 @@ internal class ChannelStateLogicTest { verify(mutableState).upsertReads(eq(listOf(expectedChannelUserRead))) } + @Test + fun `Given message already in the channel state, When updateCurrentUserRead is called, Then unread count is not updated`() { + // After a restart the sync replays the events since the last sync, including the event of + // the newest message already counted (and persisted) before the restart. That message is + // already part of the state seeded from the database, so it must not be counted twice. + val initialChannelUserRead = randomChannelUserRead( + user = user, + lastReceivedEventDate = Date(10L), + unreadMessages = 1, + lastRead = Date(10L), + lastReadMessageId = randomString(), + ) + _read.value = initialChannelUserRead + + val eventDate = Date(10L) + val newMessage = randomMessage( + user = randomUser(id = "anotherUserId"), + createdAt = eventDate, + silent = false, + shadowed = false, + parentId = null, + ) + whenever(mutableState.getMessageById(newMessage.id)) doReturn newMessage + + // when + channelStateLogic.updateCurrentUserRead(eventDate, newMessage) + + // then + verify(mutableState, times(0)).upsertReads(any()) + } + @Test fun `Given no current read state exists, When updateCurrentUserRead is called, Then unread count is not updated`() { // given - no current read state @@ -787,6 +819,37 @@ internal class ChannelStateLogicTest { verify(mutableState, times(0)).upsertReads(any()) } + @Test + fun `Given no read state and local unread tracking enabled, When updateCurrentUserRead is called, Then a read is created`() { + // given - no read state, local tracking on, read events disabled server-side + _read.value = null + _channelConfig.value = Config(readEventsEnabled = false) + whenever(mutableState.isLocalUnreadCountEnabled) doReturn true + + val eventDate = Date(20L) + val newMessage = randomMessage( + user = randomUser(id = "anotherUserId"), + createdAt = eventDate, + silent = false, + shadowed = false, + parentId = null, + ) + + val expectedChannelUserRead = ChannelUserRead( + user = user, + lastReceivedEventDate = eventDate, + unreadMessages = 1, + lastRead = Date(0), + lastReadMessageId = null, + ) + + // when + channelStateLogic.updateCurrentUserRead(eventDate, newMessage) + + // then + verify(mutableState).upsertReads(eq(listOf(expectedChannelUserRead))) + } + @Test fun `Given channel is in global mutes, When syncMuteState is called, Then setMuted is called with true`() { // given @@ -890,6 +953,105 @@ internal class ChannelStateLogicTest { ) } + @Test + fun `When updateDataForChannel is called, Then the config is set before the reads are updated`() { + // The read merge checks the config to protect locally tracked reads, so the config + // must be up to date before the reads are merged. + val channel = randomChannel(read = listOf(randomChannelUserRead(user = user))) + + channelStateLogic.updateDataForChannel( + channel = channel, + messageLimit = 0, + ) + + inOrder(mutableState) { + verify(mutableState).setChannelConfig(channel.config) + verify(mutableState).upsertReads(any()) + } + } + + @Test + fun `Given locally tracked channel, When updateDataForChannel is called with more recent server read, Then local read state is preserved`() { + val localRead = ChannelUserRead( + user = user, + lastReceivedEventDate = Date(50L), + unreadMessages = 3, + lastRead = Date(40L), + lastReadMessageId = null, + ) + // Server sends a sync payload with unreadMessages = 0 and a newer event date + val serverRead = ChannelUserRead( + user = user.copy(name = "Updated Name"), + lastReceivedEventDate = Date(100L), + unreadMessages = 0, + lastRead = Date(95L), + lastReadMessageId = "server-read-id", + lastDeliveredAt = Date(85L), + lastDeliveredMessageId = "server-delivered-id", + ) + + assertLocallyTrackedMerge( + localRead = localRead, + serverRead = serverRead, + // only user info and delivered fields are merged from the server + expectedRead = localRead.copy( + user = serverRead.user, + lastDeliveredAt = serverRead.lastDeliveredAt, + lastDeliveredMessageId = serverRead.lastDeliveredMessageId, + ), + ) + } + + @Test + fun `Given locally tracked channel, When updateDataForChannel is called with a server read tying on event date, Then local read state is preserved`() { + // Server reads carry lastReceivedEventDate = last_message_at, which ties with the local value + // anchored to the same message - a recency merge would let the stale server count win here. + val lastMessageDate = Date(100L) + val localRead = ChannelUserRead( + user = user, + lastReceivedEventDate = lastMessageDate, + unreadMessages = 3, + lastRead = Date(40L), + lastReadMessageId = null, + ) + val serverRead = localRead.copy(unreadMessages = 0) + + assertLocallyTrackedMerge(localRead = localRead, serverRead = serverRead, expectedRead = localRead) + } + + @Test + fun `Given locally tracked channel without local read, When updateDataForChannel is called, Then server data is used`() { + // first channel load, no local read state yet - the server value is authoritative + val serverRead = ChannelUserRead( + user = user, + lastReceivedEventDate = Date(100L), + unreadMessages = 5, + lastRead = Date(95L), + lastReadMessageId = "server-read-id", + ) + + assertLocallyTrackedMerge(localRead = null, serverRead = serverRead, expectedRead = serverRead) + } + + /** + * On a locally tracked channel (local unread count enabled, read events disabled), feeds + * [serverRead] through `updateDataForChannel` over the given [localRead] and asserts the merge + * upserts [expectedRead]. + */ + private fun assertLocallyTrackedMerge( + localRead: ChannelUserRead?, + serverRead: ChannelUserRead, + expectedRead: ChannelUserRead, + ) { + whenever(mutableState.isLocalUnreadCountEnabled) doReturn true + _channelConfig.value = Config(readEventsEnabled = false) + _read.value = localRead + + channelStateLogic.updateDataForChannel(channel = randomChannel(read = listOf(serverRead)), messageLimit = 0) + + verify(mutableState).upsertReads(eq(listOf(expectedRead))) + } + @Test fun `Given server read state is more recent than local, When updateDataForChannel is called, Then server data is used`() { // given - server state has more recent lastReceivedEventDate diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/logic/querychannels/internal/QueryChannelsStateLogicTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/logic/querychannels/internal/QueryChannelsStateLogicTest.kt index 8bb79719c79..da82b2eb9d8 100644 --- a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/logic/querychannels/internal/QueryChannelsStateLogicTest.kt +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/logic/querychannels/internal/QueryChannelsStateLogicTest.kt @@ -29,6 +29,8 @@ import io.getstream.chat.android.models.Filters import io.getstream.chat.android.models.querysort.QuerySortByField import io.getstream.chat.android.randomCID import io.getstream.chat.android.randomChannel +import io.getstream.chat.android.randomChannelUserRead +import io.getstream.chat.android.randomConfig import io.getstream.chat.android.randomMember import io.getstream.chat.android.randomMessage import io.getstream.chat.android.randomString @@ -46,6 +48,7 @@ import org.mockito.kotlin.argumentCaptor import org.mockito.kotlin.doReturn import org.mockito.kotlin.mock import org.mockito.kotlin.never +import org.mockito.kotlin.times import org.mockito.kotlin.verify import org.mockito.kotlin.whenever @@ -138,6 +141,43 @@ internal class QueryChannelsStateLogicTest { verify(mutableState).setChannels(channels.associateBy { it.cid }) } + @Test + fun `when adding channels with read events disabled, the channel is reconciled from the state`() = runTest { + // The payload of a read-events-disabled channel carries read data the locally tracked + // unread count must win over, so the added channel is rebuilt from the per-channel state. + val payloadChannel = randomChannel( + type = type, + id = id, + config = randomConfig(readEventsEnabled = false), + ) + val stateChannel = payloadChannel.copy( + read = listOf(randomChannelUserRead(unreadMessages = 3)), + ) + val channelState: ChannelState = mock { + on(it.toChannel()) doReturn stateChannel + } + whenever(stateRegistry.isActiveChannel(testChannelId)) doReturn true + whenever(stateRegistry.channel(testChannelId)) doReturn channelState + + queryChannelsStateLogic.addChannelsState(listOf(payloadChannel)) + + verify(mutableState).setChannels(mapOf(testCid to stateChannel)) + } + + @Test + fun `when adding channels with read events enabled, the channel is not reconciled from the state`() = runTest { + val payloadChannel = randomChannel( + type = type, + id = id, + config = randomConfig(readEventsEnabled = true), + ) + + queryChannelsStateLogic.addChannelsState(listOf(payloadChannel)) + + // Only the payload channels are set; no reconciliation pass runs + verify(mutableState, times(1)).setChannels(any()) + } + @Test fun `getActiveChannelState should return channel when it is active in state registry`() { val channel = randomChannel(type = type, id = id) diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateImplLocalUnreadCountTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateImplLocalUnreadCountTest.kt new file mode 100644 index 00000000000..1f9f0677308 --- /dev/null +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateImplLocalUnreadCountTest.kt @@ -0,0 +1,292 @@ +/* + * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved. + * + * Licensed under the Stream License; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://github.com/GetStream/stream-chat-android/blob/main/LICENSE + * + * 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.getstream.chat.android.client.internal.state.plugin.state.channel.internal + +import io.getstream.chat.android.models.ChannelUserRead +import io.getstream.chat.android.models.Config +import io.getstream.chat.android.models.User +import io.getstream.chat.android.randomUser +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.test.runTest +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Test +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.Arguments +import org.junit.jupiter.params.provider.MethodSource +import java.util.Date + +/** + * Tests for the on-device unread-count tracking enabled via + * [io.getstream.chat.android.client.api.ChatClientConfig.isLocalUnreadCountEnabled]. + */ +internal class ChannelStateImplLocalUnreadCountTest : ChannelStateImplTestBase() { + + @Test + fun `markRead resets unread count locally when local tracking is enabled and read events are disabled`() = + runTest { + val state = localTrackingState() + state.setMessages(listOf(createMessage(1, timestamp = 5000))) + state.updateRead(createRead(currentUser, unreadMessages = 5, lastRead = Date(1000))) + // when + val result = state.markRead() + // then: handled on-device and the local count is reset + assertEquals(MarkReadResult.HandledLocally, result) + assertEquals(0, state.unreadCount.value) + } + + @Test + fun `markRead is ignored when local tracking is disabled and read events are disabled`() = runTest { + val state = localTrackingState(isLocalUnreadCountEnabled = false) + state.setMessages(listOf(createMessage(1, timestamp = 5000))) + state.updateRead(createRead(currentUser, unreadMessages = 5, lastRead = Date(1000))) + // when + val result = state.markRead() + // then: not handled and the count is left untouched + assertEquals(MarkReadResult.NotNeeded, result) + assertEquals(5, state.unreadCount.value) + } + + @Test + fun `markRead uses the remote path when read events are enabled even if local tracking is enabled`() = runTest { + val state = localTrackingState(readEventsEnabled = true) + state.setMessages(listOf(createMessage(1, timestamp = 5000))) + state.updateRead( + createRead(currentUser, unreadMessages = 5, lastRead = Date(1000), lastReadMessageId = "old_id"), + ) + // when + val result = state.markRead() + // then: the remote path applies and the count is reset optimistically + assertEquals(MarkReadResult.RemoteRequired, result) + assertEquals(0, state.unreadCount.value) + } + + @Test + fun `updateCurrentUserRead creates a read state and increments when none exists for a local tracking channel`() = + runTest { + val state = localTrackingState() + assertNull(state.read.value) + val message = createMessage(1, user = randomUser(id = "other_user")) + // when + state.updateCurrentUserRead(Date(2000), message) + // then + assertEquals(1, state.read.value?.unreadMessages) + } + + @Test + fun `updateCurrentUserRead does not create a read state for own messages on a local tracking channel`() = runTest { + val state = localTrackingState() + val message = createMessage(1, user = currentUser) + // when + state.updateCurrentUserRead(Date(2000), message) + // then + assertNull(state.read.value) + assertEquals(0, state.unreadCount.value) + } + + @Test + fun `updateCurrentUserRead does not create a read state when local tracking is disabled`() = runTest { + val state = localTrackingState(isLocalUnreadCountEnabled = false) + val message = createMessage(1, user = randomUser(id = "other_user")) + // when + state.updateCurrentUserRead(Date(2000), message) + // then + assertNull(state.read.value) + } + + @Test + fun `updateCurrentUserRead does not create a read state when read events are enabled`() = runTest { + // read events enabled server-side, so the server owns the read state + val state = localTrackingState(readEventsEnabled = true) + val message = createMessage(1, user = randomUser(id = "other_user")) + // when + state.updateCurrentUserRead(Date(2000), message) + // then + assertNull(state.read.value) + assertEquals(0, state.unreadCount.value) + } + + @Test + fun `local tracking counts only messages received after a mark read`() = runTest { + val state = localTrackingState() + val otherUser = randomUser(id = "other_user") + // first message arrives -> unread = 1 (the read is updated before the message is upserted, + // mirroring the event handling order) + val firstMessage = createMessage(1, timestamp = 1000, user = otherUser) + state.updateCurrentUserRead(Date(1000), firstMessage) + state.setMessages(listOf(firstMessage)) + assertEquals(1, state.unreadCount.value) + // user opens the channel -> local mark read resets to 0 + state.markRead() + assertEquals(0, state.unreadCount.value) + // a later message arrives -> unread = 1 again, not 2 + val secondMessage = createMessage(2, timestamp = 5000, user = otherUser) + state.updateCurrentUserRead(Date(5000), secondMessage) + state.setMessages(listOf(firstMessage, secondMessage)) + assertEquals(1, state.unreadCount.value) + } + + @Test + fun `updateCurrentUserRead does not count a replayed event for a message already in the state`() = runTest { + // After a restart the sync replays the events since the last sync, including the event of + // the newest message already counted (and persisted) before the restart. That message is + // already part of the state seeded from the database, so it must not be counted twice. + val state = localTrackingState() + val message = createMessage(1, timestamp = 5000, user = randomUser(id = "other_user")) + // Seeded from the database: the message and the read counting it + state.setMessages(listOf(message)) + state.updateRead( + createRead(currentUser, unreadMessages = 1, lastRead = Date(1000), lastReceivedEventDate = Date(5000)), + ) + // when: the sync replays the message event + state.updateCurrentUserRead(Date(5000), message) + // then: the count is not incremented again + assertEquals(1, state.unreadCount.value) + } + + @ParameterizedTest(name = "{0}") + @MethodSource("updateReadsMergeInput") + fun `updateReads merges the server read according to the local tracking state`( + testName: String, + isLocalUnreadCountEnabled: Boolean, + readEventsEnabled: Boolean, + serverEventDate: Date, + expectedUnreadMessages: Int, + ) = runTest { + val state = localTrackingState(isLocalUnreadCountEnabled, readEventsEnabled) + state.updateRead( + createRead(currentUser, unreadMessages = 3, lastRead = Date(1000), lastReceivedEventDate = Date(3000)), + ) + // when: the server sends a read with unreadMessages = 0 + state.updateReads( + listOf( + createRead( + user = currentUser, + unreadMessages = 0, + lastRead = Date(1000), + lastReceivedEventDate = serverEventDate, + ), + ), + ) + // then + assertEquals(expectedUnreadMessages, state.read.value?.unreadMessages) + } + + @Test + fun `updateReads merges only user info and delivered fields from the server for locally tracked reads`() = + runTest { + val state = localTrackingState() + state.updateRead( + createRead(currentUser, unreadMessages = 3, lastRead = Date(1000)), + ) + // when + val serverRead = createRead( + user = currentUser.copy(name = "Updated Name"), + unreadMessages = 0, + lastRead = Date(5000), + lastReadMessageId = "server_message_id", + lastDeliveredAt = Date(4000), + lastDeliveredMessageId = "delivered_message_id", + ) + state.updateReads(listOf(serverRead)) + // then + val read = state.read.value + assertEquals(3, read?.unreadMessages) + assertEquals(Date(1000), read?.lastRead) + assertNull(read?.lastReadMessageId) + assertEquals("Updated Name", read?.user?.name) + assertEquals(Date(4000), read?.lastDeliveredAt) + assertEquals("delivered_message_id", read?.lastDeliveredMessageId) + } + + @Test + fun `updateReads uses the server data when no local read exists for a locally tracked channel`() = runTest { + // first channel load, no local read state yet + val state = localTrackingState() + assertNull(state.read.value) + // when + state.updateReads( + listOf(createRead(currentUser, unreadMessages = 5, lastRead = Date(1000))), + ) + // then: the server value is authoritative on first load + assertEquals(5, state.read.value?.unreadMessages) + } + + @Test + fun `updateReads does not affect other users reads for locally tracked channels`() = runTest { + val state = localTrackingState() + val otherUser = randomUser(id = "other_user") + // when + state.updateReads( + listOf(createRead(otherUser, unreadMessages = 10, lastRead = Date(2000))), + ) + // then + val otherRead = state.reads.value.find { it.user.id == "other_user" } + assertEquals(10, otherRead?.unreadMessages) + } + + private fun localTrackingState( + isLocalUnreadCountEnabled: Boolean = true, + readEventsEnabled: Boolean = false, + ) = ChannelStateImpl( + channelType = CHANNEL_TYPE, + channelId = CHANNEL_ID, + currentUser = userFlow, + latestUsers = MutableStateFlow(mapOf(currentUser.id to currentUser)), + mutedUsers = MutableStateFlow(emptyList()), + liveLocations = MutableStateFlow(emptyList()), + messageLimit = null, + isLocalUnreadCountEnabled = isLocalUnreadCountEnabled, + ).apply { + setChannelConfig(Config(readEventsEnabled = readEventsEnabled)) + } + + private fun createRead( + user: User, + unreadMessages: Int, + lastRead: Date, + lastReceivedEventDate: Date = lastRead, + lastReadMessageId: String? = null, + lastDeliveredAt: Date? = null, + lastDeliveredMessageId: String? = null, + ): ChannelUserRead = ChannelUserRead( + user = user, + lastReceivedEventDate = lastReceivedEventDate, + unreadMessages = unreadMessages, + lastRead = lastRead, + lastReadMessageId = lastReadMessageId, + lastDeliveredAt = lastDeliveredAt, + lastDeliveredMessageId = lastDeliveredMessageId, + ) + + companion object { + + @JvmStatic + fun updateReadsMergeInput() = listOf( + // (test name, isLocalUnreadCountEnabled, readEventsEnabled, serverEventDate, expectedUnreadMessages) + // Locally tracked: preserved even when the server read is more recent + Arguments.of("locally tracked, newer server read", true, false, Date(5000), 3), + // Locally tracked: preserved on tying event dates (server reads carry + // lastReceivedEventDate = last_message_at, which ties the local value) + Arguments.of("locally tracked, tying server read", true, false, Date(3000), 3), + // Read events enabled: the more recent server read wins + Arguments.of("read events enabled, newer server read", true, true, Date(5000), 0), + // Local tracking disabled: the standard recency merge applies + Arguments.of("local tracking disabled, newer server read", false, false, Date(5000), 0), + ) + } +} diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateImplReadReceiptsTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateImplReadReceiptsTest.kt index 58b8969dc66..ee2d60411a5 100644 --- a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateImplReadReceiptsTest.kt +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateImplReadReceiptsTest.kt @@ -25,7 +25,6 @@ import io.getstream.chat.android.randomUser import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.test.runTest import org.junit.jupiter.api.Assertions.assertEquals -import org.junit.jupiter.api.Assertions.assertFalse import org.junit.jupiter.api.Assertions.assertNotNull import org.junit.jupiter.api.Assertions.assertNull import org.junit.jupiter.api.Assertions.assertTrue @@ -487,39 +486,39 @@ internal class ChannelStateImplReadReceiptsTest : ChannelStateImplTestBase() { inner class MarkRead { @Test - fun `markRead should return false when read events are disabled`() = runTest { + fun `markRead returns None when read events are disabled`() = runTest { // given channelState.setChannelConfig(Config(readEventsEnabled = false)) channelState.setMessages(listOf(createMessage(1))) // when val result = channelState.markRead() // then - assertFalse(result) + assertEquals(MarkReadResult.NotNeeded, result) } @Test - fun `markRead should return true when no messages exist`() = runTest { + fun `markRead returns Remote when no messages exist`() = runTest { // given channelState.setChannelConfig(Config(readEventsEnabled = true)) // when val result = channelState.markRead() // then - assertTrue(result) + assertEquals(MarkReadResult.RemoteRequired, result) } @Test - fun `markRead should return true when no read state for current user`() = runTest { + fun `markRead returns Remote when no read state for current user`() = runTest { // given channelState.setChannelConfig(Config(readEventsEnabled = true)) channelState.setMessages(listOf(createMessage(1))) // when val result = channelState.markRead() // then - assertTrue(result) + assertEquals(MarkReadResult.RemoteRequired, result) } @Test - fun `markRead should mark as read and reset unread count`() = runTest { + fun `markRead marks as read remotely and resets unread count`() = runTest { // given channelState.setChannelConfig(Config(readEventsEnabled = true)) val message = createMessage(1, timestamp = 5000) @@ -534,12 +533,12 @@ internal class ChannelStateImplReadReceiptsTest : ChannelStateImplTestBase() { // when val result = channelState.markRead() // then - assertTrue(result) + assertEquals(MarkReadResult.RemoteRequired, result) assertEquals(0, channelState.unreadCount.value) } @Test - fun `markRead should return false when last message is already read`() = runTest { + fun `markRead returns None when last message is already read`() = runTest { // given channelState.setChannelConfig(Config(readEventsEnabled = true)) val message = createMessage(1) @@ -554,7 +553,7 @@ internal class ChannelStateImplReadReceiptsTest : ChannelStateImplTestBase() { // when val result = channelState.markRead() // then - assertFalse(result) + assertEquals(MarkReadResult.NotNeeded, result) } @Test diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateLegacyImplLocalUnreadCountTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateLegacyImplLocalUnreadCountTest.kt new file mode 100644 index 00000000000..8504b6def9c --- /dev/null +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateLegacyImplLocalUnreadCountTest.kt @@ -0,0 +1,92 @@ +/* + * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved. + * + * Licensed under the Stream License; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://github.com/GetStream/stream-chat-android/blob/main/LICENSE + * + * 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.getstream.chat.android.client.internal.state.plugin.state.channel.internal + +import io.getstream.chat.android.models.User +import io.getstream.chat.android.randomChannelUserRead +import io.getstream.chat.android.randomConfig +import io.getstream.chat.android.randomMessage +import io.getstream.chat.android.test.TestCoroutineExtension +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.test.runTest +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.extension.RegisterExtension + +/** + * Tests for on-device unread tracking on the legacy channel state implementation. + */ +internal class ChannelStateLegacyImplLocalUnreadCountTest { + + private val userFlow = MutableStateFlow(currentUser) + + @Test + fun `markChannelAsRead resets the count locally when local tracking is enabled and read events are disabled`() = + runTest { + val state = localTrackingState() + state.setMessages(listOf(randomMessage(parentId = null, shadowed = false))) + state.upsertReads(listOf(randomChannelUserRead(user = currentUser, unreadMessages = 3))) + // when + val result = state.markChannelAsRead() + // then + assertEquals(MarkReadResult.HandledLocally, result) + assertEquals(0, state.read.value?.unreadMessages) + } + + @Test + fun `markChannelAsRead is ignored when local tracking is disabled and read events are disabled`() = runTest { + val state = localTrackingState(isLocalUnreadCountEnabled = false) + state.setMessages(listOf(randomMessage(parentId = null, shadowed = false))) + state.upsertReads(listOf(randomChannelUserRead(user = currentUser, unreadMessages = 3))) + // when + val result = state.markChannelAsRead() + // then: not handled and the count is left untouched + assertEquals(MarkReadResult.NotNeeded, result) + assertEquals(3, state.read.value?.unreadMessages) + } + + private fun localTrackingState( + isLocalUnreadCountEnabled: Boolean = true, + readEventsEnabled: Boolean = false, + ) = ChannelStateLegacyImpl( + channelType = CHANNEL_TYPE, + channelId = CHANNEL_ID, + userFlow = userFlow, + latestUsers = MutableStateFlow(mapOf(currentUser.id to currentUser)), + activeLiveLocations = MutableStateFlow(emptyList()), + baseMessageLimit = null, + isLocalUnreadCountEnabled = isLocalUnreadCountEnabled, + now = ::currentTime, + ).apply { + setChannelConfig(randomConfig(readEventsEnabled = readEventsEnabled)) + } + + private companion object { + @JvmField + @RegisterExtension + val testCoroutines = TestCoroutineExtension() + + const val CHANNEL_TYPE = "messaging" + const val CHANNEL_ID = "123" + + val currentUser = User(id = "tom", name = "Tom") + + @OptIn(ExperimentalCoroutinesApi::class) + fun currentTime() = testCoroutines.dispatcher.scheduler.currentTime + } +} diff --git a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateLegacyImplTest.kt b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateLegacyImplTest.kt index 367b635b423..c6b21f047e4 100644 --- a/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateLegacyImplTest.kt +++ b/stream-chat-android-client/src/test/java/io/getstream/chat/android/client/internal/state/plugin/state/channel/internal/ChannelStateLegacyImplTest.kt @@ -617,7 +617,7 @@ internal class ChannelStateLegacyImplTest { } @Test - fun `markChannelAsRead should return true when lastReadMessageId matches lastMessage but unreadMessages greater than 0`() { + fun `markChannelAsRead returns Remote when lastReadMessageId matches lastMessage but unreadMessages greater than 0`() { val lastMessage = randomMessage( parentId = null, shadowed = false, @@ -636,14 +636,14 @@ internal class ChannelStateLegacyImplTest { val actual = channelState.markChannelAsRead() - assertEquals(true, actual) + assertEquals(MarkReadResult.RemoteRequired, actual) assertEquals(lastMessage.createdLocallyAt, channelState.read.value?.lastReceivedEventDate) assertEquals(lastMessage.createdLocallyAt, channelState.read.value?.lastRead) assertEquals(0, channelState.read.value?.unreadMessages) } @Test - fun `markChannelAsRead should return false when lastReadMessageId matches and unreadMessages is 0`() = + fun `markChannelAsRead returns None when lastReadMessageId matches and unreadMessages is 0`() = runTest { val lastMessage = randomMessage( parentId = null, @@ -661,11 +661,11 @@ internal class ChannelStateLegacyImplTest { val result = channelState.markChannelAsRead() - assertEquals(false, result) + assertEquals(MarkReadResult.NotNeeded, result) } @Test - fun `markChannelAsRead should return true when lastReadMessageId differs from lastMessage`() = runTest { + fun `markChannelAsRead returns Remote when lastReadMessageId differs from lastMessage`() = runTest { val lastMessage = randomMessage( parentId = null, shadowed = false, @@ -682,14 +682,14 @@ internal class ChannelStateLegacyImplTest { val result = channelState.markChannelAsRead() - assertEquals(true, result) + assertEquals(MarkReadResult.RemoteRequired, result) assertEquals(lastMessage.createdLocallyAt, channelState.read.value?.lastReceivedEventDate) assertEquals(lastMessage.createdLocallyAt, channelState.read.value?.lastRead) assertEquals(0, channelState.read.value?.unreadMessages) } @Test - fun `markChannelAsRead should return false when readEventsEnabled is false`() = runTest { + fun `markChannelAsRead returns None when readEventsEnabled is false`() = runTest { val lastMessage = randomMessage( parentId = null, shadowed = false, @@ -705,11 +705,11 @@ internal class ChannelStateLegacyImplTest { val result = channelState.markChannelAsRead() - assertEquals(false, result) + assertEquals(MarkReadResult.NotNeeded, result) } @Test - fun `markChannelAsRead should return false when there are no messages`() = runTest { + fun `markChannelAsRead returns None when there are no messages`() = runTest { channelState.setMessages(emptyList()) channelState.setChannelConfig(randomConfig(readEventsEnabled = true)) @@ -721,11 +721,11 @@ internal class ChannelStateLegacyImplTest { val result = channelState.markChannelAsRead() - assertEquals(false, result) + assertEquals(MarkReadResult.NotNeeded, result) } @Test - fun `markChannelAsRead should return true when read state is null`() = runTest { + fun `markChannelAsRead returns Remote when read state is null`() = runTest { val lastMessage = randomMessage( parentId = null, shadowed = false, @@ -739,7 +739,7 @@ internal class ChannelStateLegacyImplTest { val result = channelState.markChannelAsRead() - assertEquals(true, result) + assertEquals(MarkReadResult.RemoteRequired, result) } @Test diff --git a/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ChatHelper.kt b/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ChatHelper.kt index 0f8dc8118bd..dfa0002aa16 100644 --- a/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ChatHelper.kt +++ b/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ChatHelper.kt @@ -25,6 +25,7 @@ import io.getstream.chat.android.client.logger.ChatLogLevel import io.getstream.chat.android.client.notifications.handler.NotificationConfig import io.getstream.chat.android.client.notifications.handler.NotificationHandlerFactory import io.getstream.chat.android.compose.sample.data.UserCredentials +import io.getstream.chat.android.compose.sample.data.customSettings import io.getstream.chat.android.compose.sample.ui.StartupActivity import io.getstream.chat.android.models.Channel import io.getstream.chat.android.models.EventType @@ -46,6 +47,20 @@ object ChatHelper { var apiKey: String = "" private set + /** The value of the local unread count setting the SDK was initialized with. */ + private var isLocalUnreadCountEnabled: Boolean = false + + /** + * Initializes the SDK, unless it is already initialized with the given API key and the current settings. + */ + fun ensureInitialized(context: Context, apiKey: String) { + if (this.apiKey != apiKey || + isLocalUnreadCountEnabled != context.customSettings().isLocalUnreadCountEnabled + ) { + initializeSdk(context, apiKey) + } + } + /** * Initializes the SDK with the given API key. */ @@ -83,7 +98,11 @@ object ChatHelper { }, ) - val chatClientConfig = ChatClientConfig(userPresence = true) + this.isLocalUnreadCountEnabled = context.customSettings().isLocalUnreadCountEnabled + val chatClientConfig = ChatClientConfig( + userPresence = true, + isLocalUnreadCountEnabled = isLocalUnreadCountEnabled, + ) val logLevel = if (BuildConfig.DEBUG) ChatLogLevel.ALL else ChatLogLevel.NOTHING diff --git a/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/data/CustomSettings.kt b/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/data/CustomSettings.kt index f238e9443dd..c27561ebfc6 100644 --- a/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/data/CustomSettings.kt +++ b/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/data/CustomSettings.kt @@ -38,6 +38,7 @@ class CustomSettings(private val context: Context) { var isComposerFloatingStyleEnabled: Boolean by booleanPref(ComposerFloatingStyle) var isSystemAttachmentPickerEnabled: Boolean by booleanPref(SystemAttachmentPicker) var isLocationSharingEnabled: Boolean by booleanPref(LocationSharing) + var isLocalUnreadCountEnabled: Boolean by booleanPref(LocalUnreadCount) private fun booleanPref(key: String, default: Boolean = false) = object : ReadWriteProperty { @@ -55,5 +56,6 @@ private const val ComposerLinkPreview = "composer_link_preview" private const val ComposerFloatingStyle = "composer_floating_style" private const val SystemAttachmentPicker = "system_attachment_picker" private const val LocationSharing = "location_sharing" +private const val LocalUnreadCount = "local_unread_count" fun Context.customSettings(): CustomSettings = CustomSettings(this) diff --git a/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/feature/channel/list/ChannelsActivity.kt b/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/feature/channel/list/ChannelsActivity.kt index d377cb3c539..0db23e4b61b 100644 --- a/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/feature/channel/list/ChannelsActivity.kt +++ b/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/feature/channel/list/ChannelsActivity.kt @@ -87,7 +87,6 @@ import io.getstream.chat.android.compose.ui.theme.ChatTheme import io.getstream.chat.android.compose.ui.theme.ChatUiConfig import io.getstream.chat.android.compose.ui.threads.ThreadsScreen import io.getstream.chat.android.compose.viewmodel.channels.ChannelListViewModel -import io.getstream.chat.android.compose.viewmodel.channels.ChannelListViewModelFactory import io.getstream.chat.android.compose.viewmodel.threads.ThreadsViewModelFactory import io.getstream.chat.android.models.Channel import io.getstream.chat.android.models.Message @@ -104,35 +103,8 @@ class ChannelsActivity : ComponentActivity() { private val settings by lazy { customSettings() } - /** - * The provided predefined filter has the following specs: - * - * **Filter:** - * ``` - * Filters.and( - * Filters.eq("type", "messaging"), - * Filters.`in`("members", listOf(currentUserId)), - * Filters.or(Filters.notExists("draft"), Filters.eq("draft", false)), - * ) - * ``` - * - * **Sort:** - * ``` - * QuerySortByField().desc("pinned_at").desc("last_updated") - * ``` - */ private val channelsViewModelFactory by lazy { - val chatClient = ChatClient.instance() - val currentUserId = chatClient.getCurrentUser()?.id ?: "" - ChannelListViewModelFactory( - chatClient = chatClient, - predefinedFilterName = "android_sample_filter", - filterValues = mapOf( - "channel_type" to "messaging", - "user_id" to currentUserId, - ), - chatEventHandlerFactory = CustomChatEventHandlerFactory(), - ) + sampleChannelListViewModelFactory(settings) } private val channelsViewModel: ChannelListViewModel by viewModels { channelsViewModelFactory } diff --git a/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/feature/channel/list/SampleChannelListViewModelFactory.kt b/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/feature/channel/list/SampleChannelListViewModelFactory.kt new file mode 100644 index 00000000000..033847f30de --- /dev/null +++ b/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/feature/channel/list/SampleChannelListViewModelFactory.kt @@ -0,0 +1,59 @@ +/* + * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved. + * + * Licensed under the Stream License; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://github.com/GetStream/stream-chat-android/blob/main/LICENSE + * + * 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.getstream.chat.android.compose.sample.feature.channel.list + +import io.getstream.chat.android.client.ChatClient +import io.getstream.chat.android.compose.sample.data.CustomSettings +import io.getstream.chat.android.compose.viewmodel.channels.ChannelListViewModelFactory +import io.getstream.chat.android.models.Channel +import io.getstream.chat.android.models.Filters +import io.getstream.chat.android.models.querysort.QuerySortByField + +/** + * Builds the channel list factory used by the sample. + * + * When the local unread count is enabled, an explicit filter including livestream channels is used, + * to make the feature testable. Otherwise, the predefined server-side filter is used, which resolves + * to: messaging channels the current user is a member of, without a draft, sorted by "pinned_at" and + * "last_updated" descending. + */ +fun sampleChannelListViewModelFactory(settings: CustomSettings): ChannelListViewModelFactory { + val chatClient = ChatClient.instance() + val currentUserId = chatClient.getCurrentUser()?.id.orEmpty() + return if (settings.isLocalUnreadCountEnabled) { + ChannelListViewModelFactory( + chatClient = chatClient, + querySort = QuerySortByField().desc("pinned_at").desc("last_updated"), + filters = Filters.and( + Filters.`in`("type", listOf("messaging", "livestream")), + Filters.`in`("members", listOf(currentUserId)), + Filters.or(Filters.notExists("draft"), Filters.eq("draft", false)), + ), + chatEventHandlerFactory = CustomChatEventHandlerFactory(), + ) + } else { + ChannelListViewModelFactory( + chatClient = chatClient, + predefinedFilterName = "android_sample_filter", + filterValues = mapOf( + "channel_type" to "messaging", + "user_id" to currentUserId, + ), + chatEventHandlerFactory = CustomChatEventHandlerFactory(), + ) + } +} diff --git a/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ui/chats/ChatsActivity.kt b/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ui/chats/ChatsActivity.kt index e45ec872d96..c79296f6c75 100644 --- a/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ui/chats/ChatsActivity.kt +++ b/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ui/chats/ChatsActivity.kt @@ -53,7 +53,7 @@ import io.getstream.chat.android.compose.sample.R import io.getstream.chat.android.compose.sample.data.customSettings import io.getstream.chat.android.compose.sample.feature.channel.add.AddChannelActivity import io.getstream.chat.android.compose.sample.feature.channel.isGroupChannel -import io.getstream.chat.android.compose.sample.feature.channel.list.CustomChatEventHandlerFactory +import io.getstream.chat.android.compose.sample.feature.channel.list.sampleChannelListViewModelFactory import io.getstream.chat.android.compose.sample.ui.SampleChatTheme import io.getstream.chat.android.compose.sample.ui.channel.MemberRolesTrailingContent import io.getstream.chat.android.compose.sample.ui.component.AppBottomBar @@ -86,7 +86,6 @@ import io.getstream.chat.android.compose.viewmodel.channel.ChannelAttachmentsVie import io.getstream.chat.android.compose.viewmodel.channel.ChannelAttachmentsViewModelFactory import io.getstream.chat.android.compose.viewmodel.channel.ChannelInfoViewModel import io.getstream.chat.android.compose.viewmodel.channel.ChannelInfoViewModelFactory -import io.getstream.chat.android.compose.viewmodel.channels.ChannelListViewModelFactory import io.getstream.chat.android.compose.viewmodel.messages.ChannelViewModelFactory import io.getstream.chat.android.compose.viewmodel.messages.ComposerOptions import io.getstream.chat.android.compose.viewmodel.pinned.PinnedMessageListViewModel @@ -132,35 +131,8 @@ class ChatsActivity : ComponentActivity() { private val settings by lazy { customSettings() } - /** - * The provided predefined filter has the following specs: - * - * **Filter:** - * ``` - * Filters.and( - * Filters.eq("type", "messaging"), - * Filters.`in`("members", listOf(currentUserId)), - * Filters.or(Filters.notExists("draft"), Filters.eq("draft", false)), - * ) - * ``` - * - * **Sort:** - * ``` - * QuerySortByField().desc("pinned_at").desc("last_updated") - * ``` - */ private val channelListViewModelFactory by lazy { - val chatClient = ChatClient.instance() - val currentUserId = chatClient.getCurrentUser()?.id ?: "" - ChannelListViewModelFactory( - chatClient = chatClient, - predefinedFilterName = "android_sample_filter", - filterValues = mapOf( - "channel_type" to "messaging", - "user_id" to currentUserId, - ), - chatEventHandlerFactory = CustomChatEventHandlerFactory(), - ) + sampleChannelListViewModelFactory(settings) } private val channelViewModelFactory by lazy { diff --git a/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ui/login/CustomLoginActivity.kt b/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ui/login/CustomLoginActivity.kt index 19f3048cd44..8637fe73634 100644 --- a/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ui/login/CustomLoginActivity.kt +++ b/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ui/login/CustomLoginActivity.kt @@ -141,6 +141,9 @@ class CustomLoginActivity : AppCompatActivity() { var isLocationSharingEnabled by remember { mutableStateOf(settings.isLocationSharingEnabled) } + var isLocalUnreadCountEnabled by remember { + mutableStateOf(settings.isLocalUnreadCountEnabled) + } val isLoginButtonEnabled = apiKeyText.isNotEmpty() && userIdText.isNotEmpty() && @@ -207,6 +210,15 @@ class CustomLoginActivity : AppCompatActivity() { settings.isLocationSharingEnabled = it }, ), + FeatureFlag( + label = stringResource(R.string.custom_login_flag_local_unread_count_label), + description = stringResource(R.string.custom_login_flag_local_unread_count_description), + value = isLocalUnreadCountEnabled, + onValueChange = { + isLocalUnreadCountEnabled = it + settings.isLocalUnreadCountEnabled = it + }, + ), ) CustomLoginInputField( diff --git a/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ui/login/UserLoginActivity.kt b/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ui/login/UserLoginActivity.kt index 344cf15157e..7cc32cf0e20 100644 --- a/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ui/login/UserLoginActivity.kt +++ b/stream-chat-android-compose-sample/src/main/java/io/getstream/chat/android/compose/sample/ui/login/UserLoginActivity.kt @@ -85,11 +85,7 @@ class UserLoginActivity : AppCompatActivity() { UserLoginScreen( onUserItemClick = { userCredentials -> lifecycleScope.launch { - if (ChatHelper.apiKey != userCredentials.apiKey) { - // If the user attempted to login with custom credentials on the custom - // login screen then we need to reinitialize the SDK with our API key. - ChatHelper.initializeSdk(applicationContext, userCredentials.apiKey) - } + ChatHelper.ensureInitialized(applicationContext, userCredentials.apiKey) ChatHelper.connectUser( userCredentials = userCredentials, onSuccess = ::openChannels, diff --git a/stream-chat-android-compose-sample/src/main/res/values/strings.xml b/stream-chat-android-compose-sample/src/main/res/values/strings.xml index 0ae6fe91339..a31921dcedd 100644 --- a/stream-chat-android-compose-sample/src/main/res/values/strings.xml +++ b/stream-chat-android-compose-sample/src/main/res/values/strings.xml @@ -45,6 +45,8 @@ Add a Location option to the attachment picker for sharing static and live locations Channel pinning Show the Pin/Unpin Chat action in the channel options menu + Local unread count + Track the unread count locally for channels with read events disabled. Also shows livestream channels in the channel list Pinned Messages