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

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/sources.ex

@@ -211,7 +211,8 @@ defmodule Pinchflat.Sources do
             SourceTasks.kickoff_indexing_task(source)
 
           %{index_frequency_minutes: _} ->
-            # TODO: delete the recurring RSS task (when I get there)
+            Tasks.delete_pending_tasks_for(source, "FastIndexingWorker")
+            Tasks.delete_pending_tasks_for(source, "MediaIndexingWorker")
             Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker")
 
           _ ->

lib/pinchflat/sources/source.ex

@@ -68,6 +68,13 @@ defmodule Pinchflat.Sources.Source do
 
   @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

@@ -24,11 +24,13 @@ 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}
@@ -90,8 +92,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

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}
 
@@ -274,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

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,26 @@ 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_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)