feat: add exposure service

packages/node/src/assignment/assignment-filter.ts

@@ -3,6 +3,11 @@ import { Cache } from '../util/cache';
 import { Assignment, AssignmentFilter } from './assignment';
 import { DAY_MILLIS } from './assignment-service';
 
+/**
+ * @deprecated Assignment tracking is deprecated. Use Exposure tracking.
+ * Making this class a synonym for InMemoryExposureFilter. They perform same function.
+ * This class can be removed in the future with little effort.
+ */
 export class InMemoryAssignmentFilter implements AssignmentFilter {
   private readonly cache: Cache<number>;
 

packages/node/src/assignment/assignment-service.ts

@@ -8,6 +8,9 @@ import { Assignment, AssignmentFilter, AssignmentService } from './assignment';
 export const DAY_MILLIS = 24 * 60 * 60 * 1000;
 export const FLAG_TYPE_MUTUAL_EXCLUSION_GROUP = 'mutual-exclusion-group';
 
+/**
+ * @deprecated Assignment tracking is deprecated. Use Exposure tracking.
+ */
 export class AmplitudeAssignmentService implements AssignmentService {
   private readonly amplitude: CoreClient;
   private readonly assignmentFilter: AssignmentFilter;
@@ -24,6 +27,9 @@ export class AmplitudeAssignmentService implements AssignmentService {
   }
 }
 
+/**
+ * @deprecated Assignment tracking is deprecated. Use Exposure tracking.
+ */
 export const toEvent = (assignment: Assignment): BaseEvent => {
   const event: BaseEvent = {
     event_type: '[Experiment] Assignment',

packages/node/src/assignment/assignment.ts

@@ -2,14 +2,25 @@ import { EvaluationVariant } from '@amplitude/experiment-core';
 
 import { ExperimentUser } from '../types/user';
 
+/**
+ * @deprecated Assignment tracking is deprecated. Use Exposure tracking.
+ */
 export interface AssignmentService {
   track(assignment: Assignment): Promise<void>;
 }
 
+/**
+ * @deprecated Assignment tracking is deprecated. Use Exposure tracking.
+ */
 export interface AssignmentFilter {
   shouldTrack(assignment: Assignment): boolean;
 }
 
+/**
+ * @deprecated Assignment tracking is deprecated. Use Exposure tracking.
+ * Making this class a synonym for Exposure. They have the same fields.
+ * This class can be removed in the future with little effort.
+ */
 export class Assignment {
   public user: ExperimentUser;
   public results: Record<string, EvaluationVariant>;

packages/node/src/exposure/exposure-filter.ts

@@ -0,0 +1,27 @@
+import { Cache } from 'src/util/cache';
+
+import { Exposure, ExposureFilter } from './exposure';
+import { DAY_MILLIS } from './exposure-service';
+
+export class InMemoryExposureFilter implements ExposureFilter {
+  public ttlMillis: number;
+  private readonly cache: Cache<number>;
+
+  constructor(size: number, ttlMillis: number = DAY_MILLIS) {
+    this.ttlMillis = ttlMillis;
+    this.cache = new Cache<number>(size, ttlMillis);
+  }
+
+  public shouldTrack(exposure: Exposure): boolean {
+    if (Object.keys(exposure.results).length === 0) {
+      // Don't track empty exposures.
+      return false;
+    }
+    const canonicalExposure = exposure.canonicalize();
+    const track = this.cache.get(canonicalExposure) == undefined;
+    if (track) {
+      this.cache.put(canonicalExposure, 0);
+    }
+    return track;
+  }
+}

packages/node/src/exposure/exposure-service.ts

@@ -0,0 +1,98 @@
+import { BaseEvent, CoreClient } from '@amplitude/analytics-types';
+import { EvaluationVariant } from '@amplitude/experiment-core';
+import { ExperimentUser } from 'src/types/user';
+import { hashCode } from 'src/util/hash';
+
+import { Exposure, ExposureFilter, ExposureService } from './exposure';
+
+export const DAY_MILLIS = 24 * 60 * 60 * 1000;
+export const FLAG_TYPE_MUTUAL_EXCLUSION_GROUP = 'mutual-exclusion-group';
+
+export class AmplitudeExposureService implements ExposureService {
+  private readonly amplitude: CoreClient;
+  private readonly exposureFilter: ExposureFilter;
+
+  constructor(amplitude: CoreClient, exposureFilter: ExposureFilter) {
+    this.amplitude = amplitude;
+    this.exposureFilter = exposureFilter;
+  }
+
+  async track(exposure: Exposure): Promise<void> {
+    if (this.exposureFilter.shouldTrack(exposure)) {
+      toExposureEvents(exposure, this.exposureFilter.ttlMillis).forEach(
+        (event) => {
+          this.amplitude.logEvent(event);
+        },
+      );
+    }
+  }
+}
+
+export const toExposureEvents = (
+  exposure: Exposure,
+  ttlMillis: number,
+): BaseEvent[] => {
+  const events: BaseEvent[] = [];
+  const canonicalExposure = exposure.canonicalize();
+  for (const flagKey in exposure.results) {
+    const variant = exposure.results[flagKey];
+
+    const trackExposure = (variant?.metadata?.trackExposure as boolean) ?? true;
+    if (!trackExposure) {
+      continue;
+    }
+
+    const flagType = variant.metadata?.flagType;
+    const isDefault: boolean = variant.metadata?.default as boolean;
+    if (isDefault) {
+      continue;
+    }
+
+    // Determine user properties to set and unset.
+    const set = {};
+    const unset = {};
+    if (flagType != FLAG_TYPE_MUTUAL_EXCLUSION_GROUP) {
+      if (variant.key) {
+        set[`[Experiment] ${flagKey}`] = variant.key;
+      } else if (variant.value) {
+        set[`[Experiment] ${flagKey}`] = variant.value;
+      }
+    }
+
+    // Build event properties.
+    const eventProperties = {};
+    eventProperties['[Experiment] Flag Key'] = flagKey;
+    if (variant.key) {
+      eventProperties['[Experiment] Variant'] = variant.key;
+    } else if (variant.value) {
+      eventProperties['[Experiment] Variant'] = variant.value;
+    }
+    if (variant.metadata) {
+      eventProperties['metadata'] = variant.metadata;
+    }
+
+    // Build event.
+    const event: BaseEvent = {
+      event_type: '[Experiment] Exposure',
+      user_id: exposure.user.user_id,
+      device_id: exposure.user.device_id,
+      event_properties: eventProperties,
+      user_properties: {
+        $set: set,
+        $unset: unset,
+      },
+      insert_id: `${exposure.user.user_id} ${
+        exposure.user.device_id
+      } ${hashCode(flagKey + ' ' + canonicalExposure)} ${Math.floor(
+        exposure.timestamp / ttlMillis,
+      )}`,
+    };
+    if (exposure.user.groups) {
+      event.groups = exposure.user.groups;
+    }
+
+    events.push(event);
+  }
+
+  return events;
+};

packages/node/src/exposure/exposure.ts

@@ -0,0 +1,40 @@
+import { EvaluationVariant } from '@amplitude/experiment-core';
+import { ExperimentUser } from 'src/types/user';
+
+export interface ExposureService {
+  track(exposure: Exposure): Promise<void>;
+}
+
+export interface ExposureFilter {
+  shouldTrack(exposure: Exposure): boolean;
+  ttlMillis: number;
+}
+
+/**
+ * Exposure is a class that represents a user's exposure to a set of flags.
+ * It implements the FilterItem interface so it can be used with the dedupe filter.
+ */
+export class Exposure {
+  public user: ExperimentUser;
+  public results: Record<string, EvaluationVariant>;
+  public timestamp: number = Date.now();
+
+  public constructor(
+    user: ExperimentUser,
+    results: Record<string, EvaluationVariant>,
+  ) {
+    this.user = user;
+    this.results = results;
+  }
+
+  public canonicalize(): string {
+    let canonical = `${this.user.user_id?.trim()} ${this.user.device_id?.trim()} `;
+    for (const key of Object.keys(this.results).sort()) {
+      const variant = this.results[key];
+      if (variant?.key) {
+        canonical += key.trim() + ' ' + variant?.key?.trim() + ' ';
+      }
+    }
+    return canonical;
+  }
+}

packages/node/src/local/client.ts

@@ -5,6 +5,9 @@ import {
   topologicalSort,
 } from '@amplitude/experiment-core';
 import EventSource from 'eventsource';
+import { Exposure, ExposureService } from 'src/exposure/exposure';
+import { InMemoryExposureFilter } from 'src/exposure/exposure-filter';
+import { AmplitudeExposureService } from 'src/exposure/exposure-service';
 
 import { Assignment, AssignmentService } from '../assignment/assignment';
 import { InMemoryAssignmentFilter } from '../assignment/assignment-filter';
@@ -15,6 +18,8 @@ import { USER_GROUP_TYPE } from '../types/cohort';
 import {
   AssignmentConfig,
   AssignmentConfigDefaults,
+  ExposureConfig,
+  ExposureConfigDefaults,
   LocalEvaluationConfig,
 } from '../types/config';
 import { FlagConfigCache } from '../types/flag';
@@ -48,6 +53,13 @@ const STREAM_TRY_DELAY_MILLIS = 1000; // The delay between attempts.
 
 const COHORT_POLLING_INTERVAL_MILLIS_MIN = 60000;
 
+export type EvaluateOptions = {
+  /**
+   * Whether to track exposure event for the evaluation.
+   */
+  tracksExposure?: boolean;
+};
+
 /**
  * Experiment client for evaluating variants for a user locally.
  * @category Core Usage
@@ -56,7 +68,11 @@ export class LocalEvaluationClient {
   private readonly logger: Logger;
   protected readonly config: LocalEvaluationConfig;
   private readonly updater: FlagConfigUpdater;
+  /**
+   * @deprecated use {@link exposureService} instead to track exposure events.
+   */
   private readonly assignmentService: AssignmentService;
+  private readonly exposureService: ExposureService;
   private readonly evaluation: EvaluationEngine;
   private readonly cohortUpdater?: CohortUpdater;
 
@@ -148,6 +164,15 @@ export class LocalEvaluationClient {
         this.config.assignmentConfig,
       );
     }
+    const exposureConfig = {
+      ...ExposureConfigDefaults,
+      ...this.config.exposureConfig,
+    };
+    if (!exposureConfig.apiKey) {
+      // Use experiment deployment key if no api key is provided.
+      exposureConfig.apiKey = apiKey;
+    }
+    this.exposureService = this.createExposureService(exposureConfig);
     this.evaluation = new EvaluationEngine();
   }
 
@@ -163,6 +188,18 @@ export class LocalEvaluationClient {
     );
   }
 
+  private createExposureService(
+    exposureConfig: ExposureConfig,
+  ): ExposureService {
+    const instance = amplitude.createInstance();
+    const { apiKey, cacheCapacity, ...ampConfig } = exposureConfig;
+    instance.init(apiKey, ampConfig);
+    return new AmplitudeExposureService(
+      instance,
+      new InMemoryExposureFilter(cacheCapacity),
+    );
+  }
+
   /**
    * Locally evaluate varints for a user.
    *
@@ -182,6 +219,7 @@ export class LocalEvaluationClient {
   public evaluateV2(
     user: ExperimentUser,
     flagKeys?: string[],
+    options?: EvaluateOptions,
   ): Record<string, Variant> {
     const flags = this.cache.getAllCached() as Record<string, EvaluationFlag>;
     this.enrichUserWithCohorts(user, flags);
@@ -190,6 +228,9 @@ export class LocalEvaluationClient {
     const sortedFlags = topologicalSort(flags, flagKeys);
     const results = this.evaluation.evaluate(context, sortedFlags);
     void this.assignmentService?.track(new Assignment(user, results));
+    if (options?.tracksExposure) {
+      void this.exposureService?.track(new Exposure(user, results));
+    }
     this.logger.debug('[Experiment] evaluate - variants: ', results);
     return evaluationVariantsToVariants(results);
   }
@@ -291,8 +332,9 @@ export class LocalEvaluationClient {
   public async evaluate(
     user: ExperimentUser,
     flagKeys?: string[],
+    options?: EvaluateOptions,
   ): Promise<Variants> {
-    const results = this.evaluateV2(user, flagKeys);
+    const results = this.evaluateV2(user, flagKeys, options);
     return filterDefaultVariants(results);
   }