diff --git a/src/channel.ts b/src/channel.ts index dcf1c6c3d..abfc7e3f2 100644 --- a/src/channel.ts +++ b/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; + }) { + 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({ 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({ 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(); } } diff --git a/src/client.ts b/src/client.ts index 82683c92c..eb68a1cf5 100644 --- a/src/client.ts +++ b/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 }); diff --git a/src/errors.ts b/src/errors.ts index 432a1c2e7..10e9cf3b1 100644 --- a/src/errors.ts +++ b/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 } diff --git a/src/index.ts b/src/index.ts index 28953be44..0cdff0925 100644 --- a/src/index.ts +++ b/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'; diff --git a/src/messageDelivery/MessageDeliveryReporter.ts b/src/messageDelivery/MessageDeliveryReporter.ts index 757d4ee71..ce8a759f0 100644 --- a/src/messageDelivery/MessageDeliveryReporter.ts +++ b/src/messageDelivery/MessageDeliveryReporter.ts @@ -359,5 +359,5 @@ export class MessageDeliveryReporter { leading: true, trailing: true, }, - ); + ).throttledFn; } diff --git a/src/messageStore/MessageStore.ts b/src/messageStore/MessageStore.ts new file mode 100644 index 000000000..fc20304c7 --- /dev/null +++ b/src/messageStore/MessageStore.ts @@ -0,0 +1,193 @@ +import type { LocalMessage } from '../types'; +import type { Unsubscribe } from '../store'; + +/** + * A batch of message-store changes delivered to a subscriber in a single notification. + * + * `changedIds` are the ids the subscriber watches whose canonical object changed reference this + * flush (an upsert, or a removal — a removal makes `store.get(id)` return `undefined`). + */ +export type MessageStoreChangeBatch = { + changedIds: ReadonlySet; +}; + +/** + * Anything that observes messages held in the {@link MessageStore}. + * + * A subscriber watches a *set* of message ids (a paginator watches all ids in its + * intervals; a thread watches its single parent id). It is notified at most once + * per store transaction with the subset of its watched ids that changed. + */ +export type MessageStoreSubscriber = { + onMessagesChanged: (batch: MessageStoreChangeBatch) => void; + /** + * Optional: emit any throttled/pending state notification immediately. Called by + * {@link MessageStore.flushSubscribers} after an optimistic (local-user) write so the change renders + * without throttle delay. A paginator implements this by flushing its throttled window publish. + */ + flushState?: () => void; +}; + +/** + * A client-global, normalized store for message content. + * + * The store holds exactly **one** canonical {@link LocalMessage} per id and lets any + * number of entities (paginators, threads, ad-hoc consumers) subscribe to individual + * ids. Every message mutation in the SDK becomes a single `upsert`, and the store + * fans the change out to exactly the entities holding that id — replacing the manual + * copy-to-copy fan-out that keeping N per-paginator copies in sync required. + * + * ## Design + * + * - **Content:** `byId` (`Map`) is the single source of truth. + * Objects are immutable snapshots: `upsert` *replaces*, never mutates in place, so + * the reference-equality short-circuit (mirroring {@link StateStore.next}) and every + * downstream selector keep working. + * - **Subscription registry / refcount:** `subscribers` (`Map>`) + * is both the per-id notification list and the reference count — when the last + * subscriber of an id unlinks, the canonical copy is garbage-collected. + * - **Batching:** `transaction` coalesces a bulk write (e.g. a page of N messages) so + * each affected subscriber is notified **once** with the full changed-id set, instead + * of N times. + * - **Signal-then-pull:** subscribers receive the set of changed ids and pull the new + * content via `get`; the store never pushes objects. + * + * This is deliberately a hand-rolled per-id registry rather than a {@link StateStore}: + * a single `StateStore` would run every subscriber's selector on every change + * (O(all subscribers) per emit), and a store-per-message would allocate a full + * `StateStore` per id with no cross-id batching. The `Map` gives + * O(subscribers-of-changed-id) fan-out — the same idiom as `PollManager`'s cache. + */ +export class MessageStore { + private byId = new Map(); + private subscribers = new Map>(); + + private transactionDepth = 0; + private pendingChanged = new Map>(); + + // ---- reads ---- + + get(id: string | undefined): LocalMessage | undefined { + return typeof id === 'string' ? this.byId.get(id) : undefined; + } + + has(id: string): boolean { + return this.byId.has(id); + } + + // ---- writes ---- + + /** + * Replaces the canonical copy of `message` and notifies every subscriber watching + * its id — except `origin`, which is expected to re-render itself (the paginator + * that performed the write already emits its own window inline, so it must not be + * notified a second time through the subscription). + */ + upsert(message: LocalMessage, origin?: MessageStoreSubscriber): void { + const { id } = message; + const previous = this.byId.get(id); + // do not notify if the value hasn't changed (mirrors StateStore.next) + if (previous === message) return; + this.byId.set(id, message); + this.markDirty(id, origin); + this.autoFlush(); + } + + // ---- subscription registry / refcount ---- + + /** Registers `subscriber` as a holder of `id` (notification target + refcount). */ + link(id: string, subscriber: MessageStoreSubscriber): void { + let holders = this.subscribers.get(id); + if (!holders) { + holders = new Set(); + this.subscribers.set(id, holders); + } + holders.add(subscriber); + } + + /** Drops `subscriber` as a holder of `id`; GCs the canonical copy when none remain. */ + unlink(id: string, subscriber: MessageStoreSubscriber): void { + const holders = this.subscribers.get(id); + if (!holders) return; + holders.delete(subscriber); + if (holders.size === 0) { + this.subscribers.delete(id); + // refcount GC: nobody holds this message any longer. + this.byId.delete(id); + } + } + + /** + * Sugar for atomic / ad-hoc consumers that watch a single id (e.g. a thread watching + * its parent message). Fires `handler` immediately with the current value (mirroring + * {@link StateStore.subscribe}) and on every subsequent change. + */ + subscribe( + id: string, + handler: (message: LocalMessage | undefined) => void, + ): Unsubscribe { + const subscriber: MessageStoreSubscriber = { + onMessagesChanged: () => handler(this.byId.get(id)), + }; + this.link(id, subscriber); + handler(this.byId.get(id)); + return () => this.unlink(id, subscriber); + } + + // ---- batching ---- + + /** + * Runs `fn`, coalescing all notifications produced by writes inside it into a single + * flush on exit. Re-entrant: nested transactions flush only when the outermost exits. + */ + transaction(fn: () => T): T { + this.transactionDepth += 1; + try { + return fn(); + } finally { + this.transactionDepth -= 1; + if (this.transactionDepth === 0) this.flush(); + } + } + + /** + * Immediately flushes any throttled/pending state publish on the holders of `id` (via + * {@link MessageStoreSubscriber.flushState}). Called after an optimistic (local-user) write to `id` + * so it renders without the throttle delay. Only that id's own holders are flushed — the write + * touched no other id — and flushing a holder with nothing pending is a no-op. + */ + flushSubscribers(id: string): void { + const holders = this.subscribers.get(id); + if (!holders) return; + for (const holder of holders) holder.flushState?.(); + } + + private markDirty(id: string, origin?: MessageStoreSubscriber): void { + const holders = this.subscribers.get(id); + if (!holders) return; + for (const holder of holders) { + if (holder === origin) continue; + let changed = this.pendingChanged.get(holder); + if (!changed) { + changed = new Set(); + this.pendingChanged.set(holder, changed); + } + changed.add(id); + } + } + + private autoFlush(): void { + if (this.transactionDepth === 0) this.flush(); + } + + private flush(): void { + if (this.pendingChanged.size === 0) return; + // swap out the pending map before notifying so writes made from within a + // subscriber accumulate into the next flush rather than mutating this one. + const changedBySubscriber = this.pendingChanged; + this.pendingChanged = new Map(); + for (const [subscriber, changedIds] of changedBySubscriber) { + subscriber.onMessagesChanged({ changedIds }); + } + } +} diff --git a/src/messageStore/MessageStoreBackedItemIndex.ts b/src/messageStore/MessageStoreBackedItemIndex.ts new file mode 100644 index 000000000..255e8987f --- /dev/null +++ b/src/messageStore/MessageStoreBackedItemIndex.ts @@ -0,0 +1,96 @@ +import type { ItemIndexApi } from '../pagination/ItemIndex'; +import type { MessageStore, MessageStoreSubscriber } from './MessageStore'; +import type { LocalMessage } from '../types'; + +export type MessageStoreBackedItemIndexOptions = { + store: MessageStore; + /** The paginator that owns this index; used as the store subscriber + refcount holder. */ + owner: MessageStoreSubscriber; + getId: (item: LocalMessage) => string; +}; + +/** + * An {@link ItemIndexApi} implementation that keeps message **content** in a shared, + * client-global {@link MessageStore} while keeping **membership** local. + * + * A paginator sees the exact same CRUD surface as a plain {@link ItemIndex}, but: + * + * - `get`/`has`/`values`/`entries` are scoped to *this* paginator's membership + * (`memberIds`), so `getItem(id)` still means "does THIS paginator hold the id" + * — even though the canonical object lives in the shared store. This is what keeps + * e.g. reaction routing (`threadPaginator.getItem(id) ? thread : channel`) correct + * and keeps the `.values()` scans from ever walking other channels' messages. + * - `setOne` writes content once into the shared store and links the owner as a holder + * (drives both notification fan-out and refcount GC). The write passes the owner as + * `origin`, so the owner is not notified of its own write (it re-emits its window + * inline); other holders of the same id ARE notified and re-project. + * - `remove`/`clear` unlink the owner rather than hard-deleting content, so a message + * still held by another paginator (e.g. a `show_in_channel` reply in both the channel + * list and its thread) survives; the store GCs it only when the last holder unlinks. + */ +export class MessageStoreBackedItemIndex implements ItemIndexApi { + private memberIds = new Set(); + private readonly store: MessageStore; + private readonly owner: MessageStoreSubscriber; + private readonly getId: (item: LocalMessage) => string; + + constructor({ store, owner, getId }: MessageStoreBackedItemIndexOptions) { + this.store = store; + this.owner = owner; + this.getId = getId; + } + + setMany(items: LocalMessage[]) { + this.store.transaction(() => { + for (const item of items) this.setOne(item); + }); + } + + setOne(item: LocalMessage) { + const id = this.getId(item); + this.store.link(id, this.owner); + this.memberIds.add(id); + this.store.upsert(item, this.owner); + } + + get(id: string): LocalMessage | undefined { + return this.memberIds.has(id) ? this.store.get(id) : undefined; + } + + has(id: string): boolean { + return this.memberIds.has(id); + } + + remove(id: string) { + if (!this.memberIds.has(id)) return; + this.memberIds.delete(id); + this.store.unlink(id, this.owner); + } + + clear() { + for (const id of this.memberIds) this.store.unlink(id, this.owner); + this.memberIds.clear(); + } + + entries(): [string, LocalMessage][] { + const result: [string, LocalMessage][] = []; + for (const id of this.memberIds) { + const item = this.store.get(id); + if (item) result.push([id, item]); + } + return result; + } + + values(): LocalMessage[] { + const result: LocalMessage[] = []; + for (const id of this.memberIds) { + const item = this.store.get(id); + if (item) result.push(item); + } + return result; + } + + batch(fn: () => R): R { + return this.store.transaction(fn); + } +} diff --git a/src/messageStore/applyReactionLocally.ts b/src/messageStore/applyReactionLocally.ts new file mode 100644 index 000000000..fb3db23d0 --- /dev/null +++ b/src/messageStore/applyReactionLocally.ts @@ -0,0 +1,112 @@ +import type { StreamChat } from '../client'; +import type { ReactionRequest, ReactionResponse, UserResponse } from '../types'; +import { + computeOwnReactions, + messageWithReactionAdded, + messageWithReactionRemoved, +} from '../utils'; + +/** + * Applies a reaction to the single canonical copy of a message in the client-global + * {@link MessageStore}, addressed purely by id — NOT through any paginator's membership. It reads the + * current message POJO (`store.get`), produces a new POJO with the reaction folded into its + * `reaction_groups` / `latest_reactions` (shared count helpers) and `own_reactions` + * ({@link computeOwnReactions}), and writes it back (`store.upsert`). The store then notifies every + * collection currently holding that id — the main-list / thread-reply paginators, a thread's + * subscribed parent, any future consumer — and each re-projects. The message stays a plain object; + * the store's per-id registry is the reactivity. + * + * Because it is keyed by id rather than by a paginator's `getItem`, it reaches a message held by ANY + * collection, including the thread parent that lives in no paginator — with no per-home wiring. + * + * Writes memory (synchronous) + the offline DB (fire-and-forget), and returns an `undo()` reversing + * both on the current state (inverse deltas, concurrency-safe), or `undefined` when the user isn't + * connected or the message isn't in the store. + */ +export const applyReactionLocally = ( + client: StreamChat, + { + enforceUnique = false, + messageId, + reaction, + removed = false, + }: { + messageId: string; + reaction: ReactionRequest; + enforceUnique?: boolean; + removed?: boolean; + }, +): (() => void) | undefined => { + const store = client.messageStore; + const user = client.user; + const existing = store.get(messageId); + if (!user || !existing) return; + + const now = new Date(); + // Spread `reaction` first so the authoritative fields below win, while still preserving any values + // the caller already carried (e.g. the original `created_at` when undo re-applies a captured + // reaction) via `?? now`. `message_id`/`user`/`user_id` are always derived from this message and + // the connected user, so they never need to come off `reaction`. + const reactionResponse: ReactionResponse = { + ...reaction, + created_at: reaction.created_at ?? now, + custom: reaction.custom ?? {}, + message_id: messageId, + score: reaction.score ?? 1, + type: reaction.type, + updated_at: reaction.updated_at ?? now, + user: user as UserResponse, + user_id: user.id, + }; + + // Capture what this op removes so undo() can restore it faithfully (reaction spread last): the + // deleted reaction for a removal, or the user's displaced reactions for an enforce_unique add. + const removedReactions: ReactionResponse[] = removed + ? (existing.own_reactions?.filter((r) => r.type === reactionResponse.type) ?? []) + : enforceUnique + ? (existing.own_reactions ?? []) + : []; + + const withCounts = removed + ? messageWithReactionRemoved(existing, reactionResponse) + : messageWithReactionAdded(existing, reactionResponse, enforceUnique); + const own_reactions = computeOwnReactions({ + current: existing.own_reactions ?? [], + enforceUnique, + reaction: reactionResponse, + removed, + userId: user.id, + }); + store.upsert({ ...withCounts, own_reactions }); + store.flushSubscribers(messageId); + + const persisted = store.get(messageId); + if (persisted) { + client.offlineDb?.executeQuerySafely( + (db) => + removed + ? db.deleteReaction({ message: persisted, reaction: reactionResponse }) + : enforceUnique + ? db.updateReaction({ message: persisted, reaction: reactionResponse }) + : db.insertReaction({ message: persisted, reaction: reactionResponse }), + { method: 'applyReactionLocally' }, + ); + } + + return () => { + if (!removed) { + applyReactionLocally(client, { + messageId, + reaction: reactionResponse, + removed: true, + }); + } + for (const removedReaction of removedReactions) { + applyReactionLocally(client, { + messageId, + reaction: removedReaction, + removed: false, + }); + } + }; +}; diff --git a/src/messageStore/index.ts b/src/messageStore/index.ts new file mode 100644 index 000000000..b392bbc58 --- /dev/null +++ b/src/messageStore/index.ts @@ -0,0 +1,2 @@ +export * from './applyReactionLocally'; +export * from './MessageStore'; diff --git a/src/offline-support/offline_support_api.ts b/src/offline-support/offline_support_api.ts index 05178047e..ef478595b 100644 --- a/src/offline-support/offline_support_api.ts +++ b/src/offline-support/offline_support_api.ts @@ -18,6 +18,7 @@ import type { PrepareBatchDBQueries, } from './types'; import { OfflineError } from './types'; +import { isEphemeral } from '../errors'; import type { StreamChat } from '../client'; import type { AxiosError } from 'axios'; import { OfflineDBSyncManager } from './offline_sync_manager'; @@ -1185,15 +1186,15 @@ export abstract class AbstractOfflineDB implements OfflineDBApi { }; /** - * A utility method that determines if a failed task should be added to the queue based on its - * error. Error code 4 — bad request data. Error code 17 — missing `own_capabilities` to execute - * the task. + * A utility method that determines if a failed task should be skipped (NOT added to the queue) - + * i.e. the failure is a definitive rejection rather than an ephemeral/retryable one. A task is + * kept in the queue only when its error is {@link isEphemeral} (connection/network error or a + * retryable server code). A non retryable server response (i.e bad request, not allowed etc) is + * skipped since retrying it would never succeed. * - * @param error - The failed task's Axios error. - * @returns `true` when the task should not be re-queued. + * @param error - The error thrown while executing the failed task. */ - private shouldSkipQueueingTask = (error: AxiosError) => - error?.response?.data?.code === 4 || error?.response?.data?.code === 17; + private shouldSkipQueueingTask = (error: AxiosError) => !isEphemeral(error); private mergeFailedMessageUpdateIntoPendingSendMessage = ({ editedMessage, diff --git a/src/pagination/ItemIndex.ts b/src/pagination/ItemIndex.ts index fe9d2df26..e712394d4 100644 --- a/src/pagination/ItemIndex.ts +++ b/src/pagination/ItemIndex.ts @@ -2,6 +2,29 @@ export type ItemIndexOptions = { getId: (item: T) => string; }; +/** + * The minimal CRUD surface a paginator relies on from its item index. + * + * Extracted so a paginator can be backed either by the default per-instance + * {@link ItemIndex} or by an adapter over a shared, client-global store + * (see `MessageStoreBackedItemIndex`) without the call sites knowing the difference. + */ +export interface ItemIndexApi { + setMany(items: T[]): void; + setOne(item: T): void; + get(id: string): T | undefined; + has(id: string): boolean; + remove(id: string): void; + clear(): void; + entries(): [string, T][]; + values(): T[]; + /** + * Runs `fn`, coalescing any change notifications it produces into a single flush. + * A plain {@link ItemIndex} has nothing to coalesce and simply runs `fn`. + */ + batch(fn: () => R): R; +} + /** * The ItemIndex is a canonical, ID-addressable storage layer for domain items. * @@ -68,7 +91,7 @@ export type ItemIndexOptions = { * * @template T The domain item type managed by the index. */ -export class ItemIndex { +export class ItemIndex implements ItemIndexApi { private byId = new Map(); private readonly getId: (item: T) => string; @@ -109,4 +132,8 @@ export class ItemIndex { values() { return [...this.byId.values()]; } + + batch(fn: () => R): R { + return fn(); + } } diff --git a/src/pagination/paginators/BasePaginator.ts b/src/pagination/paginators/BasePaginator.ts index 69e9dff0a..f65a35f23 100644 --- a/src/pagination/paginators/BasePaginator.ts +++ b/src/pagination/paginators/BasePaginator.ts @@ -3,9 +3,11 @@ import { binarySearch } from '../sortCompiler'; import { itemMatchesFilter } from '../filterCompiler'; import { isPatch, StateStore, type ValueOrPatch } from '../../store'; import { debounce, type DebouncedFunc, generateUUIDv4, sleep } from '../../utils'; +import { throttle, type Throttled } from '../../utils/throttling/throttle'; +import { isStateThrottlingEnabled } from './stateThrottling'; import type { FieldToDataResolver } from '../types.normalization'; import { ComparisonResult } from '../types.normalization'; -import { ItemIndex } from '../ItemIndex'; +import { ItemIndex, type ItemIndexApi } from '../ItemIndex'; import { isEqual } from '../../utils/mergeWith/mergeWithCore'; import { DEFAULT_QUERY_CHANNELS_MS_BETWEEN_RETRIES } from '../../constants'; @@ -316,6 +318,17 @@ export interface PaginatorPlugin { export type PaginatorOptions = { /** The number of milliseconds to debounce the search query. The default interval is 300ms. */ debounceMs?: number; + /** + * When set (and not disabled for tests — see `stateThrottling.ts`), coalesces the paginator's own + * live `state.items` publishes to at most once per `stateThrottleMs` (leading + trailing edge): a + * burst of live mutations (WS `message.new`, reactions, reads) re-projects the active window and + * publishes it ~2×/sec instead of once per event. Only the paginator's OWN writes to `state.items` + * are batched — `state.getLatestValue()` is untouched (no `StateStore` change), pagination / jump / + * query publishes stay immediate, and an immediate flush past it is available via + * {@link flushPendingPublishes}. Unset ⇒ no throttle (default). Enabled at + * 500ms for the message list — see {@link MessagePaginator}. + */ + stateThrottleMs?: number; /** * Function containing custom logic that decides, whether the next pagination query to be executed should be considered the first page query. * It makes sense to consider the next query as the first page query if filters, sort, options etc. (query params) excluding the page size have changed. @@ -334,7 +347,14 @@ export type PaginatorOptions = { /** In case of offset pagination, specify the initial offset value. */ initialOffset?: number; /** If item index is provided, this index ensures updates in a single place and all consumers have access to a single source of data. */ - itemIndex?: ItemIndex; + itemIndex?: ItemIndexApi; + /** + * Factory for the item index, invoked with the fully-constructed paginator as `owner`. + * Lets a subclass back the paginator with an adapter that needs a reference to the + * owner (e.g. a shared, client-global store) without the `this`-before-`super` problem. + * Ignored when an explicit `itemIndex` is supplied. + */ + createItemIndex?: (owner: BasePaginator) => ItemIndexApi; /** * Comparator defining in-memory item ordering for interval math and visible list rendering. * Defaults to `sortComparator` to preserve existing paginator behavior. @@ -353,11 +373,13 @@ export type PaginatorOptions = { }; type OptionalPaginatorConfigFields = + | 'stateThrottleMs' | 'deriveCursor' | 'doRequest' | 'initialCursor' | 'initialOffset' | 'itemIndex' + | 'createItemIndex' | 'itemOrderComparator' | 'throwErrors'; @@ -392,6 +414,24 @@ export abstract class BasePaginator { intervalViews: StateStore>; config: BasePaginatorConfig; + /** + * Throttle for the active-window `state.items` publish (message list). Created only when + * `config.stateThrottleMs` is set; drives {@link scheduleWindowPublish} / {@link flushPendingPublishes}. See + * `stateThrottleMs` in {@link PaginatorOptions}. + */ + private _windowPublishThrottle?: Throttled<[]>; + + /** + * Throttle for the interval-view publishes (`anchoredHead` / `logicalHead` / `logicalTail`) driven + * by sibling store updates. Independent of {@link _windowPublishThrottle} so a view refresh lands on + * its own trailing edge even when `state.items` stays quiet. Created alongside it (only when + * `config.stateThrottleMs` is set); buffers changed ids in {@link _pendingViewChangedIds}. + */ + private _viewPublishThrottle?: Throttled<[]>; + + /** Changed ids buffered since the last {@link flushIntervalViewPublish} (throttled paginators only). */ + private _pendingViewChangedIds = new Set(); + /** * Intervals keep items in disconnected ranges. * That is a scenario of jumping to non-sequential pages. @@ -405,7 +445,7 @@ export abstract class BasePaginator { * It serves as a single source of truth for all those that need to access the items * outside the paginator. */ - protected _itemIndex: ItemIndex; + protected _itemIndex: ItemIndexApi; protected _executeQueryDebounced!: DebouncedExecQueryFunction; /** Last effective query shape produced by subclass for the most recent request. */ @@ -459,6 +499,7 @@ export abstract class BasePaginator { initialCursor, initialOffset, itemIndex, + createItemIndex, ...options }: PaginatorOptions = {}) { this.config = { @@ -473,6 +514,22 @@ export abstract class BasePaginator { cursor: initialCursor, offset: initialOffset ?? 0, }); + if (this.config.stateThrottleMs) { + // Coalesce the paginator's own live `state.items` publishes (see `stateThrottleMs` doc). The + // trailing edge re-projects the active window fresh, so a burst emits ~once per interval. + this._windowPublishThrottle = throttle( + () => this.flushWindowPublish(), + this.config.stateThrottleMs, + { leading: true, trailing: true }, + ); + // Interval view publishes ride their own throttle so they coalesce like `state.items` but land + // on an independent trailing edge (see {@link _viewPublishThrottle}). + this._viewPublishThrottle = throttle( + () => this.flushIntervalViewPublish(), + this.config.stateThrottleMs, + { leading: true, trailing: true }, + ); + } this.intervalViews = new StateStore>({ logicalHead: [], logicalTail: [], @@ -481,7 +538,10 @@ export abstract class BasePaginator { this.setDebounceOptions({ debounceMs }); this.sortComparator = noOrderChange; this._filterFieldToDataResolvers = []; - this._itemIndex = itemIndex ?? new ItemIndex({ getId: this.getItemId.bind(this) }); + this._itemIndex = + itemIndex ?? + createItemIndex?.(this) ?? + new ItemIndex({ getId: this.getItemId.bind(this) }); } // --------------------------------------------------------------------------- @@ -740,6 +800,7 @@ export abstract class BasePaginator { * publishing. No-ops when the views are already empty so a reset does not emit needlessly. */ protected clearIntervalViews() { + this._pendingViewChangedIds.clear(); const { logicalHead, logicalTail, anchoredHead } = this.intervalViews.getLatestValue(); // Clear whenever any view holds items; skip only when all are already empty, so a reset on an @@ -804,6 +865,198 @@ export abstract class BasePaginator { return typeof id === 'string' ? this._itemIndex?.get(id) : undefined; } + /** + * Whether this paginator's live `state.items` publishes are currently throttled — `stateThrottleMs` + * is set, throttling is not globally disabled (tests), and item order is not locked (a locked-order + * list must emit the caller-computed, order-preserved array, not a re-projection). When false, every + * live mutation publishes immediately, exactly as before this feature. + */ + protected get isStateThrottled(): boolean { + return ( + isStateThrottlingEnabled() && + !!this._windowPublishThrottle && + !this.config.lockItemOrder + ); + } + + /** Re-project the active window from its (live, source-of-truth) interval. `undefined` when inactive. */ + private projectActiveWindow(): T[] | undefined { + if (!this._activeIntervalId) return undefined; + const active = this._itemIntervals.get(this._activeIntervalId); + return active ? this.intervalToItems(active) : undefined; + } + + /** + * Publish the active window to `state`, re-projecting it fresh from the active interval at call time + * (the throttle boundary / flush). Because the intervals are mutated synchronously by every live op, + * this always reflects the latest settled state, so intermediate values within a window coalesce + * away. Clears the visible window when the active interval is gone (e.g. the last item was removed). + */ + private flushWindowPublish(): void { + const items = this.projectActiveWindow(); + if (items) { + this.state.partialNext({ items }); + return; + } + if ((this.state.getLatestValue().items?.length ?? 0) > 0) { + this.state.partialNext({ items: [] }); + } + } + + /** + * Schedule a throttled active-window publish (leading + trailing). Called from live mutations + * (ingest / content-change / remove) INSTEAD of an inline `state.partialNext({ items })` when + * {@link isStateThrottled}. Safe to call many times within one op — they coalesce to a single emit. + */ + protected scheduleWindowPublish(): void { + this._windowPublishThrottle?.throttledFn(); + } + + /** + * Flush any pending throttled window + interval-view publishes immediately. No-op when nothing + * is pending or throttling is off. + */ + protected flushPendingPublishes(): void { + this._windowPublishThrottle?.flush(); + this._viewPublishThrottle?.flush(); + } + + /** + * {@link _viewPublishThrottle} boundary: apply every interval-view change buffered since the last + * flush, then clear the buffer. Independent of the `state.items` window publish, so a view update + * lands within one throttle interval even when the active window stays quiet (e.g. a reaction on a + * head message while a non-head window is active). No-op when nothing was buffered. + */ + private flushIntervalViewPublish(): void { + if (!this._pendingViewChangedIds.size) return; + this.refreshIntervalViewsForChangedIds(this._pendingViewChangedIds); + this._pendingViewChangedIds.clear(); + } + + /** + * Refresh any tracked {@link intervalViews} field (logical head, logical tail, anchored head) that + * holds one of `changedIds`. A sibling holder writing new content through the shared item store + * swaps the item object those views reference, but only this paginator's own ingest/remove + * ({@link commitInterval}/{@link dropInterval}) republish them — so a reaction/edit made elsewhere + * would otherwise leave stale references in `anchoredHead`/`logicalHead`/`logicalTail` even though + * the active window (`state.items`, see {@link reconcileChangedIds}) was refreshed. + * + * Called immediately for un-throttled paginators, or from {@link flushIntervalViewPublish} at the + * view-publish throttle boundary when throttled — see {@link reconcileChangedIds}. Handled + * independently of `state.items`. When the anchored head is also the active interval its content is + * projected here as well as into `state.items`; deduping that double projection is a separate, + * deferred perf follow-up. + */ + private refreshIntervalViewsForChangedIds(changedIds: ReadonlySet): void { + const head = this.liveHeadLogical; + if (head && this.intervalHoldsAnyChangedId(head, changedIds)) { + this.intervalViews.partialNext({ logicalHead: this.intervalToItems(head) }); + } + const tail = this.liveTailLogical; + if (tail && this.intervalHoldsAnyChangedId(tail, changedIds)) { + this.intervalViews.partialNext({ logicalTail: this.intervalToItems(tail) }); + } + let anchored: Interval | undefined; + for (const itv of this._itemIntervals.values()) { + if (!isLogicalInterval(itv) && itv.isHead) { + anchored = itv; + break; + } + } + if (anchored && this.intervalHoldsAnyChangedId(anchored, changedIds)) { + this.publishAsAnchoredHead(anchored); + } + } + + private intervalHoldsAnyChangedId( + interval: AnyInterval, + changedIds: ReadonlySet, + ): boolean { + for (const id of interval.itemIds) if (changedIds.has(id)) return true; + return false; + } + + /** + * Reconcile the projected window + interval views against a set of changed ids: another holder + * swapped the shared item object those views reference. Refreshes any tracked interval view that + * holds a changed id and re-projects (or slot-swaps) the active window — coalesced through the + * publish throttles when throttling is on. + */ + protected reconcileChangedIds(changedIds: ReadonlySet): void { + // A sibling holder changed shared content: refresh any tracked interval view (logical head/tail, + // anchored head) holding a changed id — independent of the active window below, since a view can + // hold a changed id the active window does not. When throttled, buffer the ids and tick the + // view-publish throttle so refreshes coalesce (once per interval) yet still land on their own + // trailing edge even if `state.items` never publishes again; otherwise publish immediately. + if (this.isStateThrottled) { + for (const id of changedIds) this._pendingViewChangedIds.add(id); + this._viewPublishThrottle?.throttledFn(); + } else { + this.refreshIntervalViewsForChangedIds(changedIds); + } + + if (!this._activeIntervalId) return; + const activeInterval = this._itemIntervals.get(this._activeIntervalId); + if (!activeInterval) return; + + // Throttled (message list): the slot-swap fast path below reads the last-published `items`, which + // lags the live intervals while throttled — so skip it. Gate on membership only and schedule a + // single coalesced re-projection; the boundary re-derives the window fresh from the interval. + if (this.isStateThrottled) { + for (const id of activeInterval.itemIds) { + if (changedIds.has(id)) { + this.scheduleWindowPublish(); + return; + } + } + return; + } + + // Fast path: a content update (an in-place edit written through a sibling holder) changes + // items in place without changing membership or order. When the current window still lines + // up with the interval one-to-one, shallow-copy it and swap only the changed slots — this + // preserves every unchanged item reference (so memoized rows bail) and avoids re-mapping and + // re-sorting the whole active window on every event. Skipped when a boost is active (a boost + // can reorder the visible window, which a slot-swap would not reflect) — then we fall through + // to the full projection below, which applies the boost order. + const currentItems = this.items; + this.clearExpiredBoosts(); + if ( + currentItems && + currentItems.length === activeInterval.itemIds.length && + (this.config.lockItemOrder || this.boosts.size === 0) + ) { + let next: T[] | undefined; + let needsFullProjection = false; + for (let i = 0; i < currentItems.length; i++) { + const id = this.getItemId(currentItems[i]); + if (!changedIds.has(id)) continue; + const updated = this._itemIndex.get(id); + if (!updated) { + // The id left the store (a removal, not an in-place update) — resync via a full projection. + needsFullProjection = true; + break; + } + if (updated === currentItems[i]) continue; + if (!next) next = currentItems.slice(); + next[i] = updated; + } + if (!needsFullProjection) { + if (next) this.state.partialNext({ items: next }); + return; + } + } + + // Fallback: membership/order drifted (or no window to patch, or a boost is active). Re-project, + // but only if a changed id is actually in the active interval. + for (const id of activeInterval.itemIds) { + if (changedIds.has(id)) { + this.state.partialNext({ items: this.intervalToItems(activeInterval) }); + return; + } + } + } + // --------------------------------------------------------------------------- // Boosts // --------------------------------------------------------------------------- @@ -898,6 +1151,14 @@ export abstract class BasePaginator { return items; } + // itemIds are maintained in itemOrder, so the mapped items are already ordered; only an active + // boost can reorder the visible window. Skip the otherwise-redundant full re-sort when none are + // active (the common case), so a page ingest / full projection is a map, not a map + sort. + this.clearExpiredBoosts(); + if (this.boosts.size === 0) { + return items; + } + // Visible ordering uses boost-aware comparator return items.sort(this.effectiveComparator.bind(this)); } @@ -1553,9 +1814,13 @@ export abstract class BasePaginator { isTail, }); - for (const item of page) { - this._itemIndex.setOne(item); - } + // Coalesce per-item change notifications into a single flush, so a page of N + // items wakes each subscribing paginator once rather than N times. + this._itemIndex.batch(() => { + for (const item of page) { + this._itemIndex.setOne(item); + } + }); const targetInterval = targetIntervalId ? this._itemIntervals.get(targetIntervalId) @@ -1712,6 +1977,8 @@ export abstract class BasePaginator { // 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(); return itemHasBeenRemoved; } @@ -1777,6 +2044,7 @@ export abstract class BasePaginator { // 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(); return itemHasBeenRemoved; } } @@ -1814,32 +2082,36 @@ export abstract class BasePaginator { this._activeIntervalId, ) ) { - const items = this.items ?? []; - /** - * Having config.lockItemOrder enabled when working with intervals will lead to - * discrepancies once active intervals are switched: - * 1. state.items [a,b,c] intervals [a,b,c], [d] - * 2. a changed and is moved to another interval state.items is now [a,b,c], intervals [b,c,], [d, a] - * 3. jumping / changing active interval to [d,a] - state.items is now [d,a], intervals [b,c], [d,a] - */ - if (keepOrderInState) { - // Item was visible before → reinsert at its old index - const nextView = items.slice(); - const insertAt = Math.min(originalIndexInState, nextView.length); - nextView.splice(insertAt, 0, ingestedItem); - this.state.partialNext({ items: nextView }); + if (this.isStateThrottled) { + this.scheduleWindowPublish(); } else { + const items = this.items ?? []; /** - * Select a correct interval from which the state.items array is derived + * Having config.lockItemOrder enabled when working with intervals will lead to + * discrepancies once active intervals are switched: + * 1. state.items [a,b,c] intervals [a,b,c], [d] + * 2. a changed and is moved to another interval state.items is now [a,b,c], intervals [b,c,], [d, a] + * 3. jumping / changing active interval to [d,a] - state.items is now [d,a], intervals [b,c], [d,a] */ - this.state.partialNext({ - items: this.intervalToItems( - this._activeIntervalId === removedItemCoordinates?.interval?.interval.id && - this._activeIntervalId !== targetInterval.id - ? removedItemCoordinates.interval.interval - : targetInterval, - ), - }); + if (keepOrderInState) { + // Item was visible before → reinsert at its old index + const nextView = items.slice(); + const insertAt = Math.min(originalIndexInState, nextView.length); + nextView.splice(insertAt, 0, ingestedItem); + this.state.partialNext({ items: nextView }); + } else { + /** + * Select a correct interval from which the state.items array is derived + */ + this.state.partialNext({ + items: this.intervalToItems( + this._activeIntervalId === removedItemCoordinates?.interval?.interval.id && + this._activeIntervalId !== targetInterval.id + ? removedItemCoordinates.interval.interval + : targetInterval, + ), + }); + } } } @@ -1877,9 +2149,11 @@ export abstract class BasePaginator { // 2) Remove from visible state.items, if present if (stateLocation && stateLocation.currentIndex > -1) { - const newItems = [...(this.items ?? [])]; - newItems.splice(stateLocation.currentIndex, 1); - this.state.partialNext({ items: newItems }); + if (!this.isStateThrottled) { + const newItems = [...(this.items ?? [])]; + newItems.splice(stateLocation.currentIndex, 1); + this.state.partialNext({ items: newItems }); + } // keep insertionIndex consistent if someone uses it later if (stateLocation.insertionIndex > stateLocation.currentIndex) { @@ -1910,7 +2184,11 @@ export abstract class BasePaginator { if (item) { const coords = this.locateByItem(item); if (!coords.state && !coords.interval) return noAction; - return this.removeItemAtCoordinates(coords); + 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(); + return result; } return noAction; @@ -2339,6 +2617,22 @@ export abstract class BasePaginator { this.clearIntervalViews(); } + /** + * Releases this paginator's hold on its item content. With a shared, refcounted item index this + * unlinks every member id from the backing store, so the store no longer strong-references this + * paginator through its subscriber registry — otherwise a discarded owner stays pinned, keeps + * receiving change notifications, and its items never garbage-collect. Call on teardown of the + * owner: a discard, not a reset (the owner is not reused; a re-appearing id gets a fresh instance; + * leftover interval/state caches die with the paginator when it is dropped). Also cancels any + * pending throttled window/view publish, so nothing emits after teardown. + */ + dispose(): void { + this._windowPublishThrottle?.cancelTimer(); + this._viewPublishThrottle?.cancelTimer(); + this._pendingViewChangedIds.clear(); + this._itemIndex.clear(); + } + toTail = (params: Omit, 'direction' | 'queryShape'> = {}) => this.executeQuery({ direction: 'tailward', ...params }); diff --git a/src/pagination/paginators/MessageIntervalPaginator.ts b/src/pagination/paginators/MessageIntervalPaginator.ts index 536c2fee9..9b9bcb7f9 100644 --- a/src/pagination/paginators/MessageIntervalPaginator.ts +++ b/src/pagination/paginators/MessageIntervalPaginator.ts @@ -28,12 +28,18 @@ import type { } from '../../types'; import type { Channel } from '../../channel'; import { StateStore } from '../../store'; -import { formatMessage, generateUUIDv4, toDeletedMessage } from '../../utils'; +import { + computeOwnReactions, + formatMessage, + generateUUIDv4, + toDeletedMessage, +} from '../../utils'; import { makeComparator } from '../sortCompiler'; import type { FieldToDataResolver } from '../types.normalization'; import { resolveDotPathValue } from '../utility.normalization'; import { lowerBound } from '../utility.search'; -import { ItemIndex } from '../ItemIndex'; +import type { ItemIndexApi } from '../ItemIndex'; +import type { MessageStoreChangeBatch } from '../../messageStore/MessageStore'; import { deriveCreatedAtAroundPaginationFlags } from '../cursorDerivation'; import { deriveIdAroundPaginationFlags } from '../cursorDerivation/idAroundPaginationFlags'; import { deriveLinearPaginationFlags } from '../cursorDerivation/linearPaginationFlags'; @@ -117,7 +123,7 @@ export const getMessageCreatedAtTimestamp = (message: LocalMessage): number | nu export type MessagePaginatorOptions = { channel: Channel; id?: string; - itemIndex?: ItemIndex; + itemIndex?: ItemIndexApi; parentMessageId?: string; /** * Sort passed to backend message/replies query. @@ -162,6 +168,26 @@ export class MessageIntervalPaginator extends BasePaginator< return !message.shadowed; } + /** + * Message-store adapter (this paginator is a `MessageStoreSubscriber`): the `MessageStore` calls this + * on each holder with the subset of its watched ids that changed. Unwraps the batch and delegates to + * the base's store-agnostic {@link BasePaginator.reconcileChangedIds}. Lives here, not on the generic + * `BasePaginator`, so the base stays free of `MessageStore` types — only message paginators are + * store-backed. + */ + onMessagesChanged({ changedIds }: MessageStoreChangeBatch): void { + this.reconcileChangedIds(changedIds); + } + + /** + * Message-store subscriber flush (the optional `MessageStoreSubscriber.flushState`): the `MessageStore` + * calls this after an optimistic (local-user) write so the change renders without throttle delay. + * Delegates to the base's generic {@link BasePaginator.flushPendingPublishes}. + */ + flushState(): void { + this.flushPendingPublishes(); + } + protected get intervalItemIdsAreHeadFirst(): boolean { // Messages are stored in chronological order (created_at asc) within an interval. // Pagination "head" (newest side) is therefore at the END of the `itemIds` array. @@ -177,7 +203,7 @@ export class MessageIntervalPaginator extends BasePaginator< constructor({ channel, id, - itemIndex = new ItemIndex({ getId: (item) => item.id }), + itemIndex, parentMessageId, requestSort, sort, @@ -798,36 +824,40 @@ export class MessageIntervalPaginator extends BasePaginator< }) => { const loadedMessages = this.items ?? []; - for (const message of loadedMessages) { - if (message.user?.id === userId) { - if (hardDelete) { - this.removeItem({ id: message.id }); - } else { - this.ingestItem( - toDeletedMessage({ - message, + // Batch: one logical operation touches many messages; coalesce the shared-store fan-out to a + // single flush (sibling holders are notified once) instead of once per affected message. + this._itemIndex.batch(() => { + for (const message of loadedMessages) { + if (message.user?.id === userId) { + if (hardDelete) { + this.removeItem({ id: message.id }); + } else { + this.ingestItem( + toDeletedMessage({ + message, + hardDelete, + deletedAt, + }) as LocalMessage, + ); + } + continue; + } + + if ( + message.quoted_message?.user?.id === userId && + message.quoted_message.type !== 'deleted' + ) { + this.ingestItem({ + ...message, + quoted_message: toDeletedMessage({ + message: formatMessage(message.quoted_message), hardDelete, deletedAt, }) as LocalMessage, - ); + }); } - continue; } - - if ( - message.quoted_message?.user?.id === userId && - message.quoted_message.type !== 'deleted' - ) { - this.ingestItem({ - ...message, - quoted_message: toDeletedMessage({ - message: formatMessage(message.quoted_message), - hardDelete, - deletedAt, - }) as LocalMessage, - }); - } - } + }); }; /** @@ -840,14 +870,18 @@ export class MessageIntervalPaginator extends BasePaginator< reflectQuotedMessageUpdate = (message: LocalMessage) => { const cachedMessages = this._itemIndex.values(); - for (const cachedMessage of cachedMessages) { - if (cachedMessage.quoted_message_id !== message.id) continue; + // Batch: several cached messages may quote the updated one; coalesce the shared-store fan-out + // to a single flush instead of one per re-ingested quoting message. + this._itemIndex.batch(() => { + for (const cachedMessage of cachedMessages) { + if (cachedMessage.quoted_message_id !== message.id) continue; - this.ingestItem({ - ...cachedMessage, - quoted_message: message, - }); - } + this.ingestItem({ + ...cachedMessage, + quoted_message: message, + }); + } + }); }; /** @@ -862,11 +896,15 @@ export class MessageIntervalPaginator extends BasePaginator< reflectUserUpdate = (user: UserResponse) => { const activeIds = new Set((this.items ?? []).map((m) => this.getItemId(m))); let activeAffected = false; - for (const message of this._itemIndex.values()) { - if (message.user?.id !== user.id) continue; - this._itemIndex.setOne({ ...message, user }); - if (activeIds.has(this.getItemId(message))) activeAffected = true; - } + // Batch: a user rename can touch many messages; coalesce the shared-store fan-out to sibling + // holders into a single flush. This paginator's own active window is re-emitted once below. + this._itemIndex.batch(() => { + for (const message of this._itemIndex.values()) { + if (message.user?.id !== user.id) continue; + this._itemIndex.setOne({ ...message, user }); + if (activeIds.has(this.getItemId(message))) activeAffected = true; + } + }); if (activeAffected) { this.state.partialNext({ items: (this.items ?? []).map((m) => this.getItem(this.getItemId(m)) ?? m), @@ -898,6 +936,15 @@ export class MessageIntervalPaginator extends BasePaginator< * @param [params.enforceUnique=false] - When adding, first clear the current user's existing * `own_reactions` so only the incoming one remains (used by `reaction.updated`, where a user's * reaction replaces their previous one). + * + * TODO(reactive-store): reflect reactions ONCE at the store level, not per-paginator. Both the + * channel handler (channel.ts) and the thread handler (thread.ts) call this on every reaction.* + * event, so a message held in more than one collection (a show_in_channel reply, or the thread + * parent) is reflected TWICE: two writes to the same canonical slot, each fanning out to the + * other holder (double re-projection) and minting a fresh ref that defeats the reconcile + * ref-equality bail. Idempotent (counts come wholesale from the event) so the result is correct, + * just wasteful. The store already fans out to every holder, so reflect once (by id, if held) and + * retire the per-collection reflect calls + parent path + enforce_unique branch. */ reflectReaction = ({ enforceUnique = false, @@ -913,35 +960,16 @@ export class MessageIntervalPaginator extends BasePaginator< const formatted = formatMessage(message); const existing = this.getItem(formatted.id); const baseOwnReactions = existing?.own_reactions ?? formatted.own_reactions ?? []; - const own_reactions = removed - ? this.removeOwnReactionOfType(baseOwnReactions, reaction) - : this.addOwnReaction(baseOwnReactions, reaction, enforceUnique); + const own_reactions = computeOwnReactions({ + current: baseOwnReactions, + enforceUnique, + reaction, + removed, + userId: this.channel.getClient().userID, + }); this.ingestItem({ ...formatted, own_reactions }); }; - private removeOwnReactionOfType( - ownReactions: ReactionResponse[], - reaction: ReactionResponse, - ): ReactionResponse[] { - return ownReactions.filter( - (r) => r.user_id !== reaction.user_id || r.type !== reaction.type, - ); - } - - private addOwnReaction( - ownReactions: ReactionResponse[], - reaction: ReactionResponse, - enforceUnique: boolean, - ): ReactionResponse[] { - const base = enforceUnique - ? [] - : this.removeOwnReactionOfType(ownReactions, reaction); - if (this.channel.getClient().userID === reaction.user_id) { - return [...base, reaction]; - } - return base; - } - /** * Map a timestamp to a loaded message — the first message in the latest (head) window whose * `created_at` is >= `timestampMs` (mirrors the legacy `ChannelState.findMessageByTimestamp` diff --git a/src/pagination/paginators/MessagePaginator.ts b/src/pagination/paginators/MessagePaginator.ts index 5591a71c2..68c0e6e2c 100644 --- a/src/pagination/paginators/MessagePaginator.ts +++ b/src/pagination/paginators/MessagePaginator.ts @@ -12,6 +12,8 @@ import { } from './MessageIntervalPaginator'; import type { LocalMessage } from '../../types'; import { StateStore } from '../../store'; +import { ItemIndex } from '../ItemIndex'; +import { MessageStoreBackedItemIndex } from '../../messageStore/MessageStoreBackedItemIndex'; export type { JumpToMessageOptions, @@ -127,7 +129,35 @@ export class MessagePaginator extends MessageIntervalPaginator { unreadReferencePolicy = 'snapshot', ...options }: MessagePaginatorOptions) { - super(options); + const { channel } = options; + super({ + ...options, + paginatorOptions: { + // Throttle message-list `state` publishes to at most once per 500ms (leading + trailing), so a + // burst of events coalesces into ~2 renders/sec instead of one per event. Optimistic + // (local-user) writes bypass the throttle via MessageStore.flushSubscribers → flushState. + // Overridable per-instance via `paginatorOptions.stateThrottleMs`. + stateThrottleMs: 500, + ...options.paginatorOptions, + // Back the channel main list and thread reply list with the client-global + // message store so a message held in more than one of them (a channel message + // also open in its thread, a `show_in_channel` reply in both) has a single + // canonical copy — no copy-to-copy fan-out. Falls back to a private index if + // the store is unavailable (e.g. a detached paginator in a test). + createItemIndex: + options.paginatorOptions?.createItemIndex ?? + ((owner) => { + const store = channel.getClient?.().messageStore; + return store + ? new MessageStoreBackedItemIndex({ + store, + owner: owner as MessageIntervalPaginator, + getId: owner.getItemId.bind(owner), + }) + : new ItemIndex({ getId: owner.getItemId.bind(owner) }); + }), + }, + }); this.unreadReferencePolicy = unreadReferencePolicy; this.unreadStateSnapshot = new StateStore({ lastReadAt: null, diff --git a/src/pagination/paginators/stateThrottling.ts b/src/pagination/paginators/stateThrottling.ts new file mode 100644 index 000000000..e7f4281a9 --- /dev/null +++ b/src/pagination/paginators/stateThrottling.ts @@ -0,0 +1,31 @@ +/** + * Global switch for the message-list `state` publish throttle (see `BasePaginator.scheduleWindowPublish`). + * + * Auto-disabled under test runners (Vitest / Jest) so the existing unit suites keep their synchronous, + * un-throttled behavior with zero per-test changes: with throttling off, every live mutation publishes + * to `state` immediately, exactly as before this feature. A dedicated throttle test flips it on + * (around fake timers) via {@link setStateThrottlingEnabled} and restores it afterwards. + * + * Production (an app bundle) defaults ON. + */ +const isTestRunner = (): boolean => { + try { + return ( + typeof process !== 'undefined' && + !!process.env && + (!!process.env.VITEST || + !!process.env.JEST_WORKER_ID || + process.env.NODE_ENV === 'test') + ); + } catch { + return false; + } +}; + +let enabled = !isTestRunner(); + +export const isStateThrottlingEnabled = (): boolean => enabled; + +export const setStateThrottlingEnabled = (value: boolean): void => { + enabled = value; +}; diff --git a/src/thread.ts b/src/thread.ts index 269204f69..9365892b9 100644 --- a/src/thread.ts +++ b/src/thread.ts @@ -1,5 +1,10 @@ import { StateStore } from './store'; -import { formatMessage, localMessageToNewMessagePayload } from './utils'; +import { + computeOwnReactions, + formatMessage, + localMessageToNewMessagePayload, +} from './utils'; +import { applyReactionLocally } from './messageStore'; import type { DraftResponse, EventAPIResponse, @@ -7,11 +12,14 @@ import type { LocalMessage, MarkReadRequest, MessageResponse, + ReactionRequest, ReadStateResponse, + SendReactionRequest, SortParamRequest, ThreadStateResponse, UserResponse, } from './types'; +import { isEphemeral } from './errors'; import type { Channel, DeleteMessageWithStateUpdateParams, @@ -219,7 +227,10 @@ export class Thread extends WithSubscriptions { }); this.messageOperations = new MessageOperations({ - ingest: (m) => this.messagePaginator.ingestItem(m), + ingest: (m) => { + this.messagePaginator.ingestItem(m); + this.channel.getClient().messageStore.flushSubscribers(m.id); + }, get: (id) => this.messagePaginator.getItem(id), normalizeOutgoingMessage: (m) => ({ ...m, @@ -366,7 +377,13 @@ export class Thread extends WithSubscriptions { isStateStale: false, }); - this.messagePaginator.mergeNewestPage(thread.messagePaginator.items ?? []); + if (parentMessage && this.hasSubscriptions) { + this.client.messageStore.upsert(parentMessage); + } + + this.messagePaginator.mergeNewestPage( + thread.messagePaginator.state.getLatestValue().items ?? [], + ); pendingReplies.forEach((reply) => this.messagePaginator.ingestItem(reply)); // Carry the re-queried thread's last-activity floor so lastMessageAt stays fresh even when the // merged page does not include the newest reply. Monotonic, so an older value is a no-op. @@ -379,6 +396,7 @@ export class Thread extends WithSubscriptions { return; } + this.addUnsubscribeFunction(this.subscribeParentMessageFromStore()); this.addUnsubscribeFunction(this.subscribeThreadUpdated()); this.addUnsubscribeFunction(this.subscribeMarkActiveThreadRead()); this.addUnsubscribeFunction(this.subscribeReloadActiveStaleThread()); @@ -525,21 +543,6 @@ export class Thread extends WithSubscriptions { this.state.partialNext({ read: nextRead }); }).unsubscribe; - private incrementReplyCountLocally = () => { - this.state.next((current) => { - const nextReplyCount = current.replyCount + 1; - - return { - ...current, - parentMessage: { - ...current.parentMessage, - reply_count: nextReplyCount, - }, - replyCount: nextReplyCount, - }; - }); - }; - private subscribeRepliesRead = () => this.client.on('message.read', (event) => { if (!event.user || !event.created_at || !event.thread) return; @@ -599,9 +602,9 @@ export class Thread extends WithSubscriptions { this.client.on(eventType, (event: PipelineEvent) => { if (!event.message) return; // A `message.updated` WS event carries `own_reactions: []`; upserting it verbatim would - // wipe the current user's reactions on a reply edit. The reply paginator is this thread's - // own source of truth (the channel no longer enriches reply events), so preserve the - // existing reply's `own_reactions`. + // wipe the current user's reactions on an edit. Preserve them off the copy we already hold + // — the reply paginator for a reply, `state.parentMessage` for the parent (the parent is + // not held in any paginator, so it needs the same treatment directly). const message = event.message.parent_id === this.id ? { @@ -610,7 +613,14 @@ export class Thread extends WithSubscriptions { this.messagePaginator.getItem(event.message.id)?.own_reactions ?? event.message.own_reactions, } - : event.message; + : !event.message.parent_id && event.message.id === this.id + ? { + ...event.message, + own_reactions: + this.state.getLatestValue().parentMessage?.own_reactions ?? + event.message.own_reactions, + } + : event.message; this.updateParentMessageOrReplyLocally(message); this.messagePaginator.reflectQuotedMessageUpdate(formatMessage(event.message)); }).unsubscribe, @@ -631,7 +641,20 @@ export class Thread extends WithSubscriptions { removed: eventType === 'reaction.deleted', }); } else if (!message.parent_id && message.id === this.id) { - this.updateParentMessageLocally({ message }); + // Reaction on the PARENT. The parent isn't in a paginator, so apply the current user's + // own_reactions delta here (mirroring the reply path's reflectReaction) rather than + // copying the WS event verbatim — which would drop own_reactions the event omits. + const own_reactions = computeOwnReactions({ + current: + this.state.getLatestValue().parentMessage?.own_reactions ?? + message.own_reactions ?? + [], + enforceUnique: eventType === 'reaction.updated', + reaction, + removed: eventType === 'reaction.deleted', + userId: this.client.userID, + }); + this.updateParentMessageLocally({ message: { ...message, own_reactions } }); } this.messagePaginator.reflectQuotedMessageUpdate(formatMessage(message)); }).unsubscribe, @@ -666,8 +689,39 @@ export class Thread extends WithSubscriptions { return () => unsubscribeFunctions.forEach((unsubscribe) => unsubscribe()); }; + // The parent message lives in the client-global message store (one canonical POJO per id); + // `state.parentMessage` and the fields derived from it are a projection of that copy. Every update + // to the parent — optimistic reaction, WS reaction/edit/delete, reply-count bump — writes the + // store, which reflects it here through this single subscription (one per thread, not one per + // message). Seeds the store on first subscribe when no other collection holds the parent yet + // (e.g. a thread opened from a notification, its parent not in the channel window). + private subscribeParentMessageFromStore = () => { + const store = this.client.messageStore; + const parent = this.state.getLatestValue().parentMessage; + if (parent && !store.has(parent.id)) store.upsert(parent); + + return store.subscribe(this.id, (message) => { + if (!message) return; + this.state.next((current) => ({ + ...current, + deletedAt: message.deleted_at ?? null, + parentMessage: message, + participants: + normalizeThreadParticipants(message.thread_participants, current.channel.cid) ?? + current.participants, + replyCount: message.reply_count ?? current.replyCount, + })); + }); + }; + public unregisterSubscriptions = () => { const symbol = super.unregisterSubscriptions(); + // Release the reply paginator's hold on the shared message store. The parent subscription is + // torn down by `super.unregisterSubscriptions()` (it was added as an unsubscribe function), but + // the reply paginator's per-id links live in its item index, not in the subscription list — so a + // removed thread would otherwise stay pinned by `messageStore.subscribers` and keep its replies + // alive. `getThread` builds a fresh instance if this thread is re-opened, so this is a discard. + this.messagePaginator.dispose(); this.state.partialNext({ isStateStale: true }); return symbol; }; @@ -712,19 +766,12 @@ export class Thread extends WithSubscriptions { throw new Error('Message does not belong to this thread'); } - this.state.next((current) => { - const formattedMessage = formatMessage(message); - - return { - ...current, - deletedAt: formattedMessage.deleted_at ?? null, - parentMessage: formattedMessage, - participants: - normalizeThreadParticipants(message.thread_participants, current.channel.cid) ?? - current.participants, - replyCount: message.reply_count ?? current.replyCount, - }; - }); + // The parent's content lives in the client-global message store; `state.parentMessage` (and the + // fields derived from it) is a projection kept in sync by `subscribeParentMessageFromStore`. + // Writing the store fans the change out to every collection holding this id and reflects it here. + if (this.client.messageStore.has(message.id)) { + this.client.messageStore.upsert(formatMessage(message)); + } }; // todo: can be removed with the next breaking change and use MessagePaginator only @@ -800,6 +847,80 @@ export class Thread extends WithSubscriptions { ); } + /** + * Adds a reaction to a reply with an optimistic local state update, mirroring + * {@link Channel.addReactionWithLocalUpdate}. The optimistic message is applied to THIS thread's + * paginator (so pure replies get optimism the channel paginator can't give); the request routes + * through the parent channel since reactions are channel-level. + */ + async addReactionWithLocalUpdate({ + messageId, + reaction, + options, + }: { + messageId: string; + reaction: ReactionRequest; + options?: Pick; + }) { + const client = this.channel.getClient(); + const undo = applyReactionLocally(client, { + enforceUnique: options?.enforce_unique ?? false, + messageId, + reaction, + }); + + try { + const response = await this.channel.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 from a reply with an optimistic local state update, + * mirroring {@link Thread.addReactionWithLocalUpdate}. + */ + async deleteReactionWithLocalUpdate({ + messageId, + type, + }: { + messageId: string; + type: string; + }) { + const client = this.channel.getClient(); + const undo = applyReactionLocally(client, { + messageId, + reaction: { type }, + removed: true, + }); + + try { + const response = await this.channel.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; + } + } + public markRead = async ({ force = false }: { force?: boolean } = {}) => { if (this.ownUnreadCount === 0 && !force) { return null; diff --git a/src/thread_manager.ts b/src/thread_manager.ts index 0b7d5bb19..2a8d22c21 100644 --- a/src/thread_manager.ts +++ b/src/thread_manager.ts @@ -228,7 +228,7 @@ export class ThreadManager extends WithSubscriptions { }, DEFAULT_CONNECTION_RECOVERY_THROTTLE_DURATION, { trailing: true }, - ); + ).throttledFn; const unsubscribeConnectionRecovered = this.client.on( 'connection.recovered', diff --git a/src/utils.ts b/src/utils.ts index ac3bd6d9f..5b98b5b8c 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -11,6 +11,7 @@ import type { OwnUserResponse, PromoteChannelParams, ReactionGroupResponse, + ReactionResponse, UpdatedMessage, UserResponse, } from './types'; @@ -331,6 +332,136 @@ export function formatMessage(message: MessageResponse | LocalMessage): LocalMes } as LocalMessage; } +/** + * Computes the current user's `own_reactions` after applying a single reaction change, off a + * `current` base. A WS reaction/edit event carries `own_reactions: []` (or stale), so consumers + * must recompute from the copy they already hold rather than trusting the event — this is the + * shared core used by both the message paginator (`reflectReaction`) and the thread's parent + * message (which is not held in any paginator, so it needs the same logic directly). + * + * - `removed`: drop the current user's reaction of this type. + * - `enforceUnique`: replace any existing own reaction with the incoming one (used by + * `reaction.updated`, where a user's reaction supersedes their previous one). + * - otherwise: add the reaction, de-duped by type. + * + * A reaction by another user never changes the current user's `own_reactions`. + */ +export function computeOwnReactions({ + current, + enforceUnique = false, + reaction, + removed = false, + userId, +}: { + current: ReactionResponse[]; + reaction: ReactionResponse; + enforceUnique?: boolean; + removed?: boolean; + userId?: string; +}): ReactionResponse[] { + const withoutType = current.filter( + (r) => r.user_id !== reaction.user_id || r.type !== reaction.type, + ); + if (removed) return withoutType; + if (userId !== reaction.user_id) return withoutType; + return enforceUnique ? [reaction] : [...withoutType, reaction]; +} + +/** + * Returns a copy of `message` with `reaction` folded into its `reaction_groups` / + * `latest_reactions`. Does not touch `own_reactions` — the reaction entry points + * ({@link MessageIntervalPaginator.reflectReaction} / `Thread.applyParentReactionLocally`) own that + * via {@link computeOwnReactions}. Shared so both the paginator and the (paginator-less) thread + * parent compute counts identically. + */ +export function messageWithReactionAdded( + message: LocalMessage, + reaction: ReactionResponse, + enforceUnique: boolean, +): LocalMessage { + const score = reaction.score ?? 1; + const reactionGroups: Record = { + ...(message.reaction_groups ?? {}), + }; + + // When enforcing uniqueness, first back the current user's existing reactions out of the groups + if (enforceUnique) { + for (const ownReaction of message.own_reactions ?? []) { + const group = reactionGroups[ownReaction.type]; + if (!group) continue; + const next = { + ...group, + count: group.count - 1, + sum_scores: group.sum_scores - (ownReaction.score ?? 1), + }; + if (next.count < 1) delete reactionGroups[ownReaction.type]; + else reactionGroups[ownReaction.type] = next; + } + } + + const existingGroup = reactionGroups[reaction.type]; + reactionGroups[reaction.type] = existingGroup + ? { + ...existingGroup, + count: existingGroup.count + 1, + last_reaction_at: reaction.created_at, + sum_scores: existingGroup.sum_scores + score, + } + : { + count: 1, + first_reaction_at: reaction.created_at, + last_reaction_at: reaction.created_at, + latest_reactions_by: [], + sum_scores: score, + }; + + const latestReactions = enforceUnique + ? [ + ...(message.latest_reactions ?? []).filter((r) => r.user_id !== reaction.user_id), + reaction, + ] + : [...(message.latest_reactions ?? []), reaction]; + + return { + ...message, + latest_reactions: latestReactions, + reaction_groups: reactionGroups, + }; +} + +/** + * Returns a copy of `message` with the current user's reaction of `reaction.type` backed out of its + * `reaction_groups` / `latest_reactions`. Does not touch `own_reactions`. + */ +export function messageWithReactionRemoved( + message: LocalMessage, + reaction: ReactionResponse, +): LocalMessage { + const reactionGroups: Record = { + ...(message.reaction_groups ?? {}), + }; + const reactionToRemove = message.own_reactions?.find((r) => r.type === reaction.type); + + if (reactionToRemove && reactionGroups[reactionToRemove.type]) { + const group = reactionGroups[reactionToRemove.type]; + const next = { + ...group, + count: group.count - 1, + sum_scores: group.sum_scores - (reactionToRemove.score ?? 1), + }; + if (next.count < 1) delete reactionGroups[reactionToRemove.type]; + else reactionGroups[reactionToRemove.type] = next; + } + + return { + ...message, + latest_reactions: message.latest_reactions?.filter( + (r) => !(r.user_id === reaction.user_id && r.type === reaction.type), + ), + reaction_groups: reactionGroups, + }; +} + export const localMessageToNewMessagePayload = ( localMessage: LocalMessage, ): MessageRequest => { @@ -624,43 +755,14 @@ export const debounce = any>( return debouncedFn; }; -// works exactly the same as lodash.throttle - -export const throttle = any>( - fn: T, - timeout = 200, - { leading = true, trailing = false }: { leading?: boolean; trailing?: boolean } = {}, -) => { - let runningTimeout: null | NodeJS.Timeout = null; - let storedArgs: Parameters | null = null; - - return (...args: Parameters) => { - if (runningTimeout) { - if (trailing) storedArgs = args; - return; - } - - if (leading) { - fn(...args); - } else if (trailing) { - storedArgs = args; - } - - const timeoutHandler = () => { - if (storedArgs) { - fn(...storedArgs); - storedArgs = null; - runningTimeout = setTimeout(timeoutHandler, timeout); - - return; - } - - runningTimeout = null; - }; - - runningTimeout = setTimeout(timeoutHandler, timeout); - }; -}; +// The single throttle implementation lives in ./utils/throttling/throttle; re-exported here so +// `import { throttle } from './utils'` keeps working (lodash.throttle-style leading/trailing). +export { throttle } from './utils/throttling/throttle'; +export type { + Throttled, + ThrottleOptions, + ThrottledCallback, +} from './utils/throttling/throttle'; const get = (obj: T, path: string): unknown => path.split('.').reduce((acc, key) => { diff --git a/src/utils/throttling/throttle.ts b/src/utils/throttling/throttle.ts new file mode 100644 index 000000000..916edf214 --- /dev/null +++ b/src/utils/throttling/throttle.ts @@ -0,0 +1,130 @@ +export type ThrottledCallback = (...args: unknown[]) => unknown; + +export type ThrottleOptions = { + /** Call on the leading edge (default: true). */ + leading?: boolean; + /** Call once at the end of the window with the latest args (default: false). */ + trailing?: boolean; +}; + +export type Throttled = { + /** + * The throttled function — call it as often as you like; it invokes `fn` at most once per window + * (leading and/or trailing per options). + */ + throttledFn: (...args: T) => void; + /** Clear a pending trailing invocation WITHOUT firing it. */ + cancelTimer: () => void; + /** + * Fire a pending trailing invocation immediately (with the latest stored args), clearing the timer. + * No-op when nothing is pending. Use to bypass the throttle delay for updates that must land now. + */ + flush: () => void; +}; + +/** + * Throttle a function so it runs at most once per `timeout` ms. + * + * - `leading`: fire immediately when the window opens + * - `trailing`: remember the latest args/this and fire once when the window closes + * + * defaults: `{ leading: true, trailing: false }` + * + * Copied verbatim from the Feeds client (`packages/feeds-client/src/utils/throttling/throttle.ts`) + * and extended with a `flush()` method (the Feeds version only exposes `cancelTimer()`). + * + * notes: + * - make one throttled instance and reuse it; re-creating it resets internal state + */ +export const throttle = ( + fn: (...args: T) => void, + timeout = 200, + { leading = true, trailing = false }: ThrottleOptions = {}, +): Throttled => { + let timer: ReturnType | null = null; + let storedArgs: T | null = null; + let storedThis: unknown = null; + let lastInvokeTime: number | undefined; + + const invoke = (args: T, thisArg: unknown) => { + lastInvokeTime = Date.now(); + fn.apply(thisArg, args); + }; + + const scheduleTrailing = (delay: number) => { + if (timer) return; + timer = setTimeout(() => { + timer = null; + if (trailing && storedArgs) { + invoke(storedArgs, storedThis); + storedArgs = null; + storedThis = null; + } + }, delay); + }; + + return { + throttledFn(this: unknown, ...args: T) { + const now = Date.now(); + + const lastInvoke = lastInvokeTime; + + if (lastInvoke == null && !leading) lastInvokeTime = now; + + const timeSinceLast = lastInvoke == null ? timeout : now - lastInvoke; + const remaining = timeout - timeSinceLast; + + if (trailing) { + storedArgs = args; + // eslint-disable-next-line @typescript-eslint/no-this-alias + storedThis = this; + } + + if (remaining <= 0) { + if (timer) { + clearTimeout(timer); + timer = null; + } + + if (leading) { + if (trailing) { + if (storedArgs === args) { + storedArgs = null; + storedThis = null; + } + } + invoke(args, this); + } else { + if (trailing) scheduleTrailing(timeout); + } + + return; + } + + if (trailing && !timer) { + scheduleTrailing(remaining); + } + }, + cancelTimer: () => { + if (timer) { + clearTimeout(timer); + timer = null; + } + // We discard the pending trailing too, otherwise a later flush() would refire + // the invocation this cancel was meant to drop. + storedArgs = null; + storedThis = null; + }, + flush: () => { + if (timer) { + clearTimeout(timer); + timer = null; + } + if (trailing && storedArgs) { + invoke(storedArgs, storedThis); + storedArgs = null; + storedThis = null; + } + }, + }; +}; diff --git a/test/unit/MessageStore.test.ts b/test/unit/MessageStore.test.ts new file mode 100644 index 000000000..fb708ad8b --- /dev/null +++ b/test/unit/MessageStore.test.ts @@ -0,0 +1,179 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { MessageStore } from '../../src/messageStore/MessageStore'; +import type { MessageStoreSubscriber } from '../../src/messageStore/MessageStore'; +import { formatMessage } from '../../src/utils'; +import { generateMsg } from './test-utils/generateMessage'; +import type { LocalMessage } from '../../src'; + +const msg = (overrides: Partial[0]> = {}): LocalMessage => + formatMessage(generateMsg(overrides)); + +const spySubscriber = (): MessageStoreSubscriber & { + onMessagesChanged: ReturnType; +} => ({ + onMessagesChanged: vi.fn(), +}); + +describe('MessageStore', () => { + let store: MessageStore; + + beforeEach(() => { + store = new MessageStore(); + }); + + describe('reads / writes', () => { + it('stores and reads a message by id', () => { + const m = msg({ id: 'm1' }); + store.upsert(m); + expect(store.get('m1')).toBe(m); + expect(store.has('m1')).toBe(true); + }); + + it('returns undefined for missing / non-string ids', () => { + expect(store.get('nope')).toBeUndefined(); + expect(store.get(undefined)).toBeUndefined(); + expect(store.has('nope')).toBe(false); + }); + + it('replaces the canonical copy on upsert (immutable)', () => { + const first = msg({ id: 'm1', text: 'a' }); + const second = { ...first, text: 'b' }; + store.upsert(first); + store.upsert(second); + expect(store.get('m1')).toBe(second); + }); + }); + + describe('subscribe (atomic)', () => { + it('fires immediately with current value, then on change, and stops after unsubscribe', () => { + const handler = vi.fn(); + store.upsert(msg({ id: 'm1', text: 'a' })); + + const unsubscribe = store.subscribe('m1', handler); + expect(handler).toHaveBeenCalledTimes(1); + expect(handler).toHaveBeenLastCalledWith(expect.objectContaining({ text: 'a' })); + + store.upsert(msg({ id: 'm1', text: 'b' })); + expect(handler).toHaveBeenCalledTimes(2); + expect(handler).toHaveBeenLastCalledWith(expect.objectContaining({ text: 'b' })); + + unsubscribe(); + store.upsert(msg({ id: 'm1', text: 'c' })); + expect(handler).toHaveBeenCalledTimes(2); + }); + + it('fires immediately with undefined when the id is absent', () => { + const handler = vi.fn(); + store.subscribe('ghost', handler); + expect(handler).toHaveBeenCalledTimes(1); + expect(handler).toHaveBeenLastCalledWith(undefined); + }); + }); + + describe('link / unlink notification routing', () => { + it('notifies only subscribers linked to the changed id', () => { + const a = spySubscriber(); + const b = spySubscriber(); + store.link('m1', a); + store.link('m2', b); + + store.upsert(msg({ id: 'm1' })); + expect(a.onMessagesChanged).toHaveBeenCalledTimes(1); + expect(b.onMessagesChanged).not.toHaveBeenCalled(); + + const batch = a.onMessagesChanged.mock.calls[0][0]; + expect([...batch.changedIds]).toEqual(['m1']); + }); + + it('skips the origin subscriber but notifies other holders', () => { + const origin = spySubscriber(); + const sibling = spySubscriber(); + store.link('m1', origin); + store.link('m1', sibling); + + store.upsert(msg({ id: 'm1' }), origin); + expect(origin.onMessagesChanged).not.toHaveBeenCalled(); + expect(sibling.onMessagesChanged).toHaveBeenCalledTimes(1); + }); + + it('does not notify a subscriber after it unlinks', () => { + const a = spySubscriber(); + store.link('m1', a); + store.unlink('m1', a); + store.upsert(msg({ id: 'm1' })); + expect(a.onMessagesChanged).not.toHaveBeenCalled(); + }); + }); + + describe('refcount GC', () => { + it('drops the canonical copy when the last holder unlinks', () => { + const a = spySubscriber(); + store.upsert(msg({ id: 'm1' })); + store.link('m1', a); + + store.unlink('m1', a); + expect(store.has('m1')).toBe(false); + expect(store.get('m1')).toBeUndefined(); + }); + + it('keeps the message alive while another holder remains', () => { + const a = spySubscriber(); + const b = spySubscriber(); + store.upsert(msg({ id: 'm1' })); + store.link('m1', a); + store.link('m1', b); + + store.unlink('m1', a); + expect(store.has('m1')).toBe(true); + + store.unlink('m1', b); + expect(store.has('m1')).toBe(false); + }); + }); + + describe('transaction batching', () => { + it('coalesces multiple writes into one notification per subscriber', () => { + const a = spySubscriber(); + store.link('m1', a); + store.link('m2', a); + store.link('m3', a); + + store.transaction(() => { + store.upsert(msg({ id: 'm1' })); + store.upsert(msg({ id: 'm2' })); + store.upsert(msg({ id: 'm3' })); + }); + + expect(a.onMessagesChanged).toHaveBeenCalledTimes(1); + const batch = a.onMessagesChanged.mock.calls[0][0]; + expect([...batch.changedIds].sort()).toEqual(['m1', 'm2', 'm3']); + }); + + it('flushes only when the outermost transaction exits', () => { + const a = spySubscriber(); + store.link('m1', a); + + store.transaction(() => { + store.transaction(() => { + store.upsert(msg({ id: 'm1' })); + }); + expect(a.onMessagesChanged).not.toHaveBeenCalled(); + }); + expect(a.onMessagesChanged).toHaveBeenCalledTimes(1); + }); + + it('does not notify a subscriber whose watched ids were untouched', () => { + const a = spySubscriber(); + const b = spySubscriber(); + store.link('m1', a); + store.link('m2', b); + + store.transaction(() => { + store.upsert(msg({ id: 'm1' })); + }); + + expect(a.onMessagesChanged).toHaveBeenCalledTimes(1); + expect(b.onMessagesChanged).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/test/unit/MessageStoreBackedItemIndex.test.ts b/test/unit/MessageStoreBackedItemIndex.test.ts new file mode 100644 index 000000000..a740036eb --- /dev/null +++ b/test/unit/MessageStoreBackedItemIndex.test.ts @@ -0,0 +1,139 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { MessageStore } from '../../src/messageStore/MessageStore'; +import type { MessageStoreSubscriber } from '../../src/messageStore/MessageStore'; +import { MessageStoreBackedItemIndex } from '../../src/messageStore/MessageStoreBackedItemIndex'; +import { formatMessage } from '../../src/utils'; +import { generateMsg } from './test-utils/generateMessage'; +import type { LocalMessage } from '../../src'; + +const msg = (overrides: Partial[0]> = {}): LocalMessage => + formatMessage(generateMsg(overrides)); + +const spyOwner = (): MessageStoreSubscriber & { + onMessagesChanged: ReturnType; +} => ({ + onMessagesChanged: vi.fn(), +}); + +const getId = (m: LocalMessage) => m.id; + +describe('MessageStoreBackedItemIndex', () => { + let store: MessageStore; + let ownerA: ReturnType; + let ownerB: ReturnType; + let a: MessageStoreBackedItemIndex; + let b: MessageStoreBackedItemIndex; + + beforeEach(() => { + store = new MessageStore(); + ownerA = spyOwner(); + ownerB = spyOwner(); + a = new MessageStoreBackedItemIndex({ store, owner: ownerA, getId }); + b = new MessageStoreBackedItemIndex({ store, owner: ownerB, getId }); + }); + + describe('membership scoping', () => { + it('reads content it holds and hides content it does not', () => { + const m = msg({ id: 'm1' }); + a.setOne(m); + expect(a.get('m1')).toBe(m); + expect(a.has('m1')).toBe(true); + // b never ingested m1: it must not see it, even though the store holds it + expect(b.get('m1')).toBeUndefined(); + expect(b.has('m1')).toBe(false); + }); + + it('values()/entries() are scoped to this index membership', () => { + a.setOne(msg({ id: 'm1' })); + a.setOne(msg({ id: 'm2' })); + b.setOne(msg({ id: 'm3' })); + expect(a.values().map(getId).sort()).toEqual(['m1', 'm2']); + expect( + a + .entries() + .map(([id]) => id) + .sort(), + ).toEqual(['m1', 'm2']); + expect(b.values().map(getId)).toEqual(['m3']); + }); + }); + + describe('shared content', () => { + it('both indexes read the single canonical copy when both hold the id', () => { + a.setOne(msg({ id: 'm1', text: 'v1' })); + b.setOne(msg({ id: 'm1', text: 'v1' })); + const updated = msg({ id: 'm1', text: 'v2' }); + a.setOne(updated); + expect(a.get('m1')).toBe(updated); + expect(b.get('m1')).toBe(updated); + }); + }); + + describe('notification', () => { + it('does not notify the writing owner but notifies other holders (the fan-out)', () => { + a.setOne(msg({ id: 'm1', text: 'v1' })); + b.setOne(msg({ id: 'm1', text: 'v1' })); + ownerA.onMessagesChanged.mockClear(); + ownerB.onMessagesChanged.mockClear(); + + a.setOne(msg({ id: 'm1', text: 'v2' })); + + expect(ownerA.onMessagesChanged).not.toHaveBeenCalled(); + expect(ownerB.onMessagesChanged).toHaveBeenCalledTimes(1); + expect([...ownerB.onMessagesChanged.mock.calls[0][0].changedIds]).toEqual(['m1']); + }); + + it('does not notify a holder of an id it does not hold', () => { + a.setOne(msg({ id: 'm1' })); + ownerB.onMessagesChanged.mockClear(); + a.setOne(msg({ id: 'm1', text: 'again' })); + expect(ownerB.onMessagesChanged).not.toHaveBeenCalled(); + }); + }); + + describe('refcount via remove/clear', () => { + it('keeps content alive while another index still holds it', () => { + a.setOne(msg({ id: 'm1' })); + b.setOne(msg({ id: 'm1' })); + + a.remove('m1'); + expect(a.get('m1')).toBeUndefined(); + expect(b.get('m1')).toBeDefined(); + expect(store.has('m1')).toBe(true); + + b.remove('m1'); + expect(store.has('m1')).toBe(false); + }); + + it('clear() unlinks only this index membership, leaving the other index untouched', () => { + a.setOne(msg({ id: 'm1' })); + a.setOne(msg({ id: 'm2' })); + b.setOne(msg({ id: 'm2' })); + + a.clear(); + expect(a.values()).toEqual([]); + expect(store.has('m1')).toBe(false); // only a held m1 -> GC'd + expect(store.has('m2')).toBe(true); // b still holds m2 + expect(b.get('m2')).toBeDefined(); + }); + }); + + describe('batching', () => { + it('setMany coalesces sibling notifications to one', () => { + // b holds m1..m3 first so it is a sibling holder for each + b.setOne(msg({ id: 'm1' })); + b.setOne(msg({ id: 'm2' })); + b.setOne(msg({ id: 'm3' })); + ownerB.onMessagesChanged.mockClear(); + + a.setMany([msg({ id: 'm1' }), msg({ id: 'm2' }), msg({ id: 'm3' })]); + + expect(ownerB.onMessagesChanged).toHaveBeenCalledTimes(1); + expect([...ownerB.onMessagesChanged.mock.calls[0][0].changedIds].sort()).toEqual([ + 'm1', + 'm2', + 'm3', + ]); + }); + }); +}); diff --git a/test/unit/channel.test.js b/test/unit/channel.test.js index e2f961d89..999a3662f 100644 --- a/test/unit/channel.test.js +++ b/test/unit/channel.test.js @@ -3126,17 +3126,14 @@ describe('delete reaction flow', () => { vi.resetAllMocks(); }); - it('calls offlineDb.deleteReaction and queues task if offlineDb exists', async () => { + it('queues task if offlineDb exists', async () => { await channel.deleteReaction(request); - expect(deleteReactionSpy).toHaveBeenCalledTimes(1); expect(queueTaskSpy).toHaveBeenCalledTimes(1); - // The optimistic reaction now carries only message_id and type. - expect(deleteReactionSpy).toHaveBeenCalledWith({ - message: channel.messagePaginator.getItem(messageId), - reaction: { message_id: messageId, type: reactionType }, - }); + // The optimistic reaction-row removal is handled by the local-update layer + // (`applyReactionLocally`); `deleteReaction` itself only queues the replay task. + expect(deleteReactionSpy).not.toHaveBeenCalled(); expect(queueTaskSpy).toHaveBeenCalledWith({ task: { @@ -3170,7 +3167,7 @@ describe('delete reaction flow', () => { }); it('falls back to _deleteReaction if offlineDb throws', async () => { - deleteReactionSpy.mockRejectedValue(new Error('Offline failure')); + queueTaskSpy.mockRejectedValue(new Error('Offline failure')); await channel.deleteReaction(request); diff --git a/test/unit/offline-support/offline_support_api.test.ts b/test/unit/offline-support/offline_support_api.test.ts index b6af100c4..2674bf3a4 100644 --- a/test/unit/offline-support/offline_support_api.test.ts +++ b/test/unit/offline-support/offline_support_api.test.ts @@ -1830,18 +1830,26 @@ describe('OfflineSupportApi', () => { offlineDb, ); - expect( - shouldSkipQueueingTask({ response: { data: { code: 4 } } } as AxiosError), - ).toBe(true); - - expect( - shouldSkipQueueingTask({ response: { data: { code: 17 } } } as AxiosError), - ).toBe(true); - - expect( - shouldSkipQueueingTask({ response: { data: { code: 999 } } } as AxiosError), - ).toBe(false); - + // The offline replay path surfaces a `StreamAPIError`, which exposes the Stream error code + // at the top-level `code` (copied from `response.data.code` by the api-client) alongside a + // `response`. `shouldSkipQueueingTask` is `!isEphemeral`, so a task is skipped only when the + // server responded with a NON-retryable code; retryable codes and pure network/connection + // failures (no `response`) are kept in the queue for a later retry. + const serverError = (code: number) => + ({ code, response: {} }) as unknown as AxiosError; + + // Retryable server codes → ephemeral → keep queued (do NOT skip). + expect(shouldSkipQueueingTask(serverError(9))).toBe(false); // RateLimitError + expect(shouldSkipQueueingTask(serverError(23))).toBe(false); // RequestTimeoutError + + // Non-retryable server codes → skip (a retry would never succeed). + expect(shouldSkipQueueingTask(serverError(4))).toBe(true); // InputError + expect(shouldSkipQueueingTask(serverError(17))).toBe(true); // NotAllowedError + + // Unknown server code (not in APIErrorCodes) → treated as non-retryable → skip. + expect(shouldSkipQueueingTask(serverError(999))).toBe(true); + + // Network/connection failure — server never responded → ephemeral → keep queued. expect(shouldSkipQueueingTask({} as AxiosError)).toBe(false); }); diff --git a/test/unit/pagination/paginators/BasePaginator.test.ts b/test/unit/pagination/paginators/BasePaginator.test.ts index e793a5f2d..2762ec8d4 100644 --- a/test/unit/pagination/paginators/BasePaginator.test.ts +++ b/test/unit/pagination/paginators/BasePaginator.test.ts @@ -2954,6 +2954,88 @@ describe('BasePaginator', () => { anchoredHead: [], }); }); + + // A sibling holder (another paginator / WS echo) writes new content for an item held in a + // tracked interval view. Only the active window used to be refreshed on a sibling update, so + // the off-window views (logicalHead/logicalTail/anchoredHead) kept stale item references. These + // two tests drive the store-agnostic reconcile hook (reconcileChangedIds) directly after + // replacing the stored item, mimicking a sibling write. + it('refreshes logicalHead on a sibling update to an item that is off the active window', () => { + const index = new ItemIndex({ getId: ({ id }) => id }); + const paginator = new Paginator({ itemIndex: index }); + paginator.sortComparator = descByAge(); + + // A bounded (non-head) window, loaded and active. + paginator.ingestPage({ + page: [makeItem('m1', 50), makeItem('m2', 40)], + isHead: false, + isTail: false, + setActive: true, + }); + // An out-of-order head item lands in the logical head (NOT the active interval). + paginator.ingestItem(makeItem('x', 100)); + expect(paginator.logicalHeadItems.map((i) => i.id)).toEqual(['x']); + + const activeItemsBefore = paginator.items; + const logicalHead = trackKey(paginator, 'logicalHead'); + const logicalTail = trackKey(paginator, 'logicalTail'); + const anchoredHead = trackKey(paginator, 'anchoredHead'); + + // A sibling holder rewrites x's content (same id/order, e.g. a reaction) and notifies. + index.setOne({ id: 'x', age: 100, name: 'x-reacted' }); + // @ts-expect-error driving the protected store-agnostic reconcile hook directly + paginator.reconcileChangedIds(new Set(['x'])); + + // logicalHead republished with the new content... + expect(logicalHead.tracker.fires).toBe(1); + expect(paginator.logicalHeadItems.map((i) => i.name)).toEqual(['x-reacted']); + // ...the active window (x is not in it) and the other views are untouched. + expect(paginator.items).toBe(activeItemsBefore); + expect(logicalTail.tracker.fires).toBe(0); + expect(anchoredHead.tracker.fires).toBe(0); + + logicalHead.unsub(); + logicalTail.unsub(); + anchoredHead.unsub(); + }); + + it('refreshes anchoredHead on a sibling update to a message it holds', () => { + const index = new ItemIndex({ getId: ({ id }) => id }); + const paginator = new Paginator({ itemIndex: index }); + paginator.sortComparator = descByAge(); + + // The isHead page populates anchoredHead (and is the active interval). + paginator.ingestPage({ + page: [makeItem('m1', 50), makeItem('m2', 40)], + isHead: true, + isTail: false, + setActive: true, + }); + expect(paginator.anchoredHeadItems.map((i) => i.id)).toEqual(['m1', 'm2']); + + const anchoredHead = trackKey(paginator, 'anchoredHead'); + const logicalHead = trackKey(paginator, 'logicalHead'); + const logicalTail = trackKey(paginator, 'logicalTail'); + + // A sibling holder rewrites m2's content (same id/order) and notifies. + index.setOne({ id: 'm2', age: 40, name: 'm2-reacted' }); + // @ts-expect-error driving the protected store-agnostic reconcile hook directly + paginator.reconcileChangedIds(new Set(['m2'])); + + // anchoredHead republished with the new content, unchanged siblings preserved by id order... + expect(anchoredHead.tracker.fires).toBe(1); + expect(paginator.anchoredHeadItems.map((i) => i.name)).toEqual([ + 'm1', + 'm2-reacted', + ]); + // ...no logical view was touched. + expect(logicalHead.tracker.fires).toBe(0); + expect(logicalTail.tracker.fires).toBe(0); + + anchoredHead.unsub(); + logicalHead.unsub(); + logicalTail.unsub(); + }); }); describe('removeItem', () => { @@ -3072,6 +3154,24 @@ describe('BasePaginator', () => { // @ts-expect-error accessing protected property expect(Array.from(paginator._itemIntervals.values())).toStrictEqual([]); }); + + it('drops item-index membership so the id is no longer addressable', () => { + // Fresh index so the shared module-level one is not polluted across tests. + const index = new ItemIndex({ getId: ({ id }) => id }); + const paginator = new Paginator({ itemIndex: index }); + paginator.ingestPage({ page: [item1, item2, item3], setActive: true }); + expect(paginator.getItem(item2.id)).toStrictEqual(item2); + + paginator.removeItem({ id: item2.id }); + + // Regression: removeItem used to remove from intervals/state.items but leave the id in the + // item index. With a store-backed index that leaked a refcount (the message was never + // GC'd) and left getItem returning a no-longer-listed "ghost". Membership must drop too. + expect(paginator.getItem(item2.id)).toBeUndefined(); + // untouched siblings stay addressable + expect(paginator.getItem(item1.id)).toStrictEqual(item1); + expect(paginator.getItem(item3.id)).toStrictEqual(item3); + }); }); describe('headItems (newest loaded window)', () => { diff --git a/test/unit/pagination/paginators/MessagePaginatorThrottle.test.ts b/test/unit/pagination/paginators/MessagePaginatorThrottle.test.ts new file mode 100644 index 000000000..e1608a61b --- /dev/null +++ b/test/unit/pagination/paginators/MessagePaginatorThrottle.test.ts @@ -0,0 +1,354 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { MessagePaginator } from '../../../../src/pagination/paginators/MessagePaginator'; +import { setStateThrottlingEnabled } from '../../../../src/pagination/paginators/stateThrottling'; +import { MessageStore } from '../../../../src/messageStore/MessageStore'; +import { applyReactionLocally } from '../../../../src/messageStore/applyReactionLocally'; +import { formatMessage } from '../../../../src'; +import { generateMsg } from '../../test-utils/generateMessage'; +import type { Channel } from '../../../../src/channel'; +import type { StreamChat } from '../../../../src/client'; +import type { LocalMessage, MessageResponse, Reaction } from '../../../../src/types'; + +const msg = (id: string, day: number): LocalMessage => + formatMessage( + generateMsg({ + id, + cid: 'channel-id', + created_at: `2020-01-${String(day).padStart(2, '0')}T00:00:00.000Z`, + }) as MessageResponse, + ); + +const ids = (p: MessagePaginator) => p.items?.map((m) => m.id); + +const THROTTLE = 200; + +describe('MessagePaginator — state publish throttling', () => { + let channel: Channel; + + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(0); + // Vitest auto-disables state throttling for the rest of the suite; turn it on here. + setStateThrottlingEnabled(true); + channel = { + cid: 'channel-id', + getReplies: vi.fn(), + query: vi.fn(), + } as unknown as Channel; + }); + + afterEach(() => { + setStateThrottlingEnabled(false); + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + const make = () => + new MessagePaginator({ channel, paginatorOptions: { stateThrottleMs: THROTTLE } }); + + const seed = (p: MessagePaginator) => + p.setItems({ + valueOrFactory: [msg('m1', 1), msg('m2', 2)], + isFirstPage: true, + isLastPage: true, + }); + + it('coalesces a burst of ingests: leading emits the first, the rest land on the trailing edge', () => { + const p = make(); + seed(p); + expect(ids(p)).toEqual(['m1', 'm2']); + + p.ingestItem(msg('m3', 3)); // leading edge -> immediate + expect(ids(p)).toEqual(['m1', 'm2', 'm3']); + + p.ingestItem(msg('m4', 4)); // within window -> deferred + p.ingestItem(msg('m5', 5)); // within window -> deferred + expect(ids(p)).toEqual(['m1', 'm2', 'm3']); // not published yet + + vi.advanceTimersByTime(THROTTLE); // trailing edge + expect(ids(p)).toEqual(['m1', 'm2', 'm3', 'm4', 'm5']); + }); + + it('notifies `state` subscribers at throttled cadence, not once per event', () => { + const p = make(); + seed(p); + const handler = vi.fn(); + p.state.subscribe(handler); // fires once immediately with the seeded value + handler.mockClear(); + + for (let i = 3; i <= 12; i++) p.ingestItem(msg(`m${i}`, i)); // 10 ingests + // only the leading edge published so far; the other 9 are coalesced into a pending trailing + expect(handler).toHaveBeenCalledTimes(1); + + vi.advanceTimersByTime(THROTTLE); // single trailing notification + expect(handler).toHaveBeenCalledTimes(2); + expect(ids(p)).toEqual([ + 'm1', + 'm2', + 'm3', + 'm4', + 'm5', + 'm6', + 'm7', + 'm8', + 'm9', + 'm10', + 'm11', + 'm12', + ]); + }); + + it('flushState() bypasses the throttle for an optimistic write', () => { + const p = make(); + seed(p); + p.ingestItem(msg('m3', 3)); // leading + p.ingestItem(msg('m4', 4)); // deferred (trailing pending) + expect(ids(p)).toEqual(['m1', 'm2', 'm3']); + + p.flushState(); // optimistic bypass -> emit now + expect(ids(p)).toEqual(['m1', 'm2', 'm3', 'm4']); + + vi.advanceTimersByTime(THROTTLE); // no duplicate / extra emission + expect(ids(p)).toEqual(['m1', 'm2', 'm3', 'm4']); + }); + + it('an in-window content change (reaction/edit) coalesces via onMessagesChanged', () => { + const p = make(); + seed(p); + // open the window with a leading ingest so subsequent activity is deferred + p.ingestItem(msg('m3', 3)); + expect(ids(p)).toEqual(['m1', 'm2', 'm3']); + + // simulate a content change on a held id (what MessageStore.onMessagesChanged delivers) + const changed = { ...msg('m3', 3), text: 'edited' } as LocalMessage; + // update the backing index in place, then notify + p.getItem('m3'); // sanity: it is held + ( + p as unknown as { _itemIndex: { setOne: (m: LocalMessage) => void } } + )._itemIndex.setOne(changed); + p.onMessagesChanged({ changedIds: new Set(['m3']) }); + // deferred: the visible text is refreshed only on the trailing edge / flush + p.flushState(); + expect(p.items?.find((m) => m.id === 'm3')?.text).toBe('edited'); + }); + + it('does not throttle when globally disabled (the test default): every ingest is immediate', () => { + setStateThrottlingEnabled(false); + const p = make(); + seed(p); + p.ingestItem(msg('m3', 3)); + p.ingestItem(msg('m4', 4)); + expect(ids(p)).toEqual(['m1', 'm2', 'm3', 'm4']); // no timer advance needed + }); +}); + +// End-to-end optimistic path: a real store-backed paginator (MessageStoreBackedItemIndex), driven through +// the actual optimistic wiring (applyReactionLocally → store.upsert + store.flushSubscribers() → +// holder.flushState() → throttle.flush()). This is the path the "your own sends/reactions appear +// instantly" guarantee rides on — distinct from calling paginator.flushState() directly above. +describe('MessagePaginator — optimistic (local-user) writes bypass the throttle (end-to-end)', () => { + let store: MessageStore; + let client: StreamChat; + let channel: Channel; + + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(0); + setStateThrottlingEnabled(true); + store = new MessageStore(); + client = { + messageStore: store, + user: { id: 'me' }, + } as unknown as StreamChat; + channel = { + cid: 'channel-id', + getReplies: vi.fn(), + query: vi.fn(), + // MessagePaginator.createItemIndex reads this to build a MessageStoreBackedItemIndex, so the paginator + // is a real subscriber of `store` (gets onMessagesChanged + flushState). + getClient: () => client, + } as unknown as Channel; + }); + + afterEach(() => { + setStateThrottlingEnabled(false); + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + const makeStoreBacked = () => + new MessagePaginator({ channel, paginatorOptions: { stateThrottleMs: THROTTLE } }); + + const seedAndOpenWindow = (p: MessagePaginator) => { + p.setItems({ + valueOrFactory: [msg('m1', 1), msg('m2', 2)], + isFirstPage: true, + isLastPage: true, + }); + p.ingestItem(msg('m3', 3)); // leading edge — opens the throttle window + expect(ids(p)).toEqual(['m1', 'm2', 'm3']); + }; + + const hasLike = (p: MessagePaginator, id: string) => + !!p.items?.find((m) => m.id === id)?.own_reactions?.some((r) => r.type === 'like'); + + it('a local reaction renders immediately — no timer advance (flushSubscribers → flushState → flush)', () => { + const p = makeStoreBacked(); + seedAndOpenWindow(p); + expect(hasLike(p, 'm3')).toBe(false); + + // The real optimistic entry point: upserts the reacted message (deferred by the throttle) and + // then flushSubscribers() to bypass it. + applyReactionLocally(client, { + messageId: 'm3', + reaction: { type: 'like' } as unknown as Reaction, + }); + + // Reflected WITHOUT advancing the throttle timer. + expect(hasLike(p, 'm3')).toBe(true); + }); + + it('a non-optimistic store change (no flush) stays throttled until the window closes', () => { + const p = makeStoreBacked(); + seedAndOpenWindow(p); + + const current = store.get('m3'); + if (!current) throw new Error('expected m3 to be held by the store'); + // Simulate a WS-driven change that does NOT go through the optimistic flush. + store.upsert({ ...current, text: 'from-server' }); + + // Deferred — not visible until the trailing edge. + expect(p.items?.find((m) => m.id === 'm3')?.text).not.toBe('from-server'); + vi.advanceTimersByTime(THROTTLE); + expect(p.items?.find((m) => m.id === 'm3')?.text).toBe('from-server'); + }); +}); + +// Interval views (anchoredHead / logicalHead / logicalTail) publish on their OWN throttle, so a +// sibling content update to an off-active-window view still lands within one interval — it does not +// depend on `state.items` ever publishing again. +describe('MessagePaginator — interval-view (anchoredHead) publish throttling', () => { + let channel: Channel; + + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(0); + setStateThrottlingEnabled(true); + channel = { + cid: 'channel-id', + getReplies: vi.fn(), + query: vi.fn(), + } as unknown as Channel; + }); + + afterEach(() => { + setStateThrottlingEnabled(false); + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + const make = () => + new MessagePaginator({ channel, paginatorOptions: { stateThrottleMs: THROTTLE } }); + + // Count anchoredHead publishes, ignoring the initial synchronous emit (as `useStateStore` would). + const trackAnchored = (p: MessagePaginator) => { + let fires = 0; + const unsub = p.intervalViews.subscribeWithSelector( + (s) => ({ items: s.anchoredHead }), + () => { + fires += 1; + }, + ); + fires = 0; + return { count: () => fires, unsub }; + }; + + // Simulate a sibling holder rewriting a held message's content in the shared store. + const editInIndex = (p: MessagePaginator, id: string, text: string) => { + const current = p.getItem(id); + if (!current) throw new Error(`expected ${id} to be held`); + ( + p as unknown as { _itemIndex: { setOne: (m: LocalMessage) => void } } + )._itemIndex.setOne({ ...current, text }); + }; + + const notify = (p: MessagePaginator, id: string) => + p.onMessagesChanged({ changedIds: new Set([id]) }); + + it('coalesces a burst of sibling content updates into leading + trailing anchoredHead publishes', () => { + const p = make(); + p.ingestPage({ + page: [msg('m1', 1), msg('m2', 2), msg('m3', 3)], + isHead: true, + isTail: false, + setActive: true, + }); + expect(p.anchoredHeadItems.map((m) => m.id)).toEqual(['m1', 'm2', 'm3']); + + const anchored = trackAnchored(p); + + editInIndex(p, 'm1', 'e1'); + notify(p, 'm1'); // leading edge -> immediate publish + editInIndex(p, 'm2', 'e2'); + notify(p, 'm2'); // within window -> deferred + editInIndex(p, 'm3', 'e3'); + notify(p, 'm3'); // within window -> deferred + expect(anchored.count()).toBe(1); // only the leading edge so far + + vi.advanceTimersByTime(THROTTLE); // trailing edge + expect(anchored.count()).toBe(2); // coalesced 3 -> 2, not one publish per event + expect(p.anchoredHeadItems.find((m) => m.id === 'm1')?.text).toBe('e1'); + expect(p.anchoredHeadItems.find((m) => m.id === 'm2')?.text).toBe('e2'); + expect(p.anchoredHeadItems.find((m) => m.id === 'm3')?.text).toBe('e3'); + + anchored.unsub(); + }); + + it('refreshes anchoredHead on its own throttle even when the active window is a different, quiet interval', () => { + const p = make(); + // Head page becomes the anchored head (and the active window). + p.ingestPage({ + page: [msg('h1', 10), msg('h2', 11)], + isHead: true, + isTail: false, + setActive: true, + }); + // Jump to an older, disjoint window and make it the active one. + p.ingestPage({ + page: [msg('o1', 1), msg('o2', 2)], + isHead: false, + isTail: true, + setActive: true, + }); + expect(ids(p)).toEqual(['o1', 'o2']); // active window = the old page + expect(p.anchoredHeadItems.map((m) => m.id)).toEqual(['h1', 'h2']); + + let stateFires = 0; + const stateUnsub = p.state.subscribeWithSelector( + (s) => ({ items: s.items }), + () => { + stateFires += 1; + }, + ); + stateFires = 0; + const anchored = trackAnchored(p); + + // A sibling content change to a HEAD message — NOT in the active (old) window. + editInIndex(p, 'h1', 'edited-in-head'); + notify(p, 'h1'); + + // anchoredHead refreshes via its own throttle's leading edge, without any timer advance... + expect(anchored.count()).toBe(1); + expect(p.anchoredHeadItems.find((m) => m.id === 'h1')?.text).toBe('edited-in-head'); + // ...while `state.items` (the quiet active window) never publishes — h1 isn't in it. + expect(stateFires).toBe(0); + expect(ids(p)).toEqual(['o1', 'o2']); + + // A single change fires only the leading edge — nothing stray on the trailing edge. + vi.advanceTimersByTime(THROTTLE); + expect(anchored.count()).toBe(1); + + stateUnsub(); + anchored.unsub(); + }); +}); diff --git a/test/unit/reactions.optimistic.test.ts b/test/unit/reactions.optimistic.test.ts new file mode 100644 index 000000000..673f9f0d9 --- /dev/null +++ b/test/unit/reactions.optimistic.test.ts @@ -0,0 +1,583 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { formatMessage, StreamChat, Thread } from '../../src'; +import type { + Channel, + Event, + MessageResponse, + ReactionAPIResponse, + ReactionGroupResponse, + ReactionResponse, +} from '../../src'; +import { generateUUIDv4 as uuidv4 } from '../../src/utils'; +import { MockOfflineDB } from './offline-support/MockOfflineDB'; +import { generateChannel } from './test-utils/generateChannel'; +import { generateMsg } from './test-utils/generateMessage'; + +const CURRENT_USER = { id: 'me' }; + +const connect = () => { + const client = new StreamChat('apiKey'); + client.user = CURRENT_USER; + // `userID` is now a read-only getter derived from `client.user`, so setting `client.user` above + // is sufficient; assigning `client.userID` directly throws. + return client; +}; + +const createChannel = (client: StreamChat) => { + const { channel: channelResponse } = generateChannel(); + const channel = client.channel(channelResponse.type, channelResponse.id); + channel.initialized = true; + return channel; +}; + +const enableOfflineDb = (client: StreamChat) => { + client.setOfflineDBApi(new MockOfflineDB({ client })); + const db = client.offlineDb as MockOfflineDB; + db.state.partialNext({ initialized: true }); + db.insertReaction.mockResolvedValue([]); + db.updateReaction.mockResolvedValue([]); + db.deleteReaction.mockResolvedValue([]); + return db; +}; + +const ownReaction = (type: string, messageId: string): ReactionResponse => ({ + created_at: '2020-01-01T00:00:00.000Z', + message_id: messageId, + type, + updated_at: '2020-01-01T00:00:00.000Z', + user: CURRENT_USER, + user_id: CURRENT_USER.id, +}); + +const buildMessage = ( + ownTypes: string[] = [], + overrides: Partial = {}, +) => { + const id = overrides.id ?? uuidv4(); + const reactions = ownTypes.map((type) => ownReaction(type, id)); + const reaction_groups = ownTypes.reduce>( + (groups, type) => { + groups[type] = { count: 1, sum_scores: 1 }; + return groups; + }, + {}, + ); + return generateMsg({ + id, + latest_reactions: reactions, + own_reactions: reactions, + reaction_groups, + ...overrides, + }); +}; + +const seed = (channel: Channel, message: MessageResponse) => + channel.messagePaginator.ingestPage({ + isHead: true, + isTail: true, + page: [formatMessage(message)], + setActive: true, + }); + +const ownReactionTypes = (paginator: Channel['messagePaginator'], id: string) => + (paginator.getItem(id)?.own_reactions ?? []).map((reaction) => reaction.type); + +const apiReactionResponse = (message: MessageResponse) => + ({ duration: '0.0ms', message, reaction: {} }) as unknown as ReactionAPIResponse; + +const networkError = () => new Error('network down'); + +const apiError = (code: number) => + Object.assign(new Error(`api-error-${code}`), { code, response: { data: {} } }); + +describe('optimistic reactions', () => { + let client: StreamChat; + let channel: Channel; + + beforeEach(() => { + client = connect(); + channel = createChannel(client); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + describe('apply', () => { + it('adds the reaction to own_reactions immediately, then reconciles to the server response', async () => { + const message = buildMessage(); + seed(channel, message); + const serverMessage = generateMsg({ + id: message.id, + reaction_groups: { love: { count: 9, sum_scores: 9 } }, + }); + const sendReaction = vi + .spyOn(channel, 'sendReaction') + .mockResolvedValue(apiReactionResponse(serverMessage)); + + const pending = channel.addReactionWithLocalUpdate({ + messageId: message.id, + reaction: { type: 'love' }, + }); + + expect(ownReactionTypes(channel.messagePaginator, message.id)).toContain('love'); + expect(sendReaction).toHaveBeenCalledWith({ + id: message.id, + reaction: { type: 'love' }, + }); + + await pending; + + expect( + channel.messagePaginator.getItem(message.id)?.reaction_groups?.love?.count, + ).toBe(9); + }); + + it('replaces the existing own reaction when enforce_unique is set', async () => { + const message = buildMessage(['like']); + seed(channel, message); + vi.spyOn(channel, 'sendReaction').mockResolvedValue( + apiReactionResponse(generateMsg({ id: message.id })), + ); + + const pending = channel.addReactionWithLocalUpdate({ + messageId: message.id, + reaction: { type: 'love' }, + options: { enforce_unique: true }, + }); + + expect(ownReactionTypes(channel.messagePaginator, message.id)).toEqual(['love']); + + await pending; + }); + + it('removes the own reaction immediately', async () => { + const message = buildMessage(['love']); + seed(channel, message); + vi.spyOn(channel, 'deleteReaction').mockResolvedValue( + apiReactionResponse(generateMsg({ id: message.id })), + ); + + const pending = channel.deleteReactionWithLocalUpdate({ + messageId: message.id, + type: 'love', + }); + + expect(ownReactionTypes(channel.messagePaginator, message.id)).toEqual([]); + + await pending; + }); + }); + + describe('revert on failure without offline support', () => { + it('reverts an added reaction', async () => { + const message = buildMessage(); + seed(channel, message); + vi.spyOn(channel, 'sendReaction').mockRejectedValue(networkError()); + + await expect( + channel.addReactionWithLocalUpdate({ + messageId: message.id, + reaction: { type: 'love' }, + }), + ).rejects.toThrow('network down'); + + expect(ownReactionTypes(channel.messagePaginator, message.id)).toEqual([]); + }); + + it('restores a removed reaction', async () => { + const message = buildMessage(['love']); + seed(channel, message); + vi.spyOn(channel, 'deleteReaction').mockRejectedValue(networkError()); + + await expect( + channel.deleteReactionWithLocalUpdate({ messageId: message.id, type: 'love' }), + ).rejects.toThrow('network down'); + + expect(ownReactionTypes(channel.messagePaginator, message.id)).toEqual(['love']); + }); + + it('restores the displaced reaction when an enforce_unique add fails', async () => { + const message = buildMessage(['like']); + seed(channel, message); + vi.spyOn(channel, 'sendReaction').mockRejectedValue(networkError()); + + await expect( + channel.addReactionWithLocalUpdate({ + messageId: message.id, + reaction: { type: 'love' }, + options: { enforce_unique: true }, + }), + ).rejects.toThrow('network down'); + + expect(ownReactionTypes(channel.messagePaginator, message.id)).toEqual(['like']); + }); + }); + + describe('keep-vs-revert with offline support', () => { + beforeEach(() => { + enableOfflineDb(client); + }); + + it('keeps the optimistic reaction on a network error (no response)', async () => { + const message = buildMessage(); + seed(channel, message); + vi.spyOn(channel, 'sendReaction').mockRejectedValue(networkError()); + + await expect( + channel.addReactionWithLocalUpdate({ + messageId: message.id, + reaction: { type: 'love' }, + }), + ).rejects.toThrow(); + + expect(ownReactionTypes(channel.messagePaginator, message.id)).toContain('love'); + }); + + it('keeps the optimistic reaction when the server responds with a retryable code', async () => { + const message = buildMessage(); + seed(channel, message); + vi.spyOn(channel, 'sendReaction').mockRejectedValue(apiError(9)); + + await expect( + channel.addReactionWithLocalUpdate({ + messageId: message.id, + reaction: { type: 'love' }, + }), + ).rejects.toThrow(); + + expect(ownReactionTypes(channel.messagePaginator, message.id)).toContain('love'); + }); + + it('reverts when the server responds with a non-retryable code', async () => { + const message = buildMessage(); + seed(channel, message); + vi.spyOn(channel, 'sendReaction').mockRejectedValue(apiError(4)); + + await expect( + channel.addReactionWithLocalUpdate({ + messageId: message.id, + reaction: { type: 'love' }, + }), + ).rejects.toThrow(); + + expect(ownReactionTypes(channel.messagePaginator, message.id)).toEqual([]); + }); + }); + + describe('offline DB persistence', () => { + it('writes the reaction row on add and deletes it on a terminal rollback', async () => { + const db = enableOfflineDb(client); + const message = buildMessage(); + seed(channel, message); + vi.spyOn(channel, 'sendReaction').mockRejectedValue(apiError(4)); + + await expect( + channel.addReactionWithLocalUpdate({ + messageId: message.id, + reaction: { type: 'love' }, + }), + ).rejects.toThrow(); + + expect(db.insertReaction).toHaveBeenCalledTimes(1); + expect(db.deleteReaction).toHaveBeenCalledTimes(1); + }); + + it('uses updateReaction for enforce_unique and restores the displaced row on rollback', async () => { + const db = enableOfflineDb(client); + const message = buildMessage(['like']); + seed(channel, message); + vi.spyOn(channel, 'sendReaction').mockRejectedValue(apiError(4)); + + await expect( + channel.addReactionWithLocalUpdate({ + messageId: message.id, + reaction: { type: 'love' }, + options: { enforce_unique: true }, + }), + ).rejects.toThrow(); + + expect(db.updateReaction).toHaveBeenCalledTimes(1); + expect(db.deleteReaction).toHaveBeenCalledTimes(1); + expect(db.insertReaction).toHaveBeenCalledTimes(1); + }); + + it('deletes the row on remove and re-inserts it on rollback', async () => { + const db = enableOfflineDb(client); + const message = buildMessage(['love']); + seed(channel, message); + vi.spyOn(channel, 'deleteReaction').mockRejectedValue(apiError(4)); + + await expect( + channel.deleteReactionWithLocalUpdate({ messageId: message.id, type: 'love' }), + ).rejects.toThrow(); + + expect(db.deleteReaction).toHaveBeenCalledTimes(1); + expect(db.insertReaction).toHaveBeenCalledTimes(1); + }); + }); + + describe('show_in_channel fan-out', () => { + const setupDualHomed = () => { + const parentId = uuidv4(); + const reply = generateMsg({ + cid: channel.cid, + parent_id: parentId, + show_in_channel: true, + }); + seed(channel, reply); + + const thread = new Thread({ + client, + channel, + parentMessage: generateMsg({ cid: channel.cid, id: parentId }), + }); + thread.messagePaginator.ingestPage({ + isHead: true, + isTail: true, + page: [formatMessage(reply)], + setActive: true, + }); + client.threads.state.next((current) => ({ + ...current, + threads: [thread, ...current.threads], + })); + + return { reply, thread }; + }; + + it('mirrors a channel-side reaction onto the thread copy', async () => { + const { reply, thread } = setupDualHomed(); + vi.spyOn(channel, 'sendReaction').mockResolvedValue( + apiReactionResponse(generateMsg({ id: reply.id })), + ); + + const pending = channel.addReactionWithLocalUpdate({ + messageId: reply.id, + reaction: { type: 'love' }, + }); + + expect(ownReactionTypes(channel.messagePaginator, reply.id)).toContain('love'); + expect(ownReactionTypes(thread.messagePaginator, reply.id)).toContain('love'); + + await pending; + }); + + it('mirrors a thread-side reaction onto the channel copy', async () => { + const { reply, thread } = setupDualHomed(); + vi.spyOn(channel, 'sendReaction').mockResolvedValue( + apiReactionResponse(generateMsg({ id: reply.id })), + ); + + const pending = thread.addReactionWithLocalUpdate({ + messageId: reply.id, + reaction: { type: 'love' }, + }); + + expect(ownReactionTypes(thread.messagePaginator, reply.id)).toContain('love'); + expect(ownReactionTypes(channel.messagePaginator, reply.id)).toContain('love'); + + await pending; + }); + + it('reverts both copies when the request fails', async () => { + const { reply, thread } = setupDualHomed(); + vi.spyOn(channel, 'sendReaction').mockRejectedValue(networkError()); + + const pending = channel.addReactionWithLocalUpdate({ + messageId: reply.id, + reaction: { type: 'love' }, + }); + + expect(ownReactionTypes(channel.messagePaginator, reply.id)).toContain('love'); + expect(ownReactionTypes(thread.messagePaginator, reply.id)).toContain('love'); + + await expect(pending).rejects.toThrow('network down'); + + expect(ownReactionTypes(channel.messagePaginator, reply.id)).toEqual([]); + expect(ownReactionTypes(thread.messagePaginator, reply.id)).toEqual([]); + }); + }); + + describe('thread parent message (own_reactions preservation)', () => { + const parentOwnReactionTypes = (thread: Thread) => + (thread.state.getLatestValue().parentMessage?.own_reactions ?? []).map( + (r) => r.type, + ); + + const setupParentThread = (ownTypes: string[] = []) => { + const parentId = uuidv4(); + const thread = new Thread({ + client, + channel, + parentMessage: buildMessage(ownTypes, { cid: channel.cid, id: parentId }), + }); + thread.registerSubscriptions(); + return { parentId, thread }; + }; + + it('keeps the user other own_reactions when reacting to the parent (the dropped-reactions bug)', () => { + const { parentId, thread } = setupParentThread(['like']); + + client.dispatchEvent({ + type: 'reaction.new', + message: generateMsg({ id: parentId, own_reactions: [] }), + reaction: ownReaction('love', parentId), + } as unknown as Event); + + expect(parentOwnReactionTypes(thread)).toEqual( + expect.arrayContaining(['like', 'love']), + ); + thread.unregisterSubscriptions(); + }); + + it('removes only the un-reacted type from the parent own_reactions', () => { + const { parentId, thread } = setupParentThread(['like', 'love']); + + client.dispatchEvent({ + type: 'reaction.deleted', + message: generateMsg({ id: parentId, own_reactions: [] }), + reaction: ownReaction('love', parentId), + } as unknown as Event); + + expect(parentOwnReactionTypes(thread)).toEqual(['like']); + thread.unregisterSubscriptions(); + }); + + it('preserves parent own_reactions across an edit (message.updated)', () => { + const { parentId, thread } = setupParentThread(['love']); + + client.dispatchEvent({ + type: 'message.updated', + message: generateMsg({ id: parentId, text: 'edited', own_reactions: [] }), + } as unknown as Event); + + const parent = thread.state.getLatestValue().parentMessage; + expect(parent?.text).toBe('edited'); + expect(parentOwnReactionTypes(thread)).toEqual(['love']); + thread.unregisterSubscriptions(); + }); + }); + + describe('thread parent message (optimistic reactions)', () => { + const parentState = (thread: Thread) => thread.state.getLatestValue().parentMessage; + const parentOwnTypes = (thread: Thread) => + (parentState(thread)?.own_reactions ?? []).map((r) => r.type); + + // `registerSubscriptions` seeds the parent into the client-global message store and subscribes + // `state.parentMessage` to it, so an update to that id anywhere reflects on the thread. + const setupParentThread = ( + ownTypes: string[] = [], + { seedInChannel = false }: { seedInChannel?: boolean } = {}, + ) => { + const parentId = uuidv4(); + const parentMessage = buildMessage(ownTypes, { cid: channel.cid, id: parentId }); + if (seedInChannel) seed(channel, parentMessage); + const thread = new Thread({ client, channel, parentMessage }); + thread.registerSubscriptions(); + return { parentId, thread }; + }; + + it('applies an added reaction to the parent immediately even when the channel has not loaded it', async () => { + const { parentId, thread } = setupParentThread(); + vi.spyOn(channel, 'sendReaction').mockResolvedValue( + apiReactionResponse(generateMsg({ id: parentId })), + ); + + const pending = channel.addReactionWithLocalUpdate({ + messageId: parentId, + reaction: { type: 'love' }, + }); + + expect(parentOwnTypes(thread)).toContain('love'); + expect(parentState(thread)?.reaction_groups?.love?.count).toBe(1); + // the channel never held the parent, so there is no channel-side copy to update + expect(channel.messagePaginator.getItem(parentId)).toBeUndefined(); + + await pending; + }); + + it('mirrors the reaction onto both the channel copy and the thread parent when both hold it', async () => { + const { parentId, thread } = setupParentThread([], { seedInChannel: true }); + vi.spyOn(channel, 'sendReaction').mockResolvedValue( + apiReactionResponse(generateMsg({ id: parentId })), + ); + + const pending = channel.addReactionWithLocalUpdate({ + messageId: parentId, + reaction: { type: 'love' }, + }); + + expect(ownReactionTypes(channel.messagePaginator, parentId)).toContain('love'); + expect(parentOwnTypes(thread)).toContain('love'); + + await pending; + }); + + it('preserves the user other own_reactions when adding to the parent', async () => { + const { parentId, thread } = setupParentThread(['like']); + vi.spyOn(channel, 'sendReaction').mockResolvedValue( + apiReactionResponse(generateMsg({ id: parentId })), + ); + + const pending = channel.addReactionWithLocalUpdate({ + messageId: parentId, + reaction: { type: 'love' }, + }); + + expect(parentOwnTypes(thread)).toEqual(expect.arrayContaining(['like', 'love'])); + + await pending; + }); + + it('replaces the existing own reaction on the parent when enforce_unique is set', async () => { + const { parentId, thread } = setupParentThread(['like']); + vi.spyOn(channel, 'sendReaction').mockResolvedValue( + apiReactionResponse(generateMsg({ id: parentId })), + ); + + const pending = channel.addReactionWithLocalUpdate({ + messageId: parentId, + reaction: { type: 'love' }, + options: { enforce_unique: true }, + }); + + expect(parentOwnTypes(thread)).toEqual(['love']); + + await pending; + }); + + it('removes the own reaction from the parent immediately', async () => { + const { parentId, thread } = setupParentThread(['love']); + vi.spyOn(channel, 'deleteReaction').mockResolvedValue( + apiReactionResponse(generateMsg({ id: parentId })), + ); + + const pending = channel.deleteReactionWithLocalUpdate({ + messageId: parentId, + type: 'love', + }); + + expect(parentOwnTypes(thread)).toEqual([]); + + await pending; + }); + + it('reverts the parent reaction on a terminal failure', async () => { + const { parentId, thread } = setupParentThread(); + vi.spyOn(channel, 'sendReaction').mockRejectedValue(networkError()); + + const pending = channel.addReactionWithLocalUpdate({ + messageId: parentId, + reaction: { type: 'love' }, + }); + + expect(parentOwnTypes(thread)).toContain('love'); + + await expect(pending).rejects.toThrow('network down'); + + expect(parentOwnTypes(thread)).toEqual([]); + expect(parentState(thread)?.reaction_groups?.love).toBeUndefined(); + }); + }); +}); diff --git a/test/unit/threads.test.ts b/test/unit/threads.test.ts index 728fc500e..06427b0a9 100644 --- a/test/unit/threads.test.ts +++ b/test/unit/threads.test.ts @@ -319,6 +319,9 @@ describe('Threads 2.0', () => { it('updates parent message and related top-level properties', () => { const thread = createTestThread(); + // `state.parentMessage` is a projection of the client-global message store; the update + // reflects through the store subscription registered here. + thread.registerSubscriptions(); const stateBefore = thread.state.getLatestValue(); expect(stateBefore.deletedAt).to.be.null; @@ -481,6 +484,75 @@ describe('Threads 2.0', () => { // Merging a partial newest window must not clear "load older". expect(paginatorState.hasMoreTail).to.be.true; }); + + it('refreshes the store parent so the projection is not clobbered by a stale copy', () => { + const thread = createTestThread(); + thread.registerSubscriptions(); + // registering seeds the store with the parent the thread currently holds + expect(client.messageStore.get(thread.id)?.text).to.equal( + parentMessageResponse.text, + ); + + // A fresh re-query (reload / reconnect) carries an edited parent. + const hydrationThread = createTestThread({ + parentMessageOverrides: { text: 'edited-on-server' }, + }); + thread.hydrateState(hydrationThread); + + // Both the projection AND the store's canonical copy are the hydrated parent (no + // divergence). Regression: hydrateState used to update only state.parentMessage, leaving + // the store on the pre-hydrate copy, so the next store write for this id would fan the + // stale parent back over the edit. + expect(thread.state.getLatestValue().parentMessage.text).to.equal( + 'edited-on-server', + ); + expect(client.messageStore.get(thread.id)?.text).to.equal('edited-on-server'); + }); + }); + + describe('unregisterSubscriptions', () => { + it('releases the reply paginator hold on the shared message store', () => { + const reply = makeReply(); + const thread = createTestThread(); + thread.registerSubscriptions(); + thread.upsertReplyLocally({ message: reply }); + + // the reply is held in the client-global store by the reply paginator + expect(client.messageStore.has(reply.id)).to.be.true; + + thread.unregisterSubscriptions(); + + // Regression: a removed thread used to keep its reply paginator linked as a store + // subscriber, pinning it (and its replies) forever. With no other holder the store GCs + // the reply and the paginator no longer holds it. + expect(client.messageStore.has(reply.id)).to.be.false; + expect(thread.messagePaginator.getItem(reply.id)).to.be.undefined; + }); + + // Behavioral characterization of the store's refcount/reclaim contract via the PUBLIC read + // API (messageStore.has) — a dual-homed message must survive one holder leaving and only be + // reclaimed when the LAST holder releases it. This pins the contract independently of how the + // store implements routing/refcount internally (so it survives a store rearchitecture). + it('keeps a message a sibling collection still holds after the thread is torn down (dual-home refcount)', () => { + const thread = createTestThread(); + // The channel list also holds the parent message (it is a normal channel message). ingestItem + // links it into the shared store under the channel paginator, regardless of rendering/filter. + channel.messagePaginator.ingestItem(formatMessage(parentMessageResponse)); + expect(client.messageStore.has(thread.id)).to.be.true; + + // Opening the thread adds a SECOND holder (its parent-message store subscription). + thread.registerSubscriptions(); + expect(client.messageStore.has(thread.id)).to.be.true; + + // Tearing down the thread releases ITS hold — but the channel still holds the message, so + // (unlike the pure-reply case above) it must NOT be reclaimed from the store. + thread.unregisterSubscriptions(); + expect(client.messageStore.has(thread.id)).to.be.true; + + // Only once the last holder (the channel) drops it is the canonical copy reclaimed. + channel.messagePaginator.removeItem({ id: thread.id }); + expect(client.messageStore.has(thread.id)).to.be.false; + }); }); describe('reload', () => { diff --git a/test/unit/utils.test.ts b/test/unit/utils.test.ts index f0f5f8472..ed112af41 100644 --- a/test/unit/utils.test.ts +++ b/test/unit/utils.test.ts @@ -13,7 +13,6 @@ import { channelTracksReadLocally, userHasReadReceipts, formatMessage, - throttle, generateChannelTempCid, shouldConsiderArchivedChannels, shouldConsiderPinnedChannels, @@ -26,12 +25,65 @@ import { uniqBy, runDetached, sleep, + computeOwnReactions, } from '../../src/utils'; -import type { ChannelFilters, ChannelOwnCapability, ChannelSort } from '../../src'; +import type { + ChannelFilters, + ChannelOwnCapability, + ChannelSort, + ReactionResponse, +} from '../../src'; import { StreamChat, Channel } from '../../src'; import { chatLoggerSystem } from '../../src/logger'; +describe('computeOwnReactions', () => { + const ME = 'me'; + const other = 'someone-else'; + const reaction = (type: string, userId: string): ReactionResponse => + ({ type, user_id: userId }) as ReactionResponse; + + it('adds the current user reaction, de-duped by type', () => { + expect( + computeOwnReactions({ current: [], reaction: reaction('love', ME), userId: ME }), + ).toEqual([reaction('love', ME)]); + }); + + it('enforceUnique replaces the current user existing reaction', () => { + const result = computeOwnReactions({ + current: [reaction('like', ME)], + reaction: reaction('love', ME), + userId: ME, + enforceUnique: true, + }); + expect(result).toEqual([reaction('love', ME)]); + }); + + it('removed drops the current user reaction of that type', () => { + expect( + computeOwnReactions({ + current: [reaction('love', ME)], + reaction: reaction('love', ME), + userId: ME, + removed: true, + }), + ).toEqual([]); + }); + + // Regression: a cross-user reaction.updated (enforceUnique is passed for every reaction.updated) + // must NOT touch the current user's own_reactions. It previously returned [] in this case, wiping + // the current user's reaction highlight until a refresh. + it('preserves current-user own_reactions on a cross-user enforceUnique reaction.updated', () => { + const result = computeOwnReactions({ + current: [reaction('love', ME)], + reaction: reaction('like', other), + userId: ME, + enforceUnique: true, + }); + expect(result).toEqual([reaction('love', ME)]); + }); +}); + describe('findIndexInSortedArray', () => { it('finds index in the middle of haystack (asc)', () => { const needle = 5; @@ -1134,81 +1186,3 @@ describe('userHasReadReceipts', () => { expect(userHasReadReceipts(makeClient(undefined))).toBe(true); }); }); - -describe('throttle', () => { - beforeEach(() => { - vi.useFakeTimers(); - }); - afterEach(() => { - vi.useRealTimers(); - }); - - it('fires a single (non-burst) call on the trailing edge when leading is false', () => { - const fn = vi.fn(); - const throttled = throttle(fn, 1000, { leading: false, trailing: true }); - - throttled('a'); - expect(fn).not.toHaveBeenCalled(); // leading:false so nothing on the leading edge - - vi.advanceTimersByTime(1000); - expect(fn).toHaveBeenCalledTimes(1); - expect(fn).toHaveBeenLastCalledWith('a'); // the lone call's args reach the trailing edge - }); - - it('collapses a burst to one trailing call with the last args when leading is false', () => { - const fn = vi.fn(); - const throttled = throttle(fn, 1000, { leading: false, trailing: true }); - - throttled('a'); - throttled('b'); - throttled('c'); - expect(fn).not.toHaveBeenCalled(); - - vi.advanceTimersByTime(1000); - expect(fn).toHaveBeenCalledTimes(1); - expect(fn).toHaveBeenLastCalledWith('c'); - }); - - it('fires on the leading edge and drops within-window calls when trailing is false', () => { - const fn = vi.fn(); - const throttled = throttle(fn, 1000); // defaults { leading: true, trailing: false } - - throttled('a'); - expect(fn).toHaveBeenCalledTimes(1); - expect(fn).toHaveBeenLastCalledWith('a'); - - throttled('b'); - throttled('c'); - expect(fn).toHaveBeenCalledTimes(1); // trailing: false so no trailing invocation - - vi.advanceTimersByTime(1000); - expect(fn).toHaveBeenCalledTimes(1); - }); - - it('fires on both leading and trailing edges when both are enabled', () => { - const fn = vi.fn(); - const throttled = throttle(fn, 1000, { leading: true, trailing: true }); - - throttled('a'); // leading edge - expect(fn).toHaveBeenCalledTimes(1); - expect(fn).toHaveBeenLastCalledWith('a'); - - throttled('b'); // captured for the trailing edge - expect(fn).toHaveBeenCalledTimes(1); - - vi.advanceTimersByTime(1000); - expect(fn).toHaveBeenCalledTimes(2); // trailing edge - expect(fn).toHaveBeenLastCalledWith('b'); - }); - - it('does not fire a duplicate trailing call for a solitary leading+trailing call', () => { - const fn = vi.fn(); - const throttled = throttle(fn, 1000, { leading: true, trailing: true }); - - throttled('a'); // leading only so no second call to schedule a trailing - expect(fn).toHaveBeenCalledTimes(1); - - vi.advanceTimersByTime(1000); - expect(fn).toHaveBeenCalledTimes(1); // no duplicate trailing - }); -}); diff --git a/test/unit/utils/throttle.test.ts b/test/unit/utils/throttle.test.ts new file mode 100644 index 000000000..110087509 --- /dev/null +++ b/test/unit/utils/throttle.test.ts @@ -0,0 +1,284 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { ThrottledCallback } from '../../../src/utils/throttling/throttle'; +import { throttle } from '../../../src/utils/throttling/throttle'; + +// Ported from the Feeds client throttle test suite (leading/trailing/cancel), plus a `flush()` +// block for the addition made in this SDK. + +const advance = (ms: number) => vi.advanceTimersByTime(ms); + +describe('throttle', () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(0); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it('cancelTimer discards a pending trailing so a later flush does not re-fire it', () => { + const spy = vi.fn(); + const t = throttle(spy as ThrottledCallback, 200, { leading: true, trailing: true }); + + t.throttledFn('a'); // leading edge fires 'a' + t.throttledFn('b'); // schedules a trailing invocation carrying 'b' + spy.mockClear(); + + t.cancelTimer(); // should discard the pending trailing entirely + t.flush(); // must be a no-op — the trailing was cancelled + + expect(spy).not.toHaveBeenCalled(); + }); + + it('leading:true, trailing:false (default): fires immediately, drops during window, fires again after window on next call', () => { + const spy = vi.fn(); + const t = throttle(spy as ThrottledCallback, 200).throttledFn; + + t('a'); + expect(spy).toHaveBeenCalledTimes(1); + expect(spy).toHaveBeenLastCalledWith('a'); + + t('b'); + expect(spy).toHaveBeenCalledTimes(1); + + advance(199); + t('c'); + expect(spy).toHaveBeenCalledTimes(1); + + advance(1); + t('d'); + expect(spy).toHaveBeenCalledTimes(2); + expect(spy).toHaveBeenLastCalledWith('d'); + }); + + it('leading:true, trailing:true: first call fires immediately; subsequent calls within window schedule one trailing with latest args', () => { + const spy = vi.fn(); + const t = throttle(spy as ThrottledCallback, 200, { trailing: true }).throttledFn; + + t('a'); + expect(spy).toHaveBeenCalledTimes(1); + expect(spy).toHaveBeenLastCalledWith('a'); + + advance(50); + t('b'); + advance(50); + t('c'); + advance(99); + expect(spy).toHaveBeenCalledTimes(1); + + advance(1); + expect(spy).toHaveBeenCalledTimes(2); + expect(spy).toHaveBeenLastCalledWith('c'); + }); + + it('leading:true, trailing:true: no double-invoke at boundary (new leading cancels pending trailing)', () => { + const spy = vi.fn(); + const t = throttle(spy as ThrottledCallback, 200, { trailing: true }).throttledFn; + + t('a'); + expect(spy).toHaveBeenCalledTimes(1); + + advance(190); + t('b'); + t('c'); + expect(spy).toHaveBeenCalledTimes(1); + + vi.setSystemTime(200); + t('d'); + expect(spy).toHaveBeenCalledTimes(2); + expect(spy).toHaveBeenLastCalledWith('d'); + + vi.runOnlyPendingTimers(); + expect(spy).toHaveBeenCalledTimes(2); + }); + + it('leading:true, trailing:true: single call does not later trigger trailing (guard against double with same args)', () => { + const spy = vi.fn(); + const t = throttle(spy as ThrottledCallback, 200, { trailing: true }).throttledFn; + + t('a'); + expect(spy).toHaveBeenCalledTimes(1); + + vi.runAllTimers(); + expect(spy).toHaveBeenCalledTimes(1); + }); + + it('leading:false, trailing:true: does not call immediately; calls once at end of window with latest args', () => { + const spy = vi.fn(); + const t = throttle(spy as ThrottledCallback, 200, { + leading: false, + trailing: true, + }).throttledFn; + + t('a'); + expect(spy).toHaveBeenCalledTimes(0); + + advance(50); + t('b'); + expect(spy).toHaveBeenCalledTimes(0); + + advance(150); + expect(spy).toHaveBeenCalledTimes(1); + expect(spy).toHaveBeenLastCalledWith('b'); + + advance(50); + t('c'); + expect(spy).toHaveBeenCalledTimes(1); + advance(99); + expect(spy).toHaveBeenCalledTimes(1); + advance(51); + expect(spy).toHaveBeenCalledTimes(2); + expect(spy).toHaveBeenLastCalledWith('c'); + }); + + it('leading:false, trailing:false: never calls', () => { + const spy = vi.fn(); + const t = throttle(spy as ThrottledCallback, 200, { + leading: false, + trailing: false, + }).throttledFn; + + t('a'); + t('b'); + advance(1000); + expect(spy).toHaveBeenCalledTimes(0); + }); + + it('preserves `this` on trailing (apply)', () => { + const seen: unknown[][] = []; + const obj = { + x: 42, + fn(this: unknown, v: string) { + seen.push([this, v]); + }, + }; + const throttled = throttle(obj.fn, 200, { + leading: false, + trailing: true, + }).throttledFn; + + (obj as unknown as { call: typeof throttled }).call = throttled; + (obj as unknown as { call: typeof throttled }).call('hello'); + advance(200); + + expect(seen.length).toBe(1); + expect(seen[0][0]).toBe(obj); + expect(seen[0][1]).toBe('hello'); + }); + + it('schedules trailing for the exact remaining time, not the full timeout', () => { + const spy = vi.fn(); + const t = throttle(spy as ThrottledCallback, 200, { trailing: true }).throttledFn; + + t('a'); + advance(50); + t('b'); + + advance(149); + expect(spy).toHaveBeenCalledTimes(1); + + advance(1); + expect(spy).toHaveBeenCalledTimes(2); + expect(spy).toHaveBeenLastCalledWith('b'); + }); + + it('multiple calls in a burst within a window still produce at most one trailing (latest args)', () => { + const spy = vi.fn(); + const t = throttle(spy as ThrottledCallback, 200, { trailing: true }).throttledFn; + + t(1); + for (let i = 2; i <= 10; i++) t(i); + expect(spy).toHaveBeenCalledTimes(1); + + advance(200); + expect(spy).toHaveBeenCalledTimes(2); + expect(spy).toHaveBeenLastCalledWith(10); + }); + + it('does not leak extra invocations after long idle periods', () => { + const spy = vi.fn(); + const t = throttle(spy as ThrottledCallback, 200, { trailing: true }).throttledFn; + + t('a'); + advance(180); + t('b'); + + advance(20); + expect(spy).toHaveBeenCalledTimes(2); + + advance(10000); + expect(spy).toHaveBeenCalledTimes(2); + }); + + it('should cancel the timer when cancelTimer is called', () => { + const spy = vi.fn(); + const { throttledFn: t, cancelTimer: cancel } = throttle( + spy as ThrottledCallback, + 200, + { trailing: true }, + ); + + t('a'); + t('b'); + cancel(); + + vi.runOnlyPendingTimers(); + expect(spy).toHaveBeenCalledTimes(1); + }); + + describe('flush()', () => { + it('fires a pending trailing immediately with the latest args', () => { + const spy = vi.fn(); + const { throttledFn: t, flush } = throttle(spy as ThrottledCallback, 200, { + trailing: true, + }); + + t('a'); // leading + expect(spy).toHaveBeenCalledTimes(1); + advance(50); + t('b'); // trailing scheduled + t('c'); + expect(spy).toHaveBeenCalledTimes(1); + + flush(); + expect(spy).toHaveBeenCalledTimes(2); + expect(spy).toHaveBeenLastCalledWith('c'); + + // nothing lingering after the flush + vi.runOnlyPendingTimers(); + expect(spy).toHaveBeenCalledTimes(2); + }); + + it('is a no-op when nothing is pending', () => { + const spy = vi.fn(); + const { throttledFn: t, flush } = throttle(spy as ThrottledCallback, 200, { + trailing: true, + }); + + t('a'); // leading; no trailing queued + expect(spy).toHaveBeenCalledTimes(1); + flush(); + expect(spy).toHaveBeenCalledTimes(1); + }); + + it('the flushed invocation resets the window (a later call leads again)', () => { + const spy = vi.fn(); + const { throttledFn: t, flush } = throttle(spy as ThrottledCallback, 200, { + trailing: true, + }); + + t('a'); // leading @0 + advance(10); + t('b'); // trailing scheduled + flush(); // fire 'b' now @10 + expect(spy).toHaveBeenCalledTimes(2); + + advance(200); // well past the window since the flush + t('c'); + expect(spy).toHaveBeenCalledTimes(3); + expect(spy).toHaveBeenLastCalledWith('c'); + }); + }); +});