moving to separate dir

4 files changed, 726 insertions, 0 deletions

diff --git a/benchmarks/load_testing/fetcher.ex b/benchmarks/load_testing/fetcher.ex
new file mode 100644
index 000000000..0ff2f28c0
--- /dev/null
+++ b/benchmarks/load_testing/fetcher.ex
@@ -0,0 +1,229 @@
+defmodule Pleroma.LoadTesting.Fetcher do
+  use Pleroma.LoadTesting.Helper
+
+  def fetch_user(user) do
+    Benchee.run(%{
+      "By id" => fn -> Repo.get_by(User, id: user.id) end,
+      "By ap_id" => fn -> Repo.get_by(User, ap_id: user.ap_id) end,
+      "By email" => fn -> Repo.get_by(User, email: user.email) end,
+      "By nickname" => fn -> Repo.get_by(User, nickname: user.nickname) end
+    })
+  end
+
+  def query_timelines(user) do
+    home_timeline_params = %{
+      "count" => 20,
+      "with_muted" => true,
+      "type" => ["Create", "Announce"],
+      "blocking_user" => user,
+      "muting_user" => user,
+      "user" => user
+    }
+
+    mastodon_public_timeline_params = %{
+      "count" => 20,
+      "local_only" => true,
+      "only_media" => "false",
+      "type" => ["Create", "Announce"],
+      "with_muted" => "true",
+      "blocking_user" => user,
+      "muting_user" => user
+    }
+
+    mastodon_federated_timeline_params = %{
+      "count" => 20,
+      "only_media" => "false",
+      "type" => ["Create", "Announce"],
+      "with_muted" => "true",
+      "blocking_user" => user,
+      "muting_user" => user
+    }
+
+    Benchee.run(%{
+      "User home timeline" => fn ->
+        Pleroma.Web.ActivityPub.ActivityPub.fetch_activities(
+          [user.ap_id | user.following],
+          home_timeline_params
+        )
+      end,
+      "User mastodon public timeline" => fn ->
+        Pleroma.Web.ActivityPub.ActivityPub.fetch_public_activities(
+          mastodon_public_timeline_params
+        )
+      end,
+      "User mastodon federated public timeline" => fn ->
+        Pleroma.Web.ActivityPub.ActivityPub.fetch_public_activities(
+          mastodon_federated_timeline_params
+        )
+      end
+    })
+
+    home_activities =
+      Pleroma.Web.ActivityPub.ActivityPub.fetch_activities(
+        [user.ap_id | user.following],
+        home_timeline_params
+      )
+
+    public_activities =
+      Pleroma.Web.ActivityPub.ActivityPub.fetch_public_activities(mastodon_public_timeline_params)
+
+    public_federated_activities =
+      Pleroma.Web.ActivityPub.ActivityPub.fetch_public_activities(
+        mastodon_federated_timeline_params
+      )
+
+    Benchee.run(%{
+      "Rendering home timeline" => fn ->
+        Pleroma.Web.MastodonAPI.StatusView.render("index.json", %{
+          activities: home_activities,
+          for: user,
+          as: :activity
+        })
+      end,
+      "Rendering public timeline" => fn ->
+        Pleroma.Web.MastodonAPI.StatusView.render("index.json", %{
+          activities: public_activities,
+          for: user,
+          as: :activity
+        })
+      end,
+      "Rendering public federated timeline" => fn ->
+        Pleroma.Web.MastodonAPI.StatusView.render("index.json", %{
+          activities: public_federated_activities,
+          for: user,
+          as: :activity
+        })
+      end
+    })
+  end
+
+  def query_notifications(user) do
+    without_muted_params = %{"count" => "20", "with_muted" => "false"}
+    with_muted_params = %{"count" => "20", "with_muted" => "true"}
+
+    Benchee.run(%{
+      "Notifications without muted" => fn ->
+        Pleroma.Web.MastodonAPI.MastodonAPI.get_notifications(user, without_muted_params)
+      end,
+      "Notifications with muted" => fn ->
+        Pleroma.Web.MastodonAPI.MastodonAPI.get_notifications(user, with_muted_params)
+      end
+    })
+
+    without_muted_notifications =
+      Pleroma.Web.MastodonAPI.MastodonAPI.get_notifications(user, without_muted_params)
+
+    with_muted_notifications =
+      Pleroma.Web.MastodonAPI.MastodonAPI.get_notifications(user, with_muted_params)
+
+    Benchee.run(%{
+      "Render notifications without muted" => fn ->
+        Pleroma.Web.MastodonAPI.NotificationView.render("index.json", %{
+          notifications: without_muted_notifications,
+          for: user
+        })
+      end,
+      "Render notifications with muted" => fn ->
+        Pleroma.Web.MastodonAPI.NotificationView.render("index.json", %{
+          notifications: with_muted_notifications,
+          for: user
+        })
+      end
+    })
+  end
+
+  def query_dms(user) do
+    params = %{
+      "count" => "20",
+      "with_muted" => "true",
+      "type" => "Create",
+      "blocking_user" => user,
+      "user" => user,
+      visibility: "direct"
+    }
+
+    Benchee.run(%{
+      "Direct messages with muted" => fn ->
+        Pleroma.Web.ActivityPub.ActivityPub.fetch_activities_query([user.ap_id], params)
+        |> Pleroma.Pagination.fetch_paginated(params)
+      end,
+      "Direct messages without muted" => fn ->
+        Pleroma.Web.ActivityPub.ActivityPub.fetch_activities_query([user.ap_id], params)
+        |> Pleroma.Pagination.fetch_paginated(Map.put(params, "with_muted", false))
+      end
+    })
+
+    dms_with_muted =
+      Pleroma.Web.ActivityPub.ActivityPub.fetch_activities_query([user.ap_id], params)
+      |> Pleroma.Pagination.fetch_paginated(params)
+
+    dms_without_muted =
+      Pleroma.Web.ActivityPub.ActivityPub.fetch_activities_query([user.ap_id], params)
+      |> Pleroma.Pagination.fetch_paginated(Map.put(params, "with_muted", false))
+
+    Benchee.run(%{
+      "Rendering dms with muted" => fn ->
+        Pleroma.Web.MastodonAPI.StatusView.render("index.json", %{
+          activities: dms_with_muted,
+          for: user,
+          as: :activity
+        })
+      end,
+      "Rendering dms without muted" => fn ->
+        Pleroma.Web.MastodonAPI.StatusView.render("index.json", %{
+          activities: dms_without_muted,
+          for: user,
+          as: :activity
+        })
+      end
+    })
+  end
+
+  def query_long_thread(user, activity) do
+    Benchee.run(%{
+      "Fetch main post" => fn ->
+        Pleroma.Activity.get_by_id_with_object(activity.id)
+      end,
+      "Fetch context of main post" => fn ->
+        Pleroma.Web.ActivityPub.ActivityPub.fetch_activities_for_context(
+          activity.data["context"],
+          %{
+            "blocking_user" => user,
+            "user" => user,
+            "exclude_id" => activity.id
+          }
+        )
+      end
+    })
+
+    activity = Pleroma.Activity.get_by_id_with_object(activity.id)
+
+    context =
+      Pleroma.Web.ActivityPub.ActivityPub.fetch_activities_for_context(
+        activity.data["context"],
+        %{
+          "blocking_user" => user,
+          "user" => user,
+          "exclude_id" => activity.id
+        }
+      )
+
+    Benchee.run(%{
+      "Render status" => fn ->
+        Pleroma.Web.MastodonAPI.StatusView.render("status.json", %{
+          activity: activity,
+          for: user
+        })
+      end,
+      "Render context" => fn ->
+        Pleroma.Web.MastodonAPI.StatusView.render(
+          "index.json",
+          for: user,
+          activities: context,
+          as: :activity
+        )
+        |> Enum.reverse()
+      end
+    })
+  end
+end
diff --git a/benchmarks/load_testing/generator.ex b/benchmarks/load_testing/generator.ex
new file mode 100644
index 000000000..5c5a5c122
--- /dev/null
+++ b/benchmarks/load_testing/generator.ex
@@ -0,0 +1,352 @@
+defmodule Pleroma.LoadTesting.Generator do
+  use Pleroma.LoadTesting.Helper
+  alias Pleroma.Web.CommonAPI
+
+  def generate_users(opts) do
+    IO.puts("Starting generating #{opts[:users_max]} users...")
+    {time, _} = :timer.tc(fn -> do_generate_users(opts) end)
+
+    IO.puts("Inserting users take #{to_sec(time)} sec.\n")
+  end
+
+  defp do_generate_users(opts) do
+    max = Keyword.get(opts, :users_max)
+
+    Task.async_stream(
+      1..max,
+      &generate_user_data(&1),
+      max_concurrency: 10,
+      timeout: 30_000
+    )
+    |> Enum.to_list()
+  end
+
+  defp generate_user_data(i) do
+    remote = Enum.random([true, false])
+
+    user = %User{
+      name: "Test テスト User #{i}",
+      email: "user#{i}@example.com",
+      nickname: "nick#{i}",
+      password_hash:
+        "$pbkdf2-sha512$160000$bU.OSFI7H/yqWb5DPEqyjw$uKp/2rmXw12QqnRRTqTtuk2DTwZfF8VR4MYW2xMeIlqPR/UX1nT1CEKVUx2CowFMZ5JON8aDvURrZpJjSgqXrg",
+      bio: "Tester Number #{i}",
+      info: %{},
+      local: remote
+    }
+
+    user_urls =
+      if remote do
+        base_url =
+          Enum.random(["https://domain1.com", "https://domain2.com", "https://domain3.com"])
+
+        ap_id = "#{base_url}/users/#{user.nickname}"
+
+        %{
+          ap_id: ap_id,
+          follower_address: ap_id <> "/followers",
+          following_address: ap_id <> "/following",
+          following: [ap_id]
+        }
+      else
+        %{
+          ap_id: User.ap_id(user),
+          follower_address: User.ap_followers(user),
+          following_address: User.ap_following(user),
+          following: [User.ap_id(user)]
+        }
+      end
+
+    user = Map.merge(user, user_urls)
+
+    Repo.insert!(user)
+  end
+
+  def generate_activities(user, users) do
+    do_generate_activities(user, users)
+  end
+
+  defp do_generate_activities(user, users) do
+    IO.puts("Starting generating 20000 common activities...")
+
+    {time, _} =
+      :timer.tc(fn ->
+        Task.async_stream(
+          1..20_000,
+          fn _ ->
+            do_generate_activity([user | users])
+          end,
+          max_concurrency: 10,
+          timeout: 30_000
+        )
+        |> Stream.run()
+      end)
+
+    IO.puts("Inserting common activities take #{to_sec(time)} sec.\n")
+
+    IO.puts("Starting generating 20000 activities with mentions...")
+
+    {time, _} =
+      :timer.tc(fn ->
+        Task.async_stream(
+          1..20_000,
+          fn _ ->
+            do_generate_activity_with_mention(user, users)
+          end,
+          max_concurrency: 10,
+          timeout: 30_000
+        )
+        |> Stream.run()
+      end)
+
+    IO.puts("Inserting activities with menthions take #{to_sec(time)} sec.\n")
+
+    IO.puts("Starting generating 10000 activities with threads...")
+
+    {time, _} =
+      :timer.tc(fn ->
+        Task.async_stream(
+          1..10_000,
+          fn _ ->
+            do_generate_threads([user | users])
+          end,
+          max_concurrency: 10,
+          timeout: 30_000
+        )
+        |> Stream.run()
+      end)
+
+    IO.puts("Inserting activities with threads take #{to_sec(time)} sec.\n")
+  end
+
+  defp do_generate_activity(users) do
+    post = %{
+      "status" => "Some status without mention with random user"
+    }
+
+    CommonAPI.post(Enum.random(users), post)
+  end
+
+  defp do_generate_activity_with_mention(user, users) do
+    mentions_cnt = Enum.random([2, 3, 4, 5])
+    with_user = Enum.random([true, false])
+    users = Enum.shuffle(users)
+    mentions_users = Enum.take(users, mentions_cnt)
+    mentions_users = if with_user, do: [user | mentions_users], else: mentions_users
+
+    mentions_str =
+      Enum.map(mentions_users, fn user -> "@" <> user.nickname end) |> Enum.join(", ")
+
+    post = %{
+      "status" => mentions_str <> "some status with mentions random users"
+    }
+
+    CommonAPI.post(Enum.random(users), post)
+  end
+
+  defp do_generate_threads(users) do
+    thread_length = Enum.random([2, 3, 4, 5])
+    actor = Enum.random(users)
+
+    post = %{
+      "status" => "Start of the thread"
+    }
+
+    {:ok, activity} = CommonAPI.post(actor, post)
+
+    Enum.each(1..thread_length, fn _ ->
+      user = Enum.random(users)
+
+      post = %{
+        "status" => "@#{actor.nickname} reply to thread",
+        "in_reply_to_status_id" => activity.id
+      }
+
+      CommonAPI.post(user, post)
+    end)
+  end
+
+  def generate_remote_activities(user, users) do
+    do_generate_remote_activities(user, users)
+  end
+
+  defp do_generate_remote_activities(user, users) do
+    IO.puts("Starting generating 10000 remote activities...")
+
+    {time, _} =
+      :timer.tc(fn ->
+        Task.async_stream(
+          1..10_000,
+          fn i ->
+            do_generate_remote_activity(i, user, users)
+          end,
+          max_concurrency: 10,
+          timeout: 30_000
+        )
+        |> Stream.run()
+      end)
+
+    IO.puts("Inserting remote activities take #{to_sec(time)} sec.\n")
+  end
+
+  defp do_generate_remote_activity(i, user, users) do
+    actor = Enum.random(users)
+    %{host: host} = URI.parse(actor.ap_id)
+    date = Date.utc_today()
+    datetime = DateTime.utc_now()
+
+    map = %{
+      "actor" => actor.ap_id,
+      "cc" => [actor.follower_address, user.ap_id],
+      "context" => "tag:mastodon.example.org,#{date}:objectId=#{i}:objectType=Conversation",
+      "id" => actor.ap_id <> "/statuses/#{i}/activity",
+      "object" => %{
+        "actor" => actor.ap_id,
+        "atomUri" => actor.ap_id <> "/statuses/#{i}",
+        "attachment" => [],
+        "attributedTo" => actor.ap_id,
+        "bcc" => [],
+        "bto" => [],
+        "cc" => [actor.follower_address, user.ap_id],
+        "content" =>
+          "<p><span class=\"h-card\"><a href=\"" <>
+            user.ap_id <>
+            "\" class=\"u-url mention\">@<span>" <> user.nickname <> "</span></a></span></p>",
+        "context" => "tag:mastodon.example.org,#{date}:objectId=#{i}:objectType=Conversation",
+        "conversation" =>
+          "tag:mastodon.example.org,#{date}:objectId=#{i}:objectType=Conversation",
+        "emoji" => %{},
+        "id" => actor.ap_id <> "/statuses/#{i}",
+        "inReplyTo" => nil,
+        "inReplyToAtomUri" => nil,
+        "published" => datetime,
+        "sensitive" => true,
+        "summary" => "cw",
+        "tag" => [
+          %{
+            "href" => user.ap_id,
+            "name" => "@#{user.nickname}@#{host}",
+            "type" => "Mention"
+          }
+        ],
+        "to" => ["https://www.w3.org/ns/activitystreams#Public"],
+        "type" => "Note",
+        "url" => "http://#{host}/@#{actor.nickname}/#{i}"
+      },
+      "published" => datetime,
+      "to" => ["https://www.w3.org/ns/activitystreams#Public"],
+      "type" => "Create"
+    }
+
+    Pleroma.Web.ActivityPub.ActivityPub.insert(map, false)
+  end
+
+  def generate_dms(user, users, opts) do
+    IO.puts("Starting generating #{opts[:dms_max]} DMs")
+    {time, _} = :timer.tc(fn -> do_generate_dms(user, users, opts) end)
+    IO.puts("Inserting dms take #{to_sec(time)} sec.\n")
+  end
+
+  defp do_generate_dms(user, users, opts) do
+    Task.async_stream(
+      1..opts[:dms_max],
+      fn _ ->
+        do_generate_dm(user, users)
+      end,
+      max_concurrency: 10,
+      timeout: 30_000
+    )
+    |> Stream.run()
+  end
+
+  defp do_generate_dm(user, users) do
+    post = %{
+      "status" => "@#{user.nickname} some direct message",
+      "visibility" => "direct"
+    }
+
+    CommonAPI.post(Enum.random(users), post)
+  end
+
+  def generate_long_thread(user, users, opts) do
+    IO.puts("Starting generating long thread with #{opts[:thread_length]} replies")
+    {time, activity} = :timer.tc(fn -> do_generate_long_thread(user, users, opts) end)
+    IO.puts("Inserting long thread replies take #{to_sec(time)} sec.\n")
+    {:ok, activity}
+  end
+
+  defp do_generate_long_thread(user, users, opts) do
+    {:ok, %{id: id} = activity} = CommonAPI.post(user, %{"status" => "Start of long thread"})
+
+    Task.async_stream(
+      1..opts[:thread_length],
+      fn _ -> do_generate_thread(users, id) end,
+      max_concurrency: 10,
+      timeout: 30_000
+    )
+    |> Stream.run()
+
+    activity
+  end
+
+  defp do_generate_thread(users, activity_id) do
+    CommonAPI.post(Enum.random(users), %{
+      "status" => "reply to main post",
+      "in_reply_to_status_id" => activity_id
+    })
+  end
+
+  def generate_non_visible_message(user, users) do
+    IO.puts("Starting generating 1000 non visible posts")
+
+    {time, _} =
+      :timer.tc(fn ->
+        do_generate_non_visible_posts(user, users)
+      end)
+
+    IO.puts("Inserting non visible posts take #{to_sec(time)} sec.\n")
+  end
+
+  defp do_generate_non_visible_posts(user, users) do
+    [not_friend | users] = users
+
+    make_friends(user, users)
+
+    Task.async_stream(1..1000, fn _ -> do_generate_non_visible_post(not_friend, users) end,
+      max_concurrency: 10,
+      timeout: 30_000
+    )
+    |> Stream.run()
+  end
+
+  defp make_friends(_user, []), do: nil
+
+  defp make_friends(user, [friend | users]) do
+    {:ok, _} = User.follow(user, friend)
+    {:ok, _} = User.follow(friend, user)
+    make_friends(user, users)
+  end
+
+  defp do_generate_non_visible_post(not_friend, users) do
+    post = %{
+      "status" => "some non visible post",
+      "visibility" => "private"
+    }
+
+    {:ok, activity} = CommonAPI.post(not_friend, post)
+
+    thread_length = Enum.random([2, 3, 4, 5])
+
+    Enum.each(1..thread_length, fn _ ->
+      user = Enum.random(users)
+
+      post = %{
+        "status" => "@#{not_friend.nickname} reply to non visible post",
+        "in_reply_to_status_id" => activity.id,
+        "visibility" => "private"
+      }
+
+      CommonAPI.post(user, post)
+    end)
+  end
+end
diff --git a/benchmarks/load_testing/helper.ex b/benchmarks/load_testing/helper.ex
new file mode 100644
index 000000000..47b25c65f
--- /dev/null
+++ b/benchmarks/load_testing/helper.ex
@@ -0,0 +1,11 @@
+defmodule Pleroma.LoadTesting.Helper do
+  defmacro __using__(_) do
+    quote do
+      import Ecto.Query
+      alias Pleroma.Repo
+      alias Pleroma.User
+
+      defp to_sec(microseconds), do: microseconds / 1_000_000
+    end
+  end
+end
diff --git a/benchmarks/mix/tasks/pleroma/load_testing.ex b/benchmarks/mix/tasks/pleroma/load_testing.ex
new file mode 100644
index 000000000..7b2293acc
--- /dev/null
+++ b/benchmarks/mix/tasks/pleroma/load_testing.ex
@@ -0,0 +1,134 @@
+defmodule Mix.Tasks.Pleroma.LoadTesting do
+  use Mix.Task
+  use Pleroma.LoadTesting.Helper
+  import Mix.Pleroma
+  import Pleroma.LoadTesting.Generator
+  import Pleroma.LoadTesting.Fetcher
+
+  @shortdoc "Factory for generation data"
+  @moduledoc """
+  Generates data like:
+  - local/remote users
+  - local/remote activities with notifications
+  - direct messages
+  - long thread
+  - non visible posts
+
+  ## Generate data
+      MIX_ENV=benchmark mix pleroma.load_testing --users 20000 --dms 20000 --thread_length 2000
+      MIX_ENV=benchmark mix pleroma.load_testing -u 20000 -d 20000 -t 2000
+
+  Options:
+  - `--users NUMBER` - number of users to generate. Defaults to: 20000. Alias: `-u`
+  - `--dms NUMBER` - number of direct messages to generate. Defaults to: 20000. Alias `-d`
+  - `--thread_length` - number of messages in thread. Defaults to: 2000. ALias `-t`
+  """
+
+  @aliases [u: :users, d: :dms, t: :thread_length]
+  @switches [
+    users: :integer,
+    dms: :integer,
+    thread_length: :integer
+  ]
+  @users_default 20_000
+  @dms_default 20_000
+  @thread_length_default 2_000
+
+  def run(args) do
+    start_pleroma()
+    Pleroma.Config.put([:instance, :skip_thread_containment], true)
+    {opts, _} = OptionParser.parse!(args, strict: @switches, aliases: @aliases)
+
+    users_max = Keyword.get(opts, :users, @users_default)
+    dms_max = Keyword.get(opts, :dms, @dms_default)
+    thread_length = Keyword.get(opts, :thread_length, @thread_length_default)
+
+    clean_tables()
+
+    opts =
+      Keyword.put(opts, :users_max, users_max)
+      |> Keyword.put(:dms_max, dms_max)
+      |> Keyword.put(:thread_length, thread_length)
+
+    generate_users(opts)
+
+    # main user for queries
+    IO.puts("Fetching local main user...")
+
+    {time, user} =
+      :timer.tc(fn ->
+        Repo.one(
+          from(u in User, where: u.local == true, order_by: fragment("RANDOM()"), limit: 1)
+        )
+      end)
+
+    IO.puts("Fetching main user take #{to_sec(time)} sec.\n")
+
+    IO.puts("Fetching local users...")
+
+    {time, users} =
+      :timer.tc(fn ->
+        Repo.all(
+          from(u in User,
+            where: u.id != ^user.id,
+            where: u.local == true,
+            order_by: fragment("RANDOM()"),
+            limit: 10
+          )
+        )
+      end)
+
+    IO.puts("Fetching local users take #{to_sec(time)} sec.\n")
+
+    IO.puts("Fetching remote users...")
+
+    {time, remote_users} =
+      :timer.tc(fn ->
+        Repo.all(
+          from(u in User,
+            where: u.id != ^user.id,
+            where: u.local == false,
+            order_by: fragment("RANDOM()"),
+            limit: 10
+          )
+        )
+      end)
+
+    IO.puts("Fetching remote users take #{to_sec(time)} sec.\n")
+
+    generate_activities(user, users)
+
+    generate_remote_activities(user, remote_users)
+
+    generate_dms(user, users, opts)
+
+    {:ok, activity} = generate_long_thread(user, users, opts)
+
+    generate_non_visible_message(user, users)
+
+    IO.puts("Users in DB: #{Repo.aggregate(from(u in User), :count, :id)}")
+
+    IO.puts("Activities in DB: #{Repo.aggregate(from(a in Pleroma.Activity), :count, :id)}")
+
+    IO.puts("Objects in DB: #{Repo.aggregate(from(o in Pleroma.Object), :count, :id)}")
+
+    IO.puts(
+      "Notifications in DB: #{Repo.aggregate(from(n in Pleroma.Notification), :count, :id)}"
+    )
+
+    fetch_user(user)
+    query_timelines(user)
+    query_notifications(user)
+    query_dms(user)
+    query_long_thread(user, activity)
+    Pleroma.Config.put([:instance, :skip_thread_containment], false)
+    query_timelines(user)
+  end
+
+  defp clean_tables do
+    IO.puts("Deleting old data...\n")
+    Ecto.Adapters.SQL.query!(Repo, "TRUNCATE users CASCADE;")
+    Ecto.Adapters.SQL.query!(Repo, "TRUNCATE activities CASCADE;")
+    Ecto.Adapters.SQL.query!(Repo, "TRUNCATE objects CASCADE;")
+  end
+end