Merge branch 'csaurus/pleroma-feature/mstdn-direct-api' into develop

6 files changed, 76 insertions, 8 deletions

diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
index 1a1bfbffd..4e0be5ba2 100644
--- a/lib/pleroma/web/activity_pub/activity_pub.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
@@ -53,15 +53,24 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   end
 
   def stream_out(activity) do
+    public = "https://www.w3.org/ns/activitystreams#Public"
+
     if activity.data["type"] in ["Create", "Announce"] do
       Pleroma.Web.Streamer.stream("user", activity)
 
-      if Enum.member?(activity.data["to"], "https://www.w3.org/ns/activitystreams#Public") do
+      if Enum.member?(activity.data["to"], public) do
         Pleroma.Web.Streamer.stream("public", activity)
 
         if activity.local do
           Pleroma.Web.Streamer.stream("public:local", activity)
         end
+      else
+        if !Enum.member?(activity.data["cc"] || [], public) &&
+             !Enum.member?(
+               activity.data["to"],
+               User.get_by_ap_id(activity.data["actor"]).follower_address
+             ),
+           do: Pleroma.Web.Streamer.stream("direct", activity)
       end
     end
   end
@@ -293,6 +302,32 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     |> Enum.reverse()
   end
 
+  @valid_visibilities ~w[direct unlisted public private]
+
+  defp restrict_visibility(query, %{visibility: "direct"}) do
+    public = "https://www.w3.org/ns/activitystreams#Public"
+
+    from(
+      activity in query,
+      join: sender in User,
+      on: sender.ap_id == activity.actor,
+      # Are non-direct statuses with no to/cc possible?
+      where:
+        fragment(
+          "not (? && ?)",
+          [^public, sender.follower_address],
+          activity.recipients
+        )
+    )
+  end
+
+  defp restrict_visibility(_query, %{visibility: visibility})
+       when visibility not in @valid_visibilities do
+    Logger.error("Could not restrict visibility to #{visibility}")
+  end
+
+  defp restrict_visibility(query, _visibility), do: query
+
   def fetch_user_activities(user, reading_user, params \\ %{}) do
     params =
       params
@@ -447,6 +482,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     |> restrict_recent(opts)
     |> restrict_blocked(opts)
     |> restrict_media(opts)
+    |> restrict_visibility(opts)
   end
 
   def fetch_activities(recipients, opts \\ %{}) do
diff --git a/lib/pleroma/web/mastodon_api/mastodon_api_controller.ex b/lib/pleroma/web/mastodon_api/mastodon_api_controller.ex
index e12d3fb5b..8dbbe3871 100644
--- a/lib/pleroma/web/mastodon_api/mastodon_api_controller.ex
+++ b/lib/pleroma/web/mastodon_api/mastodon_api_controller.ex
@@ -220,6 +220,15 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
     end
   end
 
+  def dm_timeline(%{assigns: %{user: user}} = conn, params) do
+    query = ActivityPub.fetch_activities_query([user.ap_id], %{visibility: "direct"})
+    activities = Repo.all(query)
+
+    conn
+    |> add_link_headers(:user_statuses, activities, user.ap_id)
+    |> render(StatusView, "index.json", %{activities: activities, for: user, as: :activity})
+  end
+
   def get_status(%{assigns: %{user: user}} = conn, %{"id" => id}) do
     with %Activity{} = activity <- Repo.get(Activity, id),
          true <- ActivityPub.visible_for_user?(activity, user) do
diff --git a/lib/pleroma/web/mastodon_api/mastodon_socket.ex b/lib/pleroma/web/mastodon_api/mastodon_socket.ex
index f3e062941..080f62b31 100644
--- a/lib/pleroma/web/mastodon_api/mastodon_socket.ex
+++ b/lib/pleroma/web/mastodon_api/mastodon_socket.ex
@@ -15,7 +15,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonSocket do
     with token when not is_nil(token) <- params["access_token"],
          %Token{user_id: user_id} <- Repo.get_by(Token, token: token),
          %User{} = user <- Repo.get(User, user_id),
-         stream when stream in ["public", "public:local", "user"] <- params["stream"] do
+         stream when stream in ["public", "public:local", "user", "direct"] <- params["stream"] do
       socket =
         socket
         |> assign(:topic, params["stream"])
diff --git a/lib/pleroma/web/mastodon_api/views/status_view.ex b/lib/pleroma/web/mastodon_api/views/status_view.ex
index 5c6fd05f3..d1d48cd0a 100644
--- a/lib/pleroma/web/mastodon_api/views/status_view.ex
+++ b/lib/pleroma/web/mastodon_api/views/status_view.ex
@@ -193,10 +193,18 @@ defmodule Pleroma.Web.MastodonAPI.StatusView do
     cc = object["cc"] || []
 
     cond do
-      public in to -> "public"
-      public in cc -> "unlisted"
-      Enum.any?(to, &String.contains?(&1, "/followers")) -> "private"
-      true -> "direct"
+      public in to ->
+        "public"
+
+      public in cc ->
+        "unlisted"
+
+      # this should use the sql for the object's activity
+      Enum.any?(to, &String.contains?(&1, "/followers")) ->
+        "private"
+
+      true ->
+        "direct"
     end
   end
 end
diff --git a/lib/pleroma/web/router.ex b/lib/pleroma/web/router.ex
index 726275158..924254895 100644
--- a/lib/pleroma/web/router.ex
+++ b/lib/pleroma/web/router.ex
@@ -107,6 +107,8 @@ defmodule Pleroma.Web.Router do
 
     get("/timelines/home", MastodonAPIController, :home_timeline)
 
+    get("/timelines/direct", MastodonAPIController, :dm_timeline)
+
     get("/favourites", MastodonAPIController, :favourites)
 
     post("/statuses", MastodonAPIController, :post_status)
diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex
index 33041ec12..6aac472dc 100644
--- a/lib/pleroma/web/streamer.ex
+++ b/lib/pleroma/web/streamer.ex
@@ -46,6 +46,19 @@ defmodule Pleroma.Web.Streamer do
     {:noreply, topics}
   end
 
+  def handle_cast(%{action: :stream, topic: "direct", item: item}, topics) do
+    recipient_topics =
+      User.get_recipients_from_activity(item)
+      |> Enum.map(fn %{id: id} -> "direct:#{id}" end)
+
+    Enum.each(recipient_topics || [], fn user_topic ->
+      Logger.debug("Trying to push direct message to #{user_topic}\n\n")
+      push_to_socket(topics, user_topic, item)
+    end)
+
+    {:noreply, topics}
+  end
+
   def handle_cast(%{action: :stream, topic: "user", item: %Notification{} = item}, topics) do
     topic = "user:#{item.user_id}"
 
@@ -137,8 +150,8 @@ defmodule Pleroma.Web.Streamer do
     end)
   end
 
-  defp internal_topic("user", socket) do
-    "user:#{socket.assigns[:user].id}"
+  defp internal_topic(topic, socket) when topic in ~w[user, direct] do
+    "#{topic}:#{socket.assigns[:user].id}"
   end
 
   defp internal_topic(topic, _), do: topic