locals {
  name_prefix              = lower("${var.env_prefix}-${var.project_group}")
  acccount_id              = data.aws_caller_identity.current.account_id
  policy_base_arn          = "arn:aws:iam::${local.acccount_id}:policy/${local.name_prefix}"
  enable_processing        = var.enable_processing ? "true" : "false"
  convesion_configuration  = var.enable_conversion ? ["Enabled"] : []
  processing_configuration = var.enable_processing ? ["Enabled"] : []
  s3_backup_mode           = var.enable_s3_backup_mode ? "Enabled" : "Disabled"
  s3_backup_configuration  = var.enable_s3_backup_mode ? ["Enabled"] : []
  s3_prefix                = "${var.s3_prefix}${var.s3_path}"
  s3_error_output_prefix   = "${var.s3_error_output_prefix}${var.s3_error_output_path}"
  s3_backup_prefix         = var.s3_backup_id == "" ? "${var.project}_stream/${var.s3_prefix}" : "${var.s3_backup_id}${var.s3_prefix}"
  s3_backup_path           = var.s3_backup_id == "" ? "${var.project}_stream/${var.s3_prefix}${var.s3_path}" : "${var.s3_backup_id}${var.s3_prefix}${var.s3_path}"

  glue_catalog_database_name = element(concat(aws_glue_catalog_database.glue_catalog_database.*.name, [""]), 0)
  glue_catalog_table_name    = element(concat(aws_glue_catalog_table.glue_catalog_table.*.name, [""]), 0)
  glue_iam_role_arn          = element(concat(aws_iam_role.glue_iam_role.*.arn, [""]), 0)
  glue_iam_role_name         = element(concat(aws_iam_role.glue_iam_role.*.name, [""]), 0)
  glue_iam_policy_arn        = element(concat(aws_iam_policy.glue_iam_policy.*.arn, [""]), 0)
}

data "aws_caller_identity" "current" {}

resource "aws_cloudwatch_log_group" "firehose_cloudwatch_log_group" {
  name = "/aws/kinesisfirehose/${local.name_prefix}-${var.project}_stream"

  tags = merge(
    var.common_tags,
    {
      project                  = "Chartmetric",
      service                  = "CloudWatch"
      plat_env_project_service = "${var.aggregated_tag}_CHTMTRC_CW"
    }
  )
}

resource "aws_cloudwatch_log_stream" "firehose_cloudwatch_log_stream" {
  name           = "S3Delivery"
  log_group_name = aws_cloudwatch_log_group.firehose_cloudwatch_log_group.name
}

resource "aws_glue_catalog_database" "glue_catalog_database" {
  count = var.create_glue_database ? 1 : 0

  name = "${local.name_prefix}-${var.project}_db"
}

resource "aws_glue_catalog_table" "glue_catalog_table" {
  count = var.create_glue_database ? 1 : 0

  name          = "tbl${var.project}"
  database_name = element(concat(aws_glue_catalog_database.glue_catalog_database.*.name, [""]), 0)
}

resource "aws_kinesis_firehose_delivery_stream" "kinesis_firehose_delivery_stream" {
  name        = "${local.name_prefix}-${var.project}_stream"
  destination = "extended_s3"

  extended_s3_configuration {
    role_arn            = aws_iam_role.firehose_iam_role.arn
    bucket_arn          = var.bucket_arn
    compression_format  = var.compression_format
    buffering_size      = var.buffer_size
    buffering_interval  = var.buffer_interval
    error_output_prefix = local.s3_error_output_prefix
    prefix              = local.s3_prefix
    s3_backup_mode      = local.s3_backup_mode

    dynamic "data_format_conversion_configuration" {
      for_each = local.convesion_configuration

      content {
        input_format_configuration {
          deserializer {
            hive_json_ser_de {}
          }
        }

        output_format_configuration {
          serializer {
            parquet_ser_de {}
          }
        }

        schema_configuration {
          database_name = local.glue_catalog_database_name
          role_arn      = local.glue_iam_role_arn
          table_name    = local.glue_catalog_table_name
        }
      }
    }

    dynamic "processing_configuration" {
      for_each = local.processing_configuration

      content {
        enabled = local.enable_processing

        processors {
          type = "Lambda"

          parameters {
            parameter_name  = "LambdaArn"
            parameter_value = "${var.lambda_process_arn}:$LATEST"
          }

          /* parameters {
            parameter_name  = "BufferSizeInMBs"
            parameter_value = "1"
          } */

          parameters {
            parameter_name  = "BufferIntervalInSeconds"
            parameter_value = 900
          }
        }
      }
    }

    cloudwatch_logging_options {
      enabled         = true
      log_group_name  = aws_cloudwatch_log_group.firehose_cloudwatch_log_group.name
      log_stream_name = aws_cloudwatch_log_stream.firehose_cloudwatch_log_stream.name
    }

    dynamic "s3_backup_configuration" {
      for_each = local.s3_backup_configuration

      content {
        bucket_arn          = var.backup_bucket_arn
        role_arn            = aws_iam_role.firehose_iam_role.arn
        compression_format  = var.backup_compression_format
        buffering_size      = var.buffer_size
        buffering_interval  = var.buffer_interval
        prefix              = local.s3_backup_path
        error_output_prefix = "!{firehose:error-output-type}/"

        cloudwatch_logging_options {
          enabled         = true
          log_group_name  = aws_cloudwatch_log_group.firehose_cloudwatch_log_group.name
          log_stream_name = aws_cloudwatch_log_stream.firehose_cloudwatch_log_stream.name
        }
      }
    }
  }

  tags = merge(var.common_tags,
    {
      project                  = "Chartmetric",
      service                  = "Kinesis Firehose",
      plat_env_project_service = "${var.aggregated_tag}_CHTMTRC_KFH"
    }
  )
}

resource "aws_iam_role" "firehose_iam_role" {
  name = "${local.name_prefix}-${var.project}_role"

  assume_role_policy = <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Action": "sts:AssumeRole",
      "Principal": {
        "Service": "firehose.amazonaws.com"
      },
      "Effect": "Allow",
      "Sid": ""
    }
  ]
}
EOF
}

data "aws_iam_policy_document" "firehose_iam_policy_document" {
  statement {
    effect = "Allow"

    actions = [
      "s3:AbortMultipartUpload",
      "s3:GetBucketLocation",
      "s3:GetObject",
      "s3:ListBucket",
      "s3:ListBucketMultipartUploads",
      "s3:PutObject",
    ]

    resources = [
      var.bucket_arn,
      var.backup_bucket_arn,
      "${var.bucket_arn}/${var.s3_prefix}*",
      "${var.bucket_arn}/${var.s3_error_output_prefix}*",
      "${var.backup_bucket_arn}/${local.s3_backup_prefix}*",
    ]
  }

  statement {
    effect = "Allow"

    actions = [
      "kinesis:DescribeStream",
      "kinesis:GetShardIterator",
      "kinesis:GetRecords",
    ]

    resources = [
      "arn:aws:kinesis:${var.region}:${local.acccount_id}:stream/${local.name_prefix}-${var.project}_stream"
    ]
  }

  statement {
    effect = "Allow"

    actions = [
      "logs:PutLogEvents",
    ]

    resources = [
      "arn:aws:logs:${var.region}:${local.acccount_id}:log-group:/aws/kinesisfirehose/${local.name_prefix}-${var.project}_stream:log-stream:*"
    ]
  }

  dynamic "statement" {
    for_each = local.processing_configuration

    content {
      effect = "Allow"

      actions = [
        "lambda:InvokeFunction",
        "lambda:GetFunctionConfiguration",
      ]

      resources = [
        "${var.lambda_process_arn}:$LATEST"
      ]
    }
  }
}

resource "aws_iam_policy" "firehose_iam_policy" {
  name   = "${local.name_prefix}-${var.project}-firehose-s3"
  policy = data.aws_iam_policy_document.firehose_iam_policy_document.json
}

resource "aws_iam_role_policy_attachment" "firehose_iam_role_policy_attachment" {
  policy_arn = aws_iam_policy.firehose_iam_policy.arn
  role       = aws_iam_role.firehose_iam_role.name
}

resource "aws_iam_role" "glue_iam_role" {
  count = var.create_glue_database ? 1 : 0

  name = "${local.name_prefix}-glue_role"

  assume_role_policy = <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Action": "sts:AssumeRole",
      "Principal": {
        "Service": "glue.amazonaws.com"
      },
      "Effect": "Allow",
      "Sid": ""
    }
  ]
}
EOF
}

data "aws_iam_policy_document" "glue_iam_policy_document" {
  count = var.create_glue_database ? 1 : 0

  statement {
    effect = "Allow"
    actions = [
      "glue:*",
    ]
    resources = [
      "arn:aws:glue:${var.region}:${local.acccount_id}:catalog",
      "arn:aws:glue:${var.region}:${local.acccount_id}:database/${local.glue_catalog_database_name}",
      "arn:aws:glue:${var.region}:${local.acccount_id}:table/${local.glue_catalog_database_name}/${local.glue_catalog_table_name}",
    ]
  }
}

resource "aws_iam_policy" "glue_iam_policy" {
  count = var.create_glue_database ? 1 : 0

  name   = "${local.name_prefix}-${var.project}-glue"
  policy = element(concat(data.aws_iam_policy_document.glue_iam_policy_document.*.json, [""]), 0)
}

resource "aws_iam_role_policy_attachment" "glue_iam_role_policy_attachment" {
  count = var.create_glue_database ? 1 : 0

  policy_arn = local.glue_iam_policy_arn
  role       = local.glue_iam_role_name
}