diff --git a/assets/yarn.lock b/assets/yarn.lock new file mode 100644 index 0000000..fb57ccd --- /dev/null +++ b/assets/yarn.lock @@ -0,0 +1,4 @@ +# THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY. +# yarn lockfile v1 + + diff --git a/config/config.exs b/config/config.exs index a859fb3..30edd54 100644 --- a/config/config.exs +++ b/config/config.exs @@ -27,6 +27,13 @@ config :pinchflat, PinchflatWeb.Endpoint, pubsub_server: Pinchflat.PubSub, live_view: [signing_salt: "/t5878kO"] +config :pinchflat, Oban, + repo: Pinchflat.Repo, + # Keep old jobs for 30 days for display in the UI + plugins: [{Oban.Plugins.Pruner, max_age: 30 * 24 * 60 * 60}], + # TODO: consider making this an env var or something? + queues: [default: 10, media_indexing: 2, media_fetching: 2] + # Configures the mailer # # By default it uses the "Local" adapter which stores the emails diff --git a/config/test.exs b/config/test.exs index 1ca2159..2a2bdce 100644 --- a/config/test.exs +++ b/config/test.exs @@ -5,6 +5,8 @@ config :pinchflat, yt_dlp_executable: Path.join([File.cwd!(), "/test/support/scripts/yt-dlp-mocks/repeater.sh"]), media_directory: Path.join([System.tmp_dir!(), "yt-dlp"]) +config :pinchflat, Oban, testing: :manual + # Configure your database # # The MIX_TEST_PARTITION environment variable can be used diff --git a/lib/pinchflat/application.ex b/lib/pinchflat/application.ex index 774cc5d..6842111 100644 --- a/lib/pinchflat/application.ex +++ b/lib/pinchflat/application.ex @@ -10,6 +10,7 @@ defmodule Pinchflat.Application do children = [ PinchflatWeb.Telemetry, Pinchflat.Repo, + {Oban, Application.fetch_env!(:pinchflat, Oban)}, {DNSCluster, query: Application.get_env(:pinchflat, :dns_cluster_query) || :ignore}, {Phoenix.PubSub, name: Pinchflat.PubSub}, # Start the Finch HTTP client for sending emails diff --git a/lib/pinchflat/media.ex b/lib/pinchflat/media.ex new file mode 100644 index 0000000..b8cb20a --- /dev/null +++ b/lib/pinchflat/media.ex @@ -0,0 +1,56 @@ +defmodule Pinchflat.Media do + @moduledoc """ + The Media context. + """ + + import Ecto.Query, warn: false + alias Pinchflat.Repo + + alias Pinchflat.Media.MediaItem + + @doc """ + Returns the list of media_items. Returns [%MediaItem{}, ...]. + """ + def list_media_items do + Repo.all(MediaItem) + end + + @doc """ + Gets a single media_item. + + Returns %MediaItem{}. Raises `Ecto.NoResultsError` if the Media item does not exist. + """ + def get_media_item!(id), do: Repo.get!(MediaItem, id) + + @doc """ + Creates a media_item. Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}. + """ + def create_media_item(attrs \\ %{}) do + %MediaItem{} + |> MediaItem.changeset(attrs) + |> Repo.insert() + end + + @doc """ + Updates a media_item. Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}. + """ + def update_media_item(%MediaItem{} = media_item, attrs) do + media_item + |> MediaItem.changeset(attrs) + |> Repo.update() + end + + @doc """ + Deletes a media_item. Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}. + """ + def delete_media_item(%MediaItem{} = media_item) do + Repo.delete(media_item) + end + + @doc """ + Returns an `%Ecto.Changeset{}` for tracking media_item changes. + """ + def change_media_item(%MediaItem{} = media_item, attrs \\ %{}) do + MediaItem.changeset(media_item, attrs) + end +end diff --git a/lib/pinchflat/media/media_item.ex b/lib/pinchflat/media/media_item.ex new file mode 100644 index 0000000..5856a4b --- /dev/null +++ b/lib/pinchflat/media/media_item.ex @@ -0,0 +1,33 @@ +defmodule Pinchflat.Media.MediaItem do + @moduledoc """ + The MediaItem schema. + """ + + use Ecto.Schema + import Ecto.Changeset + + alias Pinchflat.MediaSource.Channel + + @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 + field :video_filepath, :string + + belongs_to :channel, Channel + + timestamps(type: :utc_datetime) + end + + @doc false + def changeset(media_item, attrs) do + media_item + |> cast(attrs, @allowed_fields) + |> validate_required(@required_fields) + |> unique_constraint([:media_id, :channel_id]) + end +end diff --git a/lib/pinchflat/media_client/backends/yt_dlp/channel.ex b/lib/pinchflat/media_client/backends/yt_dlp/channel.ex index 39d7348..b83093e 100644 --- a/lib/pinchflat/media_client/backends/yt_dlp/channel.ex +++ b/lib/pinchflat/media_client/backends/yt_dlp/channel.ex @@ -15,7 +15,7 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.Channel do Returns {:ok, %ChannelDetails{}} | {:error, any, ...}. """ - def get_channel_info(channel_url) do + def get_channel_details(channel_url) do opts = [print: "%(.{channel,channel_id})j", playlist_end: 1] with {:ok, output} <- backend_runner().run(channel_url, opts), diff --git a/lib/pinchflat/media_client/channel_details.ex b/lib/pinchflat/media_client/channel_details.ex index 3e952bf..021238b 100644 --- a/lib/pinchflat/media_client/channel_details.ex +++ b/lib/pinchflat/media_client/channel_details.ex @@ -21,7 +21,16 @@ defmodule Pinchflat.MediaClient.ChannelDetails do Returns {:ok, map()} | {:error, any, ...}. """ def get_channel_details(channel_url, backend \\ :yt_dlp) do - channel_module(backend).get_channel_info(channel_url) + channel_module(backend).get_channel_details(channel_url) + end + + @doc """ + Returns a list of video IDs for the given channel URL, using the given backend. + + Returns {:ok, list(binary())} | {:error, any, ...}. + """ + def get_video_ids(channel_url, backend \\ :yt_dlp) do + channel_module(backend).get_video_ids(channel_url) end defp channel_module(backend) do diff --git a/lib/pinchflat/media_source.ex b/lib/pinchflat/media_source.ex index 6b98bf7..44847e7 100644 --- a/lib/pinchflat/media_source.ex +++ b/lib/pinchflat/media_source.ex @@ -6,6 +6,9 @@ defmodule Pinchflat.MediaSource do import Ecto.Query, warn: false alias Pinchflat.Repo + alias Pinchflat.Tasks + alias Pinchflat.Media + alias Pinchflat.Tasks.ChannelTasks alias Pinchflat.MediaSource.Channel alias Pinchflat.MediaClient.ChannelDetails @@ -24,27 +27,61 @@ defmodule Pinchflat.MediaSource do def get_channel!(id), do: Repo.get!(Channel, id) @doc """ - Creates a channel. Returns {:ok, %Channel{}} | {:error, %Ecto.Changeset{}} + Creates a channel. May attempt to pull additional channel details from the + original_url (if provided). Will attempt to start indexing the channel's + media if successfully inserted. + + Returns {:ok, %Channel{}} | {:error, %Ecto.Changeset{}} """ def create_channel(attrs \\ %{}) do %Channel{} |> change_channel_from_url(attrs) - |> Repo.insert() + |> commit_and_start_indexing() end @doc """ - Updates a channel. Returns {:ok, %Channel{}} | {:error, %Ecto.Changeset{}} + Given a media source, creates (indexes) the media by creating media_items for each + media ID in the source. + + Returns [%MediaItem{}, ...] | [%Ecto.Changeset{}, ...] + """ + def index_media_items(%Channel{} = channel) do + {:ok, media_ids} = ChannelDetails.get_video_ids(channel.original_url) + + media_ids + |> Enum.map(fn media_id -> + attrs = %{channel_id: channel.id, media_id: media_id} + + case Media.create_media_item(attrs) do + {:ok, media_item} -> media_item + {:error, changeset} -> changeset + end + end) + end + + @doc """ + Updates a channel. May attempt to pull additional channel details from the + original_url (if changed). May attempt to start indexing the channel's + media if the indexing frequency has been changed. + + Existing indexing tasks will be cancelled if the indexing frequency has been + changed (logic in `ChannelTasks.kickoff_indexing_task`) + + Returns {:ok, %Channel{}} | {:error, %Ecto.Changeset{}} """ def update_channel(%Channel{} = channel, attrs) do channel |> change_channel_from_url(attrs) - |> Repo.update() + |> commit_and_start_indexing() end @doc """ - Deletes a channel. Returns {:ok, %Channel{}} | {:error, %Ecto.Changeset{}} + Deletes a channel and it's associated tasks (of any state). + + Returns {:ok, %Channel{}} | {:error, %Ecto.Changeset{}} """ def delete_channel(%Channel{} = channel) do + Tasks.delete_tasks_for(channel) Repo.delete(channel) end @@ -96,4 +133,31 @@ defmodule Pinchflat.MediaSource do ) end end + + defp commit_and_start_indexing(changeset) do + case Repo.insert_or_update(changeset) do + {:ok, %Channel{} = channel} -> + maybe_run_indexing_task(changeset, channel) + + {:ok, channel} + + err -> + err + end + end + + defp maybe_run_indexing_task(changeset, channel) do + case changeset.data do + # If the changeset is new (not persisted), start indexing no matter what + %{__meta__: %{state: :built}} -> + ChannelTasks.kickoff_indexing_task(channel) + + # If the record has been persisted, only run indexing if the + # indexing frequency has been changed + %{__meta__: %{state: :loaded}} -> + if Map.has_key?(changeset.changes, :index_frequency_minutes) do + ChannelTasks.kickoff_indexing_task(channel) + end + end + end end diff --git a/lib/pinchflat/media_source/channel.ex b/lib/pinchflat/media_source/channel.ex index 4861f65..e70030c 100644 --- a/lib/pinchflat/media_source/channel.ex +++ b/lib/pinchflat/media_source/channel.ex @@ -6,20 +6,24 @@ defmodule Pinchflat.MediaSource.Channel do use Ecto.Schema import Ecto.Changeset + alias Pinchflat.Media.MediaItem alias Pinchflat.Profiles.MediaProfile - @required_fields ~w(name channel_id original_url media_profile_id)a - @allowed_fields @required_fields + @allowed_fields ~w(name channel_id index_frequency_minutes original_url media_profile_id)a + @required_fields @allowed_fields -- ~w(index_frequency_minutes)a schema "channels" do field :name, :string field :channel_id, :string + field :index_frequency_minutes, :integer # This should only be used for user reference going forward # as the channel_id should be used for all API calls field :original_url, :string belongs_to :media_profile, MediaProfile + has_many :media_items, MediaItem + timestamps(type: :utc_datetime) end diff --git a/lib/pinchflat/tasks.ex b/lib/pinchflat/tasks.ex new file mode 100644 index 0000000..f19461c --- /dev/null +++ b/lib/pinchflat/tasks.ex @@ -0,0 +1,129 @@ +defmodule Pinchflat.Tasks do + @moduledoc """ + The Tasks context. + """ + + import Ecto.Query, warn: false + alias Pinchflat.Repo + + alias Pinchflat.Tasks.Task + alias Pinchflat.MediaSource.Channel + + @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 job states to include. + + Returns [%Task{}, ...] + """ + def list_tasks_for(attached_record_type, attached_record_id, job_states \\ Oban.Job.states()) do + stringified_states = Enum.map(job_states, &to_string/1) + + Repo.all( + from t in Task, + join: j in assoc(t, :job), + where: field(t, ^attached_record_type) == ^attached_record_id, + where: j.state in ^stringified_states + ) + end + + @doc """ + Returns the list of pending tasks for a given record type and ID. + + Returns [%Task{}, ...] + """ + def list_pending_tasks_for(attached_record_type, attached_record_id) do + list_tasks_for( + attached_record_type, + attached_record_id, + [: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. Returns {:ok, %Task{}} | {:error, %Ecto.Changeset{}}. + """ + def create_task(attrs \\ %{}) do + %Task{} + |> Task.changeset(attrs) + |> Repo.insert() + end + + # This one's function signature is designed to help simplify + # usage of `create_job_with_task/2` + def create_task(%Oban.Job{} = job, %Channel{} = channel) do + %Task{} + |> Task.changeset(%{job_id: job.id, channel_id: channel.id}) + |> Repo.insert() + end + + @doc """ + Creates a job from given attrs, creating a task with an attached record + if successful. + + Returns {:ok, %Task{}} | {:error, %Ecto.Changeset{}}. + """ + def create_job_with_task(job_attrs, task_attached_record) do + case Oban.insert(job_attrs) do + {:ok, job} -> create_task(job, task_attached_record) + 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. + + Returns :ok + """ + def delete_tasks_for(%Channel{} = channel) do + tasks = list_tasks_for(:channel_id, channel.id) + + Enum.each(tasks, fn task -> + delete_task(task) + end) + end + + @doc """ + Deletes all _pending_ tasks attached to a given record, cancelling any attached jobs. + + Returns :ok + """ + def delete_pending_tasks_for(%Channel{} = channel) do + tasks = list_pending_tasks_for(:channel_id, channel.id) + + Enum.each(tasks, fn task -> + delete_task(task) + end) + end + + @doc """ + Returns an `%Ecto.Changeset{}` for tracking task changes. + """ + def change_task(%Task{} = task, attrs \\ %{}) do + Task.changeset(task, attrs) + end +end diff --git a/lib/pinchflat/tasks/channel_tasks.ex b/lib/pinchflat/tasks/channel_tasks.ex new file mode 100644 index 0000000..095c90f --- /dev/null +++ b/lib/pinchflat/tasks/channel_tasks.ex @@ -0,0 +1,26 @@ +defmodule Pinchflat.Tasks.ChannelTasks do + @moduledoc """ + This module contains methods for managing tasks (workers) related to channels. + """ + + alias Pinchflat.Tasks + alias Pinchflat.MediaSource.Channel + alias Pinchflat.Workers.MediaIndexingWorker + + @doc """ + 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) + + if channel.index_frequency_minutes <= 0 do + {:ok, :should_not_index} + else + channel + |> Map.take([:id]) + # Schedule this one immediately, but future ones will be on an interval + |> MediaIndexingWorker.new() + |> Tasks.create_job_with_task(channel) + end + end +end diff --git a/lib/pinchflat/tasks/task.ex b/lib/pinchflat/tasks/task.ex new file mode 100644 index 0000000..4184ee8 --- /dev/null +++ b/lib/pinchflat/tasks/task.ex @@ -0,0 +1,24 @@ +defmodule Pinchflat.Tasks.Task do + @moduledoc """ + The Task schema. + """ + + use Ecto.Schema + import Ecto.Changeset + + alias Pinchflat.MediaSource.Channel + + schema "tasks" do + belongs_to :job, Oban.Job + belongs_to :channel, Channel + + timestamps(type: :utc_datetime) + end + + @doc false + def changeset(task, attrs) do + task + |> cast(attrs, [:job_id, :channel_id]) + |> validate_required([:job_id]) + end +end diff --git a/lib/pinchflat/workers/media_indexing_worker.ex b/lib/pinchflat/workers/media_indexing_worker.ex new file mode 100644 index 0000000..69c14d3 --- /dev/null +++ b/lib/pinchflat/workers/media_indexing_worker.ex @@ -0,0 +1,48 @@ +defmodule Pinchflat.Workers.MediaIndexingWorker do + @moduledoc false + + use Oban.Worker, + queue: :media_indexing, + unique: [period: :infinity, states: [:available, :scheduled, :retryable]], + tags: ["media_source", "media_indexing"] + + alias __MODULE__ + alias Pinchflat.Tasks + alias Pinchflat.MediaSource + + @impl Oban.Worker + @doc """ + The ID is that of a channel _record_, not a YouTube channel ID. + + NOTE: 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: + + - Benefit: No chance for jobs to overlap if a job takes longer than the + scheduled interval. Less likely to hit API rate limits. + - Side effect: Intervals are "soft" and _always_ walk forward. This may cause + user confusion since a 30-minute job scheduled for every hour will + 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. + """ + def perform(%Oban.Job{args: %{"id" => channel_id}}) do + channel = MediaSource.get_channel!(channel_id) + + if channel.index_frequency_minutes <= 0 do + :ok + else + index_media_and_reschedule(channel) + end + end + + defp index_media_and_reschedule(channel) do + MediaSource.index_media_items(channel) + + channel + |> Map.take([:id]) + |> MediaIndexingWorker.new(schedule_in: channel.index_frequency_minutes * 60) + |> Tasks.create_job_with_task(channel) + end +end diff --git a/lib/pinchflat_web/controllers/media_sources/channel_html.ex b/lib/pinchflat_web/controllers/media_sources/channel_html.ex index 37bc279..0feb191 100644 --- a/lib/pinchflat_web/controllers/media_sources/channel_html.ex +++ b/lib/pinchflat_web/controllers/media_sources/channel_html.ex @@ -11,4 +11,17 @@ defmodule PinchflatWeb.MediaSources.ChannelHTML do attr :media_profiles, :list, required: true def channel_form(assigns) + + def friendly_index_frequencies do + [ + {"Never", -1}, + {"1 Hour", 60}, + {"3 Hours", 3 * 60}, + {"6 Hours", 6 * 60}, + {"12 Hours", 12 * 60}, + {"Daily (recommended)", 24 * 60}, + {"Weekly", 7 * 24 * 60}, + {"Monthly", 30 * 24 * 60} + ] + end end diff --git a/lib/pinchflat_web/controllers/media_sources/channel_html/channel_form.html.heex b/lib/pinchflat_web/controllers/media_sources/channel_html/channel_form.html.heex index 00ad253..b8b4d4e 100644 --- a/lib/pinchflat_web/controllers/media_sources/channel_html/channel_form.html.heex +++ b/lib/pinchflat_web/controllers/media_sources/channel_html/channel_form.html.heex @@ -12,6 +12,13 @@ <.input field={f[:original_url]} type="text" label="Channel URL" /> + <.input + field={f[:index_frequency_minutes]} + options={friendly_index_frequencies()} + type="select" + label="Index Frequency" + /> + <:actions> <.button>Save Channel diff --git a/mix.exs b/mix.exs index cc91b60..d89ada4 100644 --- a/mix.exs +++ b/mix.exs @@ -51,9 +51,11 @@ defmodule Pinchflat.MixProject do {:jason, "~> 1.2"}, {:dns_cluster, "~> 0.1.1"}, {:plug_cowboy, "~> 2.5"}, + {:oban, "~> 2.16"}, + {:nimble_parsec, "~> 1.4"}, {:mox, "~> 1.0", only: :test}, {:credo, "~> 1.7", only: [:dev, :test], runtime: false}, - {:nimble_parsec, "~> 1.4"} + {:faker, "~> 0.17", only: :test} ] end diff --git a/mix.lock b/mix.lock index 2ea91e9..157a94f 100644 --- a/mix.lock +++ b/mix.lock @@ -12,6 +12,7 @@ "ecto_sql": {:hex, :ecto_sql, "3.11.1", "e9abf28ae27ef3916b43545f9578b4750956ccea444853606472089e7d169470", [:mix], [{:db_connection, "~> 2.4.1 or ~> 2.5", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.11.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.6.0", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.16.0 or ~> 0.17.0 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "ce14063ab3514424276e7e360108ad6c2308f6d88164a076aac8a387e1fea634"}, "esbuild": {:hex, :esbuild, "0.8.1", "0cbf919f0eccb136d2eeef0df49c4acf55336de864e63594adcea3814f3edf41", [:mix], [{:castore, ">= 0.0.0", [hex: :castore, repo: "hexpm", optional: false]}, {:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "25fc876a67c13cb0a776e7b5d7974851556baeda2085296c14ab48555ea7560f"}, "expo": {:hex, :expo, "0.5.1", "249e826a897cac48f591deba863b26c16682b43711dd15ee86b92f25eafd96d9", [:mix], [], "hexpm", "68a4233b0658a3d12ee00d27d37d856b1ba48607e7ce20fd376958d0ba6ce92b"}, + "faker": {:hex, :faker, "0.17.0", "671019d0652f63aefd8723b72167ecdb284baf7d47ad3a82a15e9b8a6df5d1fa", [:mix], [], "hexpm", "a7d4ad84a93fd25c5f5303510753789fc2433ff241bf3b4144d3f6f291658a6a"}, "file_system": {:hex, :file_system, "0.2.10", "fb082005a9cd1711c05b5248710f8826b02d7d1784e7c3451f9c1231d4fc162d", [:mix], [], "hexpm", "41195edbfb562a593726eda3b3e8b103a309b733ad25f3d642ba49696bf715dc"}, "finch": {:hex, :finch, "0.17.0", "17d06e1d44d891d20dbd437335eebe844e2426a0cd7e3a3e220b461127c73f70", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: false]}, {:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.3", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 0.2.6 or ~> 1.0", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "8d014a661bb6a437263d4b5abf0bcbd3cf0deb26b1e8596f2a271d22e48934c7"}, "floki": {:hex, :floki, "0.35.2", "87f8c75ed8654b9635b311774308b2760b47e9a579dabf2e4d5f1e1d42c39e0b", [:mix], [], "hexpm", "6b05289a8e9eac475f644f09c2e4ba7e19201fd002b89c28c1293e7bd16773d9"}, @@ -24,6 +25,7 @@ "nimble_options": {:hex, :nimble_options, "1.1.0", "3b31a57ede9cb1502071fade751ab0c7b8dbe75a9a4c2b5bbb0943a690b63172", [:mix], [], "hexpm", "8bbbb3941af3ca9acc7835f5655ea062111c9c27bcac53e004460dfd19008a99"}, "nimble_parsec": {:hex, :nimble_parsec, "1.4.0", "51f9b613ea62cfa97b25ccc2c1b4216e81df970acd8e16e8d1bdc58fef21370d", [:mix], [], "hexpm", "9c565862810fb383e9838c1dd2d7d2c437b3d13b267414ba6af33e50d2d1cf28"}, "nimble_pool": {:hex, :nimble_pool, "1.0.0", "5eb82705d138f4dd4423f69ceb19ac667b3b492ae570c9f5c900bb3d2f50a847", [:mix], [], "hexpm", "80be3b882d2d351882256087078e1b1952a28bf98d0a287be87e4a24a710b67a"}, + "oban": {:hex, :oban, "2.17.3", "ddfd5710aadcd550d2e174c8d73ce5f1865601418cf54a91775f20443fb832b7", [:mix], [{:ecto_sql, "~> 3.6", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:ecto_sqlite3, "~> 0.9", [hex: :ecto_sqlite3, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.16", [hex: :postgrex, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "452eada8bfe0d0fefd0740ab5fa8cf3ef6c375df0b4a3c3805d179022a04738a"}, "phoenix": {:hex, :phoenix, "1.7.10", "02189140a61b2ce85bb633a9b6fd02dff705a5f1596869547aeb2b2b95edd729", [:mix], [{:castore, ">= 0.0.0", [hex: :castore, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.1", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: true]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.6", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:plug_crypto, "~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:websock_adapter, "~> 0.5.3", [hex: :websock_adapter, repo: "hexpm", optional: false]}], "hexpm", "cf784932e010fd736d656d7fead6a584a4498efefe5b8227e9f383bf15bb79d0"}, "phoenix_ecto": {:hex, :phoenix_ecto, "4.4.3", "86e9878f833829c3f66da03d75254c155d91d72a201eb56ae83482328dc7ca93", [:mix], [{:ecto, "~> 3.5", [hex: :ecto, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 2.14.2 or ~> 3.0 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:plug, "~> 1.9", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "d36c401206f3011fefd63d04e8ef626ec8791975d9d107f9a0817d426f61ac07"}, "phoenix_html": {:hex, :phoenix_html, "3.3.3", "380b8fb45912b5638d2f1d925a3771b4516b9a78587249cabe394e0a5d579dc9", [:mix], [{:plug, "~> 1.5", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "923ebe6fec6e2e3b3e569dfbdc6560de932cd54b000ada0208b5f45024bdd76c"}, diff --git a/priv/repo/migrations/20240125025325_create_media_items.exs b/priv/repo/migrations/20240125025325_create_media_items.exs new file mode 100644 index 0000000..cfaa25f --- /dev/null +++ b/priv/repo/migrations/20240125025325_create_media_items.exs @@ -0,0 +1,17 @@ +defmodule Pinchflat.Repo.Migrations.CreateMediaItems do + use Ecto.Migration + + def change do + create table(:media_items) do + add :media_id, :string, null: false + add :title, :string + add :video_filepath, :string + add :channel_id, references(:channels, on_delete: :restrict), null: false + + timestamps(type: :utc_datetime) + end + + create index(:media_items, [:channel_id]) + create unique_index(:media_items, [:media_id, :channel_id]) + end +end diff --git a/priv/repo/migrations/20240125043813_add_oban_jobs_table.exs b/priv/repo/migrations/20240125043813_add_oban_jobs_table.exs new file mode 100644 index 0000000..8044fc0 --- /dev/null +++ b/priv/repo/migrations/20240125043813_add_oban_jobs_table.exs @@ -0,0 +1,13 @@ +defmodule Pinchflat.Repo.Migrations.AddObanJobsTable do + use Ecto.Migration + + def up do + Oban.Migration.up(version: 11) + end + + # We specify `version: 1` in `down`, ensuring that we'll roll all the way back down if + # necessary, regardless of which version we've migrated `up` to. + def down do + Oban.Migration.down(version: 1) + end +end diff --git a/priv/repo/migrations/20240125165519_add_index_frequency_to_channels.exs b/priv/repo/migrations/20240125165519_add_index_frequency_to_channels.exs new file mode 100644 index 0000000..c03b25b --- /dev/null +++ b/priv/repo/migrations/20240125165519_add_index_frequency_to_channels.exs @@ -0,0 +1,9 @@ +defmodule Pinchflat.Repo.Migrations.AddIndexFrequencyToChannels do + use Ecto.Migration + + def change do + alter table(:channels) do + add :index_frequency_minutes, :integer, default: 60 * 24, null: false + end + end +end diff --git a/priv/repo/migrations/20240125212753_create_tasks.exs b/priv/repo/migrations/20240125212753_create_tasks.exs new file mode 100644 index 0000000..7a7b5e3 --- /dev/null +++ b/priv/repo/migrations/20240125212753_create_tasks.exs @@ -0,0 +1,16 @@ +defmodule Pinchflat.Repo.Migrations.CreateTasks do + use Ecto.Migration + + def change do + create table(:tasks) do + add :job_id, references(:oban_jobs, on_delete: :delete_all), null: false + # `restrict` because we need to be sure to delete pending tasks when a channel is deleted + add :channel_id, references(:channels, on_delete: :restrict), null: true + + timestamps(type: :utc_datetime) + end + + create index(:tasks, [:job_id]) + create index(:tasks, [:channel_id]) + end +end diff --git a/test/pinchflat/media_client/backends/yt_dlp/channel_test.exs b/test/pinchflat/media_client/backends/yt_dlp/channel_test.exs index 3e12a6c..14cc187 100644 --- a/test/pinchflat/media_client/backends/yt_dlp/channel_test.exs +++ b/test/pinchflat/media_client/backends/yt_dlp/channel_test.exs @@ -9,13 +9,13 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.ChannelTest do setup :verify_on_exit! - describe "get_channel_info/1" do + describe "get_channel_details/1" do test "it returns a %ChannelDetails{} with data on success" do expect(YtDlpRunnerMock, :run, fn _url, _opts -> {:ok, "{\"channel\": \"TheUselessTrials\", \"channel_id\": \"UCQH2\"}"} end) - assert {:ok, res} = Channel.get_channel_info(@channel_url) + assert {:ok, res} = Channel.get_channel_details(@channel_url) assert %ChannelDetails{id: "UCQH2", name: "TheUselessTrials"} = res end @@ -26,19 +26,19 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.ChannelTest do {:ok, "{}"} end) - assert {:ok, _} = Channel.get_channel_info(@channel_url) + assert {:ok, _} = Channel.get_channel_details(@channel_url) end test "it returns an error if the runner returns an error" do expect(YtDlpRunnerMock, :run, fn _url, _opts -> {:error, "Big issue", 1} end) - assert {:error, "Big issue", 1} = Channel.get_channel_info(@channel_url) + 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) - assert {:error, %Jason.DecodeError{}} = Channel.get_channel_info(@channel_url) + assert {:error, %Jason.DecodeError{}} = Channel.get_channel_details(@channel_url) end end end diff --git a/test/pinchflat/media_client/channel_details_test.exs b/test/pinchflat/media_client/channel_details_test.exs index 012161e..f6c2707 100644 --- a/test/pinchflat/media_client/channel_details_test.exs +++ b/test/pinchflat/media_client/channel_details_test.exs @@ -35,4 +35,24 @@ defmodule Pinchflat.MediaClient.ChannelDetailsTest do assert %ChannelDetails{id: "UCQH2", name: "TheUselessTrials"} = res end end + + 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}] + + {:ok, ""} + end) + + assert {:ok, _} = ChannelDetails.get_video_ids(@channel_url) + end + + test "it returns a list of strings" do + expect(YtDlpRunnerMock, :run, fn _url, _opts -> + {:ok, "video1\nvideo2\nvideo3"} + end) + + assert {:ok, ["video1", "video2", "video3"]} = ChannelDetails.get_video_ids(@channel_url) + end + end end diff --git a/test/pinchflat/media_source_test.exs b/test/pinchflat/media_source_test.exs index 4fcc8a7..3956186 100644 --- a/test/pinchflat/media_source_test.exs +++ b/test/pinchflat/media_source_test.exs @@ -1,13 +1,15 @@ defmodule Pinchflat.MediaSourceTest do use Pinchflat.DataCase import Mox - - alias Pinchflat.MediaSource - alias Pinchflat.MediaSource.Channel - + import Pinchflat.TasksFixtures import Pinchflat.ProfilesFixtures import Pinchflat.MediaSourceFixtures + alias Pinchflat.MediaSource + alias Pinchflat.Media.MediaItem + alias Pinchflat.MediaSource.Channel + alias Pinchflat.Workers.MediaIndexingWorker + @invalid_channel_attrs %{name: nil, channel_id: nil} setup :verify_on_exit! @@ -82,6 +84,72 @@ defmodule Pinchflat.MediaSourceTest do assert {:ok, %Channel{}} = MediaSource.create_channel(channel_1_attrs) assert {:ok, %Channel{}} = MediaSource.create_channel(channel_2_attrs) end + + test "creation will schedule the indexing task" do + expect(YtDlpRunnerMock, :run, &runner_function_mock/2) + + valid_attrs = %{ + media_profile_id: media_profile_fixture().id, + original_url: "https://www.youtube.com/channel/abc123" + } + + assert {:ok, %Channel{} = channel} = MediaSource.create_channel(valid_attrs) + + assert_enqueued(worker: MediaIndexingWorker, args: %{"id" => channel.id}) + end + end + + describe "index_media_items/1" do + setup do + stub(YtDlpRunnerMock, :run, fn _url, _opts -> {:ok, "video1\nvideo2\nvideo3"} end) + + {:ok, [channel: channel_fixture()]} + end + + test "it creates a media_item record for each media ID returned", %{channel: channel} do + assert media_items = MediaSource.index_media_items(channel) + + assert Enum.count(media_items) == 3 + assert ["video1", "video2", "video3"] == Enum.map(media_items, & &1.media_id) + assert Enum.all?(media_items, fn %MediaItem{} -> true end) + end + + test "it attaches all media_items to the given channel", %{channel: channel} do + channel_id = channel.id + assert media_items = MediaSource.index_media_items(channel) + + assert Enum.count(media_items) == 3 + assert Enum.all?(media_items, fn %MediaItem{channel_id: ^channel_id} -> true end) + end + + test "it won't duplicate media_items based on media_id and channel", %{channel: channel} do + _first_run = MediaSource.index_media_items(channel) + _duplicate_run = MediaSource.index_media_items(channel) + + media_items = Repo.preload(channel, :media_items).media_items + assert Enum.count(media_items) == 3 + end + + test "it can duplicate media_ids for different channels", %{channel: channel} do + other_channel = channel_fixture() + + media_items = MediaSource.index_media_items(channel) + media_items_other_channel = MediaSource.index_media_items(other_channel) + + assert Enum.count(media_items) == 3 + assert Enum.count(media_items_other_channel) == 3 + + assert Enum.map(media_items, & &1.media_id) == + Enum.map(media_items_other_channel, & &1.media_id) + end + + test "it returns a list of media_items or changesets", %{channel: channel} do + first_run = MediaSource.index_media_items(channel) + duplicate_run = MediaSource.index_media_items(channel) + + assert Enum.all?(first_run, fn %MediaItem{} -> true end) + assert Enum.all?(duplicate_run, fn %Ecto.Changeset{} -> true end) + end end describe "update_channel/2" do @@ -113,6 +181,23 @@ defmodule Pinchflat.MediaSourceTest do assert {:ok, %Channel{}} = MediaSource.update_channel(channel, update_attrs) end + test "updating the index frequency will re-schedule the indexing task" do + channel = channel_fixture() + update_attrs = %{index_frequency_minutes: 123} + + assert {:ok, %Channel{} = channel} = MediaSource.update_channel(channel, update_attrs) + assert channel.index_frequency_minutes == 123 + assert_enqueued(worker: MediaIndexingWorker, args: %{"id" => channel.id}) + end + + test "not updating the index frequency will not re-schedule the indexing task" do + channel = channel_fixture() + update_attrs = %{name: "some updated name"} + + assert {:ok, %Channel{}} = MediaSource.update_channel(channel, update_attrs) + refute_enqueued(worker: MediaIndexingWorker, args: %{"id" => channel.id}) + end + test "updates with invalid data returns error changeset" do channel = channel_fixture() @@ -134,6 +219,14 @@ defmodule Pinchflat.MediaSourceTest do channel = channel_fixture() assert %Ecto.Changeset{} = MediaSource.change_channel(channel) end + + test "deletion also deletes all associated tasks" do + channel = channel_fixture() + task = task_fixture(channel_id: channel.id) + + assert {:ok, %Channel{}} = MediaSource.delete_channel(channel) + assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end + end end describe "change_channel/2" do diff --git a/test/pinchflat/media_test.exs b/test/pinchflat/media_test.exs new file mode 100644 index 0000000..405ad4d --- /dev/null +++ b/test/pinchflat/media_test.exs @@ -0,0 +1,84 @@ +defmodule Pinchflat.MediaTest do + use Pinchflat.DataCase + + import Pinchflat.MediaFixtures + import Pinchflat.MediaSourceFixtures + + alias Pinchflat.Media + alias Pinchflat.Media.MediaItem + + @invalid_attrs %{title: nil, media_id: nil, video_filepath: nil} + + describe "list_media_items/0" do + test "it returns all media_items" do + media_item = media_item_fixture() + assert Media.list_media_items() == [media_item] + end + end + + describe "get_media_item!/1" do + test "it returns the media_item with given id" do + media_item = media_item_fixture() + assert Media.get_media_item!(media_item.id) == media_item + end + end + + describe "create_media_item/1" do + test "creating with valid data creates a media_item" do + valid_attrs = %{ + media_id: Faker.String.base64(12), + title: Faker.Commerce.product_name(), + video_filepath: "/video/#{Faker.File.file_name(:video)}", + channel_id: channel_fixture().id + } + + assert {:ok, %MediaItem{} = media_item} = Media.create_media_item(valid_attrs) + assert media_item.title == valid_attrs.title + assert media_item.media_id == valid_attrs.media_id + assert media_item.video_filepath == valid_attrs.video_filepath + end + + test "creating with invalid data returns error changeset" do + assert {:error, %Ecto.Changeset{}} = Media.create_media_item(@invalid_attrs) + end + end + + describe "update_media_item/2" do + test "updating with valid data updates the media_item" do + media_item = media_item_fixture() + + update_attrs = %{ + media_id: Faker.String.base64(12), + title: Faker.Commerce.product_name(), + video_filepath: "/video/#{Faker.File.file_name(:video)}", + channel_id: channel_fixture().id + } + + assert {:ok, %MediaItem{} = media_item} = Media.update_media_item(media_item, update_attrs) + assert media_item.title == update_attrs.title + assert media_item.media_id == update_attrs.media_id + assert media_item.video_filepath == update_attrs.video_filepath + end + + test "updating with invalid data returns error changeset" do + media_item = media_item_fixture() + assert {:error, %Ecto.Changeset{}} = Media.update_media_item(media_item, @invalid_attrs) + assert media_item == Media.get_media_item!(media_item.id) + end + end + + describe "delete_media_item/1" do + test "deletion deletes the media_item" do + media_item = media_item_fixture() + assert {:ok, %MediaItem{}} = Media.delete_media_item(media_item) + assert_raise Ecto.NoResultsError, fn -> Media.get_media_item!(media_item.id) end + end + end + + describe "change_media_item/1" do + test "change_media_item/1 returns a media_item changeset" do + media_item = media_item_fixture() + assert %Ecto.Changeset{} = Media.change_media_item(media_item) + end + end +end diff --git a/test/pinchflat/profiles_test.exs b/test/pinchflat/profiles_test.exs index 9f54b39..8635750 100644 --- a/test/pinchflat/profiles_test.exs +++ b/test/pinchflat/profiles_test.exs @@ -2,25 +2,27 @@ defmodule Pinchflat.ProfilesTest do use Pinchflat.DataCase alias Pinchflat.Profiles + alias Pinchflat.Profiles.MediaProfile + import Pinchflat.ProfilesFixtures - describe "media_profiles" do - alias Pinchflat.Profiles.MediaProfile + @invalid_attrs %{name: nil, output_path_template: nil} - import Pinchflat.ProfilesFixtures - - @invalid_attrs %{name: nil, output_path_template: nil} - - test "list_media_profiles/0 returns all media_profiles" do + describe "list_media_profiles/0" do + test "it returns all media_profiles" do media_profile = media_profile_fixture() assert Profiles.list_media_profiles() == [media_profile] end + end - test "get_media_profile!/1 returns the media_profile with given id" do + describe "get_media_profile!/1" do + test "it returns the media_profile with given id" do media_profile = media_profile_fixture() assert Profiles.get_media_profile!(media_profile.id) == media_profile end + end - test "create_media_profile/1 with valid data creates a media_profile" do + describe "create_media_profile/1" do + test "creation with valid data creates a media_profile" do valid_attrs = %{name: "some name", output_path_template: "some output_path_template"} assert {:ok, %MediaProfile{} = media_profile} = Profiles.create_media_profile(valid_attrs) @@ -28,11 +30,13 @@ defmodule Pinchflat.ProfilesTest do assert media_profile.output_path_template == "some output_path_template" end - test "create_media_profile/1 with invalid data returns error changeset" do + test "creation with invalid data returns error changeset" do assert {:error, %Ecto.Changeset{}} = Profiles.create_media_profile(@invalid_attrs) end + end - test "update_media_profile/2 with valid data updates the media_profile" do + describe "update_media_profile/2" do + test "updating with valid data updates the media_profile" do media_profile = media_profile_fixture() update_attrs = %{ @@ -47,7 +51,7 @@ defmodule Pinchflat.ProfilesTest do assert media_profile.output_path_template == "some updated output_path_template" end - test "update_media_profile/2 with invalid data returns error changeset" do + test "updating with invalid data returns error changeset" do media_profile = media_profile_fixture() assert {:error, %Ecto.Changeset{}} = @@ -55,14 +59,18 @@ defmodule Pinchflat.ProfilesTest do assert media_profile == Profiles.get_media_profile!(media_profile.id) end + end - test "delete_media_profile/1 deletes the media_profile" do + describe "delete_media_profile/1" do + test "deletion deletes the media_profile" do media_profile = media_profile_fixture() assert {:ok, %MediaProfile{}} = Profiles.delete_media_profile(media_profile) assert_raise Ecto.NoResultsError, fn -> Profiles.get_media_profile!(media_profile.id) end end + end - test "change_media_profile/1 returns a media_profile changeset" do + describe "change_media_profile/1" do + test "it returns a media_profile changeset" do media_profile = media_profile_fixture() assert %Ecto.Changeset{} = Profiles.change_media_profile(media_profile) end diff --git a/test/pinchflat/tasks/channel_tasks_test.exs b/test/pinchflat/tasks/channel_tasks_test.exs new file mode 100644 index 0000000..2314523 --- /dev/null +++ b/test/pinchflat/tasks/channel_tasks_test.exs @@ -0,0 +1,45 @@ +defmodule Pinchflat.Tasks.ChannelTasksTest do + use Pinchflat.DataCase + + import Pinchflat.TasksFixtures + import Pinchflat.MediaSourceFixtures + + alias Pinchflat.Tasks.Task + alias Pinchflat.Tasks.ChannelTasks + alias Pinchflat.Workers.MediaIndexingWorker + + describe "kickoff_indexing_task/1" do + test "it does not schedule a job if the interval is <= 0" do + channel = channel_fixture(index_frequency_minutes: -1) + + assert {:ok, :should_not_index} = ChannelTasks.kickoff_indexing_task(channel) + + refute_enqueued(worker: MediaIndexingWorker, args: %{"id" => channel.id}) + end + + test "it schedules a job if the interval is > 0" do + channel = channel_fixture(index_frequency_minutes: 1) + + assert {:ok, _} = ChannelTasks.kickoff_indexing_task(channel) + + assert_enqueued(worker: MediaIndexingWorker, args: %{"id" => channel.id}) + end + + test "it creates and attaches a task if the interval is > 0" do + channel = channel_fixture(index_frequency_minutes: 1) + + assert {:ok, %Task{} = task} = ChannelTasks.kickoff_indexing_task(channel) + + assert task.channel_id == channel.id + end + + test "it deletes any pending tasks for the channel" do + channel = channel_fixture() + task = task_fixture(channel_id: channel.id) + + assert {:ok, _} = ChannelTasks.kickoff_indexing_task(channel) + + assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end + end + end +end diff --git a/test/pinchflat/tasks_test.exs b/test/pinchflat/tasks_test.exs new file mode 100644 index 0000000..9228c7d --- /dev/null +++ b/test/pinchflat/tasks_test.exs @@ -0,0 +1,167 @@ +defmodule Pinchflat.TasksTest do + use Pinchflat.DataCase + import Pinchflat.JobFixtures + import Pinchflat.TasksFixtures + import Pinchflat.MediaSourceFixtures + + alias Pinchflat.Tasks + alias Pinchflat.Tasks.Task + alias Pinchflat.JobFixtures.TestJobWorker + + @invalid_attrs %{job_id: nil} + + describe "schema" do + test "it deletes a task when the job gets deleted" do + task = Repo.preload(task_fixture(), [:job]) + + {:ok, _} = Repo.delete(task.job) + + assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end + end + + test "it does not delete the other record when a job gets deleted" do + task = Repo.preload(task_fixture(), [:channel, :job]) + + {:ok, _} = Repo.delete(task.job) + + assert Repo.reload!(task.channel) + end + end + + describe "list_tasks/0" do + test "it returns all tasks" do + task = task_fixture() + assert Tasks.list_tasks() == [task] + end + end + + describe "list_tasks_for/3" do + test "it lets you specify which record type/ID to join on" do + task = task_fixture() + + assert Tasks.list_tasks_for(:channel_id, task.channel_id) == [task] + end + + test "it lets you specify which job states to include" do + task = task_fixture() + + assert Tasks.list_tasks_for(:channel_id, task.channel_id, [:available]) == [task] + assert Tasks.list_tasks_for(:channel_id, task.channel_id, [:cancelled]) == [] + end + end + + describe "list_pending_tasks_for/2" do + test "it lists pending tasks" do + task = task_fixture() + + assert Tasks.list_pending_tasks_for(:channel_id, task.channel_id) == [task] + end + + test "it does not list non-pending tasks" do + task = Repo.preload(task_fixture(), :job) + :ok = Oban.cancel_job(task.job) + + assert Tasks.list_pending_tasks_for(:channel_id, task.channel_id) == [] + end + end + + describe "get_task!/1" do + test "it returns the task with given id" do + task = task_fixture() + assert Tasks.get_task!(task.id) == task + end + end + + describe "create_task/1" do + test "creation with valid data creates a task" do + valid_attrs = %{job_id: job_fixture().id} + + assert {:ok, %Task{} = _task} = Tasks.create_task(valid_attrs) + end + + test "creation with invalid data returns error changeset" do + assert {:error, %Ecto.Changeset{}} = Tasks.create_task(@invalid_attrs) + end + + test "accepts a job and channel" do + job = job_fixture() + channel = channel_fixture() + + assert {:ok, %Task{} = task} = Tasks.create_task(job, channel) + + assert task.job_id == job.id + assert task.channel_id == channel.id + end + end + + describe "create_job_with_task/2" do + test "it enqueues the given job" do + channel = channel_fixture() + + refute_enqueued(worker: TestJobWorker) + assert {:ok, %Task{}} = Tasks.create_job_with_task(TestJobWorker.new(%{}), channel) + assert_enqueued(worker: TestJobWorker) + end + + test "it creates a task record if successful" do + channel = channel_fixture() + + assert {:ok, %Task{} = task} = Tasks.create_job_with_task(TestJobWorker.new(%{}), channel) + + assert task.channel_id == channel.id + end + end + + describe "delete_task/1" do + test "deletion deletes the task" do + task = task_fixture() + assert {:ok, %Task{}} = Tasks.delete_task(task) + assert_raise Ecto.NoResultsError, fn -> Tasks.get_task!(task.id) end + end + + test "deletion also cancels the attached job" do + task = Repo.preload(task_fixture(), :job) + + assert {:ok, %Task{}} = Tasks.delete_task(task) + job = Repo.reload!(task.job) + + assert job.state == "cancelled" + end + end + + describe "delete_tasks_for/1" do + test "it deletes tasks attached to a channel" do + channel = channel_fixture() + task = task_fixture(channel_id: channel.id) + + assert :ok = Tasks.delete_tasks_for(channel) + assert_raise Ecto.NoResultsError, fn -> Tasks.get_task!(task.id) end + end + end + + describe "delete_pending_tasks_for/1" do + test "it deletes pending tasks attached to a channel" do + channel = channel_fixture() + task = task_fixture(channel_id: channel.id) + + assert :ok = Tasks.delete_pending_tasks_for(channel) + assert_raise Ecto.NoResultsError, fn -> Tasks.get_task!(task.id) end + end + + test "it does not delete non-pending tasks" do + channel = channel_fixture() + task = Repo.preload(task_fixture(channel_id: channel.id), :job) + :ok = Oban.cancel_job(task.job) + + assert :ok = Tasks.delete_pending_tasks_for(channel) + assert Tasks.get_task!(task.id) + end + end + + describe "change_task/1" do + test "it returns a task changeset" do + task = task_fixture() + assert %Ecto.Changeset{} = Tasks.change_task(task) + end + end +end diff --git a/test/pinchflat/workers/media_indexing_worker_test.exs b/test/pinchflat/workers/media_indexing_worker_test.exs new file mode 100644 index 0000000..625d277 --- /dev/null +++ b/test/pinchflat/workers/media_indexing_worker_test.exs @@ -0,0 +1,79 @@ +defmodule Pinchflat.Workers.MediaIndexingWorkerTest do + use Pinchflat.DataCase + + import Mox + import Pinchflat.MediaSourceFixtures + + alias Pinchflat.Tasks + alias Pinchflat.Workers.MediaIndexingWorker + + 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) + + channel = channel_fixture(index_frequency_minutes: -1) + + perform_job(MediaIndexingWorker, %{id: channel.id}) + end + + test "it does not reschedule if the channel shouldn't be indexed" do + expect(YtDlpRunnerMock, :run, 0, fn _url, _opts -> {:ok, ""} end) + + channel = channel_fixture(index_frequency_minutes: -1) + perform_job(MediaIndexingWorker, %{id: channel.id}) + + refute_enqueued(worker: MediaIndexingWorker, args: %{"id" => channel.id}) + end + + test "it indexes the channel if it should be indexed" do + expect(YtDlpRunnerMock, :run, 1, fn _url, _opts -> {:ok, ""} end) + + channel = channel_fixture(index_frequency_minutes: 10) + + perform_job(MediaIndexingWorker, %{id: channel.id}) + end + + test "it reschedules the job based on the index frequency" do + expect(YtDlpRunnerMock, :run, 1, fn _url, _opts -> {:ok, ""} end) + + channel = channel_fixture(index_frequency_minutes: 10) + perform_job(MediaIndexingWorker, %{id: channel.id}) + + assert_enqueued( + worker: MediaIndexingWorker, + args: %{"id" => channel.id}, + scheduled_at: now_plus(channel.index_frequency_minutes, :minutes) + ) + end + + test "it creates a task for the rescheduled job" do + expect(YtDlpRunnerMock, :run, 1, fn _url, _opts -> {:ok, ""} end) + + channel = channel_fixture(index_frequency_minutes: 10) + task_count_fetcher = fn -> Enum.count(Tasks.list_tasks()) end + + assert_changed([from: 0, to: 1], task_count_fetcher, fn -> + perform_job(MediaIndexingWorker, %{id: channel.id}) + end) + end + + test "it creates the basic media_item records" do + expect(YtDlpRunnerMock, :run, 1, fn _url, _opts -> {:ok, "video1\nvideo2"} end) + + channel = channel_fixture(index_frequency_minutes: 10) + + media_item_fetcher = fn -> + channel + |> Repo.preload(:media_items) + |> Map.get(:media_items) + |> Enum.map(fn media_item -> media_item.media_id end) + end + + assert_changed([from: [], to: ["video1", "video2"]], media_item_fetcher, fn -> + perform_job(MediaIndexingWorker, %{id: channel.id}) + end) + end + end +end diff --git a/test/support/conn_case.ex b/test/support/conn_case.ex index 012ef2f..ff478a0 100644 --- a/test/support/conn_case.ex +++ b/test/support/conn_case.ex @@ -28,6 +28,7 @@ defmodule PinchflatWeb.ConnCase do import Plug.Conn import Phoenix.ConnTest import PinchflatWeb.ConnCase + import Pinchflat.TestingHelperMethods end end diff --git a/test/support/data_case.ex b/test/support/data_case.ex index df05c53..d217749 100644 --- a/test/support/data_case.ex +++ b/test/support/data_case.ex @@ -20,10 +20,13 @@ defmodule Pinchflat.DataCase do quote do alias Pinchflat.Repo + use Oban.Testing, repo: Repo + import Ecto import Ecto.Changeset import Ecto.Query import Pinchflat.DataCase + import Pinchflat.TestingHelperMethods end end diff --git a/test/support/fixtures/job_fixtures.ex b/test/support/fixtures/job_fixtures.ex new file mode 100644 index 0000000..4aab0dd --- /dev/null +++ b/test/support/fixtures/job_fixtures.ex @@ -0,0 +1,19 @@ +defmodule Pinchflat.JobFixtures do + @moduledoc false + + defmodule TestJobWorker do + @moduledoc false + use Oban.Worker, queue: :default + + @impl Oban.Worker + def perform(%Oban.Job{}) do + :ok + end + end + + def job_fixture() do + {:ok, job} = Oban.insert(TestJobWorker.new(%{})) + + job + end +end diff --git a/test/support/fixtures/media_fixtures.ex b/test/support/fixtures/media_fixtures.ex new file mode 100644 index 0000000..9db7014 --- /dev/null +++ b/test/support/fixtures/media_fixtures.ex @@ -0,0 +1,25 @@ +defmodule Pinchflat.MediaFixtures do + @moduledoc """ + This module defines test helpers for creating + entities via the `Pinchflat.Media` context. + """ + + alias Pinchflat.MediaSourceFixtures + + @doc """ + Generate a media_item. + """ + def media_item_fixture(attrs \\ %{}) do + {:ok, media_item} = + attrs + |> Enum.into(%{ + media_id: Faker.String.base64(12), + title: Faker.Commerce.product_name(), + video_filepath: "/video/#{Faker.File.file_name(:video)}", + channel_id: MediaSourceFixtures.channel_fixture().id + }) + |> Pinchflat.Media.create_media_item() + + media_item + end +end diff --git a/test/support/fixtures/media_source_fixtures.ex b/test/support/fixtures/media_source_fixtures.ex index 5be9bd3..007c9d4 100644 --- a/test/support/fixtures/media_source_fixtures.ex +++ b/test/support/fixtures/media_source_fixtures.ex @@ -18,8 +18,9 @@ defmodule Pinchflat.MediaSourceFixtures do Enum.into(attrs, %{ name: "Channel ##{:rand.uniform(1_000_000)}", channel_id: Base.encode16(:crypto.hash(:md5, "#{:rand.uniform(1_000_000)}")), - original_url: "https://www.youtube.com/channel/#{:rand.uniform(1_000_000)}", - media_profile_id: ProfilesFixtures.media_profile_fixture().id + original_url: "https://www.youtube.com/channel/#{Faker.String.base64(12)}", + media_profile_id: ProfilesFixtures.media_profile_fixture().id, + index_frequency_minutes: 60 }) ) |> Repo.insert() diff --git a/test/support/fixtures/tasks_fixtures.ex b/test/support/fixtures/tasks_fixtures.ex new file mode 100644 index 0000000..c310ea6 --- /dev/null +++ b/test/support/fixtures/tasks_fixtures.ex @@ -0,0 +1,24 @@ +defmodule Pinchflat.TasksFixtures do + @moduledoc """ + This module defines test helpers for creating + entities via the `Pinchflat.Tasks` context. + """ + + alias Pinchflat.JobFixtures + alias Pinchflat.MediaSourceFixtures + + @doc """ + Generate a task. + """ + def task_fixture(attrs \\ %{}) do + {:ok, task} = + attrs + |> Enum.into(%{ + channel_id: MediaSourceFixtures.channel_fixture().id, + job_id: JobFixtures.job_fixture().id + }) + |> Pinchflat.Tasks.create_task() + + task + end +end diff --git a/test/support/testing_helper_methods.ex b/test/support/testing_helper_methods.ex new file mode 100644 index 0000000..ec3818e --- /dev/null +++ b/test/support/testing_helper_methods.ex @@ -0,0 +1,30 @@ +defmodule Pinchflat.TestingHelperMethods do + @moduledoc false + + use ExUnit.CaseTemplate + + def now do + DateTime.utc_now() + end + + def now_plus(offset, unit) when unit in [:minute, :minutes] do + DateTime.add(now(), offset, :minute) + end + + def assert_changed(checker_fun, action_fn) do + before_res = checker_fun.() + action_fn.() + after_res = checker_fun.() + + assert before_res != after_res + end + + def assert_changed([from: from, to: to], checker_fun, action_fn) do + before_res = checker_fun.() + action_fn.() + after_res = checker_fun.() + + assert before_res == from + assert after_res == to + end +end diff --git a/test/test_helper.exs b/test/test_helper.exs index 54cb003..40d1202 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -3,3 +3,4 @@ Application.put_env(:pinchflat, :yt_dlp_runner, YtDlpRunnerMock) ExUnit.start() Ecto.Adapters.SQL.Sandbox.mode(Pinchflat.Repo, :manual) +Faker.start()