diff --git a/advanced/samples/go/go.mod b/advanced/samples/go/go.mod index 27cbe994..3270bd77 100644 --- a/advanced/samples/go/go.mod +++ b/advanced/samples/go/go.mod @@ -6,7 +6,7 @@ require ( github.com/nexus-rpc/sdk-go v0.7.0 github.com/stretchr/testify v1.10.0 go.temporal.io/api v1.63.5 - go.temporal.io/sdk v1.49.1-0.20261006234944-b7c855cd3699 + go.temporal.io/sdk v1.49.1-0.20261007160123-1fa533e9c8c0 google.golang.org/protobuf v1.36.11 ) diff --git a/advanced/samples/go/go.sum b/advanced/samples/go/go.sum index e32ee5a8..ae5c56c2 100644 --- a/advanced/samples/go/go.sum +++ b/advanced/samples/go/go.sum @@ -59,8 +59,8 @@ go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/ go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= go.temporal.io/api v1.63.5 h1:c11+kPYHkXXL3UiShPdbMD+xtvqGsbTibUA9ypmiCa4= go.temporal.io/api v1.63.5/go.mod h1:SrlW2JMwVlDP4nRWSNznUFqnSHd+YeMDS1BkYo63HCQ= -go.temporal.io/sdk v1.49.1-0.20261006234944-b7c855cd3699 h1:H0lzIOKvMyITmdudf0Aon4wcMGbDBPQw3YyVsZABU2I= -go.temporal.io/sdk v1.49.1-0.20261006234944-b7c855cd3699/go.mod h1:xP0FulN5JJSfisESUP60LlWsrKz2tLSStGjVdk8r5cc= +go.temporal.io/sdk v1.49.1-0.20261007160123-1fa533e9c8c0 h1:Gw6hViAZbsqVT2iNAv8xJHf9zPKbxRQX8+p/RaLa1mw= +go.temporal.io/sdk v1.49.1-0.20261007160123-1fa533e9c8c0/go.mod h1:xP0FulN5JJSfisESUP60LlWsrKz2tLSStGjVdk8r5cc= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= diff --git a/advanced/samples/go/tests/workflowservice_test.go b/advanced/samples/go/tests/workflowservice_test.go index 3e3e5a04..a80a88f3 100644 --- a/advanced/samples/go/tests/workflowservice_test.go +++ b/advanced/samples/go/tests/workflowservice_test.go @@ -3,6 +3,8 @@ package tests import ( "context" "errors" + "reflect" + "sync" "testing" "time" @@ -12,8 +14,11 @@ import ( enums "go.temporal.io/api/enums/v1" workflowservicepb "go.temporal.io/api/workflowservice/v1" "go.temporal.io/sdk/converter" + "go.temporal.io/sdk/interceptor" + "go.temporal.io/sdk/internal" "go.temporal.io/sdk/temporal" "go.temporal.io/sdk/testsuite" + "go.temporal.io/sdk/worker" "go.temporal.io/sdk/workflow" ws "go.temporal.io/sdk/advanced/samples/go/workflowservice" @@ -50,12 +55,14 @@ type recordingDataConverter struct { converter.DataConverter serializationContext converter.SerializationContext recorded *[]recordedEncode + mu *sync.Mutex } func newRecordingDataConverter() recordingDataConverter { return recordingDataConverter{ DataConverter: converter.GetDefaultDataConverter(), recorded: &[]recordedEncode{}, + mu: &sync.Mutex{}, } } @@ -64,26 +71,91 @@ func (c recordingDataConverter) WithSerializationContext(ctx converter.Serializa DataConverter: converter.WithDataConverterSerializationContext(c.DataConverter, ctx), serializationContext: ctx, recorded: c.recorded, + mu: c.mu, } } func (c recordingDataConverter) ToPayload(value any) (*common.Payload, error) { + c.mu.Lock() *c.recorded = append(*c.recorded, recordedEncode{c.serializationContext, value}) + c.mu.Unlock() return c.DataConverter.ToPayload(value) } func (c recordingDataConverter) ToPayloads(values ...any) (*common.Payloads, error) { + c.mu.Lock() for _, value := range values { *c.recorded = append(*c.recorded, recordedEncode{c.serializationContext, value}) } + c.mu.Unlock() return c.DataConverter.ToPayloads(values...) } +func (c recordingDataConverter) snapshot() []recordedEncode { + c.mu.Lock() + defer c.mu.Unlock() + return append([]recordedEncode(nil), (*c.recorded)...) +} + +// Binding hides WithSerializationContext, so a bound converter cannot be scoped +// again. The SDK must select the operation's inner converter from the root. +type oneShotRecordingDataConverter struct { + recordingDataConverter +} + +func (c oneShotRecordingDataConverter) WithSerializationContext(ctx converter.SerializationContext) converter.DataConverter { + return struct{ converter.DataConverter }{c.recordingDataConverter.WithSerializationContext(ctx)} +} + +func nexusOperationSerializationContext() converter.NexusSerializationContext { + return converter.NexusSerializationContext{ + Endpoint: "__temporal_system", Service: workflowServiceName, + Operation: "SignalWithStartWorkflowExecution", + } +} + +// replaceNexusInputInterceptor lets the test replace a native input before the +// SDK selects its policy, without exposing the generated request type. +type replaceNexusInputInterceptor struct { + interceptor.WorkerInterceptorBase + replace func(any) any +} + +func (i *replaceNexusInputInterceptor) InterceptWorkflow(ctx workflow.Context, next interceptor.WorkflowInboundInterceptor) interceptor.WorkflowInboundInterceptor { + return &replaceNexusInputInbound{ + WorkflowInboundInterceptorBase: interceptor.WorkflowInboundInterceptorBase{Next: next}, + replace: i.replace, + } +} + +type replaceNexusInputInbound struct { + interceptor.WorkflowInboundInterceptorBase + replace func(any) any +} + +func (i *replaceNexusInputInbound) Init(next interceptor.WorkflowOutboundInterceptor) error { + return i.Next.Init(&replaceNexusInputOutbound{ + WorkflowOutboundInterceptorBase: interceptor.WorkflowOutboundInterceptorBase{Next: next}, + replace: i.replace, + }) +} + +type replaceNexusInputOutbound struct { + interceptor.WorkflowOutboundInterceptorBase + replace func(any) any +} + +func (i *replaceNexusInputOutbound) ExecuteNexusOperation(ctx workflow.Context, input interceptor.ExecuteNexusOperationInput) workflow.NexusOperationFuture { + input.Input = i.replace(input.Input) + return i.Next.ExecuteNexusOperation(ctx, input) +} + type WorkflowServiceIntegrationSuite struct { suite.Suite testsuite.WorkflowTestSuite - env *testsuite.TestWorkflowEnvironment - calls []*workflowservicepb.SignalWithStartWorkflowExecutionRequest + env *testsuite.TestWorkflowEnvironment + calls []*workflowservicepb.SignalWithStartWorkflowExecutionRequest + callsMu sync.Mutex } func (s *WorkflowServiceIntegrationSuite) SetupTest() { @@ -92,7 +164,9 @@ func (s *WorkflowServiceIntegrationSuite) SetupTest() { signalWithStart := nexus.NewSyncOperation("SignalWithStartWorkflowExecution", func(ctx context.Context, input *workflowservicepb.SignalWithStartWorkflowExecutionRequest, opts nexus.StartOperationOptions) (*workflowservicepb.SignalWithStartWorkflowExecutionResponse, error) { + s.callsMu.Lock() s.calls = append(s.calls, input) + s.callsMu.Unlock() return &workflowservicepb.SignalWithStartWorkflowExecutionResponse{}, nil }) @@ -242,29 +316,218 @@ func (s *WorkflowServiceIntegrationSuite) TestConversionFailureSurfacesOnTheFutu // TestModelIsConvertedInsideThePayloadConverter is the acceptance test for the // transfer-type port. The generated code no longer builds the proto itself; // the SDK's transfer-type machinery does it during payload conversion. The -// observable consequence is that the model's inner user payloads (Args, -// SignalArgs) are encoded by the data converter -- and under the serialization -// context -- that the SDK selected for the Nexus operation. +// inner user payloads use the target workflow context, while the outer proto +// uses the Nexus operation context. func (s *WorkflowServiceIntegrationSuite) TestModelIsConvertedInsideThePayloadConverter() { + s.checkSerializationContexts(false) +} + +func (s *WorkflowServiceIntegrationSuite) TestOneShotConverterIsRescopedFromRoot() { + s.checkSerializationContexts(true) +} + +func (s *WorkflowServiceIntegrationSuite) TestOperationRegistryUsesWireKeyAndNativeRequest() { + key := internal.NexusOperationKey{ + Service: workflowServiceName, Operation: "SignalWithStartWorkflowExecution", + } + original, ok := ws.NexusOperationRegistry[key] + s.Require().True(ok, "registry must use the wire service and operation names") + s.Require().NotNil(original.SerializationContext) + + var requests []any + s.env.SetWorkerOptions(worker.Options{Interceptors: []interceptor.WorkerInterceptor{ + &replaceNexusInputInterceptor{replace: func(input any) any { + requests = append(requests, input) + return input + }}, + }}) + + s.checkSerializationContexts(true) + + s.Require().Len(requests, 1) + s.Equal(converter.WorkflowSerializationContext{ + Namespace: "default-test-namespace", WorkflowID: "target-workflow-id", + }, original.SerializationContext(requests[0])) + // The native request type is intentionally unexported. Inspect its fields + // without converting it to a proto or adding a public model API for tests. + request := reflect.ValueOf(requests[0]) + s.Require().Equal(reflect.Struct, request.Kind(), "callback must receive a native request value") + s.Equal(reflect.TypeOf(ws.SignalWithStartWorkflowOptions{}).PkgPath(), request.Type().PkgPath()) + for name, want := range map[string]string{ + "namespace": "default-test-namespace", "ID": "target-workflow-id", + "TaskQueue": "my-task-queue", "Workflow": "ExampleWorkflow", "Signal": "wake-up", + } { + field := request.FieldByName(name) + s.Require().True(field.IsValid(), "native request field %s is missing", name) + s.Require().Equal(reflect.String, field.Kind(), name) + s.Equal(want, field.String(), name) + } + for name, want := range map[string][]any{ + "Args": {"workflow-input"}, "SignalArgs": {"signal-value"}, + } { + field := request.FieldByName(name) + s.Require().True(field.IsValid(), "native request field %s is missing", name) + s.Equal(want, field.Interface(), name) + } +} + +func (s *WorkflowServiceIntegrationSuite) TestRawExecuteOperationUsesRegisteredNativeRequest() { + key := internal.NexusOperationKey{ + Service: workflowServiceName, Operation: "SignalWithStartWorkflowExecution", + } + + // Capture the unexported native model without inventing a public constructor + // or substituting a protobuf that would bypass the native policy. + var nativeRequest any recorder := newRecordingDataConverter() - s.env.SetDataConverter(recorder) + s.env.SetDataConverter(oneShotRecordingDataConverter{recorder}) + endpoints := []string{"__temporal_system", "temporal-system", "ordinary-endpoint"} + s.env.SetWorkerOptions(worker.Options{Interceptors: []interceptor.WorkerInterceptor{ + &replaceNexusInputInterceptor{replace: func(input any) any { + // Preserve the unexported sourced fields while replacing the exported ID. + value := reflect.ValueOf(input) + replacement := reflect.New(value.Type()).Elem() + replacement.Set(value) + replacement.FieldByName("ID").SetString("intercepted-target") + nativeRequest = replacement.Interface() + return nativeRequest + }}, + }}) s.env.ExecuteWorkflow(func(ctx workflow.Context) error { - return ws.SignalWithStartWorkflow( + if err := ws.SignalWithStartWorkflow(ctx, + ws.SignalWithStartWorkflowOptions{ID: "raw-target"}, + "wake-up", "raw-signal", "ExampleWorkflow", "raw-input", + ).Get(ctx, nil); err != nil { + return err + } + if nativeRequest == nil { + return errors.New("workflow interceptor did not capture the native request") + } + for _, endpoint := range endpoints { + // The public constructor rejects the reserved canonical endpoint. + var client workflow.NexusClient + if endpoint == "__temporal_system" { + client = internal.NewSystemNexusClient(workflowServiceName) + } else { + client = workflow.NewNexusClient(endpoint, workflowServiceName) + } + future := client.ExecuteOperation(ctx, key.Operation, nativeRequest, workflow.NexusOperationOptions{}) + var response ws.SignalWithStartWorkflowResponse + if err := future.Get(ctx, &response); err != nil { + return err + } + } + return nil + }) + + s.Require().NoError(s.env.GetWorkflowError()) + s.Require().Len(s.calls, 4) + for _, call := range s.calls { + s.Equal("intercepted-target", call.GetWorkflowId()) + s.Equal("default-test-namespace", call.GetNamespace()) + } + target := converter.WorkflowSerializationContext{ + Namespace: "default-test-namespace", WorkflowID: "intercepted-target", + } + innerEncodes := map[converter.SerializationContext]int{} + envelopes := map[string]int{} + for _, encoded := range recorder.snapshot() { + switch value := encoded.Value.(type) { + case string: + if value == "raw-input" { + innerEncodes[encoded.Context]++ + } + case *workflowservicepb.SignalWithStartWorkflowExecutionRequest: + if sc, ok := encoded.Context.(converter.NexusSerializationContext); ok { + s.Equal(workflowServiceName, sc.Service) + s.Equal(key.Operation, sc.Operation) + envelopes[sc.Endpoint]++ + } + } + } + s.Equal(map[converter.SerializationContext]int{target: 4}, innerEncodes) + s.Equal(map[string]int{ + "__temporal_system": 2, "temporal-system": 1, "ordinary-endpoint": 1, + }, envelopes, "outer envelopes remain Nexus-scoped on every endpoint") +} + +func (s *WorkflowServiceIntegrationSuite) TestOperationRegistryCopiesEntries() { + key := internal.NexusOperationKey{ + Service: workflowServiceName, Operation: "SignalWithStartWorkflowExecution", + } + original, ok := ws.NexusOperationRegistry[key] + s.Require().True(ok) + s.T().Cleanup(func() { ws.NexusOperationRegistry[key] = original }) + entry := original + calls := 0 + entry.SerializationContext = func(any) converter.SerializationContext { + calls++ + return converter.WorkflowSerializationContext{} + } + ws.NexusOperationRegistry[key] = entry + + s.checkSerializationContexts(true) + s.Zero(calls, "changing the generated map must not replace an SDK-owned entry") +} + +func (s *WorkflowServiceIntegrationSuite) checkSerializationContexts(oneShot bool) { + s.checkSerializationContextsWithTarget(oneShot, converter.WorkflowSerializationContext{ + Namespace: "default-test-namespace", WorkflowID: "target-workflow-id", + }) +} + +func (s *WorkflowServiceIntegrationSuite) checkSerializationContextsWithTarget(oneShot bool, targetContext converter.WorkflowSerializationContext) { + recorder := newRecordingDataConverter() + if oneShot { + root := oneShotRecordingDataConverter{recorder} + bound := root.WithSerializationContext(converter.WorkflowSerializationContext{}) + _, canRescope := bound.(converter.DataConverterWithSerializationContext) + s.Require().False(canRescope, "test converter must not allow rebinding") + s.env.SetDataConverter(root) + } else { + s.env.SetDataConverter(recorder) + } + + s.env.ExecuteWorkflow(func(ctx workflow.Context) error { + err := ws.SignalWithStartWorkflow( ctx, - ws.SignalWithStartWorkflowOptions{ID: "target-workflow-id", TaskQueue: "my-task-queue"}, + ws.SignalWithStartWorkflowOptions{ + ID: "target-workflow-id", TaskQueue: "my-task-queue", + Memo: map[string]any{"memo-key": "target-memo"}, + UserMetadata: ws.UserMetadata{StaticSummary: "target-summary", StaticDetails: "target-details"}, + }, "wake-up", "signal-value", "ExampleWorkflow", "workflow-input", ).Get(ctx, nil) + if err != nil { + return err + } + _, err = internal.GetDataConverterFromWorkflowContext(ctx).ToPayload("caller-after") + return err }) s.True(s.env.IsWorkflowCompleted()) s.NoError(s.env.GetWorkflowError()) s.Require().Len(s.calls, 1) + s.Equal("default-test-namespace", s.calls[0].GetNamespace()) + s.Equal("target-workflow-id", s.calls[0].GetWorkflowId()) + // Headers are API-omitted and have no source in this sample. + s.Nil(s.calls[0].GetHeader()) + for value, payload := range map[string]*common.Payload{ + "target-memo": s.calls[0].GetMemo().GetFields()["memo-key"], + "target-summary": s.calls[0].GetUserMetadata().GetSummary(), + "target-details": s.calls[0].GetUserMetadata().GetDetails(), + } { + s.Require().NotNil(payload, value) + var decoded string + s.Require().NoError(converter.GetDefaultDataConverter().FromPayload(payload, &decoded)) + s.Equal(value, decoded) + } - recorded := *recorder.recorded + recorded := recorder.snapshot() indexOf := func(match func(recordedEncode) bool) int { for i, entry := range recorded { if match(entry) { @@ -299,22 +562,112 @@ func (s *WorkflowServiceIntegrationSuite) TestModelIsConvertedInsideThePayloadCo // The transfer value -- the proto envelope -- is encoded under the Nexus // operation's serialization context. s.Equal( - converter.NexusSerializationContext{ - Endpoint: "__temporal_system", - Service: workflowServiceName, - Operation: "SignalWithStartWorkflowExecution", - }, + nexusOperationSerializationContext(), recorded[request].Context, ) - // The inner user payloads are encoded under the workflow serialization - // context reachable from the workflow.Context handed to the transfer - // converter. That context is the *calling* workflow's today; redirecting it - // to the target workflow is what @nexus.serialization-context will do. - callerContext := converter.WorkflowSerializationContext{ - Namespace: "default-test-namespace", - WorkflowID: "default-test-workflow-id", + s.Equal(targetContext, recorded[workflowArg].Context) + s.Equal(targetContext, recorded[signalArg].Context) + // Memo uses the user converter with the current SDK memo flag enabled; + // this sample's metadata helper also uses the workflow-context converter. + for _, value := range []string{"target-memo", "target-summary", "target-details"} { + i := stringAt(value) + s.Require().NotEqual(-1, i, "%s was not encoded by the user converter", value) + s.Less(i, request) + s.Equal(targetContext, recorded[i].Context, value) + } + caller := stringAt("caller-after") + s.Require().NotEqual(-1, caller) + s.Equal(converter.WorkflowSerializationContext{ + Namespace: "default-test-namespace", WorkflowID: "default-test-workflow-id", + }, recorded[caller].Context, "operation must not change the caller's converter") +} + +func (s *WorkflowServiceIntegrationSuite) TestConcurrentOperationsKeepTargetContextsIsolated() { + recorder := newRecordingDataConverter() + s.env.SetDataConverter(oneShotRecordingDataConverter{recorder}) + targets := []string{"target-a", "target-b"} + + s.env.ExecuteWorkflow(func(ctx workflow.Context) error { + futures := make([]workflow.Future, 0, len(targets)) + // Schedule both operations before awaiting either, then await in reverse + // order to catch accidental reuse of the last operation's context. + for _, target := range targets { + futures = append(futures, ws.SignalWithStartWorkflow(ctx, + ws.SignalWithStartWorkflowOptions{ + ID: target, + Memo: map[string]any{"memo-key": target + "-memo"}, + UserMetadata: ws.UserMetadata{ + StaticSummary: target + "-summary", StaticDetails: target + "-details", + }, + }, + "wake-up", target+"-signal", "ExampleWorkflow", target+"-input", + )) + } + for i := len(futures) - 1; i >= 0; i-- { + var result ws.SignalWithStartWorkflowResponse + if err := futures[i].Get(ctx, &result); err != nil { + return err + } + } + _, err := internal.GetDataConverterFromWorkflowContext(ctx).ToPayload("caller-after") + return err + }) + + s.True(s.env.IsWorkflowCompleted()) + s.Require().NoError(s.env.GetWorkflowError()) + s.Require().Len(s.calls, 2) + var calledTargets []string + for _, call := range s.calls { + calledTargets = append(calledTargets, call.GetWorkflowId()) + s.Equal("default-test-namespace", call.GetNamespace()) + var input, signal string + s.Require().NoError(converter.GetDefaultDataConverter().FromPayloads(call.GetInput(), &input)) + s.Require().NoError(converter.GetDefaultDataConverter().FromPayloads(call.GetSignalInput(), &signal)) + s.Equal(call.GetWorkflowId()+"-input", input) + s.Equal(call.GetWorkflowId()+"-signal", signal) + } + s.ElementsMatch(targets, calledTargets) + + want := map[string]converter.SerializationContext{ + "caller-after": converter.WorkflowSerializationContext{ + Namespace: "default-test-namespace", WorkflowID: "default-test-workflow-id", + }, + } + for _, target := range targets { + for _, suffix := range []string{"-input", "-signal", "-memo", "-summary", "-details"} { + want[target+suffix] = converter.WorkflowSerializationContext{ + Namespace: "default-test-namespace", WorkflowID: target, + } + } + } + seen := make(map[string]int) + envelopes := make(map[string]int) + for _, entry := range recorder.snapshot() { + switch value := entry.Value.(type) { + case string: + if expected, ok := want[value]; ok { + s.Equal(expected, entry.Context, value) + seen[value]++ + } + case *workflowservicepb.SignalWithStartWorkflowExecutionRequest: + target := value.GetWorkflowId() + s.Contains(targets, target) + if _, ok := entry.Context.(converter.NexusSerializationContext); ok { + s.Equal(nexusOperationSerializationContext(), entry.Context, target) + envelopes[target]++ + } else { + // testNexusHandler.StartOperation rebuilds the decoded input as + // a LazyValue using the test environment's caller-scoped converter. + // This is separate from the real outer operation encoding above. + s.Equal(want["caller-after"], entry.Context, "test harness re-encode for %s", target) + } + } + } + for value := range want { + s.Equal(1, seen[value], "%s must be encoded once in its own context", value) + } + for _, target := range targets { + s.Equal(1, envelopes[target], "outer operation envelope for %s must be encoded once in the Nexus context", target) } - s.Equal(callerContext, recorded[workflowArg].Context) - s.Equal(callerContext, recorded[signalArg].Context) } diff --git a/advanced/samples/go/workflowservice/support.go b/advanced/samples/go/workflowservice/support.go index 2661fbd8..da71b352 100644 --- a/advanced/samples/go/workflowservice/support.go +++ b/advanced/samples/go/workflowservice/support.go @@ -2,8 +2,6 @@ package workflowservice import ( "fmt" - "time" - common "go.temporal.io/api/common/v1" enums "go.temporal.io/api/enums/v1" failurepb "go.temporal.io/api/failure/v1" @@ -16,6 +14,7 @@ import ( "go.temporal.io/sdk/worker" "go.temporal.io/sdk/workflow" "google.golang.org/protobuf/types/known/durationpb" + "time" ) // --- Duration (google.protobuf.Duration) --- @@ -322,3 +321,10 @@ func versioningOverrideFromProto(_ workflow.Context, versioningOverride *workflo } return &value, nil } + +func signalWithStartWorkflowSerializationContext(request signalWithStartWorkflowRequest) converter.WorkflowSerializationContext { + return converter.WorkflowSerializationContext{ + Namespace: request.namespace, + WorkflowID: request.ID, + } +} diff --git a/advanced/samples/go/workflowservice/workflowservice.go b/advanced/samples/go/workflowservice/workflowservice.go index 5a5e31d0..5a98a07b 100644 --- a/advanced/samples/go/workflowservice/workflowservice.go +++ b/advanced/samples/go/workflowservice/workflowservice.go @@ -667,3 +667,20 @@ func SignalWithStartWorkflowTyped[WorkflowArg any, WorkflowResult any]( UserMetadata: &opts.UserMetadata, }) } + +// --- Registry --- + +var NexusOperationRegistry = map[internal.NexusOperationKey]internal.NexusOperationRegistryEntry{ + { + Service: "temporal.api.workflowservice.v1.WorkflowService", + Operation: "SignalWithStartWorkflowExecution", + }: { + SerializationContext: func(request any) converter.SerializationContext { + return signalWithStartWorkflowSerializationContext(request.(signalWithStartWorkflowRequest)) + }, + }, +} + +func init() { + internal.RegisterNexusOperationRegistry(NexusOperationRegistry) +} diff --git a/advanced/samples/inputs/go/WorkflowServiceSupport.go b/advanced/samples/inputs/go/WorkflowServiceSupport.go new file mode 100644 index 00000000..9e7d1490 --- /dev/null +++ b/advanced/samples/inputs/go/WorkflowServiceSupport.go @@ -0,0 +1,10 @@ +package workflowservice + +import "go.temporal.io/sdk/converter" + +func signalWithStartWorkflowSerializationContext(request signalWithStartWorkflowRequest) converter.WorkflowSerializationContext { + return converter.WorkflowSerializationContext{ + Namespace: request.namespace, + WorkflowID: request.ID, + } +} diff --git a/advanced/samples/inputs/workflow-service.wit b/advanced/samples/inputs/workflow-service.wit index 7e47b377..60a296a3 100644 --- a/advanced/samples/inputs/workflow-service.wit +++ b/advanced/samples/inputs/workflow-service.wit @@ -1,4 +1,4 @@ -/// @nexus.support dotnet="dotnet/WorkflowServiceSupport.cs" +/// @nexus.support dotnet="dotnet/WorkflowServiceSupport.cs" go="go/WorkflowServiceSupport.go" /// @nexus.support-namespace dotnet="Nexgen.Support" package temporal:nexus@1.0.0; @@ -138,7 +138,7 @@ interface workflow-service { /// dotnet-type="Temporalio.Workflows.ExternalWorkflowHandle" /// dotnet="Temporalio.Workflows.Workflow.GetExternalWorkflowHandle(request.Id, result.RunId)" /// @nexus.operation name="SignalWithStartWorkflowExecution" - /// @nexus.serialization-context python="signal_with_start_workflow_serialization_context" typescript="signalWithStartWorkflowSerializationContext" dotnet="WorkflowServiceSerializationContexts.SignalWithStartWorkflow" + /// @nexus.serialization-context python="signal_with_start_workflow_serialization_context" typescript="signalWithStartWorkflowSerializationContext" dotnet="WorkflowServiceSerializationContexts.SignalWithStartWorkflow" go="signalWithStartWorkflowSerializationContext" /// @nexus.experimental signal-with-start-workflow: func( request: signal-with-start-workflow-request, diff --git a/src/generator/go.rs b/src/generator/go.rs index f10b599e..0b11ef71 100644 --- a/src/generator/go.rs +++ b/src/generator/go.rs @@ -503,6 +503,7 @@ fn planned_value_type(value_type: &PlannedType, spec: &PlannedSpec) -> PlannedVa pub(in crate::generator) struct GoPackageContext { package_name: String, import_path: Option, + pub(in crate::generator) has_serialization_context: bool, } impl GoPackageContext { @@ -539,6 +540,16 @@ impl GoPackageContext { Ok(Self { package_name, import_path, + has_serialization_context: api_plan + .services + .iter() + .flat_map(|service| &service.operations) + .any(|operation| { + operation + .serialization_context + .for_language(Language::Go) + .is_some() + }), }) } @@ -624,6 +635,13 @@ impl GoPackageContext { self.qualified_expr("errors", "errors.New") } + pub(in crate::generator) fn serialization_context_type(&self) -> String { + self.qualified_expr( + "go.temporal.io/sdk/converter", + "converter.SerializationContext", + ) + } + fn qualified_expr(&self, import_path: &str, code_expr: &str) -> String { if self.is_self_import(import_path) { self.unqualify_self_expr(code_expr) @@ -853,6 +871,12 @@ struct GoModelFragments { body: String, } +#[derive(Debug, Default)] +struct GoOperationRegistry { + imports: Vec<(String, String)>, + body: String, +} + enum GoExternalModels { Json(json::ModelBackend), Proto(proto::ModelBackend), @@ -1073,6 +1097,14 @@ impl<'a> ApiPlanner<'a> { primary_origin: GeneratedFileOrigin, ) -> Result { let mut services = Vec::new(); + if self.package.has_serialization_context + && self.external_models.renders_operation_references() + { + return Err(Error::UnsupportedGoProtoConversion { + context: "Go serialization context".to_string(), + reason: "serialization-context requires proto-backed operation wrappers, not operation references".to_string(), + }); + } if !self.external_models.renders_operation_references() { for service in &self.api_plan.services { let mut operations = Vec::new(); @@ -1127,6 +1159,14 @@ impl<'a> ApiPlanner<'a> { // so the backend renders final Go names directly. self.populate_model_wire_conversions()?; self.populate_operation_bindings(&mut services)?; + for operation in services.iter().flat_map(|service| &service.operations) { + if operation.serialization_context_expr.is_some() && operation.wire_binding.is_none() { + return Err(Error::UnsupportedGoProtoConversion { + context: format!("operation `{}`", operation.name), + reason: "serialization-context requires a proto-backed operation".to_string(), + }); + } + } let needs_function_name_inlining = services.iter().any(service_uses_function_name_inlining); @@ -1204,7 +1244,39 @@ impl<'a> ApiPlanner<'a> { self.imports .insert("go.temporal.io/sdk/workflow".to_string()); } - let external_imports = self.external_models.imports(); + let mut external_imports = self.external_models.imports(); + let registry = if self.package.has_serialization_context { + render_operation_registry( + &self.package, + &services, + &self.imports, + &external_imports, + support_fragments, + )? + } else { + GoOperationRegistry::default() + }; + for (path, alias) in ®istry.imports { + if let Some((_, previous_alias)) = external_imports + .iter() + .find(|(existing, _)| existing == path) + { + if previous_alias != alias { + return Err(Error::InvalidWitDirective { + path: PathBuf::from(""), + context: "Go operation registry".to_string(), + directive: "@nexus.serialization-context".to_string(), + reason: format!( + "import `{path}` uses both `{previous_alias}` and `{alias}` in the generated API file" + ), + }); + } + } else if alias == path.rsplit('/').next().unwrap_or(path) { + self.imports.insert(path.clone()); + } else { + external_imports.push((path.clone(), alias.clone())); + } + } let output = render_file( &self.package, &self.imports, @@ -1221,6 +1293,7 @@ impl<'a> ApiPlanner<'a> { &self.external_models, self.api_plan, &visibility, + ®istry.body, ); let mut files = GeneratedFileMap::default(); @@ -1385,6 +1458,10 @@ impl<'a> ApiPlanner<'a> { .collect() }); + let serialization_context_expr = operation.serialization_context.for_language(Language::Go); + if let Some(helper) = serialization_context_expr { + validate_go_serialization_context_helper(helper, &operation.name)?; + } Ok(RenderedOperation { name: operation.name.as_str(), wire_name: operation.wire_name.as_str(), @@ -1395,6 +1472,7 @@ impl<'a> ApiPlanner<'a> { model_type_parameters, output_type, raw_output_type, + serialization_context_expr, output_transform_expr: go_output_transform.map(|(_, expr)| expr), output_transform_type: go_output_transform .map(|(type_name, _)| self.package.go_type_expr(type_name)), @@ -1821,6 +1899,172 @@ fn go_api_file_name(api_plan: &PlannedSpec) -> PathBuf { PathBuf::from(format!("{stem}.go")) } +fn validate_go_serialization_context_helper(helper: &str, operation: &str) -> Result<()> { + let parts: Vec<_> = helper.split('.').collect(); + if parts.len() > 2 + || parts.iter().any(|part| { + part.is_empty() + || *part == "_" + || is_go_keyword(part) + || !part + .as_bytes() + .first() + .is_some_and(|byte| is_go_ident_start(*byte)) + || !part.bytes().all(is_go_ident_continue) + }) + { + return Err(Error::InvalidWitDirective { + path: PathBuf::from(""), + context: format!("operation `{operation}`"), + directive: "@nexus.serialization-context".to_string(), + reason: format!("Go support helper `{helper}` must be an identifier or pkg.Func"), + }); + } + Ok(()) +} + +fn render_operation_registry( + package: &GoPackageContext, + services: &[RenderedService<'_>], + model_imports: &BTreeSet, + external_imports: &[(String, String)], + support_fragments: &[SupportFragmentSpec], +) -> Result { + let services: Vec<_> = services + .iter() + .filter(|service| service.is_system_endpoint()) + .collect(); + let has_helpers = services + .iter() + .flat_map(|service| &service.operations) + .any(|operation| operation.serialization_context_expr.is_some()); + let helper_packages: BTreeSet<_> = services + .iter() + .flat_map(|service| &service.operations) + .filter_map(|operation| { + operation + .serialization_context_expr? + .split_once('.') + .map(|(name, _)| name) + }) + .collect(); + let mut candidates = external_imports.to_vec(); + candidates.extend(model_imports.iter().map(|path| { + ( + path.clone(), + path.rsplit('/').next().unwrap_or(path).to_string(), + ) + })); + for fragment in support_fragments { + for import in parse_support_fragment(&fragment.contents).imports { + let (alias, path) = import.split_once('"').unwrap_or(("", "")); + let path = path.split('"').next().unwrap_or_default(); + let alias = if alias.trim().is_empty() { + path.rsplit('/').next().unwrap_or_default() + } else { + alias.trim() + }; + candidates.push((path.to_string(), alias.to_string())); + } + } + let mut imports = BTreeMap::new(); + if has_helpers && !package.is_self_import("go.temporal.io/sdk/converter") { + imports.insert( + "converter".to_string(), + "go.temporal.io/sdk/converter".to_string(), + ); + } + if !package.is_self_import("go.temporal.io/sdk/internal") { + imports.insert( + "internal".to_string(), + "go.temporal.io/sdk/internal".to_string(), + ); + } + for (path, alias) in candidates { + let used_by_input_type = + services + .iter() + .flat_map(|service| &service.operations) + .any(|operation| { + operation.serialization_context_expr.is_some() + && operation.input_type.contains(&format!("{alias}.")) + }); + if (!helper_packages.contains(alias.as_str()) && !used_by_input_type) + || package.is_self_import(&path) + { + continue; + } + if let Some(previous) = imports.insert(alias.clone(), path.clone()) { + if previous != path { + return Err(Error::InvalidWitDirective { + path: PathBuf::from(""), + context: "Go operation registry".to_string(), + directive: "@nexus.serialization-context".to_string(), + reason: format!( + "helper qualifier `{alias}` refers to both `{previous}` and `{path}`" + ), + }); + } + } + } + let mut output = String::from("\n// --- Registry ---\n"); + let sc = package.serialization_context_type(); + let key = package.qualified_expr("go.temporal.io/sdk/internal", "internal.NexusOperationKey"); + let info = package.qualified_expr( + "go.temporal.io/sdk/internal", + "internal.NexusOperationRegistryEntry", + ); + output.push_str(&format!( + "\nvar NexusOperationRegistry = map[{key}]{info}{{\n" + )); + let mut keys = BTreeSet::new(); + for service in services { + for operation in &service.operations { + let Some(helper) = operation.serialization_context_expr else { + continue; + }; + if !keys.insert((service.wire_name, operation.wire_name)) { + return Err(Error::InvalidWitDirective { + path: PathBuf::from(""), + context: "Go operation registry".to_string(), + directive: "@nexus.serialization-context".to_string(), + reason: format!( + "duplicate registry key: service `{}`, operation `{}`", + service.wire_name, operation.wire_name + ), + }); + } + let service_name = go_string_literal(service.wire_name); + let operation_name = go_string_literal(operation.wire_name); + output.push_str(&format!( + "\t{{\n\t\tService: {service_name},\n\t\tOperation: {operation_name},\n\t}}: {{\n" + )); + let input_type = &operation.input_type; + output.push_str(&format!( + "\t\tSerializationContext: func(request any) {sc} {{\n\t\t\treturn {helper}(request.({input_type}))\n\t\t}},\n" + )); + output.push_str("\t},\n"); + } + } + output.push_str("}\n"); + if !keys.is_empty() { + let register = package.qualified_expr( + "go.temporal.io/sdk/internal", + "internal.RegisterNexusOperationRegistry", + ); + output.push_str(&format!( + "\nfunc init() {{\n\t{register}(NexusOperationRegistry)\n}}\n" + )); + } + Ok(GoOperationRegistry { + imports: imports + .into_iter() + .map(|(alias, path)| (path, alias)) + .collect(), + body: output, + }) +} + fn render_support_file(fragments: &[SupportFragmentSpec], package_name: &str) -> String { if let [fragment] = fragments { return rewrite_support_package(&fragment.contents, package_name); @@ -1973,6 +2217,15 @@ pub(in crate::generator) struct RenderedService<'a> { pub(in crate::generator) resources: Vec, } +impl RenderedService<'_> { + pub(in crate::generator) fn is_system_endpoint(&self) -> bool { + matches!( + self.endpoint.as_deref(), + Some("__temporal_system" | "temporal-system") + ) + } +} + /// A resolved Nexus operation with enough information to render the operation /// constant and an unexported caller function. #[derive(Debug)] @@ -2005,6 +2258,8 @@ pub(in crate::generator) struct RenderedOperation<'a> { /// The expression is rendered with `request` and raw `result` in scope and /// must evaluate to `(output_transform_type, error)`. pub(in crate::generator) output_transform_expr: Option<&'a str>, + /// Helper selecting the context for user payloads nested in this operation. + pub(in crate::generator) serialization_context_expr: Option<&'a str>, /// Go return type for an `@nexus.output-transform`, with any embedded /// import path stripped from the WIT annotation. pub(in crate::generator) output_transform_type: Option, @@ -3188,6 +3443,7 @@ fn render_file( external_models: &GoExternalModels, api_plan: &PlannedSpec, visibility: &GoVisibility, + registry: &str, ) -> String { let mut output = String::new(); output.push_str(GENERATED_HEADER); @@ -3244,6 +3500,7 @@ fn render_file( output.push_str(")\n"); } + let body_start = output.len(); if let Some(operation_references) = operation_references { if !operation_references.is_empty() { output.push('\n'); @@ -3252,6 +3509,7 @@ fn render_file( if !external_model_body.is_empty() { output.push_str(external_model_body); } + output.push_str(registry); return output; } if !external_model_body.is_empty() @@ -3263,6 +3521,7 @@ fn render_file( { output.push('\n'); output.push_str(external_model_body); + output.push_str(registry); return output; } @@ -3446,6 +3705,23 @@ fn render_file( } } + output.push_str(registry); + + // Imports collected while planning can become unused when the SDK owns + // input conversion instead of the generated wrapper. + let unused_external_imports: Vec<_> = external_imports + .iter() + .filter(|(_, alias)| !output[body_start..].contains(&format!("{alias}."))) + .collect(); + for (path, alias) in unused_external_imports { + let default_alias = path.rsplit('/').next().unwrap_or(path); + let import = if alias == default_alias { + format!("\t\"{path}\"\n") + } else { + format!("\t{alias} \"{path}\"\n") + }; + output = output.replacen(&import, "", 1); + } output } @@ -5147,6 +5423,178 @@ fn render_forwarding_wrapper( #[cfg(test)] mod tests { + use std::collections::BTreeMap; + use std::path::PathBuf; + + use prost::Message; + use prost_types::{ + DescriptorProto, FieldDescriptorProto, FileDescriptorProto, FileDescriptorSet, FileOptions, + field_descriptor_proto::{Label, Type}, + }; + + use crate::error::{Error, Result}; + use crate::{GenerateRequest, generate_to_file}; + + fn generate_context_operation( + endpoint: &str, + native_type: &str, + annotated: bool, + owned_request: bool, + ) -> Result> { + let temp = tempfile::tempdir().unwrap(); + let input = temp.path().join("context.wit"); + let annotation = if annotated { + "/// @nexus.serialization-context go=\"requestContext\"" + } else { + "" + }; + let declaration = if owned_request { + r#"/// @nexus.proto "test.Request" + record request { retry-policy: option }"# + } else { + "type request = retry-policy;" + }; + std::fs::write( + &input, + format!( + r#"package test:context@1.0.0; + world system {{ export sample-service; }} + /// @nexus.endpoint "{endpoint}" + interface sample-service {{ + type placeholder = string; + /// @nexus.proto "test.RetryPolicy" + /// @nexus.type go="go.temporal.io/sdk/{native_type}" + type retry-policy = placeholder; + {declaration} + {annotation} + send: func(request: request) -> retry-policy; + }}"# + ), + ) + .unwrap(); + let descriptor = temp.path().join("wire.bin"); + let descriptors = FileDescriptorSet { + file: vec![FileDescriptorProto { + name: Some("wire.proto".to_string()), + package: Some("test".to_string()), + syntax: Some("proto3".to_string()), + options: Some(FileOptions { + go_package: Some("example.com/wire;wire".to_string()), + ..Default::default() + }), + message_type: vec![ + DescriptorProto { + name: Some("RetryPolicy".to_string()), + ..Default::default() + }, + DescriptorProto { + name: Some("Request".to_string()), + field: vec![FieldDescriptorProto { + name: Some("retry_policy".to_string()), + number: Some(1), + label: Some(Label::Optional as i32), + r#type: Some(Type::Message as i32), + type_name: Some(".test.RetryPolicy".to_string()), + ..Default::default() + }], + ..Default::default() + }, + ], + ..Default::default() + }], + }; + std::fs::write(&descriptor, descriptors.encode_to_vec()).unwrap(); + let output = temp.path().join("generated"); + generate_to_file(&GenerateRequest { + config: crate::nexgen_config::NexgenConfig { + mode: crate::generator::GenerationMode::NativeApi, + ..Default::default() + }, + language: crate::language::Language::Go, + input_paths: vec![input], + support_paths: Vec::new(), + descriptor_paths: vec![descriptor], + output_path: output.clone(), + format: false, + java_package_name: None, + ts_date_time_types: Default::default(), + })?; + Ok(std::fs::read_dir(output) + .unwrap() + .map(|entry| { + let entry = entry.unwrap(); + ( + PathBuf::from(entry.file_name()), + std::fs::read_to_string(entry.path()).unwrap(), + ) + }) + .collect()) + } + + #[test] + fn serialization_context_rejects_external_operation_inputs() { + for endpoint in ["__temporal_system", "temporal-system"] { + for native_type in ["temporal.RetryPolicy", "*temporal.RetryPolicy"] { + let error = + generate_context_operation(endpoint, native_type, true, false).unwrap_err(); + let Error::UnsupportedGoProtoConversion { context, reason } = error else { + panic!("unexpected error: {error}"); + }; + assert!(context.contains("Send"), "{context}"); + assert!(context.contains(native_type), "{context}"); + assert!(reason.contains("TransferTypeConverter"), "{reason}"); + assert!(reason.contains("locally owned request model"), "{reason}"); + } + } + } + + #[test] + fn external_inputs_keep_eager_conversion_without_system_policy() { + for (endpoint, annotated) in [ + ("sample-service", true), + ("temporal-system", false), + ("__temporal_system", false), + ] { + let files = + generate_context_operation(endpoint, "temporal.RetryPolicy", annotated, false) + .unwrap(); + let source = &files[&PathBuf::from("sampleservice.go")]; + assert!( + source.contains("requestProto, err := retryPolicyToProto(ctx, &request)"), + "{source}" + ); + assert!( + source.contains("c.ExecuteOperation(ctx, \"Send\", requestProto,"), + "{source}" + ); + assert!( + source.contains("retryPolicyFromProto(ctx, &result)"), + "{source}" + ); + } + } + + #[test] + fn system_policy_supports_generated_request_with_nested_external_type() { + let files = + generate_context_operation("temporal-system", "temporal.RetryPolicy", true, true) + .unwrap(); + let source = &files[&PathBuf::from("sampleservice.go")]; + assert!( + source.contains("c.ExecuteOperation(ctx, \"Send\", request,"), + "{source}" + ); + assert!(source.contains("TransferTypeConverter()"), "{source}"); + assert!( + source.contains("retryPolicyToProto(ctx, m.RetryPolicy)"), + "{source}" + ); + assert!( + source.contains("retryPolicyFromProto(ctx, &result)"), + "{source}" + ); + } + #[test] fn recognizes_builtin_nilable_types() { for type_expr in ["any", "error", "[]byte", "map[string]any"] { diff --git a/src/generator/proto/go.rs b/src/generator/proto/go.rs index 884a5814..21f4f753 100644 --- a/src/generator/proto/go.rs +++ b/src/generator/proto/go.rs @@ -642,6 +642,7 @@ pub(in crate::generator) struct OperationBinding { /// converter -- the SDK then performs the conversion inside the payload /// converter, and the model is passed to `ExecuteOperation` as-is. input_to_proto: Option, + /// Assignments that populate the request model's `@nexus.source` fields /// from their source expressions, emitted at the top of the operation /// function where `ctx` is in scope. @@ -752,6 +753,7 @@ impl ModelBackend { services: &mut [RenderedService<'_>], ) -> Result<()> { for (service, planned_service) in services.iter_mut().zip(api_plan.services.iter()) { + let is_system_endpoint = service.is_system_endpoint(); for (rendered_op, planned_op) in service .operations .iter_mut() @@ -765,19 +767,30 @@ impl ModelBackend { continue; }; let output = operation_output(planned_op, api_plan); - let Some((_input_wire_type, input_conv)) = + let Some((_, input_conv)) = operation_message_binding(&input, &planned_op.name, "input", self)? else { continue; }; + // Generated models carry a transfer-type converter, so the SDK // converts them to proto inside the payload converter. Anything // else (hand-written override converters over types nexgen does - // not own) still converts eagerly here. + // not own) uses an eager converter unless policy selection needs + // the native request in the SDK. let input_to_proto = if input_conv.kind == GoConversionKind::ModelConverter { self.mark_transfer_model_checked(api_plan, &input, &planned_op.name, "input")?; None } else { + if is_system_endpoint && rendered_op.serialization_context_expr.is_some() { + return Err(Error::UnsupportedGoProtoConversion { + context: format!( + "operation `{}` input `{}`", + planned_op.name, rendered_op.input_type + ), + reason: "cannot attach a TransferTypeConverter to an external input type for a system operation with @nexus.serialization-context; use a generated, locally owned request model instead".to_string(), + }); + } let input_arg = match input_conv.kind { GoConversionKind::OverrideConverter => "&request".to_string(), _ => "request".to_string(), @@ -893,6 +906,7 @@ impl ModelBackend { rendered_op.wire_binding = Some(OperationBinding { input_to_proto, + input_sourced_assignments, output_proto_type, output_model_type, @@ -953,8 +967,8 @@ fn is_proto_generic_carrier(kind: &PlannedType) -> bool { ) } -/// Renders an operation function that serializes its request to proto before -/// the SDK call and deserializes the proto response afterwards. +/// Renders an operation function using transfer converters where available, +/// with eager conversion for externally owned models and transformed results. pub(in crate::generator) fn render_operation_function_proto( output: &mut String, service: &crate::generator::go::RenderedService<'_>, @@ -979,6 +993,7 @@ pub(in crate::generator) fn render_operation_function_proto( output.push_str(assignment); output.push('\n'); } + let input_arg = match binding.input_to_proto.as_deref() { Some(input_to_proto) => { output.push_str("\trequestProto, err := "); @@ -993,8 +1008,6 @@ pub(in crate::generator) fn render_operation_function_proto( output.push_str("\t}\n"); "requestProto" } - // The request model implements `workflow.ValueWithTransferTypeConverter`, - // so the SDK converts it to proto while encoding the payload. None => "request", }; let endpoint = service @@ -1736,9 +1749,7 @@ fn render_model_transfer_type_converter( output.push_str(&format!( "func ({model_ident}) TransferTypeConverter() ({converter_type}, error) {{\n" )); - output.push_str(&format!( - "\treturn {new_converter}(\n" - )); + output.push_str(&format!("\treturn {new_converter}(\n")); output.push_str(&format!( "\t\tfunc({context_context}, *{model_ident}) (*{proto_value_type}, error) {{\n\t\t\treturn nil, {errors_new}(\"nexgen: transfer type converter outside a workflow\")\n\t\t}},\n" )); diff --git a/tests/add_rpc.rs b/tests/add_rpc.rs index 061ce5b4..430e4622 100644 --- a/tests/add_rpc.rs +++ b/tests/add_rpc.rs @@ -45,14 +45,19 @@ fn write_temp_wit(name: &str, contents: &str) -> PathBuf { fs::create_dir_all(&temp_dir).unwrap(); let path = temp_dir.join("input.wit"); fs::write(&path, contents).unwrap(); - if contents.contains("dotnet/WorkflowServiceSupport.cs") { - let dotnet_dir = temp_dir.join("dotnet"); - fs::create_dir_all(&dotnet_dir).unwrap(); - fs::copy( - project_root().join("advanced/samples/inputs/dotnet/WorkflowServiceSupport.cs"), - dotnet_dir.join("WorkflowServiceSupport.cs"), - ) - .unwrap(); + for support in [ + "dotnet/WorkflowServiceSupport.cs", + "go/WorkflowServiceSupport.go", + ] { + if contents.contains(support) { + let destination = temp_dir.join(support); + fs::create_dir_all(destination.parent().unwrap()).unwrap(); + fs::copy( + project_root().join("advanced/samples/inputs").join(support), + destination, + ) + .unwrap(); + } } path } diff --git a/tests/generate_go.rs b/tests/generate_go.rs index 70647343..20309dbd 100644 --- a/tests/generate_go.rs +++ b/tests/generate_go.rs @@ -1409,9 +1409,7 @@ fn go_type_roundtrip_generates_proto_conversions() { assert!(rendered.contains("out.RetryPolicy = *converted")); // The model constructs a transfer-type converter, so the SDK // runs model<->proto conversion inside the payload converter. - assert!(rendered.contains( - "return converter.NewContextualTransferTypeConverter(" - )); + assert!(rendered.contains("return converter.NewContextualTransferTypeConverter(")); assert!(rendered.contains( "func (ActivityOptions) TransferTypeConverter() (converter.TransferTypeConverter, error) {\n\treturn converter.NewContextualTransferTypeConverter" )); @@ -1458,9 +1456,7 @@ fn go_transfer_type_converter_defers_model_conversion_to_the_sdk() { assert!(rendered.contains("\trequest.namespace = workflow.GetInfo(ctx).Namespace\n")); // Construction happens in the method; the SDK owns caching by model type. - assert!(rendered.contains( - "return converter.NewContextualTransferTypeConverter(" - )); + assert!(rendered.contains("return converter.NewContextualTransferTypeConverter(")); // Value receiver, so both the model and a pointer to it opt in. assert!(rendered.contains( "func (signalWithStartWorkflowRequest) TransferTypeConverter() (converter.TransferTypeConverter, error) {\n\treturn converter.NewContextualTransferTypeConverter" @@ -1761,6 +1757,1013 @@ world api { assert!(status.success(), "generated Go model runtime tests failed"); } +// Keep these fixtures independent of the authored examples: helper selection and +// invalid references must be exercised without changing shared sample inputs. +fn render_go_serialization_context( + directive: &str, + proto_backed: bool, +) -> Result> { + let temp_dir = unique_output_path("go-serialization-context"); + fs::create_dir_all(&temp_dir)?; + let wit_path = temp_dir.join("serialization-context.wit"); + let proto = if proto_backed { + "/// @nexus.proto \"temporal.api.workflowservice.v1.RequestCancelWorkflowExecutionRequest\"" + } else { + "" + }; + fs::write( + &wit_path, + format!( + r#"/// @nexus.support go="context.go" +package temporal:serialization-context@1.0.0; + +world system {{ + export sample-service; +}} + +/// @nexus.endpoint "temporal-system" +interface sample-service {{ + type placeholder = string; + + {proto} + record cancel-request {{ + /// @nexus.source go="workflow.GetInfo(ctx).Namespace" + namespace: string, + /// @nexus.omit + workflow-execution: placeholder, + /// @nexus.omit + reason: placeholder, + /// @nexus.omit + identity: placeholder, + /// @nexus.omit + request-id: placeholder, + /// @nexus.omit + first-execution-run-id: placeholder, + /// @nexus.omit + links: placeholder, + }} + + {directive} + cancel: func(request: cancel-request); +}} +"# + ), + )?; + fs::write( + temp_dir.join("context.go"), + "package generated\n\nimport \"go.temporal.io/sdk/converter\"\nimport helpers \"example.com/contexthelpers\"\n\nfunc requestContext(request cancelRequest) converter.SerializationContext { return nil }\n", + )?; + let result = generate_to_string_with_inputs( + nexgen::language::Language::Go, + &[wit_path], + &[descriptor_path(&project_root())], + ); + fs::remove_dir_all(temp_dir)?; + result +} + +#[test] +fn go_serialization_context_is_only_emitted_for_a_selected_helper() { + for directive in [ + "", + "/// @nexus.serialization-context python=\"python_context\" typescript=\"typescriptContext\"", + ] { + let rendered = render_go_serialization_context(directive, true).unwrap(); + assert!(!rendered.contains("### serialization_context.go")); + assert!(!rendered.contains("### registry.go")); + assert!(!rendered.contains("// --- Registry ---")); + assert!(!rendered.contains("nexgenWithOperationSerializationContext")); + assert!(!rendered.contains("SerializationContext() converter.SerializationContext")); + assert!(!rendered.contains("nexgenPayloadContext")); + assert!(rendered.contains("return m.toProto(ctx)")); + } +} + +#[test] +fn go_serialization_context_prefers_go_helper_over_default() { + for (directive, helper) in [ + ( + "/// @nexus.serialization-context \"requestContext\"", + "requestContext", + ), + ( + "/// @nexus.serialization-context \"defaultContext\" go=\"requestContext\" python=\"python_context\"", + "requestContext", + ), + ] { + let rendered = render_go_serialization_context(directive, true).unwrap(); + assert!(rendered.contains(&format!("return {helper}(request.("))); + assert!(!rendered.contains("return defaultContext(")); + assert!(!rendered.contains("python_context(request)")); + } +} + +#[test] +fn go_serialization_context_accepts_only_go_helper_references() { + for helper in [ + "requestContext", + "helpers.Context", + "_requestContext", + "Context2", + ] { + let directive = format!("/// @nexus.serialization-context go=\"{helper}\""); + let rendered = render_go_serialization_context(&directive, true).unwrap(); + assert!(!rendered.contains("### serialization_context.go")); + assert!(!rendered.contains("### registry.go")); + assert!(rendered.contains("// --- Registry ---")); + assert!(rendered.contains(&format!("return {helper}(request.("))); + } + for helper in [ + "_", + "break", + "default", + "func", + "interface", + "select", + "case", + "defer", + "go", + "map", + "struct", + "chan", + "else", + "goto", + "package", + "switch", + "const", + "fallthrough", + "if", + "range", + "type", + "continue", + "for", + "import", + "return", + "var", + "helpers._", + "helpers.func", + "_.Context", + "type.Context", + "helpers.nested.Context", + "requestContext()", + "helpers/Context", + "1context", + "helpers.1context", + "request-context", + ] { + let directive = format!("/// @nexus.serialization-context go=\"{helper}\""); + let error = render_go_serialization_context(&directive, true) + .expect_err(&format!("invalid Go helper {helper:?} must be rejected")) + .to_string(); + assert!(error.contains("serialization-context"), "{helper}: {error}"); + assert!(error.contains(helper), "{helper}: {error}"); + } +} + +#[test] +fn go_serialization_context_wrapper_assigns_sources_before_sdk() { + let rendered = render_go_serialization_context( + "/// @nexus.serialization-context go=\"requestContext\"", + true, + ) + .unwrap(); + let sourced = rendered + .find("request.namespace = workflow.GetInfo(ctx).Namespace") + .unwrap(); + let execute = rendered + .find("c.ExecuteOperation(ctx, \"Cancel\", request,") + .unwrap(); + assert!(sourced < execute); + assert!(!rendered.contains("nexgenWithOperationSerializationContext")); + assert!(rendered.contains("\tnamespace string\n")); + assert!(rendered.contains("m.toProto(ctx)")); + assert!(rendered.contains("cancelRequestFromProto(ctx, message)")); + assert!(!rendered.contains("SerializationContext()")); + assert!(!rendered.contains("SerializationContextProvider")); + assert!(!rendered.contains("GetNexusOperationInnerDataConverter")); + // Nested model conversion inherits the local converter; it must not reselect it. + let to_proto = rendered + .split("func (m cancelRequest) toProto(") + .nth(1) + .unwrap(); + let body = to_proto.split("\n}\n").next().unwrap(); + assert!(!body.contains("nexgenPayloadContext")); + assert!(!body.contains("nexgenWithOperationSerializationContext")); + assert!(!rendered.contains("ExecuteOperation(nexgenPayloadContext(ctx)")); + assert!(!rendered.contains("### registry.go")); + assert!(rendered.contains("// --- Registry ---")); + assert!(!rendered.contains("### serialization_context.go")); +} + +#[test] +fn go_serialization_context_rejects_non_proto_operation_requests() { + let error = render_go_serialization_context( + "/// @nexus.serialization-context go=\"requestContext\"", + false, + ) + .unwrap_err() + .to_string(); + assert!(error.contains("serialization-context"), "{error}"); + assert!( + error.contains("unsupported") || error.contains("proto"), + "{error}" + ); +} + +#[test] +fn go_serialization_context_deferred_models_use_sdk_context() { + let root = project_root(); + // Deferred transfer callbacks use the context supplied by the SDK. + let rendered = generate_to_string_with_inputs( + nexgen::language::Language::Go, + &example_input_paths(&root, "workflow-service"), + &[descriptor_path(&root)], + ) + .unwrap(); + assert!(!rendered.contains("### serialization_context.go")); + assert!(!rendered.contains("### registry.go")); + assert!(rendered.contains("// --- Registry ---")); + assert!(!rendered.contains("nexgenWithOperationSerializationContext")); + assert!(!rendered.contains("nexgenPayloadContext")); + let support = rendered.split("### support.go\n").nth(1).unwrap(); + let support = support.split("\n### ").next().unwrap(); + assert!(support.contains("func signalWithStartWorkflowSerializationContext(")); + assert!(support.contains("func retryPolicyToProto(")); + assert_eq!( + support.matches("\"go.temporal.io/sdk/converter\"").count(), + 1 + ); + assert!(!support.contains("func nexgenPayloadContext(")); +} + +fn render_go_serialization_context_fixture( + declarations: &str, + operations: &str, + support: &str, +) -> BTreeMap { + try_render_go_serialization_context_fixture(declarations, operations, support).unwrap() +} + +fn try_render_go_serialization_context_fixture( + declarations: &str, + operations: &str, + support: &str, +) -> Result, Box> { + try_render_go_serialization_context_fixture_at_endpoint( + declarations, + operations, + support, + "temporal-system", + ) +} + +fn try_render_go_serialization_context_fixture_at_endpoint( + declarations: &str, + operations: &str, + support: &str, + endpoint: &str, +) -> Result, Box> { + let temp_dir = unique_output_path("go-serialization-context-conversions"); + fs::create_dir_all(&temp_dir).unwrap(); + let wit_path = temp_dir.join("conversions.wit"); + fs::write( + &wit_path, + format!( + r#"/// @nexus.support go="context.go" +package temporal:context-conversions@1.0.0; +world system {{ export sample-service; }} +/// @nexus.endpoint "{endpoint}" +interface sample-service {{ + {declarations} + {operations} +}} +"# + ), + ) + .unwrap(); + fs::write(temp_dir.join("context.go"), support).unwrap(); + let output_path = temp_dir.join("output"); + let result = generate_to_file(&GenerateRequest { + config: nexgen::nexgen_config::NexgenConfig { + mode: nexgen::generator::GenerationMode::NativeApi, + ..Default::default() + }, + language: nexgen::language::Language::Go, + input_paths: vec![wit_path], + support_paths: Vec::new(), + descriptor_paths: vec![descriptor_path(&project_root())], + output_path: output_path.clone(), + // Run gofmt over every file to check the IIFE's Go syntax as well as + // inspecting its pointer types and context scope below. + format: true, + java_package_name: None, + ts_date_time_types: Default::default(), + }); + let files = result.map(|_| read_go_output_files(&output_path)); + fs::remove_dir_all(temp_dir).unwrap(); + Ok(files?) +} + +fn assert_no_generated_context_plumbing(files: &BTreeMap) { + assert!(!files.contains_key(Path::new("serialization_context.go"))); + assert!(!files.contains_key(Path::new("registry.go"))); + let source = &files[Path::new("sampleservice.go")]; + for forbidden in [ + "nexgenWithOperationSerializationContext", + "nexgenPayloadContext", + "nexgenSerializationContextKey", + "WithDataConverter", + "WithValue", + "WithRootDataConverterSerializationContext", + ".SerializationContext(request)", + ] { + assert!(!source.contains(forbidden), "sampleservice.go: {forbidden}"); + } +} + +fn go_registry_main(files: &BTreeMap) -> &str { + assert!(!files.contains_key(Path::new("registry.go"))); + let source = &files[Path::new("sampleservice.go")]; + assert_eq!(source.matches("// --- Registry ---").count(), 1); + source +} + +fn go_registry_section(source: &str) -> &str { + source.split("// --- Registry ---").nth(1).unwrap() +} + +fn assert_registry_entry(registry: &str, service: &str, operation: &str, helper: Option<&str>) { + let compact: String = registry.chars().filter(|c| !c.is_whitespace()).collect(); + let key = format!("Service:\"{service}\",Operation:\"{operation}\""); + assert!(!compact.contains("InputType:"), "{registry}"); + let Some(helper) = helper else { + assert!(!compact.contains(&key), "{registry}"); + return; + }; + assert_eq!(compact.matches(&key).count(), 1, "{registry}"); + let lines = registry + .lines() + .map(str::trim) + .collect::>() + .join("\n"); + assert!( + lines.contains(&format!( + "Service: \"{service}\",\nOperation: \"{operation}\",\n" + )), + "{registry}" + ); + let entry = compact.split(&key).nth(1).unwrap(); + let value = entry.trim_start_matches(',').strip_prefix("}:").unwrap(); + let input_type = value + .split("request.(") + .nth(1) + .unwrap() + .split(')') + .next() + .unwrap(); + assert!( + value.starts_with(&format!( + "{{SerializationContext:func(requestany)converter.SerializationContext{{return{helper}(request.({input_type}))}}," + )), + "{registry}" + ); +} + +#[test] +fn go_serialization_context_registry_uses_wire_names_and_typed_adapter() { + let files = render_go_serialization_context_fixture( + &format!("{GO_CONTEXT_OWNED_REQUEST}\nrecord plain-request {{ name: string }}"), + r#"/// @nexus.operation name="WireFirst" + + /// @nexus.serialization-context go="requestContext" + first: func(request: context-request); + /// @nexus.operation name="WireSecond" + second: func(request: context-request); + third: func(request: plain-request);"#, + "package generated\n", + ); + // The fixture uses default configuration: --system-nexus is not required. + let registry = go_registry_main(&files); + let section = go_registry_section(registry); + assert!( + registry.find("c.ExecuteOperation(").unwrap() + < registry.find("// --- Registry ---").unwrap() + ); + assert_registry_entry( + registry, + "SampleService", + "WireFirst", + Some("requestContext"), + ); + assert_registry_entry(registry, "SampleService", "WireSecond", None); + assert_registry_entry(registry, "SampleService", "Third", None); + assert!(!registry.contains("Operation: \"First\"")); + let compact = registry.split_whitespace().collect::>().join(" "); + assert!(!registry.contains("type NexusOperationKey")); + assert!(!registry.contains("NexusOperationInfo")); + assert!(!registry.contains("nexgenOperationInfo")); + assert!(!registry.contains("// NexusOperationRegistry")); + assert!(compact.contains( + "func init() { internal.RegisterNexusOperationRegistry(NexusOperationRegistry) }" + )); + assert!(compact.contains( + "var NexusOperationRegistry = map[internal.NexusOperationKey]internal.NexusOperationRegistryEntry" + )); + assert!(compact.contains("return requestContext(request.(contextRequest))")); + assert_eq!(registry.matches("\"reflect\"").count(), 1); + assert_eq!( + registry.matches("\"go.temporal.io/sdk/converter\"").count(), + 1 + ); + assert!(!registry.contains("SerializationContextProvider")); + assert_no_generated_context_plumbing(&files); + assert_eq!( + registry.matches("\"go.temporal.io/sdk/internal\"").count(), + 1 + ); + assert!(!section.contains("c.ExecuteOperation(")); +} + +fn render_go_registry_services( + second_endpoint: &str, +) -> Result> { + // Sharing wire keys across endpoints must only conflict when both are system endpoints. + let temp_dir = unique_output_path("go-context-duplicate-wire-key"); + fs::create_dir_all(&temp_dir).unwrap(); + let wit_path = temp_dir.join("services.wit"); + fs::write( + &wit_path, + r#" +package temporal:duplicate-context-keys@1.0.0; +world system { export first-service; export second-service; } +interface shared-types { + /// @nexus.proto "temporal.api.common.v1.WorkflowType" + record context-request { name: string } +} +/// @nexus.endpoint "__temporal_system" +/// @nexus.service-name "SameWireService" +interface first-service { + use shared-types.{context-request}; + /// @nexus.operation name="SameWireOperation" + /// @nexus.serialization-context go="requestContext" + first: func(request: context-request); +} +/// @nexus.endpoint "SECOND_ENDPOINT" +/// @nexus.service-name "SameWireService" +interface second-service { + use shared-types.{context-request}; + /// @nexus.operation name="SameWireOperation" + /// @nexus.serialization-context go="otherContext" + second: func(request: context-request); + /// @nexus.serialization-context go="otherContext" + ordinary-only: func(request: context-request); +} +"# + .replace("SECOND_ENDPOINT", second_endpoint), + ) + .unwrap(); + let result = generate_to_string_with_inputs( + nexgen::language::Language::Go, + &[wit_path], + &[descriptor_path(&project_root())], + ); + fs::remove_dir_all(temp_dir).unwrap(); + result +} + +#[test] +fn go_serialization_context_registry_rejects_duplicate_wire_keys() { + for endpoint in ["__temporal_system", "temporal-system"] { + let error = render_go_registry_services(endpoint) + .unwrap_err() + .to_string(); + assert!(error.contains("SameWireService"), "{error}"); + assert!(error.contains("SameWireOperation"), "{error}"); + assert!( + error.contains("duplicate") || error.contains("collision"), + "{error}" + ); + } +} + +#[test] +fn go_serialization_context_mixed_endpoints_register_only_system_operations() { + let rendered = render_go_registry_services("ordinary-endpoint").unwrap(); + assert!(!rendered.contains("### registry.go")); + let registry = rendered + .split("// --- Registry ---") + .nth(1) + .unwrap() + .split("\n### ") + .next() + .unwrap(); + assert_registry_entry( + registry, + "SameWireService", + "SameWireOperation", + Some("requestContext"), + ); + assert!(!registry.contains("OrdinaryOnly")); + assert!(!registry.contains("otherContext")); + assert_eq!( + registry + .matches("RegisterNexusOperationRegistry(NexusOperationRegistry)") + .count(), + 1 + ); + assert!(rendered.contains("c.ExecuteOperation(ctx, \"OrdinaryOnly\", request,")); + assert!( + rendered.contains("workflow.NewNexusClient(\"ordinary-endpoint\", \"SameWireService\")") + ); +} + +const GO_CONTEXT_OWNED_REQUEST: &str = r#" + /// @nexus.proto "temporal.api.common.v1.WorkflowType" + record context-request { name: string } +"#; + +const GO_CONTEXT_OVERRIDE_TYPES: &str = r#" + type placeholder = string; + /// @nexus.proto "temporal.api.common.v1.RetryPolicy" + /// @nexus.type go="go.temporal.io/sdk/temporal.RetryPolicy" + type retry-policy = placeholder; +"#; + +const GO_CONTEXT_OVERRIDE_SUPPORT: &str = r#"package generated + +import ( + common "go.temporal.io/api/common/v1" + "go.temporal.io/sdk/converter" + "go.temporal.io/sdk/temporal" + "go.temporal.io/sdk/workflow" +) + +func requestContext(request temporal.RetryPolicy) converter.SerializationContext { + return converter.WorkflowSerializationContext{Namespace: "target-namespace", WorkflowID: "target-workflow"} +} +func retryPolicyToProto(ctx workflow.Context, value *temporal.RetryPolicy) (*common.RetryPolicy, error) { + return &common.RetryPolicy{}, nil +} +func retryPolicyFromProto(ctx workflow.Context, value *common.RetryPolicy) (*temporal.RetryPolicy, error) { + return &temporal.RetryPolicy{}, nil +} +func transformPolicy(ctx workflow.Context, value *common.RetryPolicy) (string, error) { + return "transformed", nil +} +"#; + +#[test] +fn go_serialization_context_rejects_external_request() { + for endpoint in ["__temporal_system", "temporal-system"] { + for native_type in ["temporal.RetryPolicy", "*temporal.RetryPolicy"] { + for operation in [ + "roundtrip: func(request: retry-policy) -> retry-policy;", + "send: func(request: retry-policy);", + ] { + let error = try_render_go_serialization_context_fixture_at_endpoint( + &GO_CONTEXT_OVERRIDE_TYPES.replace("temporal.RetryPolicy", native_type), + &format!("/// @nexus.serialization-context go=\"requestContext\"\n{operation}"), + &GO_CONTEXT_OVERRIDE_SUPPORT.replace( + "request temporal.RetryPolicy", + &format!("request {native_type}"), + ), + endpoint, + ) + .unwrap_err() + .to_string(); + let operation_name = if operation.starts_with("roundtrip") { + "Roundtrip" + } else { + "Send" + }; + assert!(error.contains(operation_name), "{error}"); + assert!(error.contains(native_type), "{error}"); + assert!(error.contains("TransferTypeConverter"), "{error}"); + assert!(error.contains("@nexus.serialization-context"), "{error}"); + assert!(error.contains("locally owned request model"), "{error}"); + } + } + } +} + +#[test] +fn go_serialization_context_ordinary_endpoint_converts_external_input_in_wrapper() { + let files = try_render_go_serialization_context_fixture_at_endpoint( + GO_CONTEXT_OVERRIDE_TYPES, + r#"/// @nexus.serialization-context go="requestContext" + roundtrip: func(request: retry-policy) -> retry-policy;"#, + GO_CONTEXT_OVERRIDE_SUPPORT, + "sample-service", + ) + .unwrap(); + let operation = &files[Path::new("sampleservice.go")]; + let main = go_registry_main(&files); + let registry = go_registry_section(main); + let conversion = operation + .find("requestProto, err := retryPolicyToProto(ctx, &request)") + .unwrap(); + let error = operation[conversion..] + .find("resultSettable.SetError(err)") + .unwrap() + + conversion; + let execute = operation + .find("c.ExecuteOperation(ctx, \"Roundtrip\", requestProto,") + .unwrap(); + assert!(conversion < error && error < execute, "{operation}"); + assert!(operation.contains("workflow.NewNexusClient(\"sample-service\", \"SampleService\")")); + assert_eq!(operation.matches("retryPolicyToProto(").count(), 1); + assert!(operation.contains("value, err := retryPolicyFromProto(ctx, &result)")); + assert!(!operation.contains("requestContext(")); + assert_eq!(main.matches("\"go.temporal.io/sdk/temporal\"").count(), 1); + let compact = registry.split_whitespace().collect::>().join(" "); + assert!( + compact.contains("var NexusOperationRegistry = map[internal.NexusOperationKey]internal.NexusOperationRegistryEntry{}") + ); + assert!(!registry.contains("func init()")); + assert!(!registry.contains("RegisterNexusOperationRegistry")); + assert!(!registry.contains("nexgenOperationInfo")); + assert!(!registry.contains("requestContext")); + assert!(!main.contains("\"reflect\"")); + assert!(!main.contains("\"go.temporal.io/sdk/converter\"")); + assert_no_generated_context_plumbing(&files); +} + +#[test] +fn go_serialization_context_preserves_owned_input_and_uses_ctx_for_eager_output() { + let files = render_go_serialization_context_fixture( + &format!("{GO_CONTEXT_OWNED_REQUEST}\n{GO_CONTEXT_OVERRIDE_TYPES}"), + r#"/// @nexus.serialization-context go="requestContext" + roundtrip: func(request: context-request) -> retry-policy;"#, + &GO_CONTEXT_OVERRIDE_SUPPORT + .replace("request temporal.RetryPolicy", "request contextRequest"), + ); + let operation = &files[Path::new("sampleservice.go")]; + let compact = operation.split_whitespace().collect::>().join(" "); + assert!(!operation.contains("requestProto")); + + assert!( + compact.contains("value, err := retryPolicyFromProto(ctx, &result)"), + "{operation}" + ); + assert!(operation.contains("return m.toProto(ctx)")); + let execute = operation + .find("c.ExecuteOperation(ctx, \"Roundtrip\", request,") + .unwrap(); + assert!(!operation.contains("nexgenWithOperationSerializationContext")); + let decode = operation.find("fut.Get(ctx, &result)").unwrap(); + let output = decode + operation[decode..].find("value, err :=").unwrap(); + assert!(execute < decode && decode < output); + assert!(operation.contains("var result common.RetryPolicy")); + assert!( + operation.contains("resultSettable.Set(*value, nil)"), + "{operation}" + ); + assert_eq!( + operation + .matches("value, err := retryPolicyFromProto(ctx, &result)") + .count(), + 1 + ); + assert!(!operation.contains("ctx = nexgenPayloadContext(ctx")); + assert_no_generated_context_plumbing(&files); + assert!(operation.contains("\"go.temporal.io/sdk/internal\"")); +} + +#[test] +fn go_serialization_context_uses_ctx_for_output_transform_expression() { + let files = render_go_serialization_context_fixture( + &format!("{GO_CONTEXT_OWNED_REQUEST}\n{GO_CONTEXT_OVERRIDE_TYPES}"), + r#"/// @nexus.serialization-context go="requestContext" + /// @nexus.output-transform go-type="string" go="transformPolicy(ctx, &result)" + transform: func(request: context-request) -> retry-policy;"#, + &GO_CONTEXT_OVERRIDE_SUPPORT + .replace("request temporal.RetryPolicy", "request contextRequest"), + ); + let operation = &files[Path::new("sampleservice.go")]; + let compact = operation.split_whitespace().collect::>().join(" "); + assert!( + compact.contains("value, err := transformPolicy(ctx, &result)"), + "{operation}" + ); + assert!(operation.contains("var result common.RetryPolicy")); + let decode = operation.find("fut.Get(ctx, &result)").unwrap(); + let transform = operation + .find("value, err := transformPolicy(ctx, &result)") + .unwrap(); + assert!(decode < transform); + let execute = operation + .find("c.ExecuteOperation(ctx, \"Transform\", request,") + .unwrap(); + assert!(execute < decode); + assert!(operation.contains("resultSettable.Set(value, nil)")); + assert!(!operation.contains("retryPolicyFromProto(")); + assert!(!operation.contains("ctx = ")); +} + +#[test] +fn go_serialization_context_shared_model_has_one_registry_entry_per_operation() { + let files = render_go_serialization_context_fixture( + GO_CONTEXT_OWNED_REQUEST, + r#"/// @nexus.serialization-context go="requestContext" + first: func(request: context-request); + /// @nexus.serialization-context "requestContext" + second: func(request: context-request);"#, + "package generated\n", + ); + let registry = go_registry_main(&files); + assert_eq!( + registry + .matches("return requestContext(request.(contextRequest))") + .count(), + 2 + ); + assert_registry_entry(registry, "SampleService", "First", Some("requestContext")); + assert_registry_entry(registry, "SampleService", "Second", Some("requestContext")); + assert!(!files[Path::new("sampleservice.go")].contains("SerializationContext()")); + assert!(!files.contains_key(Path::new("serialization_context.go"))); +} + +#[test] +fn go_serialization_context_shared_model_all_absent_emits_no_registry() { + let files = render_go_serialization_context_fixture( + GO_CONTEXT_OWNED_REQUEST, + r#"first: func(request: context-request); + /// @nexus.serialization-context python="python_context" + second: func(request: context-request);"#, + "package generated\n", + ); + let model = &files[Path::new("sampleservice.go")]; + assert!(!model.contains("SerializationContext()")); + assert!(!model.contains("nexgenPayloadContext")); + assert!(!files.contains_key(Path::new("serialization_context.go"))); + assert!(!files.contains_key(Path::new("registry.go"))); + assert!(!model.contains("// --- Registry ---")); + assert!(!model.contains("NexusOperationRegistry")); +} + +#[test] +fn go_serialization_context_shared_models_and_aliases_have_independent_policies() { + for second_policy in [ + "/// @nexus.serialization-context go=\"otherContext\"", + "", + "/// @nexus.serialization-context python=\"python_context\"", + ] { + for alias in [false, true] { + let declarations = if alias { + format!("{GO_CONTEXT_OWNED_REQUEST}\n type request-alias = context-request;") + } else { + GO_CONTEXT_OWNED_REQUEST.to_owned() + }; + let second_model = if alias { + "request-alias" + } else { + "context-request" + }; + let files = try_render_go_serialization_context_fixture( + &declarations, + &format!( + r#"/// @nexus.serialization-context go="requestContext" + first: func(request: context-request); + {second_policy} + second: func(request: {second_model});"# + ), + "package generated\n", + ) + .unwrap(); + let registry = go_registry_main(&files); + assert_registry_entry(registry, "SampleService", "First", Some("requestContext")); + let second_helper = second_policy + .contains("otherContext") + .then_some("otherContext"); + assert_registry_entry(registry, "SampleService", "Second", second_helper); + let operation = &files[Path::new("sampleservice.go")]; + assert_no_generated_context_plumbing(&files); + assert!(!operation.contains("SerializationContext()")); + } + } +} + +#[test] +fn go_serialization_context_shared_models_across_services_have_independent_policies() { + for second_policy in [ + "/// @nexus.serialization-context go=\"requestContext\"", + "/// @nexus.serialization-context go=\"otherContext\"", + "", + ] { + let temp_dir = unique_output_path("go-context-cross-service"); + fs::create_dir_all(&temp_dir).unwrap(); + let wit_path = temp_dir.join("services.wit"); + fs::write( + &wit_path, + format!( + r#" +package temporal:context-services@1.0.0; +world system {{ export first-service; export second-service; }} +interface shared-types {{ + /// @nexus.proto "temporal.api.common.v1.WorkflowType" + record context-request {{ name: string }} +}} +/// @nexus.endpoint "__temporal_system" +/// @nexus.service-name "wire.first" +interface first-service {{ + use shared-types.{{context-request}}; + /// @nexus.serialization-context go="requestContext" + first: func(request: context-request); +}} +/// @nexus.endpoint "temporal-system" +interface second-service {{ + use shared-types.{{context-request as request-alias}}; + {second_policy} + second: func(request: request-alias); +}} +"# + ), + ) + .unwrap(); + let result = generate_to_string_with_inputs( + nexgen::language::Language::Go, + &[wit_path], + &[descriptor_path(&project_root())], + ); + fs::remove_dir_all(temp_dir).unwrap(); + let rendered = result.unwrap(); + assert!(!rendered.contains("### registry.go")); + let registry = rendered + .split("// --- Registry ---") + .nth(1) + .unwrap() + .split("\n### ") + .next() + .unwrap(); + assert_registry_entry(registry, "wire.first", "First", Some("requestContext")); + assert!(!rendered.contains("nexgenWithOperationSerializationContext")); + let helper = if second_policy.contains("requestContext") { + Some("requestContext") + } else if second_policy.contains("otherContext") { + Some("otherContext") + } else { + None + }; + assert_registry_entry(registry, "SecondService", "Second", helper); + assert!(!rendered.contains("SerializationContext()")); + } +} + +#[test] +fn go_serialization_context_distinct_authored_models_sharing_proto_have_independent_policies() { + let files = render_go_serialization_context_fixture( + &format!( + r#"{GO_CONTEXT_OWNED_REQUEST} + /// @nexus.proto "temporal.api.common.v1.WorkflowType" + record other-request {{ name: string }}"# + ), + r#"/// @nexus.serialization-context go="requestContext" + first: func(request: context-request); + /// @nexus.serialization-context go="otherContext" + second: func(request: other-request);"#, + "package generated\n", + ); + let registry = go_registry_main(&files); + assert_registry_entry(registry, "SampleService", "First", Some("requestContext")); + assert_registry_entry(registry, "SampleService", "Second", Some("otherContext")); +} + +#[test] +fn go_serialization_context_allows_model_field_named_serialization_context() { + let declarations = GO_CONTEXT_OWNED_REQUEST.replace( + "name: string", + "/// @nexus.name go=\"SerializationContext\"\n name: string", + ); + let files = render_go_serialization_context_fixture( + &declarations, + r#"/// @nexus.serialization-context go="requestContext" + first: func(request: context-request);"#, + "package generated\n", + ); + let model = &files[Path::new("sampleservice.go")]; + assert!(model.contains("SerializationContext string")); + assert!(!model.contains("SerializationContext()")); + assert_registry_entry( + go_registry_main(&files), + "SampleService", + "First", + Some("requestContext"), + ); +} + +#[test] +fn go_serialization_context_constructs_resource_return_after_decode() { + let files = render_go_serialization_context_fixture( + r#"/// @nexus.proto "temporal.api.common.v1.WorkflowType" + record start-request { name: string } + resource started-workflow { + constructor(name: string, run-id: string); + } + /// @nexus.proto "temporal.api.common.v1.WorkflowExecution" + type start-response = own;"#, + r#"/// @nexus.serialization-context go="requestContext" + start: func(request: start-request) -> start-response;"#, + r#"package generated +import "go.temporal.io/sdk/converter" +func requestContext(request startRequest) converter.SerializationContext { return nil } +"#, + ); + let operation = &files[Path::new("sampleservice.go")]; + let execute = operation + .find("c.ExecuteOperation(ctx, \"Start\", request,") + .unwrap(); + assert!(!operation.contains("nexgenWithOperationSerializationContext")); + let callback = operation[execute..] + .find("workflow.Go(ctx, func(ctx workflow.Context) {") + .unwrap() + + execute; + let decode = operation[callback..].find("fut.Get(ctx, &result)").unwrap() + callback; + let construct = operation + .find("value := NewStartedWorkflow(request.Name, result.GetRunId())") + .unwrap(); + assert!(execute < callback && callback < decode && decode < construct); + assert!(!operation.contains("ctx = ")); + assert!(operation.contains("var result common.WorkflowExecution")); + assert!(operation.contains("resultSettable.Set(value, nil)")); +} + +#[test] +fn go_serialization_context_registry_can_resolve_known_model_imports() { + // This source-generation fixture tests import resolution, not helper linkage. + // No support fragment supplies the temporal import used by the override. + let files = render_go_serialization_context_fixture( + GO_CONTEXT_OVERRIDE_TYPES, + r#"/// @nexus.serialization-context go="temporal.Context" + roundtrip: func(request: retry-policy) -> retry-policy;"#, + "package generated\n", + ); + let registry = go_registry_main(&files); + assert_eq!( + registry.matches("\"go.temporal.io/sdk/temporal\"").count(), + 1 + ); + assert_registry_entry( + registry, + "SampleService", + "Roundtrip", + Some("temporal.Context"), + ); +} + +#[test] +fn go_serialization_context_accepts_concrete_helper_return_type() { + let files = render_go_serialization_context_fixture( + GO_CONTEXT_OWNED_REQUEST, + r#"/// @nexus.serialization-context go="requestContext" + first: func(request: context-request);"#, + r#"package generated +import "go.temporal.io/sdk/converter" +func requestContext(request contextRequest) converter.WorkflowSerializationContext { + return converter.WorkflowSerializationContext{WorkflowID: request.Name} +} +"#, + ); + assert_registry_entry( + go_registry_main(&files), + "SampleService", + "First", + Some("requestContext"), + ); + assert!(!files[Path::new("sampleservice.go")].contains("SerializationContext()")); +} + +#[test] +fn go_serialization_context_imports_qualified_helper_alias_once_in_main() { + let files = render_go_serialization_context_fixture( + GO_CONTEXT_OWNED_REQUEST, + r#"/// @nexus.serialization-context go="helpers.Context" + roundtrip: func(request: context-request);"#, + &GO_CONTEXT_OVERRIDE_SUPPORT + .replace( + "import (", + "import (\n helpers \"example.com/contexthelpers\"", + ) + .replace("request temporal.RetryPolicy", "request contextRequest") + .replace("return nil }", "return helpers.Context(request) }"), + ); + let registry = go_registry_main(&files); + assert!( + registry.contains("helpers \"example.com/contexthelpers\""), + "{registry}" + ); + assert_eq!( + registry.matches("\"example.com/contexthelpers\"").count(), + 1 + ); + assert_registry_entry( + registry, + "SampleService", + "Roundtrip", + Some("helpers.Context"), + ); + assert!(files[Path::new("support.go")].contains("helpers \"example.com/contexthelpers\"")); +} + #[test] fn go_proto_resource_return_converts_request_and_constructs_resource() { let root = project_root();