feat: threads 2.0

package.json

@@ -95,15 +95,15 @@
     "eslint-plugin-typescript-sort-keys": "1.5.0",
     "husky": "^4.3.8",
     "lint-staged": "^15.2.2",
-    "mocha": "^9.1.3",
+    "mocha": "^10.7.0",
     "nyc": "^15.1.0",
     "prettier": "^2.2.1",
     "rollup": "^2.41.0",
     "rollup-plugin-peer-deps-external": "^2.2.4",
     "rollup-plugin-terser": "^7.0.2",
     "sinon": "^12.0.1",
     "standard-version": "^9.3.2",
-    "typescript": "^4.2.3",
+    "typescript": "^5.5.4",
     "uuid": "^8.3.2"
   },
   "scripts": {
@@ -118,7 +118,7 @@
     "test-types": "node test/typescript/index.js && tsc --esModuleInterop true --noEmit true --strictNullChecks true --noImplicitAny true --strict true test/typescript/*.ts",
     "eslint": "eslint '**/*.{js,md,ts}' --max-warnings 0 --ignore-path ./.eslintignore",
     "eslint-fix": "npx eslint --fix '**/*.{js,md,ts}' --max-warnings 0 --ignore-path ./.eslintignore",
-    "test-unit": "NODE_ENV=test mocha --exit --bail --timeout 20000 --require ./babel-register test/unit/*.js",
+    "test-unit": "NODE_ENV=test mocha --exit --bail --timeout 20000 --require ./babel-register test/unit/*.{js,test.ts}",
     "test-coverage": "nyc yarn test-unit",
     "test": "yarn test-unit",
     "testwatch": "NODE_ENV=test nodemon ./node_modules/.bin/mocha --timeout 20000 --require test-entry.js test/test.js",

src/channel.ts

@@ -820,6 +820,7 @@ export class Channel<StreamChatGenerics extends ExtendableGenerics = DefaultGene
     return response;
   }
 
+  // TODO: move to thread
   /**
    * getReplies - List the message replies for a parent message
    *

src/channel_state.ts

@@ -13,7 +13,7 @@ import {
   ReactionResponse,
   UserResponse,
 } from './types';
-import { addToMessageList } from './utils';
+import { addToMessageList, formatMessage } from './utils';
 
 type ChannelReadStatus<StreamChatGenerics extends ExtendableGenerics = DefaultGenerics> = Record<
   string,
@@ -61,6 +61,7 @@ export class ChannelState<StreamChatGenerics extends ExtendableGenerics = Defaul
     isLatest: boolean;
     messages: Array<ReturnType<ChannelState<StreamChatGenerics>['formatMessage']>>;
   }[] = [];
+
   constructor(channel: Channel<StreamChatGenerics>) {
     this._channel = channel;
     this.watcher_count = 0;
@@ -132,26 +133,13 @@ export class ChannelState<StreamChatGenerics extends ExtendableGenerics = Defaul
   }
 
   /**
-   * formatMessage - Takes the message object. Parses the dates, sets __html
-   * and sets the status to received if missing. Returns a message object
-   *
-   * @param {MessageResponse<StreamChatGenerics>} message a message object
+   * Takes the message object, parses the dates, sets `__html`
+   * and sets the status to `received` if missing; returns a new message object.
    *
+   * @param {MessageResponse<StreamChatGenerics>} message `MessageResponse` object
    */
-  formatMessage(message: MessageResponse<StreamChatGenerics>): FormatMessageResponse<StreamChatGenerics> {
-    return {
-      ...message,
-      /**
-       * @deprecated please use `html`
-       */
-      __html: message.html,
-      // parse the date..
-      pinned_at: message.pinned_at ? new Date(message.pinned_at) : null,
-      created_at: message.created_at ? new Date(message.created_at) : new Date(),
-      updated_at: message.updated_at ? new Date(message.updated_at) : new Date(),
-      status: message.status || 'received',
-    };
-  }
+  formatMessage = (message: MessageResponse<StreamChatGenerics>): FormatMessageResponse<StreamChatGenerics> =>
+    formatMessage<StreamChatGenerics>(message);
 
   /**
    * addMessagesSorted - Add the list of messages to state and resorts the messages
@@ -295,15 +283,16 @@ export class ChannelState<StreamChatGenerics extends ExtendableGenerics = Defaul
     this.pinnedMessages = result;
   }
 
+  // TODO: clean this up
   addReaction(
     reaction: ReactionResponse<StreamChatGenerics>,
     message?: MessageResponse<StreamChatGenerics>,
-    enforce_unique?: boolean,
+    enforceUnique?: boolean,
   ) {
     if (!message) return;
     const messageWithReaction = message;
     this._updateMessage(message, (msg) => {
-      messageWithReaction.own_reactions = this._addOwnReactionToMessage(msg.own_reactions, reaction, enforce_unique);
+      messageWithReaction.own_reactions = this._addOwnReactionToMessage(msg.own_reactions, reaction, enforceUnique);
       return this.formatMessage(messageWithReaction);
     });
     return messageWithReaction;
@@ -664,7 +653,7 @@ export class ChannelState<StreamChatGenerics extends ExtendableGenerics = Defaul
           messages[i] = {
             ...m,
             type: 'deleted',
-            deleted_at: user.deleted_at,
+            deleted_at: user.deleted_at ? new Date(user.deleted_at) : null,
           };
         }
       }

src/client.ts

@@ -207,6 +207,7 @@ import {
 import { InsightMetrics, postInsights } from './insights';
 import { Thread } from './thread';
 import { Moderation } from './moderation';
+import { ThreadManager } from './thread_manager';
 
 function isString(x: unknown): x is string {
   return typeof x === 'string' || x instanceof String;
@@ -219,6 +220,7 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
   activeChannels: {
     [key: string]: Channel<StreamChatGenerics>;
   };
+  threads: ThreadManager<StreamChatGenerics>;
   anonymous: boolean;
   persistUserOnConnectionFailure?: boolean;
   axiosInstance: AxiosInstance;
@@ -338,6 +340,7 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
     this.setUserPromise = null;
     // keeps a reference to all the channels that are in use
     this.activeChannels = {};
+
     // mapping between channel groups and configs
     this.configs = {};
     this.anonymous = false;
@@ -349,8 +352,8 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
     this.consecutiveFailures = 0;
     this.insightMetrics = new InsightMetrics();
 
-    this.defaultWSTimeoutWithFallback = 6000;
-    this.defaultWSTimeout = 15000;
+    this.defaultWSTimeoutWithFallback = 6 * 1000;
+    this.defaultWSTimeout = 15 * 1000;
 
     this.axiosInstance.defaults.paramsSerializer = axiosParamsSerializer;
 
@@ -404,6 +407,9 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
      */
     this.logger = isFunction(inputOptions.logger) ? inputOptions.logger : () => null;
     this.recoverStateOnReconnect = this.options.recoverStateOnReconnect;
+
+    // reusing the same name the channel has (Channel.threads)
+    this.threads = new ThreadManager({ client: this });
   }
 
   /**
@@ -604,7 +610,7 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
         tags: ['connection', 'client'],
       });
 
-      return Promise.resolve();
+      return;
     }
 
     this.clientID = `${this.userID}--${randomId()}`;
@@ -1432,7 +1438,7 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
       return await this.wsConnection.connect(
         this.options.enableWSFallback ? this.defaultWSTimeoutWithFallback : this.defaultWSTimeout,
       );
-    } catch (err) {
+    } catch (err: any) {
       // run fallback only if it's WS/Network error and not a normal API error
       // make sure browser is online before even trying the longpoll
       if (this.options.enableWSFallback && isWSFailure(err) && isOnline()) {
@@ -2721,7 +2727,7 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
     const res = await this.post<QueryThreadsAPIResponse<StreamChatGenerics>>(this.baseURL + `/threads`, opts);
 
     return {
-      threads: res.threads.map((thread) => new Thread(this, thread)),
+      threads: res.threads.map((thread) => new Thread({ client: this, threadData: thread })),
       next: res.next,
     };
   }
@@ -2751,7 +2757,7 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
 
     const res = await this.get<GetThreadAPIResponse<StreamChatGenerics>>(this.baseURL + `/threads/${messageId}`, opts);
 
-    return new Thread<StreamChatGenerics>(this, res.thread);
+    return new Thread<StreamChatGenerics>({ client: this, threadData: res.thread });
   }
 
   /**

src/connection.ts

@@ -113,7 +113,7 @@ export class StableWSConnection<StreamChatGenerics extends ExtendableGenerics =
       this.consecutiveFailures = 0;
 
       this._log(`connect() - Established ws connection with healthcheck: ${healthCheck}`);
-    } catch (error) {
+    } catch (error: any) {
       this.isHealthy = false;
       this.consecutiveFailures += 1;
 
@@ -148,7 +148,7 @@ export class StableWSConnection<StreamChatGenerics extends ExtendableGenerics =
         for (let i = 0; i <= timeout; i += interval) {
           try {
             return await this.connectionOpen;
-          } catch (error) {
+          } catch (error: any) {
             if (i === timeout) {
               throw new Error(
                 JSON.stringify({
@@ -298,7 +298,7 @@ export class StableWSConnection<StreamChatGenerics extends ExtendableGenerics =
         }
         return response;
       }
-    } catch (err) {
+    } catch (err: any) {
       this.isConnecting = false;
       this._log(`_connect() - Error - `, err);
       if (this.client.options.enableInsights) {
@@ -366,7 +366,7 @@ export class StableWSConnection<StreamChatGenerics extends ExtendableGenerics =
       this._log('_reconnect() - Finished recoverCallBack');
 
       this.consecutiveFailures = 0;
-    } catch (error) {
+    } catch (error: any) {
       this.isHealthy = false;
       this.consecutiveFailures += 1;
       if (error.code === chatCodes.TOKEN_EXPIRED && !this.client.tokenManager.isStatic()) {

src/connection_fallback.ts

@@ -81,7 +81,7 @@ export class WSConnectionFallback<StreamChatGenerics extends ExtendableGenerics
 
       this.consecutiveFailures = 0; // always reset in case of no error
       return res;
-    } catch (err) {
+    } catch (err: any) {
       this.consecutiveFailures += 1;
 
       if (retry && isErrorRetryable(err)) {
@@ -107,7 +107,7 @@ export class WSConnectionFallback<StreamChatGenerics extends ExtendableGenerics
             this.client.dispatchEvent(data.events[i]);
           }
         }
-      } catch (err) {
+      } catch (err: any) {
         if (axios.isCancel(err)) {
           this._log(`_poll() - axios canceled request`);
           return;

src/index.ts

@@ -4,6 +4,7 @@ export * from './client_state';
 export * from './channel';
 export * from './channel_state';
 export * from './thread';
+export * from './thread_manager';
 export * from './connection';
 export * from './events';
 export * from './moderation';
@@ -15,3 +16,4 @@ export * from './types';
 export * from './segment';
 export * from './campaign';
 export { isOwnUser, chatCodes, logChatPromiseExecution, formatMessage } from './utils';
+export * from './store';

src/store.ts

@@ -0,0 +1,57 @@
+export type Patch<T> = (value: T) => T;
+export type Handler<T> = (nextValue: T, previousValue: T | undefined) => void;
+export type Unsubscibe = () => void;
+
+function isPatch<T>(value: T | Patch<T>): value is Patch<T> {
+  return typeof value === 'function';
+}
+
+export class StateStore<T extends Record<string, unknown>> {
+  private handlerSet = new Set<Handler<T>>();
+
+  constructor(private value: T) {}
+
+  public next = (newValueOrPatch: T | Patch<T>): void => {
+    // newValue (or patch output) should never be mutated previous value
+    const newValue = isPatch(newValueOrPatch) ? newValueOrPatch(this.value) : newValueOrPatch;
+
+    // do not notify subscribers if the value hasn't changed
+    if (newValue === this.value) return;
+
+    const oldValue = this.value;
+    this.value = newValue;
+
+    this.handlerSet.forEach((handler) => handler(this.value, oldValue));
+  };
+
+  public partialNext = (partial: Partial<T>): void => this.next((current) => ({ ...current, ...partial }));
+
+  public getLatestValue = (): T => this.value;
+
+  public subscribe = (handler: Handler<T>): Unsubscibe => {
+    handler(this.value, undefined);
+    this.handlerSet.add(handler);
+    return () => {
+      this.handlerSet.delete(handler);
+    };
+  };
+
+  public subscribeWithSelector = <O extends readonly unknown[]>(selector: (nextValue: T) => O, handler: Handler<O>) => {
+    // begin with undefined to reduce amount of selector calls
+    let selectedValues: O | undefined;
+
+    const wrappedHandler: Handler<T> = (nextValue) => {
+      const newlySelectedValues = selector(nextValue);
+      const hasUpdatedValues = selectedValues?.some((value, index) => value !== newlySelectedValues[index]) ?? true;
+
+      if (!hasUpdatedValues) return;
+
+      const oldSelectedValues = selectedValues;
+      selectedValues = newlySelectedValues;
+
+      handler(newlySelectedValues, oldSelectedValues);
+    };
+
+    return this.subscribe(wrappedHandler);
+  };
+}