Add admin endpoint to flush Oban jobs by state

lib/towerops_web/controllers/admin_controller.ex

@@ -19,4 +19,27 @@ defmodule ToweropsWeb.AdminController do
   def stop_impersonate(conn, _params) do
     UserAuth.stop_impersonation(conn)
   end
+
+  @doc """
+  Flush Oban jobs by state. Cancels all jobs in the given state(s).
+  POST /admin/oban/flush?states=scheduled,retryable
+  """
+  def flush_oban(conn, %{"states" => states_param}) do
+    import Ecto.Query
+
+    valid_states = ~w(scheduled retryable available)
+    requested = String.split(states_param, ",") |> Enum.map(&String.trim/1)
+    states = Enum.filter(requested, &(&1 in valid_states))
+
+    if states == [] do
+      conn |> put_status(400) |> json(%{error: "No valid states. Use: scheduled, retryable, available"})
+    else
+      {count, _} =
+        Oban.Job
+        |> where([j], j.state in ^states)
+        |> Towerops.Repo.delete_all()
+
+      conn |> json(%{ok: true, deleted: count, states: states})
+    end
+  end
 end

lib/towerops_web/router.ex

@@ -277,6 +277,7 @@ defmodule ToweropsWeb.Router do
 
     post "/admin/impersonate/:user_id", AdminController, :start_impersonate
     delete "/admin/impersonate", AdminController, :stop_impersonate
+    post "/admin/oban/flush", AdminController, :flush_oban
   end
 
   live_session :require_superuser,