defmodule UI.Queue do @moduledoc """ Priority queue for UI operations to ensure proper serialization of output and user interactions. ## What is a UI Context? A UI context is a logical grouping of related UI operations that should execute together without interruption from other UI operations. For example, a confirmation dialog that shows information, asks a question, and displays the result should all execute as one atomic unit. Each UI context has a unique token that allows UI operations to either: - Execute immediately (if called from within the same context) - Queue for later execution (if called from outside any context) ## Context Wrappers for Interactive UI Use these wrapper functions when calling interactive UI functions from contexts that could cause deadlocks: - `run_from_genserver/1` - For GenServer callbacks (starts fresh UI context) - `run_from_task/1` - For Services.Globals.Spawn.async (preserves parent UI context) **Interactive UI functions that require wrapping:** - `UI.confirm/1` - `UI.choose/2` - `UI.prompt/1` **GenServer example:** def handle_call(:confirm_delete, _from, state) do result = UI.Queue.run_from_genserver(fn -> UI.confirm("Delete this item?") end) {:reply, result, state} end **Services.Globals.Spawn.async example:** task = Services.Globals.Spawn.async(fn -> UI.Queue.run_from_task(fn -> UI.confirm("Process this item?") end) end) result = Task.await(task) Non-interactive UI functions (`UI.info/2`, `UI.error/2`, etc.) can be called directly. """ use GenServer require Logger # ---------------------------------------------------------------------------- # Client API # ---------------------------------------------------------------------------- def start_link(opts \\ []) do init_opts = Keyword.drop(opts, [:name]) with {:ok, pid} <- GenServer.start_link(__MODULE__, :ok, init_opts) do Services.Instance.register(__MODULE__, pid) {:ok, pid} end end @doc """ The queue pid serving the current process tree. Public because some call sites (UI.Output.Production, tests) pass the server explicitly - the positional optional `server` argument makes partial calls ambiguous. """ @spec instance() :: pid() def instance(), do: Services.Instance.fetch!(__MODULE__) @spec default_server() :: pid() defp default_server(), do: Services.Instance.fetch!(__MODULE__) # Normal output (fast-path if in interaction context) def puts(server \\ default_server(), io_device \\ :stdio, data, timeout \\ :infinity) do if in_ctx?(server) do exec({:puts, io_device, data}) else GenServer.call(server, {:puts, io_device, data}, timeout) end end # Logger proxy (fast-path if in interaction context) def log(server \\ default_server(), level, chardata, md \\ [], timeout \\ :infinity) do GenServer.call(server, {:log, level, chardata, md}, timeout) end # Interactive (priority). If already in context, run inline - but FIRST # tell the GenServer to pause logs so concurrent {:log, ...} calls don't # scribble over the active prompt. Without this, the fast-path leaves the # GenServer happily processing log messages while the inline interaction # blocks on stdin in a different process. # # One exception: when the caller IS the queue process. The queued handler # (handle_call({:interact, ...})) executes funs inline via exec/1, which # tokens the queue's own pdict - so a nested interact from within a queued # fun lands here with in_ctx? true. No pause is needed (the GenServer can't # process {:log, ...} calls while it's busy executing the fun), and a # GenServer.call back into ourselves would crash with :calling_self. def interact(server \\ default_server(), fun, timeout \\ :infinity) when is_function(fun, 0) do cond do not in_ctx?(server) -> GenServer.call(server, {:interact, fun}, timeout) to_pid(server) == self() -> exec({:interact, fun}) true -> :ok = pause_logs(server, timeout) try do exec({:interact, fun}) after :ok = unpause_logs(server, timeout) end end end # Increment the GenServer's interactive-context depth. While depth > 0, # incoming {:log, ...} calls are buffered. Synchronous so the caller has a # happens-before relationship with subsequent log calls. Safe to nest from # multiple processes; only the last unpause flushes. @doc false def pause_logs(server \\ default_server(), timeout \\ :infinity), do: GenServer.call(server, :pause_logs, timeout) @doc false def unpause_logs(server \\ default_server(), timeout \\ :infinity), do: GenServer.call(server, :unpause_logs, timeout) # ---------------------------------------------------------------------------- # Context utilities for spawned processes # ---------------------------------------------------------------------------- def interaction_token(server \\ default_server()) do Process.get(pd_key(server)) end def bind(server \\ default_server(), token, fun) when is_function(fun, 0) do Process.put(pd_key(server), token) try do fun.() after Process.delete(pd_key(server)) end end @doc """ Wrapper for running interactive UI calls from GenServer callbacks. This **starts a fresh UI interaction context**, which prevents deadlocks when GenServer callbacks need to make interactive UI calls that could send messages back to the same GenServer. Use this for interactive functions like `UI.confirm/1`, `UI.choose/2`, and `UI.prompt/1` when called from `handle_call/3`, `handle_cast/2`, or `handle_info/2`. ## Example def handle_call(:confirm_delete, _from, state) do result = UI.Queue.run_from_genserver(fn -> UI.confirm("Delete this item?") end) {:reply, result, state} end """ def run_from_genserver(server \\ default_server(), fun) when is_function(fun, 0) do bind(server, make_ref(), fun) end @doc """ Wrapper for running interactive UI calls from async tasks (Services.Globals.Spawn.async). This **preserves the parent process's UI interaction context**, allowing async tasks to participate in the same UI interaction as their parent. This is essential when tasks need to make interactive UI calls as part of an ongoing user interaction. Use this when spawning tasks that might make interactive UI calls and you want them to be part of the current user interaction session. ## Example # In a function that's already in a UI interaction context task = Services.Globals.Spawn.async(fn -> UI.Queue.run_from_task(fn -> # This UI call will be part of the parent's interaction UI.confirm("Process this item?") end) end) result = Task.await(task) """ def run_from_task(server \\ default_server(), fun) when is_function(fun, 0) do parent_token = interaction_token(server) bind(server, parent_token, fun) end def spawn_bound(server \\ default_server(), fun) when is_function(fun, 0) do tok = interaction_token(server) Services.Globals.Spawn.spawn(fn -> bind(server, tok, fun) end) end # ---------------------------------------------------------------------------- # GenServer callbacks # ---------------------------------------------------------------------------- @impl true def init(:ok) do state = %{ busy: false, hq: :queue.new(), q: :queue.new(), paused_logs: false, # Tracks concurrent fast-path interact callers (pause_logs / unpause_logs). # paused_logs derives from interact_depth > 0; we keep the explicit # boolean only so the existing handle_call(:log, ...) clauses stay # cheap pattern matches. interact_depth: 0, log_buffer: :queue.new() } {:ok, state} end # INTERACT @impl true def handle_call({:interact, fun}, from, %{busy: false} = st) do # mark busy and pause logging st1 = %{st | busy: true, paused_logs: true} result = exec({:interact, fun}) # Only restore paused_logs if no fast-path callers are still holding the # pause. Otherwise their unpause_logs/0 will flush when the count hits 0. st2 = if st1.interact_depth == 0 do flush_log_buffer(%{st1 | paused_logs: false}) else st1 end GenServer.reply(from, result) {:noreply, drain(st2)} end def handle_call({:interact, fun}, from, %{busy: true} = st) do {:noreply, enqueue(:hq, {from, {:interact, fun}}, st)} end # FAST-PATH PAUSE / UNPAUSE — used by interact/3's in-context branch so the # GenServer buffers concurrent {:log, ...} calls while another process holds # an interactive context. def handle_call(:pause_logs, _from, %{interact_depth: n} = st) do {:reply, :ok, %{st | interact_depth: n + 1, paused_logs: true}} end def handle_call(:unpause_logs, _from, %{interact_depth: 1} = st) do st2 = flush_log_buffer(%{st | interact_depth: 0, paused_logs: false}) {:reply, :ok, st2} end def handle_call(:unpause_logs, _from, %{interact_depth: n} = st) when n > 1 do {:reply, :ok, %{st | interact_depth: n - 1}} end # Defensive: extra unpause without matching pause. Don't crash, just leave # state alone. Indicates a code bug somewhere but isn't user-facing damage. def handle_call(:unpause_logs, _from, st) do {:reply, :ok, st} end # PUTS def handle_call({:puts, dev, data}, from, %{busy: false} = st) do st = %{st | busy: true} result = exec({:puts, dev, data}) GenServer.reply(from, result) {:noreply, drain(st)} end def handle_call({:puts, dev, data}, from, %{busy: true} = st) do {:noreply, enqueue(:q, {from, {:puts, dev, data}}, st)} end # LOG def handle_call({:log, level, chardata, md}, _from, %{paused_logs: true} = st) do st2 = buffer_log(st, {level, chardata, md}) {:reply, :ok, st2} end def handle_call({:log, level, chardata, md}, from, %{busy: false} = st) do st = %{st | busy: true} result = exec({:log, level, chardata, md}) GenServer.reply(from, result) {:noreply, drain(st)} end def handle_call({:log, level, chardata, md}, from, %{busy: true} = st) do {:noreply, enqueue(:q, {from, {:log, level, chardata, md}}, st)} end @impl true def handle_info(_msg, st), do: {:noreply, st} # ---------------------------------------------------------------------------- # Internals # ---------------------------------------------------------------------------- defp enqueue(:hq, item, %{hq: q} = st), do: %{st | hq: :queue.in(item, q)} defp enqueue(:q, item, %{q: q} = st), do: %{st | q: :queue.in(item, q)} defp buffer_log(st, {level, chardata, md}) do new_buffer = :queue.in({level, chardata, md}, st.log_buffer) %{st | log_buffer: new_buffer} end defp flush_log_buffer(st) do # Unpause logging and replay buffered entries case :queue.out(st.log_buffer) do {{:value, {level, chardata, md}}, buffer} -> _ = exec({:log, level, chardata, md}) flush_log_buffer(%{st | log_buffer: buffer}) {:empty, _} -> %{st | log_buffer: :queue.new()} end end defp drain(%{busy: true} = st) do case take_next(st) do {:none, st2} -> %{st2 | busy: false} {{from, job}, st2} -> result = exec(job) GenServer.reply(from, result) drain(st2) end end defp take_next(%{hq: hq} = st) do case :queue.out(hq) do {{:value, item}, hq2} -> {item, %{st | hq: hq2}} {:empty, _} -> case :queue.out(st.q) do {{:value, item}, q2} -> {item, %{st | q: q2}} {:empty, _} -> {:none, st} end end end # ---------------------------------------------------------------------------- # Execution with tokened interaction context # ---------------------------------------------------------------------------- defp exec({:interact, fun}) do token = make_ref() # key by server pid put_token(self(), token) try do {:ok, fun.()} rescue e -> {:error, {e, __STACKTRACE__}} catch kind, val -> {:error, {kind, val}} after del_token(self()) end end defp exec({:puts, dev, data}) do try do case dev do :stdio -> # Owl expects Owl.Data.t() and will convert it via # Owl.Data.to_chardata/1. Owl.IO.puts(data) UI.Tee.write(data) _ -> IO.puts(dev, sanitize_chardata(data)) end :ok rescue e -> {:error, {e, __STACKTRACE__}} end end defp exec({:log, level, chardata, md}) do try do chardata = sanitize_chardata(chardata) Logger.log(level, chardata, md) :ok rescue e -> {:error, {e, __STACKTRACE__}} end end # ---------------------------------------------------------------------------- # PD helpers (keyed by server pid) # ---------------------------------------------------------------------------- defp to_pid(server) when is_pid(server), do: server defp to_pid(server) do case Process.whereis(server) do nil -> server pid -> pid end end defp pd_key(server), do: {:uiq_ctx, to_pid(server)} defp in_ctx?(server), do: Process.get(pd_key(server)) != nil defp put_token(server, token), do: Process.put(pd_key(server), token) defp del_token(server), do: Process.delete(pd_key(server)) # Sanitize any chardata or input into valid UTF-8 binary defp sanitize_chardata(input) do binary = cond do is_binary(input) -> input is_list(input) -> try do IO.iodata_to_binary(input) rescue _ -> inspect(input) end true -> inspect(input) end String.replace_invalid(binary, "�") end end