Benoit/simplify replica

packages/core/src/domain/configuration/endpointBuilder.ts

@@ -22,16 +22,17 @@ export type EndpointBuilder = ReturnType<typeof createEndpointBuilder>
 export function createEndpointBuilder(
   initConfiguration: InitConfiguration,
   trackType: TrackType,
-  configurationTags: string[]
+  extraTags: string[],
+  extraParameters?: string[]
 ) {
   const buildUrlWithParameters = createEndpointUrlWithParametersBuilder(initConfiguration, trackType)
 
   return {
     build(api: ApiType, payload: Payload) {
-      const parameters = buildEndpointParameters(initConfiguration, trackType, configurationTags, api, payload)
+      const parameters = buildEndpointParameters(initConfiguration, trackType, api, payload, extraTags, extraParameters)
       return buildUrlWithParameters(parameters)
     },
-    tags: configurationTags,
+    tags: extraTags,
     urlPrefix: buildUrlWithParameters(''),
     trackType,
   }
@@ -89,11 +90,12 @@ export function buildEndpointHost(
 function buildEndpointParameters(
   { clientToken, internalAnalyticsSubdomain }: InitConfiguration,
   trackType: TrackType,
-  configurationTags: string[],
   api: ApiType,
-  { retry, encoding }: Payload
+  { retry, encoding }: Payload,
+  extraTags: string[],
+  extraParameters: string[] = []
 ) {
-  const tags = [`sdk_version:${__BUILD_ENV__SDK_VERSION__}`, `api:${api}`].concat(configurationTags)
+  const tags = [`sdk_version:${__BUILD_ENV__SDK_VERSION__}`, `api:${api}`].concat(extraTags)
   if (retry) {
     tags.push(`retry_count:${retry.count}`, `retry_after:${retry.lastFailureStatus}`)
   }
@@ -105,7 +107,7 @@ function buildEndpointParameters(
     `dd-evp-origin-version=${encodeURIComponent(__BUILD_ENV__SDK_VERSION__)}`,
     'dd-evp-origin=browser',
     `dd-request-id=${generateUUID()}`,
-  ]
+  ].concat(extraParameters)
 
   if (encoding) {
     parameters.push(`dd-evp-encoding=${encoding}`)

packages/core/src/domain/configuration/transportConfiguration.ts

@@ -15,7 +15,6 @@ export interface TransportConfiguration {
 }
 
 export interface ReplicaConfiguration {
-  applicationId?: string
   logsEndpointBuilder: EndpointBuilder
   rumEndpointBuilder: EndpointBuilder
 }
@@ -58,12 +57,12 @@ function computeReplicaConfiguration(
     clientToken: initConfiguration.replica.clientToken,
   }
 
-  const replicaEndpointBuilders = {
+  return {
     logsEndpointBuilder: createEndpointBuilder(replicaConfiguration, 'logs', tags),
-    rumEndpointBuilder: createEndpointBuilder(replicaConfiguration, 'rum', tags),
+    rumEndpointBuilder: createEndpointBuilder(replicaConfiguration, 'rum', tags, [
+      `application.id=${initConfiguration.replica.applicationId}`,
+    ]),
   }
-
-  return { applicationId: initConfiguration.replica.applicationId, ...replicaEndpointBuilders }
 }
 
 export function isIntakeUrl(url: string): boolean {

packages/core/src/domain/deflate/deflate.types.ts

@@ -51,7 +51,5 @@ export type DeflateEncoder = Encoder<Uint8Array> & { stop: () => void }
 export const enum DeflateEncoderStreamId {
   REPLAY = 1,
   RUM = 2,
-  RUM_REPLICA = 3,
   TELEMETRY = 4,
-  TELEMETRY_REPLICA = 5,
 }

packages/core/src/domain/telemetry/telemetry.ts

@@ -18,7 +18,13 @@ import type { StackTrace } from '../../tools/stackTrace/computeStackTrace'
 import { computeStackTrace } from '../../tools/stackTrace/computeStackTrace'
 import { getConnectivity } from '../connectivity'
 import { createBoundedBuffer } from '../../tools/boundedBuffer'
-import { canUseEventBridge, getEventBridge, startBatchWithReplica } from '../../transport'
+import {
+  canUseEventBridge,
+  createFlushController,
+  createHttpRequest,
+  getEventBridge,
+  createBatch,
+} from '../../transport'
 import type { Encoder } from '../../tools/encoder'
 import type { PageMayExitEvent } from '../../browser/pageMayExitObservable'
 import { DeflateEncoderStreamId } from '../deflate'
@@ -167,27 +173,27 @@ function startTelemetryTransport(
     const telemetrySubscription = telemetryObservable.subscribe((event) => bridge.send('internal_telemetry', event))
     cleanupTasks.push(() => telemetrySubscription.unsubscribe())
   } else {
-    const telemetryBatch = startBatchWithReplica(
-      configuration,
-      {
-        endpoint: configuration.rumEndpointBuilder,
-        encoder: createEncoder(DeflateEncoderStreamId.TELEMETRY),
-      },
-      configuration.replica && {
-        endpoint: configuration.replica.rumEndpointBuilder,
-        encoder: createEncoder(DeflateEncoderStreamId.TELEMETRY_REPLICA),
-      },
-      reportError,
-      pageMayExitObservable,
-
-      // We don't use an actual session expire observable here, to make telemetry collection
-      // independent of the session. This allows to start and send telemetry events ealier.
-      new Observable()
-    )
+    const endpoints = [configuration.rumEndpointBuilder]
+    if (configuration.replica && isTelemetryReplicationAllowed(configuration)) {
+      endpoints.push(configuration.replica.rumEndpointBuilder)
+    }
+    const telemetryBatch = createBatch({
+      encoder: createEncoder(DeflateEncoderStreamId.TELEMETRY),
+      request: createHttpRequest(endpoints, configuration.batchBytesLimit, reportError),
+      flushController: createFlushController({
+        messagesLimit: configuration.batchMessagesLimit,
+        bytesLimit: configuration.batchBytesLimit,
+        durationLimit: configuration.flushTimeout,
+        pageMayExitObservable,
+
+        // We don't use an actual session expire observable here, to make telemetry collection
+        // independent of the session. This allows to start and send telemetry events earlier.
+        sessionExpireObservable: new Observable(),
+      }),
+      messageBytesLimit: configuration.messageBytesLimit,
+    })
     cleanupTasks.push(() => telemetryBatch.stop())
-    const telemetrySubscription = telemetryObservable.subscribe((event) =>
-      telemetryBatch.add(event, isTelemetryReplicationAllowed(configuration))
-    )
+    const telemetrySubscription = telemetryObservable.subscribe((event) => telemetryBatch.add(event))
     cleanupTasks.push(() => telemetrySubscription.unsubscribe())
   }
 

packages/core/src/index.ts

@@ -70,7 +70,7 @@ export {
   getEventBridge,
   bridgeSupports,
   BridgeCapability,
-  startBatchWithReplica,
+  createBatch,
   createFlushController,
 } from './transport'
 export * from './tools/display'

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

@@ -25,7 +25,7 @@ describe('httpRequest', () => {
     interceptor = interceptRequests()
     requests = interceptor.requests
     endpointBuilder = mockEndpointBuilder(ENDPOINT_URL)
-    request = createHttpRequest(endpointBuilder, BATCH_BYTES_LIMIT, noop)
+    request = createHttpRequest([endpointBuilder], BATCH_BYTES_LIMIT, noop)
   })
 
   describe('send', () => {
@@ -244,7 +244,7 @@ describe('httpRequest intake parameters', () => {
     interceptor = interceptRequests()
     requests = interceptor.requests
     endpointBuilder = createEndpointBuilder({ clientToken }, 'logs', [])
-    request = createHttpRequest(endpointBuilder, BATCH_BYTES_LIMIT, noop)
+    request = createHttpRequest([endpointBuilder], BATCH_BYTES_LIMIT, noop)
   })
 
   it('should have a unique request id', async () => {

packages/core/src/transport/httpRequest.ts

@@ -2,6 +2,7 @@ import type { EndpointBuilder } from '../domain/configuration'
 import type { Context } from '../tools/serialisation/context'
 import { monitor, monitorError } from '../tools/monitor'
 import type { RawError } from '../domain/error/error.types'
+import type { SendStrategy } from './sendWithRetryStrategy'
 import { newRetryState, sendWithRetryStrategy } from './sendWithRetryStrategy'
 
 /**
@@ -33,24 +34,28 @@ export interface RetryInfo {
 }
 
 export function createHttpRequest(
-  endpointBuilder: EndpointBuilder,
+  endpointBuilders: EndpointBuilder[],
   bytesLimit: number,
   reportError: (error: RawError) => void
 ) {
   const retryState = newRetryState()
-  const sendStrategyForRetry = (payload: Payload, onResponse: (r: HttpResponse) => void) =>
-    fetchKeepAliveStrategy(endpointBuilder, bytesLimit, payload, onResponse)
 
   return {
     send: (payload: Payload) => {
-      sendWithRetryStrategy(payload, retryState, sendStrategyForRetry, endpointBuilder.trackType, reportError)
+      for (const endpointBuilder of endpointBuilders) {
+        const sendStrategyForRetry: SendStrategy = (payload, onResponse) =>
+          fetchKeepAliveStrategy(endpointBuilder, bytesLimit, payload, onResponse)
+        sendWithRetryStrategy(payload, retryState, sendStrategyForRetry, endpointBuilder.trackType, reportError)
+      }
     },
     /**
      * Since fetch keepalive behaves like regular fetch on Firefox,
      * keep using sendBeaconStrategy on exit
      */
     sendOnExit: (payload: Payload) => {
-      sendBeaconStrategy(endpointBuilder, bytesLimit, payload)
+      for (const endpointBuilder of endpointBuilders) {
+        sendBeaconStrategy(endpointBuilder, bytesLimit, payload)
+      }
     },
   }
 }

packages/core/src/transport/index.ts

@@ -2,6 +2,6 @@ export type { HttpRequest, Payload, RetryInfo } from './httpRequest'
 export { createHttpRequest } from './httpRequest'
 export type { BrowserWindowWithEventBridge, DatadogEventBridge } from './eventBridge'
 export { canUseEventBridge, bridgeSupports, getEventBridge, BridgeCapability } from './eventBridge'
-export { startBatchWithReplica } from './startBatchWithReplica'
+export { createBatch } from './batch'
 export type { FlushController, FlushEvent, FlushReason } from './flushController'
 export { createFlushController } from './flushController'

packages/core/src/transport/sendWithRetryStrategy.ts

@@ -32,7 +32,7 @@ export interface RetryState {
   queueFullReported: boolean
 }
 
-type SendStrategy = (payload: Payload, onResponse: (r: HttpResponse) => void) => void
+export type SendStrategy = (payload: Payload, onResponse: (r: HttpResponse) => void) => void
 
 export function sendWithRetryStrategy(
   payload: Payload,