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     |> transaction_response
 125   end
 126
 127   defp maybe_add_jobs(multi, true) do
 128     multi
 129     |> Multi.run(:scheduled_activity_job, fn _repo, %{scheduled_activity: activity} ->
 130       %{activity_id: activity.id}
 131       |> ScheduledActivityWorker.new(scheduled_at: activity.scheduled_at)
 132       |> Oban.insert()
 133     end)
 134   end
 135
 136   defp maybe_add_jobs(multi, _), do: multi
 137
 138   def get(%User{} = user, scheduled_activity_id) do
 139     ScheduledActivity
 140     |> where(user_id: ^user.id)
 141     |> where(id: ^scheduled_activity_id)
 142     |> Repo.one()
 143   end
 144
 145   @spec update(ScheduledActivity.t(), map()) ::
 146           {:ok, ScheduledActivity.t()} | {:error, Ecto.Changeset.t()}
 147   def update(%ScheduledActivity{id: id} = scheduled_activity, attrs) do
 148     with {:error, %Ecto.Changeset{valid?: true} = changeset} <-
 149            {:error, update_changeset(scheduled_activity, attrs)} do
 150       Multi.new()
 151       |> Multi.update(:scheduled_activity, changeset)
 152       |> Multi.update_all(:scheduled_job, job_query(id),
 153         set: [scheduled_at: get_field(changeset, :scheduled_at)]
 154       )
 155       |> Repo.transaction()
 156       |> transaction_response
 157     end
 158   end
 159
 160   @doc "Deletes a ScheduledActivity and linked jobs."
 161   @spec delete(ScheduledActivity.t() | binary() | integer) ::
 162           {:ok, ScheduledActivity.t()} | {:error, Ecto.Changeset.t()}
 163   def delete(%ScheduledActivity{id: id} = scheduled_activity) do
 164     Multi.new()
 165     |> Multi.delete(:scheduled_activity, scheduled_activity, stale_error_field: :id)
 166     |> Multi.delete_all(:jobs, job_query(id))
 167     |> Repo.transaction()
 168     |> transaction_response
 169   end
 170
 171   def delete(id) when is_binary(id) or is_integer(id) do
 172     delete(%__MODULE__{id: id})
 173   end
 174
 175   defp transaction_response(result) do
 176     case result do
 177       {:ok, %{scheduled_activity: scheduled_activity}} ->
 178         {:ok, scheduled_activity}
 179
 180       {:error, _, changeset, _} ->
 181         {:error, changeset}
 182     end
 183   end
 184
 185   def for_user_query(%User{} = user) do
 186     ScheduledActivity
 187     |> where(user_id: ^user.id)
 188   end
 189
 190   def due_activities(offset \\ 0) do
 191     naive_datetime =
 192       NaiveDateTime.utc_now()
 193       |> NaiveDateTime.add(offset, :millisecond)
 194
 195     ScheduledActivity
 196     |> where([sa], sa.scheduled_at < ^naive_datetime)
 197     |> Repo.all()
 198   end
 199
 200   def job_query(scheduled_activity_id) do
 201     from(j in Oban.Job,
 202       where: j.queue == "scheduled_activities",
 203       where: fragment("args ->> 'activity_id' = ?::text", ^to_string(scheduled_activity_id))
 204     )
 205   end
 206 end