diff --git a/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex b/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex
index a38ccf7..b9d5a44 100644
--- a/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex
+++ b/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex
@@ -25,7 +25,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do
function starts individual indexing tasks for each new media item. I think it does
make sense grammatically, but I could see how that's confusing.
- Returns :ok
+ Returns [binary()] where each binary is the media ID of a new media item.
"""
def kickoff_indexing_tasks_from_youtube_rss_feed(%Source{} = source) do
{:ok, media_ids} = YoutubeRss.get_recent_media_ids_from_rss(source)
@@ -37,6 +37,8 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do
MediaIndexingWorker.kickoff_with_task(source, url)
end)
+
+ new_media_ids
end
@doc """
diff --git a/lib/pinchflat/fast_indexing/fast_indexing_worker.ex b/lib/pinchflat/fast_indexing/fast_indexing_worker.ex
index 1e0f700..15d287f 100644
--- a/lib/pinchflat/fast_indexing/fast_indexing_worker.ex
+++ b/lib/pinchflat/fast_indexing/fast_indexing_worker.ex
@@ -11,8 +11,10 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do
alias __MODULE__
alias Pinchflat.Tasks
alias Pinchflat.Sources
+ alias Pinchflat.Settings
alias Pinchflat.Sources.Source
alias Pinchflat.FastIndexing.FastIndexingHelpers
+ alias Pinchflat.Notifications.SourceNotifications
@doc """
Starts the source fast indexing worker and creates a task for the source.
@@ -37,8 +39,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do
source = Sources.get_source!(source_id)
if source.fast_index do
- FastIndexingHelpers.kickoff_indexing_tasks_from_youtube_rss_feed(source)
-
+ perform_indexing_and_notification(source)
reschedule_indexing(source)
else
:ok
@@ -48,6 +49,13 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do
Ecto.StaleEntryError -> Logger.info("#{__MODULE__} discarded: source #{source_id} stale")
end
+ defp perform_indexing_and_notification(source) do
+ apprise_server = Settings.get!(:apprise_server)
+ new_media_items = FastIndexingHelpers.kickoff_indexing_tasks_from_youtube_rss_feed(source)
+
+ SourceNotifications.send_new_media_notification(apprise_server, source, length(new_media_items))
+ end
+
defp reschedule_indexing(source) do
next_run_in = Source.fast_index_frequency() * 60
diff --git a/lib/pinchflat/notifications/command_runner.ex b/lib/pinchflat/notifications/command_runner.ex
index 3a3005f..edea85c 100644
--- a/lib/pinchflat/notifications/command_runner.ex
+++ b/lib/pinchflat/notifications/command_runner.ex
@@ -16,9 +16,13 @@ defmodule Pinchflat.Notifications.CommandRunner do
Can take a single server string or a list of servers as well as additional
arguments to pass to the command.
- Returns {:ok, binary()} | {:error, binary()}.
+ Returns {:ok, binary()} | {:error, :no_servers} | {:error, binary()}
"""
@impl AppriseCommandRunner
+ def run(nil, _), do: {:error, :no_servers}
+ def run("", _), do: {:error, :no_servers}
+ def run([], _), do: {:error, :no_servers}
+
def run(endpoints, args) do
endpoints = List.wrap(endpoints)
parsed_args = CliUtils.parse_options(args)
diff --git a/lib/pinchflat/notifications/source_notifications.ex b/lib/pinchflat/notifications/source_notifications.ex
new file mode 100644
index 0000000..9ecb5c9
--- /dev/null
+++ b/lib/pinchflat/notifications/source_notifications.ex
@@ -0,0 +1,77 @@
+defmodule Pinchflat.Notifications.SourceNotifications do
+ @moduledoc """
+ Contains utilities for sending notifications about sources
+ """
+
+ require Logger
+
+ alias Pinchflat.Repo
+ alias Pinchflat.Media.MediaQuery
+
+ @doc """
+ Wraps a function that may change the number of pending or downloaded
+ media items for a source, sending an apprise notification if
+ the count changes.
+
+ Returns the return value of the provided function
+ """
+ def wrap_new_media_notification(servers, source, func) do
+ before_count = relevant_media_item_count(source)
+ retval = func.()
+ after_count = relevant_media_item_count(source)
+
+ send_new_media_notification(servers, source, after_count - before_count)
+
+ retval
+ end
+
+ @doc """
+ Sends a notification if the count of new media items has changed
+
+ Returns :ok
+ """
+ def send_new_media_notification(_, _, count) when count <= 0, do: :ok
+
+ def send_new_media_notification(servers, source, changed_count) do
+ opts = [
+ title: "[Pinchflat] New media found!",
+ body: "Found #{changed_count} new media item(s) for #{source.custom_name}. Working on downloading them now!"
+ ]
+
+ case backend_runner().run(servers, opts) do
+ {:ok, _} ->
+ Logger.info("Sent new media notification for source #{source.id}")
+
+ {:error, :no_servers} ->
+ Logger.info("No notification servers provided for source #{source.id}")
+
+ {:error, err} ->
+ Logger.error("Failed to send new media notification for source #{source.id}: #{err}")
+ end
+
+ :ok
+ end
+
+ defp relevant_media_item_count(source) do
+ pending_media_item_count(source) + downloaded_media_item_count(source)
+ end
+
+ defp pending_media_item_count(source) do
+ MediaQuery.new()
+ |> MediaQuery.for_source(source)
+ |> MediaQuery.with_media_pending_download()
+ |> Repo.aggregate(:count)
+ end
+
+ defp downloaded_media_item_count(source) do
+ MediaQuery.new()
+ |> MediaQuery.for_source(source)
+ |> MediaQuery.with_media_filepath()
+ |> Repo.aggregate(:count)
+ end
+
+ defp backend_runner do
+ # This approach lets us mock the command for testing
+ Application.get_env(:pinchflat, :apprise_runner)
+ end
+end
diff --git a/lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex b/lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex
index 70f9c2e..0e982eb 100644
--- a/lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex
+++ b/lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex
@@ -11,9 +11,11 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorker do
alias __MODULE__
alias Pinchflat.Tasks
alias Pinchflat.Sources
+ alias Pinchflat.Settings
alias Pinchflat.Sources.Source
alias Pinchflat.FastIndexing.FastIndexingWorker
alias Pinchflat.SlowIndexing.SlowIndexingHelpers
+ alias Pinchflat.Notifications.SourceNotifications
@doc """
Starts the source slow indexing worker and creates a task for the source.
@@ -78,21 +80,21 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorker do
case {source.index_frequency_minutes, source.last_indexed_at} do
{index_freq, _} when index_freq > 0 ->
# If the indexing is on a schedule simply run indexing and reschedule
- SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+ perform_indexing_and_notification(source)
maybe_enqueue_fast_indexing_task(source)
reschedule_indexing(source)
{_, nil} ->
# If the source has never been indexed, index it once
# even if it's not meant to reschedule
- SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+ perform_indexing_and_notification(source)
:ok
_ ->
# If the source HAS been indexed and is not meant to reschedule,
# perform a no-op (unless forced)
if args["force"] do
- SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+ perform_indexing_and_notification(source)
end
:ok
@@ -102,6 +104,14 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorker do
Ecto.StaleEntryError -> Logger.info("#{__MODULE__} discarded: source #{source_id} stale")
end
+ defp perform_indexing_and_notification(source) do
+ apprise_server = Settings.get!(:apprise_server)
+
+ SourceNotifications.wrap_new_media_notification(apprise_server, source, fn ->
+ SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source)
+ end)
+ end
+
defp reschedule_indexing(source) do
next_run_in = source.index_frequency_minutes * 60
diff --git a/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex b/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex
index c95306c..b6b118a 100644
--- a/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex
+++ b/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex
@@ -60,7 +60,7 @@ defmodule Pinchflat.SlowIndexing.SlowIndexingHelpers do
def index_and_enqueue_download_for_media_items(%Source{} = source) do
# See the method definition below for more info on how file watchers work
# (important reading if you're not familiar with it)
- {:ok, media_attributes} = get_media_attributes_for_collection_and_setup_file_watcher(source)
+ {:ok, media_attributes} = setup_file_watcher_and_kickoff_indexing(source)
# Reload because the source may have been updated during the (long-running) indexing process
# and important settings like `download_media` may have changed.
source = Repo.reload!(source)
@@ -84,15 +84,15 @@ defmodule Pinchflat.SlowIndexing.SlowIndexingHelpers do
# lines (ie: you should gracefully fail if you can't parse a line).
#
# This works in-tandem with the normal (blocking) media indexing behaviour. When
- # the `get_media_attributes_for_collection` method completes it'll return the FULL result to
- # the caller for parsing. Ideally, every item in the list will have already
+ # the `setup_file_watcher_and_kickoff_indexing` method completes it'll return the
+ # FULL result to the caller for parsing. Ideally, every item in the list will have already
# been processed by the file follower, but if not, the caller handles creation
# of any media items that were missed/initially failed.
#
# It attempts a graceful shutdown of the file follower after the indexing is done,
# but the FileFollowerServer will also stop itself if it doesn't see any activity
# for a sufficiently long time.
- defp get_media_attributes_for_collection_and_setup_file_watcher(source) do
+ defp setup_file_watcher_and_kickoff_indexing(source) do
{:ok, pid} = FileFollowerServer.start_link()
handler = fn filepath -> setup_file_follower_watcher(pid, filepath, source) end
diff --git a/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs b/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs
index 6c55051..23350c0 100644
--- a/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs
+++ b/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs
@@ -24,7 +24,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpersTest do
test "enqueues a new worker for each new media_id in the source's RSS feed", %{source: source} do
expect(HTTPClientMock, :get, fn _url -> {:ok, "test_1"} end)
- assert :ok = FastIndexingHelpers.kickoff_indexing_tasks_from_youtube_rss_feed(source)
+ assert [_] = FastIndexingHelpers.kickoff_indexing_tasks_from_youtube_rss_feed(source)
assert [worker] = all_enqueued(worker: MediaIndexingWorker)
assert worker.args["id"] == source.id
@@ -35,10 +35,16 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpersTest do
expect(HTTPClientMock, :get, fn _url -> {:ok, "test_1"} end)
media_item_fixture(source_id: source.id, media_id: "test_1")
- assert :ok = FastIndexingHelpers.kickoff_indexing_tasks_from_youtube_rss_feed(source)
+ assert [] = FastIndexingHelpers.kickoff_indexing_tasks_from_youtube_rss_feed(source)
refute_enqueued(worker: MediaIndexingWorker)
end
+
+ test "returns the IDs of the found media items", %{source: source} do
+ expect(HTTPClientMock, :get, fn _url -> {:ok, "test_1"} end)
+
+ assert ["test_1"] = FastIndexingHelpers.kickoff_indexing_tasks_from_youtube_rss_feed(source)
+ end
end
describe "index_and_enqueue_download_for_media_item/2" do
diff --git a/test/pinchflat/fast_indexing/fast_indexing_worker_test.exs b/test/pinchflat/fast_indexing/fast_indexing_worker_test.exs
index a69a99d..55007ba 100644
--- a/test/pinchflat/fast_indexing/fast_indexing_worker_test.exs
+++ b/test/pinchflat/fast_indexing/fast_indexing_worker_test.exs
@@ -4,6 +4,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorkerTest do
import Mox
import Pinchflat.SourcesFixtures
+ alias Pinchflat.Settings
alias Pinchflat.Sources.Source
alias Pinchflat.FastIndexing.FastIndexingWorker
@@ -74,4 +75,28 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorkerTest do
assert :ok = perform_job(FastIndexingWorker, %{id: 0})
end
end
+
+ describe "perform/1 when testing notifications" do
+ setup do
+ Settings.set(apprise_server: "server_1")
+
+ :ok
+ end
+
+ test "sends a notification if new media was found" do
+ source = source_fixture(fast_index: true)
+
+ expect(HTTPClientMock, :get, fn _url -> {:ok, "test_1"} end)
+
+ expect(AppriseRunnerMock, :run, fn servers, opts ->
+ assert "server_1" = servers
+ assert is_binary(Keyword.get(opts, :title))
+ assert is_binary(Keyword.get(opts, :body))
+
+ {:ok, ""}
+ end)
+
+ perform_job(FastIndexingWorker, %{id: source.id})
+ end
+ end
end
diff --git a/test/pinchflat/notifications/command_runner_test.exs b/test/pinchflat/notifications/command_runner_test.exs
index 658dfed..0ed4cc6 100644
--- a/test/pinchflat/notifications/command_runner_test.exs
+++ b/test/pinchflat/notifications/command_runner_test.exs
@@ -11,7 +11,7 @@ defmodule Pinchflat.Notifications.CommandRunnerTest do
describe "run/2" do
test "returns :ok when the command succeeds" do
- assert {:ok, _} = Runner.run("", [])
+ assert {:ok, _} = Runner.run("server_1", [])
end
test "includes the servers as the first argument" do
@@ -27,16 +27,22 @@ defmodule Pinchflat.Notifications.CommandRunnerTest do
end
test "passes all arguments to the command" do
- assert {:ok, output} = Runner.run("", ["--dry-run"])
+ assert {:ok, output} = Runner.run("server_1", ["--dry-run"])
assert String.contains?(output, "--dry-run")
end
test "returns the output when the command fails" do
wrap_executable("/bin/false", fn ->
- assert {:error, ""} = Runner.run("", [])
+ assert {:error, ""} = Runner.run("server_1", [])
end)
end
+
+ test "returns a relevant error if no servers are provided" do
+ assert {:error, :no_servers} = Runner.run(nil, [])
+ assert {:error, :no_servers} = Runner.run("", [])
+ assert {:error, :no_servers} = Runner.run([], [])
+ end
end
describe "version/0" do
diff --git a/test/pinchflat/notifications/source_notifications_test.exs b/test/pinchflat/notifications/source_notifications_test.exs
new file mode 100644
index 0000000..788ae7f
--- /dev/null
+++ b/test/pinchflat/notifications/source_notifications_test.exs
@@ -0,0 +1,100 @@
+defmodule Pinchflat.Notifications.SourceNotificationsTest do
+ use Pinchflat.DataCase
+
+ import Mox
+ import Pinchflat.MediaFixtures
+ import Pinchflat.SourcesFixtures
+
+ alias Pinchflat.Notifications.SourceNotifications
+
+ @apprise_servers ["server_1", "server_2"]
+
+ setup :verify_on_exit!
+
+ describe "wrap_new_media_notification/3" do
+ test "sends a notification when the pending count changes" do
+ source = source_fixture()
+
+ expect(AppriseRunnerMock, :run, fn servers, opts ->
+ assert servers == @apprise_servers
+
+ assert opts == [
+ title: "[Pinchflat] New media found!",
+ body: "Found 1 new media item(s) for #{source.custom_name}. Working on downloading them now!"
+ ]
+
+ {:ok, ""}
+ end)
+
+ SourceNotifications.wrap_new_media_notification(@apprise_servers, source, fn ->
+ media_item_fixture(%{source_id: source.id, media_filepath: nil})
+ end)
+ end
+
+ test "sends a notification when the downloaded count changes" do
+ source = source_fixture()
+
+ expect(AppriseRunnerMock, :run, fn servers, opts ->
+ assert servers == @apprise_servers
+
+ assert opts == [
+ title: "[Pinchflat] New media found!",
+ body: "Found 1 new media item(s) for #{source.custom_name}. Working on downloading them now!"
+ ]
+
+ {:ok, ""}
+ end)
+
+ SourceNotifications.wrap_new_media_notification(@apprise_servers, source, fn ->
+ media_item_fixture(%{source_id: source.id, media_filepath: "file.mp4"})
+ end)
+ end
+
+ test "does not send a notification when the count does not change" do
+ source = source_fixture()
+
+ expect(AppriseRunnerMock, :run, 0, fn _, _ -> {:ok, ""} end)
+
+ SourceNotifications.wrap_new_media_notification(@apprise_servers, source, fn ->
+ media_item_fixture(%{source_id: source.id, prevent_download: true, media_filepath: nil})
+ end)
+ end
+
+ test "returns the value of the function" do
+ source = source_fixture()
+ expect(AppriseRunnerMock, :run, 0, fn _, _ -> {:ok, ""} end)
+
+ retval = SourceNotifications.wrap_new_media_notification(@apprise_servers, source, fn -> "value" end)
+
+ assert retval == "value"
+ end
+ end
+
+ describe "send_new_media_notification/3" do
+ test "sends a notification when count is positive" do
+ source = source_fixture()
+
+ expect(AppriseRunnerMock, :run, fn servers, opts ->
+ assert servers == @apprise_servers
+
+ assert opts == [
+ title: "[Pinchflat] New media found!",
+ body: "Found 1 new media item(s) for #{source.custom_name}. Working on downloading them now!"
+ ]
+
+ {:ok, ""}
+ end)
+
+ :ok = SourceNotifications.send_new_media_notification(@apprise_servers, source, 1)
+ end
+
+ test "does not send a notification when count not positive" do
+ source = source_fixture()
+
+ expect(AppriseRunnerMock, :run, 0, fn _, _ -> {:ok, ""} end)
+
+ :ok = SourceNotifications.send_new_media_notification(@apprise_servers, source, 0)
+ :ok = SourceNotifications.send_new_media_notification(@apprise_servers, source, -1)
+ end
+ end
+end
diff --git a/test/pinchflat/slow_indexing/media_collection_indexing_worker_test.exs b/test/pinchflat/slow_indexing/media_collection_indexing_worker_test.exs
index 6a3c140..0830435 100644
--- a/test/pinchflat/slow_indexing/media_collection_indexing_worker_test.exs
+++ b/test/pinchflat/slow_indexing/media_collection_indexing_worker_test.exs
@@ -7,6 +7,7 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorkerTest do
import Pinchflat.SourcesFixtures
alias Pinchflat.Tasks
+ alias Pinchflat.Settings
alias Pinchflat.Sources.Source
alias Pinchflat.FastIndexing.FastIndexingWorker
alias Pinchflat.Downloading.MediaDownloadWorker
@@ -51,6 +52,12 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorkerTest do
end
describe "perform/1" do
+ setup do
+ stub(AppriseRunnerMock, :run, fn _, _ -> {:ok, ""} end)
+
+ :ok
+ end
+
test "it indexes the source if it should be indexed" do
expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end)
@@ -210,4 +217,30 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorkerTest do
assert :ok = perform_job(MediaCollectionIndexingWorker, %{id: 0})
end
end
+
+ describe "perform/1 when testing apprise notifications" do
+ setup do
+ Settings.set(apprise_server: "server_1")
+
+ :ok
+ end
+
+ test "sends a notification if new media was found" do
+ source = source_fixture()
+
+ expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts ->
+ {:ok, source_attributes_return_fixture()}
+ end)
+
+ expect(AppriseRunnerMock, :run, fn servers, opts ->
+ assert "server_1" = servers
+ assert is_binary(Keyword.get(opts, :title))
+ assert is_binary(Keyword.get(opts, :body))
+
+ {:ok, ""}
+ end)
+
+ perform_job(MediaCollectionIndexingWorker, %{id: source.id})
+ end
+ end
end