diff --git a/config/config.exs b/config/config.exs index 70b5080..4396f4b 100644 --- a/config/config.exs +++ b/config/config.exs @@ -42,6 +42,7 @@ config :pinchflat, Oban, # TODO: consider making this an env var or something? queues: [ default: 10, + fast_indexing: 6, media_indexing: 2, media_collection_indexing: 2, media_fetching: 2, diff --git a/lib/pinchflat/api/youtube_rss.ex b/lib/pinchflat/api/youtube_rss.ex index 146f1b1..7eef959 100644 --- a/lib/pinchflat/api/youtube_rss.ex +++ b/lib/pinchflat/api/youtube_rss.ex @@ -3,6 +3,8 @@ defmodule Pinchflat.Api.YoutubeRss do Methods for interacting with YouTube RSS feeds """ + require Logger + alias Pinchflat.Sources.Source @doc """ @@ -11,6 +13,8 @@ defmodule Pinchflat.Api.YoutubeRss do Returns {:ok, [binary()]} | {:error, binary()} """ def get_recent_media_ids_from_rss(%Source{} = source) do + Logger.debug("Fetching recent media IDs from YouTube RSS feed for source: #{source.collection_id}") + case http_client().get(rss_url_for_source(source)) do {:ok, response} -> response = to_string(response) @@ -25,6 +29,8 @@ defmodule Pinchflat.Api.YoutubeRss do |> Enum.filter(&(String.length(&1) > 0)) |> Enum.uniq() + Logger.debug("Media ids fetched from RSS: #{inspect(media_ids)}") + {:ok, media_ids} {:error, _reason} -> diff --git a/lib/pinchflat/sources.ex b/lib/pinchflat/sources.ex index 95bfde8..eee7c64 100644 --- a/lib/pinchflat/sources.ex +++ b/lib/pinchflat/sources.ex @@ -46,6 +46,7 @@ defmodule Pinchflat.Sources do def create_source(attrs) do %Source{} |> change_source_from_url(attrs) + |> maybe_change_indexing_frequency() |> commit_and_handle_tasks() end @@ -62,6 +63,7 @@ defmodule Pinchflat.Sources do def update_source(%Source{} = source, attrs) do source |> change_source_from_url(attrs) + |> maybe_change_indexing_frequency() |> commit_and_handle_tasks() end @@ -151,6 +153,20 @@ defmodule Pinchflat.Sources do change_source(source, Map.merge(changes, collection_changes)) end + defp maybe_change_indexing_frequency(changeset) do + fast_index = Ecto.Changeset.get_field(changeset, :fast_index) + + if fast_index do + Ecto.Changeset.put_change( + changeset, + :index_frequency_minutes, + Source.index_frequency_when_fast_indexing() + ) + else + changeset + end + end + defp commit_and_handle_tasks(changeset) do case Repo.insert_or_update(changeset) do {:ok, %Source{} = source} -> @@ -188,13 +204,38 @@ defmodule Pinchflat.Sources do # If the record has been persisted, only run indexing if the # indexing frequency has been changed and is now greater than 0 %{__meta__: %{state: :loaded}} -> - case changeset.changes do - %{index_frequency_minutes: mins} when mins > 0 -> SourceTasks.kickoff_indexing_task(source) - %{index_frequency_minutes: _} -> Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker") - _ -> :ok - end + maybe_update_slow_indexing_task(changeset, source) + maybe_update_fast_indexing_task(changeset, source) end {:ok, source} end + + defp maybe_update_slow_indexing_task(changeset, source) do + case changeset.changes do + %{index_frequency_minutes: mins} when mins > 0 -> + SourceTasks.kickoff_indexing_task(source) + + %{index_frequency_minutes: _} -> + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") + Tasks.delete_pending_tasks_for(source, "MediaIndexingWorker") + Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker") + + _ -> + :ok + end + end + + defp maybe_update_fast_indexing_task(changeset, source) do + case changeset.changes do + %{fast_index: true} -> + SourceTasks.kickoff_fast_indexing_task(source) + + %{fast_index: false} -> + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") + + _ -> + :ok + end + end end diff --git a/lib/pinchflat/sources/source.ex b/lib/pinchflat/sources/source.ex index 6282e28..0ed8aaa 100644 --- a/lib/pinchflat/sources/source.ex +++ b/lib/pinchflat/sources/source.ex @@ -17,6 +17,7 @@ defmodule Pinchflat.Sources.Source do collection_type custom_name index_frequency_minutes + fast_index download_media last_indexed_at original_url @@ -29,6 +30,7 @@ defmodule Pinchflat.Sources.Source do collection_type custom_name index_frequency_minutes + fast_index download_media original_url media_profile_id @@ -40,6 +42,7 @@ defmodule Pinchflat.Sources.Source do field :collection_id, :string field :collection_type, Ecto.Enum, values: [:channel, :playlist] field :index_frequency_minutes, :integer, default: 60 * 24 + 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 @@ -62,4 +65,16 @@ defmodule Pinchflat.Sources.Source do |> validate_required(@required_fields) |> unique_constraint([:collection_id, :media_profile_id]) end + + @doc false + def index_frequency_when_fast_indexing do + # 30 days in minutes + 60 * 24 * 30 + end + + @doc false + def fast_index_frequency do + # minutes + 15 + end end diff --git a/lib/pinchflat/tasks/source_tasks.ex b/lib/pinchflat/tasks/source_tasks.ex index 8fab791..4c786ef 100644 --- a/lib/pinchflat/tasks/source_tasks.ex +++ b/lib/pinchflat/tasks/source_tasks.ex @@ -14,6 +14,7 @@ defmodule Pinchflat.Tasks.SourceTasks do alias Pinchflat.Sources.Source alias Pinchflat.Api.YoutubeRss alias Pinchflat.Media.MediaItem + alias Pinchflat.Workers.FastIndexingWorker alias Pinchflat.Workers.MediaDownloadWorker alias Pinchflat.Workers.MediaIndexingWorker alias Pinchflat.YtDlp.Backend.MediaCollection @@ -24,22 +25,38 @@ defmodule Pinchflat.Tasks.SourceTasks do @doc """ Starts tasks for indexing a source's media regardless of the source's indexing - frequency. It's assumed the caller will check for that. + frequency. It's assumed the caller will check for indexing frequency. Returns {:ok, %Task{}}. """ def kickoff_indexing_task(%Source{} = source) do + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") + Tasks.delete_pending_tasks_for(source, "MediaIndexingWorker") Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker") %{id: source.id} # Schedule this one immediately, but future ones will be on an interval |> MediaCollectionIndexingWorker.new() |> Tasks.create_job_with_task(source) - |> 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 + + @doc """ + Starts tasks for running a fast indexing task for a source's media + regardless of the source's fast_index state. It's assumed the + caller will check for fast_index. + + This is used for running fast index tasks on update. On creation, the + fast index is enqueued after the slow index is complete. + + Returns {:ok, %Task{}}. + """ + def kickoff_fast_indexing_task(%Source{} = source) do + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") + + %{id: source.id} + # Schedule this one immediately, but future ones will be on an interval + |> FastIndexingWorker.new() + |> Tasks.create_job_with_task(source) end @doc """ @@ -90,8 +107,8 @@ defmodule Pinchflat.Tasks.SourceTasks do # See the method definition below for more info on how file watchers work # (important reading if you're not familiar with it) {:ok, media_attributes} = get_media_attributes_for_collection_and_setup_file_watcher(source) - result = Enum.map(media_attributes, fn media_attrs -> create_media_item_from_attributes(source, media_attrs) end) + Sources.update_source(source, %{last_indexed_at: DateTime.utc_now()}) enqueue_pending_media_tasks(source) diff --git a/lib/pinchflat/workers/fast_indexing_worker.ex b/lib/pinchflat/workers/fast_indexing_worker.ex new file mode 100644 index 0000000..e2f135e --- /dev/null +++ b/lib/pinchflat/workers/fast_indexing_worker.ex @@ -0,0 +1,42 @@ +defmodule Pinchflat.Workers.FastIndexingWorker do + @moduledoc false + + use Oban.Worker, + queue: :fast_indexing, + unique: [period: :infinity, states: [:available, :scheduled, :retryable]], + tags: ["media_source", "fast_indexing"] + + alias __MODULE__ + alias Pinchflat.Tasks + alias Pinchflat.Sources + alias Pinchflat.Sources.Source + alias Pinchflat.Tasks.SourceTasks + + @impl Oban.Worker + @doc """ + TODO + """ + def perform(%Oban.Job{args: %{"id" => source_id}}) do + source = Sources.get_source!(source_id) + + if source.fast_index do + SourceTasks.kickoff_indexing_tasks_from_youtube_rss_feed(source) + + reschedule_indexing(source) + else + :ok + end + end + + defp reschedule_indexing(source) do + next_run_in = Source.fast_index_frequency() * 60 + + %{id: source.id} + |> FastIndexingWorker.new(schedule_in: next_run_in) + |> Tasks.create_job_with_task(source) + |> case do + {:ok, task} -> {:ok, task} + {:error, :duplicate_job} -> {:ok, :job_exists} + end + end +end diff --git a/lib/pinchflat/workers/media_collection_indexing_worker.ex b/lib/pinchflat/workers/media_collection_indexing_worker.ex index d5aaa6e..3f7122e 100644 --- a/lib/pinchflat/workers/media_collection_indexing_worker.ex +++ b/lib/pinchflat/workers/media_collection_indexing_worker.ex @@ -9,7 +9,9 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do alias __MODULE__ alias Pinchflat.Tasks alias Pinchflat.Sources + alias Pinchflat.Sources.Source alias Pinchflat.Tasks.SourceTasks + alias Pinchflat.Workers.FastIndexingWorker @impl Oban.Worker @doc """ @@ -30,6 +32,27 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker 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. + Order of operations: + 1. The user saves a source + 2. This job is automatically scheduled immediately. This happens in all cases. + 3. This job indexes all content for the given source. A download job is + enqueued for each media item that should be downloaded. This can be impacted + by the `download_media` field on the source as well as the profile's + shorts/livestream behaviour. At this step we also attach a file reader + to the `yt-dlp` output file so we can create media items as they come in + for a little speedup (see SourceTasks comments for more) + 4. If this job is meant to reschedule (ie: has an index frequency > 0), + it reschedules itself. If not, it runs once and does not reschedule + 5. If the source uses fast indexing, that job is kicked off as well. It + uses RSS to run a smaller, faster, and more frequent index. That job + handles rescheduling itself but largely has a similar behaviour to this + job in that it kicks off index and maybe download jobs. The biggest difference + is that an index job is kicked off _for each new media item_ as opposed + to one larger index job. Check out `MediaIndexingWorker` comments for more. + 6. If the job reschedules, the cycle from step 3 repeats until the heat death + of the universe. The user changing things like the index frequency can + dequeue or reschedule jobs as well + 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? @@ -44,6 +67,7 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do {index_freq, _} when index_freq > 0 -> # If the indexing is on a schedule simply run indexing and reschedule SourceTasks.index_and_enqueue_download_for_media_items(source) + maybe_enqueue_fast_indexing_task(source) reschedule_indexing(source) {_, nil} -> @@ -60,12 +84,26 @@ defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do end defp reschedule_indexing(source) do + next_run_in = source.index_frequency_minutes * 60 + %{id: source.id} - |> MediaCollectionIndexingWorker.new(schedule_in: source.index_frequency_minutes * 60) + |> MediaCollectionIndexingWorker.new(schedule_in: next_run_in) |> Tasks.create_job_with_task(source) |> case do {:ok, task} -> {:ok, task} {:error, :duplicate_job} -> {:ok, :job_exists} end end + + defp maybe_enqueue_fast_indexing_task(source) do + if source.fast_index do + Tasks.delete_pending_tasks_for(source, "FastIndexingWorker") + + next_run_in = Source.fast_index_frequency() * 60 + + %{id: source.id} + |> FastIndexingWorker.new(schedule_in: next_run_in) + |> Tasks.create_job_with_task(source) + end + end end diff --git a/lib/pinchflat_web/controllers/sources/source_controller.ex b/lib/pinchflat_web/controllers/sources/source_controller.ex index 757fb44..60039f4 100644 --- a/lib/pinchflat_web/controllers/sources/source_controller.ex +++ b/lib/pinchflat_web/controllers/sources/source_controller.ex @@ -3,8 +3,9 @@ defmodule PinchflatWeb.Sources.SourceController do alias Pinchflat.Repo alias Pinchflat.Media - alias Pinchflat.Profiles + alias Pinchflat.Tasks alias Pinchflat.Sources + alias Pinchflat.Profiles alias Pinchflat.Sources.Source def index(conn, _params) do @@ -43,15 +44,18 @@ defmodule PinchflatWeb.Sources.SourceController do end def show(conn, %{"id" => id}) do - source = - id - |> Sources.get_source!() - |> Repo.preload([:media_profile, tasks: [:job]]) + source = Repo.preload(Sources.get_source!(id), :media_profile) + pending_tasks = Repo.preload(Tasks.list_pending_tasks_for(:source_id, source.id), :job) pending_media = Media.list_pending_media_items_for(source, limit: 100) downloaded_media = Media.list_downloaded_media_items_for(source, limit: 100) - render(conn, :show, source: source, pending_media: pending_media, downloaded_media: downloaded_media) + render(conn, :show, + source: source, + pending_tasks: pending_tasks, + pending_media: pending_media, + downloaded_media: downloaded_media + ) end def edit(conn, %{"id" => id}) do diff --git a/lib/pinchflat_web/controllers/sources/source_html/fast_indexing_help.html.heex b/lib/pinchflat_web/controllers/sources/source_html/fast_indexing_help.html.heex new file mode 100644 index 0000000..710b2ed --- /dev/null +++ b/lib/pinchflat_web/controllers/sources/source_html/fast_indexing_help.html.heex @@ -0,0 +1,21 @@ + diff --git a/lib/pinchflat_web/controllers/sources/source_html/show.html.heex b/lib/pinchflat_web/controllers/sources/source_html/show.html.heex index d8c69d2..1b08883 100644 --- a/lib/pinchflat_web/controllers/sources/source_html/show.html.heex +++ b/lib/pinchflat_web/controllers/sources/source_html/show.html.heex @@ -84,9 +84,9 @@
Nothing Here!
<% end %> - <:tab title="Tasks"> - <%= if match?([_|_], @source.tasks) do %> - <.table rows={@source.tasks} table_class="text-black dark:text-white"> + <:tab title="Pending Tasks"> + <%= if match?([_|_], @pending_tasks) do %> + <.table rows={@pending_tasks} table_class="text-black dark:text-white"> <:col :let={task} label="Worker"> <%= task.job.worker %> diff --git a/lib/pinchflat_web/controllers/sources/source_html/source_form.html.heex b/lib/pinchflat_web/controllers/sources/source_html/source_form.html.heex index c538c96..8124106 100644 --- a/lib/pinchflat_web/controllers/sources/source_html/source_form.html.heex +++ b/lib/pinchflat_web/controllers/sources/source_html/source_form.html.heex @@ -3,6 +3,10 @@ Oops, something went wrong! Please check the errors below. +