From c39e30ad0b41b732c8e24f912e2a01b8872ed3c8 Mon Sep 17 00:00:00 2001 From: Kieran Eglin Date: Fri, 15 Mar 2024 09:25:34 -0700 Subject: [PATCH] Refactored the way I call workers --- .../downloading/downloading_helpers.ex | 6 +- .../downloading/media_download_worker.ex | 15 ++++- .../fast_indexing/fast_indexing_helpers.ex | 29 +-------- .../fast_indexing/fast_indexing_worker.ex | 18 ++++-- .../fast_indexing/media_indexing_worker.ex | 15 ++++- .../source_metadata_storage_worker.ex | 7 +- .../media_collection_indexing_worker.ex | 13 +++- .../slow_indexing/slow_indexing_helpers.ex | 9 +-- lib/pinchflat/sources/sources.ex | 19 ++---- .../media_download_worker_test.exs | 13 ++++ .../fast_indexing_helpers_test.exs | 31 --------- .../fast_indexing_worker_test.exs | 17 +++++ .../media_indexing_worker_test.exs | 15 ++++- test/pinchflat/sources_test.exs | 64 ------------------- 14 files changed, 110 insertions(+), 161 deletions(-) diff --git a/lib/pinchflat/downloading/downloading_helpers.ex b/lib/pinchflat/downloading/downloading_helpers.ex index 9217548..1e61545 100644 --- a/lib/pinchflat/downloading/downloading_helpers.ex +++ b/lib/pinchflat/downloading/downloading_helpers.ex @@ -26,11 +26,7 @@ defmodule Pinchflat.Downloading.DownloadingHelpers do 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) + |> Enum.each(&MediaDownloadWorker.kickoff_with_task/1) end def enqueue_pending_download_tasks(%Source{download_media: false}) do diff --git a/lib/pinchflat/downloading/media_download_worker.ex b/lib/pinchflat/downloading/media_download_worker.ex index 9ede5ae..386c1f3 100644 --- a/lib/pinchflat/downloading/media_download_worker.ex +++ b/lib/pinchflat/downloading/media_download_worker.ex @@ -6,17 +6,30 @@ defmodule Pinchflat.Downloading.MediaDownloadWorker do unique: [period: :infinity, states: [:available, :scheduled, :retryable, :executing]], tags: ["media_item", "media_fetching"] + alias __MODULE__ + alias Pinchflat.Tasks alias Pinchflat.Repo alias Pinchflat.Media alias Pinchflat.Downloading.MediaDownloader - @impl Oban.Worker + @doc """ + Starts the media_item media download worker and creates a task for the media_item. + + Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}} + """ + def kickoff_with_task(media_item, opts \\ []) do + %{id: media_item.id} + |> MediaDownloadWorker.new(opts) + |> Tasks.create_job_with_task(media_item) + end + @doc """ For a given media item, download the media alongside any options. Does not download media if its source is set to not download media. Returns :ok | {:ok, %MediaItem{}} | {:error, any, ...any} """ + @impl Oban.Worker def perform(%Oban.Job{args: %{"id" => media_item_id}}) do media_item = media_item_id diff --git a/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex b/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex index 6355df7..d1d72da 100644 --- a/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex +++ b/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex @@ -6,34 +6,13 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do """ alias Pinchflat.Media - alias Pinchflat.Tasks alias Pinchflat.Sources.Source alias Pinchflat.FastIndexing.YoutubeRss - alias Pinchflat.FastIndexing.FastIndexingWorker alias Pinchflat.Downloading.MediaDownloadWorker alias Pinchflat.FastIndexing.MediaIndexingWorker alias Pinchflat.YtDlp.Media, as: YtDlpMedia - @doc """ - Starts tasks for running a fast indexing task for a source's media - regardless of the source's fast_index state. It's assumed the - caller will check for fast_index. - - This is used for running fast index tasks on update. On creation, the - fast index is enqueued after the slow index is complete. - - Returns {:ok, %Task{}}. - """ - def kickoff_fast_indexing_task(%Source{} = source) do - Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") - - %{id: source.id} - # Schedule this one immediately, but future ones will be on an interval - |> FastIndexingWorker.new() - |> Tasks.create_job_with_task(source) - end - @doc """ Fetches new media IDs from a source's YouTube RSS feed and kicks off indexing tasks for any new media items. See comments in `MediaIndexingWorker` for more info on the @@ -54,9 +33,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do Enum.each(new_media_ids, fn media_id -> url = "https://www.youtube.com/watch?v=#{media_id}" - %{id: source.id, media_url: url} - |> MediaIndexingWorker.new() - |> Tasks.create_job_with_task(source) + MediaIndexingWorker.kickoff_with_task(source, url) end) end @@ -74,9 +51,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do case maybe_media_item do {:ok, media_item} -> if source.download_media && Media.pending_download?(media_item) do - %{id: media_item.id} - |> MediaDownloadWorker.new() - |> Tasks.create_job_with_task(media_item) + MediaDownloadWorker.kickoff_with_task(media_item) end {:ok, media_item} diff --git a/lib/pinchflat/fast_indexing/fast_indexing_worker.ex b/lib/pinchflat/fast_indexing/fast_indexing_worker.ex index d0a0e9a..5e16a68 100644 --- a/lib/pinchflat/fast_indexing/fast_indexing_worker.ex +++ b/lib/pinchflat/fast_indexing/fast_indexing_worker.ex @@ -12,7 +12,17 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do alias Pinchflat.Sources.Source alias Pinchflat.FastIndexing.FastIndexingHelpers - @impl Oban.Worker + @doc """ + Starts the source fast indexing worker and creates a task for the source. + + Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}} + """ + def kickoff_with_task(source, opts \\ []) do + %{id: source.id} + |> FastIndexingWorker.new(opts) + |> Tasks.create_job_with_task(source) + end + @doc """ Kicks off the fast indexing process for a source, reschedules the job to run again once complete. See `MediaCollectionIndexingWorker` and `MediaIndexingWorker` comments @@ -20,6 +30,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do Returns :ok | {:ok, :job_exists} | {:ok, %Task{}} """ + @impl Oban.Worker def perform(%Oban.Job{args: %{"id" => source_id}}) do source = Sources.get_source!(source_id) @@ -35,10 +46,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do defp reschedule_indexing(source) do next_run_in = Source.fast_index_frequency() * 60 - %{id: source.id} - |> FastIndexingWorker.new(schedule_in: next_run_in) - |> Tasks.create_job_with_task(source) - |> case do + case kickoff_with_task(source, schedule_in: next_run_in) do {:ok, task} -> {:ok, task} {:error, :duplicate_job} -> {:ok, :job_exists} end diff --git a/lib/pinchflat/fast_indexing/media_indexing_worker.ex b/lib/pinchflat/fast_indexing/media_indexing_worker.ex index 1bcc9d3..8792f60 100644 --- a/lib/pinchflat/fast_indexing/media_indexing_worker.ex +++ b/lib/pinchflat/fast_indexing/media_indexing_worker.ex @@ -8,10 +8,22 @@ defmodule Pinchflat.FastIndexing.MediaIndexingWorker do require Logger + alias __MODULE__ + alias Pinchflat.Tasks alias Pinchflat.Sources alias Pinchflat.FastIndexing.FastIndexingHelpers - @impl Oban.Worker + @doc """ + Starts the fast media indexing worker and creates a task for the source. + + Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}} + """ + def kickoff_with_task(source, media_url, opts \\ []) do + %{id: source.id, media_url: media_url} + |> MediaIndexingWorker.new(opts) + |> Tasks.create_job_with_task(source) + end + @doc """ Similar to `MediaCollectionIndexingWorker`, but for individual media items. Does not reschedule or check anything to do with a source's indexing @@ -37,6 +49,7 @@ defmodule Pinchflat.FastIndexing.MediaIndexingWorker do Returns :ok """ + @impl Oban.Worker def perform(%Oban.Job{args: %{"id" => source_id, "media_url" => media_url}}) do source = Sources.get_source!(source_id) diff --git a/lib/pinchflat/metadata/source_metadata_storage_worker.ex b/lib/pinchflat/metadata/source_metadata_storage_worker.ex index 2500324..c0bc885 100644 --- a/lib/pinchflat/metadata/source_metadata_storage_worker.ex +++ b/lib/pinchflat/metadata/source_metadata_storage_worker.ex @@ -19,14 +19,11 @@ defmodule Pinchflat.Metadata.SourceMetadataStorageWorker do @doc """ Starts the source metadata storage worker and creates a task for the source. - IDEA: testing out this method of handling job kickoff. I think I like it, so - I may use it in other places. Just testing it for now - Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}} """ - def kickoff_with_task(source) do + def kickoff_with_task(source, opts \\ []) do %{id: source.id} - |> SourceMetadataStorageWorker.new() + |> SourceMetadataStorageWorker.new(opts) |> Tasks.create_job_with_task(source) end diff --git a/lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex b/lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex index 46b2ef0..2c5cee3 100644 --- a/lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex +++ b/lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex @@ -13,7 +13,17 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorker do alias Pinchflat.FastIndexing.FastIndexingWorker alias Pinchflat.SlowIndexing.SlowIndexingHelpers - @impl Oban.Worker + @doc """ + Starts the source slow indexing worker and creates a task for the source. + + Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}} + """ + def kickoff_with_task(source, opts \\ []) do + %{id: source.id} + |> MediaCollectionIndexingWorker.new(opts) + |> Tasks.create_job_with_task(source) + end + @doc """ The ID is that of a source _record_, not a YouTube channel/playlist ID. Indexes the provided source, kicks off downloads for each new MediaItem, and @@ -58,6 +68,7 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorker do Returns :ok | {:ok, %Task{}} """ + @impl Oban.Worker def perform(%Oban.Job{args: %{"id" => source_id}}) do source = Sources.get_source!(source_id) diff --git a/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex b/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex index 4598d1b..cb4d63b 100644 --- a/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex +++ b/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex @@ -31,10 +31,7 @@ defmodule Pinchflat.SlowIndexing.SlowIndexingHelpers do 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) + MediaCollectionIndexingWorker.kickoff_with_task(source) end @doc """ @@ -125,9 +122,7 @@ defmodule Pinchflat.SlowIndexing.SlowIndexingHelpers do 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) + MediaDownloadWorker.kickoff_with_task(media_item) end {:error, changeset} -> diff --git a/lib/pinchflat/sources/sources.ex b/lib/pinchflat/sources/sources.ex index ab85257..ea2b2d3 100644 --- a/lib/pinchflat/sources/sources.ex +++ b/lib/pinchflat/sources/sources.ex @@ -14,7 +14,7 @@ defmodule Pinchflat.Sources do alias Pinchflat.Metadata.SourceMetadata alias Pinchflat.Filesystem.FilesystemHelpers alias Pinchflat.Downloading.DownloadingHelpers - alias Pinchflat.FastIndexing.FastIndexingHelpers + alias Pinchflat.FastIndexing.FastIndexingWorker alias Pinchflat.SlowIndexing.SlowIndexingHelpers alias Pinchflat.Metadata.SourceMetadataStorageWorker @@ -121,17 +121,9 @@ defmodule Pinchflat.Sources do Source.changeset(source, attrs, validation_stage) end - @doc """ - Returns an `%Ecto.Changeset{}` for tracking source changes and additionally - fetches source details from the original_url (if provided). If the source - details cannot be fetched, an error is added to the changeset. - - NOTE: When operating in the ideal path, this effectively adds an API call - to the source creation/update process. Should be used only when needed. - - NOTE: this can almost certainly be made private now - """ - def maybe_change_source_from_url(%Source{} = source, attrs) do + # NOTE: When operating in the ideal path, this effectively adds an API call + # to the source creation/update process. Should be used only when needed. + defp maybe_change_source_from_url(%Source{} = source, attrs) do case change_source(source, attrs) do %Ecto.Changeset{changes: %{original_url: _}} = changeset -> add_source_details_to_changeset(source, changeset) @@ -271,7 +263,8 @@ defmodule Pinchflat.Sources do defp maybe_update_fast_indexing_task(changeset, source) do case changeset.changes do %{fast_index: true} -> - FastIndexingHelpers.kickoff_fast_indexing_task(source) + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") + FastIndexingWorker.kickoff_with_task(source) %{fast_index: false} -> Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") diff --git a/test/pinchflat/downloading/media_download_worker_test.exs b/test/pinchflat/downloading/media_download_worker_test.exs index fdebac6..aab984b 100644 --- a/test/pinchflat/downloading/media_download_worker_test.exs +++ b/test/pinchflat/downloading/media_download_worker_test.exs @@ -23,6 +23,19 @@ defmodule Pinchflat.Downloading.MediaDownloadWorkerTest do {:ok, %{media_item: media_item}} end + describe "kickoff_with_task/2" do + test "starts the worker", %{media_item: media_item} do + assert [] = all_enqueued(worker: MediaDownloadWorker) + assert {:ok, _} = MediaDownloadWorker.kickoff_with_task(media_item) + assert [_] = all_enqueued(worker: MediaDownloadWorker) + end + + test "attaches a task", %{media_item: media_item} do + assert {:ok, task} = MediaDownloadWorker.kickoff_with_task(media_item) + assert task.media_item_id == media_item.id + end + end + describe "perform/1" do test "it saves attributes to the media_item", %{media_item: media_item} do expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> diff --git a/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs b/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs index 0f0e12d..6c55051 100644 --- a/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs +++ b/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs @@ -2,51 +2,20 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpersTest 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.Downloading.MediaDownloadWorker alias Pinchflat.FastIndexing.MediaIndexingWorker alias Pinchflat.FastIndexing.FastIndexingHelpers - alias Pinchflat.FastIndexing.FastIndexingWorker setup :verify_on_exit! @media_url "https://www.youtube.com/watch?v=test_1" - describe "kickoff_fast_indexing_task/1" do - test "it schedules a job" do - source = source_fixture() - - assert {:ok, _} = FastIndexingHelpers.kickoff_fast_indexing_task(source) - - assert_enqueued(worker: FastIndexingWorker, args: %{"id" => source.id}) - end - - test "it creates and attaches a task" do - source = source_fixture() - - assert {:ok, %Task{} = task} = FastIndexingHelpers.kickoff_fast_indexing_task(source) - - assert task.source_id == source.id - 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, _} = FastIndexingHelpers.kickoff_fast_indexing_task(source) - - assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end - end - end - describe "kickoff_indexing_tasks_from_youtube_rss_feed/1" do setup do {:ok, [source: source_fixture()]} diff --git a/test/pinchflat/fast_indexing/fast_indexing_worker_test.exs b/test/pinchflat/fast_indexing/fast_indexing_worker_test.exs index 5108865..da3ff50 100644 --- a/test/pinchflat/fast_indexing/fast_indexing_worker_test.exs +++ b/test/pinchflat/fast_indexing/fast_indexing_worker_test.exs @@ -9,6 +9,23 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorkerTest do setup :verify_on_exit! + describe "kickoff_with_task/2" do + test "starts the worker" do + source = source_fixture(fast_index: true) + + assert [] = all_enqueued(worker: FastIndexingWorker) + assert {:ok, _} = FastIndexingWorker.kickoff_with_task(source) + assert [_] = all_enqueued(worker: FastIndexingWorker) + end + + test "attaches a task" do + source = source_fixture(fast_index: true) + + assert {:ok, task} = FastIndexingWorker.kickoff_with_task(source) + assert task.source_id == source.id + end + end + describe "perform/1" do test "calls out to Youtube RSS if enabled" do expect(HTTPClientMock, :get, fn _url -> {:ok, ""} end) diff --git a/test/pinchflat/fast_indexing/media_indexing_worker_test.exs b/test/pinchflat/fast_indexing/media_indexing_worker_test.exs index 96cc932..db31cc8 100644 --- a/test/pinchflat/fast_indexing/media_indexing_worker_test.exs +++ b/test/pinchflat/fast_indexing/media_indexing_worker_test.exs @@ -6,8 +6,8 @@ defmodule Pinchflat.FastIndexing.MediaIndexingWorkerTest do import Pinchflat.SourcesFixtures alias Pinchflat.Media.MediaItem - alias Pinchflat.FastIndexing.MediaIndexingWorker alias Pinchflat.Downloading.MediaDownloadWorker + alias Pinchflat.FastIndexing.MediaIndexingWorker @media_url "https://www.youtube.com/watch?v=1234567890" @@ -19,6 +19,19 @@ defmodule Pinchflat.FastIndexing.MediaIndexingWorkerTest do {:ok, source: source} end + describe "kickoff_with_task/2" do + test "starts the worker", %{source: source} do + assert [] = all_enqueued(worker: MediaIndexingWorker) + assert {:ok, _} = MediaIndexingWorker.kickoff_with_task(source, @media_url) + assert [_] = all_enqueued(worker: MediaIndexingWorker) + end + + test "attaches a task", %{source: source} do + assert {:ok, task} = MediaIndexingWorker.kickoff_with_task(source, @media_url) + assert task.source_id == source.id + end + end + describe "perform/1" do test "indexes the media item and saves it to the database", %{source: source} do expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> diff --git a/test/pinchflat/sources_test.exs b/test/pinchflat/sources_test.exs index 59f9a84..103a33f 100644 --- a/test/pinchflat/sources_test.exs +++ b/test/pinchflat/sources_test.exs @@ -508,70 +508,6 @@ defmodule Pinchflat.SourcesTest do end end - describe "maybe_change_source_from_url/2" do - test "it returns a changeset" do - stub(YtDlpRunnerMock, :run, &channel_mock/3) - source = source_fixture() - - assert %Ecto.Changeset{} = Sources.maybe_change_source_from_url(source, %{}) - end - - test "it does not fetch source details if the original_url isn't in the changeset" do - expect(YtDlpRunnerMock, :run, 0, &channel_mock/3) - - changeset = Sources.maybe_change_source_from_url(%Source{}, %{name: "some updated name"}) - - assert %Ecto.Changeset{} = changeset - end - - test "it fetches source details if the original_url is in the changeset" do - expect(YtDlpRunnerMock, :run, &channel_mock/3) - - changeset = - Sources.maybe_change_source_from_url(%Source{}, %{ - original_url: "https://www.youtube.com/channel/abc123" - }) - - assert %Ecto.Changeset{} = changeset - end - - test "it adds source details to the changeset, keeping the orignal details" do - expect(YtDlpRunnerMock, :run, &channel_mock/3) - - media_profile = media_profile_fixture() - media_profile_id = media_profile.id - - changeset = - Sources.maybe_change_source_from_url(%Source{}, %{ - original_url: "https://www.youtube.com/channel/abc123", - media_profile_id: media_profile.id - }) - - assert %Ecto.Changeset{} = changeset - assert String.starts_with?(changeset.changes.collection_id, "some_channel_id_") - - assert %{ - collection_name: "some channel name", - media_profile_id: ^media_profile_id, - original_url: "https://www.youtube.com/channel/abc123" - } = changeset.changes - end - - test "it adds an error to the changeset if the runner fails" do - expect(YtDlpRunnerMock, :run, 1, fn _url, _opts, _ot -> - {:error, "some error", 1} - end) - - changeset = - Sources.maybe_change_source_from_url(%Source{}, %{ - original_url: "https://www.youtube.com/channel/abc123" - }) - - assert %Ecto.Changeset{} = changeset - assert errors_on(changeset).original_url == ["could not fetch source details from URL"] - end - end - defp playlist_mock(_url, _opts, _ot) do { :ok,