🤖 feat: add WATCH support to aggregated API server storage

internal/aggregated/storage/template.go

@@ -8,13 +8,15 @@ import (
 	"os"
 	"reflect"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/google/uuid"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/apiserver/pkg/registry/rest"
 
 	aggregationv1alpha1 "github.com/coder/coder-k8s/api/aggregation/v1alpha1"
@@ -47,6 +49,7 @@ var (
 	_ rest.Storage              = (*TemplateStorage)(nil)
 	_ rest.Getter               = (*TemplateStorage)(nil)
 	_ rest.Lister               = (*TemplateStorage)(nil)
+	_ rest.Watcher              = (*TemplateStorage)(nil)
 	_ rest.Creater              = (*TemplateStorage)(nil) //nolint:misspell // Kubernetes rest interface name is Creater.
 	_ rest.Updater              = (*TemplateStorage)(nil)
 	_ rest.GracefulDeleter      = (*TemplateStorage)(nil)
@@ -60,6 +63,10 @@ var (
 type TemplateStorage struct {
 	provider       coder.ClientProvider
 	tableConvertor rest.TableConvertor
+	broadcaster    *watch.Broadcaster
+	watchEvents    chan watch.Event
+	watchEventsWG  sync.WaitGroup
+	destroyOnce    sync.Once
 }
 
 // NewTemplateStorage builds codersdk-backed storage for CoderTemplate resources.
@@ -68,10 +75,16 @@ func NewTemplateStorage(provider coder.ClientProvider) *TemplateStorage {
 		panic("assertion failed: template client provider must not be nil")
 	}
 
-	return &TemplateStorage{
+	storage := &TemplateStorage{
 		provider:       provider,
 		tableConvertor: rest.NewDefaultTableConvertor(aggregationv1alpha1.Resource("codertemplates")),
+		broadcaster:    watch.NewBroadcaster(watchBroadcasterQueueLen, watch.WaitIfChannelFull),
+		watchEvents:    make(chan watch.Event, watchBroadcasterQueueLen),
 	}
+	storage.watchEventsWG.Add(1)
+	go storage.dispatchWatchEvents()
+
+	return storage
 }
 
 // New returns an empty CoderTemplate object.
@@ -80,7 +93,23 @@ func (s *TemplateStorage) New() runtime.Object {
 }
 
 // Destroy cleans up storage resources.
-func (s *TemplateStorage) Destroy() {}
+func (s *TemplateStorage) Destroy() {
+	if s == nil {
+		return
+	}
+
+	s.destroyOnce.Do(func() {
+		if s.watchEvents == nil {
+			panic("assertion failed: template watch event queue must not be nil")
+		}
+		close(s.watchEvents)
+		s.watchEventsWG.Wait()
+
+		if s.broadcaster != nil {
+			s.broadcaster.Shutdown()
+		}
+	})
+}
 
 // NamespaceScoped returns true because CoderTemplate is namespaced.
 func (s *TemplateStorage) NamespaceScoped() bool {
@@ -189,6 +218,64 @@ func (s *TemplateStorage) List(ctx context.Context, _ *metainternalversion.ListO
 	return list, nil
 }
 
+// Watch watches CoderTemplate objects backed by codersdk.
+func (s *TemplateStorage) Watch(ctx context.Context, opts *metainternalversion.ListOptions) (watch.Interface, error) {
+	if s == nil {
+		return nil, fmt.Errorf("assertion failed: template storage must not be nil")
+	}
+	if ctx == nil {
+		return nil, fmt.Errorf("assertion failed: context must not be nil")
+	}
+	if s.broadcaster == nil {
+		return nil, fmt.Errorf("assertion failed: template broadcaster must not be nil")
+	}
+
+	requestNamespace, err := namespaceFromRequestContext(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := validateUnsupportedWatchListOptions(opts); err != nil {
+		return nil, apierrors.NewBadRequest(fmt.Sprintf("invalid watch options: %v", err))
+	}
+
+	filter, err := filterForListOptions(requestNamespace, opts)
+	if err != nil {
+		return nil, apierrors.NewBadRequest(fmt.Sprintf("invalid watch options: %v", err))
+	}
+
+	w, err := s.broadcaster.Watch()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create template watcher: %w", err)
+	}
+
+	var timeoutTimer *time.Timer
+	if opts != nil && opts.TimeoutSeconds != nil && *opts.TimeoutSeconds > 0 {
+		timeoutTimer = time.NewTimer(time.Duration(*opts.TimeoutSeconds) * time.Second)
+	}
+
+	go func() {
+		if timeoutTimer == nil {
+			<-ctx.Done()
+			w.Stop()
+			return
+		}
+
+		defer timeoutTimer.Stop()
+		select {
+		case <-ctx.Done():
+		case <-timeoutTimer.C:
+		}
+		w.Stop()
+	}()
+
+	if filter != nil {
+		return watch.Filter(w, filter), nil
+	}
+
+	return w, nil
+}
+
 // Create creates a CoderTemplate through codersdk.
 func (s *TemplateStorage) Create(
 	ctx context.Context,
@@ -205,6 +292,9 @@ func (s *TemplateStorage) Create(
 	if obj == nil {
 		return nil, fmt.Errorf("assertion failed: object must not be nil")
 	}
+	if s.broadcaster == nil {
+		return nil, fmt.Errorf("assertion failed: template broadcaster must not be nil")
+	}
 
 	templateObj, ok := obj.(*aggregationv1alpha1.CoderTemplate)
 	if !ok {
@@ -284,7 +374,14 @@ func (s *TemplateStorage) Create(
 			return nil, coder.MapCoderError(err, aggregationv1alpha1.Resource("codertemplates"), templateObj.Name)
 		}
 
-		return convert.TemplateToK8s(namespace, createdTemplate), nil
+		result := convert.TemplateToK8s(namespace, createdTemplate)
+		if result == nil {
+			return nil, fmt.Errorf("assertion failed: converted template must not be nil")
+		}
+
+		s.enqueueWatchEvent(watch.Added, result.DeepCopy())
+
+		return result, nil
 	}
 
 	request, err := convert.TemplateCreateRequestFromK8s(templateObj, templateName)
@@ -297,7 +394,14 @@ func (s *TemplateStorage) Create(
 		return nil, coder.MapCoderError(err, aggregationv1alpha1.Resource("codertemplates"), templateObj.Name)
 	}
 
-	return convert.TemplateToK8s(namespace, createdTemplate), nil
+	result := convert.TemplateToK8s(namespace, createdTemplate)
+	if result == nil {
+		return nil, fmt.Errorf("assertion failed: converted template must not be nil")
+	}
+
+	s.enqueueWatchEvent(watch.Added, result.DeepCopy())
+
+	return result, nil
 }
 
 // Update applies a template metadata/source reconcile.
@@ -322,6 +426,9 @@ func (s *TemplateStorage) Update(
 	if objInfo == nil {
 		return nil, false, fmt.Errorf("assertion failed: updated object info must not be nil")
 	}
+	if s.broadcaster == nil {
+		return nil, false, fmt.Errorf("assertion failed: template broadcaster must not be nil")
+	}
 	if forceAllowCreate {
 		return nil, false, apierrors.NewMethodNotSupported(
 			aggregationv1alpha1.Resource("codertemplates"),
@@ -531,7 +638,17 @@ func (s *TemplateStorage) Update(
 		return nil, false, err
 	}
 
-	return refreshedObj, false, nil
+	result, ok := refreshedObj.(*aggregationv1alpha1.CoderTemplate)
+	if !ok {
+		return nil, false, fmt.Errorf("assertion failed: expected *CoderTemplate, got %T", refreshedObj)
+	}
+	if result == nil {
+		return nil, false, fmt.Errorf("assertion failed: refreshed template must not be nil")
+	}
+
+	s.enqueueWatchEvent(watch.Modified, result.DeepCopy())
+
+	return result, false, nil
 }
 
 // Delete deletes a CoderTemplate through codersdk.
@@ -550,6 +667,9 @@ func (s *TemplateStorage) Delete(
 	if name == "" {
 		return nil, false, fmt.Errorf("assertion failed: template name must not be empty")
 	}
+	if s.broadcaster == nil {
+		return nil, false, fmt.Errorf("assertion failed: template broadcaster must not be nil")
+	}
 
 	namespace, badNamespaceErr := requiredNamespaceFromRequestContext(ctx)
 	if badNamespaceErr != nil {
@@ -586,9 +706,55 @@ func (s *TemplateStorage) Delete(
 		return nil, false, coder.MapCoderError(err, aggregationv1alpha1.Resource("codertemplates"), name)
 	}
 
+	templateObj := convert.TemplateToK8s(namespace, template)
+	if templateObj == nil {
+		return nil, false, fmt.Errorf("assertion failed: converted template must not be nil")
+	}
+
+	// Emit a Deleted event with the last-known template state.
+	s.enqueueWatchEvent(watch.Deleted, templateObj.DeepCopy())
+
 	return &metav1.Status{Status: metav1.StatusSuccess}, true, nil
 }
 
+func (s *TemplateStorage) dispatchWatchEvents() {
+	defer s.watchEventsWG.Done()
+
+	for event := range s.watchEvents {
+		_ = s.broadcaster.Action(event.Type, event.Object)
+	}
+}
+
+func (s *TemplateStorage) enqueueWatchEvent(action watch.EventType, obj runtime.Object) {
+	if s == nil {
+		panic("assertion failed: template storage must not be nil")
+	}
+	if s.watchEvents == nil {
+		panic("assertion failed: template watch event queue must not be nil")
+	}
+	if obj == nil {
+		panic("assertion failed: template watch event object must not be nil")
+	}
+
+	event := watch.Event{Type: action, Object: obj}
+	select {
+	case s.watchEvents <- event:
+		return
+	default:
+	}
+
+	// Keep mutation request handlers non-blocking under watch backpressure by
+	// dropping the oldest queued event and retaining the most recent state.
+	select {
+	case <-s.watchEvents:
+	default:
+	}
+	select {
+	case s.watchEvents <- event:
+	default:
+	}
+}
+
 // ConvertToTable converts a template object or list into kubectl table output.
 func (s *TemplateStorage) ConvertToTable(ctx context.Context, object, tableOptions runtime.Object) (*metav1.Table, error) {
 	if s == nil {

internal/aggregated/storage/watch.go

@@ -0,0 +1,110 @@
+package storage
+
+import (
+	"fmt"
+
+	metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/watch"
+)
+
+// watchBroadcasterQueueLen is the default queue length for watch broadcasters.
+const watchBroadcasterQueueLen = 100
+
+// supportedWatchFieldSelectors lists the metadata fields supported for watch field selectors.
+var supportedWatchFieldSelectors = map[string]bool{
+	"metadata.name":      true,
+	"metadata.namespace": true,
+}
+
+// validateFieldSelector checks that all field selector requirements use supported fields.
+func validateFieldSelector(sel fields.Selector) error {
+	if sel == nil || sel.Empty() {
+		return nil
+	}
+
+	reqs := sel.Requirements()
+	for _, req := range reqs {
+		if !supportedWatchFieldSelectors[req.Field] {
+			return fmt.Errorf(
+				"field selector %q is not supported; supported fields: metadata.name, metadata.namespace",
+				req.Field,
+			)
+		}
+	}
+
+	return nil
+}
+
+// validateUnsupportedWatchListOptions rejects watch-list options that this storage
+// does not implement yet.
+func validateUnsupportedWatchListOptions(opts *metainternalversion.ListOptions) error {
+	if opts == nil {
+		return nil
+	}
+
+	if opts.SendInitialEvents != nil && *opts.SendInitialEvents {
+		return fmt.Errorf("sendInitialEvents=true is not supported for this watch endpoint")
+	}
+	if opts.ResourceVersionMatch != "" {
+		return fmt.Errorf(
+			"resourceVersionMatch %q is not supported for this watch endpoint",
+			opts.ResourceVersionMatch,
+		)
+	}
+
+	return nil
+}
+
+// filterForListOptions builds a watch.FilterFunc that applies namespace, label, and field selector filtering.
+// Returns nil if no filtering is needed.
+func filterForListOptions(requestNamespace string, opts *metainternalversion.ListOptions) (watch.FilterFunc, error) {
+	var labelSel labels.Selector
+	var fieldSel fields.Selector
+
+	if opts != nil {
+		if opts.LabelSelector != nil && !opts.LabelSelector.Empty() {
+			labelSel = opts.LabelSelector
+		}
+		if opts.FieldSelector != nil && !opts.FieldSelector.Empty() {
+			if err := validateFieldSelector(opts.FieldSelector); err != nil {
+				return nil, err
+			}
+			fieldSel = opts.FieldSelector
+		}
+	}
+
+	if requestNamespace == "" && labelSel == nil && fieldSel == nil {
+		return nil, nil
+	}
+
+	return func(in watch.Event) (watch.Event, bool) {
+		obj, ok := in.Object.(metav1.ObjectMetaAccessor)
+		if !ok {
+			return in, true
+		}
+
+		meta := obj.GetObjectMeta()
+		if requestNamespace != "" && meta.GetNamespace() != requestNamespace {
+			return in, false
+		}
+
+		if labelSel != nil && !labelSel.Matches(labels.Set(meta.GetLabels())) {
+			return in, false
+		}
+
+		if fieldSel != nil {
+			fieldSet := fields.Set{
+				"metadata.name":      meta.GetName(),
+				"metadata.namespace": meta.GetNamespace(),
+			}
+			if !fieldSel.Matches(fieldSet) {
+				return in, false
+			}
+		}
+
+		return in, true
+	}, nil
+}