From 0dec8bb9c90229cbf46f9720b755515e305a4119 Mon Sep 17 00:00:00 2001 From: Gabriel Colson Date: Wed, 3 Sep 2025 19:56:24 +0200 Subject: [PATCH 1/3] feat: add batch utility --- .projenrc.ts | 13 ++ packages/powertools-batch/.gitattributes | 22 ++++ packages/powertools-batch/.gitignore | 45 +++++++ packages/powertools-batch/.npmignore | 20 +++ packages/powertools-batch/.projen/deps.json | 33 +++++ packages/powertools-batch/.projen/files.json | 20 +++ packages/powertools-batch/.projen/tasks.json | 120 ++++++++++++++++++ packages/powertools-batch/LICENSE | 19 +++ packages/powertools-batch/README.md | 77 +++++++++++ packages/powertools-batch/docgen.json | 8 ++ packages/powertools-batch/examples/example.ts | 16 +++ packages/powertools-batch/package.json | 50 ++++++++ .../powertools-batch/src/batch-processor.ts | 59 +++++++++ packages/powertools-batch/src/constants.ts | 9 ++ packages/powertools-batch/src/index.ts | 4 + .../src/process-partial-response.ts | 12 ++ packages/powertools-batch/src/processor.ts | 15 +++ packages/powertools-batch/tsconfig.cjs.json | 10 ++ packages/powertools-batch/tsconfig.dev.json | 20 +++ packages/powertools-batch/tsconfig.esm.json | 10 ++ .../powertools-batch/tsconfig.examples.json | 17 +++ packages/powertools-batch/tsconfig.json | 16 +++ packages/powertools-batch/tsconfig.src.json | 12 ++ packages/powertools-batch/vitest.config.ts | 6 + pnpm-lock.yaml | 39 +++++- pnpm-workspace.yaml | 1 + tsconfig.base.json | 9 ++ tsconfig.build.json | 3 + tsconfig.json | 3 + vitest.shared.ts | 1 + 30 files changed, 685 insertions(+), 4 deletions(-) create mode 100644 packages/powertools-batch/.gitattributes create mode 100644 packages/powertools-batch/.gitignore create mode 100644 packages/powertools-batch/.npmignore create mode 100644 packages/powertools-batch/.projen/deps.json create mode 100644 packages/powertools-batch/.projen/files.json create mode 100644 packages/powertools-batch/.projen/tasks.json create mode 100644 packages/powertools-batch/LICENSE create mode 100644 packages/powertools-batch/README.md create mode 100644 packages/powertools-batch/docgen.json create mode 100644 packages/powertools-batch/examples/example.ts create mode 100644 packages/powertools-batch/package.json create mode 100644 packages/powertools-batch/src/batch-processor.ts create mode 100644 packages/powertools-batch/src/constants.ts create mode 100644 packages/powertools-batch/src/index.ts create mode 100644 packages/powertools-batch/src/process-partial-response.ts create mode 100644 packages/powertools-batch/src/processor.ts create mode 100644 packages/powertools-batch/tsconfig.cjs.json create mode 100644 packages/powertools-batch/tsconfig.dev.json create mode 100644 packages/powertools-batch/tsconfig.esm.json create mode 100644 packages/powertools-batch/tsconfig.examples.json create mode 100644 packages/powertools-batch/tsconfig.json create mode 100644 packages/powertools-batch/tsconfig.src.json create mode 100644 packages/powertools-batch/vitest.config.ts diff --git a/.projenrc.ts b/.projenrc.ts index b6efd77a..825be304 100644 --- a/.projenrc.ts +++ b/.projenrc.ts @@ -149,6 +149,19 @@ const tracer = new TypeScriptLibProject({ addExamples: true, }); +new TypeScriptLibProject({ + parent: project, + name: "powertools-batch", + description: "Effectful AWS Lambda Powertools Batch Processing", + devDeps: [ + ...effectDeps, + `${lambda.package.packageName}@workspace:^`, + "@types/aws-lambda", + ], + peerDeps: commonPeerDeps, + addExamples: true, +}); + tracer.tsconfigExamples?.file.addToArray( "references", { path: path.relative(tracer.outdir, lambda.outdir) }, diff --git a/packages/powertools-batch/.gitattributes b/packages/powertools-batch/.gitattributes new file mode 100644 index 00000000..1a1cfd32 --- /dev/null +++ b/packages/powertools-batch/.gitattributes @@ -0,0 +1,22 @@ +# ~~ Generated by projen. To modify, edit .projenrc.js and run "npx projen". + +* text=auto eol=lf +/.gitattributes linguist-generated +/.gitignore linguist-generated +/.npmignore linguist-generated +/.npmrc linguist-generated +/.projen/** linguist-generated +/.projen/deps.json linguist-generated +/.projen/files.json linguist-generated +/.projen/tasks.json linguist-generated +/docgen.json linguist-generated +/LICENSE linguist-generated +/package.json linguist-generated +/pnpm-lock.yaml linguist-generated +/tsconfig.cjs.json linguist-generated +/tsconfig.dev.json linguist-generated +/tsconfig.esm.json linguist-generated +/tsconfig.examples.json linguist-generated +/tsconfig.json linguist-generated +/tsconfig.src.json linguist-generated +/vitest.config.ts linguist-generated \ No newline at end of file diff --git a/packages/powertools-batch/.gitignore b/packages/powertools-batch/.gitignore new file mode 100644 index 00000000..b4315b2a --- /dev/null +++ b/packages/powertools-batch/.gitignore @@ -0,0 +1,45 @@ +# ~~ Generated by projen. To modify, edit .projenrc.js and run "npx projen". +!/.gitattributes +!/.projen/tasks.json +!/.projen/deps.json +!/.projen/files.json +!/package.json +!/LICENSE +!/.npmignore +logs +*.log +npm-debug.log* +yarn-debug.log* +yarn-error.log* +lerna-debug.log* +report.[0-9]*.[0-9]*.[0-9]*.[0-9]*.json +pids +*.pid +*.seed +*.pid.lock +lib-cov +coverage +*.lcov +.nyc_output +build/Release +node_modules/ +jspm_packages/ +*.tsbuildinfo +.eslintcache +*.tgz +.yarn-integrity +.cache +!/.npmrc +!/test/ +!/tsconfig.json +!/src/ +/build +/dist/ +!/tsconfig.src.json +!/tsconfig.dev.json +!/tsconfig.examples.json +!/tsconfig.esm.json +!/tsconfig.cjs.json +!/docgen.json +docs/ +!/vitest.config.ts diff --git a/packages/powertools-batch/.npmignore b/packages/powertools-batch/.npmignore new file mode 100644 index 00000000..74fb267c --- /dev/null +++ b/packages/powertools-batch/.npmignore @@ -0,0 +1,20 @@ +# ~~ Generated by projen. To modify, edit .projenrc.js and run "npx projen". +/.projen/ +/test/ +/src/ +!/build/ +!/build/**/*.js +!/build/**/*.d.ts +dist +/tsconfig.json +/.github/ +/.vscode/ +/.idea/ +/.projenrc.js +tsconfig.tsbuildinfo +/tsconfig.src.json +/tsconfig.dev.json +/tsconfig.examples.json +/tsconfig.esm.json +/tsconfig.cjs.json +/.gitattributes diff --git a/packages/powertools-batch/.projen/deps.json b/packages/powertools-batch/.projen/deps.json new file mode 100644 index 00000000..0fe69d0a --- /dev/null +++ b/packages/powertools-batch/.projen/deps.json @@ -0,0 +1,33 @@ +{ + "dependencies": [ + { + "name": "@effect-aws/lambda", + "version": "workspace:^", + "type": "build" + }, + { + "name": "@types/aws-lambda", + "type": "build" + }, + { + "name": "@types/node", + "version": "ts5.4", + "type": "build" + }, + { + "name": "effect", + "type": "build" + }, + { + "name": "typescript", + "version": "^5.4.2", + "type": "build" + }, + { + "name": "effect", + "version": ">=3.0.4 <4.0.0", + "type": "peer" + } + ], + "//": "~~ Generated by projen. To modify, edit .projenrc.js and run \"npx projen\"." +} diff --git a/packages/powertools-batch/.projen/files.json b/packages/powertools-batch/.projen/files.json new file mode 100644 index 00000000..cf98c2f0 --- /dev/null +++ b/packages/powertools-batch/.projen/files.json @@ -0,0 +1,20 @@ +{ + "files": [ + ".gitattributes", + ".gitignore", + ".npmignore", + ".projen/deps.json", + ".projen/files.json", + ".projen/tasks.json", + "docgen.json", + "LICENSE", + "tsconfig.cjs.json", + "tsconfig.dev.json", + "tsconfig.esm.json", + "tsconfig.examples.json", + "tsconfig.json", + "tsconfig.src.json", + "vitest.config.ts" + ], + "//": "~~ Generated by projen. To modify, edit .projenrc.js and run \"npx projen\"." +} diff --git a/packages/powertools-batch/.projen/tasks.json b/packages/powertools-batch/.projen/tasks.json new file mode 100644 index 00000000..ad983f13 --- /dev/null +++ b/packages/powertools-batch/.projen/tasks.json @@ -0,0 +1,120 @@ +{ + "tasks": { + "build": { + "name": "build", + "description": "Full release build", + "steps": [ + { + "spawn": "pre-compile" + }, + { + "spawn": "compile" + }, + { + "spawn": "post-compile" + }, + { + "spawn": "test" + }, + { + "spawn": "package" + } + ] + }, + "compile": { + "name": "compile", + "description": "Only compile", + "steps": [ + { + "exec": "tsc -b ./tsconfig.cjs.json ./tsconfig.esm.json" + } + ] + }, + "default": { + "name": "default", + "description": "Synthesize project files" + }, + "eslint": { + "name": "eslint", + "description": "Runs eslint against the codebase", + "steps": [ + { + "exec": "eslint $@ src test", + "receiveArgs": true + } + ] + }, + "install": { + "name": "install", + "description": "Install project dependencies and update lockfile (non-frozen)", + "steps": [ + { + "exec": "pnpm i --no-frozen-lockfile" + } + ] + }, + "install:ci": { + "name": "install:ci", + "description": "Install project dependencies using frozen lockfile", + "steps": [ + { + "exec": "pnpm i --frozen-lockfile" + } + ] + }, + "package": { + "name": "package", + "description": "Creates the distribution package", + "steps": [ + { + "exec": "build-utils pack-v2" + } + ] + }, + "post-compile": { + "name": "post-compile", + "description": "Runs after successful compilation" + }, + "pre-compile": { + "name": "pre-compile", + "description": "Prepare the project for compilation", + "steps": [ + { + "spawn": "eslint" + } + ] + }, + "test": { + "name": "test", + "description": "Run tests", + "steps": [ + { + "exec": "vitest run --reporter verbose", + "receiveArgs": true + } + ] + }, + "test:watch": { + "name": "test:watch", + "description": "Run tests in watch mode", + "steps": [ + { + "exec": "vitest --reporter verbose" + } + ] + }, + "watch": { + "name": "watch", + "description": "Watch & compile in the background", + "steps": [ + { + "exec": "tsc --build -w" + } + ] + } + }, + "env": { + "PATH": "$(pnpm -c exec \"node --print process.env.PATH\")" + }, + "//": "~~ Generated by projen. To modify, edit .projenrc.js and run \"npx projen\"." +} diff --git a/packages/powertools-batch/LICENSE b/packages/powertools-batch/LICENSE new file mode 100644 index 00000000..ced0788c --- /dev/null +++ b/packages/powertools-batch/LICENSE @@ -0,0 +1,19 @@ +Copyright (c) 2025 Victor Korzunin + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/packages/powertools-batch/README.md b/packages/powertools-batch/README.md new file mode 100644 index 00000000..a72414dd --- /dev/null +++ b/packages/powertools-batch/README.md @@ -0,0 +1,77 @@ +# @effect-aws/powertools-batch + +Process batch of records in AWS Lambda using [Effect](https://www.effect.website/). Inspired by [`@aws-lambda-powertools/batch`](https://docs.powertools.aws.dev/lambda/typescript/latest/features/batch) package. + +[![npm version](https://img.shields.io/npm/v/%40effect-aws%2Fpowertools-batch?color=brightgreen&label=npm%20package)](https://www.npmjs.com/package/@effect-aws/powertools-batch) +[![npm downloads](https://img.shields.io/npm/dm/%40effect-aws%2Fpowertools-batch)](https://www.npmjs.com/package/@effect-aws/powertools-batch) + +## Installation + +```bash +npm install --save @effect-aws/powertools-batch +``` + +## Usage + +With Kinesis Data Streams: + +```typescript +import { LambdaHandler } from "@effect-aws/lambda"; +import { BatchProcessor, EventType, processPartialResponse } from "@effect-aws/powertools-batch"; +import type { KinesisStreamEvent, KinesisStreamRecord } from "aws-lambda"; +import { Effect } from "effect"; + +const recordHandler = (record: KinesisStreamRecord) => + Effect.gen(function*() { + return yield* Effect.logInfo(`Processing record: ${record.kinesis.data}`); + }); + +const HandlerLive = BatchProcessor({ eventType: EventType.KinesisDataStreams }); + +export const handler = LambdaHandler.make({ + handler: (event: KinesisStreamEvent) => processPartialResponse(event, recordHandler), + layer: HandlerLive, +}); +``` + +With DynamoDB Streams: + +```typescript +import { LambdaHandler } from "@effect-aws/lambda"; +import { BatchProcessor, EventType, processPartialResponse } from "@effect-aws/powertools-batch"; +import type { DynamoDBStreamEvent, DynamoDBStreamRecord } from "aws-lambda"; +import { Effect } from "effect"; + +const recordHandler = (record: DynamoDBStreamRecord) => + Effect.gen(function*() { + return yield* Effect.logInfo(`Processing record: ${record.dynamodb.NewImage}`); + }); + +const HandlerLive = BatchProcessor({ eventType: EventType.DynamoDBStreams }); + +export const handler = LambdaHandler.make({ + handler: (event: DynamoDBStreamEvent) => processPartialResponse(event, recordHandler), + layer: HandlerLive, +}); +``` + +With SQS: + +```typescript +import { LambdaHandler } from "@effect-aws/lambda"; +import { BatchProcessor, EventType, processPartialResponse } from "@effect-aws/powertools-batch"; +import type { SQSEvent, SQSRecord } from "aws-lambda"; +import { Effect } from "effect"; + +const recordHandler = (record: SQSRecord) => + Effect.gen(function*() { + return yield* Effect.logInfo(`Processing record: ${record.body}`); + }); + +const HandlerLive = BatchProcessor({ eventType: EventType.SQS }); + +export const handler = LambdaHandler.make({ + handler: (event: SQSEvent) => processPartialResponse(event, recordHandler), + layer: HandlerLive, +}); +``` diff --git a/packages/powertools-batch/docgen.json b/packages/powertools-batch/docgen.json new file mode 100644 index 00000000..cc12dbc6 --- /dev/null +++ b/packages/powertools-batch/docgen.json @@ -0,0 +1,8 @@ +{ + "$schema": "../../node_modules/@effect/docgen/schema.json", + "exclude": [ + "src/internal/**/*.ts", + "src/Errors.ts" + ], + "//": "~~ Generated by projen. To modify, edit .projenrc.js and run \"npx projen\"." +} diff --git a/packages/powertools-batch/examples/example.ts b/packages/powertools-batch/examples/example.ts new file mode 100644 index 00000000..e2c5e929 --- /dev/null +++ b/packages/powertools-batch/examples/example.ts @@ -0,0 +1,16 @@ +import { LambdaHandler } from "@effect-aws/lambda"; +import type { KinesisStreamEvent, KinesisStreamRecord } from "aws-lambda"; +import { Effect } from "effect"; +import { BatchProcessor, EventType, processPartialResponse } from "../src/index.js"; + +const recordHandler = (record: KinesisStreamRecord) => + Effect.gen(function*() { + return yield* Effect.logInfo(`Processing record: ${record.kinesis.data}`); + }); + +const HandlerLive = BatchProcessor({ eventType: EventType.KinesisDataStreams }); + +export const handler = LambdaHandler.make({ + handler: (event: KinesisStreamEvent) => processPartialResponse(event, recordHandler), + layer: HandlerLive, +}); diff --git a/packages/powertools-batch/package.json b/packages/powertools-batch/package.json new file mode 100644 index 00000000..7ab2cc3c --- /dev/null +++ b/packages/powertools-batch/package.json @@ -0,0 +1,50 @@ +{ + "name": "@effect-aws/powertools-batch", + "description": "Effectful AWS Lambda Powertools Batch Processing", + "repository": { + "type": "git", + "url": "github:floydspace/effect-aws", + "directory": "packages/powertools-batch" + }, + "scripts": { + "build": "npx projen build", + "compile": "npx projen compile", + "default": "npx projen default", + "eslint": "npx projen eslint", + "package": "npx projen package", + "post-compile": "npx projen post-compile", + "pre-compile": "npx projen pre-compile", + "test": "npx projen test", + "test:watch": "npx projen test:watch", + "watch": "npx projen watch", + "docgen": "docgen" + }, + "author": { + "name": "Victor Korzunin", + "email": "ifloydrose@gmail.com", + "organization": false + }, + "devDependencies": { + "@effect-aws/lambda": "workspace:^", + "@types/aws-lambda": "^8.10.149", + "@types/node": "ts5.4", + "effect": "^3.16.4", + "typescript": "^5.4.2" + }, + "peerDependencies": { + "effect": ">=3.0.4 <4.0.0" + }, + "main": "build/cjs/index.js", + "license": "MIT", + "homepage": "https://floydspace.github.io/effect-aws/docs/powertools-batch", + "publishConfig": { + "access": "public", + "directory": "dist" + }, + "version": "0.0.0", + "types": "build/dts/index.d.ts", + "type": "module", + "module": "build/esm/index.js", + "sideEffects": [], + "//": "~~ Generated by projen. To modify, edit .projenrc.js and run \"npx projen\"." +} diff --git a/packages/powertools-batch/src/batch-processor.ts b/packages/powertools-batch/src/batch-processor.ts new file mode 100644 index 00000000..28385923 --- /dev/null +++ b/packages/powertools-batch/src/batch-processor.ts @@ -0,0 +1,59 @@ +import type { DynamoDBRecord, KinesisStreamRecord, SQSRecord } from "aws-lambda"; +import { Array, Effect, Either, Layer, Option, Schema } from "effect"; +import type { BaseRecord } from "./constants.js"; +import { EventType } from "./constants.js"; +import { FullBatchError, Processor } from "./processor.js"; + +class BatchProcessError extends Schema.TaggedError()("BatchProcessError", { + record: Schema.Unknown, + cause: Schema.Defect, +}) {} + +const identifierMapper = (record: BaseRecord, eventType: EventType): string => { + switch (eventType) { + case EventType.SQS: + return (record as SQSRecord).messageId; + case EventType.KinesisDataStreams: + return (record as KinesisStreamRecord).kinesis.sequenceNumber; + case EventType.DynamoDBStreams: + return (record as DynamoDBRecord).dynamodb?.SequenceNumber ?? ""; + } +}; + +export interface BatchProcessorOptions { + eventType: EventType; +} + +export const BatchProcessor = (options: BatchProcessorOptions) => + Layer.succeed( + Processor, + Processor.of({ + process: ( + event: { Records: Array }, + handler: (record: Record) => Effect.Effect, + ) => + Effect.gen(function*() { + const effects = event.Records.map((record) => + handler(record).pipe(Effect.mapError((e) => new BatchProcessError({ record, cause: e }))) + ); + + const results = yield* Effect.all(effects, { mode: "either" }); + + const batchItemFailures = Array.filterMap(results, (result) => { + if (Either.isLeft(result)) { + return Option.some({ + itemIdentifier: identifierMapper(result.left.record as BaseRecord, options.eventType), + }); + } + + return Option.none(); + }); + + if (batchItemFailures.length === event.Records.length) { + return yield* Effect.fail(new FullBatchError()); + } + + return { batchItemFailures }; + }), + }), + ); diff --git a/packages/powertools-batch/src/constants.ts b/packages/powertools-batch/src/constants.ts new file mode 100644 index 00000000..a434e058 --- /dev/null +++ b/packages/powertools-batch/src/constants.ts @@ -0,0 +1,9 @@ +import type { DynamoDBRecord, KinesisStreamRecord, SQSRecord } from "aws-lambda"; + +export type BaseRecord = { [key: string]: unknown } | SQSRecord | KinesisStreamRecord | DynamoDBRecord; + +export enum EventType { + SQS = "SQS", + KinesisDataStreams = "KinesisDataStreams", + DynamoDBStreams = "DynamoDBStreams", +} diff --git a/packages/powertools-batch/src/index.ts b/packages/powertools-batch/src/index.ts new file mode 100644 index 00000000..0b172ac5 --- /dev/null +++ b/packages/powertools-batch/src/index.ts @@ -0,0 +1,4 @@ +export * from "./batch-processor.js"; +export * from "./constants.js"; +export * from "./process-partial-response.js"; +export * from "./processor.js"; diff --git a/packages/powertools-batch/src/process-partial-response.ts b/packages/powertools-batch/src/process-partial-response.ts new file mode 100644 index 00000000..ae4567a9 --- /dev/null +++ b/packages/powertools-batch/src/process-partial-response.ts @@ -0,0 +1,12 @@ +import { Effect } from "effect"; +import type { BaseRecord } from "./constants.js"; +import { Processor } from "./processor.js"; + +export const processPartialResponse = ( + event: { Records: Array }, + recordHandler: (record: Record) => Effect.Effect, +) => + Effect.gen(function*() { + const processor = yield* Processor; + return yield* processor.process(event, recordHandler); + }); diff --git a/packages/powertools-batch/src/processor.ts b/packages/powertools-batch/src/processor.ts new file mode 100644 index 00000000..50102262 --- /dev/null +++ b/packages/powertools-batch/src/processor.ts @@ -0,0 +1,15 @@ +import { Context, Schema } from "effect"; +import type { Effect } from "effect"; +import type { BaseRecord } from "./constants.js"; + +export class FullBatchError extends Schema.TaggedError()("FullBatchError", {}) {} + +export class Processor extends Context.Tag("@effect-aws/powertools-batch/Processor")< + Processor, + { + readonly process: ( + event: { Records: Array }, + handler: (record: Record) => Effect.Effect, + ) => Effect.Effect<{ batchItemFailures: Array<{ itemIdentifier: string }> }, FullBatchError, R>; + } +>() {} diff --git a/packages/powertools-batch/tsconfig.cjs.json b/packages/powertools-batch/tsconfig.cjs.json new file mode 100644 index 00000000..5d9330be --- /dev/null +++ b/packages/powertools-batch/tsconfig.cjs.json @@ -0,0 +1,10 @@ +// ~~ Generated by projen. To modify, edit .projenrc.js and run "npx projen". +{ + "extends": "./tsconfig.src.json", + "compilerOptions": { + "tsBuildInfoFile": ".tsbuildinfo/cjs.tsbuildinfo", + "outDir": "build/cjs", + "moduleResolution": "node", + "module": "CommonJS" + } +} diff --git a/packages/powertools-batch/tsconfig.dev.json b/packages/powertools-batch/tsconfig.dev.json new file mode 100644 index 00000000..f97384ef --- /dev/null +++ b/packages/powertools-batch/tsconfig.dev.json @@ -0,0 +1,20 @@ +// ~~ Generated by projen. To modify, edit .projenrc.js and run "npx projen". +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "tsBuildInfoFile": ".tsbuildinfo/test.tsbuildinfo", + "noEmit": true, + "rootDir": "test", + "types": [ + "../../vitest.d.ts" + ] + }, + "include": [ + "test" + ], + "references": [ + { + "path": "tsconfig.src.json" + } + ] +} diff --git a/packages/powertools-batch/tsconfig.esm.json b/packages/powertools-batch/tsconfig.esm.json new file mode 100644 index 00000000..0725cdeb --- /dev/null +++ b/packages/powertools-batch/tsconfig.esm.json @@ -0,0 +1,10 @@ +// ~~ Generated by projen. To modify, edit .projenrc.js and run "npx projen". +{ + "extends": "./tsconfig.src.json", + "compilerOptions": { + "tsBuildInfoFile": ".tsbuildinfo/esm.tsbuildinfo", + "outDir": "build/esm", + "declarationDir": "build/dts", + "stripInternal": true + } +} diff --git a/packages/powertools-batch/tsconfig.examples.json b/packages/powertools-batch/tsconfig.examples.json new file mode 100644 index 00000000..5b03e9f4 --- /dev/null +++ b/packages/powertools-batch/tsconfig.examples.json @@ -0,0 +1,17 @@ +// ~~ Generated by projen. To modify, edit .projenrc.js and run "npx projen". +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "tsBuildInfoFile": ".tsbuildinfo/examples.tsbuildinfo", + "noEmit": true, + "rootDir": "examples" + }, + "include": [ + "examples" + ], + "references": [ + { + "path": "tsconfig.src.json" + } + ] +} diff --git a/packages/powertools-batch/tsconfig.json b/packages/powertools-batch/tsconfig.json new file mode 100644 index 00000000..8339ee1b --- /dev/null +++ b/packages/powertools-batch/tsconfig.json @@ -0,0 +1,16 @@ +// ~~ Generated by projen. To modify, edit .projenrc.js and run "npx projen". +{ + "extends": "../../tsconfig.base.json", + "include": [], + "references": [ + { + "path": "tsconfig.src.json" + }, + { + "path": "tsconfig.dev.json" + }, + { + "path": "tsconfig.examples.json" + } + ] +} diff --git a/packages/powertools-batch/tsconfig.src.json b/packages/powertools-batch/tsconfig.src.json new file mode 100644 index 00000000..dd8b4304 --- /dev/null +++ b/packages/powertools-batch/tsconfig.src.json @@ -0,0 +1,12 @@ +// ~~ Generated by projen. To modify, edit .projenrc.js and run "npx projen". +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "tsBuildInfoFile": ".tsbuildinfo/src.tsbuildinfo", + "outDir": "build/src", + "rootDir": "src" + }, + "include": [ + "src" + ] +} diff --git a/packages/powertools-batch/vitest.config.ts b/packages/powertools-batch/vitest.config.ts new file mode 100644 index 00000000..2cf045fa --- /dev/null +++ b/packages/powertools-batch/vitest.config.ts @@ -0,0 +1,6 @@ +import { mergeConfig, type UserConfigExport } from "vitest/config"; +import configShared from "../../vitest.shared.js"; + +const config: UserConfigExport = {}; + +export default mergeConfig(configShared, config); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 103208bd..d9289f4d 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1289,6 +1289,25 @@ importers: version: 5.4.5 publishDirectory: dist + packages/powertools-batch: + devDependencies: + '@effect-aws/lambda': + specifier: workspace:^ + version: link:../lambda/dist + '@types/aws-lambda': + specifier: ^8.10.149 + version: 8.10.149 + '@types/node': + specifier: ts5.4 + version: 24.3.0 + effect: + specifier: ^3.16.4 + version: 3.16.4 + typescript: + specifier: ^5.4.2 + version: 5.4.5 + publishDirectory: dist + packages/powertools-logger: devDependencies: '@aws-lambda-powertools/commons': @@ -3244,6 +3263,9 @@ packages: '@types/node@24.1.0': resolution: {integrity: sha512-ut5FthK5moxFKH2T1CUOC6ctR67rQRvvHdFLCD2Ql6KXmMuCrjsSsRI9UsLCm9M18BMwClv4pn327UvB7eeO1w==} + '@types/node@24.3.0': + resolution: {integrity: sha512-aPTXCrfwnDLj4VvXrm+UUCQjNEvJgNA8s5F1cvwQU+3KNltTOkBm1j30uNLyqqPNe7gE3KFzImYoZEfLhp4Yow==} + '@types/normalize-package-data@2.4.4': resolution: {integrity: sha512-37i+OaWTh9qeK4LSHPsyRC7NahnGotNuZvjLSgcPzblpHB3rrCJxAOgI5gCdKm7coonsaX1Of0ILiTcnZjbfxA==} @@ -5488,6 +5510,9 @@ packages: undici-types@6.19.8: resolution: {integrity: sha512-ve2KP6f/JnbPBFyobGHuerC9g1FYGn/F8n1LWTwNxCEzd6IfqTwUQcNXgEtmmQ6DlRrC1hrSrBnCZPokRrDHjw==} + undici-types@7.10.0: + resolution: {integrity: sha512-t5Fy/nfn+14LuOc2KNYg75vZqClpAiqscVvMygNnlsHBFpSXdJaYtXMcdNLpl/Qvc3P2cB3s6lOV51nqsFq4ag==} + undici-types@7.8.0: resolution: {integrity: sha512-9UJ2xGDvQ43tYyVMpuHlsgApydB8ZKfVYTsLDhXkFL/6gfkp+U8xTGdh8pMJv1SpZna0zxG1DwsKZsreLbXBxw==} @@ -9512,7 +9537,7 @@ snapshots: '@jest/schemas': 29.6.3 '@types/istanbul-lib-coverage': 2.0.6 '@types/istanbul-reports': 3.0.4 - '@types/node': 24.1.0 + '@types/node': 24.3.0 '@types/yargs': 17.0.33 chalk: 4.1.2 @@ -10144,7 +10169,7 @@ snapshots: '@types/cls-hooked@4.3.9': dependencies: - '@types/node': 24.1.0 + '@types/node': 24.3.0 '@types/dedent@0.7.0': {} @@ -10158,7 +10183,7 @@ snapshots: '@types/glob@7.1.3': dependencies: '@types/minimatch': 5.1.2 - '@types/node': 24.1.0 + '@types/node': 24.3.0 '@types/hast@3.0.4': dependencies: @@ -10211,6 +10236,10 @@ snapshots: dependencies: undici-types: 7.8.0 + '@types/node@24.3.0': + dependencies: + undici-types: 7.10.0 + '@types/normalize-package-data@2.4.4': {} '@types/sinon@17.0.4': @@ -11726,7 +11755,7 @@ snapshots: jest-util@29.7.0: dependencies: '@jest/types': 29.6.3 - '@types/node': 24.1.0 + '@types/node': 24.3.0 chalk: 4.1.2 ci-info: 3.9.0 graceful-fs: 4.2.11 @@ -12731,6 +12760,8 @@ snapshots: undici-types@6.19.8: {} + undici-types@7.10.0: {} + undici-types@7.8.0: {} undici@7.10.0: {} diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 83608099..6cff6893 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -58,6 +58,7 @@ packages: - packages/http-handler - packages/lambda - packages/lib-dynamodb + - packages/powertools-batch - packages/powertools-logger - packages/powertools-tracer - packages/s3 diff --git a/tsconfig.base.json b/tsconfig.base.json index 353cb515..949af6e2 100644 --- a/tsconfig.base.json +++ b/tsconfig.base.json @@ -561,6 +561,15 @@ "@effect-aws/powertools-tracer/test/*": [ "./packages/powertools-tracer/test/*.js" ], + "@effect-aws/powertools-batch": [ + "./packages/powertools-batch/src/index.js" + ], + "@effect-aws/powertools-batch/*": [ + "./packages/powertools-batch/src/*.js" + ], + "@effect-aws/powertools-batch/test/*": [ + "./packages/powertools-batch/test/*.js" + ], "@effect-aws/secrets-manager": [ "./packages/secrets-manager/src/index.js" ], diff --git a/tsconfig.build.json b/tsconfig.build.json index 740a4983..8030cf71 100644 --- a/tsconfig.build.json +++ b/tsconfig.build.json @@ -174,6 +174,9 @@ { "path": "packages/powertools-tracer/tsconfig.esm.json" }, + { + "path": "packages/powertools-batch/tsconfig.esm.json" + }, { "path": "packages/secrets-manager/tsconfig.esm.json" }, diff --git a/tsconfig.json b/tsconfig.json index 51c52b64..418d6ae5 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -179,6 +179,9 @@ { "path": "packages/powertools-tracer" }, + { + "path": "packages/powertools-batch" + }, { "path": "packages/secrets-manager" }, diff --git a/vitest.shared.ts b/vitest.shared.ts index 352d158d..d033ad2d 100644 --- a/vitest.shared.ts +++ b/vitest.shared.ts @@ -75,6 +75,7 @@ const config: UserConfig = { ...alias("lambda"), ...alias("powertools-logger"), ...alias("powertools-tracer"), + ...alias("powertools-batch"), ...alias("secrets-manager"), ...alias("ssm"), ...alias("s3"), From 345576c1109761049b3428e463b0b069999748da Mon Sep 17 00:00:00 2001 From: Gabriel Colson Date: Wed, 3 Sep 2025 20:32:32 +0200 Subject: [PATCH 2/3] chore: add tests --- .../powertools-batch/src/batch-processor.ts | 2 +- .../test/batch-processor.test.ts | 419 ++++++++++++++++++ 2 files changed, 420 insertions(+), 1 deletion(-) create mode 100644 packages/powertools-batch/test/batch-processor.test.ts diff --git a/packages/powertools-batch/src/batch-processor.ts b/packages/powertools-batch/src/batch-processor.ts index 28385923..8b3ee2fd 100644 --- a/packages/powertools-batch/src/batch-processor.ts +++ b/packages/powertools-batch/src/batch-processor.ts @@ -49,7 +49,7 @@ export const BatchProcessor = (options: BatchProcessorOptions) => return Option.none(); }); - if (batchItemFailures.length === event.Records.length) { + if (batchItemFailures.length === event.Records.length && event.Records.length > 0) { return yield* Effect.fail(new FullBatchError()); } diff --git a/packages/powertools-batch/test/batch-processor.test.ts b/packages/powertools-batch/test/batch-processor.test.ts new file mode 100644 index 00000000..caabc84d --- /dev/null +++ b/packages/powertools-batch/test/batch-processor.test.ts @@ -0,0 +1,419 @@ +import { BatchProcessor, EventType, processPartialResponse } from "@effect-aws/powertools-batch"; +import type { DynamoDBRecord, KinesisStreamRecord, SQSRecord } from "aws-lambda"; +import { Effect, Either, Layer } from "effect"; +import { describe, expect, it } from "vitest"; + +describe("BatchProcessor", () => { + describe("SQS Event Processing", () => { + const createSQSRecord = (messageId: string, body: string): SQSRecord => ({ + messageId, + receiptHandle: `receipt-${messageId}`, + body, + attributes: { + ApproximateReceiveCount: "1", + SentTimestamp: "1695820800000", + SenderId: "123456789012", + ApproximateFirstReceiveTimestamp: "1695820800000", + }, + messageAttributes: {}, + md5OfBody: "test-md5", + eventSource: "aws:sqs", + eventSourceARN: "arn:aws:sqs:us-east-1:123456789012:test-queue", + awsRegion: "us-east-1", + }); + + it("should process all SQS records successfully", async () => { + const records = [ + createSQSRecord("msg-1", "message 1"), + createSQSRecord("msg-2", "message 2"), + createSQSRecord("msg-3", "message 3"), + ]; + + const handler = (record: SQSRecord) => Effect.succeed({ messageId: record.messageId, processed: true }); + + const layer = BatchProcessor({ eventType: EventType.SQS }); + + const result = await processPartialResponse({ Records: records }, handler).pipe( + Effect.provide(layer), + Effect.runPromise, + ); + + expect(result).toEqual({ batchItemFailures: [] }); + }); + + it("should handle partial failures for SQS", async () => { + const records = [ + createSQSRecord("msg-1", "message 1"), + createSQSRecord("msg-2", "message 2"), + createSQSRecord("msg-3", "message 3"), + ]; + + const handler = (record: SQSRecord) => { + if (record.messageId === "msg-2") { + return Effect.fail(new Error("Processing failed")); + } + return Effect.succeed({ messageId: record.messageId, processed: true }); + }; + + const layer = BatchProcessor({ eventType: EventType.SQS }); + + const result = await processPartialResponse({ Records: records }, handler).pipe( + Effect.provide(layer), + Effect.runPromise, + ); + + expect(result).toEqual({ + batchItemFailures: [{ itemIdentifier: "msg-2" }], + }); + }); + + it("should handle full batch failure for SQS", async () => { + const records = [ + createSQSRecord("msg-1", "message 1"), + createSQSRecord("msg-2", "message 2"), + ]; + + const handler = (_: SQSRecord) => Effect.fail(new Error("Processing failed")); + + const layer = BatchProcessor({ eventType: EventType.SQS }); + + const result = await processPartialResponse({ Records: records }, handler).pipe( + Effect.provide(layer), + Effect.either, + Effect.runPromise, + ); + + expect(Either.isLeft(result)).toBe(true); + if (Either.isLeft(result)) { + expect(result.left._tag).toBe("FullBatchError"); + } + }); + }); + + describe("Kinesis Event Processing", () => { + const createKinesisRecord = (sequenceNumber: string, data: string): KinesisStreamRecord => ({ + kinesis: { + sequenceNumber, + approximateArrivalTimestamp: 1695820800, + data, + partitionKey: "partition-1", + kinesisSchemaVersion: "1.0", + }, + eventSource: "aws:kinesis", + eventSourceARN: "arn:aws:kinesis:us-east-1:123456789012:stream/test-stream", + eventID: `shardId-000000000000:${sequenceNumber}`, + eventName: "aws:kinesis:record", + invokeIdentityArn: "arn:aws:iam::123456789012:role/test-role", + awsRegion: "us-east-1", + eventVersion: "1.0", + }); + + it("should process all Kinesis records successfully", async () => { + const records = [ + createKinesisRecord("1000", "data-1"), + createKinesisRecord("2000", "data-2"), + createKinesisRecord("3000", "data-3"), + ]; + + const handler = (record: KinesisStreamRecord) => + Effect.succeed({ sequenceNumber: record.kinesis.sequenceNumber, processed: true }); + + const layer = BatchProcessor({ eventType: EventType.KinesisDataStreams }); + + const result = await processPartialResponse({ Records: records }, handler).pipe( + Effect.provide(layer), + Effect.runPromise, + ); + + expect(result).toEqual({ batchItemFailures: [] }); + }); + + it("should handle partial failures for Kinesis", async () => { + const records = [ + createKinesisRecord("1000", "data-1"), + createKinesisRecord("2000", "data-2"), + createKinesisRecord("3000", "data-3"), + ]; + + const handler = (record: KinesisStreamRecord) => { + if (record.kinesis.sequenceNumber === "2000") { + return Effect.fail(new Error("Processing failed")); + } + return Effect.succeed({ sequenceNumber: record.kinesis.sequenceNumber, processed: true }); + }; + + const layer = BatchProcessor({ eventType: EventType.KinesisDataStreams }); + + const result = await processPartialResponse({ Records: records }, handler).pipe( + Effect.provide(layer), + Effect.runPromise, + ); + + expect(result).toEqual({ + batchItemFailures: [{ itemIdentifier: "2000" }], + }); + }); + }); + + describe("DynamoDB Streams Event Processing", () => { + const createDynamoDBRecord = ( + sequenceNumber: string, + eventName: "INSERT" | "MODIFY" | "REMOVE", + ): DynamoDBRecord => ({ + eventID: `event-${sequenceNumber}`, + eventName, + eventVersion: "1.1", + eventSource: "aws:dynamodb", + awsRegion: "us-east-1", + dynamodb: { + SequenceNumber: sequenceNumber, + SizeBytes: 100, + StreamViewType: "NEW_AND_OLD_IMAGES", + }, + eventSourceARN: "arn:aws:dynamodb:us-east-1:123456789012:table/test-table/stream/2023-01-01T00:00:00.000", + }); + + it("should process all DynamoDB records successfully", async () => { + const records = [ + createDynamoDBRecord("100", "INSERT"), + createDynamoDBRecord("200", "MODIFY"), + createDynamoDBRecord("300", "REMOVE"), + ]; + + const handler = (record: DynamoDBRecord) => + Effect.succeed({ + sequenceNumber: record.dynamodb?.SequenceNumber, + eventName: record.eventName, + processed: true, + }); + + const layer = BatchProcessor({ eventType: EventType.DynamoDBStreams }); + + const result = await processPartialResponse({ Records: records }, handler).pipe( + Effect.provide(layer), + Effect.runPromise, + ); + + expect(result).toEqual({ batchItemFailures: [] }); + }); + + it("should handle partial failures for DynamoDB Streams", async () => { + const records = [ + createDynamoDBRecord("100", "INSERT"), + createDynamoDBRecord("200", "MODIFY"), + createDynamoDBRecord("300", "REMOVE"), + ]; + + const handler = (record: DynamoDBRecord) => { + if (record.dynamodb?.SequenceNumber === "200") { + return Effect.fail(new Error("Processing failed")); + } + return Effect.succeed({ + sequenceNumber: record.dynamodb?.SequenceNumber, + eventName: record.eventName, + processed: true, + }); + }; + + const layer = BatchProcessor({ eventType: EventType.DynamoDBStreams }); + + const result = await processPartialResponse({ Records: records }, handler).pipe( + Effect.provide(layer), + Effect.runPromise, + ); + + expect(result).toEqual({ + batchItemFailures: [{ itemIdentifier: "200" }], + }); + }); + + it("should handle missing sequence number for DynamoDB", async () => { + const recordWithoutSequenceNumber: DynamoDBRecord = { + eventID: "event-missing", + eventName: "INSERT", + eventVersion: "1.1", + eventSource: "aws:dynamodb", + awsRegion: "us-east-1", + dynamodb: { + SizeBytes: 100, + StreamViewType: "NEW_AND_OLD_IMAGES", + }, + eventSourceARN: "arn:aws:dynamodb:us-east-1:123456789012:table/test-table/stream/2023-01-01T00:00:00.000", + }; + + const handler = (_: DynamoDBRecord) => Effect.fail(new Error("Processing failed")); + + const layer = BatchProcessor({ eventType: EventType.DynamoDBStreams }); + + const result = await processPartialResponse({ Records: [recordWithoutSequenceNumber] }, handler).pipe( + Effect.provide(layer), + Effect.either, + Effect.runPromise, + ); + + // Single record failure results in FullBatchError + expect(Either.isLeft(result)).toBe(true); + if (Either.isLeft(result)) { + expect(result.left._tag).toBe("FullBatchError"); + } + }); + }); + + describe("processPartialResponse", () => { + it("should process partial response with SQS records", async () => { + const records = [ + { + messageId: "msg-1", + receiptHandle: "receipt-1", + body: "message 1", + attributes: { + ApproximateReceiveCount: "1", + SentTimestamp: "1695820800000", + SenderId: "123456789012", + ApproximateFirstReceiveTimestamp: "1695820800000", + }, + messageAttributes: {}, + md5OfBody: "test-md5", + eventSource: "aws:sqs", + eventSourceARN: "arn:aws:sqs:us-east-1:123456789012:test-queue", + awsRegion: "us-east-1", + } as SQSRecord, + ]; + + const handler = (record: SQSRecord) => Effect.succeed({ messageId: record.messageId, processed: true }); + + const layer = BatchProcessor({ eventType: EventType.SQS }); + + const result = await processPartialResponse({ Records: records }, handler).pipe( + Effect.provide(layer), + Effect.runPromise, + ); + + expect(result).toEqual({ batchItemFailures: [] }); + }); + + it("should handle errors in processPartialResponse", async () => { + const records = [ + { + messageId: "msg-1", + receiptHandle: "receipt-1", + body: "message 1", + attributes: { + ApproximateReceiveCount: "1", + SentTimestamp: "1695820800000", + SenderId: "123456789012", + ApproximateFirstReceiveTimestamp: "1695820800000", + }, + messageAttributes: {}, + md5OfBody: "test-md5", + eventSource: "aws:sqs", + eventSourceARN: "arn:aws:sqs:us-east-1:123456789012:test-queue", + awsRegion: "us-east-1", + } as SQSRecord, + ]; + + const handler = (_: SQSRecord) => Effect.fail(new Error("Processing failed")); + + const layer = BatchProcessor({ eventType: EventType.SQS }); + + const result = await processPartialResponse({ Records: records }, handler).pipe( + Effect.provide(layer), + Effect.either, + Effect.runPromise, + ); + + expect(Either.isLeft(result)).toBe(true); + if (Either.isLeft(result)) { + expect(result.left._tag).toBe("FullBatchError"); + } + }); + }); + + describe("Handler with dependencies", () => { + class TestService extends Effect.Tag("TestService") string }>() {} + + const TestServiceLive = Layer.succeed(TestService, { process: (data: string) => `processed: ${data}` }); + + it("should work with handlers that have dependencies", async () => { + const records = [ + { + messageId: "msg-1", + receiptHandle: "receipt-1", + body: "message 1", + attributes: { + ApproximateReceiveCount: "1", + SentTimestamp: "1695820800000", + SenderId: "123456789012", + ApproximateFirstReceiveTimestamp: "1695820800000", + }, + messageAttributes: {}, + md5OfBody: "test-md5", + eventSource: "aws:sqs", + eventSourceARN: "arn:aws:sqs:us-east-1:123456789012:test-queue", + awsRegion: "us-east-1", + } as SQSRecord, + ]; + + const handler = (record: SQSRecord) => + Effect.gen(function*() { + const service = yield* TestService; + return service.process(record.body); + }); + + const layer = BatchProcessor({ eventType: EventType.SQS }); + const fullLayer = Layer.merge(layer, TestServiceLive); + + const result = await processPartialResponse({ Records: records }, handler).pipe( + Effect.provide(fullLayer), + Effect.runPromise, + ); + + expect(result).toEqual({ batchItemFailures: [] }); + }); + }); + + describe("Empty batch handling", () => { + it("should handle empty batch", async () => { + const handler = (_: SQSRecord) => Effect.succeed({ processed: true }); + + const layer = BatchProcessor({ eventType: EventType.SQS }); + + const result = await processPartialResponse({ Records: [] }, handler).pipe( + Effect.provide(layer), + Effect.runPromise, + ); + + expect(result).toEqual({ batchItemFailures: [] }); + }); + }); + + describe("Multiple failure scenarios", () => { + it("should handle multiple failures correctly", async () => { + const records = [ + { messageId: "msg-1", body: "1" } as SQSRecord, + { messageId: "msg-2", body: "2" } as SQSRecord, + { messageId: "msg-3", body: "3" } as SQSRecord, + { messageId: "msg-4", body: "4" } as SQSRecord, + { messageId: "msg-5", body: "5" } as SQSRecord, + ]; + + const handler = (record: SQSRecord) => { + if (["msg-2", "msg-4"].includes(record.messageId)) { + return Effect.fail(new Error(`Failed to process ${record.messageId}`)); + } + return Effect.succeed({ messageId: record.messageId, processed: true }); + }; + + const layer = BatchProcessor({ eventType: EventType.SQS }); + + const result = await processPartialResponse({ Records: records }, handler).pipe( + Effect.provide(layer), + Effect.runPromise, + ); + + expect(result).toEqual({ + batchItemFailures: [{ itemIdentifier: "msg-2" }, { itemIdentifier: "msg-4" }], + }); + }); + }); +}); From df4cbf6bc868f1b82c8317226b603affb46d2231 Mon Sep 17 00:00:00 2001 From: Gabriel Colson Date: Thu, 4 Sep 2025 16:45:22 +0200 Subject: [PATCH 3/3] chore: add docgen --- packages/powertools-batch/src/batch-processor.ts | 11 +++++++++++ packages/powertools-batch/src/constants.ts | 11 +++++++++++ packages/powertools-batch/src/index.ts | 15 +++++++++++++++ .../src/process-partial-response.ts | 9 +++++++++ packages/powertools-batch/src/processor.ts | 11 +++++++++++ 5 files changed, 57 insertions(+) diff --git a/packages/powertools-batch/src/batch-processor.ts b/packages/powertools-batch/src/batch-processor.ts index 8b3ee2fd..f9f0ae42 100644 --- a/packages/powertools-batch/src/batch-processor.ts +++ b/packages/powertools-batch/src/batch-processor.ts @@ -1,3 +1,6 @@ +/** + * @since 1.0.0 + */ import type { DynamoDBRecord, KinesisStreamRecord, SQSRecord } from "aws-lambda"; import { Array, Effect, Either, Layer, Option, Schema } from "effect"; import type { BaseRecord } from "./constants.js"; @@ -20,10 +23,18 @@ const identifierMapper = (record: BaseRecord, eventType: EventType): string => { } }; +/** + * The options for the BatchProcessor. + * @since 1.0.0 + */ export interface BatchProcessorOptions { eventType: EventType; } +/** + * The default BatchProcessor layer. + * @since 1.0.0 + */ export const BatchProcessor = (options: BatchProcessorOptions) => Layer.succeed( Processor, diff --git a/packages/powertools-batch/src/constants.ts b/packages/powertools-batch/src/constants.ts index a434e058..31717c67 100644 --- a/packages/powertools-batch/src/constants.ts +++ b/packages/powertools-batch/src/constants.ts @@ -1,7 +1,18 @@ +/** + * @since 1.0.0 + */ import type { DynamoDBRecord, KinesisStreamRecord, SQSRecord } from "aws-lambda"; +/** + * The base record type for all event types. + * @since 1.0.0 + */ export type BaseRecord = { [key: string]: unknown } | SQSRecord | KinesisStreamRecord | DynamoDBRecord; +/** + * The event type for all event types. + * @since 1.0.0 + */ export enum EventType { SQS = "SQS", KinesisDataStreams = "KinesisDataStreams", diff --git a/packages/powertools-batch/src/index.ts b/packages/powertools-batch/src/index.ts index 0b172ac5..5deb86b7 100644 --- a/packages/powertools-batch/src/index.ts +++ b/packages/powertools-batch/src/index.ts @@ -1,4 +1,19 @@ +/** + * @since 1.0.0 + */ export * from "./batch-processor.js"; + +/** + * @since 1.0.0 + */ export * from "./constants.js"; + +/** + * @since 1.0.0 + */ export * from "./process-partial-response.js"; + +/** + * @since 1.0.0 + */ export * from "./processor.js"; diff --git a/packages/powertools-batch/src/process-partial-response.ts b/packages/powertools-batch/src/process-partial-response.ts index ae4567a9..6842b11d 100644 --- a/packages/powertools-batch/src/process-partial-response.ts +++ b/packages/powertools-batch/src/process-partial-response.ts @@ -1,7 +1,16 @@ +/** + * @since 1.0.0 + */ import { Effect } from "effect"; import type { BaseRecord } from "./constants.js"; import { Processor } from "./processor.js"; +/** + * @param event - The Lambda event + * @param recordHandler - The handler for processing each record + * @returns The partial response + * @since 1.0.0 + */ export const processPartialResponse = ( event: { Records: Array }, recordHandler: (record: Record) => Effect.Effect, diff --git a/packages/powertools-batch/src/processor.ts b/packages/powertools-batch/src/processor.ts index 50102262..a007c5ca 100644 --- a/packages/powertools-batch/src/processor.ts +++ b/packages/powertools-batch/src/processor.ts @@ -1,9 +1,20 @@ +/** + * @since 1.0.0 + */ import { Context, Schema } from "effect"; import type { Effect } from "effect"; import type { BaseRecord } from "./constants.js"; +/** + * The error thrown when a full batch is encountered. + * @since 1.0.0 + */ export class FullBatchError extends Schema.TaggedError()("FullBatchError", {}) {} +/** + * The processor context. + * @since 1.0.0 + */ export class Processor extends Context.Tag("@effect-aws/powertools-batch/Processor")< Processor, {