From 5b0eab5f66b388db7ed01ca83036adf14cb7091d Mon Sep 17 00:00:00 2001 From: Doug Perkes Date: Fri, 17 Jul 2026 09:22:55 -0500 Subject: [PATCH] New serverless pattern - cdk-dotnet-durablefunction-batchprocessing --- .../.gitignore | 345 ++++++++++++++++++ .../README.md | 277 ++++++++++++++ ...otnet-durablefunction-batchprocessing.json | 63 ++++ .../cdk.json | 34 ++ .../download-testdata.sh | 77 ++++ .../src/.globalconfig | 13 + .../src/BatchProcessor/BatchProcessor.csproj | 21 ++ .../BatchProcessor/FileProcessingService.cs | 74 ++++ .../src/BatchProcessor/Function.cs | 142 +++++++ .../src/BatchProcessor/Models.cs | 33 ++ .../src/BatchProcessor/ReportWriter.cs | 84 +++++ ...kDotnetDurablefunctionBatchprocessing.slnx | 9 + .../src/Infra/GlobalSuppressions.cs | 1 + .../src/Infra/Infra.csproj | 16 + .../src/Infra/InfraStack.cs | 123 +++++++ .../src/Infra/Program.cs | 20 + .../upload-testdata.sh | 69 ++++ 17 files changed, 1401 insertions(+) create mode 100644 cdk-dotnet-durablefunction-batchprocessing/.gitignore create mode 100644 cdk-dotnet-durablefunction-batchprocessing/README.md create mode 100644 cdk-dotnet-durablefunction-batchprocessing/cdk-dotnet-durablefunction-batchprocessing.json create mode 100644 cdk-dotnet-durablefunction-batchprocessing/cdk.json create mode 100755 cdk-dotnet-durablefunction-batchprocessing/download-testdata.sh create mode 100644 cdk-dotnet-durablefunction-batchprocessing/src/.globalconfig create mode 100644 cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/BatchProcessor.csproj create mode 100644 cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/FileProcessingService.cs create mode 100644 cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/Function.cs create mode 100644 cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/Models.cs create mode 100644 cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/ReportWriter.cs create mode 100644 cdk-dotnet-durablefunction-batchprocessing/src/CdkDotnetDurablefunctionBatchprocessing.slnx create mode 100644 cdk-dotnet-durablefunction-batchprocessing/src/Infra/GlobalSuppressions.cs create mode 100644 cdk-dotnet-durablefunction-batchprocessing/src/Infra/Infra.csproj create mode 100644 cdk-dotnet-durablefunction-batchprocessing/src/Infra/InfraStack.cs create mode 100644 cdk-dotnet-durablefunction-batchprocessing/src/Infra/Program.cs create mode 100755 cdk-dotnet-durablefunction-batchprocessing/upload-testdata.sh diff --git a/cdk-dotnet-durablefunction-batchprocessing/.gitignore b/cdk-dotnet-durablefunction-batchprocessing/.gitignore new file mode 100644 index 0000000000..ae49af835f --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/.gitignore @@ -0,0 +1,345 @@ +# CDK asset staging directory +.cdk.staging +cdk.out + +# Test data downloaded from open data +testdata/ + +# Created by https://www.gitignore.io/api/csharp + +### Csharp ### +## Ignore Visual Studio temporary files, build results, and +## files generated by popular Visual Studio add-ons. +## +## Get latest from https://github.com/github/gitignore/blob/main/VisualStudio.gitignore + +# User-specific files +*.suo +*.user +*.userosscache +*.sln.docstates + +# User-specific files (MonoDevelop/Xamarin Studio) +*.userprefs + +# Build results +[Dd]ebug/ +[Dd]ebugPublic/ +[Rr]elease/ +[Rr]eleases/ +x64/ +x86/ +bld/ +[Bb]in/ +[Oo]bj/ +[Ll]og/ + +# Visual Studio 2015/2017 cache/options directory +.vs/ +# Uncomment if you have tasks that create the project's static files in wwwroot +#wwwroot/ + +# Visual Studio 2017 auto generated files +Generated\ Files/ + +# MSTest test Results +[Tt]est[Rr]esult*/ +[Bb]uild[Ll]og.* + +# NUNIT +*.VisualState.xml +TestResult.xml + +# Build Results of an ATL Project +[Dd]ebugPS/ +[Rr]eleasePS/ +dlldata.c + +# Benchmark Results +BenchmarkDotNet.Artifacts/ + +# .NET Core +project.lock.json +project.fragment.lock.json +artifacts/ + +# StyleCop +StyleCopReport.xml + +# Files built by Visual Studio +*_i.c +*_p.c +*_i.h +*.ilk +*.meta +*.obj +*.iobj +*.pch +*.pdb +*.ipdb +*.pgc +*.pgd +*.rsp +*.sbr +*.tlb +*.tli +*.tlh +*.tmp +*.tmp_proj +*.log +*.vspscc +*.vssscc +.builds +*.pidb +*.svclog +*.scc + +# Chutzpah Test files +_Chutzpah* + +# Visual C++ cache files +ipch/ +*.aps +*.ncb +*.opendb +*.opensdf +*.sdf +*.cachefile +*.VC.db +*.VC.VC.opendb + +# Visual Studio profiler +*.psess +*.vsp +*.vspx +*.sap + +# Visual Studio Trace Files +*.e2e + +# TFS 2012 Local Workspace +$tf/ + +# Guidance Automation Toolkit +*.gpState + +# ReSharper is a .NET coding add-in +_ReSharper*/ +*.[Rr]e[Ss]harper +*.DotSettings.user + +# JustCode is a .NET coding add-in +.JustCode + +# TeamCity is a build add-in +_TeamCity* + +# DotCover is a Code Coverage Tool +*.dotCover + +# AxoCover is a Code Coverage Tool +.axoCover/* +!.axoCover/settings.json + +# Visual Studio code coverage results +*.coverage +*.coveragexml + +# NCrunch +_NCrunch_* +.*crunch*.local.xml +nCrunchTemp_* + +# MightyMoose +*.mm.* +AutoTest.Net/ + +# Web workbench (sass) +.sass-cache/ + +# Installshield output folder +[Ee]xpress/ + +# DocProject is a documentation generator add-in +DocProject/buildhelp/ +DocProject/Help/*.HxT +DocProject/Help/*.HxC +DocProject/Help/*.hhc +DocProject/Help/*.hhk +DocProject/Help/*.hhp +DocProject/Help/Html2 +DocProject/Help/html + +# Click-Once directory +publish/ + +# Publish Web Output +*.[Pp]ublish.xml +*.azurePubxml +# Note: Comment the next line if you want to checkin your web deploy settings, +# but database connection strings (with potential passwords) will be unencrypted +*.pubxml +*.publishproj + +# Microsoft Azure Web App publish settings. Comment the next line if you want to +# checkin your Azure Web App publish settings, but sensitive information contained +# in these scripts will be unencrypted +PublishScripts/ + +# NuGet Packages +*.nupkg +# The packages folder can be ignored because of Package Restore +**/[Pp]ackages/* +# except build/, which is used as an MSBuild target. +!**/[Pp]ackages/build/ +# Uncomment if necessary however generally it will be regenerated when needed +#!**/[Pp]ackages/repositories.config +# NuGet v3's project.json files produces more ignorable files +*.nuget.props +*.nuget.targets + +# Microsoft Azure Build Output +csx/ +*.build.csdef + +# Microsoft Azure Emulator +ecf/ +rcf/ + +# Windows Store app package directories and files +AppPackages/ +BundleArtifacts/ +Package.StoreAssociation.xml +_pkginfo.txt +*.appx + +# Visual Studio cache files +# files ending in .cache can be ignored +*.[Cc]ache +# but keep track of directories ending in .cache +!*.[Cc]ache/ + +# Others +ClientBin/ +~$* +*~ +*.dbmdl +*.dbproj.schemaview +*.jfm +*.pfx +*.publishsettings +orleans.codegen.cs + +# Including strong name files can present a security risk +# (https://github.com/github/gitignore/pull/2483#issue-259490424) +#*.snk + +# Since there are multiple workflows, uncomment next line to ignore bower_components +# (https://github.com/github/gitignore/pull/1529#issuecomment-104372622) +#bower_components/ + +# RIA/Silverlight projects +Generated_Code/ + +# Backup & report files from converting an old project file +# to a newer Visual Studio version. Backup files are not needed, +# because we have git ;-) +_UpgradeReport_Files/ +Backup*/ +UpgradeLog*.XML +UpgradeLog*.htm +ServiceFabricBackup/ +*.rptproj.bak + +# SQL Server files +*.mdf +*.ldf +*.ndf + +# Business Intelligence projects +*.rdl.data +*.bim.layout +*.bim_*.settings +*.rptproj.rsuser + +# Microsoft Fakes +FakesAssemblies/ + +# GhostDoc plugin setting file +*.GhostDoc.xml + +# Node.js Tools for Visual Studio +.ntvs_analysis.dat +node_modules/ + +# Visual Studio 6 build log +*.plg + +# Visual Studio 6 workspace options file +*.opt + +# Visual Studio 6 auto-generated workspace file (contains which files were open etc.) +*.vbw + +# Visual Studio LightSwitch build output +**/*.HTMLClient/GeneratedArtifacts +**/*.DesktopClient/GeneratedArtifacts +**/*.DesktopClient/ModelManifest.xml +**/*.Server/GeneratedArtifacts +**/*.Server/ModelManifest.xml +_Pvt_Extensions + +# Paket dependency manager +.paket/paket.exe +paket-files/ + +# FAKE - F# Make +.fake/ + +# JetBrains Rider +.idea/ +*.sln.iml + +# CodeRush +.cr/ + +# Python Tools for Visual Studio (PTVS) +__pycache__/ +*.pyc + +# Cake - Uncomment if you are using it +# tools/** +# !tools/packages.config + +# Tabs Studio +*.tss + +# Telerik's JustMock configuration file +*.jmconfig + +# BizTalk build output +*.btp.cs +*.btm.cs +*.odx.cs +*.xsd.cs + +# OpenCover UI analysis results +OpenCover/ + +# Azure Stream Analytics local run output +ASALocalRun/ + +# MSBuild Binary and Structured Log +*.binlog + +# NVidia Nsight GPU debugger configuration file +*.nvuser + +# MFractors (Xamarin productivity tool) working folder +.mfractor/ + +# Local History for Visual Studio +.localhistory/ + + +# End of https://www.gitignore.io/api/csharp \ No newline at end of file diff --git a/cdk-dotnet-durablefunction-batchprocessing/README.md b/cdk-dotnet-durablefunction-batchprocessing/README.md new file mode 100644 index 0000000000..85f6765006 --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/README.md @@ -0,0 +1,277 @@ +# AWS Lambda Durable Function — Batch File Processing (Dynamic Fan-Out/Fan-In) + +This sample demonstrates how to build a dynamic batch processing pipeline using **AWS Lambda Durable Functions** for .NET. Unlike the image processing sample where parallel branches are hardcoded, this workflow discovers files at runtime and dynamically creates a parallel task for each one — the number of branches is unknown at design time. + +## Architecture + +``` +┌──────────────┐ ┌──────────────────────────────────────────────────────────────┐ +│ Invoke with │ │ Lambda Durable Function │ +│ S3 prefix │──────▶│ │ +└──────────────┘ │ ┌───────────────────────────────────────────────────┐ │ + │ │ Step: List files under prefix │ │ + │ │ (discovers N files at runtime) │ │ + │ └────────────────────┬──────────────────────────────┘ │ + │ ▼ │ + │ ┌────────── ParallelAsync (N branches) ─────────┐ │ + │ │ │ │ + │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ + │ │ │ File 1 │ │ File 2 │ ... │ File N │ │ │ + │ │ │ process │ │ process │ │ process │ │ │ + │ │ │ + write │ │ + write │ │ + write │ │ │ + │ │ └────┬────┘ └────┬────┘ └────┬────┘ │ │ + │ │ │ │ │ │ │ + │ └───────┼───────────┼──────────────────┼────────┘ │ + │ └───────────┼──────────────────┘ │ + │ ▼ │ + │ ┌───────────────────────────────────────────────────┐ │ + │ │ Step: Aggregate results + write summary to S3 │ │ + │ └───────────────────────────────────────────────────┘ │ + └──────────────────────────────────────────────────────────────┘ +``` + +## What It Demonstrates + +- **Dynamic fan-out** — The number of parallel branches is determined at runtime by listing S3 objects. The workflow handles 1 file or 1,000 files with the same code. +- **Automatic checkpointing** — Each file's processing is independently checkpointed. If the Lambda is interrupted, completed files resume from cache. +- **Partial failure tolerance** — Uses `CompletionConfig.AllCompleted()` so all files are processed even if some fail. The summary report captures both successes and failures. +- **Per-file processing** — Each branch downloads the file, computes line count, word count, byte size, and SHA-256 hash. +- **Summary report** — After all files complete, a JSON report is written to S3 with aggregated statistics. +- **CDK infrastructure** — Deploys S3 bucket, DynamoDB table, and Lambda function with least-privilege IAM permissions. + +## Project Structure + +``` +├── cdk.json +├── src/ +│ ├── CdkDotnetDurablefunctionBatchprocessing.slnx +│ ├── .globalconfig +│ ├── Infra/ +│ │ ├── Program.cs +│ │ ├── InfraStack.cs +│ │ ├── GlobalSuppressions.cs +│ │ └── Infra.csproj +│ └── BatchProcessor/ +│ ├── Function.cs # Durable workflow with dynamic ParallelAsync +│ ├── FileProcessingService.cs # S3 listing + per-file processing +│ ├── ReportWriter.cs # DynamoDB + S3 report persistence +│ ├── Models.cs # Input/output record types +│ └── BatchProcessor.csproj +└── README.md +``` + +--- + +## Prerequisites + +- [.NET 10 SDK](https://dotnet.microsoft.com/download/dotnet/10.0) +- [AWS CDK CLI](https://docs.aws.amazon.com/cdk/v2/guide/getting-started.html) (`npm install -g aws-cdk`) +- AWS account with credentials configured (`aws configure`) +- CDK bootstrapped in your target account/region (`cdk bootstrap`) + +--- + +## Build & Deploy + +### Step 1: Build the solution + +```bash +dotnet build src/CdkDotnetDurablefunctionBatchprocessing.slnx -c Release +``` + +### Step 2: Deploy + +```bash +cdk deploy +``` + +The CDK automatically publishes the Lambda function during synthesis. This creates: + +| Resource | Description | +|----------|-------------| +| S3 Bucket | Stores input files and JSON summary reports | +| DynamoDB Table | Stores per-file processing results (`BatchId` + `FileName`) | +| Lambda Function | Durable workflow (512 MB, 60 min execution timeout) | +| IAM Role | Least-privilege: S3 read/write + DynamoDB write | + +--- + +## Testing + +### Capture stack outputs + +```bash +STACK_NAME="CdkDotnetDurablefunctionBatchprocessingStack" + +BUCKET_NAME=$(aws cloudformation describe-stacks \ + --stack-name $STACK_NAME \ + --query "Stacks[0].Outputs[?OutputKey=='FilesBucketName'].OutputValue" \ + --output text) + +FUNCTION_ARN=$(aws cloudformation describe-stacks \ + --stack-name $STACK_NAME \ + --query "Stacks[0].Outputs[?OutputKey=='BatchProcessorFunctionArn'].OutputValue" \ + --output text) +``` + +### Upload test files to S3 + +The project includes scripts to download real-world data from [NOAA's Global Historical Climatology Network (GHCN)](https://registry.opendata.aws/noaa-ghcn/) on Amazon S3 Open Data and upload it for processing. + +```bash +# Download ~75 small CSV weather station files to ./testdata/ (excluded from git) +./download-testdata.sh + +# Upload them to the deployed S3 bucket +./upload-testdata.sh +``` + +You can adjust the file count: `./download-testdata.sh 100` downloads 100 files. + +Alternatively, create your own test files manually: + +```bash +# Create some sample text files +for i in $(seq 1 5); do + echo "This is test file number $i with some sample content." > /tmp/file-$i.txt + echo "Line 2 of file $i." >> /tmp/file-$i.txt + echo "Line 3 with extra words to count." >> /tmp/file-$i.txt +done + +# Upload them under a common prefix +for i in $(seq 1 5); do + aws s3 cp /tmp/file-$i.txt s3://$BUCKET_NAME/input/batch-test/file-$i.txt +done +``` + +### Invoke the durable function + +```bash +aws lambda invoke \ + --function-name "$FUNCTION_ARN:\$LATEST" \ + --invocation-type Event \ + --cli-binary-format raw-in-base64-out \ + --payload '{"Bucket":"'$BUCKET_NAME'","Prefix":"input/batch-test/"}' \ + /tmp/invoke-response.json +``` + +> **Note:** Durable functions require a qualified ARN (append `:\$LATEST`). Functions with `ExecutionTimeout` greater than 15 minutes must be invoked asynchronously (`--invocation-type Event`). + +### Check execution status + +```bash +aws lambda list-durable-executions-by-function \ + --function-name $FUNCTION_ARN +``` + +### Verify the summary report + +```bash +# List reports +aws s3 ls s3://$BUCKET_NAME/reports/ + +# Download and view the latest report +REPORT_KEY=$(aws s3 ls s3://$BUCKET_NAME/reports/ --recursive | sort | tail -1 | awk '{print $4}') +aws s3 cp s3://$BUCKET_NAME/$REPORT_KEY - | jq . +``` + +Expected report structure: + +```json +{ + "BatchId": "batch-20260713-143000", + "SourceBucket": "", + "SourcePrefix": "input/batch-test/", + "TotalFiles": 5, + "TotalBytes": 475, + "SuccessCount": 5, + "FailureCount": 0, + "CompletedAt": "2026-07-13T14:30:05.0000000Z", + "Files": [ + { + "FileName": "file-1.txt", + "SizeBytes": 95, + "LineCount": 3, + "WordCount": 17, + "ContentHash": "a1b2c3..." + } + ] +} +``` + +### Verify per-file results in DynamoDB + +```bash +TABLE_NAME=$(aws cloudformation describe-stacks \ + --stack-name $STACK_NAME \ + --query "Stacks[0].Outputs[?OutputKey=='ResultsTableName'].OutputValue" \ + --output text) + +aws dynamodb scan --table-name $TABLE_NAME +``` + +--- + +## How the Dynamic Fan-Out Works + +The key difference from static parallelism is that branches are built from runtime data: + +```csharp +// Step 1: Discover files (count unknown at design time) +var fileKeys = await ctx.StepAsync( + async (_, ct) => await _fileService.ListFilesAsync(bucket, prefix, ct), + name: "list-files"); + +// Step 2: Dynamically create one branch per file +var branches = fileKeys.Select(key => + new DurableBranch( + Path.GetFileName(key), + async (branchCtx, ct) => + { + var result = await branchCtx.StepAsync( + async (_, t) => await _fileService.ProcessFileAsync(bucket, key, t), + name: "process-file", + cancellationToken: ct); + return result; + })) + .ToList(); + +// Step 3: Execute all branches (max 10 concurrent) +var batch = await ctx.ParallelAsync(branches, name: "process-all-files", + config: new ParallelConfig + { + MaxConcurrency = 10, + CompletionConfig = CompletionConfig.AllCompleted() + }); +``` + +`MaxConcurrency = 10` bounds how many branches run at once, preventing downstream service overload while still processing files in parallel. + +--- + +## Cleanup + +```bash +cdk destroy +``` + +--- + +## Useful Commands + +| Command | Description | +|---------|-------------| +| `dotnet build src/CdkDotnetDurablefunctionBatchprocessing.slnx` | Build the entire solution | +| `cdk synth` | Emit the synthesized CloudFormation template (publishes Lambda automatically) | +| `cdk diff` | Compare deployed stack with current state | +| `cdk deploy` | Deploy this stack to your default AWS account/region | +| `cdk destroy` | Tear down the stack and all resources | + +--- + +## References + +- [Amazon.Lambda.DurableExecution SDK](https://github.com/aws/aws-lambda-dotnet/tree/master/Libraries/src/Amazon.Lambda.DurableExecution) +- [Parallel documentation](https://github.com/aws/aws-lambda-dotnet/blob/master/Libraries/src/Amazon.Lambda.DurableExecution/docs/core/parallel.md) +- [Steps documentation](https://github.com/aws/aws-lambda-dotnet/blob/master/Libraries/src/Amazon.Lambda.DurableExecution/docs/core/steps.md) +- [AWS CDK .NET Reference](https://docs.aws.amazon.com/cdk/api/v2/dotnet/) diff --git a/cdk-dotnet-durablefunction-batchprocessing/cdk-dotnet-durablefunction-batchprocessing.json b/cdk-dotnet-durablefunction-batchprocessing/cdk-dotnet-durablefunction-batchprocessing.json new file mode 100644 index 0000000000..9e272808fc --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/cdk-dotnet-durablefunction-batchprocessing.json @@ -0,0 +1,63 @@ +{ + "title": "Lambda Durable Function — Dynamic Batch Processing with Fan-Out/Fan-In (.NET)", + "description": "A durable function that discovers files in S3 at runtime, dynamically fans out a parallel task per file, then aggregates results into a summary report.", + "language": ".NET", + "level": "300", + "framework": "CDK", + "introBox": { + "headline": "How it works", + "text": [ + "This pattern deploys a Lambda durable function that implements dynamic fan-out/fan-in. Given an S3 prefix, it lists all files at runtime and creates a parallel processing task for each one — the parallelism is not hardcoded at design time.", + "Each file is processed independently (parsed, transformed, written to an output prefix), and all branches are awaited with ParallelAsync. Once all files complete, the function aggregates results and writes a summary report to S3.", + "This demonstrates dynamic parallelism with durable functions: the number of branches is determined at runtime, each branch is independently checkpointed, and the orchestration is resilient to interruptions." + ] + }, + "gitHub": { + "template": { + "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/cdk-dotnet-durablefunction-batchprocessing", + "templateURL": "serverless-patterns/cdk-dotnet-durablefunction-batchprocessing", + "projectFolder": "cdk-dotnet-durablefunction-batchprocessing", + "templateFile": "cdk.json" + } + }, + "resources": { + "bullets": [ + { + "text": "Lambda Durable Functions Documentation", + "link": "https://docs.aws.amazon.com/lambda/latest/dg/durable-functions.html" + }, + { + "text": "Build multi-step applications and AI workflows with Lambda durable functions", + "link": "https://aws.amazon.com/blogs/aws/build-multi-step-applications-and-ai-workflows-with-aws-lambda-durable-functions/" + }, + { + "text": "Amazon S3", + "link": "https://docs.aws.amazon.com/AmazonS3/latest/userguide/Welcome.html" + } + ] + }, + "deploy": { + "text": [ + "cdk deploy" + ] + }, + "testing": { + "text": [ + "See the GitHub repo for detailed testing instructions." + ] + }, + "cleanup": { + "text": [ + "Delete the stack: cdk destroy." + ] + }, + "authors": [ + { + "name": "Doug Perkes", + "image": "", + "bio": "Senior Solutions Architect at AWS, focused on .NET and serverless.", + "linkedin": "dougperkes", + "twitter": "" + } + ] +} diff --git a/cdk-dotnet-durablefunction-batchprocessing/cdk.json b/cdk-dotnet-durablefunction-batchprocessing/cdk.json new file mode 100644 index 0000000000..f8e339a45e --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/cdk.json @@ -0,0 +1,34 @@ +{ + "app": "dotnet run --project src/Infra/Infra.csproj", + "watch": { + "include": [ + "**" + ], + "exclude": [ + "README.md", + "cdk*.json", + "src/*/obj", + "src/*/bin", + "src/*.sln", + "src/*/GlobalSuppressions.cs", + "src/*/*.csproj" + ] + }, + "context": { + "@aws-cdk/aws-lambda:recognizeLayerVersion": true, + "@aws-cdk/core:checkSecretUsage": true, + "@aws-cdk/core:target-partitions": [ + "aws", + "aws-cn" + ], + "@aws-cdk/aws-iam:minimizePolicies": true, + "@aws-cdk/core:validateSnapshotRemovalPolicy": true, + "@aws-cdk/aws-s3:createDefaultLoggingPolicy": true, + "@aws-cdk/aws-ec2:restrictDefaultSecurityGroup": true, + "@aws-cdk/core:enablePartitionLiterals": true, + "@aws-cdk/aws-s3:serverAccessLogsUseBucketPolicy": true, + "@aws-cdk/customresources:installLatestAwsSdkDefault": false, + "@aws-cdk/aws-s3:keepNotificationInImportedBucket": false, + "@aws-cdk/aws-s3:publicAccessBlockedByDefault": true + } +} diff --git a/cdk-dotnet-durablefunction-batchprocessing/download-testdata.sh b/cdk-dotnet-durablefunction-batchprocessing/download-testdata.sh new file mode 100755 index 0000000000..bbb6a34e0c --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/download-testdata.sh @@ -0,0 +1,77 @@ +#!/usr/bin/env bash +set -euo pipefail + +# ────────────────────────────────────────────────────────────────────────────── +# download-testdata.sh +# +# Downloads public domain books from Project Gutenberg and splits them into +# individual chapter/section files for batch processing. Each file becomes +# one parallel task in the durable function workflow. +# +# Files are saved to ./testdata/ which is excluded from source control. +# +# Usage: +# ./download-testdata.sh +# ────────────────────────────────────────────────────────────────────────────── + +OUTPUT_DIR="testdata" + +# Project Gutenberg plain text URLs (public domain books) +declare -a BOOKS=( + "https://www.gutenberg.org/cache/epub/1342/pg1342.txt" # Pride and Prejudice + "https://www.gutenberg.org/cache/epub/84/pg84.txt" # Frankenstein + "https://www.gutenberg.org/cache/epub/11/pg11.txt" # Alice in Wonderland + "https://www.gutenberg.org/cache/epub/1661/pg1661.txt" # Sherlock Holmes + "https://www.gutenberg.org/cache/epub/2701/pg2701.txt" # Moby Dick + "https://www.gutenberg.org/cache/epub/98/pg98.txt" # A Tale of Two Cities +) + +declare -a NAMES=( + "pride-and-prejudice" + "frankenstein" + "alice-in-wonderland" + "sherlock-holmes" + "moby-dick" + "tale-of-two-cities" +) + +echo "▶ Downloading books from Project Gutenberg and splitting into files..." +echo " Target: ./${OUTPUT_DIR}/" +echo "" + +rm -rf "$OUTPUT_DIR" +mkdir -p "$OUTPUT_DIR" + +TOTAL_FILES=0 + +for i in "${!BOOKS[@]}"; do + URL="${BOOKS[$i]}" + NAME="${NAMES[$i]}" + TEMP_FILE="/tmp/${NAME}.txt" + + echo " Downloading: ${NAME}..." + curl -sL "$URL" -o "$TEMP_FILE" + + # Split the book into chunks of ~500 lines each + # This produces roughly 10-15 files per book, giving us 60-90 total + CHUNK_NUM=0 + while IFS= read -r line; do + CHUNK_FILE="${OUTPUT_DIR}/${NAME}-$(printf '%03d' $CHUNK_NUM).txt" + echo "$line" >> "$CHUNK_FILE" + + # Start a new chunk every 500 lines + if [[ $(wc -l < "$CHUNK_FILE" | tr -d ' ') -ge 500 ]]; then + CHUNK_NUM=$((CHUNK_NUM + 1)) + fi + done < "$TEMP_FILE" + + CHUNK_COUNT=$((CHUNK_NUM + 1)) + TOTAL_FILES=$((TOTAL_FILES + CHUNK_COUNT)) + echo " → Split into ${CHUNK_COUNT} files" + + rm -f "$TEMP_FILE" +done + +echo "" +echo "✓ Created ${TOTAL_FILES} files in ./${OUTPUT_DIR}/" +echo " Total size: $(du -sh "$OUTPUT_DIR" | cut -f1)" diff --git a/cdk-dotnet-durablefunction-batchprocessing/src/.globalconfig b/cdk-dotnet-durablefunction-batchprocessing/src/.globalconfig new file mode 100644 index 0000000000..01268a03a6 --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/src/.globalconfig @@ -0,0 +1,13 @@ +is_global = true + +# CA2007: Do not directly await a Task without ConfigureAwait +# Not applicable in application code running on default synchronization context (Lambda, console apps). +dotnet_diagnostic.CA2007.severity = none + +# CA1711: Identifiers should not have incorrect suffix +# 'Stack' is the standard naming convention for AWS CDK stacks. +dotnet_diagnostic.CA1711.severity = none + +# CA1806: Do not ignore method results +# CDK constructs register themselves with the construct tree via side-effects in constructors. +dotnet_diagnostic.CA1806.severity = none diff --git a/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/BatchProcessor.csproj b/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/BatchProcessor.csproj new file mode 100644 index 0000000000..0308041f8c --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/BatchProcessor.csproj @@ -0,0 +1,21 @@ + + + + Exe + net10.0 + enable + enable + BatchProcessor + BatchProcessor + + + + + + + + + + + + diff --git a/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/FileProcessingService.cs b/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/FileProcessingService.cs new file mode 100644 index 0000000000..37370e9c00 --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/FileProcessingService.cs @@ -0,0 +1,74 @@ +using System.Security.Cryptography; +using Amazon.S3; +using Amazon.S3.Model; + +namespace BatchProcessor; + +/// +/// Service that processes individual files: reads content, computes stats, and returns results. +/// +internal sealed class FileProcessingService(IAmazonS3 s3Client) +{ + private readonly IAmazonS3 _s3Client = s3Client; + + /// + /// Lists all object keys under the given prefix. + /// + public async Task> ListFilesAsync(string bucket, string prefix, CancellationToken ct) + { + var keys = new List(); + string? continuationToken = null; + + do + { + var response = await _s3Client.ListObjectsV2Async(new ListObjectsV2Request + { + BucketName = bucket, + Prefix = prefix, + ContinuationToken = continuationToken + }, ct); + + foreach (var obj in response.S3Objects) + { + // Skip "directory" markers (keys ending in /) + if (!obj.Key.EndsWith('/')) + { + keys.Add(obj.Key); + } + } + + continuationToken = response.IsTruncated == true ? response.NextContinuationToken : null; + } + while (continuationToken is not null); + + return keys; + } + + /// + /// Processes a single file: downloads it, computes line count, word count, size, and SHA-256 hash. + /// + public async Task ProcessFileAsync(string bucket, string key, CancellationToken ct) + { + using var response = await _s3Client.GetObjectAsync(bucket, key, ct); + using var stream = response.ResponseStream; + using var reader = new StreamReader(stream); + + var content = await reader.ReadToEndAsync(ct); + + var lineCount = content.Split('\n').Length; + var wordCount = content.Split([' ', '\t', '\n', '\r'], StringSplitOptions.RemoveEmptyEntries).Length; + var sizeBytes = response.ContentLength; + + var hashBytes = SHA256.HashData(System.Text.Encoding.UTF8.GetBytes(content)); + var contentHash = Convert.ToHexStringLower(hashBytes); + + var fileName = Path.GetFileName(key); + + return new FileProcessingResult( + FileName: fileName, + SizeBytes: sizeBytes, + LineCount: lineCount, + WordCount: wordCount, + ContentHash: contentHash); + } +} diff --git a/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/Function.cs b/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/Function.cs new file mode 100644 index 0000000000..3b82459a2b --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/Function.cs @@ -0,0 +1,142 @@ +using Amazon.DynamoDBv2; +using Amazon.Lambda.Core; +using Amazon.Lambda.DurableExecution; +using Amazon.Lambda.RuntimeSupport; +using Amazon.Lambda.Serialization.SystemTextJson; +using Amazon.S3; + +[assembly: LambdaSerializer(typeof(DefaultLambdaJsonSerializer))] + +namespace BatchProcessor; + +/// +/// Lambda Durable Function that processes a batch of files using dynamic fan-out/fan-in. +/// +/// Workflow: +/// 1. List all files under the specified S3 prefix (unknown count at design time). +/// 2. Dynamically create a parallel branch for each discovered file. +/// 3. Wait for all branches to complete (each processes one file). +/// 4. Aggregate results and write a summary report to S3. +/// +internal sealed class BatchProcessorHandler +{ + private static readonly IAmazonS3 S3Client = new AmazonS3Client(); + private static readonly IAmazonDynamoDB DynamoDbClient = new AmazonDynamoDBClient(); + + private static readonly string TableName = + Environment.GetEnvironmentVariable("RESULTS_TABLE_NAME") + ?? throw new InvalidOperationException("RESULTS_TABLE_NAME environment variable is not set."); + + private static readonly string BucketName = + Environment.GetEnvironmentVariable("FILES_BUCKET_NAME") + ?? throw new InvalidOperationException("FILES_BUCKET_NAME environment variable is not set."); + + private readonly FileProcessingService _fileService = new(S3Client); + private readonly ReportWriter _reportWriter = new(S3Client, DynamoDbClient, TableName); + + public static async Task Main() + { + var handler = new BatchProcessorHandler(); + var serializer = new DefaultLambdaJsonSerializer(); + using var wrapper = HandlerWrapper.GetHandlerWrapper( + handler.Handler, serializer); + using var bootstrap = new LambdaBootstrap(wrapper); + await bootstrap.RunAsync(); + } + + public Task Handler( + DurableExecutionInvocationInput input, ILambdaContext context) + => DurableFunction.WrapAsync(Workflow, input, context); + + private async Task Workflow(BatchInput input, IDurableContext ctx) + { + var bucket = input.Bucket ?? BucketName; + var batchId = $"batch-{DateTime.UtcNow:yyyyMMdd-HHmmss}"; + + // ────────────────────────────────────────────────────────────────── + // Step 1: List all files under the prefix (checkpointed). + // The file count is unknown at design time — discovered at runtime. + // ────────────────────────────────────────────────────────────────── + var fileKeys = await ctx.StepAsync( + async (_, ct) => await _fileService.ListFilesAsync(bucket, input.Prefix, ct), + name: "list-files", + config: new StepConfig { RetryStrategy = RetryStrategy.Default }); + + if (fileKeys.Count == 0) + { + // No files found — write an empty summary + return new BatchSummary( + BatchId: batchId, + SourceBucket: bucket, + SourcePrefix: input.Prefix, + TotalFiles: 0, + TotalBytes: 0, + SuccessCount: 0, + FailureCount: 0, + ReportKey: "", + CompletedAt: DateTime.UtcNow); + } + + // ────────────────────────────────────────────────────────────────── + // Step 2: Dynamic Fan-Out — create a parallel branch for EACH file. + // The number of branches is determined at runtime by the list result. + // ────────────────────────────────────────────────────────────────── + var branches = fileKeys.Select(key => + new DurableBranch( + Path.GetFileName(key), + async (branchCtx, ct) => + { + // Process the file + var result = await branchCtx.StepAsync( + async (_, t) => await _fileService.ProcessFileAsync(bucket, key, t), + name: "process-file", + config: new StepConfig { RetryStrategy = RetryStrategy.Default }, + cancellationToken: ct); + + // Write per-file result to DynamoDB + await branchCtx.StepAsync( + async (_, t) => await _reportWriter.WriteFileResultAsync(batchId, result, t), + name: "write-result", + config: new StepConfig { RetryStrategy = RetryStrategy.Default }, + cancellationToken: ct); + + return result; + })) + .ToList(); + + var batch = await ctx.ParallelAsync( + branches, + name: "process-all-files", + config: new ParallelConfig + { + MaxConcurrency = 10, + CompletionConfig = CompletionConfig.AllCompleted() + }); + + // ────────────────────────────────────────────────────────────────── + // Step 3: Fan-In — aggregate results and write summary report. + // ────────────────────────────────────────────────────────────────── + var successfulResults = batch.GetResults(); + var totalBytes = successfulResults.Sum(r => r.SizeBytes); + + var summary = new BatchSummary( + BatchId: batchId, + SourceBucket: bucket, + SourcePrefix: input.Prefix, + TotalFiles: batch.TotalCount, + TotalBytes: totalBytes, + SuccessCount: batch.SuccessCount, + FailureCount: batch.FailureCount, + ReportKey: "", + CompletedAt: DateTime.UtcNow); + + // Write the JSON summary report to S3 + var reportKey = await ctx.StepAsync( + async (_, ct) => await _reportWriter.WriteSummaryReportAsync( + bucket, summary, successfulResults, ct), + name: "write-summary-report", + config: new StepConfig { RetryStrategy = RetryStrategy.Default }); + + return summary with { ReportKey = reportKey }; + } +} diff --git a/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/Models.cs b/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/Models.cs new file mode 100644 index 0000000000..07ce68e4b1 --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/Models.cs @@ -0,0 +1,33 @@ +using System.Diagnostics.CodeAnalysis; + +namespace BatchProcessor; + +/// +/// Input to the durable workflow — specifies which S3 prefix to scan for files. +/// +[SuppressMessage("Performance", "CA1812:Avoid uninstantiated internal classes", Justification = "Deserialized by Lambda runtime")] +internal sealed record BatchInput(string Bucket, string Prefix); + +/// +/// Result of processing a single file. +/// +internal sealed record FileProcessingResult( + string FileName, + long SizeBytes, + int LineCount, + int WordCount, + string ContentHash); + +/// +/// Final summary report written to S3 after all files are processed. +/// +internal sealed record BatchSummary( + string BatchId, + string SourceBucket, + string SourcePrefix, + int TotalFiles, + long TotalBytes, + int SuccessCount, + int FailureCount, + string ReportKey, + DateTime CompletedAt); diff --git a/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/ReportWriter.cs b/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/ReportWriter.cs new file mode 100644 index 0000000000..f15da498fd --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/src/BatchProcessor/ReportWriter.cs @@ -0,0 +1,84 @@ +using System.Globalization; +using System.Text.Json; +using Amazon.DynamoDBv2; +using Amazon.DynamoDBv2.Model; +using Amazon.S3; +using Amazon.S3.Model; + +namespace BatchProcessor; + +/// +/// Writes per-file results to DynamoDB and the final summary report to S3. +/// +internal sealed class ReportWriter(IAmazonS3 s3Client, IAmazonDynamoDB dynamoDb, string tableName) +{ + private readonly IAmazonS3 _s3Client = s3Client; + private readonly IAmazonDynamoDB _dynamoDb = dynamoDb; + private readonly string _tableName = tableName; + + /// + /// Writes a single file's processing result to DynamoDB. + /// + public async Task WriteFileResultAsync(string batchId, FileProcessingResult result, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(result); + + var item = new Dictionary + { + ["BatchId"] = new() { S = batchId }, + ["FileName"] = new() { S = result.FileName }, + ["SizeBytes"] = new() { N = result.SizeBytes.ToString(CultureInfo.InvariantCulture) }, + ["LineCount"] = new() { N = result.LineCount.ToString(CultureInfo.InvariantCulture) }, + ["WordCount"] = new() { N = result.WordCount.ToString(CultureInfo.InvariantCulture) }, + ["ContentHash"] = new() { S = result.ContentHash } + }; + + await _dynamoDb.PutItemAsync(new PutItemRequest + { + TableName = _tableName, + Item = item + }, ct); + } + + /// + /// Writes the final batch summary report as JSON to S3. + /// + public async Task WriteSummaryReportAsync( + string bucket, + BatchSummary summary, + IReadOnlyList results, + CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(summary); + + var report = new + { + summary.BatchId, + summary.SourceBucket, + summary.SourcePrefix, + summary.TotalFiles, + summary.TotalBytes, + summary.SuccessCount, + summary.FailureCount, + summary.CompletedAt, + Files = results + }; + + var reportJson = JsonSerializer.Serialize(report, new JsonSerializerOptions + { + WriteIndented = true + }); + + var reportKey = $"reports/{summary.BatchId}.json"; + + await _s3Client.PutObjectAsync(new PutObjectRequest + { + BucketName = bucket, + Key = reportKey, + ContentBody = reportJson, + ContentType = "application/json" + }, ct); + + return reportKey; + } +} diff --git a/cdk-dotnet-durablefunction-batchprocessing/src/CdkDotnetDurablefunctionBatchprocessing.slnx b/cdk-dotnet-durablefunction-batchprocessing/src/CdkDotnetDurablefunctionBatchprocessing.slnx new file mode 100644 index 0000000000..6629a3c8a0 --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/src/CdkDotnetDurablefunctionBatchprocessing.slnx @@ -0,0 +1,9 @@ + + + + + + + + + diff --git a/cdk-dotnet-durablefunction-batchprocessing/src/Infra/GlobalSuppressions.cs b/cdk-dotnet-durablefunction-batchprocessing/src/Infra/GlobalSuppressions.cs new file mode 100644 index 0000000000..26233fcb54 --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/src/Infra/GlobalSuppressions.cs @@ -0,0 +1 @@ +[assembly: System.Diagnostics.CodeAnalysis.SuppressMessage("Potential Code Quality Issues", "RECS0026:Possible unassigned object created by 'new'", Justification = "Constructs add themselves to the scope in which they are created")] diff --git a/cdk-dotnet-durablefunction-batchprocessing/src/Infra/Infra.csproj b/cdk-dotnet-durablefunction-batchprocessing/src/Infra/Infra.csproj new file mode 100644 index 0000000000..7707ef69ea --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/src/Infra/Infra.csproj @@ -0,0 +1,16 @@ + + + + Exe + net10.0 + enable + enable + Infra + + + + + + + + diff --git a/cdk-dotnet-durablefunction-batchprocessing/src/Infra/InfraStack.cs b/cdk-dotnet-durablefunction-batchprocessing/src/Infra/InfraStack.cs new file mode 100644 index 0000000000..07087c9678 --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/src/Infra/InfraStack.cs @@ -0,0 +1,123 @@ +using System.Diagnostics; +using Amazon.CDK; +using Amazon.CDK.AWS.DynamoDB; +using Amazon.CDK.AWS.Lambda; +using Amazon.CDK.AWS.S3; +using Constructs; + +namespace Infra; + +internal sealed class InfraStack : Stack +{ + internal InfraStack(Construct scope, string id, IStackProps? props = null) + : base(scope, id, props) + { + // ────────────────────────────────────────────────────────────────── + // S3 Bucket — stores input files and the summary report + // ────────────────────────────────────────────────────────────────── + var filesBucket = new Bucket(this, "FilesBucket", new BucketProps + { + RemovalPolicy = RemovalPolicy.DESTROY, + AutoDeleteObjects = true, + BlockPublicAccess = BlockPublicAccess.BLOCK_ALL, + Encryption = BucketEncryption.S3_MANAGED + }); + + // ────────────────────────────────────────────────────────────────── + // DynamoDB Table — stores per-file processing results + // ────────────────────────────────────────────────────────────────── + var resultsTable = new Table(this, "ResultsTable", new TableProps + { + PartitionKey = new Amazon.CDK.AWS.DynamoDB.Attribute + { + Name = "BatchId", + Type = AttributeType.STRING + }, + SortKey = new Amazon.CDK.AWS.DynamoDB.Attribute + { + Name = "FileName", + Type = AttributeType.STRING + }, + BillingMode = BillingMode.PAY_PER_REQUEST, + RemovalPolicy = RemovalPolicy.DESTROY + }); + + // ────────────────────────────────────────────────────────────────── + // Lambda Function — Durable batch file processing workflow + // ────────────────────────────────────────────────────────────────── + var publishOutputPath = PublishLambdaProject("src/BatchProcessor/BatchProcessor.csproj"); + + var batchProcessorFunction = new Function(this, "BatchProcessorFunction", new FunctionProps + { + Runtime = Runtime.DOTNET_10, + Handler = "BatchProcessor::BatchProcessor.BatchProcessorHandler::Handler", + Code = Code.FromAsset(publishOutputPath), + MemorySize = 512, + Timeout = Duration.Seconds(900), + DurableConfig = new DurableConfig + { + ExecutionTimeout = Duration.Minutes(60), + RetentionPeriod = Duration.Days(7) + }, + Environment = new Dictionary + { + ["RESULTS_TABLE_NAME"] = resultsTable.TableName, + ["FILES_BUCKET_NAME"] = filesBucket.BucketName + }, + Description = "Durable Function: Batch file processing with dynamic fan-out/fan-in" + }); + + // Grant permissions + filesBucket.GrantReadWrite(batchProcessorFunction); + resultsTable.GrantWriteData(batchProcessorFunction); + + // ────────────────────────────────────────────────────────────────── + // Outputs + // ────────────────────────────────────────────────────────────────── + _ = new CfnOutput(this, "FilesBucketName", new CfnOutputProps + { + Value = filesBucket.BucketName, + Description = "S3 bucket for input files and summary reports" + }); + + _ = new CfnOutput(this, "ResultsTableName", new CfnOutputProps + { + Value = resultsTable.TableName, + Description = "DynamoDB table storing per-file processing results" + }); + + _ = new CfnOutput(this, "BatchProcessorFunctionArn", new CfnOutputProps + { + Value = batchProcessorFunction.FunctionArn, + Description = "ARN of the batch processor Lambda function" + }); + } + + private static string PublishLambdaProject(string projectPath) + { + var publishDir = Path.Combine( + Path.GetDirectoryName(projectPath)!, + "bin", "Release", "net10.0", "publish"); + + var process = Process.Start(new ProcessStartInfo + { + FileName = "dotnet", + Arguments = $"publish \"{projectPath}\" -c Release -r linux-x64 --self-contained false -o \"{publishDir}\"", + RedirectStandardOutput = true, + RedirectStandardError = true, + UseShellExecute = false, + }) ?? throw new InvalidOperationException("Failed to start dotnet publish process."); + + var stdout = process.StandardOutput.ReadToEnd(); + var stderr = process.StandardError.ReadToEnd(); + process.WaitForExit(); + + if (process.ExitCode != 0) + { + throw new InvalidOperationException( + $"dotnet publish failed for '{projectPath}' (exit code {process.ExitCode}):\n{stderr}\n{stdout}"); + } + + return publishDir; + } +} diff --git a/cdk-dotnet-durablefunction-batchprocessing/src/Infra/Program.cs b/cdk-dotnet-durablefunction-batchprocessing/src/Infra/Program.cs new file mode 100644 index 0000000000..baab852094 --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/src/Infra/Program.cs @@ -0,0 +1,20 @@ +using Amazon.CDK; + +namespace Infra; + +sealed class Program +{ + public static void Main() + { + var app = new App(); + new InfraStack(app, "CdkDotnetDurablefunctionBatchprocessingStack", new StackProps + { + Env = new Amazon.CDK.Environment + { + Account = System.Environment.GetEnvironmentVariable("CDK_DEFAULT_ACCOUNT"), + Region = System.Environment.GetEnvironmentVariable("CDK_DEFAULT_REGION"), + } + }); + app.Synth(); + } +} diff --git a/cdk-dotnet-durablefunction-batchprocessing/upload-testdata.sh b/cdk-dotnet-durablefunction-batchprocessing/upload-testdata.sh new file mode 100755 index 0000000000..fbad20b8e2 --- /dev/null +++ b/cdk-dotnet-durablefunction-batchprocessing/upload-testdata.sh @@ -0,0 +1,69 @@ +#!/usr/bin/env bash +set -euo pipefail + +# ────────────────────────────────────────────────────────────────────────────── +# upload-testdata.sh +# +# Uploads all files from ./testdata/ to the S3 bucket created by the CDK stack. +# Files are placed under the input/batch/ prefix which can then be passed to +# the durable function for processing. +# +# Usage: +# ./upload-testdata.sh [STACK_NAME] [PREFIX] +# STACK_NAME defaults to CdkDotnetDurablefunctionBatchprocessingStack +# PREFIX defaults to input/batch/ +# ────────────────────────────────────────────────────────────────────────────── + +STACK_NAME="${1:-CdkDotnetDurablefunctionBatchprocessingStack}" +PREFIX="${2:-input/batch/}" +INPUT_DIR="testdata" + +if [[ ! -d "$INPUT_DIR" ]]; then + echo "✗ Directory ./${INPUT_DIR}/ not found. Run ./download-testdata.sh first." >&2 + exit 1 +fi + +FILE_COUNT=$(find "$INPUT_DIR" -type f | wc -l | tr -d ' ') +if [[ "$FILE_COUNT" -eq 0 ]]; then + echo "✗ No files found in ./${INPUT_DIR}/. Run ./download-testdata.sh first." >&2 + exit 1 +fi + +echo "▶ Retrieving S3 bucket name from stack '${STACK_NAME}'..." +BUCKET_NAME=$(aws cloudformation describe-stacks \ + --stack-name "$STACK_NAME" \ + --query "Stacks[0].Outputs[?OutputKey=='FilesBucketName'].OutputValue" \ + --output text) + +if [[ -z "$BUCKET_NAME" || "$BUCKET_NAME" == "None" ]]; then + echo "✗ Could not find FilesBucketName output. Is the stack deployed?" >&2 + exit 1 +fi + +echo " Bucket: ${BUCKET_NAME}" +echo " Prefix: ${PREFIX}" +echo " Files: ${FILE_COUNT}" +echo "" +echo "▶ Uploading files to s3://${BUCKET_NAME}/${PREFIX}..." + +aws s3 sync "$INPUT_DIR" "s3://${BUCKET_NAME}/${PREFIX}" \ + --content-type "text/csv" \ + --quiet + +echo "" +echo "✓ Uploaded ${FILE_COUNT} files to s3://${BUCKET_NAME}/${PREFIX}" +echo "" +echo "To invoke the batch processor, run:" +echo "" + +FUNCTION_ARN=$(aws cloudformation describe-stacks \ + --stack-name "$STACK_NAME" \ + --query "Stacks[0].Outputs[?OutputKey=='BatchProcessorFunctionArn'].OutputValue" \ + --output text) + +echo " aws lambda invoke \\" +echo " --function-name \"${FUNCTION_ARN}:\\\$LATEST\" \\" +echo " --invocation-type Event \\" +echo " --cli-binary-format raw-in-base64-out \\" +echo " --payload '{\"Bucket\":\"${BUCKET_NAME}\",\"Prefix\":\"${PREFIX}\"}' \\" +echo " /tmp/invoke-response.json"