StateMachine

Source: src/AWS/StepFunctions/StateMachine.ts

An AWS Step Functions state machine (workflow).

StateMachine owns the lifecycle of a STANDARD or EXPRESS workflow. The Amazon States Language definition may be provided as a plain object — Output values (like Lambda function ARNs) inside it are resolved before serialization — and an execution role is created automatically unless an explicit roleArn is given. Lambda functions referenced in the definition are granted lambda:InvokeFunction on the auto-created role.

Creating State Machines

Standard Workflow with a Pass State

import * as StepFunctions from "alchemy/AWS/StepFunctions";

const machine = yield* StepFunctions.StateMachine("OrderWorkflow", {
  definition: {
    StartAt: "Done",
    States: {
      Done: { Type: "Pass", End: true },
    },
  },
});

Express Workflow

const machine = yield* StepFunctions.StateMachine("FastWorkflow", {
  type: "EXPRESS",
  definition: {
    StartAt: "Echo",
    States: {
      Echo: { Type: "Pass", End: true },
    },
  },
});

Orchestrating Lambda Functions

Reference a function ARN in a Task state — lambda:InvokeFunction is granted on the auto-created execution role automatically.

const machine = yield* StepFunctions.StateMachine("Pipeline", {
  definition: {
    StartAt: "Process",
    States: {
      Process: {
        Type: "Task",
        Resource: fn.functionArn,
        End: true,
      },
    },
  },
});

Service Integrations

const machine = yield* StepFunctions.StateMachine("Callback", {
  definition: {
    StartAt: "WaitForApproval",
    States: {
      WaitForApproval: {
        Type: "Task",
        Resource: "arn:aws:states:::sqs:sendMessage.waitForTaskToken",
        Parameters: {
          QueueUrl: queue.queueUrl,
          MessageBody: { "token.$": "$$.Task.Token" },
        },
        End: true,
      },
    },
  },
  policyStatements: [
    {
      Effect: "Allow",
      Action: ["sqs:SendMessage"],
      Resource: [queue.queueArn],
    },
  ],
});

Starting Executions at Runtime

Bind execution operations in the init phase and use them in runtime handlers.

Start a workflow from a handler

// init
const startExecution = yield* StepFunctions.StartExecution(machine);

return {
  fetch: Effect.gen(function* () {
    // runtime
    const execution = yield* startExecution({
      input: JSON.stringify({ orderId: "123" }),
    });
    return HttpServerResponse.json({ executionArn: execution.executionArn });
  }),
};

Run an EXPRESS workflow synchronously

// init
const startSyncExecution = yield* StepFunctions.StartSyncExecution(machine);

// runtime
const result = yield* startSyncExecution({
  input: JSON.stringify({ value: 21 }),
});
// result.status === "SUCCEEDED", result.output is the workflow output

Typed Programs

Author the workflow as a typed Sfn program (mirroring Effect’s names — Sfn.gen, Sfn.invoke, Sfn.when, Sfn.forEach, Sfn.catchTag, …) and compile it with StateMachine.fromProgram. The compiler emits a plain ASL definition plus the IAM policy statements its task states need; the raw definition path above stays fully usable underneath.

import { Sfn, StateMachine } from "alchemy/AWS/StepFunctions";

const machine = yield* StateMachine.fromProgram("OrderWorkflow", {
  type: "EXPRESS",
  program: Sfn.gen(function* (input: Sfn.Expr<{ value: number }>) {
    const result = yield* Sfn.invoke<{ doubled: number }>(doubler, {
      value: input.value,
    });
    const size = yield* Sfn.when(
      Sfn.gt(result.doubled, 10),
      Sfn.succeed("big"),
      Sfn.succeed("small"),
    );
    return { doubled: result.doubled, size };
  }),
});