Merge pull request 'Apply architecture review quick wins' (#7941) from architecture-quick-wins into develop

Reviewed-on: https://git.pleroma.social/pleroma/pleroma/pulls/7941

changelog.d/oauth-user-preload.change

@@ -0,0 +1 @@
+Avoid a duplicate user query during OAuth authentication.

changelog.d/publisher-worker-atoms.fix

@@ -0,0 +1 @@
+Avoid creating atoms from serialized PublisherWorker parameters.

changelog.d/stream-domain-block-following.fix

@@ -0,0 +1 @@
+Keep followed users visible in streams when their domain is blocked.

changelog.d/user-cache-keys.fix

@@ -0,0 +1 @@
+Fix inconsistent User cache keys and invalidate numeric ID aliases.

lib/pleroma/user.ex

@@ -272,13 +272,13 @@ defmodule Pleroma.User do
   end
 
   def cached_blocked_users_ap_ids(user) do
-    @cachex.fetch!(:user_cache, "blocked_users_ap_ids:#{user.ap_id}", fn _ ->
+    @cachex.fetch!(:user_cache, cache_key(:blocked_users_ap_ids, user), fn _ ->
       blocked_users_ap_ids(user)
     end)
   end
 
   def cached_muted_users_ap_ids(user) do
-    @cachex.fetch!(:user_cache, "muted_users_ap_ids:#{user.ap_id}", fn _ ->
+    @cachex.fetch!(:user_cache, cache_key(:muted_users_ap_ids, user), fn _ ->
       muted_users_ap_ids(user)
     end)
   end
@@ -1253,13 +1253,18 @@ defmodule Pleroma.User do
     get_cached_by_nickname(nickname)
   end
 
+  defp cache_key(:id, %User{id: id}), do: cache_key(:id, id)
+  defp cache_key(:nickname, %User{nickname: nickname}), do: cache_key(:nickname, nickname)
+  defp cache_key(namespace, %User{ap_id: ap_id}), do: cache_key(namespace, ap_id)
+  defp cache_key(namespace, value), do: "#{namespace}:#{value}"
+
   def set_cache({:ok, user}), do: set_cache(user)
   def set_cache({:error, err}), do: {:error, err}
 
   def set_cache(%User{} = user) do
-    @cachex.put(:user_cache, "ap_id:#{user.ap_id}", user)
-    @cachex.put(:user_cache, "nickname:#{user.nickname}", user)
-    @cachex.put(:user_cache, "friends_ap_ids:#{user.nickname}", get_user_friends_ap_ids(user))
+    @cachex.put(:user_cache, cache_key(:ap_id, user), user)
+    @cachex.put(:user_cache, cache_key(:nickname, user), user)
+    @cachex.put(:user_cache, cache_key(:friends_ap_ids, user), get_user_friends_ap_ids(user))
     {:ok, user}
   end
 
@@ -1287,22 +1292,23 @@ defmodule Pleroma.User do
 
   @spec get_cached_user_friends_ap_ids(User.t()) :: [String.t()]
   def get_cached_user_friends_ap_ids(user) do
-    @cachex.fetch!(:user_cache, "friends_ap_ids:#{user.ap_id}", fn _ ->
+    @cachex.fetch!(:user_cache, cache_key(:friends_ap_ids, user), fn _ ->
       get_user_friends_ap_ids(user)
     end)
   end
 
   def invalidate_cache(user) do
-    @cachex.del(:user_cache, "ap_id:#{user.ap_id}")
-    @cachex.del(:user_cache, "nickname:#{user.nickname}")
-    @cachex.del(:user_cache, "friends_ap_ids:#{user.ap_id}")
-    @cachex.del(:user_cache, "blocked_users_ap_ids:#{user.ap_id}")
-    @cachex.del(:user_cache, "muted_users_ap_ids:#{user.ap_id}")
+    @cachex.del(:user_cache, cache_key(:ap_id, user))
+    @cachex.del(:user_cache, cache_key(:nickname, user))
+    @cachex.del(:user_cache, cache_key(:id, user))
+    @cachex.del(:user_cache, cache_key(:friends_ap_ids, user))
+    @cachex.del(:user_cache, cache_key(:blocked_users_ap_ids, user))
+    @cachex.del(:user_cache, cache_key(:muted_users_ap_ids, user))
   end
 
   @spec get_cached_by_ap_id(String.t()) :: User.t() | nil
   def get_cached_by_ap_id(ap_id) do
-    key = "ap_id:#{ap_id}"
+    key = cache_key(:ap_id, ap_id)
 
     with {:ok, nil} <- @cachex.get(:user_cache, key),
          user when not is_nil(user) <- get_by_ap_id(ap_id),
@@ -1315,14 +1321,14 @@ defmodule Pleroma.User do
   end
 
   def get_cached_by_id(id) do
-    key = "id:#{id}"
+    key = cache_key(:id, id)
 
     ap_id =
       @cachex.fetch!(:user_cache, key, fn _ ->
         user = get_by_id(id)
 
         if user do
-          @cachex.put(:user_cache, "ap_id:#{user.ap_id}", user)
+          @cachex.put(:user_cache, cache_key(:ap_id, user), user)
           {:commit, user.ap_id}
         else
           {:ignore, ""}
@@ -1333,7 +1339,7 @@ defmodule Pleroma.User do
   end
 
   def get_cached_by_nickname(nickname) do
-    key = "nickname:#{nickname}"
+    key = cache_key(:nickname, nickname)
 
     @cachex.fetch!(:user_cache, key, fn _ ->
       case get_or_fetch_by_nickname(nickname) do
@@ -1651,7 +1657,7 @@ defmodule Pleroma.User do
         |> Oban.insert()
       end
 
-      @cachex.del(:user_cache, "muted_users_ap_ids:#{muter.ap_id}")
+      @cachex.del(:user_cache, cache_key(:muted_users_ap_ids, muter))
 
       {:ok, Enum.filter([user_mute, user_notification_mute], & &1)}
     end
@@ -1661,7 +1667,7 @@ defmodule Pleroma.User do
     with {:ok, user_mute} <- UserRelationship.delete_mute(muter, mutee),
          {:ok, user_notification_mute} <-
            UserRelationship.delete_notification_mute(muter, mutee) do
-      @cachex.del(:user_cache, "muted_users_ap_ids:#{muter.ap_id}")
+      @cachex.del(:user_cache, cache_key(:muted_users_ap_ids, muter))
       {:ok, [user_mute, user_notification_mute]}
     end
   end
@@ -2815,7 +2821,7 @@ defmodule Pleroma.User do
           {:ok, UserRelationship.t()} | {:error, Ecto.Changeset.t()}
   defp add_to_block(%User{} = user, %User{} = blocked, expires_at) do
     with {:ok, relationship} <- UserRelationship.create_block(user, blocked, expires_at) do
-      @cachex.del(:user_cache, "blocked_users_ap_ids:#{user.ap_id}")
+      @cachex.del(:user_cache, cache_key(:blocked_users_ap_ids, user))
       {:ok, relationship}
     end
   end
@@ -2824,7 +2830,7 @@ defmodule Pleroma.User do
           {:ok, UserRelationship.t()} | {:ok, nil} | {:error, Ecto.Changeset.t()}
   defp remove_from_block(%User{} = user, %User{} = blocked) do
     with {:ok, relationship} <- UserRelationship.delete_block(user, blocked) do
-      @cachex.del(:user_cache, "blocked_users_ap_ids:#{user.ap_id}")
+      @cachex.del(:user_cache, cache_key(:blocked_users_ap_ids, user))
       {:ok, relationship}
     end
   end

lib/pleroma/web/federator.ex

@@ -89,7 +89,8 @@ defmodule Pleroma.Web.Federator do
 
   # Job Worker Callbacks
 
-  @spec perform(atom(), any()) :: {:ok, any()} | {:error, any()}
+  @impl true
+  @spec perform(atom(), any()) :: :ok | {:ok, any()} | {:error, any()}
   def perform(:publish_one, params) do
     Publisher.prepare_one(params)
     |> Publisher.publish_one()

lib/pleroma/web/federator/publishing.ex

@@ -3,5 +3,6 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.Federator.Publishing do
+  @callback perform(atom(), any()) :: :ok | {:ok, any()} | {:error, any()}
   @callback publish(map()) :: any()
 end

lib/pleroma/web/mastodon_api/controllers/status_controller.ex

@@ -716,7 +716,10 @@ defmodule Pleroma.Web.MastodonAPI.StatusController do
     end
   end
 
-  defp put_application(params, %{assigns: %{token: %Token{user: %User{} = user} = token}} = _conn) do
+  defp put_application(
+         params,
+         %{assigns: %{token: %Token{} = token, user: %User{} = user}} = _conn
+       ) do
     if user.disclose_client do
       %{client_name: client_name, website: website} = Repo.preload(token, :app).app
       Map.put(params, :generator, %{type: "Application", name: client_name, url: website})

lib/pleroma/web/plugs/o_auth_plug.ex

@@ -52,7 +52,7 @@ defmodule Pleroma.Web.Plugs.OAuthPlug do
         where: t.token == ^token
       )
 
-    with %Token{user_id: user_id} = token_record <- Repo.one(token_query) |> Repo.preload(:user),
+    with %Token{user_id: user_id} = token_record <- Repo.one(token_query),
          false <- is_nil(user_id),
          %User{} = user <- User.get_cached_by_id(user_id) do
       {:ok, user, token_record}

lib/pleroma/web/streamer.ex

@@ -195,6 +195,9 @@ defmodule Pleroma.Web.Streamer do
     recipients = MapSet.new(item.recipients)
     domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.domain_blocks)
 
+    following_ap_ids =
+      if user.domain_blocks == [], do: [], else: User.get_cached_user_friends_ap_ids(user)
+
     with parent <- Object.normalize(item, fetch: false) || item,
          true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(item.actor not in &1)),
          true <- item.data["type"] != "Announce" || item.actor not in reblog_muted_ap_ids,
@@ -205,8 +208,12 @@ defmodule Pleroma.Web.Streamer do
          true <- MapSet.disjoint?(recipients, recipient_blocks),
          %{host: item_host} <- URI.parse(item.actor),
          %{host: parent_host} <- URI.parse(parent.data["actor"]),
-         false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
-         false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
+         false <-
+           item.actor not in following_ap_ids and
+             Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
+         false <-
+           parent.data["actor"] not in following_ap_ids and
+             Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
          true <- thread_containment(item, user),
          false <- CommonAPI.thread_muted?(parent, user) do
       false

lib/pleroma/workers/publisher_worker.ex

@@ -4,25 +4,39 @@
 
 defmodule Pleroma.Workers.PublisherWorker do
   alias Pleroma.Activity
+  alias Pleroma.Config
   alias Pleroma.Instances
   alias Pleroma.Web.Federator
 
   use Oban.Worker, queue: :federator_outgoing, max_attempts: 13
 
+  @publish_one_param_keys %{
+    "activity_id" => :activity_id,
+    "cc" => :cc,
+    "inbox" => :inbox,
+    "unreachable_since" => :unreachable_since
+  }
+
   @impl true
   def perform(%Job{args: %{"op" => "publish", "activity_id" => activity_id}}) do
     activity = Activity.get_by_id(activity_id)
-    Federator.perform(:publish, activity)
+    federator().perform(:publish, activity)
   end
 
   def perform(%Job{args: %{"op" => "publish_one", "params" => params}} = job) do
-    params = Map.new(params, fn {k, v} -> {String.to_atom(k), v} end)
+    params =
+      Enum.reduce(params, %{}, fn {key, value}, acc ->
+        case @publish_one_param_keys do
+          %{^key => atom_key} -> Map.put(acc, atom_key, value)
+          _ -> acc
+        end
+      end)
 
     # Cancel / skip the job if this server believed to be unreachable now
     if not Instances.reachable?(params.inbox) do
       {:cancel, :unreachable}
     else
-      case Federator.perform(:publish_one, params) do
+      case federator().perform(:publish_one, params) do
         {:ok, _} ->
           :ok
 
@@ -44,6 +58,8 @@ defmodule Pleroma.Workers.PublisherWorker do
   @impl true
   def timeout(_job), do: :timer.seconds(10)
 
+  defp federator, do: Config.get([:pipeline, :federator], Federator)
+
   @base_backoff 15
   @pow 5
   @impl true

test/pleroma/user_test.exs

@@ -1988,14 +1988,26 @@ defmodule Pleroma.UserTest do
   end
 
   describe "caching" do
+    test "set_cache stores friends under the AP ID cache key" do
+      user = insert(:user)
+
+      User.set_cache(user)
+
+      assert {:ok, []} = Cachex.get(:user_cache, "friends_ap_ids:#{user.ap_id}")
+      assert {:ok, nil} = Cachex.get(:user_cache, "friends_ap_ids:#{user.nickname}")
+    end
+
     test "invalidate_cache works" do
       user = insert(:user)
 
       User.set_cache(user)
+      User.get_cached_by_id(user.id)
       User.invalidate_cache(user)
 
       {:ok, nil} = Cachex.get(:user_cache, "ap_id:#{user.ap_id}")
       {:ok, nil} = Cachex.get(:user_cache, "nickname:#{user.nickname}")
+      {:ok, nil} = Cachex.get(:user_cache, "friends_ap_ids:#{user.ap_id}")
+      {:ok, nil} = Cachex.get(:user_cache, "id:#{user.id}")
     end
 
     test "User.delete() plugs any possible zombie objects" do

test/pleroma/web/plugs/o_auth_plug_test.exs

@@ -44,6 +44,18 @@ defmodule Pleroma.Web.Plugs.OAuthPlugTest do
     assert conn.assigns[:user] == opts[:user]
   end
 
+  test "with a valid token, it does not preload the duplicate user association", %{
+    conn: conn,
+    token: token
+  } do
+    conn =
+      conn
+      |> put_req_header("authorization", "bearer #{token.token}")
+      |> OAuthPlug.call(%{})
+
+    assert %Ecto.Association.NotLoaded{} = conn.assigns.token.user
+  end
+
   test "with valid token (downcase) in url parameters, it assigns the user", opts do
     conn =
       :get

test/pleroma/web/streamer_test.exs

@@ -475,6 +475,20 @@ defmodule Pleroma.Web.StreamerTest do
       assert Streamer.filtered_by_user?(user, favorite_activity)
     end
 
+    test "it does not filter followed users on blocked domains", %{user: user} do
+      followed =
+        insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/friend"})
+
+      {:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
+      {:ok, activity} = CommonAPI.post(followed, %{status: "still a friend"})
+
+      assert Streamer.filtered_by_user?(user, activity)
+
+      {:ok, user, _followed} = User.follow(user, followed)
+
+      refute Streamer.filtered_by_user?(user, activity)
+    end
+
     test "it sends follow activities to the 'user:notification' stream", %{
       user: user,
       token: oauth_token

test/pleroma/workers/publisher_worker_test.exs

@@ -6,8 +6,8 @@ defmodule Pleroma.Workers.PublisherWorkerTest do
   use Pleroma.DataCase, async: false
   use Oban.Testing, repo: Pleroma.Repo
 
+  import Mox
   import Pleroma.Factory
-  import Mock
 
   alias Pleroma.Instances
   alias Pleroma.Object
@@ -15,6 +15,8 @@ defmodule Pleroma.Workers.PublisherWorkerTest do
   alias Pleroma.Web.ActivityPub.Builder
   alias Pleroma.Web.CommonAPI
   alias Pleroma.Web.Federator
+  alias Pleroma.Web.FederatorMock
+  alias Pleroma.Workers.PublisherWorker
 
   describe "Oban job priority:" do
     setup do
@@ -55,49 +57,80 @@ defmodule Pleroma.Workers.PublisherWorkerTest do
     end
 
     test "marks server as unreachable only on final failure", %{activity: activity} do
-      with_mock Pleroma.Web.Federator,
-        perform: fn :publish_one, _params -> {:error, :connection_error} end do
-        # First attempt
-        job = %Oban.Job{
-          args: %{
-            "op" => "publish_one",
-            "params" => %{
-              "inbox" => "https://example.com/inbox",
-              "activity_id" => activity.id
-            }
-          },
-          attempt: 1,
-          max_attempts: 5
-        }
+      expect(FederatorMock, :perform, 2, fn :publish_one, _params ->
+        {:error, :connection_error}
+      end)
 
-        assert {:error, :connection_error} = Pleroma.Workers.PublisherWorker.perform(job)
-        assert Instances.reachable?("https://example.com/inbox")
+      # First attempt
+      job = %Oban.Job{
+        args: %{
+          "op" => "publish_one",
+          "params" => %{
+            "inbox" => "https://example.com/inbox",
+            "activity_id" => activity.id
+          }
+        },
+        attempt: 1,
+        max_attempts: 5
+      }
 
-        # Final attempt
-        job = %{job | attempt: 5}
-        assert {:error, :connection_error} = Pleroma.Workers.PublisherWorker.perform(job)
-        refute Instances.reachable?("https://example.com/inbox")
-      end
+      assert {:error, :connection_error} = PublisherWorker.perform(job)
+      assert Instances.reachable?("https://example.com/inbox")
+
+      # Final attempt
+      job = %{job | attempt: 5}
+      assert {:error, :connection_error} = PublisherWorker.perform(job)
+      refute Instances.reachable?("https://example.com/inbox")
     end
 
     test "does not mark server as unreachable on successful publish", %{activity: activity} do
-      with_mock Pleroma.Web.Federator,
-        perform: fn :publish_one, _params -> {:ok, %{status: 200}} end do
-        job = %Oban.Job{
-          args: %{
-            "op" => "publish_one",
-            "params" => %{
-              "inbox" => "https://example.com/inbox",
-              "activity_id" => activity.id
-            }
-          },
-          attempt: 1,
-          max_attempts: 5
-        }
+      expect(FederatorMock, :perform, fn :publish_one, _params -> {:ok, %{status: 200}} end)
 
-        assert :ok = Pleroma.Workers.PublisherWorker.perform(job)
-        assert Instances.reachable?("https://example.com/inbox")
-      end
+      job = %Oban.Job{
+        args: %{
+          "op" => "publish_one",
+          "params" => %{
+            "inbox" => "https://example.com/inbox",
+            "activity_id" => activity.id
+          }
+        },
+        attempt: 1,
+        max_attempts: 5
+      }
+
+      assert :ok = PublisherWorker.perform(job)
+      assert Instances.reachable?("https://example.com/inbox")
+    end
+
+    test "does not create atoms from unexpected job params", %{activity: activity} do
+      unknown_key = "unknown_#{System.unique_integer([:positive])}"
+
+      assert_raise ArgumentError, fn -> String.to_existing_atom(unknown_key) end
+
+      expect(FederatorMock, :perform, fn :publish_one, params ->
+        assert params == %{
+                 inbox: "https://example.com/inbox",
+                 activity_id: activity.id
+               }
+
+        {:ok, %{status: 200}}
+      end)
+
+      job = %Oban.Job{
+        args: %{
+          "op" => "publish_one",
+          "params" => %{
+            "inbox" => "https://example.com/inbox",
+            "activity_id" => activity.id,
+            unknown_key => "ignored"
+          }
+        },
+        attempt: 1,
+        max_attempts: 5
+      }
+
+      assert :ok = PublisherWorker.perform(job)
+      assert_raise ArgumentError, fn -> String.to_existing_atom(unknown_key) end
     end
 
     test "cancels job if server is unreachable", %{activity: activity} do
@@ -117,7 +150,7 @@ defmodule Pleroma.Workers.PublisherWorkerTest do
         max_attempts: 5
       }
 
-      assert {:cancel, :unreachable} = Pleroma.Workers.PublisherWorker.perform(job)
+      assert {:cancel, :unreachable} = PublisherWorker.perform(job)
     end
   end
 end