gateway_controller.go

// SPDX-License-Identifier: AGPL-3.0-only

package controller

import (
	"context"
	"crypto/tls"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"time"

	cmv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
	cmmeta "github.com/cert-manager/cert-manager/pkg/apis/meta/v1"
	envoygatewayv1alpha1 "github.com/envoyproxy/gateway/api/v1alpha1"
	"github.com/prometheus/client_golang/prometheus"
	corev1 "k8s.io/api/core/v1"
	discoveryv1 "k8s.io/api/discovery/v1"
	"k8s.io/apimachinery/pkg/api/equality"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	apimeta "k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/utils/ptr"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
	"sigs.k8s.io/controller-runtime/pkg/cluster"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
	gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
	mcbuilder "sigs.k8s.io/multicluster-runtime/pkg/builder"
	mchandler "sigs.k8s.io/multicluster-runtime/pkg/handler"
	mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager"
	"sigs.k8s.io/multicluster-runtime/pkg/multicluster"
	mcreconcile "sigs.k8s.io/multicluster-runtime/pkg/reconcile"
	mcsource "sigs.k8s.io/multicluster-runtime/pkg/source"

	networkingv1alpha "go.datum.net/network-services-operator/api/v1alpha"
	"go.datum.net/network-services-operator/internal/config"
	downstreamclient "go.datum.net/network-services-operator/internal/downstreamclient"
	gatewayutil "go.datum.net/network-services-operator/internal/util/gateway"
	"go.datum.net/network-services-operator/internal/util/resourcename"
	dnsv1alpha1 "go.miloapis.com/dns-operator/api/v1alpha1"
)

const gatewayControllerFinalizer = "gateway.networking.datumapis.com/gateway-controller"
const gatewayControllerGCFinalizer = "gateway.networking.datumapis.com/gateway-controller-gc"
const certificateIssuerTLSOption = "gateway.networking.datumapis.com/certificate-issuer"

// autoIssuerSentinel is the placeholder value the defaulting webhook stamps
// onto the operator-injected default-https listener via
// listenerTLSOptions. It means "use whichever real issuer the user's other
// TLS listener on this gateway specified" — see resolveAutoIssuer.
const autoIssuerSentinel = "auto"
const annotationReissuanceCount = "networking.datumapis.com/reissuance-count"
const KindGateway = "Gateway"
const KindHTTPRoute = "HTTPRoute"
const KindService = "Service"
const KindEndpointSlice = "EndpointSlice"

// GatewayReconciler reconciles a Gateway object
type GatewayReconciler struct {
	mgr    mcmanager.Manager
	Config config.NetworkServicesOperator

	DownstreamCluster cluster.Cluster
}

// +kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=core,resources=services/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=core,resources=services/finalizers,verbs=update
// +kubebuilder:rbac:groups=discovery.k8s.io,resources=endpointslices,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=discovery.k8s.io,resources=endpointslices/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=discovery.k8s.io,resources=endpointslices/finalizers,verbs=update

// +kubebuilder:rbac:groups=gateway.networking.k8s.io,resources=gateways,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=gateway.networking.k8s.io,resources=gateways/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=gateway.networking.k8s.io,resources=gateways/finalizers,verbs=update
// +kubebuilder:rbac:groups=gateway.networking.k8s.io,resources=httproutes,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=gateway.networking.k8s.io,resources=httproutes/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=gateway.networking.k8s.io,resources=httproutes/finalizers,verbs=update
// +kubebuilder:rbac:groups=gateway.networking.k8s.io,resources=backendtlspolicies,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=gateway.networking.k8s.io,resources=backendtlspolicies/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=gateway.networking.k8s.io,resources=backendtlspolicies/finalizers,verbs=update

// +kubebuilder:rbac:groups=cert-manager.io,resources=certificates,verbs=get;list;watch;create;update;patch;delete

// +kubebuilder:rbac:groups=externaldns.k8s.io,resources=dnsendpoints,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=externaldns.k8s.io,resources=dnsendpoints/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=externaldns.k8s.io,resources=dnsendpoints/finalizers,verbs=update

func (r *GatewayReconciler) Reconcile(ctx context.Context, req mcreconcile.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx, "cluster", req.ClusterName, "namespace", req.Namespace, jsonKeyName, req.Name)
	ctx = log.IntoContext(ctx, logger)

	logger.Info("gateway reconcile dequeued")

	cl, err := r.mgr.GetCluster(ctx, req.ClusterName)
	if err != nil {
		logger.Error(err, "failed to get cluster")
		return ctrl.Result{}, err
	}

	logger.Info("got cluster, fetching gateway")

	var gateway gatewayv1.Gateway
	if err := cl.GetClient().Get(ctx, req.NamespacedName, &gateway); err != nil {
		if apierrors.IsNotFound(err) {
			logger.Info("gateway not found, skipping")
			return ctrl.Result{}, nil
		}
		logger.Error(err, "failed to get gateway")
		return ctrl.Result{}, err
	}

	// Look up the GatewayClass to determine if it's applicable to this controller
	var upstreamGatewayClass gatewayv1.GatewayClass
	if err := cl.GetClient().Get(ctx, types.NamespacedName{Name: string(gateway.Spec.GatewayClassName)}, &upstreamGatewayClass); err != nil {
		if apierrors.IsNotFound(err) {
			logger.Info("gateway class not found, skipping")
			return ctrl.Result{}, nil
		}
		logger.Error(err, "failed to get gateway class")
		return ctrl.Result{}, err
	}

	if upstreamGatewayClass.Spec.ControllerName != r.Config.Gateway.ControllerName {
		logger.Info("gateway class controller name mismatch, skipping", "expected", r.Config.Gateway.ControllerName, "actual", upstreamGatewayClass.Spec.ControllerName)
		return ctrl.Result{}, nil
	}

	downstreamStrategy := downstreamclient.NewMappedNamespaceResourceStrategy(string(req.ClusterName), cl.GetClient(), r.DownstreamCluster.GetClient())

	if !gateway.DeletionTimestamp.IsZero() {
		if controllerutil.ContainsFinalizer(&gateway, gatewayControllerFinalizer) {
			if result := r.finalizeGateway(ctx, string(req.ClusterName), cl.GetClient(), &gateway, downstreamStrategy); result.ShouldReturn() {
				return result.Complete(ctx)
			}

			controllerutil.RemoveFinalizer(&gateway, gatewayControllerFinalizer)
			if err := cl.GetClient().Update(ctx, &gateway); err != nil {
				return ctrl.Result{}, fmt.Errorf("failed to remove finalizer from gateway: %w", err)
			}
		}

		return ctrl.Result{}, nil
	}

	if r.prepareUpstreamGateway(&gateway) {
		logger.Info("preparing upstream gateway (adding finalizer/defaults)")
		if err := cl.GetClient().Update(ctx, &gateway); err != nil {
			logger.Error(err, "failed preparing upstream gateway")
			return ctrl.Result{}, fmt.Errorf("failed preparing upstream gateway: %w", err)
		}

		return ctrl.Result{}, nil
	}

	logger.Info("reconciling gateway")
	defer logger.Info("reconcile complete")

	result, _ := r.ensureDownstreamGateway(ctx, string(req.ClusterName), cl.GetClient(), &gateway, downstreamStrategy)
	if result.ShouldReturn() {
		return result.Complete(ctx)
	}

	return result.Complete(ctx)
}

// prepareUpstreamGateway adds a finalizer to the upstream gateway and ensures
// that default listeners have their hostname fields set.
func (r *GatewayReconciler) prepareUpstreamGateway(gateway *gatewayv1.Gateway) (needsUpdate bool) {
	if !controllerutil.ContainsFinalizer(gateway, gatewayControllerFinalizer) {
		controllerutil.AddFinalizer(gateway, gatewayControllerFinalizer)
		needsUpdate = true
	}

	// Ensure that the default listeners have hostnames set. This is not done in
	// the defaulting webhook as the UID is not available, but we still want to
	// let users create gateways without listeners and receive sane defaults.

	gatewayDefaultHostname := ptr.To(gatewayv1.Hostname(r.gatewayCanonicalHostname(gateway)))

	defaultHTTPListener := gatewayutil.GetListenerByName(gateway.Spec.Listeners, gatewayutil.DefaultHTTPListenerName)
	if defaultHTTPListener != nil && defaultHTTPListener.Hostname == nil {
		needsUpdate = true
		defaultHTTPListener.Hostname = gatewayDefaultHostname
	}

	defaultHTTPSListener := gatewayutil.GetListenerByName(gateway.Spec.Listeners, gatewayutil.DefaultHTTPSListenerName)
	if defaultHTTPSListener != nil && defaultHTTPSListener.Hostname == nil {
		needsUpdate = true
		defaultHTTPSListener.Hostname = gatewayDefaultHostname
	}

	return needsUpdate
}

func (r *GatewayReconciler) ensureDownstreamGateway(
	ctx context.Context,
	upstreamClusterName string,
	upstreamClient client.Client,
	upstreamGateway *gatewayv1.Gateway,
	downstreamStrategy downstreamclient.ResourceStrategy,
) (result Result, downstreamGateway *gatewayv1.Gateway) {

	// Get the upstream gateway class so that we can pull the controller name out
	// of it and use it in route status updates.
	var upstreamGatewayClass gatewayv1.GatewayClass
	if err := upstreamClient.Get(ctx, types.NamespacedName{Name: string(upstreamGateway.Spec.GatewayClassName)}, &upstreamGatewayClass); err != nil {
		result.Err = err
		return result, nil
	}
	upstreamGatewayClassControllerName := string(upstreamGatewayClass.Spec.ControllerName)

	var targetDomainHostnames []string
	// Keep existing addresses in the status if present.
	if len(upstreamGateway.Status.Addresses) > 0 {
		for _, addr := range upstreamGateway.Status.Addresses {
			if ptr.Deref(addr.Type, "") == gatewayv1.HostnameAddressType {
				targetDomainHostnames = append(targetDomainHostnames, addr.Value)
			}
		}
	} else {
		gatewayDNSAddress := r.gatewayCanonicalHostname(upstreamGateway)

		// targetDomainHostnames are default hostnames that are unique to each gateway, and
		// will have DNS records created for them. Any custom hostnames provided in
		// listeners WILL NOT be added to the addresses list in the gateway status.
		targetDomainHostnames = append(targetDomainHostnames, gatewayDNSAddress)

		if r.Config.Gateway.IPv4Enabled() {
			targetDomainHostnames = append(targetDomainHostnames, fmt.Sprintf("v4.%s", gatewayDNSAddress))
		}

		if r.Config.Gateway.IPv6Enabled() {
			targetDomainHostnames = append(targetDomainHostnames, fmt.Sprintf("v6.%s", gatewayDNSAddress))
		}
	}

	downstreamClient := downstreamStrategy.GetClient()
	downstreamGatewayObjectMeta, err := downstreamStrategy.ObjectMetaFromUpstreamObject(ctx, upstreamGateway)
	if err != nil {
		result.Err = fmt.Errorf("failed to get downstream gateway object metadata: %w", err)
		return result, nil
	}

	downstreamGateway = &gatewayv1.Gateway{
		ObjectMeta: downstreamGatewayObjectMeta,
	}

	if err := downstreamClient.Get(ctx, client.ObjectKeyFromObject(downstreamGateway), downstreamGateway); client.IgnoreNotFound(err) != nil {
		result.Err = fmt.Errorf("failed to get downstream gateway: %w", err)
		return result, nil
	}

	verifiedHostnames, claimedHostnames, notClaimedHostnames, err := r.ensureHostnamesClaimed(
		ctx,
		upstreamClusterName,
		upstreamClient,
		upstreamGateway,
		downstreamGateway,
	)
	if err != nil {
		result.Err = err
		return result, nil
	}

	// Evaluate certificate health once so the listener we propagate and the
	// status we report to the user can never disagree.
	listenerCertHealth := r.evaluateListenerCertHealth(
		ctx,
		downstreamClient,
		downstreamGateway.Namespace,
		upstreamGateway,
		claimedHostnames,
	)

	desiredDownstreamGateway := r.getDesiredDownstreamGateway(
		ctx,
		upstreamGateway,
		claimedHostnames,
		listenerCertHealth,
	)

	if downstreamGateway.CreationTimestamp.IsZero() {
		if err := downstreamStrategy.SetControllerReference(ctx, upstreamGateway, downstreamGateway); err != nil {
			result.Err = fmt.Errorf("failed to set controller reference on downstream gateway: %w", err)
			return result, nil
		}

		downstreamGateway.Annotations = desiredDownstreamGateway.Annotations
		downstreamGateway.Spec = desiredDownstreamGateway.Spec

		if err := downstreamClient.Create(ctx, downstreamGateway); err != nil {
			result.Err = fmt.Errorf("failed creating downstream gateway: %w", err)
			return result, nil
		}
	} else {
		if !equality.Semantic.DeepEqual(downstreamGateway.Annotations, desiredDownstreamGateway.Annotations) ||
			!equality.Semantic.DeepEqual(downstreamGateway.Spec, desiredDownstreamGateway.Spec) {
			downstreamGateway.Annotations = desiredDownstreamGateway.Annotations
			downstreamGateway.Spec = desiredDownstreamGateway.Spec
			if err := downstreamClient.Update(ctx, downstreamGateway); err != nil {
				result.Err = fmt.Errorf("failed updating downstream gateway: %w", err)
				return result, nil
			}
		}
	}

	certResult := r.ensureListenerCertificates(
		ctx,
		upstreamGateway,
		downstreamGateway,
		downstreamClient,
		downstreamStrategy,
		claimedHostnames,
	)
	if certResult.ShouldReturn() {
		return certResult.Merge(result), nil
	}

	dnsResult := r.ensureDownstreamGatewayDNSEndpoints(
		ctx,
		downstreamGateway,
		downstreamStrategy,
		targetDomainHostnames,
	)
	if dnsResult.Err != nil || dnsResult.StopProcessing {
		return dnsResult.Merge(result), nil
	}
	// Carry RequeueAfter from dnsResult (e.g. IPs not yet available) without
	// blocking downstream HTTPRoute creation or gateway status updates.
	result = result.Merge(dnsResult)

	hostnameStatuses, dnsProgramResult := r.ensureDNSRecordSets(
		ctx,
		upstreamClient,
		upstreamGateway,
		claimedHostnames,
	)
	if dnsProgramResult.ShouldReturn() {
		return dnsProgramResult.Merge(result), nil
	}
	result = result.Merge(dnsProgramResult)

	if len(hostnameStatuses) > 0 {
		dnsStatusResult := r.reconcileDNSStatus(upstreamClient, upstreamGateway, hostnameStatuses)
		result = result.Merge(dnsStatusResult)
	}

	gatewayStatusResult := r.reconcileGatewayStatus(
		ctx,
		upstreamClient,
		upstreamGateway,
		downstreamGateway,
	)
	if gatewayStatusResult.Err != nil || gatewayStatusResult.StopProcessing {
		return gatewayStatusResult.Merge(result), nil
	}
	// Carry RequeueAfter from gatewayStatusResult (e.g. downstream not yet programmed)
	// without blocking HTTPRoute creation.
	result = result.Merge(gatewayStatusResult)

	httpRouteResult := r.ensureDownstreamGatewayHTTPRoutes(
		ctx,
		upstreamClient,
		upstreamGateway,
		upstreamGatewayClassControllerName,
		downstreamGateway,
		downstreamStrategy,
		verifiedHostnames,
		notClaimedHostnames,
		listenerCertHealth,
	)

	// When a listener is only waiting on a certificate to be issued, check back
	// soon so it starts serving promptly once the certificate is ready.
	for _, status := range listenerCertHealth {
		if !status.healthy && status.pending {
			result.RequeueAfter = max(result.RequeueAfter, 1*time.Minute)
			break
		}
	}

	addresses := make([]gatewayv1.GatewayStatusAddress, 0, len(targetDomainHostnames))

	for _, hostname := range targetDomainHostnames {
		addresses = append(addresses, gatewayv1.GatewayStatusAddress{
			Type:  ptr.To(gatewayv1.HostnameAddressType),
			Value: hostname,
		})
	}

	if !equality.Semantic.DeepEqual(upstreamGateway.Status.Addresses, addresses) {
		upstreamGateway.Status.Addresses = addresses
		result.AddStatusUpdate(upstreamClient, upstreamGateway)
	}

	return httpRouteResult.Merge(result), downstreamGateway
}

// listenerCertExpiryMargin keeps a certificate that is mid-renewal from briefly
// looking expired and dropping the listener.
const listenerCertExpiryMargin = 1 * time.Minute

// listenerCertStatus describes whether a listener's TLS certificate is usable.
// A listener with no entry is never gated: its certificate is managed elsewhere
// (for example the shared platform certificate) and a customer's certificate
// failure must never disable it.
type listenerCertStatus struct {
	healthy bool
	// reason is the listener status reason shown when the certificate is unusable.
	reason gatewayv1.ListenerConditionReason
	// message is a plain-language explanation shown to the customer.
	message string
	// pending means the listener is only waiting on a certificate to be issued,
	// so the reconcile should check back to let it recover.
	pending bool
	// notAfter is the certificate's expiry time, populated only when the
	// certificate is healthy and its expiry is known. Used to set the expiry
	// gauge so operators can alert before a cert expires.
	notAfter *metav1.Time
	// secretName is the downstream Secret that holds the certificate material.
	// Carried here so the expiry gauge can be labelled with the secret name
	// without recomputing it outside listenerCertHealth.
	secretName string
}

// clearListenerCertMetrics removes every certificate-health gauge series for a
// gateway. Used both before re-recording each reconcile and on gateway deletion
// so a removed listener never leaves a series stuck at its last value. The gating
// counter is left alone because it is cumulative.
func clearListenerCertMetrics(namespace, name string) {
	labels := prometheus.Labels{jsonKeyNamespace: namespace, jsonKeyName: name}
	gatewayListenerCertWithheld.DeletePartialMatch(labels)
	gatewayListenerCertExpiryTime.DeletePartialMatch(labels)
	gatewayListenerCertManaged.DeletePartialMatch(labels)
}

// evaluateListenerCertHealth reports which listeners have a usable certificate.
// It only considers listeners that own a per-hostname certificate; the same
// selection the rest of the controller uses keeps what we propagate and what we
// report to the customer consistent. A listener with no certificate of its own
// is left out so it is never gated.
func (r *GatewayReconciler) evaluateListenerCertHealth(
	ctx context.Context,
	downstreamClient client.Client,
	downstreamNamespace string,
	upstreamGateway *gatewayv1.Gateway,
	claimedHostnames []string,
) map[gatewayv1.SectionName]listenerCertStatus {
	logger := log.FromContext(ctx)
	health := make(map[gatewayv1.SectionName]listenerCertStatus)

	wildcardSuffix := "." + r.Config.Gateway.TargetDomain
	hasSharedSecret := r.Config.Gateway.HasDefaultListenerTLSSecret()
	now := time.Now()

	// Drop this gateway's previous certificate metrics up front and re-record the
	// current state below, so series for listeners that recovered, changed
	// hostname, or were removed don't linger at a stale value.
	clearListenerCertMetrics(upstreamGateway.Namespace, upstreamGateway.Name)

	for _, l := range upstreamGateway.Spec.Listeners {
		// Only listeners that own a per-hostname certificate are gated.
		if l.TLS == nil || l.TLS.Options[certificateIssuerTLSOption] == "" || l.Hostname == nil {
			continue
		}
		hostname := string(*l.Hostname)
		if !slices.Contains(claimedHostnames, hostname) {
			continue
		}
		// The shared platform certificate is managed by us, not the customer,
		// so never gate listeners that use it.
		if hasSharedSecret && (strings.HasSuffix(hostname, wildcardSuffix) || hostname == r.Config.Gateway.TargetDomain) {
			continue
		}

		status := r.listenerCertHealth(ctx, downstreamClient, downstreamNamespace, upstreamGateway.Name, l.Name, hostname, now)
		health[l.Name] = status

		// Mark this listener as managed regardless of its health, so the
		// SLI ratio (withheld / managed) can be computed fleet-wide.
		gatewayListenerCertManaged.WithLabelValues(
			upstreamGateway.Namespace, upstreamGateway.Name, string(l.Name), hostname,
		).Set(1)

		if !status.healthy {
			logArgs := []any{
				"listener", l.Name, "hostname", hostname,
				"reason", status.reason, "message", status.message,
			}
			// Include the expiry timestamp when available so log queries can
			// identify exactly when the cert stopped being valid.
			if status.notAfter != nil {
				logArgs = append(logArgs, "cert_not_after", status.notAfter.UTC().Format(time.RFC3339))
			}
			logger.Info("listener certificate unhealthy", logArgs...)

			gatewayListenerCertWithheld.WithLabelValues(
				upstreamGateway.Namespace, upstreamGateway.Name,
				string(l.Name), hostname, string(status.reason),
			).Set(1)
			gatewayListenerCertGatingTotal.WithLabelValues(
				upstreamGateway.Namespace, upstreamGateway.Name,
				string(l.Name), hostname, string(status.reason),
			).Inc()
		} else if status.notAfter != nil {
			// Record the expiry timestamp for healthy certs so operators can
			// alert before the next expiry rather than after.
			gatewayListenerCertExpiryTime.WithLabelValues(
				upstreamGateway.Namespace, upstreamGateway.Name,
				string(l.Name), hostname, status.secretName,
			).Set(float64(status.notAfter.Unix()))
		}
	}

	return health
}

// listenerCertHealth reports whether a single listener's certificate can be
// used to serve HTTPS right now, covering every way it can be unusable: still
// being issued, not yet valid, expired, or missing.
func (r *GatewayReconciler) listenerCertHealth(
	ctx context.Context,
	downstreamClient client.Client,
	downstreamNamespace string,
	gatewayName string,
	listenerName gatewayv1.SectionName,
	hostname string,
	now time.Time,
) listenerCertStatus {
	logger := log.FromContext(ctx)

	certName := listenerCertificateName(gatewayName, listenerName)
	secretName := listenerCertificateSecretName(gatewayName, listenerName)

	// The certificate must have been issued.
	var cert cmv1.Certificate
	if err := downstreamClient.Get(ctx, client.ObjectKey{Namespace: downstreamNamespace, Name: certName}, &cert); err != nil {
		if apierrors.IsNotFound(err) {
			return listenerCertStatus{
				reason:     gatewayv1.ListenerReasonInvalidCertificateRef,
				message:    certIssuanceFailingMessage(hostname),
				pending:    true,
				secretName: secretName,
			}
		}
		// On a read error, hold the listener back but allow it to recover later.
		logger.Error(err, "failed to get listener Certificate", "certificate", certName)
		return listenerCertStatus{
			reason:     gatewayv1.ListenerReasonInvalidCertificateRef,
			message:    certIssuanceFailingMessage(hostname),
			pending:    true,
			secretName: secretName,
		}
	}

	if !certIsReady(&cert) {
		return listenerCertStatus{
			reason:     gatewayv1.ListenerReasonInvalidCertificateRef,
			message:    certIssuanceFailingMessage(hostname),
			pending:    true,
			secretName: secretName,
		}
	}

	// The certificate must be within its valid dates.
	if cert.Status.NotBefore != nil && cert.Status.NotBefore.After(now) {
		return listenerCertStatus{
			reason:     gatewayv1.ListenerReasonInvalidCertificateRef,
			message:    certNotYetValidMessage(hostname),
			secretName: secretName,
		}
	}
	if cert.Status.NotAfter != nil && !cert.Status.NotAfter.After(now.Add(listenerCertExpiryMargin)) {
		return listenerCertStatus{
			reason:     gatewayv1.ListenerReasonInvalidCertificateRef,
			message:    certExpiredMessage(hostname),
			notAfter:   cert.Status.NotAfter,
			secretName: secretName,
		}
	}

	// Finally, load the stored certificate and key and confirm they match and
	// are still valid. This catches a broken or mismatched certificate that
	// would otherwise be served and break HTTPS for the listener.
	var secret corev1.Secret
	if err := downstreamClient.Get(ctx, client.ObjectKey{Namespace: downstreamNamespace, Name: secretName}, &secret); err != nil {
		if apierrors.IsNotFound(err) {
			return listenerCertStatus{
				reason:     gatewayv1.ListenerReasonInvalidCertificateRef,
				message:    certMissingMessage(hostname),
				pending:    true,
				secretName: secretName,
			}
		}
		logger.Error(err, "failed to get listener Secret", "secret", secretName)
		return listenerCertStatus{
			reason:     gatewayv1.ListenerReasonInvalidCertificateRef,
			message:    certMissingMessage(hostname),
			pending:    true,
			secretName: secretName,
		}
	}

	keyPair, err := tls.X509KeyPair(secret.Data["tls.crt"], secret.Data["tls.key"])
	if err != nil {
		return listenerCertStatus{
			reason:     gatewayv1.ListenerReasonInvalidCertificateRef,
			message:    certMissingMessage(hostname),
			secretName: secretName,
		}
	}
	if leaf := keyPair.Leaf; leaf != nil && !leaf.NotAfter.After(now) {
		return listenerCertStatus{
			reason:     gatewayv1.ListenerReasonInvalidCertificateRef,
			message:    certExpiredMessage(hostname),
			secretName: secretName,
		}
	}

	return listenerCertStatus{healthy: true, notAfter: cert.Status.NotAfter, secretName: secretName}
}

// certIsReady reports whether a cert-manager Certificate has Ready=True.
func certIsReady(cert *cmv1.Certificate) bool {
	for _, c := range cert.Status.Conditions {
		if c.Type == cmv1.CertificateConditionReady {
			return c.Status == cmmeta.ConditionTrue
		}
	}
	return false
}

// These messages are shown to customers, so they stay plain and name the
// hostname affected.

func certExpiredMessage(hostname string) string {
	return fmt.Sprintf("The TLS certificate for %s has expired, so HTTPS for this hostname is paused until a valid certificate is issued. This usually means the domain no longer points to Datum — please check its DNS.", hostname)
}

func certNotYetValidMessage(hostname string) string {
	return fmt.Sprintf("The TLS certificate for %s isn't valid yet. HTTPS will start automatically once its start date is reached.", hostname)
}

func certIssuanceFailingMessage(hostname string) string {
	return fmt.Sprintf("We couldn't issue a TLS certificate for %s, so HTTPS for this hostname is unavailable. Please confirm the domain points to Datum.", hostname)
}

func certMissingMessage(hostname string) string {
	return fmt.Sprintf("The TLS certificate for %s is missing or unreadable, so HTTPS for this hostname is unavailable.", hostname)
}

func (r *GatewayReconciler) getDesiredDownstreamGateway(
	ctx context.Context,
	upstreamGateway *gatewayv1.Gateway,
	claimedHostnames []string,
	listenerCertHealth map[gatewayv1.SectionName]listenerCertStatus,
) *gatewayv1.Gateway {
	logger := log.FromContext(ctx)
	var downstreamGateway gatewayv1.Gateway

	var listeners []gatewayv1.Listener

	wildcardSuffix := "." + r.Config.Gateway.TargetDomain

	for listenerIndex, l := range upstreamGateway.Spec.Listeners {
		if l.Hostname != nil && !slices.Contains(claimedHostnames, string(*l.Hostname)) {
			logger.Info("skipping downstream gateway listener with unclaimed hostname", "upstream_listener_index", listenerIndex, "hostname", *l.Hostname)
			continue
		}

		// Leave out a listener whose certificate isn't usable. A bad certificate
		// then only affects its own hostname, and every other hostname on the
		// gateway keeps serving.
		if status, gated := listenerCertHealth[l.Name]; gated {
			if !status.healthy {
				logger.Info("skipping downstream gateway listener with unhealthy certificate",
					"upstream_listener_index", listenerIndex, "listener", l.Name,
					"reason", status.reason, "message", status.message)
				continue
			}
			// The listener was evaluated (it owns a per-hostname cert) and is
			// healthy: it is admitted to the downstream gateway. Log this so
			// operators can confirm recovery without correlating metric state.
			logger.Info("admitting downstream gateway listener with healthy certificate",
				"upstream_listener_index", listenerIndex, "listener", l.Name)
		}

		// Per-listener TLS decision: hostnames covered by the wildcard
		// (*.targetDomain) reference the pre-provisioned shared secret;
		// all others reference a per-listener secret populated by a
		// Certificate resource created in ensureListenerCertificates.
		hostnameUnderWildcard := false
		if l.Hostname != nil {
			h := string(*l.Hostname)
			hostnameUnderWildcard = strings.HasSuffix(h, wildcardSuffix) || h == r.Config.Gateway.TargetDomain
		}
		useSharedTLS := hostnameUnderWildcard && r.Config.Gateway.HasDefaultListenerTLSSecret()

		if l.Hostname != nil {
			listenerCopy := l.DeepCopy()
			if l.TLS != nil && l.TLS.Options[certificateIssuerTLSOption] != "" {
				delete(listenerCopy.TLS.Options, certificateIssuerTLSOption)

				tlsMode := gatewayv1.TLSModeTerminate
				if useSharedTLS {
					listenerCopy.TLS = &gatewayv1.ListenerTLSConfig{
						Mode: &tlsMode,
						CertificateRefs: []gatewayv1.SecretObjectReference{
							{
								Group: ptr.To(gatewayv1.Group("")),
								Kind:  ptr.To(gatewayv1.Kind("Secret")),
								Name:  gatewayv1.ObjectName(r.Config.Gateway.DefaultListenerTLSSecretName),
							},
						},
					}
				} else {
					// Secret name must match the Certificate created by
					// ensureListenerCertificates for this listener.
					listenerCopy.TLS = &gatewayv1.ListenerTLSConfig{
						Mode: &tlsMode,
						CertificateRefs: []gatewayv1.SecretObjectReference{
							{
								Group: ptr.To(gatewayv1.Group("")),
								Kind:  ptr.To(gatewayv1.Kind("Secret")),
								Name:  gatewayv1.ObjectName(listenerCertificateSecretName(upstreamGateway.Name, l.Name)),
							},
						},
					}
				}
			}

			listeners = append(listeners, *listenerCopy)
		}
	}

	// TODO(jreese) get from "scheduler"
	downstreamGateway.Spec.GatewayClassName = gatewayv1.ObjectName(r.Config.Gateway.DownstreamGatewayClassName)

	downstreamGateway.Spec.Listeners = listeners

	return &downstreamGateway
}

// listenerCertificateSecretName returns the deterministic Secret name that a
// Certificate resource will populate for a given gateway listener.
func listenerCertificateSecretName(gatewayName string, listenerName gatewayv1.SectionName) string {
	return resourcename.GetValidDNS1123Name(fmt.Sprintf("%s-%s", gatewayName, listenerName))
}

// listenerCertificateName returns the deterministic Certificate resource name
// for a given gateway listener.
func listenerCertificateName(gatewayName string, listenerName gatewayv1.SectionName) string {
	return resourcename.GetValidDNS1123Name(fmt.Sprintf("%s-%s", gatewayName, listenerName))
}

// resolveAutoIssuer returns the first non-`auto` certificate-issuer value
// found on any TLS listener of the gateway, mapped through ClusterIssuerMap.
// Returns "" when no listener carries a real issuer.
//
// This restores the gateway-shim era semantic where the operator set a single
// cert-manager.io/cluster-issuer annotation on the downstream Gateway from
// the first TLS listener with an issuer (see the `if !HasAnnotation` guard
// in pre-`feat: gateway controller certificates` code) and gateway-shim
// minted Certificates for every TLS listener — including the auto-injected
// default-https — using that one annotation.
func (r *GatewayReconciler) resolveAutoIssuer(gateway *gatewayv1.Gateway) string {
	for _, l := range gateway.Spec.Listeners {
		if l.TLS == nil {
			continue
		}
		issuer := string(l.TLS.Options[certificateIssuerTLSOption])
		if issuer == "" || issuer == autoIssuerSentinel {
			continue
		}
		if mapped := r.Config.Gateway.ClusterIssuerMap[issuer]; mapped != "" {
			return mapped
		}
		return issuer
	}
	return ""
}

// ensureListenerCertificates creates, updates, or deletes cert-manager
// Certificate resources for each listener that requires an individual TLS
// certificate. Listeners whose hostnames fall under the wildcard target domain
// use the shared TLS secret and do not get a Certificate.
func (r *GatewayReconciler) ensureListenerCertificates(
	ctx context.Context,
	upstreamGateway *gatewayv1.Gateway,
	downstreamGateway *gatewayv1.Gateway,
	downstreamClient client.Client,
	downstreamStrategy downstreamclient.ResourceStrategy,
	claimedHostnames []string,
) (result Result) {
	logger := log.FromContext(ctx)
	wildcardSuffix := "." + r.Config.Gateway.TargetDomain

	desiredCerts := make(map[string]bool)
	gatewayNeedsUpdate := false

	// Wildcard-covered listeners are covered by the shared TLS secret only when
	// one is configured (DefaultListenerTLSSecretName). Otherwise they need a
	// per-listener Certificate just like any other listener — the downstream
	// listener already references a per-listener secret name via
	// ensureDownstreamGateway, and nothing else creates that secret.
	hasSharedSecret := r.Config.Gateway.HasDefaultListenerTLSSecret()

	// The defaulting webhook stamps the operator-injected default-https
	// listener with `certificate-issuer: auto` (the listenerTLSOptions
	// default). Pre-`feat: gateway controller certificates`, the operator set
	// a single cert-manager.io/cluster-issuer annotation on the downstream
	// Gateway from the first listener with a real issuer, and gateway-shim
	// minted Certificates for every TLS listener from that one annotation —
	// so `auto` was implicitly resolved to whatever the user's other listener
	// specified. With per-listener Certificate creation, that implicit
	// resolution disappeared. Restore it: when a listener's issuer is `auto`,
	// fall back to the first real issuer found on any TLS listener of this
	// gateway, mapped through ClusterIssuerMap.
	autoResolved := r.resolveAutoIssuer(upstreamGateway)

	for _, l := range upstreamGateway.Spec.Listeners {
		if l.TLS == nil || l.TLS.Options[certificateIssuerTLSOption] == "" || l.Hostname == nil {
			continue
		}
		hostname := string(*l.Hostname)
		if !slices.Contains(claimedHostnames, hostname) {
			continue
		}
		if hasSharedSecret && (strings.HasSuffix(hostname, wildcardSuffix) || hostname == r.Config.Gateway.TargetDomain) {
			continue
		}

		// Apply ClusterIssuerMap translation first — this is the admin-level
		// override (e.g. mapping the `auto` sentinel to a real ClusterIssuer
		// name). Only fall back to the gateway-shim era inter-listener
		// resolution when the sentinel survives mapping unchanged, i.e. when
		// the admin hasn't provided a translation.
		clusterIssuerName := string(l.TLS.Options[certificateIssuerTLSOption])
		if mapped := r.Config.Gateway.ClusterIssuerMap[clusterIssuerName]; mapped != "" {
			clusterIssuerName = mapped
		}
		if clusterIssuerName == autoIssuerSentinel {
			if autoResolved == "" {
				// No real issuer on this gateway to inherit from — match the
				// pre-migration behavior of leaving the listener un-programmed
				// rather than creating a Certificate with an unresolvable
				// IssuerRef.
				continue
			}
			clusterIssuerName = autoResolved
		}

		certName := listenerCertificateName(upstreamGateway.Name, l.Name)
		secretName := listenerCertificateSecretName(upstreamGateway.Name, l.Name)
		desiredCerts[certName] = true

		cert := &cmv1.Certificate{
			ObjectMeta: metav1.ObjectMeta{
				Name:      certName,
				Namespace: downstreamGateway.Namespace,
			},
		}

		if err := downstreamClient.Get(ctx, client.ObjectKeyFromObject(cert), cert); client.IgnoreNotFound(err) != nil {
			result.Err = fmt.Errorf("failed to get Certificate %s: %w", certName, err)
			return result
		}

		isNew := cert.CreationTimestamp.IsZero()
		if isNew {
			if err := downstreamStrategy.SetControllerReference(ctx, upstreamGateway, cert); err != nil {
				result.Err = fmt.Errorf("failed to set strategy reference on Certificate %s: %w", certName, err)
				return result
			}
		}

		// Ensure the downstream Gateway is the controller owner. The
		// downstream certificate solver controller walks the ownership
		// chain (Challenge → Order → Certificate → Gateway) to locate
		// the Gateway when creating solver HTTPRoutes for HTTP-01
		// challenges.
		ownerRefChanged := !metav1.IsControlledBy(cert, downstreamGateway)
		if ownerRefChanged {
			if err := controllerutil.SetControllerReference(downstreamGateway, cert, downstreamClient.Scheme()); err != nil {
				result.Err = fmt.Errorf("failed to set controller reference on Certificate %s: %w", certName, err)
				return result
			}
		}

		desiredSpec := cmv1.CertificateSpec{
			SecretName: secretName,
			SecretTemplate: &cmv1.CertificateSecretTemplate{
				Labels: map[string]string{
					downstreamclient.UpstreamOwnerClusterNameLabel: cert.Labels[downstreamclient.UpstreamOwnerClusterNameLabel],
				},
			},
			DNSNames: []string{hostname},
			IssuerRef: cmmeta.ObjectReference{
				Name: clusterIssuerName,
				Kind: KindClusterIssuer,
			},
		}

		var opResult string
		var err error
		if isNew {
			reissuanceCount := getReissuanceCount(downstreamGateway, certName)
			if reissuanceCount > 0 {
				if cert.Annotations == nil {
					cert.Annotations = make(map[string]string)
				}
				cert.Annotations[annotationReissuanceCount] = fmt.Sprintf("%d", reissuanceCount)
			}
			cert.Spec = desiredSpec
			err = downstreamClient.Create(ctx, cert)
			opResult = "created"
		} else if !equality.Semantic.DeepEqual(cert.Spec, desiredSpec) || ownerRefChanged {
			cert.Spec = desiredSpec
			err = downstreamClient.Update(ctx, cert)
			opResult = "updated"
		}
		if err != nil {
			result.Err = fmt.Errorf("failed to ensure Certificate %s: %w", certName, err)
			return result
		}
		if opResult != "" {
			logger.Info("Certificate reconciled", "certificate", certName, "operation", opResult)
		}

		// Check if an existing Certificate is stuck in a failed state and
		// should be deleted so we can recreate it fresh on the next reconcile.
		if !isNew {
			requeueAfter, gwChanged := r.reissueFailedCertificate(ctx, cert, certName, downstreamGateway, downstreamClient)
			if gwChanged {
				gatewayNeedsUpdate = true
			}
			if requeueAfter > 0 && (result.RequeueAfter == 0 || requeueAfter < result.RequeueAfter) {
				result.RequeueAfter = requeueAfter
			}
		}
	}

	// Persist reissuance tracking annotations on the downstream Gateway if
	// any Certificates were deleted for re-issuance.
	if gatewayNeedsUpdate {
		if err := downstreamClient.Update(ctx, downstreamGateway); err != nil {
			result.Err = fmt.Errorf("failed to update downstream gateway reissuance annotations: %w", err)
			return result
		}
	}

	// Clean up Certificate resources for listeners that no longer need them.
	// Two ownership patterns are checked:
	//  1. Certificates created by NSO via the downstream strategy (upstream-owner labels)
	//  2. Legacy Certificates created by cert-manager's gateway-shim (controller ownerRef
	//     to the downstream gateway, no upstream-owner labels)
	var certList cmv1.CertificateList
	if err := downstreamClient.List(ctx, &certList,
		client.InNamespace(downstreamGateway.Namespace),
	); err != nil {
		result.Err = fmt.Errorf("failed to list Certificates: %w", err)
		return result
	}

	for i := range certList.Items {
		cert := &certList.Items[i]
		if desiredCerts[cert.Name] {
			continue
		}

		ownedByStrategy := cert.Labels[downstreamclient.UpstreamOwnerKindLabel] == KindGateway &&
			cert.Labels[downstreamclient.UpstreamOwnerNameLabel] == upstreamGateway.Name &&
			cert.Labels[downstreamclient.UpstreamOwnerNamespaceLabel] == upstreamGateway.Namespace
		ownedByGatewayShim := metav1.IsControlledBy(cert, downstreamGateway)

		if !ownedByStrategy && !ownedByGatewayShim {
			continue
		}

		logger.Info("deleting stale Certificate", "certificate", cert.Name)
		if err := downstreamClient.Delete(ctx, cert); client.IgnoreNotFound(err) != nil {
			result.Err = fmt.Errorf("failed to delete stale Certificate %s: %w", cert.Name, err)
			return result
		}
	}

	return result
}

// reissueFailedCertificate checks whether a Certificate is stuck in a failed
// state and should be deleted so the gateway controller recreates it fresh on
// the next reconcile (bypassing cert-manager's exponential backoff).
//
// The reissuance count is tracked as an annotation on the downstream Gateway
// (keyed by certificate name) so it survives Certificate deletion.
//
// Returns:
//   - requeueAfter: non-zero if the caller should requeue after this duration
//   - gatewayChanged: true if the downstream gateway annotations were modified
//     and the caller must persist the change
func (r *GatewayReconciler) reissueFailedCertificate(
	ctx context.Context,
	cert *cmv1.Certificate,
	certName string,
	downstreamGateway *gatewayv1.Gateway,
	downstreamClient client.Client,
) (requeueAfter time.Duration, gatewayChanged bool) {
	logger := log.FromContext(ctx)
	reissuanceCfg := &r.Config.Gateway.CertificateReissuance

	if cert.Status.LastFailureTime == nil {
		if clearReissuanceCount(downstreamGateway, certName) {
			logger.V(1).Info("cleared reissuance count for healthy Certificate", "certificate", certName)
			gatewayChanged = true
		}
		return 0, gatewayChanged
	}

	retryCount := getReissuanceCount(downstreamGateway, certName)
	maxRetries := reissuanceCfg.GetMaxRetries()
	if retryCount >= maxRetries {
		logger.V(1).Info("Certificate has exhausted fast-track re-issuance attempts, deferring to cert-manager backoff",
			"certificate", certName,
			"reissuanceCount", retryCount,
			"maxRetries", maxRetries,
		)
		return 0, false
	}

	retryInterval := reissuanceCfg.GetRetryInterval()
	sinceFailure := time.Since(cert.Status.LastFailureTime.Time)
	if sinceFailure < retryInterval {
		remaining := retryInterval - sinceFailure
		logger.V(1).Info("Certificate failed but retry interval has not elapsed",
			"certificate", certName,
			"sinceFailure", sinceFailure,
			"retryInterval", retryInterval,
			"requeueAfter", remaining,
		)
		return remaining, false
	}

	setReissuanceCount(downstreamGateway, certName, retryCount+1)

	logger.Info("deleting failed Certificate for re-issuance",
		"certificate", certName,
		"lastFailureTime", cert.Status.LastFailureTime.Time,
		"reissuanceCount", retryCount+1,
		"maxRetries", maxRetries,
	)

	if err := downstreamClient.Delete(ctx, cert); err != nil {
		if !apierrors.IsNotFound(err) {
			logger.Error(err, "failed to delete Certificate for re-issuance", "certificate", certName)
		}
		return 0, true
	}

	return 0, true
}

// reissuanceAnnotationKey returns the gateway annotation key used to track
// reissuance count for a given certificate name.
func reissuanceAnnotationKey(certName string) string {
	return annotationReissuanceCount + "/" + certName
}

func getReissuanceCount(gw *gatewayv1.Gateway, certName string) int {
	if gw.Annotations == nil {
		return 0
	}
	count, err := strconv.Atoi(gw.Annotations[reissuanceAnnotationKey(certName)])
	if err != nil {
		return 0
	}
	return count
}

func setReissuanceCount(gw *gatewayv1.Gateway, certName string, count int) {
	if gw.Annotations == nil {
		gw.Annotations = make(map[string]string)
	}
	gw.Annotations[reissuanceAnnotationKey(certName)] = fmt.Sprintf("%d", count)
}

// clearReissuanceCount removes the reissuance tracking annotation for a
// certificate that has recovered. Returns true if the annotation was present.
func clearReissuanceCount(gw *gatewayv1.Gateway, certName string) bool {
	key := reissuanceAnnotationKey(certName)
	if _, ok := gw.Annotations[key]; ok {
		delete(gw.Annotations, key)
		return true
	}
	return false
}

func (r *GatewayReconciler) reconcileGatewayStatus(
	ctx context.Context,
	upstreamClient client.Client,
	upstreamGateway *gatewayv1.Gateway,
	downstreamGateway *gatewayv1.Gateway,
) (result Result) {
	logger := log.FromContext(ctx)

	acceptedReady := false
	programmedReady := false

	if c := apimeta.FindStatusCondition(downstreamGateway.Status.Conditions, string(gatewayv1.GatewayConditionAccepted)); c != nil {
		message := "The Gateway has not been scheduled by Datum Gateway"
		if c.Status == metav1.ConditionTrue {
			message = "The Gateway has been scheduled by Datum Gateway"
			acceptedReady = true
		}

		apimeta.SetStatusCondition(&upstreamGateway.Status.Conditions, metav1.Condition{
			Message:            message,
			Type:               string(gatewayv1.GatewayConditionAccepted),
			Reason:             c.Reason,
			Status:             c.Status,
			ObservedGeneration: upstreamGateway.Generation,
		})

		result.AddStatusUpdate(upstreamClient, upstreamGateway)
	}

	if c := apimeta.FindStatusCondition(downstreamGateway.Status.Conditions, string(gatewayv1.GatewayConditionProgrammed)); c != nil {
		message := "The Gateway has not been programmed"
		if c.Status == metav1.ConditionTrue {
			message = "The Gateway has been programmed"
			programmedReady = true
		}

		apimeta.SetStatusCondition(&upstreamGateway.Status.Conditions, metav1.Condition{
			Message:            message,
			Type:               string(gatewayv1.GatewayConditionProgrammed),
			Reason:             c.Reason,
			Status:             c.Status,
			ObservedGeneration: upstreamGateway.Generation,
		})

		result.AddStatusUpdate(upstreamClient, upstreamGateway)
	}

	// Track per-gateway programmed state. Set 1 when programmed, 0 when not.
	// Dashboard: sum(nso_gateway_programmed_total) gives fleet-wide count;
	// a value below total gateway count indicates partial fleet failures.
	programmedValue := 0.0
	if programmedReady {
		programmedValue = 1.0
	}
	gatewayProgrammedTotal.WithLabelValues(upstreamGateway.Namespace, upstreamGateway.Name).Set(programmedValue)

	// If the downstream gateway hasn't been scheduled and programmed yet,
	// requeue after a short delay. This handles cache-staleness races where
	// the downstream watch fires before the cache reflects EG's status update
	// (same pattern as the IP-address requeue in ensureDownstreamGatewayDNSEndpoints).
	if !acceptedReady || !programmedReady {
		logger.Info("downstream gateway not yet accepted/programmed, requeueing",
			"accepted", acceptedReady, "programmed", programmedReady,
			"downstream_conditions", len(downstreamGateway.Status.Conditions))
		result.RequeueAfter = 5 * time.Second
	}

	return result
}

func (r *GatewayReconciler) ensureHostnamesClaimed(
	ctx context.Context,
	upstreamClusterName string,
	upstreamClient client.Client,
	upstreamGateway *gatewayv1.Gateway,
	downstreamGateway *gatewayv1.Gateway,
) (verifiedHostnames, claimedHostnames, notClaimedHostnames []string, err error) {

	verifiedHostnames, err = r.ensureHostnameVerification(ctx, upstreamClient, upstreamGateway, downstreamGateway)
	if err != nil {
		return nil, nil, nil, err
	}

	downstreamClient := r.DownstreamCluster.GetClient()

	upstreamGatewayReferenceName := fmt.Sprintf("%s/%s/%s", upstreamClusterName, upstreamGateway.Namespace, upstreamGateway.Name)

	// Track each hostname in a ConfigMap in the downstream control plane.
	// This will need to be adjusted as the number of hostnames grows to be large,
	// but is considered sufficient for now.

	for _, hostname := range verifiedHostnames {
		// No accounting for platform-managed gateway hostnames.
		if r.isDatumManagedGatewayHostname(upstreamGateway, hostname) {
			claimedHostnames = append(claimedHostnames, hostname)
			continue
		}

		objectKey := client.ObjectKey{
			Namespace: r.Config.Gateway.DownstreamHostnameAccountingNamespace,
			Name:      hostname,
		}

		var hostnameConfigMap corev1.ConfigMap
		if err := downstreamClient.Get(ctx, objectKey, &hostnameConfigMap); client.IgnoreNotFound(err) != nil {
			return nil, nil, nil, err
		}

		if hostnameConfigMap.CreationTimestamp.IsZero() {
			hostnameConfigMap = corev1.ConfigMap{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: objectKey.Namespace,
					Name:      objectKey.Name,
					Labels: map[string]string{
						downstreamclient.UpstreamOwnerClusterNameLabel: fmt.Sprintf("cluster-%s", strings.ReplaceAll(upstreamClusterName, "/", "_")),
						downstreamclient.UpstreamOwnerNamespaceLabel:   upstreamGateway.Namespace,
						downstreamclient.UpstreamOwnerNameLabel:        upstreamGateway.Name,
					},
				},
				Data: map[string]string{
					jsonKeyOwner: upstreamGatewayReferenceName,
				},
			}

			if err := downstreamClient.Create(ctx, &hostnameConfigMap); err != nil {
				if apierrors.IsConflict(err) {
					notClaimedHostnames = append(notClaimedHostnames, hostname)
					continue
				}
				return nil, nil, nil, err
			}
		} else if hostnameConfigMap.Data[jsonKeyOwner] != upstreamGatewayReferenceName {
			notClaimedHostnames = append(notClaimedHostnames, hostname)
			continue
		}

		claimedHostnames = append(claimedHostnames, hostname)
	}

	// Delete any claimed hostnames that are no longer in use
	// Delete all configmaps that are claiming hostnames
	downstreamClusterClient := r.DownstreamCluster.GetClient()
	listOpts := []client.ListOption{
		client.InNamespace(r.Config.Gateway.DownstreamHostnameAccountingNamespace),
		client.MatchingLabels{
			downstreamclient.UpstreamOwnerClusterNameLabel: fmt.Sprintf("cluster-%s", strings.ReplaceAll(upstreamClusterName, "/", "_")),
			downstreamclient.UpstreamOwnerNamespaceLabel:   upstreamGateway.Namespace,
			downstreamclient.UpstreamOwnerNameLabel:        upstreamGateway.Name,
		},
	}

	var hostnameConfigMapList corev1.ConfigMapList
	if err := downstreamClusterClient.List(ctx, &hostnameConfigMapList, listOpts...); err != nil {
		return nil, nil, nil, fmt.Errorf("failed to list hostname claim configmaps: %w", err)
	}

	if len(hostnameConfigMapList.Items) > 0 {
		for _, configMap := range hostnameConfigMapList.Items {
			if slices.Contains(claimedHostnames, configMap.Name) {
				// Still in use
				continue
			}
			if err := downstreamClusterClient.Delete(ctx, &configMap); err != nil {
				return nil, nil, nil, fmt.Errorf("failed to delete hostname claim configmap: %w", err)
			}
		}
	}

	slices.Sort(claimedHostnames)

	return verifiedHostnames, claimedHostnames, notClaimedHostnames, nil
}

func (r *GatewayReconciler) isDatumManagedGatewayHostname(upstreamGateway *gatewayv1.Gateway, hostname string) bool {
	targetDomain := r.Config.Gateway.TargetDomain
	gatewayUID := string(upstreamGateway.UID)
	legacyUIDWithoutDashes := strings.ReplaceAll(gatewayUID, "-", "")

	managedBaseHostnames := []string{
		r.gatewayCanonicalHostname(upstreamGateway),
		r.Config.Gateway.GatewayDNSAddress(upstreamGateway),
		fmt.Sprintf("%s.%s", legacyUIDWithoutDashes, targetDomain),
		fmt.Sprintf("%s.%s", gatewayUID, targetDomain),
	}

	for _, managedHostname := range managedBaseHostnames {
		if hostname == managedHostname ||
			hostname == fmt.Sprintf("v4.%s", managedHostname) ||
			hostname == fmt.Sprintf("v6.%s", managedHostname) {
			return true
		}
	}

	return false
}

// isHostnameVerified returns hostnames found on listeners that are verified. A
// hostname is considered verified if any verified Domain is found in the same
// namespace with a `spec.domainName` value that matches the hostname exactly,
// or the hostname is a sub domain. If no matching Domain is found, one will be
// created.
func (r *GatewayReconciler) ensureHostnameVerification(
	ctx context.Context,
	upstreamClient client.Client,
	upstreamGateway *gatewayv1.Gateway,
	downstreamGateway *gatewayv1.Gateway,
) ([]string, error) {
	logger := log.FromContext(ctx)

	gatewayDefaultHostname := r.gatewayCanonicalHostname(upstreamGateway)

	// Allow hostnames which have been successfully configured on the downstream
	// gateway stay on the gateway, regardless of whether or not there is a
	// matching Domain that's verified. This is done in case the user deletes the
	// Domain record after verification has completed. In the future, it would be
	// better to place a lien on the Domain via a ValidatingAdmissionPolicyBinding
	// or higher level abstraction. A finalizer is undesirable as many resources
	// could leverage the Domain, and we'd want to prohibit deletion, not just
	// keep a deleted item from going away.
	//
	// For more thoughts on liens, the following issue provides good context:
	// https://github.com/kubernetes/kubernetes/issues/10179#issuecomment-2889238042
	verifiedHostnames := sets.New[string]()
	addressHostnames := sets.New[string]()
	addressHostnames.Insert(gatewayDefaultHostname)
	if dt := downstreamGateway.DeletionTimestamp; dt.IsZero() {
		for _, listener := range downstreamGateway.Spec.Listeners {
			if listener.Hostname != nil && !strings.HasSuffix(string(*listener.Hostname), r.Config.Gateway.TargetDomain) {
				verifiedHostnames.Insert(string(*listener.Hostname))
			}
		}
	}

	// Consider any addresses programmed to be verified - this is necessary for
	// transitioning default proxy hostnames.

	for _, address := range upstreamGateway.Status.Addresses {
		if ptr.Deref(address.Type, "") == gatewayv1.HostnameAddressType {
			addressHostnames.Insert(address.Value)
		}
	}

	verifiedHostnames.Insert(addressHostnames.UnsortedList()...)

	logger.Info("collected verified hostnames from listener conditions", "hostnames", verifiedHostnames.UnsortedList())

	// Get a unique set of hostnames
	hostnames := sets.New[string]()
	for _, l := range upstreamGateway.Spec.Listeners {
		if l.Hostname != nil {
			hostnames.Insert(string(*l.Hostname))
		}
	}

	if r.Config.Gateway.DisableHostnameVerification {
		verifiedHostnamesSlice := hostnames.UnsortedList()
		slices.Sort(verifiedHostnamesSlice)
		return verifiedHostnamesSlice, nil
	}

	// List all Domains in the same namespace as the upstream gateway. A field
	// selector will not work here, as set based operations are not supported, and
	// there's not way to check for a suffix match.

	var domainList networkingv1alpha.DomainList
	if err := upstreamClient.List(ctx, &domainList, client.InNamespace(upstreamGateway.Namespace)); err != nil {
		return nil, fmt.Errorf("failed listing domains: %w", err)
	}

	logger.Info("processing domains in same namespace", "domain_count", len(domainList.Items))

	domainsToCreate := sets.New[string]()
	for _, hostname := range hostnames.UnsortedList() {
		// Gateway DNS address hostname is exempt from verification
		if addressHostnames.Has(hostname) {
			continue
		}
		foundMatchingDomain := false
		for _, domain := range domainList.Items {
			if hostname == domain.Spec.DomainName || strings.HasSuffix(hostname, "."+domain.Spec.DomainName) {
				foundMatchingDomain = true
				if !apimeta.IsStatusConditionTrue(domain.Status.Conditions, networkingv1alpha.DomainConditionVerified) {
					logger.Info("domain is not verified", "domain", domain.Name)
					continue
				}
				verifiedHostnames.Insert(hostname)
				break
			}
		}

		if !foundMatchingDomain {
			parts := strings.Split(hostname, ".")
			if len(parts) < 2 {
				logger.Info("malformed hostname, expected at least two parts", "hostname", hostname)
				continue
			}

			domainsToCreate.Insert(strings.Join(parts[len(parts)-2:], "."))
		}
	}

	if len(domainsToCreate) > 0 {
		logger.Info("creating domain resources for hostnames with no matching domain", "domains", domainsToCreate)

		// Create a Domain resource with the same name as the value that will be
		// placed in spec.domainName. This is done to avoid duplication upon cache
		// sync delays. AlreadyExists errors will be ignored.
		for _, domainName := range domainsToCreate.UnsortedList() {
			domain := &networkingv1alpha.Domain{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: upstreamGateway.Namespace,
					Name:      domainName,
				},
				Spec: networkingv1alpha.DomainSpec{
					DomainName: domainName,
				},
			}

			if err := upstreamClient.Create(ctx, domain); client.IgnoreAlreadyExists(err) != nil {
				return nil, fmt.Errorf("failed creating domain: %w", err)
			}

			logger.Info("domain created", "domain", domain.Name)
		}
	}

	verifiedHostnamesSlice := verifiedHostnames.UnsortedList()
	slices.Sort(verifiedHostnamesSlice)

	return verifiedHostnamesSlice, nil
}

// gatewayCanonicalHostname returns the managed canonical hostname for a gateway.
// It prefers an existing status address in the configured target domain to avoid
// renaming already-programmed gateways during hostname format transitions.
func (r *GatewayReconciler) gatewayCanonicalHostname(upstreamGateway *gatewayv1.Gateway) string {
	return gatewayCanonicalHostnameForConfig(r.Config.Gateway, upstreamGateway)
}

// gatewayCanonicalHostnameForConfig is the shared implementation of
// gatewayCanonicalHostname, usable by any reconciler that has a GatewayConfig.
func gatewayCanonicalHostnameForConfig(gatewayCfg config.GatewayConfig, gw *gatewayv1.Gateway) string {
	if existing := managedGatewayHostnameFromStatus(
		gw.Status.Addresses,
		gatewayCfg.TargetDomain,
	); existing != "" {
		return existing
	}
	return gatewayCfg.GatewayDNSAddress(gw)
}

// managedGatewayHostnameFromStatus returns the base hostname address (not v4/v6
// variants) in the target domain from gateway status, if present.
func managedGatewayHostnameFromStatus(
	addresses []gatewayv1.GatewayStatusAddress,
	targetDomain string,
) string {
	suffix := "." + targetDomain

	for _, addr := range addresses {
		if ptr.Deref(addr.Type, "") != gatewayv1.HostnameAddressType {
			continue
		}
		if addr.Value != targetDomain && !strings.HasSuffix(addr.Value, suffix) {
			continue
		}
		if strings.HasPrefix(addr.Value, "v4.") || strings.HasPrefix(addr.Value, "v6.") {
			continue
		}
		return addr.Value
	}

	return ""
}

func (r *GatewayReconciler) ensureDownstreamGatewayDNSEndpoints(
	ctx context.Context,
	downstreamGateway *gatewayv1.Gateway,
	downstreamStrategy downstreamclient.ResourceStrategy,
	hostnames []string,
) (result Result) {
	logger := log.FromContext(ctx)

	// Extract IP addresses from the downstream gateway's status
	// Using the `any`` type due to deep copy logic requirements in the unstructured
	// lib used to set DNSEndpoint values.
	var v4IPs, v6IPs []any
	for _, addr := range downstreamGateway.Status.Addresses {
		if addr.Type == nil {
			continue
		}
		switch *addr.Type {
		case gatewayv1.IPAddressType:
			// Check if it's an IPv4 or IPv6 address
			if strings.Contains(addr.Value, ":") {
				v6IPs = append(v6IPs, addr.Value)
			} else {
				v4IPs = append(v4IPs, addr.Value)
			}
		}
	}

	// Return early if no IP addresses were found. Requeue after a short delay
	// so we don't rely solely on the downstream Gateway watch to re-trigger
	// reconciliation (the watch may fire before the cache reflects the status
	// update, leaving us with stale data on this cycle).
	if (r.Config.Gateway.IPv4Enabled() && len(v4IPs) == 0) || (r.Config.Gateway.IPv6Enabled() && len(v6IPs) == 0) {
		logger.Info(
			"IP addresses not yet available on downstream gateway",
			"ipv4", v4IPs, "ipv4_enabled", r.Config.Gateway.IPv4Enabled(),
			"ipv6", v6IPs, "ipv6_enabled", r.Config.Gateway.IPv6Enabled(),
		)
		result.RequeueAfter = 5 * time.Second
		return result
	}

	endpoints := []any{}
	var gatewayDNSEndpoint unstructured.Unstructured
	gatewayDNSEndpoint.SetGroupVersionKind(schema.GroupVersionKind{
		Group:   "externaldns.k8s.io",
		Version: versionV1Alpha1,
		Kind:    "DNSEndpoint",
	})
	gatewayDNSEndpoint.SetNamespace(downstreamGateway.Namespace)
	gatewayDNSEndpoint.SetName(downstreamGateway.Name)

	for _, hostname := range hostnames {
		if len(v4IPs) > 0 && !strings.HasPrefix(hostname, "v6") {
			// v4 specific hostname, or hostname that includes both v4 and v6
			endpoints = append(endpoints, map[string]any{
				"dnsName":    hostname,
				"targets":    v4IPs,
				"recordType": "A",
				"recordTTL":  int64(300),
			})
		}

		if len(v6IPs) > 0 && !strings.HasPrefix(hostname, "v4") {
			// v6 specific hostname, or hostname that includes both v4 and v6
			endpoints = append(endpoints, map[string]any{
				"dnsName":    hostname,
				"targets":    v6IPs,
				"recordType": "AAAA",
				"recordTTL":  int64(300),
			})
		}
	}

	if _, err := controllerutil.CreateOrUpdate(ctx, downstreamStrategy.GetClient(), &gatewayDNSEndpoint, func() error {
		if err := controllerutil.SetControllerReference(downstreamGateway, &gatewayDNSEndpoint, downstreamStrategy.GetClient().Scheme()); err != nil {
			return err
		}
		return unstructured.SetNestedSlice(gatewayDNSEndpoint.Object, endpoints, "spec", "endpoints")
	}); err != nil {
		result.Err = err
		return result
	}

	return result
}

func (r *GatewayReconciler) finalizeGateway(
	ctx context.Context,
	upstreamClusterName string,
	upstreamClient client.Client,
	upstreamGateway *gatewayv1.Gateway,
	downstreamStrategy downstreamclient.ResourceStrategy,
) (result Result) {
	logger := log.FromContext(ctx)
	logger.Info("finalizing gateway")

	// Remove per-gateway metric series so deleted gateways do not leave stale
	// label sets that misrepresent fleet state.
	gatewayProgrammedTotal.DeleteLabelValues(upstreamGateway.Namespace, upstreamGateway.Name)
	// Clear this gateway's cert-health series now that it is gone.
	clearListenerCertMetrics(upstreamGateway.Namespace, upstreamGateway.Name)

	// Clean up DNS records created by this gateway
	if r.Config.Gateway.EnableDNSIntegration {
		if cleanupResult := r.cleanupDNSRecordSets(ctx, upstreamClient, upstreamGateway); cleanupResult.ShouldReturn() {
			return cleanupResult
		}
	}

	// Go through downstream http routes that are attached to the downstream
	// gateway and remove the parentRef from the status. If it's the last parent
	// ref, delete the downstream route. If there's a race condition on delete/create,
	// it'll be reconciled again in the next cycle and fighting is not expected.

	downstreamClient := downstreamStrategy.GetClient()

	downstreamGatewayObjectMeta, err := downstreamStrategy.ObjectMetaFromUpstreamObject(ctx, upstreamGateway)
	if err != nil {
		result.Err = err
		return result
	}
	downstreamGateway := &gatewayv1.Gateway{
		ObjectMeta: downstreamGatewayObjectMeta,
	}

	if err := downstreamClient.Get(ctx, client.ObjectKeyFromObject(downstreamGateway), downstreamGateway); err != nil {
		if apierrors.IsNotFound(err) {
			return result
		}
		result.Err = err
		return result
	}

	// Detach HTTPRoutes from the upstream gateway
	logger.Info("detaching httproutes from upstream gateway")
	detachResult := r.detachHTTPRoutes(ctx, upstreamClient, upstreamGateway, false)
	if detachResult.ShouldReturn() {
		return detachResult
	}

	// Detach HTTPRoutes from the downstream gateway
	logger.Info("detaching httproutes from downstream gateway")
	detachResult = r.detachHTTPRoutes(ctx, downstreamClient, downstreamGateway, true)
	if detachResult.ShouldReturn() {
		return detachResult
	}

	logger.Info("deleting anchor for upstream gateway")
	if err := downstreamStrategy.DeleteAnchorForObject(ctx, upstreamGateway); err != nil {
		result.Err = err
		return result
	}

	// Delete all configmaps that are claiming hostnames
	downstreamClusterClient := r.DownstreamCluster.GetClient()
	listOpts := []client.ListOption{
		client.InNamespace(r.Config.Gateway.DownstreamHostnameAccountingNamespace),
		client.MatchingLabels{
			downstreamclient.UpstreamOwnerClusterNameLabel: fmt.Sprintf("cluster-%s", strings.ReplaceAll(upstreamClusterName, "/", "_")),
			downstreamclient.UpstreamOwnerNamespaceLabel:   upstreamGateway.Namespace,
			downstreamclient.UpstreamOwnerNameLabel:        upstreamGateway.Name,
		},
	}

	var hostnameConfigMapList corev1.ConfigMapList
	if err := downstreamClusterClient.List(ctx, &hostnameConfigMapList, listOpts...); err != nil {
		result.Err = err
		return result
	}

	if len(hostnameConfigMapList.Items) > 0 {
		for _, configMap := range hostnameConfigMapList.Items {
			if err := downstreamClusterClient.Delete(ctx, &configMap); err != nil {
				result.Err = fmt.Errorf("failed to delete hostname claim configmap: %w", err)
				return result
			}
		}
	}

	return result
}

// cleanupDNSRecordSets deletes all DNSRecordSet resources that were created by
// this gateway. This is called during gateway finalization to ensure DNS records
// are cleaned up when the gateway is deleted, since owner reference-based garbage
// collection doesn't work (the dns-operator sets itself as the controller owner).
func (r *GatewayReconciler) cleanupDNSRecordSets(
	ctx context.Context,
	upstreamClient client.Client,
	upstreamGateway *gatewayv1.Gateway,
) (result Result) {
	logger := log.FromContext(ctx)

	var recordSetList dnsv1alpha1.DNSRecordSetList
	if err := upstreamClient.List(ctx, &recordSetList,
		client.InNamespace(upstreamGateway.Namespace),
		client.MatchingLabels{
			labelDNSManaged:    labelValueTrue,
			labelManagedBy:     labelManagedByValue,
			labelDNSSourceKind: KindGateway,
			labelDNSSourceName: upstreamGateway.Name,
			labelDNSSourceNS:   upstreamGateway.Namespace,
		},
	); err != nil {
		// If the CRD doesn't exist, there's nothing to clean up
		if apimeta.IsNoMatchError(err) {
			return result
		}
		result.Err = fmt.Errorf("failed listing DNSRecordSets for cleanup: %w", err)
		return result
	}

	for _, rs := range recordSetList.Items {
		logger.Info("deleting DNSRecordSet during gateway finalization",
			jsonKeyName, rs.Name,
			"hostname", rs.Annotations[annotationDNSHostname],
		)
		if err := upstreamClient.Delete(ctx, &rs); err != nil && !apierrors.IsNotFound(err) {
			result.Err = fmt.Errorf("failed to delete DNSRecordSet %q: %w", rs.Name, err)
			return result
		}
	}

	if len(recordSetList.Items) > 0 {
		logger.Info("deleted DNSRecordSets during gateway finalization", "count", len(recordSetList.Items))
	}

	return result
}

// TODO(jreese) revisit the parameters here to clean them up. It's a bit messy
// as this function is used against both the upstream and downstream resources.
func (r *GatewayReconciler) detachHTTPRoutes(
	ctx context.Context,
	gatewayClient client.Client,
	gateway *gatewayv1.Gateway,
	deleteWhenNoParents bool,
) (result Result) {
	logger := log.FromContext(ctx)

	var httpRoutes gatewayv1.HTTPRouteList
	if err := gatewayClient.List(ctx, &httpRoutes, client.InNamespace(gateway.Namespace)); err != nil {
		result.Err = err
		return result
	}

	logger.Info("found httproutes", "count", len(httpRoutes.Items))

	for _, route := range httpRoutes.Items {
		if !route.DeletionTimestamp.IsZero() {
			continue
		}

		var remainingRefs []gatewayv1.ParentReference
		for _, ref := range route.Spec.ParentRefs {
			if ptr.Deref(ref.Group, gatewayv1.GroupName) == gatewayv1.GroupName &&
				ptr.Deref(ref.Kind, KindGateway) == KindGateway &&
				string(ref.Name) == gateway.Name {
				logger.Info("removing parent ref from httproute", jsonKeyName, route.Name, "parent", ref.Name)
				continue
			}
			remainingRefs = append(remainingRefs, ref)
		}

		if len(remainingRefs) == len(route.Spec.ParentRefs) {
			continue
		}

		if len(remainingRefs) == 0 && deleteWhenNoParents {
			logger.Info("deleting httproute due to no parents", jsonKeyName, route.Name)
			if err := gatewayClient.Delete(ctx, &route); err != nil {
				result.Err = err
				return result
			}
		} else {
			route.Spec.ParentRefs = remainingRefs
			if err := gatewayClient.Update(ctx, &route); err != nil {
				result.Err = err
				return result
			}
		}
	}
	return result
}

func (r *GatewayReconciler) ensureDownstreamGatewayHTTPRoutes(
	ctx context.Context,
	upstreamClient client.Client,
	upstreamGateway *gatewayv1.Gateway,
	upstreamGatewayClassControllerName string,
	downstreamGateway *gatewayv1.Gateway,
	downstreamStrategy downstreamclient.ResourceStrategy,
	verifiedHostnames []string,
	notClaimedHostnames []string,
	listenerCertHealth map[gatewayv1.SectionName]listenerCertStatus,
) (result Result) {
	logger := log.FromContext(ctx)

	// Get HTTPRoutes in the same namespace as the upstream gateway
	var httpRoutes gatewayv1.HTTPRouteList
	if err := upstreamClient.List(ctx, &httpRoutes, client.InNamespace(upstreamGateway.Namespace)); err != nil {
		result.Err = err
		return result
	}

	upstreamNS := gatewayv1.Namespace(upstreamGateway.Namespace)

	// Collect routes attached to the gateway.
	attachedRoutes := map[client.ObjectKey]gatewayv1.HTTPRoute{}
	attachedRouteCount := make(map[gatewayv1.SectionName]int32, len(upstreamGateway.Spec.Listeners))
	for _, route := range httpRoutes.Items {
		if parentRefs := route.Spec.ParentRefs; parentRefs != nil {
			for _, parentRef := range parentRefs {
				if ptr.Deref(parentRef.Namespace, upstreamNS) != upstreamNS {
					// At the time of this writing, validation prohibits namespaces to be
					// set on parent refs. This check is here as a safety for when that
					// limitation is removed, as additional programming logic will need
					// to exist to translate downstream namespace names.
					result.Err = fmt.Errorf("unexpected namespace in parent ref: %s", *parentRef.Namespace)
					return result
				} else if parentRef.Namespace != nil {
					// Translate the namespace to the downstream namespace
					parentRef.Namespace = ptr.To(gatewayv1.Namespace(downstreamGateway.Namespace))
				}

				if ptr.Deref(parentRef.Group, gatewayv1.GroupName) == gatewayv1.GroupName &&
					ptr.Deref(parentRef.Kind, KindGateway) == KindGateway &&
					string(parentRef.Name) == upstreamGateway.Name {
					// If the parentRef has a section name, only attach the route if the
					// listener exists in the gateway.
					if parentRef.SectionName != nil {
						foundSectionName := false
						for _, listener := range upstreamGateway.Spec.Listeners {
							if listener.Name == *parentRef.SectionName {
								foundSectionName = true
								break
							}
						}
						if !foundSectionName {
							logger.Info("section name not found in gateway", "section_name", *parentRef.SectionName)
							continue
						}

						attachedRouteCount[*parentRef.SectionName]++
					} else {
						// Attached to all sections, update all counts
						for _, l := range upstreamGateway.Spec.Listeners {
							attachedRouteCount[l.Name]++
						}
					}

					attachedRoutes[client.ObjectKeyFromObject(&route)] = route
				}
			}
		}
	}

	logger.Info("attached routes", "count", len(attachedRoutes))

	for _, route := range attachedRoutes {
		if !route.DeletionTimestamp.IsZero() {
			logger.Info("skipping httproute due to deletion timestamp", jsonKeyName, route.Name)
			continue
		}

		if !controllerutil.ContainsFinalizer(&route, gatewayControllerGCFinalizer) {
			controllerutil.AddFinalizer(&route, gatewayControllerGCFinalizer)
			if err := upstreamClient.Update(ctx, &route); err != nil {
				result.Err = fmt.Errorf("failed to add finalizer to httproute: %w", err)
				return result
			}
		}

		httpRouteResult := r.ensureDownstreamHTTPRoute(
			ctx,
			upstreamClient,
			upstreamGateway,
			upstreamGatewayClassControllerName,
			downstreamGateway,
			downstreamStrategy,
			route,
		)
		if result.Err != nil {
			return result
		}
		result = result.Merge(httpRouteResult)
	}

	logger.Info("updating listener status", "verified_hostnames", verifiedHostnames, "not_claimed_hostnames", notClaimedHostnames)

	currentListenerStatus := map[gatewayv1.SectionName]gatewayv1.ListenerStatus{}
	for _, listener := range upstreamGateway.Status.Listeners {
		currentListenerStatus[listener.Name] = *listener.DeepCopy()
	}

	// Update listener status for the upstream gateway
	listenerStatus := make([]gatewayv1.ListenerStatus, 0, len(upstreamGateway.Spec.Listeners))
	for _, listener := range upstreamGateway.Spec.Listeners {
		status, ok := currentListenerStatus[listener.Name]
		if !ok {
			status = gatewayv1.ListenerStatus{
				Name: listener.Name,
				SupportedKinds: []gatewayv1.RouteGroupKind{
					{
						Group: ptr.To(gatewayv1.Group(gatewayv1.GroupName)),
						Kind:  KindHTTPRoute,
					},
				},
			}
		}

		status.AttachedRoutes = attachedRouteCount[listener.Name]

		acceptedCondition := metav1.Condition{
			Type:               string(gatewayv1.ListenerConditionAccepted),
			Status:             metav1.ConditionTrue,
			Reason:             string(gatewayv1.ListenerReasonAccepted),
			Message:            "The listener has been accepted by the Datum Gateway",
			ObservedGeneration: upstreamGateway.Generation,
		}

		programmedCondition := metav1.Condition{
			Type:               string(gatewayv1.ListenerConditionProgrammed),
			Status:             metav1.ConditionTrue,
			Reason:             string(gatewayv1.ListenerReasonProgrammed),
			Message:            "The listener has been programmed by the Datum Gateway",
			ObservedGeneration: upstreamGateway.Generation,
		}

		resolvedRefsCondition := metav1.Condition{
			Type:               string(gatewayv1.ListenerConditionResolvedRefs),
			Status:             metav1.ConditionTrue,
			Reason:             string(gatewayv1.ListenerReasonResolvedRefs),
			Message:            "The listener has been resolved by the Datum Gateway",
			ObservedGeneration: upstreamGateway.Generation,
		}

		// A hostname problem is reported instead of a certificate problem, since
		// the hostname has to be sorted out first.
		hostnameProblem := false

		if listener.Hostname != nil {

			if !slices.Contains(verifiedHostnames, string(*listener.Hostname)) {
				hostnameProblem = true
				acceptedCondition.Status = metav1.ConditionFalse
				acceptedCondition.Reason = networkingv1alpha.UnverifiedHostnamesPresent
				acceptedCondition.Message = fmt.Sprintf("The hostname %q has not been verified. Check status of Domains in the same namespace.", *listener.Hostname)

				programmedCondition.Status = metav1.ConditionFalse
				programmedCondition.Reason = acceptedCondition.Reason
				programmedCondition.Message = acceptedCondition.Message
			} else if slices.Contains(notClaimedHostnames, string(*listener.Hostname)) {
				hostnameProblem = true
				acceptedCondition.Status = metav1.ConditionFalse
				acceptedCondition.Reason = networkingv1alpha.HostnameInUseReason
				acceptedCondition.Message = fmt.Sprintf("The hostname %q is already attached to a resource.", *listener.Hostname)

				programmedCondition.Status = metav1.ConditionFalse
				programmedCondition.Reason = acceptedCondition.Reason
				programmedCondition.Message = acceptedCondition.Message
			}
		}

		// Tell the customer when a listener's certificate is the reason HTTPS
		// isn't running. The listener stays accepted because the configuration
		// is fine; only the certificate needs attention.
		if !hostnameProblem {
			if certStatus, gated := listenerCertHealth[listener.Name]; gated && !certStatus.healthy {
				resolvedRefsCondition.Status = metav1.ConditionFalse
				resolvedRefsCondition.Reason = string(gatewayv1.ListenerReasonInvalidCertificateRef)
				resolvedRefsCondition.Message = certStatus.message

				programmedCondition.Status = metav1.ConditionFalse
				programmedCondition.Reason = string(gatewayv1.ListenerReasonInvalid)
				programmedCondition.Message = certStatus.message
			}
		}

		apimeta.SetStatusCondition(&status.Conditions, acceptedCondition)
		apimeta.SetStatusCondition(&status.Conditions, programmedCondition)
		apimeta.SetStatusCondition(&status.Conditions, resolvedRefsCondition)

		listenerStatus = append(listenerStatus, status)
	}

	if !equality.Semantic.DeepEqual(upstreamGateway.Status.Listeners, listenerStatus) {
		logger.Info("listener status changes present")
		upstreamGateway.Status.Listeners = listenerStatus
		result.AddStatusUpdate(upstreamClient, upstreamGateway)
	} else {
		logger.Info("listener status unchanged")
	}

	return result
}

func (r *GatewayReconciler) ensureDownstreamHTTPRoute(
	ctx context.Context,
	upstreamClient client.Client,
	upstreamGateway *gatewayv1.Gateway,
	upstreamGatewayClassControllerName string,
	downstreamGateway *gatewayv1.Gateway,
	downstreamStrategy downstreamclient.ResourceStrategy,
	upstreamRoute gatewayv1.HTTPRoute,
) (result Result) {
	logger := log.FromContext(ctx)
	logger.Info("processing httproute", jsonKeyName, upstreamRoute.Name)

	downstreamClient := downstreamStrategy.GetClient()
	downstreamRouteObjectMeta, err := downstreamStrategy.ObjectMetaFromUpstreamObject(ctx, &upstreamRoute)
	if err != nil {
		result.Err = fmt.Errorf("failed to get downstream httproute object metadata: %w", err)
		return result
	}

	downstreamRoute := &gatewayv1.HTTPRoute{
		ObjectMeta: downstreamRouteObjectMeta,
	}

	rules, downstreamResources, downstreamResourcesToDelete, err := r.processDownstreamHTTPRouteRules(
		ctx,
		upstreamClient,
		upstreamGateway,
		upstreamRoute,
		downstreamGateway,
		downstreamStrategy,
	)
	if err != nil {
		result.Err = err
		return result
	}

	routeResult, err := controllerutil.CreateOrUpdate(ctx, downstreamClient, downstreamRoute, func() error {
		if err := downstreamStrategy.SetControllerReference(ctx, &upstreamRoute, downstreamRoute); err != nil {
			return fmt.Errorf("failed to set controller reference on downstream httproute: %w", err)
		}

		downstreamRoute.Spec = gatewayv1.HTTPRouteSpec{
			CommonRouteSpec: gatewayv1.CommonRouteSpec{
				// We currently only support same-namespace references, so just copy over
				// parentRefs from the upstream route.
				ParentRefs: upstreamRoute.Spec.ParentRefs,
			},
			Hostnames: upstreamRoute.Spec.Hostnames,
			Rules:     rules,
		}

		return nil
	})
	if err != nil {
		if apierrors.IsConflict(err) {
			result.RequeueAfter = 1 * time.Second
			return result
		}
		result.Err = err
		return result
	}

	// Create required downstream resources. Currently they're all specific to
	// the HTTPRoute resource, so we set it as the owner and let them get
	// cleaned up when the HTTPRoute is deleted.
	for _, resource := range downstreamResources {
		if err := controllerutil.SetControllerReference(downstreamRoute, resource, downstreamClient.Scheme()); err != nil {
			result.Err = err
			return result
		}

		desiredDownstreamResource := resource.DeepCopyObject()
		resourceResult, err := controllerutil.CreateOrUpdate(ctx, downstreamClient, resource, func() error {
			switch obj := resource.(type) {
			case *corev1.Service:
				desired := desiredDownstreamResource.(*corev1.Service)
				obj.Spec.Type = desired.Spec.Type
				obj.Spec.ClusterIP = desired.Spec.ClusterIP
				obj.Spec.InternalTrafficPolicy = desired.Spec.InternalTrafficPolicy
				obj.Spec.TrafficDistribution = desired.Spec.TrafficDistribution

				// Merge ports by name rather than overwriting the slice, so
				// server-defaulted fields like TargetPort are preserved. All other
				// user-controlled fields (Port, Protocol, AppProtocol) must be copied
				// from desired so changes to the upstream HTTPProxy backend (e.g.
				// flipping the origin scheme from https to http) propagate to the
				// downstream Service.
				for _, dp := range desired.Spec.Ports {
					found := false
					for i, ep := range obj.Spec.Ports {
						if ep.Name == dp.Name {
							obj.Spec.Ports[i].Port = dp.Port
							obj.Spec.Ports[i].Protocol = dp.Protocol
							obj.Spec.Ports[i].AppProtocol = dp.AppProtocol
							found = true
							break
						}
					}
					if !found {
						obj.Spec.Ports = append(obj.Spec.Ports, dp)
					}
				}
			case *discoveryv1.EndpointSlice:
				desiredEndpointSlice := desiredDownstreamResource.(*discoveryv1.EndpointSlice)
				// Since endpointslices get duplicated for routes, add them as a controller
				// owner in the downstream control plane.
				if err := controllerutil.SetControllerReference(downstreamRoute, obj, downstreamClient.Scheme()); err != nil {
					return fmt.Errorf("failed setting owner on endpointslice: %w", err)
				}
				obj.AddressType = desiredEndpointSlice.AddressType
				obj.Endpoints = desiredEndpointSlice.Endpoints
				obj.Ports = desiredEndpointSlice.Ports
			case *gatewayv1.BackendTLSPolicy:
				obj.Spec = desiredDownstreamResource.(*gatewayv1.BackendTLSPolicy).Spec
			}
			return nil
		})
		if err != nil {
			result.Err = err
			return result
		}

		gvk, err := apiutil.GVKForObject(resource, downstreamClient.Scheme())
		if err != nil {
			result.Err = err
			return result
		}

		logger.Info("downstream resource processed",
			"operation_result", resourceResult,
			jsonKeyKind, gvk.Kind,
			"namespace", resource.GetNamespace(),
			jsonKeyName, resource.GetName(),
		)
	}

	// Delete downstream resources that were previously desired but no longer
	// are. This catches cases like an HTTPProxy backend flipping from https to
	// http, which should remove the BackendTLSPolicy that had been created for
	// the previous state. Resources owned by the downstream HTTPRoute are
	// garbage collected when the route is deleted, but that does not cover
	// in-place transitions, so orphans must be cleaned up explicitly here.
	for _, resource := range downstreamResourcesToDelete {
		gvk, err := apiutil.GVKForObject(resource, downstreamClient.Scheme())
		if err != nil {
			result.Err = err
			return result
		}

		if err := downstreamClient.Delete(ctx, resource); err != nil && !apierrors.IsNotFound(err) {
			result.Err = fmt.Errorf("failed deleting stale downstream resource %s/%s: %w",
				resource.GetNamespace(), resource.GetName(), err)
			return result
		}

		logger.Info("stale downstream resource removed",
			jsonKeyKind, gvk.Kind,
			"namespace", resource.GetNamespace(),
			jsonKeyName, resource.GetName(),
		)
	}

	// Update the upstream route's parent status information
	var parentStatus *gatewayv1.RouteParentStatus
	for i, parent := range upstreamRoute.Status.Parents {
		if ptr.Deref(parent.ParentRef.Group, gatewayv1.GroupName) == gatewayv1.GroupName &&
			ptr.Deref(parent.ParentRef.Kind, KindGateway) == KindGateway &&
			string(parent.ParentRef.Name) == upstreamGateway.Name {
			parentStatus = &upstreamRoute.Status.Parents[i]
			break
		}
	}

	var insertParentStatus bool
	if parentStatus == nil {
		insertParentStatus = true
		parentStatus = &gatewayv1.RouteParentStatus{
			ControllerName: gatewayv1.GatewayController(upstreamGatewayClassControllerName),
			ParentRef: gatewayv1.ParentReference{
				Name: gatewayv1.ObjectName(upstreamGateway.Name),
			},
		}
	}

	// Get the status of this parent from the downstream route
	var downstreamParentStatus *gatewayv1.RouteParentStatus
	for _, parent := range downstreamRoute.Status.Parents {
		if string(parent.ParentRef.Name) == downstreamGateway.Name {
			downstreamParentStatus = &parent
			break
		}
	}

	if downstreamParentStatus != nil {
		if c := apimeta.FindStatusCondition(downstreamParentStatus.Conditions, string(gatewayv1.RouteConditionAccepted)); c != nil {
			message := "Route has not been accepted"
			if c.Status == metav1.ConditionTrue {
				message = "Route is accepted"
			}

			apimeta.SetStatusCondition(&parentStatus.Conditions, metav1.Condition{
				Message:            message,
				Type:               string(gatewayv1.RouteConditionAccepted),
				Reason:             c.Reason,
				Status:             c.Status,
				ObservedGeneration: upstreamRoute.Generation,
			})
		}

		if c := apimeta.FindStatusCondition(downstreamParentStatus.Conditions, string(gatewayv1.RouteConditionResolvedRefs)); c != nil {
			message := "Object references for the Route have not been resolved"
			if c.Status == metav1.ConditionTrue {
				message = "Resolved all the Object references for the Route"
			}

			apimeta.SetStatusCondition(&parentStatus.Conditions, metav1.Condition{
				Message:            message,
				Type:               string(gatewayv1.RouteConditionResolvedRefs),
				Reason:             c.Reason,
				Status:             c.Status,
				ObservedGeneration: upstreamRoute.Generation,
			})
		}
	} else {
		logger.Info("did not find downstream parent status for gateway")
	}

	if insertParentStatus {
		upstreamRoute.Status.Parents = append(upstreamRoute.Status.Parents, *parentStatus)
	}

	result.AddStatusUpdate(upstreamClient, &upstreamRoute)

	logger.Info("downstream httproute processed", "operation_result", routeResult)

	return result
}

// processDownstreamHTTPRouteRules is a helper function that processes the
// rules of an HTTPRoute and returns the rules, the downstream resources that
// need to be created or updated, and the downstream resources that must be
// deleted because they were created by a previous reconcile but are no longer
// desired (for example a BackendTLSPolicy left over from when the backend was
// https before the user switched it to http).
func (r *GatewayReconciler) processDownstreamHTTPRouteRules(
	ctx context.Context,
	upstreamClient client.Client,
	upstreamGateway *gatewayv1.Gateway,
	upstreamRoute gatewayv1.HTTPRoute,
	downstreamGateway *gatewayv1.Gateway,
	downstreamStrategy downstreamclient.ResourceStrategy,
) (rules []gatewayv1.HTTPRouteRule, downstreamResources []client.Object, downstreamResourcesToDelete []client.Object, err error) {

	// We need to create a Service for each (BackendRef, EndpointSlice)
	// combination as different backendRefs may use different hostnames in URL
	// rewrite filters which are used to drive the creation of BackendTLSPolicies.
	// A BackendTLSPolicy is associated with either an entire service, or a
	// specific port of a service, and we cannot build port aliases in the
	// downstream service, as TargetPort is ignored for headless services.
	//
	// As a result of this, we also need to create an EndpointSlice for each
	// downstream service, because an EndpointSlice may only be associated with
	// a single service.

	logger := log.FromContext(ctx)

	for ruleIdx, rule := range upstreamRoute.Spec.Rules {
		var backendRefs []gatewayv1.HTTPBackendRef
		for backendRefIdx, backendRef := range rule.BackendRefs {

			if backendRef.Kind == nil {
				// Should not happen, as the default kind is Service
				continue
			}

			switch *backendRef.Kind {
			// Transform EndpointSlice references into Service references.
			case KindEndpointSlice:
				// Fetch the upstream EndpointSlice
				var upstreamEndpointSlice discoveryv1.EndpointSlice
				if err := upstreamClient.Get(ctx, types.NamespacedName{
					Namespace: string(ptr.Deref(backendRef.Namespace, gatewayv1.Namespace(upstreamGateway.Namespace))),
					Name:      string(backendRef.Name),
				}, &upstreamEndpointSlice); err != nil {
					return nil, nil, nil, err
				}

				if backendRef.Port == nil {
					// Should be protected by validation, but check just in case.
					logger.Info("no port defined in backendRef", "backendRef", backendRef)
					return nil, nil, nil, fmt.Errorf("no port defined in backendRef")
				}

				if !controllerutil.ContainsFinalizer(&upstreamEndpointSlice, gatewayControllerGCFinalizer) {
					controllerutil.AddFinalizer(&upstreamEndpointSlice, gatewayControllerGCFinalizer)
					if err := upstreamClient.Update(ctx, &upstreamEndpointSlice); err != nil {
						return nil, nil, nil, fmt.Errorf("failed to add finalizer to endpointslice: %w", err)
					}
				}

				var ports []corev1.ServicePort
				var appProtocol *string
				var endpointPort *discoveryv1.EndpointPort
				for _, port := range upstreamEndpointSlice.Ports {
					ports = append(ports, corev1.ServicePort{
						Name:        ptr.Deref(port.Name, ""),
						Protocol:    ptr.Deref(port.Protocol, corev1.ProtocolTCP),
						AppProtocol: port.AppProtocol,
						Port:        *port.Port,
					})

					if *backendRef.Port == *port.Port {
						if port.Name == nil {
							// This should be protected by validation, but check just in case.
							logger.Info("no port name defined in upstream endpointslice", "endpointslice", upstreamEndpointSlice.Name, "port", port)
							return nil, nil, nil, fmt.Errorf("no port name defined in upstream endpointslice")
						}
						appProtocol = port.AppProtocol
						endpointPort = ptr.To(port)
					}
				}

				if endpointPort == nil {
					logger.Info("port not found in upstream endpointslice", "endpointslice", upstreamEndpointSlice.Name, "port", *backendRef.Port)
					return nil, nil, nil, fmt.Errorf("port not found in upstream endpointslice")
				}

				// Construct a name to use for the service and endpointslice that the
				// downstream backendRef will reference.
				resourceName := fmt.Sprintf("route-%s-rule-%d-backendref-%d", upstreamRoute.UID, ruleIdx, backendRefIdx)

				downstreamService := &corev1.Service{
					ObjectMeta: metav1.ObjectMeta{
						Namespace: downstreamGateway.Namespace,
						Name:      resourceName,
					},
					Spec: corev1.ServiceSpec{
						Type:                  corev1.ServiceTypeClusterIP,
						ClusterIP:             clusterIPNone,
						Ports:                 ports,
						InternalTrafficPolicy: ptr.To(corev1.ServiceInternalTrafficPolicyCluster),
						TrafficDistribution:   ptr.To(corev1.ServiceTrafficDistributionPreferClose),
					},
				}
				downstreamResources = append(downstreamResources, downstreamService)

				downstreamEndpointSlice := &discoveryv1.EndpointSlice{
					ObjectMeta: metav1.ObjectMeta{
						Namespace: downstreamGateway.Namespace,
						Name:      resourceName,
						Labels: map[string]string{
							downstreamclient.UpstreamOwnerNameLabel: upstreamEndpointSlice.Name,
							discoveryv1.LabelServiceName:            downstreamService.Name,
						},
					},
					AddressType: upstreamEndpointSlice.AddressType,
					Endpoints:   upstreamEndpointSlice.Endpoints,
					Ports:       upstreamEndpointSlice.Ports,
				}

				if err := downstreamStrategy.SetControllerReference(ctx, &upstreamEndpointSlice, downstreamEndpointSlice); err != nil {
					return nil, nil, nil, fmt.Errorf("failed to set controller reference on downstream endpointslice: %w", err)
				}

				downstreamResources = append(downstreamResources, downstreamEndpointSlice)

				backendObjectReference := gatewayv1.BackendObjectReference{
					Namespace: ptr.To(gatewayv1.Namespace(downstreamGateway.Namespace)),
					Kind:      ptr.To(gatewayv1.Kind(KindService)),
					Name:      gatewayv1.ObjectName(downstreamService.Name),
					Port:      backendRef.Port,
				}

				downstreamHTTPBackendRef := gatewayv1.HTTPBackendRef{
					BackendRef: gatewayv1.BackendRef{
						Weight:                 backendRef.Weight,
						BackendObjectReference: backendObjectReference,
					},
					Filters: backendRef.Filters,
				}

				backendRefs = append(backendRefs, downstreamHTTPBackendRef)

				if appProtocol != nil && *appProtocol == SchemeHTTPS {
					var hostname *gatewayv1.PreciseHostname

					// Prefer the cert hostname recorded by the httpproxy
					// controller on the upstream EndpointSlice. URLRewrite
					// may now carry a user-supplied Host header override
					// instead of the backend FQDN, so it's no longer a
					// reliable source for BackendTLSPolicy SAN validation.
					if v, ok := upstreamEndpointSlice.Annotations[BackendCertHostnameAnnotation]; ok && v != "" {
						hostname = ptr.To(gatewayv1.PreciseHostname(v))
					}

					// Fall back to looking at rule filters for a hostname
					// (preserves behaviour for EndpointSlices that predate
					// the annotation).
					if hostname == nil {
						for _, filter := range rule.Filters {
							if filter.URLRewrite != nil {
								hostname = filter.URLRewrite.Hostname
								break
							}
						}
					}

					if hostname == nil {
						// TODO(jreese) set the RouteConditionResolvedRefs condition to
						// False, as the hostname is not present.
						return nil, nil, nil, fmt.Errorf("no hostname found in URLRewrite filters or EndpointSlice annotation on backendRef or Route %q", upstreamRoute.Name)
					}

					// BackendTLSPolicy graduated from v1alpha3 to v1 in gateway-api v1.5.
					backendTLSPolicy := &gatewayv1.BackendTLSPolicy{
						ObjectMeta: metav1.ObjectMeta{
							Namespace: downstreamGateway.Namespace,
							Name:      resourceName,
						},
						Spec: gatewayv1.BackendTLSPolicySpec{
							TargetRefs: []gatewayv1.LocalPolicyTargetReferenceWithSectionName{
								// TODO(jreese): We may have multiple ports that we need to set
								// the policy on.
								{
									LocalPolicyTargetReference: gatewayv1.LocalPolicyTargetReference{
										Kind: gatewayv1.Kind(KindService),
										Name: gatewayv1.ObjectName(downstreamService.Name),
									},
									SectionName: ptr.To(gatewayv1.SectionName(*endpointPort.Name)),
								},
							},
							Validation: gatewayv1.BackendTLSPolicyValidation{
								WellKnownCACertificates: ptr.To(gatewayv1.WellKnownCACertificatesSystem),
								Hostname:                *hostname,
							},
						},
					}

					downstreamResources = append(downstreamResources, backendTLSPolicy)
				} else {
					// The backend is not https, so any BackendTLSPolicy that may
					// have been created by a previous reconcile (when the backend
					// was https) must be removed. The policy name is deterministic
					// from the route UID and backend indices, so we can target it
					// directly without listing.
					downstreamResourcesToDelete = append(downstreamResourcesToDelete, &gatewayv1.BackendTLSPolicy{
						ObjectMeta: metav1.ObjectMeta{
							Namespace: downstreamGateway.Namespace,
							Name:      resourceName,
						},
					})
				}

			case "Service":
				fallthrough
			case envoygatewayv1alpha1.KindBackend:
				backendRefs = append(backendRefs, backendRef)

			// Other types of backend refs will be handled in the future.
			default:
				logger.Info("unknown backend ref kind", jsonKeyKind, *backendRef.Kind)
				continue
			}
		}

		rules = append(rules, gatewayv1.HTTPRouteRule{
			Name:               rule.Name,
			Filters:            rule.Filters,
			Matches:            rule.Matches,
			BackendRefs:        backendRefs,
			Timeouts:           rule.Timeouts,
			Retry:              rule.Retry,
			SessionPersistence: rule.SessionPersistence,
		})
	}

	return rules, downstreamResources, downstreamResourcesToDelete, nil
}

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

	downstreamGatewaySource := mcsource.TypedKind(
		&gatewayv1.Gateway{},
		downstreamclient.TypedEnqueueRequestForUpstreamOwner[*gatewayv1.Gateway](&gatewayv1.Gateway{}),
	)

	downstreamGatewayClusterSource, _, _ := downstreamGatewaySource.ForCluster("", r.DownstreamCluster)

	downstreamHTTPRouteSource := mcsource.Kind(
		&gatewayv1.HTTPRoute{},
		r.listGatewaysAttachedByDownstreamHTTPRoute,
	)

	downstreamHTTPRouteClusterSource, _, _ := downstreamHTTPRouteSource.ForCluster("", r.DownstreamCluster)

	downstreamCertificateSource := mcsource.TypedKind(
		&cmv1.Certificate{},
		downstreamclient.TypedEnqueueRequestForUpstreamOwner[*cmv1.Certificate](&gatewayv1.Gateway{}),
	)

	downstreamCertificateClusterSource, _, _ := downstreamCertificateSource.ForCluster("", r.DownstreamCluster)

	builder := mcbuilder.ControllerManagedBy(mgr).
		For(&gatewayv1.Gateway{}).
		Watches(
			&gatewayv1.HTTPRoute{},
			mchandler.EnqueueRequestsFromMapFunc(r.listGatewaysAttachedByHTTPRoute),
		).
		Watches(
			&discoveryv1.EndpointSlice{},
			r.listGatewaysForEndpointSliceFunc,
		).
		Watches(
			&networkingv1alpha.Domain{},
			r.listGatewaysForDomainFunc,
		).
		Watches(
			&envoygatewayv1alpha1.HTTPRouteFilter{},
			r.listGatewaysForHTTPRouteFilterFunc,
		).
		WatchesRawSource(downstreamGatewayClusterSource).
		WatchesRawSource(downstreamHTTPRouteClusterSource).
		WatchesRawSource(downstreamCertificateClusterSource)

	if r.Config.Gateway.EnableDNSIntegration {
		builder = builder.
			Watches(
				&dnsv1alpha1.DNSZone{},
				r.listGatewaysForDNSZoneFunc,
			).
			Watches(
				&dnsv1alpha1.DNSRecordSet{},
				r.listGatewaysForDNSRecordSetFunc,
			)
	}

	return builder.
		WithOptions(controller.TypedOptions[mcreconcile.Request]{
			MaxConcurrentReconciles: r.Config.Gateway.MaxConcurrentReconciles,
		}).
		Named("gateway").Complete(r)
}

// listGatewaysAttachedByHTTPRoute is a watch predicate which finds all Gateways mentioned
// in HTTPRoutes' Parents field.
func (r *GatewayReconciler) listGatewaysAttachedByHTTPRoute(ctx context.Context, obj client.Object) []ctrl.Request {
	logger := log.FromContext(ctx)

	httpRoute, ok := obj.(*gatewayv1.HTTPRoute)
	if !ok {
		logger.Error(
			fmt.Errorf("unexpected object type"),
			"HTTPRoute watch predicate received unexpected object type",
			"expected", "*gatewayapi.HTTPRoute", "found", fmt.Sprintf("%T", obj),
		)
		return nil
	}

	var reqs []ctrl.Request

	for _, parentRef := range httpRoute.Spec.ParentRefs {
		if ptr.Deref(parentRef.Group, gatewayv1.GroupName) == gatewayv1.GroupName &&
			ptr.Deref(parentRef.Kind, KindGateway) == KindGateway {
			reqs = append(reqs, ctrl.Request{
				NamespacedName: types.NamespacedName{
					Namespace: string(ptr.Deref(parentRef.Namespace, gatewayv1.Namespace(httpRoute.Namespace))),
					Name:      string(parentRef.Name),
				},
			})
		}
	}

	return reqs
}

// listGatewaysAttachedByDownstreamHTTPRoute enqueues reconciliation requests for Gateways
// referenced by a downstream HTTPRoute's ParentRefs.
func (r *GatewayReconciler) listGatewaysAttachedByDownstreamHTTPRoute(clusterName multicluster.ClusterName, cl cluster.Cluster) handler.TypedEventHandler[*gatewayv1.HTTPRoute, mcreconcile.Request] {
	return handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, httpRoute *gatewayv1.HTTPRoute) []mcreconcile.Request {
		logger := log.FromContext(ctx)
		logger.Info("enqueueing upstream gateway for downstream httproute", jsonKeyName, httpRoute.Name)

		var reqs []mcreconcile.Request
		if _, ok := httpRoute.Labels[downstreamclient.UpstreamOwnerClusterNameLabel]; ok {
			for _, parentRef := range httpRoute.Spec.ParentRefs {
				reqs = append(reqs, mcreconcile.Request{
					Request: ctrl.Request{
						NamespacedName: types.NamespacedName{
							Namespace: httpRoute.Labels[downstreamclient.UpstreamOwnerNamespaceLabel],
							Name:      string(parentRef.Name),
						},
					},
					ClusterName: multicluster.ClusterName(downstreamclient.UpstreamClusterNameFromLabel(httpRoute.Labels[downstreamclient.UpstreamOwnerClusterNameLabel])),
				})

			}
		}
		return reqs
	})
}

// listGatewaysForEndpointSliceFunc creates an event handler that watches EndpointSlice changes
// and determines which Gateways need to be reconciled as a result of those changes.
//
// This function implements the watch pattern for EndpointSlice resources in multi-cluster scenarios.
// When an EndpointSlice changes (created, updated, or deleted), this handler:
//
//  1. Examines all HTTPRoutes in the cluster to find those that reference the changed EndpointSlice
//     as a backend (via BackendRefs with Kind=EndpointSlice)
//  2. For each matching HTTPRoute, identifies the parent Gateways referenced in ParentRefs
//  3. Returns reconcile requests for those Gateways so they can update their configuration
//
// This ensures that Gateway resources are automatically updated when their backend EndpointSlices
// change, maintaining proper traffic routing and load balancing.
func (r *GatewayReconciler) listGatewaysForEndpointSliceFunc(clusterName multicluster.ClusterName, cl cluster.Cluster) handler.TypedEventHandler[client.Object, mcreconcile.Request] {
	return handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj client.Object) []mcreconcile.Request {
		endpointSlice := obj.(*discoveryv1.EndpointSlice)
		logger := log.FromContext(ctx)

		var httpRoutes gatewayv1.HTTPRouteList
		if err := cl.GetClient().List(ctx, &httpRoutes); err != nil {
			logger.Error(err, "failed to list HTTPRoutes")
			return nil
		}

		var requests []mcreconcile.Request

		for _, route := range httpRoutes.Items {
			for _, rule := range route.Spec.Rules {
				for _, backendRef := range rule.BackendRefs {
					if ptr.Deref(backendRef.Kind, "") == KindEndpointSlice {
						backendNamespace := string(ptr.Deref(backendRef.Namespace, gatewayv1.Namespace(route.Namespace)))

						if backendNamespace == endpointSlice.Namespace && string(backendRef.Name) == endpointSlice.Name {
							for _, parentRef := range route.Spec.ParentRefs {
								if ptr.Deref(parentRef.Group, gatewayv1.GroupName) == gatewayv1.GroupName &&
									ptr.Deref(parentRef.Kind, KindGateway) == KindGateway {
									gatewayNamespace := string(ptr.Deref(parentRef.Namespace, gatewayv1.Namespace(route.Namespace)))

									requests = append(requests, mcreconcile.Request{
										ClusterName: clusterName,
										Request: reconcile.Request{
											NamespacedName: types.NamespacedName{
												Namespace: gatewayNamespace,
												Name:      string(parentRef.Name),
											},
										},
									})
								}
							}
						}
					}
				}
			}
		}

		return requests
	})
}

func (r *GatewayReconciler) listGatewaysForDomainFunc(clusterName multicluster.ClusterName, cl cluster.Cluster) handler.TypedEventHandler[client.Object, mcreconcile.Request] {
	return handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj client.Object) []mcreconcile.Request {
		domain := obj.(*networkingv1alpha.Domain)

		// Only enqueue if the domain is verified
		if !apimeta.IsStatusConditionTrue(domain.Status.Conditions, networkingv1alpha.DomainConditionVerified) {
			return nil
		}

		logger := log.FromContext(ctx)

		var gatewayList gatewayv1.GatewayList
		if err := cl.GetClient().List(ctx, &gatewayList, client.InNamespace(domain.Namespace)); err != nil {
			logger.Error(err, "failed to list Gateways")
			return nil
		}

		var requests []mcreconcile.Request

		for _, gateway := range gatewayList.Items {
			for _, l := range gateway.Spec.Listeners {
				if l.Hostname == nil {
					continue
				}
				hostname := string(*l.Hostname)

				if hostname == domain.Spec.DomainName || strings.HasSuffix(hostname, "."+domain.Spec.DomainName) {
					requests = append(requests, mcreconcile.Request{
						ClusterName: clusterName,
						Request: reconcile.Request{
							NamespacedName: client.ObjectKeyFromObject(&gateway),
						},
					})
					continue
				}
			}
		}

		return requests
	})
}

func (r *GatewayReconciler) listGatewaysForHTTPRouteFilterFunc(clusterName multicluster.ClusterName, cl cluster.Cluster) handler.TypedEventHandler[client.Object, mcreconcile.Request] {
	return handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj client.Object) []mcreconcile.Request {
		httpRouteFilter := obj.(*envoygatewayv1alpha1.HTTPRouteFilter)

		logger := log.FromContext(ctx)

		var httpRoutes gatewayv1.HTTPRouteList
		if err := cl.GetClient().List(ctx, &httpRoutes, client.InNamespace(httpRouteFilter.Namespace)); err != nil {
			logger.Error(err, "failed to list HTTPRoutes")
			return nil
		}

		var requests []mcreconcile.Request

		for _, route := range httpRoutes.Items {
			for _, rule := range route.Spec.Rules {
				for _, backendRef := range rule.BackendRefs {
					for _, filter := range backendRef.Filters {
						if filter.ExtensionRef != nil &&
							filter.ExtensionRef.Group == envoygatewayv1alpha1.GroupName &&
							filter.ExtensionRef.Kind == envoygatewayv1alpha1.KindHTTPRouteFilter {
							for _, parentRef := range route.Spec.ParentRefs {
								if ptr.Deref(parentRef.Group, gatewayv1.GroupName) == gatewayv1.GroupName &&
									ptr.Deref(parentRef.Kind, KindGateway) == KindGateway {
									gatewayNamespace := string(ptr.Deref(parentRef.Namespace, gatewayv1.Namespace(route.Namespace)))

									requests = append(requests, mcreconcile.Request{
										ClusterName: clusterName,
										Request: reconcile.Request{
											NamespacedName: types.NamespacedName{
												Namespace: gatewayNamespace,
												Name:      string(parentRef.Name),
											},
										},
									})
								}
							}
						}
					}
				}
			}
		}

		return requests
	})
}