Handle WebSub delta updates without full undeploy/redeploy

event-gateway/gateway-runtime/go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/knadh/koanf/providers/env v1.1.0
 	github.com/knadh/koanf/providers/file v1.2.1
 	github.com/knadh/koanf/v2 v2.3.0
+	github.com/stretchr/testify v1.11.1
 	github.com/twmb/franz-go v1.18.1
 	github.com/twmb/franz-go/pkg/kadm v1.14.0
 	github.com/wso2/api-platform/common v0.0.0
@@ -26,6 +27,7 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cncf/xds/go v0.0.0-20251210132809-ee656c7534f5 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/envoyproxy/protoc-gen-validate v1.3.0 // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
@@ -39,6 +41,7 @@ require (
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/client_golang v1.23.2 // indirect
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.66.1 // indirect

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

@@ -22,6 +22,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"sync"
 	"time"
 
 	"github.com/wso2/api-platform/event-gateway/gateway-runtime/internal/binding"
@@ -55,6 +56,7 @@ type WebSubReceiver struct {
 	syncProducer   *subscription.SyncProducer
 	brokerDriver   connectors.BrokerDriver
 	channel        connectors.ChannelInfo
+	channelMu      sync.RWMutex
 	opts           Options
 }
 
@@ -98,36 +100,39 @@ func NewReceiver(cfg connectors.ReceiverConfig, opts Options) (connectors.Receiv
 
 	verificationTimeout := time.Duration(opts.VerificationTimeoutSeconds) * time.Second
 
+	receiver := &WebSubReceiver{
+		deliverer:    deliverer,
+		topics:       topics,
+		store:        store,
+		consumerMgr:  consumerMgr,
+		syncProducer: syncProducer,
+		brokerDriver: cfg.BrokerDriver,
+		channel:      cfg.Channel,
+		opts:         opts,
+	}
+
 	// Create HubHandler for subscribe/unsubscribe on {context}/{version}/hub.
 	hubHandler := NewHubHandler(
 		topics, store, verificationTimeout, opts.DefaultLeaseSeconds,
 		cfg.Processor, cfg.BrokerDriver, cfg.Channel.Name,
-		cfg.Channel.Channels, consumerMgr, syncProducer,
+		cfg.Channel.Channels, &receiver.channelMu, consumerMgr, syncProducer,
 	)
 
 	// Create WebhookReceiverHandler for ingress on {context}/{version}/webhook-receiver.
 	webhookHandler := NewWebhookReceiverHandler(
 		topics, cfg.Processor, cfg.BrokerDriver,
-		cfg.Channel.Name, cfg.Channel.Channels,
+		cfg.Channel.Name, cfg.Channel.Channels, &receiver.channelMu,
 	)
 
 	// Register handlers on shared mux.
 	basePath := binding.WebSubApiBasePath(cfg.Channel.Context, cfg.Channel.Version)
 	cfg.Mux.Handle(basePath+"/hub", hubHandler)
 	cfg.Mux.Handle(basePath+"/webhook-receiver", webhookHandler)
 
-	return &WebSubReceiver{
-		hubHandler:     hubHandler,
-		webhookHandler: webhookHandler,
-		deliverer:      deliverer,
-		topics:         topics,
-		store:          store,
-		consumerMgr:    consumerMgr,
-		syncProducer:   syncProducer,
-		brokerDriver:   cfg.BrokerDriver,
-		channel:        cfg.Channel,
-		opts:           opts,
-	}, nil
+	receiver.hubHandler = hubHandler
+	receiver.webhookHandler = webhookHandler
+
+	return receiver, nil
 }
 
 // Start ensures Kafka topics exist and sets up the consumer manager context.

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

@@ -0,0 +1,88 @@
+/*
+ * Copyright (c) 2026, WSO2 LLC. (https://www.wso2.com).
+ *
+ * WSO2 LLC. licenses this file to you under the Apache License,
+ * Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package websub
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+)
+
+// ApplyBindingDelta mutates the live receiver for channel add/remove changes
+// without recreating the receiver or the subscription sync topic.
+func (e *WebSubReceiver) ApplyBindingDelta(ctx context.Context, removedChannels map[string]string, addedChannels map[string]string) error {
+	var tombstoneErrs []error
+
+	for channelName := range removedChannels {
+		subscriptions := e.store.GetByTopic(channelName)
+		for _, sub := range subscriptions {
+			if e.syncProducer != nil {
+				if err := e.syncProducer.PublishTombstone(ctx, channelName, sub.CallbackURL); err != nil {
+					slog.Error("Failed to tombstone subscription for removed WebSub channel",
+						"api", e.channel.Name,
+						"channel", channelName,
+						"callback", sub.CallbackURL,
+						"error", err)
+					tombstoneErrs = append(tombstoneErrs,
+						fmt.Errorf("failed to tombstone subscription for removed channel %q callback %q: %w", channelName, sub.CallbackURL, err))
+				}
+			}
+		}
+	}
+
+	for channelName, kafkaTopic := range removedChannels {
+		e.topics.Deregister(channelName)
+
+		subscriptions := e.store.GetByTopic(channelName)
+		for _, sub := range subscriptions {
+			if err := e.consumerMgr.RemoveSubscription(sub.CallbackURL, kafkaTopic); err != nil {
+				slog.Error("Failed to remove consumer for deleted WebSub channel",
+					"api", e.channel.Name,
+					"channel", channelName,
+					"callback", sub.CallbackURL,
+					"error", err)
+			}
+			if err := e.store.Remove(channelName, sub.CallbackURL); err != nil {
+				slog.Error("Failed to remove subscription for deleted WebSub channel",
+					"api", e.channel.Name,
+					"channel", channelName,
+					"callback", sub.CallbackURL,
+					"error", err)
+			}
+		}
+
+		e.channelMu.Lock()
+		delete(e.channel.Channels, channelName)
+		e.channelMu.Unlock()
+	}
+
+	for channelName, kafkaTopic := range addedChannels {
+		e.channelMu.Lock()
+		e.channel.Channels[channelName] = kafkaTopic
+		e.channelMu.Unlock()
+		e.topics.Register(channelName)
+	}
+
+	if len(tombstoneErrs) > 0 {
+		return errors.Join(tombstoneErrs...)
+	}
+
+	return nil
+}

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

@@ -0,0 +1,162 @@
+package websub
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/wso2/api-platform/event-gateway/gateway-runtime/internal/connectors"
+	"github.com/wso2/api-platform/event-gateway/gateway-runtime/internal/subscription"
+)
+
+type deltaTestBrokerDriver struct {
+	publishErrs map[string]error
+	published   []string
+}
+
+func (d *deltaTestBrokerDriver) Publish(_ context.Context, topic string, msg *connectors.Message) error {
+	key := string(msg.Key)
+	d.published = append(d.published, key)
+	if err, ok := d.publishErrs[key]; ok {
+		return err
+	}
+	return nil
+}
+
+func (d *deltaTestBrokerDriver) Subscribe(string, []string, connectors.MessageHandler) (connectors.Receiver, error) {
+	return nil, nil
+}
+
+func (d *deltaTestBrokerDriver) SubscribeManual(string, []string, connectors.MessageHandler) (connectors.Receiver, error) {
+	return nil, nil
+}
+
+func (d *deltaTestBrokerDriver) Replay(context.Context, string, connectors.MessageHandler) error {
+	return nil
+}
+
+func (d *deltaTestBrokerDriver) TopicExists(context.Context, string) (bool, error) {
+	return false, nil
+}
+
+func (d *deltaTestBrokerDriver) EnsureTopics(context.Context, []string) error {
+	return nil
+}
+
+func (d *deltaTestBrokerDriver) EnsureCompactedTopic(context.Context, string) error {
+	return nil
+}
+
+func (d *deltaTestBrokerDriver) DeleteTopics(context.Context, []string) error {
+	return nil
+}
+
+func (d *deltaTestBrokerDriver) Close() error {
+	return nil
+}
+
+type deltaTestReceiver struct {
+	stopCalls int
+}
+
+func (r *deltaTestReceiver) Start(context.Context) error {
+	return nil
+}
+
+func (r *deltaTestReceiver) Stop(context.Context) error {
+	r.stopCalls++
+	return nil
+}
+
+func TestApplyBindingDeltaContinuesAfterTombstoneFailure(t *testing.T) {
+	store := subscription.NewInMemoryStore("runtime-1")
+	require.NoError(t, store.Add(&subscription.Subscription{
+		ID:          "sub-1",
+		Topic:       "removed-a",
+		CallbackURL: "https://callback-a.example",
+		State:       subscription.StateActive,
+	}))
+	require.NoError(t, store.Add(&subscription.Subscription{
+		ID:          "sub-2",
+		Topic:       "removed-b",
+		CallbackURL: "https://callback-b.example",
+		State:       subscription.StateActive,
+	}))
+
+	topics := NewTopicRegistry()
+	topics.Register("removed-a")
+	topics.Register("removed-b")
+
+	consumerA := &deltaTestReceiver{}
+	consumerB := &deltaTestReceiver{}
+	consumerMgr := &ConsumerManager{
+		consumers: map[string]*managedConsumer{
+			"https://callback-a.example": {
+				consumer: consumerA,
+				topics:   map[string]bool{"kafka-a": true},
+			},
+			"https://callback-b.example": {
+				consumer: consumerB,
+				topics:   map[string]bool{"kafka-b": true},
+			},
+		},
+	}
+
+	driver := &deltaTestBrokerDriver{
+		publishErrs: map[string]error{
+			"removed-a:https://callback-a.example": errors.New("boom"),
+		},
+	}
+
+	receiver := &WebSubReceiver{
+		topics:      topics,
+		store:       store,
+		consumerMgr: consumerMgr,
+		syncProducer: subscription.NewSyncProducer(
+			driver,
+			"runtime-1",
+			"internal-sub-topic",
+		),
+		channel: connectors.ChannelInfo{
+			Name: "test-api",
+			Channels: map[string]string{
+				"removed-a": "kafka-a",
+				"removed-b": "kafka-b",
+			},
+		},
+	}
+
+	err := receiver.ApplyBindingDelta(context.Background(),
+		map[string]string{
+			"removed-a": "kafka-a",
+			"removed-b": "kafka-b",
+		},
+		map[string]string{
+			"added-c": "kafka-c",
+		},
+	)
+
+	require.Error(t, err)
+	require.ErrorContains(t, err, `failed to tombstone subscription for removed channel "removed-a"`)
+
+	require.ElementsMatch(t, []string{
+		"removed-a:https://callback-a.example",
+		"removed-b:https://callback-b.example",
+	}, driver.published)
+
+	require.Nil(t, store.GetByTopic("removed-a"))
+	require.Nil(t, store.GetByTopic("removed-b"))
+
+	require.Equal(t, 1, consumerA.stopCalls)
+	require.Equal(t, 1, consumerB.stopCalls)
+	require.Empty(t, consumerMgr.consumers)
+
+	require.NotContains(t, receiver.channel.Channels, "removed-a")
+	require.NotContains(t, receiver.channel.Channels, "removed-b")
+	require.Equal(t, "kafka-c", receiver.channel.Channels["added-c"])
+
+	require.False(t, topics.IsRegistered("removed-a"))
+	require.False(t, topics.IsRegistered("removed-b"))
+	require.True(t, topics.IsRegistered("added-c"))
+}