Squash all the commits from the other branch bc I broke things (#88)

config/config.exs

@@ -48,7 +48,8 @@ config :pinchflat, Oban,
     media_indexing: 2,
     media_collection_indexing: 2,
     media_fetching: 2,
-    media_local_metadata: 8
+    local_metadata: 8,
+    remote_metadata: 4
   ]
 
 # Configures the mailer

lib/pinchflat/boot/data_backfill_worker.ex

@@ -2,7 +2,7 @@ defmodule Pinchflat.Boot.DataBackfillWorker do
   @moduledoc false
 
   use Oban.Worker,
-    queue: :media_local_metadata,
+    queue: :local_metadata,
     unique: [period: :infinity, states: [:available, :scheduled, :retryable]],
     tags: ["media_item", "media_metadata", "local_metadata", "data_backfill"]
 

lib/pinchflat/filesystem/filesystem_data_worker.ex

@@ -2,7 +2,7 @@ defmodule Pinchflat.Filesystem.FilesystemDataWorker do
   @moduledoc false
 
   use Oban.Worker,
-    queue: :media_local_metadata,
+    queue: :local_metadata,
     tags: ["media_item", "media_metadata", "local_metadata"],
     max_attempts: 1
 
@@ -13,6 +13,10 @@ defmodule Pinchflat.Filesystem.FilesystemDataWorker do
   @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

lib/pinchflat/media/media.ex

@@ -144,6 +144,8 @@ defmodule Pinchflat.Media do
   @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
@@ -162,6 +164,8 @@ defmodule Pinchflat.Media do
   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
@@ -227,6 +231,7 @@ defmodule Pinchflat.Media do
   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)
     end
@@ -242,6 +247,7 @@ 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)
 

lib/pinchflat/metadata/source_metadata.ex

@@ -0,0 +1,36 @@
+defmodule Pinchflat.Metadata.SourceMetadata do
+  @moduledoc """
+  The SourceMetadata schema.
+
+  Look. Don't @ me about Metadata vs. Metadatum. I'm very sensitive.
+  """
+
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  alias Pinchflat.Sources.Source
+
+  @allowed_fields ~w(metadata_filepath)a
+  @required_fields ~w(metadata_filepath)a
+
+  schema "source_metadata" do
+    field :metadata_filepath, :string
+
+    belongs_to :source, Source
+
+    timestamps(type: :utc_datetime)
+  end
+
+  @doc false
+  def changeset(source_metadata, attrs) do
+    source_metadata
+    |> cast(attrs, @allowed_fields)
+    |> validate_required(@required_fields)
+    |> unique_constraint([:source_id])
+  end
+
+  @doc false
+  def filepath_attributes do
+    ~w(metadata_filepath)a
+  end
+end

lib/pinchflat/metadata/source_metadata_storage_worker.ex

@@ -0,0 +1,48 @@
+defmodule Pinchflat.Metadata.SourceMetadataStorageWorker do
+  @moduledoc false
+
+  use Oban.Worker,
+    queue: :remote_metadata,
+    tags: ["media_source", "source_metadata", "remote_metadata"],
+    max_attempts: 1
+
+  alias __MODULE__
+  alias Pinchflat.Repo
+  alias Pinchflat.Tasks
+  alias Pinchflat.Sources
+  alias Pinchflat.YtDlp.MediaCollection
+  alias Pinchflat.Metadata.MetadataFileHelpers
+
+  @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
+    %{id: source.id}
+    |> SourceMetadataStorageWorker.new()
+    |> 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
+  """
+  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)
+
+    Sources.update_source(source, %{
+      metadata: %{
+        metadata_filepath: MetadataFileHelpers.compress_and_store_metadata_for(source, metadata)
+      }
+    })
+
+    :ok
+  end
+end

lib/pinchflat/sources/source.ex

@@ -10,6 +10,7 @@ defmodule Pinchflat.Sources.Source do
   alias Pinchflat.Tasks.Task
   alias Pinchflat.Media.MediaItem
   alias Pinchflat.Profiles.MediaProfile
+  alias Pinchflat.Metadata.SourceMetadata
 
   @allowed_fields ~w(
     collection_name
@@ -28,7 +29,8 @@ defmodule Pinchflat.Sources.Source do
   # Expensive API calls are made when a source is inserted/updated so
   # we want to ensure that the source is valid before making the call.
   # This way, we check that the other attributes are valid before ensuring
-  # that all fields are valid.
+  # that all fields are valid. This is still only one DB insert but it's
+  # a two-stage validation process to fail fast before the API call.
   @initially_required_fields ~w(
     index_frequency_minutes
     fast_index
@@ -60,6 +62,8 @@ defmodule Pinchflat.Sources.Source do
 
     belongs_to :media_profile, MediaProfile
 
+    has_one :metadata, SourceMetadata, on_replace: :update
+
     has_many :tasks, Task
     has_many :media_items, MediaItem, foreign_key: :source_id
 
@@ -80,6 +84,7 @@ defmodule Pinchflat.Sources.Source do
     |> cast(attrs, @allowed_fields)
     |> dynamic_default(:custom_name, fn cs -> get_field(cs, :collection_name) end)
     |> validate_required(required_fields)
+    |> cast_assoc(:metadata, with: &SourceMetadata.changeset/2, required: false)
     |> unique_constraint([:collection_id, :media_profile_id])
   end
 

lib/pinchflat/sources/sources.ex

@@ -11,9 +11,11 @@ defmodule Pinchflat.Sources do
   alias Pinchflat.Sources.Source
   alias Pinchflat.Profiles.MediaProfile
   alias Pinchflat.YtDlp.MediaCollection
+  alias Pinchflat.Metadata.SourceMetadata
   alias Pinchflat.Downloading.DownloadingHelpers
   alias Pinchflat.FastIndexing.FastIndexingHelpers
   alias Pinchflat.SlowIndexing.SlowIndexingHelpers
+  alias Pinchflat.Metadata.SourceMetadataStorageWorker
 
   @doc """
   Returns the list of sources. Returns [%Source{}, ...]
@@ -54,7 +56,7 @@ defmodule Pinchflat.Sources do
     case change_source(%Source{}, attrs, :initial) do
       %Ecto.Changeset{valid?: true} ->
         %Source{}
-        |> change_source_from_url(attrs)
+        |> maybe_change_source_from_url(attrs)
         |> maybe_change_indexing_frequency()
         |> commit_and_handle_tasks()
 
@@ -82,7 +84,7 @@ defmodule Pinchflat.Sources do
     case change_source(source, attrs, :initial) do
       %Ecto.Changeset{valid?: true} ->
         source
-        |> change_source_from_url(attrs)
+        |> maybe_change_source_from_url(attrs)
         |> maybe_change_indexing_frequency()
         |> commit_and_handle_tasks()
 
@@ -107,6 +109,7 @@ defmodule Pinchflat.Sources do
     end)
 
     Tasks.delete_tasks_for(source)
+    delete_source_metadata_files(source)
     Repo.delete(source)
   end
 
@@ -122,14 +125,12 @@ defmodule Pinchflat.Sources do
   fetches source details from the original_url (if provided). If the source
   details cannot be fetched, an error is added to the changeset.
 
-  Note that this fetches source details as long as the `original_url` is present.
-  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.
-
   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 change_source_from_url(%Source{} = source, attrs) do
+  def 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)
@@ -139,6 +140,18 @@ defmodule Pinchflat.Sources do
     end
   end
 
+  defp delete_source_metadata_files(source) do
+    metadata = Repo.preload(source, :metadata).metadata || %SourceMetadata{}
+    mapped_struct = Map.from_struct(metadata)
+
+    filepaths =
+      SourceMetadata.filepath_attributes()
+      |> Enum.map(fn field -> mapped_struct[field] end)
+      |> Enum.filter(&is_binary/1)
+
+    Enum.each(filepaths, &File.rm/1)
+  end
+
   defp add_source_details_to_changeset(source, changeset) do
     %Ecto.Changeset{changes: changes} = changeset
 
@@ -196,6 +209,9 @@ defmodule Pinchflat.Sources do
       {:ok, %Source{} = source} ->
         maybe_handle_media_tasks(changeset, source)
         maybe_run_indexing_task(changeset, source)
+        run_metadata_storage_task(source)
+
+        {:ok, source}
 
       err ->
         err
@@ -215,8 +231,6 @@ defmodule Pinchflat.Sources do
       _ ->
         :ok
     end
-
-    {:ok, source}
   end
 
   defp maybe_run_indexing_task(changeset, source) do
@@ -231,8 +245,11 @@ defmodule Pinchflat.Sources do
         maybe_update_slow_indexing_task(changeset, source)
         maybe_update_fast_indexing_task(changeset, source)
     end
+  end
 
-    {:ok, source}
+  # This runs every time to pick up any changes to the metadata
+  defp run_metadata_storage_task(source) do
+    SourceMetadataStorageWorker.kickoff_with_task(source)
   end
 
   defp maybe_update_slow_indexing_task(changeset, source) do

lib/pinchflat/tasks/tasks.ex

@@ -20,6 +20,8 @@ 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

lib/pinchflat/yt_dlp/media_collection.ex

@@ -50,8 +50,8 @@ defmodule Pinchflat.YtDlp.MediaCollection do
   @doc """
   Gets a source's ID and name from its URL.
 
-  yt-dlp does not _really_ have source-specific functions, so
+  yt-dlp does not _really_ have source-specific functions that return what
-  instead we're fetching just the first video (using playlist_end: 1)
+  we need, so instead we're fetching just the first video (using playlist_end: 1)
   and parsing the source ID and name from _its_ metadata
 
   Returns {:ok, map()} | {:error, any, ...}.
@@ -71,7 +71,35 @@ defmodule Pinchflat.YtDlp.MediaCollection do
     end
   end
 
+  @doc """
+  Gets a source's metadata from its URL.
+
+  This is mostly for things like getting the source's avatar and banner image
+  (if applicable). However, this yt-dlp call doesn't have enough overlap with
+  `get_source_details/1` to allow combining them - this one has _almost_ everything
+  we need, but it doesn't contain enough information to tell 100% if the url is a channel
+  or a playlist.
+
+  The main purpose of this (past using as a fetcher for _other_ metadata) is to live
+  as a compressed blob for possible future use. That's why it's not getting formatted like
+  `get_source_details/1`
+
+  Returns {:ok, map()} | {:error, any, ...}.
+  """
+  def get_source_metadata(source_url) do
+    opts = [playlist_items: 0]
+    output_template = "playlist:%()j"
+
+    with {:ok, output} <- backend_runner().run(source_url, opts, output_template),
+         {:ok, parsed_json} <- Phoenix.json_library().decode(output) do
+      {:ok, parsed_json}
+    else
+      err -> err
+    end
+  end
+
   defp format_source_details(response) do
+    # NOTE: I should probably make this a struct some day
     %{
       channel_id: response["channel_id"],
       channel_name: response["channel"],

priv/repo/migrations/20240314174731_create_source_metadata.exs

@@ -0,0 +1,14 @@
+defmodule Pinchflat.Repo.Migrations.CreateSourceMetadata do
+  use Ecto.Migration
+
+  def change do
+    create table(:source_metadata) do
+      add :metadata_filepath, :string, null: false
+      add :source_id, references(:sources, on_delete: :delete_all), null: false
+
+      timestamps(type: :utc_datetime)
+    end
+
+    create unique_index(:source_metadata, [:source_id])
+  end
+end

test/pinchflat/metadata/source_metadata_storage_worker_test.exs

@@ -0,0 +1,55 @@
+defmodule Pinchflat.Metadata.SourceMetadataStorageWorkerTest do
+  use Pinchflat.DataCase
+  import Mox
+  import Pinchflat.SourcesFixtures
+
+  alias Pinchflat.Metadata.MetadataFileHelpers
+  alias Pinchflat.Metadata.SourceMetadataStorageWorker
+
+  setup :verify_on_exit!
+
+  describe "kickoff_with_task/1" do
+    test "enqueues a new worker for the source" do
+      source = source_fixture()
+
+      assert {:ok, _} = SourceMetadataStorageWorker.kickoff_with_task(source)
+
+      assert_enqueued(worker: SourceMetadataStorageWorker, args: %{"id" => source.id})
+    end
+
+    test "creates a new task for the source" do
+      source = source_fixture()
+
+      assert {:ok, task} = SourceMetadataStorageWorker.kickoff_with_task(source)
+
+      assert task.source_id == source.id
+    end
+  end
+
+  describe "perform/1" do
+    test "sets metadata location for source" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, "{}"} end)
+      source = Repo.preload(source_fixture(), :metadata)
+
+      refute source.metadata
+      perform_job(SourceMetadataStorageWorker, %{id: source.id})
+      source = Repo.preload(Repo.reload(source), :metadata)
+
+      assert source.metadata.metadata_filepath
+
+      File.rm!(source.metadata.metadata_filepath)
+    end
+
+    test "fetches and stores returned metadata for source" do
+      source = source_fixture()
+      file_contents = Phoenix.json_library().encode!(%{"title" => "test"})
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, file_contents} end)
+
+      perform_job(SourceMetadataStorageWorker, %{id: source.id})
+      source = Repo.preload(Repo.reload(source), :metadata)
+      {:ok, metadata} = MetadataFileHelpers.read_compressed_metadata(source.metadata.metadata_filepath)
+
+      assert metadata == %{"title" => "test"}
+    end
+  end
+end

test/pinchflat/slow_indexing/media_collection_indexing_worker_test.exs

@@ -101,7 +101,10 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorkerTest do
       expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end)
 
       source = source_fixture(index_frequency_minutes: 10)
-      task_count_fetcher = fn -> Enum.count(Tasks.list_tasks()) end
+
+      task_count_fetcher = fn ->
+        Enum.count(Tasks.list_tasks_for(:source_id, source.id, "MediaCollectionIndexingWorker"))
+      end
 
       assert_changed([from: 0, to: 1], task_count_fetcher, fn ->
         perform_job(MediaCollectionIndexingWorker, %{id: source.id})

test/pinchflat/sources_test.exs

@@ -8,16 +8,32 @@ defmodule Pinchflat.SourcesTest do
 
   alias Pinchflat.Sources
   alias Pinchflat.Sources.Source
+  alias Pinchflat.Metadata.MetadataFileHelpers
   alias Pinchflat.Downloading.DownloadingHelpers
   alias Pinchflat.FastIndexing.FastIndexingWorker
   alias Pinchflat.Downloading.MediaDownloadWorker
   alias Pinchflat.FastIndexing.MediaIndexingWorker
+  alias Pinchflat.Metadata.SourceMetadataStorageWorker
   alias Pinchflat.SlowIndexing.MediaCollectionIndexingWorker
 
   @invalid_source_attrs %{name: nil, collection_id: nil}
 
   setup :verify_on_exit!
 
+  describe "schema" do
+    test "source_metadata is deleted when the source is deleted" do
+      source =
+        source_fixture(%{metadata: %{metadata_filepath: "/metadata.json.gz"}})
+
+      metadata = source.metadata
+      assert {:ok, %Source{}} = Sources.delete_source(source)
+
+      assert_raise Ecto.NoResultsError, fn ->
+        Repo.reload!(metadata)
+      end
+    end
+  end
+
   describe "list_sources/0" do
     test "it returns all sources" do
       source = source_fixture()
@@ -220,6 +236,21 @@ defmodule Pinchflat.SourcesTest do
 
       assert source.index_frequency_minutes == 0
     end
+
+    test "creating will kickoff a metadata storage worker" do
+      expect(YtDlpRunnerMock, :run, &channel_mock/3)
+
+      valid_attrs = %{
+        media_profile_id: media_profile_fixture().id,
+        original_url: "https://www.youtube.com/channel/abc123",
+        fast_index: false,
+        index_frequency_minutes: 0
+      }
+
+      assert {:ok, %Source{} = source} = Sources.create_source(valid_attrs)
+
+      assert_enqueued(worker: SourceMetadataStorageWorker, args: %{"id" => source.id})
+    end
   end
 
   describe "update_source/2" do
@@ -384,6 +415,15 @@ defmodule Pinchflat.SourcesTest do
 
       assert source.index_frequency_minutes == 0
     end
+
+    test "updating will kickoff a metadata storage worker" do
+      source = source_fixture()
+      update_attrs = %{name: "some updated name"}
+
+      assert {:ok, %Source{} = source} = Sources.update_source(source, update_attrs)
+
+      assert_enqueued(worker: SourceMetadataStorageWorker, args: %{"id" => source.id})
+    end
   end
 
   describe "delete_source/2" do
@@ -421,6 +461,22 @@ defmodule Pinchflat.SourcesTest do
       assert {:ok, %Source{}} = Sources.delete_source(source)
       assert File.exists?(media_item.media_filepath)
     end
+
+    test "deletes the source's metadata files" do
+      stub(HTTPClientMock, :get, fn _url, _headers, _opts -> {:ok, ""} end)
+      source = Repo.preload(source_fixture(), :metadata)
+
+      update_attrs = %{
+        metadata: %{
+          metadata_filepath: MetadataFileHelpers.compress_and_store_metadata_for(source, %{})
+        }
+      }
+
+      {:ok, updated_source} = Sources.update_source(source, update_attrs)
+
+      assert {:ok, _} = Sources.delete_source(updated_source, delete_files: true)
+      refute File.exists?(updated_source.metadata.metadata_filepath)
+    end
   end
 
   describe "delete_source/2 when deleting files" do
@@ -452,18 +508,18 @@ defmodule Pinchflat.SourcesTest do
     end
   end
 
-  describe "change_source_from_url/2" do
+  describe "maybe_change_source_from_url/2" do
     test "it returns a changeset" do
       stub(YtDlpRunnerMock, :run, &channel_mock/3)
       source = source_fixture()
 
-      assert %Ecto.Changeset{} = Sources.change_source_from_url(source, %{})
+      assert %Ecto.Changeset{} = Sources.maybe_change_source_from_url(source, %{})
     end
 
     test "it does not fetch source details if the original_url isn't in the changeset" do
       expect(YtDlpRunnerMock, :run, 0, &channel_mock/3)
 
-      changeset = Sources.change_source_from_url(%Source{}, %{name: "some updated name"})
+      changeset = Sources.maybe_change_source_from_url(%Source{}, %{name: "some updated name"})
 
       assert %Ecto.Changeset{} = changeset
     end
@@ -472,7 +528,7 @@ defmodule Pinchflat.SourcesTest do
       expect(YtDlpRunnerMock, :run, &channel_mock/3)
 
       changeset =
-        Sources.change_source_from_url(%Source{}, %{
+        Sources.maybe_change_source_from_url(%Source{}, %{
           original_url: "https://www.youtube.com/channel/abc123"
         })
 
@@ -486,7 +542,7 @@ defmodule Pinchflat.SourcesTest do
       media_profile_id = media_profile.id
 
       changeset =
-        Sources.change_source_from_url(%Source{}, %{
+        Sources.maybe_change_source_from_url(%Source{}, %{
           original_url: "https://www.youtube.com/channel/abc123",
           media_profile_id: media_profile.id
         })
@@ -507,7 +563,7 @@ defmodule Pinchflat.SourcesTest do
       end)
 
       changeset =
-        Sources.change_source_from_url(%Source{}, %{
+        Sources.maybe_change_source_from_url(%Source{}, %{
           original_url: "https://www.youtube.com/channel/abc123"
         })
 

test/pinchflat/yt_dlp/media_collection_test.exs

@@ -108,4 +108,39 @@ defmodule Pinchflat.YtDlp.MediaCollectionTest do
       assert {:error, %Jason.DecodeError{}} = MediaCollection.get_source_details(@channel_url)
     end
   end
+
+  describe "get_source_metadata/1" do
+    test "it returns a map with data on success" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
+        Phoenix.json_library().encode(%{channel: "TheUselessTrials"})
+      end)
+
+      assert {:ok, res} = MediaCollection.get_source_metadata(@channel_url)
+
+      assert %{"channel" => "TheUselessTrials"} = res
+    end
+
+    test "it passes the expected args to the backend runner" do
+      expect(YtDlpRunnerMock, :run, fn @channel_url, opts, ot ->
+        assert opts == [playlist_items: 0]
+        assert ot == "playlist:%()j"
+
+        {:ok, "{}"}
+      end)
+
+      assert {:ok, _} = MediaCollection.get_source_metadata(@channel_url)
+    end
+
+    test "it returns an error if the runner returns an error" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:error, "Big issue", 1} end)
+
+      assert {:error, "Big issue", 1} = MediaCollection.get_source_metadata(@channel_url)
+    end
+
+    test "it returns an error if the output is not JSON" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, "Not JSON"} end)
+
+      assert {:error, %Jason.DecodeError{}} = MediaCollection.get_source_metadata(@channel_url)
+    end
+  end
 end

test/support/fixtures/sources_fixtures.ex

@@ -34,6 +34,20 @@ defmodule Pinchflat.SourcesFixtures do
     source
   end
 
+  @doc """
+  Generate a source with metadata.
+  """
+  def source_with_metadata(attrs \\ %{}) do
+    merged_attrs =
+      Map.merge(attrs, %{
+        metadata: %{
+          metadata_filepath: Application.get_env(:pinchflat, :metadata_directory) <> "/metadata.json.gz"
+        }
+      })
+
+    source_fixture(merged_attrs)
+  end
+
   def source_attributes_return_fixture do
     source_attributes = [
       %{