From dc6d281d47c4b4fae431926d4dfb31901903c522 Mon Sep 17 00:00:00 2001 From: Kieran Eglin Date: Thu, 21 Nov 2024 10:21:21 -0800 Subject: [PATCH] Updated sources context to do the right thing when a source is updated --- .../fast_indexing/fast_indexing_helpers.ex | 8 ++ .../slow_indexing/slow_indexing_helpers.ex | 8 ++ lib/pinchflat/sources/sources.ex | 59 +++++++++---- .../media_profile_html/index.html.heex | 3 +- .../sources/source_html/index_table_live.ex | 1 + test/pinchflat/sources_test.exs | 82 ++++++++++++++++++- 6 files changed, 141 insertions(+), 20 deletions(-) diff --git a/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex b/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex index 9daeda1..4036509 100644 --- a/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex +++ b/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex @@ -11,14 +11,22 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do alias Pinchflat.Repo alias Pinchflat.Media + alias Pinchflat.Tasks alias Pinchflat.Sources.Source alias Pinchflat.FastIndexing.YoutubeRss alias Pinchflat.FastIndexing.YoutubeApi alias Pinchflat.Downloading.DownloadingHelpers + alias Pinchflat.FastIndexing.FastIndexingWorker alias Pinchflat.Downloading.DownloadOptionBuilder alias Pinchflat.YtDlp.Media, as: YtDlpMedia + # TODO: test + def kickoff_indexing_task(%Source{} = source) do + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker", include_executing: true) + FastIndexingWorker.kickoff_with_task(source) + end + @doc """ Fetches new media IDs for a source from YT's API or RSS, indexes them, and kicks off downloading tasks for any pending media items. See comments in `FastIndexingWorker` for more info on the diff --git a/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex b/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex index 74e5f4c..2e483bf 100644 --- a/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex +++ b/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex @@ -34,6 +34,14 @@ defmodule Pinchflat.SlowIndexing.SlowIndexingHelpers do MediaCollectionIndexingWorker.kickoff_with_task(source, job_args, job_opts) end + # TODO: test + def delete_indexing_tasks(%Source{} = source, opts \\ []) do + include_executing = Keyword.get(opts, :include_executing, false) + + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker", include_executing: include_executing) + Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker", include_executing: include_executing) + end + @doc """ Given a media source, creates (indexes) the media by creating media_items for each media ID in the source. Afterward, kicks off a download task for each pending media diff --git a/lib/pinchflat/sources/sources.ex b/lib/pinchflat/sources/sources.ex index 7052614..f4482c0 100644 --- a/lib/pinchflat/sources/sources.ex +++ b/lib/pinchflat/sources/sources.ex @@ -15,8 +15,8 @@ defmodule Pinchflat.Sources do alias Pinchflat.Metadata.SourceMetadata alias Pinchflat.Utils.FilesystemUtils alias Pinchflat.Downloading.DownloadingHelpers - alias Pinchflat.FastIndexing.FastIndexingWorker alias Pinchflat.SlowIndexing.SlowIndexingHelpers + alias Pinchflat.FastIndexing.FastIndexingHelpers alias Pinchflat.Metadata.SourceMetadataStorageWorker @doc """ @@ -272,18 +272,18 @@ defmodule Pinchflat.Sources do # current changes or if that's how it already was in the database. # Rephrased, we're essentially using it in place of `get_field/2` case {current_changes, applied_changes} do - {%{download_media: false}, _} -> - DownloadingHelpers.dequeue_pending_download_tasks(source) - {%{download_media: true}, %{enabled: true}} -> DownloadingHelpers.enqueue_pending_download_tasks(source) - {%{enabled: false}, _} -> - DownloadingHelpers.dequeue_pending_download_tasks(source) - {%{enabled: true}, %{download_media: true}} -> DownloadingHelpers.enqueue_pending_download_tasks(source) + {%{download_media: false}, _} -> + DownloadingHelpers.dequeue_pending_download_tasks(source) + + {%{enabled: false}, _} -> + DownloadingHelpers.dequeue_pending_download_tasks(source) + _ -> nil end @@ -322,13 +322,24 @@ defmodule Pinchflat.Sources do end defp maybe_update_slow_indexing_task(changeset, source) do - case changeset.changes do - %{index_frequency_minutes: mins} when mins > 0 -> + # See comment in `maybe_handle_media_tasks` as to why we need these + current_changes = changeset.changes + applied_changes = Ecto.Changeset.apply_changes(changeset) + + case {current_changes, applied_changes} do + {%{index_frequency_minutes: mins}, %{enabled: true}} when mins > 0 -> + # TODO: consider scheduling the task for the future based on the index_frequency_minutes + # compared to `last_indexed_at` SlowIndexingHelpers.kickoff_indexing_task(source) - %{index_frequency_minutes: _} -> - Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") - Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker") + {%{enabled: true}, %{index_frequency_minutes: mins}} when mins > 0 -> + SlowIndexingHelpers.kickoff_indexing_task(source) + + {%{index_frequency_minutes: mins}, _} when mins <= 0 -> + SlowIndexingHelpers.delete_indexing_tasks(source, include_executing: true) + + {%{enabled: false}, _} -> + SlowIndexingHelpers.delete_indexing_tasks(source, include_executing: true) _ -> :ok @@ -336,13 +347,25 @@ defmodule Pinchflat.Sources do end defp maybe_update_fast_indexing_task(changeset, source) do - case changeset.changes do - %{fast_index: true} -> - Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") - FastIndexingWorker.kickoff_with_task(source) + # See comment in `maybe_handle_media_tasks` as to why we need these + current_changes = changeset.changes + applied_changes = Ecto.Changeset.apply_changes(changeset) - %{fast_index: false} -> - Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") + # This technically could be simplified since `maybe_update_slow_indexing_task` + # has some overlap re: deleting pending tasks, but I'm keeping it separate + # for clarity and explicitness. + case {current_changes, applied_changes} do + {%{fast_index: true}, %{enabled: true}} -> + FastIndexingHelpers.kickoff_indexing_task(source) + + {%{enabled: true}, %{fast_index: true}} -> + FastIndexingHelpers.kickoff_indexing_task(source) + + {%{fast_index: false}, _} -> + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker", include_executing: true) + + {%{enabled: false}, _} -> + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker", include_executing: true) _ -> :ok diff --git a/lib/pinchflat_web/controllers/media_profiles/media_profile_html/index.html.heex b/lib/pinchflat_web/controllers/media_profiles/media_profile_html/index.html.heex index 9252ad4..913213c 100644 --- a/lib/pinchflat_web/controllers/media_profiles/media_profile_html/index.html.heex +++ b/lib/pinchflat_web/controllers/media_profiles/media_profile_html/index.html.heex @@ -10,7 +10,8 @@ - +<%!-- TODO: link the sources count to the sources tab --%> +<%!-- TODO: consider doing the same for other tables, like the source index's media item counts --%>
diff --git a/lib/pinchflat_web/controllers/sources/source_html/index_table_live.ex b/lib/pinchflat_web/controllers/sources/source_html/index_table_live.ex index eae6613..475ac96 100644 --- a/lib/pinchflat_web/controllers/sources/source_html/index_table_live.ex +++ b/lib/pinchflat_web/controllers/sources/source_html/index_table_live.ex @@ -8,6 +8,7 @@ defmodule PinchflatWeb.Sources.IndexTableLive do alias Pinchflat.Media.MediaItem # TODO: test (and maybe remove existing index tests) + # TODO: see comments in media profile index view def render(assigns) do ~H""" <.table rows={@sources} table_class="text-white"> diff --git a/test/pinchflat/sources_test.exs b/test/pinchflat/sources_test.exs index f365a41..a87917e 100644 --- a/test/pinchflat/sources_test.exs +++ b/test/pinchflat/sources_test.exs @@ -511,7 +511,7 @@ defmodule Pinchflat.SourcesTest do end end - describe "update_source/3 when testing indexing" do + describe "update_source/3 when testing slow indexing" do test "updating the index frequency to >0 will re-schedule the indexing task" do source = source_fixture() update_attrs = %{index_frequency_minutes: 123} @@ -556,6 +556,47 @@ defmodule Pinchflat.SourcesTest do refute_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id}) end + test "disabling a source will delete any pending tasks" do + source = source_fixture() + update_attrs = %{enabled: false} + + {:ok, job} = Oban.insert(MediaCollectionIndexingWorker.new(%{"id" => source.id})) + task = task_fixture(source_id: source.id, job_id: job.id) + + assert {:ok, %Source{}} = Sources.update_source(source, update_attrs) + + assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end + end + + test "updating the index frequency will not create a task if the source is disabled" do + source = source_fixture(enabled: false) + update_attrs = %{index_frequency_minutes: 123} + + refute_enqueued(worker: MediaCollectionIndexingWorker) + assert {:ok, %Source{}} = Sources.update_source(source, update_attrs) + refute_enqueued(worker: MediaCollectionIndexingWorker) + end + + test "enabling a source will create a task if the index frequency is >0" do + source = source_fixture(enabled: false, index_frequency_minutes: 123) + update_attrs = %{enabled: true} + + refute_enqueued(worker: MediaCollectionIndexingWorker) + assert {:ok, %Source{}} = Sources.update_source(source, update_attrs) + assert_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id}) + end + + test "enabling a source will not create a task if the index frequency is 0" do + source = source_fixture(enabled: false, index_frequency_minutes: 0) + update_attrs = %{enabled: true} + + refute_enqueued(worker: MediaCollectionIndexingWorker) + assert {:ok, %Source{}} = Sources.update_source(source, update_attrs) + refute_enqueued(worker: MediaCollectionIndexingWorker) + end + end + + describe "update_source/3 when testing fast indexing" do test "enabling fast_index will schedule a fast indexing task" do source = source_fixture(fast_index: false) update_attrs = %{fast_index: true} @@ -593,6 +634,45 @@ defmodule Pinchflat.SourcesTest do assert source.index_frequency_minutes == 0 end + + test "disabling a source will delete any pending tasks" do + source = source_fixture() + update_attrs = %{enabled: false} + + {:ok, job} = Oban.insert(FastIndexingWorker.new(%{"id" => source.id})) + task = task_fixture(source_id: source.id, job_id: job.id) + + assert {:ok, %Source{}} = Sources.update_source(source, update_attrs) + + assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end + end + + test "updating fast indexing will not create a task if the source is disabled" do + source = source_fixture(enabled: false, fast_index: false) + update_attrs = %{fast_index: true} + + refute_enqueued(worker: FastIndexingWorker) + assert {:ok, %Source{}} = Sources.update_source(source, update_attrs) + refute_enqueued(worker: FastIndexingWorker) + end + + test "enabling a source will create a task if fast_index is true" do + source = source_fixture(enabled: false, fast_index: true) + update_attrs = %{enabled: true} + + refute_enqueued(worker: FastIndexingWorker) + assert {:ok, %Source{}} = Sources.update_source(source, update_attrs) + assert_enqueued(worker: FastIndexingWorker, args: %{"id" => source.id}) + end + + test "enabling a source will not create a task if fast_index is false" do + source = source_fixture(enabled: false, fast_index: false) + update_attrs = %{enabled: true} + + refute_enqueued(worker: FastIndexingWorker) + assert {:ok, %Source{}} = Sources.update_source(source, update_attrs) + refute_enqueued(worker: FastIndexingWorker) + end end describe "update_source/3 when testing options" do