locals {
  raw_topics = var.topics

  topics = {
    for topic in local.raw_topics : topic.name => {
      name        = topic.name
      owner       = topic.owner
      cluster_ref = topic.clusterRef
      cluster     = var.kafka_cluster_refs[topic.clusterRef]

      partitions = try(
        tonumber(topic.partitions[var.environment]),
        try(tonumber(topic.partitions._default), try(tonumber(topic.partitions), var.kafka_cluster_refs[topic.clusterRef].default_partitions))
      )

      replication_factor = try(
        tonumber(topic.replicationFactor[var.environment]),
        try(tonumber(topic.replicationFactor._default), try(tonumber(topic.replicationFactor), var.kafka_cluster_refs[topic.clusterRef].default_replication_factor))
      )

      config = {
        for key, value in merge(
          try(topic.inheritDefaultConfig, true) ? try(var.kafka_cluster_refs[topic.clusterRef].default_topic_config, {}) : {},
          try(topic.config, {})
        ) : key => try(tostring(value[var.environment]), try(tostring(value._default), tostring(value)))
      }

      deletion_policy = try(topic.deletionPolicy, "orphan")

      owner_roles = try(topic.ownerRoles, null) == null ? var.default_user_roles : topic.ownerRoles
    }
  }

  user_grants = flatten([
    for user in var.users : [
      for permission in try(user.permissions, []) : [
        for role in permission.roles : {
          cluster_ref = user.clusterRef
          user        = user.name
          topic_name  = permission.topic
          role        = role
        }
      ]
    ]
  ])

  permission_tuples = concat(
    flatten([
      for topic in values(local.topics) : [
        for role in topic.owner_roles : {
          cluster_ref = topic.cluster_ref
          user        = topic.owner
          topic_name  = topic.name
          role        = role
        }
      ]
    ]),
    local.user_grants
  )

  user_clusters = {
    for item in distinct(concat(
      [
        for tuple in local.permission_tuples : "${tuple.cluster_ref}:${tuple.user}"
      ],
      [
        for user in var.users : "${user.clusterRef}:${user.name}"
      ]
      )) : item => {
      cluster_ref = split(":", item)[0]
      user        = split(":", item)[1]
      cluster_id  = var.kafka_cluster_refs[split(":", item)[0]].cluster_id
    }
  }

  user_permissions = {
    for user_key, user in local.user_clusters : user_key => distinct([
      for tuple in local.permission_tuples : {
        topic_name = tuple.topic_name
        role       = tuple.role
      }
      if "${tuple.cluster_ref}:${tuple.user}" == user_key
    ])
  }

}

resource "random_password" "kafka_user" {
  for_each = var.create_users ? local.user_clusters : {}

  length  = var.user_password_length
  special = false
  upper   = true
  lower   = true
  numeric = true

  lifecycle {
    ignore_changes = all
  }
}

resource "yandex_mdb_kafka_topic" "this" {
  for_each = local.topics

  cluster_id         = each.value.cluster.cluster_id
  name               = each.value.name
  partitions         = each.value.partitions
  replication_factor = each.value.replication_factor

  dynamic "topic_config" {
    for_each = length(each.value.config) == 0 ? [] : [each.value.config]
    content {
      cleanup_policy = (
        try(topic_config.value["cleanup.policy"], null) == "delete" ? "CLEANUP_POLICY_DELETE" :
        try(topic_config.value["cleanup.policy"], null) == "compact" ? "CLEANUP_POLICY_COMPACT" :
        null
      )
      compression_type = try(topic_config.value["compression.type"], null)
      min_insync_replicas = try(
        tonumber(topic_config.value["min.insync.replicas"]),
        null
      )
      retention_bytes = try(
        tonumber(topic_config.value["retention.bytes"]),
        null
      )
      retention_ms = try(
        tonumber(topic_config.value["retention.ms"]),
        null
      )
      segment_bytes = try(
        tonumber(topic_config.value["segment.bytes"]),
        null
      )
    }
  }

  lifecycle {
    prevent_destroy = true
  }
}

resource "yandex_mdb_kafka_user" "this" {
  for_each = var.create_users ? local.user_clusters : {}

  cluster_id = each.value.cluster_id
  name       = each.value.user
  password   = random_password.kafka_user[each.key].result

  dynamic "permission" {
    for_each = try(local.user_permissions[each.key], [])
    content {
      topic_name = permission.value.topic_name
      role       = permission.value.role
    }
  }

  lifecycle {
    prevent_destroy = true
    ignore_changes  = [password]
  }

  depends_on = [yandex_mdb_kafka_topic.this]
}