# `PgFlow.Flow`
[🔗](https://github.com/agoodway/pgflow/blob/v0.3.1/lib/pgflow/flow.ex#L1)

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.

# `map`
*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`
*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.

---

*Consult [api-reference.md](api-reference.md) for complete listing*