Description: >
  This template deploys AWS Glue jobs with Apache Iceberg connector in AWS Marketplace, which extract Amazon Customer Review Datasets and run Iceberg operations such as creating an Iceberg table, reading/writing data from/to the table, merging two tables and performing time-travels.
Parameters:
  IcebergDatabaseName:
    Description: >
      This is a database name for registering Iceberg tables. If necessary, you can specify a custom database name.
    Type: String
    Default: iceberg_blog_default
  DynamoDBTableName:
    Description: >
      This is a name of DynamoDB table which a Glue job for Iceberg will newly create.
      This table will be used for Iceberg commit lock. 
      Please note you don't specify the name of existing table.
    Type: String
    Default: myGlueLockTable

Resources:
  IcebergDataS3Bucket:
    Type: 'AWS::S3::Bucket'
    DeletionPolicy: Delete
    Properties:
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: AES256
      Tags:
        - Key: CreatedBy
          Value: GlueIcebergConnectorBlog
  IcebergResourcePrepLambdaRole:
    Type: 'AWS::IAM::Role'
    Properties:
      RoleName: IcebergResourcePrepLambdaRole
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - lambda.amazonaws.com
            Action:
              - 'sts:AssumeRole'
      Path: /
      Tags:
        - Key: CreatedBy
          Value: GlueIcebergConnectorBlog
  IcebergResourcePrepLambdaPolicy:
    Type: 'AWS::IAM::ManagedPolicy'
    DependsOn:
      - IcebergDataS3Bucket
      - IcebergResourcePrepLambdaRole
    Properties:
      ManagedPolicyName: IcebergResourcePrepLambda
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Action:
              - 's3:GetObject'
              - 's3:PutObject'
              - 's3:ListObject'
              - 's3:DeleteObject'
              - 's3:ListBucket'
            Resource:
              - !Join ['', ['arn:aws:s3:::', !Ref IcebergDataS3Bucket]]
              - !Join ['', ['arn:aws:s3:::', !Ref IcebergDataS3Bucket, '/*']]
          - Effect: Allow
            Action:
              - 's3:GetObject'
              - 's3:ListBucket'
            Resource:
              - 'arn:aws:s3:::aws-bigdata-blog'
              - 'arn:aws:s3:::aws-bigdata-blog/artifacts/glue-iceberg-connector/*'
          - Effect: Allow
            Action:
              - 'dynamodb:DescribeTable'
              - 'dynamodb:DeleteTable'
            Resource:
              - !Sub arn:aws:dynamodb:${AWS::Region}:${AWS::AccountId}:table/${DynamoDBTableName}
          - Effect: Allow
            Action:
              - 'lambda:TagResource'
              - 'lambda:ListTags'
            Resource: !GetAtt IcebergResourcePrepLambda.Arn
          - Effect: Allow
            Action:
              - 'logs:*'
            Resource: !Sub arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/*
      Roles:
        - IcebergResourcePrepLambdaRole
  IcebergConnectorGlueJobRole:
    Type: 'AWS::IAM::Role'
    Properties:
      RoleName: IcebergConnectorGlueJobRole
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - glue.amazonaws.com
            Action:
              - 'sts:AssumeRole'
      ManagedPolicyArns:
        - 'arn:aws:iam::aws:policy/AmazonEC2ContainerRegistryReadOnly'
        - 'arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole'
      Path: /
      Tags:
        - Key: CreatedBy
          Value: GlueIcebergConnectorBlog
  IcebergConnectorGlueJobPolicy:
    Type: 'AWS::IAM::ManagedPolicy'
    DependsOn: IcebergConnectorGlueJobRole
    Properties: 
      ManagedPolicyName: IcebergConnectorGlueJobPolicy
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Action:
              - 'dynamodb:describeTable'
              - 'dynamodb:CreateTable'
              - 'dynamodb:BatchGetItem'
              - 'dynamodb:BatchWriteItem'
              - 'dynamodb:DeleteItem'
              - 'dynamodb:GetItem'
              - 'dynamodb:PutItem'
              - 'dynamodb:UpdateItem'
              - 'dynamodb:Query'
              - 'dynamodb:Scan'
            Resource: !Sub arn:aws:dynamodb:${AWS::Region}:${AWS::AccountId}:table/${DynamoDBTableName}
          - Effect: Allow
            Action:
              - 'iam:PassRole'
            Resource: '*'
          - Effect: Allow
            Action:
              - 's3:GetBucketLocation'
              - 's3:ListBucket'
              - 's3:GetBucketAcl'
              - 's3:GetObject'
              - 's3:PutObject'
              - 's3:DeleteObject'
            Resource:
              - !Join ['', ['arn:aws:s3:::', !Ref IcebergDataS3Bucket]]
              - !Join ['', ['arn:aws:s3:::', !Ref IcebergDataS3Bucket, '/*']]
              - 'arn:aws:s3:::amazon-reviews-pds'
              - 'arn:aws:s3:::amazon-reviews-pds/*'
      Roles:
        - IcebergConnectorGlueJobRole
  IcebergDatabase:
    Type: AWS::Glue::Database
    Properties:
      CatalogId: !Ref AWS::AccountId
      DatabaseInput:
        Name: !Ref IcebergDatabaseName
  IcebergResourcePrepLambda:
    Type: 'AWS::Lambda::Function'
    Properties:
      Code:
        ZipFile: |
          import json
          import boto3
          import urllib3
          DEFAULT_KEYS = ['StackId', 'RequestId', 'LogicalResourceId']
          DDB_TABLE_DELETE_TAG_NAME = 'deleteDynamoDBTable'
          USER_CONFIG_KEY = 'config/user_config.json'


          def exists_dynamodb_table(client, table_name: str) -> bool:
              """
              Check if a DynamoDB table exists
              :param client: boto3 dynamodb client
              :param table_name: DynamoDB table name that a user specified in a CloudFormation template
              :return: If the specified DynamoDB table exists, return True, otherwise returns False
              """
              try:
                  client.describe_table(TableName=table_name)
                  return True
              except client.exceptions.ResourceNotFoundException:
                  return False


          def has_dynamo_table_delete_tag_in_lambda(client, arn: str, tag_name: str) -> bool:
              """
              Check if a Lambda function has a tag to delete a DynamoDB table when a user clean up resources
              :param client: boto3 lambda function client
              :param arn: a lambda function ARN
              :param tag_name:
              :return: If the tag exists, return True, otherwise False.
              """
              tags = client.list_tags(Resource=arn)['Tags']
              return tag_name in tags


          def response_constructor(response: dict, event: dict, event_keys: list, **kwargs) -> dict:
              """
              Construct a custom response that is passed to CloudFormation
              :param response:
              :param event: custom resource event that is passed to a Lambda function from CloudFormation
              :param event_keys: event keys that is used to create a custom resource response
              :return: return a custom response for a pre-signed URL of a custom resource event 
              """
              for event_key in event_keys: response.update({event_key: event[event_key]})
              for key in kwargs: response.update({key: kwargs[key]})
              return response


          http = urllib3.PoolManager()
          ddb = boto3.client('dynamodb')
          lam = boto3.client('lambda')
          s3c = boto3.client('s3')
          s3r = boto3.resource('s3')


          def lambda_handler(event, context):
              """
              This lambda handler provides custom functionalities to prepare and remove the blog related resources. Specifically,
              In a 'Create' request,
                  1. check if a user-specified DynamoDB table exists, if it exists, raise NameError
                  2. if the table doesn't exist, add a tag to this lambda function. This tag determines if this Lambda function removes a DynamoDB table created by Iceberg in resource-cleanup process.
                      If this tag is set to the lambda function, this lambda function delete the table,
                      If this tag is not set, this lambda function doesn't the table. This tag also prevents from deleting a table by lambda even if a user specifies a existing table name. 
                  3. upload user config that is defined in a CloudFormation template by a user
                  4. then, return a SUCCESS response to the custom resource pre-signed URL
              In a 'Delete' request, 
                  1. delete all objects in a S3 bucket
                  2. check if the DynamoDB table exists and this lambda function has the tag in itself. If yes, this lambda function deletes the table.
                  3. then, return a SUCCESS response to the customer resource pre-signed URL
              In a 'UPDATE' request, nothing happens and return a SUCCESS to the customer resource pre-signed URL
              Otherwise, raise NameError
              """
              request_type = event['RequestType']
              resource_props = event['ResourceProperties']
              lambda_arn = resource_props['ServiceToken']
              database_name = resource_props['CatalogDatabaseName']
              table_name = resource_props['CatalogTableName']
              report_table_name = resource_props['CatalogReportTableName']
              dynamodb_table = resource_props['DynamoDBTable']
              s3_bucket = resource_props['S3Data']
              user_config = {
                  'database_name': database_name,
                  'table_name': table_name,
                  'report_table_name': report_table_name,
                  'warehouse_path': f's3://{s3_bucket}/data',
                  'dynamodb_table': dynamodb_table
              }
              if request_type == 'Create':
                  try:
                      if exists_dynamodb_table(client=ddb, table_name=dynamodb_table):
                          raise NameError('The specified DynamoDB table already exists. Type a non-exsiting table name.')
                      lam.tag_resource(Resource=lambda_arn, Tags={DDB_TABLE_DELETE_TAG_NAME: 'yes'})  # Add a tag to this lambda function
                      s3r.Object(bucket_name=s3_bucket, key=USER_CONFIG_KEY).put(Body=json.dumps(user_config))  # Upload user config
                      response = response_constructor(response={}, event=event, event_keys=DEFAULT_KEYS, Status='SUCCESS', PhysicalResourceId='InitialResources-Create')
                  except Exception as e:
                      response = response_constructor(response={}, event=event, event_keys=DEFAULT_KEYS, Status='FAILED', PhysicalResourceId='InitialResources-CreateFailure', Reason=str(e))
              elif request_type == 'Delete':
                  try:
                      s3r.Bucket(name=s3_bucket).objects.delete()  # Delete all objects in a specified S3 bucket
                      if exists_dynamodb_table(client=ddb, table_name=dynamodb_table) \
                              and has_dynamo_table_delete_tag_in_lambda(client=lam, arn=lambda_arn, tag_name=DDB_TABLE_DELETE_TAG_NAME):
                          ddb.delete_table(TableName=dynamodb_table)  # Delete a DynamoDB table
                      response = response_constructor(response={}, event=event, event_keys=DEFAULT_KEYS, Status='SUCCESS', PhysicalResourceId='InitialResources-Delete')
                  except Exception as e:
                      response = response_constructor(response={}, event=event, event_keys=DEFAULT_KEYS, Status='FAILED', PhysicalResourceId='InitialResources-DeleteFailure', Reason=str(e))
              elif request_type == 'Update':
                  response = response_constructor(response={}, event=event, event_keys=DEFAULT_KEYS, Status='SUCCESS', PhysicalResourceId='InitialResources-NoChanges')
              else:
                  raise NameError(f'The RequestType: {request_type} is invalid.')
              http.request('PUT', event['ResponseURL'], body=json.dumps(response), headers={'Content-Type': 'application/json'}, retries=False)  # Return a SUCCESS or FAILED response to the event presigned-URL
              print(response)
              return response

      FunctionName: iceberg_glue_job_resource_prep
      MemorySize: 128
      Timeout: 900
      Runtime: python3.9
      Description: >
        A Lambda fucntion which create Glue jobs for Iceberg operations and copy Glue job scripts from S3.
        It also helps cleanup resources when deleting this CloudFromation stack.
      Handler: index.lambda_handler
      Role: !GetAtt IcebergResourcePrepLambdaRole.Arn
      Tags: 
        - Key: CreatedBy
          Value: GlueIcebergConnectorBlog
  ExecuteIcebergResourceProcess:
    Type: 'Custom::ExecuteIcebergResourceProcess'
    DependsOn:
      - IcebergDataS3Bucket
      - IcebergResourcePrepLambdaRole
      - IcebergResourcePrepLambdaPolicy
      - IcebergResourcePrepLambda
    Properties:
      ServiceToken: !GetAtt IcebergResourcePrepLambda.Arn
      CatalogDatabaseName: !Ref IcebergDatabaseName
      CatalogTableName: 'acr_iceberg'
      CatalogReportTableName: 'acr_iceberg_report'
      DynamoDBTable: !Ref DynamoDBTableName
      S3Data: !Ref IcebergDataS3Bucket

Outputs:
  IcebergDataS3Bucket:
    Description: An Amazon S3 bucket name for Iceberg
    Value: !Ref IcebergDataS3Bucket
