diff --git a/lib/pinchflat/api/youtube_rss.ex b/lib/pinchflat/api/youtube_rss.ex index d9fa7d9..146f1b1 100644 --- a/lib/pinchflat/api/youtube_rss.ex +++ b/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} diff --git a/lib/pinchflat/media.ex b/lib/pinchflat/media.ex index f26d895..04d1379 100644 --- a/lib/pinchflat/media.ex +++ b/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 diff --git a/lib/pinchflat/tasks/source_tasks.ex b/lib/pinchflat/tasks/source_tasks.ex index ca4856c..fa8f973 100644 --- a/lib/pinchflat/tasks/source_tasks.ex +++ b/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 diff --git a/lib/pinchflat/workers/media_indexing_worker.ex b/lib/pinchflat/workers/media_indexing_worker.ex index 9579191..74222b9 100644 --- a/lib/pinchflat/workers/media_indexing_worker.ex +++ b/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 - another job to keep the indexing queue moving quickly. + and the media matches the profile's format preferences) + + 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 - _source = Sources.get_source!(source_id) + case MediaItemTasks.index_and_enqueue_download_for_media_item(source, media_url) do + {: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 diff --git a/test/pinchflat/api/youtube_rss_test.exs b/test/pinchflat/api/youtube_rss_test.exs index eb9acb6..34c386c 100644 --- a/test/pinchflat/api/youtube_rss_test.exs +++ b/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, "test_1test_1"} + end) + + assert {:ok, ["test_1"]} = YoutubeRss.get_recent_media_ids_from_rss(source) + end end end diff --git a/test/pinchflat/media_test.exs b/test/pinchflat/media_test.exs index 1719a6c..a7a992f 100644 --- a/test/pinchflat/media_test.exs +++ b/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() diff --git a/test/pinchflat/tasks/source_tasks_test.exs b/test/pinchflat/tasks/source_tasks_test.exs index eba4dec..3d688ee 100644 --- a/test/pinchflat/tasks/source_tasks_test.exs +++ b/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, "test_1"} 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, "test_1"} 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 -> diff --git a/test/pinchflat/workers/media_indexing_worker_test.exs b/test/pinchflat/workers/media_indexing_worker_test.exs new file mode 100644 index 0000000..9db9b19 --- /dev/null +++ b/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