provider "aws" {
  region = "us-east-1"
}

# Terraform backends cannot contain interpolations
terraform {
  backend "s3" {
    bucket  = "orcd-terraform-state"
    key     = "prod/kafka-infra/kafka-cluster-db-deploy/msk-connect/terraform.tfstate"
    region  = "us-east-1"
    encrypt = "true"
  }
}

data "aws_caller_identity" "current" {}

module "mskconnect_neo_sink_db_secret" { 
  source = "git@github.com:theorchard/terraform-secrets-manager.git//?ref=1.0.6" 

  environment                    = var.environment 
  application_family             = var.application_family 
  additional_tags                = var.additional_tags 
  service_name                   = "mskconnect-neo-sink-db"
  secret_name                    = "NEO4J_CREDENTIALS" 
  secret_recovery_window_in_days = 7 
} 

data "aws_iam_policy_document" "mskconnect_db_deploy_secretsmanager_policy_document" {
  statement {
    effect = "Allow"
    actions = [
      "secretsmanager:GetResourcePolicy",
      "secretsmanager:GetSecretValue",
      "secretsmanager:DescribeSecret",
      "secretsmanager:ListSecretVersionIds",
    ]
    resources = [module.mskconnect_neo_sink_db_secret.secret_arn]
  }
}

data "aws_iam_policy_document" "assume_role_policy" {
  statement {
    effect  = "Allow"
    actions = ["sts:AssumeRole"]
    principals {
      type        = "Service"
      identifiers = ["kafkaconnect.amazonaws.com"]
    }
    condition {
      test     = "StringEquals"
      variable = "aws:SourceAccount"
      values   = [data.aws_caller_identity.current.account_id]
    }
    condition {
      test     = "ArnLike"
      variable = "aws:SourceArn"
      values   = ["arn:aws:kafkaconnect:${var.aws_region}:${data.aws_caller_identity.current.account_id}:connector/${var.environment}-neo4j-sink-*"]
    }
  }
}

# MSK-Connect secretsmanager policy.
resource "aws_iam_policy" "mskconnect_db_deploy_secretsmanager_policy" {
  name   = "MSKConnect-${var.environment}-db-deploy-secretsmanager-policy"
  policy = data.aws_iam_policy_document.mskconnect_db_deploy_secretsmanager_policy_document.json
}

# MSK-Connect execution role.
resource "aws_iam_role" "mskconnect_db_deploy_execution_role" {
  name               = "MSKConnect-${var.environment}-db-deploy-execution-role"
  assume_role_policy = data.aws_iam_policy_document.assume_role_policy.json
}

# Attach policy to the role.
resource "aws_iam_role_policy_attachment" "mskconnect_db_deploy_execution_role_policy_attachment" {
  role       = aws_iam_role.mskconnect_db_deploy_execution_role.id
  policy_arn = aws_iam_policy.mskconnect_db_deploy_secretsmanager_policy.arn
}

resource "aws_mskconnect_worker_configuration" "mskconnect-neo-sink-worker" {
  name                    = "${var.environment}-mskconnect-neo-sink-worker"
  properties_file_content = file("neo-sink-static-properties.conf")
}

resource "aws_mskconnect_custom_plugin" "mskconnect-neo-sink-plugin" {
  name         = "${var.environment}-mskconnect-neo-sink-source-plugin"
  content_type = "ZIP"
  location {
    s3 {
      bucket_arn = "arn:aws:s3:::${var.environment}-kafka-connect"
      file_key   = "mskconnect-plugins/neo4j-sink-source-2.0.2.zip"
    }
  }
}