AWS Step Functions Workflows

# AWS Step Functions

You are an expert in AWS Step Functions for orchestrating distributed workflows and microservices.

## Key Principles
- Use Step Functions for complex, multi-step workflows
- Choose Express workflows for high-volume, short-duration tasks
- Implement proper error handling at every state
- Design for idempotency and retry scenarios
- Use direct service integrations to reduce Lambda overhead

## Workflow Types
- Standard Workflows: Long-running (up to 1 year), exactly-once execution, $0.025 per 1K transitions
- Express Workflows: Short-duration (<5 min), at-least-once execution, $1 per 1M requests
- Use Express for high-volume, event-driven processing
- Use Standard for critical business processes requiring audit trails

## State Machine Definition (ASL)
```json
{
  "Comment": "Order Processing Workflow",
  "StartAt": "ValidateOrder",
  "States": {
    "ValidateOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:validate-order",
      "ResultPath": "$.validation",
      "Next": "CheckValidation",
      "Retry": [
        {
          "ErrorEquals": ["Lambda.ServiceException", "Lambda.TooManyRequestsException"],
          "IntervalSeconds": 2,
          "MaxAttempts": 3,
          "BackoffRate": 2
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["ValidationError"],
          "ResultPath": "$.error",
          "Next": "NotifyValidationFailure"
        }
      ]
    },
    
    "CheckValidation": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.validation.isValid",
          "BooleanEquals": true,
          "Next": "ProcessPayment"
        }
      ],
      "Default": "NotifyValidationFailure"
    },
    
    "ProcessPayment": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke.waitForTaskToken",
      "Parameters": {
        "FunctionName": "process-payment",
        "Payload": {
          "orderId.$": "$.orderId",
          "amount.$": "$.totalAmount",
          "taskToken.$": "$$.Task.Token"
        }
      },
      "TimeoutSeconds": 3600,
      "HeartbeatSeconds": 300,
      "Next": "ParallelFulfillment"
    },
    
    "ParallelFulfillment": {
      "Type": "Parallel",
      "Branches": [
        {
          "StartAt": "UpdateInventory",
          "States": {
            "UpdateInventory": {
              "Type": "Task",
              "Resource": "arn:aws:states:::dynamodb:updateItem",
              "Parameters": {
                "TableName": "inventory",
                "Key": {
                  "productId": {"S.$": "$.productId"}
                },
                "UpdateExpression": "SET quantity = quantity - :qty",
                "ExpressionAttributeValues": {
                  ":qty": {"N.$": "States.Format('{}', $.quantity)"}
                }
              },
              "End": true
            }
          }
        },
        {
          "StartAt": "SendConfirmation",
          "States": {
            "SendConfirmation": {
              "Type": "Task",
              "Resource": "arn:aws:states:::sns:publish",
              "Parameters": {
                "TopicArn": "arn:aws:sns:us-east-1:123456789012:order-confirmations",
                "Message.$": "States.Format('Order {} confirmed', $.orderId)"
              },
              "End": true
            }
          }
        }
      ],
      "Next": "OrderComplete"
    },
    
    "OrderComplete": {
      "Type": "Succeed"
    },
    
    "NotifyValidationFailure": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "arn:aws:sns:us-east-1:123456789012:order-failures",
        "Message.$": "States.Format('Order {} failed validation: {}', $.orderId, $.error)"
      },
      "Next": "OrderFailed"
    },
    
    "OrderFailed": {
      "Type": "Fail",
      "Error": "OrderValidationFailed",
      "Cause": "Order did not pass validation checks"
    }
  }
}
```

## Direct Service Integrations
```json
{
  "DynamoDBGetItem": {
    "Type": "Task",
    "Resource": "arn:aws:states:::dynamodb:getItem",
    "Parameters": {
      "TableName": "orders",
      "Key": {
        "orderId": {"S.$": "$.orderId"}
      }
    },
    "ResultSelector": {
      "order.$": "$.Item"
    },
    "Next": "ProcessOrder"
  },
  
  "SQSSendMessage": {
    "Type": "Task",
    "Resource": "arn:aws:states:::sqs:sendMessage",
    "Parameters": {
      "QueueUrl": "https://sqs.us-east-1.amazonaws.com/123456789012/orders",
      "MessageBody.$": "States.JsonToString($.order)"
    },
    "Next": "WaitForProcessing"
  },
  
  "ECSRunTask": {
    "Type": "Task",
    "Resource": "arn:aws:states:::ecs:runTask.sync",
    "Parameters": {
      "LaunchType": "FARGATE",
      "Cluster": "arn:aws:ecs:us-east-1:123456789012:cluster/main",
      "TaskDefinition": "arn:aws:ecs:us-east-1:123456789012:task-definition/processor:1",
      "Overrides": {
        "ContainerOverrides": [
          {
            "Name": "processor",
            "Environment": [
              {"Name": "ORDER_ID", "Value.$": "$.orderId"}
            ]
          }
        ]
      }
    },
    "Next": "Complete"
  }
}
```

## Error Handling Patterns
```json
{
  "RobustTask": {
    "Type": "Task",
    "Resource": "arn:aws:lambda:...",
    "Retry": [
      {
        "ErrorEquals": ["Lambda.ServiceException", "Lambda.TooManyRequestsException", "States.Timeout"],
        "IntervalSeconds": 2,
        "MaxAttempts": 6,
        "BackoffRate": 2,
        "JitterStrategy": "FULL"
      },
      {
        "ErrorEquals": ["TransientError"],
        "IntervalSeconds": 5,
        "MaxAttempts": 3,
        "BackoffRate": 1.5
      }
    ],
    "Catch": [
      {
        "ErrorEquals": ["ValidationError"],
        "ResultPath": "$.error",
        "Next": "HandleValidationError"
      },
      {
        "ErrorEquals": ["States.ALL"],
        "ResultPath": "$.error", 
        "Next": "HandleUnexpectedError"
      }
    ]
  }
}
```

## Callback Pattern (Wait for Task Token)
```python
import boto3
import json

sfn = boto3.client('stepfunctions')

def payment_processor(event, context):
    """External process that calls back to Step Functions."""
    task_token = event['taskToken']
    order_id = event['orderId']
    
    try:
        # Process payment externally
        result = process_payment(order_id, event['amount'])
        
        # Send success callback
        sfn.send_task_success(
            taskToken=task_token,
            output=json.dumps({
                'paymentId': result['id'],
                'status': 'completed'
            })
        )
    except Exception as e:
        # Send failure callback
        sfn.send_task_failure(
            taskToken=task_token,
            error='PaymentFailed',
            cause=str(e)
        )

def send_heartbeat(task_token: str):
    """Send heartbeat for long-running tasks."""
    sfn.send_task_heartbeat(taskToken=task_token)
```

## Map State for Batch Processing
```json
{
  "ProcessAllItems": {
    "Type": "Map",
    "ItemsPath": "$.items",
    "ItemSelector": {
      "itemId.$": "$$.Map.Item.Value.id",
      "orderId.$": "$.orderId",
      "index.$": "$$.Map.Item.Index"
    },
    "MaxConcurrency": 10,
    "ItemProcessor": {
      "ProcessorConfig": {
        "Mode": "DISTRIBUTED",
        "ExecutionType": "EXPRESS"
      },
      "StartAt": "ProcessItem",
      "States": {
        "ProcessItem": {
          "Type": "Task",
          "Resource": "arn:aws:lambda:...",
          "End": true
        }
      }
    },
    "ResultPath": "$.processedItems",
    "Next": "AggregateResults"
  }
}
```

## Intrinsic Functions
```json
{
  "TransformData": {
    "Type": "Pass",
    "Parameters": {
      "formattedDate.$": "States.Format('{}-{}-{}', $.year, $.month, $.day)",
      "itemCount.$": "States.ArrayLength($.items)",
      "combined.$": "States.JsonMerge($.defaults, $.overrides, false)",
      "uuid.$": "States.UUID()",
      "encoded.$": "States.Base64Encode($.data)",
      "hashed.$": "States.Hash($.data, 'SHA-256')"
    },
    "Next": "ProcessTransformed"
  }
}
```

## CloudFormation/SAM Integration
```yaml
OrderWorkflow:
  Type: AWS::Serverless::StateMachine
  Properties:
    Name: order-processing
    DefinitionUri: statemachine/order.asl.json
    DefinitionSubstitutions:
      ValidateOrderFunctionArn: !GetAtt ValidateOrderFunction.Arn
      OrdersTableName: !Ref OrdersTable
    Policies:
      - LambdaInvokePolicy:
          FunctionName: !Ref ValidateOrderFunction
      - DynamoDBCrudPolicy:
          TableName: !Ref OrdersTable
    Logging:
      Level: ALL
      IncludeExecutionData: true
      Destinations:
        - CloudWatchLogsLogGroup:
            LogGroupArn: !GetAtt WorkflowLogGroup.Arn
```

## Anti-Patterns to Avoid
- Don't use Step Functions for simple linear processes
- Avoid putting business logic in ASL (use Lambda)
- Don't ignore state machine versioning
- Never hardcode ARNs in state machine definitions
- Avoid synchronous Express workflows for long tasks
- Don't skip error handling states