Updated sources context to do the right thing when a source is updated

This commit is contained in:
Kieran Eglin 2024-11-21 10:21:21 -08:00
parent d20bdf86d9
commit dc6d281d47
No known key found for this signature in database
GPG key ID: 193984967FCF432D
6 changed files with 141 additions and 20 deletions

View file

@ -11,14 +11,22 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do
alias Pinchflat.Repo alias Pinchflat.Repo
alias Pinchflat.Media alias Pinchflat.Media
alias Pinchflat.Tasks
alias Pinchflat.Sources.Source alias Pinchflat.Sources.Source
alias Pinchflat.FastIndexing.YoutubeRss alias Pinchflat.FastIndexing.YoutubeRss
alias Pinchflat.FastIndexing.YoutubeApi alias Pinchflat.FastIndexing.YoutubeApi
alias Pinchflat.Downloading.DownloadingHelpers alias Pinchflat.Downloading.DownloadingHelpers
alias Pinchflat.FastIndexing.FastIndexingWorker
alias Pinchflat.Downloading.DownloadOptionBuilder alias Pinchflat.Downloading.DownloadOptionBuilder
alias Pinchflat.YtDlp.Media, as: YtDlpMedia alias Pinchflat.YtDlp.Media, as: YtDlpMedia
# TODO: test
def kickoff_indexing_task(%Source{} = source) do
Tasks.delete_pending_tasks_for(source, "FastIndexingWorker", include_executing: true)
FastIndexingWorker.kickoff_with_task(source)
end
@doc """ @doc """
Fetches new media IDs for a source from YT's API or RSS, indexes them, and kicks off downloading Fetches new media IDs for a source from YT's API or RSS, indexes them, and kicks off downloading
tasks for any pending media items. See comments in `FastIndexingWorker` for more info on the tasks for any pending media items. See comments in `FastIndexingWorker` for more info on the

View file

@ -34,6 +34,14 @@ defmodule Pinchflat.SlowIndexing.SlowIndexingHelpers do
MediaCollectionIndexingWorker.kickoff_with_task(source, job_args, job_opts) MediaCollectionIndexingWorker.kickoff_with_task(source, job_args, job_opts)
end end
# TODO: test
def delete_indexing_tasks(%Source{} = source, opts \\ []) do
include_executing = Keyword.get(opts, :include_executing, false)
Tasks.delete_pending_tasks_for(source, "FastIndexingWorker", include_executing: include_executing)
Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker", include_executing: include_executing)
end
@doc """ @doc """
Given a media source, creates (indexes) the media by creating media_items for each 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 media ID in the source. Afterward, kicks off a download task for each pending media

View file

@ -15,8 +15,8 @@ defmodule Pinchflat.Sources do
alias Pinchflat.Metadata.SourceMetadata alias Pinchflat.Metadata.SourceMetadata
alias Pinchflat.Utils.FilesystemUtils alias Pinchflat.Utils.FilesystemUtils
alias Pinchflat.Downloading.DownloadingHelpers alias Pinchflat.Downloading.DownloadingHelpers
alias Pinchflat.FastIndexing.FastIndexingWorker
alias Pinchflat.SlowIndexing.SlowIndexingHelpers alias Pinchflat.SlowIndexing.SlowIndexingHelpers
alias Pinchflat.FastIndexing.FastIndexingHelpers
alias Pinchflat.Metadata.SourceMetadataStorageWorker alias Pinchflat.Metadata.SourceMetadataStorageWorker
@doc """ @doc """
@ -272,18 +272,18 @@ defmodule Pinchflat.Sources do
# current changes or if that's how it already was in the database. # current changes or if that's how it already was in the database.
# Rephrased, we're essentially using it in place of `get_field/2` # Rephrased, we're essentially using it in place of `get_field/2`
case {current_changes, applied_changes} do case {current_changes, applied_changes} do
{%{download_media: false}, _} ->
DownloadingHelpers.dequeue_pending_download_tasks(source)
{%{download_media: true}, %{enabled: true}} -> {%{download_media: true}, %{enabled: true}} ->
DownloadingHelpers.enqueue_pending_download_tasks(source) DownloadingHelpers.enqueue_pending_download_tasks(source)
{%{enabled: false}, _} ->
DownloadingHelpers.dequeue_pending_download_tasks(source)
{%{enabled: true}, %{download_media: true}} -> {%{enabled: true}, %{download_media: true}} ->
DownloadingHelpers.enqueue_pending_download_tasks(source) DownloadingHelpers.enqueue_pending_download_tasks(source)
{%{download_media: false}, _} ->
DownloadingHelpers.dequeue_pending_download_tasks(source)
{%{enabled: false}, _} ->
DownloadingHelpers.dequeue_pending_download_tasks(source)
_ -> _ ->
nil nil
end end
@ -322,13 +322,24 @@ defmodule Pinchflat.Sources do
end end
defp maybe_update_slow_indexing_task(changeset, source) do defp maybe_update_slow_indexing_task(changeset, source) do
case changeset.changes do # See comment in `maybe_handle_media_tasks` as to why we need these
%{index_frequency_minutes: mins} when mins > 0 -> current_changes = changeset.changes
applied_changes = Ecto.Changeset.apply_changes(changeset)
case {current_changes, applied_changes} do
{%{index_frequency_minutes: mins}, %{enabled: true}} when mins > 0 ->
# TODO: consider scheduling the task for the future based on the index_frequency_minutes
# compared to `last_indexed_at`
SlowIndexingHelpers.kickoff_indexing_task(source) SlowIndexingHelpers.kickoff_indexing_task(source)
%{index_frequency_minutes: _} -> {%{enabled: true}, %{index_frequency_minutes: mins}} when mins > 0 ->
Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") SlowIndexingHelpers.kickoff_indexing_task(source)
Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker")
{%{index_frequency_minutes: mins}, _} when mins <= 0 ->
SlowIndexingHelpers.delete_indexing_tasks(source, include_executing: true)
{%{enabled: false}, _} ->
SlowIndexingHelpers.delete_indexing_tasks(source, include_executing: true)
_ -> _ ->
:ok :ok
@ -336,13 +347,25 @@ defmodule Pinchflat.Sources do
end end
defp maybe_update_fast_indexing_task(changeset, source) do defp maybe_update_fast_indexing_task(changeset, source) do
case changeset.changes do # See comment in `maybe_handle_media_tasks` as to why we need these
%{fast_index: true} -> current_changes = changeset.changes
Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") applied_changes = Ecto.Changeset.apply_changes(changeset)
FastIndexingWorker.kickoff_with_task(source)
%{fast_index: false} -> # This technically could be simplified since `maybe_update_slow_indexing_task`
Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") # has some overlap re: deleting pending tasks, but I'm keeping it separate
# for clarity and explicitness.
case {current_changes, applied_changes} do
{%{fast_index: true}, %{enabled: true}} ->
FastIndexingHelpers.kickoff_indexing_task(source)
{%{enabled: true}, %{fast_index: true}} ->
FastIndexingHelpers.kickoff_indexing_task(source)
{%{fast_index: false}, _} ->
Tasks.delete_pending_tasks_for(source, "FastIndexingWorker", include_executing: true)
{%{enabled: false}, _} ->
Tasks.delete_pending_tasks_for(source, "FastIndexingWorker", include_executing: true)
_ -> _ ->
:ok :ok

View file

@ -10,7 +10,8 @@
</.link> </.link>
</nav> </nav>
</div> </div>
<%!-- TODO: link the sources count to the sources tab --%>
<%!-- TODO: consider doing the same for other tables, like the source index's media item counts --%>
<div class="rounded-sm border border-stroke bg-white shadow-default dark:border-strokedark dark:bg-boxdark"> <div class="rounded-sm border border-stroke bg-white shadow-default dark:border-strokedark dark:bg-boxdark">
<div class="max-w-full overflow-x-auto"> <div class="max-w-full overflow-x-auto">
<div class="flex flex-col gap-10 min-w-max"> <div class="flex flex-col gap-10 min-w-max">

View file

@ -8,6 +8,7 @@ defmodule PinchflatWeb.Sources.IndexTableLive do
alias Pinchflat.Media.MediaItem alias Pinchflat.Media.MediaItem
# TODO: test (and maybe remove existing index tests) # TODO: test (and maybe remove existing index tests)
# TODO: see comments in media profile index view
def render(assigns) do def render(assigns) do
~H""" ~H"""
<.table rows={@sources} table_class="text-white"> <.table rows={@sources} table_class="text-white">

View file

@ -511,7 +511,7 @@ defmodule Pinchflat.SourcesTest do
end end
end end
describe "update_source/3 when testing indexing" do describe "update_source/3 when testing slow indexing" do
test "updating the index frequency to >0 will re-schedule the indexing task" do test "updating the index frequency to >0 will re-schedule the indexing task" do
source = source_fixture() source = source_fixture()
update_attrs = %{index_frequency_minutes: 123} update_attrs = %{index_frequency_minutes: 123}
@ -556,6 +556,47 @@ defmodule Pinchflat.SourcesTest do
refute_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id}) refute_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id})
end end
test "disabling a source will delete any pending tasks" do
source = source_fixture()
update_attrs = %{enabled: false}
{:ok, job} = Oban.insert(MediaCollectionIndexingWorker.new(%{"id" => source.id}))
task = task_fixture(source_id: source.id, job_id: job.id)
assert {:ok, %Source{}} = Sources.update_source(source, update_attrs)
assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end
end
test "updating the index frequency will not create a task if the source is disabled" do
source = source_fixture(enabled: false)
update_attrs = %{index_frequency_minutes: 123}
refute_enqueued(worker: MediaCollectionIndexingWorker)
assert {:ok, %Source{}} = Sources.update_source(source, update_attrs)
refute_enqueued(worker: MediaCollectionIndexingWorker)
end
test "enabling a source will create a task if the index frequency is >0" do
source = source_fixture(enabled: false, index_frequency_minutes: 123)
update_attrs = %{enabled: true}
refute_enqueued(worker: MediaCollectionIndexingWorker)
assert {:ok, %Source{}} = Sources.update_source(source, update_attrs)
assert_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id})
end
test "enabling a source will not create a task if the index frequency is 0" do
source = source_fixture(enabled: false, index_frequency_minutes: 0)
update_attrs = %{enabled: true}
refute_enqueued(worker: MediaCollectionIndexingWorker)
assert {:ok, %Source{}} = Sources.update_source(source, update_attrs)
refute_enqueued(worker: MediaCollectionIndexingWorker)
end
end
describe "update_source/3 when testing fast indexing" do
test "enabling fast_index will schedule a fast indexing task" do test "enabling fast_index will schedule a fast indexing task" do
source = source_fixture(fast_index: false) source = source_fixture(fast_index: false)
update_attrs = %{fast_index: true} update_attrs = %{fast_index: true}
@ -593,6 +634,45 @@ defmodule Pinchflat.SourcesTest do
assert source.index_frequency_minutes == 0 assert source.index_frequency_minutes == 0
end end
test "disabling a source will delete any pending tasks" do
source = source_fixture()
update_attrs = %{enabled: false}
{:ok, job} = Oban.insert(FastIndexingWorker.new(%{"id" => source.id}))
task = task_fixture(source_id: source.id, job_id: job.id)
assert {:ok, %Source{}} = Sources.update_source(source, update_attrs)
assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end
end
test "updating fast indexing will not create a task if the source is disabled" do
source = source_fixture(enabled: false, fast_index: false)
update_attrs = %{fast_index: true}
refute_enqueued(worker: FastIndexingWorker)
assert {:ok, %Source{}} = Sources.update_source(source, update_attrs)
refute_enqueued(worker: FastIndexingWorker)
end
test "enabling a source will create a task if fast_index is true" do
source = source_fixture(enabled: false, fast_index: true)
update_attrs = %{enabled: true}
refute_enqueued(worker: FastIndexingWorker)
assert {:ok, %Source{}} = Sources.update_source(source, update_attrs)
assert_enqueued(worker: FastIndexingWorker, args: %{"id" => source.id})
end
test "enabling a source will not create a task if fast_index is false" do
source = source_fixture(enabled: false, fast_index: false)
update_attrs = %{enabled: true}
refute_enqueued(worker: FastIndexingWorker)
assert {:ok, %Source{}} = Sources.update_source(source, update_attrs)
refute_enqueued(worker: FastIndexingWorker)
end
end end
describe "update_source/3 when testing options" do describe "update_source/3 when testing options" do