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()))