Handled fast index worker on source update

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} ->

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

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 """

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

lib/pinchflat_web/controllers/sources/source_html/show.html.heex

@@ -84,9 +84,9 @@
           <p class="text-black dark:text-white">Nothing Here!</p>
         <% end %>
       </:tab>
-      <: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 %>
             </:col>

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()
 

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()]}