Merge pull request #2003 from senthuran16/websub-topic-hash

Hash WebSub API topics

Author: senthuran16

event-gateway/ARCHITECTURE.md

@@ -1220,7 +1220,7 @@ type BrokerDriver interface {
     Publish(ctx context.Context, topic string, msg *Message) error
     Subscribe(groupID string, topics []string, handler MessageHandler) (Receiver, error)
     TopicExists(ctx context.Context, topic string) (bool, error)
-    EnsureTopics(ctx context.Context, topics []string) error
+    EnsureTopics(ctx context.Context, topics []string, metadata map[string]map[string]string) error
     DeleteTopics(ctx context.Context, topics []string) error
     Close() error
 }

event-gateway/gateway-runtime/internal/binding/types.go

@@ -19,6 +19,7 @@
 package binding
 
 import (
+	"crypto/sha256"
 	"fmt"
 	"path"
 	"strings"
@@ -140,18 +141,22 @@ type ChannelsConfig struct {
 	Channels []Binding `yaml:"channels"`
 }
 
-// JoinNormalizedTopic derives a Kafka topic name by normalizing each logical
-// segment and joining them with underscores.
+// JoinNormalizedTopic derives a Kafka topic name by hashing the parts joined with underscores.
+// Each part is written as `<length>:<value>|` to ensure uniqueness and prevent collisions.
+// eg: "my-api", "v1", "/orders" -> "6:my-api|2:v1|7:/orders|" -> SHA-256 hash of that string.
+// Returns the SHA-256 hash of the joined string.
 func JoinNormalizedTopic(parts ...string) string {
 	if len(parts) == 0 {
 		return ""
 	}
-
-	normalizedParts := make([]string, 0, len(parts))
+	var joined strings.Builder
 	for _, part := range parts {
-		normalizedParts = append(normalizedParts, NormalizeTopicSegment(part))
+		fmt.Fprintf(&joined, "%d:%s|", len(part), part)
 	}
-	return strings.Join(normalizedParts, "_")
+	// Calculate SHA-256 hash
+	hash := sha256.Sum256([]byte(joined.String()))
+	// Return hex-encoded hash
+	return fmt.Sprintf("%x", hash)
 }
 
 // WebSubApiTopicName derives a Kafka topic name for a WebSubApi channel.

event-gateway/gateway-runtime/internal/connectors/brokerdriver/kafka/endpoint.go

@@ -100,22 +100,36 @@ func (e *KafkaBrokerDriver) TopicExists(ctx context.Context, topic string) (bool
 }
 
 // EnsureTopics creates topics if they don't already exist (idempotent).
-func (e *KafkaBrokerDriver) EnsureTopics(ctx context.Context, topics []string) error {
-	resp, err := e.admin.CreateTopics(ctx, 1, 1, nil, topics...)
-	if err != nil {
-		return fmt.Errorf("failed to create topics: %w", err)
-	}
+// metadata is an optional map of topic name -> key-value pairs for logging/tracking purposes.
+// For WebSubAPI topics, metadata typically includes: apiName, apiVersion, channelName.
+// Note: metadata is logged but not stored in Kafka (Kafka only accepts specific configuration keys).
+func (e *KafkaBrokerDriver) EnsureTopics(ctx context.Context, topics []string, metadata map[string]map[string]string) error {
+	// Create topics one by one to allow logging metadata per topic
+	for _, topic := range topics {
+		resp, err := e.admin.CreateTopics(ctx, 1, 1, nil, topic)
+		if err != nil {
+			return fmt.Errorf("failed to create topic %s: %w", topic, err)
+		}
 
-	for _, t := range resp.Sorted() {
-		if t.Err != nil {
-			// "TOPIC_ALREADY_EXISTS" is not a real failure for idempotent creates.
-			if isTopicAlreadyExistsErr(t.Err) {
-				slog.Debug("Topic already exists", "topic", t.Topic)
-				continue
+		for _, t := range resp.Sorted() {
+			if t.Err != nil {
+				// "TOPIC_ALREADY_EXISTS" is not a real failure for idempotent creates.
+				if isTopicAlreadyExistsErr(t.Err) {
+					slog.Debug("Topic already exists", "topic", t.Topic)
+					continue
+				}
+				return fmt.Errorf("failed to create topic %s: %w", t.Topic, t.Err)
+			}
+			slog.Info("Created topic", "topic", t.Topic)
+			// Log metadata for debugging/tracking purposes
+			if metadata != nil && metadata[t.Topic] != nil {
+				slog.Info("Topic metadata",
+					"topic_hash", t.Topic,
+					"apiName", metadata[t.Topic]["apiName"],
+					"apiVersion", metadata[t.Topic]["apiVersion"],
+					"channelName", metadata[t.Topic]["channelName"])
 			}
-			return fmt.Errorf("failed to create topic %s: %w", t.Topic, t.Err)
 		}
-		slog.Info("Created topic", "topic", t.Topic)
 	}
 
 	return nil

event-gateway/gateway-runtime/internal/connectors/receiver/websocket/broker_api_connector.go

@@ -102,7 +102,7 @@ func (e *WebBrokerApiReceiver) Start(ctx context.Context) error {
 		slog.Info("Ensuring Kafka topics exist",
 			"channel", e.channel.Name,
 			"topics", e.opts.Topics)
-		if err := e.brokerDriver.EnsureTopics(ctx, e.opts.Topics); err != nil {
+		if err := e.brokerDriver.EnsureTopics(ctx, e.opts.Topics, nil); err != nil {
 			return fmt.Errorf("failed to ensure kafka topics: %w", err)
 		}
 		slog.Info("Kafka topics verified",

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

@@ -133,14 +133,22 @@ func NewReceiver(cfg connectors.ReceiverConfig, opts Options) (connectors.Receiv
 // Start ensures Kafka topics exist and sets up the consumer manager context.
 // The HTTP server is managed by the runtime.
 func (e *WebSubReceiver) Start(ctx context.Context) error {
-	// Collect all Kafka topics to ensure.
+	// Collect all Kafka topics to ensure and build metadata for WebSubAPI topics.
 	var topicsToEnsure []string
-	for _, kafkaTopic := range e.channel.Channels {
+	topicMetadata := make(map[string]map[string]string)
+
+	for channelName, kafkaTopic := range e.channel.Channels {
 		topicsToEnsure = append(topicsToEnsure, kafkaTopic)
+		// Add metadata for WebSubAPI topics
+		topicMetadata[kafkaTopic] = map[string]string{
+			"apiName":     e.channel.Name,
+			"apiVersion":  e.channel.Version,
+			"channelName": channelName,
+		}
 	}
 
 	if len(topicsToEnsure) > 0 {
-		if err := e.brokerDriver.EnsureTopics(ctx, topicsToEnsure); err != nil {
+		if err := e.brokerDriver.EnsureTopics(ctx, topicsToEnsure, topicMetadata); err != nil {
 			return fmt.Errorf("failed to ensure kafka topics: %w", err)
 		}
 	}

event-gateway/gateway-runtime/internal/connectors/types.go

@@ -65,7 +65,7 @@ type BrokerDriver interface {
 	SubscribeManual(groupID string, topics []string, handler MessageHandler) (Receiver, error)
 	Replay(ctx context.Context, topic string, handler MessageHandler) error
 	TopicExists(ctx context.Context, topic string) (bool, error)
-	EnsureTopics(ctx context.Context, topics []string) error
+	EnsureTopics(ctx context.Context, topics []string, metadata map[string]map[string]string) error
 	EnsureCompactedTopic(ctx context.Context, topic string) error
 	DeleteTopics(ctx context.Context, topics []string) error
 	Close() error

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

@@ -132,7 +132,8 @@ func TestRemoveWebSubApiBinding_ClearsStalePolicyChains(t *testing.T) {
 
 func TestJoinNormalizedTopic_NormalizesUnsupportedCharacters(t *testing.T) {
 	got := binding.JoinNormalizedTopic("/orders/eu", "v1/test", "order_events")
-	want := "_2f_orders_2f_eu_v1_2f_test_order__events"
+	// SHA-256 hash of "10:/orders/eu|7:v1/test|12:order_events|"
+	want := "2836e285c333e251f0929fbbbe31dec3bb7d61423761103d790bcef2429b194b"
 	if got != want {
 		t.Fatalf("JoinNormalizedTopic() = %q, want %q", got, want)
 	}