diff --git a/.changeset/httpapi-client-stream-request-payloads.md b/.changeset/httpapi-client-stream-request-payloads.md new file mode 100644 index 00000000000..2a92b81442c --- /dev/null +++ b/.changeset/httpapi-client-stream-request-payloads.md @@ -0,0 +1,7 @@ +--- +"effect": patch +--- + +Support `HttpApiSchema.StreamUint8Array` request payloads in `HttpApiClient` + +Previously, declaring an endpoint with `payload: HttpApiSchema.StreamUint8Array()` type-checked (the client accepted a `Stream`), but the payload encoder had no stream case and fell back to Json encoding, sending `JSON.stringify(stream)` — the literal body `null` — over the wire. Stream payload schemas are now preserved through endpoint construction and encoded as streamed request bodies with the schema's content type. diff --git a/packages/effect/src/unstable/httpapi/HttpApiClient.ts b/packages/effect/src/unstable/httpapi/HttpApiClient.ts index 2c3c08211df..81155acf91d 100644 --- a/packages/effect/src/unstable/httpapi/HttpApiClient.ts +++ b/packages/effect/src/unstable/httpapi/HttpApiClient.ts @@ -991,10 +991,33 @@ function getEncodePayloadSchema( const bodyFromPayloadCache = new WeakMap() +// Stream schemas share a single AST, with the content type carried on the +// schema object itself, so their encoders are cached by schema identity +const streamBodyFromPayloadCache = new WeakMap() + function getEncodePayloadSchemaFromBody( schema: Schema.Constraint, method: HttpMethod.HttpMethod ): Schema.Top { + if (HttpApiSchema.isStreamUint8Array(schema)) { + const cachedStream = streamBodyFromPayloadCache.get(schema) + if (cachedStream !== undefined) { + return cachedStream + } + const out = $HttpBody.pipe(Schema.decodeTo( + schema, + SchemaTransformation.transformOrFail, HttpBody.HttpBody>({ + decode(httpBody) { + return Effect.fail(new SchemaIssue.Forbidden(Option.some(httpBody), { message: "Encode only schema" })) + }, + encode(stream) { + return Effect.succeed(HttpBody.stream(stream, schema.contentType)) + } + }) + )) + streamBodyFromPayloadCache.set(schema, out) + return out + } const ast = schema.ast const cached = bodyFromPayloadCache.get(ast) if (cached !== undefined) { diff --git a/packages/effect/src/unstable/httpapi/HttpApiEndpoint.ts b/packages/effect/src/unstable/httpapi/HttpApiEndpoint.ts index d5b4a89e966..85a8a7a7270 100644 --- a/packages/effect/src/unstable/httpapi/HttpApiEndpoint.ts +++ b/packages/effect/src/unstable/httpapi/HttpApiEndpoint.ts @@ -1287,6 +1287,12 @@ function transformResponse(schema: Schema.Top): Schema.Top { } function transformPayload(schema: Schema.Top, method: HttpMethod): Schema.Top { + // Stream schemas carry their metadata on the schema object itself, so they + // must be preserved as-is for the client to detect them when encoding the + // request body + if (HttpApiSchema.isStreamSchema(schema)) { + return schema + } const encoding = HttpApiSchema.getPayloadEncoding(schema.ast, method) switch (encoding._tag) { case "Json": diff --git a/packages/effect/test/unstable/httpapi/HttpApiClient.test.ts b/packages/effect/test/unstable/httpapi/HttpApiClient.test.ts index fe5bcb8b7b2..b92317b4cc7 100644 --- a/packages/effect/test/unstable/httpapi/HttpApiClient.test.ts +++ b/packages/effect/test/unstable/httpapi/HttpApiClient.test.ts @@ -2,6 +2,7 @@ import { assert, describe, it } from "@effect/vitest" import { strictEqual } from "@effect/vitest/utils" import { Cause, Effect, Schema, Stream } from "effect" import { Sse } from "effect/unstable/encoding" +import type { HttpBody } from "effect/unstable/http" import { HttpClient, HttpClientError, HttpClientRequest, HttpClientResponse } from "effect/unstable/http" import { HttpApi, HttpApiClient, HttpApiEndpoint, HttpApiGroup, HttpApiSchema } from "effect/unstable/httpapi" @@ -209,6 +210,51 @@ describe("HttpApiClient", () => { })) }) + describe("streaming request payloads", () => { + it.effect("sends StreamUint8Array payloads as streamed bodies", () => + Effect.gen(function*() { + let captured: HttpBody.HttpBody | undefined + const client = yield* HttpApiClient.makeWith(UploadApi, { + baseUrl: "http://test", + httpClient: HttpClient.make((request) => { + captured = request.body + return Effect.succeed(HttpClientResponse.fromWeb(request, new Response(undefined, { status: 200 }))) + }) + }) + + yield* client.test.upload({ + payload: Stream.make(textEncoder.encode("hello "), textEncoder.encode("world")) + }) + + assert.strictEqual(captured?._tag, "Stream") + const body = captured as HttpBody.Stream + assert.strictEqual(body.contentType, "application/octet-stream") + const chunks = yield* Stream.runCollect(body.stream) + const textDecoder = new TextDecoder() + strictEqual(chunks.map((chunk) => textDecoder.decode(chunk, { stream: true })).join(""), "hello world") + })) + + it.effect("preserves each endpoint's stream content type", () => + Effect.gen(function*() { + const captured: Array = [] + const client = yield* HttpApiClient.makeWith(UploadApi, { + baseUrl: "http://test", + httpClient: HttpClient.make((request) => { + captured.push(request.body) + return Effect.succeed(HttpClientResponse.fromWeb(request, new Response(undefined, { status: 200 }))) + }) + }) + + yield* client.test.upload({ payload: Stream.make(textEncoder.encode("a")) }) + yield* client.test.uploadCustom({ payload: Stream.make(textEncoder.encode("b")) }) + + assert.strictEqual(captured[0]?._tag, "Stream") + assert.strictEqual((captured[0] as HttpBody.Stream).contentType, "application/octet-stream") + assert.strictEqual(captured[1]?._tag, "Stream") + assert.strictEqual((captured[1] as HttpBody.Stream).contentType, "application/vnd.custom") + })) + }) + describe("error responses", () => { const makeClient = (response: () => Response) => HttpApiClient.makeWith(ErrorContentTypeApi, { @@ -651,6 +697,20 @@ const ErrorContentTypeApi = HttpApi.make("ErrorContentTypeApi").add( ) ) +const UploadApi = HttpApi.make("UploadApi").add( + HttpApiGroup.make("test").add( + HttpApiEndpoint.post("upload", "/upload", { + payload: HttpApiSchema.StreamUint8Array(), + success: HttpApiSchema.Empty(200) + }) + ).add( + HttpApiEndpoint.post("uploadCustom", "/uploadCustom", { + payload: HttpApiSchema.StreamUint8Array({ contentType: "application/vnd.custom" }), + success: HttpApiSchema.Empty(200) + }) + ) +) + const clientFromResponse = (response: () => Response): HttpClient.HttpClient => HttpClient.make((request): Effect.Effect => Effect.succeed(HttpClientResponse.fromWeb(request, response()))