clusterwatchrule_controller.go

// SPDX-License-Identifier: Apache-2.0

package controller

import (
	"context"
	"errors"
	"fmt"

	"github.com/go-logr/logr"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	logf "sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	ctrlreconcile "sigs.k8s.io/controller-runtime/pkg/reconcile"
	"sigs.k8s.io/controller-runtime/pkg/source"

	configbutleraiv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3"
	"github.com/ConfigButler/gitops-reverser/internal/rulestore"
	"github.com/ConfigButler/gitops-reverser/internal/telemetry"
	reverserTypes "github.com/ConfigButler/gitops-reverser/internal/types"
	"github.com/ConfigButler/gitops-reverser/internal/watch"
)

// ClusterWatchRule status condition reasons.
const (
	ClusterWatchRuleReasonValidating            = "Validating"
	ClusterWatchRuleReasonAccessDenied          = "AccessDenied"
	ClusterWatchRuleReasonGitTargetNotFound     = "GitTargetNotFound"
	ClusterWatchRuleReasonGitDestinationInvalid = "GitDestinationInvalid"
	ClusterWatchRuleReasonReady                 = ReasonSucceeded
	// ClusterWatchRuleReasonCatalogNotReady is ResourcesResolved=False: the source cluster's discovery
	// catalog is not ready, so no selector can be resolved yet. A selector that matches nothing
	// the cluster serves is not this: it resolves True, watching zero types.
	ClusterWatchRuleReasonCatalogNotReady = "CatalogNotReady"

	// ClusterWatchRuleReasonGitTargetNamespaceNotAuthorized is the terminal reason when the
	// referenced GitTarget's namespace is not admitted by that target's KubeProvider. It is
	// re-exported from internal/watch, where the shared compile path both bootstrap and this
	// reconciler call decides it, so the two can never drift.
	ClusterWatchRuleReasonGitTargetNamespaceNotAuthorized = watch.ClusterWatchRuleReasonGitTargetNamespaceNotAuthorized
)

// ClusterWatchRuleReconciler reconciles a ClusterWatchRule object.
type ClusterWatchRuleReconciler struct {
	client.Client

	Scheme       *runtime.Scheme
	RuleStore    *rulestore.RuleStore
	WatchManager WatchManagerInterface
	// Recorder emits a Kubernetes Event on every persisted Ready transition; nil disables Events.
	Recorder record.EventRecorder
}

// +kubebuilder:rbac:groups=configbutler.ai,resources=clusterwatchrules,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=configbutler.ai,resources=clusterwatchrules/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=configbutler.ai,resources=gittargets,verbs=get;list;watch
// +kubebuilder:rbac:groups=configbutler.ai,resources=gitproviders,verbs=get;list;watch
// +kubebuilder:rbac:groups=configbutler.ai,resources=kubeproviders,verbs=get;list;watch
// +kubebuilder:rbac:groups="",resources=namespaces,verbs=get;list;watch

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// clusterWatchRuleGitTarget names the GitTarget a ClusterWatchRule writes through. Unlike a
// WatchRule's, its gitTargetRef carries a namespace of its own.
//
// It carries no UID, and that is correct rather than an omission: a rule-derived reference has
// none to carry, and the watch-plane owner resolves the trigger against the UID the GitTarget
// controller captured. See resolveGitTargetUID.
func clusterWatchRuleGitTarget(rule *configbutleraiv1alpha3.ClusterWatchRule) reverserTypes.ResourceReference {
	return reverserTypes.NewResourceReference(rule.Spec.GitTargetRef.Name, rule.Spec.GitTargetRef.Namespace)
}

func (r *ClusterWatchRuleReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := logf.FromContext(ctx).WithName("ClusterWatchRuleReconciler")

	log.Info("Starting reconciliation", "name", req.Name)

	// Fetch the ClusterWatchRule instance
	var clusterRule configbutleraiv1alpha3.ClusterWatchRule
	if err := r.Get(ctx, req.NamespacedName, &clusterRule); err != nil {
		if client.IgnoreNotFound(err) == nil {
			log.Info("ClusterWatchRule not found, was likely deleted", "name", req.Name)
			// Resource was deleted. Remove it from the store.
			r.RuleStore.DeleteClusterWatchRule(req.NamespacedName)
			log.Info("ClusterWatchRule deleted, removed from store", "name", req.Name)

			// The rule is gone, so the GitTarget it named cannot be read off it. Mark them all;
			// each one's pass is a cheap diff against a plan that has not moved.
			if r.WatchManager != nil {
				r.WatchManager.TriggerAllRuleChange()
			}

			// Stop publishing its conditions for the same reason the store entry goes: a series
			// that outlives its object reports Ready=False forever and never clears.
			telemetry.ForgetResourceConditions(conditionKindClusterWatchRule, req.Namespace, req.Name)

			return ctrl.Result{}, nil
		}
		log.Error(err, "unable to fetch ClusterWatchRule", "name", req.Name)
		return ctrl.Result{}, err
	}

	log.Info("Starting ClusterWatchRule validation",
		"name", clusterRule.Name,
		"target", clusterRule.Spec.GitTargetRef,
		"generation", clusterRule.Generation,
		"resourceVersion", clusterRule.ResourceVersion)
	st := beginStatus(r.Client, r.Recorder, &clusterRule)

	// Seed the axis conditions as not-yet-evaluated. There is deliberately no placeholder write of
	// the Ready/Reconciling/Stalled trio here: every path below ends in exactly one applyReadiness.
	st.set(
		ConditionTypeStreamsRunning,
		metav1.ConditionUnknown,
		GitTargetStreamsRunningReasonNotReady,
		"Blocked by validation; streams not evaluated",
	)
	st.set(
		ConditionTypeGitTargetReady,
		metav1.ConditionUnknown,
		ReasonProgressing,
		"Blocked by validation; GitTarget not evaluated",
	)

	// Delegate to target-based reconciliation
	return r.reconcileClusterWatchRuleViaTarget(ctx, st, &clusterRule)
}

// reconcileClusterWatchRuleViaTarget validates and stores a ClusterWatchRule that references a GitTarget.
func (r *ClusterWatchRuleReconciler) reconcileClusterWatchRuleViaTarget(
	ctx context.Context,
	st *reconcileStatus,
	clusterRule *configbutleraiv1alpha3.ClusterWatchRule,
) (ctrl.Result, error) {
	log := logf.FromContext(ctx).WithName("reconcileClusterWatchRuleViaTarget")

	// Target is required
	if clusterRule.Spec.GitTargetRef.Name == "" {
		st.set(
			ConditionTypeGitTargetReady,
			metav1.ConditionFalse,
			ClusterWatchRuleReasonGitDestinationInvalid,
			"Target.name must be specified for ClusterWatchRule",
		)
		return r.stallRule(ctx, st, ClusterWatchRuleReasonGitDestinationInvalid,
			"Target.name must be specified for ClusterWatchRule")
	}

	// For ClusterWatchRule, target namespace must be specified
	targetNS := clusterRule.Spec.GitTargetRef.Namespace
	if targetNS == "" {
		st.set(
			ConditionTypeGitTargetReady,
			metav1.ConditionFalse,
			ClusterWatchRuleReasonGitDestinationInvalid,
			"Target.namespace must be specified for ClusterWatchRule",
		)
		return r.stallRule(ctx, st, ClusterWatchRuleReasonGitDestinationInvalid,
			"Target.namespace must be specified for ClusterWatchRule")
	}

	// Fetch GitTarget
	var target configbutleraiv1alpha3.GitTarget
	targetKey := types.NamespacedName{Name: clusterRule.Spec.GitTargetRef.Name, Namespace: targetNS}
	if err := r.Get(ctx, targetKey, &target); err != nil {
		log.Error(err, "Failed to get referenced GitTarget",
			"gitTargetName", clusterRule.Spec.GitTargetRef.Name,
			"gitTargetNamespace", targetNS)
		st.set(
			ConditionTypeGitTargetReady,
			metav1.ConditionFalse,
			ClusterWatchRuleReasonGitTargetNotFound,
			fmt.Sprintf("Referenced GitTarget '%s/%s' not found: %v",
				targetNS, clusterRule.Spec.GitTargetRef.Name, err),
		)
		return r.stallRule(ctx, st, ClusterWatchRuleReasonGitTargetNotFound, "Referenced GitTarget not found")
	}
	r.setGitTargetReadyCondition(st, target)

	// Admission AND compilation, in that order and in one call — see gateClusterWatchRule. There is
	// deliberately no AddOrUpdateClusterWatchRule here: routing every compilation through
	// watch.CompileClusterWatchRule is what stops the startup bootstrap from being a second,
	// ungated path into the store.
	if handled, result, err := r.gateClusterWatchRule(ctx, st, clusterRule, target, log); handled {
		return result, err
	}

	// Submit the change and return. The pass runs on the watch-plane owner, after this target has
	// been quiet for the settle window, so the stream summary read below describes the state from
	// BEFORE it — see the status contract in docs/design/watch-manager-ownership.md.
	if r.WatchManager != nil {
		r.WatchManager.TriggerRuleChange(clusterWatchRuleGitTarget(clusterRule))
		r.setResourceResolutionCondition(ctx, st, clusterRule)
		r.setStreamsReadyCondition(st, clusterRule, r.WatchManager.StreamSummaryForClusterWatchRule(*clusterRule))
	} else {
		r.setStreamsReadyCondition(st, clusterRule, noResolvedStreamsSummary())
	}

	log.Info("ClusterWatchRule reconciliation via GitTarget successful", "name", clusterRule.Name)

	msg := fmt.Sprintf(
		"ClusterWatchRule is ready and monitoring resources via GitTarget '%s/%s'",
		clusterRule.Spec.GitTargetRef.Namespace,
		clusterRule.Spec.GitTargetRef.Name,
	)
	return commitRule(ctx, st,
		ruleReadiness(clusterRule.Status.Conditions, gitTargetStalled(target), "ClusterWatchRule", msg))
}

// gateClusterWatchRule is the ClusterWatchRule gate and the ONE place this controller compiles a
// cluster rule. It runs the shared compile path, whose refusal is the KubeProvider namespace
// admission of the referenced GitTarget: a ClusterWatchRule is cluster-scoped and its gitTargetRef
// carries a REQUIRED namespace, so it may name a GitTarget in ANY namespace and widen that target's
// mirror scope cluster-wide. Compiling such a rule without re-applying the target's own provider
// admission would let it mirror through a credential whose grants never admitted that target.
//
// It lives in internal/watch rather than here because the startup bootstrap must apply exactly the
// same refusal BEFORE the first reconcile — otherwise every restart reopens the window it closes.
//
// It returns handled=false when the rule compiled and the reconcile should continue; handled=true
// means the reconcile is over and the caller must return the accompanying result and error
// unchanged.
func (r *ClusterWatchRuleReconciler) gateClusterWatchRule(
	ctx context.Context,
	st *reconcileStatus,
	clusterRule *configbutleraiv1alpha3.ClusterWatchRule,
	target configbutleraiv1alpha3.GitTarget,
	log logr.Logger,
) (bool, ctrl.Result, error) {
	var selected watch.ClusterResourceSelection
	if r.WatchManager != nil {
		selected = r.WatchManager.ClusterResourcesSelectedFor(target)
	}
	decision, err := watch.CompileClusterWatchRule(
		ctx, r.Client, r.RuleStore, *clusterRule, target, selected)
	var selectorErr *watch.ObjectSelectorError
	if errors.As(err, &selectorErr) {
		result, stallErr := r.refuseRuleResources(ctx, st, clusterRule, watch.ReasonInvalidObjectSelector,
			selectorErr.Message, log)
		return true, result, stallErr
	}
	if err != nil {
		// A transient apiserver failure must NOT tear down a running stream: leave the compiled
		// rule in place and requeue with the error so the gate re-runs on real data.
		log.Error(err, "Failed to evaluate admission for ClusterWatchRule",
			"gitTargetName", target.Name, "gitTargetNamespace", target.Namespace)
		return true, ctrl.Result{}, err
	}
	if decision.Admitted {
		return false, ctrl.Result{}, nil
	}
	if decision.ResourcesRefused {
		result, stallErr := r.refuseRuleResources(ctx, st, clusterRule, decision.Reason, decision.Message, log)
		return true, result, stallErr
	}

	result, refuseErr := r.refuseClusterWatchRule(ctx, st, clusterRule, decision, log)
	return true, result, refuseErr
}

// refuseRuleResources is the refusal of the rule's own resources rather than of its GitTarget, so it
// lands on ResourcesResolved rather than on the condition refuseClusterWatchRule writes. The compiled
// rule is already out of the store; the replan tears down what it kept alive.
func (r *ClusterWatchRuleReconciler) refuseRuleResources(
	ctx context.Context,
	st *reconcileStatus,
	clusterRule *configbutleraiv1alpha3.ClusterWatchRule,
	reason, message string,
	log logr.Logger,
) (ctrl.Result, error) {
	log.Info("Refusing ClusterWatchRule", "name", clusterRule.Name, "reason", reason, "message", message)
	if r.WatchManager != nil {
		r.WatchManager.TriggerRuleChange(clusterWatchRuleGitTarget(clusterRule))
	}
	st.set(ConditionTypeResourcesResolved, metav1.ConditionFalse, reason, message)
	st.set(ConditionTypeStreamsRunning, metav1.ConditionFalse, reason, "No streams: the ClusterWatchRule was refused")
	return r.stallRule(ctx, st, reason, message)
}

// refuseClusterWatchRule is the denial half of the ClusterWatchRule gate.
//
// Order is the contract, not an implementation detail: CompileClusterWatchRule has ALREADY removed
// the compiled rule, this replans the watch manager, and only then is the terminal status
// published. A gate that only writes a condition is not a gate — it leaves the stream running while
// announcing that it is not. Any test that asserts the terminal condition must therefore also be
// able to assert the rule is already gone.
//
// The refusal is terminal (Stalled=True, Reconciling=False) rather than a retry: nothing this
// controller does will change the verdict. Recovery arrives as an event — a KubeProvider policy
// change or a Namespace label change — through the mappers registered in SetupWithManager, or as an
// edit converting the rule to the cluster-only model.
func (r *ClusterWatchRuleReconciler) refuseClusterWatchRule(
	ctx context.Context,
	st *reconcileStatus,
	clusterRule *configbutleraiv1alpha3.ClusterWatchRule,
	decision watch.ClusterWatchRuleDecision,
	log logr.Logger,
) (ctrl.Result, error) {
	log.Info("Refusing ClusterWatchRule",
		"name", clusterRule.Name,
		"reason", decision.Reason,
		"message", decision.Message)

	// The compiled rule is already out of the store; trigger a replan so the watch manager tears
	// down a stream this rule was keeping alive.
	if r.WatchManager != nil {
		r.WatchManager.TriggerRuleChange(clusterWatchRuleGitTarget(clusterRule))
	}

	st.set(
		ConditionTypeGitTargetReady,
		metav1.ConditionFalse,
		decision.Reason,
		decision.Message,
	)
	st.set(
		ConditionTypeStreamsRunning,
		metav1.ConditionFalse,
		decision.Reason,
		"No streams: the ClusterWatchRule was refused",
	)

	return r.stallRule(ctx, st, decision.Reason, decision.Message)
}

// stallRule publishes a terminal ClusterWatchRule outcome and ends the reconcile.
func (r *ClusterWatchRuleReconciler) stallRule(
	ctx context.Context,
	st *reconcileStatus,
	reason, message string,
) (ctrl.Result, error) {
	rd := newRuleReadiness("ClusterWatchRule", "")
	rd.stalled(reason, message)
	return commitRule(ctx, st, rd)
}

func (r *ClusterWatchRuleReconciler) setGitTargetReadyCondition(
	st *reconcileStatus,
	target configbutleraiv1alpha3.GitTarget,
) {
	ready := gitTargetReadyCondition(target)
	st.setValue(ConditionTypeGitTargetReady, ready)
}

func (r *ClusterWatchRuleReconciler) setResourceResolutionCondition(
	ctx context.Context,
	st *reconcileStatus,
	clusterRule *configbutleraiv1alpha3.ClusterWatchRule,
) {
	if refused, message := r.WatchManager.CollectionOverlapForClusterWatchRule(*clusterRule); refused {
		st.set(ConditionTypeResourcesResolved, metav1.ConditionFalse, watch.ReasonCollectionOverlap, message)
		return
	}
	resolved, message := r.WatchManager.ResolveClusterWatchRuleResources(ctx, *clusterRule)
	status := metav1.ConditionFalse
	reason := ClusterWatchRuleReasonCatalogNotReady
	if resolved {
		status = metav1.ConditionTrue
		reason = ReasonSucceeded
	}
	st.set(ConditionTypeResourcesResolved, status, reason, message)
}

func (r *ClusterWatchRuleReconciler) setStreamsReadyCondition(
	st *reconcileStatus,
	clusterRule *configbutleraiv1alpha3.ClusterWatchRule,
	streams watch.StreamSummary,
) {
	clusterRule.Status.Streams = streamsStatus(streams)
	st.set(ConditionTypeStreamsRunning, streamConditionStatus(streams), streams.Reason, streams.Message)
}

// SetupWithManager sets up the controller with the Manager.
func (r *ClusterWatchRuleReconciler) SetupWithManager(mgr ctrl.Manager) error {
	b := ctrl.NewControllerManagedBy(mgr).
		// A For() predicate is not an optimisation here, it closes a self-triggering edge: a status
		// write bumps resourceVersion and fires an Update watch event that EnqueueRequestForObject
		// turns straight back into a queued request, un-rate-limited. reconcileStatus.commit()
		// already suppresses no-op writes, so the loop has no fuel; this makes it structural, and
		// matches what GitProvider and KubeProvider already do.
		For(&configbutleraiv1alpha3.ClusterWatchRule{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
		Named("clusterwatchrule")
	b = ruleDependencyWatches(b, r.gitTargetToClusterWatchRules, r.kubeProviderToClusterWatchRules,
		r.namespaceToClusterWatchRules)

	// React to a stream of this rule's GitTarget reaching or leaving Streaming, for the reason the
	// WatchRule controller wires the same channel: the data plane is the only thing that knows when
	// a stream came up, and without the edge the rule publishes StreamsRunning=False until its 10s
	// settle requeue.
	if r.WatchManager != nil {
		if events := r.WatchManager.StreamStateEvents(); events != nil {
			b = b.WatchesRawSource(source.Channel(
				events,
				handler.EnqueueRequestsFromMapFunc(r.gitTargetToClusterWatchRules),
			))
		}
	}

	return b.Complete(r)
}

// kubeProviderToClusterWatchRules maps a KubeProvider policy change to every ClusterWatchRule
// whose referenced GitTarget mirrors through that provider, so an admission grant or revocation
// converges on the event rather than on the periodic reconcile.
func (r *ClusterWatchRuleReconciler) kubeProviderToClusterWatchRules(
	ctx context.Context,
	obj client.Object,
) []ctrlreconcile.Request {
	var targets configbutleraiv1alpha3.GitTargetList
	if err := r.List(ctx, &targets); err != nil {
		logDependencyListError(ctx, err, "GitTargets", obj)
		return nil
	}

	affected := make(map[types.NamespacedName]struct{}, len(targets.Items))
	for i := range targets.Items {
		t := &targets.Items[i]
		if t.KubeProviderKey() != client.ObjectKeyFromObject(obj) {
			continue
		}
		affected[types.NamespacedName{Name: t.Name, Namespace: t.Namespace}] = struct{}{}
	}

	return r.clusterWatchRulesTargeting(ctx, affected, obj)
}

// namespaceToClusterWatchRules maps a Namespace label change to every ClusterWatchRule whose
// referenced GitTarget lives in that namespace — what a Selector grant reads.
func (r *ClusterWatchRuleReconciler) namespaceToClusterWatchRules(
	ctx context.Context,
	obj client.Object,
) []ctrlreconcile.Request {
	var targets configbutleraiv1alpha3.GitTargetList
	if err := r.List(ctx, &targets, client.InNamespace(obj.GetName())); err != nil {
		logDependencyListError(ctx, err, "GitTargets", obj)
		return nil
	}

	affected := make(map[types.NamespacedName]struct{}, len(targets.Items))
	for i := range targets.Items {
		t := &targets.Items[i]
		affected[types.NamespacedName{Name: t.Name, Namespace: t.Namespace}] = struct{}{}
	}

	return r.clusterWatchRulesTargeting(ctx, affected, obj)
}

// clusterWatchRulesTargeting returns a request for every ClusterWatchRule whose gitTargetRef names one
// of the given GitTargets. Requests carry a name only — ClusterWatchRule is cluster-scoped.
func (r *ClusterWatchRuleReconciler) clusterWatchRulesTargeting(
	ctx context.Context,
	targets map[types.NamespacedName]struct{},
	obj client.Object,
) []ctrlreconcile.Request {
	if len(targets) == 0 {
		return nil
	}

	var rules configbutleraiv1alpha3.ClusterWatchRuleList
	if err := r.List(ctx, &rules); err != nil {
		logDependencyListError(ctx, err, "ClusterWatchRules", obj)
		return nil
	}

	var requests []ctrlreconcile.Request
	for i := range rules.Items {
		rule := &rules.Items[i]
		key := types.NamespacedName{
			Name:      rule.Spec.GitTargetRef.Name,
			Namespace: rule.Spec.GitTargetRef.Namespace,
		}
		if _, ok := targets[key]; !ok {
			continue
		}
		requests = append(requests, ctrlreconcile.Request{
			NamespacedName: types.NamespacedName{Name: rule.Name},
		})
	}
	return requests
}

// gitTargetToClusterWatchRules maps a GitTarget event to every ClusterWatchRule
// whose gitTargetRef matches it. ClusterWatchRule is cluster-scoped, so the lookup
// is cluster-wide.
func (r *ClusterWatchRuleReconciler) gitTargetToClusterWatchRules(
	ctx context.Context,
	obj client.Object,
) []ctrlreconcile.Request {
	var rules configbutleraiv1alpha3.ClusterWatchRuleList
	if err := r.List(ctx, &rules); err != nil {
		logDependencyListError(ctx, err, "ClusterWatchRules", obj)
		return nil
	}

	var requests []ctrlreconcile.Request
	for i := range rules.Items {
		rule := &rules.Items[i]
		if rule.Spec.GitTargetRef.Name != obj.GetName() {
			continue
		}
		if rule.Spec.GitTargetRef.Namespace != obj.GetNamespace() {
			continue
		}
		requests = append(requests, ctrlreconcile.Request{
			NamespacedName: types.NamespacedName{Name: rule.Name},
		})
	}
	return requests
}