From 874a0fe17951f86500aa679a874d495f92e6edc8 Mon Sep 17 00:00:00 2001 From: Kieran Eglin Date: Sat, 9 Mar 2024 14:47:59 -0800 Subject: [PATCH] Added fast indexing worker and updated other modules to start using it --- config/config.exs | 1 + lib/pinchflat/sources.ex | 3 +- lib/pinchflat/sources/source.ex | 7 +++ lib/pinchflat/tasks/source_tasks.ex | 6 ++- lib/pinchflat/workers/fast_indexing_worker.ex | 42 +++++++++++++++++ .../media_collection_indexing_worker.ex | 40 +++++++++++++++- test/pinchflat/sources_test.exs | 17 +++++-- test/pinchflat/tasks/source_tasks_test.exs | 23 +++++++++- .../workers/fast_indexing_worker_test.exs | 46 +++++++++++++++++++ .../media_collection_indexing_worker_test.exs | 39 +++++++++++++++- 10 files changed, 214 insertions(+), 10 deletions(-) create mode 100644 lib/pinchflat/workers/fast_indexing_worker.ex 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/sources.ex b/lib/pinchflat/sources.ex index 3948c8a..9baf37c 100644 --- a/lib/pinchflat/sources.ex +++ b/lib/pinchflat/sources.ex @@ -211,7 +211,8 @@ defmodule Pinchflat.Sources do SourceTasks.kickoff_indexing_task(source) %{index_frequency_minutes: _} -> - # TODO: delete the recurring RSS task (when I get there) + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") + Tasks.delete_pending_tasks_for(source, "MediaIndexingWorker") Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker") _ -> diff --git a/lib/pinchflat/sources/source.ex b/lib/pinchflat/sources/source.ex index 66f20af..0ed8aaa 100644 --- a/lib/pinchflat/sources/source.ex +++ b/lib/pinchflat/sources/source.ex @@ -68,6 +68,13 @@ defmodule Pinchflat.Sources.Source do @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..e110ffe 100644 --- a/lib/pinchflat/tasks/source_tasks.ex +++ b/lib/pinchflat/tasks/source_tasks.ex @@ -24,11 +24,13 @@ 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} @@ -90,8 +92,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/test/pinchflat/sources_test.exs b/test/pinchflat/sources_test.exs index 2532c6c..8805ad5 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} @@ -274,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 diff --git a/test/pinchflat/tasks/source_tasks_test.exs b/test/pinchflat/tasks/source_tasks_test.exs index 74bec27..49f9241 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,26 @@ 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_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)