156 lines
5.7 KiB
TypeScript
156 lines
5.7 KiB
TypeScript
import { Effect, Stream } from "effect"
|
|
import { Headers, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
|
|
import { Auth } from "../auth"
|
|
import { render as renderEndpoint } from "../endpoint"
|
|
import { Framing } from "../framing"
|
|
import type { HttpMiddleware, Transport, TransportPrepareInput } from "./index"
|
|
import * as ProviderShared from "../../protocols/shared"
|
|
import { LLMError, mergeJsonRecords, type LLMRequest } from "../../schema"
|
|
|
|
export type JsonRequestInput<Body> = TransportPrepareInput<Body>
|
|
|
|
export interface JsonRequestParts<Body = unknown> {
|
|
readonly url: string
|
|
readonly jsonBody: Body | Record<string, unknown>
|
|
readonly bodyText: string
|
|
readonly headers: Headers.Headers
|
|
}
|
|
|
|
export interface HttpPrepared<Frame> {
|
|
readonly request: HttpClientRequest.HttpClientRequest
|
|
readonly web: Request
|
|
readonly framing: Framing.Definition<Frame>
|
|
readonly middleware?: HttpMiddleware
|
|
}
|
|
|
|
const applyQuery = (url: string, query: Record<string, string> | 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: 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 = <Body>(input: JsonRequestInput<Body>) =>
|
|
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: Framing.Definition<Frame>
|
|
}
|
|
|
|
export type HttpJsonPatch<Body, Frame> = Partial<HttpJsonInput<Body, Frame>>
|
|
|
|
export interface HttpJsonTransport<Body, Frame> extends Transport<Body, HttpPrepared<Frame>, Frame> {
|
|
readonly with: (patch: HttpJsonPatch<Body, Frame>) => HttpJsonTransport<Body, Frame>
|
|
}
|
|
|
|
export const httpJson = <Body, Frame>(input: HttpJsonInput<Body, Frame>): HttpJsonTransport<Body, Frame> => ({
|
|
id: "http-json",
|
|
with: (patch) => httpJson({ ...input, ...patch }),
|
|
prepare: (prepareInput) =>
|
|
Effect.gen(function* () {
|
|
const parts = yield* jsonRequestParts({ ...prepareInput })
|
|
const request = ProviderShared.jsonPost({
|
|
url: parts.url,
|
|
body: parts.bodyText,
|
|
headers: parts.headers,
|
|
})
|
|
return {
|
|
request,
|
|
web: new Request(parts.url, { method: "POST", headers: parts.headers, body: parts.bodyText }),
|
|
framing: input.framing,
|
|
middleware: prepareInput.middleware,
|
|
}
|
|
}),
|
|
frames: (prepared, request, runtime) =>
|
|
Stream.unwrap(
|
|
Effect.gen(function* () {
|
|
const request = prepared.web
|
|
const execute = (input: Request) =>
|
|
Effect.tryPromise({
|
|
try: () => input.text(),
|
|
catch: (cause) => (cause instanceof Error ? cause : new Error(String(cause))),
|
|
}).pipe(
|
|
Effect.flatMap((body) =>
|
|
runtime.http.execute(
|
|
ProviderShared.jsonPost({
|
|
url: input.url,
|
|
body,
|
|
headers: Headers.fromInput(input.headers),
|
|
}),
|
|
),
|
|
),
|
|
Effect.flatMap((response) =>
|
|
Stream.toReadableStreamEffect(response.stream).pipe(
|
|
Effect.map(
|
|
(body) =>
|
|
new Response(body, {
|
|
status: response.status,
|
|
headers: response.headers,
|
|
}),
|
|
),
|
|
),
|
|
),
|
|
)
|
|
return yield* prepared.middleware ? prepared.middleware(request, execute) : execute(request)
|
|
})
|
|
.pipe(
|
|
Effect.mapError((error) =>
|
|
error instanceof LLMError
|
|
? error
|
|
: ProviderShared.eventError(
|
|
`${request.model.provider}/${request.model.route.id}`,
|
|
`Failed to execute ${request.model.provider}/${request.model.route.id} request`,
|
|
ProviderShared.errorText(error),
|
|
),
|
|
),
|
|
Effect.map((response) => HttpClientResponse.fromWeb(prepared.request, response)),
|
|
)
|
|
.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: <Body>() => httpJson<Body, string>({ framing: Framing.sse }),
|
|
} as const
|