Draft - target factories

Author: arnautov-anton

src/client.ts

@@ -82,6 +82,7 @@ import {
   ErrorFromResponse,
   Event,
   EventHandler,
+  EventTypes,
   ExportChannelOptions,
   ExportChannelRequest,
   ExportChannelResponse,
@@ -218,6 +219,11 @@ function isString(x: unknown): x is string {
   return typeof x === 'string' || x instanceof String;
 }
 
+export type TargetFactory<
+  SCG extends ExtendableGenerics = DefaultGenerics,
+  T extends Exclude<EventTypes, 'all'> = Exclude<EventTypes, 'all'>
+> = (event: Event<SCG>) => `${T}${string}` | `${string}${T}` | `${string}${T}${string}` | null;
+
 export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultGenerics> {
   private static _instance?: unknown | StreamChat; // type is undefined|StreamChat, unknown is due to TS limitations with statics
 
@@ -269,6 +275,7 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
   defaultWSTimeoutWithFallback: number;
   defaultWSTimeout: number;
   private nextRequestAbortController: AbortController | null = null;
+  private targetFactoriesByType = new Map<Exclude<EventTypes, 'all'>, Set<TargetFactory<StreamChatGenerics>>>();
 
   /**
    * Initialize a client
@@ -909,6 +916,28 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
     return JWTUserToken(this.secret, userID, extra, {});
   }
 
+  public registerTargetFactory = <T extends Exclude<EventTypes, 'all'>>(
+    eventType: T,
+    factory: TargetFactory<StreamChatGenerics, T>,
+  ) => {
+    let set = this.targetFactoriesByType.get(eventType);
+
+    if (!set) {
+      set = new Set();
+      this.targetFactoriesByType.set(eventType, set);
+    }
+
+    set.add(factory);
+
+    return () => {
+      set.delete(factory);
+
+      if (!set.size) {
+        this.targetFactoriesByType.delete(eventType);
+      }
+    };
+  };
+
   /**
    * on - Listen to events on all channels and users your watching
    *
@@ -936,6 +965,7 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
       tags: ['event', 'client'],
     });
     this.listeners[key].push(callback);
+
     return {
       unsubscribe: () => {
         this.logger('info', `Removing listener for ${key} event`, {
@@ -1364,8 +1394,21 @@ export class StreamChat<StreamChatGenerics extends ExtendableGenerics = DefaultG
     if (client.listeners.all) {
       listeners.push(...client.listeners.all);
     }
-    if (client.listeners[event.type]) {
-      listeners.push(...client.listeners[event.type]);
+
+    const eventTypes: string[] = [event.type];
+
+    // factories
+    const factorySet = client.targetFactoriesByType.get(event.type as Exclude<EventTypes, 'all'>);
+    factorySet?.forEach((factory) => {
+      // a specific value could be missing from the event payload so factory can return "null" to be skipped
+      const targetedEventType = factory(event);
+      if (targetedEventType) eventTypes.push(targetedEventType);
+    });
+
+    for (const eventType of eventTypes) {
+      if (client.listeners[eventType]) {
+        listeners.push(...client.listeners[eventType]);
+      }
     }
 
     // call the event and send it to the listeners

src/thread.ts

@@ -1,5 +1,5 @@
 import type { Channel } from './channel';
-import type { StreamChat } from './client';
+import type { StreamChat, TargetFactory } from './client';
 import { StateStore } from './store';
 import type {
   AscDesc,
@@ -62,6 +62,14 @@ export type ThreadReadState<SCG extends ExtendableGenerics = DefaultGenerics> =
   ThreadUserReadState<SCG> | undefined
 >;
 
+// TODO: figure out generics here
+const messageNewFactory: TargetFactory<DefaultGenerics, 'message.new'> = (event) => {
+  return event.parent_id ? `message.new-${event.parent_id}` : null;
+};
+const threadMessageReadFactory: TargetFactory<DefaultGenerics, 'message.read'> = (event) => {
+  return event.thread ? `message.read-${event.thread.parent_message_id}` : null;
+};
+
 const DEFAULT_PAGE_LIMIT = 50;
 const DEFAULT_SORT: { created_at: AscDesc }[] = [{ created_at: -1 }];
 const MARK_AS_READ_THROTTLE_TIMEOUT = 1000;
@@ -186,6 +194,9 @@ export class Thread<SCG extends ExtendableGenerics = DefaultGenerics> {
       return;
     }
 
+    this.unsubscribeFunctions.add(this.client.registerTargetFactory('message.new', messageNewFactory));
+    this.unsubscribeFunctions.add(this.client.registerTargetFactory('message.read', threadMessageReadFactory));
+
     this.unsubscribeFunctions.add(this.subscribeMarkActiveThreadRead());
     this.unsubscribeFunctions.add(this.subscribeReloadActiveStaleThread());
     this.unsubscribeFunctions.add(this.subscribeMarkThreadStale());
@@ -230,7 +241,7 @@ export class Thread<SCG extends ExtendableGenerics = DefaultGenerics> {
     }).unsubscribe;
 
   private subscribeNewReplies = () =>
-    this.client.on('message.new', (event) => {
+    this.client.on(`message.new-${this.id}`, (event) => {
       if (!this.client.userID || event.message?.parent_id !== this.id) {
         return;
       }
@@ -284,7 +295,7 @@ export class Thread<SCG extends ExtendableGenerics = DefaultGenerics> {
     }).unsubscribe;
 
   private subscribeRepliesRead = () =>
-    this.client.on('message.read', (event) => {
+    this.client.on(`message.read-${this.id}`, (event) => {
       if (!event.user || !event.created_at || !event.thread) return;
       if (event.thread.parent_message_id !== this.id) return;