Current section

Files

Jump to
journey lib examples basic.livemd
Raw

lib/examples/basic.livemd

<!-- livebook:{"persist_outputs":true} -->
# x + y = _ , with Journey
```elixir
Mix.install(
[
{:ecto_sql, "~> 3.10"},
{:postgrex, "~> 0.21"},
{:jason, "~> 1.4"},
{:journey, "~> 0.10", organization: "shipworthy"},
{:kino_vega_lite, "~> 0.1.11"},
{:kino, "~> 0.16.1"}
],
start_applications: false
)
Application.put_env(:journey, :log_level, :warning)
Application.put_env(:journey, Journey.Repo,
database: "journey_dev",
username: "postgres",
password: "postgres",
hostname: "localhost",
log: false,
port: 5438
)
Application.put_env(:journey, :ecto_repos, [Journey.Repo])
Application.loaded_applications()
|> Enum.map(fn {app, _, _} -> app end)
|> Enum.each(&Application.ensure_all_started/1)
```
## Getting things done with Journey
This livebook shows using Journey for a ridiculously basic flow: computing the sum of two numbers. It has two input values (`x` and `y`) and one computation (`sum`).
This livebook shows creating a blueprint (graph) for computing the sum, and then executing an instance of the blueprint to perform a computation.
A few things to note:
* every input value (:x, :y), or computation result (:sum) is persisted,
* the :sum computation happens reliably (with a retry policy),
* the :sum computation is as horizontally distributed as your app,
* the :sum computation is proactive: it will be computed when `x` and `y` become available,
* the executions of this flow can take as long as needed (milliseconds? months?), and will live through system restarts, crashes, redeployments, page reloads, etc.
## Define the Blueprint of the Application
Our application is very simple, given two numbers, it computes the sum. ; )
## "Business logic": `f_add(x, y)`
```elixir
# The function for adding two numbers. Duh. ; )
f_add = fn %{x: x, y: y} -> {:ok, x + y} end
```
<!-- livebook:{"output":true} -->
```
#Function<42.81571850/1 in :erl_eval.expr/6>
```
## The flow
```elixir
import Journey.Node
import Journey.Node.Conditions
import Journey.Node.UpstreamDependencies
graph = Journey.new_graph(
# graph name.
"g1",
# graph version.
"v1",
# graph nodes.
[
input(:x),
input(:y),
# the `:sum` computation is waiting on :x and :y.
compute(:sum, [:x, :y], f_add)
]
)
```
<!-- livebook:{"output":true} -->
```
%Journey.Graph{
name: "g1",
version: "v1",
nodes: [
%Journey.Graph.Input{name: :execution_id, type: :input},
%Journey.Graph.Input{name: :last_updated_at, type: :input},
%Journey.Graph.Input{name: :x, type: :input},
%Journey.Graph.Input{name: :y, type: :input},
%Journey.Graph.Step{
name: :sum,
gated_by: [:x, :y],
f_compute: #Function<42.81571850/1 in :erl_eval.expr/6>,
f_on_save: nil,
type: :compute,
mutates: nil,
max_retries: 3,
abandon_after_seconds: 60
}
]
}
```
## Flow, visualized
```elixir
graph
|> Journey.Tools.generate_mermaid_graph()
|> Kino.Mermaid.new()
```
<!-- livebook:{"output":true} -->
```mermaid
graph TD
%% Graph
subgraph Graph["🧩 'g1', version v1"]
execution_id[execution_id]
last_updated_at[last_updated_at]
x[x]
y[y]
sum["sum<br/>(anonymous fn)"]
x --> sum
y --> sum
end
%% Styling
classDef inputNode fill:#e1f5fe,stroke:#01579b,stroke-width:2px,color:#000000
classDef computeNode fill:#f3e5f5,stroke:#4a148c,stroke-width:2px,color:#000000
classDef scheduleNode fill:#fff3e0,stroke:#e65100,stroke-width:2px,color:#000000
classDef mutateNode fill:#e8f5e8,stroke:#2e7d32,stroke-width:2px,color:#000000
%% Apply styles to actual nodes
class y,x,last_updated_at,execution_id inputNode
class sum computeNode
```
## Executing instances of the blueprint
## Starting an new execution
```elixir
execution = Journey.start_execution(graph)
# Take a note of the id of the execution, so we can reload it in case the data center reboots.
execution_id = execution.id
```
<!-- livebook:{"output":true} -->
```
"EXECZ5J17YRXL83G0Y9HVBEH"
```
```elixir
# No values are set, except for system-provided values.
Journey.values_all(execution)
```
<!-- livebook:{"output":true} -->
```
%{
sum: :not_set,
y: :not_set,
x: :not_set,
last_updated_at: {:set, 1755151863},
execution_id: {:set, "EXECZ5J17YRXL83G0Y9HVBEH"}
}
```
## Once we have `:x` and `:y`, `:sum` gets computed
```elixir
execution = Journey.set_value(execution, :x, 12); :ok
```
<!-- livebook:{"output":true} -->
```
:ok
```
Btw, if the world crashed (or got redeployed), no worries. As long as we took a note of the ID of the execution, we can load the execution as soon as things are back up, and proceed as if nothing happened.
```elixir
reloaded_execution = Journey.load(execution_id)
```
<!-- livebook:{"output":true} -->
```
%Journey.Persistence.Schema.Execution{
__meta__: #Ecto.Schema.Metadata<:loaded, "executions">,
id: "EXECZ5J17YRXL83G0Y9HVBEH",
graph_name: "g1",
graph_version: "v1",
archived_at: nil,
values: [
%Journey.Persistence.Schema.Execution.Value{
__meta__: #Ecto.Schema.Metadata<:loaded, "values">,
id: "VALBMDGBLBGBG6T403Z37YJ",
execution_id: "EXECZ5J17YRXL83G0Y9HVBEH",
execution: #Ecto.Association.NotLoaded<association :execution is not loaded>,
node_name: :last_updated_at,
node_type: :input,
node_value: 1755151863,
set_time: 1755151863,
ex_revision: 1,
inserted_at: 1755151863,
updated_at: 1755151863
},
%Journey.Persistence.Schema.Execution.Value{
__meta__: #Ecto.Schema.Metadata<:loaded, "values">,
id: "VALD4A75R8E7XJ92RHVMJVE",
execution_id: "EXECZ5J17YRXL83G0Y9HVBEH",
execution: #Ecto.Association.NotLoaded<association :execution is not loaded>,
node_name: :x,
node_type: :input,
node_value: 12,
set_time: 1755151863,
ex_revision: 1,
inserted_at: 1755151863,
updated_at: 1755151863
},
%Journey.Persistence.Schema.Execution.Value{
__meta__: #Ecto.Schema.Metadata<:loaded, "values">,
id: "VALE18VJVA3GX0VB454LJ63",
execution_id: "EXECZ5J17YRXL83G0Y9HVBEH",
execution: #Ecto.Association.NotLoaded<association :execution is not loaded>,
node_name: :execution_id,
node_type: :input,
node_value: "EXECZ5J17YRXL83G0Y9HVBEH",
set_time: 1755151863,
ex_revision: 0,
inserted_at: 1755151863,
updated_at: 1755151863
},
%Journey.Persistence.Schema.Execution.Value{
__meta__: #Ecto.Schema.Metadata<:loaded, "values">,
id: "VALJATV0Y26YY8TAE4L5RRH",
execution_id: "EXECZ5J17YRXL83G0Y9HVBEH",
execution: #Ecto.Association.NotLoaded<association :execution is not loaded>,
node_name: :y,
node_type: :input,
node_value: nil,
set_time: nil,
ex_revision: 0,
inserted_at: 1755151863,
updated_at: 1755151863
},
%Journey.Persistence.Schema.Execution.Value{
__meta__: #Ecto.Schema.Metadata<:loaded, "values">,
id: "VALH3YV7MBZMY6Y1HG7GVHV",
execution_id: "EXECZ5J17YRXL83G0Y9HVBEH",
execution: #Ecto.Association.NotLoaded<association :execution is not loaded>,
node_name: :sum,
node_type: :compute,
node_value: nil,
set_time: nil,
ex_revision: 0,
inserted_at: 1755151863,
updated_at: 1755151863
}
],
computations: [
%Journey.Persistence.Schema.Execution.Computation{
__meta__: #Ecto.Schema.Metadata<:loaded, "computations">,
id: "CMPB8LZ37J6972AYH6Z0ZVM",
execution_id: "EXECZ5J17YRXL83G0Y9HVBEH",
execution: #Ecto.Association.NotLoaded<association :execution is not loaded>,
node_name: :sum,
computation_type: :compute,
state: :not_set,
ex_revision_at_start: nil,
ex_revision_at_completion: nil,
scheduled_time: nil,
start_time: nil,
completion_time: nil,
deadline: nil,
error_details: nil,
computed_with: nil,
inserted_at: 1755151863,
updated_at: 1755151863
}
],
revision: 1,
inserted_at: 1755151863,
updated_at: 1755151863
}
```
```elixir
reloaded_execution = Journey.set_value(reloaded_execution, :y, 2); :ok
```
<!-- livebook:{"output":true} -->
```
:ok
```
```elixir
Journey.values_all(reloaded_execution)
```
<!-- livebook:{"output":true} -->
```
%{
sum: :not_set,
y: {:set, 2},
x: {:set, 12},
last_updated_at: {:set, 1755151863},
execution_id: {:set, "EXECZ5J17YRXL83G0Y9HVBEH"}
}
```
at this point, the customer provided `:x` and `:y`, and we can get the computed `:sum`:
```elixir
Journey.get_value(reloaded_execution, :sum, wait_any: true)
```
<!-- livebook:{"output":true} -->
```
{:ok, 14}
```
"oh, wait, :x was actually something else!"
```elixir
reloaded_execution = Journey.set_value(reloaded_execution, :x, 133); :ok
```
<!-- livebook:{"output":true} -->
```
:ok
```
"no worries, here is the updated sum"
```elixir
Journey.get_value(reloaded_execution, :sum, wait_new: true)
```
<!-- livebook:{"output":true} -->
```
{:ok, 135}
```
This basic computation happened with peristence, resiliency, and scalability.
## Search through computation records
"find me all the records where the sum is greater than 20, and whose x is less than 20!"
```elixir
Journey.list_executions(
graph_name: graph.name,
graph_version: graph.version,
order_by_execution_fields: [:inserted_at],
value_filters: [{:sum, :gt, 20}, {:x, :lt, 20}]
)
|> Enum.map(fn e ->
Journey.values(e)
end)
```
<!-- livebook:{"output":true} -->
```
[
%{y: 9, x: 2, last_updated_at: 1755145434, execution_id: "EXECTYHV1RM721EHH3V89A42"},
%{y: 9, x: 2, last_updated_at: 1755145483, execution_id: "EXECY4RE369XYJJ0VJX48ZH7"}
]
```
## Flow Analytics: executions stats
```elixir
Journey.Insights.Status.status()
```
<!-- livebook:{"output":true} -->
```
%{
status: :healthy,
graphs: [
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 4, failed: 0, not_set: 2, computing: 0},
most_recently_created: "2025-08-12T18:43:19Z",
most_recently_updated: "2025-08-12T18:43:19Z"
},
executions: %{
active: 3,
archived: 0,
most_recently_created: "2025-08-12T18:40:47Z",
most_recently_updated: "2025-08-12T18:43:19Z"
}
},
graph_name: "basic graph, greetings 1",
graph_version: "1.0.0"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 4, failed: 0, not_set: 0, computing: 0},
most_recently_created: "2025-08-12T17:16:04Z",
most_recently_updated: "2025-08-12T17:16:04Z"
},
executions: %{
active: 3,
archived: 0,
most_recently_created: "2025-08-12T17:08:11Z",
most_recently_updated: "2025-08-12T17:16:04Z"
}
},
graph_name: "basic graph, greetings 123",
graph_version: "1.0.0"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 0, failed: 1, not_set: 1, computing: 0},
most_recently_created: "2025-08-04T17:15:37Z",
most_recently_updated: "2025-08-04T17:15:37Z"
},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-08-04T17:15:37Z",
most_recently_updated: "2025-08-04T17:15:37Z"
}
},
graph_name: "test failure graph 698",
graph_version: "v1.0.0"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 0, failed: 2, not_set: 0, computing: 0},
most_recently_created: "2025-08-04T17:11:24Z",
most_recently_updated: "2025-08-04T17:11:31Z"
},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-08-04T17:11:24Z",
most_recently_updated: "2025-08-04T17:11:31Z"
}
},
graph_name: "test failure graph 869",
graph_version: "v1.0.0"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 0, failed: 0, not_set: 1, computing: 0},
most_recently_created: "2025-08-04T17:09:51Z",
most_recently_updated: "2025-08-04T17:09:51Z"
},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-08-04T17:09:51Z",
most_recently_updated: "2025-08-04T17:09:51Z"
}
},
graph_name: "test",
graph_version: "v1"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 1, failed: 0, not_set: 0, computing: 0},
most_recently_created: "2025-08-12T14:34:43Z",
most_recently_updated: "2025-08-12T14:34:51Z"
},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-08-12T14:34:43Z",
most_recently_updated: "2025-08-12T14:34:51Z"
}
},
graph_name: "workflow with history",
graph_version: "v1.0.0"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 0, failed: 2, not_set: 0, computing: 0},
most_recently_created: "2025-08-04T17:10:26Z",
most_recently_updated: "2025-08-04T17:10:27Z"
},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-08-04T17:10:25Z",
most_recently_updated: "2025-08-04T17:10:27Z"
}
},
graph_name: "test failure graph",
graph_version: "v1.0.0"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 0, failed: 2, not_set: 0, computing: 0},
most_recently_created: "2025-08-04T17:14:51Z",
most_recently_updated: "2025-08-04T17:15:00Z"
},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-08-04T17:14:51Z",
most_recently_updated: "2025-08-04T17:15:00Z"
}
},
graph_name: "test failure graph 12",
graph_version: "v1.0.0"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 0, failed: 4, not_set: 0, computing: 0},
most_recently_created: "2025-08-04T17:27:48Z",
most_recently_updated: "2025-08-04T17:27:55Z"
},
executions: %{
active: 2,
archived: 0,
most_recently_created: "2025-08-04T17:27:48Z",
most_recently_updated: "2025-08-04T17:27:55Z"
}
},
graph_name: "astrological sign workflow, failure compute ",
graph_version: "v2.0.0"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 0, failed: 2, not_set: 0, computing: 0},
most_recently_created: "2025-08-04T17:15:03Z",
most_recently_updated: "2025-08-04T17:15:12Z"
},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-08-04T17:15:03Z",
most_recently_updated: "2025-08-04T17:15:12Z"
}
},
graph_name: "test failure graph 872",
graph_version: "v1.0.0"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 0, failed: 2, not_set: 0, computing: 0},
most_recently_created: "2025-08-04T17:13:59Z",
most_recently_updated: "2025-08-04T17:14:05Z"
},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-08-04T17:13:59Z",
most_recently_updated: "2025-08-04T17:14:05Z"
}
},
graph_name: "test failure graph 798",
graph_version: "v1.0.0"
},
%{
stats: %{
computations: %{by_state: %{}, most_recently_created: nil, most_recently_updated: nil},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-08-13T17:11:44Z",
most_recently_updated: "2025-08-13T17:11:44Z"
}
},
graph_name: "graphtest2025-08-13",
graph_version: "v2"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 0, failed: 2, not_set: 0, computing: 0},
most_recently_created: "2025-08-04T17:15:25Z",
most_recently_updated: "2025-08-04T17:15:34Z"
},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-08-04T17:15:25Z",
most_recently_updated: "2025-08-04T17:15:34Z"
}
},
graph_name: "test failure graph 588",
graph_version: "v1.0.0"
},
%{
stats: %{
computations: %{
by_state: %{
abandoned: 1023,
cancelled: 0,
success: 61390,
failed: 0,
not_set: 151685,
computing: 33
},
most_recently_created: "2025-08-14T05:31:04Z",
most_recently_updated: "2025-08-14T05:31:05Z"
},
executions: %{
active: 12652,
archived: 5107,
most_recently_created: "2025-08-14T05:29:39Z",
most_recently_updated: "2025-08-14T05:31:05Z"
}
},
graph_name: "Credit Card Application flow graph",
graph_version: "v1.0.0"
},
%{
stats: %{
computations: %{by_state: %{}, most_recently_created: nil, most_recently_updated: nil},
executions: %{
active: 2,
archived: 0,
most_recently_created: "2025-08-13T17:11:40Z",
most_recently_updated: "2025-08-13T17:11:40Z"
}
},
graph_name: "graphtest2025-08-13",
graph_version: "v1"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 0, failed: 2, not_set: 0, computing: 0},
most_recently_created: "2025-08-04T17:15:15Z",
most_recently_updated: "2025-08-04T17:15:24Z"
},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-08-04T17:15:14Z",
most_recently_updated: "2025-08-04T17:15:24Z"
}
},
graph_name: "test failure graph 925",
graph_version: "v1.0.0"
},
%{
stats: %{
computations: %{
by_state: %{
abandoned: 0,
cancelled: 0,
success: 1696,
failed: 0,
not_set: 2250,
computing: 54
},
most_recently_created: "2025-08-01T22:05:06Z",
most_recently_updated: "2025-08-01T22:05:09Z"
},
executions: %{
active: 900,
archived: 100,
most_recently_created: "2025-08-01T22:05:06Z",
most_recently_updated: "2025-08-01T22:05:09Z"
}
},
graph_name: "flow_analytics_perf_test",
graph_version: "1.0.0"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 0, failed: 2, not_set: 0, computing: 0},
most_recently_created: "2025-08-04T17:11:13Z",
most_recently_updated: "2025-08-04T17:11:14Z"
},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-08-04T17:11:13Z",
most_recently_updated: "2025-08-04T17:11:14Z"
}
},
graph_name: "test failure graph 821",
graph_version: "v1.0.0"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 2, cancelled: 0, success: 32, failed: 7, not_set: 4, computing: 0},
most_recently_created: "2025-08-14T06:11:03Z",
most_recently_updated: "2025-08-14T06:11:03Z"
},
executions: %{
active: 22,
archived: 0,
most_recently_created: "2025-08-14T06:11:03Z",
most_recently_updated: "2025-08-14T06:11:03Z"
}
},
graph_name: "g1",
graph_version: "v1"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 0, failed: 2, not_set: 0, computing: 0},
most_recently_created: "2025-08-04T17:14:40Z",
most_recently_updated: "2025-08-04T17:14:50Z"
},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-08-04T17:14:40Z",
most_recently_updated: "2025-08-04T17:14:50Z"
}
},
graph_name: "test failure graph 279",
graph_version: "v1.0.0"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 1, failed: 0, not_set: 0, computing: 0},
most_recently_created: "2025-08-11T06:04:56Z",
most_recently_updated: "2025-08-11T06:04:56Z"
},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-08-11T06:04:56Z",
most_recently_updated: "2025-08-11T06:04:56Z"
}
},
graph_name: "basic graph, greetings TestDebug test_debug",
graph_version: "1.0.0"
},
%{
stats: %{
computations: %{
by_state: %{abandoned: 0, cancelled: 0, success: 0, failed: 0, not_set: 3, computing: 0},
most_recently_created: "2025-07-30T04:45:26Z",
most_recently_updated: "2025-07-30T04:45:26Z"
},
executions: %{
active: 1,
archived: 0,
most_recently_created: "2025-07-30T04:45:26Z",
most_recently_updated: "2025-07-30T04:59:16Z"
}
},
graph_name: "horoscope workflow, success Elixir.Journey.Examples.Horoscope-64eAJ",
graph_version: "v1.0.0"
}
],
database_connected: true
}
```
```elixir
# get some analytics for the executions flowing through the system: how many, what does the funnel look like
Journey.Insights.FlowAnalytics.flow_analytics(graph.name, graph.version)
```
<!-- livebook:{"output":true} -->
```
%{
graph_name: "g1",
graph_version: "v1",
node_stats: %{
nodes: [
%{
node_type: :input,
node_name: :x,
flow_ends_here_count: 0,
reached_count: 18,
reached_percentage: 81.8,
average_time_to_reach: 56,
flow_ends_here_percentage_of_all: 0.0,
flow_ends_here_percentage_of_reached: 0.0
},
%{
node_type: :input,
node_name: :y,
flow_ends_here_count: 0,
reached_count: 18,
reached_percentage: 81.8,
average_time_to_reach: 646,
flow_ends_here_percentage_of_all: 0.0,
flow_ends_here_percentage_of_reached: 0.0
},
%{
node_type: :compute,
node_name: :sum,
flow_ends_here_count: 0,
reached_count: 16,
reached_percentage: 72.7,
average_time_to_reach: 84,
flow_ends_here_percentage_of_all: 0.0,
flow_ends_here_percentage_of_reached: 0.0
}
]
},
analyzed_at: "2025-08-14T06:11:04.181883Z",
executions: %{
count: 22,
duration_median_seconds_to_last_update: 1,
duration_avg_seconds_to_last_update: 556
}
}
```
## In summary
This showed:
* an application defined as a graph + business logic,
* an execution of the flow take place, step by step,
* an execution of the flow be interrupted and resumed, as if nothing happened,
* analytics describing the "funnel" of executions of your graph,
Behind the scenes (not visible in this simple example):
* computations were subject to a retry policy and retries in case of failures,
* computations scale seamlessly: they run on any replica of your application.
This all happened without application data getting shipped to a third party, or requiring a SAAS dependency.
See Journey documentation for examples of more complex applications (e.g. a Horoscope app, or a Credit Card Application flow, which includes Mutations, and one-time and recurring Scheduled events).
<!-- livebook:{"offset":24238,"stamp":{"token":"XCP.tyeaWKp4UqJH-S4yW3OHCHSW4NQKyddd4ZRqBRYI68hzqLwOmuwF4lnKudI6jmFS_JfILnQmw7YwmVdN9u4u5opWapV_8X3MJ_EM9XUer3_X12IeDDtMcc64","version":2}} -->