ci

defmodule OsCmd do
  @moduledoc """
  Managed execution of external commands.

  This module provides similar functionality to `System.cmd/3`, with the difference that the
  execution of commands is managed, which provides the following benefits:

  1. The external OS process is logically linked to the parent BEAM process (the process which
     started it). If the parent process terminates, the OS process will be taken down.
  2. The external OS process will also be taken down if the entire BEAM instance goes down.
  3. Support for timeout-based and manual termination of the OS process.
  4. Polite-first termination of the OS process (SIGTERM followed by SIGKILL) in the spirit of
     OTP termination (shutdown strategies).

  In this regard, `OsCmd` is similar to [erlexec](http://saleyn.github.io/erlexec/) and
  [porcelain](https://github.com/alco/porcelain), though it doesn't have all the features of those
  projects.

  For usage details, see `start_link/1`.
  """

  defmodule Error do
    @moduledoc "Error struct returned by various OsCmd operations."

    @type t :: %__MODULE__{message: String.t(), exit_status: term, output: String.t()}

    defexception [:message, :exit_status, :output]
  end

  use GenServer, shutdown: :infinity
  alias OsCmd.Faker

  @type start_opt ::
          {:name, GenServer.name()}
          | {:handler, handler}
          | {:timeout, pos_integer() | :infinity}
          | {:cd, String.t()}
          | {:env, [{String.t() | atom, String.t() | nil}]}
          | {:pty, boolean}
          | {:propagate_exit?, boolean}
          | {:terminate_cmd, String.t()}

  @type handler :: (event -> any) | {acc, (event, acc -> acc)}
  @type acc :: any

  @type event ::
          :starting
          | {:output, output}
          | {:stopped, exit_status}

  @type mock ::
          String.t()
          | (command :: String.t(), [start_opt] -> {:ok, output} | {:error, Error.t()})

  @type output :: String.t()
  @type exit_status :: non_neg_integer() | (exit_reason :: :timeout | any)

  @doc """
  Starts the command owner process.

  The owner process will start the command, handle its events, and stop when the command finishes.
  The started process will synchronously stop the command during while being terminated. The owner
  process never stops before the command finishes (unless the owner process is forcefully
  terminated with the reason `:kill`), which makes it safe to run under a supervisor or `Parent`.

  The command is a "free-form string" in the shape of: `"command arg1 arg2 ..."`. The command has
  to be an executable that exists in standard search paths.

  Args can be separated by one or more whitespaces, tabs, or newline characters. If any arg has
  to contain whitespace characters, you can encircle it in double or single quotes. Inside the
  quoted argument, you can use `\\"` or `\\'` to inject the quote character, and `\\\\` to inject
  the backslash character.

  Examples:

      OsCmd.start_link("echo 1")`

      OsCmd.start_link(~s/
        some_cmd
          arg1
          "arg \\" \\\\ 2"
          'arg \\' \\\\ 3'
      /)

  Due to support for free-form execution, it is possible to execute complex scripts, by starting
  the shell process

      OsCmd.start_link(~s/bash -c "..."/)

  However, this is usually not advised, because `OsCmd` can't keep all of its guarantees. Any
  child process started inside the shell is not guaranteed to be terminated before the owner
  process stops, and some of them might even linger on forever.

  ## Options

  You can pass additional options by invoking `OsCmd.start_link({cmd, opts})`. The following
  options are supported:

    - `:cd` - Folder in which the command will be started
    - `:env`- OS environment variables which will be set in the command's own environment. Note
      that the command OS process inherits the environment from the BEAM process. If you want to
      unset some of the inherited variables, you can include `{var_to_unset, nil}` in this list.
    - `:pty` - If set to `true`, the command will be started with a pseudo-terminal interface.
      If the OS doesn't support pseudo-terminal, this flag is ignored. Defaults to `false`.
    - `:timeout` - The duration after which the command will be automatically terminated.
      Defaults to `:infinity`. If the command is timed out, the process will exit with the reason
      `:timeout`, irrespective of the `propagate_exit?` setting.
    - `propagate_exit?` - When set to `true` and the exit reason of the command is not zero, the
      process will exit with `{:failed, exit_status}`. Otherwise, the process will always exit
      with the reason `:normal` (unless the command times out).
    - `terminate_cmd` - Custom command to use in place of SIGTERM when stopping the OS process.
      See the "Command termination" section for details.
    - `handler` - Custom event handler. See the "Event handling" section for details.
    - `name` - Registered name of the process. If not provided, the process won't be registered.

  ## Event handling

  During the lifetime of the command, the following events are emitted:

    - `:starting` - the command is being started
    - `{:output, output}` - stdout or stderr output
    - `{:stopped, exit_status}` - the command stopped with the given exit status

  You can install multiple custom handlers to deal with these events. By default, no handler is
  created, which means that the command is executed silently. Handlers are functions which are
  executed inside the command owner process. Handlers can be stateless or stateful.

  A stateless handler is a function in the shape of `fun(event) -> ... end`. This function holds
  no state, so its result is ignored. A stateful handler can be specified as
  `{initial_acc, fun(event, acc) -> next_acc end}`. You can provide multiple handlers, and they
  don't have to be of the same type.

  Since handlers are executed in the owner process, an unhandled exception in the handler will
  lead to process termination. The OS process will be properly taken down before the owner process
  stops, but no additional event (including the `:stopped` event) will be fired.

  It is advised to minimize the logic inside handlers. Handlers are best suited for very simple
  tasks such as logging or notifying other processes.

  ### Output

  The output events will contain output fragments, as they are received. It is therefore possible
  that some fragment contains only a part of the output line, while another spans multiple lines.
  It is the responsibility of the client to assemble these fragments according to its needs.

  `OsCmd` operates on the assumption that output is in utf8 encoding, so it may not work correctly
  for other encodings, such as plain binary.

  ## Command termination

  When command is being externally terminated (e.g. due to a timeout), a polite termination is
  first attempted, by sending a SIGTERM signal to the OS process (if the OS supports such signal),
  or alternatively invoking a custom terminated command provided via `:terminate_cmd`. If the OS
  process doesn't stop in 5 seconds (currently not configurable), a SIGKILL signal will be sent.

  ## Internals

  The owner process starts the command as the Erlang port. The command is not started directly.
  Instead a bridge program (implemented in go) is used to start and manage the OS process. Each
  command uses its own bridge process. This approach ensures proper cleanup guarantees even if the
  BEAM OS process is taken down.

  As a result, compared to `System.cmd/3`, `OsCmd` will consume more resources (2x more OS process
  instances) and require more hops to pass the output back to Elixir. Most often this won't matter,
  but be aware of these trade-offs if you're starting a large number of external processes.

  ## Mocking in tests

  Command execution can be mocked, which may be useful if you want to avoid starting long-running
  commands in tests.

  Mocking can done with `expect/1` and `stub/1`, and explicit allowances can be issued with
  `allow/1`.
  """
  @spec start_link(String.t()) :: GenServer.on_start()
  def start_link(command) when is_binary(command), do: start_link({command, []})

  @spec start_link({String.t(), [start_opt]}) :: GenServer.on_start()
  def start_link({command, opts}) do
    opts = normalize_opts(opts)
    GenServer.start_link(__MODULE__, {command, opts}, Keyword.take(opts, [:name]))
  end

  @doc "Stops the command and the owner process."
  @spec stop(GenServer.server(), :infinity | pos_integer()) :: :ok
  def stop(server, timeout \\ :infinity) do
    pid = whereis!(server)
    mref = Process.monitor(pid)
    GenServer.cast(pid, :stop)

    receive do
      {:DOWN, ^mref, :process, ^pid, _reason} -> :ok
    after
      timeout -> exit(:timeout)
    end
  end

  @doc """
  Returns a lazy stream of events.

  This function is internally used by `run/2` and `await/1`. If you want to use it yourself, you
  need to pass the handler `&send(some_pid, {self(), &1})` when starting the command. This
  function can only be invoked in the process which receives the event messages.
  """
  @spec events(GenServer.server()) :: Enumerable.t()
  def events(server) do
    pid = GenServer.whereis(server)

    Stream.resource(
      fn -> Process.monitor(pid) end,
      fn
        nil ->
          {:halt, nil}

        mref ->
          receive do
            {^pid, {:stopped, _} = stopped} ->
              Process.demonitor(mref, [:flush])
              {[stopped], nil}

            {^pid, message} ->
              {[message], mref}

            {:DOWN, ^mref, :process, ^pid, reason} ->
              {[{:stopped, reason}], nil}
          end
      end,
      fn
        nil -> :ok
        mref -> Process.demonitor(mref, [:flush])
      end
    )
  end

  @doc """
  Synchronously runs the command.

  This function will start the owner process, wait for it to finish, and return the result which
  will include the complete output of the command.

  If the command exits with a zero exit status, an `:ok` tuple is returned. Otherwise, the function
  returns an error tuple.

  See `start_link/1` for detailed explanation.
  """
  @spec run(String.t(), [start_opt]) :: {:ok, output} | {:error, Error.t()}
  def run(cmd, opts \\ []) do
    caller = self()
    start_arg = {cmd, [handler: &send(caller, {self(), &1})] ++ opts}

    start_fun =
      case Keyword.fetch(opts, :start) do
        :error -> fn -> start_link(start_arg) end
        {:ok, fun} -> fn -> fun.({__MODULE__, start_arg}) end
      end

    with {:ok, pid} <- start_fun.() do
      try do
        await(pid)
      after
        stop(pid)
      end
    end
  end

  @doc """
  Awaits for the started command to finish.

  This function is internally used by `run/2`. If you want to use it yourself, you need to pass the
  handler `&send(some_pid, {self(), &1})` when starting the command. This function can only be
  invoked in the process which receives the event messages.
  """
  @spec await(GenServer.server()) :: {:ok, output :: String.t()} | {:error, Error.t()}
  def await(server) do
    server
    |> whereis!()
    |> events()
    |> Enum.reduce(
      %{output: [], exit_status: nil},
      fn
        :starting, acc -> acc
        {:output, output}, acc -> update_in(acc.output, &[&1, output])
        {:stopped, exit_status}, acc -> %{acc | exit_status: exit_status}
      end
    )
    |> case do
      %{exit_status: 0} = result ->
        {:ok, to_string(result.output)}

      result ->
        {
          :error,
          %Error{
            message: "command failed",
            output: to_string(result.output),
            exit_status: result.exit_status
          }
        }
    end
  end

  @doc """
  Returns a specification for running the command as a `Job` action.

  The corresponding action will return `{:ok, output} | {:error, %OsCmdError{}}`
  See `Job.start_action/2` for details.
  """
  @spec action(String.t(), [start_opt | Job.action_opt()]) :: Job.action()
  def action(cmd, opts \\ []) do
    fn responder ->
      {action_opts, opts} = Keyword.split(opts, ~w/telemetry_id temeletry_meta/a)
      action_opts = Config.Reader.merge(action_opts, telemetry_meta: %{cmd: cmd})
      handler_state = %{responder: responder, cmd: cmd, opts: opts, output: []}
      {{__MODULE__, {cmd, [handler: {handler_state, &handle_event/2}] ++ opts}}, action_opts}
    end
  end

  @doc """
  Issues an explicit mock allowance to another process.

  Note that mocks are automatically inherited by descendants, so you only need to use this for non
  descendant processes. See `Mox.allow/3` for details.
  """
  @spec allow(GenServer.server()) :: :ok
  def allow(server), do: Faker.allow(whereis!(server))

  @doc """
  Sets up a mock expectation.

  The argument can be either a string (the exact command text), or a function. If the string is
  passed, the mocked command will succeed with the empty output. If the function is passed, it will
  be invoked when the command is started. The function can then return ok or error tuple.

  The expectation will be inherited by all descendant processes (unless overridden somewhere down
  the process tree).

  See `Mox.expect/4` for details on expectations.
  """
  @spec expect(mock) :: :ok
  def expect(fun), do: Faker.expect(fun)

  @doc """
  Sets up a mock stub.

  This function works similarly to `expect/1`, except it sets up a stub. See `Mox.stub/3` for
  details on stubs.
  """
  @spec stub(mock) :: :ok
  def stub(fun), do: Faker.stub(fun)

  @impl GenServer
  def init({cmd, opts}) do
    Process.flag(:trap_exit, true)

    state = %{
      port: nil,
      handlers: Keyword.fetch!(opts, :handlers),
      propagate_exit?: Keyword.get(opts, :propagate_exit?, false),
      buffer: "",
      exit_reason: nil
    }

    state = invoke_handler(state, :starting)

    with {:ok, timeout} <- Keyword.fetch(opts, :timeout),
         do: Process.send_after(self(), :timeout, timeout)

    starter =
      case Faker.fetch() do
        {:ok, pid} ->
          Mox.allow(Faker.Port, pid, self())
          Faker.Port

        :error ->
          OsCmd.Port
      end

    case starter.start(cmd, opts) do
      {:ok, port} -> {:ok, %{state | port: port}}
      {:error, reason} -> {:stop, reason}
    end
  end

  @impl GenServer
  def handle_info({port, {:exit_status, exit_status}}, %{port: port} = state),
    # Delegating to `handle_continue` because we must invoke a custom handler which can crash, so
    # we need to make sure that the correct state is committed.
    do: {:noreply, %{state | port: nil}, {:continue, {:stop, exit_status}}}

  def handle_info({port, {:data, message}}, %{port: port} = state) do
    state = invoke_handler(state, message)
    {:noreply, state}
  end

  def handle_info(:timeout, state) do
    send_stop_command(state)
    {:noreply, %{state | exit_reason: :timeout}}
  end

  @impl GenServer
  def handle_continue({:stop, exit_status}, state) do
    state = invoke_handler(state, {:stopped, exit_status})

    exit_reason =
      cond do
        not is_nil(state.exit_reason) -> state.exit_reason
        not state.propagate_exit? or exit_status == 0 -> :normal
        true -> {:failed, exit_status}
      end

    {:stop, exit_reason, %{state | port: nil}}
  end

  @impl GenServer
  def handle_cast(:stop, state) do
    send_stop_command(state)
    {:noreply, %{state | exit_reason: :normal}}
  end

  @impl GenServer
  def terminate(_reason, %{port: port} = state) do
    unless is_nil(port) do
      send_stop_command(state)

      # If we end up here, we still didn't receive the exit_status command, so we'll await for it
      # indefinitely. We assume that the go bridge works flawlessly and that it will stop the
      # program eventually, so there's no timeout clause. If there's a bug, this process will hang,
      # but at least we won't leak OS processes.
      receive do
        {^port, {:exit_status, _exit_status}} -> :ok
      end
    end
  end

  defp normalize_opts(opts) do
    {handlers, opts} = Keyword.pop_values(opts, :handler)

    env =
      opts
      |> Keyword.get(:env, [])
      |> Enum.map(fn
        {name, nil} -> {env_name_to_charlist(name), false}
        {name, value} -> {env_name_to_charlist(name), to_charlist(value)}
      end)

    Keyword.merge(opts, handlers: handlers, env: env)
  end

  defp env_name_to_charlist(atom) when is_atom(atom),
    do: atom |> to_string() |> String.upcase() |> to_charlist()

  defp env_name_to_charlist(name), do: to_charlist(name)

  defp invoke_handler(state, message) do
    message = with message when is_binary(message) <- message, do: :erlang.binary_to_term(message)
    {message, state} = normalize_message(message, state)

    handlers =
      Enum.map(
        state.handlers,
        fn
          {acc, fun} ->
            {fun.(message, acc), fun}

          fun ->
            fun.(message)
            fun
        end
      )

    %{state | handlers: handlers}
  end

  defp normalize_message({:output, output}, state) do
    {output, rest} = get_utf8_chars(state.buffer <> output)
    {{:output, to_string(output)}, %{state | buffer: rest}}
  end

  defp normalize_message(message, state), do: {message, state}

  defp get_utf8_chars(<<char::utf8, rest::binary>>) do
    {remaining_bytes, rest} = get_utf8_chars(rest)
    {[char | remaining_bytes], rest}
  end

  defp get_utf8_chars(other), do: {[], other}

  defp send_stop_command(state) do
    if not is_nil(state.port) do
      try do
        Port.command(state.port, "stop")
      catch
        _, _ -> :ok
      end
    end
  end

  defp handle_event({:output, output}, state),
    do: update_in(state.output, &[&1, output])

  defp handle_event({:stopped, exit_status}, state) do
    output = to_string(state.output)

    response =
      if exit_status == 0 do
        {:ok, output}
      else
        message = "#{state.cmd} exited with status #{exit_status}"
        {:error, %Error{exit_status: exit_status, message: message, output: output}}
      end

    state.responder.(response)
    nil
  end

  defp handle_event(_event, state), do: state

  defp whereis!(server) do
    case GenServer.whereis(server) do
      pid when is_pid(pid) -> pid
      nil -> raise "process #{inspect(server)} not found"
    end
  end

  defmodule Program do
    @moduledoc false
    @type id :: any

    @callback start(cmd :: String.t() | [String.t()], opts :: Keyword.t()) ::
                {:ok, id} | {:error, reason :: any}
  end
end