fix: v4 changes

k8-operator/internal/api/api.go

@@ -3,7 +3,7 @@ package api
 import (
 	"fmt"
 
-	"github.com/Infisical/infisical/k8-operator/packages/model"
+	"github.com/Infisical/infisical/k8-operator/internal/model"
 	"github.com/go-resty/resty/v2"
 )
 

k8-operator/internal/controller/infisicalsecret_controller.go

@@ -203,14 +203,14 @@ func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	if infisicalSecretCRD.Spec.InstantUpdates {
 		logger.Info("Instant updates are enabled")
 
-		if err := r.EnsureEventStream(ctx, logger, &infisicalSecretCRD); err != nil {
+		if err := handler.OpenInstantUpdatesStream(ctx, logger, &infisicalSecretCRD, infisicalSecretResourceVariablesMap, r.SourceCh); err != nil {
 			logger.Error(err, fmt.Sprintf("unable to ensure event stream. Will requeue after [requeueTime=%v]", requeueTime))
 			return ctrl.Result{
 				RequeueAfter: requeueTime,
 			}, nil
 		}
 	} else {
-		r.CloseEventStream(ctx, logger, &infisicalSecretCRD)
+		handler.CloseInstantUpdatesStream(ctx, logger, &infisicalSecretCRD, infisicalSecretResourceVariablesMap)
 	}
 
 	// Sync again after the specified time

k8-operator/internal/services/infisicalsecret/handler.go

@@ -7,6 +7,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 
 	"github.com/Infisical/infisical/k8-operator/api/v1alpha1"
 	"github.com/Infisical/infisical/k8-operator/internal/api"
@@ -100,3 +101,22 @@ func (h *InfisicalSecretHandler) SetInfisicalAutoRedeploymentReady(ctx context.C
 	}
 	reconciler.SetInfisicalAutoRedeploymentReady(ctx, logger, infisicalSecret, numDeployments, errorToConditionOn)
 }
+
+func (h *InfisicalSecretHandler) CloseInstantUpdatesStream(ctx context.Context, logger logr.Logger, infisicalSecret *v1alpha1.InfisicalSecret, resourceVariablesMap map[string]util.ResourceVariables) error {
+	reconciler := &InfisicalSecretReconciler{
+		Client:            h.Client,
+		Scheme:            h.Scheme,
+		IsNamespaceScoped: h.IsNamespaceScoped,
+	}
+	return reconciler.CloseInstantUpdatesStream(ctx, logger, infisicalSecret, resourceVariablesMap)
+}
+
+// Ensures that SSE stream is open, incase if the stream is already opened - this is a noop
+func (h *InfisicalSecretHandler) OpenInstantUpdatesStream(ctx context.Context, logger logr.Logger, infisicalSecret *v1alpha1.InfisicalSecret, resourceVariablesMap map[string]util.ResourceVariables, eventCh chan<- event.TypedGenericEvent[client.Object]) error {
+	reconciler := &InfisicalSecretReconciler{
+		Client:            h.Client,
+		Scheme:            h.Scheme,
+		IsNamespaceScoped: h.IsNamespaceScoped,
+	}
+	return reconciler.OpenInstantUpdatesStream(ctx, logger, infisicalSecret, resourceVariablesMap, eventCh)
+}

k8-operator/internal/services/infisicalsecret/reconciler.go

@@ -36,9 +36,7 @@ const FINALIZER_NAME = "secrets.finalizers.infisical.com"
 
 type InfisicalSecretReconciler struct {
 	client.Client
-	Scheme *runtime.Scheme
-
-	SourceCh          chan event.TypedGenericEvent[client.Object]
+	Scheme            *runtime.Scheme
 	IsNamespaceScoped bool
 }
 
@@ -533,7 +531,33 @@ func (r *InfisicalSecretReconciler) ReconcileInfisicalSecret(ctx context.Context
 	return secretsCount, nil
 }
 
-func (r *InfisicalSecretReconciler) EnsureEventStream(ctx context.Context, logger logr.Logger, secret *v1alpha1.InfisicalSecret, variables util.ResourceVariables) error {
+func (r *InfisicalSecretReconciler) CloseInstantUpdatesStream(ctx context.Context, logger logr.Logger, infisicalSecret *v1alpha1.InfisicalSecret, resourceVariablesMap map[string]util.ResourceVariables) error {
+	if infisicalSecret == nil {
+		return fmt.Errorf("infisicalSecret is nil")
+	}
+
+	variables := r.getResourceVariables(*infisicalSecret, resourceVariablesMap)
+
+	if !variables.AuthDetails.IsMachineIdentityAuth {
+		return fmt.Errorf("only machine identity is supported for subscriptions")
+	}
+
+	conn := variables.ServerSentEvents
+
+	if _, ok := conn.Get(); ok {
+		conn.Close()
+	}
+
+	return nil
+}
+
+func (r *InfisicalSecretReconciler) OpenInstantUpdatesStream(ctx context.Context, logger logr.Logger, infisicalSecret *v1alpha1.InfisicalSecret, resourceVariablesMap map[string]util.ResourceVariables, eventCh chan<- event.TypedGenericEvent[client.Object]) error {
+	if infisicalSecret == nil {
+		return fmt.Errorf("infisicalSecret is nil")
+	}
+
+	variables := r.getResourceVariables(*infisicalSecret, resourceVariablesMap)
+
 	if !variables.AuthDetails.IsMachineIdentityAuth {
 		return fmt.Errorf("only machine identity is supported for subscriptions")
 	}
@@ -613,8 +637,8 @@ func (r *InfisicalSecretReconciler) EnsureEventStream(ctx context.Context, logge
 			select {
 			case ev := <-events:
 				logger.Info("Received SSE Event", "event", ev)
-				r.SourceCh <- event.TypedGenericEvent[client.Object]{
-					Object: secret,
+				eventCh <- event.TypedGenericEvent[client.Object]{
+					Object: infisicalSecret,
 				}
 			case err := <-errors:
 				logger.Error(err, "Error occurred")
@@ -627,17 +651,3 @@ func (r *InfisicalSecretReconciler) EnsureEventStream(ctx context.Context, logge
 
 	return nil
 }
-
-func (r *InfisicalSecretReconciler) CloseEventStream(ctx context.Context, logger logr.Logger, variables util.ResourceVariables) error {
-	if !variables.AuthDetails.IsMachineIdentityAuth {
-		return fmt.Errorf("only machine identity is supported for subscriptions")
-	}
-
-	conn := variables.ServerSentEvents
-
-	if _, ok := conn.Get(); ok {
-		conn.Close()
-	}
-
-	return nil
-}

k8-operator/internal/util/models.go

@@ -3,7 +3,7 @@ package util
 import (
 	"context"
 
-	"github.com/Infisical/infisical/k8-operator/packages/util/sse"
+	"github.com/Infisical/infisical/k8-operator/internal/util/sse"
 	infisicalSdk "github.com/infisical/go-sdk"
 )