feat: threads 2.0

src/index.ts

@@ -16,4 +16,4 @@ export * from './types';
 export * from './segment';
 export * from './campaign';
 export { isOwnUser, chatCodes, logChatPromiseExecution, formatMessage } from './utils';
-export * from './store/SimpleStateStore';
+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);
+  };
+}

src/thread.ts

@@ -1,6 +1,6 @@
 import type { Channel } from './channel';
 import type { StreamChat } from './client';
-import { SimpleStateStore } from './store/SimpleStateStore';
+import { StateStore } from './store';
 import type {
   AscDesc,
   DefaultGenerics,
@@ -81,7 +81,7 @@ export const DEFAULT_MARK_AS_READ_THROTTLE_DURATION = 1000;
 // TODO: store users someplace and reference them from state as now replies might contain users with stale information
 
 export class Thread<Scg extends ExtendableGenerics = DefaultGenerics> {
-  public readonly state: SimpleStateStore<ThreadState<Scg>>;
+  public readonly state: StateStore<ThreadState<Scg>>;
   public id: string;
 
   private client: StreamChat<Scg>;
@@ -100,7 +100,7 @@ export class Thread<Scg extends ExtendableGenerics = DefaultGenerics> {
 
     const placeholderDate = new Date();
 
-    this.state = new SimpleStateStore<ThreadState<Scg>>({
+    this.state = new StateStore<ThreadState<Scg>>({
       // used as handler helper - actively mark read all of the incoming messages
       // if the thread is active (visibly selected in the UI or main focal point in advanced list)
       active: false,
@@ -142,11 +142,11 @@ export class Thread<Scg extends ExtendableGenerics = DefaultGenerics> {
   }
 
   public activate = () => {
-    this.state.patchedNext('active', true);
+    this.state.partialNext({ active: true });
   };
 
   public deactivate = () => {
-    this.state.patchedNext('active', false);
+    this.state.partialNext({ active: false });
   };
 
   // take state of one instance and merge it to the current instance
@@ -265,7 +265,7 @@ export class Thread<Scg extends ExtendableGenerics = DefaultGenerics> {
 
         if (event.channel.cid !== channel.cid) return;
 
-        this.state.patchedNext('isStateStale', true);
+        this.state.partialNext({ isStateStale: true });
       }).unsubscribe,
     );
 
@@ -476,7 +476,7 @@ export class Thread<Scg extends ExtendableGenerics = DefaultGenerics> {
     sort,
     limit = DEFAULT_PAGE_LIMIT,
   }: Pick<QueryRepliesOptions<Scg>, 'sort' | 'limit'> = {}) => {
-    this.state.patchedNext('loadingNextPage', true);
+    this.state.partialNext({ loadingNextPage: true });
 
     const { loadingNextPage, nextCursor } = this.state.getLatestValue();
 
@@ -502,7 +502,7 @@ export class Thread<Scg extends ExtendableGenerics = DefaultGenerics> {
     } catch (error) {
       this.client.logger('error', (error as Error).message);
     } finally {
-      this.state.patchedNext('loadingNextPage', false);
+      this.state.partialNext({ loadingNextPage: false });
     }
   };
 
@@ -514,7 +514,7 @@ export class Thread<Scg extends ExtendableGenerics = DefaultGenerics> {
 
     if (loadingPreviousPage || previousCursor === null) return;
 
-    this.state.patchedNext('loadingPreviousPage', true);
+    this.state.partialNext({ loadingPreviousPage: true });
 
     try {
       const data = await this.queryReplies({
@@ -535,7 +535,7 @@ export class Thread<Scg extends ExtendableGenerics = DefaultGenerics> {
     } catch (error) {
       this.client.logger('error', (error as Error).message);
     } finally {
-      this.state.patchedNext('loadingPreviousPage', false);
+      this.state.partialNext({ loadingPreviousPage: false });
     }
   };
 }

src/thread_manager.ts

@@ -1,6 +1,6 @@
 import type { StreamChat } from './client';
-import type { Handler } from './store/SimpleStateStore';
-import { SimpleStateStore } from './store/SimpleStateStore';
+import type { Handler } from './store';
+import { StateStore } from './store';
 import type { Thread } from './thread';
 import type { DefaultGenerics, Event, ExtendableGenerics, QueryThreadsOptions } from './types';
 import { throttle } from './utils';
@@ -25,13 +25,13 @@ export type ThreadManagerState<Scg extends ExtendableGenerics = DefaultGenerics>
 type WithRequired<T, K extends keyof T> = T & { [P in K]-?: T[P] };
 
 export class ThreadManager<Scg extends ExtendableGenerics = DefaultGenerics> {
-  public readonly state: SimpleStateStore<ThreadManagerState<Scg>>;
+  public readonly state: StateStore<ThreadManagerState<Scg>>;
   private client: StreamChat<Scg>;
   private unsubscribeFunctions: Set<() => void> = new Set();
 
   constructor({ client }: { client: StreamChat<Scg> }) {
     this.client = client;
-    this.state = new SimpleStateStore<ThreadManagerState<Scg>>({
+    this.state = new StateStore<ThreadManagerState<Scg>>({
       active: false,
       threads: [],
       threadIdIndexMap: {},
@@ -46,11 +46,11 @@ export class ThreadManager<Scg extends ExtendableGenerics = DefaultGenerics> {
   }
 
   public activate = () => {
-    this.state.patchedNext('active', true);
+    this.state.partialNext({ active: true });
   };
 
   public deactivate = () => {
-    this.state.patchedNext('active', false);
+    this.state.partialNext({ active: false });
   };
 
   // eslint-disable-next-line sonarjs/cognitive-complexity
@@ -96,7 +96,7 @@ export class ThreadManager<Scg extends ExtendableGenerics = DefaultGenerics> {
         try {
           // FIXME: syncing does not work for me
           await this.client.sync(Array.from(channelCids), lastConnectionDownAt.toISOString(), { watch: true });
-          this.state.patchedNext('lastConnectionDownAt', null);
+          this.state.partialNext({ lastConnectionDownAt: null });
         } catch (error) {
           // TODO: if error mentions that the amount of events is more than 2k
           // do a reload-type recovery (re-query threads and merge states)
@@ -122,7 +122,7 @@ export class ThreadManager<Scg extends ExtendableGenerics = DefaultGenerics> {
         const { lastConnectionDownAt } = this.state.getLatestValue();
 
         if (!event.online && !lastConnectionDownAt) {
-          this.state.patchedNext('lastConnectionDownAt', new Date());
+          this.state.partialNext({ lastConnectionDownAt: new Date() });
         }
       }).unsubscribe,
     );

src/types.ts

@@ -503,7 +503,6 @@ export type ThreadResponse<StreamChatGenerics extends ExtendableGenerics = Defau
   channel: ChannelResponse<StreamChatGenerics>;
   channel_cid: string;
   created_at: string;
-  deleted_at: string;
   latest_replies: MessageResponse<StreamChatGenerics>[];
   parent_message_id: string;
   reply_count: number;
@@ -513,6 +512,7 @@ export type ThreadResponse<StreamChatGenerics extends ExtendableGenerics = Defau
   }[];
   title: string;
   updated_at: string;
+  deleted_at?: string;
   parent_message?: MessageResponse<StreamChatGenerics>;
   read?: {
     last_read: string;