Updated file watcher to enqueue download; refactored download worker methods

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.

lib/pinchflat/tasks.ex

@@ -2,7 +2,6 @@ defmodule Pinchflat.Tasks do
   @moduledoc """
   The Tasks context.
   """
-
   import Ecto.Query, warn: false
   alias Pinchflat.Repo
 

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 ->
+    result
-      create_media_item_from_attributes(source, media_attrs)
-    end)
   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,

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])

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 =

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)
+      _first_run = SourceTasks.index_and_enqueue_download_for_media_items(source)
-      _duplicate_run = SourceTasks.index_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 = SourceTasks.index_and_enqueue_download_for_media_items(source)
-      media_items_other_source = SourceTasks.index_media_items(other_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)
+      first_run = SourceTasks.index_and_enqueue_download_for_media_items(source)
-      duplicate_run = SourceTasks.index_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