Merge branch 'develop' into issue/1276

config/config.exs

@@ -51,20 +51,6 @@ config :pleroma, Pleroma.Repo,
   telemetry_event: [Pleroma.Repo.Instrumenter],
   migration_lock: nil
 
-scheduled_jobs =
-  with digest_config <- Application.get_env(:pleroma, :email_notifications)[:digest],
-       true <- digest_config[:active] do
-    [{digest_config[:schedule], {Pleroma.Daemons.DigestEmailDaemon, :perform, []}}]
-  else
-    _ -> []
-  end
-
-config :pleroma, Pleroma.Scheduler,
-  global: true,
-  overlap: true,
-  timezone: :utc,
-  jobs: scheduled_jobs
-
 config :pleroma, Pleroma.Captcha,
   enabled: true,
   seconds_valid: 300,
@@ -495,6 +481,12 @@ config :pleroma, Oban,
     scheduled_activities: 10,
     background: 5,
     attachments_cleanup: 5
+  ],
+  crontab: [
+    {"0 0 * * *", Pleroma.Workers.Cron.ClearOauthTokenWorker},
+    {"0 * * * *", Pleroma.Workers.Cron.StatsWorker},
+    {"* * * * *", Pleroma.Workers.Cron.PurgeExpiredActivitiesWorker},
+    {"0 0 * * 0", Pleroma.Workers.Cron.DigestEmailsWorker}
   ]
 
 config :pleroma, :workers,
@@ -578,7 +570,6 @@ config :pleroma, Pleroma.ScheduledActivity,
 config :pleroma, :email_notifications,
   digest: %{
     active: false,
-    schedule: "0 0 * * 0",
     interval: 7,
     inactivity_threshold: 7
   }
@@ -586,8 +577,7 @@ config :pleroma, :email_notifications,
 config :pleroma, :oauth2,
   token_expires_in: 600,
   issue_new_refresh_token: true,
-  clean_expired_tokens: false,
+  clean_expired_tokens: false
-  clean_expired_tokens_interval: 86_400_000
 
 config :pleroma, :database, rum_enabled: false
 
@@ -622,7 +612,6 @@ config :pleroma, :modules, runtime_dir: "instance/modules"
 
 config :pleroma, configurable_from_database: false
 
-config :swarm, node_blacklist: [~r/myhtml_.*$/]
 # Import environment specific config. This must remain at the bottom
 # of this file so it overrides the configuration defined above.
 import_config "#{Mix.env()}.exs"

config/description.exs

@@ -2519,13 +2519,6 @@ config :pleroma, :config_description, [
         key: :clean_expired_tokens,
         type: :boolean,
         description: "Enable a background job to clean expired oauth tokens. Default: `false`."
-      },
-      %{
-        key: :clean_expired_tokens_interval,
-        type: :integer,
-        description:
-          "Interval to run the job to clean expired tokens. Default: 86_400_000 (24 hours).",
-        suggestions: [86_400_000]
       }
     ]
   },

config/test.exs

@@ -68,10 +68,6 @@ config :pleroma, Oban,
   queues: false,
   prune: :disabled
 
-config :pleroma, Pleroma.Scheduler,
-  jobs: [],
-  global: false
-
 config :pleroma, Pleroma.ScheduledActivity,
   daily_user_limit: 2,
   total_user_limit: 3,

docs/administration/backup.md

@@ -1,17 +1,35 @@
-# Backup/Restore your instance
+# Backup/Restore/Move/Remove your instance
 
 ## Backup
 
 1. Stop the Pleroma service.
 2. Go to the working directory of Pleroma (default is `/opt/pleroma`)
-3. Run `sudo -Hu postgres pg_dump -d <pleroma_db> --format=custom -f </path/to/backup_location/pleroma.pgdump>`
+3. Run `sudo -Hu postgres pg_dump -d <pleroma_db> --format=custom -f </path/to/backup_location/pleroma.pgdump>` (make sure the postgres user has write access to the destination file)
 4. Copy `pleroma.pgdump`, `config/prod.secret.exs` and the `uploads` folder to your backup destination. If you have other modifications, copy those changes too.
 5. Restart the Pleroma service.
 
-## Restore
+## Restore/Move
 
-1. Stop the Pleroma service.
+1. Optionally reinstall Pleroma (either on the same server or on another server if you want to move servers). Try to use the same database name.
-2. Go to the working directory of Pleroma (default is `/opt/pleroma`)
+2. Stop the Pleroma service.
-3. Copy the above mentioned files back to their original position.
+3. Go to the working directory of Pleroma (default is `/opt/pleroma`)
-4. Run `sudo -Hu postgres pg_restore -d <pleroma_db> -v -1 </path/to/backup_location/pleroma.pgdump>`
+4. Copy the above mentioned files back to their original position.
-5. Restart the Pleroma service.
+5. Drop the existing database and recreate an empty one `sudo -Hu postgres psql -c 'DROP DATABASE <pleroma_db>;';` `sudo -Hu postgres psql -c 'CREATE DATABASE <pleroma_db>;';`
+6. Run `sudo -Hu postgres pg_restore -d <pleroma_db> -v -1 </path/to/backup_location/pleroma.pgdump>`
+7. If you installed a newer Pleroma version, you should run `mix ecto.migrate`[^1]. This task performs database migrations, if there were any.
+8. Restart the Pleroma service.
+
+[^1]: Prefix with `MIX_ENV=prod` to run it using the production config file.
+
+## Remove
+
+1. Optionally you can remove the users of your instance. This will trigger delete requests for their accounts and posts. Note that this is 'best effort' and doesn't mean that all traces of your instance will be gone from the fediverse.
+    * You can do this from the admin-FE where you can select all local users and delete the accounts using the *Moderate multiple users* dropdown.
+    * You can also list local users and delete them individualy using the CLI tasks for [Managing users](./CLI_tasks/user.md).
+2. Stop the Pleroma service `systemctl stop pleroma`
+3. Disable pleroma from systemd `systemctl disable pleroma`
+4. Remove the files and folders you created during installation (see installation guide). This includes the pleroma, nginx and systemd files and folders.
+5. Reload nginx now that the configuration is removed `systemctl reload nginx`
+6. Remove the database and database user `sudo -Hu postgres psql -c 'DROP DATABASE <pleroma_db>;';` `sudo -Hu postgres psql -c 'DROP USER <pleroma_db>;';`
+7. Remove the system user `userdel pleroma`
+8. Remove the dependencies that you don't need anymore (see installation guide). Make sure you don't remove packages that are still needed for other software that you have running!

docs/configuration/cheatsheet.md

@@ -513,6 +513,7 @@ Configuration options described in [Oban readme](https://github.com/sorentwo/oba
 * `verbose` - logs verbosity
 * `prune` - non-retryable jobs [pruning settings](https://github.com/sorentwo/oban#pruning) (`:disabled` / `{:maxlen, value}` / `{:maxage, value}`)
 * `queues` - job queues (see below)
+* `crontab` - periodic jobs, see [`Oban.Cron`](#obancron)
 
 Pleroma has the following queues:
 
@@ -524,6 +525,12 @@ Pleroma has the following queues:
 * `web_push` - Web push notifications
 * `scheduled_activities` - Scheduled activities, see [`Pleroma.ScheduledActivity`](#pleromascheduledactivity)
 
+#### Oban.Cron
+
+Pleroma has these periodic job workers:
+
+`Pleroma.Workers.Cron.ClearOauthTokenWorker` - a job worker to cleanup expired oauth tokens.
+
 Example:
 
 ```elixir
@@ -534,6 +541,9 @@ config :pleroma, Oban,
   queues: [
     federator_incoming: 50,
     federator_outgoing: 50
+  ],
+  crontab: [
+    {"0 0 * * *", Pleroma.Workers.Cron.ClearOauthTokenWorker}
   ]
 ```
 
@@ -816,8 +826,7 @@ Configure OAuth 2 provider capabilities:
 
 * `token_expires_in` - The lifetime in seconds of the access token.
 * `issue_new_refresh_token` - Keeps old refresh token or generate new refresh token when to obtain an access token.
-* `clean_expired_tokens` - Enable a background job to clean expired oauth tokens. Defaults to `false`.
+* `clean_expired_tokens` - Enable a background job to clean expired oauth tokens. Defaults to `false`. Interval settings sets in configuration periodic jobs [`Oban.Cron`](#obancron)
-* `clean_expired_tokens_interval` - Interval to run the job to clean expired tokens. Defaults to `86_400_000` (24 hours).
 
 ## Link parsing
 

lib/pleroma/application.ex

@@ -42,12 +42,9 @@ defmodule Pleroma.Application do
     children =
       [
         Pleroma.Repo,
-        Pleroma.Scheduler,
         Pleroma.Config.TransferTask,
         Pleroma.Emoji,
         Pleroma.Captcha,
-        Pleroma.Daemons.ScheduledActivityDaemon,
-        Pleroma.Daemons.ActivityExpirationDaemon,
         Pleroma.Plugs.RateLimiter.Supervisor
       ] ++
         cachex_children() ++
@@ -58,7 +55,6 @@ defmodule Pleroma.Application do
           {Oban, Pleroma.Config.get(Oban)}
         ] ++
         task_children(@env) ++
-        oauth_cleanup_child(oauth_cleanup_enabled?()) ++
         streamer_child(@env) ++
         chat_child(@env, chat_enabled?()) ++
         [
@@ -160,20 +156,12 @@ defmodule Pleroma.Application do
 
   defp chat_enabled?, do: Pleroma.Config.get([:chat, :enabled])
 
-  defp oauth_cleanup_enabled?,
-    do: Pleroma.Config.get([:oauth2, :clean_expired_tokens], false)
-
   defp streamer_child(:test), do: []
 
   defp streamer_child(_) do
     [Pleroma.Web.Streamer.supervisor()]
   end
 
-  defp oauth_cleanup_child(true),
-    do: [Pleroma.Web.OAuth.Token.CleanWorker]
-
-  defp oauth_cleanup_child(_), do: []
-
   defp chat_child(_env, true) do
     [Pleroma.Web.ChatChannel.ChatChannelState]
   end

lib/pleroma/daemons/activity_expiration_daemon.ex

@@ -1,66 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Daemons.ActivityExpirationDaemon do
-  alias Pleroma.Activity
-  alias Pleroma.ActivityExpiration
-  alias Pleroma.Config
-  alias Pleroma.Repo
-  alias Pleroma.User
-  alias Pleroma.Web.CommonAPI
-
-  require Logger
-  use GenServer
-  import Ecto.Query
-
-  @schedule_interval :timer.minutes(1)
-
-  def start_link(_) do
-    GenServer.start_link(__MODULE__, nil)
-  end
-
-  @impl true
-  def init(_) do
-    if Config.get([ActivityExpiration, :enabled]) do
-      schedule_next()
-      {:ok, nil}
-    else
-      :ignore
-    end
-  end
-
-  def perform(:execute, expiration_id) do
-    try do
-      expiration =
-        ActivityExpiration
-        |> where([e], e.id == ^expiration_id)
-        |> Repo.one!()
-
-      activity = Activity.get_by_id_with_object(expiration.activity_id)
-      user = User.get_by_ap_id(activity.object.data["actor"])
-      CommonAPI.delete(activity.id, user)
-    rescue
-      error ->
-        Logger.error("#{__MODULE__} Couldn't delete expired activity: #{inspect(error)}")
-    end
-  end
-
-  @impl true
-  def handle_info(:perform, state) do
-    ActivityExpiration.due_expirations(@schedule_interval)
-    |> Enum.each(fn expiration ->
-      Pleroma.Workers.ActivityExpirationWorker.enqueue(
-        "activity_expiration",
-        %{"activity_expiration_id" => expiration.id}
-      )
-    end)
-
-    schedule_next()
-    {:noreply, state}
-  end
-
-  defp schedule_next do
-    Process.send_after(self(), :perform, @schedule_interval)
-  end
-end

lib/pleroma/daemons/digest_email_daemon.ex

@@ -1,42 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Daemons.DigestEmailDaemon do
-  alias Pleroma.Repo
-  alias Pleroma.Workers.DigestEmailsWorker
-
-  import Ecto.Query
-
-  def perform do
-    config = Pleroma.Config.get([:email_notifications, :digest])
-    negative_interval = -Map.fetch!(config, :interval)
-    inactivity_threshold = Map.fetch!(config, :inactivity_threshold)
-    inactive_users_query = Pleroma.User.list_inactive_users_query(inactivity_threshold)
-
-    now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second)
-
-    from(u in inactive_users_query,
-      where: fragment(~s(? ->'digest' @> 'true'), u.email_notifications),
-      where: u.last_digest_emailed_at < datetime_add(^now, ^negative_interval, "day"),
-      select: u
-    )
-    |> Repo.all()
-    |> Enum.each(fn user ->
-      DigestEmailsWorker.enqueue("digest_email", %{"user_id" => user.id})
-    end)
-  end
-
-  @doc """
-  Send digest email to the given user.
-  Updates `last_digest_emailed_at` field for the user and returns the updated user.
-  """
-  @spec perform(Pleroma.User.t()) :: Pleroma.User.t()
-  def perform(user) do
-    with %Swoosh.Email{} = email <- Pleroma.Emails.UserEmail.digest_email(user) do
-      Pleroma.Emails.Mailer.deliver_async(email)
-    end
-
-    Pleroma.User.touch_last_digest_emailed_at(user)
-  end
-end

lib/pleroma/daemons/scheduled_activity_daemon.ex

@@ -1,62 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Daemons.ScheduledActivityDaemon do
-  @moduledoc """
-  Sends scheduled activities to the job queue.
-  """
-
-  alias Pleroma.Config
-  alias Pleroma.ScheduledActivity
-  alias Pleroma.User
-  alias Pleroma.Web.CommonAPI
-
-  use GenServer
-  require Logger
-
-  @schedule_interval :timer.minutes(1)
-
-  def start_link(_) do
-    GenServer.start_link(__MODULE__, nil)
-  end
-
-  def init(_) do
-    if Config.get([ScheduledActivity, :enabled]) do
-      schedule_next()
-      {:ok, nil}
-    else
-      :ignore
-    end
-  end
-
-  def perform(:execute, scheduled_activity_id) do
-    try do
-      {:ok, scheduled_activity} = ScheduledActivity.delete(scheduled_activity_id)
-      %User{} = user = User.get_cached_by_id(scheduled_activity.user_id)
-      {:ok, _result} = CommonAPI.post(user, scheduled_activity.params)
-    rescue
-      error ->
-        Logger.error(
-          "#{__MODULE__} Couldn't create a status from the scheduled activity: #{inspect(error)}"
-        )
-    end
-  end
-
-  def handle_info(:perform, state) do
-    ScheduledActivity.due_activities(@schedule_interval)
-    |> Enum.each(fn scheduled_activity ->
-      Pleroma.Workers.ScheduledActivityWorker.enqueue(
-        "execute",
-        %{"activity_id" => scheduled_activity.id}
-      )
-    end)
-
-    schedule_next()
-    {:noreply, state}
-  end
-
-  defp schedule_next do
-    Process.send_after(self(), :perform, @schedule_interval)
-  end
-end

lib/pleroma/docs/generator.ex

@@ -13,7 +13,7 @@ defmodule Pleroma.Docs.Generator do
       |> Enum.filter(&String.ends_with?(&1, ".ex"))
       |> Enum.map(fn filename ->
         module = filename |> String.trim_trailing(".ex") |> Macro.camelize()
-        String.to_existing_atom(start <> module)
+        String.to_atom(start <> module)
       end)
     end
   end

lib/pleroma/scheduled_activity.ex

@@ -5,15 +5,19 @@
 defmodule Pleroma.ScheduledActivity do
   use Ecto.Schema
 
+  alias Ecto.Multi
   alias Pleroma.Config
   alias Pleroma.Repo
   alias Pleroma.ScheduledActivity
   alias Pleroma.User
   alias Pleroma.Web.CommonAPI.Utils
+  alias Pleroma.Workers.ScheduledActivityWorker
 
   import Ecto.Query
   import Ecto.Changeset
 
+  @type t :: %__MODULE__{}
+
   @min_offset :timer.minutes(5)
 
   schema "scheduled_activities" do
@@ -105,16 +109,32 @@ defmodule Pleroma.ScheduledActivity do
   end
 
   def new(%User{} = user, attrs) do
-    %ScheduledActivity{user_id: user.id}
+    changeset(%ScheduledActivity{user_id: user.id}, attrs)
-    |> changeset(attrs)
   end
 
+  @doc """
+  Creates ScheduledActivity and add to queue to perform at scheduled_at date
+  """
+  @spec create(User.t(), map()) :: {:ok, ScheduledActivity.t()} | {:error, Ecto.Changeset.t()}
   def create(%User{} = user, attrs) do
-    user
+    Multi.new()
-    |> new(attrs)
+    |> Multi.insert(:scheduled_activity, new(user, attrs))
-    |> Repo.insert()
+    |> maybe_add_jobs(Config.get([ScheduledActivity, :enabled]))
+    |> Repo.transaction()
+    |> transaction_response
   end
 
+  defp maybe_add_jobs(multi, true) do
+    multi
+    |> Multi.run(:scheduled_activity_job, fn _repo, %{scheduled_activity: activity} ->
+      %{activity_id: activity.id}
+      |> ScheduledActivityWorker.new(scheduled_at: activity.scheduled_at)
+      |> Oban.insert()
+    end)
+  end
+
+  defp maybe_add_jobs(multi, _), do: multi
+
   def get(%User{} = user, scheduled_activity_id) do
     ScheduledActivity
     |> where(user_id: ^user.id)
@@ -122,25 +142,43 @@ defmodule Pleroma.ScheduledActivity do
     |> Repo.one()
   end
 
-  def update(%ScheduledActivity{} = scheduled_activity, attrs) do
+  @spec update(ScheduledActivity.t(), map()) ::
-    scheduled_activity
+          {:ok, ScheduledActivity.t()} | {:error, Ecto.Changeset.t()}
-    |> update_changeset(attrs)
+  def update(%ScheduledActivity{id: id} = scheduled_activity, attrs) do
-    |> Repo.update()
+    with {:error, %Ecto.Changeset{valid?: true} = changeset} <-
+           {:error, update_changeset(scheduled_activity, attrs)} do
+      Multi.new()
+      |> Multi.update(:scheduled_activity, changeset)
+      |> Multi.update_all(:scheduled_job, job_query(id),
+        set: [scheduled_at: get_field(changeset, :scheduled_at)]
+      )
+      |> Repo.transaction()
+      |> transaction_response
+    end
   end
 
-  def delete(%ScheduledActivity{} = scheduled_activity) do
+  @doc "Deletes a ScheduledActivity and linked jobs."
-    scheduled_activity
+  @spec delete(ScheduledActivity.t() | binary() | integer) ::
-    |> Repo.delete()
+          {:ok, ScheduledActivity.t()} | {:error, Ecto.Changeset.t()}
+  def delete(%ScheduledActivity{id: id} = scheduled_activity) do
+    Multi.new()
+    |> Multi.delete(:scheduled_activity, scheduled_activity, stale_error_field: :id)
+    |> Multi.delete_all(:jobs, job_query(id))
+    |> Repo.transaction()
+    |> transaction_response
   end
 
   def delete(id) when is_binary(id) or is_integer(id) do
-    ScheduledActivity
+    delete(%__MODULE__{id: id})
-    |> where(id: ^id)
+  end
-    |> select([sa], sa)
+
-    |> Repo.delete_all()
+  defp transaction_response(result) do
-    |> case do
+    case result do
-      {1, [scheduled_activity]} -> {:ok, scheduled_activity}
+      {:ok, %{scheduled_activity: scheduled_activity}} ->
-      _ -> :error
+        {:ok, scheduled_activity}
+
+      {:error, _, changeset, _} ->
+        {:error, changeset}
     end
   end
 
@@ -158,4 +196,11 @@ defmodule Pleroma.ScheduledActivity do
     |> where([sa], sa.scheduled_at < ^naive_datetime)
     |> Repo.all()
   end
+
+  def job_query(scheduled_activity_id) do
+    from(j in Oban.Job,
+      where: j.queue == "scheduled_activities",
+      where: fragment("args ->> 'activity_id' = ?::text", ^to_string(scheduled_activity_id))
+    )
+  end
 end

lib/pleroma/scheduler.ex

@@ -1,7 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Scheduler do
-  use Quantum.Scheduler, otp_app: :pleroma
-end

lib/pleroma/stats.ex

@@ -9,22 +9,43 @@ defmodule Pleroma.Stats do
 
   use GenServer
 
-  @interval 1000 * 60 * 60
+  @init_state %{
+    peers: [],
+    stats: %{
+      domain_count: 0,
+      status_count: 0,
+      user_count: 0
+    }
+  }
 
   def start_link(_) do
-    GenServer.start_link(__MODULE__, initial_data(), name: __MODULE__)
+    GenServer.start_link(
+      __MODULE__,
+      @init_state,
+      name: __MODULE__
+    )
   end
 
+  @doc "Performs update stats"
   def force_update do
     GenServer.call(__MODULE__, :force_update)
   end
 
+  @doc "Performs collect stats"
+  def do_collect do
+    GenServer.cast(__MODULE__, :run_update)
+  end
+
+  @doc "Returns stats data"
+  @spec get_stats() :: %{domain_count: integer(), status_count: integer(), user_count: integer()}
   def get_stats do
     %{stats: stats} = GenServer.call(__MODULE__, :get_state)
 
     stats
   end
 
+  @doc "Returns list peers"
+  @spec get_peers() :: list(String.t())
   def get_peers do
     %{peers: peers} = GenServer.call(__MODULE__, :get_state)
 
@@ -32,7 +53,6 @@ defmodule Pleroma.Stats do
   end
 
   def init(args) do
-    Process.send(self(), :run_update, [])
     {:ok, args}
   end
 
@@ -45,17 +65,12 @@ defmodule Pleroma.Stats do
     {:reply, state, state}
   end
 
-  def handle_info(:run_update, _state) do
+  def handle_cast(:run_update, _state) do
     new_stats = get_stat_data()
 
-    Process.send_after(self(), :run_update, @interval)
     {:noreply, new_stats}
   end
 
-  defp initial_data do
-    %{peers: [], stats: %{}}
-  end
-
   defp get_stat_data do
     peers =
       from(
@@ -74,7 +89,11 @@ defmodule Pleroma.Stats do
 
     %{
       peers: peers,
-      stats: %{domain_count: domain_count, status_count: status_count, user_count: user_count}
+      stats: %{
+        domain_count: domain_count,
+        status_count: status_count,
+        user_count: user_count
+      }
     }
   end
 end

lib/pleroma/web/mastodon_api/controllers/status_controller.ex

@@ -124,15 +124,18 @@ defmodule Pleroma.Web.MastodonAPI.StatusController do
       ) do
     params = Map.put(params, "in_reply_to_status_id", params["in_reply_to_id"])
 
-    if ScheduledActivity.far_enough?(scheduled_at) do
+    with {:far_enough, true} <- {:far_enough, ScheduledActivity.far_enough?(scheduled_at)},
-      with {:ok, scheduled_activity} <-
+         attrs <- %{"params" => params, "scheduled_at" => scheduled_at},
-             ScheduledActivity.create(user, %{"params" => params, "scheduled_at" => scheduled_at}) do
+         {:ok, scheduled_activity} <- ScheduledActivity.create(user, attrs) do
-        conn
+      conn
-        |> put_view(ScheduledActivityView)
+      |> put_view(ScheduledActivityView)
-        |> render("show.json", scheduled_activity: scheduled_activity)
+      |> render("show.json", scheduled_activity: scheduled_activity)
-      end
     else
-      create(conn, Map.drop(params, ["scheduled_at"]))
+      {:far_enough, _} ->
+        create(conn, Map.drop(params, ["scheduled_at"]))
+
+      error ->
+        error
     end
   end
 

lib/pleroma/web/oauth/token/clean_worker.ex

@@ -1,34 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Web.OAuth.Token.CleanWorker do
-  @moduledoc """
-  The module represents functions to clean an expired oauth tokens.
-  """
-  use GenServer
-
-  @ten_seconds 10_000
-  @one_day 86_400_000
-
-  alias Pleroma.Web.OAuth.Token
-  alias Pleroma.Workers.BackgroundWorker
-
-  def start_link(_), do: GenServer.start_link(__MODULE__, %{})
-
-  def init(_) do
-    Process.send_after(self(), :perform, @ten_seconds)
-    {:ok, nil}
-  end
-
-  @doc false
-  def handle_info(:perform, state) do
-    BackgroundWorker.enqueue("clean_expired_tokens", %{})
-    interval = Pleroma.Config.get([:oauth2, :clean_expired_tokens_interval], @one_day)
-
-    Process.send_after(self(), :perform, interval)
-    {:noreply, state}
-  end
-
-  def perform(:clean), do: Token.delete_expired_tokens()
-end

lib/pleroma/workers/activity_expiration_worker.ex

@@ -1,18 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Workers.ActivityExpirationWorker do
-  use Pleroma.Workers.WorkerHelper, queue: "activity_expiration"
-
-  @impl Oban.Worker
-  def perform(
-        %{
-          "op" => "activity_expiration",
-          "activity_expiration_id" => activity_expiration_id
-        },
-        _job
-      ) do
-    Pleroma.Daemons.ActivityExpirationDaemon.perform(:execute, activity_expiration_id)
-  end
-end

lib/pleroma/workers/background_worker.ex

@@ -6,7 +6,6 @@ defmodule Pleroma.Workers.BackgroundWorker do
   alias Pleroma.Activity
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy
-  alias Pleroma.Web.OAuth.Token.CleanWorker
 
   use Pleroma.Workers.WorkerHelper, queue: "background"
 
@@ -55,10 +54,6 @@ defmodule Pleroma.Workers.BackgroundWorker do
     User.perform(:follow_import, follower, followed_identifiers)
   end
 
-  def perform(%{"op" => "clean_expired_tokens"}, _job) do
-    CleanWorker.perform(:clean)
-  end
-
   def perform(%{"op" => "media_proxy_preload", "message" => message}, _job) do
     MediaProxyWarmingPolicy.perform(:preload, message)
   end

lib/pleroma/workers/cron/clear_oauth_token_worker.ex

@@ -0,0 +1,21 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Cron.ClearOauthTokenWorker do
+  @moduledoc """
+  The worker to cleanup expired oAuth tokens.
+  """
+
+  use Oban.Worker, queue: "background"
+
+  alias Pleroma.Config
+  alias Pleroma.Web.OAuth.Token
+
+  @impl Oban.Worker
+  def perform(_opts, _job) do
+    if Config.get([:oauth2, :clean_expired_tokens], false) do
+      Token.delete_expired_tokens()
+    end
+  end
+end

lib/pleroma/workers/cron/digest_emails_worker.ex

@@ -0,0 +1,58 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Cron.DigestEmailsWorker do
+  @moduledoc """
+  The worker to send digest emails.
+  """
+
+  use Oban.Worker, queue: "digest_emails"
+
+  alias Pleroma.Config
+  alias Pleroma.Emails
+  alias Pleroma.Repo
+  alias Pleroma.User
+
+  import Ecto.Query
+
+  require Logger
+
+  @impl Oban.Worker
+  def perform(_opts, _job) do
+    config = Config.get([:email_notifications, :digest])
+
+    if config[:active] do
+      negative_interval = -Map.fetch!(config, :interval)
+      inactivity_threshold = Map.fetch!(config, :inactivity_threshold)
+      inactive_users_query = User.list_inactive_users_query(inactivity_threshold)
+
+      now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second)
+
+      from(u in inactive_users_query,
+        where: fragment(~s(? ->'digest' @> 'true'), u.email_notifications),
+        where: u.last_digest_emailed_at < datetime_add(^now, ^negative_interval, "day"),
+        select: u
+      )
+      |> Repo.all()
+      |> send_emails
+    end
+  end
+
+  def send_emails(users) do
+    Enum.each(users, &send_email/1)
+  end
+
+  @doc """
+  Send digest email to the given user.
+  Updates `last_digest_emailed_at` field for the user and returns the updated user.
+  """
+  @spec send_email(User.t()) :: User.t()
+  def send_email(user) do
+    with %Swoosh.Email{} = email <- Emails.UserEmail.digest_email(user) do
+      Emails.Mailer.deliver_async(email)
+    end
+
+    User.touch_last_digest_emailed_at(user)
+  end
+end

lib/pleroma/workers/cron/purge_expired_activities_worker.ex

@@ -0,0 +1,46 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Cron.PurgeExpiredActivitiesWorker do
+  @moduledoc """
+  The worker to purge expired activities.
+  """
+
+  use Oban.Worker, queue: "activity_expiration"
+
+  alias Pleroma.Activity
+  alias Pleroma.ActivityExpiration
+  alias Pleroma.Config
+  alias Pleroma.User
+  alias Pleroma.Web.CommonAPI
+
+  require Logger
+
+  @interval :timer.minutes(1)
+
+  @impl Oban.Worker
+  def perform(_opts, _job) do
+    if Config.get([ActivityExpiration, :enabled]) do
+      Enum.each(ActivityExpiration.due_expirations(@interval), &delete_activity/1)
+    end
+  end
+
+  def delete_activity(%ActivityExpiration{activity_id: activity_id}) do
+    with {:activity, %Activity{} = activity} <-
+           {:activity, Activity.get_by_id_with_object(activity_id)},
+         {:user, %User{} = user} <- {:user, User.get_by_ap_id(activity.object.data["actor"])} do
+      CommonAPI.delete(activity.id, user)
+    else
+      {:activity, _} ->
+        Logger.error(
+          "#{__MODULE__} Couldn't delete expired activity: not found activity ##{activity_id}"
+        )
+
+      {:user, _} ->
+        Logger.error(
+          "#{__MODULE__} Couldn't delete expired activity: not found actorof ##{activity_id}"
+        )
+    end
+  end
+end

lib/pleroma/workers/cron/stats_worker.ex

@@ -0,0 +1,16 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Cron.StatsWorker do
+  @moduledoc """
+  The worker to update peers statistics.
+  """
+
+  use Oban.Worker, queue: "background"
+
+  @impl Oban.Worker
+  def perform(_opts, _job) do
+    Pleroma.Stats.do_collect()
+  end
+end

lib/pleroma/workers/digest_emails_worker.ex

@@ -1,16 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Workers.DigestEmailsWorker do
-  alias Pleroma.User
-
-  use Pleroma.Workers.WorkerHelper, queue: "digest_emails"
-
-  @impl Oban.Worker
-  def perform(%{"op" => "digest_email", "user_id" => user_id}, _job) do
-    user_id
-    |> User.get_cached_by_id()
-    |> Pleroma.Daemons.DigestEmailDaemon.perform()
-  end
-end

lib/pleroma/workers/scheduled_activity_worker.ex

@@ -3,10 +3,42 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.ScheduledActivityWorker do
+  @moduledoc """
+  The worker to post scheduled activity.
+  """
+
   use Pleroma.Workers.WorkerHelper, queue: "scheduled_activities"
 
+  alias Pleroma.Config
+  alias Pleroma.ScheduledActivity
+  alias Pleroma.User
+  alias Pleroma.Web.CommonAPI
+
+  require Logger
+
   @impl Oban.Worker
-  def perform(%{"op" => "execute", "activity_id" => activity_id}, _job) do
+  def perform(%{"activity_id" => activity_id}, _job) do
-    Pleroma.Daemons.ScheduledActivityDaemon.perform(:execute, activity_id)
+    if Config.get([ScheduledActivity, :enabled]) do
+      case Pleroma.Repo.get(ScheduledActivity, activity_id) do
+        %ScheduledActivity{} = scheduled_activity ->
+          post_activity(scheduled_activity)
+
+        _ ->
+          Logger.error("#{__MODULE__} Couldn't find scheduled activity: #{activity_id}")
+      end
+    end
+  end
+
+  defp post_activity(%ScheduledActivity{user_id: user_id, params: params} = scheduled_activity) do
+    with {:delete, {:ok, _}} <- {:delete, ScheduledActivity.delete(scheduled_activity)},
+         {:user, %User{} = user} <- {:user, User.get_cached_by_id(user_id)},
+         {:post, {:ok, _}} <- {:post, CommonAPI.post(user, params)} do
+      :ok
+    else
+      error ->
+        Logger.error(
+          "#{__MODULE__} Couldn't create a status from the scheduled activity: #{inspect(error)}"
+        )
+    end
   end
 end

mix.exs

@@ -63,7 +63,7 @@ defmodule Pleroma.Mixfile do
   def application do
     [
       mod: {Pleroma.Application, []},
-      extra_applications: [:logger, :runtime_tools, :comeonin, :quack, :fast_sanitize, :swarm],
+      extra_applications: [:logger, :runtime_tools, :comeonin, :quack, :fast_sanitize],
       included_applications: [:ex_syslogger]
     ]
   end
@@ -108,8 +108,7 @@ defmodule Pleroma.Mixfile do
       {:ecto_enum, "~> 1.4"},
       {:ecto_sql, "~> 3.3.2"},
       {:postgrex, ">= 0.13.5"},
-      {:oban, "~> 0.12.0"},
+      {:oban, "~> 0.12.1"},
-      {:quantum, "~> 2.3"},
       {:gettext, "~> 0.15"},
       {:comeonin, "~> 4.1.1"},
       {:pbkdf2_elixir, "~> 0.12.3"},
@@ -163,7 +162,7 @@ defmodule Pleroma.Mixfile do
       {:esshd, "~> 0.1.0", runtime: Application.get_env(:esshd, :enabled, false)},
       {:ex_const, "~> 0.2"},
       {:plug_static_index_html, "~> 1.0.0"},
-      {:excoveralls, "~> 0.11.1", only: :test},
+      {:excoveralls, "~> 0.12.1", only: :test},
       {:flake_id, "~> 0.1.0"},
       {:remote_ip,
        git: "https://git.pleroma.social/pleroma/remote_ip.git",

mix.lock

@@ -36,7 +36,7 @@
   "ex_doc": {:hex, :ex_doc, "0.21.2", "caca5bc28ed7b3bdc0b662f8afe2bee1eedb5c3cf7b322feeeb7c6ebbde089d6", [:mix], [{:earmark, "~> 1.3.3 or ~> 1.4", [hex: :earmark, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}], "hexpm", "f1155337ae17ff7a1255217b4c1ceefcd1860b7ceb1a1874031e7a861b052e39"},
   "ex_machina": {:hex, :ex_machina, "2.3.0", "92a5ad0a8b10ea6314b876a99c8c9e3f25f4dde71a2a835845b136b9adaf199a", [:mix], [{:ecto, "~> 2.2 or ~> 3.0", [hex: :ecto, repo: "hexpm", optional: true]}, {:ecto_sql, "~> 3.0", [hex: :ecto_sql, repo: "hexpm", optional: true]}], "hexpm", "b84f6af156264530b312a8ab98ac6088f6b77ae5fe2058305c81434aa01fbaf9"},
   "ex_syslogger": {:git, "https://github.com/slashmili/ex_syslogger.git", "f3963399047af17e038897c69e20d552e6899e1d", [tag: "1.4.0"]},
-  "excoveralls": {:hex, :excoveralls, "0.11.2", "0c6f2c8db7683b0caa9d490fb8125709c54580b4255ffa7ad35f3264b075a643", [:mix], [{:hackney, "~> 1.0", [hex: :hackney, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "e11a4490976aabeed3eb9dc70ec94a4f2d11fed5c9d4b5dc5d89bfa0a215abb5"},
+  "excoveralls": {:hex, :excoveralls, "0.12.2", "a513defac45c59e310ac42fcf2b8ae96f1f85746410f30b1ff2b710a4b6cd44b", [:mix], [{:hackney, "~> 1.0", [hex: :hackney, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "151c476331d49b45601ffc45f43cb3a8beb396b02a34e3777fea0ad34ae57d89"},
   "fast_html": {:hex, :fast_html, "1.0.1", "5bc7df4dc4607ec2c314c16414e4111d79a209956c4f5df96602d194c61197f9", [:make, :mix], [], "hexpm", "18e627dd62051a375ef94b197f41e8027c3e8eef0180ab8f81e0543b3dc6900a"},
   "fast_sanitize": {:hex, :fast_sanitize, "0.1.6", "60a5ae96879956dea409a91a77f5dd2994c24cc10f80eefd8f9892ee4c0c7b25", [:mix], [{:fast_html, "~> 1.0", [hex: :fast_html, repo: "hexpm", optional: false]}, {:plug, "~> 1.8", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "b73f50f0cb522dd0331ea8e8c90b408de42c50f37641219d6364f0e3e7efd22c"},
   "flake_id": {:hex, :flake_id, "0.1.0", "7716b086d2e405d09b647121a166498a0d93d1a623bead243e1f74216079ccb3", [:mix], [{:base62, "~> 1.2", [hex: :base62, repo: "hexpm", optional: false]}, {:ecto, ">= 2.0.0", [hex: :ecto, repo: "hexpm", optional: true]}], "hexpm", "31fc8090fde1acd267c07c36ea7365b8604055f897d3a53dd967658c691bd827"},

test/activity_expiration_test.exs

@@ -7,6 +7,8 @@ defmodule Pleroma.ActivityExpirationTest do
   alias Pleroma.ActivityExpiration
   import Pleroma.Factory
 
+  clear_config([ActivityExpiration, :enabled])
+
   test "finds activities due to be deleted only" do
     activity = insert(:note_activity)
     expiration_due = insert(:expiration_in_the_past, %{activity_id: activity.id})
@@ -24,4 +26,27 @@ defmodule Pleroma.ActivityExpirationTest do
     now = NaiveDateTime.utc_now()
     assert {:error, _} = ActivityExpiration.create(activity, now)
   end
+
+  test "deletes an expiration activity" do
+    Pleroma.Config.put([ActivityExpiration, :enabled], true)
+    activity = insert(:note_activity)
+
+    naive_datetime =
+      NaiveDateTime.add(
+        NaiveDateTime.utc_now(),
+        -:timer.minutes(2),
+        :millisecond
+      )
+
+    expiration =
+      insert(
+        :expiration_in_the_past,
+        %{activity_id: activity.id, scheduled_at: naive_datetime}
+      )
+
+    Pleroma.Workers.Cron.PurgeExpiredActivitiesWorker.perform(:ops, :pid)
+
+    refute Pleroma.Repo.get(Pleroma.Activity, activity.id)
+    refute Pleroma.Repo.get(Pleroma.ActivityExpiration, expiration.id)
+  end
 end

test/daemons/activity_expiration_daemon_test.exs

@@ -1,17 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.ActivityExpirationWorkerTest do
-  use Pleroma.DataCase
-  alias Pleroma.Activity
-  import Pleroma.Factory
-
-  test "deletes an activity" do
-    activity = insert(:note_activity)
-    expiration = insert(:expiration_in_the_past, %{activity_id: activity.id})
-    Pleroma.Daemons.ActivityExpirationDaemon.perform(:execute, expiration.id)
-
-    refute Repo.get(Activity, activity.id)
-  end
-end

test/daemons/scheduled_activity_daemon_test.exs

@@ -1,19 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.ScheduledActivityDaemonTest do
-  use Pleroma.DataCase
-  alias Pleroma.ScheduledActivity
-  import Pleroma.Factory
-
-  test "creates a status from the scheduled activity" do
-    user = insert(:user)
-    scheduled_activity = insert(:scheduled_activity, user: user, params: %{status: "hi"})
-    Pleroma.Daemons.ScheduledActivityDaemon.perform(:execute, scheduled_activity.id)
-
-    refute Repo.get(ScheduledActivity, scheduled_activity.id)
-    activity = Repo.all(Pleroma.Activity) |> Enum.find(&(&1.actor == user.ap_id))
-    assert Pleroma.Object.normalize(activity).data["content"] == "hi"
-  end
-end

test/scheduled_activity_test.exs

@@ -8,11 +8,51 @@ defmodule Pleroma.ScheduledActivityTest do
   alias Pleroma.ScheduledActivity
   import Pleroma.Factory
 
+  clear_config([ScheduledActivity, :enabled])
+
   setup context do
     DataCase.ensure_local_uploader(context)
   end
 
   describe "creation" do
+    test "scheduled activities with jobs when ScheduledActivity enabled" do
+      Pleroma.Config.put([ScheduledActivity, :enabled], true)
+      user = insert(:user)
+
+      today =
+        NaiveDateTime.utc_now()
+        |> NaiveDateTime.add(:timer.minutes(6), :millisecond)
+        |> NaiveDateTime.to_iso8601()
+
+      attrs = %{params: %{}, scheduled_at: today}
+      {:ok, sa1} = ScheduledActivity.create(user, attrs)
+      {:ok, sa2} = ScheduledActivity.create(user, attrs)
+
+      jobs =
+        Repo.all(from(j in Oban.Job, where: j.queue == "scheduled_activities", select: j.args))
+
+      assert jobs == [%{"activity_id" => sa1.id}, %{"activity_id" => sa2.id}]
+    end
+
+    test "scheduled activities without jobs when ScheduledActivity disabled" do
+      Pleroma.Config.put([ScheduledActivity, :enabled], false)
+      user = insert(:user)
+
+      today =
+        NaiveDateTime.utc_now()
+        |> NaiveDateTime.add(:timer.minutes(6), :millisecond)
+        |> NaiveDateTime.to_iso8601()
+
+      attrs = %{params: %{}, scheduled_at: today}
+      {:ok, _sa1} = ScheduledActivity.create(user, attrs)
+      {:ok, _sa2} = ScheduledActivity.create(user, attrs)
+
+      jobs =
+        Repo.all(from(j in Oban.Job, where: j.queue == "scheduled_activities", select: j.args))
+
+      assert jobs == []
+    end
+
     test "when daily user limit is exceeded" do
       user = insert(:user)
 
@@ -24,6 +64,7 @@ defmodule Pleroma.ScheduledActivityTest do
       attrs = %{params: %{}, scheduled_at: today}
       {:ok, _} = ScheduledActivity.create(user, attrs)
       {:ok, _} = ScheduledActivity.create(user, attrs)
+
       {:error, changeset} = ScheduledActivity.create(user, attrs)
       assert changeset.errors == [scheduled_at: {"daily limit exceeded", []}]
     end

test/support/helpers.ex

@@ -54,6 +54,12 @@ defmodule Pleroma.Tests.Helpers do
           clear_config_all: 2
         ]
 
+      def to_datetime(naive_datetime) do
+        naive_datetime
+        |> DateTime.from_naive!("Etc/UTC")
+        |> DateTime.truncate(:second)
+      end
+
       def collect_ids(collection) do
         collection
         |> Enum.map(& &1.id)

test/web/mastodon_api/controllers/scheduled_activity_controller_test.exs

@@ -9,6 +9,9 @@ defmodule Pleroma.Web.MastodonAPI.ScheduledActivityControllerTest do
   alias Pleroma.ScheduledActivity
 
   import Pleroma.Factory
+  import Ecto.Query
+
+  clear_config([ScheduledActivity, :enabled])
 
   test "shows scheduled activities" do
     %{user: user, conn: conn} = oauth_access(["read:statuses"])
@@ -52,11 +55,26 @@ defmodule Pleroma.Web.MastodonAPI.ScheduledActivityControllerTest do
   end
 
   test "updates a scheduled activity" do
+    Pleroma.Config.put([ScheduledActivity, :enabled], true)
     %{user: user, conn: conn} = oauth_access(["write:statuses"])
-    scheduled_activity = insert(:scheduled_activity, user: user)
 
-    new_scheduled_at =
+    scheduled_at = Timex.shift(NaiveDateTime.utc_now(), minutes: 60)
-      NaiveDateTime.add(NaiveDateTime.utc_now(), :timer.minutes(120), :millisecond)
+
+    {:ok, scheduled_activity} =
+      ScheduledActivity.create(
+        user,
+        %{
+          scheduled_at: scheduled_at,
+          params: build(:note).data
+        }
+      )
+
+    job = Repo.one(from(j in Oban.Job, where: j.queue == "scheduled_activities"))
+
+    assert job.args == %{"activity_id" => scheduled_activity.id}
+    assert DateTime.truncate(job.scheduled_at, :second) == to_datetime(scheduled_at)
+
+    new_scheduled_at = Timex.shift(NaiveDateTime.utc_now(), minutes: 120)
 
     res_conn =
       put(conn, "/api/v1/scheduled_statuses/#{scheduled_activity.id}", %{
@@ -65,6 +83,9 @@ defmodule Pleroma.Web.MastodonAPI.ScheduledActivityControllerTest do
 
     assert %{"scheduled_at" => expected_scheduled_at} = json_response(res_conn, 200)
     assert expected_scheduled_at == Pleroma.Web.CommonAPI.Utils.to_masto_date(new_scheduled_at)
+    job = refresh_record(job)
+
+    assert DateTime.truncate(job.scheduled_at, :second) == to_datetime(new_scheduled_at)
 
     res_conn = put(conn, "/api/v1/scheduled_statuses/404", %{scheduled_at: new_scheduled_at})
 
@@ -72,8 +93,22 @@ defmodule Pleroma.Web.MastodonAPI.ScheduledActivityControllerTest do
   end
 
   test "deletes a scheduled activity" do
+    Pleroma.Config.put([ScheduledActivity, :enabled], true)
     %{user: user, conn: conn} = oauth_access(["write:statuses"])
-    scheduled_activity = insert(:scheduled_activity, user: user)
+    scheduled_at = Timex.shift(NaiveDateTime.utc_now(), minutes: 60)
+
+    {:ok, scheduled_activity} =
+      ScheduledActivity.create(
+        user,
+        %{
+          scheduled_at: scheduled_at,
+          params: build(:note).data
+        }
+      )
+
+    job = Repo.one(from(j in Oban.Job, where: j.queue == "scheduled_activities"))
+
+    assert job.args == %{"activity_id" => scheduled_activity.id}
 
     res_conn =
       conn
@@ -81,7 +116,8 @@ defmodule Pleroma.Web.MastodonAPI.ScheduledActivityControllerTest do
       |> delete("/api/v1/scheduled_statuses/#{scheduled_activity.id}")
 
     assert %{} = json_response(res_conn, 200)
-    assert nil == Repo.get(ScheduledActivity, scheduled_activity.id)
+    refute Repo.get(ScheduledActivity, scheduled_activity.id)
+    refute Repo.get(Oban.Job, job.id)
 
     res_conn =
       conn

test/web/node_info_test.exs

@@ -6,6 +6,7 @@ defmodule Pleroma.Web.NodeInfoTest do
   use Pleroma.Web.ConnCase
 
   import Pleroma.Factory
+  clear_config([:mrf_simple])
 
   test "GET /.well-known/nodeinfo", %{conn: conn} do
     links =

test/workers/cron/clear_oauth_token_worker_test.exs

@@ -0,0 +1,22 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Cron.ClearOauthTokenWorkerTest do
+  use Pleroma.DataCase
+
+  import Pleroma.Factory
+  alias Pleroma.Workers.Cron.ClearOauthTokenWorker
+
+  clear_config([:oauth2, :clean_expired_tokens])
+
+  test "deletes expired tokens" do
+    insert(:oauth_token,
+      valid_until: NaiveDateTime.add(NaiveDateTime.utc_now(), -60 * 10)
+    )
+
+    Pleroma.Config.put([:oauth2, :clean_expired_tokens], true)
+    ClearOauthTokenWorker.perform(:opts, :job)
+    assert Pleroma.Repo.all(Pleroma.Web.OAuth.Token) == []
+  end
+end

test/workers/cron/digest_emails_worker_test.exs

@@ -2,16 +2,24 @@
 # Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
 # SPDX-License-Identifier: AGPL-3.0-only
 
-defmodule Pleroma.DigestEmailDaemonTest do
+defmodule Pleroma.Workers.Cron.DigestEmailsWorkerTest do
   use Pleroma.DataCase
+
   import Pleroma.Factory
 
-  alias Pleroma.Daemons.DigestEmailDaemon
   alias Pleroma.Tests.ObanHelpers
   alias Pleroma.User
   alias Pleroma.Web.CommonAPI
 
+  clear_config([:email_notifications, :digest])
+
   test "it sends digest emails" do
+    Pleroma.Config.put([:email_notifications, :digest], %{
+      active: true,
+      inactivity_threshold: 7,
+      interval: 7
+    })
+
     user = insert(:user)
 
     date =
@@ -23,8 +31,7 @@ defmodule Pleroma.DigestEmailDaemonTest do
     {:ok, _} = User.switch_email_notifications(user2, "digest", true)
     CommonAPI.post(user, %{"status" => "hey @#{user2.nickname}!"})
 
-    DigestEmailDaemon.perform()
+    Pleroma.Workers.Cron.DigestEmailsWorker.perform(:opts, :pid)
-    ObanHelpers.perform_all()
     # Performing job(s) enqueued at previous step
     ObanHelpers.perform_all()
 

test/workers/cron/purge_expired_activities_worker_test.exs

@@ -0,0 +1,56 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Cron.PurgeExpiredActivitiesWorkerTest do
+  use Pleroma.DataCase
+
+  alias Pleroma.ActivityExpiration
+  alias Pleroma.Workers.Cron.PurgeExpiredActivitiesWorker
+
+  import Pleroma.Factory
+  import ExUnit.CaptureLog
+
+  clear_config([ActivityExpiration, :enabled])
+
+  test "deletes an expiration activity" do
+    Pleroma.Config.put([ActivityExpiration, :enabled], true)
+    activity = insert(:note_activity)
+
+    naive_datetime =
+      NaiveDateTime.add(
+        NaiveDateTime.utc_now(),
+        -:timer.minutes(2),
+        :millisecond
+      )
+
+    expiration =
+      insert(
+        :expiration_in_the_past,
+        %{activity_id: activity.id, scheduled_at: naive_datetime}
+      )
+
+    Pleroma.Workers.Cron.PurgeExpiredActivitiesWorker.perform(:ops, :pid)
+
+    refute Pleroma.Repo.get(Pleroma.Activity, activity.id)
+    refute Pleroma.Repo.get(Pleroma.ActivityExpiration, expiration.id)
+  end
+
+  describe "delete_activity/1" do
+    test "adds log message if activity isn't find" do
+      assert capture_log([level: :error], fn ->
+               PurgeExpiredActivitiesWorker.delete_activity(%ActivityExpiration{
+                 activity_id: "test-activity"
+               })
+             end) =~ "Couldn't delete expired activity: not found activity"
+    end
+
+    test "adds log message if actor isn't find" do
+      assert capture_log([level: :error], fn ->
+               PurgeExpiredActivitiesWorker.delete_activity(%ActivityExpiration{
+                 activity_id: "test-activity"
+               })
+             end) =~ "Couldn't delete expired activity: not found activity"
+    end
+  end
+end

test/workers/scheduled_activity_worker_test.exs

@@ -0,0 +1,52 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.ScheduledActivityWorkerTest do
+  use Pleroma.DataCase
+
+  alias Pleroma.ScheduledActivity
+  alias Pleroma.Workers.ScheduledActivityWorker
+
+  import Pleroma.Factory
+  import ExUnit.CaptureLog
+
+  clear_config([ScheduledActivity, :enabled])
+
+  test "creates a status from the scheduled activity" do
+    Pleroma.Config.put([ScheduledActivity, :enabled], true)
+    user = insert(:user)
+
+    naive_datetime =
+      NaiveDateTime.add(
+        NaiveDateTime.utc_now(),
+        -:timer.minutes(2),
+        :millisecond
+      )
+
+    scheduled_activity =
+      insert(
+        :scheduled_activity,
+        scheduled_at: naive_datetime,
+        user: user,
+        params: %{status: "hi"}
+      )
+
+    ScheduledActivityWorker.perform(
+      %{"activity_id" => scheduled_activity.id},
+      :pid
+    )
+
+    refute Repo.get(ScheduledActivity, scheduled_activity.id)
+    activity = Repo.all(Pleroma.Activity) |> Enum.find(&(&1.actor == user.ap_id))
+    assert Pleroma.Object.normalize(activity).data["content"] == "hi"
+  end
+
+  test "adds log message if ScheduledActivity isn't find" do
+    Pleroma.Config.put([ScheduledActivity, :enabled], true)
+
+    assert capture_log([level: :error], fn ->
+             ScheduledActivityWorker.perform(%{"activity_id" => 42}, :pid)
+           end) =~ "Couldn't find scheduled activity"
+  end
+end