fetch.ts

import type { StandardSchemaV1 } from '@standard-schema/spec';
import merge from 'lodash.merge';
import { BreakerError, BreakerState, Circuit, Module, Ratelimit, RatelimitError, Retry, RetryMode, SlidingCountBreaker, TimeoutError } from 'mollitia';
import { CONTEXT, ContextHeader, ContextKey, ContextKeyHttp, ContextKeyHttpRequest, ContextKeyHttpResponse } from '@smooai/logger/Logger';
import { handleSchemaValidation, HumanReadableSchemaError } from '@smooai/utils/validation/standardSchema';
import { contextLogger } from './logger';
import { redactBody, redactHeaders, redactQueryString, redactUrl } from './redact';

export { RatelimitError, TimeoutError } from 'mollitia';
export * from 'mollitia';

declare global {
    interface Window {
        fetch: typeof fetch;
    }
    interface Self {
        fetch: typeof fetch;
    }
    // Add declarations for window and self globals
    const window: Window & typeof globalThis;
    const self: Self & typeof globalThis;
}

/**
 * Determine the appropriate fetch implementation based on the environment.
 * Uses global.fetch in Node.js environments and window.fetch in browser environments.
 */
const globalFetch = (): typeof global.fetch => {
    // Browser environment
    if (typeof window !== 'undefined' && window.fetch) {
        return window.fetch.bind(window);
    }

    // Web Worker environment
    if (typeof self !== 'undefined' && self.fetch) {
        return self.fetch.bind(self);
    }

    // Node.js environment
    if (typeof global !== 'undefined') {
        // Modern Node.js (v18+) with built-in fetch
        if (global.fetch) {
            return global.fetch.bind(global);
        }

        // For older Node.js versions, suggest installing node-fetch
        throw new Error(
            'No fetch implementation found in Node.js environment. ' + 'For Node.js versions < 18, please install node-fetch: npm install node-fetch',
        );
    }

    throw new Error('No fetch implementation found. ' + 'Please ensure fetch is available in your environment.');
};

/**
 * Cache of undici `Agent` dispatchers keyed by connect-timeout ms, so we build
 * one Agent per distinct timeout instead of per request.
 */
const connectTimeoutDispatchers = new Map<number, unknown>();

/**
 * Resolve an undici `Agent` dispatcher that bounds the connect phase to
 * `connectTimeoutMs`.
 *
 * Node only: browsers and Web Workers expose no connect-timeout knob at all, so
 * this returns `undefined` there and no `dispatcher` is passed — behavior stays
 * byte-identical to today, as documented on `connectTimeoutMs`.
 *
 * `undici` is an OPTIONAL peer dependency, imported lazily so it is never loaded
 * unless a connect timeout is actually requested. When it is requested and undici
 * is absent, this throws rather than quietly running without the bound the caller
 * asked for — a timeout that silently isn't applied is the failure mode this
 * whole feature exists to prevent.
 */
async function getConnectTimeoutDispatcher(connectTimeoutMs: number): Promise<unknown | undefined> {
    // Only Node's fetch (undici) honors a `dispatcher`. Bail everywhere else.
    if (typeof window !== 'undefined' || typeof process === 'undefined' || !process.versions?.node) {
        return undefined;
    }
    const cached = connectTimeoutDispatchers.get(connectTimeoutMs);
    if (cached) return cached;
    let Agent: new (options: { connect: { timeout: number } }) => unknown;
    try {
        ({ Agent } = await import('undici'));
    } catch (error) {
        throw new Error(
            'connectTimeoutMs requires the optional peer dependency `undici` (Node only). Install it alongside @smooai/fetch, or drop connectTimeoutMs to use the whole-request timeout alone.',
            { cause: error },
        );
    }
    const dispatcher = new Agent({ connect: { timeout: connectTimeoutMs } });
    connectTimeoutDispatchers.set(connectTimeoutMs, dispatcher);
    return dispatcher;
}

/**
 * Interface for browser-compatible logging functionality.
 * Provides methods for different log levels with context support.
 */
export interface LoggerInterface {
    /**
     * Log a debug message with optional context.
     * @param message - The message to log
     * @param args - Additional arguments to include in the log
     */
    debug(message: string, ...args: any[]): void;

    /**
     * Log an info message with optional context.
     * @param message - The message to log
     * @param args - Additional arguments to include in the log
     */
    info(message: string, ...args: any[]): void;

    /**
     * Log a warning message with optional context.
     * @param message - The message to log
     * @param args - Additional arguments to include in the log
     */
    warn(message: string, ...args: any[]): void;

    /**
     * Log an error message with the error object and optional context.
     * @param error - The error object to log
     * @param message - The error message
     * @param args - Additional arguments to include in the log
     */
    error(error: Error | unknown, message: string, ...args: any[]): void;
}

const contextLoggerToUse = contextLogger();

/**
 * Extended Response type that includes parsed body data and metadata.
 * @template T - The type of the response body data
 */
export type ResponseWithBody<T = any> = Response & {
    /** The parsed response body data */
    data?: T;
    /** Whether the response body is JSON */
    isJson: boolean;
    /** The raw response body as a string */
    dataString: string;
};

type Headers = globalThis.Headers;
export type HeadersInit = string[][] | Record<string, string | ReadonlyArray<string>> | Headers;

type RequestInit = globalThis.RequestInit;
export type RequestInfo = string | URL | Request;

type Request = globalThis.Request;
type Response = globalThis.Response;

export type { Headers, RequestInit, Request, Response };

type ResponseType<Schema extends StandardSchemaV1 = never> = Schema extends StandardSchemaV1 ? StandardSchemaV1.InferOutput<Schema> : any;

/**
 * Defaults set below:
 *
 * - DEFAULTS
 * - DEFAULT_RETRY_OPTIONS
 * - DEFAULT_RATE_LIMIT_RETRY_OPTIONS
 */

/**
 * Error thrown when an HTTP request fails with a non-2xx status code.
 * Includes the response data and attempts to extract error information.
 * @template T - The type of the response body data
 */
export class HTTPResponseError<T = any> extends Error {
    /** The response object containing the error details */
    public response: ResponseWithBody<T>;
    constructor(response: ResponseWithBody<T>, msg?: string) {
        let errorStr = '';
        let errIsSet = false;
        if (response.isJson && response.data) {
            const data = response.data as any;
            if (data.error) {
                if (!Array.isArray(data.error)) {
                    if (data.error.type) {
                        errorStr += `(${data.error.type}): `;
                        errIsSet = true;
                    }
                    if (data.error.code) {
                        errorStr += `(${data.error.code}): `;
                        errIsSet = true;
                    }
                    if (data.error.message) {
                        errorStr += `${data.error.message}`;
                        errIsSet = true;
                    }
                    if (typeof data.error === 'string') {
                        errorStr += `${data.error}`;
                        errIsSet = true;
                    }
                }
            }
            if (data.errorMessages) {
                if (Array.isArray(data.errorMessages)) {
                    errorStr += `${data.errorMessages.join('; ')}`;
                    errIsSet = true;
                }
            }
        }
        if (!errIsSet) {
            errorStr = response.dataString || 'Unknown error';
        }
        super(`${msg ? `${msg}; ` : ''}${errorStr}; HTTP Error Response: ${response.status} ${response.statusText}`);
        this.response = response;
    }
}

export function isRetryable(status: number) {
    return status === 429 || status >= 500;
}

export class RetryError<T = any> extends HTTPResponseError<T> {
    constructor(response: ResponseWithBody<T>) {
        super(response, 'Retry Error: Ran out of retry attempts.');
        this.response = response;
    }
}

export type ErrorCallback = (err: any) => boolean;
export type RetryCallback = (err: any, attempt: number) => boolean | number;

/**
 * Configuration options for retry behavior.
 */
export interface RetryOptions {
    /** Number of retry attempts */
    attempts: number;
    /** Initial delay between retries in milliseconds */
    initialIntervalMs: number;
    /** Retry mode (e.g., exponential backoff, jitter) */
    mode?: RetryMode;
    /** Factor to multiply the interval by for each retry */
    factor?: number;
    /** Whether to attempt the first retry immediately */
    fastFirst?: boolean;
    /** Maximum delay between retries in milliseconds */
    maxInterval?: number;
    /** Amount of random jitter to add to retry delays */
    jitterAdjustment?: number;
    /** Callback to determine if and when to retry */
    onRejection?: RetryCallback;
    /**
     * Retry non-idempotent methods (POST, PATCH, ...) too. Off by default.
     *
     * Only idempotent methods (RFC 9110 §9.2.2: GET, HEAD, OPTIONS, TRACE, PUT,
     * DELETE) are retried unless this is set or the request carries an
     * `Idempotency-Key` header. A POST that timed out or got a 429/5xx may
     * already have done its work server-side — re-sending it bills a second
     * image, sends a second message, charges a card twice. Set this only when
     * the endpoint tolerates a duplicate, and prefer an `Idempotency-Key`.
     *
     * This overrides `onRejection`: an ineligible request makes exactly one
     * attempt and `onRejection` is never consulted.
     */
    allowNonIdempotent?: boolean;
}

/** Header that, when set to a non-empty value, makes a non-idempotent request retry-eligible. */
export const IDEMPOTENCY_KEY_HEADER = 'Idempotency-Key';

const IDEMPOTENT_METHODS = new Set(['GET', 'HEAD', 'OPTIONS', 'TRACE', 'PUT', 'DELETE']);

/**
 * Whether an HTTP method is idempotent per RFC 9110 §9.2.2 — i.e. sending it
 * twice has the same effect on the server as sending it once. Case-insensitive;
 * anything unrecognised is treated as non-idempotent.
 */
export function isIdempotentMethod(method: string | undefined): boolean {
    return IDEMPOTENT_METHODS.has((method ?? 'GET').toUpperCase());
}

function hasIdempotencyKey(headers: RequestInit['headers']): boolean {
    if (!headers) return false;
    let value: string | null = null;
    if (headers instanceof Headers) {
        value = headers.get(IDEMPOTENCY_KEY_HEADER);
    } else {
        const entries = Array.isArray(headers) ? headers : Object.entries(headers);
        const found = entries.find(([key]) => key.toLowerCase() === IDEMPOTENCY_KEY_HEADER.toLowerCase());
        value = found ? String(found[1]) : null;
    }
    return !!value && value.trim().length > 0;
}

/**
 * Whether a failed attempt of this request may be re-sent: its method is
 * idempotent, the caller set `retry.allowNonIdempotent`, or it carries a
 * non-empty `Idempotency-Key` header (the server then dedupes for us).
 */
export function isRetryEligible(init: Pick<RequestInit, 'method' | 'headers'>, retry?: Partial<RetryOptions>): boolean {
    return isIdempotentMethod(init.method) || !!retry?.allowNonIdempotent || hasIdempotencyKey(init.headers);
}

export const DEFAULT_RETRY_OPTIONS: RetryOptions = {
    attempts: 2,
    initialIntervalMs: 500,
    mode: RetryMode.JITTER,
    factor: 2,
    jitterAdjustment: 0.5,
    onRejection: (error) => {
        if (error instanceof HTTPResponseError) {
            if (isRetryable(error.response.status) && error.response.headers?.has('Retry-After')) {
                return parseInt(error.response.headers.get('Retry-After')!) * 1000;
            }
            return isRetryable(error.response.status);
        } else if (error instanceof RatelimitError) {
            return error.remainingTimeInRatelimit;
        } else if (error instanceof TimeoutError) {
            return true;
        } else if (error instanceof HumanReadableSchemaError) {
            return false;
        }

        return true;
    },
};

export const DEFAULT_RATE_LIMIT_RETRY_OPTIONS: RetryOptions = {
    attempts: 1,
    initialIntervalMs: 500,
    onRejection: (error) => {
        if (error instanceof RatelimitError) {
            return error.remainingTimeInRatelimit + 50;
        }

        return false;
    },
};

/**
 * Hook that runs before the request is made, allowing modification of the request.
 *
 * May return synchronously or async — the fetch pipeline awaits the result, so
 * callers can perform asynchronous work (e.g., minting a fresh auth token).
 */
export type PreRequestHook = (url: string, init: RequestInit) => [string, RequestInit] | void | Promise<[string, RequestInit] | void>;

/**
 * Provider that returns an auth token to inject as the `Authorization` header.
 * May be sync or async. Invoked before every request.
 */
export type AuthTokenProvider = () => string | Promise<string>;

/**
 * Hook that runs after a successful response, allowing modification of the response
 */
export type PostResponseSuccessHook<T = any> = (url: string, init: Readonly<RequestInit>, response: ResponseWithBody<T>) => ResponseWithBody<T> | void;

/**
 * Hook that runs after a failed response, allowing modification of the error
 */
export type PostResponseErrorHook<T = any> = (url: string, init: Readonly<RequestInit>, error: Error, response?: ResponseWithBody<T>) => Error | void;

/**
 * Collection of lifecycle hooks for request/response handling
 */
export interface LifecycleHooks<T = any> {
    /** Hook that runs before the request is made */
    preRequest?: PreRequestHook;
    /** Hook that runs after a successful response */
    postResponseSuccess?: PostResponseSuccessHook<T>;
    /** Hook that runs after a failed response */
    postResponseError?: PostResponseErrorHook<T>;
}

/**
 * Configuration options for HTTP requests.
 * @template Schema - The schema type for response validation. Must be a StandardSchemaV1 compatible schema (e.g., Zod schema)
 */
export interface RequestOptions<Schema extends StandardSchemaV1 = never> {
    /** Custom logger for request logging */
    logger?: LoggerInterface;
    /**
     * Per-attempt timeout. When it fires the attempt is ABORTED (its
     * connection is closed, so the server sees the cancel) before any retry.
     */
    timeout?: {
        /** Timeout duration in milliseconds */
        timeoutMs: number;
    };
    /**
     * Connect timeout in milliseconds. Bounds only the connection-establishment
     * phase (Node only, via an undici `Agent` dispatcher). When set, a connect
     * that never completes (e.g. a black-holed SYN to a dead pod IP still
     * lingering in a ClusterIP's iptables) fails in ~this window and the
     * configured retry can land on a live endpoint — instead of stalling until
     * the whole-request {@link timeout}. Slow-but-alive handlers are unaffected.
     * Leaving it unset preserves the previous no-connect-timeout behavior.
     * Ignored in browser/worker environments (no connect-timeout knob there).
     */
    connectTimeoutMs?: number;
    /**
     * Retry configuration, deep-merged over {@link DEFAULT_RETRY_OPTIONS}, so
     * `retry: { allowNonIdempotent: true }` keeps every other default.
     * POST/PATCH are not retried unless `allowNonIdempotent` is set or the
     * request carries an `Idempotency-Key` — see {@link RetryOptions.allowNonIdempotent}.
     */
    retry?: Partial<RetryOptions>;
    /** Schema for response validation. Must be a StandardSchemaV1 compatible schema (e.g., Zod schema) */
    schema?: Schema;
    /** Lifecycle hooks for request/response handling */
    hooks?: LifecycleHooks<ResponseType<Schema>>;
}

const DEFAULTS: RequestOptions<never> = {
    logger: contextLoggerToUse,
    retry: DEFAULT_RETRY_OPTIONS,
    timeout: {
        timeoutMs: 10000,
    },
};

// RateLimit, CircuitBreaker below is using https://genesys.github.io/mollitia/overview/introduction
/**
 * Configuration options for fetch container features like rate limiting and circuit breaking.
 */
export interface FetchContainerOptions {
    /** Rate limiting configuration */
    rateLimit?: {
        /** Maximum number of requests allowed in the period */
        limitForPeriod: number;
        /** Duration of the rate limit period in milliseconds */
        limitPeriodMs: number;
        /** Retry configuration for rate limit handling */
        retry?: RetryOptions;
    };
    /** Circuit breaker configuration */
    circuitBreaker?: {
        /** Current state of the circuit breaker */
        state?: BreakerState;
        /** Failure rate threshold as a percentage */
        failureRateThreshold?: number;
        /** Slow call rate threshold as a percentage */
        slowCallRateThreshold?: number;
        /** Duration threshold for slow calls in milliseconds */
        slowCallDurationThresholdMs?: number;
        /** Number of calls allowed in half-open state */
        permittedNumberOfCallsInHalfOpenState?: number;
        /** Maximum delay in half-open state in milliseconds */
        halfOpenStateMaxDelayMs?: number;
        /** Size of the sliding window for failure rate calculation */
        slidingWindowSize?: number;
        /** Minimum number of calls for failure rate calculation */
        minimumNumberOfCalls?: number;
        /** Time to stay in open state in milliseconds */
        openStateDelayMs?: number;
        /** Callback to determine if an error should trigger circuit breaker */
        onError?: ErrorCallback;
    };
}

function mergeHeadersWithContext(headersInit?: HeadersInit): HeadersInit {
    let newHeadersInit: HeadersInit = {};
    if (
        CONTEXT[ContextKey.Http] &&
        CONTEXT[ContextKey.Http][ContextKeyHttp.Request] &&
        CONTEXT[ContextKey.Http][ContextKeyHttp.Request][ContextKeyHttpRequest.UserAgent]
    ) {
        newHeadersInit[ContextHeader.UserAgent] = CONTEXT[ContextKey.Http][ContextKeyHttp.Request][ContextKeyHttpRequest.UserAgent];
    }
    if (CONTEXT[ContextKey.CorrelationId]) {
        newHeadersInit[ContextHeader.CorrelationId] = CONTEXT[ContextKey.CorrelationId];
    }
    let initialHeaders = headersInit;
    if ((initialHeaders && initialHeaders instanceof Headers) || initialHeaders instanceof Map) {
        initialHeaders = Array.from(initialHeaders.entries()).reduce((result, entry) => {
            result[entry[0]] = entry[1];
            return result;
        }, {} as any);
    }
    newHeadersInit = merge({}, initialHeaders, newHeadersInit);
    return newHeadersInit;
}

// Both helpers below feed logger payloads only — never the outgoing request —
// so they redact credentials in place. `@smooai/logger` does no redaction of its
// own, so if it isn't scrubbed here it reaches CloudWatch in plaintext.
function getHeadersObject(headers: Headers | Record<string, string> | undefined): Record<string, string> {
    if (!headers) return {};
    if (headers instanceof Headers) {
        const result: Record<string, string> = {};
        headers.forEach((value, key) => {
            result[key] = value;
        });
        return redactHeaders(result);
    }
    return redactHeaders(headers);
}

function getRequestBody(body: any): string | undefined {
    return redactBody(body);
}

function prepareDefaultInit(init?: RequestInit): RequestInit {
    const requestInit: RequestInit = {
        headers: {
            ...mergeHeadersWithContext(init?.headers),
        },
        method: init && init.method ? init.method : 'GET',
    };
    if (init && init.headers) {
        delete init.headers;
    }
    return merge({}, init, requestInit);
}

function prepareDefaultOptions<Schema extends StandardSchemaV1 = never>(options?: RequestOptions<Schema>): RequestOptions<Schema> {
    return merge({}, DEFAULTS, options);
}

// Mollitia modules need a name unique within the process; these are labels that
// show up in logs, nothing more. This used to be `${prefix}-${faker.color.human()}-
// ${faker.animal.type()}`, which made @faker-js/faker — a test-data generator —
// a RUNTIME dependency of a published HTTP client, for two cosmetic words.
let moduleSequence = 0;
function generateRandomName(prefix: string): string {
    return `${prefix}-${++moduleSequence}`;
}

function prepareCircuitModules<Schema extends StandardSchemaV1 = never>(options: RequestOptions<Schema>, init: RequestInit): Module[] {
    const modules: Module[] = [];
    const logger = options.logger || contextLoggerToUse;
    const retryEligible = isRetryEligible(init, options.retry);
    const onRejection = options.retry?.onRejection;

    // No Retry module at all for an ineligible request: one attempt, and the
    // underlying error surfaces as-is rather than as a RetryError.
    //
    // There is deliberately no mollitia `Timeout` module here either. It only
    // RACED the attempt — the losing fetch kept running, so a timed-out POST
    // stayed in flight server-side while the retry sent it again. The timeout
    // now lives in `doGlobalFetch`, where it can abort the request itself.
    if (options.retry && retryEligible) {
        modules.push(
            new Retry({
                name: generateRandomName('smooai-fetch-retry'),
                logger: logger,
                attempts: options.retry.attempts,
                interval: options.retry.initialIntervalMs,
                mode: options.retry.mode,
                factor: options.retry.factor,
                jitterAdjustment: options.retry.jitterAdjustment,
                // A request the CALLER aborted is finished, not failed — never re-send it.
                onRejection: (error, attempt) => (init.signal?.aborted ? false : onRejection ? onRejection(error, attempt) : true),
            }),
        );
    }

    return modules;
}

function prepareFetchContainerModules(options: RequestOptions, containerOptions?: FetchContainerOptions): Module[] {
    const modules: Module[] = [];
    const logger = options.logger || contextLoggerToUse;

    if (containerOptions) {
        if (containerOptions.rateLimit) {
            const retryOptions = containerOptions.rateLimit.retry || DEFAULT_RATE_LIMIT_RETRY_OPTIONS;
            modules.push(
                new Retry({
                    name: generateRandomName('smooai-fetch-rate-limit-retry'),
                    logger: logger,
                    attempts: retryOptions.attempts,
                    interval: retryOptions.initialIntervalMs,
                    mode: retryOptions.mode,
                    factor: retryOptions.factor,
                    jitterAdjustment: retryOptions.jitterAdjustment,
                    onRejection: retryOptions.onRejection,
                }),
            );

            modules.push(
                new Ratelimit({
                    name: generateRandomName('smooai-fetch-rate-limit'),
                    logger: logger,
                    limitPeriod: containerOptions.rateLimit.limitPeriodMs,
                    limitForPeriod: containerOptions.rateLimit.limitForPeriod,
                }),
            );
        }

        if (containerOptions.circuitBreaker) {
            modules.push(
                new SlidingCountBreaker({
                    name: generateRandomName('smooai-fetch-circuit-breaker'),
                    logger: logger,
                    state: containerOptions.circuitBreaker.state,
                    failureRateThreshold: containerOptions.circuitBreaker.failureRateThreshold,
                    slowCallRateThreshold: containerOptions.circuitBreaker.slowCallRateThreshold,
                    slowCallDurationThreshold: containerOptions.circuitBreaker.slowCallDurationThresholdMs,
                    permittedNumberOfCallsInHalfOpenState: containerOptions.circuitBreaker.permittedNumberOfCallsInHalfOpenState,
                    halfOpenStateMaxDelay: containerOptions.circuitBreaker.halfOpenStateMaxDelayMs,
                    slidingWindowSize: containerOptions.circuitBreaker.slidingWindowSize,
                    minimumNumberOfCalls: containerOptions.circuitBreaker.minimumNumberOfCalls,
                    openStateDelay: containerOptions.circuitBreaker.openStateDelayMs,
                    onError: containerOptions.circuitBreaker.onError,
                }),
            );
        }
    }

    return modules;
}

/**
 * Cached, best-effort handle on `@opentelemetry/api`.
 *
 * The package is an OPTIONAL peer dependency: an OSS HTTP client must not force
 * OpenTelemetry on anyone. When it is absent the import rejects, this resolves
 * `null`, and injection becomes a no-op — no crash, no behaviour change.
 */
let otelApi: Promise<typeof import('@opentelemetry/api') | null> | undefined;

/**
 * Inject W3C trace context (`traceparent`/`tracestate`) into an outbound request.
 *
 * # Why this exists
 *
 * api-prime already EXTRACTS `traceparent` on ingress, but nothing on the
 * platform ever injected it on egress — so every service-to-service call began a
 * brand new root trace. Measured on 2026-08-14 over a three-hour window: 34,961
 * traces touched exactly one service, and 4 touched two.
 *
 * This is the client, which is the correct place for propagation: services are
 * already required to use `@smooai/fetch` over raw HTTP, so wiring it once here
 * covers the fleet. Mirrors `rust/fetch/src/client.rs::inject_trace_context`.
 *
 * # Three guards, each for a reason
 *
 * 1. **Optional dependency.** No `@opentelemetry/api` installed → no-op.
 * 2. **Valid span contexts only.** No registered SDK / no active span yields
 *    INVALID_SPAN_CONTEXT (all-zero ids). Injecting that writes a malformed
 *    `traceparent` a downstream service will either reject or, worse, adopt —
 *    poisoning its trace. The sibling logger shipped exactly this bug, where
 *    all-zero ids overwrote a real correlation id.
 * 3. **Caller wins.** An explicitly-set `traceparent` is left alone. A client
 *    that silently rewrites an intentional header is worse than one that does
 *    nothing.
 */
async function injectTraceContext(init: RequestInit): Promise<void> {
    otelApi ??= import('@opentelemetry/api').catch(() => null);
    const otel = await otelApi;
    if (!otel) return;

    const activeContext = otel.context.active();
    const spanContext = otel.trace.getSpanContext(activeContext);
    if (!spanContext || !otel.isSpanContextValid(spanContext)) return;

    // Copy rather than mutate: the caller's `init` is reused across retries, and a
    // header we wrote ourselves must never be mistaken for the caller's on the next
    // attempt — that would pin every retry to the traceparent of the first one.
    const headers = new Headers(init.headers as HeadersInit | undefined);
    if (headers.has('traceparent')) return;

    const carrier: Record<string, string> = {};
    otel.propagation.inject(activeContext, carrier);
    if (Object.keys(carrier).length === 0) return;
    for (const [key, value] of Object.entries(carrier)) {
        headers.set(key, value);
    }
    init.headers = headers;
}

async function doGlobalFetch<Schema extends StandardSchemaV1 = never>(
    url: RequestInfo,
    init?: RequestInit,
    options?: RequestOptions<Schema>,
): Promise<ResponseWithBody<ResponseType<Schema>>> {
    // Defaults FIRST, caller's init LAST — `merge` lets later sources win, so
    // the old order (`merge({}, init, { redirect: 'follow' })`) silently
    // overwrote an explicit `redirect: 'manual' | 'error'` with 'follow'.
    //
    // Following redirects is the right default, but it must be a default and
    // not a mandate: a caller that resolved a hostname and checked it against
    // an SSRF allowlist has that guard defeated by a 302 to an internal
    // address, and RFC 8461 forbids following them at all when fetching an
    // MTA-STS policy.
    const useInit: RequestInit = merge({}, { redirect: 'follow' } as RequestInit, init);

    // Stringify JSON body if needed
    if ((useInit?.headers as Record<string, string>)?.['Content-Type'] === 'application/json' && typeof useInit.body === 'object') {
        useInit.body = JSON.stringify(useInit.body);
    }

    // Continue the caller's trace across the hop. This is the single-request site —
    // the innermost place that performs exactly one HTTP call — so every retry and
    // every re-issued request carries a CURRENT traceparent instead of a stale one.
    await injectTraceContext(useInit);

    // Bound the connect phase via an undici dispatcher (Node only). Only attach
    // when explicitly requested so the default path stays byte-identical.
    if (options?.connectTimeoutMs) {
        const dispatcher = await getConnectTimeoutDispatcher(options.connectTimeoutMs);
        if (dispatcher) {
            (useInit as Record<string, unknown>).dispatcher = dispatcher;
        }
    }

    // Per-attempt timeout that ABORTS the request. The caller's own signal still
    // works: either one aborting cancels the fetch.
    const timeoutMs = options?.timeout?.timeoutMs;
    const timeoutController = new AbortController();
    let timer: ReturnType<typeof setTimeout> | undefined;
    if (timeoutMs !== undefined) {
        timer = setTimeout(() => timeoutController.abort(new TimeoutError()), timeoutMs);
        useInit.signal = init?.signal ? anySignal([init.signal, timeoutController.signal]) : timeoutController.signal;
    }

    try {
        const attempt = sendAndRead<Schema>(url, useInit, options);
        if (timeoutMs === undefined) return await attempt;
        // Race as well as abort: a fetch implementation that ignores `signal` (a
        // polyfill, a test double) must still time out on schedule. With a
        // compliant fetch the abort has already torn the request down.
        return await Promise.race([
            attempt,
            new Promise<never>((_, reject) => {
                timeoutController.signal.addEventListener('abort', () => reject(timeoutController.signal.reason), { once: true });
            }),
        ]);
    } catch (error) {
        // undici rejects with the abort reason, a browser with an AbortError —
        // normalise both to the TimeoutError the retry policy and callers expect.
        if (timeoutController.signal.aborted && !init?.signal?.aborted) {
            throw timeoutController.signal.reason instanceof TimeoutError ? timeoutController.signal.reason : new TimeoutError();
        }
        throw error;
    } finally {
        clearTimeout(timer);
    }
}

/** `AbortSignal.any` where available (Node 20+, modern browsers), else a manual fan-in. */
function anySignal(signals: AbortSignal[]): AbortSignal {
    const any = (AbortSignal as unknown as { any?: (signals: AbortSignal[]) => AbortSignal }).any;
    if (any) return any(signals);
    const controller = new AbortController();
    for (const signal of signals) {
        if (signal.aborted) {
            controller.abort(signal.reason);
            break;
        }
        signal.addEventListener('abort', () => controller.abort(signal.reason), { once: true });
    }
    return controller.signal;
}

async function sendAndRead<Schema extends StandardSchemaV1 = never>(
    url: RequestInfo,
    useInit: RequestInit,
    options?: RequestOptions<Schema>,
): Promise<ResponseWithBody<ResponseType<Schema>>> {
    const response = await globalFetch()(url, useInit);
    let isJson = false;
    let data: ResponseType<Schema> | undefined;
    let dataString: string = '';
    let read = false;

    const responseClone = response.clone();

    if (responseClone.headers?.has('Content-Type') && responseClone.headers?.get('Content-Type')?.includes('application/json')) {
        dataString = await responseClone.text();
        read = true;
        try {
            const parsedData = JSON.parse(dataString);
            if (responseClone.ok && options?.schema) {
                data = (await handleSchemaValidation(options.schema, parsedData)) as Schema extends StandardSchemaV1
                    ? StandardSchemaV1.InferOutput<Schema>
                    : any;
            } else {
                data = parsedData as ResponseType<Schema>;
            }
            isJson = true;
        } catch (error) {
            if (error instanceof HumanReadableSchemaError) {
                throw error;
            }
            isJson = false;
        }
    } else {
        isJson = false;
    }

    const responseWithBody = response as any;
    responseWithBody.isJson = isJson;
    responseWithBody.dataString = dataString;
    responseWithBody.data = data;

    // A 3xx under `redirect: 'manual'` is the ANSWER, not a failure — the
    // caller asked to see the redirect rather than follow it. It is neither
    // `ok` nor `redirected` (nothing was followed), so without this it would
    // throw and the option would be useless even once it is honoured.
    const isManualRedirect = useInit.redirect === 'manual' && responseClone.status >= 300 && responseClone.status < 400;

    if (responseClone.ok || responseClone.redirected || isManualRedirect) {
        return responseWithBody;
    } else {
        if (!read) {
            responseWithBody.dataString = await responseClone.text();
        }
        throw new HTTPResponseError<ResponseType<Schema>>(responseWithBody);
    }
}

async function doFetch<Schema extends StandardSchemaV1 = never>(
    url: RequestInfo,
    init: RequestInit,
    options: RequestOptions<Schema>,
): Promise<ResponseWithBody<ResponseType<Schema>>> {
    const logger = options.logger || contextLoggerToUse;

    // Apply pre-request hook if present (supports async hooks)
    let modifiedInit = init;
    if (options.hooks?.preRequest) {
        const hookResult = await options.hooks.preRequest(url.toString(), init);
        if (hookResult) {
            modifiedInit = hookResult[1];
            url = hookResult[0];
        }
    }

    // Built AFTER the pre-request hook: retry eligibility depends on the final
    // method and headers, and a hook may be what adds the Idempotency-Key.
    const retryEligible = isRetryEligible(modifiedInit, options.retry);
    const circuit = new Circuit({
        name: 'node-fetch-circuit',
        func: doGlobalFetch,
        options: {
            modules: prepareCircuitModules(options, modifiedInit),
        },
    });

    const urlObj = new URL(url.toString());
    const safeUrl = redactUrl(url.toString());

    logger.debug(`Sending HTTP request "${modifiedInit.method} ${safeUrl}"`, {
        [ContextKey.Http]: {
            [ContextKeyHttp.Request]: {
                [ContextKeyHttpRequest.Method]: modifiedInit.method,
                [ContextKeyHttpRequest.Host]: urlObj.host,
                [ContextKeyHttpRequest.Path]: urlObj.pathname,
                [ContextKeyHttpRequest.QueryString]: redactQueryString(urlObj.search),
                [ContextKeyHttpRequest.Headers]: getHeadersObject(modifiedInit.headers as Headers),
                [ContextKeyHttpRequest.Body]: getRequestBody(modifiedInit.body),
            },
        },
    });

    let response: ResponseWithBody<ResponseType<Schema>>;
    try {
        response = await circuit.execute(url, modifiedInit, options);

        // Apply post-response success hook if present
        if (options.hooks?.postResponseSuccess) {
            const hookResult = options.hooks.postResponseSuccess(url.toString(), init, response);
            if (hookResult) {
                response = hookResult;
            }
        }

        return response;
    } catch (error) {
        // Apply post-response error hook if present
        if (options.hooks?.postResponseError) {
            const hookResult = options.hooks.postResponseError(
                url.toString(),
                init,
                error as Error,
                error instanceof HTTPResponseError ? error.response : undefined,
            );
            if (hookResult) {
                throw hookResult;
            }
        }

        if (error instanceof TimeoutError) {
            logger.error(error, `HTTP request "${modifiedInit.method} ${safeUrl}" timed out (${error.name}) after ${options.timeout!.timeoutMs} ms`, {
                [ContextKey.Http]: {
                    [ContextKeyHttp.Request]: {
                        [ContextKeyHttpRequest.Method]: modifiedInit.method,
                        [ContextKeyHttpRequest.Host]: urlObj.host,
                        [ContextKeyHttpRequest.Path]: urlObj.pathname,
                        [ContextKeyHttpRequest.QueryString]: redactQueryString(urlObj.search),
                        [ContextKeyHttpRequest.Headers]: getHeadersObject(modifiedInit.headers as Headers),
                        [ContextKeyHttpRequest.Body]: getRequestBody(modifiedInit.body),
                    },
                },
            });
        } else if (options.retry && retryEligible && error instanceof HTTPResponseError) {
            if (options.retry.onRejection && options.retry.onRejection(error, 1)) {
                logger.error(error, `HTTP request "${modifiedInit.method} ${safeUrl}" retries failed after ${options.retry.attempts} retries`, {
                    [ContextKey.Http]: {
                        [ContextKeyHttp.Request]: {
                            [ContextKeyHttpRequest.Method]: modifiedInit.method,
                            [ContextKeyHttpRequest.Host]: urlObj.host,
                            [ContextKeyHttpRequest.Path]: urlObj.pathname,
                            [ContextKeyHttpRequest.QueryString]: redactQueryString(urlObj.search),
                            [ContextKeyHttpRequest.Headers]: getHeadersObject(modifiedInit.headers as Headers),
                            [ContextKeyHttpRequest.Body]: getRequestBody(modifiedInit.body),
                        },
                        [ContextKeyHttp.Response]: {
                            [ContextKeyHttpResponse.StatusCode]: error.response.status,
                            [ContextKeyHttpResponse.Headers]: getHeadersObject(error.response.headers),
                            [ContextKeyHttpResponse.Body]: redactBody(error.response.dataString),
                        },
                    },
                });
                throw new RetryError<ResponseType<Schema>>(error.response);
            }
        }
        throw error;
    }
}

export type RequestInitWithOptions<Schema extends StandardSchemaV1 = never> = RequestInit & {
    options?: RequestOptions<Schema>;
};

// Overload signatures
export default async function fetch(url: RequestInfo, init?: RequestInitWithOptions<never>): Promise<ResponseWithBody<any>>;

export default async function fetch<Schema extends StandardSchemaV1>(
    url: RequestInfo,
    init?: RequestInitWithOptions<Schema>,
): Promise<ResponseWithBody<ResponseType<Schema>>>;

// Implementation
export default async function fetch<Schema extends StandardSchemaV1 = never>(
    url: RequestInfo,
    init?: RequestInitWithOptions<Schema>,
): Promise<ResponseWithBody<ResponseType<Schema>>> {
    const { options, ...requestInit } = init || {};
    return doFetch<Schema>(url, prepareDefaultInit(requestInit), prepareDefaultOptions(options));
}

/**
 * Example usage:
 *
 * const fetch = generateFetchWithOptions({
    containerOptions: {
        rateLimit: {
            name: 'finch-node-fetch-rate-limit',
            limitForPeriod: 2,
            limitPeriodMs: 62 * 1000,
            retry: DEFAULT_RATE_LIMIT_RETRY_OPTIONS,
        },
    },
    requestOptions: {
        logger,
        timeout: {
            name: 'finch-node-fetch-timeout',
            timeoutMs: 60 * 1000,
        },
        retry: DEFAULT_RETRY_OPTIONS,
    },
});
 * @param options
 * @returns
 */
function generateFetchWithOptions(options: { init?: RequestInit; requestOptions?: RequestOptions<never>; containerOptions?: FetchContainerOptions }): {
    (url: RequestInfo, init?: RequestInitWithOptions<never>): Promise<ResponseWithBody<any>>;
    <Schema extends StandardSchemaV1>(url: RequestInfo, init?: RequestInitWithOptions<Schema>): Promise<ResponseWithBody<ResponseType<Schema>>>;
} {
    const _init = prepareDefaultInit(options.init);
    const _requestOptions = prepareDefaultOptions(options.requestOptions);
    const circuit = new Circuit({
        name: 'node-fetch-container-circuit',
        options: {
            modules: prepareFetchContainerModules(_requestOptions, options.containerOptions),
        },
    });
    return <Schema extends StandardSchemaV1 = never>(
        url: RequestInfo,
        init?: RequestInitWithOptions<Schema>,
    ): Promise<ResponseWithBody<ResponseType<Schema>>> => {
        circuit.fn(doFetch);
        const { options: requestOptions, ...requestInit } = init || {};
        const __init = prepareDefaultInit(merge({}, _init, requestInit));
        const __requestOptions = prepareDefaultOptions(merge({}, _requestOptions, requestOptions));
        const logger = __requestOptions.logger || contextLoggerToUse;
        const urlObj = new URL(url.toString());
        const safeUrl = redactUrl(url.toString());
        const headers = getHeadersObject(__init.headers as Headers);
        const requestBody = getRequestBody(__init.body);

        try {
            return circuit.execute(url, prepareDefaultInit(__init), prepareDefaultOptions(__requestOptions));
        } catch (error) {
            if (error instanceof RatelimitError) {
                logger.error(
                    error,
                    `HTTP request "${__init.method} ${safeUrl}" rate limited (${error.name}) - more than ${
                        options.containerOptions!.rateLimit?.limitForPeriod
                    } in ${options.containerOptions!.rateLimit?.limitPeriodMs} ms - ${error.remainingTimeInRatelimit} ms left in rate limit`,
                    {
                        [ContextKey.Http]: {
                            [ContextKeyHttp.Request]: {
                                [ContextKeyHttpRequest.Method]: __init.method,
                                [ContextKeyHttpRequest.Host]: urlObj.host,
                                [ContextKeyHttpRequest.Path]: urlObj.pathname,
                                [ContextKeyHttpRequest.QueryString]: redactQueryString(urlObj.search),
                                [ContextKeyHttpRequest.Headers]: headers,
                                [ContextKeyHttpRequest.Body]: requestBody,
                            },
                        },
                    },
                );
            } else if (error instanceof BreakerError) {
                logger.error(error, `HTTP request "${__init.method} ${safeUrl}" circuit open (${error.name})`, {
                    [ContextKey.Http]: {
                        [ContextKeyHttp.Request]: {
                            [ContextKeyHttpRequest.Method]: __init.method,
                            [ContextKeyHttpRequest.Host]: urlObj.host,
                            [ContextKeyHttpRequest.Path]: urlObj.pathname,
                            [ContextKeyHttpRequest.QueryString]: redactQueryString(urlObj.search),
                            [ContextKeyHttpRequest.Headers]: headers,
                            [ContextKeyHttpRequest.Body]: requestBody,
                        },
                    },
                });
            }
            throw error;
        }
    };
}

/**
 * Builder class for creating configured fetch instances with retry, rate limiting, and circuit breaking.
 * Provides a fluent interface for configuring fetch options.
 * @template Schema - The schema type for response validation. Must be a StandardSchemaV1 compatible schema (e.g., Zod schema)
 */
export class FetchBuilder<Schema extends StandardSchemaV1 = never> {
    private _init?: RequestInit;
    private _requestOptions?: RequestOptions<Schema>;
    private _containerOptions?: FetchContainerOptions;

    constructor(schema?: Schema) {
        if (schema) this.withSchema(schema);
    }

    /**
     * Sets the initial request configuration.
     * @param init - The initial request configuration
     * @returns The builder instance for method chaining
     */
    withInit(init: RequestInit): FetchBuilder<Schema> {
        this._init = init;
        return this;
    }

    /**
     * Sets the request timeout.
     * @param timeoutMs - Timeout duration in milliseconds
     * @returns The builder instance for method chaining
     */
    withTimeout(timeoutMs: number): FetchBuilder<Schema> {
        this._requestOptions = {
            ...this._requestOptions,
            timeout: { timeoutMs },
        };
        return this;
    }

    /**
     * Sets the connect timeout (Node only). Bounds only the
     * connection-establishment phase so a black-holed connect fails fast and
     * retry can land on a live endpoint, instead of stalling until the
     * whole-request {@link withTimeout} elapses. Leaving it unset preserves the
     * previous no-connect-timeout behavior; ignored in browser/worker builds.
     * Mirrors the Rust `FetchBuilder::with_connect_timeout`.
     * @param connectTimeoutMs - Connect timeout duration in milliseconds
     * @returns The builder instance for method chaining
     */
    withConnectTimeout(connectTimeoutMs: number): FetchBuilder<Schema> {
        this._requestOptions = {
            ...this._requestOptions,
            connectTimeoutMs,
        };
        return this;
    }

    /**
     * Configures retry behavior for failed requests.
     * If not specified, uses DEFAULT_RETRY_OPTIONS; a partial is merged over it.
     * POST/PATCH are only retried with `allowNonIdempotent: true` or an
     * `Idempotency-Key` header.
     * @param options - Retry configuration options
     * @returns The builder instance for method chaining
     */
    withRetry(options: Partial<RetryOptions> = DEFAULT_RETRY_OPTIONS): FetchBuilder<Schema> {
        this._requestOptions = {
            ...this._requestOptions,
            retry: options,
        };
        return this;
    }

    /**
     * Configures rate limiting for requests.
     * If retryOptions is not specified, uses DEFAULT_RATE_LIMIT_RETRY_OPTIONS.
     * @param limitForPeriod - Maximum number of requests allowed in the period
     * @param limitPeriodMs - Duration of the rate limit period in milliseconds
     * @param retryOptions - Optional retry configuration for rate limit handling
     * @returns The builder instance for method chaining
     */
    withRateLimit(limitForPeriod: number, limitPeriodMs: number, retryOptions: RetryOptions = DEFAULT_RATE_LIMIT_RETRY_OPTIONS): FetchBuilder<Schema> {
        this._containerOptions = {
            ...this._containerOptions,
            rateLimit: {
                limitForPeriod,
                limitPeriodMs,
                retry: retryOptions,
            },
        };
        return this;
    }

    /**
     * Configures circuit breaker behavior.
     *
     * Note: This is not typically used, but can be useful for advanced use cases.
     *
     * @param options - Circuit breaker configuration options
     * @returns The builder instance for method chaining
     */
    withCircuitBreaker(options: FetchContainerOptions['circuitBreaker']): FetchBuilder<Schema> {
        this._containerOptions = {
            ...this._containerOptions,
            circuitBreaker: options,
        };
        return this;
    }

    /**
     * Sets container options directly.
     *
     * Note: This is not typically used, but can be useful for advanced use cases.
     *
     * @param options - Container configuration options
     * @returns The builder instance for method chaining
     */
    withContainerOptions(options: FetchContainerOptions): FetchBuilder<Schema> {
        this._containerOptions = {
            ...this._containerOptions,
            ...options,
        };
        return this;
    }

    /**
     * Sets a custom logger for request logging.
     * If not specified, uses the default contextLogger.
     * @param logger - The logger instance to use
     * @returns The builder instance for method chaining
     */
    withLogger(logger: LoggerInterface = contextLoggerToUse): FetchBuilder<Schema> {
        this._requestOptions = {
            ...this._requestOptions,
            logger,
        };
        return this;
    }

    /**
     * Sets a schema for response validation.
     * The schema must be StandardSchemaV1 compatible (e.g., a Zod schema).
     * The response body will be validated against this schema before being returned.
     *
     * @example
     * ```typescript
     * const schema = z.object({
     *   id: z.string(),
     *   name: z.string()
     * });
     *
     * const fetch = new FetchBuilder()
     *   .withSchema(schema)
     *   .build();
     * ```
     *
     * @param schema - The StandardSchemaV1 compatible schema to use for validation
     * @returns The builder instance for method chaining
     */
    private withSchema(schema: Schema): FetchBuilder<Schema> {
        this._requestOptions = {
            ...this._requestOptions,
            schema,
        };
        return this;
    }

    /**
     * Sets lifecycle hooks for request/response handling
     * @param hooks - The lifecycle hooks to use
     * @returns The builder instance for method chaining
     */
    withHooks(hooks: LifecycleHooks<ResponseType<Schema>>): FetchBuilder<Schema> {
        this._requestOptions = {
            ...this._requestOptions,
            hooks,
        };
        return this;
    }

    /**
     * Registers an async (or sync) auth token provider that is invoked before
     * every request and injects an `Authorization` header.
     *
     * Composes cleanly with `withHooks(...)` — if a `preRequest` hook is
     * already configured, the existing hook runs first and the
     * `Authorization` header is appended afterwards. The provider is awaited,
     * so callers can fetch / refresh / mint tokens lazily.
     *
     * @example
     * ```typescript
     * const fetch = new FetchBuilder()
     *     .withAuthTokenProvider(async () => await tokenStore.getFreshToken(), 'Bearer')
     *     .build();
     * ```
     *
     * @param provider - Sync or async function returning the bare token (no scheme prefix).
     * @param scheme - The auth scheme to prefix the token with. Defaults to "Bearer".
     * @returns The builder instance for method chaining.
     */
    withAuthTokenProvider(provider: AuthTokenProvider, scheme: string = 'Bearer'): FetchBuilder<Schema> {
        const existingHooks = this._requestOptions?.hooks;
        const existingPreRequest = existingHooks?.preRequest;

        const composedPreRequest: PreRequestHook = async (url, init) => {
            // Run any existing pre-request hook first so it can adjust the URL/init.
            let nextUrl = url;
            let nextInit = init;
            if (existingPreRequest) {
                const prior = await existingPreRequest(url, init);
                if (prior) {
                    nextUrl = prior[0];
                    nextInit = prior[1];
                }
            }

            const token = await provider();
            const headers = new Headers(nextInit.headers ?? {});
            headers.set('Authorization', `${scheme} ${token}`);
            return [nextUrl, { ...nextInit, headers }];
        };

        this._requestOptions = {
            ...this._requestOptions,
            hooks: {
                ...existingHooks,
                preRequest: composedPreRequest,
            },
        };
        return this;
    }

    /**
     * Builds and returns a configured fetch function.
     * Applies default options for any unset configurations.
     * @returns A configured fetch function with the specified options
     */
    build(): {
        (url: RequestInfo, init?: RequestInitWithOptions<Schema>): Promise<ResponseWithBody<ResponseType<Schema>>>;
    } {
        // Apply defaults for request options
        const requestOptions = {
            ...DEFAULTS,
            ...this._requestOptions,
        };

        // Apply defaults for container options if rate limit is set
        const containerOptions = this._containerOptions?.rateLimit
            ? {
                  ...this._containerOptions,
                  rateLimit: {
                      ...this._containerOptions.rateLimit,
                      retry: this._containerOptions.rateLimit.retry || DEFAULT_RATE_LIMIT_RETRY_OPTIONS,
                  },
              }
            : this._containerOptions;

        return generateFetchWithOptions({
            init: this._init,
            requestOptions: requestOptions as RequestOptions<never>,
            containerOptions,
        });
    }
}

// Example usage:
/*
const fetch = new FetchBuilder()
    .withTimeout(60 * 1000)
    .withRetry(DEFAULT_RETRY_OPTIONS)
    .withRateLimit(2, 62 * 1000, DEFAULT_RATE_LIMIT_RETRY_OPTIONS)
    .withLogger(logger)
    .withHooks({
        preRequest: (url, init) => {
            // Modify URL and request before sending
            const modifiedUrl = new URL(url.toString());
            modifiedUrl.searchParams.set('timestamp', Date.now().toString());
            
            init.headers = {
                ...init.headers,
                'Custom-Header': 'value'
            };
            
            return [modifiedUrl, init];
        },
        postResponseSuccess: (url, init, response) => {
            // Modify successful response with access to original request details
            if (response.isJson && response.data) {
                response.data = {
                    ...response.data,
                    processed: true,
                    requestUrl: url.toString(),
                    requestMethod: init.method
                };
            }
            return response;
        },
        postResponseError: (url, init, error, response) => {
            // Handle or modify error with access to request details
            if (error instanceof HTTPResponseError) {
                console.error(`HTTP Error for ${init.method} ${url}:`, error.response.status);
                // Could modify error message to include request details
                return new Error(`Request to ${url} failed with status ${error.response.status}`);
            }
            return error;
        }
    })
    .build();
*/