diff --git a/ai-docs/v10-breaking-changes.md b/ai-docs/v10-breaking-changes.md
index 805369d46..0b202a98a 100644
--- a/ai-docs/v10-breaking-changes.md
+++ b/ai-docs/v10-breaking-changes.md
@@ -431,7 +431,7 @@ Common rules for every generated or rewritten `StreamChat` method: path paramete
**Kind:** signature · **Impact:** medium
- **v9:** `queryThreads(options?)` -> `{ threads: Thread[], next }` (defaults `limit 10, participant_limit 10, reply_limit 3, watch true`, poll cache hydrated); `getThread(messageId, options?)` -> `Thread` (defaults `participant_limit 100, reply_limit 3, watch true`); `queryReactions(messageID, filter, sort?, options?)` read the offline DB first and dispatched `offline_reactions.queried`.
-- **v10:** `queryThreads(request?)`, `getThread({ message_id }, request?)`, `queryReactions({ id }, request?)` are raw generated calls with no defaults and no state. Hydrating versions: `queryThreadsAndHydrate(options?, requestOptions?)`, `getThreadAndHydrate({ message_id }, options?, requestOptions?)` (path params like `getThread`; thread instances resolved through `client.threads`), `queryReactionsAndHydrate({ id }, request?)`.
+- **v10:** `queryThreads(request?)`, `getThread({ message_id }, request?)`, `queryReactions({ id }, request?)` are raw generated calls with no defaults and no state. Hydrating versions: `queryThreadsAndHydrate(options?, requestOptions?)`, `getThreadAndHydrate({ message_id }, options?, requestOptions?)` (path params like `getThread`; a thread registered in `client.threads` is returned as it is, unless it is stale and not open: then it takes the answer, its replies replaced), `queryReactionsAndHydrate({ id }, request?)`.
- **Migrate:** switch calls that need `Thread` instances or offline reactions to the `*AndHydrate` variants (`getThread(id)` -> `getThreadAndHydrate({ message_id: id })`); when calling the raw ones pass `watch: true` yourself if needed.
- **Source:** `src/client.ts:1643-1670,2059-2142` (v10); `src/client.ts:2307-2349,3725-3808` (v9)
@@ -1923,9 +1923,20 @@ Client-level `queryThreads` / `getThread` becoming raw: [threads and reactions](
- **v9:** `new Thread({ client, threadData: ThreadResponse })` (required); apps used `client.threads.threadsById[id] ?? new Thread(...)`.
- **v10:** constructor `new Thread({ client, threadData?: ThreadStateResponse, channel?, parentMessage?, draft? })`. Without `threadData`, `channel` and `parentMessage` (with an `id`) are required, else it throws. A thread built from a parent message starts with `isStateStale: true` when the parent has replies or the channel is not watched, and loads once when activated. With `threadData`, `threadData.channel` and `threadData.parent_message` are required (throws otherwise) and the channel comes from `client.channelManager.ensure` (seeded with `channel.custom`, not `{ name }`). `ThreadResponse` -> `ThreadStateResponse` ([same-name changes](#same-name-meaning-changes)).
-- **Migrate:** `const thread = client.threads.ensure({ channel, parentMessage })`. Use `client.getThreadAndHydrate({ message_id: id })` / `client.queryThreadsAndHydrate()` for fetched threads; both return the already-registered instance when one exists.
+- **Migrate:** `const thread = client.threads.ensure({ channel, parentMessage })`. Use `client.getThreadAndHydrate({ message_id: id })` / `client.queryThreadsAndHydrate()` for fetched threads; both return the already-registered instance when one exists, hydrated from the answer only when it is stale and not open.
- **Source:** `src/thread.ts:133-231`, `src/thread_manager.ts:174-204`, `src/client.ts:2059-2142` (v10); `src/thread.ts:123-196` (v9)
+
+
+### `Thread.hydrateState` takes a thread response
+
+**Kind:** signature · **Impact:** low
+
+- **v9:** `thread.hydrateState(otherThread: Thread)` copied another instance's state and replies into this one (no-op when passed itself).
+- **v10:** `thread.hydrateState(threadData: ThreadStateResponse, options?: { reconcile? })` applies a `getThread` / `queryThreads` answer: the state, the channel's members and custom data, and the newest replies merged into the loaded window with `reconcile` (replacing it without). Throws for a response of another thread.
+- **Migrate:** pass the response (`thread.hydrateState(response.thread)`) instead of a second `Thread`, or call `client.getThreadAndHydrate({ message_id })`, which returns the registered thread (hydrated in place when it is stale and not open).
+- **Source:** `src/thread.ts:401-454`, `src/thread_manager.ts:169-190` (v10); `src/thread.ts:221-262` (v9)
+
### Thread state derivation: `replyCount`, `custom`, parent message via message store
@@ -1955,9 +1966,9 @@ Client-level `queryThreads` / `getThread` becoming raw: [threads and reactions](
**Kind:** behavior · **Impact:** low
- **v9:** `client.getThread(id, { watch: true })` (server default reply limit), replaced `state.replies` (plus failed replies), rethrew every error.
-- **v10:** `client.getThreadAndHydrate({ message_id: id }, { watch: true, reply_limit: max(loaded replies, pageSize) })`, merges into the reply paginator in place and re-ingests failed replies. A not-found error (Stream code or HTTP 404) resolves silently when `replyCount === 0`; other errors rethrow.
+- **v10:** `client.getThread({ message_id: id }, { watch: true, participant_limit: 100, reply_limit: max(loaded replies, pageSize) })`, applied to the thread itself with `hydrateState` (merged into the reply paginator in place), and re-ingests failed replies. A not-found error (Stream code or HTTP 404) resolves silently when `replyCount === 0`; other errors rethrow.
- **Migrate:** none.
-- **Source:** `src/thread.ts:413-466` (v10); `src/thread.ts:213-226` (v9)
+- **Source:** `src/thread.ts:329-384` (v10); `src/thread.ts:213-226` (v9)
diff --git a/ai-docs/v10-followups.md b/ai-docs/v10-followups.md
index c3a2c2464..0ba74c1e9 100644
--- a/ai-docs/v10-followups.md
+++ b/ai-docs/v10-followups.md
@@ -913,11 +913,11 @@ Non-breaking; land any time. Breaking simplifications are in [section 1](#fix-be
`simplification` · confidence: medium · pre-release: no · threads / reminders / delivery
- **Where / What:**
- - `src/pagination/paginators/ThreadPaginator.ts:113-125`: `toStoredInstance` repeats `ThreadManager.resolveQueried` (`src/thread_manager.ts:197-204`), which `queryThreadsAndHydrate` already applies (`src/client.ts:2093-2095`); only matters for a standalone paginator with a private store.
+ - `src/pagination/paginators/ThreadPaginator.ts:113-125`: `toStoredInstance` repeats `ThreadManager.resolveQueried` (`src/thread_manager.ts:197-204`), which `queryThreadsAndHydrate` already applies (`src/client.ts:2093-2095`); only matters for a standalone paginator with a private store. [SOLVED 25d24e8c]
- `src/reminders/ReminderManager.ts:209-221`: `reminder!` assertions plus "TODO: OAPI discrepancy?" right after `isReminderWsEventPayload` already checked `!!event.reminder` (`:190-193`).
- `src/thread.ts:129, 495-498, 872-878`: `failedRepliesMap` ("todo: do we really need the failedRepliesMap?"); `reload()` already reads failed replies from the paginator, only `hydrateState` uses the map.
- `src/messageDelivery/MessageReceiptsTracker.ts:732`: `furthestRefByOthers` reads the deprecated `client.userID` (the test mock only provides `userID`).
-- **Suggested change:** Delegate both resolve paths to one helper; make the guard a type predicate (`event is ReminderEvent & { reminder: ReminderResponseData }`); read pending replies from `messagePaginator.items` and drop the map; use `client.userId` and update the mock.
+- **Suggested change:** Delegate both resolve paths to one helper [SOLVED 25d24e8c]; make the guard a type predicate (`event is ReminderEvent & { reminder: ReminderResponseData }`); read pending replies from `messagePaginator.items` and drop the map; use `client.userId` and update the mock.
diff --git a/src/client.ts b/src/client.ts
index af387bdff..fbd17c5bb 100644
--- a/src/client.ts
+++ b/src/client.ts
@@ -61,7 +61,7 @@ import type {
import { isWSFailure } from './errors';
import type { APIError } from './errors';
import { chatLoggerSystem } from './logger';
-import { Thread } from './thread';
+import { DEFAULT_THREAD_PARTICIPANT_LIMIT, type Thread } from './thread';
import { Moderation } from './moderation';
import { ThreadManager } from './thread_manager';
import { DEFAULT_QUERY_CHANNELS_MESSAGE_LIST_PAGE_SIZE } from './constants';
@@ -1852,8 +1852,8 @@ export class StreamChat extends ChatApi {
const parentMessages = response.threads.map((thread) => thread.parent_message!);
this.polls.hydratePollCache(parentMessages);
- const threads = response.threads.map((thread) =>
- this.threads.resolveQueried(new Thread({ client: this, threadData: thread })),
+ const threads = response.threads.map((threadData) =>
+ this.threads.resolveQueried(threadData),
);
if (optionsWithDefaults.watch) threads.forEach(markThreadChannelWatched);
@@ -1884,7 +1884,7 @@ export class StreamChat extends ChatApi {
}
const optionsWithDefaults = {
- participant_limit: 100,
+ participant_limit: DEFAULT_THREAD_PARTICIPANT_LIMIT,
reply_limit: 3,
watch: true,
...options,
@@ -1896,9 +1896,7 @@ export class StreamChat extends ChatApi {
requestOptions,
);
- const thread = this.threads.resolveQueried(
- new Thread({ client: this, threadData: response.thread }),
- );
+ const thread = this.threads.resolveQueried(response.thread);
if (optionsWithDefaults.watch) markThreadChannelWatched(thread);
return thread;
diff --git a/src/pagination/paginators/ThreadPaginator.ts b/src/pagination/paginators/ThreadPaginator.ts
index a17caeb32..c57603e75 100644
--- a/src/pagination/paginators/ThreadPaginator.ts
+++ b/src/pagination/paginators/ThreadPaginator.ts
@@ -102,21 +102,10 @@ export class ThreadPaginator extends BasePaginator
PaginationQueryReturnValue
> => {
const { threads, next } = await this.client.queryThreadsAndHydrate(queryShape);
- return { items: threads.map(this.toStoredInstance), tailward: next };
- };
-
- /**
- * Keeps one instance per ID (the index upserts, so a second one would replace the stored one). A
- * stale stored thread is hydrated from the query unless it is active: it reloads itself, and
- * hydrating would replace the replies on screen.
- */
- private toStoredInstance = (incoming: Thread): Thread => {
- const stored = this.store.get(incoming.id);
- if (!stored) return incoming;
- if (stored.hasStaleState && !stored.state.getLatestValue().active) {
- stored.hydrateState(incoming);
- }
- return stored;
+ return {
+ items: threads.map((thread) => this.store.get(thread.id) ?? thread),
+ tailward: next,
+ };
};
filterQueryResults = (threads: Thread[]) => threads;
diff --git a/src/thread.ts b/src/thread.ts
index 8c7301ad8..f97bc137c 100644
--- a/src/thread.ts
+++ b/src/thread.ts
@@ -133,42 +133,12 @@ export class Thread extends WithSubscriptions {
}) {
super();
if (threadData) {
- if (!threadData.channel) {
- throw new Error('Thread channel is required when threadData is provided');
- }
- if (!threadData.parent_message) {
- throw new Error('Thread parent_message is required when threadData is provided');
- }
- const threadChannel = client.channelManager.ensure({
- type: threadData.channel.type,
- id: threadData.channel.id,
- data: { custom: threadData.channel.custom },
- });
- threadChannel._hydrateMembers({
- members: threadData.channel.members ?? [],
- overrideCurrentState: false,
- });
-
this.state = new StateStore({
active: false,
isLoading: false,
isStateStale: false,
- channel: threadChannel,
- createdAt: threadData.created_at,
- deletedAt: threadData.deleted_at ?? null,
- parentMessage: formatMessage(threadData.parent_message),
- participants: threadData.thread_participants,
- read: formatReadState(
- !threadData.read || threadData.read.length === 0
- ? getPlaceholderReadResponse(client.userId)
- : threadData.read,
- ),
- // From the parent message: the top-level `threadData.reply_count` excludes soft-deleted
- // replies, the parent's (like the channel's) includes them.
- replyCount: threadData.parent_message.reply_count ?? 0,
- updatedAt: threadData.updated_at ?? null,
- title: threadData.title,
- custom: threadData.custom ?? {},
+ ...formatThreadState(threadData, client.userId),
+ channel: ensureThreadChannel(client, threadData),
});
this.id = threadData.parent_message_id;
@@ -241,7 +211,7 @@ export class Thread extends WithSubscriptions {
// the head, and the tail only if it holds every reply.
if (threadData?.latest_replies?.length) {
const replies = threadData.latest_replies.map(formatMessage);
- const isLastPage = replies.length === (threadData.reply_count ?? 0);
+ const isLastPage = holdsEveryReply(replies, threadData);
this.messagePaginator.setItems({
valueOrFactory: replies,
cursor: getCursorFromReplies(replies, isLastPage),
@@ -369,14 +339,16 @@ export class Thread extends WithSubscriptions {
Math.max(loadedReplyCount, this.messagePaginator.pageSize ?? 0) || undefined;
const reconcileCandidateIds = new Set(loadedReplies.map((reply) => reply.id));
const failedBefore = loadedReplies.filter((reply) => reply.status === 'failed');
- const thread = await this.client.getThreadAndHydrate(
+ const { thread: threadData } = await this.client.getThread(
{ message_id: this.id },
{
+ participant_limit: DEFAULT_THREAD_PARTICIPANT_LIMIT,
watch: true,
reply_limit: requestedReplyLimit,
},
);
- this.hydrateState(thread, {
+ this.channel.watchStatus = ChannelWatchStatus.Watching;
+ this.hydrateState(threadData, {
reconcile: {
requestedLimit: requestedReplyLimit,
candidateIds: reconcileCandidateIds,
@@ -410,59 +382,38 @@ export class Thread extends WithSubscriptions {
}
};
+ /**
+ * Updates the thread's state and its channel's members and custom data, and merges the replies
+ * into the loaded window (with `reconcile`) or replaces it (without).
+ */
public hydrateState = (
- thread: Thread,
+ threadData: ThreadStateResponse,
options?: { reconcile?: MergeNewestPageOptions },
) => {
- if (thread === this) {
- return;
+ if (threadData.parent_message_id !== this.id) {
+ throw new Error("Cannot hydrate thread's state from another thread's response");
}
- if (thread.id !== this.id) {
- throw new Error(
- "Cannot hydrate thread's state using thread with different threadId",
- );
- }
-
- const {
- createdAt,
- custom,
- title,
- deletedAt,
- parentMessage,
- participants,
- read,
- replyCount,
- updatedAt,
- } = thread.state.getLatestValue();
+ const serverState = formatThreadState(threadData, this.client.userId);
+ const reconcile = options?.reconcile;
// Failed replies survive the hydrate: they are re-ingested after the incoming page is merged.
const pendingReplies = Array.from(this.failedRepliesMap.values());
- this.state.partialNext({
- title,
- createdAt,
- custom,
- deletedAt,
- parentMessage,
- participants,
- read,
- replyCount,
- updatedAt,
- isStateStale: false,
- });
+ ensureThreadChannel(this.client, threadData);
+ this.state.partialNext({ ...serverState, isStateStale: false });
- if (parentMessage && this.hasSubscriptions) {
- this.client.messageStore.upsert(parentMessage);
+ if (this.hasSubscriptions) {
+ this.client.messageStore.upsert(serverState.parentMessage);
}
- const incomingReplies = thread.messagePaginator.state.getLatestValue().items ?? [];
-
- if (!options?.reconcile) this.messagePaginator.clearStateAndCache();
+ const incomingReplies = (threadData.latest_replies ?? []).map(formatMessage);
+ if (!reconcile) this.messagePaginator.clearStateAndCache();
if (typeof this.messagePaginator.items === 'undefined') {
- // Nothing loaded yet (e.g. a thread opened from a message): `mergeNewestPage` no-ops without
- // an anchored head and would drop the fetched replies, so seed as the constructor does.
- const isLastPage = incomingReplies.length === replyCount;
+ // Nothing loaded yet (e.g. a thread opened from a message): `mergeNewestPage` no-ops
+ // without an anchored head and would drop the fetched replies, so seed as the constructor
+ // does.
+ const isLastPage = holdsEveryReply(incomingReplies, threadData);
this.messagePaginator.setItems({
valueOrFactory: incomingReplies,
cursor: getCursorFromReplies(incomingReplies, isLastPage),
@@ -470,11 +421,11 @@ export class Thread extends WithSubscriptions {
isLastPage,
});
} else {
- this.messagePaginator.mergeNewestPage(incomingReplies, options?.reconcile);
+ this.messagePaginator.mergeNewestPage(incomingReplies, reconcile);
}
pendingReplies.forEach((reply) => this.messagePaginator.ingestItem(reply));
// Keeps `lastMessageAt` fresh when the merged page lacks the newest reply. Monotonic.
- this.messagePaginator.seedLastMessageAt(thread.messagePaginator.lastMessageAt);
+ this.messagePaginator.seedLastMessageAt(threadData.last_message_at);
};
public registerSubscriptions = () => {
@@ -856,6 +807,58 @@ const normalizeThreadParticipants = (
);
};
+/** `participant_limit` of the single-thread requests (`getThreadAndHydrate`, `reload()`). */
+export const DEFAULT_THREAD_PARTICIPANT_LIMIT = 100;
+
+/** The thread's channel, with the members and custom data of `threadData` applied. */
+export const ensureThreadChannel = (
+ client: StreamChat,
+ threadData: ThreadStateResponse,
+) => {
+ if (!threadData.channel) {
+ throw new Error('Thread channel is required when threadData is provided');
+ }
+ const channel = client.channelManager.ensure({
+ type: threadData.channel.type,
+ id: threadData.channel.id,
+ data: { custom: threadData.channel.custom },
+ });
+ channel._hydrateMembers({
+ members: threadData.channel.members ?? [],
+ overrideCurrentState: false,
+ });
+ return channel;
+};
+
+/** Against the top-level `reply_count`: unlike the parent's, it excludes soft-deleted replies. */
+const holdsEveryReply = (replies: LocalMessage[], threadData: ThreadStateResponse) =>
+ replies.length === (threadData.reply_count ?? 0);
+
+const formatThreadState = (threadData: ThreadStateResponse, currentUserId?: string) => {
+ const parentMessage = threadData.parent_message;
+ if (!parentMessage) {
+ throw new Error('Thread parent_message is required when threadData is provided');
+ }
+
+ return {
+ createdAt: threadData.created_at,
+ custom: threadData.custom ?? {},
+ deletedAt: threadData.deleted_at ?? null,
+ parentMessage: formatMessage(parentMessage),
+ participants: threadData.thread_participants,
+ read: formatReadState(
+ !threadData.read || threadData.read.length === 0
+ ? getPlaceholderReadResponse(currentUserId)
+ : threadData.read,
+ ),
+ // From the parent message: the top-level `threadData.reply_count` excludes soft-deleted
+ // replies, the parent's (like the channel's) includes them.
+ replyCount: parentMessage.reply_count ?? 0,
+ title: threadData.title,
+ updatedAt: threadData.updated_at ?? null,
+ };
+};
+
const formatReadState = (read: ReadStateResponse[]): ThreadReadState =>
read.reduce((state, userRead) => {
state[userRead.user.id] = {
diff --git a/src/thread_manager.ts b/src/thread_manager.ts
index 330d81223..0e220827a 100644
--- a/src/thread_manager.ts
+++ b/src/thread_manager.ts
@@ -8,7 +8,7 @@ import { ThreadPaginator } from './pagination/paginators/ThreadPaginator';
import type { Channel } from './channel';
import type { StreamChat } from './client';
-import { Thread } from './thread';
+import { ensureThreadChannel, Thread } from './thread';
import type {
Event,
EventPayload,
@@ -16,6 +16,7 @@ import type {
LocalMessage,
MessageResponse,
OwnUserResponse,
+ ThreadStateResponse,
} from './types';
import { WithSubscriptions } from './utils/WithSubscriptions';
@@ -165,19 +166,21 @@ export class ThreadManager extends WithSubscriptions {
};
/**
- * The registered instance for a thread a query just built, else the query's own, so events keep
- * reaching the one managed instance. A stale registered thread is refreshed from the query unless
- * it is active (it reloads itself).
+ * The registered thread, or a new one when none is registered. Only a stale thread nobody has
+ * open is hydrated from `threadData`: an open one reloads itself, and events keep any other
+ * current (an answer that was in flight would roll them back). The channel's members and custom
+ * data are applied either way, as the constructor does for a new thread.
*
* @internal
*/
- public resolveQueried = (incoming: Thread): Thread => {
- const stored = this.threadStore.get(incoming.id);
- if (!stored) return incoming;
- if (stored.hasStaleState && !stored.state.getLatestValue().active) {
- stored.hydrateState(incoming);
- }
- return stored;
+ public resolveQueried = (threadData: ThreadStateResponse): Thread => {
+ const registered = this.threadStore.get(threadData.parent_message_id);
+ if (!registered) return new Thread({ client: this.client, threadData });
+
+ const { active, isStateStale } = registered.state.getLatestValue();
+ if (isStateStale && !active) registered.hydrateState(threadData);
+ else ensureThreadChannel(this.client, threadData);
+ return registered;
};
/**
diff --git a/test/unit/pagination/paginators/ThreadPaginator.test.ts b/test/unit/pagination/paginators/ThreadPaginator.test.ts
index f4bb5d0f1..e9c860b10 100644
--- a/test/unit/pagination/paginators/ThreadPaginator.test.ts
+++ b/test/unit/pagination/paginators/ThreadPaginator.test.ts
@@ -1,7 +1,7 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
-import type { ChannelResponse, MessageResponse, StreamChat } from '../../../../src';
-import { Thread } from '../../../../src';
+import type { ChannelResponse, MessageResponse } from '../../../../src';
+import { EntityStore, StreamChat, Thread, ThreadPaginator } from '../../../../src';
import { generateUUIDv4 as uuidv4 } from '../../../../src/utils';
import { nowNs } from '../../../../src/utils/time';
import { generateChannel } from '../../test-utils/generateChannel';
@@ -13,14 +13,10 @@ describe('ThreadPaginator', () => {
let client: StreamChat;
let channelResponse: ChannelResponse;
+ const threadResponse = (id = uuidv4()) =>
+ generateThreadResponse(channelResponse, generateMsg({ id }) as MessageResponse);
const makeThread = (id = uuidv4()) =>
- new Thread({
- client,
- threadData: generateThreadResponse(
- channelResponse,
- generateMsg({ id }) as MessageResponse,
- ),
- });
+ new Thread({ client, threadData: threadResponse(id) });
const ids = () => (client.threads.paginator.items ?? []).map((thread) => thread.id);
const respond = (threads: Thread[], next?: string) =>
vi.spyOn(client, 'queryThreadsAndHydrate').mockResolvedValueOnce({ next, threads });
@@ -31,6 +27,21 @@ describe('ThreadPaginator', () => {
.channel as ChannelResponse;
});
+ it('keeps the instances a store of its own holds across reloads', async () => {
+ const paginator = new ThreadPaginator({
+ client,
+ store: new EntityStore({ getEntityId: (thread) => thread.id }),
+ });
+ respond([makeThread('a')]);
+ await paginator.reload();
+ const [listed] = paginator.items ?? [];
+
+ respond([makeThread('a')]);
+ await paginator.reload();
+
+ expect(paginator.items?.[0]).toBe(listed);
+ });
+
it('publishes isLoading on the first load, like any paginator, with no items until it lands', async () => {
respond([makeThread()]);
@@ -191,7 +202,13 @@ describe('ThreadPaginator', () => {
client.threads.paginator.state.subscribe(({ isLoading, items }) =>
snapshots.push({ hasItems: items !== undefined, isLoading }),
);
- respond([c, makeThread('a')]);
+ // The raw answer, through the real `queryThreadsAndHydrate`: `a` resolves to its listed instance.
+ vi.spyOn(client, 'queryThreads').mockResolvedValueOnce({
+ threads: [threadResponse('c'), threadResponse('a')],
+ } as never);
+ vi.spyOn(client, 'queryThreadsAndHydrate').mockImplementationOnce((...args) =>
+ StreamChat.prototype.queryThreadsAndHydrate.apply(client, args),
+ );
await client.threads.reload({ force: true });
diff --git a/test/unit/threads.test.ts b/test/unit/threads.test.ts
index 3fb003794..839f3c229 100644
--- a/test/unit/threads.test.ts
+++ b/test/unit/threads.test.ts
@@ -50,6 +50,22 @@ describe('Threads 2.0', () => {
});
}
+ // What `client.getThread` answers, for stubbing `Thread.reload()`.
+ function createGetThreadResponse({
+ parentMessageOverrides = {},
+ ...overrides
+ }: Partial & {
+ parentMessageOverrides?: Partial;
+ } = {}) {
+ return {
+ thread: generateThreadResponse(
+ channelResponse,
+ { ...parentMessageResponse, ...parentMessageOverrides },
+ overrides,
+ ),
+ };
+ }
+
function createMinimalThread({
parentMessageOverrides = {},
draft,
@@ -411,14 +427,37 @@ describe('Threads 2.0', () => {
});
describe('hydrateState', () => {
- it('prevents hydrating state from the instance with a different id', () => {
+ it('prevents hydrating state from a response for another thread', () => {
const thread = createTestThread();
- const otherThread = createTestThread({
+ const otherThreadData = createGetThreadResponse({
parentMessageOverrides: { id: uuidv4() },
- });
+ }).thread;
- expect(thread.id).to.not.equal(otherThread.id);
- expect(() => thread.hydrateState(otherThread)).to.throw();
+ expect(thread.id).to.not.equal(otherThreadData.parent_message_id);
+ expect(() => thread.hydrateState(otherThreadData)).to.throw();
+ });
+
+ it('seeds a reply window that holds every reply as the last page, with soft-deleted replies counted out', () => {
+ // The parent's `reply_count` includes a soft-deleted reply the server does not return;
+ // the top-level one does not, as for the constructor.
+ const thread = createMinimalThread();
+ const replies = ['2020-01-01T00:00:01.000Z', '2020-01-01T00:00:02.000Z'].map(
+ (createdAt) =>
+ generateMsg({
+ parent_id: parentMessageResponse.id,
+ created_at: convertDateToTimestamp(createdAt),
+ }) as MessageResponse,
+ );
+
+ thread.hydrateState(
+ createGetThreadResponse({
+ latest_replies: replies,
+ parentMessageOverrides: { reply_count: 3 },
+ reply_count: 2,
+ }).thread,
+ );
+
+ expect(thread.messagePaginator.state.getLatestValue().hasMoreTail).to.be.false;
});
it('seeds the reply window when the thread has never loaded one', () => {
@@ -434,15 +473,15 @@ describe('Threads 2.0', () => {
parent_id: parentMessageResponse.id,
created_at: convertDateToTimestamp('2020-01-01T00:00:00.000Z'),
}) as MessageResponse;
- const hydrationThread = createTestThread({
+ const threadData = createGetThreadResponse({
latest_replies: [reply],
// The count that matters is the parent message's — the thread endpoint's top-level
// `reply_count` excludes soft-deleted replies, so the LLC reads the parent's.
parentMessageOverrides: { reply_count: 1 },
reply_count: 1,
- });
+ }).thread;
- thread.hydrateState(hydrationThread);
+ thread.hydrateState(threadData);
const paginatorState = thread.messagePaginator.state.getLatestValue();
expect(paginatorState.items?.map((item) => item.id)).to.deep.equal([reply.id]);
@@ -462,13 +501,13 @@ describe('Threads 2.0', () => {
parent_id: parentMessageResponse.id,
created_at: convertDateToTimestamp('2020-01-03T00:00:00.000Z'),
}) as MessageResponse;
- const hydrationThread = createTestThread({
+ const threadData = createGetThreadResponse({
latest_replies: [newer, older],
parentMessageOverrides: { reply_count: 5 },
reply_count: 5,
- });
+ }).thread;
- thread.hydrateState(hydrationThread);
+ thread.hydrateState(threadData);
const getRepliesStub = sinon
.stub(thread.channel.getClient(), 'getReplies')
.resolves({ messages: [], duration: '' } as unknown as ReturnType<
@@ -480,30 +519,34 @@ describe('Threads 2.0', () => {
expect(getRepliesStub.firstCall.args[1]?.id_lt).to.equal(older.id);
});
- it('copies state of the instance with the same id', () => {
+ it('takes the state of a response for the same thread', () => {
const thread = createTestThread();
- const hydrationThread = createTestThread();
- thread.hydrateState(hydrationThread);
+ const threadData = createGetThreadResponse({
+ parentMessageOverrides: { text: 'edited-on-server' },
+ thread_participants: [
+ { user_id: 'u1' },
+ ] as ThreadStateResponse['thread_participants'],
+ title: 'renamed',
+ }).thread;
+ thread.hydrateState(threadData);
const stateAfter = thread.state.getLatestValue();
- const hydrationState = hydrationThread.state.getLatestValue();
- // compare non-primitive values only
- expect(stateAfter.read).to.equal(hydrationState.read);
- expect(stateAfter.parentMessage).to.equal(hydrationState.parentMessage);
- expect(stateAfter.participants).to.equal(hydrationState.participants);
+ expect(stateAfter.title).to.equal('renamed');
+ expect(stateAfter.participants).to.equal(threadData.thread_participants);
+ expect(stateAfter.parentMessage.text).to.equal('edited-on-server');
});
it('retains failed replies after hydration', () => {
const thread = createTestThread();
- const hydrationThread = createTestThread({
+ const threadData = createGetThreadResponse({
latest_replies: [
makeReply({
created_at: convertDateToTimestamp('2020-01-01T00:00:01.000Z'),
}),
],
reply_count: 1,
- });
+ }).thread;
// Pinned, failed reply NEWEST: `makeReply()` reads `created_at` from the clock, so implicit
// timestamps landed in random order and an older-than-window reply sits below it, not in
@@ -514,7 +557,7 @@ describe('Threads 2.0', () => {
});
thread.upsertReplyLocally({ message: failedMessage });
- thread.hydrateState(hydrationThread);
+ thread.hydrateState(threadData);
// The failed reply survives the hydrate (it is re-ingested into the paginator).
expect(repliesOf(thread).map((reply) => reply.id)).to.include(failedMessage.id);
@@ -540,11 +583,11 @@ describe('Threads 2.0', () => {
// Reconnect hydrate (Thread.reload → hydrateState → mergeNewestPage) must RE-DERIVE the cursor
// from the merged reply window so pagination works again.
- const hydrationThread = createTestThread({
+ const threadData = createGetThreadResponse({
latest_replies: [existingReply],
reply_count: 10,
- });
- thread.hydrateState(hydrationThread, {
+ }).thread;
+ thread.hydrateState(threadData, {
reconcile: { candidateIds: new Set([existingReply.id]) },
});
@@ -576,12 +619,12 @@ describe('Threads 2.0', () => {
parent_id: parentMessageResponse.id,
created_at: convertDateToTimestamp('2020-01-02T00:00:00.000Z'),
}) as MessageResponse;
- const hydrationThread = createTestThread({
+ const threadData = createGetThreadResponse({
latest_replies: [editedReply, newReply],
reply_count: 11,
- });
+ }).thread;
- thread.hydrateState(hydrationThread, {
+ thread.hydrateState(threadData, {
reconcile: { candidateIds: new Set([existingReply.id]) },
});
@@ -611,12 +654,12 @@ describe('Threads 2.0', () => {
reply_count: 5,
});
// The page a list query brings: only the newest replies, fetched without a snapshot.
- const hydrationThread = createTestThread({
+ const threadData = createGetThreadResponse({
latest_replies: [{ ...kept, text: 'edited' }],
reply_count: 4,
- });
+ }).thread;
- thread.hydrateState(hydrationThread);
+ thread.hydrateState(threadData);
expect(repliesOf(thread).map((reply) => reply.id)).to.deep.equal([kept.id]);
expect(thread.messagePaginator.getItem(kept.id)?.text).to.equal('edited');
@@ -633,10 +676,10 @@ describe('Threads 2.0', () => {
);
// A fresh re-query (reload / reconnect) carries an edited parent.
- const hydrationThread = createTestThread({
+ const threadData = createGetThreadResponse({
parentMessageOverrides: { text: 'edited-on-server' },
- });
- thread.hydrateState(hydrationThread);
+ }).thread;
+ thread.hydrateState(threadData);
// Both the projection AND the store's canonical copy are the hydrated parent (no
// divergence). Regression: hydrateState used to update only state.parentMessage, leaving
@@ -741,6 +784,37 @@ describe('Threads 2.0', () => {
expect(stub.thirdCall.args[1]?.reply_limit).to.equal(pageSize);
});
+ it('applies the fetched replies to a registered thread', async () => {
+ // Stubs `getThread`, not `getThreadAndHydrate`: the registry lookup is what used to hand
+ // the thread back to itself, with nothing applied.
+ const r1 = makeReply({
+ id: 'r1',
+ created_at: convertDateToTimestamp('2020-01-01T00:00:01.000Z'),
+ });
+ const r2 = makeReply({
+ id: 'r2',
+ created_at: convertDateToTimestamp('2020-01-01T00:00:02.000Z'),
+ });
+ const thread = createTestThread({ latest_replies: [r1], reply_count: 1 });
+ subscribeThread(thread);
+ const release = thread.activate();
+ sinon.stub(client, 'getThread').resolves({
+ thread: generateThreadResponse(
+ channelResponse,
+ { ...parentMessageResponse, reply_count: 2 },
+ { latest_replies: [r1, r2], reply_count: 2 },
+ ),
+ });
+
+ await thread.reload();
+
+ expect(repliesOf(thread).map((reply) => reply.id)).to.eql(['r1', 'r2']);
+ expect(client.threads.get(thread.id)).to.equal(thread);
+ expect(client.messageStore.holderNames('r2')).to.have.length(1);
+ release();
+ thread.unregisterSubscriptions();
+ });
+
it('merges the fetched page into a thread whose only reply was ingested live', async () => {
// A thread created this session has an EMPTY first reply page, so nothing anchors its
// window and the sent reply lands in a LOGICAL head. `mergeNewestPage` used to require an
@@ -768,8 +842,8 @@ describe('Threads 2.0', () => {
id: 'peer2',
created_at: convertDateToTimestamp('2020-01-01T00:00:03.000Z'),
});
- sinon.stub(client, 'getThreadAndHydrate').resolves(
- createTestThread({
+ sinon.stub(client, 'getThread').resolves(
+ createGetThreadResponse({
latest_replies: [
makeReply({
id: 'mine',
@@ -819,8 +893,8 @@ describe('Threads 2.0', () => {
const activeBefore = repliesOf(thread).map((r) => r.id);
expect(activeBefore).to.eql(['old1', 'old2']);
- sinon.stub(client, 'getThreadAndHydrate').resolves(
- createTestThread({
+ sinon.stub(client, 'getThread').resolves(
+ createGetThreadResponse({
latest_replies: [
makeReply({
id: 'mine',
@@ -853,7 +927,7 @@ describe('Threads 2.0', () => {
StatusCode: 404,
},
);
- sinon.stub(client, 'getThreadAndHydrate').rejects(notFound);
+ sinon.stub(client, 'getThread').rejects(notFound);
// Resolves rather than rejecting: nothing to reload is not a failure, and rethrowing would
// make every caller log "reload failed" on a brand-new thread.
@@ -870,7 +944,7 @@ describe('Threads 2.0', () => {
status: 404,
},
);
- sinon.stub(client, 'getThreadAndHydrate').rejects(notFound);
+ sinon.stub(client, 'getThread').rejects(notFound);
await expect(thread.reload()).resolves.toBeUndefined();
});
@@ -898,7 +972,7 @@ describe('Threads 2.0', () => {
status: 404,
},
);
- sinon.stub(client, 'getThreadAndHydrate').rejects(notFound);
+ sinon.stub(client, 'getThread').rejects(notFound);
await expect(thread.reload()).rejects.toThrow(notFound);
});
@@ -906,7 +980,7 @@ describe('Threads 2.0', () => {
it('still rethrows a real failure', async () => {
const thread = createTestThread({ latest_replies: [], reply_count: 0 });
const failure = Object.assign(new Error('boom'), { code: 9, StatusCode: 500 });
- sinon.stub(client, 'getThreadAndHydrate').rejects(failure);
+ sinon.stub(client, 'getThread').rejects(failure);
await expect(thread.reload()).rejects.toThrow(failure);
});
@@ -961,14 +1035,14 @@ describe('Threads 2.0', () => {
});
// The server's authoritative page (computed before r4 existed) has r3 hard-deleted, no r4.
- const hydrationThread = createTestThread({
+ const getThreadResponse = createGetThreadResponse({
latest_replies: [r1, r2],
reply_count: 2,
});
- sinon.stub(client, 'getThreadAndHydrate').callsFake(async () => {
+ sinon.stub(client, 'getThread').callsFake(async () => {
thread.messagePaginator.ingestItem(formatMessage(r4)); // live arrival during the await
- return hydrationThread;
+ return getThreadResponse;
});
await thread.reload();
@@ -1014,11 +1088,11 @@ describe('Threads 2.0', () => {
id: 'far2',
created_at: convertDateToTimestamp('2021-06-01T00:00:02.000Z'),
});
- const hydrationThread = createTestThread({
+ const getThreadResponse = createGetThreadResponse({
latest_replies: [far1, far2],
reply_count: 2,
});
- sinon.stub(client, 'getThreadAndHydrate').resolves(hydrationThread);
+ sinon.stub(client, 'getThread').resolves(getThreadResponse);
await thread.reload();
@@ -1028,7 +1102,7 @@ describe('Threads 2.0', () => {
it('rethrows a reload failure and releases isLoading', async () => {
const thread = createTestThread({ latest_replies: [], reply_count: 0 });
const failure = new Error('thread reload failed');
- const stub = sinon.stub(client, 'getThreadAndHydrate').rejects(failure);
+ const stub = sinon.stub(client, 'getThread').rejects(failure);
await expect(thread.reload()).rejects.toThrow(failure);
// Not left mid-flight — otherwise the isLoading guard would swallow every later reload.
@@ -1037,8 +1111,8 @@ describe('Threads 2.0', () => {
// Which is exactly what a later reload proves: it must reach the client again.
stub.restore();
const succeeding = sinon
- .stub(client, 'getThreadAndHydrate')
- .resolves(createTestThread({ latest_replies: [], reply_count: 0 }));
+ .stub(client, 'getThread')
+ .resolves(createGetThreadResponse({ latest_replies: [], reply_count: 0 }));
await thread.reload();
expect(succeeding.calledOnce).to.equal(true);
});
@@ -1275,8 +1349,8 @@ describe('Threads 2.0', () => {
const reloadedReply = makeReply({
created_at: convertDateToTimestamp('2020-03-01T00:00:01.000Z'),
});
- const stubbedGetThread = sinon.stub(client, 'getThreadAndHydrate').resolves(
- createTestThread({
+ const stubbedGetThread = sinon.stub(client, 'getThread').resolves(
+ createGetThreadResponse({
latest_replies: [initialReply, reloadedReply],
reply_count: 2,
}),
@@ -2403,6 +2477,19 @@ describe('Threads 2.0', () => {
});
await manager.paginator.reload();
};
+ /**
+ * Serves the next thread query from what the server answers, through the real
+ * `queryThreadsAndHydrate`, so registered threads are matched as in the SDK.
+ */
+ const respondFromServer = (threads: ThreadStateResponse[], next?: string) => {
+ vi.spyOn(client, 'queryThreads').mockResolvedValueOnce({ threads, next } as never);
+ vi.spyOn(client, 'queryThreadsAndHydrate').mockImplementationOnce((...args) =>
+ StreamChat.prototype.queryThreadsAndHydrate.apply(client, args),
+ );
+ };
+ const threadResponse = (
+ overrides: Parameters[0] = {},
+ ) => createGetThreadResponse(overrides).thread as ThreadStateResponse;
/** Replaces a loaded list through a real reload. */
const replaceList = async (manager: ThreadManager, threads: Thread[]) => {
const query = sinon.stub(manager.paginator, 'query').resolves({ items: threads });
@@ -2872,7 +2959,10 @@ describe('Threads 2.0', () => {
expect(client.threads.get(parentMessage.id)).to.equal(thread);
expect(client.threads.ensure({ channel, parentMessage })).to.equal(thread);
- await loadList(client.threads, [createThreadWithId(parentMessage.id)]);
+ respondFromServer([
+ threadResponse({ parentMessageOverrides: { id: parentMessage.id } }),
+ ]);
+ await client.threads.paginator.reload();
expect(listOf(client.threads)).to.have.lengthOf(1);
expect(listOf(client.threads)[0]).to.equal(thread);
});
@@ -3014,13 +3104,14 @@ describe('Threads 2.0', () => {
await loadList(client.threads, [listed]);
listed.state.partialNext({ isStateStale: true });
- await loadList(client.threads, [
- createTestThread({
+ respondFromServer([
+ threadResponse({
latest_replies: [kept],
parentMessageOverrides: { id },
reply_count: 2,
}),
]);
+ await client.threads.paginator.reload();
expect(listOf(client.threads)[0]).to.equal(listed);
expect(repliesOf(listed).map((r) => r.id)).to.deep.equal([kept.id]);
@@ -3046,13 +3137,14 @@ describe('Threads 2.0', () => {
listed.activate();
listed.state.partialNext({ isStateStale: true });
- await loadList(client.threads, [
- createTestThread({
+ respondFromServer([
+ threadResponse({
latest_replies: [kept],
parentMessageOverrides: { id },
reply_count: 2,
}),
]);
+ await client.threads.paginator.reload();
expect(listOf(client.threads)[0]).to.equal(listed);
expect(repliesOf(listed).map((r) => r.id)).to.deep.equal([older.id, kept.id]);
@@ -3176,9 +3268,10 @@ describe('Threads 2.0', () => {
it('reuses an active instance when a reload returns its thread', async () => {
const thread = createThreadWithId();
- const copy = createThreadWithId(thread.id);
thread.activate();
- stubbedQueryThreads.resolves({ threads: [copy], next: undefined });
+ respondFromServer([
+ threadResponse({ parentMessageOverrides: { id: thread.id } }),
+ ]);
await client.threads.reload({ force: true });
@@ -3188,10 +3281,11 @@ describe('Threads 2.0', () => {
it('reuses an active instance when the next page returns its thread', async () => {
const listed = createThreadWithId();
const thread = createThreadWithId();
- const copy = createThreadWithId(thread.id);
thread.activate();
setList(client.threads, [listed], 'cursor');
- stubbedQueryThreads.resolves({ threads: [copy], next: undefined });
+ respondFromServer([
+ threadResponse({ parentMessageOverrides: { id: thread.id } }),
+ ]);
await client.threads.paginator.toTail();
@@ -3202,8 +3296,10 @@ describe('Threads 2.0', () => {
client.threads.registerSubscriptions();
const thread = createThreadWithId();
const getThread = sinon
- .stub(client, 'getThreadAndHydrate')
- .resolves(createThreadWithId(thread.id));
+ .stub(client, 'getThread')
+ .resolves(
+ createGetThreadResponse({ parentMessageOverrides: { id: thread.id } }),
+ );
const releaseFirst = thread.activate();
releaseFirst();
@@ -3331,25 +3427,24 @@ describe('Threads 2.0', () => {
});
it('hydrates existing stale threads when reloading', async () => {
+ // `client.threads`: queried threads are matched against the client's registry.
const existingThread = createTestThread();
existingThread.state.partialNext({ isStateStale: true });
- const newThread = createTestThread({
- thread_participants: [
- { user_id: 'u1' },
- ] as ThreadStateResponse['thread_participants'],
- });
- setList(threadManager, [existingThread]);
- threadManager.state.partialNext({
- unseenThreadIds: [newThread.id],
- });
- stubbedQueryThreads.resolves({
- threads: [newThread],
- next: undefined,
- });
+ setList(client.threads, [existingThread]);
+ client.threads.state.partialNext({
+ unseenThreadIds: [existingThread.id],
+ });
+ respondFromServer([
+ threadResponse({
+ thread_participants: [
+ { user_id: 'u1' },
+ ] as ThreadStateResponse['thread_participants'],
+ }),
+ ]);
- await threadManager.reload();
+ await client.threads.reload();
- const threads = listOf(threadManager);
+ const threads = listOf(client.threads);
expect(threads).to.have.lengthOf(1);
expect(threads).to.contain(existingThread);
@@ -3593,6 +3688,77 @@ describe('Threads 2.0', () => {
expect(channel.watchStatus).to.equal(ChannelWatchStatus.Watching);
});
+ it('getThreadAndHydrate returns a registered thread as it is, unless it is stale and closed', async () => {
+ const r1 = makeReply({
+ id: 'r1',
+ created_at: convertDateToTimestamp('2020-01-01T00:00:01.000Z'),
+ });
+ const r2 = makeReply({
+ id: 'r2',
+ created_at: convertDateToTimestamp('2020-01-01T00:00:02.000Z'),
+ });
+ const thread = createTestThread({ latest_replies: [r1], reply_count: 1 });
+ subscribeThread(thread);
+ const { replyCount } = thread.state.getLatestValue();
+ const answer = createGetThreadResponse({
+ latest_replies: [r1, r2],
+ parentMessageOverrides: { reply_count: 2 },
+ reply_count: 2,
+ thread_participants: [
+ { user_id: 'p1' },
+ ] as ThreadStateResponse['thread_participants'],
+ });
+ vi.spyOn(client, 'getThread').mockResolvedValue(answer as never);
+
+ const release = thread.activate();
+ vi.spyOn(thread, 'reload').mockResolvedValue(undefined);
+ thread.state.partialNext({ isStateStale: true });
+ const open = await client.getThreadAndHydrate({ message_id: thread.id });
+
+ // Open: it reloads itself.
+ expect(open).to.equal(thread);
+ expect(thread.state.getLatestValue().replyCount).to.equal(replyCount);
+ expect(repliesOf(thread).map((reply) => reply.id)).to.eql(['r1']);
+ expect(thread.hasStaleState).to.equal(true);
+
+ release();
+ thread.state.partialNext({ isStateStale: false });
+ await client.getThreadAndHydrate({ message_id: thread.id });
+
+ // Current: events keep it so, and an older answer would roll it back.
+ expect(thread.state.getLatestValue().replyCount).to.equal(replyCount);
+ expect(thread.state.getLatestValue().participants).to.eql([]);
+ expect(repliesOf(thread).map((reply) => reply.id)).to.eql(['r1']);
+
+ thread.state.partialNext({ isStateStale: true });
+ const stale = await client.getThreadAndHydrate({ message_id: thread.id });
+
+ expect(stale).to.equal(thread);
+ expect(thread.state.getLatestValue().replyCount).to.equal(2);
+ expect(repliesOf(thread).map((reply) => reply.id)).to.eql(['r1', 'r2']);
+ expect(thread.hasStaleState).to.equal(false);
+ thread.unregisterSubscriptions();
+ });
+
+ it('getThreadAndHydrate applies the channel members and custom data to a registered thread it leaves as it is', async () => {
+ const thread = createTestThread();
+ subscribeThread(thread);
+ const answer = createGetThreadResponse({
+ channel: {
+ ...channelResponse,
+ custom: { name: 'Renamed' },
+ members: [{ user: { id: 'member' }, user_id: 'member' }],
+ } as ThreadStateResponse['channel'],
+ });
+ vi.spyOn(client, 'getThread').mockResolvedValue(answer as never);
+
+ await client.getThreadAndHydrate({ message_id: thread.id });
+
+ expect(thread.channel.state.members).to.have.property('member');
+ expect(thread.channel.data?.custom).to.deep.include({ name: 'Renamed' });
+ thread.unregisterSubscriptions();
+ });
+
it('queryThreadsAndHydrate marks the threads’ channels watched, unless asked not to watch', async () => {
vi.spyOn(client, 'queryThreads').mockResolvedValue({
threads: [threadResponse()],