diff --git a/README.md b/README.md index 3ea3a85..1131517 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,22 @@ npm install @scylladb/alternator-client @aws-sdk/client-dynamodb For document commands, also install `@aws-sdk/lib-dynamodb`. +Node.js 20 or newer is required for the Node runtime. + +The package uses conditional exports. Node resolves the Node build; browser, +worker, and default ESM conditions resolve the Edge build, which contains no +Node built-in imports. To force the Edge entrypoint in a non-edge test harness: + +```ts +import { AlternatorDynamoDBClient } from "@scylladb/alternator-client/edge"; +``` + +For document commands in an Edge bundle, use: + +```ts +import { AlternatorDynamoDBDocumentClient } from "@scylladb/alternator-client/document/edge"; +``` + ## Low-Level Client ```ts @@ -145,7 +161,6 @@ new AlternatorDynamoDBClient({ scheme: "http", port: 8080, routing: routing.cluster(), - runtime: "node", logger: console, discovery: { diff --git a/eslint.config.js b/eslint.config.js index b8932ce..752882c 100644 --- a/eslint.config.js +++ b/eslint.config.js @@ -2,7 +2,7 @@ import js from "@eslint/js"; import globals from "globals"; import tseslint from "typescript-eslint"; -const tsFiles = ["src/**/*.ts", "test/**/*.ts", "vitest.config.ts"]; +const tsFiles = ["src/**/*.ts", "test/**/*.ts", "vitest.config.ts", "vitest.integration.config.ts", "tsup.config.ts"]; export default tseslint.config( { diff --git a/package-lock.json b/package-lock.json index 748d060..519ca0a 100644 --- a/package-lock.json +++ b/package-lock.json @@ -10,13 +10,13 @@ "license": "Apache-2.0", "dependencies": { "@aws-sdk/client-dynamodb": "^3.1075.0", - "@aws-sdk/lib-dynamodb": "^3.1075.0", "@smithy/fetch-http-handler": "^5.5.2", "@smithy/node-http-handler": "^4.8.2", "@smithy/protocol-http": "^5.5.2", "@smithy/types": "^4.15.0" }, "devDependencies": { + "@aws-sdk/lib-dynamodb": "^3.1075.0", "@eslint/js": "^10.0.1", "@types/node": "^24.0.0", "eslint": "^10.5.0", @@ -26,6 +26,9 @@ "typescript-eslint": "^8.62.0", "vitest": "^4.1.9" }, + "engines": { + "node": ">=20" + }, "peerDependencies": { "@aws-sdk/client-dynamodb": "^3.1075.0", "@aws-sdk/lib-dynamodb": "^3.1075.0" @@ -321,6 +324,7 @@ "version": "3.1075.0", "resolved": "https://registry.npmjs.org/@aws-sdk/lib-dynamodb/-/lib-dynamodb-3.1075.0.tgz", "integrity": "sha512-3OxmMDgk8wvK6QOaVrQuOq9t6xlgS5047aDPxAZxTSwGBcEMc6FtiamtoK3kfClfOt7k3t+NWIo0bIfhWQ4Pzg==", + "dev": true, "license": "Apache-2.0", "dependencies": { "@aws-sdk/core": "^3.974.23", @@ -422,6 +426,7 @@ "version": "3.996.5", "resolved": "https://registry.npmjs.org/@aws-sdk/util-dynamodb/-/util-dynamodb-3.996.5.tgz", "integrity": "sha512-m9bdmYq3WtbMHAKGALw9XWiMBfKu5T8ukgdJT7Mc/d2oOwDGNFmhsnnkQ18xomoXo/ZHxAuIDi3Y6slsblW1Mg==", + "dev": true, "license": "Apache-2.0", "dependencies": { "tslib": "^2.6.2" diff --git a/package.json b/package.json index 96fedc2..ef6de68 100644 --- a/package.json +++ b/package.json @@ -6,17 +6,51 @@ "type": "module", "main": "./dist/index.cjs", "module": "./dist/index.js", + "browser": "./dist/edge.js", "types": "./dist/index.d.ts", + "engines": { + "node": ">=20" + }, "exports": { ".": { "types": "./dist/index.d.ts", - "import": "./dist/index.js", - "require": "./dist/index.cjs" + "browser": { + "import": "./dist/edge.js" + }, + "worker": { + "import": "./dist/edge.js" + }, + "node": { + "import": "./dist/index.js", + "require": "./dist/index.cjs" + }, + "default": { + "import": "./dist/edge.js" + } + }, + "./edge": { + "types": "./dist/edge.d.ts", + "import": "./dist/edge.js" }, "./document": { "types": "./dist/document.d.ts", - "import": "./dist/document.js", - "require": "./dist/document.cjs" + "browser": { + "import": "./dist/document-edge.js" + }, + "worker": { + "import": "./dist/document-edge.js" + }, + "node": { + "import": "./dist/document.js", + "require": "./dist/document.cjs" + }, + "default": { + "import": "./dist/document-edge.js" + } + }, + "./document/edge": { + "types": "./dist/document-edge.d.ts", + "import": "./dist/document-edge.js" }, "./package.json": "./package.json" }, @@ -27,7 +61,7 @@ ], "sideEffects": false, "scripts": { - "build": "tsup src/index.ts src/document.ts --format esm,cjs --dts --sourcemap --clean", + "build": "tsup", "lint": "eslint . --max-warnings=0", "typecheck": "tsc --noEmit", "test": "vitest run", @@ -36,13 +70,13 @@ }, "dependencies": { "@aws-sdk/client-dynamodb": "^3.1075.0", - "@aws-sdk/lib-dynamodb": "^3.1075.0", "@smithy/fetch-http-handler": "^5.5.2", "@smithy/node-http-handler": "^4.8.2", "@smithy/protocol-http": "^5.5.2", "@smithy/types": "^4.15.0" }, "devDependencies": { + "@aws-sdk/lib-dynamodb": "^3.1075.0", "@eslint/js": "^10.0.1", "@types/node": "^24.0.0", "eslint": "^10.5.0", diff --git a/src/client-base.ts b/src/client-base.ts new file mode 100644 index 0000000..c958fe3 --- /dev/null +++ b/src/client-base.ts @@ -0,0 +1,149 @@ +import { + DescribeTableCommand, + DynamoDBClient, + type DynamoDBClientConfig, + type ServiceInputTypes, + type ServiceOutputTypes, +} from "@aws-sdk/client-dynamodb"; +import type { HttpHandler, HttpHandlerUserInput } from "@smithy/protocol-http"; +import type { HttpHandlerOptions } from "@smithy/types"; +import { DEFAULT_REGION, firstEndpointUrl, NO_AUTH_CREDENTIALS, normalizeConfig } from "./config.js"; +import { AlternatorDiscovery } from "./discovery.js"; +import { KeyRouteAffinityPlanner } from "./affinity.js"; +import { createAlternatorPostSigningMiddleware, createAlternatorRequestMiddleware } from "./middleware.js"; +import type { AlternatorBodyCompressor } from "./compression-types.js"; +import type { AlternatorDynamoDBClientConfig, AlternatorNode, NormalizedAlternatorConfig } from "./types.js"; + +export interface AlternatorDynamoDBClientApi { + nodes(): AlternatorNode[]; + refreshNodes(): Promise; + supportsScopedDiscovery(): Promise; + validateRouting(): Promise; + partitionKey(tableName: string): string | undefined; +} + +export interface AlternatorRuntimePlatform { + assertRuntimeSupport(config: NormalizedAlternatorConfig): void; + createRequestHandler( + input: AlternatorDynamoDBClientConfig, + config: NormalizedAlternatorConfig, + ): HttpHandlerUserInput; + compressBody: AlternatorBodyCompressor; +} + +export abstract class AlternatorDynamoDBClientBase extends DynamoDBClient { + readonly alternator: AlternatorDynamoDBClientApi; + private readonly alternatorConfig: NormalizedAlternatorConfig; + private readonly discovery: AlternatorDiscovery; + private readonly keyAffinity: KeyRouteAffinityPlanner; + + protected constructor(config: AlternatorDynamoDBClientConfig, platform: AlternatorRuntimePlatform) { + const alternatorConfig = normalizeConfig(config); + platform.assertRuntimeSupport(alternatorConfig); + + const requestHandler = platform.createRequestHandler(config, alternatorConfig); + const dynamoConfig = buildDynamoConfig(config, alternatorConfig, requestHandler); + + super(dynamoConfig); + + this.alternatorConfig = alternatorConfig; + this.discovery = new AlternatorDiscovery( + alternatorConfig, + this.config.requestHandler as HttpHandler, + ); + this.keyAffinity = new KeyRouteAffinityPlanner( + alternatorConfig.keyRouteAffinity, + (tableName) => this.discoverPartitionKey(tableName), + alternatorConfig.logger, + ); + this.alternator = { + nodes: () => this.discovery.getLiveNodes(), + refreshNodes: () => this.discovery.refreshLiveNodes(), + supportsScopedDiscovery: () => this.discovery.checkRackDatacenterSupport(), + validateRouting: () => this.discovery.checkIfRackAndDatacenterSetCorrectly(), + partitionKey: (tableName) => this.keyAffinity.getPartitionKeyName(tableName), + }; + + this.middlewareStack.addRelativeTo( + createAlternatorRequestMiddleware({ + discovery: this.discovery, + config: alternatorConfig, + keyAffinity: this.keyAffinity, + compressBody: platform.compressBody, + }), + { + relation: "before", + toMiddleware: "httpSigningMiddleware", + name: "alternatorRequestMiddleware", + override: true, + }, + ); + + this.middlewareStack.addRelativeTo( + createAlternatorPostSigningMiddleware(alternatorConfig), + { + relation: "after", + toMiddleware: "httpSigningMiddleware", + name: "alternatorPostSigningMiddleware", + override: true, + }, + ); + } + + override destroy(): void { + this.discovery.destroy(); + super.destroy(); + } + + private async discoverPartitionKey(tableName: string): Promise { + const response = await this.send(new DescribeTableCommand({ TableName: tableName })); + const keyName = response.Table?.KeySchema?.find((key) => key.KeyType === "HASH")?.AttributeName; + if (!keyName) { + this.alternatorConfig.logger.warn?.("alternator key affinity: DescribeTable returned no partition key", { + tableName, + }); + return; + } + this.keyAffinity.setPartitionKeyName(tableName, keyName); + } +} + +function buildDynamoConfig( + input: AlternatorDynamoDBClientConfig, + alternatorConfig: NormalizedAlternatorConfig, + requestHandler: HttpHandlerUserInput, +): DynamoDBClientConfig { + const { + seeds: _seeds, + scheme: _scheme, + port: _port, + routing: _routing, + runtime: _runtime, + compression: _compression, + headerOptimization: _headerOptimization, + userAgent: _userAgent, + keyRouteAffinity: _keyRouteAffinity, + tls: _tls, + discovery: _discovery, + connection: _connection, + logger: _logger, + credentials, + region, + ...awsConfigWithEndpoint + } = input as AlternatorDynamoDBClientConfig & { endpoint?: unknown }; + const { endpoint: _endpoint, ...awsConfig } = awsConfigWithEndpoint; + + const dynamoConfig: DynamoDBClientConfig & { applyChecksum?: boolean } = { + ...awsConfig, + endpoint: firstEndpointUrl(alternatorConfig), + region: region ?? DEFAULT_REGION, + credentials: credentials ?? NO_AUTH_CREDENTIALS, + requestHandler, + }; + if (alternatorConfig.headerOptimization.enabled) { + dynamoConfig.applyChecksum = false; + } + return dynamoConfig; +} + +export type AlternatorRequestHandler = HttpHandler; diff --git a/src/client-edge.ts b/src/client-edge.ts new file mode 100644 index 0000000..a8db254 --- /dev/null +++ b/src/client-edge.ts @@ -0,0 +1,14 @@ +import { AlternatorDynamoDBClientBase } from "./client-base.js"; +import { edgeRuntimePlatform } from "./runtime-edge.js"; +import type { AlternatorDynamoDBClientConfig } from "./types.js"; + +export class AlternatorDynamoDBClient extends AlternatorDynamoDBClientBase { + constructor(config: AlternatorDynamoDBClientConfig) { + super(config, edgeRuntimePlatform); + } +} + +export type { + AlternatorDynamoDBClientApi, + AlternatorRequestHandler, +} from "./client-base.js"; diff --git a/src/client.ts b/src/client.ts index 3e3405c..b938cb9 100644 --- a/src/client.ts +++ b/src/client.ts @@ -1,140 +1,14 @@ -import { - DescribeTableCommand, - DynamoDBClient, - type DynamoDBClientConfig, - type ServiceInputTypes, - type ServiceOutputTypes, -} from "@aws-sdk/client-dynamodb"; -import type { HttpHandler, HttpHandlerUserInput } from "@smithy/protocol-http"; -import type { HttpHandlerOptions } from "@smithy/types"; -import { DEFAULT_REGION, firstEndpointUrl, NO_AUTH_CREDENTIALS, normalizeConfig } from "./config.js"; -import { AlternatorDiscovery } from "./discovery.js"; -import { KeyRouteAffinityPlanner } from "./affinity.js"; -import { createAlternatorPostSigningMiddleware, createAlternatorRequestMiddleware } from "./middleware.js"; -import { assertRuntimeSupport, createRequestHandler } from "./runtime.js"; -import type { AlternatorDynamoDBClientConfig, AlternatorNode, NormalizedAlternatorConfig } from "./types.js"; - -export interface AlternatorDynamoDBClientApi { - nodes(): AlternatorNode[]; - refreshNodes(): Promise; - supportsScopedDiscovery(): Promise; - validateRouting(): Promise; - partitionKey(tableName: string): string | undefined; -} - -export class AlternatorDynamoDBClient extends DynamoDBClient { - readonly alternator: AlternatorDynamoDBClientApi; - private readonly alternatorConfig: NormalizedAlternatorConfig; - private readonly discovery: AlternatorDiscovery; - private readonly keyAffinity: KeyRouteAffinityPlanner; +import { AlternatorDynamoDBClientBase } from "./client-base.js"; +import { nodeRuntimePlatform } from "./runtime-node.js"; +import type { AlternatorDynamoDBClientConfig } from "./types.js"; +export class AlternatorDynamoDBClient extends AlternatorDynamoDBClientBase { constructor(config: AlternatorDynamoDBClientConfig) { - const alternatorConfig = normalizeConfig(config); - assertRuntimeSupport(alternatorConfig); - - const requestHandler = createRequestHandler(config, alternatorConfig); - const dynamoConfig = buildDynamoConfig(config, alternatorConfig, requestHandler); - - super(dynamoConfig); - - this.alternatorConfig = alternatorConfig; - this.discovery = new AlternatorDiscovery( - alternatorConfig, - this.config.requestHandler as HttpHandler, - ); - this.keyAffinity = new KeyRouteAffinityPlanner( - alternatorConfig.keyRouteAffinity, - (tableName) => this.discoverPartitionKey(tableName), - alternatorConfig.logger, - ); - this.alternator = { - nodes: () => this.discovery.getLiveNodes(), - refreshNodes: () => this.discovery.refreshLiveNodes(), - supportsScopedDiscovery: () => this.discovery.checkRackDatacenterSupport(), - validateRouting: () => this.discovery.checkIfRackAndDatacenterSetCorrectly(), - partitionKey: (tableName) => this.keyAffinity.getPartitionKeyName(tableName), - }; - - this.middlewareStack.addRelativeTo( - createAlternatorRequestMiddleware({ - discovery: this.discovery, - config: alternatorConfig, - keyAffinity: this.keyAffinity, - }), - { - relation: "before", - toMiddleware: "httpSigningMiddleware", - name: "alternatorRequestMiddleware", - override: true, - }, - ); - - this.middlewareStack.addRelativeTo( - createAlternatorPostSigningMiddleware(alternatorConfig), - { - relation: "after", - toMiddleware: "httpSigningMiddleware", - name: "alternatorPostSigningMiddleware", - override: true, - }, - ); - - } - - override destroy(): void { - this.discovery.destroy(); - super.destroy(); - } - - private async discoverPartitionKey(tableName: string): Promise { - const response = await this.send(new DescribeTableCommand({ TableName: tableName })); - const keyName = response.Table?.KeySchema?.find((key) => key.KeyType === "HASH")?.AttributeName; - if (!keyName) { - this.alternatorConfig.logger.warn?.("alternator key affinity: DescribeTable returned no partition key", { - tableName, - }); - return; - } - this.keyAffinity.setPartitionKeyName(tableName, keyName); - } -} - -function buildDynamoConfig( - input: AlternatorDynamoDBClientConfig, - alternatorConfig: NormalizedAlternatorConfig, - requestHandler: HttpHandlerUserInput, -): DynamoDBClientConfig { - const { - seeds: _seeds, - scheme: _scheme, - port: _port, - routing: _routing, - runtime: _runtime, - compression: _compression, - headerOptimization: _headerOptimization, - userAgent: _userAgent, - keyRouteAffinity: _keyRouteAffinity, - tls: _tls, - discovery: _discovery, - connection: _connection, - logger: _logger, - credentials, - region, - ...awsConfigWithEndpoint - } = input as AlternatorDynamoDBClientConfig & { endpoint?: unknown }; - const { endpoint: _endpoint, ...awsConfig } = awsConfigWithEndpoint; - - const dynamoConfig: DynamoDBClientConfig & { applyChecksum?: boolean } = { - ...awsConfig, - endpoint: firstEndpointUrl(alternatorConfig), - region: region ?? DEFAULT_REGION, - credentials: credentials ?? NO_AUTH_CREDENTIALS, - requestHandler, - }; - if (alternatorConfig.headerOptimization.enabled) { - dynamoConfig.applyChecksum = false; + super(config, nodeRuntimePlatform); } - return dynamoConfig; } -export type AlternatorRequestHandler = HttpHandler; +export type { + AlternatorDynamoDBClientApi, + AlternatorRequestHandler, +} from "./client-base.js"; diff --git a/src/compression-edge.ts b/src/compression-edge.ts new file mode 100644 index 0000000..9e86d74 --- /dev/null +++ b/src/compression-edge.ts @@ -0,0 +1,58 @@ +import type { HttpResponse } from "@smithy/protocol-http"; +import { bodyToBytes } from "./body.js"; +import { + bodyToReadableStream, + mapCompressedResponse, +} from "./compression-shared.js"; +import type { CompressedBody } from "./compression-types.js"; +import type { + AlternatorResponseCompressionAlgorithm, + NormalizedRequestCompressionOptions, +} from "./types.js"; + +export async function compressBody( + body: unknown, + config: NormalizedRequestCompressionOptions, +): Promise { + const bytes = bodyToBytes(body); + if (!bytes) { + return undefined; + } + + if (config.compressor) { + const compressed = await config.compressor(bytes); + return { + body: compressed.body, + contentEncoding: compressed.contentEncoding, + contentLength: compressed.contentLength ?? compressed.body.byteLength, + }; + } + + if (typeof CompressionStream === "undefined") { + throw new Error("gzip compression requires CompressionStream support in edge runtime"); + } + const arrayBuffer = bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength) as ArrayBuffer; + const stream = new Blob([arrayBuffer]).stream().pipeThrough(new CompressionStream("gzip")); + const compressed = new Uint8Array(await new Response(stream).arrayBuffer()); + return { + body: compressed, + contentEncoding: "gzip", + contentLength: compressed.byteLength, + }; +} + +export async function decompressResponse(response: HttpResponse): Promise { + return mapCompressedResponse(response, decompressWebResponseBody); +} + +async function decompressWebResponseBody( + body: unknown, + encoding: AlternatorResponseCompressionAlgorithm, +): Promise { + 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)); +} diff --git a/src/compression-node.ts b/src/compression-node.ts new file mode 100644 index 0000000..6b1a245 --- /dev/null +++ b/src/compression-node.ts @@ -0,0 +1,75 @@ +import type { HttpResponse } from "@smithy/protocol-http"; +import * as zlib from "node:zlib"; +import { bodyToBytes } from "./body.js"; +import { + bodyToAsyncBytes, + mapCompressedResponse, +} from "./compression-shared.js"; +import type { CompressedBody } from "./compression-types.js"; +import type { + AlternatorResponseCompressionAlgorithm, + NormalizedRequestCompressionOptions, +} from "./types.js"; + +export async function compressBody( + body: unknown, + config: NormalizedRequestCompressionOptions, +): Promise { + const bytes = bodyToBytes(body); + if (!bytes) { + return undefined; + } + + if (config.compressor) { + const compressed = await config.compressor(bytes); + return { + body: compressed.body, + contentEncoding: compressed.contentEncoding, + contentLength: compressed.contentLength ?? compressed.body.byteLength, + }; + } + + const compressed = zlib.gzipSync(bytes, { + level: config.gzipLevel, + }); + return { + body: compressed, + contentEncoding: "gzip", + contentLength: compressed.byteLength, + }; +} + +export async function decompressResponse(response: HttpResponse): Promise { + return mapCompressedResponse(response, decompressNodeResponseBody); +} + +async function decompressNodeResponseBody( + body: unknown, + encoding: AlternatorResponseCompressionAlgorithm, +): Promise { + if (isNodePipeableBody(body)) { + const decoder = encoding === "gzip" + ? zlib.createGunzip() + : zlib.createInflate(); + return body.pipe(decoder); + } + + const bytes = await bodyToAsyncBytes(body); + return encoding === "gzip" + ? zlib.gunzipSync(bytes) + : zlib.inflateSync(bytes); +} + +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 isObject(value: unknown): value is object { + return value !== null && typeof value === "object"; +} diff --git a/src/compression.ts b/src/compression-shared.ts similarity index 62% rename from src/compression.ts rename to src/compression-shared.ts index fc3b83a..04e4cbd 100644 --- a/src/compression.ts +++ b/src/compression-shared.ts @@ -1,54 +1,6 @@ import { HttpResponse } from "@smithy/protocol-http"; import { bodyToBytes } from "./body.js"; -import type { - AlternatorResponseCompressionAlgorithm, - AlternatorRuntime, - NormalizedRequestCompressionOptions, -} from "./types.js"; - -export async function compressBody( - body: unknown, - runtime: AlternatorRuntime, - config: NormalizedRequestCompressionOptions, -): Promise<{ body: Uint8Array; contentEncoding: string; contentLength: number } | undefined> { - const bytes = bodyToBytes(body); - if (!bytes) { - return undefined; - } - - if (config.compressor) { - const compressed = await config.compressor(bytes); - return { - body: compressed.body, - contentEncoding: compressed.contentEncoding, - contentLength: compressed.contentLength ?? compressed.body.byteLength, - }; - } - - if (runtime === "edge") { - if (typeof CompressionStream === "undefined") { - throw new Error("gzip compression requires CompressionStream support in edge runtime"); - } - const arrayBuffer = bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength) as ArrayBuffer; - const stream = new Blob([arrayBuffer]).stream().pipeThrough(new CompressionStream("gzip")); - const compressed = new Uint8Array(await new Response(stream).arrayBuffer()); - return { - body: compressed, - contentEncoding: "gzip", - contentLength: compressed.byteLength, - }; - } - - const zlib = await import("node:zlib"); - const compressed = zlib.gzipSync(bytes, { - level: config.gzipLevel, - }); - return { - body: compressed, - contentEncoding: "gzip", - contentLength: compressed.byteLength, - }; -} +import type { AlternatorResponseCompressionAlgorithm } from "./types.js"; export function applyResponseEncodingHeaders( headers: Record, @@ -87,9 +39,12 @@ export function responseAcceptEncoding( return parts.join(", "); } -export async function decompressResponse( +export async function mapCompressedResponse( response: HttpResponse, - runtime: AlternatorRuntime, + decompressBody: ( + body: unknown, + encoding: AlternatorResponseCompressionAlgorithm, + ) => Promise, ): Promise { const encoding = responseContentEncoding(getHeader(response.headers, "content-encoding")); const body: unknown = response.body; @@ -97,7 +52,7 @@ export async function decompressResponse( return response; } - const decodedBody = await decompressResponseBody(body, runtime, encoding); + const decodedBody = await decompressBody(body, encoding); return new HttpResponse({ statusCode: response.statusCode, ...(response.reason !== undefined ? { reason: response.reason } : {}), @@ -106,49 +61,7 @@ export async function decompressResponse( }); } -async function decompressResponseBody( - body: unknown, - runtime: AlternatorRuntime, - encoding: AlternatorResponseCompressionAlgorithm, -): Promise { - if (runtime === "edge") { - return decompressWebResponseBody(body, encoding); - } - return decompressNodeResponseBody(body, encoding); -} - -async function decompressNodeResponseBody( - body: unknown, - encoding: AlternatorResponseCompressionAlgorithm, -): Promise { - const zlib = await import("node:zlib"); - - if (isNodePipeableBody(body)) { - const decoder = encoding === "gzip" - ? zlib.createGunzip() - : zlib.createInflate(); - return body.pipe(decoder); - } - - const bytes = await bodyToAsyncBytes(body); - return encoding === "gzip" - ? zlib.gunzipSync(bytes) - : zlib.inflateSync(bytes); -} - -async function decompressWebResponseBody( - body: unknown, - encoding: AlternatorResponseCompressionAlgorithm, -): Promise { - 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 { +export async function bodyToReadableStream(body: unknown): Promise { if (typeof ReadableStream !== "undefined" && body instanceof ReadableStream) { return body; } @@ -160,7 +73,7 @@ async function bodyToReadableStream(body: unknown): Promise { return new Blob([bytesToArrayBuffer(bytes)]).stream(); } -async function bodyToAsyncBytes(body: unknown): Promise { +export async function bodyToAsyncBytes(body: unknown): Promise { const bytes = bodyToBytes(body); if (bytes) { return bytes; @@ -267,16 +180,6 @@ function bytesToArrayBuffer(bytes: Uint8Array): ArrayBuffer { 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; } { diff --git a/src/compression-types.ts b/src/compression-types.ts new file mode 100644 index 0000000..b211a56 --- /dev/null +++ b/src/compression-types.ts @@ -0,0 +1,12 @@ +import type { NormalizedRequestCompressionOptions } from "./types.js"; + +export interface CompressedBody { + readonly body: Uint8Array; + readonly contentEncoding: string; + readonly contentLength: number; +} + +export type AlternatorBodyCompressor = ( + body: unknown, + config: NormalizedRequestCompressionOptions, +) => Promise; diff --git a/src/config.ts b/src/config.ts index 6d5e3e8..15ea7de 100644 --- a/src/config.ts +++ b/src/config.ts @@ -13,6 +13,7 @@ import type { AlternatorRuntime, NormalizedAlternatorConfig, NormalizedCompressionOptions, + NormalizedDiscoveryOptions, NormalizedRequestCompressionOptions, NormalizedResponseCompressionOptions, } from "./types.js"; @@ -31,7 +32,6 @@ const DEFAULT_RESPONSE_COMPRESSION_ALGORITHMS = [ export const DEFAULT_REGION = "us-east-1"; export const DEFAULT_SCHEME = "http"; export const DEFAULT_PORT = 8080; -export const HTTPS_ALTERNATOR_PORT = 8043; export const NO_AUTH_CREDENTIALS = { accessKeyId: "alternator", @@ -236,8 +236,14 @@ function normalizeRequestCompression( } assertAllowedKeys(input, ["thresholdBytes", "gzipLevel", "compressor"], "compression.request"); const options = input as Exclude; + if (options.thresholdBytes !== undefined) { + assertNonNegative(options.thresholdBytes, "compression.request.thresholdBytes"); + } + if (options.compressor !== undefined && typeof options.compressor !== "function") { + throw new TypeError("compression.request.compressor must be a function"); + } const gzipLevel = options.gzipLevel; - if (gzipLevel !== undefined && (gzipLevel < -1 || gzipLevel > 9)) { + if (gzipLevel !== undefined && (!Number.isInteger(gzipLevel) || gzipLevel < -1 || gzipLevel > 9)) { throw new TypeError("compression.request.gzipLevel must be between -1 and 9"); } return { @@ -303,17 +309,36 @@ function normalizeHeaderOptimization(input: AlternatorDynamoDBClientConfig["head allowedHeaders: defaultAllowedHeaders(noAuth), }; } - assertAllowedKeys(input as Record, ["allowedHeaders", "additionalAllowedHeaders"], "headerOptimization"); - const baseHeaders = input.allowedHeaders ?? defaultAllowedHeaders(noAuth); + if (!isRecord(input)) { + throw new TypeError("headerOptimization must be a boolean or options object"); + } + assertAllowedKeys(input, ["allowedHeaders", "additionalAllowedHeaders"], "headerOptimization"); + const baseHeaders = input.allowedHeaders === undefined + ? defaultAllowedHeaders(noAuth) + : normalizeHeaderList(input.allowedHeaders, "headerOptimization.allowedHeaders"); + const additionalHeaders = input.additionalAllowedHeaders === undefined + ? [] + : normalizeHeaderList(input.additionalAllowedHeaders, "headerOptimization.additionalAllowedHeaders"); return { enabled: true, allowedHeaders: [ ...baseHeaders, - ...(input.additionalAllowedHeaders ?? []), + ...additionalHeaders, ], }; } +function normalizeHeaderList(value: unknown, label: string): string[] { + if (!isNonEmptyStringArray(value)) { + throw new TypeError(`${label} must be an array of non-empty strings`); + } + return [...value]; +} + +function isNonEmptyStringArray(value: unknown): value is string[] { + return Array.isArray(value) && value.every((header) => typeof header === "string" && header.trim() !== ""); +} + function normalizeKeyRouteAffinity(input: AlternatorDynamoDBClientConfig["keyRouteAffinity"]) { if (input === undefined || input === false) { return { @@ -385,17 +410,28 @@ function normalizeKeyRouteAffinityMode(mode: unknown): AlternatorKeyRouteAffinit function normalizeDiscovery( input: AlternatorDynamoDBClientConfig["discovery"], runtime: AlternatorRuntime, -) { - const refreshIntervalMs = input?.refreshIntervalMs ?? 60_000; - const requestRefreshIntervalMs = input?.requestRefreshIntervalMs ?? 60_000; - const timeoutMs = input?.timeoutMs ?? 2_000; +): NormalizedDiscoveryOptions { + const options = input as Record | undefined; + if (input !== undefined) { + if (!isRecord(options)) { + throw new TypeError("discovery must be an options object"); + } + assertAllowedKeys(options, ["background", "refreshIntervalMs", "requestRefreshIntervalMs", "timeoutMs"], "discovery"); + if (options.background !== undefined && typeof options.background !== "boolean") { + throw new TypeError("discovery.background must be a boolean"); + } + } + + const refreshIntervalMs = numberOption(options?.refreshIntervalMs, 60_000, "discovery.refreshIntervalMs"); + const requestRefreshIntervalMs = numberOption(options?.requestRefreshIntervalMs, 60_000, "discovery.requestRefreshIntervalMs"); + const timeoutMs = numberOption(options?.timeoutMs, 2_000, "discovery.timeoutMs"); assertNonNegative(refreshIntervalMs, "discovery.refreshIntervalMs"); assertNonNegative(requestRefreshIntervalMs, "discovery.requestRefreshIntervalMs"); assertNonNegative(timeoutMs, "discovery.timeoutMs"); return { - background: input?.background ?? runtime === "node", + background: typeof options?.background === "boolean" ? options.background : runtime === "node", refreshIntervalMs, requestRefreshIntervalMs, timeoutMs, @@ -403,24 +439,58 @@ function normalizeDiscovery( } function normalizeConnection(input: AlternatorConnectionOptions): AlternatorConnectionOptions { - if ("node" in input && input.node !== undefined && ("httpAgent" in input.node || "httpsAgent" in input.node)) { + if (!isRecord(input)) { + throw new TypeError("connection must be an options object"); + } + const options = input; + assertAllowedKeys(options, ["keepAlive", "maxSockets", "timeouts", "throwOnRequestTimeout", "node", "fetch"], "connection"); + if (options.keepAlive !== undefined && typeof options.keepAlive !== "boolean") { + throw new TypeError("connection.keepAlive must be a boolean"); + } + if (options.throwOnRequestTimeout !== undefined && typeof options.throwOnRequestTimeout !== "boolean") { + throw new TypeError("connection.throwOnRequestTimeout must be a boolean"); + } + if (options.timeouts !== undefined) { + if (!isRecord(options.timeouts)) { + throw new TypeError("connection.timeouts must be an options object"); + } + assertAllowedKeys(options.timeouts, ["connectMs", "requestMs", "socketMs"], "connection.timeouts"); + } + if (options.node !== undefined && !isRecord(options.node)) { + throw new TypeError("connection.node must be an options object"); + } + if (options.fetch !== undefined && !isRecord(options.fetch)) { + throw new TypeError("connection.fetch must be an options object"); + } + if (isRecord(options.node) && ("httpAgent" in options.node || "httpsAgent" in options.node)) { throw new TypeError("connection.node cannot include httpAgent or httpsAgent; use Alternator connection and tls options"); } - if ("maxSockets" in input && input.maxSockets !== undefined) { - assertPositive(input.maxSockets, "connection.maxSockets"); + if (options.maxSockets !== undefined) { + assertPositive(numberOption(options.maxSockets, undefined, "connection.maxSockets"), "connection.maxSockets"); } - if (input.timeouts && "connectMs" in input.timeouts && input.timeouts.connectMs !== undefined) { - assertNonNegative(input.timeouts.connectMs, "connection.timeouts.connectMs"); + const timeouts = isRecord(options.timeouts) ? options.timeouts : undefined; + if (timeouts?.connectMs !== undefined) { + assertNonNegative(numberOption(timeouts.connectMs, undefined, "connection.timeouts.connectMs"), "connection.timeouts.connectMs"); } - if (input.timeouts?.requestMs !== undefined) { - assertNonNegative(input.timeouts.requestMs, "connection.timeouts.requestMs"); + if (timeouts?.requestMs !== undefined) { + assertNonNegative(numberOption(timeouts.requestMs, undefined, "connection.timeouts.requestMs"), "connection.timeouts.requestMs"); } - if (input.timeouts && "socketMs" in input.timeouts && input.timeouts.socketMs !== undefined) { - assertNonNegative(input.timeouts.socketMs, "connection.timeouts.socketMs"); + if (timeouts?.socketMs !== undefined) { + assertNonNegative(numberOption(timeouts.socketMs, undefined, "connection.timeouts.socketMs"), "connection.timeouts.socketMs"); } return input; } +function numberOption(value: unknown, fallback: number, label: string): number; +function numberOption(value: unknown, fallback: undefined, label: string): number; +function numberOption(value: unknown, fallback: number | undefined, label: string): number { + const normalized = value === undefined ? fallback : value; + if (typeof normalized !== "number") { + throw new TypeError(`${label} must be a number`); + } + return normalized; +} + function assertPositive(value: number, label: string): void { if (!Number.isFinite(value) || value <= 0) { throw new TypeError(`${label} must be a positive number`); diff --git a/src/document-edge.ts b/src/document-edge.ts new file mode 100644 index 0000000..2c228b6 --- /dev/null +++ b/src/document-edge.ts @@ -0,0 +1,40 @@ +import type { DynamoDBClient } from "@aws-sdk/client-dynamodb"; +import { + DynamoDBDocumentClient, + type TranslateConfig, +} from "@aws-sdk/lib-dynamodb"; +import { AlternatorDynamoDBClient } from "./client-edge.js"; +import type { AlternatorDynamoDBClientConfig } from "./types.js"; + +export class AlternatorDynamoDBDocumentClient extends DynamoDBDocumentClient { + private constructor( + client: DynamoDBClient, + translateConfig?: TranslateConfig, + private readonly ownedClient?: AlternatorDynamoDBClient, + ) { + super(client, translateConfig); + } + + static override from( + client: DynamoDBClient, + translateConfig?: TranslateConfig, + ): AlternatorDynamoDBDocumentClient { + return new AlternatorDynamoDBDocumentClient(client, translateConfig); + } + + static fromConfig( + config: AlternatorDynamoDBClientConfig, + translateConfig?: TranslateConfig, + ): AlternatorDynamoDBDocumentClient { + const client = new AlternatorDynamoDBClient(config); + return new AlternatorDynamoDBDocumentClient(client, translateConfig, client); + } + + override destroy(): void { + this.ownedClient?.destroy(); + super.destroy(); + } +} + +export type { TranslateConfig } from "@aws-sdk/lib-dynamodb"; +export * from "@aws-sdk/lib-dynamodb"; diff --git a/src/edge.ts b/src/edge.ts new file mode 100644 index 0000000..e8534c3 --- /dev/null +++ b/src/edge.ts @@ -0,0 +1,49 @@ +export { AlternatorDynamoDBClient } from "./client-edge.js"; +export { routing } from "./routing.js"; +export type { + AlternatorDynamoDBClientApi, + AlternatorRequestHandler, +} from "./client-edge.js"; +export type { + AlternatorCompressionOptions, + AlternatorConnectionOptions, + AlternatorConnectionTimeoutOptions, + AlternatorDiscoveryOptions, + AlternatorDynamoDBClientConfig, + AlternatorEdgeConnectionOptions, + AlternatorEdgeDynamoDBClientConfig, + AlternatorHeaderOptimizationOptions, + AlternatorKeyRouteAffinityConfig, + AlternatorKeyRouteAffinityMode, + AlternatorKeyRouteAffinityOptions, + AlternatorLogger, + AlternatorNodeConnectionOptions, + AlternatorNodeDynamoDBClientConfig, + AlternatorNode, + AlternatorPartitionKeyByTable, + AlternatorRequestCompressionConfig, + AlternatorRequestCompressionOptions, + AlternatorResponseCompressionAlgorithm, + AlternatorResponseCompressionConfig, + AlternatorResponseCompressionOptions, + AlternatorRuntime, + AlternatorScheme, + AlternatorRequestCompressor, + AlternatorRequestCompressorResult, + AlternatorTlsOptions, + AlternatorTlsMaterial, + AlternatorUserAgentConfig, + AlternatorUserAgentOptions, + AlternatorUserAgentTransformer, + NonEmptyReadonlyArray, +} from "./types.js"; +export type { + AlternatorClusterRoutingScope, + AlternatorDatacenterRoutingScope, + AlternatorDatacenterRoutingScopeOptions, + AlternatorRackRoutingScope, + AlternatorRackRoutingScopeOptions, + AlternatorRoutingFallbackOptions, + AlternatorRoutingScope, + AlternatorRoutingScopeKind, +} from "./routing.js"; diff --git a/src/middleware.ts b/src/middleware.ts index 1db2459..1233957 100644 --- a/src/middleware.ts +++ b/src/middleware.ts @@ -1,15 +1,13 @@ import { HttpRequest } from "@smithy/protocol-http"; import type { FinalizeRequestMiddleware, HandlerExecutionContext } from "@smithy/types"; -import { - applyResponseEncodingHeaders, - compressBody, -} from "./compression.js"; +import { applyResponseEncodingHeaders } from "./compression-shared.js"; import { hostForUrl } from "./config.js"; import type { AlternatorDiscovery } from "./discovery.js"; import type { KeyRouteAffinityPlanner } from "./affinity.js"; import type { AlternatorQueryPlan } from "./query-plan.js"; import type { AlternatorNode, NormalizedAlternatorConfig } from "./types.js"; import { applyUserAgent } from "./user-agent.js"; +import type { AlternatorBodyCompressor } from "./compression-types.js"; const queryPlanKey = "__alternatorQueryPlan"; @@ -17,12 +15,14 @@ export interface AlternatorMiddlewareOptions { discovery: AlternatorDiscovery; config: NormalizedAlternatorConfig; keyAffinity: KeyRouteAffinityPlanner; + compressBody: AlternatorBodyCompressor; } export function createAlternatorRequestMiddleware({ discovery, config, keyAffinity, + compressBody, }: AlternatorMiddlewareOptions): FinalizeRequestMiddleware { return (next, context) => async (args) => { if (!HttpRequest.isInstance(args.request)) { @@ -50,7 +50,7 @@ export function createAlternatorRequestMiddleware { if (request.headers["content-encoding"] || request.headers["Content-Encoding"]) { return request; @@ -183,7 +184,7 @@ async function maybeCompressRequest( return request; } - const compressedBody = await compressBody(request.body, config.runtime, config.compression.request); + const compressedBody = await compressBody(request.body, config.compression.request); if (!compressedBody) { return request; } diff --git a/src/murmur.ts b/src/murmur.ts index 414aae2..e2b3a8d 100644 --- a/src/murmur.ts +++ b/src/murmur.ts @@ -107,8 +107,3 @@ function rotl64(value: bigint, rotate: bigint): bigint { function toSigned64(value: bigint): bigint { return value >= SIGN_BIT ? value - (1n << 64n) : value; } - -export function positiveModulo(value: bigint, divisor: number): number { - const mod = value % BigInt(divisor); - return Number(mod < 0n ? mod + BigInt(divisor) : mod); -} diff --git a/src/routing.ts b/src/routing.ts index 64a1d23..a976840 100644 --- a/src/routing.ts +++ b/src/routing.ts @@ -76,24 +76,7 @@ export const routing = { rack, }; -export function routingQueries(scope: AlternatorRoutingScope): LocalNodesQuery[] { - const query = queryForScope(scope); - const fallback = "fallback" in scope ? scope.fallback : undefined; - return fallback ? [query, ...routingQueries(fallback)] : [query]; -} - export function routingChain(scope: AlternatorRoutingScope): AlternatorRoutingScope[] { const fallback = "fallback" in scope ? scope.fallback : undefined; return fallback ? [scope, ...routingChain(fallback)] : [scope]; } - -function queryForScope(scope: AlternatorRoutingScope): LocalNodesQuery { - switch (scope.kind) { - case "cluster": - return {}; - case "datacenter": - return { dc: scope.datacenter }; - case "rack": - return { dc: scope.datacenter, rack: scope.rack }; - } -} diff --git a/src/runtime-common.ts b/src/runtime-common.ts new file mode 100644 index 0000000..0443010 --- /dev/null +++ b/src/runtime-common.ts @@ -0,0 +1,71 @@ +import type { HttpHandler, HttpHandlerUserInput, HttpRequest, HttpResponse } from "@smithy/protocol-http"; +import type { + HttpHandlerOptions, + RequestHandlerMetadata, +} from "@smithy/types"; + +type GenericHttpHandler = HttpHandler>; +export type ResponseDecompressor = (response: HttpResponse) => Promise; + +class ResponseCompressionHttpHandler implements GenericHttpHandler { + readonly metadata: RequestHandlerMetadata; + + constructor( + private readonly delegate: GenericHttpHandler, + private readonly decompressResponse: ResponseDecompressor, + ) { + this.metadata = delegate.metadata ?? { handlerProtocol: "http/1.1" }; + } + + async handle( + request: HttpRequest, + options?: HttpHandlerOptions, + ): Promise<{ response: HttpResponse }> { + const result = await this.delegate.handle(request, options); + return { + response: await this.decompressResponse(result.response), + }; + } + + destroy(): void { + this.delegate.destroy?.(); + } + + updateHttpClientConfig( + key: keyof Record, + value: Record[typeof key], + ): void { + this.delegate.updateHttpClientConfig(key, value); + } + + httpHandlerConfigs(): Record { + return this.delegate.httpHandlerConfigs(); + } +} + +export function withResponseCompression( + requestHandler: HttpHandlerUserInput, + enabled: boolean, + decompressResponse: ResponseDecompressor, +): HttpHandlerUserInput { + if (!enabled) { + return requestHandler; + } + if (!isHttpHandler(requestHandler)) { + throw new TypeError("compression.response requires requestHandler to be an HTTP handler instance"); + } + return new ResponseCompressionHttpHandler(requestHandler, decompressResponse); +} + +function isHttpHandler(requestHandler: HttpHandlerUserInput): requestHandler is GenericHttpHandler { + return ( + typeof requestHandler === "object" && + requestHandler !== null && + "handle" in requestHandler && + typeof (requestHandler as { handle?: unknown }).handle === "function" && + "updateHttpClientConfig" in requestHandler && + typeof (requestHandler as { updateHttpClientConfig?: unknown }).updateHttpClientConfig === "function" && + "httpHandlerConfigs" in requestHandler && + typeof (requestHandler as { httpHandlerConfigs?: unknown }).httpHandlerConfigs === "function" + ); +} diff --git a/src/runtime-edge.ts b/src/runtime-edge.ts new file mode 100644 index 0000000..4f15138 --- /dev/null +++ b/src/runtime-edge.ts @@ -0,0 +1,74 @@ +import { FetchHttpHandler } from "@smithy/fetch-http-handler"; +import type { HttpHandlerUserInput } from "@smithy/protocol-http"; +import { compressBody, decompressResponse } from "./compression-edge.js"; +import { withResponseCompression } from "./runtime-common.js"; +import type { + AlternatorDynamoDBClientConfig, + NormalizedAlternatorConfig, +} from "./types.js"; + +export const edgeRuntimePlatform = { + assertRuntimeSupport, + createRequestHandler, + compressBody, +}; + +function assertRuntimeSupport(config: NormalizedAlternatorConfig): void { + if (config.runtime !== "edge") { + throw new Error("Alternator edge entrypoint requires runtime edge"); + } + + if (config.tls && Object.keys(config.tls).length > 0) { + throw new Error("Alternator edge runtime does not support custom TLS or CA options"); + } + + const connection = config.connection; + if (connection && "maxSockets" in connection && connection.maxSockets !== undefined) { + throw new Error("Alternator edge runtime does not support socket pool tuning"); + } + if ( + (connection?.timeouts && "socketMs" in connection.timeouts && connection.timeouts.socketMs !== undefined) || + (connection?.timeouts && "connectMs" in connection.timeouts && connection.timeouts.connectMs !== undefined) + ) { + throw new Error("Alternator edge runtime does not support Node socket timeout options"); + } + if (connection && "node" in connection && connection.node !== undefined) { + throw new Error("Alternator edge runtime does not support Node HTTP handler options"); + } + + const requestCompression = config.compression.request; + if (requestCompression.enabled && !requestCompression.compressor && typeof CompressionStream === "undefined") { + throw new Error("Alternator edge runtime gzip compression requires CompressionStream support"); + } + if (config.compression.response.enabled && typeof DecompressionStream === "undefined") { + throw new Error("Alternator edge runtime response compression requires DecompressionStream support"); + } +} + +function createRequestHandler( + input: AlternatorDynamoDBClientConfig, + config: NormalizedAlternatorConfig, +): HttpHandlerUserInput { + if (input.requestHandler) { + return withResponseCompression( + input.requestHandler, + config.compression.response.enabled, + decompressResponse, + ); + } + + const fetchOptions = { + ...(config.connection && "fetch" in config.connection ? config.connection.fetch : undefined), + }; + if (config.connection?.timeouts?.requestMs !== undefined) { + fetchOptions.requestTimeout = config.connection.timeouts.requestMs; + } + if (config.connection?.keepAlive !== undefined) { + fetchOptions.keepAlive = config.connection.keepAlive; + } + return withResponseCompression( + new FetchHttpHandler(fetchOptions), + config.compression.response.enabled, + decompressResponse, + ); +} diff --git a/src/runtime-node.ts b/src/runtime-node.ts new file mode 100644 index 0000000..2acf7ec --- /dev/null +++ b/src/runtime-node.ts @@ -0,0 +1,157 @@ +import { NodeHttpHandler } from "@smithy/node-http-handler"; +import type { HttpHandler, HttpHandlerUserInput, HttpRequest, HttpResponse } from "@smithy/protocol-http"; +import type { + HttpHandlerOptions, + NodeHttpHandlerOptions, +} from "@smithy/types"; +import { readFile } from "node:fs/promises"; +import { compressBody, decompressResponse } from "./compression-node.js"; +import { withResponseCompression } from "./runtime-common.js"; +import type { + AlternatorDynamoDBClientConfig, + AlternatorTlsMaterial, + NormalizedAlternatorConfig, +} from "./types.js"; + +type Handler = HttpHandler; + +export const nodeRuntimePlatform = { + assertRuntimeSupport, + createRequestHandler, + compressBody, +}; + +function assertRuntimeSupport(config: NormalizedAlternatorConfig): void { + if (config.runtime !== "node") { + throw new Error("Alternator Node entrypoint requires runtime node; use @scylladb/alternator-client/edge for edge runtime"); + } +} + +function createRequestHandler( + input: AlternatorDynamoDBClientConfig, + config: NormalizedAlternatorConfig, +): HttpHandlerUserInput { + if (input.requestHandler) { + return withResponseCompression( + input.requestHandler, + config.compression.response.enabled, + decompressResponse, + ); + } + + return withResponseCompression( + new LazyNodeHttpHandler(() => buildNodeHandlerOptions(config)), + config.compression.response.enabled, + decompressResponse, + ); +} + +class LazyNodeHttpHandler implements Handler { + readonly metadata = { handlerProtocol: "http/1.1" }; + private delegate?: Handler; + private pendingUpdates = new Map(); + + constructor(private readonly optionsProvider: () => Promise) {} + + async handle( + request: HttpRequest, + options?: HttpHandlerOptions, + ): Promise<{ response: HttpResponse }> { + const delegate = await this.getDelegate(); + return delegate.handle(request, options); + } + + destroy(): void { + this.delegate?.destroy?.(); + } + + updateHttpClientConfig( + key: K, + value: NodeHttpHandlerOptions[K], + ): void { + if (this.delegate && "updateHttpClientConfig" in this.delegate) { + this.delegate.updateHttpClientConfig?.(key, value); + return; + } + this.pendingUpdates.set(key, value); + } + + httpHandlerConfigs(): NodeHttpHandlerOptions { + if (this.delegate && "httpHandlerConfigs" in this.delegate) { + return this.delegate.httpHandlerConfigs?.() ?? {}; + } + return Object.fromEntries(this.pendingUpdates); + } + + private async getDelegate(): Promise { + if (!this.delegate) { + const options = await this.optionsProvider(); + for (const [key, value] of this.pendingUpdates) { + (options as Record)[key] = value; + } + this.delegate = new NodeHttpHandler(options); + } + return this.delegate; + } +} + +async function buildNodeHandlerOptions( + config: NormalizedAlternatorConfig, +): Promise { + const connection = config.connection; + const tls = config.tls; + const keepAlive = connection?.keepAlive ?? true; + const maxSockets = connection && "maxSockets" in connection ? connection.maxSockets ?? 50 : 50; + + const httpAgent: Record = { keepAlive, maxSockets }; + const httpsAgent: Record = { keepAlive, maxSockets }; + + if (tls) { + if (tls.ca !== undefined) { + httpsAgent.ca = await tlsMaterialValue(tls.ca); + } + if (tls.cert !== undefined) { + httpsAgent.cert = await tlsMaterialValue(tls.cert); + } + if (tls.key !== undefined) { + httpsAgent.key = await tlsMaterialValue(tls.key); + } + if (tls.rejectUnauthorized !== undefined) { + httpsAgent.rejectUnauthorized = tls.rejectUnauthorized; + } + if (tls.sessionCache === false) { + httpsAgent.maxCachedSessions = 0; + } + } + + const options: NodeHttpHandlerOptions = { + httpAgent, + httpsAgent, + ...(connection && "node" in connection ? connection.node : undefined), + }; + + if (connection?.timeouts?.requestMs !== undefined) { + options.requestTimeout = connection.timeouts.requestMs; + } + if (connection?.timeouts && "connectMs" in connection.timeouts && connection.timeouts.connectMs !== undefined) { + options.connectionTimeout = connection.timeouts.connectMs; + } + if (connection?.timeouts && "socketMs" in connection.timeouts && connection.timeouts.socketMs !== undefined) { + options.socketTimeout = connection.timeouts.socketMs; + } + if (connection && "throwOnRequestTimeout" in connection && connection.throwOnRequestTimeout !== undefined) { + options.throwOnRequestTimeout = connection.throwOnRequestTimeout; + } + + return options; +} + +async function tlsMaterialValue(material: AlternatorTlsMaterial): Promise { + if ("file" in material) { + return readFile(material.file); + } + if ("bytes" in material) { + return material.bytes; + } + return material.text; +} diff --git a/src/runtime.ts b/src/runtime.ts deleted file mode 100644 index aed5cb8..0000000 --- a/src/runtime.ts +++ /dev/null @@ -1,247 +0,0 @@ -import { FetchHttpHandler } from "@smithy/fetch-http-handler"; -import type { - HttpHandlerOptions, - NodeHttpHandlerOptions, - RequestHandlerMetadata, -} from "@smithy/types"; -import type { HttpHandler, HttpHandlerUserInput, HttpRequest, HttpResponse } from "@smithy/protocol-http"; -import { decompressResponse } from "./compression.js"; -import type { - AlternatorDynamoDBClientConfig, - AlternatorTlsMaterial, - NormalizedAlternatorConfig, -} from "./types.js"; - -type Handler = HttpHandler; -type GenericHttpHandler = HttpHandler>; - -export function assertRuntimeSupport(config: NormalizedAlternatorConfig): void { - if (config.runtime !== "edge") { - return; - } - - if (config.tls && Object.keys(config.tls).length > 0) { - throw new Error("Alternator edge runtime does not support custom TLS or CA options"); - } - - const connection = config.connection; - if (connection && "maxSockets" in connection && connection.maxSockets !== undefined) { - throw new Error("Alternator edge runtime does not support socket pool tuning"); - } - if ( - (connection?.timeouts && "socketMs" in connection.timeouts && connection.timeouts.socketMs !== undefined) || - (connection?.timeouts && "connectMs" in connection.timeouts && connection.timeouts.connectMs !== undefined) - ) { - throw new Error("Alternator edge runtime does not support Node socket timeout options"); - } - if (connection && "node" in connection && connection.node !== undefined) { - throw new Error("Alternator edge runtime does not support Node HTTP handler options"); - } - - const requestCompression = config.compression.request; - if (requestCompression.enabled && !requestCompression.compressor && typeof CompressionStream === "undefined") { - throw new Error("Alternator edge runtime gzip compression requires CompressionStream support"); - } - if (config.compression.response.enabled && typeof DecompressionStream === "undefined") { - throw new Error("Alternator edge runtime response compression requires DecompressionStream support"); - } -} - -export function createRequestHandler( - input: AlternatorDynamoDBClientConfig, - config: NormalizedAlternatorConfig, -): HttpHandlerUserInput { - if (input.requestHandler) { - return withResponseCompression(input.requestHandler, config); - } - - if (config.runtime === "edge") { - const fetchOptions = { - ...(config.connection && "fetch" in config.connection ? config.connection.fetch : undefined), - }; - if (config.connection?.timeouts?.requestMs !== undefined) { - fetchOptions.requestTimeout = config.connection.timeouts.requestMs; - } - if (config.connection?.keepAlive !== undefined) { - fetchOptions.keepAlive = config.connection.keepAlive; - } - return withResponseCompression(new FetchHttpHandler(fetchOptions), config); - } - - return withResponseCompression(new LazyNodeHttpHandler(() => buildNodeHandlerOptions(config)), config); -} - -class LazyNodeHttpHandler implements Handler { - readonly metadata = { handlerProtocol: "http/1.1" }; - private delegate?: Handler; - private pendingUpdates = new Map(); - - constructor(private readonly optionsProvider: () => Promise) {} - - async handle( - request: HttpRequest, - options?: HttpHandlerOptions, - ): Promise<{ response: HttpResponse }> { - const delegate = await this.getDelegate(); - return delegate.handle(request, options); - } - - destroy(): void { - this.delegate?.destroy?.(); - } - - updateHttpClientConfig( - key: K, - value: NodeHttpHandlerOptions[K], - ): void { - if (this.delegate && "updateHttpClientConfig" in this.delegate) { - this.delegate.updateHttpClientConfig?.(key, value); - return; - } - this.pendingUpdates.set(key, value); - } - - httpHandlerConfigs(): NodeHttpHandlerOptions { - if (this.delegate && "httpHandlerConfigs" in this.delegate) { - return this.delegate.httpHandlerConfigs?.() ?? {}; - } - return Object.fromEntries(this.pendingUpdates); - } - - private async getDelegate(): Promise { - if (!this.delegate) { - const { NodeHttpHandler } = await import("@smithy/node-http-handler"); - const options = await this.optionsProvider(); - for (const [key, value] of this.pendingUpdates) { - (options as Record)[key] = value; - } - this.delegate = new NodeHttpHandler(options); - } - return this.delegate; - } -} - -class ResponseCompressionHttpHandler implements GenericHttpHandler { - readonly metadata: RequestHandlerMetadata; - - constructor( - private readonly delegate: GenericHttpHandler, - private readonly runtime: NormalizedAlternatorConfig["runtime"], - ) { - this.metadata = delegate.metadata ?? { handlerProtocol: "http/1.1" }; - } - - async handle( - request: HttpRequest, - options?: HttpHandlerOptions, - ): Promise<{ response: HttpResponse }> { - const result = await this.delegate.handle(request, options); - return { - response: await decompressResponse(result.response, this.runtime), - }; - } - - destroy(): void { - this.delegate.destroy?.(); - } - - updateHttpClientConfig( - key: keyof Record, - value: Record[typeof key], - ): void { - this.delegate.updateHttpClientConfig(key, value); - } - - httpHandlerConfigs(): Record { - return this.delegate.httpHandlerConfigs(); - } -} - -function withResponseCompression( - requestHandler: HttpHandlerUserInput, - config: NormalizedAlternatorConfig, -): HttpHandlerUserInput { - if (!config.compression.response.enabled) { - return requestHandler; - } - if (!isHttpHandler(requestHandler)) { - throw new TypeError("compression.response requires requestHandler to be an HTTP handler instance"); - } - return new ResponseCompressionHttpHandler(requestHandler, config.runtime); -} - -function isHttpHandler(requestHandler: HttpHandlerUserInput): requestHandler is GenericHttpHandler { - return ( - typeof requestHandler === "object" && - requestHandler !== null && - "handle" in requestHandler && - typeof (requestHandler as { handle?: unknown }).handle === "function" && - "updateHttpClientConfig" in requestHandler && - typeof (requestHandler as { updateHttpClientConfig?: unknown }).updateHttpClientConfig === "function" && - "httpHandlerConfigs" in requestHandler && - typeof (requestHandler as { httpHandlerConfigs?: unknown }).httpHandlerConfigs === "function" - ); -} - -async function buildNodeHandlerOptions( - config: NormalizedAlternatorConfig, -): Promise { - const connection = config.connection; - const tls = config.tls; - const keepAlive = connection?.keepAlive ?? true; - const maxSockets = connection && "maxSockets" in connection ? connection.maxSockets ?? 50 : 50; - - const httpAgent: Record = { keepAlive, maxSockets }; - const httpsAgent: Record = { keepAlive, maxSockets }; - - if (tls) { - if (tls.ca !== undefined) { - httpsAgent.ca = await tlsMaterialValue(tls.ca); - } - if (tls.cert !== undefined) { - httpsAgent.cert = await tlsMaterialValue(tls.cert); - } - if (tls.key !== undefined) { - httpsAgent.key = await tlsMaterialValue(tls.key); - } - if (tls.rejectUnauthorized !== undefined) { - httpsAgent.rejectUnauthorized = tls.rejectUnauthorized; - } - if (tls.sessionCache === false) { - httpsAgent.maxCachedSessions = 0; - } - - } - - const options: NodeHttpHandlerOptions = { - httpAgent, - httpsAgent, - ...(connection && "node" in connection ? connection.node : undefined), - }; - - if (connection?.timeouts?.requestMs !== undefined) { - options.requestTimeout = connection.timeouts.requestMs; - } - if (connection?.timeouts && "connectMs" in connection.timeouts && connection.timeouts.connectMs !== undefined) { - options.connectionTimeout = connection.timeouts.connectMs; - } - if (connection?.timeouts && "socketMs" in connection.timeouts && connection.timeouts.socketMs !== undefined) { - options.socketTimeout = connection.timeouts.socketMs; - } - if (connection && "throwOnRequestTimeout" in connection && connection.throwOnRequestTimeout !== undefined) { - options.throwOnRequestTimeout = connection.throwOnRequestTimeout; - } - - return options; -} - -async function tlsMaterialValue(material: AlternatorTlsMaterial): Promise { - if ("file" in material) { - const fs = await import("node:fs/promises"); - return fs.readFile(material.file); - } - if ("bytes" in material) { - return material.bytes; - } - return material.text; -} diff --git a/src/user-agent.ts b/src/user-agent.ts index f22dd00..4639878 100644 --- a/src/user-agent.ts +++ b/src/user-agent.ts @@ -1,4 +1,3 @@ -import packageJson from "../package.json" with { type: "json" }; import type { AlternatorUserAgentConfig, AlternatorUserAgentOptions, @@ -6,10 +5,12 @@ import type { NormalizedUserAgentOptions, } from "./types.js"; +declare const __PACKAGE_VERSION__: string; + export const ALTERNATOR_USER_AGENT_PRODUCT = "scylladb-alternator-client-javascript"; export function alternatorUserAgentToken(): string { - return `${ALTERNATOR_USER_AGENT_PRODUCT}/${normalizeVersion(packageJson.version)}`; + return `${ALTERNATOR_USER_AGENT_PRODUCT}/${normalizeVersion(packageVersion())}`; } export function normalizeUserAgent(input: AlternatorUserAgentConfig | undefined): NormalizedUserAgentOptions { @@ -104,3 +105,7 @@ function normalizeVersion(version: string | undefined): string { } return trimmed.replace(/\s+/g, "_"); } + +function packageVersion(): string | undefined { + return typeof __PACKAGE_VERSION__ === "string" ? __PACKAGE_VERSION__ : undefined; +} diff --git a/test/config.test.ts b/test/config.test.ts index 316d471..ddadb11 100644 --- a/test/config.test.ts +++ b/test/config.test.ts @@ -1,6 +1,7 @@ import { ListTablesCommand } from "@aws-sdk/client-dynamodb"; import { describe, expect, it } from "vitest"; import { AlternatorDynamoDBClient } from "../src/index.js"; +import { AlternatorDynamoDBClient as EdgeAlternatorDynamoDBClient } from "../src/edge.js"; import { normalizeConfig } from "../src/config.js"; import { RecordingHandler } from "./helpers.js"; @@ -70,7 +71,7 @@ describe("AlternatorDynamoDBClient config", () => { expect( () => - new AlternatorDynamoDBClient({ + new EdgeAlternatorDynamoDBClient({ seeds: ["localhost"], runtime: "edge", tls: { ca: { file: "/tmp/ca.pem" } }, @@ -79,7 +80,7 @@ describe("AlternatorDynamoDBClient config", () => { expect( () => - new AlternatorDynamoDBClient({ + new EdgeAlternatorDynamoDBClient({ seeds: ["localhost"], runtime: "edge", connection: { maxSockets: 8 }, @@ -108,6 +109,42 @@ describe("AlternatorDynamoDBClient config", () => { ).toThrow(/mutually exclusive/); }); + it("validates object-shaped JavaScript options", () => { + expect( + () => + new AlternatorDynamoDBClient({ + seeds: ["localhost"], + headerOptimization: "" as never, + }), + ).toThrow(/headerOptimization/); + + expect( + () => + new AlternatorDynamoDBClient({ + seeds: ["localhost"], + headerOptimization: { + allowedHeaders: "Host", + } as never, + }), + ).toThrow(/allowedHeaders/); + + expect( + () => + new AlternatorDynamoDBClient({ + seeds: ["localhost"], + discovery: "bad" as never, + }), + ).toThrow(/discovery/); + + expect( + () => + new AlternatorDynamoDBClient({ + seeds: ["localhost"], + connection: "bad" as never, + }), + ).toThrow(/connection/); + }); + it("rejects direct Node agent overrides in connection.node", () => { expect( () => diff --git a/test/discovery.test.ts b/test/discovery.test.ts index 6888d9f..f0c3c5b 100644 --- a/test/discovery.test.ts +++ b/test/discovery.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "vitest"; import { AlternatorDynamoDBClient, routing } from "../src/index.js"; +import { AlternatorDynamoDBClient as EdgeAlternatorDynamoDBClient } from "../src/edge.js"; import { RecordingHandler } from "./helpers.js"; import { ListTablesCommand } from "@aws-sdk/client-dynamodb"; @@ -257,7 +258,7 @@ describe("Alternator discovery", () => { } return { TableNames: [] }; }); - const client = new AlternatorDynamoDBClient({ + const client = new EdgeAlternatorDynamoDBClient({ seeds: ["seed"], runtime: "edge", requestHandler: handler, diff --git a/tsconfig.json b/tsconfig.json index 3fe1beb..18d47a4 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -18,5 +18,5 @@ "sourceMap": true, "outDir": "dist" }, - "include": ["src/**/*.ts", "test/**/*.ts", "vitest.config.ts"] + "include": ["src/**/*.ts", "test/**/*.ts", "vitest.config.ts", "vitest.integration.config.ts", "tsup.config.ts"] } diff --git a/tsup.config.ts b/tsup.config.ts new file mode 100644 index 0000000..a9a246e --- /dev/null +++ b/tsup.config.ts @@ -0,0 +1,19 @@ +import { defineConfig } from "tsup"; +import packageJson from "./package.json" with { type: "json" }; + +export default defineConfig({ + entry: [ + "src/index.ts", + "src/edge.ts", + "src/document.ts", + "src/document-edge.ts", + ], + format: ["esm", "cjs"], + dts: true, + sourcemap: true, + clean: true, + target: "es2022", + define: { + __PACKAGE_VERSION__: JSON.stringify(packageJson.version), + }, +}); diff --git a/vitest.config.ts b/vitest.config.ts index f90096c..ee7bf05 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -1,6 +1,10 @@ import { defineConfig } from "vitest/config"; +import packageJson from "./package.json" with { type: "json" }; export default defineConfig({ + define: { + __PACKAGE_VERSION__: JSON.stringify(packageJson.version), + }, test: { environment: "node", include: ["test/*.test.ts", "test/unit/**/*.test.ts"], diff --git a/vitest.integration.config.ts b/vitest.integration.config.ts index b0df46f..14776f5 100644 --- a/vitest.integration.config.ts +++ b/vitest.integration.config.ts @@ -1,6 +1,10 @@ import { defineConfig } from "vitest/config"; +import packageJson from "./package.json" with { type: "json" }; export default defineConfig({ + define: { + __PACKAGE_VERSION__: JSON.stringify(packageJson.version), + }, test: { environment: "node", include: ["test/integration-test/**/*.test.ts"],