This Sample Workflow Execute 3 Steps In Sequence

This sample workflow execute 3 steps in sequence.

description: globs: alwaysApply: false

This file provides rules and context for generating or understanding Go code related to a custom Domain Specific Language (DSL) for defining Temporal workflows within this project.

DSL Overview:

This project uses a specific Go-based DSL to define Temporal workflows declaratively. The core idea is to represent workflow logic using nested structures rather than imperative Go code directly within the workflow function.

Core DSL Structures (Defined in dsl package):

• Workflow: The top-level structure representing the entire workflow definition.
  • Variables map[string]string: Initial variables available to the workflow. These act as the initial state or input.
  • Root Statement: The starting point of the workflow logic (can be a single activity, a sequence, or parallel steps).
• Statement: A building block representing a single step or a composite structure. A Statement can contain one of the following:
  • Activity *ActivityInvocation: Represents a call to a Temporal Activity.
  • Sequence *Sequence: Represents a series of statements executed sequentially.
  • Parallel *Parallel: Represents a set of statements executed in parallel.
• Sequence: Contains a slice Elements []*Statement which are executed one after another.
• Parallel: Contains a slice Branches []*Statement which are executed concurrently. The parallel execution waits for all branches to complete. If one branch errors, others are cancelled.
• ActivityInvocation: Defines how to invoke a specific Temporal Activity.
  • Name string: The registered name of the Temporal Activity to call.
  • Arguments []string: A list of variable names (keys from the bindings map) whose values should be passed as arguments to the activity.
  • Result string: The name of the variable (key in the bindings map) where the activity's result should be stored.

Execution Flow (SimpleDSLWorkflow):

• The entry point for workflows defined using this DSL is the SimpleDSLWorkflow function: func SimpleDSLWorkflow(ctx workflow.Context, dslWorkflow Workflow) ([]byte, error).
• It initializes a bindings map from the dslWorkflow.Variables.
• It sets default workflow.ActivityOptions (e.g., StartToCloseTimeout).
• It uses workflow.GetLogger(ctx) for logging.
• It recursively calls the execute method on the Root statement, passing the ctx and bindings map.

Data Flow (bindings):

• The bindings map[string]string acts as the shared state or memory for the workflow execution.
• Initial values come from Workflow.Variables.
• ActivityInvocation.Arguments specifies which values from bindings to use as input for an activity.
• ActivityInvocation.Result specifies the key in bindings where the activity's return value should be stored.
• The makeInput helper function retrieves argument values from the bindings map based on the names listed in ActivityInvocation.Arguments.

Concurrency (Parallel execution):

• The Parallel.execute method uses workflow.Go to launch each branch concurrently.
• It uses workflow.NewSelector and Future.Get to wait for branches to complete.
• workflow.WithCancel is used to cancel pending branches if one branch encounters an error.

Working with the DSL:

• When asked to create or modify workflows, structure the logic using the Workflow, Statement, Sequence, Parallel, and ActivityInvocation types.
• Define the flow of execution by nesting Sequence and Parallel structures within Statements.
• Specify activity calls using ActivityInvocation, ensuring Name, Arguments, and Result are correctly defined based on the available activities and the desired data flow through the bindings map.
• Remember that the actual activity implementation exists separately (see activities.mdc) and is invoked by name.

package dsl

import (
	"time"

	"go.temporal.io/sdk/workflow"
)

type (
	// Workflow is the type used to express the workflow definition. Variables are a map of valuables. Variables can be
	// used as input to Activity.
	Workflow struct {
		Variables map[string]string
		Root      Statement
	}

	// Statement is the building block of dsl workflow. A Statement can be a simple ActivityInvocation or it
	// could be a Sequence or Parallel.
	Statement struct {
		Activity *ActivityInvocation
		Sequence *Sequence
		Parallel *Parallel
	}

	// Sequence consist of a collection of Statements that runs in sequential.
	Sequence struct {
		Elements []*Statement
	}

	// Parallel can be a collection of Statements that runs in parallel.
	Parallel struct {
		Branches []*Statement
	}

	// ActivityInvocation is used to express invoking an Activity. The Arguments defined expected arguments as input to
	// the Activity, the result specify the name of variable that it will store the result as which can then be used as
	// arguments to subsequent ActivityInvocation.
	ActivityInvocation struct {
		Name      string
		Arguments []string
		Result    string
	}

	executable interface {
		execute(ctx workflow.Context, bindings map[string]string) error
	}
)

// SimpleDSLWorkflow workflow definition
func SimpleDSLWorkflow(ctx workflow.Context, dslWorkflow Workflow) ([]byte, error) {
	bindings := make(map[string]string)
	//workflowcheck:ignore Only iterates for building another map
	for k, v := range dslWorkflow.Variables {
		bindings[k] = v
	}

	ao := workflow.ActivityOptions{
		StartToCloseTimeout: 10 * time.Second,
	}
	ctx = workflow.WithActivityOptions(ctx, ao)
	logger := workflow.GetLogger(ctx)

	err := dslWorkflow.Root.execute(ctx, bindings)
	if err != nil {
		logger.Error("DSL Workflow failed.", "Error", err)
		return nil, err
	}

	logger.Info("DSL Workflow completed.")
	return nil, err
}

func (b *Statement) execute(ctx workflow.Context, bindings map[string]string) error {
	if b.Parallel != nil {
		err := b.Parallel.execute(ctx, bindings)
		if err != nil {
			return err
		}
	}
	if b.Sequence != nil {
		err := b.Sequence.execute(ctx, bindings)
		if err != nil {
			return err
		}
	}
	if b.Activity != nil {
		err := b.Activity.execute(ctx, bindings)
		if err != nil {
			return err
		}
	}
	return nil
}

func (a ActivityInvocation) execute(ctx workflow.Context, bindings map[string]string) error {
	inputParam := makeInput(a.Arguments, bindings)
	var result string
	err := workflow.ExecuteActivity(ctx, a.Name, inputParam).Get(ctx, &result)
	if err != nil {
		return err
	}
	if a.Result != "" {
		bindings[a.Result] = result
	}
	return nil
}

func (s Sequence) execute(ctx workflow.Context, bindings map[string]string) error {
	for _, a := range s.Elements {
		err := a.execute(ctx, bindings)
		if err != nil {
			return err
		}
	}
	return nil
}

func (p Parallel) execute(ctx workflow.Context, bindings map[string]string) error {
	//
	// You can use the context passed in to activity as a way to cancel the activity like standard GO way.
	// Cancelling a parent context will cancel all the derived contexts as well.
	//

	// In the parallel block, we want to execute all of them in parallel and wait for all of them.
	// if one activity fails then we want to cancel all the rest of them as well.
	childCtx, cancelHandler := workflow.WithCancel(ctx)
	selector := workflow.NewSelector(ctx)
	var activityErr error
	for _, s := range p.Branches {
		f := executeAsync(s, childCtx, bindings)
		selector.AddFuture(f, func(f workflow.Future) {
			err := f.Get(ctx, nil)
			if err != nil {
				// cancel all pending activities
				cancelHandler()
				activityErr = err
			}
		})
	}

	for i := 0; i < len(p.Branches); i++ {
		selector.Select(ctx) // this will wait for one branch
		if activityErr != nil {
			return activityErr
		}
	}

	return nil
}

func executeAsync(exe executable, ctx workflow.Context, bindings map[string]string) workflow.Future {
	future, settable := workflow.NewFuture(ctx)
	workflow.Go(ctx, func(ctx workflow.Context) {
		err := exe.execute(ctx, bindings)
		settable.Set(nil, err)
	})
	return future
}

func makeInput(argNames []string, argsMap map[string]string) []string {
	var args []string
	for _, arg := range argNames {
		args = append(args, argsMap[arg])
	}
	return args
}

Example Workflows:

Here are some example workflow implementations:

• Serial workflow:

# This sample workflow execute 3 steps in sequence.
# 1) sampleActivity1, takes arg1 as input, and put result as result1.
# 2) sampleActivity2, takes result1 as input, and put result as result2.
# 3) sampleActivity3, takes args2 and result2 as input, and put result as result3.
variables:
  arg1: value1
  arg2: value2

root:
  sequence:
    elements:
     - activity:
        name: SampleActivity1
        arguments:
          - arg1
        result: result1
     - activity:
        name: SampleActivity2
        arguments:
          - result1
        result: result2
     - activity:
        name: SampleActivity3
        arguments:
          - arg2
          - result2
        result: result3

• Parallel workflow:

# This sample workflow execute 3 steps in sequence.
# 1) activity1, takes arg1 as input, and put result as result1.
# 2) it runs a parallel block which runs below sequence branches in parallel
#  2.1) sequence 1
#    2.1.1) activity2, takes result1 as input, and put result as result2
#    2.1.2) activity3, takes arg2 and result2 as input, and put result as result3
#  2.2) sequence 2
#    2.2.1) activity4, takes result1 as input, and put result as result4
#    2.2.2) activity5, takes arg3 and result4 as input, and put result as result5
# 3) activity1, takes result3 and result5 as input, and put result as result6.
variables:
  arg1: value1
  arg2: value2
  arg3: value3

root:
  sequence:
    elements:
      - activity:
         name: SampleActivity1
         arguments:
           - arg1
         result: result1
      - parallel:
          branches:
            - sequence:
                elements:
                 - activity:
                    name: SampleActivity2
                    arguments:
                      - result1
                    result: result2
                 - activity:
                    name: SampleActivity3
                    arguments:
                      - arg2
                      - result2
                    result: result3
            - sequence:
                elements:
                 - activity:
                    name: SampleActivity4
                    arguments:
                      - result1
                    result: result4
                 - activity:
                    name: SampleActivity5
                    arguments:
                      - arg3
                      - result4
                    result: result5
      - activity:
         name: SampleActivity1
         arguments:
           - result3
           - result5
         result: result6