package service

import (
	"context"
	"maps"
	"slices"
	"strings"

	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/emptypb"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/equality"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	k8smeta "k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"sigs.k8s.io/controller-runtime/pkg/client"

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

func (s *AgentService) CreateGKEHeterogeneousGatekeeperAssign(ctx context.Context, req *apiv1.CreateGKEHeterogeneousGatekeeperAssignRequest) (*apiv1.GKEHeterogeneousGatekeeperAssign, error) {
	clusterID, err := s.validateGKEClusterAgentRequest(req.GetClusterId(), "gke heterogeneous APIs")
	if err != nil {
		return nil, err
	}
	component, err := resolveGKEHeterogeneousGatekeeperComponent(req.GetComponentType(), req.GetTidbGroupId(), req.GetTidbGroupKind())
	if err != nil {
		return nil, err
	}
	computeClassName, err := s.computeClassNameForGatekeeperAssign(ctx, req.GetComputeClassName())
	if err != nil {
		return nil, err
	}

	name := defaultGKEHeterogeneousResourceName(clusterID, component, "compute-class-assign")
	topologySpreadName := defaultGKEHeterogeneousResourceName(clusterID, component, "topology-spread-assign")

	settings, err := s.inferGKEHeterogeneousAssignSettings(ctx, clusterID, component)
	if err != nil {
		return nil, err
	}
	obj, err := buildGatekeeperAssign(name, clusterID, component, computeClassName, settings)
	if err != nil {
		return nil, err
	}
	token := gatekeeperAssignExemptToken(name)
	topologySpreadAssign, err := buildGatekeeperTopologySpreadAssign(topologySpreadName, clusterID, component, settings, token)
	if err != nil {
		return nil, err
	}
	if err := preflightManagedUnstructured(ctx, s.dataplaneCli, obj, validateGKEHeterogeneousGatekeeperAssignOwner); err != nil {
		return nil, err
	}
	if err := preflightManagedUnstructured(ctx, s.dataplaneCli, topologySpreadAssign, validateGKEHeterogeneousTopologySpreadGatekeeperAssignOwner); err != nil {
		return nil, err
	}
	if err := s.markGatekeeperExemptPods(ctx, settings.pods, token); err != nil {
		return nil, err
	}
	topologySpreadCreated := false
	if _, topologySpreadCreated, err = createManagedUnstructuredWithCreated(ctx, s.dataplaneCli, topologySpreadAssign, validateGKEHeterogeneousTopologySpreadGatekeeperAssignOwner); err != nil {
		if topologySpreadCreated {
			if rollbackErr := s.dataplaneCli.Delete(ctx, topologySpreadAssign); rollbackErr != nil && !apierrors.IsNotFound(rollbackErr) {
				return nil, status.Errorf(codes.Internal, "create topology spread gatekeeper assign failed: %s; rollback topology spread gatekeeper assign %s failed: %s", err, topologySpreadAssign.GetName(), rollbackErr)
			}
		}
		return nil, err
	}
	obj, err = createManagedUnstructured(ctx, s.dataplaneCli, obj, validateGKEHeterogeneousGatekeeperAssignOwner)
	if err != nil {
		if topologySpreadCreated {
			if rollbackErr := s.dataplaneCli.Delete(ctx, topologySpreadAssign); rollbackErr != nil && !apierrors.IsNotFound(rollbackErr) {
				return nil, status.Errorf(codes.Internal, "create gatekeeper assign failed: %s; rollback topology spread gatekeeper assign %s failed: %s", err, topologySpreadAssign.GetName(), rollbackErr)
			}
		}
		return nil, err
	}
	return gkeHeterogeneousGatekeeperAssignResponse(clusterID, obj)
}

func (s *AgentService) GetGKEHeterogeneousGatekeeperAssign(ctx context.Context, req *apiv1.GetGKEHeterogeneousGatekeeperAssignRequest) (*apiv1.GKEHeterogeneousGatekeeperAssign, error) {
	clusterID, err := s.validateGKEClusterAgentRequest(req.GetClusterId(), "gke heterogeneous APIs")
	if err != nil {
		return nil, err
	}
	component, err := resolveGKEHeterogeneousComponent(req.GetComponentType(), req.GetTidbGroupId())
	if err != nil {
		return nil, err
	}
	obj, err := getOwnedGKEHeterogeneousGatekeeperAssignForComponent(ctx, s.dataplaneCli, clusterID, component)
	if err != nil {
		return nil, err
	}
	resp, err := gkeHeterogeneousGatekeeperAssignResponse(clusterID, obj)
	if err != nil {
		return nil, err
	}
	if err := validateGKEHeterogeneousGatekeeperAssignCompanion(ctx, s.dataplaneCli, clusterID, component, obj); err != nil {
		return nil, err
	}
	return resp, nil
}

func (s *AgentService) UpdateGKEHeterogeneousGatekeeperAssign(ctx context.Context, req *apiv1.UpdateGKEHeterogeneousGatekeeperAssignRequest) (*apiv1.GKEHeterogeneousGatekeeperAssign, error) {
	clusterID, err := s.validateGKEClusterAgentRequest(req.GetClusterId(), "gke heterogeneous APIs")
	if err != nil {
		return nil, err
	}
	obj, err := getOwnedGKEHeterogeneousGatekeeperAssign(ctx, s.dataplaneCli, clusterID, req.GetName())
	if err != nil {
		return nil, err
	}
	if gatekeeperAssignPurpose(obj) != gatekeeperAssignPurposeNodeSelector {
		return nil, status.Errorf(codes.FailedPrecondition, "gatekeeper assign %s is not a node selector assign", obj.GetName())
	}
	computeClassName, err := s.computeClassNameForGatekeeperAssign(ctx, req.GetComputeClassName())
	if err != nil {
		return nil, err
	}
	base := obj.DeepCopy()
	if err := setGatekeeperAssignComputeClass(obj, computeClassName); err != nil {
		return nil, err
	}
	if err := s.dataplaneCli.Patch(ctx, obj, client.MergeFrom(base)); err != nil {
		return nil, gkeHeterogeneousResourceClientError(gatekeeperAssignLabel, err, "update gatekeeper assign")
	}
	return gkeHeterogeneousGatekeeperAssignResponse(clusterID, obj)
}

func (s *AgentService) DeleteGKEHeterogeneousGatekeeperAssign(ctx context.Context, req *apiv1.DeleteGKEHeterogeneousGatekeeperAssignRequest) (*emptypb.Empty, error) {
	clusterID, err := s.validateGKEClusterAgentRequest(req.GetClusterId(), "gke heterogeneous APIs")
	if err != nil {
		return nil, err
	}
	obj, err := getOwnedGKEHeterogeneousGatekeeperAssign(ctx, s.dataplaneCli, clusterID, req.GetName())
	if err != nil {
		if status.Code(err) == codes.NotFound {
			if err := deleteOrphanGKEHeterogeneousTopologySpreadGatekeeperAssignForPrimaryName(ctx, s.dataplaneCli, clusterID, req.GetName()); err != nil {
				return nil, err
			}
			return &emptypb.Empty{}, nil
		}
		return nil, err
	}
	component, err := gkeHeterogeneousComponentFromGatekeeperAssign(obj)
	if err != nil {
		return nil, err
	}
	if gatekeeperAssignPurpose(obj) == gatekeeperAssignPurposeNodeSelector {
		if err := deleteGKEHeterogeneousTopologySpreadGatekeeperAssign(ctx, s.dataplaneCli, clusterID, component); err != nil {
			return nil, err
		}
	}
	if err := s.dataplaneCli.Delete(ctx, obj); err != nil && !apierrors.IsNotFound(err) {
		return nil, gkeHeterogeneousResourceClientError(gatekeeperAssignLabel, err, "delete gatekeeper assign")
	}
	return &emptypb.Empty{}, nil
}

func (s *AgentService) computeClassNameForGatekeeperAssign(ctx context.Context, name string) (string, error) {
	computeClassName, err := cloudv1.NormalizeGKEComputeClassName(name)
	if err != nil {
		return "", status.Error(codes.InvalidArgument, err.Error())
	}
	if _, err := getSharedManagedGKEHeterogeneousComputeClass(ctx, s.dataplaneCli, computeClassName); err != nil {
		return "", err
	}
	return computeClassName, nil
}

func (s *AgentService) markGatekeeperExemptPods(ctx context.Context, pods []corev1.Pod, token string) error {
	// Mark only currently live Pods. The label must never be written to Pod
	// templates, so future scale-out or recreate Pods still match this Assign.
	for i := range pods {
		pod := &pods[i]
		if pod.GetLabels()[gkeHeterogeneousGatekeeperExemptLabelKey] == token {
			continue
		}
		base := pod.DeepCopy()
		labels := maps.Clone(pod.GetLabels())
		if labels == nil {
			labels = map[string]string{}
		}
		labels[gkeHeterogeneousGatekeeperExemptLabelKey] = token
		pod.SetLabels(labels)
		if err := s.dataplaneCli.Patch(ctx, pod, client.MergeFrom(base)); err != nil {
			return status.Errorf(codes.Internal, "mark gatekeeper exempt pod %s/%s: %s", pod.Namespace, pod.Name, err)
		}
	}
	return nil
}

func gkeHeterogeneousComponentFromGatekeeperAssign(obj *unstructured.Unstructured) (*gkeHeterogeneousDataplaneComponent, error) {
	componentLabel := strings.TrimSpace(obj.GetLabels()[gkeHeterogeneousComponentLabelKey])
	switch cloudv1.ComponentType(componentLabel) {
	case cloudv1.ComponentTiKV:
		return &gkeHeterogeneousDataplaneComponent{componentType: cloudv1.ComponentTiKV}, nil
	case cloudv1.ComponentTiDB:
		groupID := strings.TrimSpace(obj.GetLabels()[consts.LabelKeyTiDBGroupId])
		if groupID == "" {
			return nil, status.Errorf(codes.FailedPrecondition, "gatekeeper assign %s is missing tidb group ownership, cannot manage TiDB heterogeneous assignment", obj.GetName())
		}
		return &gkeHeterogeneousDataplaneComponent{componentType: cloudv1.ComponentTiDB, tidbGroupID: groupID}, nil
	case "":
		return nil, status.Errorf(codes.FailedPrecondition, "gatekeeper assign %s is missing component ownership", obj.GetName())
	default:
		return nil, status.Errorf(codes.FailedPrecondition, "gatekeeper assign %s has unsupported component ownership %q", obj.GetName(), componentLabel)
	}
}

func setGatekeeperAssignComputeClass(obj *unstructured.Unstructured, computeClassName string) error {
	value, found, err := gatekeeperAssignValueMap(obj)
	if err != nil {
		return status.Errorf(codes.Internal, "read gatekeeper assign value: %s", err)
	}
	if !found {
		return status.Errorf(codes.FailedPrecondition, "gatekeeper assign %s is missing assign value", obj.GetName())
	}
	value[consts.GKEComputeClassLabelKey] = computeClassName
	if err := unstructured.SetNestedMap(obj.Object, value, "spec", "parameters", "assign", "value"); err != nil {
		return status.Errorf(codes.Internal, "set gatekeeper assign compute class: %s", err)
	}
	labels := obj.GetLabels()
	labels[gkeHeterogeneousComputeClassLabelKey] = computeClassName
	obj.SetLabels(labels)
	return nil
}

func gatekeeperAssignValueMap(obj *unstructured.Unstructured) (map[string]interface{}, bool, error) {
	return unstructured.NestedMap(obj.Object, "spec", "parameters", "assign", "value")
}

func gatekeeperAssignComputeClassValue(obj *unstructured.Unstructured) (string, bool, error) {
	return unstructured.NestedString(obj.Object, "spec", "parameters", "assign", "value", consts.GKEComputeClassLabelKey)
}

func gatekeeperAssignPurpose(obj *unstructured.Unstructured) string {
	return strings.TrimSpace(obj.GetLabels()[gkeHeterogeneousGatekeeperAssignPurposeKey])
}

func (s *AgentService) inferGKEHeterogeneousAssignSettings(ctx context.Context, clusterID string, component *gkeHeterogeneousDataplaneComponent) (*gkeHeterogeneousAssignSettings, error) {
	pods, matchLabels, err := s.getGKEHeterogeneousReferencePods(ctx, clusterID, component)
	if err != nil {
		return nil, err
	}
	settings := gkeHeterogeneousAssignSettingsFromPod(&pods[0])
	settings.matchLabels = matchLabels
	settings.pods = pods
	for i := 1; i < len(pods); i++ {
		next := gkeHeterogeneousAssignSettingsFromPod(&pods[i])
		if settings.namespace != next.namespace ||
			!maps.Equal(settings.nodeSelector, next.nodeSelector) {
			return nil, status.Errorf(codes.FailedPrecondition, "%s pods have inconsistent node selectors for heterogeneous deploy; wait for rollout to finish before updating gatekeeper assign", component.name())
		}
	}
	return settings, nil
}

func (s *AgentService) getGKEHeterogeneousReferencePods(ctx context.Context, clusterID string, component *gkeHeterogeneousDataplaneComponent) ([]corev1.Pod, map[string]string, error) {
	var pods corev1.PodList
	matchLabels := gkeHeterogeneousPodMatchLabels(component)
	if err := s.dataplaneCli.List(ctx, &pods, client.InNamespace(buildNamespace(clusterID)), client.MatchingLabels(matchLabels)); err != nil {
		return nil, nil, status.Errorf(codes.Internal, "list component pods: %s", err)
	}
	filteredPods, filteredMatchLabels, err := filterTiDBGatekeeperReferencePods(component, pods.Items, matchLabels)
	if err != nil {
		return nil, nil, err
	}
	pods.Items = filteredPods
	matchLabels = filteredMatchLabels
	if len(pods.Items) == 0 {
		return nil, nil, status.Errorf(codes.FailedPrecondition, "no %s pod found in cluster %s", component.name(), clusterID)
	}
	slices.SortFunc(pods.Items, func(a, b corev1.Pod) int {
		if a.Namespace != b.Namespace {
			return strings.Compare(a.Namespace, b.Namespace)
		}
		return strings.Compare(a.Name, b.Name)
	})
	return pods.Items, matchLabels, nil
}

func filterTiDBGatekeeperReferencePods(component *gkeHeterogeneousDataplaneComponent, pods []corev1.Pod, matchLabels map[string]string) ([]corev1.Pod, map[string]string, error) {
	if !component.isTiDB() {
		return pods, matchLabels, nil
	}

	expectedInstance := consts.ClusterPrefix
	switch component.tidbGroupKind {
	case apiv1.TiDBGroupKind_TIDB_GROUP_KIND_DEFAULT:
	case apiv1.TiDBGroupKind_TIDB_GROUP_KIND_EXTRA:
		expectedInstance = gkeHeterogeneousTiDBGroupInstanceLabel(component.tidbGroupID)
	default:
		return nil, nil, status.Error(codes.FailedPrecondition, "tidb group kind is required for TiDB heterogeneous deploy")
	}

	groupPods := make([]corev1.Pod, 0, len(pods))
	for _, pod := range pods {
		if pod.GetLabels()[consts.InstanceLabelKey] == expectedInstance {
			groupPods = append(groupPods, pod)
		}
	}
	matchLabels = maps.Clone(matchLabels)
	matchLabels[consts.InstanceLabelKey] = expectedInstance
	return groupPods, matchLabels, nil
}

func gkeHeterogeneousAssignSettingsFromPod(pod *corev1.Pod) *gkeHeterogeneousAssignSettings {
	nodeSelector := make(map[string]string, len(pod.Spec.NodeSelector))
	for key, value := range pod.Spec.NodeSelector {
		if key == consts.InstanceTypeLabelKey || key == consts.GKEComputeClassLabelKey {
			continue
		}
		nodeSelector[key] = value
	}
	return &gkeHeterogeneousAssignSettings{
		namespace:    pod.Namespace,
		nodeSelector: nodeSelector,
	}
}

func buildGatekeeperAssign(name, clusterID string, component *gkeHeterogeneousDataplaneComponent, computeClassName string, settings *gkeHeterogeneousAssignSettings) (*unstructured.Unstructured, error) {
	if settings == nil || len(settings.matchLabels) == 0 {
		return nil, status.Error(codes.Internal, "gatekeeper assign match labels are required")
	}

	labels := gkeHeterogeneousLabels(clusterID, gatekeeperAssignDeployLabelValue, component)
	labels[gkeHeterogeneousComputeClassLabelKey] = computeClassName
	labels[gkeHeterogeneousGatekeeperAssignPurposeKey] = gatekeeperAssignPurposeNodeSelector
	token := gatekeeperAssignExemptToken(name)
	obj := &unstructured.Unstructured{
		Object: map[string]interface{}{
			"apiVersion": gatekeeperMutationGroup + "/" + gatekeeperMutationVersion,
			"kind":       gatekeeperAssignKind,
			"metadata": map[string]interface{}{
				"name":   name,
				"labels": labels,
			},
			"spec": buildGatekeeperAssignSpec(computeClassName, settings, token),
		},
	}
	obj.SetGroupVersionKind(gatekeeperAssignGVK)
	return obj, nil
}

func buildGatekeeperTopologySpreadAssign(name, clusterID string, component *gkeHeterogeneousDataplaneComponent, settings *gkeHeterogeneousAssignSettings, token string) (*unstructured.Unstructured, error) {
	if settings == nil || len(settings.matchLabels) == 0 {
		return nil, status.Error(codes.Internal, "gatekeeper assign match labels are required")
	}
	if strings.TrimSpace(token) == "" {
		return nil, status.Error(codes.Internal, "gatekeeper assign exemption token is required")
	}

	labels := gkeHeterogeneousLabels(clusterID, gatekeeperAssignDeployLabelValue, component)
	labels[gkeHeterogeneousGatekeeperAssignPurposeKey] = gatekeeperAssignPurposeTopologySpread
	obj := &unstructured.Unstructured{
		Object: map[string]interface{}{
			"apiVersion": gatekeeperMutationGroup + "/" + gatekeeperMutationVersion,
			"kind":       gatekeeperAssignKind,
			"metadata": map[string]interface{}{
				"name":   name,
				"labels": labels,
			},
			"spec": buildGatekeeperTopologySpreadAssignSpec(settings, token),
		},
	}
	obj.SetGroupVersionKind(gatekeeperAssignGVK)
	return obj, nil
}

func buildGatekeeperAssignSpec(computeClassName string, settings *gkeHeterogeneousAssignSettings, token string) map[string]interface{} {
	nodeSelector := map[string]interface{}{}
	for key, value := range settings.nodeSelector {
		nodeSelector[key] = value
	}
	nodeSelector[consts.GKEComputeClassLabelKey] = computeClassName
	return map[string]interface{}{
		"applyTo":  buildGatekeeperPodApplyTo(),
		"match":    buildGatekeeperAssignMatch(settings, token),
		"location": "spec.nodeSelector",
		"parameters": map[string]interface{}{
			"assign": map[string]interface{}{
				"value": nodeSelector,
			},
		},
	}
}

func buildGatekeeperTopologySpreadAssignSpec(settings *gkeHeterogeneousAssignSettings, token string) map[string]interface{} {
	return map[string]interface{}{
		"applyTo":  buildGatekeeperPodApplyTo(),
		"match":    buildGatekeeperAssignMatch(settings, token),
		"location": gatekeeperTopologySpreadNodeAffinityLocation,
		"parameters": map[string]interface{}{
			"pathTests": []interface{}{
				map[string]interface{}{
					"subPath":   gatekeeperTopologySpreadConstraintLocation,
					"condition": "MustExist",
				},
			},
			"assign": map[string]interface{}{
				"value": string(corev1.NodeInclusionPolicyIgnore),
			},
		},
	}
}

func buildGatekeeperPodApplyTo() []interface{} {
	return []interface{}{
		map[string]interface{}{
			"groups":   []interface{}{""},
			"versions": []interface{}{"v1"},
			"kinds":    []interface{}{"Pod"},
		},
	}
}

func buildGatekeeperAssignMatch(settings *gkeHeterogeneousAssignSettings, token string) map[string]interface{} {
	matchLabels := map[string]interface{}{}
	for key, value := range settings.matchLabels {
		matchLabels[key] = value
	}
	// NotIn also matches Pods without the label, so future CREATE/recreate Pods
	// still receive the Gatekeeper mutation while marked live Pods are excluded.
	labelSelector := map[string]interface{}{
		"matchLabels": matchLabels,
		"matchExpressions": []interface{}{
			map[string]interface{}{
				"key":      gkeHeterogeneousGatekeeperExemptLabelKey,
				"operator": "NotIn",
				"values":   []interface{}{token},
			},
		},
	}
	return map[string]interface{}{
		"scope":      "Namespaced",
		"namespaces": []interface{}{settings.namespace},
		"kinds": []interface{}{
			map[string]interface{}{
				"apiGroups": []interface{}{""},
				"kinds":     []interface{}{"Pod"},
			},
		},
		"labelSelector": labelSelector,
	}
}

func getOwnedGKEHeterogeneousGatekeeperAssign(ctx context.Context, cli client.Client, clusterID, name string) (*unstructured.Unstructured, error) {
	obj, err := getOwnedGKEHeterogeneousResource(
		ctx,
		cli,
		clusterID,
		name,
		gatekeeperAssignLabel,
		gatekeeperAssignGVK,
		gatekeeperAssignDeployLabelValue,
	)
	if err != nil {
		return nil, err
	}
	if err := validateGKEHeterogeneousGatekeeperAssignScope(obj, clusterID); err != nil {
		return nil, err
	}
	if _, err := gkeHeterogeneousComponentFromGatekeeperAssign(obj); err != nil {
		return nil, err
	}
	return obj, nil
}

func getOwnedGKEHeterogeneousGatekeeperAssignForComponent(ctx context.Context, cli client.Client, clusterID string, component *gkeHeterogeneousDataplaneComponent) (*unstructured.Unstructured, error) {
	items := &unstructured.UnstructuredList{}
	items.SetGroupVersionKind(gatekeeperAssignListGVK)
	labels := gkeHeterogeneousComponentResourceLabels(clusterID, component)
	labels[gkeHeterogeneousDeployLabelKey] = gatekeeperAssignDeployLabelValue
	if err := cli.List(ctx, items, client.MatchingLabels(labels)); err != nil {
		if k8smeta.IsNoMatchError(err) {
			return nil, status.Error(codes.NotFound, "gatekeeper assign API is not installed")
		}
		return nil, gkeHeterogeneousResourceClientError(gatekeeperAssignLabel, err, "list %s", gatekeeperAssignLabel)
	}

	nodeSelectorAssigns := make([]unstructured.Unstructured, 0, len(items.Items))
	for i := range items.Items {
		switch purpose := gatekeeperAssignPurpose(&items.Items[i]); purpose {
		case gatekeeperAssignPurposeNodeSelector:
			nodeSelectorAssigns = append(nodeSelectorAssigns, items.Items[i])
		case gatekeeperAssignPurposeTopologySpread:
		default:
			return nil, status.Errorf(codes.FailedPrecondition, "gatekeeper assign %s has unsupported purpose %q", items.Items[i].GetName(), purpose)
		}
	}

	switch len(nodeSelectorAssigns) {
	case 0:
		return nil, status.Errorf(codes.NotFound, "%s for %s not found", gatekeeperAssignLabel, gkeHeterogeneousComponentResourceLabel(component))
	case 1:
		obj := &nodeSelectorAssigns[0]
		if err := validateGKEHeterogeneousGatekeeperAssignScope(obj, clusterID); err != nil {
			return nil, err
		}
		return obj, nil
	default:
		names := make([]string, 0, len(nodeSelectorAssigns))
		for i := range nodeSelectorAssigns {
			names = append(names, nodeSelectorAssigns[i].GetName())
		}
		return nil, status.Errorf(codes.FailedPrecondition, "multiple %s resources found for %s: %s", gatekeeperAssignLabel, gkeHeterogeneousComponentResourceLabel(component), strings.Join(names, ","))
	}
}

func validateGKEHeterogeneousGatekeeperAssignCompanion(
	ctx context.Context,
	cli client.Client,
	clusterID string,
	component *gkeHeterogeneousDataplaneComponent,
	primary *unstructured.Unstructured,
) error {
	name := defaultGKEHeterogeneousResourceName(clusterID, component, "topology-spread-assign")
	obj, err := getOwnedGKEHeterogeneousGatekeeperAssign(ctx, cli, clusterID, name)
	if err != nil {
		if status.Code(err) == codes.NotFound {
			return status.Errorf(codes.FailedPrecondition, "topology spread gatekeeper assign %s for %s is missing", name, gkeHeterogeneousComponentResourceLabel(component))
		}
		return err
	}
	actualComponent, err := gkeHeterogeneousComponentFromGatekeeperAssign(obj)
	if err != nil {
		return err
	}
	if actualComponent.componentType != component.componentType ||
		strings.TrimSpace(actualComponent.tidbGroupID) != strings.TrimSpace(component.tidbGroupID) {
		return status.Errorf(
			codes.FailedPrecondition,
			"topology spread gatekeeper assign %s component mismatch: got %s, want %s",
			name,
			gkeHeterogeneousComponentResourceLabel(actualComponent),
			gkeHeterogeneousComponentResourceLabel(component),
		)
	}
	if err := validateGKEHeterogeneousTopologySpreadGatekeeperAssignShape(obj); err != nil {
		return err
	}
	return validateGKEHeterogeneousGatekeeperAssignCompanionMatch(primary, obj)
}

func validateGKEHeterogeneousGatekeeperAssignCompanionMatch(primary, companion *unstructured.Unstructured) error {
	if primary == nil {
		return status.Error(codes.Internal, "primary gatekeeper assign is nil")
	}
	if companion == nil {
		return status.Error(codes.Internal, "topology spread gatekeeper assign is nil")
	}

	primaryMatch, found, err := unstructured.NestedMap(primary.Object, "spec", "match")
	if err != nil {
		return status.Errorf(codes.FailedPrecondition, "read gatekeeper assign %s match: %s", primary.GetName(), err)
	}
	if !found {
		return status.Errorf(codes.FailedPrecondition, "gatekeeper assign %s is missing match", primary.GetName())
	}

	companionMatch, found, err := unstructured.NestedMap(companion.Object, "spec", "match")
	if err != nil {
		return status.Errorf(codes.FailedPrecondition, "read topology spread gatekeeper assign %s match: %s", companion.GetName(), err)
	}
	if !found {
		return status.Errorf(codes.FailedPrecondition, "topology spread gatekeeper assign %s is missing match", companion.GetName())
	}

	if !equality.Semantic.DeepEqual(primaryMatch, companionMatch) {
		return status.Errorf(codes.FailedPrecondition, "topology spread gatekeeper assign %s match does not match primary gatekeeper assign %s",
			companion.GetName(), primary.GetName())
	}

	return nil
}

func deleteGKEHeterogeneousTopologySpreadGatekeeperAssign(ctx context.Context, cli client.Client, clusterID string, component *gkeHeterogeneousDataplaneComponent) error {
	name := defaultGKEHeterogeneousResourceName(clusterID, component, "topology-spread-assign")
	obj, err := getOwnedGKEHeterogeneousGatekeeperAssign(ctx, cli, clusterID, name)
	if err != nil {
		if status.Code(err) == codes.NotFound {
			return nil
		}
		return err
	}
	if gatekeeperAssignPurpose(obj) != gatekeeperAssignPurposeTopologySpread {
		return nil
	}
	if err := cli.Delete(ctx, obj); err != nil && !apierrors.IsNotFound(err) {
		return gkeHeterogeneousResourceClientError(gatekeeperAssignLabel, err, "delete topology spread gatekeeper assign")
	}
	return nil
}

func deleteOrphanGKEHeterogeneousTopologySpreadGatekeeperAssignForPrimaryName(ctx context.Context, cli client.Client, clusterID, primaryName string) error {
	items := &unstructured.UnstructuredList{}
	items.SetGroupVersionKind(gatekeeperAssignListGVK)
	labels := map[string]string{
		consts.ManagedByLabelKey:                   gkeHeterogeneousManagedByClusterAgent,
		consts.ClusterLabelKey:                     clusterID,
		gkeHeterogeneousDeployLabelKey:             gatekeeperAssignDeployLabelValue,
		gkeHeterogeneousGatekeeperAssignPurposeKey: gatekeeperAssignPurposeTopologySpread,
	}
	if err := cli.List(ctx, items, client.MatchingLabels(labels)); err != nil {
		return gkeHeterogeneousResourceClientError(gatekeeperAssignLabel, err, "list topology spread gatekeeper assign")
	}

	for i := range items.Items {
		obj := &items.Items[i]
		component, err := gkeHeterogeneousComponentFromGatekeeperAssign(obj)
		if err != nil {
			continue
		}
		if defaultGKEHeterogeneousResourceName(clusterID, component, "compute-class-assign") != primaryName {
			continue
		}
		if defaultGKEHeterogeneousResourceName(clusterID, component, "topology-spread-assign") != obj.GetName() {
			continue
		}
		if err := validateGKEHeterogeneousGatekeeperAssignScope(obj, clusterID); err != nil {
			return err
		}
		if err := cli.Delete(ctx, obj); err != nil && !apierrors.IsNotFound(err) {
			return gkeHeterogeneousResourceClientError(gatekeeperAssignLabel, err, "delete orphan topology spread gatekeeper assign")
		}
	}
	return nil
}

func validateGKEHeterogeneousTopologySpreadGatekeeperAssignShape(obj *unstructured.Unstructured) error {
	if gatekeeperAssignPurpose(obj) != gatekeeperAssignPurposeTopologySpread {
		return status.Errorf(codes.FailedPrecondition, "gatekeeper assign %s is not a topology spread assign", obj.GetName())
	}
	if strings.TrimSpace(obj.GetLabels()[gkeHeterogeneousComputeClassLabelKey]) != "" {
		return status.Errorf(codes.FailedPrecondition, "topology spread gatekeeper assign %s must not have compute class label", obj.GetName())
	}
	location, found, err := unstructured.NestedString(obj.Object, "spec", "location")
	if err != nil {
		return status.Errorf(codes.FailedPrecondition, "read topology spread gatekeeper assign %s location: %s", obj.GetName(), err)
	}
	if !found || strings.TrimSpace(location) != gatekeeperTopologySpreadNodeAffinityLocation {
		return status.Errorf(codes.FailedPrecondition, "topology spread gatekeeper assign %s must assign %s", obj.GetName(), gatekeeperTopologySpreadNodeAffinityLocation)
	}
	value, found, err := unstructured.NestedString(obj.Object, "spec", "parameters", "assign", "value")
	if err != nil {
		return status.Errorf(codes.FailedPrecondition, "read topology spread gatekeeper assign %s assign value: %s", obj.GetName(), err)
	}
	if !found || strings.TrimSpace(value) != string(corev1.NodeInclusionPolicyIgnore) {
		return status.Errorf(codes.FailedPrecondition, "topology spread gatekeeper assign %s must assign %s", obj.GetName(), corev1.NodeInclusionPolicyIgnore)
	}
	return nil
}

func validateGKEHeterogeneousGatekeeperAssignOwner(current, desired *unstructured.Unstructured) error {
	if err := validateGKEHeterogeneousOwner(current, desired); err != nil {
		return err
	}
	clusterID := strings.TrimSpace(desired.GetLabels()[consts.ClusterLabelKey])
	if err := validateGKEHeterogeneousGatekeeperAssignScope(current, clusterID); err != nil {
		return err
	}
	if err := validateGKEHeterogeneousGatekeeperAssignPurpose(current, desired); err != nil {
		return err
	}
	return validateGKEHeterogeneousGatekeeperAssignComputeClass(current, desired)
}

func validateGKEHeterogeneousTopologySpreadGatekeeperAssignOwner(current, desired *unstructured.Unstructured) error {
	if err := validateGKEHeterogeneousOwner(current, desired); err != nil {
		return err
	}
	clusterID := strings.TrimSpace(desired.GetLabels()[consts.ClusterLabelKey])
	if err := validateGKEHeterogeneousGatekeeperAssignScope(current, clusterID); err != nil {
		return err
	}
	return validateGKEHeterogeneousGatekeeperAssignPurpose(current, desired)
}

func validateGKEHeterogeneousGatekeeperAssignPurpose(current, desired *unstructured.Unstructured) error {
	currentPurpose := gatekeeperAssignPurpose(current)
	desiredPurpose := gatekeeperAssignPurpose(desired)
	if currentPurpose != desiredPurpose {
		return status.Errorf(codes.AlreadyExists, "%s %s already exists with different purpose", desired.GetKind(), desired.GetName())
	}
	return nil
}

func validateGKEHeterogeneousGatekeeperAssignComputeClass(current, desired *unstructured.Unstructured) error {
	currentLabel := strings.TrimSpace(current.GetLabels()[gkeHeterogeneousComputeClassLabelKey])
	desiredLabel := strings.TrimSpace(desired.GetLabels()[gkeHeterogeneousComputeClassLabelKey])
	if currentLabel == "" {
		return status.Errorf(codes.AlreadyExists, "%s %s already exists without compute class label", desired.GetKind(), desired.GetName())
	}
	if currentLabel != desiredLabel {
		return status.Errorf(codes.AlreadyExists, "%s %s already exists with different compute class label", desired.GetKind(), desired.GetName())
	}

	currentValue, found, err := gatekeeperAssignComputeClassValue(current)
	if err != nil {
		return status.Errorf(codes.AlreadyExists, "read %s %s compute class assign value: %s", desired.GetKind(), desired.GetName(), err)
	}
	if !found || strings.TrimSpace(currentValue) == "" {
		return status.Errorf(codes.AlreadyExists, "%s %s already exists without compute class assign value", desired.GetKind(), desired.GetName())
	}
	if strings.TrimSpace(currentValue) != currentLabel {
		return status.Errorf(codes.AlreadyExists, "%s %s already exists with mismatched compute class label and assign value", desired.GetKind(), desired.GetName())
	}

	desiredValue, found, err := gatekeeperAssignComputeClassValue(desired)
	if err != nil {
		return status.Errorf(codes.Internal, "read desired %s %s compute class assign value: %s", desired.GetKind(), desired.GetName(), err)
	}
	if !found || strings.TrimSpace(desiredValue) != desiredLabel {
		return status.Errorf(codes.Internal, "desired %s %s has mismatched compute class label and assign value", desired.GetKind(), desired.GetName())
	}
	return nil
}

func validateGKEHeterogeneousGatekeeperAssignScope(obj *unstructured.Unstructured, clusterID string) error {
	expectedNamespace := buildNamespace(clusterID)
	scope, found, err := unstructured.NestedString(obj.Object, "spec", "match", "scope")
	if err != nil {
		return status.Errorf(codes.FailedPrecondition, "read gatekeeper assign %s match scope: %s", obj.GetName(), err)
	}
	if !found || strings.TrimSpace(scope) != "Namespaced" {
		return status.Errorf(codes.FailedPrecondition, "gatekeeper assign %s must match Namespaced scope", obj.GetName())
	}

	namespaces, found, err := unstructured.NestedStringSlice(obj.Object, "spec", "match", "namespaces")
	if err != nil {
		return status.Errorf(codes.FailedPrecondition, "read gatekeeper assign %s match namespaces: %s", obj.GetName(), err)
	}
	if !found || len(namespaces) != 1 || strings.TrimSpace(namespaces[0]) != expectedNamespace {
		return status.Errorf(codes.PermissionDenied, "gatekeeper assign %s does not target cluster namespace %s", obj.GetName(), expectedNamespace)
	}
	return nil
}

func gkeHeterogeneousGatekeeperAssignResponse(clusterID string, obj *unstructured.Unstructured) (*apiv1.GKEHeterogeneousGatekeeperAssign, error) {
	response := &apiv1.GKEHeterogeneousGatekeeperAssign{
		ClusterId:     clusterID,
		Name:          obj.GetName(),
		ComponentType: gkeHeterogeneousComponentType(obj.GetLabels()[gkeHeterogeneousComponentLabelKey]),
	}
	if namespaces, found, err := unstructured.NestedStringSlice(obj.Object, "spec", "match", "namespaces"); err != nil {
		return nil, status.Errorf(codes.FailedPrecondition, "read gatekeeper assign %s match namespaces: %s", obj.GetName(), err)
	} else if found && len(namespaces) > 0 {
		response.Namespace = namespaces[0]
	}
	computeClassName, found, err := gatekeeperAssignComputeClassValue(obj)
	if err != nil {
		return nil, status.Errorf(codes.FailedPrecondition, "read gatekeeper assign %s compute class assign value: %s", obj.GetName(), err)
	}
	if !found || strings.TrimSpace(computeClassName) == "" {
		return nil, status.Errorf(codes.FailedPrecondition, "gatekeeper assign %s is missing compute class assign value", obj.GetName())
	}
	response.ComputeClassName = computeClassName
	if selectorVersion, found, err := unstructured.NestedString(obj.Object, "spec", "parameters", "assign", "value", consts.K8sVersionLabelKey); err != nil {
		return nil, status.Errorf(codes.FailedPrecondition, "read gatekeeper assign %s selector version: %s", obj.GetName(), err)
	} else if found {
		response.SelectorVersion = selectorVersion
	}
	if groupID := strings.TrimSpace(obj.GetLabels()[consts.LabelKeyTiDBGroupId]); groupID != "" {
		response.TidbGroupId = groupID
	}
	nodeSelector, found, err := unstructured.NestedStringMap(obj.Object, "spec", "parameters", "assign", "value")
	if err != nil {
		return nil, status.Errorf(codes.FailedPrecondition, "read gatekeeper assign %s node selector: %s", obj.GetName(), err)
	}
	if !found {
		return nil, status.Errorf(codes.FailedPrecondition, "gatekeeper assign %s is missing node selector", obj.GetName())
	}
	response.NodeSelector = nodeSelector
	if matchLabels, found, err := unstructured.NestedStringMap(obj.Object, "spec", "match", "labelSelector", "matchLabels"); err != nil {
		return nil, status.Errorf(codes.FailedPrecondition, "read gatekeeper assign %s match labels: %s", obj.GetName(), err)
	} else if found {
		response.MatchLabels = matchLabels
		if cloudv1.ComponentType(obj.GetLabels()[gkeHeterogeneousComponentLabelKey]) == cloudv1.ComponentTiDB {
			switch matchLabels[consts.InstanceLabelKey] {
			case consts.ClusterPrefix:
				response.TidbGroupKind = apiv1.TiDBGroupKind_TIDB_GROUP_KIND_DEFAULT
			case "":
			default:
				response.TidbGroupKind = apiv1.TiDBGroupKind_TIDB_GROUP_KIND_EXTRA
			}
		}
	}
	return response, nil
}