Add Mezzanine transport abstraction

Author: nshkrdotcom

bridges/mezzanine_bridge/lib/app_kit/bridges/mezzanine_bridge/transport.ex

@@ -0,0 +1,14 @@
+defmodule AppKit.Bridges.MezzanineBridge.Transport do
+  @moduledoc """
+  Caller-owned transport contract for AppKit calls into Mezzanine.
+
+  The same AppKit bridge can use an in-process Mezzanine facade, a local
+  distributed facade, or deterministic fixtures without changing the product
+  surface DTOs that AppKit owns.
+  """
+
+  @type result :: {:ok, map()} | {:error, map()}
+
+  @callback submit_work(request :: map(), opts :: keyword()) :: result()
+  @callback readback(ref :: String.t(), opts :: keyword()) :: result()
+end

bridges/mezzanine_bridge/lib/app_kit/bridges/mezzanine_bridge/transport/direct.ex

@@ -0,0 +1,80 @@
+defmodule AppKit.Bridges.MezzanineBridge.Transport.Direct do
+  @moduledoc """
+  In-process Mezzanine transport for monolith mode.
+
+  The target facade is supplied explicitly through `:target` or `:module`.
+  This keeps runtime authority out of ambient application environment.
+  """
+
+  @behaviour AppKit.Bridges.MezzanineBridge.Transport
+
+  @impl true
+  def submit_work(request, opts) when is_map(request) and is_list(opts) do
+    call(opts, :submit_work, [request, opts])
+  end
+
+  @impl true
+  def readback(ref, opts) when is_binary(ref) and is_list(opts) do
+    call(opts, :readback, [ref, opts])
+  end
+
+  defp call(opts, callback, args) do
+    with {:ok, target} <- fetch_target(opts),
+         {:ok, function} <- fetch_function(opts, callback),
+         {:ok, apply_args} <- apply_args(target, function, args) do
+      target
+      |> apply(function, apply_args)
+      |> normalize_result()
+    end
+  end
+
+  defp fetch_target(opts) do
+    case Keyword.get(opts, :target, Keyword.get(opts, :module)) do
+      target when is_atom(target) -> {:ok, target}
+      _other -> {:error, error(:missing_direct_target)}
+    end
+  end
+
+  defp fetch_function(opts, callback) do
+    function = Keyword.get(opts, function_option(callback), callback)
+
+    if is_atom(function) do
+      {:ok, function}
+    else
+      {:error, error(:invalid_direct_function)}
+    end
+  end
+
+  defp apply_args(target, function, args) do
+    args_without_opts = Enum.drop(args, -1)
+
+    cond do
+      function_exported?(target, function, length(args)) ->
+        {:ok, args}
+
+      function_exported?(target, function, length(args_without_opts)) ->
+        {:ok, args_without_opts}
+
+      true ->
+        {:error,
+         error(:direct_target_unavailable, %{
+           "target" => inspect(target),
+           "function" => Atom.to_string(function)
+         })}
+    end
+  end
+
+  defp function_option(:submit_work), do: :submit_work_function
+  defp function_option(:readback), do: :readback_function
+
+  defp normalize_result({:ok, result}) when is_map(result), do: {:ok, result}
+  defp normalize_result({:error, reason}) when is_map(reason), do: {:error, reason}
+  defp normalize_result(result) when is_map(result), do: {:ok, result}
+
+  defp normalize_result(reason),
+    do: {:error, error(:invalid_direct_response, %{"reason" => inspect(reason)})}
+
+  defp error(code, attrs \\ %{}) do
+    Map.merge(%{"code" => Atom.to_string(code), "transport" => "direct"}, attrs)
+  end
+end

bridges/mezzanine_bridge/lib/app_kit/bridges/mezzanine_bridge/transport/distributed.ex

@@ -0,0 +1,89 @@
+defmodule AppKit.Bridges.MezzanineBridge.Transport.Distributed do
+  @moduledoc """
+  Local distributed Mezzanine transport for StackLab proof topology.
+
+  This module uses bounded `:erpc` calls to owner facades. It is a transport
+  adapter, not a domain business API.
+  """
+
+  @behaviour AppKit.Bridges.MezzanineBridge.Transport
+
+  @default_timeout 5_000
+
+  @impl true
+  def submit_work(request, opts) when is_map(request) and is_list(opts) do
+    remote_call(opts, :submit_work, [request, opts])
+  end
+
+  @impl true
+  def readback(ref, opts) when is_binary(ref) and is_list(opts) do
+    remote_call(opts, :readback, [ref, opts])
+  end
+
+  defp remote_call(opts, callback, args) do
+    with {:ok, node} <- fetch_node(opts),
+         {:ok, module} <- fetch_module(opts),
+         {:ok, function} <- fetch_function(opts, callback) do
+      timeout = Keyword.get(opts, :timeout, @default_timeout)
+
+      try do
+        node
+        |> :erpc.call(module, function, args, timeout)
+        |> normalize_result()
+      rescue
+        exception -> {:error, error(:unreachable, %{"reason" => Exception.message(exception)})}
+      catch
+        :exit, reason -> {:error, error(classify_exit(reason), %{"reason" => inspect(reason)})}
+      end
+    end
+  end
+
+  defp fetch_node(opts) do
+    case Keyword.get(opts, :node) do
+      node when is_atom(node) -> {:ok, node}
+      _other -> {:error, error(:missing_node)}
+    end
+  end
+
+  defp fetch_module(opts) do
+    case Keyword.get(opts, :facade_module, Keyword.get(opts, :module)) do
+      module when is_atom(module) -> {:ok, module}
+      _other -> {:error, error(:missing_facade_module)}
+    end
+  end
+
+  defp fetch_function(opts, callback) do
+    function = Keyword.get(opts, function_option(callback), callback)
+
+    if is_atom(function) do
+      {:ok, function}
+    else
+      {:error, error(:invalid_remote_function)}
+    end
+  end
+
+  defp function_option(:submit_work), do: :submit_work_function
+  defp function_option(:readback), do: :readback_function
+
+  defp normalize_result({:ok, result}) when is_map(result), do: {:ok, result}
+  defp normalize_result({:error, reason}) when is_map(reason), do: {:error, reason}
+  defp normalize_result(result) when is_map(result), do: {:ok, result}
+
+  defp normalize_result(reason),
+    do: {:error, error(:invalid_remote_response, %{"reason" => inspect(reason)})}
+
+  defp classify_exit(reason) do
+    reason
+    |> inspect()
+    |> String.downcase()
+    |> String.contains?("timeout")
+    |> case do
+      true -> :timeout
+      false -> :unreachable
+    end
+  end
+
+  defp error(code, attrs \\ %{}) do
+    Map.merge(%{"code" => Atom.to_string(code), "transport" => "distributed"}, attrs)
+  end
+end

bridges/mezzanine_bridge/lib/app_kit/bridges/mezzanine_bridge/transport/fixture.ex

@@ -0,0 +1,56 @@
+defmodule AppKit.Bridges.MezzanineBridge.Transport.Fixture do
+  @moduledoc """
+  Deterministic AppKit-to-Mezzanine transport for tests.
+  """
+
+  @behaviour AppKit.Bridges.MezzanineBridge.Transport
+
+  @impl true
+  def submit_work(request, opts) when is_map(request) and is_list(opts) do
+    reply(opts, :submit_work, [request, opts], fn ->
+      {:ok,
+       %{
+         "status" => "accepted",
+         "accepted_ref" => Map.get(request, "idempotency_key", "work://fixture/accepted"),
+         "correlation_ref" => Map.get(request, "correlation_ref", "correlation://fixture/appkit")
+       }}
+    end)
+  end
+
+  @impl true
+  def readback(ref, opts) when is_binary(ref) and is_list(opts) do
+    reply(opts, :readback, [ref, opts], fn ->
+      {:ok, %{"status" => "completed", "accepted_ref" => ref}}
+    end)
+  end
+
+  defp reply(opts, callback, args, default) do
+    opts
+    |> configured_response(callback)
+    |> case do
+      nil -> default.()
+      fun when is_function(fun, length(args)) -> apply(fun, args)
+      fun when is_function(fun, length(args) - 1) -> apply(fun, Enum.drop(args, -1))
+      result -> result
+    end
+    |> normalize_result()
+  end
+
+  defp configured_response(opts, callback) do
+    responses = Keyword.get(opts, :responses, %{})
+
+    Keyword.get(opts, callback) || Map.get(responses, callback) ||
+      Map.get(responses, Atom.to_string(callback))
+  end
+
+  defp normalize_result({:ok, result}) when is_map(result), do: {:ok, result}
+  defp normalize_result({:error, reason}) when is_map(reason), do: {:error, reason}
+  defp normalize_result(result) when is_map(result), do: {:ok, result}
+
+  defp normalize_result(reason),
+    do: {:error, error(:invalid_fixture_response, %{"reason" => inspect(reason)})}
+
+  defp error(code, attrs) do
+    Map.merge(%{"code" => Atom.to_string(code), "transport" => "fixture"}, attrs)
+  end
+end

bridges/mezzanine_bridge/lib/app_kit/bridges/mezzanine_bridge/transport/runtime_deps.ex

@@ -0,0 +1,53 @@
+defmodule AppKit.Bridges.MezzanineBridge.Transport.RuntimeDeps do
+  @moduledoc """
+  Explicit runtime dependency holder for AppKit's Mezzanine transport.
+  """
+
+  alias AppKit.Bridges.MezzanineBridge.Transport
+
+  defstruct transport: Transport.Direct, transport_opts: []
+
+  @type t :: %__MODULE__{transport: module(), transport_opts: keyword()}
+
+  @spec new(keyword()) :: {:ok, t()} | {:error, map()}
+  def new(opts \\ []) when is_list(opts) do
+    transport = Keyword.get(opts, :transport, Transport.Direct)
+    transport_opts = Keyword.get(opts, :transport_opts, [])
+
+    with :ok <- validate_transport(transport),
+         :ok <- validate_transport_opts(transport_opts) do
+      {:ok, %__MODULE__{transport: transport, transport_opts: transport_opts}}
+    end
+  end
+
+  @spec submit_work(t(), map(), keyword()) :: Transport.result()
+  def submit_work(%__MODULE__{} = deps, request, opts \\ [])
+      when is_map(request) and is_list(opts) do
+    deps.transport.submit_work(request, Keyword.merge(deps.transport_opts, opts))
+  end
+
+  @spec readback(t(), String.t(), keyword()) :: Transport.result()
+  def readback(%__MODULE__{} = deps, ref, opts \\ []) when is_binary(ref) and is_list(opts) do
+    deps.transport.readback(ref, Keyword.merge(deps.transport_opts, opts))
+  end
+
+  defp validate_transport(transport) when is_atom(transport) do
+    case Code.ensure_loaded(transport) do
+      {:module, ^transport} ->
+        if function_exported?(transport, :submit_work, 2) and
+             function_exported?(transport, :readback, 2) do
+          :ok
+        else
+          {:error, %{"code" => "invalid_transport", "transport" => inspect(transport)}}
+        end
+
+      _other ->
+        {:error, %{"code" => "invalid_transport", "transport" => inspect(transport)}}
+    end
+  end
+
+  defp validate_transport(_transport), do: {:error, %{"code" => "invalid_transport"}}
+
+  defp validate_transport_opts(opts) when is_list(opts), do: :ok
+  defp validate_transport_opts(_opts), do: {:error, %{"code" => "invalid_transport_opts"}}
+end

bridges/mezzanine_bridge/test/app_kit/bridges/mezzanine_bridge/transport_test.exs

@@ -0,0 +1,56 @@
+defmodule AppKit.Bridges.MezzanineBridge.TransportTest do
+  use ExUnit.Case, async: true
+
+  alias AppKit.Bridges.MezzanineBridge.Transport
+
+  defmodule DirectTarget do
+    def submit_work(request, opts) do
+      {:ok, %{"mode" => "direct", "request" => request, "timeout" => opts[:timeout]}}
+    end
+
+    def readback(ref), do: {:ok, %{"mode" => "direct", "ref" => ref}}
+  end
+
+  test "direct transport calls an explicitly supplied in-process target" do
+    assert {:ok, result} =
+             Transport.Direct.submit_work(%{"work" => "one"}, target: DirectTarget, timeout: 25)
+
+    assert result["mode"] == "direct"
+    assert result["timeout"] == 25
+
+    assert {:ok, %{"ref" => "work://1"}} =
+             Transport.Direct.readback("work://1", target: DirectTarget)
+  end
+
+  test "distributed transport calls an explicitly supplied owner facade" do
+    assert {:ok, result} =
+             Transport.Distributed.submit_work(%{"work" => "one"},
+               node: Node.self(),
+               facade_module: DirectTarget,
+               timeout: 1_000
+             )
+
+    assert result["mode"] == "direct"
+  end
+
+  test "fixture transport is deterministic and overridable" do
+    assert {:ok, %{"accepted_ref" => "idem-1"}} =
+             Transport.Fixture.submit_work(%{"idempotency_key" => "idem-1"}, [])
+
+    assert {:ok, %{"status" => "custom"}} =
+             Transport.Fixture.readback("work://1",
+               responses: %{readback: fn ref -> {:ok, %{"status" => "custom", "ref" => ref}} end}
+             )
+  end
+
+  test "runtime deps select a transport explicitly" do
+    assert {:ok, deps} =
+             Transport.RuntimeDeps.new(
+               transport: Transport.Fixture,
+               transport_opts: [submit_work: {:ok, %{"accepted_ref" => "fixture://work"}}]
+             )
+
+    assert {:ok, %{"accepted_ref" => "fixture://work"}} =
+             Transport.RuntimeDeps.submit_work(deps, %{})
+  end
+end