Misc refactors 2024-03-14 (#89)

* Adds method to improve cleanup of empty directories

* resolved bug where source metadata worker could call itself in an infinite loop

* Refactored file deletion for media items

* Removed useless filesystem data worker

* Updated task listing fns to take a record directly

* Refactored the way I call workers

* Improved some tests

lib/pinchflat/boot/pre_job_startup_tasks.ex

@@ -14,6 +14,7 @@ defmodule Pinchflat.Boot.PreJobStartupTasks do
 
   alias Pinchflat.Repo
   alias Pinchflat.Settings
+  alias Pinchflat.Filesystem.FilesystemHelpers
 
   def start_link(opts \\ []) do
     GenServer.start_link(__MODULE__, %{}, opts)
@@ -31,6 +32,7 @@ defmodule Pinchflat.Boot.PreJobStartupTasks do
   @impl true
   def init(state) do
     apply_default_settings()
+    ensure_directories_are_writeable()
     rename_old_job_workers()
 
     {:ok, state}
@@ -41,6 +43,21 @@ defmodule Pinchflat.Boot.PreJobStartupTasks do
     Settings.fetch!(:pro_enabled, false)
   end
 
+  defp ensure_directories_are_writeable do
+    directories = [
+      Application.get_env(:pinchflat, :media_directory),
+      Application.get_env(:pinchflat, :tmpfile_directory),
+      Application.get_env(:pinchflat, :metadata_directory)
+    ]
+
+    Enum.each(directories, fn dir ->
+      file = Path.join([dir, ".keep"])
+
+      # This will fail if the directory is not writeable, stopping boot
+      FilesystemHelpers.write_p!(file, "")
+    end)
+  end
+
   # As part of a large refactor, I ended up moving a bunch of workers around. This
   # is a problem because the workers are stored in the database and the runner
   # will try to run the OLD jobs. This is also why these tasks run before the job
@@ -52,7 +69,6 @@ defmodule Pinchflat.Boot.PreJobStartupTasks do
     rename_map = [
       ["Pinchflat.Workers.MediaIndexingWorker", "Pinchflat.FastIndexing.MediaIndexingWorker"],
       ["Pinchflat.Workers.MediaDownloadWorker", "Pinchflat.Downloading.MediaDownloadWorker"],
-      ["Pinchflat.Workers.FilesystemDataWorker", "Pinchflat.Filesystem.FilesystemDataWorker"],
       ["Pinchflat.Workers.FastIndexingWorker", "Pinchflat.FastIndexing.FastIndexingWorker"],
       ["Pinchflat.Workers.MediaCollectionIndexingWorker", "Pinchflat.SlowIndexing.MediaCollectionIndexingWorker"],
       ["Pinchflat.Workers.DataBackfillWorker", "Pinchflat.Boot.DataBackfillWorker"]

lib/pinchflat/downloading/download_option_builder.ex

@@ -99,7 +99,6 @@ defmodule Pinchflat.Downloading.DownloadOptionBuilder do
       :"480p" -> [format_sort: "res:480,#{video_codec_options}"]
       :"720p" -> [format_sort: "res:720,#{video_codec_options}"]
       :"1080p" -> [format_sort: "res:1080,#{video_codec_options}"]
-      :"1440p" -> [format_sort: "res:1440,#{video_codec_options}"]
       :"2160p" -> [format_sort: "res:2160,#{video_codec_options}"]
     end
   end

lib/pinchflat/downloading/downloading_helpers.ex

@@ -26,11 +26,7 @@ defmodule Pinchflat.Downloading.DownloadingHelpers do
   def enqueue_pending_download_tasks(%Source{download_media: true} = source) do
     source
     |> Media.list_pending_media_items_for()
-    |> Enum.each(fn media_item ->
-      %{id: media_item.id}
-      |> MediaDownloadWorker.new()
-      |> Tasks.create_job_with_task(media_item)
-    end)
+    |> Enum.each(&MediaDownloadWorker.kickoff_with_task/1)
   end
 
   def enqueue_pending_download_tasks(%Source{download_media: false}) do

lib/pinchflat/downloading/media_download_worker.ex

@@ -6,19 +6,30 @@ defmodule Pinchflat.Downloading.MediaDownloadWorker do
     unique: [period: :infinity, states: [:available, :scheduled, :retryable, :executing]],
     tags: ["media_item", "media_fetching"]
 
+  alias __MODULE__
+  alias Pinchflat.Tasks
   alias Pinchflat.Repo
   alias Pinchflat.Media
-  alias Pinchflat.Tasks
   alias Pinchflat.Downloading.MediaDownloader
-  alias Pinchflat.Filesystem.FilesystemDataWorker
 
-  @impl Oban.Worker
+  @doc """
+  Starts the media_item media download worker and creates a task for the media_item.
+
+  Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}}
+  """
+  def kickoff_with_task(media_item, opts \\ []) do
+    %{id: media_item.id}
+    |> MediaDownloadWorker.new(opts)
+    |> Tasks.create_job_with_task(media_item)
+  end
+
   @doc """
   For a given media item, download the media alongside any options.
   Does not download media if its source is set to not download media.
 
   Returns :ok | {:ok, %MediaItem{}} | {:error, any, ...any}
   """
+  @impl Oban.Worker
   def perform(%Oban.Job{args: %{"id" => media_item_id}}) do
     media_item =
       media_item_id
@@ -35,22 +46,23 @@ defmodule Pinchflat.Downloading.MediaDownloadWorker do
 
   defp download_media_and_schedule_jobs(media_item) do
     case MediaDownloader.download_for_media_item(media_item) do
-      {:ok, _} ->
-        schedule_filesystem_data_worker(media_item)
-        {:ok, media_item}
+      {:ok, updated_media_item} ->
+        compute_and_save_media_filesize(updated_media_item)
+
+        {:ok, updated_media_item}
 
       err ->
         err
     end
   end
 
-  defp schedule_filesystem_data_worker(media_item) do
-    %{id: media_item.id}
-    |> FilesystemDataWorker.new()
-    |> Tasks.create_job_with_task(media_item)
-    |> case do
-      {:ok, task} -> {:ok, task}
-      {:error, :duplicate_job} -> {:ok, :job_exists}
+  defp compute_and_save_media_filesize(media_item) do
+    case File.stat(media_item.media_filepath) do
+      {:ok, %{size: size}} ->
+        Media.update_media_item(media_item, %{media_size_bytes: size})
+
+      _ ->
+        :ok
     end
   end
 end

lib/pinchflat/downloading/media_downloader.ex

@@ -38,6 +38,8 @@ defmodule Pinchflat.Downloading.MediaDownloader do
             media_downloaded_at: DateTime.utc_now(),
             nfo_filepath: determine_nfo_filepath(item_with_preloads, parsed_json),
             metadata: %{
+              # IDEA: might be worth kicking off a job for this since thumbnail fetching
+              # could fail and I want to handle that in isolation
               metadata_filepath: MetadataFileHelpers.compress_and_store_metadata_for(media_item, parsed_json),
               thumbnail_filepath: MetadataFileHelpers.download_and_store_thumbnail_for(media_item, parsed_json)
             }

lib/pinchflat/fast_indexing/fast_indexing_helpers.ex

@@ -6,34 +6,13 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do
   """
 
   alias Pinchflat.Media
-  alias Pinchflat.Tasks
   alias Pinchflat.Sources.Source
   alias Pinchflat.FastIndexing.YoutubeRss
-  alias Pinchflat.FastIndexing.FastIndexingWorker
   alias Pinchflat.Downloading.MediaDownloadWorker
   alias Pinchflat.FastIndexing.MediaIndexingWorker
 
   alias Pinchflat.YtDlp.Media, as: YtDlpMedia
 
-  @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 """
   Fetches new media IDs from a source's YouTube RSS feed and kicks off indexing tasks
   for any new media items. See comments in `MediaIndexingWorker` for more info on the
@@ -54,9 +33,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do
     Enum.each(new_media_ids, fn media_id ->
       url = "https://www.youtube.com/watch?v=#{media_id}"
 
-      %{id: source.id, media_url: url}
-      |> MediaIndexingWorker.new()
-      |> Tasks.create_job_with_task(source)
+      MediaIndexingWorker.kickoff_with_task(source, url)
     end)
   end
 
@@ -74,9 +51,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do
     case maybe_media_item do
       {:ok, media_item} ->
         if source.download_media && Media.pending_download?(media_item) do
-          %{id: media_item.id}
-          |> MediaDownloadWorker.new()
-          |> Tasks.create_job_with_task(media_item)
+          MediaDownloadWorker.kickoff_with_task(media_item)
         end
 
         {:ok, media_item}

lib/pinchflat/fast_indexing/fast_indexing_worker.ex

@@ -12,7 +12,17 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do
   alias Pinchflat.Sources.Source
   alias Pinchflat.FastIndexing.FastIndexingHelpers
 
-  @impl Oban.Worker
+  @doc """
+  Starts the source fast indexing worker and creates a task for the source.
+
+  Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}}
+  """
+  def kickoff_with_task(source, opts \\ []) do
+    %{id: source.id}
+    |> FastIndexingWorker.new(opts)
+    |> Tasks.create_job_with_task(source)
+  end
+
   @doc """
   Kicks off the fast indexing process for a source, reschedules the job to run again
   once complete. See `MediaCollectionIndexingWorker` and `MediaIndexingWorker` comments
@@ -20,6 +30,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do
 
   Returns :ok | {:ok, :job_exists} | {:ok, %Task{}}
   """
+  @impl Oban.Worker
   def perform(%Oban.Job{args: %{"id" => source_id}}) do
     source = Sources.get_source!(source_id)
 
@@ -35,10 +46,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do
   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
+    case kickoff_with_task(source, schedule_in: next_run_in) do
       {:ok, task} -> {:ok, task}
       {:error, :duplicate_job} -> {:ok, :job_exists}
     end

lib/pinchflat/fast_indexing/media_indexing_worker.ex

@@ -8,10 +8,22 @@ defmodule Pinchflat.FastIndexing.MediaIndexingWorker do
 
   require Logger
 
+  alias __MODULE__
+  alias Pinchflat.Tasks
   alias Pinchflat.Sources
   alias Pinchflat.FastIndexing.FastIndexingHelpers
 
-  @impl Oban.Worker
+  @doc """
+  Starts the fast media indexing worker and creates a task for the source.
+
+  Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}}
+  """
+  def kickoff_with_task(source, media_url, opts \\ []) do
+    %{id: source.id, media_url: media_url}
+    |> MediaIndexingWorker.new(opts)
+    |> Tasks.create_job_with_task(source)
+  end
+
   @doc """
   Similar to `MediaCollectionIndexingWorker`, but for individual media items.
   Does not reschedule or check anything to do with a source's indexing
@@ -37,6 +49,7 @@ defmodule Pinchflat.FastIndexing.MediaIndexingWorker do
 
   Returns :ok
   """
+  @impl Oban.Worker
   def perform(%Oban.Job{args: %{"id" => source_id, "media_url" => media_url}}) do
     source = Sources.get_source!(source_id)
 

lib/pinchflat/filesystem/filesystem_data_worker.ex

@@ -1,31 +0,0 @@
-defmodule Pinchflat.Filesystem.FilesystemDataWorker do
-  @moduledoc false
-
-  use Oban.Worker,
-    queue: :local_metadata,
-    tags: ["media_item", "media_metadata", "local_metadata"],
-    max_attempts: 1
-
-  alias Pinchflat.Media
-  alias Pinchflat.Filesystem.FilesystemHelpers
-
-  @impl Oban.Worker
-  @doc """
-  For a given media item, compute and save metadata about the file on-disk.
-
-  IDEA: does this have to be a standalone job? I originally split it out
-  so a failure here wouldn't cause a downloader job retry, but I can match
-  for failures so it doesn't retry.
-
-  Returns :ok
-  """
-  def perform(%Oban.Job{args: %{"id" => media_item_id}}) do
-    media_item = Media.get_media_item!(media_item_id)
-
-    FilesystemHelpers.compute_and_save_media_filesize(media_item)
-
-    # Don't retry on failure - if it didn't work immediately there's no
-    # reason to believe it will work later.
-    :ok
-  end
-end

lib/pinchflat/filesystem/filesystem_helpers.ex

@@ -48,4 +48,36 @@ defmodule Pinchflat.Filesystem.FilesystemHelpers do
         err
     end
   end
+
+  @doc """
+  Deletes a file and removes any empty directories in the path.
+  Does NOT remove any directories that are not empty.
+
+  Returns :ok | {:error, any()}
+  """
+  def delete_file_and_remove_empty_directories(filepath) do
+    case File.rm(filepath) do
+      :ok ->
+        filepath
+        |> Path.dirname()
+        |> recursively_delete_empty_directories()
+
+      err ->
+        err
+    end
+  end
+
+  defp recursively_delete_empty_directories(directory) do
+    case File.rmdir(directory) do
+      :ok ->
+        directory
+        |> Path.dirname()
+        |> recursively_delete_empty_directories()
+
+      err ->
+        err
+    end
+
+    :ok
+  end
 end

lib/pinchflat/media/media.ex

@@ -7,9 +7,10 @@ defmodule Pinchflat.Media do
 
   alias Pinchflat.Repo
   alias Pinchflat.Tasks
-  alias Pinchflat.Media.MediaItem
   alias Pinchflat.Sources.Source
+  alias Pinchflat.Media.MediaItem
   alias Pinchflat.Metadata.MediaMetadata
+  alias Pinchflat.Filesystem.FilesystemHelpers
 
   @doc """
   Returns the list of media_items.
@@ -141,42 +142,6 @@ defmodule Pinchflat.Media do
   """
   def get_media_item!(id), do: Repo.get!(MediaItem, id)
 
-  @doc """
-  Produces a flat list of the filesystem paths for a media_item's downloaded files
-
-  NOTE: this can almost certainly be made private
-
-  Returns [binary()]
-  """
-  def media_filepaths(media_item) do
-    mapped_struct = Map.from_struct(media_item)
-
-    MediaItem.filepath_attributes()
-    |> Enum.map(fn
-      :subtitle_filepaths = field -> Enum.map(mapped_struct[field], fn [_, filepath] -> filepath end)
-      field -> List.wrap(mapped_struct[field])
-    end)
-    |> List.flatten()
-    |> Enum.filter(&is_binary/1)
-  end
-
-  @doc """
-  Produces a flat list of the filesystem paths for a media_item's metadata files.
-  Returns an empty list if the media_item has no metadata.
-
-  NOTE: this can almost certainly be made private
-
-  Returns [binary()] | []
-  """
-  def metadata_filepaths(media_item) do
-    metadata = Repo.preload(media_item, :metadata).metadata || %MediaMetadata{}
-    mapped_struct = Map.from_struct(metadata)
-
-    MediaMetadata.filepath_attributes()
-    |> Enum.map(fn field -> mapped_struct[field] end)
-    |> Enum.filter(&is_binary/1)
-  end
-
   @doc """
   Creates a media_item.
 
@@ -223,19 +188,20 @@ defmodule Pinchflat.Media do
   end
 
   @doc """
-  Deletes a media_item and its associated tasks.
-  Can optionally delete the media_item's files.
+  Deletes a media_item, its associated tasks, and our internal metadata files.
+  Can optionally delete the media_item's media files (media, thumbnail, subtitles, etc).
 
   Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}
   """
   def delete_media_item(%MediaItem{} = media_item, opts \\ []) do
     delete_files = Keyword.get(opts, :delete_files, false)
 
-    # NOTE: this should delete metadata no matter what
     if delete_files do
-      {:ok, _} = delete_all_attachments(media_item)
+      {:ok, _} = delete_media_files(media_item)
     end
 
+    # Should delete these no matter what
+    delete_internal_metadata_files(media_item)
     Tasks.delete_tasks_for(media_item)
     Repo.delete(media_item)
   end
@@ -247,27 +213,31 @@ defmodule Pinchflat.Media do
     MediaItem.changeset(media_item, attrs)
   end
 
-  # NOTE: refactor this
-  defp delete_all_attachments(media_item) do
-    media_item = Repo.preload(media_item, :metadata)
+  defp delete_media_files(media_item) do
+    mapped_struct = Map.from_struct(media_item)
 
-    media_item
-    |> media_filepaths()
-    |> Enum.concat(metadata_filepaths(media_item))
-    |> Enum.each(&File.rm/1)
-
-    # rmdir will attempt to delete the directory, but only if it is empty
-    if media_item.media_filepath do
-      File.rmdir(Path.dirname(media_item.media_filepath))
-    end
-
-    if media_item.metadata && media_item.metadata.metadata_filepath do
-      File.rmdir(Path.dirname(media_item.metadata.metadata_filepath))
-    end
+    MediaItem.filepath_attributes()
+    |> Enum.map(fn
+      :subtitle_filepaths = field -> Enum.map(mapped_struct[field], fn [_, filepath] -> filepath end)
+      field -> List.wrap(mapped_struct[field])
+    end)
+    |> List.flatten()
+    |> Enum.filter(&is_binary/1)
+    |> Enum.each(&FilesystemHelpers.delete_file_and_remove_empty_directories/1)
 
     {:ok, media_item}
   end
 
+  defp delete_internal_metadata_files(media_item) do
+    metadata = Repo.preload(media_item, :metadata).metadata || %MediaMetadata{}
+    mapped_struct = Map.from_struct(metadata)
+
+    MediaMetadata.filepath_attributes()
+    |> Enum.map(fn field -> mapped_struct[field] end)
+    |> Enum.filter(&is_binary/1)
+    |> Enum.each(&FilesystemHelpers.delete_file_and_remove_empty_directories/1)
+  end
+
   defp maybe_apply_cutoff_date(source) do
     if source.download_cutoff_date do
       dynamic([mi], mi.upload_date >= ^source.download_cutoff_date)

lib/pinchflat/metadata/source_metadata_storage_worker.ex

@@ -4,7 +4,10 @@ defmodule Pinchflat.Metadata.SourceMetadataStorageWorker do
   use Oban.Worker,
     queue: :remote_metadata,
     tags: ["media_source", "source_metadata", "remote_metadata"],
-    max_attempts: 1
+    max_attempts: 1,
+    # This is the only thing stopping this job from calling itself
+    # in an infinite loop.
+    unique: [period: 600]
 
   alias __MODULE__
   alias Pinchflat.Repo
@@ -16,27 +19,27 @@ defmodule Pinchflat.Metadata.SourceMetadataStorageWorker do
   @doc """
   Starts the source metadata storage worker and creates a task for the source.
 
-  IDEA: testing out this method of handling job kickoff. I think I like it, so
-  I may use it in other places. Just testing it for now
-
   Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}}
   """
-  def kickoff_with_task(source) do
+  def kickoff_with_task(source, opts \\ []) do
     %{id: source.id}
-    |> SourceMetadataStorageWorker.new()
+    |> SourceMetadataStorageWorker.new(opts)
     |> Tasks.create_job_with_task(source)
   end
 
-  @impl Oban.Worker
   @doc """
   Fetches and stores metadata for a source in the secret metadata location.
 
   Returns :ok
   """
+  @impl Oban.Worker
   def perform(%Oban.Job{args: %{"id" => source_id}}) do
     source = Repo.preload(Sources.get_source!(source_id), :metadata)
     {:ok, metadata} = MediaCollection.get_source_metadata(source.original_url)
 
+    # Since updating a source kicks this job off again, we enforce job uniqueness (above)
+    # to once, per source, per x minutes. This is to prevent a job from calling itself
+    # in an infinite loop.
     Sources.update_source(source, %{
       metadata: %{
         metadata_filepath: MetadataFileHelpers.compress_and_store_metadata_for(source, metadata)

lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex

@@ -13,7 +13,17 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorker do
   alias Pinchflat.FastIndexing.FastIndexingWorker
   alias Pinchflat.SlowIndexing.SlowIndexingHelpers
 
-  @impl Oban.Worker
+  @doc """
+  Starts the source slow indexing worker and creates a task for the source.
+
+  Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}}
+  """
+  def kickoff_with_task(source, opts \\ []) do
+    %{id: source.id}
+    |> MediaCollectionIndexingWorker.new(opts)
+    |> Tasks.create_job_with_task(source)
+  end
+
   @doc """
   The ID is that of a source _record_, not a YouTube channel/playlist ID. Indexes
   the provided source, kicks off downloads for each new MediaItem, and
@@ -58,6 +68,7 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorker do
 
   Returns :ok | {:ok, %Task{}}
   """
+  @impl Oban.Worker
   def perform(%Oban.Job{args: %{"id" => source_id}}) do
     source = Sources.get_source!(source_id)
 

lib/pinchflat/slow_indexing/slow_indexing_helpers.ex

@@ -31,10 +31,7 @@ defmodule Pinchflat.SlowIndexing.SlowIndexingHelpers do
     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)
+    MediaCollectionIndexingWorker.kickoff_with_task(source)
   end
 
   @doc """
@@ -125,9 +122,7 @@ defmodule Pinchflat.SlowIndexing.SlowIndexingHelpers do
         if source.download_media && Media.pending_download?(media_item) do
           Logger.debug("FileFollowerServer Handler: Enqueuing download task for #{inspect(media_attrs)}")
 
-          %{id: media_item.id}
-          |> MediaDownloadWorker.new()
-          |> Tasks.create_job_with_task(media_item)
+          MediaDownloadWorker.kickoff_with_task(media_item)
         end
 
       {:error, changeset} ->

lib/pinchflat/sources/sources.ex

@@ -12,8 +12,9 @@ defmodule Pinchflat.Sources do
   alias Pinchflat.Profiles.MediaProfile
   alias Pinchflat.YtDlp.MediaCollection
   alias Pinchflat.Metadata.SourceMetadata
+  alias Pinchflat.Filesystem.FilesystemHelpers
   alias Pinchflat.Downloading.DownloadingHelpers
-  alias Pinchflat.FastIndexing.FastIndexingHelpers
+  alias Pinchflat.FastIndexing.FastIndexingWorker
   alias Pinchflat.SlowIndexing.SlowIndexingHelpers
   alias Pinchflat.Metadata.SourceMetadataStorageWorker
 
@@ -108,8 +109,8 @@ defmodule Pinchflat.Sources do
       Media.delete_media_item(media_item, delete_files: delete_files)
     end)
 
-    Tasks.delete_tasks_for(source)
     delete_source_metadata_files(source)
+    Tasks.delete_tasks_for(source)
     Repo.delete(source)
   end
 
@@ -120,17 +121,9 @@ defmodule Pinchflat.Sources do
     Source.changeset(source, attrs, validation_stage)
   end
 
-  @doc """
-  Returns an `%Ecto.Changeset{}` for tracking source changes and additionally
-  fetches source details from the original_url (if provided). If the source
-  details cannot be fetched, an error is added to the changeset.
-
-  NOTE: When operating in the ideal path, this effectively adds an API call
-  to the source creation/update process. Should be used only when needed.
-
-  NOTE: this can almost certainly be made private now
-  """
-  def maybe_change_source_from_url(%Source{} = source, attrs) do
+  # NOTE: When operating in the ideal path, this effectively adds an API call
+  # to the source creation/update process. Should be used only when needed.
+  defp maybe_change_source_from_url(%Source{} = source, attrs) do
     case change_source(source, attrs) do
       %Ecto.Changeset{changes: %{original_url: _}} = changeset ->
         add_source_details_to_changeset(source, changeset)
@@ -149,7 +142,7 @@ defmodule Pinchflat.Sources do
       |> Enum.map(fn field -> mapped_struct[field] end)
       |> Enum.filter(&is_binary/1)
 
-    Enum.each(filepaths, &File.rm/1)
+    Enum.each(filepaths, &FilesystemHelpers.delete_file_and_remove_empty_directories/1)
   end
 
   defp add_source_details_to_changeset(source, changeset) do
@@ -270,7 +263,8 @@ defmodule Pinchflat.Sources do
   defp maybe_update_fast_indexing_task(changeset, source) do
     case changeset.changes do
       %{fast_index: true} ->
-        FastIndexingHelpers.kickoff_fast_indexing_task(source)
+        Tasks.delete_pending_tasks_for(source, "FastIndexingWorker")
+        FastIndexingWorker.kickoff_with_task(source)
 
       %{fast_index: false} ->
         Tasks.delete_pending_tasks_for(source, "FastIndexingWorker")

lib/pinchflat/tasks/tasks.ex

@@ -20,13 +20,17 @@ defmodule Pinchflat.Tasks do
   Returns the list of tasks for a given record type and ID. Optionally allows you to specify
   which worker or job states to include.
 
-  IDEA: this should be updated to take a struct instead of a record type and ID
-
   Returns [%Task{}, ...]
   """
-  def list_tasks_for(attached_record_type, attached_record_id, worker_name \\ nil, job_states \\ Oban.Job.states()) do
+  def list_tasks_for(record, worker_name \\ nil, job_states \\ Oban.Job.states()) do
     stringified_states = Enum.map(job_states, &to_string/1)
 
+    record_type =
+      case record do
+        %Source{} -> :source_id
+        %MediaItem{} -> :media_item_id
+      end
+
     worker_name_finder =
       if worker_name do
         # Workers are the full module name - we want to match on the string ENDING with
@@ -43,7 +47,7 @@ defmodule Pinchflat.Tasks do
     Repo.all(
       from t in Task,
         join: j in assoc(t, :job),
-        where: field(t, ^attached_record_type) == ^attached_record_id,
+        where: field(t, ^record_type) == ^record.id,
         where: ^worker_name_finder,
         where: j.state in ^stringified_states
     )
@@ -55,10 +59,9 @@ defmodule Pinchflat.Tasks do
 
   Returns [%Task{}, ...]
   """
-  def list_pending_tasks_for(attached_record_type, attached_record_id, worker_name \\ nil) do
+  def list_pending_tasks_for(record, worker_name \\ nil) do
     list_tasks_for(
-      attached_record_type,
-      attached_record_id,
+      record,
       worker_name,
       [:available, :scheduled, :retryable]
     )
@@ -128,14 +131,10 @@ defmodule Pinchflat.Tasks do
 
   Returns :ok
   """
-  def delete_tasks_for(attached_record, worker_name \\ nil) do
-    tasks =
-      case attached_record do
-        %Source{} = source -> list_tasks_for(:source_id, source.id, worker_name)
-        %MediaItem{} = media_item -> list_tasks_for(:media_item_id, media_item.id, worker_name)
-      end
-
-    Enum.each(tasks, &delete_task/1)
+  def delete_tasks_for(record, worker_name \\ nil) do
+    record
+    |> list_tasks_for(worker_name)
+    |> Enum.each(&delete_task/1)
   end
 
   @doc """
@@ -144,14 +143,10 @@ defmodule Pinchflat.Tasks do
 
   Returns :ok
   """
-  def delete_pending_tasks_for(attached_record, worker_name \\ nil) do
-    tasks =
-      case attached_record do
-        %Source{} = source -> list_pending_tasks_for(:source_id, source.id, worker_name)
-        %MediaItem{} = media_item -> list_pending_tasks_for(:media_item_id, media_item.id, worker_name)
-      end
-
-    Enum.each(tasks, &delete_task/1)
+  def delete_pending_tasks_for(record, worker_name \\ nil) do
+    record
+    |> list_pending_tasks_for(worker_name)
+    |> Enum.each(&delete_task/1)
   end
 
   @doc """

lib/pinchflat/yt_dlp/command_runner.ex

@@ -37,7 +37,8 @@ defmodule Pinchflat.YtDlp.CommandRunner do
 
     case System.cmd(command, formatted_command_opts, stderr_to_stdout: true) do
       {_, 0} ->
-        # IDEA: consider deleting the file after reading it
+        # IDEA: consider deleting the file after reading it. It's in the tmp dir, so it's not
+        # a huge deal, but it's still a good idea to clean up after ourselves.
         # (even on error? especially on error?)
         File.read(output_filepath)
 

lib/pinchflat/yt_dlp/media_collection.ex

@@ -6,7 +6,6 @@ defmodule Pinchflat.YtDlp.MediaCollection do
 
   require Logger
 
-  alias Pinchflat.Utils.FunctionUtils
   alias Pinchflat.Filesystem.FilesystemHelpers
   alias Pinchflat.YtDlp.Media, as: YtDlpMedia
 
@@ -36,11 +35,20 @@ defmodule Pinchflat.YtDlp.MediaCollection do
 
     case runner.run(url, command_opts, output_template, output_filepath: output_filepath) do
       {:ok, output} ->
-        output
-        |> String.split("\n", trim: true)
-        |> Enum.map(&Phoenix.json_library().decode!/1)
-        |> Enum.map(&YtDlpMedia.response_to_struct/1)
-        |> FunctionUtils.wrap_ok()
+        parsed_lines =
+          output
+          |> String.split("\n", trim: true)
+          |> Enum.map(fn line ->
+            case Phoenix.json_library().decode(line) do
+              {:ok, parsed_json} ->
+                YtDlpMedia.response_to_struct(parsed_json)
+
+              _ ->
+                nil
+            end
+          end)
+
+        {:ok, Enum.filter(parsed_lines, &(&1 != nil))}
 
       res ->
         res