s3-lambda-ddb-pipeline

AWSTemplateFormatVersion: "2010-09-09"
Description: S3 -> Lambda -> DynamoDB event pipeline

Resources:

  EventsKey:
    Type: AWS::KMS::Key
    Properties:
      Description: CMK for pipeline-events DynamoDB table
      EnableKeyRotation: true
      KeyPolicy:
        Version: "2012-10-17"
        Statement:
          - Sid: EnableRootAccess
            Effect: Allow
            Principal:
              AWS: !Sub "arn:aws:iam::${AWS::AccountId}:root"
            Action: "kms:*"
            Resource: "*"

  EventsTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: pipeline-events
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: object_key
          AttributeType: S
      KeySchema:
        - AttributeName: object_key
          KeyType: HASH
      SSESpecification:
        SSEEnabled: true
        SSEType: KMS
        KMSMasterKeyId: !GetAtt EventsKey.Arn

  LambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Policies:
        - PolicyName: WriteEvents
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - dynamodb:PutItem
                Resource: !GetAtt EventsTable.Arn
              - Effect: Allow
                Action:
                  - kms:Decrypt
                  - kms:GenerateDataKey
                Resource: !GetAtt EventsKey.Arn

  ProcessorFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: pipeline-processor
      Runtime: python3.11
      Handler: index.handler
      Role: !GetAtt LambdaExecutionRole.Arn
      Timeout: 30
      Environment:
        Variables:
          AWS_ENDPOINT_URL: http://localstack:4566
          EVENTS_TABLE: !Ref EventsTable
      Code:
        ZipFile: |
          import os
          import urllib.parse
          import boto3

          TABLE = os.environ["EVENTS_TABLE"]
          ENDPOINT = os.environ.get("AWS_ENDPOINT_URL")
          ddb = boto3.client("dynamodb", endpoint_url=ENDPOINT) if ENDPOINT else boto3.client("dynamodb")

          def handler(event, context):
              for record in event.get("Records", []):
                  s3 = record["s3"]
                  key = urllib.parse.unquote_plus(s3["object"]["key"])
                  bucket = s3["bucket"]["name"]
                  size = int(s3["object"].get("size", 0))
                  event_time = record.get("eventTime", "")
                  ddb.put_item(
                      TableName=TABLE,
                      Item={
                          "object_key": {"S": key},
                          "bucket": {"S": bucket},
                          "size": {"N": str(size)},
                          "event_time": {"S": event_time},
                      },
                  )
              return {"status": "ok"}

  InvokePermission:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !GetAtt ProcessorFunction.Arn
      Action: lambda:InvokeFunction
      Principal: s3.amazonaws.com
      SourceAccount: !Ref AWS::AccountId
      SourceArn: !Sub "arn:aws:s3:::pipeline-source"

  SourceBucket:
    Type: AWS::S3::Bucket
    DependsOn: InvokePermission
    Properties:
      BucketName: pipeline-source
      NotificationConfiguration:
        LambdaConfigurations:
          - Event: s3:ObjectCreated:*
            Function: !GetAtt ProcessorFunction.Arn

Outputs:
  BucketName:
    Value: !Ref SourceBucket
  TableName:
    Value: !Ref EventsTable
  FunctionName:
    Value: !Ref ProcessorFunction
  KeyArn:
    Value: !GetAtt EventsKey.Arn