Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 15 additions & 4 deletions ai-docs/v10-breaking-changes.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down Expand Up @@ -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)

<a id="thread-hydrate-state"></a>

### `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)

<a id="thread-state-derivation"></a>

### Thread state derivation: `replyCount`, `custom`, parent message via message store
Expand Down Expand Up @@ -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)

<a id="thread-mark-read"></a>

Expand Down
4 changes: 2 additions & 2 deletions ai-docs/v10-followups.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

<a id="simp-composer"></a>

Expand Down
12 changes: 5 additions & 7 deletions src/client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -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,
Expand All @@ -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;
Expand Down
19 changes: 4 additions & 15 deletions src/pagination/paginators/ThreadPaginator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -102,21 +102,10 @@ export class ThreadPaginator extends BasePaginator<Thread, QueryThreadsRequest>
PaginationQueryReturnValue<Thread>
> => {
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;
Expand Down
157 changes: 80 additions & 77 deletions src/thread.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<ThreadState>({
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;
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -410,71 +382,50 @@ 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),
isFirstPage: true,
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 = () => {
Expand Down Expand Up @@ -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<ThreadReadState>((state, userRead) => {
state[userRead.user.id] = {
Expand Down
Loading
Loading