fix(aws): stabilization batch for AWS Lambda + generic adapters

docs/1.guide/9.aws-lambda.md

@@ -74,36 +74,40 @@ The function handles all conversions internally:
 
 AWS Lambda supports [response streaming](https://docs.aws.amazon.com/lambda/latest/dg/configuration-response-streaming.html) which allows you to send response data progressively as it becomes available. This improves Time To First Byte (TTFB) and enables streaming large responses (up to 20MB vs 6MB buffered limit).
 
-To use response streaming, wrap your handler with `awslambda.streamifyResponse()` and use `handleLambdaEventWithStream`:
+To use response streaming, build a streaming handler with `toLambdaStreamHandler` and wrap it with `awslambda.streamifyResponse()`. Just like `toLambdaHandler`, it accepts a full `ServerOptions` object, so `middleware`, `plugins`, `error`, and `trustProxy` all apply to the streaming path:
 
 ```ts
-import { handleLambdaEventWithStream, type AWSLambdaStreamingHandler } from "srvx/aws-lambda";
-
-const fetchHandler = async (request: Request) => {
-  // Create a streaming response
-  const stream = new ReadableStream({
-    async start(controller) {
-      controller.enqueue(new TextEncoder().encode("Hello, "));
-      await new Promise((r) => setTimeout(r, 100));
-      controller.enqueue(new TextEncoder().encode("streaming "));
-      await new Promise((r) => setTimeout(r, 100));
-      controller.enqueue(new TextEncoder().encode("world!"));
-      controller.close();
-    },
-  });
-
-  return new Response(stream, {
-    headers: { "Content-Type": "text/plain" },
-  });
-};
+import { toLambdaStreamHandler, type AWSLambdaStreamingHandler } from "srvx/aws-lambda";
+import { serveStatic } from "srvx/static";
 
 // Export a streaming handler
 export const handler: AWSLambdaStreamingHandler = awslambda.streamifyResponse(
-  (event, responseStream, context) =>
-    handleLambdaEventWithStream(fetchHandler, event, responseStream, context),
+  toLambdaStreamHandler({
+    middleware: [serveStatic({ dir: "public" })],
+    fetch(req: Request) {
+      // Create a streaming response
+      const stream = new ReadableStream({
+        async start(controller) {
+          controller.enqueue(new TextEncoder().encode("Hello, "));
+          await new Promise((r) => setTimeout(r, 100));
+          controller.enqueue(new TextEncoder().encode("streaming "));
+          await new Promise((r) => setTimeout(r, 100));
+          controller.enqueue(new TextEncoder().encode("world!"));
+          controller.close();
+        },
+      });
+
+      return new Response(stream, {
+        headers: { "Content-Type": "text/plain" },
+      });
+    },
+  }),
 );
 ```
 
+> [!TIP]
+> The lower-level `handleLambdaEventWithStream(fetchHandler, event, responseStream, context)` is still available if you want to wire a bare fetch handler yourself, but it bypasses `middleware`/`plugins`/`error`/`trustProxy`. Prefer `toLambdaStreamHandler` unless you have a specific reason not to.
+
 > [!NOTE]
 > Response streaming requires a Lambda Function URL with `--invoke-mode RESPONSE_STREAM`, or an API Gateway **REST API (v1)** integration configured with `responseTransferMode: STREAM` (both use the `InvokeWithResponseStream` API under the hood, and srvx's request/response handling is identical for both). HTTP API (v2) and Application Load Balancer do not support progressive streaming yet.
 >

src/adapters/_aws/utils.ts

@@ -34,15 +34,27 @@ export function awsRequest(
   context: AWSContext,
   trustProxy?: TrustProxyOption,
 ): ServerRequest {
+  // Real API Gateway always sends a `headers` object; a null/non-object here
+  // means a malformed/hand-built event and would otherwise surface as an opaque
+  // `TypeError` deep inside header parsing.
+  if (!event.headers || typeof event.headers !== "object") {
+    throw new TypeError("[srvx] Invalid AWS Lambda event: `headers` must be an object.");
+  }
+
   // Resolve the immediate-peer address and trust decision once and pass them
   // down; both the URL and the client IP derivation need them.
   const sourceIp = awsEventIP(event);
   const trusted = isTrustedProxy(trustProxy, sourceIp);
 
+  // Per the fetch spec a GET/HEAD request cannot carry a body; passing one to
+  // `new Request` throws. Raw bytes stay reachable via `runtime.awsLambda.event`.
+  const method = awsEventMethod(event);
+  const hasBody = method !== "GET" && method !== "HEAD";
+
   const req = new Request(awsEventURL(event, trusted), {
-    method: awsEventMethod(event),
+    method,
     headers: awsEventHeaders(event),
-    body: awsEventBody(event),
+    body: hasBody ? awsEventBody(event) : undefined,
   }) as ServerRequest;
 
   req.runtime = {
@@ -129,11 +141,31 @@ function awsEventQuery(event: APIGatewayProxyEvent | APIGatewayProxyEventV2) {
 
 function awsEventHeaders(event: APIGatewayProxyEvent | APIGatewayProxyEventV2): Headers {
   const headers = new Headers();
+
+  // v1 (REST API) events carry repeated headers in `multiValueHeaders`; the
+  // single-valued `headers` map only keeps the last value for each key. Prefer
+  // the multi-value form and skip those keys in the single map to avoid
+  // duplicating a header that appears in both.
+  const multiValueHeaders = (event as APIGatewayProxyEvent).multiValueHeaders;
+  const covered = new Set<string>();
+  if (multiValueHeaders) {
+    for (const [key, values] of Object.entries(multiValueHeaders)) {
+      if (!values) continue;
+      covered.add(key.toLowerCase());
+      for (const value of values) {
+        if (value != null) {
+          headers.append(key, value);
+        }
+      }
+    }
+  }
+
   for (const [key, value] of Object.entries(event.headers)) {
-    if (value) {
+    if (value && !covered.has(key.toLowerCase())) {
       headers.set(key, value);
     }
   }
+
   if ("cookies" in event && event.cookies) {
     for (const cookie of event.cookies) {
       headers.append("cookie", cookie);
@@ -160,15 +192,19 @@ export function awsResponseHeaders(
   response: Response,
   event?: APIGatewayProxyEvent | APIGatewayProxyEventV2,
 ): AWSResponseHeaders {
+  const cookies = response.headers.getSetCookie();
+
   const headers = Object.create(null);
   for (const [key, value] of response.headers) {
+    // `set-cookie` is delivered via `cookies` (v2) / `multiValueHeaders` (v1).
+    // Emitting it here too makes API Gateway merge both and send the last
+    // cookie a second time.
+    if (key === "set-cookie") continue;
     if (value) {
-      headers[key] = Array.isArray(value) ? value.join(",") : String(value);
+      headers[key] = value;
     }
   }
 
-  const cookies = response.headers.getSetCookie();
-
   if (cookies.length === 0) {
     return { headers };
   }
@@ -193,7 +229,11 @@ export async function awsResponseBody(
   }
   const buffer = await toBuffer(response.body as any);
   const contentType = response.headers.get("content-type") || "";
-  return isTextType(contentType)
+  // A compressed body (e.g. `content-encoding: gzip`) is binary regardless of
+  // its content-type; running it through `toString("utf8")` mangles the bytes.
+  const contentEncoding = (response.headers.get("content-encoding") || "").trim().toLowerCase();
+  const isEncoded = contentEncoding !== "" && contentEncoding !== "identity";
+  return !isEncoded && isTextType(contentType)
     ? { body: buffer.toString("utf8") }
     : { body: buffer.toString("base64"), isBase64Encoded: true };
 }
@@ -299,12 +339,18 @@ export async function requestToAwsEvent(request: Request): Promise<AwsLambdaEven
   const url = new URL(request.url);
 
   const headers: Record<string, string> = {};
+  const multiValueHeaders: Record<string, string[]> = {};
   const cookies: string[] = [];
   for (const [key, value] of request.headers) {
     if (key.toLowerCase() === "cookie") {
+      // Real v2 API Gateway events strip `cookie` from `headers` and carry it in
+      // `cookies`; keeping it in the header maps too would double it once
+      // `awsEventHeaders` re-appends `event.cookies` on the round trip.
       cookies.push(value);
+      continue;
     }
     headers[key] = value;
+    (multiValueHeaders[key] ??= []).push(value);
   }
 
   let body: string | undefined;
@@ -332,7 +378,7 @@ export async function requestToAwsEvent(request: Request): Promise<AwsLambdaEven
     multiValueQueryStringParameters: parseMultiValueQuery(url.searchParams),
     pathParameters: undefined,
     stageVariables: undefined,
-    multiValueHeaders: Object.fromEntries([...request.headers].map(([k, v]) => [k, [v]])),
+    multiValueHeaders,
 
     // v2 (HTTP API) fields
     version: "2.0",
@@ -457,7 +503,13 @@ export function awsResultToResponse(result: AwsLambdaResult): Response {
 
   const statusCode = typeof result.statusCode === "number" ? result.statusCode : 200;
 
-  return new Response(body, {
+  // `new Response(body, ...)` throws for null-body statuses when `body` is a
+  // (even empty) string, which broke the documented local-testing round trip
+  // for any 204/304 handler.
+  const nullBody =
+    statusCode === 101 || statusCode === 204 || statusCode === 205 || statusCode === 304;
+
+  return new Response(nullBody ? null : body, {
     status: statusCode,
     headers,
   });

src/adapters/aws-lambda.ts

@@ -3,6 +3,7 @@ import type { FetchHandler, Server, ServerOptions } from "../types.ts";
 import type { TrustProxyOption } from "../_trust-proxy.ts";
 import { wrapFetch } from "../_middleware.ts";
 import { errorPlugin } from "../_plugins.ts";
+import { createWaitUntil } from "../_utils.ts";
 import {
   awsRequest,
   awsResponseBody,
@@ -36,19 +37,39 @@ export function toLambdaHandler(options: ServerOptions): AWSLambdaHandler {
   return (event, context) => server.fetch(event, context);
 }
 
+/**
+ * Streaming counterpart to {@link toLambdaHandler}.
+ *
+ * The returned handler goes through the same server contract as the buffered
+ * one (`plugins`, `middleware`, `error`, `trustProxy` all apply) and is meant to
+ * be wrapped with `awslambda.streamifyResponse(...)`.
+ */
+export function toLambdaStreamHandler(options: ServerOptions): AWSLambdaStreamingHandler {
+  const server = new AWSLambdaServer(options);
+  return (event, responseStream, context) => server.fetchStream(event, responseStream, context);
+}
+
 export async function handleLambdaEvent(
   fetchHandler: FetchHandler,
   event: AwsLambdaEvent,
   context: AWS.Context,
   trustProxy?: TrustProxyOption,
 ): Promise<AWS.APIGatewayProxyResult | AWS.APIGatewayProxyResultV2> {
+  const wait = createWaitUntil();
   const request = awsRequest(event, context, trustProxy);
-  const response = await fetchHandler(request);
-  return {
-    statusCode: response.status,
-    ...awsResponseHeaders(response, event),
-    ...(await awsResponseBody(response)),
-  };
+  Object.defineProperty(request, "waitUntil", { value: wait.waitUntil, configurable: true });
+  try {
+    const response = await fetchHandler(request);
+    return {
+      statusCode: response.status,
+      ...awsResponseHeaders(response, event),
+      ...(await awsResponseBody(response)),
+    };
+  } finally {
+    // Await background tasks registered via `request.waitUntil` before the
+    // invocation returns; the process would otherwise be frozen mid-flight.
+    await wait.wait();
+  }
 }
 
 export async function handleLambdaEventWithStream(
@@ -58,9 +79,15 @@ export async function handleLambdaEventWithStream(
   context: AWS.Context,
   trustProxy?: TrustProxyOption,
 ): Promise<void> {
+  const wait = createWaitUntil();
   const request = awsRequest(event, context, trustProxy);
-  const response = await fetchHandler(request);
-  await awsStreamResponse(response, responseStream, event);
+  Object.defineProperty(request, "waitUntil", { value: wait.waitUntil, configurable: true });
+  try {
+    const response = await fetchHandler(request);
+    await awsStreamResponse(response, responseStream, event);
+  } finally {
+    await wait.wait();
+  }
 }
 
 export async function invokeLambdaHandler(
@@ -76,6 +103,7 @@ class AWSLambdaServer implements Server<AWSLambdaHandler> {
   readonly runtime = "aws-lambda";
   readonly options: Server["options"];
   readonly fetch: AWSLambdaHandler;
+  readonly fetchStream: AWSLambdaStreamingHandler;
 
   constructor(options: ServerOptions) {
     this.options = { ...options, middleware: [...(options.middleware || [])] };
@@ -87,6 +115,19 @@ class AWSLambdaServer implements Server<AWSLambdaHandler> {
 
     this.fetch = (event: AwsLambdaEvent, context: AWS.Context) =>
       handleLambdaEvent(fetchHandler, event, context, this.options.trustProxy);
+
+    this.fetchStream = (
+      event: AwsLambdaEvent,
+      responseStream: AWSLambdaResponseStream,
+      context: AWS.Context,
+    ) =>
+      handleLambdaEventWithStream(
+        fetchHandler,
+        event,
+        responseStream,
+        context,
+        this.options.trustProxy,
+      );
   }
 
   serve() {}

src/adapters/generic.ts

@@ -32,6 +32,17 @@ class GenericServer implements Server {
     this.fetch = (request: Request) => {
       Object.defineProperties(request, {
         waitUntil: { value: this.#wait.waitUntil },
+        runtime: { enumerable: true, value: { name: "generic" } },
+        ip: {
+          enumerable: true,
+          // The generic adapter has no native transport to read a peer address
+          // from, so `ip` is derived from `x-forwarded-for` when present.
+          // Configurable so a trustProxy setup can still override it.
+          configurable: true,
+          get() {
+            return request.headers.get("x-forwarded-for")?.split(",")[0]?.trim() || undefined;
+          },
+        },
       });
       return Promise.resolve(fetchHandler(request));
     };