feat: support scheduled messages (CHA-5934)

src/channel.ts

@@ -211,6 +211,8 @@ export class Channel {
    * @param {boolean} [options.pending] Make this message pending
    * @param {Record<string,string>} [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<SendMessageAPIResponse>} The Server Response
    */
@@ -221,13 +223,29 @@ export class Channel {
       message: rawMessage,
     });
 
-    return await client.post<SendMessageAPIResponse>(this._channelURL() + '/message', {
-      message,
-      ...options,
-    });
+    const scheduledMessagesToken = client.scheduledMessages.snapshotToken();
+    const response = await client.post<SendMessageAPIResponse>(
+      this._channelURL() + '/message',
+      {
+        message,
+        ...options,
+      },
+    );
+    if (response.scheduled_message) {
+      client.scheduledMessages.upsertResponseToState(
+        response.scheduled_message,
+        scheduledMessagesToken,
+      );
+    }
+    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) {
@@ -1585,6 +1603,7 @@ export class Channel {
       );
     }
 
+    const scheduledMessagesToken = this.getClient().scheduledMessages.snapshotToken();
     let queryURL = `${this.getClient().baseURL}/channels/${encodeURIComponent(
       this.type,
     )}`;
@@ -1657,6 +1676,11 @@ 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,
+      scheduledMessagesToken,
+    );
 
     this.messageComposer.initStateFromChannelResponse(state);
 

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<string, MessageComposer>(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();
 
@@ -2233,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.
@@ -2260,6 +2270,7 @@ export class StreamChat {
     options: ChannelOptions = {},
     stateOptions: ChannelStateOptions = {},
   ): Promise<Channel[] | QueryChannelsResponseWithChannels> {
+    const scheduledMessagesToken = this.scheduledMessages.snapshotToken();
     const queryChannelsResponse = await this.queryChannelsRequestWithResponse(
       filterConditions,
       sort,
@@ -2283,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) {
@@ -5244,6 +5264,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<ScheduledMessageAPIResponse>} The scheduled message as stored after the change
+   */
+  async updateScheduledMessage(id: string, options: UpdateScheduledMessageOptions) {
+    return await this.patch<ScheduledMessageAPIResponse>(
+      `${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<ScheduledMessageAPIResponse>} The scheduled message as stored after the change
+   */
+  async deleteScheduledMessage(id: string) {
+    return await this.delete<ScheduledMessageAPIResponse>(
+      `${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<ScheduledMessageAPIResponse>} The scheduled message as stored after the change
+   */
+  async sendScheduledMessageNow(id: string) {
+    return await this.post<ScheduledMessageAPIResponse>(
+      `${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<QueryScheduledMessagesResponse>}
+   */
+  async queryScheduledMessages({ sort, ...rest }: QueryScheduledMessagesOptions = {}) {
+    return await this.post<QueryScheduledMessagesResponse>(
+      `${this.baseURL}/scheduled_messages/query`,
+      {
+        ...rest,
+        sort: sort && normalizeQuerySort(sort),
+      },
+    );
+  }
+
   /**
    * queryTeamUsageStats - Queries team-level usage statistics from the warehouse database
    *

src/errors.ts

@@ -29,6 +29,17 @@ export const APIErrorCodes: Record<string, { name: string; retryable: boolean }>
   '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 & {

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,

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';