controller.go

package controller

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"os"
	"strings"
	"sync"

	"github.com/authzed/controller-idioms/adopt"
	"github.com/authzed/controller-idioms/cachekeys"
	"github.com/authzed/controller-idioms/component"
	"github.com/authzed/controller-idioms/fileinformer"
	"github.com/authzed/controller-idioms/handler"
	"github.com/authzed/controller-idioms/hash"
	"github.com/authzed/controller-idioms/manager"
	"github.com/authzed/controller-idioms/middleware"
	"github.com/authzed/controller-idioms/typed"
	"github.com/cespare/xxhash/v2"
	"github.com/go-logr/logr"
	"go.uber.org/atomic"
	appsv1 "k8s.io/api/apps/v1"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	policyv1 "k8s.io/api/policy/v1"
	rbacv1 "k8s.io/api/rbac/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/yaml"
	applyappsv1 "k8s.io/client-go/applyconfigurations/apps/v1"
	applybatchv1 "k8s.io/client-go/applyconfigurations/batch/v1"
	applycorev1 "k8s.io/client-go/applyconfigurations/core/v1"
	applypolicyv1 "k8s.io/client-go/applyconfigurations/policy/v1"
	applyrbacv1 "k8s.io/client-go/applyconfigurations/rbac/v1"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/dynamic/dynamicinformer"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/record"
	_ "k8s.io/component-base/metrics/prometheus/workqueue" // for workqueue metric registration
	"k8s.io/klog/v2"
	"k8s.io/klog/v2/textlogger"
	"k8s.io/kubectl/pkg/util/openapi"

	"github.com/authzed/spicedb-operator/pkg/apis/authzed/v1alpha1"
	"github.com/authzed/spicedb-operator/pkg/config"
	"github.com/authzed/spicedb-operator/pkg/metadata"
)

// +kubebuilder:rbac:groups="authzed.com",resources=spicedbclusters,verbs=get;watch;list;create;update;patch;delete
// +kubebuilder:rbac:groups="authzed.com",resources=spicedbclusters/status,verbs=get;watch;list;create;update;patch;delete
// +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=batch,resources=jobs,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=policy,resources=poddisruptionbudgets,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch;delete
// +kubebuilder:rbac:groups="",resources=jobs,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=events,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=serviceaccounts,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=endpoints,verbs=get;list;watch
// +kubebuilder:rbac:groups=discovery.k8s.io,resources=endpointslices,verbs=get;list;watch
// +kubebuilder:rbac:groups="rbac.authorization.k8s.io",resources=roles,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="rbac.authorization.k8s.io",resources=rolebindings,verbs=get;list;watch;create;update;patch;delete

func init() {
	utilruntime.Must(v1alpha1.AddToScheme(scheme.Scheme))
}

var v1alpha1ClusterGVR = v1alpha1.SchemeGroupVersion.WithResource(v1alpha1.SpiceDBClusterResourceName)

func localClusterNamespace(ns string) string {
	if ns == "" {
		return "local"
	}
	return strings.Join([]string{"local", ns}, "/")
}

func OwnedFactoryKey(namespace string) typed.FactoryKey {
	return typed.NewFactoryKey(v1alpha1.SpiceDBClusterResourceName, localClusterNamespace(namespace), "unfiltered")
}

func DependentFactoryKey(namespace string) typed.FactoryKey {
	return typed.NewFactoryKey(v1alpha1.SpiceDBClusterResourceName, localClusterNamespace(namespace), "dependents")
}

type Controller struct {
	*manager.OwnedResourceController
	namespaces  []string
	client      dynamic.Interface
	kclient     kubernetes.Interface
	resources   openapi.Resources
	mainHandler handler.Handler

	// config
	configLock     sync.RWMutex
	config         config.OperatorConfig
	baseImage      string
	lastConfigHash atomic.Uint64
}

func NewController(ctx context.Context, registry *typed.Registry, dclient dynamic.Interface, kclient kubernetes.Interface, resources openapi.Resources, configFilePath, baseImage string, broadcaster record.EventBroadcaster, namespaces []string) (*Controller, error) {
	// If no namespaces are provided, watch all namespaces
	if len(namespaces) == 0 {
		namespaces = []string{metav1.NamespaceAll}
	}

	c := Controller{
		client:     dclient,
		kclient:    kclient,
		resources:  resources,
		namespaces: namespaces,
		baseImage:  baseImage,
	}
	c.OwnedResourceController = manager.NewOwnedResourceController(
		textlogger.NewLogger(textlogger.NewConfig()),
		v1alpha1.SpiceDBClusterResourceName,
		v1alpha1ClusterGVR,
		QueueOps,
		registry,
		broadcaster,
		c.syncOwnedResource,
	)

	fileInformerFactory, err := fileinformer.NewFileInformerFactory(textlogger.NewLogger(textlogger.NewConfig()))
	if err != nil {
		return nil, err
	}

	if len(configFilePath) > 0 {
		inf := fileInformerFactory.ForResource(fileinformer.FileGroupVersion.WithResource(configFilePath)).Informer()
		if _, err := inf.AddEventHandler(cache.ResourceEventHandlerFuncs{
			AddFunc:    func(_ any) { c.loadConfig(configFilePath) },
			UpdateFunc: func(_, _ any) { c.loadConfig(configFilePath) },
			DeleteFunc: func(_ any) { c.loadConfig(configFilePath) },
		}); err != nil {
			return nil, err
		}
	} else {
		logr.FromContextOrDiscard(ctx).V(3).Info("no operator configuration provided", "path", configFilePath)
	}

	ownedInformerFactories := make([]dynamicinformer.DynamicSharedInformerFactory, 0, len(namespaces))
	for _, ns := range namespaces {
		ownedInformerFactory := registry.MustNewFilteredDynamicSharedInformerFactory(
			OwnedFactoryKey(ns),
			dclient,
			0,
			ns,
			nil,
		)
		if _, err := ownedInformerFactory.ForResource(v1alpha1ClusterGVR).Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
			AddFunc:    func(obj any) { c.enqueue(v1alpha1ClusterGVR, obj) },
			UpdateFunc: func(_, obj any) { c.enqueue(v1alpha1ClusterGVR, obj) },
			// Delete is not used right now, we rely on ownerrefs to clean up
		}); err != nil {
			return nil, err
		}
		ownedInformerFactories = append(ownedInformerFactories, ownedInformerFactory)
	}

	externalInformerFactories := make([]dynamicinformer.DynamicSharedInformerFactory, 0, len(namespaces))
	for _, ns := range namespaces {
		externalInformerFactory := registry.MustNewFilteredDynamicSharedInformerFactory(
			DependentFactoryKey(ns),
			dclient,
			0,
			ns,
			func(options *metav1.ListOptions) {
				options.LabelSelector = metadata.ManagedDependentSelector.String()
			},
		)

		for _, gvr := range []schema.GroupVersionResource{
			appsv1.SchemeGroupVersion.WithResource("deployments"),
			corev1.SchemeGroupVersion.WithResource("serviceaccounts"),
			corev1.SchemeGroupVersion.WithResource("services"),
			corev1.SchemeGroupVersion.WithResource("pods"),
			batchv1.SchemeGroupVersion.WithResource("jobs"),
			rbacv1.SchemeGroupVersion.WithResource("roles"),
			rbacv1.SchemeGroupVersion.WithResource("rolebindings"),
			policyv1.SchemeGroupVersion.WithResource("poddisruptionbudgets"),
		} {
			inf := externalInformerFactory.ForResource(gvr).Informer()
			if err := inf.AddIndexers(cache.Indexers{metadata.OwningClusterIndex: metadata.GetClusterKeyFromMeta}); err != nil {
				return nil, err
			}
			if _, err := inf.AddEventHandler(cache.ResourceEventHandlerFuncs{
				AddFunc:    func(obj any) { c.syncExternalResource(obj) },
				UpdateFunc: func(_, obj any) { c.syncExternalResource(obj) },
				DeleteFunc: func(obj any) { c.syncExternalResource(obj) },
			}); err != nil {
				return nil, err
			}
		}
		// Secrets get per-type indexes (one per credential role) so that each
		// adoption handler only sees secrets of its own type and never
		// misidentifies another role's secret as stale. OwningClusterIndex is
		// kept for the legacy SecretRef path and for syncExternalResource, which
		// uses it to re-enqueue the owning cluster when any secret changes.
		secretsInf := externalInformerFactory.ForResource(corev1.SchemeGroupVersion.WithResource("secrets")).Informer()
		if err := secretsInf.AddIndexers(cache.Indexers{
			metadata.OwningClusterIndex:                 metadata.GetClusterKeyFromMeta,
			metadata.OwningClusterDatastoreURIIndex:     metadata.GetClusterKeyFromMetaForType(metadata.CredentialTypeDatastoreURI),
			metadata.OwningClusterPresharedKeyIndex:     metadata.GetClusterKeyFromMetaForType(metadata.CredentialTypePresharedKey),
			metadata.OwningClusterMigrationSecretsIndex: metadata.GetClusterKeyFromMetaForType(metadata.CredentialTypeMigrationSecrets),
		}); err != nil {
			return nil, err
		}
		if _, err := secretsInf.AddEventHandler(cache.ResourceEventHandlerFuncs{
			AddFunc:    func(obj any) { c.syncExternalResource(obj) },
			UpdateFunc: func(_, obj any) { c.syncExternalResource(obj) },
			DeleteFunc: func(obj any) { c.syncExternalResource(obj) },
		}); err != nil {
			return nil, err
		}
		externalInformerFactories = append(externalInformerFactories, externalInformerFactory)
	}

	// start informers
	for _, ownedInformerFactory := range ownedInformerFactories {
		ownedInformerFactory.Start(ctx.Done())
	}

	for _, externalInformerFactory := range externalInformerFactories {
		externalInformerFactory.Start(ctx.Done())
	}

	fileInformerFactory.Start(ctx.Done())

	// wait for caches to sync
	for _, ownedInformerFactory := range ownedInformerFactories {
		ownedInformerFactory.WaitForCacheSync(ctx.Done())
	}

	for _, externalInformerFactory := range externalInformerFactories {
		externalInformerFactory.WaitForCacheSync(ctx.Done())
	}
	fileInformerFactory.WaitForCacheSync(ctx.Done())

	// Build mainHandler handler
	mw := middleware.NewHandlerLoggingMiddleware(4)
	chain := middleware.ChainWithMiddleware(mw)
	parallel := middleware.ParallelWithMiddleware(mw)

	deploymentHandlerChain := chain(
		c.ensureDeployment,
		c.ensurePDB,
		c.cleanupJob,
	).Handler(HandlerDeploymentKey)

	waitForMigrationsChain := c.waitForMigrationsHandler(
		deploymentHandlerChain,
		c.selfPauseCluster(handler.NoopHandler),
	).WithID(HandlerWaitForMigrationsKey)

	c.mainHandler = chain(
		c.pauseCluster,
		c.validateSpec,
		c.secretAdopter,
		c.collectSecrets,
		c.checkConfigChanged,
		c.validateConfig,
		parallel(
			c.ensureServiceAccount,
			c.ensureRole,
			c.ensureService,
		),
		c.ensureRoleBinding,
		CtxDeployments.BoxBuilder("deploymentsPre"),
		CtxJobs.BoxBuilder("jobsPre"),
		parallel(
			c.getDeployments,
			c.getJobs,
		),
		c.checkMigrations(
			deploymentHandlerChain,
			chain(
				c.runMigration,
				waitForMigrationsChain.Builder(),
			).Handler(HandlerMigrationRunKey),
			waitForMigrationsChain,
		).Builder(),
	).Handler("controller")

	return &c, nil
}

func (c *Controller) loadConfig(path string) {
	if len(path) == 0 {
		return
	}

	logger := textlogger.NewLogger(textlogger.NewConfig())
	logger.V(3).Info("loading config", "path", path)

	file, err := os.Open(path)
	if err != nil {
		panic(err)
	}
	defer func() {
		utilruntime.HandleError(file.Close())
	}()
	contents, err := io.ReadAll(file)
	if err != nil {
		panic(err)
	}

	decoder := yaml.NewYAMLOrJSONDecoder(bytes.NewReader(contents), 100)
	var cfg config.OperatorConfig
	if err := decoder.Decode(&cfg); err != nil {
		panic(err)
	}

	if h := xxhash.Sum64(contents); h != c.lastConfigHash.Load() {
		func() {
			c.configLock.Lock()
			defer c.configLock.Unlock()
			c.config = cfg
			// Override ImageName with flag if provided
			if c.baseImage != "" {
				logger.V(3).Info("overriding graph-defined base image with startup flag", "baseImage", c.baseImage)
				c.config.ImageName = c.baseImage
			}
		}()
		c.lastConfigHash.Store(h)
	} else {
		// config hasn't changed
		logger.V(4).Info("config hasn't changed", "old hash", c.lastConfigHash.Load(), "new hash", h)
		return
	}

	logger.V(3).Info("updated config", "path", path, "config", c.config)

	// requeue all clusters
	for _, ns := range c.namespaces {
		lister := typed.MustListerForKey[*v1alpha1.SpiceDBCluster](c.Registry, typed.NewRegistryKey(OwnedFactoryKey(ns), v1alpha1ClusterGVR))
		clusters, err := lister.List(labels.Everything())
		if err != nil {
			utilruntime.HandleError(err)
			return
		}
		for _, cluster := range clusters {
			c.enqueue(v1alpha1ClusterGVR, cluster)
		}
	}
}

func (c *Controller) enqueue(gvr schema.GroupVersionResource, obj interface{}) {
	key, err := cachekeys.GVRMetaNamespaceKeyFunc(gvr, obj)
	if err != nil {
		utilruntime.HandleError(err)
		return
	}
	c.Queue.AddRateLimited(key)
}

// syncOwnedResource is called when SpiceDBCluster is updated
func (c *Controller) syncOwnedResource(ctx context.Context, gvr schema.GroupVersionResource, namespace, name string) {
	// configure watch namespace; used to lookup the right caches to use
	if len(c.namespaces) == 1 && c.namespaces[0] == corev1.NamespaceAll {
		ctx = CtxCacheNamespace.WithValue(ctx, corev1.NamespaceAll)
	} else {
		ctx = CtxCacheNamespace.WithValue(ctx, namespace)
	}

	cluster, err := typed.MustListerForKey[*v1alpha1.SpiceDBCluster](c.Registry, typed.NewRegistryKey(OwnedFactoryKey(CtxCacheNamespace.Value(ctx)), v1alpha1ClusterGVR)).ByNamespace(namespace).Get(name)
	if err != nil {
		utilruntime.HandleError(fmt.Errorf("syncOwnedResource called on unknown object (%s::%s/%s): %w", gvr.String(), namespace, name, err))
		QueueOps.Done(ctx)
		return
	}

	logger := textlogger.NewLogger(textlogger.NewConfig()).WithValues(
		"syncID", middleware.NewSyncID(5),
		"controller", c.Name(),
		"obj", klog.KObj(cluster).MarshalLog(),
	)
	ctx = logr.NewContext(ctx, logger)

	ctx = CtxCluster.WithValue(ctx, cluster.DeepCopy())
	ctx = CtxClusterNN.WithValue(ctx, cluster.NamespacedName())

	// Derive the primary secret name from SecretRef or Credentials (first non-skipped ref).
	secretNames := credentialSecretNames(cluster.DeepCopy())
	primarySecretName := ""
	if len(secretNames) > 0 {
		primarySecretName = secretNames[0]
	}
	ctx = CtxSecretNN.WithValue(ctx, types.NamespacedName{
		Name:      primarySecretName,
		Namespace: cluster.Namespace,
	})
	ctx = CtxSecretNames.WithValue(ctx, secretNames)

	c.configLock.RLock()
	cfg := c.config.Copy()
	ctx = CtxOperatorConfig.WithValue(ctx, &cfg)
	c.configLock.RUnlock()

	logger.V(4).Info("syncing owned object", "gvr", gvr)

	c.Handle(ctx)
}

// credentialSecretNames returns the unique list of secret names required by the cluster's credentials.
// If Credentials is nil and SecretRef is set, returns the single SecretRef name (promotion path).
func credentialSecretNames(cluster *v1alpha1.SpiceDBCluster) []string {
	creds := cluster.Spec.Credentials
	if creds == nil && cluster.Spec.SecretRef != "" {
		return []string{cluster.Spec.SecretRef}
	}
	if creds == nil {
		return nil
	}
	seen := map[string]struct{}{}
	var names []string
	for _, ref := range []*v1alpha1.CredentialRef{creds.DatastoreURI, creds.PresharedKey, creds.MigrationSecrets} {
		if ref != nil && !ref.Skip && ref.SecretName != "" {
			if _, ok := seen[ref.SecretName]; !ok {
				seen[ref.SecretName] = struct{}{}
				names = append(names, ref.SecretName)
			}
		}
	}
	return names
}

type credentialSecret struct {
	name     string
	credType string
}

// credentialSecretsForCluster returns one entry per non-skipped credential ref,
// preserving the credential type for each. Unlike credentialSecretNames, the
// same name may appear more than once when a secret is shared across multiple
// roles — each role runs its own adoption handler with its own type label.
func credentialSecretsForCluster(cluster *v1alpha1.SpiceDBCluster) []credentialSecret {
	creds := cluster.Spec.Credentials
	if creds == nil && cluster.Spec.SecretRef != "" {
		return []credentialSecret{{name: cluster.Spec.SecretRef, credType: ""}}
	}
	if creds == nil {
		return nil
	}
	roleSecrets := []struct {
		ref      *v1alpha1.CredentialRef
		credType string
	}{
		{creds.DatastoreURI, metadata.CredentialTypeDatastoreURI},
		{creds.PresharedKey, metadata.CredentialTypePresharedKey},
		{creds.MigrationSecrets, metadata.CredentialTypeMigrationSecrets},
	}
	var result []credentialSecret
	for _, rs := range roleSecrets {
		if rs.ref != nil && !rs.ref.Skip && rs.ref.SecretName != "" {
			result = append(result, credentialSecret{name: rs.ref.SecretName, credType: rs.credType})
		}
	}
	return result
}

// syncExternalResource is called when a dependent resource is updated:
// It queues the owning SpiceDBCluster for reconciliation based on the labels.
// No other reconciliation should take place here; we keep a single state
// machine for SpiceDBCluster with an entrypoint in the mainHandler Handler
func (c *Controller) syncExternalResource(obj interface{}) {
	objMeta, err := meta.Accessor(obj)
	if err != nil {
		utilruntime.HandleError(err)
		return
	}

	logger := textlogger.NewLogger(textlogger.NewConfig()).WithValues(
		"syncID", middleware.NewSyncID(5),
		"controller", c.Name(),
		"obj", klog.KObj(objMeta),
	)
	logger.V(4).Info("syncing external object")

	keys, err := adopt.OwnerKeysFromMeta(metadata.OwnerAnnotationKeyPrefix)(obj)
	if err != nil {
		utilruntime.HandleError(err)
		return
	}

	for _, k := range keys {
		c.Queue.AddRateLimited(cachekeys.GVRMetaNamespaceKeyer(v1alpha1ClusterGVR, k))
	}
}

// Handle inspects the current SpiceDBCluster object and ensures
// the desired state is persisted on the cluster.
func (c *Controller) Handle(ctx context.Context) {
	c.mainHandler.Handle(ctx)
}

func (c *Controller) ensureDeployment(next ...handler.Handler) handler.Handler {
	return handler.NewTypeHandler(&DeploymentHandler{
		applyDeployment: func(ctx context.Context, dep *applyappsv1.DeploymentApplyConfiguration) (*appsv1.Deployment, error) {
			logr.FromContextOrDiscard(ctx).V(4).Info("updating deployment", "namespace", *dep.Namespace, "name", *dep.Name)
			return c.kclient.AppsV1().Deployments(*dep.Namespace).Apply(ctx, dep, metadata.ApplyForceOwned)
		},
		deleteDeployment: func(ctx context.Context, nn types.NamespacedName) error {
			logr.FromContextOrDiscard(ctx).V(4).Info("deleting deployment", "namespace", nn.Namespace, "name", nn.Name)
			return c.kclient.AppsV1().Deployments(nn.Namespace).Delete(ctx, nn.Name, metav1.DeleteOptions{})
		},
		getDeploymentPods: func(ctx context.Context) []*corev1.Pod {
			return component.NewIndexedComponent(
				typed.MustIndexerForKey[*corev1.Pod](c.Registry, typed.NewRegistryKey(DependentFactoryKey(CtxCacheNamespace.Value(ctx)), corev1.SchemeGroupVersion.WithResource("pods"))),
				metadata.OwningClusterIndex,
				func(ctx context.Context) labels.Selector {
					return metadata.SelectorForComponent(CtxClusterNN.MustValue(ctx).Name, metadata.ComponentSpiceDBLabelValue)
				},
			).List(ctx, CtxClusterNN.MustValue(ctx))
		},
		patchStatus: c.PatchStatus,
		next:        handler.Handlers(next).MustOne(),
	})
}

func (c *Controller) ensurePDB(next ...handler.Handler) handler.Handler {
	applyPDB := func(ctx context.Context, apply *applypolicyv1.PodDisruptionBudgetApplyConfiguration) (*policyv1.PodDisruptionBudget, error) {
		logr.FromContextOrDiscard(ctx).V(4).Info("applying pdb", "namespace", *apply.Namespace, "name", *apply.Name)
		return c.kclient.PolicyV1().PodDisruptionBudgets(*apply.Namespace).Apply(ctx, apply, metadata.ApplyForceOwned)
	}
	deletePDB := func(ctx context.Context, nn types.NamespacedName) error {
		logr.FromContextOrDiscard(ctx).V(4).Info("deleting pdb", "namespace", nn.Namespace, "name", nn.Name)
		return c.kclient.PolicyV1().PodDisruptionBudgets(nn.Namespace).Delete(ctx, nn.Name, metav1.DeleteOptions{})
	}
	return handler.NewHandlerFromFunc(func(ctx context.Context) {
		// look up the index to find PDBs owned by the cluster
		ownedPDBIndexer := typed.MustIndexerForKey[*policyv1.PodDisruptionBudget](
			c.Registry,
			typed.NewRegistryKey(
				DependentFactoryKey(CtxCacheNamespace.Value(ctx)),
				policyv1.SchemeGroupVersion.WithResource("poddisruptionbudgets"),
			),
		)

		// define the PDB "component" - the object that we want to ensure exists
		// and is owned by the cluster
		pdbComponent := component.NewIndexedComponent(
			ownedPDBIndexer,
			metadata.OwningClusterIndex,
			func(ctx context.Context) labels.Selector {
				return metadata.SelectorForComponent(CtxClusterNN.MustValue(ctx).Name, metadata.ComponentPDBLabel)
			},
		)

		// build the handler that ensures the PDB with the proper definition
		// exists and run it
		cfg := CtxConfig.MustValue(ctx)
		component.NewEnsureComponentByHash(
			component.NewHashableComponent(pdbComponent, hash.NewObjectHash(), "authzed.com/controller-component-hash"),
			CtxClusterNN,
			QueueOps,
			applyPDB,
			deletePDB,
			func(_ context.Context) *applypolicyv1.PodDisruptionBudgetApplyConfiguration {
				return cfg.PodDisruptionBudget()
			},
		).Handle(ctx)

		// if the context is canceled, the sub-handler wants to stop processing
		if errors.Is(ctx.Err(), context.Canceled) {
			return
		}

		handler.Handlers(next).MustOne().Handle(ctx)
	}, "ensurePDB")
}

func (c *Controller) cleanupJob(...handler.Handler) handler.Handler {
	return handler.NewTypeHandler(&JobCleanupHandler{
		registry: c.Registry,
		getJobs: func(ctx context.Context) []*batchv1.Job {
			return component.NewIndexedComponent(
				typed.MustIndexerForKey[*batchv1.Job](c.Registry, typed.NewRegistryKey(DependentFactoryKey(CtxCacheNamespace.Value(ctx)), batchv1.SchemeGroupVersion.WithResource("jobs"))),
				metadata.OwningClusterIndex,
				func(ctx context.Context) labels.Selector {
					return metadata.SelectorForComponent(CtxClusterNN.MustValue(ctx).Name, metadata.ComponentMigrationJobLabelValue)
				}).List(ctx, CtxClusterNN.MustValue(ctx))
		},
		getJobPods: func(ctx context.Context) []*corev1.Pod {
			return component.NewIndexedComponent(
				typed.MustIndexerForKey[*corev1.Pod](c.Registry, typed.NewRegistryKey(DependentFactoryKey(CtxCacheNamespace.Value(ctx)), corev1.SchemeGroupVersion.WithResource("pods"))),
				metadata.OwningClusterIndex,
				func(ctx context.Context) labels.Selector {
					return metadata.SelectorForComponent(CtxClusterNN.MustValue(ctx).Name, metadata.ComponentMigrationJobLabelValue)
				},
			).List(ctx, CtxClusterNN.MustValue(ctx))
		},
		deleteJob: func(ctx context.Context, nn types.NamespacedName) error {
			logr.FromContextOrDiscard(ctx).V(4).Info("deleting job", "namespace", nn.Namespace, "name", nn.Name)
			backgroundPolicy := metav1.DeletePropagationBackground
			return c.kclient.BatchV1().Jobs(nn.Namespace).Delete(ctx, nn.Name, metav1.DeleteOptions{PropagationPolicy: &backgroundPolicy})
		},
		deletePod: func(ctx context.Context, nn types.NamespacedName) error {
			logr.FromContextOrDiscard(ctx).V(4).Info("deleting job pod", "namespace", nn.Namespace, "name", nn.Name)
			return c.kclient.CoreV1().Pods(nn.Namespace).Delete(ctx, nn.Name, metav1.DeleteOptions{})
		},
	})
}

func (c *Controller) waitForMigrationsHandler(next ...handler.Handler) handler.Handler {
	return handler.NewTypeHandler(&WaitForMigrationsHandler{
		recorder:              c.Recorder,
		nextSelfPause:         HandlerSelfPauseKey.MustFind(next),
		nextDeploymentHandler: HandlerDeploymentKey.MustFind(next),
	})
}

func (c *Controller) pauseCluster(next ...handler.Handler) handler.Handler {
	return NewPauseHandler(c.PatchStatus, handler.Handlers(next).MustOne())
}

func (c *Controller) selfPauseCluster(...handler.Handler) handler.Handler {
	return NewSelfPauseHandler(c.Patch, c.PatchStatus)
}

func (c *Controller) secretAdopter(next ...handler.Handler) handler.Handler {
	secretsGVR := corev1.SchemeGroupVersion.WithResource("secrets")
	return handler.NewHandlerFromFunc(func(ctx context.Context) {
		cluster := CtxCluster.MustValue(ctx)
		pairs := credentialSecretsForCluster(cluster)
		if len(pairs) == 0 {
			handler.Handlers(next).MustOne().Handle(ctx)
			return
		}
		for _, cs := range pairs {
			credType := cs.credType
			secretCtx := CtxSecretNN.WithValue(ctx, types.NamespacedName{
				Name:      cs.name,
				Namespace: cluster.Namespace,
			})
			NewSecretAdoptionHandler(
				c.Recorder,
				func(ctx context.Context) (*corev1.Secret, error) {
					secret, err := typed.MustListerForKey[*corev1.Secret](c.Registry, typed.NewRegistryKey(DependentFactoryKey(CtxCacheNamespace.Value(ctx)), secretsGVR)).ByNamespace(CtxSecretNN.MustValue(ctx).Namespace).Get(CtxSecretNN.MustValue(ctx).Name)
					if err != nil {
						return nil, err
					}
					// If the secret lacks this role's per-type label key, return
					// NotFound so the adoption handler applies the full label set.
					// This migrates secrets from operator versions that did not
					// set per-role labels.
					if lk := metadata.LabelKeyForCredentialType(credType); lk != "" {
						if _, ok := secret.Labels[lk]; !ok {
							return nil, apierrors.NewNotFound(corev1.SchemeGroupVersion.WithResource("secrets").GroupResource(), CtxSecretNN.MustValue(ctx).Name)
						}
					}
					return secret, nil
				},
				func(ctx context.Context, err error) {
					cluster := CtxCluster.MustValue(ctx)
					status := &v1alpha1.SpiceDBCluster{
						TypeMeta: metav1.TypeMeta{
							Kind:       v1alpha1.SpiceDBClusterKind,
							APIVersion: v1alpha1.SchemeGroupVersion.String(),
						},
						ObjectMeta: metav1.ObjectMeta{Namespace: cluster.Namespace, Name: cluster.Name},
						Status:     *cluster.Status.DeepCopy(),
					}
					status.Status.ObservedGeneration = cluster.GetGeneration()
					status.SetStatusCondition(v1alpha1.NewMissingSecretCondition(CtxSecretNN.MustValue(ctx)))
					if err := c.PatchStatus(ctx, status); err != nil {
						QueueOps.RequeueAPIErr(ctx, err)
						return
					}
					QueueOps.RequeueErr(ctx, err)
				},
				typed.MustIndexerForKey[*corev1.Secret](c.Registry, typed.NewRegistryKey(DependentFactoryKey(CtxCacheNamespace.Value(ctx)), secretsGVR)),
				func(ctx context.Context, secret *applycorev1.SecretApplyConfiguration, options metav1.ApplyOptions) (*corev1.Secret, error) {
					return c.kclient.CoreV1().Secrets(*secret.Namespace).Apply(ctx, secret, options)
				},
				func(ctx context.Context, nn types.NamespacedName) error {
					_, err := c.kclient.CoreV1().Secrets(nn.Namespace).Get(ctx, nn.Name, metav1.GetOptions{})
					return err
				},
				credType,
				handler.NoopHandler,
			).Handle(secretCtx)
			if errors.Is(secretCtx.Err(), context.Canceled) {
				return
			}
		}
		// Call next with the original ctx; collectSecrets fetches all secrets
		// from cache into CtxSecrets using CtxSecretNames.
		handler.Handlers(next).MustOne().Handle(ctx)
	}, "adoptSecrets")
}

// collectSecrets looks up all credential secrets from the cache and populates
// CtxSecrets. This runs after secretAdopter.
func (c *Controller) collectSecrets(next ...handler.Handler) handler.Handler {
	secretsGVR := corev1.SchemeGroupVersion.WithResource("secrets")
	return handler.NewHandlerFromFunc(func(ctx context.Context) {
		cluster := CtxCluster.MustValue(ctx)
		names := CtxSecretNames.Value(ctx)
		if names == nil {
			names = credentialSecretNames(cluster)
		}
		secretsMap := make(map[string]*corev1.Secret, len(names))
		lister := typed.MustListerForKey[*corev1.Secret](c.Registry, typed.NewRegistryKey(DependentFactoryKey(CtxCacheNamespace.Value(ctx)), secretsGVR)).ByNamespace(cluster.Namespace)
		for _, name := range names {
			sec, err := lister.Get(name)
			if err == nil && sec != nil {
				secretsMap[name] = sec
			}
		}
		ctx = CtxSecrets.WithValue(ctx, secretsMap)
		handler.Handlers(next).MustOne().Handle(ctx)
	}, "collectSecrets")
}

func (c *Controller) checkConfigChanged(next ...handler.Handler) handler.Handler {
	return handler.NewTypeHandler(&ConfigChangedHandler{
		patchStatus: c.PatchStatus,
		next:        handler.Handlers(next).MustOne(),
	})
}

func (c *Controller) validateSpec(next ...handler.Handler) handler.Handler {
	return handler.NewTypeHandler(&ValidateSpecHandler{
		patchStatus: c.PatchStatus,
		recorder:    c.Recorder,
		next:        handler.Handlers(next).MustOne(),
	})
}

func (c *Controller) validateConfig(next ...handler.Handler) handler.Handler {
	return handler.NewTypeHandler(&ValidateConfigHandler{
		patchStatus: c.PatchStatus,
		recorder:    c.Recorder,
		resources:   c.resources,
		next:        handler.Handlers(next).MustOne(),
	})
}

func (c *Controller) getDeployments(next ...handler.Handler) handler.Handler {
	return handler.NewHandlerFromFunc(func(ctx context.Context) {
		component.NewComponentContextHandler[*appsv1.Deployment](
			CtxDeployments,
			component.NewIndexedComponent(
				typed.MustIndexerForKey[*appsv1.Deployment](c.Registry, typed.NewRegistryKey(DependentFactoryKey(CtxCacheNamespace.Value(ctx)), appsv1.SchemeGroupVersion.WithResource("deployments"))),
				metadata.OwningClusterIndex,
				func(ctx context.Context) labels.Selector {
					return metadata.SelectorForComponent(CtxClusterNN.MustValue(ctx).Name, metadata.ComponentSpiceDBLabelValue)
				}),
			CtxClusterNN,
			handler.NoopHandler,
		).Handle(ctx)
		if errors.Is(ctx.Err(), context.Canceled) {
			return
		}
		handler.Handlers(next).MustOne().Handle(ctx)
	}, "getDeployments")
}

func (c *Controller) getJobs(next ...handler.Handler) handler.Handler {
	return handler.NewHandlerFromFunc(func(ctx context.Context) {
		component.NewComponentContextHandler[*batchv1.Job](
			CtxJobs,
			component.NewIndexedComponent(
				typed.MustIndexerForKey[*batchv1.Job](c.Registry, typed.NewRegistryKey(DependentFactoryKey(CtxCacheNamespace.Value(ctx)), batchv1.SchemeGroupVersion.WithResource("jobs"))),
				metadata.OwningClusterIndex,
				func(ctx context.Context) labels.Selector {
					return metadata.SelectorForComponent(CtxClusterNN.MustValue(ctx).Name, metadata.ComponentMigrationJobLabelValue)
				}),
			CtxClusterNN,
			handler.NoopHandler,
		).Handle(ctx)
		if errors.Is(ctx.Err(), context.Canceled) {
			return
		}
		handler.Handlers(next).MustOne().Handle(ctx)
	}, "getJobs")
}

func (c *Controller) runMigration(next ...handler.Handler) handler.Handler {
	return handler.NewTypeHandler(&MigrationRunHandler{
		applyJob: func(ctx context.Context, job *applybatchv1.JobApplyConfiguration) error {
			_, err := c.kclient.BatchV1().Jobs(*job.Namespace).Apply(ctx, job, metadata.ApplyForceOwned)
			return err
		},
		deleteJob: func(ctx context.Context, nn types.NamespacedName) error {
			return c.kclient.BatchV1().Jobs(nn.Namespace).Delete(ctx, nn.Name, metav1.DeleteOptions{})
		},
		patchStatus: c.PatchStatus,
		next:        handler.Handlers(next).MustOne(),
	})
}

func (c *Controller) checkMigrations(next ...handler.Handler) handler.Handler {
	return handler.NewTypeHandler(&MigrationCheckHandler{
		recorder:                c.Recorder,
		nextMigrationRunHandler: HandlerMigrationRunKey.MustFind(next),
		nextWaitForJobHandler:   HandlerWaitForMigrationsKey.MustFind(next),
		nextDeploymentHandler:   HandlerDeploymentKey.MustFind(next),
	})
}

func (c *Controller) ensureServiceAccount(next ...handler.Handler) handler.Handler {
	return handler.NewHandlerFromFunc(func(ctx context.Context) {
		component.NewEnsureComponentByHash(
			component.NewHashableComponent(
				component.NewIndexedComponent(
					typed.MustIndexerForKey[*corev1.ServiceAccount](
						c.Registry,
						typed.NewRegistryKey(
							DependentFactoryKey(CtxCacheNamespace.Value(ctx)),
							corev1.SchemeGroupVersion.WithResource("serviceaccounts"),
						)),
					metadata.OwningClusterIndex,
					func(ctx context.Context) labels.Selector {
						return metadata.SelectorForComponent(CtxClusterNN.MustValue(ctx).Name, metadata.ComponentServiceAccountLabel)
					}),
				hash.NewObjectHash(), "authzed.com/controller-component-hash"),
			CtxClusterNN,
			QueueOps,
			func(ctx context.Context, apply *applycorev1.ServiceAccountApplyConfiguration) (*corev1.ServiceAccount, error) {
				logr.FromContextOrDiscard(ctx).V(4).Info("applying serviceaccount", "namespace", *apply.Namespace, "name", *apply.Name)
				return c.kclient.CoreV1().ServiceAccounts(*apply.Namespace).Apply(ctx, apply, metadata.ApplyForceOwned)
			},
			func(ctx context.Context, nn types.NamespacedName) error {
				logr.FromContextOrDiscard(ctx).V(4).Info("deleting serviceaccount", "namespace", nn.Namespace, "name", nn.Name)
				return c.kclient.CoreV1().ServiceAccounts(nn.Namespace).Delete(ctx, nn.Name, metav1.DeleteOptions{})
			},
			func(ctx context.Context) *applycorev1.ServiceAccountApplyConfiguration {
				return CtxConfig.MustValue(ctx).ServiceAccount()
			}).Handle(ctx)
		if errors.Is(ctx.Err(), context.Canceled) {
			return
		}
		handler.Handlers(next).MustOne().Handle(ctx)
	}, "ensureServiceAccount")
}

func (c *Controller) ensureRole(next ...handler.Handler) handler.Handler {
	return handler.NewHandlerFromFunc(func(ctx context.Context) {
		component.NewEnsureComponentByHash(
			component.NewHashableComponent(
				component.NewIndexedComponent(
					typed.MustIndexerForKey[*rbacv1.Role](
						c.Registry,
						typed.NewRegistryKey(
							DependentFactoryKey(CtxCacheNamespace.Value(ctx)),
							rbacv1.SchemeGroupVersion.WithResource("roles"),
						)),
					metadata.OwningClusterIndex,
					func(ctx context.Context) labels.Selector {
						return metadata.SelectorForComponent(CtxClusterNN.MustValue(ctx).Name, metadata.ComponentRoleLabel)
					}),
				hash.NewObjectHash(), "authzed.com/controller-component-hash"),
			CtxClusterNN,
			QueueOps,
			func(ctx context.Context, apply *applyrbacv1.RoleApplyConfiguration) (*rbacv1.Role, error) {
				logr.FromContextOrDiscard(ctx).V(4).Info("applying role", "namespace", *apply.Namespace, "name", *apply.Name)
				return c.kclient.RbacV1().Roles(*apply.Namespace).Apply(ctx, apply, metadata.ApplyForceOwned)
			},
			func(ctx context.Context, nn types.NamespacedName) error {
				logr.FromContextOrDiscard(ctx).V(4).Info("deleting role", "namespace", nn.Namespace, "name", nn.Name)
				return c.kclient.RbacV1().Roles(nn.Namespace).Delete(ctx, nn.Name, metav1.DeleteOptions{})
			},
			func(ctx context.Context) *applyrbacv1.RoleApplyConfiguration {
				return CtxConfig.MustValue(ctx).Role()
			}).Handle(ctx)
		if errors.Is(ctx.Err(), context.Canceled) {
			return
		}
		handler.Handlers(next).MustOne().Handle(ctx)
	}, "ensureRole")
}

func (c *Controller) ensureRoleBinding(next ...handler.Handler) handler.Handler {
	return handler.NewHandlerFromFunc(func(ctx context.Context) {
		component.NewEnsureComponentByHash(
			component.NewHashableComponent(
				component.NewIndexedComponent(
					typed.MustIndexerForKey[*rbacv1.RoleBinding](
						c.Registry,
						typed.NewRegistryKey(
							DependentFactoryKey(CtxCacheNamespace.Value(ctx)),
							rbacv1.SchemeGroupVersion.WithResource("rolebindings"),
						)),
					metadata.OwningClusterIndex,
					func(ctx context.Context) labels.Selector {
						return metadata.SelectorForComponent(CtxClusterNN.MustValue(ctx).Name, metadata.ComponentRoleBindingLabel)
					}),
				hash.NewObjectHash(), "authzed.com/controller-component-hash"),
			CtxClusterNN,
			QueueOps,
			func(ctx context.Context, apply *applyrbacv1.RoleBindingApplyConfiguration) (*rbacv1.RoleBinding, error) {
				logr.FromContextOrDiscard(ctx).V(4).Info("applying rolebinding", "namespace", *apply.Namespace, "name", *apply.Name)
				return c.kclient.RbacV1().RoleBindings(*apply.Namespace).Apply(ctx, apply, metadata.ApplyForceOwned)
			},
			func(ctx context.Context, nn types.NamespacedName) error {
				logr.FromContextOrDiscard(ctx).V(4).Info("deleting rolebinding", "namespace", nn.Namespace, "name", nn.Name)
				return c.kclient.RbacV1().RoleBindings(nn.Namespace).Delete(ctx, nn.Name, metav1.DeleteOptions{})
			},
			func(ctx context.Context) *applyrbacv1.RoleBindingApplyConfiguration {
				return CtxConfig.MustValue(ctx).RoleBinding()
			},
		).Handle(ctx)
		if errors.Is(ctx.Err(), context.Canceled) {
			return
		}
		handler.Handlers(next).MustOne().Handle(ctx)
	}, "ensureRoleBinding")
}

func (c *Controller) ensureService(next ...handler.Handler) handler.Handler {
	return handler.NewHandlerFromFunc(func(ctx context.Context) {
		component.NewEnsureComponentByHash(
			component.NewHashableComponent(
				component.NewIndexedComponent(
					typed.MustIndexerForKey[*corev1.Service](
						c.Registry,
						typed.NewRegistryKey(
							DependentFactoryKey(CtxCacheNamespace.Value(ctx)),
							corev1.SchemeGroupVersion.WithResource("services"),
						)),
					metadata.OwningClusterIndex,
					func(ctx context.Context) labels.Selector {
						return metadata.SelectorForComponent(CtxClusterNN.MustValue(ctx).Name, metadata.ComponentServiceLabel)
					}),
				hash.NewObjectHash(), "authzed.com/controller-component-hash"),
			CtxClusterNN,
			QueueOps,
			func(ctx context.Context, apply *applycorev1.ServiceApplyConfiguration) (*corev1.Service, error) {
				logr.FromContextOrDiscard(ctx).V(4).Info("applying service", "namespace", *apply.Namespace, "name", *apply.Name)
				return c.kclient.CoreV1().Services(*apply.Namespace).Apply(ctx, apply, metadata.ApplyForceOwned)
			},
			func(ctx context.Context, nn types.NamespacedName) error {
				logr.FromContextOrDiscard(ctx).V(4).Info("deleting service", "namespace", nn.Namespace, "name", nn.Name)
				return c.kclient.CoreV1().Services(nn.Namespace).Delete(ctx, nn.Name, metav1.DeleteOptions{})
			},
			func(ctx context.Context) *applycorev1.ServiceApplyConfiguration {
				return CtxConfig.MustValue(ctx).Service()
			}).Handle(ctx)
		if errors.Is(ctx.Err(), context.Canceled) {
			return
		}
		handler.Handlers(next).MustOne().Handle(ctx)
	}, "ensureService")
}