locals {
  manifest = yamldecode(file(var.topic_manifest_path))

  raw_topics = try(local.manifest.kafka.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(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_clusters = {
    for item in distinct([
      for topic in values(local.topics) : "${topic.cluster_ref}:${topic.owner}"
      ]) : item => {
      cluster_ref = split(":", item)[0]
      owner       = split(":", item)[1]
      cluster_id  = var.kafka_cluster_refs[split(":", item)[0]].cluster_id
    }
  }

  owner_permissions = {
    for owner_key, owner in local.owner_clusters : owner_key => flatten([
      for topic in values(local.topics) : [
        for role in var.default_user_roles : {
          topic_name = topic.name
          role       = role
        }
      ]
      if topic.cluster_ref == owner.cluster_ref && topic.owner == owner.owner
    ])
  }
}

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

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

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.owner_clusters : {}

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

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

  lifecycle {
    prevent_destroy = true
  }

  depends_on = [yandex_mdb_kafka_topic.this]
}