AWSTemplateFormatVersion: "2010-09-09"
Description: "This is the template for creating a solution to analyze Amazon Connect Cases logs using Amazon Athena as described in the blog"
Parameters:
    ProjectName:
        Type: "String"
        Default: "test123"
        Description: Enter an all-lowercase single-word name which is between 3 and 42 chars long. An S3 bucket will be created with this name to store all collected data.  
        MinLength : 3
        MaxLength : 42
        AllowedPattern : "^[a-z0-9]*$"
    CreateANewGlueDatabaseAndS3BucketForTheProject:
        Type: "String"
        Default: "Use existing AWS Glue database and Amazon S3 data collection bucket created for this project"
        AllowedValues:
          - "Create a new AWS Glue database and Amazon S3 data collection bucket for this project"
          - "Use existing AWS Glue database and Amazon S3 data collection bucket created for this project"
        Description: Choose whether to create a AWS Glue database and an Amazon S3 bucket for this project. If there is an existing bucket named '{ProjectName}' and a database named '{ProjectName}', then choose to reuse them. 

Conditions:
    CreateCommonGlueDatabaseAndDestinationBucket: !Equals 
        - !Ref CreateANewGlueDatabaseAndS3BucketForTheProject
        - Create a new AWS Glue database and Amazon S3 data collection bucket for this project
        
Resources:

    DataDestinationBucket:
        Type: "AWS::S3::Bucket"
        Condition: CreateCommonGlueDatabaseAndDestinationBucket
        DeletionPolicy: Retain
        Properties:
            BucketName: !Sub "${ProjectName}"

    GlueDatabase:
        Type: "AWS::Glue::Database"
        Condition: CreateCommonGlueDatabaseAndDestinationBucket
        DependsOn: DataDestinationBucket
        DeletionPolicy: Retain
        Properties:
            DatabaseInput:
                Name: !Sub "${ProjectName}"
            CatalogId: !Ref "AWS::AccountId"

    DbAndBucketCreationWaitHandle: 
        Condition: CreateCommonGlueDatabaseAndDestinationBucket
        DependsOn: GlueDatabase 
        Type: "AWS::CloudFormation::WaitConditionHandle"

    WaitHandle: 
        Type: "AWS::CloudFormation::WaitConditionHandle"

    DbAndBucketWaitCondition: 
        Type: "AWS::CloudFormation::WaitCondition"
        Properties: 
            Handle: !If [CreateCommonGlueDatabaseAndDestinationBucket, !Ref DbAndBucketCreationWaitHandle, !Ref WaitHandle]
            Timeout: "1"
            Count: 0

    EventBridgePolicyForCases:
        Type: "AWS::IAM::ManagedPolicy"
        DependsOn: DbAndBucketWaitCondition
        Properties:
            ManagedPolicyName: !Sub "${ProjectName}-EB-Cases"
            Path: "/service-role/"
            PolicyDocument: !Sub |
                {
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Effect": "Allow",
                            "Action": [
                                "firehose:PutRecord",
                                "firehose:PutRecordBatch"
                            ],
                            "Resource": [
                                "${KinesisFirehoseDeliveryStreamForCases.Arn}",
                                "${KinesisFirehoseDeliveryStreamForCaseRelatedItem.Arn}"
                            ]
                        }
                    ]
                }


    EventBridgeRoleForCases:
        Type: "AWS::IAM::Role"
        DependsOn: EventBridgePolicyForCases
        Properties:
            Path: "/service-role/"
            RoleName: !Sub "${ProjectName}-EB-Cases"
            AssumeRolePolicyDocument: "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Service\":\"events.amazonaws.com\"},\"Action\":\"sts:AssumeRole\"}]}"
            MaxSessionDuration: 3600
            ManagedPolicyArns:
              - !Ref EventBridgePolicyForCases



    EventsRuleForCasesEvent:
        Type: "AWS::Events::Rule"
        Properties:
            Name: !Sub "${ProjectName}_CasesEvent"
            EventPattern: !Sub |
                {
                  "source": ["aws.cases"],
                  "detail": {
                    "case": {
                      "templateId": [{
                        "exists": true
                      }]
                    }
                  }
                }                
            State: "ENABLED"
            Targets:
              -
                Arn: !GetAtt KinesisFirehoseDeliveryStreamForCases.Arn
                Id: "Id4a10c2a7-48a3-418f-a525-ecb1272e51c4"
                RoleArn: !GetAtt EventBridgeRoleForCases.Arn
            EventBusName: "default"

    EventsRuleForCaseRelatedItemEvent:
        Type: "AWS::Events::Rule"
        Properties:
            Name: !Sub "${ProjectName}_CaseRelatedItemEvent"
            EventPattern: "{\"source\":[\"aws.cases\"],\"detail\":{\"relatedItem\":[{\"exists\":true}]}}"             
            EventPattern: !Sub |
                {
                  "source": ["aws.cases"],
                  "detail": {
                    "relatedItem": {
                      "relatedItemId": [{
                        "exists": true
                      }]
                    }
                  }
                }   
            State: "ENABLED"
            Targets:
              -
                Arn: !GetAtt KinesisFirehoseDeliveryStreamForCaseRelatedItem.Arn
                Id: "Id4a10c2a7-48a3-418f-a525-ecb1272e51c4"
                RoleArn: !GetAtt EventBridgeRoleForCases.Arn
            EventBusName: "default"


    FirehoseTransformationLambdaForCasesBasicPolicy:
        Type: "AWS::IAM::ManagedPolicy"
        DependsOn: DbAndBucketWaitCondition
        Properties:
            ManagedPolicyName: !Sub "${ProjectName}-Lambda-Cases"
            Path: "/"
            PolicyDocument: !Sub |
                {
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Action": [
                                "logs:CreateLogStream",
                                "logs:CreateLogGroup",
                                "logs:PutLogEvents"
                            ],
                            "Resource": "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/*",
                            "Effect": "Allow",
                            "Sid": "WriteCloudWatchLogs"
                        }
                    ]
                }

    FirehoseTransformationLambdaRoleForCases:
        Type: "AWS::IAM::Role"
        Properties:
            Path: "/service-role/"
            RoleName: !Sub "${ProjectName}-Lambda-Cases"
            AssumeRolePolicyDocument: !Sub |
                {
                    "Version":"2012-10-17",
                    "Statement":[
                        {
                            "Effect":"Allow",
                            "Principal":{
                                "Service":"lambda.amazonaws.com"
                                },
                            "Action":"sts:AssumeRole"
                        }
                    ]
                }
            MaxSessionDuration: 3600
            ManagedPolicyArns:
              - !Ref FirehoseTransformationLambdaForCasesBasicPolicy

    TransformationLambdaFunctionForCases:
        Type: "AWS::Lambda::Function"
        Properties:
            Description: "Kinesis firehose uses this processor for custom transformation logic."
            FunctionName: !Sub "${ProjectName}-TransformCases"
            Handler: "index.lambda_handler"
            Code:
                ZipFile: |
                    import base64
                    import json
                    import boto3
                    import urllib.parse

                    print('Loading function')

                    def transformRecord(recordToTransform):
                        ## This method receives the record to transform and returns the transformed record. 
                        transformedRecord = {}
                        print("Cases log before transformation: " + json.dumps(recordToTransform, indent=2))
                        eventDetails = recordToTransform['detail']
                        transformedRecord['eventType'] = eventDetails['eventType']
                        transformedRecord['approximateChangeTime'] = eventDetails['approximateChangeTime']
                     
                        # standard fields in the case event
                        transformedRecord['account'] = recordToTransform['account']
                        transformedRecord['eventid'] = recordToTransform['id']
                        #transformedRecord['casedomainarn'] = eventDetails['case']['caseId']
                        #transformedRecord['caseId'] = eventDetails['case']['caseId']

                        if eventDetails['eventType'].startswith("CASE"):
                            caseDetails = eventDetails['case']
                            transformedRecord['templateId'] = caseDetails['templateId']
                            transformedRecord['caseId'] = caseDetails['caseId']
                            transformedRecord['createdDateTime'] = caseDetails['createdDateTime']
                            
                            fieldDetails = caseDetails['fields']
                            
                            if ('last_updated_datetime' in fieldDetails):
                                transformedRecord['last_updated_datetime'] = fieldDetails['last_updated_datetime']["value"]["stringValue"]
                            if ('last_closed_datetime' in fieldDetails):
                                transformedRecord['last_closed_datetime'] = fieldDetails['last_closed_datetime']["value"]["stringValue"]
                            if ('title' in fieldDetails):
                                transformedRecord['title'] = fieldDetails['title']['value']['stringValue']
                            if ('customer_id' in fieldDetails):
                                transformedRecord['customer_id'] = fieldDetails['customer_id']['value']['stringValue']
                            if ('status' in fieldDetails):
                                transformedRecord['status'] = fieldDetails['status']['value']['stringValue']        
                            if ('summary' in fieldDetails):
                                transformedRecord['summary'] = fieldDetails['summary']['value']['stringValue']
                            if ('case_reason' in fieldDetails):
                                transformedRecord['case_reason'] = fieldDetails['case_reason']['value']['stringValue']
                            if ('reference_number' in fieldDetails):
                                transformedRecord['reference_number'] = fieldDetails['reference_number']['value']['stringValue']
                                
                            # custom fields in the case event
                            #transformedRecords = [transformRecord(eachRecord) for eachRecord in eachKFHRecordData['Records']]

                            print("Cases log after transformation: " + json.dumps(transformedRecord, indent=2))    
                            return json.dumps(transformedRecord)
                            
                        if  eventDetails['eventType'].startswith("RELATED_ITEM"):
                            relatedItemDetails = eventDetails['relatedItem']
                            # standard fields in the related item event
                            transformedRecord['caseId'] = relatedItemDetails['caseId']
                            transformedRecord['relatedItemId'] = relatedItemDetails['relatedItemId']
                            transformedRecord['createdDateTime'] = relatedItemDetails['createdDateTime']
                            transformedRecord['relatedItemType'] = relatedItemDetails['relatedItemType']

                            if ('comment' in relatedItemDetails):
                                transformedRecord['comment'] = relatedItemDetails['comment']['body']
                            if ('contact' in relatedItemDetails):
                                transformedRecord['contactArn'] = relatedItemDetails['contact']['contactArn']
                                transformedRecord['contactChannel'] = relatedItemDetails['contact']['channel']
                                transformedRecord['contactConnectedToSystemTime'] = relatedItemDetails['contact']['connectedToSystemTime']
                            
                            print("Cases-related item log after transformation: " + json.dumps(transformedRecord, indent=2))    
                            return json.dumps(transformedRecord)
                            
                        raise Exception ('Unknown eventType' + eventDetails['eventType'])

                    def processKFHRecord(eachKFHRecord):
                        ## Each Kinesis Firehose record has the incoming record in 'data' field and an id that needs to be returned with transformed response.
                        recordId = eachKFHRecord['recordId']
                        eachKFHRecordData = json.loads(base64.b64decode(eachKFHRecord['data']))
                        #print("Record from firehose:" + json.dumps(eachKFHRecordData, indent=2))
                        if not ('detail' in eachKFHRecordData):
                            ## Drop the test events  - there is no data in these records
                            return {'result': 'Dropped','recordId': recordId}
                        else:
                            try:
                                ## For each valid record, get the data, transform the data and return it with 'Ok' result.  
                                transformedRecord = transformRecord(eachKFHRecordData)
                                encodedData = base64.b64encode(''.join(transformedRecord).encode("utf-8"))
                                return {'data': encodedData,'result': 'Ok','recordId': recordId}
                            except Exception as e:
                                print ('Unknown Exception')
                                print (e)
                            return {'result': 'Dropped','recordId': recordId}

                    def lambda_handler(event, context):
                        #print (json.dumps(event, indent=2))
                        output = [processKFHRecord(eachKFHRecord) for eachKFHRecord in event['records']]
                        print("{} record(s) processed successfully".format(len(output)))
                        return {'records': output}


            MemorySize: 128
            Role: !GetAtt FirehoseTransformationLambdaRoleForCases.Arn
            Runtime: "python3.9"
            Timeout: 60
            TracingConfig:
                Mode: "PassThrough"

    KinesisFirehoseServicePolicyForCases:
        Type: "AWS::IAM::ManagedPolicy"
        DependsOn: GlueTableForCases
        Properties:
            ManagedPolicyName: !Sub "${ProjectName}-KFH-Cases"
            Path: "/service-role/"
            PolicyDocument: !Sub |
                {
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Sid": "",
                            "Effect": "Allow",
                            "Action": [
                                "glue:GetTable",
                                "glue:GetTableVersion",
                                "glue:GetTableVersions"
                            ],
                            "Resource": [
                                "arn:aws:glue:${AWS::Region}:${AWS::AccountId}:catalog",
                                "arn:aws:glue:${AWS::Region}:${AWS::AccountId}:database/${ProjectName}",
                                "arn:aws:glue:${AWS::Region}:${AWS::AccountId}:table/${ProjectName}/connect_case",
                                "arn:aws:glue:${AWS::Region}:${AWS::AccountId}:table/${ProjectName}/connect_case_relateditem"
                            ]
                        },
                        {
                            "Sid": "",
                            "Effect": "Allow",
                            "Action": [
                                "s3:AbortMultipartUpload",
                                "s3:GetBucketLocation",
                                "s3:GetObject",
                                "s3:ListBucket",
                                "s3:ListBucketMultipartUploads",
                                "s3:PutObject"
                            ],
                            "Resource": [
                                "arn:aws:s3:::${ProjectName}",
                                "arn:aws:s3:::${ProjectName}/*"
                            ]
                        },
                        {
                            "Sid": "",
                            "Effect": "Allow",
                            "Action": [
                                "lambda:InvokeFunction",
                                "lambda:GetFunctionConfiguration"
                            ],
                            "Resource": "arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:${TransformationLambdaFunctionForCases}"
                        },
                        {
                            "Sid": "",
                            "Effect": "Allow",
                            "Action": [
                                "logs:PutLogEvents"
                            ],
                            "Resource": [   
                                "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/kinesisfirehose/${ProjectName}-Cases:log-stream:*",
                                "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/kinesisfirehose/${ProjectName}-CaseRelatedItem:log-stream:*"
                            ]
                        }
                    ]
                }

    KinesisFirehoseServiceRoleForCases:
        Type: "AWS::IAM::Role"
        Properties:
            Path: "/service-role/"
            RoleName: !Sub "${ProjectName}-KFH-Cases"
            AssumeRolePolicyDocument: !Sub |
                {
                    "Version":"2012-10-17",
                    "Statement":[
                        {
                            "Effect":"Allow",
                            "Principal":{"Service":"firehose.amazonaws.com"},
                            "Action":"sts:AssumeRole"
                        }
                    ]
                }
            MaxSessionDuration: 3600
            ManagedPolicyArns:
              - !Ref KinesisFirehoseServicePolicyForCases


    GlueTableForCases:
        Type: "AWS::Glue::Table"
        DependsOn: "DbAndBucketWaitCondition"
        Properties:
            DatabaseName: !Sub "${ProjectName}"
            CatalogId: !Ref "AWS::AccountId"
            TableInput:
                Owner: "hadoop"
                TableType: "EXTERNAL_TABLE"
                Parameters:
                    EXTERNAL: "TRUE"
                    "parquet.compression": "SNAPPY"
                    "projection.year.type": "integer"
                    "projection.year.range": "2022,2024"
                    "projection.month.type": "integer"
                    "projection.month.range": "01,12"
                    "projection.month.digits": "2"
                    "projection.day.type": "integer"
                    "projection.day.range": "01,31"
                    "projection.day.digits": "2"
                    "projection.enabled": "true"
                    transient_lastDdlTime: "1628198798"
                StorageDescriptor:
                    Columns:
                      -
                        Name: "account"
                        Type: "string"
                      -
                        Name: "eventid"
                        Type: "string"
                      -
                        Name: "eventtype"
                        Type: "string"
                      -
                        Name: "casedomainarn"
                        Type: "string"
                      -
                        Name: "casearn"
                        Type: "string"
                      -
                        Name: "approximatechangetime"
                        Type: "string"
                      -
                        Name: "caseId"
                        Type: "string"
                      -
                        Name: "templateId"
                        Type: "string"
                      -
                        Name: "createddatetime"
                        Type: "string"
                      -
                        Name: "last_updated_datetime"
                        Type: "string"
                      -
                        Name: "last_closed_datetime"
                        Type: "string"
                      -
                        Name: "status"
                        Type: "string"
                      -
                        Name: "case_reason"
                        Type: "string"
                      -
                        Name: "customer_id"
                        Type: "string"
                      -
                        Name: "title"
                        Type: "string"
                      -
                        Name: "reference_number"
                        Type: "string"
                      -
                        Name: "summary"
                        Type: "string"
                      -
                        Name: "customfields"
                        Type: "array<struct<fieldname:string,fieldvalue:string,fieldnumericvalue:double,fieldstringvalue:string>>"
                    Location: !Sub "s3://${ProjectName}/case-base"
                    InputFormat: "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat"
                    OutputFormat: "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat"
                    Compressed: false
                    NumberOfBuckets: -1
                    SerdeInfo:
                        SerializationLibrary: "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe"
                        Parameters:
                            "serialization.format": "1"
                    Parameters: {}
                    SkewedInfo:
                        SkewedColumnValueLocationMaps: {}
                    StoredAsSubDirectories: false
                PartitionKeys:
                  -
                    Name: "year"
                    Type: "int"
                  -
                    Name: "month"
                    Type: "int"
                  -
                    Name: "day"
                    Type: "int"
                Retention: 0
                Name: "connect_case"



    KinesisFirehoseDeliveryStreamForCases:
        Type: "AWS::KinesisFirehose::DeliveryStream"
        DependsOn: 
            - GlueTableForCases
            - TransformationLambdaFunctionForCases
        Properties:
            DeliveryStreamName: !Sub "${ProjectName}-Cases"
            DeliveryStreamType: "DirectPut"
            ExtendedS3DestinationConfiguration:
                BucketARN: !Sub "arn:aws:s3:::${ProjectName}"
                BufferingHints:
                    SizeInMBs: 128
                    IntervalInSeconds: 60
                CloudWatchLoggingOptions:
                    Enabled: true
                    LogGroupName: !Sub "/aws/kinesisfirehose/${ProjectName}-Cases"
                    LogStreamName: "DestinationDelivery"
                CompressionFormat: "UNCOMPRESSED"
                DataFormatConversionConfiguration:
                    SchemaConfiguration:
                        RoleARN: !GetAtt KinesisFirehoseServiceRoleForCases.Arn
                        DatabaseName: !Sub "${ProjectName}"
                        TableName: "connect_case"
                        Region: !Ref AWS::Region
                        VersionId: "LATEST"
                    InputFormatConfiguration:
                        Deserializer:
                            OpenXJsonSerDe: {}
                    OutputFormatConfiguration:
                        Serializer:
                            ParquetSerDe: {}
                    Enabled: true
                EncryptionConfiguration:
                    NoEncryptionConfig: "NoEncryption"
                Prefix: "case-base/year=!{timestamp:YYYY}/month=!{timestamp:MM}/day=!{timestamp:dd}/"
                ErrorOutputPrefix: "case-erroroutputbase/!{timestamp:yyy/MM/dd}/!{firehose:random-string}/!{firehose:error-output-type}/!{timestamp:yyy/MM/dd}/"
                RoleARN: !GetAtt KinesisFirehoseServiceRoleForCases.Arn
                ProcessingConfiguration:
                    Enabled: true
                    Processors:
                      -
                        Type: "Lambda"
                        Parameters:
                          -
                            ParameterName: "LambdaArn"
                            ParameterValue: !GetAtt TransformationLambdaFunctionForCases.Arn 
                          -
                            ParameterName: "NumberOfRetries"
                            ParameterValue: "3"
                          -
                            ParameterName: "RoleArn"
                            ParameterValue: !GetAtt KinesisFirehoseServiceRoleForCases.Arn
                          -
                            ParameterName: "BufferSizeInMBs"
                            ParameterValue: "3"
                          -
                            ParameterName: "BufferIntervalInSeconds"
                            ParameterValue: "60"
                S3BackupMode: "Disabled"

    GlueTableForCaseRelatedItem:
        Type: "AWS::Glue::Table"
        DependsOn: "DbAndBucketWaitCondition"
        Properties:
            DatabaseName: !Sub "${ProjectName}"
            CatalogId: !Ref "AWS::AccountId"
            TableInput:
                Owner: "hadoop"
                TableType: "EXTERNAL_TABLE"
                Parameters:
                    EXTERNAL: "TRUE"
                    "parquet.compression": "SNAPPY"
                    "projection.year.type": "integer"
                    "projection.year.range": "2022,2024"
                    "projection.month.type": "integer"
                    "projection.month.range": "01,12"
                    "projection.month.digits": "2"
                    "projection.day.type": "integer"
                    "projection.day.range": "01,31"
                    "projection.day.digits": "2"
                    "projection.enabled": "true"
                    transient_lastDdlTime: "1628198798"
                StorageDescriptor:
                    Columns:
                      -
                        Name: "account"
                        Type: "string"
                      -
                        Name: "eventid"
                        Type: "string"
                      -
                        Name: "eventtype"
                        Type: "string"
                      -
                        Name: "casedomainarn"
                        Type: "string"
                      -
                        Name: "caserelateditemarn"
                        Type: "string"
                      -
                        Name: "approximatechangetime"
                        Type: "string"
                      -
                        Name: "caseid"
                        Type: "string"
                      -
                        Name: "relateditemid"
                        Type: "string"
                      -
                        Name: "relateditemtype"
                        Type: "string"
                      -
                        Name: "comment"
                        Type: "string"
                      -
                        Name: "contactarn"
                        Type: "string"
                      -
                        Name: "contactchannel"
                        Type: "string"
                      -
                        Name: "contactconnectedtosystemtime"
                        Type: "string"
                      -
                        Name: "createddatetime"
                        Type: "string"
                      -
                        Name: "last_updated_datetime"
                        Type: "string"
                    Location: !Sub "s3://${ProjectName}/case-relateditem-base"
                    InputFormat: "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat"
                    OutputFormat: "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat"
                    Compressed: false
                    NumberOfBuckets: -1
                    SerdeInfo:
                        SerializationLibrary: "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe"
                        Parameters:
                            "serialization.format": "1"
                    Parameters: {}
                    SkewedInfo:
                        SkewedColumnValueLocationMaps: {}
                    StoredAsSubDirectories: false
                PartitionKeys:
                  -
                    Name: "year"
                    Type: "int"
                  -
                    Name: "month"
                    Type: "int"
                  -
                    Name: "day"
                    Type: "int"
                Retention: 0
                Name: "connect_case_relateditem"


    KinesisFirehoseDeliveryStreamForCaseRelatedItem:
        Type: "AWS::KinesisFirehose::DeliveryStream"
        DependsOn: 
            - GlueTableForCaseRelatedItem
            - TransformationLambdaFunctionForCases
        Properties:
            DeliveryStreamName: !Sub "${ProjectName}-CaseRelatedItem"
            DeliveryStreamType: "DirectPut"
            ExtendedS3DestinationConfiguration:
                BucketARN: !Sub "arn:aws:s3:::${ProjectName}"
                BufferingHints:
                    SizeInMBs: 128
                    IntervalInSeconds: 60
                CloudWatchLoggingOptions:
                    Enabled: true
                    LogGroupName: !Sub "/aws/kinesisfirehose/${ProjectName}-CaseRelatedItem"
                    LogStreamName: "DestinationDelivery"
                CompressionFormat: "UNCOMPRESSED"
                DataFormatConversionConfiguration:
                    SchemaConfiguration:
                        RoleARN: !GetAtt KinesisFirehoseServiceRoleForCases.Arn
                        DatabaseName: !Sub "${ProjectName}"
                        TableName: "connect_case_relatedItem"
                        Region: !Ref AWS::Region
                        VersionId: "LATEST"
                    InputFormatConfiguration:
                        Deserializer:
                            OpenXJsonSerDe: {}
                    OutputFormatConfiguration:
                        Serializer:
                            ParquetSerDe: {}
                    Enabled: true
                EncryptionConfiguration:
                    NoEncryptionConfig: "NoEncryption"
                Prefix: "case-relateditem-base/year=!{timestamp:YYYY}/month=!{timestamp:MM}/day=!{timestamp:dd}/"
                ErrorOutputPrefix: "case-relateditem-erroroutputbase/!{timestamp:yyy/MM/dd}/!{firehose:random-string}/!{firehose:error-output-type}/!{timestamp:yyy/MM/dd}/"
                RoleARN: !GetAtt KinesisFirehoseServiceRoleForCases.Arn
                ProcessingConfiguration:
                    Enabled: true
                    Processors:
                      -
                        Type: "Lambda"
                        Parameters:
                          -
                            ParameterName: "LambdaArn"
                            ParameterValue: !GetAtt TransformationLambdaFunctionForCases.Arn 
                          -
                            ParameterName: "NumberOfRetries"
                            ParameterValue: "3"
                          -
                            ParameterName: "RoleArn"
                            ParameterValue: !GetAtt KinesisFirehoseServiceRoleForCases.Arn
                          -
                            ParameterName: "BufferSizeInMBs"
                            ParameterValue: "3"
                          -
                            ParameterName: "BufferIntervalInSeconds"
                            ParameterValue: "60"
                S3BackupMode: "Disabled"


    ViewForLatestCases:
        Type: "AWS::Glue::Table"
        DependsOn: 
            - GlueTableForCases
            - GlueTableForCaseRelatedItem
        Properties:
            DatabaseName: !Sub "${ProjectName}"
            CatalogId: !Ref "AWS::AccountId"
            TableInput: 
                ViewOriginalText: !Join
                  - ''
                  - - '/* Presto View: '
                    - Fn::Base64: 
                        Fn::Sub: |
                          {"originalSql":"WITH\n  latest_data AS (\n   SELECT\n     \"max\"(approximatechangetime) latestapproximatechangetime\n   , caseid caseid\n   FROM\n     \"${ProjectName}\".\"connect_case\"\n   GROUP BY caseid\n) \n, countofcontactspercase_data AS (\n   SELECT\n     \"count\"(caseid) countofcontactspercase\n   , caseid caseid\n   FROM\n     \"${ProjectName}\".\"connect_case_relateditem\" cri\n   WHERE ((cri.contactarn IS NOT NULL) AND (contactchannel <> 'TASK'))\n   GROUP BY caseid\n) \n, countoftaskspercase_data AS (\n   SELECT\n     \"count\"(caseid) countoftaskspercase\n   , caseid caseid\n   FROM\n     \"${ProjectName}\".\"connect_case_relateditem\" cri\n   WHERE ((cri.contactarn IS NOT NULL) AND (contactchannel = 'TASK'))\n   GROUP BY caseid\n) \nSELECT\n  CAST(\"from_iso8601_timestamp\"(c.approximatechangetime) AS timestamp) approximatechangetime\n, c.caseid\n, c.templateid\n, CAST(\"from_iso8601_timestamp\"(c.createddatetime) AS timestamp) createddatetime\n, CAST(\"from_iso8601_timestamp\"(c.last_updated_datetime) AS timestamp) last_updated_datetime\n, CAST(\"from_iso8601_timestamp\"(c.last_closed_datetime) AS timestamp) last_closed_datetime\n, c.status\n, (CASE WHEN (c.status = 'open') THEN 1 END) status_open\n, (CASE WHEN (c.status = 'closed') THEN 1 END) status_closed\n, \"date_diff\"('hour', \"from_iso8601_timestamp\"(c.createddatetime), \"from_iso8601_timestamp\"(c.last_closed_datetime)) creation_to_close_diff_hours\n, c.case_reason\n, c.title\n, c.reference_number\n, c.summary\n, ccpc.countofcontactspercase\n, (CASE WHEN (ctpc.countoftaskspercase IS NULL) THEN 0 ELSE ctpc.countoftaskspercase END) countoftaskspercase\nFROM\n  (((\"${ProjectName}\".\"connect_case\" c\nINNER JOIN latest_data ON ((latest_data.caseid = c.caseid) AND (c.approximatechangetime = latest_data.latestapproximatechangetime)))\nLEFT JOIN countofcontactspercase_data ccpc ON (ccpc.caseid = c.caseid))\nLEFT JOIN countoftaskspercase_data ctpc ON (ctpc.caseid = c.caseid))\n","catalog":"awsdatacatalog","schema":"${ProjectName}","columns":[{"name":"approximatechangetime","type":"timestamp"},{"name":"caseid","type":"varchar"},{"name":"templateid","type":"varchar"},{"name":"createddatetime","type":"timestamp"},{"name":"last_updated_datetime","type":"timestamp"},{"name":"last_closed_datetime","type":"timestamp"},{"name":"status","type":"varchar"},{"name":"status_open","type":"integer"},{"name":"status_closed","type":"integer"},{"name":"creation_to_close_diff_hours","type":"bigint"},{"name":"case_reason","type":"varchar"},{"name":"title","type":"varchar"},{"name":"reference_number","type":"varchar"},{"name":"summary","type":"varchar"},{"name":"countofcontactspercase","type":"bigint"},{"name":"countoftaskspercase","type":"bigint"}]}
                    - ' */'
                TableType: "VIRTUAL_VIEW"
                Parameters: 
                    comment: "Presto View"
                    presto_view: "true"
                ViewExpandedText: "/* Presto View */"
                StorageDescriptor: 
                    Columns: 
                      - 
                        Name: "approximatechangetime"
                        Type: "timestamp"
                      - 
                        Name: "caseid"
                        Type: "string"
                      - 
                        Name: "templateid"
                        Type: "string"
                      - 
                        Name: "createddatetime"
                        Type: "timestamp"
                      - 
                        Name: "last_updated_datetime"
                        Type: "timestamp"
                      - 
                        Name: "last_closed_datetime"
                        Type: "timestamp"
                      - 
                        Name: "status"
                        Type: "string"
                      - 
                        Name: "status_open"
                        Type: "int"
                      - 
                        Name: "status_closed"
                        Type: "int"
                      - 
                        Name: "creation_to_close_diff_hours"
                        Type: "bigint"
                      - 
                        Name: "case_reason"
                        Type: "string"
                      - 
                        Name: "title"
                        Type: "string"
                      - 
                        Name: "reference_number"
                        Type: "string"
                      - 
                        Name: "summary"
                        Type: "string"
                      - 
                        Name: "countofcontactspercase"
                        Type: "bigint"
                      - 
                        Name: "countoftaskspercase"
                        Type: "bigint"
                    Location: ""
                    Compressed: false
                    NumberOfBuckets: 0
                    SerdeInfo: {}
                    StoredAsSubDirectories: false
                Retention: 0
                Name: "connect_cases_latest"

    ViewForLatestCaseFields:
        Type: "AWS::Glue::Table"
        DependsOn: 
            - GlueTableForCases
            - GlueTableForCaseRelatedItem
        Properties:
            DatabaseName: !Sub "${ProjectName}"
            CatalogId: !Ref "AWS::AccountId"
            TableInput: 
                ViewOriginalText: !Join
                  - ''
                  - - '/* Presto View: '
                    - Fn::Base64: 
                        Fn::Sub: |
                          {"originalSql":"WITH\n  latest_data AS (\n   SELECT\n     \"max\"(approximatechangetime) latestapproximatechangetime\n   , caseid foreachcaseid\n   FROM\n     \"${ProjectName}\".\"connect_case\"\n   GROUP BY caseid\n) \nSELECT\n  c.account\n, c.eventid\n, c.eventtype\n, c.casedomainarn\n, c.casearn\n, CAST(\"from_iso8601_timestamp\"(c.approximatechangetime) AS timestamp) approximatechangetime\n, c.caseid\n, c.templateid\n, CAST(\"from_iso8601_timestamp\"(c.createddatetime) AS timestamp) createddatetime\n, CAST(\"from_iso8601_timestamp\"(c.last_updated_datetime) AS timestamp) last_updated_datetime\n, CAST(\"from_iso8601_timestamp\"(c.last_closed_datetime) AS timestamp) last_closed_datetime\n, c.status\n, c.case_reason\n, c.customer_id\n, c.title\n, c.reference_number\n, c.summary\n, c.year\n, c.month\n, c.day\n, customfieldset.fieldname\n, customfieldset.fieldvalue\nFROM\n  \"${ProjectName}\".\"connect_case\" c\n, UNNEST(c.customfields) t (customfieldset)\n, latest_data\nWHERE ((latest_data.foreachcaseid = c.caseid) AND (c.approximatechangetime = latest_data.latestapproximatechangetime))\n","catalog":"awsdatacatalog","schema":"${ProjectName}","columns":[{"name":"account","type":"varchar"},{"name":"eventid","type":"varchar"},{"name":"eventtype","type":"varchar"},{"name":"casedomainarn","type":"varchar"},{"name":"casearn","type":"varchar"},{"name":"approximatechangetime","type":"timestamp"},{"name":"caseid","type":"varchar"},{"name":"templateid","type":"varchar"},{"name":"createddatetime","type":"timestamp"},{"name":"last_updated_datetime","type":"timestamp"},{"name":"last_closed_datetime","type":"timestamp"},{"name":"status","type":"varchar"},{"name":"case_reason","type":"varchar"},{"name":"customer_id","type":"varchar"},{"name":"title","type":"varchar"},{"name":"reference_number","type":"varchar"},{"name":"summary","type":"varchar"},{"name":"year","type":"integer"},{"name":"month","type":"integer"},{"name":"day","type":"integer"},{"name":"fieldname","type":"varchar"},{"name":"fieldvalue","type":"varchar"}]}
                    - ' */'
                TableType: "VIRTUAL_VIEW"
                Parameters: 
                    comment: "Presto View"
                    presto_view: "true"
                ViewExpandedText: "/* Presto View */"
                StorageDescriptor: 
                    Columns: 
                      - 
                        Name: "account"
                        Type: "string"
                      - 
                        Name: "eventid"
                        Type: "string"
                      - 
                        Name: "eventtype"
                        Type: "string"
                      - 
                        Name: "casedomainarn"
                        Type: "string"
                      - 
                        Name: "casearn"
                        Type: "string"
                      - 
                        Name: "approximatechangetime"
                        Type: "timestamp"
                      - 
                        Name: "caseid"
                        Type: "string"
                      - 
                        Name: "templateid"
                        Type: "string"
                      - 
                        Name: "createddatetime"
                        Type: "timestamp"
                      - 
                        Name: "last_updated_datetime"
                        Type: "timestamp"
                      - 
                        Name: "last_closed_datetime"
                        Type: "timestamp"
                      - 
                        Name: "status"
                        Type: "string"
                      - 
                        Name: "case_reason"
                        Type: "string"
                      - 
                        Name: "customer_id"
                        Type: "string"
                      - 
                        Name: "title"
                        Type: "string"
                      - 
                        Name: "reference_number"
                        Type: "string"
                      - 
                        Name: "summary"
                        Type: "string"
                      - 
                        Name: "year"
                        Type: "int"
                      - 
                        Name: "month"
                        Type: "int"
                      - 
                        Name: "day"
                        Type: "int"
                      - 
                        Name: "fieldname"
                        Type: "string"
                      - 
                        Name: "fieldvalue"
                        Type: "string"
                    Location: ""
                    Compressed: false
                    NumberOfBuckets: 0
                    SerdeInfo: {}
                    StoredAsSubDirectories: false
                Retention: 0
                Name: "connect_casefields_latest"

