# Protocol & Connectivity Gap — Implementation Plan
## MW-Core · MercuryPay TMS

> **Purpose:** This document is a developer-facing implementation plan to close the protocol and connectivity gaps identified in [`docs/middleware-comparison.md`](middleware-comparison.md) — Section 3 (Protocol & Connectivity Support).
>
> Every protocol is implemented using the existing `MwKernel.Adapter` behaviour contract and umbrella app pattern described in [`docs/adapter_development_guide.md`](adapter_development_guide.md). No changes to the kernel or router are required.
>
> **Revised 2026-05-03 (v3):** Phase 1 approach updated based on [`docs/n2o-evaluation.md`](n2o-evaluation.md) and clarification of the existing TMS architecture. The TMS already runs **VerneMQ + Tortoise MQTT client**. MW-Core Phase 1 reuses VerneMQ as the broker and connects with `emqtt` (the Erlang/Elixir MQTT client library) — a direct parallel to the Tortoise pattern. **N2O is not used in `gateway_mqtt`** at all. The only N2O component retained is `n2o_ring`, added as a small dependency to `mw_router` for multi-node tenant affinity. Kafka/AMQP remain Phase 2. Phases 3–5 are on-demand only.

---

## Table of Contents

1. [Gap Analysis](#1-gap-analysis)
2. [Priority Framework](#2-priority-framework)
3. [Implementation Roadmap](#3-implementation-roadmap)
4. [Phase 1 — MQTT Gateway (VerneMQ + emqtt)](#4-phase-1--mqtt-gateway-vernemq--emqtt) ← **Start here**
5. [Phase 2 — Messaging Backbone (Kafka + AMQP)](#5-phase-2--messaging-backbone-kafka--amqp)
6. [Phase 3 — Modern API Protocols (GraphQL + gRPC)](#6-phase-3--modern-api-protocols-graphql--grpc) ⚠️ On-demand
7. [Phase 4 — Enterprise Messaging (JMS + EDI)](#7-phase-4--enterprise-messaging-jms--edi) ⚠️ On-demand
8. [Phase 5 — Financial Standards (FIX + SWIFT)](#8-phase-5--financial-standards-fix--swift) 🔒 Hold
9. [Cross-Cutting Concerns](#9-cross-cutting-concerns)
10. [Testing Strategy](#10-testing-strategy)
11. [Definition of Done (Per Protocol)](#11-definition-of-done-per-protocol)
12. [Full Timeline Summary](#12-full-timeline-summary)

---

## 1. Gap Analysis

From the comparison document, the following protocols are missing from MW-Core:

| Protocol | Gap Type | Status | Business Driver |
|---|---|---|---|
| **MQTT** | North gateway + South adapter | ❌ Not implemented | POS terminals, ATMs, IoT payment devices, soft-POS apps all speak MQTT natively |
| **Kafka** | South adapter + North gateway | ❌ (Broadway supports as add-on) | Event-driven architectures, high-throughput transaction streaming, audit log streaming |
| **AMQP / RabbitMQ** | South adapter | ❌ Not implemented | Enterprise partners using RabbitMQ; async decoupling for payment workflows |
| **GraphQL** | North gateway extension | ❌ Not implemented | Modern fintech clients prefer GraphQL; flexible field selection reduces over-fetching |
| **gRPC** | South adapter + North gateway | ❌ Not implemented | Internal microservice communication; high-performance binary RPC for backend systems |
| **JMS / ActiveMQ** | South adapter (via CloudI) | ❌ Not implemented | Legacy banking systems; IBM MQ-adjacent integration; durable messaging guarantees |
| **EDI (EDIFACT / X12)** | South adapter | ❌ Not implemented | B2B partner integrations; interchange file formats for settlements and reconciliation |
| **SWIFT** | South adapter (via CloudI) | ❌ Not implemented | International wire transfer messaging; correspondent banking; inter-bank settlements |
| **FIX Protocol** | South adapter (via CloudI) | ❌ Not implemented | Capital markets / securities trading; real-time order routing |

### What Already Exists (Do Not Re-implement)

| Protocol | Existing App | Notes |
|---|---|---|
| REST / JSON | `gateway_api` | Versioned `/api/v1/`, `/api/v2/` |
| WebSocket | `gateway_ws` | Phoenix Channels, 100K+ concurrent |
| ISO 8583 | `adapter_banking` | Binary encoder/decoder, Finch pool |
| SFTP / FTP | `adapter_file` | SSHEx client, streaming |
| CSV / XML / Flat | `adapter_file` | NimbleCSV + SweetXML |
| SOAP / HTTP | `adapter_http` | Finch + XML handling |
| CloudI bus | `adapter_cloudi` | Language-agnostic, Java/Python/Go |
| Email | `adapter_aritic_mail` | Already implemented |
| Marketing automation | `adapter_aritic_ma` | Already implemented |

---

## 2. Priority Framework

Priorities are set by three factors: **business impact** (does a current or near-term project need this?), **implementation effort** (how complex is the Elixir work?), and **risk** (protocol maturity and library quality on BEAM).

```
         HIGH IMPACT
              │
  Phase 1 ───┼─── Phase 2
  MQTT        │    Kafka
  (VerneMQ)   │    AMQP
              │
LOW EFFORT ───┼──────────────── HIGH EFFORT
              │
  Phase 3 ───┼─── Phase 4 / 5
  GraphQL     │    JMS / EDI
  gRPC        │    FIX / SWIFT
   (on-demand)│    (on-demand / hold)
              │
         LOW IMPACT (until confirmed need)
```

| Phase | Protocols | Weeks | Status | Trigger |
|---|---|---|---|---|
| **Phase 1** | MQTT (`gateway_mqtt`, `adapter_mqtt`; `n2o_ring` in `mw_router`) | 1–4 | **Start now** | Foundational for POS/device connectivity; reuses existing VerneMQ broker |
| **Phase 2** | Kafka, AMQP/RabbitMQ | 5–8 | **Follows Phase 1** | Enterprise streaming + partner messaging |
| **Phase 3** | GraphQL, gRPC | On-demand | ⚠️ **Only on confirmed client/service need** | A client team requests GraphQL, or a backend exposes gRPC |
| **Phase 4** | JMS/ActiveMQ, EDI | On-demand | ⚠️ **Only on confirmed partner need** | A specific legacy partner requires JMS or EDI file exchange |
| **Phase 5** | FIX, SWIFT | Hold | 🔒 **Hold — do not implement speculatively** | MercuryPay enters capital markets or international wire business |

> **Why MQTT before Kafka?**
> MQTT is the native protocol of POS terminals, ATMs, and embedded payment devices — the *sources* of transactions in a TMS. Without it, those devices must poll via REST or use non-standard WebSocket framing. Kafka and AMQP are infrastructure protocols that connect systems; MQTT connects the devices that generate business. MQTT is also bounded (4 weeks, full implementation), while VerneMQ is already running in the TMS — there is no new broker to operate.

---

## 3. Implementation Roadmap

```
Week  1  2  3  4  5  6  7  8   On-demand          Hold
      ├──────────┤  ├──────────┤  ┌──────────┐  ┌──────────────┐
P1    │  MQTT    │  │  Kafka   │  │ GraphQL  │  │ FIX          │
      │  emqtt   │  │  AMQP    │  │  gRPC    │  │ SWIFT        │
      │  VerneMQ │  │          │  │ JMS/EDI  │  │              │
      └──────────┘  └──────────┘  └──────────┘  └──────────────┘
       Phase 1        Phase 2      Phase 3 & 4      Phase 5
       (commit)       (commit)     (on-demand)       (hold)
```

**Resulting umbrella apps:**

| App Name | Protocol | Type | Phase | Status |
|---|---|---|---|---|
| `gateway_mqtt` | MQTT 3.1.1 (VerneMQ + emqtt) | North gateway | 1 | **Implement now** |
| `adapter_mqtt` | MQTT south publish | South adapter | 1 | **Implement now** |
| `adapter_kafka` | Apache Kafka | South adapter + North gateway | 2 | Implement after Phase 1 |
| `adapter_amqp` | AMQP 0-9-1 / RabbitMQ | South adapter | 2 | Implement after Phase 1 |
| `gateway_graphql` | GraphQL (Absinthe) | North gateway | 3 | On-demand |
| `adapter_grpc` | gRPC / Protobuf | South adapter | 3 | On-demand |
| `gateway_grpc` | gRPC | North gateway | 3 | On-demand |
| `adapter_jms` | JMS / ActiveMQ (via CloudI) | South adapter | 4 | On-demand |
| `adapter_edi` | EDI EDIFACT / X12 | South adapter | 4 | On-demand |
| `adapter_fix` | FIX Protocol (via CloudI) | South adapter | 5 | Hold |
| `adapter_swift` | SWIFT MT/MX (via CloudI) | South adapter | 5 | Hold |

---

## 4. Phase 1 — MQTT Gateway (VerneMQ + emqtt)

### Overview

MQTT (ISO/IEC 20922) is the standard protocol for POS terminals, ATMs, IoT payment devices, and soft-POS mobile apps communicating over cellular or low-bandwidth networks.

**Existing TMS stack:**
```
TMS:      POS Device → VerneMQ (broker) → Tortoise client → TMS app code
```

**MW-Core Phase 1 follows the identical pattern, different client library:**
```
MW-Core:  POS Device → VerneMQ (same broker) → emqtt client → gateway_mqtt → mw_router
```

`emqtt` and Tortoise are both MQTT client libraries — `emqtt` is Erlang-native (no JVM, no wrapper), which fits the BEAM runtime better. VerneMQ is reused as-is; no new broker to operate or configure.

**N2O is not a dependency of `gateway_mqtt`.** The only N2O component used in Phase 1 is `n2o_ring` — a ~200-line consistent hash ring added to `mw_router` for tenant-to-node affinity. It has no relationship to the MQTT gateway code.

Phase 1 delivers three things:

1. **`gateway_mqtt`** — new north-plane gateway; `emqtt` connects to VerneMQ, incoming MQTT messages are translated to `MwKernel.Message` and dispatched through the existing pipeline
2. **`adapter_mqtt`** — south-plane adapter; MW-Core publishes notification payloads back to MQTT topics after transaction processing
3. **`n2o_ring` in `mw_router`** — consistent hash ring for tenant-to-node affinity; also benefits Phase 2 Kafka partition routing

### Architecture

```
POS Terminals / ATMs / IoT Devices / soft-POS apps
        │  TCP :1883 / TLS :8883
        ▼
┌───────────────────────────────────────────────┐
│  VerneMQ Broker  (existing, already running)  │
│  Handles: QoS 0/1/2, persistent sessions,     │
│  retained msgs, will msgs, topic ACL, TLS      │
└──────────────────────┬────────────────────────┘
                       │  emqtt client connection
                       │  (same role as Tortoise in TMS)
                       ▼
┌──────────────────────────────────────────────────────────┐
│  gateway_mqtt  (new umbrella app — MW-Core BEAM node)    │
│                                                          │
│  GatewayMqtt.Bridge  (pure emqtt — no N2O)              │
│    subscribes: mw/{tenant_id}/{message_type}             │
│         │                                                │
│  GatewayMqtt.Transformer                                 │
│    MQTT topic + payload → MwKernel.Message               │
│         │                                                │
│  MwRouter.Dispatcher.dispatch/1  ← unchanged pipeline   │
│    mw_auth · mw_router · mw_transform · mw_audit        │
│         │                                                │
│  GatewayMqtt.Publisher                                   │
│    publishes: mw/{tenant_id}/{type}/response             │
└──────────────────────────────────────────────────────────┘

mw_router also has:
  MwRouter.Ring  (n2o_ring — only N2O component in Phase 1)
    maps tenant_id hash → responsible BEAM node
    used by both gateway_mqtt and adapter_kafka (Phase 2)
```

> **Why reuse VerneMQ, not add EMQX?**
> VerneMQ is already operated by the TMS team — it handles QoS guarantees, persistent sessions, retained messages, and topic ACL out of the box. Adding a second MQTT broker (EMQX) would split operational knowledge and double the broker infrastructure to monitor. MW-Core simply connects to it as a client, the same way the TMS already does with Tortoise.

> **Why `emqtt` instead of Tortoise?**
> Tortoise is an Elixir MQTT client backed by a GenServer. `emqtt` is the official Erlang MQTT client from the EMQX team — it runs directly on the BEAM, has no external runtime dependency, handles reconnection natively, and integrates cleanly with OTP supervisors. The API surface is almost identical to what you already know from Tortoise.

---

### 4.1 `gateway_mqtt` — North Gateway

#### 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},
    {:emqtt,           "~> 1.6"},   # Erlang MQTT client — equivalent of Tortoise in TMS
    {:jason,           "~> 1.4"},
    {:telemetry,       "~> 1.2"}
  ]
end
```

> No `{:n2o, ...}` dependency here. `n2o_ring` lives only in `mw_router/mix.exs`.

#### File Structure

```
apps/gateway_mqtt/
├── lib/
│   ├── gateway_mqtt.ex
│   ├── gateway_mqtt/
│   │   ├── application.ex           # Supervisor: Bridge + SubscriptionManager
│   │   ├── bridge.ex                # GenServer: emqtt connection lifecycle
│   │   ├── subscription_manager.ex  # GenServer: registers default topic subscriptions
│   │   ├── transformer.ex           # MQTT topic/payload ↔ MwKernel.Message
│   │   ├── publisher.ex             # Publishes responses back to VerneMQ
│   │   └── telemetry.ex             # :telemetry events for Prometheus
├── test/
│   ├── gateway_mqtt_test.exs
│   └── support/mqtt_mock.ex
└── mix.exs
```

#### Core Implementation

**Bridge GenServer — emqtt connection to VerneMQ**

```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_ms 5_000

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

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

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

  @impl true
  def handle_info(:connect, state) do
    opts = [
      host:        Application.fetch_env!(:gateway_mqtt, :vernemq_host),
      port:        Application.get_env(:gateway_mqtt, :vernemq_port, 1883),
      clientid:    "mw-core-#{node()}",
      username:    Application.get_env(:gateway_mqtt, :vernemq_username, "mw-core"),
      password:    Application.get_env(:gateway_mqtt, :vernemq_password, ""),
      clean_start: false,   # persistent session — re-subscribe on reconnect is handled below
      keepalive:   60,
      reconnect:   true
    ]

    case :emqtt.start_link(opts) do
      {:ok, client} ->
        {:ok, _connack} = :emqtt.connect(client)
        Logger.info("[GatewayMqtt] Connected to VerneMQ at #{opts[:host]}:#{opts[:port]}")
        # Re-register any subscriptions accumulated before reconnect
        Enum.each(state.subscriptions, fn {t, q} -> :emqtt.subscribe(client, t, q) end)
        {:noreply, %{state | client: client}}

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

  # Hot path — incoming MQTT publish from VerneMQ
  @impl true
  def handle_info({:publish, %{topic: topic, payload: payload, qos: qos}}, state) do
    t0 = System.monotonic_time()

    case Transformer.from_mqtt(topic, payload) do
      {:ok, mw_msg} ->
        case Dispatcher.dispatch(mw_msg) do
          {:ok, response} ->
            Publisher.publish(state.client, "#{topic}/response",
                              Transformer.to_mqtt(response), qos)
          {:error, err} ->
            Publisher.publish_error(state.client, topic, err)
        end

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

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

    {:noreply, state}
  end

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

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

  def handle_call({:subscribe, topic, qos}, _from, state) do
    # Client not yet connected — queue subscription for when it connects
    {:reply, :queued, %{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"] || new_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 new_trace_id, do: :crypto.strong_rand_bytes(16) |> Base.encode16(case: :lower)
end
```

**Subscription Manager — default topic registrations**

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

  # QoS 1 for payment-critical topics, QoS 0 for fire-and-forget telemetry
  # VerneMQ wildcard + matches any single topic segment (one tenant level)
  @defaults [
    {"mw/+/transaction.payment",  1},
    {"mw/+/transaction.inquiry",  0},
    {"mw/+/device.telemetry",     0},
    {"mw/+/auth.token.refresh",   1}
  ]

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

  @impl true
  def init(_opts) do
    send(self(), :subscribe_defaults)
    {:ok, %{}}
  end

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

**Publisher — response and error publishing back to VerneMQ**

```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
    payload = Jason.encode!(%{
      error:     err.code,
      detail:    err.detail,
      timestamp: DateTime.utc_now() |> DateTime.to_iso8601()
    })
    :emqtt.publish(client, "#{topic}/error", 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,
  vernemq_host:     "localhost",
  vernemq_port:     1883,
  vernemq_username: "mw-core",
  vernemq_password: ""

# config/runtime.exs  — point to the existing VerneMQ instance
config :gateway_mqtt,
  vernemq_host:     System.get_env("VERNEMQ_HOST", "vernemq"),
  vernemq_port:     System.get_env("VERNEMQ_PORT", "1883") |> String.to_integer(),
  vernemq_username: System.get_env("VERNEMQ_USERNAME", "mw-core"),
  vernemq_password: System.get_env("VERNEMQ_PASSWORD", "")
```

**MQTT topic naming convention**

```
Request (device → VerneMQ → MW-Core):   mw/{tenant_id}/{message_type}
Response (MW-Core → VerneMQ → device):  mw/{tenant_id}/{message_type}/response
Error:                                   mw/{tenant_id}/{message_type}/error

QoS assignments:
  mw/+/transaction.payment    QoS 1  — at-least-once; MW-Core idempotency handles duplicates
  mw/+/transaction.inquiry    QoS 0  — lossy ok; device retries on timeout
  mw/+/device.telemetry       QoS 0  — fire-and-forget metrics
  mw/+/auth.token.refresh     QoS 1  — must not lose
```

---

### 4.2 `adapter_mqtt` — South Adapter

Used when MW-Core needs to **push notifications to MQTT subscribers after processing** — e.g., notifying POS terminals of payment confirmation, broadcasting config changes to devices. MW-Core publishes to VerneMQ; subscribed devices receive the message.

```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, :vernemq_host)
    port  = config[:port]  || Application.get_env(:adapter_mqtt, :vernemq_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}, %MwKernel.Message{} = msg) do
    payload = Transformer.to_mqtt(msg)

    case :emqtt.publish(client, topic, payload, _qos = 1) do
      :ok ->
        :telemetry.execute([:adapter_mqtt, :publish], %{count: 1}, %{topic: topic, status: :ok})
        {:ok, msg}

      {:error, reason} ->
        :telemetry.execute([:adapter_mqtt, :publish], %{count: 1}, %{topic: topic, status: :error})
        {: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
      err   -> {:error, err}
    end
  end

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

**Config**

```elixir
# config/config.exs
config :adapter_mqtt,
  vernemq_host:  "localhost",
  vernemq_port:  1883,
  default_topic: "mw-core.notifications"

# config/runtime.exs
config :adapter_mqtt,
  vernemq_host: System.get_env("VERNEMQ_HOST", "vernemq"),
  vernemq_port: System.get_env("VERNEMQ_PORT", "1883") |> String.to_integer()
```

**Route registration**

```elixir
config :mw_router, :default_routes, [
  %{message_type: "device.notification",  adapter_module: "Elixir.AdapterMqtt"},
  %{message_type: "payment.confirmation", adapter_module: "Elixir.AdapterMqtt"}
]
```

---

### 4.3 `n2o_ring` in `mw_router` — Consistent Hash Ring

> **Scope:** This is the **only N2O component in Phase 1.** It lives entirely in `mw_router`. `gateway_mqtt` and `adapter_mqtt` have zero N2O dependencies.

`n2o_ring` is a ~200-line Erlang consistent hash ring backed by `gb_trees`. It ensures the same tenant's messages always route to the same BEAM node in a multi-node cluster — preventing duplicate MQTT message processing when multiple nodes are subscribed to VerneMQ.

**Add to `mw_router/mix.exs`** (only change needed):

```elixir
# apps/mw_router/mix.exs
defp deps do
  [
    # ... existing deps ...
    {:n2o, "~> 13.4"}   # only for n2o_ring — no other N2O features used
  ]
end
```

**Ring module**

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

  def init do
    :n2o_ring.create(@ring_name)
    :n2o_ring.add(@ring_name, node(), @vnodes)
  end

  # Returns the BEAM node responsible for this tenant's messages
  def responsible_node(tenant_id),
    do: :n2o_ring.lookup(@ring_name, :erlang.phash2(tenant_id))

  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 existing `libcluster` node 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, :init_ring}}

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

  @impl true
  def handle_info({:nodeup, n},   s), do: MwRouter.Ring.add_node(n)    |> then(fn _ -> {:noreply, s} end)
  def handle_info({:nodedown, n}, s), do: MwRouter.Ring.remove_node(n) |> then(fn _ -> {:noreply, s} end)
end
```

**How it is used in `gateway_mqtt`** (no N2O import needed — pure Elixir call):

```elixir
# In GatewayMqtt.Bridge — after Transformer.from_mqtt/2 succeeds
node = MwRouter.Ring.responsible_node(mw_msg.context.tenant_id)

if node == node() do
  Dispatcher.dispatch(mw_msg)
else
  # Forward to the responsible node — avoids duplicate processing
  :rpc.call(node, MwRouter.Dispatcher, :dispatch, [mw_msg])
end
```

---

### 4.4 VerneMQ — Local Dev and Production

**Local development — VerneMQ in Docker:**

```yaml
# docker-compose.yml  (add if VerneMQ is not already in your local stack)
services:
  vernemq:
    image: vernemq/vernemq:1.13.0-alpine
    ports:
      - "1883:1883"     # MQTT
      - "8883:8883"     # MQTT/TLS
      - "8080:8080"     # MQTT-over-WebSocket
      - "8888:8888"     # VerneMQ HTTP API
    environment:
      DOCKER_VERNEMQ_ALLOW_ANONYMOUS: "on"         # dev only — disable in prod
      DOCKER_VERNEMQ_ACCEPT_EULA: "yes"
    healthcheck:
      test: ["CMD", "vernemq", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5
```

> **Production:** Point `VERNEMQ_HOST` env var at the existing VerneMQ cluster already operated by the TMS team. MW-Core connects as a client — no VerneMQ config changes are required except adding an ACL rule for the `mw-core` username.

**VerneMQ ACL entry for MW-Core client** (add to `vmq.acl` or use VerneMQ's HTTP API):

```
# Allow mw-core to subscribe to all mw/+ topics and publish responses
user mw-core
topic read mw/#
topic write mw/#
```

**Kubernetes — MW-Core deployment env vars** (production):

```yaml
env:
  - name: VERNEMQ_HOST
    valueFrom:
      secretKeyRef:
        name: vernemq-credentials
        key: host
  - name: VERNEMQ_PORT
    value: "1883"
  - name: VERNEMQ_USERNAME
    valueFrom:
      secretKeyRef:
        name: vernemq-credentials
        key: username
  - name: VERNEMQ_PASSWORD
    valueFrom:
      secretKeyRef:
        name: vernemq-credentials
        key: password
```

---

### 4.5 Phase 1 Week-by-Week Plan

| Week | Work | Deliverable |
|---|---|---|
| 1 | `gateway_mqtt` — Bridge + Transformer + unit tests; VerneMQ in Docker locally | MQTT → MW-Core pipeline working locally; round-trip test passes |
| 2 | Publisher + SubscriptionManager + `mw_auth` device credential check + telemetry | Auth, rate limiting, audit all wired; integration test with real VerneMQ passes |
| 3 | `adapter_mqtt` south adapter + `n2o_ring` in `mw_router` + K8s env vars | South MQTT publish live; ring active on multi-node; K8s points to existing VerneMQ |
| 4 | TLS config (`:8883`) + QoS 1 + idempotency interaction test + k6 load test | P99 < 300 ms at 1,000 concurrent MQTT clients; `GET /health/ready` checks VerneMQ |

---

## 5. Phase 2 — Messaging Backbone (Kafka + AMQP)

> **When to start:** After Phase 1 is complete and in production. `n2o_ring` built in Phase 1 benefits Kafka partition routing here.

### 5.1 `adapter_kafka` — Apache Kafka

#### Overview

Kafka integration serves two directions:
- **South (producer):** MW-Core publishes `MwKernel.Message` structs to Kafka topics after processing (audit streaming, event sourcing, downstream notification).
- **North (consumer gateway):** A new Broadway-based Kafka consumer reads from Kafka topics and feeds messages into the MW-Core pipeline — enabling event-driven transaction ingestion from external systems.

#### Dependencies

```elixir
# apps/adapter_kafka/mix.exs
defp deps do
  [
    {:mw_kernel,        in_umbrella: true},
    {:infra_telemetry,  in_umbrella: true},
    {:brod,             "~> 3.16"},       # Kafka Erlang client (battle-tested)
    {:broadway_kafka,   "~> 0.10"},       # Broadway producer wrapping brod
    {:jason,            "~> 1.4"},
    {:telemetry,        "~> 1.2"}
  ]
end
```

> **Why `brod` over `kafka_ex`?** `brod` is the Erlang-native Kafka client used by Klarna, Discord, and others in financial systems. `broadway_kafka` wraps it with back-pressure built-in.

#### File Structure

```
apps/adapter_kafka/
├── lib/
│   ├── adapter_kafka.ex
│   ├── adapter_kafka/
│   │   ├── application.ex
│   │   ├── producer.ex
│   │   ├── consumer_pipeline.ex
│   │   ├── transformer.ex
│   │   └── telemetry.ex
├── test/
│   ├── adapter_kafka_test.exs
│   └── support/kafka_mock.ex
└── mix.exs
```

#### Core Implementation

**South adapter (producer)**

```elixir
# apps/adapter_kafka/lib/adapter_kafka.ex
defmodule AdapterKafka do
  @behaviour MwKernel.Adapter

  alias AdapterKafka.{Producer, Transformer}

  @impl MwKernel.Adapter
  def connect(config) do
    brokers = config[:brokers] || Application.fetch_env!(:adapter_kafka, :brokers)
    topic   = config[:topic]   || Application.fetch_env!(:adapter_kafka, :default_topic)
    {:ok, %{brokers: brokers, topic: topic, client: :brod_client_mw}}
  end

  @impl MwKernel.Adapter
  def send(state, %MwKernel.Message{} = msg) do
    payload = Transformer.to_kafka(msg)

    case Producer.publish(state.client, state.topic, msg.id, payload) do
      :ok ->
        :telemetry.execute([:adapter_kafka, :publish], %{count: 1}, %{topic: state.topic, status: :ok})
        {:ok, MwKernel.Message.put_meta(msg, :kafka_offset, :produced)}

      {:error, reason} ->
        :telemetry.execute([:adapter_kafka, :publish], %{count: 1}, %{topic: state.topic, status: :error})
        {:error, %MwKernel.Error{code: :kafka_publish_failed, detail: inspect(reason)}}
    end
  end

  @impl MwKernel.Adapter
  def health_check(%{client: client}) do
    case :brod.get_metadata(client, :all) do
      {:ok, _}         -> :ok
      {:error, reason} -> {:error, reason}
    end
  end

  @impl MwKernel.Adapter
  def disconnect(_state), do: :ok
end
```

**Producer module**

```elixir
defmodule AdapterKafka.Producer do
  def publish(client, topic, key, value) when is_binary(value) do
    :brod.produce_sync(client, topic, _partition = :hash, key, value)
  end
end
```

**North-side consumer pipeline (Broadway)**

```elixir
defmodule AdapterKafka.ConsumerPipeline do
  use Broadway

  alias AdapterKafka.Transformer
  alias MwRouter.Dispatcher

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {BroadwayKafka.Producer, [
          hosts:            Application.fetch_env!(:adapter_kafka, :brokers),
          group_id:         Application.get_env(:adapter_kafka, :consumer_group, "mw-core"),
          topics:           Application.fetch_env!(:adapter_kafka, :consumer_topics),
          offset_commit_on: :acks
        ]},
        concurrency: 1
      ],
      processors: [default: [concurrency: Application.get_env(:adapter_kafka, :processor_concurrency, 10)]],
      batchers:   [default: [batch_size: 50, batch_timeout: 500]]
    )
  end

  @impl true
  def handle_message(_processor, message, _context) do
    case Transformer.from_kafka(message.data) do
      {:ok, mw_message} ->
        case Dispatcher.dispatch(mw_message) do
          {:ok, _}      -> message
          {:error, _}   -> Broadway.Message.failed(message, :dispatch_error)
        end
      {:error, reason} ->
        Broadway.Message.failed(message, reason)
    end
  end

  @impl true
  def handle_failed(messages, _context) do
    Enum.each(messages, fn msg ->
      InfraQueue.DLQ.enqueue(%{source: :kafka, payload: msg.data, reason: msg.status})
    end)
    messages
  end
end
```

**Transformer**

```elixir
defmodule AdapterKafka.Transformer do
  alias MwKernel.Message

  def to_kafka(%Message{} = msg) do
    Jason.encode!(%{
      id:         msg.id,
      type:       msg.type,
      tenant_id:  msg.context.tenant_id,
      trace_id:   msg.context.trace_id,
      payload:    msg.payload,
      emitted_at: DateTime.utc_now() |> DateTime.to_iso8601()
    })
  end

  def from_kafka(raw_binary) do
    with {:ok, map} <- Jason.decode(raw_binary),
         type       <- String.to_existing_atom(map["type"] || "kafka.event"),
         msg        <- Message.new(type, map["payload"] || %{}, :kafka) do
      {:ok, msg}
    else
      _ -> {:error, :invalid_kafka_payload}
    end
  end
end
```

**Application supervisor**

```elixir
defmodule AdapterKafka.Application do
  use Application

  @impl true
  def start(_type, _args) do
    brokers = Application.fetch_env!(:adapter_kafka, :brokers)

    children = [
      %{id: :brod_client_mw,
        start: {:brod_client, :start_link, [brokers, :brod_client_mw, []]}},
      maybe_consumer_pipeline()
    ]
    |> Enum.reject(&is_nil/1)

    Supervisor.start_link(children, strategy: :one_for_one, name: AdapterKafka.Supervisor)
  end

  defp maybe_consumer_pipeline do
    case Application.get_env(:adapter_kafka, :consumer_topics) do
      topics when is_list(topics) and length(topics) > 0 -> AdapterKafka.ConsumerPipeline
      _ -> nil
    end
  end
end
```

**Config**

```elixir
# config/config.exs
config :adapter_kafka,
  brokers:               [{"localhost", 9092}],
  default_topic:         "mw-core.transactions",
  consumer_topics:       ["mw-core.inbound"],
  consumer_group:        "mw-core",
  processor_concurrency: 10

# config/runtime.exs
config :adapter_kafka,
  brokers: System.get_env("KAFKA_BROKERS", "localhost:9092")
           |> String.split(",")
           |> Enum.map(fn b ->
             [h, p] = String.split(b, ":")
             {h, String.to_integer(p)}
           end)
```

**Route registration**

```elixir
config :mw_router, :default_routes, [
  %{message_type: "transaction.stream", adapter_module: "Elixir.AdapterKafka"},
  %{message_type: "audit.stream",       adapter_module: "Elixir.AdapterKafka"}
]
```

---

### 5.2 `adapter_amqp` — AMQP 0-9-1 / RabbitMQ

#### Overview

AMQP adapter enables MW-Core to publish processed messages to a RabbitMQ exchange (south side) and consume from queues (north side via a GenServer). Used for partner integrations where RabbitMQ is the enterprise messaging bus.

#### Dependencies

```elixir
defp deps do
  [
    {:mw_kernel,        in_umbrella: true},
    {:infra_telemetry,  in_umbrella: true},
    {:amqp,             "~> 3.3"},
    {:jason,            "~> 1.4"},
    {:telemetry,        "~> 1.2"}
  ]
end
```

#### File Structure

```
apps/adapter_amqp/
├── lib/
│   ├── adapter_amqp.ex
│   ├── adapter_amqp/
│   │   ├── application.ex
│   │   ├── channel_pool.ex
│   │   ├── publisher.ex
│   │   ├── consumer.ex
│   │   └── transformer.ex
├── test/
│   └── adapter_amqp_test.exs
└── mix.exs
```

#### Core Implementation

```elixir
defmodule AdapterAmqp do
  @behaviour MwKernel.Adapter

  alias AdapterAmqp.{Publisher, Transformer}

  @impl MwKernel.Adapter
  def connect(config) do
    url      = config[:url]      || Application.fetch_env!(:adapter_amqp, :url)
    exchange = config[:exchange] || Application.fetch_env!(:adapter_amqp, :default_exchange)
    {:ok, %{url: url, exchange: exchange}}
  end

  @impl MwKernel.Adapter
  def send(state, %MwKernel.Message{} = msg) do
    routing_key = msg.type |> to_string() |> String.replace(".", "_")
    payload     = Transformer.to_amqp(msg)

    case Publisher.publish(state.exchange, routing_key, payload) do
      :ok ->
        :telemetry.execute([:adapter_amqp, :publish], %{count: 1},
          %{exchange: state.exchange, status: :ok})
        {:ok, MwKernel.Message.put_meta(msg, :amqp_routing_key, routing_key)}

      {:error, reason} ->
        {:error, %MwKernel.Error{code: :amqp_publish_failed, detail: inspect(reason)}}
    end
  end

  @impl MwKernel.Adapter
  def health_check(%{url: url}) do
    case AMQP.Connection.open(url) do
      {:ok, conn} -> AMQP.Connection.close(conn); :ok
      {:error, r} -> {:error, r}
    end
  end

  @impl MwKernel.Adapter
  def disconnect(_state), do: :ok
end
```

**Channel pool**

```elixir
defmodule AdapterAmqp.ChannelPool do
  use GenServer

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

  @impl true
  def init(_opts) do
    {:ok, conn} = AMQP.Connection.open(Application.fetch_env!(:adapter_amqp, :url))
    {:ok, chan} = AMQP.Channel.open(conn)
    {:ok, %{conn: conn, chan: chan}}
  end

  @impl true
  def handle_call(:get_channel, _from, state), do: {:reply, state.chan, state}

  @impl true
  def terminate(_reason, %{conn: conn}), do: AMQP.Connection.close(conn)
end
```

**Consumer (north-side queue reader)**

```elixir
defmodule AdapterAmqp.Consumer do
  use GenServer

  alias AdapterAmqp.Transformer
  alias MwRouter.Dispatcher

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

  @impl true
  def init(_opts) do
    queue = Application.get_env(:adapter_amqp, :consumer_queue)
    if queue do
      chan = AdapterAmqp.ChannelPool.get_channel()
      AMQP.Queue.declare(chan, queue, durable: true)
      AMQP.Basic.consume(chan, queue)
      {:ok, %{chan: chan, queue: queue}}
    else
      :ignore
    end
  end

  @impl true
  def handle_info({:basic_deliver, payload, meta}, state) do
    case Transformer.from_amqp(payload) do
      {:ok, mw_msg} ->
        case Dispatcher.dispatch(mw_msg) do
          {:ok, _}    -> AMQP.Basic.ack(state.chan, meta.delivery_tag)
          {:error, _} -> AMQP.Basic.nack(state.chan, meta.delivery_tag, requeue: false)
        end
      {:error, _} ->
        AMQP.Basic.nack(state.chan, meta.delivery_tag, requeue: false)
    end
    {:noreply, state}
  end

  def handle_info({:basic_consume_ok, _}, state), do: {:noreply, state}
  def handle_info({:basic_cancel, _},    state),  do: {:stop, :cancelled, state}
end
```

**Config**

```elixir
config :adapter_amqp,
  url:              "amqp://guest:guest@localhost",
  default_exchange: "mw-core.transactions",
  consumer_queue:   nil

# config/runtime.exs
config :adapter_amqp,
  url: System.get_env("AMQP_URL", "amqp://guest:guest@localhost")
```

---

## 6. Phase 3 — Modern API Protocols (GraphQL + gRPC)

> ⚠️ **On-demand only.**
>
> **Trigger for GraphQL:** A client engineering team explicitly requests GraphQL — they cannot achieve their use case with REST alone.
> **Trigger for gRPC:** A specific backend system MW-Core needs to call exposes only a gRPC endpoint.

### 6.1 `gateway_graphql` — GraphQL Gateway

#### Dependencies

```elixir
defp deps do
  [
    {:mw_kernel,     in_umbrella: true},
    {:mw_auth,       in_umbrella: true},
    {:mw_router,     in_umbrella: true},
    {:mw_audit,      in_umbrella: true},
    {:phoenix,       "~> 1.7"},
    {:absinthe,      "~> 1.7"},
    {:absinthe_plug, "~> 1.5"},
    {:jason,         "~> 1.4"},
    {:plug_cowboy,   "~> 2.6"}
  ]
end
```

#### File Structure

```
apps/gateway_graphql/
├── lib/gateway_graphql/
│   ├── application.ex
│   ├── endpoint.ex
│   ├── schema.ex
│   ├── resolvers/
│   │   ├── transaction.ex
│   │   └── account.ex
│   └── transformer.ex
├── test/gateway_graphql_test.exs
└── mix.exs
```

#### Core Implementation

```elixir
defmodule GatewayGraphql.Schema do
  use Absinthe.Schema
  alias GatewayGraphql.Resolvers

  object :transaction_result do
    field :id,           :string
    field :status,       :string
    field :amount,       :integer
    field :currency,     :string
    field :trace_id,     :string
    field :processed_at, :string
  end

  query do
    field :account_balance, :transaction_result do
      arg :account_id, non_null(:string)
      resolve &Resolvers.Account.balance/3
    end
  end

  mutation do
    field :process_payment, :transaction_result do
      arg :amount,          non_null(:integer)
      arg :currency,        non_null(:string)
      arg :merchant_id,     non_null(:string)
      arg :idempotency_key, :string
      resolve &Resolvers.Transaction.process_payment/3
    end
  end
end
```

```elixir
defmodule GatewayGraphql.Resolvers.Transaction do
  alias GatewayGraphql.Transformer
  alias MwRouter.Dispatcher

  def process_payment(_parent, args, %{context: ctx}) do
    msg = Transformer.from_graphql_mutation(:process_payment, args, ctx)
    case Dispatcher.dispatch(msg) do
      {:ok, response}                   -> {:ok, Transformer.to_graphql(response)}
      {:error, %MwKernel.Error{} = err} -> {:error, err.detail}
    end
  end
end
```

```elixir
defmodule GatewayGraphql.Transformer do
  alias MwKernel.{Message, Context}

  def from_graphql_mutation(type, args, ctx) do
    Message.new(:"transaction.#{type}", args, :gateway_graphql, %Context{
      trace_id:  :crypto.strong_rand_bytes(16) |> Base.encode16(case: :lower),
      tenant_id: ctx[:tenant_id],
      user:      ctx[:user],
      roles:     ctx[:roles]
    })
  end

  def to_graphql(%Message{payload: p, id: id, context: ctx}) do
    %{id: id, status: p[:status] || "ok", amount: p[:amount],
      currency: p[:currency], trace_id: ctx.trace_id,
      processed_at: DateTime.utc_now() |> DateTime.to_iso8601()}
  end
end
```

```elixir
defmodule GatewayGraphql.Endpoint do
  use Plug.Router

  plug Plug.Parsers,
    parsers: [:urlencoded, :multipart, :json, Absinthe.Plug.Parser],
    pass: ["*/*"], json_decoder: Jason

  plug MwAuth.Plug
  plug MwRouter.RateLimiterPlug

  plug Absinthe.Plug,
    schema: GatewayGraphql.Schema,
    context: &__MODULE__.build_context/1

  def build_context(conn),
    do: %{tenant_id: conn.assigns[:tenant_id], user: conn.assigns[:user], roles: conn.assigns[:roles]}
end
```

**Config**

```elixir
config :gateway_graphql, GatewayGraphql.Endpoint,
  http: [port: 4020],
  server: true
```

> **Port allocation:** REST `:4000` · WebSocket `:4010` · Mobile `:4015` · Admin `:4001` · MQTT `:1883` (VerneMQ) · GraphQL `:4020`

---

### 6.2 `adapter_grpc` — gRPC / Protocol Buffers

#### Dependencies

```elixir
defp deps do
  [
    {:mw_kernel,     in_umbrella: true},
    {:infra_telemetry, in_umbrella: true},
    {:grpc,          "~> 0.7"},
    {:protobuf,      "~> 0.12"},
    {:google_protos, "~> 0.3"},
    {:jason,         "~> 1.4"},
    {:telemetry,     "~> 1.2"}
  ]
end
```

#### Proto definition

```protobuf
syntax = "proto3";
package mw;

service TransactionService {
  rpc ProcessTransaction (TransactionRequest) returns (TransactionResponse);
}

message TransactionRequest {
  string request_id   = 1;
  string type         = 2;
  string tenant_id    = 3;
  bytes  payload_json = 4;
}

message TransactionResponse {
  string request_id  = 1;
  string status      = 2;
  bytes  result_json = 3;
  string trace_id    = 4;
}
```

```bash
mix protobuf.generate --include-docs true \
  priv/proto/transaction_service.proto \
  --output-path lib/adapter_grpc/proto/
```

#### Core Implementation

```elixir
defmodule AdapterGrpc do
  @behaviour MwKernel.Adapter

  alias AdapterGrpc.{ChannelManager, Transformer}
  alias Mw.TransactionService.Stub

  @impl MwKernel.Adapter
  def connect(config) do
    host = config[:host] || Application.fetch_env!(:adapter_grpc, :host)
    port = config[:port] || Application.fetch_env!(:adapter_grpc, :port)
    case ChannelManager.get_or_create({host, port}) do
      {:ok, channel} -> {:ok, %{channel: channel}}
      {:error, r}    -> {:error, r}
    end
  end

  @impl MwKernel.Adapter
  def send(%{channel: channel}, %MwKernel.Message{} = msg) do
    case Stub.process_transaction(channel, Transformer.to_grpc(msg), timeout: 5_000) do
      {:ok, response} ->
        :telemetry.execute([:adapter_grpc, :request], %{count: 1}, %{status: :ok})
        {:ok, Transformer.from_grpc(response, msg)}
      {:error, %GRPC.RPCError{} = err} ->
        :telemetry.execute([:adapter_grpc, :request], %{count: 1}, %{status: :error})
        {:error, %MwKernel.Error{code: :grpc_error, detail: err.message}}
    end
  end

  @impl MwKernel.Adapter
  def health_check(%{channel: channel}) do
    case GRPC.Channel.get(channel) do
      {:ok, _} -> :ok
      _        -> {:error, :channel_unavailable}
    end
  end

  @impl MwKernel.Adapter
  def disconnect(%{channel: channel}), do: GRPC.Stub.disconnect(channel)
end
```

```elixir
defmodule AdapterGrpc.ChannelManager do
  use GenServer

  def start_link(_), do: GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
  def get_or_create(key), do: GenServer.call(__MODULE__, {:get_or_create, key})

  @impl true
  def init(state), do: {:ok, state}

  @impl true
  def handle_call({:get_or_create, {host, port} = key}, _from, channels) do
    case Map.get(channels, key) do
      nil ->
        case GRPC.Stub.connect("#{host}:#{port}") do
          {:ok, ch} -> {:reply, {:ok, ch}, Map.put(channels, key, ch)}
          err       -> {:reply, err, channels}
        end
      ch -> {:reply, {:ok, ch}, channels}
    end
  end
end
```

**Config**

```elixir
config :adapter_grpc,
  host: System.get_env("GRPC_HOST", "localhost"),
  port: System.get_env("GRPC_PORT", "50051") |> String.to_integer()
```

---

## 7. Phase 4 — Enterprise Messaging (JMS + EDI)

> ⚠️ **On-demand only.**
>
> **Trigger for JMS:** A named integration partner or legacy banking system requires JMS/ActiveMQ that cannot be served via REST, MQTT, or Kafka.
> **Trigger for EDI:** A B2B partner sends settlement or remittance files in EDIFACT or X12 format.

### 7.1 `adapter_jms` — JMS / ActiveMQ via CloudI

```elixir
defmodule AdapterJms do
  @behaviour MwKernel.Adapter
  alias AdapterJms.Transformer

  @cloudi_service "/mw/jms/send"

  @impl MwKernel.Adapter
  def connect(config) do
    {:ok, %{queue: config[:queue] || Application.fetch_env!(:adapter_jms, :default_queue)}}
  end

  @impl MwKernel.Adapter
  def send(state, %MwKernel.Message{} = msg) do
    cloudi_msg = MwKernel.Message.new(:"jms.send", Transformer.to_jms(msg, state.queue), :adapter_jms)
    case AdapterClouDi.send(%{service_name: @cloudi_service}, cloudi_msg) do
      {:ok, response} -> {:ok, Transformer.from_jms(response, msg)}
      {:error, _} = e -> e
    end
  end

  @impl MwKernel.Adapter
  def health_check(_state) do
    case AdapterClouDi.send(%{service_name: "/mw/jms/health"},
                             MwKernel.Message.new(:"jms.health", %{}, :adapter_jms)) do
      {:ok, _}         -> :ok
      {:error, reason} -> {:error, reason}
    end
  end

  @impl MwKernel.Adapter
  def disconnect(_state), do: :ok
end
```

**Java CloudI service skeleton**

```java
public class JmsService implements Runnable {
    private final API api;
    private final ConnectionFactory factory;

    public JmsService(int threadIndex) throws API.InvalidInputException {
        this.api = new API(threadIndex);
        this.factory = new ActiveMQConnectionFactory(System.getenv("ACTIVEMQ_URL"));
    }

    public void run() {
        try {
            api.subscribe("mw/jms/send",   this::sendMessage);
            api.subscribe("mw/jms/health", this::healthCheck);
            api.poll(-1);
        } catch (Exception e) { e.printStackTrace(); }
    }
    // sendMessage: parse JSON → JMS TextMessage → send → return {"status":"sent"}
}
```

---

### 7.2 `adapter_edi` — EDI EDIFACT / X12

```elixir
defmodule AdapterEdi do
  @behaviour MwKernel.Adapter
  alias AdapterEdi.{EdifactParser, X12Parser, Transformer}

  @impl MwKernel.Adapter
  def connect(config), do: {:ok, %{format: config[:format] || :edifact}}

  @impl MwKernel.Adapter
  def send(%{format: fmt}, %MwKernel.Message{payload: %{edi_data: raw}} = msg) do
    parser = if fmt == :x12, do: X12Parser, else: EdifactParser
    case parser.parse(raw) do
      {:ok, parsed} ->
        {:ok, MwKernel.Message.new(:edi_parsed, Transformer.to_canonical(parsed, fmt),
                                    :adapter_edi, msg.context)}
      {:error, reason} ->
        {:error, %MwKernel.Error{code: :edi_parse_failed, detail: inspect(reason)}}
    end
  end

  @impl MwKernel.Adapter
  def health_check(_state), do: :ok

  @impl MwKernel.Adapter
  def disconnect(_state), do: :ok
end
```

```elixir
defmodule AdapterEdi.EdifactParser do
  import NimbleParsec

  segment_tag  = ascii_string([?A..?Z], min: 3, max: 6)
  data_element = repeat(utf8_char(not: [?+, ?', ?:])) |> reduce({Enum, :join, [""]})
  segment      = segment_tag
                 |> ignore(ascii_char([?+]))
                 |> repeat(data_element |> ignore(optional(ascii_char([?+]))))
                 |> ignore(ascii_char([?']))
                 |> tag(:segment)

  defparsec :parse_message, repeat(segment)

  def parse(edi_string) do
    case parse_message(edi_string) do
      {:ok, segs, "", _, _, _} ->
        {:ok, Enum.reduce(segs, %{}, fn {:segment, [tag | elems]}, acc ->
          Map.update(acc, tag, [elems], &[elems | &1])
        end)}
      {:error, r, rest, _, _, _} -> {:error, {r, rest}}
    end
  end
end
```

```elixir
defmodule AdapterEdi.X12Parser do
  def parse(edi_string) do
    segments =
      edi_string
      |> String.split("~")
      |> Enum.map(&String.trim/1)
      |> Enum.reject(&(&1 == ""))
      |> Enum.map(&String.split(&1, "*"))
      |> Enum.map(fn [tag | elems] -> {tag, elems} end)
    {:ok, Map.new(segments)}
  end
end
```

---

## 8. Phase 5 — Financial Standards (FIX + SWIFT)

> 🔒 **Hold — do not implement without a confirmed capital markets or international wire roadmap.**
>
> **Trigger for FIX:** MercuryPay enters equities/FX/derivatives order routing.
> **Trigger for SWIFT:** MercuryPay provides correspondent banking, international wire, or inter-bank settlement services.

### 8.1 `adapter_fix` — FIX Protocol via CloudI

```elixir
defmodule AdapterFix do
  @behaviour MwKernel.Adapter
  alias AdapterFix.Transformer

  @impl MwKernel.Adapter
  def connect(config),
    do: {:ok, %{session_id: config[:session_id] || Application.fetch_env!(:adapter_fix, :session_id)}}

  @impl MwKernel.Adapter
  def send(state, %MwKernel.Message{} = msg) do
    cloudi_msg = MwKernel.Message.new(:"fix.new_order", Transformer.to_fix(msg, state.session_id), :adapter_fix)
    case AdapterClouDi.send(%{service_name: "/mw/fix/send"}, cloudi_msg) do
      {:ok, response} -> {:ok, Transformer.from_fix(response, msg)}
      {:error, _} = e -> e
    end
  end

  @impl MwKernel.Adapter
  def health_check(_state) do
    case AdapterClouDi.send(%{service_name: "/mw/fix/health"},
                             MwKernel.Message.new(:"fix.health", %{}, :adapter_fix)) do
      {:ok, _}         -> :ok
      {:error, reason} -> {:error, reason}
    end
  end

  @impl MwKernel.Adapter
  def disconnect(_state), do: :ok
end
```

---

### 8.2 `adapter_swift` — SWIFT MT/MX via CloudI

```elixir
defmodule AdapterSwift do
  @behaviour MwKernel.Adapter
  alias AdapterSwift.Transformer

  @impl MwKernel.Adapter
  def connect(config),
    do: {:ok, %{message_type: config[:message_type] || Application.get_env(:adapter_swift, :default_message_type, :mt103)}}

  @impl MwKernel.Adapter
  def send(%{message_type: mt}, %MwKernel.Message{} = msg) do
    cloudi_msg = MwKernel.Message.new(:"swift.#{mt}", Transformer.to_swift(msg, mt), :adapter_swift)
    case AdapterClouDi.send(%{service_name: "/mw/swift/send/#{mt}"}, cloudi_msg) do
      {:ok, response} -> {:ok, Transformer.from_swift(response, msg)}
      {:error, _} = e -> e
    end
  end

  @impl MwKernel.Adapter
  def health_check(_state) do
    case AdapterClouDi.send(%{service_name: "/mw/swift/health"},
                             MwKernel.Message.new(:"swift.health", %{}, :adapter_swift)) do
      {:ok, _}         -> :ok
      {:error, reason} -> {:error, reason}
    end
  end

  @impl MwKernel.Adapter
  def disconnect(_state), do: :ok
end
```

---

## 9. Cross-Cutting Concerns

These apply to **every protocol**. Implement from day one — do not defer.

### 9.1 Telemetry (mandatory per adapter)

```elixir
:telemetry.execute(
  [:adapter_<name>, :request],
  %{count: 1, duration: System.monotonic_time() - start_time},
  %{protocol: :<name>, status: :ok | :error, tenant_id: msg.context.tenant_id}
)
```

Attach in `infra_telemetry`:

```elixir
:telemetry.attach_many("adapter-mqtt-metrics",
  [[:gateway_mqtt, :message], [:adapter_mqtt, :publish]],
  &InfraTelemetry.Prometheus.handle_event/4, nil)
```

### 9.2 Circuit Breaker (automatic — no code needed)

`:fuse` is applied per adapter automatically by `MwRouter.Dispatcher`. Reset in tests:

```elixir
setup do: :fuse.reset(:"AdapterMqtt.fuse")
```

### 9.3 `n2o_ring` — Multi-Node Dispatch (from Phase 1)

`n2o_ring` lives in `mw_router` only. After Phase 1 ships, Phase 2 Kafka adapters use it for partition affinity:

```elixir
# In AdapterKafka.ConsumerPipeline — route to responsible node
node = MwRouter.Ring.responsible_node(mw_message.context.tenant_id)
if node == node(),
  do:   Dispatcher.dispatch(mw_message),
  else: :rpc.call(node, MwRouter.Dispatcher, :dispatch, [mw_message])
```

### 9.4 Dead Letter Queue

All consumer-side adapters (MQTT, Kafka, AMQP, JMS) must enqueue on failure:

```elixir
InfraQueue.DLQ.enqueue(%{source: :mqtt, payload: raw, reason: reason})
```

### 9.5 Health Check Propagation

Every adapter's `health_check/1` is discovered automatically by `MwRouter.HealthCheck`. No registration needed.

### 9.6 Multi-Tenant Routing

```elixir
MwRouter.RouteTable.upsert_rule(%{
  tenant_id:      "tenant_acme",
  message_type:   "transaction.payment",
  adapter_module: "Elixir.AdapterMqtt",
  config:         %{topic: "mw/tenant_acme/payment.confirmation"}
})
```

---

## 10. Testing Strategy

### Unit tests (per adapter)

```elixir
defmodule AdapterMqttTest do
  use ExUnit.Case, async: true

  setup do: :fuse.reset(:"AdapterMqtt.fuse")

  describe "connect/1" do
    test "returns state with client and topic" do
      # stub :emqtt.start_link + :emqtt.connect via Mox
    end
  end

  describe "send/2" do
    test "publishes and returns {:ok, msg}" do ... end
    test "returns {:error, %MwKernel.Error{}} on publish failure" do ... end
  end

  describe "health_check/1" do
    test ":ok when :emqtt.ping returns :pong" do ... end
    test "{:error, _} when unreachable" do ... end
  end
end
```

### Integration tests (Docker Compose)

| Adapter | Docker image | Test |
|---|---|---|
| `gateway_mqtt` / `adapter_mqtt` | `vernemq/vernemq:1.13.0-alpine` | Publish via MQTT client → MW-Core → response on reply topic |
| `adapter_kafka` | `confluentinc/cp-kafka:7.5` | Produce → consume → pipeline round-trip |
| `adapter_amqp` | `rabbitmq:3.12-management` | Publish → queue → consume → dispatch |
| `gateway_graphql` | None (in-process) | HTTP POST GraphQL mutation → canonical routing |
| `adapter_grpc` | Custom test server | Protobuf round-trip |
| `adapter_edi` | None (file fixture) | Parse `test/fixtures/*.edi` |

**`docker-compose.test.yml`:**

```yaml
services:
  vernemq:
    image: vernemq/vernemq:1.13.0-alpine
    ports: ["1883:1883"]
    environment:
      DOCKER_VERNEMQ_ALLOW_ANONYMOUS: "on"
      DOCKER_VERNEMQ_ACCEPT_EULA: "yes"

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    environment:
      KAFKA_KRAFT_MODE: "true"
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_NODE_ID: 1
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      CLUSTER_ID: "MkU3OEVBNTcwNTJENDM2Qk"
    ports: ["9092:9092"]

  rabbitmq:
    image: rabbitmq:3.12-management-alpine
    ports: ["5672:5672", "15672:15672"]
```

```bash
docker-compose -f docker-compose.test.yml up -d
mix test --only integration
docker-compose -f docker-compose.test.yml down
```

### Load tests (k6)

```bash
# Phase 1: 1,000 concurrent MQTT device connections
k6 run tests/load/mqtt_devices.js --vus 1000 --duration 60s

# Phase 2: Kafka producer throughput
k6 run tests/load/kafka_producer.js --vus 200 --duration 60s
```

**Threshold:** P99 < 300 ms (same SLA as existing REST endpoints).

---

## 11. Definition of Done (Per Protocol)

| Criterion | Verified by |
|---|---|
| All four `MwKernel.Adapter` callbacks implemented | `mix compile --warnings-as-errors` |
| Unit tests for all callbacks pass | `mix test apps/adapter_<name>/` |
| Integration test against real infrastructure passes | `mix test --only integration` |
| Route registered via Flow Builder and published | Manual: matching `message_type` routes correctly |
| Health check returns `:ok` when broker reachable | `GET /health/ready` returns 200 |
| Circuit breaker opens on 5 consecutive failures | Unit test with fuse mock |
| Telemetry events emitted correctly | Prometheus scrape verification |
| DLQ populated on consumer failure (if applicable) | Integration test + admin UI |
| P99 < 300 ms at target VUs | k6 load test |
| `@moduledoc` / `@doc` complete | Code review |
| Transformer handles nil/missing fields gracefully | Unit test edge cases |

---

## 12. Full Timeline Summary

### Committed (start immediately)

| Week | Work | Deliverable |
|---|---|---|
| 1 | `gateway_mqtt` — Bridge + Transformer + unit tests; VerneMQ in Docker | MQTT → pipeline working locally; round-trip test passes |
| 2 | Publisher + SubscriptionManager + `mw_auth` device auth + telemetry | Auth, audit, rate-limiting wired; integration test passes |
| 3 | `adapter_mqtt` south adapter + `n2o_ring` in `mw_router` + K8s env vars for VerneMQ | South MQTT publish live; ring active on multi-node |
| 4 | TLS (`:8883`) + QoS 1 + idempotency interaction test + k6 (1,000 MQTT clients) | P99 < 300 ms; `GET /health/ready` includes VerneMQ check |
| 5–6 | `adapter_kafka` producer + Broadway consumer + unit tests | Kafka south adapter + north consumer live |
| 6–7 | Kafka integration tests + k6 + ring-based partition routing | Kafka P99 verified; tenant affinity across nodes |
| 7–8 | `adapter_amqp` publisher + consumer GenServer + integration tests | RabbitMQ adapter live; Flow Builder route published |

**Total committed: 8 weeks · 3 new umbrella apps · 1 `mw_router` enhancement**

### On-demand (start only with confirmed project)

| Phase | Work | Weeks (from trigger) | Trigger |
|---|---|---|---|
| 3 | `gateway_graphql` (Absinthe) | 3 weeks | Client team explicitly requests GraphQL |
| 3 | `adapter_grpc` (Protobuf) | 2 weeks | Backend system exposes gRPC endpoint |
| 4 | `adapter_jms` + Java CloudI service | 3 weeks | Named legacy partner requires JMS |
| 4 | `adapter_edi` (EDIFACT + X12) | 2 weeks | Named partner sends EDI settlement files |

### Hold (do not implement without roadmap confirmation)

| Phase | Work | Weeks | Hold condition |
|---|---|---|---|
| 5 | `adapter_fix` + QuickFIX/J CloudI service | 4 weeks | Capital markets / order routing roadmap confirmed |
| 5 | `adapter_swift` + SWIFT MT/MX CloudI service | 4 weeks | International wire / correspondent banking confirmed |

---

*Document owner: Platform Engineering · Last updated: 2026-05-03 · Related: [`middleware-comparison.md`](middleware-comparison.md) · [`n2o-evaluation.md`](n2o-evaluation.md) · [`adapter_development_guide.md`](adapter_development_guide.md)*