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: ^user.id,
  60             unread_count: fragment("SUM( CASE WHEN seen = false THEN 1 ELSE 0 END ) as unread_count")
  61           }
  62         )
  63
  64       {:ok, Repo.one(query)}
  65     end)
  66     |> Multi.insert(
  67       :marker,
  68       fn %{counters: attrs} ->
  69         Marker
  70         |> struct(attrs)
  71         |> Ecto.Changeset.change()
  72       end,
  73       returning: true,
  74       on_conflict: {:replace, [:last_read_id, :unread_count]},
  75       conflict_target: [:user_id, :timeline]
  76     )
  77   end
  78
  79   def set_unread_count(%User{} = user, timeline) do
  80     Multi.new()
  81     |> multi_set_unread_count(user, timeline)
  82     |> Repo.transaction()
  83   end
  84
  85   defp get_marker(user, timeline) do
  86     case Repo.find_resource(get_query(user, timeline)) do
  87       {:ok, marker} -> %__MODULE__{marker | user: user}
  88       _ -> %__MODULE__{timeline: timeline, user_id: user.id}
  89     end
  90   end
  91
  92   @doc false
  93   defp changeset(marker, attrs) do
  94     marker
  95     |> cast(attrs, [:last_read_id, :unread_count])
  96     |> validate_required([:user_id, :timeline, :last_read_id])
  97     |> validate_inclusion(:timeline, @timelines)
  98   end
  99
 100   defp by_timeline(query, timeline) do
 101     from(m in query, where: m.timeline in ^List.wrap(timeline))
 102   end
 103
 104   defp by_user_id(query, id), do: from(m in query, where: m.user_id == ^id)
 105
 106   defp get_query(user, timelines) do
 107     __MODULE__
 108     |> by_user_id(user.id)
 109     |> by_timeline(timelines)
 110   end
 111 end