package aks

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"regexp"
	"strings"

	"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/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/providerconfig"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
	storagev1beta1 "github.com/upbound/provider-azure/apis/storage/v1beta1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/labels"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	storageAccountLabelKey     = "tags.tidbcloud.com/storage-account-id"
	permittedDataplaneLabelKey = "tags.tidbcloud.com/permitted-dataplane-id"
	principalIDLabelKey        = "tags.tidbcloud.com/principal-id"
	roleDefinitionBlobData     = "Storage Blob Data Owner"

	maxStorageAccountNameLength = 24
	randomBytesLength           = 8
)

var (
	storageAccountNameRegexp = regexp.MustCompile(`[^a-z0-9]`)
	containerNameRegexp      = regexp.MustCompile(`[^a-z0-9-]`)
)

func storageAccountName(prefix, dataplaneID string) (string, error) {
	randomBytes := make([]byte, randomBytesLength)
	_, err := rand.Read(randomBytes)
	if err != nil {
		return "", err
	}
	origin := append([]byte(dataplaneID), randomBytes...)
	hash := sha256.Sum256(origin)
	hashedName := hex.EncodeToString(hash[:])
	hashedName = fmt.Sprintf("%s%s", prefix, hashedName)
	// make it a valid storage account name
	hashedName = storageAccountNameRegexp.ReplaceAllString(strings.ToLower(hashedName), "")
	if len(hashedName) > maxStorageAccountNameLength {
		hashedName = hashedName[:maxStorageAccountNameLength]
	}
	return hashedName, nil
}

func containerName(bucketID string) string {
	return containerNameRegexp.ReplaceAllString(strings.ToLower(bucketID), "")
}

func lifecycleRuleName(containerName, ruleID string) string {
	return fmt.Sprintf("%s:%s", containerName, ruleID)
}

func lifecycleRulePrefix(containerName, rulePrefix string) string {
	prefix := containerName + "/"
	if rulePrefix != "" {
		prefix = prefix + rulePrefix
		if prefix[len(prefix)-1] != '/' {
			prefix = prefix + "/"
		}
	}
	return prefix
}

func isLifecycleRuleOfContainer(ruleName, containerName string) bool {
	return strings.HasPrefix(ruleName, containerName+":")
}

func getStorageAccountForDataplane(ctx context.Context, cli client.Client, dataplaneID string) (*storagev1beta1.Account, error) {
	accountList := &storagev1beta1.AccountList{}
	listOpts := &client.ListOptions{
		LabelSelector: labels.SelectorFromSet(map[string]string{
			consts.DataplaneLabelKey: dataplaneID,
		}),
	}
	if err := cli.List(ctx, accountList, listOpts); err != nil {
		return nil, fmt.Errorf("list storage accounts error: %w", err)
	}
	if len(accountList.Items) == 0 {
		return nil, nil
	}
	if len(accountList.Items) > 1 {
		return nil, fmt.Errorf("multiple storage accounts found for dataplane %s", dataplaneID)
	}
	return &accountList.Items[0], nil
}

func getBucketResourceGroup(bucket *v1alpha1.Bucket) (string, error) {
	if bucket.Spec.AccountRef == nil {
		return "", fmt.Errorf("azure account reference is missing for bucket %s", bucket.Name)
	}

	if resourceGroupName := strings.TrimSpace(bucket.Spec.AccountRef.ResourceGroupName); resourceGroupName != "" {
		return resourceGroupName, nil
	}

	accountKey := bucket.Spec.AccountRef.AccountID
	if accountKey == "" {
		return "", fmt.Errorf("azure account id is missing for bucket %s", bucket.Name)
	}

	resourceGroupName, ok := providerconfig.GetAzureAccountCache(accountKey)
	if !ok {
		return "", fmt.Errorf("bucket resource group is not set for subscription %s", accountKey)
	}
	return resourceGroupName, nil
}

func getLabelsAndTags(resourceLabels, parentLabels map[string]string, rconfig config.RootConfig) (map[string]string, map[string]string) {
	labels := util.MergeMaps(resourceLabels, parentLabels, rconfig.ResourceLabels)
	tags := util.MergeMaps(labels, rconfig.AdditionalCloudTags)
	tags = controllers.ConvertToCloudTags(tags, consts.ProviderTypeAKS)
	return labels, tags
}

func getNodePrincipalID(ctx context.Context, cli client.Client, dataplaneID, namespace string) (string, error) {
	dataPlaneCR := &v1alpha1.DataPlane{}
	if err := cli.Get(ctx, client.ObjectKey{
		Namespace: namespace,
		Name:      dataplaneID,
	}, dataPlaneCR); err != nil {
		if apierrors.IsNotFound(err) {
			return "", nil
		}
		return "", fmt.Errorf("get dataplane '%s' error: %w", dataplaneID, err)
	}

	if !dataPlaneCR.DeletionTimestamp.IsZero() {
		return "", nil
	}

	appBoxCR := &v1alpha1.APPBox{}
	if err := cli.Get(ctx, client.ObjectKey{
		Namespace: namespace,
		Name:      dataPlaneCR.Status.AppBox.Name,
	}, appBoxCR); err != nil {
		if apierrors.IsNotFound(err) {
			return "", nil
		}
		return "", fmt.Errorf("get appbox '%s' for dataplane '%s' error: %w", dataPlaneCR.Status.AppBox.Name, dataplaneID, err)
	}

	if !appBoxCR.DeletionTimestamp.IsZero() {
		return "", nil
	}

	if appBoxCR.Status.IAM.KubeletIdentityObjectID == "" {
		return "", fmt.Errorf("kubeletIdentityObjectID not found for appbox '%s' for dataplane '%s'", dataPlaneCR.Status.AppBox.Name, dataplaneID)
	}

	return appBoxCR.Status.IAM.KubeletIdentityObjectID, nil
}