feat(spam): Introduce checking new accounts, events & comments for spam with the help of Akismet

Signed-off-by: Thomas Citharel <tcit@tcit.fr>

lib/mobilizon/actors/actors.ex

@@ -320,8 +320,8 @@ defmodule Mobilizon.Actors do
           String.t(),
           String.t(),
           String.t(),
-          boolean,
-          boolean,
+          boolean | nil,
+          boolean | nil,
           integer | nil,
           integer | nil
         ) :: Page.t(Actor.t())
@@ -380,8 +380,8 @@ defmodule Mobilizon.Actors do
           String.t(),
           String.t(),
           String.t(),
-          boolean(),
-          boolean()
+          boolean() | nil,
+          boolean() | nil
         ) ::
           Ecto.Query.t()
   defp filter_actors(
@@ -417,10 +417,12 @@ defmodule Mobilizon.Actors do
 
   defp filter_remote(query, true), do: filter_local(query)
   defp filter_remote(query, false), do: filter_external(query)
+  defp filter_remote(query, nil), do: query
 
-  @spec filter_suspended(Ecto.Queryable.t(), boolean()) :: Ecto.Query.t()
+  @spec filter_suspended(Ecto.Queryable.t(), boolean() | nil) :: Ecto.Query.t()
   defp filter_suspended(query, true), do: where(query, [a], a.suspended)
   defp filter_suspended(query, false), do: where(query, [a], not a.suspended)
+  defp filter_suspended(query, nil), do: query
 
   @spec filter_out_anonymous_actor_id(Ecto.Queryable.t(), integer() | String.t()) ::
           Ecto.Query.t()
@@ -1766,4 +1768,26 @@ defmodule Mobilizon.Actors do
     )
     |> Repo.all()
   end
+
+  @spec stream_persons(
+          String.t(),
+          String.t(),
+          String.t(),
+          boolean | nil,
+          boolean | nil,
+          integer()
+        ) :: Enum.t()
+  def stream_persons(
+        preferred_username \\ "",
+        name \\ "",
+        domain \\ "",
+        local \\ true,
+        suspended \\ false,
+        chunk_size \\ 500
+      ) do
+    person_query()
+    |> filter_actors(preferred_username, name, domain, local, suspended)
+    |> preload([:user])
+    |> Page.chunk(chunk_size)
+  end
 end

lib/mobilizon/events/events.ex

@@ -1819,7 +1819,14 @@ defmodule Mobilizon.Events do
 
   @spec filter_local(Ecto.Queryable.t()) :: Ecto.Query.t()
   defp filter_local(query) do
-    where(query, [q], q.local == true)
+    filter_local(query, true)
+  end
+
+  @spec filter_local(Ecto.Queryable.t(), boolean() | nil) :: Ecto.Query.t()
+  defp filter_local(query, nil), do: query
+
+  defp filter_local(query, value) when is_boolean(value) do
+    where(query, [q], q.local == ^value)
   end
 
   @spec filter_local_or_from_followed_instances_events(Ecto.Queryable.t()) ::
@@ -1938,4 +1945,13 @@ defmodule Mobilizon.Events do
 
   @spec preload_for_event(Ecto.Queryable.t()) :: Ecto.Query.t()
   defp preload_for_event(query), do: preload(query, ^@event_preloads)
+
+  @spec stream_events(boolean() | nil, integer()) :: Enum.t()
+  def stream_events(local \\ true, chunk_size \\ 500) do
+    Event
+    |> filter_draft()
+    |> filter_local(local)
+    |> preload_for_event()
+    |> Page.chunk(chunk_size)
+  end
 end

lib/mobilizon/storage/page.ex

@@ -48,4 +48,34 @@ defmodule Mobilizon.Storage.Page do
   def paginate(query, page, size) do
     from(query, limit: ^size, offset: ^((page - 1) * size))
   end
+
+  @doc """
+  Stream chunks of results from the given queryable.
+
+  Unlike Repo.stream, this function does not keep a long running transaction open.
+  Hence, consistency is not guarenteed in the presence of rows being deleted or sort criteria changing.
+
+  ## Example
+
+    Ecto.Query.from(u in Users, order_by: [asc: :created_at])
+    |> Repo.chunk(100)
+    |> Stream.map(&process_batch_of_users)
+    |> Stream.run()
+
+  ## Source
+  https://elixirforum.com/t/what-is-the-best-approach-for-fetching-large-amount-of-records-from-postgresql-with-ecto/3766/8
+  """
+  @spec chunk(Ecto.Queryable.t(), integer) :: Stream.t()
+  def chunk(queryable, chunk_size) do
+    chunk_stream =
+      Stream.unfold(1, fn page_number ->
+        page = queryable |> paginate(page_number, chunk_size) |> Repo.all()
+        {page, page_number + 1}
+      end)
+
+    Stream.take_while(chunk_stream, fn
+      [] -> false
+      _ -> true
+    end)
+  end
 end