Merge branch 'conversations_three' into 'develop'

Conversations once more.

See merge request pleroma/pleroma!1119

lib/pleroma/conversation.ex

@@ -0,0 +1,75 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Conversation do
+  alias Pleroma.Conversation.Participation
+  alias Pleroma.Repo
+  alias Pleroma.User
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  schema "conversations" do
+    # This is the context ap id.
+    field(:ap_id, :string)
+    has_many(:participations, Participation)
+    has_many(:users, through: [:participations, :user])
+
+    timestamps()
+  end
+
+  def creation_cng(struct, params) do
+    struct
+    |> cast(params, [:ap_id])
+    |> validate_required([:ap_id])
+    |> unique_constraint(:ap_id)
+  end
+
+  def create_for_ap_id(ap_id) do
+    %__MODULE__{}
+    |> creation_cng(%{ap_id: ap_id})
+    |> Repo.insert(
+      on_conflict: [set: [updated_at: NaiveDateTime.utc_now()]],
+      returning: true,
+      conflict_target: :ap_id
+    )
+  end
+
+  def get_for_ap_id(ap_id) do
+    Repo.get_by(__MODULE__, ap_id: ap_id)
+  end
+
+  @doc """
+  This will
+  1. Create a conversation if there isn't one already
+  2. Create a participation for all the people involved who don't have one already
+  3. Bump all relevant participations to 'unread'
+  """
+  def create_or_bump_for(activity) do
+    with true <- Pleroma.Web.ActivityPub.Visibility.is_direct?(activity),
+         object <- Pleroma.Object.normalize(activity),
+         "Create" <- activity.data["type"],
+         "Note" <- object.data["type"],
+         ap_id when is_binary(ap_id) and byte_size(ap_id) > 0 <- object.data["context"] do
+      {:ok, conversation} = create_for_ap_id(ap_id)
+
+      users = User.get_users_from_set(activity.recipients, false)
+
+      participations =
+        Enum.map(users, fn user ->
+          {:ok, participation} =
+            Participation.create_for_user_and_conversation(user, conversation)
+
+          participation
+        end)
+
+      {:ok,
+       %{
+         conversation
+         | participations: participations
+       }}
+    else
+      e -> {:error, e}
+    end
+  end
+end

lib/pleroma/conversation/participation.ex

@@ -0,0 +1,81 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Conversation.Participation do
+  use Ecto.Schema
+  alias Pleroma.Conversation
+  alias Pleroma.Repo
+  alias Pleroma.User
+  alias Pleroma.Web.ActivityPub.ActivityPub
+  import Ecto.Changeset
+  import Ecto.Query
+
+  schema "conversation_participations" do
+    belongs_to(:user, User, type: Pleroma.FlakeId)
+    belongs_to(:conversation, Conversation)
+    field(:read, :boolean, default: false)
+    field(:last_activity_id, Pleroma.FlakeId, virtual: true)
+
+    timestamps()
+  end
+
+  def creation_cng(struct, params) do
+    struct
+    |> cast(params, [:user_id, :conversation_id])
+    |> validate_required([:user_id, :conversation_id])
+  end
+
+  def create_for_user_and_conversation(user, conversation) do
+    %__MODULE__{}
+    |> creation_cng(%{user_id: user.id, conversation_id: conversation.id})
+    |> Repo.insert(
+      on_conflict: [set: [read: false, updated_at: NaiveDateTime.utc_now()]],
+      returning: true,
+      conflict_target: [:user_id, :conversation_id]
+    )
+  end
+
+  def read_cng(struct, params) do
+    struct
+    |> cast(params, [:read])
+    |> validate_required([:read])
+  end
+
+  def mark_as_read(participation) do
+    participation
+    |> read_cng(%{read: true})
+    |> Repo.update()
+  end
+
+  def mark_as_unread(participation) do
+    participation
+    |> read_cng(%{read: false})
+    |> Repo.update()
+  end
+
+  def for_user(user, params \\ %{}) do
+    from(p in __MODULE__,
+      where: p.user_id == ^user.id,
+      order_by: [desc: p.updated_at]
+    )
+    |> Pleroma.Pagination.fetch_paginated(params)
+    |> Repo.preload(conversation: [:users])
+  end
+
+  def for_user_with_last_activity_id(user, params \\ %{}) do
+    for_user(user, params)
+    |> Enum.map(fn participation ->
+      activity_id =
+        ActivityPub.fetch_latest_activity_id_for_context(participation.conversation.ap_id, %{
+          "user" => user,
+          "blocking_user" => user
+        })
+
+      %{
+        participation
+        | last_activity_id: activity_id
+      }
+    end)
+  end
+end

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -4,6 +4,7 @@
 
 defmodule Pleroma.Web.ActivityPub.ActivityPub do
   alias Pleroma.Activity
+  alias Pleroma.Conversation
   alias Pleroma.Instances
   alias Pleroma.Notification
   alias Pleroma.Object
@@ -141,7 +142,14 @@ def insert(map, local \\ true, fake \\ false) when is_map(map) do
       end)
 
       Notification.create_notifications(activity)
+
+      participations =
+        activity
+        |> Conversation.create_or_bump_for()
+        |> get_participations()
+
       stream_out(activity)
+      stream_out_participations(participations)
       {:ok, activity}
     else
       %Activity{} = activity ->
@@ -164,6 +172,19 @@ def insert(map, local \\ true, fake \\ false) when is_map(map) do
     end
   end
 
+  defp get_participations({:ok, %{participations: participations}}), do: participations
+  defp get_participations(_), do: []
+
+  def stream_out_participations(participations) do
+    participations =
+      participations
+      |> Repo.preload(:user)
+
+    Enum.each(participations, fn participation ->
+      Pleroma.Web.Streamer.stream("participation", participation)
+    end)
+  end
+
   def stream_out(activity) do
     public = "https://www.w3.org/ns/activitystreams#Public"
 
@@ -195,6 +216,7 @@ def stream_out(activity) do
           end
         end
       else
+        # TODO: Write test, replace with visibility test
         if !Enum.member?(activity.data["cc"] || [], public) &&
              !Enum.member?(
                activity.data["to"],
@@ -457,35 +479,44 @@ def flag(
     end
   end
 
-  def fetch_activities_for_context(context, opts \\ %{}) do
+  defp fetch_activities_for_context_query(context, opts) do
     public = ["https://www.w3.org/ns/activitystreams#Public"]
 
     recipients =
       if opts["user"], do: [opts["user"].ap_id | opts["user"].following] ++ public, else: public
 
-    query = from(activity in Activity)
+    from(activity in Activity)
-
+    |> restrict_blocked(opts)
-    query =
+    |> restrict_recipients(recipients, opts["user"])
-      query
+    |> where(
-      |> restrict_blocked(opts)
+      [activity],
-      |> restrict_recipients(recipients, opts["user"])
+      fragment(
-
+        "?->>'type' = ? and ?->>'context' = ?",
-    query =
+        activity.data,
-      from(
+        "Create",
-        activity in query,
+        activity.data,
-        where:
+        ^context
-          fragment(
-            "?->>'type' = ? and ?->>'context' = ?",
-            activity.data,
-            "Create",
-            activity.data,
-            ^context
-          ),
-        order_by: [desc: :id]
       )
-      |> Activity.with_preloaded_object()
+    )
+    |> order_by([activity], desc: activity.id)
+  end
 
-    Repo.all(query)
+  @spec fetch_activities_for_context(String.t(), keyword() | map()) :: [Activity.t()]
+  def fetch_activities_for_context(context, opts \\ %{}) do
+    context
+    |> fetch_activities_for_context_query(opts)
+    |> Activity.with_preloaded_object()
+    |> Repo.all()
+  end
+
+  @spec fetch_latest_activity_id_for_context(String.t(), keyword() | map()) ::
+          Pleroma.FlakeId.t() | nil
+  def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
+    context
+    |> fetch_activities_for_context_query(opts)
+    |> limit(1)
+    |> select([a], a.id)
+    |> Repo.one()
   end
 
   def fetch_public_activities(opts \\ %{}) do

lib/pleroma/web/mastodon_api/mastodon_api_controller.ex

@@ -8,6 +8,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
   alias Pleroma.Activity
   alias Pleroma.Bookmark
   alias Pleroma.Config
+  alias Pleroma.Conversation.Participation
   alias Pleroma.Filter
   alias Pleroma.Formatter
   alias Pleroma.Notification
@@ -24,6 +25,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
   alias Pleroma.Web.CommonAPI
   alias Pleroma.Web.MastodonAPI.AccountView
   alias Pleroma.Web.MastodonAPI.AppView
+  alias Pleroma.Web.MastodonAPI.ConversationView
   alias Pleroma.Web.MastodonAPI.FilterView
   alias Pleroma.Web.MastodonAPI.ListView
   alias Pleroma.Web.MastodonAPI.MastodonAPI
@@ -165,7 +167,7 @@ def user(%{assigns: %{user: for_user}} = conn, %{"id" => nickname_or_id}) do
     end
   end
 
-  @mastodon_api_level "2.5.0"
+  @mastodon_api_level "2.6.5"
 
   def masto_instance(conn, _params) do
     instance = Config.get(:instance)
@@ -1712,6 +1714,31 @@ def reports(%{assigns: %{user: user}} = conn, params) do
     end
   end
 
+  def conversations(%{assigns: %{user: user}} = conn, params) do
+    participations = Participation.for_user_with_last_activity_id(user, params)
+
+    conversations =
+      Enum.map(participations, fn participation ->
+        ConversationView.render("participation.json", %{participation: participation, user: user})
+      end)
+
+    conn
+    |> add_link_headers(:conversations, participations)
+    |> json(conversations)
+  end
+
+  def conversation_read(%{assigns: %{user: user}} = conn, %{"id" => participation_id}) do
+    with %Participation{} = participation <-
+           Repo.get_by(Participation, id: participation_id, user_id: user.id),
+         {:ok, participation} <- Participation.mark_as_read(participation) do
+      participation_view =
+        ConversationView.render("participation.json", %{participation: participation, user: user})
+
+      conn
+      |> json(participation_view)
+    end
+  end
+
   def try_render(conn, target, params)
       when is_binary(target) do
     res = render(conn, target, params)

lib/pleroma/web/mastodon_api/views/conversation_view.ex

@@ -0,0 +1,38 @@
+defmodule Pleroma.Web.MastodonAPI.ConversationView do
+  use Pleroma.Web, :view
+
+  alias Pleroma.Activity
+  alias Pleroma.Repo
+  alias Pleroma.Web.ActivityPub.ActivityPub
+  alias Pleroma.Web.MastodonAPI.AccountView
+  alias Pleroma.Web.MastodonAPI.StatusView
+
+  def render("participation.json", %{participation: participation, user: user}) do
+    participation = Repo.preload(participation, conversation: :users)
+
+    last_activity_id =
+      with nil <- participation.last_activity_id do
+        ActivityPub.fetch_latest_activity_id_for_context(participation.conversation.ap_id, %{
+          "user" => user,
+          "blocking_user" => user
+        })
+      end
+
+    activity = Activity.get_by_id_with_object(last_activity_id)
+
+    last_status = StatusView.render("status.json", %{activity: activity, for: user})
+
+    accounts =
+      AccountView.render("accounts.json", %{
+        users: participation.conversation.users,
+        as: :user
+      })
+
+    %{
+      id: participation.id |> to_string(),
+      accounts: accounts,
+      unread: !participation.read,
+      last_status: last_status
+    }
+  end
+end

lib/pleroma/web/router.ex

@@ -276,6 +276,9 @@ defmodule Pleroma.Web.Router do
 
       get("/suggestions", MastodonAPIController, :suggestions)
 
+      get("/conversations", MastodonAPIController, :conversations)
+      post("/conversations/:id/read", MastodonAPIController, :conversation_read)
+
       get("/endorsements", MastodonAPIController, :empty_array)
 
       get("/pleroma/flavour", MastodonAPIController, :get_flavour)

lib/pleroma/web/streamer.ex

@@ -6,6 +6,7 @@ defmodule Pleroma.Web.Streamer do
   use GenServer
   require Logger
   alias Pleroma.Activity
+  alias Pleroma.Conversation.Participation
   alias Pleroma.Notification
   alias Pleroma.Object
   alias Pleroma.User
@@ -71,6 +72,15 @@ def handle_cast(%{action: :stream, topic: "direct", item: item}, topics) do
     {:noreply, topics}
   end
 
+  def handle_cast(%{action: :stream, topic: "participation", item: participation}, topics) do
+    user_topic = "direct:#{participation.user_id}"
+    Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")
+
+    push_to_socket(topics, user_topic, participation)
+
+    {:noreply, topics}
+  end
+
   def handle_cast(%{action: :stream, topic: "list", item: item}, topics) do
     # filter the recipient list if the activity is not public, see #270.
     recipient_lists =
@@ -192,6 +202,19 @@ defp represent_update(%Activity{} = activity) do
     |> Jason.encode!()
   end
 
+  def represent_conversation(%Participation{} = participation) do
+    %{
+      event: "conversation",
+      payload:
+        Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{
+          participation: participation,
+          user: participation.user
+        })
+        |> Jason.encode!()
+    }
+    |> Jason.encode!()
+  end
+
   def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do
     Enum.each(topics[topic] || [], fn socket ->
       # Get the current user so we have up-to-date blocks etc.
@@ -214,6 +237,12 @@ def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = ite
     end)
   end
 
+  def push_to_socket(topics, topic, %Participation{} = participation) do
+    Enum.each(topics[topic] || [], fn socket ->
+      send(socket.transport_pid, {:text, represent_conversation(participation)})
+    end)
+  end
+
   def push_to_socket(topics, topic, %Activity{
         data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
       }) do

mix.exs

@@ -87,7 +87,7 @@ defp deps do
       {:bbcode, "~> 0.1"},
       {:ex_machina, "~> 2.3", only: :test},
       {:credo, "~> 0.9.3", only: [:dev, :test]},
-      {:mock, "~> 0.3.1", only: :test},
+      {:mock, "~> 0.3.3", only: :test},
       {:crypt,
        git: "https://github.com/msantos/crypt", ref: "1f2b58927ab57e72910191a7ebaeff984382a1d3"},
       {:cors_plug, "~> 1.5"},

mix.lock

@@ -46,7 +46,7 @@
   "mime": {:hex, :mime, "1.3.1", "30ce04ab3175b6ad0bdce0035cba77bba68b813d523d1aac73d9781b4d193cf8", [:mix], [], "hexpm"},
   "mimerl": {:hex, :mimerl, "1.2.0", "67e2d3f571088d5cfd3e550c383094b47159f3eee8ffa08e64106cdf5e981be3", [:rebar3], [], "hexpm"},
   "mochiweb": {:hex, :mochiweb, "2.15.0", "e1daac474df07651e5d17cc1e642c4069c7850dc4508d3db7263a0651330aacc", [:rebar3], [], "hexpm"},
-  "mock": {:hex, :mock, "0.3.1", "994f00150f79a0ea50dc9d86134cd9ebd0d177ad60bd04d1e46336cdfdb98ff9", [:mix], [{:meck, "~> 0.8.8", [hex: :meck, repo: "hexpm", optional: false]}], "hexpm"},
+  "mock": {:hex, :mock, "0.3.3", "42a433794b1291a9cf1525c6d26b38e039e0d3a360732b5e467bfc77ef26c914", [:mix], [{:meck, "~> 0.8.13", [hex: :meck, repo: "hexpm", optional: false]}], "hexpm"},
   "mogrify": {:hex, :mogrify, "0.6.1", "de1b527514f2d95a7bbe9642eb556061afb337e220cf97adbf3a4e6438ed70af", [:mix], [], "hexpm"},
   "nimble_parsec": {:hex, :nimble_parsec, "0.5.0", "90e2eca3d0266e5c53f8fbe0079694740b9c91b6747f2b7e3c5d21966bba8300", [:mix], [], "hexpm"},
   "parse_trans": {:hex, :parse_trans, "3.3.0", "09765507a3c7590a784615cfd421d101aec25098d50b89d7aa1d66646bc571c1", [:rebar3], [], "hexpm"},

priv/repo/migrations/20190408123347_create_conversations.exs

@@ -0,0 +1,26 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Repo.Migrations.CreateConversations do
+  use Ecto.Migration
+
+  def change do
+    create table(:conversations) do
+      add(:ap_id, :string, null: false)
+      timestamps()
+    end
+
+    create table(:conversation_participations) do
+      add(:user_id, references(:users, type: :uuid, on_delete: :delete_all))
+      add(:conversation_id, references(:conversations, on_delete: :delete_all))
+      add(:read, :boolean, default: false)
+
+      timestamps()
+    end
+
+    create index(:conversation_participations, [:conversation_id])
+    create unique_index(:conversation_participations, [:user_id, :conversation_id])
+    create unique_index(:conversations, [:ap_id])
+  end
+end

priv/repo/migrations/20190410152859_add_participation_updated_at_index.exs

@@ -0,0 +1,7 @@
+defmodule Pleroma.Repo.Migrations.AddParticipationUpdatedAtIndex do
+  use Ecto.Migration
+
+  def change do
+    create index(:conversation_participations, ["updated_at desc"])
+  end
+end

test/conversation/participation_test.exs

@@ -0,0 +1,89 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Conversation.ParticipationTest do
+  use Pleroma.DataCase
+  import Pleroma.Factory
+  alias Pleroma.Conversation.Participation
+  alias Pleroma.Web.CommonAPI
+
+  test "it creates a participation for a conversation and a user" do
+    user = insert(:user)
+    conversation = insert(:conversation)
+
+    {:ok, %Participation{} = participation} =
+      Participation.create_for_user_and_conversation(user, conversation)
+
+    assert participation.user_id == user.id
+    assert participation.conversation_id == conversation.id
+
+    :timer.sleep(1000)
+    # Creating again returns the same participation
+    {:ok, %Participation{} = participation_two} =
+      Participation.create_for_user_and_conversation(user, conversation)
+
+    assert participation.id == participation_two.id
+    refute participation.updated_at == participation_two.updated_at
+  end
+
+  test "recreating an existing participations sets it to unread" do
+    participation = insert(:participation, %{read: true})
+
+    {:ok, participation} =
+      Participation.create_for_user_and_conversation(
+        participation.user,
+        participation.conversation
+      )
+
+    refute participation.read
+  end
+
+  test "it marks a participation as read" do
+    participation = insert(:participation, %{read: false})
+    {:ok, participation} = Participation.mark_as_read(participation)
+
+    assert participation.read
+  end
+
+  test "it marks a participation as unread" do
+    participation = insert(:participation, %{read: true})
+    {:ok, participation} = Participation.mark_as_unread(participation)
+
+    refute participation.read
+  end
+
+  test "gets all the participations for a user, ordered by updated at descending" do
+    user = insert(:user)
+    {:ok, activity_one} = CommonAPI.post(user, %{"status" => "x", "visibility" => "direct"})
+    :timer.sleep(1000)
+    {:ok, activity_two} = CommonAPI.post(user, %{"status" => "x", "visibility" => "direct"})
+    :timer.sleep(1000)
+
+    {:ok, activity_three} =
+      CommonAPI.post(user, %{
+        "status" => "x",
+        "visibility" => "direct",
+        "in_reply_to_status_id" => activity_one.id
+      })
+
+    assert [participation_one, participation_two] = Participation.for_user(user)
+
+    object2 = Pleroma.Object.normalize(activity_two)
+    object3 = Pleroma.Object.normalize(activity_three)
+
+    assert participation_one.conversation.ap_id == object3.data["context"]
+    assert participation_two.conversation.ap_id == object2.data["context"]
+
+    # Pagination
+    assert [participation_one] = Participation.for_user(user, %{"limit" => 1})
+
+    assert participation_one.conversation.ap_id == object3.data["context"]
+
+    # With last_activity_id
+    assert [participation_one] =
+             Participation.for_user_with_last_activity_id(user, %{"limit" => 1})
+
+    assert participation_one.last_activity_id == activity_three.id
+  end
+end

test/conversation_test.exs

@@ -0,0 +1,137 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.ConversationTest do
+  use Pleroma.DataCase
+  alias Pleroma.Conversation
+  alias Pleroma.Web.CommonAPI
+
+  import Pleroma.Factory
+
+  test "it creates a conversation for given ap_id" do
+    assert {:ok, %Conversation{} = conversation} =
+             Conversation.create_for_ap_id("https://some_ap_id")
+
+    # Inserting again returns the same
+    assert {:ok, conversation_two} = Conversation.create_for_ap_id("https://some_ap_id")
+    assert conversation_two.id == conversation.id
+  end
+
+  test "public posts don't create conversations" do
+    user = insert(:user)
+    {:ok, activity} = CommonAPI.post(user, %{"status" => "Hey"})
+
+    object = Pleroma.Object.normalize(activity)
+    context = object.data["context"]
+
+    conversation = Conversation.get_for_ap_id(context)
+
+    refute conversation
+  end
+
+  test "it creates or updates a conversation and participations for a given DM" do
+    har = insert(:user)
+    jafnhar = insert(:user, local: false)
+    tridi = insert(:user)
+
+    {:ok, activity} =
+      CommonAPI.post(har, %{"status" => "Hey @#{jafnhar.nickname}", "visibility" => "direct"})
+
+    object = Pleroma.Object.normalize(activity)
+    context = object.data["context"]
+
+    conversation =
+      Conversation.get_for_ap_id(context)
+      |> Repo.preload(:participations)
+
+    assert conversation
+
+    assert Enum.find(conversation.participations, fn %{user_id: user_id} -> har.id == user_id end)
+
+    assert Enum.find(conversation.participations, fn %{user_id: user_id} ->
+             jafnhar.id == user_id
+           end)
+
+    {:ok, activity} =
+      CommonAPI.post(jafnhar, %{
+        "status" => "Hey @#{har.nickname}",
+        "visibility" => "direct",
+        "in_reply_to_status_id" => activity.id
+      })
+
+    object = Pleroma.Object.normalize(activity)
+    context = object.data["context"]
+
+    conversation_two =
+      Conversation.get_for_ap_id(context)
+      |> Repo.preload(:participations)
+
+    assert conversation_two.id == conversation.id
+
+    assert Enum.find(conversation_two.participations, fn %{user_id: user_id} ->
+             har.id == user_id
+           end)
+
+    assert Enum.find(conversation_two.participations, fn %{user_id: user_id} ->
+             jafnhar.id == user_id
+           end)
+
+    {:ok, activity} =
+      CommonAPI.post(tridi, %{
+        "status" => "Hey @#{har.nickname}",
+        "visibility" => "direct",
+        "in_reply_to_status_id" => activity.id
+      })
+
+    object = Pleroma.Object.normalize(activity)
+    context = object.data["context"]
+
+    conversation_three =
+      Conversation.get_for_ap_id(context)
+      |> Repo.preload([:participations, :users])
+
+    assert conversation_three.id == conversation.id
+
+    assert Enum.find(conversation_three.participations, fn %{user_id: user_id} ->
+             har.id == user_id
+           end)
+
+    assert Enum.find(conversation_three.participations, fn %{user_id: user_id} ->
+             jafnhar.id == user_id
+           end)
+
+    assert Enum.find(conversation_three.participations, fn %{user_id: user_id} ->
+             tridi.id == user_id
+           end)
+
+    assert Enum.find(conversation_three.users, fn %{id: user_id} ->
+             har.id == user_id
+           end)
+
+    assert Enum.find(conversation_three.users, fn %{id: user_id} ->
+             jafnhar.id == user_id
+           end)
+
+    assert Enum.find(conversation_three.users, fn %{id: user_id} ->
+             tridi.id == user_id
+           end)
+  end
+
+  test "create_or_bump_for returns the conversation with participations" do
+    har = insert(:user)
+    jafnhar = insert(:user, local: false)
+
+    {:ok, activity} =
+      CommonAPI.post(har, %{"status" => "Hey @#{jafnhar.nickname}", "visibility" => "direct"})
+
+    {:ok, conversation} = Conversation.create_or_bump_for(activity)
+
+    assert length(conversation.participations) == 2
+
+    {:ok, activity} =
+      CommonAPI.post(har, %{"status" => "Hey @#{jafnhar.nickname}", "visibility" => "public"})
+
+    assert {:error, _} = Conversation.create_or_bump_for(activity)
+  end
+end

test/support/factory.ex

@@ -5,6 +5,23 @@
 defmodule Pleroma.Factory do
   use ExMachina.Ecto, repo: Pleroma.Repo
 
+  def participation_factory do
+    conversation = insert(:conversation)
+    user = insert(:user)
+
+    %Pleroma.Conversation.Participation{
+      conversation: conversation,
+      user: user,
+      read: false
+    }
+  end
+
+  def conversation_factory do
+    %Pleroma.Conversation{
+      ap_id: sequence(:ap_id, &"https://some_conversation/#{&1}")
+    }
+  end
+
   def user_factory do
     user = %Pleroma.User{
       name: sequence(:name, &"Test テスト User #{&1}"),

test/web/activity_pub/activity_pub_test.exs

@@ -22,6 +22,28 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
     :ok
   end
 
+  describe "streaming out participations" do
+    test "it streams them out" do
+      user = insert(:user)
+      {:ok, activity} = CommonAPI.post(user, %{"status" => ".", "visibility" => "direct"})
+
+      {:ok, conversation} = Pleroma.Conversation.create_or_bump_for(activity)
+
+      participations =
+        conversation.participations
+        |> Repo.preload(:user)
+
+      with_mock Pleroma.Web.Streamer,
+        stream: fn _, _ -> nil end do
+        ActivityPub.stream_out_participations(conversation.participations)
+
+        Enum.each(participations, fn participation ->
+          assert called(Pleroma.Web.Streamer.stream("participation", participation))
+        end)
+      end
+    end
+  end
+
   describe "fetching restricted by visibility" do
     test "it restricts by the appropriate visibility" do
       user = insert(:user)
@@ -130,9 +152,15 @@ test "drops activities beyond a certain limit" do
     end
 
     test "doesn't drop activities with content being null" do
+      user = insert(:user)
+
       data = %{
-        "ok" => true,
+        "actor" => user.ap_id,
+        "to" => [],
         "object" => %{
+          "actor" => user.ap_id,
+          "to" => [],
+          "type" => "Note",
           "content" => nil
         }
       }
@@ -148,8 +176,17 @@ test "returns the activity if one with the same id is already in" do
     end
 
     test "inserts a given map into the activity database, giving it an id if it has none." do
+      user = insert(:user)
+
       data = %{
-        "ok" => true
+        "actor" => user.ap_id,
+        "to" => [],
+        "object" => %{
+          "actor" => user.ap_id,
+          "to" => [],
+          "type" => "Note",
+          "content" => "hey"
+        }
       }
 
       {:ok, %Activity{} = activity} = ActivityPub.insert(data)
@@ -159,9 +196,16 @@ test "inserts a given map into the activity database, giving it an id if it has
       given_id = "bla"
 
       data = %{
-        "ok" => true,
         "id" => given_id,
-        "context" => "blabla"
+        "actor" => user.ap_id,
+        "to" => [],
+        "context" => "blabla",
+        "object" => %{
+          "actor" => user.ap_id,
+          "to" => [],
+          "type" => "Note",
+          "content" => "hey"
+        }
       }
 
       {:ok, %Activity{} = activity} = ActivityPub.insert(data)
@@ -172,26 +216,39 @@ test "inserts a given map into the activity database, giving it an id if it has
     end
 
     test "adds a context when none is there" do
+      user = insert(:user)
+
       data = %{
-        "id" => "some_id",
+        "actor" => user.ap_id,
+        "to" => [],
         "object" => %{
-          "id" => "object_id"
+          "actor" => user.ap_id,
+          "to" => [],
+          "type" => "Note",
+          "content" => "hey"
         }
       }
 
       {:ok, %Activity{} = activity} = ActivityPub.insert(data)
+      object = Pleroma.Object.normalize(activity)
 
       assert is_binary(activity.data["context"])
-      assert is_binary(activity.data["object"]["context"])
+      assert is_binary(object.data["context"])
       assert activity.data["context_id"]
-      assert activity.data["object"]["context_id"]
+      assert object.data["context_id"]
     end
 
     test "adds an id to a given object if it lacks one and is a note and inserts it to the object database" do
+      user = insert(:user)
+
       data = %{
+        "actor" => user.ap_id,
+        "to" => [],
         "object" => %{
+          "actor" => user.ap_id,
+          "to" => [],
           "type" => "Note",
-          "ok" => true
+          "content" => "hey"
         }
       }
 

test/web/mastodon_api/mastodon_api_controller_test.exs

@@ -300,6 +300,65 @@ test "direct timeline", %{conn: conn} do
     assert status["url"] != direct.data["id"]
   end
 
+  test "Conversations", %{conn: conn} do
+    user_one = insert(:user)
+    user_two = insert(:user)
+
+    {:ok, user_two} = User.follow(user_two, user_one)
+
+    {:ok, direct} =
+      CommonAPI.post(user_one, %{
+        "status" => "Hi @#{user_two.nickname}!",
+        "visibility" => "direct"
+      })
+
+    {:ok, _follower_only} =
+      CommonAPI.post(user_one, %{
+        "status" => "Hi @#{user_two.nickname}!",
+        "visibility" => "private"
+      })
+
+    res_conn =
+      conn
+      |> assign(:user, user_one)
+      |> get("/api/v1/conversations")
+
+    assert response = json_response(res_conn, 200)
+
+    assert [
+             %{
+               "id" => res_id,
+               "accounts" => res_accounts,
+               "last_status" => res_last_status,
+               "unread" => unread
+             }
+           ] = response
+
+    assert length(res_accounts) == 2
+    assert is_binary(res_id)
+    assert unread == true
+    assert res_last_status["id"] == direct.id
+
+    # Apparently undocumented API endpoint
+    res_conn =
+      conn
+      |> assign(:user, user_one)
+      |> post("/api/v1/conversations/#{res_id}/read")
+
+    assert response = json_response(res_conn, 200)
+    assert length(response["accounts"]) == 2
+    assert response["last_status"]["id"] == direct.id
+    assert response["unread"] == false
+
+    # (vanilla) Mastodon frontend behaviour
+    res_conn =
+      conn
+      |> assign(:user, user_one)
+      |> get("/api/v1/statuses/#{res_last_status["id"]}/context")
+
+    assert %{"ancestors" => [], "descendants" => []} == json_response(res_conn, 200)
+  end
+
   test "doesn't include DMs from blocked users", %{conn: conn} do
     blocker = insert(:user)
     blocked = insert(:user)