ovsdb_ex

defmodule OVSDB.ServerSession do
  @moduledoc """
  Per-connection server-side OVSDB session. Takes ownership of an
  accepted Transport, receives client requests, dispatches to a
  `Handler` behaviour module, and sends responses.

  ## Role asymmetry with ClientSession

  `ClientSession` originates requests and correlates responses.
  `ServerSession` does the inverse: receives requests, produces
  responses. It can also *originate* notifications (like `update`
  messages for monitors) via `notify/3`.

  ## Handler behaviour

  Applications implement `OVSDB.ServerSession.Handler` to respond to
  RFC 7047 methods. One handler module serves one server config;
  handler state is per-connection.

      defmodule MySimHandler do
        @behaviour OVSDB.ServerSession.Handler

        @impl true
        def init(opts), do: {:ok, %{sim_pid: Keyword.fetch!(opts, :sim_pid)}}

        @impl true
        def handle_list_dbs(state), do: {:ok, ["Open_vSwitch"], state}

        @impl true
        def handle_get_schema("Open_vSwitch", state) do
          {:ok, state.sim_pid |> Sim.schema(), state}
        end

        @impl true
        def handle_transact("Open_vSwitch", ops, state) do
          # Apply ops, return per-op results
        end

        # ...
      end

  ## Unimplemented methods

  Handler callbacks are all optional. A method with no matching
  callback returns `{"error": "not supported"}` per RFC 7047 §4 —
  the default behavior of `Handler`'s default implementations.

  ## Notifications

  `notify/3` pushes a notification to the connected client:

      ServerSession.notify(session, "update", ["monitor-id", updates])
  """

  use GenServer

  require Logger

  alias OVSDB.{Protocol, Transport}

  defmodule Handler do
    @moduledoc """
    Behaviour for applications serving OVSDB requests.

    All callbacks are optional. Methods not implemented return a
    `"not supported"` error to the client.
    """

    @type state :: term()
    @type error_string :: String.t()
    @type result :: {:ok, term(), state()} | {:error, error_string(), state()}

    @callback init(opts :: keyword()) :: {:ok, state()} | {:error, term()}
    @callback terminate(reason :: term(), state()) :: :ok

    @callback handle_list_dbs(state()) :: result()
    @callback handle_get_schema(db :: String.t(), state()) :: result()
    @callback handle_transact(db :: String.t(), ops :: [map()], state()) :: result()
    @callback handle_cancel(id :: term(), state()) :: result()
    @callback handle_monitor(
                db :: String.t(),
                monitor_id :: term(),
                requests :: map(),
                state()
              ) :: result()
    @callback handle_monitor_cancel(monitor_id :: term(), state()) :: result()
    @callback handle_lock(lock :: String.t(), state()) :: result()
    @callback handle_steal(lock :: String.t(), state()) :: result()
    @callback handle_unlock(lock :: String.t(), state()) :: result()
    @callback handle_echo(args :: [term()], state()) :: result()

    @optional_callbacks terminate: 2,
                        handle_list_dbs: 1,
                        handle_get_schema: 2,
                        handle_transact: 3,
                        handle_cancel: 2,
                        handle_monitor: 4,
                        handle_monitor_cancel: 2,
                        handle_lock: 2,
                        handle_steal: 2,
                        handle_unlock: 2,
                        handle_echo: 2
  end

  @type t :: pid()

  defstruct [
    :transport,
    :handler_mod,
    :handler_state,
    # Used during the window between start_link and :socket_owned cast.
    :pending_socket,
    :pending_transport_mod
  ]

  # ---------------------------------------------------------------------------
  # Public API
  # ---------------------------------------------------------------------------

  @doc """
  Starts a ServerSession on an accepted socket. Wraps the socket
  in a Transport owned by the session.

  ## Options

    * `:transport` — `:gen_tcp` (default) or `:ssl`
    * `:handler` — module implementing `Handler` behaviour (required)
    * `:handler_opts` — options passed to `handler.init/1`
  """
  @spec start_link(port() | :ssl.sslsocket(), keyword()) ::
          {:ok, t()} | {:error, term()}
  def start_link(socket, opts) do
    GenServer.start_link(__MODULE__, {socket, opts})
  end

  @doc """
  Sends a notification to the connected client.

  Implemented as a cast so it's safe to call from within a handler
  callback (which runs in the session's own process — a synchronous
  `GenServer.call` would deadlock). Failures to send are logged;
  callers that need delivery guarantees should not use this path.
  """
  @spec notify(t(), String.t(), list()) :: :ok
  def notify(session, method, params) when is_binary(method) and is_list(params) do
    GenServer.cast(session, {:notify, method, params})
  end

  @doc "Stops the session (closes the connection)."
  @spec close(t()) :: :ok
  def close(session), do: GenServer.stop(session, :normal)

  # ---------------------------------------------------------------------------
  # GenServer callbacks
  # ---------------------------------------------------------------------------

  @impl true
  def init({socket, opts}) do
    handler_mod = Keyword.fetch!(opts, :handler)
    handler_opts = Keyword.get(opts, :handler_opts, [])
    transport_mod = Keyword.get(opts, :transport, :gen_tcp)

    case handler_mod.init(handler_opts) do
      {:ok, handler_state} ->
        # Don't touch the socket yet — the acceptor still owns it.
        # Wait for the :socket_owned cast before calling Transport.wrap.
        {:ok,
         %__MODULE__{
           transport: nil,
           handler_mod: handler_mod,
           handler_state: handler_state,
           pending_socket: socket,
           pending_transport_mod: transport_mod
         }}

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

  @impl true
  def handle_cast(:socket_owned, %{pending_socket: socket, pending_transport_mod: tm} = state) do
    case Transport.wrap(socket, transport: tm, controller: self()) do
      {:ok, transport} ->
        {:noreply, %{state | transport: transport, pending_socket: nil, pending_transport_mod: nil}}

      {:error, reason} ->
        Logger.warning("OVSDB.ServerSession failed to wrap socket: #{inspect(reason)}")
        {:stop, :normal, state}
    end
  end

  def handle_cast({:notify, method, params}, state) do
    notification = Protocol.notification(method, params)

    case Transport.send(state.transport, notification) do
      :ok ->
        {:noreply, state}

      {:error, reason} ->
        Logger.warning("OVSDB.ServerSession notify send failed: #{inspect(reason)}")
        {:noreply, state}
    end
  end

  @impl true
  def handle_info({:ovsdb_message, transport, msg}, %{transport: transport} = state) do
    case Protocol.classify(msg) do
      {:ok, {:request, %{id: id, method: method, params: params}}} ->
        {:noreply, dispatch_request(state, id, method, params)}

      {:ok, {:notification, %{method: "echo", params: params}}} ->
        # Client keepalive — reply with echo notification back.
        _ = Transport.send(transport, Protocol.notification("echo", params))
        {:noreply, state}

      {:ok, {:notification, %{method: method, params: _params}}} ->
        Logger.debug("OVSDB.ServerSession ignoring notification: #{method}")
        {:noreply, state}

      {:ok, {:response, _}} ->
        # Servers don't typically send requests that expect responses.
        Logger.warning("OVSDB.ServerSession got unexpected response")
        {:noreply, state}

      {:error, reason} ->
        Logger.warning("OVSDB.ServerSession received malformed: #{inspect(reason)}")
        {:noreply, state}
    end
  end

  def handle_info({:ovsdb_closed, transport}, %{transport: transport} = state) do
    {:stop, :normal, state}
  end

  def handle_info({:ovsdb_error, transport, reason}, %{transport: transport} = state) do
    Logger.warning("OVSDB.ServerSession transport error: #{inspect(reason)}")
    {:stop, :normal, state}
  end

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

  @impl true
  def terminate(reason, state) do
    if function_exported?(state.handler_mod, :terminate, 2) do
      _ = state.handler_mod.terminate(reason, state.handler_state)
    end

    :ok
  end

  # ---------------------------------------------------------------------------
  # Private — request dispatch
  # ---------------------------------------------------------------------------

  defp dispatch_request(state, id, method, params) do
    case do_dispatch(state.handler_mod, state.handler_state, method, params) do
      {:ok, result, new_state} ->
        send_response(state, Protocol.response(id, result))
        %{state | handler_state: new_state}

      {:error, error_string, new_state} ->
        send_response(state, Protocol.error_response(id, error_string))
        %{state | handler_state: new_state}

      :not_supported ->
        send_response(state, Protocol.error_response(id, "not supported"))
        state
    end
  end

  defp send_response(state, response) do
    case Transport.send(state.transport, response) do
      :ok ->
        :ok

      {:error, reason} ->
        Logger.warning("OVSDB.ServerSession failed to send response: #{inspect(reason)}")
    end
  end

  # Route methods to handler callbacks. Each method's expected
  # param shape is per RFC 7047 §4.1.
  defp do_dispatch(mod, state, "list_dbs", []) do
    if function_exported?(mod, :handle_list_dbs, 1) do
      mod.handle_list_dbs(state)
    else
      :not_supported
    end
  end

  defp do_dispatch(mod, state, "get_schema", [db]) when is_binary(db) do
    if function_exported?(mod, :handle_get_schema, 2) do
      mod.handle_get_schema(db, state)
    else
      :not_supported
    end
  end

  defp do_dispatch(mod, state, "transact", [db | ops]) when is_binary(db) do
    if function_exported?(mod, :handle_transact, 3) do
      mod.handle_transact(db, ops, state)
    else
      :not_supported
    end
  end

  defp do_dispatch(mod, state, "cancel", [id]) do
    if function_exported?(mod, :handle_cancel, 2) do
      mod.handle_cancel(id, state)
    else
      :not_supported
    end
  end

  defp do_dispatch(mod, state, "monitor", [db, monitor_id, requests])
       when is_binary(db) and is_map(requests) do
    if function_exported?(mod, :handle_monitor, 4) do
      mod.handle_monitor(db, monitor_id, requests, state)
    else
      :not_supported
    end
  end

  defp do_dispatch(mod, state, "monitor_cancel", [monitor_id]) do
    if function_exported?(mod, :handle_monitor_cancel, 2) do
      mod.handle_monitor_cancel(monitor_id, state)
    else
      :not_supported
    end
  end

  defp do_dispatch(mod, state, "lock", [lock]) when is_binary(lock) do
    if function_exported?(mod, :handle_lock, 2),
      do: mod.handle_lock(lock, state),
      else: :not_supported
  end

  defp do_dispatch(mod, state, "steal", [lock]) when is_binary(lock) do
    if function_exported?(mod, :handle_steal, 2),
      do: mod.handle_steal(lock, state),
      else: :not_supported
  end

  defp do_dispatch(mod, state, "unlock", [lock]) when is_binary(lock) do
    if function_exported?(mod, :handle_unlock, 2),
      do: mod.handle_unlock(lock, state),
      else: :not_supported
  end

  defp do_dispatch(mod, state, "echo", args) when is_list(args) do
    if function_exported?(mod, :handle_echo, 2) do
      mod.handle_echo(args, state)
    else
      # Echo has a sensible default: return the args unchanged.
      {:ok, args, state}
    end
  end

  defp do_dispatch(_mod, _state, _method, _params), do: :not_supported
end