mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
zigeHandGitHub
3f92925356
## What changed? only error message ## Why? Fix error message for catchup input validation ## How did you test it? - [ ] built - [x] run locally and tested manually - [ ] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s) ## Potential risks no risk
60 lines
1.3 KiB
Go
60 lines
1.3 KiB
Go
package migration
|
|
|
|
import (
|
|
"time"
|
|
|
|
"go.temporal.io/sdk/temporal"
|
|
"go.temporal.io/sdk/workflow"
|
|
)
|
|
|
|
const (
|
|
catchupWorkflowName = "catchup"
|
|
)
|
|
|
|
type (
|
|
CatchUpParams struct {
|
|
Namespace string
|
|
CatchupCluster string
|
|
TargetCluster string
|
|
}
|
|
|
|
CatchUpOutput struct{}
|
|
)
|
|
|
|
func CatchupWorkflow(ctx workflow.Context, params CatchUpParams) (CatchUpOutput, error) {
|
|
if err := validateCatchupParams(¶ms); err != nil {
|
|
return CatchUpOutput{}, err
|
|
}
|
|
|
|
retryPolicy := &temporal.RetryPolicy{
|
|
InitialInterval: time.Second,
|
|
MaximumInterval: time.Second,
|
|
BackoffCoefficient: 1,
|
|
}
|
|
ao := workflow.ActivityOptions{
|
|
StartToCloseTimeout: time.Hour,
|
|
HeartbeatTimeout: time.Second * 10,
|
|
RetryPolicy: retryPolicy,
|
|
}
|
|
ctx1 := workflow.WithActivityOptions(ctx, ao)
|
|
|
|
var a *activities
|
|
err := workflow.ExecuteActivity(ctx1, a.WaitCatchup, params).Get(ctx, nil)
|
|
if err != nil {
|
|
return CatchUpOutput{}, err
|
|
}
|
|
|
|
return CatchUpOutput{}, err
|
|
}
|
|
|
|
func validateCatchupParams(params *CatchUpParams) error {
|
|
if len(params.Namespace) == 0 {
|
|
return temporal.NewNonRetryableApplicationError("InvalidArgument: Namespace is required", "InvalidArgument", nil)
|
|
}
|
|
if len(params.CatchupCluster) == 0 {
|
|
return temporal.NewNonRetryableApplicationError("InvalidArgument: CatchupCluster is required", "InvalidArgument", nil)
|
|
}
|
|
|
|
return nil
|
|
}
|