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

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

@@ -53,37 +53,25 @@ 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).
    */
   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).
+   * WEB-130: inject relay iframe (non-blocking init) and run Pass 2 sync.
    *
    * @returns true when matched behaviors changed after relay sync
    */

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

@@ -1,5 +1,17 @@
+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).
+ */
+export function eventDedupKey(event: {
+  event_type: string;
+  timestamp: number;
+}): string {
+  return `${event.event_type}:${event.timestamp}`;
+}
+
 /**
  * Represents a stored event record.
  */
@@ -33,15 +45,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();
@@ -86,6 +101,94 @@ export class EventStorageManager {
 
     // Trigger debounced write to localStorage
     this.scheduleDebouncedWrite();
+
+    // Fire-and-forget relay write when cross-subdomain sync is enabled
+    this.relayClient?.writeEvent(event);
+  }
+
+  /**
+   * 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(): Promise<boolean> {
+    const relay = 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);
+          }
+        }
+        relay.flush();
+      }
+
+      const relayStore = await relay.readEvents();
+      this.mergeFromRelay(relayStore);
+      return true;
+    } catch {
+      return false;
+    }
   }
 
   /**
@@ -148,6 +251,7 @@ export class EventStorageManager {
    */
   flush(): void {
     this.flushToLocalStorage();
+    this.flushRelay();
   }
 
   /**
@@ -254,6 +358,7 @@ export class EventStorageManager {
    */
   private handleBeforeUnload = (): void => {
     this.flushToLocalStorage();
+    this.flushRelay();
   };
 
   /**
@@ -262,6 +367,7 @@ export class EventStorageManager {
   private handleVisibilityChange = (): void => {
     if (document.visibilityState === 'hidden') {
       this.flushToLocalStorage();
+      this.flushRelay();
     }
   };
 
@@ -272,6 +378,7 @@ export class EventStorageManager {
   cleanup(): void {
     // Flush any pending writes
     this.flushToLocalStorage();
+    this.flushRelay();
 
     // Clear debounce timeout
     if (this.debouncedWriteTimeout) {

packages/experiment-tag/src/behavioral-targeting/relay-client.ts

@@ -189,15 +189,27 @@ export class RelayClient {
     if (this.destroyed) {
       return;
     }
+
+    // Queue until async write confirms — flush() can resend in-flight events on unload.
+    this.pendingWrites.push(event);
+
     if (!this.available || !this.iframeWindow) {
-      this.pendingWrites.push(event);
       return;
     }
-    void this.sendRequest(
-      this.createRelayRequest('WRITE_EVENT', { event }),
-    ).catch(() => {
-      // fire-and-forget
-    });
+
+    void this.sendRequest(this.createRelayRequest('WRITE_EVENT', { event }))
+      .then((response) => {
+        if (!response.ok) {
+          return;
+        }
+        const idx = this.pendingWrites.indexOf(event);
+        if (idx !== -1) {
+          this.pendingWrites.splice(idx, 1);
+        }
+      })
+      .catch(() => {
+        // Keep in pendingWrites for flush()
+      });
   }
 
   flush(): void {

packages/experiment-tag/test/behavioral-targeting/behavioral-targeting-manager-relay.test.ts

@@ -44,7 +44,7 @@ describe('BehavioralTargetingManager relay wiring', () => {
     return iframeWindow;
   };
 
-  test('beginRelaySync injects relay iframe and returns false without storage hooks', async () => {
+  test('beginRelaySync injects relay iframe and completes Pass 2 sync attempt', async () => {
     const manager = new BehavioralTargetingManager(API_KEY, {
       flag_a: {
         behavior_1: [
@@ -73,6 +73,5 @@ describe('BehavioralTargetingManager relay wiring', () => {
 
     expect(document.querySelector('iframe')).not.toBeNull();
     expect(relayClient.relayAvailable).toBe(true);
-    expect(await manager.syncFromRelay()).toBe(false);
   });
 });