2 files changed, 159 insertions, 0 deletions
diff --git a/lib/pleroma/conversation.ex b/lib/pleroma/conversation.ex
new file mode 100644
index 000000000..a77a7cd6e
--- /dev/null
+++ b/lib/pleroma/conversation.ex
@@ -0,0 +1,70 @@
+# 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)
+
+    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)
+
+      participations =
+        Enum.map(users, fn user ->
+          {:ok, participation} =
+            Participation.create_for_user_and_conversation(user, conversation)
+
+          participation
+        end)
+
+      %{
+        conversation
+        | participations: participations
+      }
+    end
+  end
+end
diff --git a/lib/pleroma/conversation/participation.ex b/lib/pleroma/conversation/participation.ex
new file mode 100644
index 000000000..1a2ceafeb
--- /dev/null
+++ b/lib/pleroma/conversation/participation.ex
@@ -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.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)
+  end
+
+  def for_user_with_last_activity_id(user, params \\ %{}) do
+    for_user(user, params)
+    |> Repo.preload(:conversation)
+    |> Enum.map(fn participation ->
+      # TODO: Don't load all those activities, just get the most recent
+      # Involves splitting up the query.
+      activities =
+        ActivityPub.fetch_activities_for_context(participation.conversation.ap_id, %{
+          "user" => user,
+          "blocking_user" => user
+        })
+
+      activity_id =
+        case activities do
+          [activity | _] -> activity.id
+          _ -> nil
+        end
+
+      %{
+        participation
+        | last_activity_id: activity_id
+      }
+    end)
+  end
+end