From 19ec13e64cd801287443205d9d2d4095a5677099 Mon Sep 17 00:00:00 2001 From: Kieran Date: Sun, 10 Mar 2024 15:17:15 -0700 Subject: [PATCH] Local data worker improvements (#66) * Adds new worker for backfilling data * Adds backfill job to startup tasks --- lib/pinchflat/application.ex | 4 +- lib/pinchflat/startup_tasks.ex | 15 +++- lib/pinchflat/workers/data_backfill_worker.ex | 75 +++++++++++++++++++ lib/pinchflat/workers/fast_indexing_worker.ex | 6 +- .../workers/data_backfill_worker_test.exs | 70 +++++++++++++++++ 5 files changed, 166 insertions(+), 4 deletions(-) create mode 100644 lib/pinchflat/workers/data_backfill_worker.ex create mode 100644 test/pinchflat/workers/data_backfill_worker_test.exs diff --git a/lib/pinchflat/application.ex b/lib/pinchflat/application.ex index 1527d7d..6f838bc 100644 --- a/lib/pinchflat/application.ex +++ b/lib/pinchflat/application.ex @@ -10,9 +10,9 @@ defmodule Pinchflat.Application do children = [ PinchflatWeb.Telemetry, Pinchflat.Repo, - # {Task, &run_startup_tasks/0}, - Pinchflat.StartupTasks, + # Must be before startup tasks {Oban, Application.fetch_env!(:pinchflat, Oban)}, + Pinchflat.StartupTasks, {DNSCluster, query: Application.get_env(:pinchflat, :dns_cluster_query) || :ignore}, {Phoenix.PubSub, name: Pinchflat.PubSub}, # Start the Finch HTTP client for sending emails diff --git a/lib/pinchflat/startup_tasks.ex b/lib/pinchflat/startup_tasks.ex index e4b1aec..8e4be23 100644 --- a/lib/pinchflat/startup_tasks.ex +++ b/lib/pinchflat/startup_tasks.ex @@ -8,8 +8,11 @@ defmodule Pinchflat.StartupTasks do # restart: :temporary means that this process will never be restarted (ie: will run once and then die) use GenServer, restart: :temporary + import Ecto.Query, warn: false + alias Pinchflat.Repo alias Pinchflat.Settings + alias Pinchflat.Workers.DataBackfillWorker def start_link(opts \\ []) do GenServer.start_link(__MODULE__, %{}, opts) @@ -21,11 +24,13 @@ defmodule Pinchflat.StartupTasks do Any code defined here will run every time the application starts. You must make sure that the code is idempotent and safe to run multiple times. - This is a good place to set up default settings, create initial records, stuff like that + This is a good place to set up default settings, create initial records, stuff like that. + Should be fast - anything with the potential to be slow should be kicked off as a job instead. """ @impl true def init(state) do apply_default_settings() + enqueue_backfill_worker() {:ok, state} end @@ -34,4 +39,12 @@ defmodule Pinchflat.StartupTasks do Settings.fetch!(:onboarding, true) Settings.fetch!(:pro_enabled, false) end + + defp enqueue_backfill_worker do + DataBackfillWorker.cancel_pending_backfill_jobs() + + %{} + |> DataBackfillWorker.new() + |> Repo.insert_unique_job() + end end diff --git a/lib/pinchflat/workers/data_backfill_worker.ex b/lib/pinchflat/workers/data_backfill_worker.ex new file mode 100644 index 0000000..15a7041 --- /dev/null +++ b/lib/pinchflat/workers/data_backfill_worker.ex @@ -0,0 +1,75 @@ +defmodule Pinchflat.Workers.DataBackfillWorker do + @moduledoc false + + use Oban.Worker, + queue: :media_local_metadata, + unique: [period: :infinity, states: [:available, :scheduled, :retryable]], + tags: ["media_item", "media_metadata", "local_metadata", "data_backfill"] + + # This one is going to be a little more self-contained + # instead of relying on outside modules for the methods. + # That's because, for now, these methods are not intended + # to be used elsewhere. + # + # I'm just trying out that pattern and seeing if I like it better + # so this may change. + import Ecto.Query, warn: false + require Logger + + alias __MODULE__ + alias Pinchflat.Repo + alias Pinchflat.Media.MediaItem + + @doc """ + Cancels all pending backfill jobs. Useful for ensuring worker runs immediately + on app boot. + + Returns {:ok, integer()} + """ + def cancel_pending_backfill_jobs do + Oban.Job + |> where(worker: "Pinchflat.Workers.DataBackfillWorker") + |> Oban.cancel_all_jobs() + end + + @impl Oban.Worker + @doc """ + Performs one-off tasks to get data in the right shape. + This can be needed when we add new features or change the way + we store data. Must be idempotent. All new data should already + conform to the expected schema so this should only be needed + for existing data. Still runs periodically to be safe. + + Returns :ok + """ + def perform(%Oban.Job{}) do + Logger.info("Running data backfill worker") + backfill_shorts_data() + + reschedule_backfill() + + :ok + end + + defp backfill_shorts_data do + query = + from( + m in MediaItem, + where: fragment("? like ?", m.original_url, "%/shorts/%"), + where: m.short_form_content == false + ) + + {count, _} = Repo.update_all(query, set: [short_form_content: true]) + + Logger.info("Backfill worker set short_form_content to true for #{count} media items.") + end + + defp reschedule_backfill do + # Run hourly + next_run_in = 60 * 60 + + %{} + |> DataBackfillWorker.new(schedule_in: next_run_in) + |> Repo.insert_unique_job() + end +end diff --git a/lib/pinchflat/workers/fast_indexing_worker.ex b/lib/pinchflat/workers/fast_indexing_worker.ex index e2f135e..96ac85a 100644 --- a/lib/pinchflat/workers/fast_indexing_worker.ex +++ b/lib/pinchflat/workers/fast_indexing_worker.ex @@ -14,7 +14,11 @@ defmodule Pinchflat.Workers.FastIndexingWorker do @impl Oban.Worker @doc """ - TODO + Kicks off the fast indexing process for a source, reschedules the job to run again + once complete. See `MediaCollectionIndexingWorker` and `MediaIndexingWorker` comments + for more + + Returns :ok | {:ok, :job_exists} | {:ok, %Task{}} """ def perform(%Oban.Job{args: %{"id" => source_id}}) do source = Sources.get_source!(source_id) diff --git a/test/pinchflat/workers/data_backfill_worker_test.exs b/test/pinchflat/workers/data_backfill_worker_test.exs new file mode 100644 index 0000000..856bfb6 --- /dev/null +++ b/test/pinchflat/workers/data_backfill_worker_test.exs @@ -0,0 +1,70 @@ +defmodule Pinchflat.Workers.DataBackfillWorkerTest do + use Pinchflat.DataCase + + import Pinchflat.MediaFixtures + + alias Pinchflat.Workers.DataBackfillWorker + alias Pinchflat.Workers.FilesystemDataWorker + + describe "cancel_pending_backfill_jobs/0" do + test "cancels all pending backfill jobs" do + %{} + |> DataBackfillWorker.new() + |> Repo.insert_unique_job() + + assert_enqueued(worker: DataBackfillWorker) + + DataBackfillWorker.cancel_pending_backfill_jobs() + + refute_enqueued(worker: DataBackfillWorker) + end + + test "does not cancel jobs for other workers" do + %{id: 0} + |> FilesystemDataWorker.new() + |> Repo.insert_unique_job() + + assert_enqueued(worker: FilesystemDataWorker) + + DataBackfillWorker.cancel_pending_backfill_jobs() + + assert_enqueued(worker: FilesystemDataWorker) + end + end + + describe "perform/1" do + setup do + DataBackfillWorker.cancel_pending_backfill_jobs() + + :ok + end + + test "reschedules itself once complete" do + perform_job(DataBackfillWorker, %{}) + + assert_enqueued(worker: DataBackfillWorker, scheduled_at: now_plus(60, :minutes)) + end + end + + describe "perform/1 when testing backfill_shorts_data" do + test "sets short_form_content to true for media items with shorts in the URL" do + media_item = media_item_with_attachments(%{original_url: "https://example.com/shorts/123"}) + + refute media_item.short_form_content + + perform_job(DataBackfillWorker, %{}) + + assert Repo.reload!(media_item).short_form_content + end + + test "does not set short_form_content to true for media items without shorts in the URL" do + media_item = media_item_with_attachments(%{original_url: "https://example.com/longs/123"}) + + refute media_item.short_form_content + + perform_job(DataBackfillWorker, %{}) + + refute Repo.reload!(media_item).short_form_content + end + end +end