From 06644426bed9c9e664868f993f179cae63953da3 Mon Sep 17 00:00:00 2001 From: Yotam Ishak Date: Wed, 20 May 2026 13:44:53 +0300 Subject: [PATCH] feat: add lambda http api streaming --- .changeset/streaming-http-api-lambda.md | 5 + packages/lambda/README.md | 51 +++++++ packages/lambda/src/LambdaHandler.ts | 153 +++++++++++++++++---- packages/lambda/test/LambdaHandler.test.ts | 100 ++++++++++++++ 4 files changed, 281 insertions(+), 28 deletions(-) create mode 100644 .changeset/streaming-http-api-lambda.md diff --git a/.changeset/streaming-http-api-lambda.md b/.changeset/streaming-http-api-lambda.md new file mode 100644 index 00000000..4fac1353 --- /dev/null +++ b/.changeset/streaming-http-api-lambda.md @@ -0,0 +1,5 @@ +--- +"@effect-aws/lambda": minor +--- + +Add `LambdaHandler.streamFromHttpApi` for streaming `HttpApi` responses through Lambda Function URLs with response streaming enabled. diff --git a/packages/lambda/README.md b/packages/lambda/README.md index fe42f17f..eb97df60 100644 --- a/packages/lambda/README.md +++ b/packages/lambda/README.md @@ -110,3 +110,54 @@ export const handler: Handler = LambdaHandler.stream({ layer: S3.defaultLayer }) ``` + +HttpApi streaming with Lambda Function URLs: + +```ts +import { LambdaHandler } from "@effect-aws/lambda" +import { + HttpApi, + HttpApiBuilder, + HttpApiEndpoint, + HttpApiGroup, + HttpApiSchema, + HttpServer, + HttpServerResponse +} from "@effect/platform" +import { Effect, Layer, Stream } from "effect" + +const encoder = new TextEncoder() + +const streamEndpoint = HttpApiEndpoint.get("stream")`/stream`.addSuccess( + HttpApiSchema.Text() +) +const pingEndpoint = HttpApiEndpoint.get("ping")`/ping`.addSuccess( + HttpApiSchema.Text() +) + +const group = HttpApiGroup.make("api").add(streamEndpoint).add(pingEndpoint) +const api = HttpApi.make("Api").add(group) + +const routes = HttpApiBuilder.group(api, "api", (handlers) => + handlers + .handle("ping", () => Effect.succeed("pong\n")) + .handleRaw("stream", () => + Effect.succeed( + HttpServerResponse.stream( + Stream.fromIterable(["chunk 1\n", "chunk 2\n", "done\n"]).pipe( + Stream.map((chunk) => encoder.encode(chunk)) + ), + { contentType: "text/plain; charset=utf-8" } + ) + ) + ) +) + +const apiLive = HttpApiBuilder.api(api).pipe(Layer.provide(routes)) + +export const handler = LambdaHandler.streamFromHttpApi( + Layer.mergeAll(apiLive, HttpServer.layerContext) +) +``` + +Use `handle` for normal schema responses and `handleRaw` for streaming `HttpServerResponse` values. The Lambda must be invoked through a Lambda Function URL with response streaming enabled, for example `InvokeMode: RESPONSE_STREAM`. API Gateway HTTP API integrations do not support Lambda response streaming. diff --git a/packages/lambda/src/LambdaHandler.ts b/packages/lambda/src/LambdaHandler.ts index 33bc1739..f21374a2 100644 --- a/packages/lambda/src/LambdaHandler.ts +++ b/packages/lambda/src/LambdaHandler.ts @@ -1,6 +1,7 @@ /** * @since 1.4.0 */ +import type * as PlatformError from "@effect/platform/Error"; import type * as HttpApi from "@effect/platform/HttpApi"; import * as HttpApiBuilder from "@effect/platform/HttpApiBuilder"; import * as HttpApp from "@effect/platform/HttpApp"; @@ -10,9 +11,10 @@ import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Function from "effect/Function"; import * as Layer from "effect/Layer"; +import { Readable } from "node:stream"; import { getEventSource } from "./internal/index.js"; import * as internal from "./internal/lambdaHandler.js"; -import { pipeTo } from "./internal/stream.js"; +import { pipeline, pipeTo } from "./internal/stream.js"; import type { EventSource, ResponseValues } from "./internal/types.js"; import { encodeBase64, isContentEncodingBinary, isContentTypeBinary } from "./internal/utils.js"; import * as LambdaRuntime from "./LambdaRuntime.js"; @@ -227,6 +229,47 @@ interface HttpApiOptions { type WebHandler = ReturnType; const WebHandler = Context.GenericTag("@effect-aws/lambda/WebHandler"); +type EffectStreamifyHandler = ( + event: T, + responseStream: awslambda.HttpResponseStream, + context: LambdaContext, +) => Effect.Effect; + +const requestFromEvent = (event: LambdaHandler.Event): Request => { + const eventSource = getEventSource(event) as EventSource; + const requestValues = eventSource.getRequest(event); + + return new Request( + `https://${requestValues.remoteAddress}${requestValues.path}`, + { + method: requestValues.method, + headers: requestValues.headers, + body: requestValues.body, + }, + ); +}; + +const responseHeaders = (res: Response) => { + const headers: Record = {}; + const cookies = res.headers.has("set-cookie") + ? res.headers.getSetCookie + ? res.headers.getSetCookie() + : Array.from((res.headers as any).entries()) + .filter(([k]: any) => k === "set-cookie") + .map(([, v]: any) => v) + : []; + + if (Array.isArray(cookies) && cookies.length > 0) { + res.headers.delete("set-cookie"); + } + + res.headers.forEach((value, key) => { + headers[key] = value; + }); + + return { headers, cookies }; +}; + /** * Construct an `WebHandler` from an `HttpApi` instance. * @@ -268,16 +311,7 @@ export const httpApiHandler = (options?: Pick): Ef (event, context) => Effect.gen(function*() { const eventSource = getEventSource(event) as EventSource; - const requestValues = eventSource.getRequest(event); - - const req = new Request( - `https://${requestValues.remoteAddress}${requestValues.path}`, - { - method: requestValues.method, - headers: requestValues.headers, - body: requestValues.body, - }, - ); + const req = requestFromEvent(event); const res = yield* makeWebHandler(options).pipe( Effect.provideService(internal.lambdaEventTag, event), @@ -297,34 +331,57 @@ export const httpApiHandler = (options?: Pick): Ef ? encodeBase64(yield* Effect.promise(() => res.arrayBuffer())) : yield* Effect.promise(() => res.text()); - const headers: ResponseValues["headers"] = {}; - - if (res.headers.has("set-cookie")) { - const cookies = res.headers.getSetCookie - ? res.headers.getSetCookie() - : Array.from((res.headers as any).entries()) - .filter(([k]: any) => k === "set-cookie") - .map(([, v]: any) => v); + const { cookies, headers } = responseHeaders(res); + const responseHeadersMap: ResponseValues["headers"] = headers; - if (Array.isArray(cookies)) { - headers["set-cookie"] = cookies; - res.headers.delete("set-cookie"); - } + if (cookies.length > 0) { + responseHeadersMap["set-cookie"] = cookies; } - res.headers.forEach((value, key) => { - headers[key] = value; - }); - return eventSource.getResponse({ event, statusCode: res.status, body, - headers, + headers: responseHeadersMap, isBase64Encoded, }); }); +/** + * Construct a streaming `StreamifyHandler` from an `HttpApi` instance for Lambda Function URLs. + * + * @since 1.7.0 + * @category constructors + */ +export const httpApiStreamHandler = (options?: Pick): EffectStreamifyHandler< + LambdaHandler.Event, + HttpApi.Api | HttpApiBuilder.Router | HttpRouter.HttpRouter.DefaultServices | HttpApiBuilder.Middleware, + Cause.UnknownException | PlatformError.PlatformError +> => +(event, responseStream, context) => + Effect.gen(function*() { + const req = requestFromEvent(event); + + const res = yield* makeWebHandler(options).pipe( + Effect.provideService(internal.lambdaEventTag, event), + Effect.provideService(internal.lambdaContextTag, context), + Effect.andThen((handler) => handler(req)), + ); + + const { cookies, headers } = responseHeaders(res); + const httpResponseStream = global.awslambda.HttpResponseStream.from(responseStream, { + statusCode: res.status, + headers, + cookies, + }); + + if (res.body) { + yield* pipeline(Readable.fromWeb(res.body as any), httpResponseStream); + } else { + httpResponseStream.end(); + } + }); + /** * Construct a lambda handler from an `HttpApi` instance. * @@ -358,3 +415,43 @@ export const fromHttpApi = ( const httpApiLayer = Layer.mergeAll(layer, HttpApiBuilder.Router.Live, HttpApiBuilder.Middleware.layer); return make({ handler: httpApiHandler(options), layer: httpApiLayer, memoMap: options?.memoMap }); }; + +/** + * Construct a streaming lambda handler from an `HttpApi` instance for Lambda Function URLs. + * + * Lambda response streaming is supported by Lambda Function URLs with response streaming enabled, + * not by API Gateway HTTP API integrations. + * + * @example + * ```ts + * import { LambdaHandler } from "@effect-aws/lambda" + * import { HttpApi, HttpApiBuilder, HttpServer } from "@effect/platform" + * import { Layer } from "effect" + * + * class MyApi extends HttpApi.make("api") {} + * + * const MyApiLive = HttpApiBuilder.api(MyApi) + * + * export const handler = LambdaHandler.streamFromHttpApi( + * Layer.mergeAll( + * MyApiLive, + * HttpServer.layerContext + * ) + * ) + * ``` + * + * @since 1.7.0 + * @category constructors + */ +export const streamFromHttpApi = ( + layer: Layer.Layer, + options?: HttpApiOptions, +): Handler => { + const httpApiLayer = Layer.mergeAll(layer, HttpApiBuilder.Router.Live, HttpApiBuilder.Middleware.layer); + const runtime = LambdaRuntime.fromLayer(httpApiLayer, { memoMap: options?.memoMap }); + + return global.awslambda?.streamifyResponse(async (event, responseStream, context) => { + context.callbackWaitsForEmptyEventLoop = false; + return httpApiStreamHandler(options)(event, responseStream, context).pipe(runtime.runPromise); + }); +}; diff --git a/packages/lambda/test/LambdaHandler.test.ts b/packages/lambda/test/LambdaHandler.test.ts index b173cbaf..60cbafcd 100644 --- a/packages/lambda/test/LambdaHandler.test.ts +++ b/packages/lambda/test/LambdaHandler.test.ts @@ -19,6 +19,7 @@ import { HttpServerResponse, } from "@effect/platform"; import { Context, Effect, Layer } from "effect"; +import { Writable } from "node:stream"; import { describe, expect, it, vi } from "vitest"; import { albEvent } from "./fixtures/alb-event.js"; import { apiGatewayV1Event } from "./fixtures/api-gateway-v1-event.js"; @@ -324,4 +325,103 @@ describe("LambdaHandler", () => { ); }); }); + + describe("LambdaHandler.streamFromHttpApi", () => { + it("should stream Lambda Function URL responses", async () => { + const originalAwslambda = global.awslambda; + const chunks: Array = []; + let responseMetadata: { + readonly statusCode: number; + readonly headers: Record; + readonly cookies: Array; + } | undefined; + type TestStreamifyHandler = ( + event: unknown, + responseStream: awslambda.HttpResponseStream, + context: LambdaContext, + ) => Promise; + + const streamifyResponse = vi.fn( + (streamHandler: TestStreamifyHandler) => async (event: unknown, context: LambdaContext) => { + const responseStream = new Writable({ + write(chunk: Buffer | string, _encoding: BufferEncoding, callback: (error?: Error | null) => void) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + callback(); + }, + }); + + await streamHandler(event, responseStream as awslambda.HttpResponseStream, context); + }, + ); + + const from = vi.fn(( + responseStream: awslambda.HttpResponseStream, + metadata: NonNullable, + ) => { + responseMetadata = metadata; + return responseStream; + }); + + (global as any).awslambda = { + streamifyResponse, + HttpResponseStream: { from }, + }; + + try { + const context = { + functionVersion: "$LATEST", + functionName: "demo-effect-app-dev-api", + invokedFunctionArn: "arn:aws:lambda:eu-fake-1:000000000000:function:demo-effect-app-dev-api", + awsRequestId: "8ad41330-f092-4037-bc7c-63ffb7d6d4e7", + } as LambdaContext; + + const hello = HttpApiEndpoint.post("hello")`/my/path`.addSuccess(HttpApiSchema.Text()); + + const quotesGroup = HttpApiGroup.make("hello").add(hello); + + const MyApi = HttpApi.make("MyApi").add(quotesGroup); + + const HelloLive = HttpApiBuilder.group( + MyApi, + "hello", + (handlers) => + handlers.handle( + "hello", + () => + HttpApp.appendPreResponseHandler((_req, response) => + Effect.orDie( + HttpServerResponse.setCookie(response, "cookie key", "cookie value"), + ) + ).pipe( + Effect.flatMap(() => Effect.succeed("Hello, World!")), + ), + ), + ); + + const MyApiLive = HttpApiBuilder.api(MyApi).pipe(Layer.provide(HelloLive)); + + const handler = LambdaHandler.streamFromHttpApi(Layer.mergeAll(MyApiLive, HttpServer.layerContext)); + + // Lambda Function URL events currently use the same payload format as APIGatewayProxyEventV2. + const result = await handler(apiGatewayV2Event, context); + + expect(result).toBeUndefined(); + expect(streamifyResponse).toHaveBeenCalledTimes(1); + expect(from).toHaveBeenCalledTimes(1); + expect(responseMetadata).toStrictEqual({ + statusCode: 200, + headers: { + "content-length": "13", + "content-type": "text/plain", + }, + cookies: [ + "cookie key=cookie%20value", + ], + }); + expect(Buffer.concat(chunks).toString()).toBe("Hello, World!"); + } finally { + (global as any).awslambda = originalAwslambda; + } + }); + }); });