package eks

import (
	"context"
	"fmt"
	"testing"
	"time"

	awssdk "github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/ec2"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	ec2v1beta1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1beta1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/go-logr/logr"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/util"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/privatelinkservice/task"
	respls "github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/privatelinkservice"
	resvpc "github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/vpc"
	"github.com/tidbcloud/infra-provider/regional-server/test/kubemock"
	mockprivatelinkservice "github.com/tidbcloud/infra-provider/regional-server/test/resourcemanager/mockprivatelinkservice"
	"github.com/tidbcloud/infra-provider/regional-server/test/testdata"
	"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"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/yaml"
)

func fakeTaskContext(t *testing.T) *task.Context {
	s := runtime.NewScheme()
	v1alpha1.AddToScheme(s)
	ec2v1beta1.SchemeBuilder.AddToScheme(s)
	ec2v1alpha1.SchemeBuilder.AddToScheme(s)
	fakeVPC := &v1alpha1.VPC{}
	bs, err := testdata.LoadDefaultVPCYAML()
	require.NoError(t, err)
	require.NoError(t, yaml.Unmarshal(bs, fakeVPC))
	fakepls := &v1alpha1.PrivateLinkService{}
	bs, err = testdata.LoadEKSPrivateLinkService()
	require.NoError(t, err)
	assert.NoError(t, yaml.Unmarshal(bs, fakepls))
	cli := fake.NewClientBuilder().WithScheme(s).WithObjects(fakeVPC, fakepls).Build()
	c := &task.Context{
		Local:                k8s.NewMutClientWithOwner(cli, fakeVPC.DeepCopy(), namer.NewNamer()).Client(),
		Logger:               logr.Discard(),
		VPC:                  fakeVPC.DeepCopy(),
		VPCR:                 resvpc.NewVPCReader(cli),
		Config:               config.RootConfig{},
		PrivateLinkService:   fakepls.DeepCopy(),
		PrivateLinkServiceRW: respls.NewPrivateLinkServiceReaderWriter(cli),
		ClusterCli:           nil,
	}
	c, err = c.Default()
	assert.Nil(t, err)
	return c
}

func TestEndpointService_Create(t *testing.T) {
	t.Run("create endpointservice failed", func(t *testing.T) {
		task := NewEndpointService(fakeTaskContext(t), nil)
		ctl := gomock.NewController(t)
		k8scli := kubemock.NewMockClient(ctl)
		task.Local = k8scli
		k8scli.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(apierrors.NewNotFound(schema.GroupResource{}, "not found"))
		k8scli.EXPECT().Create(gomock.Any(), gomock.Any(), gomock.Any()).Return(assert.AnError)
		result := task.Create(context.Background())
		res, err, retriableErr := deploy.MinRequeue([]deploy.Result{result})
		assert.Nil(t, err)
		assert.ErrorContains(t, retriableErr, "assert.AnError")
		assert.Equal(t, time.Second*30, res.RequeueAfter)
	})
	t.Run("create endpointservice success", func(t *testing.T) {
		task := NewEndpointService(fakeTaskContext(t), nil)
		tags := cloudv1.TagsFromLabels(task.PrivateLinkService.Labels)
		tags = util.AddResourceName(tags, task.PrivateLinkService.Name)

		ctl := gomock.NewController(t)
		k8scli := kubemock.NewMockClient(ctl)
		task.Local = k8scli
		k8scli.EXPECT().Get(gomock.Any(), types.NamespacedName{
			Name: task.endpointServiceName(),
		}, gomock.Any()).Return(apierrors.NewNotFound(schema.GroupResource{}, "not found"))
		k8scli.EXPECT().Create(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, obj runtime.Object, opts ...client.CreateOption) error {
			ves, ok := obj.(*ec2v1alpha1.VPCEndpointServiceConfiguration)
			if !ok {
				return assert.AnError
			}
			assert.Equal(t, task.endpointServiceName(), ves.Name)
			// assert.Equal(t, labels.NewOwnerLabels(task.PrivateLinkService, nil), ves.Labels)
			assert.Equal(t, task.ProviderConfigName(), ves.Spec.ProviderConfigReference.Name)
			assert.Equal(t, task.PrivateLinkService.Spec.Region, ves.Spec.ForProvider.Region)
			assert.Equal(t, pointer.Bool(true), ves.Spec.ForProvider.AcceptanceRequired)
			assert.Equal(t, pointer.String(task.privateDNSName()), ves.Spec.ForProvider.PrivateDNSName)
			assert.Equal(t, []*ec2v1alpha1.TagSpecification{
				{
					ResourceType: pointer.String("vpc-endpoint-service"),
					Tags:         util.GetEC2V1AlphaTags(tags),
				},
			}, ves.Spec.ForProvider.TagSpecifications)
			return nil
		})
		// k8scli.EXPECT().Patch(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
		result := task.Create(context.Background())
		assert.Equal(t, deploy.Done(), result)
	})
}

func TestEndpointService_PostCreateReconcilesDualstackAfterSettingPrincipals(t *testing.T) {
	c := fakeTaskContext(t)
	c.PrivateLinkService.Spec.TargetIPAddressType = v1alpha1.IPAddressTypeDualstack
	c.PrivateLinkService.Status.AllowedPrincipals = nil

	awsClient := &fakeEndpointServiceAWS{addressType: v1alpha1.IPAddressTypeIPv4}
	task := NewEndpointService(c, nil)
	task.ec2Cli = awsClient
	ves := endpointServiceCR()
	ves.Name = task.endpointServiceName()
	ves.Status.SetConditions(xpv1.Available())
	ves.Status.AtProvider.ServiceConfiguration.ServiceName = awssdk.String("com.amazonaws.vpce.us-west-2.vpce-svc-1")
	ves.Status.AtProvider.ServiceConfiguration.PrivateDNSName = awssdk.String("privatelink.example.com")

	ctl := gomock.NewController(t)
	k8scli := kubemock.NewMockClient(ctl)
	k8scli.EXPECT().Get(gomock.Any(), types.NamespacedName{Name: task.endpointServiceName()}, gomock.Any()).DoAndReturn(
		func(_ context.Context, _ types.NamespacedName, obj client.Object, _ ...client.GetOption) error {
			*obj.(*ec2v1alpha1.VPCEndpointServiceConfiguration) = *ves.DeepCopy()
			return nil
		},
	)
	task.Local = k8scli

	plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
	plsrw.EXPECT().UpdateStatus(gomock.Any(), task.PrivateLinkService).Return(nil).Times(2)
	task.PrivateLinkServiceRW = plsrw

	result := task.PostCreate(context.Background())

	assert.False(t, result.IsDoneState())
	_, err, retriableErr := deploy.MinRequeue([]deploy.Result{result})
	assert.NoError(t, err)
	require.Error(t, retriableErr)
	require.Len(t, awsClient.permissionInputs, 1)
	assert.Equal(t, []string{privateLinkAllowedPrincipalsAll}, awsClient.permissionInputs[0].AddAllowedPrincipals)
	assert.True(t, awsClient.modified)
	assert.Equal(t, []string{privateLinkAllowedPrincipalsAll}, task.PrivateLinkService.Status.AllowedPrincipals)
	assert.Equal(t, 1, awsClient.describeCalls)
}

func TestEndpointService_PostCreateDoesNotDelayLegacyIPv4AfterSettingPrincipals(t *testing.T) {
	c := fakeTaskContext(t)
	c.PrivateLinkService.Spec.TargetIPAddressType = ""
	c.PrivateLinkService.Status.AllowedPrincipals = nil

	awsClient := &fakeEndpointServiceAWS{addressType: v1alpha1.IPAddressTypeIPv4}
	task := NewEndpointService(c, nil)
	task.ec2Cli = awsClient
	ves := endpointServiceCR()
	ves.Name = task.endpointServiceName()
	ves.Status.SetConditions(xpv1.Available())
	ves.Status.AtProvider.ServiceConfiguration.ServiceName = awssdk.String("com.amazonaws.vpce.us-west-2.vpce-svc-1")
	ves.Status.AtProvider.ServiceConfiguration.PrivateDNSName = awssdk.String("privatelink.example.com")

	ctl := gomock.NewController(t)
	k8scli := kubemock.NewMockClient(ctl)
	k8scli.EXPECT().Get(gomock.Any(), types.NamespacedName{Name: task.endpointServiceName()}, gomock.Any()).DoAndReturn(
		func(_ context.Context, _ types.NamespacedName, obj client.Object, _ ...client.GetOption) error {
			*obj.(*ec2v1alpha1.VPCEndpointServiceConfiguration) = *ves.DeepCopy()
			return nil
		},
	)
	task.Local = k8scli

	plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
	plsrw.EXPECT().UpdateStatus(gomock.Any(), task.PrivateLinkService).Return(nil).Times(2)
	task.PrivateLinkServiceRW = plsrw

	result := task.PostCreate(context.Background())

	assert.True(t, result.IsDoneState())
	require.Len(t, awsClient.permissionInputs, 1)
	assert.Zero(t, awsClient.describeCalls)
	assert.False(t, awsClient.modified)
}

func TestEndpointServiceManagedAddressType(t *testing.T) {
	ipv4 := v1alpha1.IPAddressTypeIPv4
	dualstack := v1alpha1.IPAddressTypeDualstack
	tests := []struct {
		name        string
		observed    v1alpha1.IPAddressType
		target      v1alpha1.IPAddressType
		legacy      bool
		nlbARNs     []string
		wantModify  bool
		wantReady   bool
		wantObserve bool
		wantAdd     []string
		wantRemove  []string
		connections []ec2types.VpcEndpointConnection
	}{
		{name: "enable dualstack", observed: ipv4, target: dualstack, wantModify: true, wantObserve: true, wantAdd: []string{"ipv6"}},
		{name: "unpaused legacy uses managed dualstack target", observed: ipv4, target: dualstack, legacy: true, nlbARNs: []string{"arn-1"}, wantModify: true, wantObserve: true, wantAdd: []string{"ipv6"}},
		{name: "unpaused legacy rejects mismatched imported NLB", observed: ipv4, target: dualstack, legacy: true, nlbARNs: []string{"arn-other"}, wantObserve: true},
		{name: "endpoint service readback is ready", observed: dualstack, target: dualstack, wantReady: true, wantObserve: true},
		{name: "disable dualstack", observed: dualstack, target: ipv4, wantModify: true, wantObserve: true, wantRemove: []string{"ipv6"}},
		{
			name: "disable waits for live IPv6 connection", observed: dualstack, target: ipv4, wantObserve: true,
			connections: []ec2types.VpcEndpointConnection{{
				VpcEndpointId: awssdk.String("vpce-ipv6"), VpcEndpointState: ec2types.StateAvailable,
				IpAddressType: ec2types.IpAddressType("ipv6"),
			}},
		},
		{
			name: "disable waits for deleting IPv6 connection", observed: dualstack, target: ipv4, wantObserve: true,
			connections: []ec2types.VpcEndpointConnection{{
				VpcEndpointId: awssdk.String("vpce-deleting"), VpcEndpointState: ec2types.StateDeleting,
				IpAddressType: ec2types.IpAddressType("dualstack"),
			}},
		},
		{
			name: "disable retains live IPv4 connection", observed: dualstack, target: ipv4, wantObserve: true,
			wantModify: true, wantRemove: []string{"ipv6"},
			connections: []ec2types.VpcEndpointConnection{{
				VpcEndpointId: awssdk.String("vpce-ipv4"), VpcEndpointState: ec2types.StateAvailable,
				IpAddressType: ec2types.IpAddressTypeIpv4,
			}},
		},
		{
			name: "disable ignores deleted connection", observed: dualstack, target: ipv4, wantModify: true, wantObserve: true,
			wantRemove: []string{"ipv6"},
			connections: []ec2types.VpcEndpointConnection{{
				VpcEndpointId: awssdk.String("vpce-deleted"), VpcEndpointState: ec2types.StateDeleted,
				IpAddressType: ec2types.IpAddressType("dualstack"),
			}},
		},
		{name: "managed ipv4 readback is ready", observed: ipv4, target: ipv4, wantReady: true, wantObserve: true},
		{name: "legacy address type does not observe endpoint service address type", observed: dualstack, wantReady: true},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			pls := &v1alpha1.PrivateLinkService{
				ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{}},
				Spec: v1alpha1.PrivateLinkServiceSpec{
					TargetIPAddressType: tt.target,
					LoadBalancers:       []v1alpha1.LoadBalancer{{ClusterID: "cluster-1", ID: "group-1"}},
				},
				Status: v1alpha1.PrivateLinkServiceStatus{
					LoadBalancers: []v1alpha1.LoadBalancerStatus{{ID: "group-1", CloudResourceId: "arn-1"}},
				},
			}
			if tt.legacy {
				pls.Labels[v1alpha1.IsLegacyKey] = "true"
			}
			ctx := &task.Context{PrivateLinkService: pls}
			awsClient := &fakeEndpointServiceAWS{
				addressType: tt.observed, connections: tt.connections, nlbARNs: tt.nlbARNs,
			}
			endpointService := NewEndpointService(ctx, nil)
			endpointService.ec2Cli = awsClient

			result := endpointService.reconcileManagedAddressType(context.Background(), endpointServiceCR())

			assert.Equal(t, tt.wantModify, awsClient.modified)
			assert.Equal(t, tt.wantReady, result.IsDoneState())
			wantDescribeCalls := 0
			if tt.wantObserve {
				wantDescribeCalls = 1
			}
			assert.Equal(t, wantDescribeCalls, awsClient.describeCalls)
			if tt.wantModify {
				require.Len(t, awsClient.modifyInputs, 1)
				assert.Equal(t, tt.wantAdd, awsClient.modifyInputs[0].AddSupportedIpAddressTypes)
				assert.Equal(t, tt.wantRemove, awsClient.modifyInputs[0].RemoveSupportedIpAddressTypes)
			}
		})
	}
}

func endpointServiceCR() *ec2v1alpha1.VPCEndpointServiceConfiguration {
	return &ec2v1alpha1.VPCEndpointServiceConfiguration{
		Status: ec2v1alpha1.VPCEndpointServiceConfigurationStatus{
			AtProvider: ec2v1alpha1.VPCEndpointServiceConfigurationObservation{
				ServiceConfiguration: &ec2v1alpha1.ServiceConfiguration{ServiceID: awssdk.String("vpce-svc-1")},
			},
		},
	}
}

type fakeEndpointServiceAWS struct {
	addressType      v1alpha1.IPAddressType
	describeCalls    int
	modified         bool
	modifyInputs     []*ec2.ModifyVpcEndpointServiceConfigurationInput
	permissionInputs []*ec2.ModifyVpcEndpointServicePermissionsInput
	connections      []ec2types.VpcEndpointConnection
	nlbARNs          []string
}

func (f *fakeEndpointServiceAWS) DescribeVpcEndpointConnections(
	context.Context,
	*ec2.DescribeVpcEndpointConnectionsInput,
	...func(*ec2.Options),
) (*ec2.DescribeVpcEndpointConnectionsOutput, error) {
	return &ec2.DescribeVpcEndpointConnectionsOutput{VpcEndpointConnections: f.connections}, nil
}

func (f *fakeEndpointServiceAWS) DescribeAvailabilityZones(
	context.Context,
	*ec2.DescribeAvailabilityZonesInput,
	...func(*ec2.Options),
) (*ec2.DescribeAvailabilityZonesOutput, error) {
	return &ec2.DescribeAvailabilityZonesOutput{}, nil
}

func (f *fakeEndpointServiceAWS) DescribeVpcEndpointServiceConfigurations(
	context.Context,
	*ec2.DescribeVpcEndpointServiceConfigurationsInput,
	...func(*ec2.Options),
) (*ec2.DescribeVpcEndpointServiceConfigurationsOutput, error) {
	f.describeCalls++
	types := []ec2types.ServiceConnectivityType{ec2types.ServiceConnectivityTypeIpv4}
	if f.addressType == v1alpha1.IPAddressTypeDualstack {
		types = append(types, ec2types.ServiceConnectivityTypeIpv6)
	}
	return &ec2.DescribeVpcEndpointServiceConfigurationsOutput{
		ServiceConfigurations: []ec2types.ServiceConfiguration{{
			ServiceId:               awssdk.String("vpce-svc-1"),
			ServiceState:            ec2types.ServiceStateAvailable,
			SupportedIpAddressTypes: types,
			NetworkLoadBalancerArns: append([]string(nil), f.nlbARNs...),
		}},
	}, nil
}

func (f *fakeEndpointServiceAWS) ModifyVpcEndpointServicePermissions(
	_ context.Context,
	input *ec2.ModifyVpcEndpointServicePermissionsInput,
	_ ...func(*ec2.Options),
) (*ec2.ModifyVpcEndpointServicePermissionsOutput, error) {
	f.permissionInputs = append(f.permissionInputs, input)
	return &ec2.ModifyVpcEndpointServicePermissionsOutput{ReturnValue: awssdk.Bool(true)}, nil
}

func (f *fakeEndpointServiceAWS) ModifyVpcEndpointServiceConfiguration(
	_ context.Context,
	input *ec2.ModifyVpcEndpointServiceConfigurationInput,
	_ ...func(*ec2.Options),
) (*ec2.ModifyVpcEndpointServiceConfigurationOutput, error) {
	f.modified = true
	f.modifyInputs = append(f.modifyInputs, input)
	return &ec2.ModifyVpcEndpointServiceConfigurationOutput{}, nil
}

func Test_privateDNSName(t *testing.T) {
	task := NewEndpointService(fakeTaskContext(t), nil)
	task.PrivateLinkService.Name = "test"
	assert.Equal(t, fmt.Sprintf("privatelink-%s.%s.%s", task.PrivateLinkService.Name, task.PrivateLinkService.Status.DNSIdentify, task.Config.PublicHostedZoneDomain), task.privateDNSName())

	task.PrivateLinkService.Name = "test.test"
	assert.Equal(t, fmt.Sprintf("privatelink-%s.%s.%s", task.PrivateLinkService.Name[:8], task.PrivateLinkService.Status.DNSIdentify, task.Config.PublicHostedZoneDomain), task.privateDNSName())
}