package gatewaycontroller

import (
	"context"
	"fmt"
	"math"
	"reflect"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/haproxytech/client-native/v5/models"
	haproxyv1 "github.com/haproxytech/kubernetes-ingress/v3/crs/api/ingress/v1"
	monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
	appsv1 "k8s.io/api/apps/v1"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/intstr"
	utilversion "k8s.io/apimachinery/pkg/util/version"
	"k8s.io/client-go/discovery"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	logf "sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	gatewayv1alpha1 "github.com/tidbcloud/aws-shared-provider/apis/gateway/v1alpha1"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/query"
)

var (
	ownerKey = ".metadata.controller"
	apiGVStr = gatewayv1alpha1.GroupVersion.String()
)

const (
	ingressClassPublic        = "public"
	ingressClassPrivateLink   = "privatelink"
	ingressClassPrivateLinkZR = "privatelink-zr"
	gatewayMetricsInterval    = monitoringv1.Duration("30s")
)

func matchIngressClass(ingressClass, base string) bool {
	return ingressClass == base || strings.HasPrefix(ingressClass, base+"-")
}

// GatewayReconciler reconciles a Gateway object
type GatewayReconciler struct {
	client.Client
	Scheme *runtime.Scheme

	Config  *Config
	Querier query.PromQuerier

	kubernetesVersion *utilversion.Version
}

func (r *GatewayReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := logf.FromContext(ctx)

	var gateway gatewayv1alpha1.Gateway
	if err := r.Get(ctx, req.NamespacedName, &gateway); err != nil {
		logger.Error(err, "unable to fetch Gateway")
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	// NOTE: gateway.tidbcloud.com/dnsName seems not used in privatelink mode
	deployment, err := r.getOrCreateNewVersionDeployment(ctx, &gateway)
	if err != nil {
		logger.Error(err, "failed to get or create deployment")
		return ctrl.Result{}, err
	}

	if !isDeploymentReady(deployment) {
		logger.Info("Deployment is not ready", "name", deployment.Name, "namespace", deployment.Namespace)
		return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
	}

	_, err = r.changeHPATarget(ctx, &gateway)
	if err != nil {
		logger.Error(err, "failed to change HPA target")
		return ctrl.Result{}, err
	}

	_, err = r.changeServiceLabel(ctx, &gateway)
	if err != nil {
		logger.Error(err, "failed to change service label")
		return ctrl.Result{}, err
	}

	err = r.createMonitorResources(ctx, &gateway)
	if err != nil {
		logger.Error(err, "failed to create monitor resources")
		return ctrl.Result{}, err
	}

	_, err = r.createTCPIfNotExists(ctx, &gateway)
	if err != nil {
		logger.Error(err, "failed to create TCP resource")
		return ctrl.Result{}, err
	}

	if gateway.Spec.Version != gateway.Status.Version {
		gateway.Status.Version = gateway.Spec.Version
		gateway.Status.ApplyTime = metav1.Now()
		if err := r.Status().Update(ctx, &gateway); err != nil {
			logger.Error(err, "failed to update Gateway status")
			return ctrl.Result{}, fmt.Errorf("failed to update Gateway status: %w", err)
		}
		logger.Info("Gateway status updated", "version", gateway.Status.Version, "applyTime", gateway.Status.ApplyTime)
	}

	return r.recycleOldVersionDeployments(ctx, &gateway)
}

func (r *GatewayReconciler) getOrCreateNewVersionDeployment(ctx context.Context, gateway *gatewayv1alpha1.Gateway) (*appsv1.Deployment, error) {
	logger := logf.FromContext(ctx)
	versionLabel := getVersionLabel(gateway)

	var deployment appsv1.Deployment
	if err := r.Get(ctx, client.ObjectKey{Name: versionLabel, Namespace: gateway.Namespace}, &deployment); err != nil {
		if client.IgnoreNotFound(err) != nil {
			return nil, fmt.Errorf("failed to get deployment: %w", err)
		}

		// sqlConfig.GracefulWaitBeforeShutdown = 60, reserve enough time for tiproxy to shutdown gracefully
		gracefulWaitBeforeCloseServer := gateway.Spec.TerminationGracePeriodSeconds - 2*60
		if gracefulWaitBeforeCloseServer < 0 {
			gracefulWaitBeforeCloseServer = 0
		}

		// Deployment not found, create a new one
		deployment = appsv1.Deployment{
			TypeMeta: metav1.TypeMeta{
				Kind:       "Deployment",
				APIVersion: "apps/v1",
			},
			ObjectMeta: metav1.ObjectMeta{
				Name:      versionLabel,
				Namespace: gateway.Namespace,
				Labels: map[string]string{
					"control-plane": "gateway-controller",
					"version":       versionLabel,
					"app":           gateway.Name,
				},
			},
			Spec: appsv1.DeploymentSpec{
				Replicas: int32Ptr(gateway.Spec.Replicas),
				Selector: &metav1.LabelSelector{
					MatchLabels: map[string]string{
						"version": versionLabel,
					},
				},
				Strategy: appsv1.DeploymentStrategy{
					Type: appsv1.RollingUpdateDeploymentStrategyType,
					RollingUpdate: &appsv1.RollingUpdateDeployment{
						MaxSurge:       &intstr.IntOrString{IntVal: 1},
						MaxUnavailable: &intstr.IntOrString{IntVal: 0},
					},
				},
				Template: corev1.PodTemplateSpec{
					ObjectMeta: metav1.ObjectMeta{
						Labels: r.buildPodLabels(gateway, versionLabel),
						Annotations: map[string]string{
							"prometheus.io/path":   "/metrics",
							"prometheus.io/port":   "9090",
							"prometheus.io/scrape": "true",
						},
					},
					Spec: corev1.PodSpec{
						Affinity:                  r.buildNodeAffinity(gateway, versionLabel),
						TopologySpreadConstraints: r.buildTopologySpreadConstraints(gateway, versionLabel),
						Containers: []corev1.Container{
							{
								Name:            "tidb-gateway",
								Image:           fmt.Sprintf("%s:%s", r.Config.ImageRepo, gateway.Spec.Version),
								Command:         []string{"/tidb-gateway"},
								Args:            r.buildContainerArgs(gateway, gracefulWaitBeforeCloseServer),
								ImagePullPolicy: corev1.PullIfNotPresent,
								Ports: []corev1.ContainerPort{
									{ContainerPort: 3306},
									{ContainerPort: 4000},
								},
								Env: []corev1.EnvVar{
									{
										Name: "POD_NAMESPACE",
										ValueFrom: &corev1.EnvVarSource{
											FieldRef: &corev1.ObjectFieldSelector{
												FieldPath: "metadata.namespace",
											},
										},
									},
								},
								SecurityContext: &corev1.SecurityContext{
									RunAsNonRoot: boolPtr(true),
								},
								LivenessProbe: &corev1.Probe{
									ProbeHandler: corev1.ProbeHandler{
										HTTPGet: &corev1.HTTPGetAction{
											Path: "/healthz",
											Port: intstr.FromInt(8081),
										},
									},
									InitialDelaySeconds: 12,
									PeriodSeconds:       20,
								},
								ReadinessProbe: &corev1.Probe{
									ProbeHandler: corev1.ProbeHandler{
										HTTPGet: &corev1.HTTPGetAction{
											Path: "/readyz",
											Port: intstr.FromInt(8081),
										},
									},
									InitialDelaySeconds: 12,
									PeriodSeconds:       10,
								},
								Resources: buildGatewayResourceRequirements(gateway),
								VolumeMounts: []corev1.VolumeMount{
									{
										Name:      "tls-certs",
										MountPath: "/gateway-tls",
									},
									{
										Name:      "infra-api-kubeconfig",
										MountPath: "/infra",
									},
									{
										Name:      "cluster-cert",
										MountPath: "/cluster-cert",
										ReadOnly:  true,
									},
									{
										Name:      "sql-cert",
										MountPath: "/sql-cert",
										ReadOnly:  true,
									},
									{
										Name:      "infra-redis-client-cert",
										MountPath: "/infra-redis-client-cert",
										ReadOnly:  true,
									},
									{
										Name:      "dynamic-config",
										MountPath: "/etc/gateway",
										ReadOnly:  true,
									},
								},
							},
						},
						Volumes: []corev1.Volume{
							{
								Name: "tls-certs",
								VolumeSource: corev1.VolumeSource{
									Secret: &corev1.SecretVolumeSource{
										SecretName:  "gateway-letsencrypt-secret",
										DefaultMode: int32Ptr(0440),
									},
								},
							},
							{
								Name: "infra-api-kubeconfig",
								VolumeSource: corev1.VolumeSource{
									Secret: &corev1.SecretVolumeSource{
										SecretName: "infra-api-kubeconfig",
									},
								},
							},
							{
								Name: "sql-cert",
								VolumeSource: corev1.VolumeSource{
									Secret: &corev1.SecretVolumeSource{
										SecretName:  "tidb-gateway-cluster-secret",
										DefaultMode: int32Ptr(0440),
									},
								},
							},
							{
								Name: "cluster-cert",
								VolumeSource: corev1.VolumeSource{
									Secret: &corev1.SecretVolumeSource{
										SecretName:  "tidb-gateway-cluster-secret",
										DefaultMode: int32Ptr(0440),
									},
								},
							},
							{
								Name: "infra-redis-client-cert",
								VolumeSource: corev1.VolumeSource{
									Secret: &corev1.SecretVolumeSource{
										SecretName:  "tidb-gateway-infra-redis-client-secret",
										DefaultMode: int32Ptr(0440),
									},
								},
							},
							{
								Name: "dynamic-config",
								VolumeSource: corev1.VolumeSource{
									ConfigMap: &corev1.ConfigMapVolumeSource{
										LocalObjectReference: corev1.LocalObjectReference{
											Name: "tidb-gateway-dynamic-config",
										},
										DefaultMode: int32Ptr(0444),
									},
								},
							},
						},
						Tolerations: []corev1.Toleration{
							{
								Key:      "serverless.tidbcloud.com/node",
								Operator: corev1.TolerationOpEqual,
								Value:    "critical-services",
								Effect:   corev1.TaintEffectNoSchedule,
							},
						},
						ServiceAccountName: "tidb-gateway",
					},
				},
			},
		}

		if r.Config.CloudProvider == "alicloud" {
			deployment.Spec.Template.Annotations["goatscaler.io/safe-to-evict"] = "false"
		} else {
			deployment.Spec.Template.Annotations["cluster-autoscaler.kubernetes.io/safe-to-evict"] = "false"
		}

		if r.Config.InfraRegion != "" {
			deployment.Spec.Template.Spec.Containers[0].Env = append(
				deployment.Spec.Template.Spec.Containers[0].Env,
				corev1.EnvVar{
					Name:  "AWS_ROLE_ARN",
					Value: r.Config.InfraRegionRole,
				},
				corev1.EnvVar{
					Name:  "AWS_WEB_IDENTITY_TOKEN_FILE",
					Value: "/var/run/secrets/eks.amazonaws.com/serviceaccount/token",
				},
				corev1.EnvVar{
					Name:  "AWS_REGION",
					Value: r.Config.InfraRegion,
				},
				corev1.EnvVar{
					Name:  "AWS_ENDPOINT_URL_DYNAMODB",
					Value: r.Config.DynamoDBEndpoint,
				},
			)
			deployment.Spec.Template.Spec.Volumes = append(
				deployment.Spec.Template.Spec.Volumes,
				corev1.Volume{
					Name: "aws-iam-token",
					VolumeSource: corev1.VolumeSource{
						Projected: &corev1.ProjectedVolumeSource{
							Sources: []corev1.VolumeProjection{
								{
									ServiceAccountToken: &corev1.ServiceAccountTokenProjection{
										Path:              "token",
										ExpirationSeconds: int64Ptr(86400),
										Audience:          "sts.amazonaws.com",
									},
								},
							},
						},
					},
				},
			)
			deployment.Spec.Template.Spec.Containers[0].VolumeMounts = append(
				deployment.Spec.Template.Spec.Containers[0].VolumeMounts,
				corev1.VolumeMount{
					Name:      "aws-iam-token",
					MountPath: "/var/run/secrets/eks.amazonaws.com/serviceaccount",
					ReadOnly:  true,
				},
			)
		}

		if r.Config.E2eEnv {
			deployment.Spec.Template.Spec.Containers[0].Resources = corev1.ResourceRequirements{}
		}

		if gateway.Spec.TerminationGracePeriodSeconds > 0 {
			deployment.Spec.Template.Spec.TerminationGracePeriodSeconds = &gateway.Spec.TerminationGracePeriodSeconds
		}

		if err := controllerutil.SetControllerReference(gateway, &deployment, r.Scheme); err != nil {
			return nil, fmt.Errorf("failed to set controller reference: %w", err)
		}

		logger.Info("creating deployment", "name", deployment.Name, "namespace", deployment.Namespace)
		if err := r.Create(ctx, &deployment); err != nil {
			return nil, fmt.Errorf("failed to create deployment: %w", err)
		}
	}

	desiredResources := buildGatewayResourceRequirements(gateway)
	if !gatewayResourceRequirementsEqual(deployment.Spec.Template.Spec.Containers[0].Resources, desiredResources) && !r.Config.E2eEnv {
		deployment.Spec.Template.Spec.Containers[0].Resources = desiredResources
		logger.Info("updating deployment resources", "name", deployment.Name, "namespace", deployment.Namespace)
		if err := r.Update(ctx, &deployment); err != nil {
			return nil, fmt.Errorf("failed to update deployment resources: %w", err)
		}
	}

	desiredTopologySpreadConstraints := r.buildTopologySpreadConstraints(gateway, versionLabel)
	if !reflect.DeepEqual(deployment.Spec.Template.Spec.TopologySpreadConstraints, desiredTopologySpreadConstraints) {
		deployment.Spec.Template.Spec.TopologySpreadConstraints = desiredTopologySpreadConstraints
		logger.Info("updating deployment topology spread constraints", "name", deployment.Name, "namespace", deployment.Namespace)
		if err := r.Update(ctx, &deployment); err != nil {
			return nil, fmt.Errorf("failed to update deployment topology spread constraints: %w", err)
		}
	}

	return &deployment, nil
}

func buildGatewayResourceRequirements(gateway *gatewayv1alpha1.Gateway) corev1.ResourceRequirements {
	cpuLimit := resource.MustParse(gateway.Spec.CpuLimit)
	memoryLimit := resource.MustParse(gateway.Spec.MemoryLimit)

	return corev1.ResourceRequirements{
		Requests: corev1.ResourceList{
			corev1.ResourceCPU:    cpuRequestFromLimit(cpuLimit, gateway.Spec.CpuRequestRatio),
			corev1.ResourceMemory: memoryLimit,
		},
		Limits: corev1.ResourceList{
			corev1.ResourceCPU:    cpuLimit,
			corev1.ResourceMemory: memoryLimit,
		},
	}
}

func cpuRequestFromLimit(cpuLimit resource.Quantity, ratio float64) resource.Quantity {
	if ratio <= 0 {
		ratio = 1
	}

	requestMilliCPU := int64(math.Round(float64(cpuLimit.MilliValue()) * ratio))
	if requestMilliCPU < 1 {
		requestMilliCPU = 1
	}
	return *resource.NewMilliQuantity(requestMilliCPU, resource.DecimalSI)
}

func gatewayResourceRequirementsEqual(a, b corev1.ResourceRequirements) bool {
	return resourceListEqual(a.Requests, b.Requests) && resourceListEqual(a.Limits, b.Limits)
}

func resourceListEqual(a, b corev1.ResourceList) bool {
	if len(a) != len(b) {
		return false
	}
	for name, aQuantity := range a {
		bQuantity, ok := b[name]
		if !ok || aQuantity.Cmp(bQuantity) != 0 {
			return false
		}
	}
	return true
}

func (r *GatewayReconciler) changeHPATarget(ctx context.Context, gateway *gatewayv1alpha1.Gateway) (*autoscalingv2.HorizontalPodAutoscaler, error) {
	logger := logf.FromContext(ctx)
	versionLabel := getVersionLabel(gateway)
	hpaName := fmt.Sprintf("%s-hpa", gateway.Name)

	var hpa autoscalingv2.HorizontalPodAutoscaler
	if err := r.Get(ctx, client.ObjectKey{Name: hpaName, Namespace: gateway.Namespace}, &hpa); err != nil {
		if client.IgnoreNotFound(err) != nil {
			return nil, fmt.Errorf("failed to get HPA: %w", err)
		}

		hpa = autoscalingv2.HorizontalPodAutoscaler{
			TypeMeta: metav1.TypeMeta{
				Kind:       "HorizontalPodAutoscaler",
				APIVersion: autoscalingv2.SchemeGroupVersion.String(),
			},
			ObjectMeta: metav1.ObjectMeta{
				Name:      hpaName,
				Namespace: gateway.Namespace,
				Labels: map[string]string{
					"control-plane": "gateway-controller",
				},
			},
			Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
				ScaleTargetRef: autoscalingv2.CrossVersionObjectReference{
					Kind:       "Deployment",
					Name:       versionLabel,
					APIVersion: "apps/v1",
				},
				MinReplicas: int32Ptr(gateway.Spec.Replicas),
				MaxReplicas: 20,
				Metrics: []autoscalingv2.MetricSpec{
					{
						Type: autoscalingv2.ResourceMetricSourceType,
						Resource: &autoscalingv2.ResourceMetricSource{
							Name: corev1.ResourceCPU,
							Target: autoscalingv2.MetricTarget{
								Type:               autoscalingv2.UtilizationMetricType,
								AverageUtilization: int32Ptr(50),
							},
						},
					},
					{
						Type: autoscalingv2.ResourceMetricSourceType,
						Resource: &autoscalingv2.ResourceMetricSource{
							Name: corev1.ResourceMemory,
							Target: autoscalingv2.MetricTarget{
								Type:               autoscalingv2.UtilizationMetricType,
								AverageUtilization: int32Ptr(50),
							},
						},
					},
				},
				Behavior: &autoscalingv2.HorizontalPodAutoscalerBehavior{
					ScaleDown: &autoscalingv2.HPAScalingRules{
						StabilizationWindowSeconds: int32Ptr(1800),
					},
				},
			},
		}

		if err := controllerutil.SetControllerReference(gateway, &hpa, r.Scheme); err != nil {
			return nil, fmt.Errorf("failed to set controller reference for HPA resource: %w", err)
		}

		logger.Info("creating HPA resource", "name", hpa.Name, "namespace", hpa.Namespace)
		if err := r.Create(ctx, &hpa); err != nil {
			return nil, fmt.Errorf("failed to create HPA resource: %w", err)
		}
	}

	if hpa.Spec.ScaleTargetRef.Name != versionLabel {
		hpa.Spec.ScaleTargetRef.Name = versionLabel
		if err := r.Update(ctx, &hpa); err != nil {
			return nil, fmt.Errorf("failed to update HPA resource: %w", err)
		}
		logger.Info("updated HPA resource target", "name", hpa.Name, "namespace", hpa.Namespace, "version", versionLabel)
	}
	if hpa.Spec.MinReplicas == nil || *hpa.Spec.MinReplicas != gateway.Spec.Replicas {
		hpa.Spec.MinReplicas = int32Ptr(gateway.Spec.Replicas)
		if err := r.Update(ctx, &hpa); err != nil {
			return nil, fmt.Errorf("failed to update HPA resource: %w", err)
		}
		logger.Info("updated HPA resource min replicas", "name", hpa.Name, "namespace", hpa.Namespace, "minReplicas", gateway.Spec.Replicas)
	}
	if *hpa.Spec.Metrics[0].Resource.Target.AverageUtilization != 50 || *hpa.Spec.Metrics[1].Resource.Target.AverageUtilization != 50 {
		hpa.Spec.Metrics[0].Resource.Target.AverageUtilization = int32Ptr(50)
		hpa.Spec.Metrics[1].Resource.Target.AverageUtilization = int32Ptr(50)
		if err := r.Update(ctx, &hpa); err != nil {
			return nil, fmt.Errorf("failed to update HPA resource: %w", err)
		}
		logger.Info("updated HPA resource metrics", "name", hpa.Name, "namespace", hpa.Namespace)
	}
	if hpa.Spec.MaxReplicas != 20 {
		hpa.Spec.MaxReplicas = 20
		if err := r.Update(ctx, &hpa); err != nil {
			return nil, fmt.Errorf("failed to update HPA resource max replicas: %w", err)
		}
		logger.Info("updated HPA resource max replicas", "name", hpa.Name, "namespace", hpa.Namespace, "maxReplicas", 20)
	}

	return &hpa, nil
}

func (r *GatewayReconciler) changeServiceLabel(ctx context.Context, gateway *gatewayv1alpha1.Gateway) (*corev1.Service, error) {
	logger := logf.FromContext(ctx)
	versionLabel := getVersionLabel(gateway)

	var service corev1.Service
	if err := r.Get(ctx, client.ObjectKey{Name: gateway.Name, Namespace: gateway.Namespace}, &service); err != nil {
		if client.IgnoreNotFound(err) != nil {
			return nil, fmt.Errorf("failed to get service: %w", err)
		}

		serviceSpec := corev1.ServiceSpec{
			Selector: map[string]string{
				"version": versionLabel,
			},
			Ports: []corev1.ServicePort{
				{
					Name:       "mysql-client-4000",
					Port:       4000,
					Protocol:   corev1.ProtocolTCP,
					TargetPort: intstr.FromInt(4000),
				},
				{
					Name:       "mysql-client-3306",
					Port:       3306,
					Protocol:   corev1.ProtocolTCP,
					TargetPort: intstr.FromInt(3306),
				},
			},
		}
		if r.serviceTrafficDistributionSupported() {
			serviceSpec.TrafficDistribution = stringPtr(corev1.ServiceTrafficDistributionPreferClose)
		}

		service = corev1.Service{
			TypeMeta: metav1.TypeMeta{
				APIVersion: "v1",
				Kind:       "Service",
			},
			ObjectMeta: metav1.ObjectMeta{
				Name:      gateway.Name,
				Namespace: gateway.Namespace,
				Labels: map[string]string{
					"control-plane": "gateway-controller",
				},
				Annotations: map[string]string{
					"haproxy.org/send-proxy-protocol": "proxy-v2",
				},
			},
			Spec: serviceSpec,
		}
		if err := controllerutil.SetControllerReference(gateway, &service, r.Scheme); err != nil {
			return nil, fmt.Errorf("failed to set controller reference: %w", err)
		}

		logger.Info("creating service", "name", service.Name, "namespace", service.Namespace)
		if err := r.Create(ctx, &service); err != nil {
			return nil, fmt.Errorf("failed to create service: %w", err)
		}

		return &service, nil
	}

	needUpdate := false
	if service.Spec.Selector == nil || service.Spec.Selector["version"] != versionLabel {
		service.Spec.Selector = map[string]string{"version": versionLabel}
		needUpdate = true
	}

	// Privatelink and ZR HAProxy should be blind to proxy protocol, remove the annotation if exists
	_, sendProxy := service.Annotations["haproxy.org/send-proxy-protocol"]
	if (gateway.Spec.IngressClass == ingressClassPrivateLink || gateway.Spec.IngressClass == ingressClassPrivateLinkZR) && sendProxy {
		service.Annotations = nil
		needUpdate = true
	}

	if service.Spec.TrafficDistribution == nil && r.serviceTrafficDistributionSupported() {
		service.Spec.TrafficDistribution = stringPtr(corev1.ServiceTrafficDistributionPreferClose)
		needUpdate = true
	}

	if needUpdate {
		if err := r.Update(ctx, &service); err != nil {
			return nil, fmt.Errorf("failed to update service: %w", err)
		}
		logger.Info("updated service selector", "name", service.Name, "namespace", service.Namespace)
	}
	return &service, nil
}

func (r *GatewayReconciler) createMonitorResources(ctx context.Context, gateway *gatewayv1alpha1.Gateway) error {
	logger := logf.FromContext(ctx)
	serviceName := fmt.Sprintf("%s-metrics", gateway.Name)

	var service corev1.Service
	if err := r.Get(ctx, client.ObjectKey{Name: serviceName, Namespace: gateway.Namespace}, &service); err != nil {
		if client.IgnoreNotFound(err) != nil {
			return fmt.Errorf("failed to get service: %w", err)
		}

		service = corev1.Service{
			TypeMeta: metav1.TypeMeta{
				APIVersion: "v1",
				Kind:       "Service",
			},
			ObjectMeta: metav1.ObjectMeta{
				Name:      serviceName,
				Namespace: gateway.Namespace,
				Labels: map[string]string{
					"control-plane": "gateway-controller",
					"app":           gateway.Name,
				},
			},
			Spec: corev1.ServiceSpec{
				Selector: map[string]string{
					"control-plane": "gateway-controller",
				},
				Ports: []corev1.ServicePort{
					{
						Name:       "metrics",
						Port:       9090,
						Protocol:   corev1.ProtocolTCP,
						TargetPort: intstr.FromInt(9090),
					},
				},
			},
		}
		if err := controllerutil.SetControllerReference(gateway, &service, r.Scheme); err != nil {
			return fmt.Errorf("failed to set controller reference: %w", err)
		}

		logger.Info("creating service", "name", service.Name, "namespace", service.Namespace)
		if err := r.Create(ctx, &service); err != nil {
			return fmt.Errorf("failed to create service: %w", err)
		}

		return nil
	}

	if service.Spec.Selector == nil || service.Spec.Selector["control-plane"] != "gateway-controller" {
		service.Spec.Selector = map[string]string{"control-plane": "gateway-controller", "app": gateway.Name}
		if err := r.Update(ctx, &service); err != nil {
			return fmt.Errorf("failed to update service: %w", err)
		}
		logger.Info("updated service selector", "name", service.Name, "namespace", service.Namespace)
	}

	var serviceMonitor monitoringv1.ServiceMonitor
	if err := r.Get(ctx, client.ObjectKey{Name: serviceName, Namespace: gateway.Namespace}, &serviceMonitor); err != nil {
		if client.IgnoreNotFound(err) != nil {
			return fmt.Errorf("failed to get ServiceMonitor: %w", err)
		}
		serviceMonitor = monitoringv1.ServiceMonitor{
			TypeMeta: metav1.TypeMeta{
				Kind:       "ServiceMonitor",
				APIVersion: monitoringv1.SchemeGroupVersion.String(),
			},
			ObjectMeta: metav1.ObjectMeta{
				Name:      serviceName,
				Namespace: gateway.Namespace,
				Labels: map[string]string{
					"control-plane": "gateway-controller",
				},
			},
			Spec: monitoringv1.ServiceMonitorSpec{
				Selector: metav1.LabelSelector{
					MatchLabels: map[string]string{
						"control-plane": "gateway-controller",
						"app":           gateway.Name,
					},
				},
				Endpoints: []monitoringv1.Endpoint{
					{
						Port:        "metrics",
						Path:        "/metrics",
						HonorLabels: true,
						Interval:    gatewayMetricsInterval,
					},
				},
				NamespaceSelector: monitoringv1.NamespaceSelector{
					MatchNames: []string{gateway.Namespace},
				},
			},
		}
		if err := controllerutil.SetControllerReference(gateway, &serviceMonitor, r.Scheme); err != nil {
			return fmt.Errorf("failed to set controller reference for ServiceMonitor: %w", err)
		}
		logger.Info("creating ServiceMonitor", "name", serviceMonitor.Name, "namespace", serviceMonitor.Namespace)
		if err := r.Create(ctx, &serviceMonitor); err != nil {
			return fmt.Errorf("failed to create ServiceMonitor: %w", err)
		}
	} else if len(serviceMonitor.Spec.Endpoints) == 0 ||
		serviceMonitor.Spec.Endpoints[0].Interval != gatewayMetricsInterval {
		if len(serviceMonitor.Spec.Endpoints) == 0 {
			serviceMonitor.Spec.Endpoints = []monitoringv1.Endpoint{{
				Port:        "metrics",
				Path:        "/metrics",
				HonorLabels: true,
				Interval:    gatewayMetricsInterval,
			}}
		} else {
			serviceMonitor.Spec.Endpoints[0].Interval = gatewayMetricsInterval
		}
		if err := r.Update(ctx, &serviceMonitor); err != nil {
			return fmt.Errorf("failed to update ServiceMonitor: %w", err)
		}
		logger.Info("updated ServiceMonitor scrape interval", "name", serviceMonitor.Name,
			"namespace", serviceMonitor.Namespace, "interval", gatewayMetricsInterval)
	}

	return nil
}

func (r *GatewayReconciler) createTCPIfNotExists(ctx context.Context, gateway *gatewayv1alpha1.Gateway) (*haproxyv1.TCP, error) {
	var tcp haproxyv1.TCP
	logger := logf.FromContext(ctx)
	if err := r.Get(ctx, client.ObjectKey{Name: gateway.Name, Namespace: gateway.Namespace}, &tcp); err != nil {
		if client.IgnoreNotFound(err) != nil {
			return nil, fmt.Errorf("failed to get TCP resource: %w", err)
		}

		tcp := haproxyv1.TCP{
			ObjectMeta: metav1.ObjectMeta{
				Name:      gateway.Name,
				Namespace: gateway.Namespace,
				Labels:    map[string]string{"control-plane": "gateway-controller"},
				Annotations: map[string]string{
					"ingress.class": gateway.Spec.IngressClass,
				},
			},
			Spec: haproxyv1.TCPSpec{
				{
					Name: "3306",
					Frontend: haproxyv1.SectionFrontend{
						Frontend: models.Frontend{
							Name:          fmt.Sprintf("%s-3306-frontend", gateway.Name),
							Tcplog:        true,
							Clitcpka:      "enabled",
							ClitcpkaIdle:  int64Ptr(300000),
							ClitcpkaIntvl: int64Ptr(75000),
						},
						Binds: []*models.Bind{
							{
								BindParams: models.BindParams{
									Name: ":3306",
								},
								Port: int64Ptr(3306),
							},
						},
					},
					Service: haproxyv1.TCPService{
						Name: gateway.Name,
						Port: 3306,
					},
				},
				{
					Name: "4000",
					Frontend: haproxyv1.SectionFrontend{
						Frontend: models.Frontend{
							Name:          fmt.Sprintf("%s-4000-frontend", gateway.Name),
							Tcplog:        true,
							Clitcpka:      "enabled",
							ClitcpkaIdle:  int64Ptr(300000),
							ClitcpkaIntvl: int64Ptr(75000),
						},
						Binds: []*models.Bind{
							{
								BindParams: models.BindParams{
									Name: ":4000",
								},
								Port: int64Ptr(4000),
							},
						},
					},
					Service: haproxyv1.TCPService{
						Name: gateway.Name,
						Port: 4000,
					},
				},
			},
		}
		if err := controllerutil.SetControllerReference(gateway, &tcp, r.Scheme); err != nil {
			return nil, fmt.Errorf("failed to set controller reference for TCP resource: %w", err)
		}

		logger.Info("creating TCP resource", "name", tcp.Name, "namespace", tcp.Namespace)
		if err := r.Create(ctx, &tcp); err != nil {
			return nil, fmt.Errorf("failed to create TCP resource: %w", err)
		}
		return &tcp, nil
	}

	needUpdate := false
	for i := range tcp.Spec {
		if tcp.Spec[i].Frontend.Clitcpka != "enabled" || tcp.Spec[i].Frontend.Name != fmt.Sprintf("%s-%s-frontend", gateway.Name, tcp.Spec[i].Name) {
			tcp.Spec[i].Frontend.Name = fmt.Sprintf("%s-%s-frontend", gateway.Name, tcp.Spec[i].Name)
			tcp.Spec[i].Frontend.Clitcpka = "enabled"
			tcp.Spec[i].Frontend.ClitcpkaIdle = int64Ptr(300000)
			tcp.Spec[i].Frontend.ClitcpkaIntvl = int64Ptr(75000)
			needUpdate = true
		}
	}
	if needUpdate {
		if err := r.Update(ctx, &tcp); err != nil {
			return nil, fmt.Errorf("failed to update TCP resource: %w", err)
		}
		logger.Info("updated TCP resource", "name", tcp.Name, "namespace", tcp.Namespace)
	}

	return &tcp, nil
}

func (r *GatewayReconciler) recycleOldVersionDeployments(ctx context.Context, gateway *gatewayv1alpha1.Gateway) (ctrl.Result, error) {
	logger := logf.FromContext(ctx)
	versionLabel := getVersionLabel(gateway)

	var deployments appsv1.DeploymentList
	if err := r.List(ctx, &deployments, client.InNamespace(gateway.Namespace), client.MatchingFields{ownerKey: gateway.Name}); err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to list deployments: %w", err)
	}

	elapsed := time.Since(gateway.Status.ApplyTime.Time)
	timeout := gateway.Spec.GracefulCloseConnectionsTimeout.Duration
	if elapsed < timeout {
		if !r.Config.EnableActiveConnectionDetection {
			logger.Info("skipping active connection detection, waiting for timeout", "elapsed", elapsed, "timeout", timeout)
			return ctrl.Result{RequeueAfter: time.Minute * 5}, nil
		}
		for _, deployment := range deployments.Items {
			if r.shouldSkipVersion(&deployment, versionLabel) {
				continue
			}
			activeConnections, err := r.getActiveConnections(ctx, &deployment)
			if err != nil {
				logger.Error(err, "failed to get active connections", "name", deployment.Name, "namespace", deployment.Namespace)
				continue
			}
			if activeConnections > 0 {
				continue
			}
			if err := r.scaleDeploymentToZero(ctx, &deployment); err != nil {
				logger.Error(err, "failed to scale deployment to zero", "name", deployment.Name, "namespace", deployment.Namespace)
				continue
			}
			logger.Info("scaled deployment to zero when no active connections", "name", deployment.Name, "namespace", deployment.Namespace)
		}
		return ctrl.Result{RequeueAfter: time.Minute * 5}, nil
	} else {
		for _, deployment := range deployments.Items {
			if r.shouldSkipVersion(&deployment, versionLabel) {
				continue
			}
			err := r.scaleDeploymentToZero(ctx, &deployment)
			if err != nil {
				logger.Error(err, "failed to scale deployment to zero", "name", deployment.Name, "namespace", deployment.Namespace)
				return ctrl.Result{}, err
			}
			logger.Info("scaled deployment to zero when timeout", "name", deployment.Name, "namespace", deployment.Namespace)
		}

		if len(deployments.Items) > int(gateway.Spec.MaxDeployments) {
			logger.Info("recycling old deployments", "currentDeployments", len(deployments.Items), "maxDeployments", gateway.Spec.MaxDeployments)
			sort.Slice(deployments.Items, func(i, j int) bool {
				return deployments.Items[i].CreationTimestamp.Before(&deployments.Items[j].CreationTimestamp)
			})
			for i := 0; i < len(deployments.Items)-int(gateway.Spec.MaxDeployments); i++ {
				deployment := deployments.Items[i]
				if *deployment.Spec.Replicas != 0 {
					// skip non zero replica deployments
					continue
				}
				if err := r.Delete(ctx, &deployment); err != nil {
					logger.Error(err, "failed to delete old deployment", "name", deployment.Name, "namespace", deployment.Namespace)
					return ctrl.Result{}, fmt.Errorf("failed to delete old deployment %s: %w", deployment.Name, err)
				}
				logger.Info("deleted old deployment", "name", deployment.Name, "namespace", deployment.Namespace)
			}
		}
		return ctrl.Result{}, nil
	}
}

func (r *GatewayReconciler) shouldSkipVersion(deployment *appsv1.Deployment, currentVersion string) bool {
	if deployment.Name == currentVersion {
		// skipping current version deployment
		return true
	}
	if *deployment.Spec.Replicas == 0 {
		// skip zero replica deployments
		return true
	}
	return false
}

func (r *GatewayReconciler) scaleDeploymentToZero(ctx context.Context, deployment *appsv1.Deployment) error {
	deployment.Spec.Replicas = int32Ptr(0)
	if err := r.Update(ctx, deployment); err != nil {
		return fmt.Errorf("failed to update deployment %s: %w", deployment.Name, err)
	}
	return nil
}

func (r *GatewayReconciler) getActiveConnections(ctx context.Context, deployment *appsv1.Deployment) (int64, error) {
	if r.Querier == nil {
		return 0, fmt.Errorf("prometheus querier is not initialized")
	}

	v := map[string]string{
		"query": fmt.Sprintf(`sum(gateway_connection_gauge{pod=~"%s(.*)"})`, deployment.Name),
		"_":     strconv.FormatInt(time.Now().Unix(), 10), // a dummy value just avoid cache
	}
	resp, err := r.Querier.QueryMetrics(ctx, "/api/v1/query", v)
	if err != nil {
		return 0, fmt.Errorf("failed to query active connections: %w", err)
	}

	if len(resp.Data.Result) == 0 {
		return 0, nil
	}

	value, err := strconv.ParseInt(resp.Data.Result[0].Value[1].(string), 10, 64)
	if err != nil {
		return 0, fmt.Errorf("failed to parse active connections value: %w", err)
	}

	return value, nil
}

func getVersionLabel(gateway *gatewayv1alpha1.Gateway) string {
	return fmt.Sprintf("%s-%s", gateway.Name, gateway.Spec.Version)
}

func isDeploymentReady(deployment *appsv1.Deployment) bool {
	if deployment.Status.Replicas == 0 {
		return false
	}
	if deployment.Status.ReadyReplicas < deployment.Status.Replicas {
		return false
	}
	if deployment.Status.AvailableReplicas < deployment.Status.Replicas {
		return false
	}
	return true
}

func int32Ptr(i int32) *int32 {
	return &i
}

func int64Ptr(i int64) *int64 {
	return &i
}

func boolPtr(b bool) *bool {
	return &b
}

func stringPtr(s string) *string {
	return &s
}

func (r *GatewayReconciler) serviceTrafficDistributionSupported() bool {
	if r.kubernetesVersion == nil {
		return false
	}

	return r.kubernetesVersion.AtLeast(utilversion.MajorMinor(1, 31))
}

// SetupWithManager sets up the controller with the Manager.
func (r *GatewayReconciler) SetupWithManager(mgr ctrl.Manager) error {
	r.setupKubernetesVersion(mgr)

	NameFilter := func(object client.Object) bool {
		return object.GetNamespace() == r.Config.Namespace
	}
	byNamespaceFuncs := predicate.NewPredicateFuncs(NameFilter)
	if err := mgr.GetFieldIndexer().IndexField(context.Background(), &appsv1.Deployment{}, ownerKey, func(rawObj client.Object) []string {
		deployment := rawObj.(*appsv1.Deployment)
		owner := metav1.GetControllerOf(deployment)
		if owner == nil {
			return nil
		}
		if owner.APIVersion != apiGVStr || owner.Kind != "Gateway" {
			return nil
		}
		return []string{owner.Name}
	}); err != nil {
		return err
	}
	return ctrl.NewControllerManagedBy(mgr).
		WithEventFilter(byNamespaceFuncs).
		For(&gatewayv1alpha1.Gateway{}).
		WithEventFilter(byNamespaceFuncs).
		Owns(&appsv1.Deployment{}).
		Owns(&corev1.Service{}).
		Owns(&haproxyv1.TCP{}).
		Owns(&autoscalingv2.HorizontalPodAutoscaler{}).
		Owns(&monitoringv1.ServiceMonitor{}).
		Named("gateway").
		Complete(r)
}

func (r *GatewayReconciler) setupKubernetesVersion(mgr ctrl.Manager) {
	logger := logf.Log.WithName("gateway")

	discoveryClient, err := discovery.NewDiscoveryClientForConfig(mgr.GetConfig())
	if err != nil {
		logger.Error(err, "failed to create discovery client")
		return
	}

	serverVersion, err := discoveryClient.ServerVersion()
	if err != nil {
		logger.Error(err, "failed to get Kubernetes server version")
		return
	}

	r.kubernetesVersion = utilversion.MustParseMajorMinor(serverVersion.String())

	logger.Info("detected Kubernetes server version",
		"serverVersion", serverVersion,
		"kubernetesVersion", r.kubernetesVersion,
		"serviceTrafficDistributionSupported", r.serviceTrafficDistributionSupported())
}

// buildContainerArgs builds the container arguments based on gateway type
func (r *GatewayReconciler) buildContainerArgs(gateway *gatewayv1alpha1.Gateway, gracefulWaitBeforeCloseServer int64) []string {
	baseArgs := []string{
		"--addr=:4000,:3306",
		"--infra-kubeconfig=/infra/kubeconfig",
		"--tls-cert=/gateway-tls/tls.crt",
		"--tls-key=/gateway-tls/tls.key",
		fmt.Sprintf("--region=%s", r.Config.Region),
		fmt.Sprintf("--infra-redis-addr=%s:6380", r.Config.InfraRedisAddr),
		fmt.Sprintf("--infra-cluster-table=serverless_cluster_%s", r.Config.Env),
		fmt.Sprintf("--cloud-provider-type=%s-free-tier", r.Config.CloudProvider),
		fmt.Sprintf("--kube-cluster-name=%s", r.Config.ClusterName),
		fmt.Sprintf("--push-traffic-metrics-addr=%s", r.Config.PushTrafficMetricsAddr),
		fmt.Sprintf("--push-traffic-metrics-interval=%d", r.Config.PushTrafficMetricsInterval),
		"--sql-tls-ca=/sql-cert/ca.crt",
		"--sql-tls-cert=/sql-cert/tls.crt",
		"--sql-tls-key=/sql-cert/tls.key",
		"--cluster-tls-ca=/cluster-cert/ca.crt",
		"--cluster-tls-cert=/cluster-cert/tls.crt",
		"--cluster-tls-key=/cluster-cert/tls.key",
		"--require-backend-tls=true",
		fmt.Sprintf("--tidb-server-version=%s", r.Config.TiDBServerVersion),
		fmt.Sprintf("--enable-infra-mysql-client=%v", r.Config.EnableInfraMysqlClient),
		fmt.Sprintf("--infra-rds-secret-arn=%s", r.Config.InfraRdsSecretArn),
		fmt.Sprintf("--cache-server-addr=%s", r.Config.CacheServerAddr),
		fmt.Sprintf("--graceful-wait-before-close-server=%d", gracefulWaitBeforeCloseServer),
		fmt.Sprintf("--frontend-read-timeout=%d", r.Config.FrontendReadTimeout),
	}
	if r.Config.E2eEnv {
		baseArgs = append(baseArgs, "--e2e-env=true")
	}

	// Add gateway-mode based on ingress class
	switch {
	case matchIngressClass(gateway.Spec.IngressClass, ingressClassPrivateLinkZR):
		baseArgs = append(baseArgs, "--gateway-mode=privatelink")
		baseArgs = append(baseArgs, "--only-for-zr=true")
	case matchIngressClass(gateway.Spec.IngressClass, ingressClassPrivateLink):
		baseArgs = append(baseArgs, "--gateway-mode=privatelink")
	case matchIngressClass(gateway.Spec.IngressClass, ingressClassPublic):
		baseArgs = append(baseArgs, "--gateway-mode=public")
		baseArgs = append(baseArgs, "--tls-version=TLSv1.2")
		baseArgs = append(baseArgs, "--tls-only=true")
		// Add TLS skip user prefixes if configured
		for _, prefix := range r.Config.TLSOnlySkipUserPrefixes {
			baseArgs = append(baseArgs, fmt.Sprintf("--tls-only-skip-user-prefix=%s", prefix))
		}
	}

	// Add OTS instance name if configured
	if r.Config.OtsInstanceName != "" {
		baseArgs = append(baseArgs, fmt.Sprintf("--ots-instance-name=%s", r.Config.OtsInstanceName))
	}

	return baseArgs
}

func (r *GatewayReconciler) getConfiguredZoneAffinityValues(gateway *gatewayv1alpha1.Gateway) ([]string, bool) {
	if r.Config == nil {
		return nil, false
	}
	if zones, ok := r.Config.ZoneAffinity[gateway.Spec.IngressClass]; ok && len(zones) > 0 {
		return zones, true
	}
	return nil, false
}

// buildPodLabels builds pod labels based on ingress class
func (r *GatewayReconciler) buildPodLabels(gateway *gatewayv1alpha1.Gateway, versionLabel string) map[string]string {
	baseLabels := map[string]string{
		"control-plane":               "gateway-controller",
		"version":                     versionLabel,
		"app":                         gateway.Name,
		"app.kubernetes.io/instance":  "serverless-cluster",
		"app.kubernetes.io/component": "tidb-gateway",
		"app.kubernetes.io/name":      "gateway",
	}

	// Add ingress class specific labels if needed
	baseLabels["gateway.tidbcloud.com/ingress-class"] = gateway.Spec.IngressClass

	if gateway.Spec.IngressClass != ingressClassPrivateLinkZR {
		baseLabels["zr-evicter.tidbcloud.com/ignored"] = "true"
	}

	return baseLabels
}

// buildNodeAffinity builds node affinity based on gateway type and cloud provider
func (r *GatewayReconciler) buildNodeAffinity(gateway *gatewayv1alpha1.Gateway, versionLabel string) *corev1.Affinity {
	affinity := &corev1.Affinity{
		NodeAffinity: &corev1.NodeAffinity{
			RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
				NodeSelectorTerms: []corev1.NodeSelectorTerm{
					{
						MatchExpressions: []corev1.NodeSelectorRequirement{
							{
								Key:      "serverless.tidbcloud.com/node",
								Operator: corev1.NodeSelectorOpIn,
								Values:   []string{"critical-services"},
							},
						},
					},
				},
			},
		},
		PodAntiAffinity: &corev1.PodAntiAffinity{
			RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
				{
					LabelSelector: &metav1.LabelSelector{
						MatchLabels: map[string]string{
							"version": versionLabel,
						},
					},
					TopologyKey: "kubernetes.io/hostname",
				},
			},
		},
	}

	// Add zone affinity for AWS
	if r.Config.CloudProvider == "aws" {
		affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions = append(
			affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions,
			corev1.NodeSelectorRequirement{
				Key:      "serverless.tidbcloud.com/zone",
				Operator: corev1.NodeSelectorOpIn,
				Values:   []string{"primary", "standby"},
			},
		)
	}
	// Add zone affinity for AliCloud if configured
	if r.Config.CloudProvider == "alicloud" {
		zoneValues, ok := r.getConfiguredZoneAffinityValues(gateway)
		if ok {
			affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions = append(
				affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions,
				corev1.NodeSelectorRequirement{
					Key:      "topology.kubernetes.io/zone",
					Operator: corev1.NodeSelectorOpIn,
					Values:   zoneValues,
				},
			)
		}
	}
	if r.Config.KubeletVersion != "" {
		affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions = append(
			affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions,
			corev1.NodeSelectorRequirement{
				Key:      "serverless.tidbcloud.com/kubelet-version",
				Operator: corev1.NodeSelectorOpIn,
				Values:   []string{r.Config.KubeletVersion},
			},
		)
	}

	return affinity
}

// shouldSpreadAcrossZones reports whether a gateway's pods can actually be
// scheduled on nodes in multiple zones. Single-zone-pinned gateways (AWS
// public/privatelink) would otherwise end up with a no-op constraint that
// misleads future readers.
func (r *GatewayReconciler) shouldSpreadAcrossZones(gateway *gatewayv1alpha1.Gateway) bool {
	if gateway.Spec.IngressClass == ingressClassPrivateLinkZR {
		return true
	}
	if r.Config != nil && r.Config.CloudProvider == "alicloud" {
		if _, ok := r.getConfiguredZoneAffinityValues(gateway); !ok {
			return true
		}
	}
	return false
}

func (r *GatewayReconciler) buildTopologySpreadConstraints(gateway *gatewayv1alpha1.Gateway, versionLabel string) []corev1.TopologySpreadConstraint {
	if !r.shouldSpreadAcrossZones(gateway) {
		return nil
	}
	return []corev1.TopologySpreadConstraint{
		{
			MaxSkew:           1,
			TopologyKey:       corev1.LabelTopologyZone,
			WhenUnsatisfiable: corev1.DoNotSchedule,
			LabelSelector: &metav1.LabelSelector{
				MatchLabels: map[string]string{
					"version": versionLabel,
				},
			},
		},
	}
}