workflows support reuse id policy (#637)

* workflows support reuse id policy

Signed-off-by: Fabian Martinez <[email protected]>

* add new type

Signed-off-by: Fabian Martinez <[email protected]>

* lint

Signed-off-by: Fabian Martinez <[email protected]>

* wrap all types

Signed-off-by: Fabian Martinez <[email protected]>

* use existing type

Signed-off-by: Fabian Martinez <[email protected]>

---------

Signed-off-by: Fabian Martinez <[email protected]>

Author: famarting

workflow/client.go

@@ -31,6 +31,19 @@ type Client struct {
 	taskHubClient *durabletaskclient.TaskHubGrpcClient
 }
 
+type WorkflowIDReusePolicy struct {
+	OperationStatus []Status
+	Action          CreateWorkflowAction
+}
+
+type CreateWorkflowAction = api.CreateOrchestrationAction
+
+const (
+	ReuseIDActionError     CreateWorkflowAction = api.REUSE_ID_ACTION_ERROR
+	ReuseIDActionIgnore    CreateWorkflowAction = api.REUSE_ID_ACTION_IGNORE
+	ReuseIDActionTerminate CreateWorkflowAction = api.REUSE_ID_ACTION_TERMINATE
+)
+
 // WithInstanceID is an option to set an InstanceID when scheduling a new workflow.
 func WithInstanceID(id string) api.NewOrchestrationOptions {
 	return api.WithInstanceID(api.InstanceID(id))
@@ -53,6 +66,13 @@ func WithStartTime(time time.Time) api.NewOrchestrationOptions {
 	return api.WithStartTime(time)
 }
 
+func WithReuseIDPolicy(policy WorkflowIDReusePolicy) api.NewOrchestrationOptions {
+	return api.WithOrchestrationIdReusePolicy(&api.OrchestrationIdReusePolicy{
+		OperationStatus: convertStatusSlice(policy.OperationStatus),
+		Action:          policy.Action,
+	})
+}
+
 // WithFetchPayloads is an option to return the payload from a workflow.
 func WithFetchPayloads(fetchPayloads bool) api.FetchOrchestrationMetadataOptions {
 	return api.WithFetchPayloads(fetchPayloads)

workflow/client_test.go

@@ -54,7 +54,10 @@ func TestClientMethods(t *testing.T) {
 	}
 	ctx := context.Background()
 	t.Run("ScheduleNewWorkflow - empty wf name", func(t *testing.T) {
-		id, err := testClient.ScheduleNewWorkflow(ctx, "", nil)
+		id, err := testClient.ScheduleNewWorkflow(ctx, "", WithReuseIDPolicy(WorkflowIDReusePolicy{
+			OperationStatus: []Status{StatusCompleted},
+			Action:          ReuseIDActionIgnore,
+		}))
 		require.Error(t, err)
 		assert.Empty(t, id)
 	})

workflow/state.go

@@ -48,6 +48,28 @@ func (s Status) String() string {
 	return status[s]
 }
 
+func (s Status) RuntimeStatus() api.OrchestrationStatus {
+	switch s {
+	case StatusRunning:
+		return api.RUNTIME_STATUS_RUNNING
+	case StatusCompleted:
+		return api.RUNTIME_STATUS_COMPLETED
+	case StatusContinuedAsNew:
+		return api.RUNTIME_STATUS_CONTINUED_AS_NEW
+	case StatusFailed:
+		return api.RUNTIME_STATUS_FAILED
+	case StatusCanceled:
+		return api.RUNTIME_STATUS_CANCELED
+	case StatusTerminated:
+		return api.RUNTIME_STATUS_TERMINATED
+	case StatusPending:
+		return api.RUNTIME_STATUS_PENDING
+	case StatusSuspended:
+		return api.RUNTIME_STATUS_SUSPENDED
+	}
+	return -1
+}
+
 type WorkflowState struct {
 	Metadata api.OrchestrationMetadata
 }
@@ -57,3 +79,11 @@ func (wfs *WorkflowState) RuntimeStatus() Status {
 	s := Status(wfs.Metadata.RuntimeStatus.Number())
 	return s
 }
+
+func convertStatusSlice(ss []Status) []api.OrchestrationStatus {
+	out := []api.OrchestrationStatus{}
+	for _, s := range ss {
+		out = append(out, s.RuntimeStatus())
+	}
+	return out
+}