import { Effect, Stream } from "effect" import { Headers, HttpClientRequest } from "effect/unstable/http" import { Auth } from "../auth" import { render as renderEndpoint } from "../endpoint" import { Framing, type Framing as FramingDef } from "../framing" import type { Transport, TransportPrepareInput } from "./index" import * as ProviderShared from "../../protocols/shared" import { mergeJsonRecords, type LLMRequest } from "../../schema" export type JsonRequestInput = TransportPrepareInput export interface JsonRequestParts { readonly url: string readonly jsonBody: Body | Record readonly bodyText: string readonly headers: Headers.Headers } export interface HttpPrepared { readonly request: HttpClientRequest.HttpClientRequest readonly framing: FramingDef } const applyQuery = (url: string, query: Record | undefined) => { if (!query) return url const next = new URL(url) Object.entries(query).forEach(([key, value]) => next.searchParams.set(key, value)) return next.toString() } const bodyWithOverlay = (body: Body, request: LLMRequest, encodeBody: (body: Body) => string) => Effect.gen(function* () { if (request.http?.body === undefined) return { jsonBody: body, bodyText: encodeBody(body) } if (ProviderShared.isRecord(body)) { const overlaid = mergeJsonRecords(body, request.http.body) ?? {} return { jsonBody: overlaid, bodyText: ProviderShared.encodeJson(overlaid) } } return yield* ProviderShared.invalidRequest("http.body can only overlay JSON object request bodies") }) export const jsonRequestParts = (input: JsonRequestInput) => Effect.gen(function* () { const url = applyQuery( renderEndpoint(input.endpoint, { request: input.request, body: input.body }).toString(), input.request.http?.query, ) const body = yield* bodyWithOverlay(input.body, input.request, input.encodeBody) const headers = yield* Auth.toEffect(input.auth)({ request: input.request, method: "POST", url, body: body.bodyText, headers: Headers.fromInput({ ...input.headers?.({ request: input.request }), ...input.request.http?.headers, }), }) return { url, jsonBody: body.jsonBody, bodyText: body.bodyText, headers } }) export interface HttpJsonInput<_Body, Frame> { readonly framing: FramingDef } export type HttpJsonPatch = Partial> export interface HttpJsonTransport extends Transport, Frame> { readonly with: (patch: HttpJsonPatch) => HttpJsonTransport } export const httpJson = (input: HttpJsonInput): HttpJsonTransport => ({ id: "http-json", with: (patch) => httpJson({ ...input, ...patch }), prepare: (prepareInput) => jsonRequestParts({ ...prepareInput, }).pipe( Effect.map((parts) => ({ request: ProviderShared.jsonPost({ url: parts.url, body: parts.bodyText, headers: parts.headers }), framing: input.framing, })), ), frames: (prepared, request, runtime) => Stream.unwrap( runtime.http .execute(prepared.request) .pipe( Effect.map((response) => prepared.framing.frame( response.stream.pipe( Stream.mapError((error) => ProviderShared.eventError( `${request.model.provider}/${request.model.route.id}`, `Failed to read ${request.model.provider}/${request.model.route.id} stream`, ProviderShared.errorText(error), ), ), ), ), ), ), ), }) export const sseJson = { id: "http-json/sse", with: () => httpJson({ framing: Framing.sse }), } as const