package aks

import (
	"context"
	"fmt"
	"time"

	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/containerservice/armcontainerservice/v6"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	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/util"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/azure"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/cidr"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/appbox/task"
	"github.com/upbound/provider-azure/apis/containerservice/v1beta1"
	networkv1beta1 "github.com/upbound/provider-azure/apis/network/v1beta1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

type AKS struct {
	*task.BaseTask
}

const (
	AKSNameLimit             = 40
	DefaultNodeMaxSize       = 10
	DefaultNodeMinSize       = 2
	DefaultInstanceType      = "Standard_DS2_v2"
	NetworkPlugin            = "azure"
	IdentityType             = "SystemAssigned"
	UserAssignedIdentityType = "UserAssigned"
	// defaultRequeuInterval is the default requeue interval for the task.
	defaultRequeuInterval   = time.Second * 30
	adminNodePoolName       = "adminpool"
	nodeResourceGroupSuffix = "-node-rg"
	userAssignedNATGateway  = "userAssignedNATGateway"
)

func NewAKS(ctx *task.Context) *AKS {
	return &AKS{
		BaseTask: &task.BaseTask{Context: ctx},
	}
}

func (t *AKS) Name() string {
	return "AKS"
}

func (t *AKS) Create(ctx context.Context) deploy.Result {
	t.BeginAt = util.Pointer(time.Now())
	aksName := t.getAKSName()
	tags := cloudv1.TagsFromLabels(t.APPBox.Labels)
	tags = util.AddResourceName(tags, aksName)
	existingAKS, existingAKSFound, err := getExistingAKS(ctx, t.Local, aksName)
	if err != nil {
		return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
	}
	enableAPIServerVNetIntegration := shouldEnableAPIServerVNetIntegration(t.APPBox, existingAKS, existingAKSFound)

	adminNodeConfig := t.APPBox.Spec.KubernetesConfig.AdminNodeConfig
	instanceType := DefaultInstanceType
	if adminNodeConfig.InstanceType != "" {
		instanceType = adminNodeConfig.InstanceType
	}
	nodeMaxSize := DefaultNodeMaxSize
	if adminNodeConfig.MaxSize != nil {
		nodeMaxSize = int(*adminNodeConfig.MaxSize)
	}
	nodeMinSize := DefaultNodeMinSize
	if adminNodeConfig.MinSize != nil {
		nodeMinSize = int(*adminNodeConfig.MinSize)
	}
	cidrDispatcher, err := cidr.NewAKSCIDRDispatcher(t.VPC.Spec.CIDR)
	if err != nil {
		return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
	}
	if err := cidrDispatcher.Allocate(); err != nil {
		return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
	}

	// if no subnet was found, `t.getSubnet` will return an error.
	nodeSubnets, err := t.getSubnet(ctx, v1alpha1.NodeSubnetType)
	if err != nil {
		return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
	}
	nodeSubnet := nodeSubnets.Items[0]

	podSubnets, err := t.getSubnet(ctx, v1alpha1.PodSubnetType)
	if err != nil {
		return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
	}
	podSubnet := podSubnets.Items[0]
	var apiServerSubnet *networkv1beta1.Subnet
	if enableAPIServerVNetIntegration {
		apiServerSubnets, err := t.getSubnet(ctx, v1alpha1.APIServerSubnetType)
		if err != nil {
			return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
		}
		apiServerSubnet = &apiServerSubnets.Items[0]
	}
	aks := &v1beta1.KubernetesCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: aksName,
			Labels: map[string]string{
				v1alpha1.APPBoxNameKey: t.APPBox.Name,
			},
		},
		Spec: v1beta1.KubernetesClusterSpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: t.ProviderConfigName(),
				},
			},
			ForProvider: v1beta1.KubernetesClusterParameters{
				KubernetesVersion:       ptr.To(t.APPBox.Spec.KubernetesConfig.Version),
				Location:                ptr.To(t.APPBox.Spec.Region),
				OidcIssuerEnabled:       ptr.To(true),
				WorkloadIdentityEnabled: ptr.To(true),
				ResourceGroupName:       ptr.To(controllers.GetResourceGroupName(t.VPC)),
				// NodeResourceGroupName will autogenerated syntax of MC_resourceGroupName_resourceName_AzureRegion must be no greater than 80
				// So we need to set it customly to avoid the name too long
				// https://learn.microsoft.com/en-us/troubleshoot/azure/azure-kubernetes/create-upgrade-delete/aks-common-issues-faq
				NodeResourceGroup:    ptr.To(t.getNodeResourceGroupName()),
				NodeOsChannelUpgrade: ptr.To("None"),
				Identity:             []v1beta1.IdentityParameters{{Type: ptr.To(IdentityType)}},
				NetworkProfile: []v1beta1.NetworkProfileParameters{
					{
						NetworkPlugin: ptr.To(NetworkPlugin),
						ServiceCidr:   ptr.To(cidrDispatcher.ServiceIPv4CIDRBlock()),
						DNSServiceIP:  ptr.To(cidrDispatcher.DNSServiceIP()),
						OutboundType:  ptr.To(userAssignedNATGateway),
					},
				},
				DNSPrefix: ptr.To(aksName),
				AutoScalerProfile: []v1beta1.AutoScalerProfileParameters{
					{
						BalanceSimilarNodeGroups: ptr.To(true),
					},
				},
				DefaultNodePool: []v1beta1.DefaultNodePoolParameters{
					{
						EnableAutoScaling: ptr.To(true),
						MaxCount:          ptr.To(float64(nodeMaxSize)),
						MinCount:          ptr.To(float64(nodeMinSize)),
						// use the aks name as the admin node pool name
						OsSku:           ptr.To("AzureLinux"),
						Name:            ptr.To(adminNodePoolName),
						VMSize:          ptr.To(instanceType),
						VnetSubnetIDRef: &xpv1.Reference{Name: nodeSubnet.Name},
						PodSubnetIDRef:  &xpv1.Reference{Name: podSubnet.Name},
						UpgradeSettings: []v1beta1.UpgradeSettingsParameters{
							{
								// if it is not set, the default value is 10%, and aks will be frequently update by crossplane
								// because it is not set by crossplane lateinitialize
								// https://github.com/crossplane-contrib/provider-upjet-azure/issues/686
								MaxSurge: ptr.To(string("10%")),
							},
						},
						Tags: util.ToMapPointer(tags),
					},
				},
				SkuTier: ptr.To("Standard"),
				Tags:    util.ToMapPointer(tags),
				AzureActiveDirectoryRoleBasedAccessControl: []v1beta1.AzureActiveDirectoryRoleBasedAccessControlParameters{
					{
						Managed: ptr.To(true),
					},
				},
			},
		},
	}
	adminGroupIDs := []*string{}
	for _, id := range t.Config.APPBoxConfig.AdminGroupObjectIDs {
		adminGroupIDs = append(adminGroupIDs, ptr.To(id))
	}
	if len(adminGroupIDs) > 0 {
		aks.Spec.ForProvider.AzureActiveDirectoryRoleBasedAccessControl[0].AdminGroupObjectIds = adminGroupIDs
	}
	if enableAPIServerVNetIntegration {
		controlPlaneIdentityResourceID, err := aksControlPlaneManagedIdentityResourceID(t.Context)
		if err != nil {
			return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
		}
		aks.Spec.ForProvider.Identity = []v1beta1.IdentityParameters{
			{
				Type:        ptr.To(UserAssignedIdentityType),
				IdentityIds: []*string{ptr.To(controlPlaneIdentityResourceID)},
			},
		}
		util.SetAnnotation(aks, v1alpha1.AnnotationKeyAKSAPIServerVNetIntegrationRequested, "true")
	}

	accessProfile := v1beta1.APIServerAccessProfileParameters{}
	hasAccessProfile := false
	if enableAPIServerVNetIntegration {
		accessProfile.SubnetIDRef = &xpv1.Reference{Name: apiServerSubnet.Name}
		accessProfile.VnetIntegrationEnabled = ptr.To(true)
		hasAccessProfile = true
	}
	allowCidrs := t.Config.APPBoxConfig.ApiserverPublicAccessCidrs
	if len(allowCidrs) > 0 && !utils.IsIncludeString(allowCidrs, "0.0.0.0/0") {
		internet, err := t.getActiveInternet()
		if err != nil {
			return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
		}
		ips, err := t.getNATIPs(internet)
		if err != nil {
			return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
		}
		allowCidrs = append(append([]string(nil), allowCidrs...), ips...)
	}
	if len(allowCidrs) > 0 {
		accessProfile.AuthorizedIPRanges = util.ToSlicePointer(allowCidrs)
		hasAccessProfile = true
	}
	if hasAccessProfile {
		aks.Spec.ForProvider.APIServerAccessProfile = []v1beta1.APIServerAccessProfileParameters{accessProfile}
	}

	if err := k8s.Apply(ctx, t.Local, aks); err != nil {
		return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
	}

	return deploy.Done()
}

func getExistingAKS(ctx context.Context, cli client.Client, name string) (*v1beta1.KubernetesCluster, bool, error) {
	cluster := &v1beta1.KubernetesCluster{}
	if err := cli.Get(ctx, types.NamespacedName{Name: name}, cluster); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, false, nil
		}
		return nil, false, err
	}
	return cluster, true, nil
}

func shouldEnableAPIServerVNetIntegration(appBox *v1alpha1.APPBox, cluster *v1beta1.KubernetesCluster, clusterExists bool) bool {
	if !clusterExists {
		return true
	}
	if !isAKSProvisioned(cluster) {
		return true
	}
	if appBox.GetAnnotations()[v1alpha1.AnnotationKeyEnableAKSAPIServerVNetIntegration] == "true" {
		return true
	}
	if isAKSAPIServerVNetIntegrationObservedEnabled(cluster) {
		return true
	}
	if cluster.GetAnnotations()[v1alpha1.AnnotationKeyAKSAPIServerVNetIntegrationRequested] == "true" {
		return true
	}
	return false
}

func isAKSProvisioned(cluster *v1beta1.KubernetesCluster) bool {
	return cluster != nil && ptr.Deref(cluster.Status.AtProvider.ID, "") != ""
}

func isAKSAPIServerVNetIntegrationObservedEnabled(cluster *v1beta1.KubernetesCluster) bool {
	if cluster == nil {
		return false
	}
	for _, profile := range cluster.Status.AtProvider.APIServerAccessProfile {
		if ptr.Deref(profile.VnetIntegrationEnabled, false) {
			return true
		}
	}
	return false
}

func (t *AKS) getNATIPs(internet *v1alpha1.Internet) ([]string, error) {
	if len(internet.Status.ExternalIPs) == 0 {
		return nil, fmt.Errorf("no external ip found in internet %s", internet.Name)
	}
	ips := make([]string, 0, len(internet.Status.ExternalIPs))
	for _, ip := range internet.Status.ExternalIPs {
		ips = append(ips, fmt.Sprintf("%s/32", ip))
	}
	return ips, nil
}

func (t *AKS) getActiveInternet() (*v1alpha1.Internet, error) {
	internets := &v1alpha1.InternetList{}
	ops := &client.ListOptions{
		LabelSelector: labels.SelectorFromSet(map[string]string{
			v1alpha1.VPCNameKey: t.VPC.Name,
		}),
	}
	if err := t.InternetR.List(context.Background(), internets, ops); err != nil {
		return nil, err
	}
	if len(internets.Items) == 0 {
		return nil, fmt.Errorf("no internet found for vpc %s", t.VPC.Name)
	}
	for _, internet := range internets.Items {
		if internet.Status.Phase == v1alpha1.InternetPhaseActive {
			return ptr.To(internet), nil
		}
	}
	return nil, fmt.Errorf("no active internet found for vpc %s", t.VPC.Name)
}

func (t *AKS) getAKSName() string {
	return aksClusterName(t.APPBox)
}

func (t *AKS) getNodeResourceGroupName() string {
	return fmt.Sprintf("%s%s", t.getAKSName(), nodeResourceGroupSuffix)
}

func (t *AKS) getSubnet(ctx context.Context, subnetType string) (*networkv1beta1.SubnetList, error) {
	subnets := &networkv1beta1.SubnetList{}
	if err := t.Local.List(ctx, subnets, &client.ListOptions{
		LabelSelector: labels.SelectorFromSet(map[string]string{
			v1alpha1.VPCNameKey:    t.VPC.Name,
			v1alpha1.SubnetTypeKey: subnetType,
		}),
	}); err != nil {
		return nil, err
	}
	if len(subnets.Items) == 0 {
		return nil, fmt.Errorf("no subnet found for vpc %s and type %s", t.VPC.Name, subnetType)
	}
	return subnets, nil
}

func (t *AKS) PostCreate(ctx context.Context) (res deploy.Result) {
	key := types.NamespacedName{Name: t.getAKSName()}
	aks := &v1beta1.KubernetesCluster{}
	err := t.Local.Get(ctx, key, aks)
	if err != nil {
		return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
	}
	t.Cluster = task.NewAKSClusterIdentifier(aks)
	t.APPBox.Status.KubernetesClusterName = aks.Name

	condition := aks.Status.GetCondition(xpv1.TypeReady)
	if condition.Status != corev1.ConditionTrue {
		t.APPBox.Status.Conditions.SetCondition(metav1.Condition{
			Type:               v1alpha1.APPBoxConditionK8SReady,
			Status:             metav1.ConditionFalse,
			Reason:             "AKSCreating",
			Message:            condition.Message,
			LastTransitionTime: metav1.Now(),
		})
		if err := t.APPBoxRW.UpdateStatus(ctx, t.APPBox); err != nil {
			t.Logger.Error(err, "failed to update appbox status")
		}
		return deploy.Panic(fmt.Errorf("AKS Cluster is creating")).WithRequeuAfter(defaultRequeuInterval)
	}

	t.APPBox.Status.Conditions.SetCondition(metav1.Condition{
		Type:               v1alpha1.APPBoxConditionK8SReady,
		Status:             metav1.ConditionTrue,
		Reason:             "AKSReady",
		Message:            "AKS Creation succeeded",
		LastTransitionTime: metav1.Now(),
	})

	t.AKSCluster = aks

	secret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      t.getAKSName(),
			Namespace: t.APPBox.Namespace,
			Labels: map[string]string{
				v1alpha1.APPBoxNameKey: t.APPBox.Name,
			},
		},
	}
	if err := t.ensureKubeconfigSecret(ctx, secret, aks); err != nil {
		return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
	}

	t.APPBox.Status.KubeConfigSecretRef = &corev1.SecretReference{
		Name:      secret.Name,
		Namespace: secret.Namespace,
	}

	return deploy.PanicOrDone(t.APPBoxRW.UpdateStatus(ctx, t.APPBox))
}

func (t *AKS) ensureKubeconfigSecret(ctx context.Context, secret *corev1.Secret, cluster *v1beta1.KubernetesCluster) error {
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(secret), secret); err != nil {
		if !apierrors.IsNotFound(err) {
			return fmt.Errorf("failed to get secret: %w", err)
		}
		return t.createKubeconfigSecret(ctx, secret, cluster)
	}

	return nil
}

func (t *AKS) createKubeconfigSecret(ctx context.Context, secret *corev1.Secret, cluster *v1beta1.KubernetesCluster) error {
	if cluster.Status.AtProvider.Fqdn == nil {
		return fmt.Errorf("fqdn is empty")
	}
	credentials, err := t.AKSCli.ListClusterUserCredentials(ctx, controllers.GetResourceGroupName(t.VPC), t.getAKSName(), &armcontainerservice.ManagedClustersClientListClusterUserCredentialsOptions{
		ServerFqdn: cluster.Status.AtProvider.Fqdn,
	})
	if err != nil {
		return err
	}
	if len(credentials.Kubeconfigs) == 0 {
		return fmt.Errorf("no kubeconfig found")
	}
	original := credentials.Kubeconfigs[0].Value
	kubeconfig, err := azure.ConvertToWorkloadIdentityLogin(original)
	if err != nil {
		return err
	}
	secret.Data = map[string][]byte{
		"kubeconfig": kubeconfig,
	}
	return k8s.Apply(ctx, t.Local, secret)
}

func (t *AKS) Delete(ctx context.Context) deploy.Result {
	aks := &v1beta1.KubernetesCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.getAKSName(),
		},
	}
	if err := k8s.DeleteObject(ctx, t.Local, aks); err != nil {
		return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
	}

	return deploy.PanicOrDone(k8s.DeleteObject(ctx, t.Local, &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      t.getAKSName(),
			Namespace: t.APPBox.Namespace,
		},
	}))
}

func (t *AKS) PostDelete(ctx context.Context) deploy.Result {
	if err := k8s.GetDeletedObject(ctx, t.Local, types.NamespacedName{
		Name: t.getAKSName(),
	}, &v1beta1.KubernetesCluster{}); err != nil {
		return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
	}
	if err := k8s.GetDeletedObject(ctx, t.Local, types.NamespacedName{
		Name:      t.getAKSName(),
		Namespace: t.APPBox.Namespace,
	}, &corev1.Secret{}); err != nil {
		return deploy.Panic(err).WithRequeuAfter(defaultRequeuInterval)
	}
	return deploy.Done()
}