package controller

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"sort"
	"strings"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"

	bindv1alpha1 "git.unkin.net/unkin/bind-operator/api/v1alpha1"
	"git.unkin.net/unkin/bind-operator/internal/bind"
)

// BindClusterReconciler manages the StatefulSet, Services, ConfigMap and
// Secrets backing a BindCluster, and re-renders named.conf when dependent
// objects (ACLs, views, policies, keys, catalog) change.
type BindClusterReconciler struct {
	client.Client
	Scheme *runtime.Scheme
	Exec   *bind.Executor
}

// +kubebuilder:rbac:groups=bind.unkin.net,resources=bindclusters,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=bind.unkin.net,resources=bindclusters/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=bind.unkin.net,resources=bindacls;bindviews;bindpolicies;binddnssecpolicies;bindcatalogzones;bindtsigkeys,verbs=get;list;watch
// +kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=services;configmaps;secrets,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch
// +kubebuilder:rbac:groups="",resources=pods/exec,verbs=create;get

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

	var cluster bindv1alpha1.BindCluster
	if err := r.Get(ctx, req.NamespacedName, &cluster); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	if err := r.reconcileRNDCSecret(ctx, &cluster); err != nil {
		return ctrl.Result{}, fmt.Errorf("rndc secret: %w", err)
	}
	if err := r.reconcileKeysSecret(ctx, &cluster); err != nil {
		return ctrl.Result{}, fmt.Errorf("keys secret: %w", err)
	}
	if err := r.reconcileConfigMap(ctx, &cluster); err != nil {
		return ctrl.Result{}, fmt.Errorf("configmap: %w", err)
	}
	if err := r.reconcileServices(ctx, &cluster); err != nil {
		return ctrl.Result{}, fmt.Errorf("services: %w", err)
	}
	sts, err := r.reconcileStatefulSet(ctx, &cluster)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("statefulset: %w", err)
	}

	// Best-effort: reload configuration on ready pods so ConfigMap changes take
	// effect without a rollout.
	r.reloadReadyPods(ctx, &cluster)

	// Status.
	cluster.Status.ObservedGeneration = cluster.Generation
	cluster.Status.Replicas = cluster.Spec.Replicas
	cluster.Status.ReadyReplicas = sts.Status.ReadyReplicas
	cluster.Status.PrimaryPod = primaryPodName(cluster.Name)
	cluster.Status.PrimaryService = primaryAddress(cluster.Name, cluster.Namespace)
	ready := sts.Status.ReadyReplicas == cluster.Spec.Replicas && cluster.Spec.Replicas > 0
	if ready {
		cluster.Status.Phase = "Ready"
	} else {
		cluster.Status.Phase = "Progressing"
	}
	setReady(&cluster.Status.Conditions, cluster.Generation, ready, "Reconciled",
		fmt.Sprintf("%d/%d replicas ready", sts.Status.ReadyReplicas, cluster.Spec.Replicas))
	if err := r.Status().Update(ctx, &cluster); err != nil {
		return ctrl.Result{}, err
	}

	if !ready {
		return ctrl.Result{RequeueAfter: requeueShort}, nil
	}
	logger.V(1).Info("cluster reconciled", "cluster", cluster.Name, "ready", sts.Status.ReadyReplicas)
	return ctrl.Result{}, nil
}

func (r *BindClusterReconciler) reconcileRNDCSecret(ctx context.Context, c *bindv1alpha1.BindCluster) error {
	name := rndcSecretName(c.Name)
	var existing corev1.Secret
	err := r.Get(ctx, types.NamespacedName{Namespace: c.Namespace, Name: name}, &existing)
	if err == nil {
		return nil // rndc key is generated once and preserved
	}
	if !apierrors.IsNotFound(err) {
		return err
	}
	secret, genErr := bind.GenerateSecret(32)
	if genErr != nil {
		return genErr
	}
	keyClause := bind.KeyClause("rndc-key", "hmac-sha256", secret)
	rndcConf := fmt.Sprintf("include \"%s\";\noptions {\n    default-key \"rndc-key\";\n    default-server 127.0.0.1;\n    default-port 953;\n};\n", bind.RndcKeyPath)
	s := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: c.Namespace, Labels: commonLabels(c.Name)},
		Data: map[string][]byte{
			"rndc.key":  []byte(keyClause),
			"rndc.conf": []byte(rndcConf),
		},
	}
	if err := ctrl.SetControllerReference(c, s, r.Scheme); err != nil {
		return err
	}
	return r.Create(ctx, s)
}

func (r *BindClusterReconciler) reconcileKeysSecret(ctx context.Context, c *bindv1alpha1.BindCluster) error {
	var keys bindv1alpha1.BindTSIGKeyList
	if err := r.List(ctx, &keys, client.InNamespace(c.Namespace)); err != nil {
		return err
	}
	// Include keys scoped to this cluster (spec.clusterRef == name) and shared
	// keys (empty clusterRef). This keeps keys from leaking across clusters that
	// share a namespace.
	var items []bindv1alpha1.BindTSIGKey
	for _, k := range keys.Items {
		if k.Spec.ClusterRef == "" || k.Spec.ClusterRef == c.Name {
			items = append(items, k)
		}
	}
	sort.Slice(items, func(i, j int) bool { return items[i].Name < items[j].Name })

	var b strings.Builder
	b.WriteString("// Managed by bind-operator.\n")
	for _, k := range items {
		secretName := k.Status.SecretName
		if secretName == "" {
			secretName = k.Spec.SecretName
		}
		if secretName == "" {
			secretName = k.Name + "-tsig"
		}
		var secret corev1.Secret
		if err := r.Get(ctx, types.NamespacedName{Namespace: c.Namespace, Name: secretName}, &secret); err != nil {
			continue // key not yet materialised; skip until its controller runs
		}
		keyName := k.Spec.KeyName
		if keyName == "" {
			keyName = k.Name
		}
		alg := string(secret.Data["algorithm"])
		if alg == "" {
			alg = string(bindv1alpha1.TSIGHMACSHA256)
		}
		b.WriteString(bind.KeyClause(keyName, alg, string(secret.Data["secret"])))
	}

	return r.upsertSecret(ctx, c, keysSecretName(c.Name), map[string][]byte{"keys.conf": []byte(b.String())})
}

func (r *BindClusterReconciler) reconcileConfigMap(ctx context.Context, c *bindv1alpha1.BindCluster) error {
	// BIND primaries/default-primaries need an IP address, not a DNS name. Use
	// the stable primary Service ClusterIP so secondaries keep transferring
	// across primary pod restarts (falls back to the pod IP when no primary
	// Service exists; the Pod/Service watches re-render when it changes).
	in := bind.RenderInput{Cluster: c, PrimaryAddress: primaryTransferAddress(ctx, r.Client, c)}

	var acls bindv1alpha1.BindACLList
	if err := r.List(ctx, &acls, client.InNamespace(c.Namespace)); err == nil {
		for _, a := range acls.Items {
			if a.Spec.ClusterRef == "" || a.Spec.ClusterRef == c.Name {
				in.ACLs = append(in.ACLs, a)
			}
		}
	}
	var views bindv1alpha1.BindViewList
	if err := r.List(ctx, &views, client.InNamespace(c.Namespace)); err == nil {
		for _, v := range views.Items {
			if v.Spec.ClusterRef == c.Name {
				in.Views = append(in.Views, v)
			}
		}
	}
	var policies bindv1alpha1.BindPolicyList
	if err := r.List(ctx, &policies, client.InNamespace(c.Namespace)); err == nil {
		for _, p := range policies.Items {
			if p.Spec.ClusterRef == c.Name {
				in.Policies = append(in.Policies, p)
			}
		}
	}
	var dnssec bindv1alpha1.BindDNSSECPolicyList
	if err := r.List(ctx, &dnssec, client.InNamespace(c.Namespace)); err == nil {
		for _, d := range dnssec.Items {
			if d.Spec.ClusterRef == c.Name {
				in.DNSSECPolicies = append(in.DNSSECPolicies, d)
			}
		}
	}

	// Forward zones are configuration (no data), so they are rendered into
	// named.conf on every pod rather than added dynamically to the primary.
	var zones bindv1alpha1.BindZoneList
	if err := r.List(ctx, &zones, client.InNamespace(c.Namespace)); err == nil {
		for _, z := range zones.Items {
			if z.Spec.ClusterRef == c.Name && z.Spec.Type == bindv1alpha1.ZoneForward {
				in.Forwards = append(in.Forwards, z)
			}
		}
	}

	var catalogs bindv1alpha1.BindCatalogZoneList
	if err := r.List(ctx, &catalogs, client.InNamespace(c.Namespace)); err == nil {
		for i := range catalogs.Items {
			if catalogs.Items[i].Spec.ClusterRef == c.Name {
				in.Catalog = &catalogs.Items[i]
				break
			}
		}
	}

	// Secondaries accept intra-cluster NOTIFYs signed with the cluster's catalog
	// transfer TSIG key (the primary signs its also-notify NOTIFYs with it — see
	// bindzone_controller). Render `allow-notify { key "<name>"; }` — a static key
	// element, NO pod IPs — so it never changes on pod churn and cannot re-render
	// the restart-scoped config (the v0.2.5 roll loop). Resolve the catalog's
	// TransferKeyRef to the BIND key name (KeyName override, else the ref).
	if in.Catalog != nil && in.Catalog.Spec.TransferKeyRef != "" {
		in.NotifyKeyName = tsigKeyName(ctx, r.Client, c.Namespace, in.Catalog.Spec.TransferKeyRef)
	}

	primaryConf, secondaryConf := bind.RenderNamedConf(in)
	data := map[string]string{
		"named.conf.primary":   primaryConf,
		"named.conf.secondary": secondaryConf,
		"entrypoint.sh":        entrypointScript(),
	}
	return r.upsertConfigMap(ctx, c, configMapName(c.Name), data)
}

func (r *BindClusterReconciler) reconcileServices(ctx context.Context, c *bindv1alpha1.BindCluster) error {
	dnsPorts := []corev1.ServicePort{
		{Name: "dns-udp", Port: 53, Protocol: corev1.ProtocolUDP, TargetPort: intstrFromInt(53)},
		{Name: "dns-tcp", Port: 53, Protocol: corev1.ProtocolTCP, TargetPort: intstrFromInt(53)},
	}

	headless := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{Name: headlessServiceName(c.Name), Namespace: c.Namespace, Labels: commonLabels(c.Name)},
		Spec: corev1.ServiceSpec{
			ClusterIP:                corev1.ClusterIPNone,
			PublishNotReadyAddresses: true,
			Selector:                 commonLabels(c.Name),
			Ports:                    dnsPorts,
		},
	}
	if err := r.upsertService(ctx, c, headless); err != nil {
		return err
	}

	svcType := c.Spec.Service.Type
	if svcType == "" {
		svcType = corev1.ServiceTypeClusterIP
	}
	client := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:        clientServiceName(c.Name),
			Namespace:   c.Namespace,
			Labels:      commonLabels(c.Name),
			Annotations: c.Spec.Service.Annotations,
		},
		Spec: corev1.ServiceSpec{
			Type:           svcType,
			Selector:       commonLabels(c.Name),
			Ports:          dnsPorts,
			LoadBalancerIP: c.Spec.Service.LoadBalancerIP,
		},
	}
	// externalTrafficPolicy is only valid for LoadBalancer/NodePort Services.
	if svcType == corev1.ServiceTypeLoadBalancer || svcType == corev1.ServiceTypeNodePort {
		client.Spec.ExternalTrafficPolicy = c.Spec.Service.ExternalTrafficPolicy
	}
	if err := r.upsertService(ctx, c, client); err != nil {
		return err
	}

	// Primary (write) Service: routes only to pod-0. Created when configured,
	// deleted when removed.
	return r.reconcilePrimaryService(ctx, c, dnsPorts)
}

func (r *BindClusterReconciler) reconcilePrimaryService(ctx context.Context, c *bindv1alpha1.BindCluster, dnsPorts []corev1.ServicePort) error {
	name := primaryServiceName(c.Name)
	if c.Spec.PrimaryService == nil {
		var existing corev1.Service
		err := r.Get(ctx, types.NamespacedName{Namespace: c.Namespace, Name: name}, &existing)
		if apierrors.IsNotFound(err) {
			return nil
		}
		if err != nil {
			return err
		}
		return client.IgnoreNotFound(r.Delete(ctx, &existing))
	}

	ps := c.Spec.PrimaryService
	psType := ps.Type
	if psType == "" {
		psType = corev1.ServiceTypeClusterIP
	}
	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:        name,
			Namespace:   c.Namespace,
			Labels:      commonLabels(c.Name),
			Annotations: ps.Annotations,
		},
		Spec: corev1.ServiceSpec{
			Type:           psType,
			Selector:       primaryPodSelector(c.Name),
			Ports:          dnsPorts,
			LoadBalancerIP: ps.LoadBalancerIP,
		},
	}
	if psType == corev1.ServiceTypeLoadBalancer || psType == corev1.ServiceTypeNodePort {
		svc.Spec.ExternalTrafficPolicy = ps.ExternalTrafficPolicy
	}
	return r.upsertService(ctx, c, svc)
}

func (r *BindClusterReconciler) reconcileStatefulSet(ctx context.Context, c *bindv1alpha1.BindCluster) (*appsv1.StatefulSet, error) {
	labels := commonLabels(c.Name)
	replicas := c.Spec.Replicas
	image := c.Spec.Image
	if image == "" {
		image = defaultBindImage
	}
	storageSize := c.Spec.StorageSize
	if storageSize == "" {
		storageSize = "1Gi"
	}
	qty, err := resource.ParseQuantity(storageSize)
	if err != nil {
		return nil, fmt.Errorf("parse storageSize: %w", err)
	}

	projected := corev1.Volume{
		Name: "bind-etc",
		VolumeSource: corev1.VolumeSource{Projected: &corev1.ProjectedVolumeSource{Sources: []corev1.VolumeProjection{
			{ConfigMap: &corev1.ConfigMapProjection{LocalObjectReference: corev1.LocalObjectReference{Name: configMapName(c.Name)}}},
			{Secret: &corev1.SecretProjection{LocalObjectReference: corev1.LocalObjectReference{Name: keysSecretName(c.Name)}}},
			{Secret: &corev1.SecretProjection{LocalObjectReference: corev1.LocalObjectReference{Name: rndcSecretName(c.Name)}}},
		}}},
	}

	// Pods copy config from the projected volume into an emptyDir once at
	// startup; a ConfigMap/keys change never reaches a running pod (rndc
	// reconfig re-reads the stale startup copy). Stamp a hash of the projected
	// config onto the pod template so a config change rolls the StatefulSet,
	// which is the only way the change takes effect. The operator owns the
	// template, so this restart is operator-driven and not reverted.
	sts := &appsv1.StatefulSet{
		ObjectMeta: metav1.ObjectMeta{Name: c.Name, Namespace: c.Namespace, Labels: labels},
		Spec: appsv1.StatefulSetSpec{
			ServiceName: headlessServiceName(c.Name),
			Replicas:    &replicas,
			Selector:    &metav1.LabelSelector{MatchLabels: labels},
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{Labels: labels, Annotations: map[string]string{configHashAnnotation: r.configHash(ctx, c)}},
				Spec: corev1.PodSpec{
					NodeSelector: c.Spec.NodeSelector,
					Tolerations:  c.Spec.Tolerations,
					Affinity:     c.Spec.Affinity,
					Containers: []corev1.Container{{
						Name:            bind.ContainerName,
						Image:           image,
						ImagePullPolicy: c.Spec.ImagePullPolicy,
						Command:         []string{"/bin/sh", bind.EntrypointPath},
						Ports: []corev1.ContainerPort{
							{Name: "dns-udp", ContainerPort: 53, Protocol: corev1.ProtocolUDP},
							{Name: "dns-tcp", ContainerPort: 53, Protocol: corev1.ProtocolTCP},
						},
						Resources: c.Spec.Resources,
						VolumeMounts: []corev1.VolumeMount{
							{Name: "bind-etc", MountPath: bind.ConfigDir, ReadOnly: true},
							{Name: "run", MountPath: bind.RunDir},
							{Name: "data", MountPath: bind.DataDir},
						},
						ReadinessProbe: &corev1.Probe{
							ProbeHandler:        corev1.ProbeHandler{TCPSocket: &corev1.TCPSocketAction{Port: intstrFromInt(53)}},
							InitialDelaySeconds: 5,
							PeriodSeconds:       10,
						},
						LivenessProbe: &corev1.Probe{
							ProbeHandler:        corev1.ProbeHandler{TCPSocket: &corev1.TCPSocketAction{Port: intstrFromInt(53)}},
							InitialDelaySeconds: 15,
							PeriodSeconds:       20,
						},
					}},
					Volumes: []corev1.Volume{
						projected,
						{Name: "run", VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{}}},
					},
				},
			},
			VolumeClaimTemplates: []corev1.PersistentVolumeClaim{{
				ObjectMeta: metav1.ObjectMeta{Name: "data"},
				Spec: corev1.PersistentVolumeClaimSpec{
					AccessModes:      []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
					StorageClassName: c.Spec.StorageClassName,
					Resources:        corev1.VolumeResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceStorage: qty}},
				},
			}},
		},
	}
	if err := ctrl.SetControllerReference(c, sts, r.Scheme); err != nil {
		return nil, err
	}

	var existing appsv1.StatefulSet
	err = r.Get(ctx, types.NamespacedName{Namespace: c.Namespace, Name: c.Name}, &existing)
	if apierrors.IsNotFound(err) {
		return sts, r.Create(ctx, sts)
	}
	if err != nil {
		return nil, err
	}
	// VolumeClaimTemplates are immutable; only mutate the mutable fields.
	existing.Spec.Replicas = sts.Spec.Replicas
	existing.Spec.Template = sts.Spec.Template
	if err := r.Update(ctx, &existing); err != nil {
		return nil, err
	}
	return &existing, nil
}

// configHashAnnotation carries a hash of the projected config on the pod
// template; changing it triggers a rolling restart so pods pick up new config.
const configHashAnnotation = "bind.unkin.net/config-hash"

// configHash returns a deterministic hash of the config projected into the pods
// (the rendered ConfigMap and the keys.conf Secret). It is read after those are
// reconciled, so it reflects the current desired config. A stable hash means no
// spurious restarts; any config or TSIG-key change flips it and rolls the pods.
func (r *BindClusterReconciler) configHash(ctx context.Context, c *bindv1alpha1.BindCluster) string {
	var buf bytes.Buffer
	var cm corev1.ConfigMap
	if err := r.Get(ctx, types.NamespacedName{Namespace: c.Namespace, Name: configMapName(c.Name)}, &cm); err == nil {
		for _, k := range sortedKeys(cm.Data) {
			fmt.Fprintf(&buf, "%s\x00%s\x00", k, cm.Data[k])
		}
	}
	var keys corev1.Secret
	if err := r.Get(ctx, types.NamespacedName{Namespace: c.Namespace, Name: keysSecretName(c.Name)}, &keys); err == nil {
		data := make(map[string]string, len(keys.Data))
		for k, v := range keys.Data {
			data[k] = string(v)
		}
		for _, k := range sortedKeys(data) {
			fmt.Fprintf(&buf, "%s\x00%s\x00", k, data[k])
		}
	}
	sum := sha256.Sum256(buf.Bytes())
	return hex.EncodeToString(sum[:])
}

func sortedKeys(m map[string]string) []string {
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	return keys
}

func (r *BindClusterReconciler) reloadReadyPods(ctx context.Context, c *bindv1alpha1.BindCluster) {
	if r.Exec == nil {
		return
	}
	logger := log.FromContext(ctx)
	var pods corev1.PodList
	if err := r.List(ctx, &pods, client.InNamespace(c.Namespace), client.MatchingLabels(commonLabels(c.Name))); err != nil {
		return
	}
	for i := range pods.Items {
		pod := &pods.Items[i]
		if !podReady(pod) {
			continue
		}
		if err := r.Exec.Reconfig(ctx, c.Namespace, pod.Name); err != nil {
			logger.V(1).Info("rndc reconfig failed", "pod", pod.Name, "err", err.Error())
		}
	}
}

func (r *BindClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
	mapToCluster := func(clusterRef, namespace string) []reconcile.Request {
		if clusterRef == "" {
			return nil
		}
		return []reconcile.Request{{NamespacedName: types.NamespacedName{Namespace: namespace, Name: clusterRef}}}
	}
	return ctrl.NewControllerManagedBy(mgr).
		For(&bindv1alpha1.BindCluster{}).
		Owns(&appsv1.StatefulSet{}).
		Owns(&corev1.Service{}).
		Owns(&corev1.ConfigMap{}).
		Owns(&corev1.Secret{}).
		Watches(&corev1.Pod{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, o client.Object) []reconcile.Request {
			// Re-render named.conf when a cluster pod's IP appears or changes, so
			// secondaries always point primaries/default-primaries at the current
			// primary pod IP.
			return mapToCluster(o.GetLabels()[clusterLabel], o.GetNamespace())
		})).
		Watches(&bindv1alpha1.BindACL{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, o client.Object) []reconcile.Request {
			return mapToCluster(o.(*bindv1alpha1.BindACL).Spec.ClusterRef, o.GetNamespace())
		})).
		Watches(&bindv1alpha1.BindView{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, o client.Object) []reconcile.Request {
			return mapToCluster(o.(*bindv1alpha1.BindView).Spec.ClusterRef, o.GetNamespace())
		})).
		Watches(&bindv1alpha1.BindPolicy{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, o client.Object) []reconcile.Request {
			return mapToCluster(o.(*bindv1alpha1.BindPolicy).Spec.ClusterRef, o.GetNamespace())
		})).
		Watches(&bindv1alpha1.BindDNSSECPolicy{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, o client.Object) []reconcile.Request {
			return mapToCluster(o.(*bindv1alpha1.BindDNSSECPolicy).Spec.ClusterRef, o.GetNamespace())
		})).
		Watches(&bindv1alpha1.BindZone{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, o client.Object) []reconcile.Request {
			// Only forward zones affect named.conf; primary/secondary zones are
			// managed dynamically by the BindZone controller.
			z := o.(*bindv1alpha1.BindZone)
			if z.Spec.Type != bindv1alpha1.ZoneForward {
				return nil
			}
			return mapToCluster(z.Spec.ClusterRef, o.GetNamespace())
		})).
		Watches(&bindv1alpha1.BindCatalogZone{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, o client.Object) []reconcile.Request {
			return mapToCluster(o.(*bindv1alpha1.BindCatalogZone).Spec.ClusterRef, o.GetNamespace())
		})).
		Watches(&bindv1alpha1.BindTSIGKey{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, o client.Object) []reconcile.Request {
			// TSIG keys are namespace-wide; re-render every cluster in the namespace.
			var clusters bindv1alpha1.BindClusterList
			if err := r.List(ctx, &clusters, client.InNamespace(o.GetNamespace())); err != nil {
				return nil
			}
			var reqs []reconcile.Request
			for _, cl := range clusters.Items {
				reqs = append(reqs, reconcile.Request{NamespacedName: types.NamespacedName{Namespace: cl.Namespace, Name: cl.Name}})
			}
			return reqs
		})).
		Complete(r)
}