feat(experiment-tag): Dual-write RTBT events to relay

packages/experiment-tag/src/behavioral-targeting/behavioral-targeting-manager.ts

@@ -3,6 +3,7 @@ import { BehavioralTargetingRules } from '../types';
 import { BehavioralTargetingEvaluator } from './evaluator';
 import { EventStorageManager } from './event-storage';
 import { RelayClient } from './relay-client';
+import { RelaySyncResult } from './relay-sync-result';
 import { SessionManager } from './session-manager';
 import { BehavioralTargeting } from './types';
 
@@ -53,52 +54,54 @@ export class BehavioralTargetingManager {
   }
 
   /**
-   * Attach the relay client for cross-subdomain event dual-write (#334).
+   * Attach the relay client for cross-subdomain event dual-write.
    */
   public setRelayClient(relayClient: RelayClient | null): void {
-    (
-      this.eventStorage as {
-        setRelayClient?: (client: RelayClient | null) => void;
-      }
-    ).setRelayClient?.(relayClient);
+    this.eventStorage.setRelayClient(relayClient);
   }
 
   /**
    * Pass 2: migrate local events to relay if needed, merge relay store, re-evaluate.
-   * Returns true when relay store was merged (#334).
+   * Returns true when relay store was merged.
    */
   public async syncFromRelay(): Promise<boolean> {
-    const sync = (
-      this.eventStorage as { syncFromRelay?: () => Promise<boolean> }
-    ).syncFromRelay;
-    if (!sync) {
-      return false;
-    }
-    const synced = await sync.call(this.eventStorage);
+    const synced = await this.eventStorage.syncFromRelay();
     if (synced) {
       this.evaluateAll();
     }
     return synced;
   }
 
   /**
-   * WEB-130: inject relay iframe (non-blocking init) and run Pass 2 sync when
-   * event-storage relay hooks are present (#334).
+   * Inject relay iframe (non-blocking init) and run Pass 2 sync.
    *
-   * @returns true when matched behaviors changed after relay sync
+   * Relay is attached to storage only after init succeeds so Pass 2 migration
+   * does not duplicate events dual-written during iframe startup.
    */
-  public async beginRelaySync(relayClient: RelayClient): Promise<boolean> {
+  public async beginRelaySync(
+    relayClient: RelayClient,
+  ): Promise<RelaySyncResult> {
     const behaviorsBefore = this.serializeMatchedBehaviors();
+
     await relayClient.init();
     if (!relayClient.relayAvailable) {
-      return false;
+      await relayClient.waitForAvailable();
+    }
+    if (!relayClient.relayAvailable) {
+      return { status: 'unavailable' };
     }
+
+    const synced = await this.eventStorage.syncFromRelay(relayClient);
     this.setRelayClient(relayClient);
-    const synced = await this.syncFromRelay();
+
     if (!synced) {
-      return false;
+      return { status: 'sync_failed' };
     }
-    return behaviorsBefore !== this.serializeMatchedBehaviors();
+
+    this.evaluateAll();
+    return behaviorsBefore !== this.serializeMatchedBehaviors()
+      ? { status: 'behaviors_changed' }
+      : { status: 'unchanged' };
   }
 
   /**

packages/experiment-tag/src/behavioral-targeting/event-storage.ts

@@ -1,5 +1,19 @@
+import { RelayClient } from './relay-client';
+import { RelayEventStorage } from './relay-protocol';
 import { SessionManager } from './session-manager';
 
+/**
+ * Dedup key for cross-subdomain merge (matches relay.js MIGRATE_EVENTS for
+ * migration; includes id so same-millisecond events do not collapse on merge).
+ */
+export function eventDedupKey(event: {
+  event_type: string;
+  timestamp: number;
+  id: number;
+}): string {
+  return `${event.event_type}:${event.timestamp}:${event.id}`;
+}
+
 /**
  * Represents a stored event record.
  */
@@ -33,15 +47,18 @@ export class EventStorageManager {
   private hasPendingWrites = false; // Track if cache has unsaved changes
   private persistedEvents?: Set<string>; // Optional set of event types to persist
   private storageKey: string;
+  private relayClient: RelayClient | null = null;
 
   constructor(
     apiKey: string,
     sessionManager: SessionManager,
     persistedEvents?: Set<string>,
+    relayClient?: RelayClient | null,
   ) {
     this.storageKey = `EXP_${apiKey.slice(0, 10)}_rtbt_events`;
     this.sessionManager = sessionManager;
     this.persistedEvents = persistedEvents;
+    this.relayClient = relayClient ?? null;
 
     // Load from localStorage into memory on initialization
     this.memoryCache = this.loadFromLocalStorage();
@@ -78,14 +95,105 @@ export class EventStorageManager {
     this.memoryCache.events.push(event);
 
     // Apply FIFO limit
-    if (this.memoryCache.events.length > EventStorageManager.MAX_EVENTS) {
+    const fifoTrimmed =
+      this.memoryCache.events.length > EventStorageManager.MAX_EVENTS;
+    if (fifoTrimmed) {
       this.memoryCache.events = this.memoryCache.events.slice(
         -EventStorageManager.MAX_EVENTS,
       );
     }
 
     // Trigger debounced write to localStorage
     this.scheduleDebouncedWrite();
+
+    this.relayClient?.writeEvent(event);
+    if (fifoTrimmed) {
+      this.reconcileRelayAfterFifoTrim();
+    }
+  }
+
+  /**
+   * Attach or detach the relay client for cross-subdomain dual-write.
+   */
+  setRelayClient(relayClient: RelayClient | null): void {
+    this.relayClient = relayClient;
+  }
+
+  /**
+   * Flushes pending relay writes (e.g. on page unload).
+   */
+  flushRelay(): void {
+    this.relayClient?.flush();
+  }
+
+  /**
+   * Merges relay events into the in-memory cache. Relay wins on dedup conflicts.
+   */
+  mergeFromRelay(relayStore: RelayEventStorage): void {
+    const byKey = new Map<string, EventRecord>();
+    for (const event of this.memoryCache.events) {
+      byKey.set(eventDedupKey(event), event);
+    }
+    for (const event of relayStore.events) {
+      byKey.set(eventDedupKey(event), event);
+    }
+
+    let events = Array.from(byKey.values()).sort(
+      (a, b) => a.timestamp - b.timestamp,
+    );
+    if (events.length > EventStorageManager.MAX_EVENTS) {
+      events = events.slice(-EventStorageManager.MAX_EVENTS);
+    }
+
+    let nextId = Math.max(this.memoryCache.nextId, relayStore.nextId);
+    for (const event of events) {
+      if (event.id + 1 > nextId) {
+        nextId = event.id + 1;
+      }
+    }
+
+    this.memoryCache = { events, nextId };
+    this.hasPendingWrites = true;
+    this.scheduleDebouncedWrite();
+  }
+
+  /**
+   * Pass 2 sync: migrate local store to relay if needed, then merge relay events.
+   * Returns true when sync completed; false when relay unavailable or RPC failed.
+   */
+  async syncFromRelay(relayClient?: RelayClient): Promise<boolean> {
+    const relay = relayClient ?? this.relayClient;
+    if (!relay?.relayAvailable) {
+      return false;
+    }
+
+    try {
+      const origin = window.location.origin;
+      const migrated = await relay.checkMigrated(origin);
+
+      if (!migrated && this.memoryCache.events.length > 0) {
+        await relay.migrateEvents(origin, {
+          events: [...this.memoryCache.events],
+          nextId: this.memoryCache.nextId,
+        });
+      } else if (migrated && this.memoryCache.events.length > 0) {
+        const existingRelayStore = await relay.readEvents();
+        const relayKeys = new Set(
+          existingRelayStore.events.map((e) => eventDedupKey(e)),
+        );
+        for (const event of this.memoryCache.events) {
+          if (!relayKeys.has(eventDedupKey(event))) {
+            relay.writeEvent(event);
+          }
+        }
+      }
+
+      const relayStore = await relay.readEvents();
+      this.mergeFromRelay(relayStore);
+      return true;
+    } catch {
+      return false;
+    }
   }
 
   /**
@@ -148,6 +256,7 @@ export class EventStorageManager {
    */
   flush(): void {
     this.flushToLocalStorage();
+    this.flushRelay();
   }
 
   /**
@@ -201,6 +310,23 @@ export class EventStorageManager {
    * Schedules a debounced write to localStorage.
    * Resets the timer on each call to batch multiple writes together.
    */
+  private reconcileRelayAfterFifoTrim(): void {
+    const relay = this.relayClient;
+    if (!relay?.relayAvailable) {
+      return;
+    }
+    void relay
+      .migrateEvents(window.location.origin, {
+        events: [...this.memoryCache.events],
+        nextId: this.memoryCache.nextId,
+      })
+      .catch(() => {
+        for (const cached of this.memoryCache.events) {
+          relay.writeEvent(cached);
+        }
+      });
+  }
+
   private scheduleDebouncedWrite(): void {
     this.hasPendingWrites = true;
 
@@ -253,15 +379,15 @@ export class EventStorageManager {
    * Handler for beforeunload event.
    */
   private handleBeforeUnload = (): void => {
-    this.flushToLocalStorage();
+    this.flush();
   };
 
   /**
    * Handler for visibilitychange event.
    */
   private handleVisibilityChange = (): void => {
     if (document.visibilityState === 'hidden') {
-      this.flushToLocalStorage();
+      this.flush();
     }
   };
 
@@ -272,6 +398,7 @@ export class EventStorageManager {
   cleanup(): void {
     // Flush any pending writes
     this.flushToLocalStorage();
+    this.flushRelay();
 
     // Clear debounce timeout
     if (this.debouncedWriteTimeout) {