HexPort

Hexagonal architecture ports for Elixir.

The problem

Clean Architecture and Hexagonal Architecture tell you to put your domain logic behind port boundaries — but Elixir doesn’t give you a standard way to define those boundaries. You end up with ad-hoc behaviours, manual delegation modules, test doubles that aren’t process-safe, and no way to log or inspect what crossed a boundary during a test.

What HexPort does

HexPort gives you a single macro — defport — that generates typed contracts, behaviours, and dispatch facades. In production, dispatch reads from application config. In tests, dispatch uses process-scoped handlers via NimbleOwnership (the same ownership library that Mox uses to track mock ownership — a battle-tested, low-risk dependency), giving you full async test isolation with zero global state.

Two entry points

HexPort has two macros for two separate concerns:

• use HexPort.Contract — defines the contract (pure interface definition). Generates X.Behaviour (callbacks) and X.__port_operations__/0 (introspection). No otp_app, no dispatch facade.

• use HexPort.Port, contract: X, otp_app: :my_app — generates the dispatch facade. Reads X.__port_operations__/0 at compile time and creates facade functions, bang variants, and key helpers. The consuming application controls which otp_app to use.

This separation means library-provided contracts (like HexPort.Repo) don’t hardcode an otp_app — the consuming application decides how dispatch works.

Quick example

Define a contract

defmodule MyApp.Todos do
  use HexPort.Contract

  defport get_todo(tenant_id :: String.t(), id :: String.t()) ::
    {:ok, Todo.t()} | {:error, term()}

  defport list_todos(tenant_id :: String.t()) :: [Todo.t()]

  defport create_todo!(params :: map()) :: Todo.t(), bang: false
end

This generates:

• MyApp.Todos.Behaviour — standard @behaviour with @callbacks
• MyApp.Todos.__port_operations__/0 — introspection for the contract

Generate a dispatch facade

# In a separate file (contract must compile first)
defmodule MyApp.Todos.Port do
  use HexPort.Port, contract: MyApp.Todos, otp_app: :my_app
end

This generates facade functions (get_todo/2, list_todos/1, create_todo!/1) that dispatch via HexPort.Dispatch.

Implement the behaviour

defmodule MyApp.Todos.Ecto do
  @behaviour MyApp.Todos.Behaviour

  @impl true
  def get_todo(tenant_id, id) do
    case MyApp.Repo.get_by(Todo, tenant_id: tenant_id, id: id) do
      nil -> {:error, :not_found}
      todo -> {:ok, todo}
    end
  end

  # ...
end

Configure for production

# config/config.exs
config :my_app, MyApp.Todos, impl: MyApp.Todos.Ecto

Test with process-scoped handlers

# test/test_helper.exs
HexPort.Testing.start()

# test/my_test.exs
defmodule MyApp.TodosTest do
  use ExUnit.Case, async: true

  setup do
    HexPort.Testing.set_fn_handler(MyApp.Todos, fn
      :get_todo, [_tenant, id] -> {:ok, %Todo{id: id, title: "Test"}}
      :list_todos, [_tenant] -> [%Todo{id: "1", title: "Test"}]
      :create_todo!, [params] -> struct!(Todo, params)
    end)
    :ok
  end

  test "gets a todo" do
    assert {:ok, %Todo{id: "42"}} = MyApp.Todos.Port.get_todo("t1", "42")
  end
end

Testing features

Function handlers

Map operations to return values with a simple function:

HexPort.Testing.set_fn_handler(MyApp.Todos, fn
  :get_todo, [_, id] -> {:ok, %Todo{id: id}}
  :list_todos, [_] -> []
end)

Stateful handlers

Maintain state across calls with atomic updates:

HexPort.Testing.set_stateful_handler(
  MyApp.Todos,
  fn
    :create_todo!, [params], state ->
      todo = struct!(Todo, params)
      {todo, %{state | todos: [todo | state.todos]}}

    :list_todos, [_tenant], state ->
      {state.todos, state}
  end,
  %{todos: []}   # initial state
)

See the InMemory adapter section below for the built-in stateful Repo implementation with its 3-stage dispatch model.

Dispatch logging

Record and inspect every call that crosses a port boundary:

setup do
  HexPort.Testing.enable_log(MyApp.Todos)
  HexPort.Testing.set_fn_handler(MyApp.Todos, fn
    :get_todo, [_, id] -> {:ok, %Todo{id: id}}
  end)
  :ok
end

test "logs dispatch calls" do
  MyApp.Todos.Port.get_todo("t1", "42")

  assert [{:get_todo, ["t1", "42"], {:ok, %Todo{id: "42"}}}] =
    HexPort.Testing.get_log(MyApp.Todos)
end

Async safety

All test handlers are process-scoped via NimbleOwnership. async: true tests run in full isolation. Task.async children automatically inherit their parent’s handlers.

HexPort.Testing.allow(MyApp.Todos, self(), some_pid)

Built-in Repo contract

HexPort includes a ready-made 15-operation Ecto Repo contract covering insert, update, delete, update_all, delete_all, get, get!, get_by, get_by!, one, one!, all, exists?, aggregate, and transact.

# HexPort.Repo defines the contract (no otp_app)
# Your app creates its own Port module:
defmodule MyApp.Repo.Port do
  use HexPort.Port, contract: HexPort.Repo, otp_app: :my_app
end

Three implementations are provided:

InMemory adapter

Repo.InMemory is a stateful test double that models a consistent store. It stores records in a nested map (Schema => %{pk => record}) and uses a 3-stage dispatch model for reads that reflects what the store can and cannot answer authoritatively.

The key insight: the InMemory store only contains records that have been explicitly inserted during the test. It is not a complete model of the logical store. When a record is not found in state, InMemory cannot know whether it exists in the logical store — so it must not silently return nil or []. Instead, it falls through to a user-supplied fallback function, or raises a clear error.

Operation categories

Basic usage — writes and PK reads

If your test only needs writes and PK-based lookups, no fallback is needed:

setup do
  HexPort.Testing.set_stateful_handler(
    HexPort.Repo,
    &HexPort.Repo.InMemory.dispatch/3,
    HexPort.Repo.InMemory.new()
  )
  :ok
end

test "insert then get by PK" do
  {:ok, user} = MyApp.Repo.Port.insert(User.changeset(%{name: "Alice"}))
  assert ^user = MyApp.Repo.Port.get(User, user.id)
end

Fallback function for non-PK reads

For operations the state cannot answer, supply a fallback_fn. It receives (operation, args) and returns the result. If it raises FunctionClauseError (no matching clause), dispatch falls through to a clear error:

setup do
  alice = %User{id: 1, name: "Alice", email: "alice@example.com"}

  state = HexPort.Repo.InMemory.new(
    seed: [alice],
    fallback_fn: fn
      :get_by, [User, [email: "alice@example.com"]] -> alice
      :all, [User] -> [alice]
      :exists?, [User] -> true
      :aggregate, [User, :count, :id] -> 1
    end
  )

  HexPort.Testing.set_stateful_handler(
    HexPort.Repo,
    &HexPort.Repo.InMemory.dispatch/3,
    state
  )
  :ok
end

test "PK read comes from state, non-PK reads use fallback" do
  # PK read — served from state
  assert %User{name: "Alice"} = MyApp.Repo.Port.get(User, 1)

  # Non-PK reads — served by fallback
  assert %User{name: "Alice"} = MyApp.Repo.Port.get_by(User, email: "alice@example.com")
  assert [%User{}] = MyApp.Repo.Port.all(User)
  assert MyApp.Repo.Port.exists?(User) == true
end

Error on unhandled operations

If an operation is not handled by the state (for PK reads) or the fallback function, InMemory raises ArgumentError with a message suggesting how to add a fallback clause:

** (ArgumentError) HexPort.Repo.InMemory cannot service :get_by with args [User, [name: "Bob"]].

    The InMemory adapter can only answer authoritatively for:
      - Write operations (insert, update, delete)
      - PK-based reads (get, get!) when the record exists in state

    For all other operations, register a fallback function:

        HexPort.Repo.InMemory.new(
          fallback_fn: fn
            :get_by, [User, [name: "Bob"]] -> # your result here
          end
        )

Transactions with transact

transact/2 mirrors Ecto.Repo.transact/2 — it accepts either a function or an Ecto.Multi as the first argument.

With a function:

MyApp.Repo.Port.transact(fn ->
  {:ok, user} = MyApp.Repo.Port.insert(user_changeset)
  {:ok, profile} = MyApp.Repo.Port.insert(profile_changeset(user))
  {:ok, {user, profile}}
end, [])

The function must return {:ok, result} or {:error, reason}.

With an Ecto.Multi:

Ecto.Multi.new()
|> Ecto.Multi.insert(:user, user_changeset)
|> Ecto.Multi.run(:profile, fn repo, %{user: user} ->
  repo.insert(profile_changeset(user))
end)
|> MyApp.Repo.Port.transact([])

On success, returns {:ok, changes} where changes is a map of operation names to results. On failure, returns {:error, failed_op, failed_value, changes_so_far}.

Multi :run callbacks receive the Port facade as the repo argument in Test and InMemory adapters (so repo.insert(cs) dispatches correctly), or the underlying Ecto Repo module in the Ecto adapter.

Supported Multi operations: insert, update, delete, run, put, error, inspect, merge, insert_all, update_all, delete_all. Bulk operations (insert_all, update_all, delete_all) return {0, nil} in the Test adapter. In InMemory, bulk operations go through the fallback function or raise (see InMemory adapter).

No transact! bang variant is generated (consistent with Ecto, which does not define Repo.transact! either).

Concurrency limitations of transact in test adapters

The Ecto adapter provides real database transactions with full ACID isolation — this is the production path and works correctly under concurrent access.

The Test and InMemory adapters do not provide true transaction isolation. In the Test adapter, transact simply calls the function (or steps through the Multi) without any locking. In the InMemory adapter, transact uses {:defer, fn} to avoid NimbleOwnership deadlocks — the function runs outside the lock, and each sub-operation acquires the lock individually.

This means:

• There is no rollback on error — side effects from earlier operations in a function-based transact are not undone (Multi-based transact in test adapters also does not roll back successful operations on failure).
• Concurrent writes to the InMemory store within a transaction are not isolated from each other.

This is acceptable for test-only adapters where transactions are typically exercised in serial, single-process tests. If your tests require true transaction isolation, use the Ecto adapter with a real database and Ecto’s sandbox.

Dispatch resolution

HexPort.Dispatch.call/4 resolves handlers in order:

1. Test handler — NimbleOwnership process-scoped lookup (zero-cost in production: GenServer.whereis returns nil when the ownership server isn’t started)
2. Application config — Application.get_env(otp_app, contract)[:impl]
3. Raise — clear error message if nothing is configured

Installation

Add hex_port to your dependencies in mix.exs:

def deps do
  [
    {:hex_port, "~> 0.1"}
  ]
end

Ecto is an optional dependency. If you want the built-in Repo contract, add Ecto to your own deps.

Relationship to Skuld

HexPort extracts the port system from Skuld (algebraic effects for Elixir) into a standalone library. You get typed contracts, async-safe test doubles, and dispatch logging without needing Skuld’s effect system. Skuld depends on HexPort and layers effectful dispatch on top.

License

MIT License - see LICENSE for details.