feat(llc): Implement local unread counts

docs/docs_screenshots/test/src/mocks.dart

@@ -128,6 +128,7 @@ void setupMockChannel({
   final allMembers = members.isNotEmpty ? members : _defaultMembers(channel.id);
 
   when(() => client.state).thenReturn(clientState);
+  when(() => client.isLocalUnreadCountEnabled).thenReturn(false);
   when(() => channel.lastMessageAt).thenReturn(DateTime.parse('2020-06-22 12:00:00'));
   when(() => channel.lastMessageAtStream).thenAnswer((_) => Stream.value(DateTime.parse('2020-06-22 12:00:00')));
   when(() => channel.currentUserLastMessageAt).thenReturn(DateTime.parse('2020-06-22 12:00:00'));

packages/stream_chat/CHANGELOG.md

@@ -1,5 +1,9 @@
 ## Upcoming
 
+✅ Added
+
+- Added `StreamChatClient.isLocalUnreadCountEnabled` (default `false`). When enabled, channels that have read events disabled (e.g. livestream channel types) track their unread count locally, on-device: incoming messages increment it, hard-deleted messages decrement it, and `Channel.markRead` / `markUnread` / `markUnreadByTimestamp` update it locally without a network request — including `Read.lastReadMessageId`, so the unread divider and jump-to-unread button anchor to the right message. Channels that support read receipts are unaffected and keep relying on server-driven unread counts.
+
 🔄 Changed
 
 - `StreamChatClient.updateSystemEnvironment` now sanitizes the passed `SystemEnvironment`: `sdkName`, `sdkVersion`, and `osName` are locked to internal defaults, and `sdkIdentifier` only accepts the `dart` → `flutter` promotion (other values, including a `flutter` → `dart` demotion, are ignored). `appName`, `appVersion`, `osVersion`, and `deviceModel` continue to pass through as-is.

packages/stream_chat/lib/src/client/channel.dart

@@ -1852,9 +1852,17 @@ class Channel {
   ///
   /// Optionally provide a [messageId] if you want to mark channel as
   /// read from a particular message onwards.
+  ///
+  /// If [usesLocalUnreadCount] is `true` for this channel, this updates the
+  /// unread count locally, on-device, without making a network request.
   Future<EmptyResponse> markRead({String? messageId}) async {
     _checkInitialized();
 
+    if (usesLocalUnreadCount) {
+      state!.markReadLocally(messageId: messageId);
+      return EmptyResponse();
+    }
+
     if (!canUseReadReceipts) {
       throw const StreamChatError(
         'Cannot mark as read: Channel does not support read events. '
@@ -1867,10 +1875,43 @@ class Channel {
 
   /// Marks the channel as unread by a given [messageId].
   ///
-  /// All messages from the provided message onwards will be marked as unread.
+  /// All messages from the provided message onwards will be marked as unread,
+  /// **including** the message itself. Contrast with
+  /// [markUnreadByTimestamp], which is exclusive: a message created at
+  /// exactly the given timestamp stays read.
+  ///
+  /// If [usesLocalUnreadCount] is `true` for this channel, this updates the
+  /// unread count locally, on-device, without making a network request. The
+  /// message must be part of the locally-known messages ([Channel.messages])
+  /// for the count to be recomputed.
   Future<EmptyResponse> markUnread(String messageId) async {
     _checkInitialized();
 
+    if (usesLocalUnreadCount) {
+      // [ChannelClientState.messages] is sorted ascending by `createdAt`, so
+      // the entry before the anchor is the newest message that stays read.
+      final messages = state!.messages;
+      final anchorIndex = messages.indexWhere((it) => it.id == messageId);
+      if (anchorIndex < 0) {
+        throw StreamChatError(
+          'Cannot mark as unread: Message "$messageId" was not found in the '
+          'locally-known messages for this channel.',
+        );
+      }
+
+      final anchor = messages[anchorIndex];
+      final previous = anchorIndex > 0 ? messages[anchorIndex - 1] : null;
+
+      // Subtract a microsecond so the anchor message itself is treated as
+      // "after" the new read boundary, matching the "from the provided
+      // message onwards" semantics described above. Preferred over using
+      // `previous.createdAt` as the boundary, which would leak the anchor
+      // back into the read set if the two share an identical `createdAt`.
+      final lastRead = anchor.createdAt.subtract(const Duration(microseconds: 1));
+      state!.markUnreadLocally(lastRead: lastRead, lastReadMessageId: previous?.id);
+      return EmptyResponse();
+    }
+
     if (!canUseReadReceipts) {
       throw const StreamChatError(
         'Cannot mark as unread: Channel does not support read events. '
@@ -1883,10 +1924,32 @@ class Channel {
 
   /// Marks the channel as unread by a given [timestamp].
   ///
-  /// All messages after the provided timestamp will be marked as unread.
+  /// All messages after the provided timestamp will be marked as unread. This
+  /// boundary is **exclusive**: a message created at exactly [timestamp] stays
+  /// read. Contrast with [markUnread], which is inclusive of the message it is
+  /// given — `markUnread(m.id)` is equivalent to
+  /// `markUnreadByTimestamp(m.createdAt - 1µs)`, not to
+  /// `markUnreadByTimestamp(m.createdAt)`.
+  ///
+  /// If [usesLocalUnreadCount] is `true` for this channel, this updates the
+  /// unread count locally, on-device, without making a network request.
   Future<EmptyResponse> markUnreadByTimestamp(DateTime timestamp) async {
     _checkInitialized();
 
+    if (usesLocalUnreadCount) {
+      // The newest locally-known message at or before the boundary is the last
+      // one that stays read.
+      final lastReadMessage = state!.messages.lastWhereOrNull(
+        (it) => !it.createdAt.isAfter(timestamp),
+      );
+
+      state!.markUnreadLocally(
+        lastRead: timestamp,
+        lastReadMessageId: lastReadMessage?.id,
+      );
+      return EmptyResponse();
+    }
+
     if (!canUseReadReceipts) {
       throw const StreamChatError(
         'Cannot mark as unread: Channel does not support read events. '
@@ -2096,7 +2159,7 @@ class Channel {
         };
 
         if (isQueryingAround) this.state?.truncate();
-        this.state?.updateChannelState(channelState);
+        this.state?.updateChannelStateFromServer(channelState);
       }
 
       // Submit for delivery reporting only when fetching the latest messages.
@@ -3209,6 +3272,15 @@ class ChannelClientState {
           deletedForMe: event.deletedForMe,
         );
 
+        // Decrement the locally-tracked unread count for hard-deleted
+        // messages that would have counted as unread. Soft-deleted messages
+        // keep their slot. Only applies to channels that track unread counts
+        // locally (see [Channel.usesLocalUnreadCount]) — server-driven
+        // channels get corrected counts from server read events instead.
+        if (hardDelete && _channel.usesLocalUnreadCount && MessageRules.canCountAsUnread(message, _channel)) {
+          unreadCount = math.max(0, unreadCount - 1);
+        }
+
         return deleteMessage(message, hardDelete: hardDelete);
       }),
     );
@@ -3563,6 +3635,75 @@ class ChannelClientState {
     return updateRead([existingUserRead.copyWith(unreadMessages: count)]);
   }
 
+  /// Marks the channel as read locally, without making a network request.
+  ///
+  /// Used for channels that track unread counts locally (see
+  /// [Channel.usesLocalUnreadCount]), since the server rejects the mark-read
+  /// endpoint for channels that have read events disabled.
+  void markReadLocally({String? messageId}) {
+    final currentUser = _client.state.currentUser;
+    if (currentUser == null) return;
+
+    final now = DateTime.now();
+    final lastReadMessageId = messageId ?? messages.lastOrNull?.id;
+
+    final existingUserRead = currentUserRead;
+    updateRead([
+      Read(
+        user: currentUser,
+        lastRead: now,
+        lastReadMessageId: lastReadMessageId,
+        lastDeliveredAt: existingUserRead?.lastDeliveredAt,
+        lastDeliveredMessageId: existingUserRead?.lastDeliveredMessageId,
+      ),
+    ]);
+
+    // Read supersedes delivered, so drop any pending delivery candidate the
+    // new read boundary just made ineligible. `delivery_events` is configured
+    // independently of `read_events`, so a channel tracking unread counts
+    // locally can still have delivery receipts enabled. Mirrors what the
+    // `message.read` event listener does for server-driven channels.
+    _client.channelDeliveryReporter.reconcileDelivery([_channel]);
+  }
+
+  /// Marks the channel as unread locally, without making a network request.
+  ///
+  /// [lastRead] and [lastReadMessageId] define the new read boundary: any
+  /// locally-known message that is still eligible per
+  /// [MessageRules.canCountAsUnread] once this boundary is applied is counted
+  /// as unread.
+  ///
+  /// Used for channels that track unread counts locally (see
+  /// [Channel.usesLocalUnreadCount]), since the server rejects the
+  /// mark-unread endpoint for channels that have read events disabled.
+  void markUnreadLocally({
+    required DateTime lastRead,
+    String? lastReadMessageId,
+  }) {
+    final currentUser = _client.state.currentUser;
+    if (currentUser == null) return;
+
+    final existingUserRead = currentUserRead;
+
+    // Apply the new read boundary first so `MessageRules.canCountAsUnread`
+    // (which reads `channel.state?.currentUserRead`) evaluates against it.
+    updateRead([
+      Read(
+        user: currentUser,
+        lastRead: lastRead,
+        lastReadMessageId: lastReadMessageId,
+        lastDeliveredAt: existingUserRead?.lastDeliveredAt,
+        lastDeliveredMessageId: existingUserRead?.lastDeliveredMessageId,
+      ),
+    ]);
+
+    // Recompute the unread count from the locally-known messages now that
+    // the boundary above is in effect.
+    final unread = messages.where((it) => MessageRules.canCountAsUnread(it, _channel)).length;
+
+    unreadCount = unread;
+  }
+
   /// Counts the number of unread messages mentioning the current user.
   ///
   /// **NOTE**: The method relies on the [Channel.messages] list and doesn't do
@@ -3645,6 +3786,47 @@ class ChannelClientState {
     );
   }
 
+  /// Applies a [remoteState] received from the server or offline storage
+  /// (e.g. a `query`/`watch` response), merging it into local state.
+  ///
+  /// Unlike [updateChannelState], this preserves the current user's
+  /// locally-tracked read state for channels that track unread counts
+  /// on-device (see [Channel.usesLocalUnreadCount]) — their `lastRead`,
+  /// `lastReadMessageId`, and `unreadMessages` are kept as-is instead of
+  /// being overwritten by the remote payload; only delivery fields are
+  /// still applied from it.
+  ///
+  /// Call this instead of [updateChannelState] whenever [remoteState]
+  /// genuinely comes from the network or offline storage.
+  void updateChannelStateFromServer(ChannelState remoteState) {
+    updateChannelState(_preserveLocalUnreadState(remoteState));
+  }
+
+  /// Rewrites the current user's [Read] in [remoteState], if present, to
+  /// keep the locally-tracked `lastRead` / `lastReadMessageId` /
+  /// `unreadMessages` while still adopting the remote delivery fields.
+  ///
+  /// No-op unless [Channel.usesLocalUnreadCount] is enabled and a local read
+  /// already exists for the current user.
+  ChannelState _preserveLocalUnreadState(ChannelState remoteState) {
+    if (!_channel.usesLocalUnreadCount) return remoteState;
+
+    final localRead = currentUserRead;
+    final remoteReads = remoteState.read;
+    if (localRead == null || remoteReads == null) return remoteState;
+
+    final currentUserId = localRead.user.id;
+    final preservedReads = remoteReads.map((read) {
+      if (read.user.id != currentUserId) return read;
+      return localRead.copyWith(
+        lastDeliveredAt: read.lastDeliveredAt,
+        lastDeliveredMessageId: read.lastDeliveredMessageId,
+      );
+    });
+
+    return remoteState.copyWith(read: preservedReads.toList());
+  }
+
   int _sortByCreatedAt(Message a, Message b) => a.createdAt.compareTo(b.createdAt);
 
   /// The channel state related to this client.
@@ -4554,6 +4736,17 @@ extension ChannelCapabilityCheck on Channel {
     return ownCapabilities.contains(ChannelCapability.readEvents);
   }
 
+  /// True, if unread counts for this channel should be tracked locally,
+  /// on-device, rather than relying on the server.
+  ///
+  /// This is the case when [StreamChatClient.isLocalUnreadCountEnabled] is
+  /// enabled and the channel doesn't support read receipts (for example,
+  /// livestream channel types that disable read events). Channels that
+  /// support read receipts always rely on server-driven unread counts.
+  bool get usesLocalUnreadCount {
+    return _client.isLocalUnreadCountEnabled && !canUseReadReceipts;
+  }
+
   /// True, if the current user has connect events capability.
   bool get canReceiveConnectEvents {
     return ownCapabilities.contains(ChannelCapability.connectEvents);

packages/stream_chat/lib/src/client/client.dart

@@ -101,6 +101,7 @@ class StreamChatClient {
     Iterable<Interceptor>? chatApiInterceptors,
     HttpClientAdapter? httpClientAdapter,
     bool recoverStateOnReconnect = true,
+    this.isLocalUnreadCountEnabled = false,
   }) : _recoverStateOnReconnect = recoverStateOnReconnect {
     logger.info('Initiating new StreamChatClient');
 
@@ -212,6 +213,21 @@ class StreamChatClient {
   /// Chat persistence client
   ChatPersistenceClient? chatPersistenceClient;
 
+  /// Whether the SDK should track unread counts locally, on-device, for
+  /// channels that have read events disabled (e.g. livestream channel
+  /// types).
+  ///
+  /// Channels with read events disabled never receive `message.read` /
+  /// `notification.mark_*` events from the server, and reject the mark-read
+  /// endpoint, so their unread count is always `0` by default.
+  ///
+  /// When this is `true`, [Channel.unreadCount] is instead incremented
+  /// locally as new messages arrive and reset locally (with no network
+  /// request) when [Channel.markRead] is called, for those channels only.
+  /// Channels with read events enabled are unaffected and keep relying on
+  /// server-driven unread counts.
+  final bool isLocalUnreadCountEnabled;
+
   /// Returns `True` if the [chatPersistenceClient] is available and connected.
   /// Otherwise, returns `False`.
   bool get persistenceEnabled {
@@ -1014,7 +1030,7 @@ class StreamChatClient {
     for (final channelState in channelStates) {
       final channel = channels[channelState.channel!.cid];
       if (channel != null) {
-        channel.state?.updateChannelState(channelState);
+        channel.state?.updateChannelStateFromServer(channelState);
         newChannels.add(channel);
       } else {
         final newChannel = Channel.fromState(this, channelState);

packages/stream_chat/lib/src/core/util/message_rules.dart

@@ -68,7 +68,8 @@ class MessageRules {
   /// Returns `false` for the current user's own messages, messages from muted
   /// users, silent/shadowed/ephemeral messages, thread-only replies, restricted
   /// messages, and messages already read. Also returns `false` if the channel
-  /// is muted or doesn't support read events.
+  /// is muted, or doesn't support read events unless
+  /// [Channel.usesLocalUnreadCount] is enabled for the channel.
   static bool canCountAsUnread(
     Message message,
     Channel channel,
@@ -80,8 +81,12 @@ class MessageRules {
     // Don't count if the user has disabled read receipts.
     if (!currentUser.isReadReceiptsEnabled) return false;
 
-    // Don't count if the channel doesn't support read receipts.
-    if (!channel.canUseReadReceipts) return false;
+    // Don't count if the channel doesn't support read receipts, unless local
+    // unread tracking owns the count for this channel (see
+    // [Channel.usesLocalUnreadCount]).
+    if (!channel.canUseReadReceipts && !channel.usesLocalUnreadCount) {
+      return false;
+    }
 
     // Don't count if the channel is muted.
     if (channel.isMuted) return false;