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

lib/frenzy/pipeline/filter_engine.ex

@@ -0,0 +1,120 @@
+defmodule Frenzy.Pipeline.FilterEngine do
+  def validate_filter(filter) when is_map(filter) do
+    cond do
+      not is_map(filter) ->
+        {:error, "filter must be a map"}
+
+      not (Map.has_key?(filter, "mode") and is_binary(filter["mode"]) and
+               filter["mode"] in ["accept", "reject"]) ->
+        {:error, "mode must be a string, either 'accept' or 'reject'"}
+
+      not (Map.has_key?(filter, "score") and is_integer(filter["score"])) ->
+        {:error, "score must be an integer"}
+
+      not (Map.has_key?(filter, "rules") and is_list(filter["rules"])) ->
+        {:error, "rules must be a list of rules"}
+
+      true ->
+        validate_rules(filter)
+    end
+  end
+
+  def validate_rules(%{"rules" => rules} = filter) do
+    case validate_rules(rules) do
+      {:ok, rules} ->
+        {:ok, Map.put(filter, "rules", rules)}
+
+      {:error, _reason} = err ->
+        err
+    end
+  end
+
+  def validate_rules(rules) when is_list(rules) do
+    rules
+    |> Enum.with_index()
+    |> Enum.reduce_while({:ok, []}, fn {rule, index}, {:ok, new_rules} ->
+      case validate_rule(rule) do
+        {:ok, rule} ->
+          {:cont, {:ok, [rule | new_rules]}}
+
+        {:error, reason} ->
+          {:halt, {:error, "invalid rule #{index}: #{reason}"}}
+      end
+    end)
+  end
+
+  def validate_rules(_rules), do: {:error, "rules must be a list"}
+
+  def validate_rule(rule) do
+    cond do
+      not is_map(rule) ->
+        {:error, "rule must be a map"}
+
+      not (Map.has_key?(rule, "mode") and is_binary(rule["mode"]) and
+               rule["mode"] in ["contains_string", "matches_regex"]) ->
+        {:error, "mode property must be a string, either 'contains_string' or 'matches_regex'"}
+
+      not (Map.has_key?(rule, "property") and is_binary(rule["property"]) and
+               rule["property"] in ["url", "title", "author"]) ->
+        {:error, "property property must be a string, either 'url', 'title', or 'author'"}
+
+      not (Map.has_key?(rule, "param") and is_binary(rule["param"])) ->
+        {:error, "param property must be a string"}
+
+      not (Map.has_key?(rule, "weight") and is_integer(rule["weight"])) ->
+        {:error, "weight property must be an integer"}
+
+      true ->
+        {:ok, rule}
+    end
+  end
+
+  def test(%{"mode" => mode, "score" => score, "rules" => rules}, item_params) do
+    item_score =
+      rules
+      |> Enum.filter(&test_rule(&1, item_params))
+      |> Enum.map(& &1["weight"])
+      |> Enum.sum()
+
+    matches = item_score >= score
+
+    case {mode, matches} do
+      {"accept", true} ->
+        true
+
+      {"reject", false} ->
+        true
+
+      _ ->
+        false
+    end
+  end
+
+  def test_rule(
+        %{"mode" => mode, "property" => property, "param" => param},
+        item_params
+      ) do
+    with prop_value <- get_property(item_params, property),
+         true <- is_binary(prop_value),
+         true <- matches(prop_value, mode, param) do
+      true
+    else
+      _ ->
+        false
+    end
+  end
+
+  defp matches(value, "contains_string", param) do
+    String.contains?(value, param)
+  end
+
+  defp matches(value, "matches_regex", param) do
+    {:ok, regex} = Regex.compile(param)
+    String.match?(value, regex)
+  end
+
+  defp get_property(item_params, "url"), do: item_params.url
+  defp get_property(item_params, "title"), do: item_params.title
+  defp get_property(item_params, "author"), do: item_params.author
+  defp get_property(_item_params, _property), do: {:error, "invalid property"}
+end

lib/frenzy/pipeline/filter_stage.ex

@@ -1,128 +1,20 @@
 defmodule Frenzy.Pipeline.FilterStage do
   require Logger
-  alias Frenzy.Pipeline.Stage
+  alias Frenzy.Pipeline.{Stage, FilterEngine}
   @behaviour Stage
 
   @impl Stage
-  def apply(%{"mode" => mode, "score" => score, "rules" => rules}, item_params)
+  def apply(filter, item_params) do
-      when is_binary(mode) and is_integer(score) and is_list(rules) do
+    if FilterEngine.test(filter, item_params) do
-    item_score =
+      {:ok, item_params}
-      rules
-      |> Enum.map(fn rule -> test(rule, item_params) end)
-      |> Enum.sum()
-
-    matches = item_score >= score
-
-    case {mode, matches} do
-      {"accept", true} ->
-        {:ok, item_params}
-
-      {"reject", false} ->
-        {:ok, item_params}
-
-      _ ->
-        Logger.debug("Skipping item #{item_params.url} due to feed filter")
-        :tombstone
-    end
-  end
-
-  @impl Stage
-  def apply(opts, item_params) do
-    Logger.warn("Received invalid filter opts: #{opts}")
-    {:ok, item_params}
-  end
-
-  @impl Stage
-  def validate_opts(opts) when is_map(opts) do
-    cond do
-      not (Map.has_key?(opts, "mode") and is_binary(opts["mode"]) and
-               opts["mode"] in ["accept", "reject"]) ->
-        {:error, "mode must be a string, either 'accept' or 'reject'"}
-
-      not (Map.has_key?(opts, "score") and is_integer(opts["score"])) ->
-        {:error, "score must be an integer"}
-
-      not (Map.has_key?(opts, "rules") and is_list(opts["rules"])) ->
-        {:error, "rules must be a list of rules"}
-
-      true ->
-        validate_rules(opts)
-    end
-  end
-
-  @impl Stage
-  def validate_opts(_opts), do: {:error, "options must be a map"}
-
-  defp validate_rules(%{"rules" => rules} = opts) do
-    rules
-    |> Enum.with_index()
-    |> Enum.reduce_while(:ok, fn {rule, index}, :ok ->
-      case validate_rule(rule) do
-        :ok ->
-          {:cont, :ok}
-
-        {:error, reason} ->
-          {:halt, {:error, "invalid rule #{index}: #{reason}"}}
-      end
-    end)
-    |> case do
-      :ok ->
-        {:ok, opts}
-
-      {:error, _reason} = err ->
-        err
-    end
-  end
-
-  defp validate_rule(rule) do
-    cond do
-      not is_map(rule) ->
-        {:error, "rule must be a map"}
-
-      not (Map.has_key?(rule, "mode") and is_binary(rule["mode"]) and
-               rule["mode"] in ["contains_string", "matches_regex"]) ->
-        {:error, "mode property must be a string, either 'contains_string' or 'matches_regex'"}
-
-      not (Map.has_key?(rule, "property") and is_binary(rule["property"]) and
-               rule["property"] in ["url", "title", "author"]) ->
-        {:error, "property property must be a string, either 'url', 'title', or 'author'"}
-
-      not (Map.has_key?(rule, "param") and is_binary(rule["param"])) ->
-        {:error, "param property must be a string"}
-
-      not (Map.has_key?(rule, "weight") and is_integer(rule["weight"])) ->
-        {:error, "weight property must be an integer"}
-
-      true ->
-        :ok
-    end
-  end
-
-  defp test(
-         %{"mode" => mode, "property" => property, "param" => param, "weight" => weight},
-         item_params
-       ) do
-    with prop_value <- get_property(item_params, property),
-         true <- is_binary(prop_value),
-         true <- matches(prop_value, mode, param) do
-      weight
     else
-      _ ->
+      Logger.debug("Skipping item #{item_params.url} due to feed filter")
-        0
+      :tombstone
     end
   end
 
-  def matches(value, "contains_string", param) do
+  @impl Stage
-    String.contains?(value, param)
+  def validate_opts(opts) do
+    FilterEngine.validate_filter(opts)
   end
-
-  def matches(value, "matches_regex", param) do
-    {:ok, regex} = Regex.compile(param)
-    String.match?(value, regex)
-  end
-
-  defp get_property(item_params, "url"), do: item_params.url
-  defp get_property(item_params, "title"), do: item_params.title
-  defp get_property(item_params, "author"), do: item_params.author
-  defp get_property(_item_params, _property), do: {:error, "invalid property"}
 end