diff --git a/packages/client/src/promise/client.ts b/packages/client/src/promise/client.ts index ff07c1caa24..0db239d9cc5 100644 --- a/packages/client/src/promise/client.ts +++ b/packages/client/src/promise/client.ts @@ -9,7 +9,7 @@ export type { ClientOptions, RequestOptions } from "./generated/client.js" export function make(options: ClientOptions) { const raw = OpenCode.make(options) - const events = SharedEvents.make((signal) => raw.event.subscribe({ signal })) + const events = SharedEvents.make((signal, onActivity) => raw.event.subscribe({ signal, onActivity })) return { ...raw, rpc: Object.assign(makeRpc(raw, events), raw.rpc), diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index fb55ccc0b63..3241880cd91 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -278,6 +278,8 @@ export interface ClientOptions { export interface RequestOptions { readonly signal?: AbortSignal readonly headers?: RequestInit["headers"] + /** Reports every chunk a streaming response receives, including keepalive comments that yield no event. */ + readonly onActivity?: () => void } interface RequestDescriptor { @@ -369,6 +371,7 @@ export function make(options: ClientOptions) { } catch (cause) { throw new ClientError("Transport", { cause }) } + if (!next.done) requestOptions?.onActivity?.() buffer += decoder.decode(next.value, { stream: !next.done }) if (buffer.length > maxSseEventBytes) throw new ClientError("SseEventTooLarge") const trailingCarriageReturn = !next.done && buffer.endsWith("\r") diff --git a/packages/client/src/shared-events.ts b/packages/client/src/shared-events.ts index 322368551d4..f8e4ce919ed 100644 --- a/packages/client/src/shared-events.ts +++ b/packages/client/src/shared-events.ts @@ -1,10 +1,19 @@ export * as SharedEvents from "./shared-events.js" -export function make(connect: (signal: AbortSignal) => AsyncIterable) { +export type SubscribeOptions = { + readonly signal?: AbortSignal + /** Reports transport activity on the shared stream, including keepalive frames that carry no event. */ + readonly onActivity?: () => void +} + +export function make( + connect: (signal: AbortSignal, onActivity: () => void) => AsyncIterable, +) { type Completion = { readonly error: unknown } | Record type Subscriber = { push: (value: A) => void finish: (completion: Completion) => void + activity?: () => void } type Connection = { controller: AbortController @@ -26,7 +35,9 @@ export function make(connect: (signal: Abor let completion: Completion = {} try { if (connection.controller.signal.aborted) return - iterator = connect(connection.controller.signal)[Symbol.asyncIterator]() + iterator = connect(connection.controller.signal, () => { + connection.subscribers.forEach((subscriber) => subscriber.activity?.()) + })[Symbol.asyncIterator]() while (!connection.controller.signal.aborted) { const item = await iterator.next() if (item.done || connection.controller.signal.aborted) break @@ -47,7 +58,7 @@ export function make(connect: (signal: Abor } return { - subscribe(options?: { readonly signal?: AbortSignal }): AsyncIterable { + subscribe(options?: SubscribeOptions): AsyncIterable { return { [Symbol.asyncIterator]() { const pending: ReturnType>>[] = [] @@ -72,6 +83,7 @@ export function make(connect: (signal: Abor } const subscriber: Subscriber = { + activity: options?.onActivity, finish(result) { finish(result, false) }, diff --git a/packages/client/src/solid/connection.ts b/packages/client/src/solid/connection.ts index 32f5af5f43c..01e6b75b28c 100644 --- a/packages/client/src/solid/connection.ts +++ b/packages/client/src/solid/connection.ts @@ -1,4 +1,4 @@ -import { batch, onCleanup, onMount } from "solid-js" +import { batch, onCleanup } from "solid-js" import { createStore } from "solid-js/store" import type { OpenCodeClient, OpenCodeEvent } from "../promise" @@ -18,6 +18,11 @@ export type ClientConnectionOptions = { readonly onEvent: (event: OpenCodeEvent) => void readonly flushInterval?: number readonly pageLifecycle?: boolean + /** + * Abort and reconnect a stream that receives no bytes for this long. The server writes a keepalive + * comment every 15 seconds, so a quiet but healthy stream never trips this. + */ + readonly idleTimeout?: number readonly log?: { readonly debug?: (message: string, data?: Readonly>) => void readonly info?: (message: string, data?: Readonly>) => void @@ -27,10 +32,15 @@ export type ClientConnectionOptions = { const connectTimeout = 2_000 const reconnectDelay = 1_000 const connectionHistoryLimit = 50 +export const defaultIdleTimeout = 45_000 +// Longer than one server keepalive interval: a stream that is silent this long when the page +// returns to the foreground is probably half-open after the device slept. +export const foregroundIdleThreshold = 20_000 export function createClientConnection(initialApi: OpenCodeClient, options: ClientConnectionOptions) { const abort = new AbortController() const history: ClientConnectionEvent[] = [] + const idleTimeout = options.idleTimeout ?? defaultIdleTimeout const [connection, setConnection] = createStore<{ status: ClientConnectionStatus attempt: number @@ -40,9 +50,12 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie let pending: OpenCodeEvent[] = [] let flushTimer: ReturnType | undefined let stream: AbortController | undefined + let current: AbortController | undefined let run: Promise | undefined let started = false let generation = 0 + let lastActivity = 0 + let forced = false function record(status: ClientConnectionEvent["data"]["status"], attempt: number, error?: string) { history.push({ type: "client.connection", created: Date.now(), data: { status, attempt, error } }) @@ -63,14 +76,25 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie async function connect(signal: AbortSignal, attempt: number) { let connectedAt: number | undefined const request = new AbortController() + current = request const cancel = () => request.abort(signal.reason) const timeout = setTimeout(() => request.abort(new Error("Timed out connecting to server")), connectTimeout) signal.addEventListener("abort", cancel, { once: true }) + // Any received bytes, including keepalive comments, push the stall deadline out. A timer whose + // deadline passed while the page was suspended fires as soon as the page resumes. + let watchdog: ReturnType | undefined + const touch = () => { + lastActivity = Date.now() + if (connectedAt === undefined) return + clearTimeout(watchdog) + watchdog = setTimeout(() => request.abort(new Error("Event stream stalled")), idleTimeout) + } + try { record(attempt === 0 ? "connecting" : "reconnecting", attempt) options.log?.info?.("event stream connecting", { attempt }) - const iterator = api.event.subscribe({ signal: request.signal })[Symbol.asyncIterator]() + const iterator = api.event.subscribe({ signal: request.signal, onActivity: touch })[Symbol.asyncIterator]() const first = await iterator.next() if (signal.aborted) return { error: undefined, connectedAt } if (first.done) @@ -85,6 +109,7 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie clearTimeout(timeout) record("connected", attempt) connectedAt = Date.now() + touch() options.log?.info?.("event stream connected") publish(first.value) setConnection({ status: "connected", attempt: 0, error: undefined }) @@ -92,7 +117,13 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie while (!signal.aborted) { const event = await iterator.next() if (signal.aborted) return { error: undefined, connectedAt } - if (event.done) return { error: new Error("Event stream disconnected"), connectedAt } + if (event.done) + return { + error: + request.signal.reason instanceof Error ? request.signal.reason : new Error("Event stream disconnected"), + connectedAt, + } + touch() if ("durable" in event.value && event.value.durable) options.log?.debug?.("event", { type: event.value.type, @@ -106,7 +137,9 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie return { error, connectedAt } } finally { request.abort() + if (current === request) current = undefined clearTimeout(timeout) + clearTimeout(watchdog) signal.removeEventListener("abort", cancel) } } @@ -140,6 +173,11 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie if (attempt === 1) continue } } + // A deliberate resync already knows the old socket is gone; reconnect without backing off. + if (forced) { + forced = false + continue + } await wait(reconnectDelay, controller.signal) } } @@ -147,6 +185,7 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie function start() { if (started) return run started = true + forced = false const active = ++generation const previous = run const current = (async () => { @@ -161,26 +200,45 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie } function stop() { + if (!started) return started = false generation += 1 stream?.abort() + // Nothing is listening once stopped, so consumers must treat their data as stale until start() reconnects. + setConnection({ status: "connecting", attempt: 0, error: undefined }) } - onMount(() => { - if (options.pageLifecycle) { - const pagehide = () => stop() - const pageshow = (event: PageTransitionEvent) => { - if (event.persisted) void start() - } - window.addEventListener("pagehide", pagehide) - window.addEventListener("pageshow", pageshow) - onCleanup(() => { - window.removeEventListener("pagehide", pagehide) - window.removeEventListener("pageshow", pageshow) - }) + // Drop the live request so the reconnect loop replaces it now instead of waiting for the idle watchdog. + function resync(reason: string) { + if (!started || connection.status !== "connected") return + options.log?.info?.("event stream resync", { reason, idle: Date.now() - lastActivity }) + forced = true + current?.abort(new Error(reason)) + } + + if (options.pageLifecycle) { + const pagehide = () => stop() + const pageshow = () => void start() + // Locking a phone or switching apps hides the document without a pagehide; the socket usually + // dies while the page is suspended, and the browser may never report that on the hung read. + const visibility = () => { + if (document.visibilityState !== "visible") return + if (Date.now() - lastActivity < foregroundIdleThreshold) return + resync("Page returned to the foreground after the event stream went quiet") } - void start() - }) + const online = () => resync("Network connection restored") + window.addEventListener("pagehide", pagehide) + window.addEventListener("pageshow", pageshow) + window.addEventListener("online", online) + document.addEventListener("visibilitychange", visibility) + onCleanup(() => { + window.removeEventListener("pagehide", pagehide) + window.removeEventListener("pageshow", pageshow) + window.removeEventListener("online", online) + document.removeEventListener("visibilitychange", visibility) + }) + } + void start() onCleanup(() => { stop() @@ -195,6 +253,7 @@ export function createClientConnection(initialApi: OpenCodeClient, options: Clie error: () => connection.error, internal: { history: () => history.slice(), + resync, }, } } diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index 63f2a2328aa..9c5b7554e50 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -675,6 +675,31 @@ test("event.subscribe ignores server heartbeat comments", async () => { expect(received).toEqual([event]) }) +test("event.subscribe reports heartbeat comments as stream activity", async () => { + const event = { id: "evt_sentinel", created: 1, type: "server.connected", data: {} } + const encoder = new TextEncoder() + const client = OpenCode.make({ + baseUrl: "http://localhost:3000", + fetch: async () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(": heartbeat\n\n")) + controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`)) + controller.enqueue(encoder.encode(": heartbeat\n\n")) + controller.close() + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ), + }) + let activity = 0 + const received = [] + for await (const item of client.event.subscribe({ onActivity: () => activity++ })) received.push(item) + expect(received).toEqual([event]) + expect(activity).toBe(3) +}) + // Moved from packages/app/e2e/regression/session-timeline-transport.spec.ts test("event transport passes through ordinary health requests", async () => { const requests: string[] = [] diff --git a/packages/client/test/shared-events.test.ts b/packages/client/test/shared-events.test.ts index bd99ca853f4..bf3370be5e8 100644 --- a/packages/client/test/shared-events.test.ts +++ b/packages/client/test/shared-events.test.ts @@ -359,3 +359,24 @@ test("synchronous source creation failures reject subscribers without automatic await expect(shared.subscribe()[Symbol.asyncIterator]().next()).rejects.toBe(failure) expect(attempts).toHaveLength(2) }) + +test("source activity fans out to every subscriber that asked for it", async () => { + const events = source() + let activity: (() => void) | undefined + const shared = SharedEvents.make((signal, onActivity) => { + activity = onActivity + return events.connect(signal) + }) + const counts = { first: 0, second: 0 } + const first = shared.subscribe({ onActivity: () => counts.first++ })[Symbol.asyncIterator]() + const second = shared.subscribe()[Symbol.asyncIterator]() + const reads = Promise.all([first.next(), second.next()]) + activity!() + activity!() + events.connections[0].push({ type: "server.connected" }) + await reads + expect(counts).toEqual({ first: 2, second: 0 }) + await first.return!() + await second.return!() + await events.connections[0].closed +}) diff --git a/packages/client/test/solid-connection.test.ts b/packages/client/test/solid-connection.test.ts new file mode 100644 index 00000000000..87448aead8f --- /dev/null +++ b/packages/client/test/solid-connection.test.ts @@ -0,0 +1,153 @@ +import { expect, test } from "bun:test" +import { createRoot } from "solid-js" +import { createClientConnection } from "../src/solid" +import { OpenCode, type OpenCodeEvent } from "../src/promise" + +const connected = { id: "evt_connected", created: 1, type: "server.connected", data: {} } + +// One fake server whose event streams stay open until the test writes to them or the client aborts. +function server() { + const encoder = new TextEncoder() + const streams: { + write: (text: string) => void + close: () => void + aborted: boolean + }[] = [] + const api = OpenCode.make({ + baseUrl: "http://opencode.local", + fetch: async (input, init) => { + const request = input instanceof Request ? input : new Request(input, init) + let controller!: ReadableStreamDefaultController + const entry = { + write: (text: string) => controller.enqueue(encoder.encode(text)), + close: () => controller.close(), + aborted: false, + } + const body = new ReadableStream({ + start(value) { + controller = value + }, + cancel() { + entry.aborted = true + }, + }) + request.signal.addEventListener("abort", () => { + entry.aborted = true + controller.error(request.signal.reason) + }) + streams.push(entry) + return new Response(body, { headers: { "content-type": "text/event-stream" } }) + }, + }) + return { api, streams } +} + +function setup(input: ReturnType, idleTimeout: number) { + const events: OpenCodeEvent[] = [] + return createRoot((dispose) => ({ + events, + dispose, + connection: createClientConnection(input.api, { + idleTimeout, + flushInterval: 0, + onEvent: (event) => events.push(event), + }), + })) +} + +async function until(check: () => boolean, timeout = 2_000) { + const deadline = Date.now() + timeout + while (!check()) { + if (Date.now() > deadline) throw new Error("Timed out waiting for condition") + await new Promise((resolve) => setTimeout(resolve, 5)) + } +} + +test("a stream that goes silent past the idle timeout is replaced", async () => { + const fake = server() + const ctx = setup(fake, 60) + try { + await until(() => fake.streams.length === 1) + fake.streams[0].write(`data: ${JSON.stringify(connected)}\n\n`) + await until(() => ctx.connection.status() === "connected") + + await until(() => fake.streams.length === 2) + expect(fake.streams[0].aborted).toBe(true) + expect(ctx.connection.internal.history().map((item) => item.data)).toContainEqual({ + status: "disconnected", + attempt: 1, + error: "Event stream stalled", + }) + + fake.streams[1].write(`data: ${JSON.stringify({ ...connected, id: "evt_connected_2" })}\n\n`) + await until(() => ctx.connection.status() === "connected" && ctx.events.length === 2) + expect(ctx.connection.error()).toBeUndefined() + } finally { + ctx.dispose() + } +}) + +test("keepalive comments hold a quiet stream open", async () => { + const fake = server() + const ctx = setup(fake, 60) + try { + await until(() => fake.streams.length === 1) + fake.streams[0].write(`data: ${JSON.stringify(connected)}\n\n`) + await until(() => ctx.connection.status() === "connected") + + const heartbeat = setInterval(() => fake.streams[0].write(": heartbeat\n\n"), 20) + await new Promise((resolve) => setTimeout(resolve, 250)) + clearInterval(heartbeat) + + expect(fake.streams).toHaveLength(1) + expect(fake.streams[0].aborted).toBe(false) + expect(ctx.connection.status()).toBe("connected") + expect(ctx.events).toHaveLength(1) + } finally { + ctx.dispose() + } +}) + +test("a forced resync replaces the stream immediately and only while connected", async () => { + const fake = server() + const ctx = setup(fake, 10_000) + try { + ctx.connection.internal.resync("too early") + await until(() => fake.streams.length === 1) + expect(fake.streams[0].aborted).toBe(false) + fake.streams[0].write(`data: ${JSON.stringify(connected)}\n\n`) + await until(() => ctx.connection.status() === "connected") + + const started = Date.now() + ctx.connection.internal.resync("Network connection restored") + await until(() => fake.streams.length === 2) + expect(Date.now() - started).toBeLessThan(500) + expect(fake.streams[0].aborted).toBe(true) + expect(ctx.connection.internal.history().map((item) => item.data)).toContainEqual({ + status: "disconnected", + attempt: 1, + error: "Network connection restored", + }) + fake.streams[1].write(`data: ${JSON.stringify(connected)}\n\n`) + await until(() => ctx.connection.status() === "connected") + } finally { + ctx.dispose() + } +}) + +test("a stream the server closes reconnects and reports the disconnect", async () => { + const fake = server() + const ctx = setup(fake, 10_000) + try { + await until(() => fake.streams.length === 1) + fake.streams[0].write(`data: ${JSON.stringify(connected)}\n\n`) + await until(() => ctx.connection.status() === "connected") + + fake.streams[0].close() + await until(() => ctx.connection.status() === "reconnecting") + expect(ctx.connection.error()).toBe("Event stream disconnected") + await until(() => fake.streams.length === 2) + } finally { + ctx.dispose() + } +}) diff --git a/packages/httpapi-codegen/src/index.ts b/packages/httpapi-codegen/src/index.ts index f8de6a2650d..001c5ae2123 100644 --- a/packages/httpapi-codegen/src/index.ts +++ b/packages/httpapi-codegen/src/index.ts @@ -961,7 +961,7 @@ function renderPromiseClient(groups: ReadonlyArray) { if (group.endpoints[0]?.topLevel) return fields return `${JSON.stringify(group.identifier)}: { ${fields} }` }) - return `import type { ${imports.join(", ")} } from "./types.js"\nimport { ClientError } from "./client-error.js"\n\nexport interface ClientOptions {\n readonly baseUrl: string\n readonly fetch?: typeof globalThis.fetch\n readonly headers?: RequestInit["headers"]\n}\n\nexport interface RequestOptions {\n readonly signal?: AbortSignal\n readonly headers?: RequestInit["headers"]\n}\n\ninterface RequestDescriptor {\n readonly method: string\n readonly path: string\n readonly query?: Record\n readonly headers?: Record\n readonly body?: unknown\n readonly successStatus: number\n readonly declaredStatuses: ReadonlyArray\n readonly empty: boolean\n}\n\nconst maxSseEventBytes = 16 * 1024 * 1024\n\nexport function make(options: ClientOptions) {\n const fetch = options.fetch ?? globalThis.fetch\n\n const prepare = (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n const url = new URL(descriptor.path, options.baseUrl)\n for (const [key, value] of Object.entries(descriptor.query ?? {})) appendQuery(url.searchParams, key, value)\n const headers = new Headers(options.headers)\n for (const [key, value] of Object.entries(descriptor.headers ?? {})) {\n if (value !== undefined && value !== null) headers.set(key, String(value))\n }\n for (const [key, value] of new Headers(requestOptions?.headers)) headers.set(key, value)\n if (descriptor.body !== undefined && !headers.has("content-type")) headers.set("content-type", "application/json")\n return {\n url,\n init: {\n method: descriptor.method,\n signal: requestOptions?.signal,\n headers,\n body: descriptor.body === undefined ? undefined : JSON.stringify(descriptor.body),\n } satisfies RequestInit,\n }\n }\n\n const execute = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n try {\n const prepared = prepare(descriptor, requestOptions)\n return await fetch(prepared.url, prepared.init)\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n }\n\n const responseError = async (response: Response, descriptor: RequestDescriptor): Promise => {\n if (descriptor.declaredStatuses.includes(response.status)) throw await json(response)\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnexpectedStatus", { cause: { status: response.status } })\n }\n\n const request = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions): Promise => {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) return responseError(response, descriptor)\n if (descriptor.empty) {\n try {\n await response.body?.cancel()\n } catch {}\n return undefined as A\n }\n return await json(response) as A\n }\n\n const sse = (descriptor: RequestDescriptor, requestOptions?: RequestOptions): AsyncIterable => ({\n async *[Symbol.asyncIterator]() {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) await responseError(response, descriptor)\n if (!isContentType(response, "text/event-stream")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n if (response.body === null) throw new ClientError("MalformedResponse")\n const reader = response.body.getReader()\n const decoder = new TextDecoder()\n let buffer = ""\n try {\n while (true) {\n let next: ReadableStreamReadResult\n try {\n next = await reader.read()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n buffer += decoder.decode(next.value, { stream: !next.done })\n if (buffer.length > maxSseEventBytes) throw new ClientError("SseEventTooLarge")\n const trailingCarriageReturn = !next.done && buffer.endsWith("\\r")\n if (trailingCarriageReturn) buffer = buffer.slice(0, -1)\n buffer = buffer.replaceAll("\\r\\n", "\\n").replaceAll("\\r", "\\n")\n if (trailingCarriageReturn) buffer += "\\r"\n if (next.done && buffer !== "") buffer += "\\n\\n"\n let boundary = buffer.indexOf("\\n\\n")\n while (boundary >= 0) {\n const block = buffer.slice(0, boundary)\n buffer = buffer.slice(boundary + 2)\n const data = block.split("\\n").flatMap((line) => line.startsWith("data:") ? [line.slice(5).trimStart()] : []).join("\\n")\n if (data !== "") {\n try {\n yield JSON.parse(data) as A\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n }\n boundary = buffer.indexOf("\\n\\n")\n }\n if (next.done) return\n }\n } finally {\n try {\n await reader.cancel()\n } catch {}\n reader.releaseLock()\n }\n },\n })\n\n return { ${fields.join(", ")} }\n}\n\nfunction appendQuery(params: URLSearchParams, key: string, value: unknown): void {\n if (value === undefined) return\n if (value === null) {\n params.append(key, "null")\n return\n }\n if (Array.isArray(value)) {\n for (const item of value) appendQuery(params, key, item)\n return\n }\n if (typeof value === "object") {\n for (const [child, item] of Object.entries(value)) appendQuery(params, \`\${key}[\${child}]\`, item)\n return\n }\n params.append(key, String(value))\n}\n\nasync function json(response: Response): Promise {\n if (!isContentType(response, "application/json") && !response.headers.get("content-type")?.includes("+json")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n let text: string\n try {\n text = await response.text()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n if (text === "") throw new ClientError("MalformedResponse")\n try {\n return JSON.parse(text)\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n}\n\nfunction isContentType(response: Response, expected: string) {\n return response.headers.get("content-type")?.split(";", 1)[0]?.trim().toLowerCase() === expected\n}\n` + return `import type { ${imports.join(", ")} } from "./types.js"\nimport { ClientError } from "./client-error.js"\n\nexport interface ClientOptions {\n readonly baseUrl: string\n readonly fetch?: typeof globalThis.fetch\n readonly headers?: RequestInit["headers"]\n}\n\nexport interface RequestOptions {\n readonly signal?: AbortSignal\n readonly headers?: RequestInit["headers"]\n /** Reports every chunk a streaming response receives, including keepalive comments that yield no event. */\n readonly onActivity?: () => void\n}\n\ninterface RequestDescriptor {\n readonly method: string\n readonly path: string\n readonly query?: Record\n readonly headers?: Record\n readonly body?: unknown\n readonly successStatus: number\n readonly declaredStatuses: ReadonlyArray\n readonly empty: boolean\n}\n\nconst maxSseEventBytes = 16 * 1024 * 1024\n\nexport function make(options: ClientOptions) {\n const fetch = options.fetch ?? globalThis.fetch\n\n const prepare = (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n const url = new URL(descriptor.path, options.baseUrl)\n for (const [key, value] of Object.entries(descriptor.query ?? {})) appendQuery(url.searchParams, key, value)\n const headers = new Headers(options.headers)\n for (const [key, value] of Object.entries(descriptor.headers ?? {})) {\n if (value !== undefined && value !== null) headers.set(key, String(value))\n }\n for (const [key, value] of new Headers(requestOptions?.headers)) headers.set(key, value)\n if (descriptor.body !== undefined && !headers.has("content-type")) headers.set("content-type", "application/json")\n return {\n url,\n init: {\n method: descriptor.method,\n signal: requestOptions?.signal,\n headers,\n body: descriptor.body === undefined ? undefined : JSON.stringify(descriptor.body),\n } satisfies RequestInit,\n }\n }\n\n const execute = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n try {\n const prepared = prepare(descriptor, requestOptions)\n return await fetch(prepared.url, prepared.init)\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n }\n\n const responseError = async (response: Response, descriptor: RequestDescriptor): Promise => {\n if (descriptor.declaredStatuses.includes(response.status)) throw await json(response)\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnexpectedStatus", { cause: { status: response.status } })\n }\n\n const request = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions): Promise => {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) return responseError(response, descriptor)\n if (descriptor.empty) {\n try {\n await response.body?.cancel()\n } catch {}\n return undefined as A\n }\n return await json(response) as A\n }\n\n const sse = (descriptor: RequestDescriptor, requestOptions?: RequestOptions): AsyncIterable => ({\n async *[Symbol.asyncIterator]() {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) await responseError(response, descriptor)\n if (!isContentType(response, "text/event-stream")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n if (response.body === null) throw new ClientError("MalformedResponse")\n const reader = response.body.getReader()\n const decoder = new TextDecoder()\n let buffer = ""\n try {\n while (true) {\n let next: ReadableStreamReadResult\n try {\n next = await reader.read()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n if (!next.done) requestOptions?.onActivity?.()\n buffer += decoder.decode(next.value, { stream: !next.done })\n if (buffer.length > maxSseEventBytes) throw new ClientError("SseEventTooLarge")\n const trailingCarriageReturn = !next.done && buffer.endsWith("\\r")\n if (trailingCarriageReturn) buffer = buffer.slice(0, -1)\n buffer = buffer.replaceAll("\\r\\n", "\\n").replaceAll("\\r", "\\n")\n if (trailingCarriageReturn) buffer += "\\r"\n if (next.done && buffer !== "") buffer += "\\n\\n"\n let boundary = buffer.indexOf("\\n\\n")\n while (boundary >= 0) {\n const block = buffer.slice(0, boundary)\n buffer = buffer.slice(boundary + 2)\n const data = block.split("\\n").flatMap((line) => line.startsWith("data:") ? [line.slice(5).trimStart()] : []).join("\\n")\n if (data !== "") {\n try {\n yield JSON.parse(data) as A\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n }\n boundary = buffer.indexOf("\\n\\n")\n }\n if (next.done) return\n }\n } finally {\n try {\n await reader.cancel()\n } catch {}\n reader.releaseLock()\n }\n },\n })\n\n return { ${fields.join(", ")} }\n}\n\nfunction appendQuery(params: URLSearchParams, key: string, value: unknown): void {\n if (value === undefined) return\n if (value === null) {\n params.append(key, "null")\n return\n }\n if (Array.isArray(value)) {\n for (const item of value) appendQuery(params, key, item)\n return\n }\n if (typeof value === "object") {\n for (const [child, item] of Object.entries(value)) appendQuery(params, \`\${key}[\${child}]\`, item)\n return\n }\n params.append(key, String(value))\n}\n\nasync function json(response: Response): Promise {\n if (!isContentType(response, "application/json") && !response.headers.get("content-type")?.includes("+json")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n let text: string\n try {\n text = await response.text()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n if (text === "") throw new ClientError("MalformedResponse")\n try {\n return JSON.parse(text)\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n}\n\nfunction isContentType(response: Response, expected: string) {\n return response.headers.get("content-type")?.split(";", 1)[0]?.trim().toLowerCase() === expected\n}\n` } function promiseTypePrefix(group: string, path: ReadonlyArray) {