quorate/services/project-delivery/controller/controller.go

package main

import (
	"context"
	"crypto/sha256"
	"encoding/base32"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"slices"
	"sort"
	"strings"
	"time"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/util/retry"
)

var (
	projectDeliveryGVR = schema.GroupVersionResource{Group: "delivery.quorate.dev", Version: "v1alpha1", Resource: "projectdeliveries"}
	workflowGVR        = schema.GroupVersionResource{Group: "argoproj.io", Version: "v1alpha1", Resource: "workflows"}
)

type controllerConfig struct {
	pollInterval   time.Duration
	bucketInterval time.Duration
	requestTimeout time.Duration
	templateQuiet  time.Duration
}

type controller struct {
	client     dynamic.Interface
	source     sourceReader
	now        func() time.Time
	config     controllerConfig
	qualCursor string
}

func (c *controller) run(ctx context.Context) error {
	ticker := time.NewTicker(c.config.pollInterval)
	defer ticker.Stop()
	for {
		c.reconcileAll(ctx)
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
		}
	}
}

func (c *controller) reconcileAll(ctx context.Context) {
	if err := c.releaseOrphans(ctx); err != nil {
		slog.Error("release orphan records", "error", err)
	}
	if err := c.reconcileTemplates(ctx); err != nil {
		slog.Error("reconcile templates", "error", err)
	}
	if err := c.qualifyAll(ctx); err != nil {
		slog.Error("reconcile qualification", "error", err)
	}
	items, err := c.client.Resource(projectDeliveryGVR).Namespace(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
	if err != nil {
		slog.Error("list ProjectDeliveries", "error", err)
		return
	}
	for index := range items.Items {
		item := &items.Items[index]
		if err := c.reconcileKey(ctx, item.GetNamespace(), item.GetName()); err != nil {
			slog.Error("reconcile ProjectDelivery", "namespace", item.GetNamespace(), "name", item.GetName(), "error", err)
			c.recordError(ctx, item.GetNamespace(), item.GetName(), err)
		}
	}
}

func (c *controller) reconcileKey(ctx context.Context, namespace, name string) error {
	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
		object, err := c.client.Resource(projectDeliveryGVR).Namespace(namespace).Get(ctx, name, metav1.GetOptions{})
		if apierrors.IsNotFound(err) {
			return nil
		}
		if err != nil {
			return err
		}
		return c.reconcile(ctx, object)
	})
}

func (c *controller) reconcile(ctx context.Context, object *unstructured.Unstructured) error {
	delivery, err := decodeProjectDelivery(object)
	if err != nil {
		return err
	}
	if delivery.Metadata.DeletionTimestamp != nil {
		return nil
	}
	if err := validateTargets(delivery.Spec.Targets); err != nil {
		return err
	}
	recovered, err := c.recoverBuilds(ctx, &delivery)
	if err != nil {
		return err
	}
	active, err := c.resumeAttempts(ctx, &delivery)
	if err != nil {
		return err
	}
	if active || recovered {
		return c.updateStatus(ctx, object, &delivery)
	}
	retried, err := c.reconcileRetry(ctx, &delivery)
	if err != nil {
		return err
	}
	var reconcileErr error
	switch {
	case retried:
	case delivery.Spec.Source.Continuous != nil:
		refs, err := c.readRefs(ctx, delivery.Spec.Source.Repository)
		if err != nil {
			return fmt.Errorf("read Forgejo refs: %w", err)
		}
		reconcileErr = c.reconcileContinuous(ctx, &delivery, refs)
	case delivery.Spec.Source.Point != nil:
		reconcileErr = c.reconcilePoint(ctx, &delivery)
	default:
		reconcileErr = errors.New("delivery source has no mode")
	}
	if reconcileErr != nil {
		return reconcileErr
	}
	delivery.Status.ObservedGeneration = delivery.Metadata.Generation
	return c.updateStatus(ctx, object, &delivery)
}

func (c *controller) resumeAttempts(ctx context.Context, delivery *projectDelivery) (bool, error) {
	active := false
	if status := delivery.Status.Continuous; status != nil && status.Attempt != nil && !terminalPhase(status.Attempt.Phase) {
		if err := c.advanceAttempt(ctx, delivery, status.Attempt); err != nil {
			return false, err
		}
		recordDelivered(status)
		delivery.Status.Message = "build attempt: " + phaseOrPending(status.Attempt.Phase)
		active = true
	}
	for index := range delivery.Status.Points {
		attempt := &delivery.Status.Points[index]
		if terminalPhase(attempt.Phase) {
			continue
		}
		if err := c.advanceAttempt(ctx, delivery, attempt); err != nil {
			return false, err
		}
		delivery.Status.Message = "point attempt: " + phaseOrPending(attempt.Phase)
		active = true
	}
	return active, nil
}

func (c *controller) reconcileRetry(ctx context.Context, delivery *projectDelivery) (bool, error) {
	request := delivery.Metadata.Annotations[retryAnnotation]
	if request == "" || request == delivery.Status.LastHandledRetry {
		return false, nil
	}
	target, continuous := retryTarget(delivery, request)
	if target == nil {
		return false, fmt.Errorf("retry Workflow %q is not a recorded attempt", request)
	}
	if otherAttemptActive(delivery, request) {
		return false, nil
	}
	if !terminalPhase(target.Phase) {
		return false, nil
	}
	successor := attemptStatus{
		Ref:            target.Ref,
		Revision:       target.Revision,
		SourceIdentity: target.SourceIdentity,
		RetryOf:        request,
		Phase:          "Pending",
	}
	if err := c.advanceAttempt(ctx, delivery, &successor); err != nil {
		return false, err
	}
	if successor.WorkflowName == "" {
		delivery.Status.Message = templatePending.Error()
		return true, nil
	}
	if continuous {
		rememberAttempt(delivery.Status.Continuous, &successor)
		target = delivery.Status.Continuous.Attempt
	} else {
		delivery.Status.Points = append(delivery.Status.Points, successor)
		target = &delivery.Status.Points[len(delivery.Status.Points)-1]
	}
	delivery.Status.LastHandledRetry = request
	delivery.Status.Message = "retry attempt: " + phaseOrPending(target.Phase)
	return true, nil
}

func retryTarget(delivery *projectDelivery, workflow string) (*attemptStatus, bool) {
	if delivery.Status.Continuous != nil {
		attempt := delivery.Status.Continuous.Attempt
		if attempt != nil && attempt.WorkflowName == workflow {
			return attempt, true
		}
	}
	for index := range delivery.Status.Points {
		attempt := &delivery.Status.Points[index]
		if attempt.WorkflowName == workflow {
			return attempt, false
		}
	}
	return nil, false
}

func otherAttemptActive(delivery *projectDelivery, excluded string) bool {
	if delivery.Status.Continuous != nil {
		attempt := delivery.Status.Continuous.Attempt
		if attempt != nil && attempt.WorkflowName != excluded && !terminalPhase(attempt.Phase) {
			return true
		}
	}
	for index := range delivery.Status.Points {
		attempt := &delivery.Status.Points[index]
		if attempt.WorkflowName != excluded && !terminalPhase(attempt.Phase) {
			return true
		}
	}
	return false
}

func (c *controller) reconcileContinuous(
	ctx context.Context,
	delivery *projectDelivery,
	refs map[string]string,
) error {
	ref := delivery.Spec.Source.Continuous.Ref
	revision, found := refs[ref]
	if !found {
		return fmt.Errorf("continuous ref %q is absent", ref)
	}
	if delivery.Status.Continuous == nil {
		delivery.Status.Continuous = &continuousStatus{}
	}
	status := delivery.Status.Continuous
	sourceIdentity, err := c.sourceIdentity(ctx, delivery, revision)
	if err != nil {
		return fmt.Errorf("read continuous path identity: %w", err)
	}
	filtered := len(delivery.Spec.Source.Continuous.Paths) != 0
	if consumedIdentity(status, sourceIdentity) {
		clearCandidate(status)
		delivery.Status.Message = "continuous source already attempted"
		return nil
	}
	if deliveredMatches(status.Delivered, sourceIdentity) ||
		status.Delivered == nil && status.Attempt != nil && status.Attempt.Phase == "Succeeded" &&
			attemptMatches(status.Attempt, sourceIdentity, revision, filtered) {
		clearCandidate(status)
		delivery.Status.Message = "continuous source already delivered"
		return nil
	}
	if status.Attempt != nil && terminalPhase(status.Attempt.Phase) &&
		attemptMatches(status.Attempt, sourceIdentity, revision, filtered) {
		delivery.Status.Message = "continuous source already attempted"
		return nil
	}
	now := c.now().UTC()
	bucket := now.Truncate(c.config.bucketInterval).Format(time.RFC3339)
	if status.CandidateIdentity != sourceIdentity {
		status.CandidateRevision = revision
		status.CandidateIdentity = sourceIdentity
		status.CandidateBuckets = []string{bucket}
		status.CandidateSince = &now
		if delivery.Spec.Source.Continuous.Debounce == "" {
			delivery.Status.Message = "observed continuous candidate 1/3"
		} else {
			delivery.Status.Message = "observed continuous candidate"
		}
		return nil
	}
	status.CandidateRevision = revision
	if configured := delivery.Spec.Source.Continuous.Debounce; configured != "" {
		debounce, err := time.ParseDuration(configured)
		if err != nil || debounce <= 0 {
			return fmt.Errorf("invalid continuous debounce %q", configured)
		}
		if status.CandidateSince == nil {
			status.CandidateSince = &now
			delivery.Status.Message = "observed continuous candidate"
			return nil
		}
		if now.Before(status.CandidateSince.Add(debounce)) {
			delivery.Status.Message = "continuous candidate debouncing"
			return nil
		}
	} else {
		if !contains(status.CandidateBuckets, bucket) && len(status.CandidateBuckets) < 3 {
			status.CandidateBuckets = append(status.CandidateBuckets, bucket)
		}
		if len(status.CandidateBuckets) < 3 {
			delivery.Status.Message = fmt.Sprintf("observed continuous candidate %d/3", len(status.CandidateBuckets))
			return nil
		}
	}
	latest, err := c.readRefs(ctx, delivery.Spec.Source.Repository)
	if err != nil {
		return fmt.Errorf("reread continuous ref: %w", err)
	}
	latestRevision, found := latest[ref]
	if !found {
		return fmt.Errorf("continuous ref %q is absent after debounce", ref)
	}
	latestIdentity, err := c.sourceIdentity(ctx, delivery, latestRevision)
	if err != nil {
		return fmt.Errorf("reread continuous path identity: %w", err)
	}
	if consumedIdentity(status, latestIdentity) {
		clearCandidate(status)
		delivery.Status.Message = "continuous source returned to attempted state"
		return nil
	}
	if deliveredMatches(status.Delivered, latestIdentity) {
		clearCandidate(status)
		delivery.Status.Message = "continuous source returned to delivered state"
		return nil
	}
	if latestIdentity != sourceIdentity {
		status.CandidateRevision = latestRevision
		status.CandidateIdentity = latestIdentity
		status.CandidateBuckets = nil
		status.CandidateSince = nil
		delivery.Status.Message = "continuous source changed before Workflow creation"
		return nil
	}
	revision = latestRevision
	status.CandidateRevision = revision
	attempt := &attemptStatus{
		Ref:            ref,
		Revision:       revision,
		SourceIdentity: sourceIdentity,
		Phase:          "Pending",
	}
	if err := c.advanceAttempt(ctx, delivery, attempt); err != nil {
		return err
	}
	if attempt.WorkflowName == "" {
		delivery.Status.Message = templatePending.Error()
		return nil
	}
	rememberAttempt(status, attempt)
	recordDelivered(status)
	delivery.Status.Message = "build attempt: " + phaseOrPending(attempt.Phase)
	return nil
}

func (c *controller) reconcilePoint(
	ctx context.Context,
	delivery *projectDelivery,
) error {
	for _, revision := range delivery.Spec.Source.Point.Revisions {
		if pointAttempted(delivery.Status.Points, revision) {
			continue
		}
		attempt := attemptStatus{
			Revision: revision,
			Phase:    "Pending",
		}
		if err := c.advanceAttempt(ctx, delivery, &attempt); err != nil {
			return err
		}
		if attempt.WorkflowName == "" {
			delivery.Status.Message = templatePending.Error()
			return nil
		}
		delivery.Status.Points = append(delivery.Status.Points, attempt)
		delivery.Status.Message = "point attempt: " + phaseOrPending(attempt.Phase)
		return nil
	}

	delivery.Status.Message = "point revisions already attempted"
	return nil
}

func (c *controller) advanceAttempt(
	ctx context.Context,
	delivery *projectDelivery,
	attempt *attemptStatus,
) error {
	if attempt.Input == nil {
		targets, err := json.Marshal(delivery.Spec.Targets)
		if err != nil {
			return err
		}
		attempt.Input = map[string]string{
			"source-repository": delivery.Spec.Source.Repository,
			"source-ref":        attempt.Ref,
			"source-revision":   attempt.Revision,
			"targets":           string(targets),
		}
	}
	name := attempt.WorkflowName
	if name == "" {
		name = attemptName(delivery, attempt)
		if err := c.ensureWorkflow(ctx, delivery, name, delivery.Spec.WorkflowTemplateRef.Name, "build", attempt.Input, map[string]string{sourceKey: attempt.SourceIdentity, retryKey: attempt.RetryOf}); err != nil {
			if errors.Is(err, templatePending) {
				return nil
			}
			return err
		}
	}
	state, err := c.workflow(ctx, delivery.Metadata.Namespace, name)
	if apierrors.IsNotFound(err) && attempt.WorkflowName != "" {
		attempt.Phase = "Error"
		return nil
	}
	if err != nil {
		return err
	}
	attempt.WorkflowName = name
	attempt.CreatedAt = state.CreatedAt
	attempt.Input = state.Input
	attempt.Ref = state.Input["source-ref"]
	attempt.Revision = state.Input["source-revision"]
	attempt.Phase = phaseOrPending(state.Phase)
	if result, found := state.Outputs["delivery-result"]; found {
		attempt.Result = opaqueResult(result)
	}
	observeQualBuild(delivery, attempt)
	return nil
}

func (c *controller) ensureWorkflow(
	ctx context.Context,
	delivery *projectDelivery,
	name, templateName, purpose string,
	parameters map[string]string,
	capture map[string]string,
) error {
	resource := c.client.Resource(workflowGVR).Namespace(delivery.Metadata.Namespace)
	if _, err := resource.Get(ctx, name, metav1.GetOptions{}); err == nil {
		return nil
	} else if !apierrors.IsNotFound(err) {
		return err
	}
	published, err := c.readyTemplate(ctx, delivery.Metadata.Namespace, templateName)
	if errors.Is(err, templatePending) {
		if _, getErr := resource.Get(ctx, name, metav1.GetOptions{}); getErr == nil {
			return nil
		} else if !apierrors.IsNotFound(getErr) {
			return getErr
		}
	}
	if err != nil {
		return err
	}
	metadata := map[string]any{
		"name":        name,
		"namespace":   delivery.Metadata.Namespace,
		"annotations": map[string]any{templateKey: templateName},
		"finalizers":  []any{recordKey},
		"labels": map[string]any{
			"app.kubernetes.io/name":       "project-delivery",
			"app.kubernetes.io/instance":   delivery.Metadata.Name,
			"delivery.quorate.dev/purpose": purpose,
			recordKey:                      "pending",
		},
		"ownerReferences": []any{map[string]any{
			"apiVersion":         "delivery.quorate.dev/v1alpha1",
			"kind":               "ProjectDelivery",
			"name":               delivery.Metadata.Name,
			"uid":                delivery.Metadata.UID,
			"controller":         true,
			"blockOwnerDeletion": false,
		}},
	}
	for key, value := range capture {
		metadata["annotations"].(map[string]any)[key] = value
	}
	names := make([]string, 0, len(parameters))
	for parameter := range parameters {
		names = append(names, parameter)
	}
	sort.Strings(names)
	arguments := make([]any, 0, len(names))
	for _, parameter := range names {
		arguments = append(arguments, map[string]any{"name": parameter, "value": parameters[parameter]})
	}
	workflow := &unstructured.Unstructured{Object: map[string]any{
		"apiVersion": "argoproj.io/v1alpha1",
		"kind":       "Workflow",
		"metadata":   metadata,
		"spec": map[string]any{
			"workflowTemplateRef": map[string]any{"name": published},
			"arguments":           map[string]any{"parameters": arguments},
			"synchronization": map[string]any{
				"mutexes": []any{map[string]any{"name": templateName}},
			},
		},
	}}
	_, err = resource.Create(ctx, workflow, metav1.CreateOptions{})
	if apierrors.IsAlreadyExists(err) {
		return nil
	}
	return err
}

func (c *controller) workflow(ctx context.Context, namespace, name string) (workflowState, error) {
	object, err := c.client.Resource(workflowGVR).Namespace(namespace).Get(ctx, name, metav1.GetOptions{})
	if err != nil {
		return workflowState{}, err
	}
	phase, _, _ := unstructured.NestedString(object.Object, "status", "phase")
	outputs := map[string]string{}
	parameters, _, _ := unstructured.NestedSlice(object.Object, "status", "outputs", "parameters")
	rootParameters, _, _ := unstructured.NestedSlice(object.Object, "status", "nodes", name, "outputs", "parameters")
	parameters = append(parameters, rootParameters...)
	for _, item := range parameters {
		parameter, ok := item.(map[string]any)
		if !ok {
			continue
		}
		name, _ := parameter["name"].(string)
		value, _ := parameter["value"].(string)
		if name != "" {
			outputs[name] = value
		}
	}
	input := map[string]string{}
	arguments, _, _ := unstructured.NestedSlice(object.Object, "spec", "arguments", "parameters")
	for _, item := range arguments {
		parameter := item.(map[string]any)
		name, _ := parameter["name"].(string)
		value, _ := parameter["value"].(string)
		input[name] = value
	}
	var startedAt *time.Time
	if value, _, _ := unstructured.NestedString(object.Object, "status", "startedAt"); value != "" {
		timestamp, err := time.Parse(time.RFC3339, value)
		if err != nil {
			return workflowState{}, err
		}
		if !timestamp.IsZero() {
			startedAt = &timestamp
		}
	}
	ownerUID := ""
	for _, owner := range object.GetOwnerReferences() {
		if owner.Kind == "ProjectDelivery" && owner.APIVersion == "delivery.quorate.dev/v1alpha1" {
			ownerUID = string(owner.UID)
		}
	}
	createdAt := object.GetCreationTimestamp()
	return workflowState{Phase: phase, Outputs: outputs, Input: input, StartedAt: startedAt, OwnerUID: ownerUID, CreatedAt: &createdAt.Time}, nil
}

func consumedIdentity(status *continuousStatus, identity string) bool {
	return status.Attempt != nil && status.Attempt.History != nil && slices.Contains(status.Attempt.History.Identities, identity)
}

func rememberAttempt(status *continuousStatus, attempt *attemptStatus) {
	history := &attemptHistory{}
	if previous := status.Attempt; previous != nil {
		if previous.History != nil {
			history.Workflows = slices.Clone(previous.History.Workflows)
			history.Identities = slices.Clone(previous.History.Identities)
		}
		addHistory(history, previous)
	}
	addHistory(history, attempt)
	attempt.History = history
	status.Attempt = attempt
}

func addHistory(history *attemptHistory, attempt *attemptStatus) {
	if attempt.WorkflowName != "" && !slices.Contains(history.Workflows, attempt.WorkflowName) {
		history.Workflows = append(history.Workflows, attempt.WorkflowName)
	}
	if attempt.SourceIdentity != "" && !slices.Contains(history.Identities, attempt.SourceIdentity) {
		history.Identities = append(history.Identities, attempt.SourceIdentity)
	}
}

func (c *controller) updateStatus(ctx context.Context, object *unstructured.Unstructured, delivery *projectDelivery) error {
	status, err := toMap(delivery.Status)
	if err != nil {
		return err
	}
	copy := object.DeepCopy()
	copy.Object["status"] = status
	_, err = c.client.Resource(projectDeliveryGVR).Namespace(copy.GetNamespace()).UpdateStatus(ctx, copy, metav1.UpdateOptions{})
	if apierrors.IsNotFound(err) {
		return c.releaseOrphans(ctx)
	}
	if err != nil {
		return err
	}
	return c.releaseRecords(ctx, delivery)
}

func (c *controller) recordError(ctx context.Context, namespace, name string, reconcileError error) {
	object, err := c.client.Resource(projectDeliveryGVR).Namespace(namespace).Get(ctx, name, metav1.GetOptions{})
	if err != nil {
		return
	}
	delivery, err := decodeProjectDelivery(object)
	if err != nil {
		return
	}
	delivery.Status.Message = reconcileError.Error()
	delivery.Status.ObservedGeneration = delivery.Metadata.Generation
	_ = c.updateStatus(ctx, object, &delivery)
}

func decodeProjectDelivery(object *unstructured.Unstructured) (projectDelivery, error) {
	content, err := json.Marshal(object.Object)
	if err != nil {
		return projectDelivery{}, err
	}
	var delivery projectDelivery
	if err := json.Unmarshal(content, &delivery); err != nil {
		return projectDelivery{}, err
	}
	return delivery, nil
}

func validateTargets(targets []json.RawMessage) error {
	seen := make(map[string]struct{}, len(targets))
	for index, target := range targets {
		key := string(target)
		if _, found := seen[key]; found {
			return fmt.Errorf("duplicate target at index %d", index)
		}
		seen[key] = struct{}{}
	}
	return nil
}

func toMap(value any) (map[string]any, error) {
	content, err := json.Marshal(value)
	if err != nil {
		return nil, err
	}
	var result map[string]any
	err = json.Unmarshal(content, &result)
	return result, err
}

func workflowName(purpose string, fields ...string) string {
	digest := sha256.Sum256([]byte("v1\x00" + purpose + "\x00" + strings.Join(fields, "\x00")))
	encoded := base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(digest[:20])
	return "pd-" + purpose + "-" + strings.ToLower(encoded)
}

func attemptName(delivery *projectDelivery, attempt *attemptStatus) string {
	sourceMode := ""
	identity := attempt.Revision
	if attempt.Ref != "" {
		sourceMode = "continuous"
		if attempt.SourceIdentity != "" {
			identity = attempt.SourceIdentity
		}
	}
	fields := []string{delivery.Metadata.UID, sourceMode, identity}
	if attempt.RetryOf != "" {
		fields = append(fields, attempt.RetryOf)
	}
	return workflowName("build", fields...)
}

func (c *controller) readRefs(ctx context.Context, repository string) (map[string]string, error) {
	requestContext, cancel := context.WithTimeout(ctx, c.config.requestTimeout)
	defer cancel()
	return c.source.Refs(requestContext, repository)
}

func (c *controller) sourceIdentity(
	ctx context.Context,
	delivery *projectDelivery,
	revision string,
) (string, error) {
	continuous := delivery.Spec.Source.Continuous
	paths := append([]string(nil), continuous.Paths...)
	sort.Strings(paths)
	if len(paths) == 0 {
		return revision, nil
	}
	requestContext, cancel := context.WithTimeout(ctx, c.config.requestTimeout)
	defer cancel()
	value, err := c.source.PathIdentity(requestContext, delivery.Spec.Source.Repository, revision, paths)
	if err != nil {
		return "", err
	}
	return value, nil
}

func recordDelivered(status *continuousStatus) {
	attempt := status.Attempt
	if attempt == nil || attempt.Phase != "Succeeded" || attempt.SourceIdentity == "" {
		return
	}
	status.Delivered = &deliveredSource{
		Revision: attempt.Revision,
		Identity: attempt.SourceIdentity,
	}
}

func deliveredMatches(delivered *deliveredSource, sourceIdentity string) bool {
	return delivered != nil && delivered.Identity == sourceIdentity
}

func attemptMatches(attempt *attemptStatus, sourceIdentity, revision string, filtered bool) bool {
	if attempt.SourceIdentity != "" {
		return attempt.SourceIdentity == sourceIdentity
	}
	return !filtered && attempt.Revision == revision
}

func clearCandidate(status *continuousStatus) {
	status.CandidateRevision = ""
	status.CandidateIdentity = ""
	status.CandidateBuckets = nil
	status.CandidateSince = nil
}

func contains(values []string, expected string) bool {
	for _, value := range values {
		if value == expected {
			return true
		}
	}
	return false
}

func terminalPhase(phase string) bool {
	return phase == "Succeeded" || phase == "Failed" || phase == "Error"
}

func pointAttempted(attempts []attemptStatus, revision string) bool {
	for _, attempt := range attempts {
		if attempt.Revision == revision {
			return true
		}
	}
	return false
}

func phaseOrPending(phase string) string {
	if phase == "" {
		return "Pending"
	}
	return phase
}