Local data worker improvements (#66)

* Adds new worker for backfilling data

* Adds backfill job to startup tasks

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

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

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

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)

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