fix: changes

k8-operator/controllers/infisicalsecret/infisicalsecret_controller.go

@@ -7,14 +7,11 @@ import (
 
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/types"
 	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/event"
-	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
-	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	defaultErrors "errors"
@@ -62,6 +59,8 @@ func (r *InfisicalSecretReconciler) GetLogger(req ctrl.Request) logr.Logger {
 func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	logger := r.GetLogger(req)
 
+	logger.Info("Reconcile called")
+
 	var infisicalSecretCRD secretsv1alpha1.InfisicalSecret
 	requeueTime := time.Minute // seconds
 
@@ -181,8 +180,8 @@ func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	}
 
 	if infisicalSecretCRD.Spec.InstantUpdates {
-		logger.Info("Event watcher enabled")
-		// ensure event watcher is open
+		logger.Info("Instant updates are enabled")
+
 		if err := r.EnsureEventStream(ctx, logger, &infisicalSecretCRD); err != nil {
 			logger.Error(err, fmt.Sprintf("unable to ensure event stream. Will requeue after [requeueTime=%v]", requeueTime))
 			return ctrl.Result{
@@ -190,7 +189,6 @@ func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 			}, nil
 		}
 	} else {
-		// ensure event stream is closed
 		r.CloseEventStream(ctx, logger, &infisicalSecretCRD)
 	}
 
@@ -207,15 +205,10 @@ func (r *InfisicalSecretReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		Watches(
 			&source.Channel{Source: r.SourceCh},
-			handler.EnqueueRequestsFromMapFunc(r.findSecretsForCluster),
+			&util.EnqueueDelayedEventHandler{Delay: time.Second * 3},
 		).
 		For(&secretsv1alpha1.InfisicalSecret{}, builder.WithPredicates(predicate.Funcs{
-			GenericFunc: func(ge event.GenericEvent) bool {
-				fmt.Println("Generic event recieved")
-				return true
-			},
 			UpdateFunc: func(e event.UpdateEvent) bool {
-				println("UpdateFunc event recieved")
 				if e.ObjectOld.GetGeneration() == e.ObjectNew.GetGeneration() {
 					return false // Skip reconciliation for status-only changes
 				}
@@ -241,31 +234,3 @@ func (r *InfisicalSecretReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Complete(r)
 
 }
-
-func (r *InfisicalSecretReconciler) findSecretsForCluster(o client.Object) []reconcile.Request {
-	ctx := context.Background()
-	secrets := &secretsv1alpha1.InfisicalSecretList{}
-
-	requests := []reconcile.Request{}
-
-	if err := r.List(ctx, secrets); err != nil {
-		fmt.Println(err)
-		return requests
-	}
-
-	for _, sec := range secrets.Items {
-		if sec.GetName() == o.GetName() && sec.GetNamespace() == o.GetNamespace() {
-			requests = append(requests, reconcile.Request{
-				NamespacedName: types.NamespacedName{
-					Namespace: o.GetNamespace(),
-					Name:      o.GetName(),
-				},
-			})
-			break
-		}
-	}
-
-	fmt.Println(requests)
-
-	return requests
-}

k8-operator/controllers/infisicalsecret/infisicalsecret_helper.go

@@ -469,7 +469,7 @@ func (r *InfisicalSecretReconciler) getResourceVariables(infisicalSecret v1alpha
 			InfisicalClient:  client,
 			CancelCtx:        cancel,
 			AuthDetails:      util.AuthenticationDetails{},
-			ServerSentEvents: sse.NewConnectionRegistry(),
+			ServerSentEvents: sse.NewConnectionRegistry(ctx),
 		}
 
 		resourceVariables = infisicalSecretResourceVariablesMap[string(infisicalSecret.UID)]
@@ -509,7 +509,7 @@ func (r *InfisicalSecretReconciler) ReconcileInfisicalSecret(ctx context.Context
 			InfisicalClient:  infisicalClient,
 			CancelCtx:        cancelCtx,
 			AuthDetails:      authDetails,
-			ServerSentEvents: sse.NewConnectionRegistry(),
+			ServerSentEvents: sse.NewConnectionRegistry(ctx),
 		})
 	}
 
@@ -638,41 +638,43 @@ func (r *InfisicalSecretReconciler) EnsureEventStream(ctx context.Context, logge
 
 		req, err := http.NewRequestWithContext(ctx, "POST", fmt.Sprintf("%s/v1/events/subscribe/project-events", api.API_HOST_URL), strings.NewReader(string(body)))
 
+		if err != nil {
+			return nil, err
+		}
+
 		for k, v := range headers {
 			req.Header.Set(k, v)
 		}
 
-		return req, err
+		return req, nil
 	})
 
 	if err != nil {
 		return fmt.Errorf("unable to connect to SSE server [err=%s]", err)
 	}
 
-outer:
-	for {
-		select {
-		case ev := <-events:
-			logger.Info("Received event", "secret", secret, "event", ev)
-			r.SourceCh <- event.GenericEvent{
-				Object: secret,
+	go func() {
+	outer:
+		for {
+			select {
+			case ev := <-events:
+				logger.Info("Received SSE Event", "event", ev)
+				r.SourceCh <- event.GenericEvent{
+					Object: secret,
+				}
+			case err := <-errors:
+				logger.Error(err, "Error occurred")
+				break outer
+			case <-ctx.Done():
+				break outer
 			}
-			logger.Info("Send to channel")
-		case err := <-errors:
-			logger.Error(err, "Error occurred")
-			break outer
-		case <-ctx.Done():
-			logger.Info("Context done")
-			break outer
 		}
-	}
+	}()
 
 	return nil
 }
 
 func (r *InfisicalSecretReconciler) CloseEventStream(ctx context.Context, logger logr.Logger, secret *secretsv1alpha1.InfisicalSecret) error {
-	logger.Info("Event watcher disabled")
-
 	if secret == nil {
 		return fmt.Errorf("infisicalSecret is nil")
 	}

k8-operator/packages/util/handler.go

@@ -1,7 +1,6 @@
 package util
 
 import (
-	"fmt"
 	"math/rand"
 	"time"
 
@@ -37,7 +36,6 @@ func (e *EnqueueDelayedEventHandler) Delete(_ event.DeleteEvent, _ workqueue.Rat
 }
 
 func (e *EnqueueDelayedEventHandler) Generic(evt event.GenericEvent, q workqueue.RateLimitingInterface) {
-	fmt.Println(evt)
 	if evt.Object == nil {
 		return
 	}

k8-operator/packages/util/models.go

@@ -11,5 +11,5 @@ type ResourceVariables struct {
 	InfisicalClient  infisicalSdk.InfisicalClientInterface
 	CancelCtx        context.CancelFunc
 	AuthDetails      AuthenticationDetails
-	ServerSentEvents sse.ConnectionRegistry
+	ServerSentEvents *sse.ConnectionRegistry
 }

k8-operator/packages/util/sse/sse.go

@@ -4,134 +4,162 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"sync"
 	"time"
 )
 
-// ConnectionMeta holds metadata about an active SSE connection
 type ConnectionMeta struct {
 	EventChan  <-chan SSEEvent
 	ErrorChan  <-chan error
-	Cancel     context.CancelFunc
 	LastPingAt time.Time
+	Cancel     context.CancelFunc
 }
 
-// ConnectionRegistry manages a single SSE connection with a shared client
 type ConnectionRegistry struct {
-	Ticker *time.Ticker
+	Ctx    context.Context
 	meta   *ConnectionMeta
 	client SSEClient
+	mu     sync.RWMutex
+
+	monitorCancel context.CancelFunc
+	monitorCtx    context.Context
 }
 
-// NewConnectionRegistry creates a new registry
-func NewConnectionRegistry() ConnectionRegistry {
-	return ConnectionRegistry{
-		Ticker: time.NewTicker(time.Second * 30),
-		client: NewClient(),
+func NewConnectionRegistry(ctx context.Context) *ConnectionRegistry {
+	monitorCtx, monitorCancel := context.WithCancel(ctx)
+	return &ConnectionRegistry{
+		Ctx:           ctx,
+		client:        NewClient(),
+		monitorCtx:    monitorCtx,
+		monitorCancel: monitorCancel,
 	}
 }
 
-// GetOrCreate returns existing connection or creates a new one
-func (r *ConnectionRegistry) GetOrCreate(
-	onBuild func() (*http.Request, error),
-) (*ConnectionMeta, error) {
-	// First try to get existing connection
-	if r.meta != nil {
-		return r.meta, nil
-	}
-
-	// Create new connection
-	req, err := onBuild()
-	if err != nil {
-		return nil, fmt.Errorf("failed to build request: %w", err)
-	}
-
-	// Add cancellation context
-	ctx, cancel := context.WithCancel(context.Background())
-
+// create creates a new connection
+func (r *ConnectionRegistry) create(req *http.Request) (*ConnectionMeta, error) {
+	// Create new connection using provided request
 	eventChan, errorChan, err := r.client.Connect(req)
 	if err != nil {
-		cancel()
 		return nil, fmt.Errorf("failed to connect: %w", err)
 	}
 
 	meta := &ConnectionMeta{
 		EventChan:  eventChan,
 		ErrorChan:  errorChan,
-		Cancel:     cancel,
 		LastPingAt: time.Now(),
 	}
 
 	r.meta = meta
 
-	// Start cleanup monitor for this connection
-	go r.monitor(ctx, meta)
+	// Start cleanup monitor for this connection (NON-BLOCKING)
+	go r.monitor(meta)
 
+	println("Creating new connection\n")
 	return meta, nil
 }
 
 // Get retrieves the existing connection
 func (r *ConnectionRegistry) Get() (*ConnectionMeta, bool) {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
 	return r.meta, r.meta != nil
 }
 
 // Close closes the connection
 func (r *ConnectionRegistry) Close() {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
 	if r.meta != nil {
-		r.meta.Cancel()
+		if r.meta.Cancel != nil {
+			r.meta.Cancel()
+		}
 		r.meta = nil
 	}
+
+	// Cancel the monitor
+	if r.monitorCancel != nil {
+		r.monitorCancel()
+	}
 }
 
 // IsConnected returns whether there's an active connection
 func (r *ConnectionRegistry) IsConnected() bool {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
 	return r.meta != nil
 }
 
-// monitorConnection watches for connection closure and cleans up
-func (r *ConnectionRegistry) monitor(ctx context.Context, meta *ConnectionMeta) {
-outer:
-	for range r.Ticker.C {
+// UpdateLastPing updates the last ping time
+func (r *ConnectionRegistry) UpdateLastPing() {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.meta != nil {
+		r.meta.LastPingAt = time.Now()
+	}
+}
+
+// monitor watches for connection closure and cleans up
+func (r *ConnectionRegistry) monitor(meta *ConnectionMeta) {
+	ticker := time.NewTicker(30 * time.Second)
+	defer ticker.Stop()
+
+	for {
 		select {
-		case <-ctx.Done():
-			break outer
-		default:
-			if r.IsConnected() && time.Since(r.meta.LastPingAt) > 2*time.Minute {
-				fmt.Println("Last ping was more than 2 minutes ago")
-				r.Close()
-				break outer
-			} else {
-				fmt.Println("Last ping was within the last 2 minutes")
+		case <-r.monitorCtx.Done():
+			// Context cancelled, exit monitor
+			return
+
+		case <-ticker.C:
+			r.mu.RLock()
+			currentMeta := r.meta
+			r.mu.RUnlock()
+
+			// Check if this monitor is still relevant
+			if currentMeta != meta {
+				// This connection has been replaced, exit monitor
+				return
+			}
+
+			if currentMeta != nil && time.Since(currentMeta.LastPingAt) > 2*time.Minute {
+				fmt.Println("Last ping was more than 2 minutes ago, closing connection")
+				r.mu.Lock()
+				if r.meta == meta { // Double-check under lock
+					if r.meta.Cancel != nil {
+						r.meta.Cancel()
+					}
+					r.meta = nil
+				}
+				r.mu.Unlock()
+				return // Exit monitor after cleanup
 			}
 		}
 	}
-
-	// Clean up from registry
-	if r.meta == meta {
-		r.meta = nil
-	}
-}
-
-// ConnectionInfo provides read-only info about a connection
-type ConnectionInfo struct {
-	LastPingAt time.Time
 }
 
 // Subscribe provides a convenient way to get events from the connection
-func (r *ConnectionRegistry) Subscribe(
-	onBuild func() (*http.Request, error),
-) (<-chan SSEEvent, <-chan error, error) {
-	meta, err := r.GetOrCreate(onBuild)
+func (r *ConnectionRegistry) Subscribe(build func() (*http.Request, error)) (<-chan SSEEvent, <-chan error, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	// Get existing connection if available
+	if r.meta != nil {
+		return r.meta.EventChan, r.meta.ErrorChan, nil
+	}
+
+	req, err := build()
+
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Create new connection if none exists
+	meta, err := r.create(req)
+
 	if err != nil {
 		return nil, nil, err
 	}
 
 	return meta.EventChan, meta.ErrorChan, nil
 }
-
-// Reconnect closes existing connection and creates a new one
-func (r *ConnectionRegistry) Reconnect(
-	onBuild func() (*http.Request, error),
-) (*ConnectionMeta, error) {
-	r.Close()
-	return r.GetOrCreate(onBuild)
-}