AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Run large numbers of voice synthesis tasks through Amazon Polly asynchronously.

Globals:
  Function:
    Timeout: 3

Parameters: 
  NotificationEmail: 
    Description: The email address for notification of job completion
    Type: String
  WorkBucket:
    Description: A new S3 bucket where generated voice files will go
    Type: String

Resources:

  # The Set Processor reads set files that are placed in the working bucket 
  SetProcessorLambdaFunction:
    Type: 'AWS::Serverless::Function'
    Properties:
      CodeUri: amazon_polly_async_batch/
      Handler: setprocessor.process_set
      Runtime: python3.7
      FunctionName: polly-batch-set-processor
      MemorySize: 4096
      Timeout: 900
      Description: Submits text-to-speech jobs specified in a YAML file to SQS queue
      Environment:
        Variables:
          WORK_BUCKET: !Ref WorkBucket
          ITEM_QUEUE: polly-batch.fifo
          TASK_TABLE: polly-batch-tasks
          SET_TABLE: polly-batch-sets
          RESPONSE_TOPIC: polly-batch-topic
          SET_COMPLETE_WAITER_ARN: !Ref PollyBatchSetWaiter
      Policies:
        - Statement:
          - Effect: Allow
            Action:
              - 'logs:CreateLogStream'
              - 'logs:CreateLogGroup'
              - 'logs:PutLogEvents'
            Resource:
              - !Sub >-
                arn:${AWS::Partition}:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/polly-batch-set-processor:*:*
          - Effect: Allow
            Action:
              - 'sqs:GetQueueUrl'
              - 'sqs:SendMessage'
            Resource:
              - !GetAtt 
                - ItemQueue
                - Arn
          - Effect: Allow
            Action:
              - 's3:GetObject'
            Resource: 
            - !Join 
              - ''
              - - 'arn:'
                - !Ref 'AWS::Partition'
                - ':s3:::'
                - !Ref WorkBucket
                - /*
          - Effect: Allow
            Action:
              - 'dynamodb:PutItem'
            Resource:
              - !GetAtt 
                - SetTable
                - Arn
          - Effect: Allow
            Action:
              - 'states:StartExecution'
            Resource: !Ref PollyBatchSetWaiter
      Events:
        S3NewObjectEvent:
          Type: S3
          Properties:
            Bucket: !Ref PollyBatchWorkBucket
            Events: s3:ObjectCreated:*
            Filter:
              S3Key:
                Rules:
                  - Name: suffix
                    Value: ".yml"
    DependsOn:
      - SetProcessorLogGroup
  SetProcessorLogGroup:
    Type: 'AWS::Logs::LogGroup'
    Properties:
      LogGroupName: /aws/lambda/polly-batch-set-processor
  PollyBatchWorkBucket:
    Type: 'AWS::S3::Bucket'
    Properties:
      BucketName: !Ref WorkBucket
      BucketEncryption: 
        ServerSideEncryptionConfiguration: 
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: 'aws:kms'
        
      PublicAccessBlockConfiguration: 
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true


  SetProcessorLambdaPermissionPollyS3:
    Type: 'AWS::Lambda::Permission'
    Properties:
      FunctionName: !GetAtt 
        - SetProcessorLambdaFunction
        - Arn
      Action: 'lambda:InvokeFunction'
      Principal: s3.amazonaws.com
      SourceArn: !Join 
        - ''
        - - 'arn:'
          - !Ref 'AWS::Partition'
          - ':s3:::'
          - !Ref WorkBucket
      SourceAccount: !Ref 'AWS::AccountId'
  SetProcessorLambdaPermissionPollyBatch:
    Type: 'AWS::Lambda::Permission'
    Properties:
      FunctionName: !GetAtt 
        - SetProcessorLambdaFunction
        - Arn
      Action: 'lambda:InvokeFunction'
      Principal: s3.amazonaws.com
      SourceArn: !Join 
        - ''
        - - 'arn:'
          - !Ref 'AWS::Partition'
          - ':s3:::'
          - !Ref WorkBucket
      SourceAccount: !Ref 'AWS::AccountId'
  SetNotification:
    Type: 'AWS::SNS::Topic'
    Properties:
      DisplayName: Amazon Polly Async Batch
      TopicName: polly-batch-set-complete
      KmsMasterKeyId: alias/aws/sns
        
  # Policy for SNS Topic polly-Batch-Set-Complete
  SNSPollyCompleteBatchPolicy:
    Type: AWS::SNS::TopicPolicy
    Properties:
      PolicyDocument:
        Id: TopicPolicy
        Version: '2012-10-17'
        Statement:
        - Sid: Topic-Statement-1
          Effect: Allow
          Principal:
            Service: lambda.amazonaws.com
          Action: sns:Publish
          Resource: !Ref SetNotification
      Topics:
      - !Ref SetNotification
  # Policy for SNS Topic polly-Batch
  SNSPollyBatchPolicy:
    Type: AWS::SNS::TopicPolicy
    Properties:
      PolicyDocument:
        Id: TopicBatchPolicy
        Version: '2012-10-17'
        Statement:
        - Sid: Topic-Statement-1
          Effect: Allow
          Principal:
            Service: lambda.amazonaws.com
          Action: sns:Publish
          Resource: !Ref SNSTopicPollyBatch
      Topics:
      - !Ref SNSTopicPollyBatch
  
  SNSTopicPollyBatchSubscription:
    Type: AWS::SNS::Subscription
    Properties:
      Endpoint: !Ref NotificationEmail
      Protocol: email
      TopicArn: !Ref SetNotification
    DependsOn : SetNotification

  # The Item Processor takes items off the SQS queue and sends them to Polly
  ItemProcessorLambdaFunction:
    Type: 'AWS::Serverless::Function'
    Properties:
      CodeUri: amazon_polly_async_batch/
      Handler: itemprocessor.process_sqs_message
      Runtime: python3.7
      FunctionName: polly-batch-item-processor
      MemorySize: 128
      Timeout: 6
      Description: Reads items from SQS queue and submits them as tasks to Polly
      Environment:
        Variables:
          WORK_BUCKET: !Ref WorkBucket
          ITEM_QUEUE: polly-batch.fifo
          TASK_TABLE: polly-batch-tasks
          SET_TABLE: polly-batch-sets
          RESPONSE_TOPIC: polly-batch-topic
          RESPONSE_TOPIC_ARN: !Ref SNSTopicPollyBatch
      Policies:
        - Statement:
            - Effect: Allow
              Action:
                - 'logs:CreateLogStream'
                - 'logs:CreateLogGroup'
                - 'logs:PutLogEvents'
              Resource:
                - !Sub >-
                  arn:${AWS::Partition}:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/polly-batch-item-processor:*:*
            - Effect: Allow
              Action:
                - 'sqs:ReceiveMessage'
                - 'sqs:DeleteMessage'
                - 'sqs:GetQueueAttributes'
              Resource:
                - !GetAtt 
                  - ItemQueue
                  - Arn
            - Effect: Allow
              Action:
                - 'polly:StartSpeechSynthesisTask'
              Resource: '*'
            - Effect: Allow
              Action:
                - 's3:PutObject'
              Resource:
                - !Join 
                  - ''
                  - - 'arn:'
                    - !Ref 'AWS::Partition'
                    - ':s3:::'
                    - !Ref WorkBucket
                    - /*
            - Effect: Allow
              Action:
                - 'sns:Publish'
              Resource: !Ref SNSTopicPollyBatch
            - Effect: Allow
              Action:
                - 'dynamodb:DeleteItem'
                - 'dynamodb:GetItem'
                - 'dynamodb:PutItem'
                - 'dynamodb:UpdateItem'
              Resource:
                - !GetAtt 
                  - TaskTable
                  - Arn
                - !GetAtt 
                  - SetTable
                  - Arn
            - Effect: Allow
              Action:
                - 'sqs:ReceiveMessage'
                - 'sqs:DeleteMessage'
                - 'sqs:GetQueueAttributes'
              Resource:
                - !GetAtt 
                  - ItemQueue
                  - Arn
    DependsOn:
      - ItemProcessorLogGroup
  ItemProcessorLogGroup:
    Type: 'AWS::Logs::LogGroup'
    Properties:
      LogGroupName: /aws/lambda/polly-batch-item-processor
  ItemProcessorEventSourceMappingSQSItemQueue:
    Type: 'AWS::Lambda::EventSourceMapping'
    Properties:
      BatchSize: 1
      EventSourceArn: !GetAtt 
        - ItemQueue
        - Arn
      FunctionName: !GetAtt 
        - ItemProcessorLambdaFunction
        - Arn
      Enabled: true
      
  # The queue of work is FIFO to minimize fanning out due to service quota limits; see setprocessor.py
  ItemQueue:
    Type: 'AWS::SQS::Queue'
    Properties:
      QueueName: polly-batch.fifo
      FifoQueue: true
      KmsMasterKeyId: alias/aws/sqs
      VisibilityTimeout: 60

  # The Response Processor handles notifications from Polly
  ResponseProcessorLambdaFunction:
    Type: 'AWS::Serverless::Function'
    Properties:
      CodeUri: amazon_polly_async_batch/
      Handler: responseprocessor.process_sns_message
      Runtime: python3.7
      FunctionName: polly-batch-response-processor
      MemorySize: 128
      Description: Given a response from Polly updates DynamoDB records
      Environment:
        Variables:
          WORK_BUCKET: !Ref WorkBucket
          ITEM_QUEUE: polly-batch.fifo
          TASK_TABLE: polly-batch-tasks
          SET_TABLE: polly-batch-sets
          RESPONSE_TOPIC: polly-batch-topic
      Policies: 
        - Statement:
          - Effect: Allow
            Action:
              - 'logs:CreateLogStream'
              - 'logs:CreateLogGroup'
              - 'logs:PutLogEvents'
            Resource:
              - !Sub >-
                arn:${AWS::Partition}:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/polly-batch-response-processor:*:*
          - Effect: Allow
            Action:
              - 'dynamodb:GetItem'
              - 'dynamodb:PutItem'
              - 'dynamodb:UpdateItem'
            Resource:
              - !GetAtt 
                - TaskTable
                - Arn
              - !GetAtt 
                - SetTable
                - Arn
          - Effect: Allow
            Action:
              - 's3:GetObject'
              - 's3:DeleteObject'
              - 's3:PutObject'
            Resource:
             - !Join 
              - ''
              - - 'arn:'
                - !Ref 'AWS::Partition'
                - ':s3:::'
                - !Ref WorkBucket
                - '/*'
    DependsOn:
      - ResponseProcessorLogGroup
  ResponseProcessorLogGroup:
    Type: 'AWS::Logs::LogGroup'
    Properties:
      LogGroupName: /aws/lambda/polly-batch-response-processor
  SNSTopicPollyBatch:
    Type: 'AWS::SNS::Topic'
    Properties:
      TopicName: polly-batch-topic
      Subscription: 
      - Endpoint: !GetAtt 
            - ResponseProcessorLambdaFunction
            - Arn
        Protocol: lambda
      KmsMasterKeyId: alias/aws/sns
  ResponseProcessorLambdaPermissionPollyBatchSNS:
    Type: 'AWS::Lambda::Permission'
    Properties:
      FunctionName: !Ref 'ResponseProcessorLambdaFunction'
      Action: 'lambda:InvokeFunction'
      Principal: sns.amazonaws.com
      SourceArn: !Ref SNSTopicPollyBatch

  # The Set Completion step function watches for a set completing
  SetCompletionLoaderLambdaFunction:
    Type: 'AWS::Serverless::Function'
    Properties:
      CodeUri: amazon_polly_async_batch/
      Handler: setcompletionwaiter.load_set_details
      Runtime: python3.7
      FunctionName: polly-batch-set-completion-loader
      MemorySize: 128
      Description: Loads a set from the database to see if it's finished
      Environment:
        Variables:
          SET_COMPLETE_TOPIC_ARN: !Ref SetNotification
          RESPONSE_TOPIC: polly-batch-topic
          RESPONSE_TOPIC_ARN: !Ref SNSTopicPollyBatch
          SET_TABLE: polly-batch-sets
          WORK_BUCKET: !Ref WorkBucket
      Policies: 
        - Statement:
          - Effect: Allow
            Action:
              - 'logs:CreateLogStream'
              - 'logs:CreateLogGroup'
              - 'logs:PutLogEvents'
            Resource:
              - !Sub >-
                arn:${AWS::Partition}:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/polly-batch-set-completion-loader:*:*
          - Effect: Allow
            Action:
              - 'dynamodb:GetItem'
            Resource:
              - !GetAtt 
                - SetTable
                - Arn
    DependsOn:
      - SetCompletionLoaderLogGroup
  SetCompletionNotifierLambdaFunction:
    Type: 'AWS::Serverless::Function'
    Properties:
      CodeUri: amazon_polly_async_batch/
      Handler: setcompletionwaiter.notify_of_set_completion
      Runtime: python3.7
      FunctionName: polly-batch-set-completion-notifier
      MemorySize: 128
      Description: 'After a set is completed, notifies the appropriate topic'
      Environment:
        Variables:
          SET_COMPLETE_TOPIC_ARN: !Ref SetNotification
          SET_TABLE: polly-batch-sets
          WORK_BUCKET: !Ref WorkBucket
      Policies: 
      - Statement:
        - Effect: Allow
          Action:
            - 'logs:CreateLogStream'
            - 'logs:CreateLogGroup'
            - 'logs:PutLogEvents'
          Resource:
            - !Sub >-
              arn:${AWS::Partition}:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/polly-batch-set-completion-notifier:*:*
        - Effect: Allow
          Action:
            - 'dynamodb:GetItem'
          Resource:
            - !GetAtt 
              - SetTable
              - Arn
        - Effect: Allow
          Action:
            - 'sns:Publish'
          Resource: !Ref SetNotification
    DependsOn:
      - SetCompletionNotifierLogGroup
  SetProblemNotifierLambdaFunction:
    Type: 'AWS::Serverless::Function'
    Properties:
      CodeUri: amazon_polly_async_batch/
      Handler: setcompletionwaiter.notify_of_set_problem
      Runtime: python3.7
      FunctionName: polly-batch-set-problem-notifier
      MemorySize: 128
      Description: 'After a set is completed, notifies the appropriate topic'
      Environment:
        Variables:
          SET_COMPLETE_TOPIC_ARN: !Ref SetNotification
          SET_TABLE: polly-batch-sets
          WORK_BUCKET: !Ref WorkBucket
      Policies: 
        - Statement:
          - Effect: Allow
            Action:
              - 'logs:CreateLogStream'
              - 'logs:CreateLogGroup'
              - 'logs:PutLogEvents'
            Resource:
              - !Sub >-
                arn:${AWS::Partition}:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/polly-batch-dev-set-problem-notifier:*:*
          - Effect: Allow
            Action:
              - 'dynamodb:GetItem'
            Resource:
              - !GetAtt 
                - SetTable
                - Arn
          - Effect: Allow
            Action:
              - 'sns:Publish'
            Resource: !Ref SetNotification
    DependsOn:
      - SetProblemNotifierLogGroup
  SetCompletionLoaderLogGroup:
    Type: 'AWS::Logs::LogGroup'
    Properties:
      LogGroupName: /aws/lambda/polly-batch-dev-set-completion-loader
  SetCompletionNotifierLogGroup:
    Type: 'AWS::Logs::LogGroup'
    Properties:
      LogGroupName: /aws/lambda/polly-batch-dev-set-completion-notifier
  SetProblemNotifierLogGroup:
    Type: 'AWS::Logs::LogGroup'
    Properties:
      LogGroupName: /aws/lambda/polly-batch-dev-set-problem-notifier
  PollyBatchSetWaiterRole:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service: states.amazonaws.com
            Action: 'sts:AssumeRole'
      Policies:
        - PolicyName: polly-batch-statemachine
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - 'lambda:InvokeFunction'
                Resource:
                                
                  - !Sub >-
                     arn:${AWS::Partition}:lambda:${AWS::Region}:${AWS::AccountId}:function:polly-batch-set-completion-loader
                  - !Sub >-
                     arn:${AWS::Partition}:lambda:${AWS::Region}:${AWS::AccountId}:function:polly-batch-set-completion-notifier
                  - !Sub >-
                     arn:${AWS::Partition}:lambda:${AWS::Region}:${AWS::AccountId}:function:polly-batch-set-problem-notifier
  PollyBatchSetWaiter:
    Type: 'AWS::Serverless::StateMachine'
    Properties:
      DefinitionUri: amazon_polly_async_batch/pollybatchsetwaiter.asl.json  
      Role: !GetAtt 
        - PollyBatchSetWaiterRole
        - Arn

  # Work is tracked in DynamoDb
  TaskTable:
    Type: 'AWS::DynamoDB::Table'
    Properties:
      TableName: polly-batch-tasks
      AttributeDefinitions:
        - AttributeName: taskId
          AttributeType: S
      KeySchema:
        - AttributeName: taskId
          KeyType: HASH
      BillingMode: PAY_PER_REQUEST
      SSESpecification: 
        SSEEnabled:  false
     
  SetTable:
    Type: 'AWS::DynamoDB::Table'
    Properties:
      TableName: polly-batch-sets
      SSESpecification: 
        SSEEnabled:  false
      AttributeDefinitions:
        - AttributeName: setName
          AttributeType: S
      KeySchema:
        - AttributeName: setName
          KeyType: HASH
      BillingMode: PAY_PER_REQUEST