Index a channel (#9)

* Ran a MediaItem generator; Reformatted to my liking

* [WIP] added basic index function

* setup oban

* Added basic Oban job for indexing

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

* Added a task model with a phx generator

* Tied together tasks with jobs and channels

assets/yarn.lock

@@ -0,0 +1,4 @@
+# THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY.
+# yarn lockfile v1
+
+

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

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

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

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

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

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),

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

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

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 ~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
 
     belongs_to :media_profile, MediaProfile
 
+    has_many :media_items, MediaItem
+
     timestamps(type: :utc_datetime)
   end
 

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

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

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

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

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>

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
 

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"},

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

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

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

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

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

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,15 @@
 defmodule Pinchflat.MediaSourceTest do
   use Pinchflat.DataCase
   import Mox
-
+  import Pinchflat.TasksFixtures
-  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 +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

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

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
+  @invalid_attrs %{name: nil, output_path_template: nil}
-    alias Pinchflat.Profiles.MediaProfile
 
-    import Pinchflat.ProfilesFixtures
+  describe "list_media_profiles/0" do
-
+    test "it returns all media_profiles" do
-    @invalid_attrs %{name: nil, output_path_template: nil}
-
-    test "list_media_profiles/0 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

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

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

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

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/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

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

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)}",
+          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/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

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

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()