Hooked up apprise notification module

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

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
 

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)

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

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
 

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 `setup_file_watcher_and_kickoff_indexing` method completes it'll return the
-  # the caller for parsing. Ideally, every item in the list will have already
+  # 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

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, "<yt:videoId>test_1</yt:videoId>"} 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, "<yt:videoId>test_1</yt:videoId>"} 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, "<yt:videoId>test_1</yt:videoId>"} 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

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, "<yt:videoId>test_1</yt:videoId>"} 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

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

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

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