package privatelinkservice

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"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"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"

	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"
	respls "github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/privatelinkservice"
)

const (
	groupAddressTestClusterID = "cluster-1"
	groupAddressTestGroupID   = "group-1"
	groupAddressTestNamespace = "cluster-namespace"
)

func TestNormalGroupPrivateLinkService(t *testing.T) {
	tests := []struct {
		name       string
		objects    []client.Object
		wantErr    bool
		wantCode   bizerr.ErrorCode
		wantDetail string
	}{
		{name: "normal PLS", objects: []client.Object{groupAddressTestAPPRoute(), groupAddressTestPLS(groupAddressTestGroupID)}},
		{name: "normal PLS with generated CR name", objects: []client.Object{groupAddressTestAPPRoute(), groupAddressTestPLS("generated-pls-id")}},
		{name: "legacy normal PLS", objects: []client.Object{groupAddressTestAPPRoute(), groupAddressTestLegacyPLS("legacy-pls")}},
		{
			name: "paused legacy PLS is not managed by this workflow",
			objects: []client.Object{
				groupAddressTestAPPRoute(), groupAddressTestImportedLegacyPLS("imported-pls"),
			},
			wantErr: true, wantCode: bizerr.PreconditionFailed, wantDetail: "not ready",
		},
		{
			name: "missing normal PLS", objects: []client.Object{groupAddressTestAPPRoute()},
			wantErr: true, wantCode: bizerr.PrivateLinkAddressPending, wantDetail: "not ready",
		},
		{
			name: "multiple native PLSs bind the same group",
			objects: []client.Object{
				groupAddressTestAPPRoute(), groupAddressTestPLS("native-pls-1"), groupAddressTestPLS("native-pls-2"),
			},
			wantErr: true, wantCode: bizerr.PreconditionFailed, wantDetail: "got 2",
		},
		{
			name: "native and imported legacy PLSs bind the same group",
			objects: []client.Object{
				groupAddressTestAPPRoute(), groupAddressTestPLS("native-pls"), groupAddressTestImportedLegacyPLS("imported-pls"),
			},
			wantErr: true, wantCode: bizerr.PreconditionFailed, wantDetail: "got 2",
		},
		{
			name: "native PLS binds multiple groups",
			objects: []client.Object{
				groupAddressTestAPPRoute(), groupAddressTestPLSWithAdditionalBinding(groupAddressTestPLS("native-pls")),
			},
			wantErr: true, wantCode: bizerr.PreconditionFailed, wantDetail: "must bind exactly one",
		},
		{
			name: "imported legacy PLS binds multiple groups",
			objects: []client.Object{
				groupAddressTestAPPRoute(), groupAddressTestPLSWithAdditionalBinding(groupAddressTestImportedLegacyPLS("imported-pls")),
			},
			wantErr: true, wantCode: bizerr.PreconditionFailed, wantDetail: "must bind exactly one",
		},
		{
			name: "native paused PLS is not managed by this workflow",
			objects: []client.Object{
				groupAddressTestAPPRoute(), groupAddressTestPausedNativePLS(),
			},
			wantErr: true, wantCode: bizerr.PreconditionFailed, wantDetail: "not ready",
		},
		{
			name: "PLS VPC differs from APPRoute VPC",
			objects: []client.Object{
				groupAddressTestAPPRoute(), groupAddressTestPLSWithVPC("vpc-other"),
			},
			wantErr: true, wantCode: bizerr.PreconditionFailed, wantDetail: "not the TiDB Group APPRoute VPC",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			objects := []client.Object{
				groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeIPv4),
				groupAddressTestVPC(),
			}
			objects = append(objects, tt.objects...)
			kubeClient := groupAddressTestClient(t, objects...)
			service := &PrivateLinkServiceService{
				cli:          kubeClient,
				groupManager: privatelinkgroup.NewManager(kubeClient, nil),
			}

			pls, err := service.normalGroupPrivateLinkService(
				context.Background(),
				groupAddressTestClusterID,
				groupAddressTestGroupID,
				groupAddressTestVPC(),
				"arn-1",
			)

			if !tt.wantErr {
				require.NoError(t, err)
				require.NotNil(t, pls.pls)
				if tt.name == "legacy normal PLS" {
					assert.Equal(t, "legacy-pls", pls.pls.Name)
				} else {
					if tt.name == "normal PLS with generated CR name" {
						assert.Equal(t, "generated-pls-id", pls.pls.Name)
					} else {
						assert.Equal(t, groupAddressTestGroupID, pls.pls.Name)
					}
				}
				return
			}
			require.Error(t, err)
			assert.True(t, bizerr.Is(err, tt.wantCode))
			assert.ErrorContains(t, err, tt.wantDetail)
		})
	}
}

func TestObserveEndpointServiceAddressTypeScopesGroupNLBCheckToLegacyPLS(t *testing.T) {
	for _, tt := range []struct {
		name    string
		arnsXML string
		legacy  bool
		wantErr bool
	}{
		{name: "native trusts controller ownership", arnsXML: "<item>arn-other</item>"},
		{name: "legacy rejects different NLB", arnsXML: "<item>arn-other</item>", legacy: true, wantErr: true},
		{name: "legacy rejects additional NLB", arnsXML: "<item>arn-1</item><item>arn-other</item>", legacy: true, wantErr: true},
	} {
		t.Run(tt.name, func(t *testing.T) {
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				require.NoError(t, r.ParseForm())
				require.Equal(t, "DescribeVpcEndpointServiceConfigurations", r.Form.Get("Action"))
				w.Header().Set("Content-Type", "text/xml")
				_, _ = w.Write([]byte(`<DescribeVpcEndpointServiceConfigurationsResponse>
<serviceConfigurationSet><item><serviceId>vpce-svc-1</serviceId><serviceState>Available</serviceState>
<supportedIpAddressTypeSet><item>ipv4</item></supportedIpAddressTypeSet>
<networkLoadBalancerArnSet>` + tt.arnsXML + `</networkLoadBalancerArnSet>
</item></serviceConfigurationSet></DescribeVpcEndpointServiceConfigurationsResponse>`))
			}))
			t.Cleanup(server.Close)

			pls := groupAddressTestPLS(groupAddressTestGroupID)
			if tt.legacy {
				pls = groupAddressTestLegacyPLS("legacy-pls")
			}
			kubeClient := groupAddressTestClient(
				t,
				groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeIPv4),
				groupAddressTestAPPRoute(),
				pls,
				groupAddressTestVPC(),
			)
			service := &PrivateLinkServiceService{
				cli:          kubeClient,
				awsCli:       groupAddressTestAWSClient(t, server),
				groupManager: privatelinkgroup.NewManager(kubeClient, nil),
			}
			normal, err := service.normalGroupPrivateLinkService(
				context.Background(),
				groupAddressTestClusterID,
				groupAddressTestGroupID,
				groupAddressTestVPC(),
				"arn-1",
			)
			require.NoError(t, err)

			observed, available, err := service.observeEndpointServiceAddressType(context.Background(), normal)

			if tt.wantErr {
				require.Error(t, err)
				assert.True(t, bizerr.Is(err, bizerr.PreconditionFailed))
				assert.ErrorContains(t, err, "must be associated only with the TiDB Group private NLB arn-1")
				return
			}
			require.NoError(t, err)
			assert.Equal(t, v1alpha1.IPAddressTypeIPv4, observed)
			assert.True(t, available)
		})
	}
}

func TestValidateGroupIPv4DowngradeConnectionsClassifiesNonUserBlockers(t *testing.T) {
	tests := []struct {
		name     string
		state    string
		wantCode bizerr.ErrorCode
	}{
		{name: "deleting connection is pending", state: "deleting", wantCode: bizerr.PrivateLinkAddressPending},
		{name: "unknown connection state is a provider error", state: "future-state", wantCode: bizerr.AWSClientError},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				require.NoError(t, r.ParseForm())
				require.Equal(t, "DescribeVpcEndpointConnections", r.Form.Get("Action"))
				w.Header().Set("Content-Type", "text/xml")
				_, _ = w.Write([]byte(`<DescribeVpcEndpointConnectionsResponse>
<vpcEndpointConnectionSet><item><vpcEndpointId>vpce-dualstack</vpcEndpointId>
<vpcEndpointState>` + tt.state + `</vpcEndpointState><ipAddressType>dualstack</ipAddressType>
</item></vpcEndpointConnectionSet></DescribeVpcEndpointConnectionsResponse>`))
			}))
			t.Cleanup(server.Close)

			cli := ec2.NewFromConfig(awssdk.Config{
				Region: "us-west-2", Credentials: awssdk.AnonymousCredentials{},
				HTTPClient: server.Client(), BaseEndpoint: awssdk.String(server.URL),
			})
			err := (&PrivateLinkServiceService{}).validateGroupIPv4DowngradeConnections(
				context.Background(), cli, groupAddressTestPLS(groupAddressTestGroupID),
			)

			require.Error(t, err)
			assert.True(t, bizerr.Is(err, tt.wantCode))
		})
	}
}

func TestValidatePrivateLinkIPv4OnlyIsReadOnlyAndClassifiesConnections(t *testing.T) {
	for _, tt := range []struct {
		name      string
		state     string
		requested cloudv1.PrivateLinkIPAddressType
		plsTarget v1alpha1.IPAddressType
		wantCode  bizerr.ErrorCode
	}{
		{name: "rejected connection blocks admission", state: "rejected", requested: cloudv1.PrivateLinkIPAddressTypeDualstack, plsTarget: v1alpha1.IPAddressTypeDualstack, wantCode: bizerr.PrivateLinkAddressBlocked},
		{name: "failed connection blocks admission", state: "failed", requested: cloudv1.PrivateLinkIPAddressTypeDualstack, plsTarget: v1alpha1.IPAddressTypeDualstack, wantCode: bizerr.PrivateLinkAddressBlocked},
		{
			name:  "failed connection blocks resumed admission before AWS observation reaches IPv4",
			state: "failed", requested: cloudv1.PrivateLinkIPAddressTypeIPv4,
			plsTarget: v1alpha1.IPAddressTypeIPv4, wantCode: bizerr.PrivateLinkAddressBlocked,
		},
		{name: "deleting connection leaves admission pending", state: "deleting", requested: cloudv1.PrivateLinkIPAddressTypeDualstack, plsTarget: v1alpha1.IPAddressTypeDualstack, wantCode: bizerr.PrivateLinkAddressPending},
		{name: "deleted connection allows admission", state: "deleted", requested: cloudv1.PrivateLinkIPAddressTypeDualstack, plsTarget: v1alpha1.IPAddressTypeDualstack},
	} {
		t.Run(tt.name, func(t *testing.T) {
			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><item>ipv6</item></supportedIpAddressTypeSet>
<networkLoadBalancerArnSet><item>arn-1</item></networkLoadBalancerArnSet>
</item></serviceConfigurationSet></DescribeVpcEndpointServiceConfigurationsResponse>`))
				case "DescribeVpcEndpointConnections":
					_, _ = w.Write([]byte(`<DescribeVpcEndpointConnectionsResponse>
<vpcEndpointConnectionSet><item><vpcEndpointId>vpce-dualstack</vpcEndpointId>
<vpcEndpointState>` + tt.state + `</vpcEndpointState><ipAddressType>dualstack</ipAddressType>
</item></vpcEndpointConnectionSet></DescribeVpcEndpointConnectionsResponse>`))
				default:
					t.Errorf("precheck must not call mutating AWS action %q", r.Form.Get("Action"))
					http.Error(w, "unexpected action", http.StatusBadRequest)
				}
			}))
			t.Cleanup(server.Close)

			ctl := gomock.NewController(t)
			awsCli := regionalaws.NewMockClient(ctl)
			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)
					}
					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...)
				},
			)

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

			err := service.ValidatePrivateLinkIPv4Only(t.Context(), &regionalapiv1.GetPrivateLinkIPAddressTypeRequest{
				ClusterId: groupAddressTestClusterID, TidbGroupId: groupAddressTestGroupID,
			})

			if tt.wantCode == 0 {
				require.NoError(t, err)
			} else {
				require.Error(t, err)
				assert.True(t, bizerr.Is(err, tt.wantCode))
			}
			persisted := &v1alpha1.PrivateLinkService{}
			require.NoError(t, kubeClient.Get(t.Context(), client.ObjectKeyFromObject(pls), persisted))
			assert.Equal(t, tt.plsTarget, persisted.Spec.TargetIPAddressType)
		})
	}
}

func TestSetPrivateLinkIPAddressTypeDoesNotPatchPLSWhenIPv6ConnectionExists(t *testing.T) {
	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><item>ipv6</item></supportedIpAddressTypeSet>
<networkLoadBalancerArnSet><item>arn-1</item></networkLoadBalancerArnSet>
</item></serviceConfigurationSet>
</DescribeVpcEndpointServiceConfigurationsResponse>`))
		case "DescribeVpcEndpointConnections":
			_, _ = w.Write([]byte(`<DescribeVpcEndpointConnectionsResponse>
<vpcEndpointConnectionSet><item>
<vpcEndpointId>vpce-ipv6</vpcEndpointId><vpcEndpointState>available</vpcEndpointState><ipAddressType>ipv6</ipAddressType>
</item></vpcEndpointConnectionSet>
</DescribeVpcEndpointConnectionsResponse>`))
		default:
			http.Error(w, "unexpected action "+r.Form.Get("Action"), http.StatusBadRequest)
		}
	}))
	t.Cleanup(server.Close)

	ctl := gomock.NewController(t)
	awsCli := regionalaws.NewMockClient(ctl)
	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...)
		},
	)
	pls := groupAddressTestPLS(groupAddressTestGroupID)
	pls.Spec.Region = "us-west-2"
	pls.Spec.VPCRef = v1alpha1.Reference{Name: "vpc-1", Namespace: groupAddressTestNamespace}
	pls.Spec.TargetIPAddressType = v1alpha1.IPAddressTypeDualstack
	kubeClient := groupAddressTestClient(t,
		groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeDualstack),
		groupAddressTestAPPRoute(),
		pls,
		groupAddressTestVPC(),
	)
	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.ErrorContains(t, err, "vpce-ipv6")
	persisted := &v1alpha1.PrivateLinkService{}
	require.NoError(t, kubeClient.Get(context.Background(), client.ObjectKey{
		Name: groupAddressTestGroupID, Namespace: groupAddressTestNamespace,
	}, persisted))
	assert.Equal(t, v1alpha1.IPAddressTypeDualstack, persisted.Spec.TargetIPAddressType)
}

func TestLegacyDisableChangesGroupNLBAfterControllerReadback(t *testing.T) {
	observed := "dualstack"
	modifyCalls := 0
	awsServer := 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":
			types := "<item>ipv4</item>"
			if observed == "dualstack" {
				types += "<item>ipv6</item>"
			}
			_, _ = w.Write([]byte(`<DescribeVpcEndpointServiceConfigurationsResponse>
<serviceConfigurationSet><item><serviceId>vpce-svc-1</serviceId><serviceState>Available</serviceState>
<supportedIpAddressTypeSet>` + types + `</supportedIpAddressTypeSet>
<networkLoadBalancerArnSet><item>arn-1</item></networkLoadBalancerArnSet></item></serviceConfigurationSet>
</DescribeVpcEndpointServiceConfigurationsResponse>`))
		case "DescribeVpcEndpointConnections":
			_, _ = w.Write([]byte(`<DescribeVpcEndpointConnectionsResponse><vpcEndpointConnectionSet/></DescribeVpcEndpointConnectionsResponse>`))
		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(awsServer.Close)
	groupWrites := 0
	var kubeClient client.Client
	clusterServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		groupWrites++
		var body struct {
			RequestedIPAddressType string `json:"requestedIpAddressType"`
			NLBIPAddressType       string `json:"nlbIpAddressType"`
		}
		require.NoError(t, json.NewDecoder(r.Body).Decode(&body))
		cluster := &cloudv1.Cluster{}
		require.NoError(t, kubeClient.Get(r.Context(), client.ObjectKey{
			Name: "cluster", Namespace: groupAddressTestNamespace,
		}, cluster))
		if body.RequestedIPAddressType == "PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4" {
			cluster.Spec.SqlEndpoints.PrivateLinkIPAddressType = cloudv1.PrivateLinkIPAddressTypeIPv4
		}
		if body.NLBIPAddressType == "PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4" {
			cluster.Spec.SqlEndpoints.PrivateLinkNLBIPAddressType = cloudv1.PrivateLinkIPAddressTypeIPv4
		}
		require.NoError(t, kubeClient.Update(r.Context(), cluster))
		w.WriteHeader(http.StatusOK)
	}))
	t.Cleanup(clusterServer.Close)

	ctl := gomock.NewController(t)
	awsCli := regionalaws.NewMockClient(ctl)
	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: awsServer.Client(), BaseEndpoint: awssdk.String(awsServer.URL),
			}
			for _, opt := range opts {
				opt(&cfg)
			}
			return ec2.NewFromConfig(cfg)
		},
	)
	awsCli.EXPECT().ELBV2(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(
		func(_ *regionalaws.AssumedRole, opts ...regionalaws.Option) *elasticloadbalancingv2.Client {
			return groupAddressTestELBV2Client(t, awsServer, opts...)
		},
	)
	pls := groupAddressTestLegacyPLS("legacy-pls")
	pls.Spec.Region = "us-west-2"
	pls.Spec.VPCRef = v1alpha1.Reference{Name: "vpc-1", Namespace: groupAddressTestNamespace}
	pls.Spec.TargetIPAddressType = v1alpha1.IPAddressTypeDualstack
	kubeClient = groupAddressTestClient(t,
		groupAddressTestCluster(cloudv1.PrivateLinkIPAddressTypeDualstack),
		groupAddressTestAPPRoute(), pls, groupAddressTestVPC(),
	)
	service := &PrivateLinkServiceService{
		cli: kubeClient, awsCli: awsCli,
		config: privateLinkGroupTestConfig(), groupClusterClient: clustercli.NewClusterCli(clusterServer.URL),
		groupManager:         privatelinkgroup.NewManager(kubeClient, awsCli),
		privatelinkservicerw: respls.NewPrivateLinkServiceReaderWriter(kubeClient),
	}
	req := privateLinkGroupSetRequest(regionalapiv1.PrivateLinkIPAddressType_PRIVATE_LINK_IP_ADDRESS_TYPE_IPV4)

	require.NoError(t, service.SetPrivateLinkIPAddressType(context.Background(), req))
	assert.Zero(t, modifyCalls)
	assert.Equal(t, 1, groupWrites, "the first request persists the transition direction")
	persistedPLS := &v1alpha1.PrivateLinkService{}
	require.NoError(t, kubeClient.Get(context.Background(), client.ObjectKeyFromObject(pls), persistedPLS))
	assert.Equal(t, v1alpha1.IPAddressTypeDualstack, persistedPLS.Spec.TargetIPAddressType)
	persistedCluster := &cloudv1.Cluster{}
	require.NoError(t, kubeClient.Get(context.Background(), client.ObjectKey{
		Name: "cluster", Namespace: groupAddressTestNamespace,
	}, persistedCluster))
	assert.Equal(t, cloudv1.PrivateLinkIPAddressTypeIPv4, persistedCluster.Spec.SqlEndpoints.PrivateLinkIPAddressType)
	assert.Equal(t, cloudv1.PrivateLinkIPAddressTypeDualstack, persistedCluster.Spec.SqlEndpoints.PrivateLinkNLBIPAddressType)
	persistedPLS = &v1alpha1.PrivateLinkService{}
	require.NoError(t, kubeClient.Get(context.Background(), client.ObjectKeyFromObject(pls), persistedPLS))
	assert.Equal(t, v1alpha1.IPAddressTypeDualstack, persistedPLS.Spec.TargetIPAddressType)

	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_DUALSTACK,
		status.GetObservedIpAddressType(),
	)
	assert.Equal(
		t,
		"waiting for requested target ipv4 and NLB target dualstack to converge",
		status.GetMessage(),
	)

	require.NoError(t, service.SetPrivateLinkIPAddressType(context.Background(), req))
	assert.Zero(t, modifyCalls, "the second request persists the IPv4 PLS target")
	assert.Equal(t, 1, groupWrites)
	persistedPLS = &v1alpha1.PrivateLinkService{}
	require.NoError(t, kubeClient.Get(context.Background(), client.ObjectKeyFromObject(pls), persistedPLS))
	assert.Equal(t, v1alpha1.IPAddressTypeIPv4, persistedPLS.Spec.TargetIPAddressType)

	// The EndpointService controller is the only AWS writer. Simulate its
	// successful provider readback before advancing the NLB target.
	observed = "ipv4"
	require.NoError(t, service.SetPrivateLinkIPAddressType(context.Background(), req))
	assert.Zero(t, modifyCalls, "Regional must not modify the Endpoint Service")
	assert.Equal(t, 2, groupWrites, "the third request changes the NLB target only after live IPv4 readback")
}

func groupAddressTestClient(t *testing.T, objects ...client.Object) client.Client {
	t.Helper()
	scheme := runtime.NewScheme()
	require.NoError(t, cloudv1.AddToScheme(scheme))
	require.NoError(t, v1alpha1.AddToScheme(scheme))
	return fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build()
}

func groupAddressTestAPPRoute() *v1alpha1.APPRoute {
	return &v1alpha1.APPRoute{
		ObjectMeta: metav1.ObjectMeta{
			Name: groupAddressTestClusterID, Namespace: groupAddressTestNamespace,
			Labels: map[string]string{v1alpha1.ClusterIDKey: groupAddressTestClusterID},
		},
		Spec: v1alpha1.APPRouteSpec{VPCRef: &v1alpha1.Reference{
			Name: "vpc-1", Namespace: groupAddressTestNamespace,
		}},
	}
}

func groupAddressTestCluster(addressType cloudv1.PrivateLinkIPAddressType) *cloudv1.Cluster {
	return &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster", Namespace: groupAddressTestNamespace,
			Labels: map[string]string{cloudv1.ClusterTagKey: groupAddressTestClusterID},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: groupAddressTestGroupID,
			},
		},
		Spec: cloudv1.ClusterSpec{
			Kubernetes: cloudv1.Kubernetes{Provider: consts.ProviderTypeEKS},
			SqlEndpoints: cloudv1.SqlEndpoints{
				Private:                     &cloudv1.Service{Enabled: awssdk.Bool(true)},
				PrivateLinkIPAddressType:    addressType,
				PrivateLinkNLBIPAddressType: addressType,
			},
		},
		Status: cloudv1.ClusterStatus{SqlEndpoints: cloudv1.SqlEndpointsStatus{
			PrivateEndpoint: &cloudv1.SqlEndpointStatus{ResourceID: "arn-1"},
		}},
	}
}

func groupAddressTestInfrastructureResponse(
	t *testing.T,
	w http.ResponseWriter,
	r *http.Request,
	addressType string,
) bool {
	t.Helper()
	require.NoError(t, r.ParseForm())
	w.Header().Set("Content-Type", "text/xml")
	switch r.Form.Get("Action") {
	case "DescribeVpcs":
		_, _ = w.Write([]byte(`<DescribeVpcsResponse><vpcSet><item><vpcId>vpc-1</vpcId>
<ipv6CidrBlockAssociationSet><item><ipv6CidrBlock>2600:1::/56</ipv6CidrBlock>
<ipv6CidrBlockState><state>associated</state></ipv6CidrBlockState>
</item></ipv6CidrBlockAssociationSet></item></vpcSet></DescribeVpcsResponse>`))
	case "DescribeSubnets":
		_, _ = w.Write([]byte(`<DescribeSubnetsResponse><subnetSet><item><subnetId>subnet-1</subnetId><vpcId>vpc-1</vpcId>
<ipv6CidrBlockAssociationSet><item><ipv6CidrBlock>2600:1::/64</ipv6CidrBlock>
<ipv6CidrBlockState><state>associated</state></ipv6CidrBlockState>
</item></ipv6CidrBlockAssociationSet></item></subnetSet></DescribeSubnetsResponse>`))
	case "DescribeLoadBalancers":
		_, _ = w.Write([]byte(`<DescribeLoadBalancersResponse><DescribeLoadBalancersResult><LoadBalancers><member>
<LoadBalancerArn>arn-1</LoadBalancerArn><VpcId>vpc-1</VpcId><IpAddressType>` + addressType + `</IpAddressType>
<State><Code>active</Code></State><AvailabilityZones><member><SubnetId>subnet-1</SubnetId></member></AvailabilityZones>
</member></LoadBalancers></DescribeLoadBalancersResult></DescribeLoadBalancersResponse>`))
	default:
		return false
	}
	return true
}

func groupAddressTestELBV2Client(
	t *testing.T,
	server *httptest.Server,
	opts ...regionalaws.Option,
) *elasticloadbalancingv2.Client {
	t.Helper()
	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 elasticloadbalancingv2.NewFromConfig(cfg)
}

func groupAddressTestPLS(name string) *v1alpha1.PrivateLinkService {
	return &v1alpha1.PrivateLinkService{
		ObjectMeta: metav1.ObjectMeta{
			Name: name, Namespace: groupAddressTestNamespace,
			Labels: map[string]string{
				v1alpha1.ClusterIDKey:      groupAddressTestClusterID,
				v1alpha1.LoadBalancerIDKey: groupAddressTestGroupID,
			},
		},
		Spec: v1alpha1.PrivateLinkServiceSpec{
			Provider: consts.ProviderTypeEKS,
			Region:   "us-west-2",
			VPCRef: v1alpha1.Reference{
				Name: "vpc-1", Namespace: groupAddressTestNamespace,
			},
			LoadBalancers: []v1alpha1.LoadBalancer{{
				ClusterID: groupAddressTestClusterID, ID: groupAddressTestGroupID,
			}},
		},
		Status: v1alpha1.PrivateLinkServiceStatus{
			Phase: v1alpha1.PrivateLinkServicePhaseActive,
			EndpointService: v1alpha1.EndpointServiceStatus{
				ServiceID: "vpce-svc-1",
			},
		},
	}
}

func groupAddressTestPLSWithAdditionalBinding(pls *v1alpha1.PrivateLinkService) *v1alpha1.PrivateLinkService {
	pls.Spec.LoadBalancers = append(pls.Spec.LoadBalancers, v1alpha1.LoadBalancer{
		ClusterID: groupAddressTestClusterID,
		ID:        "group-2",
	})
	return pls
}

func groupAddressTestPLSWithVPC(vpcName string) *v1alpha1.PrivateLinkService {
	pls := groupAddressTestPLS(groupAddressTestGroupID)
	pls.Spec.VPCRef.Name = vpcName
	return pls
}

func groupAddressTestImportedLegacyPLS(name string) *v1alpha1.PrivateLinkService {
	pls := groupAddressTestPLS(name)
	pls.Labels[v1alpha1.IsLegacyKey] = "true"
	pls.Annotations = map[string]string{v1alpha1.AnnotationKeyReconciliationPaused: "true"}
	return pls
}

func groupAddressTestLegacyPLS(name string) *v1alpha1.PrivateLinkService {
	pls := groupAddressTestPLS(name)
	pls.Labels[v1alpha1.IsLegacyKey] = "true"
	return pls
}

func groupAddressTestPausedNativePLS() *v1alpha1.PrivateLinkService {
	pls := groupAddressTestPLS(groupAddressTestGroupID)
	pls.Annotations = map[string]string{v1alpha1.AnnotationKeyReconciliationPaused: "true"}
	return pls
}