defmodule Banzai do @moduledoc """ A token-based pipeline pattern for sequential function execution with automatic error handling. Banzai implements a workflow pattern where a **token** (a map or struct) is passed through a series of **steps** (functions) that transform it. Each step receives the token, performs its operation, and returns an updated token. If any step fails, execution stops immediately and the error is returned. ## Core Concepts ### Token The token is the data structure that flows through the pipeline. It can be a map or struct and accumulates state as it passes through each step. In practice, the token is often an embedded Ecto schema representing the action's state. ### Steps Steps are functions that receive the token and return either: - `{:ok, updated_token}` - Success, continue to next step - `{:error, reason}` - Failure, stop execution and return error Steps are executed in the order they are added to the pipeline. ### Early Termination When a step returns `{:error, reason}`, execution stops immediately. Subsequent steps are not executed, and the error is returned to the caller. ## Basic Example Here's a simple example showing token transformation through multiple steps: defmodule CalculateTotal do use Banzai def perform(params) do %{value: 0} |> new(%{}) |> step(fn token -> {:ok, Map.put(token, :value, token.value + 10)} end) |> step(fn token -> {:ok, Map.put(token, :value, token.value * 2)} end) |> step(fn token -> {:ok, Map.put(token, :value, token.value + 5)} end) |> run() end end # Result: {:ok, %{value: 25}} # Step 1: 0 + 10 = 10 # Step 2: 10 * 2 = 20 # Step 3: 20 + 5 = 25 ## Error Handling When a step returns `{:error, reason}`, Banzai: - Stops execution immediately (no subsequent steps run) - Logs the error with context (action_id, step_index, error) - Tracks which step failed in `__META__.failed_step` - Returns `{:error, reason}` to the caller Example with error handling: defmodule ProcessOrder do use Banzai def perform(params) do %{order_id: params.order_id} |> new(%{}) |> step(&validate_order/1) |> step(&calculate_total/1) |> step(&charge_customer/1) |> run() end defp validate_order(token) do if order_exists?(token.order_id) do {:ok, token} else {:error, :order_not_found} end end # This step won't run if validate_order fails defp calculate_total(token) do {:ok, Map.put(token, :total, 100.0)} end end ## Usage Pattern The typical pattern when using Banzai is: 1. Define an embedded schema (or use a map) as your token 2. Implement `perform/1` that builds the pipeline 3. Define private step functions that transform the token 4. Each step returns `{:ok, updated_token}` or `{:error, reason}` The token-based pattern provides several benefits: - Clear, linear flow of execution - Automatic error propagation - Easy to test individual steps - Composable and reusable step functions - Built-in logging and debugging support """ require Logger @type meta() :: %{ id: Ecto.UUID.t(), failed_step: integer() | nil } @type t() :: %__MODULE__{ __META__: meta(), token: map() | struct(), funcs: list(function()), error: term() | nil } defstruct token: %{}, funcs: [], error: nil, __META__: %{id: nil, failed_step: nil} @doc """ Execute the event """ @callback perform(map()) :: {:ok, term()} | {:error, term()} defmacro __using__(_opts) do require Logger quote generated: true do alias Ecto.Changeset import Ecto.Changeset use Ecto.Schema @behaviour Banzai import Banzai, only: [ run: 1, step: 2, new: 2, new: 3 ] end end @spec new(map() | struct(), map()) :: t() def new(token, initial_state \\ %{}, opts \\ []) do keys_to_atoms? = Keyword.get(opts, :keys_to_atoms?, false) initial_state = if keys_to_atoms?, do: keys_to_atoms(initial_state), else: initial_state %__MODULE__{ __META__: %{ id: Ecto.UUID.generate(), failed_step: nil }, token: Map.merge(token, initial_state), funcs: [], error: nil } end @spec step(t(), function(), Keyword.t()) :: t() def step(%__MODULE__{funcs: funcs} = action, func, _opts \\ []) do %__MODULE__{ action | funcs: [func | funcs] } end defp run_step(action = %__MODULE__{error: error}, _, _) when error != nil, do: action defp run_step(%__MODULE__{} = action, func, step_index) do case func.(action.token) do {:ok, updated_token} -> %__MODULE__{action | token: updated_token} {:error, reason} -> Logger.error("Banzai step failed", action_id: action.__META__.id, step_index: step_index, error: inspect(reason) ) %__MODULE__{ action | error: reason, __META__: %{ action.__META__ | failed_step: step_index } } invalid_response -> error_msg = "Banzai step returned invalid response. Expected {:ok, term} | {:error, term}, got: #{inspect(invalid_response)}" Logger.error("Banzai step returned invalid response", action_id: action.__META__.id, step_index: step_index, response: inspect(invalid_response) ) %__MODULE__{ action | error: error_msg, __META__: %{ action.__META__ | failed_step: step_index } } end end def run(%__MODULE__{} = action) do action_name = get_action_name(action.token) total_steps = length(action.funcs) Logger.info("Starting Banzai action", action_id: action.__META__.id, action_name: action_name, total_steps: total_steps ) final_action = execute(action) case final_action do %__MODULE__{error: reason} when reason != nil -> Logger.error("Banzai action failed", action_id: action.__META__.id, action_name: action_name, failed_step: final_action.__META__.failed_step, error: inspect(reason) ) {:error, reason} %__MODULE__{token: token} -> Logger.info("Banzai action completed successfully", action_id: action.__META__.id, action_name: action_name, total_steps: total_steps ) {:ok, token} end end defp execute(%__MODULE__{funcs: funcs} = action) do funcs = Enum.reverse(funcs) funcs |> Enum.with_index(1) |> Enum.reduce(action, fn {func, index}, acc -> run_step(acc, func, index) end) end defp get_action_name(token) when is_map(token) do case Map.get(token, :__struct__) do nil -> "Unknown" struct -> struct |> Module.split() |> List.last() end end defp get_action_name(_), do: "Unknown" def keys_to_atoms(data) do do_transform_to_atoms(data) end defp do_transform_to_atoms(data) when is_struct(data) do struct!( data.__struct__, data |> Map.from_struct() |> Map.new(fn {key, val} -> {key, do_transform_to_atoms(val)} end) ) end defp do_transform_to_atoms(data) when is_map(data) do Map.new(data, fn {key, val} -> new_key = if is_binary(key), do: String.to_existing_atom(key), else: key {new_key, do_transform_to_atoms(val)} end) end defp do_transform_to_atoms(data) when is_list(data) do Enum.map(data, &do_transform_to_atoms/1) end defp do_transform_to_atoms(data), do: data end