From 7f8c3ae7705e4d924c3d84b4327613ad30985b96 Mon Sep 17 00:00:00 2001 From: Kieran Eglin Date: Mon, 4 Mar 2024 10:01:47 -0800 Subject: [PATCH] Updated file watcher to enqueue download; refactored download worker methods --- lib/pinchflat/media.ex | 18 +++ lib/pinchflat/tasks.ex | 1 - lib/pinchflat/tasks/source_tasks.ex | 42 +++++-- .../workers/media_indexing_worker.ex | 11 +- test/pinchflat/media_test.exs | 28 +++++ test/pinchflat/tasks/source_tasks_test.exs | 116 +++++++++++++++--- 6 files changed, 184 insertions(+), 32 deletions(-) diff --git a/lib/pinchflat/media.ex b/lib/pinchflat/media.ex index 6f93d1b..db304c1 100644 --- a/lib/pinchflat/media.ex +++ b/lib/pinchflat/media.ex @@ -66,6 +66,24 @@ defmodule Pinchflat.Media do |> Repo.all() end + @doc """ + For a given media_item, tells you if it is pending download. This is defined as + the media_item having a `media_filepath` of `nil` and matching the format selection + rules of the parent media_profile. + + Intentionally does not take the `download_media` setting of the source into account. + + Returns boolean() + """ + def pending_download?(%MediaItem{} = media_item) do + media_profile = Repo.preload(media_item, source: :media_profile).source.media_profile + + MediaItem + |> where([mi], mi.id == ^media_item.id and is_nil(mi.media_filepath)) + |> where(^build_format_clauses(media_profile)) + |> Repo.exists?() + end + @doc """ Returns a list of media_items that match the search term. Adds a `matching_search_term` virtual field to the result set. diff --git a/lib/pinchflat/tasks.ex b/lib/pinchflat/tasks.ex index e7ae431..6426450 100644 --- a/lib/pinchflat/tasks.ex +++ b/lib/pinchflat/tasks.ex @@ -2,7 +2,6 @@ defmodule Pinchflat.Tasks do @moduledoc """ The Tasks context. """ - import Ecto.Query, warn: false alias Pinchflat.Repo diff --git a/lib/pinchflat/tasks/source_tasks.ex b/lib/pinchflat/tasks/source_tasks.ex index b893ac4..302e785 100644 --- a/lib/pinchflat/tasks/source_tasks.ex +++ b/lib/pinchflat/tasks/source_tasks.ex @@ -12,6 +12,7 @@ defmodule Pinchflat.Tasks.SourceTasks do alias Pinchflat.Tasks alias Pinchflat.Sources alias Pinchflat.Sources.Source + alias Pinchflat.Media.MediaItem alias Pinchflat.MediaClient.SourceDetails alias Pinchflat.Workers.MediaIndexingWorker alias Pinchflat.Workers.VideoDownloadWorker @@ -40,7 +41,8 @@ defmodule Pinchflat.Tasks.SourceTasks do @doc """ Given a media source, creates (indexes) the media by creating media_items for each - media ID in the source. + media ID in the source. Afterward, kicks off a download task for each pending media + item belonging to the source. You can't tell me the method name isn't descriptive! Indexing is slow and usually returns a list of all media data at once for record creation. To help with this, we use a file follower to watch the file that yt-dlp writes to @@ -48,6 +50,11 @@ defmodule Pinchflat.Tasks.SourceTasks do clarity to the user experience. This has a few things to be aware of which are documented below in the file watcher setup method. + NOTE: downloads are only enqueued if the source is set to download media. Downloads are + also enqueued for ALL pending media items, not just the ones that were indexed in this + job run. This should ensure that any stragglers are caught if, for some reason, they + weren't enqueued or somehow got de-queued. + Since indexing returns all media data EVERY TIME, we rely on the unique index of the media_id to prevent duplicates. Due to both the file follower and the fact that future indexing will index a lot of existing data, this method will MOSTLY return error @@ -55,16 +62,16 @@ defmodule Pinchflat.Tasks.SourceTasks do Returns [%MediaItem{}, ...] | [%Ecto.Changeset{}, ...] """ - def index_media_items(%Source{} = source) do + def index_and_enqueue_download_for_media_items(%Source{} = source) 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_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) - Enum.map(media_attributes, fn media_attrs -> - create_media_item_from_attributes(source, media_attrs) - end) + result end @doc """ @@ -76,8 +83,6 @@ defmodule Pinchflat.Tasks.SourceTasks do that any stragglers are caught if, for some reason, they weren't enqueued or somehow got de-queued. - I'm not sure of a case where this would happen, but it's cheap insurance. - Returns :ok """ def enqueue_pending_media_tasks(%Source{download_media: true} = source) do @@ -123,8 +128,8 @@ defmodule Pinchflat.Tasks.SourceTasks do {:ok, pid} = FileFollowerServer.start_link() handler = fn filepath -> setup_file_follower_watcher(pid, filepath, source) end - result = SourceDetails.get_media_attributes(source.original_url, file_listener_handler: handler) + FileFollowerServer.stop(pid) result @@ -136,7 +141,7 @@ defmodule Pinchflat.Tasks.SourceTasks do {:ok, media_attrs} -> Logger.debug("FileFollowerServer Handler: Got media attributes: #{inspect(media_attrs)}") - create_media_item_from_attributes(source, media_attrs) + create_media_item_and_enqueue_download(source, media_attrs) err -> Logger.debug("FileFollowerServer Handler: Error decoding JSON: #{inspect(err)}") @@ -146,6 +151,25 @@ defmodule Pinchflat.Tasks.SourceTasks do end) end + defp create_media_item_and_enqueue_download(source, media_attrs) do + maybe_media_item = create_media_item_from_attributes(source, media_attrs) + + case maybe_media_item do + %MediaItem{} = media_item -> + if source.download_media && Media.pending_download?(media_item) do + Logger.debug("FileFollowerServer Handler: Enqueuing download task for #{inspect(media_attrs)}") + + media_item + |> Map.take([:id]) + |> VideoDownloadWorker.new() + |> Tasks.create_job_with_task(media_item) + end + + changeset -> + changeset + end + end + defp create_media_item_from_attributes(source, media_attrs) do attrs = %{ source_id: source.id, diff --git a/lib/pinchflat/workers/media_indexing_worker.ex b/lib/pinchflat/workers/media_indexing_worker.ex index ae54ac1..d2ff676 100644 --- a/lib/pinchflat/workers/media_indexing_worker.ex +++ b/lib/pinchflat/workers/media_indexing_worker.ex @@ -43,13 +43,14 @@ defmodule Pinchflat.Workers.MediaIndexingWorker do case {source.index_frequency_minutes, source.last_indexed_at} do {index_freq, _} when index_freq > 0 -> # If the indexing is on a schedule simply run indexing and reschedule - index_media(source) + SourceTasks.index_and_enqueue_download_for_media_items(source) reschedule_indexing(source) {_, nil} -> # If the source has never been indexed, index it once # even if it's not meant to reschedule - index_media(source) + SourceTasks.index_and_enqueue_download_for_media_items(source) + :ok _ -> # If the source HAS been indexed and is not meant to reschedule, @@ -58,12 +59,6 @@ defmodule Pinchflat.Workers.MediaIndexingWorker do end end - defp index_media(source) do - SourceTasks.index_media_items(source) - # This method handles the case where a source is set to not download media - SourceTasks.enqueue_pending_media_tasks(source) - end - defp reschedule_indexing(source) do source |> Map.take([:id]) diff --git a/test/pinchflat/media_test.exs b/test/pinchflat/media_test.exs index 15ca76d..fc5521f 100644 --- a/test/pinchflat/media_test.exs +++ b/test/pinchflat/media_test.exs @@ -215,6 +215,34 @@ defmodule Pinchflat.MediaTest do end end + describe "pending_download?/1" do + test "returns true when the media hasn't been downloaded" do + media_item = media_item_fixture(%{media_filepath: nil}) + + assert Media.pending_download?(media_item) + end + + test "returns false if the media has been downloaded" do + media_item = media_item_fixture(%{media_filepath: "/video/#{Faker.File.file_name(:video)}"}) + + refute Media.pending_download?(media_item) + end + + test "returns false if the media hasn't been downloaded but the profile doesn't DL shorts" do + source = source_fixture(%{media_profile_id: media_profile_fixture(%{shorts_behaviour: :exclude}).id}) + media_item = media_item_fixture(%{source_id: source.id, media_filepath: nil, original_url: "/shorts/"}) + + refute Media.pending_download?(media_item) + end + + test "returns false if the media hasn't been downloaded but the profile doesn't DL livestreams" do + source = source_fixture(%{media_profile_id: media_profile_fixture(%{livestream_behaviour: :exclude}).id}) + media_item = media_item_fixture(%{source_id: source.id, media_filepath: nil, livestream: true}) + + refute Media.pending_download?(media_item) + end + end + describe "search/1" do setup do media_item = diff --git a/test/pinchflat/tasks/source_tasks_test.exs b/test/pinchflat/tasks/source_tasks_test.exs index 19d7936..7f5e988 100644 --- a/test/pinchflat/tasks/source_tasks_test.exs +++ b/test/pinchflat/tasks/source_tasks_test.exs @@ -5,6 +5,7 @@ defmodule Pinchflat.Tasks.SourceTasksTest do import Pinchflat.TasksFixtures import Pinchflat.MediaFixtures import Pinchflat.SourcesFixtures + import Pinchflat.ProfilesFixtures alias Pinchflat.Tasks alias Pinchflat.Tasks.Task @@ -43,7 +44,7 @@ defmodule Pinchflat.Tasks.SourceTasksTest do end end - describe "index_media_items/1" do + describe "index_and_enqueue_download_for_media_items/1" do setup do stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, source_attributes_return_fixture()} @@ -53,7 +54,7 @@ defmodule Pinchflat.Tasks.SourceTasksTest do end test "it creates a media_item record for each media ID returned", %{source: source} do - assert media_items = SourceTasks.index_media_items(source) + assert media_items = SourceTasks.index_and_enqueue_download_for_media_items(source) assert Enum.count(media_items) == 3 assert ["video1", "video2", "video3"] == Enum.map(media_items, & &1.media_id) @@ -65,15 +66,15 @@ defmodule Pinchflat.Tasks.SourceTasksTest do test "it attaches all media_items to the given source", %{source: source} do source_id = source.id - assert media_items = SourceTasks.index_media_items(source) + assert media_items = SourceTasks.index_and_enqueue_download_for_media_items(source) assert Enum.count(media_items) == 3 assert Enum.all?(media_items, fn %MediaItem{source_id: ^source_id} -> true end) end test "it won't duplicate media_items based on media_id and source", %{source: source} do - _first_run = SourceTasks.index_media_items(source) - _duplicate_run = SourceTasks.index_media_items(source) + _first_run = SourceTasks.index_and_enqueue_download_for_media_items(source) + _duplicate_run = SourceTasks.index_and_enqueue_download_for_media_items(source) media_items = Repo.preload(source, :media_items).media_items assert Enum.count(media_items) == 3 @@ -82,8 +83,8 @@ defmodule Pinchflat.Tasks.SourceTasksTest do test "it can duplicate media_ids for different sources", %{source: source} do other_source = source_fixture() - media_items = SourceTasks.index_media_items(source) - media_items_other_source = SourceTasks.index_media_items(other_source) + media_items = SourceTasks.index_and_enqueue_download_for_media_items(source) + media_items_other_source = SourceTasks.index_and_enqueue_download_for_media_items(other_source) assert Enum.count(media_items) == 3 assert Enum.count(media_items_other_source) == 3 @@ -93,8 +94,8 @@ defmodule Pinchflat.Tasks.SourceTasksTest do end test "it returns a list of media_items or changesets", %{source: source} do - first_run = SourceTasks.index_media_items(source) - duplicate_run = SourceTasks.index_media_items(source) + first_run = SourceTasks.index_and_enqueue_download_for_media_items(source) + duplicate_run = SourceTasks.index_and_enqueue_download_for_media_items(source) assert Enum.all?(first_run, fn %MediaItem{} -> true end) assert Enum.all?(duplicate_run, fn %Ecto.Changeset{} -> true end) @@ -103,19 +104,37 @@ defmodule Pinchflat.Tasks.SourceTasksTest do test "it updates the source's last_indexed_at field", %{source: source} do assert source.last_indexed_at == nil - SourceTasks.index_media_items(source) + SourceTasks.index_and_enqueue_download_for_media_items(source) source = Repo.reload!(source) assert DateTime.diff(DateTime.utc_now(), source.last_indexed_at) < 2 end + + test "it enqueues a job for each pending media item" do + source = source_fixture() + media_item = media_item_fixture(source_id: source.id, media_filepath: nil) + + SourceTasks.index_and_enqueue_download_for_media_items(source) + + assert_enqueued(worker: VideoDownloadWorker, args: %{"id" => media_item.id}) + end + + test "it does not attach tasks if the source is set to not download" do + source = source_fixture(download_media: false) + media_item = media_item_fixture(source_id: source.id, media_filepath: nil) + + SourceTasks.index_and_enqueue_download_for_media_items(source) + + assert [] = Tasks.list_tasks_for(:media_item_id, media_item.id) + end end - describe "index_media_items/1 when testing file watcher" do + describe "index_and_enqueue_download_for_media_items/1 when testing file watcher" do setup do {:ok, [source: source_fixture()]} end - test "it creates a new media item for everything already in the file", %{source: source} do + test "creates a new media item for everything already in the file", %{source: source} do watcher_poll_interval = Application.get_env(:pinchflat, :file_watcher_poll_interval) stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot, addl_opts -> @@ -124,14 +143,83 @@ defmodule Pinchflat.Tasks.SourceTasksTest do # Need to add a delay to ensure the file watcher has time to read the file :timer.sleep(watcher_poll_interval * 2) - + # We know we're testing the file watcher since the syncronous call will only + # return an empty string (creating no records) {:ok, ""} end) assert Repo.aggregate(MediaItem, :count, :id) == 0 - SourceTasks.index_media_items(source) + SourceTasks.index_and_enqueue_download_for_media_items(source) assert Repo.aggregate(MediaItem, :count, :id) == 3 end + + test "enqueues a download for everything already in the file", %{source: source} do + watcher_poll_interval = Application.get_env(:pinchflat, :file_watcher_poll_interval) + + stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot, addl_opts -> + filepath = Keyword.get(addl_opts, :output_filepath) + File.write(filepath, source_attributes_return_fixture()) + + # Need to add a delay to ensure the file watcher has time to read the file + :timer.sleep(watcher_poll_interval * 2) + # We know we're testing the file watcher since the syncronous call will only + # return an empty string (creating no records) + {:ok, ""} + end) + + refute_enqueued(worker: VideoDownloadWorker) + SourceTasks.index_and_enqueue_download_for_media_items(source) + assert_enqueued(worker: VideoDownloadWorker) + end + + test "does not enqueue downloads if the source is set to not download" do + watcher_poll_interval = Application.get_env(:pinchflat, :file_watcher_poll_interval) + source = source_fixture(download_media: false) + + stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot, addl_opts -> + filepath = Keyword.get(addl_opts, :output_filepath) + File.write(filepath, source_attributes_return_fixture()) + + # Need to add a delay to ensure the file watcher has time to read the file + :timer.sleep(watcher_poll_interval * 2) + # We know we're testing the file watcher since the syncronous call will only + # return an empty string (creating no records) + {:ok, ""} + end) + + SourceTasks.index_and_enqueue_download_for_media_items(source) + refute_enqueued(worker: VideoDownloadWorker) + end + + test "does not enqueue downloads for media that doesn't match the profile's format options" do + watcher_poll_interval = Application.get_env(:pinchflat, :file_watcher_poll_interval) + profile = media_profile_fixture(%{shorts_behaviour: :exclude}) + source = source_fixture(%{media_profile_id: profile.id}) + + stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot, addl_opts -> + filepath = Keyword.get(addl_opts, :output_filepath) + + contents = + Phoenix.json_library().encode!(%{ + id: "video2", + title: "Video 2", + original_url: "https://example.com/shorts/video2", + was_live: true, + description: "desc2" + }) + + File.write(filepath, contents) + + # Need to add a delay to ensure the file watcher has time to read the file + :timer.sleep(watcher_poll_interval * 2) + # We know we're testing the file watcher since the syncronous call will only + # return an empty string (creating no records) + {:ok, ""} + end) + + SourceTasks.index_and_enqueue_download_for_media_items(source) + refute_enqueued(worker: VideoDownloadWorker) + end end describe "enqueue_pending_media_tasks/1" do