provider "aws" {
  region = var.aws_region
}

terraform {
  backend "s3" {
    bucket  = "dev-orcd-terraform-state"
    key     = "dev/kdh/flink/message-tranformation/terraform.tfstate"
    region  = "us-east-1"
    encrypt = "true"
  }
}

data "aws_caller_identity" "current" {}

module "flink_dynamodb_table" {
  source = "git@github.com:theorchard/terraform-dynamodb.git//?ref=3.3.1"

  env                            = var.environment
  table_name                     = "${var.service_name}-v1"
  hash_key                       = var.hash_key
  range_key                      = var.range_key
  range_key_enabled              = true
  server_side_encryption_enabled = true
  application_family             = var.application_family
  use_on_demand                  = true

  attribute = [
    {
      name = var.hash_key
      type = "S"
    },
    {
      name = var.range_key
      type = "S"
    }
  ]
}


data "aws_iam_policy_document" "flink_studio_assume_role" {
  statement {
    actions = ["sts:AssumeRole"]
    principals {
      type = "Service"
      identifiers = ["kinesisanalytics.amazonaws.com"]
    }
  }
}

resource "aws_iam_role" "flink_studio_role" {
  name               = "${var.environment}-${var.service_name}-role"
  assume_role_policy = data.aws_iam_policy_document.flink_studio_assume_role.json
}

data "aws_iam_policy_document" "flink_studio_policy_doc" {
  statement {
    actions = [
      "kafka:*"
    ]
    resources = [
      "arn:aws:kafka:us-east-1:103233932089:cluster/dev-managed-kafka-cdc-destination/194e93a4-1a4e-4e96-85bd-99b4ebce8df0-11",
    ]
  }

  statement {
    actions = [
      "s3:Create*",
      "s3:Delete*",
      "s3:Get*",
      "s3:Put*",
      "s3:Replicate*",
      "s3:Restore*",
      "s3:Update*",
    ]
    resources = [
      "arn:aws:s3:::dev-cucumbers",
      "arn:aws:s3:::dev-cucumbers/flink/*"
    ]
  }

  statement {
    actions = [
      "dynamodb:PutItem",
      "dynamodb:UpdateItem",
      "dynamodb:GetItem",
      "dynamodb:Query",
      "dynamodb:Scan"
    ]
    resources = [
      "arn:aws:dynamodb:us-east-1:103233932089:table/dev_flink-message-transformation-v1",
    ]
  }

  statement {
    actions = [
      "logs:PutLogEvents",
      "logs:CreateLogStream",
      "logs:CreateLogGroup"
    ]
    resources = [
      "arn:aws:logs:${var.aws_region}:${data.aws_caller_identity.current.account_id}:log-group:${var.environment}-${var.service_name}",
      "arn:aws:logs:${var.aws_region}:${data.aws_caller_identity.current.account_id}:log-group:${var.environment}-${var.service_name}:log-stream:*"
    ]
  }
}

resource "aws_iam_policy" "flink_studio_policy" {
  name   = "${var.environment}-${var.service_name}-flink-studio-policy"
  policy = data.aws_iam_policy_document.flink_studio_policy_doc.json
}

resource "aws_iam_role_policy_attachment" "flink_studio_attach" {
  role       = aws_iam_role.flink_studio_role.name
  policy_arn = aws_iam_policy.flink_studio_policy.arn
}