| 29 | const CONNECTION_LIMIT_REACHED_CODE = "websocket_connection_limit_reached" |
| 30 | |
| 31 | export function createWebSocketFetch(options?: CreateWebSocketFetchOptions) { |
| 32 | const httpFetch = options?.httpFetch ?? globalThis.fetch |
| 33 | const pool = new Map<string, PoolEntry>() |
| 34 | const connectTimeout = options?.connectTimeout ?? DEFAULT_CONNECT_TIMEOUT |
| 35 | const idleTimeout = options?.idleTimeout ?? DEFAULT_IDLE_TIMEOUT |
| 36 | const maxConnectionAge = options?.maxConnectionAge ?? DEFAULT_MAX_CONNECTION_AGE |
| 37 | const streamRetries = options?.streamRetries ?? 5 |
| 38 | const pruneTimer = setInterval(() => prune(), Math.min(idleTimeout, 60_000)) |
| 39 | if (typeof pruneTimer === "object" && "unref" in pruneTimer && typeof pruneTimer.unref === "function") { |
| 40 | pruneTimer.unref() |
| 41 | } |
| 42 | |
| 43 | async function websocketFetch(input: RequestInfo | URL, init?: RequestInit): Promise<Response> { |
| 44 | const url = input instanceof URL ? input.toString() : typeof input === "string" ? input : input.url |
| 45 | const internalHeaders = OpenAIWebSocket.normalizeHeaders(init?.headers) |
| 46 | const httpInit = withoutInternalHeaders(init) |
| 47 | |
| 48 | if (init?.method !== "POST" || !new URL(url).pathname.endsWith("/responses")) { |
| 49 | return httpFetch(input, httpInit) |
| 50 | } |
| 51 | |
| 52 | const body = (() => { |
| 53 | try { |
| 54 | if (typeof init?.body !== "string") return undefined |
| 55 | const parsed = JSON.parse(init.body) |
| 56 | return typeof parsed === "object" && parsed !== null ? parsed : undefined |
| 57 | } catch { |
| 58 | return undefined |
| 59 | } |
| 60 | })() |
| 61 | if (!body?.stream) return httpFetch(input, httpInit) |
| 62 | if (internalHeaders[TITLE_HEADER] === "true") { |
| 63 | return httpFetch(input, httpInit) |
| 64 | } |
| 65 | |
| 66 | const sessionID = internalHeaders["x-session-affinity"] ?? internalHeaders["session-id"] |
| 67 | if (!sessionID) { |
| 68 | return httpFetch(input, httpInit) |
| 69 | } |
| 70 | const key = `${sessionID}:conversation` |
| 71 | |
| 72 | const entry = pool.get(key) ?? { lastUsedAt: Date.now(), busy: false, fallback: false, streamFailures: 0 } |
| 73 | pool.set(key, entry) |
| 74 | |
| 75 | if (entry.fallback) { |
| 76 | return httpFetch(input, httpInit) |
| 77 | } |
| 78 | if (entry.busy) { |
| 79 | return httpFetch(input, httpInit) |
| 80 | } |
| 81 | |
| 82 | entry.busy = true |
| 83 | entry.lastUsedAt = Date.now() |
| 84 | try { |
| 85 | entry.socket = await socket( |
| 86 | entry, |
| 87 | options?.url ?? url, |
| 88 | OpenAIWebSocket.normalizeHeaders(httpInit?.headers), |