fix: preserve nullable nested input presence in gRPC compiler (#1447)

## Summary
- treat omitted and null nested input messages as absent during protobuf
compilation
- preserve explicit empty objects instead of conflating them with null
- add a regression test covering omitted, null, empty-object and
explicit-value cases

## Testing
- go test ./v2/pkg/engine/datasource/grpc_datasource

Refs wundergraph/cosmo#2650

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Bug Fixes**
* More accurate detection of present vs absent nested inputs, stricter
enforcement of required fields, and correct skipping of empty optional
scalar values during gRPC datasource compilation.

* **Tests**
* Added tests exercising omitted, null, empty-object, and explicit
nested input scenarios to validate compiler behavior.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: fengyuwusong

v2/pkg/engine/datasource/grpc_datasource/compiler.go

@@ -1021,6 +1021,7 @@ func (p *RPCCompiler) buildProtoMessage(inputMessage Message, rpcMessage *RPCMes
 				fieldMsg protoref.Message
 				err      error
 			)
+			fieldData := data.Get(rpcField.JSONPath)
 
 			switch {
 			case rpcField.IsListType:
@@ -1031,7 +1032,7 @@ func (p *RPCCompiler) buildProtoMessage(inputMessage Message, rpcMessage *RPCMes
 				//   ListOfBoolean is_published = 1;
 				//   ListOfListOfString related_topics = 2;
 				// }
-				if !data.Get(rpcField.JSONPath).Exists() {
+				if !isNonNullValue(fieldData) {
 					if !rpcField.Optional {
 						return nil, fmt.Errorf("field %s is required but has no value", rpcField.JSONPath)
 					}
@@ -1056,8 +1057,8 @@ func (p *RPCCompiler) buildProtoMessage(inputMessage Message, rpcMessage *RPCMes
 				// If the field is optional, we are handling a scalar value that is wrapped in a message
 				// as protobuf scalar types are not nullable.
 
-				if !data.Get(rpcField.JSONPath).Exists() {
-					// If we don't have a value for an optional field, we skip it to provide a null message.
+				if !isNonNullValue(fieldData) {
+					// If we don't have a concrete value for an optional field, leave it unset (absent).
 					continue
 				}
 
@@ -1072,7 +1073,15 @@ func (p *RPCCompiler) buildProtoMessage(inputMessage Message, rpcMessage *RPCMes
 					return nil, err
 				}
 			default:
-				fieldMsg, err = p.buildProtoMessage(p.doc.Messages[field.MessageRef], rpcField.Message, data.Get(rpcField.JSONPath))
+				if !isNonNullValue(fieldData) {
+					if !rpcField.Optional {
+						return nil, fmt.Errorf("field %s is required but has no value", rpcField.JSONPath)
+					}
+
+					continue
+				}
+
+				fieldMsg, err = p.buildProtoMessage(p.doc.Messages[field.MessageRef], rpcField.Message, fieldData)
 				if err != nil {
 					return nil, err
 				}
@@ -1103,6 +1112,10 @@ func (p *RPCCompiler) buildProtoMessage(inputMessage Message, rpcMessage *RPCMes
 	return message, nil
 }
 
+func isNonNullValue(data gjson.Result) bool {
+	return data.Exists() && data.Type != gjson.Null
+}
+
 // buildListMessage creates a new protobuf message, which reflects a wrapper type to work with a list in GraphQL.
 // A list wrapper type has an inner message type, which contains a repeated field.
 // We need this to make sure we can differentiate between a null list and an empty list, as repeated fields are not nullable.

v2/pkg/engine/datasource/grpc_datasource/compiler_input_presence_test.go

@@ -0,0 +1,141 @@
+package grpcdatasource
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/tidwall/gjson"
+	"google.golang.org/protobuf/reflect/protoreflect"
+)
+
+const protoSchemaWithOptionalNestedInputs = `
+syntax = "proto3";
+package rule.v1;
+
+import "google/protobuf/wrappers.proto";
+
+service RuleService {
+  rpc UpdateRule(UpdateRuleRequest) returns (UpdateRuleResponse) {}
+}
+
+message UpdateRuleRequest {
+  ConditionsInput conditions = 1;
+  google.protobuf.StringValue params = 2;
+}
+
+message ConditionsInput {
+  string key = 1;
+}
+
+message UpdateRuleResponse {}
+`
+
+func TestCompileOptionalNestedInputsTreatsNullAsAbsent(t *testing.T) {
+	t.Parallel()
+
+	testCases := []struct {
+		name                string
+		input               string
+		expectConditions    bool
+		expectConditionsKey bool
+		expectParams        bool
+	}{
+		{
+			name:             "omitted optional fields stay absent",
+			input:            `{}`,
+			expectConditions: false,
+			expectParams:     false,
+		},
+		{
+			name:             "null optional fields stay absent",
+			input:            `{"conditions":null,"params":null}`,
+			expectConditions: false,
+			expectParams:     false,
+		},
+		{
+			name:                "explicit empty object keeps nested message",
+			input:               `{"conditions":{}}`,
+			expectConditions:    true,
+			expectConditionsKey: false,
+			expectParams:        false,
+		},
+		{
+			name:                "explicit nested value is preserved",
+			input:               `{"conditions":{"key":"status"},"params":"{\"value\":\"1212\"}"}`,
+			expectConditions:    true,
+			expectConditionsKey: true,
+			expectParams:        true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			compiler, err := NewProtoCompiler(protoSchemaWithOptionalNestedInputs, nil)
+			require.NoError(t, err)
+
+			inputMessage, ok := compiler.doc.MessageByName("UpdateRuleRequest")
+			require.True(t, ok)
+
+			request, err := compiler.buildProtoMessage(inputMessage, optionalNestedInputsMessage(), gjson.Parse(tc.input))
+			require.NoError(t, err)
+
+			conditionsDesc := request.Descriptor().Fields().ByName(protoreflect.Name("conditions"))
+			require.NotNil(t, conditionsDesc)
+			require.Equal(t, tc.expectConditions, request.Has(conditionsDesc))
+
+			if tc.expectConditions {
+				conditions := request.Get(conditionsDesc).Message()
+				keyDesc := conditions.Descriptor().Fields().ByName(protoreflect.Name("key"))
+				require.NotNil(t, keyDesc)
+				require.Equal(t, tc.expectConditionsKey, conditions.Has(keyDesc))
+				if tc.expectConditionsKey {
+					require.Equal(t, "status", conditions.Get(keyDesc).String())
+				}
+			}
+
+			paramsDesc := request.Descriptor().Fields().ByName(protoreflect.Name("params"))
+			require.NotNil(t, paramsDesc)
+			require.Equal(t, tc.expectParams, request.Has(paramsDesc))
+
+			if tc.expectParams {
+				params := request.Get(paramsDesc).Message()
+				valueDesc := params.Descriptor().Fields().ByName(protoreflect.Name("value"))
+				require.NotNil(t, valueDesc)
+				require.True(t, params.Has(valueDesc))
+				require.Equal(t, `{"value":"1212"}`, params.Get(valueDesc).String())
+			}
+		})
+	}
+}
+
+func optionalNestedInputsMessage() *RPCMessage {
+	return &RPCMessage{
+		Name: "UpdateRuleRequest",
+		Fields: RPCFields{
+			{
+				Name:          "conditions",
+				ProtoTypeName: DataTypeMessage,
+				JSONPath:      "conditions",
+				Optional:      true,
+				Message: &RPCMessage{
+					Name: "ConditionsInput",
+					Fields: RPCFields{
+						{
+							Name:          "key",
+							ProtoTypeName: DataTypeString,
+							JSONPath:      "key",
+							Optional:      true,
+						},
+					},
+				},
+			},
+			{
+				Name:          "params",
+				ProtoTypeName: DataTypeString,
+				JSONPath:      "params",
+				Optional:      true,
+			},
+		},
+	}
+}