package stages

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/kelseyhightower/envconfig"
	gocache "github.com/patrickmn/go-cache"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/fields"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	kerrors "k8s.io/apimachinery/pkg/util/errors"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/util/retry"
	"k8s.io/utils/ptr"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/source"

	rolloutsapi "github.com/akuity/kargo/api/stubs/rollouts/v1alpha1"
	kargoapi "github.com/akuity/kargo/api/v1alpha1"
	"github.com/akuity/kargo/pkg/api"
	"github.com/akuity/kargo/pkg/conditions"
	"github.com/akuity/kargo/pkg/controller"
	argocdapi "github.com/akuity/kargo/pkg/controller/argocd/api/v1alpha1"
	kargoEvent "github.com/akuity/kargo/pkg/event"
	k8sevent "github.com/akuity/kargo/pkg/event/kubernetes"
	exprfn "github.com/akuity/kargo/pkg/expressions/function"
	"github.com/akuity/kargo/pkg/health"
	"github.com/akuity/kargo/pkg/indexer"
	"github.com/akuity/kargo/pkg/kargo"
	"github.com/akuity/kargo/pkg/kubeclient"
	"github.com/akuity/kargo/pkg/kubernetes"
	libEvent "github.com/akuity/kargo/pkg/kubernetes/event"
	"github.com/akuity/kargo/pkg/logging"
	"github.com/akuity/kargo/pkg/pattern"
	intpredicate "github.com/akuity/kargo/pkg/predicate"
	"github.com/akuity/kargo/pkg/rollouts"
)

// ReconcilerConfig represents configuration for the stage reconciler.
type ReconcilerConfig struct {
	IsDefaultController                bool   `envconfig:"IS_DEFAULT_CONTROLLER"`
	ShardName                          string `envconfig:"SHARD_NAME"`
	RolloutsIntegrationEnabled         bool   `envconfig:"ROLLOUTS_INTEGRATION_ENABLED"`
	RolloutsControllerInstanceID       string `envconfig:"ROLLOUTS_CONTROLLER_INSTANCE_ID"`
	MaxConcurrentControlFlowReconciles int    `envconfig:"MAX_CONCURRENT_CONTROL_FLOW_RECONCILES" default:"4"`
	MaxConcurrentReconciles            int    `envconfig:"MAX_CONCURRENT_STAGE_RECONCILES" default:"4"`
}

// Name returns the name of the Stage controller.
func (c ReconcilerConfig) Name() string {
	const name = "stage-controller"
	if c.ShardName != "" {
		return name + "-" + c.ShardName
	}
	return name
}

// ReconcilerConfigFromEnv returns a new ReconcilerConfig populated from the
// environment variables.
func ReconcilerConfigFromEnv() ReconcilerConfig {
	cfg := ReconcilerConfig{}
	envconfig.MustProcess("", &cfg)
	return cfg
}

type RegularStageReconciler struct {
	cfg            ReconcilerConfig
	client         client.Client
	eventSender    kargoEvent.Sender
	healthChecker  health.AggregatingChecker
	shardPredicate controller.ResponsibleFor[kargoapi.Stage]

	backoffCfg wait.Backoff
}

// NewRegularStageReconciler creates a new Stages reconciler.
func NewRegularStageReconciler(
	cfg ReconcilerConfig,
	healthChecker health.AggregatingChecker,
) *RegularStageReconciler {
	return &RegularStageReconciler{
		cfg:           cfg,
		healthChecker: healthChecker,
		shardPredicate: controller.ResponsibleFor[kargoapi.Stage]{
			IsDefaultController: cfg.IsDefaultController,
			ShardName:           cfg.ShardName,
		},
		backoffCfg: wait.Backoff{
			Duration: 1 * time.Second,
			Factor:   2,
			Steps:    10,
			Cap:      2 * time.Minute,
			Jitter:   0.1,
		},
	}
}

// SetupWithManager sets up the Stage reconciler with the given controller
// manager. It registers the reconciler with the manager and sets up watches
// on the required objects.
func (r *RegularStageReconciler) SetupWithManager(
	ctx context.Context,
	kargoMgr, argocdMgr ctrl.Manager,
	sharedIndexer client.FieldIndexer,
) error {
	// Configure client and event recorder using manager.
	r.client = kargoMgr.GetClient()
	r.eventSender = k8sevent.NewEventSender(
		libEvent.NewRecorder(ctx, kargoMgr.GetScheme(), kargoMgr.GetClient(), r.cfg.Name()),
	)

	// This index is used to find all Promotions that are associated with a
	// specific Stage.
	if err := sharedIndexer.IndexField(
		ctx,
		&kargoapi.Promotion{},
		indexer.PromotionsByStageField,
		indexer.PromotionsByStage,
	); err != nil {
		return fmt.Errorf("error setting up index for Promotions by Stage: %w", err)
	}

	// This index is used to determine if a Promotion already exists for a
	// Stage and Freight combination.
	if err := sharedIndexer.IndexField(
		ctx,
		&kargoapi.Promotion{},
		indexer.PromotionsByStageAndFreightField,
		indexer.PromotionsByStageAndFreight,
	); err != nil {
		return fmt.Errorf("error setting up index for Promotions by Stage and Freight: %w", err)
	}

	// This index is used to find Promotions that are non-terminal.
	if err := sharedIndexer.IndexField(
		ctx,
		&kargoapi.Promotion{},
		indexer.PromotionsByTerminalField,
		indexer.PromotionsByTerminal,
	); err != nil {
		return fmt.Errorf(
			"error setting up index for Promotions by terminal phase: %w", err,
		)
	}

	// This index is used to find Freight that are directly available from a
	// Warehouse and can be automatically promoted to a Stage.
	if err := sharedIndexer.IndexField(
		ctx,
		&kargoapi.Freight{},
		indexer.FreightByWarehouseField,
		indexer.FreightByWarehouse,
	); err != nil {
		return fmt.Errorf("error setting up index for Freight by Warehouse: %w", err)
	}

	// This index is used to find all Freight that have been verified in upstream
	// Stages and can be automatically promoted to a Stage.
	if err := sharedIndexer.IndexField(
		ctx,
		&kargoapi.Freight{},
		indexer.FreightByVerifiedStagesField,
		indexer.FreightByVerifiedStages,
	); err != nil {
		return fmt.Errorf("error setting up index for Freight by Stages in which it has been verified: %w", err)
	}

	// This index is used to find all Freight that have been explicitly approved
	// for a Stage and can be automatically promoted to that Stage.
	if err := sharedIndexer.IndexField(
		ctx,
		&kargoapi.Freight{},
		indexer.FreightApprovedForStagesField,
		indexer.FreightApprovedForStages,
	); err != nil {
		return fmt.Errorf("index Freight by Stages for which it has been approved: %w", err)
	}

	// Build the controller with the reconciler.
	c, err := ctrl.NewControllerManagedBy(kargoMgr).
		For(&kargoapi.Stage{}).
		WithEventFilter(controller.ResponsibleFor[client.Object]{
			IsDefaultController: r.cfg.IsDefaultController,
			ShardName:           r.cfg.ShardName,
		}).
		WithEventFilter(intpredicate.IgnoreDelete[client.Object]{}).
		WithEventFilter(
			predicate.And(
				IsControlFlowStage(false),
				predicate.Or(
					predicate.GenerationChangedPredicate{},
					kargo.RefreshRequested{},
					kargo.ReverifyRequested{},
					kargo.VerificationAbortRequested{},
				),
			),
		).
		WithOptions(controller.CommonOptions(r.cfg.MaxConcurrentReconciles)).
		Build(r)
	if err != nil {
		return fmt.Errorf("error building Stage reconciler: %w", err)
	}

	// Configure the watches.
	// Changes to these objects that match the constraints from the predicates
	// will enqueue a reconciliation for the related Stage(s).
	logger := logging.LoggerFromContext(ctx)

	// Watch for Promotions for which the phase changed and enqueue the related
	// Stage for reconciliation.
	if err = c.Watch(
		source.Kind(
			kargoMgr.GetCache(),
			&kargoapi.Promotion{},
			handler.TypedEnqueueRequestForOwner[*kargoapi.Promotion](
				kargoMgr.GetScheme(),
				kargoMgr.GetRESTMapper(),
				&kargoapi.Stage{},
				handler.OnlyControllerOwner(),
			),
			kargo.NewPromoPhaseChangedPredicate(logger),
		),
	); err != nil {
		return fmt.Errorf("unable to watch Promotions: %w", err)
	}

	// Watch for Freight that have been newly promoted to a Stage or newly marked
	// as verified in a Stage and enqueue downstream Stages for reconciliation.
	if err = c.Watch(
		source.Kind(
			kargoMgr.GetCache(),
			&kargoapi.Freight{},
			&downstreamStageEnqueuer[*kargoapi.Freight]{
				kargoClient: kargoMgr.GetClient(),
			},
		),
	); err != nil {
		return fmt.Errorf("unable to watch Freight from upstream Stages: %w", err)
	}

	// Watch for Freight that has been approved for a Stage and enqueue the Stage
	// for reconciliation.
	if err = c.Watch(
		source.Kind(
			kargoMgr.GetCache(),
			&kargoapi.Freight{},
			&stageEnqueuerForApprovedFreight[*kargoapi.Freight]{
				kargoClient: kargoMgr.GetClient(),
			},
		),
	); err != nil {
		return fmt.Errorf("unable to watch approved Freight: %w", err)
	}

	// Watch for newly produced Freight from a Warehouse and enqueue the related
	// Stages for reconciliation.
	if err = c.Watch(
		source.Kind(
			kargoMgr.GetCache(),
			&kargoapi.Freight{},
			&warehouseStageEnqueuer[*kargoapi.Freight]{
				kargoClient: kargoMgr.GetClient(),
			},
		),
	); err != nil {
		return fmt.Errorf("unable to watch Freight produced by Warehouse: %w", err)
	}

	// If we have an ArgoCD manager, then we should watch for changes to
	// ArgCD Applications and enqueue the related Stages for reconciliation.
	if argocdMgr != nil {
		if err = c.Watch(
			source.Kind(
				argocdMgr.GetCache(),
				&argocdapi.Application{},
				&stageEnqueuerForArgoCDChanges[*argocdapi.Application]{
					kargoClient: kargoMgr.GetClient(),
				},
			),
		); err != nil {
			return fmt.Errorf("unable to watch Applications: %w", err)
		}
	}

	// If the Argo Rollouts integration is enabled, then we should watch for
	// changes to AnalysisRuns and enqueue the related Stages for reconciliation.
	if r.cfg.RolloutsIntegrationEnabled {
		if err = sharedIndexer.IndexField(
			ctx,
			&kargoapi.Stage{},
			indexer.StagesByAnalysisRunField,
			indexer.StagesByAnalysisRun(r.cfg.ShardName, r.cfg.IsDefaultController),
		); err != nil {
			return fmt.Errorf("error setting up index for Stages by AnalysisRun: %w", err)
		}

		if err = c.Watch(
			source.Kind(
				kargoMgr.GetCache(),
				&rolloutsapi.AnalysisRun{},
				&stageEnqueuerForAnalysisRuns[*rolloutsapi.AnalysisRun]{
					kargoClient: kargoMgr.GetClient(),
				},
			),
		); err != nil {
			return fmt.Errorf("unable to watch AnalysisRuns: %w", err)
		}
	}

	logging.LoggerFromContext(ctx).Info(
		"Initialized regular Stage reconciler",
		"maxConcurrentReconciles", r.cfg.MaxConcurrentControlFlowReconciles,
	)

	return nil
}

func (r *RegularStageReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := logging.LoggerFromContext(ctx).WithValues(
		"namespace", req.Namespace,
		"stage", req.Name,
		"controlFlow", false,
	)
	ctx = logging.ContextWithLogger(ctx, logger)

	// Find the Stage.
	stage := &kargoapi.Stage{}
	if err := r.client.Get(ctx, req.NamespacedName, stage); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	// Safety check: do not reconcile Stages that are control flow Stages.
	if stage.IsControlFlow() {
		return ctrl.Result{}, nil
	}

	if !r.shardPredicate.IsResponsible(stage) {
		logger.Debug("ignoring Stage because it is not assigned to this shard")
		return ctrl.Result{}, nil
	}

	// Handle deletion of the Stage.
	if !stage.DeletionTimestamp.IsZero() {
		return ctrl.Result{}, r.handleDelete(ctx, stage)
	}

	// Ensure the Stage has a finalizer and requeue if it was added.
	// The reason to requeue is to ensure that a possible deletion of the Stage
	// directly after the finalizer was added is handled without delay.
	if ok, err := api.EnsureFinalizer(ctx, r.client, stage); ok || err != nil {
		return ctrl.Result{RequeueAfter: 100 * time.Millisecond}, err
	}

	// Reconcile the Stage.
	logger.Debug("reconciling Stage")
	newStatus, needsRequeue, reconcileErr := r.reconcile(ctx, stage, time.Now())
	logger.Debug("done reconciling Stage")

	// Record the current refresh token as having been handled.
	if token, ok := api.RefreshAnnotationValue(stage.GetAnnotations()); ok {
		newStatus.LastHandledRefresh = token
	}

	// Patch the status of the Stage.
	if err := kubeclient.PatchStatus(ctx, r.client, stage, func(status *kargoapi.StageStatus) {
		*status = newStatus
	}); err != nil {
		// Prioritize the reconcile error if it exists.
		if reconcileErr != nil {
			logger.Error(err, "failed to update Stage status after reconciliation error")
			return ctrl.Result{}, reconcileErr
		}
		return ctrl.Result{}, fmt.Errorf("failed to update Stage status: %w", err)
	}

	// Return the reconcile error if it exists.
	if reconcileErr != nil {
		return ctrl.Result{}, reconcileErr
	}
	// Immediate requeue if needed.
	if needsRequeue {
		return ctrl.Result{RequeueAfter: 100 * time.Millisecond}, nil
	}
	// Otherwise, requeue after a delay.
	// TODO: Make the requeue delay configurable.
	return ctrl.Result{RequeueAfter: 5 * time.Minute}, nil
}

func (r *RegularStageReconciler) reconcile(
	ctx context.Context,
	stage *kargoapi.Stage,
	startTime time.Time,
) (kargoapi.StageStatus, bool, error) {
	logger := logging.LoggerFromContext(ctx)
	newStatus := *stage.Status.DeepCopy()

	// Mark the Stage as reconciling.
	conditions.Set(&newStatus, &metav1.Condition{
		Type:               kargoapi.ConditionTypeReconciling,
		Status:             metav1.ConditionTrue,
		Reason:             "Reconciling",
		ObservedGeneration: stage.Generation,
	})

	var requestRequeue bool
	subReconcilers := []struct {
		name      string
		reconcile func() (kargoapi.StageStatus, error)
	}{
		{
			name: "syncing Promotions",
			reconcile: func() (kargoapi.StageStatus, error) {
				status, hasPendingPromotions, err := r.syncPromotions(ctx, stage)
				if err != nil {
					err = fmt.Errorf("failed to sync Promotions: %w", err)
				}
				// If we have no current Promotion and there are pending Promotions,
				// then we should request an immediate requeue to ensure that we
				// process the next Promotion as soon as possible.
				if status.CurrentPromotion == nil && hasPendingPromotions {
					requestRequeue = true
				}
				return status, err
			},
		},
		{
			name: "syncing Freight",
			reconcile: func() (kargoapi.StageStatus, error) {
				if err := r.syncFreight(ctx, stage); err != nil {
					return stage.Status, fmt.Errorf("failed to sync Freight: %w", err)
				}
				return stage.Status, nil
			},
		},
		{
			name: "assessing health",
			reconcile: func() (kargoapi.StageStatus, error) {
				status := r.assessHealth(ctx, stage)
				if status.Health != nil && status.Health.Status == kargoapi.HealthStateUnknown {
					// If Stage health has evaluated to Unknown, there are two specific
					// scenarios between which we must distinguish and handle differently
					// from one another:
					//
					//  1. If Stage health is unknown specifically because the last
					//     Promotion did not succeed, we know we cannot obtain a more
					//     definitive assessment of Stage health until a new Promotion has
					//     restored the Stage to a consistent state by executing to
					//     completion. In such a case, we're DONE reconciling the Stage,
					//     for now. We would like the Stage's conditions to reflect that
					//     and for the next reconciliation attempt to occur after the
					//     usual interval. This can be accomplished by returning a nil
					//     error.
					//
					//  2. If Stage health is unknown for any other reason, we MAY
					//     possibly obtain a more definitive assessment simply by
					//     re-attempting reconciliation. In such a case, we would like
					//     the Stage's conditions to reflect that we're still trying to
					//     reconcile, with subsequent attempts observing a progressive
					//     backoff. This can be accomplished by returning an error.
					if lastPromo := status.LastPromotion; lastPromo == nil ||
						lastPromo.Status == nil ||
						!lastPromo.Status.Phase.IsTerminal() ||
						lastPromo.Status.Phase == kargoapi.PromotionPhaseSucceeded {
						// Scenario 2: There was no last Promotion or there was and it was
						// successful. Whatever the reason the Stage health evaluated to
						// Unknown, an unsuccessful Promotion wasn't it.
						return status, errors.New("Stage health evaluated to Unknown") // nolint: staticcheck
					}
				}
				// Health assessment was definitive OR scenario 1: Stage health is
				// unknown specifically because the last Promotion did not succeed.
				return status, nil
			},
		},
		{
			name: "verifying Stage Freight",
			reconcile: func() (kargoapi.StageStatus, error) {
				status, err := r.verifyStageFreight(ctx, stage, startTime, time.Now)
				if err != nil {
					err = fmt.Errorf("failed to verify Stage Freight: %w", err)
				}
				// If we have a non-terminal verification for the current Freight,
				// then we should rely on the watcher to requeue the Stage when the
				// verification completes.
				curFreightCol := status.FreightHistory.Current()
				if curFreightCol != nil && curFreightCol.HasNonTerminalVerification() {
					requestRequeue = false
				}
				return status, err
			},
		},
		{
			name: "verifying Freight for Stage",
			reconcile: func() (kargoapi.StageStatus, error) {
				status, err := r.markFreightVerifiedForStage(ctx, stage)
				if err != nil {
					err = fmt.Errorf("failed to verify Freight for Stage: %w", err)
				}
				return status, err
			},
		},
		{
			name: "auto-promoting Freight",
			reconcile: func() (kargoapi.StageStatus, error) {
				status, err := r.autoPromoteFreight(ctx, stage)
				if err != nil {
					err = fmt.Errorf("failed to auto-promote Freight: %w", err)
				}
				return status, err
			},
		},
	}
	for _, subR := range subReconcilers {
		logger.Debug(subR.name)

		// Reconcile the Stage with the sub-reconciler.
		var err error
		newStatus, err = subR.reconcile()

		// Summarize the conditions after each sub-reconciler to ensure that
		// we have a consistent view of the Stage status.
		summarizeConditions(stage, &newStatus, err)

		// If an error occurred during the sub-reconciler, then we should
		// return the error which will cause the Stage to be requeued.
		if err != nil {
			return newStatus, false, err
		}

		// Patch the status of the Stage after each sub-reconciler to show
		// progress.
		if err = kubeclient.PatchStatus(ctx, r.client, stage, func(status *kargoapi.StageStatus) {
			*status = newStatus
		}); err != nil {
			logger.Error(err, fmt.Sprintf("failed to update Stage status after %s", subR.name))
		}
	}

	// If an immediate requeue was not requested, then we can delete the
	// Reconciling condition as we have finished reconciling the Stage
	// and did not encounter any errors.
	if !requestRequeue {
		conditions.Delete(&newStatus, kargoapi.ConditionTypeReconciling)
	}

	return newStatus, requestRequeue, nil
}

// syncPromotions synchronizes the Promotions for a Stage. It determines the
// current state of the Stage based on the Promotions that are running or have
// completed.
func (r *RegularStageReconciler) syncPromotions(
	ctx context.Context,
	stage *kargoapi.Stage,
) (kargoapi.StageStatus, bool, error) {
	logger := logging.LoggerFromContext(ctx)
	newStatus := *stage.Status.DeepCopy()

	// List all Promotions for the Stage.
	promotions := &kargoapi.PromotionList{}
	if err := r.client.List(
		ctx,
		promotions,
		client.InNamespace(stage.Namespace),
		client.MatchingFieldsSelector{
			Selector: fields.OneTermEqualSelector(indexer.PromotionsByStageField, stage.Name),
		},
	); err != nil {
		err = fmt.Errorf(
			"failed to list Promotions for Stage %q in namespace %q: %w",
			stage.Name, stage.Namespace, err,
		)

		conditions.Set(&newStatus, &metav1.Condition{
			Type:               kargoapi.ConditionTypePromoting,
			Status:             metav1.ConditionUnknown,
			Reason:             "ListPromotionsFailed",
			Message:            err.Error(),
			ObservedGeneration: stage.Generation,
		})

		return newStatus, false, err
	}

	// If there are no Promotions, then we are not promoting any Freight.
	if len(promotions.Items) == 0 {
		logger.Debug("no Promotions found for Stage")

		// Ensure we delete any existing "current" Promotion related information
		// from the Stage status.
		conditions.Delete(&newStatus, kargoapi.ConditionTypePromoting)
		newStatus.CurrentPromotion = nil

		return newStatus, false, nil
	}

	// Sort the Promotions by phase and creation time to determine the current
	// state of the Stage.
	slices.SortFunc(promotions.Items, api.ComparePromotionByPhaseAndCreationTime)

	// The Promotion with the highest priority (i.e. a Running or Pending phase)
	// is the one that we will consider for the current state of the Stage.
	highestPrioPromo := promotions.Items[0]

	// The Promotion which is currently running on the Stage.
	currentPromo := stage.Status.CurrentPromotion

	// The last Promotion which ran on the Stage.
	lastPromo := stage.Status.LastPromotion

	// Track if there are any non-terminal promotions that need handling.
	// This is later used to determine if we should issue an immediate
	// requeue.
	var hasNonTerminalPromotions bool
	for _, promo := range promotions.Items {
		if !promo.Status.Phase.IsTerminal() {
			hasNonTerminalPromotions = true
			break
		}
	}

	// If the current Promotion is not the highest priority Promotion, or the
	// highest priority Promotion is in a terminal phase, then we must have
	// finished promoting.
	if currentPromo != nil && (currentPromo.Name != highestPrioPromo.Name || highestPrioPromo.Status.Phase.IsTerminal()) {
		// Gather information about the Promotions that have terminated since the
		// last reconciliation.
		var newPromotions []kargoapi.PromotionReference
		for _, promo := range promotions.Items {
			// Update the conditions to reflect that we are no longer promoting.
			conditions.Delete(&newStatus, kargoapi.ConditionTypePromoting)
			newStatus.CurrentPromotion = nil

			if lastPromo != nil {
				// We can break here since we know that all subsequent Promotions
				// will be older than the last Promotion we saw.
				// NB: This makes use of the fact that Promotion names are
				// generated, and contain a timestamp component which will ensure
				// that they can be sorted in a consistent order.
				if strings.Compare(promo.Name, lastPromo.Name) <= 0 {
					break
				}
			}

			if promo.Status.Phase.IsTerminal() {
				info := kargoapi.PromotionReference{
					Name:       promo.Name,
					Status:     promo.Status.DeepCopy(),
					FinishedAt: promo.Status.FinishedAt,
				}
				if promo.Status.Freight != nil {
					info.Freight = promo.Status.Freight.DeepCopy()
				}
				newPromotions = append(newPromotions, info)
			}
		}

		// As we will be appending to the Freight history, we need to ensure that
		// we order the Promotions from oldest to newest. This is because the
		// Freight history is garbage collected based on the number of entries,
		// and we want to ensure that the oldest entries are removed first.
		slices.SortFunc(newPromotions, func(a, b kargoapi.PromotionReference) int {
			return strings.Compare(a.Name, b.Name)
		})

		// Update the Stage status with the information about the newly terminated
		// Promotions, and any new Freight that was successfully promoted.
		for _, p := range newPromotions {
			promo := p
			newStatus.LastPromotion = &promo
			if p.Status.Phase == kargoapi.PromotionPhaseSucceeded {
				// If the Promotion was successful, then we should add the Freight
				// to the history of successfully promoted Freight.
				newStatus.FreightHistory.Record(promo.Status.FreightCollection)

				// Erase any health checks that were performed for the previous
				// Freight, as they are no longer relevant.
				newStatus.Health = nil
				conditions.Set(&newStatus, &metav1.Condition{
					Type:               kargoapi.ConditionTypeHealthy,
					Status:             metav1.ConditionUnknown,
					Reason:             "WaitingForHealthCheck",
					Message:            "Waiting for health check to be performed after successful promotion",
					ObservedGeneration: stage.Generation,
				})

				// Set verified condition to unknown to indicate that the
				// new Freight needs to be verified.
				conditions.Set(&newStatus, &metav1.Condition{
					Type:               kargoapi.ConditionTypeVerified,
					Status:             metav1.ConditionUnknown,
					Reason:             "WaitingForVerification",
					Message:            "Waiting for verification to be performed after successful promotion",
					ObservedGeneration: stage.Generation,
				})

				// Annotate the Stage with the latest information related to
				// ArgoCD Applications. This is used to provide deep links to the
				// ArgoCD UI for the Stage in the Kargo UI.
				//
				// NB: If the health checks do not include ArgoCD Applications,
				// then the annotation will be removed.
				if err := api.AnnotateStageWithArgoCDContext(ctx, r.client, p.Status.HealthChecks, stage); err != nil {
					// Let the error be logged, but do not return it as it is not
					// critical to the operation of the Stage.
					logger.Error(err, "failed to annotate Stage with ArgoCD context")
				}
			}
		}

		// Return at this point to allow the new Freight to be verified.
		return newStatus, hasNonTerminalPromotions, nil
	}

	// If the current Freight exists and has a non-terminal verification, wait
	// for it to complete regardless of health state to ensure we capture the
	// results.
	if curFreight := stage.Status.FreightHistory.Current(); curFreight != nil {
		if curFreight.HasNonTerminalVerification() {
			logger.Debug(
				"current Freight has a non-terminal verification: " +
					"wait for it to complete before allowing new promotions to start",
			)
			conditions.Delete(&newStatus, kargoapi.ConditionTypePromoting)
			return newStatus, hasNonTerminalPromotions, nil
		}

		// If we are in a healthy state, the current Freight needs to be verified
		// before we can allow the next Promotion to start. If we are unhealthy
		// or the verification failed, then we can allow the next Promotion to
		// start immediately as the expectation is that the Promotion can fix the
		// issue.
		if stage.Status.Health == nil || stage.Status.Health.Status == kargoapi.HealthStateHealthy {
			curVI := curFreight.VerificationHistory.Current()
			if curVI == nil || !curVI.Phase.IsTerminal() {
				logger.Debug("current Freight needs to be verified before allowing new promotions to start")
				conditions.Delete(&newStatus, kargoapi.ConditionTypePromoting)
				return newStatus, hasNonTerminalPromotions, nil
			}
		}
	}

	// If the highest priority Promotion is not in a terminal phase, then we
	// are promoting the Freight.
	if !highestPrioPromo.Status.Phase.IsTerminal() {
		conditions.Set(&newStatus, &metav1.Condition{
			Type:   kargoapi.ConditionTypePromoting,
			Status: metav1.ConditionTrue,
			Reason: "ActivePromotion",
			Message: fmt.Sprintf(
				"Promotion %q is currently %s",
				highestPrioPromo.Name, highestPrioPromo.Status.Phase,
			),
			ObservedGeneration: stage.Generation,
		})

		newStatus.CurrentPromotion = &kargoapi.PromotionReference{
			Name: highestPrioPromo.Name,
		}
		if freight := highestPrioPromo.Status.Freight; freight != nil {
			newStatus.CurrentPromotion.Freight = freight.DeepCopy()
		}
		return newStatus, hasNonTerminalPromotions, nil
	}

	// If the highest priority Promotion is in a terminal phase, then we are
	// not promoting.
	conditions.Delete(&newStatus, kargoapi.ConditionTypePromoting)
	return newStatus, hasNonTerminalPromotions, nil
}

// assessHealth assesses the health of a Stage based on the health checks from
// the last Promotion.
func (r *RegularStageReconciler) assessHealth(ctx context.Context, stage *kargoapi.Stage) kargoapi.StageStatus {
	logger := logging.LoggerFromContext(ctx)
	newStatus := *stage.Status.DeepCopy()

	if currentPromo := stage.Status.CurrentPromotion; currentPromo != nil {
		logger.Debug("Promotion is in progress: no health checks to perform")
		conditions.Set(&newStatus, &metav1.Condition{
			Type:               kargoapi.ConditionTypeHealthy,
			Status:             metav1.ConditionUnknown,
			Reason:             "ActivePromotion",
			Message:            "Stage has a Promotion in progress",
			ObservedGeneration: stage.Generation,
		})
		newStatus.Health = &kargoapi.Health{
			Status: kargoapi.HealthStateUnknown,
			Issues: []string{
				"Cannot assess health because a Promotion is currently in progress",
			},
		}
		return newStatus
	}

	lastPromo := stage.Status.LastPromotion
	if lastPromo == nil {
		logger.Debug("Stage has no current Freight: no health checks to perform")
		conditions.Set(&newStatus, &metav1.Condition{
			Type:               kargoapi.ConditionTypeHealthy,
			Status:             metav1.ConditionUnknown,
			Reason:             "NoFreight",
			Message:            "Stage has no current Freight",
			ObservedGeneration: stage.Generation,
		})
		newStatus.Health = nil
		return newStatus
	}

	// If the last Promotion did not succeed, then we cannot perform any health
	// checks because they are only available after a successful Promotion.
	// Since we lack enough information to determine health, we mark it as Unknown.
	//
	// TODO(hidde): Long term, this should probably be changed to allow to
	//  continue to run health checks from the last successful Promotion,
	//  even if the current Promotion did not succeed (e.g. because it was
	//  aborted).
	if lastPromo.Status.Phase != kargoapi.PromotionPhaseSucceeded {
		logger.Debug("Last promotion did not succeed: defaulting Stage health to Unknown")
		conditions.Set(&newStatus, &metav1.Condition{
			Type:               kargoapi.ConditionTypeHealthy,
			Status:             metav1.ConditionUnknown,
			Reason:             fmt.Sprintf("LastPromotion%s", lastPromo.Status.Phase),
			Message:            "Cannot assess health because last Promotion did not succeed",
			ObservedGeneration: stage.Generation,
		})
		newStatus.Health = &kargoapi.Health{
			Status: kargoapi.HealthStateUnknown,
			Issues: []string{"Cannot assess health because last Promotion did not succeed"},
		}
		return newStatus
	}

	// Compose the health check criteria.
	healthChecks := lastPromo.Status.HealthChecks
	var criteria []health.Criteria
	for _, check := range healthChecks {
		criteria = append(criteria, health.Criteria{
			Kind:  check.Uses,
			Input: check.GetConfig(),
		})
	}

	// Run the hlth checks.
	hlth := r.healthChecker.Check(ctx, stage.Namespace, stage.Name, criteria)
	newStatus.Health = &hlth

	// Set the Healthy condition based on the health status.
	switch hlth.Status {
	case kargoapi.HealthStateHealthy:
		conditions.Set(&newStatus, &metav1.Condition{
			Type:               kargoapi.ConditionTypeHealthy,
			Status:             metav1.ConditionTrue,
			Reason:             string(hlth.Status),
			Message:            fmt.Sprintf("Stage is healthy (performed %d health checks)", len(healthChecks)),
			ObservedGeneration: stage.Generation,
		})
	case kargoapi.HealthStateUnhealthy:
		conditions.Set(&newStatus, &metav1.Condition{
			Type:   kargoapi.ConditionTypeHealthy,
			Status: metav1.ConditionFalse,
			Reason: string(hlth.Status),
			Message: fmt.Sprintf(
				"Stage is unhealthy (%d issues in %d health checks)",
				len(hlth.Issues), len(healthChecks),
			),
			ObservedGeneration: stage.Generation,
		})
	default:
		conditions.Set(&newStatus, &metav1.Condition{
			Type:               kargoapi.ConditionTypeHealthy,
			Status:             metav1.ConditionUnknown,
			Reason:             string(hlth.Status),
			ObservedGeneration: stage.Generation,
		})
	}

	return newStatus
}

// syncFreight ensures that all Freight statuses accurately reflect whether they
// are currently in use by the Stage.
func (r *RegularStageReconciler) syncFreight(ctx context.Context, stage *kargoapi.Stage) error {
	// Get the Stage's current FreightCollection.
	curFreight := stage.Status.FreightHistory.Current()
	// Find all Freight that think they're currently in use by this Stage.
	var freight []kargoapi.Freight
	freight, err := api.ListFreightByCurrentStage(ctx, r.client, stage)
	if err != nil {
		return err
	}
	// Step through all the Freight that think they're currently used by this
	// Stage and, if they're not, patch their status to accurately reflect that.
	for _, f := range freight {
		if !curFreight.Includes(f.Name) {
			newStatus := f.Status.DeepCopy()
			newStatus.RemoveCurrentStage(stage.Name)
			if err := kubeclient.PatchStatus(ctx, r.client, &f, func(status *kargoapi.FreightStatus) {
				*status = *newStatus
			}); err != nil {
				return fmt.Errorf(
					"error patching status of Freight %q in namespace %q: %w",
					f.Name, f.Namespace, err,
				)
			}
		}
	}
	// Iterate over every piece of Freight that the Stage is actually using to
	// make sure that their status accurately reflects that.
	//
	// This is the timestamp we'll use to track when the Freight came into use
	// by the Stage. There are edge cases where this won't be perfectly accurate,
	// but it's close enough, especially given that we use it for calculating
	// "soak time," which requires a certain MINIMUM amount of time to pass. Any
	// inaccuracy in the timestamp only means the Freight has actually "soaked"
	// LONGER than what we calculate.
	now := time.Now()
	for _, fr := range curFreight.References() {
		f, err := api.GetFreight(
			ctx,
			r.client,
			types.NamespacedName{
				Namespace: stage.Namespace,
				Name:      fr.Name,
			},
		)
		if err != nil {
			return fmt.Errorf(
				"error getting Freight %q in namespace %q: %w",
				fr.Name, stage.Namespace, err,
			)
		}
		if f == nil {
			// nolint:staticcheck
			return fmt.Errorf("Freight %q not found in namespace %q", fr.Name, stage.Namespace)
		}
		if !f.IsCurrentlyIn(stage.Name) {
			newStatus := f.Status.DeepCopy()
			newStatus.AddCurrentStage(stage.Name, now)
			if err = kubeclient.PatchStatus(ctx, r.client, f, func(status *kargoapi.FreightStatus) {
				*status = *newStatus
			}); err != nil {
				return fmt.Errorf(
					"error patching status of Freight %q in namespace %q: %w",
					f.Name, f.Namespace, err,
				)
			}
		}
	}
	return nil
}

// verifyStageFreight verifies the current Freight of a Stage. If the Stage has
// no current Freight, or the Freight has already been verified, then no action
// is taken. If the Freight has not been verified yet, then a new verification
// is started.
//
// An annotation can be set on the Stage to request the verification to be
// aborted. This is useful if the verification is taking too long, or if the
// Freight is no longer needed.
//
// In addition, an annotation can be set on the Stage to request re-verification
// of the Freight. This can be useful to ensure that the current Freight is
// still in a good state.
//
// When the Stage is unhealthy, or a Promotion is currently running, then the
// verification is skipped.
func (r *RegularStageReconciler) verifyStageFreight(
	ctx context.Context,
	stage *kargoapi.Stage,
	startTime time.Time,
	endTime func() time.Time,
) (newStatus kargoapi.StageStatus, err error) {
	logger := logging.LoggerFromContext(ctx)
	newStatus = *stage.Status.DeepCopy()

	// If there is no current Freight, then we have nothing to verify.
	curFreight := stage.Status.FreightHistory.Current()
	if curFreight == nil {
		logger.Debug("Stage has no current Freight: no verification to perform")
		conditions.Set(&newStatus, &metav1.Condition{
			Type:               kargoapi.ConditionTypeVerified,
			Status:             metav1.ConditionUnknown,
			Reason:             "NoFreight",
			Message:            "Stage has no current Freight to verify",
			ObservedGeneration: stage.Generation,
		})
		return newStatus, nil
	}

	// If we are currently promoting Freight, then we are not in a stable state
	// and should wait until the promotion is complete.
	if curPromotion := stage.Status.CurrentPromotion; curPromotion != nil {
		logger.Debug("Stage is currently promoting Freight: skipping verification")
		return newStatus, nil
	}

	defer func() {
		curFreight = newStatus.FreightHistory.Current()
		if curFreight == nil || len(curFreight.VerificationHistory) == 0 {
			return
		}

		for _, vi := range curFreight.VerificationHistory {
			if vi.Phase == kargoapi.VerificationPhaseSuccessful {
				// If the Freight has at least one successful verification,
				// then we can consider the Freight to be verified.
				conditions.Set(&newStatus, &metav1.Condition{
					Type:               kargoapi.ConditionTypeVerified,
					Status:             metav1.ConditionTrue,
					Reason:             "Verified",
					Message:            "Freight has been verified",
					ObservedGeneration: stage.Generation,
				})
				return
			}
		}

		// If the Freight has no successful verification, then we should look
		// for the most recent verification and set the status accordingly.
		lastVerification := curFreight.VerificationHistory.Current()
		if lastVerification != nil {
			switch lastVerification.Phase {
			case kargoapi.VerificationPhasePending:
				conditions.Set(&newStatus, &metav1.Condition{
					Type:               kargoapi.ConditionTypeVerified,
					Status:             metav1.ConditionUnknown,
					Reason:             "VerificationPending",
					Message:            "Freight is pending verification",
					ObservedGeneration: stage.Generation,
				})
			case kargoapi.VerificationPhaseRunning:
				conditions.Set(&newStatus, &metav1.Condition{
					Type:               kargoapi.ConditionTypeVerified,
					Status:             metav1.ConditionUnknown,
					Reason:             "VerificationRunning",
					Message:            "Freight is currently being verified",
					ObservedGeneration: stage.Generation,
				})
			case kargoapi.VerificationPhaseFailed, kargoapi.VerificationPhaseError:
				conditions.Set(&newStatus, &metav1.Condition{
					Type:               kargoapi.ConditionTypeVerified,
					Status:             metav1.ConditionFalse,
					Reason:             fmt.Sprintf("Verification%s", lastVerification.Phase),
					Message:            lastVerification.Message,
					ObservedGeneration: stage.Generation,
				})
			case kargoapi.VerificationPhaseAborted:
				conditions.Set(&newStatus, &metav1.Condition{
					Type:               kargoapi.ConditionTypeVerified,
					Status:             metav1.ConditionFalse,
					Reason:             "VerificationAborted",
					Message:            lastVerification.Message,
					ObservedGeneration: stage.Generation,
				})
			case kargoapi.VerificationPhaseInconclusive:
				conditions.Set(&newStatus, &metav1.Condition{
					Type:               kargoapi.ConditionTypeVerified,
					Status:             metav1.ConditionUnknown,
					Reason:             "VerificationInconclusive",
					Message:            lastVerification.Message,
					ObservedGeneration: stage.Generation,
				})
			default:
				conditions.Set(&newStatus, &metav1.Condition{
					Type:    kargoapi.ConditionTypeVerified,
					Status:  metav1.ConditionUnknown,
					Reason:  "UnknownVerificationPhase",
					Message: fmt.Sprintf("Freight verification is in an unknown phase: %s", lastVerification.Phase),
				})
			}
		}
	}()

	// Get the re-verification request, if any.
	reverifyReq, _ := api.ReverifyAnnotationValue(stage.GetAnnotations())

	// Check if the current Freight has already been verified.
	var newVI *kargoapi.VerificationInfo
	if lastVerification := curFreight.VerificationHistory.Current(); lastVerification != nil {
		// If the last verification is not terminal, then we should check if
		// we need to abort the verification, or if we need to get the verification
		// result.
		if !lastVerification.Phase.IsTerminal() {
			// Check if we need to abort the verification.
			abortReq, _ := api.AbortVerificationAnnotationValue(stage.GetAnnotations())
			if abortReq.ForID(lastVerification.ID) {
				logger.Debug("aborting verification of Stage Freight")

				// Abort the verification.
				newVI, err = r.abortVerification(ctx, *curFreight, abortReq)
				if newVI != nil {
					newStatus.FreightHistory.Current().VerificationHistory.UpdateOrPush(*newVI)
				}

				// Issue an event for the aborted verification.
				for _, ref := range curFreight.Freight {
					r.recordFreightVerificationEvent(stage, ref, newVI)
				}

				return newStatus, err
			}

			// Get the latest result of the verification.
			newVI, err = r.getVerificationResult(ctx, *curFreight)
			if newVI != nil {
				newStatus.FreightHistory.Current().VerificationHistory.UpdateOrPush(*newVI)

				// If the verification is terminal, we should issue an event for
				// each Freight that was verified.
				if newVI.Phase.IsTerminal() {
					for _, ref := range curFreight.Freight {
						r.recordFreightVerificationEvent(stage, ref, newVI)
					}
				}
			}
			return newStatus, err
		}

		// If the last verification is terminal, and we are not re-verifying
		// the Freight, then we have nothing to do.
		if !reverifyReq.ForID(lastVerification.ID) {
			logger.Debug("Stage Freight has already been verified")
			return newStatus, nil
		}
	}

	// If the Stage is not passed any health checks (yet), then we should not
	// verify the Freight.
	if stage.Status.Health == nil || stage.Status.Health.Status != kargoapi.HealthStateHealthy {
		logger.Debug("Stage has not passed health checks: skipping verification")
		return newStatus, nil
	}

	// If we have no specific verification configuration, then we can mark the
	// verification as successful.
	if stage.Spec.Verification == nil {
		newVI := kargoapi.VerificationInfo{
			ID:         uuid.NewString(),
			StartTime:  ptr.To(metav1.NewTime(startTime)),
			FinishTime: ptr.To(metav1.NewTime(endTime())),
			Phase:      kargoapi.VerificationPhaseSuccessful,
		}
		newStatus.FreightHistory.Current().VerificationHistory.UpdateOrPush(newVI)

		// Issue an event for each Freight that was verified.
		for _, ref := range curFreight.Freight {
			r.recordFreightVerificationEvent(stage, ref, &newVI)
		}
		return newStatus, nil
	}

	// Start a new (re-)verification.
	newVI, err = r.startVerification(ctx, stage, *curFreight, reverifyReq, startTime)
	if newVI != nil {
		newStatus.FreightHistory.Current().VerificationHistory.UpdateOrPush(*newVI)

		// There is a chance for the verification to be terminal immediately
		// after starting it. For example, if the rollouts integration is not
		// enabled. In this case, we should issue an event for the verification.
		if newVI.Phase.IsTerminal() {
			for _, ref := range curFreight.Freight {
				r.recordFreightVerificationEvent(stage, ref, newVI)
			}
		}
	}
	return newStatus, err
}

// markFreightVerifiedForStage marks the Freight that is associated with the
// Stage as verified. If the Freight has already been verified, then no action
// is taken.
func (r *RegularStageReconciler) markFreightVerifiedForStage(
	ctx context.Context,
	stage *kargoapi.Stage,
) (kargoapi.StageStatus, error) {
	logger := logging.LoggerFromContext(ctx)
	newStatus := *stage.Status.DeepCopy()

	// If the Stage is unhealthy, then we should not verify the Freight.
	if stage.Status.Health == nil || stage.Status.Health.Status != kargoapi.HealthStateHealthy {
		return newStatus, nil
	}

	// If there is no current Freight, or the Stage has not been verified yet
	// after the last Promotion, then we are not ready to verify the Freight.
	curFreight := stage.Status.FreightHistory.Current()
	if curFreight == nil ||
		len(curFreight.VerificationHistory) == 0 ||
		curFreight.HasNonTerminalVerification() ||
		curFreight.VerificationHistory.Current().Phase != kargoapi.VerificationPhaseSuccessful {
		return newStatus, nil
	}

	// At this point, all preconditions for verifying the Freight have been met,
	// and we can proceed with the verification.
	for _, ref := range curFreight.Freight {
		freight := &kargoapi.Freight{}
		if err := r.client.Get(ctx, types.NamespacedName{
			Namespace: stage.Namespace,
			Name:      ref.Name,
		}, freight); err != nil {
			return newStatus, fmt.Errorf(
				"error getting Freight %q in namespace %q: %w",
				ref.Name, stage.Namespace, err,
			)
		}

		// If the Freight has already been verified, then there is no need to
		// verify it again.
		if freight.IsVerifiedIn(stage.Name) {
			logger.Debug("Freight has already been verified in Stage")
			continue
		}

		// Verify the Freight.
		if err := kubeclient.PatchStatus(ctx, r.client, freight, func(status *kargoapi.FreightStatus) {
			if status.VerifiedIn == nil {
				status.VerifiedIn = make(map[string]kargoapi.VerifiedStage)
			}
			status.AddVerifiedStage(stage.Name, curFreight.VerificationHistory.Current().FinishTime.Time)
		}); err != nil {
			return newStatus, fmt.Errorf(
				"error marking Freight %q as verified in Stage: %w",
				freight.Name, err,
			)
		}
		logger.Debug("marked Freight as verified in Stage", "freight", freight.Name)
	}

	return newStatus, nil
}

// recordFreightVerificationEvent records an event for the verification of a
// Freight. The event contains information about the Freight, the verification,
// and the Stage that triggered the verification.
func (r *RegularStageReconciler) recordFreightVerificationEvent(
	stage *kargoapi.Stage,
	freightRef kargoapi.FreightReference,
	vi *kargoapi.VerificationInfo,
) {
	freight := &kargoapi.Freight{}
	if err := r.client.Get(context.Background(), types.NamespacedName{
		Namespace: stage.Namespace,
		Name:      freightRef.Name,
	}, freight); err != nil {
		logging.LoggerFromContext(context.Background()).Error(
			err, "failed to get Freight for verification event",
			"freight", freightRef.Name,
		)
		return
	}

	var analysisTriggeredByPromotion *string
	// Extract metadata from the AnalysisRun if available
	if vi.HasAnalysisRun() {
		ar := &rolloutsapi.AnalysisRun{}
		if err := r.client.Get(context.Background(), types.NamespacedName{
			Namespace: vi.AnalysisRun.Namespace,
			Name:      vi.AnalysisRun.Name,
		}, ar); err != nil {
			// Log the error but do not fail the event recording.
			logging.LoggerFromContext(context.Background()).Error(
				err, "failed to get AnalysisRun for verification event",
				"analysisRun", vi.AnalysisRun.Name, "freight", freightRef.Name,
			)
		}
		// AnalysisRun that triggered by a Promotion contains the Promotion name
		if promoName, ok := ar.Annotations[kargoapi.AnnotationKeyPromotion]; ok {
			analysisTriggeredByPromotion = &promoName
		}
	}

	evtActor := api.FormatEventControllerActor(r.cfg.Name())

	// If the verification is manually triggered (e.g. reverify),
	// override the actor with the one who triggered the verification.
	if vi.Actor != "" {
		evtActor = vi.Actor
	}

	var evt kargoEvent.FreightVerificationEventMeta

	switch vi.Phase {
	case kargoapi.VerificationPhaseSuccessful:
		e := kargoEvent.NewFreightVerificationSucceeded(evtActor, stage.Name, freight, vi)
		e.Message = "Freight verification succeeded"
		evt = e
	case kargoapi.VerificationPhaseFailed:
		evt = kargoEvent.NewFreightVerificationFailed(evtActor, stage.Name, freight, vi)
	case kargoapi.VerificationPhaseError:
		evt = kargoEvent.NewFreightVerificationErrored(evtActor, stage.Name, freight, vi)
	case kargoapi.VerificationPhaseAborted:
		evt = kargoEvent.NewFreightVerificationAborted(evtActor, stage.Name, freight, vi)
	case kargoapi.VerificationPhaseInconclusive:
		evt = kargoEvent.NewFreightVerificationInconclusive(evtActor, stage.Name, freight, vi)
	default:
		evt = kargoEvent.NewFreightVerificationUnknown(evtActor, stage.Name, freight, vi)
	}

	evt.SetTriggeredByPromotion(analysisTriggeredByPromotion)

	if err := r.eventSender.Send(context.Background(), evt); err != nil {
		logging.LoggerFromContext(context.Background()).Error(
			err, "failed to send verification event",
			"freight", freightRef.Name,
		)
	}
}

// startVerification starts a new verification for the Freight that is associated
// with the Stage. If the Freight has already been verified, then no verification
// is started unless a re-verification is requested.
//
// If there is no verification configuration for the Stage, then the verification
// is automatically considered successful and no verification is started.
//
// If the Rollouts integration is disabled, then the verification is marked as
// failed with an appropriate message.
//
// To start a verification, the Stage must be healthy.
func (r *RegularStageReconciler) startVerification(
	ctx context.Context,
	stage *kargoapi.Stage,
	freight kargoapi.FreightCollection,
	req *kargoapi.VerificationRequest,
	startTime time.Time,
) (*kargoapi.VerificationInfo, error) {
	newVI := &kargoapi.VerificationInfo{
		ID:        uuid.NewString(),
		StartTime: &metav1.Time{Time: startTime},
	}

	// If we have a verification request, we should enrich the information
	// with the actor who requested the verification.
	curVI := freight.VerificationHistory.Current()
	if curVI != nil && req.ForID(curVI.ID) {
		newVI.Actor = req.Actor
	}

	// Return early, as we cannot start the verification if the Rollouts
	// integration is disabled.
	if !r.cfg.RolloutsIntegrationEnabled {
		newVI.FinishTime = ptr.To(metav1.Now())
		newVI.Phase = kargoapi.VerificationPhaseError
		newVI.Message = "Rollouts integration is disabled on this controller: cannot start verification"
		return newVI, nil
	}

	logger := logging.LoggerFromContext(ctx)

	// If this is not a re-verification request, check if there is an existing
	// AnalysisRun for the Stage and Freight. If there is, return the status
	// of the existing AnalysisRun.
	if req == nil {
		existingAnalysisRun, err := r.findExistingAnalysisRun(ctx, types.NamespacedName{
			Namespace: stage.Namespace,
			Name:      stage.Name,
		}, freight.ID)
		if err != nil {
			newVI.FinishTime = ptr.To(metav1.Now())
			newVI.Phase = kargoapi.VerificationPhaseError
			newVI.Message = err.Error()
			return newVI, nil
		}

		if existingAnalysisRun != nil {
			logger.Debug("AnalysisRun already exists for FreightCollection")

			newVI.FinishTime = existingAnalysisRun.Status.CompletedAt()
			newVI.Phase = kargoapi.VerificationPhase(existingAnalysisRun.Status.Phase)
			newVI.AnalysisRun = &kargoapi.AnalysisRunReference{
				Name:      existingAnalysisRun.Name,
				Namespace: existingAnalysisRun.Namespace,
				Phase:     string(existingAnalysisRun.Status.Phase),
			}
			newVI.FinishTime = existingAnalysisRun.Status.CompletedAt()
			return newVI, nil
		}
	}

	// At this point, we know that we need to start a new AnalysisRun for the
	// verification.
	shortStageName := kubernetes.ShortenLabelValue(stage.Name)
	builder := rollouts.NewAnalysisRunBuilder(r.client, rollouts.Config{
		ControllerInstanceID: r.cfg.RolloutsControllerInstanceID,
	})
	builderOpts := []rollouts.AnalysisRunOption{
		rollouts.WithNamePrefix(stage.Name),
		rollouts.WithNameSuffix(freight.ID),
		rollouts.WithExtraLabels(map[string]string{
			kargoapi.LabelKeyStage:             shortStageName,
			kargoapi.LabelKeyFreightCollection: freight.ID,
		}),
		rollouts.WithArgumentEvaluationConfig{
			Env: map[string]any{
				"ctx": map[string]any{
					"project": stage.Namespace,
					"stage":   stage.Name,
				},
			},
			Options: slices.Concat(
				exprfn.DataOperations(
					ctx,
					r.client,
					gocache.New(gocache.NoExpiration, gocache.NoExpiration),
					stage.Namespace,
				),
				exprfn.FreightOperations(
					ctx,
					r.client,
					stage.Namespace,
					stage.Spec.RequestedFreight,
					freight.References(),
				),
				exprfn.UtilityOperations(),
			),
			Vars: stage.Spec.Vars,
		},
	}

	if stage.Name != shortStageName {
		builderOpts = append(builderOpts, rollouts.WithExtraAnnotations(map[string]string{
			kargoapi.AnnotationKeyStage: stage.Name,
		}))
	}

	for _, freightRef := range freight.Freight {
		builderOpts = append(builderOpts, rollouts.WithOwner{
			APIVersion: kargoapi.GroupVersion.String(),
			Kind:       "Freight",
			Reference:  types.NamespacedName{Namespace: stage.Namespace, Name: freightRef.Name},
		})
	}
	if curVI == nil || (req.ForID(curVI.ID) && req.ControlPlane && req.Actor != "") {
		if stage.Status.LastPromotion != nil {
			builderOpts = append(builderOpts, rollouts.WithExtraAnnotations{
				kargoapi.AnnotationKeyPromotion: stage.Status.LastPromotion.Name,
			})
		}
	}
	ar, err := builder.Build(ctx, stage.Namespace, stage.Spec.Verification, builderOpts...)
	if err != nil {
		newVI.FinishTime = ptr.To(metav1.Now())
		newVI.Phase = kargoapi.VerificationPhaseError
		newVI.Message = fmt.Errorf(
			"error building AnalysisRun for Stage %q and Freight collection %q in namespace %q: %w",
			stage.Name,
			freight.ID,
			stage.Namespace,
			err,
		).Error()
		return newVI, nil
	}
	if err = r.client.Create(ctx, ar); err != nil {
		newVI.FinishTime = ptr.To(metav1.Now())
		newVI.Phase = kargoapi.VerificationPhaseError
		newVI.Message = fmt.Errorf(
			"error creating AnalysisRun %q in namespace %q: %w",
			ar.Name,
			ar.Namespace,
			err,
		).Error()
		return newVI, kubeclient.IgnoreInvalid(err) // Ignore errors which are due to validation issues
	}

	// Mark the verification as pending.
	newVI.FinishTime = ptr.To(ar.CreationTimestamp)
	newVI.Phase = kargoapi.VerificationPhasePending
	newVI.AnalysisRun = &kargoapi.AnalysisRunReference{
		Name:      ar.Name,
		Namespace: ar.Namespace,
		Phase:     string(ar.Status.Phase),
	}
	return newVI, nil
}

// getVerificationResult gets the result of the verification for the current
// Freight of a Stage.
//
// If the Stage does not have an AnalysisRun associated with the verification,
// an error is returned.
//
// If the Rollouts integration is disabled, then the verification is marked as
// failed with an appropriate message.
func (r *RegularStageReconciler) getVerificationResult(
	ctx context.Context,
	freight kargoapi.FreightCollection,
) (*kargoapi.VerificationInfo, error) {
	// Ensure all necessary information is available to get the verification.
	currentVI := freight.VerificationHistory.Current()
	if currentVI == nil {
		return nil, fmt.Errorf("no current verification info for Freight collection %q", freight.ID)
	}
	if currentVI.AnalysisRun == nil {
		return nil, fmt.Errorf(
			"no AnalysisRun reference in current verification info for Freight collection %q",
			freight.ID,
		)
	}

	// If the Rollouts integration is disabled, then we cannot get the
	// verification.
	if !r.cfg.RolloutsIntegrationEnabled {
		return &kargoapi.VerificationInfo{
			ID:         currentVI.ID,
			StartTime:  currentVI.StartTime,
			FinishTime: ptr.To(metav1.Now()),
			Phase:      kargoapi.VerificationPhaseError,
			Message:    "Rollouts integration is disabled on this controller: cannot get verification result",
		}, nil
	}

	// TODO(hidde): This retry logic has been put in place because we have
	// observed the cache not being up-to-date with the API server in some
	// edge case scenarios. While this is not a long-term solution, it cures
	// the symptoms for now. We should investigate the root cause of this
	// issue and remove this retry logic when the root cause has been resolved.
	ar := rolloutsapi.AnalysisRun{}
	if err := retry.OnError(r.backoffCfg, func(err error) bool {
		return apierrors.IsNotFound(err)
	}, func() error {
		return r.client.Get(ctx, types.NamespacedName{
			Namespace: currentVI.AnalysisRun.Namespace,
			Name:      currentVI.AnalysisRun.Name,
		}, &ar)
	}); err != nil {
		return &kargoapi.VerificationInfo{
			ID:         currentVI.ID,
			Actor:      currentVI.Actor,
			StartTime:  currentVI.StartTime,
			FinishTime: currentVI.FinishTime,
			Phase:      kargoapi.VerificationPhaseError,
			Message: fmt.Errorf(
				"error getting AnalysisRun %q in namespace %q: %w",
				currentVI.AnalysisRun.Name,
				currentVI.AnalysisRun.Namespace,
				err,
			).Error(),
			AnalysisRun: currentVI.AnalysisRun.DeepCopy(),
		}, err
	}

	// Return a new VerificationInfo with the same ID and the information from
	// the current state of the AnalysisRun.
	return &kargoapi.VerificationInfo{
		ID:         currentVI.ID,
		Actor:      currentVI.Actor,
		StartTime:  currentVI.StartTime,
		FinishTime: ar.Status.CompletedAt(),
		Phase:      kargoapi.VerificationPhase(ar.Status.Phase),
		Message:    ar.Status.Message,
		AnalysisRun: &kargoapi.AnalysisRunReference{
			Name:      ar.Name,
			Namespace: ar.Namespace,
			Phase:     string(ar.Status.Phase),
		},
	}, nil
}

// abortVerification aborts the verification for the current Freight of a Stage.
func (r *RegularStageReconciler) abortVerification(
	ctx context.Context,
	freight kargoapi.FreightCollection,
	req *kargoapi.VerificationRequest,
) (*kargoapi.VerificationInfo, error) {
	// Ensure all necessary information is available to abort the verification.
	currentVI := freight.VerificationHistory.Current()
	if currentVI == nil {
		return nil, fmt.Errorf("no current verification info for Freight collection %q", freight.ID)
	}
	if currentVI.AnalysisRun == nil {
		return nil, fmt.Errorf(
			"no AnalysisRun reference in current verification info for Freight collection %q",
			freight.ID,
		)
	}

	// If the current verification is already terminal, then there is no need
	// to abort it.
	if currentVI.Phase.IsTerminal() {
		return currentVI, nil
	}

	// Determine the actor who requested the abort.
	actor := currentVI.Actor
	if req.ForID(currentVI.ID) {
		actor = req.Actor
	}

	// If the Rollouts integration is disabled, then we cannot abort the
	// verification.
	if !r.cfg.RolloutsIntegrationEnabled {
		return &kargoapi.VerificationInfo{
			ID:          currentVI.ID,
			Actor:       actor,
			StartTime:   currentVI.StartTime,
			FinishTime:  ptr.To(metav1.Now()),
			Phase:       kargoapi.VerificationPhaseError,
			Message:     "Rollouts integration is disabled on this controller: cannot abort verification",
			AnalysisRun: currentVI.AnalysisRun.DeepCopy(),
		}, nil
	}

	// Patch the AnalysisRun to request the abort.
	ar := &rolloutsapi.AnalysisRun{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: currentVI.AnalysisRun.Namespace,
			Name:      currentVI.AnalysisRun.Name,
		},
	}
	if err := r.client.Patch(
		ctx,
		ar,
		client.RawPatch(types.MergePatchType, []byte(`{"spec":{"terminate":true}}`)),
	); err != nil {
		// TODO(hidde): we should consider better error handling here to e.g.
		// retry an abort request when the Kubernetes API server is under heavy
		// load.
		return &kargoapi.VerificationInfo{
			ID:         currentVI.ID,
			Actor:      actor,
			StartTime:  currentVI.StartTime,
			FinishTime: ptr.To(metav1.Now()),
			Phase:      kargoapi.VerificationPhaseError,
			Message: fmt.Errorf(
				"error terminating AnalysisRun %q in namespace %q: %w", ar.Name, ar.Namespace, err,
			).Error(),
			AnalysisRun: currentVI.AnalysisRun.DeepCopy(),
		}, nil
	}

	// Return a new VerificationInfo with the same ID and a message indicating
	// that the verification was aborted. The Phase will be set to Failed, as
	// the verification was not successful.
	// We do not use the further information from the AnalysisRun, as this
	// will indicate a "Succeeded" phase due to Argo Rollouts behavior.
	return &kargoapi.VerificationInfo{
		ID:          currentVI.ID,
		Actor:       actor,
		StartTime:   currentVI.StartTime,
		FinishTime:  ptr.To(metav1.Now()),
		Phase:       kargoapi.VerificationPhaseFailed,
		Message:     "Verification aborted by user",
		AnalysisRun: currentVI.AnalysisRun.DeepCopy(),
	}, nil
}

// findExistingAnalysisRun finds the most recent AnalysisRun for a Stage and
// Freight collection in the namespace of the Stage. If no AnalysisRun is found,
// it returns nil.
func (r *RegularStageReconciler) findExistingAnalysisRun(
	ctx context.Context,
	stage types.NamespacedName,
	freightColID string,
) (*rolloutsapi.AnalysisRun, error) {
	analysisRuns := &rolloutsapi.AnalysisRunList{}
	if err := r.client.List(
		ctx,
		analysisRuns,
		client.InNamespace(stage.Namespace),
		client.MatchingLabelsSelector{
			Selector: labels.SelectorFromSet(map[string]string{
				kargoapi.LabelKeyStage:             kubernetes.ShortenLabelValue(stage.Name),
				kargoapi.LabelKeyFreightCollection: freightColID,
			}),
		},
	); err != nil {
		return nil, fmt.Errorf(
			"error listing AnalysisRuns for Stage %q and Freight collection %q in namespace %q: %w",
			stage.Name, freightColID, stage.Namespace, err,
		)
	}

	if len(analysisRuns.Items) == 0 {
		return nil, nil
	}

	// Sort the AnalysisRuns by creation timestamp, so that the most recent
	// one is first.
	slices.SortFunc(analysisRuns.Items, func(lhs, rhs rolloutsapi.AnalysisRun) int {
		return rhs.CreationTimestamp.Compare(lhs.CreationTimestamp.Time)
	})
	return &analysisRuns.Items[0], nil
}

// autoPromoteFreight automatically promotes the latest promotable (i.e.
// verified) Freight for a Stage if auto-promotion is allowed (see
// autoPromotionAllowed).
func (r *RegularStageReconciler) autoPromoteFreight(
	ctx context.Context,
	stage *kargoapi.Stage,
) (kargoapi.StageStatus, error) {
	logger := logging.LoggerFromContext(ctx)
	newStatus := *stage.Status.DeepCopy()

	// If the Stage has no requested Freight, then there is nothing to promote.
	// NB: This should not happen in practice, as a Stage cannot exist without
	// requested Freight.
	if len(stage.Spec.RequestedFreight) == 0 {
		return newStatus, nil
	}

	// Confirm that auto-promotion is allowed for the Stage.
	if autoPromotionAllowed, err := r.autoPromotionAllowed(ctx, stage.ObjectMeta); err != nil || !autoPromotionAllowed {
		newStatus.AutoPromotionEnabled = false
		return newStatus, err
	}
	newStatus.AutoPromotionEnabled = true

	// Retrieve promotable Freight for the Stage.
	promotableFreight, err := r.getPromotableFreight(ctx, stage)
	if err != nil {
		return newStatus, err
	}

	// If the Stage has no current Freight, then we can promote any available
	currentFreight := newStatus.FreightHistory.Current()

	// Check if there is any new Freight which can be auto-promoted.
	for _, req := range stage.Spec.RequestedFreight {
		origin := req.Origin.String()
		freight, exists := promotableFreight[origin]
		if !exists || len(freight) == 0 {
			logger.Debug("no Freight from origin available for auto-promotion", "origin", origin)
			continue
		}

		// Quick sanity check...
		if req.Sources.AutoPromotionOptions != nil &&
			req.Sources.AutoPromotionOptions.SelectionPolicy == kargoapi.AutoPromotionSelectionPolicyMatchUpstream &&
			len(freight) > 1 {
			return newStatus, fmt.Errorf(
				"unexpectedly found %d available Freight running immediately "+
					"upstream from Stage %q in namespace %q; this should not be "+
					"possible",
				len(freight), stage.Name, stage.Namespace,
			)
		}

		// Find the latest Freight by sorting the available Freight by creation time
		// in descending order.
		slices.SortFunc(freight, func(lhs, rhs kargoapi.Freight) int {
			return rhs.CreationTimestamp.Compare(lhs.CreationTimestamp.Time)
		})
		latestFreight := freight[0]

		freightLogger := logger.WithValues("origin", origin, "freight", latestFreight.Name)

		// Only proceed if the latest available Freight is different from the
		// current Freight in the Stage.
		if currentFreight != nil && len(currentFreight.Freight) > 0 {
			if freightRef, ok := currentFreight.Freight[origin]; ok && freightRef.Name == latestFreight.Name {
				freightLogger.Debug("Stage already has latest available Freight for origin")
				continue
			}
		}

		// How we proceed depends on the auto-promotion Freight selection policy...
		if req.Sources.AutoPromotionOptions != nil &&
			req.Sources.AutoPromotionOptions.SelectionPolicy == kargoapi.AutoPromotionSelectionPolicyMatchUpstream {
			// With the MatchUpstream policy, we want to allow for the possibility of
			// a Stage needing to RETURN to a state it's previously been in. The only
			// conditions under which we WON'T proceed with auto-promotion are if
			// there's already a Promotion like the one we'd create that is not
			// terminal (e.g. Pending or Running) OR the last terminal Promotion like
			// the one we'd create wasn't successful. Not creating a new Promotion
			// under these conditions avoids the formation of an infinite loop of new
			// Promotions.
			//
			// If a non-terminal Promotion already exists for this Stage and Freight,
			// then we should not create a new one.
			promotions := &kargoapi.PromotionList{}
			if err = r.client.List(
				ctx,
				promotions,
				client.InNamespace(stage.Namespace),
				client.MatchingFieldsSelector{
					Selector: fields.AndSelectors(
						fields.OneTermEqualSelector(
							indexer.PromotionsByStageAndFreightField,
							indexer.StageAndFreightKey(stage.Name, latestFreight.Name),
						),
						fields.OneTermEqualSelector(
							indexer.PromotionsByTerminalField,
							"false",
						),
					),
				},
				client.Limit(1),
			); err != nil {
				return newStatus, fmt.Errorf(
					"error listing existing non-terminal Promotions for Freight %q "+
						"in namespace %q: %w",
					latestFreight.Name, stage.Namespace, err,
				)
			}
			if len(promotions.Items) > 0 {
				freightLogger.Debug("at least one non-terminal Promotion already " +
					"exists for Stage and Freight")
				continue
			}
			// If the most recent terminal Promotion for this Stage and Freight wasn't
			// successful, then we should not create a new one -- otherwise we're
			// likely to end up in an infinite loop of failed auto-promotions.
			promotions = &kargoapi.PromotionList{}
			if err = r.client.List(
				ctx,
				promotions,
				client.InNamespace(stage.Namespace),
				client.MatchingFieldsSelector{
					Selector: fields.AndSelectors(
						fields.OneTermEqualSelector(
							indexer.PromotionsByStageAndFreightField,
							indexer.StageAndFreightKey(stage.Name, latestFreight.Name),
						),
						fields.OneTermEqualSelector(
							indexer.PromotionsByTerminalField,
							"true",
						),
					),
				},
				client.Limit(1),
			); err != nil {
				return newStatus, fmt.Errorf(
					"error listing existing terminal Promotions for Freight %q in "+
						"namespace %q: %w",
					latestFreight.Name, stage.Namespace, err,
				)
			}
			if len(promotions.Items) > 0 {
				// Sort the terminal Promotions by creation time in descending order
				slices.SortFunc(promotions.Items, func(lhs, rhs kargoapi.Promotion) int {
					return rhs.CreationTimestamp.Compare(lhs.CreationTimestamp.Time)
				})
				newestPromotion := promotions.Items[0]
				if newestPromotion.Status.Phase != kargoapi.PromotionPhaseSucceeded {
					freightLogger.Debug(
						"most recent terminal Promotion for Stage and Freight was not "+
							"successful; skipping auto-promotion to avoid an infinite loop",
						"lastPromotion", newestPromotion.Name,
						"lastPromotionPhase", newestPromotion.Status.Phase,
					)
					continue
				}
			}
		} else {
			// If we get to here, auto-promotion is based on promoting the newest
			// Freight. We want to avoid the scenario where we RE-auto-promote the
			// newest Freight AFTER a user has manually re-promoted the Stage to some
			// older state. The check we'll perform is more basic than in the case of
			// the MatchUpstream policy. If ANY Promotion already exists referencing
			// the same Stage and Freight as the one we'd create, it means this Stage
			// has seen this Freight before and it shouldn't be RE-auto-promoted to
			// it. (Freight that's actually NEW and never before promoted to this
			// Stage will still make it through.)
			//
			// TODO(krancour): After #3016 is addressed, we can probably rely on
			// "pinning" older Freight to a Stage to prevent a user's will from being
			// preempted by auto-promotion of the newest Freight, and then the less
			// stringent check used above for the MatchUpstream selection policy
			// will probably suffice for this case as well.
			//
			// See: https://github.com/akuity/kargo/issues/3016
			promotions := &kargoapi.PromotionList{}
			if err = r.client.List(
				ctx,
				promotions,
				client.InNamespace(stage.Namespace),
				client.MatchingFieldsSelector{
					Selector: fields.OneTermEqualSelector(
						indexer.PromotionsByStageAndFreightField,
						indexer.StageAndFreightKey(stage.Name, latestFreight.Name),
					),
				},
				client.Limit(1),
			); err != nil {
				return newStatus, fmt.Errorf(
					"error listing existing Promotions for Freight %q in namespace %q: %w",
					latestFreight.Name, stage.Namespace, err,
				)
			}
			if len(promotions.Items) > 0 {
				freightLogger.Debug("Promotion already exists for Freight")
				continue
			}
		}

		// Auto promote the latest available Freight and record an event.
		promotion, err := kargo.NewPromotionBuilder(r.client).
			Build(ctx, *stage, latestFreight.Name)
		if err != nil {
			return newStatus, fmt.Errorf(
				"error building Promotion for Freight %q in namespace %q: %w",
				latestFreight.Name, stage.Namespace, err,
			)
		}
		if err = r.client.Create(ctx, promotion); err != nil {
			return newStatus, fmt.Errorf(
				"error creating Promotion for Freight %q in namespace %q: %w",
				latestFreight.Name, stage.Namespace, err,
			)
		}
		evt := kargoEvent.NewPromotionCreated(
			fmt.Sprintf("Automatically promoted Freight from origin %q for Stage %q",
				origin,
				promotion.Spec.Stage),
			api.FormatEventControllerActor(r.cfg.Name()),
			promotion,
			&latestFreight,
		)
		if err := r.eventSender.Send(ctx, evt); err != nil {
			logger.Error(err, "failed to send promotion event")
		}
		logger.Debug(
			"created Promotion resource",
			"promotion", promotion.Name,
		)
	}

	return newStatus, nil
}

// autoPromotionAllowed checks if auto-promotion is allowed for the given Stage.
func (r *RegularStageReconciler) autoPromotionAllowed(
	ctx context.Context,
	stage metav1.ObjectMeta,
) (bool, error) {
	logger := logging.LoggerFromContext(ctx)

	projectCfg := &kargoapi.ProjectConfig{}
	if err := r.client.Get(ctx, types.NamespacedName{
		Name:      stage.Namespace,
		Namespace: stage.Namespace,
	}, projectCfg); err != nil {
		if apierrors.IsNotFound(err) {
			logger.Debug("found no ProjectConfig associated with Project; auto-promotion is disabled")
			return false, nil
		}
		return false, fmt.Errorf("error getting ProjectConfig for Project %q: %w", stage.Namespace, err)
	}

	if len(projectCfg.Spec.PromotionPolicies) == 0 {
		logger.Debug("found no PromotionPolicy associated with Stage")
		return false, nil
	}

	for _, policy := range projectCfg.Spec.PromotionPolicies {
		if policy.StageSelector == nil {
			// Maintain backward compatibility with older versions of the
			// PromotionPolicy where the selector was not available.
			policy.StageSelector = &kargoapi.PromotionPolicySelector{
				Name: policy.Stage, // nolint:staticcheck
			}
		}

		// Match the Stage name with the exact PromotionPolicy name
		if nameSelector := policy.StageSelector.Name; nameSelector != "" {
			m, err := pattern.ParseNamePattern(nameSelector)
			if err != nil {
				return false, fmt.Errorf("error parsing PromotionPolicy name pattern %q: %w", nameSelector, err)
			}
			if !m.Matches(stage.Name) {
				continue
			}
		}

		// Match the Stage labels with the PromotionPolicy label selector.
		if labelSelector := policy.StageSelector.LabelSelector; labelSelector != nil {
			s, err := metav1.LabelSelectorAsSelector(labelSelector)
			if err != nil {
				return false, fmt.Errorf("error parsing PromotionPolicy label selector %q: %w", labelSelector, err)
			}
			if !s.Matches(labels.Set(stage.Labels)) {
				continue
			}
		}

		// If we reach this point, we have found a matching PromotionPolicy.
		logger.Debug(
			"found PromotionPolicy associated with Stage",
			"autoPromotionEnabled", policy.AutoPromotionEnabled,
		)
		return policy.AutoPromotionEnabled, nil
	}

	logger.Debug("found no PromotionPolicy associated with Stage")
	return false, nil
}

// getPromotableFreight retrieves a map of []Freight promotable to the specified
// Stage, indexed by origin.
func (r *RegularStageReconciler) getPromotableFreight(
	ctx context.Context,
	stage *kargoapi.Stage,
) (map[string][]kargoapi.Freight, error) {
	availableFreight, err := api.ListFreightAvailableToStage(ctx, r.client, stage)
	if err != nil {
		return nil, fmt.Errorf(
			"error listing available Freight for Stage %q: %w",
			stage.Name, err,
		)
	}

	promotableFreight := make(map[string][]kargoapi.Freight)
	for _, freight := range availableFreight {
		originID := freight.Origin.String()
		if _, ok := promotableFreight[originID]; !ok {
			promotableFreight[originID] = []kargoapi.Freight{freight}
		} else {
			promotableFreight[originID] = append(promotableFreight[originID], freight)
		}
	}

	return promotableFreight, nil
}

// handleDelete handles the deletion of the given Stage. It clears the
// verification status of all Freight that have been verified in the Stage, the
// approval status of all Freight that have been approved for the Stage, and
// deletes all AnalysisRuns that are associated with the Stage.
//
// It returns an error aggregate of all errors that occurred during the deletion
// process.
func (r *RegularStageReconciler) handleDelete(ctx context.Context, stage *kargoapi.Stage) error {
	// If the Stage does not have the finalizer, there is nothing to do.
	if !controllerutil.ContainsFinalizer(stage, kargoapi.FinalizerName) {
		return nil
	}

	// Clear the verification and approval status of all Freight that have been
	// verified or approved for the Stage, and delete all AnalysisRuns.
	toClear := []func(context.Context, *kargoapi.Stage) error{
		r.clearVerifications,
		r.clearApprovals,
		r.clearAnalysisRuns,
	}
	var errs []error
	for _, c := range toClear {
		if err := c(ctx, stage); err != nil {
			errs = append(errs, err)
		}
	}
	if err := kerrors.Flatten(kerrors.NewAggregate(errs)); err != nil {
		// We ran into an error, return it before proceeding with removing the
		// finalizer.
		return fmt.Errorf("error handling deletion of Stage: %w", err)
	}

	// Remove the finalizer from the Stage.
	if err := api.RemoveFinalizer(ctx, r.client, stage); err != nil {
		return fmt.Errorf("error removing finalizer from Stage: %w", err)
	}

	return nil
}

// clearVerifications clears the verification status of all Freight that have
// been verified in the given Stage. It removes the Stage from the VerifiedIn
// map of each Freight.
func (r *RegularStageReconciler) clearVerifications(ctx context.Context, stage *kargoapi.Stage) error {
	verified := kargoapi.FreightList{}
	if err := r.client.List(
		ctx,
		&verified,
		client.InNamespace(stage.Namespace),
		client.MatchingFieldsSelector{
			Selector: fields.OneTermEqualSelector(
				indexer.FreightByVerifiedStagesField,
				stage.Name,
			),
		},
	); err != nil {
		return fmt.Errorf(
			"error listing Freight verified in Stage %q in namespace %q: %w",
			stage.Name,
			stage.Namespace,
			err,
		)
	}

	var errs []error
	for _, f := range verified.Items {
		newStatus := *f.Status.DeepCopy()
		if newStatus.VerifiedIn == nil {
			continue
		}
		delete(newStatus.VerifiedIn, stage.Name)

		if err := kubeclient.PatchStatus(ctx, r.client, &f, func(status *kargoapi.FreightStatus) {
			*status = newStatus
		}); client.IgnoreNotFound(err) != nil {
			errs = append(errs, fmt.Errorf(
				"error clearing verification status of Freight %q in namespace %q: %w",
				f.Name, f.Namespace, err,
			))
		}
	}
	return kerrors.NewAggregate(errs)
}

// clearApprovals clears the approval status of all Freight that have been
// approved for the given Stage. It removes the Stage from the ApprovedFor map
// of each Freight.
func (r *RegularStageReconciler) clearApprovals(ctx context.Context, stage *kargoapi.Stage) error {
	approved := kargoapi.FreightList{}
	if err := r.client.List(
		ctx,
		&approved,
		client.InNamespace(stage.Namespace),
		client.MatchingFieldsSelector{
			Selector: fields.OneTermEqualSelector(
				indexer.FreightApprovedForStagesField,
				stage.Name,
			),
		},
	); err != nil {
		return fmt.Errorf("error listing Freight approved for Stage %q in namespace %q: %w",
			stage.Name,
			stage.Namespace,
			err,
		)
	}

	var errs []error
	for _, f := range approved.Items {
		newStatus := *f.Status.DeepCopy()
		if newStatus.ApprovedFor == nil {
			continue
		}
		delete(newStatus.ApprovedFor, stage.Name)

		if err := kubeclient.PatchStatus(ctx, r.client, &f, func(status *kargoapi.FreightStatus) {
			*status = newStatus
		}); client.IgnoreNotFound(err) != nil {
			errs = append(errs, fmt.Errorf(
				"error clearing approval status of Freight %q in namespace %q: %w",
				f.Name, f.Namespace, err,
			))
		}
	}
	return kerrors.NewAggregate(errs)
}

// clearAnalysisRuns clears all AnalysisRuns that are associated with the given
// Stage. This is only done if the Rollouts integration is enabled.
func (r *RegularStageReconciler) clearAnalysisRuns(ctx context.Context, stage *kargoapi.Stage) error {
	if !r.cfg.RolloutsIntegrationEnabled {
		return nil
	}

	if err := r.client.DeleteAllOf(
		ctx,
		&rolloutsapi.AnalysisRun{},
		client.InNamespace(stage.Namespace),
		client.MatchingLabels(map[string]string{
			kargoapi.LabelKeyStage: kubernetes.ShortenLabelValue(stage.Name),
		}),
	); err != nil {
		return fmt.Errorf("error deleting AnalysisRuns for Stage %q in namespace %q: %w",
			stage.Name,
			stage.Namespace,
			err,
		)
	}
	return nil
}

// summarizeConditions summarizes the conditions of the given Stage. It sets the
// Ready condition based on the Promoting, Healthy, and Verified conditions.
// If there is an error, the Ready condition is set to False until the error is
// resolved.
func summarizeConditions(stage *kargoapi.Stage, newStatus *kargoapi.StageStatus, err error) {
	// If there is an error, then we are not Ready until the error is resolved.
	if err != nil {
		conditions.Set(newStatus, &metav1.Condition{
			Type:               kargoapi.ConditionTypeReady,
			Status:             metav1.ConditionFalse,
			Reason:             "ReconcileError",
			Message:            err.Error(),
			ObservedGeneration: stage.Generation,
		})

		conditions.Set(newStatus, &metav1.Condition{
			Type:               kargoapi.ConditionTypeReconciling,
			Status:             metav1.ConditionTrue,
			Reason:             "RetryAfterError",
			ObservedGeneration: stage.Generation,
		})
		return
	}

	// Set the Freight summary.
	newStatus.FreightSummary = buildFreightSummary(len(stage.Spec.RequestedFreight), newStatus.FreightHistory.Current())

	// If we are currently Promoting, then we are not Ready.
	promoCond := conditions.Get(newStatus, kargoapi.ConditionTypePromoting)
	if promoCond != nil {
		conditions.Set(newStatus, &metav1.Condition{
			Type:               kargoapi.ConditionTypeReady,
			Status:             metav1.ConditionFalse,
			Reason:             promoCond.Reason,
			Message:            promoCond.Message,
			ObservedGeneration: stage.Generation,
		})
		return
	}

	// If we are not currently Promoting but the last promotion failed,
	// then we are not Ready.
	if lastPromo := newStatus.LastPromotion; lastPromo != nil && lastPromo.Status != nil &&
		lastPromo.Status.Phase.IsTerminal() && lastPromo.Status.Phase != kargoapi.PromotionPhaseSucceeded {
		conditions.Set(newStatus, &metav1.Condition{
			Type:               kargoapi.ConditionTypeReady,
			Status:             metav1.ConditionFalse,
			Reason:             fmt.Sprintf("LastPromotion%s", string(lastPromo.Status.Phase)),
			Message:            lastPromo.Status.Message,
			ObservedGeneration: stage.Generation,
		})
		return
	}

	// If we are not Healthy, then we are not Ready.
	healthCond := conditions.Get(newStatus, kargoapi.ConditionTypeHealthy)
	if healthCond == nil || healthCond.Status != metav1.ConditionTrue {
		readyCond := &metav1.Condition{
			Type:               kargoapi.ConditionTypeReady,
			Status:             metav1.ConditionFalse,
			Reason:             "Unhealthy",
			Message:            "Stage is not healthy",
			ObservedGeneration: stage.Generation,
		}
		if healthCond != nil {
			readyCond.Reason = healthCond.Reason
			readyCond.Message = healthCond.Message
		}
		conditions.Set(newStatus, readyCond)
		return
	}

	// If we are not verified, then we are not Ready.
	verificationCond := conditions.Get(newStatus, kargoapi.ConditionTypeVerified)
	if verificationCond == nil || verificationCond.Status != metav1.ConditionTrue {
		readyCond := &metav1.Condition{
			Type:               kargoapi.ConditionTypeReady,
			Status:             metav1.ConditionFalse,
			Reason:             "PendingVerification",
			Message:            "Stage is not verified",
			ObservedGeneration: stage.Generation,
		}
		if verificationCond != nil {
			readyCond.Reason = verificationCond.Reason
			readyCond.Message = verificationCond.Message
		}
		conditions.Set(newStatus, readyCond)
		return
	}

	// At this point, we can propagate the Ready condition from the Verified
	// condition.
	conditions.Set(newStatus, &metav1.Condition{
		Type:               kargoapi.ConditionTypeReady,
		Status:             metav1.ConditionTrue,
		Reason:             verificationCond.Reason,
		Message:            verificationCond.Message,
		ObservedGeneration: stage.Generation,
	})
	conditions.Delete(newStatus, kargoapi.ConditionTypeReconciling)

	// If we are Ready, then we can also mark the current generation as
	// observed.
	newStatus.ObservedGeneration = stage.Generation
}

func buildFreightSummary(requested int, current *kargoapi.FreightCollection) string {
	if current == nil {
		return fmt.Sprintf("0/%d Fulfilled", requested)
	}
	if requested == 1 && len(current.Freight) == 1 {
		for _, f := range current.Freight {
			return f.Name
		}
	}
	return fmt.Sprintf("%d/%d Fulfilled", len(current.Freight), requested)
}
