Improved task fetching and deletion methods

lib/pinchflat/sources.ex

@@ -190,7 +190,7 @@ defmodule Pinchflat.Sources do
       %{__meta__: %{state: :loaded}} ->
         case changeset.changes do
           %{index_frequency_minutes: mins} when mins > 0 -> SourceTasks.kickoff_indexing_task(source)
-          %{index_frequency_minutes: _} -> Tasks.delete_pending_tasks_for(source)
+          %{index_frequency_minutes: _} -> Tasks.delete_pending_tasks_for(source, "MediaIndexingWorker")
           _ -> :ok
         end
     end

lib/pinchflat/tasks.ex

@@ -19,30 +19,46 @@ defmodule Pinchflat.Tasks do
 
   @doc """
   Returns the list of tasks for a given record type and ID. Optionally allows you to specify
-  which job states to include.
+  which worker or job states to include.
 
   Returns [%Task{}, ...]
   """
-  def list_tasks_for(attached_record_type, attached_record_id, job_states \\ Oban.Job.states()) do
+  def list_tasks_for(attached_record_type, attached_record_id, worker_name \\ nil, job_states \\ Oban.Job.states()) do
     stringified_states = Enum.map(job_states, &to_string/1)
 
+    worker_name_finder =
+      if worker_name do
+        # Workers are the full module name - we want to match on the string ENDING with
+        # the passed worker name and it should be preceeded with a . so we aren't matching
+        # on a substring. You can pass in more fragments of the worker name if you need
+        # to disambiguate. eg: "TestWorker" or "FooBar.TestWorker"
+        worker_finder = "%.#{worker_name}"
+
+        dynamic([_t, j], fragment("? LIKE ?", j.worker, ^worker_finder))
+      else
+        true
+      end
+
     Repo.all(
       from t in Task,
         join: j in assoc(t, :job),
         where: field(t, ^attached_record_type) == ^attached_record_id,
+        where: ^worker_name_finder,
         where: j.state in ^stringified_states
     )
   end
 
   @doc """
-  Returns the list of pending tasks for a given record type and ID.
+  Returns the list of pending tasks for a given record type and ID. Optionally allows you to specify
+  which worker to include.
 
   Returns [%Task{}, ...]
   """
-  def list_pending_tasks_for(attached_record_type, attached_record_id) do
+  def list_pending_tasks_for(attached_record_type, attached_record_id, worker_name \\ nil) do
     list_tasks_for(
       attached_record_type,
       attached_record_id,
+      worker_name,
       [:available, :scheduled, :retryable]
     )
   end
@@ -107,14 +123,15 @@ defmodule Pinchflat.Tasks do
 
   @doc """
   Deletes all tasks attached to a given record, cancelling any attached jobs.
+  Optionally allows you to specify which worker to include.
 
   Returns :ok
   """
-  def delete_tasks_for(attached_record) do
+  def delete_tasks_for(attached_record, worker_name \\ nil) do
     tasks =
       case attached_record do
-        %Source{} = source -> list_tasks_for(:source_id, source.id)
-        %MediaItem{} = media_item -> list_tasks_for(:media_item_id, media_item.id)
+        %Source{} = source -> list_tasks_for(:source_id, source.id, worker_name)
+        %MediaItem{} = media_item -> list_tasks_for(:media_item_id, media_item.id, worker_name)
       end
 
     Enum.each(tasks, &delete_task/1)
@@ -122,14 +139,15 @@ defmodule Pinchflat.Tasks do
 
   @doc """
   Deletes all _pending_ tasks attached to a given record, cancelling any attached jobs.
+  Optionally allows you to specify which worker to include.
 
   Returns :ok
   """
-  def delete_pending_tasks_for(attached_record) do
+  def delete_pending_tasks_for(attached_record, worker_name \\ nil) do
     tasks =
       case attached_record do
-        %Source{} = source -> list_pending_tasks_for(:source_id, source.id)
-        %MediaItem{} = media_item -> list_pending_tasks_for(:media_item_id, media_item.id)
+        %Source{} = source -> list_pending_tasks_for(:source_id, source.id, worker_name)
+        %MediaItem{} = media_item -> list_pending_tasks_for(:media_item_id, media_item.id, worker_name)
       end
 
     Enum.each(tasks, &delete_task/1)

lib/pinchflat/tasks/source_tasks.ex

@@ -18,7 +18,7 @@ defmodule Pinchflat.Tasks.SourceTasks do
   Returns {:ok, %Task{}}.
   """
   def kickoff_indexing_task(%Source{} = source) do
-    Tasks.delete_pending_tasks_for(source)
+    Tasks.delete_pending_tasks_for(source, "MediaIndexingWorker")
 
     source
     |> Map.take([:id])

test/pinchflat/sources_test.exs

@@ -244,7 +244,8 @@ defmodule Pinchflat.SourcesTest do
 
     test "updating the index frequency to 0 will delete any pending tasks" do
       source = source_fixture()
-      task = task_fixture(source_id: source.id)
+      {:ok, job} = Oban.insert(MediaIndexingWorker.new(%{"id" => source.id}))
+      task = task_fixture(source_id: source.id, job_id: job.id)
       update_attrs = %{index_frequency_minutes: 0}
 
       assert {:ok, %Source{}} = Sources.update_source(source, update_attrs)

test/pinchflat/tasks/source_tasks_test.exs

@@ -34,7 +34,8 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
 
     test "it deletes any pending tasks for the source" do
       source = source_fixture()
-      task = task_fixture(source_id: source.id)
+      {:ok, job} = Oban.insert(MediaIndexingWorker.new(%{"id" => source.id}))
+      task = task_fixture(source_id: source.id, job_id: job.id)
 
       assert {:ok, _} = SourceTasks.kickoff_indexing_task(source)
 

test/pinchflat/tasks_test.exs

@@ -36,7 +36,7 @@ defmodule Pinchflat.TasksTest do
     end
   end
 
-  describe "list_tasks_for/3" do
+  describe "list_tasks_for/4" do
     test "it lets you specify which record type/ID to join on" do
       task = task_fixture()
 
@@ -46,12 +46,25 @@ defmodule Pinchflat.TasksTest do
     test "it lets you specify which job states to include" do
       task = task_fixture()
 
-      assert Tasks.list_tasks_for(:source_id, task.source_id, [:available]) == [task]
-      assert Tasks.list_tasks_for(:source_id, task.source_id, [:cancelled]) == []
+      assert Tasks.list_tasks_for(:source_id, task.source_id, nil, [:available]) == [task]
+      assert Tasks.list_tasks_for(:source_id, task.source_id, nil, [:cancelled]) == []
+    end
+
+    test "it lets you specify which worker to include" do
+      task = task_fixture()
+
+      assert Tasks.list_tasks_for(:source_id, task.source_id, "TestJobWorker") == [task]
+      assert Tasks.list_tasks_for(:source_id, task.source_id, "FooBarWorker") == []
+    end
+
+    test "it includes all workers if no worker is specified" do
+      task = task_fixture()
+
+      assert Tasks.list_tasks_for(:source_id, task.source_id, nil) == [task]
     end
   end
 
-  describe "list_pending_tasks_for/2" do
+  describe "list_pending_tasks_for/3" do
     test "it lists pending tasks" do
       task = task_fixture()
 
@@ -64,6 +77,13 @@ defmodule Pinchflat.TasksTest do
 
       assert Tasks.list_pending_tasks_for(:source_id, task.source_id) == []
     end
+
+    test "it lets you specify which worker to include" do
+      task = task_fixture()
+
+      assert Tasks.list_pending_tasks_for(:source_id, task.source_id, "TestJobWorker") == [task]
+      assert Tasks.list_pending_tasks_for(:source_id, task.source_id, "FooBarWorker") == []
+    end
   end
 
   describe "get_task!/1" do
@@ -154,7 +174,7 @@ defmodule Pinchflat.TasksTest do
     end
   end
 
-  describe "delete_tasks_for/1" do
+  describe "delete_tasks_for/2" do
     test "it deletes tasks attached to a source" do
       source = source_fixture()
       task = task_fixture(source_id: source.id)
@@ -170,6 +190,17 @@ defmodule Pinchflat.TasksTest do
       assert :ok = Tasks.delete_tasks_for(media_item)
       assert_raise Ecto.NoResultsError, fn -> Tasks.get_task!(task.id) end
     end
+
+    test "deletion can specify which worker to include" do
+      media_item = media_item_fixture()
+      task = task_fixture(media_item_id: media_item.id)
+
+      assert :ok = Tasks.delete_tasks_for(media_item, "FooBarWorker")
+      assert Repo.reload!(task)
+
+      assert :ok = Tasks.delete_tasks_for(media_item, "TestJobWorker")
+      assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end
+    end
   end
 
   describe "delete_pending_tasks_for/1" do
@@ -200,6 +231,17 @@ defmodule Pinchflat.TasksTest do
       assert Tasks.get_task!(cancelled_task.id)
       assert_raise Ecto.NoResultsError, fn -> Repo.reload!(pending_task) end
     end
+
+    test "deletion can specify which worker to include" do
+      media_item = media_item_fixture()
+      task = task_fixture(media_item_id: media_item.id)
+
+      assert :ok = Tasks.delete_pending_tasks_for(media_item, "FooBarWorker")
+      assert Repo.reload!(task)
+
+      assert :ok = Tasks.delete_pending_tasks_for(media_item, "TestJobWorker")
+      assert_raise Ecto.NoResultsError, fn -> Repo.reload!(task) end
+    end
   end
 
   describe "change_task/1" do