Stream follow updates

benchmarks/load_testing/users.ex

@@ -109,8 +109,8 @@ def make_friends(main_user, max) when is_integer(max) do
   end
 
   def make_friends(%User{} = main_user, %User{} = user) do
-    {:ok, _} = User.follow(main_user, user)
+    {:ok, _, _} = User.follow(main_user, user)
-    {:ok, _} = User.follow(user, main_user)
+    {:ok, _, _} = User.follow(user, main_user)
   end
 
   @spec get_users(User.t(), keyword()) :: [User.t()]

benchmarks/mix/tasks/pleroma/benchmarks/timelines.ex

@@ -50,7 +50,7 @@ def run(_args) do
     )
 
     users
-    |> Enum.each(fn {:ok, follower} -> Pleroma.User.follow(follower, user) end)
+    |> Enum.each(fn {:ok, follower, user} -> Pleroma.User.follow(follower, user) end)
 
     Benchee.run(
       %{

lib/pleroma/following_relationship.ex

@@ -62,23 +62,47 @@ def update(%User{} = follower, %User{} = following, state) do
         follow(follower, following, state)
 
       following_relationship ->
-        following_relationship
+        with {:ok, _following_relationship} <-
-        |> cast(%{state: state}, [:state])
+               following_relationship
-        |> validate_required([:state])
+               |> cast(%{state: state}, [:state])
-        |> Repo.update()
+               |> validate_required([:state])
+               |> Repo.update() do
+          after_update(state, follower, following)
+        end
     end
   end
 
   def follow(%User{} = follower, %User{} = following, state \\ :follow_accept) do
-    %__MODULE__{}
+    with {:ok, _following_relationship} <-
-    |> changeset(%{follower: follower, following: following, state: state})
+           %__MODULE__{}
-    |> Repo.insert(on_conflict: :nothing)
+           |> changeset(%{follower: follower, following: following, state: state})
+           |> Repo.insert(on_conflict: :nothing) do
+      after_update(state, follower, following)
+    end
   end
 
   def unfollow(%User{} = follower, %User{} = following) do
     case get(follower, following) do
-      %__MODULE__{} = following_relationship -> Repo.delete(following_relationship)
+      %__MODULE__{} = following_relationship ->
-      _ -> {:ok, nil}
+        with {:ok, _following_relationship} <- Repo.delete(following_relationship) do
+          after_update(:unfollow, follower, following)
+        end
+
+      _ ->
+        {:ok, nil}
+    end
+  end
+
+  defp after_update(state, %User{} = follower, %User{} = following) do
+    with {:ok, following} <- User.update_follower_count(following),
+         {:ok, follower} <- User.update_following_count(follower) do
+      Pleroma.Web.Streamer.stream("relationships:update", %{
+        state: state,
+        following: following,
+        follower: follower
+      })
+
+      {:ok, follower, following}
     end
   end
 

lib/pleroma/user.ex

@@ -882,7 +882,7 @@ def maybe_direct_follow(%User{} = follower, %User{} = followed) do
     if not ap_enabled?(followed) do
       follow(follower, followed)
     else
-      {:ok, follower}
+      {:ok, follower, followed}
     end
   end
 
@@ -908,11 +908,6 @@ def follow(%User{} = follower, %User{} = followed, state \\ :follow_accept) do
 
       true ->
         FollowingRelationship.follow(follower, followed, state)
-
-        {:ok, _} = update_follower_count(followed)
-
-        follower
-        |> update_following_count()
     end
   end
 
@@ -936,11 +931,6 @@ defp do_unfollow(%User{} = follower, %User{} = followed) do
     case get_follow_state(follower, followed) do
       state when state in [:follow_pending, :follow_accept] ->
         FollowingRelationship.unfollow(follower, followed)
-        {:ok, followed} = update_follower_count(followed)
-
-        {:ok, follower} = update_following_count(follower)
-
-        {:ok, follower, followed}
 
       nil ->
         {:error, "Not subscribed!"}

lib/pleroma/user/import.ex

@@ -45,7 +45,7 @@ def perform(:follow_import, %User{} = follower, [_ | _] = identifiers) do
       identifiers,
       fn identifier ->
         with {:ok, %User{} = followed} <- User.get_or_fetch(identifier),
-             {:ok, follower} <- User.maybe_direct_follow(follower, followed),
+             {:ok, follower, followed} <- User.maybe_direct_follow(follower, followed),
              {:ok, _, _, _} <- CommonAPI.follow(follower, followed) do
           followed
         else

lib/pleroma/web/activity_pub/side_effects.ex

@@ -47,10 +47,9 @@ def handle(
          %User{} = followed <- User.get_cached_by_ap_id(actor),
          %User{} = follower <- User.get_cached_by_ap_id(follower_id),
          {:ok, follow_activity} <- Utils.update_follow_state_for_all(follow_activity, "accept"),
-         {:ok, _relationship} <- FollowingRelationship.update(follower, followed, :follow_accept) do
+         {:ok, _follower, followed} <-
+           FollowingRelationship.update(follower, followed, :follow_accept) do
       Notification.update_notification_type(followed, follow_activity)
-      User.update_follower_count(followed)
-      User.update_following_count(follower)
     end
 
     {:ok, object, meta}
@@ -99,7 +98,7 @@ def handle(
       ) do
     with %User{} = follower <- User.get_cached_by_ap_id(following_user),
          %User{} = followed <- User.get_cached_by_ap_id(followed_user),
-         {_, {:ok, _}, _, _} <-
+         {_, {:ok, _, _}, _, _} <-
            {:following, User.follow(follower, followed, :follow_pending), follower, followed} do
       if followed.local && !followed.is_locked do
         {:ok, accept_data, _} = Builder.accept(followed, object)

lib/pleroma/web/streamer.ex

@@ -36,9 +36,8 @@ def registry, do: @registry
         ) ::
           {:ok, topic :: String.t()} | {:error, :bad_topic} | {:error, :unauthorized}
   def get_topic_and_add_socket(stream, user, oauth_token, params \\ %{}) do
-    case get_topic(stream, user, oauth_token, params) do
+    with {:ok, topic} <- get_topic(stream, user, oauth_token, params) do
-      {:ok, topic} -> add_socket(topic, user)
+      add_socket(topic, user)
-      error -> error
     end
   end
 
@@ -70,10 +69,10 @@ def get_topic("public:remote:media", _user, _oauth_token, %{"instance" => instan
   def get_topic(
         stream,
         %User{id: user_id} = user,
-        %Token{user_id: token_user_id} = oauth_token,
+        %Token{user_id: user_id} = oauth_token,
         _params
       )
-      when stream in @user_streams and user_id == token_user_id do
+      when stream in @user_streams do
     # Note: "read" works for all user streams (not mentioning it since it's an ancestor scope)
     required_scopes =
       if stream == "user:notification" do
@@ -97,10 +96,9 @@ def get_topic(stream, _user, _oauth_token, _params) when stream in @user_streams
   def get_topic(
         "list",
         %User{id: user_id} = user,
-        %Token{user_id: token_user_id} = oauth_token,
+        %Token{user_id: user_id} = oauth_token,
         %{"list" => id}
-      )
+      ) do
-      when user_id == token_user_id do
     cond do
       OAuthScopesPlug.filter_descendants(["read", "read:lists"], oauth_token.scopes) == [] ->
         {:error, :unauthorized}
@@ -137,16 +135,10 @@ def remove_socket(topic) do
 
   def stream(topics, items) do
     if should_env_send?() do
-      List.wrap(topics)
+      for topic <- List.wrap(topics), item <- List.wrap(items) do
-      |> Enum.each(fn topic ->
+        spawn(fn -> do_stream(topic, item) end)
-        List.wrap(items)
+      end
-        |> Enum.each(fn item ->
-          spawn(fn -> do_stream(topic, item) end)
-        end)
-      end)
     end
-
-    :ok
   end
 
   def filtered_by_user?(user, item, streamed_type \\ :activity)
@@ -160,8 +152,7 @@ def filtered_by_user?(%User{} = user, %Activity{} = item, streamed_type) do
     domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.domain_blocks)
 
     with parent <- Object.normalize(item) || item,
-         true <-
+         true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(item.actor not in &1)),
-           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,
          true <-
            !(streamed_type == :activity && item.data["type"] == "Announce" &&
@@ -195,6 +186,22 @@ defp do_stream("direct", item) do
     end)
   end
 
+  defp do_stream("relationships:update", item) do
+    text = StreamerView.render("relationships_update.json", item)
+
+    [item.follower, item.following]
+    |> Enum.map(fn %{id: id} -> "user:#{id}" end)
+    |> Enum.each(fn user_topic ->
+      Logger.debug("Trying to push relationships:update to #{user_topic}\n\n")
+
+      Registry.dispatch(@registry, user_topic, fn list ->
+        Enum.each(list, fn {pid, _auth} ->
+          send(pid, {:text, text})
+        end)
+      end)
+    end)
+  end
+
   defp do_stream("participation", participation) do
     user_topic = "direct:#{participation.user_id}"
     Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")

lib/pleroma/web/views/streamer_view.ex

@@ -74,6 +74,28 @@ def render("chat_update.json", %{chat_message_reference: cm_ref}) do
     |> Jason.encode!()
   end
 
+  def render("relationships_update.json", item) do
+    %{
+      event: "pleroma:relationships_update",
+      payload:
+        %{
+          state: item.state,
+          follower: %{
+            id: item.follower.id,
+            follower_count: item.follower.follower_count,
+            following_count: item.follower.following_count
+          },
+          following: %{
+            id: item.following.id,
+            follower_count: item.following.follower_count,
+            following_count: item.following.following_count
+          }
+        }
+        |> Jason.encode!()
+    }
+    |> Jason.encode!()
+  end
+
   def render("conversation.json", %Participation{} = participation) do
     %{
       event: "conversation",

test/mix/tasks/pleroma/database_test.exs

@@ -73,7 +73,7 @@ test "it prunes old objects from the database" do
   describe "running update_users_following_followers_counts" do
     test "following and followers count are updated" do
       [user, user2] = insert_pair(:user)
-      {:ok, %User{} = user} = User.follow(user, user2)
+      {:ok, %User{} = user, _user2} = User.follow(user, user2)
 
       following = User.following(user)
 

test/mix/tasks/pleroma/user_test.exs

@@ -503,7 +503,7 @@ test "it returns users matching" do
       moot = insert(:user, nickname: "moot")
       kawen = insert(:user, nickname: "kawen", name: "fediverse expert moon")
 
-      {:ok, user} = User.follow(user, moon)
+      {:ok, user, moon} = User.follow(user, moon)
 
       assert [moon.id, kawen.id] == User.Search.search("moon") |> Enum.map(& &1.id)
 

test/pleroma/bbs/handler_test.exs

@@ -19,7 +19,7 @@ test "getting the home timeline" do
     user = insert(:user)
     followed = insert(:user)
 
-    {:ok, user} = User.follow(user, followed)
+    {:ok, user, followed} = User.follow(user, followed)
 
     {:ok, _first} = CommonAPI.post(user, %{status: "hey"})
     {:ok, _second} = CommonAPI.post(followed, %{status: "hello"})

test/pleroma/notification_test.exs

@@ -779,7 +779,7 @@ test "it returns following domain-blocking recipient in enabled recipients list"
       other_user = insert(:user)
 
       {:ok, other_user} = User.block_domain(other_user, blocked_domain)
-      {:ok, other_user} = User.follow(other_user, user)
+      {:ok, other_user, user} = User.follow(other_user, user)
 
       {:ok, activity} = CommonAPI.post(user, %{status: "hey @#{other_user.nickname}!"})
 
@@ -1070,7 +1070,7 @@ test "it returns notifications for domain-blocked but followed user" do
       blocked = insert(:user, ap_id: "http://some-domain.com")
 
       {:ok, user} = User.block_domain(user, "some-domain.com")
-      {:ok, _} = User.follow(user, blocked)
+      {:ok, _, _} = User.follow(user, blocked)
 
       {:ok, _activity} = CommonAPI.post(blocked, %{status: "hey @#{user.nickname}"})
 

test/pleroma/user/import_test.exs

@@ -30,7 +30,7 @@ test "it imports user followings from list" do
 
       assert {:ok, result} = ObanHelpers.perform(job)
       assert is_list(result)
-      assert result == [user2, user3]
+      assert result == [refresh_record(user2), refresh_record(user3)]
       assert User.following?(user1, user2)
       assert User.following?(user1, user3)
     end

test/pleroma/user_search_test.exs

@@ -151,8 +151,8 @@ test "finds users, boosting ranks of friends and followers" do
       follower = insert(:user, %{name: "Doe"})
       friend = insert(:user, %{name: "Doe"})
 
-      {:ok, follower} = User.follow(follower, u1)
+      {:ok, follower, u1} = User.follow(follower, u1)
-      {:ok, u1} = User.follow(u1, friend)
+      {:ok, u1, friend} = User.follow(u1, friend)
 
       assert [friend.id, follower.id, u2.id] --
                Enum.map(User.search("doe", resolve: false, for_user: u1), & &1.id) == []
@@ -165,9 +165,9 @@ test "finds followings of user by partial name" do
       following_jimi = insert(:user, %{name: "Lizz Wright"})
       follower_lizz = insert(:user, %{name: "Jimi"})
 
-      {:ok, lizz} = User.follow(lizz, following_lizz)
+      {:ok, lizz, following_lizz} = User.follow(lizz, following_lizz)
-      {:ok, _jimi} = User.follow(jimi, following_jimi)
+      {:ok, _jimi, _following_jimi} = User.follow(jimi, following_jimi)
-      {:ok, _follower_lizz} = User.follow(follower_lizz, lizz)
+      {:ok, _follower_lizz, _lizz} = User.follow(follower_lizz, lizz)
 
       assert Enum.map(User.search("jimi", following: true, for_user: lizz), & &1.id) == [
                following_lizz.id

test/pleroma/user_test.exs

@@ -233,7 +233,7 @@ test "follow_all follows mutliple users" do
     {:ok, _user_relationship} = User.block(user, blocked)
     {:ok, _user_relationship} = User.block(reverse_blocked, user)
 
-    {:ok, user} = User.follow(user, followed_zero)
+    {:ok, user, followed_zero} = User.follow(user, followed_zero)
 
     {:ok, user} = User.follow_all(user, [followed_one, followed_two, blocked, reverse_blocked])
 
@@ -262,7 +262,7 @@ test "follow takes a user and another user" do
     user = insert(:user)
     followed = insert(:user)
 
-    {:ok, user} = User.follow(user, followed)
+    {:ok, user, followed} = User.follow(user, followed)
 
     user = User.get_cached_by_id(user.id)
     followed = User.get_cached_by_ap_id(followed.ap_id)
@@ -302,7 +302,7 @@ test "local users do not automatically follow local locked accounts" do
     follower = insert(:user, is_locked: true)
     followed = insert(:user, is_locked: true)
 
-    {:ok, follower} = User.maybe_direct_follow(follower, followed)
+    {:ok, follower, followed} = User.maybe_direct_follow(follower, followed)
 
     refute User.following?(follower, followed)
   end
@@ -330,7 +330,7 @@ test "unfollow with syncronizes external user" do
           following_address: "http://localhost:4001/users/fuser2/following"
         })
 
-      {:ok, user} = User.follow(user, followed, :follow_accept)
+      {:ok, user, followed} = User.follow(user, followed, :follow_accept)
 
       {:ok, user, _activity} = User.unfollow(user, followed)
 
@@ -343,7 +343,7 @@ test "unfollow takes a user and another user" do
       followed = insert(:user)
       user = insert(:user)
 
-      {:ok, user} = User.follow(user, followed, :follow_accept)
+      {:ok, user, followed} = User.follow(user, followed, :follow_accept)
 
       assert User.following(user) == [user.follower_address, followed.follower_address]
 
@@ -904,8 +904,8 @@ test "gets all followers for a given user" do
       follower_two = insert(:user)
       not_follower = insert(:user)
 
-      {:ok, follower_one} = User.follow(follower_one, user)
+      {:ok, follower_one, user} = User.follow(follower_one, user)
-      {:ok, follower_two} = User.follow(follower_two, user)
+      {:ok, follower_two, user} = User.follow(follower_two, user)
 
       res = User.get_followers(user)
 
@@ -920,8 +920,8 @@ test "gets all friends (followed users) for a given user" do
       followed_two = insert(:user)
       not_followed = insert(:user)
 
-      {:ok, user} = User.follow(user, followed_one)
+      {:ok, user, followed_one} = User.follow(user, followed_one)
-      {:ok, user} = User.follow(user, followed_two)
+      {:ok, user, followed_two} = User.follow(user, followed_two)
 
       res = User.get_friends(user)
 
@@ -1091,8 +1091,8 @@ test "blocks tear down cyclical follow relationships" do
       blocker = insert(:user)
       blocked = insert(:user)
 
-      {:ok, blocker} = User.follow(blocker, blocked)
+      {:ok, blocker, blocked} = User.follow(blocker, blocked)
-      {:ok, blocked} = User.follow(blocked, blocker)
+      {:ok, blocked, blocker} = User.follow(blocked, blocker)
 
       assert User.following?(blocker, blocked)
       assert User.following?(blocked, blocker)
@@ -1110,7 +1110,7 @@ test "blocks tear down blocker->blocked follow relationships" do
       blocker = insert(:user)
       blocked = insert(:user)
 
-      {:ok, blocker} = User.follow(blocker, blocked)
+      {:ok, blocker, blocked} = User.follow(blocker, blocked)
 
       assert User.following?(blocker, blocked)
       refute User.following?(blocked, blocker)
@@ -1128,7 +1128,7 @@ test "blocks tear down blocked->blocker follow relationships" do
       blocker = insert(:user)
       blocked = insert(:user)
 
-      {:ok, blocked} = User.follow(blocked, blocker)
+      {:ok, blocked, blocker} = User.follow(blocked, blocker)
 
       refute User.following?(blocker, blocked)
       assert User.following?(blocked, blocker)
@@ -1226,7 +1226,7 @@ test "follows take precedence over domain blocks" do
       good_eggo = insert(:user, %{ap_id: "https://meanies.social/user/cuteposter"})
 
       {:ok, user} = User.block_domain(user, "meanies.social")
-      {:ok, user} = User.follow(user, good_eggo)
+      {:ok, user, good_eggo} = User.follow(user, good_eggo)
 
       refute User.blocks?(user, good_eggo)
     end
@@ -1260,8 +1260,8 @@ test "get recipients" do
       assert Enum.map([actor, addressed], & &1.ap_id) --
                Enum.map(User.get_recipients_from_activity(activity), & &1.ap_id) == []
 
-      {:ok, user} = User.follow(user, actor)
+      {:ok, user, actor} = User.follow(user, actor)
-      {:ok, _user_two} = User.follow(user_two, actor)
+      {:ok, _user_two, _actor} = User.follow(user_two, actor)
       recipients = User.get_recipients_from_activity(activity)
       assert length(recipients) == 3
       assert user in recipients
@@ -1282,8 +1282,8 @@ test "has following" do
       assert Enum.map([actor, addressed], & &1.ap_id) --
                Enum.map(User.get_recipients_from_activity(activity), & &1.ap_id) == []
 
-      {:ok, _actor} = User.follow(actor, user)
+      {:ok, _actor, _user} = User.follow(actor, user)
-      {:ok, _actor} = User.follow(actor, user_two)
+      {:ok, _actor, _user_two} = User.follow(actor, user_two)
       recipients = User.get_recipients_from_activity(activity)
       assert length(recipients) == 2
       assert addressed in recipients
@@ -1304,7 +1304,7 @@ test "hide a user from followers" do
       user = insert(:user)
       user2 = insert(:user)
 
-      {:ok, user} = User.follow(user, user2)
+      {:ok, user, user2} = User.follow(user, user2)
       {:ok, _user} = User.deactivate(user)
 
       user2 = User.get_cached_by_id(user2.id)
@@ -1317,7 +1317,7 @@ test "hide a user from friends" do
       user = insert(:user)
       user2 = insert(:user)
 
-      {:ok, user2} = User.follow(user2, user)
+      {:ok, user2, user} = User.follow(user2, user)
       assert user2.following_count == 1
       assert User.following_count(user2) == 1
 
@@ -1335,7 +1335,7 @@ test "hide a user's statuses from timelines and notifications" do
       user = insert(:user)
       user2 = insert(:user)
 
-      {:ok, user2} = User.follow(user2, user)
+      {:ok, user2, user} = User.follow(user2, user)
 
       {:ok, activity} = CommonAPI.post(user, %{status: "hey @#{user2.nickname}"})
 
@@ -1408,10 +1408,10 @@ test ".delete_user_activities deletes all create activities", %{user: user} do
 
     test "it deactivates a user, all follow relationships and all activities", %{user: user} do
       follower = insert(:user)
-      {:ok, follower} = User.follow(follower, user)
+      {:ok, follower, user} = User.follow(follower, user)
 
       locked_user = insert(:user, name: "locked", is_locked: true)
-      {:ok, _} = User.follow(user, locked_user, :follow_pending)
+      {:ok, _, _} = User.follow(user, locked_user, :follow_pending)
 
       object = insert(:note, user: user)
       activity = insert(:note_activity, user: user, note: object)
@@ -1769,9 +1769,9 @@ test "follower count is updated when a follower is blocked" do
     follower2 = insert(:user)
     follower3 = insert(:user)
 
-    {:ok, follower} = User.follow(follower, user)
+    {:ok, follower, user} = User.follow(follower, user)
-    {:ok, _follower2} = User.follow(follower2, user)
+    {:ok, _follower2, _user} = User.follow(follower2, user)
-    {:ok, _follower3} = User.follow(follower3, user)
+    {:ok, _follower3, _user} = User.follow(follower3, user)
 
     {:ok, _user_relationship} = User.block(user, follower)
     user = refresh_record(user)
@@ -2012,8 +2012,7 @@ test "updates the counters normally on following/getting a follow when disabled"
       assert other_user.following_count == 0
       assert other_user.follower_count == 0
 
-      {:ok, user} = Pleroma.User.follow(user, other_user)
+      {:ok, user, other_user} = Pleroma.User.follow(user, other_user)
-      other_user = Pleroma.User.get_by_id(other_user.id)
 
       assert user.following_count == 1
       assert other_user.follower_count == 1
@@ -2036,8 +2035,7 @@ test "syncronizes the counters with the remote instance for the followed when en
       assert other_user.follower_count == 0
 
       Pleroma.Config.put([:instance, :external_user_synchronization], true)
-      {:ok, _user} = User.follow(user, other_user)
+      {:ok, _user, other_user} = User.follow(user, other_user)
-      other_user = User.get_by_id(other_user.id)
 
       assert other_user.follower_count == 437
     end
@@ -2059,7 +2057,7 @@ test "syncronizes the counters with the remote instance for the follower when en
       assert other_user.follower_count == 0
 
       Pleroma.Config.put([:instance, :external_user_synchronization], true)
-      {:ok, other_user} = User.follow(other_user, user)
+      {:ok, other_user, _user} = User.follow(other_user, user)
 
       assert other_user.following_count == 152
     end

test/pleroma/web/activity_pub/activity_pub_controller_test.exs

@@ -675,7 +675,7 @@ test "it accepts messages from actors that are followed by the user", %{
       recipient = insert(:user)
       actor = insert(:user, %{ap_id: "http://mastodon.example.org/users/actor"})
 
-      {:ok, recipient} = User.follow(recipient, actor)
+      {:ok, recipient, actor} = User.follow(recipient, actor)
 
       object =
         data["object"]

test/pleroma/web/activity_pub/activity_pub_test.exs

@@ -726,7 +726,7 @@ test "does return activities from followed users on blocked domains" do
     domain_user = insert(:user, %{ap_id: "https://#{domain}/@pundit"})
     blocker = insert(:user)
 
-    {:ok, blocker} = User.follow(blocker, domain_user)
+    {:ok, blocker, domain_user} = User.follow(blocker, domain_user)
     {:ok, blocker} = User.block_domain(blocker, domain)
 
     assert User.following?(blocker, domain_user)
@@ -853,7 +853,7 @@ test "does include announces on request" do
     user = insert(:user)
     booster = insert(:user)
 
-    {:ok, user} = User.follow(user, booster)
+    {:ok, user, booster} = User.follow(user, booster)
 
     {:ok, announce} = CommonAPI.repeat(activity_three.id, booster)
 
@@ -1158,13 +1158,13 @@ test "it filters broken threads" do
       user2 = insert(:user)
       user3 = insert(:user)
 
-      {:ok, user1} = User.follow(user1, user3)
+      {:ok, user1, user3} = User.follow(user1, user3)
       assert User.following?(user1, user3)
 
-      {:ok, user2} = User.follow(user2, user3)
+      {:ok, user2, user3} = User.follow(user2, user3)
       assert User.following?(user2, user3)
 
-      {:ok, user3} = User.follow(user3, user2)
+      {:ok, user3, user2} = User.follow(user3, user2)
       assert User.following?(user3, user2)
 
       {:ok, public_activity} = CommonAPI.post(user3, %{status: "hi 1"})
@@ -1931,13 +1931,13 @@ test "home timeline with default reply_visibility `self`", %{
 
   defp public_messages(_) do
     [u1, u2, u3, u4] = insert_list(4, :user)
-    {:ok, u1} = User.follow(u1, u2)
+    {:ok, u1, u2} = User.follow(u1, u2)
-    {:ok, u2} = User.follow(u2, u1)
+    {:ok, u2, u1} = User.follow(u2, u1)
-    {:ok, u1} = User.follow(u1, u4)
+    {:ok, u1, u4} = User.follow(u1, u4)
-    {:ok, u4} = User.follow(u4, u1)
+    {:ok, u4, u1} = User.follow(u4, u1)
 
-    {:ok, u2} = User.follow(u2, u3)
+    {:ok, u2, u3} = User.follow(u2, u3)
-    {:ok, u3} = User.follow(u3, u2)
+    {:ok, u3, u2} = User.follow(u3, u2)
 
     {:ok, a1} = CommonAPI.post(u1, %{status: "Status"})
 
@@ -2030,15 +2030,15 @@ defp public_messages(_) do
 
   defp private_messages(_) do
     [u1, u2, u3, u4] = insert_list(4, :user)
-    {:ok, u1} = User.follow(u1, u2)
+    {:ok, u1, u2} = User.follow(u1, u2)
-    {:ok, u2} = User.follow(u2, u1)
+    {:ok, u2, u1} = User.follow(u2, u1)
-    {:ok, u1} = User.follow(u1, u3)
+    {:ok, u1, u3} = User.follow(u1, u3)
-    {:ok, u3} = User.follow(u3, u1)
+    {:ok, u3, u1} = User.follow(u3, u1)
-    {:ok, u1} = User.follow(u1, u4)
+    {:ok, u1, u4} = User.follow(u1, u4)
-    {:ok, u4} = User.follow(u4, u1)
+    {:ok, u4, u1} = User.follow(u4, u1)
 
-    {:ok, u2} = User.follow(u2, u3)
+    {:ok, u2, u3} = User.follow(u2, u3)
-    {:ok, u3} = User.follow(u3, u2)
+    {:ok, u3, u2} = User.follow(u3, u2)
 
     {:ok, a1} = CommonAPI.post(u1, %{status: "Status", visibility: "private"})
 

test/pleroma/web/activity_pub/publisher_test.exs

@@ -281,8 +281,7 @@ test "publish to url with with different ports" do
       actor = insert(:user, follower_address: follower.ap_id)
       user = insert(:user)
 
-      {:ok, _follower_one} = Pleroma.User.follow(follower, actor)
+      {:ok, follower, actor} = Pleroma.User.follow(follower, actor)
-      actor = refresh_record(actor)
 
       note_activity =
         insert(:note_activity,

test/pleroma/web/activity_pub/transmogrifier/accept_handling_test.exs

@@ -15,7 +15,7 @@ test "it works for incoming accepts which were pre-accepted" do
     follower = insert(:user)
     followed = insert(:user)
 
-    {:ok, follower} = User.follow(follower, followed)
+    {:ok, follower, followed} = User.follow(follower, followed)
     assert User.following?(follower, followed) == true
 
     {:ok, _, _, follow_activity} = CommonAPI.follow(follower, followed)

test/pleroma/web/activity_pub/transmogrifier/block_handling_test.exs

@@ -40,8 +40,8 @@ test "incoming blocks successfully tear down any follow relationship" do
       |> Map.put("object", blocked.ap_id)
       |> Map.put("actor", blocker.ap_id)
 
-    {:ok, blocker} = User.follow(blocker, blocked)
+    {:ok, blocker, blocked} = User.follow(blocker, blocked)
-    {:ok, blocked} = User.follow(blocked, blocker)
+    {:ok, blocked, blocker} = User.follow(blocked, blocker)
 
     assert User.following?(blocker, blocked)
     assert User.following?(blocked, blocker)

test/pleroma/web/activity_pub/transmogrifier/reject_handling_test.exs

@@ -35,7 +35,7 @@ test "it works for incoming rejects which are referenced by IRI only" do
     follower = insert(:user)
     followed = insert(:user, is_locked: true)
 
-    {:ok, follower} = User.follow(follower, followed)
+    {:ok, follower, followed} = User.follow(follower, followed)
     {:ok, _, _, follow_activity} = CommonAPI.follow(follower, followed)
 
     assert User.following?(follower, followed) == true

test/pleroma/web/activity_pub/visibility_test.exs

@@ -15,7 +15,7 @@ defmodule Pleroma.Web.ActivityPub.VisibilityTest do
     mentioned = insert(:user)
     following = insert(:user)
     unrelated = insert(:user)
-    {:ok, following} = Pleroma.User.follow(following, user)
+    {:ok, following, user} = Pleroma.User.follow(following, user)
     {:ok, list} = Pleroma.List.create("foo", user)
 
     Pleroma.List.follow(list, unrelated)

test/pleroma/web/mastodon_api/controllers/account_controller_test.exs

@@ -320,7 +320,7 @@ test "gets users statuses", %{conn: conn} do
       user_two = insert(:user)
       user_three = insert(:user)
 
-      {:ok, _user_three} = User.follow(user_three, user_one)
+      {:ok, _user_three, _user_one} = User.follow(user_three, user_one)
 
       {:ok, activity} = CommonAPI.post(user_one, %{status: "HI!!!"})
 
@@ -568,7 +568,7 @@ test "if user is authenticated", %{local: local, remote: remote} do
 
     test "getting followers", %{user: user, conn: conn} do
       other_user = insert(:user)
-      {:ok, %{id: user_id}} = User.follow(user, other_user)
+      {:ok, %{id: user_id}, other_user} = User.follow(user, other_user)
 
       conn = get(conn, "/api/v1/accounts/#{other_user.id}/followers")
 
@@ -577,7 +577,7 @@ test "getting followers", %{user: user, conn: conn} do
 
     test "getting followers, hide_followers", %{user: user, conn: conn} do
       other_user = insert(:user, hide_followers: true)
-      {:ok, _user} = User.follow(user, other_user)
+      {:ok, _user, _other_user} = User.follow(user, other_user)
 
       conn = get(conn, "/api/v1/accounts/#{other_user.id}/followers")
 
@@ -587,7 +587,7 @@ test "getting followers, hide_followers", %{user: user, conn: conn} do
     test "getting followers, hide_followers, same user requesting" do
       user = insert(:user)
       other_user = insert(:user, hide_followers: true)
-      {:ok, _user} = User.follow(user, other_user)
+      {:ok, _user, _other_user} = User.follow(user, other_user)
 
       conn =
         build_conn()
@@ -599,9 +599,9 @@ test "getting followers, hide_followers, same user requesting" do
     end
 
     test "getting followers, pagination", %{user: user, conn: conn} do
-      {:ok, %User{id: follower1_id}} = :user |> insert() |> User.follow(user)
+      {:ok, %User{id: follower1_id}, _user} = :user |> insert() |> User.follow(user)
-      {:ok, %User{id: follower2_id}} = :user |> insert() |> User.follow(user)
+      {:ok, %User{id: follower2_id}, _user} = :user |> insert() |> User.follow(user)
-      {:ok, %User{id: follower3_id}} = :user |> insert() |> User.follow(user)
+      {:ok, %User{id: follower3_id}, _user} = :user |> insert() |> User.follow(user)
 
       assert [%{"id" => ^follower3_id}, %{"id" => ^follower2_id}] =
                conn
@@ -637,7 +637,7 @@ test "getting followers, pagination", %{user: user, conn: conn} do
 
     test "getting following", %{user: user, conn: conn} do
       other_user = insert(:user)
-      {:ok, user} = User.follow(user, other_user)
+      {:ok, user, other_user} = User.follow(user, other_user)
 
       conn = get(conn, "/api/v1/accounts/#{user.id}/following")
 
@@ -648,7 +648,7 @@ test "getting following", %{user: user, conn: conn} do
     test "getting following, hide_follows, other user requesting" do
       user = insert(:user, hide_follows: true)
       other_user = insert(:user)
-      {:ok, user} = User.follow(user, other_user)
+      {:ok, user, other_user} = User.follow(user, other_user)
 
       conn =
         build_conn()
@@ -662,7 +662,7 @@ test "getting following, hide_follows, other user requesting" do
     test "getting following, hide_follows, same user requesting" do
       user = insert(:user, hide_follows: true)
       other_user = insert(:user)
-      {:ok, user} = User.follow(user, other_user)
+      {:ok, user, _other_user} = User.follow(user, other_user)
 
       conn =
         build_conn()
@@ -677,9 +677,9 @@ test "getting following, pagination", %{user: user, conn: conn} do
       following1 = insert(:user)
       following2 = insert(:user)
       following3 = insert(:user)
-      {:ok, _} = User.follow(user, following1)
+      {:ok, _, _} = User.follow(user, following1)
-      {:ok, _} = User.follow(user, following2)
+      {:ok, _, _} = User.follow(user, following2)
-      {:ok, _} = User.follow(user, following3)
+      {:ok, _, _} = User.follow(user, following3)
 
       res_conn = get(conn, "/api/v1/accounts/#{user.id}/following?since_id=#{following1.id}")
 
@@ -1520,7 +1520,7 @@ test "locked accounts" do
 
     test "returns the relationships for the current user", %{user: user, conn: conn} do
       %{id: other_user_id} = other_user = insert(:user)
-      {:ok, _user} = User.follow(user, other_user)
+      {:ok, _user, _other_user} = User.follow(user, other_user)
 
       assert [%{"id" => ^other_user_id}] =
                conn

test/pleroma/web/mastodon_api/controllers/conversation_controller_test.exs

@@ -18,7 +18,7 @@ defmodule Pleroma.Web.MastodonAPI.ConversationControllerTest do
       user_two = insert(:user)
       user_three = insert(:user)
 
-      {:ok, user_two} = User.follow(user_two, user_one)
+      {:ok, user_two, user_one} = User.follow(user_two, user_one)
 
       {:ok, %{user: user_one, user_two: user_two, user_three: user_three, conn: conn}}
     end

test/pleroma/web/mastodon_api/controllers/follow_request_controller_test.exs

@@ -21,7 +21,7 @@ test "/api/v1/follow_requests works", %{user: user, conn: conn} do
       other_user = insert(:user)
 
       {:ok, _, _, _activity} = CommonAPI.follow(other_user, user)
-      {:ok, other_user} = User.follow(other_user, user, :follow_pending)
+      {:ok, other_user, user} = User.follow(other_user, user, :follow_pending)
 
       assert User.following?(other_user, user) == false
 
@@ -35,7 +35,7 @@ test "/api/v1/follow_requests/:id/authorize works", %{user: user, conn: conn} do
       other_user = insert(:user)
 
       {:ok, _, _, _activity} = CommonAPI.follow(other_user, user)
-      {:ok, other_user} = User.follow(other_user, user, :follow_pending)
+      {:ok, other_user, user} = User.follow(other_user, user, :follow_pending)
 
       user = User.get_cached_by_id(user.id)
       other_user = User.get_cached_by_id(other_user.id)

test/pleroma/web/mastodon_api/controllers/timeline_controller_test.exs

@@ -136,7 +136,7 @@ test "the public timeline includes only public statuses for an authenticated use
     test "doesn't return replies if follower is posting with blocked user" do
       %{conn: conn, user: blocker} = oauth_access(["read:statuses"])
       [blockee, friend] = insert_list(2, :user)
-      {:ok, blocker} = User.follow(blocker, friend)
+      {:ok, blocker, friend} = User.follow(blocker, friend)
       {:ok, _} = User.block(blocker, blockee)
 
       conn = assign(conn, :user, blocker)
@@ -165,7 +165,7 @@ test "doesn't return replies if follow is posting with users from blocked domain
       %{conn: conn, user: blocker} = oauth_access(["read:statuses"])
       friend = insert(:user)
       blockee = insert(:user, ap_id: "https://example.com/users/blocked")
-      {:ok, blocker} = User.follow(blocker, friend)
+      {:ok, blocker, friend} = User.follow(blocker, friend)
       {:ok, blocker} = User.block_domain(blocker, "example.com")
 
       conn = assign(conn, :user, blocker)
@@ -336,7 +336,7 @@ test "direct timeline", %{conn: conn} do
       user_one = insert(:user)
       user_two = insert(:user)
 
-      {:ok, user_two} = User.follow(user_two, user_one)
+      {:ok, user_two, user_one} = User.follow(user_two, user_one)
 
       {:ok, direct} =
         CommonAPI.post(user_one, %{

test/pleroma/web/mastodon_api/mastodon_api_test.exs

@@ -30,7 +30,7 @@ test "following for user" do
     test "returns ok if user already followed" do
       follower = insert(:user)
       user = insert(:user)
-      {:ok, follower} = User.follow(follower, user)
+      {:ok, follower, user} = User.follow(follower, user)
       {:ok, follower} = MastodonAPI.follow(follower, refresh_record(user))
       assert User.following?(follower, user)
     end
@@ -41,8 +41,8 @@ test "returns user followers" do
       follower1_user = insert(:user)
       follower2_user = insert(:user)
       user = insert(:user)
-      {:ok, _follower1_user} = User.follow(follower1_user, user)
+      {:ok, _follower1_user, _user} = User.follow(follower1_user, user)
-      {:ok, follower2_user} = User.follow(follower2_user, user)
+      {:ok, follower2_user, _user} = User.follow(follower2_user, user)
 
       assert MastodonAPI.get_followers(user, %{"limit" => 1}) == [follower2_user]
     end
@@ -55,9 +55,9 @@ test "returns user friends" do
       followed_two = insert(:user)
       followed_three = insert(:user)
 
-      {:ok, user} = User.follow(user, followed_one)
+      {:ok, user, followed_one} = User.follow(user, followed_one)
-      {:ok, user} = User.follow(user, followed_two)
+      {:ok, user, followed_two} = User.follow(user, followed_two)
-      {:ok, user} = User.follow(user, followed_three)
+      {:ok, user, followed_three} = User.follow(user, followed_three)
       res = MastodonAPI.get_friends(user)
 
       assert length(res) == 3

test/pleroma/web/mastodon_api/views/account_view_test.exs

@@ -274,8 +274,8 @@ test "represent a relationship for the following and followed user" do
       user = insert(:user)
       other_user = insert(:user)
 
-      {:ok, user} = User.follow(user, other_user)
+      {:ok, user, other_user} = User.follow(user, other_user)
-      {:ok, other_user} = User.follow(other_user, user)
+      {:ok, other_user, user} = User.follow(other_user, user)
       {:ok, _subscription} = User.subscribe(user, other_user)
       {:ok, _user_relationships} = User.mute(user, other_user, %{notifications: true})
       {:ok, _reblog_mute} = CommonAPI.hide_reblogs(user, other_user)
@@ -301,7 +301,7 @@ test "represent a relationship for the blocking and blocked user" do
       user = insert(:user)
       other_user = insert(:user)
 
-      {:ok, user} = User.follow(user, other_user)
+      {:ok, user, other_user} = User.follow(user, other_user)
       {:ok, _subscription} = User.subscribe(user, other_user)
       {:ok, _user_relationship} = User.block(user, other_user)
       {:ok, _user_relationship} = User.block(other_user, user)

test/pleroma/web/pleroma_api/controllers/user_import_controller_test.exs

@@ -47,7 +47,8 @@ test "it imports follow lists from file", %{conn: conn} do
                  |> json_response_and_validate_schema(200)
 
         assert [{:ok, job_result}] = ObanHelpers.perform_all()
-        assert job_result == [user2]
+        assert job_result == [refresh_record(user2)]
+        assert [%Pleroma.User{follower_count: 1}] = job_result
       end
     end
 
@@ -108,7 +109,7 @@ test "it imports follows with different nickname variations", %{conn: conn} do
                |> json_response_and_validate_schema(200)
 
       assert [{:ok, job_result}] = ObanHelpers.perform_all()
-      assert job_result == users
+      assert job_result == Enum.map(users, &refresh_record/1)
     end
   end
 

test/pleroma/web/streamer_test.exs

@@ -403,6 +403,73 @@ test "it sends follow activities to the 'user:notification' stream", %{
       assert notif.activity.id == follow_activity.id
       refute Streamer.filtered_by_user?(user, notif)
     end
+
+    test "it sends relationships updates to the 'user' stream", %{
+      user: user,
+      token: oauth_token
+    } do
+      user_id = user.id
+      user_url = user.ap_id
+      follower = insert(:user)
+      follower_token = insert(:oauth_token, user: follower)
+      follower_id = follower.id
+
+      body =
+        File.read!("test/fixtures/users_mock/localhost.json")
+        |> String.replace("{{nickname}}", user.nickname)
+        |> Jason.encode!()
+
+      Tesla.Mock.mock_global(fn
+        %{method: :get, url: ^user_url} ->
+          %Tesla.Env{status: 200, body: body}
+      end)
+
+      Streamer.get_topic_and_add_socket("user", user, oauth_token)
+      Streamer.get_topic_and_add_socket("user", follower, follower_token)
+      {:ok, _follower, _followed, _follow_activity} = CommonAPI.follow(follower, user)
+
+      # follow_pending event sent to both follower and following
+      assert_receive {:text, event}
+      assert_receive {:text, ^event}
+
+      assert %{"event" => "pleroma:relationships_update", "payload" => payload} =
+               Jason.decode!(event)
+
+      assert %{
+               "follower" => %{
+                 "follower_count" => 0,
+                 "following_count" => 0,
+                 "id" => ^follower_id
+               },
+               "following" => %{
+                 "follower_count" => 0,
+                 "following_count" => 0,
+                 "id" => ^user_id
+               },
+               "state" => "follow_pending"
+             } = Jason.decode!(payload)
+
+      # follow_accept event sent to both follower and following
+      assert_receive {:text, event}
+      assert_receive {:text, ^event}
+
+      assert %{"event" => "pleroma:relationships_update", "payload" => payload} =
+               Jason.decode!(event)
+
+      assert %{
+               "follower" => %{
+                 "follower_count" => 0,
+                 "following_count" => 1,
+                 "id" => ^follower_id
+               },
+               "following" => %{
+                 "follower_count" => 1,
+                 "following_count" => 0,
+                 "id" => ^user_id
+               },
+               "state" => "follow_accept"
+             } = Jason.decode!(payload)
+    end
   end
 
   describe "public streams" do
@@ -563,7 +630,7 @@ test "it doesn't send unwanted DMs to list", %{user: user_a, token: user_a_token
       user_b = insert(:user)
       user_c = insert(:user)
 
-      {:ok, user_a} = User.follow(user_a, user_b)
+      {:ok, user_a, user_b} = User.follow(user_a, user_b)
 
       {:ok, list} = List.create("Test", user_a)
       {:ok, list} = List.follow(list, user_b)
@@ -599,7 +666,7 @@ test "it doesn't send unwanted private posts to list", %{user: user_a, token: us
     test "it sends wanted private posts to list", %{user: user_a, token: user_a_token} do
       user_b = insert(:user)
 
-      {:ok, user_a} = User.follow(user_a, user_b)
+      {:ok, user_a, user_b} = User.follow(user_a, user_b)
 
       {:ok, list} = List.create("Test", user_a)
       {:ok, list} = List.follow(list, user_b)