feat: destination filter support for device mode plugins (#685)

package.json

@@ -15,11 +15,6 @@
     "lint": "eslint \"**/*.{js,ts,tsx}\"",
     "release": "yarn workspaces run release"
   },
-  "jest": {
-    "projects": [
-      "<rootDir>/packages/*"
-    ]
-  },
   "devDependencies": {
     "@changesets/cli": "^2.16.0",
     "@commitlint/config-conventional": "^16.2.4",

packages/core/src/__tests__/__helpers__/mockSegmentStore.ts

@@ -9,7 +9,9 @@ import type {
 import type {
   Context,
   DeepPartial,
+  DestinationFilters,
   IntegrationSettings,
+  RoutingRule,
   SegmentAPIIntegrations,
   UserInfoState,
 } from '../../types';
@@ -20,6 +22,7 @@ export type StoreData = {
   isReady: boolean;
   context?: DeepPartial<Context>;
   settings: SegmentAPIIntegrations;
+  filters: DestinationFilters;
   userInfo: UserInfoState;
   deepLinkData: DeepLinkData;
 };
@@ -30,6 +33,7 @@ const INITIAL_VALUES: StoreData = {
   settings: {
     [SEGMENT_DESTINATION_KEY]: {},
   },
+  filters: {},
   userInfo: {
     anonymousId: 'anonymousId',
     userId: undefined,
@@ -67,6 +71,7 @@ export class MockSegmentStore implements Storage {
   private callbacks = {
     context: createCallbackManager<DeepPartial<Context> | undefined>(),
     settings: createCallbackManager<SegmentAPIIntegrations>(),
+    filters: createCallbackManager<DestinationFilters>(),
     userInfo: createCallbackManager<UserInfoState>(),
     deepLinkData: createCallbackManager<DeepLinkData>(),
   };
@@ -100,9 +105,8 @@ export class MockSegmentStore implements Storage {
     Settable<SegmentAPIIntegrations> &
     Dictionary<string, IntegrationSettings> = {
     get: createMockStoreGetter(() => this.data.settings),
-    onChange: (
-      callback: (value?: SegmentAPIIntegrations | undefined) => void
-    ) => this.callbacks.settings.register(callback),
+    onChange: (callback: (value?: SegmentAPIIntegrations) => void) =>
+      this.callbacks.settings.register(callback),
     set: (value) => {
       this.data.settings =
         value instanceof Function
@@ -117,6 +121,26 @@ export class MockSegmentStore implements Storage {
     },
   };
 
+  readonly filters: Watchable<DestinationFilters | undefined> &
+    Settable<DestinationFilters> &
+    Dictionary<string, RoutingRule> = {
+    get: createMockStoreGetter(() => this.data.filters),
+    onChange: (callback: (value?: DestinationFilters) => void) =>
+      this.callbacks.filters.register(callback),
+    set: (value) => {
+      this.data.filters =
+        value instanceof Function
+          ? value(this.data.filters ?? {})
+          : { ...value };
+      this.callbacks.filters.run(this.data.filters);
+      return this.data.filters;
+    },
+    add: (key: string, value: RoutingRule) => {
+      this.data.filters[key] = value;
+      this.callbacks.filters.run(this.data.filters);
+    },
+  };
+
   readonly userInfo: Watchable<UserInfoState> & Settable<UserInfoState> = {
     get: createMockStoreGetter(() => this.data.userInfo),
     onChange: (callback: (value: UserInfoState) => void) =>

packages/core/src/analytics.ts

@@ -3,7 +3,7 @@ import { Unsubscribe } from '@segment/sovran-react-native';
 import deepmerge from 'deepmerge';
 import allSettled from 'promise.allsettled';
 import { AppState, AppStateStatus } from 'react-native';
-import { settingsCDN } from './constants';
+import { settingsCDN, workspaceDestinationFilterKey } from './constants';
 import { getContext } from './context';
 import {
   applyRawEventData,
@@ -42,8 +42,11 @@ import {
 } from './types';
 import { getPluginsWithFlush, getPluginsWithReset } from './util';
 import { getUUID } from './uuid';
+import type { SegmentAPISettings, DestinationFilters } from './types';
+import type { Rule } from '@segment/tsub/dist/store';
 
 type OnContextLoadCallback = (type: UpdateType) => void | Promise<void>;
+type OnPluginAddedCallback = (plugin: Plugin) => void;
 
 export class SegmentClient {
   // the config parameters for the client - a merge of user provided and default options
@@ -82,7 +85,8 @@ export class SegmentClient {
 
   private isContextLoaded = false;
 
-  private onContextLoadedCallback: OnContextLoadCallback | undefined;
+  private onContextLoadedObservers: OnContextLoadCallback[] = [];
+  private onPluginAddedObservers: OnPluginAddedCallback[] = [];
 
   private readonly platformPlugins: PlatformPlugin[] = [
     new InjectUserInfo(),
@@ -106,6 +110,11 @@ export class SegmentClient {
    */
   readonly settings: Watchable<SegmentAPIIntegrations | undefined>;
 
+  /**
+   * Access or subscribe to destination filter settings
+   */
+  readonly filters: Watchable<DestinationFilters | undefined>;
+
   /**
    * Access or subscribe to user info (anonymousId, userId, traits)
    */
@@ -181,6 +190,11 @@ export class SegmentClient {
       onChange: this.store.settings.onChange,
     };
 
+    this.filters = {
+      get: this.store.filters.get,
+      onChange: this.store.filters.onChange,
+    };
+
     this.userInfo = {
       get: this.store.userInfo.get,
       set: this.store.userInfo.set,
@@ -238,15 +252,32 @@ export class SegmentClient {
     this.isInitialized = true;
   }
 
+  private generateFiltersMap(rules: Rule[]): DestinationFilters {
+    const map: DestinationFilters = {};
+
+    for (const r of rules) {
+      const key = r.destinationName ?? workspaceDestinationFilterKey;
+      map[key] = r;
+    }
+
+    return map;
+  }
+
   async fetchSettings() {
     const settingsEndpoint = `${settingsCDN}/${this.config.writeKey}/settings`;
 
     try {
       const res = await fetch(settingsEndpoint);
-      const resJson = await res.json();
+      const resJson: SegmentAPISettings = await res.json();
       const integrations = resJson.integrations;
+      const filters = this.generateFiltersMap(
+        resJson.middlewareSettings?.routingRules ?? []
+      );
       this.logger.info(`Received settings from Segment succesfully.`);
-      await this.store.settings.set(integrations);
+      await Promise.all([
+        this.store.settings.set(integrations),
+        this.store.filters.set(filters),
+      ]);
     } catch {
       this.logger.warn(
         `Could not receive settings from Segment. ${
@@ -366,6 +397,7 @@ export class SegmentClient {
   private addPlugin(plugin: Plugin) {
     plugin.configure(this);
     this.timeline.add(plugin);
+    this.triggerOnPluginLoaded(plugin);
   }
 
   /**
@@ -537,8 +569,8 @@ export class SegmentClient {
     await this.store.context.set(deepmerge(previousContext ?? {}, context));
 
     // Only callback during the intial context load
-    if (this.onContextLoadedCallback !== undefined && !this.isContextLoaded) {
-      this.onContextLoadedCallback(UpdateType.initial);
+    if (!this.isContextLoaded) {
+      this.triggerOnContextLoad(UpdateType.initial);
     }
 
     this.isContextLoaded = true;
@@ -651,9 +683,25 @@ export class SegmentClient {
    * @param callback Function to call when context is ready.
    */
   onContextLoaded(callback: OnContextLoadCallback) {
-    this.onContextLoadedCallback = callback;
+    this.onContextLoadedObservers.push(callback);
     if (this.isContextLoaded) {
-      this.onContextLoadedCallback(UpdateType.initial);
+      this.triggerOnContextLoad(UpdateType.initial);
     }
   }
+
+  private triggerOnContextLoad(type: UpdateType) {
+    this.onContextLoadedObservers.map((f) => f?.(type));
+  }
+
+  /**
+   * Registers a callback for each plugin that gets added to the analytics client.
+   * @param callback Function to call
+   */
+  onPluginLoaded(callback: OnPluginAddedCallback) {
+    this.onPluginAddedObservers.push(callback);
+  }
+
+  private triggerOnPluginLoaded(plugin: Plugin) {
+    this.onPluginAddedObservers.map((f) => f?.(plugin));
+  }
 }

packages/core/src/constants.ts

@@ -14,3 +14,5 @@ export const defaultConfig: Config = {
   trackAppLifecycleEvents: false,
   autoAddSegmentDestination: true,
 };
+
+export const workspaceDestinationFilterKey = '';

packages/core/src/storage/sovranStorage.ts

@@ -12,6 +12,8 @@ import type {
   DeepPartial,
   Context,
   UserInfoState,
+  RoutingRule,
+  DestinationFilters,
 } from '..';
 import { getUUID } from '../uuid';
 import { createGetter } from './helpers';
@@ -26,20 +28,20 @@ import type {
 } from './types';
 
 type Data = {
-  isReady: boolean;
   events: SegmentEvent[];
   eventsToRetry: SegmentEvent[];
   context: DeepPartial<Context>;
   settings: SegmentAPIIntegrations;
   userInfo: UserInfoState;
+  filters: DestinationFilters;
 };
 
 const INITIAL_VALUES: Data = {
-  isReady: true,
   events: [],
   eventsToRetry: [],
   context: {},
   settings: {},
+  filters: {},
   userInfo: {
     anonymousId: getUUID(),
     userId: undefined,
@@ -52,6 +54,7 @@ interface ReadinessStore {
   hasRestoredContext: boolean;
   hasRestoredSettings: boolean;
   hasRestoredUserInfo: boolean;
+  hasRestoredFilters: boolean;
 }
 
 const isEverythingReady = (state: ReadinessStore) =>
@@ -89,7 +92,7 @@ registerBridgeStore({
 });
 
 function createStoreGetter<
-  U,
+  U extends {},
   Z extends keyof U | undefined = undefined,
   V = undefined
 >(store: Store<U>, key?: Z): getStateFunc<Z extends keyof U ? V : U> {
@@ -120,6 +123,7 @@ export class SovranStorage implements Storage {
   private settingsStore: Store<{ settings: SegmentAPIIntegrations }>;
   private userInfoStore: Store<{ userInfo: UserInfoState }>;
   private deepLinkStore: Store<DeepLinkData> = deepLinkStore;
+  private filtersStore: Store<DestinationFilters>;
 
   readonly isReady: Watchable<boolean>;
 
@@ -130,6 +134,10 @@ export class SovranStorage implements Storage {
     Settable<SegmentAPIIntegrations> &
     Dictionary<string, IntegrationSettings>;
 
+  readonly filters: Watchable<DestinationFilters | undefined> &
+    Settable<DestinationFilters> &
+    Dictionary<string, RoutingRule>;
+
   readonly userInfo: Watchable<UserInfoState> & Settable<UserInfoState>;
 
   readonly deepLinkData: Watchable<DeepLinkData>;
@@ -142,6 +150,7 @@ export class SovranStorage implements Storage {
       hasRestoredContext: false,
       hasRestoredSettings: false,
       hasRestoredUserInfo: false,
+      hasRestoredFilters: false,
     });
 
     const markAsReadyGenerator = (key: keyof ReadinessStore) => () => {
@@ -240,6 +249,40 @@ export class SovranStorage implements Storage {
       },
     };
 
+    // Filters
+
+    this.filtersStore = createStore(INITIAL_VALUES.filters, {
+      persist: {
+        storeId: `${this.storeId}-filters`,
+        persistor: this.storePersistor,
+        onInitialized: markAsReadyGenerator('hasRestoredFilters'),
+      },
+    });
+
+    this.filters = {
+      get: createStoreGetter(this.filtersStore),
+      onChange: (callback: (value?: DestinationFilters | undefined) => void) =>
+        this.filtersStore.subscribe((store) => callback(store)),
+      set: async (value) => {
+        const filters = await this.filtersStore.dispatch((state) => {
+          let newState: typeof state;
+          if (value instanceof Function) {
+            newState = value(state);
+          } else {
+            newState = { ...state, ...value };
+          }
+          return newState;
+        });
+        return filters;
+      },
+      add: (key, value) => {
+        this.filtersStore.dispatch((state) => ({
+          ...state,
+          [key]: value,
+        }));
+      },
+    };
+
     // User Info Store
 
     this.userInfoStore = createStore(

packages/core/src/storage/types.ts

@@ -2,7 +2,9 @@ import type { Unsubscribe, Persistor } from '@segment/sovran-react-native';
 import type {
   Context,
   DeepPartial,
+  DestinationFilters,
   IntegrationSettings,
+  RoutingRule,
   SegmentAPIIntegrations,
   UserInfoState,
 } from '../types';
@@ -62,6 +64,10 @@ export interface Storage {
     Settable<SegmentAPIIntegrations> &
     Dictionary<string, IntegrationSettings>;
 
+  readonly filters: Watchable<DestinationFilters | undefined> &
+    Settable<DestinationFilters> &
+    Dictionary<string, RoutingRule>;
+
   readonly userInfo: Watchable<UserInfoState> & Settable<UserInfoState>;
 
   readonly deepLinkData: Watchable<DeepLinkData>;

packages/core/src/types.ts

@@ -1,4 +1,5 @@
 import type { Persistor } from '@segment/sovran-react-native';
+import type { Rule } from '@segment/tsub/dist/store';
 
 export type JsonValue =
   | boolean
@@ -270,8 +271,25 @@ export type SegmentAPIIntegrations = {
   [key: string]: IntegrationSettings;
 };
 
+export type RoutingRule = Rule;
+
+export interface MetricsOptions {
+  host?: string;
+  sampleRate?: number;
+  flushTimer?: number;
+  maxQueueSize?: number;
+}
+
+export interface DestinationFilters {
+  [key: string]: RoutingRule;
+}
+
 export type SegmentAPISettings = {
   integrations: SegmentAPIIntegrations;
+  middlewareSettings?: {
+    routingRules: RoutingRule[];
+  };
+  metrics?: MetricsOptions;
 };
 
 export type DestinationMetadata = {