test: Add E2E integration and load tests for EventStream pipeline (#1115)

* test: Add E2E integration and load tests for EventStream pipeline

Full-stack integration tests covering the Kafka -> EventSource -> FanOut
-> Router -> Handler -> WebSocket client pipeline:

- Kafka testcontainer + miniredis full-stack E2E test
- InProcessFanOut-based E2E delivery test
- Cross-tenant isolation verification (zero leakage)
- Multi-tenant concurrent event delivery
- Channel-scoped subscription filtering
- 100 concurrent WebSocket connections load test (50k events/sec)
- 5000 events/sec sustained throughput test
- Go benchmark for connection concurrency

* fix: Use cancel-only context in benchmark to prevent timeout with large b.N

* refactor: Address CodeRabbit review feedback

- Export ConsumerGroupID constant to prevent test drift between
  integration_test.go and kafka_source.go
- Replace tautological const assertion with runtime registry check

---------

Co-authored-by: Ben Coombs <bjcoombs@users.noreply.github.com>

Author: bjcoombs

services/gateway/eventstream/adapters/kafka_source.go

@@ -25,7 +25,8 @@ const (
 	headerCorrelationID = "correlation_id"
 	headerCausationID   = "causation_id"
 
-	consumerGroupID = "ops-console-events"
+	// ConsumerGroupID is the Kafka consumer group used by KafkaEventSource.
+	ConsumerGroupID = "ops-console-events"
 )
 
 // Sentinel errors returned by KafkaEventSource constructors.
@@ -90,7 +91,7 @@ func NewKafkaEventSource(
 
 	opts := []kgo.Opt{
 		kgo.SeedBrokers(brokers...),
-		kgo.ConsumerGroup(consumerGroupID),
+		kgo.ConsumerGroup(ConsumerGroupID),
 		kgo.ConsumeTopics(topics...),
 		kgo.ConsumeResetOffset(kgo.NewOffset().AtEnd()),
 		kgo.BlockRebalanceOnPoll(),

services/gateway/eventstream/adapters/kafka_source_test.go

@@ -349,7 +349,7 @@ func TestKafkaEventSource_Integration(t *testing.T) {
 			AtMost(10 * time.Second).
 			PollInterval(200 * time.Millisecond).
 			UntilNoError(func() error {
-				return consumerGroupActive(ctx, brokerAddr, consumerGroupID)
+				return consumerGroupActive(ctx, brokerAddr, ConsumerGroupID)
 			}); err != nil {
 			t.Fatalf("consumer group did not become active: %v", err)
 		}
@@ -409,7 +409,7 @@ func TestKafkaEventSource_Integration(t *testing.T) {
 			AtMost(5 * time.Second).
 			PollInterval(100 * time.Millisecond).
 			UntilNoError(func() error {
-				return consumerGroupActive(ctx, brokerAddr, consumerGroupID)
+				return consumerGroupActive(ctx, brokerAddr, ConsumerGroupID)
 			}); err != nil {
 			t.Logf("consumer group may not be fully active: %v (continuing)", err)
 		}
@@ -464,7 +464,7 @@ func TestKafkaEventSource_Integration(t *testing.T) {
 			AtMost(15 * time.Second).
 			PollInterval(200 * time.Millisecond).
 			UntilNoError(func() error {
-				return consumerGroupActive(ctx3, brokerAddr, consumerGroupID)
+				return consumerGroupActive(ctx3, brokerAddr, ConsumerGroupID)
 			}); err != nil {
 			t.Fatalf("consumer group did not become active: %v", err)
 		}