Creates a graph node that mutates the value of another node.

Examples:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "`archive()` doctest graph (a useless machine that archives itself immediately;)",
...>       "v1.0.0",
...>       [
...>         input(:name),
...>         archive(:archive, [:name])
...>       ]
...>     )
iex> execution = graph |> Journey.start_execution()
iex> execution.archived_at == nil
true
iex> execution = Journey.set_value(execution, :name, "Mario")
iex> {:ok, _} = Journey.get_value(execution, :archive, wait_any: true)
iex> Journey.load(execution)
nil
iex> execution = Journey.load(execution, include_archived: true)
iex> execution.archived_at == nil
false

Creates a self-computing node.

name is an atom uniquely identifying the node in this graph.

gated_by defines when this node becomes eligible to compute. Accepts either:

• A list of atom node names, e.g. [:a, :b], indicating the node becomes unblocked when all of the listed nodes have a value.
• A keyword list with conditions, e.g. [a: fn node -> node.node_value > 10 end], for conditional dependencies.
• A mixed list combining atoms and keyword conditions, e.g. [:a, :b, c: fn node -> node.node_value > 5 end].
• A structured condition (see unblocked_when/1 ) allowing for logical operators (:and, :or) and custom value predicates (e.g. unblocked_when({:and, [{:a, &provided?/1}, {:b, &provided?/1}]})).

f_compute is the function that computes the value of the node, once the upstream dependencies are satisfied. The function is provided a map of the upstream nodes and their values as its argument and returns a tuple:

• {:ok, value} or
• {:error, reason}. The function is called when the upstream nodes are set, and the value is set to the result of the function.

Note that return values are JSON-serialized for storage. If the returned value or reason contains atoms (e.g., {:ok, :pending} or {:ok, %{status: :active}}), those atoms will be converted to strings when retrieved via get_value/3.

In the case of a failure, the function is automatically retried, up to max_retries times. If the function fails after max_retries attempts, the node is marked as failed. If the function does not return within abandon_after_seconds, it is considered abandoned, and it will be retried (up to max_retries times).

Examples:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "`compute()` doctest graph (pig-latinize-ish a name)",
...>       "v1.0.0",
...>       [
...>         input(:name),
...>         compute(
...>           :pig_latin_ish_name,
...>           [:name],
...>           fn %{name: name} ->
...>             {:ok, "#{name}-ay"}
...>           end,
...>           max_retries: 4, # Optional (default: 3)
...>           abandon_after_seconds: 60, # Optional (default: 60)
...>           f_on_save: fn _execution_id, _params ->
...>             # Optional callback to be called when the value is saved.
...>             # This is useful for notifying other systems (e.g. a LiveView
...>             # via PubSub.notify()) – that the value has been saved.
...>             :ok
...>           end
...>         )
...>       ]
...>     )
iex> execution = graph |> Journey.start_execution() |> Journey.set_value(:name, "Alice")
iex> execution |> Journey.get_value(:pig_latin_ish_name, wait_any: true)
{:ok, "Alice-ay"}
iex> execution |> Journey.values() |> redact([:execution_id, :last_updated_at])
%{name: "Alice", pig_latin_ish_name: "Alice-ay", execution_id: "...", last_updated_at: 1_234_567_890}

Keyword List Syntax for Conditional Dependencies

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "threshold alert example",
...>       "v1.0.0",
...>       [
...>         input(:temperature),
...>         # Using keyword list syntax for conditional dependency
...>         compute(
...>           :high_temp_alert,
...>           [temperature: fn node -> node.node_value > 30 end],
...>           fn %{temperature: temp} ->
...>             {:ok, "High temperature alert: #{temp}°C"}
...>           end
...>         )
...>       ]
...>     )
iex> execution = graph |> Journey.start_execution()
iex> execution = Journey.set_value(execution, :temperature, 25)
iex> Journey.get_value(execution, :high_temp_alert)
{:error, :not_set}
iex> execution = Journey.set_value(execution, :temperature, 35)
iex> Journey.get_value(execution, :high_temp_alert, wait_any: true)
{:ok, "High temperature alert: 35°C"}

Return Values

The f_compute function must return {:ok, value} or {:error, reason}. Note that atoms in the returned value and reason will be converted to strings when persisted.

Creates a graph input node. The value of an input node is set with Journey.set_value/3. The name of the node must be an atom.

Examples:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "`input()` doctest graph (just a few input nodes)",
...>       "v1.0.0",
...>       [
...>         input(:first_name),
...>         input(:last_name),
...>         input(:zip_code)
...>        ]
...>     )
iex> execution = graph |> Journey.start_execution() |> Journey.set_value(:first_name, "Mario")
iex> Journey.values(execution) |> redact([:execution_id, :last_updated_at])
%{first_name: "Mario", execution_id: "...", last_updated_at: 1_234_567_890}

Creates a graph node that mutates the value of another node.

Examples:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "`mutate()` doctest graph (a useless machine;)",
...>       "v1.0.0",
...>       [
...>         input(:name),
...>         mutate(
...>           :remove_pii,
...>           [:name],
...>           fn %{name: _name} ->
...>             # Return the new value for the "name" node.
...>             {:ok, "redacted"}
...>           end,
...>           mutates: :name # The name of an existing node whose value will be mutated.
...>         )
...>       ]
...>     )
iex> execution =
...>     graph
...>     |> Journey.start_execution()
...>     |> Journey.set_value(:name, "Mario")
iex> execution |> Journey.get_value(:remove_pii, wait_any: true)
{:ok, "updated :name"}
iex> execution |> Journey.values() |> redact([:execution_id, :last_updated_at])
%{name: "redacted", remove_pii: "updated :name",  execution_id: "...", last_updated_at: 1_234_567_890}

Return Values

The f_compute function must return {:ok, value} or {:error, reason}. Note that atoms in the returned value and reason will be converted to strings when persisted.

Creates a graph node that declares its readiness at a specific time, once.

Once this node is unblocked, it will be executed to set the time at which it will unblock its downstream dependencies.

Examples:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "`schedule_once()` doctest graph (it reminds you to take a nap in a couple of seconds;)",
...>       "v1.0.0",
...>       [
...>         input(:name),
...>         schedule_once(
...>           :schedule_a_nap,
...>           [:name],
...>           fn %{name: _name} ->
...>             # This function is to return the time (in epoch seconds) at which
...>             # its downstream dependencies should be unblocked.
...>             in_two_seconds = System.system_time(:second) + 2
...>             {:ok, in_two_seconds}
...>           end
...>         ),
...>         compute(
...>           :nap_time,
...>           [:name, :schedule_a_nap],
...>           fn %{name: name, schedule_a_nap: _time_to_take_a_nap} ->
...>             {:ok, "It's time to take a nap, #{name}!"}
...>           end
...>         )
...>       ]
...>     )
iex> execution =
...>     graph
...>     |> Journey.start_execution()
...>     |> Journey.set_value(:name, "Mario")
iex> execution |> Journey.values() |> Map.get(:name)
"Mario"
iex> # This is only needed in a test, to simulate the background processing that happens in non-tests automatically.
iex> background_sweeps_task = Journey.Scheduler.Background.Periodic.start_background_sweeps_in_test(execution.id)
iex> execution |> Journey.get_value(:nap_time, wait_any: true)
{:ok, "It's time to take a nap, Mario!"}
iex> Journey.Scheduler.Background.Periodic.stop_background_sweeps_in_test(background_sweeps_task)

Return Values

The f_compute function must return {:ok, value} or {:error, reason}. Note that atoms in the returned value and reason will be converted to strings when persisted.

Creates a graph node that declares its readiness at a specific time, time after time.

Once this node is unblocked, it will be repeatedly computed, to set the time at which it will unblock its downstream dependencies.

This is useful for triggering recurring tasks, such as sending reminders or notifications.

Examples:

iex> import Journey.Node
iex> graph = Journey.new_graph(
...>       "`schedule_recurring()` doctest graph (it issues 'reminders' every few seconds)",
...>       "v1.0.0",
...>       [
...>         input(:name),
...>         schedule_recurring(
...>           :schedule_a_reminder,
...>           [:name],
...>           fn _ ->
...>             soon = System.system_time(:second) + 2
...>             {:ok, soon}
...>           end
...>         ),
...>         compute(
...>           :send_a_reminder,
...>           [:name, :schedule_a_reminder],
...>           fn %{name: name} = v ->
...>             reminder_count = Map.get(v, :send_a_reminder, 0) + 1
...>             IO.puts("[#{System.system_time(:second)}] #{name}, here is your scheduled reminder # #{reminder_count}.")
...>             {:ok, reminder_count}
...>           end
...>         )
...>       ]
...>     )
iex> execution =
...>     graph
...>     |> Journey.start_execution()
...>     |> Journey.set_value(:name, "Mario")
iex> execution |> Journey.values() |> Map.get(:name)
"Mario"
iex> # This is only needed in a test, to simulate the background processing that happens in non-tests automatically.
iex> background_sweeps_task = Journey.Scheduler.Background.Periodic.start_background_sweeps_in_test(execution.id)
iex> # Wait for initial reminders
iex> {:ok, count1} = Journey.get_value(execution, :send_a_reminder, wait_any: true)
iex> count1 >= 1
true
iex> # Wait for more reminders to verify recurring behavior
iex> execution = Journey.load(execution)
iex> {:ok, count2} = Journey.get_value(execution, :send_a_reminder, wait_new: true)
iex> count2 > count1
true
iex> execution = Journey.load(execution)
iex> {:ok, count3} = Journey.get_value(execution, :send_a_reminder, wait_new: true)
iex> count3 > count2
true
iex> Journey.Scheduler.Background.Periodic.stop_background_sweeps_in_test(background_sweeps_task)

Return Values

The f_compute function must return {:ok, value} or {:error, reason}. Note that atoms in the returned value and reason will be converted to strings when persisted.