package gkeheterogeneous

import (
	"context"
	"fmt"
	"sort"
	"strings"

	"github.com/pingcap/errors"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	k8smeta "k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"sigs.k8s.io/controller-runtime/pkg/client"

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

const (
	gatekeeperMutationGroup   = "mutations.gatekeeper.sh"
	gatekeeperMutationVersion = "v1"
	gatekeeperAssignKind      = "Assign"

	GatekeeperAssignComputeClassLabelKey  = "tidbcloud.com/gke-compute-class"
	HeterogeneousDeployLabelKey           = "tidbcloud.com/heterogeneous-deploy"
	HeterogeneousComponentLabelKey        = "tidbcloud.com/component"
	GatekeeperAssignPurposeLabelKey       = "tidbcloud.com/gke-heterogeneous-gatekeeper-assign-purpose"
	GatekeeperAssignDeployLabelValue      = "gatekeeper-assign"
	GatekeeperAssignPurposeNodeSelector   = "compute-class-node-selector"
	GatekeeperAssignPurposeTopologySpread = "topology-spread-node-affinity-policy"
	ManagedByClusterAgentLabelValue       = "cluster-agent"

	gatekeeperTopologySpreadConstraintLocation   = "spec.topologySpreadConstraints[topologyKey:'topology.kubernetes.io/zone']"
	gatekeeperTopologySpreadNodeAffinityLocation = gatekeeperTopologySpreadConstraintLocation + ".nodeAffinityPolicy"
	gatekeeperTopologySpreadNodeAffinityIgnore   = "Ignore"
	gatekeeperAssignPrimarySuffix                = "compute-class-assign"
	gatekeeperAssignTopologySpreadSuffix         = "topology-spread-assign"
	tidbClusterNamespacePrefix                   = "tidb"
)

var (
	gatekeeperAssignGVK     = schema.GroupVersionKind{Group: gatekeeperMutationGroup, Version: gatekeeperMutationVersion, Kind: gatekeeperAssignKind}
	gatekeeperAssignListGVK = schema.GroupVersionKind{Group: gatekeeperMutationGroup, Version: gatekeeperMutationVersion, Kind: gatekeeperAssignKind + "List"}
)

type GatekeeperAssignComputeClassRef struct {
	Name                  string
	Namespace             string
	LabelComputeClassName string
	SpecComputeClassName  string
}

func GatekeeperAssignRefsForComputeClass(ctx context.Context, cli client.Client, computeClassName string) ([]GatekeeperAssignComputeClassRef, error) {
	computeClassName = strings.TrimSpace(computeClassName)
	if computeClassName == "" {
		return nil, errors.New("compute class name is empty")
	}
	if cli == nil {
		return nil, errors.New("kubernetes client is nil")
	}

	assigns := &unstructured.UnstructuredList{}
	assigns.SetGroupVersionKind(gatekeeperAssignListGVK)
	if err := cli.List(ctx, assigns); err != nil {
		return nil, err
	}

	refs := make([]GatekeeperAssignComputeClassRef, 0)
	for i := range assigns.Items {
		assign := &assigns.Items[i]
		if isGKEHeterogeneousTopologySpreadGatekeeperAssign(assign) {
			continue
		}
		labelComputeClassName := strings.TrimSpace(assign.GetLabels()[GatekeeperAssignComputeClassLabelKey])
		if labelComputeClassName != "" && labelComputeClassName != computeClassName {
			continue
		}
		specComputeClassName, found, err := unstructured.NestedString(assign.Object, "spec", "parameters", "assign", "value", consts.GKEComputeClassLabelKey)
		if err != nil {
			if isManagedGKEHeterogeneousGatekeeperAssign(assign, labelComputeClassName) {
				return nil, errors.Wrapf(err, "parse compute class value from gatekeeper assign %s", assign.GetName())
			}
			continue
		}
		specComputeClassName = strings.TrimSpace(specComputeClassName)
		if labelComputeClassName != computeClassName && (!found || specComputeClassName != computeClassName) {
			continue
		}
		refs = append(refs, GatekeeperAssignComputeClassRef{
			Name:                  assign.GetName(),
			Namespace:             assign.GetNamespace(),
			LabelComputeClassName: labelComputeClassName,
			SpecComputeClassName:  specComputeClassName,
		})
	}
	return refs, nil
}

func DeleteGatekeeperAssignForComponentIfExists(ctx context.Context, cli client.Client, clusterID string, componentType cloudv1.ComponentType, tidbGroupID string) error {
	clusterID = strings.TrimSpace(clusterID)
	tidbGroupID = strings.TrimSpace(tidbGroupID)
	if clusterID == "" {
		return errors.New("cluster id is empty")
	}
	if cli == nil {
		return errors.New("kubernetes client is nil")
	}
	if componentType != cloudv1.ComponentTiKV && componentType != cloudv1.ComponentTiDB {
		return errors.Errorf("unsupported heterogeneous component type %q", componentType)
	}
	if componentType == cloudv1.ComponentTiDB && tidbGroupID == "" {
		return errors.New("tidb group id is required for TiDB gatekeeper assign deletion")
	}

	assigns := &unstructured.UnstructuredList{}
	assigns.SetGroupVersionKind(gatekeeperAssignListGVK)
	labels := gatekeeperAssignComponentLabels(clusterID, componentType, tidbGroupID)
	if err := cli.List(ctx, assigns, client.MatchingLabels(labels)); err != nil {
		if isGatekeeperAssignMissingError(err) {
			return nil
		}
		return errors.Wrap(err, "list gatekeeper assigns")
	}

	primaryName := ResourceName(clusterID, componentType, tidbGroupID, gatekeeperAssignPrimarySuffix)
	companionName := ResourceName(clusterID, componentType, tidbGroupID, gatekeeperAssignTopologySpreadSuffix)
	primaries := make([]unstructured.Unstructured, 0, len(assigns.Items))
	companions := make([]unstructured.Unstructured, 0, len(assigns.Items))
	for i := range assigns.Items {
		assign := &assigns.Items[i]
		switch purpose := strings.TrimSpace(assign.GetLabels()[GatekeeperAssignPurposeLabelKey]); purpose {
		case GatekeeperAssignPurposeNodeSelector:
			primaries = append(primaries, *assign)
		case GatekeeperAssignPurposeTopologySpread:
			companions = append(companions, *assign)
		default:
			return errors.Errorf("gatekeeper assign %s has unsupported purpose %q", assign.GetName(), purpose)
		}
	}
	if len(companions) > 1 {
		names := make([]string, 0, len(companions))
		for i := range companions {
			names = append(names, companions[i].GetName())
		}
		sort.Strings(names)
		return errors.Errorf("multiple topology spread gatekeeper assigns found for %s/%s/%s: %s", clusterID, componentType, tidbGroupID, strings.Join(names, ","))
	}
	if len(companions) == 1 {
		if err := validateGatekeeperAssignForDelete(&companions[0], clusterID, componentType, tidbGroupID, companionName, GatekeeperAssignPurposeTopologySpread); err != nil {
			return err
		}
	}

	switch len(primaries) {
	case 0:
		primary := newGatekeeperAssign()
		if err := cli.Get(ctx, client.ObjectKey{Name: primaryName}, primary); err != nil {
			if isGatekeeperAssignMissingError(err) {
				return deleteGatekeeperAssignByNameIfExists(ctx, cli, companionName, clusterID, componentType, tidbGroupID, GatekeeperAssignPurposeTopologySpread)
			}
			return errors.Wrapf(err, "get gatekeeper assign %s", primaryName)
		}
		if err := validateGatekeeperAssignForDelete(primary, clusterID, componentType, tidbGroupID, primaryName, GatekeeperAssignPurposeNodeSelector); err != nil {
			return err
		}
		return deleteGatekeeperAssignPair(ctx, cli, primary, clusterID, componentType, tidbGroupID)
	case 1:
		primary := &primaries[0]
		if err := validateGatekeeperAssignForDelete(primary, clusterID, componentType, tidbGroupID, primaryName, GatekeeperAssignPurposeNodeSelector); err != nil {
			return err
		}
		return deleteGatekeeperAssignPair(ctx, cli, primary, clusterID, componentType, tidbGroupID)
	default:
		names := make([]string, 0, len(primaries))
		for i := range primaries {
			names = append(names, primaries[i].GetName())
		}
		sort.Strings(names)
		return errors.Errorf("multiple primary gatekeeper assigns found for %s/%s/%s: %s", clusterID, componentType, tidbGroupID, strings.Join(names, ","))
	}
}

func deleteGatekeeperAssignPair(ctx context.Context, cli client.Client, primary *unstructured.Unstructured, clusterID string, componentType cloudv1.ComponentType, tidbGroupID string) error {
	companionName := ResourceName(clusterID, componentType, tidbGroupID, gatekeeperAssignTopologySpreadSuffix)
	if err := deleteGatekeeperAssignByNameIfExists(ctx, cli, companionName, clusterID, componentType, tidbGroupID, GatekeeperAssignPurposeTopologySpread); err != nil {
		return err
	}
	if err := cli.Delete(ctx, primary); err != nil && !isGatekeeperAssignMissingError(err) {
		return errors.Wrapf(err, "delete gatekeeper assign %s", primary.GetName())
	}
	return nil
}

func deleteGatekeeperAssignByNameIfExists(ctx context.Context, cli client.Client, name, clusterID string, componentType cloudv1.ComponentType, tidbGroupID, purpose string) error {
	assign := newGatekeeperAssign()
	if err := cli.Get(ctx, client.ObjectKey{Name: name}, assign); err != nil {
		if isGatekeeperAssignMissingError(err) {
			return nil
		}
		return errors.Wrapf(err, "get gatekeeper assign %s", name)
	}
	if err := validateGatekeeperAssignForDelete(assign, clusterID, componentType, tidbGroupID, name, purpose); err != nil {
		return err
	}
	if err := cli.Delete(ctx, assign); err != nil && !isGatekeeperAssignMissingError(err) {
		return errors.Wrapf(err, "delete gatekeeper assign %s", name)
	}
	return nil
}

func newGatekeeperAssign() *unstructured.Unstructured {
	obj := &unstructured.Unstructured{}
	obj.SetGroupVersionKind(gatekeeperAssignGVK)
	return obj
}

func gatekeeperAssignComponentLabels(clusterID string, componentType cloudv1.ComponentType, tidbGroupID string) map[string]string {
	labels := map[string]string{
		consts.ManagedByLabelKey:       ManagedByClusterAgentLabelValue,
		consts.ClusterLabelKey:         strings.TrimSpace(clusterID),
		HeterogeneousDeployLabelKey:    GatekeeperAssignDeployLabelValue,
		HeterogeneousComponentLabelKey: string(componentType),
	}
	if componentType == cloudv1.ComponentTiDB && strings.TrimSpace(tidbGroupID) != "" {
		labels[consts.LabelKeyTiDBGroupId] = strings.TrimSpace(tidbGroupID)
	}
	return labels
}

func validateGatekeeperAssignForDelete(assign *unstructured.Unstructured, clusterID string, componentType cloudv1.ComponentType, tidbGroupID, name, purpose string) error {
	if assign.GetName() != name {
		return errors.Errorf("gatekeeper assign %s does not match expected name %s", assign.GetName(), name)
	}
	labels := gatekeeperAssignComponentLabels(clusterID, componentType, tidbGroupID)
	for key, want := range labels {
		if got := strings.TrimSpace(assign.GetLabels()[key]); got != want {
			return errors.Errorf("gatekeeper assign %s label %s = %q, want %q", assign.GetName(), key, got, want)
		}
	}
	if componentType != cloudv1.ComponentTiDB && strings.TrimSpace(assign.GetLabels()[consts.LabelKeyTiDBGroupId]) != "" {
		return errors.Errorf("gatekeeper assign %s has unexpected tidb group label %q", assign.GetName(), assign.GetLabels()[consts.LabelKeyTiDBGroupId])
	}
	if componentType == cloudv1.ComponentTiDB && strings.TrimSpace(tidbGroupID) == "" && strings.TrimSpace(assign.GetLabels()[consts.LabelKeyTiDBGroupId]) != "" {
		return errors.Errorf("gatekeeper assign %s has unexpected tidb group label %q", assign.GetName(), assign.GetLabels()[consts.LabelKeyTiDBGroupId])
	}
	if got := strings.TrimSpace(assign.GetLabels()[GatekeeperAssignPurposeLabelKey]); got != purpose {
		return errors.Errorf("gatekeeper assign %s purpose = %q, want %q", assign.GetName(), got, purpose)
	}
	if purpose == GatekeeperAssignPurposeTopologySpread && !isGKEHeterogeneousTopologySpreadGatekeeperAssign(assign) {
		return errors.Errorf("topology spread gatekeeper assign %s has invalid shape", assign.GetName())
	}
	return validateGatekeeperAssignNamespaceScope(assign, clusterID)
}

func validateGatekeeperAssignNamespaceScope(assign *unstructured.Unstructured, clusterID string) error {
	scope, found, err := unstructured.NestedString(assign.Object, "spec", "match", "scope")
	if err != nil {
		return errors.Wrapf(err, "read gatekeeper assign %s match scope", assign.GetName())
	}
	if !found || strings.TrimSpace(scope) != "Namespaced" {
		return errors.Errorf("gatekeeper assign %s must match Namespaced scope", assign.GetName())
	}
	namespaces, found, err := unstructured.NestedStringSlice(assign.Object, "spec", "match", "namespaces")
	if err != nil {
		return errors.Wrapf(err, "read gatekeeper assign %s match namespaces", assign.GetName())
	}
	expectedNamespace := tidbClusterNamespacePrefix + strings.TrimSpace(clusterID)
	if !found || len(namespaces) != 1 || strings.TrimSpace(namespaces[0]) != expectedNamespace {
		return errors.Errorf("gatekeeper assign %s does not target cluster namespace %s", assign.GetName(), expectedNamespace)
	}
	return nil
}

func isGatekeeperAssignMissingError(err error) bool {
	return apierrors.IsNotFound(err) || k8smeta.IsNoMatchError(err)
}

func isGKEHeterogeneousTopologySpreadGatekeeperAssign(assign *unstructured.Unstructured) bool {
	labels := assign.GetLabels()
	if strings.TrimSpace(labels[GatekeeperAssignPurposeLabelKey]) != GatekeeperAssignPurposeTopologySpread ||
		strings.TrimSpace(labels[consts.ManagedByLabelKey]) != ManagedByClusterAgentLabelValue ||
		strings.TrimSpace(labels[HeterogeneousDeployLabelKey]) != GatekeeperAssignDeployLabelValue ||
		strings.TrimSpace(labels[GatekeeperAssignComputeClassLabelKey]) != "" {
		return false
	}
	location, found, err := unstructured.NestedString(assign.Object, "spec", "location")
	if err != nil || !found || strings.TrimSpace(location) != gatekeeperTopologySpreadNodeAffinityLocation {
		return false
	}
	value, found, err := unstructured.NestedString(assign.Object, "spec", "parameters", "assign", "value")
	return err == nil && found && strings.TrimSpace(value) == gatekeeperTopologySpreadNodeAffinityIgnore
}

func isManagedGKEHeterogeneousGatekeeperAssign(assign *unstructured.Unstructured, labelComputeClassName string) bool {
	labels := assign.GetLabels()
	return strings.TrimSpace(labelComputeClassName) != "" ||
		(strings.TrimSpace(labels[consts.ManagedByLabelKey]) == ManagedByClusterAgentLabelValue &&
			strings.TrimSpace(labels[HeterogeneousDeployLabelKey]) == GatekeeperAssignDeployLabelValue)
}

func FormatGatekeeperAssignComputeClassRefs(refs []GatekeeperAssignComputeClassRef) string {
	parts := make([]string, 0, len(refs))
	for _, ref := range refs {
		namespace := strings.TrimSpace(ref.Namespace)
		name := strings.TrimSpace(ref.Name)
		if namespace != "" {
			name = namespace + "/" + name
		}
		labelComputeClassName := strings.TrimSpace(ref.LabelComputeClassName)
		if labelComputeClassName == "" {
			labelComputeClassName = "<empty>"
		}
		specComputeClassName := strings.TrimSpace(ref.SpecComputeClassName)
		if specComputeClassName == "" {
			specComputeClassName = "<empty>"
		}
		parts = append(parts, fmt.Sprintf(
			"assign=%s, label=%s, spec=%s",
			name,
			labelComputeClassName,
			specComputeClassName,
		))
	}
	sort.Strings(parts)
	return strings.Join(parts, "; ")
}