Refactored the way I call workers

lib/pinchflat/downloading/downloading_helpers.ex

@@ -26,11 +26,7 @@ defmodule Pinchflat.Downloading.DownloadingHelpers do
   def enqueue_pending_download_tasks(%Source{download_media: true} = source) do
     source
     |> Media.list_pending_media_items_for()
-    |> Enum.each(fn media_item ->
+    |> Enum.each(&MediaDownloadWorker.kickoff_with_task/1)
-      %{id: media_item.id}
-      |> MediaDownloadWorker.new()
-      |> Tasks.create_job_with_task(media_item)
-    end)
   end
 
   def enqueue_pending_download_tasks(%Source{download_media: false}) do

lib/pinchflat/downloading/media_download_worker.ex

@@ -6,17 +6,30 @@ defmodule Pinchflat.Downloading.MediaDownloadWorker do
     unique: [period: :infinity, states: [:available, :scheduled, :retryable, :executing]],
     tags: ["media_item", "media_fetching"]
 
+  alias __MODULE__
+  alias Pinchflat.Tasks
   alias Pinchflat.Repo
   alias Pinchflat.Media
   alias Pinchflat.Downloading.MediaDownloader
 
-  @impl Oban.Worker
+  @doc """
+  Starts the media_item media download worker and creates a task for the media_item.
+
+  Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}}
+  """
+  def kickoff_with_task(media_item, opts \\ []) do
+    %{id: media_item.id}
+    |> MediaDownloadWorker.new(opts)
+    |> Tasks.create_job_with_task(media_item)
+  end
+
   @doc """
   For a given media item, download the media alongside any options.
   Does not download media if its source is set to not download media.
 
   Returns :ok | {:ok, %MediaItem{}} | {:error, any, ...any}
   """
+  @impl Oban.Worker
   def perform(%Oban.Job{args: %{"id" => media_item_id}}) do
     media_item =
       media_item_id

lib/pinchflat/fast_indexing/fast_indexing_helpers.ex

@@ -6,34 +6,13 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do
   """
 
   alias Pinchflat.Media
-  alias Pinchflat.Tasks
   alias Pinchflat.Sources.Source
   alias Pinchflat.FastIndexing.YoutubeRss
-  alias Pinchflat.FastIndexing.FastIndexingWorker
   alias Pinchflat.Downloading.MediaDownloadWorker
   alias Pinchflat.FastIndexing.MediaIndexingWorker
 
   alias Pinchflat.YtDlp.Media, as: YtDlpMedia
 
-  @doc """
-  Starts tasks for running a fast indexing task for a source's media
-  regardless of the source's fast_index state. It's assumed the
-  caller will check for fast_index.
-
-  This is used for running fast index tasks on update. On creation, the
-  fast index is enqueued after the slow index is complete.
-
-  Returns {:ok, %Task{}}.
-  """
-  def kickoff_fast_indexing_task(%Source{} = source) do
-    Tasks.delete_pending_tasks_for(source, "FastIndexingWorker")
-
-    %{id: source.id}
-    # Schedule this one immediately, but future ones will be on an interval
-    |> FastIndexingWorker.new()
-    |> Tasks.create_job_with_task(source)
-  end
-
   @doc """
   Fetches new media IDs from a source's YouTube RSS feed and kicks off indexing tasks
   for any new media items. See comments in `MediaIndexingWorker` for more info on the
@@ -54,9 +33,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do
     Enum.each(new_media_ids, fn media_id ->
       url = "https://www.youtube.com/watch?v=#{media_id}"
 
-      %{id: source.id, media_url: url}
+      MediaIndexingWorker.kickoff_with_task(source, url)
-      |> MediaIndexingWorker.new()
-      |> Tasks.create_job_with_task(source)
     end)
   end
 
@@ -74,9 +51,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do
     case maybe_media_item do
       {:ok, media_item} ->
         if source.download_media && Media.pending_download?(media_item) do
-          %{id: media_item.id}
+          MediaDownloadWorker.kickoff_with_task(media_item)
-          |> MediaDownloadWorker.new()
-          |> Tasks.create_job_with_task(media_item)
         end
 
         {:ok, media_item}

lib/pinchflat/fast_indexing/fast_indexing_worker.ex

@@ -12,7 +12,17 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do
   alias Pinchflat.Sources.Source
   alias Pinchflat.FastIndexing.FastIndexingHelpers
 
-  @impl Oban.Worker
+  @doc """
+  Starts the source fast indexing worker and creates a task for the source.
+
+  Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}}
+  """
+  def kickoff_with_task(source, opts \\ []) do
+    %{id: source.id}
+    |> FastIndexingWorker.new(opts)
+    |> Tasks.create_job_with_task(source)
+  end
+
   @doc """
   Kicks off the fast indexing process for a source, reschedules the job to run again
   once complete. See `MediaCollectionIndexingWorker` and `MediaIndexingWorker` comments
@@ -20,6 +30,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do
 
   Returns :ok | {:ok, :job_exists} | {:ok, %Task{}}
   """
+  @impl Oban.Worker
   def perform(%Oban.Job{args: %{"id" => source_id}}) do
     source = Sources.get_source!(source_id)
 
@@ -35,10 +46,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do
   defp reschedule_indexing(source) do
     next_run_in = Source.fast_index_frequency() * 60
 
-    %{id: source.id}
+    case kickoff_with_task(source, schedule_in: next_run_in) do
-    |> FastIndexingWorker.new(schedule_in: next_run_in)
-    |> Tasks.create_job_with_task(source)
-    |> case do
       {:ok, task} -> {:ok, task}
       {:error, :duplicate_job} -> {:ok, :job_exists}
     end

lib/pinchflat/fast_indexing/media_indexing_worker.ex

@@ -8,10 +8,22 @@ defmodule Pinchflat.FastIndexing.MediaIndexingWorker do
 
   require Logger
 
+  alias __MODULE__
+  alias Pinchflat.Tasks
   alias Pinchflat.Sources
   alias Pinchflat.FastIndexing.FastIndexingHelpers
 
-  @impl Oban.Worker
+  @doc """
+  Starts the fast media indexing worker and creates a task for the source.
+
+  Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}}
+  """
+  def kickoff_with_task(source, media_url, opts \\ []) do
+    %{id: source.id, media_url: media_url}
+    |> MediaIndexingWorker.new(opts)
+    |> Tasks.create_job_with_task(source)
+  end
+
   @doc """
   Similar to `MediaCollectionIndexingWorker`, but for individual media items.
   Does not reschedule or check anything to do with a source's indexing
@@ -37,6 +49,7 @@ defmodule Pinchflat.FastIndexing.MediaIndexingWorker do
 
   Returns :ok
   """
+  @impl Oban.Worker
   def perform(%Oban.Job{args: %{"id" => source_id, "media_url" => media_url}}) do
     source = Sources.get_source!(source_id)
 

lib/pinchflat/metadata/source_metadata_storage_worker.ex

@@ -19,14 +19,11 @@ defmodule Pinchflat.Metadata.SourceMetadataStorageWorker do
   @doc """
   Starts the source metadata storage worker and creates a task for the source.
 
-  IDEA: testing out this method of handling job kickoff. I think I like it, so
-  I may use it in other places. Just testing it for now
-
   Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}}
   """
-  def kickoff_with_task(source) do
+  def kickoff_with_task(source, opts \\ []) do
     %{id: source.id}
-    |> SourceMetadataStorageWorker.new()
+    |> SourceMetadataStorageWorker.new(opts)
     |> Tasks.create_job_with_task(source)
   end
 

lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex

@@ -13,7 +13,17 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorker do
   alias Pinchflat.FastIndexing.FastIndexingWorker
   alias Pinchflat.SlowIndexing.SlowIndexingHelpers
 
-  @impl Oban.Worker
+  @doc """
+  Starts the source slow indexing worker and creates a task for the source.
+
+  Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}}
+  """
+  def kickoff_with_task(source, opts \\ []) do
+    %{id: source.id}
+    |> MediaCollectionIndexingWorker.new(opts)
+    |> Tasks.create_job_with_task(source)
+  end
+
   @doc """
   The ID is that of a source _record_, not a YouTube channel/playlist ID. Indexes
   the provided source, kicks off downloads for each new MediaItem, and
@@ -58,6 +68,7 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorker do
 
   Returns :ok | {:ok, %Task{}}
   """
+  @impl Oban.Worker
   def perform(%Oban.Job{args: %{"id" => source_id}}) do
     source = Sources.get_source!(source_id)
 

lib/pinchflat/slow_indexing/slow_indexing_helpers.ex

@@ -31,10 +31,7 @@ defmodule Pinchflat.SlowIndexing.SlowIndexingHelpers do
     Tasks.delete_pending_tasks_for(source, "MediaIndexingWorker")
     Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker")
 
-    %{id: source.id}
+    MediaCollectionIndexingWorker.kickoff_with_task(source)
-    # Schedule this one immediately, but future ones will be on an interval
-    |> MediaCollectionIndexingWorker.new()
-    |> Tasks.create_job_with_task(source)
   end
 
   @doc """
@@ -125,9 +122,7 @@ defmodule Pinchflat.SlowIndexing.SlowIndexingHelpers do
         if source.download_media && Media.pending_download?(media_item) do
           Logger.debug("FileFollowerServer Handler: Enqueuing download task for #{inspect(media_attrs)}")
 
-          %{id: media_item.id}
+          MediaDownloadWorker.kickoff_with_task(media_item)
-          |> MediaDownloadWorker.new()
-          |> Tasks.create_job_with_task(media_item)
         end
 
       {:error, changeset} ->

lib/pinchflat/sources/sources.ex

@@ -14,7 +14,7 @@ defmodule Pinchflat.Sources do
   alias Pinchflat.Metadata.SourceMetadata
   alias Pinchflat.Filesystem.FilesystemHelpers
   alias Pinchflat.Downloading.DownloadingHelpers
-  alias Pinchflat.FastIndexing.FastIndexingHelpers
+  alias Pinchflat.FastIndexing.FastIndexingWorker
   alias Pinchflat.SlowIndexing.SlowIndexingHelpers
   alias Pinchflat.Metadata.SourceMetadataStorageWorker
 
@@ -121,17 +121,9 @@ defmodule Pinchflat.Sources do
     Source.changeset(source, attrs, validation_stage)
   end
 
-  @doc """
+  # NOTE: When operating in the ideal path, this effectively adds an API call
-  Returns an `%Ecto.Changeset{}` for tracking source changes and additionally
+  # to the source creation/update process. Should be used only when needed.
-  fetches source details from the original_url (if provided). If the source
+  defp maybe_change_source_from_url(%Source{} = source, attrs) do
-  details cannot be fetched, an error is added to the changeset.
-
-  NOTE: When operating in the ideal path, this effectively adds an API call
-  to the source creation/update process. Should be used only when needed.
-
-  NOTE: this can almost certainly be made private now
-  """
-  def maybe_change_source_from_url(%Source{} = source, attrs) do
     case change_source(source, attrs) do
       %Ecto.Changeset{changes: %{original_url: _}} = changeset ->
         add_source_details_to_changeset(source, changeset)
@@ -271,7 +263,8 @@ defmodule Pinchflat.Sources do
   defp maybe_update_fast_indexing_task(changeset, source) do
     case changeset.changes do
       %{fast_index: true} ->
-        FastIndexingHelpers.kickoff_fast_indexing_task(source)
+        Tasks.delete_pending_tasks_for(source, "FastIndexingWorker")
+        FastIndexingWorker.kickoff_with_task(source)
 
       %{fast_index: false} ->
         Tasks.delete_pending_tasks_for(source, "FastIndexingWorker")

test/pinchflat/downloading/media_download_worker_test.exs

@@ -23,6 +23,19 @@ defmodule Pinchflat.Downloading.MediaDownloadWorkerTest do
     {:ok, %{media_item: media_item}}
   end
 
+  describe "kickoff_with_task/2" do
+    test "starts the worker", %{media_item: media_item} do
+      assert [] = all_enqueued(worker: MediaDownloadWorker)
+      assert {:ok, _} = MediaDownloadWorker.kickoff_with_task(media_item)
+      assert [_] = all_enqueued(worker: MediaDownloadWorker)
+    end
+
+    test "attaches a task", %{media_item: media_item} do
+      assert {:ok, task} = MediaDownloadWorker.kickoff_with_task(media_item)
+      assert task.media_item_id == media_item.id
+    end
+  end
+
   describe "perform/1" do
     test "it saves attributes to the media_item", %{media_item: media_item} do
       expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->

test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs

@@ -2,51 +2,20 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpersTest do
   use Pinchflat.DataCase
 
   import Mox
-  import Pinchflat.TasksFixtures
   import Pinchflat.MediaFixtures
   import Pinchflat.SourcesFixtures
   import Pinchflat.ProfilesFixtures
 
   alias Pinchflat.Tasks
-  alias Pinchflat.Tasks.Task
   alias Pinchflat.Media.MediaItem
   alias Pinchflat.Downloading.MediaDownloadWorker
   alias Pinchflat.FastIndexing.MediaIndexingWorker
   alias Pinchflat.FastIndexing.FastIndexingHelpers
-  alias Pinchflat.FastIndexing.FastIndexingWorker
 
   setup :verify_on_exit!
 
   @media_url "https://www.youtube.com/watch?v=test_1"
 
-  describe "kickoff_fast_indexing_task/1" do
-    test "it schedules a job" do
-      source = source_fixture()
-
-      assert {:ok, _} = FastIndexingHelpers.kickoff_fast_indexing_task(source)
-
-      assert_enqueued(worker: FastIndexingWorker, args: %{"id" => source.id})
-    end
-
-    test "it creates and attaches a task" do
-      source = source_fixture()
-
-      assert {:ok, %Task{} = task} = FastIndexingHelpers.kickoff_fast_indexing_task(source)
-
-      assert task.source_id == source.id
-    end
-
-    test "it deletes any fast indexing tasks for the source" do
-      source = source_fixture()
-      {:ok, job} = Oban.insert(FastIndexingWorker.new(%{"id" => source.id}))
-      task = task_fixture(source_id: source.id, job_id: job.id)
-
-      assert {:ok, _} = FastIndexingHelpers.kickoff_fast_indexing_task(source)
-
-      assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end
-    end
-  end
-
   describe "kickoff_indexing_tasks_from_youtube_rss_feed/1" do
     setup do
       {:ok, [source: source_fixture()]}

test/pinchflat/fast_indexing/fast_indexing_worker_test.exs

@@ -9,6 +9,23 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorkerTest do
 
   setup :verify_on_exit!
 
+  describe "kickoff_with_task/2" do
+    test "starts the worker" do
+      source = source_fixture(fast_index: true)
+
+      assert [] = all_enqueued(worker: FastIndexingWorker)
+      assert {:ok, _} = FastIndexingWorker.kickoff_with_task(source)
+      assert [_] = all_enqueued(worker: FastIndexingWorker)
+    end
+
+    test "attaches a task" do
+      source = source_fixture(fast_index: true)
+
+      assert {:ok, task} = FastIndexingWorker.kickoff_with_task(source)
+      assert task.source_id == source.id
+    end
+  end
+
   describe "perform/1" do
     test "calls out to Youtube RSS if enabled" do
       expect(HTTPClientMock, :get, fn _url -> {:ok, ""} end)

test/pinchflat/fast_indexing/media_indexing_worker_test.exs

@@ -6,8 +6,8 @@ defmodule Pinchflat.FastIndexing.MediaIndexingWorkerTest do
   import Pinchflat.SourcesFixtures
 
   alias Pinchflat.Media.MediaItem
-  alias Pinchflat.FastIndexing.MediaIndexingWorker
   alias Pinchflat.Downloading.MediaDownloadWorker
+  alias Pinchflat.FastIndexing.MediaIndexingWorker
 
   @media_url "https://www.youtube.com/watch?v=1234567890"
 
@@ -19,6 +19,19 @@ defmodule Pinchflat.FastIndexing.MediaIndexingWorkerTest do
     {:ok, source: source}
   end
 
+  describe "kickoff_with_task/2" do
+    test "starts the worker", %{source: source} do
+      assert [] = all_enqueued(worker: MediaIndexingWorker)
+      assert {:ok, _} = MediaIndexingWorker.kickoff_with_task(source, @media_url)
+      assert [_] = all_enqueued(worker: MediaIndexingWorker)
+    end
+
+    test "attaches a task", %{source: source} do
+      assert {:ok, task} = MediaIndexingWorker.kickoff_with_task(source, @media_url)
+      assert task.source_id == source.id
+    end
+  end
+
   describe "perform/1" do
     test "indexes the media item and saves it to the database", %{source: source} do
       expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->

test/pinchflat/sources_test.exs

@@ -508,70 +508,6 @@ defmodule Pinchflat.SourcesTest do
     end
   end
 
-  describe "maybe_change_source_from_url/2" do
-    test "it returns a changeset" do
-      stub(YtDlpRunnerMock, :run, &channel_mock/3)
-      source = source_fixture()
-
-      assert %Ecto.Changeset{} = Sources.maybe_change_source_from_url(source, %{})
-    end
-
-    test "it does not fetch source details if the original_url isn't in the changeset" do
-      expect(YtDlpRunnerMock, :run, 0, &channel_mock/3)
-
-      changeset = Sources.maybe_change_source_from_url(%Source{}, %{name: "some updated name"})
-
-      assert %Ecto.Changeset{} = changeset
-    end
-
-    test "it fetches source details if the original_url is in the changeset" do
-      expect(YtDlpRunnerMock, :run, &channel_mock/3)
-
-      changeset =
-        Sources.maybe_change_source_from_url(%Source{}, %{
-          original_url: "https://www.youtube.com/channel/abc123"
-        })
-
-      assert %Ecto.Changeset{} = changeset
-    end
-
-    test "it adds source details to the changeset, keeping the orignal details" do
-      expect(YtDlpRunnerMock, :run, &channel_mock/3)
-
-      media_profile = media_profile_fixture()
-      media_profile_id = media_profile.id
-
-      changeset =
-        Sources.maybe_change_source_from_url(%Source{}, %{
-          original_url: "https://www.youtube.com/channel/abc123",
-          media_profile_id: media_profile.id
-        })
-
-      assert %Ecto.Changeset{} = changeset
-      assert String.starts_with?(changeset.changes.collection_id, "some_channel_id_")
-
-      assert %{
-               collection_name: "some channel name",
-               media_profile_id: ^media_profile_id,
-               original_url: "https://www.youtube.com/channel/abc123"
-             } = changeset.changes
-    end
-
-    test "it adds an error to the changeset if the runner fails" do
-      expect(YtDlpRunnerMock, :run, 1, fn _url, _opts, _ot ->
-        {:error, "some error", 1}
-      end)
-
-      changeset =
-        Sources.maybe_change_source_from_url(%Source{}, %{
-          original_url: "https://www.youtube.com/channel/abc123"
-        })
-
-      assert %Ecto.Changeset{} = changeset
-      assert errors_on(changeset).original_url == ["could not fetch source details from URL"]
-    end
-  end
-
   defp playlist_mock(_url, _opts, _ot) do
     {
       :ok,