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

  Statement

  s.
• 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