package privatelinkservice

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"sync/atomic"
	"testing"

	awssdk "github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/ec2"
	"github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"go.uber.org/mock/gomock"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"sigs.k8s.io/controller-runtime/pkg/client"

	regionalapiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	regionalaws "github.com/tidbcloud/infra-provider/pkg/cloud/aws"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/clustercli"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/privatelinkgroup"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	respls "github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/privatelinkservice"
)

func TestSetPrivateLinkIPAddressTypeAdvancesPLSAfterGroupNLBIsReady(t *testing.T) {
	type addressUpdate struct {
		RequestedIPAddressType string `json:"requestedIpAddressType"`
		NLBIPAddressType       string `json:"nlbIpAddressType"`
	}
	var updates []addressUpdate
	var nlbDualstack atomic.Bool
	var groupMutationCalls atomic.Int32
	var reader client.Client
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		groupMutationCalls.Add(1)
		var update addressUpdate
		require.NoError(t, json.NewDecoder(r.Body).Decode(&update))
		updates = append(updates, update)
		cluster := &cloudv1.Cluster{}
		require.NoError(t, reader.Get(r.Context(), client.ObjectKey{
			Name: "cluster", Namespace: groupAddressTestNamespace,
		}, cluster))
		if update.RequestedIPAddressType == "PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK" {
			cluster.Spec.SqlEndpoints.PrivateLinkIPAddressType = cloudv1.PrivateLinkIPAddressTypeDualstack
		}
		if update.NLBIPAddressType == "PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK" {
			cluster.Spec.SqlEndpoints.PrivateLinkNLBIPAddressType = cloudv1.PrivateLinkIPAddressTypeDualstack
			nlbDualstack.Store(true)
		}
		require.NoError(t, reader.Update(r.Context(), cluster))
		w.WriteHeader(http.StatusOK)
	}))
	defer server.Close()
	awsServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		addressType := "ipv4"
		if nlbDualstack.Load() {
			addressType = "dualstack"
		}
		if groupAddressTestInfrastructureResponse(t, w, r, addressType) {
			return
		}
		if r.Form.Get("Action") == "DescribeVpcEndpointServiceConfigurations" {
			_, _ = w.Write([]byte(`<DescribeVpcEndpointServiceConfigurationsResponse>
<serviceConfigurationSet><item><serviceId>vpce-svc-1</serviceId><serviceState>Available</serviceState>
<supportedIpAddressTypeSet><item>ipv4</item></supportedIpAddressTypeSet>
<networkLoadBalancerArnSet><item>arn-1</item></networkLoadBalancerArnSet></item></serviceConfigurationSet>
</DescribeVpcEndpointServiceConfigurationsResponse>`))
			return
		}
		http.Error(w, "unexpected action "+r.Form.Get("Action"), http.StatusBadRequest)
	}))
	t.Cleanup(awsServer.Close)
	reader = groupAddressTestClient(t,
		groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeIPv4),
		groupAddressTestAPPRoute(),
		groupAddressTestPLS(groupAddressTestGroupID),
		groupAddressTestVPC(),
	)
	service := &PrivateLinkServiceService{
		cli:                  reader,
		privatelinkservicerw: respls.NewPrivateLinkServiceReaderWriter(reader),
		config:               privateLinkGroupTestConfig(),
		groupClusterClient:   clustercli.NewClusterCli(server.URL),
	}
	awsCli := groupAddressTestAWSClient(t, awsServer)
	service.awsCli = awsCli
	service.groupManager = privatelinkgroup.NewManager(reader, awsCli)

	err := service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK,
	))

	require.NoError(t, err)
	require.Len(t, updates, 1)
	assert.Equal(t, "PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK", updates[0].RequestedIPAddressType)
	assert.Equal(t, "PRIVATE_LINK_IP_ADDRESS_TYPE_UNSPECIFIED", updates[0].NLBIPAddressType)
	persisted := &v1alpha1.PrivateLinkService{}
	require.NoError(t, reader.Get(context.Background(), client.ObjectKey{
		Name: groupAddressTestGroupID, Namespace: groupAddressTestNamespace,
	}, persisted))
	assert.Empty(t, persisted.Spec.TargetIPAddressType, "the first request advances only the Group infrastructure")

	require.NoError(t, service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK,
	)))
	require.Len(t, updates, 2)
	assert.Equal(t, "PRIVATE_LINK_IP_ADDRESS_TYPE_UNSPECIFIED", updates[1].RequestedIPAddressType)
	assert.Equal(t, "PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK", updates[1].NLBIPAddressType)
	require.NoError(t, reader.Get(context.Background(), client.ObjectKey{
		Name: groupAddressTestGroupID, Namespace: groupAddressTestNamespace,
	}, persisted))
	assert.Empty(t, persisted.Spec.TargetIPAddressType)

	require.NoError(t, service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK,
	)))
	require.NoError(t, reader.Get(context.Background(), client.ObjectKey{
		Name: groupAddressTestGroupID, Namespace: groupAddressTestNamespace,
	}, persisted))
	assert.Equal(t, v1alpha1.IPAddressTypeDualstack, persisted.Spec.TargetIPAddressType)

	require.NoError(t, service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK,
	)))
	assert.Equal(t, int32(2), groupMutationCalls.Load(), "replaying the same target must not repeat the Group mutation")
}

func TestAdvanceGroupToIPv4KeepsEmptyTargetWhenAWSIsAlreadyIPv4(t *testing.T) {
	pls := groupAddressTestPLS(groupAddressTestGroupID)
	service := &PrivateLinkServiceService{}
	state := &privateLinkIPAddressState{
		infrastructure: privatelinkgroup.Evaluation{
			Requested: v1alpha1.IPAddressTypeIPv4,
			NLBTarget: v1alpha1.IPAddressTypeIPv4,
		},
		normal:    &normalPrivateLinkService{pls: pls},
		plsTarget: v1alpha1.IPAddressTypeIPv4,
		observed:  v1alpha1.IPAddressTypeIPv4,
		available: true,
	}

	require.NoError(t, service.advanceGroupToIPv4(
		context.Background(), groupAddressTestClusterID, groupAddressTestGroupID, state,
	))
	assert.Empty(t, pls.Spec.TargetIPAddressType)
}

func TestSetPrivateLinkIPAddressTypeWaitsForIPv6Prerequisites(t *testing.T) {
	awsServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.NoError(t, r.ParseForm())
		w.Header().Set("Content-Type", "text/xml")
		switch r.Form.Get("Action") {
		case "DescribeLoadBalancers":
			_, _ = w.Write([]byte(`<DescribeLoadBalancersResponse><DescribeLoadBalancersResult><LoadBalancers><member>
<LoadBalancerArn>arn-1</LoadBalancerArn><VpcId>vpc-1</VpcId><IpAddressType>ipv4</IpAddressType>
<State><Code>active</Code></State><AvailabilityZones><member><SubnetId>subnet-1</SubnetId></member></AvailabilityZones>
</member></LoadBalancers></DescribeLoadBalancersResult></DescribeLoadBalancersResponse>`))
		case "DescribeVpcs":
			_, _ = w.Write([]byte(`<DescribeVpcsResponse><vpcSet><item><vpcId>vpc-1</vpcId>
<ipv6CidrBlockAssociationSet><item><ipv6CidrBlock>2600:1::/56</ipv6CidrBlock>
<ipv6CidrBlockState><state>associating</state></ipv6CidrBlockState>
</item></ipv6CidrBlockAssociationSet></item></vpcSet></DescribeVpcsResponse>`))
		case "DescribeVpcEndpointServiceConfigurations":
			_, _ = w.Write([]byte(`<DescribeVpcEndpointServiceConfigurationsResponse>
<serviceConfigurationSet><item><serviceId>vpce-svc-1</serviceId><serviceState>Available</serviceState>
<supportedIpAddressTypeSet><item>ipv4</item></supportedIpAddressTypeSet>
<networkLoadBalancerArnSet><item>arn-1</item></networkLoadBalancerArnSet></item></serviceConfigurationSet>
</DescribeVpcEndpointServiceConfigurationsResponse>`))
		default:
			http.Error(w, "unexpected action "+r.Form.Get("Action"), http.StatusBadRequest)
		}
	}))
	t.Cleanup(awsServer.Close)

	var clusterServerCalls atomic.Int32
	var kubeClient client.Client
	clusterServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		clusterServerCalls.Add(1)
		var body struct {
			RequestedIPAddressType string `json:"requestedIpAddressType"`
			NLBIPAddressType       string `json:"nlbIpAddressType"`
		}
		require.NoError(t, json.NewDecoder(r.Body).Decode(&body))
		assert.Equal(t, "PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK", body.RequestedIPAddressType)
		assert.Equal(t, "PRIVATE_LINK_IP_ADDRESS_TYPE_UNSPECIFIED", body.NLBIPAddressType)
		cluster := &cloudv1.Cluster{}
		require.NoError(t, kubeClient.Get(r.Context(), client.ObjectKey{
			Name: "cluster", Namespace: groupAddressTestNamespace,
		}, cluster))
		cluster.Spec.SqlEndpoints.PrivateLinkIPAddressType = cloudv1.PrivateLinkIPAddressTypeDualstack
		require.NoError(t, kubeClient.Update(r.Context(), cluster))
		w.WriteHeader(http.StatusOK)
	}))
	t.Cleanup(clusterServer.Close)

	pls := groupAddressTestPLS(groupAddressTestGroupID)
	kubeClient = groupAddressTestClient(t,
		groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeIPv4),
		groupAddressTestAPPRoute(),
		pls,
		groupAddressTestVPC(),
	)
	awsCli := groupAddressTestAWSClient(t, awsServer)
	service := &PrivateLinkServiceService{
		cli:                kubeClient,
		awsCli:             awsCli,
		config:             privateLinkGroupTestConfig(),
		groupManager:       privatelinkgroup.NewManager(kubeClient, awsCli),
		groupClusterClient: clustercli.NewClusterCli(clusterServer.URL),
	}

	err := service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK,
	))

	require.NoError(t, err)
	assert.Equal(t, int32(1), clusterServerCalls.Load())
	persisted := &v1alpha1.PrivateLinkService{}
	require.NoError(t, kubeClient.Get(context.Background(), client.ObjectKeyFromObject(pls), persisted))
	assert.Empty(t, persisted.Spec.TargetIPAddressType)

	status, err := service.GetPrivateLinkIPAddressType(
		context.Background(),
		&regionalapiv1.GetPrivateLinkIPAddressTypeRequest{
			ClusterId: groupAddressTestClusterID, TidbGroupId: groupAddressTestGroupID,
		},
	)
	require.NoError(t, err)
	assert.Equal(t, regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK, status.GetDesiredIpAddressType())
	assert.Equal(t, regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4, status.GetObservedIpAddressType())
	assert.False(t, status.GetReady())
	assert.Contains(t, status.GetMessage(), "waiting for VPC and attached NLB subnet IPv6 CIDRs")

	require.NoError(t, service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK,
	)))
	assert.Equal(t, int32(1), clusterServerCalls.Load(), "waiting for prerequisites must not advance the NLB target")
}

func TestSetPrivateLinkIPAddressTypeTreatsMissingNormalPLSAsPending(t *testing.T) {
	awsServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if groupAddressTestInfrastructureResponse(t, w, r, "ipv4") {
			return
		}
		http.Error(w, "unexpected action "+r.Form.Get("Action"), http.StatusBadRequest)
	}))
	t.Cleanup(awsServer.Close)

	kubeClient := groupAddressTestClient(
		t,
		groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeIPv4),
		groupAddressTestAPPRoute(),
		groupAddressTestVPC(),
	)
	awsCli := groupAddressTestAWSClient(t, awsServer)
	service := &PrivateLinkServiceService{
		cli:          kubeClient,
		awsCli:       awsCli,
		config:       privateLinkGroupTestConfig(),
		groupManager: privatelinkgroup.NewManager(kubeClient, awsCli),
	}

	err := service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4,
	))

	require.NoError(t, err)
	status, err := service.GetPrivateLinkIPAddressType(
		context.Background(),
		&regionalapiv1.GetPrivateLinkIPAddressTypeRequest{
			ClusterId: groupAddressTestClusterID, TidbGroupId: groupAddressTestGroupID,
		},
	)
	require.NoError(t, err)
	assert.False(t, status.GetReady())
	assert.Contains(t, status.GetMessage(), "not ready")
}

func TestPrivateLinkIPAddressTypeTreatsGroupPropagationAsPending(t *testing.T) {
	cluster := groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeIPv4)
	cluster.Annotations[cloudv1.DefaultLoadBalancerIDAnnotationKey] = "existing-group"
	kubeClient := groupAddressTestClient(t, cluster)
	service := &PrivateLinkServiceService{
		cli:          kubeClient,
		config:       privateLinkGroupTestConfig(),
		groupManager: privatelinkgroup.NewManager(kubeClient, nil),
	}

	status, err := service.GetPrivateLinkIPAddressType(
		context.Background(),
		&regionalapiv1.GetPrivateLinkIPAddressTypeRequest{
			ClusterId: groupAddressTestClusterID, TidbGroupId: groupAddressTestGroupID,
		},
	)
	require.NoError(t, err)
	assert.False(t, status.GetReady())
	assert.Equal(
		t,
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4,
		status.GetDesiredIpAddressType(),
	)
	assert.Equal(
		t,
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_UNSPECIFIED,
		status.GetObservedIpAddressType(),
	)
	assert.Contains(t, status.GetMessage(), "not ready in Cluster")

	err = service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK,
	))
	require.NoError(t, err)
}

func TestPrivateLinkIPAddressTypeTreatsEndpointServiceProvisioningAsPending(t *testing.T) {
	var endpointServiceCalls atomic.Int32
	awsServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if groupAddressTestInfrastructureResponse(t, w, r, "ipv4") {
			return
		}
		if r.Form.Get("Action") == "DescribeVpcEndpointServiceConfigurations" {
			endpointServiceCalls.Add(1)
		}
		http.Error(w, "unexpected action "+r.Form.Get("Action"), http.StatusBadRequest)
	}))
	t.Cleanup(awsServer.Close)

	pls := groupAddressTestPLS(groupAddressTestGroupID)
	pls.Status.EndpointService.ServiceID = ""
	kubeClient := groupAddressTestClient(
		t,
		groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeIPv4),
		groupAddressTestAPPRoute(),
		pls,
		groupAddressTestVPC(),
	)
	awsCli := groupAddressTestAWSClient(t, awsServer)
	service := &PrivateLinkServiceService{
		cli:          kubeClient,
		awsCli:       awsCli,
		config:       privateLinkGroupTestConfig(),
		groupManager: privatelinkgroup.NewManager(kubeClient, awsCli),
	}

	status, err := service.GetPrivateLinkIPAddressType(
		context.Background(),
		&regionalapiv1.GetPrivateLinkIPAddressTypeRequest{
			ClusterId: groupAddressTestClusterID, TidbGroupId: groupAddressTestGroupID,
		},
	)
	require.NoError(t, err)
	assert.False(t, status.GetReady())
	assert.Equal(
		t,
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4,
		status.GetDesiredIpAddressType(),
	)
	assert.Equal(
		t,
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_UNSPECIFIED,
		status.GetObservedIpAddressType(),
	)
	assert.Contains(t, status.GetMessage(), "Endpoint Service is not ready")

	err = service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4,
	))
	require.NoError(t, err)

	err = service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK,
	))
	require.NoError(t, err)
	assert.Zero(t, endpointServiceCalls.Load())
}

func TestPrivateLinkIPAddressTypeRejectsDisabledPrivateNLBAsPrecondition(t *testing.T) {
	cluster := groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeIPv4)
	cluster.Spec.SqlEndpoints.Private.Enabled = awssdk.Bool(false)
	kubeClient := groupAddressTestClient(t, cluster)
	service := &PrivateLinkServiceService{
		cli:          kubeClient,
		config:       privateLinkGroupTestConfig(),
		groupManager: privatelinkgroup.NewManager(kubeClient, nil),
	}

	_, err := service.GetPrivateLinkIPAddressType(
		context.Background(),
		&regionalapiv1.GetPrivateLinkIPAddressTypeRequest{
			ClusterId: groupAddressTestClusterID, TidbGroupId: groupAddressTestGroupID,
		},
	)
	require.Error(t, err)
	assert.True(t, bizerr.Is(err, bizerr.PreconditionFailed))
	assert.Contains(t, err.Error(), "must be enabled")

	err = service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4,
	))
	require.Error(t, err)
	assert.True(t, bizerr.Is(err, bizerr.PreconditionFailed))
}

func TestPrivateLinkIPAddressTypeRejectsNonEKSProviderAsPrecondition(t *testing.T) {
	service := &PrivateLinkServiceService{
		config: config.RootConfig{ControllerConfig: config.ControllerConfig{Provider: consts.ProviderTypeAWS}},
	}

	_, err := service.GetPrivateLinkIPAddressType(
		context.Background(),
		&regionalapiv1.GetPrivateLinkIPAddressTypeRequest{
			ClusterId: groupAddressTestClusterID, TidbGroupId: groupAddressTestGroupID,
		},
	)
	require.Error(t, err)
	assert.True(t, bizerr.Is(err, bizerr.PreconditionFailed))

	err = service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4,
	))
	require.Error(t, err)
	assert.True(t, bizerr.Is(err, bizerr.PreconditionFailed))
}

func TestPrivateLinkIPAddressTypeTreatsPrivateNLBProvisioningAsPending(t *testing.T) {
	cluster := groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeIPv4)
	cluster.Status.SqlEndpoints.PrivateEndpoint = nil
	kubeClient := groupAddressTestClient(t, cluster, groupAddressTestAPPRoute(), groupAddressTestVPC())
	service := &PrivateLinkServiceService{
		cli:          kubeClient,
		config:       privateLinkGroupTestConfig(),
		groupManager: privatelinkgroup.NewManager(kubeClient, nil),
	}

	status, err := service.GetPrivateLinkIPAddressType(
		context.Background(),
		&regionalapiv1.GetPrivateLinkIPAddressTypeRequest{
			ClusterId: groupAddressTestClusterID, TidbGroupId: groupAddressTestGroupID,
		},
	)
	require.NoError(t, err)
	assert.False(t, status.GetReady())
	assert.Equal(t,
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4,
		status.GetDesiredIpAddressType(),
	)

	err = service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4,
	))
	require.NoError(t, err)
}

func TestSetPrivateLinkIPAddressTypeTreatsIPv6OnlyAsDowngradeBlocker(t *testing.T) {
	awsServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if groupAddressTestInfrastructureResponse(t, w, r, "dualstack") {
			return
		}
		if r.Form.Get("Action") == "DescribeVpcEndpointServiceConfigurations" {
			_, _ = w.Write([]byte(`<DescribeVpcEndpointServiceConfigurationsResponse>
<serviceConfigurationSet><item><serviceId>vpce-svc-1</serviceId><serviceState>Available</serviceState>
<supportedIpAddressTypeSet><item>ipv6</item></supportedIpAddressTypeSet>
<networkLoadBalancerArnSet><item>arn-1</item></networkLoadBalancerArnSet></item></serviceConfigurationSet>
</DescribeVpcEndpointServiceConfigurationsResponse>`))
			return
		}
		http.Error(w, "unexpected action "+r.Form.Get("Action"), http.StatusBadRequest)
	}))
	t.Cleanup(awsServer.Close)

	kubeClient := groupAddressTestClient(t,
		groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeDualstack),
		groupAddressTestAPPRoute(),
		groupAddressTestDualstackPLS(),
		groupAddressTestVPC(),
	)
	awsCli := groupAddressTestAWSClient(t, awsServer)
	service := &PrivateLinkServiceService{
		cli: kubeClient, awsCli: awsCli, config: privateLinkGroupTestConfig(),
		groupManager: privatelinkgroup.NewManager(kubeClient, awsCli),
	}

	err := service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4,
	))

	require.Error(t, err)
	assert.True(t, bizerr.Is(err, bizerr.PrivateLinkAddressBlocked))
	assert.Contains(t, err.Error(), "IPv6-only")
}

func TestGetPrivateLinkIPAddressTypeDescribesMismatchedTargetsWithoutGuessingDirection(t *testing.T) {
	awsServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if groupAddressTestInfrastructureResponse(t, w, r, "dualstack") {
			return
		}
		switch r.Form.Get("Action") {
		case "DescribeVpcEndpointServiceConfigurations":
			_, _ = w.Write([]byte(`<DescribeVpcEndpointServiceConfigurationsResponse>
<serviceConfigurationSet><item><serviceId>vpce-svc-1</serviceId><serviceState>Available</serviceState>
<supportedIpAddressTypeSet><item>ipv4</item></supportedIpAddressTypeSet>
<networkLoadBalancerArnSet><item>arn-1</item></networkLoadBalancerArnSet></item></serviceConfigurationSet>
</DescribeVpcEndpointServiceConfigurationsResponse>`))
		default:
			http.Error(w, "unexpected action "+r.Form.Get("Action"), http.StatusBadRequest)
		}
	}))
	t.Cleanup(awsServer.Close)

	awsCli := groupAddressTestAWSClient(t, awsServer)
	pls := groupAddressTestPLS(groupAddressTestGroupID)
	pls.Spec.Region = "us-west-2"
	pls.Spec.VPCRef = v1alpha1.Reference{Name: "vpc-1", Namespace: groupAddressTestNamespace}
	kubeClient := groupAddressTestClient(t,
		groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeDualstack),
		groupAddressTestAPPRoute(),
		pls,
		groupAddressTestVPC(),
	)
	service := &PrivateLinkServiceService{
		cli: kubeClient, awsCli: awsCli, config: privateLinkGroupTestConfig(),
		groupManager: privatelinkgroup.NewManager(kubeClient, awsCli),
	}

	status, err := service.GetPrivateLinkIPAddressType(
		context.Background(),
		&regionalapiv1.GetPrivateLinkIPAddressTypeRequest{
			ClusterId: groupAddressTestClusterID, TidbGroupId: groupAddressTestGroupID,
		},
	)

	require.NoError(t, err)
	assert.False(t, status.GetReady())
	assert.Equal(
		t,
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK,
		status.GetDesiredIpAddressType(),
	)
	assert.Equal(
		t,
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4,
		status.GetObservedIpAddressType(),
	)
	assert.Equal(
		t,
		"waiting for requested target dualstack, normal PrivateLinkService target ipv4, and observed Endpoint Service address type ipv4 to converge",
		status.GetMessage(),
	)
}

func TestInterruptedIPv4DowngradeResumesNLBStepAndRejectsReverse(t *testing.T) {
	awsServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if groupAddressTestInfrastructureResponse(t, w, r, "dualstack") {
			return
		}
		switch r.Form.Get("Action") {
		case "DescribeVpcEndpointServiceConfigurations":
			_, _ = w.Write([]byte(`<DescribeVpcEndpointServiceConfigurationsResponse>
<serviceConfigurationSet><item><serviceId>vpce-svc-1</serviceId><serviceState>Available</serviceState>
<supportedIpAddressTypeSet><item>ipv4</item></supportedIpAddressTypeSet>
<networkLoadBalancerArnSet><item>arn-1</item></networkLoadBalancerArnSet></item></serviceConfigurationSet>
</DescribeVpcEndpointServiceConfigurationsResponse>`))
		default:
			http.Error(w, "unexpected action "+r.Form.Get("Action"), http.StatusBadRequest)
		}
	}))
	t.Cleanup(awsServer.Close)

	var clusterServerCalls atomic.Int32
	var body struct {
		RequestedIPAddressType string `json:"requestedIpAddressType"`
		NLBIPAddressType       string `json:"nlbIpAddressType"`
	}
	clusterServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		clusterServerCalls.Add(1)
		require.NoError(t, json.NewDecoder(r.Body).Decode(&body))
		w.WriteHeader(http.StatusOK)
	}))
	t.Cleanup(clusterServer.Close)

	cluster := groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeDualstack)
	cluster.Spec.SqlEndpoints.PrivateLinkIPAddressType = cloudv1.PrivateLinkIPAddressTypeIPv4
	pls := groupAddressTestPLS(groupAddressTestGroupID)
	pls.Spec.TargetIPAddressType = v1alpha1.IPAddressTypeIPv4
	kubeClient := groupAddressTestClient(
		t,
		cluster,
		groupAddressTestAPPRoute(),
		pls,
		groupAddressTestVPC(),
	)
	awsCli := groupAddressTestAWSClient(t, awsServer)
	service := &PrivateLinkServiceService{
		cli:                kubeClient,
		awsCli:             awsCli,
		config:             privateLinkGroupTestConfig(),
		groupManager:       privatelinkgroup.NewManager(kubeClient, awsCli),
		groupClusterClient: clustercli.NewClusterCli(clusterServer.URL),
	}

	err := service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK,
	))
	require.Error(t, err)
	assert.True(t, bizerr.Is(err, bizerr.PreconditionFailed))
	assert.Contains(t, err.Error(), "transition to ipv4 is still in progress")
	assert.Zero(t, clusterServerCalls.Load())

	require.NoError(t, service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4,
	)))
	assert.Equal(t, int32(1), clusterServerCalls.Load())
	assert.Equal(t, "PRIVATE_LINK_IP_ADDRESS_TYPE_UNSPECIFIED", body.RequestedIPAddressType)
	assert.Equal(t, "PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4", body.NLBIPAddressType)
}

func TestSetGroupAddressTypesPreservesClusterServerError(t *testing.T) {
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusInternalServerError)
		_, _ = w.Write([]byte(`{"code":13,"message":"injected Cluster Server failure"}`))
	}))
	t.Cleanup(server.Close)
	service := &PrivateLinkServiceService{groupClusterClient: clustercli.NewClusterCli(server.URL)}

	err := service.setGroupAddressTypes(
		context.Background(),
		groupAddressTestClusterID,
		groupAddressTestGroupID,
		v1alpha1.IPAddressTypeDualstack,
		v1alpha1.IPAddressTypeDualstack,
	)

	require.Error(t, err)
	assert.True(t, bizerr.Is(err, bizerr.InternalServerError))
	assert.Contains(t, err.Error(), "injected Cluster Server failure")
}

func TestSetPrivateLinkServiceAddressTypeRetriesConflictWithFreshGet(t *testing.T) {
	pls := groupAddressTestPLS(groupAddressTestGroupID)
	pls.UID = "pls-uid"
	kubeClient := groupAddressTestClient(
		t,
		groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeIPv4),
		groupAddressTestAPPRoute(),
		groupAddressTestVPC(),
		pls,
	)
	conflictClient := &conflictOncePrivateLinkServiceClient{Client: kubeClient}
	service := &PrivateLinkServiceService{
		cli:                  conflictClient,
		privatelinkservicerw: respls.NewPrivateLinkServiceReaderWriter(conflictClient),
		groupManager:         privatelinkgroup.NewManager(conflictClient, nil),
	}
	original := &v1alpha1.PrivateLinkService{}
	require.NoError(t, conflictClient.Get(
		context.Background(), client.ObjectKeyFromObject(pls), original,
	))

	require.NoError(t, service.setPrivateLinkServiceAddressType(
		context.Background(), groupAddressTestClusterID, groupAddressTestGroupID,
		&normalPrivateLinkService{
			pls:           original,
			vpc:           groupAddressTestVPC(),
			privateNLBARN: "arn-1",
		},
		v1alpha1.IPAddressTypeDualstack,
	))
	assert.Equal(t, 2, conflictClient.updateCalls)

	persisted := &v1alpha1.PrivateLinkService{}
	require.NoError(t, kubeClient.Get(
		context.Background(), client.ObjectKeyFromObject(pls), persisted,
	))
	assert.Equal(t, v1alpha1.IPAddressTypeDualstack, persisted.Spec.TargetIPAddressType)
	assert.Equal(t, "preserved", persisted.Annotations["concurrent-update"])
}

func TestSetPrivateLinkServiceAddressTypeRejectsMultipleBindingsAfterConflict(t *testing.T) {
	pls := groupAddressTestPLS(groupAddressTestGroupID)
	pls.UID = "pls-uid"
	kubeClient := groupAddressTestClient(
		t,
		groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeIPv4),
		groupAddressTestAPPRoute(),
		groupAddressTestVPC(),
		pls,
	)
	conflictClient := &conflictOncePrivateLinkServiceClient{
		Client: kubeClient,
		onConflict: func(current *v1alpha1.PrivateLinkService) {
			groupAddressTestPLSWithAdditionalBinding(current)
		},
	}
	service := &PrivateLinkServiceService{
		cli:                  conflictClient,
		privatelinkservicerw: respls.NewPrivateLinkServiceReaderWriter(conflictClient),
		groupManager:         privatelinkgroup.NewManager(conflictClient, nil),
	}
	original := &v1alpha1.PrivateLinkService{}
	require.NoError(t, conflictClient.Get(
		context.Background(), client.ObjectKeyFromObject(pls), original,
	))

	err := service.setPrivateLinkServiceAddressType(
		context.Background(), groupAddressTestClusterID, groupAddressTestGroupID,
		&normalPrivateLinkService{
			pls:           original,
			vpc:           groupAddressTestVPC(),
			privateNLBARN: "arn-1",
		},
		v1alpha1.IPAddressTypeDualstack,
	)

	require.Error(t, err)
	assert.True(t, bizerr.Is(err, bizerr.PreconditionFailed))
	assert.Equal(t, 1, conflictClient.updateCalls)
	persisted := &v1alpha1.PrivateLinkService{}
	require.NoError(t, kubeClient.Get(
		context.Background(), client.ObjectKeyFromObject(pls), persisted,
	))
	assert.Len(t, persisted.Spec.LoadBalancers, 2)
	assert.Empty(t, persisted.Spec.TargetIPAddressType)
}

type conflictOncePrivateLinkServiceClient struct {
	client.Client
	updateCalls int
	onConflict  func(*v1alpha1.PrivateLinkService)
}

func (c *conflictOncePrivateLinkServiceClient) Update(
	ctx context.Context,
	obj client.Object,
	opts ...client.UpdateOption,
) error {
	c.updateCalls++
	if c.updateCalls != 1 {
		return c.Client.Update(ctx, obj, opts...)
	}

	current := &v1alpha1.PrivateLinkService{}
	if err := c.Client.Get(ctx, client.ObjectKeyFromObject(obj), current); err != nil {
		return err
	}
	if c.onConflict != nil {
		c.onConflict(current)
	} else {
		if current.Annotations == nil {
			current.Annotations = map[string]string{}
		}
		current.Annotations["concurrent-update"] = "preserved"
	}
	if err := c.Client.Update(ctx, current); err != nil {
		return err
	}
	return apierrors.NewConflict(
		schema.GroupResource{
			Group: v1alpha1.GroupVersion.Group, Resource: "privatelinkservices",
		},
		obj.GetName(),
		errors.New("injected conflict"),
	)
}

func TestUnpausedLegacyPrivateLinkServiceUsesManagedReconciliation(t *testing.T) {
	modifyCalls := 0
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if groupAddressTestInfrastructureResponse(t, w, r, "dualstack") {
			return
		}
		require.NoError(t, r.ParseForm())
		w.Header().Set("Content-Type", "text/xml")
		switch r.Form.Get("Action") {
		case "DescribeVpcEndpointServiceConfigurations":
			_, _ = w.Write([]byte(`<DescribeVpcEndpointServiceConfigurationsResponse>
<serviceConfigurationSet><item><serviceId>vpce-svc-1</serviceId><serviceState>Available</serviceState>
<supportedIpAddressTypeSet><item>ipv4</item></supportedIpAddressTypeSet>
<networkLoadBalancerArnSet><item>arn-1</item></networkLoadBalancerArnSet></item></serviceConfigurationSet>
</DescribeVpcEndpointServiceConfigurationsResponse>`))
		case "ModifyVpcEndpointServiceConfiguration":
			modifyCalls++
			_, _ = w.Write([]byte(`<ModifyVpcEndpointServiceConfigurationResponse><return>true</return></ModifyVpcEndpointServiceConfigurationResponse>`))
		default:
			http.Error(w, "unexpected action "+r.Form.Get("Action"), http.StatusBadRequest)
		}
	}))
	t.Cleanup(server.Close)

	awsCli := groupAddressTestAWSClient(t, server)
	pls := groupAddressTestLegacyPLS("legacy-pls")
	pls.Spec.Region = "us-west-2"
	pls.Spec.VPCRef = v1alpha1.Reference{Name: "vpc-1", Namespace: groupAddressTestNamespace}
	kubeClient := groupAddressTestClient(t,
		groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeDualstack),
		groupAddressTestAPPRoute(), pls, groupAddressTestVPC(),
	)
	service := &PrivateLinkServiceService{
		cli: kubeClient, awsCli: awsCli, config: privateLinkGroupTestConfig(),
		groupManager:         privatelinkgroup.NewManager(kubeClient, awsCli),
		privatelinkservicerw: respls.NewPrivateLinkServiceReaderWriter(kubeClient),
	}

	require.NoError(t, service.SetPrivateLinkIPAddressType(context.Background(), privateLinkGroupSetRequest(
		regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_DUALSTACK,
	)))
	assert.Zero(t, modifyCalls)
	persisted := &v1alpha1.PrivateLinkService{}
	require.NoError(t, kubeClient.Get(context.Background(), client.ObjectKeyFromObject(pls), persisted))
	assert.Equal(t, v1alpha1.IPAddressTypeDualstack, persisted.Spec.TargetIPAddressType)
	assert.NotEqual(t, "true", persisted.Annotations[v1alpha1.AnnotationKeyReconciliationPaused])
}

func groupAddressTestVPC() *v1alpha1.VPC {
	return &v1alpha1.VPC{
		ObjectMeta: metav1.ObjectMeta{Name: "vpc-1", Namespace: groupAddressTestNamespace},
		Spec: v1alpha1.VPCSpec{
			Provider: consts.ProviderTypeEKS,
			Region:   "us-west-2",
			AccountRef: &v1alpha1.AccountReference{
				AccountID: "123456789012", Role: "regional-server",
			},
		},
		Status: v1alpha1.VPCStatus{VPC: v1alpha1.VPCRes{VPCID: "vpc-1"}},
	}
}

func groupAddressTestAWSClient(t *testing.T, server *httptest.Server) regionalaws.Client {
	t.Helper()
	awsCli := regionalaws.NewMockClient(gomock.NewController(t))
	awsCli.EXPECT().EC2(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(
		func(_ *regionalaws.AssumedRole, opts ...regionalaws.Option) *ec2.Client {
			cfg := awssdk.Config{
				Region:       "wrong-region",
				Credentials:  awssdk.AnonymousCredentials{},
				HTTPClient:   server.Client(),
				BaseEndpoint: awssdk.String(server.URL),
			}
			for _, opt := range opts {
				opt(&cfg)
			}
			assert.Equal(t, "us-west-2", cfg.Region)
			return ec2.NewFromConfig(cfg)
		},
	)
	awsCli.EXPECT().ELBV2(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(
		func(_ *regionalaws.AssumedRole, opts ...regionalaws.Option) *elasticloadbalancingv2.Client {
			return groupAddressTestELBV2Client(t, server, opts...)
		},
	)
	return awsCli
}

func privateLinkGroupTestConfig() config.RootConfig {
	return config.RootConfig{ControllerConfig: config.ControllerConfig{Provider: consts.ProviderTypeEKS}}
}

func privateLinkGroupSetRequest(addressType regionalapiv1.PrivateLinkIPAddressType) *regionalapiv1.SetPrivateLinkIPAddressTypeRequest {
	return &regionalapiv1.SetPrivateLinkIPAddressTypeRequest{
		ClusterId:     groupAddressTestClusterID,
		TidbGroupId:   groupAddressTestGroupID,
		IpAddressType: addressType,
	}
}

func groupAddressTestDualstackPLS() *v1alpha1.PrivateLinkService {
	pls := groupAddressTestPLS(groupAddressTestGroupID)
	pls.Spec.TargetIPAddressType = v1alpha1.IPAddressTypeDualstack
	return pls
}