package controllers

import (
	"context"
	"crypto/md5"
	"fmt"
	"regexp"
	"strings"

	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/pingcap/log"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"go.uber.org/zap"
	"gomodules.xyz/jsonpatch/v2"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	storagev1 "k8s.io/api/storage/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/helm"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/k8s"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/namer"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
)

const (
	VPCNameField           = ".spec.vpcRef.name"
	DataPlaneNameField     = ".spec.dataPlane.name"
	APPBoxNameField        = ".spec.appBoxRef.name"
	KubernetesVersionField = ".spec.kubernetesConfig.version"

	// VNetNameLimit is the limit of network name.
	// At present, subnet name is `{vnetName}-node`, so we need to limit the length of vnet name.
	VNetNameLimit = 57

	hashSize = 8 // Size of the hash used in unique identifiers
)

var (
	// --- AWS Tagging Conventions ---
	// Key: Max 128 characters. Can contain letters, numbers, spaces, and the following characters: + - = . _ : / @
	// Value: Max 256 characters. Can contain letters, numbers, spaces, and the following characters: + - = . _ : / @
	// Case-sensitive.
	awsKeyRegexp   = regexp.MustCompile(`[^a-zA-Z0-9+\-=._:/@]`)
	awsValueRegexp = regexp.MustCompile(`[^a-zA-Z0-9+\-=._:/@]`)

	// --- GCP Tagging Conventions ---
	// Key: Max 63 characters. Must begin with a lowercase letter, and contain only lowercase letters, numeric characters, underscores, and hyphens.
	// Value: Max 63 characters. Can be empty. Must contain only lowercase letters, numeric characters, underscores, and hyphens.
	// Case-insensitive for keys (internally converted to lowercase). Values are case-sensitive.
	gcpKeyRegexp   = regexp.MustCompile(`[^a-z0-9_-]`)
	gcpValueRegexp = regexp.MustCompile(`[^a-z0-9_-]`)

	// --- Azure Tagging Conventions ---
	// Key: Max 512 characters. Can contain letters, numbers, spaces, and the following characters: _ . -
	// Value: Max 256 characters. Can contain letters, numbers, spaces, and the following characters: _ . -
	// Case-insensitive for keys. Values are case-sensitive.
	azureKeyRegexp   = regexp.MustCompile(`[^a-zA-Z0-9_.\- ]`)
	azureValueRegexp = regexp.MustCompile(`[^a-zA-Z0-9_.\- ]`)

	// --- Aliyun Tagging Conventions ---
	// Key: Can be up to 128 characters in length and cannot contain http:// or https://. It cannot start with acs: or aliyun
	// Value: Can be up to 128 characters in length and cannot contain http:// or https://. It cannot start with acs: or aliyun
	aliyunTagRegexp = regexp.MustCompile(`https?://`)
)

var (
	bindingMode   = storagev1.VolumeBindingWaitForFirstConsumer
	reclaimPolicy = corev1.PersistentVolumeReclaimDelete
)

// ConvertAWSStorageClass converts the AWS cloudv1.StorageClass to storagev1.StorageClass
func ConvertAWSStorageClass(sc cloudv1.StorageClass) *storagev1.StorageClass {
	res := &storagev1.StorageClass{
		ObjectMeta: metav1.ObjectMeta{
			Name: sc.Name,
		},
		VolumeBindingMode:    &bindingMode,
		AllowVolumeExpansion: pointer.Bool(true),
		ReclaimPolicy:        &reclaimPolicy,
		Provisioner:          "ebs.csi.aws.com",
		Parameters: map[string]string{
			"csi.storage.k8s.io/fstype": "ext4",
			"type":                      sc.Type,
			"iops":                      fmt.Sprint(sc.IOPS),
			"throughput":                fmt.Sprint(sc.Throughput),
			"encrypted":                 "true",
			// Pass the tags to the persistent volumes
			// See https://github.com/kubernetes-sigs/aws-ebs-csi-driver/blob/master/docs/TAGGING.md for more information
			"tagSpecification": fmt.Sprintf("%s=%s", consts.TagKeyCrossplaneVendor, consts.TagValCrossplaneVendor),
		},
		MountOptions: []string{"nodelalloc,noatime"},
	}
	if sc.Encryption != nil && sc.Encryption.AwsKMSARN != nil {
		res.Parameters["kmsKeyId"] = *sc.Encryption.AwsKMSARN
	}
	return res
}

func EnsureChartDeploymentReady(ctx context.Context, getter k8s.ClientGetter, key types.NamespacedName) error {
	cli, err := getter.GetClient(ctx)
	if err != nil {
		return err
	}

	app := new(appsv1.Deployment)
	if err := cli.Get(ctx, key, app); err != nil {
		return err
	}

	if IsStatusConditionTrue(app.Status.Conditions, appsv1.DeploymentAvailable) {
		return nil
	}
	return fmt.Errorf("chart deployment %s not ready", key.String())
}

func EnsureChartStatefulSetReady(ctx context.Context, getter k8s.ClientGetter, key types.NamespacedName) error {
	cli, err := getter.GetClient(ctx)
	if err != nil {
		return err
	}

	app := new(appsv1.StatefulSet)
	if err := cli.Get(ctx, key, app); err != nil {
		return err
	}

	if app.Status.ReadyReplicas == *app.Spec.Replicas {
		return nil
	}
	return fmt.Errorf("chart statefulset %s not ready", key.String())
}

func EnsureDaemonSetReady(ctx context.Context, getter k8s.ClientGetter, key types.NamespacedName) error {
	cli, err := getter.GetClient(ctx)
	if err != nil {
		return err
	}
	var app appsv1.DaemonSet
	if err := cli.Get(ctx, key, &app); err != nil {
		return err
	}

	if app.Status.NumberReady != app.Status.DesiredNumberScheduled {
		return fmt.Errorf("DaemonSet %s is not ready", key.String())
	}
	return nil
}

var _ helm.Applier = &MockHelmApplier{}

type MockHelmApplier struct {
	err   error
	chart helm.Package
}

func (m *MockHelmApplier) ExpectedApplyErr(err error) *MockHelmApplier {
	m.err = err
	return m
}

func (m *MockHelmApplier) Apply(ctx context.Context, cli k8s.ClientGetter, chart helm.Package) error {
	m.chart = chart
	return m.err
}

func (m *MockHelmApplier) GetAppliedChart() helm.Package {
	return m.chart
}

func GetUnacceptedVPCEndpoints(connsStatus []v1alpha1.ConnectionStatus) []string {
	eps := []string{}
	for _, conn := range connsStatus {
		if conn.Phase == v1alpha1.ConnectionPhasePending {
			eps = append(eps, conn.Endpoint)
		}
	}

	return eps
}

// ConvertToCloudTags converts a map with tidbcloud.com prefixed keys to cloud provider compliant tags
// based on the specified cloudType.
func ConvertToCloudTags(input map[string]string, cloudType string) map[string]string {
	var resultTags = make(map[string]string)

	for key, value := range input {
		if !strings.HasPrefix(key, consts.TagLabelPrefix) {
			continue // Skip keys that don't have our prefix
		}

		// Extract the actual tag name after the prefix
		tagName := strings.TrimPrefix(key, consts.TagLabelPrefix)

		var convertedKey, convertedValue string

		switch cloudType {
		case consts.ProviderTypeEKS:
			convertedKey = toAWSTagKey(tagName)
			convertedValue = toAWSTagValue(value)
		case consts.ProviderTypeGKE:
			convertedKey = toGCPTagKey(tagName)
			convertedValue = toGCPTagValue(value)
		case consts.ProviderTypeAKS:
			convertedKey = toAzureTagKey(tagName)
			convertedValue = toAzureTagValue(value)
		case consts.ProviderTypeACK:
			convertedKey = toAliyunValidTag(tagName)
			convertedValue = toAliyunValidTag(value)
		default:
			// Handle unknown cloud type or return empty if not supported
			log.Warn("Warning: Unknown cloud type %s. Skipping tag conversion for key %s.", zap.String("cloud", cloudType))
			continue
		}

		if convertedKey != "" { // Only add if the key is valid after conversion
			resultTags[convertedKey] = convertedValue
		}
	}
	return resultTags
}

func EffectiveCloudTags(
	cloudType string,
	additionalCloudTags map[string]string,
	resourceLabels map[string]string,
	resourceSpecTags map[string]string,
	dataPlaneCloudTags map[string]string,
) map[string]string {
	return util.MergeMaps(
		ConvertToCloudTags(resourceLabels, cloudType),
		additionalCloudTags,
		ConvertToCloudTags(resourceSpecTags, cloudType),
		ConvertToCloudTags(dataPlaneCloudTags, cloudType),
	)
}

// EnsureDataPlaneIDLabel backfills the stable dataplane ownership label on child CRs.
func EnsureDataPlaneIDLabel(obj client.Object, dataPlaneID string) {
	if obj == nil || dataPlaneID == "" {
		return
	}
	labels := obj.GetLabels()
	if labels == nil {
		labels = map[string]string{}
	}
	labels[v1alpha1.DataPlaneIDKey] = dataPlaneID
	obj.SetLabels(labels)
}

func toAWSTagKey(s string) string {
	s = strings.ReplaceAll(s, " ", "_") // AWS allows spaces, but underscores are safer for general use
	s = awsKeyRegexp.ReplaceAllString(s, "")
	if len(s) > 128 {
		s = s[:128]
	}
	return s
}

func toAWSTagValue(s string) string {
	s = awsValueRegexp.ReplaceAllString(s, "")
	if len(s) > 256 {
		s = s[:256]
	}
	return s
}

func toGCPTagKey(s string) string {
	s = strings.ToLower(s)
	s = strings.ReplaceAll(s, " ", "-") // Spaces are not allowed, convert to hyphen
	s = gcpKeyRegexp.ReplaceAllString(s, "")
	if len(s) > 63 {
		s = s[:63]
	}
	// GCP keys must start with a lowercase letter
	if len(s) > 0 && !regexp.MustCompile(`^[a-z]`).MatchString(s) {
		return "" // Simplification: if it can't start with a letter, just discard
	}
	return s
}

func toGCPTagValue(s string) string {
	s = strings.ToLower(s) // GCP values are generally lowercase for consistency, though not strictly enforced for values
	s = strings.ReplaceAll(s, " ", "-")
	s = gcpValueRegexp.ReplaceAllString(s, "")
	if len(s) > 63 {
		s = s[:63]
	}
	return s
}

func toAzureTagKey(s string) string {
	// Replace "/" with "-" to maintain readability before removing other invalid chars
	s = strings.ReplaceAll(s, "/", "-")
	s = azureKeyRegexp.ReplaceAllString(s, "")
	if len(s) > 512 {
		s = s[:512]
	}
	return s
}

// ToAzureTagKey converts a tag key to Azure-compatible format.
// Azure tag keys cannot contain "/" characters, so they are replaced with "-".
// Other invalid characters are removed according to Azure's tag naming rules.
// This is a public wrapper for use in Azure-specific resource implementations.
func ToAzureTagKey(s string) string {
	return toAzureTagKey(s)
}

func toAzureTagValue(s string) string {
	s = azureValueRegexp.ReplaceAllString(s, "")
	if len(s) > 256 {
		s = s[:256]
	}
	return s
}

// toAliyunValidTag: both tag key and value can be up to 128 characters in length and cannot contain http:// or https://.
// It cannot start with acs: or aliyun
func toAliyunValidTag(s string) string {
	s = aliyunTagRegexp.ReplaceAllString(s, "")

	for {
		if strings.HasPrefix(s, "acs:") {
			s = s[4:]
		} else if strings.HasPrefix(s, "aliyun") {
			s = s[6:]
		} else {
			break
		}
	}

	if len(s) > 128 {
		s = s[:128]
	}
	return s
}

func UniqueIdentifier(meta metav1.ObjectMeta, options ...namer.IdentifierOptions) string {
	const (
		labelLengthLimit = 63
		resultLimit      = labelLengthLimit - hashSize - 1
	)
	namespacedName := fmt.Sprintf("%s/%s", meta.GetNamespace(), meta.GetName())
	hash := fmt.Sprintf("%x", md5.Sum([]byte(namespacedName)))[:hashSize]

	result := meta.GetName()

	for _, opt := range options {
		result = opt(result)
	}
	if len(result) > resultLimit {
		result = result[:resultLimit]
	}
	return fmt.Sprintf("%s-%s", result, hash)
}

// GetResourceGroupName returns the target Azure resource group name for a dataplane VPC.
// It prefers explicit VPC.Spec.AccountRef.ResourceGroupName, and falls back to generated legacy naming.
func GetResourceGroupName(vpc *v1alpha1.VPC, prefix string) string {
	if vpc != nil {
		if vpc.Spec.AccountRef != nil {
			if rgName := strings.TrimSpace(vpc.Spec.AccountRef.ResourceGroupName); rgName != "" {
				return rgName
			}
		}
	}
	return UniqueIdentifier(vpc.ObjectMeta, namer.WithPrefix(prefix))
}

func IsResourceReady(status xpv1.ResourceStatus) bool {
	var ready bool
	for _, condition := range status.Conditions {
		if condition.Type == xpv1.TypeReady {
			if condition.Status == corev1.ConditionTrue && condition.Reason == xpv1.ReasonAvailable {
				ready = true
				break
			}
		}
	}
	return ready
}

// IsStatusConditionTrue returns true when the conditionType is present and set to `true`
func IsStatusConditionTrue(conditions []appsv1.DeploymentCondition, conditionType appsv1.DeploymentConditionType) bool {
	return IsStatusConditionPresentAndEqual(conditions, conditionType, corev1.ConditionTrue)
}

// IsStatusConditionPresentAndEqual returns true when conditionType is present and equal to status.
func IsStatusConditionPresentAndEqual(conditions []appsv1.DeploymentCondition, conditionType appsv1.DeploymentConditionType, status corev1.ConditionStatus) bool {
	for _, condition := range conditions {
		if condition.Type == conditionType {
			return condition.Status == status
		}
	}
	return false
}

func GetResourceManagementPolicies() xpv1.ManagementPolicies {
	return xpv1.ManagementPolicies{
		xpv1.ManagementActionCreate,
		xpv1.ManagementActionUpdate,
		xpv1.ManagementActionObserve,
		xpv1.ManagementActionLateInitialize,
	}
}

var ManagementPoliciesPatch k8s.ObjectPatch = func(obj client.Object) *jsonpatch.Operation {
	return &jsonpatch.Operation{
		Operation: "replace",
		Path:      "/spec/managementPolicies",
		Value:     xpv1.ManagementPolicies{xpv1.ManagementActionAll},
	}
}

func DeleteUpboundResource(ctx context.Context, cli client.Client, obj client.Object) error {
	if err := cli.Get(ctx, client.ObjectKeyFromObject(obj), obj); err != nil {
		if apierrors.IsNotFound(err) {
			return nil
		}
		return err
	}
	if obj.GetDeletionTimestamp() != nil {
		return nil
	}

	if err := k8s.PatchObject(ctx, cli, obj, ManagementPoliciesPatch); err != nil {
		return err
	}
	return k8s.DeleteObject(ctx, cli, obj)
}

func DeletionConfirmed(obj client.Object) bool {
	confirmed, ok := obj.GetAnnotations()[consts.AnnotationKeyConfirmDeletion]
	return ok && confirmed == "true"
}