Added jobs for enqueuing/dequeuing media based on source's status

This commit is contained in:
Kieran Eglin 2024-02-09 16:43:35 -08:00
parent 8bdd189fa5
commit aeef2c7588
No known key found for this signature in database
GPG key ID: 193984967FCF432D
6 changed files with 87 additions and 20 deletions

View file

@ -35,7 +35,7 @@ defmodule Pinchflat.MediaSource do
def create_source(attrs) do
%Source{}
|> change_source_from_url(attrs)
|> commit_and_start_indexing()
|> commit_and_handle_tasks()
end
@doc """
@ -51,7 +51,7 @@ defmodule Pinchflat.MediaSource do
def update_source(%Source{} = source, attrs) do
source
|> change_source_from_url(attrs)
|> commit_and_start_indexing()
|> commit_and_handle_tasks()
end
@doc """
@ -139,13 +139,34 @@ defmodule Pinchflat.MediaSource do
change_source(source, Map.merge(changes, collection_changes))
end
defp commit_and_start_indexing(changeset) do
defp commit_and_handle_tasks(changeset) do
case Repo.insert_or_update(changeset) do
{:ok, %Source{} = source} -> maybe_run_indexing_task(changeset, source)
err -> err
{:ok, %Source{} = source} ->
maybe_handle_media_tasks(changeset, source)
maybe_run_indexing_task(changeset, source)
err ->
err
end
end
# If the source is NOT new (ie: updated) and the download_media flag has changed,
# enqueue or dequeue media download tasks as necessary.
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)
{%{__meta__: %{state: :loaded}}, %{download_media: false}} ->
SourceTasks.dequeue_pending_media_tasks(source)
_ ->
:ok
end
{:ok, source}
end
defp maybe_run_indexing_task(changeset, source) do
case changeset.data do
# If the changeset is new (not persisted), attempt indexing no matter what

View file

@ -117,9 +117,7 @@ defmodule Pinchflat.Tasks do
%MediaItem{} = media_item -> list_tasks_for(:media_item_id, media_item.id)
end
Enum.each(tasks, fn task ->
delete_task(task)
end)
Enum.each(tasks, &delete_task/1)
end
@doc """
@ -134,9 +132,7 @@ defmodule Pinchflat.Tasks do
%MediaItem{} = media_item -> list_pending_tasks_for(:media_item_id, media_item.id)
end
Enum.each(tasks, fn task ->
delete_task(task)
end)
Enum.each(tasks, &delete_task/1)
end
@doc """

View file

@ -73,7 +73,7 @@ defmodule Pinchflat.Tasks.SourceTasks do
Returns :ok
"""
def enqueue_pending_media_downloads(%Source{download_media: true} = source) do
def enqueue_pending_media_tasks(%Source{download_media: true} = source) do
source
|> Media.list_pending_media_items_for()
|> Enum.each(fn media_item ->
@ -84,7 +84,18 @@ defmodule Pinchflat.Tasks.SourceTasks do
end)
end
def enqueue_pending_media_downloads(%Source{download_media: false} = _source) do
def enqueue_pending_media_tasks(%Source{download_media: false} = _source) 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
end

View file

@ -49,7 +49,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorker do
defp index_media_and_reschedule(source) do
SourceTasks.index_media_items(source)
# This method handles the case where a source is set to not download media
SourceTasks.enqueue_pending_media_downloads(source)
SourceTasks.enqueue_pending_media_tasks(source)
source
|> Map.take([:id])

View file

@ -2,12 +2,15 @@ defmodule Pinchflat.MediaSourceTest do
use Pinchflat.DataCase
import Mox
import Pinchflat.TasksFixtures
import Pinchflat.MediaFixtures
import Pinchflat.ProfilesFixtures
import Pinchflat.MediaSourceFixtures
alias Pinchflat.MediaSource
alias Pinchflat.Tasks.SourceTasks
alias Pinchflat.MediaSource.Source
alias Pinchflat.Workers.MediaIndexingWorker
alias Pinchflat.Workers.VideoDownloadWorker
@invalid_source_attrs %{name: nil, collection_id: nil}
@ -173,6 +176,27 @@ defmodule Pinchflat.MediaSourceTest do
refute_enqueued(worker: MediaIndexingWorker, args: %{"id" => source.id})
end
test "enabling the download_media attribute will schedule a download task" do
source = source_fixture(download_media: false)
media_item = media_item_fixture(source_id: source.id, media_filepath: nil)
update_attrs = %{download_media: true}
refute_enqueued(worker: VideoDownloadWorker)
assert {:ok, %Source{}} = MediaSource.update_source(source, update_attrs)
assert_enqueued(worker: VideoDownloadWorker, args: %{"id" => media_item.id})
end
test "disabling the download_media attribute will cancel the download task" 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)
assert_enqueued(worker: VideoDownloadWorker, args: %{"id" => media_item.id})
assert {:ok, %Source{}} = MediaSource.update_source(source, update_attrs)
refute_enqueued(worker: VideoDownloadWorker)
end
test "updates with invalid data returns error changeset" do
source = source_fixture()

View file

@ -105,12 +105,12 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
end
end
describe "enqueue_pending_media_downloads/1" do
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_downloads(source)
assert :ok = SourceTasks.enqueue_pending_media_tasks(source)
assert_enqueued(worker: VideoDownloadWorker, args: %{"id" => media_item.id})
end
@ -119,7 +119,7 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
source = source_fixture()
_media_item = media_item_fixture(source_id: source.id, media_filepath: "some/filepath.mp4")
assert :ok = SourceTasks.enqueue_pending_media_downloads(source)
assert :ok = SourceTasks.enqueue_pending_media_tasks(source)
refute_enqueued(worker: VideoDownloadWorker)
end
@ -130,7 +130,7 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
assert [] = Tasks.list_tasks_for(:media_item_id, media_item.id)
assert :ok = SourceTasks.enqueue_pending_media_downloads(source)
assert :ok = SourceTasks.enqueue_pending_media_tasks(source)
assert [_] = Tasks.list_tasks_for(:media_item_id, media_item.id)
end
@ -138,7 +138,7 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
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_downloads(source)
assert :ok = SourceTasks.enqueue_pending_media_tasks(source)
refute_enqueued(worker: VideoDownloadWorker)
end
@ -147,7 +147,22 @@ defmodule Pinchflat.Tasks.SourceTasksTest 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_downloads(source)
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: VideoDownloadWorker, args: %{"id" => media_item.id})
assert :ok = SourceTasks.dequeue_pending_media_tasks(source)
refute_enqueued(worker: VideoDownloadWorker)
assert [] = Tasks.list_tasks_for(:media_item_id, media_item.id)
end
end