feat(router): provide field arguments in operation context

router-tests/modules/start_subscription_test.go

@@ -711,4 +711,76 @@ func TestStartSubscriptionHook(t *testing.T) {
 			assert.Equal(t, int32(1), customModule.HookCallCount.Load())
 		})
 	})
+
+	t.Run("Test StartSubscription hook can access field arguments", func(t *testing.T) {
+		t.Parallel()
+
+		// This test verifies that the subscription start hook can access GraphQL field arguments
+		// via ctx.Operation().Arguments().
+
+		var capturedEmployeeID int
+
+		customModule := &start_subscription.StartSubscriptionModule{
+			HookCallCount: &atomic.Int32{},
+			Callback: func(ctx core.SubscriptionOnStartHandlerContext) error {
+				args := ctx.Operation().Arguments()
+				if args != nil {
+					employeeIDArg := args.Get("employeeUpdatedMyKafka", "employeeID")
+					if employeeIDArg != nil {
+						capturedEmployeeID = employeeIDArg.GetInt()
+					}
+				}
+				return nil
+			},
+		}
+
+		cfg := config.Config{
+			Graph: config.Graph{},
+			Modules: map[string]interface{}{
+				"startSubscriptionModule": customModule,
+			},
+		}
+
+		testenv.Run(t, &testenv.Config{
+			RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
+			EnableKafka:              true,
+			RouterOptions: []core.Option{
+				core.WithModulesConfig(cfg.Modules),
+				core.WithCustomModules(&start_subscription.StartSubscriptionModule{}),
+			},
+		}, func(t *testing.T, xEnv *testenv.Environment) {
+			var subscriptionOne struct {
+				employeeUpdatedMyKafka struct {
+					ID float64 `graphql:"id"`
+				} `graphql:"employeeUpdatedMyKafka(employeeID: $employeeID)"`
+			}
+
+			surl := xEnv.GraphQLWebSocketSubscriptionURL()
+			client := graphql.NewSubscriptionClient(surl)
+
+			vars := map[string]interface{}{
+				"employeeID": 7,
+			}
+			subscriptionOneID, err := client.Subscribe(&subscriptionOne, vars, func(dataValue []byte, errValue error) error {
+				return nil
+			})
+			require.NoError(t, err)
+			require.NotEmpty(t, subscriptionOneID)
+
+			clientRunCh := make(chan error)
+			go func() {
+				clientRunCh <- client.Run()
+			}()
+
+			xEnv.WaitForSubscriptionCount(1, time.Second*10)
+
+			require.NoError(t, client.Close())
+			testenv.AwaitChannelWithT(t, time.Second*10, clientRunCh, func(t *testing.T, err error) {
+				require.NoError(t, err)
+			}, "unable to close client before timeout")
+
+			assert.Equal(t, int32(1), customModule.HookCallCount.Load())
+			assert.Equal(t, 7, capturedEmployeeID, "expected to capture employeeID argument value")
+		})
+	})
 }

router-tests/modules/stream_publish_test.go

@@ -358,4 +358,52 @@ func TestPublishHook(t *testing.T) {
 			require.Equal(t, []byte("3"), header.Value)
 		})
 	})
+
+	t.Run("Test Publish hook can access field arguments", func(t *testing.T) {
+		t.Parallel()
+
+		// This test verifies that the publish hook can access GraphQL field arguments
+		// via ctx.Operation().Arguments().
+
+		var capturedEmployeeID int
+
+		customModule := stream_publish.PublishModule{
+			HookCallCount: &atomic.Int32{},
+			Callback: func(ctx core.StreamPublishEventHandlerContext, events datasource.StreamEvents) (datasource.StreamEvents, error) {
+				args := ctx.Operation().Arguments()
+				if args != nil {
+					employeeIDArg := args.Get("updateEmployeeMyKafka", "employeeID")
+					if employeeIDArg != nil {
+						capturedEmployeeID = employeeIDArg.GetInt()
+					}
+				}
+				return events, nil
+			},
+		}
+
+		cfg := config.Config{
+			Graph: config.Graph{},
+			Modules: map[string]interface{}{
+				"publishModule": customModule,
+			},
+		}
+
+		testenv.Run(t, &testenv.Config{
+			RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
+			EnableKafka:              true,
+			RouterOptions: []core.Option{
+				core.WithModulesConfig(cfg.Modules),
+				core.WithCustomModules(&stream_publish.PublishModule{}),
+			},
+		}, func(t *testing.T, xEnv *testenv.Environment) {
+			events.KafkaEnsureTopicExists(t, xEnv, time.Second, "employeeUpdated")
+			resOne := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{
+				Query: `mutation { updateEmployeeMyKafka(employeeID: 5, update: {name: "test"}) { success } }`,
+			})
+			require.JSONEq(t, `{"data":{"updateEmployeeMyKafka":{"success":true}}}`, resOne.Body)
+
+			assert.Equal(t, int32(1), customModule.HookCallCount.Load())
+			assert.Equal(t, 5, capturedEmployeeID, "expected to capture employeeID argument value")
+		})
+	})
 }

router-tests/modules/stream_receive_test.go

@@ -963,4 +963,91 @@ func TestReceiveHook(t *testing.T) {
 			assert.Equal(t, int32(3), customModule.HookCallCount.Load())
 		})
 	})
+
+	t.Run("Test Receive hook can access field arguments", func(t *testing.T) {
+		t.Parallel()
+
+		// This test verifies that the receive hook can access GraphQL field arguments
+		// via ctx.Operation().Arguments().
+
+		var capturedEmployeeID int
+
+		customModule := stream_receive.StreamReceiveModule{
+			HookCallCount: &atomic.Int32{},
+			Callback: func(ctx core.StreamReceiveEventHandlerContext, events datasource.StreamEvents) (datasource.StreamEvents, error) {
+				args := ctx.Operation().Arguments()
+				if args != nil {
+					employeeIDArg := args.Get("employeeUpdatedMyKafka", "employeeID")
+					if employeeIDArg != nil {
+						capturedEmployeeID = employeeIDArg.GetInt()
+					}
+				}
+				return events, nil
+			},
+		}
+
+		cfg := config.Config{
+			Graph: config.Graph{},
+			Modules: map[string]interface{}{
+				"streamReceiveModule": customModule,
+			},
+		}
+
+		testenv.Run(t, &testenv.Config{
+			RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
+			EnableKafka:              true,
+			RouterOptions: []core.Option{
+				core.WithModulesConfig(cfg.Modules),
+				core.WithCustomModules(&stream_receive.StreamReceiveModule{}),
+			},
+		}, func(t *testing.T, xEnv *testenv.Environment) {
+			topics := []string{"employeeUpdated"}
+			events.KafkaEnsureTopicExists(t, xEnv, time.Second, topics...)
+
+			var subscriptionOne struct {
+				employeeUpdatedMyKafka struct {
+					ID float64 `graphql:"id"`
+				} `graphql:"employeeUpdatedMyKafka(employeeID: 3)"`
+			}
+
+			surl := xEnv.GraphQLWebSocketSubscriptionURL()
+			client := graphql.NewSubscriptionClient(surl)
+
+			type kafkaSubscriptionArgs struct {
+				dataValue []byte
+				errValue  error
+			}
+			subscriptionArgsCh := make(chan kafkaSubscriptionArgs)
+			subscriptionOneID, err := client.Subscribe(&subscriptionOne, nil, func(dataValue []byte, errValue error) error {
+				subscriptionArgsCh <- kafkaSubscriptionArgs{
+					dataValue: dataValue,
+					errValue:  errValue,
+				}
+				return nil
+			})
+			require.NoError(t, err)
+			require.NotEmpty(t, subscriptionOneID)
+
+			clientRunCh := make(chan error)
+			go func() {
+				clientRunCh <- client.Run()
+			}()
+
+			xEnv.WaitForSubscriptionCount(1, Timeout)
+
+			events.ProduceKafkaMessage(t, xEnv, Timeout, topics[0], `{"__typename":"Employee","id": 1,"update":{"name":"foo"}}`)
+
+			testenv.AwaitChannelWithT(t, Timeout, subscriptionArgsCh, func(t *testing.T, args kafkaSubscriptionArgs) {
+				require.NoError(t, args.errValue)
+			})
+
+			require.NoError(t, client.Close())
+			testenv.AwaitChannelWithT(t, Timeout, clientRunCh, func(t *testing.T, err error) {
+				require.NoError(t, err)
+			}, "unable to close client before timeout")
+
+			assert.Equal(t, int32(1), customModule.HookCallCount.Load())
+			assert.Equal(t, 3, capturedEmployeeID, "expected to capture employeeID argument value")
+		})
+	})
 }

router/core/arguments.go

@@ -0,0 +1,43 @@
+package core
+
+import "github.com/wundergraph/astjson"
+
+// Arguments allow access to GraphQL field arguments used by clients.
+type Arguments struct {
+	// First key is the path to the field in dot notation
+	// i.e. root_field.subfield1.subfield2.
+	// Second argument is the name of the argument of that field.
+	data map[string]map[string]*astjson.Value
+}
+
+// Get will return the value of argument a from field f.
+//
+// To access the an argument of a root level field, you need to pass the
+// name of the field as the first argument to Get and the name of the argument
+// as the second argument, e.g. Get("rootfield_name", "argument_name") .
+//
+// The field needs to be a dot notated path to the position of the field,
+// if it is nested.
+// For example you can access arg1 on field2 on the operation
+//
+//	subscription {
+//		mySub(arg1: "val1", arg2: "val2") {
+//			field1
+//			field2(arg1: "val3", arg2: "val4")
+//		}
+//
+// You need to call Get("mySub.field2", "arg1") .
+//
+// If f or a cannot be found nil is returned.
+func (fa *Arguments) Get(f string, a string) *astjson.Value {
+	if fa == nil {
+		return nil
+	}
+
+	args, found := fa.data[f]
+	if !found {
+		return nil
+	}
+
+	return args[a]
+}

router/core/context.go

@@ -482,16 +482,16 @@ type OperationContext interface {
 	Hash() uint64
 	// Content is the content of the operation
 	Content() string
-	// Variables is the variables of the operation
+	// Arguments allow access to GraphQL operation field arguments.
+	Arguments() *Arguments
+	// Variables allow access to GraphQL operation variables.
 	Variables() *astjson.Value
 	// ClientInfo returns information about the client that initiated this operation
 	ClientInfo() ClientInfo
-
 	// Sha256Hash returns the SHA256 hash of the original operation
 	// It is important to note that this hash is not calculated just because this method has been called
 	// and is only calculated based on other existing logic (such as if sha256Hash is used in expressions)
 	Sha256Hash() string
-
 	// QueryPlanStats returns some statistics about the query plan for the operation
 	// if called too early in request chain, it may be inaccurate for modules, using
 	// in Middleware is recommended
@@ -524,10 +524,11 @@ type operationContext struct {
 	// RawContent is the raw content of the operation
 	rawContent string
 	// Content is the normalized content of the operation
-	content    string
-	variables  *astjson.Value
-	files      []*httpclient.FileUpload
-	clientInfo *ClientInfo
+	content        string
+	fieldArguments Arguments
+	variables      *astjson.Value
+	files          []*httpclient.FileUpload
+	clientInfo     *ClientInfo
 	// preparedPlan is the prepared plan of the operation
 	preparedPlan     *planWithMetaData
 	traceOptions     resolve.TraceOptions
@@ -560,6 +561,10 @@ func (o *operationContext) Variables() *astjson.Value {
 	return o.variables
 }
 
+func (o *operationContext) Arguments() *Arguments {
+	return &o.fieldArguments
+}
+
 func (o *operationContext) Files() []*httpclient.FileUpload {
 	return o.files
 }