Merge pull request #4321 from Infisical/sid/k8s-operator

feat: support `InstantUpdates` in k8s operator

backend/src/ee/services/event/event-sse-stream.ts

@@ -123,7 +123,7 @@ export function createEventStreamClient(redis: Redis, options: IEventStreamClien
 
     await redis.set(key, "1", "EX", 60);
 
-    stream.push("1");
+    send({ type: "ping" });
   };
 
   const close = () => {

backend/src/ee/services/permission/default-roles.ts

@@ -13,6 +13,7 @@ import {
   ProjectPermissionPkiSubscriberActions,
   ProjectPermissionPkiTemplateActions,
   ProjectPermissionSecretActions,
+  ProjectPermissionSecretEventActions,
   ProjectPermissionSecretRotationActions,
   ProjectPermissionSecretScanningConfigActions,
   ProjectPermissionSecretScanningDataSourceActions,
@@ -252,6 +253,16 @@ const buildAdminPermissionRules = () => {
     ProjectPermissionSub.SecretScanningConfigs
   );
 
+  can(
+    [
+      ProjectPermissionSecretEventActions.SubscribeCreated,
+      ProjectPermissionSecretEventActions.SubscribeDeleted,
+      ProjectPermissionSecretEventActions.SubscribeUpdated,
+      ProjectPermissionSecretEventActions.SubscribeImportMutations
+    ],
+    ProjectPermissionSub.SecretEvents
+  );
+
   return rules;
 };
 
@@ -455,6 +466,16 @@ const buildMemberPermissionRules = () => {
 
   can([ProjectPermissionSecretScanningConfigActions.Read], ProjectPermissionSub.SecretScanningConfigs);
 
+  can(
+    [
+      ProjectPermissionSecretEventActions.SubscribeCreated,
+      ProjectPermissionSecretEventActions.SubscribeDeleted,
+      ProjectPermissionSecretEventActions.SubscribeUpdated,
+      ProjectPermissionSecretEventActions.SubscribeImportMutations
+    ],
+    ProjectPermissionSub.SecretEvents
+  );
+
   return rules;
 };
 
@@ -505,6 +526,16 @@ const buildViewerPermissionRules = () => {
 
   can([ProjectPermissionSecretScanningConfigActions.Read], ProjectPermissionSub.SecretScanningConfigs);
 
+  can(
+    [
+      ProjectPermissionSecretEventActions.SubscribeCreated,
+      ProjectPermissionSecretEventActions.SubscribeDeleted,
+      ProjectPermissionSecretEventActions.SubscribeUpdated,
+      ProjectPermissionSecretEventActions.SubscribeImportMutations
+    ],
+    ProjectPermissionSub.SecretEvents
+  );
+
   return rules;
 };
 

k8-operator/api/v1alpha1/infisicalsecret_types.go

@@ -160,6 +160,9 @@ type InfisicalSecretSpec struct {
 
 	// +kubebuilder:validation:Optional
 	TLS TLSConfig `json:"tls"`
+
+	// +kubebuilder:default:=false
+	InstantUpdates bool `json:"instantUpdates"`
 }
 
 // InfisicalSecretStatus defines the observed state of InfisicalSecret

k8-operator/config/crd/bases/secrets.infisical.com_infisicalsecrets.yaml

@@ -314,6 +314,9 @@ spec:
               hostAPI:
                 description: Infisical host to pull secrets from
                 type: string
+              instantUpdates:
+                default: false
+                type: boolean
               managedKubeConfigMapReferences:
                 items:
                   properties:
@@ -469,6 +472,7 @@ spec:
                 - secretNamespace
                 type: object
             required:
+            - instantUpdates
             - resyncInterval
             type: object
           status:

k8-operator/config/samples/crd/infisicalsecret/infisicalSecretCrd.yaml

@@ -9,6 +9,7 @@ metadata:
 spec:
   hostAPI: http://localhost:8080/api
   resyncInterval: 10
+  instantUpdates: false
   # tls:
   #   caRef:
   #     secretName: custom-ca-certificate

k8-operator/config/samples/serviceTokenSecret.yaml

@@ -1,7 +1,7 @@
-apiVersion: v1
-kind: Secret
-metadata:
-  name: service-token
-type: Opaque
-data:
-  infisicalToken: <base64 infisical token here>
+# apiVersion: v1
+# kind: Secret
+# metadata:
+#   name: service-token
+# type: Opaque
+# data:
+#   infisicalToken: <base64 infisical token here>

k8-operator/config/samples/universalAuthIdentitySecret.yaml

@@ -4,5 +4,5 @@ metadata:
   name: universal-auth-credentials
 type: Opaque
 stringData:
-  clientId: da81e27e-1885-47d9-9ea3-ec7d4d807bb6
-  clientSecret: 2772414d440fe04d8b975f5fe25acd0fbfe71b2a4a420409eb9ac6f5ae6c1e98
+  clientId: your-client-id-here
+  clientSecret: your-client-secret-here

k8-operator/internal/api/api.go

@@ -1,8 +1,11 @@
 package api
 
 import (
+	"encoding/json"
 	"fmt"
+	"net/http"
 
+	"github.com/Infisical/infisical/k8-operator/internal/model"
 	"github.com/go-resty/resty/v2"
 )
 
@@ -146,3 +149,85 @@ func CallGetProjectByID(httpClient *resty.Client, request GetProjectByIDRequest)
 	return projectResponse, nil
 
 }
+
+func CallGetProjectByIDv2(httpClient *resty.Client, request GetProjectByIDRequest) (model.Project, error) {
+	var projectResponse model.Project
+
+	response, err := httpClient.
+		R().SetResult(&projectResponse).
+		SetHeader("User-Agent", USER_AGENT_NAME).
+		Get(fmt.Sprintf("%s/v2/workspace/%s", API_HOST_URL, request.ProjectID))
+
+	if err != nil {
+		return model.Project{}, fmt.Errorf("CallGetProject: Unable to complete api request [err=%s]", err)
+	}
+
+	if response.IsError() {
+		return model.Project{}, fmt.Errorf("CallGetProject: Unsuccessful response: [response=%s]", response)
+	}
+
+	return projectResponse, nil
+
+}
+
+func CallSubscribeProjectEvents(httpClient *resty.Client, projectId, secretsPath, envSlug, token string) (*http.Response, error) {
+	conditions := &SubscribeProjectEventsRequestCondition{
+		SecretPath:      secretsPath,
+		EnvironmentSlug: envSlug,
+	}
+
+	body, err := json.Marshal(&SubscribeProjectEventsRequest{
+		ProjectID: projectId,
+		Register: []SubscribeProjectEventsRequestRegister{
+			{
+				Event:      "secret:create",
+				Conditions: conditions,
+			},
+			{
+				Event:      "secret:update",
+				Conditions: conditions,
+			},
+			{
+				Event:      "secret:delete",
+				Conditions: conditions,
+			},
+			{
+				Event:      "secret:import-mutation",
+				Conditions: conditions,
+			},
+		},
+	})
+
+	if err != nil {
+		return nil, fmt.Errorf("CallSubscribeProjectEvents: Unable to marshal body [err=%s]", err)
+	}
+
+	response, err := httpClient.
+		R().
+		SetDoNotParseResponse(true).
+		SetHeader("User-Agent", USER_AGENT_NAME).
+		SetHeader("Content-Type", "application/json").
+		SetHeader("Accept", "text/event-stream").
+		SetHeader("Connection", "keep-alive").
+		SetHeader("Authorization", fmt.Sprint("Bearer ", token)).
+		SetBody(body).
+		Post(fmt.Sprintf("%s/v1/events/subscribe/project-events", API_HOST_URL))
+
+	if err != nil {
+		return nil, fmt.Errorf("CallSubscribeProjectEvents: Unable to complete api request [err=%s]", err)
+	}
+
+	if response.IsError() {
+		data := struct {
+			Message string `json:"message"`
+		}{}
+
+		if err := json.NewDecoder(response.RawBody()).Decode(&data); err != nil {
+			return nil, err
+		}
+
+		return nil, fmt.Errorf("CallSubscribeProjectEvents: Unsuccessful response: [message=%s]", data.Message)
+	}
+
+	return response.RawResponse, nil
+}

k8-operator/internal/api/models.go

@@ -206,3 +206,20 @@ type GetProjectByIDRequest struct {
 type GetProjectByIDResponse struct {
 	Project model.Project `json:"workspace"`
 }
+
+type SubscribeProjectEventsRequestRegister struct {
+	Event      string                                  `json:"event"`
+	Conditions *SubscribeProjectEventsRequestCondition `json:"conditions"`
+}
+
+type SubscribeProjectEventsRequestCondition struct {
+	EnvironmentSlug string `json:"environmentSlug"`
+	SecretPath      string `json:"secretPath"`
+}
+
+type SubscribeProjectEventsRequest struct {
+	ProjectID string                                  `json:"projectId"`
+	Register  []SubscribeProjectEventsRequestRegister `json:"register"`
+}
+
+type SubscribeProjectEventsResponse struct{}

k8-operator/internal/controller/infisicalpushsecret_controller.go

@@ -231,7 +231,6 @@ func (r *InfisicalPushSecretReconciler) Reconcile(ctx context.Context, req ctrl.
 }
 
 func (r *InfisicalPushSecretReconciler) SetupWithManager(mgr ctrl.Manager) error {
-
 	// Custom predicate that allows both spec changes and deletions
 	specChangeOrDelete := predicate.Funcs{
 		UpdateFunc: func(e event.UpdateEvent) bool {

k8-operator/internal/controller/infisicalsecret_controller.go

@@ -31,6 +31,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	secretsv1alpha1 "github.com/Infisical/infisical/k8-operator/api/v1alpha1"
 	"github.com/Infisical/infisical/k8-operator/internal/controllerhelpers"
@@ -41,8 +42,10 @@ import (
 // InfisicalSecretReconciler reconciles a InfisicalSecret object
 type InfisicalSecretReconciler struct {
 	client.Client
-	BaseLogger        logr.Logger
-	Scheme            *runtime.Scheme
+	BaseLogger logr.Logger
+	Scheme     *runtime.Scheme
+
+	SourceCh          chan event.TypedGenericEvent[client.Object]
 	Namespace         string
 	IsNamespaceScoped bool
 }
@@ -74,7 +77,6 @@ func (r *InfisicalSecretReconciler) GetLogger(req ctrl.Request) logr.Logger {
 // For more details, check Reconcile and its Result here:
 // - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.21.0/pkg/reconcile
 func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-
 	logger := r.GetLogger(req)
 
 	var infisicalSecretCRD secretsv1alpha1.InfisicalSecret
@@ -196,6 +198,20 @@ func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		}, nil
 	}
 
+	if infisicalSecretCRD.Spec.InstantUpdates {
+		if err := handler.OpenInstantUpdatesStream(ctx, logger, &infisicalSecretCRD, infisicalSecretResourceVariablesMap, r.SourceCh); err != nil {
+			requeueTime = time.Second * 10
+			logger.Info(fmt.Sprintf("event stream failed. Will requeue after [requeueTime=%v] [error=%s]", requeueTime, err.Error()))
+			return ctrl.Result{
+				RequeueAfter: requeueTime,
+			}, nil
+		}
+
+		logger.Info("Instant updates are enabled")
+	} else {
+		handler.CloseInstantUpdatesStream(ctx, logger, &infisicalSecretCRD, infisicalSecretResourceVariablesMap)
+	}
+
 	// Sync again after the specified time
 	logger.Info(fmt.Sprintf("Successfully synced %d secrets. Operator will requeue after [%v]", secretsCount, requeueTime))
 	return ctrl.Result{
@@ -204,7 +220,12 @@ func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 }
 
 func (r *InfisicalSecretReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	r.SourceCh = make(chan event.TypedGenericEvent[client.Object])
+
 	return ctrl.NewControllerManagedBy(mgr).
+		WatchesRawSource(
+			source.Channel[client.Object](r.SourceCh, &util.EnqueueDelayedEventHandler{Delay: time.Second * 10}),
+		).
 		For(&secretsv1alpha1.InfisicalSecret{}, builder.WithPredicates(predicate.Funcs{
 			UpdateFunc: func(e event.UpdateEvent) bool {
 				if e.ObjectOld.GetGeneration() == e.ObjectNew.GetGeneration() {
@@ -230,4 +251,5 @@ func (r *InfisicalSecretReconciler) SetupWithManager(mgr ctrl.Manager) error {
 			},
 		})).
 		Complete(r)
+
 }

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

@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/http"
 	"strings"
 	tpl "text/template"
 
@@ -15,11 +16,14 @@ import (
 	"github.com/Infisical/infisical/k8-operator/internal/model"
 	"github.com/Infisical/infisical/k8-operator/internal/template"
 	"github.com/Infisical/infisical/k8-operator/internal/util"
+	"github.com/Infisical/infisical/k8-operator/internal/util/sse"
 	"github.com/go-logr/logr"
+	"github.com/go-resty/resty/v2"
 
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 
 	infisicalSdk "github.com/infisical/go-sdk"
 	corev1 "k8s.io/api/core/v1"
@@ -409,9 +413,10 @@ func (r *InfisicalSecretReconciler) getResourceVariables(infisicalSecret v1alpha
 		})
 
 		resourceVariablesMap[string(infisicalSecret.UID)] = util.ResourceVariables{
-			InfisicalClient: client,
-			CancelCtx:       cancel,
-			AuthDetails:     util.AuthenticationDetails{},
+			InfisicalClient:  client,
+			CancelCtx:        cancel,
+			AuthDetails:      util.AuthenticationDetails{},
+			ServerSentEvents: sse.NewConnectionRegistry(ctx),
 		}
 
 		resourceVariables = resourceVariablesMap[string(infisicalSecret.UID)]
@@ -421,7 +426,6 @@ func (r *InfisicalSecretReconciler) getResourceVariables(infisicalSecret v1alpha
 	}
 
 	return resourceVariables
-
 }
 
 func (r *InfisicalSecretReconciler) updateResourceVariables(infisicalSecret v1alpha1.InfisicalSecret, resourceVariables util.ResourceVariables, resourceVariablesMap map[string]util.ResourceVariables) {
@@ -454,9 +458,10 @@ func (r *InfisicalSecretReconciler) ReconcileInfisicalSecret(ctx context.Context
 		}
 
 		r.updateResourceVariables(*infisicalSecret, util.ResourceVariables{
-			InfisicalClient: infisicalClient,
-			CancelCtx:       cancelCtx,
-			AuthDetails:     authDetails,
+			InfisicalClient:  infisicalClient,
+			CancelCtx:        cancelCtx,
+			AuthDetails:      authDetails,
+			ServerSentEvents: sse.NewConnectionRegistry(ctx),
 		}, resourceVariablesMap)
 	}
 
@@ -525,3 +530,94 @@ func (r *InfisicalSecretReconciler) ReconcileInfisicalSecret(ctx context.Context
 
 	return secretsCount, nil
 }
+
+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")
+	}
+
+	projectSlug := variables.AuthDetails.MachineIdentityScope.ProjectSlug
+	secretsPath := variables.AuthDetails.MachineIdentityScope.SecretsPath
+	envSlug := variables.AuthDetails.MachineIdentityScope.EnvSlug
+
+	infiscalClient := variables.InfisicalClient
+	sseRegistry := variables.ServerSentEvents
+
+	token := infiscalClient.Auth().GetAccessToken()
+
+	project, err := util.GetProjectBySlug(token, projectSlug)
+
+	if err != nil {
+		return fmt.Errorf("failed to get project [err=%s]", err)
+	}
+
+	if variables.AuthDetails.MachineIdentityScope.Recursive {
+		secretsPath = fmt.Sprint(secretsPath, "**")
+	}
+
+	if err != nil {
+		return fmt.Errorf("CallSubscribeProjectEvents: unable to marshal body [err=%s]", err)
+	}
+
+	events, errors, err := sseRegistry.Subscribe(func() (*http.Response, error) {
+		httpClient := resty.New()
+
+		req, err := api.CallSubscribeProjectEvents(httpClient, project.ID, secretsPath, envSlug, token)
+
+		if err != nil {
+			return nil, err
+		}
+
+		return req, nil
+	})
+
+	if err != nil {
+		return fmt.Errorf("unable to connect sse [err=%s]", err)
+	}
+
+	go func() {
+	outer:
+		for {
+			select {
+			case ev := <-events:
+				logger.Info("Received SSE Event", "event", ev)
+				eventCh <- event.TypedGenericEvent[client.Object]{
+					Object: infisicalSecret,
+				}
+			case err := <-errors:
+				logger.Error(err, "Error occurred")
+				break outer
+			case <-ctx.Done():
+				break outer
+			}
+		}
+	}()
+
+	return nil
+}

k8-operator/internal/util/handler.go

@@ -0,0 +1,59 @@
+package util
+
+import (
+	"context"
+	"math/rand"
+	"time"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+// computeMaxJitterDuration returns a random duration between 0 and max.
+// This is useful for introducing jitter to event processing.
+func computeMaxJitterDuration(max time.Duration) (time.Duration, time.Duration) {
+	if max <= 0 {
+		return 0, 0
+	}
+	jitter := time.Duration(rand.Int63n(int64(max)))
+	return max, jitter
+}
+
+// EnqueueDelayedEventHandler enqueues reconcile requests with a random delay (jitter)
+// to spread the load and avoid thundering herd issues.
+type EnqueueDelayedEventHandler struct {
+	Delay time.Duration
+}
+
+func (e *EnqueueDelayedEventHandler) Create(_ context.Context, _ event.TypedCreateEvent[client.Object], _ workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+}
+
+func (e *EnqueueDelayedEventHandler) Update(_ context.Context, _ event.TypedUpdateEvent[client.Object], _ workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+}
+
+func (e *EnqueueDelayedEventHandler) Delete(_ context.Context, _ event.TypedDeleteEvent[client.Object], _ workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+}
+
+func (e *EnqueueDelayedEventHandler) Generic(_ context.Context, evt event.TypedGenericEvent[client.Object], q workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+	if evt.Object == nil {
+		return
+	}
+
+	req := reconcile.Request{
+		NamespacedName: types.NamespacedName{
+			Namespace: evt.Object.GetNamespace(),
+			Name:      evt.Object.GetName(),
+		},
+	}
+
+	_, delay := computeMaxJitterDuration(e.Delay)
+
+	if delay > 0 {
+		q.AddAfter(req, delay)
+	} else {
+		q.Add(req)
+	}
+}

k8-operator/internal/util/models.go

@@ -3,11 +3,13 @@ package util
 import (
 	"context"
 
+	"github.com/Infisical/infisical/k8-operator/internal/util/sse"
 	infisicalSdk "github.com/infisical/go-sdk"
 )
 
 type ResourceVariables struct {
-	InfisicalClient infisicalSdk.InfisicalClientInterface
-	CancelCtx       context.CancelFunc
-	AuthDetails     AuthenticationDetails
+	InfisicalClient  infisicalSdk.InfisicalClientInterface
+	CancelCtx        context.CancelFunc
+	AuthDetails      AuthenticationDetails
+	ServerSentEvents *sse.ConnectionRegistry
 }

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

@@ -0,0 +1,331 @@
+package sse
+
+import (
+	"bufio"
+	"context"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// Event represents a Server-Sent Event
+type Event struct {
+	ID    string
+	Event string
+	Data  string
+	Retry int
+}
+
+// ConnectionMeta holds metadata about an SSE connection
+type ConnectionMeta struct {
+	EventChan  <-chan Event
+	ErrorChan  <-chan error
+	lastPingAt atomic.Value // stores time.Time
+	cancel     context.CancelFunc
+}
+
+// LastPing returns the last ping time
+func (c *ConnectionMeta) LastPing() time.Time {
+	if t, ok := c.lastPingAt.Load().(time.Time); ok {
+		return t
+	}
+	return time.Time{}
+}
+
+// UpdateLastPing atomically updates the last ping time
+func (c *ConnectionMeta) UpdateLastPing() {
+	c.lastPingAt.Store(time.Now())
+}
+
+// Cancel terminates the connection
+func (c *ConnectionMeta) Cancel() {
+	if c.cancel != nil {
+		c.cancel()
+	}
+}
+
+// ConnectionRegistry manages SSE connections with high performance
+type ConnectionRegistry struct {
+	mu   sync.RWMutex
+	conn *ConnectionMeta
+
+	monitorOnce sync.Once
+	monitorStop chan struct{}
+
+	onPing func() // Callback for ping events
+}
+
+// NewConnectionRegistry creates a new high-performance connection registry
+func NewConnectionRegistry(ctx context.Context) *ConnectionRegistry {
+	r := &ConnectionRegistry{
+		monitorStop: make(chan struct{}),
+	}
+
+	// Configure ping handler
+	r.onPing = func() {
+		r.UpdateLastPing()
+	}
+
+	return r
+}
+
+// Subscribe provides SSE events, creating a connection if needed
+func (r *ConnectionRegistry) Subscribe(request func() (*http.Response, error)) (<-chan Event, <-chan error, error) {
+	// Fast path: check if connection exists
+	if conn := r.getConnection(); conn != nil {
+		return conn.EventChan, conn.ErrorChan, nil
+	}
+
+	// Slow path: create new connection under lock
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	// Double-check after acquiring lock
+	if r.conn != nil {
+		return r.conn.EventChan, r.conn.ErrorChan, nil
+	}
+
+	res, err := request()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	conn, err := r.createStream(res)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	r.conn = conn
+
+	// Start monitor once
+	r.monitorOnce.Do(func() {
+		go r.monitorConnections()
+	})
+
+	return conn.EventChan, conn.ErrorChan, nil
+}
+
+// Get retrieves the current connection
+func (r *ConnectionRegistry) Get() (*ConnectionMeta, bool) {
+	conn := r.getConnection()
+	return conn, conn != nil
+}
+
+// IsConnected checks if there's an active connection
+func (r *ConnectionRegistry) IsConnected() bool {
+	return r.getConnection() != nil
+}
+
+// UpdateLastPing updates the last ping time for the current connection
+func (r *ConnectionRegistry) UpdateLastPing() {
+	if conn := r.getConnection(); conn != nil {
+		conn.UpdateLastPing()
+	}
+}
+
+// Close gracefully shuts down the registry
+func (r *ConnectionRegistry) Close() {
+	// Stop monitor first
+	select {
+	case <-r.monitorStop:
+		// Already closed
+	default:
+		close(r.monitorStop)
+	}
+
+	// Close connection
+	r.mu.Lock()
+	if r.conn != nil {
+		r.conn.Cancel()
+		r.conn = nil
+	}
+	r.mu.Unlock()
+}
+
+// getConnection returns the current connection without locking
+func (r *ConnectionRegistry) getConnection() *ConnectionMeta {
+	r.mu.RLock()
+	conn := r.conn
+	r.mu.RUnlock()
+	return conn
+}
+
+func (r *ConnectionRegistry) createStream(res *http.Response) (*ConnectionMeta, error) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	eventChan, errorChan, err := r.stream(ctx, res)
+	if err != nil {
+		cancel()
+		return nil, err
+	}
+
+	meta := &ConnectionMeta{
+		EventChan: eventChan,
+		ErrorChan: errorChan,
+		cancel:    cancel,
+	}
+	meta.UpdateLastPing()
+
+	return meta, nil
+}
+
+// stream processes SSE data from an HTTP response
+func (r *ConnectionRegistry) stream(ctx context.Context, res *http.Response) (<-chan Event, <-chan error, error) {
+	eventChan := make(chan Event, 10)
+	errorChan := make(chan error, 1)
+
+	go r.processStream(ctx, res.Body, eventChan, errorChan)
+
+	return eventChan, errorChan, nil
+}
+
+// processStream reads and parses SSE events from the response body
+func (r *ConnectionRegistry) processStream(ctx context.Context, body io.ReadCloser, eventChan chan<- Event, errorChan chan<- error) {
+	defer body.Close()
+	defer close(eventChan)
+	defer close(errorChan)
+
+	scanner := bufio.NewScanner(body)
+
+	var currentEvent Event
+	var dataBuilder strings.Builder
+
+	for scanner.Scan() {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
+		line := scanner.Text()
+
+		// Empty line indicates end of event
+		if len(line) == 0 {
+			if currentEvent.Data != "" || currentEvent.Event != "" {
+				// Finalize data
+				if dataBuilder.Len() > 0 {
+					currentEvent.Data = dataBuilder.String()
+					dataBuilder.Reset()
+				}
+
+				// Handle ping events
+				if r.isPingEvent(currentEvent) {
+					if r.onPing != nil {
+						r.onPing()
+					}
+				} else {
+					// Send non-ping events
+					select {
+					case eventChan <- currentEvent:
+					case <-ctx.Done():
+						return
+					}
+				}
+
+				// Reset for next event
+				currentEvent = Event{}
+			}
+			continue
+		}
+
+		// Parse line efficiently
+		r.parseLine(line, &currentEvent, &dataBuilder)
+	}
+
+	if err := scanner.Err(); err != nil {
+		select {
+		case errorChan <- err:
+		case <-ctx.Done():
+		}
+	}
+}
+
+// parseLine efficiently parses SSE protocol lines
+func (r *ConnectionRegistry) parseLine(line string, event *Event, dataBuilder *strings.Builder) {
+	colonIndex := strings.IndexByte(line, ':')
+	if colonIndex == -1 {
+		return // Invalid line format
+	}
+
+	field := line[:colonIndex]
+	value := line[colonIndex+1:]
+
+	// Trim leading space from value (SSE spec)
+	if len(value) > 0 && value[0] == ' ' {
+		value = value[1:]
+	}
+
+	switch field {
+	case "data":
+		if dataBuilder.Len() > 0 {
+			dataBuilder.WriteByte('\n')
+		}
+		dataBuilder.WriteString(value)
+	case "event":
+		event.Event = value
+	case "id":
+		event.ID = value
+	case "retry":
+		// Parse retry value if needed
+		// This could be used to configure reconnection delay
+	case "":
+		// Comment line, ignore
+	}
+}
+
+// isPingEvent checks if an event is a ping/keepalive
+func (r *ConnectionRegistry) isPingEvent(event Event) bool {
+	// Check for common ping patterns
+	if event.Event == "ping" {
+		return true
+	}
+
+	// Check for heartbeat data (common pattern is "1" or similar)
+	if event.Event == "" && strings.TrimSpace(event.Data) == "1" {
+		return true
+	}
+
+	return false
+}
+
+// monitorConnections checks connection health periodically
+func (r *ConnectionRegistry) monitorConnections() {
+	const (
+		checkInterval = 30 * time.Second
+		pingTimeout   = 2 * time.Minute
+	)
+
+	ticker := time.NewTicker(checkInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-r.monitorStop:
+			return
+		case <-ticker.C:
+			r.checkConnectionHealth(pingTimeout)
+		}
+	}
+}
+
+// checkConnectionHealth verifies connection is still alive
+func (r *ConnectionRegistry) checkConnectionHealth(timeout time.Duration) {
+	conn := r.getConnection()
+	if conn == nil {
+		return
+	}
+
+	if time.Since(conn.LastPing()) > timeout {
+		// Connection is stale, close it
+		r.mu.Lock()
+		if r.conn == conn { // Verify it's still the same connection
+			r.conn.Cancel()
+			r.monitorStop <- struct{}{}
+			r.conn = nil
+		}
+		r.mu.Unlock()
+	}
+}

k8-operator/internal/util/workspace.go

@@ -9,7 +9,6 @@ import (
 )
 
 func GetProjectByID(accessToken string, projectId string) (model.Project, error) {
-
 	httpClient := resty.New()
 	httpClient.
 		SetAuthScheme("Bearer").
@@ -25,3 +24,21 @@ func GetProjectByID(accessToken string, projectId string) (model.Project, error)
 
 	return projectDetails.Project, nil
 }
+
+func GetProjectBySlug(accessToken string, projectSlug string) (model.Project, error) {
+	httpClient := resty.New()
+	httpClient.
+		SetAuthScheme("Bearer").
+		SetAuthToken(accessToken).
+		SetHeader("Accept", "application/json")
+
+	project, err := api.CallGetProjectByIDv2(httpClient, api.GetProjectByIDRequest{
+		ProjectID: projectSlug,
+	})
+
+	if err != nil {
+		return model.Project{}, fmt.Errorf("unable to get project by slug. [err=%v]", err)
+	}
+
+	return project, nil
+}