mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
7fec4a9a3f
## What changed? This PR adds `SignalWithStartWorkflowExecution` as a synchronous Nexus operation exposed via `__temporal_system endpoint`, allowing workflows to signal-with-start other workflows through the CHASM Nexus operation framework. Key changes: - `chasm/lib/workflow/nexus_service.go`: `workflowServiceNexusHandler` implements the `SignalWithStartWorkflowExecution` Nexus sync operation by resolving the namespace and delegating to the History service. A `SignalWithStartOperationProcessor` handles input enrichment (namespace, request ID, links) and routing via CHASM's `NexusOperationProcessorResult`. - `chasm/lib/workflow/library.go` — library now holds the `workflowServiceNexusHandler`, the workflow `Config`, the SA mapper provider, and the SA validator. `newLibrary` (used by fx) takes those dependencies, while the public `NewLibrary` keeps its old signature for external callers. Adds `NexusServices()` so the library registers its Nexus service via CHASM. - `chasm/lib/workflow/validator.go`: `RequestValidator` consolidates the `SignalWithStartWorkflowExecution` validation logic (previously inlined in `WorkflowHandler`) into a reusable, injectable struct. This same validator is used by both the frontend handler and the new CHASM processor. - common/dynamicconfig/constants.go — adds `EnableSignalWithStartFromWorkflow` (namespace-scoped, default false). - `service/frontend/workflow_handler.go`: Removed the `SignalWithStart` validation block - `service/history/fx.go`: Provides a `HistoryServiceServerProvider` so the CHASM workflow library can call the history handler directly. - `temporal/fx.go`: Removes the now-redundant `ChasmLibraryOptions` grouping; each service module registers its own CHASM libraries. - `components/nexusoperations/workflow/commands.go`: `NotFound` and `InvalidArgument` errors during Nexus command handling are now surfaced as workflow task failures instead of being treated as transient handler errors. - `common/payloads`: Adds `EncodeSingle`, `MustEncodeSingle`, and `MustEncode` helpers used in tests. - `cmd/tools/getproto`: Adds support for nexus-proto-annotations proto imports. - `tests/signal_with_start_from_workflow_test.go`: Functional test suite covering the happy path, duplicate detection, conflict policies, and validation rejection for the new Nexus operation. ## Why? This functionality is one of our most requested GitHub issues. ## How did you test it? - [X] built - [X] run locally and tested manually - [ ] covered by existing tests - [X] added new unit test(s) - [X] added new functional test(s) ## Potential risks The history service now directly exposes a `HistoryServiceServer` interface via `fx` for injection into the CHASM workflow library. This tight coupling between the CHASM workflow library and the history handler could complicate future layering — callers outside the history service should not adopt this pattern. The feature is gated by `history.enableSignalWithStartFromWorkflow` for rollout. --------- Co-authored-by: Roey Berman <[email protected]>
71 lines
1.8 KiB
Go
71 lines
1.8 KiB
Go
package payloads
|
|
|
|
import (
|
|
"fmt"
|
|
"strings"
|
|
|
|
commonpb "go.temporal.io/api/common/v1"
|
|
"go.temporal.io/sdk/converter"
|
|
)
|
|
|
|
var (
|
|
defaultDataConverter = converter.GetDefaultDataConverter()
|
|
)
|
|
|
|
func EncodeString(str string) *commonpb.Payloads {
|
|
// Error can be safely ignored here becase string always can be converted.
|
|
ps, _ := defaultDataConverter.ToPayloads(str)
|
|
return ps
|
|
}
|
|
|
|
func EncodeInt(i int) *commonpb.Payloads {
|
|
// Error can be safely ignored here becase int always can be converted.
|
|
ps, _ := defaultDataConverter.ToPayloads(i)
|
|
return ps
|
|
}
|
|
|
|
func EncodeBytes(bytes []byte) *commonpb.Payloads {
|
|
// Error can be safely ignored here becase []byte always can be raw encoded.
|
|
ps, _ := defaultDataConverter.ToPayloads(bytes)
|
|
return ps
|
|
}
|
|
|
|
func Encode(value ...any) (*commonpb.Payloads, error) {
|
|
return defaultDataConverter.ToPayloads(value...)
|
|
}
|
|
|
|
func EncodeSingle(value any) (*commonpb.Payload, error) {
|
|
ps, err := defaultDataConverter.ToPayloads(value)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(ps.GetPayloads()) < 1 {
|
|
return nil, nil
|
|
}
|
|
return ps.GetPayloads()[0], nil
|
|
}
|
|
|
|
func MustEncodeSingle(value any) *commonpb.Payload {
|
|
p, err := EncodeSingle(value)
|
|
if err != nil {
|
|
panic(fmt.Sprintf("unable to encode single payload: %v", err)) //nolint:forbidigo // Must-helper: callers opt into panic on encode failure
|
|
}
|
|
return p
|
|
}
|
|
|
|
func MustEncode(value ...any) *commonpb.Payloads {
|
|
p, err := defaultDataConverter.ToPayloads(value...)
|
|
if err != nil {
|
|
panic(fmt.Sprintf("unable to encode payloads: %v", err)) //nolint:forbidigo // Must-helper: callers opt into panic on encode failure
|
|
}
|
|
return p
|
|
}
|
|
|
|
func Decode(ps *commonpb.Payloads, valuePtr ...any) error {
|
|
return defaultDataConverter.FromPayloads(ps, valuePtr...)
|
|
}
|
|
|
|
func ToString(ps *commonpb.Payloads) string {
|
|
return fmt.Sprintf("[%s]", strings.Join(defaultDataConverter.ToStrings(ps), ", "))
|
|
}
|