From 816def21bb16aae71f03d9faa93bdcdff7ad6087 Mon Sep 17 00:00:00 2001 From: Kieran Date: Sat, 9 Mar 2024 20:09:11 -0800 Subject: [PATCH] Fast index UI (#63) * Added fast_index field and adds it to source form * Added fast indexing to source changeset operations * Added fast indexing worker and updated other modules to start using it * Handled fast index worker on source update --- config/config.exs | 1 + lib/pinchflat/api/youtube_rss.ex | 6 ++ lib/pinchflat/sources.ex | 51 +++++++++-- lib/pinchflat/sources/source.ex | 15 ++++ lib/pinchflat/tasks/source_tasks.ex | 31 +++++-- lib/pinchflat/workers/fast_indexing_worker.ex | 42 +++++++++ .../media_collection_indexing_worker.ex | 40 ++++++++- .../controllers/sources/source_controller.ex | 16 ++-- .../source_html/fast_indexing_help.html.heex | 21 +++++ .../sources/source_html/show.html.heex | 6 +- .../sources/source_html/source_form.html.heex | 28 +++++- ...240309182418_add_fast_index_to_sources.exs | 9 ++ test/pinchflat/sources_test.exs | 85 ++++++++++++++++++- test/pinchflat/tasks/source_tasks_test.exs | 51 ++++++++++- .../workers/fast_indexing_worker_test.exs | 46 ++++++++++ .../media_collection_indexing_worker_test.exs | 39 ++++++++- 16 files changed, 456 insertions(+), 31 deletions(-) create mode 100644 lib/pinchflat/workers/fast_indexing_worker.ex create mode 100644 lib/pinchflat_web/controllers/sources/source_html/fast_indexing_help.html.heex create mode 100644 priv/repo/migrations/20240309182418_add_fast_index_to_sources.exs create mode 100644 test/pinchflat/workers/fast_indexing_worker_test.exs diff --git a/config/config.exs b/config/config.exs index 70b5080..4396f4b 100644 --- a/config/config.exs +++ b/config/config.exs @@ -42,6 +42,7 @@ config :pinchflat, Oban, # TODO: consider making this an env var or something? queues: [ default: 10, + fast_indexing: 6, media_indexing: 2, media_collection_indexing: 2, media_fetching: 2, diff --git a/lib/pinchflat/api/youtube_rss.ex b/lib/pinchflat/api/youtube_rss.ex index 146f1b1..7eef959 100644 --- a/lib/pinchflat/api/youtube_rss.ex +++ b/lib/pinchflat/api/youtube_rss.ex @@ -3,6 +3,8 @@ defmodule Pinchflat.Api.YoutubeRss do Methods for interacting with YouTube RSS feeds """ + require Logger + alias Pinchflat.Sources.Source @doc """ @@ -11,6 +13,8 @@ defmodule Pinchflat.Api.YoutubeRss do Returns {:ok, [binary()]} | {:error, binary()} """ def get_recent_media_ids_from_rss(%Source{} = source) do + Logger.debug("Fetching recent media IDs from YouTube RSS feed for source: #{source.collection_id}") + case http_client().get(rss_url_for_source(source)) do {:ok, response} -> response = to_string(response) @@ -25,6 +29,8 @@ defmodule Pinchflat.Api.YoutubeRss do |> Enum.filter(&(String.length(&1) > 0)) |> Enum.uniq() + Logger.debug("Media ids fetched from RSS: #{inspect(media_ids)}") + {:ok, media_ids} {:error, _reason} -> diff --git a/lib/pinchflat/sources.ex b/lib/pinchflat/sources.ex index 95bfde8..eee7c64 100644 --- a/lib/pinchflat/sources.ex +++ b/lib/pinchflat/sources.ex @@ -46,6 +46,7 @@ defmodule Pinchflat.Sources do def create_source(attrs) do %Source{} |> change_source_from_url(attrs) + |> maybe_change_indexing_frequency() |> commit_and_handle_tasks() end @@ -62,6 +63,7 @@ defmodule Pinchflat.Sources do def update_source(%Source{} = source, attrs) do source |> change_source_from_url(attrs) + |> maybe_change_indexing_frequency() |> commit_and_handle_tasks() end @@ -151,6 +153,20 @@ defmodule Pinchflat.Sources do change_source(source, Map.merge(changes, collection_changes)) end + defp maybe_change_indexing_frequency(changeset) do + fast_index = Ecto.Changeset.get_field(changeset, :fast_index) + + if fast_index do + Ecto.Changeset.put_change( + changeset, + :index_frequency_minutes, + Source.index_frequency_when_fast_indexing() + ) + else + changeset + end + end + defp commit_and_handle_tasks(changeset) do case Repo.insert_or_update(changeset) do {:ok, %Source{} = source} -> @@ -188,13 +204,38 @@ defmodule Pinchflat.Sources do # If the record has been persisted, only run indexing if the # indexing frequency has been changed and is now greater than 0 %{__meta__: %{state: :loaded}} -> - case changeset.changes do - %{index_frequency_minutes: mins} when mins > 0 -> SourceTasks.kickoff_indexing_task(source) - %{index_frequency_minutes: _} -> Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker") - _ -> :ok - end + maybe_update_slow_indexing_task(changeset, source) + maybe_update_fast_indexing_task(changeset, source) end {:ok, source} end + + 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) + + %{index_frequency_minutes: _} -> + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") + Tasks.delete_pending_tasks_for(source, "MediaIndexingWorker") + Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker") + + _ -> + :ok + end + end + + defp maybe_update_fast_indexing_task(changeset, source) do + case changeset.changes do + %{fast_index: true} -> + SourceTasks.kickoff_fast_indexing_task(source) + + %{fast_index: false} -> + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") + + _ -> + :ok + end + end end diff --git a/lib/pinchflat/sources/source.ex b/lib/pinchflat/sources/source.ex index 6282e28..0ed8aaa 100644 --- a/lib/pinchflat/sources/source.ex +++ b/lib/pinchflat/sources/source.ex @@ -17,6 +17,7 @@ defmodule Pinchflat.Sources.Source do collection_type custom_name index_frequency_minutes + fast_index download_media last_indexed_at original_url @@ -29,6 +30,7 @@ defmodule Pinchflat.Sources.Source do collection_type custom_name index_frequency_minutes + fast_index download_media original_url media_profile_id @@ -40,6 +42,7 @@ defmodule Pinchflat.Sources.Source do field :collection_id, :string field :collection_type, Ecto.Enum, values: [:channel, :playlist] field :index_frequency_minutes, :integer, default: 60 * 24 + field :fast_index, :boolean, default: false field :download_media, :boolean, default: true field :last_indexed_at, :utc_datetime # This should only be used for user reference going forward @@ -62,4 +65,16 @@ defmodule Pinchflat.Sources.Source do |> validate_required(@required_fields) |> unique_constraint([:collection_id, :media_profile_id]) end + + @doc false + def index_frequency_when_fast_indexing do + # 30 days in minutes + 60 * 24 * 30 + end + + @doc false + def fast_index_frequency do + # minutes + 15 + end end diff --git a/lib/pinchflat/tasks/source_tasks.ex b/lib/pinchflat/tasks/source_tasks.ex index 8fab791..4c786ef 100644 --- a/lib/pinchflat/tasks/source_tasks.ex +++ b/lib/pinchflat/tasks/source_tasks.ex @@ -14,6 +14,7 @@ defmodule Pinchflat.Tasks.SourceTasks do alias Pinchflat.Sources.Source alias Pinchflat.Api.YoutubeRss alias Pinchflat.Media.MediaItem + alias Pinchflat.Workers.FastIndexingWorker alias Pinchflat.Workers.MediaDownloadWorker alias Pinchflat.Workers.MediaIndexingWorker alias Pinchflat.YtDlp.Backend.MediaCollection @@ -24,22 +25,38 @@ defmodule Pinchflat.Tasks.SourceTasks do @doc """ Starts tasks for indexing a source's media regardless of the source's indexing - frequency. It's assumed the caller will check for that. + 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) - |> case do - # This should never return {:error, :duplicate_job} since we just deleted - # any pending tasks. I'm being assertive about it so it's obvious if I'm wrong - {:ok, task} -> {:ok, task} - end + end + + @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 """ @@ -90,8 +107,8 @@ defmodule Pinchflat.Tasks.SourceTasks 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) diff --git a/lib/pinchflat/workers/fast_indexing_worker.ex b/lib/pinchflat/workers/fast_indexing_worker.ex new file mode 100644 index 0000000..e2f135e --- /dev/null +++ b/lib/pinchflat/workers/fast_indexing_worker.ex @@ -0,0 +1,42 @@ +defmodule Pinchflat.Workers.FastIndexingWorker do + @moduledoc false + + use Oban.Worker, + queue: :fast_indexing, + unique: [period: :infinity, states: [:available, :scheduled, :retryable]], + tags: ["media_source", "fast_indexing"] + + alias __MODULE__ + alias Pinchflat.Tasks + alias Pinchflat.Sources + alias Pinchflat.Sources.Source + alias Pinchflat.Tasks.SourceTasks + + @impl Oban.Worker + @doc """ + TODO + """ + def perform(%Oban.Job{args: %{"id" => source_id}}) do + source = Sources.get_source!(source_id) + + if source.fast_index do + SourceTasks.kickoff_indexing_tasks_from_youtube_rss_feed(source) + + reschedule_indexing(source) + else + :ok + end + end + + 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 + {:ok, task} -> {:ok, task} + {:error, :duplicate_job} -> {:ok, :job_exists} + end + end +end diff --git a/lib/pinchflat/workers/media_collection_indexing_worker.ex b/lib/pinchflat/workers/media_collection_indexing_worker.ex index d5aaa6e..3f7122e 100644 --- a/lib/pinchflat/workers/media_collection_indexing_worker.ex +++ b/lib/pinchflat/workers/media_collection_indexing_worker.ex @@ -9,7 +9,9 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do alias __MODULE__ alias Pinchflat.Tasks alias Pinchflat.Sources + alias Pinchflat.Sources.Source alias Pinchflat.Tasks.SourceTasks + alias Pinchflat.Workers.FastIndexingWorker @impl Oban.Worker @doc """ @@ -30,6 +32,27 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do actually run every 1 hour and 30 minutes. The tradeoff of not inundating the API with requests and also not overlapping jobs is worth it, IMO. + Order of operations: + 1. The user saves a source + 2. This job is automatically scheduled immediately. This happens in all cases. + 3. This job indexes all content for the given source. A download job is + enqueued for each media item that should be downloaded. This can be impacted + 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) + 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 + uses RSS to run a smaller, faster, and more frequent index. That job + handles rescheduling itself but largely has a similar behaviour to this + job in that it kicks off index and maybe download jobs. The biggest difference + is that an index job is kicked off _for each new media item_ as opposed + to one larger index job. Check out `MediaIndexingWorker` comments for more. + 6. If the job reschedules, the cycle from step 3 repeats until the heat death + of the universe. The user changing things like the index frequency can + dequeue or reschedule jobs as well + 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? @@ -44,6 +67,7 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker 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) + maybe_enqueue_fast_indexing_task(source) reschedule_indexing(source) {_, nil} -> @@ -60,12 +84,26 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do end defp reschedule_indexing(source) do + next_run_in = source.index_frequency_minutes * 60 + %{id: source.id} - |> MediaCollectionIndexingWorker.new(schedule_in: source.index_frequency_minutes * 60) + |> MediaCollectionIndexingWorker.new(schedule_in: next_run_in) |> Tasks.create_job_with_task(source) |> case do {:ok, task} -> {:ok, task} {:error, :duplicate_job} -> {:ok, :job_exists} end end + + defp maybe_enqueue_fast_indexing_task(source) do + if source.fast_index do + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") + + next_run_in = Source.fast_index_frequency() * 60 + + %{id: source.id} + |> FastIndexingWorker.new(schedule_in: next_run_in) + |> Tasks.create_job_with_task(source) + end + end end diff --git a/lib/pinchflat_web/controllers/sources/source_controller.ex b/lib/pinchflat_web/controllers/sources/source_controller.ex index 757fb44..60039f4 100644 --- a/lib/pinchflat_web/controllers/sources/source_controller.ex +++ b/lib/pinchflat_web/controllers/sources/source_controller.ex @@ -3,8 +3,9 @@ defmodule PinchflatWeb.Sources.SourceController do alias Pinchflat.Repo alias Pinchflat.Media - alias Pinchflat.Profiles + alias Pinchflat.Tasks alias Pinchflat.Sources + alias Pinchflat.Profiles alias Pinchflat.Sources.Source def index(conn, _params) do @@ -43,15 +44,18 @@ defmodule PinchflatWeb.Sources.SourceController do end def show(conn, %{"id" => id}) do - source = - id - |> Sources.get_source!() - |> Repo.preload([:media_profile, tasks: [:job]]) + source = Repo.preload(Sources.get_source!(id), :media_profile) + pending_tasks = Repo.preload(Tasks.list_pending_tasks_for(:source_id, source.id), :job) pending_media = Media.list_pending_media_items_for(source, limit: 100) downloaded_media = Media.list_downloaded_media_items_for(source, limit: 100) - render(conn, :show, source: source, pending_media: pending_media, downloaded_media: downloaded_media) + render(conn, :show, + source: source, + pending_tasks: pending_tasks, + pending_media: pending_media, + downloaded_media: downloaded_media + ) end def edit(conn, %{"id" => id}) do diff --git a/lib/pinchflat_web/controllers/sources/source_html/fast_indexing_help.html.heex b/lib/pinchflat_web/controllers/sources/source_html/fast_indexing_help.html.heex new file mode 100644 index 0000000..710b2ed --- /dev/null +++ b/lib/pinchflat_web/controllers/sources/source_html/fast_indexing_help.html.heex @@ -0,0 +1,21 @@ + diff --git a/lib/pinchflat_web/controllers/sources/source_html/show.html.heex b/lib/pinchflat_web/controllers/sources/source_html/show.html.heex index d8c69d2..1b08883 100644 --- a/lib/pinchflat_web/controllers/sources/source_html/show.html.heex +++ b/lib/pinchflat_web/controllers/sources/source_html/show.html.heex @@ -84,9 +84,9 @@

Nothing Here!

<% end %> - <:tab title="Tasks"> - <%= if match?([_|_], @source.tasks) do %> - <.table rows={@source.tasks} table_class="text-black dark:text-white"> + <:tab title="Pending Tasks"> + <%= if match?([_|_], @pending_tasks) do %> + <.table rows={@pending_tasks} table_class="text-black dark:text-white"> <:col :let={task} label="Worker"> <%= task.job.worker %> diff --git a/lib/pinchflat_web/controllers/sources/source_html/source_form.html.heex b/lib/pinchflat_web/controllers/sources/source_html/source_form.html.heex index c538c96..8124106 100644 --- a/lib/pinchflat_web/controllers/sources/source_html/source_form.html.heex +++ b/lib/pinchflat_web/controllers/sources/source_html/source_form.html.heex @@ -3,6 +3,10 @@ Oops, something went wrong! Please check the errors below. +

+ General Options +

+ <.input field={f[:custom_name]} type="text" @@ -19,6 +23,10 @@ label="Media Profile" /> +

+ Indexing Options +

+ <.input field={f[:index_frequency_minutes]} options={friendly_index_frequencies()} @@ -27,6 +35,18 @@ help="Time between one index of this source finishing and the next one starting. Setting to 'On Create' will still run an initial index but no subsequent ones" /> + <%!-- TODO: use Alpine to disable the index frequency when fast indexing is enabled --%> + <.input + field={f[:fast_index]} + type="toggle" + label="Use Fast Indexing?" + help="Experimental. Ignores 'Index Frequency'. Recommended for large channels that upload frequently. See below for more info" + /> + +

+ Downloading Options +

+ <.input field={f[:download_media]} type="toggle" @@ -34,7 +54,9 @@ help="Unchecking still indexes media but it won't be downloaded until you enable this option" /> - <:actions> - <.button class="my-10 sm:mb-7.5 w-full sm:w-auto">Save Source - + <.button class="my-10 sm:mb-7.5 w-full sm:w-auto">Save Source + +
+ <.fast_indexing_help /> +
diff --git a/priv/repo/migrations/20240309182418_add_fast_index_to_sources.exs b/priv/repo/migrations/20240309182418_add_fast_index_to_sources.exs new file mode 100644 index 0000000..26b3a6c --- /dev/null +++ b/priv/repo/migrations/20240309182418_add_fast_index_to_sources.exs @@ -0,0 +1,9 @@ +defmodule Pinchflat.Repo.Migrations.AddFastIndexToSources do + use Ecto.Migration + + def change do + alter table(:sources) do + add :fast_index, :boolean, null: false, default: false + end + end +end diff --git a/test/pinchflat/sources_test.exs b/test/pinchflat/sources_test.exs index 5e8c6e2..fe1db7e 100644 --- a/test/pinchflat/sources_test.exs +++ b/test/pinchflat/sources_test.exs @@ -9,8 +9,10 @@ defmodule Pinchflat.SourcesTest do alias Pinchflat.Sources alias Pinchflat.Tasks.SourceTasks alias Pinchflat.Sources.Source - alias Pinchflat.Workers.MediaCollectionIndexingWorker + alias Pinchflat.Workers.FastIndexingWorker alias Pinchflat.Workers.MediaDownloadWorker + alias Pinchflat.Workers.MediaIndexingWorker + alias Pinchflat.Workers.MediaCollectionIndexingWorker @invalid_source_attrs %{name: nil, collection_id: nil} @@ -182,6 +184,36 @@ defmodule Pinchflat.SourcesTest do assert_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id}) end + + test "fast_index forces the index frequency to be a default value" do + expect(YtDlpRunnerMock, :run, &channel_mock/3) + + valid_attrs = %{ + media_profile_id: media_profile_fixture().id, + original_url: "https://www.youtube.com/channel/abc123", + fast_index: true, + index_frequency_minutes: 0 + } + + assert {:ok, %Source{} = source} = Sources.create_source(valid_attrs) + + assert source.index_frequency_minutes == Source.index_frequency_when_fast_indexing() + end + + test "disabling fast index will not change the index frequency" do + expect(YtDlpRunnerMock, :run, &channel_mock/3) + + valid_attrs = %{ + media_profile_id: media_profile_fixture().id, + original_url: "https://www.youtube.com/channel/abc123", + fast_index: false, + index_frequency_minutes: 0 + } + + assert {:ok, %Source{} = source} = Sources.create_source(valid_attrs) + + assert source.index_frequency_minutes == 0 + end end describe "update_source/2" do @@ -244,13 +276,20 @@ defmodule Pinchflat.SourcesTest do test "updating the index frequency to 0 will delete any pending tasks" do source = source_fixture() - {:ok, job} = Oban.insert(MediaCollectionIndexingWorker.new(%{"id" => source.id})) - task = task_fixture(source_id: source.id, job_id: job.id) update_attrs = %{index_frequency_minutes: 0} + {:ok, job_1} = Oban.insert(FastIndexingWorker.new(%{"id" => source.id})) + task_1 = task_fixture(source_id: source.id, job_id: job_1.id) + {:ok, job_2} = Oban.insert(MediaIndexingWorker.new(%{"id" => source.id})) + task_2 = task_fixture(source_id: source.id, job_id: job_2.id) + {:ok, job_3} = Oban.insert(MediaCollectionIndexingWorker.new(%{"id" => source.id})) + task_3 = task_fixture(source_id: source.id, job_id: job_3.id) + assert {:ok, %Source{}} = Sources.update_source(source, update_attrs) - assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end + assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task_1) end + assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task_2) end + assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task_3) end end test "not updating the index frequency will not re-schedule the indexing task or delete tasks" do @@ -285,6 +324,26 @@ defmodule Pinchflat.SourcesTest do refute_enqueued(worker: MediaDownloadWorker) end + test "enabling fast_index will schedule a fast indexing task" do + source = source_fixture(fast_index: false) + update_attrs = %{fast_index: true} + + refute_enqueued(worker: FastIndexingWorker) + assert {:ok, %Source{}} = Sources.update_source(source, update_attrs) + assert_enqueued(worker: FastIndexingWorker, args: %{"id" => source.id}) + end + + test "disabling fast_index will cancel the fast indexing task" do + source = source_fixture(fast_index: true) + update_attrs = %{fast_index: false} + {:ok, job} = Oban.insert(FastIndexingWorker.new(%{"id" => source.id})) + task_fixture(source_id: source.id, job_id: job.id) + + assert_enqueued(worker: FastIndexingWorker, args: %{"id" => source.id}) + assert {:ok, %Source{}} = Sources.update_source(source, update_attrs) + refute_enqueued(worker: FastIndexingWorker) + end + test "updates with invalid data returns error changeset" do source = source_fixture() @@ -293,6 +352,24 @@ defmodule Pinchflat.SourcesTest do assert source == Sources.get_source!(source.id) end + + test "fast_index forces the index frequency to be a default value" do + source = source_fixture(%{fast_index: true}) + update_attrs = %{index_frequency_minutes: 0} + + assert {:ok, source} = Sources.update_source(source, update_attrs) + + assert source.index_frequency_minutes == Source.index_frequency_when_fast_indexing() + end + + test "disabling fast index will not change the index frequency" do + source = source_fixture(%{fast_index: false}) + update_attrs = %{index_frequency_minutes: 0} + + assert {:ok, source} = Sources.update_source(source, update_attrs) + + assert source.index_frequency_minutes == 0 + end end describe "delete_source/2" do diff --git a/test/pinchflat/tasks/source_tasks_test.exs b/test/pinchflat/tasks/source_tasks_test.exs index 74bec27..55df00c 100644 --- a/test/pinchflat/tasks/source_tasks_test.exs +++ b/test/pinchflat/tasks/source_tasks_test.exs @@ -11,6 +11,7 @@ defmodule Pinchflat.Tasks.SourceTasksTest do alias Pinchflat.Tasks.Task alias Pinchflat.Tasks.SourceTasks alias Pinchflat.Media.MediaItem + alias Pinchflat.Workers.FastIndexingWorker alias Pinchflat.Workers.MediaDownloadWorker alias Pinchflat.Workers.MediaIndexingWorker alias Pinchflat.Workers.MediaCollectionIndexingWorker @@ -34,7 +35,7 @@ defmodule Pinchflat.Tasks.SourceTasksTest do assert task.source_id == source.id end - test "it deletes any pending tasks for the source" do + 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) @@ -43,6 +44,54 @@ defmodule Pinchflat.Tasks.SourceTasksTest do 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 "kickoff_fast_indexing_task/1" do + test "it schedules a job" do + source = source_fixture() + + assert {:ok, _} = SourceTasks.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} = SourceTasks.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, _} = SourceTasks.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 diff --git a/test/pinchflat/workers/fast_indexing_worker_test.exs b/test/pinchflat/workers/fast_indexing_worker_test.exs new file mode 100644 index 0000000..701b827 --- /dev/null +++ b/test/pinchflat/workers/fast_indexing_worker_test.exs @@ -0,0 +1,46 @@ +defmodule Pinchflat.Workers.FastIndexingWorkerTest do + use Pinchflat.DataCase + + import Mox + import Pinchflat.SourcesFixtures + + alias Pinchflat.Sources.Source + alias Pinchflat.Workers.FastIndexingWorker + + setup :verify_on_exit! + + describe "perform/1" do + test "calls out to Youtube RSS if enabled" do + expect(HTTPClientMock, :get, fn _url -> {:ok, ""} end) + source = source_fixture(fast_index: true) + + perform_job(FastIndexingWorker, %{"id" => source.id}) + end + + test "reschedules itself if fast indexing is enabled" do + expect(HTTPClientMock, :get, fn _url -> {:ok, ""} end) + source = source_fixture(fast_index: true) + perform_job(FastIndexingWorker, %{"id" => source.id}) + + assert_enqueued( + worker: FastIndexingWorker, + args: %{"id" => source.id}, + scheduled_at: now_plus(Source.fast_index_frequency(), :minutes) + ) + end + + test "does not call out to Youtube RSS if disabled" do + expect(HTTPClientMock, :get, 0, fn _url -> {:ok, ""} end) + source = source_fixture(fast_index: false) + + perform_job(FastIndexingWorker, %{"id" => source.id}) + end + + test "does not reschedule itself if fast indexing is disabled" do + source = source_fixture(fast_index: false) + perform_job(FastIndexingWorker, %{"id" => source.id}) + + refute_enqueued(worker: FastIndexingWorker, args: %{"id" => source.id}) + end + end +end diff --git a/test/pinchflat/workers/media_collection_indexing_worker_test.exs b/test/pinchflat/workers/media_collection_indexing_worker_test.exs index e60ab8b..4d3c5b7 100644 --- a/test/pinchflat/workers/media_collection_indexing_worker_test.exs +++ b/test/pinchflat/workers/media_collection_indexing_worker_test.exs @@ -2,12 +2,15 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorkerTest do use Pinchflat.DataCase import Mox + import Pinchflat.TasksFixtures import Pinchflat.MediaFixtures import Pinchflat.SourcesFixtures alias Pinchflat.Tasks - alias Pinchflat.Workers.MediaCollectionIndexingWorker + alias Pinchflat.Sources.Source + alias Pinchflat.Workers.FastIndexingWorker alias Pinchflat.Workers.MediaDownloadWorker + alias Pinchflat.Workers.MediaCollectionIndexingWorker setup :verify_on_exit! @@ -105,6 +108,40 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorkerTest do end) end + test "it creates a future task for fast indexing if appropriate" do + expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end) + + source = source_fixture(index_frequency_minutes: 10, fast_index: true) + perform_job(MediaCollectionIndexingWorker, %{id: source.id}) + + assert_enqueued( + worker: FastIndexingWorker, + args: %{"id" => source.id}, + scheduled_at: now_plus(Source.fast_index_frequency(), :minutes) + ) + end + + test "it deletes existing fast indexing tasks if a new one is created" do + expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end) + + source = source_fixture(index_frequency_minutes: 10, fast_index: true) + {:ok, job} = Oban.insert(FastIndexingWorker.new(%{"id" => source.id})) + task = task_fixture(source_id: source.id, job_id: job.id) + + perform_job(MediaCollectionIndexingWorker, %{id: source.id}) + + assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end + end + + test "it does not create a task for fast indexing otherwise" do + expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end) + + source = source_fixture(index_frequency_minutes: 10, fast_index: false) + perform_job(MediaCollectionIndexingWorker, %{id: source.id}) + + refute_enqueued(worker: FastIndexingWorker) + end + test "it creates the basic media_item records" do expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, source_attributes_return_fixture()} end)