feat(experiment-tag): dual-write RTBT events to relay (WEB-131)

EventStorageManager writes to relay on addEvent, merges relay store on
syncFromRelay (migrate if needed, relay wins on dedup), and flushes relay
on unload/visibility. BehavioralTargetingManager exposes setRelayClient
and syncFromRelay for Pass 2 evaluation.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: tyiuhc

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

@@ -2,6 +2,7 @@ import { BehavioralTargetingRules } from '../types';
 
 import { BehavioralTargetingEvaluator } from './evaluator';
 import { EventStorageManager } from './event-storage';
+import { RelayClient } from './relay-client';
 import { SessionManager } from './session-manager';
 import { BehavioralTargeting } from './types';
 
@@ -51,6 +52,24 @@ export class BehavioralTargetingManager {
     this.evaluateEvent(eventType);
   }
 
+  /**
+   * Attach the relay client for cross-subdomain event dual-write.
+   */
+  public setRelayClient(relayClient: RelayClient | null): void {
+    this.eventStorage.setRelayClient(relayClient);
+  }
+
+  /**
+   * Pass 2: migrate local events to relay if needed, merge relay store, re-evaluate.
+   */
+  public async syncFromRelay(): Promise<boolean> {
+    const synced = await this.eventStorage.syncFromRelay();
+    if (synced) {
+      this.evaluateAll();
+    }
+    return synced;
+  }
+
   /**
    * Check if a flag has behavioral targeting rules.
    * @param flagKey The flag key to check

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,83 @@ 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,
+        });
+      }
+
+      const relayStore = await relay.readEvents();
+      this.mergeFromRelay(relayStore);
+      return true;
+    } catch {
+      return false;
+    }
   }
 
   /**
@@ -148,6 +240,7 @@ export class EventStorageManager {
    */
   flush(): void {
     this.flushToLocalStorage();
+    this.flushRelay();
   }
 
   /**
@@ -254,6 +347,7 @@ export class EventStorageManager {
    */
   private handleBeforeUnload = (): void => {
     this.flushToLocalStorage();
+    this.flushRelay();
   };
 
   /**
@@ -262,6 +356,7 @@ export class EventStorageManager {
   private handleVisibilityChange = (): void => {
     if (document.visibilityState === 'hidden') {
       this.flushToLocalStorage();
+      this.flushRelay();
     }
   };
 
@@ -272,6 +367,7 @@ export class EventStorageManager {
   cleanup(): void {
     // Flush any pending writes
     this.flushToLocalStorage();
+    this.flushRelay();
 
     // Clear debounce timeout
     if (this.debouncedWriteTimeout) {

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

@@ -1,6 +1,20 @@
-import { EventStorageManager } from 'src/behavioral-targeting/event-storage';
+import {
+  EventStorageManager,
+  eventDedupKey,
+} from 'src/behavioral-targeting/event-storage';
+import { RelayClient } from 'src/behavioral-targeting/relay-client';
 import { SessionManager } from 'src/behavioral-targeting/session-manager';
 
+type MockRelayClient = Pick<
+  RelayClient,
+  | 'relayAvailable'
+  | 'writeEvent'
+  | 'flush'
+  | 'readEvents'
+  | 'checkMigrated'
+  | 'migrateEvents'
+>;
+
 describe('EventStorageManager', () => {
   let eventStorage: EventStorageManager;
   let sessionManager: SessionManager;
@@ -411,4 +425,133 @@ describe('EventStorageManager', () => {
       expect(events[0].timestamp).toBeLessThanOrEqual(after);
     });
   });
+
+  describe('eventDedupKey', () => {
+    test('uses event_type and timestamp', () => {
+      expect(eventDedupKey({ event_type: 'click', timestamp: 1000 })).toBe(
+        'click:1000',
+      );
+    });
+  });
+
+  describe('relay dual-write', () => {
+    const createMockRelay = (available = true) => ({
+      relayAvailable: available,
+      writeEvent: jest.fn(),
+      flush: jest.fn(),
+      readEvents: jest.fn().mockResolvedValue({ events: [], nextId: 1 }),
+      checkMigrated: jest.fn().mockResolvedValue(true),
+      migrateEvents: jest.fn().mockResolvedValue(undefined),
+    });
+
+    test('writes events to relay on addEvent when relay is attached', () => {
+      const relay = createMockRelay();
+      eventStorage.setRelayClient(relay as unknown as RelayClient);
+      eventStorage.addEvent('click', { page: 'home' });
+
+      expect(relay.writeEvent).toHaveBeenCalledWith(
+        expect.objectContaining({
+          event_type: 'click',
+          properties: { page: 'home' },
+          id: 1,
+        }),
+      );
+    });
+
+    test('does not write to relay when relay is not attached', () => {
+      const relay = createMockRelay();
+      eventStorage.addEvent('click');
+      expect(relay.writeEvent).not.toHaveBeenCalled();
+    });
+
+    test('flush calls relay flush', () => {
+      const relay = createMockRelay();
+      eventStorage.setRelayClient(relay as unknown as RelayClient);
+      eventStorage.flush();
+      expect(relay.flush).toHaveBeenCalled();
+    });
+  });
+
+  describe('mergeFromRelay', () => {
+    test('merges relay events and relay wins on dedup key conflict', () => {
+      eventStorage.addEvent('click', { source: 'local' }, 1000);
+      const local = eventStorage.getAllEvents()[0];
+
+      eventStorage.mergeFromRelay({
+        events: [
+          {
+            ...local,
+            properties: { source: 'relay' },
+          },
+          {
+            id: 99,
+            event_type: 'view',
+            timestamp: 1001,
+            session_id: local.session_id,
+            properties: {},
+          },
+        ],
+        nextId: 100,
+      });
+
+      const events = eventStorage.getAllEvents();
+      expect(events).toHaveLength(2);
+      expect(events[0].properties).toEqual({ source: 'relay' });
+      expect(events[1].event_type).toBe('view');
+    });
+  });
+
+  describe('syncFromRelay', () => {
+    test('migrates local store when origin not migrated', async () => {
+      eventStorage.addEvent('click', {}, 1000);
+      const relay = {
+        relayAvailable: true,
+        writeEvent: jest.fn(),
+        flush: jest.fn(),
+        checkMigrated: jest.fn().mockResolvedValue(false),
+        migrateEvents: jest.fn().mockResolvedValue(undefined),
+        readEvents: jest.fn().mockResolvedValue({
+          events: [
+            {
+              id: 1,
+              event_type: 'click',
+              timestamp: 1000,
+              session_id: 's1',
+              properties: {},
+            },
+          ],
+          nextId: 2,
+        }),
+      };
+      eventStorage.setRelayClient(relay as unknown as RelayClient);
+
+      const synced = await eventStorage.syncFromRelay();
+
+      expect(synced).toBe(true);
+      expect(relay.migrateEvents).toHaveBeenCalledWith(
+        window.location.origin,
+        expect.objectContaining({
+          events: expect.arrayContaining([
+            expect.objectContaining({ event_type: 'click' }),
+          ]),
+        }),
+      );
+      expect(relay.readEvents).toHaveBeenCalled();
+    });
+
+    test('returns false when relay is unavailable', async () => {
+      const relay = {
+        relayAvailable: false,
+        writeEvent: jest.fn(),
+        flush: jest.fn(),
+        readEvents: jest.fn(),
+        checkMigrated: jest.fn(),
+        migrateEvents: jest.fn(),
+      };
+      eventStorage.setRelayClient(relay as unknown as RelayClient);
+
+      expect(await eventStorage.syncFromRelay()).toBe(false);
+      expect(relay.readEvents).not.toHaveBeenCalled();
+    });
+  });
 });