Merge branch 'develop' of https://git.pleroma.social/pleroma/pleroma into develop
[akkoma] / 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