dsql-cdc-eventbridge-fanout-cdk: DSQL CDC to EventBridge fan-out

dsql-cdc-eventbridge-fanout-cdk/README.md

@@ -0,0 +1,109 @@
+# Amazon Aurora DSQL CDC to Amazon EventBridge
+
+This pattern deploys an event-driven pipeline that captures real-time database changes from Amazon Aurora DSQL using Change Data Capture (CDC), streams them through Amazon Kinesis Data Streams, processes them with AWS Lambda, and publishes typed events to an Amazon EventBridge custom event bus for downstream consumption.
+
+Learn more about this pattern at Serverless Land Patterns: https://serverlessland.com/patterns/dsql-cdc-eventbridge-fanout-cdk
+
+Important: this application uses various AWS services and there are costs associated with these services after the Free Tier usage - please see the [AWS Pricing page](https://aws.amazon.com/pricing/) for details.
+
+## Architecture
+
+```
+┌──────────────────┐     ┌─────────────────────┐     ┌──────────────────────┐     ┌─────────────────────┐
+│ Amazon Aurora    │────▶│ Amazon Kinesis Data  │────▶│ AWS Lambda           │────▶│ Amazon EventBridge   │
+│ DSQL (CDC)       │     │ Streams              │     │ (CDC Processor)      │     │ (Custom Bus)         │
+└──────────────────┘     └─────────────────────┘     └──────────────────────┘     └─────────────────────┘
+```
+
+**How it works:**
+
+1. Amazon Aurora DSQL captures every committed row-level change (INSERT, UPDATE, DELETE) and delivers it as a structured JSON record to Amazon Kinesis Data Streams.
+2. AWS Lambda consumes the Amazon Kinesis stream, parses the CDC payload (Debezium-style op codes), classifies the operation type, and publishes typed events to an Amazon EventBridge custom event bus.
+3. Amazon EventBridge receives events with source `dsql.cdc` and detail-type `INSERT`, `UPDATE`, or `DELETE`. Add your own rules and targets to route events to any downstream consumer.
+
+## Requirements
+
+- [AWS CDK v2](https://docs.aws.amazon.com/cdk/v2/guide/getting_started.html) installed and configured
+- [Node.js 20+](https://nodejs.org/) with npm
+- AWS account [bootstrapped for CDK](https://docs.aws.amazon.com/cdk/v2/guide/bootstrapping.html)
+- An existing Amazon Aurora DSQL cluster
+- Python 3.12 (for AWS Lambda functions)
+
+## Deployment
+
+1. Create an Amazon Aurora DSQL cluster (if you don't have one):
+
+    ```bash
+    aws dsql create-cluster --region us-east-1
+    ```
+
+    Note the `identifier` from the response.
+
+2. Install dependencies and build:
+
+    ```bash
+    cd dsql-cdc-eventbridge-fanout-cdk/cdk
+    npm install
+    npm run build
+    ```
+
+3. Deploy the stack:
+
+    ```bash
+    npx cdk deploy --parameters DsqlClusterId=<your-cluster-id>
+    ```
+
+## Testing
+
+After deploying, insert data into your Amazon Aurora DSQL cluster to trigger CDC events:
+
+```sql
+CREATE TABLE orders (
+  id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
+  customer_name TEXT NOT NULL,
+  amount DECIMAL(10,2),
+  created_at TIMESTAMP DEFAULT now()
+);
+
+INSERT INTO orders (customer_name, amount) VALUES ('Acme Corp', 1250.00);
+UPDATE orders SET amount = 1500.00 WHERE customer_name = 'Acme Corp';
+DELETE FROM orders WHERE customer_name = 'Acme Corp';
+```
+
+Then verify events arrive on the custom event bus by adding a temporary rule:
+
+```bash
+aws events put-rule \
+  --name test-cdc-rule \
+  --event-bus-name dsql-cdc-events \
+  --event-pattern '{"source": ["dsql.cdc"]}'
+```
+
+## Extending This Pattern
+
+Add Amazon EventBridge rules and targets to route CDC events to any consumer:
+
+- Route ALL changes to Amazon SQS for audit
+- Route INSERT events to AWS Step Functions for validation
+- Route DELETE events to Amazon SNS for alerting
+
+## Cleanup
+
+> **Warning:** This will delete all resources. The Amazon Aurora DSQL cluster is NOT deleted (it was created externally).
+
+```bash
+npx cdk destroy
+```
+
+## Services Used
+
+| Service | Role |
+|---------|------|
+| Amazon Aurora DSQL | Source database with CDC enabled |
+| Amazon Kinesis Data Streams | Receives CDC event stream from Amazon Aurora DSQL |
+| AWS Lambda | Processes CDC events, classifies operations, publishes to Amazon EventBridge |
+| Amazon EventBridge | Custom event bus for content-based routing of CDC events |
+
+----
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0

dsql-cdc-eventbridge-fanout-cdk/cdk/.gitignore

@@ -0,0 +1,6 @@
+node_modules
+cdk.out
+cdk.context.json
+build
+*.js
+*.d.ts

dsql-cdc-eventbridge-fanout-cdk/cdk/bin/app.ts

@@ -0,0 +1,12 @@
+#!/usr/bin/env node
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT-0 (2026)
+
+import 'source-map-support/register';
+import * as cdk from 'aws-cdk-lib';
+import { DsqlCdcEventbridgeFanoutStack } from '../lib/dsql-cdc-eventbridge-fanout-stack';
+
+const app = new cdk.App();
+new DsqlCdcEventbridgeFanoutStack(app, 'DsqlCdcEventbridgeFanoutStack', {
+  description: 'Amazon Aurora DSQL CDC to Amazon EventBridge fan-out pattern (uksb-1tupboc57)',
+});

dsql-cdc-eventbridge-fanout-cdk/cdk/cdk.json

@@ -0,0 +1,3 @@
+{
+  "app": "npx ts-node --prefer-ts-exts bin/app.ts"
+}

dsql-cdc-eventbridge-fanout-cdk/cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts

@@ -0,0 +1,181 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT-0 (2026)
+
+import * as cdk from 'aws-cdk-lib';
+import * as kinesis from 'aws-cdk-lib/aws-kinesis';
+import * as lambda from 'aws-cdk-lib/aws-lambda';
+import * as iam from 'aws-cdk-lib/aws-iam';
+import * as events from 'aws-cdk-lib/aws-events';
+import * as cr from 'aws-cdk-lib/custom-resources';
+import * as kinesisEvtSrc from 'aws-cdk-lib/aws-lambda-event-sources';
+import { Construct } from 'constructs';
+import * as path from 'path';
+
+export class DsqlCdcEventbridgeFanoutStack extends cdk.Stack {
+  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
+    super(scope, id, props);
+
+    const region = cdk.Stack.of(this).region;
+    const account = cdk.Stack.of(this).account;
+
+    // --- Context parameters ---
+    const dsqlClusterId = new cdk.CfnParameter(this, 'DsqlClusterId', {
+      type: 'String',
+      description: 'Amazon Aurora DSQL cluster identifier',
+    });
+
+    // =========================================================
+    // 1. Amazon Kinesis Data Stream (CDC target)
+    // =========================================================
+    const cdcStream = new kinesis.Stream(this, 'CdcStream', {
+      streamName: `dsql-cdc-${cdk.Names.uniqueId(this).slice(-8).toLowerCase()}`,
+      shardCount: 1,
+      retentionPeriod: cdk.Duration.hours(24),
+      encryption: kinesis.StreamEncryption.MANAGED,
+    });
+
+    // =========================================================
+    // 2. IAM Role for Amazon Aurora DSQL to write to Amazon Kinesis
+    // =========================================================
+    const dsqlCdcRole = new iam.Role(this, 'DsqlCdcRole', {
+      assumedBy: new iam.ServicePrincipal('dsql.amazonaws.com'),
+      description: 'Allows Amazon Aurora DSQL CDC to write to Amazon Kinesis',
+      inlinePolicies: {
+        KinesisWrite: new iam.PolicyDocument({
+          statements: [
+            new iam.PolicyStatement({
+              effect: iam.Effect.ALLOW,
+              actions: [
+                'kinesis:PutRecord',
+                'kinesis:PutRecords',
+                'kinesis:DescribeStream',
+              ],
+              resources: [cdcStream.streamArn],
+            }),
+          ],
+        }),
+      },
+    });
+
+    // =========================================================
+    // 3. Custom Resource: Create/Delete DSQL CDC Stream
+    //    (No CFN resource type for DSQL streams yet — use SDK)
+    // =========================================================
+    const streamManagerFn = new lambda.Function(this, 'CdcStreamManagerFn', {
+      runtime: lambda.Runtime.PYTHON_3_12,
+      handler: 'handler.on_event',
+      code: lambda.Code.fromAsset(path.join(__dirname, '../../lambdas/cdc-stream-manager'), {
+        bundling: {
+          image: lambda.Runtime.PYTHON_3_12.bundlingImage,
+          command: [
+            'bash', '-c',
+            'pip install -r requirements.txt -t /asset-output && cp handler.py /asset-output/',
+          ],
+        },
+      }),
+      timeout: cdk.Duration.minutes(5),
+      memorySize: 256,
+      description: 'Custom Resource: manages Amazon Aurora DSQL CDC stream lifecycle',
+      environment: {
+        CLUSTER_ID: dsqlClusterId.valueAsString,
+        KINESIS_STREAM_ARN: cdcStream.streamArn,
+        CDC_ROLE_ARN: dsqlCdcRole.roleArn,
+      },
+    });
+
+    streamManagerFn.addToRolePolicy(new iam.PolicyStatement({
+      effect: iam.Effect.ALLOW,
+      actions: [
+        'dsql:CreateStream',
+        'dsql:DeleteStream',
+        'dsql:GetStream',
+        'dsql:ListStreams',
+      ],
+      resources: [
+        `arn:aws:dsql:${region}:${account}:cluster/${dsqlClusterId.valueAsString}`,
+        `arn:aws:dsql:${region}:${account}:cluster/${dsqlClusterId.valueAsString}/stream/*`,
+      ],
+    }));
+
+    streamManagerFn.addToRolePolicy(new iam.PolicyStatement({
+      effect: iam.Effect.ALLOW,
+      actions: ['iam:PassRole'],
+      resources: [dsqlCdcRole.roleArn],
+      conditions: {
+        StringEquals: { 'iam:PassedToService': 'dsql.amazonaws.com' },
+      },
+    }));
+
+    const cdcStreamCr = new cr.Provider(this, 'CdcStreamProvider', {
+      onEventHandler: streamManagerFn,
+    });
+
+    const cdcStreamResource = new cdk.CustomResource(this, 'DsqlCdcStream', {
+      serviceToken: cdcStreamCr.serviceToken,
+      properties: {
+        ClusterId: dsqlClusterId.valueAsString,
+        KinesisStreamArn: cdcStream.streamArn,
+        RoleArn: dsqlCdcRole.roleArn,
+      },
+    });
+
+    // =========================================================
+    // 4. Amazon EventBridge Custom Event Bus
+    // =========================================================
+    const cdcEventBus = new events.EventBus(this, 'CdcEventBus', {
+      eventBusName: 'dsql-cdc-events',
+    });
+
+    // =========================================================
+    // 5. AWS Lambda: CDC Processor (Amazon Kinesis → Amazon EventBridge)
+    // =========================================================
+    const cdcProcessorFn = new lambda.Function(this, 'CdcProcessorFn', {
+      runtime: lambda.Runtime.PYTHON_3_12,
+      handler: 'handler.lambda_handler',
+      code: lambda.Code.fromAsset(path.join(__dirname, '../../lambdas/cdc-processor')),
+      timeout: cdk.Duration.seconds(60),
+      memorySize: 256,
+      description: 'Processes Amazon Aurora DSQL CDC events and publishes to Amazon EventBridge',
+      environment: {
+        EVENT_BUS_NAME: cdcEventBus.eventBusName,
+      },
+    });
+
+    cdcProcessorFn.addToRolePolicy(new iam.PolicyStatement({
+      effect: iam.Effect.ALLOW,
+      actions: ['events:PutEvents'],
+      resources: [cdcEventBus.eventBusArn],
+    }));
+
+    cdcProcessorFn.addEventSource(new kinesisEvtSrc.KinesisEventSource(cdcStream, {
+      startingPosition: lambda.StartingPosition.TRIM_HORIZON,
+      batchSize: 100,
+      maxBatchingWindow: cdk.Duration.seconds(5),
+      retryAttempts: 3,
+      bisectBatchOnError: true,
+    }));
+
+    // =========================================================
+    // Outputs
+    // =========================================================
+    new cdk.CfnOutput(this, 'KinesisStreamArn', {
+      value: cdcStream.streamArn,
+      description: 'Amazon Kinesis Data Stream ARN (CDC target)',
+    });
+
+    new cdk.CfnOutput(this, 'EventBusArn', {
+      value: cdcEventBus.eventBusArn,
+      description: 'Amazon EventBridge custom event bus ARN — add rules and targets to consume CDC events',
+    });
+
+    new cdk.CfnOutput(this, 'CdcProcessorFunctionName', {
+      value: cdcProcessorFn.functionName,
+      description: 'AWS Lambda CDC processor function name',
+    });
+
+    new cdk.CfnOutput(this, 'CdcStreamId', {
+      value: cdcStreamResource.getAttString('StreamId'),
+      description: 'Amazon Aurora DSQL CDC stream identifier',
+    });
+  }
+}

dsql-cdc-eventbridge-fanout-cdk/cdk/package.json

@@ -0,0 +1,20 @@
+{
+  "name": "dsql-cdc-eventbridge-fanout-cdk",
+  "version": "1.0.0",
+  "bin": { "app": "bin/app.ts" },
+  "scripts": {
+    "build": "tsc",
+    "synth": "cdk synth",
+    "deploy": "cdk deploy",
+    "destroy": "cdk destroy"
+  },
+  "dependencies": {
+    "aws-cdk-lib": "2.185.0",
+    "constructs": "^10.0.0",
+    "source-map-support": "^0.5.21"
+  },
+  "devDependencies": {
+    "typescript": "~5.4.0",
+    "ts-node": "^10.9.0"
+  }
+}

dsql-cdc-eventbridge-fanout-cdk/cdk/tsconfig.json

@@ -0,0 +1,24 @@
+{
+  "compilerOptions": {
+    "target": "ES2020",
+    "module": "commonjs",
+    "lib": ["es2020"],
+    "declaration": true,
+    "strict": true,
+    "noImplicitAny": true,
+    "strictNullChecks": true,
+    "noImplicitThis": true,
+    "alwaysStrict": true,
+    "noUnusedLocals": false,
+    "noUnusedParameters": false,
+    "noImplicitReturns": true,
+    "noFallthroughCasesInSwitch": false,
+    "inlineSourceMap": true,
+    "inlineSources": true,
+    "experimentalDecorators": true,
+    "strictPropertyInitialization": false,
+    "outDir": "./build",
+    "rootDir": "."
+  },
+  "exclude": ["node_modules", "build"]
+}