resource "aws_lb_target_group" "mysql" {
  count = local.privatelink_enabled ? 1 : 0

  name        = "${var.name_prefix}-mysql"
  port        = local.ports.mysql
  protocol    = "TCP"
  target_type = "instance"
  vpc_id      = aws_vpc.this.id
  tags        = local.deployment_tags

  health_check {
    enabled             = true
    healthy_threshold   = 2
    interval            = 10
    protocol            = "TCP"
    unhealthy_threshold = 2
  }
}

resource "aws_lb_target_group_attachment" "mysql" {
  count = local.privatelink_enabled ? 1 : 0

  target_group_arn = aws_lb_target_group.mysql[0].arn
  target_id        = aws_instance.this.id
  port             = local.ports.mysql
}

resource "aws_lb_listener" "mysql" {
  count = local.privatelink_enabled ? 1 : 0

  load_balancer_arn = aws_lb.kafka[0].arn
  port              = local.ports.mysql
  protocol          = "TCP"
  tags              = local.deployment_tags

  default_action {
    type             = "forward"
    target_group_arn = aws_lb_target_group.mysql[0].arn
  }
}

resource "aws_lb" "kafka" {
  count = local.privatelink_enabled ? 1 : 0

  name                                                         = "${var.name_prefix}-kafka"
  internal                                                     = true
  load_balancer_type                                           = "network"
  subnets                                                      = [for subnet in aws_subnet.public : subnet.id]
  security_groups                                              = [aws_security_group.kafka_nlb[0].id]
  enable_cross_zone_load_balancing                             = true
  enforce_security_group_inbound_rules_on_private_link_traffic = "off"

  tags = merge(local.deployment_tags, {
    Name = "${var.name_prefix}-mysql-kafka"
  })
}

resource "aws_lb_target_group" "kafka_public" {
  count = local.privatelink_enabled ? 1 : 0

  name        = "${var.name_prefix}-kafka-pub"
  port        = local.ports.kafka_pl_public
  protocol    = "TCP"
  target_type = "instance"
  vpc_id      = aws_vpc.this.id
  tags        = local.deployment_tags

  health_check {
    enabled             = true
    healthy_threshold   = 2
    interval            = 10
    protocol            = "TCP"
    unhealthy_threshold = 2
  }
}

resource "aws_lb_target_group" "kafka_sasl" {
  count = local.privatelink_enabled ? 1 : 0

  name        = "${var.name_prefix}-kafka-auth"
  port        = local.ports.kafka_pl_sasl
  protocol    = "TCP"
  target_type = "instance"
  vpc_id      = aws_vpc.this.id
  tags        = local.deployment_tags

  health_check {
    enabled             = true
    healthy_threshold   = 2
    interval            = 10
    protocol            = "TCP"
    unhealthy_threshold = 2
  }
}

resource "aws_lb_target_group_attachment" "kafka_public" {
  count = local.privatelink_enabled ? 1 : 0

  target_group_arn = aws_lb_target_group.kafka_public[0].arn
  target_id        = aws_instance.this.id
  port             = local.ports.kafka_pl_public
}

resource "aws_lb_target_group_attachment" "kafka_sasl" {
  count = local.privatelink_enabled ? 1 : 0

  target_group_arn = aws_lb_target_group.kafka_sasl[0].arn
  target_id        = aws_instance.this.id
  port             = local.ports.kafka_pl_sasl
}

resource "aws_lb_listener" "kafka_public" {
  count = local.privatelink_enabled ? 1 : 0

  load_balancer_arn = aws_lb.kafka[0].arn
  port              = local.ports.kafka_pl_public
  protocol          = "TCP"
  tags              = local.deployment_tags

  default_action {
    type             = "forward"
    target_group_arn = aws_lb_target_group.kafka_public[0].arn
  }
}

resource "aws_lb_listener" "kafka_sasl" {
  count = local.privatelink_enabled ? 1 : 0

  load_balancer_arn = aws_lb.kafka[0].arn
  port              = local.ports.kafka_pl_sasl
  protocol          = "TCP"
  tags              = local.deployment_tags

  default_action {
    type             = "forward"
    target_group_arn = aws_lb_target_group.kafka_sasl[0].arn
  }
}

resource "aws_vpc_endpoint_service" "kafka" {
  count = local.privatelink_enabled ? 1 : 0

  acceptance_required        = false
  network_load_balancer_arns = [aws_lb.kafka[0].arn]

  tags = merge(local.deployment_tags, {
    Name = "${var.name_prefix}-mysql-kafka"
  })
}

# The AWS provider exposes Endpoint Service availability_zones as computed-only.
# Reconcile the producer service after its NLB has enabled every regional AZ.
resource "terraform_data" "endpoint_service_availability_zones" {
  count = local.privatelink_enabled ? 1 : 0

  triggers_replace = [
    aws_vpc_endpoint_service.kafka[0].id,
    join(",", sort(local.availability_zones)),
  ]

  provisioner "local-exec" {
    command     = "${path.module}/../../scripts/sync-aws-endpoint-service-zones.sh"
    interpreter = ["/usr/bin/env", "bash", "-c"]
    environment = {
      AWS_PROFILE                     = var.aws_profile
      AWS_REGION                      = var.aws_region
      ENDPOINT_SERVICE_ID             = aws_vpc_endpoint_service.kafka[0].id
      DESIRED_AVAILABILITY_ZONE_NAMES = join(" ", local.availability_zones)
    }
  }

  depends_on = [aws_lb.kafka]
}

resource "aws_vpc_endpoint_service_allowed_principal" "kafka" {
  for_each = local.privatelink_enabled ? local.allowed_principal_arns : toset([])

  vpc_endpoint_service_id = aws_vpc_endpoint_service.kafka[0].id
  principal_arn           = each.value
}