denox

defmodule Denox.Run.Base do
  @moduledoc """
  Shared GenServer dispatch logic for Run modules.

  This module provides a `__using__` macro that injects GenServer boilerplate,
  a common public API, and stdout dispatch logic for both the NIF-backed
  `Denox.Run` and the CLI-backed `Denox.CLI.Run`.

  ## Public API (injected into using modules)

  **Lifecycle:** `start_link/1`, `stop/1`

  **I/O:** `send/2`, `recv/2`, `send_and_recv/3`

  **Pub/Sub:** `subscribe/1`, `unsubscribe/1`

  **Introspection:** `alive?/1`, `os_pid/1`

  **Convenience (one-shot):**
    - `capture/1` — start, collect all lines, stop; returns `{:ok, [String.t()]}`
    - `stream/1` — start + lazy `Stream` (auto-stops on halt)
    - `stream_from/2` — lazy `Stream` from an existing PID (caller manages lifecycle)
    - `with_runtime/2` — start + user function + guaranteed stop via `after`

  ## Implementing a Backend

  Modules that `use Denox.Run.Base, backend: :my_backend` must implement four
  callbacks:

    * `c:init_backend/1` — start the backend (NIF resource or Port)
    * `c:send_backend/2` — write data to the backend's stdin
    * `c:stop_backend/1` — shut down the backend
    * `c:alive_backend?/1` — check if the backend is still running

  See `Denox.Run` (NIF backend) and `Denox.CLI.Run` (subprocess backend) for
  concrete implementations.
  """

  @callback init_backend(keyword()) ::
              {:ok, backend_state :: term()} | {:error, term()}

  @callback send_backend(backend_state :: term(), String.t()) ::
              :ok | {:error, term()}

  @callback stop_backend(backend_state :: term()) :: :ok

  @callback alive_backend?(backend_state :: term()) :: boolean()

  @doc """
  Resolve a specifier to a form suitable for the Deno runtime.

  ## Rules

    * Specifiers already prefixed with `npm:`, `jsr:`, `http://`, `https://`,
      or `file://` are passed through unchanged.
    * Scoped package names starting with `@` are prefixed with `npm:`.
    * Everything else is returned as-is (treated as a file path by the backend).

  ## Examples

      iex> Denox.Run.Base.resolve_specifier("npm:cowsay")
      "npm:cowsay"

      iex> Denox.Run.Base.resolve_specifier("@scope/pkg")
      "npm:@scope/pkg"

      iex> Denox.Run.Base.resolve_specifier("server.ts")
      "server.ts"
  """
  @spec resolve_specifier(String.t()) :: String.t()
  def resolve_specifier(spec) do
    cond do
      String.starts_with?(spec, ["npm:", "jsr:", "http://", "https://", "file://"]) -> spec
      String.starts_with?(spec, "@") -> "npm:#{spec}"
      true -> spec
    end
  end

  defmacro __using__(opts) do
    backend_type = Keyword.fetch!(opts, :backend)

    # credo:disable-for-next-line Credo.Check.Refactor.LongQuoteBlocks
    quote location: :keep do
      @behaviour Denox.Run.Base

      use GenServer

      require Logger

      defstruct [
        :backend_state,
        :package,
        :file,
        :exit_status,
        subscribers: [],
        recv_waiters: :queue.new(),
        stdout_buffer: :queue.new()
      ]

      @type t :: %__MODULE__{}

      @backend_type unquote(backend_type)

      # --- Public API ---

      @doc """
      Start a managed Deno runtime.

      ## Options

        - `:package` - JSR/npm package specifier
        - `:file` - local file path to run
        - `:permissions` - permission mode; defaults to `:none` (deny all) when omitted:
          - `:all` — allow all permissions (`-A` in Deno CLI)
          - `:none` — deny all permissions (Deno's default behaviour)
          - keyword list — granular permissions, e.g. `[allow_net: true, allow_read: ["/tmp"],
            deny_env: true]`. Supports both `allow_*` and `deny_*` keys.
        - `:env` - map of environment variables
        - `:args` - extra arguments after the specifier
        - `:name` - GenServer name for registration
        - `:buffer_size` - (NIF backend only) stdout channel capacity in lines;
          controls how many lines the NIF can buffer before applying backpressure.
          Range `[1, 100_000]`; `0` or omitted uses the default of `1024` lines.
      """
      @spec start_link(keyword()) :: GenServer.on_start()
      def start_link(opts) do
        gen_opts = Keyword.take(opts, [:name])
        GenServer.start_link(__MODULE__, opts, gen_opts)
      end

      @doc """
      Send data to stdin of the running process.

      A newline (`\\n`) is automatically appended if `data` does not
      already end with one.

      Returns `:ok` on success, or `{:error, :closed}` if the process
      has already exited.
      """
      @spec send(GenServer.server(), String.t()) :: :ok | {:error, term()}
      def send(server, data) do
        GenServer.call(server, {:send, data})
      end

      @doc """
      Receive the next line from stdout.

      ## Options
        - `:timeout` - milliseconds to wait (default: 5000)
      """
      @spec recv(GenServer.server(), keyword()) ::
              {:ok, String.t()} | {:error, :timeout | :closed}
      def recv(server, opts \\ []) do
        timeout = Keyword.get(opts, :timeout, 5000)
        # Add a generous buffer so the GenServer always replies (via its internal
        # timer) before the outer GenServer.call times out, ensuring stale waiters
        # are never left in recv_waiters.
        GenServer.call(server, {:recv, timeout}, timeout + 1000)
      end

      @doc """
      Send data to stdin and wait for the next stdout line.

      A convenience wrapper around `send/2` + `recv/2` for the common
      request-response pattern (e.g. JSON-RPC over stdio, MCP servers).

      Returns `{:ok, response_line}` or `{:error, reason}` where reason is
      `:timeout`, `:closed`, or the send error.

      ## Options

        - `:timeout` - milliseconds to wait for a response (default: 5000)

      ## Example

          request = Jason.encode!(%{jsonrpc: "2.0", method: "ping", id: 1})
          {:ok, response_line} = Denox.Run.send_and_recv(pid, request, timeout: 5000)
          response = Jason.decode!(response_line)

      """
      @spec send_and_recv(GenServer.server(), String.t(), keyword()) ::
              {:ok, String.t()} | {:error, term()}
      def send_and_recv(server, data, opts \\ []) do
        case __MODULE__.send(server, data) do
          :ok -> recv(server, opts)
          {:error, _} = error -> error
        end
      end

      @doc """
      Stream stdout lines from an already-running server as a lazy `Stream`.

      Unlike `stream/1`, which starts a new runtime, `stream_from/2` works
      with an existing server PID. The stream halts when the runtime exits or
      a per-line timeout is reached. The server is **not** stopped when the
      stream halts — the caller retains ownership.

      This pairs naturally with `with_runtime/2` when you want lazy output
      enumeration after sending input:

          Denox.Run.with_runtime([file: "script.ts", permissions: :all], fn pid ->
            :ok = Denox.Run.send(pid, Jason.encode!(request))
            Denox.Run.stream_from(pid) |> Enum.to_list()
          end)

      ## Options

        - `:timeout` - per-line receive timeout in milliseconds (default: `5000`).
          If no new line arrives within this window, the stream halts.

      """
      @spec stream_from(GenServer.server(), keyword()) :: Enumerable.t()
      def stream_from(server, opts \\ []) do
        timeout = Keyword.get(opts, :timeout, 5_000)

        Stream.resource(
          fn -> server end,
          fn pid ->
            case recv(pid, timeout: timeout) do
              {:ok, line} -> {[line], pid}
              {:error, :closed} -> {:halt, pid}
              {:error, :timeout} -> {:halt, pid}
            end
          end,
          fn _pid -> :ok end
        )
      end

      @doc """
      Execute a function with a managed runtime, ensuring cleanup on exit.

      Starts a runtime with `opts`, calls `fun` with the server PID, then
      stops the runtime — even if `fun` raises an exception.

      Returns the value returned by `fun`. If `fun` returns an error tuple,
      that tuple is passed through as-is. Returns `{:error, reason}` if the
      runtime failed to start.

      ## Example

          Denox.Run.with_runtime([file: "scripts/tool.ts", permissions: :all], fn pid ->
            :ok = Denox.Run.send(pid, Jason.encode!(%{method: "init"}))
            {:ok, line} = Denox.Run.recv(pid, timeout: 10_000)
            Jason.decode!(line)
          end)

      """
      @spec with_runtime(keyword(), (GenServer.server() -> result)) ::
              result | {:error, term()}
            when result: term()
      def with_runtime(opts, fun) when is_list(opts) and is_function(fun, 1) do
        case start_link(opts) do
          {:ok, pid} ->
            try do
              fun.(pid)
            after
              stop(pid)
            end

          {:error, reason} ->
            {:error, reason}
        end
      end

      @doc """
      Subscribe the calling process to stdout messages.

      After subscribing, the calling process will receive:

        - `{:denox_run_stdout, server_pid, line}` for each stdout line
        - `{:denox_run_exit, server_pid, exit_status}` when the process exits

      If the subscribing process dies, it is automatically removed from the
      subscriber list without needing an explicit `unsubscribe/1` call.
      """
      @spec subscribe(GenServer.server()) :: :ok
      def subscribe(server) do
        GenServer.call(server, {:subscribe, self()})
      end

      @doc """
      Unsubscribe from stdout messages.

      The calling process will stop receiving `{:denox_run_stdout, ...}` and
      `{:denox_run_exit, ...}` messages from this server.
      """
      @spec unsubscribe(GenServer.server()) :: :ok
      def unsubscribe(server) do
        GenServer.call(server, {:unsubscribe, self()})
      end

      @doc "Check if the runtime is still running."
      @spec alive?(GenServer.server()) :: boolean()
      def alive?(server) do
        GenServer.call(server, :alive?)
      end

      @doc """
      Return the OS PID of the process, if available.

      Returns `{:ok, pid}` for CLI-backed runtimes or `{:error, :not_available}`
      for NIF-backed runtimes where no separate OS process exists.
      """
      @spec os_pid(GenServer.server()) ::
              {:ok, non_neg_integer()} | {:error, :not_available | :not_running}
      def os_pid(server) do
        GenServer.call(server, :os_pid)
      end

      @doc "Stop the runtime."
      @spec stop(GenServer.server()) :: :ok
      def stop(server) do
        GenServer.stop(server, :normal)
      end

      @doc """
      Run a script and capture all stdout output.

      Starts a managed runtime, collects all output lines until the script
      exits or the per-line timeout is reached, then stops the runtime.

      Returns `{:ok, lines}` where `lines` is a list of stdout lines in order,
      or `{:error, reason}` if the runtime failed to start.

      ## Options

      Same as `start_link/1`, plus:

        - `:timeout` - per-line receive timeout in milliseconds (default: `30_000`).
          If no new line arrives within this window, collection stops and the
          lines collected so far are returned.

      ## Example

          {:ok, lines} = Denox.Run.capture(
            file: "scripts/generate.ts",
            permissions: :all
          )
          # lines => ["line 1", "line 2", ...]

      """
      @spec capture(keyword()) :: {:ok, [String.t()]} | {:error, term()}
      def capture(opts) do
        timeout = Keyword.get(opts, :timeout, 30_000)
        start_opts = Keyword.delete(opts, :timeout)

        case start_link(start_opts) do
          {:ok, pid} ->
            lines = do_capture_loop(pid, [], timeout)
            stop(pid)
            {:ok, lines}

          {:error, reason} ->
            {:error, reason}
        end
      end

      defp do_capture_loop(pid, acc, timeout) do
        case recv(pid, timeout: timeout) do
          {:ok, line} -> do_capture_loop(pid, [line | acc], timeout)
          {:error, :closed} -> Enum.reverse(acc)
          {:error, :timeout} -> Enum.reverse(acc)
        end
      end

      @doc """
      Stream stdout lines from a script as a lazy `Stream`.

      Starts a managed runtime and returns an `Enumerable` that yields
      stdout lines one at a time. The runtime is automatically stopped
      when the stream is fully consumed or halted (e.g. via `Stream.take/2`
      or `Enum.take/2`).

      This is useful for processing large outputs without buffering all
      lines in memory, or for early termination once a condition is met.

      ## Options

      Same as `start_link/1`, plus:

        - `:timeout` - per-line receive timeout in milliseconds (default: `30_000`).
          If no new line arrives within this window, the stream halts.

      ## Example

          # Collect the first 5 lines of a long-running script
          Denox.Run.stream(file: "server.ts", permissions: :all)
          |> Enum.take(5)

          # Filter lines matching a pattern
          Denox.Run.stream(file: "generate.ts", permissions: :all)
          |> Stream.filter(&String.contains?(&1, "ERROR"))
          |> Enum.to_list()

      > #### Errors on start {: .warning}
      >
      > If the runtime fails to start, the stream raises a `RuntimeError`
      > when enumerated. Use `capture/1` if you prefer an `{:error, reason}`
      > tuple instead.

      """
      @spec stream(keyword()) :: Enumerable.t()
      def stream(opts) do
        timeout = Keyword.get(opts, :timeout, 30_000)
        start_opts = Keyword.delete(opts, :timeout)

        Stream.resource(
          fn ->
            case start_link(start_opts) do
              {:ok, pid} -> {:running, pid}
              {:error, reason} -> {:failed, reason}
            end
          end,
          fn
            {:failed, reason} ->
              raise RuntimeError,
                message: "Denox.Run failed to start: #{inspect(reason)}"

            {:running, pid} = state ->
              case recv(pid, timeout: timeout) do
                {:ok, line} -> {[line], state}
                {:error, :closed} -> {:halt, state}
                {:error, :timeout} -> {:halt, state}
              end
          end,
          fn
            {:failed, _} -> :ok
            {:running, pid} -> stop(pid)
          end
        )
      end

      # --- GenServer callbacks ---

      @spec init(keyword()) :: {:ok, t()} | {:stop, term()}
      @impl true
      def init(opts) do
        package = Keyword.get(opts, :package)
        file = Keyword.get(opts, :file)

        if is_nil(package) and is_nil(file) do
          # Return :normal so exit(:normal) is called, which does not
          # propagate to linked processes as an abnormal exit signal.
          {:stop, :normal}
        else
          case init_backend(opts) do
            {:ok, backend_state} ->
              state = %__MODULE__{
                backend_state: backend_state,
                package: package,
                file: file
              }

              :telemetry.execute(
                [:denox, :run, :start],
                %{system_time: System.system_time()},
                %{package: package, file: file, backend: @backend_type}
              )

              {:ok, state}

            {:error, reason} ->
              {:stop, reason}
          end
        end
      end

      @spec handle_call(term(), GenServer.from(), t()) ::
              {:reply, term(), t()} | {:noreply, t()}
      @impl true
      def handle_call(msg, from, state) do
        Denox.Run.Base.__handle_call__(__MODULE__, msg, from, state)
      end

      @impl true
      def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
        # Remove subscriber with this monitor ref
        subscribers = Enum.reject(state.subscribers, fn {_p, sref} -> sref == ref end)

        # Remove any recv_waiters with this monitor ref, cancelling their timers to
        # suppress stale :recv_timeout messages (4-tuple: {from, mon_ref, timeout_ref, timer_ref}).
        waiters =
          :queue.filter(
            fn {_from, wref, _tref, timer_ref} ->
              if wref == ref, do: Process.cancel_timer(timer_ref)
              wref != ref
            end,
            state.recv_waiters
          )

        {:noreply, %{state | subscribers: subscribers, recv_waiters: waiters}}
      end

      def handle_info({:recv_timeout, timeout_ref}, state) do
        Denox.Run.Base.__handle_recv_timeout__(timeout_ref, state)
      end

      def handle_info(_msg, state) do
        {:noreply, state}
      end

      @impl true
      def terminate(_reason, %{exit_status: nil} = state) do
        stop_backend(state.backend_state)
        :ok
      end

      def terminate(_reason, _state), do: :ok

      # --- Shared dispatch helpers ---

      @doc false
      @spec dispatch_line(String.t(), t()) :: t()
      def dispatch_line(line, state) do
        :telemetry.execute(
          [:denox, :run, :recv],
          %{system_time: System.system_time()},
          %{line_bytes: byte_size(line), backend: @backend_type}
        )

        for {pid, _ref} <- state.subscribers do
          Kernel.send(pid, {:denox_run_stdout, self(), line})
        end

        case :queue.out(state.recv_waiters) do
          {{:value, {from, mon_ref, _tref, timer_ref}}, rest} ->
            Process.cancel_timer(timer_ref)
            Process.demonitor(mon_ref, [:flush])
            GenServer.reply(from, {:ok, line})
            %{state | recv_waiters: rest}

          {:empty, _} ->
            %{state | stdout_buffer: :queue.in(line, state.stdout_buffer)}
        end
      end

      @doc false
      @spec handle_exit(non_neg_integer(), t()) :: t()
      def handle_exit(status, state) do
        state = %{state | exit_status: status}

        for {pid, _ref} <- state.subscribers do
          Kernel.send(pid, {:denox_run_exit, self(), status})
        end

        state = drain_waiters(state)

        :telemetry.execute(
          [:denox, :run, :stop],
          %{system_time: System.system_time()},
          %{
            package: state.package,
            file: state.file,
            exit_status: status,
            backend: @backend_type
          }
        )

        state
      end

      defp drain_waiters(state) do
        case :queue.out(state.recv_waiters) do
          {{:value, {from, mon_ref, _tref, timer_ref}}, rest} ->
            Process.cancel_timer(timer_ref)
            Process.demonitor(mon_ref, [:flush])
            GenServer.reply(from, {:error, :closed})
            drain_waiters(%{state | recv_waiters: rest})

          {:empty, _} ->
            state
        end
      end

      defoverridable handle_call: 3, handle_info: 2, terminate: 2
    end
  end

  # Shared handle_call implementation, called from the using module.
  # This avoids the deprecated `super` pattern.
  @doc false
  def __handle_call__(_module, {:send, data}, _from, %{exit_status: nil} = state) do
    data = if String.ends_with?(data, "\n"), do: data, else: data <> "\n"
    result = state.__struct__.send_backend(state.backend_state, data)
    {:reply, result, state}
  end

  def __handle_call__(_module, {:send, _data}, _from, state) do
    {:reply, {:error, :closed}, state}
  end

  def __handle_call__(_module, {:recv, timeout}, from, %{stdout_buffer: buffer} = state) do
    case :queue.out(buffer) do
      {{:value, line}, rest} ->
        {:reply, {:ok, line}, %{state | stdout_buffer: rest}}

      {:empty, _} ->
        if state.exit_status != nil do
          {:reply, {:error, :closed}, state}
        else
          {pid, _tag} = from
          mon_ref = Process.monitor(pid)
          timeout_ref = make_ref()
          timer_ref = Process.send_after(self(), {:recv_timeout, timeout_ref}, timeout)
          waiters = :queue.in({from, mon_ref, timeout_ref, timer_ref}, state.recv_waiters)
          {:noreply, %{state | recv_waiters: waiters}}
        end
    end
  end

  def __handle_call__(_module, {:subscribe, pid}, _from, state) do
    if Enum.any?(state.subscribers, fn {p, _ref} -> p == pid end) do
      {:reply, :ok, state}
    else
      ref = Process.monitor(pid)
      {:reply, :ok, %{state | subscribers: [{pid, ref} | state.subscribers]}}
    end
  end

  def __handle_call__(_module, {:unsubscribe, pid}, _from, state) do
    {removed, kept} = Enum.split_with(state.subscribers, fn {p, _ref} -> p == pid end)
    Enum.each(removed, fn {_p, ref} -> Process.demonitor(ref, [:flush]) end)
    {:reply, :ok, %{state | subscribers: kept}}
  end

  def __handle_call__(module, :alive?, _from, state) do
    alive = state.exit_status == nil and module.alive_backend?(state.backend_state)
    {:reply, alive, state}
  end

  def __handle_call__(
        _module,
        :os_pid,
        _from,
        %{backend_state: %{os_pid: os_pid}, exit_status: nil} = state
      ) do
    {:reply, {:ok, os_pid}, state}
  end

  def __handle_call__(_module, :os_pid, _from, %{exit_status: status} = state)
      when not is_nil(status) do
    {:reply, {:error, :not_running}, state}
  end

  # Backend without os_pid (e.g., NIF-backed Denox.Run)
  def __handle_call__(_module, :os_pid, _from, state) do
    {:reply, {:error, :not_available}, state}
  end

  def __handle_call__(_module, msg, _from, state) do
    {:reply, {:error, {:unknown_call, msg}}, state}
  end

  @doc false
  def __handle_recv_timeout__(timeout_ref, state) do
    # Find the waiter with this timeout_ref and reply :timeout, removing it.
    # If not found, the line already arrived and consumed the waiter — do nothing.
    waiter_list = :queue.to_list(state.recv_waiters)

    {matching, rest} =
      Enum.split_with(waiter_list, fn {_from, _mref, tref, _tref2} -> tref == timeout_ref end)

    case matching do
      [] ->
        {:noreply, state}

      [{from, mon_ref, _tref, _timer_ref} | _] ->
        Process.demonitor(mon_ref, [:flush])
        GenServer.reply(from, {:error, :timeout})
        {:noreply, %{state | recv_waiters: :queue.from_list(rest)}}
    end
  end
end