feat: localized unread count (#1787)

## CLA

- [ ] I have signed the [Stream
CLA](https://docs.google.com/forms/d/e/1FAIpQLScFKsKkAJI7mhCr7K9rEIOpqIDThrWxuvxnwUq2XkHyG154vQ/viewform)
(required).
- [ ] Code changes are tested

## Description of the changes, What, Why and How?

Spec:
https://stream-wiki.notion.site/Local-Unread-Count-specification-38b6a5d7f9f68073a5e1c73298fe0d71
Adds an optin, client side unread count for channels that have read
events disabled (e.g. livestreams). Those channels normally don't track
unread at all and so `_countMessageAsUnread` bails out whenever
`own_capabilities` lacks `read-events`, so integrators have no "N new
messages" affordance to show while scrolled up.
With the new `isLocalUnreadCountEnabled` client option (default
`false`), the SDK maintains a purely local unread count for those
channels. It increments on incoming messages and is reset via the new
`channel.markReadLocally()` method, which dispatches a client only
`message.local_read` event. That event reuses the exact `message.read`
state logic in `_handleChannelEvent` (so read state updates stay in one
place) but skips `syncDeliveredCandidates()`, so the count is never sent
to the backend. `hasReadEvents()` centralizes the capability check.
When offline support is enabled, the local read/unread state is
persisted for channels with disabled `read-events` so the count survives
a cold start. The offline DB handles `message.local_read` (guarded so it
can only affect channels with disabled `read-events` channels with the
option on) and falls back gracefully when `state.read[userId]` is
absent, since the server never sends a read for these channels.
Behavior is fully gated behind the flag, so default off means no change
for existing integrations.
## Changelog
- Add `isLocalUnreadCountEnabled` client option for a client side unread
count on channels with disabled `read-events` (e.g. livestreams)
- Add `Channel.markReadLocally()` to reset the local unread count with
no backend call
- Add `Channel.hasReadEvents()` helper
- Add client only `message.local_read` event
- Persist local unread state to the offline DB so it survives app
restarts

Author: isekovanic

src/channel.ts

@@ -4,6 +4,7 @@ import { CooldownTimer } from './CooldownTimer';
 import { MessageComposer } from './messageComposer';
 import { MessageReceiptsTracker } from './messageDelivery';
 import {
+  channelHasReadEvents,
   generateChannelTempCid,
   logChatPromiseExecution,
   messageSetPagination,
@@ -1247,6 +1248,36 @@ export class Channel {
     });
   }
 
+  /**
+   * markReadLocally - Resets this user's unread count locally, without any backend call. Intended for
+   * channels that have read events disabled (e.g. livestreams) when the client is created with the
+   * `isLocalUnreadCountEnabled` option. Dispatches a dedicated, client-only `message.read_locally` event
+   * that runs through the same `_handleChannelEvent` read logic as a real `message.read` (minus the
+   * delivery-report network sync), so the read-state update lives in one place. When offline support
+   * is enabled, the offline DB persists the reset for read-events-disabled channels, so the local
+   * count stays consistent across app restarts.
+   *
+   * @return {Event | undefined} The dispatched `message.read_locally` event, or `undefined` if there is no connected user.
+   */
+  markReadLocally() {
+    const client = this.getClient();
+    if (!client.userID) return;
+
+    const event: Event = {
+      channel_id: this.id,
+      channel_type: this.type,
+      cid: this.cid,
+      created_at: new Date().toISOString(),
+      last_read_message_id: this.lastMessage()?.id,
+      team: this.data?.team,
+      type: 'message.read_locally',
+      user: client.user,
+    };
+    client.dispatchEvent(event);
+
+    return event;
+  }
+
   /**
    * clean - Cleans the channel state and fires stop typing if needed
    */
@@ -1429,10 +1460,11 @@ export class Channel {
     if (message.user?.id && this.getClient().userMuteStatus(message.user.id))
       return false;
 
-    // Return false if channel doesn't allow read events.
+    // Return false if channel doesn't allow read events, unless the client opted into a local
+    // unread count (e.g. livestreams where read events are disabled). See `isLocalUnreadCountEnabled`.
     if (
-      Array.isArray(this.data?.own_capabilities) &&
-      !this.data?.own_capabilities.includes('read-events')
+      !this.getClient().options.isLocalUnreadCountEnabled &&
+      !channelHasReadEvents(this)
     ) {
       return false;
     }
@@ -1965,6 +1997,11 @@ export class Channel {
           delete channelState.typing[event.user.id];
         }
         break;
+      // `message.read_locally` is the client-only event dispatched by `markReadLocally()` when read
+      // events are disabled (e.g. livestreams with `isLocalUnreadCountEnabled`). It reuses the exact
+      // `message.read` state logic so the read-state update lives in one place — only the
+      // delivery-report network sync below is skipped for it.
+      case 'message.read_locally':
       case 'message.read':
         if (event.user?.id && event.created_at) {
           const previousReadState = channelState.read[event.user.id];
@@ -1987,7 +2024,11 @@ export class Channel {
 
           if (isOwnEvent) {
             channelState.unreadCount = 0;
-            client.syncDeliveredCandidates([this]);
+            // Delivery reporting buffers a `markChannelsDelivered` network request; the local read
+            // must not hit the backend, so only sync for the real server `message.read`.
+            if (event.type === 'message.read') {
+              client.syncDeliveredCandidates([this]);
+            }
           }
         }
         break;

src/client.ts

@@ -470,6 +470,7 @@ export class StreamChat {
       warmUp: false,
       recoverStateOnReconnect: true,
       disableCache: false,
+      isLocalUnreadCountEnabled: false,
       wsUrlParams: new URLSearchParams({}),
       ...inputOptions,
     };

src/events.ts

@@ -63,6 +63,7 @@ export const EVENT_MAP = {
   'ai_indicator.clear': true,
 
   // local events
+  'message.read_locally': true,
   'channels.queried': true,
   'offline_reactions.queried': true,
   'connection.changed': true,

src/messageDelivery/MessageDeliveryReporter.ts

@@ -10,7 +10,7 @@ import type {
   MarkReadOptions,
 } from '../types';
 import { type APIErrorResponse } from '../types';
-import { throttle } from '../utils';
+import { throttle, userHasReadReceipts } from '../utils';
 import { isAPIError, isErrorRetryable } from '../errors';
 
 const MAX_DELIVERED_MESSAGE_COUNT_IN_PAYLOAD = 100 as const;
@@ -279,6 +279,8 @@ export class MessageDeliveryReporter {
    * @param options
    */
   public markRead = async (collection: Channel | Thread, options?: MarkReadOptions) => {
+    if (!userHasReadReceipts(this.client)) return null;
+
     let result: EventAPIResponse | null = null;
     if (isChannel(collection)) {
       result = await collection.markAsReadRequest(options);

src/offline-support/offline_support_api.ts

@@ -18,7 +18,12 @@ import type { StreamChat } from '../client';
 import type { AxiosError } from 'axios';
 import { OfflineDBSyncManager } from './offline_sync_manager';
 import { StateStore } from '../store';
-import { localMessageToNewMessagePayload, runDetached } from '../utils';
+import {
+  channelHasReadEvents,
+  channelTracksReadLocally,
+  localMessageToNewMessagePayload,
+  runDetached,
+} from '../utils';
 import { isMessageUpdateReplayable } from './util';
 
 /**
@@ -609,16 +614,22 @@ export abstract class AbstractOfflineDB implements OfflineDBApi {
         if (cid && client.user && client.user.id !== user?.id) {
           const userId = client.user.id;
           const channel = client.activeChannels[cid];
-          if (channel) {
+          // Persist the current user's read state for channels that track reads server-side. When the
+          // client opted into a local unread count, also persist it for read-events-disabled channels
+          // (e.g. livestreams) so the client-local count survives a cold start. The server never sends
+          // a read for those, so state.read[userId] may be absent here; fall back accordingly and rely
+          // on countUnread() (the aggregate the local count maintains) for the value.
+          const tracksReadLocally = channelTracksReadLocally(channel);
+          if (channel && (channelHasReadEvents(channel) || tracksReadLocally)) {
             const ownReads = channel.state.read[userId];
             const unreadCount = channel.countUnread();
             const upsertReadsQueries = await this.upsertReads({
               cid,
               execute: false,
               reads: [
                 {
-                  last_read: ownReads.last_read.toISOString() as string,
-                  last_read_message_id: ownReads.last_read_message_id,
+                  last_read: (ownReads?.last_read ?? new Date(0)).toISOString() as string,
+                  last_read_message_id: ownReads?.last_read_message_id,
                   unread_messages: unreadCount,
                   user: client.user,
                 },
@@ -867,32 +878,38 @@ export abstract class AbstractOfflineDB implements OfflineDBApi {
         execute: false,
       });
 
+      let finalQueries = [...truncateQueries];
+
+      // Persist read state for channels that track reads server-side. When the client opted into a
+      // local unread count, also persist it for read-events-disabled channels (e.g. livestreams) so
+      // the client-local count survives a cold start. state.read[userId] may be absent for those, so
+      // fall back accordingly.
       const userId = ownUser.id;
       const activeChannel = this.client.activeChannels[cid];
-      const ownReads = activeChannel.state.read[userId];
+      const tracksReadLocally = channelTracksReadLocally(activeChannel);
+      if (activeChannel && (channelHasReadEvents(activeChannel) || tracksReadLocally)) {
+        const ownReads = activeChannel.state.read[userId];
 
-      let unreadCount = 0;
+        let unreadCount = 0;
+        if (truncated_at) {
+          unreadCount = activeChannel.countUnread(new Date(truncated_at));
+        }
 
-      if (truncated_at) {
-        const truncatedAt = new Date(truncated_at);
-        unreadCount = activeChannel.countUnread(truncatedAt);
+        const upsertReadQueries = await this.upsertReads({
+          cid,
+          execute: false,
+          reads: [
+            {
+              last_read: (ownReads?.last_read ?? new Date(0)).toString() as string,
+              last_read_message_id: ownReads?.last_read_message_id,
+              unread_messages: unreadCount,
+              user: ownUser,
+            },
+          ],
+        });
+        finalQueries = [...finalQueries, ...upsertReadQueries];
       }
 
-      const upsertReadQueries = await this.upsertReads({
-        cid,
-        execute: false,
-        reads: [
-          {
-            last_read: ownReads.last_read.toString() as string,
-            last_read_message_id: ownReads.last_read_message_id,
-            unread_messages: unreadCount,
-            user: ownUser,
-          },
-        ],
-      });
-
-      const finalQueries = [...truncateQueries, ...upsertReadQueries];
-
       if (execute) {
         await this.executeSqlBatch(finalQueries);
       }
@@ -1021,6 +1038,18 @@ export abstract class AbstractOfflineDB implements OfflineDBApi {
       return this.handleRead({ event, unreadMessages: 0, execute });
     }
 
+    // `message.read_locally` is the client-only reset dispatched by `Channel.markReadLocally()`. Only
+    // persist it for read-events-disabled channels with the local unread count opted in (the only
+    // situation it is ever dispatched), so it can never touch channels that track reads server-side.
+    if (type === 'message.read_locally') {
+      const localChannel = event.cid ? this.client.activeChannels[event.cid] : undefined;
+      // channelTracksReadLocally returns false when localChannel is undefined, so no extra guard.
+      if (channelTracksReadLocally(localChannel)) {
+        return this.handleRead({ event, unreadMessages: 0, execute });
+      }
+      return [];
+    }
+
     if (type === 'notification.mark_unread') {
       return this.handleRead({ event, execute });
     }

src/types.ts

@@ -1591,6 +1591,12 @@ export type StreamChatOptions = AxiosRequestConfig & {
    */
   disableCache?: boolean;
   enableInsights?: boolean;
+  /**
+   * When true, maintains a client-local unread count on channels that have read events disabled
+   * (e.g. livestreams). The count increments on incoming messages and is reset via
+   * `channel.markReadLocally()`. It is never sent to the backend, but is persisted to the offline DB.
+   */
+  isLocalUnreadCountEnabled?: boolean;
   /** experimental feature, please contact support if you want this feature enabled for you */
   enableWSFallback?: boolean;
   logger?: Logger;

src/utils.ts

@@ -110,6 +110,28 @@ export function isOwnUserBaseProperty(property: string) {
   return ownUserBaseProperties[property as keyof OwnUserBase];
 }
 
+/**
+ * channelHasReadEvents - Whether read events are enabled for the current user on a channel.
+ */
+export const channelHasReadEvents = (channel?: Channel) => {
+  const ownCapabilities = channel?.data?.own_capabilities;
+  return !(Array.isArray(ownCapabilities) && !ownCapabilities.includes('read-events'));
+};
+
+/**
+ * channelTracksReadLocally - Whether a channel maintains a client local unread count.
+ */
+export const channelTracksReadLocally = (channel?: Channel) =>
+  !channelHasReadEvents(channel) &&
+  !!channel?.getClient().options.isLocalUnreadCountEnabled;
+
+/**
+ * userHasReadReceipts - Whether the current user allows read receipts, per their privacy settings.
+ * Read receipts are treated as enabled unless the user has explicitly disabled them.
+ */
+export const userHasReadReceipts = (client: StreamChat) =>
+  client.user?.privacy_settings?.read_receipts?.enabled ?? true;
+
 export function addFileToFormData(
   uri: string | NodeJS.ReadableStream | Buffer | File,
   name?: string,