fix: PR changes

k8-operator/internal/api/api.go

@@ -1,7 +1,9 @@
 package api
 
 import (
+	"encoding/json"
 	"fmt"
+	"net/http"
 
 	"github.com/Infisical/infisical/k8-operator/internal/model"
 	"github.com/go-resty/resty/v2"
@@ -167,3 +169,65 @@ func CallGetProjectByIDv2(httpClient *resty.Client, request GetProjectByIDReques
 	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/controller/infisicalsecret_controller.go

@@ -199,14 +199,16 @@ func (r *InfisicalSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	}
 
 	if infisicalSecretCRD.Spec.InstantUpdates {
-		logger.Info("Instant updates are enabled")
-
 		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))
+			requeueTime = time.Second * 10
+			logger.Info(err.Error())
+			logger.Info(fmt.Sprintf("unable to ensure event stream. Will requeue after [requeueTime=%v]", requeueTime))
 			return ctrl.Result{
 				RequeueAfter: requeueTime,
 			}, nil
 		}
+
+		logger.Info("Instant updates are enabled")
 	} else {
 		handler.CloseInstantUpdatesStream(ctx, logger, &infisicalSecretCRD, infisicalSecretResourceVariablesMap)
 	}
@@ -223,7 +225,7 @@ func (r *InfisicalSecretReconciler) SetupWithManager(mgr ctrl.Manager) error {
 
 	return ctrl.NewControllerManagedBy(mgr).
 		WatchesRawSource(
-			source.Channel[client.Object](r.SourceCh, &util.EnqueueDelayedEventHandler{Delay: time.Second * 3}),
+			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 {

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

@@ -3,7 +3,6 @@ package infisicalsecret
 import (
 	"bytes"
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"net/http"
@@ -19,6 +18,7 @@ import (
 	"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"
@@ -581,59 +581,24 @@ func (r *InfisicalSecretReconciler) OpenInstantUpdatesStream(ctx context.Context
 		secretsPath = fmt.Sprint(secretsPath, "**")
 	}
 
-	conditions := &api.SubscribeProjectEventsRequestCondition{
-		SecretPath:      secretsPath,
-		EnvironmentSlug: envSlug,
-	}
-
-	body, err := json.Marshal(api.SubscribeProjectEventsRequest{
-		ProjectID: project.ID,
-		Register: []api.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 fmt.Errorf("CallSubscribeProjectEvents: unable to marshal body [err=%s]", err)
 	}
 
-	events, errors, err := sseRegistry.Subscribe(func() (*http.Request, error) {
+	events, errors, err := sseRegistry.Subscribe(func() (*http.Response, error) {
-		headers := map[string]string{
+		httpClient := resty.New()
-			"User-Agent":    api.USER_AGENT_NAME,
-			"Authorization": fmt.Sprint("Bearer ", token),
-			"Content-Type":  "application/json",
-		}
 
-		req, err := http.NewRequest("POST", fmt.Sprintf("%s/v1/events/subscribe/project-events", api.API_HOST_URL), strings.NewReader(string(body)))
+		req, err := api.CallSubscribeProjectEvents(httpClient, project.ID, secretsPath, envSlug, token)
 
 		if err != nil {
 			return nil, err
 		}
 
-		for k, v := range headers {
-			req.Header.Set(k, v)
-		}
-
 		return req, nil
 	})
 
 	if err != nil {
-		return fmt.Errorf("unable to connect to SSE server [err=%s]", err)
+		return fmt.Errorf("unable to connect sse [err=%s]", err)
 	}
 
 	go func() {

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

@@ -1,187 +0,0 @@
-package sse
-
-import (
-	"bufio"
-	"context"
-	"fmt"
-	"io"
-	"net/http"
-	"strings"
-	"time"
-)
-
-// Event represents a Server-Sent Event
-type Event struct {
-	ID    string
-	Event string
-	Data  string
-	Retry int
-}
-
-// Client handles SSE connections with high performance
-type Client struct {
-	httpClient *http.Client
-	onPing     func() // Callback for ping events
-}
-
-// NewClient creates a new high-performance SSE client
-func NewClient() Client {
-	return Client{
-		httpClient: &http.Client{
-			Timeout: 0, // No timeout for streaming
-			Transport: &http.Transport{
-				MaxIdleConns:       100,
-				IdleConnTimeout:    90 * time.Second,
-				DisableCompression: true, // SSE typically doesn't benefit from compression
-			},
-		},
-	}
-}
-
-// WithPingHandler sets a callback for ping events
-func (c *Client) WithPingHandler(handler func()) *Client {
-	c.onPing = handler
-	return c
-}
-
-// Connect establishes an SSE connection and returns event channels
-func (c *Client) Connect(ctx context.Context, req *http.Request) (<-chan Event, <-chan error, error) {
-	// Configure SSE headers
-	req.Header.Set("Cache-Control", "no-cache")
-	req.Header.Set("Accept", "text/event-stream")
-	req.Header.Set("Connection", "keep-alive")
-
-	// Add context to request
-	req = req.WithContext(ctx)
-
-	resp, err := c.httpClient.Do(req)
-	if err != nil {
-		return nil, nil, fmt.Errorf("request failed: %w", err)
-	}
-
-	if resp.StatusCode != http.StatusOK {
-		resp.Body.Close()
-		return nil, nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
-	}
-
-	eventChan := make(chan Event, 10)
-	errorChan := make(chan error, 1)
-
-	go c.stream(ctx, resp.Body, eventChan, errorChan)
-
-	return eventChan, errorChan, nil
-}
-
-func (c *Client) stream(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 c.isPingEvent(currentEvent) {
-					if c.onPing != nil {
-						c.onPing()
-					}
-				} else {
-					// Send non-ping events
-					select {
-					case eventChan <- currentEvent:
-					case <-ctx.Done():
-						return
-					}
-				}
-
-				// Reset for next event
-				currentEvent = Event{}
-			}
-			continue
-		}
-
-		// Parse line efficiently
-		c.parseLine(line, &currentEvent, &dataBuilder)
-	}
-
-	if err := scanner.Err(); err != nil {
-		select {
-		case errorChan <- err:
-		case <-ctx.Done():
-		}
-	}
-}
-
-// parseLine efficiently parses SSE protocol lines
-func (c *Client) 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 (c *Client) 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
-}
-
-// WithHTTPClient sets a custom HTTP client
-func (c *Client) WithHTTPClient(client *http.Client) *Client {
-	c.httpClient = client
-	return c
-}

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

@@ -1,14 +1,24 @@
 package sse
 
 import (
+	"bufio"
 	"context"
-	"fmt"
+	"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
@@ -39,13 +49,13 @@ func (c *ConnectionMeta) Cancel() {
 
 // ConnectionRegistry manages SSE connections with high performance
 type ConnectionRegistry struct {
-	client Client
-
 	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
@@ -54,17 +64,16 @@ func NewConnectionRegistry(ctx context.Context) *ConnectionRegistry {
 		monitorStop: make(chan struct{}),
 	}
 
-	// Configure client with ping handler
+	// Configure ping handler
-	r.client = NewClient()
+	r.onPing = func() {
-	r.client.WithPingHandler(func() {
 		r.UpdateLastPing()
-	})
+	}
 
 	return r
 }
 
 // Subscribe provides SSE events, creating a connection if needed
-func (r *ConnectionRegistry) Subscribe(buildRequest func() (*http.Request, error)) (<-chan Event, <-chan error, error) {
+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
@@ -79,12 +88,12 @@ func (r *ConnectionRegistry) Subscribe(buildRequest func() (*http.Request, error
 		return r.conn.EventChan, r.conn.ErrorChan, nil
 	}
 
-	req, err := buildRequest()
+	res, err := request()
 	if err != nil {
-		return nil, nil, fmt.Errorf("failed to build request: %w", err)
+		return nil, nil, err
 	}
 
-	conn, err := r.createConnection(req)
+	conn, err := r.createStream(res)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -144,14 +153,13 @@ func (r *ConnectionRegistry) getConnection() *ConnectionMeta {
 	return conn
 }
 
-// createConnection creates a new SSE connection
+func (r *ConnectionRegistry) createStream(res *http.Response) (*ConnectionMeta, error) {
-func (r *ConnectionRegistry) createConnection(req *http.Request) (*ConnectionMeta, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 
-	eventChan, errorChan, err := r.client.Connect(ctx, req)
+	eventChan, errorChan, err := r.stream(ctx, res)
 	if err != nil {
 		cancel()
-		return nil, fmt.Errorf("failed to connect: %w", err)
+		return nil, err
 	}
 
 	meta := &ConnectionMeta{
@@ -164,6 +172,125 @@ func (r *ConnectionRegistry) createConnection(req *http.Request) (*ConnectionMet
 	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 (