diff --git a/vendor/oban_web/lib/oban/web/components/core.ex b/vendor/oban_web/lib/oban/web/components/core.ex index ac7e4267..db971d36 100644 --- a/vendor/oban_web/lib/oban/web/components/core.ex +++ b/vendor/oban_web/lib/oban/web/components/core.ex @@ -212,6 +212,9 @@ defmodule Oban.Web.Components.Core do defp badge_icon(%{name: "lock_closed"} = assigns), do: ~H[] + defp badge_icon(%{name: "signal"} = assigns), + do: ~H[] + defp badge_icon(%{name: "sparkles"} = assigns), do: ~H[] diff --git a/vendor/oban_web/lib/oban/web/cron_expr.ex b/vendor/oban_web/lib/oban/web/cron_expr.ex index 7ba4d2f6..1d9e0cee 100644 --- a/vendor/oban_web/lib/oban/web/cron_expr.ex +++ b/vendor/oban_web/lib/oban/web/cron_expr.ex @@ -45,10 +45,10 @@ defmodule Oban.Web.CronExpr do defp describe_parsed(expression) do with [min, hrs, dom, "*", dow] <- String.split(expression, " ", parts: 5), - {:ok, parsed_min} <- parse_field(min), - {:ok, parsed_hrs} <- parse_field(hrs), - {:ok, parsed_dom} <- parse_field(dom), - {:ok, parsed_dow} <- parse_field(dow, @days_of_week_translations) do + {:ok, parsed_min} <- parse_field(min, 0..59), + {:ok, parsed_hrs} <- parse_field(hrs, 0..23), + {:ok, parsed_dom} <- parse_field(dom, 1..31), + {:ok, parsed_dow} <- parse_field(dow, 0..7, @days_of_week_translations) do combine_description(parsed_min, parsed_hrs, parsed_dom, parsed_dow) else _ -> nil @@ -57,11 +57,11 @@ defmodule Oban.Web.CronExpr do # Field Parsing - defp parse_field(field, translations \\ %{}) do + defp parse_field(field, bounds, translations \\ %{}) do parts = field |> String.split(",") - |> Enum.map(&parse_part(&1, translations)) + |> Enum.map(&parse_part(&1, bounds, translations)) if Enum.any?(parts, &(&1 == :error)) do :error @@ -70,37 +70,40 @@ defmodule Oban.Web.CronExpr do end end - defp parse_part("*", _translations), do: :wildcard + defp parse_part("*", _bounds, _translations), do: :wildcard - defp parse_part("*/" <> step, _translations) do + defp parse_part("*/" <> step, bounds, _translations) do case Integer.parse(step) do - {num, ""} when num > 0 -> {:step, num} + {num, ""} when num > bounds.first and num <= bounds.last -> {:step, num} _ -> :error end end - defp parse_part(part, translations) do + defp parse_part(part, bounds, translations) do if String.contains?(part, "-") do - parse_range(part, translations) + parse_range(part, bounds, translations) else - parse_value(part, translations) + parse_value(part, bounds, translations) end end - defp parse_range(part, translations) do + defp parse_range(part, bounds, translations) do with [start_str, end_str] <- String.split(part, "-", parts: 2), {:ok, start_val} <- translate_or_parse(start_str, translations), - {:ok, end_val} <- translate_or_parse(end_str, translations) do + {:ok, end_val} <- translate_or_parse(end_str, translations), + true <- start_val in bounds and end_val in bounds and end_val >= start_val do {:range, start_val, end_val} else _ -> :error end end - defp parse_value(part, translations) do - case translate_or_parse(part, translations) do - {:ok, val} -> {:value, val} - :error -> :error + defp parse_value(part, bounds, translations) do + with {:ok, val} <- translate_or_parse(part, translations), + true <- val in bounds do + {:value, val} + else + _ -> :error end end diff --git a/vendor/oban_web/lib/oban/web/helpers.ex b/vendor/oban_web/lib/oban/web/helpers.ex index 5f2c297f..9d290048 100644 --- a/vendor/oban_web/lib/oban/web/helpers.ex +++ b/vendor/oban_web/lib/oban/web/helpers.ex @@ -28,7 +28,7 @@ defmodule Oban.Web.Helpers do def oban_path(route, params) when is_list(route) do route - |> Enum.join("/") + |> Enum.map_join("/", &encode_segment/1) |> oban_path(params) end @@ -50,6 +50,10 @@ defmodule Oban.Web.Helpers do end end + defp encode_segment(segment) do + segment |> to_string() |> URI.encode(&URI.char_unreserved?/1) + end + @doc """ Toggle filterable params into a patch path. """ diff --git a/vendor/oban_web/lib/oban/web/live/crons/helpers.ex b/vendor/oban_web/lib/oban/web/live/crons/helpers.ex index 56317e72..6844fa5e 100644 --- a/vendor/oban_web/lib/oban/web/live/crons/helpers.ex +++ b/vendor/oban_web/lib/oban/web/live/crons/helpers.ex @@ -43,14 +43,14 @@ defmodule Oban.Web.Crons.Helpers do @doc """ Converts a NaiveDateTime to unix milliseconds, returning empty string for nil. """ - def maybe_to_unix(nil), do: "" - - def maybe_to_unix(timestamp) do + def maybe_to_unix(timestamp) when is_struct(timestamp) do timestamp |> DateTime.from_naive!("Etc/UTC") |> DateTime.to_unix(:millisecond) end + def maybe_to_unix(_timestamp), do: "" + def show_name?(%{dynamic?: true, name: name, worker: worker}), do: name != worker def show_name?(_cron), do: false end diff --git a/vendor/oban_web/lib/oban/web/live/jobs/detail_component.ex b/vendor/oban_web/lib/oban/web/live/jobs/detail_component.ex index 4be1662f..03a7695f 100644 --- a/vendor/oban_web/lib/oban/web/live/jobs/detail_component.ex +++ b/vendor/oban_web/lib/oban/web/live/jobs/detail_component.ex @@ -64,6 +64,7 @@ defmodule Oban.Web.Jobs.DetailComponent do + @@ -502,7 +503,7 @@ defmodule Oban.Web.Jobs.DetailComponent do
-
+
+ +
+
+
+
+

+ {signal_heading(@job)} +

+ <.pro_badge id="signal-pro-badge" tooltip="Awaitable signal from Oban.Pro.Worker" /> +
+ +
+
{format_signal(@job, @resolver)}
+
+
""" @@ -785,14 +813,16 @@ defmodule Oban.Web.Jobs.DetailComponent do def handle_event("save-job", params, socket) do job = socket.assigns.job - changes = - params - |> parse_edit_params(job) - |> Enum.reject(fn {_key, val} -> is_nil(val) end) - |> Map.new() + if can?(:update_jobs, socket.assigns.access) do + changes = + params + |> parse_edit_params(job) + |> Enum.reject(fn {_key, val} -> is_nil(val) end) + |> Map.new() - if map_size(changes) > 0 do - send(self(), {:update_job, job, changes}) + if map_size(changes) > 0 do + send(self(), {:update_job, job, changes}) + end end {:noreply, assign(socket, edit_changed?: false)} @@ -805,12 +835,7 @@ defmodule Oban.Web.Jobs.DetailComponent do end defp format_meta(%{meta: meta} = job, resolver) do - job = - if meta["recorded"] do - %{job | meta: Map.delete(meta, "return")} - else - job - end + job = %{job | meta: Map.drop(meta, ["return", "signal"])} Resolver.call_with_fallback(resolver, :format_job_meta, [job]) end @@ -828,6 +853,41 @@ defmodule Oban.Web.Jobs.DetailComponent do end end + defp format_signal(%{meta: %{"signal" => value}} = job, resolver) do + Resolver.call_with_fallback(resolver, :format_signal, [value, job]) + end + + defp format_signal(%{meta: %{"wait_until" => wait_until}}, _resolver) do + case wait_until_to_datetime(wait_until) do + {:ok, datetime} -> "Deadline #{Timing.datetime_to_words(datetime)}" + :infinity -> "No deadline" + :error -> "" + end + end + + defp signal_status(%{meta: %{"signal" => _}}), do: :received + defp signal_status(%{meta: %{"wait_until" => _}}), do: :awaiting + defp signal_status(_), do: :none + + defp signal_heading(job) do + case signal_status(job) do + :received -> "Received Signal" + :awaiting -> "Awaiting Signal" + :none -> "Signal" + end + end + + defp wait_until_to_datetime("infinity"), do: :infinity + + defp wait_until_to_datetime(ms) when is_integer(ms) do + case DateTime.from_unix(ms, :millisecond) do + {:ok, datetime} -> {:ok, datetime |> DateTime.to_naive() |> NaiveDateTime.truncate(:second)} + _ -> :error + end + end + + defp wait_until_to_datetime(_), do: :error + defp error_entry(assigns) do error = assigns.errors diff --git a/vendor/oban_web/lib/oban/web/live/jobs/new_component.ex b/vendor/oban_web/lib/oban/web/live/jobs/new_component.ex index 81314ab6..993d72b3 100644 --- a/vendor/oban_web/lib/oban/web/live/jobs/new_component.ex +++ b/vendor/oban_web/lib/oban/web/live/jobs/new_component.ex @@ -273,7 +273,7 @@ defmodule Oban.Web.Jobs.NewComponent do defp parse_scheduled_at(str) when is_binary(str) do case DateTime.from_iso8601(str <> ":00Z") do - {:ok, datetime, _offset} -> datetime + {:ok, datetime} -> datetime _ -> nil end end diff --git a/vendor/oban_web/lib/oban/web/pages/workflows_page.ex b/vendor/oban_web/lib/oban/web/pages/workflows_page.ex index b83dbe34..3adde7b4 100644 --- a/vendor/oban_web/lib/oban/web/pages/workflows_page.ex +++ b/vendor/oban_web/lib/oban/web/pages/workflows_page.ex @@ -211,7 +211,7 @@ defmodule Oban.Web.WorkflowsPage do ) true -> - workflows = WorkflowQuery.all_workflows(params, conf) + workflows = WorkflowQuery.all_workflows(conf, params) limit = params.limit assign(socket, diff --git a/vendor/oban_web/lib/oban/web/queries/cron_query.ex b/vendor/oban_web/lib/oban/web/queries/cron_query.ex index 6b00be25..05a607c5 100644 --- a/vendor/oban_web/lib/oban/web/queries/cron_query.ex +++ b/vendor/oban_web/lib/oban/web/queries/cron_query.ex @@ -195,7 +195,11 @@ defmodule Oban.Web.CronQuery do 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) + finished_at: + type( + fragment("COALESCE(?, ?, ?)", j.completed_at, j.cancelled_at, j.discarded_at), + :utc_datetime_usec + ) }) conf @@ -280,7 +284,10 @@ defmodule Oban.Web.CronQuery do attempted_at: o.attempted_at, scheduled_at: o.scheduled_at, finished_at: - fragment("COALESCE(?, ?, ?)", o.completed_at, o.cancelled_at, o.discarded_at) + type( + fragment("COALESCE(?, ?, ?)", o.completed_at, o.cancelled_at, o.discarded_at), + :utc_datetime_usec + ) }, select_merge: %{ rn: over(row_number(), partition_by: o.meta["cron_name"], order_by: [desc: o.id]) diff --git a/vendor/oban_web/lib/oban/web/queries/workflow_query.ex b/vendor/oban_web/lib/oban/web/queries/workflow_query.ex index 9787deee..f7168a80 100644 --- a/vendor/oban_web/lib/oban/web/queries/workflow_query.ex +++ b/vendor/oban_web/lib/oban/web/queries/workflow_query.ex @@ -126,30 +126,32 @@ defmodule Oban.Web.WorkflowQuery do end end - defmacrop sub_workflow_parent_dep(sub_workflow_id, sup_workflow_id) do + defmacrop sub_workflow_parent_dep(prefix, sub_workflow_id, sup_workflow_id) do quote do fragment( """ (SELECT dep->>1 - FROM oban_jobs j2, + FROM ?.oban_jobs j2, LATERAL jsonb_array_elements(j2.meta->'deps') AS dep WHERE j2.meta->>'workflow_id' = ? AND jsonb_typeof(dep) = 'array' AND dep->>0 = ? LIMIT 1) """, + identifier(^unquote(prefix)), unquote(sub_workflow_id), unquote(sup_workflow_id) ) end end - defmacrop sub_workflow_states(parent_id) do + defmacrop sub_workflow_states(prefix, parent_id) do quote do fragment( """ - (SELECT COALESCE(array_agg(state), '{}') FROM oban_workflows WHERE parent_id = ?) + (SELECT COALESCE(array_agg(state), '{}') FROM ?.oban_workflows WHERE parent_id = ?) """, + identifier(^unquote(prefix)), unquote(parent_id) ) end @@ -279,18 +281,19 @@ defmodule Oban.Web.WorkflowQuery do # Querying - def all_workflows(params, conf) do + def all_workflows(conf, params) do limit = Map.get(params, :limit, 10) sort_by = Map.get(params, :sort_by, "inserted") sort_dir = Map.get(params, :sort_dir, "desc") dir = String.to_existing_atom(sort_dir) + prefix = conf.prefix Workflow |> where([wf], is_nil(wf.parent_id)) |> apply_filters(params) |> apply_sort(sort_by, dir) |> limit(^limit) - |> select([wf], {wf, sub_workflow_states(wf.id)}) + |> select([wf], {wf, sub_workflow_states(prefix, wf.id)}) |> then(&Repo.all(conf, &1)) |> Enum.map(&build_workflow/1) end @@ -446,6 +449,8 @@ defmodule Oban.Web.WorkflowQuery do end defp workflow_graph_subs(conf, workflow_id) do + prefix = conf.prefix + base_query = Job |> where([j], has_sup_workflow_id(j.meta, ^workflow_id)) @@ -464,7 +469,7 @@ defmodule Oban.Web.WorkflowQuery do workflow_name: s.workflow_name, sub_name: s.sub_name, state: s.state, - parent_dep: sub_workflow_parent_dep(s.workflow_id, ^workflow_id) + parent_dep: sub_workflow_parent_dep(prefix, s.workflow_id, ^workflow_id) } ) diff --git a/vendor/oban_web/lib/oban/web/resolver.ex b/vendor/oban_web/lib/oban/web/resolver.ex index b638b2de..44c77f07 100644 --- a/vendor/oban_web/lib/oban/web/resolver.ex +++ b/vendor/oban_web/lib/oban/web/resolver.ex @@ -41,6 +41,13 @@ defmodule Oban.Web.Resolver do |> inspect(charlists: :as_lists, pretty: true) end + @impl true + def format_signal(signal, _job) do + signal + |> Oban.Web.Resolver.decode_signal() + |> inspect(charlists: :as_lists, pretty: true) + end + @impl true def jobs_query_limit(_state), do: 100_000 @@ -138,6 +145,7 @@ defmodule Oban.Web.Resolver do | :scale_queues | :stop_queues | :update_crons + | :update_jobs @type qualifier :: :args | :meta | :nodes | :queues | :tags | :workers @@ -234,6 +242,35 @@ defmodule Oban.Web.Resolver do """ @callback format_recorded(recorded :: term(), job :: Job.t()) :: iodata() + @doc """ + Customize the formatting of awaitable signal payloads wherever they are displayed. + + This callback is similar to `c:format_recorded/2`, but it accepts both the signal binary and + the job to help augment the output. + + Note that you **must decode the signal binary** prior to inspecting it. + + ## Examples + + Disable pretty printing and change the output width to 98 characters: + + def format_signal(signal, _job) do + signal + |> Oban.Web.Resolver.decode_signal() + |> inspect(pretty: false, width: 98) + end + + Decode the signal value without the `:safe` flag set, to allow decoding terms with unknown + atoms: + + def format_signal(signal, _job) do + signal + |> Oban.Web.Resolver.decode_signal([]) + |> inspect(pretty: false, width: 98) + end + """ + @callback format_signal(signal :: term(), job :: Job.t()) :: iodata() + @doc """ Extract the current user from a `Plug.Conn` when the dashboard mounts. @@ -283,6 +320,7 @@ defmodule Oban.Web.Resolver do * `:scale_queues` * `:stop_queues` * `:update_crons` + * `:update_jobs` Actions which aren't listed are considered disabled. @@ -411,6 +449,7 @@ defmodule Oban.Web.Resolver do @optional_callbacks format_job_args: 1, format_job_meta: 1, format_recorded: 2, + format_signal: 2, bulk_action_limit: 1, hint_query_limit: 1, jobs_query_limit: 1, @@ -451,6 +490,15 @@ defmodule Oban.Web.Resolver do end end + @doc """ + Decode an awaitable signal payload from a compressed, base64 binary into proper terms. + + Signals share the same wire format as recorded output, so this is a thin alias around + `decode_recorded/2` provided for documentation symmetry. + """ + @spec decode_signal(binary(), [:safe]) :: term() + def decode_signal(bin, opts \\ [:safe]), do: decode_recorded(bin, opts) + @doc false def call_with_fallback(resolver, fun, args) when is_atom(fun) and is_list(args) do resolver = @@ -482,6 +530,13 @@ defmodule Oban.Web.Resolver do |> inspect(@inspect_opts) end + @doc false + def format_signal(signal, _job) do + signal + |> decode_signal() + |> inspect(@inspect_opts) + end + @doc false def resolve_user(_conn), do: nil diff --git a/vendor/oban_web/mix.exs b/vendor/oban_web/mix.exs index 5106b58c..7d99fa91 100644 --- a/vendor/oban_web/mix.exs +++ b/vendor/oban_web/mix.exs @@ -2,7 +2,7 @@ defmodule Oban.Web.MixProject do use Mix.Project @source_url "https://github.com/oban-bg/oban_web" - @version "2.12.3" + @version "2.12.5" def project do [