Add HTTP response compression

README.md

@@ -128,6 +128,7 @@ least one seed per datacenter.
 | Socket pool tuning | Yes | No |
 | TLS session cache tuning | Yes | No |
 | Gzip request compression | Yes | Only with `CompressionStream` |
+| Gzip/deflate response compression | Yes | Only with `DecompressionStream` |
 
 Unsupported edge combinations throw at construction time with clear errors.
 
@@ -154,6 +155,11 @@ new AlternatorDynamoDBClient({
     gzipLevel: -1,
   },
 
+  responseCompression: {
+    enabled: true,
+    encodings: [ResponseCompressionGzip],
+  },
+
   headerOptimization: {
     enabled: true,
     allowedHeaders: ["Host", "X-Amz-Target", "Content-Length", "Accept-Encoding", "Content-Encoding"],
@@ -229,6 +235,28 @@ Request compression is disabled by default. When enabled, it compresses every
 request body with gzip. Use `gzipLevel` to select the zlib level, or provide
 `compressor` for a custom compressor.
 
+Response compression is disabled by default. Enable it with
+`responseCompression: true` to accept both supported encodings, or pass an
+options object with an explicit list:
+
+```ts
+import {
+  ResponseCompressionDeflate,
+  ResponseCompressionGzip,
+} from "@scylladb/alternator-client";
+
+new AlternatorDynamoDBClient({
+  seeds: ["scylla-0.internal"],
+  responseCompression: {
+    enabled: true,
+    encodings: [ResponseCompressionGzip, ResponseCompressionDeflate],
+  },
+});
+```
+
+When enabled, the client sends `Accept-Encoding` and transparently decodes
+`gzip` and `deflate` response bodies before the AWS SDK deserializes them.
+
 Key-route affinity supports these modes:
 
 ```ts

src/client.ts

@@ -118,6 +118,7 @@ function buildDynamoConfig(
     routing: _routing,
     runtime: _runtime,
     compression: _compression,
+    responseCompression: _responseCompression,
     headerOptimization: _headerOptimization,
     userAgent: _userAgent,
     keyRouteAffinity: _keyRouteAffinity,

src/compression.ts

@@ -1,5 +1,11 @@
+import { HttpResponse } from "@smithy/protocol-http";
 import { bodyToBytes } from "./body.js";
-import type { AlternatorRuntime, NormalizedCompressionOptions } from "./types.js";
+import { ResponseCompressionDeflate, ResponseCompressionGzip } from "./types.js";
+import type {
+  AlternatorResponseCompression,
+  AlternatorRuntime,
+  NormalizedCompressionOptions,
+} from "./types.js";
 
 export async function compressBody(
   body: unknown,
@@ -44,3 +50,245 @@ export async function compressBody(
     contentLength: compressed.byteLength,
   };
 }
+
+export function applyResponseCompressionRequestHeaders(
+  headers: Record<string, string | undefined>,
+  encodings: readonly AlternatorResponseCompression[],
+): Record<string, string> {
+  const acceptEncoding = responseCompressionAcceptEncoding(encodings);
+  if (acceptEncoding === "") {
+    return copyDefinedHeaders(headers);
+  }
+
+  const currentAcceptEncoding = getHeader(headers, "accept-encoding")?.trim();
+  if (currentAcceptEncoding && currentAcceptEncoding.toLowerCase() !== "identity") {
+    return copyDefinedHeaders(headers);
+  }
+
+  return {
+    ...removeHeaders(headers, ["accept-encoding"]),
+    "accept-encoding": acceptEncoding,
+  };
+}
+
+export function responseCompressionAcceptEncoding(
+  encodings: readonly AlternatorResponseCompression[],
+): string {
+  const seen = new Set<AlternatorResponseCompression>();
+  const parts: AlternatorResponseCompression[] = [];
+
+  for (const encoding of encodings) {
+    if (seen.has(encoding)) {
+      continue;
+    }
+    seen.add(encoding);
+    parts.push(encoding);
+  }
+
+  return parts.join(", ");
+}
+
+export async function decompressResponse(
+  response: HttpResponse,
+  runtime: AlternatorRuntime,
+): Promise<HttpResponse> {
+  const encoding = responseCompressionContentEncoding(getHeader(response.headers, "content-encoding"));
+  const body: unknown = response.body;
+  if (!encoding || body === undefined || body === null) {
+    return response;
+  }
+
+  const decodedBody = await decompressResponseBody(body, runtime, encoding);
+  return new HttpResponse({
+    statusCode: response.statusCode,
+    ...(response.reason !== undefined ? { reason: response.reason } : {}),
+    headers: removeHeaders(response.headers, ["content-encoding", "content-length"]),
+    body: decodedBody,
+  });
+}
+
+async function decompressResponseBody(
+  body: unknown,
+  runtime: AlternatorRuntime,
+  encoding: AlternatorResponseCompression,
+): Promise<unknown> {
+  if (runtime === "edge") {
+    return decompressWebResponseBody(body, encoding);
+  }
+  return decompressNodeResponseBody(body, encoding);
+}
+
+async function decompressNodeResponseBody(
+  body: unknown,
+  encoding: AlternatorResponseCompression,
+): Promise<unknown> {
+  const zlib = await import("node:zlib");
+
+  if (isNodePipeableBody(body)) {
+    const decoder = encoding === ResponseCompressionGzip
+      ? zlib.createGunzip()
+      : zlib.createInflate();
+    return body.pipe(decoder);
+  }
+
+  const bytes = await bodyToAsyncBytes(body);
+  return encoding === ResponseCompressionGzip
+    ? zlib.gunzipSync(bytes)
+    : zlib.inflateSync(bytes);
+}
+
+async function decompressWebResponseBody(
+  body: unknown,
+  encoding: AlternatorResponseCompression,
+): Promise<unknown> {
+  if (typeof DecompressionStream === "undefined") {
+    throw new Error("response compression requires DecompressionStream support in edge runtime");
+  }
+
+  const stream = await bodyToReadableStream(body);
+  return stream.pipeThrough(new DecompressionStream(encoding));
+}
+
+async function bodyToReadableStream(body: unknown): Promise<ReadableStream> {
+  if (typeof ReadableStream !== "undefined" && body instanceof ReadableStream) {
+    return body;
+  }
+  if (typeof Blob !== "undefined" && body instanceof Blob) {
+    return body.stream();
+  }
+
+  const bytes = await bodyToAsyncBytes(body);
+  return new Blob([bytesToArrayBuffer(bytes)]).stream();
+}
+
+async function bodyToAsyncBytes(body: unknown): Promise<Uint8Array> {
+  const bytes = bodyToBytes(body);
+  if (bytes) {
+    return bytes;
+  }
+  if (typeof Blob !== "undefined" && body instanceof Blob) {
+    return new Uint8Array(await body.arrayBuffer());
+  }
+  if (typeof ReadableStream !== "undefined" && body instanceof ReadableStream) {
+    return new Uint8Array(await new Response(body).arrayBuffer());
+  }
+  if (isTransformableByteBody(body)) {
+    return body.transformToByteArray();
+  }
+  if (isAsyncIterable(body)) {
+    const chunks: Uint8Array[] = [];
+    for await (const chunk of body) {
+      chunks.push(chunkToBytes(chunk));
+    }
+    return concatBytes(chunks);
+  }
+
+  throw new Error("compressed response body is not readable");
+}
+
+function responseCompressionContentEncoding(
+  value: string | undefined,
+): AlternatorResponseCompression | undefined {
+  switch (value?.trim().toLowerCase()) {
+    case ResponseCompressionGzip:
+      return ResponseCompressionGzip;
+    case ResponseCompressionDeflate:
+      return ResponseCompressionDeflate;
+    default:
+      return undefined;
+  }
+}
+
+function copyDefinedHeaders(headers: Record<string, string | undefined>): Record<string, string> {
+  const nextHeaders: Record<string, string> = {};
+  for (const [name, value] of Object.entries(headers)) {
+    if (value !== undefined) {
+      nextHeaders[name] = value;
+    }
+  }
+  return nextHeaders;
+}
+
+function removeHeaders(
+  headers: Record<string, string | undefined>,
+  names: readonly string[],
+): Record<string, string> {
+  const removed = new Set(names.map((name) => name.toLowerCase()));
+  const nextHeaders: Record<string, string> = {};
+
+  for (const [name, value] of Object.entries(headers)) {
+    if (value === undefined || removed.has(name.toLowerCase())) {
+      continue;
+    }
+    nextHeaders[name] = value;
+  }
+
+  return nextHeaders;
+}
+
+function getHeader(headers: Record<string, string | undefined>, name: string): string | undefined {
+  const lowerName = name.toLowerCase();
+  for (const [headerName, value] of Object.entries(headers)) {
+    if (headerName.toLowerCase() === lowerName) {
+      return value;
+    }
+  }
+  return undefined;
+}
+
+function chunkToBytes(chunk: unknown): Uint8Array {
+  const bytes = bodyToBytes(chunk);
+  if (bytes) {
+    return bytes;
+  }
+  return new TextEncoder().encode(String(chunk));
+}
+
+function concatBytes(chunks: readonly Uint8Array[]): Uint8Array {
+  const size = chunks.reduce((sum, chunk) => sum + chunk.byteLength, 0);
+  const bytes = new Uint8Array(size);
+  let offset = 0;
+  for (const chunk of chunks) {
+    bytes.set(chunk, offset);
+    offset += chunk.byteLength;
+  }
+  return bytes;
+}
+
+function bytesToArrayBuffer(bytes: Uint8Array): ArrayBuffer {
+  const copy = new Uint8Array(bytes.byteLength);
+  copy.set(bytes);
+  return copy.buffer;
+}
+
+function isNodePipeableBody(body: unknown): body is {
+  pipe(destination: NodeJS.WritableStream): NodeJS.ReadableStream;
+} {
+  return (
+    isObject(body) &&
+    "pipe" in body &&
+    typeof (body as { pipe?: unknown }).pipe === "function"
+  );
+}
+
+function isTransformableByteBody(body: unknown): body is {
+  transformToByteArray(): Promise<Uint8Array>;
+} {
+  return (
+    isObject(body) &&
+    "transformToByteArray" in body &&
+    typeof (body as { transformToByteArray?: unknown }).transformToByteArray === "function"
+  );
+}
+
+function isAsyncIterable(body: unknown): body is AsyncIterable<unknown> {
+  return (
+    isObject(body) &&
+    Symbol.asyncIterator in body &&
+    typeof (body as { [Symbol.asyncIterator]?: unknown })[Symbol.asyncIterator] === "function"
+  );
+}
+
+function isObject(value: unknown): value is object {
+  return value !== null && typeof value === "object";
+}