state-machine.asl.yml

QueryLanguage: JSONPath
StartAt: Normalize Input
States:
  # Start of pre-processing
  Normalize Input:
    # Sample input:
    #     { Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […] } }
    # Sample output:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "DONE"
    #     }
    Type: Task
    QueryLanguage: JSONPath
    Comment: Normalizes execution input
    Retry:
      - { ErrorEquals: [MissingTaskTypeError], MaxAttempts: 0 } # Don't retry
      - { ErrorEquals: [States.ALL], IntervalSeconds: 1, MaxAttempts: 2, BackoffRate: 1 }
    Catch:
      # Add the error object to the state input before passing to the catcher
      - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: Normalize Input Catcher }

    Resource: "${NormalizeInputLambdaFunctionArn}"
    Parameters: # Payload provided to the Lambda function
      Input.$: "$"

    # Proceed down the happy path
    Next: JobReceived Callbacks Map
  Normalize Input Catcher:
    # Sample input
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […] },
    #       Error: {…}
    #     }
    # Sample output:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […] },
    #       Error: {…},
    #       State: NORMALIZE_INPUT_ERROR
    #     }
    Type: Pass
    QueryLanguage: JSONPath
    Comment: Add a field to the state so normalize errors can be identified

    # Set the output State to a static string value
    Result: NORMALIZE_INPUT_ERROR
    ResultPath: "$.State"

    # Skip to post-processing
    Next: Add Empty TasksMapResults

  JobReceived Callbacks Map:
    # Sample input:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "DONE"
    #     }
    # Sample output:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "DONE",
    #       JobReceivedMapResultNulls: [null, null]
    #     }
    Type: Map
    QueryLanguage: JSONPath
    Comment: Iterates over all callback endpoints to indicate the job was received
    # Retries are handled directly on the callback state in the iterator
    # If callbacks fail, there's no point in having a fallback catcher, since we would only use that to send callbacks

    # Add the result of the Map (an array containing the results of each iteration) to the default output
    # (This field will not be shown in future sample state data)
    ResultPath: "$.JobReceivedMapResultNulls"

    # Proceed down the happy path
    Next: Ingest Source File

    MaxConcurrency: 0
    ItemsPath: "$.Job.Callbacks" # This references post-InputPath input
    ItemSelector: # Initial state for each ItemProcessor iteration
      Callback.$: "$$.Map.Item.Value" # Reference the current item being processed from the array indicated by ItemsPath
      StateMachine:
        Id.$: "$$.StateMachine.Id"
      Execution:
        Id.$: "$$.Execution.Id"
      Message: # The callbacks Lambda function uses the JSON of this value as the message sent to each callback
        JobReceived: # The `JobReceived` field MUST be included in these callback messages per our published API
          Job:
            Id.$: "$.Job.Id"
          Execution:
            Id.$: "$$.Execution.Id"
          State: RECEIVED
    ItemProcessor:
      ProcessorConfig: { Mode: INLINE }
      StartAt: Send JobReceived Callback
      States:
        Send JobReceived Callback:
          # Sample input:
          #     {
          #       Callback: {…},
          #       StateMachine: { Id: "wxyz" },
          #       Execution: { Id: "a1b2c3" },
          #       Message: {…}
          #     }
          # Sample output: (will become an element in JobReceivedMapResultNulls)
          #     null
          Type: Task
          QueryLanguage: JSONPath
          Comment: >-
            Sends a job received message for a single callback endpoint in
            the iterator
          Resource: "${CallbackLambdaFunctionArn}"
          Retry:
            - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }

          End: true

  Ingest Source File:
    # Sample input:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "DONE"
    #     }
    # Sample output:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "DONE",
    #       Artifact: {…}
    #     }
    Type: Task
    QueryLanguage: JSONPath
    Comment: Creates an artifact in S3 of the source file
    Retry:
      - { ErrorEquals: [UnknownSourceModeError], MaxAttempts: 0 } # Don't retry
      - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }
    Catch:
      # Add the error object to the state input before passing to the catcher
      - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: Ingest Source File Catcher }

    Resource: "${IngestLambdaFunctionArn}"
    Parameters: # Payload provided to the Lambda function
      Job:
        Source.$: "$.Job.Source"
      Execution:
        Id.$: "$$.Execution.Id"

    # Add the return value of the Lambda function to the default state output
    ResultPath: "$.Artifact"

    # Proceed down the happy path
    Next: Detect Source File Type
  Ingest Source File Catcher:
    # Sample input:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "DONE",
    #       Error: {}
    #     }
    # Sample output:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: SOURCE_FILE_INGEST_ERROR,
    #       Error: {}
    #     }
    Type: Pass
    QueryLanguage: JSONPath
    Comment: Add a field to the state so ingest errors can be identified

    # Set the output State to a static string value
    Result: SOURCE_FILE_INGEST_ERROR
    ResultPath: "$.State"

    # Skip to post-processing
    Next: Add Empty TasksMapResults

  Detect Source File Type:
    # Sample input:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "DONE",
    #       Artifact: {…}
    #     }
    # Sample output:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "DONE",
    #       Artifact: { …, Descriptor: {…} }
    #     }
    Comment: Detects the file type of the source file artifact
    Type: Task
    QueryLanguage: JSONPath
    Retry:
      - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }
    Catch:
      # Add the error object to the state input before passing to the catcher
      - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: Detect Source File Type Catcher }

    Resource: "${SourceTypeLambdaFunctionArn}"
    Parameters: # Payload provided to the Lambda function
      Artifact.$: "$.Artifact"

    # Add the return value of the Lambda function to the default state output
    ResultPath: "$.Artifact.Descriptor"

    # Proceed down the happy path
    Next: Tasks Map
  Detect Source File Type Catcher:
    # Sample input:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "DONE",
    #       Artifact: {…},
    #       Error: {…}
    #     }
    # Sample output:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "SOURCE_FILE_TYPE_DETECTION_ERROR",
    #       Artifact: {…},
    #       Error: {…}
    #     }
    Type: Pass
    QueryLanguage: JSONPath
    Comment: Add a field to the state so type detection errors can be identified

    # Set the output State to a static string value
    Result: SOURCE_FILE_TYPE_DETECTION_ERROR
    ResultPath: "$.State"

    # Skip to post-processing
    Next: Add Empty TasksMapResults
  # End of pre-processing

  # Start of task execution
  Tasks Map:
    # The `TasksMapResults` field in the output will be an array, where each element is the final output of an iteration of
    # the map. For example, if a job has three tasks, `TasksMapResults` will be an array with three elements, regardless of
    # whether those tasks succeeded or failed.
    #
    # For successful tasks (those that aren't caught for any reason), the element in the array will be an object with
    # the task's results, like { Task: "___SOME_TASK__TYPE___", …other task-specifc fields }.
    #
    # For unsuccessful tasks (those caught for any reason), the element will be the original task definition from the
    # job, such as { Type: "___SOME_TASK__TYPE___" }.
    #
    # Note the distinction between the `Task` and `Type` fields in the two different cases.
    #
    # Sample input:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "DONE",
    #       Artifact: { Descriptor: {…} }
    #     }
    # Sample output:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "DONE",
    #       Artifact: { Descriptor: {…} },
    #       TasksMapResults: […]
    #     }
    Type: Map
    QueryLanguage: JSONPath
    Comment: Iterates over all tasks included in the job
    Catch:
      # Add the error object to the state input before passing to the catcher
      - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: Tasks Map Catcher }

    # Add the result of the Map (an array containing the results of each iteration) to the default output
    ResultPath: "$.TasksMapResults"

    # Proceed down the happy path
    Next: JobResult Callback Map

    MaxConcurrency: 0
    ItemsPath: "$.Job.Tasks" # This references post-InputPath input
    ItemSelector: # Initial state for each ItemProcessor iteration
      Job:
        Id.$: "$.Job.Id"
      Execution:
        Id.$: "$$.Execution.Id"
      Artifact.$: "$.Artifact"
      TaskIndex.$: "$$.Map.Item.Index"
      Task.$: "$$.Map.Item.Value" # This is the current iteration's element from Job.Tasks, e.g., { "Type": "Copy" }
      Callbacks.$: "$.Job.Callbacks"
    ItemProcessor:
      ProcessorConfig: { Mode: INLINE }
      StartAt: Route Task By Type
      States:
        # The general shape of a task map iteration is:
        #   1. Route the task, based on Task.Type
        #   2. One or more task-specific states that do the work
        #   3. Send callbacks with the task results
        # One of two callback states will send callbacks, depending on if the
        # task was successful. Because these callback states are the final
        # states of the iteration, their output is important: it will be what
        # ends up in the map's ResultPath ($.TasksMapResults).
        #
        # All task-specific states, not just the final state, should catch
        # to the TaskResult Error Callback Map.
        #
        # The ResultPath for states that are returning the final output of a
        # task's operation (i.e., states immediately prior to the callbacks
        # map) must be $.TaskResult, and their return value must be the
        # expected standard task output (e.g., { Task: 'Copy' })

        Route Task By Type:
          # Sample input/output: See map's ItemSelector
          Type: Choice
          QueryLanguage: JSONPath
          Comment: >-
            For the current task being iterated, routes to the appropriate
            state

          Default: Unknown Task Type
          Choices:
            - Variable: "$.Task.Type"
              StringEquals: Inspect
              Next: Inspect Task Handler
            - And:
                - Variable: "$.Task.Type"
                  StringEquals: Copy
                - Variable: "$.Task.Mode"
                  StringEquals: AWS/S3
              Next: Copy Task Handler
            - And:
                - Variable: "$.Task.Type"
                  StringEquals: Copy
                - Variable: "$.Task.Mode"
                  StringEquals: FTP/Passive
              Next: FTP Copy Task Fargate Execution
            - And:
                - Variable: "$.Task.Type"
                  StringEquals: Copy
                - Variable: "$.Task.Mode"
                  StringEquals: FTP/Active
              Next: FTP Copy Task Fargate Execution
            - And:
                - Variable: "$.Task.Type"
                  StringEquals: Copy
                - Variable: "$.Task.Mode"
                  StringEquals: FTP/Auto
              Next: FTP Copy Task Fargate Execution
            - Variable: "$.Task.Type"
              StringEquals: Transcode
              Next: Transcode Task Fargate Execution
            - Variable: "$.Task.Type"
              StringEquals: HLS
              Next: HLS Task Fargate Execution
            - Variable: "$.Task.Type"
              StringEquals: Image
              Next: Image Task Handler
            - Variable: "$.Task.Type"
              StringEquals: Transcribe
              Next: Transcribe Task Job Start
            - Variable: "$.Task.Type"
              StringEquals: WavWrap
              Next: WAV Wrap Task Handler
            - Variable: "$.Task.Type"
              StringEquals: DetectSilence
              Next: Detect Silence Task Handler
            - Variable: "$.Task.Type"
              StringEquals: DetectTone
              Next: Detect Tone Task Handler
            - Variable: "$.Task.Type"
              StringEquals: DetectEas
              Next: Detect EAS Task Handler
            - Variable: "$.Task.Type"
              StringEquals: Waveform
              Next: Waveform Task Handler

        # This is a no-op. No callbacks are sent, and any record of this task
        # will be filtered out of the job result callbacks
        Unknown Task Type:
          # Sample output:
          #     { Task: "UnknownTaskType" }
          Type: Pass
          QueryLanguage: JSONPath
          Comment: Handles and blackholes tasks with an unknown type

          # Even though we do no work for this task, we still return a
          # reasonably-well-formed result from the ItemProcessor, which will
          # appear in the Map state's results array.
          Result:
            Task: "UnknownTaskType"

          End: true

        Inspect Task Handler:
          # Sample input: See map's ItemSelector
          # Sample output: ItemSelector with TaskResult
          Type: Task
          QueryLanguage: JSONPath
          Comment: Inspects the artifact file
          Retry:
            - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }
          Catch:
            - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: TaskResult Error Callback Map }

          Resource: "${InspectMediaLambdaFunctionArn}"
          Parameters: # Payload provided to the Lambda function
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"

          # Add the return value of the Lambda function to the default state output
          ResultPath: "$.TaskResult"

          Next: TaskResult Callbacks Map

        Copy Task Handler:
          # Sample input: See map's ItemSelector
          # Sample output: ItemSelector with TaskResult
          Type: Task
          QueryLanguage: JSONPath
          Comment: Copies the artifact
          Retry:
            - { ErrorEquals: [UnknownCopyTaskModeError], MaxAttempts: 0 }
            - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }
          Catch:
            - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: TaskResult Error Callback Map }

          Resource: "${CopyLambdaFunctionArn}"
          Parameters: # Payload provided to the Lambda function
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"

          # Add the return value of the Lambda function to the default state output
          ResultPath: "$.TaskResult"

          Next: TaskResult Callbacks Map

        FTP Copy Task Fargate Execution:
          # Sample input: See map's ItemSelector
          # Sample output: ItemSelector with TaskResult
          Type: Task
          QueryLanguage: JSONPath
          Comment: Sends artifact to an FTP destination
          Retry:
            - { ErrorEquals: [States.ALL], IntervalSeconds: 10, MaxAttempts: 3, BackoffRate: 2 }
          Catch:
            - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: TaskResult Error Callback Map }

          # Add the return value of the Fargate task to the default state output
          ResultPath: "$.TaskResult"

          Next: TaskResult Callbacks Map

          Resource: arn:aws:states:::ecs:runTask.waitForTaskToken
          Parameters: # Fargate RunTask request parameters
            Cluster: "${EcsClusterArn}"
            LaunchType: FARGATE
            Overrides:
              ContainerOverrides:
                - Environment:
                    - Name: STATE_MACHINE_ARN
                      Value.$: "$$.StateMachine.Id"
                    - Name: STATE_MACHINE_NAME
                      Value.$: "$$.StateMachine.Name"
                    - Name: STATE_MACHINE_EXECUTION_ID
                      Value.$: "$$.Execution.Id"
                    - Name: STATE_MACHINE_JOB_ID
                      Value.$: "$.Job.Id"
                    - Name: STATE_MACHINE_TASK_INDEX
                      Value.$: States.Format('{}', $.TaskIndex)
                    - Name: STATE_MACHINE_AWS_REGION
                      Value: "${AwsRegion}"
                    - Name: STATE_MACHINE_ARTIFACT_BUCKET_NAME
                      Value.$: "$.Artifact.BucketName"
                    - Name: STATE_MACHINE_ARTIFACT_OBJECT_KEY
                      Value.$: "$.Artifact.ObjectKey"
                    - Name: STATE_MACHINE_TASK_JSON
                      Value.$: States.JsonToString($.Task)
                    - Name: STATE_MACHINE_TASK_TOKEN
                      Value.$: "$$.Task.Token"
                    - Name: STATE_MACHINE_TASK_TYPE
                      Value.$: "$.Task.Type"
                  Name: "${FtpCopyContainerName}"
            NetworkConfiguration:
              AwsvpcConfiguration:
                AssignPublicIp: ENABLED
                SecurityGroups:
                  - "${FtpCopyEcsTaskSecurityGroupId}"
                Subnets:
                  - "${VpcPublicSubnet1}"
                  - "${VpcPublicSubnet2}"
            PropagateTags: TASK_DEFINITION
            TaskDefinition: "${FtpCopyEcsTaskDefinitionArn}"

        Transcode Task Fargate Execution:
          # Sample input: See map's ItemSelector
          # Sample output: ItemSelector with TaskResult
          Type: Task
          QueryLanguage: JSONPath
          Comment: Transcodes multimedia artifact
          Retry:
            - { ErrorEquals: [States.ALL], IntervalSeconds: 15, MaxAttempts: 5, BackoffRate: 2 }
          Catch:
            - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: TaskResult Error Callback Map }

          # Add the return value of the Fargate task to the default state output
          ResultPath: "$.TaskResult"

          Next: TaskResult Callbacks Map

          Resource: arn:aws:states:::ecs:runTask.waitForTaskToken
          Parameters: # Fargate RunTask request parameters
            Cluster: "${EcsClusterArn}"
            LaunchType: FARGATE
            Overrides:
              ContainerOverrides:
                - Environment:
                    - Name: STATE_MACHINE_ARN
                      Value.$: "$$.StateMachine.Id"
                    - Name: STATE_MACHINE_NAME
                      Value.$: "$$.StateMachine.Name"
                    - Name: STATE_MACHINE_EXECUTION_ID
                      Value.$: "$$.Execution.Id"
                    - Name: STATE_MACHINE_JOB_ID
                      Value.$: "$.Job.Id"
                    - Name: STATE_MACHINE_TASK_INDEX
                      Value.$: States.Format('{}', $.TaskIndex)
                    - Name: STATE_MACHINE_S3_DESTINATION_WRITER_ROLE
                      Value: "${S3DestinationWriterRoleArn}"
                    - Name: STATE_MACHINE_AWS_REGION
                      Value: "${AwsRegion}"
                    - Name: STATE_MACHINE_ARTIFACT_BUCKET_NAME
                      Value.$: "$.Artifact.BucketName"
                    - Name: STATE_MACHINE_ARTIFACT_OBJECT_KEY
                      Value.$: "$.Artifact.ObjectKey"
                    - Name: STATE_MACHINE_DESTINATION_JSON
                      Value.$: States.JsonToString($.Task.Destination)
                    - Name: STATE_MACHINE_DESTINATION_MODE
                      Value.$: "$.Task.Destination.Mode"
                    - Name: STATE_MACHINE_DESTINATION_BUCKET_NAME
                      Value.$: "$.Task.Destination.BucketName"
                    - Name: STATE_MACHINE_DESTINATION_OBJECT_KEY
                      Value.$: "$.Task.Destination.ObjectKey"
                    - Name: STATE_MACHINE_DESTINATION_FORMAT
                      Value.$: "$.Task.Format"
                    - Name: STATE_MACHINE_FFMPEG_GLOBAL_OPTIONS
                      Value.$: "$.Task.FFmpeg.GlobalOptions"
                    - Name: STATE_MACHINE_FFMPEG_INPUT_FILE_OPTIONS
                      Value.$: "$.Task.FFmpeg.InputFileOptions"
                    - Name: STATE_MACHINE_FFMPEG_OUTPUT_FILE_OPTIONS
                      Value.$: "$.Task.FFmpeg.OutputFileOptions"
                    # Guaranteed and required values should be listed and
                    # accessed directly. Optional or conditional values can be
                    # pulled from the JSON, to avoid state machine runtime
                    # errors
                    - Name: STATE_MACHINE_TASK_JSON
                      Value.$: States.JsonToString($.Task)
                    - Name: STATE_MACHINE_ARTIFACT_JSON
                      Value.$: States.JsonToString($.Artifact)
                    - Name: STATE_MACHINE_TASK_TOKEN
                      Value.$: "$$.Task.Token"
                    - Name: STATE_MACHINE_TASK_TYPE
                      Value.$: "$.Task.Type"
                  Name: "${TranscodeContainerName}"
            NetworkConfiguration:
              AwsvpcConfiguration:
                AssignPublicIp: ENABLED
                Subnets:
                  - "${VpcPublicSubnet1}"
                  - "${VpcPublicSubnet2}"
            PropagateTags: TASK_DEFINITION
            TaskDefinition: "${TranscodeEcsTaskDefinitionArn}"

        Image Task Handler:
          # Sample input: See map's ItemSelector
          # Sample output: ItemSelector with TaskResult
          Type: Task
          QueryLanguage: JSONPath
          Comment: Manipulates and converts image artifacts
          Retry:
            - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }
          Catch:
            - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: TaskResult Error Callback Map }

          Resource: "${ImageTransformLambdaFunctionArn}"
          Parameters: # Payload provided to the Lambda function
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"

          # Add the return value of the Lambda function to the default state output
          ResultPath: "$.TaskResult"

          Next: TaskResult Callbacks Map

        WAV Wrap Task Handler:
          # Sample input: See map's ItemSelector
          # Sample output: ItemSelector with TaskResult
          Type: Task
          QueryLanguage: JSONPath
          Comment: WAV wraps audio artifacts
          Retry:
            - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }
          Catch:
            - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: TaskResult Error Callback Map }

          Resource: "${WavWrapLambdaFunctionArn}"
          Parameters: # Payload provided to the Lambda function
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"

          # Add the return value of the Lambda function to the default state output
          ResultPath: "$.TaskResult"

          Next: TaskResult Callbacks Map

        Detect Silence Task Handler:
          # Sample input: See map's ItemSelector
          # Sample output: ItemSelector with TaskResult
          Type: Task
          QueryLanguage: JSONPath
          Comment: Detects silence in audio files
          Retry:
            - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }
          Catch:
            - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: TaskResult Error Callback Map }

          Resource: "${SilenceDetectionLambdaFunctionArn}"
          Parameters: # Payload provided to the Lambda function
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"

          # Add the return value of the Lambda function to the default state output
          ResultPath: "$.TaskResult"

          Next: TaskResult Callbacks Map

        Detect Tone Task Handler:
          # Sample input: See map's ItemSelector
          # Sample output: ItemSelector with TaskResult
          Type: Task
          QueryLanguage: JSONPath
          Comment: Detects tones in audio files
          Retry:
            - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }
          Catch:
            - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: TaskResult Error Callback Map }

          Resource: "${ToneDetectionLambdaFunctionArn}"
          Parameters: # Payload provided to the Lambda function
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"

          # Add the return value of the Lambda function to the default state output
          ResultPath: "$.TaskResult"

          Next: TaskResult Callbacks Map

        Detect EAS Task Handler:
          # Sample input: See map's ItemSelector
          # Sample output: ItemSelector with TaskResult
          Type: Task
          QueryLanguage: JSONPath
          Comment: Detects alert tones in audio files
          Retry:
            - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }
          Catch:
            - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: TaskResult Error Callback Map }

          Resource: "${EasDetectionLambdaFunctionArn}"
          Parameters: # Payload provided to the Lambda function
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"

          # Add the return value of the Lambda function to the default state output
          ResultPath: "$.TaskResult"

          Next: TaskResult Callbacks Map

        Waveform Task Handler:
          # Sample input: See map's ItemSelector
          # Sample output: ItemSelector with TaskResult
          Type: Task
          QueryLanguage: JSONPath
          Comment: Generates waveform data from audio files
          Retry:
            - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }
          Catch:
            - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: TaskResult Error Callback Map }

          Resource: "${WaveformLambdaFunctionArn}"
          Parameters: # Payload provided to the Lambda function
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"

          # Add the return value of the Lambda function to the default state output
          ResultPath: "$.TaskResult"

          Next: TaskResult Callbacks Map

        Transcribe Task Job Start:
          # Sample input: See map's ItemSelector
          # Sample output: ItemSelector with TranscriptionJob
          Type: Task
          QueryLanguage: JSONPath
          Comment: Transcribes audio from artifacts
          Retry:
            - { ErrorEquals: [InvalidTranscribeTaskInputError, UnknownDestinationModeError], MaxAttempts: 0 }
            - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }
          Catch:
            - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: TaskResult Error Callback Map }

          Resource: arn:${AwsPartition}:states:::lambda:invoke.waitForTaskToken
          Parameters:
            FunctionName: "${TranscriptionJobStartLambdaFunctionArn}"
            Payload: # Payload provided to the Lambda function
              Job:
                Id.$: "$.Job.Id"
              Execution:
                Id.$: "$$.Execution.Id"
              Artifact.$: "$.Artifact"
              Task.$: "$.Task"
              TaskIteratorIndex.$: "$.TaskIndex"
              TaskToken.$: "$$.Task.Token"

          # Add the return value of the Lambda function to the default state output
          ResultPath: "$.TranscriptionJob"

          Next: Transcribe Task Job Success Handler
        Transcribe Task Job Success Handler:
          # Sample input: ItemSelector with TranscriptionJob
          # Sample output: Input with TaskResult
          Type: Task
          QueryLanguage: JSONPath
          Comment: Formats the output of a transcribe task
          Retry:
            - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }
          Catch:
            - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: TaskResult Error Callback Map }

          Resource: "${TranscriptionJobResultsLambdaFunctionArn}"
          Parameters: # Payload provided to the Lambda function
            Task.$: "$.Task"
            TranscriptionJob.$: "$.TranscriptionJob"

          # Add the return value of the Lambda function to the default state output
          ResultPath: "$.TaskResult"

          Next: TaskResult Callbacks Map

        HLS Task Fargate Execution:
          # Sample input: See map's ItemSelector
          # Sample output: ItemSelector with TaskResult
          Type: Task
          QueryLanguage: JSONPath
          Comment: Prepares an artifact for HLS delivery
          Retry:
            - { ErrorEquals: [States.ALL], IntervalSeconds: 15, MaxAttempts: 5, BackoffRate: 2 }
          Catch:
            - { ErrorEquals: [States.ALL], ResultPath: "$.Error", Next: TaskResult Error Callback Map }

          # Add the return value of the Fargate task to the default state output
          ResultPath: "$.TaskResult"

          Next: TaskResult Callbacks Map

          Resource: arn:aws:states:::ecs:runTask.waitForTaskToken
          # Parameters: # ECS RunTask request parameters
          #   CapacityProviderStrategy:
          #     - CapacityProvider: "${BasicCapacityProviderName}"
          #       Base: 0
          #       Weight: 1
          #   Cluster: "${EcsClusterArn}"
          Parameters: # ECS RunTask request parameters
            Cluster: "${EcsClusterArn}"
            LaunchType: FARGATE
            Overrides:
              ContainerOverrides:
                - Environment:
                    - Name: STATE_MACHINE_ARN
                      Value.$: "$$.StateMachine.Id"
                    - Name: STATE_MACHINE_NAME
                      Value.$: "$$.StateMachine.Name"
                    - Name: STATE_MACHINE_EXECUTION_ID
                      Value.$: "$$.Execution.Id"
                    - Name: STATE_MACHINE_JOB_ID
                      Value.$: "$.Job.Id"
                    - Name: STATE_MACHINE_TASK_INDEX
                      Value.$: States.Format('{}', $.TaskIndex)
                    - Name: STATE_MACHINE_S3_DESTINATION_WRITER_ROLE
                      Value: "${S3DestinationWriterRoleArn}"
                    - Name: STATE_MACHINE_AWS_REGION
                      Value: "${AwsRegion}"
                    - Name: STATE_MACHINE_ARTIFACT_BUCKET_NAME
                      Value.$: "$.Artifact.BucketName"
                    - Name: STATE_MACHINE_ARTIFACT_OBJECT_KEY
                      Value.$: "$.Artifact.ObjectKey"
                    - Name: STATE_MACHINE_DESTINATION_JSON
                      Value.$: States.JsonToString($.Task.Destination)
                    - Name: STATE_MACHINE_DESTINATION_MODE
                      Value.$: "$.Task.Destination.Mode"
                    - Name: STATE_MACHINE_DESTINATION_BUCKET_NAME
                      Value.$: "$.Task.Destination.BucketName"
                    - Name: STATE_MACHINE_DESTINATION_OBJECT_KEY_PREFIX
                      Value.$: "$.Task.Destination.ObjectKeyPrefix"
                    # Guaranteed and required values should be listed and
                    # accessed directly. Optional or conditional values can be
                    # pulled from the JSON, to avoid state machine runtime
                    # errors
                    - Name: STATE_MACHINE_TASK_JSON
                      Value.$: States.JsonToString($.Task)
                    - Name: STATE_MACHINE_ARTIFACT_JSON
                      Value.$: States.JsonToString($.Artifact)
                    - Name: STATE_MACHINE_TASK_TOKEN
                      Value.$: "$$.Task.Token"
                    - Name: STATE_MACHINE_TASK_TYPE
                      Value.$: "$.Task.Type"
                  Name: "${HlsContainerName}"
            NetworkConfiguration:
              AwsvpcConfiguration:
                AssignPublicIp: ENABLED
                Subnets:
                  - "${VpcPublicSubnet1}"
                  - "${VpcPublicSubnet2}"
                  - "${VpcPublicSubnet3}"
            PropagateTags: TASK_DEFINITION
            TaskDefinition: "${HlsEcsTaskDefinitionArn}"

        # Task callbacks
        # All valid tasks operations end up at one of these callback states,
        # depending on whether the task was successful or not. The value for
        # a given task in the TasksMapResults array coming out of the Tasks Map
        # will be the output whichever of these callback states is used.
        #
        # For failed tasks, the task callback will include information about
        # the error. This information is not included in the job result
        # callback.
        #
        # The output for a successful task is the result of the task.
        # The output for a failed task is the original input task definition.
        TaskResult Callbacks Map:
          # Sample input:
          #     {
          #       Job: { Id: "123" },
          #       Execution: { Id: "a1b2c3" },
          #       Artifact: {…},
          #       TaskIndex: 1/2/3/etc,
          #       Task: {…},
          #       Callbacks: {…},
          #       TaskResult: { Task: "___SOME_TASK_TYPE___", …other_task_fields },
          #     }
          # Sample output: (will become an element in TasksMapResults)
          #     { Task: "___SOME_TASK_TYPE___", …other_result_fields }
          Type: Map
          QueryLanguage: JSONPath
          Comment: >-
            Iterates over all callback endpoints to send messages when tasks
            are successful

          # Add the result of the Map (an array containing the results of each iteration) to the default output
          ResultPath: "$.TaskResultCallbacksMapResultNulls"

          # Use the TaskResult as the final output of this iteration
          OutputPath: "$.TaskResult"

          End: true

          MaxConcurrency: 0
          ItemsPath: "$.Callbacks" # This references post-InputPath input
          ItemSelector: # Initial state for each ItemProcessor iteration
            Callback.$: "$$.Map.Item.Value"
            StateMachine:
              Id.$: "$$.StateMachine.Id"
            Execution:
              Id.$: "$$.Execution.Id"
            TaskIteratorIndex.$: "$.TaskIndex"
            Message: # The callbacks Lambda function uses the JSON of this value as the message sent to each callback
              Task.$: "$.Task"
              TaskResult: # The `TaskResult` field MUST be included in these callback messages per our published API
                Job:
                  Id.$: "$.Job.Id"
                Execution:
                  Id.$: "$$.Execution.Id"
                Result.$: "$.TaskResult"
          ItemProcessor:
            ProcessorConfig: { Mode: INLINE }
            StartAt: Send TaskResult Callback
            States:
              Send TaskResult Callback:
                # Sample input: See map's ItemSelector
                # Sample output: (will become an element in TaskResultCallbacksMapResultNulls)
                #     null
                Type: Task
                Comment: >-
                  Sends a callback message to a single endpoint when tasks
                  are successful
                Retry:
                  - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }

                Resource: "${CallbackLambdaFunctionArn}"

                End: true
        TaskResult Error Callback Map:
          # Sample input:
          #     {
          #       Job: { Id: "123" },
          #       Execution: { Id: "a1b2c3" },
          #       Artifact: {…},
          #       TaskIndex: 1/2/3/etc,
          #       Task: {…},
          #       Callbacks: {…},
          #       Error: {…},
          #     }
          # Sample output: (will become an element in TasksMapResults)
          #     { Type: "___SOME_TASK_TYPE___", …other_task_fields }
          Type: Map
          QueryLanguage: JSONPath
          Comment: >-
            Iterates over all callback endpoints to send messages when tasks
            fail

          # Add the result of the Map (an array containing the results of each iteration) to the default output
          ResultPath: "$.TaskResultErrorCallbacksMapResultNulls"

          # Use the Task as the final output of this iteration
          OutputPath: "$.Task"

          End: true

          MaxConcurrency: 0
          ItemsPath: "$.Callbacks" # This references post-InputPath input
          ItemSelector: # Initial state for each ItemProcessor iteration
            Callback.$: "$$.Map.Item.Value"
            StateMachine:
              Id.$: "$$.StateMachine.Id"
            Execution:
              Id.$: "$$.Execution.Id"
            TaskIteratorIndex.$: "$.TaskIndex"
            Message: # The JSON of this value will be sent to endpoints
              Task.$: "$.Task"
              TaskResult: # The `TaskResult` field MUST be included in these callback messages per our published API
                Job:
                  Id.$: "$.Job.Id"
                Execution:
                  Id.$: "$$.Execution.Id"
                # $.Error will exist when a state takes the Catch path (because
                # of `ResultPath: "$.Error"`) and will generally be an object
                # like:
                # { Error: "SomeError", Cause: "Something went wrong" }
                Error.$: "$.Error"
          ItemProcessor:
            ProcessorConfig: { Mode: INLINE }
            StartAt: Send TaskResult Error Callback
            States:
              Send TaskResult Error Callback:
                # Sample input: See map's ItemSelector
                # Sample output: (will become an element in TaskResultErrorCallbacksMapResultNulls)
                #     null
                Type: Task
                Comment: >-
                  Sends a callback message to a single endpoint when tasks
                  fail
                Retry:
                  - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }

                Resource: "${CallbackLambdaFunctionArn}"

                End: true
  Tasks Map Catcher:
    # This state generally doesn't come into play if a task within the Task Map fails, as those failures are caught
    # within the map itself. Certain conditions can cause the Tasks Map state to fail hard enough to end up here,
    # though.
    #
    # Sample input:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "DONE",
    #       Artifact: { Descriptor: {…} },
    #       Error: {…}
    #     }
    # Sample output:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "ITERATOR_ERROR",
    #       Artifact: { Descriptor: {…} },
    #       Error: {…}
    #     }
    Type: Pass
    QueryLanguage: JSONPath
    Comment: >-
      Add a field so task map errors can be identified. Is NOT responsible
      for catching individual task execution errors (like if a Copy task fails)

    # Set the output State to a static string value
    Result: ITERATOR_ERROR
    ResultPath: "$.State"

    # Proceed to post-processing through the failure path
    Next: Add Empty TasksMapResults
  # End of task execution

  Add Empty TasksMapResults:
    # Sample input:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "___SOME_ERROR___",
    #       Error: {…}
    #       …may contain other fields
    #     }
    # Sample output:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "___SOME_ERROR___",
    #       Error: {…},
    #       TasksMapResults: [],
    #       …may contain other fields
    #     }
    Type: Pass
    QueryLanguage: JSONPath
    Comment: >-
      Add a TasksMapResults key with an empty array value, for cases where the task
      iterator does not succeed and it's not otherwise added

    # Set the TasksMapResults field in the state to an empty array
    Result: []
    ResultPath: "$.TasksMapResults"

    Next: JobResult Callback Map

  # States below here will expect $.TasksMapResults to exist on their input

  # Start of post-processing
  JobResult Callback Map:
    # Sample input:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "___SOME_STATE___",
    #       TasksMapResults: […],
    #
    #       May contain…
    #       Error: {…},
    #       …other fields
    #     }
    # Sample output:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "___SOME_STATE___",
    #       TasksMapResults: […],
    #       JobResultMapResultNulls: [null, null]
    #
    #       May contain…
    #       Error: {…},
    #       …other fields
    #     }
    Type: Map
    QueryLanguage: JSONPath
    Comment: Iterates over all callback endpoints to send job results
    # Retries are handled directly on the callback state in the iterator
    # If callbacks fail, there's no point in having a fallback catcher, since we would only use that to send callbacks

    # Add the result of the Map (an array containing the results of each iteration) to the default output
    # (This field will not be shown in future sample state data)
    ResultPath: "$.JobResultMapResultNulls"

    Next: Serialized Jobs Map

    MaxConcurrency: 0
    ItemsPath: "$.Job.Callbacks" # This references post-InputPath input
    ItemSelector: # Initial state for each ItemProcessor iteration
      Callback.$: "$$.Map.Item.Value"
      StateMachine:
        Id.$: "$$.StateMachine.Id"
      Execution:
        Id.$: "$$.Execution.Id"
      Message: # The callbacks Lambda function uses the JSON of this value as the message sent to each callback
        JobResult:
          Job:
            Id.$: "$.Job.Id"
          Execution:
            Id.$: "$$.Execution.Id"
          State.$: "$.State"
          # Task/Type = Null elements are filtered out.
          # Successful tasks will have a Task property
          TaskResults.$: "$.TasksMapResults.[?(@.Task && @.Task != 'UnknownTaskType')]"
          # Failed tasks will have a Type property
          FailedTasks.$: "$.TasksMapResults.[?(@.Type && @.Type != 'UnknownTaskType')]"
    ItemProcessor:
      ProcessorConfig: { Mode: INLINE }
      StartAt: Send JobResult Callback
      States:
        Send JobResult Callback:
          # Sample input:
          #     {
          #       Callback: {…},
          #       StateMachine: { Id: "wxyz" },
          #       Execution: { Id: "a1b2c3" },
          #       Message: {…}
          #     }
          # Sample output: (will become an element in JobResultMapResultNulls)
          #     null
          Type: Task
          QueryLanguage: JSONPath
          Comment: >-
            Sends a callback message to a single endpoint in the iterator with
            a job result
          Retry:
            - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }

          Resource: "${CallbackLambdaFunctionArn}"

          End: true

  Serialized Jobs Map:
    # Sample input:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "___SOME_STATE___",
    #       TasksMapResults: [],
    #       SerializedJobsMapResultNulls: [null, null]
    #
    #       May contain…
    #       Error: {…},
    #       …other fields
    #     }
    # Sample output:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "___SOME_STATE___",
    #       TasksMapResults: [],
    #       Void: [null, null]
    #
    #       May contain…
    #       Error: {…},
    #       …other fields
    #     }
    Type: Map
    QueryLanguage: JSONPath
    Comment: Iterates over all serialized jobs and sends them to SNS

    # Add the result of the Map (an array containing the results of each iteration) to the default output
    # (This field will not be shown in future sample state data)
    ResultPath: "$.SerializedJobsMapResultNulls"

    Next: Normalize Output

    MaxConcurrency: 0
    ItemsPath: "$.Job.SerializedJobs" # This references post-InputPath input
    ItemSelector: # Initial state for each ItemProcessor iteration
      Execution:
        Id.$: "$$.Execution.Id"
      ExecutionTrace.$: "$.Job.ExecutionTrace"
      SerializedJob.$: "$$.Map.Item.Value"
    ItemProcessor:
      ProcessorConfig: { Mode: INLINE }
      StartAt: Start Serialized Job Execution
      States:
        Start Serialized Job Execution:
          # Sample input:
          #     {
          #       Execution: { Id: "a1b2c3" },
          #       ExecutionTrace: {…},
          #       SerializedJob: {…}
          #     }
          # Sample output: (will become an element in SerializedJobsMapResultNulls)
          #     null
          Type: Task
          QueryLanguage: JSONPath
          Comment: Sends a serialized job to the job execution SNS topic
          Retry:
            - { ErrorEquals: [States.ALL], IntervalSeconds: 5, MaxAttempts: 3, BackoffRate: 2 }

          Resource: "${JobSerializerLambdaFunctionArn}"

          End: true

  Normalize Output:
    # This normalizes the actual Step Function execution's output, which,
    # currently, nothing consumes. This normalization step does not affect
    # data sent as callbacks. The output of this step is intended to match the
    # job result callback data, though.
    #
    # Sample input:
    #     {
    #       Job: { Id: "123", Source: {…}, Tasks: […], Callbacks: […], SerializedJobs: […], ExecutionTrace: […] },
    #       State: "DONE",
    #       Artifact: { …, Descriptor: {…} },
    #       TasksMapResults: [{…}, {…}, {…}]
    #     }
    # Sample output:
    #     {
    #       Time: "2026-…",
    #       Timestamp: 1787150969.658,
    #       JobResult: {…}
    #     }
    Type: Task
    QueryLanguage: JSONPath
    Comment: Normalizes output data

    Retry:
      # There's no Catch because at this point we have already sent callbacks,
      # so there's no one else to tell if things go wrong.
      - { ErrorEquals: [States.ALL], IntervalSeconds: 1, MaxAttempts: 2, BackoffRate: 1 }

    Resource: "${NormalizeOutputLambdaFunctionArn}"
    Parameters: # Payload provided to the Lambda function
      StateMachine:
        Id.$: "$$.StateMachine.Id"
      Message:
        JobResult:
          Job:
            Id.$: "$.Job.Id"
          Execution:
            Id.$: "$$.Execution.Id"
          State.$: "$.State"
          TaskResults.$: "$.TasksMapResults.[?(@.Task && @.Task != 'UnknownTaskType')]"
          FailedTasks.$: "$.TasksMapResults.[?(@.Type && @.Type != 'UnknownTaskType')]"

    End: true
  # End of post-processing