feat: add lambda http api streaming

.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.

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.

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<typeof HttpApp.toWebHandler>;
 const WebHandler = Context.GenericTag<WebHandler>("@effect-aws/lambda/WebHandler");
 
+type EffectStreamifyHandler<T, R, E = never> = (
+  event: T,
+  responseStream: awslambda.HttpResponseStream,
+  context: LambdaContext,
+) => Effect.Effect<void, E, R>;
+
+const requestFromEvent = (event: LambdaHandler.Event): Request => {
+  const eventSource = getEventSource(event) as EventSource<LambdaHandler.Event, LambdaHandler.Result>;
+  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<string, string> = {};
+  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<HttpApiOptions, "middleware">): Ef
 (event, context) =>
   Effect.gen(function*() {
     const eventSource = getEventSource(event) as EventSource<LambdaHandler.Event, LambdaHandler.Result>;
-    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<HttpApiOptions, "middleware">): Ef
       ? encodeBase64(yield* Effect.promise(() => res.arrayBuffer()))
       : yield* Effect.promise(() => res.text());
 
-    const headers: ResponseValues<LambdaHandler.Event>["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<LambdaHandler.Event>["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<HttpApiOptions, "middleware">): 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 = <LA, LE>(
   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 = <LA, LE>(
+  layer: Layer.Layer<LA | HttpApi.Api | HttpRouter.HttpRouter.DefaultServices, LE>,
+  options?: HttpApiOptions,
+): Handler<LambdaHandler.Event, void> => {
+  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);
+  });
+};

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<Buffer> = [];
+      let responseMetadata: {
+        readonly statusCode: number;
+        readonly headers: Record<string, string>;
+        readonly cookies: Array<string>;
+      } | undefined;
+      type TestStreamifyHandler = (
+        event: unknown,
+        responseStream: awslambda.HttpResponseStream,
+        context: LambdaContext,
+      ) => Promise<void>;
+
+      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<typeof responseMetadata>,
+      ) => {
+        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;
+      }
+    });
+  });
 });