fix(router): ignore header rules for pubsub trigger sources

router-tests/events/trigger_test.go

@@ -1,12 +1,15 @@
 package events_test
 
 import (
+	"net/http"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/require"
 	"github.com/wundergraph/cosmo/router-tests/testenv"
+	"github.com/wundergraph/cosmo/router/core"
+	"github.com/wundergraph/cosmo/router/pkg/config"
 )
 
 // TestEDFSTriggerDeduplication verifies trigger ID generation for Cosmo Streams subscriptions.
@@ -33,8 +36,9 @@ func TestEDFSTriggerDeduplication(t *testing.T) {
 				// provider ID, so they must resolve to the same trigger ID — wait for exactly
 				// one trigger to be initialized before asserting.
 				xEnv.WaitForTriggerCount(1, time.Second*10)
-				xEnv.RequireTriggerCount(1)
 				xEnv.NATSPublishUntilReceived(xEnv.NatsConnectionDefault, xEnv.GetPubSubName("employeeUpdated.3"), []byte(`{"id":3,"__typename":"Employee"}`), 2, time.Second*10)
+				// Assert the exact count only after both subscriptions have been served
+				xEnv.RequireTriggerCount(1)
 			}()
 
 			// Subscription 1: selects only id.
@@ -104,29 +108,43 @@ func TestEDFSTriggerDeduplication(t *testing.T) {
 		})
 	})
 
-	// Two subscriptions with the same query but different initial_payload (headers) should
+	// Two subscriptions with the same query but different headers should
 	// still share a single NATS trigger because the trigger ID is based on the NATS subject,
 	// not on connection-level metadata like headers.
-	t.Run("same subject different initial payload shares one trigger", func(t *testing.T) {
+	t.Run("same subject different headers shares one trigger", func(t *testing.T) {
 		t.Parallel()
 		testenv.Run(t, &testenv.Config{
 			RouterConfigJSONTemplate: testenv.ConfigWithEdfsNatsJSONTemplate,
 			EnableNats:               true,
+			RouterOptions: []core.Option{
+				core.WithHeaderRules(config.HeaderRules{
+					All: &config.GlobalHeaderRule{
+						Request: []*config.RequestHeaderRule{
+							{Operation: config.HeaderRuleOperationPropagate, Named: "Authorization"},
+						},
+					},
+				}),
+			},
 		}, func(t *testing.T, xEnv *testenv.Environment) {
 			var done sync.WaitGroup
 			done.Add(2)
 
 			go func() {
 				xEnv.WaitForSubscriptionCount(2, time.Second*10)
 				xEnv.WaitForTriggerCount(1, time.Second*10)
-				xEnv.RequireTriggerCount(1)
 				xEnv.NATSPublishUntilReceived(xEnv.NatsConnectionDefault, xEnv.GetPubSubName("employeeUpdated.3"), []byte(`{"id":3,"__typename":"Employee"}`), 2, time.Second*10)
+				// Asserted after both subscriptions have been served, see the comment in the
+				// "different selected fields" subtest above.
+				xEnv.RequireTriggerCount(1)
 			}()
 
-			// Subscription 1: sends Authorization header A in connection_init.
+			// Subscription 1: sends Authorization header A
 			go func() {
 				defer done.Done()
-				conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, []byte(`{"headers":{"Authorization":"Bearer token-a"}}`))
+				conn := xEnv.InitGraphQLWebSocketConnection(
+					http.Header{"Authorization": []string{"Bearer token-a"}}, nil,
+					[]byte(`{"headers":{"Authorization":"Bearer token-a"}}`),
+				)
 				defer conn.Close()
 
 				err := testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{
@@ -154,10 +172,13 @@ func TestEDFSTriggerDeduplication(t *testing.T) {
 				require.Equal(t, "1", complete.ID)
 			}()
 
-			// Subscription 2: sends a different Authorization header in connection_init.
+			// Subscription 2: sends Authorization header B
 			go func() {
 				defer done.Done()
-				conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, []byte(`{"headers":{"Authorization":"Bearer token-b"}}`))
+				conn := xEnv.InitGraphQLWebSocketConnection(
+					http.Header{"Authorization": []string{"Bearer token-b"}}, nil,
+					[]byte(`{"headers":{"Authorization":"Bearer token-b"}}`),
+				)
 				defer conn.Close()
 
 				err := testenv.WSWriteJSON(t, conn, &testenv.WebSocketMessage{

router/core/context.go

@@ -24,6 +24,7 @@ import (
 	"github.com/wundergraph/cosmo/router/pkg/authentication"
 	"github.com/wundergraph/cosmo/router/pkg/config"
 	"github.com/wundergraph/cosmo/router/pkg/graphqlschemausage"
+	pubsub "github.com/wundergraph/cosmo/router/pkg/pubsub/datasource"
 	ctrace "github.com/wundergraph/cosmo/router/pkg/trace"
 
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/httpclient"
@@ -334,6 +335,15 @@ func SubgraphHeadersBuilder(ctx *requestContext, headerPropagation *HeaderPropag
 		headers := make(map[string]*HeaderWithHash, len(p.Response.Response.DataSources)+1)
 		makeHeaders(headers, p.Response.Response.DataSources)
 
+		// avoid adding header rules for pubsub triggers sources, as no headers are passed to them.
+		_, isPubSub := p.Response.Trigger.Source.(pubsub.SubscriptionDataSource)
+		if isPubSub {
+			return &headerBuilder{
+				headers: headers,
+				allHash: keyGen.Sum64(),
+			}
+		}
+
 		h, hh := headerPropagation.BuildRequestHeaderForSubgraph(p.Response.Trigger.SourceName, ctx)
 		headers[p.Response.Trigger.SourceName] = &HeaderWithHash{
 			Header: h,

router/core/header_rule_engine_buildheader_test.go

@@ -10,6 +10,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/wundergraph/cosmo/router/pkg/config"
+	pubsub "github.com/wundergraph/cosmo/router/pkg/pubsub/datasource"
+	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/graphql_datasource"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/plan"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/resolve"
 	"go.uber.org/zap"
@@ -324,6 +326,109 @@ func TestSubgraphHeadersBuilder_SubscriptionPlan_IncludesTriggerAndResponse(t *t
 	assert.Equal(t, hashTrig, hashTrig2)
 }
 
+func TestSubgraphHeadersBuilder_SubscriptionPlan_SkipsPubSubTriggerSource(t *testing.T) {
+	ht, err := NewHeaderPropagation(t.Context(), zap.NewNop(), &config.HeaderRules{
+		All: &config.GlobalHeaderRule{
+			Request: []*config.RequestHeaderRule{
+				{Operation: "propagate", Named: "X-A"},
+				{Operation: "set", Name: "X-Static", Value: "static"},
+			},
+		},
+	}, nil)
+	require.NoError(t, err)
+
+	newCtx := func(headerValue string) *requestContext {
+		clientReq := httptest.NewRequest("POST", "http://localhost", nil)
+		clientReq.Header.Set("X-A", headerValue)
+
+		return &requestContext{
+			logger:           zap.NewNop(),
+			responseWriter:   httptest.NewRecorder(),
+			request:          clientReq,
+			operation:        &operationContext{},
+			subgraphResolver: NewSubgraphResolver(nil),
+		}
+	}
+
+	newPlan := func(source resolve.SubscriptionDataSource) *plan.SubscriptionResponsePlan {
+		return &plan.SubscriptionResponsePlan{
+			Response: &resolve.GraphQLSubscription{
+				Response: &resolve.GraphQLResponse{
+					DataSources: []resolve.DataSourceInfo{{Name: "sg-resp"}},
+				},
+				Trigger: resolve.GraphQLSubscriptionTrigger{
+					SourceName: "sg-trigger",
+					Source:     source,
+				},
+			},
+		}
+	}
+
+	t.Run("pubsub trigger source gets no headers", func(t *testing.T) {
+		ctx := newCtx("va")
+		pubSubDS := &pubsub.PubSubSubscriptionDataSource[pubsub.SubscriptionEventConfiguration]{}
+
+		hb := SubgraphHeadersBuilder(ctx, ht, newPlan(pubSubDS))
+		require.NotNil(t, hb)
+
+		// The trigger source is a pubsub data source, so no header rules are built for it.
+		hTrig, hashTrig := hb.HeadersForSubgraph("sg-trigger")
+		assert.Nil(t, hTrig)
+		assert.Zero(t, hashTrig)
+
+		// Regular response data sources are unaffected.
+		hResp, hashResp := hb.HeadersForSubgraph("sg-resp")
+		require.NotNil(t, hResp)
+		assert.Equal(t, "va", hResp.Get("X-A"))
+		assert.Equal(t, "static", hResp.Get("X-Static"))
+		require.NotZero(t, hashResp)
+	})
+
+	t.Run("pubsub trigger source is excluded from the all hash", func(t *testing.T) {
+		pubSubDS := &pubsub.PubSubSubscriptionDataSource[pubsub.SubscriptionEventConfiguration]{}
+		nonPubSubDS := &graphql_datasource.SubscriptionSource{}
+
+		pubSubHash := SubgraphHeadersBuilder(newCtx("va"), ht,
+			newPlan(pubSubDS)).HashAll()
+
+		// A plan containing only the response data source must produce the same hash, proving
+		// the trigger source no longer contributes to it.
+		responseOnlyHash := SubgraphHeadersBuilder(newCtx("va"), ht, &plan.SynchronousResponsePlan{
+			Response: &resolve.GraphQLResponse{
+				DataSources: []resolve.DataSourceInfo{{Name: "sg-resp"}},
+			},
+		}).HashAll()
+		assert.Equal(t, responseOnlyHash, pubSubHash)
+
+		// A non-pubsub trigger source still contributes to the hash.
+		regularHash := SubgraphHeadersBuilder(newCtx("va"), ht,
+			newPlan(nonPubSubDS)).HashAll()
+		assert.NotEqual(t, regularHash, pubSubHash)
+	})
+
+	t.Run("pubsub all hash is independent of propagated client headers", func(t *testing.T) {
+		pubSubDS := &pubsub.PubSubSubscriptionDataSource[pubsub.SubscriptionEventConfiguration]{}
+		pubSubOnlyPlan := func() *plan.SubscriptionResponsePlan {
+			return &plan.SubscriptionResponsePlan{
+				Response: &resolve.GraphQLSubscription{
+					Response: &resolve.GraphQLResponse{},
+					Trigger: resolve.GraphQLSubscriptionTrigger{
+						SourceName: "sg-trigger",
+						Source:     pubSubDS,
+					},
+				},
+			}
+		}
+
+		// Two clients sending different values for a propagated header must produce the same
+		// hash, so they share a single trigger instead of each creating their own.
+		assert.Equal(t,
+			SubgraphHeadersBuilder(newCtx("va"), ht, pubSubOnlyPlan()).HashAll(),
+			SubgraphHeadersBuilder(newCtx("vb"), ht, pubSubOnlyPlan()).HashAll(),
+		)
+	})
+}
+
 func TestSubgraphHeadersBuilder_MissingPrePopulatedCache(t *testing.T) {
 	ht, err := NewHeaderPropagation(t.Context(), zap.NewNop(), &config.HeaderRules{
 		All: &config.GlobalHeaderRule{