Download indexed videos (#10)

* Clarified documentation

* more comments

* [WIP] hooked up basic video downloading; starting work on metadata

* Added metadata model and parsing

Adding the metadata model made me realize that, in many cases, yt-dlp
returns undesired input in stdout, breaking parsing. In order to get
the metadata model working, I had to change the way in which the app
interacts with yt-dlp. Now, output is written as a file to disk which
is immediately re-read and returned.

* Added tests for video download worker

* Hooked up video downloading to the channel indexing pipeline

* Adds tasks for media items

* Updated video metadata parser to extract the title

.gitignore

@@ -38,3 +38,4 @@ npm-debug.log
 
 /.elixir_ls
 .env
+.DS_Store

Dockerfile

@@ -29,5 +29,7 @@ RUN chmod +x ./docker-run.sh
 
 # Install Elixir deps
 RUN mix deps.get
+# Gives us iex shell history
+ENV ERL_AFLAGS="-kernel shell_history enabled"
 
 EXPOSE 4008

config/config.exs

@@ -14,7 +14,8 @@ config :pinchflat,
   yt_dlp_executable: System.find_executable("yt-dlp"),
   yt_dlp_runner: Pinchflat.MediaClient.Backends.YtDlp.CommandRunner,
   # TODO: figure this out
-  media_directory: :not_implemented
+  media_directory: :not_implemented,
+  metadata_directory: Path.join([System.tmp_dir!(), "pinchflat", "metadata"])
 
 # Configures the endpoint
 config :pinchflat, PinchflatWeb.Endpoint,

config/dev.exs

@@ -1,7 +1,8 @@
 import Config
 
 config :pinchflat,
-  media_directory: Path.join([System.tmp_dir!(), "yt-dlp"])
+  media_directory: Path.join([File.cwd!(), "tmp", "videos"]),
+  metadata_directory: Path.join([File.cwd!(), "tmp", "metadata"])
 
 # Configure your database
 config :pinchflat, Pinchflat.Repo,

config/test.exs

@@ -3,7 +3,8 @@ import Config
 config :pinchflat,
   # Specifying backend data here makes mocking and local testing SUPER easy
   yt_dlp_executable: Path.join([File.cwd!(), "/test/support/scripts/yt-dlp-mocks/repeater.sh"]),
-  media_directory: Path.join([System.tmp_dir!(), "yt-dlp"])
+  media_directory: Path.join([System.tmp_dir!(), "videos"]),
+  metadata_directory: Path.join([System.tmp_dir!(), "metadata"])
 
 config :pinchflat, Oban, testing: :manual
 

lib/pinchflat/media.ex

@@ -4,9 +4,11 @@ defmodule Pinchflat.Media do
   """
 
   import Ecto.Query, warn: false
-  alias Pinchflat.Repo
 
+  alias Pinchflat.Repo
+  alias Pinchflat.Tasks
   alias Pinchflat.Media.MediaItem
+  alias Pinchflat.MediaSource.Channel
 
   @doc """
   Returns the list of media_items. Returns [%MediaItem{}, ...].
@@ -15,6 +17,20 @@ defmodule Pinchflat.Media do
     Repo.all(MediaItem)
   end
 
+  @doc """
+  Returns a list of pending media_items for a given channel, where
+  pending means the `video_filepath` is `nil`.
+
+  Returns [%MediaItem{}, ...].
+  """
+  def list_pending_media_items_for(%Channel{} = channel) do
+    from(
+      m in MediaItem,
+      where: m.channel_id == ^channel.id and is_nil(m.video_filepath)
+    )
+    |> Repo.all()
+  end
+
   @doc """
   Gets a single media_item.
 
@@ -25,7 +41,7 @@ defmodule Pinchflat.Media do
   @doc """
   Creates a media_item. Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}.
   """
-  def create_media_item(attrs \\ %{}) do
+  def create_media_item(attrs) do
     %MediaItem{}
     |> MediaItem.changeset(attrs)
     |> Repo.insert()
@@ -41,9 +57,12 @@ defmodule Pinchflat.Media do
   end
 
   @doc """
-  Deletes a media_item. Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}.
+  Deletes a media_item and its associated tasks.
+
+  Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}.
   """
   def delete_media_item(%MediaItem{} = media_item) do
+    Tasks.delete_tasks_for(media_item)
     Repo.delete(media_item)
   end
 

lib/pinchflat/media/media_item.ex

@@ -6,13 +6,13 @@ defmodule Pinchflat.Media.MediaItem do
   use Ecto.Schema
   import Ecto.Changeset
 
+  alias Pinchflat.Tasks.Task
   alias Pinchflat.MediaSource.Channel
+  alias Pinchflat.Media.MediaMetadata
 
   @required_fields ~w(media_id channel_id)a
   @allowed_fields ~w(title media_id video_filepath channel_id)a
 
-  # IDEA: consider making an attached `metadata` model to store the JSON response from whatever backend is used
-
   schema "media_items" do
     field :title, :string
     field :media_id, :string
@@ -20,6 +20,10 @@ defmodule Pinchflat.Media.MediaItem do
 
     belongs_to :channel, Channel
 
+    has_one :metadata, MediaMetadata, on_replace: :update
+
+    has_many :tasks, Task
+
     timestamps(type: :utc_datetime)
   end
 
@@ -27,6 +31,7 @@ defmodule Pinchflat.Media.MediaItem do
   def changeset(media_item, attrs) do
     media_item
     |> cast(attrs, @allowed_fields)
+    |> cast_assoc(:metadata, with: &MediaMetadata.changeset/2, required: false)
     |> validate_required(@required_fields)
     |> unique_constraint([:media_id, :channel_id])
   end

lib/pinchflat/media/media_metadata.ex

@@ -0,0 +1,28 @@
+defmodule Pinchflat.Media.MediaMetadata do
+  @moduledoc """
+  The MediaMetadata schema.
+
+  Look. Don't @ me about Metadata vs. Metadatum. I'm very sensitive.
+  """
+
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  alias Pinchflat.Media.MediaItem
+
+  schema "media_metadata" do
+    field :client_response, :map
+
+    belongs_to :media_item, MediaItem
+
+    timestamps(type: :utc_datetime)
+  end
+
+  @doc false
+  def changeset(media_metadata, attrs) do
+    media_metadata
+    |> cast(attrs, [:client_response])
+    |> validate_required([:client_response])
+    |> unique_constraint([:media_item_id])
+  end
+end

lib/pinchflat/media_client/backends/backend_command_runner.ex

@@ -3,5 +3,5 @@ defmodule Pinchflat.MediaClient.Backends.BackendCommandRunner do
   A behaviour for running CLI commands against a downloader backend
   """
 
-  @callback run(binary(), keyword()) :: {:ok, binary()} | {:error, binary(), integer()}
+  @callback run(binary(), keyword(), binary()) :: {:ok, binary()} | {:error, binary(), integer()}
 end

lib/pinchflat/media_client/backends/yt_dlp/channel.ex

@@ -16,9 +16,9 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.Channel do
   Returns {:ok, %ChannelDetails{}} | {:error, any, ...}.
   """
   def get_channel_details(channel_url) do
-    opts = [print: "%(.{channel,channel_id})j", playlist_end: 1]
+    opts = [playlist_end: 1]
 
-    with {:ok, output} <- backend_runner().run(channel_url, opts),
+    with {:ok, output} <- backend_runner().run(channel_url, opts, "%(.{channel,channel_id})j"),
          {:ok, parsed_json} <- Phoenix.json_library().decode(output) do
       {:ok, ChannelDetails.new(parsed_json["channel_id"], parsed_json["channel"])}
     else

lib/pinchflat/media_client/backends/yt_dlp/command_runner.ex

@@ -3,31 +3,57 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.CommandRunner do
   Runs yt-dlp commands using the `System.cmd/3` function
   """
 
+  require Logger
+
   alias Pinchflat.Utils.StringUtils
   alias Pinchflat.MediaClient.Backends.BackendCommandRunner
 
   @behaviour BackendCommandRunner
 
   @doc """
-  Runs a yt-dlp command and returns the string output
+  Runs a yt-dlp command and returns the string output. Saves the output to
+  a file and then returns its contents because yt-dlp will return warnings
+  to stdout even if the command is successful, but these will break JSON parsing.
 
   Returns {:ok, binary()} | {:error, output, status}.
 
-  # IDEA: deduplicate command opts, keeping the last one on conflict
-          although possibly not needed (and a LOT easier) if yt-dlp
-          just ignores duplicate options (ie: look into that)
+  IDEA: Indexing takes a long time, but the output is actually streamed to stdout.
+  Maybe we could listen to that stream instead so we can index videos as they're discovered.
+  See: https://stackoverflow.com/a/49061086/5665799
   """
   @impl BackendCommandRunner
-  def run(url, command_opts) do
+  def run(url, command_opts, output_template) do
     command = backend_executable()
-    formatted_command_opts = parse_options(command_opts) ++ [url]
+    # These must stay in exactly this order, hence why I'm giving it its own variable.
+    # Also, can't use RAM file since yt-dlp needs a concrete filepath.
+    json_output_path = generate_json_output_path()
+    print_to_file_opts = [{:print_to_file, output_template}, json_output_path]
+    formatted_command_opts = [url] ++ parse_options(command_opts ++ print_to_file_opts)
+
+    Logger.debug("[yt-dlp] called with: #{Enum.join(formatted_command_opts, " ")}")
 
     case System.cmd(command, formatted_command_opts, stderr_to_stdout: true) do
-      {output, 0} -> {:ok, output}
-      {output, status} -> {:error, output, status}
+      {_, 0} ->
+        # IDEA: consider deleting the file after reading it
+        # (even on error? especially on error?)
+        File.read(json_output_path)
+
+      {output, status} ->
+        {:error, output, status}
     end
   end
 
+  defp generate_json_output_path do
+    metadata_directory = Application.get_env(:pinchflat, :metadata_directory)
+    filepath = Path.join([metadata_directory, "#{StringUtils.random_string(64)}.json"])
+
+    # Ensure the file can be created and written to BEFORE we run the `yt-dlp` command
+    :ok = File.mkdir_p!(Path.dirname(filepath))
+    :ok = File.write(filepath, "")
+
+    filepath
+  end
+
   # We want to satisfy the following behaviours:
   #
   # 1. If the key is an atom, convert it to a string and convert it to kebab case (for convenience)
@@ -55,7 +81,7 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.CommandRunner do
   end
 
   defp parse_option(arg, acc) when is_binary(arg) do
-    [arg | acc]
+    acc ++ [arg]
   end
 
   defp backend_executable do

lib/pinchflat/media_client/backends/yt_dlp/metadata_parser.ex

@@ -0,0 +1,27 @@
+defmodule Pinchflat.MediaClient.Backends.YtDlp.MetadataParser do
+  @moduledoc """
+  yt-dlp offers a LOT of metadata in its JSON response, some of which
+  needs to be extracted and included in various models.
+
+  For now, also squirrel all of it away in the `media_metadata` table.
+  I might revisit this or pare it down later, but I'd rather need it
+  and not have it, ya know?
+  """
+
+  @doc """
+  Parses the given JSON response from yt-dlp and returns a map of
+  the needful media_item attributes, along with anything needed for
+  its associations.
+
+  Returns map()
+  """
+  def parse_for_media_item(metadata) do
+    %{
+      title: metadata["title"],
+      video_filepath: metadata["filepath"],
+      metadata: %{
+        client_response: metadata
+      }
+    }
+  end
+end

lib/pinchflat/media_client/backends/yt_dlp/video.ex

@@ -4,15 +4,15 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.Video do
   """
 
   @doc """
-  Downloads a single video (and possible metadata) directly to its
+  Downloads a single video (and possibly its metadata) directly to its
   final destination. Returns the parsed JSON output from yt-dlp.
 
   Returns {:ok, map()} | {:error, any, ...}.
   """
   def download(url, command_opts \\ []) do
-    opts = [:no_simulate, print: "%()j"] ++ command_opts
+    opts = [:no_simulate] ++ command_opts
 
-    with {:ok, output} <- backend_runner().run(url, opts),
+    with {:ok, output} <- backend_runner().run(url, opts, "after_move:%()j"),
          {:ok, parsed_json} <- Phoenix.json_library().decode(output) do
       {:ok, parsed_json}
     else

lib/pinchflat/media_client/backends/yt_dlp/video_collection.ex

@@ -16,9 +16,9 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.VideoCollection do
       """
       def get_video_ids(url, command_opts \\ []) do
         runner = Application.get_env(:pinchflat, :yt_dlp_runner)
-        opts = command_opts ++ [:simulate, :skip_download, print: :id]
+        opts = command_opts ++ [:simulate, :skip_download]
 
-        case runner.run(url, opts) do
+        case runner.run(url, opts, "%(id)s") do
           {:ok, output} -> {:ok, String.split(output, "\n", trim: true)}
           res -> res
         end

lib/pinchflat/media_client/video_downloader.ex

@@ -8,17 +8,41 @@ defmodule Pinchflat.MediaClient.VideoDownloader do
   it open-ish for future expansion (just in case).
   """
 
+  alias Pinchflat.Repo
+  alias Pinchflat.Media
+  alias Pinchflat.Media.MediaItem
   alias Pinchflat.Profiles.MediaProfile
 
   alias Pinchflat.MediaClient.Backends.YtDlp.Video, as: YtDlpVideo
   alias Pinchflat.Profiles.Options.YtDlp.OptionBuilder, as: YtDlpOptionBuilder
+  alias Pinchflat.MediaClient.Backends.YtDlp.MetadataParser, as: YtDlpMetadataParser
 
   @doc """
-  Downloads a single video based on the settings in the given media profile.
+  Downloads a video for a media item, updating the media item based on the metadata
+  returned by the backend. Also saves the entire metadata response to the associated
+  media_metadata record.
 
-  Returns {:ok, %ChannelDetails{}} | {:error, any, ...}.
+  Returns {:ok, %MediaItem{}} | {:error, any, ...any}
   """
-  def download_for_media_profile(url, %MediaProfile{} = media_profile, backend \\ :yt_dlp) do
+  def download_for_media_item(%MediaItem{} = media_item, backend \\ :yt_dlp) do
+    item_with_preloads = Repo.preload(media_item, [:metadata, channel: :media_profile])
+    media_profile = item_with_preloads.channel.media_profile
+
+    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)
+
+        # Don't forgor to use preloaded associations or updates to
+        # associations won't work!
+        Media.update_media_item(item_with_preloads, parsed_attrs)
+
+      err ->
+        err
+    end
+  end
+
+  defp download_for_media_profile(url, %MediaProfile{} = media_profile, backend) do
     option_builder = option_builder(backend)
     video_backend = video_backend(backend)
     {:ok, options} = option_builder.build(media_profile)
@@ -37,4 +61,10 @@ defmodule Pinchflat.MediaClient.VideoDownloader do
       :yt_dlp -> YtDlpVideo
     end
   end
+
+  defp metadata_parser(backend) do
+    case backend do
+      :yt_dlp -> YtDlpMetadataParser
+    end
+  end
 end

lib/pinchflat/media_source.ex

@@ -33,7 +33,7 @@ defmodule Pinchflat.MediaSource do
 
   Returns {:ok, %Channel{}} | {:error, %Ecto.Changeset{}}
   """
-  def create_channel(attrs \\ %{}) do
+  def create_channel(attrs) do
     %Channel{}
     |> change_channel_from_url(attrs)
     |> commit_and_start_indexing()
@@ -101,7 +101,7 @@ defmodule Pinchflat.MediaSource do
   This means that it'll go for it even if a changeset is otherwise invalid. This
   is pretty easy to change, but for MVP I'm not concerned.
   """
-  def change_channel_from_url(%Channel{} = channel, attrs \\ %{}) do
+  def change_channel_from_url(%Channel{} = channel, attrs) do
     case change_channel(channel, attrs) do
       %Ecto.Changeset{changes: %{original_url: _}} = changeset ->
         add_channel_details_to_changeset(channel, changeset)
@@ -148,11 +148,11 @@ defmodule Pinchflat.MediaSource do
 
   defp maybe_run_indexing_task(changeset, channel) do
     case changeset.data do
-      # If the changeset is new (not persisted), start indexing no matter what
+      # If the changeset is new (not persisted), attempt indexing no matter what
       %{__meta__: %{state: :built}} ->
         ChannelTasks.kickoff_indexing_task(channel)
 
-      # If the record has been persisted, only run indexing if the
+      # If the record has been persisted, only attempt indexing if the
       # indexing frequency has been changed
       %{__meta__: %{state: :loaded}} ->
         if Map.has_key?(changeset.changes, :index_frequency_minutes) do

lib/pinchflat/profiles.ex

@@ -25,7 +25,7 @@ defmodule Pinchflat.Profiles do
   @doc """
   Creates a media_profile. Returns {:ok, %MediaProfile{}} | {:error, %Ecto.Changeset{}}
   """
-  def create_media_profile(attrs \\ %{}) do
+  def create_media_profile(attrs) do
     %MediaProfile{}
     |> MediaProfile.changeset(attrs)
     |> Repo.insert()

lib/pinchflat/profiles/options/yt_dlp/option_builder.ex

@@ -24,15 +24,10 @@ defmodule Pinchflat.Profiles.Options.YtDlp.OptionBuilder do
     # see: https://github.com/yt-dlp/yt-dlp#output-template
     {:ok,
      [
-       :write_thumbnail,
-       :write_subs,
        :embed_metadata,
        :embed_thumbnail,
        :embed_subs,
-       :write_info_json,
-       :write_auto_subs,
        :no_progress,
-       convert_thumbnails: "jpg",
        sub_langs: "en.*",
        output: Path.join(base_directory(), output_path)
      ]}

lib/pinchflat/repo.ex

@@ -2,4 +2,18 @@ defmodule Pinchflat.Repo do
   use Ecto.Repo,
     otp_app: :pinchflat,
     adapter: Ecto.Adapters.Postgres
+
+  @doc """
+  It's not immediately obvious if an Oban job qualifies as unique, so this method
+  attempts creating a job and checks for the `conflict?` field in the returned job.
+
+  Returns {:ok, %Oban.Job{}} | {:duplicate, %Oban.Job{}} | {:error, any()}.
+  """
+  def insert_unique_job(job_struct) do
+    case Oban.insert(job_struct) do
+      {:ok, %Oban.Job{conflict?: false} = job} -> {:ok, job}
+      {:ok, %Oban.Job{conflict?: true} = job} -> {:duplicate, job}
+      err -> err
+    end
+  end
 end

lib/pinchflat/tasks.ex

@@ -7,6 +7,7 @@ defmodule Pinchflat.Tasks do
   alias Pinchflat.Repo
 
   alias Pinchflat.Tasks.Task
+  alias Pinchflat.Media.MediaItem
   alias Pinchflat.MediaSource.Channel
 
   @doc """
@@ -54,31 +55,41 @@ defmodule Pinchflat.Tasks do
   def get_task!(id), do: Repo.get!(Task, id)
 
   @doc """
-  Creates a task. Returns {:ok, %Task{}} | {:error, %Ecto.Changeset{}}.
+  Creates a task.
+
+  Accepts map() | %Oban.Job{}, %Channel{} | %Oban.Job{}, %MediaItem{}.
+  Returns {:ok, %Task{}} | {:error, %Ecto.Changeset{}}.
   """
-  def create_task(attrs \\ %{}) do
+  def create_task(attrs) do
     %Task{}
     |> Task.changeset(attrs)
     |> Repo.insert()
   end
 
-  # This one's function signature is designed to help simplify
+  # This function's signature is designed to help simplify
   # usage of `create_job_with_task/2`
-  def create_task(%Oban.Job{} = job, %Channel{} = channel) do
+  def create_task(%Oban.Job{} = job, attached_record) do
+    attached_record_attr =
+      case attached_record do
+        %Channel{} = channel -> %{channel_id: channel.id}
+        %MediaItem{} = media_item -> %{media_item_id: media_item.id}
+      end
+
     %Task{}
-    |> Task.changeset(%{job_id: job.id, channel_id: channel.id})
+    |> Task.changeset(Map.merge(%{job_id: job.id}, attached_record_attr))
     |> Repo.insert()
   end
 
   @doc """
   Creates a job from given attrs, creating a task with an attached record
-  if successful.
+  if successful. Returns an error if the job already exists.
 
-  Returns {:ok, %Task{}} | {:error, %Ecto.Changeset{}}.
+  Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}}.
   """
   def create_job_with_task(job_attrs, task_attached_record) do
-    case Oban.insert(job_attrs) do
+    case Repo.insert_unique_job(job_attrs) do
       {:ok, job} -> create_task(job, task_attached_record)
+      {:duplicate, _} -> {:error, :duplicate_job}
       err -> err
     end
   end
@@ -99,8 +110,12 @@ defmodule Pinchflat.Tasks do
 
   Returns :ok
   """
-  def delete_tasks_for(%Channel{} = channel) do
-    tasks = list_tasks_for(:channel_id, channel.id)
+  def delete_tasks_for(attached_record) do
+    tasks =
+      case attached_record do
+        %Channel{} = channel -> list_tasks_for(:channel_id, channel.id)
+        %MediaItem{} = media_item -> list_tasks_for(:media_item_id, media_item.id)
+      end
 
     Enum.each(tasks, fn task ->
       delete_task(task)
@@ -112,8 +127,12 @@ defmodule Pinchflat.Tasks do
 
   Returns :ok
   """
-  def delete_pending_tasks_for(%Channel{} = channel) do
-    tasks = list_pending_tasks_for(:channel_id, channel.id)
+  def delete_pending_tasks_for(attached_record) do
+    tasks =
+      case attached_record do
+        %Channel{} = channel -> list_pending_tasks_for(:channel_id, channel.id)
+        %MediaItem{} = media_item -> list_pending_tasks_for(:media_item_id, media_item.id)
+      end
 
     Enum.each(tasks, fn task ->
       delete_task(task)

lib/pinchflat/tasks/channel_tasks.ex

@@ -8,7 +8,9 @@ defmodule Pinchflat.Tasks.ChannelTasks do
   alias Pinchflat.Workers.MediaIndexingWorker
 
   @doc """
-  Starts tasks for indexing a channel's media. Returns {:ok, :should_not_index} | {:ok, %Task{}}.
+  Starts tasks for indexing a channel's media.
+
+  Returns {:ok, :should_not_index} | {:ok, %Task{}}.
   """
   def kickoff_indexing_task(%Channel{} = channel) do
     Tasks.delete_pending_tasks_for(channel)
@@ -21,6 +23,11 @@ defmodule Pinchflat.Tasks.ChannelTasks do
       # Schedule this one immediately, but future ones will be on an interval
       |> MediaIndexingWorker.new()
       |> Tasks.create_job_with_task(channel)
+      |> case do
+        # This should never return {:error, :duplicate_job} since we just deleted
+        # any pending tasks. I'm being assertive about it so it's obvious if I'm wrong
+        {:ok, task} -> {:ok, task}
+      end
     end
   end
 end

lib/pinchflat/tasks/task.ex

@@ -6,11 +6,13 @@ defmodule Pinchflat.Tasks.Task do
   use Ecto.Schema
   import Ecto.Changeset
 
+  alias Pinchflat.Media.MediaItem
   alias Pinchflat.MediaSource.Channel
 
   schema "tasks" do
     belongs_to :job, Oban.Job
     belongs_to :channel, Channel
+    belongs_to :media_item, MediaItem
 
     timestamps(type: :utc_datetime)
   end
@@ -18,7 +20,7 @@ defmodule Pinchflat.Tasks.Task do
   @doc false
   def changeset(task, attrs) do
     task
-    |> cast(attrs, [:job_id, :channel_id])
+    |> cast(attrs, [:job_id, :channel_id, :media_item_id])
     |> validate_required([:job_id])
   end
 end

lib/pinchflat/utils/string_utils.ex

@@ -5,10 +5,23 @@ defmodule Pinchflat.Utils.StringUtils do
 
   @doc """
   Converts a string to kebab-case (ie: `hello world` -> `hello-world`)
+
+  Returns binary()
   """
   def to_kebab_case(string) do
     string
     |> String.replace(~r/[\s_]/, "-")
     |> String.downcase()
   end
+
+  @doc """
+  Returns a random string of the given length. Base 16 encoded, lower case.
+
+  Returns binary()
+  """
+  def random_string(length \\ 32) do
+    :crypto.strong_rand_bytes(length)
+    |> Base.encode16(case: :lower)
+    |> String.slice(0..(length - 1))
+  end
 end

lib/pinchflat/workers/media_indexing_worker.ex

@@ -7,14 +7,19 @@ defmodule Pinchflat.Workers.MediaIndexingWorker do
     tags: ["media_source", "media_indexing"]
 
   alias __MODULE__
+  alias Pinchflat.Media
   alias Pinchflat.Tasks
   alias Pinchflat.MediaSource
+  alias Pinchflat.Workers.VideoDownloadWorker
 
   @impl Oban.Worker
   @doc """
-  The ID is that of a channel _record_, not a YouTube channel ID.
+  The ID is that of a channel _record_, not a YouTube channel ID. Indexes
+  the provided channel, kicks off downloads for each new MediaItem, and
+  reschedules the job to run again in the future (as determined by the
+  channel's `index_frequency_minutes` field).
 
-  NOTE: Re-scheduling here works a little different than you may expect.
+  README: Re-scheduling here works a little different than you may expect.
   The reschedule time is relative to the time the job has actually _completed_.
   This has some benefits but also side effects to be aware of:
 
@@ -25,7 +30,12 @@ defmodule Pinchflat.Workers.MediaIndexingWorker 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.
 
-  Returns :ok | {:ok, %Task{}}. Not that it matters.
+  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?
+
+  IDEA: Should I use paging and do indexing in chunks? Is that even faster?
+
+  Returns :ok | {:ok, %Task{}}
   """
   def perform(%Oban.Job{args: %{"id" => channel_id}}) do
     channel = MediaSource.get_channel!(channel_id)
@@ -39,10 +49,32 @@ defmodule Pinchflat.Workers.MediaIndexingWorker do
 
   defp index_media_and_reschedule(channel) do
     MediaSource.index_media_items(channel)
+    enqueue_video_downloads(channel)
 
     channel
     |> Map.take([:id])
     |> MediaIndexingWorker.new(schedule_in: channel.index_frequency_minutes * 60)
     |> Tasks.create_job_with_task(channel)
+    |> case do
+      {:ok, task} -> {:ok, task}
+      {:error, :duplicate_job} -> {:ok, :job_exists}
+    end
+  end
+
+  # NOTE: this starts a download for each media item that is pending,
+  # not just the ones that were indexed in this job run. This should ensure
+  # that any stragglers are caught if, for some reason, they weren't enqueued
+  # or somehow got de-queued.
+  #
+  # I'm not sure of a case where this would happen, but it's cheap insurance.
+  defp enqueue_video_downloads(channel) do
+    channel
+    |> Media.list_pending_media_items_for()
+    |> Enum.each(fn media_item ->
+      media_item
+      |> Map.take([:id])
+      |> VideoDownloadWorker.new()
+      |> Tasks.create_job_with_task(media_item)
+    end)
   end
 end

lib/pinchflat/workers/video_download_worker.ex

@@ -0,0 +1,26 @@
+defmodule Pinchflat.Workers.VideoDownloadWorker do
+  @moduledoc false
+
+  use Oban.Worker,
+    queue: :media_fetching,
+    unique: [period: :infinity, states: [:available, :scheduled, :retryable, :executing]],
+    tags: ["media_item", "media_fetching"]
+
+  alias Pinchflat.Media
+  alias Pinchflat.MediaClient.VideoDownloader
+
+  @impl Oban.Worker
+  @doc """
+  For a given media item, download the video and save the metadata.
+
+  Returns {:ok, %MediaItem{}} | {:error, any, ...any}
+  """
+  def perform(%Oban.Job{args: %{"id" => media_item_id}}) do
+    media_item = Media.get_media_item!(media_item_id)
+
+    case VideoDownloader.download_for_media_item(media_item) do
+      {:ok, _} -> {:ok, media_item}
+      err -> err
+    end
+  end
+end

priv/repo/migrations/20240129015810_create_media_metadata.exs

@@ -0,0 +1,15 @@
+defmodule Pinchflat.Repo.Migrations.CreateMediaMetadata do
+  use Ecto.Migration
+
+  def change do
+    create table(:media_metadata) do
+      add :client_response, :jsonb, null: false
+      add :media_item_id, references(:media_items, on_delete: :delete_all), null: false
+
+      timestamps(type: :utc_datetime)
+    end
+
+    create unique_index(:media_metadata, [:media_item_id])
+    create index(:media_metadata, [:client_response], using: :gin)
+  end
+end

priv/repo/migrations/20240130161316_add_media_item_to_tasks.exs

@@ -0,0 +1,12 @@
+defmodule Pinchflat.Repo.Migrations.AddMediaItemToTasks do
+  use Ecto.Migration
+
+  def change do
+    alter table(:tasks) do
+      # `restrict` because we need to be sure to delete pending tasks when a channel is deleted
+      add :media_item_id, references(:media_items, on_delete: :restrict), null: true
+    end
+
+    create index(:tasks, [:media_item_id])
+  end
+end

test/pinchflat/media_client/backends/yt_dlp/channel_test.exs

@@ -11,7 +11,7 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.ChannelTest do
 
   describe "get_channel_details/1" do
     test "it returns a %ChannelDetails{} with data on success" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts ->
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
         {:ok, "{\"channel\": \"TheUselessTrials\", \"channel_id\": \"UCQH2\"}"}
       end)
 
@@ -20,8 +20,9 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.ChannelTest do
     end
 
     test "it passes the expected args to the backend runner" do
-      expect(YtDlpRunnerMock, :run, fn @channel_url, opts ->
-        assert opts == [{:print, "%(.{channel,channel_id})j"}, {:playlist_end, 1}]
+      expect(YtDlpRunnerMock, :run, fn @channel_url, opts, ot ->
+        assert opts == [playlist_end: 1]
+        assert ot == "%(.{channel,channel_id})j"
 
         {:ok, "{}"}
       end)
@@ -30,13 +31,13 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.ChannelTest do
     end
 
     test "it returns an error if the runner returns an error" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts -> {:error, "Big issue", 1} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:error, "Big issue", 1} end)
 
       assert {:error, "Big issue", 1} = Channel.get_channel_details(@channel_url)
     end
 
     test "it returns an error if the output is not JSON" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts -> {:ok, "Not JSON"} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, "Not JSON"} end)
 
       assert {:error, %Jason.DecodeError{}} = Channel.get_channel_details(@channel_url)
     end

test/pinchflat/media_client/backends/yt_dlp/command_runner_test.exs

@@ -12,50 +12,49 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.CommandRunnerTest do
 
   describe "run/2" do
     test "it returns the output and status when the command succeeds" do
-      assert {:ok, _output} = Runner.run(@video_url, [])
+      assert {:ok, _output} = Runner.run(@video_url, [], "")
     end
 
     test "it converts symbol k-v arg keys to kebab case" do
-      assert {:ok, output} = Runner.run(@video_url, buffer_size: 1024)
+      assert {:ok, output} = Runner.run(@video_url, [buffer_size: 1024], "")
 
       assert String.contains?(output, "--buffer-size 1024")
     end
 
     test "it keeps string k-v arg keys untouched" do
-      assert {:ok, output} = Runner.run(@video_url, [{"--under_score", 1024}])
+      assert {:ok, output} = Runner.run(@video_url, [{"--under_score", 1024}], "")
 
       assert String.contains?(output, "--under_score 1024")
     end
 
     test "it converts symbol arg keys to kebab case" do
-      assert {:ok, output} = Runner.run(@video_url, [:ignore_errors])
+      assert {:ok, output} = Runner.run(@video_url, [:ignore_errors], "")
 
       assert String.contains?(output, "--ignore-errors")
     end
 
     test "it keeps string arg keys untouched" do
-      assert {:ok, output} = Runner.run(@video_url, ["-v"])
+      assert {:ok, output} = Runner.run(@video_url, ["-v"], "")
 
       assert String.contains?(output, "-v")
       refute String.contains?(output, "--v")
     end
 
-    test "it places arg keys (flags) at the beginning of the command" do
-      assert {:ok, output} =
-               Runner.run(@video_url, [{"--under_score", 1024}, :ignore_errors])
+    test "it includes the video url as the first argument" do
+      assert {:ok, output} = Runner.run(@video_url, [:ignore_errors], "")
 
-      assert String.contains?(output, "--ignore-errors --under_score 1024")
+      assert String.contains?(output, "#{@video_url} --ignore-errors")
     end
 
-    test "it includes the video url as the last argument" do
-      assert {:ok, output} = Runner.run(@video_url, [:ignore_errors])
+    test "it automatically includes the --print-to-file flag" do
+      assert {:ok, output} = Runner.run(@video_url, [], "%(id)s")
 
-      assert String.contains?(output, "--ignore-errors #{@video_url}\n")
+      assert String.contains?(output, "--print-to-file %(id)s /tmp/")
     end
 
     test "it returns the output and status when the command fails" do
       wrap_executable("/bin/false", fn ->
-        assert {:error, "", 1} = Runner.run(@video_url, [])
+        assert {:error, "", 1} = Runner.run(@video_url, [], "")
       end)
     end
   end

test/pinchflat/media_client/backends/yt_dlp/metadata_parser_test.exs

@@ -0,0 +1,45 @@
+defmodule Pinchflat.MediaClient.Backends.YtDlp.MediaParserTest do
+  use ExUnit.Case, async: true
+
+  alias Pinchflat.MediaClient.Backends.YtDlp.MetadataParser, as: Parser
+
+  setup do
+    json_filepath =
+      Path.join([
+        File.cwd!(),
+        "test",
+        "support",
+        "files",
+        "media_metadata.json"
+      ])
+
+    {:ok, file_body} = File.read(json_filepath)
+    {:ok, parsed_json} = Phoenix.json_library().decode(file_body)
+
+    {:ok,
+     %{
+       metadata: parsed_json
+     }}
+  end
+
+  describe "parse_for_media_item/1" do
+    test "it extracts the video filepath", %{metadata: metadata} do
+      result = Parser.parse_for_media_item(metadata)
+
+      assert String.contains?(result.video_filepath, "bwRHIkYqYJo")
+      assert String.ends_with?(result.video_filepath, ".mkv")
+    end
+
+    test "it extracts the title", %{metadata: metadata} do
+      result = Parser.parse_for_media_item(metadata)
+
+      assert result.title == "Trying to Wheelie Without the Rear Brake"
+    end
+
+    test "it returns the metadata as a map", %{metadata: metadata} do
+      result = Parser.parse_for_media_item(metadata)
+
+      assert result.metadata.client_response == metadata
+    end
+  end
+end

test/pinchflat/media_client/backends/yt_dlp/video_collection_test.exs

@@ -14,14 +14,15 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.VideoCollectionTest do
 
   describe "get_video_ids/2" do
     test "returns a list of video ids with no blank elements" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts -> {:ok, "id1\nid2\n\nid3\n"} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, "id1\nid2\n\nid3\n"} end)
 
       assert {:ok, ["id1", "id2", "id3"]} = VideoCollectionUser.get_video_ids(@channel_url)
     end
 
     test "it passes the expected default args" do
-      expect(YtDlpRunnerMock, :run, fn _url, opts ->
-        assert opts == [:simulate, :skip_download, {:print, :id}]
+      expect(YtDlpRunnerMock, :run, fn _url, opts, ot ->
+        assert opts == [:simulate, :skip_download]
+        assert ot == "%(id)s"
 
         {:ok, ""}
       end)
@@ -30,8 +31,8 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.VideoCollectionTest do
     end
 
     test "it passes the expected custom args" do
-      expect(YtDlpRunnerMock, :run, fn _url, opts ->
-        assert opts == [:custom_arg, :simulate, :skip_download, {:print, :id}]
+      expect(YtDlpRunnerMock, :run, fn _url, opts, _ot ->
+        assert opts == [:custom_arg, :simulate, :skip_download]
 
         {:ok, ""}
       end)
@@ -40,7 +41,7 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.VideoCollectionTest do
     end
 
     test "returns the error straight through when the command fails" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts -> {:error, "Big issue", 1} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:error, "Big issue", 1} end)
 
       assert {:error, "Big issue", 1} = VideoCollectionUser.get_video_ids(@channel_url)
     end

test/pinchflat/media_client/backends/yt_dlp/video_test.exs

@@ -1,5 +1,5 @@
 defmodule Pinchflat.MediaClient.Backends.YtDlp.VideoTest do
-  use ExUnit.Case, async: true
+  use Pinchflat.DataCase
   import Mox
 
   alias Pinchflat.MediaClient.Backends.YtDlp.Video
@@ -8,20 +8,27 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.VideoTest do
 
   setup :verify_on_exit!
 
+  # expect(YtDlpRunnerMock, :run, fn _url, [_, _, json_output_path | _] ->
+  # copy_metadata(json_output_path)
+
+  #   {:ok, ""}
+  # end)
+
   describe "download/2" do
     test "it calls the backend runner with the expected arguments" do
-      expect(YtDlpRunnerMock, :run, fn @video_url, opts ->
-        assert opts == [:no_simulate, {:print, "%()j"}]
+      expect(YtDlpRunnerMock, :run, fn @video_url, opts, ot ->
+        assert [:no_simulate] = opts
+        assert "after_move:%()j" = ot
 
-        {:ok, "{}"}
+        {:ok, render_metadata(:media_metadata)}
       end)
 
       assert {:ok, _} = Video.download(@video_url)
     end
 
     test "it passes along additional options" do
-      expect(YtDlpRunnerMock, :run, fn _url, opts ->
-        assert opts == [:no_simulate, {:print, "%()j"}, :custom_arg]
+      expect(YtDlpRunnerMock, :run, fn _url, opts, _ot ->
+        assert [:no_simulate, :custom_arg] = opts
 
         {:ok, "{}"}
       end)
@@ -29,22 +36,21 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.VideoTest do
       assert {:ok, _} = Video.download(@video_url, [:custom_arg])
     end
 
-    test "it parses the output as JSON" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts -> {:ok, "{\"title\": \"Test\"}"} end)
+    test "it parses and returns the generated file as JSON" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
+        {:ok, render_metadata(:media_metadata)}
+      end)
 
-      assert {:ok, %{"title" => "Test"}} = Video.download(@video_url)
+      assert {:ok, %{"title" => "Trying to Wheelie Without the Rear Brake"}} =
+               Video.download(@video_url)
     end
 
-    test "it returns an error if the output is not JSON" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts -> {:ok, "Not JSON"} end)
+    test "it returns errors" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opt, _ot ->
+        {:error, "something"}
+      end)
 
-      assert {:error, %Jason.DecodeError{}} = Video.download(@video_url)
-    end
-
-    test "it directly passes along any errors" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts -> {:error, "Big issue", 1} end)
-
-      assert {:error, "Big issue", 1} = Video.download(@video_url)
+      assert {:error, "something"} = Video.download(@video_url)
     end
   end
 end

test/pinchflat/media_client/channel_details_test.exs

@@ -17,8 +17,9 @@ defmodule Pinchflat.MediaClient.ChannelDetailsTest do
 
   describe "get_channel_details/2" do
     test "it passes the expected arguments to the backend" do
-      expect(YtDlpRunnerMock, :run, fn @channel_url, opts ->
-        assert opts == [{:print, "%(.{channel,channel_id})j"}, {:playlist_end, 1}]
+      expect(YtDlpRunnerMock, :run, fn @channel_url, opts, ot ->
+        assert opts == [playlist_end: 1]
+        assert ot == "%(.{channel,channel_id})j"
 
         {:ok, "{\"channel\": \"TheUselessTrials\", \"channel_id\": \"UCQH2\"}"}
       end)
@@ -27,7 +28,7 @@ defmodule Pinchflat.MediaClient.ChannelDetailsTest do
     end
 
     test "it returns a struct composed of the returned data" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts ->
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
         {:ok, "{\"channel\": \"TheUselessTrials\", \"channel_id\": \"UCQH2\"}"}
       end)
 
@@ -38,8 +39,9 @@ defmodule Pinchflat.MediaClient.ChannelDetailsTest do
 
   describe "get_video_ids/2" do
     test "it passes the expected arguments to the backend" do
-      expect(YtDlpRunnerMock, :run, fn @channel_url, opts ->
-        assert opts == [:simulate, :skip_download, {:print, :id}]
+      expect(YtDlpRunnerMock, :run, fn @channel_url, opts, ot ->
+        assert opts == [:simulate, :skip_download]
+        assert ot == "%(id)s"
 
         {:ok, ""}
       end)
@@ -48,7 +50,7 @@ defmodule Pinchflat.MediaClient.ChannelDetailsTest do
     end
 
     test "it returns a list of strings" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts ->
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
         {:ok, "video1\nvideo2\nvideo3"}
       end)
 

test/pinchflat/media_client/video_downloader_test.exs

@@ -1,35 +1,61 @@
 defmodule Pinchflat.MediaClient.VideoDownloaderTest do
-  use ExUnit.Case, async: true
+  use Pinchflat.DataCase
   import Mox
+  import Pinchflat.MediaFixtures
 
-  alias Pinchflat.Profiles.MediaProfile
   alias Pinchflat.MediaClient.VideoDownloader
 
-  @video_url "https://www.youtube.com/watch?v=TiZPUDkDYbk"
-  @media_profile %MediaProfile{
-    output_path_template: "videos/{{ title }}.%(ext)s"
-  }
-
   setup :verify_on_exit!
 
-  describe "download_for_media_profile/3" do
-    test "it calls the backend runner with the arguments built from the media profile" do
-      expect(YtDlpRunnerMock, :run, fn @video_url, opts ->
-        assert :no_simulate in opts
-        assert {:print, "%()j"} in opts
-        assert {:output, "/tmp/yt-dlp/videos/%(title)S.%(ext)s"} in opts
+  setup do
+    media_item =
+      Repo.preload(
+        media_item_fixture(%{title: nil, video_filepath: nil}),
+        [:metadata, channel: :media_profile]
+      )
 
-        {:ok, "{}"}
+    {:ok, %{media_item: media_item}}
+  end
+
+  describe "download_for_media_item/3" do
+    test "it calls the backend runner", %{media_item: media_item} do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, ot ->
+        assert ot == "after_move:%()j"
+
+        {:ok, render_metadata(:media_metadata)}
       end)
 
-      assert {:ok, _} = VideoDownloader.download_for_media_profile(@video_url, @media_profile)
+      assert {:ok, _} = VideoDownloader.download_for_media_item(media_item)
     end
 
-    test "it returns the parsed JSON output" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts -> {:ok, "{\"title\": \"Test\"}"} end)
+    test "it writes attributes to the media item", %{media_item: media_item} do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
+        {:ok, render_metadata(:media_metadata)}
+      end)
 
-      assert {:ok, %{"title" => "Test"}} =
-               VideoDownloader.download_for_media_profile(@video_url, @media_profile)
+      assert %{video_filepath: nil, title: nil} = media_item
+      assert {:ok, updated_media_item} = VideoDownloader.download_for_media_item(media_item)
+      assert updated_media_item.video_filepath
+      assert updated_media_item.title
+    end
+
+    test "it saves the metadata to the database", %{media_item: media_item} do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
+        {:ok, render_metadata(:media_metadata)}
+      end)
+
+      assert is_nil(media_item.metadata)
+      assert {:ok, updated_media_item} = VideoDownloader.download_for_media_item(media_item)
+      assert updated_media_item.metadata
+      assert is_map(updated_media_item.metadata.client_response)
+    end
+
+    test "errors are passed through", %{media_item: media_item} do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
+        {:error, :some_error}
+      end)
+
+      assert {:error, :some_error} = VideoDownloader.download_for_media_item(media_item)
     end
   end
 end

test/pinchflat/media_source_test.exs

@@ -30,7 +30,7 @@ defmodule Pinchflat.MediaSourceTest do
 
   describe "create_channel/1" do
     test "creates a channel and adds name + ID from runner response" do
-      expect(YtDlpRunnerMock, :run, &runner_function_mock/2)
+      expect(YtDlpRunnerMock, :run, &runner_function_mock/3)
 
       valid_attrs = %{
         media_profile_id: media_profile_fixture().id,
@@ -47,7 +47,7 @@ defmodule Pinchflat.MediaSourceTest do
     end
 
     test "creation enforces uniqueness of channel_id scoped to the media_profile" do
-      expect(YtDlpRunnerMock, :run, 2, fn _url, _opts ->
+      expect(YtDlpRunnerMock, :run, 2, fn _url, _opts, _ot ->
         {:ok,
          Phoenix.json_library().encode!(%{
            channel: "some name",
@@ -65,7 +65,7 @@ defmodule Pinchflat.MediaSourceTest do
     end
 
     test "creation lets you duplicate channel_ids as long as the media profile is different" do
-      expect(YtDlpRunnerMock, :run, 2, fn _url, _opts ->
+      expect(YtDlpRunnerMock, :run, 2, fn _url, _opts, _ot ->
         {:ok,
          Phoenix.json_library().encode!(%{
            channel: "some name",
@@ -86,7 +86,7 @@ defmodule Pinchflat.MediaSourceTest do
     end
 
     test "creation will schedule the indexing task" do
-      expect(YtDlpRunnerMock, :run, &runner_function_mock/2)
+      expect(YtDlpRunnerMock, :run, &runner_function_mock/3)
 
       valid_attrs = %{
         media_profile_id: media_profile_fixture().id,
@@ -101,7 +101,7 @@ defmodule Pinchflat.MediaSourceTest do
 
   describe "index_media_items/1" do
     setup do
-      stub(YtDlpRunnerMock, :run, fn _url, _opts -> {:ok, "video1\nvideo2\nvideo3"} end)
+      stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, "video1\nvideo2\nvideo3"} end)
 
       {:ok, [channel: channel_fixture()]}
     end
@@ -162,7 +162,7 @@ defmodule Pinchflat.MediaSourceTest do
     end
 
     test "updating the original_url will re-fetch the channel details" do
-      expect(YtDlpRunnerMock, :run, &runner_function_mock/2)
+      expect(YtDlpRunnerMock, :run, &runner_function_mock/3)
 
       channel = channel_fixture()
       update_attrs = %{original_url: "https://www.youtube.com/channel/abc123"}
@@ -173,7 +173,7 @@ defmodule Pinchflat.MediaSourceTest do
     end
 
     test "not updating the original_url will not re-fetch the channel details" do
-      expect(YtDlpRunnerMock, :run, 0, &runner_function_mock/2)
+      expect(YtDlpRunnerMock, :run, 0, &runner_function_mock/3)
 
       channel = channel_fixture()
       update_attrs = %{name: "some updated name"}
@@ -239,14 +239,14 @@ defmodule Pinchflat.MediaSourceTest do
 
   describe "change_channel_from_url/2" do
     test "it returns a changeset" do
-      stub(YtDlpRunnerMock, :run, &runner_function_mock/2)
+      stub(YtDlpRunnerMock, :run, &runner_function_mock/3)
       channel = channel_fixture()
 
-      assert %Ecto.Changeset{} = MediaSource.change_channel_from_url(channel)
+      assert %Ecto.Changeset{} = MediaSource.change_channel_from_url(channel, %{})
     end
 
     test "it does not fetch channel details if the original_url isn't in the changeset" do
-      expect(YtDlpRunnerMock, :run, 0, &runner_function_mock/2)
+      expect(YtDlpRunnerMock, :run, 0, &runner_function_mock/3)
 
       changeset = MediaSource.change_channel_from_url(%Channel{}, %{name: "some updated name"})
 
@@ -254,7 +254,7 @@ defmodule Pinchflat.MediaSourceTest do
     end
 
     test "it fetches channel details if the original_url is in the changeset" do
-      expect(YtDlpRunnerMock, :run, &runner_function_mock/2)
+      expect(YtDlpRunnerMock, :run, &runner_function_mock/3)
 
       changeset =
         MediaSource.change_channel_from_url(%Channel{}, %{
@@ -265,7 +265,7 @@ defmodule Pinchflat.MediaSourceTest do
     end
 
     test "it adds channel details to the changeset, keeping the orignal details" do
-      expect(YtDlpRunnerMock, :run, &runner_function_mock/2)
+      expect(YtDlpRunnerMock, :run, &runner_function_mock/3)
 
       media_profile = media_profile_fixture()
       media_profile_id = media_profile.id
@@ -287,7 +287,7 @@ defmodule Pinchflat.MediaSourceTest do
     end
 
     test "it adds an error to the changeset if the runner fails" do
-      expect(YtDlpRunnerMock, :run, 1, fn _url, _opts ->
+      expect(YtDlpRunnerMock, :run, 1, fn _url, _opts, _ot ->
         {:error, "some error", 1}
       end)
 
@@ -301,7 +301,7 @@ defmodule Pinchflat.MediaSourceTest do
     end
   end
 
-  defp runner_function_mock(_url, _opts) do
+  defp runner_function_mock(_url, _opts, _ot) do
     {
       :ok,
       Phoenix.json_library().encode!(%{

test/pinchflat/media_test.exs

@@ -1,6 +1,7 @@
 defmodule Pinchflat.MediaTest do
   use Pinchflat.DataCase
 
+  import Pinchflat.TasksFixtures
   import Pinchflat.MediaFixtures
   import Pinchflat.MediaSourceFixtures
 
@@ -9,6 +10,18 @@ defmodule Pinchflat.MediaTest do
 
   @invalid_attrs %{title: nil, media_id: nil, video_filepath: nil}
 
+  describe "schema" do
+    test "media_metadata is deleted when media_item is deleted" do
+      media_item = media_item_fixture(%{metadata: %{client_response: %{foo: "bar"}}})
+      metadata = media_item.metadata
+      assert {:ok, %MediaItem{}} = Media.delete_media_item(media_item)
+
+      assert_raise Ecto.NoResultsError, fn ->
+        Repo.reload!(metadata)
+      end
+    end
+  end
+
   describe "list_media_items/0" do
     test "it returns all media_items" do
       media_item = media_item_fixture()
@@ -16,6 +29,27 @@ defmodule Pinchflat.MediaTest do
     end
   end
 
+  describe "list_pending_media_items_for/1" do
+    test "it returns pending media_items for a given channel" do
+      channel = channel_fixture()
+      media_item = media_item_fixture(%{channel_id: channel.id, video_filepath: nil})
+
+      assert Media.list_pending_media_items_for(channel) == [media_item]
+    end
+
+    test "it does not return media_items with video_filepath" do
+      channel = channel_fixture()
+
+      _media_item =
+        media_item_fixture(%{
+          channel_id: channel.id,
+          video_filepath: "/video/#{Faker.File.file_name(:video)}"
+        })
+
+      assert Media.list_pending_media_items_for(channel) == []
+    end
+  end
+
   describe "get_media_item!/1" do
     test "it returns the media_item with given id" do
       media_item = media_item_fixture()
@@ -73,6 +107,14 @@ defmodule Pinchflat.MediaTest do
       assert {:ok, %MediaItem{}} = Media.delete_media_item(media_item)
       assert_raise Ecto.NoResultsError, fn -> Media.get_media_item!(media_item.id) end
     end
+
+    test "it also deletes attached tasks" do
+      media_item = media_item_fixture()
+      task = task_fixture(%{media_item_id: media_item.id})
+
+      assert {:ok, %MediaItem{}} = Media.delete_media_item(media_item)
+      assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end
+    end
   end
 
   describe "change_media_item/1" do

test/pinchflat/profiles/options/yt_dlp/option_builder_test.exs

@@ -5,14 +5,14 @@ defmodule Pinchflat.Profiles.Options.YtDlp.OptionBuilderTest do
   alias Pinchflat.Profiles.Options.YtDlp.OptionBuilder
 
   @media_profile %MediaProfile{
-    output_path_template: "videos/{{ title }}.%(ext)s"
+    output_path_template: "{{ title }}.%(ext)s"
   }
 
   describe "build/1" do
     test "it generates an expanded output path based on the given template" do
       assert {:ok, res} = OptionBuilder.build(@media_profile)
 
-      assert {:output, "/tmp/yt-dlp/videos/%(title)S.%(ext)s"} in res
+      assert {:output, "/tmp/videos/%(title)S.%(ext)s"} in res
     end
   end
 end

test/pinchflat/repo_test.exs

@@ -0,0 +1,26 @@
+defmodule Pinchflat.RepoTest do
+  use Pinchflat.DataCase
+
+  alias Pinchflat.JobFixtures.TestJobWorker
+
+  describe "insert_unique_job/1" do
+    test "returns {:ok, job} if there is no conflict" do
+      job = TestJobWorker.new(%{})
+
+      assert {:ok, %Oban.Job{}} = Pinchflat.Repo.insert_unique_job(job)
+    end
+
+    test "returns {:duplicate, original_job} if there is a conflict" do
+      job = TestJobWorker.new(%{foo: "bar"}, unique: [period: :infinity])
+
+      {:ok, saved_job_1} = Pinchflat.Repo.insert_unique_job(job)
+
+      assert {:duplicate, saved_job_2} = Pinchflat.Repo.insert_unique_job(job)
+      assert saved_job_1.id == saved_job_2.id
+    end
+
+    test "returns the error if there is an error" do
+      assert {:error, _} = Pinchflat.Repo.insert_unique_job(%Ecto.Changeset{})
+    end
+  end
+end

test/pinchflat/tasks_test.exs

@@ -2,6 +2,7 @@ defmodule Pinchflat.TasksTest do
   use Pinchflat.DataCase
   import Pinchflat.JobFixtures
   import Pinchflat.TasksFixtures
+  import Pinchflat.MediaFixtures
   import Pinchflat.MediaSourceFixtures
 
   alias Pinchflat.Tasks
@@ -92,14 +93,24 @@ defmodule Pinchflat.TasksTest do
       assert task.job_id == job.id
       assert task.channel_id == channel.id
     end
+
+    test "accepts a job and media item" do
+      job = job_fixture()
+      media_item = media_item_fixture()
+
+      assert {:ok, %Task{} = task} = Tasks.create_task(job, media_item)
+
+      assert task.job_id == job.id
+      assert task.media_item_id == media_item.id
+    end
   end
 
   describe "create_job_with_task/2" do
     test "it enqueues the given job" do
-      channel = channel_fixture()
+      media_item = media_item_fixture()
 
       refute_enqueued(worker: TestJobWorker)
-      assert {:ok, %Task{}} = Tasks.create_job_with_task(TestJobWorker.new(%{}), channel)
+      assert {:ok, %Task{}} = Tasks.create_job_with_task(TestJobWorker.new(%{}), media_item)
       assert_enqueued(worker: TestJobWorker)
     end
 
@@ -110,6 +121,20 @@ defmodule Pinchflat.TasksTest do
 
       assert task.channel_id == channel.id
     end
+
+    test "it returns an error if the job already exists" do
+      channel = channel_fixture()
+      job = TestJobWorker.new(%{foo: "bar"}, unique: [period: :infinity])
+
+      assert {:ok, %Task{}} = Tasks.create_job_with_task(job, channel)
+      assert {:error, :duplicate_job} = Tasks.create_job_with_task(job, channel)
+    end
+
+    test "it returns an error if the job fails to enqueue" do
+      channel = channel_fixture()
+
+      assert {:error, %Ecto.Changeset{}} = Tasks.create_job_with_task(%Ecto.Changeset{}, channel)
+    end
   end
 
   describe "delete_task/1" do
@@ -137,6 +162,14 @@ defmodule Pinchflat.TasksTest do
       assert :ok = Tasks.delete_tasks_for(channel)
       assert_raise Ecto.NoResultsError, fn -> Tasks.get_task!(task.id) end
     end
+
+    test "it deletes the tasks attached to a media_item" do
+      media_item = media_item_fixture()
+      task = task_fixture(media_item_id: media_item.id)
+
+      assert :ok = Tasks.delete_tasks_for(media_item)
+      assert_raise Ecto.NoResultsError, fn -> Tasks.get_task!(task.id) end
+    end
   end
 
   describe "delete_pending_tasks_for/1" do
@@ -156,6 +189,17 @@ defmodule Pinchflat.TasksTest do
       assert :ok = Tasks.delete_pending_tasks_for(channel)
       assert Tasks.get_task!(task.id)
     end
+
+    test "it works on media_items" do
+      media_item = media_item_fixture()
+      pending_task = task_fixture(media_item_id: media_item.id)
+      cancelled_task = Repo.preload(task_fixture(media_item_id: media_item.id), :job)
+      :ok = Oban.cancel_job(cancelled_task.job)
+
+      assert :ok = Tasks.delete_pending_tasks_for(media_item)
+      assert Tasks.get_task!(cancelled_task.id)
+      assert_raise Ecto.NoResultsError, fn -> Repo.reload!(pending_task) end
+    end
   end
 
   describe "change_task/1" do

test/pinchflat/utils/string_utils_test.exs

@@ -12,4 +12,19 @@ defmodule Pinchflat.Utils.StringUtilsTest do
       assert StringUtils.to_kebab_case("hello_world") == "hello-world"
     end
   end
+
+  describe "random_string/1" do
+    test "generates a random string" do
+      assert is_binary(StringUtils.random_string())
+      assert StringUtils.random_string() != StringUtils.random_string()
+    end
+
+    test "has a defined default length" do
+      assert String.length(StringUtils.random_string()) == 32
+    end
+
+    test "can generate a string of a given length" do
+      assert String.length(StringUtils.random_string(64)) == 64
+    end
+  end
 end

test/pinchflat/workers/media_indexing_worker_test.exs

@@ -2,16 +2,18 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
   use Pinchflat.DataCase
 
   import Mox
+  import Pinchflat.MediaFixtures
   import Pinchflat.MediaSourceFixtures
 
   alias Pinchflat.Tasks
   alias Pinchflat.Workers.MediaIndexingWorker
+  alias Pinchflat.Workers.VideoDownloadWorker
 
   setup :verify_on_exit!
 
   describe "perform/1" do
     test "it does not do any indexing if the channel shouldn't be indexed" do
-      expect(YtDlpRunnerMock, :run, 0, fn _url, _opts -> {:ok, ""} end)
+      expect(YtDlpRunnerMock, :run, 0, fn _url, _opts, _ot -> {:ok, ""} end)
 
       channel = channel_fixture(index_frequency_minutes: -1)
 
@@ -19,7 +21,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
     end
 
     test "it does not reschedule if the channel shouldn't be indexed" do
-      expect(YtDlpRunnerMock, :run, 0, fn _url, _opts -> {:ok, ""} end)
+      expect(YtDlpRunnerMock, :run, 0, fn _url, _opts, _ot -> {:ok, ""} end)
 
       channel = channel_fixture(index_frequency_minutes: -1)
       perform_job(MediaIndexingWorker, %{id: channel.id})
@@ -28,15 +30,44 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
     end
 
     test "it indexes the channel if it should be indexed" do
-      expect(YtDlpRunnerMock, :run, 1, fn _url, _opts -> {:ok, ""} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, ""} end)
 
       channel = channel_fixture(index_frequency_minutes: 10)
 
       perform_job(MediaIndexingWorker, %{id: channel.id})
     end
 
+    test "it kicks off a download job for each pending media item" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, "video1"} end)
+
+      channel = channel_fixture(index_frequency_minutes: 10)
+      perform_job(MediaIndexingWorker, %{id: channel.id})
+
+      assert [_] = all_enqueued(worker: VideoDownloadWorker)
+    end
+
+    test "it starts a job for any pending media item even if it's from another run" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, "video1"} end)
+
+      channel = channel_fixture(index_frequency_minutes: 10)
+      media_item_fixture(%{channel_id: channel.id, video_filepath: nil})
+      perform_job(MediaIndexingWorker, %{id: channel.id})
+
+      assert [_, _] = all_enqueued(worker: VideoDownloadWorker)
+    end
+
+    test "it does not kick off a job for media items that could not be saved" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, "video1\nvideo1"} end)
+
+      channel = channel_fixture(index_frequency_minutes: 10)
+      perform_job(MediaIndexingWorker, %{id: channel.id})
+
+      # Only one job should be enqueued, since the second video is a duplicate
+      assert [_] = all_enqueued(worker: VideoDownloadWorker)
+    end
+
     test "it reschedules the job based on the index frequency" do
-      expect(YtDlpRunnerMock, :run, 1, fn _url, _opts -> {:ok, ""} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, ""} end)
 
       channel = channel_fixture(index_frequency_minutes: 10)
       perform_job(MediaIndexingWorker, %{id: channel.id})
@@ -49,7 +80,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
     end
 
     test "it creates a task for the rescheduled job" do
-      expect(YtDlpRunnerMock, :run, 1, fn _url, _opts -> {:ok, ""} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, ""} end)
 
       channel = channel_fixture(index_frequency_minutes: 10)
       task_count_fetcher = fn -> Enum.count(Tasks.list_tasks()) end
@@ -60,7 +91,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
     end
 
     test "it creates the basic media_item records" do
-      expect(YtDlpRunnerMock, :run, 1, fn _url, _opts -> {:ok, "video1\nvideo2"} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, "video1\nvideo2"} end)
 
       channel = channel_fixture(index_frequency_minutes: 10)
 

test/pinchflat/workers/video_download_worker_test.exs

@@ -0,0 +1,59 @@
+defmodule Pinchflat.Workers.VideoDownloadWorkerTest do
+  use Pinchflat.DataCase
+
+  import Mox
+  import Pinchflat.MediaFixtures
+
+  alias Pinchflat.Workers.VideoDownloadWorker
+
+  setup :verify_on_exit!
+
+  setup do
+    media_item =
+      Repo.preload(
+        media_item_fixture(%{video_filepath: nil}),
+        [:metadata, channel: :media_profile]
+      )
+
+    {:ok, %{media_item: media_item}}
+  end
+
+  describe "perform/1" do
+    test "it saves attributes to the media_item", %{media_item: media_item} do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
+        {:ok, render_metadata(:media_metadata)}
+      end)
+
+      assert media_item.video_filepath == nil
+      perform_job(VideoDownloadWorker, %{id: media_item.id})
+      assert Repo.reload(media_item).video_filepath != nil
+    end
+
+    test "it saves the metadata to the media_item", %{media_item: media_item} do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
+        {:ok, render_metadata(:media_metadata)}
+      end)
+
+      assert media_item.metadata == nil
+      perform_job(VideoDownloadWorker, %{id: media_item.id})
+      assert Repo.reload(media_item).metadata != nil
+    end
+
+    test "it won't double-schedule downloading jobs", %{media_item: media_item} do
+      Oban.insert(VideoDownloadWorker.new(%{id: media_item.id}))
+      Oban.insert(VideoDownloadWorker.new(%{id: media_item.id}))
+
+      assert [_] = all_enqueued(worker: VideoDownloadWorker)
+    end
+
+    test "it sets the job to retryable if the download fails", %{media_item: media_item} do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:error, "error"} end)
+
+      Oban.Testing.with_testing_mode(:inline, fn ->
+        {:ok, job} = Oban.insert(VideoDownloadWorker.new(%{id: media_item.id}))
+
+        assert job.state == "retryable"
+      end)
+    end
+  end
+end

test/pinchflat_web/controllers/channel_controller_test.exs

@@ -41,7 +41,7 @@ defmodule PinchflatWeb.ChannelControllerTest do
 
   describe "create channel" do
     test "redirects to show when data is valid", %{conn: conn, create_attrs: create_attrs} do
-      expect(YtDlpRunnerMock, :run, 1, &runner_function_mock/2)
+      expect(YtDlpRunnerMock, :run, 1, &runner_function_mock/3)
       conn = post(conn, ~p"/media_sources/channels", channel: create_attrs)
 
       assert %{id: id} = redirected_params(conn)
@@ -70,7 +70,7 @@ defmodule PinchflatWeb.ChannelControllerTest do
     setup [:create_channel]
 
     test "redirects when data is valid", %{conn: conn, channel: channel, update_attrs: update_attrs} do
-      expect(YtDlpRunnerMock, :run, 1, &runner_function_mock/2)
+      expect(YtDlpRunnerMock, :run, 1, &runner_function_mock/3)
 
       conn = put(conn, ~p"/media_sources/channels/#{channel}", channel: update_attrs)
       assert redirected_to(conn) == ~p"/media_sources/channels/#{channel}"
@@ -107,7 +107,7 @@ defmodule PinchflatWeb.ChannelControllerTest do
     %{channel: channel}
   end
 
-  defp runner_function_mock(_url, _opts) do
+  defp runner_function_mock(_url, _opts, _ot) do
     {
       :ok,
       Phoenix.json_library().encode!(%{

test/support/fixtures/media_fixtures.ex

@@ -22,4 +22,24 @@ defmodule Pinchflat.MediaFixtures do
 
     media_item
   end
+
+  @doc """
+  Generate a media_item with metadata.
+  """
+  def media_item_with_metadata(attrs \\ %{}) do
+    json_filepath =
+      Path.join([
+        Path.dirname(__ENV__.file),
+        "support",
+        "fixtures",
+        "files",
+        "media_metadata.json"
+      ])
+
+    {:ok, file_body} = File.read(json_filepath)
+    {:ok, parsed_json} = Phoenix.json_library().decode(file_body)
+    merged_attrs = Map.merge(attrs, %{metadata: %{client_respinse: parsed_json}})
+
+    media_item_fixture(merged_attrs)
+  end
 end

test/support/fixtures/profiles_fixtures.ex

@@ -12,7 +12,7 @@ defmodule Pinchflat.ProfilesFixtures do
       attrs
       |> Enum.into(%{
         name: "Media Profile ##{:rand.uniform(1_000_000)}",
-        output_path_template: "/video/{{title}}.{{ext}}"
+        output_path_template: "{{title}}.{{ext}}"
       })
       |> Pinchflat.Profiles.create_media_profile()
 

test/support/scripts/yt-dlp-mocks/repeater.sh

@@ -1,7 +1,16 @@
 #!/bin/bash
 
-if [[ "$@" == *"--dump-json"* ]]; then
-  echo '{ "args": "'$@'"}'
-else
-  echo $@
-fi
+# Args come in the format of "<unknown number of args> --print-to-file <output template> <file location> <unknown number of args>".
+# I need to extract <file location> and write all args to it.
+
+# Extract the file location (in an unknown position BUT it's 2 args after --print-to-file).
+for ((i = 1; i <= $#; i++)); do
+  if [ "${!i}" == "--print-to-file" ]; then
+    # Extract the file location.
+    file_location="${@:i+2:1}"
+    break
+  fi
+done
+
+# Write all args to the file
+echo "$@" >"$file_location"

test/support/testing_helper_methods.ex

@@ -27,4 +27,17 @@ defmodule Pinchflat.TestingHelperMethods do
     assert before_res == from
     assert after_res == to
   end
+
+  def render_metadata(metadata_name) do
+    json_filepath =
+      Path.join([
+        File.cwd!(),
+        "test",
+        "support",
+        "files",
+        "#{metadata_name}.json"
+      ])
+
+    File.read!(json_filepath)
+  end
 end