--- /dev/null
+# 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),
+ "Create" <- activity.data["type"],
+ "Note" <- activity.data["object"]["type"],
+ ap_id when is_binary(ap_id) <- activity.data["object"]["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)
+
+ %{
+ conversation
+ | participations: participations
+ }
+ end
+ end
+end
--- /dev/null
+# 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
defmodule Pleroma.Web.ActivityPub.ActivityPub do
alias Pleroma.Activity
+ alias Pleroma.Conversation
alias Pleroma.Instances
alias Pleroma.Notification
alias Pleroma.Object
end)
Notification.create_notifications(activity)
+ Conversation.create_or_bump_for(activity)
stream_out(activity)
{:ok, activity}
else
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)
-
- query =
- query
- |> restrict_blocked(opts)
- |> restrict_recipients(recipients, opts["user"])
-
- query =
- from(
- activity in query,
- where:
- fragment(
- "?->>'type' = ? and ?->>'context' = ?",
- activity.data,
- "Create",
- activity.data,
- ^context
- ),
- order_by: [desc: :id]
+ from(activity in Activity)
+ |> restrict_blocked(opts)
+ |> restrict_recipients(recipients, opts["user"])
+ |> where(
+ [activity],
+ fragment(
+ "?->>'type' = ? and ?->>'context' = ?",
+ activity.data,
+ "Create",
+ activity.data,
+ ^context
)
- |> Activity.with_preloaded_object()
+ )
+ |> order_by([activity], desc: activity.id)
+ end
+
+ @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
- Repo.all(query)
+ @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
alias Pleroma.Activity
alias Pleroma.Bookmark
alias Pleroma.Config
+ alias Pleroma.Conversation.Participation
alias Pleroma.Filter
alias Pleroma.Notification
alias Pleroma.Object
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
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)
--- /dev/null
+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
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)
--- /dev/null
+# 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
--- /dev/null
+defmodule Pleroma.Repo.Migrations.AddParticipationUpdatedAtIndex do
+ use Ecto.Migration
+
+ def change do
+ create index(:conversation_participations, ["updated_at desc"])
+ end
+end
--- /dev/null
+# 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)
+ |> Repo.preload(:conversation)
+
+ assert participation_one.conversation.ap_id == activity_three.data["object"]["context"]
+ assert participation_two.conversation.ap_id == activity_two.data["object"]["context"]
+
+ # Pagination
+ assert [participation_one] =
+ Participation.for_user(user, %{limit: 1})
+ |> Repo.preload(:conversation)
+
+ assert participation_one.conversation.ap_id == activity_three.data["object"]["context"]
+
+ # With last_activity_id
+ assert [participation_one] =
+ Participation.for_user_with_last_activity_id(user, %{limit: 1})
+ |> Repo.preload(:conversation)
+
+ assert participation_one.last_activity_id == activity_three.id
+ end
+end
--- /dev/null
+# 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"})
+
+ context = activity.data["object"]["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"})
+
+ context = activity.data["object"]["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
+ })
+
+ context = activity.data["object"]["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
+ })
+
+ context = activity.data["object"]["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
+end
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}"),
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
}
}
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)
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)
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"
}
}
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"
}
}
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)
projects / akkoma / commitdiff