Download cutoff date for sources (#69)

* Added new uploaded_at column to media items

* Updated indexer to pull upload date

* Updates media item creation to update on conflict

* Added download cutoff date to sources

* Applies cutoff date logic to pending media logic

* Updated docs

lib/pinchflat/media.ex

@@ -64,6 +64,7 @@ defmodule Pinchflat.Media do
     MediaItem
     |> where([mi], mi.source_id == ^source.id and is_nil(mi.media_filepath))
     |> where(^build_format_clauses(media_profile))
+    |> where(^maybe_apply_cutoff_date(source))
     |> Repo.maybe_limit(limit)
     |> Repo.all()
   end
@@ -92,11 +93,12 @@ defmodule Pinchflat.Media do
   Returns boolean()
   """
   def pending_download?(%MediaItem{} = media_item) do
-    media_profile = Repo.preload(media_item, source: :media_profile).source.media_profile
+    media_item = Repo.preload(media_item, source: :media_profile)
 
     MediaItem
     |> where([mi], mi.id == ^media_item.id and is_nil(mi.media_filepath))
-    |> where(^build_format_clauses(media_profile))
+    |> where(^build_format_clauses(media_item.source.media_profile))
+    |> where(^maybe_apply_cutoff_date(media_item.source))
     |> Repo.exists?()
   end
 
@@ -184,14 +186,25 @@ defmodule Pinchflat.Media do
 
   @doc """
   Creates a media item from the attributes returned by the video backend
-  (read: yt-dlp)
+  (read: yt-dlp).
+
+  Unlike `create_media_item`, this will attempt an update if the media_item
+  already exists. This is so that future indexing can pick up attributes that
+  we may not have asked for in the past (eg: upload_date)
 
   Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}
   """
   def create_media_item_from_backend_attrs(source, media_attrs_struct) do
-    %{source_id: source.id}
-    |> Map.merge(Map.from_struct(media_attrs_struct))
-    |> create_media_item()
+    attrs = Map.merge(%{source_id: source.id}, Map.from_struct(media_attrs_struct))
+
+    %MediaItem{}
+    |> MediaItem.changeset(attrs)
+    |> Repo.insert(
+      on_conflict: [
+        set: Map.to_list(attrs)
+      ],
+      conflict_target: [:source_id, :media_id]
+    )
   end
 
   @doc """
@@ -249,6 +262,14 @@ defmodule Pinchflat.Media do
     {:ok, media_item}
   end
 
+  defp maybe_apply_cutoff_date(source) do
+    if source.download_cutoff_date do
+      dynamic([mi], mi.upload_date >= ^source.download_cutoff_date)
+    else
+      dynamic(true)
+    end
+  end
+
   defp build_format_clauses(media_profile) do
     mapped_struct = Map.from_struct(media_profile)
 

lib/pinchflat/media/media_item.ex

@@ -20,6 +20,7 @@ defmodule Pinchflat.Media.MediaItem do
     :livestream,
     :source_id,
     :short_form_content,
+    :upload_date,
     # these fields are captured only on download
     :media_downloaded_at,
     :media_filepath,
@@ -28,7 +29,16 @@ defmodule Pinchflat.Media.MediaItem do
     :thumbnail_filepath,
     :metadata_filepath
   ]
-  @required_fields ~w(title original_url livestream media_id source_id short_form_content)a
+  # Pretty much all the fields captured at index are required.
+  @required_fields ~w(
+    title
+    original_url
+    livestream
+    media_id
+    source_id
+    upload_date
+    short_form_content
+    )a
 
   schema "media_items" do
     field :title, :string
@@ -38,6 +48,7 @@ defmodule Pinchflat.Media.MediaItem do
     field :livestream, :boolean, default: false
     field :short_form_content, :boolean, default: false
     field :media_downloaded_at, :utc_datetime
+    field :upload_date, :date
 
     field :media_filepath, :string
     field :media_size_bytes, :integer

lib/pinchflat/sources.ex

@@ -41,13 +41,24 @@ defmodule Pinchflat.Sources do
   original_url (if provided). Will attempt to start indexing the source's
   media if successfully inserted.
 
+  Runs an initial `change_source` check to ensure most of the source is valid
+  before making an expensive API call. Runs it through `Repo.insert` even
+  though we know it's going to fail so it picks up any addl. database errors
+  and fulfills our return contract.
+
   Returns {:ok, %Source{}} | {:error, %Ecto.Changeset{}}
   """
   def create_source(attrs) do
-    %Source{}
-    |> change_source_from_url(attrs)
-    |> maybe_change_indexing_frequency()
-    |> commit_and_handle_tasks()
+    case change_source(%Source{}, attrs, :initial) do
+      %Ecto.Changeset{valid?: true} ->
+        %Source{}
+        |> change_source_from_url(attrs)
+        |> maybe_change_indexing_frequency()
+        |> commit_and_handle_tasks()
+
+      changeset ->
+        Repo.insert(changeset)
+    end
   end
 
   @doc """
@@ -58,13 +69,24 @@ defmodule Pinchflat.Sources do
   Existing indexing tasks will be cancelled if the indexing frequency has been
   changed (logic in `SourceTasks.kickoff_indexing_task`)
 
+  Runs an initial `change_source` check to ensure most of the source is valid
+  before making an expensive API call. Runs it through `Repo.update` even
+  though we know it's going to fail so it picks up any addl. database errors
+  and fulfills our return contract.
+
   Returns {:ok, %Source{}} | {:error, %Ecto.Changeset{}}
   """
   def update_source(%Source{} = source, attrs) do
-    source
-    |> change_source_from_url(attrs)
-    |> maybe_change_indexing_frequency()
-    |> commit_and_handle_tasks()
+    case change_source(source, attrs, :initial) do
+      %Ecto.Changeset{valid?: true} ->
+        source
+        |> change_source_from_url(attrs)
+        |> maybe_change_indexing_frequency()
+        |> commit_and_handle_tasks()
+
+      changeset ->
+        Repo.update(changeset)
+    end
   end
 
   @doc """
@@ -89,8 +111,8 @@ defmodule Pinchflat.Sources do
   @doc """
   Returns an `%Ecto.Changeset{}` for tracking source changes.
   """
-  def change_source(%Source{} = source, attrs \\ %{}) do
-    Source.changeset(source, attrs)
+  def change_source(%Source{} = source, attrs \\ %{}, validation_stage \\ :pre_insert) do
+    Source.changeset(source, attrs, validation_stage)
   end
 
   @doc """

lib/pinchflat/sources/source.ex

@@ -21,14 +21,15 @@ defmodule Pinchflat.Sources.Source do
     download_media
     last_indexed_at
     original_url
+    download_cutoff_date
     media_profile_id
   )a
 
-  @required_fields ~w(
-    collection_name
-    collection_id
-    collection_type
-    custom_name
+  # 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.
+  @initially_required_fields ~w(
     index_frequency_minutes
     fast_index
     download_media
@@ -36,6 +37,14 @@ defmodule Pinchflat.Sources.Source do
     media_profile_id
   )a
 
+  @pre_insert_required_fields @initially_required_fields ++
+                                ~w(
+                                  custom_name
+                                  collection_name
+                                  collection_id
+                                  collection_type
+                                )a
+
   schema "sources" do
     field :custom_name, :string
     field :collection_name, :string
@@ -45,8 +54,8 @@ defmodule Pinchflat.Sources.Source do
     field :fast_index, :boolean, default: false
     field :download_media, :boolean, default: true
     field :last_indexed_at, :utc_datetime
-    # This should only be used for user reference going forward
-    # as the collection_id should be used for all API calls
+    # Only download media items that were published after this date
+    field :download_cutoff_date, :date
     field :original_url, :string
 
     belongs_to :media_profile, MediaProfile
@@ -58,11 +67,19 @@ defmodule Pinchflat.Sources.Source do
   end
 
   @doc false
-  def changeset(source, attrs) do
+  def changeset(source, attrs, validation_stage) do
+    # See above for rationale
+    required_fields =
+      if validation_stage == :initial do
+        @initially_required_fields
+      else
+        @pre_insert_required_fields
+      end
+
     source
     |> cast(attrs, @allowed_fields)
     |> dynamic_default(:custom_name, fn cs -> get_field(cs, :collection_name) end)
-    |> validate_required(@required_fields)
+    |> validate_required(required_fields)
     |> unique_constraint([:collection_id, :media_profile_id])
   end
 

lib/pinchflat/tasks/source_tasks.ex

@@ -96,12 +96,10 @@ defmodule Pinchflat.Tasks.SourceTasks do
   job run. This should ensure that any stragglers are caught if, for some reason, they
   weren't enqueued or somehow got de-queued.
 
-  Since indexing returns all media data EVERY TIME, we rely on the unique index of the
-  media_id to prevent duplicates. Due to both the file follower and the fact that future
-  indexing will index a lot of existing data, this method will MOSTLY return error
-  changesets (from the unique index violation) and not media items. This is intended.
+  Since indexing returns all media data EVERY TIME, we that that opportunity to update
+  indexing metadata for media items that have already been created.
 
-  Returns [%MediaItem{}, ...] | [%Ecto.Changeset{}, ...]
+  Returns [%MediaItem{}, ...]
   """
   def index_and_enqueue_download_for_media_items(%Source{} = source) do
     # See the method definition below for more info on how file watchers work

lib/pinchflat/yt_dlp/backend/media.ex

@@ -9,7 +9,8 @@ defmodule Pinchflat.YtDlp.Backend.Media do
     :description,
     :original_url,
     :livestream,
-    :short_form_content
+    :short_form_content,
+    :upload_date
   ]
 
   defstruct [
@@ -18,7 +19,8 @@ defmodule Pinchflat.YtDlp.Backend.Media do
     :description,
     :original_url,
     :livestream,
-    :short_form_content
+    :short_form_content,
+    :upload_date
   ]
 
   alias __MODULE__
@@ -67,7 +69,7 @@ defmodule Pinchflat.YtDlp.Backend.Media do
   Returns the output template for yt-dlp's indexing command.
   """
   def indexing_output_template do
-    "%(.{id,title,was_live,webpage_url,description,aspect_ratio,duration})j"
+    "%(.{id,title,was_live,webpage_url,description,aspect_ratio,duration,upload_date})j"
   end
 
   @doc """
@@ -83,7 +85,8 @@ defmodule Pinchflat.YtDlp.Backend.Media do
       description: response["description"],
       original_url: response["webpage_url"],
       livestream: response["was_live"],
-      short_form_content: short_form_content?(response)
+      short_form_content: short_form_content?(response),
+      upload_date: parse_upload_date(response["upload_date"])
     }
   end
 
@@ -100,6 +103,12 @@ defmodule Pinchflat.YtDlp.Backend.Media do
     end
   end
 
+  defp parse_upload_date(upload_date) do
+    <<year::binary-size(4)>> <> <<month::binary-size(2)>> <> <<day::binary-size(2)>> = upload_date
+
+    Date.from_iso8601!("#{year}-#{month}-#{day}")
+  end
+
   defp backend_runner do
     # This approach lets us mock the command for testing
     Application.get_env(:pinchflat, :yt_dlp_runner)