# N2O Framework Evaluation for MW-Core
## Can `synrc/n2o` Close the MQTT & Protocol Gaps?

> **Evaluation context:** MW-Core currently has no MQTT support and several other protocol gaps identified in [`docs/middleware-comparison.md`](middleware-comparison.md) and the [`docs/protocol-connectivity-implementation-plan.md`](protocol-connectivity-implementation-plan.md). This document evaluates [N2O (synrc/n2o)](https://github.com/synrc/n2o) as a candidate to address those gaps — specifically MQTT — and identifies exactly where it complements MW-Core versus where it conflicts.

---

## Table of Contents

1. [What N2O Is (Technical Summary)](#1-what-n2o-is-technical-summary)
2. [N2O Ecosystem Map](#2-n2o-ecosystem-map)
3. [MQTT Gap — Why It Matters for MW-Core](#3-mqtt-gap--why-it-matters-for-mw-core)
4. [Fit Assessment Matrix](#4-fit-assessment-matrix)
5. [Integration Point 1 — `gateway_mqtt` (Highest Value)](#5-integration-point-1--gateway_mqtt-highest-value)
6. [Integration Point 2 — `n2o_ring` for Multi-Node Dispatch](#6-integration-point-2--n2o_ring-for-multi-node-dispatch)
7. [Integration Point 3 — BPE for BPMN 2.0 Flow Builder](#7-integration-point-3--bpe-for-bpmn-20-flow-builder)
8. [Integration Point 4 — `adapter_mqtt` (Alternative Simpler Path)](#8-integration-point-4--adapter_mqtt-alternative-simpler-path)
9. [What NOT to Adopt from N2O](#9-what-not-to-adopt-from-n2o)
10. [MQTT Options Comparison: N2O vs Direct EMQX Client vs HiveMQ](#10-mqtt-options-comparison-n2o-vs-direct-emqx-client-vs-hivemq)
11. [Recommended Architecture](#11-recommended-architecture)
12. [Implementation Plan — `gateway_mqtt` with N2O](#12-implementation-plan--gateway_mqtt-with-n2o)
13. [Risk Register](#13-risk-register)
14. [Verdict](#14-verdict)

---

## 1. What N2O Is (Technical Summary)

N2O is an **embeddable, protocol-agnostic message loop library** written in Erlang (~90%) with Elixir support (~10%). It is not a web framework — it is a **thin protocol relay and process orchestration layer** that runs on top of server hosts (Cowboy, EMQ, Mochiweb) and provides a unified API for sessions, message queues, caching, encoding, and distributed node coordination.

### Key Characteristics

| Property | Value |
|---|---|
| **Core size** | ~700 lines of Erlang + ~500 lines of JavaScript |
| **External dependencies** | Zero (`rebar.config: {deps, []}`) |
| **OTP compatibility** | OTP 25, 26, 27, 28 (CI-verified) |
| **Elixir / Mix support** | Yes — available as `:n2o` on Hex |
| **License** | ISC (per-file, permissive) |
| **Current version** | 13.4.15 |
| **Protocols** | WebSocket, MQTT, TCP, HTTP, FTP, UDP, QUIC (experimental) |
| **Pub/sub backends** | GPROC, SYN, PG2/PG (pluggable) |
| **Encoding** | BERT (default), JSON, ASN.1, XML, MessagePack (pluggable) |
| **Encryption** | AES/GCM-256 built-in via `n2o_secret` |
| **Production use** | Ukrainian banking, ERP, healthcare (HL7), government PKI |

### The Core Abstraction — `n2o_proto`

N2O's heart is a **handler-chain dispatch loop** — conceptually very close to MW-Core's own Plug pipeline:

```
Incoming frame (WS / MQTT / TCP)
         │
         ▼
  n2o_proto.push/5
         │
   ┌─────▼──────┐   {unknown,...}   ┌─────────────┐
   │ n2o_heart  │ ─────────────────► │ your_handler│
   └─────┬──────┘                   └──────┬───────┘
         │ {reply,...}                     │ {reply,...}
         ▼                                 ▼
    send frame back                  send frame back
```

Each handler returns `{reply, Message, Request, State}` to claim the message or `{unknown, ...}` to pass it to the next handler. This is structurally identical to how MW-Core's `MwRouter.Pipeline` dispatches through Plug stages.

---

## 2. N2O Ecosystem Map

N2O is the kernel of a wider ecosystem. The projects most relevant to MW-Core are highlighted:

```
synrc/n2o  (kernel — EVALUATE ✅)
  │
  ├── synrc/mqtt      → MQTT protocol handler    (EVALUATE ✅)
  ├── synrc/n2o_ring  → Consistent hash ring     (EVALUATE ✅)
  ├── synrc/bpmn      → BPMN 2.0 workflow engine (EVALUATE ✅)
  ├── synrc/kvs       → Abstract key-value store (LOW PRIORITY)
  ├── synrc/nitro     → Server-side HTML UI      (NOT RELEVANT — conflicts with LiveView)
  ├── synrc/rest      → Cowboy REST framework    (NOT RELEVANT — MW-Core uses Plug/Phoenix)
  ├── synrc/rpc       → gRPC/BERT/SOAP RPC       (COVERED — Phase 2 plan has adapter_grpc)
  └── synrc/chat      → IM platform (X.509)      (NOT RELEVANT)
```

---

## 3. MQTT Gap — Why It Matters for MW-Core

MQTT (ISO/IEC 20922) is the dominant protocol for:

| Use Case | Relevance to MW-Core |
|---|---|
| **IoT / POS terminal connectivity** | POS devices at merchant locations typically speak MQTT over cellular |
| **Mobile payment push** | QR-code scanners, tap-to-pay terminals, soft POS apps use MQTT for async status updates |
| **Branch / ATM telemetry** | ATM health metrics, cash levels, fault alerts — MQTT pub/sub |
| **Partner bank systems** | Some correspondent banking APIs use MQTT for low-latency transaction confirmations |
| **Audit log streaming** | Lightweight MQTT-based streaming to SIEM systems (Splunk, ELK) |

Without MQTT, MW-Core forces all device/terminal clients to either:
- Poll via REST (high latency, high server load), or
- Use WebSocket with a custom framing protocol (works but non-standard for embedded devices)

MQTT's lightweight binary framing (2-byte fixed header, QoS 0/1/2) is designed exactly for constrained devices where Phoenix Channels would be too heavy.

**Current MW-Core state:**

```
gateway_ws (Phoenix Channels) — works for browser/mobile WebSocket clients
     ✅ handles 100,000+ concurrent connections
     ❌ NOT the protocol POS terminals and IoT devices speak
     ❌ requires custom framing; no standard MQTT client library support
```

**Target state with N2O:**

```
gateway_mqtt (N2O + EMQ) — new north-plane gateway for MQTT devices
     ✅ standard MQTT 3.1.1 / 5.0 protocol
     ✅ QoS 0 (at-most-once), QoS 1 (at-least-once), QoS 2 (exactly-once)
     ✅ retained messages, will messages, topic-based routing
     ✅ bridges into existing mw_router pipeline
     ✅ audit, auth, rate limiting unchanged
```

---

## 4. Fit Assessment Matrix

For each N2O component, the score considers: **protocol alignment** (does it fill a real gap?), **integration effort** (how hard to wire into existing MW-Core?), and **risk** (API stability, maintenance burden).

| N2O Component | What It Does | MW-Core Gap It Fills | Fit Score | Recommendation |
|---|---|---|---|---|
| **`n2o` kernel + MQTT** | Protocol loop + MQTT listener | MQTT gateway (missing entirely) | ⭐⭐⭐⭐⭐ | **Adopt — `gateway_mqtt`** |
| **`n2o_ring`** | Consistent hash ring for distributed dispatch | Multi-node Kafka/MQTT topic partitioning | ⭐⭐⭐ | **Selective adopt — augment `mw_router`** |
| **`synrc/bpmn` (BPE)** | BPMN 2.0 workflow engine | Flow Builder limited to custom DAG; no BPMN import | ⭐⭐⭐ | **Evaluate for Phase 3** |
| **`n2o_pi`** | Named gen_server replacement | None — OTP supervisors work fine | ⭐ | **Do not adopt** |
| **`n2o_heart` over WS** | WebSocket PING protocol | Phoenix already handles heartbeats | ⭐ | **Do not adopt** |
| **`synrc/nitro`** | Server-side HTML/UI framework | Conflicts with Phoenix LiveView admin UI | ❌ | **Reject — direct conflict** |
| **`synrc/rest`** | Cowboy REST framework | Conflicts with existing Plug/Phoenix REST | ❌ | **Reject — direct conflict** |
| **`synrc/kvs`** | Abstract key-value storage | `infra_cache` (ETS + Redis) already works | ⭐ | **Do not adopt** |
| **`synrc/rpc`** | gRPC/BERT/SOAP | Phase 2 already covers `adapter_grpc` | ⭐⭐ | **Consider as alternative to Phase 2** |

**Legend:** ⭐⭐⭐⭐⭐ Strongly adopt · ⭐⭐⭐ Selective adopt · ⭐ Skip · ❌ Reject

---

## 5. Integration Point 1 — `gateway_mqtt` (Highest Value)

### Architecture

N2O provides two complementary pieces for MQTT:
1. **N2O MQTT protocol handler** — the `n2o_proto` loop with MQTT message type handlers
2. **EMQ (EMQX) broker** — runs as a separate Docker container; N2O connects via Erlang/MQTT bridge

The recommended architecture for MW-Core is a **bridged gateway** — not embedding an MQTT broker inside the BEAM node, but running EMQX externally and connecting N2O as a bridge subscriber that feeds messages into the existing `mw_router` pipeline:

```
MQTT Devices (POS terminals, IoT, mobile)
        │  TCP :1883 / TLS :8883 / WS :8083
        ▼
┌──────────────────────────────────┐
│  EMQX Broker (Docker container)  │  ← handles MQTT connection lifecycle,
│  Port: 1883 (MQTT)               │    QoS guarantees, will messages,
│  Port: 8883 (MQTT/TLS)           │    retained messages, topic ACL
│  Port: 8083 (MQTT-over-WS)       │
└──────────────────┬───────────────┘
                   │ EMQX ExHook / MQTT bridge
                   ▼
┌──────────────────────────────────────────────────────┐
│  gateway_mqtt  (new umbrella app — MW-Core BEAM node) │
│                                                       │
│  N2O MQTT loop:                                       │
│    n2o:subscribe(topic) → n2o_proto.push(msg)        │
│         │                                             │
│         ▼                                             │
│  MqttTransformer.from_mqtt/1                          │
│    → MwKernel.Message                                 │
│         │                                             │
│         ▼                                             │
│  MwRouter.Dispatcher.dispatch/1  ← same pipeline     │
│    → mw_auth, mw_router, mw_transform, mw_audit      │
│         │                                             │
│         ▼ response                                    │
│  n2o:publish(response_topic, payload)                 │
└──────────────────────────────────────────────────────┘
```

### Why EMQX externally rather than N2O as broker

| Approach | EMQX external + N2O bridge | N2O as sole broker |
|---|---|---|
| **QoS 2 (exactly-once)** | ✅ EMQX handles session state | ⚠️ N2O delegates QoS to EMQ anyway |
| **MQTT 5.0** | ✅ EMQX fully supports MQTT 5.0 | ⚠️ N2O's MQTT support targets 3.1.1 |
| **Persistent sessions** | ✅ EMQX persists across broker restarts | ❌ N2O's ETS sessions are ephemeral |
| **Topic ACL / security** | ✅ EMQX has full ACL + TLS + OAuth 2.0 | ⚠️ N2O defers to MW-Core auth |
| **Scaling MQTT independently** | ✅ EMQX cluster scales separately | ❌ MQTT tied to BEAM node count |
| **Ops simplicity** | ⚠️ Two containers to operate | ✅ Single BEAM node |
| **Best for MW-Core** | ✅ Matches existing multi-container K8s model | ❌ Over-complicates BEAM release |

**Decision: Run EMQX as a sidecar/separate container. Use N2O's bridge client in `gateway_mqtt` to subscribe and publish.**

### New umbrella app structure

```
apps/gateway_mqtt/
├── lib/
│   ├── gateway_mqtt.ex              # OTP Application entry point
│   ├── gateway_mqtt/
│   │   ├── application.ex           # Supervisor: N2O bridge + subscription manager
│   │   ├── bridge.ex                # N2O MQTT connection to EMQX + subscription loop
│   │   ├── subscription_manager.ex  # GenServer: manages topic → route_type mappings
│   │   ├── transformer.ex           # MQTT payload ↔ MwKernel.Message
│   │   ├── publisher.ex             # Publishes responses back to MQTT topics
│   │   └── telemetry.ex             # :telemetry events for Prometheus
├── test/
│   ├── gateway_mqtt_test.exs
│   └── support/
│       └── mqtt_mock.ex
└── mix.exs
```

### Dependencies

```elixir
# apps/gateway_mqtt/mix.exs
defp deps do
  [
    {:mw_kernel,        in_umbrella: true},
    {:mw_auth,          in_umbrella: true},
    {:mw_router,        in_umbrella: true},
    {:mw_audit,         in_umbrella: true},
    {:infra_telemetry,  in_umbrella: true},
    {:n2o,              "~> 13.4"},      # N2O kernel + MQTT loop
    {:emqtt,            "~> 1.6"},       # EMQX official Erlang/Elixir MQTT client
    {:jason,            "~> 1.4"},
    {:telemetry,        "~> 1.2"}
  ]
end
```

> **`emqtt` vs N2O's internal MQTT client:**  
> N2O's MQTT handling delegates to EMQ at the broker level. For the **client connection** from MW-Core to EMQX, `emqtt` (EMQX's official Erlang client, available on Hex) is more idiomatic and better maintained than using N2O's bridge internals. Use N2O for the **protocol loop and handler chain** (where it excels at zero cost); use `emqtt` for the broker connection.

### Core Implementation

**Bridge GenServer (N2O + emqtt)**

```elixir
# apps/gateway_mqtt/lib/gateway_mqtt/bridge.ex
defmodule GatewayMqtt.Bridge do
  use GenServer
  require Logger

  alias GatewayMqtt.{Transformer, Publisher}
  alias MwRouter.Dispatcher

  @reconnect_interval 5_000

  # ── Public API ──────────────────────────────────────────────────────────────

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)

  def subscribe(topic, qos \\ 1) do
    GenServer.call(__MODULE__, {:subscribe, topic, qos})
  end

  # ── Callbacks ───────────────────────────────────────────────────────────────

  @impl true
  def init(_opts) do
    send(self(), :connect)
    {:ok, %{client: nil, subscriptions: []}}
  end

  @impl true
  def handle_info(:connect, state) do
    emqx_host = Application.fetch_env!(:gateway_mqtt, :emqx_host)
    emqx_port = Application.get_env(:gateway_mqtt, :emqx_port, 1883)

    case :emqtt.start_link([
      host:         emqx_host,
      port:         emqx_port,
      clientid:     "mw-core-#{node()}",
      username:     Application.get_env(:gateway_mqtt, :emqx_username, "mw-core"),
      password:     Application.get_env(:gateway_mqtt, :emqx_password, ""),
      clean_start:  false,
      keepalive:    60,
      reconnect:    true
    ]) do
      {:ok, client} ->
        {:ok, _connack} = :emqtt.connect(client)
        Logger.info("[GatewayMqtt] Connected to EMQX at #{emqx_host}:#{emqx_port}")

        # Re-subscribe to all previously registered topics after reconnect
        Enum.each(state.subscriptions, fn {topic, qos} ->
          :emqtt.subscribe(client, topic, qos)
        end)

        {:noreply, %{state | client: client}}

      {:error, reason} ->
        Logger.warning("[GatewayMqtt] Connection failed: #{inspect(reason)}, retrying in #{@reconnect_interval}ms")
        Process.send_after(self(), :connect, @reconnect_interval)
        {:noreply, state}
    end
  end

  # Incoming MQTT message — this is the hot path
  @impl true
  def handle_info({:publish, %{topic: topic, payload: payload, qos: qos} = packet}, state) do
    start_time = System.monotonic_time()

    case Transformer.from_mqtt(topic, payload) do
      {:ok, mw_message} ->
        case Dispatcher.dispatch(mw_message) do
          {:ok, response} ->
            # Publish response to the reply topic (request/response pattern)
            reply_topic = "#{topic}/response"
            Publisher.publish(state.client, reply_topic, Transformer.to_mqtt(response), qos)

          {:error, err} ->
            Logger.warning("[GatewayMqtt] Dispatch failed for topic #{topic}: #{inspect(err)}")
            Publisher.publish_error(state.client, topic, err)
        end

      {:error, reason} ->
        Logger.warning("[GatewayMqtt] Cannot parse MQTT payload on #{topic}: #{inspect(reason)}")
    end

    :telemetry.execute(
      [:gateway_mqtt, :message],
      %{count: 1, duration: System.monotonic_time() - start_time},
      %{topic: topic, qos: qos}
    )

    {:noreply, state}
  end

  # Disconnected — trigger reconnect
  @impl true
  def handle_info({:disconnected, _reason, _}, state) do
    Logger.warning("[GatewayMqtt] Disconnected from EMQX, reconnecting...")
    Process.send_after(self(), :connect, @reconnect_interval)
    {:noreply, %{state | client: nil}}
  end

  @impl true
  def handle_call({:subscribe, topic, qos}, _from, %{client: client} = state) when not is_nil(client) do
    :emqtt.subscribe(client, topic, qos)
    {:reply, :ok, %{state | subscriptions: [{topic, qos} | state.subscriptions]}}
  end
end
```

**Transformer — MQTT topic/payload → `MwKernel.Message`**

```elixir
# apps/gateway_mqtt/lib/gateway_mqtt/transformer.ex
defmodule GatewayMqtt.Transformer do
  alias MwKernel.{Message, Context}

  # Topic convention: mw/{tenant_id}/{message_type}
  # Example: mw/tenant_acme/transaction.payment
  @topic_regex ~r|^mw/(?<tenant>[^/]+)/(?<type>[^/]+)$|

  def from_mqtt(topic, payload) when is_binary(payload) do
    with {:ok, %{"tenant" => tenant_id, "type" => msg_type}} <- parse_topic(topic),
         {:ok, body} <- Jason.decode(payload),
         type        <- String.to_existing_atom(msg_type) do

      msg = Message.new(
        type,
        body["payload"] || %{},
        :gateway_mqtt,
        %Context{
          trace_id:    body["trace_id"] || generate_trace_id(),
          tenant_id:   tenant_id,
          user:        body["client_id"] || "mqtt_device",
          roles:       ["device"]
        }
      )
      {:ok, msg}
    else
      _ -> {:error, :invalid_mqtt_message}
    end
  end

  def to_mqtt(%Message{payload: p, id: id, context: ctx}) do
    Jason.encode!(%{
      id:        id,
      status:    p[:status] || "ok",
      payload:   p,
      trace_id:  ctx.trace_id,
      timestamp: DateTime.utc_now() |> DateTime.to_iso8601()
    })
  end

  defp parse_topic(topic) do
    case Regex.named_captures(@topic_regex, topic) do
      nil    -> {:error, :invalid_topic}
      result -> {:ok, result}
    end
  end

  defp generate_trace_id, do: :crypto.strong_rand_bytes(16) |> Base.encode16(case: :lower)
end
```

**Subscription Manager — topic ↔ route mapping**

```elixir
# apps/gateway_mqtt/lib/gateway_mqtt/subscription_manager.ex
defmodule GatewayMqtt.SubscriptionManager do
  use GenServer

  @default_subscriptions [
    {"mw/+/transaction.payment",  1},   # QoS 1 — payment requests
    {"mw/+/transaction.inquiry",  0},   # QoS 0 — balance queries (lossy ok)
    {"mw/+/device.telemetry",     0},   # QoS 0 — POS health metrics
    {"mw/+/auth.token.refresh",   1}    # QoS 1 — token refresh from devices
  ]

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)

  @impl true
  def init(_opts) do
    # Subscribe after bridge is connected
    send(self(), :subscribe_defaults)
    {:ok, %{}}
  end

  @impl true
  def handle_info(:subscribe_defaults, state) do
    Enum.each(@default_subscriptions, fn {topic, qos} ->
      GatewayMqtt.Bridge.subscribe(topic, qos)
    end)
    {:noreply, state}
  end
end
```

**Publisher**

```elixir
# apps/gateway_mqtt/lib/gateway_mqtt/publisher.ex
defmodule GatewayMqtt.Publisher do
  def publish(client, topic, payload, qos \\ 1) when is_binary(payload) do
    :emqtt.publish(client, topic, payload, qos)
  end

  def publish_error(client, topic, %MwKernel.Error{} = err) do
    error_topic = "#{topic}/error"
    payload = Jason.encode!(%{
      error: err.code,
      detail: err.detail,
      timestamp: DateTime.utc_now() |> DateTime.to_iso8601()
    })
    :emqtt.publish(client, error_topic, payload, 1)
  end
end
```

**Application supervisor**

```elixir
# apps/gateway_mqtt/lib/gateway_mqtt/application.ex
defmodule GatewayMqtt.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      GatewayMqtt.Bridge,
      GatewayMqtt.SubscriptionManager
    ]
    Supervisor.start_link(children, strategy: :one_for_one, name: GatewayMqtt.Supervisor)
  end
end
```

**Config**

```elixir
# config/config.exs
config :gateway_mqtt,
  emqx_host:     "localhost",
  emqx_port:     1883,
  emqx_username: "mw-core",
  emqx_password: ""

# config/runtime.exs
config :gateway_mqtt,
  emqx_host:     System.get_env("EMQX_HOST", "emqx"),
  emqx_port:     System.get_env("EMQX_PORT", "1883") |> String.to_integer(),
  emqx_username: System.get_env("EMQX_USERNAME", "mw-core"),
  emqx_password: System.get_env("EMQX_PASSWORD", "")
```

**Kubernetes sidecar / Docker Compose**

```yaml
# docker-compose.yml (add to existing services)
services:
  emqx:
    image: emqx/emqx:5.6.0
    ports:
      - "1883:1883"     # MQTT
      - "8883:8883"     # MQTT/TLS
      - "8083:8083"     # MQTT-over-WebSocket
      - "18083:18083"   # EMQX Dashboard
    environment:
      EMQX_NODE_NAME:            "emqx@localhost"
      EMQX_DASHBOARD__DEFAULT_PASSWORD: "${EMQX_DASHBOARD_PASSWORD}"
    volumes:
      - emqx_data:/opt/emqx/data
```

**Topic naming convention for MW-Core**

```
mw/{tenant_id}/{message_type}

Request (device → MW-Core):   mw/tenant_acme/transaction.payment
Response (MW-Core → device):  mw/tenant_acme/transaction.payment/response
Error:                         mw/tenant_acme/transaction.payment/error

Device telemetry (QoS 0):     mw/tenant_acme/device.telemetry
Token refresh (QoS 1):        mw/tenant_acme/auth.token.refresh
```

---

## 6. Integration Point 2 — `n2o_ring` for Multi-Node Dispatch

### What it does

`n2o_ring` is a **consistent hash ring** (backed by `gb_trees`) that distributes work across virtual nodes:

```erlang
% n2o_ring API
n2o_ring:add(Name, Node, VNodes)    % add a physical node
n2o_ring:remove(Name, Node)         % remove a node
n2o_ring:lookup(Name, Key)          % find responsible node for a key
n2o_ring:members(Name)              % list all nodes
```

### Where it helps MW-Core

MW-Core already uses `libcluster` for node discovery and ETS-based routing. `n2o_ring` would be **additive** — not a replacement — for these two specific problems:

| Problem | Current MW-Core approach | With `n2o_ring` |
|---|---|---|
| **Kafka partition → node assignment** | Broadway processes all partitions on one node | Ring maps partition key to responsible BEAM node; each node processes its own partitions |
| **MQTT topic → node assignment** | `gateway_mqtt` on every node subscribes to every topic | Ring maps tenant_id hash to a single responsible node; avoids duplicate processing |
| **Stateful adapter session affinity** | None — adapters are stateless by design | Ring ensures same device/tenant always hits same node (useful for connection pooling) |

### How to integrate

```elixir
# apps/mw_router/lib/mw_router/ring.ex
defmodule MwRouter.Ring do
  @ring_name :mw_ring
  @vnodes    10   # virtual nodes per physical node

  def init do
    :n2o_ring.create(@ring_name)
    # Register self on startup; libcluster will trigger for other nodes
    :n2o_ring.add(@ring_name, node(), @vnodes)
  end

  def responsible_node(tenant_id) do
    :n2o_ring.lookup(@ring_name, :erlang.phash2(tenant_id))
  end

  def add_node(node), do: :n2o_ring.add(@ring_name, node, @vnodes)
  def remove_node(node), do: :n2o_ring.remove(@ring_name, node)
end
```

Wire into `libcluster` node up/down events:

```elixir
# apps/mw_router/lib/mw_router/cluster_observer.ex
defmodule MwRouter.ClusterObserver do
  use GenServer

  @impl true
  def init(_), do: {:ok, nil, {:continue, :subscribe}}

  @impl true
  def handle_continue(:subscribe, state) do
    :net_kernel.monitor_nodes(true)
    MwRouter.Ring.init()
    {:noreply, state}
  end

  @impl true
  def handle_info({:nodeup, node},   state) do
    MwRouter.Ring.add_node(node)
    {:noreply, state}
  end

  @impl true
  def handle_info({:nodedown, node}, state) do
    MwRouter.Ring.remove_node(node)
    {:noreply, state}
  end
end
```

**Effort:** ~1 week. **Risk:** Low — `n2o_ring` is read-only at lookup time; failure falls back gracefully to local dispatch.

---

## 7. Integration Point 3 — BPE for BPMN 2.0 Flow Builder

### What `synrc/bpmn` is

`synrc/bpmn` (BPE — Business Process Engine) is a BPMN 2.0 compliant process orchestration engine built on the BEAM. It manages process instances with states: `Created → Active → Suspended → Completed/Terminated`. Each step maps to a BPMN element (startEvent, userTask, serviceTask, gateway, endEvent).

### Where it complements MW-Core's Flow Builder

MW-Core's current Flow Builder (`gateway_web`) uses a custom DAG — it is a proprietary graph format. BPE would allow:

| Capability | Current Flow Builder | With BPE |
|---|---|---|
| **Import from standard BPMN tool** | ❌ Custom JSON format only | ✅ Import `.bpmn` files from Camunda, Bizagi, Draw.io |
| **BPMN compliance** | ❌ None | ✅ ISO 19510:2015 |
| **Human-task / approval steps** | ❌ Not modelled | ✅ `userTask` with assignment and deadline |
| **Timer boundary events** | ❌ Not available | ✅ `timerEvent` for SLA expiry handling |
| **Compensation / rollback flows** | ❌ Manual error handling | ✅ BPMN compensation events |
| **Workflow persistence** | ✅ DB-persisted DAG | ✅ Mnesia-backed process state machine |

### Integration approach (Phase 3 — non-blocking)

BPE would be a **new UI feature** in `gateway_web` alongside the existing Flow Builder — operators could choose between the custom DAG editor and a BPMN 2.0 importer. The BPMN process steps would resolve to `MwKernel.Adapter` modules at execution time, exactly like DAG nodes today.

```elixir
# BPE service task → adapter dispatch bridge
defmodule GatewayWeb.BpeAdapterBridge do
  def execute_service_task(%{adapter: module_name} = task, context) do
    adapter = Module.concat([module_name])
    {:ok, state} = adapter.connect(%{})
    msg = MwKernel.Message.new(task.message_type, task.payload, :bpe, context)
    adapter.send(state, msg)
  end
end
```

**Effort:** 3–4 weeks. **Dependency:** Requires Mnesia (add to umbrella) or adapt BPE to use `infra_repo` (MySQL).

---

## 8. Integration Point 4 — `adapter_mqtt` (Alternative Simpler Path)

If the full `gateway_mqtt` (north-side gateway) is too large a scope for immediate delivery, a **south-side MQTT adapter** is a simpler starting point. This follows the exact pattern from `adapter_development_guide.md` — MW-Core publishes to an MQTT topic as a downstream notification channel.

```elixir
# apps/adapter_mqtt/lib/adapter_mqtt.ex
defmodule AdapterMqtt do
  @behaviour MwKernel.Adapter

  alias AdapterMqtt.Transformer

  @impl MwKernel.Adapter
  def connect(config) do
    host  = config[:host]  || Application.fetch_env!(:adapter_mqtt, :emqx_host)
    port  = config[:port]  || Application.get_env(:adapter_mqtt, :emqx_port, 1883)
    topic = config[:topic] || Application.fetch_env!(:adapter_mqtt, :default_topic)

    {:ok, client} = :emqtt.start_link(host: host, port: port,
                                       clientid: "mw-adapter-#{inspect(self())}")
    {:ok, _}      = :emqtt.connect(client)
    {:ok, %{client: client, topic: topic}}
  end

  @impl MwKernel.Adapter
  def send(%{client: client, topic: topic} = _state, %MwKernel.Message{} = msg) do
    payload = Transformer.to_mqtt(msg)
    case :emqtt.publish(client, topic, payload, _qos = 1) do
      :ok              -> {:ok, msg}
      {:error, reason} -> {:error, %MwKernel.Error{code: :mqtt_publish_failed, detail: inspect(reason)}}
    end
  end

  @impl MwKernel.Adapter
  def health_check(%{client: client}) do
    case :emqtt.ping(client) do
      :pong            -> :ok
      {:error, reason} -> {:error, reason}
    end
  end

  @impl MwKernel.Adapter
  def disconnect(%{client: client}), do: :emqtt.disconnect(client)
end
```

**Use case:** After a transaction completes, MW-Core notifies subscribed POS terminals via MQTT without those devices needing to maintain REST long-polling.

---

## 9. What NOT to Adopt from N2O

These components look relevant on the surface but would create conflicts or unnecessary complexity in MW-Core:

### `synrc/nitro` — Server-side UI framework
**Conflict:** Nitro renders HTML server-side using Erlang records and templates (Nitrogen framework model). MW-Core's admin dashboard is built with **Phoenix LiveView** — a far more productive, idiomatic Elixir approach. Adopting Nitro would require maintaining two different server-side rendering systems.

### `synrc/rest` — Cowboy REST framework
**Conflict:** MW-Core already has a production-grade REST gateway in `gateway_api` built on **Plug + Bandit + Phoenix Router**. The Cowboy-based `synrc/rest` would duplicate the REST layer with a different API style and no Phoenix integration.

### `n2o_pi` — Named process instance manager
**Not needed:** MW-Core uses standard `GenServer` + `Horde.DynamicSupervisor` for distributed singletons. `n2o_pi` is an alternative process registry with ETS backing — it overlaps with what `Horde` already provides and would add a second process registry to reason about.

### `synrc/kvs` — Abstract key-value storage
**Not needed:** MW-Core has `infra_cache` (ETS + optional Redis) and `infra_repo` (MySQL via Ecto). KVS abstracts over Mnesia, RocksDB, and others — none of which MW-Core needs. Adding another storage abstraction would introduce unnecessary indirection without filling a real gap.

### N2O as a full WebSocket replacement for `gateway_ws`
**High risk, low value:** Phoenix Channels are the industry standard for Elixir WebSocket development. They are deeply integrated with Phoenix PubSub, LiveView, and the Plug pipeline. Replacing `gateway_ws` with N2O's WebSocket loop would:
- Break Phoenix PubSub integration
- Lose LiveDashboard metrics
- Require rewriting all client JavaScript to use BERT framing instead of Phoenix's JSON framing
- Provide zero functional improvement for browser clients

---

## 10. MQTT Options Comparison: N2O vs Direct EMQX Client vs HiveMQ

| Approach | Library | Broker required | MW-Core fit | Effort | Recommendation |
|---|---|---|---|---|---|
| **N2O MQTT bridge + EMQX** | `{:n2o, "~> 13.4"}` + `{:emqtt, "~> 1.6"}` | EMQX (external) | ✅ Best — N2O protocol loop, emqtt for connection, EMQX for broker | ~2 weeks | ✅ **Recommended for `gateway_mqtt`** |
| **Pure `emqtt` client** | `{:emqtt, "~> 1.6"}` | EMQX (external) | ✅ Good — simpler, no N2O dependency | ~1 week | ✅ **Recommended for `adapter_mqtt`** |
| **`tortoise311` MQTT client** | `{:tortoise311, "~> 0.12"}` | Any MQTT broker | ⚠️ Pure Elixir but less battle-tested for banking scale | ~1 week | ⚠️ Consider if `emqtt` has issues |
| **VerneMQ (Elixir-native broker)** | Built-in | VerneMQ self-hosted | ⚠️ Full broker in Elixir, but heavyweight; less widely deployed than EMQX | ~3 weeks | ⚠️ Over-engineered for current scope |
| **HiveMQ Cloud** | HiveMQ MQTT client | HiveMQ Cloud (SaaS) | ⚠️ Vendor lock-in; adds SaaS dependency | ~1 week | ❌ Avoid for financial system |
| **AWS IoT Core** | AWS SDK | AWS IoT Core | ❌ High vendor lock-in; expensive at scale | ~1 week setup, ongoing cost | ❌ Avoid |

---

## 11. Recommended Architecture

After this evaluation, the recommended N2O integration into MW-Core is:

```
┌─────────────────────────────────────────────────────────────────────────────┐
│                           EXTERNAL CLIENTS                                  │
│                                                                             │
│  REST  WebSocket  Browser  Mobile  MQTT Devices / POS Terminals             │
│                                        │                                   │
└────────────────────────────────────────┼───────────────────────────────────┘
                                         │ TCP :1883 / TLS :8883
                                         ▼
                              ┌──────────────────┐
                              │   EMQX Broker    │  ← external Docker container
                              │ (5.x, clustered) │    handles QoS, sessions,
                              └────────┬─────────┘    topic ACL, TLS
                                       │ MQTT bridge (emqtt client)
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         NORTH PLANE  —  Gateways                            │
│                                                                             │
│  gateway_api  gateway_ws  gateway_web  gateway_mobile  gateway_mqtt (NEW)  │
│                                                          │                  │
│                                              N2O proto loop + emqtt        │
│                                              Transformer: topic → Message   │
└─────────────────────────────────────────────────────────┬───────────────────┘
                                                          │ MwKernel.Message
┌─────────────────────────────────────────────────────────▼───────────────────┐
│                          CORE PLANE  —  unchanged                           │
│                                                                             │
│         mw_auth → mw_router (+ n2o_ring) → mw_transform → mw_audit        │
│                                                                             │
└─────────────────────────────────────────────────────────┬───────────────────┘
                                                          │
┌─────────────────────────────────────────────────────────▼───────────────────┐
│                         SOUTH PLANE  —  Adapters                            │
│                                                                             │
│  adapter_banking  adapter_dw  adapter_http  adapter_file  adapter_mqtt(NEW)│
└─────────────────────────────────────────────────────────────────────────────┘
```

**N2O components used:**
- `n2o` kernel → `gateway_mqtt` protocol loop (handler chain for MQTT message routing)
- `n2o_ring` → augments `mw_router` for tenant-to-node affinity in multi-node deployments

**N2O components NOT used:** `n2o_pi`, `synrc/nitro`, `synrc/rest`, `synrc/kvs`

---

## 12. Implementation Plan — `gateway_mqtt` with N2O

### Week 1 — Foundation

- [ ] `mix new gateway_mqtt --sup` in `apps/`
- [ ] Add `{:n2o, "~> 13.4"}` and `{:emqtt, "~> 1.6"}` to deps
- [ ] Implement `GatewayMqtt.Bridge` GenServer (connect + reconnect loop)
- [ ] Implement `GatewayMqtt.Transformer` (topic parsing + JSON ↔ `MwKernel.Message`)
- [ ] Add EMQX to `docker-compose.yml` for local development
- [ ] Unit tests: transformer, topic parsing, error cases

### Week 2 — Pipeline integration + telemetry

- [ ] Implement `GatewayMqtt.Publisher` (response and error publishing)
- [ ] Implement `GatewayMqtt.SubscriptionManager` (default topic subscriptions)
- [ ] Wire `mw_auth` plug for device authentication (JWT in MQTT `password` field or `CONNECT` username/password)
- [ ] Add `:telemetry` events: `[:gateway_mqtt, :message]`, `[:gateway_mqtt, :connect]`, `[:gateway_mqtt, :disconnect]`
- [ ] Integration test: end-to-end MQTT → MW-Core pipeline → response

### Week 3 — `adapter_mqtt` (south side) + `n2o_ring`

- [ ] `mix new adapter_mqtt --sup` in `apps/`
- [ ] Implement `AdapterMqtt` (south adapter: publish responses to MQTT topics)
- [ ] Register adapter in Flow Builder + static config
- [ ] Implement `MwRouter.Ring` using `n2o_ring` (optional, non-blocking)
- [ ] Wire `MwRouter.ClusterObserver` for automatic ring membership
- [ ] K8s deployment config: EMQX StatefulSet + MW-Core deployment with env vars

### Week 4 — Hardening + load test

- [ ] TLS configuration for EMQX (`8883` port, cert rotation via K8s Secret)
- [ ] Device authentication via MQTT `CONNECT` username/password → `mw_auth` API key check
- [ ] QoS 2 end-to-end test (exactly-once for payment requests)
- [ ] k6 load test: 1,000 concurrent MQTT clients, 60s sustained
- [ ] `GET /health/ready` includes EMQX connectivity check

### Definition of Done

| Criterion | How verified |
|---|---|
| MQTT device can publish to `mw/{tenant}/{type}` and receive response | Integration test |
| Auth rejects devices without valid credentials | Unit test + integration test |
| Circuit breaker opens if EMQX is unreachable | Chaos test (kill EMQX container) |
| Reconnect succeeds after EMQX restart (< 10 s) | Integration test |
| `GET /health/ready` returns 503 when EMQX down | Integration test |
| Telemetry events appear in Prometheus | Manual verification |
| P99 < 300 ms at 1,000 concurrent MQTT clients | k6 load test |
| DLQ populated on dispatch failure | Integration test |

---

## 13. Risk Register

| Risk | Likelihood | Impact | Mitigation |
|---|---|---|---|
| **N2O API changes between minor versions** | Medium | Low | Pin to exact version; N2O 13.x is stable with 6-month compatibility window per docs |
| **`emqtt` connection leak under high load** | Low | High | Use connection pool pattern; monitor with `:observer` during load test |
| **EMQX broker becomes a single point of failure** | Medium | High | Deploy EMQX as a 3-node cluster in K8s; MW-Core reconnects automatically |
| **MQTT topic ACL misconfiguration allows cross-tenant access** | Low | Critical | EMQX ACL rules enforce `mw/{tenant_id}/#` per device credential; unit test boundary |
| **N2O Erlang API not idiomatic in Elixir codebase** | High | Low | Wrapper modules (`GatewayMqtt.Bridge`) isolate all N2O/emqtt calls; rest of codebase stays pure Elixir |
| **MQTT QoS 2 exactly-once conflicts with MW-Core idempotency layer** | Low | Medium | MQTT QoS 2 + `Idempotency-Key` in payload provide double protection; document interaction |
| **BPE (Mnesia) conflicts with existing MySQL-only infra** | Medium | Medium | BPE integration is Phase 3, non-blocking; evaluate Mnesia isolation before committing |

---

## 14. Verdict

### Use N2O for:

| What | Why | When |
|---|---|---|
| **`gateway_mqtt`** — MQTT north-side gateway | N2O's protocol loop + `emqtt` client is the cleanest BEAM-native MQTT implementation; closes the most significant protocol gap; directly serves POS/IoT/embedded device use cases | **Phase 1 addition — Week 1–4** |
| **`adapter_mqtt`** — MQTT south-side adapter | Allows MW-Core to publish notifications to MQTT subscribers after transaction processing; ~1 week effort | **Phase 1 addition — Week 3** |
| **`n2o_ring`** — consistent hash ring in `mw_router` | Adds tenant/topic affinity across BEAM nodes; useful for Kafka + MQTT multi-node deployments; low effort, zero risk | **Phase 1 addition — Week 3** |
| **`synrc/bpmn` (BPE)** — BPMN 2.0 Flow Builder extension | Adds ISO 19510 compliance to the Flow Builder; allows import from Camunda/Bizagi; extends MW-Core toward a true BPM platform | **Phase 3 — evaluate after Phase 1–2 complete** |

### Do not use N2O for:

| What | Why |
|---|---|
| **WebSocket replacement for `gateway_ws`** | Phoenix Channels are superior for browser clients; no functional improvement; high migration cost |
| **REST gateway replacement** | `gateway_api` on Plug/Bandit is production-proven; Cowboy `synrc/rest` is not Phoenix-integrated |
| **Admin UI replacement** | Phoenix LiveView is the right choice for `gateway_web`; Nitro would regress the developer experience |
| **Process registry (`n2o_pi`)** | `Horde.DynamicSupervisor` already solves distributed singletons; no gap to fill |
| **Storage abstraction (`synrc/kvs`)** | ETS + Ecto/MySQL is sufficient and well-understood by the team |

### Summary score

| N2O component | Verdict | Effort | Value |
|---|---|---|---|
| `n2o` kernel + MQTT | **Adopt** | 4 weeks | ⭐⭐⭐⭐⭐ Closes critical MQTT gap |
| `n2o_ring` | **Adopt (selective)** | 1 week | ⭐⭐⭐ Augments multi-node routing |
| `synrc/bpmn` BPE | **Evaluate Phase 3** | 3–4 weeks | ⭐⭐⭐ Extends Flow Builder |
| `synrc/nitro` | **Reject** | — | ❌ Conflicts with LiveView |
| `synrc/rest` | **Reject** | — | ❌ Conflicts with Plug/Phoenix |
| `n2o_pi` | **Skip** | — | ⭐ No gap to fill |
| `synrc/kvs` | **Skip** | — | ⭐ No gap to fill |

> **Bottom line:** N2O is not a framework to adopt wholesale — its Nitro/REST/UI layer conflicts directly with Phoenix. But its **MQTT protocol handling and `n2o_ring` distribution primitive are genuine complements** to MW-Core that fill real gaps with minimal complexity. The `emqtt` library (EMQX's official client) should be used alongside N2O for the broker connection. With a 4-week implementation for `gateway_mqtt` and 1 week for `n2o_ring`, MW-Core gains full MQTT support while remaining 100% Phoenix/OTP for all other capabilities.

---

*Document owner: Platform Engineering · Evaluated: 2026-05-03 · N2O version evaluated: 13.4.15 (OTP 27/28) · References: [github.com/synrc/n2o](https://github.com/synrc/n2o)*