Add conditional stage

Allows applying another pipeline stage based on a condition, which can
either be a whole filter or a single filter rule.

lib/frenzy/pipeline/conditional_stage.ex

@@ -0,0 +1,87 @@
+defmodule Frenzy.Pipeline.ConditionalStage do
+  require Logger
+  alias Frenzy.Pipeline.{Stage, FilterEngine}
+  @behaviour Stage
+
+  @filter_modes ["accept", "reject"]
+  @rule_modes ["contains_string", "matches_regex"]
+
+  @impl Stage
+  def apply(%{"stage" => stage, "opts" => stage_opts, "condition" => condition}, item_params) do
+    if test_condition(condition, item_params) do
+      apply(String.to_existing_atom("Elixir." <> stage), :apply, [stage_opts, item_params])
+    else
+      {:ok, item_params}
+    end
+  end
+
+  @impl Stage
+  def apply(opts, item_params) do
+    Logger.warn("Received invalid conditional opts: #{inspect(opts)}")
+    {:ok, item_params}
+  end
+
+  defp test_condition(%{"mode" => mode} = filter, item_params) when mode in @filter_modes do
+    FilterEngine.test(filter, item_params)
+  end
+
+  defp test_condition(%{"mode" => mode} = rule, item_params) when mode in @rule_modes do
+    FilterEngine.test_rule(rule, item_params)
+  end
+
+  defp test_condition(condition, _item_params) do
+    Logger.warn("Received invalid condition: #{inspect(condition)}")
+    false
+  end
+
+  @impl Stage
+  def validate_opts(opts) do
+    cond do
+      not (Map.has_key?(opts, "stage") and is_binary(opts["stage"]) and
+               module_exists(opts["stage"])) ->
+        {:error, "stage must be a string containg a module that exists"}
+
+      not (Map.has_key?(opts, "opts") and is_map(opts["opts"])) ->
+        {:error, "opts must be a map"}
+
+      not (Map.has_key?(opts, "condition") and is_map(opts["condition"])) ->
+        {:error, "condition must be a map"}
+
+      true ->
+        with {:ok, stage_opts} <-
+               apply(String.to_existing_atom("Elixir." <> opts["stage"]), :validate_opts, [
+                 opts["opts"]
+               ]),
+             {:ok, condition} <- validate_condition(opts["condition"]) do
+          {
+            :ok,
+            opts
+            |> Map.put("opts", stage_opts)
+            |> Map.put("condition", condition)
+          }
+        else
+          {:error, _reason} = err ->
+            err
+        end
+    end
+  end
+
+  defp module_exists(module_name) do
+    try do
+      String.to_existing_atom("Elixir." <> module_name)
+      true
+    rescue
+      ArgumentError -> false
+    end
+  end
+
+  defp validate_condition(%{"mode" => mode} = filter) when mode in @filter_modes do
+    FilterEngine.validate_filter(filter)
+  end
+
+  defp validate_condition(%{"mode" => mode} = rule) when mode in @rule_modes do
+    FilterEngine.validate_rule(rule)
+  end
+
+  defp validate_condition(_condition), do: {:error, "condition must be either a filter or a rule"}
+end