Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions .changeset/httpapi-client-stream-request-payloads.md
Original file line number Diff line number Diff line change
@@ -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.
15 changes: 15 additions & 0 deletions packages/effect/src/unstable/httpapi/HttpApiClient.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1000,6 +1000,21 @@ function getEncodePayloadSchemaFromBody(
if (cached !== undefined) {
return cached
}
if (HttpApiSchema.isStreamUint8Array(schema)) {
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))
}
})
))
bodyFromPayloadCache.set(ast, out)
return out
Comment on lines +1003 to +1016

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

Do not cache stream encoders by the shared AST.

StreamUint8Array and StreamSse instances share streamSchema.ast, but their contentType and mode live on the schema object. Because the AST cache is checked before this branch, the first stream schema can determine the encoder for later requests, causing custom content types to be silently wrong or bypassing this branch entirely.

Use a cache keyed by schema identity, or skip bodyFromPayloadCache for stream schemas. Add a regression test with a custom content type.

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/effect/src/unstable/httpapi/HttpApiClient.ts` around lines 1003 -
1016, Update the stream-schema handling around HttpApiSchema.isStreamUint8Array
so StreamUint8Array and StreamSse encoders are not cached or retrieved by shared
AST identity. Key bodyFromPayloadCache by the schema object for stream schemas,
or bypass that cache for them, ensuring each schema preserves its own
contentType and mode; add a regression test covering distinct custom content
types.

}
const encoding = HttpApiSchema.getPayloadEncoding(ast, method)
const out = $HttpBody.pipe(Schema.decodeTo(
schema,
Expand Down
6 changes: 6 additions & 0 deletions packages/effect/src/unstable/httpapi/HttpApiEndpoint.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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":
Expand Down
35 changes: 35 additions & 0 deletions packages/effect/test/unstable/httpapi/HttpApiClient.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down Expand Up @@ -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, {
Expand Down Expand Up @@ -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<HttpClientResponse.HttpClientResponse, never, never> =>
Effect.succeed(HttpClientResponse.fromWeb(request, response()))
Expand Down
Loading