package task

import (
	"context"
	"crypto/md5"
	"encoding/hex"
	"fmt"
	"net/url"
	"strings"

	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/upbound/provider-azure/apis/containerservice/v1beta1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

type azureProvider struct {
	tenantID                string
	subscriptionID          string
	clientID                string
	az                      cloudv1.CloudAzure
	kubernetesName          string
	kubeVersion             string
	dataplaneSubscriptionID string
	storageAccount          string
	container               string
	prefix                  string
	sasToken                string
	providerConfigName      string

	// resource tag
	backupTenantID  string
	backupProjectID string
}

func NewAzureProvider(tenantID, subscriptionID, clientID string, dataplaneSubscriptionID, kubernetesName, kubeVersion string, az cloudv1.CloudAzure, bucket string, providerConfigName string, backupTenantID string, backupProjectID string) (CloudProvider, error) {
	storageAccount, container, prefix, sasToken, err := util.ParseAzureBlobURI(bucket)
	if err != nil {
		return nil, err
	}

	return &azureProvider{
		tenantID:                tenantID,
		subscriptionID:          subscriptionID,
		clientID:                clientID,
		kubernetesName:          kubernetesName,
		kubeVersion:             kubeVersion,
		dataplaneSubscriptionID: dataplaneSubscriptionID,
		az:                      az,
		storageAccount:          storageAccount,
		container:               container,
		prefix:                  prefix,
		sasToken:                sasToken,
		providerConfigName:      providerConfigName,
		backupTenantID:          backupTenantID,
		backupProjectID:         backupProjectID,
	}, nil
}

func (p *azureProvider) GetProviderConfigName() string {
	// use the providerconfig of dataplane
	return p.providerConfigName
}

// Nodepool and node labels should be
func (p *azureProvider) GetNodepoolLabels() map[string]string {
	return map[string]string{
		consts.TagKeyTenant:         p.backupTenantID,
		consts.TagKeyProject:        p.backupProjectID,
		consts.TagKeyCluster:        consts.ClusterShared,
		consts.TagKeyUsedBy:         consts.UsedByCustomerTiDB,
		cloudv1.TagLabelServiceType: cloudv1.ServiceTypeDedicated,
		// used for node schedule
		namer.ComponentSelectorKey: string(cloudv1.ComponentBackup),
		consts.K8sVersionLabelKey:  p.kubeVersion,
	}
}

func (p *azureProvider) GetPodAffinity() *corev1.Affinity {
	return &corev1.Affinity{
		PodAntiAffinity: &corev1.PodAntiAffinity{
			RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
				{
					LabelSelector: &metav1.LabelSelector{
						MatchLabels: map[string]string{
							consts.ComponentLabelKey: string(cloudv1.ComponentBackup),
						},
					},
					TopologyKey: "kubernetes.io/hostname",
				},
			},
		},
		NodeAffinity: &corev1.NodeAffinity{
			RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
				NodeSelectorTerms: []corev1.NodeSelectorTerm{
					{
						MatchExpressions: []corev1.NodeSelectorRequirement{
							{
								Key:      namer.ComponentSelectorKey,
								Operator: corev1.NodeSelectorOpIn,
								Values:   []string{string(cloudv1.ComponentBackup)},
							},
						},
					},
				},
			},
		},
	}
}

func (p *azureProvider) BuildNodePool(ctx context.Context, pc string, np *NodePool) ([]client.Object, error) {
	taints := []*string{}
	for _, t := range np.Taints {
		taint := fmt.Sprintf("%s=%s:%s", t.Key, t.Value, t.Effect)
		taints = append(taints, ptr.To(taint))
	}
	if len(np.NodeSubnets) < 1 || len(np.PodSubnets) < 1 {
		return nil, fmt.Errorf("missing node or pod subnet")
	}
	zones := make([]string, len(np.Zones))
	for i, z := range np.Zones {
		s := strings.Split(z, "-")
		if len(s) != 2 {
			return nil, fmt.Errorf("invalid zone %s", z)
		}
		zones[i] = s[len(s)-1]
	}
	diskSize := ptr.To(float64(defaultRootDiskSizeGB))
	if np.RootDiskSizeGB != 0 {
		diskSize = ptr.To(float64(np.RootDiskSizeGB))
	}
	kcnp := &v1beta1.KubernetesClusterNodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name:   p.nodePoolName(np.Name),
			Labels: np.Labels,
		},
		Spec: v1beta1.KubernetesClusterNodePoolSpec{
			ForProvider: v1beta1.KubernetesClusterNodePoolParameters{
				VMSize:              ptr.To(np.InstanceType),
				OsSku:               ptr.To("AzureLinux"),
				OsDiskSizeGb:        diskSize,
				Zones:               util.ToSlicePointer(zones),
				Tags:                util.ToMapPointer(np.Tags),
				NodeLabels:          util.ToMapPointer(np.Labels),
				NodeTaints:          taints,
				KubernetesClusterID: ptr.To(np.KubeClusterID),
				VnetSubnetID:        ptr.To(np.NodeSubnets[0]),
				PodSubnetID:         ptr.To(np.PodSubnets[0]),
				EnableAutoScaling:   ptr.To(true), // always enable autoscaling
				MinCount:            ptr.To(float64(0)),
				MaxCount:            ptr.To(float64(maxNodeCountPerZone)),
				// 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
				UpgradeSettings: []v1beta1.KubernetesClusterNodePoolUpgradeSettingsParameters{
					{
						MaxSurge: ptr.To("10%"),
					},
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: pc,
				},
				ManagementPolicies: []xpv1.ManagementAction{
					xpv1.ManagementActionCreate,
					xpv1.ManagementActionObserve,
					xpv1.ManagementActionUpdate,
					xpv1.ManagementActionDelete,
				},
			},
		},
	}
	return []client.Object{kcnp}, nil
}

func (p *azureProvider) nodePoolName(name string) string {
	hash := md5.Sum([]byte(name))
	hashString := hex.EncodeToString(hash[:])
	// prefix with `n` to avoid name starts with a number
	return fmt.Sprintf("n%s", hashString[:11])
}

func (p *azureProvider) GetKubernetesID() string {
	return fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.ContainerService/managedClusters/%s", p.dataplaneSubscriptionID, p.az.ResourceGroup, p.kubernetesName)
}

func (p *azureProvider) GetKubernetesName() string {
	return p.kubernetesName
}

func (p *azureProvider) GetKubernetesVersion() string {
	return p.kubeVersion
}

func (p *azureProvider) GetNodeSubnets() []string {
	// https://learn.microsoft.com/en-us/azure/aks/configure-azure-cni-dynamic-ip-allocation
	return []string{p.az.NodeSubnets[0].Name}
}

func (p *azureProvider) GetNodeZones() []string {
	return p.az.NodeZones
}

func (p *azureProvider) GetPodSubnets() []string {
	return []string{p.az.PodSubnets[0].Name}
}

// aks does not support per nodepool managed identity, we don't use this identity at the moment
// https://learn.microsoft.com/en-us/azure/aks/use-managed-identity
func (p *azureProvider) GetNodeIdentity() string {
	return ""
}

func (p *azureProvider) Overlay(backup *opv1alpha1.Backup) error {
	// follow previous convention: append backup CR name to prefix
	prefix := strings.Trim(fmt.Sprintf("%s/%s", p.prefix, backup.Name), "/")
	backup.Spec.StorageProvider = opv1alpha1.StorageProvider{
		Azblob: &opv1alpha1.AzblobStorageProvider{
			StorageAccount: p.storageAccount,
			Container:      p.container,
			SasToken:       p.sasToken,
			Prefix:         prefix,
		},
	}
	backup.Spec.BR.SendCredToTikv = ptr.To(true)
	return nil
}

func (p *azureProvider) OverlayStatus(cloudBackup *cloudv1.Backup, opBackup *opv1alpha1.Backup) error {
	cloudBackup.Status.BucketName = p.storageAccount

	backupMode := ptr.Deref(cloudBackup.Spec.Mode, "")
	if backupMode == cloudv1.BackupModeLog {
		backupPath := ptr.Deref(cloudBackup.Spec.Storage, "")
		u, err := url.Parse(backupPath)
		if err != nil {
			return err
		}
		// the backup path is required in format of `https://<storage-account>.blob.core.windows.net/<container>/<prefix>/backup-name`,
		// however, the opBackup.status.BackupPath is in format of `azure://<container>/<prefix>/backup-name`.
		// We need to overwrite it here.
		cloudBackup.Status.BackupPath = fmt.Sprintf("%s://%s%s/%s", u.Scheme, u.Host, u.Path, cloudBackup.Name)
	}
	return nil
}