AWSTemplateFormatVersion: "2010-09-09"
Parameters:

  DynamoDBTable:
    Description: The name of your DynamoDB table to be created
    Type: String
    Default: salesitems
    AllowedPattern: "([a-z]|[0-9])+"
  DatabaseName:
    Description: The name of the first database to be created when the Redshift cluster is
      created
    Type: String
    Default: salesdwh
    AllowedPattern: "([a-z]|[0-9])+"
  MasterUsername:
    Description: The user name that is associated with the master user account for
      the cluster that is being created
    Type: String
    Default: awsuser
    AllowedPattern: "([a-z])([a-z]|[0-9])*"
  MasterUserPassword:
    Description: The password that is associated with the master user account for
      the cluster that is being created. Minimum eight characters, at least one uppercase letter, one lowercase letter and one number.
    Type: String
    AllowedPattern: ^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)[a-zA-Z\d]{8,}$
    NoEcho: 'true'

  PortNumber:
    Description: The port number on which the Redshift cluster accepts incoming connections.
    Type: Number
    Default: '5439'

  VPCCidr:
    Description: The IPv4 network range for the VPC
    Type: String
    Default: '10.0.0.0/16'

  PublicSubnetOneCidr:
    Description: The IPv4 network range for public subnet one
    Type: String
    Default: '10.0.0.1/20'

  PublicSubnetTwoCidr:
    Description: The IPv4 network range for public subnet two
    Type: String
    Default: '10.0.16.1/20'


Resources:
  
  ## S3 bucket resources
  ####################################################################################################################

  RawBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub ${AWS::StackName}-${AWS::AccountId}-raw

  DataLakeBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub ${AWS::StackName}-${AWS::AccountId}-datalake

  RawArchiveBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub ${AWS::StackName}-${AWS::AccountId}-raw-archive

  DatalakeArchiveBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub ${AWS::StackName}-${AWS::AccountId}-datalake-archive


  AthenaBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub 'athena-ddb-dwh-blog-result-${AWS::AccountId}-${AWS::Region}'


  ## NEED IAM Role for all resources
  ################################################################################################################################
  AllowBasicS3Access:
    Type: AWS::IAM::ManagedPolicy
    Properties:
      PolicyDocument:
        Version: 2012-10-17
        Statement:
         - Sid: AllowBasicS3Actions
           Effect: Allow
           Action:
             - s3:PutObject
             - s3:GetObject
             - s3:GetEncryptionConfiguration
             - s3:ListBucket
             - s3:DeleteObject
           Resource:
             - !Sub "arn:aws:s3:::${AWS::StackName}-${AWS::AccountId}-datalake"
             - !Sub "arn:aws:s3:::${AWS::StackName}-${AWS::AccountId}-datalake/*"
             - !Sub "arn:aws:s3:::${AWS::StackName}-${AWS::AccountId}-raw"
             - !Sub "arn:aws:s3:::${AWS::StackName}-${AWS::AccountId}-raw/*"
             - !Sub "arn:aws:s3:::${AWS::StackName}-${AWS::AccountId}-datalake-archive"
             - !Sub "arn:aws:s3:::${AWS::StackName}-${AWS::AccountId}-datalake-archive/*"
             - !Sub "arn:aws:s3:::${AWS::StackName}-${AWS::AccountId}-raw-archive"
             - !Sub "arn:aws:s3:::${AWS::StackName}-${AWS::AccountId}-raw-archive/*"
             - !Sub "arn:aws:s3:::aws-glue-assets-${AWS::AccountId}-${AWS::Region}"
             - !Sub "arn:aws:s3:::aws-glue-assets-${AWS::AccountId}-${AWS::Region}/*"


  DynamoDBexport:
    Type: AWS::IAM::ManagedPolicy
    Properties:
      PolicyDocument:
        Version: 2012-10-17
        Statement:
         - Sid: AllowDynamoDBExportAction
           Effect: Allow
           Action:
             - dynamodb:*
           Resource:
             - !Sub "arn:aws:dynamodb:${AWS::Region}:${AWS::AccountId}:table/${DynamoDBTable}"
             - !Sub "arn:aws:dynamodb:${AWS::Region}:${AWS::AccountId}:table/${DynamoDBTable}/*"

         - Sid: AllowWriteToDestinationBucket
           Effect: Allow
           Action:
             - s3:AbortMultipartUpload
             - s3:PutObject
             - s3:PutObjectAcl
           Resource:
             - !Sub arn:aws:s3:::${AWS::StackName}-${AWS::AccountId}-raw/*
             - !Sub "arn:aws:s3:::aws-glue-assets-${AWS::AccountId}-${AWS::Region}/*"


  GlueGetActions:
    Type: AWS::IAM::ManagedPolicy
    Properties:
      PolicyDocument:
        Version: 2012-10-17
        Statement:
         - Sid: GetGlueActions
           Effect: Allow
           Action:
             - glue:GetTable
             - glue:GetTables
             - glue:GetDatabase
             - glue:GetDatabases
             - glue:GetConnection
             - glue:GetConnections
           Resource:
             - "*"
 

  # We create our own service role for EC2 instances that allows full access to our S3 bucket
  MyGlueServiceRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - glue.amazonaws.com
            Action:
              - "sts:AssumeRole"
      Description: "Service role for Glue"
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonRedshiftDataFullAccess
        - arn:aws:iam::aws:policy/AmazonAthenaFullAccess
        - arn:aws:iam::aws:policy/AWSOpsWorksCloudWatchLogs
        - arn:aws:iam::aws:policy/AmazonEC2FullAccess
        - !Ref DynamoDBexport
        - !Ref AllowBasicS3Access
        - !Ref GlueGetActions

        
  MyStepFunctionServiceRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - states.amazonaws.com
            Action:
              - "sts:AssumeRole"
      Description: "Service role for Step Functions"
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonRedshiftFullAccess
        - arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole
        - arn:aws:iam::aws:policy/AWSOpsWorksCloudWatchLogs
        - !Ref AllowBasicS3Access

  GetPLResourceRole:
    Type: AWS::IAM::Role
    Properties:
        AssumeRolePolicyDocument:
          Version: 2012-10-17
          Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
        Path: "/"
        Policies:
          -
            PolicyName: "EC2Permissions"
            PolicyDocument:
              Version: '2012-10-17'
              Statement:
                -
                  Effect: Allow
                  Action: ec2:DescribePrefixLists
                  Resource: "*"
          -
            PolicyName: "LambdaPermissions"
            PolicyDocument:
              Version: '2012-10-17'
              Statement:
                -
                  Effect: Allow
                  Action: logs:CreateLogGroup
                  Resource:
                    - !Sub 'arn:aws:logs:${AWS::Region}:${AWS::AccountId}:*'
                -
                  Effect: Allow
                  Action:
                    - logs:CreateLogStream
                    - logs:PutLogEvents
                  Resource:
                    - !Sub 'arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/*'
        ManagedPolicyArns:
          - !Ref AllowBasicS3Access


  ## Lambda PL Resource
  ######################################################

  GetPLResourceFunction:
    Type: AWS::Lambda::Function
    Properties:
      Description: CloudFormation Custom resource for getting endpoint prefix list ID
      Handler: index.handler
      Runtime: python3.9
      Timeout: 300
      Role: !GetAtt GetPLResourceRole.Arn
      Code:
        ZipFile: |
          
          from boto3 import client
          from botocore.exceptions import ClientError
          import os

          import cfnresponse

          def handler(event, context):
            # This comment was added to force an update on this function's code
            try: 
              region = os.environ['AWS_REGION']
              prefix_list_name = event['ResourceProperties']['PrefixListName']

            except ClientError as error:
                cfnresponse.send(event, context, cfnresponse.FAILED, {"code": error.response['Error']['Code']})
                return

            physical_resource_id = 'RetrievedPrefixList'
            response_data = get_pl_id(prefix_list_name, region)
            cfnresponse.send(event, context, cfnresponse.SUCCESS, response_data,physical_resource_id)

          def get_pl_id(pl_name, region):

            try:
                ec2 = client('ec2', region_name=region)
                response = ec2.describe_prefix_lists(
                    Filters=[
                        {
                            'Name': 'prefix-list-name',
                            'Values': [
                                pl_name
                            ]
                        }
                    ]
                )
            except ClientError as error:
                cfnresponse.send(event, context, cfnresponse.FAILED, {"code": error.response['Error']['Code']})
                return

            prefix_list_id = response['PrefixLists'][0]['PrefixListId']
            resp = {'PrefixListID': prefix_list_id}
            return resp      

  S3PrefixListID:
    Type: Custom::GetPL
    Properties:
      ServiceToken: !GetAtt GetPLResourceFunction.Arn
      PrefixListName: !Sub "com.amazonaws.${AWS::Region}.s3"

  ## DynamoDB
  ######################################################################################################################

  MyDynamoDBTable: 
    Type: AWS::DynamoDB::Table
    Properties: 
      AttributeDefinitions: 
        - 
          AttributeName: "pk"
          AttributeType: "S"
        - 
          AttributeName: "sk"
          AttributeType: "S"
        - 
          AttributeName: "GSI1pk"
          AttributeType: "S"
        - 
          AttributeName: "GSI1sk"
          AttributeType: "S"

      KeySchema: 
        - 
          AttributeName: "pk"
          KeyType: "HASH"
        - 
          AttributeName: "sk"
          KeyType: "RANGE"
      
      BillingMode: PAY_PER_REQUEST
      TableName: 
        Ref: DynamoDBTable
      PointInTimeRecoverySpecification:
        PointInTimeRecoveryEnabled: true
      GlobalSecondaryIndexes: 
        - 
          IndexName: "gsi_one"
          KeySchema: 
            - 
              AttributeName: "GSI1pk"
              KeyType: "HASH"
            - 
              AttributeName: "GSI1sk"
              KeyType: "RANGE"
          Projection: 
            ProjectionType: "ALL"

  ################################################
   ### Custom VPC is created ###
  VPC:
    Type: 'AWS::EC2::VPC'
    Properties:
      CidrBlock: !Ref VPCCidr #'10.0.0.0/16'
      EnableDnsSupport: true
      EnableDnsHostnames: true
      InstanceTenancy: default
      Tags:
        - Key: Name
          Value: DemoRedshiftVPC 

  ### InternetGateway is created ###
  InternetGateway:
    Type: 'AWS::EC2::InternetGateway'

  ### InternetGateway is attached to the custom VPC ###
  VPCGatewayAttachment:
    Type: 'AWS::EC2::VPCGatewayAttachment'
    Properties:
      VpcId: !Ref VPC
      InternetGatewayId: !Ref InternetGateway

  ### A Public Subnet is created in the custom VPC ###
  PublicSubnetOne:
    Type: 'AWS::EC2::Subnet'
    Properties:
      AvailabilityZone: !Select [0, !GetAZs '']
      CidrBlock: !Ref PublicSubnetOneCidr  #'10.0.0.1/20'
      MapPublicIpOnLaunch: false
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: DemoRedshiftOneSubnet

  ### Second Subnet (where the Redshift cluster will be provisioned) is created in the custom VPC ###
  PublicSubnetTwo:
    Type: 'AWS::EC2::Subnet'
    Properties:
      AvailabilityZone: !Select [1, !GetAZs '']
      CidrBlock: !Ref PublicSubnetTwoCidr  #'10.0.16.1/20'
      MapPublicIpOnLaunch: false
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: DemoRedshiftSubnetTwo

  ### A Route table is created in the custom VPC ###
  RouteTablePublicSubnet:
    Type: 'AWS::EC2::RouteTable'
    Properties:
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: DemoRedshiftPublicRouteTable
  
  ### Another Route table is created in the custom VPC ###
  RouteTablePrivateSubnet:
    Type: 'AWS::EC2::RouteTable'
    Properties:
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: DemoRedshiftPrivateRouteTable

  ### A Route table is associated with the Public Subnet ###
  RouteTableAssociationPublicSubnet:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      SubnetId: !Ref PublicSubnetOne
      RouteTableId: !Ref RouteTablePublicSubnet
  
  ### A Route table is associated with the Private Subnet ###
  RouteTableAssociationPrivateSubnet:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      SubnetId: !Ref PublicSubnetTwo
      RouteTableId: !Ref RouteTablePrivateSubnet

  RouteTablePublicSubnetInternetRoute:
    Type: 'AWS::EC2::Route'
    DependsOn: VPCGatewayAttachment
    Properties:
      RouteTableId: !Ref RouteTablePublicSubnet
      DestinationCidrBlock: '0.0.0.0/0'
      GatewayId: !Ref InternetGateway

  ### NACL is created for the custom VPC ###
  NetworkAclPublic:
    Type: 'AWS::EC2::NetworkAcl'
    Properties:
      VpcId: !Ref VPC

  ### Associate NACL to Public and Private Subnets ###
  SubnetNetworkAclAssociationPublicSubnet:
    Type: 'AWS::EC2::SubnetNetworkAclAssociation'
    Properties:
      SubnetId: !Ref PublicSubnetOne
      NetworkAclId: !Ref NetworkAclPublic
  SubnetNetworkAclAssociationPrivateSubnet:
    Type: 'AWS::EC2::SubnetNetworkAclAssociation'
    Properties:
      SubnetId: !Ref PublicSubnetTwo
      NetworkAclId: !Ref NetworkAclPublic

  ### NACL Inbound Rule ###
  NetworkAclEntryInPublicAllowAll:
    Type: 'AWS::EC2::NetworkAclEntry'
    Properties:
      NetworkAclId: !Ref NetworkAclPublic
      RuleNumber: 99
      Protocol: -1
      RuleAction: allow
      Egress: false
      CidrBlock: '0.0.0.0/0'
  
### NACL Outbound Rule ###
  NetworkAclEntryOutPublicAllowAll:
    Type: 'AWS::EC2::NetworkAclEntry'
    Properties:
      NetworkAclId: !Ref NetworkAclPublic
      RuleNumber: 99
      Protocol: -1
      RuleAction: allow
      Egress: true
      CidrBlock: '0.0.0.0/0'

  ### This Security Group is created to controls access to the Redshift cluster and communication between Glue components ### 
  GlueRedshiftSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupName: GlueSecurityGroup
      GroupDescription: Self Referring Security Group for Glue services to interact with each others
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: GlueRedshiftSecurityGroup
  
  ### This self-referencing Inbound SG rule allows AWS Glue components to communicate with each other. ###  
  SecurityGroupIngress1:
    Type: AWS::EC2::SecurityGroupIngress
    Properties:
      GroupId: !Ref GlueRedshiftSecurityGroup
      IpProtocol: tcp
      SourceSecurityGroupId: !Ref GlueRedshiftSecurityGroup
      FromPort: 0
      ToPort: 65535
  

  ### This self-referencing Outbound SG rule allows AWS Glue components to communicate with each other. ###  
  SecurityGroupEgress1:
    Type: AWS::EC2::SecurityGroupEgress
    Properties:
      GroupId: !Ref GlueRedshiftSecurityGroup
      IpProtocol: tcp
      FromPort: 0
      ToPort: 65535
      DestinationSecurityGroupId: !Ref GlueRedshiftSecurityGroup
  

  ### This Outbound SG rule allows an HTTPS rule for Amazon S3 access when a VPC endpoint is defined. ###
  ### The s3-prefix-list-id is required in the security group rule to  ###
  ### allow traffic from the VPC to the Amazon S3 VPC endpoint.        ### 
  SecurityGroupEgress2:
    Type: AWS::EC2::SecurityGroupEgress
    Properties:
      GroupId: !Ref GlueRedshiftSecurityGroup 
      IpProtocol: tcp
      FromPort: 443
      ToPort: 443
      DestinationPrefixListId: !GetAtt S3PrefixListID.PrefixListID


  ### A S3 VPC endpoint is created so that all the communication to and from the S3 bucket ###
  ### happens through VPC instead of passing through public internet. ###
  S3VPCEndpoint:
    Type: AWS::EC2::VPCEndpoint
    Properties:
      PolicyDocument: {
            "Statement":[{
                "Effect":"Allow",
                "Principal": "*",
                "Action":"*",
                "Resource":"*"
            }]
        }

      RouteTableIds:
        - !Ref RouteTablePublicSubnet
        - !Ref RouteTablePrivateSubnet
      ServiceName: !Sub 'com.amazonaws.${AWS::Region}.s3'
      VpcEndpointType: Gateway
      VpcId: !Ref VPC


  ## Redshift
  #####################################################################################################################

  RedshiftCluster:
    Type: AWS::Redshift::Cluster
    DependsOn: VPCGatewayAttachment
    Properties:
      ClusterType: "multi-node"
      NumberOfNodes: "2"
      NodeType: "ra3.4xlarge"
      DBName:
        Ref: DatabaseName
      MasterUsername:
        Ref: MasterUsername
      MasterUserPassword:
        Ref: MasterUserPassword
      ClusterParameterGroupName:
        Ref: RedshiftClusterParameterGroup
      VpcSecurityGroupIds:
      - Ref: GlueRedshiftSecurityGroup
      ClusterSubnetGroupName:
        Ref: RedshiftClusterSubnetGroup
      PubliclyAccessible: 'true'
      Port:
        Ref: PortNumber
      Encrypted: true
  RedshiftClusterParameterGroup:
    Type: AWS::Redshift::ClusterParameterGroup
    Properties:
      Description: Cluster parameter group
      ParameterGroupFamily: redshift-1.0
      Parameters:
      - ParameterName: enable_user_activity_logging
        ParameterValue: 'true'
  RedshiftClusterSubnetGroup:
    Type: AWS::Redshift::ClusterSubnetGroup
    Properties:
      Description: Cluster subnet group
      SubnetIds:
      - Ref: PublicSubnetOne
      - Ref: PublicSubnetTwo
 

## Glue create 2 databases
########################################################################################################################

  GlueDatalakeDatabase:
    Type: AWS::Glue::Database
    Properties: 
      CatalogId: !Ref AWS::AccountId
      DatabaseInput: 
        Name: "ddb_dwh_datalake"

  GlueRawDatabase:
    Type: AWS::Glue::Database
    Properties: 
      CatalogId: !Ref AWS::AccountId
      DatabaseInput: 
        Name: "ddb_dwh_raw"

  GlueRedshiftDatabase:
    Type: AWS::Glue::Database
    Properties: 
      CatalogId: !Ref AWS::AccountId
      DatabaseInput: 
        Name: "ddb_redshift"

### Glue JDBC Connection ###
  GlueConnection:
    Type: AWS::Glue::Connection
    DependsOn:
      - RedshiftCluster
    Properties:
      CatalogId: !Ref AWS::AccountId
      ConnectionInput:
        ConnectionProperties: {
          "JDBC_CONNECTION_URL": !Join ['', ['jdbc:redshift://', !GetAtt RedshiftCluster.Endpoint.Address, ':', !GetAtt RedshiftCluster.Endpoint.Port, '/', !Ref DatabaseName]],
          "USERNAME": awsuser,
          "PASSWORD": !Ref MasterUserPassword
        }
        ConnectionType: "JDBC"
        Description: Redshift Cluster Connection
        Name: DDB-Redshift-Blog-Connection
        PhysicalConnectionRequirements:
          AvailabilityZone: { "Fn::GetAtt" : [ "PublicSubnetOne", "AvailabilityZone" ] }
          SecurityGroupIdList:
            - !Ref GlueRedshiftSecurityGroup
          SubnetId: !Ref PublicSubnetOne


## Resource outputs
########################################################################################################################

Outputs:
  ClusterEndpoint:
    Description: Cluster endpoint
    Value: !Sub "${RedshiftCluster.Endpoint.Address}:${RedshiftCluster.Endpoint.Port}"
  ClusterName:
    Description: Name of cluster
    Value:
      Ref: RedshiftCluster
  ParameterGroupName:
    Description: Name of parameter group
    Value:
      Ref: RedshiftClusterParameterGroup
  RedshiftClusterSubnetGroupName:
    Description: Name of cluster subnet group
    Value:
      Ref: RedshiftClusterSubnetGroup
  RedshiftClusterSecurityGroupName:
    Description: Name of cluster security groups
    Value:
      Ref: GlueRedshiftSecurityGroup

  DynamoTable:
    Value: !Ref 'MyDynamoDBTable'
    Description: Table name of the newly created DynamoDB table

  RawBucket:
    Value: !Ref 'RawBucket'

  DataLakeBucket:
    Value: !Ref 'DataLakeBucket'

  RawArchiveBucket:
    Value: !Ref 'RawArchiveBucket'

  DatalakeArchiveBucket:
    Value: !Ref 'DatalakeArchiveBucket'

  AthenaBucket:
    Value: !Ref 'AthenaBucket'

  GlueDatalakeDatabase:
    Value: !Ref 'GlueDatalakeDatabase'

  GlueRawDatabase:
    Value: !Ref 'GlueRawDatabase'

  GlueRedshiftDatabase:
    Value: !Ref 'GlueRedshiftDatabase'

  MyGlueServiceRole:
    Value: !Ref 'MyGlueServiceRole'

  MyStepFunctionServiceRole:
    Value: !Ref 'MyStepFunctionServiceRole'

  S3PrefixListID:
    Value: !GetAtt S3PrefixListID.PrefixListID
  
