PgFlow.Flow (PgFlow v0.3.1)

Copy Markdown View Source

A macro-based DSL for defining pgflow workflows.

This module provides a declarative way to define workflow steps with dependencies, retries, timeouts, and array processing capabilities. Use it by calling use PgFlow.Flow in your flow module.

Example

defmodule MyApp.Flows.Example do
  use PgFlow.Flow

  @flow queue: :example, max_attempts: 3, base_delay: 5, timeout: 60

  step :first do
    fn input, ctx ->
      %{result: input["value"] * 2}
    end
  end

  step :second, depends_on: [:first] do
    fn deps, ctx ->
      %{doubled: deps.first["result"]}
    end
  end

  map :process_items, array: :second do
    fn item, ctx ->
      %{processed: item}
    end
  end
end

Flow Options

The @flow module attribute accepts the following options:

  • :queue - (required) atom identifier for the flow queue (also accepts :slug as alias)
  • :max_attempts - maximum retry attempts for failed steps (default: 1)
  • :base_delay - base delay in seconds for exponential backoff (default: 1)
  • :timeout - step execution timeout in seconds (default: 30)
  • :cron - (optional) schedule this flow via pg_cron with sub-options:
    • :schedule - (required) cron schedule string (e.g., "@hourly", "0 9 *")
    • :input - (optional) static input map passed to each scheduled run

Step Options

Steps defined with step/2 or step/3 accept these options:

  • :depends_on - list of step atoms this step depends on
  • :handler - module implementing PgFlow.StepHandler (alternative to block)
  • :max_attempts - override flow-level max_attempts
  • :base_delay - override flow-level base_delay
  • :timeout - override flow-level timeout
  • :start_delay - seconds to delay before starting this step
  • :if - map pattern that must match for the step to run (optional)
  • :if_not - map pattern that must not match for the step to run (optional)
  • :when_unmet - what happens when :if or :if_not is not satisfied: :fail, :skip, or :skip_cascade (default :skip when a condition is set)
  • :when_exhausted - what happens when retries are exhausted: :fail, :skip, or :skip_cascade (default :fail)

Conditional Step Execution

Steps can be made conditional using :if or :if_not options with PostgreSQL @> (contains) pattern matching on JSON objects:

Root Steps

For root steps (those with no dependencies), the pattern is matched against the flow's input map:

step :process, if: %{"mode" => "active"} do
  fn input, _ctx ->
    # Only runs if input contains {"mode" => "active"}
    %{processed: true}
  end
end

Dependent Steps

For steps with dependencies, the pattern is matched against a map built from the dependencies' outputs: %{dep_slug => output}:

step :charge, depends_on: [:validate] do
  fn deps, _ctx ->
    %{charged: true}
  end
end

step :send_receipt, depends_on: [:charge], if: %{"charge" => %{"success" => true}} do
  fn deps, _ctx ->
    # Only runs if deps.charge contains {"success" => true}
    %{sent: true}
  end
end

Condition Behavior

  • :when_unmet - Action when condition is not satisfied (requires :if or :if_not; default :skip):

    • :skip - Skip this step; its key is omitted from dependent step inputs
    • :skip_cascade - Skip this step and all steps downstream of it (transitively)
    • :fail - Fail this step (bypassing handler execution)

  • :when_exhausted - Action when retries exhaust (default :fail):

    • :skip - Skip after max retries
    • :skip_cascade - Skip after max retries and cascade to dependents
    • :fail - Fail after max retries

Omitted-Key Contract

When a non-cascade-skipped dependency is skipped, its key is omitted from the dependent handler's input map. This allows handlers to distinguish "dependency not run" from "dependency ran but returned nil":

# If :validate is skipped (non-cascade), deps will be %{}
# If :validate ran and returned %{"valid" => false}, deps will be %{"validate" => %{"valid" => false}}
step :charge, depends_on: [:validate] do
  fn deps, _ctx ->
    if Map.has_key?(deps, "validate") do
      # validate ran
      %{charged: true}
    else
      # validate was skipped — this step executes but should handle the missing dep
      %{charged: false, reason: :dependency_skipped}
    end
  end
end

Type Violation Exception

Retry exhaustion (:when_exhausted) does not apply to TYPE_VIOLATION errors; such errors fail immediately regardless of retry settings.

Map Options

Map steps defined with map/2 or map/3 accept step options plus:

  • :array - step slug whose output array to process (for dependent maps)

Generated Functions

Using this module generates the following callback functions:

  • __pgflow_definition__/0 - returns a PgFlow.Flow.Definition struct
  • __pgflow_slug__/0 - returns the flow slug atom
  • __pgflow_steps__/0 - returns the raw step definitions
  • __pgflow_handler__/1 - pattern-matched functions for each step

Runtime Flow Definition

For dynamic (runtime-defined) flows, see PgFlow.Client.upsert_flow/2, which accepts the same conditional options as the DSL.

Summary

Functions

map(slug, opts \\ [], list)

Defines an array processing step that executes a handler for each item.

step(slug, opts \\ [], list)

Defines a single execution step in the workflow.

Functions

map(slug, opts \\ [], list)

(macro)

Defines an array processing step that executes a handler for each item.

Options

Map steps accept all step options (see step/3), plus:

  • :array - (optional) step slug whose output array to process. If omitted, the map is a root map processing the flow's input array.

Conditional options (:if, :if_not, :when_unmet, :when_exhausted) apply to the map step itself, not to individual items.

Examples

# Root map over the flow's input array
map :process_users do
  fn user, ctx ->
    %{processed: process_user(user)}
  end
end

# Map over output from another step
map :enrich_items, array: :fetch_items do
  fn item, ctx ->
    %{enriched: enrich(item)}
  end
end

# Map with conditional execution (skipped if condition unmet)
map :premium_enrichment, array: :items, if: %{"plan" => "premium"} do
  fn item, ctx ->
    %{enriched: enrich_premium(item)}
  end
end

# Map with module handler
map :validate_each, array: :items, handler: MyApp.ValidateItemHandler

# Map with custom settings
map :slow_processing, array: :items, timeout: 120 do
  fn item, ctx ->
    %{result: slow_process(item)}
  end
end

For more on conditional execution, see the "Conditional Step Execution" section in the moduledoc.

step(slug, opts \\ [], list)

(macro)

Defines a single execution step in the workflow.

Options

See the moduledoc "Step Options" section for the complete list. Notably:

  • :if - Map the input/deps must match (JSON @> containment)
  • :if_not - Map the input/deps must not match

  • :when_unmet - Action when condition fails (:fail | :skip | :skip_cascade, default :skip)

  • :when_exhausted - Action when retries exhaust (:fail | :skip | :skip_cascade, default :fail)

Examples

# Basic step with inline handler
step :fetch_data do
  fn input, ctx ->
    %{data: fetch_from_api(input["url"])}
  end
end

# Step with dependencies
step :transform, depends_on: [:fetch_data] do
  fn deps, ctx ->
    %{transformed: transform(deps.fetch_data["data"])}
  end
end

# Step with conditional execution (matches against input)
step :premium_step, if: %{"plan" => "premium"} do
  fn input, ctx ->
    %{premium_feature: true}
  end
end

# Step with conditional execution (matches against dependencies)
step :charge, depends_on: [:validate], if: %{"validate" => %{"valid" => true}} do
  fn deps, ctx ->
    %{charged: true}
  end
end

# Step with module handler
step :validate, handler: MyApp.ValidateHandler

# Step with custom retry settings
step :flaky_operation, max_attempts: 5, base_delay: 10 do
  fn input, ctx ->
    perform_operation()
  end
end

For more on conditional execution, see the "Conditional Step Execution" section in the moduledoc.