feat: destructive reconciliation

src/channel.ts

@@ -17,7 +17,6 @@ import {
 } from './utils';
 import type { StreamChat } from './client';
 import { chatLoggerSystem } from './logger';
-import { DEFAULT_QUERY_CHANNEL_MESSAGE_LIST_PAGE_SIZE } from './constants';
 import type {
   AIState,
   APIResponse,
@@ -175,6 +174,8 @@ export class Channel extends ChannelApi {
   lastTypingEvent: Date | null;
   isTyping: boolean;
   disconnected: boolean;
+  /** Re-entrancy guard for {@link Channel.reload} (mirrors Thread.reload's isLoading guard). */
+  private _reloading = false;
   push_preferences?: Gen_ChannelPushPreferencesResponse;
   public readonly configState = new StateStore<ChannelInstanceConfig>({});
   public readonly messageComposer: MessageComposer;
@@ -1297,6 +1298,48 @@ export class Channel extends ChannelApi {
     return state;
   }
 
+  /**
+   * Re-watch the channel and refresh its FULL loaded message window — the channel analog of
+   * {@link Thread.reload}. Used on reconnect to catch up AND reconcile hard deletes that happened
+   * while offline: a hard delete reaches other clients via no event, so an offline client only learns
+   * of it by diffing the re-queried page.
+   *
+   * This is intentionally thin: it only re-issues `watch()` with a limit sized to the loaded window
+   * (`items.length`, so the whole loaded window is refreshed — not the smaller channel-list page). The
+   * actual fold + destructive reconciliation happens inside `query()` → `seedFirstPageSync`
+   * (the same path the channel-list re-hydrate and React's `recoverState` use), driven by the loaded-id
+   * snapshot `query()` captures before its await. Owning that single path is what lets the SDK stop
+   * passing the reconciliation window/snapshot itself.
+   *
+   * Preserves failed (unsent) messages: an overlap merge keeps them (the reconcile's provenance guard
+   * never prunes a non-server message); only a disjoint rebuild can drop them, so any that actually
+   * fell out are re-ingested below.
+   */
+  async reload() {
+    if (this._reloading || (!this.initialized && !this.offlineMode)) return;
+    this._reloading = true;
+    try {
+      const paginator = this.messagePaginator;
+      const headItems = paginator.headItems;
+      const requestedLimit = headItems.length || paginator.pageSize;
+      const failedBefore = headItems.filter((message) => message.status === 'failed');
+
+      await this.watch({ messages: { limit: requestedLimit } });
+      this.offlineMode = false;
+
+      paginator.batch(
+        () => {
+          for (const failed of failedBefore) {
+            if (!paginator.getItem(failed.id)) paginator.ingestItem(failed);
+          }
+        },
+        { coalesce: true },
+      );
+    } finally {
+      this._reloading = false;
+    }
+  }
+
   /**
    * Stops watching the channel.
    *
@@ -1482,13 +1525,44 @@ export class Channel extends ChannelApi {
     options: ChannelGetOrCreateRequest = {},
     messageSetToAddToIfDoesNotExist: MessageSetType = 'current',
   ) {
+    // Snapshot the loaded message ids BEFORE the network await, for a latest-window (re)seed only.
+    // When this query re-seeds an already-loaded window (reconnect / re-hydrate), seedFirstPageSync
+    // reconciles the fresh page against what is loaded, and the snapshot lets it tell an offline
+    // hard-delete (in the snapshot, absent from the page) from a message that arrives live
+    // during the fetch (not in the snapshot). Captured here — the only place with the pre-await state —
+    // so callers (channel.reload, watch) need not thread it. Empty on a cold open, so it is harmless.
+    // TODO(perf/cleanup): `headItems` materializes full message objects (intervalToItems) just to map
+    // them down to ids. A cheaper, clearer equivalent is a straight copy of the paginator index's own
+    // id set — expose `memberIds` on StoreBackedItemIndex (e.g. `snapshotMembers()` returning
+    // `new Set(this.memberIds)`) and use it here AND in client.queryChannelsAndHydrate. The broader
+    // scope (all intervals vs just the head) is inert: the reconcile only consults head ids, older
+    // island ids are never at/above-newest, and local messages are guarded by isServerConfirmedMessage.
+    const candidateIds =
+      messageSetToAddToIfDoesNotExist === 'latest'
+        ? new Set(this.messagePaginator.headItems.map((message) => message.id))
+        : undefined;
+
+    // The INITIAL channel-open query honors the paginator's OWN pageSize (light on native, 25) rather
+    // than the server's larger default — opening loads the same page size it paginates by. A caller
+    // that already knows how much to fetch passes an explicit messages.limit, which is respected as-is:
+    // a reconnect/re-hydrate sizes it to the loaded window (channel.reload → items.length), and
+    // pagination/around pass their own cursors + limit.
+    const requestedPageSize = options?.messages?.limit ?? this.messagePaginator.pageSize;
+
     // Make sure we wait for the connect promise if there is a pending one
     await this.getClient().wsPromise;
 
     const queryPayload: ChannelGetOrCreateRequest = {
       data: this._data,
       state: true,
       ...options,
+      // Ask the server for exactly the initial-open page size (not its default), so the loaded window
+      // matches the paginator's pageSize. Explicit messages (reconnect/around/pagination) pass through.
+      messages:
+        options?.messages ??
+        (messageSetToAddToIfDoesNotExist === 'latest'
+          ? { limit: requestedPageSize }
+          : undefined),
     };
 
     const state = this.id
@@ -1542,15 +1616,15 @@ export class Channel extends ChannelApi {
     // latest-page open paths (watch/create) pass 'latest' — the paginator's own pagination queries
     // use 'current' and must not be reseeded as a first page here.
     if (messageSetToAddToIfDoesNotExist === 'latest' && Array.isArray(state.messages)) {
-      const requestedPageSize =
-        options?.messages?.limit ?? DEFAULT_QUERY_CHANNEL_MESSAGE_LIST_PAGE_SIZE;
       // Pass the query's message pagination options through: a channel can be opened AROUND a
       // message (id_around / created_at_around), in which case the fetched page is a jump window,
       // not the latest page — the paginator must reconcile it with jump semantics.
       this.messagePaginator.seedFirstPageSync(
         state.messages.map(formatMessage),
         requestedPageSize,
         options?.messages,
+        // Re-seed of an already-loaded window folds + reconciles instead of blanking (see above).
+        { candidateIds, reconcile: true },
       );
     }
 

src/client.ts

@@ -1332,6 +1332,14 @@ export class StreamChat extends ChatApi {
     options?: QueryChannelsRequest,
     stateOptions: ChannelStateOptions = {},
   ): Promise<Channel[] | QueryChannelsResponseWithChannels> {
+    // TODO(perf/cleanup): prefer a `memberIds` snapshot over `headItems.map` here too — see the
+    // matching TODO in channel.query() for the full rationale.
+    const candidateIdsByCid = new Map<string, ReadonlySet<string>>();
+    for (const cid of Object.keys(this.activeChannels)) {
+      const head = this.activeChannels[cid]?.messagePaginator?.headItems;
+      if (head?.length)
+        candidateIdsByCid.set(cid, new Set(head.map((message) => message.id)));
+    }
     const queryChannelsResponse = await this.queryChannels(options);
     const channels = queryChannelsResponse.channels;
 
@@ -1349,7 +1357,12 @@ export class StreamChat extends ChatApi {
       });
     }
 
-    const hydratedChannels = this.hydrateActiveChannels(channels, stateOptions, options);
+    const hydratedChannels = this.hydrateActiveChannels(
+      channels,
+      stateOptions,
+      options,
+      candidateIdsByCid,
+    );
 
     if (stateOptions.withResponse) {
       return {
@@ -1405,6 +1418,7 @@ export class StreamChat extends ChatApi {
     channelsFromApi: ChannelStateResponseFields[] = [],
     stateOptions: ChannelStateOptions = {},
     queryChannelsOptions?: ChannelOptions,
+    candidateIdsByCid?: Map<string, ReadonlySet<string>>,
   ) {
     const { skipInitialization, offlineMode = false } = stateOptions;
     const channels: Channel[] = [];
@@ -1445,6 +1459,8 @@ export class StreamChat extends ChatApi {
         c.messagePaginator.seedFirstPageSync(
           channelState.messages.map(formatMessage),
           requestedPageSize,
+          undefined,
+          { reconcile: true, candidateIds: candidateIdsByCid?.get(c.cid) },
         );
       }
 

src/offline-support/offline_support_api.ts

@@ -746,6 +746,34 @@ export abstract class AbstractOfflineDB implements OfflineDBApi {
     return queries;
   };
 
+  /**
+   * Hard-delete a set of messages by id in a single transaction. A convenience over N individual
+   * {@link hardDeleteMessage} calls: collects each delete's queries (`execute: false`) and runs them
+   * as one batch. Used e.g. by destructive reconciliation on reconnect to mirror a set of in-memory
+   * removals into the DB. No-op for an empty set.
+   *
+   * @param payload.ids - The ids of the messages to hard-delete.
+   * @param payload.execute - Whether to immediately execute the operation (optional, defaults to `true`).
+   */
+  public hardDeleteMessages = async ({
+    ids,
+    execute = true,
+  }: {
+    ids: string[];
+    execute?: boolean;
+  }) => {
+    if (!ids.length) return [];
+    const queries = (
+      await Promise.all(ids.map((id) => this.hardDeleteMessage({ id, execute: false })))
+    ).flat();
+
+    if (execute) {
+      await this.executeSqlBatch(queries);
+    }
+
+    return queries;
+  };
+
   /**
    * A utility method to handle read events. It will calculate the state of the reads if
    * present in the event, or optionally rely on the hard override in unreadMessages.

src/pagination/paginators/BasePaginator.ts

@@ -443,6 +443,15 @@ export abstract class BasePaginator<T, Q> {
   /** Changed ids buffered since the last {@link flushIntervalViewPublish} (throttled paginators only). */
   private _pendingViewChangedIds = new Set<string>();
 
+  /**
+   * Depth of active {@link batch} `coalesce` scopes. While > 0, `ingestItem` / `removeItem` record
+   * that the active window changed (see {@link _suspendedWindowDirty}) instead of publishing it, so
+   * the whole batch produces a single `state.items` emit — independent of state throttling.
+   */
+  private _windowPublishSuspendDepth = 0;
+  /** Set by a suspended op that changed the active window, so {@link batch} publishes once on exit. */
+  private _suspendedWindowDirty = false;
+
   /**
    * Intervals keep items in disconnected ranges.
    * That is a scenario of jumping to non-sequential pages.
@@ -896,6 +905,11 @@ export abstract class BasePaginator<T, Q> {
     );
   }
 
+  /** True while a coalescing {@link batch} scope is suspending this paginator's own window publishes. */
+  protected get isWindowPublishSuspended(): boolean {
+    return this._windowPublishSuspendDepth > 0;
+  }
+
   /** Re-project the active window from its (live, source-of-truth) interval. `undefined` when inactive. */
   private projectActiveWindow(): T[] | undefined {
     if (!this._activeIntervalId) return undefined;
@@ -2108,7 +2122,9 @@ export abstract class BasePaginator<T, Q> {
     // 3. If it no longer matches the filter, we’re done (it has been removed above).
     if (!this.matchesFilter(ingestedItem)) {
       // Throttled: the removal above deferred its emit — publish the (settled) window once.
-      if (this.isStateThrottled && itemHasBeenRemoved) this.scheduleWindowPublish();
+      // Suspended (coalescing batch): removeItemAtCoordinates recorded it; batch() emits once on exit.
+      if (!this.isWindowPublishSuspended && this.isStateThrottled && itemHasBeenRemoved)
+        this.scheduleWindowPublish();
       return itemHasBeenRemoved;
     }
 
@@ -2190,7 +2206,12 @@ export abstract class BasePaginator<T, Q> {
           // Falls somewhere *inside* the global bounds, but we don't have that page loaded.
           // We’ve already removed any old occurrence, so from the paginator's perspective
           // this item won't be visible again until the relevant page is fetched.
-          if (this.isStateThrottled && itemHasBeenRemoved) this.scheduleWindowPublish();
+          if (
+            !this.isWindowPublishSuspended &&
+            this.isStateThrottled &&
+            itemHasBeenRemoved
+          )
+            this.scheduleWindowPublish();
           return itemHasBeenRemoved;
         }
       }
@@ -2211,7 +2232,10 @@ export abstract class BasePaginator<T, Q> {
       activeIntervalIdBeforeRemoval === removedIntervalId &&
       targetInterval.id === removedIntervalId
     ) {
-      this.setActiveInterval(targetInterval);
+      this.setActiveInterval(
+        targetInterval,
+        this.isWindowPublishSuspended ? { updateState: false } : undefined,
+      );
     }
 
     const addedNewInterval = !this._itemIntervals.has(targetInterval.id);
@@ -2228,7 +2252,10 @@ export abstract class BasePaginator<T, Q> {
         this._activeIntervalId,
       )
     ) {
-      if (this.isStateThrottled) {
+      if (this.isWindowPublishSuspended) {
+        // Coalescing batch: record the change; batch() emits the settled window once on exit.
+        this._suspendedWindowDirty = true;
+      } else if (this.isStateThrottled) {
         this.scheduleWindowPublish();
       } else {
         const items = this.items ?? [];
@@ -2264,6 +2291,41 @@ export abstract class BasePaginator<T, Q> {
     return true;
   }
 
+  /**
+   * Run `fn` as one batched mutation over this paginator's items, collapsing the redundant emits a
+   * naive per-item loop would produce:
+   *
+   * - **Shared-store fan-out** (always, via the `_itemIndex.batch` wrapper): per-item store
+   *   notifications fold into a single flush, so sibling holders of the same ids (e.g. a thread /
+   *   pinned paginator sharing the entity) re-project once, not once per item. Inert for single-home
+   *   paginators (channels, reminders, user groups) whose index is over a private store with no
+   *   sibling subscribers — there it is a plain passthrough.
+   * - **This paginator's own active window** (`coalesce: true`): every `ingestItem` / `removeItem`
+   *   inside `fn` records that the window changed instead of publishing it, then `batch` emits the
+   *   settled window exactly once on exit via {@link flushWindowPublish}. This is deterministic and
+   *   independent of state throttling — unlike leaving it off, where an un-throttled paginator emits
+   *   once per item and a throttled one merely coalesces within its 500ms window. Use it for one-shot
+   *   operations (reconciliation, reload re-ingest) that must settle in a single update. Omit it in WS
+   *   event handlers, where per-event publishes should ride the throttle so successive events coalesce
+   *   across its trailing edge rather than each forcing a synchronous emit.
+   */
+  batch(fn: () => void, { coalesce = false }: { coalesce?: boolean } = {}): void {
+    if (!coalesce) {
+      this._itemIndex.batch(fn);
+      return;
+    }
+    this._windowPublishSuspendDepth += 1;
+    try {
+      this._itemIndex.batch(fn);
+    } finally {
+      this._windowPublishSuspendDepth -= 1;
+    }
+    if (this._windowPublishSuspendDepth === 0 && this._suspendedWindowDirty) {
+      this._suspendedWindowDirty = false;
+      this.flushWindowPublish();
+    }
+  }
+
   // ---------------------------------------------------------------------------
   // Remove / contains
   // ---------------------------------------------------------------------------
@@ -2295,7 +2357,9 @@ export abstract class BasePaginator<T, Q> {
 
     // 2) Remove from visible state.items, if present
     if (stateLocation && stateLocation.currentIndex > -1) {
-      if (!this.isStateThrottled) {
+      if (this.isWindowPublishSuspended) {
+        this._suspendedWindowDirty = true;
+      } else if (!this.isStateThrottled) {
         const newItems = [...(this.items ?? [])];
         newItems.splice(stateLocation.currentIndex, 1);
         this.state.partialNext({ items: newItems });
@@ -2333,7 +2397,8 @@ export abstract class BasePaginator<T, Q> {
       const result = this.removeItemAtCoordinates(coords);
       this._itemIndex.remove(this.getItemId(item));
       // Throttled: removeItemAtCoordinates deferred its emit — publish the (settled) window once.
-      if (this.isStateThrottled) this.scheduleWindowPublish();
+      if (!this.isWindowPublishSuspended && this.isStateThrottled)
+        this.scheduleWindowPublish();
       return result;
     }