git.squeep.com Git - akkoma/blob - lib/pleroma/marker.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.Marker do
   6   use Ecto.Schema
   7
   8   import Ecto.Changeset
   9   import Ecto.Query
  10
  11   alias Ecto.Multi
  12   alias Pleroma.Repo
  13   alias Pleroma.User
  14   alias __MODULE__
  15
  16   @timelines ["notifications"]
  17
  18   schema "markers" do
  19     field(:last_read_id, :string, default: "")
  20     field(:timeline, :string, default: "")
  21     field(:lock_version, :integer, default: 0)
  22     field(:unread_count, :integer, default: 0)
  23
  24     belongs_to(:user, User, type: FlakeId.Ecto.CompatType)
  25     timestamps()
  26   end
  27
  28   def get_markers(user, timelines \\ []) do
  29     Repo.all(get_query(user, timelines))
  30   end
  31
  32   def upsert(%User{} = user, attrs) do
  33     attrs
  34     |> Map.take(@timelines)
  35     |> Enum.reduce(Multi.new(), fn {timeline, timeline_attrs}, multi ->
  36       marker =
  37         user
  38         |> get_marker(timeline)
  39         |> changeset(timeline_attrs)
  40
  41       Multi.insert(multi, timeline, marker,
  42         returning: true,
  43         on_conflict: {:replace, [:last_read_id, :unread_count]},
  44         conflict_target: [:user_id, :timeline]
  45       )
  46     end)
  47     |> Repo.transaction()
  48   end
  49
  50   @spec multi_set_unread_count(Multi.t(), User.t(), String.t()) :: Multi.t()
  51   def multi_set_unread_count(multi, %User{} = user, "notifications") do
  52     multi
  53     |> Multi.run(:counters, fn _repo, _changes ->
  54       query =
  55         from(q in Pleroma.Notification,
  56           where: q.user_id == ^user.id,
  57           select: %{
  58             timeline: "notifications",
  59             user_id: type(^user.id, :string),
  60             unread_count: fragment("SUM( CASE WHEN seen = false THEN 1 ELSE 0 END )"),
  61             last_read_id: type(fragment("MAX( CASE WHEN seen = true THEN id ELSE null END )"), :string)
  62           }
  63         )
  64
  65       {:ok, Repo.one(query)}
  66     end)
  67     |> Multi.insert(
  68       :marker,
  69       fn %{counters: attrs} ->
  70         Marker
  71         |> struct(attrs)
  72         |> Ecto.Changeset.change()
  73       end,
  74       returning: true,
  75       on_conflict: {:replace, [:last_read_id, :unread_count]},
  76       conflict_target: [:user_id, :timeline]
  77     )
  78   end
  79
  80   def set_unread_count(%User{} = user, timeline) do
  81     Multi.new()
  82     |> multi_set_unread_count(user, timeline)
  83     |> Repo.transaction()
  84   end
  85
  86   defp get_marker(user, timeline) do
  87     case Repo.find_resource(get_query(user, timeline)) do
  88       {:ok, marker} -> %__MODULE__{marker | user: user}
  89       _ -> %__MODULE__{timeline: timeline, user_id: user.id}
  90     end
  91   end
  92
  93   @doc false
  94   defp changeset(marker, attrs) do
  95     marker
  96     |> cast(attrs, [:last_read_id, :unread_count])
  97     |> validate_required([:user_id, :timeline, :last_read_id])
  98     |> validate_inclusion(:timeline, @timelines)
  99   end
 100
 101   defp by_timeline(query, timeline) do
 102     from(m in query, where: m.timeline in ^List.wrap(timeline))
 103   end
 104
 105   defp by_user_id(query, id), do: from(m in query, where: m.user_id == ^id)
 106
 107   defp get_query(user, timelines) do
 108     __MODULE__
 109     |> by_user_id(user.id)
 110     |> by_timeline(timelines)
 111   end
 112 end