✨ [RUM-2203] Forward replay records to the bridge

packages/core/src/transport/eventBridge.spec.ts

@@ -1,4 +1,5 @@
-import { deleteEventBridgeStub, initEventBridgeStub } from '../../test'
+import { PRIVACY_LEVEL_FROM_EVENT_BRIDGE, deleteEventBridgeStub, initEventBridgeStub } from '../../test'
+import type { DatadogEventBridge } from './eventBridge'
 import { getEventBridge, canUseEventBridge } from './eventBridge'
 
 describe('canUseEventBridge', () => {
@@ -33,7 +34,7 @@ describe('canUseEventBridge', () => {
   })
 })
 
-describe('getEventBridge', () => {
+describe('event bridge send', () => {
   let sendSpy: jasmine.Spy<(msg: string) => void>
 
   beforeEach(() => {
@@ -45,11 +46,44 @@ describe('getEventBridge', () => {
     deleteEventBridgeStub()
   })
 
-  it('event bridge should serialize sent events', () => {
+  it('should serialize sent events without view', () => {
     const eventBridge = getEventBridge()!
 
     eventBridge.send('view', { foo: 'bar' })
 
     expect(sendSpy).toHaveBeenCalledOnceWith('{"eventType":"view","event":{"foo":"bar"}}')
   })
+
+  it('should serialize sent events with view', () => {
+    const eventBridge = getEventBridge()!
+
+    eventBridge.send('view', { foo: 'bar' }, '123')
+
+    expect(sendSpy).toHaveBeenCalledOnceWith('{"eventType":"view","event":{"foo":"bar"},"view":{"id":"123"}}')
+  })
+})
+
+describe('event bridge getPrivacyLevel', () => {
+  let eventBridgeStub: DatadogEventBridge
+
+  beforeEach(() => {
+    eventBridgeStub = initEventBridgeStub()
+  })
+
+  afterEach(() => {
+    deleteEventBridgeStub()
+  })
+
+  it('should return the privacy level', () => {
+    const eventBridge = getEventBridge()!
+
+    expect(eventBridge.getPrivacyLevel()).toEqual(PRIVACY_LEVEL_FROM_EVENT_BRIDGE)
+  })
+
+  it('should return undefined if getPrivacyLevel not present in the bridge', () => {
+    delete eventBridgeStub.getPrivacyLevel
+    const eventBridge = getEventBridge()!
+
+    expect(eventBridge.getPrivacyLevel()).toBeUndefined()
+  })
 })

packages/core/src/transport/eventBridge.ts

@@ -6,6 +6,7 @@ export interface BrowserWindowWithEventBridge extends Window {
 }
 
 export interface DatadogEventBridge {
+  getPrivacyLevel?(): string
   getAllowedWebViewHosts(): string
   send(msg: string): void
 }
@@ -18,11 +19,15 @@ export function getEventBridge<T, E>() {
   }
 
   return {
+    getPrivacyLevel() {
+      return eventBridgeGlobal.getPrivacyLevel?.()
+    },
     getAllowedWebViewHosts() {
       return JSON.parse(eventBridgeGlobal.getAllowedWebViewHosts()) as string[]
     },
-    send(eventType: T, event: E) {
-      eventBridgeGlobal.send(JSON.stringify({ eventType, event }))
+    send(eventType: T, event: E, viewId?: string) {
+      const view = viewId ? { id: viewId } : undefined
+      eventBridgeGlobal.send(JSON.stringify({ eventType, event, view }))
     },
   }
 }

packages/core/test/emulate/eventBridge.ts

@@ -1,9 +1,11 @@
 import type { BrowserWindowWithEventBridge } from '../../src/transport'
 
-export function initEventBridgeStub(allowedWebViewHosts: string[] = [window.location.hostname]) {
+export const PRIVACY_LEVEL_FROM_EVENT_BRIDGE = 'allow'
+export function initEventBridgeStub(allowedWebViewHosts = [window.location.hostname]) {
   const eventBridgeStub = {
     send: (_msg: string) => undefined,
     getAllowedWebViewHosts: () => JSON.stringify(allowedWebViewHosts),
+    getPrivacyLevel: () => PRIVACY_LEVEL_FROM_EVENT_BRIDGE,
   }
   ;(window as BrowserWindowWithEventBridge).DatadogEventBridge = eventBridgeStub
   return eventBridgeStub

packages/logs/src/transport/startLogsBridge.ts

@@ -8,6 +8,6 @@ export function startLogsBridge(lifeCycle: LifeCycle) {
   const bridge = getEventBridge<'log', LogsEvent>()!
 
   lifeCycle.subscribe(LifeCycleEventType.LOG_COLLECTED, (serverLogsEvent: LogsEvent & Context) => {
-    bridge.send('log', serverLogsEvent)
+    bridge.send('log', serverLogsEvent, serverLogsEvent.view.id)
   })
 }

packages/rum-core/src/boot/rumPublicApi.spec.ts

@@ -23,6 +23,7 @@ import {
   deleteEventBridgeStub,
   cleanupSyntheticsWorkerValues,
   mockSyntheticsWorkerValues,
+  PRIVACY_LEVEL_FROM_EVENT_BRIDGE,
 } from '@datadog/browser-core/test'
 import type { TestSetupBuilder } from '../../test'
 import { setup, noopRecorderApi } from '../../test'
@@ -112,18 +113,34 @@ describe('rum public api', () => {
         deleteEventBridgeStub()
       })
 
-      it('init should accept empty application id and client token', () => {
+      it('should accept empty application id and client token', () => {
         const hybridInitConfiguration: HybridInitConfiguration = {}
         rumPublicApi.init(hybridInitConfiguration as RumInitConfiguration)
         expect(display.error).not.toHaveBeenCalled()
       })
 
-      it('init should force session sample rate to 100', () => {
+      it('should force session sample rate to 100', () => {
         const invalidConfiguration: HybridInitConfiguration = { sessionSampleRate: 50 }
         rumPublicApi.init(invalidConfiguration as RumInitConfiguration)
         expect(rumPublicApi.getInitConfiguration()?.sessionSampleRate).toEqual(100)
       })
 
+      it('should set the default privacy level received from the bridge if the not provided in the init configuration', () => {
+        const hybridInitConfiguration: HybridInitConfiguration = {}
+        rumPublicApi.init(hybridInitConfiguration as RumInitConfiguration)
+        expect((rumPublicApi.getInitConfiguration() as RumInitConfiguration)?.defaultPrivacyLevel).toEqual(
+          PRIVACY_LEVEL_FROM_EVENT_BRIDGE
+        )
+      })
+
+      it('should set the default privacy level from the init configuration if provided', () => {
+        const hybridInitConfiguration: HybridInitConfiguration = { defaultPrivacyLevel: 'mask' }
+        rumPublicApi.init(hybridInitConfiguration as RumInitConfiguration)
+        expect((rumPublicApi.getInitConfiguration() as RumInitConfiguration)?.defaultPrivacyLevel).toEqual(
+          hybridInitConfiguration.defaultPrivacyLevel
+        )
+      })
+
       it('should initialize even if session cannot be handled', () => {
         spyOnProperty(document, 'cookie', 'get').and.returnValue('')
         const rumPublicApi = makeRumPublicApi(startRumSpy, noopRecorderApi, {})

packages/rum-core/src/boot/rumPublicApi.ts

@@ -31,6 +31,7 @@ import {
   CustomerDataCompressionStatus,
   createCustomerDataTrackerManager,
   storeContextManager,
+  getEventBridge,
 } from '@datadog/browser-core'
 import type { LifeCycle } from '../domain/lifeCycle'
 import type { ViewContexts } from '../domain/contexts/viewContexts'
@@ -341,11 +342,12 @@ export function makeRumPublicApi(
     return true
   }
 
-  function overrideInitConfigurationForBridge<C extends InitConfiguration>(initConfiguration: C): C {
+  function overrideInitConfigurationForBridge<C extends RumInitConfiguration>(initConfiguration: C): C {
     return assign({}, initConfiguration, {
       applicationId: '00000000-aaaa-0000-aaaa-000000000000',
       clientToken: 'empty',
       sessionSampleRate: 100,
+      defaultPrivacyLevel: initConfiguration.defaultPrivacyLevel ?? getEventBridge()?.getPrivacyLevel(),
     })
   }
 }

packages/rum-core/src/boot/startRum.spec.ts

@@ -381,8 +381,13 @@ describe('view events', () => {
     clock.tick(VIEW_DURATION)
     window.dispatchEvent(createNewEvent('beforeunload'))
 
-    const lastBridgeMessage = JSON.parse(sendSpy.calls.mostRecent().args[0]) as { eventType: 'rum'; event: RumEvent }
+    const lastBridgeMessage = JSON.parse(sendSpy.calls.mostRecent().args[0]) as {
+      eventType: 'rum'
+      event: RumEvent
+      view: { id: string }
+    }
     expect(lastBridgeMessage.event.type).toBe('view')
     expect(lastBridgeMessage.event.view.time_spent).toBe(toServerDuration(VIEW_DURATION))
+    expect(lastBridgeMessage.view.id).toBeDefined()
   })
 })

packages/rum-core/src/domain/rumSessionManager.ts

@@ -58,7 +58,7 @@ export function startRumSessionManager(configuration: RumConfiguration, lifeCycl
 export function startRumSessionManagerStub(): RumSessionManager {
   const session: RumSession = {
     id: '00000000-aaaa-0000-aaaa-000000000000',
-    sessionReplayAllowed: false,
+    sessionReplayAllowed: true,
   }
   return {
     findTrackedSession: () => session,

packages/rum-core/src/transport/startRumEventBridge.ts

@@ -8,6 +8,6 @@ export function startRumEventBridge(lifeCycle: LifeCycle) {
   const bridge = getEventBridge<'rum', RumEvent>()!
 
   lifeCycle.subscribe(LifeCycleEventType.RUM_EVENT_COLLECTED, (serverRumEvent: RumEvent & Context) => {
-    bridge.send('rum', serverRumEvent)
+    bridge.send('rum', serverRumEvent, serverRumEvent.view.id)
   })
 }

packages/rum/src/boot/recorderApi.ts

@@ -1,5 +1,5 @@
 import type { DeflateEncoder } from '@datadog/browser-core'
-import { DeflateEncoderStreamId, canUseEventBridge, noop, runOnReadyState } from '@datadog/browser-core'
+import { DeflateEncoderStreamId, noop, runOnReadyState } from '@datadog/browser-core'
 import type {
   LifeCycle,
   ViewContexts,
@@ -53,7 +53,7 @@ export function makeRecorderApi(
   startRecordingImpl: StartRecording,
   createDeflateWorkerImpl?: CreateDeflateWorker
 ): RecorderApi {
-  if (canUseEventBridge() || !isBrowserSupported()) {
+  if (!isBrowserSupported()) {
     return {
       start: noop,
       stop: noop,

packages/rum/src/boot/startRecording.spec.ts

@@ -3,7 +3,12 @@ import { PageExitReason, DefaultPrivacyLevel, noop, isIE, DeflateEncoderStreamId
 import type { LifeCycle, ViewCreatedEvent, RumConfiguration } from '@datadog/browser-rum-core'
 import { LifeCycleEventType } from '@datadog/browser-rum-core'
 import type { Clock } from '@datadog/browser-core/test'
-import { collectAsyncCalls, createNewEvent } from '@datadog/browser-core/test'
+import {
+  collectAsyncCalls,
+  createNewEvent,
+  deleteEventBridgeStub,
+  initEventBridgeStub,
+} from '@datadog/browser-core/test'
 import type { RumSessionManagerMock, TestSetupBuilder } from '../../../rum-core/test'
 import { appendElement, createRumSessionManagerMock, setup } from '../../../rum-core/test'
 
@@ -81,6 +86,7 @@ describe('startRecording', () => {
     setupBuilder.cleanup()
     clock?.cleanup()
     resetDeflateWorkerState()
+    deleteEventBridgeStub()
   })
 
   it('sends recorded segments with valid context', async () => {
@@ -209,6 +215,22 @@ describe('startRecording', () => {
     })
   })
 
+  it('should send records through the bridge when it is present', () => {
+    const eventBridgeStub = initEventBridgeStub()
+    setupBuilder.build()
+    const sendSpy = spyOn(eventBridgeStub, 'send')
+
+    document.body.dispatchEvent(createNewEvent('click', { clientX: 1, clientY: 2 }))
+
+    const lastBridgeMessage = JSON.parse(sendSpy.calls.mostRecent().args[0])
+
+    expect(lastBridgeMessage).toEqual({
+      eventType: 'record',
+      event: jasmine.objectContaining({ type: RecordType.IncrementalSnapshot }),
+      view: { id: viewId },
+    })
+  })
+
   function changeView(lifeCycle: LifeCycle) {
     lifeCycle.notify(LifeCycleEventType.VIEW_ENDED, {} as any)
     viewId = 'view-id-2'

packages/rum/src/boot/startRecording.ts

@@ -1,10 +1,12 @@
 import type { RawError, HttpRequest, DeflateEncoder } from '@datadog/browser-core'
-import { createHttpRequest, addTelemetryDebug } from '@datadog/browser-core'
+import { createHttpRequest, addTelemetryDebug, noop, canUseEventBridge } from '@datadog/browser-core'
 import type { LifeCycle, ViewContexts, RumConfiguration, RumSessionManager } from '@datadog/browser-rum-core'
 import { LifeCycleEventType } from '@datadog/browser-rum-core'
 
 import { record } from '../domain/record'
 import { startSegmentCollection, SEGMENT_BYTES_LIMIT } from '../domain/segmentCollection'
+import type { BrowserRecord } from '../types'
+import { startRecordBridge } from '../domain/startRecordBridge'
 
 export function startRecording(
   lifeCycle: LifeCycle,
@@ -23,14 +25,21 @@ export function startRecording(
     httpRequest ||
     createHttpRequest(configuration, configuration.sessionReplayEndpointBuilder, SEGMENT_BYTES_LIMIT, reportError)
 
-  const { addRecord, stop: stopSegmentCollection } = startSegmentCollection(
-    lifeCycle,
-    configuration,
-    sessionManager,
-    viewContexts,
-    replayRequest,
-    encoder
-  )
+  let addRecord = (_record: BrowserRecord) => {}
+  let stopSegmentCollection = noop
+
+  if (!canUseEventBridge()) {
+    ;({ addRecord, stop: stopSegmentCollection } = startSegmentCollection(
+      lifeCycle,
+      configuration,
+      sessionManager,
+      viewContexts,
+      replayRequest,
+      encoder
+    ))
+  } else {
+    ;({ addRecord } = startRecordBridge(viewContexts))
+  }
 
   const { stop: stopRecording } = record({
     emit: addRecord,

packages/rum/src/domain/startRecordBridge.ts

@@ -0,0 +1,17 @@
+import { getEventBridge } from '@datadog/browser-core'
+import type { ViewContexts } from '@datadog/browser-rum-core'
+import type { BrowserRecord } from '../types'
+
+export function startRecordBridge(viewContexts: ViewContexts) {
+  const bridge = getEventBridge<'record', BrowserRecord>()!
+
+  return {
+    addRecord: (record: BrowserRecord) => {
+      const view = viewContexts.findView()
+      if (!view) {
+        return
+      }
+      bridge.send('record', record, view.id)
+    },
+  }
+}

test/e2e/lib/framework/intakeRegistry.ts

@@ -121,6 +121,10 @@ export class IntakeRegistry {
   get replaySegments() {
     return this.replayRequests.map((request) => request.segment)
   }
+
+  get replayRecords() {
+    return this.replayRequests.flatMap((request) => request.segment.records)
+  }
 }
 
 function isLogsIntakeRequest(request: IntakeRequest): request is LogsIntakeRequest {

test/e2e/lib/framework/serverApps/intake.ts

@@ -58,7 +58,7 @@ function computeIntakeRequestInfos(req: express.Request): IntakeRequestInfos {
     return {
       isBridge: true,
       encoding,
-      intakeType: eventType === 'log' ? 'logs' : 'rum',
+      intakeType: eventType === 'log' ? 'logs' : eventType === 'record' ? 'replay' : 'rum',
     }
   }
 
@@ -104,6 +104,23 @@ function readReplayIntakeRequest(
   infos: IntakeRequestInfos & { intakeType: 'replay' }
 ): Promise<ReplayIntakeRequest> {
   return new Promise((resolve, reject) => {
+    if (infos.isBridge) {
+      readStream(req)
+        .then((rawBody) => {
+          resolve({
+            ...infos,
+            segment: {
+              records: rawBody
+                .toString('utf-8')
+                .split('\n')
+                .map((line): any => JSON.parse(line)),
+            },
+          } as ReplayIntakeRequest)
+        })
+        .catch(reject)
+      return
+    }
+
     let segmentPromise: Promise<{
       encoding: string
       filename: string

test/e2e/scenario/eventBridge.scenario.ts

@@ -91,4 +91,14 @@ describe('bridge present', () => {
       expect(intakeRegistry.logsEvents.length).toBe(1)
       expect(intakeRegistry.hasOnlyBridgeRequests).toBe(true)
     })
+
+  createTest('send records to the bridge')
+    .withRum()
+    .withEventBridge()
+    .run(async ({ intakeRegistry }) => {
+      await flushEvents()
+
+      expect(intakeRegistry.replayRecords.length).toBeGreaterThan(0)
+      expect(intakeRegistry.hasOnlyBridgeRequests).toBe(true)
+    })
 })