Add Secret Management UI Implementation

gateway/gateway-controller/pkg/controlplane/client.go

@@ -110,6 +110,7 @@ type ControlPlaneClient interface {
 // *secrets.SecretService satisfies this interface.
 type secretSyncer interface {
 	UpsertFromPlatform(handle, displayName, plaintext string) error
+	Delete(handle, correlationID string) error
 }
 
 // WebhookSecretSnapshotRefresher is the extension point through which an
@@ -158,6 +159,7 @@ type Client struct {
 	webhookSecretSnapshotManager WebhookSecretSnapshotRefresher
 	secretSyncer                 secretSyncer
 	secretHashCache              sync.Map // handle → last-known Platform API hash (string)
+	secretRevisionCache          sync.Map // handle → last-applied event Revision (int64); never cleared on evict, so a stale event can't undo a later one
 	eventGatewayHooks            ControlPlaneEventGatewayHooks
 
 	// DP->CP push retry tuning.
@@ -1342,9 +1344,19 @@ func (c *Client) handleMessage(messageType int, message []byte) {
 		return
 	}
 
-	// Parse as generic event to extract type
+	// Parse as generic event to extract type. UseNumber() keeps JSON numbers as
+	// json.Number (exact decimal text) instead of the default float64 — float64
+	// only has ~53 bits of integer precision, which silently rounds a UnixNano
+	// revision (~60 bits) to the nearest ~256ns at this magnitude. Two events for
+	// the same handle within that window would decode to the identical revision,
+	// letting a stale, reordered event's revision compare as "not older" than the
+	// newer one already applied (see isStaleSecretEvent). utils.MapToStruct's own
+	// marshal/unmarshal re-encodes json.Number as the original digits verbatim, so
+	// this one change is sufficient — no downstream struct/comparison needs to change.
 	var event map[string]interface{}
-	if err := json.Unmarshal(message, &event); err != nil {
+	dec := json.NewDecoder(bytes.NewReader(message))
+	dec.UseNumber()
+	if err := dec.Decode(&event); err != nil {
 		c.logger.Error("Failed to parse WebSocket message",
 			slog.Any("error", err),
 			slog.String("message", string(message)),
@@ -1440,6 +1452,10 @@ func (c *Client) handleMessage(messageType int, message []byte) {
 		c.dispatchEventGatewayHook(event["type"], func(h ControlPlaneEventGatewayHooks) { h.HandleWebBrokerAPIDeleted(c, event) })
 	case "application.updated":
 		c.handleApplicationUpdatedEvent(event)
+	case "secret.updated":
+		c.handleSecretUpdatedEvent(event)
+	case "secret.deleted":
+		c.handleSecretDeletedEvent(event)
 	default:
 		c.logger.Info("Received unknown event type (will be processed when handlers are implemented)",
 			slog.String("type", eventType),
@@ -3745,6 +3761,126 @@ func (c *Client) handleSubscriptionPlanDeletedEvent(event map[string]interface{}
 	}
 }
 
+// handleSecretUpdatedEvent processes secret.updated events, pushed when a secret is
+// rotated. It re-fetches the plaintext over the authenticated internal secret-value
+// endpoint (the event payload never carries it) and upserts it into local storage, so
+// {{ secret "handle" }} placeholders resolve to the new value immediately instead of
+// waiting for the next reconnect's incremental sync.
+func (c *Client) handleSecretUpdatedEvent(event map[string]interface{}) {
+	baseLogger := c.logger
+	if c.apiUtilsService == nil || c.secretSyncer == nil {
+		baseLogger.Debug("Skipping secret.updated event: secret sync not configured")
+		return
+	}
+
+	var updated SecretUpdatedEvent
+	if err := utils.MapToStruct(event, &updated); err != nil {
+		baseLogger.Error("Failed to parse secret.updated event", slog.Any("error", err))
+		return
+	}
+	payload := updated.Payload
+	if payload.Handle == "" {
+		baseLogger.Error("secret.updated event missing handle")
+		return
+	}
+	logger := baseLogger.With(
+		slog.String("correlation_id", updated.CorrelationID),
+		slog.String("secret_handle", payload.Handle),
+	)
+
+	c.applySecretUpdatedPayload(payload, logger, c.apiUtilsService.FetchPlatformSecretValue)
+}
+
+// applySecretUpdatedPayload is the testable core of handleSecretUpdatedEvent: it
+// fetches the rotated plaintext via fetchValue and upserts it into local storage.
+// Extracted so unit tests can stub fetchValue instead of the concrete
+// *utils.APIUtilsService (mirroring syncSecretsIncrementalFromMetas in
+// sync_secrets_test.go, which works around the same constraint).
+func (c *Client) applySecretUpdatedPayload(payload SecretUpdatedEventPayload, logger *slog.Logger, fetchValue func(handle string) (string, error)) {
+	if c.isStaleSecretEvent(payload.Handle, payload.Revision, logger) {
+		return
+	}
+
+	plaintext, err := fetchValue(payload.Handle)
+	if err != nil {
+		logger.Error("Failed to fetch rotated secret value", slog.Any("error", err))
+		return
+	}
+
+	if err := c.secretSyncer.UpsertFromPlatform(payload.Handle, payload.DisplayName, plaintext); err != nil {
+		logger.Error("Failed to upsert rotated secret", slog.Any("error", err))
+		return
+	}
+
+	c.secretHashCache.Store(payload.Handle, payload.Hash)
+	c.secretRevisionCache.Store(payload.Handle, payload.Revision)
+	logger.Info("Applied secret rotation from secret.updated event")
+}
+
+// isStaleSecretEvent reports whether revision is older than the last revision this
+// gateway already applied for handle. The comparison is strict-less-than so a
+// redelivery of the same event (revision equal to the cached one — e.g. at-least-once
+// retry from the EventHub) still applies normally, preserving existing idempotency.
+// The cache entry is never cleared on eviction (see handleSecretDeletedEvent), so a
+// deletion followed by the same handle being reused by a later create, followed by a
+// stale, redelivered copy of the original deletion, cannot evict the newly created
+// secret: the stale deletion's revision is lower than the new secret's.
+func (c *Client) isStaleSecretEvent(handle string, revision int64, logger *slog.Logger) bool {
+	if cached, ok := c.secretRevisionCache.Load(handle); ok {
+		if lastApplied, ok := cached.(int64); ok && revision < lastApplied {
+			logger.Warn("Ignoring stale secret event",
+				slog.Int64("event_revision", revision),
+				slog.Int64("last_applied_revision", lastApplied),
+			)
+			return true
+		}
+	}
+	return false
+}
+
+// handleSecretDeletedEvent processes secret.deleted events, pushed when a secret is
+// permanently deleted. Deletion only succeeds once no artifact — current config or
+// any deployed snapshot, on any gateway — still references the handle, so evicting
+// the local copy here is always safe.
+func (c *Client) handleSecretDeletedEvent(event map[string]interface{}) {
+	baseLogger := c.logger
+	if c.secretSyncer == nil {
+		baseLogger.Debug("Skipping secret.deleted event: secret sync not configured")
+		return
+	}
+
+	var deleted SecretDeletedEvent
+	if err := utils.MapToStruct(event, &deleted); err != nil {
+		baseLogger.Error("Failed to parse secret.deleted event", slog.Any("error", err))
+		return
+	}
+	payload := deleted.Payload
+	if payload.Handle == "" {
+		baseLogger.Error("secret.deleted event missing handle")
+		return
+	}
+	logger := baseLogger.With(
+		slog.String("correlation_id", deleted.CorrelationID),
+		slog.String("secret_handle", payload.Handle),
+	)
+
+	if c.isStaleSecretEvent(payload.Handle, payload.Revision, logger) {
+		return
+	}
+
+	if err := c.secretSyncer.Delete(payload.Handle, deleted.CorrelationID); err != nil {
+		logger.Warn("Failed to evict deleted secret from local store", slog.Any("error", err))
+		return
+	}
+	c.secretHashCache.Delete(payload.Handle)
+	// Deliberately Store, not Delete: a later secret.updated for the same handle (the
+	// handle reused by a newly created secret) must still be able to detect a
+	// subsequently-redelivered copy of *this* deletion event as stale. Clearing the
+	// cache entry here would let that stale redelivery through and evict the new secret.
+	c.secretRevisionCache.Store(payload.Handle, payload.Revision)
+	logger.Info("Evicted deleted secret from local store")
+}
+
 // setState updates the connection state
 func (c *Client) setState(newState State) {
 	c.state.mu.Lock()

gateway/gateway-controller/pkg/controlplane/events.go

@@ -399,6 +399,46 @@ type SubscriptionPlanDeletedEvent struct {
 	CorrelationID string                              `json:"correlationId"`
 }
 
+// SecretUpdatedEventPayload represents the payload of a secret.updated event, fired
+// when a secret is rotated. It never carries the plaintext value — Hash is the
+// HMAC-SHA256 change-detection digest, safe to transmit since it cannot be reversed
+// into the plaintext. The receiving handler fetches the fresh plaintext separately
+// over the authenticated internal secret-value endpoint.
+type SecretUpdatedEventPayload struct {
+	Handle      string `json:"handle"`
+	DisplayName string `json:"name"`
+	Hash        string `json:"hash"`
+	// Revision orders events for the same handle so a redelivered or reordered
+	// event cannot undo a change already applied locally. See Client.secretRevisionCache.
+	Revision int64 `json:"revision"`
+}
+
+// SecretUpdatedEvent represents the complete secret.updated event.
+type SecretUpdatedEvent struct {
+	Type          string                    `json:"type"`
+	Payload       SecretUpdatedEventPayload `json:"payload"`
+	Timestamp     string                    `json:"timestamp"`
+	CorrelationID string                    `json:"correlationId"`
+}
+
+// SecretDeletedEventPayload represents the payload of a secret.deleted event,
+// fired when a secret is permanently deleted.
+type SecretDeletedEventPayload struct {
+	Handle string `json:"handle"`
+	// Revision — see SecretUpdatedEventPayload.Revision. Compared against the same
+	// cache so a late deletion cannot evict a secret that was recreated under the
+	// same handle after it.
+	Revision int64 `json:"revision"`
+}
+
+// SecretDeletedEvent represents the complete secret.deleted event.
+type SecretDeletedEvent struct {
+	Type          string                    `json:"type"`
+	Payload       SecretDeletedEventPayload `json:"payload"`
+	Timestamp     string                    `json:"timestamp"`
+	CorrelationID string                    `json:"correlationId"`
+}
+
 // ApplicationKeyMappingPayload represents a single application to API key mapping entry.
 type ApplicationKeyMappingPayload struct {
 	ApiKeyUuid string `json:"apiKeyUuid"`

gateway/gateway-controller/pkg/controlplane/sync_secrets.go

@@ -22,6 +22,7 @@ import (
 	"log/slog"
 
 	"github.com/wso2/api-platform/gateway/gateway-controller/pkg/constants"
+	"github.com/wso2/api-platform/gateway/gateway-controller/pkg/utils"
 )
 
 // syncSecrets pulls secrets from the Platform API and upserts them into local
@@ -66,19 +67,27 @@ func (c *Client) syncSecrets() {
 func (c *Client) syncSecretsBulk() {
 	c.logger.Info("Starting bulk Platform API secret sync (startup)")
 
+	// Snapshot before the (potentially slow) fetch so a handle concurrently added
+	// to secretHashCache while the fetch is in flight — e.g. syncSecretRefsFromYAML
+	// running on a deployment event handled on another goroutine — is never
+	// considered for eviction below: activeHandles couldn't possibly reflect it.
+	preFetchHandles := c.snapshotSecretHashCacheKeys()
+
 	metas, err := c.apiUtilsService.FetchPlatformSecrets(nil, true)
 	if err != nil {
 		c.logger.Error("Failed to bulk fetch platform secrets", slog.Any("error", err))
 		return
 	}
 
 	synced, skipped, failed := 0, 0, 0
+	activeHandles := make(map[string]struct{}, len(metas))
 
 	for _, meta := range metas {
 		if meta.Status != "ACTIVE" {
 			skipped++
 			continue
 		}
+		activeHandles[meta.Handle] = struct{}{}
 
 		if meta.Value == nil {
 			c.logger.Warn("Bulk fetch returned no value for secret — skipping",
@@ -101,10 +110,13 @@ func (c *Client) syncSecretsBulk() {
 		synced++
 	}
 
+	evicted := c.evictSecretsNotIn(activeHandles, preFetchHandles)
+
 	c.logger.Info("Bulk Platform API secret sync complete",
 		slog.Int("synced", synced),
 		slog.Int("skipped", skipped),
 		slog.Int("failed", failed),
+		slog.Int("evicted", evicted),
 	)
 }
 
@@ -113,19 +125,25 @@ func (c *Client) syncSecretsBulk() {
 func (c *Client) syncSecretsIncremental() {
 	c.logger.Info("Starting incremental Platform API secret sync (reconnect)")
 
+	// See the matching comment in syncSecretsBulk: this snapshot must be taken
+	// before the fetch begins, not after.
+	preFetchHandles := c.snapshotSecretHashCacheKeys()
+
 	metas, err := c.apiUtilsService.FetchPlatformSecrets(nil, false)
 	if err != nil {
 		c.logger.Error("Failed to fetch platform secrets metadata", slog.Any("error", err))
 		return
 	}
 
 	synced, skipped, failed := 0, 0, 0
+	activeHandles := make(map[string]struct{}, len(metas))
 
 	for _, meta := range metas {
 		if meta.Status != "ACTIVE" {
 			skipped++
 			continue
 		}
+		activeHandles[meta.Handle] = struct{}{}
 
 		// Skip if hash unchanged since last sync.
 		if cached, ok := c.secretHashCache.Load(meta.Handle); ok && cached.(string) == meta.Hash {
@@ -156,13 +174,96 @@ func (c *Client) syncSecretsIncremental() {
 		synced++
 	}
 
+	evicted := c.evictSecretsNotIn(activeHandles, preFetchHandles)
+
 	c.logger.Info("Incremental Platform API secret sync complete",
 		slog.Int("synced", synced),
 		slog.Int("skipped", skipped),
 		slog.Int("failed", failed),
+		slog.Int("evicted", evicted),
 	)
 }
 
+// snapshotSecretHashCacheKeys captures the set of handles present in
+// secretHashCache at a point in time, so a subsequent evictSecretsNotIn call
+// can tell which handles existed before an in-flight platform fetch started.
+func (c *Client) snapshotSecretHashCacheKeys() map[string]struct{} {
+	snapshot := make(map[string]struct{})
+	c.secretHashCache.Range(func(key, _ any) bool {
+		if handle, ok := key.(string); ok {
+			snapshot[handle] = struct{}{}
+		}
+		return true
+	})
+	return snapshot
+}
+
+// evictSecretsNotIn is the poll-based recovery path for the same eviction that
+// handleSecretDeletedEvent applies live: a gateway that is disconnected at the
+// moment a secret is deleted only receives that secret.deleted WebSocket event if
+// it's connected when the event is broadcast. This diffs the
+// latest Platform API response against secretHashCache so that, on the next
+// reconnect/poll, any cached handle that is no longer ACTIVE (permanently deleted,
+// or flipped to a non-ACTIVE status) gets evicted from local storage even though
+// the live event was missed.
+//
+// activeHandles is the set of handles the just-completed poll returned with
+// status ACTIVE; every other handle currently in secretHashCache is a
+// candidate for eviction. preFetchHandles is the secretHashCache key snapshot
+// taken immediately before the platform fetch began (see syncSecretsBulk /
+// syncSecretsIncremental): a handle absent from it was added concurrently
+// while the fetch was in flight (e.g. by syncSecretRefsFromYAML handling a
+// deployment event on another goroutine) and is never evicted here, since
+// activeHandles — reflecting a fetch that started before this handle
+// existed — can say nothing about whether it's actually still active.
+func (c *Client) evictSecretsNotIn(activeHandles, preFetchHandles map[string]struct{}) int {
+	var stale []string
+	c.secretHashCache.Range(func(key, _ any) bool {
+		handle, ok := key.(string)
+		if !ok {
+			return true
+		}
+		if _, existedBeforeFetch := preFetchHandles[handle]; !existedBeforeFetch {
+			return true // added during the in-flight fetch — not eligible for eviction
+		}
+		if _, ok := activeHandles[handle]; !ok {
+			stale = append(stale, handle)
+		}
+		return true
+	})
+
+	for _, handle := range stale {
+		// A fresh random ID per attempt, not a deterministic one: this eviction is
+		// decided locally (no incoming event carries its own correlation ID to
+		// reuse, unlike handleSecretDeletedEvent's live path), and repeated retries
+		// across poll cycles for the same handle must be distinguishable in logs
+		// rather than colliding whenever they land in the same time bucket.
+		correlationID, err := utils.GenerateUUID()
+		if err != nil {
+			c.logger.Warn("Failed to generate correlation ID for secret eviction",
+				slog.String("secret_handle", handle),
+				slog.Any("error", err),
+			)
+			correlationID = "unknown"
+		}
+		if err := c.secretSyncer.Delete(handle, correlationID); err != nil {
+			c.logger.Warn("Failed to evict stale secret from local store",
+				slog.String("secret_handle", handle),
+				slog.String("correlation_id", correlationID),
+				slog.Any("error", err),
+			)
+			continue
+		}
+		c.secretHashCache.Delete(handle)
+		c.logger.Info("Evicted stale secret from local store during poll sync",
+			slog.String("secret_handle", handle),
+			slog.String("correlation_id", correlationID),
+		)
+	}
+
+	return len(stale)
+}
+
 // syncSecretRefsFromYAML extracts {{ secret "handle" }} placeholders from the
 // supplied YAML, then fetches and upserts any handle that is not already in the
 // local hash cache. This is called from deployment event handlers so that secrets