From 3e898be95498bc4baeff9e5263bd2afd6d2ee254 Mon Sep 17 00:00:00 2001 From: nijeeshjoshy Date: Thu, 15 Oct 2026 00:15:07 +0200 Subject: [PATCH 1/4] feat: support scheduled messages (CHA-5934) Add scheduled_at to sendMessage options, the update/delete/send_now/query scheduled message endpoints, the scheduled_message.* events and a ScheduledMessageManager (client.scheduledMessages) that keeps the current user's scheduled messages in sync from the query, channel queries with include_scheduled_messages, and events. Co-Authored-By: Claude Opus 5.5 --- src/channel.ts | 26 +- src/client.ts | 67 ++++ src/errors.ts | 11 + src/events.ts | 5 + src/index.ts | 1 + .../ScheduledMessageManager.ts | 293 ++++++++++++++ src/scheduledMessages/index.ts | 1 + src/types.ts | 158 ++++++++ .../ScheduledMessageManager.test.ts | 369 ++++++++++++++++++ .../scheduled_messages.api.test.ts | 185 +++++++++ 10 files changed, 1112 insertions(+), 4 deletions(-) create mode 100644 src/scheduledMessages/ScheduledMessageManager.ts create mode 100644 src/scheduledMessages/index.ts create mode 100644 test/unit/scheduledMessages/ScheduledMessageManager.test.ts create mode 100644 test/unit/scheduledMessages/scheduled_messages.api.test.ts diff --git a/src/channel.ts b/src/channel.ts index 078f8af93c..039b3d8901 100644 --- a/src/channel.ts +++ b/src/channel.ts @@ -211,6 +211,8 @@ export class Channel { * @param {boolean} [options.pending] Make this message pending * @param {Record} [options.pending_message_metadata] Metadata for the pending message * @param {boolean} [options.force_moderation] Apply force moderation for server-side requests + * @param {string | Date} [options.scheduled_at] Do not send the message now but schedule it to be sent at this time. + * The response then carries `scheduled_message`, and `message` is not added to the channel. * * @return {Promise} The Server Response */ @@ -221,13 +223,25 @@ export class Channel { message: rawMessage, }); - return await client.post(this._channelURL() + '/message', { - message, - ...options, - }); + const response = await client.post( + this._channelURL() + '/message', + { + message, + ...options, + }, + ); + if (response.scheduled_message) { + client.scheduledMessages.upsertToState(response.scheduled_message); + } + return response; } async sendMessage(message: Message, options?: SendMessageOptions) { + // A scheduled send is not queued offline: replaying the queued task adds the + // response message to the channel state as if it had been sent. + if (options?.scheduled_at) { + return await this._sendMessage(message, options); + } try { const offlineDb = this.getClient().offlineDb; if (offlineDb) { @@ -1657,6 +1671,10 @@ export class Channel { this.getClient().polls.hydratePollCache(state.messages, true); this.getClient().reminders.hydrateState(state.messages); + this.getClient().scheduledMessages.hydrateChannelState( + this.cid, + state.scheduled_messages, + ); this.messageComposer.initStateFromChannelResponse(state); diff --git a/src/client.ts b/src/client.ts index 36da6c9285..f02a13840f 100644 --- a/src/client.ts +++ b/src/client.ts @@ -204,6 +204,8 @@ import type { QueryReactionsOptions, QueryRemindersOptions, QueryRemindersResponse, + QueryScheduledMessagesOptions, + QueryScheduledMessagesResponse, QuerySegmentsOptions, QuerySegmentTargetsFilter, QueryTeamUsageStatsOptions, @@ -225,6 +227,7 @@ import type { RequestOptions, ReviewFlagReportOptions, ReviewFlagReportResponse, + ScheduledMessageAPIResponse, SdkIdentifier, SearchAPIResponse, SearchMessageSortBase, @@ -266,6 +269,7 @@ import type { UpdatePollOptionAPIResponse, UpdatePredefinedFilterOptions, UpdateReminderOptions, + UpdateScheduledMessageOptions, UpdateSegmentData, UpdateUserGroupOptions, UpdateUserGroupResponse, @@ -294,6 +298,7 @@ import { ChannelManager } from './channel_manager'; import { MessageDeliveryReporter } from './messageDelivery'; import { NotificationManager } from './notifications'; import { ReminderManager } from './reminders'; +import { ScheduledMessageManager } from './scheduledMessages'; import { StateStore } from './store'; import type { MessageComposer } from './messageComposer'; import type { AbstractOfflineDB } from './offline-support'; @@ -350,6 +355,7 @@ export class StreamChat { offlineDb?: AbstractOfflineDB; notifications: NotificationManager; reminders: ReminderManager; + scheduledMessages: ScheduledMessageManager; anonymous: boolean; persistUserOnConnectionFailure?: boolean; axiosInstance: AxiosInstance; @@ -586,6 +592,7 @@ export class StreamChat { this.threads = new ThreadManager({ client: this }); this.polls = new PollManager({ client: this }); this.reminders = new ReminderManager({ client: this }); + this.scheduledMessages = new ScheduledMessageManager({ client: this }); this.messageDeliveryReporter = new MessageDeliveryReporter({ client: this }); this.messageComposerCache = new FixedSizeQueueCache(64); } @@ -1135,6 +1142,8 @@ export class StreamChat { this.state = new ClientState({ client: this }); // reset thread manager this.threads.resetState(); + // scheduled messages are private to the user + this.scheduledMessages.resetState(); this.uploadManager.reset(); this.messageComposerCache.clear(); @@ -5244,6 +5253,64 @@ export class StreamChat { }); } + /** + * updateScheduledMessage - Edits a scheduled message. Only messages with status + * `scheduled` can be edited. Messages are scheduled with `channel.sendMessage(message, { scheduled_at })`. + * + * @param {string} id The ID of the scheduled message + * @param {UpdateScheduledMessageOptions} options `message` replaces the scheduled message as a whole; `scheduled_at` moves the send time. At least one is required. + * @returns {Promise} The scheduled message as stored after the change + */ + async updateScheduledMessage(id: string, options: UpdateScheduledMessageOptions) { + return await this.patch( + `${this.baseURL}/scheduled_messages/${encodeURIComponent(id)}`, + options, + ); + } + + /** + * deleteScheduledMessage - Cancels a scheduled message, or removes a failed one. + * + * @param {string} id The ID of the scheduled message + * @returns {Promise} The scheduled message as stored after the change + */ + async deleteScheduledMessage(id: string) { + return await this.delete( + `${this.baseURL}/scheduled_messages/${encodeURIComponent(id)}`, + ); + } + + /** + * sendScheduledMessageNow - Sends a scheduled message now instead of at its scheduled time. + * The send happens asynchronously: the `scheduled_message.sent` (or `scheduled_message.failed`) + * event reports the outcome. + * + * @param {string} id The ID of the scheduled message + * @returns {Promise} The scheduled message as stored after the change + */ + async sendScheduledMessageNow(id: string) { + return await this.post( + `${this.baseURL}/scheduled_messages/${encodeURIComponent(id)}/send_now`, + ); + } + + /** + * queryScheduledMessages - Lists the scheduled messages of the current user (or, server-side, of `user_id`). + * Pending messages come first, then failed ones, each ordered by `scheduled_at`. + * + * @param {QueryScheduledMessagesOptions} options filter (`cid`, `status`), sort (only `scheduled_at` ascending), limit (1-100, default 20) and `next` cursor + * @returns {Promise} + */ + async queryScheduledMessages({ sort, ...rest }: QueryScheduledMessagesOptions = {}) { + return await this.post( + `${this.baseURL}/scheduled_messages/query`, + { + ...rest, + sort: sort && normalizeQuerySort(sort), + }, + ); + } + /** * queryTeamUsageStats - Queries team-level usage statistics from the warehouse database * diff --git a/src/errors.ts b/src/errors.ts index 3688a656c1..70f4317a08 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -29,6 +29,17 @@ export const APIErrorCodes: Record '70': { name: 'ErrQueryChannelPermissions', retryable: false }, '71': { name: 'ErrTooManyConnections', retryable: true }, '99': { name: 'AppSuspendedError', retryable: false }, + // Scheduled messages. The response's `exception_fields.reason` carries the + // snake_case name of these errors too. + '114': { name: 'ScheduledMessagesDisabledError', retryable: false }, + '115': { name: 'ScheduledAtTooSoonError', retryable: false }, + '116': { name: 'ScheduledAtTooFarError', retryable: false }, + '117': { name: 'ScheduledLimitChannelError', retryable: false }, + '118': { name: 'ScheduledLimitUserError', retryable: false }, + '119': { name: 'ScheduledMessageIDTakenError', retryable: false }, + '120': { name: 'ScheduledMessageNotEditableError', retryable: false }, + '121': { name: 'ScheduledMessageSendingError', retryable: false }, + '122': { name: 'ScheduledMessageConflictError', retryable: true }, }; export type APIError = Error & { diff --git a/src/events.ts b/src/events.ts index 7f8805cd8d..e5b8617e78 100644 --- a/src/events.ts +++ b/src/events.ts @@ -45,6 +45,11 @@ export const EVENT_MAP = { 'reminder.created': true, 'reminder.deleted': true, 'reminder.updated': true, + 'scheduled_message.created': true, + 'scheduled_message.deleted': true, + 'scheduled_message.failed': true, + 'scheduled_message.sent': true, + 'scheduled_message.updated': true, 'thread.updated': true, 'typing.start': true, 'typing.stop': true, diff --git a/src/index.ts b/src/index.ts index 4723a09b6f..025008665d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -19,6 +19,7 @@ export * from './permissions'; export * from './poll'; export * from './poll_manager'; export * from './reminders'; +export * from './scheduledMessages'; export * from './search'; export * from './segment'; export * from './signing'; diff --git a/src/scheduledMessages/ScheduledMessageManager.ts b/src/scheduledMessages/ScheduledMessageManager.ts new file mode 100644 index 0000000000..ae77cb15fc --- /dev/null +++ b/src/scheduledMessages/ScheduledMessageManager.ts @@ -0,0 +1,293 @@ +import { StateStore } from '../store'; +import { WithSubscriptions } from '../utils/WithSubscriptions'; +import type { StreamChat } from '../client'; +import type { + Event, + EventTypes, + QueryScheduledMessagesResponse, + ScheduledMessageResponse, + UpdateScheduledMessageOptions, +} from '../types'; + +type ScheduledMessageId = string; + +/** + * Page size used when loading the current user's scheduled messages. It is the + * maximum the API allows, and one page holds every pending message of a user + * (the per-user limit of pending scheduled messages is at most 100). + */ +export const SCHEDULED_MESSAGES_LOAD_LIMIT = 100; + +export type ScheduledMessageEvent = Event & { + cid: string; + scheduled_message: ScheduledMessageResponse; + type: EventTypes; +}; + +export type ScheduledMessageManagerState = { + /** The current user's known scheduled messages, keyed by scheduled message ID. */ + scheduledMessages: Map; + /** Whether more scheduled messages exist than were loaded (see `loadMore`). */ + hasMore: boolean; + isLoading: boolean; + /** Whether `load` has completed at least once. */ + isLoaded: boolean; + lastQueryError?: Error; + /** Cursor of the next page of `load`. */ + next?: string; +}; + +export type ScheduledMessageManagerOptions = { + client: StreamChat; +}; + +const byScheduledAt = (a: ScheduledMessageResponse, b: ScheduledMessageResponse) => + new Date(a.scheduled_at).getTime() - new Date(b.scheduled_at).getTime() || + (a.id < b.id ? -1 : a.id > b.id ? 1 : 0); + +/** + * Keeps the current user's scheduled messages in a client-side store. + * + * The server does not include scheduled messages in channel state (unless a channel is + * queried with `include_scheduled_messages: true`), so the store is filled by `load` + * (up to 100 messages; call it once the user is connected) and kept in sync by the + * `scheduled_message.*` events once `registerSubscriptions` was called: + * - `scheduled_message.created` / `scheduled_message.updated`: upserted + * - `scheduled_message.failed`: upserted with status `failed` and its `failure_reason` + * - `scheduled_message.sent` / `scheduled_message.deleted`: removed + * + * After a connection recovery, a loaded store is reloaded, because events may have + * been missed while the connection was down. + */ +export class ScheduledMessageManager extends WithSubscriptions { + private client: StreamChat; + /** + * IDs of scheduled messages known to be sent or deleted. A scheduled message ID is + * never reused, so a later upsert of one of these comes from a response or event + * that was overtaken (e.g. the `sendScheduledMessageNow` response arriving after the + * `scheduled_message.sent` event) and must not bring the message back. + */ + private removedIds = new Set(); + state: StateStore; + + constructor({ client }: ScheduledMessageManagerOptions) { + super(); + this.client = client; + this.state = new StateStore(this.initialState); + } + + get initialState(): ScheduledMessageManagerState { + return { + scheduledMessages: new Map(), + hasMore: false, + isLoading: false, + isLoaded: false, + lastQueryError: undefined, + next: undefined, + }; + } + + // State API START // + get scheduledMessages() { + return this.state.getLatestValue().scheduledMessages; + } + + getFromState(id: ScheduledMessageId) { + return this.scheduledMessages.get(id); + } + + /** + * Returns the known scheduled messages of a channel (or, with `parentId`, of a thread), + * ordered by `scheduled_at`. + */ + getForChannel = (cid: string, { parentId }: { parentId?: string } = {}) => + Array.from(this.scheduledMessages.values()) + .filter( + (sm) => + sm.cid === cid && + (parentId === undefined || (sm.message.parent_id ?? undefined) === parentId), + ) + .sort(byScheduledAt); + + upsertToState = (scheduledMessage: ScheduledMessageResponse) => { + if (!this.client._cacheEnabled()) return; + // Terminal states are reported by the delete endpoint and by the sent event; + // such messages no longer exist on the server. + if (scheduledMessage.status === 'canceled' || scheduledMessage.status === 'sent') { + this.removeFromState(scheduledMessage.id); + return; + } + if (this.removedIds.has(scheduledMessage.id)) return; + const scheduledMessages = new Map(this.scheduledMessages); + scheduledMessages.set(scheduledMessage.id, scheduledMessage); + this.state.partialNext({ scheduledMessages }); + }; + + removeFromState = (id: ScheduledMessageId) => { + this.removedIds.add(id); + if (!this.scheduledMessages.has(id)) return; + const scheduledMessages = new Map(this.scheduledMessages); + scheduledMessages.delete(id); + this.state.partialNext({ scheduledMessages }); + }; + + /** + * Replaces the known scheduled messages of a channel with the authoritative list a + * channel query returned (`include_scheduled_messages: true`). Nothing happens when + * the list is absent, which means the server did not know it. + */ + hydrateChannelState = ( + cid: string, + scheduledMessages: ScheduledMessageResponse[] | undefined, + ) => { + if (!this.client._cacheEnabled() || !Array.isArray(scheduledMessages)) return; + const next = new Map(this.scheduledMessages); + for (const [id, sm] of next) { + if (sm.cid === cid) next.delete(id); + } + for (const sm of scheduledMessages) { + if (!this.removedIds.has(sm.id)) next.set(sm.id, sm); + } + this.state.partialNext({ scheduledMessages: next }); + }; + + resetState = () => { + this.removedIds.clear(); + this.state.next(this.initialState); + }; + // State API END // + + // WS event handling START // + static isScheduledMessageEvent = (event: Event): event is ScheduledMessageEvent => + !!event.scheduled_message && event.type.startsWith('scheduled_message.'); + + public registerSubscriptions = () => { + if (this.hasSubscriptions) return; + this.addUnsubscribeFunction(this.subscribeUpsertEvents()); + this.addUnsubscribeFunction(this.subscribeRemoveEvents()); + this.addUnsubscribeFunction(this.subscribeConnectionRecovered()); + }; + + private subscribeUpsertEvents = () => { + const subscriptions = ( + [ + 'scheduled_message.created', + 'scheduled_message.updated', + 'scheduled_message.failed', + ] as const + ).map( + (type) => + this.client.on(type, (event) => { + if (!ScheduledMessageManager.isScheduledMessageEvent(event)) return; + this.upsertToState(event.scheduled_message); + }).unsubscribe, + ); + return () => subscriptions.forEach((unsubscribe) => unsubscribe()); + }; + + private subscribeRemoveEvents = () => { + const subscriptions = ( + ['scheduled_message.sent', 'scheduled_message.deleted'] as const + ).map( + (type) => + this.client.on(type, (event) => { + if (!ScheduledMessageManager.isScheduledMessageEvent(event)) return; + this.removeFromState(event.scheduled_message.id); + }).unsubscribe, + ); + return () => subscriptions.forEach((unsubscribe) => unsubscribe()); + }; + + private subscribeConnectionRecovered = () => + this.client.on('connection.recovered', () => { + if (!this.state.getLatestValue().isLoaded) return; + void this.load().catch((error) => { + this.client.logger('warn', 'scheduledMessages: reload after recovery failed', { + tags: ['scheduled_messages'], + error, + }); + }); + }).unsubscribe; + // WS event handling END // + + // API calls START // + /** + * Loads the current user's scheduled messages (up to 100) and replaces the store with + * them. Call it once the user is connected. + */ + load = async () => { + this.state.partialNext({ isLoading: true }); + try { + const response = await this.client.queryScheduledMessages({ + limit: SCHEDULED_MESSAGES_LOAD_LIMIT, + }); + const scheduledMessages = new Map( + response.scheduled_messages + .filter((sm) => !this.removedIds.has(sm.id)) + .map((sm) => [sm.id, sm] as const), + ); + this.state.partialNext({ + scheduledMessages: this.client._cacheEnabled() ? scheduledMessages : new Map(), + hasMore: !!response.next, + isLoaded: true, + lastQueryError: undefined, + next: response.next, + }); + return response; + } catch (error) { + this.state.partialNext({ lastQueryError: error as Error }); + throw error; + } finally { + this.state.partialNext({ isLoading: false }); + } + }; + + /** + * Loads the next page after `load` and adds it to the store. + */ + loadMore = async (): Promise => { + const { next, isLoading } = this.state.getLatestValue(); + if (!next || isLoading) return; + this.state.partialNext({ isLoading: true }); + try { + const response = await this.client.queryScheduledMessages({ + limit: SCHEDULED_MESSAGES_LOAD_LIMIT, + next, + }); + response.scheduled_messages.forEach(this.upsertToState); + this.state.partialNext({ + hasMore: !!response.next, + lastQueryError: undefined, + next: response.next, + }); + return response; + } catch (error) { + this.state.partialNext({ lastQueryError: error as Error }); + throw error; + } finally { + this.state.partialNext({ isLoading: false }); + } + }; + + updateScheduledMessage = async ( + id: ScheduledMessageId, + options: UpdateScheduledMessageOptions, + ) => { + const { scheduled_message } = await this.client.updateScheduledMessage(id, options); + this.upsertToState(scheduled_message); + return scheduled_message; + }; + + deleteScheduledMessage = async (id: ScheduledMessageId) => { + const { scheduled_message } = await this.client.deleteScheduledMessage(id); + this.removeFromState(id); + return scheduled_message; + }; + + sendScheduledMessageNow = async (id: ScheduledMessageId) => { + const { scheduled_message } = await this.client.sendScheduledMessageNow(id); + this.upsertToState(scheduled_message); + return scheduled_message; + }; + // API calls END // +} diff --git a/src/scheduledMessages/index.ts b/src/scheduledMessages/index.ts new file mode 100644 index 0000000000..4ab5083689 --- /dev/null +++ b/src/scheduledMessages/index.ts @@ -0,0 +1 @@ +export * from './ScheduledMessageManager'; diff --git a/src/types.ts b/src/types.ts index 11a62c3f85..bd139ba7f4 100644 --- a/src/types.ts +++ b/src/types.ts @@ -398,6 +398,13 @@ export type ChannelAPIResponse = { watcher_count?: number; watchers?: UserResponse[]; active_live_locations?: SharedLocationResponse[]; + /** + * The current user's own scheduled, sending and failed messages in this channel. + * Set only when the channel was queried with `include_scheduled_messages: true`. + * Absent when not requested or not known (e.g. more than 100); use + * `client.queryScheduledMessages` then. + */ + scheduled_messages?: ScheduledMessageResponse[]; }; export type ChannelUpdateOptions = { @@ -1007,6 +1014,12 @@ export type SendFileAPIResponse = APIResponse & { file: string; thumb_url?: stri export type SendMessageAPIResponse = APIResponse & { message: MessageResponse; pending_message_metadata?: Record | null; + /** + * Set only when the message was sent with `scheduled_at`. `message` then holds the + * content as it will be sent; it is not in the channel and its `id` is the + * scheduled message ID, not the ID the message gets when sent. + */ + scheduled_message?: ScheduledMessageResponse; }; export type SyncResponse = APIResponse & { @@ -1191,6 +1204,11 @@ export type ChannelQueryOptions = { created_by_id?: UserResponse['id']; data?: ChannelResponse; hide_for_creator?: boolean; + /** + * Client-side only. Adds the current user's scheduled messages in this channel + * to the response (`scheduled_messages`). + */ + include_scheduled_messages?: boolean; members?: PaginationOptions; messages?: MessagePaginationOptions; presence?: boolean; @@ -1752,6 +1770,7 @@ export type Event = CustomEventData & { reaction?: ReactionResponse; received_at?: string | Date; reminder?: ReminderResponse; + scheduled_message?: ScheduledMessageResponse; shadow?: boolean; team?: string; thread?: ThreadResponse; @@ -3058,6 +3077,12 @@ export type SendMessageOptions = { keep_channel_hidden?: boolean; pending?: boolean; pending_message_metadata?: Record; + /** + * When set, the message is not sent now: it is stored and sent at this time. + * Must be at least 1 minute and at most the app's maximum lead time from now. + * The response then carries `scheduled_message`. Cannot be combined with `pending`. + */ + scheduled_at?: string | Date; skip_enrich_url?: boolean; skip_push?: boolean; }; @@ -3578,6 +3603,7 @@ export type APIErrorResponse = { StatusCode: number; code?: number; details?: ErrorResponseDetails; + exception_fields?: Record; }; export class ErrorFromResponse extends Error { @@ -4857,6 +4883,138 @@ export type QueryRemindersResponse = { next?: string; }; +/** + * Scheduled messages + */ + +/** + * Status of a scheduled message. + * - `scheduled`: waiting for its send time; can be edited, sent now or deleted + * - `sending`: being sent (or a send was tried and is waiting to be retried); cannot change + * - `failed`: the send failed (see `failure_reason`); can only be deleted + * - `canceled`: deleted by its owner (reported by the delete endpoint and event only) + * - `sent`: sent (reported by the `scheduled_message.sent` event only) + */ +export type ScheduledMessageStatus = + | 'scheduled' + | 'sending' + | 'failed' + | 'canceled' + | 'sent'; + +/** Why a scheduled message failed to send. Only set when the status is `failed`. */ +export type ScheduledMessageFailureReason = + | 'banned' + | 'no_permission' + | 'channel_frozen' + | 'channel_deleted' + | 'user_deleted' + | 'moderation_blocked' + | 'hook_rejected' + | 'expired' + | 'internal_error' + | 'invalid_message' + | 'pending_not_supported'; + +/** The location a scheduled message will share. */ +export type ScheduledMessageLocation = { + created_by_device_id: string; + end_at?: string; + latitude?: number; + longitude?: number; +}; + +/** + * The message a scheduled message will send. It has no `id`: the ID of the + * sent message is only revealed by the `scheduled_message.sent` event + * (`message_id`). `mentioned_users` holds user IDs, not user objects. + * Custom message fields are returned at the top level, as on messages. + */ +export type ScheduledMessageContent = CustomMessageData & { + text: string; + attachments?: Attachment[]; + html?: string; + mentioned_channel?: boolean; + mentioned_group_ids?: string[]; + mentioned_here?: boolean; + mentioned_roles?: string[]; + mentioned_users?: string[]; + mml?: string; + parent_id?: string; + pin_expires?: string; + pinned?: boolean; + poll_id?: string; + quoted_message_id?: string; + restricted_visibility?: string[]; + shared_location?: ScheduledMessageLocation; + show_in_channel?: boolean; + silent?: boolean; + type?: MessageLabel; +}; + +/** Enough of a channel to list a scheduled message without loading the channel. */ +export type ScheduledMessageChannelResponse = { + cid: string; + id: string; + type: string; + image?: string; + name?: string; +}; + +export type ScheduledMessageResponse = { + /** Public handle of the scheduled message. Not the ID the message gets when sent. */ + id: string; + /** CID of the channel the message will be sent to. */ + cid: string; + created_at: string; + message: ScheduledMessageContent; + /** Date/time (ISO string) the message is scheduled to be sent. */ + scheduled_at: string; + status: ScheduledMessageStatus; + updated_at: string; + /** The user who scheduled the message. */ + user: UserResponse; + /** Set only in `queryScheduledMessages` results. */ + channel?: ScheduledMessageChannelResponse; + /** Set only when `status` is `failed`. */ + failure_reason?: ScheduledMessageFailureReason; +}; + +export type ScheduledMessageAPIResponse = APIResponse & { + scheduled_message: ScheduledMessageResponse; +}; + +export type UpdateScheduledMessageOptions = { + /** Replaces the scheduled message as a whole. */ + message?: Message; + /** New send time. Must be at least 1 minute and at most the app's maximum lead time from now. */ + scheduled_at?: string | Date; +}; + +export type ScheduledMessageFilters = { + /** Only scheduled messages of this channel. */ + cid?: string; + /** Only these statuses. */ + status?: Array>; +}; + +/** Only `scheduled_at` ascending is supported. */ +export type ScheduledMessageSort = Sort<{ scheduled_at: AscDesc }>; + +export type QueryScheduledMessagesOptions = Omit & { + filter?: ScheduledMessageFilters; + sort?: ScheduledMessageSort | ScheduledMessageSort[]; + /** Server-side only: the user whose scheduled messages to list. */ + user_id?: string; +}; + +export type QueryScheduledMessagesResponse = APIResponse & { + /** Pending messages first (`scheduled`, `sending`), then `failed` ones, each by `scheduled_at`. */ + scheduled_messages: ScheduledMessageResponse[]; + /** Cursor of the next page; absent on the last page. */ + next?: string; +}; + export type UserGroupMemberResponse = { group_id: string; user_id: string; diff --git a/test/unit/scheduledMessages/ScheduledMessageManager.test.ts b/test/unit/scheduledMessages/ScheduledMessageManager.test.ts new file mode 100644 index 0000000000..9f312ede04 --- /dev/null +++ b/test/unit/scheduledMessages/ScheduledMessageManager.test.ts @@ -0,0 +1,369 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { + EventTypes, + ScheduledMessageManager, + ScheduledMessageResponse, + StreamChat, +} from '../../../src'; +import { getClientWithUser } from '../test-utils/getClient'; + +const cid = 'messaging:general'; +const otherCid = 'messaging:random'; + +const generateScheduledMessage = ( + data: Partial = {}, +): ScheduledMessageResponse => { + const now = new Date().toISOString(); + return { + id: `sm-${Math.random().toString(36).slice(2)}`, + cid, + user: { id: 'user1' }, + message: { text: 'see you tomorrow' }, + scheduled_at: new Date(Date.now() + 60 * 60 * 1000).toISOString(), + status: 'scheduled', + created_at: now, + updated_at: now, + ...data, + }; +}; + +const generateEvent = ( + type: EventTypes, + scheduled_message: ScheduledMessageResponse, + extra: Record = {}, +) => ({ + type, + cid: scheduled_message.cid, + created_at: new Date().toISOString(), + user: scheduled_message.user, + scheduled_message, + ...extra, +}); + +const setup = () => { + const client = getClientWithUser({ id: 'user1' }) as StreamChat; + const manager = client.scheduledMessages; + return { client, manager }; +}; + +describe('ScheduledMessageManager', () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('is created by the client with an empty store', () => { + const { manager } = setup(); + expect(manager).toBeInstanceOf(ScheduledMessageManager); + expect(manager.scheduledMessages.size).toBe(0); + expect(manager.state.getLatestValue()).toEqual( + expect.objectContaining({ hasMore: false, isLoaded: false, isLoading: false }), + ); + }); + + describe('state', () => { + it('upserts and removes scheduled messages', () => { + const { manager } = setup(); + const sm = generateScheduledMessage(); + manager.upsertToState(sm); + expect(manager.getFromState(sm.id)).toBe(sm); + + const edited = { ...sm, message: { text: 'edited' } }; + manager.upsertToState(edited); + expect(manager.getFromState(sm.id)).toBe(edited); + expect(manager.scheduledMessages.size).toBe(1); + + manager.removeFromState(sm.id); + expect(manager.getFromState(sm.id)).toBeUndefined(); + }); + + it('removes instead of storing canceled and sent scheduled messages', () => { + const { manager } = setup(); + const canceled = generateScheduledMessage(); + const sent = generateScheduledMessage(); + manager.upsertToState(canceled); + manager.upsertToState(sent); + + manager.upsertToState({ ...canceled, status: 'canceled' }); + manager.upsertToState({ ...sent, status: 'sent' }); + + expect(manager.scheduledMessages.size).toBe(0); + }); + + it('does not bring back a removed scheduled message', () => { + const { manager } = setup(); + const sm = generateScheduledMessage(); + manager.upsertToState(sm); + manager.removeFromState(sm.id); + + // e.g. the send_now response arriving after the scheduled_message.sent event + manager.upsertToState(sm); + + expect(manager.getFromState(sm.id)).toBeUndefined(); + }); + + it('does not store anything when the cache is disabled', () => { + const { client, manager } = setup(); + vi.spyOn(client, '_cacheEnabled').mockReturnValue(false); + manager.upsertToState(generateScheduledMessage()); + manager.hydrateChannelState(cid, [generateScheduledMessage()]); + expect(manager.scheduledMessages.size).toBe(0); + }); + + it('returns the scheduled messages of a channel or thread ordered by scheduled_at', () => { + const { manager } = setup(); + const later = generateScheduledMessage({ scheduled_at: '2030-01-02T00:00:00Z' }); + const sooner = generateScheduledMessage({ scheduled_at: '2030-01-01T00:00:00Z' }); + const reply = generateScheduledMessage({ + scheduled_at: '2030-01-03T00:00:00Z', + message: { text: 'reply', parent_id: 'parent' }, + }); + const elsewhere = generateScheduledMessage({ cid: otherCid }); + [later, reply, elsewhere, sooner].forEach(manager.upsertToState); + + expect(manager.getForChannel(cid)).toEqual([sooner, later, reply]); + expect(manager.getForChannel(cid, { parentId: 'parent' })).toEqual([reply]); + expect(manager.getForChannel(otherCid)).toEqual([elsewhere]); + }); + + it('replaces the scheduled messages of a channel when hydrating channel state', () => { + const { manager } = setup(); + const stale = generateScheduledMessage(); + const elsewhere = generateScheduledMessage({ cid: otherCid }); + manager.upsertToState(stale); + manager.upsertToState(elsewhere); + + const fresh = generateScheduledMessage(); + manager.hydrateChannelState(cid, [fresh]); + + expect(manager.getForChannel(cid)).toEqual([fresh]); + expect(manager.getFromState(elsewhere.id)).toBe(elsewhere); + }); + + it('keeps the store when the channel response has no scheduled messages field', () => { + const { manager } = setup(); + const sm = generateScheduledMessage(); + manager.upsertToState(sm); + + manager.hydrateChannelState(cid, undefined); + + expect(manager.getFromState(sm.id)).toBe(sm); + }); + + it('empties the channel when the channel response has an empty list', () => { + const { manager } = setup(); + manager.upsertToState(generateScheduledMessage()); + + manager.hydrateChannelState(cid, []); + + expect(manager.getForChannel(cid)).toEqual([]); + }); + + it('is reset when the user disconnects', async () => { + const { client, manager } = setup(); + manager.upsertToState(generateScheduledMessage()); + vi.spyOn(client, 'closeConnection').mockResolvedValue(undefined); + + await client.disconnectUser(); + + expect(manager.scheduledMessages.size).toBe(0); + }); + }); + + describe('events', () => { + it.each(['scheduled_message.created', 'scheduled_message.updated'] as EventTypes[])( + 'upserts on %s', + (type) => { + const { client, manager } = setup(); + manager.registerSubscriptions(); + const sm = generateScheduledMessage(); + + client.dispatchEvent(generateEvent(type, sm)); + + expect(manager.getFromState(sm.id)).toEqual(sm); + }, + ); + + it('marks the scheduled message failed on scheduled_message.failed', () => { + const { client, manager } = setup(); + manager.registerSubscriptions(); + const sm = generateScheduledMessage(); + manager.upsertToState(sm); + + client.dispatchEvent( + generateEvent('scheduled_message.failed', { + ...sm, + status: 'failed', + failure_reason: 'channel_frozen', + }), + ); + + expect(manager.getFromState(sm.id)).toEqual( + expect.objectContaining({ status: 'failed', failure_reason: 'channel_frozen' }), + ); + }); + + it('removes on scheduled_message.sent', () => { + const { client, manager } = setup(); + manager.registerSubscriptions(); + const sm = generateScheduledMessage(); + manager.upsertToState(sm); + + client.dispatchEvent( + generateEvent( + 'scheduled_message.sent', + { ...sm, status: 'sent' }, + { message_id: 'final-message-id' }, + ), + ); + + expect(manager.getFromState(sm.id)).toBeUndefined(); + }); + + it('removes on scheduled_message.deleted', () => { + const { client, manager } = setup(); + manager.registerSubscriptions(); + const sm = generateScheduledMessage(); + manager.upsertToState(sm); + + client.dispatchEvent( + generateEvent('scheduled_message.deleted', { ...sm, status: 'canceled' }), + ); + + expect(manager.getFromState(sm.id)).toBeUndefined(); + }); + + it('ignores scheduled message events before registerSubscriptions and after unregistering', () => { + const { client, manager } = setup(); + const sm = generateScheduledMessage(); + + client.dispatchEvent(generateEvent('scheduled_message.created', sm)); + expect(manager.getFromState(sm.id)).toBeUndefined(); + + manager.registerSubscriptions(); + manager.unregisterSubscriptions(); + client.dispatchEvent(generateEvent('scheduled_message.created', sm)); + expect(manager.getFromState(sm.id)).toBeUndefined(); + }); + + it('reloads a loaded store on connection.recovered', async () => { + const { client, manager } = setup(); + manager.registerSubscriptions(); + const querySpy = vi + .spyOn(client, 'queryScheduledMessages') + .mockResolvedValue({ scheduled_messages: [], duration: '' }); + + client.dispatchEvent({ type: 'connection.recovered' }); + expect(querySpy).not.toHaveBeenCalled(); + + await manager.load(); + client.dispatchEvent({ type: 'connection.recovered' }); + expect(querySpy).toHaveBeenCalledTimes(2); + }); + }); + + describe('API', () => { + it('loads up to 100 scheduled messages and replaces the store', async () => { + const { client, manager } = setup(); + const stale = generateScheduledMessage(); + manager.upsertToState(stale); + const loaded = [generateScheduledMessage(), generateScheduledMessage()]; + const querySpy = vi + .spyOn(client, 'queryScheduledMessages') + .mockResolvedValue({ scheduled_messages: loaded, next: 'cursor', duration: '' }); + + await manager.load(); + + expect(querySpy).toHaveBeenCalledWith({ limit: 100 }); + expect(Array.from(manager.scheduledMessages.values())).toEqual(loaded); + expect(manager.state.getLatestValue()).toEqual( + expect.objectContaining({ + hasMore: true, + isLoaded: true, + isLoading: false, + next: 'cursor', + }), + ); + }); + + it('loads the next page and adds it to the store', async () => { + const { client, manager } = setup(); + const first = generateScheduledMessage(); + const second = generateScheduledMessage(); + const querySpy = vi + .spyOn(client, 'queryScheduledMessages') + .mockResolvedValueOnce({ + scheduled_messages: [first], + next: 'cursor', + duration: '', + }) + .mockResolvedValueOnce({ scheduled_messages: [second], duration: '' }); + + await manager.load(); + await manager.loadMore(); + await manager.loadMore(); // no next cursor: no request + + expect(querySpy).toHaveBeenCalledTimes(2); + expect(querySpy).toHaveBeenLastCalledWith({ limit: 100, next: 'cursor' }); + expect(manager.scheduledMessages.size).toBe(2); + expect(manager.state.getLatestValue().hasMore).toBe(false); + }); + + it('keeps the error of a failed load', async () => { + const { client, manager } = setup(); + const error = new Error('boom'); + vi.spyOn(client, 'queryScheduledMessages').mockRejectedValue(error); + + await expect(manager.load()).rejects.toThrow('boom'); + + expect(manager.state.getLatestValue()).toEqual( + expect.objectContaining({ + isLoading: false, + isLoaded: false, + lastQueryError: error, + }), + ); + }); + + it('updates a scheduled message and stores the result', async () => { + const { client, manager } = setup(); + const sm = generateScheduledMessage(); + const updated = { ...sm, scheduled_at: '2030-01-01T00:00:00Z' }; + const spy = vi + .spyOn(client, 'updateScheduledMessage') + .mockResolvedValue({ scheduled_message: updated, duration: '' }); + + await manager.updateScheduledMessage(sm.id, { scheduled_at: updated.scheduled_at }); + + expect(spy).toHaveBeenCalledWith(sm.id, { scheduled_at: updated.scheduled_at }); + expect(manager.getFromState(sm.id)).toBe(updated); + }); + + it('deletes a scheduled message and removes it from the store', async () => { + const { client, manager } = setup(); + const sm = generateScheduledMessage(); + manager.upsertToState(sm); + vi.spyOn(client, 'deleteScheduledMessage').mockResolvedValue({ + scheduled_message: { ...sm, status: 'canceled' }, + duration: '', + }); + + await manager.deleteScheduledMessage(sm.id); + + expect(manager.getFromState(sm.id)).toBeUndefined(); + }); + + it('sends a scheduled message now and stores the result', async () => { + const { client, manager } = setup(); + const sm = generateScheduledMessage(); + const due = { ...sm, scheduled_at: new Date().toISOString() }; + vi.spyOn(client, 'sendScheduledMessageNow').mockResolvedValue({ + scheduled_message: due, + duration: '', + }); + + await manager.sendScheduledMessageNow(sm.id); + + expect(manager.getFromState(sm.id)).toBe(due); + }); + }); +}); diff --git a/test/unit/scheduledMessages/scheduled_messages.api.test.ts b/test/unit/scheduledMessages/scheduled_messages.api.test.ts new file mode 100644 index 0000000000..1d0473c9ff --- /dev/null +++ b/test/unit/scheduledMessages/scheduled_messages.api.test.ts @@ -0,0 +1,185 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { ScheduledMessageResponse, StreamChat } from '../../../src'; +import { APIErrorCodes } from '../../../src/errors'; +import { getClientWithUser } from '../test-utils/getClient'; +import { mockChannelQueryResponse } from '../test-utils/mockChannelQueryResponse'; + +const scheduledMessage: ScheduledMessageResponse = { + id: 'sm-1', + cid: 'messaging:general', + user: { id: 'user1' }, + message: { text: 'see you tomorrow', mood: 'happy' }, + scheduled_at: '2030-01-01T00:00:00Z', + status: 'scheduled', + created_at: '2026-10-15T00:00:00Z', + updated_at: '2026-10-15T00:00:00Z', +}; + +const setup = () => getClientWithUser({ id: 'user1' }) as StreamChat; + +describe('scheduled messages API', () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + describe('client', () => { + it('updateScheduledMessage sends a PATCH to the scheduled message', async () => { + const client = setup(); + const spy = vi + .spyOn(client, 'patch') + .mockResolvedValue({ scheduled_message: scheduledMessage }); + + await client.updateScheduledMessage('sm/1', { + message: { text: 'edited' }, + scheduled_at: '2030-01-01T00:00:00Z', + }); + + expect(spy).toHaveBeenCalledWith(`${client.baseURL}/scheduled_messages/sm%2F1`, { + message: { text: 'edited' }, + scheduled_at: '2030-01-01T00:00:00Z', + }); + }); + + it('deleteScheduledMessage sends a DELETE to the scheduled message', async () => { + const client = setup(); + const spy = vi + .spyOn(client, 'delete') + .mockResolvedValue({ scheduled_message: scheduledMessage }); + + await client.deleteScheduledMessage('sm-1'); + + expect(spy).toHaveBeenCalledWith(`${client.baseURL}/scheduled_messages/sm-1`); + }); + + it('sendScheduledMessageNow sends a POST to send_now', async () => { + const client = setup(); + const spy = vi + .spyOn(client, 'post') + .mockResolvedValue({ scheduled_message: scheduledMessage }); + + await client.sendScheduledMessageNow('sm-1'); + + expect(spy).toHaveBeenCalledWith( + `${client.baseURL}/scheduled_messages/sm-1/send_now`, + ); + }); + + it('queryScheduledMessages posts the filter, normalized sort and pagination', async () => { + const client = setup(); + const spy = vi + .spyOn(client, 'post') + .mockResolvedValue({ scheduled_messages: [scheduledMessage], next: 'n2' }); + + const response = await client.queryScheduledMessages({ + filter: { cid: 'messaging:general', status: ['scheduled', 'failed'] }, + sort: { scheduled_at: 1 }, + limit: 100, + next: 'n1', + }); + + expect(spy).toHaveBeenCalledWith(`${client.baseURL}/scheduled_messages/query`, { + filter: { cid: 'messaging:general', status: ['scheduled', 'failed'] }, + sort: [{ field: 'scheduled_at', direction: 1 }], + limit: 100, + next: 'n1', + }); + expect(response.scheduled_messages).toEqual([scheduledMessage]); + }); + + it('queryScheduledMessages works without options', async () => { + const client = setup(); + const spy = vi.spyOn(client, 'post').mockResolvedValue({ scheduled_messages: [] }); + + await client.queryScheduledMessages(); + + expect(spy).toHaveBeenCalledWith(`${client.baseURL}/scheduled_messages/query`, { + sort: undefined, + }); + }); + + it('knows the scheduled message error codes', () => { + expect(APIErrorCodes['115']).toEqual({ + name: 'ScheduledAtTooSoonError', + retryable: false, + }); + expect(APIErrorCodes['122']).toEqual({ + name: 'ScheduledMessageConflictError', + retryable: true, + }); + }); + }); + + describe('channel', () => { + it('sendMessage passes scheduled_at through and stores the scheduled message', async () => { + const client = setup(); + const channel = client.channel('messaging', 'general'); + const spy = vi.spyOn(client, 'post').mockResolvedValue({ + message: { id: 'sm-1', text: 'see you tomorrow' }, + scheduled_message: scheduledMessage, + }); + + const response = await channel.sendMessage( + { text: 'see you tomorrow' }, + { scheduled_at: '2030-01-01T00:00:00Z' }, + ); + + expect(spy).toHaveBeenCalledWith( + `${client.baseURL}/channels/messaging/general/message`, + { + message: { text: 'see you tomorrow' }, + scheduled_at: '2030-01-01T00:00:00Z', + }, + ); + expect(response.scheduled_message).toEqual(scheduledMessage); + expect(client.scheduledMessages.getFromState('sm-1')).toEqual(scheduledMessage); + expect(channel.state.messages).toHaveLength(0); + }); + + it('sendMessage does not queue a scheduled send in the offline DB', async () => { + const client = setup(); + const queueTask = vi.fn(); + // @ts-expect-error partial offline DB + client.offlineDb = { queueTask }; + const channel = client.channel('messaging', 'general'); + vi.spyOn(client, 'post').mockResolvedValue({ + message: { id: 'sm-1' }, + scheduled_message: scheduledMessage, + }); + + await channel.sendMessage( + { id: 'sm-1', text: 'see you tomorrow' }, + { scheduled_at: '2030-01-01T00:00:00Z' }, + ); + + expect(queueTask).not.toHaveBeenCalled(); + }); + + it('sendMessage without scheduled_at does not touch the scheduled messages store', async () => { + const client = setup(); + const channel = client.channel('messaging', 'general'); + vi.spyOn(client, 'post').mockResolvedValue({ message: { id: 'm-1' } }); + + await channel.sendMessage({ text: 'now' }); + + expect(client.scheduledMessages.scheduledMessages.size).toBe(0); + }); + + it('query passes include_scheduled_messages and hydrates the store', async () => { + const client = setup(); + const { cid, id, type } = mockChannelQueryResponse.channel; + const channel = client.channel(type, id); + const inChannel = { ...scheduledMessage, cid }; + const spy = vi.spyOn(client, 'post').mockResolvedValue({ + ...mockChannelQueryResponse, + scheduled_messages: [inChannel], + }); + + await channel.query({ include_scheduled_messages: true }); + + expect(spy.mock.calls[0][1]).toEqual( + expect.objectContaining({ include_scheduled_messages: true }), + ); + expect(client.scheduledMessages.getForChannel(cid)).toEqual([inChannel]); + }); + }); +}); From cac2635e34fbd2e3776a76793bdcf05a7df02478 Mon Sep 17 00:00:00 2001 From: nijeeshjoshy Date: Thu, 15 Oct 2026 04:34:16 +0200 Subject: [PATCH 2/4] docs: align scheduled message docs with backend changes Messages awaiting a send retry report status sending, pending: false is accepted with scheduled_at, and the scheduled send response hydrates mentioned_users. Co-Authored-By: Claude Opus 5.5 --- src/types.ts | 11 ++++++----- .../ScheduledMessageManager.test.ts | 14 ++++++++++++++ 2 files changed, 20 insertions(+), 5 deletions(-) diff --git a/src/types.ts b/src/types.ts index bd139ba7f4..5613d1022d 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1016,8 +1016,8 @@ export type SendMessageAPIResponse = APIResponse & { pending_message_metadata?: Record | null; /** * Set only when the message was sent with `scheduled_at`. `message` then holds the - * content as it will be sent; it is not in the channel and its `id` is the - * scheduled message ID, not the ID the message gets when sent. + * content as it will be sent (with hydrated `mentioned_users`); it is not in the + * channel and its `id` is the scheduled message ID, not the ID the message gets when sent. */ scheduled_message?: ScheduledMessageResponse; }; @@ -3080,7 +3080,7 @@ export type SendMessageOptions = { /** * When set, the message is not sent now: it is stored and sent at this time. * Must be at least 1 minute and at most the app's maximum lead time from now. - * The response then carries `scheduled_message`. Cannot be combined with `pending`. + * The response then carries `scheduled_message`. Cannot be combined with `pending: true`. */ scheduled_at?: string | Date; skip_enrich_url?: boolean; @@ -4890,7 +4890,8 @@ export type QueryRemindersResponse = { /** * Status of a scheduled message. * - `scheduled`: waiting for its send time; can be edited, sent now or deleted - * - `sending`: being sent (or a send was tried and is waiting to be retried); cannot change + * - `sending`: being sent, or a send was tried and is waiting to be retried; cannot be + * edited, sent now or deleted until the send finishes (409 `scheduled_message_sending`) * - `failed`: the send failed (see `failure_reason`); can only be deleted * - `canceled`: deleted by its owner (reported by the delete endpoint and event only) * - `sent`: sent (reported by the `scheduled_message.sent` event only) @@ -4994,7 +4995,7 @@ export type UpdateScheduledMessageOptions = { export type ScheduledMessageFilters = { /** Only scheduled messages of this channel. */ cid?: string; - /** Only these statuses. */ + /** Only these statuses, matched against the reported `status`. */ status?: Array>; }; diff --git a/test/unit/scheduledMessages/ScheduledMessageManager.test.ts b/test/unit/scheduledMessages/ScheduledMessageManager.test.ts index 9f312ede04..b9dd442666 100644 --- a/test/unit/scheduledMessages/ScheduledMessageManager.test.ts +++ b/test/unit/scheduledMessages/ScheduledMessageManager.test.ts @@ -183,6 +183,20 @@ describe('ScheduledMessageManager', () => { }, ); + it('keeps a message awaiting a send retry as sending', () => { + const { client, manager } = setup(); + manager.registerSubscriptions(); + const sm = generateScheduledMessage(); + manager.upsertToState(sm); + + client.dispatchEvent( + generateEvent('scheduled_message.updated', { ...sm, status: 'sending' }), + ); + + expect(manager.getFromState(sm.id)?.status).toBe('sending'); + expect(manager.getForChannel(cid)).toHaveLength(1); + }); + it('marks the scheduled message failed on scheduled_message.failed', () => { const { client, manager } = setup(); manager.registerSubscriptions(); From 8296921c1e16f77cfecc07a6cca61267175639e2 Mon Sep 17 00:00:00 2001 From: nijeeshjoshy Date: Thu, 15 Oct 2026 12:12:52 +0200 Subject: [PATCH 3/4] fix: keep scheduled message store ordered across responses and events - Ignore an upsert older (by updated_at) than the stored version, so the send-now response or its updated event cannot revert a failed message to scheduled. - load() and channel query hydration no longer undo changes that arrived while they were in flight. - Drop responses to requests made before disconnectUser(), so a new user never sees the previous user's scheduled messages. - getForChannel accepts parentId: null for messages outside threads. Co-Authored-By: Claude Opus 5.5 --- src/channel.ts | 8 +- .../ScheduledMessageManager.ts | 139 ++++++++++++--- .../ScheduledMessageManager.test.ts | 166 ++++++++++++++++++ .../scheduled_messages.api.test.ts | 23 +++ 4 files changed, 309 insertions(+), 27 deletions(-) diff --git a/src/channel.ts b/src/channel.ts index 039b3d8901..a70d5eac9a 100644 --- a/src/channel.ts +++ b/src/channel.ts @@ -223,6 +223,7 @@ export class Channel { message: rawMessage, }); + const scheduledMessagesToken = client.scheduledMessages.snapshotToken(); const response = await client.post( this._channelURL() + '/message', { @@ -231,7 +232,10 @@ export class Channel { }, ); if (response.scheduled_message) { - client.scheduledMessages.upsertToState(response.scheduled_message); + client.scheduledMessages.upsertResponseToState( + response.scheduled_message, + scheduledMessagesToken, + ); } return response; } @@ -1599,6 +1603,7 @@ export class Channel { ); } + const scheduledMessagesToken = this.getClient().scheduledMessages.snapshotToken(); let queryURL = `${this.getClient().baseURL}/channels/${encodeURIComponent( this.type, )}`; @@ -1674,6 +1679,7 @@ export class Channel { this.getClient().scheduledMessages.hydrateChannelState( this.cid, state.scheduled_messages, + scheduledMessagesToken, ); this.messageComposer.initStateFromChannelResponse(state); diff --git a/src/scheduledMessages/ScheduledMessageManager.ts b/src/scheduledMessages/ScheduledMessageManager.ts index ae77cb15fc..cd8623c043 100644 --- a/src/scheduledMessages/ScheduledMessageManager.ts +++ b/src/scheduledMessages/ScheduledMessageManager.ts @@ -41,6 +41,16 @@ export type ScheduledMessageManagerOptions = { client: StreamChat; }; +/** + * Taken before a request whose response replaces part of the store (see + * `snapshotToken`), so the response cannot undo what happened while it was in flight. + */ +export type ScheduledMessageSnapshotToken = { epoch: number; seq: number }; + +/** Whether `a` is a later version of the same scheduled message than `b`. */ +const isNewer = (a: ScheduledMessageResponse, b: ScheduledMessageResponse) => + new Date(a.updated_at).getTime() > new Date(b.updated_at).getTime(); + const byScheduledAt = (a: ScheduledMessageResponse, b: ScheduledMessageResponse) => new Date(a.scheduled_at).getTime() - new Date(b.scheduled_at).getTime() || (a.id < b.id ? -1 : a.id > b.id ? 1 : 0); @@ -58,6 +68,11 @@ const byScheduledAt = (a: ScheduledMessageResponse, b: ScheduledMessageResponse) * * After a connection recovery, a loaded store is reloaded, because events may have * been missed while the connection was down. + * + * Events and responses can arrive out of order, so an update older (by `updated_at`) + * than the stored version is ignored, and a list (`load`, a channel query) does not + * undo changes that arrived while it was being fetched. Responses to requests made + * before `disconnectUser` are dropped. */ export class ScheduledMessageManager extends WithSubscriptions { private client: StreamChat; @@ -68,6 +83,11 @@ export class ScheduledMessageManager extends WithSubscriptions { * `scheduled_message.sent` event) and must not bring the message back. */ private removedIds = new Set(); + /** Bumped by `resetState`: responses to requests made before then are dropped. */ + private epoch = 0; + /** Bumped on every store change; `touchedAt` records the value per ID. */ + private seq = 0; + private touchedAt = new Map(); state: StateStore; constructor({ client }: ScheduledMessageManagerOptions) { @@ -97,18 +117,72 @@ export class ScheduledMessageManager extends WithSubscriptions { } /** - * Returns the known scheduled messages of a channel (or, with `parentId`, of a thread), - * ordered by `scheduled_at`. + * Returns the known scheduled messages of a channel, ordered by `scheduled_at`. + * `parentId` narrows the list: a message ID returns the replies of that thread, + * `null` returns only the messages that are not thread replies, and leaving it out + * returns both. */ - getForChannel = (cid: string, { parentId }: { parentId?: string } = {}) => + getForChannel = (cid: string, { parentId }: { parentId?: string | null } = {}) => Array.from(this.scheduledMessages.values()) .filter( (sm) => sm.cid === cid && - (parentId === undefined || (sm.message.parent_id ?? undefined) === parentId), + (parentId === undefined || (sm.message.parent_id ?? null) === parentId), ) .sort(byScheduledAt); + /** + * Call before a request whose response is passed to `hydrateChannelState`. The + * response is then not applied over changes made while it was in flight, and not + * at all after `resetState` (the user disconnected). + */ + snapshotToken = (): ScheduledMessageSnapshotToken => ({ + epoch: this.epoch, + seq: this.seq, + }); + + private touch = (id: ScheduledMessageId) => { + this.touchedAt.set(id, ++this.seq); + }; + + /** + * Replaces the entries `inScope` selects with `snapshot`, keeping the entries that + * changed after `token` was taken where the snapshot is not newer. + */ + private applySnapshot = ( + snapshot: ScheduledMessageResponse[], + inScope: (sm: ScheduledMessageResponse) => boolean, + token: ScheduledMessageSnapshotToken, + ) => { + const changedSince = (id: ScheduledMessageId) => + (this.touchedAt.get(id) ?? 0) > token.seq; + const next = new Map(this.scheduledMessages); + for (const [id, sm] of next) { + if (inScope(sm) && !changedSince(id)) next.delete(id); + } + for (const sm of snapshot) { + if (this.removedIds.has(sm.id)) continue; + const current = next.get(sm.id); + if (current && isNewer(current, sm)) continue; + next.set(sm.id, sm); + } + this.state.partialNext({ scheduledMessages: next }); + }; + + /** + * Stores a scheduled message from an API response, like `upsertToState`. `token` + * comes from `snapshotToken`, taken before the request was sent: a response that + * arrives after the user disconnected belongs to the previous user and is dropped. + */ + upsertResponseToState = ( + scheduledMessage: ScheduledMessageResponse, + token: ScheduledMessageSnapshotToken, + ) => { + if (token.epoch !== this.epoch) return; + this.upsertToState(scheduledMessage); + }; + + /** Stores a scheduled message, unless a newer version of it is already stored. */ upsertToState = (scheduledMessage: ScheduledMessageResponse) => { if (!this.client._cacheEnabled()) return; // Terminal states are reported by the delete endpoint and by the sent event; @@ -118,6 +192,12 @@ export class ScheduledMessageManager extends WithSubscriptions { return; } if (this.removedIds.has(scheduledMessage.id)) return; + // Why: responses and events can arrive out of order, e.g. the send-now response + // or its `scheduled_message.updated` event after `scheduled_message.failed` + // (the send is queued before the response is written). + const current = this.scheduledMessages.get(scheduledMessage.id); + if (current && isNewer(current, scheduledMessage)) return; + this.touch(scheduledMessage.id); const scheduledMessages = new Map(this.scheduledMessages); scheduledMessages.set(scheduledMessage.id, scheduledMessage); this.state.partialNext({ scheduledMessages }); @@ -125,6 +205,7 @@ export class ScheduledMessageManager extends WithSubscriptions { removeFromState = (id: ScheduledMessageId) => { this.removedIds.add(id); + this.touch(id); if (!this.scheduledMessages.has(id)) return; const scheduledMessages = new Map(this.scheduledMessages); scheduledMessages.delete(id); @@ -134,24 +215,22 @@ export class ScheduledMessageManager extends WithSubscriptions { /** * Replaces the known scheduled messages of a channel with the authoritative list a * channel query returned (`include_scheduled_messages: true`). Nothing happens when - * the list is absent, which means the server did not know it. + * the list is absent, which means the server did not know it. `token` comes from + * `snapshotToken`, taken before the query was sent. */ hydrateChannelState = ( cid: string, scheduledMessages: ScheduledMessageResponse[] | undefined, + token: ScheduledMessageSnapshotToken = this.snapshotToken(), ) => { if (!this.client._cacheEnabled() || !Array.isArray(scheduledMessages)) return; - const next = new Map(this.scheduledMessages); - for (const [id, sm] of next) { - if (sm.cid === cid) next.delete(id); - } - for (const sm of scheduledMessages) { - if (!this.removedIds.has(sm.id)) next.set(sm.id, sm); - } - this.state.partialNext({ scheduledMessages: next }); + if (token.epoch !== this.epoch) return; + this.applySnapshot(scheduledMessages, (sm) => sm.cid === cid, token); }; resetState = () => { + this.epoch++; + this.touchedAt.clear(); this.removedIds.clear(); this.state.next(this.initialState); }; @@ -216,18 +295,18 @@ export class ScheduledMessageManager extends WithSubscriptions { * them. Call it once the user is connected. */ load = async () => { + const token = this.snapshotToken(); this.state.partialNext({ isLoading: true }); try { const response = await this.client.queryScheduledMessages({ limit: SCHEDULED_MESSAGES_LOAD_LIMIT, }); - const scheduledMessages = new Map( - response.scheduled_messages - .filter((sm) => !this.removedIds.has(sm.id)) - .map((sm) => [sm.id, sm] as const), - ); + // The user disconnected meanwhile: the response belongs to the previous user. + if (token.epoch !== this.epoch) return response; + if (this.client._cacheEnabled()) { + this.applySnapshot(response.scheduled_messages, () => true, token); + } this.state.partialNext({ - scheduledMessages: this.client._cacheEnabled() ? scheduledMessages : new Map(), hasMore: !!response.next, isLoaded: true, lastQueryError: undefined, @@ -235,10 +314,12 @@ export class ScheduledMessageManager extends WithSubscriptions { }); return response; } catch (error) { - this.state.partialNext({ lastQueryError: error as Error }); + if (token.epoch === this.epoch) { + this.state.partialNext({ lastQueryError: error as Error }); + } throw error; } finally { - this.state.partialNext({ isLoading: false }); + if (token.epoch === this.epoch) this.state.partialNext({ isLoading: false }); } }; @@ -248,12 +329,14 @@ export class ScheduledMessageManager extends WithSubscriptions { loadMore = async (): Promise => { const { next, isLoading } = this.state.getLatestValue(); if (!next || isLoading) return; + const { epoch } = this; this.state.partialNext({ isLoading: true }); try { const response = await this.client.queryScheduledMessages({ limit: SCHEDULED_MESSAGES_LOAD_LIMIT, next, }); + if (epoch !== this.epoch) return response; response.scheduled_messages.forEach(this.upsertToState); this.state.partialNext({ hasMore: !!response.next, @@ -262,10 +345,11 @@ export class ScheduledMessageManager extends WithSubscriptions { }); return response; } catch (error) { - this.state.partialNext({ lastQueryError: error as Error }); + if (epoch === this.epoch) + this.state.partialNext({ lastQueryError: error as Error }); throw error; } finally { - this.state.partialNext({ isLoading: false }); + if (epoch === this.epoch) this.state.partialNext({ isLoading: false }); } }; @@ -273,20 +357,23 @@ export class ScheduledMessageManager extends WithSubscriptions { id: ScheduledMessageId, options: UpdateScheduledMessageOptions, ) => { + const token = this.snapshotToken(); const { scheduled_message } = await this.client.updateScheduledMessage(id, options); - this.upsertToState(scheduled_message); + this.upsertResponseToState(scheduled_message, token); return scheduled_message; }; deleteScheduledMessage = async (id: ScheduledMessageId) => { + const { epoch } = this; const { scheduled_message } = await this.client.deleteScheduledMessage(id); - this.removeFromState(id); + if (epoch === this.epoch) this.removeFromState(id); return scheduled_message; }; sendScheduledMessageNow = async (id: ScheduledMessageId) => { + const token = this.snapshotToken(); const { scheduled_message } = await this.client.sendScheduledMessageNow(id); - this.upsertToState(scheduled_message); + this.upsertResponseToState(scheduled_message, token); return scheduled_message; }; // API calls END // diff --git a/test/unit/scheduledMessages/ScheduledMessageManager.test.ts b/test/unit/scheduledMessages/ScheduledMessageManager.test.ts index b9dd442666..37f5d3fc59 100644 --- a/test/unit/scheduledMessages/ScheduledMessageManager.test.ts +++ b/test/unit/scheduledMessages/ScheduledMessageManager.test.ts @@ -40,6 +40,14 @@ const generateEvent = ( ...extra, }); +const at = (second: number) => new Date(Date.UTC(2030, 0, 1, 0, 0, second)).toISOString(); + +const deferred = () => { + let resolve!: (value: T) => void; + const promise = new Promise((r) => (resolve = r)); + return { promise, resolve }; +}; + const setup = () => { const client = getClientWithUser({ id: 'user1' }) as StreamChat; const manager = client.scheduledMessages; @@ -122,6 +130,7 @@ describe('ScheduledMessageManager', () => { expect(manager.getForChannel(cid)).toEqual([sooner, later, reply]); expect(manager.getForChannel(cid, { parentId: 'parent' })).toEqual([reply]); + expect(manager.getForChannel(cid, { parentId: null })).toEqual([sooner, later]); expect(manager.getForChannel(otherCid)).toEqual([elsewhere]); }); @@ -380,4 +389,161 @@ describe('ScheduledMessageManager', () => { expect(manager.getFromState(sm.id)).toBe(due); }); }); + + describe('ordering', () => { + it('keeps a failed message failed when the send-now response arrives after the failed event', async () => { + const { client, manager } = setup(); + manager.registerSubscriptions(); + const sm = generateScheduledMessage({ updated_at: at(0) }); + manager.upsertToState(sm); + const response = deferred(); + vi.spyOn(client, 'post').mockReturnValue(response.promise as never); + + const sending = manager.sendScheduledMessageNow(sm.id); + client.dispatchEvent( + generateEvent('scheduled_message.failed', { + ...sm, + status: 'failed', + failure_reason: 'channel_frozen', + updated_at: at(20), + }), + ); + response.resolve({ + scheduled_message: { ...sm, scheduled_at: at(10), updated_at: at(10) }, + }); + await sending; + + expect(manager.getFromState(sm.id)?.status).toBe('failed'); + }); + + it('ignores an updated event older than the stored version', () => { + const { client, manager } = setup(); + manager.registerSubscriptions(); + const sm = generateScheduledMessage({ updated_at: at(0) }); + client.dispatchEvent( + generateEvent('scheduled_message.failed', { + ...sm, + status: 'failed', + updated_at: at(20), + }), + ); + client.dispatchEvent( + generateEvent('scheduled_message.updated', { ...sm, updated_at: at(10) }), + ); + + expect(manager.getFromState(sm.id)?.status).toBe('failed'); + }); + + it('applies an update with the same updated_at', () => { + const { manager } = setup(); + const sm = generateScheduledMessage({ updated_at: at(0) }); + manager.upsertToState(sm); + manager.upsertToState({ ...sm, status: 'sending' }); + + expect(manager.getFromState(sm.id)?.status).toBe('sending'); + }); + + it('keeps a message scheduled while load is in flight', async () => { + const { client, manager } = setup(); + const response = deferred(); + vi.spyOn(client, 'post').mockReturnValue(response.promise as never); + const existing = generateScheduledMessage(); + const scheduledMeanwhile = generateScheduledMessage(); + + const loading = manager.load(); + manager.upsertToState(scheduledMeanwhile); + response.resolve({ scheduled_messages: [existing] }); + await loading; + + expect(manager.getFromState(existing.id)).toEqual(existing); + expect(manager.getFromState(scheduledMeanwhile.id)).toEqual(scheduledMeanwhile); + }); + + it('keeps a failure received while load is in flight', async () => { + const { client, manager } = setup(); + manager.registerSubscriptions(); + const sm = generateScheduledMessage({ status: 'sending', updated_at: at(5) }); + manager.upsertToState(sm); + const response = deferred(); + vi.spyOn(client, 'post').mockReturnValue(response.promise as never); + + const loading = manager.load(); + client.dispatchEvent( + generateEvent('scheduled_message.failed', { + ...sm, + status: 'failed', + updated_at: at(20), + }), + ); + response.resolve({ scheduled_messages: [sm] }); + await loading; + + expect(manager.getFromState(sm.id)?.status).toBe('failed'); + }); + + it('still drops a message missing from the load when it did not change meanwhile', async () => { + const { client, manager } = setup(); + const gone = generateScheduledMessage(); + manager.upsertToState(gone); + vi.spyOn(client, 'post').mockResolvedValue({ scheduled_messages: [] }); + + await manager.load(); + + expect(manager.getFromState(gone.id)).toBeUndefined(); + }); + + it('keeps a message scheduled while a channel query is in flight', async () => { + const { manager } = setup(); + const token = manager.snapshotToken(); + const scheduledMeanwhile = generateScheduledMessage(); + manager.upsertToState(scheduledMeanwhile); + + manager.hydrateChannelState(cid, [], token); + + expect(manager.getFromState(scheduledMeanwhile.id)).toEqual(scheduledMeanwhile); + }); + }); + + describe('user switch', () => { + it('drops a load response that arrives after the user disconnected', async () => { + const { client, manager } = setup(); + const response = deferred(); + vi.spyOn(client, 'post').mockReturnValue(response.promise as never); + vi.spyOn(client, 'closeConnection').mockResolvedValue(undefined); + + const loading = manager.load(); + await client.disconnectUser(); + response.resolve({ scheduled_messages: [generateScheduledMessage()] }); + await loading; + + expect(manager.scheduledMessages.size).toBe(0); + expect(manager.state.getLatestValue().isLoaded).toBe(false); + }); + + it('drops a send-now response that arrives after the user disconnected', async () => { + const { client, manager } = setup(); + const sm = generateScheduledMessage(); + const response = deferred(); + vi.spyOn(client, 'post').mockReturnValue(response.promise as never); + vi.spyOn(client, 'closeConnection').mockResolvedValue(undefined); + + const sending = manager.sendScheduledMessageNow(sm.id); + await client.disconnectUser(); + response.resolve({ scheduled_message: sm }); + await sending; + + expect(manager.scheduledMessages.size).toBe(0); + }); + + it('drops a channel query response that arrives after the user disconnected', async () => { + const { client, manager } = setup(); + const token = manager.snapshotToken(); + vi.spyOn(client, 'closeConnection').mockResolvedValue(undefined); + await client.disconnectUser(); + + manager.hydrateChannelState(cid, [generateScheduledMessage()], token); + + expect(manager.scheduledMessages.size).toBe(0); + }); + }); }); diff --git a/test/unit/scheduledMessages/scheduled_messages.api.test.ts b/test/unit/scheduledMessages/scheduled_messages.api.test.ts index 1d0473c9ff..acefc4b6ae 100644 --- a/test/unit/scheduledMessages/scheduled_messages.api.test.ts +++ b/test/unit/scheduledMessages/scheduled_messages.api.test.ts @@ -164,6 +164,29 @@ describe('scheduled messages API', () => { expect(client.scheduledMessages.scheduledMessages.size).toBe(0); }); + it('sendMessage drops a scheduled response that arrives after the user disconnected', async () => { + const client = setup(); + const channel = client.channel('messaging', 'general'); + let resolve!: (value: unknown) => void; + vi.spyOn(client, 'post').mockReturnValue( + new Promise((r) => (resolve = r)) as never, + ); + vi.spyOn(client, 'closeConnection').mockResolvedValue(undefined); + + const sending = channel.sendMessage( + { text: 'see you tomorrow' }, + { scheduled_at: '2030-01-01T00:00:00Z' }, + ); + await client.disconnectUser(); + resolve({ + message: { id: 'sm-1', text: 'see you tomorrow' }, + scheduled_message: scheduledMessage, + }); + await sending; + + expect(client.scheduledMessages.scheduledMessages.size).toBe(0); + }); + it('query passes include_scheduled_messages and hydrates the store', async () => { const client = setup(); const { cid, id, type } = mockChannelQueryResponse.channel; From 5e783e088cdf5002a24584c2e96ff9b63aa398db Mon Sep 17 00:00:00 2001 From: nijeeshjoshy Date: Tue, 20 Oct 2026 11:11:24 +0200 Subject: [PATCH 4/4] feat: hydrate scheduled messages from queryChannels Pass include_scheduled_messages through queryChannels and replace each returned channel's entries in the scheduled messages store from its scheduled_messages field. An absent field leaves the store untouched so the app can fall back to queryScheduledMessages. Co-Authored-By: Claude Opus 5.5 --- src/client.ts | 13 ++- src/types.ts | 7 ++ .../scheduled_messages.api.test.ts | 87 +++++++++++++++++++ 3 files changed, 106 insertions(+), 1 deletion(-) diff --git a/src/client.ts b/src/client.ts index f02a13840f..2e5199ce25 100644 --- a/src/client.ts +++ b/src/client.ts @@ -2242,7 +2242,8 @@ export class StreamChat { * @param {ChannelFilters} filterConditions object MongoDB style filters * @param {ChannelSort} [sort] Sort options, for instance {created_at: -1}. * When using multiple fields, make sure you use array of objects to guarantee field order, for instance [{last_updated: -1}, {created_at: 1}] - * @param {ChannelOptions} [options] Options object + * @param {ChannelOptions} [options] Options object. With `include_scheduled_messages: true`, the + * scheduled messages returned with each channel replace that channel's entries in `client.scheduledMessages`. * @param {ChannelStateOptions} [stateOptions] State options object. These options will only be used for state management and won't be sent in the request. * - stateOptions.skipInitialization - Skips the initialization of the state for the channels matching the ids in the list. * - stateOptions.skipHydration - Skips returning the channels as instances of the Channel class and rather returns the raw query response. @@ -2269,6 +2270,7 @@ export class StreamChat { options: ChannelOptions = {}, stateOptions: ChannelStateOptions = {}, ): Promise { + const scheduledMessagesToken = this.scheduledMessages.snapshotToken(); const queryChannelsResponse = await this.queryChannelsRequestWithResponse( filterConditions, sort, @@ -2292,6 +2294,15 @@ export class StreamChat { }); } + // Not in hydrateActiveChannels: it also hydrates channels from the offline DB, + // which are not a fresh server snapshot of the scheduled messages. + for (const channelState of channels ?? []) { + this.scheduledMessages.hydrateChannelState( + channelState.channel.cid, + channelState.scheduled_messages, + scheduledMessagesToken, + ); + } const hydratedChannels = this.hydrateActiveChannels(channels, stateOptions, options); if (stateOptions.withResponse) { diff --git a/src/types.ts b/src/types.ts index 5613d1022d..d89cde0193 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1156,6 +1156,13 @@ export type BanUserOptions = UnBanUserOptions & { }; export type ChannelOptions = { + /** + * Client-side only. Adds the current user's scheduled messages to each returned + * channel (`scheduled_messages`). The field is absent on every channel when they + * are not known (e.g. more than 100 across the returned channels); use + * `client.queryScheduledMessages` then. + */ + include_scheduled_messages?: boolean; limit?: number; member_limit?: number; message_limit?: number; diff --git a/test/unit/scheduledMessages/scheduled_messages.api.test.ts b/test/unit/scheduledMessages/scheduled_messages.api.test.ts index acefc4b6ae..c603a509af 100644 --- a/test/unit/scheduledMessages/scheduled_messages.api.test.ts +++ b/test/unit/scheduledMessages/scheduled_messages.api.test.ts @@ -109,6 +109,93 @@ describe('scheduled messages API', () => { }); }); + describe('queryChannels', () => { + const channelResponse = ( + id: string, + scheduled_messages?: ScheduledMessageResponse[], + ) => ({ + ...mockChannelQueryResponse, + channel: { ...mockChannelQueryResponse.channel, id, cid: `messaging:${id}` }, + ...(scheduled_messages ? { scheduled_messages } : {}), + }); + + it('passes include_scheduled_messages and hydrates the store per channel', async () => { + const client = setup(); + const inA = { ...scheduledMessage, id: 'sm-a', cid: 'messaging:a' }; + const spy = vi.spyOn(client, 'post').mockResolvedValue({ + channels: [channelResponse('a', [inA]), channelResponse('b', [])], + }); + + await client.queryChannels({}, [], { include_scheduled_messages: true }); + + expect(spy.mock.calls[0][0]).toBe(`${client.baseURL}/channels`); + expect(spy.mock.calls[0][1]).toEqual( + expect.objectContaining({ include_scheduled_messages: true }), + ); + expect(client.scheduledMessages.getForChannel('messaging:a')).toEqual([inA]); + expect(client.scheduledMessages.getForChannel('messaging:b')).toEqual([]); + }); + + it('empties a channel whose scheduled_messages is an empty list', async () => { + const client = setup(); + const stale = { ...scheduledMessage, id: 'sm-stale', cid: 'messaging:a' }; + const other = { ...scheduledMessage, id: 'sm-other', cid: 'messaging:other' }; + client.scheduledMessages.upsertToState(stale); + client.scheduledMessages.upsertToState(other); + vi.spyOn(client, 'post').mockResolvedValue({ + channels: [channelResponse('a', [])], + }); + + await client.queryChannels({}, [], { include_scheduled_messages: true }); + + expect(client.scheduledMessages.getForChannel('messaging:a')).toEqual([]); + expect(client.scheduledMessages.getForChannel('messaging:other')).toEqual([other]); + }); + + it('keeps the store when scheduled_messages is absent', async () => { + const client = setup(); + const known = { ...scheduledMessage, id: 'sm-known', cid: 'messaging:a' }; + client.scheduledMessages.upsertToState(known); + vi.spyOn(client, 'post').mockResolvedValue({ + channels: [channelResponse('a')], + }); + + await client.queryChannels({}, [], { include_scheduled_messages: true }); + + expect(client.scheduledMessages.getForChannel('messaging:a')).toEqual([known]); + }); + + it('does not send include_scheduled_messages unless asked', async () => { + const client = setup(); + const spy = vi.spyOn(client, 'post').mockResolvedValue({ channels: [] }); + + await client.queryChannels({}, [], {}); + + expect(spy.mock.calls[0][1]).not.toHaveProperty('include_scheduled_messages'); + }); + + it('drops scheduled messages of a response that arrives after the user disconnected', async () => { + const client = setup(); + let resolve!: (value: unknown) => void; + vi.spyOn(client, 'post').mockReturnValue( + new Promise((r) => (resolve = r)) as never, + ); + vi.spyOn(client, 'closeConnection').mockResolvedValue(undefined); + + const querying = client.queryChannels({}, [], { include_scheduled_messages: true }); + await vi.waitFor(() => expect(client.post).toHaveBeenCalled()); + await client.disconnectUser(); + resolve({ + channels: [ + channelResponse('a', [{ ...scheduledMessage, id: 'sm-a', cid: 'messaging:a' }]), + ], + }); + await querying.catch(() => undefined); + + expect(client.scheduledMessages.scheduledMessages.size).toBe(0); + }); + }); + describe('channel', () => { it('sendMessage passes scheduled_at through and stores the scheduled message', async () => { const client = setup();