arduino_router_bridge_ex

# SPDX-License-Identifier: MIT
# Copyright (c) 2025 Scott Thompson

defmodule ArduinoRouter.Bridge do
  @moduledoc """
  Elixir interface to the [Arduino Router](https://github.com/arduino/arduino-router) Bridge.

  This module is modeled after the Sketch and Python Bridge modules
  from the [Arduino Uno Q documentation](https://docs.arduino.cc/tutorials/uno-q/user-manual/#bridge---remote-procedure-call-rpc-library).
  It provides a GenServer-based interface for communicating with other programs
  running on the device's processors through RPC calls.

  It is primarily intended for interaction between Elixir, running in the Linux
  environment of the MPU, and an Arduino Sketch running on the MCU. It can also
  be used for interprocesscommunication between programs on the MPU.

  In common usage, the module pairs with `ArduinoRouter.Socket`, taking
  advantage of the Unix socket interface provided by the arduino-router.
  However, the bridge uses an abstract behaviour, `ArduinoRouter.Transport`,
  so other transport mechanisms can be implemented.

  Start the bridge and register methods to handle incoming calls:

      {:ok, bridge} = ArduinoRouter.Bridge.start_link()
      ArduinoRouter.Bridge.provide("led_on", MyModule, :turn_on_led)

  Make calls through the bridge:

      {:ok, result} = ArduinoRouter.Bridge.call("get_sensor_reading", [])

  Send notifications (fire-and-forget):

      ArduinoRouter.Bridge.notify("log_message", ["Hello from Elixir"])
  """
  require Logger
  require :telemetry

  use GenServer

  alias ArduinoRouter.Socket
  alias ArduinoRouter.MessagePackRPC

  @type t :: %__MODULE__{
          transport_module: module(),
          transport_state: ArduinoRouter.Transport.state(),
          registered_methods: %{MessagePackRPC.method_name() => {module(), atom()}},
          calls_in_flight: %{MessagePackRPC.message_id() => pid()},
          next_id: MessagePackRPC.message_id()
        }

  defstruct transport_module: nil,
            transport_state: nil,
            registered_methods: %{},
            calls_in_flight: %{},
            next_id: 1

  @doc """
    Register the given module and function to handle an incoming method call or
    notification. The function will be invoked MFA style with the params
    extracted from the RPC message.

    The function registered is expected to return {:ok, result} or {:error, reason}.
    `result` and `reason` must be MessagePack serializable.
  """
  @spec provide(MessagePackRPC.method_name(), module(), atom()) :: {:ok, any()} | {:error, any()}
  def provide(method, module, function)
      when is_binary(method) and is_atom(module) and is_atom(function) do
    case call("$/register", [method]) do
      {:ok, _result} = result ->
        :telemetry.execute([:arduino_router, :bridge, :register_method], %{}, %{
          method: method,
          module: module,
          function: function
        })

        GenServer.call(__MODULE__, {:register, method, module, function})
        result

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

  @doc """
    Removes *all* provided methods from the arduino-router.
  """
  @spec reset() :: {:ok, MessagePackRPC.result()} | {:error, MessagePackRPC.error()}
  def reset() do
    # Reset the methods on the arduino-router side
    response = call("$/reset", [])

    # Reset the registered methods on this side too
    GenServer.call(__MODULE__, :reset_methods)

    :telemetry.execute([:arduino_router, :bridge, :reset_methods], %{}, %{})

    response
  end

  @doc """
    Asks the bridge for its version number
  """
  @spec version() :: {:ok, String.t()} | {:error, MessagePackRPC.error()}
  def version() do
    case call("$/version", []) do
      {:ok, version} = result ->
        :telemetry.execute([:arduino_router, :bridge, :version_request], %{}, %{version: version})
        result

      {:error, _reason} = error ->
        error
    end
  end

  @doc """
    Make a synchronous RPC call to the arduino-router with the given method and
    parameters. The parameters list should be MessagePack serializable.

    Following the convention of other implementations of the bridge,
    this function will block until a response is received.
  """
  @spec call(MessagePackRPC.method_name(), MessagePackRPC.params()) ::
          {:ok, MessagePackRPC.result()} | {:error, MessagePackRPC.error()}
  @spec call(MessagePackRPC.method_name(), MessagePackRPC.params(), timeout()) ::
          {:ok, MessagePackRPC.result()} | {:error, MessagePackRPC.error()}
  def call(method, params, timeout \\ 5000) when is_binary(method) and is_list(params) do
    start_time = System.monotonic_time()

    result =
      Task.async(fn ->
        GenServer.cast(__MODULE__, {:arduino_router_call, self(), method, params})

        receive do
          {:arduino_router_response, result} ->
            result
        end
      end)
      |> Task.await(timeout)

    duration = System.monotonic_time() - start_time

    :telemetry.execute(
      [:arduino_router, :bridge, :call],
      %{duration: duration},
      %{method: method, result: elem(result, 0)}
    )

    result
  end

  @doc """
    This mechanism is not really found in other implementations of the bridge,
    but makes sense in Elixir.

    Make an asynchronous RPC call to the arduino-router with the given method and
    parameters. `call_async` will return `:ok` immediately.

    When the arduino-router returns the RPC call's response, a message is
    sent to the process that invoked `call_async`, only, in the format
    `{:arduino_router_response, {:ok, result}}` or
    `{:arduino_router_response, {:error, reason}}`. Callers must
    explicitly `receive` the message to obtain the asynchronous result.
  """
  @spec call_async(MessagePackRPC.method_name(), MessagePackRPC.params()) :: :ok
  def call_async(method, params) when is_binary(method) and is_list(params) do
    GenServer.cast(__MODULE__, {:arduino_router_call, self(), method, params})

    :telemetry.execute(
      [:arduino_router, :bridge, :call_async],
      %{},
      %{method: method}
    )

    :ok
  end

  @doc """
    Send an asynchronous RPC notification to the arduino-router with the given
    method and parameters. This call will return immediately since no response
    is expected.
  """
  @spec notify(MessagePackRPC.method_name(), MessagePackRPC.params()) :: :ok
  def notify(method, params) when is_binary(method) and is_list(params) do
    GenServer.cast(__MODULE__, {:notify, method, params})
  end

  @doc """
    Retuns the Supervisor child_spec for the bridge.
  """
  @spec child_spec(keyword()) :: Supervisor.child_spec()
  def child_spec(opts) do
    %{
      id: __MODULE__,
      start: {__MODULE__, :start_link, [opts]},
      type: :worker,
      restart: :permanent,
      shutdown: 5000
    }
  end

  @doc """
    Start the bridge. It is registered under the module name.

    Supports the `:transport` option to specify a custom transport module. This
    is primarily intended for unit testing. The value of the `:transport` option
    should be a tuple of the form `{TransportModule, transport_opts}` where
    `TransportModule` implements the `ArduinoRouter.Transport` behaviour and
    `transport_opts` is a list of options to pass to the transport module's
    `start/2` function.
  """
  @spec start_link(keyword()) :: GenServer.on_start()
  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @impl GenServer
  @spec init(keyword()) :: {:ok, t()}
  def init(opts) do
    {transport_module, transport_opts} = Keyword.get(opts, :transport, {Socket, []})

    case transport_module.start(self(), transport_opts) do
      {:ok, transport_state} ->
        {:ok, %__MODULE__{transport_module: transport_module, transport_state: transport_state}}

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

  @impl GenServer
  def handle_call({:register, method, module, function}, _sender, state) do
    {:reply, :ok,
     put_in(
       state,
       [Access.key(:registered_methods, %{}), Access.key(method)],
       {module, function}
     )}
  end

  def handle_call(:reset_methods, _sender, state) do
    {:reply, :ok, %__MODULE__{state | registered_methods: %{}}}
  end

  @impl GenServer
  def handle_cast({:arduino_router_call, caller, method, params}, state) do
    # Generate a message_id
    {message_id, new_state} = next_message_id(state)

    # Register the caller under that message_id
    new_state = register_in_flight_call(new_state, message_id, caller)

    {:ok, new_transport_state} =
      state.transport_module.send_message(
        state.transport_state,
        MessagePackRPC.rpc_request(message_id, method, params)
      )

    # Send a request message through the socket
    {:noreply, %{new_state | transport_state: new_transport_state}}
  end

  def handle_cast({:notify, method, params}, state) do
    {:ok, new_transport_state} =
      state.transport_module.send_message(
        state.transport_state,
        MessagePackRPC.rpc_notification(method, params)
      )

    {:noreply, %__MODULE__{state | transport_state: new_transport_state}}
  end

  @impl GenServer
  def handle_info({:incoming_rpc, rpc_message}, state) do
    new_state = handle_incoming_rpc(rpc_message, state)
    {:noreply, new_state}
  end

  @impl GenServer
  def terminate(_reason, state) do
    state.transport_module.stop(state.transport_state)
  end

  # Handle the result of a successfully parsed incoming RPC message.
  @spec handle_incoming_rpc(MessagePackRPC.rpc_message(), t()) :: t()
  defp handle_incoming_rpc({:response, {:ok, message_id, result}}, state) do
    caller = registered_caller(state, message_id)

    if nil != caller do
      send(caller, {:arduino_router_response, {:ok, result}})
    else
      Logger.error("No registered caller found for message ID #{message_id}")
    end

    complete_call(state, message_id)
  end

  defp handle_incoming_rpc({:response, {:error, message_id, error}}, state) do
    caller = registered_caller(state, message_id)

    if nil != caller do
      send(caller, {:arduino_router_response, {:error, error}})
    else
      Logger.error("No registered caller found for message ID #{message_id}")
    end

    complete_call(state, message_id)
  end

  defp handle_incoming_rpc({:request, {message_id, method_name, params}}, state) do
    case registered_handler(state, method_name) do
      {module, function} ->
        call_registered_method(module, function, params)
        |> send_rpc_response(message_id, state)

      nil ->
        Logger.error("No handler registered for requested method #{method_name}")
        send_rpc_response({:error, "Method implementation not found"}, message_id, state)
    end
  end

  defp handle_incoming_rpc({:notify, {method_name, params}}, state) do
    case registered_handler(state, method_name) do
      {module, function} ->
        call_registered_method(module, function, params)

      # The function result is ignored as this is a notification.

      nil ->
        Logger.error("No handler registered for notify method #{method_name}")
    end

    state
  end

  @spec send_rpc_response(
          {:ok, MessagePackRPC.result()} | {:error, MessagePackRPC.error()},
          MessagePackRPC.message_id(),
          t()
        ) :: t()
  defp send_rpc_response({:ok, result}, message_id, state) do
    message = MessagePackRPC.successful_response_message(message_id, result)

    {:ok, new_transport_state} =
      state.transport_module.send_message(state.transport_state, message)

    %__MODULE__{state | transport_state: new_transport_state}
  end

  defp send_rpc_response({:error, error}, message_id, state) do
    message = MessagePackRPC.error_response_message(message_id, error)

    {:ok, new_transport_state} =
      state.transport_module.send_message(state.transport_state, message)

    %__MODULE__{state | transport_state: new_transport_state}
  end

  # Just a short utility method to call the registered module/function
  # and translate exceptions into errors.
  @spec call_registered_method(module(), atom(), MessagePackRPC.params()) ::
          {:ok, MessagePackRPC.result()} | {:error, MessagePackRPC.error()}
  defp call_registered_method(module, function, args) do
    try do
      apply(module, function, args)
    rescue
      exception -> {:error, Exception.format(:error, exception, __STACKTRACE__)}
    end
  end

  # Returns the handler ({module, function}) registered for the given method name.
  # or nil if none is registered
  @spec registered_handler(t(), MessagePackRPC.method_name()) ::
          {module(), atom()} | nil
  defp registered_handler(state, method) do
    get_in(state, [Access.key(:registered_methods, %{}), Access.key(method)])
  end

  # Registers an in-flight call so that when the response is returned by the
  # arduino-router we can send the result back to the caller.
  @spec register_in_flight_call(t(), MessagePackRPC.message_id(), pid()) :: t()
  defp register_in_flight_call(state, message_id, caller) do
    put_in(state, [Access.key(:calls_in_flight, %{}), Access.key(message_id)], caller)
  end

  # Retrieves the caller registered for the given message_id.
  @spec registered_caller(t(), MessagePackRPC.message_id()) :: pid() | nil
  defp registered_caller(state, message_id) do
    get_in(state, [Access.key(:calls_in_flight, %{}), Access.key(message_id)])
  end

  # Completes the call by removing it from the in-flight calls.
  @spec complete_call(t(), MessagePackRPC.message_id()) :: t()
  defp complete_call(state, message_id) do
    %__MODULE__{state | calls_in_flight: Map.delete(state.calls_in_flight, message_id)}
  end

  # Generates the next message ID.  Returns the new id and an updated state
  @spec next_message_id(t()) :: {MessagePackRPC.message_id(), t()}
  defp next_message_id(state) do
    {state.next_id, %__MODULE__{state | next_id: rem(state.next_id + 1, 0xFFFF)}}
  end
end