[WIP] more contexts, this time around slow indexing and downloads

lib/pinchflat/application.ex

@@ -12,7 +12,7 @@ defmodule Pinchflat.Application do
       Pinchflat.Repo,
       # Must be before startup tasks
       {Oban, Application.fetch_env!(:pinchflat, Oban)},
-      Pinchflat.StartupTasks,
+      Pinchflat.Boot.StartupTasks,
       {DNSCluster, query: Application.get_env(:pinchflat, :dns_cluster_query) || :ignore},
       {Phoenix.PubSub, name: Pinchflat.PubSub},
       # Start the Finch HTTP client for sending emails

lib/pinchflat/boot/data_backfill_worker.ex

@@ -1,4 +1,4 @@
-defmodule Pinchflat.Workers.DataBackfillWorker do
+defmodule Pinchflat.Boot.DataBackfillWorker do
   @moduledoc false
 
   use Oban.Worker,
@@ -28,7 +28,7 @@ defmodule Pinchflat.Workers.DataBackfillWorker do
   """
   def cancel_pending_backfill_jobs do
     Oban.Job
-    |> where(worker: "Pinchflat.Workers.DataBackfillWorker")
+    |> where(worker: "Pinchflat.Boot.DataBackfillWorker")
     |> Oban.cancel_all_jobs()
   end
 

lib/pinchflat/boot/startup_tasks.ex

@@ -1,6 +1,7 @@
-defmodule Pinchflat.StartupTasks do
+defmodule Pinchflat.Boot.StartupTasks do
   @moduledoc """
-  This module is responsible for running startup tasks on app boot.
+  This module is responsible for running startup tasks on app boot
+  AFTER the job runner has initiallized.
 
   It's a GenServer because that plays REALLY nicely with the existing
   Phoenix supervision tree.
@@ -12,7 +13,7 @@ defmodule Pinchflat.StartupTasks do
 
   alias Pinchflat.Repo
   alias Pinchflat.Settings
-  alias Pinchflat.Workers.DataBackfillWorker
+  alias Pinchflat.Boot.DataBackfillWorker
 
   def start_link(opts \\ []) do
     GenServer.start_link(__MODULE__, %{}, opts)

lib/pinchflat/downloading/downloading_helpers.ex

@@ -0,0 +1,54 @@
+defmodule Pinchflat.Downloading.DownloadingHelpers do
+  require Logger
+
+  alias Pinchflat.Media
+  alias Pinchflat.Tasks
+  alias Pinchflat.Sources
+  alias Pinchflat.Sources.Source
+  alias Pinchflat.FastIndexing.YoutubeRss
+  alias Pinchflat.Media.MediaItem
+  alias Pinchflat.FastIndexing.FastIndexingWorker
+  alias Pinchflat.Downloading.MediaDownloadWorker
+  alias Pinchflat.FastIndexing.MediaIndexingWorker
+  alias Pinchflat.YtDlp.Backend.MediaCollection
+  alias Pinchflat.SlowIndexing.MediaCollectionIndexingWorker
+  alias Pinchflat.Utils.FilesystemUtils.FileFollowerServer
+
+  alias Pinchflat.YtDlp.Backend.Media, as: YtDlpMedia
+
+  @doc """
+  Starts tasks for downloading media for any of a sources _pending_ media items.
+  Jobs are not enqueued if the source is set to not download media. This will return :ok.
+
+  NOTE: this starts a download for each media item that is pending,
+  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.
+
+  Returns :ok
+  """
+  def enqueue_pending_download_tasks(%Source{download_media: true} = source) do
+    source
+    |> Media.list_pending_media_items_for()
+    |> Enum.each(fn media_item ->
+      %{id: media_item.id}
+      |> MediaDownloadWorker.new()
+      |> Tasks.create_job_with_task(media_item)
+    end)
+  end
+
+  def enqueue_pending_download_tasks(%Source{download_media: false}) do
+    :ok
+  end
+
+  @doc """
+  Deletes ALL pending tasks for a source's media items.
+
+  Returns :ok
+  """
+  def dequeue_pending_download_tasks(%Source{} = source) do
+    source
+    |> Media.list_pending_media_items_for()
+    |> Enum.each(&Tasks.delete_pending_tasks_for/1)
+  end
+end

lib/pinchflat/downloading/media_download_worker.ex

@@ -1,4 +1,4 @@
-defmodule Pinchflat.Workers.MediaDownloadWorker do
+defmodule Pinchflat.Downloading.MediaDownloadWorker do
   @moduledoc false
 
   use Oban.Worker,

lib/pinchflat/fast_indexing/fast_indexing_helpers.ex

@@ -6,10 +6,10 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do
   alias Pinchflat.FastIndexing.YoutubeRss
   alias Pinchflat.Media.MediaItem
   alias Pinchflat.FastIndexing.FastIndexingWorker
-  alias Pinchflat.Workers.MediaDownloadWorker
+  alias Pinchflat.Downloading.MediaDownloadWorker
-  alias Pinchflat.Workers.MediaIndexingWorker
+  alias Pinchflat.FastIndexing.MediaIndexingWorker
   alias Pinchflat.YtDlp.Backend.MediaCollection
-  alias Pinchflat.Workers.MediaCollectionIndexingWorker
+  alias Pinchflat.SlowIndexing.MediaCollectionIndexingWorker
   alias Pinchflat.Utils.FilesystemUtils.FileFollowerServer
 
   @doc """

lib/pinchflat/fast_indexing/media_indexing_worker.ex

@@ -1,6 +1,8 @@
-defmodule Pinchflat.Workers.MediaIndexingWorker do
+defmodule Pinchflat.FastIndexing.MediaIndexingWorker do
   @moduledoc false
 
+  # TODO: make a startup task to rename all existing workers so they still run
+
   use Oban.Worker,
     queue: :media_indexing,
     unique: [period: :infinity, states: [:available, :scheduled, :retryable]],

lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex

@@ -1,4 +1,4 @@
-defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do
+defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorker do
   @moduledoc false
 
   use Oban.Worker,
@@ -12,6 +12,7 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do
   alias Pinchflat.Sources.Source
   alias Pinchflat.Tasks.SourceTasks
   alias Pinchflat.FastIndexing.FastIndexingWorker
+  alias Pinchflat.SlowIndexing.SlowIndexingHelpers
 
   @impl Oban.Worker
   @doc """
@@ -40,7 +41,7 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do
        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)
+       for a little speedup (see {Fast,Slow}IndexingHelpers 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
@@ -56,8 +57,6 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do
   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?
 
-  IDEA: Should I use paging and do indexing in chunks? Is that even faster?
-
   Returns :ok | {:ok, %Task{}}
   """
   def perform(%Oban.Job{args: %{"id" => source_id}}) do
@@ -66,14 +65,14 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker 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
-        SourceTasks.index_and_enqueue_download_for_media_items(source)
+        SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
         maybe_enqueue_fast_indexing_task(source)
         reschedule_indexing(source)
 
       {_, nil} ->
         # If the source has never been indexed, index it once
         # even if it's not meant to reschedule
-        SourceTasks.index_and_enqueue_download_for_media_items(source)
+        SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
         :ok
 
       _ ->

lib/pinchflat/slow_indexing/slow_indexing_helpers.ex

@@ -0,0 +1,133 @@
+defmodule Pinchflat.SlowIndexing.SlowIndexingHelpers do
+  require Logger
+
+  alias Pinchflat.Media
+  alias Pinchflat.Tasks
+  alias Pinchflat.Sources
+  alias Pinchflat.Sources.Source
+  alias Pinchflat.FastIndexing.YoutubeRss
+  alias Pinchflat.Media.MediaItem
+  alias Pinchflat.FastIndexing.FastIndexingWorker
+  alias Pinchflat.Downloading.MediaDownloadWorker
+  alias Pinchflat.FastIndexing.MediaIndexingWorker
+  alias Pinchflat.YtDlp.Backend.MediaCollection
+  alias Pinchflat.SlowIndexing.MediaCollectionIndexingWorker
+  alias Pinchflat.Utils.FilesystemUtils.FileFollowerServer
+  alias Pinchflat.Downloading.DownloadingHelpers
+
+  alias Pinchflat.YtDlp.Backend.Media, as: YtDlpMedia
+
+  @doc """
+  Starts tasks for indexing a source's media regardless of the source's indexing
+  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}
+    # Schedule this one immediately, but future ones will be on an interval
+    |> MediaCollectionIndexingWorker.new()
+    |> Tasks.create_job_with_task(source)
+  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
+  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
+  so we can create media items as they come in. This parallelizes the process and adds
+  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 that that opportunity to update
+  indexing metadata for media items that have already been created.
+
+  Returns [%MediaItem{}, ...]
+  """
+  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_for_collection_and_setup_file_watcher(source)
+
+    result =
+      Enum.map(media_attributes, fn media_attrs ->
+        case Media.create_media_item_from_backend_attrs(source, media_attrs) do
+          {:ok, media_item} -> media_item
+          {:error, changeset} -> changeset
+        end
+      end)
+
+    Sources.update_source(source, %{last_indexed_at: DateTime.utc_now()})
+    DownloadingHelpers.enqueue_pending_download_tasks(source)
+
+    result
+  end
+
+  # The file follower is a GenServer that watches a file for new lines and
+  # processes them. This works well, but we have to be resilliant to partially-written
+  # lines (ie: you should gracefully fail if you can't parse a line).
+  #
+  # This works in-tandem with the normal (blocking) media indexing behaviour. When
+  # the `get_media_attributes_for_collection` method completes it'll return the FULL result to
+  # the caller for parsing. Ideally, every item in the list will have already
+  # been processed by the file follower, but if not, the caller handles creation
+  # of any media items that were missed/initially failed.
+  #
+  # It attempts a graceful shutdown of the file follower after the indexing is done,
+  # but the FileFollowerServer will also stop itself if it doesn't see any activity
+  # for a sufficiently long time.
+  defp get_media_attributes_for_collection_and_setup_file_watcher(source) do
+    {:ok, pid} = FileFollowerServer.start_link()
+
+    handler = fn filepath -> setup_file_follower_watcher(pid, filepath, source) end
+    result = MediaCollection.get_media_attributes_for_collection(source.original_url, file_listener_handler: handler)
+
+    FileFollowerServer.stop(pid)
+
+    result
+  end
+
+  defp setup_file_follower_watcher(pid, filepath, source) do
+    FileFollowerServer.watch_file(pid, filepath, fn line ->
+      case Phoenix.json_library().decode(line) do
+        {:ok, media_attrs} ->
+          Logger.debug("FileFollowerServer Handler: Got media attributes: #{inspect(media_attrs)}")
+
+          media_struct = YtDlpMedia.response_to_struct(media_attrs)
+          create_media_item_and_enqueue_download(source, media_struct)
+
+        err ->
+          Logger.debug("FileFollowerServer Handler: Error decoding JSON: #{inspect(err)}")
+
+          err
+      end
+    end)
+  end
+
+  defp create_media_item_and_enqueue_download(source, media_attrs) do
+    case Media.create_media_item_from_backend_attrs(source, media_attrs) do
+      {:ok, %MediaItem{} = media_item} ->
+        if source.download_media && Media.pending_download?(media_item) do
+          Logger.debug("FileFollowerServer Handler: Enqueuing download task for #{inspect(media_attrs)}")
+
+          %{id: media_item.id}
+          |> MediaDownloadWorker.new()
+          |> Tasks.create_job_with_task(media_item)
+        end
+
+      {:error, changeset} ->
+        changeset
+    end
+  end
+end

lib/pinchflat/sources.ex

@@ -12,7 +12,9 @@ defmodule Pinchflat.Sources do
   alias Pinchflat.Tasks.SourceTasks
   alias Pinchflat.Profiles.MediaProfile
   alias Pinchflat.YtDlp.Backend.MediaCollection
+  alias Pinchflat.Downloading.DownloadingHelpers
   alias Pinchflat.FastIndexing.FastIndexingHelpers
+  alias Pinchflat.SlowIndexing.SlowIndexingHelpers
 
   @doc """
   Returns the list of sources. Returns [%Source{}, ...]
@@ -68,7 +70,7 @@ defmodule Pinchflat.Sources do
   media if the indexing frequency has been changed.
 
   Existing indexing tasks will be cancelled if the indexing frequency has been
-  changed (logic in `SourceTasks.kickoff_indexing_task`)
+  changed (logic in `SlowIndexingHelpers.kickoff_indexing_task`)
 
   Runs an initial `change_source` check to ensure most of the source is valid
   before making an expensive API call. Runs it through `Repo.update` even
@@ -206,10 +208,10 @@ defmodule Pinchflat.Sources do
   defp maybe_handle_media_tasks(changeset, source) do
     case {changeset.data, changeset.changes} do
       {%{__meta__: %{state: :loaded}}, %{download_media: true}} ->
-        SourceTasks.enqueue_pending_media_tasks(source)
+        DownloadingHelpers.enqueue_pending_download_tasks(source)
 
       {%{__meta__: %{state: :loaded}}, %{download_media: false}} ->
-        SourceTasks.dequeue_pending_media_tasks(source)
+        DownloadingHelpers.dequeue_pending_download_tasks(source)
 
       _ ->
         :ok
@@ -222,7 +224,7 @@ defmodule Pinchflat.Sources do
     case changeset.data do
       # If the changeset is new (not persisted), attempt indexing no matter what
       %{__meta__: %{state: :built}} ->
-        SourceTasks.kickoff_indexing_task(source)
+        SlowIndexingHelpers.kickoff_indexing_task(source)
 
       # If the record has been persisted, only run indexing if the
       # indexing frequency has been changed and is now greater than 0
@@ -237,7 +239,7 @@ defmodule Pinchflat.Sources do
   defp maybe_update_slow_indexing_task(changeset, source) do
     case changeset.changes do
       %{index_frequency_minutes: mins} when mins > 0 ->
-        SourceTasks.kickoff_indexing_task(source)
+        SlowIndexingHelpers.kickoff_indexing_task(source)
 
       %{index_frequency_minutes: _} ->
         Tasks.delete_pending_tasks_for(source, "FastIndexingWorker")

lib/pinchflat/tasks/media_item_tasks.ex

@@ -8,7 +8,7 @@ defmodule Pinchflat.Tasks.MediaItemTasks do
   alias Pinchflat.Media
   alias Pinchflat.Tasks
   alias Pinchflat.Sources.Source
-  alias Pinchflat.Workers.MediaDownloadWorker
+  alias Pinchflat.Downloading.MediaDownloadWorker
 
   alias Pinchflat.YtDlp.Backend.Media, as: YtDlpMedia
 

lib/pinchflat/tasks/source_tasks.ex

@@ -8,170 +8,18 @@ defmodule Pinchflat.Tasks.SourceTasks do
 
   require Logger
 
-  alias Pinchflat.Media
+  # alias Pinchflat.Media
-  alias Pinchflat.Tasks
+  # alias Pinchflat.Tasks
-  alias Pinchflat.Sources
+  # alias Pinchflat.Sources
-  alias Pinchflat.Sources.Source
+  # alias Pinchflat.Sources.Source
-  alias Pinchflat.FastIndexing.YoutubeRss
+  # alias Pinchflat.FastIndexing.YoutubeRss
-  alias Pinchflat.Media.MediaItem
+  # alias Pinchflat.Media.MediaItem
-  alias Pinchflat.FastIndexing.FastIndexingWorker
+  # alias Pinchflat.FastIndexing.FastIndexingWorker
-  alias Pinchflat.Workers.MediaDownloadWorker
+  # alias Pinchflat.Downloading.MediaDownloadWorker
-  alias Pinchflat.Workers.MediaIndexingWorker
+  # alias Pinchflat.FastIndexing.MediaIndexingWorker
-  alias Pinchflat.YtDlp.Backend.MediaCollection
+  # alias Pinchflat.YtDlp.Backend.MediaCollection
-  alias Pinchflat.Workers.MediaCollectionIndexingWorker
+  # alias Pinchflat.SlowIndexing.MediaCollectionIndexingWorker
-  alias Pinchflat.Utils.FilesystemUtils.FileFollowerServer
+  # alias Pinchflat.Utils.FilesystemUtils.FileFollowerServer
 
-  alias Pinchflat.YtDlp.Backend.Media, as: YtDlpMedia
+  # alias Pinchflat.YtDlp.Backend.Media, as: YtDlpMedia
-
-  @doc """
-  Starts tasks for indexing a source's media regardless of the source's indexing
-  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}
-    # Schedule this one immediately, but future ones will be on an interval
-    |> MediaCollectionIndexingWorker.new()
-    |> Tasks.create_job_with_task(source)
-  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
-  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
-  so we can create media items as they come in. This parallelizes the process and adds
-  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 that that opportunity to update
-  indexing metadata for media items that have already been created.
-
-  Returns [%MediaItem{}, ...]
-  """
-  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_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)
-
-    result
-  end
-
-  @doc """
-  Starts tasks for downloading media for any of a sources _pending_ media items.
-  Jobs are not enqueued if the source is set to not download media. This will return :ok.
-
-  NOTE: this starts a download for each media item that is pending,
-  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.
-
-  Returns :ok
-  """
-  def enqueue_pending_media_tasks(%Source{download_media: true} = source) do
-    source
-    |> Media.list_pending_media_items_for()
-    |> Enum.each(fn media_item ->
-      %{id: media_item.id}
-      |> MediaDownloadWorker.new()
-      |> Tasks.create_job_with_task(media_item)
-    end)
-  end
-
-  def enqueue_pending_media_tasks(%Source{download_media: false}) do
-    :ok
-  end
-
-  @doc """
-  Deletes ALL pending tasks for a source's media items.
-
-  Returns :ok
-  """
-  def dequeue_pending_media_tasks(%Source{} = source) do
-    source
-    |> Media.list_pending_media_items_for()
-    |> Enum.each(&Tasks.delete_pending_tasks_for/1)
-  end
-
-  # The file follower is a GenServer that watches a file for new lines and
-  # processes them. This works well, but we have to be resilliant to partially-written
-  # lines (ie: you should gracefully fail if you can't parse a line).
-  #
-  # This works in-tandem with the normal (blocking) media indexing behaviour. When
-  # the `get_media_attributes_for_collection` method completes it'll return the FULL result to
-  # the caller for parsing. Ideally, every item in the list will have already
-  # been processed by the file follower, but if not, the caller handles creation
-  # of any media items that were missed/initially failed.
-  #
-  # It attempts a graceful shutdown of the file follower after the indexing is done,
-  # but the FileFollowerServer will also stop itself if it doesn't see any activity
-  # for a sufficiently long time.
-  defp get_media_attributes_for_collection_and_setup_file_watcher(source) do
-    {:ok, pid} = FileFollowerServer.start_link()
-
-    handler = fn filepath -> setup_file_follower_watcher(pid, filepath, source) end
-    result = MediaCollection.get_media_attributes_for_collection(source.original_url, file_listener_handler: handler)
-
-    FileFollowerServer.stop(pid)
-
-    result
-  end
-
-  defp setup_file_follower_watcher(pid, filepath, source) do
-    FileFollowerServer.watch_file(pid, filepath, fn line ->
-      case Phoenix.json_library().decode(line) do
-        {:ok, media_attrs} ->
-          Logger.debug("FileFollowerServer Handler: Got media attributes: #{inspect(media_attrs)}")
-
-          media_struct = YtDlpMedia.response_to_struct(media_attrs)
-          create_media_item_and_enqueue_download(source, media_struct)
-
-        err ->
-          Logger.debug("FileFollowerServer Handler: Error decoding JSON: #{inspect(err)}")
-
-          err
-      end
-    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)}")
-
-          %{id: media_item.id}
-          |> MediaDownloadWorker.new()
-          |> Tasks.create_job_with_task(media_item)
-        end
-
-      changeset ->
-        changeset
-    end
-  end
-
-  defp create_media_item_from_attributes(source, media_attrs) do
-    case Media.create_media_item_from_backend_attrs(source, media_attrs) do
-      {:ok, media_item} -> media_item
-      {:error, changeset} -> changeset
-    end
-  end
 end

test/pinchflat/boot/data_backfill_worker_test.exs

@@ -1,9 +1,9 @@
-defmodule Pinchflat.Workers.DataBackfillWorkerTest do
+defmodule Pinchflat.Boot.DataBackfillWorkerTest do
   use Pinchflat.DataCase
 
   import Pinchflat.MediaFixtures
 
-  alias Pinchflat.Workers.DataBackfillWorker
+  alias Pinchflat.Boot.DataBackfillWorker
   alias Pinchflat.Workers.FilesystemDataWorker
 
   describe "cancel_pending_backfill_jobs/0" do

test/pinchflat/boot/startup_tasks_test.exs

@@ -1,4 +1,4 @@
-defmodule Pinchflat.StartupTasksTest do
+defmodule Pinchflat.Boot.StartupTasksTest do
   use Pinchflat.DataCase
 
   alias Pinchflat.Settings

test/pinchflat/downloading/downloading_helpers_test.exs

@@ -0,0 +1,83 @@
+defmodule Pinchflat.Downloading.DownloadingHelpersTest do
+  use Pinchflat.DataCase
+
+  import Mox
+  import Pinchflat.TasksFixtures
+  import Pinchflat.MediaFixtures
+  import Pinchflat.SourcesFixtures
+  import Pinchflat.ProfilesFixtures
+
+  alias Pinchflat.Tasks
+  alias Pinchflat.Tasks.Task
+  alias Pinchflat.Media.MediaItem
+  alias Pinchflat.FastIndexing.FastIndexingWorker
+  alias Pinchflat.Downloading.MediaDownloadWorker
+  alias Pinchflat.FastIndexing.MediaIndexingWorker
+  alias Pinchflat.SlowIndexing.SlowIndexingHelpers
+  alias Pinchflat.SlowIndexing.MediaCollectionIndexingWorker
+  alias Pinchflat.Downloading.DownloadingHelpers
+
+  setup :verify_on_exit!
+
+  describe "enqueue_pending_download_tasks/1" do
+    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)
+
+      assert :ok = DownloadingHelpers.enqueue_pending_download_tasks(source)
+
+      assert_enqueued(worker: MediaDownloadWorker, args: %{"id" => media_item.id})
+    end
+
+    test "it does not enqueue a job for media items with a filepath" do
+      source = source_fixture()
+      _media_item = media_item_fixture(source_id: source.id, media_filepath: "some/filepath.mp4")
+
+      assert :ok = DownloadingHelpers.enqueue_pending_download_tasks(source)
+
+      refute_enqueued(worker: MediaDownloadWorker)
+    end
+
+    test "it attaches a task to each enqueued job" do
+      source = source_fixture()
+      media_item = media_item_fixture(source_id: source.id, media_filepath: nil)
+
+      assert [] = Tasks.list_tasks_for(:media_item_id, media_item.id)
+
+      assert :ok = DownloadingHelpers.enqueue_pending_download_tasks(source)
+
+      assert [_] = Tasks.list_tasks_for(:media_item_id, media_item.id)
+    end
+
+    test "it does not create a job if the source is set to not download" do
+      source = source_fixture(download_media: false)
+
+      assert :ok = DownloadingHelpers.enqueue_pending_download_tasks(source)
+
+      refute_enqueued(worker: MediaDownloadWorker)
+    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)
+
+      assert :ok = DownloadingHelpers.enqueue_pending_download_tasks(source)
+      assert [] = Tasks.list_tasks_for(:media_item_id, media_item.id)
+    end
+  end
+
+  describe "dequeue_pending_download_tasks/1" do
+    test "it deletes all pending tasks for a source's media items" do
+      source = source_fixture()
+      media_item = media_item_fixture(source_id: source.id, media_filepath: nil)
+
+      DownloadingHelpers.enqueue_pending_download_tasks(source)
+      assert_enqueued(worker: MediaDownloadWorker, args: %{"id" => media_item.id})
+
+      assert :ok = DownloadingHelpers.dequeue_pending_download_tasks(source)
+
+      refute_enqueued(worker: MediaDownloadWorker)
+      assert [] = Tasks.list_tasks_for(:media_item_id, media_item.id)
+    end
+  end
+end

test/pinchflat/downloading/media_download_worker_test.exs

@@ -1,11 +1,11 @@
-defmodule Pinchflat.Workers.MediaDownloadWorkerTest do
+defmodule Pinchflat.Downloading.MediaDownloadWorkerTest do
   use Pinchflat.DataCase
 
   import Mox
   import Pinchflat.MediaFixtures
 
   alias Pinchflat.Sources
-  alias Pinchflat.Workers.MediaDownloadWorker
+  alias Pinchflat.Downloading.MediaDownloadWorker
   alias Pinchflat.Workers.FilesystemDataWorker
 
   setup :verify_on_exit!

test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs

@@ -11,11 +11,11 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpersTest do
   alias Pinchflat.Tasks.Task
   alias Pinchflat.Tasks.SourceTasks
   alias Pinchflat.Media.MediaItem
-  alias Pinchflat.Workers.MediaDownloadWorker
+  alias Pinchflat.Downloading.MediaDownloadWorker
-  alias Pinchflat.Workers.MediaIndexingWorker
+  alias Pinchflat.FastIndexing.MediaIndexingWorker
   alias Pinchflat.FastIndexing.FastIndexingHelpers
   alias Pinchflat.FastIndexing.FastIndexingWorker
-  alias Pinchflat.Workers.MediaCollectionIndexingWorker
+  alias Pinchflat.SlowIndexing.MediaCollectionIndexingWorker
 
   setup :verify_on_exit!
 

test/pinchflat/fast_indexing/media_indexing_worker_test.exs

@@ -1,4 +1,4 @@
-defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
+defmodule Pinchflat.FastIndexing.MediaIndexingWorkerTest do
   use Pinchflat.DataCase
 
   import Mox
@@ -6,8 +6,8 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
   import Pinchflat.SourcesFixtures
 
   alias Pinchflat.Media.MediaItem
-  alias Pinchflat.Workers.MediaIndexingWorker
+  alias Pinchflat.FastIndexing.MediaIndexingWorker
-  alias Pinchflat.Workers.MediaDownloadWorker
+  alias Pinchflat.Downloading.MediaDownloadWorker
 
   @media_url "https://www.youtube.com/watch?v=1234567890"
 

test/pinchflat/slow_indexing/media_collection_indexing_worker_test.exs

@@ -1,4 +1,4 @@
-defmodule Pinchflat.Workers.MediaCollectionIndexingWorkerTest do
+defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorkerTest do
   use Pinchflat.DataCase
 
   import Mox
@@ -9,8 +9,8 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorkerTest do
   alias Pinchflat.Tasks
   alias Pinchflat.Sources.Source
   alias Pinchflat.FastIndexing.FastIndexingWorker
-  alias Pinchflat.Workers.MediaDownloadWorker
+  alias Pinchflat.Downloading.MediaDownloadWorker
-  alias Pinchflat.Workers.MediaCollectionIndexingWorker
+  alias Pinchflat.SlowIndexing.MediaCollectionIndexingWorker
 
   setup :verify_on_exit!
 

test/pinchflat/slow_indexing/slow_indexing_helpers_test.exs

@@ -0,0 +1,270 @@
+defmodule Pinchflat.SlowIndexing.SlowIndexingHelpersTest do
+  use Pinchflat.DataCase
+
+  import Mox
+  import Pinchflat.TasksFixtures
+  import Pinchflat.MediaFixtures
+  import Pinchflat.SourcesFixtures
+  import Pinchflat.ProfilesFixtures
+
+  alias Pinchflat.Tasks
+  alias Pinchflat.Tasks.Task
+  alias Pinchflat.Media.MediaItem
+  alias Pinchflat.FastIndexing.FastIndexingWorker
+  alias Pinchflat.Downloading.MediaDownloadWorker
+  alias Pinchflat.FastIndexing.MediaIndexingWorker
+  alias Pinchflat.SlowIndexing.SlowIndexingHelpers
+  alias Pinchflat.SlowIndexing.MediaCollectionIndexingWorker
+
+  setup :verify_on_exit!
+
+  describe "kickoff_indexing_task/1" do
+    test "it schedules a job" do
+      source = source_fixture(index_frequency_minutes: 1)
+
+      assert {:ok, _} = SlowIndexingHelpers.kickoff_indexing_task(source)
+
+      assert_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id})
+    end
+
+    test "it creates and attaches a task" do
+      source = source_fixture(index_frequency_minutes: 1)
+
+      assert {:ok, %Task{} = task} = SlowIndexingHelpers.kickoff_indexing_task(source)
+
+      assert task.source_id == source.id
+    end
+
+    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)
+
+      assert {:ok, _} = SlowIndexingHelpers.kickoff_indexing_task(source)
+
+      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, _} = SlowIndexingHelpers.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, _} = SlowIndexingHelpers.kickoff_indexing_task(source)
+
+      assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end
+    end
+  end
+
+  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()}
+      end)
+
+      {:ok, [source: source_fixture()]}
+    end
+
+    test "it creates a media_item record for each media ID returned", %{source: source} do
+      assert media_items = SlowIndexingHelpers.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)
+      assert ["Video 1", "Video 2", "Video 3"] == Enum.map(media_items, & &1.title)
+      assert ["desc1", "desc2", "desc3"] == Enum.map(media_items, & &1.description)
+      assert Enum.all?(media_items, fn mi -> mi.original_url end)
+      assert Enum.all?(media_items, fn %MediaItem{} -> true end)
+    end
+
+    test "it attaches all media_items to the given source", %{source: source} do
+      source_id = source.id
+      assert media_items = SlowIndexingHelpers.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 = SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+      _duplicate_run = SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+
+      media_items = Repo.preload(source, :media_items).media_items
+      assert Enum.count(media_items) == 3
+    end
+
+    test "it can duplicate media_ids for different sources", %{source: source} do
+      other_source = source_fixture()
+
+      media_items = SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+      media_items_other_source = SlowIndexingHelpers.index_and_enqueue_download_for_media_items(other_source)
+
+      assert Enum.count(media_items) == 3
+      assert Enum.count(media_items_other_source) == 3
+
+      assert Enum.map(media_items, & &1.media_id) ==
+               Enum.map(media_items_other_source, & &1.media_id)
+    end
+
+    test "it returns a list of media_items", %{source: source} do
+      first_run = SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+      duplicate_run = SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+
+      first_ids = Enum.map(first_run, & &1.id)
+      duplicate_ids = Enum.map(duplicate_run, & &1.id)
+
+      assert first_ids == duplicate_ids
+    end
+
+    test "it updates the source's last_indexed_at field", %{source: source} do
+      assert source.last_indexed_at == nil
+
+      SlowIndexingHelpers.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)
+
+      SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+
+      assert_enqueued(worker: MediaDownloadWorker, 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)
+
+      SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+
+      assert [] = Tasks.list_tasks_for(:media_item_id, media_item.id)
+    end
+  end
+
+  describe "index_and_enqueue_download_for_media_items/1 when testing file watcher" do
+    setup do
+      {:ok, [source: source_fixture()]}
+    end
+
+    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 ->
+        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)
+
+      assert Repo.aggregate(MediaItem, :count, :id) == 0
+      SlowIndexingHelpers.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: MediaDownloadWorker)
+      SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+      assert_enqueued(worker: MediaDownloadWorker)
+    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)
+
+      SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+      refute_enqueued(worker: MediaDownloadWorker)
+    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",
+            webpage_url: "https://example.com/shorts/video2",
+            was_live: true,
+            description: "desc2",
+            aspect_ratio: 1.67,
+            duration: 345.67,
+            upload_date: "20210101"
+          })
+
+        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)
+
+      SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+      refute_enqueued(worker: MediaDownloadWorker)
+    end
+
+    test "does not enqueue multiple download jobs for the same media items", %{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)
+        # This also returns the final result to the yt-dlp call (like the real usage actually would do)
+        # so it'll attempt to create the media items and enqueue the download jobs based on this as well
+        {:ok, source_attributes_return_fixture()}
+      end)
+
+      SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+      assert Repo.aggregate(MediaItem, :count, :id) == 3
+      assert [_, _, _] = all_enqueued(worker: MediaDownloadWorker)
+    end
+  end
+end

test/pinchflat/sources_test.exs

@@ -9,10 +9,11 @@ defmodule Pinchflat.SourcesTest do
   alias Pinchflat.Sources
   alias Pinchflat.Tasks.SourceTasks
   alias Pinchflat.Sources.Source
+  alias Pinchflat.Downloading.DownloadingHelpers
   alias Pinchflat.FastIndexing.FastIndexingWorker
-  alias Pinchflat.Workers.MediaDownloadWorker
+  alias Pinchflat.Downloading.MediaDownloadWorker
-  alias Pinchflat.Workers.MediaIndexingWorker
+  alias Pinchflat.FastIndexing.MediaIndexingWorker
-  alias Pinchflat.Workers.MediaCollectionIndexingWorker
+  alias Pinchflat.SlowIndexing.MediaCollectionIndexingWorker
 
   @invalid_source_attrs %{name: nil, collection_id: nil}
 
@@ -331,7 +332,7 @@ defmodule Pinchflat.SourcesTest do
       source = source_fixture(download_media: true)
       media_item = media_item_fixture(source_id: source.id, media_filepath: nil)
       update_attrs = %{download_media: false}
-      SourceTasks.enqueue_pending_media_tasks(source)
+      DownloadingHelpers.enqueue_pending_download_tasks(source)
 
       assert_enqueued(worker: MediaDownloadWorker, args: %{"id" => media_item.id})
       assert {:ok, %Source{}} = Sources.update_source(source, update_attrs)

test/pinchflat/tasks/media_items_tasks_test.exs

@@ -9,7 +9,7 @@ defmodule Pinchflat.Tasks.MediaItemTasksTest do
   alias Pinchflat.Tasks
   alias Pinchflat.Media.MediaItem
   alias Pinchflat.Tasks.MediaItemTasks
-  alias Pinchflat.Workers.MediaDownloadWorker
+  alias Pinchflat.Downloading.MediaDownloadWorker
 
   setup :verify_on_exit!
 

test/pinchflat/tasks/source_tasks_test.exs

@@ -12,321 +12,9 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
   alias Pinchflat.Tasks.SourceTasks
   alias Pinchflat.Media.MediaItem
   alias Pinchflat.FastIndexing.FastIndexingWorker
-  alias Pinchflat.Workers.MediaDownloadWorker
+  alias Pinchflat.Downloading.MediaDownloadWorker
-  alias Pinchflat.Workers.MediaIndexingWorker
+  alias Pinchflat.FastIndexing.MediaIndexingWorker
-  alias Pinchflat.Workers.MediaCollectionIndexingWorker
+  alias Pinchflat.SlowIndexing.MediaCollectionIndexingWorker
 
   setup :verify_on_exit!
-
-  describe "kickoff_indexing_task/1" do
-    test "it schedules a job" do
-      source = source_fixture(index_frequency_minutes: 1)
-
-      assert {:ok, _} = SourceTasks.kickoff_indexing_task(source)
-
-      assert_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id})
-    end
-
-    test "it creates and attaches a task" do
-      source = source_fixture(index_frequency_minutes: 1)
-
-      assert {:ok, %Task{} = task} = SourceTasks.kickoff_indexing_task(source)
-
-      assert task.source_id == source.id
-    end
-
-    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)
-
-      assert {:ok, _} = SourceTasks.kickoff_indexing_task(source)
-
-      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 "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()}
-      end)
-
-      {:ok, [source: source_fixture()]}
-    end
-
-    test "it creates a media_item record for each media ID returned", %{source: source} do
-      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)
-      assert ["Video 1", "Video 2", "Video 3"] == Enum.map(media_items, & &1.title)
-      assert ["desc1", "desc2", "desc3"] == Enum.map(media_items, & &1.description)
-      assert Enum.all?(media_items, fn mi -> mi.original_url end)
-      assert Enum.all?(media_items, fn %MediaItem{} -> true end)
-    end
-
-    test "it attaches all media_items to the given source", %{source: source} do
-      source_id = source.id
-      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_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
-    end
-
-    test "it can duplicate media_ids for different sources", %{source: source} do
-      other_source = source_fixture()
-
-      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
-
-      assert Enum.map(media_items, & &1.media_id) ==
-               Enum.map(media_items_other_source, & &1.media_id)
-    end
-
-    test "it returns a list of media_items", %{source: source} do
-      first_run = SourceTasks.index_and_enqueue_download_for_media_items(source)
-      duplicate_run = SourceTasks.index_and_enqueue_download_for_media_items(source)
-
-      first_ids = Enum.map(first_run, & &1.id)
-      duplicate_ids = Enum.map(duplicate_run, & &1.id)
-
-      assert first_ids == duplicate_ids
-    end
-
-    test "it updates the source's last_indexed_at field", %{source: source} do
-      assert source.last_indexed_at == nil
-
-      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: MediaDownloadWorker, 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_and_enqueue_download_for_media_items/1 when testing file watcher" do
-    setup do
-      {:ok, [source: source_fixture()]}
-    end
-
-    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 ->
-        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)
-
-      assert Repo.aggregate(MediaItem, :count, :id) == 0
-      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: MediaDownloadWorker)
-      SourceTasks.index_and_enqueue_download_for_media_items(source)
-      assert_enqueued(worker: MediaDownloadWorker)
-    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: MediaDownloadWorker)
-    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",
-            webpage_url: "https://example.com/shorts/video2",
-            was_live: true,
-            description: "desc2",
-            aspect_ratio: 1.67,
-            duration: 345.67,
-            upload_date: "20210101"
-          })
-
-        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: MediaDownloadWorker)
-    end
-
-    test "does not enqueue multiple download jobs for the same media items", %{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)
-        # This also returns the final result to the yt-dlp call (like the real usage actually would do)
-        # so it'll attempt to create the media items and enqueue the download jobs based on this as well
-        {:ok, source_attributes_return_fixture()}
-      end)
-
-      SourceTasks.index_and_enqueue_download_for_media_items(source)
-      assert Repo.aggregate(MediaItem, :count, :id) == 3
-      assert [_, _, _] = all_enqueued(worker: MediaDownloadWorker)
-    end
-  end
-
-  describe "enqueue_pending_media_tasks/1" do
-    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)
-
-      assert :ok = SourceTasks.enqueue_pending_media_tasks(source)
-
-      assert_enqueued(worker: MediaDownloadWorker, args: %{"id" => media_item.id})
-    end
-
-    test "it does not enqueue a job for media items with a filepath" do
-      source = source_fixture()
-      _media_item = media_item_fixture(source_id: source.id, media_filepath: "some/filepath.mp4")
-
-      assert :ok = SourceTasks.enqueue_pending_media_tasks(source)
-
-      refute_enqueued(worker: MediaDownloadWorker)
-    end
-
-    test "it attaches a task to each enqueued job" do
-      source = source_fixture()
-      media_item = media_item_fixture(source_id: source.id, media_filepath: nil)
-
-      assert [] = Tasks.list_tasks_for(:media_item_id, media_item.id)
-
-      assert :ok = SourceTasks.enqueue_pending_media_tasks(source)
-
-      assert [_] = Tasks.list_tasks_for(:media_item_id, media_item.id)
-    end
-
-    test "it does not create a job if the source is set to not download" do
-      source = source_fixture(download_media: false)
-
-      assert :ok = SourceTasks.enqueue_pending_media_tasks(source)
-
-      refute_enqueued(worker: MediaDownloadWorker)
-    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)
-
-      assert :ok = SourceTasks.enqueue_pending_media_tasks(source)
-      assert [] = Tasks.list_tasks_for(:media_item_id, media_item.id)
-    end
-  end
-
-  describe "dequeue_pending_media_tasks/1" do
-    test "it deletes all pending tasks for a source's media items" do
-      source = source_fixture()
-      media_item = media_item_fixture(source_id: source.id, media_filepath: nil)
-
-      SourceTasks.enqueue_pending_media_tasks(source)
-      assert_enqueued(worker: MediaDownloadWorker, args: %{"id" => media_item.id})
-
-      assert :ok = SourceTasks.dequeue_pending_media_tasks(source)
-
-      refute_enqueued(worker: MediaDownloadWorker)
-      assert [] = Tasks.list_tasks_for(:media_item_id, media_item.id)
-    end
-  end
 end