git.squeep.com Git - akkoma/blob - lib/pleroma/scheduled_activity.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.ScheduledActivity do
   6   use Ecto.Schema
   7
   8   alias Ecto.Multi
   9   alias Pleroma.Config
  10   alias Pleroma.Repo
  11   alias Pleroma.ScheduledActivity
  12   alias Pleroma.User
  13   alias Pleroma.Web.CommonAPI.Utils
  14   alias Pleroma.Workers.ScheduledActivityWorker
  15
  16   import Ecto.Query
  17   import Ecto.Changeset
  18
  19   @type t :: %__MODULE__{}
  20
  21   @min_offset :timer.minutes(5)
  22
  23   schema "scheduled_activities" do
  24     belongs_to(:user, User, type: FlakeId.Ecto.CompatType)
  25     field(:scheduled_at, :naive_datetime)
  26     field(:params, :map)
  27
  28     timestamps()
  29   end
  30
  31   def changeset(%ScheduledActivity{} = scheduled_activity, attrs) do
  32     scheduled_activity
  33     |> cast(attrs, [:scheduled_at, :params])
  34     |> validate_required([:scheduled_at, :params])
  35     |> validate_scheduled_at()
  36     |> with_media_attachments()
  37   end
  38
  39   defp with_media_attachments(
  40          %{changes: %{params: %{"media_ids" => media_ids} = params}} = changeset
  41        )
  42        when is_list(media_ids) do
  43     media_attachments = Utils.attachments_from_ids(%{"media_ids" => media_ids})
  44
  45     params =
  46       params
  47       |> Map.put("media_attachments", media_attachments)
  48       |> Map.put("media_ids", media_ids)
  49
  50     put_change(changeset, :params, params)
  51   end
  52
  53   defp with_media_attachments(changeset), do: changeset
  54
  55   def update_changeset(%ScheduledActivity{} = scheduled_activity, attrs) do
  56     scheduled_activity
  57     |> cast(attrs, [:scheduled_at])
  58     |> validate_required([:scheduled_at])
  59     |> validate_scheduled_at()
  60   end
  61
  62   def validate_scheduled_at(changeset) do
  63     validate_change(changeset, :scheduled_at, fn _, scheduled_at ->
  64       cond do
  65         not far_enough?(scheduled_at) ->
  66           [scheduled_at: "must be at least 5 minutes from now"]
  67
  68         exceeds_daily_user_limit?(changeset.data.user_id, scheduled_at) ->
  69           [scheduled_at: "daily limit exceeded"]
  70
  71         exceeds_total_user_limit?(changeset.data.user_id) ->
  72           [scheduled_at: "total limit exceeded"]
  73
  74         true ->
  75           []
  76       end
  77     end)
  78   end
  79
  80   def exceeds_daily_user_limit?(user_id, scheduled_at) do
  81     ScheduledActivity
  82     |> where(user_id: ^user_id)
  83     |> where([sa], type(sa.scheduled_at, :date) == type(^scheduled_at, :date))
  84     |> select([sa], count(sa.id))
  85     |> Repo.one()
  86     |> Kernel.>=(Config.get([ScheduledActivity, :daily_user_limit]))
  87   end
  88
  89   def exceeds_total_user_limit?(user_id) do
  90     ScheduledActivity
  91     |> where(user_id: ^user_id)
  92     |> select([sa], count(sa.id))
  93     |> Repo.one()
  94     |> Kernel.>=(Config.get([ScheduledActivity, :total_user_limit]))
  95   end
  96
  97   def far_enough?(scheduled_at) when is_binary(scheduled_at) do
  98     with {:ok, scheduled_at} <- Ecto.Type.cast(:naive_datetime, scheduled_at) do
  99       far_enough?(scheduled_at)
 100     else
 101       _ -> false
 102     end
 103   end
 104
 105   def far_enough?(scheduled_at) do
 106     now = NaiveDateTime.utc_now()
 107     diff = NaiveDateTime.diff(scheduled_at, now, :millisecond)
 108     diff > @min_offset
 109   end
 110
 111   def new(%User{} = user, attrs) do
 112     changeset(%ScheduledActivity{user_id: user.id}, attrs)
 113   end
 114
 115   @doc """
 116   Creates ScheduledActivity and add to queue to perform at scheduled_at date
 117   """
 118   @spec create(User.t(), map()) :: {:ok, ScheduledActivity.t()} | {:error, Ecto.Changeset.t()}
 119   def create(%User{} = user, attrs) do
 120     Multi.new()
 121     |> Multi.insert(:scheduled_activity, new(user, attrs))
 122     |> maybe_add_jobs(Config.get([ScheduledActivity, :enabled]))
 123     |> Repo.transaction()
 124     |> case do
 125       {:ok, %{scheduled_activity: scheduled_activity}} ->
 126         {:ok, scheduled_activity}
 127
 128       {:error, _, changeset, _} ->
 129         {:error, changeset}
 130     end
 131   end
 132
 133   defp maybe_add_jobs(multi, true) do
 134     multi
 135     |> Multi.run(:scheduled_activity_job, fn _repo, %{scheduled_activity: activity} ->
 136       %{activity_id: activity.id}
 137       |> ScheduledActivityWorker.new(scheduled_at: activity.scheduled_at)
 138       |> Oban.insert()
 139     end)
 140   end
 141
 142   defp maybe_add_jobs(multi, _), do: multi
 143
 144   def get(%User{} = user, scheduled_activity_id) do
 145     ScheduledActivity
 146     |> where(user_id: ^user.id)
 147     |> where(id: ^scheduled_activity_id)
 148     |> Repo.one()
 149   end
 150
 151   @spec update(ScheduledActivity.t(), map()) ::
 152           {:ok, ScheduledActivity.t()} | {:error, Ecto.Changeset.t()}
 153   def update(%ScheduledActivity{id: id} = scheduled_activity, attrs) do
 154     with {:error, %Ecto.Changeset{valid?: true} = changeset} <-
 155            {:error, update_changeset(scheduled_activity, attrs)} do
 156       Multi.new()
 157       |> Multi.update(:scheduled_activity, changeset)
 158       |> Multi.update_all(:scheduled_job, job_query(id),
 159         set: [scheduled_at: changeset.changes[:scheduled_at]]
 160       )
 161       |> Repo.transaction()
 162       |> case do
 163         {:ok, %{scheduled_activity: scheduled_activity}} ->
 164           {:ok, scheduled_activity}
 165
 166         {:error, _, changeset, _} ->
 167           {:error, changeset}
 168       end
 169     end
 170   end
 171
 172   def delete_job(%ScheduledActivity{id: id} = _scheduled_activity) do
 173     id
 174     |> job_query
 175     |> Repo.delete_all()
 176   end
 177
 178   def delete(%ScheduledActivity{} = scheduled_activity) do
 179     Repo.delete(scheduled_activity)
 180   end
 181
 182   def delete(id) when is_binary(id) or is_integer(id) do
 183     ScheduledActivity
 184     |> where(id: ^id)
 185     |> select([sa], sa)
 186     |> Repo.delete_all()
 187     |> case do
 188       {1, [scheduled_activity]} -> {:ok, scheduled_activity}
 189       _ -> :error
 190     end
 191   end
 192
 193   def for_user_query(%User{} = user) do
 194     ScheduledActivity
 195     |> where(user_id: ^user.id)
 196   end
 197
 198   def due_activities(offset \\ 0) do
 199     naive_datetime =
 200       NaiveDateTime.utc_now()
 201       |> NaiveDateTime.add(offset, :millisecond)
 202
 203     ScheduledActivity
 204     |> where([sa], sa.scheduled_at < ^naive_datetime)
 205     |> Repo.all()
 206   end
 207
 208   def job_query(scheduled_activity_id) do
 209     from(j in Oban.Job,
 210       where: j.queue == "scheduled_activities",
 211       where: fragment("args ->> 'activity_id' = ?::text", ^to_string(scheduled_activity_id))
 212     )
 213   end
 214 end