defmodule Oban.Web.CronQuery do @moduledoc false import Ecto.Query import Oban.Web.QueryHelpers alias Oban.Cron.Expression alias Oban.{Job, Met, Repo} alias Oban.Web.{Cron, Search, Utils} @compile {:no_warn_undefined, Oban.Pro.Cron} @suggest_qualifier [ {"names:", "cron entry name", "names:my-cron"}, {"workers:", "cron worker name", "workers:MyApp.Worker"}, {"modes:", "cron mode (static/dynamic)", "modes:static"}, {"states:", "last execution state", "states:completed"} ] @suggest_state [ {"available", "last job is available", "available"}, {"cancelled", "last job was cancelled", "cancelled"}, {"completed", "last job was completed", "completed"}, {"discarded", "last job was discarded", "discarded"}, {"executing", "last job is executing", "executing"}, {"retryable", "last job is retryable", "retryable"}, {"scheduled", "last job is scheduled", "scheduled"}, {"unknown", "no previous jobs available", "unknown"} ] @suggest_mode [ {"static", "regular cron job", "static"}, {"dynamic", "dynamic cron job", "dynamic"} ] @known_qualifiers MapSet.new(@suggest_qualifier, fn {qualifier, _, _} -> qualifier end) # Searching def filterable, do: ~w(names workers states modes)a def parse(terms) when is_binary(terms) do Search.parse(terms, &parse_term/1) end def suggest(terms, conf, _opts \\ []) do terms |> String.split(~r/\s+(?=([^\"]*\"[^\"]*\")*[^\"]*$)/) |> List.last() |> to_string() |> case do "" -> @suggest_qualifier last -> case String.split(last, ":", parts: 2) do ["names", frag] -> suggest_names(frag, conf) ["workers", frag] -> suggest_workers(frag, conf) ["modes", frag] -> suggest_static(frag, @suggest_mode) ["states", frag] -> suggest_static(frag, @suggest_state) [frag] -> suggest_static(frag, @suggest_qualifier) _ -> [] end end end defp suggest_static(fragment, possibilities) do for {field, _, _} = suggest <- possibilities, String.starts_with?(field, fragment), do: suggest end defp suggest_names(fragment, conf) do static_names = conf.name |> Met.crontab() |> Enum.map(&entry_name/1) dynamic_names = if Utils.has_crons?(conf) do query = from c in Oban.Pro.Cron, select: c.name Repo.all(conf, query) else [] end Search.restrict_suggestions(static_names ++ dynamic_names, fragment) end defp suggest_workers(fragment, conf) do conf.name |> Met.crontab() |> Enum.map(&elem(&1, 1)) |> Enum.map(&to_string/1) |> Search.restrict_suggestions(fragment) end def append(terms, choice) do Search.append(terms, choice, @known_qualifiers) end def complete(terms, conf) do case suggest(terms, conf) do [] -> terms [{match, _, _} | _] -> append(terms, match) end end defp parse_term("names:" <> names) do {:names, String.split(names, ",")} end defp parse_term("workers:" <> workers) do {:workers, String.split(workers, ",")} end defp parse_term("states:" <> states) do parsed = states |> String.split(",") |> Enum.map(fn "unknown" -> nil state -> state end) {:states, parsed} end defp parse_term("modes:" <> modes) do {:modes, String.split(modes, ",")} end defp parse_term(_term), do: {:none, ""} @history_limit 60 # Querying def all_crons(params, conf) do {sort_by, sort_dir} = parse_sort(params) limit = Map.get(params, :limit, 20) crontab = static_crontab(conf) ++ dynamic_crontab(conf) history = crontab_history(crontab, conf) conditions = Map.take(params, filterable()) crontab |> Enum.map(&build_cron(&1, history)) |> Enum.filter(&filter(&1, conditions)) |> Enum.sort_by(&order(&1, sort_by), sort_dir) |> Enum.take(limit) end def get_cron(name, conf) when is_binary(name) do with entry when not is_nil(entry) <- find_cron_entry(name, conf) do history = cron_history(name, conf) build_cron(entry, %{name => history}) end end defp find_cron_entry(name, conf) do static_entry = conf |> static_crontab() |> Enum.find(fn {_, _, _, cron_name, _, _} -> cron_name == name end) cond do static_entry -> static_entry Utils.has_crons?(conf) -> query = from c in Oban.Pro.Cron, where: c.name == ^name, select: {c.expression, c.worker, c.opts, c.name, true, c.paused}, limit: 1 Repo.one(conf, query) true -> nil end end def cron_history(name, conf) do query = Job |> where(^filter_cron_name(name, conf)) |> order_by([j], desc: j.id) |> limit(@history_limit) |> select([j], %{ state: j.state, scheduled_at: j.scheduled_at, attempted_at: j.attempted_at, finished_at: fragment("COALESCE(?, ?, ?)", j.completed_at, j.cancelled_at, j.discarded_at) }) conf |> Repo.all(query) |> Enum.reverse() end defp filter_cron_name(name, conf) when is_mysql(conf) or is_sqlite(conf) do dynamic([j], fragment("json_extract(?, '$.cron_name') = ?", j.meta, ^name)) end defp filter_cron_name(name, _conf) do dynamic([j], fragment("? @> ?", j.meta, ^%{cron_name: name})) end defp static_crontab(conf) do conf.name |> Met.crontab() |> Enum.map(fn {expr, worker, opts} = entry -> {expr, worker, opts, entry_name(entry), false, false} end) end defp entry_name({_expr, _worker, opts} = entry) do Map.get_lazy(opts, "name", fn -> Oban.Plugins.Cron.entry_name(entry) end) end defp dynamic_crontab(conf) do if Utils.has_crons?(conf) do query = select(Oban.Pro.Cron, [c], {c.expression, c.worker, c.opts, c.name, true, c.paused}) Repo.all(conf, query) else [] end end # Construction defp build_cron({expr, worker, opts, name, dynamic?, paused?}, history) do jobs = Map.get(history, name, []) last_job = List.last(jobs) fields = [ name: name, expression: expr, worker: worker, opts: opts, dynamic?: dynamic?, paused?: paused?, next_at: next_at(expr), last_at: last_at_from_job(last_job), last_state: if(last_job, do: last_job.state), history: jobs ] struct!(Cron, fields) end defp last_at_from_job(nil), do: nil defp last_at_from_job(%{finished_at: at}) when not is_nil(at), do: at defp last_at_from_job(%{attempted_at: at}) when not is_nil(at), do: at defp last_at_from_job(%{scheduled_at: at}) when not is_nil(at), do: at defp last_at_from_job(_job), do: nil def crontab_history(crontab, conf) when is_mysql(conf) or is_sqlite(conf) do crontab |> Enum.map(&elem(&1, 3)) |> Map.new(fn name -> {name, cron_history(name, conf)} end) end def crontab_history(crontab, conf) do names = Enum.map(crontab, &elem(&1, 3)) inside = from o in Job, where: fragment("? @> jsonb_build_object('cron_name', ?)", o.meta, parent_as(:list).value), select: %{ cron_name: o.meta["cron_name"], state: o.state, attempted_at: o.attempted_at, scheduled_at: o.scheduled_at, finished_at: fragment("COALESCE(?, ?, ?)", o.completed_at, o.cancelled_at, o.discarded_at) }, select_merge: %{ rn: over(row_number(), partition_by: o.meta["cron_name"], order_by: [desc: o.id]) } ranked = from t in subquery(inside), where: t.rn <= @history_limit query = from f in fragment("json_array_elements_text(?)", ^names), as: :list, left_lateral_join: j in subquery(ranked), on: true, order_by: [asc: j.rn], select: {f.value, j} conf |> Repo.all(query) |> Enum.group_by(&elem(&1, 0), fn {_name, job} -> job end) |> Map.new(fn {name, jobs} -> {name, jobs |> Enum.reject(&empty_job?/1) |> Enum.reverse()} end) end defp empty_job?(nil), do: true defp empty_job?(%{scheduled_at: nil}), do: true defp empty_job?(_job), do: false defp next_at(expression) do expression |> Expression.parse!() |> Expression.next_at() end # Sorting defp parse_sort(%{sort_by: "last_run", sort_dir: dir}) do {:last_run, {String.to_existing_atom(dir), NaiveDateTime}} end defp parse_sort(%{sort_by: "next_run", sort_dir: dir}) do {:next_run, {String.to_existing_atom(dir), NaiveDateTime}} end defp parse_sort(%{sort_by: sby, sort_dir: dir}) do {String.to_existing_atom(sby), String.to_existing_atom(dir)} end defp parse_sort(_params), do: {:worker, :asc} defp order(%{last_at: nil}, :last_run), do: ~U[2000-01-01 00:00:00Z] defp order(%{last_at: last_at}, :last_run), do: last_at defp order(%{name: name}, :name), do: name defp order(%{next_at: next_at}, :next_run), do: next_at defp order(%{expression: expression}, :schedule), do: expression defp order(%{worker: worker}, :worker), do: worker # Filtering defp filter(_row, conditions) when conditions == %{}, do: true defp filter(row, conditions) when is_map(conditions) do Enum.all?(conditions, &filter(row, &1)) end defp filter(cron, {:names, names}), do: cron.name in names defp filter(cron, {:workers, workers}), do: cron.worker in workers defp filter(cron, {:states, states}), do: cron.last_state in states defp filter(cron, {:modes, modes}) do if(cron.dynamic?, do: "dynamic", else: "static") in modes end end