Packages

journey

0.10.45
0.10.58 0.10.57 0.10.56 0.10.55 0.10.54 0.10.53 0.10.52 0.10.51 0.10.50 0.10.49 0.10.48 0.10.47 0.10.46 0.10.45 0.10.44 0.10.43 0.10.41 0.10.40 0.10.39 0.10.38 0.10.37 0.10.36 0.10.35 0.10.34 0.10.33 0.10.32 0.10.31 0.10.30 0.10.29 0.10.28 0.10.27 0.10.26 0.10.25 0.10.24 0.10.23 0.10.22 0.0.9 retired 0.0.8 0.0.7 0.0.6 0.0.5 0.0.3 0.0.2

Journey is a library for defining and running durable workflows with persistence, reliability, and scalability.

HexDocs

Current section

Files

Jump to
Readme
43 Versions
8 Dependencies
Dependants
Files
Activity
journey lib journey graph.ex
Raw

lib/journey/graph.ex 

defmodule Journey.Graph do

  @moduledoc false



  import Journey.Node, only: [input: 1]



  defstruct [:name, :version, :nodes, :f_on_save, :hash, :execution_id_prefix, singleton: false]



  @type t :: %__MODULE__{

          name: String.t(),

          version: String.t(),

          nodes: list,

          f_on_save: function() | nil,

          hash: String.t(),

          execution_id_prefix: String.t(),

          singleton: boolean()

        }



  def new(name, version, nodes, opts \\ [])

      when (is_binary(name) or is_nil(name)) and (is_binary(version) or is_nil(version)) and

             is_list(nodes) do

    opts_schema = [

      f_on_save: [

        is: {:fun, 3},

        required: false,

        doc:

          "Graph-wide callback invoked after any node computation succeeds. Receives (execution_id, node_name, result)."

      ],

      execution_id_prefix: [

        is: :binary,

        required: false,

        doc: "Custom prefix for execution IDs. Will be normalized to uppercase. Defaults to 'EXEC'."

      ],

      singleton: [

        is: :boolean,

        required: false,

        doc: "When true, only one non-archived execution can exist for this graph. Defaults to false."

      ]

    ]



    KeywordValidator.validate!(opts, opts_schema)



    all_nodes = [input(:execution_id), input(:last_updated_at)] ++ nodes

    hash = compute_hash(all_nodes)



    actual_name = name || "graph_#{String.slice(hash, 0, 8) |> String.downcase()}"

    actual_version = version || "v1.0"



    %__MODULE__{

      name: actual_name,

      version: actual_version,

      nodes: all_nodes,

      f_on_save: Keyword.get(opts, :f_on_save),

      execution_id_prefix: Keyword.get(opts, :execution_id_prefix, "EXEC"),

      hash: hash,

      singleton: Keyword.get(opts, :singleton, false)

    }

  end



  def find_node_by_name(nil, node_name) when is_atom(node_name) do

    nil

  end



  def find_node_by_name(graph, node_name) when is_struct(graph, Journey.Graph) and is_atom(node_name) do

    graph.nodes

    |> Enum.find(fn n -> n.name == node_name end)

  end



  defp compute_hash(nodes) do

    nodes

    |> Enum.sort_by(& &1.name)

    |> Enum.map(fn %{name: name, type: type} -> {name, type} end)

    |> :erlang.term_to_binary()

    |> then(&:crypto.hash(:sha256, &1))

    |> Base.encode16()

  end

end