typestep

README

⟦𝚲 𝜏 . 𝜏⟧

typestep

type-safe choreography for AWS Step Functions

---

This library provides AWS CDK L3 constructs for defining AWS Step Functions using a type-safe notation in Go.

Inspiration

AWS Step Functions provide an out-of-the-box implementation of the choreography pattern. Their seamless integration with AWS services, especially AWS Lambda, makes them the default choice for AWS-hosted workloads. The main challenge, however, lies in the complexity of workflow specification and it further maintainability. Amazon has developed a domain specific language for definition of the state machine structure—ultimately requiring you to 'code' and testing in JSON. While AWS CDK improves this experience with its L2 constructs, allowing workflow choreography to be defined using general-purpose languages like TypeScript, Go, and others, the process can still be complex.

The biggest challenge is the reliance on duck typing when composing Lambdas into the workflow. A single refactoring mistake in one function can break the entire workflow, with issues only becoming visible at runtime—there is no compile-time inference when composing Lambda A with Lambda B.

typestep is a lightweight library designed to simplify the definition of state machines for AWS Step Functions. By introducing a type-safe notation, it eliminates the challenges of duck typing and ensures compile-time inference of AWS Lambda signatures. This approach enhances reliability, making workflow choreography easier to define, maintain, and refactor while reducing runtime errors.

Getting Starter

• Inspiration
• Getting Starter
  • Quick example
  • Type-safe annotation of AWS Lambda
  • Workflow composition
    • Form sources events
    • Join composes functions
    • Lift, Wrap and Unit builds nested computations
    • Yield the results
• How To Contribute
• License

The latest version of the library is available at main branch of this repository. All development, including new features and bug fixes, take place on the main branch using forking and pull requests as described in contribution guidelines. The stable version is available via Golang modules.

Use go get to retrieve the library and add it as dependency to your application.

go get -u github.com/fogfish/typestep

Quick example

Example below is most simplest illustration on how to make a type-safe composition of lambda function into AWS Step Function workflow.

a := typestep.From[string](
  awsevents.EventBus_FromEventBusArn(/* ... */),
)

b := typestep.Join(
  func(name string) (string, error) { /* ... */ },
  awslambda.Function_FromFunctionArn(/* ... */),
  a,
)

c := typestep.ToQueue(
  awssqs.Queue_FromQueueArn(/* ... */),
  b,
)

workflow := typestep.NewTypeStep(stack, jsii.String("Workflow"),
  &typestep.TypeStepProps{},
)
typestep.StateMachine(workflow, c)

More detailed examples are here

Type-safe annotation of AWS Lambda

AWS Lambda does not impose restrictions on the development runtime, allowing the use of type-safe languages like Go. However, outside of the function itself, type safety is not enforced by the AWS environment, as Lambda relies on JSON for input and output handling. As a consequence the duck typing is used when composing Lambdas. When building Go-based workflows, Lambdas must be lifted into a type-safe abstraction. Since Lambda is merely a deployment pattern, it is recommended to define workflow functions within the core domain and reference them in both the Lambda configuration and infrastructure-as-code (IaC) definitions.

// app/internal/core/biz.go
func GetUser(acc Account) (User, error) { /* ... */ } 

// app/cmd/lambda/main.go
func main() { lambda.Start(core.GetUser) }

// app/internal/cdk/workflow.go

// declares AWS Lambda resource
f := scud.NewFunctionGo(stack, jsii.String("Lambda"),
  &scud.FunctionGoProps{
    SourceCodeModule: "github.com/fogfish/app",
    SourceCodeLambda: "cmd/lambda",
  },
)

// annotates AWS Lambda with type-safe signature inside the workflow
typestep.Join(core.GetUser, f, /* ... */)

This technique allows validation of function signatures at compile time.

Workflow composition

The library uses category-theory-inspired algebra defined here to compose workflows. Its algebra is tailored for effective composition of ƒ: A ⟼ B and ƒ: A ⟼ []B types of computations.

The workflow is triggered by the AWS EventBridge event and passes through a series of transformations defined by AWS Lambda functions. The results are then either emitted back to AWS EventBridge or sent to an AWS SQS queue. Any errors encountered during execution are captured in a dead-letter queue (AWS SQS) for further analysis. The library does not provide L3 constructs for provisioning AWS EventBridge, Lambda, or SQS. Its sole focus is on defining AWS Step Functions and their state machines.

The library provide simple api for the workflow composition: From, Join, Lift, Wrap, Unit and Yeild.

Once the workflow is composed, deploy it using TypeStep L3 construct:

// Declare the workflow
a := typestep.From[core.Account](input)
// ...
f := typestep.ToQueue(reply, e)

// Deploy the workflow
ts := typestep.NewTypeStep(stack, jsii.String("Pipe"), &typestep.TypeStepProps{})
typestep.StateMachine(ts, f)

Form sources events

From binds EventBridge to an AWS Step Function, automatically configuring the consumption of all events where detail-type matches the specified type name. For example, in the snippet below, all events with detail-type set to Account will trigger the computation.

bus := awsevents.NewEventBus(/* ... */)
// ...
a := typestep.From[core.Account](bus)

Join composes functions

The simple operation above returns a workflow definition that represents an identity function ƒ: Account ⟼ Account. It can be further composed with any function of type 𝑔: Account ⟼ ?, using Join.

func GetUser(Account) (User, error) { /* ... */ }

fun := awslambda.NewFunction(/* ... */)

b := typestep.Join(GetUser, fun, a)

Lift, Wrap and Unit builds nested computations

If your first function returns a list (ƒ: A ⟼ []B) and needs to be composed with 𝑔: B ⟼ C, you must lift the computation to ensure proper composition.

func GetManyB(A) ([]B, error) { /* ... */ }
func UseJustB(B) (C, error) { /* ... */ }

b := typestep.Join(GetManyB, /* ... */)
c := typestep.Lift(UseJustB, /* ... */, b)
// ... 
x := typestep.Unit(/* ... */)

In the functional programming, this abstraction is called "free monad". We are lifting the function UseJustB within "a functorial context"--[]B list. It is a responsibility of creator of such op to do something with those nested stucts either yielding individual elements (ToQueue, ToEventBus) or uniting it. Think about it as the following construct.

for _, b := range GetManyB() {
  UseJustB(b)
}

Yield the results

The workflow completes by emitting an event to AWS SQS or EventBridge, unless explicitly persisted elsewhere through a chained AWS Lambda function.

x := typestep.ToQueue(/* ... */)

How To Contribute

The library is MIT licensed and accepts contributions via GitHub pull requests:

1. Fork it
2. Create your feature branch (git checkout -b my-new-feature)
3. Commit your changes (git commit -am 'Added some feature')
4. Push to the branch (git push origin my-new-feature)
5. Create new Pull Request

The build and testing process requires Go version 1.24 or later.

Build and run in your development console.

git clone https://github.com/fogfish/typestep
go test ./...

License

Documentation

Index

• func From[A any](in awsevents.IEventBus) duct.Morphism[A, A]
• func Join[A, B, C any](_ func(context.Context, B) (C, error), f awslambda.IFunction, ...) duct.Morphism[A, C]
• func Lift[A, B, C any](_ func(context.Context, B) (C, error), f awslambda.IFunction, ...) duct.Morphism[A, C]
• func LiftP[A, B, C any](n int, _ func(context.Context, B) (C, error), f awslambda.IFunction, ...) duct.Morphism[A, C]
• func StateMachine[A, B any](ts TypeStep, m duct.Morphism[A, B])
• func ToEventBus[A, B any](source string, bus awsevents.IEventBus, m duct.Morphism[A, B]) duct.Morphism[A, duct.Void]
• func ToQueue[A, B any](q awssqs.IQueue, m duct.Morphism[A, B]) duct.Morphism[A, duct.Void]
• func Unit[A, B any](m duct.Morphism[A, B]) duct.Morphism[A, []B]
• func Wrap[A, B any](m duct.Morphism[A, []B]) duct.Morphism[A, B]
• type TypeStep
  • func NewTypeStep(scope constructs.Construct, id *string, props *TypeStepProps) TypeStep
• type TypeStepProps

Functions

func From

func From[A any](in awsevents.IEventBus) duct.Morphism[A, A]

Creates new morphism 𝑚, binding it with EventBridge for reading category `A` events.

func Join

func Join[A, B, C any](
	_ func(context.Context, B) (C, error),
	f awslambda.IFunction,
	m duct.Morphism[A, B],
) duct.Morphism[A, C]

Compose lambda function transformer 𝑓: B ⟼ C with morphism 𝑚: A ⟼ B producing a new morphism 𝑚: A ⟼ C.

func Lift

func Lift[A, B, C any](
	_ func(context.Context, B) (C, error),
	f awslambda.IFunction,
	m duct.Morphism[A, []B],
) duct.Morphism[A, C]

Compose lambda function transformer 𝑓: B ⟼ C with morphism 𝑚: A ⟼ []B. It produces a new computation 𝑚: A ⟼ []C that enables transformation within `[]C` context without immediate collapsing (see duct.LiftF for details). In other words, it nests the computation within the slice context. It is a responsibility of lifter to do something with those nested contexts either yielding individual elements or uniting (e.g. use Unit(Join(g, Lift(f))) to leave nested context into the morphism 𝑚: A ⟼ []C).

func LiftP

func LiftP[A, B, C any](
	n int,
	_ func(context.Context, B) (C, error),
	f awslambda.IFunction,
	m duct.Morphism[A, []B],
) duct.Morphism[A, C]

See Lift for details. The function LiftP is equivalent to Lift but allows to specify the maximum number of concurrent invocations of the lambda function.

func StateMachine

func StateMachine[A, B any](ts TypeStep, m duct.Morphism[A, B])

StateMachine injects the morphism into the AWS Step Function, it constructs the state machine from the defined computation.

func ToEventBus

func ToEventBus[A, B any](source string, bus awsevents.IEventBus, m duct.Morphism[A, B]) duct.Morphism[A, duct.Void]

Yield results of 𝑚: A ⟼ B binding it with AWS EventBridge.

func ToQueue

func ToQueue[A, B any](q awssqs.IQueue, m duct.Morphism[A, B]) duct.Morphism[A, duct.Void]

Yield results of 𝑚: A ⟼ B binding it with AWS SQS.

func Unit

func Unit[A, B any](m duct.Morphism[A, B]) duct.Morphism[A, []B]

Unit finalizes a transformation context by collapsing the nested morphism. It acts as the terminal operation, ensuring that all staged compositions, such as those built with Lift and Wrap, are fully resolved into a single, consumable form.

func Wrap

func Wrap[A, B any](m duct.Morphism[A, []B]) duct.Morphism[A, B]

Wrap is equivalent to Lift but operates directly on the inner structure of the morphism 𝑚: A ⟼ []B, extracting individual elements of B while preserving the transformation context, enabling further composition.

Usable to Yield elements of []B without transformation

Types

type TypeStep

type TypeStep interface {
	constructs.IConstruct
}

TypeStep is AWS CDK L3, a builder for AWS Step Function state machine.

func NewTypeStep

func NewTypeStep(scope constructs.Construct, id *string, props *TypeStepProps) TypeStep

Create a new instance of TypeStep construct

type TypeStepProps

type TypeStepProps struct {
	// DeadLetterQueue is the queue to receive messages to if an error occurs
	// while running the computation. The message is input JSON and "error".
	DeadLetterQueue awssqs.IQueue

	// SeqConcurrency is the maximum number of lambda's invocations allowed for
	// itterators while processing the sequence of computations (morphism 𝑚: A ⟼ []B).
	SeqConcurrency *float64
}

TypeStep L3 construct properties