feat: added interleave/4

lib/sage.ex

@@ -163,6 +163,16 @@ defmodule Sage do
 
   defguardp is_transaction(value) when is_function(value, 2) or is_mfa(value)
 
+  @typedoc """
+  Just like a transaction, although it receives extra argument of the name of the previous transaction
+  """
+  @type intermediate_transaction ::
+          (effects_so_far :: effects(), attrs :: any(), previous_stage :: stage_name() ->
+             {:ok | :error | :abort, any()})
+          | mfa()
+
+  defguardp is_intermediate_transaction(value) when is_function(value, 3) or is_mfa(value)
+
   @typedoc """
   Tracer callback, can be a module that implements `Sage.Tracer` behaviour, an anonymous function, or an
   `{module, function, [arguments]}` tuple.
@@ -437,6 +447,57 @@ defmodule Sage do
   def run_async(sage, name, transaction, compensation, opts \\ []),
     do: add_stage(sage, name, build_operation!(:run_async, transaction, compensation, opts))
 
+  @doc """
+  For a given Sage S with transactions :t1 -> :t2 -> :t3, a call to `interleave(S, :name, f)`
+  will yield a saga with transactions :t1 -> :name_1 -> :t2 -> :name_2 -> :t3 -> :name_3.
+
+  This can be useful if you are trying to do a long computation and want to do something with
+  the intermediate results, such as logging or persistence.
+
+  Note: 
+  - This isn't strict interleaving because a transaction is still appended at the end.
+  - Calling this function twice with the same name will give a `Sage.DuplicateStageError`, but
+  calling the function twice with a different name will compound its effects.
+  - Calling this function before the end of your saga definition will mean any stages you add after the
+  call to `interleave` will not have the intermediate stages added after them
+  """
+  @spec interleave(
+          sage :: t(),
+          name :: stage_name(),
+          intermediate_transaction :: intermediate_transaction(),
+          compensation :: compensation()
+        ) :: t()
+  def interleave(sage, name, intermediate_transaction, compensation \\ :noop)
+      when is_intermediate_transaction(intermediate_transaction) and is_compensation(compensation) do
+    sage.stages
+    |> Enum.reverse()
+    |> Enum.with_index(1)
+    |> Enum.flat_map(fn {stage, index} ->
+      name = String.to_atom("#{name}_#{index}")
+      {stage_name, _} = stage
+
+      transaction =
+        case intermediate_transaction do
+          {m, f, a} ->
+            {m, f, [stage_name | a]}
+
+          _ ->
+            &intermediate_transaction.(&1, &2, stage_name)
+        end
+
+      [stage, {name, build_operation!(:run, transaction, compensation)}]
+    end)
+    |> Enum.reduce(new(), fn stage, sage ->
+      case stage do
+        {name, {:run, transaction, comp, _opts}} ->
+          run(sage, name, transaction, comp)
+
+        {name, {:run_async, transaction, comp, opts}} ->
+          run_async(sage, name, transaction, comp, opts)
+      end
+    end)
+  end
+
   @doc """
   Executes a Sage.
 

test/sage_test.exs

@@ -254,6 +254,70 @@ defmodule SageTest do
     end
   end
 
+  describe "interleaves/3" do
+    test "adds a step between every transaction" do
+      sage =
+        new()
+        |> run(:t1, transaction(:t1))
+        |> run(:t2, transaction(:t2))
+        |> run_async(:t_async, transaction(:t_async), :noop)
+        |> run(:t3, transaction(:t3))
+        |> interleave(:i, fn _effects, _args, previous_stage_name -> {:ok, previous_stage_name} end)
+
+      assert [i_4: _, t3: _, i_3: _, t_async: _, i_2: _, t2: _, i_1: _, t1: _] = sage.stages
+      assert {:ok, _, %{i_4: :t3, i_3: :t_async, i_2: :t2, i_1: :t1}} = execute(sage)
+    end
+
+    test "adds nothing if there are no transactions" do
+      assert [] =
+               new()
+               |> interleave(:i, fn _effects, _args, _previous_stage_name -> :ok end)
+               |> Map.get(:stages)
+    end
+
+    test "adds a transaction at the end if there is one transaction" do
+      assert [i_1: _, t1: _] =
+               new()
+               |> run(:t1, transaction(:t1))
+               |> interleave(:i, fn _effects, _args, _previous_stage_name -> :ok end)
+               |> Map.get(:stages)
+    end
+
+    test "works with mfa" do
+      sage =
+        new()
+        |> run(:t1, transaction(:t1))
+        |> run(:t2, transaction(:t2))
+        |> run(:t3, transaction(:t3))
+        |> interleave(:i, {TestIntermediateTransactionHandler, :intermediate_transaction_handler, [:foo]})
+
+      assert {:ok, _, %{i_3: {:t3, :foo}, i_2: {:t2, :foo}, i_1: {:t1, :foo}}} = execute(sage)
+    end
+
+    test "can run a compensations" do
+      new()
+      |> run(:t1, transaction(:t1))
+      |> run(:t2, transaction(:t2))
+      |> run(:t3, transaction_with_error(:t3))
+      |> interleave(:i, fn _effects, _args, _previous_stage_name -> {:ok, nil} end, fn _errored_effect, _effects_so_far, _attrs ->
+        send(self(), :compensating)
+        :ok
+      end)
+      |> execute()
+
+      for _ <- 1..2, do: assert_received(:compensating)
+    end
+
+    test "errors if used more than once" do
+      assert_raise Sage.DuplicateStageError, fn ->
+        new()
+        |> run(:t1, transaction(:t1))
+        |> interleave(:i, fn _effects, _args, _previous_stage_name -> :ok end)
+        |> interleave(:i, fn _effects, _args, _previous_stage_name -> :ok end)
+      end
+    end
+  end
+
   def dummy_transaction_for_mfa(_effects_so_far, _opts), do: raise("Not implemented")
   def dummy_compensation_for_mfa(_effect_to_compensate, _opts), do: raise("Not implemented")
   def dummy_final_cb(_status, _opts, _return), do: raise("Not implemented")

test/support/test_intermediate_transaction_handler.ex

@@ -0,0 +1,5 @@
+defmodule TestIntermediateTransactionHandler do
+  def intermediate_transaction_handler(_effects, _args, previous_stage_name, something_else) do
+    {:ok, {previous_stage_name, something_else}}
+  end
+end