Improved the way the source metadata worker updates the source

lib/pinchflat/metadata/source_metadata_storage_worker.ex

@@ -4,10 +4,7 @@ defmodule Pinchflat.Metadata.SourceMetadataStorageWorker do
   use Oban.Worker,
     queue: :remote_metadata,
     tags: ["media_source", "source_metadata", "remote_metadata"],
-    max_attempts: 3,
-    # This is the only thing stopping this job from calling itself
-    # in an infinite loop. Time is in seconds
-    unique: [period: 120]
+    max_attempts: 3
 
   require Logger
 
@@ -37,6 +34,10 @@ defmodule Pinchflat.Metadata.SourceMetadataStorageWorker do
     - The series directory for the source
     - The NFO file for the source (if specified)
 
+  The worker is kicked off after a source is inserted/updated - this can
+  take an unknown amount of time so don't rely on this data being here
+  before, say, the first indexing or downloading task is complete.
+
   Returns :ok
   """
   @impl Oban.Worker
@@ -45,16 +46,18 @@ defmodule Pinchflat.Metadata.SourceMetadataStorageWorker do
     source_metadata = fetch_source_metadata(source)
     series_directory = determine_series_directory(source)
 
-    # Since updating a source kicks this job off again, we enforce job uniqueness (above)
-    # to once, per source, per x minutes. This is to prevent a job from calling itself
-    # in an infinite loop.
-    Sources.update_source(source, %{
-      series_directory: series_directory,
-      nfo_filepath: store_source_nfo(source, series_directory, source_metadata),
-      metadata: %{
-        metadata_filepath: store_source_metadata(source, source_metadata)
-      }
-    })
+    # `run_post_commit_tasks: false` prevents this from running in an infinite loop
+    Sources.update_source(
+      source,
+      %{
+        series_directory: series_directory,
+        nfo_filepath: store_source_nfo(source, series_directory, source_metadata),
+        metadata: %{
+          metadata_filepath: store_source_metadata(source, source_metadata)
+        }
+      },
+      run_post_commit_tasks: false
+    )
 
     :ok
   rescue

lib/pinchflat/sources/sources.ex

@@ -51,15 +51,19 @@ defmodule Pinchflat.Sources do
   though we know it's going to fail so it picks up any addl. database errors
   and fulfills our return contract.
 
+  You can pass options to control the behavior of the function:
+    - `run_post_commit_tasks` (default: true) - If false, the function will not
+      enqueue any tasks in `commit_and_handle_tasks`.
+
   Returns {:ok, %Source{}} | {:error, %Ecto.Changeset{}}
   """
-  def create_source(attrs) do
+  def create_source(attrs, opts \\ []) do
     case change_source(%Source{}, attrs, :initial) do
       %Ecto.Changeset{valid?: true} ->
         %Source{}
         |> maybe_change_source_from_url(attrs)
         |> maybe_change_indexing_frequency()
-        |> commit_and_handle_tasks()
+        |> commit_and_handle_tasks(opts)
 
       changeset ->
         Repo.insert(changeset)
@@ -79,15 +83,19 @@ defmodule Pinchflat.Sources do
   though we know it's going to fail so it picks up any addl. database errors
   and fulfills our return contract.
 
+  You can pass options to control the behavior of the function:
+    - `run_post_commit_tasks` (default: true) - If false, the function will not
+      enqueue any tasks in `commit_and_handle_tasks`.
+
   Returns {:ok, %Source{}} | {:error, %Ecto.Changeset{}}
   """
-  def update_source(%Source{} = source, attrs) do
+  def update_source(%Source{} = source, attrs, opts \\ []) do
     case change_source(source, attrs, :initial) do
       %Ecto.Changeset{valid?: true} ->
         source
         |> maybe_change_source_from_url(attrs)
         |> maybe_change_indexing_frequency()
-        |> commit_and_handle_tasks()
+        |> commit_and_handle_tasks(opts)
 
       changeset ->
         Repo.update(changeset)
@@ -206,12 +214,16 @@ defmodule Pinchflat.Sources do
     end
   end
 
-  defp commit_and_handle_tasks(changeset) do
+  defp commit_and_handle_tasks(changeset, opts) do
+    run_post_commit_tasks = Keyword.get(opts, :run_post_commit_tasks, true)
+
     case Repo.insert_or_update(changeset) do
       {:ok, %Source{} = source} ->
-        maybe_handle_media_tasks(changeset, source)
-        maybe_run_indexing_task(changeset, source)
-        run_metadata_storage_task(source)
+        if run_post_commit_tasks do
+          maybe_handle_media_tasks(changeset, source)
+          maybe_run_indexing_task(changeset, source)
+          run_metadata_storage_task(source)
+        end
 
         {:ok, source}
 

test/pinchflat/metadata/source_metadata_storage_worker_test.exs

@@ -38,27 +38,9 @@ defmodule Pinchflat.Metadata.SourceMetadataStorageWorkerTest do
 
       source = source_fixture()
 
-      perform_job(SourceMetadataStorageWorker, %{id: source.id})
       perform_job(SourceMetadataStorageWorker, %{id: source.id})
 
-      assert [_] = all_enqueued(worker: SourceMetadataStorageWorker)
-    end
-
-    test "doesn't prevent over source jobs from running" do
-      stub(YtDlpRunnerMock, :run, fn
-        _url, _opts, ot when ot == @source_details_ot -> {:ok, source_details_return_fixture()}
-        _url, _opts, ot when ot == @metadata_ot -> {:ok, "{}"}
-      end)
-
-      source_1 = source_fixture()
-      source_2 = source_fixture()
-
-      perform_job(SourceMetadataStorageWorker, %{id: source_1.id})
-      perform_job(SourceMetadataStorageWorker, %{id: source_1.id})
-      perform_job(SourceMetadataStorageWorker, %{id: source_2.id})
-      perform_job(SourceMetadataStorageWorker, %{id: source_2.id})
-
-      assert [_, _] = all_enqueued(worker: SourceMetadataStorageWorker)
+      assert [] = all_enqueued(worker: SourceMetadataStorageWorker)
     end
 
     test "does not blow up if the record doesn't exist" do

test/pinchflat/sources_test.exs

@@ -58,7 +58,7 @@ defmodule Pinchflat.SourcesTest do
     end
   end
 
-  describe "create_source/1" do
+  describe "create_source/2" do
     test "creates a source and adds name + ID from runner response for channels" do
       expect(YtDlpRunnerMock, :run, &channel_mock/3)
 
@@ -254,7 +254,23 @@ defmodule Pinchflat.SourcesTest do
     end
   end
 
-  describe "update_source/2" do
+  describe "create_source/2 when testing options" do
+    test "run_post_commit_tasks: false won't enqueue post-commit tasks" do
+      expect(YtDlpRunnerMock, :run, &channel_mock/3)
+
+      valid_attrs = %{
+        media_profile_id: media_profile_fixture().id,
+        original_url: "https://www.youtube.com/channel/abc123"
+      }
+
+      assert {:ok, %Source{}} = Sources.create_source(valid_attrs, run_post_commit_tasks: false)
+
+      refute_enqueued(worker: MediaCollectionIndexingWorker)
+      refute_enqueued(worker: SourceMetadataStorageWorker)
+    end
+  end
+
+  describe "update_source/3" do
     test "updates with valid data updates the source" do
       source = source_fixture()
       update_attrs = %{collection_name: "some updated name"}
@@ -427,6 +443,20 @@ defmodule Pinchflat.SourcesTest do
     end
   end
 
+  describe "update_source/3 when testing options" do
+    test "run_post_commit_tasks: false won't enqueue post-commit tasks" do
+      source = source_fixture(%{fast_index: false, download_media: false, index_frequency_minutes: -1})
+      update_attrs = %{fast_index: true, download_media: true, index_frequency_minutes: 100}
+
+      assert {:ok, %Source{}} = Sources.update_source(source, update_attrs, run_post_commit_tasks: false)
+
+      refute_enqueued(worker: MediaCollectionIndexingWorker)
+      refute_enqueued(worker: SourceMetadataStorageWorker)
+      refute_enqueued(worker: MediaDownloadWorker)
+      refute_enqueued(worker: FastIndexingWorker)
+    end
+  end
+
   describe "delete_source/2" do
     test "it deletes the source" do
       source = source_fixture()