From f3eccc37910f226617671a49e3283a26910e4795 Mon Sep 17 00:00:00 2001 From: Leo Conforti Date: Mon, 20 Jul 2026 14:42:47 -0500 Subject: [PATCH 1/2] Support HttpApiSchema.StreamUint8Array request payloads in HttpApiClient --- .../httpapi-client-stream-request-payloads.md | 7 ++++ .../src/unstable/httpapi/HttpApiClient.ts | 15 ++++++++ .../src/unstable/httpapi/HttpApiEndpoint.ts | 6 ++++ .../unstable/httpapi/HttpApiClient.test.ts | 35 +++++++++++++++++++ 4 files changed, 63 insertions(+) create mode 100644 .changeset/httpapi-client-stream-request-payloads.md 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..23e9f434f13 100644 --- a/packages/effect/src/unstable/httpapi/HttpApiClient.ts +++ b/packages/effect/src/unstable/httpapi/HttpApiClient.ts @@ -1000,6 +1000,21 @@ function getEncodePayloadSchemaFromBody( if (cached !== undefined) { return cached } + if (HttpApiSchema.isStreamUint8Array(schema)) { + 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)) + } + }) + )) + bodyFromPayloadCache.set(ast, out) + return out + } const encoding = HttpApiSchema.getPayloadEncoding(ast, method) const out = $HttpBody.pipe(Schema.decodeTo( schema, 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..3cb8dfa98ae 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,31 @@ 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") + })) + }) + describe("error responses", () => { const makeClient = (response: () => Response) => HttpApiClient.makeWith(ErrorContentTypeApi, { @@ -651,6 +677,15 @@ 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) + }) + ) +) + const clientFromResponse = (response: () => Response): HttpClient.HttpClient => HttpClient.make((request): Effect.Effect => Effect.succeed(HttpClientResponse.fromWeb(request, response())) From e3a60c19cc549744204fe66ccaa78b94a25150e6 Mon Sep 17 00:00:00 2001 From: Leo Conforti Date: Tue, 28 Jul 2026 11:43:35 -0500 Subject: [PATCH 2/2] Cache stream payload encoders by schema identity, not shared AST StreamUint8Array schemas all share a single AST while carrying their content type on the schema object, so caching encoders by AST let the first stream endpoint's content type leak into every other stream endpoint. --- .../src/unstable/httpapi/HttpApiClient.ts | 20 ++++++++++----- .../unstable/httpapi/HttpApiClient.test.ts | 25 +++++++++++++++++++ 2 files changed, 39 insertions(+), 6 deletions(-) diff --git a/packages/effect/src/unstable/httpapi/HttpApiClient.ts b/packages/effect/src/unstable/httpapi/HttpApiClient.ts index 23e9f434f13..81155acf91d 100644 --- a/packages/effect/src/unstable/httpapi/HttpApiClient.ts +++ b/packages/effect/src/unstable/httpapi/HttpApiClient.ts @@ -991,16 +991,19 @@ 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 { - const ast = schema.ast - const cached = bodyFromPayloadCache.get(ast) - if (cached !== undefined) { - return cached - } 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>({ @@ -1012,9 +1015,14 @@ function getEncodePayloadSchemaFromBody( } }) )) - bodyFromPayloadCache.set(ast, out) + streamBodyFromPayloadCache.set(schema, out) return out } + const ast = schema.ast + const cached = bodyFromPayloadCache.get(ast) + if (cached !== undefined) { + return cached + } const encoding = HttpApiSchema.getPayloadEncoding(ast, method) const out = $HttpBody.pipe(Schema.decodeTo( schema, diff --git a/packages/effect/test/unstable/httpapi/HttpApiClient.test.ts b/packages/effect/test/unstable/httpapi/HttpApiClient.test.ts index 3cb8dfa98ae..b92317b4cc7 100644 --- a/packages/effect/test/unstable/httpapi/HttpApiClient.test.ts +++ b/packages/effect/test/unstable/httpapi/HttpApiClient.test.ts @@ -233,6 +233,26 @@ describe("HttpApiClient", () => { 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", () => { @@ -683,6 +703,11 @@ const UploadApi = HttpApi.make("UploadApi").add( payload: HttpApiSchema.StreamUint8Array(), success: HttpApiSchema.Empty(200) }) + ).add( + HttpApiEndpoint.post("uploadCustom", "/uploadCustom", { + payload: HttpApiSchema.StreamUint8Array({ contentType: "application/vnd.custom" }), + success: HttpApiSchema.Empty(200) + }) ) )