Media item lifecycle improvements (#18)

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

* adds lifecycle columns to source and media_item tables

lib/pinchflat/media/media_item.ex

@@ -15,11 +15,12 @@ defmodule Pinchflat.Media.MediaItem do
     media_id
     original_url
     livestream
+    media_downloaded_at
     media_filepath
-    source_id
     subtitle_filepaths
     thumbnail_filepath
     metadata_filepath
+    source_id
   )a
   @required_fields ~w(title original_url livestream media_id source_id)a
 
@@ -28,6 +29,8 @@ defmodule Pinchflat.Media.MediaItem do
     field :media_id, :string
     field :original_url, :string
     field :livestream, :boolean, default: false
+    field :media_downloaded_at, :utc_datetime
+
     field :media_filepath, :string
     field :thumbnail_filepath, :string
     field :metadata_filepath, :string

lib/pinchflat/media_client/video_downloader.ex

@@ -35,7 +35,11 @@ defmodule Pinchflat.MediaClient.VideoDownloader do
     case download_for_media_profile(media_item.media_id, media_profile, backend) do
       {:ok, parsed_json} ->
         parser = metadata_parser(backend)
-        parsed_attrs = parser.parse_for_media_item(parsed_json)
+
+        parsed_attrs =
+          parsed_json
+          |> parser.parse_for_media_item()
+          |> Map.merge(%{media_downloaded_at: DateTime.utc_now()})
 
         # Don't forgor to use preloaded associations or updates to
         # associations won't work!

lib/pinchflat/media_source.ex

@@ -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,38 @@ 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
+
+  # IDEA: this uses a pattern where `kickoff_indexing_task` controls whether
+  # it should run based on the source, but `maybe_handle_media_tasks` handles that
+  # logic itself. Consider updating one or the other to be consistent (once I've
+  # decided which I like more)
   defp maybe_run_indexing_task(changeset, source) do
     case changeset.data do
       # If the changeset is new (not persisted), attempt indexing no matter what

lib/pinchflat/media_source/source.ex

@@ -16,11 +16,19 @@ defmodule Pinchflat.MediaSource.Source do
     friendly_name
     index_frequency_minutes
     download_media
+    last_indexed_at
     original_url
     media_profile_id
   )a
 
-  @required_fields @allowed_fields -- ~w(index_frequency_minutes friendly_name)a
+  @required_fields ~w(
+    collection_name
+    collection_id
+    collection_type
+    download_media
+    original_url
+    media_profile_id
+  )a
 
   schema "sources" do
     field :friendly_name, :string
@@ -29,6 +37,7 @@ defmodule Pinchflat.MediaSource.Source do
     field :collection_type, Ecto.Enum, values: [:channel, :playlist]
     field :index_frequency_minutes, :integer, default: 60 * 24
     field :download_media, :boolean, default: true
+    field :last_indexed_at, :utc_datetime
     # This should only be used for user reference going forward
     # as the collection_id should be used for all API calls
     field :original_url, :string

lib/pinchflat/tasks.ex

@@ -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 """

lib/pinchflat/tasks/source_tasks.ex

@@ -5,6 +5,7 @@ defmodule Pinchflat.Tasks.SourceTasks do
 
   alias Pinchflat.Media
   alias Pinchflat.Tasks
+  alias Pinchflat.MediaSource
   alias Pinchflat.MediaSource.Source
   alias Pinchflat.MediaClient.SourceDetails
   alias Pinchflat.Workers.MediaIndexingWorker
@@ -42,6 +43,7 @@ defmodule Pinchflat.Tasks.SourceTasks do
   """
   def index_media_items(%Source{} = source) do
     {:ok, media_attributes} = SourceDetails.get_media_attributes(source.original_url)
+    MediaSource.update_source(source, %{last_indexed_at: DateTime.utc_now()})
 
     media_attributes
     |> Enum.map(fn media_attrs ->
@@ -73,7 +75,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 +86,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

lib/pinchflat/workers/media_indexing_worker.ex

@@ -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])