AWSTemplateFormatVersion: '2010-09-09'

Description: >
    "Deployable solution to enable logging, capture, and storage of diagnostic information going forward to support reporting and billing questions.
    Components Deployed: Kinesis Stream for Amazon Connect Agent Events, Kinesis Firehose for Agent Events to S3, Kinesis Firehose for Amazon Connect CTRs, An Amazon Connect Lambda to be included in Whisper Flows, an S3 triggered Lambda Function for transformation of diagnostic and log data.
    * Manual setup steps required to connect the infrastructure deployed with Amazon Connect Include: *
    - Select or create an S3 bucket for logs to be saved to.  Note that the default Amazon S3 bucket found in the Administation Section can be used.
    - Enable Data Streams in Amazon Connect Admin Console and select the Kinesis Streams / Firehose.
    - Add the Lambda Function to Amazon Connect's enabled functions for Contact Flows
    - Run the Athena Saved Queries generated by the CFN Template
    - v0.0.4"

Mappings:
    FunctionMap:
        Configuration:
            S3Bucket: "nicklo-cfn-lambda-deployments"
            S3Key: "connect-diagnostics/v0.0.4/connect-diagnostics.yaml"

Parameters:
  instanceIdParam:
    Type: String
    AllowedPattern: '\w{8}-\w{4}-\w{4}-\w{4}-\w{12}'
    ConstraintDescription: "Invalid Amazon Connect instance Id"
    Description: Amazon Connect Instance ID (Ensure you it is entered accurately in the format xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx).
  instanceNameParam:
    Type: String
    ConstraintDescription: "Invalid Amazon Connect instance alias"
    Description: Amazon Connect Instance Alias (Ensure you it is entered accurately as configured under Amazon Connect Service).
  storageDestinationS3:
    Type: String
    AllowedPattern: '(?!^(\d+\.)+\d+$)^(^([a-z0-9]|[a-z0-9][a-z0-9\-]*[a-z0-9])\.)*([a-z0-9]|[a-z0-9][a-z0-9\-]*[a-z0-9])([a-z0-9\-]+((\/))?)*([a-z0-9])+$'
    ConstraintDescription: "Invalid storage destination"
    Description: Storage bucket for saving events, contact trace records, and logs. Required.
  athenaDbName:
    Type: String
    AllowedPattern: '([a-z]?)([a-z|-])*([^-])$'
    ConstraintDescription: "Invalid database name, must be lowercase"
    Description: Database Name for use in Athena, must be unique in your instance.

Metadata:
  'AWS::CloudFormation::Interface':
    ParameterGroups:
      - Label:
          default: Amazon Connect Configuration
        Parameters:
          - instanceNameParam
          - instanceIdParam
      - Label:
          default: Storage Configuration
        Parameters:
          - storageDestinationS3
      - Label:
          default: Athena Configuration
        Parameters:
          - athenaDbName          
    ParameterLabels:
      instanceIdParam:
        default: Instance ID
      instanceNameParam:
        default: Instance Alias
      storageDestinationS3:
        default: S3 Bucket

Outputs:
    agentEventsKinesisStream:
        Description:  The ARN of the Kinesis Stream that will receive agent events.  Manual configuration required in Amazon Connect admin -> Data Streaming section.
        Value: !Ref AgentEventKinesisStream
    CTRKinesisFirehose:
        Description:  The ARN of the Kinesis Firehose that will receive CTR events.  Manual configuration required in the Amazon Connect admin -> Data Streaming section.Topic.
        Value: !Ref FirehoseCTRStream
    describeUserWhisperLambda:
        Description: The Lambda function that will log UserDetails when added to your Amazon Connect Contact Flows.  Include this function in the Whisper flow with the following parameters included for functionality - 'AgentARN' = 'Type:Agent' && 'Attribute:ARN' -- 'InstanceARN' = 'Type:System' && 'Attribute:Instance ARN'
        Value: !Ref describeUserWhisperLambda
    ContactTraceRecordsAthenaNamedQuery:
        Description: Creates Athena table for CTRs for analysis.
        Value: !Join
        - ''
        - - !Sub 'https://console.aws.amazon.com/athena/home?force&region=${AWS::Region}'
          - '#query/saved/'
          - !Ref ContactTraceRecordsAthenaNamedQuery
    APILogsAthenaNamedQuery:
        Description: Creates Athena table for API Logs for analysis.
        Value: !Join
        - ''
        - - !Sub 'https://console.aws.amazon.com/athena/home?force&region=${AWS::Region}'
          - '#query/saved/'
          - !Ref APILogsAthenaNamedQuery
    AgentEventAthenaNamedQuery:
        Description: Creates Athena table for Agent Events for analysis.
        Value: !Join
        - ''
        - - !Sub 'https://console.aws.amazon.com/athena/home?force&region=${AWS::Region}'
          - '#query/saved/'
          - !Ref AgentEventAthenaNamedQuery
    ReconciliationReportAthenaNamedQuery:
        Description: Creates Athena query for analyzing call times and phone type for billing reconcilitation.
        Value: !Join
        - ''
        - - !Sub 'https://console.aws.amazon.com/athena/home?force&region=${AWS::Region}'
          - '#query/saved/'
          - !Ref ReconciliationReportAthenaNamedQuery

Resources:

  FirehoseRole:
    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:
              - !Sub ${AWS::AccountId}
      Path: /


  FirehoseS3Policy:
    Type: AWS::IAM::Policy
    Properties:
      PolicyName: s3
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Action:
          - s3:AbortMultipartUpload
          - s3:GetBucketLocation
          - s3:GetObject
          - s3:ListBucket
          - s3:ListBucketMultipartUploads
          - s3:PutObject
          - s3:PutObjectAcl
          Resource:
          - !Sub 'arn:aws:s3:::${storageDestinationS3}/*'
          - !Sub 'arn:aws:s3:::${storageDestinationS3}'
        - Effect: Allow
          Action:
          - kinesis:GetRecords
          - kinesis:DescribeStream
          - kinesis:ListShards
          - kinesis:GetShardIterator
          Resource:
          - !Join 
            - ''
            - - !Sub arn:aws:kinesis:${AWS::Region}:${AWS::AccountId}:stream/
              - !Ref AgentEventKinesisStream
        - Effect: Allow
          Action:
          -  kms:GenerateDataKey
          -  kms:Decrypt
          Resource:
          -  '*'
      Roles:
      - Ref: FirehoseRole


  FirehoseLogsPolicy:
    Type: AWS::IAM::Policy
    Properties:
      PolicyName: logs
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Action:
          - logs:PutLogEvents
          - logs:CreateLogStream
          Resource:
          - !Sub arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/kinesisfirehose/${FirehoseCTRStream}:log-stream:*
          - !Sub arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/kinesisfirehose/${FireHoseAgentEvents}:log-stream:*
      Roles:
      - Ref: FirehoseRole

  FirehosePolicy:
    Type: AWS::IAM::Policy
    Properties:
      PolicyName: firehose
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Action:
          - firehose:PutRecord
          - firehose:PutRecordBatch
          - firehose:UpdateDestination
          Resource:
          - !Sub arn:aws:firehose:${AWS::Region}:${AWS::AccountId}:deliverystream/${FirehoseCTRStream}
          - Fn::Sub: arn:aws:firehose:${AWS::Region}:${AWS::AccountId}:deliverystream/${FireHoseAgentEvents}
          - Fn::Sub: arn:aws:firehose:${AWS::Region}:${AWS::AccountId}:deliverystream/*
      Roles:
      - Ref: FirehoseRole

  FirehoseLambdaTransformPolicy:
    Type: AWS::IAM::Policy
    Properties:  
      PolicyName: lambdatransform
      PolicyDocument:
        Version: '2012-10-17'
        Statement: 
          - Effect: Allow
            Action:
              - lambda:InvokeFunction
              - lambda:GetFunctionConfiguration
            Resource:
              - !Sub arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:${lineBreakTransformLambda}*
      Roles:
      - Ref: FirehoseRole


  FirehoseStreamsSubscribePolicy:
    Type: AWS::IAM::Policy
    DependsOn: AgentEventKinesisStream
    Properties:
      PolicyName: kinesis-streams
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Action:
          - kinesis:GetRecords
          - kinesis:DescribeStream
          - kinesis:ListShards
          - kinesis:GetShardIterator
          Resource:
          - !Join 
            - ''
            - - !Sub arn:aws:kinesis:${AWS::Region}:${AWS::AccountId}:stream/
              - !Ref AgentEventKinesisStream
      Roles:
      - Ref: FirehoseRole

  FireHoseCTRStreamLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub /aws/kinesisfirehose/${AWS::StackName}-AmazonConnect-CTR-to-S3

  FirehoseCTRStreamLogS3:
    Type: AWS::Logs::LogStream
    Properties:
      LogGroupName:
        Ref: FireHoseCTRStreamLogGroup

  FirehoseCTRStream:
    Type: AWS::KinesisFirehose::DeliveryStream
    DependsOn:
      - FireHoseCTRStreamLogGroup
      - FirehoseCTRStreamLogS3
    Properties:
      DeliveryStreamName: !Sub ${AWS::StackName}-CTR-to-S3
      ExtendedS3DestinationConfiguration:
        BucketARN: !Sub 'arn:aws:s3:::${storageDestinationS3}'
        Prefix: 'CTR/'
        ErrorOutputPrefix: 'CTRError/'
        BufferingHints: 
          IntervalInSeconds: 60
          SizeInMBs: 1
        CloudWatchLoggingOptions: 
          Enabled: true
          LogGroupName:
            Ref: FireHoseCTRStreamLogGroup
          LogStreamName:
            Ref: FirehoseCTRStreamLogS3
        CompressionFormat: UNCOMPRESSED
        RoleARN: !GetAtt FirehoseRole.Arn
  
  AgentEventKinesisStream:
    Type: AWS::Kinesis::Stream
    Properties:
      Name: !Sub ${AWS::StackName}-AgentEvents-Stream
      ShardCount: 1
      StreamEncryption:
        EncryptionType: KMS
        KeyId: alias/aws/kinesis

  FireHoseAgentEventsLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub /aws/kinesisfirehose/${AWS::StackName}-AE-to-S3

  FireHoseAgentEventsLogS3:
    Type: AWS::Logs::LogStream
    Properties:
      LogGroupName:
        Ref: FireHoseAgentEventsLogGroup

  lineBreakTransformLambda:
    Type: "AWS::Lambda::Function"
    Properties:
      Description:  >
          AWS Lambda Function to write to SQS queue for processing derived user details to S3 for diagnostic purposes.
      Handler: "index.handler"
      Role: !Sub ${describeUserWhisperLambdaRole.Arn}
      Runtime: "nodejs12.x"
      MemorySize: 128
      Timeout: 360
      Code:
        ZipFile: |
         'use strict';
          console.log('Loading function');
          const AWS = require('aws-sdk');

          //Record processing status enumerations
          const STATUS_OK = 'Ok';
          const STATUS_DROPPED = 'Dropped';
          const STATUS_FAIL = 'ProcessingFailed';

          function addLineBreak (data) {
              data = Buffer.from(data, 'base64').toString('utf-8');
              console.log('Processing data: ' + data);
              var i = data.search('}{');
              while (i>=0) {
                      data = data.substring(0,i) + '\n' + data.substring(i+2);
                      i = data.search('}{');
              }
              data = data + '\n';
              data = Buffer.from(data).toString('base64');
              return data;
          };

          function processRecord(record) {
              try {
                  record.data = addLineBreak(record.data);
                  return {
                      recordId: record.recordId,
                      result: STATUS_OK,
                      data: record.data
                  };
              } catch (err) {
                  console.log('Error occurred during transform: ' + err);
                  return {
                      recordId: record.recordId,
                      result: STATUS_FAIL,
                      data: record.data
                  };
              } 
          };

          exports.handler = (event, context, callback) => {
              /* Process the list of records and transform them */
              const output = event.records.map((record) => (
                  processRecord(record)
              ));
              console.log(`Processing completed.  Successful records ${output.length}.`);
              callback(null, { records: output });
          };

  FireHoseAgentEvents:
      Type: AWS::KinesisFirehose::DeliveryStream
      DependsOn:
        - FirehoseRole
        - FirehoseStreamsSubscribePolicy
        - AgentEventKinesisStream
        - FireHoseAgentEventsLogGroup
        - FireHoseAgentEventsLogS3
        - lineBreakTransformLambda
      Properties:
        DeliveryStreamName: !Sub ${AWS::StackName}-AE-to-S3
        DeliveryStreamType: KinesisStreamAsSource
        KinesisStreamSourceConfiguration:
          KinesisStreamARN: !GetAtt AgentEventKinesisStream.Arn
          RoleARN: !GetAtt FirehoseRole.Arn
        ExtendedS3DestinationConfiguration:
          ProcessingConfiguration:
            Enabled: 'true'
            Processors:
              - Parameters:
                  - ParameterName: LambdaArn
                    ParameterValue: !GetAtt lineBreakTransformLambda.Arn
                Type: Lambda 
          BucketARN: !Sub 'arn:aws:s3:::${storageDestinationS3}'
          Prefix: 'AgentEvents/'
          ErrorOutputPrefix: 'AgentEventsError/'
          BufferingHints: 
            IntervalInSeconds: 60
            SizeInMBs: 1
          CloudWatchLoggingOptions: 
            Enabled: true
            LogGroupName:
              Ref: FireHoseAgentEventsLogGroup
            LogStreamName:
              Ref: FireHoseAgentEventsLogS3
          CompressionFormat: UNCOMPRESSED
          RoleARN: !GetAtt FirehoseRole.Arn

# This lambda needs updating
  # FormatLogsforDBLambda:
  #   Type: 'AWS::Lambda::Function'
  #   Properties:
  #     Description:  >
  #         AWS Lambda Function to transform S3 data into a usable format for Amazon Connect Usage Diagnostics.
  #     Handler: "index.handler"
  #     Role: !Sub ${LambdaIAMRole.Arn}
  #     Runtime: "nodejs12.x"
  #     MemorySize: 128
  #     Timeout: 60
  #     Environment:
  #         Variables:
  #             bucketName: !Ref storageDestinationS3
  #     Code:
  #       ZipFile: |
  #         const AWS = require('aws-sdk');
  #         const s3 = new AWS.S3();
  #         exports.handler = async function(event, context) {
  #           console.log("EVENT: \n" + JSON.stringify(event))
  #           return 'This function will read incoming S3 records and transform them, placing them somewhere for analysis.  To be developed further'
  #         }

  # LambdaInvokePermission:
  #   Type: 'AWS::Lambda::Permission'
  #   Properties:
  #     FunctionName: !GetAtt FormatLogsforDBLambda.Arn
  #     Action: 'lambda:InvokeFunction'
  #     Principal: s3.amazonaws.com
  #     SourceAccount: !Ref 'AWS::AccountId'
  #     SourceArn: !Sub 'arn:aws:s3:::${storageDestinationS3}'

  LambdaIAMRole:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - lambda.amazonaws.com
            Action:
              - 'sts:AssumeRole'
      Path: /
      Policies:
        - PolicyName: root
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - 's3:GetBucketNotification'
                  - 's3:PutBucketNotification'
                Resource: !Sub 'arn:aws:s3:::${storageDestinationS3}'
              - Effect: Allow
                Action:
                  - 'logs:CreateLogGroup'
                  - 'logs:CreateLogStream'
                  - 'logs:PutLogEvents'
                Resource: 'arn:aws:logs:*:*:*'
              - Effect: Allow
                Action:
                - s3:AbortMultipartUpload
                - s3:GetBucketLocation
                - s3:GetObject
                - s3:ListBucket
                - s3:ListBucketMultipartUploads
                - s3:PutObject
                - s3:PutObjectAcl
                Resource:
                - !Sub 'arn:aws:s3:::${storageDestinationS3}/*'
                - !Sub 'arn:aws:s3:::${storageDestinationS3}'
              - Effect: Allow
                Action:
                - firehose:StartDeliveryStreamEncryption 
                - firehose:StopDeliveryStreamEncryption
                Resource:
                - !Sub 'arn:aws:firehose:${AWS::Region}:${AWS::AccountId}:deliverystream/${FirehoseCTRStream}'

  ConnectGlueDatabase:
    Type: AWS::Glue::Database
    Properties:
      DatabaseInput:
        Name: !Sub ${athenaDbName}
        Description: "Glue database for Athena queries for Amazon Connect Diagnostics"
      CatalogId: !Ref AWS::AccountId

  ContactTraceRecordsAthenaNamedQuery:
    Type: AWS::Athena::NamedQuery
    Properties:
      Database: !Ref ConnectGlueDatabase
      Description: "Generate table for CTR data"
      Name: !Sub ${AWS::StackName}-ctr
      QueryString: !Sub >
        CREATE EXTERNAL TABLE amazonconnect_CTR (
          AWSAccountId string,
          AWSContactTraceRecordFormatVersion string,
          Agent struct<
              ARN: string,
              AfterContactWorkDuration: int,
              AfterContactWorkEndTimestamp: string,
              AfterContactWorkStartTimestamp: string,
              AgentInteractionDuration: int,
              ConnectedToAgentTimestamp: string,
              CustomerHoldDuration: int,
              HierarchyGroups: string,
              LongestHoldDuration: int,
              NumberOfHolds: int,
              RoutingProfile: struct<
                  ARN: string,
                  Name: string
              >,
              Username: string
          >,
          AgentConnectionAttempts int,
          Attributes string,
          Channel string,
          ConnectedToSystemTimestamp string,
          ContactId string,
          CustomerEndpoint struct< 
              Address: string,
              Type: string 
          >,
          DisconnectTimestamp string,
          InitialContactId string,
          InitiationMethod string,
          InitiationTimestamp string,
          InstanceARN string,
          LastUpstringTimestamp string,
          MediaStreams array< 
              struct<
                  Type: string
              >
          >,
          NextContactId string,
          PreviousContactId string,
          Queue struct<
              ARN: string,
              DequeueTimestamp: string,
              Duration: int,
              EnqueueTimestamp: string,
              Name: string
          >,
          Recording string,
          Recordings string,
          SystemEndpoint struct<
              Address: string,
              Type: string 
          >,
          TransferCompletedTimestamp string,
          TransferredToEndpoint string  
        ) 
        ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe' LOCATION 's3://${storageDestinationS3}/CTR'

  APILogsAthenaNamedQuery:
    Type: AWS::Athena::NamedQuery
    Properties:
      Database: !Ref ConnectGlueDatabase
      Description: "Generate table for API Log data"
      Name: !Sub ${AWS::StackName}-apilogs
      QueryString: !Sub >
        CREATE EXTERNAL TABLE amazonconnect_APILogs (
          `User` struct<
              Id: string,
              Arn: string,
              Username: string,
              IdentityInfo: struct<
                  FirstName: string,
                  LastName: string,
                  Email: string
              >,
              PhoneConfig: struct<
                  PhoneType: string,
                  AutoAccept: string,
                  AfterContactWorkTimeLimit: int,
                  DeskPhoneNumber: string
              >,
              DirectoryUserId: string,
              SecurityProfileIds: array<string>,
              RoutingProfileId: string,
              HierarchyGroupId: string,
              Tags: string
          > COMMENT 'from deserializer',
          `Contact` struct<
              Attributes: string,
              Id: string
          > COMMENT 'from deserializer',
          `stepId` string COMMENT 'from deserializer',
          `dateStamp` string COMMENT 'from deserializer'
        ) 
        ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe' 
        STORED AS INPUTFORMAT 
          'org.apache.hadoop.mapred.TextInputFormat' 
        OUTPUTFORMAT 
          'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
        LOCATION 's3://${storageDestinationS3}/ApiFlowDetails'

  AgentEventAthenaNamedQuery:
    Type: AWS::Athena::NamedQuery
    Properties:
      Database: !Ref ConnectGlueDatabase
      Description: "Generate table for AgentEvent data"
      Name: !Sub ${AWS::StackName}-ae
      QueryString: !Sub >
        CREATE EXTERNAL TABLE amazonconnect_agentevents (
          AWSAccountId string,
          AgentARN string,
          CurrentAgentSnapshot struct<
              AgentStatus: struct<
                ARN: string,
                Name: string,
                StartTimestamp: string
              >,
              Configuration: struct<
                AgentHierarchyGroups: string,
                FirstName: string,
                LastName: string,
                RoutingProfile: struct<
                    ARN: string,
                    Concurrency: array<
                      struct<
                          AvailableSlots: int,
                          Channel: string,
                          MaximumSlots: int
                      >
                    >,
                    DefaultOutboundQueue: struct<
                      ARN: string,
                      Channels: array<
                          string
                      >,
                      Name: string
                    >,
                    InboundQueues: array<
                      struct<
                          ARN: string,
                          Channels: array<string>,
                          Name: string
                      >
                    >,
                    Name: string
                >,
                Username: string
              >,
              Contacts: array<string>
          >,
          EventId string,
          EventTimestamp string,
          EventType string,
          InstanceARN string,
          PreviousAgentSnapshot struct<
              AgentStatus: struct<
                ARN: string,
                Name: string,
                StartTimestamp: string
              >,
              Configuration: struct<
                AgentHierarchyGroups: array<string>,
                FirstName: string,
                LastName: string,
                RoutingProfile: struct<
                    ARN: string,
                    Concurrency: array<
                      struct<
                          AvailableSlots: int,
                          Channel: string,
                          MaximumSlots: int
                      >
                    >,
                    DefaultOutboundQueue: struct<
                      ARN: string,
                      Channels: array<string>,
                      Name: string
                    >,
                    InboundQueues: array<
                      struct<
                          ARN: string,
                          Channels: array<string>,
                          Name: string
                      >
                    >,
                    Name: string
                >,
                Username: string
              >,
              Contacts: array<
                struct<
                    Channel: string,
                    ConnectedToAgentTimestamp: string,
                    ContactId: string,
                    InitialContactId: string,
                    InitiationMethod: string,
                    Queue: struct<
                      ARN: string,
                      Name: string
                    >,
                    QueueTimestamp: string,
                    State: string,
                    StateStartTimestamp: string
                >
              >
          >,
          Version string
        ) 
        ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe' LOCATION 's3://${storageDestinationS3}/AgentEvents'

  ReconciliationReportAthenaNamedQuery:
    Type: AWS::Athena::NamedQuery
    Properties:
      Database: !Ref ConnectGlueDatabase
      Description: "Generate results for call times"
      Name: !Sub ${AWS::StackName}-ReconciliationReport
      QueryString: !Sub >
                SELECT 
                  ctr."$path" as CTR_GUID,
                  ctr.Agent.ARN as AgentARN,
                  api.User.PhoneConfig.PhoneType as AgentPhoneType,
                  ctr.initiationmethod as CallInitiationMethod,
                  ctr.ContactId,
                  split(ctr.Agent.RoutingProfile.ARN,'routing-profile/',2)[2] as CTRRoutingProfile,

                  ctr.InitiationTimestamp as CallInitiationTimestamp,
                  ctr.Agent.ConnectedToAgentTimestamp as ConnectedToAgentTimestamp,
                  ctr.DisconnectTimestamp as DisconnectTimestamp,
                  ctr.ConnectedToSystemTimestamp as ConnectedToSystemTimestamp,
                  
                  CASE (api.User.PhoneConfig.PhoneType = 'DESK_PHONE')
                    WHEN true THEN CAST(to_unixtime(from_iso8601_timestamp(ctr.DisconnectTimestamp)) as INTEGER)-CAST(to_unixtime(from_iso8601_timestamp(ctr.Agent.ConnectedToAgentTimestamp)) as INTEGER) 
                    ELSE null
                  END as AgentDeskPhone_inSeconds,

                  CASE (api.User.PhoneConfig.PhoneType = 'DESK_PHONE')
                    WHEN true then
                      CASE (CAST(to_unixtime(from_iso8601_timestamp(ctr.DisconnectTimestamp)) as INTEGER)-CAST(to_unixtime(from_iso8601_timestamp(ctr.Agent.ConnectedToAgentTimestamp)) as INTEGER) < (60))            
                        WHEN true THEN CAST(60 as INTEGER)            
                        ELSE CAST(
                        CAST( to_unixtime(from_iso8601_timestamp(ctr.DisconnectTimestamp)) as DECIMAL(20))-
                        CAST( to_unixtime( from_iso8601_timestamp(ctr.Agent.ConnectedToAgentTimestamp) ) as DECIMAL(20))
                        as INTEGER )
                      END
                    ELSE null
                  END as AgentDeskPhone_TelecomBilledSeconds,

                  CAST(to_unixtime(from_iso8601_timestamp(ctr.DisconnectTimestamp)) as INTEGER)-CAST(to_unixtime(from_iso8601_timestamp(ctr.TransferCompletedTimestamp)) as INTEGER) as TransferCallLength_inSeconds,
                  
                  CASE (CAST(to_unixtime(from_iso8601_timestamp(ctr.DisconnectTimestamp)) as INTEGER)-CAST(to_unixtime(from_iso8601_timestamp(ctr.TransferCompletedTimestamp)) as INTEGER) < (60))            
                    WHEN true THEN CAST(60 as INTEGER)            
                    ELSE CAST(
                      CAST( to_unixtime(from_iso8601_timestamp(ctr.DisconnectTimestamp)) as DECIMAL(20))-
                      CAST( to_unixtime( from_iso8601_timestamp(ctr.TransferCompletedTimestamp) ) as DECIMAL(20))
                      as INTEGER )
                    END as TransferCallLength_TelecomBilledSeconds,
                    
                  CAST(to_unixtime(from_iso8601_timestamp(ctr.DisconnectTimestamp)) as INTEGER)-CAST(to_unixtime(from_iso8601_timestamp(ctr.ConnectedToSystemTimestamp)) as INTEGER) as CustomerCallLength_inSeconds,

                  CASE (CAST(to_unixtime(from_iso8601_timestamp(ctr.DisconnectTimestamp)) as INTEGER)-CAST(to_unixtime(from_iso8601_timestamp(ctr.ConnectedToSystemTimestamp)) as INTEGER) < 10)
                    WHEN true THEN 10
                    ELSE CAST(to_unixtime(from_iso8601_timestamp(ctr.DisconnectTimestamp)) as INTEGER)-CAST(to_unixtime(from_iso8601_timestamp(ctr.ConnectedToSystemTimestamp)) as INTEGER)
                    END as CustomerCallLength_ConnectBilledSeconds,

                  CASE (CAST(to_unixtime(from_iso8601_timestamp(ctr.DisconnectTimestamp)) as INTEGER)-CAST(to_unixtime(from_iso8601_timestamp(ctr.ConnectedToSystemTimestamp)) as INTEGER) < (60))            
                    WHEN true THEN CAST(60 as INTEGER)            
                    ELSE CAST(
                      CAST( to_unixtime(from_iso8601_timestamp(ctr.DisconnectTimestamp)) as DECIMAL(20))-
                      CAST( to_unixtime( from_iso8601_timestamp(ctr.ConnectedToSystemTimestamp) ) as DECIMAL(20))
                      as INTEGER )
                    END as CustomerCallLength_TelecomBilledSeconds

                FROM "connect-diagnostics"."amazonconnect_ctr" ctr

                LEFT JOIN "connect-diagnostics"."amazonconnect_apilogs" api 
                  ON ctr.ContactId = api.Contact.Id 
                  AND split(ctr.Agent.RoutingProfile.ARN,'routing-profile/',2)[2] = api.User.RoutingProfileId

                -- WHERE
                --  from_iso8601_timestamp(ctr.InitiationTimestamp) BETWEEN date_add('hour', -8760, now()) AND now()
                --  from_iso8601_timestamp(ctr.InitiationTimestamp) BETWEEN from_iso8601_timestamp('2020-06-01T00:00:00Z') AND from_iso8601_timestamp('2020-07-01T00:00:00Z')
                
                ORDER BY ctr.InitiationTimestamp DESC


  CustomResourceLambdaFunction:
    Type: 'AWS::Lambda::Function'
    DependsOn:
      - AgentEventAthenaNamedQuery
      - APILogsAthenaNamedQuery
      - ContactTraceRecordsAthenaNamedQuery
    Properties:
      Handler: index.lambda_handler
      Role: !GetAtt LambdaIAMRole.Arn
      Code:
        ZipFile: |
            from __future__ import print_function
            import json
            import boto3
            import cfnresponse

            SUCCESS = "SUCCESS"
            FAILED = "FAILED"

            print('Loading function')
            s3 = boto3.resource('s3')
            firehose = boto3.client('firehose')

            def lambda_handler(event, context):
                print("Received event: " + json.dumps(event, indent=2))
                responseData={}
                try:
                    if event['RequestType'] == 'Delete':
                        print("Request Type:",event['RequestType'])
                        Bucket=event['ResourceProperties']['Bucket']
                        delete_notification(Bucket)
                        print("Sending response to custom resource after Delete")
                    elif event['RequestType'] == 'Create' or event['RequestType'] == 'Update':
                        print("Request Type:",event['RequestType'])
                        LambdaArn=event['ResourceProperties']['LambdaArn']
                        CTRFirehoseName=event['ResourceProperties']['CTRFirehoseName']
                        Bucket=event['ResourceProperties']['Bucket']
                        if event['RequestType'] == 'Create':
                            start_firehose_encryption(CTRFirehoseName)
                        add_notification(LambdaArn, Bucket)
                        responseData={'Bucket':Bucket}
                        print("Sending response to custom resource")
                    responseStatus = 'SUCCESS'
                except Exception as e:
                    print('Failed to process:', e)
                    responseStatus = 'FAILURE'
                    responseData = {'Failure': 'Something bad happened.'}
                cfnresponse.send(event, context, responseStatus, responseData)

            def start_firehose_encryption(DeliveryStreamName):
                response = firehose.start_delivery_stream_encryption(
                    DeliveryStreamName=DeliveryStreamName,
                    DeliveryStreamEncryptionConfigurationInput={
                        'KeyType' : 'AWS_OWNED_CMK'
                    }
                )
                print("Encryption request completed....")

                
            def add_notification(LambdaArn, Bucket):
                bucket_notification = s3.BucketNotification(Bucket)
                response = bucket_notification.put(
                  NotificationConfiguration={
                    'LambdaFunctionConfigurations': [
                      {
                          'LambdaFunctionArn': LambdaArn,
                          'Events': [
                              's3:ObjectCreated:*'
                          ]
                      }
                    ]
                  }
                )
                print("Put request completed....")
              
            def delete_notification(Bucket):
                bucket_notification = s3.BucketNotification(Bucket)
                response = bucket_notification.put(
                    NotificationConfiguration={}
                )
                print("Delete request completed....")
      Runtime: python3.9
      Timeout: 50

  # LambdaTrigger:
  #   Type: 'Custom::LambdaTrigger'
  #   DependsOn: LambdaInvokePermission
  #   Properties:
  #     ServiceToken: !GetAtt CustomResourceLambdaFunction.Arn
  #     LambdaArn: !GetAtt FormatLogsforDBLambda.Arn
  #     Bucket: !Ref storageDestinationS3
  #     CTRFirehoseName: !Ref FirehoseCTRStream
  #     StackName: !Sub ${AWS::StackName}
    
  allowConnectRequestLambdaPermission:
      Type: 'AWS::Lambda::Permission'
      Properties:
          FunctionName: !Ref describeUserWhisperLambda
          Action: 'lambda:InvokeFunction'
          Principal: connect.amazonaws.com
          SourceAccount: !Ref 'AWS::AccountId'

  describeUserWhisperLambdaRole:
      Type: "AWS::IAM::Role"
      Properties:
        AssumeRolePolicyDocument:
          Version: "2012-10-17"
          Statement:
            -
              Effect: "Allow"
              Principal:
                Service:
                  - "lambda.amazonaws.com"
              Action:
                - "sts:AssumeRole"
        Path: "/"
        Policies:
          -
            PolicyName: s3
            PolicyDocument:
              Version: '2012-10-17'
              Statement:
              - Effect: Allow
                Action:
                - s3:AbortMultipartUpload
                - s3:GetBucketLocation
                - s3:GetObject
                - s3:ListBucket
                - s3:ListBucketMultipartUploads
                - s3:PutObject
                - s3:PutObjectAcl
                Resource:
                - !Sub 'arn:aws:s3:::${storageDestinationS3}/*'
                - !Sub 'arn:aws:s3:::${storageDestinationS3}'
          -
            PolicyName: ConnectReadOnlyDiagnostics
            PolicyDocument:
              Version: "2012-10-17"
              Statement:
                -
                  Effect: "Allow"
                  Action:
                    - 'connect:DescribeUser'
                    - 'connect:DescribeUserHierarchyGroup'
                    - 'connect:DescribeUserHierarchyStructure'
                    - 'connect:GetContactAttributes'
                    - 'connect:GetCurrentMetricData'
                    - 'connect:ListUserHierarchyGroups'
                    - 'connect:ListUsers'
                    - 'connect:ListInstances'
                  Resource:
                    - !Sub 'arn:aws:connect:${AWS::Region}:${AWS::AccountId}:instance/${instanceIdParam}'
                    - !Sub 'arn:aws:connect:${AWS::Region}:${AWS::AccountId}:instance/${instanceIdParam}/*'                    
          -
            PolicyName: LogAccess
            PolicyDocument:
              Version: "2012-10-17"
              Statement:
                -
                  Effect: "Allow"
                  Action:
                    - 'logs:CreateLogGroup'
                    - 'logs:CreateLogStream'
                    - 'logs:PutLogEvents'
                  Resource: 'arn:aws:logs:*:*:*'
          -
            PolicyName: SQSAccess
            PolicyDocument:
              Version: "2012-10-17"
              Statement:
                -
                  Effect: "Allow"
                  Action:
                    - 'sqs:DeleteMessage'
                    - 'sqs:GetQueueAttributes'
                    - 'sqs:ReceiveMessage'
                    - 'sqs:SendMessage'
                  Resource: !Sub ${connectDescribeUserQueue.Arn} 

  describeUserWhisperLambda:
    Type: "AWS::Lambda::Function"
    DependsOn: 
      - describeUserWhisperLambdaRole
      - connectDescribeUserQueue
    Properties:
      Description:  >
          AWS Lambda Function to write to SQS queue for processing derived user details to S3 for diagnostic purposes.
      Handler: "index.handler"
      Role: !Sub ${describeUserWhisperLambdaRole.Arn}
      Runtime: "nodejs12.x"

      MemorySize: 128
      Timeout: 3
      Environment:
          Variables:
              bucketName: !Ref storageDestinationS3
              queueName: !Ref connectDescribeUserQueue

      Code:
        ZipFile: |
         const AWS = require('aws-sdk');
          const sqs = new AWS.SQS();
          const sqsQueue = process.env.queueName;

          async function queueConnectDescribe(queueUrl, messageBody) {
              var params = {
                  MessageBody: messageBody,
                  QueueUrl: queueUrl,
                  DelaySeconds: 0,
                };
                
                console.log(params);
                var result = await sqs.sendMessage(params).promise();
                return result;
              
          }


          exports.handler = async function(event, context) {
              console.log("EVENT: \n" + JSON.stringify(event))
              var date = new Date();
              var instanceArn = event['Details']['Parameters']['InstanceARN'];
              var instanceId = '';
              if (instanceArn) instanceId = instanceArn.split('/')[1];
              
              var userArn = event['Details']['Parameters']['AgentARN'];
              var userId = '';
              if (userArn) userId = userArn.split('/')[3];
              
              var contactId = event['Details']['Parameters']['ContactID'] || '';
              var stepId = event['Details']['Parameters']['StepID'] || '';
              
              //Add Connect API Call to Queue
              var messageBody = {"instanceId":instanceId, "userId": userId, "contactId": contactId, "stepId": stepId, "dateStamp": date.toISOString()};
              let status = await queueConnectDescribe(sqsQueue, JSON.stringify(messageBody));
              console.log(status);
              return {"status": "success"};
          }

  connectDescribeUserDLQ:
    Type: AWS::SQS::Queue
    Properties:
      KmsMasterKeyId: alias/aws/sqs
      MessageRetentionPeriod: 1209600
      ReceiveMessageWaitTimeSeconds: 0

  connectDescribeUserQueue:
    Type: AWS::SQS::Queue
    Properties: 
      KmsMasterKeyId: alias/aws/sqs
      MessageRetentionPeriod: 1200
      ReceiveMessageWaitTimeSeconds: 20
      RedrivePolicy: 
        deadLetterTargetArn : !Sub ${connectDescribeUserDLQ.Arn}
        maxReceiveCount : 3 
        
  connectLambdaWorker:
    Type: "AWS::Lambda::Function"
    DependsOn: 
      - describeUserWhisperLambdaRole
      - connectDescribeUserQueue
    Properties:
      Description:  >
          AWS Lambda Function to write API derived user details to S3 for diagnostic purposes.
      Handler: "index.handler"
      Role: !Sub ${describeUserWhisperLambdaRole.Arn}
      Runtime: "nodejs12.x"
      ReservedConcurrentExecutions: 1
      MemorySize: 128
      Timeout: 3
      Environment:
          Variables:
              bucketName: !Ref storageDestinationS3
              queueName: !Ref connectDescribeUserQueue
      Code:
        ZipFile: |
          const AWS = require('aws-sdk');
          const connect = new AWS.Connect({apiVersion: '2017-08-08'});
          const s3 = new AWS.S3();
          const s3Bucket = process.env.bucketName;

          function connectDescribeUser(instanceId, userId){
              var params = {
                  InstanceId: instanceId,
                  UserId: userId
              };
              var result;
              result = connect.describeUser(params).promise();
              return result
              
          }

          function connectGetContactAttributes(instanceId, contactId){
              var params = {
                  InitialContactId: contactId,
                  InstanceId: instanceId
              };
              var result;
              result = connect.getContactAttributes(params).promise();
              return result;
          }

          function s3putObject(data, bucket, userId, contactId, date) {
              //Update this to not overwrite if key already exists
              var Month = date.getUTCMonth()+1;
              var params = {
                  Body: JSON.stringify(data), 
                  Bucket: bucket, 
                  Key: 'ApiFlowDetails/'+date.getUTCFullYear()+'/'+Month+'/'+date.getUTCDate()+'/'+date.getUTCHours()+'/'+ userId+contactId+date.getUTCMinutes()+date.getUTCSeconds()
                };
                var s3result = s3.putObject(params).promise();
                return s3result;
          }

          async function getConnectData(instanceId, userId, contactId, attempt) {
              console.log('running Connect API Calls');
              if (userId) describeUserPromise = connectDescribeUser(instanceId, userId);
              if (contactId) getContactAttributesPromise = connectGetContactAttributes(instanceId, contactId);
              
              try {
                  var promiseResult = await Promise.all([describeUserPromise, getContactAttributesPromise]);
                  return promiseResult;
              } catch(err) {
                  console.log(err);
                  if (err && (attempt < 6 || attempt == undefined) ) {
                      if (attempt == undefined) attempt = 1;
                      console.log('Error: ' + attempt);
                      setTimeout(function(){return getConnectData(instanceId, userId, contactId, attempt+1);}, attempt*attempt*500);
                  } else {
                      console.log('Error: retry limit reached');
                      return [];
                  }
              }
              
          }

          exports.handler = async function(event, context) {
              console.log("EVENT: \n" + JSON.stringify(event))
              for (const record of event.Records) {

                  //Do Stuff To Records Here
                  var message = JSON.parse(record.body);
                  var date = new Date(message.dateStamp); //Get date from event, convert to date object
                  var instanceId = message.instanceId;
                  var userId = message.userId;        
                  var contactId = message.contactId;
                  var stepId = message.stepId;
                  
                  var describeUserPromise = '';
                  var getContactAttributesPromise = '';
                  
                  var promiseResult = await getConnectData(instanceId, userId, contactId);

                  var data = {};
                  data = promiseResult[0] || {};
                  data.Contact = promiseResult[1] || {};

                  data['Contact']['Id'] = contactId || '';
                  data.stepId = stepId || '';
                  data.dateStamp = date.toISOString();
                  var result = await s3putObject(data, s3Bucket, userId, contactId, date);
                  console.log(JSON.stringify({whisperFlowData: data, s3UploadResponse: result}));
                
              };
              
              return {"RecordsProcessed":event.Records.length};
          }

  WorkerEventSource:
    DependsOn:
      - connectLambdaWorker
      - connectDescribeUserQueue
    Type: AWS::Lambda::EventSourceMapping
    Properties: 
      BatchSize: 1
      Enabled: true
      EventSourceArn: !Sub ${connectDescribeUserQueue.Arn}
      FunctionName: !Ref connectLambdaWorker
