diff --git a/lib/pinchflat/application.ex b/lib/pinchflat/application.ex index 6f838bc..e47699c 100644 --- a/lib/pinchflat/application.ex +++ b/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 diff --git a/lib/pinchflat/workers/data_backfill_worker.ex b/lib/pinchflat/boot/data_backfill_worker.ex similarity index 94% rename from lib/pinchflat/workers/data_backfill_worker.ex rename to lib/pinchflat/boot/data_backfill_worker.ex index 15a7041..6c863e3 100644 --- a/lib/pinchflat/workers/data_backfill_worker.ex +++ b/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 diff --git a/lib/pinchflat/startup_tasks.ex b/lib/pinchflat/boot/startup_tasks.ex similarity index 91% rename from lib/pinchflat/startup_tasks.ex rename to lib/pinchflat/boot/startup_tasks.ex index 8e4be23..191970f 100644 --- a/lib/pinchflat/startup_tasks.ex +++ b/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) diff --git a/lib/pinchflat/downloading/downloading_helpers.ex b/lib/pinchflat/downloading/downloading_helpers.ex new file mode 100644 index 0000000..f0becd2 --- /dev/null +++ b/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 diff --git a/lib/pinchflat/workers/media_download_worker.ex b/lib/pinchflat/downloading/media_download_worker.ex similarity index 96% rename from lib/pinchflat/workers/media_download_worker.ex rename to lib/pinchflat/downloading/media_download_worker.ex index 2b99489..fd51571 100644 --- a/lib/pinchflat/workers/media_download_worker.ex +++ b/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, diff --git a/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex b/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex index 4006fe4..9dc5ac3 100644 --- a/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex +++ b/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.Workers.MediaIndexingWorker + alias Pinchflat.Downloading.MediaDownloadWorker + alias Pinchflat.FastIndexing.MediaIndexingWorker alias Pinchflat.YtDlp.Backend.MediaCollection - alias Pinchflat.Workers.MediaCollectionIndexingWorker + alias Pinchflat.SlowIndexing.MediaCollectionIndexingWorker alias Pinchflat.Utils.FilesystemUtils.FileFollowerServer @doc """ diff --git a/lib/pinchflat/workers/media_indexing_worker.ex b/lib/pinchflat/fast_indexing/media_indexing_worker.ex similarity index 94% rename from lib/pinchflat/workers/media_indexing_worker.ex rename to lib/pinchflat/fast_indexing/media_indexing_worker.ex index dbbdcb0..bc416db 100644 --- a/lib/pinchflat/workers/media_indexing_worker.ex +++ b/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]], diff --git a/lib/pinchflat/workers/media_collection_indexing_worker.ex b/lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex similarity index 92% rename from lib/pinchflat/workers/media_collection_indexing_worker.ex rename to lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex index bf9a356..b086169 100644 --- a/lib/pinchflat/workers/media_collection_indexing_worker.ex +++ b/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 _ -> diff --git a/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex b/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex new file mode 100644 index 0000000..664bf21 --- /dev/null +++ b/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 diff --git a/lib/pinchflat/sources.ex b/lib/pinchflat/sources.ex index 787861f..3158899 100644 --- a/lib/pinchflat/sources.ex +++ b/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") diff --git a/lib/pinchflat/tasks/media_item_tasks.ex b/lib/pinchflat/tasks/media_item_tasks.ex index afcb3d5..4345965 100644 --- a/lib/pinchflat/tasks/media_item_tasks.ex +++ b/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 diff --git a/lib/pinchflat/tasks/source_tasks.ex b/lib/pinchflat/tasks/source_tasks.ex index 2891ff5..a44ba99 100644 --- a/lib/pinchflat/tasks/source_tasks.ex +++ b/lib/pinchflat/tasks/source_tasks.ex @@ -8,170 +8,18 @@ defmodule Pinchflat.Tasks.SourceTasks 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.Workers.MediaDownloadWorker - alias Pinchflat.Workers.MediaIndexingWorker - alias Pinchflat.YtDlp.Backend.MediaCollection - alias Pinchflat.Workers.MediaCollectionIndexingWorker - alias Pinchflat.Utils.FilesystemUtils.FileFollowerServer + # 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 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 + # alias Pinchflat.YtDlp.Backend.Media, as: YtDlpMedia end diff --git a/test/pinchflat/workers/data_backfill_worker_test.exs b/test/pinchflat/boot/data_backfill_worker_test.exs similarity index 94% rename from test/pinchflat/workers/data_backfill_worker_test.exs rename to test/pinchflat/boot/data_backfill_worker_test.exs index 856bfb6..a8e8b4f 100644 --- a/test/pinchflat/workers/data_backfill_worker_test.exs +++ b/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 diff --git a/test/pinchflat/startup_tasks_test.exs b/test/pinchflat/boot/startup_tasks_test.exs similarity index 90% rename from test/pinchflat/startup_tasks_test.exs rename to test/pinchflat/boot/startup_tasks_test.exs index ef06d0a..0fece7c 100644 --- a/test/pinchflat/startup_tasks_test.exs +++ b/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 diff --git a/test/pinchflat/downloading/downloading_helpers_test.exs b/test/pinchflat/downloading/downloading_helpers_test.exs new file mode 100644 index 0000000..d9b2e21 --- /dev/null +++ b/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 diff --git a/test/pinchflat/workers/media_download_worker_test.exs b/test/pinchflat/downloading/media_download_worker_test.exs similarity index 95% rename from test/pinchflat/workers/media_download_worker_test.exs rename to test/pinchflat/downloading/media_download_worker_test.exs index 88fc001..b207ee1 100644 --- a/test/pinchflat/workers/media_download_worker_test.exs +++ b/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! diff --git a/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs b/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs index 5463a5d..c2cdd8d 100644 --- a/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs +++ b/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.Workers.MediaIndexingWorker + alias Pinchflat.Downloading.MediaDownloadWorker + alias Pinchflat.FastIndexing.MediaIndexingWorker alias Pinchflat.FastIndexing.FastIndexingHelpers alias Pinchflat.FastIndexing.FastIndexingWorker - alias Pinchflat.Workers.MediaCollectionIndexingWorker + alias Pinchflat.SlowIndexing.MediaCollectionIndexingWorker setup :verify_on_exit! diff --git a/test/pinchflat/workers/media_indexing_worker_test.exs b/test/pinchflat/fast_indexing/media_indexing_worker_test.exs similarity index 87% rename from test/pinchflat/workers/media_indexing_worker_test.exs rename to test/pinchflat/fast_indexing/media_indexing_worker_test.exs index 9db9b19..96cc932 100644 --- a/test/pinchflat/workers/media_indexing_worker_test.exs +++ b/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.Workers.MediaDownloadWorker + alias Pinchflat.FastIndexing.MediaIndexingWorker + alias Pinchflat.Downloading.MediaDownloadWorker @media_url "https://www.youtube.com/watch?v=1234567890" diff --git a/test/pinchflat/workers/media_collection_indexing_worker_test.exs b/test/pinchflat/slow_indexing/media_collection_indexing_worker_test.exs similarity index 97% rename from test/pinchflat/workers/media_collection_indexing_worker_test.exs rename to test/pinchflat/slow_indexing/media_collection_indexing_worker_test.exs index d720b74..9d1f1ce 100644 --- a/test/pinchflat/workers/media_collection_indexing_worker_test.exs +++ b/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.Workers.MediaCollectionIndexingWorker + alias Pinchflat.Downloading.MediaDownloadWorker + alias Pinchflat.SlowIndexing.MediaCollectionIndexingWorker setup :verify_on_exit! diff --git a/test/pinchflat/slow_indexing/slow_indexing_helpers_test.exs b/test/pinchflat/slow_indexing/slow_indexing_helpers_test.exs new file mode 100644 index 0000000..6d385a5 --- /dev/null +++ b/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 diff --git a/test/pinchflat/sources_test.exs b/test/pinchflat/sources_test.exs index a0903d5..b6a75c4 100644 --- a/test/pinchflat/sources_test.exs +++ b/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.Workers.MediaIndexingWorker - alias Pinchflat.Workers.MediaCollectionIndexingWorker + alias Pinchflat.Downloading.MediaDownloadWorker + alias Pinchflat.FastIndexing.MediaIndexingWorker + 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) diff --git a/test/pinchflat/tasks/media_items_tasks_test.exs b/test/pinchflat/tasks/media_items_tasks_test.exs index 80fcbbd..dba1b6d 100644 --- a/test/pinchflat/tasks/media_items_tasks_test.exs +++ b/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! diff --git a/test/pinchflat/tasks/source_tasks_test.exs b/test/pinchflat/tasks/source_tasks_test.exs index f5e7326..893c35a 100644 --- a/test/pinchflat/tasks/source_tasks_test.exs +++ b/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.Workers.MediaIndexingWorker - alias Pinchflat.Workers.MediaCollectionIndexingWorker + alias Pinchflat.Downloading.MediaDownloadWorker + alias Pinchflat.FastIndexing.MediaIndexingWorker + 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