Added methods to kick off indexing workers from RSS response

lib/pinchflat/api/youtube_rss.ex

@@ -23,6 +23,7 @@ defmodule Pinchflat.Api.YoutubeRss do
           |> Regex.scan(response)
           |> Enum.map(fn [_, id] -> String.trim(id) end)
           |> Enum.filter(&(String.length(&1) > 0))
+          |> Enum.uniq()
 
         {:ok, media_ids}
 

lib/pinchflat/media.ex

@@ -31,6 +31,22 @@ defmodule Pinchflat.Media do
     |> Repo.all()
   end
 
+  @doc """
+  Fetches all media items belonging to a given source that have a media_id in the given list.
+  Useful for determining the what media items we DON'T already have for fast indexing.
+
+  NOTE: These queries are getting a little tedious. When I have the time, I should see about
+  implementing a query pattern and having these compose queries from a common base. This would
+  also let me compose simple queries in the module using them for one-off methods
+
+  Returns [%MediaItem{}, ...].
+  """
+  def list_media_items_by_media_id_for(%Source{} = source, media_ids) do
+    MediaItem
+    |> where([mi], mi.source_id == ^source.id and mi.media_id in ^media_ids)
+    |> Repo.all()
+  end
+
   @doc """
   Returns a list of pending media_items for a given source, where
   pending means the `media_filepath` is `nil` AND the media_item

lib/pinchflat/tasks/source_tasks.ex

@@ -12,8 +12,10 @@ defmodule Pinchflat.Tasks.SourceTasks do
   alias Pinchflat.Tasks
   alias Pinchflat.Sources
   alias Pinchflat.Sources.Source
+  alias Pinchflat.Api.YoutubeRss
   alias Pinchflat.Media.MediaItem
   alias Pinchflat.Workers.MediaDownloadWorker
+  alias Pinchflat.Workers.MediaIndexingWorker
   alias Pinchflat.YtDlp.Backend.MediaCollection
   alias Pinchflat.Workers.MediaCollectionIndexingWorker
   alias Pinchflat.Utils.FilesystemUtils.FileFollowerServer
@@ -38,6 +40,27 @@ defmodule Pinchflat.Tasks.SourceTasks do
     end
   end
 
+  @doc """
+  Fetches new media IDs from a source's YouTube RSS feed and kicks off indexing tasks
+  for any new media items. See comments in `MediaIndexingWorker` for more info on the
+  order of operations and how this fits into the indexing process.
+
+  Returns :ok
+  """
+  def kickoff_indexing_tasks_from_youtube_rss_feed(%Source{} = source) do
+    {:ok, media_ids} = YoutubeRss.get_recent_media_ids_from_rss(source)
+    existing_media_items = Media.list_media_items_by_media_id_for(source, media_ids)
+    new_media_ids = media_ids -- Enum.map(existing_media_items, & &1.media_id)
+
+    Enum.each(new_media_ids, fn media_id ->
+      url = "https://www.youtube.com/watch?v=#{media_id}"
+
+      %{id: source.id, media_url: url}
+      |> MediaIndexingWorker.new()
+      |> Tasks.create_job_with_task(source)
+    end)
+  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

lib/pinchflat/workers/media_indexing_worker.ex

@@ -4,9 +4,12 @@ defmodule Pinchflat.Workers.MediaIndexingWorker do
   use Oban.Worker,
     queue: :media_indexing,
     unique: [period: :infinity, states: [:available, :scheduled, :retryable]],
-    tags: ["media_source", "media_collection_indexing"]
+    tags: ["media_source", "media_indexing"]
+
+  require Logger
 
   alias Pinchflat.Sources
+  alias Pinchflat.Tasks.MediaItemTasks
 
   @impl Oban.Worker
   @doc """
@@ -17,13 +20,34 @@ defmodule Pinchflat.Workers.MediaIndexingWorker do
   useful for downloading one-off media items based on a URL (like for fast indexing).
 
   Only downloads media that _should_ be downloaded (ie: the source is set to download
-  and the media matches the profile's format preferences). Splits downloading into
+  and the media matches the profile's format preferences)
-  another job to keep the indexing queue moving quickly.
+
+  Order of operations:
+    1. SourceTasks.kickoff_indexing_tasks_from_youtube_rss_feed/1 (which is running
+       in its own worker) periodically checks the YouTube RSS feed for new media
+    2. If new media is found, it enqueues a MediaIndexingWorker (this module) for each new media
+       item
+    3. This worker fetches the media metadata and uses that to determine if it should be
+       downloaded. If so, it enqueues a MediaDownloadWorker
+
+  Each is a worker because they all either need to be scheduled periodically or call out to
+  an external service and will be long-running. They're split into different jobs to separate
+  retry logic for each step and allow us to better optimize various queues (eg: the indexing
+  steps can keep running while the slow download steps are worked through).
 
   Returns :ok
   """
+  def perform(%Oban.Job{args: %{"id" => source_id, "media_url" => media_url}}) do
+    source = Sources.get_source!(source_id)
 
-  def perform(%Oban.Job{args: %{"id" => source_id, "media_url" => _media_url}}) do
+    case MediaItemTasks.index_and_enqueue_download_for_media_item(source, media_url) do
-    _source = Sources.get_source!(source_id)
+      {:ok, media_item} ->
+        Logger.debug("Indexed and enqueued download for url: #{media_url} (media item: #{media_item.id})")
+
+      {:error, reason} ->
+        Logger.debug("Failed to index and enqueue download for url: #{media_url} (reason: #{inspect(reason)})")
+    end
+
+    :ok
   end
 end

test/pinchflat/api/youtube_rss_test.exs

@@ -67,5 +67,13 @@ defmodule Pinchflat.Api.YoutubeRssTest do
 
       assert {:ok, ["test_1"]} = YoutubeRss.get_recent_media_ids_from_rss(source)
     end
+
+    test "removes duplicate media IDs", %{source: source} do
+      expect(HTTPClientMock, :get, fn _url ->
+        {:ok, "<yt:videoId>test_1</yt:videoId><yt:videoId>test_1</yt:videoId>"}
+      end)
+
+      assert {:ok, ["test_1"]} = YoutubeRss.get_recent_media_ids_from_rss(source)
+    end
   end
 end

test/pinchflat/media_test.exs

@@ -45,6 +45,23 @@ defmodule Pinchflat.MediaTest do
     end
   end
 
+  describe "list_media_items_by_media_id_for/2" do
+    test "returns media_items for a given source and media_ids" do
+      source = source_fixture()
+      media_item = media_item_fixture(%{source_id: source.id, media_id: "123"})
+
+      assert Media.list_media_items_by_media_id_for(source, ["123"]) == [media_item]
+    end
+
+    test "does not return matching media_ids for a different source" do
+      source = source_fixture()
+      other_source = source_fixture()
+      _media_item = media_item_fixture(%{source_id: other_source.id, media_id: "123"})
+
+      assert Media.list_media_items_by_media_id_for(source, ["123"]) == []
+    end
+  end
+
   describe "list_pending_media_items_for/1" do
     test "it returns pending without a filepath for a given source" do
       source = source_fixture()

test/pinchflat/tasks/source_tasks_test.exs

@@ -11,8 +11,9 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
   alias Pinchflat.Tasks.Task
   alias Pinchflat.Tasks.SourceTasks
   alias Pinchflat.Media.MediaItem
-  alias Pinchflat.Workers.MediaCollectionIndexingWorker
   alias Pinchflat.Workers.MediaDownloadWorker
+  alias Pinchflat.Workers.MediaIndexingWorker
+  alias Pinchflat.Workers.MediaCollectionIndexingWorker
 
   setup :verify_on_exit!
 
@@ -44,6 +45,31 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
     end
   end
 
+  describe "kickoff_indexing_tasks_from_youtube_rss_feed/1" do
+    setup do
+      {:ok, [source: source_fixture()]}
+    end
+
+    test "enqueues a new worker for each new media_id in the source's RSS feed", %{source: source} do
+      expect(HTTPClientMock, :get, fn _url -> {:ok, "<yt:videoId>test_1</yt:videoId>"} end)
+
+      assert :ok = SourceTasks.kickoff_indexing_tasks_from_youtube_rss_feed(source)
+
+      assert [worker] = all_enqueued(worker: MediaIndexingWorker)
+      assert worker.args["id"] == source.id
+      assert worker.args["media_url"] == "https://www.youtube.com/watch?v=test_1"
+    end
+
+    test "does not enqueue a new worker for the source's media IDs we already know about", %{source: source} do
+      expect(HTTPClientMock, :get, fn _url -> {:ok, "<yt:videoId>test_1</yt:videoId>"} end)
+      media_item_fixture(source_id: source.id, media_id: "test_1")
+
+      assert :ok = SourceTasks.kickoff_indexing_tasks_from_youtube_rss_feed(source)
+
+      refute_enqueued(worker: MediaIndexingWorker)
+    end
+  end
+
   describe "index_and_enqueue_download_for_media_items/1" do
     setup do
       stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts ->

test/pinchflat/workers/media_indexing_worker_test.exs

@@ -0,0 +1,44 @@
+defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
+  use Pinchflat.DataCase
+
+  import Mox
+  import Pinchflat.MediaFixtures
+  import Pinchflat.SourcesFixtures
+
+  alias Pinchflat.Media.MediaItem
+  alias Pinchflat.Workers.MediaIndexingWorker
+  alias Pinchflat.Workers.MediaDownloadWorker
+
+  @media_url "https://www.youtube.com/watch?v=1234567890"
+
+  setup :verify_on_exit!
+
+  setup do
+    source = source_fixture()
+
+    {:ok, source: source}
+  end
+
+  describe "perform/1" do
+    test "indexes the media item and saves it to the database", %{source: source} do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
+        {:ok, media_attributes_return_fixture()}
+      end)
+
+      before = Repo.aggregate(MediaItem, :count, :id)
+      perform_job(MediaIndexingWorker, %{id: source.id, media_url: @media_url})
+
+      assert Repo.aggregate(MediaItem, :count, :id) == before + 1
+    end
+
+    test "enqueues a download job for the media item", %{source: source} do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
+        {:ok, media_attributes_return_fixture()}
+      end)
+
+      perform_job(MediaIndexingWorker, %{id: source.id, media_url: @media_url})
+
+      assert [_] = all_enqueued(worker: MediaDownloadWorker)
+    end
+  end
+end