platform api revamp

Author: tharindu1st

event-gateway/gateway-runtime/internal/connectors/receiver/websub/delivery.go

@@ -144,20 +144,19 @@ func (d *Deliverer) doDeliver(ctx context.Context, callbackURL, secret string, m
 // hopByHopHeaders contains the RFC 2616 hop-by-hop headers that must never be
 // forwarded to a downstream subscriber. Keys are lower-cased for comparison.
 var hopByHopHeaders = map[string]bool{
-	"connection":          true,
-	"keep-alive":          true,
-	"proxy-authenticate":  true,
-	"proxy-authorization": true,
-	"te":                  true,
-	"trailers":            true,
-	"transfer-encoding":   true,
-	"upgrade":             true,
-	"host":                true,
+	"connection":        true,
+	"keep-alive":        true,
+	"te":                true,
+	"trailer":           true,
+	"transfer-encoding": true,
+	"upgrade":           true,
+	"host":              true,
 }
 
 // internalHeaderPrefixes lists lower-cased prefixes used by the gateway for
 // internal metadata that must not be leaked to external subscribers.
 var internalHeaderPrefixes = []string{
+	"proxy-",
 	"x-internal-",
 }
 

event-gateway/gateway-runtime/internal/connectors/receiver/websub/handler.go

@@ -106,7 +106,7 @@ func (h *HubHandler) handleSubscribe(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	if shortCircuited {
-		http.Error(w, "forbidden by policy", http.StatusForbidden)
+		writePolicyResponse(w, nil, http.StatusForbidden, "forbidden by policy")
 		return
 	}
 
@@ -210,7 +210,7 @@ func (h *HubHandler) handleUnsubscribe(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	if shortCircuited {
-		http.Error(w, "forbidden by policy", http.StatusForbidden)
+		writePolicyResponse(w, nil, http.StatusForbidden, "forbidden by policy")
 		return
 	}
 
@@ -342,7 +342,7 @@ func (h *WebhookReceiverHandler) ServeHTTP(w http.ResponseWriter, r *http.Reques
 	}
 	if shortCircuited {
 		slog.Info("Inbound request rejected by policy", "channel", channelName, "binding", h.bindingName)
-		http.Error(w, "forbidden by policy", http.StatusForbidden)
+		writePolicyResponse(w, processed, http.StatusForbidden, "forbidden by policy")
 		return
 	}
 
@@ -355,6 +355,33 @@ func (h *WebhookReceiverHandler) ServeHTTP(w http.ResponseWriter, r *http.Reques
 	w.WriteHeader(http.StatusAccepted)
 }
 
+// writePolicyResponse writes the HTTP response from a short-circuited policy execution.
+// If msg is non-nil and carries a status code in Metadata["status_code"], that status is used
+// along with any headers and body from the message. Otherwise the fallback status and body are used.
+func writePolicyResponse(w http.ResponseWriter, msg *connectors.Message, fallbackStatus int, fallbackBody string) {
+	if msg != nil {
+		statusCode := fallbackStatus
+		if sc, ok := msg.Metadata["status_code"]; ok {
+			if code, ok := sc.(int); ok && code > 0 {
+				statusCode = code
+			}
+		}
+		for k, vals := range msg.Headers {
+			for _, v := range vals {
+				w.Header().Add(k, v)
+			}
+		}
+		if len(msg.Value) > 0 {
+			w.WriteHeader(statusCode)
+			_, _ = w.Write(msg.Value)
+			return
+		}
+		w.WriteHeader(statusCode)
+		return
+	}
+	http.Error(w, fallbackBody, fallbackStatus)
+}
+
 func (h *WebhookReceiverHandler) publishToBrokerDriver(ctx context.Context, kafkaTopic string, msg *connectors.Message) error {
 	if err := h.brokerDriver.Publish(ctx, kafkaTopic, msg); err != nil {
 		return fmt.Errorf("failed to publish to broker-driver: %w", err)

event-gateway/gateway-runtime/internal/hub/hub.go

@@ -38,19 +38,20 @@ type ChannelChainKeySet struct {
 
 // ChannelBinding holds the runtime state for a registered channel.
 type ChannelBinding struct {
-	APIID             string
-	Name              string
-	Mode              string // "websub" or "protocol-mediation"
-	Context           string
-	Version           string
-	Vhost             string
-	SubscribeChainKey string
-	InboundChainKey   string
-	OutboundChainKey  string
-	BrokerDriverTopic string
-	Ordering          string                        // "ordered" or "unordered"
-	Channels          map[string]string             // channel-name → Kafka topic (WebSubApi only)
-	ChannelChainKeys  map[string]ChannelChainKeySet // channel-name → per-channel chain keys
+	APIID               string
+	Name                string
+	Mode                string // "websub" or "protocol-mediation"
+	Context             string
+	Version             string
+	Vhost               string
+	SubscribeChainKey   string
+	InboundChainKey     string
+	OutboundChainKey    string
+	BrokerDriverTopic   string
+	Ordering            string                        // "ordered" or "unordered"
+	Channels            map[string]string             // channel-name → Kafka topic (WebSubApi only)
+	KafkaTopicToChannel map[string]string             // Kafka topic → channel-name (reverse of Channels, cached)
+	ChannelChainKeys    map[string]ChannelChainKeySet // channel-name → per-channel chain keys
 }
 
 // Hub is the central message router. It holds the policy engine reference and
@@ -70,9 +71,16 @@ func NewHub(eng *engine.Engine) *Hub {
 }
 
 // RegisterBinding adds a channel binding to the hub.
+// It automatically builds the KafkaTopicToChannel reverse map from b.Channels.
 func (h *Hub) RegisterBinding(b ChannelBinding) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
+	if len(b.Channels) > 0 {
+		b.KafkaTopicToChannel = make(map[string]string, len(b.Channels))
+		for channelName, topic := range b.Channels {
+			b.KafkaTopicToChannel[topic] = channelName
+		}
+	}
 	h.bindings[b.Name] = &b
 }
 
@@ -138,7 +146,7 @@ func (h *Hub) ProcessSubscribe(ctx context.Context, bindingName string, msg *con
 			}
 			if result.ShortCircuited {
 				logShortCircuit("Subscribe request short-circuited by hub policy", bindingName, binding.SubscribeChainKey, result.ImmediateResponse)
-				return nil, true, nil
+				return immediateResponseToMessage(result.ImmediateResponse), true, nil
 			}
 			if err := ApplyRequestHeaderResult(result, msg); err != nil {
 				return nil, false, fmt.Errorf("failed to apply subscribe header result: %w", err)
@@ -158,7 +166,7 @@ func (h *Hub) ProcessSubscribe(ctx context.Context, bindingName string, msg *con
 				}
 				if result.ShortCircuited {
 					logShortCircuit("Subscribe request short-circuited by channel policy", bindingName, keys.SubscribeChainKey, result.ImmediateResponse)
-					return nil, true, nil
+					return immediateResponseToMessage(result.ImmediateResponse), true, nil
 				}
 				if err := ApplyRequestHeaderResult(result, msg); err != nil {
 					return nil, false, fmt.Errorf("failed to apply subscribe channel header result: %w", err)
@@ -190,7 +198,7 @@ func (h *Hub) ProcessInbound(ctx context.Context, bindingName string, msg *conne
 			}
 			if result.ShortCircuited {
 				logShortCircuit("Inbound message short-circuited by hub policy", bindingName, binding.InboundChainKey, result.ImmediateResponse)
-				return nil, true, nil
+				return immediateResponseToMessage(result.ImmediateResponse), true, nil
 			}
 			if err := ApplyRequestHeaderResult(result, msg); err != nil {
 				return nil, false, fmt.Errorf("failed to apply inbound header result: %w", err)
@@ -202,7 +210,7 @@ func (h *Hub) ProcessInbound(ctx context.Context, bindingName string, msg *conne
 					return nil, false, fmt.Errorf("inbound body policy execution failed: %w", err)
 				}
 				if bodyResult.ShortCircuited {
-					return nil, true, nil
+					return immediateResponseToMessage(bodyResult.ImmediateResponse), true, nil
 				}
 				if err := ApplyRequestBodyResult(bodyResult, msg); err != nil {
 					return nil, false, fmt.Errorf("failed to apply inbound body result: %w", err)
@@ -223,7 +231,7 @@ func (h *Hub) ProcessInbound(ctx context.Context, bindingName string, msg *conne
 				}
 				if result.ShortCircuited {
 					logShortCircuit("Inbound message short-circuited by channel policy", bindingName, keys.InboundChainKey, result.ImmediateResponse)
-					return nil, true, nil
+					return immediateResponseToMessage(result.ImmediateResponse), true, nil
 				}
 				if err := ApplyRequestHeaderResult(result, msg); err != nil {
 					return nil, false, fmt.Errorf("failed to apply inbound channel header result: %w", err)
@@ -235,7 +243,7 @@ func (h *Hub) ProcessInbound(ctx context.Context, bindingName string, msg *conne
 						return nil, false, fmt.Errorf("inbound channel body policy execution failed: %w", err)
 					}
 					if bodyResult.ShortCircuited {
-						return nil, true, nil
+						return immediateResponseToMessage(bodyResult.ImmediateResponse), true, nil
 					}
 					if err := ApplyRequestBodyResult(bodyResult, msg); err != nil {
 						return nil, false, fmt.Errorf("failed to apply inbound channel body result: %w", err)
@@ -292,7 +300,7 @@ func (h *Hub) ProcessOutbound(ctx context.Context, bindingName string, msg *conn
 	// Apply channel-level outbound chain if present.
 	// msg.Topic here is the Kafka topic; resolve back to channel name.
 	if msg.Topic != "" && len(binding.ChannelChainKeys) > 0 {
-		channelName := resolveChannelName(binding.Channels, msg.Topic)
+		channelName := resolveChannelName(binding.KafkaTopicToChannel, binding.Channels, msg.Topic)
 		if channelName != "" {
 			if keys, ok := binding.ChannelChainKeys[channelName]; ok && keys.OutboundChainKey != "" {
 				chain := h.engine.GetChain(keys.OutboundChainKey)
@@ -330,9 +338,13 @@ func (h *Hub) ProcessOutbound(ctx context.Context, bindingName string, msg *conn
 	return msg, false, nil
 }
 
-// resolveChannelName reverse-maps a Kafka topic name to the channel name
-// using the binding's channel-name → Kafka-topic map.
-func resolveChannelName(channels map[string]string, kafkaTopic string) string {
+// resolveChannelName reverse-maps a Kafka topic name to the channel name.
+// It first checks the pre-built kafkaTopicToChannel reverse map for an O(1) lookup,
+// and falls back to an O(n) scan over channels only if the cached map is absent.
+func resolveChannelName(kafkaTopicToChannel map[string]string, channels map[string]string, kafkaTopic string) string {
+	if len(kafkaTopicToChannel) > 0 {
+		return kafkaTopicToChannel[kafkaTopic]
+	}
 	for channelName, topic := range channels {
 		if topic == kafkaTopic {
 			return channelName
@@ -341,6 +353,28 @@ func resolveChannelName(channels map[string]string, kafkaTopic string) string {
 	return ""
 }
 
+// immediateResponseToMessage encodes an ImmediateResponseResult from the policy engine
+// into a connectors.Message so callers can write the policy-provided HTTP response.
+// The status code is stored in Metadata["status_code"] (int), body in Value,
+// and response headers (single-value) in Headers ([]string slice per key).
+// Returns nil when resp is nil.
+func immediateResponseToMessage(resp *engine.ImmediateResponseResult) *connectors.Message {
+	if resp == nil {
+		return nil
+	}
+	headers := make(map[string][]string, len(resp.Headers))
+	for k, v := range resp.Headers {
+		headers[k] = []string{v}
+	}
+	return &connectors.Message{
+		Value:   resp.Body,
+		Headers: headers,
+		Metadata: map[string]interface{}{
+			"status_code": resp.StatusCode,
+		},
+	}
+}
+
 // logShortCircuit keeps Info logs to metadata only; ImmediateResponse.Body is
 // user-visible content and must not contain sensitive information.
 func logShortCircuit(message, bindingName, chainKey string, resp *engine.ImmediateResponseResult) {

platform-api/src/internal/model/deployment.go

@@ -61,12 +61,12 @@ type DeploymentContent struct {
 type DeploymentStatus string
 
 const (
-	DeploymentStatusDeployed   DeploymentStatus = "DEPLOYED"
-	DeploymentStatusUndeployed DeploymentStatus = "UNDEPLOYED"
+	DeploymentStatusDeployed    DeploymentStatus = "DEPLOYED"
+	DeploymentStatusUndeployed  DeploymentStatus = "UNDEPLOYED"
 	DeploymentStatusDeploying   DeploymentStatus = "DEPLOYING"
 	DeploymentStatusUndeploying DeploymentStatus = "UNDEPLOYING"
 	DeploymentStatusFailed      DeploymentStatus = "FAILED"
-	DeploymentStatusArchived   DeploymentStatus = "ARCHIVED" // Derived state: exists in history but not in status table
+	DeploymentStatusArchived    DeploymentStatus = "ARCHIVED" // Derived state: exists in history but not in status table
 )
 
 // DeploymentInfo is a lightweight representation of a deployment
@@ -121,12 +121,13 @@ type WebSubAPIDeploymentYAML struct {
 
 // WebSubAPIDeploymentSpec represents the spec section of the WebSub API deployment YAML
 type WebSubAPIDeploymentSpec struct {
-	DisplayName string                       `yaml:"displayName"`
-	Version     string                       `yaml:"version"`
-	Context     string                       `yaml:"context"`
-	Vhosts      *WebSubAPIDeploymentVhosts   `yaml:"vhosts,omitempty"`
-	Channels    []WebSubAPIDeploymentChannel `yaml:"channels,omitempty"`
-	Policies    []Policy                     `yaml:"policies,omitempty"`
+	DisplayName string                     `yaml:"displayName"`
+	Version     string                     `yaml:"version"`
+	Context     string                     `yaml:"context"`
+	Vhosts      *WebSubAPIDeploymentVhosts `yaml:"vhosts,omitempty"`
+	Hub         WebSubHub                  `json:"hub" yaml:"hub"`
+	Receiver    *WebSubReceiver            `json:"receiver,omitempty" yaml:"receiver,omitempty"`
+	Delivery    *WebSubDelivery            `json:"delivery,omitempty" yaml:"delivery,omitempty"`
 }
 
 // WebSubAPIDeploymentVhosts represents vhost configuration in the WebSub API deployment YAML
@@ -140,3 +141,34 @@ type WebSubAPIDeploymentChannel struct {
 	Name   string `yaml:"name"`
 	Method string `yaml:"method"`
 }
+
+// WebSubDelivery Delivery configuration for the WebSub API - handles outbound event delivery to subscribers.
+type WebSubDelivery struct {
+	// Policies List of policies applied when delivering events to subscriber callback URLs (e.g., hmac-signature-validation)
+	Policies *[]Policy `json:"policies,omitempty" yaml:"policies,omitempty"`
+}
+
+// WebSubHub Hub configuration for the WebSub API - handles subscriber management and event fan-out.
+type WebSubHub struct {
+	// Channels List of topic channels available for subscription
+	Channels []HubChannel `json:"channels" yaml:"channels"`
+
+	// Policies List of policies applied at the hub level (e.g., api-key-auth for subscribers)
+	Policies *[]Policy `json:"policies,omitempty" yaml:"policies,omitempty"`
+}
+
+// HubChannel A subscribable topic channel within the WebSub hub.
+type HubChannel struct {
+	// Name Channel name or topic identifier relative to API context.
+	Name string `json:"name" yaml:"name"`
+	// Method The method by which the channel is identified (e.g., "SUB" for subscription-based, "PUB" for publish-only)
+	Method string `json:"method" yaml:"method"`
+	// Policies List of policies applied only to this channel (e.g., rbac)
+	Policies *[]Policy `json:"policies,omitempty" yaml:"policies,omitempty"`
+}
+
+// WebSubReceiver Receiver configuration for the WebSub API - handles inbound event publishing from publishers.
+type WebSubReceiver struct {
+	// Policies List of policies applied to inbound webhook requests (e.g., hmac-signature-validation)
+	Policies *[]Policy `json:"policies,omitempty" yaml:"policies,omitempty"`
+}