Added in workers for indexing; hooked them into record creation flow

config/config.exs

@@ -29,7 +29,8 @@ config :pinchflat, PinchflatWeb.Endpoint,
 
 config :pinchflat, Oban,
   repo: Pinchflat.Repo,
-  plugins: [Oban.Plugins.Pruner],
+  # 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]
 

config/test.exs

@@ -5,7 +5,7 @@ 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: :inline
+config :pinchflat, Oban, testing: :manual
 
 # Configure your database
 #

lib/pinchflat/media/media_item.ex

@@ -1,4 +1,8 @@
 defmodule Pinchflat.Media.MediaItem do
+  @moduledoc """
+  The MediaItem schema.
+  """
+
   use Ecto.Schema
   import Ecto.Changeset
 

lib/pinchflat/media_client/channel_details.ex

@@ -25,7 +25,9 @@ defmodule Pinchflat.MediaClient.ChannelDetails do
   end
 
   @doc """
-  TODO: test
+  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)

lib/pinchflat/media_source.ex

@@ -7,6 +7,7 @@ defmodule Pinchflat.MediaSource do
   alias Pinchflat.Repo
 
   alias Pinchflat.Media
+  alias Pinchflat.Tasks.ChannelTasks
   alias Pinchflat.MediaSource.Channel
   alias Pinchflat.MediaClient.ChannelDetails
 
@@ -25,12 +26,16 @@ 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 """
@@ -38,8 +43,6 @@ defmodule Pinchflat.MediaSource do
   media ID in the source.
 
   Returns [%MediaItem{}, ...] | [%Ecto.Changeset{}, ...]
-
-  TODO: test
   """
   def index_media_items(%Channel{} = channel) do
     {:ok, media_ids} = ChannelDetails.get_video_ids(channel.original_url)
@@ -56,12 +59,19 @@ defmodule Pinchflat.MediaSource do
   end
 
   @doc """
-  Updates a channel. Returns {:ok, %Channel{}} | {:error, %Ecto.Changeset{}}
+  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.
+
+  TODO: ensure that indexing is cancelled/rescheduled if the indexing frequency
+  has been changed.
+
+  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 """
@@ -119,4 +129,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

lib/pinchflat/media_source/channel.ex

@@ -9,12 +9,13 @@ defmodule Pinchflat.MediaSource.Channel do
   alias Pinchflat.Media.MediaItem
   alias Pinchflat.Profiles.MediaProfile
 
-  @required_fields ~w(name channel_id original_url media_profile_id)a
+  @allowed_fields ~w(name channel_id index_frequency_minutes original_url media_profile_id)a
-  @allowed_fields @required_fields
+  @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

lib/pinchflat/tasks/channel_tasks.ex

@@ -0,0 +1,27 @@
+defmodule Pinchflat.Tasks.ChannelTasks do
+  @moduledoc """
+  This module contains methods for managing tasks (workers) related to channels.
+  """
+
+  alias Pinchflat.MediaSource.Channel
+  alias Pinchflat.Workers.MediaIndexingWorker
+
+  @doc """
+  Starts tasks for indexing a channel's media.
+
+  TODO: modify so that updates cancel/reschedule existing tasks as-needed
+  TODO: modify so that deletion cancels existing tasks (or maybe can do from Postgres?)
+  TODO: modify so that starting a worker adds a Task record (not implemented yet)
+  """
+  def kickoff_indexing_task(%Channel{} = channel) do
+    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()
+      |> Oban.insert()
+    end
+  end
+end

lib/pinchflat/workers/media_indexing_worker.ex

@@ -1,17 +1,46 @@
 defmodule Pinchflat.Workers.MediaIndexingWorker do
-  use Oban.Worker, queue: :media_indexing
+  @moduledoc false
 
+  use Oban.Worker,
+    queue: :media_indexing,
+    unique: [period: :infinity, states: [:available, :scheduled]]
+
+  alias __MODULE__
   alias Pinchflat.MediaSource
 
   @impl Oban.Worker
-  # This `channel_id` is the ID of the channel _record_ in
+  @doc """
-  # the database, not the ID of the channel on YouTube.
+  The ID is that of a channel _record_, not a YouTube channel ID.
-  # TODO: test
-  def perform(%Oban.Job{args: %{"channel_id" => channel_id}}) do
-    channel_id
-    |> MediaSource.get_channel!()
-    |> MediaSource.index_media_items()
 
-    :ok
+  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, %Oban.Job{}}. 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)
+    |> Oban.insert()
   end
 end

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

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</.button>
   </:actions>

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

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

test/pinchflat/media_source_test.exs

@@ -1,13 +1,14 @@
 defmodule Pinchflat.MediaSourceTest do
   use Pinchflat.DataCase
   import Mox
-
-  alias Pinchflat.MediaSource
-  alias Pinchflat.MediaSource.Channel
-
   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 +83,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 +180,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()
 

test/pinchflat/tasks/channel_tasks_test.exs

@@ -0,0 +1,26 @@
+defmodule Pinchflat.Tasks.ChannelTasksTest do
+  use Pinchflat.DataCase
+
+  import Pinchflat.MediaSourceFixtures
+
+  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
+  end
+end

test/pinchflat/workers/media_indexing_worker_test.exs

@@ -0,0 +1,67 @@
+defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
+  use Pinchflat.DataCase
+
+  import Mox
+  import Pinchflat.MediaSourceFixtures
+
+  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 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

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
 

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
 

test/support/fixtures/media_source_fixtures.ex

@@ -19,7 +19,8 @@ defmodule Pinchflat.MediaSourceFixtures do
           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/#{Faker.String.base64(12)}",
-          media_profile_id: ProfilesFixtures.media_profile_fixture().id
+          media_profile_id: ProfilesFixtures.media_profile_fixture().id,
+          index_frequency_minutes: 60
         })
       )
       |> Repo.insert()

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