Potential optimisation

packages/sync-service/lib/electric/shapes/consumer.ex

@@ -70,9 +70,7 @@ defmodule Electric.Shapes.Consumer do
   end
 
   defp selector(%Transaction{changes: changes}, shape) do
-    changes
-    |> Stream.flat_map(&Shape.convert_change(shape, &1))
-    |> Enum.any?()
+    Shape.any_relevant?(shape, changes)
   end
 
   defp selector(_, _), do: false

packages/sync-service/lib/electric/shapes/shape.ex

@@ -164,6 +164,40 @@ defmodule Electric.Shapes.Shape do
     end
   end
 
+  def any_relevant?(_shape, []), do: false
+
+  def any_relevant?(shape, [change | changes]) do
+    if relevant?(shape, change) do
+      true
+    else
+      any_relevant?(shape, changes)
+    end
+  end
+
+  defp relevant?(%__MODULE__{root_table: table}, %{relation: relation})
+       when table != relation,
+       do: false
+
+  defp relevant?(%__MODULE__{where: nil}, _change), do: true
+  defp relevant?(%__MODULE__{where: _}, %Changes.TruncatedRelation{}), do: true
+
+  defp relevant?(%__MODULE__{where: where}, change)
+       when is_struct(change, Changes.NewRecord)
+       when is_struct(change, Changes.DeletedRecord) do
+    record = if is_struct(change, Changes.NewRecord), do: change.record, else: change.old_record
+    record_in_shape?(where, record)
+  end
+
+  defp relevant?(
+         %__MODULE__{where: where},
+         %Changes.UpdatedRecord{old_record: old_record, record: record}
+       ) do
+    old_record_in_shape = record_in_shape?(where, old_record)
+    new_record_in_shape = record_in_shape?(where, record)
+
+    old_record_in_shape || new_record_in_shape
+  end
+
   defp record_in_shape?(where, record) do
     with {:ok, refs} <- Runner.record_to_ref_values(where.used_refs, record),
          {:ok, evaluated} <- Runner.execute(where, refs) do