package ack

import (
	"context"
	"fmt"
	"strings"
	"time"

	privatelink "github.com/alibabacloud-go/privatelink-20200415/v5/client"
	"github.com/alibabacloud-go/tea/tea"
	aliecsv1alpha1 "github.com/crossplane-contrib/provider-upjet-alibabacloud/apis/ecs/v1alpha1"
	aliplv1alpha1 "github.com/crossplane-contrib/provider-upjet-alibabacloud/apis/privatelink/v1alpha1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/cloud/aliyun"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/privatelinkendpoint/task"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/deploy"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/k8s"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
)

const (
	securityGroupRuleTypeIngress = "ingress"
	securityGroupRuleTypeEgress  = "egress"
)

type endpoint struct {
	*task.BaseTask
	aliCli aliyun.Client
}

func NewEndpoint(ctx *task.Context, aliCli aliyun.Client) deploy.Task {
	return &endpoint{
		BaseTask: &task.BaseTask{Context: ctx},
		aliCli:   aliCli,
	}
}

func (t *endpoint) Name() string {
	return "Endpoint"
}

func (t *endpoint) Create(ctx context.Context) deploy.Result {
	tags := t.EffectiveCloudTags(consts.ProviderTypeACK)

	sg, err := t.createSecurityGroupAndRules(ctx, tags)
	if err != nil {
		return deploy.Panic(err).WithRequeuAfter(time.Second * 30)
	}

	ep, err := t.createVPCEndpoint(ctx, sg, tags)
	if err != nil {
		return deploy.Panic(err).WithRequeuAfter(time.Second * 30)
	}
	labels := util.MergeMaps(t.VPC.Labels, map[string]string{v1alpha1.PrivateLinkEndpointNameKey: t.PrivateLinkEndpoint.Name})
	if err := t.createVPCEndpointAssociations(ctx, ep, labels); err != nil {
		return deploy.Panic(err).WithRequeuAfter(time.Second * 30)
	}

	return deploy.Done()
}

func (t *endpoint) createSecurityGroupAndRules(ctx context.Context, tags map[string]string) (*aliecsv1alpha1.SecurityGroup, error) {
	sg, err := t.createSecurityGroup(ctx, util.DeepCopyMap(tags))
	if err != nil {
		return nil, err
	}

	labels := util.MergeMaps(t.VPC.Labels, map[string]string{v1alpha1.PrivateLinkEndpointNameKey: t.PrivateLinkEndpoint.Name})
	if err := t.createSecurityGroupRule(ctx, securityGroupRuleTypeIngress, sg, labels); err != nil {
		return nil, fmt.Errorf("create ingress security group rule error: %w", err)
	}
	if err := t.createSecurityGroupRule(ctx, securityGroupRuleTypeEgress, sg, labels); err != nil {
		return nil, fmt.Errorf("create egress security group rule error: %w", err)
	}
	return sg, nil
}

func (t *endpoint) createSecurityGroup(ctx context.Context, tags map[string]string) (*aliecsv1alpha1.SecurityGroup, error) {
	name := t.EndpointName()
	labels := util.MergeMaps(t.VPC.Labels, tags, map[string]string{v1alpha1.PrivateLinkEndpointNameKey: t.PrivateLinkEndpoint.Name})
	sg := &aliecsv1alpha1.SecurityGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name,
			Labels: labels,
		},
		Spec: aliecsv1alpha1.SecurityGroupSpec{
			ForProvider: aliecsv1alpha1.SecurityGroupParameters{
				Region:            ptr.To(t.PrivateLinkEndpoint.Spec.Region),
				SecurityGroupName: ptr.To(name),
				Description:       ptr.To("allow all security group"),
				VPCIDRef: &xpv1.Reference{
					Name: t.VPC.Status.VPC.Name,
				},
				Tags: util.ToMapPointer(tags),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: t.ProviderConfigName(),
				},
				ManagementPolicies: controllers.GetResourceManagementPolicies(),
			},
		},
	}
	if err := k8s.Apply(ctx, t.Local, sg); err != nil {
		return nil, err
	}
	return sg, nil
}

func (t *endpoint) createSecurityGroupRule(ctx context.Context, ruleType string, sg *aliecsv1alpha1.SecurityGroup, labels map[string]string) error {
	name := t.securityGroupRuleName(ruleType, sg.Name)
	sgRule := &aliecsv1alpha1.SecurityGroupRule{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name,
			Labels: labels,
		},
		Spec: aliecsv1alpha1.SecurityGroupRuleSpec{
			ForProvider: aliecsv1alpha1.SecurityGroupRuleParameters{
				Region: ptr.To(t.PrivateLinkEndpoint.Spec.Region),
				SecurityGroupIDRef: &xpv1.Reference{
					Name: sg.Name,
				},
				IPProtocol:  ptr.To("all"),
				Description: ptr.To("allow all"),
				PortRange:   ptr.To("1/65535"),
				Type:        ptr.To(ruleType),
				CidrIP:      ptr.To("0.0.0.0/0"),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: t.ProviderConfigName(),
				},
				ManagementPolicies: controllers.GetResourceManagementPolicies(),
			},
		},
	}
	return k8s.Apply(ctx, t.Local, sgRule)
}

func (t *endpoint) createVPCEndpoint(ctx context.Context, sg *aliecsv1alpha1.SecurityGroup, tags map[string]string) (*aliplv1alpha1.VpcEndpoint, error) {
	name := t.EndpointName()
	labels := util.MergeMaps(t.VPC.Labels, tags, map[string]string{v1alpha1.PrivateLinkEndpointNameKey: t.PrivateLinkEndpoint.Name})
	ep := &aliplv1alpha1.VpcEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name,
			Labels: labels,
		},
		Spec: aliplv1alpha1.VpcEndpointSpec{
			ForProvider: aliplv1alpha1.VpcEndpointParameters{
				VPCEndpointName:           ptr.To(name),
				Region:                    ptr.To(t.PrivateLinkEndpoint.Spec.Region),
				ServiceName:               ptr.To(t.PrivateLinkEndpoint.Spec.ServiceName),
				EndpointType:              ptr.To("Interface"),
				ZonePrivateIPAddressCount: ptr.To(float64(1)),
				VPCIDRef: &xpv1.Reference{
					Name: t.VPC.Status.VPC.Name,
				},
				SecurityGroupRefs: []xpv1.Reference{
					{Name: sg.Name},
				},
				Tags: util.ToMapPointer(tags),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: t.ProviderConfigName(),
				},
				ManagementPolicies: controllers.GetResourceManagementPolicies(),
			},
		},
	}
	if err := k8s.Apply(ctx, t.Local, ep); err != nil {
		return nil, err
	}
	return ep, nil
}

func (t *endpoint) createVPCEndpointAssociations(ctx context.Context, ep *aliplv1alpha1.VpcEndpoint, labels map[string]string) error {
	subnets, err := t.getMatchedSubnetsByEndpointZones()
	if err != nil {
		return fmt.Errorf("get endpoint subnets error: %w", err)
	}
	for _, subnet := range subnets {
		if err := t.createEndpointSubnetAssociation(ctx, ep, subnet, labels); err != nil {
			return fmt.Errorf("associate endpoint with subnet %s error: %w", subnet, err)
		}
	}
	return nil
}

func (t *endpoint) createEndpointSubnetAssociation(ctx context.Context, ep *aliplv1alpha1.VpcEndpoint, subnet *v1alpha1.Subnet, labels map[string]string) error {
	association := &aliplv1alpha1.VpcEndpointZone{
		ObjectMeta: metav1.ObjectMeta{
			Name:   t.endpointSubnetAssociationName(ep.Name, subnet.Name),
			Labels: labels,
		},
		Spec: aliplv1alpha1.VpcEndpointZoneSpec{
			ForProvider: aliplv1alpha1.VpcEndpointZoneParameters{
				Region: ptr.To(t.PrivateLinkEndpoint.Spec.Region),
				ZoneID: ptr.To(subnet.AvailabilityZoneID),
				VswitchIDRef: &xpv1.Reference{
					Name: subnet.Name,
				},
				EndpointIDRef: &xpv1.Reference{
					Name: ep.Name,
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: t.ProviderConfigName(),
				},
				ManagementPolicies: controllers.GetResourceManagementPolicies(),
			},
		},
	}
	return k8s.Apply(ctx, t.Local, association)
}

func (t *endpoint) getMatchedSubnetsByEndpointZones() ([]*v1alpha1.Subnet, error) {
	subnets, err := t.endpointSubnets()
	if err != nil {
		return nil, err
	}
	az2subnets := make(map[string]*v1alpha1.Subnet)
	for i := range subnets {
		subnet := &subnets[i]
		az2subnets[subnet.AvailabilityZoneID] = subnet
	}

	endpointSubnets := make([]*v1alpha1.Subnet, 0, len(t.PrivateLinkEndpoint.Spec.Zones))
	for _, azID := range t.PrivateLinkEndpoint.Spec.Zones {
		// Every requested service zone needs a consumer vSwitch in the same zone.
		subnet, ok := az2subnets[azID]
		if !ok {
			if t.usePrivateLinkEndpointSubnets() {
				return nil, fmt.Errorf("az %s is not found in private link endpoint subnets", azID)
			}
			return nil, fmt.Errorf("az %s is not found in subnets", azID)
		}
		endpointSubnets = append(endpointSubnets, subnet)
	}
	return endpointSubnets, nil
}

func (t *endpoint) endpointSubnets() ([]v1alpha1.Subnet, error) {
	if t.usePrivateLinkEndpointSubnets() {
		if len(t.VPC.Status.PrivateLinkEndpointSubnets) == 0 {
			return nil, fmt.Errorf("private link endpoint subnets are not ready")
		}
		return t.VPC.Status.PrivateLinkEndpointSubnets, nil
	}
	return t.VPC.Status.NodeSubnets, nil
}

func (t *endpoint) usePrivateLinkEndpointSubnets() bool {
	return util.ShouldUsePrivateLinkEndpointSubnets(t.VPC.Spec.PrivateLinkEndpointZones, t.VPC.Spec.NodeZones)
}

func (t *endpoint) PostCreate(ctx context.Context) deploy.Result {
	t.PrivateLinkEndpoint.Status.Phase = v1alpha1.PrivateLinkEndpointPhaseCreating

	sg, sgRules, err := t.listSecurityGroupCRs(ctx)
	if err != nil {
		return deploy.Panic(err).WithRequeuAfter(time.Second * 30)
	}
	if !controllers.IsResourceReady(sg.Status.ResourceStatus) {
		return deploy.Error(fmt.Errorf("security group is not ready"))
	}
	for _, sgRule := range sgRules {
		if !controllers.IsResourceReady(sgRule.Status.ResourceStatus) {
			return deploy.Error(fmt.Errorf("security group rule %s is not ready", sgRule.Name))
		}
	}

	ep, epZones, err := t.listVPCEndpointCRs(ctx)
	if err != nil {
		return deploy.Panic(err).WithRequeuAfter(time.Second * 30)
	}
	if !controllers.IsResourceReady(ep.Status.ResourceStatus) {
		return deploy.Error(fmt.Errorf("vpc endpoint is not ready"))
	}
	for _, epZone := range epZones {
		if !controllers.IsResourceReady(epZone.Status.ResourceStatus) {
			return deploy.Error(fmt.Errorf("vpc endpoint zone %s is not ready", epZone.Name))
		}
	}

	if err := t.syncEndpointInfo(ep); err != nil {
		return deploy.Error(err)
	}
	if t.PrivateLinkEndpoint.Status.Phase != v1alpha1.PrivateLinkEndpointPhaseEndpointAvailable {
		return deploy.Error(fmt.Errorf("private link endpoint is not available"))
	}
	return deploy.Done()
}

func (t *endpoint) syncEndpointInfo(ep *aliplv1alpha1.VpcEndpoint) error {
	t.PrivateLinkEndpoint.Status.ID = tea.StringValue(ep.Status.AtProvider.ID)
	t.PrivateLinkEndpoint.Status.EndpointAddress = tea.StringValue(ep.Status.AtProvider.EndpointDomain)
	t.syncEndpointStatus(tea.StringValue(ep.Status.AtProvider.Status), tea.StringValue(ep.Status.AtProvider.ConnectionStatus))

	// crossplane take a long time to refresh the status, try to use aws SDK to get the detail ASAP
	if t.PrivateLinkEndpoint.Status.Phase != v1alpha1.PrivateLinkEndpointPhaseEndpointAvailable {
		return t.syncEndpointInfoBySDK()
	}
	return nil
}

func (t *endpoint) syncEndpointInfoBySDK() error {
	assumedRole := &aliyun.AssumedRole{
		RoleARN: fmt.Sprintf("acs:ram::%s:role/%s", t.VPC.Spec.AccountRef.AccountID, t.Config.DataPlaneRoleName),
	}
	cli, err := t.aliCli.PrivateLink(assumedRole, aliyun.WithRegion(t.VPC.Spec.Region))
	if err != nil {
		return fmt.Errorf("init alicloud private link client error: %w", err)
	}
	req := &privatelink.GetVpcEndpointAttributeRequest{
		EndpointId: tea.String(t.PrivateLinkEndpoint.Status.ID),
		RegionId:   tea.String(t.PrivateLinkEndpoint.Spec.Region),
	}
	result, err := cli.GetVpcEndpointAttribute(req)
	if err != nil {
		return fmt.Errorf("get vpc endpoint %s attribute error: %w", t.PrivateLinkEndpoint.Status.ID, err)
	}

	t.PrivateLinkEndpoint.Status.EndpointAddress = tea.StringValue(result.Body.EndpointDomain)
	t.syncEndpointStatus(tea.StringValue(result.Body.EndpointStatus), tea.StringValue(result.Body.ConnectionStatus))
	return nil
}

func (t *endpoint) syncEndpointStatus(epStatus, connStatus string) {
	switch strings.ToLower(epStatus) {
	case "pending":
		t.PrivateLinkEndpoint.Status.Phase = v1alpha1.PrivateLinkEndpointPhasePending
	case "creating":
		t.PrivateLinkEndpoint.Status.Phase = v1alpha1.PrivateLinkEndpointPhaseCreating
	case "active":
		t.syncEndpointConnectionStatus(connStatus)
	case "deleting":
		t.PrivateLinkEndpoint.Status.Phase = v1alpha1.PrivateLinkEndpointPhaseDeleting
	}
}

func (t *endpoint) syncEndpointConnectionStatus(connStatus string) {
	switch strings.ToLower(connStatus) {
	case "pending", "disconnecting", "disconnected":
		t.PrivateLinkEndpoint.Status.Phase = v1alpha1.PrivateLinkEndpointPhasePendingAcceptance
	case "connecting":
		t.PrivateLinkEndpoint.Status.Phase = v1alpha1.PrivateLinkEndpointPhaseCreating
	case "connected":
		t.PrivateLinkEndpoint.Status.Phase = v1alpha1.PrivateLinkEndpointPhaseEndpointAvailable
	case "deleting":
		t.PrivateLinkEndpoint.Status.Phase = v1alpha1.PrivateLinkEndpointPhaseDeleting
	case "servicedeleted":
		t.PrivateLinkEndpoint.Status.Phase = v1alpha1.PrivateLinkEndpointPhaseFailed
	}
}

func (t *endpoint) listSecurityGroupCRs(ctx context.Context) (*aliecsv1alpha1.SecurityGroup, []aliecsv1alpha1.SecurityGroupRule, error) {
	sg := &aliecsv1alpha1.SecurityGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.EndpointName(),
		},
	}
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(sg), sg); err != nil {
		return nil, nil, err
	}

	labelOpt := client.MatchingLabels{
		v1alpha1.PrivateLinkEndpointNameKey: t.PrivateLinkEndpoint.Name,
	}
	sgRules := &aliecsv1alpha1.SecurityGroupRuleList{}
	if err := t.Local.List(ctx, sgRules, labelOpt); err != nil {
		return nil, nil, err
	}
	return sg, sgRules.Items, nil
}

func (t *endpoint) listVPCEndpointCRs(ctx context.Context) (*aliplv1alpha1.VpcEndpoint, []aliplv1alpha1.VpcEndpointZone, error) {
	ep := &aliplv1alpha1.VpcEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.EndpointName(),
		},
	}
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(ep), ep); err != nil {
		return nil, nil, err
	}

	labelOpt := client.MatchingLabels{
		v1alpha1.PrivateLinkEndpointNameKey: t.PrivateLinkEndpoint.Name,
	}
	epZones := &aliplv1alpha1.VpcEndpointZoneList{}
	if err := t.Local.List(ctx, epZones, labelOpt); err != nil {
		return nil, nil, err
	}
	return ep, epZones.Items, nil
}

func (t *endpoint) Delete(ctx context.Context) deploy.Result {
	labelOpt := client.MatchingLabels{
		v1alpha1.PrivateLinkEndpointNameKey: t.PrivateLinkEndpoint.Name,
	}
	epZones := &aliplv1alpha1.VpcEndpointZoneList{}
	if err := t.Local.List(ctx, epZones, labelOpt); err != nil {
		return deploy.Panic(err).WithRequeuAfter(time.Second * 30)
	}
	for i := range epZones.Items {
		if err := controllers.DeleteUpboundResource(ctx, t.Local, &epZones.Items[i]); err != nil {
			return deploy.Panic(err).WithRequeuAfter(time.Second * 30)
		}
	}

	sgRules := &aliecsv1alpha1.SecurityGroupRuleList{}
	if err := t.Local.List(ctx, sgRules, labelOpt); err != nil {
		return deploy.Panic(err).WithRequeuAfter(time.Second * 30)
	}
	for i := range sgRules.Items {
		if err := controllers.DeleteUpboundResource(ctx, t.Local, &sgRules.Items[i]); err != nil {
			return deploy.Panic(err).WithRequeuAfter(time.Second * 30)
		}
	}

	ep := &aliplv1alpha1.VpcEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.EndpointName(),
		},
	}
	if err := controllers.DeleteUpboundResource(ctx, t.Local, ep); err != nil {
		return deploy.Panic(err).WithRequeuAfter(time.Second * 30)
	}

	sg := &aliecsv1alpha1.SecurityGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.EndpointName(),
		},
	}
	if err := controllers.DeleteUpboundResource(ctx, t.Local, sg); err != nil {
		return deploy.Panic(err).WithRequeuAfter(time.Second * 30)
	}
	return deploy.Done()
}

func (t *endpoint) PostDelete(ctx context.Context) deploy.Result {
	labelOpt := client.MatchingLabels{
		v1alpha1.PrivateLinkEndpointNameKey: t.PrivateLinkEndpoint.Name,
	}
	epZones := &aliplv1alpha1.VpcEndpointZoneList{}
	if err := t.Local.List(ctx, epZones, labelOpt); err != nil {
		return deploy.Error(err).WithRequeuAfter(time.Second * 30)
	}
	if len(epZones.Items) > 0 {
		return deploy.Error(fmt.Errorf("vpc endpoint zones still exist")).WithRequeuAfter(time.Second * 30)
	}

	sgRules := &aliecsv1alpha1.SecurityGroupRuleList{}
	if err := t.Local.List(ctx, sgRules, labelOpt); err != nil {
		return deploy.Error(err).WithRequeuAfter(time.Second * 30)
	}
	if len(sgRules.Items) > 0 {
		return deploy.Error(fmt.Errorf("security group rules still exist")).WithRequeuAfter(time.Second * 30)
	}

	ep := &aliplv1alpha1.VpcEndpoint{}
	if err := k8s.GetDeletedObject(ctx, t.Local, types.NamespacedName{Name: t.EndpointName()}, ep); err != nil {
		return deploy.Error(err).WithRequeuAfter(time.Second * 30)
	}

	sg := &aliecsv1alpha1.SecurityGroup{}
	if err := k8s.GetDeletedObject(ctx, t.Local, types.NamespacedName{Name: t.EndpointName()}, sg); err != nil {
		return deploy.Error(err).WithRequeuAfter(time.Second * 30)
	}
	return deploy.Done()
}

func (t *endpoint) securityGroupRuleName(ruleType, sgName string) string {
	return fmt.Sprintf("%s-%s", ruleType, sgName)
}

func (t *endpoint) endpointSubnetAssociationName(epName, subnetName string) string {
	return fmt.Sprintf("%s-%s", epName, subnetName)
}