feat: changes to support offline feature

src/channel.ts

@@ -51,6 +51,7 @@ import {
   UpdateChannelAPIResponse,
   UserFilters,
   UserResponse,
+  QueryChannelAPIResponse,
 } from './types';
 import { Role } from './permissions';
 
@@ -64,9 +65,24 @@ export class Channel<StreamChatGenerics extends ExtendableGenerics = DefaultGene
   data: ChannelData<StreamChatGenerics> | ChannelResponse<StreamChatGenerics> | undefined;
   _data: ChannelData<StreamChatGenerics> | ChannelResponse<StreamChatGenerics>;
   cid: string;
+  /**  */
   listeners: { [key: string]: (string | EventHandler<StreamChatGenerics>)[] };
   state: ChannelState<StreamChatGenerics>;
+  /**
+   * This boolean is a vague indication of weather the channel exists on chat backend.
+   *
+   * If the value is true, then that means the channel has been initialized by either calling
+   * channel.create() or channel.query() or channel.watch().
+   *
+   * If the value is false, then channel may or may not exist on the backend. The only way to ensure
+   * is by calling channel.create() or channel.query() or channel.watch().
+   */
   initialized: boolean;
+  /**
+   * Indicates weather channel has been initialized by manually populating the state with some messages, members etc.
+   * Static state indicates that channel exists on backend, but is not being watched yet.
+   */
+  offlineMode: boolean;
   lastKeyStroke?: Date;
   lastTypingEvent: Date | null;
   isTyping: boolean;
@@ -110,6 +126,7 @@ export class Channel<StreamChatGenerics extends ExtendableGenerics = DefaultGene
     // perhaps the state variable should be private
     this.state = new ChannelState<StreamChatGenerics>(this);
     this.initialized = false;
+    this.offlineMode = false;
     this.lastTypingEvent = null;
     this.isTyping = false;
     this.disconnected = false;
@@ -730,7 +747,7 @@ export class Channel<StreamChatGenerics extends ExtendableGenerics = DefaultGene
    *
    * @param {ChannelQueryOptions<StreamChatGenerics>} options additional options for the query endpoint
    *
-   * @return {Promise<ChannelAPIResponse<StreamChatGenerics>>} The server response
+   * @return {Promise<QueryChannelAPIResponse<StreamChatGenerics>>} The server response
    */
   async watch(options?: ChannelQueryOptions<StreamChatGenerics>) {
     const defaultOptions = {
@@ -929,7 +946,7 @@ export class Channel<StreamChatGenerics extends ExtendableGenerics = DefaultGene
   /**
    * create - Creates a new channel
    *
-   * @return {Promise<ChannelAPIResponse<StreamChatGenerics>>} The Server Response
+   * @return {Promise<QueryChannelAPIResponse<StreamChatGenerics>>} The Server Response
    */
   create = async () => {
     const options = {
@@ -946,7 +963,7 @@ export class Channel<StreamChatGenerics extends ExtendableGenerics = DefaultGene
    * @param {ChannelQueryOptions<StreamChatGenerics>} options The query options
    * @param {MessageSetType} messageSetToAddToIfDoesNotExist It's possible to load disjunct sets of a channel's messages into state, use `current` to load the initial channel state or if you want to extend the currently displayed messages, use `latest` if you want to load/extend the latest messages, `new` is used for loading a specific message and it's surroundings
    *
-   * @return {Promise<ChannelAPIResponse<StreamChatGenerics>>} Returns a query response
+   * @return {Promise<QueryChannelAPIResponse<StreamChatGenerics>>} Returns a query response
    */
   async query(
     options: ChannelQueryOptions<StreamChatGenerics>,
@@ -960,7 +977,7 @@ export class Channel<StreamChatGenerics extends ExtendableGenerics = DefaultGene
       queryURL += `/${this.id}`;
     }
 
-    const state = await this.getClient().post<ChannelAPIResponse<StreamChatGenerics>>(queryURL + '/query', {
+    const state = await this.getClient().post<QueryChannelAPIResponse<StreamChatGenerics>>(queryURL + '/query', {
       data: this._data,
       state: true,
       ...options,
@@ -992,10 +1009,18 @@ export class Channel<StreamChatGenerics extends ExtendableGenerics = DefaultGene
     this.getClient()._addChannelConfig(state);
 
     // add any messages to our channel state
-    this._initializeState(state, messageSetToAddToIfDoesNotExist);
+    const { messageSet } = this._initializeState(state, messageSetToAddToIfDoesNotExist);
 
     this.data = state.channel;
 
+    this.getClient().dispatchEvent({
+      type: 'channels.queried',
+      queriedChannels: {
+        channels: [state],
+        isLatestMessageSet: messageSet.isLatest,
+      },
+    });
+
     return state;
   }
 
@@ -1271,7 +1296,8 @@ export class Channel<StreamChatGenerics extends ExtendableGenerics = DefaultGene
           });
 
           channelState.pinnedMessages.forEach(({ id, created_at: createdAt }) => {
-            if (truncatedAt > +createdAt) channelState.removePinnedMessage({ id });
+            if (truncatedAt > +createdAt)
+              channelState.removePinnedMessage({ id } as MessageResponse<StreamChatGenerics>);
           });
         } else {
           channelState.clearMessages();
@@ -1368,7 +1394,7 @@ export class Channel<StreamChatGenerics extends ExtendableGenerics = DefaultGene
   };
 
   _checkInitialized() {
-    if (!this.initialized && !this.getClient()._isUsingServerAuth()) {
+    if (!this.initialized && !this.offlineMode && !this.getClient()._isUsingServerAuth()) {
       throw Error(
         `Channel ${this.cid} hasn't been initialized yet. Make sure to call .watch() and wait for it to resolve`,
       );
@@ -1397,7 +1423,8 @@ export class Channel<StreamChatGenerics extends ExtendableGenerics = DefaultGene
     if (!this.state.messages) {
       this.state.initMessages();
     }
-    this.state.addMessagesSorted(messages, false, true, true, messageSetToAddToIfDoesNotExist);
+    const { messageSet } = this.state.addMessagesSorted(messages, false, true, true, messageSetToAddToIfDoesNotExist);
+
     if (!this.state.pinnedMessages) {
       this.state.pinnedMessages = [];
     }
@@ -1452,6 +1479,10 @@ export class Channel<StreamChatGenerics extends ExtendableGenerics = DefaultGene
         }
       }
     }
+
+    return {
+      messageSet,
+    };
   }
 
   _extendEventWithOwnReactions(event: Event<StreamChatGenerics>) {

src/channel_state.ts

@@ -244,6 +244,10 @@ export class ChannelState<StreamChatGenerics extends ExtendableGenerics = Defaul
         this.threads[parentID] = threadMessages;
       }
     }
+
+    return {
+      messageSet: this.messageSets[targetMessageSetIndex],
+    };
   }
 
   /**

src/client.ts

@@ -146,6 +146,7 @@ import {
   UserOptions,
   UserResponse,
   UserSort,
+  QueryChannelsAPIResponse,
 } from './types';
 import { InsightMetrics, postInsights } from './insights';
 
@@ -161,6 +162,7 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
     [key: string]: Channel<StreamChatGenerics>;
   };
   anonymous: boolean;
+  persistUserOnConnectionFailure?: boolean;
   axiosInstance: AxiosInstance;
   baseURL?: string;
   browser: boolean;
@@ -279,6 +281,7 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
     // mapping between channel groups and configs
     this.configs = {};
     this.anonymous = false;
+    this.persistUserOnConnectionFailure = this.options?.persistUserOnConnectionFailure;
 
     // If its a server-side client, then lets initialize the tokenManager, since token will be
     // generated from secret.
@@ -460,8 +463,12 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
     try {
       return await this.setUserPromise;
     } catch (err) {
-      // cleanup client to allow the user to retry connectUser again
-      this.disconnectUser();
+      if (this.persistUserOnConnectionFailure) {
+        // cleanup client to allow the user to retry connectUser again
+        this.closeConnection();
+      } else {
+        this.disconnectUser();
+      }
       throw err;
     }
   };
@@ -487,6 +494,7 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
      * It contains reserved properties and own user properties which are not present in `this._user`.
      */
     this.user = user;
+    this.userID = user.id;
     // this one is actually used for requests. This is a copy of current user provided to `connectUser` function.
     this._user = { ...user };
   }
@@ -707,7 +715,6 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
     for (const channel of Object.values(this.activeChannels)) {
       channel._disconnect();
     }
-
     // ensure we no longer return inactive channels
     this.activeChannels = {};
     // reset client state
@@ -1461,7 +1468,6 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
     options: ChannelOptions = {},
     stateOptions: ChannelStateOptions = {},
   ) {
-    const { skipInitialization } = stateOptions;
     const defaultOptions: ChannelOptions = { state: true, watch: true, presence: false };
 
     // Make sure we wait for the connect promise if there is a pending one
@@ -1478,22 +1484,36 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
       ...options,
     };
 
-    const data = await this.post<{ channels: ChannelAPIResponse<StreamChatGenerics>[] }>(
-      this.baseURL + '/channels',
-      payload,
-    );
+    const data = await this.post<QueryChannelsAPIResponse<StreamChatGenerics>>(this.baseURL + '/channels', payload);
 
-    const channels: Channel<StreamChatGenerics>[] = [];
+    this.dispatchEvent({
+      type: 'channels.queried',
+      queriedChannels: {
+        channels: data.channels,
+        isLatestMessageSet: true,
+      },
+    });
+
+    return this.hydrateActiveChannels(data.channels, stateOptions);
+  }
 
-    // update our cache of the configs
-    for (const channelState of data.channels) {
+  hydrateActiveChannels(
+    channelsFromApi: ChannelAPIResponse<StreamChatGenerics>[] = [],
+    stateOptions: ChannelStateOptions = {},
+  ) {
+    const { skipInitialization, offlineMode = false } = stateOptions;
+
+    for (const channelState of channelsFromApi) {
       this._addChannelConfig(channelState);
     }
 
-    for (const channelState of data.channels) {
+    const channels: Channel<StreamChatGenerics>[] = [];
+
+    for (const channelState of channelsFromApi) {
       const c = this.channel(channelState.channel.type, channelState.channel.id);
       c.data = channelState.channel;
-      c.initialized = true;
+      c.offlineMode = offlineMode;
+      c.initialized = !offlineMode;
 
       if (skipInitialization === undefined) {
         c._initializeState(channelState, 'latest');
@@ -1504,6 +1524,18 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
 
       channels.push(c);
     }
+
+    if (!offlineMode) {
+      // If the channels are coming from server, then clear out the
+      // previously help offline channels.
+      for (const key in this.activeChannels) {
+        const channel = this.activeChannels[key];
+        if (channel.offlineMode) {
+          delete this.activeChannels[key];
+        }
+      }
+    }
+
     return channels;
   }
 

src/connection.ts

@@ -257,13 +257,25 @@ export class StableWSConnection<StreamChatGenerics extends ExtendableGenerics =
    * @return {ConnectAPIResponse<ChannelType, CommandType, UserType>} Promise that completes once the first health check message is received
    */
   async _connect() {
-    if (this.isConnecting || this.isDisconnected) return; // simply ignore _connect if it's currently trying to connect
+    if (this.isConnecting || (this.isDisconnected && this.client.options.enableWSFallback)) return; // simply ignore _connect if it's currently trying to connect
     this.isConnecting = true;
     this.requestID = randomId();
     this.client.insightMetrics.connectionStartTimestamp = new Date().getTime();
+    let isTokenReady = false;
     try {
       this._log(`_connect() - waiting for token`);
       await this.client.tokenManager.tokenReady();
+      isTokenReady = true;
+    } catch (e) {
+      // token provider has failed before, so try again
+    }
+
+    try {
+      if (!isTokenReady) {
+        this._log(`_connect() - tokenProvider failed before, so going to retry`);
+        await this.client.tokenManager.loadToken();
+      }
+
       this._setupConnectionPromise();
       const wsURL = this._buildUrl();
       this._log(`_connect() - Connecting to ${wsURL}`, { wsURL, requestID: this.requestID });
@@ -288,7 +300,7 @@ export class StableWSConnection<StreamChatGenerics extends ExtendableGenerics =
       }
     } catch (err) {
       this.isConnecting = false;
-
+      this._log(`_connect() - Error - `, err);
       if (this.client.options.enableInsights) {
         this.client.insightMetrics.wsConsecutiveFailures++;
         this.client.insightMetrics.wsTotalFailures++;
@@ -333,7 +345,7 @@ export class StableWSConnection<StreamChatGenerics extends ExtendableGenerics =
       return;
     }
 
-    if (this.isDisconnected) {
+    if (this.isDisconnected && this.client.options.enableWSFallback) {
       this._log('_reconnect() - Abort (3) since disconnect() is called');
       return;
     }

src/events.ts

@@ -42,6 +42,7 @@ export const EVENT_MAP = {
   'user.watching.stop': true,
 
   // local events
+  'channels.queried': true,
   'connection.changed': true,
   'connection.recovered': true,
   'transport.changed': true,

src/types.ts

@@ -267,7 +267,14 @@ export type ChannelResponse<
   updated_at?: string;
 };
 
-export type ChannelAPIResponse<StreamChatGenerics extends ExtendableGenerics = DefaultGenerics> = APIResponse & {
+export type QueryChannelsAPIResponse<StreamChatGenerics extends ExtendableGenerics = DefaultGenerics> = APIResponse & {
+  channels: Omit<ChannelAPIResponse<StreamChatGenerics>, keyof APIResponse>[];
+};
+
+export type QueryChannelAPIResponse<StreamChatGenerics extends ExtendableGenerics = DefaultGenerics> = APIResponse &
+  ChannelAPIResponse<StreamChatGenerics>;
+
+export type ChannelAPIResponse<StreamChatGenerics extends ExtendableGenerics = DefaultGenerics> = {
   channel: ChannelResponse<StreamChatGenerics>;
   members: ChannelMemberResponse<StreamChatGenerics>[];
   messages: MessageResponse<StreamChatGenerics>[];
@@ -500,6 +507,7 @@ export type MessageResponse<
 export type MessageResponseBase<
   StreamChatGenerics extends ExtendableGenerics = DefaultGenerics
 > = MessageBase<StreamChatGenerics> & {
+  type: MessageLabel;
   args?: string;
   channel?: ChannelResponse<StreamChatGenerics>;
   cid?: string;
@@ -523,7 +531,6 @@ export type MessageResponseBase<
   silent?: boolean;
   status?: string;
   thread_participants?: UserResponse<StreamChatGenerics>[];
-  type?: MessageLabel;
   updated_at?: string;
 };
 
@@ -580,6 +587,7 @@ export type ReactionResponse<
   StreamChatGenerics extends ExtendableGenerics = DefaultGenerics
 > = Reaction<StreamChatGenerics> & {
   created_at: string;
+  message_id: string;
   updated_at: string;
 };
 
@@ -733,6 +741,7 @@ export type ChannelQueryOptions<StreamChatGenerics extends ExtendableGenerics =
 };
 
 export type ChannelStateOptions = {
+  offlineMode?: boolean;
   skipInitialization?: string[];
 };
 
@@ -897,6 +906,7 @@ export type StreamChatOptions = AxiosRequestConfig & {
   /** experimental feature, please contact support if you want this feature enabled for you */
   enableWSFallback?: boolean;
   logger?: Logger;
+  persistUserOnConnectionFailure?: boolean;
   /**
    * When network is recovered, we re-query the active channels on client. But in single query, you can recover
    * only 30 channels. So its not guaranteed that all the channels in activeChannels object have updated state.
@@ -979,8 +989,13 @@ export type Event<StreamChatGenerics extends ExtendableGenerics = DefaultGeneric
   me?: OwnUserResponse<StreamChatGenerics>;
   member?: ChannelMemberResponse<StreamChatGenerics>;
   message?: MessageResponse<StreamChatGenerics>;
+  mode?: string;
   online?: boolean;
   parent_id?: string;
+  queriedChannels?: {
+    channels: ChannelAPIResponse<StreamChatGenerics>[];
+    isLatestMessageSet?: boolean;
+  };
   reaction?: ReactionResponse<StreamChatGenerics>;
   received_at?: string | Date;
   team?: string;