git.squeep.com Git - akkoma/blob - lib/pleroma/conversation/participation.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Conversation.Participation do
   6   use Ecto.Schema
   7   alias Pleroma.Conversation
   8   alias Pleroma.Conversation.Participation.RecipientShip
   9   alias Pleroma.Repo
  10   alias Pleroma.User
  11   alias Pleroma.Web.ActivityPub.ActivityPub
  12   import Ecto.Changeset
  13   import Ecto.Query
  14
  15   schema "conversation_participations" do
  16     belongs_to(:user, User, type: FlakeId.Ecto.CompatType)
  17     belongs_to(:conversation, Conversation)
  18     field(:read, :boolean, default: false)
  19     field(:last_activity_id, FlakeId.Ecto.CompatType, virtual: true)
  20
  21     has_many(:recipient_ships, RecipientShip)
  22     has_many(:recipients, through: [:recipient_ships, :user])
  23
  24     timestamps()
  25   end
  26
  27   def creation_cng(struct, params) do
  28     struct
  29     |> cast(params, [:user_id, :conversation_id, :read])
  30     |> validate_required([:user_id, :conversation_id])
  31   end
  32
  33   def create_for_user_and_conversation(user, conversation, opts \\ []) do
  34     read = !!opts[:read]
  35     invisible_conversation = !!opts[:invisible_conversation]
  36
  37     update_on_conflict =
  38       if(invisible_conversation, do: [], else: [read: read])
  39       |> Keyword.put(:updated_at, NaiveDateTime.utc_now())
  40
  41     %__MODULE__{}
  42     |> creation_cng(%{
  43       user_id: user.id,
  44       conversation_id: conversation.id,
  45       read: invisible_conversation || read
  46     })
  47     |> Repo.insert(
  48       on_conflict: [set: update_on_conflict],
  49       returning: true,
  50       conflict_target: [:user_id, :conversation_id]
  51     )
  52   end
  53
  54   def read_cng(struct, params) do
  55     struct
  56     |> cast(params, [:read])
  57     |> validate_required([:read])
  58   end
  59
  60   def mark_as_read(%User{} = user, %Conversation{} = conversation) do
  61     with %__MODULE__{} = participation <- for_user_and_conversation(user, conversation) do
  62       mark_as_read(participation)
  63     end
  64   end
  65
  66   def mark_as_read(participation) do
  67     __MODULE__
  68     |> where(id: ^participation.id)
  69     |> update(set: [read: true])
  70     |> select([p], p)
  71     |> Repo.update_all([])
  72     |> case do
  73       {1, [participation]} ->
  74         participation = Repo.preload(participation, :user)
  75         User.set_unread_conversation_count(participation.user)
  76         {:ok, participation}
  77
  78       error ->
  79         error
  80     end
  81   end
  82
  83   def mark_all_as_read(%User{local: true} = user, %User{} = target_user) do
  84     target_conversation_ids =
  85       __MODULE__
  86       |> where([p], p.user_id == ^target_user.id)
  87       |> select([p], p.conversation_id)
  88       |> Repo.all()
  89
  90     __MODULE__
  91     |> where([p], p.user_id == ^user.id)
  92     |> where([p], p.conversation_id in ^target_conversation_ids)
  93     |> update([p], set: [read: true])
  94     |> Repo.update_all([])
  95
  96     {:ok, user} = User.set_unread_conversation_count(user)
  97     {:ok, user, []}
  98   end
  99
 100   def mark_all_as_read(%User{} = user, %User{}), do: {:ok, user, []}
 101
 102   def mark_all_as_read(%User{} = user) do
 103     {_, participations} =
 104       __MODULE__
 105       |> where([p], p.user_id == ^user.id)
 106       |> where([p], not p.read)
 107       |> update([p], set: [read: true])
 108       |> select([p], p)
 109       |> Repo.update_all([])
 110
 111     {:ok, user} = User.set_unread_conversation_count(user)
 112     {:ok, user, participations}
 113   end
 114
 115   def mark_as_unread(participation) do
 116     participation
 117     |> read_cng(%{read: false})
 118     |> Repo.update()
 119   end
 120
 121   def for_user(user, params \\ %{}) do
 122     from(p in __MODULE__,
 123       where: p.user_id == ^user.id,
 124       order_by: [desc: p.updated_at],
 125       preload: [conversation: [:users]]
 126     )
 127     |> restrict_recipients(user, params)
 128     |> Pleroma.Pagination.fetch_paginated(params)
 129   end
 130
 131   def restrict_recipients(query, user, %{"recipients" => user_ids}) do
 132     user_ids =
 133       [user.id | user_ids]
 134       |> Enum.uniq()
 135       |> Enum.reduce([], fn user_id, acc ->
 136         case FlakeId.Ecto.CompatType.dump(user_id) do
 137           {:ok, user_id} -> [user_id | acc]
 138           _ -> acc
 139         end
 140       end)
 141
 142     conversation_subquery =
 143       __MODULE__
 144       |> group_by([p], p.conversation_id)
 145       |> having(
 146         [p],
 147         count(p.user_id) == ^length(user_ids) and
 148           fragment("array_agg(?) @> ?", p.user_id, ^user_ids)
 149       )
 150       |> select([p], %{id: p.conversation_id})
 151
 152     query
 153     |> join(:inner, [p], c in subquery(conversation_subquery), on: p.conversation_id == c.id)
 154   end
 155
 156   def restrict_recipients(query, _, _), do: query
 157
 158   def for_user_and_conversation(user, conversation) do
 159     from(p in __MODULE__,
 160       where: p.user_id == ^user.id,
 161       where: p.conversation_id == ^conversation.id
 162     )
 163     |> Repo.one()
 164   end
 165
 166   def for_user_with_last_activity_id(user, params \\ %{}) do
 167     for_user(user, params)
 168     |> Enum.map(fn participation ->
 169       activity_id =
 170         ActivityPub.fetch_latest_activity_id_for_context(participation.conversation.ap_id, %{
 171           "user" => user,
 172           "blocking_user" => user
 173         })
 174
 175       %{
 176         participation
 177         | last_activity_id: activity_id
 178       }
 179     end)
 180     |> Enum.filter(& &1.last_activity_id)
 181   end
 182
 183   def get(_, _ \\ [])
 184   def get(nil, _), do: nil
 185
 186   def get(id, params) do
 187     query =
 188       if preload = params[:preload] do
 189         from(p in __MODULE__,
 190           preload: ^preload
 191         )
 192       else
 193         __MODULE__
 194       end
 195
 196     Repo.get(query, id)
 197   end
 198
 199   def set_recipients(participation, user_ids) do
 200     user_ids =
 201       [participation.user_id | user_ids]
 202       |> Enum.uniq()
 203
 204     Repo.transaction(fn ->
 205       query =
 206         from(r in RecipientShip,
 207           where: r.participation_id == ^participation.id
 208         )
 209
 210       Repo.delete_all(query)
 211
 212       users =
 213         from(u in User,
 214           where: u.id in ^user_ids
 215         )
 216         |> Repo.all()
 217
 218       RecipientShip.create(users, participation)
 219       :ok
 220     end)
 221
 222     {:ok, Repo.preload(participation, :recipients, force: true)}
 223   end
 224
 225   def unread_conversation_count_for_user(user) do
 226     from(p in __MODULE__,
 227       where: p.user_id == ^user.id,
 228       where: not p.read,
 229       select: %{count: count(p.id)}
 230     )
 231   end
 232 end