Letting Agents Debug Production

defmodule Orders.MixProject do
  use Mix.Project

  def project do
    [app: :orders, version: "0.1.0", elixir: "~> 1.20", deps: []]
  end

  def application do
    # :ssh ships with Erlang/OTP; the release must include it.
    [mod: {Orders.Application, []}, extra_applications: [:logger, :ssh]]
  end
end

defmodule Orders.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [Orders.Worker, Orders.AgentAccess]
    Supervisor.start_link(children, strategy: :one_for_one, name: Orders.Supervisor)
  end
end

defmodule Orders.Worker do
  @moduledoc """
  Charges orders one at a time. Today the payment provider hangs, so the first
  charge never returns and every later order waits in the mailbox.
  """
  use GenServer

  def start_link(_opts), do: GenServer.start_link(__MODULE__, nil, name: __MODULE__)

  def charge(order_id), do: GenServer.cast(__MODULE__, {:charge, order_id})

  @impl true
  def init(nil), do: {:ok, nil}

  @impl true
  def handle_cast({:charge, order_id}, state) do
    :ok = call_payment_provider(order_id)
    {:noreply, state}
  end

  # Stands in for an HTTP call without a timeout to a provider that never answers.
  defp call_payment_provider(_order_id) do
    receive do
      :never_sent -> :ok
    end
  end
end

defmodule Orders.AgentAccess do
  @moduledoc """
  A narrow, read-only window into the running node for coding agents.

  An SSH daemon (OTP's own `:ssh`) accepts only the keys listed in
  `authorized_keys`, offers no shell, and runs only the commands below. Every
  request, allowed or not, is written to an audit log first.
  """

  @commands %{
    "queues" => "processes with messages waiting, longest queue first (top 3)",
    "where NAME" => "what a registered process is doing (its stack)",
    "help" => "this list"
  }

  def child_spec(_opts), do: %{id: __MODULE__, start: {__MODULE__, :start_link, []}}

  def start_link do
    config = Application.fetch_env!(:orders, :agent_access)

    :ssh.daemon(config[:port],
      system_dir: String.to_charlist(config[:host_keys]),
      user_dir: String.to_charlist(config[:authorized_keys]),
      auth_methods: ~c"publickey",
      shell: :disabled,
      exec: {:direct, fn command, user -> run(to_string(command), to_string(user)) end}
    )
  end

  defp run(command, user) do
    File.write!(audit_log(), "#{user}: #{command}\n", [:append])

    result =
      case String.split(command) do
        ["queues"] -> {:ok, queues()}
        ["where", name] -> where(name)
        ["help"] -> {:ok, Enum.map_join(@commands, "\n", fn {cmd, what} -> "#{cmd}: #{what}" end)}
        _ -> {:error, "not allowed: #{command} (try: help)"}
      end

    # The SSH client prints the text as is; end it with a newline.
    case result do
      {:ok, text} -> {:ok, text <> "\n"}
      {:error, text} -> {:error, text <> "\n"}
    end
  end

  defp queues do
    Process.list()
    |> Enum.map(&{&1, Process.info(&1, [:registered_name, :message_queue_len])})
    |> Enum.filter(fn {_pid, info} -> info != nil and info[:message_queue_len] > 0 end)
    |> Enum.sort_by(fn {_pid, info} -> -info[:message_queue_len] end)
    |> Enum.take(3)
    |> Enum.map_join("\n", fn {pid, info} -> "#{label(pid, info)}: #{info[:message_queue_len]}" end)
  end

  # Only names that already exist are accepted: String.to_existing_atom/1
  # never creates atoms from untrusted input.
  defp where(name) do
    with {:ok, atom} <- existing_atom(name),
         pid when is_pid(pid) <- Process.whereis(atom),
         {:current_stacktrace, stack} <- Process.info(pid, :current_stacktrace) do
      {:ok, stack |> Enum.take(3) |> Enum.map_join("\n", &Exception.format_stacktrace_entry/1)}
    else
      _ -> {:error, "no process named #{name}"}
    end
  end

  defp existing_atom(name) do
    {:ok, String.to_existing_atom("Elixir." <> name)}
  rescue
    ArgumentError -> :error
  end

  defp label(pid, info) do
    case info[:registered_name] do
      [] -> inspect(pid)
      name -> inspect(name)
    end
  end

  defp audit_log, do: Application.fetch_env!(:orders, :agent_access)[:audit_log]
end

import Config

# Where the agent-access SSH daemon listens and keeps its keys. Bind it to an
# internal address only; never expose it to the internet.
config :orders, :agent_access,
  port: String.to_integer(System.get_env("AGENT_SSH_PORT", "4022")),
  host_keys: System.fetch_env!("AGENT_SSH_HOST_KEYS"),
  authorized_keys: System.fetch_env!("AGENT_SSH_AUTHORIZED_KEYS"),
  audit_log: System.fetch_env!("AGENT_AUDIT_LOG")

# Simulates the incident: 100 orders arrive while the payment provider hangs.
for id <- 1..100, do: Orders.Worker.charge(id)

# Wait until the first order is stuck and the other 99 are queued behind it.
worker = Process.whereis(Orders.Worker)

wait = fn again ->
  if Process.info(worker, :message_queue_len) == {:message_queue_len, 99},
    do: :ok,
    else: Process.sleep(10) && again.(again)
end

:ok = wait.(wait)