Fast index UI (#63)

* Added fast_index field and adds it to source form

* Added fast indexing to source changeset operations

* Added fast indexing worker and updated other modules to start using it

* Handled fast index worker on source update

config/config.exs

@@ -42,6 +42,7 @@ config :pinchflat, Oban,
   # TODO: consider making this an env var or something?
   queues: [
     default: 10,
+    fast_indexing: 6,
     media_indexing: 2,
     media_collection_indexing: 2,
     media_fetching: 2,

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

@@ -46,6 +46,7 @@ defmodule Pinchflat.Sources do
   def create_source(attrs) do
     %Source{}
     |> change_source_from_url(attrs)
+    |> maybe_change_indexing_frequency()
     |> commit_and_handle_tasks()
   end
 
@@ -62,6 +63,7 @@ defmodule Pinchflat.Sources do
   def update_source(%Source{} = source, attrs) do
     source
     |> change_source_from_url(attrs)
+    |> maybe_change_indexing_frequency()
     |> commit_and_handle_tasks()
   end
 
@@ -151,6 +153,20 @@ defmodule Pinchflat.Sources do
     change_source(source, Map.merge(changes, collection_changes))
   end
 
+  defp maybe_change_indexing_frequency(changeset) do
+    fast_index = Ecto.Changeset.get_field(changeset, :fast_index)
+
+    if fast_index do
+      Ecto.Changeset.put_change(
+        changeset,
+        :index_frequency_minutes,
+        Source.index_frequency_when_fast_indexing()
+      )
+    else
+      changeset
+    end
+  end
+
   defp commit_and_handle_tasks(changeset) do
     case Repo.insert_or_update(changeset) do
       {:ok, %Source{} = source} ->
@@ -188,13 +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
+        maybe_update_slow_indexing_task(changeset, source)
-          %{index_frequency_minutes: mins} when mins > 0 -> SourceTasks.kickoff_indexing_task(source)
+        maybe_update_fast_indexing_task(changeset, source)
-          %{index_frequency_minutes: _} -> Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker")
-          _ -> :ok
-        end
     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/sources/source.ex

@@ -17,6 +17,7 @@ defmodule Pinchflat.Sources.Source do
     collection_type
     custom_name
     index_frequency_minutes
+    fast_index
     download_media
     last_indexed_at
     original_url
@@ -29,6 +30,7 @@ defmodule Pinchflat.Sources.Source do
     collection_type
     custom_name
     index_frequency_minutes
+    fast_index
     download_media
     original_url
     media_profile_id
@@ -40,6 +42,7 @@ defmodule Pinchflat.Sources.Source do
     field :collection_id, :string
     field :collection_type, Ecto.Enum, values: [:channel, :playlist]
     field :index_frequency_minutes, :integer, default: 60 * 24
+    field :fast_index, :boolean, default: false
     field :download_media, :boolean, default: true
     field :last_indexed_at, :utc_datetime
     # This should only be used for user reference going forward
@@ -62,4 +65,16 @@ defmodule Pinchflat.Sources.Source do
     |> validate_required(@required_fields)
     |> unique_constraint([:collection_id, :media_profile_id])
   end
+
+  @doc false
+  def index_frequency_when_fast_indexing do
+    # 30 days in minutes
+    60 * 24 * 30
+  end
+
+  @doc false
+  def fast_index_frequency do
+    # minutes
+    15
+  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
@@ -24,22 +25,38 @@ defmodule Pinchflat.Tasks.SourceTasks do
 
   @doc """
   Starts tasks for indexing a source's media regardless of the source's indexing
-  frequency. It's assumed the caller will check for that.
+  frequency. It's assumed the caller will check for indexing frequency.
 
   Returns {:ok, %Task{}}.
   """
   def kickoff_indexing_task(%Source{} = source) do
+    Tasks.delete_pending_tasks_for(source, "FastIndexingWorker")
+    Tasks.delete_pending_tasks_for(source, "MediaIndexingWorker")
     Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker")
 
     %{id: source.id}
     # Schedule this one immediately, but future ones will be on an interval
     |> MediaCollectionIndexingWorker.new()
     |> Tasks.create_job_with_task(source)
-    |> case do
+  end
-      # 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
+  @doc """
-      {:ok, task} -> {:ok, task}
+  Starts tasks for running a fast indexing task for a source's media
-    end
+  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 """
@@ -90,8 +107,8 @@ defmodule Pinchflat.Tasks.SourceTasks do
     # See the method definition below for more info on how file watchers work
     # (important reading if you're not familiar with it)
     {:ok, media_attributes} = get_media_attributes_for_collection_and_setup_file_watcher(source)
-
     result = Enum.map(media_attributes, fn media_attrs -> create_media_item_from_attributes(source, media_attrs) end)
+
     Sources.update_source(source, %{last_indexed_at: DateTime.utc_now()})
     enqueue_pending_media_tasks(source)
 

lib/pinchflat/workers/fast_indexing_worker.ex

@@ -0,0 +1,42 @@
+defmodule Pinchflat.Workers.FastIndexingWorker do
+  @moduledoc false
+
+  use Oban.Worker,
+    queue: :fast_indexing,
+    unique: [period: :infinity, states: [:available, :scheduled, :retryable]],
+    tags: ["media_source", "fast_indexing"]
+
+  alias __MODULE__
+  alias Pinchflat.Tasks
+  alias Pinchflat.Sources
+  alias Pinchflat.Sources.Source
+  alias Pinchflat.Tasks.SourceTasks
+
+  @impl Oban.Worker
+  @doc """
+  TODO
+  """
+  def perform(%Oban.Job{args: %{"id" => source_id}}) do
+    source = Sources.get_source!(source_id)
+
+    if source.fast_index do
+      SourceTasks.kickoff_indexing_tasks_from_youtube_rss_feed(source)
+
+      reschedule_indexing(source)
+    else
+      :ok
+    end
+  end
+
+  defp reschedule_indexing(source) do
+    next_run_in = Source.fast_index_frequency() * 60
+
+    %{id: source.id}
+    |> 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
+  end
+end

lib/pinchflat/workers/media_collection_indexing_worker.ex

@@ -9,7 +9,9 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do
   alias __MODULE__
   alias Pinchflat.Tasks
   alias Pinchflat.Sources
+  alias Pinchflat.Sources.Source
   alias Pinchflat.Tasks.SourceTasks
+  alias Pinchflat.Workers.FastIndexingWorker
 
   @impl Oban.Worker
   @doc """
@@ -30,6 +32,27 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do
     actually run every 1 hour and 30 minutes. The tradeoff of not inundating
     the API with requests and also not overlapping jobs is worth it, IMO.
 
+  Order of operations:
+    1. The user saves a source
+    2. This job is automatically scheduled immediately. This happens in all cases.
+    3. This job indexes all content for the given source. A download job is
+       enqueued for each media item that should be downloaded. This can be impacted
+       by the `download_media` field on the source as well as the profile's
+       shorts/livestream behaviour. At this step we also attach a file reader
+       to the `yt-dlp` output file so we can create media items as they come in
+       for a little speedup (see SourceTasks comments for more)
+    4. If this job is meant to reschedule (ie: has an index frequency > 0),
+       it reschedules itself. If not, it runs once and does not reschedule
+    5. If the source uses fast indexing, that job is kicked off as well. It
+       uses RSS to run a smaller, faster, and more frequent index. That job
+       handles rescheduling itself but largely has a similar behaviour to this
+       job in that it kicks off index and maybe download jobs. The biggest difference
+       is that an index job is kicked off _for each new media item_ as opposed
+       to one larger index job. Check out `MediaIndexingWorker` comments for more.
+    6. If the job reschedules, the cycle from step 3 repeats until the heat death
+       of the universe. The user changing things like the index frequency can
+       dequeue or reschedule jobs as well
+
   NOTE: Since indexing can take a LONG time, I should check what happens if an
   application restart occurs while a job is running. Will the job be lost?
 
@@ -44,6 +67,7 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do
       {index_freq, _} when index_freq > 0 ->
         # If the indexing is on a schedule simply run indexing and reschedule
         SourceTasks.index_and_enqueue_download_for_media_items(source)
+        maybe_enqueue_fast_indexing_task(source)
         reschedule_indexing(source)
 
       {_, nil} ->
@@ -60,12 +84,26 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do
   end
 
   defp reschedule_indexing(source) do
+    next_run_in = source.index_frequency_minutes * 60
+
     %{id: source.id}
-    |> MediaCollectionIndexingWorker.new(schedule_in: source.index_frequency_minutes * 60)
+    |> MediaCollectionIndexingWorker.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
   end
+
+  defp maybe_enqueue_fast_indexing_task(source) do
+    if source.fast_index do
+      Tasks.delete_pending_tasks_for(source, "FastIndexingWorker")
+
+      next_run_in = Source.fast_index_frequency() * 60
+
+      %{id: source.id}
+      |> FastIndexingWorker.new(schedule_in: next_run_in)
+      |> Tasks.create_job_with_task(source)
+    end
+  end
 end

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 =
+    source = Repo.preload(Sources.get_source!(id), :media_profile)
-      id
-      |> Sources.get_source!()
-      |> Repo.preload([:media_profile, tasks: [:job]])
 
+    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/fast_indexing_help.html.heex

@@ -0,0 +1,21 @@
+<aside>
+  <h2 class="text-xl font-bold mb-2">What is fast indexing (experimental)?</h2>
+  <section class="ml-2 md:ml-4 mb-4 max-w-prose">
+    <p>
+      Indexing is the act of scanning a channel or playlist (aka: source) for new media.
+    </p>
+    <p class="mt-2">
+      Normal indexing uses <code class="text-sm">yt-dlp</code>
+      to scan the entire source on your specified frequency, but it's very slow for large sources. This is the most accurate way to find uploaded media with the tradeoff being that pairing a large source with a low index frequency will result in you spending most of your time indexing. Only so many indexing operations can be running at the same time, so this can impact your other source's ability to index.
+    </p>
+    <p class="mt-2">
+      Fast indexing takes a different approach. It still does an initial scan the slow way but after that it uses an RSS feed to frequently check for new videos. This has the potential to be hundreds of times faster, but it can miss videos if the uploader un-privates an old video or uploads dozens of videos in the space of a few minutes. It works well for most channels or playlists but it's not perfect.
+    </p>
+    <p class="mt-2">
+      To make up for this limitation, a normal index is still run monthly to catch any videos that were missed by fast indexing. Fast indexing overrides the normal index frequency.
+    </p>
+    <p class="mt-2">
+      Fast indexing is experimental so please report any issues on GitHub. It's only recommended for sources with over 200-ish videos and that upload frequently. Not recommended for small or inactive sources.
+    </p>
+  </section>
+</aside>

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">
+      <:tab title="Pending Tasks">
-        <%= if match?([_|_], @source.tasks) do %>
+        <%= if match?([_|_], @pending_tasks) do %>
-          <.table rows={@source.tasks} table_class="text-black dark:text-white">
+          <.table rows={@pending_tasks} table_class="text-black dark:text-white">
             <:col :let={task} label="Worker">
               <%= task.job.worker %>
             </:col>

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

@@ -3,6 +3,10 @@
     Oops, something went wrong! Please check the errors below.
   </.error>
 
+  <h3 class="mt-8 text-2xl text-black dark:text-white">
+    General Options
+  </h3>
+
   <.input
     field={f[:custom_name]}
     type="text"
@@ -19,6 +23,10 @@
     label="Media Profile"
   />
 
+  <h3 class="mt-8 text-2xl text-black dark:text-white">
+    Indexing Options
+  </h3>
+
   <.input
     field={f[:index_frequency_minutes]}
     options={friendly_index_frequencies()}
@@ -27,6 +35,18 @@
     help="Time between one index of this source finishing and the next one starting. Setting to 'On Create' will still run an initial index but no subsequent ones"
   />
 
+  <%!-- TODO: use Alpine to disable the index frequency when fast indexing is enabled --%>
+  <.input
+    field={f[:fast_index]}
+    type="toggle"
+    label="Use Fast Indexing?"
+    help="Experimental. Ignores 'Index Frequency'. Recommended for large channels that upload frequently. See below for more info"
+  />
+
+  <h3 class="mt-8 text-2xl text-black dark:text-white">
+    Downloading Options
+  </h3>
+
   <.input
     field={f[:download_media]}
     type="toggle"
@@ -34,7 +54,9 @@
     help="Unchecking still indexes media but it won't be downloaded until you enable this option"
   />
 
-  <:actions>
+  <.button class="my-10 sm:mb-7.5 w-full sm:w-auto">Save Source</.button>
-    <.button class="my-10 sm:mb-7.5 w-full sm:w-auto">Save Source</.button>
+
-  </:actions>
+  <div class="rounded-sm dark:bg-meta-4 p-4 md:p-6 mb-5">
+    <.fast_indexing_help />
+  </div>
 </.simple_form>

priv/repo/migrations/20240309182418_add_fast_index_to_sources.exs

@@ -0,0 +1,9 @@
+defmodule Pinchflat.Repo.Migrations.AddFastIndexToSources do
+  use Ecto.Migration
+
+  def change do
+    alter table(:sources) do
+      add :fast_index, :boolean, null: false, default: false
+    end
+  end
+end

test/pinchflat/sources_test.exs

@@ -9,8 +9,10 @@ defmodule Pinchflat.SourcesTest do
   alias Pinchflat.Sources
   alias Pinchflat.Tasks.SourceTasks
   alias Pinchflat.Sources.Source
-  alias Pinchflat.Workers.MediaCollectionIndexingWorker
+  alias Pinchflat.Workers.FastIndexingWorker
   alias Pinchflat.Workers.MediaDownloadWorker
+  alias Pinchflat.Workers.MediaIndexingWorker
+  alias Pinchflat.Workers.MediaCollectionIndexingWorker
 
   @invalid_source_attrs %{name: nil, collection_id: nil}
 
@@ -182,6 +184,36 @@ defmodule Pinchflat.SourcesTest do
 
       assert_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id})
     end
+
+    test "fast_index forces the index frequency to be a default value" do
+      expect(YtDlpRunnerMock, :run, &channel_mock/3)
+
+      valid_attrs = %{
+        media_profile_id: media_profile_fixture().id,
+        original_url: "https://www.youtube.com/channel/abc123",
+        fast_index: true,
+        index_frequency_minutes: 0
+      }
+
+      assert {:ok, %Source{} = source} = Sources.create_source(valid_attrs)
+
+      assert source.index_frequency_minutes == Source.index_frequency_when_fast_indexing()
+    end
+
+    test "disabling fast index will not change the index frequency" do
+      expect(YtDlpRunnerMock, :run, &channel_mock/3)
+
+      valid_attrs = %{
+        media_profile_id: media_profile_fixture().id,
+        original_url: "https://www.youtube.com/channel/abc123",
+        fast_index: false,
+        index_frequency_minutes: 0
+      }
+
+      assert {:ok, %Source{} = source} = Sources.create_source(valid_attrs)
+
+      assert source.index_frequency_minutes == 0
+    end
   end
 
   describe "update_source/2" do
@@ -244,13 +276,20 @@ defmodule Pinchflat.SourcesTest do
 
     test "updating the index frequency to 0 will delete any pending tasks" do
       source = source_fixture()
-      {:ok, job} = Oban.insert(MediaCollectionIndexingWorker.new(%{"id" => source.id}))
-      task = task_fixture(source_id: source.id, job_id: job.id)
       update_attrs = %{index_frequency_minutes: 0}
 
+      {:ok, job_1} = Oban.insert(FastIndexingWorker.new(%{"id" => source.id}))
+      task_1 = task_fixture(source_id: source.id, job_id: job_1.id)
+      {:ok, job_2} = Oban.insert(MediaIndexingWorker.new(%{"id" => source.id}))
+      task_2 = task_fixture(source_id: source.id, job_id: job_2.id)
+      {:ok, job_3} = Oban.insert(MediaCollectionIndexingWorker.new(%{"id" => source.id}))
+      task_3 = task_fixture(source_id: source.id, job_id: job_3.id)
+
       assert {:ok, %Source{}} = Sources.update_source(source, update_attrs)
 
-      assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end
+      assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task_1) end
+      assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task_2) end
+      assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task_3) end
     end
 
     test "not updating the index frequency will not re-schedule the indexing task or delete tasks" do
@@ -285,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()
 
@@ -293,6 +352,24 @@ defmodule Pinchflat.SourcesTest do
 
       assert source == Sources.get_source!(source.id)
     end
+
+    test "fast_index forces the index frequency to be a default value" do
+      source = source_fixture(%{fast_index: true})
+      update_attrs = %{index_frequency_minutes: 0}
+
+      assert {:ok, source} = Sources.update_source(source, update_attrs)
+
+      assert source.index_frequency_minutes == Source.index_frequency_when_fast_indexing()
+    end
+
+    test "disabling fast index will not change the index frequency" do
+      source = source_fixture(%{fast_index: false})
+      update_attrs = %{index_frequency_minutes: 0}
+
+      assert {:ok, source} = Sources.update_source(source, update_attrs)
+
+      assert source.index_frequency_minutes == 0
+    end
   end
 
   describe "delete_source/2" do

test/pinchflat/tasks/source_tasks_test.exs

@@ -11,6 +11,7 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
   alias Pinchflat.Tasks.Task
   alias Pinchflat.Tasks.SourceTasks
   alias Pinchflat.Media.MediaItem
+  alias Pinchflat.Workers.FastIndexingWorker
   alias Pinchflat.Workers.MediaDownloadWorker
   alias Pinchflat.Workers.MediaIndexingWorker
   alias Pinchflat.Workers.MediaCollectionIndexingWorker
@@ -34,7 +35,7 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
       assert task.source_id == source.id
     end
 
-    test "it deletes any pending tasks for the source" do
+    test "it deletes any pending media collection tasks for the source" do
       source = source_fixture()
       {:ok, job} = Oban.insert(MediaCollectionIndexingWorker.new(%{"id" => source.id}))
       task = task_fixture(source_id: source.id, job_id: job.id)
@@ -43,6 +44,54 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
 
       assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end
     end
+
+    test "it deletes any pending media tasks for the source" do
+      source = source_fixture()
+      {:ok, job} = Oban.insert(MediaIndexingWorker.new(%{"id" => source.id}))
+      task = task_fixture(source_id: source.id, job_id: job.id)
+
+      assert {:ok, _} = SourceTasks.kickoff_indexing_task(source)
+
+      assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end
+    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_indexing_task(source)
+
+      assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end
+    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

test/pinchflat/workers/fast_indexing_worker_test.exs

@@ -0,0 +1,46 @@
+defmodule Pinchflat.Workers.FastIndexingWorkerTest do
+  use Pinchflat.DataCase
+
+  import Mox
+  import Pinchflat.SourcesFixtures
+
+  alias Pinchflat.Sources.Source
+  alias Pinchflat.Workers.FastIndexingWorker
+
+  setup :verify_on_exit!
+
+  describe "perform/1" do
+    test "calls out to Youtube RSS if enabled" do
+      expect(HTTPClientMock, :get, fn _url -> {:ok, ""} end)
+      source = source_fixture(fast_index: true)
+
+      perform_job(FastIndexingWorker, %{"id" => source.id})
+    end
+
+    test "reschedules itself if fast indexing is enabled" do
+      expect(HTTPClientMock, :get, fn _url -> {:ok, ""} end)
+      source = source_fixture(fast_index: true)
+      perform_job(FastIndexingWorker, %{"id" => source.id})
+
+      assert_enqueued(
+        worker: FastIndexingWorker,
+        args: %{"id" => source.id},
+        scheduled_at: now_plus(Source.fast_index_frequency(), :minutes)
+      )
+    end
+
+    test "does not call out to Youtube RSS if disabled" do
+      expect(HTTPClientMock, :get, 0, fn _url -> {:ok, ""} end)
+      source = source_fixture(fast_index: false)
+
+      perform_job(FastIndexingWorker, %{"id" => source.id})
+    end
+
+    test "does not reschedule itself if fast indexing is disabled" do
+      source = source_fixture(fast_index: false)
+      perform_job(FastIndexingWorker, %{"id" => source.id})
+
+      refute_enqueued(worker: FastIndexingWorker, args: %{"id" => source.id})
+    end
+  end
+end

test/pinchflat/workers/media_collection_indexing_worker_test.exs

@@ -2,12 +2,15 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorkerTest do
   use Pinchflat.DataCase
 
   import Mox
+  import Pinchflat.TasksFixtures
   import Pinchflat.MediaFixtures
   import Pinchflat.SourcesFixtures
 
   alias Pinchflat.Tasks
-  alias Pinchflat.Workers.MediaCollectionIndexingWorker
+  alias Pinchflat.Sources.Source
+  alias Pinchflat.Workers.FastIndexingWorker
   alias Pinchflat.Workers.MediaDownloadWorker
+  alias Pinchflat.Workers.MediaCollectionIndexingWorker
 
   setup :verify_on_exit!
 
@@ -105,6 +108,40 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorkerTest do
       end)
     end
 
+    test "it creates a future task for fast indexing if appropriate" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end)
+
+      source = source_fixture(index_frequency_minutes: 10, fast_index: true)
+      perform_job(MediaCollectionIndexingWorker, %{id: source.id})
+
+      assert_enqueued(
+        worker: FastIndexingWorker,
+        args: %{"id" => source.id},
+        scheduled_at: now_plus(Source.fast_index_frequency(), :minutes)
+      )
+    end
+
+    test "it deletes existing fast indexing tasks if a new one is created" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end)
+
+      source = source_fixture(index_frequency_minutes: 10, fast_index: true)
+      {:ok, job} = Oban.insert(FastIndexingWorker.new(%{"id" => source.id}))
+      task = task_fixture(source_id: source.id, job_id: job.id)
+
+      perform_job(MediaCollectionIndexingWorker, %{id: source.id})
+
+      assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end
+    end
+
+    test "it does not create a task for fast indexing otherwise" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end)
+
+      source = source_fixture(index_frequency_minutes: 10, fast_index: false)
+      perform_job(MediaCollectionIndexingWorker, %{id: source.id})
+
+      refute_enqueued(worker: FastIndexingWorker)
+    end
+
     test "it creates the basic media_item records" do
       expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, source_attributes_return_fixture()} end)