import type { Server } from "bun";
import {
  codexWsUpstreamFetch,
  currentBunRuntimeIdentity,
  shouldUseCodexWsUpstream,
  type BunRuntimeGateInput,
} from "./ws-upstream";
import type { OcxProviderConfig } from "../../types";
import type { WsData } from "../ws-bridge";
import { waitForProviderRequestSlot } from "../../providers/request-pacing";
import { withUpstreamHttpVersion } from "../../lib/upstream-http-version";
import { providerTlsFetch } from "../../lib/provider-tls-profile";
import { testProviderFetch } from "../../lib/test-provider-fetch";
import { runtimeProviderFetch } from "../../lib/provider-runtime-fetch";
import type { CodexWsQuotaObserver } from "./codex-ws-metadata";

export { withUpstreamHttpVersion };

export function disableResponsesRequestTimeout(req: Request, server: Pick<Server<WsData>, "timeout"> | undefined): boolean {
  if (!server) return false;
  try {
    server.timeout(req, 0);
    return true;
  } catch {
    return false;
  }
}



export function safeHostLabel(url: string): string {
  try {
    return new URL(url).host;
  } catch {
    return "upstream";
  }
}

/** Canonical origin (scheme + host) for failure-attribution keys: http and
 * https for the same host must not share one ledger entry (#914 review). */
export function safeOriginLabel(url: string): string {
  try {
    return new URL(url).origin.toLowerCase();
  } catch {
    return "upstream";
  }
}



export interface PaceAwareFetch {
  waitForPacing?: (signal?: AbortSignal) => Promise<void>;
  unpacedFetch?: typeof globalThis.fetch;
}

export type ProviderFetch = typeof globalThis.fetch & PaceAwareFetch;

export class UpstreamRedirectError extends Error {
  override readonly name = "UpstreamRedirectError";

  constructor(readonly status: number) {
    super(`upstream returned ${status} redirect; configure the final upstream URL directly`);
  }
}

export interface ProviderFetchOptions {
  providerName?: string;
  modelId?: string;
  /** One pacing slot was acquired immediately before this fetch wrapper was created. */
  pacingSlotAcquired?: boolean;
  /** Explicit test/integration executor; never read from serialized provider config. */
  fetch?: typeof globalThis.fetch;
  /** Captured selected-account observer, attached before the native WS send. */
  onCodexWsQuota?: CodexWsQuotaObserver;
  /** Synchronous admission at actual credential dispatch, after pacing/backoff. */
  beforeDispatch?: (headers: Headers) => void;
  /** Revalidate/rebuild a queued request at its physical send boundary, after pacing. */
  dispatchOverride?: (input: Parameters<typeof globalThis.fetch>[0], init: RequestInit, execute: typeof globalThis.fetch) => Promise<Response>;
}

export function providerFetch(
  provider: OcxProviderConfig,
  runtime: BunRuntimeGateInput = currentBunRuntimeIdentity(),
  options: ProviderFetchOptions = {},
): ProviderFetch {
  const base = options.fetch
    ?? testProviderFetch(provider)
    ?? runtimeProviderFetch(provider, options.providerName)
    ?? (provider as OcxProviderConfig & { fetch?: typeof globalThis.fetch }).fetch
    ?? globalThis.fetch;
  const preconnect = (...args: Parameters<typeof globalThis.fetch.preconnect>): void => {
    base.preconnect?.(...args);
  };
  const transport = options.providerName
    ? providerTlsFetch(options.providerName, provider, base)
    : base;
  const httpFetch = Object.assign(
    async (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
      options.beforeDispatch?.(new Headers(init?.headers ?? (input instanceof Request ? input.headers : undefined)));
      const dispatchInit = { ...withUpstreamHttpVersion(input, init, provider), timeout: 0 };
      return options.dispatchOverride
        ? options.dispatchOverride(input, dispatchInit, transport)
        : transport(input, dispatchInit);
    },
    { preconnect },
  ) as typeof globalThis.fetch;
  // ChatGPT Codex backend: streaming turns ride the responses_websockets
  // transport (measured ~3s faster TTFT than the SSE POST queue); everything
  // else keeps the provider's HTTP fetch. See ws-upstream.ts for the details.
  const unpaced = async (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
    const upstreamWebsocket = provider.upstreamWebsocket === true;
    const wsOpts = {
      // Keep the canonical ChatGPT fast lane independent: upstreamWebsocket opts a
      // configured HTTPS /responses endpoint in, but must not silently enable Codex WS.
      wsUpstream: provider.wsUpstream,
      maxWsFrameBytes: provider.maxWsFrameBytes,
      upstreamWebsocket,
    };
    if (typeof input === "string" && init && shouldUseCodexWsUpstream(input, init, runtime, wsOpts)) {
      // The fallback has to be the same HTTP fetch the non-WS branch would have
      // used, protocol pin included: a WS turn that falls back is serving the
      // request over HTTP, and dropping the provider's `upstreamHttpVersion`
      // there would silently negotiate a transport the operator ruled out.
      return codexWsUpstreamFetch(input, init, httpFetch, runtime, wsOpts, options.onCodexWsQuota, options.beforeDispatch);
    }
    return httpFetch(input, init);
  };
  let pacingSlotAcquired = options.pacingSlotAcquired === true;
  const waitForPacing = (signal?: AbortSignal) => {
    if (pacingSlotAcquired) {
      pacingSlotAcquired = false;
      return Promise.resolve();
    }
    return options.providerName
      ? waitForProviderRequestSlot(options.providerName, provider, options.modelId, signal)
      : Promise.resolve();
  };
  const wrapped = async (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
    await waitForPacing(init?.signal ?? undefined);
    return unpaced(input, init);
  };
  return Object.assign(wrapped, {
    preconnect,
    waitForPacing,
    unpacedFetch: Object.assign(unpaced, { preconnect }),
  });
}



/**
 * Wrap a provider fetch so `onDispatch` fires immediately before the send, not before pacing.
 *
 * `fetchWithHeaderTimeout` awaits `waitForPacing` and only then calls the executor, so a caller
 * that signals at the call site records a dispatch even when a rejected pacing wait means nothing
 * reached the network. That matters when the signal bounds later recovery: the request would lose
 * its fallback on the strength of a send that never happened.
 *
 * The pacing surface is preserved deliberately. `waitForPacing` and `unpacedFetch` are read off
 * the executor by `fetchWithHeaderTimeout`, so a plain function wrapper would silently drop
 * provider pacing and double-send the slot.
 */
export function storedPoolReplayDispatchNotifier(
  executor: ProviderFetch,
  onDispatch: (() => void) | undefined,
): ProviderFetch {
  if (!onDispatch) return executor;
  let notified = false;
  const notifyOnce = (): void => {
    if (notified) return;
    notified = true;
    onDispatch();
  };
  const unpacedSource = executor.unpacedFetch ?? executor;
  const unpaced = Object.assign(
    (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
      notifyOnce();
      return unpacedSource(input, init);
    },
    { preconnect: unpacedSource.preconnect },
  ) as ProviderFetch["unpacedFetch"];
  const wrapped = async (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
    await executor.waitForPacing?.(init?.signal ?? undefined);
    return unpaced!(input, init);
  };
  return Object.assign(wrapped, {
    preconnect: executor.preconnect,
    waitForPacing: executor.waitForPacing,
    unpacedFetch: unpaced,
  }) as ProviderFetch;
}

export async function fetchWithHeaderTimeout(
  url: string,
  init: Omit<RequestInit, "signal">,
  abortSignal: AbortSignal,
  timeoutMs: number,
  preferIdentityEncoding = false,
  executor: typeof globalThis.fetch = globalThis.fetch,
  _manualRedirect = true,
): Promise<Response> {
  const pacing = executor as ProviderFetch;
  await pacing.waitForPacing?.(abortSignal);
  const fetchExecutor = pacing.unpacedFetch ?? executor;
  const timeout = new AbortController();
  const timer = setTimeout(() => {
    if (!timeout.signal.aborted) timeout.abort(new DOMException("Timeout elapsed", "TimeoutError"));
  }, timeoutMs);
  const headers = new Headers(init.headers);
  // Compressed SSE can be held until the decompressor has a complete block. Streaming calls
  // default to identity for low-latency frame delivery, while an explicit caller choice wins.
  if (preferIdentityEncoding && !headers.has("accept-encoding")) {
    headers.set("accept-encoding", "identity");
  }
  try {
    const response = await fetchExecutor(url, {
      ...init,
      headers,
      // Upstream URLs are configuration, not navigation. Refuse every redirect
      // so POST bodies and provider headers are never replayed to another hop.
      redirect: "manual" as const,
      signal: AbortSignal.any([abortSignal, timeout.signal]),
      timeout: 0,
    });
    if (response.status >= 300 && response.status < 400) {
      try { await response.body?.cancel(); } catch { /* ignore cancellation failures */ }
      throw new UpstreamRedirectError(response.status);
    }
    return response;
  } finally {
    clearTimeout(timer);
  }
}