capsule/internal/controllers/resources/namespace_trigger.go

// Copyright 2020-2026 Project Capsule Authors
// SPDX-License-Identifier: Apache-2.0

package resources

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"time"

	"github.com/go-logr/logr"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/sets"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"

	capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2"
	"github.com/projectcapsule/capsule/internal/cache"
	"github.com/projectcapsule/capsule/internal/controllers/utils"
	"github.com/projectcapsule/capsule/pkg/api"
	"github.com/projectcapsule/capsule/pkg/api/meta"
	"github.com/projectcapsule/capsule/pkg/api/processor"
	"github.com/projectcapsule/capsule/pkg/runtime/configuration"
	"github.com/projectcapsule/capsule/pkg/runtime/predicates"
	tpl "github.com/projectcapsule/capsule/pkg/template"
)

// NamespaceTrigger is watching for Namespace events, and trigger the
// GlobalTenantResource or TenantResource target the watched object:
// this is used to propagate resources in new freshly created Namespace
// without waiting for the resyncPeriod of the Resources.
type NamespaceTrigger struct {
	client        client.Client
	reader        client.Reader
	log           logr.Logger
	configuration configuration.Configuration
	impersonation *cache.ImpersonationCache
	processor     processor.Processor
	collector     Collector

	globalClients     impersonatedClientLoader[*capsulev1beta2.GlobalTenantResource]
	namespacedClients impersonatedClientLoader[*capsulev1beta2.TenantResource]
	globalStatus      scopedStatusPatcher[*capsulev1beta2.GlobalTenantResource]
	namespacedStatus  scopedStatusPatcher[*capsulev1beta2.TenantResource]
}

func (r *NamespaceTrigger) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) {
	log := ctrllog.FromContext(ctx)

	var namespace corev1.Namespace
	if err := r.client.Get(ctx, request.NamespacedName, &namespace); err != nil {
		if apierrors.IsNotFound(err) {
			log.V(5).Info("Request object not found, could have been deleted after reconcile request")

			return reconcile.Result{}, nil
		}

		return reconcile.Result{}, err
	}

	tntName, found := namespace.GetLabels()[meta.TenantLabel]
	if !found {
		log.V(5).Info("cannot retrieve Tenant ownership from Namespace's Tenant label")

		return reconcile.Result{}, nil
	}

	var tnt capsulev1beta2.Tenant
	if err := r.client.Get(ctx, types.NamespacedName{Name: tntName}, &tnt); err != nil {
		return reconcile.Result{}, err
	}

	log = log.WithValues("tenant", tnt.GetName())

	syncErr := errors.Join(
		r.replicateNamespacedResources(ctx, log, tnt, &namespace),
		r.replicateGlobalResources(ctx, log, tnt, &namespace),
	)

	return reconcile.Result{}, syncErr
}

func (r *NamespaceTrigger) SetupWithManager(mgr ctrl.Manager, ctrlConfig utils.ControllerOptions) error {
	r.client = mgr.GetClient()
	r.reader = mgr.GetAPIReader()

	r.processor = processor.Processor{
		Configuration:                r.configuration,
		GatherClient:                 mgr.GetAPIReader(),
		AllowCrossNamespaceSelection: true,

		Mapper: mgr.GetRESTMapper(),
	}
	r.collector = NewCollector(
		mgr.GetAPIReader(),
		mgr.GetRESTMapper(),
	)
	r.globalClients = impersonatedClientLoader[*capsulev1beta2.GlobalTenantResource]{
		client:        r.client,
		configuration: r.configuration,
		impersonation: r.impersonation,
		resolve:       globalServiceAccount,
	}
	r.namespacedClients = impersonatedClientLoader[*capsulev1beta2.TenantResource]{
		client:        r.client,
		configuration: r.configuration,
		impersonation: r.impersonation,
		resolve:       namespacedServiceAccount,
	}
	r.globalStatus = scopedStatusPatcher[*capsulev1beta2.GlobalTenantResource]{
		client:  r.client,
		reader:  r.reader,
		factory: func() *capsulev1beta2.GlobalTenantResource { return &capsulev1beta2.GlobalTenantResource{} },
		status: func(o *capsulev1beta2.GlobalTenantResource) *capsulev1beta2.TenantResourceCommonStatus {
			return &o.Status.TenantResourceCommonStatus
		},
	}
	r.namespacedStatus = scopedStatusPatcher[*capsulev1beta2.TenantResource]{
		client:  r.client,
		reader:  r.reader,
		factory: func() *capsulev1beta2.TenantResource { return &capsulev1beta2.TenantResource{} },
		status: func(o *capsulev1beta2.TenantResource) *capsulev1beta2.TenantResourceCommonStatus {
			return &o.Status.TenantResourceCommonStatus
		},
	}

	return ctrl.NewControllerManagedBy(mgr).
		Named("NamespaceWatcher").
		For(
			&corev1.Namespace{},
			// Only the Namespaces of a Tenant which are created from now on: the already
			// existing ones are covered by the full reconciliation of the resources.
			builder.WithPredicates(
				predicates.LabelPresentPredicate{Label: meta.TenantLabel},
				predicates.NewCreatedAfterPredicate(time.Now()),
			),
		).
		WithOptions(ctrlConfig.Runtime.ToControllerOptions()).
		Complete(r)
}

// Replicates into the given Namespace every TenantResource of the Tenant.
func (r *NamespaceTrigger) replicateNamespacedResources(
	ctx context.Context,
	log logr.Logger,
	tnt capsulev1beta2.Tenant,
	namespace *corev1.Namespace,
) (syncErr error) {
	var tr capsulev1beta2.TenantResourceList

	nsSet := sets.New[string](tnt.Status.Namespaces...)
	nsSet.Insert(namespace.GetName())

	for ns := range nsSet {
		var list capsulev1beta2.TenantResourceList
		if err := r.client.List(ctx, &list, client.InNamespace(ns)); err != nil {
			log.Error(err, "cannot retrieve TenantResourceList", "namespace", ns)

			return err
		}

		tr.Items = append(tr.Items, list.Items...)
	}

	for _, tntResource := range tr.Items {
		ilog := log.WithValues("tenantresource", tntResource.GetName(), "source", tntResource.GetNamespace())

		if skip, reason := skipScopedReplication(&tntResource, &tntResource.Spec.TenantResourceCommonSpec); skip {
			ilog.V(5).Info("TenantResource is not replicating, ignoring", "reason", reason)

			continue
		}

		if err := r.replicateNamespaced(ctx, ilog, &tntResource, tnt, namespace); err != nil {
			ilog.Error(err, "cannot replicate the TenantResource on the Namespace")

			syncErr = errors.Join(syncErr, err)
		}
	}

	return syncErr
}

// Replicates into the given Namespace every GlobalTenantResource selecting the Tenant.
func (r *NamespaceTrigger) replicateGlobalResources(
	ctx context.Context,
	log logr.Logger,
	tnt capsulev1beta2.Tenant,
	namespace *corev1.Namespace,
) (syncErr error) {
	var gtr capsulev1beta2.GlobalTenantResourceList
	if err := r.client.List(ctx, &gtr); err != nil {
		log.Error(err, "cannot retrieve GlobalTenantResourceList")

		return err
	}

	for _, tntResource := range gtr.Items {
		ilog := log.WithValues("globaltenantresource", tntResource.GetName())

		selector, err := metav1.LabelSelectorAsSelector(&tntResource.Spec.TenantSelector)
		if err != nil {
			ilog.Error(err, "cannot create MatchingLabelsSelector for Global filtering")

			continue
		}

		if !selector.Matches(labels.Set(tnt.GetLabels())) {
			ilog.V(5).Info("Tenant is not selected by the GlobalTenantResource, ignoring")

			continue
		}

		if skip, reason := skipGlobalNamespaceReplication(tntResource); skip {
			ilog.V(5).Info("GlobalTenantResource is not replicating on a Namespace basis, ignoring", "reason", reason)

			continue
		}

		if err := r.replicateGlobal(ctx, ilog, &tntResource, tnt, namespace); err != nil {
			ilog.Error(err, "cannot replicate the GlobalTenantResource on the Namespace")

			syncErr = errors.Join(syncErr, err)
		}
	}

	return syncErr
}

// States whether the given GlobalTenantResource is out of the Namespace scoped
// replication duties, along with the reason.
func skipGlobalNamespaceReplication(tntResource capsulev1beta2.GlobalTenantResource) (bool, string) {
	// Any other scope is not replicating on a per Namespace basis: the accumulated items
	// would not be addressable by a Namespace scope at all.
	if tntResource.Spec.Scope != api.ResourceScopeNamespace {
		return true, "scope is " + string(tntResource.Spec.Scope)
	}

	return skipScopedReplication(&tntResource, &tntResource.Spec.TenantResourceCommonSpec)
}

// Replicates a single GlobalTenantResource into the given Namespace of the given Tenant,
// reporting the outcome on the status without touching the items of the other Namespaces.
func (r *NamespaceTrigger) replicateGlobal(
	ctx context.Context,
	log logr.Logger,
	tntResource *capsulev1beta2.GlobalTenantResource,
	tnt capsulev1beta2.Tenant,
	namespace *corev1.Namespace,
) error {
	// The resolved ServiceAccount is intentionally discarded: posting it to the status is
	// a duty of the GlobalTenantResource controller.
	c, _, err := r.globalClients.Load(ctx, log, tntResource)
	if err != nil {
		return fmt.Errorf("failed to load serviceaccount client: %w", err)
	}

	scope := processor.Scope{
		Tenant:    tnt.GetName(),
		Namespace: namespace.GetName(),
	}

	acc := processor.Accumulator{}

	// Bailing out on a partial accumulation, as the full reconciliation does: reporting it
	// would drop the status entries of the items which could not be gathered, making the
	// full reconciliation lose track of them.
	if err := r.gatherGlobalResources(ctx, c, log, tntResource, tnt, namespace, acc); err != nil {
		return fmt.Errorf("failed to gather resources: %w", err)
	}

	owner := meta.GetLooseOwnerReference(tntResource)

	items, reconcileErr := r.processor.ReconcileNamespace(
		ctx,
		log,
		c,
		tntResource.Status.ProcessedItems,
		acc,
		scopedProcessorOptions(tntResource, &tntResource.Spec.TenantResourceCommonSpec, &owner),
		scope,
	)

	// The items are reported even along an error, since they carry the outcome of the
	// single objects which have been processed.
	statusErr := r.globalStatus.Patch(ctx, tntResource, scope, items)

	return errors.Join(reconcileErr, statusErr)
}

// Replicates a single TenantResource into the given Namespace of the given Tenant,
// reporting the outcome on the status without touching the items of the other Namespaces.
func (r *NamespaceTrigger) replicateNamespaced(
	ctx context.Context,
	log logr.Logger,
	tntResource *capsulev1beta2.TenantResource,
	tnt capsulev1beta2.Tenant,
	namespace *corev1.Namespace,
) error {
	c, _, err := r.namespacedClients.Load(ctx, log, tntResource)
	if err != nil {
		return fmt.Errorf("failed to load serviceaccount client: %w", err)
	}

	scope := processor.Scope{
		Tenant:    tnt.GetName(),
		Namespace: namespace.GetName(),
	}

	acc := processor.Accumulator{}

	if err := r.gatherNamespacedResources(ctx, c, log, tntResource, tnt, namespace, acc); err != nil {
		return fmt.Errorf("failed to gather resources: %w", err)
	}

	// A TenantResource is namespaced, hence it cannot own objects living in another
	// Namespace: no owner reference is set, mirroring the full reconciliation.
	items, reconcileErr := r.processor.ReconcileNamespace(
		ctx,
		log,
		c,
		tntResource.Status.ProcessedItems,
		acc,
		scopedProcessorOptions(tntResource, &tntResource.Spec.TenantResourceCommonSpec, nil),
		scope,
	)

	statusErr := r.namespacedStatus.Patch(ctx, tntResource, scope, items)

	return errors.Join(reconcileErr, statusErr)
}

// Accumulates the items of a GlobalTenantResource the given Namespace must be holding.
func (r *NamespaceTrigger) gatherGlobalResources(
	ctx context.Context,
	c client.Client,
	log logr.Logger,
	tntResource *capsulev1beta2.GlobalTenantResource,
	tnt capsulev1beta2.Tenant,
	namespace *corev1.Namespace,
	acc processor.Accumulator,
) error {
	opts := CollectorOptions{
		Accumulator:               acc,
		AllowClusterScopedObjects: true,
	}

	for resourceIndex, resource := range tntResource.Spec.Resources {
		targeted, err := r.targetsNamespace(log, tnt, resource, namespace, resourceIndex)
		if err != nil {
			return err
		}

		if !targeted {
			continue
		}

		// Sources are loaded cluster-wide, as they can live outside of the target Namespace.
		opts.AllowCrossNamespaceSelection = true

		sources, err := r.collector.CollectNamespacedItems(ctx, c, opts, resource, nil, tnt)
		if err != nil {
			return err
		}

		opts.AllowCrossNamespaceSelection = false

		if err := r.collector.CollectForNamespace(
			ctx,
			c,
			opts,
			tnt,
			strconv.Itoa(resourceIndex),
			resource,
			sources,
			namespace,
		); err != nil {
			return err
		}
	}

	return nil
}

// States whether the given resource specification is replicating on the given Namespace.
func (r *NamespaceTrigger) targetsNamespace(
	log logr.Logger,
	tnt capsulev1beta2.Tenant,
	resource capsulev1beta2.ResourceSpec,
	namespace *corev1.Namespace,
	resourceIndex int,
) (bool, error) {
	selector, err := r.collector.namespaceSelector(tnt, resource)
	if err != nil {
		return false, err
	}

	if !selector.Matches(labels.Set(namespace.GetLabels())) {
		log.V(5).Info("Namespace is not targeted by the resource, ignoring", "resource", resourceIndex)

		return false, nil
	}

	return true, nil
}

// Accumulates the items of a TenantResource the given Namespace must be holding.
func (r *NamespaceTrigger) gatherNamespacedResources(
	ctx context.Context,
	c client.Client,
	log logr.Logger,
	tntResource *capsulev1beta2.TenantResource,
	tnt capsulev1beta2.Tenant,
	namespace *corev1.Namespace,
	acc processor.Accumulator,
) error {
	// The Namespace has just been created, thus it may not have landed on the Tenant status
	// yet: it is a legit replication target nonetheless, and the validator must know about it
	// to not reject the items referring to it.
	allowed := sets.New[string](tnt.Status.Namespaces...)
	allowed.Insert(namespace.GetName())

	// The very same boundaries of the full reconciliation: a Tenant owner must not be able to
	// select cluster scoped objects, nor objects living outside of its own Namespaces.
	opts := CollectorOptions{
		Accumulator:                  acc,
		AllowCrossNamespaceSelection: false,
		AllowClusterScopedObjects:    false,
		ValidatorNamespaces:          tpl.NewNamespaceValidator(false, allowed),
	}

	// The sources of a TenantResource always live in the Namespace it is deployed in.
	source := &corev1.Namespace{
		ObjectMeta: metav1.ObjectMeta{Name: tntResource.GetNamespace()},
	}

	for resourceIndex, resource := range tntResource.Spec.Resources {
		targeted, err := r.targetsNamespace(log, tnt, resource, namespace, resourceIndex)
		if err != nil {
			return err
		}

		if !targeted {
			continue
		}

		sources, err := r.collector.CollectNamespacedItems(ctx, c, opts, resource, source, tnt)
		if err != nil {
			return err
		}

		if err := r.collector.CollectForNamespace(
			ctx,
			c,
			opts,
			tnt,
			strconv.Itoa(resourceIndex),
			resource,
			sources,
			namespace,
		); err != nil {
			return err
		}
	}

	return nil
}