diff --git a/lib/pinchflat/api/youtube_rss.ex b/lib/pinchflat/api/youtube_rss.ex index 146f1b1..7eef959 100644 --- a/lib/pinchflat/api/youtube_rss.ex +++ b/lib/pinchflat/api/youtube_rss.ex @@ -3,6 +3,8 @@ defmodule Pinchflat.Api.YoutubeRss do Methods for interacting with YouTube RSS feeds """ + require Logger + alias Pinchflat.Sources.Source @doc """ @@ -11,6 +13,8 @@ defmodule Pinchflat.Api.YoutubeRss do Returns {:ok, [binary()]} | {:error, binary()} """ def get_recent_media_ids_from_rss(%Source{} = source) do + Logger.debug("Fetching recent media IDs from YouTube RSS feed for source: #{source.collection_id}") + case http_client().get(rss_url_for_source(source)) do {:ok, response} -> response = to_string(response) @@ -25,6 +29,8 @@ defmodule Pinchflat.Api.YoutubeRss do |> Enum.filter(&(String.length(&1) > 0)) |> Enum.uniq() + Logger.debug("Media ids fetched from RSS: #{inspect(media_ids)}") + {:ok, media_ids} {:error, _reason} -> diff --git a/lib/pinchflat/sources.ex b/lib/pinchflat/sources.ex index 9baf37c..eee7c64 100644 --- a/lib/pinchflat/sources.ex +++ b/lib/pinchflat/sources.ex @@ -156,16 +156,14 @@ defmodule Pinchflat.Sources do defp maybe_change_indexing_frequency(changeset) do fast_index = Ecto.Changeset.get_field(changeset, :fast_index) - case {changeset.changes, fast_index} do - {%{index_frequency_minutes: _}, true} -> - Ecto.Changeset.put_change( - changeset, - :index_frequency_minutes, - Source.index_frequency_when_fast_indexing() - ) - - _ -> - changeset + if fast_index do + Ecto.Changeset.put_change( + changeset, + :index_frequency_minutes, + Source.index_frequency_when_fast_indexing() + ) + else + changeset end end @@ -206,20 +204,38 @@ defmodule Pinchflat.Sources do # If the record has been persisted, only run indexing if the # indexing frequency has been changed and is now greater than 0 %{__meta__: %{state: :loaded}} -> - case changeset.changes do - %{index_frequency_minutes: mins} when mins > 0 -> - SourceTasks.kickoff_indexing_task(source) - - %{index_frequency_minutes: _} -> - Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") - Tasks.delete_pending_tasks_for(source, "MediaIndexingWorker") - Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker") - - _ -> - :ok - end + maybe_update_slow_indexing_task(changeset, source) + maybe_update_fast_indexing_task(changeset, source) end {:ok, source} end + + defp maybe_update_slow_indexing_task(changeset, source) do + case changeset.changes do + %{index_frequency_minutes: mins} when mins > 0 -> + SourceTasks.kickoff_indexing_task(source) + + %{index_frequency_minutes: _} -> + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") + Tasks.delete_pending_tasks_for(source, "MediaIndexingWorker") + Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker") + + _ -> + :ok + end + end + + defp maybe_update_fast_indexing_task(changeset, source) do + case changeset.changes do + %{fast_index: true} -> + SourceTasks.kickoff_fast_indexing_task(source) + + %{fast_index: false} -> + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") + + _ -> + :ok + end + end end diff --git a/lib/pinchflat/tasks/source_tasks.ex b/lib/pinchflat/tasks/source_tasks.ex index e110ffe..4c786ef 100644 --- a/lib/pinchflat/tasks/source_tasks.ex +++ b/lib/pinchflat/tasks/source_tasks.ex @@ -14,6 +14,7 @@ defmodule Pinchflat.Tasks.SourceTasks do alias Pinchflat.Sources.Source alias Pinchflat.Api.YoutubeRss alias Pinchflat.Media.MediaItem + alias Pinchflat.Workers.FastIndexingWorker alias Pinchflat.Workers.MediaDownloadWorker alias Pinchflat.Workers.MediaIndexingWorker alias Pinchflat.YtDlp.Backend.MediaCollection @@ -37,11 +38,25 @@ defmodule Pinchflat.Tasks.SourceTasks do # Schedule this one immediately, but future ones will be on an interval |> MediaCollectionIndexingWorker.new() |> Tasks.create_job_with_task(source) - |> case do - # This should never return {:error, :duplicate_job} since we just deleted - # any pending tasks. I'm being assertive about it so it's obvious if I'm wrong - {:ok, task} -> {:ok, task} - end + end + + @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 """ diff --git a/lib/pinchflat_web/controllers/sources/source_controller.ex b/lib/pinchflat_web/controllers/sources/source_controller.ex index 757fb44..60039f4 100644 --- a/lib/pinchflat_web/controllers/sources/source_controller.ex +++ b/lib/pinchflat_web/controllers/sources/source_controller.ex @@ -3,8 +3,9 @@ defmodule PinchflatWeb.Sources.SourceController do alias Pinchflat.Repo alias Pinchflat.Media - alias Pinchflat.Profiles + alias Pinchflat.Tasks alias Pinchflat.Sources + alias Pinchflat.Profiles alias Pinchflat.Sources.Source def index(conn, _params) do @@ -43,15 +44,18 @@ defmodule PinchflatWeb.Sources.SourceController do end def show(conn, %{"id" => id}) do - source = - id - |> Sources.get_source!() - |> Repo.preload([:media_profile, tasks: [:job]]) + source = Repo.preload(Sources.get_source!(id), :media_profile) + pending_tasks = Repo.preload(Tasks.list_pending_tasks_for(:source_id, source.id), :job) pending_media = Media.list_pending_media_items_for(source, limit: 100) downloaded_media = Media.list_downloaded_media_items_for(source, limit: 100) - render(conn, :show, source: source, pending_media: pending_media, downloaded_media: downloaded_media) + render(conn, :show, + source: source, + pending_tasks: pending_tasks, + pending_media: pending_media, + downloaded_media: downloaded_media + ) end def edit(conn, %{"id" => id}) do diff --git a/lib/pinchflat_web/controllers/sources/source_html/show.html.heex b/lib/pinchflat_web/controllers/sources/source_html/show.html.heex index d8c69d2..1b08883 100644 --- a/lib/pinchflat_web/controllers/sources/source_html/show.html.heex +++ b/lib/pinchflat_web/controllers/sources/source_html/show.html.heex @@ -84,9 +84,9 @@

Nothing Here!

<% end %> - <:tab title="Tasks"> - <%= if match?([_|_], @source.tasks) do %> - <.table rows={@source.tasks} table_class="text-black dark:text-white"> + <:tab title="Pending Tasks"> + <%= if match?([_|_], @pending_tasks) do %> + <.table rows={@pending_tasks} table_class="text-black dark:text-white"> <:col :let={task} label="Worker"> <%= task.job.worker %> diff --git a/test/pinchflat/sources_test.exs b/test/pinchflat/sources_test.exs index 8805ad5..fe1db7e 100644 --- a/test/pinchflat/sources_test.exs +++ b/test/pinchflat/sources_test.exs @@ -324,6 +324,26 @@ defmodule Pinchflat.SourcesTest do refute_enqueued(worker: MediaDownloadWorker) end + test "enabling fast_index will schedule a fast indexing task" do + source = source_fixture(fast_index: false) + update_attrs = %{fast_index: true} + + refute_enqueued(worker: FastIndexingWorker) + assert {:ok, %Source{}} = Sources.update_source(source, update_attrs) + assert_enqueued(worker: FastIndexingWorker, args: %{"id" => source.id}) + end + + test "disabling fast_index will cancel the fast indexing task" do + source = source_fixture(fast_index: true) + update_attrs = %{fast_index: false} + {:ok, job} = Oban.insert(FastIndexingWorker.new(%{"id" => source.id})) + task_fixture(source_id: source.id, job_id: job.id) + + assert_enqueued(worker: FastIndexingWorker, args: %{"id" => source.id}) + assert {:ok, %Source{}} = Sources.update_source(source, update_attrs) + refute_enqueued(worker: FastIndexingWorker) + end + test "updates with invalid data returns error changeset" do source = source_fixture() diff --git a/test/pinchflat/tasks/source_tasks_test.exs b/test/pinchflat/tasks/source_tasks_test.exs index 49f9241..55df00c 100644 --- a/test/pinchflat/tasks/source_tasks_test.exs +++ b/test/pinchflat/tasks/source_tasks_test.exs @@ -66,6 +66,34 @@ defmodule Pinchflat.Tasks.SourceTasksTest do end end + describe "kickoff_fast_indexing_task/1" do + test "it schedules a job" do + source = source_fixture() + + assert {:ok, _} = SourceTasks.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} = SourceTasks.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, _} = SourceTasks.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()]}