fetchObservable.ts

import type { ClocksState } from '@datadog/js-core/time'
import { clocksNow } from '@datadog/js-core/time'
import { normalizeUrl, globalObject } from '@datadog/js-core/util'
import type { GlobalObject } from '@datadog/js-core/util'
import type { InstrumentedMethodCall } from '../tools/instrumentMethod'
import { instrumentMethod } from '../tools/instrumentMethod'
import { monitorError } from '../tools/monitor'
import { Observable } from '../tools/observable'
import { readBytesFromStream } from '../tools/readBytesFromStream'
import { tryToClone } from '../tools/utils/responseUtils'

interface FetchContextBase {
  method: string
  startClocks: ClocksState
  input: unknown
  init?: RequestInit
  url: string
  handlingStack?: string
  isAbortedOnStart: boolean
}

export interface FetchStartContext extends FetchContextBase {
  state: 'start'
}

export interface FetchResolveContext extends FetchContextBase {
  state: 'resolve'
  status: number
  response?: Response
  responseBody?: string
  responseType?: string
  isAborted: boolean
  error?: Error
}

export type FetchContext = FetchStartContext | FetchResolveContext

type ResponseBodyActionGetter = (context: FetchResolveContext) => ResponseBodyAction

/**
 * Action to take with the response body of a fetch request.
 * Values are ordered by priority: higher values take precedence when multiple actions are requested.
 */
export const enum ResponseBodyAction {
  IGNORE = 0,
  COLLECT = 1,
}

let fetchObservable: Observable<FetchContext> | undefined
const responseBodyActionGetters: ResponseBodyActionGetter[] = []

export function initFetchObservable({ responseBodyAction }: { responseBodyAction?: ResponseBodyActionGetter } = {}) {
  if (responseBodyAction) {
    responseBodyActionGetters.push(responseBodyAction)
  }
  if (!fetchObservable) {
    fetchObservable = createFetchObservable()
  }
  return fetchObservable
}

export function resetFetchObservable() {
  fetchObservable = undefined
  responseBodyActionGetters.length = 0
}

function createFetchObservable() {
  return new Observable<FetchContext>((observable) => {
    // eslint-disable-next-line local-rules/disallow-zone-js-patched-values
    if (!globalObject.fetch) {
      return
    }

    const { stop } = instrumentMethod(globalObject, 'fetch', (call) => beforeSend(call, observable), {
      computeHandlingStack: true,
    })

    return stop
  })
}

function beforeSend(
  { parameters, onPostCall, handlingStack }: InstrumentedMethodCall<GlobalObject, 'fetch'>,
  observable: Observable<FetchContext>
) {
  const [input, init] = parameters
  let methodFromParams = init?.method

  if (methodFromParams === undefined && input instanceof Request) {
    methodFromParams = input.method
  }

  const method = methodFromParams !== undefined ? String(methodFromParams).toUpperCase() : 'GET'
  const url = input instanceof Request ? input.url : normalizeUrl(String(input))
  const startClocks = clocksNow()

  const context: FetchStartContext = {
    state: 'start',
    init,
    input,
    method,
    startClocks,
    url,
    handlingStack,
    isAbortedOnStart: (input instanceof Request && input.signal?.aborted) || init?.signal?.aborted || false,
  }

  observable.notify(context)

  // Those properties can be changed by observable subscribers
  parameters[0] = context.input as RequestInfo | URL
  parameters[1] = context.init

  onPostCall((responsePromise) => {
    afterSend(observable, responsePromise, context).catch(monitorError)
  })
}

async function afterSend(
  observable: Observable<FetchContext>,
  responsePromise: Promise<Response>,
  startContext: FetchStartContext
) {
  const context = startContext as unknown as FetchResolveContext

  let response: Response

  try {
    response = await responsePromise
  } catch (error) {
    observable.notify({
      ...context,
      state: 'resolve',
      status: 0,
      isAborted:
        context.init?.signal?.aborted || (error instanceof DOMException && error.code === DOMException.ABORT_ERR),
      error: error as Error,
    })
    return
  }

  context.response = response
  context.status = response.status
  context.responseType = response.type
  context.isAborted = false

  const responseBodyCondition: ResponseBodyAction = responseBodyActionGetters.reduce<number>(
    (action, getter) => Math.max(action, getter(context)),
    ResponseBodyAction.IGNORE
  )

  if (responseBodyCondition === ResponseBodyAction.COLLECT) {
    const clonedResponse = tryToClone(response)
    if (clonedResponse?.body) {
      try {
        const bytes = await readBytesFromStream(clonedResponse.body)
        context.responseBody = new TextDecoder().decode(bytes)
      } catch {
        // Ignore errors when reading the response body (e.g., stream aborted, network errors)
        // This is not critical and should not be reported as an SDK error
      }
    }
  }

  observable.notify({ ...context, state: 'resolve' })
}