workflow

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
}