Support trigger a job that uses a generic action

config/config.exs

@@ -22,7 +22,8 @@ if Mix.env() == :test do
     queues: [
       triggered_process: 10,
       triggered_process_2: 10,
-      triggered_say_hello: 10
+      triggered_say_hello: 10,
+      triggered_process_generic: 10
     ]
 
   config :ash_oban, actor_persister: AshOban.Test.ActorPersister

lib/transformers/define_schedulers.ex

@@ -210,18 +210,32 @@ defmodule AshOban.Transformers.DefineSchedulers do
                   fn _ -> %{} end
               end
 
+            trigger_action = Ash.Resource.Info.action(unquote(resource), unquote(trigger.action))
+
             case AshOban.lookup_actor(args["actor"]) do
               {:ok, actor} ->
-                unquote(resource)
-                |> stream(actor)
-                |> Stream.map(fn record ->
+                if(trigger_action.type == :action) do
+                  action_input = unquote(Macro.escape(trigger.action_input || %{}))
+
                   unquote(worker_module_name).new(%{
-                    primary_key: Map.take(record, unquote(primary_key)),
-                    metadata: metadata.(record),
-                    actor: args["actor"]
+                    actor: args["actor"],
+                    action_arguments: action_input
                   })
-                end)
-                |> insert()
+                  |> Oban.insert!()
+
+                  :ok
+                else
+                  unquote(resource)
+                  |> stream(actor)
+                  |> Stream.map(fn record ->
+                    unquote(worker_module_name).new(%{
+                      primary_key: Map.take(record, unquote(primary_key)),
+                      metadata: metadata.(record),
+                      actor: args["actor"]
+                    })
+                  end)
+                  |> insert()
+                end
 
               {:error, e} ->
                 raise Ash.Error.to_ash_error(e)
@@ -574,6 +588,66 @@ defmodule AshOban.Transformers.DefineSchedulers do
     end
   end
 
+  defp work(trigger, worker, _atomic?, :action, pro?, _read_action, resource, domain) do
+    function_name =
+      if pro? do
+        :process
+      else
+        :perform
+      end
+
+    if trigger.state != :active do
+      quote location: :keep do
+        @impl unquote(worker)
+        def unquote(function_name)(_) do
+          {:discard, unquote(trigger.state)}
+        end
+      end
+    else
+      quote location: :keep, generated: true do
+        @impl unquote(worker)
+        def unquote(function_name)(%Oban.Job{args: args} = job) do
+          case AshOban.lookup_actor(args["actor"]) do
+            {:ok, actor} ->
+              authorize? = AshOban.authorize?()
+
+              AshOban.debug(
+                "Trigger #{unquote(inspect(resource))}.#{unquote(trigger.name)} triggered",
+                unquote(trigger.debug?)
+              )
+
+              action_input = unquote(Macro.escape(trigger.action_input || %{}))
+
+              input =
+                if unquote(is_nil(trigger.read_metadata)) do
+                  %{}
+                else
+                  %{metadata: args["metadata"]}
+                end
+                |> Map.merge(args["action_arguments"] || %{})
+                |> Map.merge(action_input)
+
+              unquote(resource)
+              |> Ash.ActionInput.for_action(
+                unquote(trigger.action),
+                input,
+                authorize?: authorize?,
+                actor: actor,
+                domain: unquote(domain),
+                skip_unknown_inputs: Map.keys(input)
+              )
+              |> Ash.run_action!()
+
+              :ok
+
+            {:error, error} ->
+              raise Ash.Error.to_ash_error(error)
+          end
+        end
+      end
+    end
+  end
+
   defp work(trigger, worker, atomic?, trigger_action_type, pro?, read_action, resource, domain) do
     function_name =
       if pro? do

test/ash_oban_test.exs

@@ -14,13 +14,18 @@ defmodule AshObanTest do
 
   setup do
     Enum.each(
-      [:triggered_process, :triggered_process_2, :triggered_say_hello],
+      [
+        :triggered_process,
+        :triggered_process_2,
+        :triggered_say_hello,
+        :triggered_process_generic
+      ],
       &Oban.drain_queue(queue: &1)
     )
   end
 
   test "nothing happens if no records exist" do
-    assert %{success: 2} = AshOban.Test.schedule_and_run_triggers(Triggered)
+    assert %{success: 4} = AshOban.Test.schedule_and_run_triggers(Triggered)
   end
 
   test "if a record exists, it is processed" do
@@ -66,15 +71,16 @@ defmodule AshObanTest do
     |> Ash.Changeset.for_create(:create)
     |> Ash.create!()
 
-    assert %{success: 3, failure: 1} =
+    assert %{success: 5, failure: 1} =
              AshOban.Test.schedule_and_run_triggers(Triggered)
   end
 
   test "dsl introspection" do
     assert [
              %AshOban.Trigger{action: :process},
              %AshOban.Trigger{action: :process_atomically},
-             %AshOban.Trigger{action: :process, scheduler: nil}
+             %AshOban.Trigger{action: :process, scheduler: nil},
+             %AshOban.Trigger{name: :process_generic}
            ] = AshOban.Info.oban_triggers(Triggered)
   end
 
@@ -87,7 +93,8 @@ defmodule AshObanTest do
         queues: [
           triggered_process: 10,
           triggered_process_2: 10,
-          triggered_say_hello: 10
+          triggered_say_hello: 10,
+          triggered_process_generic: 10
         ]
       )
 
@@ -97,6 +104,7 @@ defmodule AshObanTest do
                 [
                   crontab: [
                     {"0 0 1 1 *", AshOban.Test.Triggered.AshOban.ActionWorker.SayHello, []},
+                    {"* * * * *", AshOban.Test.Triggered.AshOban.Scheduler.ProcessGeneric, []},
                     {"* * * * *", AshOban.Test.Triggered.AshOban.Scheduler.ProcessAtomically, []},
                     {"* * * * *", AshOban.Test.Triggered.AshOban.Scheduler.Process, []}
                   ]
@@ -105,7 +113,8 @@ defmodule AshObanTest do
              queues: [
                triggered_process: 10,
                triggered_process_2: 10,
-               triggered_say_hello: 10
+               triggered_say_hello: 10,
+               triggered_process_generic: 10
              ]
            ] = config
   end
@@ -125,7 +134,8 @@ defmodule AshObanTest do
            queues: [
              triggered_process: 10,
              triggered_process_2: 10,
-             triggered_say_hello: 10
+             triggered_say_hello: 10,
+             triggered_process_generic: 10
            ]}
         ],
         queues: false
@@ -141,6 +151,8 @@ defmodule AshObanTest do
                   crontab: [
                     {"0 0 1 1 *", AshOban.Test.Triggered.AshOban.ActionWorker.SayHello,
                      [paused: false]},
+                    {"* * * * *", AshOban.Test.Triggered.AshOban.Scheduler.ProcessGeneric,
+                     [paused: false]},
                     {"* * * * *", AshOban.Test.Triggered.AshOban.Scheduler.ProcessAtomically,
                      [paused: false]},
                     {"* * * * *", AshOban.Test.Triggered.AshOban.Scheduler.Process,
@@ -151,7 +163,8 @@ defmodule AshObanTest do
                 queues: [
                   triggered_process: 10,
                   triggered_process_2: 10,
-                  triggered_say_hello: 10
+                  triggered_say_hello: 10,
+                  triggered_process_generic: 10
                 ]}
              ],
              queues: false

test/support/triggered.ex

@@ -30,6 +30,12 @@ defmodule AshOban.Test.Triggered do
         worker_read_action(:read)
         scheduler_cron false
       end
+
+      trigger :process_generic do
+        action :say_hello
+        max_attempts 2
+        scheduler_cron "* * * * *"
+      end
     end
 
     scheduled_actions do