Refactor modules into contexts (#78)
* [WIP] break out a few contexts, start refactoring fast index modules * [WIP] more contexts, this time around slow indexing and downloads * [WIP] got all tests passing * [WIP] Added moduledocs * Built a genserver to rename old jobs on boot * Added a module naming check; moved things around * Fixed specs
This commit is contained in:
Kieran
GitHub
@@ -1,61 +0,0 @@
|
||||
defmodule Pinchflat.Tasks.MediaItemTasks do
|
||||
@moduledoc """
|
||||
Contains methods used by OR used to create/manage tasks for media items.
|
||||
|
||||
Tasks/workers are meant to be thin wrappers so most of the actual work they
|
||||
do is also defined here. Essentially, a one-stop-shop for media-related tasks/workers.
|
||||
"""
|
||||
alias Pinchflat.Media
|
||||
alias Pinchflat.Tasks
|
||||
alias Pinchflat.Sources.Source
|
||||
alias Pinchflat.Workers.MediaDownloadWorker
|
||||
|
||||
alias Pinchflat.YtDlp.Backend.Media, as: YtDlpMedia
|
||||
|
||||
@doc """
|
||||
Fetches the file size of a media item and saves it to the database.
|
||||
|
||||
Returns {:ok, media_item} | {:error, any()}
|
||||
"""
|
||||
def compute_and_save_media_filesize(media_item) do
|
||||
case File.stat(media_item.media_filepath) do
|
||||
{:ok, %{size: size}} ->
|
||||
Media.update_media_item(media_item, %{media_size_bytes: size})
|
||||
|
||||
err ->
|
||||
err
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Indexes a single media item for a source and enqueues a download job if the
|
||||
media should be downloaded. This method creates the media item record so it's
|
||||
the one-stop-shop for adding a media item (and possibly downloading it) just
|
||||
by a URL and source.
|
||||
|
||||
Returns {:ok, media_item} | {:error, any()}
|
||||
"""
|
||||
def index_and_enqueue_download_for_media_item(%Source{} = source, url) do
|
||||
maybe_media_item = create_media_item_from_url(source, url)
|
||||
|
||||
case maybe_media_item do
|
||||
{:ok, media_item} ->
|
||||
if source.download_media && Media.pending_download?(media_item) do
|
||||
%{id: media_item.id}
|
||||
|> MediaDownloadWorker.new()
|
||||
|> Tasks.create_job_with_task(media_item)
|
||||
end
|
||||
|
||||
{:ok, media_item}
|
||||
|
||||
err ->
|
||||
err
|
||||
end
|
||||
end
|
||||
|
||||
defp create_media_item_from_url(source, url) do
|
||||
{:ok, media_attrs} = YtDlpMedia.get_media_attributes(url)
|
||||
|
||||
Media.create_media_item_from_backend_attrs(source, media_attrs)
|
||||
end
|
||||
end
|
||||
@@ -1,217 +0,0 @@
|
||||
defmodule Pinchflat.Tasks.SourceTasks do
|
||||
@moduledoc """
|
||||
Contains methods used by OR used to create/manage tasks for sources.
|
||||
|
||||
Tasks/workers are meant to be thin wrappers so most of the actual work they
|
||||
do is also defined here. Essentially, a one-stop-shop for source-related tasks/workers.
|
||||
"""
|
||||
|
||||
require Logger
|
||||
|
||||
alias Pinchflat.Media
|
||||
alias Pinchflat.Tasks
|
||||
alias Pinchflat.Sources
|
||||
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
|
||||
alias Pinchflat.Workers.MediaCollectionIndexingWorker
|
||||
alias Pinchflat.Utils.FilesystemUtils.FileFollowerServer
|
||||
|
||||
alias Pinchflat.YtDlp.Backend.Media, as: YtDlpMedia
|
||||
|
||||
@doc """
|
||||
Starts tasks for indexing a source's media regardless of the source's indexing
|
||||
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)
|
||||
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 """
|
||||
Fetches new media IDs from a source's YouTube RSS feed and kicks off indexing tasks
|
||||
for any new media items. See comments in `MediaIndexingWorker` for more info on the
|
||||
order of operations and how this fits into the indexing process.
|
||||
|
||||
Returns :ok
|
||||
"""
|
||||
def kickoff_indexing_tasks_from_youtube_rss_feed(%Source{} = source) do
|
||||
{:ok, media_ids} = YoutubeRss.get_recent_media_ids_from_rss(source)
|
||||
existing_media_items = Media.list_media_items_by_media_id_for(source, media_ids)
|
||||
new_media_ids = media_ids -- Enum.map(existing_media_items, & &1.media_id)
|
||||
|
||||
Enum.each(new_media_ids, fn media_id ->
|
||||
url = "https://www.youtube.com/watch?v=#{media_id}"
|
||||
|
||||
%{id: source.id, media_url: url}
|
||||
|> MediaIndexingWorker.new()
|
||||
|> Tasks.create_job_with_task(source)
|
||||
end)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Given a media source, creates (indexes) the media by creating media_items for each
|
||||
media ID in the source. Afterward, kicks off a download task for each pending media
|
||||
item belonging to the source. You can't tell me the method name isn't descriptive!
|
||||
|
||||
Indexing is slow and usually returns a list of all media data at once for record creation.
|
||||
To help with this, we use a file follower to watch the file that yt-dlp writes to
|
||||
so we can create media items as they come in. This parallelizes the process and adds
|
||||
clarity to the user experience. This has a few things to be aware of which are documented
|
||||
below in the file watcher setup method.
|
||||
|
||||
NOTE: downloads are only enqueued if the source is set to download media. Downloads are
|
||||
also enqueued for ALL pending media items, 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.
|
||||
|
||||
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{}, ...]
|
||||
"""
|
||||
def index_and_enqueue_download_for_media_items(%Source{} = source) 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)
|
||||
|
||||
result
|
||||
end
|
||||
|
||||
@doc """
|
||||
Starts tasks for downloading media for any of a sources _pending_ media items.
|
||||
Jobs are not enqueued if the source is set to not download media. This will return :ok.
|
||||
|
||||
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.
|
||||
|
||||
Returns :ok
|
||||
"""
|
||||
def enqueue_pending_media_tasks(%Source{download_media: true} = source) do
|
||||
source
|
||||
|> Media.list_pending_media_items_for()
|
||||
|> Enum.each(fn media_item ->
|
||||
%{id: media_item.id}
|
||||
|> MediaDownloadWorker.new()
|
||||
|> Tasks.create_job_with_task(media_item)
|
||||
end)
|
||||
end
|
||||
|
||||
def enqueue_pending_media_tasks(%Source{download_media: false} = _source) do
|
||||
:ok
|
||||
end
|
||||
|
||||
@doc """
|
||||
Deletes ALL pending tasks for a source's media items.
|
||||
|
||||
Returns :ok
|
||||
"""
|
||||
def dequeue_pending_media_tasks(%Source{} = source) do
|
||||
source
|
||||
|> Media.list_pending_media_items_for()
|
||||
|> Enum.each(&Tasks.delete_pending_tasks_for/1)
|
||||
end
|
||||
|
||||
# The file follower is a GenServer that watches a file for new lines and
|
||||
# processes them. This works well, but we have to be resilliant to partially-written
|
||||
# lines (ie: you should gracefully fail if you can't parse a line).
|
||||
#
|
||||
# This works in-tandem with the normal (blocking) media indexing behaviour. When
|
||||
# the `get_media_attributes_for_collection` method completes it'll return the FULL result to
|
||||
# the caller for parsing. Ideally, every item in the list will have already
|
||||
# been processed by the file follower, but if not, the caller handles creation
|
||||
# of any media items that were missed/initially failed.
|
||||
#
|
||||
# It attempts a graceful shutdown of the file follower after the indexing is done,
|
||||
# but the FileFollowerServer will also stop itself if it doesn't see any activity
|
||||
# for a sufficiently long time.
|
||||
defp get_media_attributes_for_collection_and_setup_file_watcher(source) do
|
||||
{:ok, pid} = FileFollowerServer.start_link()
|
||||
|
||||
handler = fn filepath -> setup_file_follower_watcher(pid, filepath, source) end
|
||||
result = MediaCollection.get_media_attributes_for_collection(source.original_url, file_listener_handler: handler)
|
||||
|
||||
FileFollowerServer.stop(pid)
|
||||
|
||||
result
|
||||
end
|
||||
|
||||
defp setup_file_follower_watcher(pid, filepath, source) do
|
||||
FileFollowerServer.watch_file(pid, filepath, fn line ->
|
||||
case Phoenix.json_library().decode(line) do
|
||||
{:ok, media_attrs} ->
|
||||
Logger.debug("FileFollowerServer Handler: Got media attributes: #{inspect(media_attrs)}")
|
||||
|
||||
media_struct = YtDlpMedia.response_to_struct(media_attrs)
|
||||
create_media_item_and_enqueue_download(source, media_struct)
|
||||
|
||||
err ->
|
||||
Logger.debug("FileFollowerServer Handler: Error decoding JSON: #{inspect(err)}")
|
||||
|
||||
err
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
defp create_media_item_and_enqueue_download(source, media_attrs) do
|
||||
maybe_media_item = create_media_item_from_attributes(source, media_attrs)
|
||||
|
||||
case maybe_media_item do
|
||||
%MediaItem{} = media_item ->
|
||||
if source.download_media && Media.pending_download?(media_item) do
|
||||
Logger.debug("FileFollowerServer Handler: Enqueuing download task for #{inspect(media_attrs)}")
|
||||
|
||||
%{id: media_item.id}
|
||||
|> MediaDownloadWorker.new()
|
||||
|> Tasks.create_job_with_task(media_item)
|
||||
end
|
||||
|
||||
changeset ->
|
||||
changeset
|
||||
end
|
||||
end
|
||||
|
||||
defp create_media_item_from_attributes(source, media_attrs) do
|
||||
case Media.create_media_item_from_backend_attrs(source, media_attrs) do
|
||||
{:ok, media_item} -> media_item
|
||||
{:error, changeset} -> changeset
|
||||
end
|
||||
end
|
||||
end
|
||||
@@ -0,0 +1,161 @@
|
||||
defmodule Pinchflat.Tasks do
|
||||
@moduledoc """
|
||||
The Tasks context.
|
||||
"""
|
||||
import Ecto.Query, warn: false
|
||||
alias Pinchflat.Repo
|
||||
|
||||
alias Pinchflat.Tasks.Task
|
||||
alias Pinchflat.Media.MediaItem
|
||||
alias Pinchflat.Sources.Source
|
||||
|
||||
@doc """
|
||||
Returns the list of tasks. Returns [%Task{}, ...]
|
||||
"""
|
||||
def list_tasks do
|
||||
Repo.all(Task)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Returns the list of tasks for a given record type and ID. Optionally allows you to specify
|
||||
which worker or job states to include.
|
||||
|
||||
Returns [%Task{}, ...]
|
||||
"""
|
||||
def list_tasks_for(attached_record_type, attached_record_id, worker_name \\ nil, job_states \\ Oban.Job.states()) do
|
||||
stringified_states = Enum.map(job_states, &to_string/1)
|
||||
|
||||
worker_name_finder =
|
||||
if worker_name do
|
||||
# Workers are the full module name - we want to match on the string ENDING with
|
||||
# the passed worker name and it should be preceeded with a . so we aren't matching
|
||||
# on a substring. You can pass in more fragments of the worker name if you need
|
||||
# to disambiguate. eg: "TestWorker" or "FooBar.TestWorker"
|
||||
worker_finder = "%.#{worker_name}"
|
||||
|
||||
dynamic([_t, j], fragment("? LIKE ?", j.worker, ^worker_finder))
|
||||
else
|
||||
true
|
||||
end
|
||||
|
||||
Repo.all(
|
||||
from t in Task,
|
||||
join: j in assoc(t, :job),
|
||||
where: field(t, ^attached_record_type) == ^attached_record_id,
|
||||
where: ^worker_name_finder,
|
||||
where: j.state in ^stringified_states
|
||||
)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Returns the list of pending tasks for a given record type and ID. Optionally allows you to specify
|
||||
which worker to include.
|
||||
|
||||
Returns [%Task{}, ...]
|
||||
"""
|
||||
def list_pending_tasks_for(attached_record_type, attached_record_id, worker_name \\ nil) do
|
||||
list_tasks_for(
|
||||
attached_record_type,
|
||||
attached_record_id,
|
||||
worker_name,
|
||||
[:available, :scheduled, :retryable]
|
||||
)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Gets a single task.
|
||||
|
||||
Returns %Task{}. Raises `Ecto.NoResultsError` if the Task does not exist.
|
||||
"""
|
||||
def get_task!(id), do: Repo.get!(Task, id)
|
||||
|
||||
@doc """
|
||||
Creates a task.
|
||||
|
||||
Accepts map() | %Oban.Job{}, %Source{} | %Oban.Job{}, %MediaItem{}.
|
||||
Returns {:ok, %Task{}} | {:error, %Ecto.Changeset{}}.
|
||||
"""
|
||||
def create_task(attrs) do
|
||||
%Task{}
|
||||
|> Task.changeset(attrs)
|
||||
|> Repo.insert()
|
||||
end
|
||||
|
||||
# This function's signature is designed to help simplify
|
||||
# usage of `create_job_with_task/2`
|
||||
def create_task(%Oban.Job{} = job, attached_record) do
|
||||
attached_record_attr =
|
||||
case attached_record do
|
||||
%Source{} = source -> %{source_id: source.id}
|
||||
%MediaItem{} = media_item -> %{media_item_id: media_item.id}
|
||||
end
|
||||
|
||||
%Task{}
|
||||
|> 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. Returns an error if the job already exists.
|
||||
|
||||
Returns {:ok, %Task{}} | {:error, :duplicate_job} | {:error, %Ecto.Changeset{}}.
|
||||
"""
|
||||
def create_job_with_task(job_attrs, task_attached_record) 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
|
||||
|
||||
@doc """
|
||||
Deletes a task. Also cancels any attached job.
|
||||
|
||||
Returns {:ok, %Task{}} | {:error, %Ecto.Changeset{}}.
|
||||
"""
|
||||
def delete_task(%Task{} = task) do
|
||||
:ok = Oban.cancel_job(task.job_id)
|
||||
|
||||
Repo.delete(task)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Deletes all tasks attached to a given record, cancelling any attached jobs.
|
||||
Optionally allows you to specify which worker to include.
|
||||
|
||||
Returns :ok
|
||||
"""
|
||||
def delete_tasks_for(attached_record, worker_name \\ nil) do
|
||||
tasks =
|
||||
case attached_record do
|
||||
%Source{} = source -> list_tasks_for(:source_id, source.id, worker_name)
|
||||
%MediaItem{} = media_item -> list_tasks_for(:media_item_id, media_item.id, worker_name)
|
||||
end
|
||||
|
||||
Enum.each(tasks, &delete_task/1)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Deletes all _pending_ tasks attached to a given record, cancelling any attached jobs.
|
||||
Optionally allows you to specify which worker to include.
|
||||
|
||||
Returns :ok
|
||||
"""
|
||||
def delete_pending_tasks_for(attached_record, worker_name \\ nil) do
|
||||
tasks =
|
||||
case attached_record do
|
||||
%Source{} = source -> list_pending_tasks_for(:source_id, source.id, worker_name)
|
||||
%MediaItem{} = media_item -> list_pending_tasks_for(:media_item_id, media_item.id, worker_name)
|
||||
end
|
||||
|
||||
Enum.each(tasks, &delete_task/1)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Returns an `%Ecto.Changeset{}` for tracking task changes.
|
||||
"""
|
||||
def change_task(%Task{} = task, attrs \\ %{}) do
|
||||
Task.changeset(task, attrs)
|
||||
end
|
||||
end
|
||||
Reference in New Issue
Block a user