xmavlink

defmodule XMAVLink.Router do
  @moduledoc """
  Connect to serial, udp and tcp ports and listen for, validate and
  forward MAVLink messages towards their destinations on other connections
  and/or Elixir processes subscribing to messages.

  The rules for MAVLink packet forwarding are described here:

    https://mavlink.io/en/guide/routing.html

  and here:

    http://ardupilot.org/dev/docs/mavlink-routing-in-ardupilot.html
  """

  use GenServer
  require Logger

  alias XMAVLink.Types
  alias XMAVLink.ConnectionSpec
  alias XMAVLink.ConnectionSupervisor
  alias XMAVLink.ConnectionWorker
  alias XMAVLink.Frame
  alias XMAVLink.Message
  alias XMAVLink.Router
  alias XMAVLink.Router.Config
  alias XMAVLink.Router.Routing
  alias XMAVLink.Signing
  alias XMAVLink.LocalConnection
  alias XMAVLink.SerialConnection
  alias XMAVLink.TCPOutConnection
  alias XMAVLink.UDPInConnection
  alias XMAVLink.UDPOutConnection

  @setup_signing_message_id 256

  @typedoc """
  Represents the state of the XMAVLink.Router. Initial values should be set in config.exs.

  ## Fields
  - dialect:            The XMAVLink dialect module generated by mix mavlink from a MAVLink definition file
  - connection_strings: Configuration strings describing the network and serial connections MAVLink messages
                        are to be received and sent over. Allowed formats are:

                        ```udpin:<local ip>:<local port>
                        udpout:<remote ip>:<remote port>
                        tcpout:<remote ip>:<remote port>
                        serial:<device>:<baud rate>```

                        Note there is no tcpin connection - tcp is rarely used for MAVLink, the exception
                        being SITL testing which requires a tcpout connection.

  - connections:        A map containing internal connection state for configured
                        transports and the local process connection. This field is
                        owned by the router and is not a supported integration API.
  - routes:             A map from a {system id, component id} tuple to the connection key a message from that
                        system/component was last received on. Used to forward messages to that system/component.
  - system_time_boot_ms:
                        A map from a {system id, component id} tuple to the last SYSTEM_TIME.time_boot_ms seen
                        from that source. Used to clear stale learned routes when a remote system reboots.
  """
  defstruct [
    # Registered router name, if any
    name: nil,
    # Generated dialect module
    dialect: nil,
    # Connection descriptions from user
    connection_strings: [],
    # Supervised connection retry delay
    connection_retry_ms: 1_000,
    # Whether remote inbound frames may be forwarded to other remote links
    remote_forwarding: true,
    # Subscription cache name for restart restoration, if any
    subscription_cache: nil,
    # Per-connection MAVLink 2 signing policy seed
    signing: nil,
    # Per-router supervised connection workers
    connection_supervisor: nil,
    connection_workers: %{},
    connection_worker_monitors: %{},
    # %{socket|port|local: XMAVLink.*_Connection}
    connections: %{},
    # Connection key last observed for each MAVLink address
    routes: %{},
    # Last SYSTEM_TIME.time_boot_ms observed for each MAVLink address
    system_time_boot_ms: %{}
  ]

  # Can't used qualified type as map key
  @type mavlink_address :: Types.mavlink_address()
  @type mavlink_connection :: Types.connection()
  @type connection_key :: :local | binary | port | {port, Types.net_address(), Types.net_port()}
  @typedoc "Delivery metadata returned by `send_message/1..3`."
  @type delivery :: %{
          source_connection: connection_key(),
          recipients: [connection_key()],
          remote_recipients: [connection_key()],
          target: XMAVLink.Dialect.target() | nil,
          message_id: Types.message_id(),
          unreachable?: boolean
        }
  @type router_name :: atom | {:global, term} | {:via, module, term}
  @type router_ref :: GenServer.server()
  @type subscribe_query :: [
          {:message, Message.t() | :unknown}
          | {subscribe_query_id_key, 0..255}
          | {:as_frame, boolean}
        ]
  @type t :: %Router{
          name: router_name | nil,
          dialect: module | nil,
          connection_strings: [String.t()],
          connection_retry_ms: non_neg_integer,
          remote_forwarding: boolean,
          signing: Signing.t() | nil,
          subscription_cache: router_name | nil,
          connection_supervisor: pid | nil,
          connection_workers: %{connection_key() => pid},
          connection_worker_monitors: %{reference => pid},
          connections: %{connection_key() => mavlink_connection()},
          routes: %{mavlink_address() => connection_key()},
          system_time_boot_ms: %{mavlink_address() => non_neg_integer}
        }

  defguardp is_router_ref(router)
            when is_pid(router) or
                   (is_atom(router) and not is_nil(router)) or
                   (is_tuple(router) and tuple_size(router) == 2 and
                      elem(router, 0) == :global) or
                   (is_tuple(router) and tuple_size(router) == 3 and
                      elem(router, 0) == :via and is_atom(elem(router, 1))) or
                   (is_tuple(router) and tuple_size(router) == 2 and
                      is_atom(elem(router, 0)) and not is_nil(elem(router, 0)) and
                      is_atom(elem(router, 1)) and not is_nil(elem(router, 1)))

  ##############
  # Router API #
  ##############

  @doc """
  Start the MAVLink Router service. You should not call this directly, use the MAVLink application
  to start this under a supervision tree with the services it expects to use.

  ## Parameters
  - dialect:            Name of the module generated by mix mavlink task to represent the enumerations
                        and messages of a particular MAVLink dialect
  - system:             The System id of this system  1..255, typically low for vehicles and high for
                        ground stations
  - component:          The component id of this system 1..255, typically 1 for the autopilot
  - connection_strings: A list of strings in the following formats:

                        udpin:<local ip>:<local port>
                        udpout:<remote ip>:<remote port>
                        tcpout:<remote ip>:<remote port>
                        serial:<device>:<baud rate>
  - connection_retry_ms:
                        Retry delay for configured connection workers after
                        open failures or TCP/serial disconnects. Defaults to
                        1000 ms.
  - remote_forwarding:
                        Whether frames received from remote links are forwarded
                        to other remote links. Defaults to true. Set false for
                        endpoint/GCS use cases that should receive remote
                        traffic locally without bridging it between links.

  - opts:               Standard GenServer options. Pass `:name` to register
                        a non-default router, or `name: nil` in the args map to
                        start an unregistered router addressed by pid.
  """
  @spec start_link(
          %{
            required(:system) => 1..255,
            required(:component) => 1..255,
            required(:dialect) => module,
            optional(:name) => router_name | nil,
            optional(:connection_strings) => [String.t()],
            optional(:connections) => [String.t()],
            optional(:connection_retry_ms) => non_neg_integer,
            optional(:remote_forwarding) => boolean,
            optional(:signing) => keyword() | nil
          },
          [{atom, any}]
        ) :: {:ok, pid}
  def start_link(args, opts \\ []) do
    args = Config.normalize!(args)
    name = Keyword.get(opts, :name, Map.get(args, :name, __MODULE__))

    GenServer.start_link(__MODULE__, Map.put(args, :name, name), start_options(opts, name))
  end

  @doc false
  def child_spec(args) do
    args = Config.normalize!(args)
    name = Map.get(args, :name, __MODULE__)

    %{
      id: child_id(args, name),
      start: {__MODULE__, :start_link, [Map.delete(args, :id)]}
    }
  end

  @doc """
  Subscribes the calling process to MAVLink messages from the installed dialect matching the query.

  ## Parameters

  - router: Optional router name or pid. Defaults to `XMAVLink.Router`.
  - query:  Keyword list of zero or more of the following query keywords:

    message:          message_module | :unknown (use latter with as_frame)
    source_system:    integer 0..255
    source_component: integer 0..255
    target_system:    integer 0..255
    target_component: integer 0..255
    as_frame:         true|false (default false, shows entire message frame with sender/target details)

  ## Example

  ```
    XMAVLink.Router.subscribe message: XMAVLink.Message.Heartbeat, source_system: 1
  ```
  """
  @type subscribe_query_id_key ::
          :source_system | :source_component | :target_system | :target_component
  @spec subscribe() :: :ok | {:error, :invalid_message}
  def subscribe(), do: subscribe(__MODULE__, [])

  @spec subscribe(router_ref | subscribe_query) :: :ok | {:error, :invalid_message}
  def subscribe(router) when is_router_ref(router), do: subscribe(router, [])
  def subscribe(nil), do: invalid_router_ref!(nil)

  def subscribe(invalid_router) when is_tuple(invalid_router),
    do: invalid_router_ref!(invalid_router)

  def subscribe(query), do: subscribe(__MODULE__, query)

  @spec subscribe(router_ref, subscribe_query) :: :ok | {:error, :invalid_message}
  def subscribe(router, query) when is_router_ref(router) do
    query = normalize_subscribe_query(query)

    with message <- Keyword.get(query, :message),
         true <-
           message == nil or message == XMAVLink.UnknownMessage or Code.ensure_loaded?(message) do
      # Synchronous: returns only after the subscription is committed to the
      # Router state. This prevents a race where the caller sends an outbound
      # message that elicits a reply before the inbound subscription is in
      # place, causing the reply to be dropped.
      GenServer.call(
        router,
        {
          :subscribe,
          [
            message: nil,
            source_system: 0,
            source_component: 0,
            target_system: 0,
            target_component: 0,
            as_frame: false
          ]
          |> Keyword.merge(query)
          |> Enum.into(%{}),
          self()
        }
      )
    else
      false ->
        {:error, :invalid_message}
    end
  end

  def subscribe(invalid_router, _query), do: invalid_router_ref!(invalid_router)

  @doc """
  Un-subscribes calling process from all existing subscriptions

  ## Example

  ```
    XMAVLink.Router.unsubscribe
  ```
  """
  @spec unsubscribe() :: :ok
  def unsubscribe(), do: unsubscribe(__MODULE__)

  @spec unsubscribe(router_ref) :: :ok
  def unsubscribe(router) when is_router_ref(router),
    do: GenServer.cast(router, {:unsubscribe, self()})

  def unsubscribe(invalid_router), do: invalid_router_ref!(invalid_router)

  @doc """
  Send a MAVLink message to one or more recipients using available
  connections. For now if destination is unreachable it will log
  a warning

  ## Parameters

  - router: Optional router name or pid. Defaults to `XMAVLink.Router`.
  - message: A MAVLink message structure from the installed dialect
  - version: Force sending using a specific MAVLink protocol (default 2)
  - opts: Optional send settings:
    - `:source_system` and `:source_component` override the router's
      configured local identity for this message. Provide both or neither.

  ## Example

  ```
    XMAVLink.Router.pack_and_send(
      %Common.RcChannelsOverride{
        target_system: 1,
        target_component: 1,
        chan1_raw: 1500,
        chan2_raw: 1500,
        chan3_raw: 1500,
        chan4_raw: 1500,
        chan5_raw: 1500,
        chan6_raw: 1500,
        chan7_raw: 1500,
        chan8_raw: 1500,
        chan9_raw: 0,
        chan10_raw: 0,
        chan11_raw: 0,
        chan12_raw: 0,
        chan13_raw: 0,
        chan14_raw: 0,
        chan15_raw: 0,
        chan16_raw: 0,
        chan17_raw: 0,
        chan18_raw: 0
      }
    )
  ```
  """
  @spec pack_and_send(Message.t()) :: :ok | {:error, :protocol_undefined}
  def pack_and_send(message), do: pack_and_send(__MODULE__, message, 2, [])

  @spec pack_and_send(router_ref, Message.t()) :: :ok | {:error, :protocol_undefined}
  def pack_and_send(router, message) when is_router_ref(router) and is_map(message),
    do: pack_and_send(router, message, 2, [])

  def pack_and_send(invalid_router, message) when is_map(message),
    do: invalid_router_ref!(invalid_router)

  @spec pack_and_send(Message.t(), Types.version() | keyword) ::
          :ok | {:error, :protocol_undefined}
  def pack_and_send(message, opts) when is_list(opts),
    do: pack_and_send(__MODULE__, message, 2, opts)

  def pack_and_send(message, version), do: pack_and_send(__MODULE__, message, version, [])

  @spec pack_and_send(router_ref, Message.t(), Types.version() | keyword) ::
          :ok | {:error, :protocol_undefined}
  def pack_and_send(router, message, opts) when is_router_ref(router) and is_list(opts),
    do: pack_and_send(router, message, 2, opts)

  def pack_and_send(router, message, version) when is_router_ref(router),
    do: pack_and_send(router, message, version, [])

  def pack_and_send(invalid_router, message, _version) when is_map(message),
    do: invalid_router_ref!(invalid_router)

  @spec pack_and_send(Message.t(), Types.version(), keyword) ::
          :ok | {:error, :protocol_undefined}
  def pack_and_send(message, version, opts), do: pack_and_send(__MODULE__, message, version, opts)

  @spec pack_and_send(router_ref, Message.t(), Types.version(), keyword) ::
          :ok | {:error, :protocol_undefined}
  def pack_and_send(router, message, version, opts) when is_router_ref(router) do
    case build_frame(message, version, opts) do
      {:ok, frame} ->
        # Resolve all router target shapes before dispatch so missing global/via
        # names fail fast instead of being silently dropped by GenServer.cast/2.
        dispatch_local(router, frame)
        :ok

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

  def pack_and_send(invalid_router, _message, _version, _opts),
    do: invalid_router_ref!(invalid_router)

  @doc """
  Synchronously pack, route, and send a MAVLink message.

  This is the structured alternative to `pack_and_send/2..4`. Pass `:version`
  in `opts` to select MAVLink 1 or 2, and pass `:source_system` plus
  `:source_component` to override the router's local identity for this message.

  The success reply includes the selected recipient connection keys. Connection
  keys are internal runtime identifiers, but they are useful for observability
  and tests.
  """
  @spec send_message(Message.t()) :: {:ok, delivery()} | {:error, term}
  def send_message(message), do: send_message(__MODULE__, message, [])

  @spec send_message(Message.t(), keyword) :: {:ok, delivery()} | {:error, term}
  def send_message(message, opts) when is_map(message) and is_list(opts),
    do: send_message(__MODULE__, message, opts)

  @spec send_message(router_ref, Message.t()) :: {:ok, delivery()} | {:error, term}
  def send_message(router, message) when is_router_ref(router) and is_map(message),
    do: send_message(router, message, [])

  @spec send_message(router_ref, Message.t(), keyword) :: {:ok, delivery()} | {:error, term}
  def send_message(router, message, opts) when is_router_ref(router) and is_map(message) do
    version = Keyword.get(opts, :version, 2)

    with {:ok, frame} <- build_frame(message, version, opts) do
      call_local(router, frame)
    end
  end

  def send_message(invalid_router, _message, _opts), do: invalid_router_ref!(invalid_router)

  defp build_frame(message, version, opts) do
    source_identity = source_identity!(opts)

    try do
      case Message.pack(message, version) do
        {:ok, message_id, {:ok, crc_extra, _, target}, payload} ->
          {target_system, target_component} =
            if target != :broadcast do
              {message.target_system, Map.get(message, :target_component, 0)}
            else
              {0, 0}
            end

          {:ok,
           struct(Frame,
             version: version,
             message_id: message_id,
             source_system: source_identity[:source_system],
             source_component: source_identity[:source_component],
             target_system: target_system,
             target_component: target_component,
             target: target,
             message: message,
             payload: payload,
             crc_extra: crc_extra
           )}

        {:error, reason} ->
          {:error, reason}
      end
    rescue
      # Need to catch Protocol.UndefinedError - happens with SimState (Common) and Simstate (APM)
      # messages because non-case-sensitive filesystems (including OSX thanks @noobz) can't tell
      # the difference between generated module beam files. Work around is comment out one of the
      # message definitions and regenerate the dialect module.
      Protocol.UndefinedError ->
        {:error, :protocol_undefined}
    end
  end

  defp dispatch_local(router, frame) do
    case GenServer.whereis(router) do
      nil -> router_not_running!(router)
      pid -> send(pid, {:local, frame})
    end
  end

  defp call_local(router, frame) do
    case GenServer.whereis(router) do
      nil -> router_not_running!(router)
      pid -> GenServer.call(pid, {:local, frame})
    end
  end

  defp child_id(%{id: id}, _name) when not is_nil(id), do: id

  defp child_id(_args, nil),
    do: raise(ArgumentError, "router child_spec requires :id when :name is nil")

  defp child_id(_args, name), do: name

  defp start_options(opts, nil), do: Keyword.delete(opts, :name)

  defp start_options(opts, name) do
    opts
    |> Keyword.delete(:name)
    |> Keyword.put(:name, name)
  end

  defp subscription_cache_name(nil), do: nil
  defp subscription_cache_name(__MODULE__), do: XMAVLink.SubscriptionCache

  defp subscription_cache_name(name), do: {:global, {__MODULE__, :subscription_cache, name}}

  defp normalize_subscribe_query(query) do
    Keyword.update(query, :message, nil, fn
      :unknown -> XMAVLink.UnknownMessage
      message -> message
    end)
  end

  defp invalid_router_ref!(router) do
    raise ArgumentError, "invalid router target #{inspect(router)}"
  end

  defp router_not_running!(router) do
    raise ArgumentError, "router target #{inspect(router)} is not running"
  end

  defp route_local(frame, state) do
    {state, _delivery} = route_local_with_delivery(frame, state)
    state
  end

  defp route_local_with_delivery(frame, state) do
    LocalConnection.handle_info({:local, frame}, state.connections.local, state.dialect)
    |> update_route_info(state)
    |> route_with_delivery()
  end

  defp source_identity!(opts) do
    source_system = Keyword.get(opts, :source_system)
    source_component = Keyword.get(opts, :source_component)

    case {source_system, source_component} do
      {nil, nil} ->
        []

      {system, component} when system in 1..255 and component in 1..255 ->
        [source_system: system, source_component: component]

      {nil, _component} ->
        raise ArgumentError, "source_system is required when source_component is set"

      {_system, nil} ->
        raise ArgumentError, "source_component is required when source_system is set"

      _ ->
        raise ArgumentError,
              "source_system and source_component must be integers in the MAVLink range 1..255"
    end
  end

  #######################
  # GenServer Callbacks #
  #######################

  @impl true
  # No dialect, no play.
  def init(%{dialect: nil}) do
    {:error, :no_mavlink_dialect_set}
  end

  # Start connections (they will send :add_connection messages
  # to us if successful) and initialise Router state
  def init(args) do
    subscription_cache = subscription_cache_name(args.name)
    {:ok, connection_supervisor} = ConnectionSupervisor.start_link()

    local_connection = LocalConnection.new(args.system, args.component, subscription_cache)
    connection_specs = Enum.map(args.connection_strings, &ConnectionSpec.parse/1)

    for connection_spec <- connection_specs do
      start_connection_worker(connection_supervisor, connection_spec, args.connection_retry_ms)
    end

    {:ok,
     %Router{
       name: args.name,
       dialect: args.dialect,
       connection_strings: args.connection_strings,
       connection_retry_ms: args.connection_retry_ms,
       remote_forwarding: args.remote_forwarding,
       connection_supervisor: connection_supervisor,
       subscription_cache: subscription_cache,
       signing: args.signing,
       connections: %{local: local_connection}
     }}
  end

  @impl true
  def handle_call({:local, frame}, _from, state) do
    {state, delivery} = route_local_with_delivery(frame, state)
    {:reply, {:ok, delivery}, state}
  end

  @impl true
  # Call to subscribe() API
  def handle_call({:subscribe, query, pid}, _from, state) do
    {:reply, :ok,
     update_in(
       state,
       [Access.key!(:connections), :local],
       &LocalConnection.subscribe(query, pid, &1)
     )}
  end

  @impl true
  # Call to unsubscribe() API
  def handle_cast({:unsubscribe, pid}, state) do
    {:noreply,
     update_in(state, [Access.key!(:connections), :local], &LocalConnection.unsubscribe(pid, &1))}
  end

  # A call to pack_and_send() targeting a non-local registered name.
  def handle_cast({:local, frame}, state) do
    {:noreply, route_local(frame, state)}
  end

  @impl true
  # Receive data from a supervised connection worker
  def handle_info({:connection_message, worker, message = {:udp, _, _, _, _}}, state) do
    handle_udp_message(message, worker, state)
  end

  # Receive data on UDP connection
  def handle_info(
        message = {:udp, _socket, _address, _port, _},
        state
      ) do
    handle_udp_message(message, nil, state)
  end

  def handle_info({:connection_message, _worker, message = {:tcp, _, _}}, state) do
    handle_tcp_message(message, state)
  end

  # Receive data on TCP connection
  def handle_info(message = {:tcp, _socket, _}, state) do
    handle_tcp_message(message, state)
  end

  def handle_info({:connection_closed, worker, connection_key}, state) do
    {:noreply, remove_worker_connection(connection_key, worker, state)}
  end

  # Unlike UDP, TCP connections can close
  def handle_info({:tcp_closed, socket}, state) do
    reconnect_worker(state.connection_workers[socket])
    {:noreply, remove_connection(socket, state)}
  end

  def handle_info({:connection_message, _worker, message = {:circuits_uart, _, raw}}, state)
      when is_binary(raw) do
    handle_serial_message(message, state)
  end

  # Received data on serial connection
  def handle_info(message = {:circuits_uart, _port, raw}, state) when is_binary(raw) do
    handle_serial_message(message, state)
  end

  # Received error on serial connection
  def handle_info({:circuits_uart, port, {:error, _reason}}, state) do
    reconnect_worker(state.connection_workers[port])
    {:noreply, remove_connection(port, state)}
  end

  # A local subscribing Elixir process or a supervised connection worker has crashed.
  def handle_info({:DOWN, ref, :process, pid, _}, state) do
    case Map.pop(state.connection_worker_monitors, ref) do
      {nil, _worker_monitors} ->
        {:noreply,
         update_in(
           state,
           [Access.key!(:connections), :local],
           &LocalConnection.subscriber_down(pid, &1)
         )}

      {worker, worker_monitors} ->
        {:noreply,
         state
         |> struct(connection_worker_monitors: worker_monitors)
         |> remove_connections_for_worker(worker)}
    end
  end

  # A call to pack_and_send() from a local Elixir process, sent as a vanilla message for symmetry with other connection types
  def handle_info({:local, frame}, state) do
    {:noreply, route_local(frame, state)}
  end

  # A supervised connection worker has successfully opened a connection and
  # wants to add it to our connection list.
  def handle_info(
        {:add_connection, connection_key, connection},
        state = %Router{connections: connections}
      ) do
    connection = connection_with_signing(connection, state.signing)

    {
      :noreply,
      struct(
        state,
        connections: Map.put(connections, connection_key, connection)
      )
    }
  end

  def handle_info(
        {:add_connection, connection_key, connection, worker},
        state = %Router{connections: connections}
      ) do
    connection = connection_with_signing(connection, state.signing)

    {
      :noreply,
      state
      |> monitor_connection_worker(worker)
      |> struct(
        connections: Map.put(connections, connection_key, connection),
        connection_workers: Map.put(state.connection_workers, connection_key, worker)
      )
    }
  end

  # Ignore weird messages
  def handle_info(_msg, state) do
    {:noreply, state}
  end

  defp handle_udp_message(
         message = {:udp, socket, address, port, _},
         worker,
         state = %Router{connections: connections, dialect: dialect, signing: signing}
       ) do
    {
      :noreply,
      case connections[socket] do
        # A `udpout:` connection is registered under the bare `socket` key
        # (see UDPOutConnection.open/2). Any UDP packet arriving on that
        # socket is a reply to our outbound traffic - attribute it to the
        # UDPOut regardless of source IP. This is required behind NAT /
        # masquerade / kube-proxy DNAT, where the reply's source IP is the
        # NAT gateway rather than the address we sent to. Without this,
        # we'd file the reply as a brand-new UDPInConnection under
        # `{socket, reply_ip, reply_port}` and the broadcast `route/1`
        # clause would forward subsequent inbound frames back out the
        # original UDPOut - i.e. echo the host's traffic back to itself.
        udpout = %UDPOutConnection{} ->
          UDPOutConnection.handle_info(message, udpout, dialect)

        _ ->
          case connections[{socket, address, port}] do
            connection = %UDPInConnection{} ->
              UDPInConnection.handle_info(message, connection, dialect)

            connection = %UDPOutConnection{} ->
              UDPOutConnection.handle_info(message, connection, dialect)

            nil ->
              # New previously unseen UDPIn client
              UDPInConnection.handle_info(message, nil, dialect, worker, signing)
          end
      end
      |> update_route_info(state)
      |> route
    }
  end

  defp handle_tcp_message(message = {:tcp, socket, _}, state) do
    case state.connections[socket] do
      nil ->
        {:noreply, state}

      connection ->
        {
          :noreply,
          TCPOutConnection.handle_info(message, connection, state.dialect)
          |> update_route_info(state)
          |> route
        }
    end
  end

  defp handle_serial_message(message = {:circuits_uart, port, _raw}, state) do
    case state.connections[port] do
      nil ->
        {:noreply, state}

      connection ->
        {
          :noreply,
          SerialConnection.handle_info(message, connection, state.dialect)
          |> update_route_info(state)
          |> route
        }
    end
  end

  ####################
  # Helper Functions #
  ####################

  defp connection_with_signing(connection, signing) do
    if Map.has_key?(connection, :signing) do
      struct(connection, signing: signing)
    else
      connection
    end
  end

  # Handle user configured connections with supervised workers. If
  # successful they send us an :add_connection message with the details. The
  # local connection gets added automatically.
  defp start_connection_worker(connection_supervisor, connection_spec, retry_ms) do
    DynamicSupervisor.start_child(
      connection_supervisor,
      {ConnectionWorker,
       Map.merge(connection_spec, %{
         router: self(),
         retry_ms: retry_ms
       })}
    )
  end

  defp reconnect_worker(nil), do: :ok
  defp reconnect_worker(worker), do: ConnectionWorker.reconnect(worker)

  defp monitor_connection_worker(state, worker) do
    if worker in Map.values(state.connection_worker_monitors) do
      state
    else
      ref = Process.monitor(worker)
      put_in(state.connection_worker_monitors[ref], worker)
    end
  end

  defp track_connection_worker(state, connection_key, %{worker: worker}) when is_pid(worker) do
    state
    |> monitor_connection_worker(worker)
    |> struct(connection_workers: Map.put(state.connection_workers, connection_key, worker))
  end

  defp track_connection_worker(state, _connection_key, _connection), do: state

  defp remove_worker_connection(connection_key, worker, state) do
    if state.connection_workers[connection_key] == worker do
      remove_connection(connection_key, state)
    else
      state
    end
  end

  defp remove_connections_for_worker(state, worker) do
    state.connection_workers
    |> Enum.filter(fn {_connection_key, connection_worker} -> connection_worker == worker end)
    |> Enum.reduce(state, fn {connection_key, _worker}, updated_state ->
      remove_connection(connection_key, updated_state)
    end)
  end

  defp remove_connection(connection_key, state = %Router{}) do
    Routing.remove_connection(connection_key, state)
  end

  defp update_route_info(result, state) do
    case Routing.update_route_info(result, state) do
      {:ok, source_connection_key, frame, state} ->
        {:ok, source_connection_key, frame,
         track_connection_worker(
           state,
           source_connection_key,
           state.connections[source_connection_key]
         )}

      {:error, reason, state} ->
        {:error, reason, track_workers_for_connections(state)}
    end
  end

  defp track_workers_for_connections(state) do
    Enum.reduce(state.connections, state, fn {connection_key, connection}, updated_state ->
      track_connection_worker(updated_state, connection_key, connection)
    end)
  end

  defp route(result) do
    {state, _delivery} = route_with_delivery(result)
    state
  end

  # SETUP_SIGNING carries key material. Inbound provisioning messages are
  # delivered locally for application handling, but never bridged to another
  # MAVLink connection by generic routing.
  defp route_with_delivery(
         {:ok, source_connection_key, frame = %Frame{message_id: @setup_signing_message_id},
          state = %Router{connections: connections}}
       )
       when source_connection_key != :local do
    recipients = [:local]

    {
      forward_and_update(:local, connections.local, frame, state),
      Routing.delivery(source_connection_key, recipients, frame, state)
    }
  end

  defp route_with_delivery(
         {:ok, source_connection_key, frame = %Frame{target: :broadcast}, state = %Router{}}
       ) do
    recipients = Routing.broadcast_recipients(source_connection_key, state)

    {
      forward_to_recipients(recipients, frame, state),
      Routing.delivery(source_connection_key, recipients, frame, state)
    }
  end

  defp route_with_delivery(
         {:ok, source_connection_key,
          frame = %Frame{
            target_system: target_system,
            target_component: target_component,
            message: %{__struct__: message_type}
          }, state = %Router{}}
       ) do
    recipients =
      Routing.targeted_recipients(target_system, target_component, source_connection_key, state)

    delivery = Routing.delivery(source_connection_key, recipients, frame, state)

    if delivery.unreachable? do
      Logger.debug(
        "Could not send message #{Atom.to_string(message_type)} to #{target_system}/#{target_component}: destination unreachable"
      )
    end

    {forward_to_recipients(recipients, frame, state), delivery}
  end

  defp route_with_delivery({:error, reason, state = %Router{}}) do
    {state, {:error, reason}}
  end

  defp forward_to_recipients(recipients, frame, state) do
    Enum.reduce(recipients, state, fn connection_key, routed_state ->
      forward_and_update(
        connection_key,
        routed_state.connections[connection_key],
        frame,
        routed_state
      )
    end)
  end

  defp forward_and_update(_connection_key, nil, _frame, state), do: state

  defp forward_and_update(connection_key, connection, frame, state) do
    updated_connection = forward(connection, frame)
    put_in(state.connections[connection_key], updated_connection)
  end

  defp forward(connection, frame) do
    case outbound_frame(connection, frame) do
      {:ok, updated_connection, outbound_frame} ->
        _ = forward_raw(updated_connection, outbound_frame)
        updated_connection

      {:error, reason, updated_connection} ->
        Logger.debug("Could not sign outbound MAVLink frame: #{inspect(reason)}")
        updated_connection
    end
  end

  defp outbound_frame(connection, frame) do
    if Map.has_key?(connection, :signing) do
      case Signing.sign_outbound(frame, connection.signing) do
        {:ok, outbound_frame, signing} ->
          {:ok, struct(connection, signing: signing), outbound_frame}

        {:error, reason, signing} ->
          {:error, reason, struct(connection, signing: signing)}
      end
    else
      {:ok, connection, frame}
    end
  end

  # Delegate sending a message to connection-type specific code
  defp forward_raw(connection = %UDPInConnection{}, frame),
    do: UDPInConnection.forward(connection, frame)

  defp forward_raw(connection = %UDPOutConnection{}, frame),
    do: UDPOutConnection.forward(connection, frame)

  defp forward_raw(connection = %TCPOutConnection{}, frame),
    do: TCPOutConnection.forward(connection, frame)

  defp forward_raw(connection = %SerialConnection{}, frame),
    do: SerialConnection.forward(connection, frame)

  defp forward_raw(connection = %LocalConnection{}, frame),
    do: LocalConnection.forward(connection, frame)
end