package task

import (
	"context"
	"crypto/md5"
	"encoding/hex"
	"fmt"
	"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
	snapshotBackupPrefix    string
	snapshotSASToken        string
	providerConfigName      string

	logBackupPrefix   string
	logBackupSASToken string

	// resource tag
	restoreTenantID  string
	restoreProjectID string
}

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.restoreTenantID,
		consts.TagKeyProject:        p.restoreProjectID,
		consts.TagKeyCluster:        consts.ClusterShared,
		consts.TagKeyUsedBy:         consts.UsedByCustomerTiDB,
		cloudv1.TagLabelServiceType: cloudv1.ServiceTypeDedicated,
		// used for node schedule
		namer.ComponentSelectorKey: string(cloudv1.ComponentRestore),
		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.ComponentRestore),
						},
					},
					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.ComponentRestore)},
							},
						},
					},
				},
			},
		},
	}
}

func NewAzureProvider(tenantID, subscriptionID, clientID string, dataplaneSubscriptionID, kubernetesName, kubeVersion string, az cloudv1.CloudAzure, providerConfigName string, restoreTenantID string, restoreProjectID string, snapshotBucket string, logBackupBucket string) (CloudProvider, error) {
	storageAccount, container, snapshotBackupPrefix, sasToken, err := util.ParseAzureBlobURI(snapshotBucket)
	if err != nil {
		return nil, err
	}

	var logBackupPrefix string
	var logBackupSASToken string
	if logBackupBucket != "" {
		_, _, logBackupPrefix, logBackupSASToken, err = util.ParseAzureBlobURI(logBackupBucket)
		if err != nil {
			return nil, err
		}
	}

	// log backup and snapshot backup use the same bucket

	return &azureProvider{
		tenantID:                tenantID,
		subscriptionID:          subscriptionID,
		clientID:                clientID,
		kubernetesName:          kubernetesName,
		kubeVersion:             kubeVersion,
		dataplaneSubscriptionID: dataplaneSubscriptionID,
		az:                      az,
		storageAccount:          storageAccount,
		container:               container,
		snapshotBackupPrefix:    snapshotBackupPrefix,
		snapshotSASToken:        sasToken,
		providerConfigName:      providerConfigName,
		restoreTenantID:         restoreTenantID,
		restoreProjectID:        restoreProjectID,
		// PITR needed
		logBackupPrefix:   logBackupPrefix,
		logBackupSASToken: logBackupSASToken,
	}, nil
}

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(_ *cloudv1.Restore, restore *opv1alpha1.Restore) error {
	restore.Spec.StorageProvider = opv1alpha1.StorageProvider{
		Azblob: &opv1alpha1.AzblobStorageProvider{
			StorageAccount: p.storageAccount,
			Container:      p.container,
			SasToken:       p.snapshotSASToken,
			Prefix:         p.snapshotBackupPrefix,
		},
	}
	restore.Spec.BR.SendCredToTikv = ptr.To(true)

	if restore.Spec.Mode == opv1alpha1.RestoreModePiTR {
		// log backup and snapshot backup use the same bucket, sas token, container
		restore.Spec.StorageProvider.Azblob.Prefix = p.logBackupPrefix
		restore.Spec.PitrFullBackupStorageProvider = opv1alpha1.StorageProvider{
			Azblob: &opv1alpha1.AzblobStorageProvider{
				StorageAccount: p.storageAccount,
				Container:      p.container,
				SasToken:       p.snapshotSASToken,
				Prefix:         p.snapshotBackupPrefix,
			},
		}
	}

	return nil
}