Support HttpApiSchema.StreamUint8Array request payloads in HttpApiClient

.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<Uint8Array>`), 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.

packages/effect/src/unstable/httpapi/HttpApiClient.ts

@@ -991,10 +991,33 @@ function getEncodePayloadSchema(
 
 const bodyFromPayloadCache = new WeakMap<SchemaAST.AST, Schema.Top>()
 
+// 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<HttpApiSchema.StreamUint8Array, Schema.Top>()
+
 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<Stream.Stream<Uint8Array, unknown>, 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) {

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":

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<HttpBody.HttpBody> = []
+        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<HttpClientResponse.HttpClientResponse, never, never> =>
     Effect.succeed(HttpClientResponse.fromWeb(request, response()))