Merge branch 'main' into beta

Author: floydspace

.changeset/tricky-apricots-wonder.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/CHANGELOG.md

@@ -30,6 +30,12 @@
 
 - [#198](https://github.com/floydspace/effect-aws/pull/198) [`57b06d0`](https://github.com/floydspace/effect-aws/commit/57b06d0474b35ae754e8e1b0a317e15669191779) Thanks [@floydspace](https://github.com/floydspace)! - Migrate to effect v4
 
+## 1.7.0
+
+### Minor Changes
+
+- [#218](https://github.com/floydspace/effect-aws/pull/218) [`1e12d67`](https://github.com/floydspace/effect-aws/commit/1e12d6795b51c3a85a526d80dd13343a7806dae9) Thanks [@yotamishak](https://github.com/yotamishak)! - Add `LambdaHandler.streamFromHttpApi` for streaming `HttpApi` responses through Lambda Function URLs with response streaming enabled.
+
 ## 1.6.0
 
 ### Minor Changes

packages/lambda/README.md

@@ -113,3 +113,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

@@ -5,12 +5,14 @@ import type * as Cause from "effect/Cause";
 import * as Context from "effect/Context";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
+import type * as PlatformError from "effect/PlatformError";
 import * as Predicate from "effect/Predicate";
 import type { HttpMiddleware } from "effect/unstable/http";
 import { HttpRouter } from "effect/unstable/http";
+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";
@@ -207,6 +209,47 @@ type WebHandler = (
 ) => Promise<globalThis.Response>;
 const WebHandler = Context.Service<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 `HttpRouter` instance.
  *
@@ -238,16 +281,7 @@ export const httpApiHandler: EffectHandler<
 > = (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 handler = yield* Effect.service(WebHandler);
 
@@ -270,34 +304,59 @@ export const httpApiHandler: EffectHandler<
       ? encodeBase64(yield* Effect.promise(() => res.arrayBuffer()))
       : yield* Effect.promise(() => res.text());
 
-    const headers: ResponseValues<LambdaHandler.Event>["headers"] = {};
+    const { cookies, headers } = responseHeaders(res);
+    const responseHeadersMap: ResponseValues<LambdaHandler.Event>["headers"] = 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);
-
-      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: EffectStreamifyHandler<
+  LambdaHandler.Event,
+  WebHandler,
+  Cause.UnknownError | PlatformError.PlatformError
+> = (event, responseStream, context) =>
+  Effect.gen(function*() {
+    const req = requestFromEvent(event);
+
+    const handler = yield* Effect.service(WebHandler);
+
+    const ctx = Context.empty().pipe(
+      Context.add(internal.lambdaEventTag, event),
+      Context.add(internal.lambdaContextTag, context),
+    );
+
+    const res: globalThis.Response = yield* Effect.tryPromise(() => handler(req, ctx));
+
+    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.
  *
@@ -327,3 +386,43 @@ export const fromHttpApi = <LA, LE>(
   const WebHandlerLive = Layer.effect(WebHandler, makeWebHandler(layer, options));
   return make({ handler: httpApiHandler, layer: WebHandlerLive, 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, LE, HttpRouter.HttpRouter>,
+  options?: HttpApiOptions,
+): Handler<LambdaHandler.Event, void> => {
+  const WebHandlerLive = Layer.effect(WebHandler, makeWebHandler(layer, options));
+  const runtime = LambdaRuntime.fromLayer(WebHandlerLive, { memoMap: options?.memoMap });
+
+  return global.awslambda?.streamifyResponse(async (event, responseStream, context) => {
+    context.callbackWaitsForEmptyEventLoop = false;
+    return httpApiStreamHandler(event, responseStream, context).pipe(runtime.runPromise);
+  });
+};

packages/lambda/test/LambdaHandler.test.ts

@@ -14,6 +14,7 @@ import * as Layer from "effect/Layer";
 import * as Schema from "effect/Schema";
 import { HttpEffect, HttpServer, HttpServerResponse } from "effect/unstable/http";
 import { HttpApi, HttpApiBuilder, HttpApiEndpoint, HttpApiGroup, HttpApiSchema } from "effect/unstable/httpapi";
+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";
@@ -339,4 +340,108 @@ 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`, {
+          success: Schema.String.pipe(HttpApiSchema.asText()),
+        });
+
+        const quotesGroup = HttpApiGroup.make("hello").add(hello);
+
+        const MyApi = HttpApi.make("MyApi").add(quotesGroup);
+
+        const HelloLive = HttpApiBuilder.group(
+          MyApi,
+          "hello",
+          (handlers) =>
+            handlers.handle(
+              "hello",
+              () =>
+                HttpEffect.appendPreResponseHandler((_req, response) =>
+                  Effect.orDie(
+                    HttpServerResponse.setCookie(response, "cookie key", "cookie value"),
+                  )
+                ).pipe(
+                  Effect.flatMap(() => Effect.succeed("Hello, World!")),
+                ),
+            ),
+        );
+
+        const MyApiLive = HttpApiBuilder.layer(MyApi).pipe(
+          Layer.provide(HelloLive),
+          Layer.provide(HttpServer.layerServices),
+        );
+
+        const handler = LambdaHandler.streamFromHttpApi(MyApiLive);
+
+        // 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;
+      }
+    });
+  });
 });