CloudFormation template that demonstrates how to start a Step Functions State Machine using DynamoDB Streams and EventBridge Pipes

cfn-ddb-sfn.yml

AWSTemplateFormatVersion: "2010-09-09"
Description: Creates a Step Functions State Machine that is executed when new items are added to a DynamoDB Table.
Resources:
  Table:
    Type: "AWS::DynamoDB::Table"
    Properties:
      AttributeDefinitions:
        - AttributeName: "PK"
          AttributeType: "S"
      KeySchema:
        - AttributeName: "PK"
          KeyType: "HASH"
      BillingMode: PAY_PER_REQUEST
      StreamSpecification:
        StreamViewType: NEW_IMAGE
    UpdateReplacePolicy: "Delete"
    DeletionPolicy: "Delete"

  StateMachineRole:
    Type: "AWS::IAM::Role"
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          Action: "sts:AssumeRole"
          Effect: "Allow"
          Principal:
            Service: "states.amazonaws.com"

  StateMachine:
    Type: "AWS::StepFunctions::StateMachine"
    Properties:
      RoleArn: !GetAtt StateMachineRole.Arn
      DefinitionString: |
        {
          "StartAt": "DummyState",
          "States": {
            "DummyState": {
              "Type": "Pass",
              "End": true
            }
          }
        }

  EventBus:
    Type: "AWS::Events::EventBus"
    Properties:
      Name: !Sub "${AWS::StackName}-bus"

  PipesRole:
    Type: "AWS::IAM::Role"
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          Action: "sts:AssumeRole"
          Effect: "Allow"
          Principal:
            Service: "pipes.amazonaws.com"

  PipesRolePolicy:
    Type: "AWS::IAM::Policy"
    Properties:
      PolicyName: !Sub "${AWS::StackName}-eventbridge-pipes-policy"
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Action: "events:PutEvents"
            Resource: !GetAtt EventBus.Arn
          - Effect: "Allow"
            Action: "dynamodb:ListStreams"
            Resource: "*"
          - Effect: "Allow"
            Action:
              - "dynamodb:DescribeStream"
              - "dynamodb:GetRecords"
              - "dynamodb:GetShardIterator"
            Resource: !GetAtt Table.StreamArn
          - Effect: "Allow"
            Action: "states:StartExecution"
            Resource: !Ref StateMachine
      Roles:
        - !Ref PipesRole

  Pipe:
    Type: "AWS::Pipes::Pipe"
    Properties:
      RoleArn: !GetAtt PipesRole.Arn
      Source: !GetAtt Table.StreamArn
      SourceParameters:
        FilterCriteria:
          Filters:
            - Pattern: |
                {
                  "eventName": ["INSERT"]
                }
        DynamoDBStreamParameters:
          BatchSize: 1
          StartingPosition: "LATEST"
      Target: !GetAtt EventBus.Arn
      TargetParameters:
        EventBridgeEventBusParameters:
          DetailType: "ItemCreated"
        InputTemplate: |
          {
            "itemId": <$.dynamodb.Keys.PK.S>,
            "foo": "bar"
          }

  EventsRuleRole:
    Type: "AWS::IAM::Role"
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          Action: "sts:AssumeRole"
          Effect: "Allow"
          Principal:
            Service: "events.amazonaws.com"
      Policies:
        - PolicyName: "events-rule-policy"
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: "Allow"
                Action: "states:StartExecution"
                Resource: !Ref StateMachine

  EventsRule:
    Type: "AWS::Events::Rule"
    Properties:
      EventBusName: !Ref EventBus
      EventPattern: |
        {
          "detail-type": ["ItemCreated"]
        }
      Targets:
        - Arn: !Ref StateMachine
          Id: StateMachineTarget
          RoleArn: !GetAtt EventsRuleRole.Arn