locals {
  e2e_kafka_dependencies = concat(
    [
      {
        networkType        = "NetworkType_Public"
        authType           = "DISABLE"
        tlsEnabled         = false
        bootstrapEndpoints = "${aws_eip.public.public_ip}:${local.ports.kafka_public}"
        user               = ""
        password           = ""
      },
      {
        networkType        = "NetworkType_Public"
        authType           = "SASL_PLAIN"
        tlsEnabled         = false
        bootstrapEndpoints = "${aws_eip.public.public_ip}:${local.ports.kafka_sasl}"
        user               = local.kafka_sasl_username
        password           = local.kafka_sasl_password
      },
    ],
    [for dependency in [
      {
        networkType            = "NetworkType_Private_Link"
        authType               = "DISABLE"
        tlsEnabled             = false
        cloudProvider          = "aws"
        region                 = var.aws_region
        projectID              = var.e2e_project_ids
        kafkaID                = var.name_prefix
        isCustomDomain         = true
        rootDomainName         = var.kafka_privatelink_root_domain_name
        dnsRecords             = { (local.primary_az_id) = local.kafka_privatelink_zone_record }
        privateLinkServiceName = try(aws_vpc_endpoint_service.kafka[0].service_name, "")
        ports                  = [local.ports.kafka_pl_public]
        user                   = ""
        password               = ""
      },
      {
        networkType            = "NetworkType_Private_Link"
        authType               = "SASL_PLAIN"
        tlsEnabled             = false
        cloudProvider          = "aws"
        region                 = var.aws_region
        projectID              = var.e2e_project_ids
        kafkaID                = "${var.name_prefix}-sasl"
        isCustomDomain         = true
        rootDomainName         = var.kafka_privatelink_root_domain_name
        dnsRecords             = { (local.primary_az_id) = local.kafka_privatelink_zone_record }
        privateLinkServiceName = try(aws_vpc_endpoint_service.kafka[0].service_name, "")
        ports                  = [local.ports.kafka_pl_sasl]
        user                   = local.kafka_sasl_username
        password               = local.kafka_sasl_password
      },
    ] : dependency if local.privatelink_enabled],
  )

  e2e_mysql_dependencies = concat(
    [
      {
        networkType = "NetworkType_Public"
        endpoint    = "${aws_eip.public.public_ip}:${local.ports.mysql}"
        user        = var.mysql_user
        password    = local.mysql_password
      },
    ],
    [for dependency in [
      {
        networkType            = "NetworkType_Private_Link"
        cloudProvider          = "aws"
        region                 = var.aws_region
        privateLinkServiceName = try(aws_vpc_endpoint_service.kafka[0].service_name, "")
        ports                  = [local.ports.mysql]
        user                   = var.mysql_user
        password               = local.mysql_password
      },
    ] : dependency if local.privatelink_enabled],
  )
}

output "aws_profile" {
  description = "AWS profile used by the stack and helper scripts."
  value       = var.aws_profile
}

output "aws_region" {
  description = "AWS region used by the stack and helper scripts."
  value       = var.aws_region
}

output "deployment_name" {
  description = "Stable unique name generated once for this Terraform deployment and used by the usedby tag."
  value       = local.deployment_name
}

output "audit_tags" {
  description = "Common tracking and audit tags applied to taggable AWS resources."
  value       = local.common_tags
}

output "instance_id" {
  description = "Service EC2 instance ID."
  value       = aws_instance.this.id
}

output "desired_state" {
  description = "Terraform-managed EC2 state."
  value       = var.desired_state
}

output "public_endpoints" {
  description = "Public test service endpoints."
  value = merge({
    public_ip = aws_eip.public.public_ip
    mysql = {
      host     = aws_eip.public.public_ip
      port     = local.ports.mysql
      endpoint = "${aws_eip.public.public_ip}:${local.ports.mysql}"
    }
    kafka_anonymous = {
      host     = aws_eip.public.public_ip
      port     = local.ports.kafka_public
      endpoint = "${aws_eip.public.public_ip}:${local.ports.kafka_public}"
    }
    kafka_sasl_plain = {
      host     = aws_eip.public.public_ip
      port     = local.ports.kafka_sasl
      endpoint = "${aws_eip.public.public_ip}:${local.ports.kafka_sasl}"
      tls      = false
    }
    }, var.enable_pulsar ? {
    pulsar_binary = {
      host     = aws_eip.public.public_ip
      port     = local.ports.pulsar_binary
      endpoint = "pulsar://${aws_eip.public.public_ip}:${local.ports.pulsar_binary}"
    }
    pulsar_admin_http = {
      host     = aws_eip.public.public_ip
      port     = local.ports.pulsar_admin_http
      endpoint = "http://${aws_eip.public.public_ip}:${local.ports.pulsar_admin_http}"
    }
    } : {}, var.enable_schema_registry ? {
    schema_registry = {
      host     = aws_eip.public.public_ip
      port     = local.ports.schema_registry
      endpoint = "http://${aws_eip.public.public_ip}:${local.ports.schema_registry}"
    }
  } : {})
}

output "privatelink" {
  description = "Shared provider-side MySQL and Kafka PrivateLink service. Null while deeply stopped."
  value = {
    enabled = local.privatelink_enabled
    shared = local.privatelink_enabled ? {
      service_id   = aws_vpc_endpoint_service.kafka[0].id
      service_name = aws_vpc_endpoint_service.kafka[0].service_name
    } : null
    mysql = local.privatelink_enabled ? {
      service_id   = aws_vpc_endpoint_service.kafka[0].id
      service_name = aws_vpc_endpoint_service.kafka[0].service_name
      port         = local.ports.mysql
    } : null
    kafka = local.privatelink_enabled ? {
      service_id              = aws_vpc_endpoint_service.kafka[0].id
      service_name            = aws_vpc_endpoint_service.kafka[0].service_name
      anonymous_port          = local.ports.kafka_pl_public
      sasl_plain_port         = local.ports.kafka_pl_sasl
      advertised_host         = local.kafka_privatelink_advertised_host
      root_domain_name        = var.kafka_privatelink_root_domain_name
      consumer_dns_record     = local.kafka_privatelink_zone_record
      consumer_dns_record_key = local.primary_az_id
    } : null
    availability_zones = [
      for key, subnet in local.subnets : {
        name = subnet.availability_zone
        id   = subnet.availability_zone_id
      }
    ]
    allowed_principal_arns = sort(tolist(local.allowed_principal_arns))
  }
}

output "privatelink_service_ids" {
  description = "Endpoint service IDs used by the deep-sleep connection preflight."
  value = {
    shared = try(aws_vpc_endpoint_service.kafka[0].id, null)
  }
}

output "s3_storage" {
  description = "CDC, backup-export, cluster-export, and import S3 locations and IAM identities."
  value = {
    bucket_name               = aws_s3_bucket.test_data.id
    bucket_arn                = aws_s3_bucket.test_data.arn
    region                    = var.aws_region
    cdc_uri                   = local.s3_cdc_uri
    backup_export_uri         = local.s3_backup_export_uri
    cluster_export_uri        = local.s3_cluster_export_uri
    import_uri                = local.s3_import_uri
    essential_export_role_arn = aws_iam_role.s3_access.arn
    role_arn                  = aws_iam_role.s3_access.arn
    role_name                 = aws_iam_role.s3_access.name
    role_trusted_principal_arns = sort(distinct(concat(
      tolist(local.s3_role_trusted_principal_arns),
      [for grant in values(var.s3_assume_role_grants) : "arn:aws:iam::${grant.account_id}:root"],
    )))
    access_key_user_arn  = aws_iam_user.s3_access.arn
    access_key_user_name = aws_iam_user.s3_access.name
  }
}

output "service_credentials" {
  description = "Test credentials. These values also exist in local state and EC2 user data."
  sensitive   = true
  value = {
    mysql = {
      root_password = local.mysql_root_password
      user          = var.mysql_user
      password      = local.mysql_password
    }
    kafka = {
      username = local.kafka_sasl_username
      password = local.kafka_sasl_password
    }
    schema_registry = {
      username = local.schema_registry_username
      password = local.schema_registry_password
    }
    pulsar = {
      token_secret = local.pulsar_token_secret
      proxy_token  = data.external.pulsar_proxy_token.result.token
      admin_token  = data.external.pulsar_admin_token.result.token
      client_token = data.external.pulsar_client_token.result.token
    }
    s3 = {
      access_key_id     = aws_iam_access_key.s3_access.id
      secret_access_key = aws_iam_access_key.s3_access.secret
      role_external_id  = var.s3_role_external_id
    }
  }
}

output "e2e_dependency_yaml" {
  description = "Sensitive YAML fragment for dataflow-service test/e2e dependency configuration."
  sensitive   = true
  value = yamlencode(merge({
    kafka = local.e2e_kafka_dependencies
    mysql = local.e2e_mysql_dependencies
    s3    = local.e2e_s3_dependencies
    backupExport = {
      aws = {
        targetURI = local.s3_backup_export_uri
        region    = var.aws_region
      }
    }
    essentialExport = {
      aws = {
        targetURI = local.s3_cluster_export_uri
        region    = var.aws_region
        roleARN   = aws_iam_role.s3_access.arn
      }
    }
    }, var.enable_pulsar ? {
    pulsar = [{
      networkType = "NetworkType_Public"
      authType    = "PULSAR_AUTH_TYPE_TOKEN"
      token       = data.external.pulsar_client_token.result.token
      protocol    = "PULSAR_PROTOCOL_PULSAR"
      endpoint    = "${aws_eip.public.public_ip}:${local.ports.pulsar_binary}"
      tenant      = "public"
      namespace   = "default"
      topic       = "date-integration-smoke"
    }]
    } : {}, var.enable_schema_registry ? {
    schemaRegistry = {
      username  = local.schema_registry_username
      password  = local.schema_registry_password
      endpoints = "http://${aws_eip.public.public_ip}:${local.ports.schema_registry}"
    }
  } : {}))
}