locals {
  k8s_secret_type_map = {
    "dockerconfigjson" = "kubernetes.io/dockerconfigjson"
    "database"         = "Opaque"
    "database_url"     = "Opaque"
    "s3"               = "Opaque"
    "valkey"           = "Opaque"
    "yc_sa"            = "Opaque"
    "rabbitmq"         = "Opaque"
    "kafka"            = "Opaque"
    "opaque"           = "Opaque"
  }

  default_yc_sa_key_names = {
    access_key         = "access_key"
    secret_key         = "secret_key"
    service_account_id = "service_account_id"
  }

  secrets_map = {
    for secret in var.secrets : "${secret.namespace}/${secret.name}" => secret
  }

  random_keys_flat = merge([
    for secret_name, secret in local.secrets_map : {
      for key, config in try(secret.random_keys, {}) : "${secret_name}:${key}" => {
        secret_name = secret_name
        key         = key
        length      = try(config.length, 32)
        special     = try(config.special, false)
      }
    }
  ]...)

  database_outputs_by_secret = {
    for name, secret in local.secrets_map : name => try(
      var.database_outputs_map["${try(secret.depends_on.cluster, "")}:${try(secret.depends_on.db, "")}:${try(secret.depends_on.user, "")}"],
      null
    )
  }

  valkey_dependency_keys = {
    for name, secret in local.secrets_map : name => {
      cluster = try(secret.depends_on.valkey_cluster, "") != "" ? try(secret.depends_on.valkey_cluster, "") : try(secret.depends_on.cluster, "")
      user    = try(secret.depends_on.valkey_user, "") != "" ? try(secret.depends_on.valkey_user, "") : try(secret.depends_on.user, "")
    }
  }

  valkey_outputs_by_secret = {
    for name, secret in local.secrets_map : name => try(
      var.valkey_outputs_map["${local.valkey_dependency_keys[name].cluster}:${local.valkey_dependency_keys[name].user}"],
      null
    )
  }

  rabbitmq_outputs_by_secret = {
    for name, secret in local.secrets_map : name => try(
      var.rabbitmq_outputs_map[
        format(
          "%s:%s:%s",
          try(secret.depends_on.rabbitmq_ref, "default"),
          try(secret.depends_on.rabbitmq_vhost, try(secret.depends_on.vhost, "")),
          try(secret.depends_on.rabbitmq_user, try(secret.depends_on.user, ""))
        )
      ],
      null
    )
  }

  kafka_outputs_by_secret = {
    for name, secret in local.secrets_map : name => try(
      var.kafka_outputs_map[
        format(
          "%s:%s",
          try(coalesce(secret.depends_on.kafka_ref, secret.depends_on.cluster_ref, secret.depends_on.cluster), ""),
          try(coalesce(secret.depends_on.kafka_user, secret.depends_on.user), "")
        )
      ],
      null
    )
  }

  rabbitmq_default_credentials_by_secret = {
    for name, secret in local.secrets_map : name => local.rabbitmq_outputs_by_secret[name] == null ? {} : {
      host                = local.rabbitmq_outputs_by_secret[name].host
      hostname            = "${local.rabbitmq_outputs_by_secret[name].host}:${local.rabbitmq_outputs_by_secret[name].port}"
      port                = local.rabbitmq_outputs_by_secret[name].port
      vhost               = local.rabbitmq_outputs_by_secret[name].vhost
      user                = local.rabbitmq_outputs_by_secret[name].username
      username            = local.rabbitmq_outputs_by_secret[name].username
      password            = local.rabbitmq_outputs_by_secret[name].password
      uri                 = local.rabbitmq_outputs_by_secret[name].uri
      management_endpoint = local.rabbitmq_outputs_by_secret[name].management_endpoint
    }
    if secret.type == "rabbitmq"
  }

  rabbitmq_credentials_by_secret = {
    for name, secret in local.secrets_map : name => length(try(secret.credential_keys, {})) == 0 ? local.rabbitmq_default_credentials_by_secret[name] : {
      for key, source in try(secret.credential_keys, {}) : key => try(local.rabbitmq_default_credentials_by_secret[name][source], "")
    }
    if secret.type == "rabbitmq" && local.rabbitmq_outputs_by_secret[name] != null
  }

  kafka_default_credentials_by_secret = {
    for name, secret in local.secrets_map : name => local.kafka_outputs_by_secret[name] == null ? {} : {
      host                   = local.kafka_outputs_by_secret[name].host
      hostname               = local.kafka_outputs_by_secret[name].hostname
      port                   = local.kafka_outputs_by_secret[name].port
      username               = local.kafka_outputs_by_secret[name].username
      user                   = local.kafka_outputs_by_secret[name].username
      password               = local.kafka_outputs_by_secret[name].password
      sasl_mechanism         = local.kafka_outputs_by_secret[name].sasl_mechanism
      security_protocol      = local.kafka_outputs_by_secret[name].security_protocol
      bootstrap_server       = local.kafka_outputs_by_secret[name].bootstrap_server
      bootstrap_servers      = local.kafka_outputs_by_secret[name].bootstrap_servers
      bootstrap_servers_json = local.kafka_outputs_by_secret[name].bootstrap_servers_json
    }
    if secret.type == "kafka"
  }

  kafka_credentials_by_secret = {
    for name, secret in local.secrets_map : name => length(try(secret.credential_keys, {})) == 0 ? local.kafka_default_credentials_by_secret[name] : {
      for key, source in try(secret.credential_keys, {}) : key => try(local.kafka_default_credentials_by_secret[name][source], "")
    }
    if secret.type == "kafka" && local.kafka_outputs_by_secret[name] != null
  }

  yc_sa_sources = {
    for name, secret in local.secrets_map : name => try(
      var.yc_service_accounts_map[try(secret.depends_on.service_account, "")],
      null
    )
    if secret.type == "yc_sa"
  }

  yc_sa_key_names_by_secret = {
    for name, secret in local.secrets_map : name => merge(
      local.default_yc_sa_key_names,
      try(secret.yc_sa_key_names, {})
    )
    if secret.type == "yc_sa"
  }

  secrets_data_override = {
    for name, secret in local.secrets_map : name => {
      for key, b64 in try(secret.data, {}) : key => base64decode(b64)
    }
    if length(try(secret.data, {})) > 0
  }

  secrets_data_computed = {
    for name, secret in local.secrets_map : name =>
    secret.type == "dockerconfigjson" ? {
      ".dockerconfigjson" = jsonencode({
        auths = {
          "${secret.registry_url}" = {
            username = try(var.env_vars.DOCKER_REGISTRY_USERNAME, "")
            password = try(var.env_vars.DOCKER_REGISTRY_PASSWORD, "")
            auth     = base64encode("${try(var.env_vars.DOCKER_REGISTRY_USERNAME, "")}:${try(var.env_vars.DOCKER_REGISTRY_PASSWORD, "")}")
          }
        }
      })
      } : secret.type == "s3" ? {
      access_key = try(var.s3_buckets_map[secret.depends_on.bucket].access_key, var.s3_outputs.access_key)
      secret_key = try(var.s3_buckets_map[secret.depends_on.bucket].secret_key, var.s3_outputs.secret_key)
      bucket     = try(var.s3_buckets_map[secret.depends_on.bucket].bucket_name, var.s3_outputs.bucket_name)
      endpoint   = try(var.constants.s3_endpoint, "https://storage.yandexcloud.net")
      } : secret.type == "yc_sa" ? merge(
      {
        (local.yc_sa_key_names_by_secret[name].access_key)         = try(local.yc_sa_sources[name].access_key, "")
        (local.yc_sa_key_names_by_secret[name].secret_key)         = try(local.yc_sa_sources[name].secret_key, "")
        (local.yc_sa_key_names_by_secret[name].service_account_id) = try(local.yc_sa_sources[name].service_account_id, "")
        endpoint                                                   = "https://storage.yandexcloud.net"
      },
      try(secret.custom_keys, {}),
      {
        for key, constant_name in try(secret.constant_keys, {}) : key => try(var.constants[constant_name], "")
      }
      ) : secret.type == "database" && local.database_outputs_by_secret[name] != null ? merge(
      {
        host     = local.database_outputs_by_secret[name].host
        port     = try(var.constants.postgres_port, "6432")
        database = local.database_outputs_by_secret[name].database_name
        username = local.database_outputs_by_secret[name].user_name
        password = local.database_outputs_by_secret[name].password
        "ca.crt" = try(var.constants.postgres_ca, "")
      },
      try(secret.custom_keys, {}),
      {
        for key, constant_name in try(secret.constant_keys, {}) : key => try(var.constants[constant_name], "")
      }
      ) : secret.type == "database_url" && local.database_outputs_by_secret[name] != null ? merge(
      {
        database_url = format(
          "postgresql+asyncpg://%s:%s@%s:%s/%s",
          local.database_outputs_by_secret[name].user_name,
          local.database_outputs_by_secret[name].password,
          local.database_outputs_by_secret[name].host,
          try(var.constants.postgres_port, "6432"),
          local.database_outputs_by_secret[name].database_name
        )
      },
      try(secret.custom_keys, {}),
      {
        for key, constant_name in try(secret.constant_keys, {}) : key => try(var.constants[constant_name], "")
      }
      ) : secret.type == "valkey" && local.valkey_outputs_by_secret[name] != null ? merge(
      {
        cert     = try(var.constants.valkey_ca, try(var.constants.postgres_ca, ""))
        host     = local.valkey_outputs_by_secret[name].host
        login    = local.valkey_outputs_by_secret[name].user_name
        password = local.valkey_outputs_by_secret[name].password
        port     = tostring(try(local.valkey_outputs_by_secret[name].port, try(var.constants.valkey_port, "6380")))
        url      = format("%s:%s", local.valkey_outputs_by_secret[name].host, tostring(try(local.valkey_outputs_by_secret[name].port, try(var.constants.valkey_port, "6380"))))
      },
      try(secret.custom_keys, {}),
      {
        for key, constant_name in try(secret.constant_keys, {}) : key => try(var.constants[constant_name], "")
      }
      ) : secret.type == "rabbitmq" && local.rabbitmq_outputs_by_secret[name] != null ? merge(
      local.rabbitmq_credentials_by_secret[name],
      try(secret.custom_keys, {}),
      {
        for key, constant_name in try(secret.constant_keys, {}) : key => try(var.constants[constant_name], "")
      }
      ) : secret.type == "kafka" && local.kafka_outputs_by_secret[name] != null ? merge(
      local.kafka_credentials_by_secret[name],
      try(secret.custom_keys, {}),
      {
        for key, constant_name in try(secret.constant_keys, {}) : key => try(var.constants[constant_name], "")
      }
      ) : merge(
      try(secret.custom_keys, {}),
      {
        for key, constant_name in try(secret.constant_keys, {}) : key => try(var.constants[constant_name], "")
      }
    )
  }

  known_ownerships = ["", "managed", "referenced", "observed", "adopt_once"]
  known_schemas = [
    "", "postgresql.v1", "kafka.v1", "rabbitmq.v1", "s3.v1", "valkey.v1",
    "dockerconfigjson.v1", "opaque.v1",
  ]

  # extra_fields flattened to "secret_name::field_key" => {source_kind, ref, value}
  extra_fields_flat = merge([
    for name, secret in local.secrets_map : {
      for key, ef in try(secret.extra_fields, {}) : "${name}::${key}" => {
        source_kind = ef.source_kind
        ref         = try(ef.ref, "")
        value       = try(ef.value, "")
      }
    }
  ]...)

  # Every vault_kv_secret_v2 read this module needs to perform for referenced
  # ownership and extra_fields secret_ref, keyed uniformly so a single
  # data-source for_each can serve both. ref format: "vault://<kv-name>#<field>".
  vault_secret_ref_targets = merge(
    {
      for name, secret in local.secrets_map : "secret:${name}" => {
        path = trimprefix(split("#", secret.source_ref)[0], "vault://")
        key  = length(split("#", secret.source_ref)) > 1 ? split("#", secret.source_ref)[1] : ""
      }
      if try(secret.ownership, "") == "referenced" && try(secret.source_kind, "") == "secret_ref" && try(secret.source_ref, "") != ""
    },
    {
      for k, ef in local.extra_fields_flat : "extra:${k}" => {
        path = trimprefix(split("#", ef.ref)[0], "vault://")
        key  = length(split("#", ef.ref)) > 1 ? split("#", ef.ref)[1] : ""
      }
      if ef.source_kind == "secret_ref" && ef.ref != ""
    }
  )

  # Resolved data for ownership=referenced secrets. dockerconfigjson.v1 is a
  # single opaque JSON blob under vault_secret_ref_targets[...].key - it lands
  # in the ".dockerconfigjson" field verbatim. Anything else with a key reads
  # that one field; without a key, the whole KV object is copied through.
  secrets_data_referenced = {
    for name, secret in local.secrets_map : name => (
      try(secret.schema, "") == "dockerconfigjson.v1"
      ? { ".dockerconfigjson" = try(data.vault_kv_secret_v2.referenced["secret:${name}"].data[local.vault_secret_ref_targets["secret:${name}"].key], "") }
      : (
        local.vault_secret_ref_targets["secret:${name}"].key != ""
        ? { (local.vault_secret_ref_targets["secret:${name}"].key) = try(data.vault_kv_secret_v2.referenced["secret:${name}"].data[local.vault_secret_ref_targets["secret:${name}"].key], "") }
        : { for k, v in try(data.vault_kv_secret_v2.referenced["secret:${name}"].data, {}) : k => tostring(v) }
      )
    )
    if try(secret.ownership, "") == "referenced" && try(secret.source_kind, "") == "secret_ref" && try(secret.source_ref, "") != ""
  }

  extra_fields_resolved = {
    for name, secret in local.secrets_map : name => {
      for key, ef in try(secret.extra_fields, {}) : key => (
        ef.source_kind == "secret_ref" && ef.ref != ""
        ? try(data.vault_kv_secret_v2.referenced["extra:${name}::${key}"].data[local.vault_secret_ref_targets["extra:${name}::${key}"].key], "")
        : try(ef.value, "")
      )
    }
  }

  # Base data before schema field-naming and extra_fields: referenced value
  # if this secret is ownership=referenced, otherwise the existing
  # type-driven computation (unchanged for every legacy secret).
  secrets_data_base = {
    for name, secret in local.secrets_map : name => (
      contains(keys(local.secrets_data_referenced), name)
      ? local.secrets_data_referenced[name]
      : local.secrets_data_computed[name]
    )
  }

  # rabbitmq.v1 differs from the legacy field-set only in "hostname": legacy
  # ships host:port there, the canonical schema wants host alone.
  secrets_data_schema_adjusted = {
    for name, secret in local.secrets_map : name => (
      try(secret.schema, "") == "rabbitmq.v1" && local.rabbitmq_outputs_by_secret[name] != null
      ? merge(local.secrets_data_base[name], { hostname = local.rabbitmq_outputs_by_secret[name].host })
      : local.secrets_data_base[name]
    )
  }

  secrets_contract_errors = {
    for name, secret in local.secrets_map : name => (
      length(try(secret.data, {})) > 0 ? "" :
      !contains(local.known_ownerships, try(secret.ownership, "")) ?
      "unknown ownership '${try(secret.ownership, "")}' (expected one of: ${join(", ", local.known_ownerships)})" :
      !contains(local.known_schemas, try(secret.schema, "")) ?
      "unknown schema '${try(secret.schema, "")}' (expected one of: ${join(", ", local.known_schemas)})" :
      length([for k, ef in try(secret.extra_fields, {}) : k if !contains(["secret_ref", "static"], ef.source_kind)]) > 0 ?
      "extra_fields has unknown source_kind (expected secret_ref|static)" :
      length([for k, ef in try(secret.extra_fields, {}) : k if ef.source_kind == "secret_ref" && try(ef.ref, "") == ""]) > 0 ?
      "extra_fields secret_ref entries require non-empty ref (vault://<path>#<field>)" :
      try(secret.ownership, "") == "observed" ? "" :
      try(secret.ownership, "") == "referenced" ? (
        try(secret.source_kind, "") != "secret_ref" ? "ownership=referenced requires source_kind=secret_ref" :
        !can(regex("^vault://.+", try(secret.source_ref, ""))) ? "ownership=referenced requires source_ref matching vault://<path>#<field>" :
        ""
      ) :
      secret.type == "database" || secret.type == "database_url" ? (
        try(secret.depends_on.cluster, "") == "" || try(secret.depends_on.db, "") == "" || try(secret.depends_on.user, "") == ""
        ? "type=${secret.type} requires depends_on.cluster, depends_on.db, depends_on.user (got cluster=${try(secret.depends_on.cluster, "")}, db=${try(secret.depends_on.db, "")}, user=${try(secret.depends_on.user, "")})"
        : ""
      ) :
      secret.type == "valkey" ? (
        local.valkey_dependency_keys[name].cluster == "" || local.valkey_dependency_keys[name].user == ""
        ? "type=valkey requires depends_on.cluster (or valkey_cluster) and depends_on.user (or valkey_user)"
        : ""
      ) :
      secret.type == "rabbitmq" ? (
        try(secret.depends_on.rabbitmq_vhost, try(secret.depends_on.vhost, "")) == "" || try(secret.depends_on.rabbitmq_user, try(secret.depends_on.user, "")) == ""
        ? "type=rabbitmq requires depends_on.vhost (or rabbitmq_vhost) and depends_on.user (or rabbitmq_user)"
        : ""
      ) :
      secret.type == "kafka" ? (
        try(coalesce(secret.depends_on.kafka_ref, secret.depends_on.cluster_ref, secret.depends_on.cluster), "") == "" || try(coalesce(secret.depends_on.kafka_user, secret.depends_on.user), "") == ""
        ? "type=kafka requires depends_on.kafka_ref (or cluster_ref/cluster) and depends_on.kafka_user (or user)"
        : ""
      ) :
      secret.type == "yc_sa" ? (
        try(secret.depends_on.service_account, "") == ""
        ? "type=yc_sa requires depends_on.service_account"
        : ""
      ) :
      secret.type == "dockerconfigjson" && !try(secret.ignore_changes, false) ? (
        trimspace(try(var.env_vars.DOCKER_REGISTRY_USERNAME, "")) == "" || trimspace(try(var.env_vars.DOCKER_REGISTRY_PASSWORD, "")) == ""
        ? "type=dockerconfigjson requires module-level env_vars.DOCKER_REGISTRY_USERNAME and env_vars.DOCKER_REGISTRY_PASSWORD to be non-empty"
        : ""
      ) :
      ""
    )
  }

  secrets_data = {
    for name, secret in local.secrets_map : name => merge(
      contains(keys(local.secrets_data_override), name)
      ? local.secrets_data_override[name]
      : local.secrets_data_schema_adjusted[name],
      try(local.extra_fields_resolved[name], {})
    )
  }

  # ownership supersedes legacy ignore_changes when set: adopt_once behaves
  # like the old ignore_changes=true, managed/referenced behave like false.
  # observed doesn't create a write resource at all (see secrets_observed_*).
  secrets_ignore_changes = {
    for name, secret in local.secrets_map : name => (
      try(secret.ownership, "") == "adopt_once" ? true :
      try(secret.ownership, "") == "" ? try(secret.ignore_changes, false) :
      false
    )
  }

  secrets_with_data = {
    for name, secret in local.secrets_map : name => {
      name           = secret.name
      namespace      = secret.namespace
      type           = secret.type
      k8s_type       = try(local.k8s_secret_type_map[secret.type], "Opaque")
      labels         = try(secret.labels, {})
      annotations    = try(secret.annotations, {})
      ignore_changes = local.secrets_ignore_changes[name]
      ownership      = try(secret.ownership, "")
      sink           = try(secret.sink, "k8s")
      vault_path     = try(secret.vault_path, "") != "" ? secret.vault_path : "${var.vault_path_prefix}/${secret.name}"
      data           = try(local.secrets_data[name], {})
      random_keys = contains(keys(local.secrets_data_override), name) ? {} : try(secret.random_keys, {})
    }
  }

  secrets_with_ignore = {
    for name, secret in local.secrets_with_data : name => secret
    if secret.sink == "k8s" && secret.ignore_changes && secret.ownership != "observed"
  }
  secrets_without_ignore = {
    for name, secret in local.secrets_with_data : name => secret
    if secret.sink == "k8s" && !secret.ignore_changes && secret.ownership != "observed"
  }

  secrets_vault = {
    for name, secret in local.secrets_with_data : name => secret
    if secret.sink == "vault" && secret.ownership != "observed"
  }

  secrets_observed_k8s = {
    for name, secret in local.secrets_with_data : name => secret
    if secret.sink == "k8s" && secret.ownership == "observed"
  }
  secrets_observed_vault = {
    for name, secret in local.secrets_with_data : name => secret
    if secret.sink == "vault" && secret.ownership == "observed"
  }
}

data "vault_kv_secret_v2" "referenced" {
  for_each = local.vault_secret_ref_targets

  mount = var.vault_kv_mount
  name  = each.value.path
}

data "kubernetes_secret" "observed" {
  for_each = local.secrets_observed_k8s

  metadata {
    name      = each.value.name
    namespace = each.value.namespace
  }
}

data "vault_kv_secret_v2" "observed" {
  for_each = local.secrets_observed_vault

  mount = var.vault_kv_mount
  name  = each.value.vault_path
}

resource "random_password" "secrets" {
  for_each = local.random_keys_flat

  length  = each.value.length
  special = each.value.special
  upper   = true
  lower   = true
  numeric = true

  lifecycle {
    ignore_changes = all
  }
}

resource "kubernetes_secret" "with_ignore" {
  for_each = local.secrets_with_ignore

  metadata {
    name        = each.value.name
    namespace   = each.value.namespace
    labels      = try(each.value.labels, {})
    annotations = try(each.value.annotations, {})
  }

  type = each.value.k8s_type

  data = merge(each.value.data, {
    for key, config in each.value.random_keys : key => random_password.secrets["${each.key}:${key}"].result
  })

  lifecycle {
    ignore_changes = [data, type, metadata, wait_for_service_account_token]
    precondition {
      condition     = local.secrets_contract_errors[each.key] == ""
      error_message = "Secret contract violation for ${each.key}: ${local.secrets_contract_errors[each.key]}"
    }
  }
}

resource "kubernetes_secret" "without_ignore" {
  for_each = local.secrets_without_ignore

  metadata {
    name        = each.value.name
    namespace   = each.value.namespace
    labels      = try(each.value.labels, {})
    annotations = try(each.value.annotations, {})
  }

  type = each.value.k8s_type

  data = merge(each.value.data, {
    for key, config in each.value.random_keys : key => random_password.secrets["${each.key}:${key}"].result
  })

  lifecycle {
    ignore_changes = [type, metadata, wait_for_service_account_token]
    precondition {
      condition     = local.secrets_contract_errors[each.key] == ""
      error_message = "Secret contract violation for ${each.key}: ${local.secrets_contract_errors[each.key]}"
    }
  }
}

resource "vault_kv_secret_v2" "this" {
  for_each = local.secrets_vault

  mount = var.vault_kv_mount
  name  = each.value.vault_path
  data_json = jsonencode(merge(each.value.data, {
    for key, config in each.value.random_keys : key => random_password.secrets["${each.key}:${key}"].result
  }))

  lifecycle {
    precondition {
      condition     = local.secrets_contract_errors[each.key] == ""
      error_message = "Secret contract violation for ${each.key}: ${local.secrets_contract_errors[each.key]}"
    }
  }
}