diff --git a/src/channel.ts b/src/channel.ts index 0845a4bf69..5e89e83a14 100644 --- a/src/channel.ts +++ b/src/channel.ts @@ -1,4 +1,4 @@ -import { ChannelState } from './channel_state'; +import { ChannelState, ChannelWatchStatus } from './channel_state'; import { CooldownTimer } from './CooldownTimer'; import { isEphemeral } from './errors'; import { applyReactionLocally } from './entityStore'; @@ -53,6 +53,7 @@ import type { UpdateMessageOptions, UserResponse, } from './types'; +import { AIStates } from './types'; import { StateStore } from './store'; import type { ChannelMemberRequest as Gen_ChannelMemberRequest, @@ -145,27 +146,13 @@ export class Channel extends ChannelApi { /** */ listeners: Map>; state: ChannelState; - /** - * This boolean is a vague indication of whether the channel exists on chat backend. - * - * If the value is true, then that means the channel has been initialized by either calling - * channel.create() or channel.query() or channel.watch(). - * - * If the value is false, then channel may or may not exist on the backend. The only way to ensure - * is by calling channel.create() or channel.query() or channel.watch(). - */ - initialized: boolean; - /** - * Indicates whether channel has been initialized by manually populating the state with some messages, members etc. - * Static state indicates that channel exists on backend, but is not being watched yet. - */ - offlineMode: boolean; lastKeyStroke?: Date; lastTypingEvent: Date | null; isTyping: boolean; - disconnected: boolean; /** Re-entrancy guard for {@link Channel.reload} (mirrors Thread.reload's isLoading guard). */ private _reloading = false; + /** Refcount backing the reactive `active` flag (a shared Channel instance can have several consumers). */ + private _activeRefCount = 0; push_preferences?: Gen_ChannelPushPreferencesResponse; public readonly configState = new StateStore({}); public readonly messageComposer: MessageComposer; @@ -211,11 +198,8 @@ export class Channel extends ChannelApi { this.listeners = new Map(); // perhaps the state variable should be private this.state = new ChannelState(this); - this.initialized = false; - this.offlineMode = false; this.lastTypingEvent = null; this.isTyping = false; - this.disconnected = false; this.messageComposer = new MessageComposer({ client: this._client, @@ -297,16 +281,24 @@ export class Channel extends ChannelApi { }, }, }); + + // Seed the reactive mute state from the client's current `mutedChannels` (a channel created + // after connect may already be muted). Kept in sync afterwards by the client fan-out on + // `notification.channel_mutes_updated` / `health.check`. + this._syncMuteStatus(); } /** - * Returns the chat client for this channel. Throws if `client.disconnect()` was called. + * Returns the chat client for this channel. Throws if the channel is pending disposal — see + * {@link Channel.pendingDisposal}. * * @returns The chat client. */ getClient(): StreamChat { - if (this.disconnected === true) { - throw Error(`You can't use a channel after client.disconnect() was called`); + if (this.pendingDisposal) { + throw Error( + `Channel ${this.cid} is pending disposal and cannot be used. Get a fresh instance via client.channel().`, + ); } return this._client; } @@ -690,7 +682,7 @@ export class Channel extends ChannelApi { const previousData = this.data; const data = await super.update(...args); this.data = data.channel; - this._syncStateFromChannelData(this.data, previousData); + this.state.syncStateFromChannelData(this.data, previousData); return data; } @@ -720,7 +712,7 @@ export class Channel extends ChannelApi { const previousData = this.data; this.data = channel; - this._syncStateFromChannelData(this.data, previousData); + this.state.syncStateFromChannelData(this.data, previousData); // If the capabiltities are changed, we trigger the `capabilities.changed` event. if (capabilitiesChanged) { this.getClient().dispatchEvent({ @@ -1106,6 +1098,27 @@ export class Channel extends ChannelApi { return this.getClient()._muteStatus(this.cid); } + /** + * Recomputes this channel's reactive `state.muteStatus` from the client's current `mutedChannels` + * and publishes it only when it actually changed — so the frequent `health.check` fan-out does not + * churn subscribers. Called from the constructor and by the client whenever `mutedChannels` + * updates. Unlike `muteStatus()`, this does not require the channel to be initialized. + */ + _syncMuteStatus() { + if (this.pendingDisposal) return; + + const next = this.getClient()._muteStatus(this.cid); + const previous = this.state.getLatestValue().muteStatus; + const unchanged = + previous.muted === next.muted && + (previous.createdAt?.getTime() ?? null) === (next.createdAt?.getTime() ?? null) && + (previous.expiresAt?.getTime() ?? null) === (next.expiresAt?.getTime() ?? null); + + if (unchanged) return; + + this.state.partialNext({ muteStatus: next }); + } + sendAction( messageId: string, formData: Record, @@ -1307,6 +1320,100 @@ export class Channel extends ChannelApi { return await super.markRead(...args); } + /** + * A vague indication of whether the channel exists on the chat backend — `true` once + * `create()`/`query()`/`watch()` has run. Store-backed and reactive: subscribe via + * `useStateStore(channel.state, (s) => ({ initialized: s.initialized }))`. + */ + get initialized() { + return this.state.getLatestValue().initialized; + } + + set initialized(initialized: boolean) { + this.state.partialNext({ initialized }); + } + + /** + * Whether the channel was initialized by manually populating its state (offline hydration) rather + * than a live watch. Store-backed and reactive. + */ + get offlineMode() { + return this.state.getLatestValue().offlineMode; + } + + set offlineMode(offlineMode: boolean) { + this.state.partialNext({ offlineMode }); + } + + /** + * Whether the channel has been torn down and is awaiting disposal (deleted, the current user + * removed, or the client disconnected). Store-backed and reactive. + * + * One-way and terminal — there is no counterpart that revives the instance. Its resources are + * already released ({@link Channel._disconnect} disposes the paginators and unregisters the + * subscriptions) and the client drops it from `activeChannels` right after, so nothing should + * touch it: it is skipped by the `client.activeChannels` lookups (so `client.channel(…)` mints a + * fresh instance), never re-watched on recovery, refused as a source of `channel.data` by the + * offline DB, and `getClient()` throws on it so a reference held across a `disconnectUser()` + * fails loudly instead of quietly requesting on a client with no user. + */ + get pendingDisposal() { + return this.state.getLatestValue().pendingDisposal; + } + + set pendingDisposal(pendingDisposal: boolean) { + this.state.partialNext({ pendingDisposal }); + } + + /** + * Whether this client holds a server-side watch on the channel, and if not, whether it should be + * restored — see {@link ChannelWatchStatus}. Store-backed and reactive: subscribe via + * `useStateStore(channel.state, (s) => ({ watchStatus: s.watchStatus }))`. + */ + get watchStatus() { + return this.state.getLatestValue().watchStatus; + } + + set watchStatus(watchStatus: ChannelWatchStatus) { + this.state.partialNext({ watchStatus }); + } + + /** + * Whether a consumer has declared this channel as the one it is currently consuming (see + * {@link Channel.activate}). Reactive — subscribe via + * `useStateStore(channel.state, (s) => ({ active: s.active }))`. + */ + get active() { + return this.state.getLatestValue().active; + } + + /** + * Declares that a consumer is now consuming this channel's own state (mirrors + * `thread.activate()`). Refcounted, as a single `Channel` instance can be held by several + * consumers at once, so it stays active until the last one deactivates. + * + * While active, the channel's own state takes precedence over bulk state writes: channel-list + * hydration does not re-seed its message list (its own `channel.reload()` owns that window). + */ + activate = () => { + this._activeRefCount += 1; + if (this._activeRefCount === 1) { + this.state.partialNext({ active: true }); + } + }; + + /** + * Declares that a consumer has stopped consuming this channel (mirrors `thread.deactivate()`). + * Only flips `active` back to `false` once the last holder deactivates. + */ + deactivate = () => { + if (this._activeRefCount === 0) return; + this._activeRefCount -= 1; + if (this._activeRefCount === 0) { + this.state.partialNext({ active: false }); + } + }; + /** * Marks the channel as unread from `messageId`. Only works when the `read_events` setting is enabled. * @@ -1368,6 +1475,15 @@ export class Channel extends ChannelApi { } this.state.clean(); + + // Clear a stuck AI indicator when we're offline (we'd miss the ending clear/stop). The cleaning + // interval keeps ticking through transient/internet drops; closeConnection stops it, and that + // path clears the indicator itself. Gate on health, not staleness — a healthy connection must + // never cut off a long-running response. + const client = this.getClient(); + if (!client.wsConnection?.isHealthy) { + this.state.resetAIState(); + } } /** @@ -1395,7 +1511,7 @@ export class Channel extends ChannelApi { this.initialized = true; const previousData = this.data; this.data = state.channel; - this._syncStateFromChannelData(this.data, previousData); + this.state.syncStateFromChannelData(this.data, previousData); // The message paginator is seeded synchronously inside query() (before read-state hydration), // so a channel opened via watch() alone — a deep-link restore, a search result, a freshly @@ -1458,6 +1574,9 @@ export class Channel extends ChannelApi { override async stopWatching(...args: Parameters) { const response = await super.stopWatching(...args); + // Deliberate: unlike a connection loss this must NOT be restored on reconnect. + this.watchStatus = ChannelWatchStatus.NotWatching; + logger.withExtraTags('stopWatching', this.cid).info('Stopped watching the channel.'); return response; @@ -1694,6 +1813,13 @@ export class Channel extends ChannelApi { ); } + // The request carrying `watch: true` came back, so the server has registered this connection as + // a watcher. Only ever set here because a `watch: false` query does NOT unwatch server-side, so it + // must not clear the flag. + if (queryPayload.watch) { + this.watchStatus = ChannelWatchStatus.Watching; + } + // Seed read/members/pinned/thread-cleanup state; the message list is in the paginator. this._initializeState(state); // The queried page is the latest set unless this was a jump/around query. @@ -1718,7 +1844,7 @@ export class Channel extends ChannelApi { .join(); const previousData = this.data; this.data = channel; - this._syncStateFromChannelData(this.data, previousData); + this.state.syncStateFromChannelData(this.data, previousData); this.offlineMode = false; this.cooldownTimer.refresh(); @@ -1980,16 +2106,16 @@ export class Channel extends ChannelApi { let hasStateChanged = false; this.messageReceiptsTracker.setPendingReadStoreReconcileMeta(reconcileMeta); - this.state.readStore.next((currentReadStoreState) => { - const nextReadState = patch(currentReadStoreState.read); + this.state.next((currentState) => { + const nextReadState = patch(currentState.read); - if (nextReadState === currentReadStoreState.read) { - return currentReadStoreState; + if (nextReadState === currentState.read) { + return currentState; } hasStateChanged = true; return { - ...currentReadStoreState, + ...currentState, read: nextReadState, }; }); @@ -2023,6 +2149,31 @@ export class Channel extends ChannelApi { return nextUserReadState; } + /** + * Sets the current user's unread count. The count lives in exactly one place — the reactive + * `read[userId].unread_messages`, which is what the unread badge reads and what + * `state.unreadCount` derives from — so channel-wide resets (`channel.truncated`, "all channels + * read") must route through here rather than writing a count of their own. + */ + _setOwnUnreadCount(unreadCount: number) { + if (this.pendingDisposal) return; + + const userId = this.getClient().userId; + if (!userId) return; + + const currentUserReadState = this.state.read[userId]; + // only reconcile an existing read entry; never fabricate one just to store a count + if (!currentUserReadState || currentUserReadState.unread_messages === unreadCount) { + return; + } + + this._upsertReadState( + userId, + () => ({ ...currentUserReadState, unread_messages: unreadCount }), + { changedUserIds: [userId] }, + ); + } + _handleChannelEvent(event: Event) { // eslint-disable-next-line @typescript-eslint/no-this-alias const channel = this; @@ -2042,6 +2193,17 @@ export class Channel extends ChannelApi { channelState.removeTypingEvent(event.user.id); } break; + case 'ai_indicator.update': + channelState.partialNext({ + aiState: (event.ai_state as AIState) ?? AIStates.Idle, + }); + break; + case 'ai_indicator.clear': + channelState.partialNext({ aiState: AIStates.Idle }); + break; + case 'ai_indicator.stop': + channelState.partialNext({ aiState: AIStates.Stop }); + break; // `message.read_locally` is the client-only event dispatched by `markReadLocally()` when read // events are disabled (e.g. livestreams with `isLocalUnreadCountEnabled`). It reuses the exact // `message.read` state logic so the read-state update lives in one place — only the @@ -2091,7 +2253,6 @@ export class Channel extends ChannelApi { const isOwnEvent = event.user?.id === client.user?.id; if (isOwnEvent) { - channelState.unreadCount = 0; // Delivery reporting buffers a `markChannelsDelivered` network request; the local read // must not hit the backend, so only sync for the real server `message.read`. if (event.type === 'message.read') { @@ -2221,15 +2382,26 @@ export class Channel extends ChannelApi { } if (preventUnreadCountUpdate) break; + // The own unread count IS `read[ownUserId].unread_messages` (see + // `ChannelState.unreadCount`), so the own row is what carries the own-unread gating: a + // message that does not count as unread (silent/shadowed/muted) never bumps it, and neither + // does one arriving while the user is viewing the latest messages — the SDK is about to mark + // it read itself, and the count/snapshot would flash until it does. The other users' rows + // are receipt bookkeeping and keep their blanket bump. + const countsAsOwnUnread = + !this.messagePaginator.isViewingLive && + this._countMessageAsUnread(event.message); + if (event.user?.id) { const eventUser = event.user; const eventUserId = eventUser.id; const createdAt = new Date(event.created_at ?? Date.now()); const eventMessageId = event.message.id; + const ownUserId = client.userId; this._patchReadState( (currentReadState) => { const userIds = Object.keys(currentReadState); - if (!userIds.length) return currentReadState; + if (!userIds.length && !countsAsOwnUnread) return currentReadState; const nextReadState = { ...currentReadState }; @@ -2242,6 +2414,9 @@ export class Channel extends ChannelApi { last_delivered_at: createdAt, last_delivered_message_id: eventMessageId, }; + } else if (userId === ownUserId && !countsAsOwnUnread) { + // does not count towards the own unread count — leave the row untouched + continue; } else { nextReadState[userId] = { ...currentReadState[userId], @@ -2251,22 +2426,26 @@ export class Channel extends ChannelApi { } } + // Seed the own row when the channel has none yet — an uninitialized channel (see + // regression #1732) or one queried with `state: false`, where `_initializeState` + // never ran to seed it. Without this the own count has nowhere to live and stops + // accumulating. `last_read` is epoch: nothing has been read, which is exactly how + // every "no last read" consumer already treats a missing value. + if (ownUserId && countsAsOwnUnread && !currentReadState[ownUserId]) { + nextReadState[ownUserId] = { + last_read: new Date(0), + unread_messages: 1, + user: (client.user ?? { id: ownUserId }) as UserResponse, + }; + } + return nextReadState; }, { changedUserIds: Object.keys(channelState.read) }, ); } - // Skip the own-unread bump when the user is actively viewing the latest messages (app - // foregrounded + newest message on screen). Without this, a message read in real time - // momentarily bumps `unreadCount`/the snapshot — the "N new" separator/banner + the - // channel-list badge would flash until the SDK's mark-read resets it. The SDK reports the - // viewing state via `messagePaginator.setViewingLive` and marks the message read itself. - // Only the OWN unread accounting is gated; the per-user read/receipt tracking above is - // intentionally left intact. - const isViewingLive = this.messagePaginator.isViewingLive; - if (!isViewingLive && this._countMessageAsUnread(event.message)) { - channelState.unreadCount = channelState.unreadCount + 1; + if (countsAsOwnUnread) { this.messagePaginator.setUnreadSnapshot({ unreadCount: channelState.unreadCount, }); @@ -2293,7 +2472,7 @@ export class Channel extends ChannelApi { if (event.channel?.truncated_at) { const truncatedAtDate = new Date(event.channel.truncated_at); - channelState.unreadCount = this.countUnread(truncatedAtDate); + this._setOwnUnreadCount(this.countUnread(truncatedAtDate)); // Partial truncation: keep messages newer than the cutoff. clearStateAndCache would wipe // the whole paginator (readers now source from it), so use the partial truncate. The // channel-wide read/unread context is reset by the truncation, so drop the unread snapshot @@ -2302,7 +2481,7 @@ export class Channel extends ChannelApi { this.messagePaginator.clearUnreadSnapshot(); this.pinnedMessagesPaginator.truncate({ truncatedAt: truncatedAtDate }); } else { - channelState.unreadCount = 0; + this._setOwnUnreadCount(0); this.messagePaginator.clearStateAndCache(); this.pinnedMessagesPaginator.clearStateAndCache(); } @@ -2385,8 +2564,6 @@ export class Channel extends ChannelApi { lastReadMessageId: channelState.read[event.user.id].last_read_message_id, unreadCount, }); - - channelState.unreadCount = unreadCount; break; } case 'channel.updated': @@ -2406,7 +2583,7 @@ export class Channel extends ChannelApi { event.channel?.own_capabilities ?? channel.data?.own_capabilities, }; channel.data = newChannelData; - channel._syncStateFromChannelData(channel.data, previousChannelData); + channel.state.syncStateFromChannelData(channel.data, previousChannelData); this.cooldownTimer.refresh(); } break; @@ -2470,7 +2647,7 @@ export class Channel extends ChannelApi { blocked: event.channel?.blocked ?? false, hidden: true, }; - channel._syncStateFromChannelData(channel.data, previousChannelData); + channel.state.syncStateFromChannelData(channel.data, previousChannelData); if (event.clear_history) { this.messagePaginator.clearStateAndCache(); this.pinnedMessagesPaginator.clearStateAndCache(); @@ -2485,7 +2662,7 @@ export class Channel extends ChannelApi { blocked: event.channel?.blocked ?? false, hidden: false, }; - channel._syncStateFromChannelData(channel.data, previousChannelData); + channel.state.syncStateFromChannelData(channel.data, previousChannelData); this.getClient().offlineDb?.handleChannelVisibilityEvent({ event }); break; } @@ -2534,14 +2711,6 @@ export class Channel extends ChannelApi { } } - _syncStateFromChannelData( - data: Channel['data'], - fallbackData: Channel['data'] = this.data, - ) { - this.state.syncOwnCapabilitiesFromChannelData(data, fallbackData); - this.state.syncMemberCountFromChannelData(data, fallbackData); - } - _initializeState(state: ChannelStateResponseFields) { const { state: clientState, user, userID } = this.getClient(); @@ -2616,10 +2785,6 @@ export class Channel extends ChannelApi { unread_messages: read.unread_messages ?? 0, user: read.user, }; - - if (read.user.id === user?.id) { - this.state.unreadCount = readUpdates[read.user.id].unread_messages; - } } } @@ -2694,12 +2859,16 @@ export class Channel extends ChannelApi { _disconnect() { logger.withExtraTags('_disconnect', this.cid).info('Disconnecting the channel.'); - this.disconnected = true; + // Tear down the channel.state subscriptions BEFORE flipping `pendingDisposal` — that setter + // now publishes to the store, so no subscriber handler runs against a half-torn-down channel. this.messageReceiptsTracker.unregisterSubscriptions(); + // A deleted channel (or one the user was removed from) must not be re-watched — see #2599. + this.watchStatus = ChannelWatchStatus.NotWatching; + this.pendingDisposal = true; 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. + // here (pending disposal + deleted from activeChannels, never reused), mirroring Thread teardown. this.messagePaginator.dispose(); this.pinnedMessagesPaginator.dispose(); } diff --git a/src/channel_state.ts b/src/channel_state.ts index 1d16d4479d..6da7c5caf7 100644 --- a/src/channel_state.ts +++ b/src/channel_state.ts @@ -1,5 +1,6 @@ import type { Channel } from './channel'; import type { + AIState, ChannelMemberResponse, Event, LocalMessage, @@ -7,6 +8,7 @@ import type { PendingMessageResponse, UserResponse, } from './types'; +import { AIStates } from './types'; import { formatMessage } from './utils'; import { StateStore } from './store'; @@ -23,9 +25,43 @@ type ChannelReadStatus = Record< } >; -export type WatcherState = { +/** + * Whether this client holds a server-side watch on the channel — i.e. whether channel events are + * being delivered to it. + * + * Three values rather than a boolean, because "not watching" alone cannot say whether a re-watch is + * wanted: a watch lost to a dead socket should be restored on reconnect, one the consumer stopped + * deliberately must not be. The server keys watches by connection ID, so a dropped socket ends every + * watch this client held — even if it reconnects moments later with a fresh ID. + */ +export const ChannelWatchStatus = { + /** Never watched, or the consumer stopped watching on purpose — do NOT re-watch. */ + NotWatching: 'notWatching', + /** Was watching until the connection dropped — the watch is gone and SHOULD be restored. */ + WasWatching: 'wasWatching', + /** Holding a live watch: events are flowing. */ + Watching: 'watching', +} as const; + +export type ChannelWatchStatus = + (typeof ChannelWatchStatus)[keyof typeof ChannelWatchStatus]; + +/** + * Everything about watching this channel: who else is watching it, and whether *we* are. + */ +export type ChannelWatchState = { watcherCount: number; watchers: Record; + /** + * See {@link ChannelWatchStatus}. Goes to `Watching` when a query carrying `watch: true` succeeds + * (`channel.watch()`, `channel.query({ watch: true })`, `client.queryChannels()`); to + * `WasWatching` when the WS connection is lost (only from `Watching` — a deliberate stop is never + * resurrected); and to `NotWatching` on `channel.stopWatching()` or teardown. + * + * Note `channel.watch()` silently downgrades to a non-watching query when the client has no + * connection ID; this is what makes that observable. + */ + watchStatus: ChannelWatchStatus; }; export type TypingUsersState = { @@ -36,163 +72,280 @@ export type ReadState = { read: ChannelReadStatus; }; -export type MutedUsersState = { - mutedUsers: Array; -}; - export type MembersState = { members: Record; + /** + * Total member count, sourced from `channel.data.member_count`. Set on query / watch / hydration + * (incl. offline-DB rehydration) and on `channel.updated` — deliberately NOT on `member.added` / + * `member.removed`. `channel.updated` fires on every member change and is the authoritative count; + * incrementing per member event double-counted when several members changed at once. + * See https://github.com/GetStream/stream-chat-js/pull/1761. + */ memberCount: number; }; +/** The current user's own membership in this channel (role, pinned_at, archived_at, …). */ +export type MembershipState = { + membership: ChannelMemberResponse; +}; + export type OwnCapabilitiesState = { ownCapabilities: string[]; }; /** - * ChannelState - A container class for the channel state. + * The channel's server-provided `data` (name, image, frozen, hidden, blocked, config, + * `member_count`, `own_capabilities`, …), mirrored reactively so consumers can subscribe to + * channel-level changes via `useStateStore(channel.state, (s) => ({ data: s.data }))`. + */ +export type ChannelDataState = { + data: Channel['data']; +}; + +/** Whether THIS channel is muted for the current user, mirrored from `client.mutedChannels`. */ +export type ChannelMuteStatus = { + muted: boolean; + createdAt: Date | null; + expiresAt: Date | null; +}; + +/** + * Reactive channel-mute state — is this channel muted for the current user. Mirrors the client-owned + * `client.mutedChannels` (updated on `notification.channel_mutes_updated` + `health.check`) and is + * subscribable via `useStateStore(channel.state, (s) => ({ muteStatus: s.muteStatus }))`. Muted + * USERS remain client-global on `client.mutedUsersStore` and are NOT part of channel state. + */ +export type MuteStatusState = { + muteStatus: ChannelMuteStatus; +}; + +/** + * Connection / initialization lifecycle flags for the channel. Previously plain fields on `Channel`; + * now store-backed so consumers can react to them via `useStateStore(channel.state, selector)`. + * Read/written through the `channel.initialized` / `channel.offlineMode` / + * `channel.pendingDisposal` getters/setters, which proxy this slice. + */ +export type ChannelLifecycleState = { + /** + * A vague indication of whether the channel exists on the chat backend. `true` once the channel + * has been initialized by `channel.create()` / `channel.query()` / `channel.watch()`. `false` + * means the channel may or may not exist — only those calls confirm it. + */ + initialized: boolean; + /** + * Whether the channel was initialized by manually populating its state (e.g. offline hydration) + * rather than a live watch. Such static state means the channel exists on the backend but is not + * being watched yet. + */ + offlineMode: boolean; + /** + * Whether the channel has been torn down and is awaiting disposal (deleted, the current user + * removed, or the client disconnected). One-way and terminal — see + * {@link Channel.pendingDisposal}: the instance is never revived. + */ + pendingDisposal: boolean; +}; + +/** + * Whether the channel is currently being consumed — consumer-declared, never returned by the API. + * Mirrors `thread.state.active`. + */ +export type ChannelActivationState = { + /** + * Whether a consumer has declared this channel as the one it is currently reading, via + * `channel.activate()` / `channel.deactivate()` (refcounted, as a single `Channel` instance can + * be held by several consumers at once). It carries no rendering semantics; it tells the client + * that the channel's own state is being consumed and takes precedence over bulk state writes — + * channel-list hydration does not re-seed the message list of an `active` channel (the channel's + * own `channel.reload()` owns that window). + */ + active: boolean; +}; + +/** + * Reactive AI-indicator state — driven by the `ai_indicator.update` / `.clear` / `.stop` events (see + * `Channel._handleChannelEvent`). Seeded to `AIStates.Idle` and subscribable via + * `useStateStore(channel.state, (s) => ({ aiState: s.aiState }))`. Reset to `Idle` on unwatch / + * disconnect (a live server sends `ai_indicator.clear` when the AI response starts streaming). + */ +export type AIIndicatorState = { + aiState: AIState; +}; + +/** + * The single, unified reactive state for a channel. All per-channel reactive state is published + * through this one `StateStore` and subscribed to with `useStateStore(channel.state, selector)` + * (mirroring `thread.state`). + * + * The shape is FLAT — subscribe to any slice via a selector, e.g. + * `useStateStore(channel.state, (s) => ({ read: s.read }))`. + */ +export type ChannelStateData = ChannelWatchState & + TypingUsersState & + ReadState & + MembersState & + MembershipState & + OwnCapabilitiesState & + ChannelDataState & + MuteStatusState & + ChannelLifecycleState & + ChannelActivationState & + AIIndicatorState; + +/** + * ChannelState - the container for a channel's reactive state. + * + * It IS a `StateStore` (so `useStateStore(channel.state, selector)` works, + * mirroring `thread.state`) while additionally exposing convenience getters/setters + * (`members`, `read`, `typing`, `watchers`, …) that read/write the same unified store. */ -export class ChannelState { +export class ChannelState extends StateStore { _channel: Channel; - readonly watcherStore: StateStore; - readonly typingStore: StateStore; - readonly readStore: StateStore; - readonly membersStore: StateStore; - readonly ownCapabilitiesStore: StateStore; - // todo: is this actually used somewhere? - readonly mutedUsersStore: StateStore; pending_messages: Array; - unreadCount: number; - membership: ChannelMemberResponse; constructor(channel: Channel) { - this._channel = channel; - this.watcherStore = new StateStore({ + super({ watcherCount: 0, watchers: {}, - }); - this.typingStore = new StateStore({ + watchStatus: ChannelWatchStatus.NotWatching, typing: {}, - }); - this.readStore = new StateStore({ read: {} }); - // a list of users to hide messages from - this.mutedUsersStore = new StateStore({ mutedUsers: [] }); - this.membersStore = new StateStore({ members: {}, memberCount: 0 }); - this.ownCapabilitiesStore = new StateStore({ + read: {}, + members: {}, + memberCount: 0, + membership: {} as ChannelMemberResponse, ownCapabilities: [], + data: channel?.data, + muteStatus: { muted: false, createdAt: null, expiresAt: null }, + initialized: false, + offlineMode: false, + pendingDisposal: false, + active: false, + aiState: AIStates.Idle, }); - this.syncMemberCountFromChannelData(channel?.data); - this.syncOwnCapabilitiesFromChannelData(channel?.data); + this._channel = channel; + this.syncStateFromChannelData(channel?.data); this.pending_messages = []; - this.membership = {} as ChannelMemberResponse; - this.unreadCount = 0; + } + + /** + * The current user's unread message count, derived from the `read` slice + * (`read[ownUserId].unread_messages`) rather than stored a second time — there is exactly one + * place holding this value, so the count the unread badge reads can never drift from the count + * `channel.countUnread()` returns. Written through `Channel._setOwnUnreadCount`. + */ + get unreadCount() { + // read `_client` directly: `getClient()` throws on a channel pending disposal, and reading a + // count off a dead instance should yield 0, not blow up. + const userId = this._channel?._client?.userId; + if (!userId) return 0; + return this.getLatestValue().read[userId]?.unread_messages ?? 0; + } + + /** The current user's own membership; store-backed so `useStateStore` can subscribe to it. */ + get membership() { + return this.getLatestValue().membership; + } + + set membership(membership: ChannelMemberResponse) { + this.partialNext({ membership }); } get members() { - return this.membersStore.getLatestValue().members; + return this.getLatestValue().members; } set members(members: Record) { - this.membersStore.partialNext({ members }); + this.partialNext({ members }); } get member_count() { - return this.membersStore.getLatestValue().memberCount; + return this.getLatestValue().memberCount; } set member_count(memberCount: number) { - this.membersStore.partialNext({ memberCount }); + this.partialNext({ memberCount }); } get read() { - return this.readStore.getLatestValue().read; + return this.getLatestValue().read; } set read(read: ChannelReadStatus) { - this.readStore.next({ read }); + this.partialNext({ read }); } get typing() { return ( - this._channel?.messageComposer?.textComposer.typing ?? - this.typingStore.getLatestValue().typing + this._channel?.messageComposer?.textComposer.typing ?? this.getLatestValue().typing ); } set typing(typing: Record) { - this.typingStore.next({ typing }); + this.partialNext({ typing }); if (this._channel?.messageComposer) { this._channel.messageComposer.textComposer.setTyping(typing); } } - syncMemberCountFromChannelData( + /** + * Reflects the channel's server-provided `data` into the unified store and derives the + * `memberCount` and `ownCapabilities` slices from it. + * + * `fallbackData` (the previous `channel.data`) makes both derived fields sticky: a data update + * that omits `member_count`/`own_capabilities` keeps the last known value rather than wiping it. + * The sticky value is written back onto `data` as a plain field (only when `data` itself is + * missing it) so raw readers — e.g. `channelHasReadEvents`, which inspects + * `channel.data.own_capabilities` directly — stay consistent with the store. `own_capabilities` + * is never coerced to `[]` while unknown, so "not yet loaded" is not mistaken for "explicitly no + * capabilities" (regression #1732). This replaces the previous `Object.defineProperty` machinery; + * direct in-place mutation of `channel.data.member_count`/`own_capabilities` no longer syncs to + * the store — reassign `channel.data` (as the WS handlers do) instead. + */ + syncStateFromChannelData( data: Channel['data'], fallbackData: Channel['data'] = this._channel?.data, ) { const fallbackMemberCount = typeof fallbackData?.member_count === 'number' ? fallbackData.member_count - : this.membersStore.getLatestValue().memberCount; - - if (!data || typeof data !== 'object') { - this.membersStore.partialNext({ memberCount: fallbackMemberCount ?? 0 }); - return; - } + : this.getLatestValue().memberCount; - const dataDescriptor = Object.getOwnPropertyDescriptor(data, 'member_count'); - let memberCount = - typeof data.member_count === 'number' + const memberCount = + typeof data?.member_count === 'number' ? data.member_count : typeof fallbackMemberCount === 'number' ? fallbackMemberCount : undefined; - this.membersStore.partialNext({ memberCount: memberCount ?? 0 }); - - Object.defineProperty(data, 'member_count', { - configurable: true, - enumerable: dataDescriptor?.enumerable ?? false, - get: () => memberCount, - set: (nextMemberCount: number | undefined) => { - memberCount = typeof nextMemberCount === 'number' ? nextMemberCount : undefined; - this.membersStore.partialNext({ memberCount: memberCount ?? 0 }); - }, - }); - } - - syncOwnCapabilitiesFromChannelData( - data: Channel['data'], - fallbackData: Channel['data'] = this._channel?.data, - ) { - if (!data || typeof data !== 'object') { - this.ownCapabilitiesStore.next({ ownCapabilities: [] }); - return; - } - - let ownCapabilities: string[] | undefined = Array.isArray(data.own_capabilities) + const ownCapabilities = Array.isArray(data?.own_capabilities) ? [...data.own_capabilities] : Array.isArray(fallbackData?.own_capabilities) ? [...fallbackData.own_capabilities] : undefined; - this.ownCapabilitiesStore.next({ ownCapabilities: ownCapabilities ?? [] }); - - // Keep the reactive getter/setter so backward-compatible assignments still sync to - // the store, but return `undefined` until capabilities are actually known. Forcing - // `[]` on an unloaded channel would make read-events–gated logic (e.g. unread - // counting, regression #1732) treat "not yet loaded" as "explicitly no capabilities". - Object.defineProperty(data, 'own_capabilities', { - configurable: true, - enumerable: true, - get: () => ownCapabilities, - set: (nextOwnCapabilities: string[] | undefined) => { - ownCapabilities = Array.isArray(nextOwnCapabilities) - ? [...nextOwnCapabilities] - : undefined; - this.ownCapabilitiesStore.next({ ownCapabilities: ownCapabilities ?? [] }); - }, + // Carry a genuinely-known previous value forward onto the new `data` object when the update + // omits it — never fabricate one (an empty channel keeps `data === {}`, its `own_capabilities` + // undefined). This is a plain assignment, not an accessor. + if (data && typeof data === 'object') { + if ( + typeof data.member_count !== 'number' && + typeof fallbackData?.member_count === 'number' + ) { + data.member_count = fallbackData.member_count; + } + if ( + !Array.isArray(data.own_capabilities) && + Array.isArray(fallbackData?.own_capabilities) + ) { + data.own_capabilities = [...fallbackData.own_capabilities]; + } + } + + this.partialNext({ + data, + memberCount: memberCount ?? 0, + ownCapabilities: ownCapabilities ?? [], }); } @@ -208,28 +361,20 @@ export class ChannelState { this.typing = typing; } - get mutedUsers() { - return this.mutedUsersStore.getLatestValue().mutedUsers; - } - - set mutedUsers(mutedUsers: Array) { - this.mutedUsersStore.next({ mutedUsers }); - } - get watchers() { - return this.watcherStore.getLatestValue().watchers; + return this.getLatestValue().watchers; } set watchers(watchers: Record) { - this.watcherStore.partialNext({ watchers }); + this.partialNext({ watchers }); } get watcher_count() { - return this.watcherStore.getLatestValue().watcherCount; + return this.getLatestValue().watcherCount; } set watcher_count(watcherCount: number) { - this.watcherStore.partialNext({ watcherCount }); + this.partialNext({ watcherCount }); } /** @@ -261,4 +406,14 @@ export class ChannelState { } } } + + /** + * Resets the AI indicator to `Idle`. Called when the connection drops, as while the connection is + * severed we miss the `ai_indicator.clear`/`.stop` that ends a response and it isn't replayed on + * reconnect, so the indicator would otherwise stay stuck on "Generating". Noop when already `Idle`. + */ + resetAIState() { + if (this.getLatestValue().aiState === AIStates.Idle) return; + this.partialNext({ aiState: AIStates.Idle }); + } } diff --git a/src/client.ts b/src/client.ts index b4a7a2afeb..d1f2cbf283 100644 --- a/src/client.ts +++ b/src/client.ts @@ -5,6 +5,7 @@ import type { AxiosInstance } from 'axios'; import axios from 'axios'; import { Channel } from './channel'; +import { ChannelWatchStatus } from './channel_state'; import { ClientState } from './client_state'; import { StableWSConnection } from './connection'; import { UploadManager } from './uploadManager'; @@ -508,6 +509,9 @@ export class StreamChat extends ChatApi { * successful disconnection. See https://developer.mozilla.org/en-US/docs/Web/API/CloseEvent (optional). */ closeConnection = async (timeout?: number) => { + this._resetAIStateOnActiveChannels(); + this._markActiveChannelsWatchInterrupted(); + if (this.cleaningIntervalRef != null) { clearInterval(this.cleaningIntervalRef); this.cleaningIntervalRef = undefined; @@ -1014,6 +1018,7 @@ export class StreamChat extends ChatApi { client.user = event.me; client.state.updateUser(event.me); client.mutedChannels = event.me.channel_mutes; + client._reflectMutedChannelsToActiveChannels(); client.mutedUsers = event.me.mutes; client.blockedUsers.partialNext({ userIds: event.me.blocked_user_ids ?? [] }); } @@ -1025,6 +1030,7 @@ export class StreamChat extends ChatApi { if (event.type === 'notification.channel_mutes_updated' && event.me?.channel_mutes) { this.mutedChannels = event.me.channel_mutes; + this._reflectMutedChannelsToActiveChannels(); } if (event.type === 'notification.mutes_updated' && event.me?.mutes) { @@ -1033,9 +1039,10 @@ export class StreamChat extends ChatApi { if (event.type === 'notification.mark_read' && event.unread_channels === 0) { const activeChannelKeys = Object.keys(this.activeChannels); - activeChannelKeys.forEach( - (activeChannelKey) => - (this.activeChannels[activeChannelKey].state.unreadCount = 0), + activeChannelKeys.forEach((activeChannelKey) => + // resets `read[userId].unread_messages`, which is what the unread badge reads, so it does + // not stay stale. + this.activeChannels[activeChannelKey]._setOwnUnreadCount(0), ); } @@ -1071,6 +1078,50 @@ export class StreamChat extends ChatApi { return postListenerCallbacks; } + /** + * Fans the client-owned `mutedChannels` out to every active channel's reactive `state.muteStatus`. + * Each channel republishes only when its own mute status actually changed, so this stays cheap on + * the frequent `health.check` path. + */ + _reflectMutedChannelsToActiveChannels() { + for (const cid in this.activeChannels) { + this.activeChannels[cid]?._syncMuteStatus(); + } + } + + /** + * Resets the AI indicator state to `Idle` on every active channel. Invoked from `closeConnection` + * as it's a deliberate shutdown and will not natively trigger a WS event. + */ + _resetAIStateOnActiveChannels() { + for (const cid in this.activeChannels) { + this.activeChannels[cid]?.state.resetAIState(); + } + } + + /** + * Demotes every actively-watched channel to `WasWatching`. The server keys watches by connection + * ID, so losing the socket ends every watch this client held — a reconnect issues a NEW id and the + * channels have to be re-queried to watch again. `WasWatching` is what records that they should be. + * + * Only `Watching` is demoted: a channel the consumer stopped on purpose, or one that was torn + * down, stays `NotWatching` and must not be resurrected by a reconnect. + * + * Invoked from two places, because neither covers the other: `StableWSConnection._setHealth(false)` + * for an abnormal close/error (immediately — NOT via the `connection.changed` event, which is + * 5s-debounced when going offline and is skipped entirely on a quick flap, both of which would + * leave the status lying), and `closeConnection()` for a deliberate shutdown (e.g. mobile + * backgrounding), which sets `isHealthy` directly and so never reaches `_setHealth`. + */ + _markActiveChannelsWatchInterrupted() { + for (const cid in this.activeChannels) { + const channel = this.activeChannels[cid]; + if (channel?.watchStatus === ChannelWatchStatus.Watching) { + channel.watchStatus = ChannelWatchStatus.WasWatching; + } + } + } + _muteStatus(cid: string) { let muteStatus; for (let i = 0; i < this.mutedChannels.length; i++) { @@ -1410,9 +1461,16 @@ export class StreamChat extends ChatApi { const c = this.channel(channelState.channel.type, channelState.channel.id); const previousData = c.data; c.data = channelState.channel; - c._syncStateFromChannelData(c.data, previousData); + c.state.syncStateFromChannelData(c.data, previousData); c.offlineMode = offlineMode; c.initialized = !offlineMode; + // Same precedence `queryChannels` applies to the request: an explicit caller choice wins, + // otherwise we watch only if there is a connection to watch on. Offline hydration populates + // state without a live watch, so it never counts - and a query that did not watch leaves the + // status untouched (it neither starts nor ends a watch). + if (!offlineMode && (queryChannelsOptions?.watch ?? this._hasConnectionID())) { + c.watchStatus = ChannelWatchStatus.Watching; + } c.push_preferences = channelState.push_preferences; const willInitialize = @@ -1432,8 +1490,12 @@ export class StreamChat extends ChatApi { // newest page to merge into that jumped interval across the gap (missing messages in the // middle). A cold paginator, or one still at the head (offline/at-latest), re-seeds normally so // cursors/hasMoreTail get (re)derived and pagination keeps working. + // Also skip the re-seed for an ACTIVE (on-screen) channel: its own `channel.reload()` owns the + // loaded window (up to 100 msgs), so a 25-msg list re-seed here is a redundant second update + // that could perturb the fuller window. (Inert until a UI SDK calls `channel.activate()`.) if ( willInitialize && + !c.active && (!c.messagePaginator.isInitialized || c.messagePaginator.isActiveIntervalAtHead) ) { c.messagePaginator.seedFirstPageSync( @@ -1580,7 +1642,7 @@ export class StreamChat extends ChatApi { // we will replace it with `cid` for (const key in this.activeChannels) { const channel = this.activeChannels[key]; - if (channel.disconnected) { + if (channel.pendingDisposal) { continue; } @@ -1634,7 +1696,7 @@ export class StreamChat extends ChatApi { if ( cid in this.activeChannels && this.activeChannels[cid] && - !this.activeChannels[cid].disconnected + !this.activeChannels[cid].pendingDisposal ) { const channel = this.activeChannels[cid]; // Only overwrite the existing channel's custom data when the caller actually provided some. @@ -1646,7 +1708,7 @@ export class StreamChat extends ChatApi { if (custom.custom !== undefined) { const previousData = channel.data; channel.data = { ...channel.data, custom: custom.custom }; - channel._syncStateFromChannelData(channel.data, previousData); + channel.state.syncStateFromChannelData(channel.data, previousData); channel._data = { ...channel._data, custom: custom.custom }; } return channel; diff --git a/src/connection.ts b/src/connection.ts index 9461a5f682..34c4d369bf 100644 --- a/src/connection.ts +++ b/src/connection.ts @@ -650,6 +650,10 @@ export class StableWSConnection { return; } + // The server keys channel watches by connection ID, so they are gone the moment the socket is. + // Done here rather than off the `connection.changed` event below, which is debounced by 5s. + this.client._markActiveChannelsWatchInterrupted(); + // we're offline, wait few seconds and fire and event if still offline setTimeout(() => { if (this.isHealthy) return; diff --git a/src/messageDelivery/MessageReceiptsTracker.ts b/src/messageDelivery/MessageReceiptsTracker.ts index c265051c6b..efb61b4c7c 100644 --- a/src/messageDelivery/MessageReceiptsTracker.ts +++ b/src/messageDelivery/MessageReceiptsTracker.ts @@ -196,14 +196,19 @@ export class MessageReceiptsTracker extends WithSubscriptions { if (this.hasSubscriptions) return; this.addUnsubscribeFunction( - this.channel.state.readStore.subscribe((next, prev) => { - this.reconcileFromReadStore({ - previousReadState: prev?.read, - nextReadState: next.read, - meta: this.pendingReadStoreReconcileMeta, - }); - this.pendingReadStoreReconcileMeta = undefined; - }), + // Subscribe to only the `read` slice of the unified channel state so this reconcile fires on + // read changes, not on every unrelated channel-state write (typing/members/…). + this.channel.state.subscribeWithSelector( + (currentState) => ({ read: currentState.read }), + (next, prev) => { + this.reconcileFromReadStore({ + previousReadState: prev?.read, + nextReadState: next.read, + meta: this.pendingReadStoreReconcileMeta, + }); + this.pendingReadStoreReconcileMeta = undefined; + }, + ), ); }; diff --git a/src/offline-support/offline_support_api.ts b/src/offline-support/offline_support_api.ts index 7a5e9434c5..89d441d67d 100644 --- a/src/offline-support/offline_support_api.ts +++ b/src/offline-support/offline_support_api.ts @@ -587,7 +587,7 @@ export abstract class AbstractOfflineDB implements OfflineDBApi { event_.channel_type, event_.channel_id, ); - if (channelFromState.initialized && !channelFromState.disconnected) { + if (channelFromState.initialized && !channelFromState.pendingDisposal) { channelData = channelFromState.data as unknown as ChannelResponse; } } diff --git a/src/pagination/paginators/MessagePaginator.ts b/src/pagination/paginators/MessagePaginator.ts index 5db0ace7ae..b9622a4c3d 100644 --- a/src/pagination/paginators/MessagePaginator.ts +++ b/src/pagination/paginators/MessagePaginator.ts @@ -328,9 +328,9 @@ export class MessagePaginator extends MessageIntervalPaginator { seedUnreadSnapshot = () => { // A paginator query (BasePaginator.executeQuery) awaits the network before running its // synchronous postQueryReconcile, which calls this on the first page. If the channel was - // disconnected while that request was in flight, reading the client below throws ("You can't - // use a channel after client.disconnect()"), so guard against that. - if (this.channel.disconnected) return; + // torn down while that request was in flight, reading the client below throws (the channel is + // pending disposal), so guard against that. + if (this.channel.pendingDisposal) return; const ownUserId = this.channel.getClient().user?.id; const ownReadState = ownUserId ? this.channel.state.read[ownUserId] : undefined; if (!ownReadState) return; diff --git a/src/types.ts b/src/types.ts index 7c8ce78092..ee81ca0edc 100644 --- a/src/types.ts +++ b/src/types.ts @@ -438,12 +438,23 @@ export type ModerationFlagOptions = Omit< >; export type AIState = + | 'AI_STATE_IDLE' | 'AI_STATE_ERROR' - | 'AI_STATE_CHECKING_SOURCES' + | 'AI_STATE_EXTERNAL_SOURCES' | 'AI_STATE_THINKING' | 'AI_STATE_GENERATING' + | 'AI_STATE_STOP' | (string & {}); +export const AIStates = { + Error: 'AI_STATE_ERROR', + ExternalSources: 'AI_STATE_EXTERNAL_SOURCES', + Generating: 'AI_STATE_GENERATING', + Idle: 'AI_STATE_IDLE', + Stop: 'AI_STATE_STOP', + Thinking: 'AI_STATE_THINKING', +} as const satisfies Record; + /** * An identifier containing information about the downstream SDK using stream-chat. It * is used to resolve the user agent. diff --git a/test/unit/channel.test.js b/test/unit/channel.test.js index 132cd20d85..8001cdcc28 100644 --- a/test/unit/channel.test.js +++ b/test/unit/channel.test.js @@ -7,7 +7,8 @@ import { getOrCreateChannelApi } from './test-utils/getOrCreateChannelApi'; import sinon from 'sinon'; import { mockChannelQueryResponse } from './test-utils/mockChannelQueryResponse'; -import { ChannelState, StreamChat } from '../../src'; +import { Channel, ChannelState, ChannelWatchStatus, StreamChat } from '../../src'; +import { StableWSConnection } from '../../src/connection'; import { DEFAULT_QUERY_CHANNEL_MESSAGE_LIST_PAGE_SIZE } from '../../src/constants'; import { MockOfflineDB } from './offline-support/MockOfflineDB'; import { formatMessage, generateUUIDv4 as uuidv4 } from '../../src/utils'; @@ -25,6 +26,21 @@ const seedLatestWindow = (channel, messages) => setActive: true, }); +// The own unread count is derived from `read[ownUserId].unread_messages` (there is no separate +// counter to assign), so seed it through the own read row. +const seedOwnUnreadCount = (channel, unread_messages) => { + const ownUser = channel.getClient().user; + channel.state.read = { + ...channel.state.read, + [ownUser.id]: { + last_read: new Date(0), + user: ownUser, + ...channel.state.read[ownUser.id], + unread_messages, + }, + }; +}; + describe('Channel count unread', function () { let lastRead; let ignoredMessages; @@ -119,9 +135,9 @@ describe('Channel count unread', function () { it('countUnread should return state.unreadCount without lastRead', function () { expect(channel.countUnread()).to.be.equal(channel.state.unreadCount); - channel.state.unreadCount = 10; + seedOwnUnreadCount(channel, 10); expect(channel.countUnread()).to.be.equal(10); - channel.state.unreadCount = 0; + seedOwnUnreadCount(channel, 0); }); it('countUnread should return correct count', function () { @@ -243,7 +259,6 @@ describe('Channel isViewingLive (unread bump gating)', function () { const channel = client.channel('messaging', 'live-mode-id'); channel.initialized = true; channel.data = { ...channel.data, own_capabilities: ['read-events'] }; - channel.state.unreadCount = 0; return { channel }; }; @@ -292,6 +307,348 @@ describe('Channel isViewingLive (unread bump gating)', function () { }); }); +describe('Channel watch status (channel.state.watchStatus)', function () { + const user = { id: 'user' }; + let client; + let channel; + + const mockQueryResponse = (channelResponse) => { + client.api.sendRequest = () => + Promise.resolve({ + body: getOrCreateChannelApi(channelResponse).response.data, + metadata: {}, + }); + }; + + beforeEach(() => { + client = new StreamChat('apiKey'); + client.user = user; + client.user = { id: user.id }; + client.wsPromise = Promise.resolve(); + // a connection id is what makes a watch possible at all + client._hasConnectionID = () => true; + client.connectionId = 'connection-id'; + channel = client.channel('messaging', 'watching-id'); + client.activeChannels[channel.cid] = channel; + mockQueryResponse(generateChannel({ channel: { id: 'watching-id' } })); + }); + + it('starts out NotWatching', () => { + expect(channel.watchStatus).to.equal(ChannelWatchStatus.NotWatching); + expect(channel.state.getLatestValue().watchStatus).to.equal( + ChannelWatchStatus.NotWatching, + ); + }); + + it('is Watching after watch() resolves', async () => { + await channel.watch(); + + expect(channel.watchStatus).to.equal(ChannelWatchStatus.Watching); + }); + + it('is reactive via the store', async () => { + const seen = []; + const unsubscribe = channel.state.subscribeWithSelector( + (s) => ({ watchStatus: s.watchStatus }), + ({ watchStatus }) => seen.push(watchStatus), + ); + + await channel.watch(); + unsubscribe(); + + expect(seen).to.eql([ChannelWatchStatus.NotWatching, ChannelWatchStatus.Watching]); + }); + + it('stays NotWatching for a query that does not ask to watch', async () => { + await channel.query({ watch: false }); + + expect(channel.watchStatus).to.equal(ChannelWatchStatus.NotWatching); + }); + + it('is NOT demoted by a later non-watching query (a watch:false query does not unwatch)', async () => { + await channel.watch(); + + await channel.query({ watch: false }); + + expect(channel.watchStatus).to.equal(ChannelWatchStatus.Watching); + }); + + it('stays NotWatching when watch() downgrades for lack of a connection id', async () => { + client._hasConnectionID = () => false; + + await channel.watch(); + + expect(channel.watchStatus).to.equal(ChannelWatchStatus.NotWatching); + }); + + it('goes to NotWatching on stopWatching() — a deliberate stop is not restored', async () => { + await channel.watch(); + + await channel.stopWatching(); + + expect(channel.watchStatus).to.equal(ChannelWatchStatus.NotWatching); + }); + + it('goes to NotWatching on teardown', async () => { + await channel.watch(); + + channel._disconnect(); + + expect(channel.state.getLatestValue().watchStatus).to.equal( + ChannelWatchStatus.NotWatching, + ); + }); + + it('demotes Watching to WasWatching when the connection is interrupted', async () => { + await channel.watch(); + + client._markActiveChannelsWatchInterrupted(); + + expect(channel.watchStatus).to.equal(ChannelWatchStatus.WasWatching); + }); + + it('leaves a deliberately-stopped channel NotWatching when the connection is interrupted', async () => { + await channel.watch(); + await channel.stopWatching(); + + client._markActiveChannelsWatchInterrupted(); + + // the whole point of the third state: a reconnect must not resurrect this watch + expect(channel.watchStatus).to.equal(ChannelWatchStatus.NotWatching); + }); + + it('leaves a never-watched channel NotWatching when the connection is interrupted', () => { + client._markActiveChannelsWatchInterrupted(); + + expect(channel.watchStatus).to.equal(ChannelWatchStatus.NotWatching); + }); + + it('does not demote WasWatching any further on a second interruption', async () => { + await channel.watch(); + client._markActiveChannelsWatchInterrupted(); + + client._markActiveChannelsWatchInterrupted(); + + expect(channel.watchStatus).to.equal(ChannelWatchStatus.WasWatching); + }); + + it('returns to Watching when a re-watch succeeds after an interruption', async () => { + await channel.watch(); + client._markActiveChannelsWatchInterrupted(); + expect(channel.watchStatus).to.equal(ChannelWatchStatus.WasWatching); + + await channel.watch(); + + expect(channel.watchStatus).to.equal(ChannelWatchStatus.Watching); + }); + + it('is demoted across every active channel at once', async () => { + await channel.watch(); + const other = client.channel('messaging', 'other-id'); + client.activeChannels[other.cid] = other; + other.watchStatus = ChannelWatchStatus.Watching; + + client._markActiveChannelsWatchInterrupted(); + + expect(channel.watchStatus).to.equal(ChannelWatchStatus.WasWatching); + expect(other.watchStatus).to.equal(ChannelWatchStatus.WasWatching); + }); + + it('is set by channel-list hydration, which watches by default', () => { + const response = generateChannel({ channel: { id: 'hydrated-id' } }); + + const [hydrated] = client.hydrateActiveChannels([response]); + + expect(hydrated.watchStatus).to.equal(ChannelWatchStatus.Watching); + }); + + it('is NOT set by hydration when the caller opted out of watching', () => { + const response = generateChannel({ channel: { id: 'unwatched-id' } }); + + const [hydrated] = client.hydrateActiveChannels([response], {}, { watch: false }); + + expect(hydrated.watchStatus).to.equal(ChannelWatchStatus.NotWatching); + }); + + it('leaves WasWatching intact when hydration does not watch', async () => { + await channel.watch(); + client._markActiveChannelsWatchInterrupted(); + const response = generateChannel({ channel: { id: 'watching-id' } }); + + client.hydrateActiveChannels([response], {}, { watch: false }); + + // a non-watching hydrate neither starts nor ends a watch + expect(channel.watchStatus).to.equal(ChannelWatchStatus.WasWatching); + }); + + it('is NOT set by hydration without a connection id', () => { + client._hasConnectionID = () => false; + const response = generateChannel({ channel: { id: 'no-connection-id' } }); + + const [hydrated] = client.hydrateActiveChannels([response]); + + expect(hydrated.watchStatus).to.equal(ChannelWatchStatus.NotWatching); + }); + + it('is NOT set by offline hydration (state without a live watch)', () => { + const response = generateChannel({ channel: { id: 'offline-id' } }); + + const [hydrated] = client.hydrateActiveChannels([response], { offlineMode: true }); + + expect(hydrated.watchStatus).to.equal(ChannelWatchStatus.NotWatching); + expect(hydrated.offlineMode).to.equal(true); + }); + + it('is demoted via closeConnection (deliberate shutdown never reaches _setHealth)', async () => { + await channel.watch(); + + await client.closeConnection(); + + expect(channel.watchStatus).to.equal(ChannelWatchStatus.WasWatching); + }); + + it('is demoted when the WS connection reports itself unhealthy', async () => { + await channel.watch(); + const sweep = vi.spyOn(client, '_markActiveChannelsWatchInterrupted'); + const connection = new StableWSConnection({ client }); + connection.isHealthy = true; + + connection._setHealth(false); + + expect(sweep).toHaveBeenCalledTimes(1); + }); + + it('skips channels pending disposal when sweeping', async () => { + await channel.watch(); + channel._disconnect(); + + expect(() => client._markActiveChannelsWatchInterrupted()).not.to.throw(); + expect(channel.state.getLatestValue().watchStatus).to.equal( + ChannelWatchStatus.NotWatching, + ); + }); +}); + +describe('Channel AI indicator state (channel.state.aiState)', function () { + const setupChannel = () => { + const client = new StreamChat('apiKey'); + client.user = { id: 'user' }; + const channel = client.channel('messaging', 'ai-state-id'); + channel.initialized = true; + return { channel }; + }; + + it('defaults to AI_STATE_IDLE', () => { + const { channel } = setupChannel(); + expect(channel.state.getLatestValue().aiState).to.equal('AI_STATE_IDLE'); + }); + + it('reflects ai_indicator.update into channel.state.aiState', () => { + const { channel } = setupChannel(); + channel._handleChannelEvent({ + type: 'ai_indicator.update', + ai_state: 'AI_STATE_GENERATING', + }); + expect(channel.state.getLatestValue().aiState).to.equal('AI_STATE_GENERATING'); + }); + + it('resets to Idle on ai_indicator.clear', () => { + const { channel } = setupChannel(); + channel._handleChannelEvent({ + type: 'ai_indicator.update', + ai_state: 'AI_STATE_THINKING', + }); + channel._handleChannelEvent({ type: 'ai_indicator.clear' }); + expect(channel.state.getLatestValue().aiState).to.equal('AI_STATE_IDLE'); + }); + + it('sets AI_STATE_STOP on ai_indicator.stop', () => { + const { channel } = setupChannel(); + channel._handleChannelEvent({ + type: 'ai_indicator.update', + ai_state: 'AI_STATE_GENERATING', + }); + channel._handleChannelEvent({ type: 'ai_indicator.stop' }); + expect(channel.state.getLatestValue().aiState).to.equal('AI_STATE_STOP'); + }); + + it('is reactive via subscribeWithSelector', () => { + const { channel } = setupChannel(); + const seen = []; + channel.state.subscribeWithSelector( + (s) => ({ aiState: s.aiState }), + ({ aiState }) => seen.push(aiState), + ); + channel._handleChannelEvent({ + type: 'ai_indicator.update', + ai_state: 'AI_STATE_THINKING', + }); + expect(seen).to.include('AI_STATE_THINKING'); + }); + + it('clean() resets aiState to Idle when the WS connection is down', () => { + const { channel } = setupChannel(); + channel.getClient().wsConnection = { isHealthy: false }; + channel._handleChannelEvent({ + type: 'ai_indicator.update', + ai_state: 'AI_STATE_GENERATING', + }); + expect(channel.state.getLatestValue().aiState).to.equal('AI_STATE_GENERATING'); + + channel.clean(); + + expect(channel.state.getLatestValue().aiState).to.equal('AI_STATE_IDLE'); + }); + + it('clean() leaves a live aiState untouched while the WS connection is healthy', () => { + const { channel } = setupChannel(); + channel.getClient().wsConnection = { isHealthy: true }; + channel._handleChannelEvent({ + type: 'ai_indicator.update', + ai_state: 'AI_STATE_GENERATING', + }); + + channel.clean(); + + expect(channel.state.getLatestValue().aiState).to.equal('AI_STATE_GENERATING'); + }); + + it('closeConnection resets aiState to Idle across active channels', async () => { + const { channel } = setupChannel(); + const client = channel.getClient(); + client.activeChannels[channel.cid] = channel; + channel._handleChannelEvent({ + type: 'ai_indicator.update', + ai_state: 'AI_STATE_GENERATING', + }); + expect(channel.state.getLatestValue().aiState).to.equal('AI_STATE_GENERATING'); + + await client.closeConnection(); + + expect(channel.state.getLatestValue().aiState).to.equal('AI_STATE_IDLE'); + }); + + it('resetAIState() is a no-op (no notify) when already Idle', () => { + const { channel } = setupChannel(); + const seen = []; + channel.state.subscribeWithSelector( + (s) => ({ aiState: s.aiState }), + ({ aiState }) => seen.push(aiState), + ); + seen.length = 0; // drop the initial subscribe emission + + channel.state.resetAIState(); + + expect(seen).to.have.length(0); + }); + + it('falls back to Idle when ai_indicator.update carries no ai_state', () => { + const { channel } = setupChannel(); + channel._handleChannelEvent({ type: 'ai_indicator.update' }); + expect(channel.state.getLatestValue().aiState).to.equal('AI_STATE_IDLE'); + }); +}); + describe('Channel localized unread count (isLocalUnreadCountEnabled)', function () { const user = { id: 'user' }; const otherUser = { id: 'other-user' }; @@ -320,7 +677,6 @@ describe('Channel localized unread count (isLocalUnreadCountEnabled)', function it('message.new increments the unread count with read events off when the flag is set', function () { const { channel } = setupChannel({ isLocalUnreadCountEnabled: true }); - channel.state.unreadCount = 0; channel._handleChannelEvent({ type: 'message.new', @@ -337,9 +693,25 @@ describe('Channel localized unread count (isLocalUnreadCountEnabled)', function expect(channel.countUnread()).to.be.equal(2); }); + it('seeds the own read row when a message counts as unread and the channel has none yet', function () { + const { channel } = setupChannel({ isLocalUnreadCountEnabled: true }); + expect(channel.state.read[user.id]).to.be.undefined; + + channel._handleChannelEvent({ + type: 'message.new', + user: otherUser, + message: generateMsg({ user: otherUser }), + }); + + // the count lives in the read row, so the row has to exist for it to be counted at all + expect(channel.state.read[user.id]).to.be.ok; + expect(channel.state.read[user.id].unread_messages).to.be.equal(1); + expect(channel.state.read[user.id].user.id).to.be.equal(user.id); + expect(channel.state.read[user.id].last_read.getTime()).to.be.equal(0); + }); + it('message.new does not increment the unread count with read events off when the flag is not set', function () { const { channel } = setupChannel({ isLocalUnreadCountEnabled: false }); - channel.state.unreadCount = 0; channel._handleChannelEvent({ type: 'message.new', @@ -357,7 +729,6 @@ describe('Channel localized unread count (isLocalUnreadCountEnabled)', function .mockResolvedValue({ body: {}, metadata: {} }); const lastMsg = generateMsg({ user: otherUser }); seedLatestWindow(channel, [lastMsg]); - channel.state.unreadCount = 5; channel.state.read[user.id] = { last_read: new Date('2020-01-01T00:00:00'), unread_messages: 5, @@ -411,7 +782,6 @@ describe('Channel localized unread count (isLocalUnreadCountEnabled)', function .mockResolvedValue({ body: {}, metadata: {} }); const lastMsg = generateMsg({ user: otherUser }); seedLatestWindow(channel, [lastMsg]); - channel.state.unreadCount = 3; delete channel.state.read[user.id]; channel.markReadLocally(); @@ -537,7 +907,7 @@ describe('Channel _handleChannelEvent', function () { describe('message.new', () => { it('message.new does not reset the unreadCount for current user messages', function () { - channel.state.unreadCount = 100; + seedOwnUnreadCount(channel, 100); channel._handleChannelEvent({ type: 'message.new', user, @@ -548,7 +918,7 @@ describe('Channel _handleChannelEvent', function () { }); it('message.new does not reset the unreadCount for own thread replies', function () { - channel.state.unreadCount = 100; + seedOwnUnreadCount(channel, 100); channel._handleChannelEvent({ type: 'message.new', user, @@ -563,7 +933,7 @@ describe('Channel _handleChannelEvent', function () { }); it('message.new does not reset the unreadCount for others thread replies', function () { - channel.state.unreadCount = 100; + seedOwnUnreadCount(channel, 100); channel._handleChannelEvent({ type: 'message.new', user: { id: 'id' }, @@ -606,7 +976,7 @@ describe('Channel _handleChannelEvent', function () { }); it('message.new increment unreadCount properly', function () { - channel.state.unreadCount = 20; + seedOwnUnreadCount(channel, 20); channel._handleChannelEvent({ type: 'message.new', user: { id: 'id' }, @@ -622,7 +992,7 @@ describe('Channel _handleChannelEvent', function () { }); it('message.new skip increment for silent/shadowed/muted messages', function () { - channel.state.unreadCount = 30; + seedOwnUnreadCount(channel, 30); channel._handleChannelEvent({ type: 'message.new', user: { id: 'id' }, @@ -958,6 +1328,27 @@ describe('Channel _handleChannelEvent', function () { expect(channel.messagePaginator.headItems.length).to.be.equal(0); }); + it('resets read[userId].unread_messages together with unreadCount so the badge does not go stale (#29)', function () { + const userId = client.user.id; + channel.state.read = { + [userId]: { + unread_messages: 5, + last_read: new Date('2021-01-01T00:00:00.000Z'), + user: { id: userId }, + }, + }; + + channel._handleChannelEvent({ + type: 'channel.truncated', + user: { id: userId }, + channel: {}, + }); + + expect(channel.state.unreadCount).to.equal(0); + // the badge reads read[userId].unread_messages — it must not stay at the stale 5 + expect(channel.state.read[userId].unread_messages).to.equal(0); + }); + it('message.truncate clears messagePaginator unread snapshot', function () { const cachedMessage = generateMsg({ date: '2020-01-01T00:00:00.000Z', @@ -1481,7 +1872,6 @@ describe('Channel _handleChannelEvent', function () { }); it('should update channel read state produced for current user', () => { - channel.state.unreadCount = initialCountUnread; channel.state.read[user.id] = initialReadState; const event = notificationMarkUnreadEvent; @@ -1530,7 +1920,6 @@ describe('Channel _handleChannelEvent', function () { }); it('should not update channel read state produced for another user or user is missing', () => { - channel.state.unreadCount = initialCountUnread; channel.state.read[user.id] = initialReadState; const { user: excludedUser, ...eventMissingUser } = notificationMarkUnreadEvent; const eventWithAnotherUser = { @@ -1584,7 +1973,6 @@ describe('Channel _handleChannelEvent', function () { }); it('should update channel read state produced for current user', () => { - channel.state.unreadCount = initialCountUnread; channel.state.read[user.id] = initialReadState; const event = messageReadEvent; @@ -1612,7 +2000,7 @@ describe('Channel _handleChannelEvent', function () { it('should update channel read state produced for another user', () => { const anotherUser = { id: 'another-user' }; - channel.state.unreadCount = initialCountUnread; + seedOwnUnreadCount(channel, initialCountUnread); channel.state.read[anotherUser.id] = initialReadState; const event = { ...messageReadEvent, user: anotherUser }; @@ -1638,13 +2026,16 @@ describe('Channel _handleChannelEvent', function () { it('should emit readStore subscription updates for single-user message.read events', () => { channel.state.read[user.id] = initialReadState; const changes = []; - const unsubscribe = channel.state.readStore.subscribe((next, prev) => { - if (!prev) return; - changes.push({ - next: next.read[user.id], - prev: prev.read[user.id], - }); - }); + const unsubscribe = channel.state.subscribeWithSelector( + (s) => ({ read: s.read }), + (next, prev) => { + if (!prev) return; + changes.push({ + next: next.read[user.id], + prev: prev.read[user.id], + }); + }, + ); channel._handleChannelEvent(messageReadEvent); unsubscribe(); @@ -1686,7 +2077,6 @@ describe('Channel _handleChannelEvent', function () { }); it('should update channel read state produced for current user', () => { - channel.state.unreadCount = initialCountUnread; channel.state.read[user.id] = initialReadState; channel._handleChannelEvent(messageDeliveredEvent); @@ -1734,7 +2124,7 @@ describe('Channel _handleChannelEvent', function () { it('should update channel read state produced for another user', () => { const anotherUser = { id: 'another-user' }; - channel.state.unreadCount = initialCountUnread; + seedOwnUnreadCount(channel, initialCountUnread); channel.state.read[anotherUser.id] = initialReadState; const event = { ...messageDeliveredEvent, user: anotherUser }; @@ -2195,6 +2585,67 @@ describe('Uninitialized Channel', () => { }); }); +describe('reactive channel mute status', () => { + let client; + let channel; + + beforeEach(() => { + client = new StreamChat('apiKey'); + client.user = { id: 'me' }; + channel = client.channel('messaging', 'mute-reactivity'); + }); + + it('seeds state.muteStatus from client.mutedChannels at construction', () => { + const preMutedClient = new StreamChat('apiKey'); + preMutedClient.user = { id: 'me' }; + preMutedClient.mutedChannels = [ + { + channel: { cid: 'messaging:premuted' }, + created_at: '2024-01-01T00:00:00.000Z', + }, + ]; + + const preMuted = preMutedClient.channel('messaging', 'premuted'); + + expect(preMuted.state.getLatestValue().muteStatus.muted).to.be.true; + }); + + it('reactively reflects notification.channel_mutes_updated into state.muteStatus', () => { + expect(channel.state.getLatestValue().muteStatus.muted).to.be.false; + + const seen = []; + const unsubscribe = channel.state.subscribeWithSelector( + (s) => ({ muted: s.muteStatus.muted }), + ({ muted }) => { + seen.push(muted); + }, + ); + + client.dispatchEvent({ + type: 'notification.channel_mutes_updated', + me: { + channel_mutes: [ + { + channel: { cid: channel.cid }, + created_at: '2024-01-01T00:00:00.000Z', + }, + ], + }, + }); + expect(channel.state.getLatestValue().muteStatus.muted).to.be.true; + + client.dispatchEvent({ + type: 'notification.channel_mutes_updated', + me: { channel_mutes: [] }, + }); + unsubscribe(); + + expect(channel.state.getLatestValue().muteStatus.muted).to.be.false; + // the selector only fires on real transitions — no health.check churn + expect(seen).to.eql([false, true, false]); + }); +}); + describe('Channels - Constructor', function () { const client = new StreamChat('key', 'secret'); // client.channel() now requires a connected user (userId derives from client.user). @@ -2235,10 +2686,10 @@ describe('Channels - Constructor', function () { it('undefined ID no options', function () { const channel = client.channel('messaging', undefined); expect(channel.id).to.eql(undefined); - // own_capabilities stays undefined ("not yet loaded") until the channel is - // hydrated; the reactive getter is still defined (hence enumerable). + // own_capabilities stays undefined ("not yet loaded") until the channel is hydrated, + // and no fields are fabricated onto an empty channel's data. expect(channel.data.own_capabilities).to.be.undefined; - expect(Object.keys(channel.data)).to.eql(['own_capabilities']); + expect(Object.keys(channel.data)).to.eql([]); }); it('short version with options', function () { @@ -3602,3 +4053,51 @@ describe('Channel.reload', () => { expect(watchSpy).toHaveBeenCalledTimes(1); }); }); +describe('Channel active flag (mark-read stays UI-driven)', () => { + let client; + let channel; + + beforeEach(() => { + client = getClientWithUser({ id: 'me' }); + channel = new Channel(client, 'messaging', 'active-x', {}); + client.activeChannels[channel.cid] = channel; + }); + + it('refcounts activate()/deactivate() behind the reactive active flag', () => { + expect(channel.active).to.equal(false); + expect(channel.state.getLatestValue().active).to.equal(false); + + channel.activate(); + expect(channel.active).to.equal(true); + + channel.activate(); // a second mount holds another ref + expect(channel.active).to.equal(true); + + channel.deactivate(); // one holder remains → still active + expect(channel.active).to.equal(true); + + channel.deactivate(); // last holder leaves → inactive + expect(channel.active).to.equal(false); + + channel.deactivate(); // underflow guard → stays inactive + expect(channel.active).to.equal(false); + }); + + it('does NOT auto-mark the channel read (mark-read is UI-driven, matching v10 destructive-reconciliation)', () => { + const spy = vi + .spyOn(client.messageDeliveryReporter, 'throttledMarkRead') + .mockImplementation(() => undefined); + + // Activate an unread channel and put it at the live edge — the channel must NOT auto-mark-read. + // Read is owned by the UI layer (MessageList marks read on viewability / scroll-to-bottom, + // gated on isViewingLive). A channel-level auto-read here would wipe the scroll-to-bottom unread + // badge while scrolled up — the "flash then vanish" regression. Parity guard so it can't return. + channel.activate(); + channel.messagePaginator.setViewingLive(true); + channel.state.read = { + me: { last_read: new Date(0), unread_messages: 3, user: { id: 'me' } }, + }; + + expect(spy).not.toHaveBeenCalled(); + }); +}); diff --git a/test/unit/channel_state.test.js b/test/unit/channel_state.test.js index 55daadec2f..bb5c3b02ed 100644 --- a/test/unit/channel_state.test.js +++ b/test/unit/channel_state.test.js @@ -52,7 +52,10 @@ describe('ChannelState members store', () => { expect(state.members).to.eql({}); expect(state.member_count).to.equal(0); - expect(state.membersStore.getLatestValue()).to.eql({ members: {}, memberCount: 0 }); + expect(state.getLatestValue()).to.deep.include({ + members: {}, + memberCount: 0, + }); }); it('keeps members getter/setter backward compatible while syncing the store', () => { @@ -64,7 +67,7 @@ describe('ChannelState members store', () => { state.members = members; expect(state.members).to.equal(members); - expect(state.membersStore.getLatestValue()).to.eql({ + expect(state.getLatestValue()).to.deep.include({ memberCount: 0, members, }); @@ -76,7 +79,7 @@ describe('ChannelState members store', () => { state.member_count = 42; expect(state.member_count).to.equal(42); - expect(state.membersStore.getLatestValue()).to.eql({ + expect(state.getLatestValue()).to.deep.include({ memberCount: 42, members: {}, }); @@ -90,7 +93,7 @@ describe('ChannelState member count bridge', () => { const state = channel.state; expect(state.member_count).to.equal(3); - expect(state.membersStore.getLatestValue()).to.eql({ + expect(state.getLatestValue()).to.deep.include({ memberCount: 3, members: {}, }); @@ -103,29 +106,52 @@ describe('ChannelState member count bridge', () => { const state = channel.state; channel.data = { ...channel.data, member_count: 7 }; - state.syncMemberCountFromChannelData(channel.data); + state.syncStateFromChannelData(channel.data); expect(state.member_count).to.equal(7); - expect(state.membersStore.getLatestValue()).to.eql({ + expect(state.getLatestValue()).to.deep.include({ memberCount: 7, members: {}, }); expect(channel.data?.member_count).to.equal(7); }); - it('keeps backward-compatible channel.data.member_count assignments in sync', () => { + it('keeps the last known member_count when a data update omits it (sticky fallback)', () => { const client = new StreamChat(); - const channel = new Channel(client, 'type', 'id', {}); + const channel = new Channel(client, 'type', 'id', { member_count: 4 }); const state = channel.state; - channel.data.member_count = 5; + const previousData = channel.data; + channel.data = { name: 'renamed' }; + state.syncStateFromChannelData(channel.data, previousData); - expect(state.member_count).to.equal(5); - expect(state.membersStore.getLatestValue()).to.eql({ - memberCount: 5, + expect(state.member_count).to.equal(4); + expect(state.getLatestValue()).to.deep.include({ + memberCount: 4, members: {}, }); - expect(channel.data.member_count).to.equal(5); + // sticky value is written back onto the raw data so raw readers stay consistent + expect(channel.data.member_count).to.equal(4); + }); +}); + +describe('ChannelState memberCount subscribers', () => { + it('does NOT re-notify memberCount subscribers on a members-only update', () => { + const client = new StreamChat(); + const state = new Channel(client, 'type', 'd', { member_count: 2 }).state; + const seen = []; + state.subscribeWithSelector( + (s) => ({ memberCount: s.memberCount }), + ({ memberCount }) => seen.push(memberCount), + ); + seen.length = 0; // drop the initial subscribe emission + + // a members map churn (presence/watchers/etc.) must not re-notify memberCount subscribers — + // this is what lets a consumer derive e.g. "is this a 1:1 channel" without paying for it + state.members = { alice: { user: { id: 'alice' } }, bob: { user: { id: 'bob' } } }; + + expect(seen).to.have.length(0); + expect(state.member_count).to.equal(2); }); }); @@ -134,7 +160,7 @@ describe('ChannelState read store', () => { const state = new ChannelState(); expect(state.read).to.eql({}); - expect(state.readStore.getLatestValue()).to.eql({ read: {} }); + expect(state.getLatestValue()).to.deep.include({ read: {} }); }); it('keeps read getter/setter backward compatible while syncing the store', () => { @@ -150,7 +176,52 @@ describe('ChannelState read store', () => { state.read = read; expect(state.read).to.equal(read); - expect(state.readStore.getLatestValue()).to.eql({ read }); + expect(state.getLatestValue()).to.deep.include({ read }); + }); +}); + +describe('ChannelState unreadCount', () => { + let client; + let channel; + + beforeEach(() => { + client = new StreamChat('apiKey'); + client.user = { id: 'me' }; + channel = new Channel(client, 'messaging', 'unread-count-id', {}); + }); + + it('derives the own count from the read state instead of storing it separately', () => { + expect(channel.state.unreadCount).to.equal(0); + + channel.state.read = { + me: { last_read: new Date(0), unread_messages: 7, user: { id: 'me' } }, + alice: { last_read: new Date(0), unread_messages: 3, user: { id: 'alice' } }, + }; + + expect(channel.state.unreadCount).to.equal(7); + expect(channel.countUnread()).to.equal(7); + }); + + it('is 0 while the current user has no read row', () => { + channel.state.read = { + alice: { last_read: new Date(0), unread_messages: 3, user: { id: 'alice' } }, + }; + + expect(channel.state.unreadCount).to.equal(0); + }); + + it('is 0 without a connected user, and does not throw on a channel pending disposal', () => { + client.user = undefined; + expect(channel.state.unreadCount).to.equal(0); + + client.user = { id: 'me' }; + channel.state.read = { + me: { last_read: new Date(0), unread_messages: 4, user: { id: 'me' } }, + }; + channel.pendingDisposal = true; + + expect(() => channel.state.unreadCount).not.to.throw(); + expect(channel.state.unreadCount).to.equal(4); }); }); @@ -159,7 +230,7 @@ describe('ChannelState watcher count store', () => { const state = new ChannelState(); expect(state.watcher_count).to.equal(0); - expect(state.watcherStore.getLatestValue()).to.eql({ + expect(state.getLatestValue()).to.deep.include({ watcherCount: 0, watchers: {}, }); @@ -171,7 +242,7 @@ describe('ChannelState watcher count store', () => { state.watcher_count = 42; expect(state.watcher_count).to.equal(42); - expect(state.watcherStore.getLatestValue()).to.eql({ + expect(state.getLatestValue()).to.deep.include({ watcherCount: 42, watchers: {}, }); @@ -183,7 +254,7 @@ describe('ChannelState watchers store', () => { const state = new ChannelState(); expect(state.watchers).to.eql({}); - expect(state.watcherStore.getLatestValue()).to.eql({ + expect(state.getLatestValue()).to.deep.include({ watcherCount: 0, watchers: {}, }); @@ -198,38 +269,19 @@ describe('ChannelState watchers store', () => { state.watchers = watchers; expect(state.watchers).to.equal(watchers); - expect(state.watcherStore.getLatestValue()).to.eql({ + expect(state.getLatestValue()).to.deep.include({ watcherCount: 0, watchers, }); }); }); -describe('ChannelState muted users store', () => { - it('initializes muted users store with an empty list', () => { - const state = new ChannelState(); - - expect(state.mutedUsers).to.eql([]); - expect(state.mutedUsersStore.getLatestValue()).to.eql({ mutedUsers: [] }); - }); - - it('keeps mutedUsers getter/setter backward compatible while syncing the store', () => { - const state = new ChannelState(); - const mutedUsers = [{ id: 'alice' }]; - - state.mutedUsers = mutedUsers; - - expect(state.mutedUsers).to.equal(mutedUsers); - expect(state.mutedUsersStore.getLatestValue()).to.eql({ mutedUsers }); - }); -}); - describe('ChannelState typing store', () => { it('initializes typing store with an empty typing map', () => { const state = new ChannelState(); expect(state.typing).to.eql({}); - expect(state.typingStore.getLatestValue()).to.eql({ typing: {} }); + expect(state.getLatestValue()).to.deep.include({ typing: {} }); }); it('keeps typing store and textComposer typing in sync via setTypingEvent/removeTypingEvent', () => { @@ -244,13 +296,13 @@ describe('ChannelState typing store', () => { state.setTypingEvent('alice', typingStartEvent); expect(state.typing).to.have.property('alice'); - expect(state.typingStore.getLatestValue().typing).to.have.property('alice'); + expect(state.getLatestValue().typing).to.have.property('alice'); expect(channel.messageComposer.textComposer.typing).to.have.property('alice'); state.removeTypingEvent('alice'); expect(state.typing).to.not.have.property('alice'); - expect(state.typingStore.getLatestValue().typing).to.not.have.property('alice'); + expect(state.getLatestValue().typing).to.not.have.property('alice'); expect(channel.messageComposer.textComposer.typing).to.not.have.property('alice'); }); }); @@ -276,7 +328,7 @@ describe('ChannelState own capabilities store', () => { }); const state = channel.state; - expect(state.ownCapabilitiesStore.getLatestValue()).to.eql({ + expect(state.getLatestValue()).to.deep.include({ ownCapabilities: ['send-message', 'upload-file'], }); expect(channel.data?.own_capabilities).to.eql(['send-message', 'upload-file']); @@ -293,28 +345,42 @@ describe('ChannelState own capabilities store', () => { ...channel.data, own_capabilities: ['pin-message'], }; - state.syncOwnCapabilitiesFromChannelData(channel.data); + state.syncStateFromChannelData(channel.data); - expect(state.ownCapabilitiesStore.getLatestValue()).to.eql({ + expect(state.getLatestValue()).to.deep.include({ ownCapabilities: ['pin-message'], }); expect(channel.data?.own_capabilities).to.eql(['pin-message']); }); - it('keeps backward-compatible channel.data.own_capabilities assignments in sync', () => { + it('keeps the last known own_capabilities when a data update omits them (sticky fallback)', () => { const client = new StreamChat(); - const channel = new Channel(client, 'type', 'id', {}); + const channel = new Channel(client, 'type', 'id', { + own_capabilities: ['send-message'], + }); const state = channel.state; - channel.data.own_capabilities = ['delete-message']; + const previousData = channel.data; + channel.data = { name: 'renamed' }; + state.syncStateFromChannelData(channel.data, previousData); - expect(state.ownCapabilitiesStore.getLatestValue()).to.eql({ - ownCapabilities: ['delete-message'], + expect(state.getLatestValue()).to.deep.include({ + ownCapabilities: ['send-message'], }); - expect(channel.data.own_capabilities).to.eql(['delete-message']); + // sticky value is written back onto the raw data so channelHasReadEvents stays consistent + expect(channel.data.own_capabilities).to.eql(['send-message']); }); - it('only wraps own_capabilities and keeps other channel.data fields as value properties', () => { + it('leaves own_capabilities undefined until known (#1732)', () => { + const client = new StreamChat(); + const channel = new Channel(client, 'type', 'id', {}); + + // unknown on the raw data, but the store slice defaults to an empty array + expect(channel.data.own_capabilities).to.be.undefined; + expect(channel.state.getLatestValue().ownCapabilities).to.eql([]); + }); + + it('exposes member_count / own_capabilities as plain value properties (no accessors)', () => { const client = new StreamChat(); const channel = new Channel(client, 'type', 'id', { hidden: false, @@ -322,29 +388,16 @@ describe('ChannelState own capabilities store', () => { own_capabilities: ['send-message'], }); - const ownCapabilitiesDescriptor = Object.getOwnPropertyDescriptor( - channel.data, - 'own_capabilities', - ); - const hiddenDescriptor = Object.getOwnPropertyDescriptor(channel.data, 'hidden'); - const memberCountDescriptor = Object.getOwnPropertyDescriptor( - channel.data, - 'member_count', - ); - - expect(ownCapabilitiesDescriptor).toBeDefined(); - expect('get' in ownCapabilitiesDescriptor).toBe(true); - expect('set' in ownCapabilitiesDescriptor).toBe(true); - expect(hiddenDescriptor).toBeDefined(); - expect('value' in hiddenDescriptor).toBe(true); - expect('get' in hiddenDescriptor).toBe(false); - expect('set' in hiddenDescriptor).toBe(false); - expect(memberCountDescriptor).toBeDefined(); - expect('get' in memberCountDescriptor).toBe(true); - expect('set' in memberCountDescriptor).toBe(true); + for (const key of ['own_capabilities', 'hidden', 'member_count']) { + const descriptor = Object.getOwnPropertyDescriptor(channel.data, key); + expect(descriptor).toBeDefined(); + expect('value' in descriptor).toBe(true); + expect('get' in descriptor).toBe(false); + expect('set' in descriptor).toBe(false); + } }); - it('does not overwrite non-capability fields when own_capabilities is updated', () => { + it('does not overwrite non-capability fields when channel.data is replaced', () => { const client = new StreamChat(); const channel = new Channel(client, 'type', 'id', { hidden: false, @@ -353,15 +406,127 @@ describe('ChannelState own capabilities store', () => { }); const state = channel.state; - channel.data.hidden = true; - channel.data.member_count = 5; - channel.data.own_capabilities = ['pin-message']; + const previousData = channel.data; + channel.data = { + ...channel.data, + hidden: true, + member_count: 5, + own_capabilities: ['pin-message'], + }; + state.syncStateFromChannelData(channel.data, previousData); expect(channel.data.hidden).to.equal(true); expect(channel.data.member_count).to.equal(5); expect(state.member_count).to.equal(5); - expect(state.ownCapabilitiesStore.getLatestValue()).to.eql({ + expect(state.getLatestValue()).to.deep.include({ ownCapabilities: ['pin-message'], }); }); }); + +describe('ChannelState unified store', () => { + it('publishes all slices through one StateStore and preserves siblings on single-key writes', () => { + const state = new ChannelState(); + const members = { alice: { user: { id: 'alice' }, user_id: 'alice' } }; + const read = { + alice: { last_read: new Date(0), unread_messages: 2, user: { id: 'alice' } }, + }; + const watchers = { bob: { id: 'bob' } }; + + state.members = members; + state.read = read; + state.watchers = watchers; + state.watcher_count = 3; + state.member_count = 5; + state.typing = { carol: { type: 'typing.start', user: { id: 'carol' } } }; + + // Every slice must survive: the setters go through partialNext, so a single-key write + // is a shallow merge — NOT a full replace that would wipe siblings. (Teeth-check: if any + // setter used `.next({ slice })` these references would be lost and this test would fail.) + const snapshot = state.getLatestValue(); + expect(snapshot.members).to.equal(members); + expect(snapshot.read).to.equal(read); + expect(snapshot.watchers).to.equal(watchers); + expect(snapshot.watcherCount).to.equal(3); + expect(snapshot.memberCount).to.equal(5); + expect(snapshot.typing).to.have.property('carol'); + }); + + it('is subscribable directly via subscribeWithSelector (useStateStore(channel.state, …))', () => { + const state = new ChannelState(); + const seen = []; + const unsubscribe = state.subscribeWithSelector( + (next) => ({ read: next.read }), + ({ read }) => { + seen.push(read); + }, + ); + + const read = { + alice: { last_read: new Date(0), unread_messages: 1, user: { id: 'alice' } }, + }; + state.read = read; + // a non-read write must NOT emit to a read selector + state.watcher_count = 9; + unsubscribe(); + + // initial emit + the read change; the watcher_count write is filtered out by the selector + expect(seen).to.have.length(2); + expect(seen[1]).to.equal(read); + }); + + it('publishes channel.data reactively so name/image/frozen changes are subscribable', () => { + const client = new StreamChat(); + const channel = new Channel(client, 'type', 'id', { name: 'orig' }); + const state = channel.state; + + expect(state.getLatestValue().data).to.deep.include({ name: 'orig' }); + + const names = []; + const unsubscribe = state.subscribeWithSelector( + (next) => ({ name: next.data?.name }), + ({ name }) => { + names.push(name); + }, + ); + + const previousData = channel.data; + channel.data = { ...channel.data, name: 'renamed' }; + state.syncStateFromChannelData(channel.data, previousData); + unsubscribe(); + + expect(names).to.eql(['orig', 'renamed']); + }); + + it('proxies the lifecycle flags (initialized/offlineMode/pendingDisposal) through the store', () => { + const client = new StreamChat(); + client.user = { id: 'me' }; + const channel = new Channel(client, 'messaging', 'lifecycle', {}); + client.activeChannels[channel.cid] = channel; + + expect(channel.initialized).to.equal(false); + expect(channel.offlineMode).to.equal(false); + expect(channel.pendingDisposal).to.equal(false); + expect(channel.state.getLatestValue()).to.deep.include({ + initialized: false, + offlineMode: false, + pendingDisposal: false, + }); + + const seen = []; + const unsubscribe = channel.state.subscribeWithSelector( + (s) => ({ initialized: s.initialized }), + ({ initialized }) => { + seen.push(initialized); + }, + ); + + // writing the getter/setter goes through the store, so subscribers are notified + channel.initialized = true; + unsubscribe(); + + expect(channel.initialized).to.equal(true); + expect(channel.state.getLatestValue().initialized).to.equal(true); + expect(seen).to.eql([false, true]); + }); +}); diff --git a/test/unit/client.test.js b/test/unit/client.test.js index d42fdb143b..050f7bcf58 100644 --- a/test/unit/client.test.js +++ b/test/unit/client.test.js @@ -395,10 +395,17 @@ describe('Client active channels cache', () => { client.user = user; client.wsPromise = Promise.resolve(); }; + const makeChannelMock = (unreadCount) => ({ + state: { unreadCount }, + _setOwnUnreadCount(next) { + this.state.unreadCount = next; + }, + }); + beforeEach(() => { client.activeChannels = { - vish: { state: { unreadCount: 1 } }, - vish2: { state: { unreadCount: 2 } }, + vish: makeChannelMock(1), + vish2: makeChannelMock(2), }; }); @@ -909,15 +916,20 @@ describe('StreamChat.queryChannels', async () => { const [channel] = await client.queryChannelsAndHydrate(); expect(channel.state.member_count).to.equal(7); - expect(channel.state.ownCapabilitiesStore.getLatestValue()).to.eql({ + expect(channel.state.getLatestValue()).to.deep.include({ ownCapabilities: ['send-message', 'read-events'], }); - channel.data.member_count = 8; - channel.data.own_capabilities = ['send-message']; + const previousData = channel.data; + channel.data = { + ...channel.data, + member_count: 8, + own_capabilities: ['send-message'], + }; + channel.state.syncStateFromChannelData(channel.data, previousData); expect(channel.state.member_count).to.equal(8); - expect(channel.state.ownCapabilitiesStore.getLatestValue()).to.eql({ + expect(channel.state.getLatestValue()).to.deep.include({ ownCapabilities: ['send-message'], }); diff --git a/test/unit/messageDelivery/MessageReceiptsTracker.test.ts b/test/unit/messageDelivery/MessageReceiptsTracker.test.ts index a43c055391..e338b25ad5 100644 --- a/test/unit/messageDelivery/MessageReceiptsTracker.test.ts +++ b/test/unit/messageDelivery/MessageReceiptsTracker.test.ts @@ -42,9 +42,8 @@ const createChannelMock = ({ return { channel: { - state: { - readStore, - }, + // `channel.state` is the reactive store itself; the tracker subscribes to its `read` slice. + state: readStore, // The default receipts locator now resolves timestamps via the message paginator; this mock // fn (still named findMessageByTimestamp in tests) backs messagePaginator.findItemByTimestamp. messagePaginator: { diff --git a/test/unit/offline-support/offline_support_api.test.ts b/test/unit/offline-support/offline_support_api.test.ts index f42649a9d5..a54c94dfa9 100644 --- a/test/unit/offline-support/offline_support_api.test.ts +++ b/test/unit/offline-support/offline_support_api.test.ts @@ -372,7 +372,7 @@ describe('OfflineSupportApi', () => { const mockChannelData = { id: '123', type: 'messaging' }; const mockChannel = { initialized: true, - disconnected: false, + pendingDisposal: false, data: mockChannelData, }; @@ -501,7 +501,7 @@ describe('OfflineSupportApi', () => { const mockChannelData = { id: '123', type: 'messaging' }; const mockChannel = { initialized: true, - disconnected: false, + pendingDisposal: false, data: mockChannelData, };