clusterextensionrevision_controller.go

//go:build !standard

package controllers

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"k8s.io/apimachinery/pkg/api/equality"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/utils/clock"
	"pkg.package-operator.run/boxcutter"
	"pkg.package-operator.run/boxcutter/machinery"
	machinerytypes "pkg.package-operator.run/boxcutter/machinery/types"
	"pkg.package-operator.run/boxcutter/ownerhandling"
	"pkg.package-operator.run/boxcutter/probing"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/event"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/source"

	ocv1 "github.com/operator-framework/operator-controller/api/v1"
	"github.com/operator-framework/operator-controller/internal/operator-controller/applier"
	"github.com/operator-framework/operator-controller/internal/operator-controller/labels"
)

const (
	clusterExtensionRevisionTeardownFinalizer = "olm.operatorframework.io/teardown"
)

// ClusterExtensionRevisionReconciler actions individual snapshots of ClusterExtensions,
// as part of the boxcutter integration.
type ClusterExtensionRevisionReconciler struct {
	Client                client.Client
	RevisionEngineFactory RevisionEngineFactory
	TrackingCache         trackingCache
	Clock                 clock.Clock
}

type trackingCache interface {
	client.Reader
	Source(handler handler.EventHandler, predicates ...predicate.Predicate) source.Source
	Watch(ctx context.Context, user client.Object, gvks sets.Set[schema.GroupVersionKind]) error
	Free(ctx context.Context, user client.Object) error
}

//+kubebuilder:rbac:groups=olm.operatorframework.io,resources=clusterextensionrevisions,verbs=get;list;watch;update;patch;create;delete
//+kubebuilder:rbac:groups=olm.operatorframework.io,resources=clusterextensionrevisions/status,verbs=update;patch
//+kubebuilder:rbac:groups=olm.operatorframework.io,resources=clusterextensionrevisions/finalizers,verbs=update

func (c *ClusterExtensionRevisionReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	l := log.FromContext(ctx).WithName("cluster-extension-revision")
	ctx = log.IntoContext(ctx, l)

	existingRev := &ocv1.ClusterExtensionRevision{}
	if err := c.Client.Get(ctx, req.NamespacedName, existingRev); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	l.Info("reconcile starting")
	defer l.Info("reconcile ending")

	reconciledRev := existingRev.DeepCopy()
	res, reconcileErr := c.reconcile(ctx, reconciledRev)

	if pd := existingRev.Spec.ProgressDeadlineMinutes; pd > 0 {
		cnd := meta.FindStatusCondition(reconciledRev.Status.Conditions, ocv1.ClusterExtensionRevisionTypeProgressing)
		isStillProgressing := cnd != nil && cnd.Status == metav1.ConditionTrue && cnd.Reason != ocv1.ReasonSucceeded
		succeeded := meta.IsStatusConditionTrue(reconciledRev.Status.Conditions, ocv1.ClusterExtensionRevisionTypeSucceeded)
		// check if we reached the progress deadline only if the revision is still progressing and has not succeeded yet
		if isStillProgressing && !succeeded {
			timeout := time.Duration(pd) * time.Minute
			if c.Clock.Since(existingRev.CreationTimestamp.Time) > timeout {
				// progress deadline reached, reset any errors and stop reconciling this revision
				markAsNotProgressing(reconciledRev, ocv1.ReasonProgressDeadlineExceeded, fmt.Sprintf("Revision has not rolled out for %d minute(s).", pd))
				reconcileErr = nil
				res = ctrl.Result{}
			} else if reconcileErr == nil {
				// We want to requeue so far in the future that the next reconciliation
				// can detect if the revision did not progress within the given timeout.
				// Thus, we plan the next reconcile slightly after (+2secs) the timeout is passed.
				drift := 2 * time.Second
				requeueAfter := existingRev.CreationTimestamp.Time.Add(timeout).Add(drift).Sub(c.Clock.Now()).Round(time.Second)
				l.Info(fmt.Sprintf("ProgressDeadline not exceeded, requeue after ~%v to check again.", requeueAfter))
				res = ctrl.Result{RequeueAfter: requeueAfter}
			}
		}
	}
	// Do checks before any Update()s, as Update() may modify the resource structure!
	updateStatus := !equality.Semantic.DeepEqual(existingRev.Status, reconciledRev.Status)

	unexpectedFieldsChanged := checkForUnexpectedClusterExtensionRevisionFieldChange(*existingRev, *reconciledRev)
	if unexpectedFieldsChanged {
		panic("spec or metadata changed by reconciler")
	}

	// NOTE: finalizer updates are performed during c.reconcile as patches, so that reconcile can
	//   continue performing logic after successfully setting the finalizer. therefore we only need
	//   to set status here.

	if updateStatus {
		if err := c.Client.Status().Update(ctx, reconciledRev); err != nil {
			reconcileErr = errors.Join(reconcileErr, fmt.Errorf("error updating status: %v", err))
		}
	}

	return res, reconcileErr
}

// Compare resources - ignoring status & metadata.finalizers
func checkForUnexpectedClusterExtensionRevisionFieldChange(a, b ocv1.ClusterExtensionRevision) bool {
	a.Status, b.Status = ocv1.ClusterExtensionRevisionStatus{}, ocv1.ClusterExtensionRevisionStatus{}

	// when finalizers are updated during reconcile, we expect finalizers, managedFields, and resourceVersion
	// to be updated, so we ignore changes in these fields.
	a.Finalizers, b.Finalizers = []string{}, []string{}
	a.ManagedFields, b.ManagedFields = nil, nil
	a.ResourceVersion, b.ResourceVersion = "", ""
	return !equality.Semantic.DeepEqual(a.Spec, b.Spec)
}

func (c *ClusterExtensionRevisionReconciler) reconcile(ctx context.Context, cer *ocv1.ClusterExtensionRevision) (ctrl.Result, error) {
	l := log.FromContext(ctx)

	if !cer.DeletionTimestamp.IsZero() {
		return c.delete(ctx, cer)
	}

	phases, opts, err := c.buildBoxcutterPhases(ctx, cer)
	if err != nil {
		setRetryingConditions(cer, err.Error())
		return ctrl.Result{}, fmt.Errorf("converting to boxcutter revision: %v", err)
	}

	siblings, err := c.siblingRevisionNames(ctx, cer)
	if err != nil {
		setRetryingConditions(cer, err.Error())
		return ctrl.Result{}, fmt.Errorf("listing sibling revisions: %v", err)
	}

	revisionEngine, err := c.RevisionEngineFactory.CreateRevisionEngine(ctx, cer)
	if err != nil {
		setRetryingConditions(cer, err.Error())
		return ctrl.Result{}, fmt.Errorf("failed to create revision engine: %v", err)
	}

	revision := boxcutter.NewRevisionWithOwner(
		cer.Name,
		cer.Spec.Revision,
		phases,
		cer,
		ownerhandling.NewNative(c.Client.Scheme()),
	)

	if cer.Spec.LifecycleState == ocv1.ClusterExtensionRevisionLifecycleStateArchived {
		if err := c.TrackingCache.Free(ctx, cer); err != nil {
			markAsAvailableUnknown(cer, ocv1.ClusterExtensionRevisionReasonReconciling, err.Error())
			return ctrl.Result{}, fmt.Errorf("error stopping informers: %v", err)
		}
		return c.archive(ctx, revisionEngine, cer, revision)
	}

	if err := c.ensureFinalizer(ctx, cer, clusterExtensionRevisionTeardownFinalizer); err != nil {
		return ctrl.Result{}, fmt.Errorf("error ensuring teardown finalizer: %v", err)
	}

	if err := c.establishWatch(ctx, cer, revision); err != nil {
		werr := fmt.Errorf("establish watch: %v", err)
		setRetryingConditions(cer, werr.Error())
		return ctrl.Result{}, werr
	}

	rres, err := revisionEngine.Reconcile(ctx, revision, opts...)
	if err != nil {
		if rres != nil {
			// Log detailed reconcile reports only in debug mode (V(1)) to reduce verbosity.
			l.V(1).Info("reconcile report", "report", rres.String())
		}
		setRetryingConditions(cer, err.Error())
		return ctrl.Result{}, fmt.Errorf("revision reconcile: %v", err)
	}

	// Retry failing preflight checks with a flat 10s retry.
	// TODO: report status, backoff?
	if verr := rres.GetValidationError(); verr != nil {
		l.Error(fmt.Errorf("%w", verr), "preflight validation failed, retrying after 10s")
		setRetryingConditions(cer, fmt.Sprintf("revision validation error: %s", verr))
		return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
	}

	for i, pres := range rres.GetPhases() {
		if verr := pres.GetValidationError(); verr != nil {
			l.Error(fmt.Errorf("%w", verr), "phase preflight validation failed, retrying after 10s", "phase", i)
			setRetryingConditions(cer, fmt.Sprintf("phase %d validation error: %s", i, verr))
			return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
		}

		var collidingObjs []string
		for _, ores := range pres.GetObjects() {
			if ores.Action() == machinery.ActionCollision {
				collidingObjs = append(collidingObjs, ores.String())
			}
			if ores.Action() == machinery.ActionProgressed && siblings != nil {
				if ref := foreignRevisionController(ores.Object(), siblings); ref != nil {
					collidingObjs = append(collidingObjs, ores.String()+fmt.Sprintf("\nConflicting Owner: %s", ref.String()))
				}
			}
		}

		if len(collidingObjs) > 0 {
			l.Error(fmt.Errorf("object collision detected"), "object collision, retrying after 10s", "phase", i, "collisions", collidingObjs)
			setRetryingConditions(cer, fmt.Sprintf("revision object collisions in phase %d\n%s", i, strings.Join(collidingObjs, "\n\n")))
			return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
		}
	}

	revVersion := cer.GetAnnotations()[labels.BundleVersionKey]
	if rres.InTransition() {
		markAsProgressing(cer, ocv1.ReasonRollingOut, fmt.Sprintf("Revision %s is rolling out.", revVersion))
	}

	//nolint:nestif
	if rres.IsComplete() {
		// Archive previous revisions
		previous, err := c.listPreviousRevisions(ctx, cer)
		if err != nil {
			return ctrl.Result{}, fmt.Errorf("listing previous revisions: %v", err)
		}
		for _, a := range previous {
			patch := []byte(`{"spec":{"lifecycleState":"Archived"}}`)
			if err := c.Client.Patch(ctx, client.Object(a), client.RawPatch(types.MergePatchType, patch)); err != nil {
				// TODO: It feels like an error here needs to propagate to a status _somewhere_.
				//   Not sure the current CER makes sense? But it also feels off to set the CE
				//   status from outside the CE reconciler.
				return ctrl.Result{}, fmt.Errorf("archive previous Revision: %w", err)
			}
		}

		markAsProgressing(cer, ocv1.ReasonSucceeded, fmt.Sprintf("Revision %s has rolled out.", revVersion))
		markAsAvailable(cer, ocv1.ClusterExtensionRevisionReasonProbesSucceeded, "Objects are available and pass all probes.")

		// We'll probably only want to remove this once we are done updating the ClusterExtension conditions
		// as its one of the interfaces between the revision and the extension. If we still have the Succeeded for now
		// that's fine.
		meta.SetStatusCondition(&cer.Status.Conditions, metav1.Condition{
			Type:               ocv1.ClusterExtensionRevisionTypeSucceeded,
			Status:             metav1.ConditionTrue,
			Reason:             ocv1.ReasonSucceeded,
			Message:            "Revision succeeded rolling out.",
			ObservedGeneration: cer.Generation,
		})
	} else {
		var probeFailureMsgs []string
		for _, pres := range rres.GetPhases() {
			if pres.IsComplete() {
				continue
			}
			for _, ores := range pres.GetObjects() {
				// we probably want an AvailabilityProbeType and run through all of them independently of whether
				// the revision is complete or not
				pr := ores.ProbeResults()[boxcutter.ProgressProbeType]
				if pr.Status == machinerytypes.ProbeStatusTrue {
					continue
				}

				obj := ores.Object()
				gvk := obj.GetObjectKind().GroupVersionKind()
				// I think these can be pretty large and verbose. We may want to
				// work a little on the formatting...?
				probeFailureMsgs = append(probeFailureMsgs, fmt.Sprintf(
					"Object %s.%s %s/%s: %v",
					gvk.Kind, gvk.GroupVersion().String(),
					obj.GetNamespace(), obj.GetName(), strings.Join(pr.Messages, " and "),
				))
				break
			}
		}

		if len(probeFailureMsgs) > 0 {
			markAsUnavailable(cer, ocv1.ClusterExtensionRevisionReasonProbeFailure, strings.Join(probeFailureMsgs, "\n"))
		} else {
			markAsUnavailable(cer, ocv1.ReasonRollingOut, fmt.Sprintf("Revision %s is rolling out.", revVersion))
		}
		if meta.FindStatusCondition(cer.Status.Conditions, ocv1.ClusterExtensionRevisionTypeProgressing) == nil {
			markAsProgressing(cer, ocv1.ReasonRollingOut, fmt.Sprintf("Revision %s is rolling out.", revVersion))
		}
	}

	return ctrl.Result{}, nil
}

func (c *ClusterExtensionRevisionReconciler) delete(ctx context.Context, cer *ocv1.ClusterExtensionRevision) (ctrl.Result, error) {
	if err := c.TrackingCache.Free(ctx, cer); err != nil {
		markAsAvailableUnknown(cer, ocv1.ClusterExtensionRevisionReasonReconciling, err.Error())
		return ctrl.Result{}, fmt.Errorf("error stopping informers: %v", err)
	}
	if err := c.removeFinalizer(ctx, cer, clusterExtensionRevisionTeardownFinalizer); err != nil {
		return ctrl.Result{}, fmt.Errorf("error removing teardown finalizer: %v", err)
	}
	return ctrl.Result{}, nil
}

func (c *ClusterExtensionRevisionReconciler) archive(ctx context.Context, revisionEngine RevisionEngine, cer *ocv1.ClusterExtensionRevision, revision boxcutter.RevisionBuilder) (ctrl.Result, error) {
	tdres, err := revisionEngine.Teardown(ctx, revision)
	if err != nil {
		err = fmt.Errorf("error archiving revision: %v", err)
		setRetryingConditions(cer, err.Error())
		return ctrl.Result{}, err
	}
	if tdres != nil && !tdres.IsComplete() {
		setRetryingConditions(cer, "removing revision resources that are not owned by another revision")
		return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
	}
	// Ensure conditions are set before removing the finalizer when archiving
	if markAsArchived(cer) {
		return ctrl.Result{}, nil
	}
	if err := c.removeFinalizer(ctx, cer, clusterExtensionRevisionTeardownFinalizer); err != nil {
		return ctrl.Result{}, fmt.Errorf("error removing teardown finalizer: %v", err)
	}
	return ctrl.Result{}, nil
}

type Sourcerer interface {
	Source(handler handler.EventHandler, predicates ...predicate.Predicate) source.Source
}

func (c *ClusterExtensionRevisionReconciler) SetupWithManager(mgr ctrl.Manager) error {
	skipProgressDeadlineExceededPredicate := predicate.Funcs{
		UpdateFunc: func(e event.UpdateEvent) bool {
			rev, ok := e.ObjectNew.(*ocv1.ClusterExtensionRevision)
			if !ok {
				return true
			}
			// allow deletions to happen
			if !rev.DeletionTimestamp.IsZero() {
				return true
			}
			if cnd := meta.FindStatusCondition(rev.Status.Conditions, ocv1.ClusterExtensionRevisionTypeProgressing); cnd != nil && cnd.Status == metav1.ConditionFalse && cnd.Reason == ocv1.ReasonProgressDeadlineExceeded {
				return false
			}
			return true
		},
	}
	c.Clock = clock.RealClock{}
	return ctrl.NewControllerManagedBy(mgr).
		For(
			&ocv1.ClusterExtensionRevision{},
			builder.WithPredicates(
				predicate.ResourceVersionChangedPredicate{},
				skipProgressDeadlineExceededPredicate,
			),
		).
		WatchesRawSource(
			c.TrackingCache.Source(
				handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &ocv1.ClusterExtensionRevision{}),
				predicate.ResourceVersionChangedPredicate{},
			),
		).
		Complete(c)
}

func (c *ClusterExtensionRevisionReconciler) establishWatch(ctx context.Context, cer *ocv1.ClusterExtensionRevision, revision boxcutter.RevisionBuilder) error {
	gvks := sets.New[schema.GroupVersionKind]()
	for _, phase := range revision.GetPhases() {
		for _, obj := range phase.GetObjects() {
			gvks.Insert(obj.GetObjectKind().GroupVersionKind())
		}
	}

	return c.TrackingCache.Watch(ctx, cer, gvks)
}

func (c *ClusterExtensionRevisionReconciler) ensureFinalizer(
	ctx context.Context, obj client.Object, finalizer string,
) error {
	if controllerutil.ContainsFinalizer(obj, finalizer) {
		return nil
	}

	controllerutil.AddFinalizer(obj, finalizer)
	patch := map[string]any{
		"metadata": map[string]any{
			"resourceVersion": obj.GetResourceVersion(),
			"finalizers":      obj.GetFinalizers(),
		},
	}
	patchJSON, err := json.Marshal(patch)
	if err != nil {
		return fmt.Errorf("marshalling patch to remove finalizer: %w", err)
	}
	if err := c.Client.Patch(ctx, obj, client.RawPatch(types.MergePatchType, patchJSON)); err != nil {
		return fmt.Errorf("adding finalizer: %w", err)
	}
	return nil
}

func (c *ClusterExtensionRevisionReconciler) removeFinalizer(ctx context.Context, obj client.Object, finalizer string) error {
	if !controllerutil.ContainsFinalizer(obj, finalizer) {
		return nil
	}

	controllerutil.RemoveFinalizer(obj, finalizer)

	patch := map[string]any{
		"metadata": map[string]any{
			"resourceVersion": obj.GetResourceVersion(),
			"finalizers":      obj.GetFinalizers(),
		},
	}
	patchJSON, err := json.Marshal(patch)
	if err != nil {
		return fmt.Errorf("marshalling patch to remove finalizer: %w", err)
	}
	if err := c.Client.Patch(ctx, obj, client.RawPatch(types.MergePatchType, patchJSON)); err != nil {
		return fmt.Errorf("removing finalizer: %w", err)
	}
	return nil
}

// listPreviousRevisions returns active revisions belonging to the same ClusterExtension with lower revision numbers.
// Filters out the current revision, archived revisions, deleting revisions, and revisions with equal or higher numbers.
func (c *ClusterExtensionRevisionReconciler) listPreviousRevisions(ctx context.Context, cer *ocv1.ClusterExtensionRevision) ([]*ocv1.ClusterExtensionRevision, error) {
	ownerLabel, ok := cer.Labels[labels.OwnerNameKey]
	if !ok {
		// No owner label means this revision isn't properly labeled - return empty list
		return nil, nil
	}

	revList := &ocv1.ClusterExtensionRevisionList{}
	if err := c.TrackingCache.List(ctx, revList, client.MatchingLabels{
		labels.OwnerNameKey: ownerLabel,
	}); err != nil {
		return nil, fmt.Errorf("listing revisions: %w", err)
	}

	previous := make([]*ocv1.ClusterExtensionRevision, 0, len(revList.Items))
	for i := range revList.Items {
		r := &revList.Items[i]
		if r.Name == cer.Name {
			continue
		}
		// Skip archived or deleting revisions
		if r.Spec.LifecycleState == ocv1.ClusterExtensionRevisionLifecycleStateArchived ||
			!r.DeletionTimestamp.IsZero() {
			continue
		}
		// Only include revisions with lower revision numbers (actual previous revisions)
		if r.Spec.Revision >= cer.Spec.Revision {
			continue
		}
		previous = append(previous, r)
	}

	return previous, nil
}

func (c *ClusterExtensionRevisionReconciler) buildBoxcutterPhases(ctx context.Context, cer *ocv1.ClusterExtensionRevision) ([]boxcutter.Phase, []boxcutter.RevisionReconcileOption, error) {
	previous, err := c.listPreviousRevisions(ctx, cer)
	if err != nil {
		return nil, nil, fmt.Errorf("listing previous revisions: %w", err)
	}

	// Convert to []client.Object for boxcutter
	previousObjs := make([]client.Object, len(previous))
	for i, rev := range previous {
		previousObjs[i] = rev
	}

	progressionProbes, err := buildProgressionProbes(cer.Spec.ProgressionProbes)
	if err != nil {
		return nil, nil, err
	}

	opts := []boxcutter.RevisionReconcileOption{
		boxcutter.WithPreviousOwners(previousObjs),
		boxcutter.WithProbe(boxcutter.ProgressProbeType, progressionProbes),
	}

	phases := make([]boxcutter.Phase, 0)
	for _, specPhase := range cer.Spec.Phases {
		objs := make([]client.Object, 0)
		for _, specObj := range specPhase.Objects {
			obj := specObj.Object.DeepCopy()

			objLabels := obj.GetLabels()
			if objLabels == nil {
				objLabels = map[string]string{}
			}
			objLabels[labels.OwnerNameKey] = cer.Labels[labels.OwnerNameKey]
			obj.SetLabels(objLabels)

			switch cp := EffectiveCollisionProtection(cer.Spec.CollisionProtection, specPhase.CollisionProtection, specObj.CollisionProtection); cp {
			case ocv1.CollisionProtectionIfNoController, ocv1.CollisionProtectionNone:
				opts = append(opts, boxcutter.WithObjectReconcileOptions(
					obj, boxcutter.WithCollisionProtection(cp)))
			}

			objs = append(objs, obj)
		}
		phases = append(phases, boxcutter.NewPhase(specPhase.Name, objs))
	}
	return phases, opts, nil
}

// EffectiveCollisionProtection resolves the collision protection value using
// the inheritance hierarchy: object > phase > spec > default ("Prevent").
func EffectiveCollisionProtection(cp ...ocv1.CollisionProtection) ocv1.CollisionProtection {
	ecp := ocv1.CollisionProtectionPrevent
	for _, c := range cp {
		if c != "" {
			ecp = c
		}
	}
	return ecp
}

// siblingRevisionNames returns the names of all ClusterExtensionRevisions that belong to
// the same ClusterExtension as cer. Returns nil when cer has no owner label.
func (c *ClusterExtensionRevisionReconciler) siblingRevisionNames(ctx context.Context, cer *ocv1.ClusterExtensionRevision) (sets.Set[string], error) {
	ownerLabel, ok := cer.Labels[labels.OwnerNameKey]
	if !ok {
		return nil, nil
	}
	revList := &ocv1.ClusterExtensionRevisionList{}
	if err := c.TrackingCache.List(ctx, revList, client.MatchingLabels{
		labels.OwnerNameKey: ownerLabel,
	}); err != nil {
		return nil, fmt.Errorf("listing sibling revisions: %w", err)
	}
	names := sets.New[string]()
	for i := range revList.Items {
		names.Insert(revList.Items[i].Name)
	}
	return names, nil
}

// foreignRevisionController returns the controller OwnerReference when obj is owned by a
// ClusterExtensionRevision that is not in siblings (i.e. belongs to a different ClusterExtension).
// Returns nil when the controller is a sibling or is not a ClusterExtensionRevision.
func foreignRevisionController(obj metav1.Object, siblings sets.Set[string]) *metav1.OwnerReference {
	refs := obj.GetOwnerReferences()
	for i := range refs {
		if refs[i].Controller != nil && *refs[i].Controller &&
			refs[i].Kind == ocv1.ClusterExtensionRevisionKind &&
			refs[i].APIVersion == ocv1.GroupVersion.String() &&
			!siblings.Has(refs[i].Name) {
			return &refs[i]
		}
	}
	return nil
}

// buildProgressionProbes creates a set of boxcutter probes from the fields provided in the CER's spec.progressionProbes.
// Returns nil and an error if encountered while attempting to build the probes.
func buildProgressionProbes(progressionProbes []ocv1.ProgressionProbe) (probing.And, error) {
	userProbes := probing.And{}
	if len(progressionProbes) < 1 {
		return userProbes, nil
	}
	for _, progressionProbe := range progressionProbes {
		// Collect all user assertions into a single 'And'
		assertions := probing.And{}
		for _, probe := range progressionProbe.Assertions {
			switch probe.Type {
			// Switch based on the union discriminator
			case ocv1.ProbeTypeConditionEqual:
				conditionProbe := probing.ConditionProbe(probe.ConditionEqual)
				assertions = append(assertions, &conditionProbe)
			case ocv1.ProbeTypeFieldsEqual:
				fieldsEqualProbe := probing.FieldsEqualProbe(probe.FieldsEqual)
				assertions = append(assertions, &fieldsEqualProbe)
			case ocv1.ProbeTypeFieldValue:
				fieldValueProbe := applier.FieldValueProbe(probe.FieldValue)
				assertions = append(assertions, &fieldValueProbe)
			default:
				return nil, fmt.Errorf("unknown progressionProbe assertion probe type: %s", probe.Type)
			}
		}

		// Create the selector probe based on user-requested type and provide the assertions
		var selectorProbe probing.Prober
		switch progressionProbe.Selector.Type {
		// Switch based on the union discriminator
		case ocv1.SelectorTypeGroupKind:
			selectorProbe = &probing.GroupKindSelector{
				GroupKind: schema.GroupKind(progressionProbe.Selector.GroupKind),
				Prober:    assertions,
			}
		case ocv1.SelectorTypeLabel:
			selector, err := metav1.LabelSelectorAsSelector(&progressionProbe.Selector.Label)
			if err != nil {
				return nil, fmt.Errorf("invalid label selector in progressionProbe (%v): %w", progressionProbe.Selector.Label, err)
			}
			selectorProbe = &probing.LabelSelector{
				Selector: selector,
				Prober:   assertions,
			}
		default:
			return nil, fmt.Errorf("unknown progressionProbe selector type: %s", progressionProbe.Selector.Type)
		}
		userProbes = append(userProbes, &probing.ObservedGenerationProbe{
			Prober: selectorProbe,
		})
	}
	return userProbes, nil
}

func setRetryingConditions(cer *ocv1.ClusterExtensionRevision, message string) {
	markAsProgressing(cer, ocv1.ClusterExtensionRevisionReasonRetrying, message)
	if meta.FindStatusCondition(cer.Status.Conditions, ocv1.ClusterExtensionRevisionTypeAvailable) != nil {
		markAsAvailableUnknown(cer, ocv1.ClusterExtensionRevisionReasonReconciling, message)
	}
}

func markAsProgressing(cer *ocv1.ClusterExtensionRevision, reason, message string) {
	meta.SetStatusCondition(&cer.Status.Conditions, metav1.Condition{
		Type:               ocv1.ClusterExtensionRevisionTypeProgressing,
		Status:             metav1.ConditionTrue,
		Reason:             reason,
		Message:            message,
		ObservedGeneration: cer.Generation,
	})
}

func markAsNotProgressing(cer *ocv1.ClusterExtensionRevision, reason, message string) bool {
	return meta.SetStatusCondition(&cer.Status.Conditions, metav1.Condition{
		Type:               ocv1.ClusterExtensionRevisionTypeProgressing,
		Status:             metav1.ConditionFalse,
		Reason:             reason,
		Message:            message,
		ObservedGeneration: cer.Generation,
	})
}

func markAsAvailable(cer *ocv1.ClusterExtensionRevision, reason, message string) bool {
	return meta.SetStatusCondition(&cer.Status.Conditions, metav1.Condition{
		Type:               ocv1.ClusterExtensionRevisionTypeAvailable,
		Status:             metav1.ConditionTrue,
		Reason:             reason,
		Message:            message,
		ObservedGeneration: cer.Generation,
	})
}

func markAsUnavailable(cer *ocv1.ClusterExtensionRevision, reason, message string) {
	meta.SetStatusCondition(&cer.Status.Conditions, metav1.Condition{
		Type:               ocv1.ClusterExtensionRevisionTypeAvailable,
		Status:             metav1.ConditionFalse,
		Reason:             reason,
		Message:            message,
		ObservedGeneration: cer.Generation,
	})
}

func markAsAvailableUnknown(cer *ocv1.ClusterExtensionRevision, reason, message string) bool {
	return meta.SetStatusCondition(&cer.Status.Conditions, metav1.Condition{
		Type:               ocv1.ClusterExtensionRevisionTypeAvailable,
		Status:             metav1.ConditionUnknown,
		Reason:             reason,
		Message:            message,
		ObservedGeneration: cer.Generation,
	})
}

func markAsArchived(cer *ocv1.ClusterExtensionRevision) bool {
	const msg = "revision is archived"
	updated := markAsNotProgressing(cer, ocv1.ClusterExtensionRevisionReasonArchived, msg)
	return markAsAvailableUnknown(cer, ocv1.ClusterExtensionRevisionReasonArchived, msg) || updated
}