feat: optimistic reactions

src/channel.ts

@@ -1,6 +1,8 @@
 import type { AxiosRequestConfig } from 'axios';
 import { ChannelState } from './channel_state';
 import { CooldownTimer } from './CooldownTimer';
+import { isEphemeral } from './errors';
+import { applyReactionLocally } from './messageStore';
 import { MessageComposer } from './messageComposer';
 import { MessageReceiptsTracker } from './messageDelivery';
 import type { ReadStoreReconcileMeta } from './messageDelivery';
@@ -46,9 +48,10 @@ import type {
   PinnedMessagesSort,
   QueryMembersPayload,
   ReactionAPIResponse,
-  ReactionResponse,
+  ReactionRequest,
   SearchPayload,
   SendMessageOptions,
+  SendReactionRequest,
   SharedLocation,
   UnBanUserOptions,
   UpdateChannelPartialRequest,
@@ -240,7 +243,10 @@ export class Channel extends ChannelApi {
     this.cooldownTimer = new CooldownTimer({ channel: this });
 
     this.messageOperations = new MessageOperations({
-      ingest: (m) => this.messagePaginator.ingestItem(m),
+      ingest: (m) => {
+        this.messagePaginator.ingestItem(m);
+        this.getClient().messageStore.flushSubscribers(m.id);
+      },
       get: (id) => this.messagePaginator.getItem(id),
       handlers: () => {
         const { requestHandlers } = this.configState.getLatestValue();
@@ -416,6 +422,76 @@ export class Channel extends ChannelApi {
     );
   }
 
+  /**
+   * Adds a reaction with an optimistic local state update: the reaction is applied to the cached
+   * message immediately ({@link applyReactionLocally}), then the request is
+   * fired via {@link Channel.sendReaction} (which owns the offline-DB write + queue). The
+   * server-authoritative counts reconcile on the response; the message is rolled back on failure.
+   */
+  async addReactionWithLocalUpdate({
+    messageId,
+    reaction,
+    options,
+  }: {
+    messageId: string;
+    reaction: ReactionRequest;
+    options?: Pick<SendReactionRequest, 'enforce_unique' | 'skip_push'>;
+  }) {
+    const client = this.getClient();
+    const undo = applyReactionLocally(client, {
+      enforceUnique: options?.enforce_unique ?? false,
+      messageId,
+      reaction,
+    });
+
+    try {
+      const response = await this.sendReaction({ id: messageId, reaction, ...options });
+      // reconcile the server copy only if we still hold it — a bare upsert of an unheld id would
+      // orphan it (the store's refcount GC only reclaims held ids).
+      if (response?.message && client.messageStore.has(response.message.id)) {
+        client.messageStore.upsert(formatMessage(response.message));
+      }
+    } catch (error) {
+      if (undo && (!client.offlineDb || !isEphemeral(error as Error))) {
+        undo();
+      }
+      throw error;
+    }
+  }
+
+  /**
+   * Removes the current user's reaction with an optimistic local state update, mirroring
+   * {@link Channel.addReactionWithLocalUpdate}.
+   */
+  async deleteReactionWithLocalUpdate({
+    messageId,
+    type,
+  }: {
+    messageId: string;
+    type: string;
+  }) {
+    const client = this.getClient();
+    const undo = applyReactionLocally(client, {
+      messageId,
+      reaction: { type },
+      removed: true,
+    });
+
+    try {
+      const response = await this.deleteReaction({ id: messageId, type });
+      // reconcile the server copy only if we still hold it — a bare upsert of an unheld id would
+      // orphan it (the store's refcount GC only reclaims held ids).
+      if (response?.message && client.messageStore.has(response.message.id)) {
+        client.messageStore.upsert(formatMessage(response.message));
+      }
+    } catch (error) {
+      if (undo && (!client.offlineDb || !isEphemeral(error as Error))) {
+        undo();
+      }
+      throw error;
+    }
+  }
+
   /**
    * Upload a file to this channel’s file endpoint (multipart). Forwards to the client’s `sendFile` implementation.
    *
@@ -549,6 +625,8 @@ export class Channel extends ChannelApi {
     try {
       const offlineDb = this.getClient().offlineDb;
       if (offlineDb) {
+        // The optimistic reaction row is written by the local-update layer
+        // (`applyReactionLocally`); here we only queue the request for replay.
         return await offlineDb.queueTask<ReactionAPIResponse>({
           task: {
             channelId: this.id as string,
@@ -578,19 +656,8 @@ export class Channel extends ChannelApi {
     try {
       const offlineDb = this.getClient().offlineDb;
       if (offlineDb) {
-        const message = this.messagePaginator.getItem(request.id);
-        const reaction = {
-          message_id: request.id,
-          type: request.type,
-        } as ReactionResponse;
-
-        if (message) {
-          await offlineDb.deleteReaction({
-            message,
-            reaction,
-          });
-        }
-
+        // The optimistic reaction-row removal is handled by the local-update layer
+        // (`applyReactionLocally`); here we only queue the request for replay.
         return await offlineDb.queueTask<ReactionAPIResponse>({
           task: {
             channelId: this.id as string,
@@ -2228,7 +2295,9 @@ export class Channel extends ChannelApi {
       case 'reaction.new':
         if (event.message && event.reaction) {
           const { reaction } = event;
-          if (!event.message?.parent_id) {
+          // Reflect main messages AND show_in_channel replies (both live in these paginators);
+          // pure replies are handled by the thread's own reaction subscription.
+          if (!event.message?.parent_id || event.message.show_in_channel) {
             this.messagePaginator.reflectReaction({ message: event.message, reaction });
             this.pinnedMessagesPaginator.reflectReaction({
               message: event.message,
@@ -2240,7 +2309,10 @@ export class Channel extends ChannelApi {
       case 'reaction.deleted':
         if (event.message && event.reaction) {
           const { reaction } = event;
-          if (event.message && !event.message.parent_id) {
+          if (
+            event.message &&
+            (!event.message.parent_id || event.message.show_in_channel)
+          ) {
             this.messagePaginator.reflectReaction({
               message: event.message,
               reaction,
@@ -2258,7 +2330,7 @@ export class Channel extends ChannelApi {
         if (event.message && event.reaction) {
           const { reaction } = event;
           // assuming reaction.updated is only called if enforce_unique is true
-          if (!event.message?.parent_id) {
+          if (!event.message?.parent_id || event.message.show_in_channel) {
             this.messagePaginator.reflectReaction({
               enforceUnique: true,
               message: event.message,
@@ -2521,5 +2593,10 @@ export class Channel extends ChannelApi {
     this.disconnected = true;
     this.messageReceiptsTracker.unregisterSubscriptions();
     this.cooldownTimer.clearTimeout();
+    // Release the store-backed paginators so the message store no longer pins this removed channel
+    // (and its whole message graph) through its subscriber registry. The channel is being discarded
+    // here (disconnected + deleted from activeChannels, never reused), mirroring Thread teardown.
+    this.messagePaginator.dispose();
+    this.pinnedMessagesPaginator.dispose();
   }
 }

src/client.ts

@@ -71,6 +71,7 @@ import { Moderation } from './moderation';
 import { ThreadManager } from './thread_manager';
 import { DEFAULT_QUERY_CHANNELS_MESSAGE_LIST_PAGE_SIZE } from './constants';
 import { PollManager } from './poll_manager';
+import { MessageStore } from './messageStore/MessageStore';
 import type {
   ChannelManagerEventHandlerOverrides,
   ChannelManagerOptions,
@@ -138,6 +139,12 @@ export class StreamChat extends ChatApi {
   };
   threads: ThreadManager;
   polls: PollManager;
+  /**
+   * Client-global, normalized store holding one canonical copy of each message. The channel main
+   * list and thread reply paginators read/write message content through it, so a message held in
+   * more than one of them stays consistent without copy-to-copy fan-out.
+   */
+  messageStore: MessageStore;
   offlineDb?: AbstractOfflineDB;
   notifications: NotificationManager;
   reminders: ReminderManager;
@@ -314,6 +321,7 @@ export class StreamChat extends ChatApi {
     this.defaultWSTimeout = 15 * 1000;
 
     this.recoverStateOnReconnect = this.options.recoverStateOnReconnect;
+    this.messageStore = new MessageStore();
     this.threads = new ThreadManager({ client: this });
     this.polls = new PollManager({ client: this });
     this.reminders = new ReminderManager({ client: this });

src/errors.ts

@@ -48,6 +48,18 @@ export function isErrorRetryable(error: APIError) {
   return err.retryable;
 }
 
+/**
+ * Whether an error is EPHEMERAL — a transient failure worth queueing/retrying rather than a
+ * definitive rejection. True when the server never responded (connection/network/offline error - no
+ * `response`, i.e an axios network error or an `OfflineError`) and when the server responded with a
+ * retryable code (see {@link APIErrorCodes}); false only when the server responded with a
+ * non-retryable code (InputError 4, DoesNotExist 16, NotAllowed 17, …).
+ */
+export function isEphemeral(error: Error): boolean {
+  if (!(error as { response?: unknown }).response) return true;
+  return isErrorRetryable(error as APIError);
+}
+
 export function isConnectionIDError(error: APIError) {
   return error.code === 46; // ConnectionIDNotFoundError
 }

src/index.ts

@@ -10,6 +10,11 @@ export * from './insights';
 export * from './logger';
 export * from './messageComposer';
 export * from './messageDelivery';
+export { MessageStore } from './messageStore/MessageStore';
+export type {
+  MessageStoreChangeBatch,
+  MessageStoreSubscriber,
+} from './messageStore/MessageStore';
 export * from './middleware';
 export * from './moderation';
 export * from './notifications';

src/messageDelivery/MessageDeliveryReporter.ts

@@ -359,5 +359,5 @@ export class MessageDeliveryReporter {
       leading: true,
       trailing: true,
     },
-  );
+  ).throttledFn;
 }