Description:  The template will create the resources needed for the CTR Pipeline

Parameters:       
    Prefix:
        Description: Prefix for resources
        Type: String
        
    CTRBucketName:
        Description: Required to avoid circular dependencies with S3 notifications
        Type: String
        
    LambdaDeployS3Bucket:
        Description: Bucket where the Lambda Zip file is located
        Type: String
        
    FirehoseAddNewLineLambdaKey:
        Description: Key where the Lambda Zip file is located
        Type: String
        
    ModifyCtrLambdaKey:
        Description: Key where the Lambda Zip file is located
        Type: String
        
    CTRModifiedS3Folder:
        Description: Where lambda will put the modified records
        Type: String

Resources:
    CTRBucket:
        Type: AWS::S3::Bucket
        DeletionPolicy: Retain
        Properties:
            BucketName: !Ref CTRBucketName
            NotificationConfiguration:
                LambdaConfigurations:
                -   Event: s3:ObjectCreated:*
                    Filter:
                        S3Key:
                            Rules:
                            -   Name: prefix
                                Value: ctr/
                    Function: !GetAtt ModifyCtrLambda.Arn
                    
    CTRBucketInvokeLambdaPermission:
        Type: AWS::Lambda::Permission
        Properties:
            Action: lambda:InvokeFunction
            FunctionName: !Ref ModifyCtrLambda
            Principal: s3.amazonaws.com
            SourceAccount: !Ref AWS::AccountId
            SourceArn: !Sub arn:aws:s3:::${CTRBucketName}
            
    FirehoseAddNewLineLambda:
        Type: AWS::Lambda::Function
        Properties: 
            Code:
                S3Bucket: !Ref LambdaDeployS3Bucket
                S3Key: !Ref FirehoseAddNewLineLambdaKey
            Environment:
                Variables:
                    LOG_LEVEL: INFO
            FunctionName: !Join ["", [!Ref Prefix, FirehoseAddNewLine]]
            Handler: lambda_function.lambda_handler
            MemorySize: 128
            PackageType: Zip
            Runtime: python3.8
            Timeout: 60
            Role: !GetAtt FirehoseAddNewLineLambdaRole.Arn
            
    ModifyCtrLambda:
        Type: AWS::Lambda::Function
        Properties: 
            Code:
                S3Bucket: !Ref LambdaDeployS3Bucket
                S3Key: !Ref ModifyCtrLambdaKey
            Environment:
                Variables:
                    LOG_LEVEL: INFO
                    CTRModifiedS3Folder: !Ref CTRModifiedS3Folder
            FunctionName: !Join ["", [!Ref Prefix, ModifyCtr]]
            Handler: lambda_function.lambda_handler
            MemorySize: 128
            PackageType: Zip
            Runtime: python3.8
            Timeout: 60
            Role: !GetAtt ModifyCtrLambdaRole.Arn
            
    FirehoseAddNewLineLambdaRole:
        Type: AWS::IAM::Role
        Properties:     
            AssumeRolePolicyDocument:
                Version: 2012-10-17
                Statement:
                -   Effect: Allow
                    Principal:
                        Service: lambda.amazonaws.com
                    Action: sts:AssumeRole
            Path: /
            Policies:
                -   PolicyName: !Join ["", [!Ref Prefix, FirehoseAddNewLinePolicy]]
                    PolicyDocument:
                        Version: 2012-10-17
                        Statement:
                        -   Effect: Allow
                            Action:
                            -   logs:CreateLogGroup
                            -   logs:CreateLogStream
                            -   logs:PutLogEvents
                            Resource: "*"
                            
    ModifyCtrLambdaRole:
        Type: AWS::IAM::Role
        Properties:     
            AssumeRolePolicyDocument:
                Version: 2012-10-17
                Statement:
                -   Effect: Allow
                    Principal:
                        Service: lambda.amazonaws.com
                    Action: sts:AssumeRole
            Path: /
            Policies:
                -   PolicyName: !Join ["", [!Ref Prefix, ModifyCtrLambdaPolicy]]
                    PolicyDocument:
                        Version: 2012-10-17
                        Statement:
                        -   Effect: Allow
                            Action:
                            -   logs:CreateLogGroup
                            -   logs:CreateLogStream
                            -   logs:PutLogEvents
                            Resource: "*"
                        -   Effect: Allow
                            Action:
                            -   s3:GetObject
                            -   s3:ListBucket
                            -   s3:PutObject
                            Resource: 
                            -   !Sub arn:aws:s3:::${CTRBucketName}
                            -   !Sub arn:aws:s3:::${CTRBucketName}/*
                            
    CTRStream:
        Type: AWS::Kinesis::Stream
        Properties:
            Name: !Join ["", [!Ref Prefix, Ctr]]
            RetentionPeriodHours: 24
            ShardCount: 1

    CTRDelivery:
        Type: AWS::KinesisFirehose::DeliveryStream
        Properties:
            DeliveryStreamName: !Join ["", [!Ref Prefix, Ctr, S3]]
            DeliveryStreamType: KinesisStreamAsSource
            KinesisStreamSourceConfiguration: 
                KinesisStreamARN: !GetAtt CTRStream.Arn
                RoleARN: !GetAtt CTRDeliveryRole.Arn
            ExtendedS3DestinationConfiguration:
                BucketARN: !GetAtt CTRBucket.Arn
                Prefix: ctr/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/
                ErrorOutputPrefix: ctrError/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/!{firehose:error-output-type}
                CompressionFormat: UNCOMPRESSED
                RoleARN: !GetAtt CTRDeliveryRole.Arn
                ProcessingConfiguration:
                    Enabled: true
                    Processors:
                    -   Parameters:
                        -   ParameterName: LambdaArn
                            ParameterValue: !GetAtt FirehoseAddNewLineLambda.Arn
                        Type: Lambda 
                BufferingHints:
                    IntervalInSeconds: 60
                    SizeInMBs: 1
                S3BackupMode: Enabled
                S3BackupConfiguration:
                    BucketARN: !GetAtt CTRBucket.Arn
                    Prefix: ctrBackup/
                    BufferingHints:
                        IntervalInSeconds: 60
                        SizeInMBs: 1
                    CompressionFormat: UNCOMPRESSED
                    RoleARN: !GetAtt CTRDeliveryRole.Arn
                CloudWatchLoggingOptions:
                    Enabled: true
                    LogGroupName: !Ref CTRDeliveryLogGroup
                    LogStreamName: !Ref CTRDeliveryLogStream

    CTRDeliveryLogGroup:
        Type: AWS::Logs::LogGroup
        Properties: 
            LogGroupName: !Join ["", [/aws/kinesis/firehose/, !Ref Prefix, CtrS3]]

    CTRDeliveryLogStream:
        Type: AWS::Logs::LogStream
        Properties: 
            LogGroupName: !Ref CTRDeliveryLogGroup
            LogStreamName: S3Delivery

    CTRDeliveryRole:
        Type: AWS::IAM::Role
        Properties:
            AssumeRolePolicyDocument:
                Version: 2012-10-17
                Statement:
                -   Effect: Allow
                    Principal:
                        Service: firehose.amazonaws.com
                    Action: sts:AssumeRole
                    Condition:
                        StringEquals:
                            sts:ExternalId: !Ref AWS::AccountId
            Path: /
            Policies:
                -   PolicyName: !Join ["", [KinesisFirehoseServicePolicy, !Ref Prefix, CtrS3]]
                    PolicyDocument:
                        Version: 2012-10-17
                        Statement:
                        -   Effect: Allow
                            Action:
                            -   s3:AbortMultipartUpload
                            -   s3:GetBucketLocation
                            -   s3:GetObject
                            -   s3:ListBucket
                            -   s3:ListBucketMultipartUploads
                            -   s3:PutObject
                            Resource:
                            -   !GetAtt CTRBucket.Arn
                            -   !Join ["", [!GetAtt CTRBucket.Arn, "/*"]]
                        -   Effect: Allow
                            Action:
                            -   lambda:InvokeFunction
                            -   lambda:GetFunctionConfiguration
                            Resource: !GetAtt FirehoseAddNewLineLambda.Arn
                        -   Effect: Allow
                            Action:
                            -   logs:PutLogEvents
                            Resource: 
                            -   !Join ["", [!GetAtt CTRDeliveryLogGroup.Arn, ":*"]]
                        -   Effect: Allow
                            Action:
                            -   kinesis:DescribeStream
                            -   kinesis:GetShardIterator
                            -   kinesis:GetRecords
                            -   kinesis:ListShards
                            Resource: !GetAtt CTRStream.Arn

Outputs:
    CTRBucket:
        Value: !Ref CTRBucket
        
    FirehoseAddNewLineLambda:
        Value: !Ref FirehoseAddNewLineLambda
        
    ModifyCtrLambda:
        Value: !Ref ModifyCtrLambda
        
    CTRStream:
        Value: !Ref CTRStream

    CTRDelivery:
        Value: !Ref CTRDelivery
        
    CTRDeliveryLogStream:
        Value: !Ref CTRDeliveryLogStream