# Adapter Development Guide

**MW-Core · MercuryPay TMS**

This document explains the complete lifecycle of a new adapter — from writing the code
to making it available in the Flow Builder UI and live routing config — from both the
**developer** and **operator/admin** perspective.

---

## What Is an Adapter?

An adapter is the south-side integration module that talks to one external system
(a core banking API, a fraud service, a data warehouse, a third-party REST endpoint, etc.).
The pipeline calls adapters polymorphically: it only knows the `MwKernel.Adapter` behaviour
contract; the concrete implementation lives entirely inside the adapter app.

```
Gateway (REST / WebSocket / Mobile)
        ↓
  MwRouter.Pipeline
        ↓  resolves route from RouteTable
  MwRouter.Dispatcher  ←→  MwRouter.DagExecutor  ←→  MwRouter.FanoutDispatcher
        ↓                                              (composite routes)
  YourAdapter.connect/1
  YourAdapter.send/2
        ↓
  External System
```

---

## Part 1 — Developer Perspective

### Step 1 · Create the umbrella app

```bash
cd apps/
mix new adapter_fraud --sup
```

Add it to the umbrella `mix.exs` — nothing to edit; Mix discovers all apps in `apps/`
automatically.

---

### Step 2 · Add umbrella dependencies to your `mix.exs`

```elixir
# apps/adapter_fraud/mix.exs
defp deps do
  [
    {:mw_kernel,   in_umbrella: true},   # required — gives you the Adapter behaviour + Message struct
    {:mw_audit,    in_umbrella: true},   # optional — if you want structured audit logging
    {:infra_cache, in_umbrella: true},   # optional — if you need Redis-backed caching
    {:infra_repo,  in_umbrella: true},   # optional — if you need DB access
    {:finch,       "~> 0.13"},           # HTTP client (or Tesla, Req, etc.)
    {:jason,       "~> 1.2"},
    {:telemetry,   "~> 1.0"}
  ]
end
```

---

### Step 3 · Implement the `MwKernel.Adapter` behaviour

The behaviour requires **four callbacks**:

| Callback | Purpose |
|---|---|
| `connect/1` | Open a connection / pool. Called per-dispatch with a config map. |
| `send/2` | Send the canonical `Message`, return `{:ok, Message}` or `{:error, reason}` |
| `health_check/1` | Called by health endpoints and circuit-breaker checks |
| `disconnect/1` | Graceful teardown |

```elixir
# apps/adapter_fraud/lib/adapter_fraud.ex
defmodule AdapterFraud do
  @moduledoc "Fraud-scoring adapter — integrates with the external FraudGuard API."

  @behaviour MwKernel.Adapter

  alias AdapterFraud.{Client, Transformer}

  @impl MwKernel.Adapter
  def connect(_config) do
    base_url = Application.get_env(:adapter_fraud, :base_url, "https://fraud.internal")
    api_key  = Application.get_env(:adapter_fraud, :api_key,  "")
    {:ok, %{base_url: base_url, api_key: api_key}}
  end

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

    case Client.post(state, "/v1/score", payload) do
      {:ok, response} ->
        :telemetry.execute([:adapter_fraud, :request], %{count: 1}, %{status: :ok})
        {:ok, Transformer.from_fraud(response)}

      {:error, reason} ->
        :telemetry.execute([:adapter_fraud, :request], %{count: 1}, %{status: :error})
        {:error, reason}
    end
  end

  @impl MwKernel.Adapter
  def health_check(state) do
    case Client.get(state, "/health") do
      {:ok, %{status: 200}} -> :ok
      {:ok, %{status: s}}   -> {:error, {:unhealthy, s}}
      {:error, r}           -> {:error, r}
    end
  end

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

> **Why `connect/1` is called per-dispatch**: MW-Core is designed for pooled,
> stateless connections. Pass a Finch pool name in `state` for real HTTP connection
> pooling; don't hold open sockets in the struct.

---

### Step 4 · Write a Transformer

Keep the adapter thin — put all field mapping in a `Transformer` module:

```elixir
# apps/adapter_fraud/lib/adapter_fraud/transformer.ex
defmodule AdapterFraud.Transformer do
  alias MwKernel.Message

  def to_fraud(%Message{payload: p, id: id}) do
    %{
      "request_id"  => id,
      "amount"      => p[:amount],
      "currency"    => p[:currency],
      "merchant_id" => p[:merchant_id],
      "ip_address"  => p[:ip_address]
    }
  end

  def from_fraud(%{body: %{"score" => score, "outcome" => outcome}}) do
    Message.new(:fraud_score, %{score: score, outcome: outcome}, :adapter_fraud)
  end

  def from_fraud(raw) do
    Message.new(:fraud_score, %{raw: raw}, :adapter_fraud)
  end
end
```

---

### Step 5 · Start a Finch pool in your Application supervisor

```elixir
# apps/adapter_fraud/lib/adapter_fraud/application.ex
defmodule AdapterFraud.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      {Finch, name: AdapterFraud.Finch,
              pools: %{
                Application.get_env(:adapter_fraud, :base_url, "https://fraud.internal") =>
                  [size: 10, count: 1]
              }}
    ]
    Supervisor.start_link(children, strategy: :one_for_one, name: AdapterFraud.Supervisor)
  end
end
```

---

### Step 6 · Add config entries

```elixir
# config/config.exs  (or config/dev.exs / runtime.exs for secrets)
config :adapter_fraud,
  base_url: "https://fraud.internal",
  api_key:  System.get_env("FRAUD_API_KEY", "")
```

---

### Step 7 · Verify the behaviour is satisfied

```bash
mix compile
# Should produce no warnings about unimplemented callbacks
```

The pipeline's `DagExecutor` resolves adapters at runtime via:

```elixir
defp resolve_adapter(name) when is_binary(name) do
  mod = Module.concat([name])
  if Code.ensure_loaded?(mod) and function_exported?(mod, :connect, 1),
    do: mod,
    else: nil
end
```

This means **the module name you type in the Flow Builder must exactly match the
Elixir module name** — e.g. `AdapterFraud` (no `Elixir.` prefix needed in the UI).

---

### Step 8 · Write tests

```elixir
# apps/adapter_fraud/test/adapter_fraud_test.exs
defmodule AdapterFraudTest do
  use ExUnit.Case, async: true

  test "connect/1 returns state with base_url" do
    assert {:ok, %{base_url: _}} = AdapterFraud.connect(%{})
  end

  test "send/2 maps message correctly" do
    msg = MwKernel.Message.new(:"transaction.payment",
      %{amount: 100, currency: "USD", merchant_id: "M1", ip_address: "1.2.3.4"},
      :test)
    # Use Bypass or Mox to stub HTTP
    # assert {:ok, %MwKernel.Message{payload: %{outcome: "pass"}}} = AdapterFraud.send(state, msg)
  end
end
```

---

## Part 2 · How the Router Discovers the Adapter

There are **three ways** a route reaches the adapter — each with different lifetime:

```
┌───────────────────────────────────────────────────────┐
│                   RouteTable (ETS)                    │
│                                                       │
│  key: "transaction.fraud_check"                       │
│  value: %{type: :single, adapter: AdapterFraud, ...}  │
│                                                       │
│  key: "payment.dag"                                   │
│  value: %{type: :dag, dag_route: %DagRoute{...}, ...} │
└───────────────────────────────────────────────────────┘
         ↑               ↑                ↑
   config.exs      Flow Builder      API / IEx
   (boot time)     (runtime)         (runtime)
```

### Route registration method A — Static config (boot time)

Survives server restarts; used for stable, always-on routes:

```elixir
# config/config.exs
config :mw_router, :default_routes, [
  # existing:
  %{message_type: "transaction.payment",   adapter_module: "Elixir.AdapterBanking"},
  %{message_type: "account.balance",       adapter_module: "Elixir.AdapterBanking"},
  # new:
  %{message_type: "transaction.fraud_check", adapter_module: "Elixir.AdapterFraud"}
]
```

Loaded by `RouteTable.init/1` at GenServer startup → lives in ETS.
**Restart required** to pick up changes.

---

### Route registration method B — Flow Builder UI (runtime, recommended)

Zero-downtime. No restart needed. Works for both simple (single-adapter) and DAG flows.

**Single adapter route:**

1. Open `/admin/flows/new`
2. Drop a **Request** node → **Adapter** node → **Response** node
3. In the Adapter property panel set **Adapter module** = `AdapterFraud`
4. Set **Message type** field in the toolbar = `transaction.fraud_check`
5. Click **Save** (persists to DB) then **▶ Publish** (writes to ETS RouteTable)

**DAG flow with Decision:**

1. Use a Phase 9 template (e.g. "Fraud Gate + Payment") as a starting point
2. Customise adapter node names (`AdapterFraud`, `PaymentAdapter`, etc.)
3. Set message_type = `payment.dag_fraud_gate`
4. **Publish** → calls `RouteTable.upsert_dag_rule/1` → DAG goes live

After Publish, the route is hot and all four gateways (:4000, :4010, :4015) will route
matching requests through the new flow immediately.

---

### Route registration method C — Programmatic / IEx (runtime)

For scripts, migrations, or testing:

```elixir
# Single adapter
MwRouter.RouteTable.upsert_rule(%{
  message_type:   "transaction.fraud_check",
  adapter_module: "Elixir.AdapterFraud",
  priority:       1,
  timeout_ms:     3_000,
  active:         true
})

# DAG flow
MwRouter.RouteTable.upsert_dag_rule(%{
  message_type: "payment.fraud_gate",
  dag_route:    %MwKernel.DagRoute{
    nodes: [...],
    edges: [...],
    merge_strategy: :first_wins
  }
})
```

---

## Part 3 — Operator / Admin Perspective (Flow Builder)

### Using your new adapter in a flow

Once the developer has deployed the adapter app (it compiles into the same BEAM node),
the module is immediately available. You do **not** need to register it anywhere before
using it in a flow — the Flow Builder resolves it by module name at Publish time.

#### Quick-start: single adapter route

```
Node Palette → drag Adapter onto canvas
Property panel → Adapter module: AdapterFraud
                 Priority: 1
                 Timeout (ms): 3000
                 Fallback on: timeout, error

Toolbar → Message type: transaction.fraud_check
         → Save → ▶ Publish
```

#### Quick-start: DAG flow with fraud gate

1. Templates → "Fraud Gate + Payment"
2. Change the `FraudCheckAdapter` node → `AdapterFraud`
3. Change `PaymentAdapter` → your payment adapter
4. Set message_type → `payment.checkout`
5. Save → Publish

---

### What happens at Publish

```
Browser (LiveView)
  → handle_event("publish_flow")
  → builds %MwKernel.DagRoute{nodes, edges, merge_strategy}
  → RouteTable.upsert_dag_rule(%{message_type: ..., dag_route: ...})
  → :ets.insert(:mw_route_table, {message_type_key, rule_map})

All gateways (same BEAM node, same ETS) see the change instantly.
```

> **Persistence**: Publish writes to ETS only. The flow is also saved to the DB
> (via Save). On server restart, the `RouteTableSync` subscriber replays published
> flows from the DB back into ETS so routes survive restarts.

---

### Checking if an adapter is available before using it

In IEx (or a dev health endpoint):

```elixir
mod = Module.concat(["AdapterFraud"])
Code.ensure_loaded?(mod)                        # true if compiled + loaded
function_exported?(mod, :connect, 1)            # true if behaviour satisfied
AdapterFraud.health_check(elem(AdapterFraud.connect(%{}), 1))  # :ok or {:error, reason}
```

---

## Part 4 — Checklist Summary

### Developer checklist

- [ ] `mix new adapter_<name> --sup` inside `apps/`
- [ ] Add `{:mw_kernel, in_umbrella: true}` to deps
- [ ] Implement all four `MwKernel.Adapter` callbacks
- [ ] Write a `Transformer` module for field mapping
- [ ] Start a Finch pool in `Application.start/2`
- [ ] Add config entries (`base_url`, secrets via `System.get_env`)
- [ ] `mix compile` — zero warnings about unimplemented callbacks
- [ ] Write unit tests (use Bypass or Mox for HTTP)

### Admin / operator checklist (Flow Builder)

- [ ] Verify adapter is deployed (same BEAM node as router)
- [ ] Open `/admin/flows/new`
- [ ] Drop nodes, set **Adapter module** = exact Elixir module name (e.g. `AdapterFraud`)
- [ ] Set **Message type** in toolbar
- [ ] **Save** → confirms DB persistence
- [ ] **▶ Publish** → hot-loads into ETS RouteTable
- [ ] Test by sending a request with the matching `message_type` to `:4000/api/v1/transactions`

---

## Part 5 — Key Architecture Points

| Topic | Detail |
|---|---|
| **Discovery** | `DagExecutor` and `Dispatcher` resolve adapters by name at runtime via `Code.ensure_loaded?` + `function_exported?`. No registration step needed. |
| **Circuit breaker** | Every adapter gets an automatic fuse named `:<AdapterModule>.fuse` (managed by `:fuse`). Blows after 5 errors in 10s; resets after 30s. No code needed in your adapter. |
| **Telemetry** | Emit `[:adapter_<name>, :request]` events with `%{status: :ok/:error}` metadata for automatic Prometheus scraping. |
| **Multi-tenancy** | A flow published without a `tenant_id` applies globally. To scope to a tenant, publish via `RouteTable.init_tenant_table/1` + `upsert_rule/2` with the tenant-scoped table. |
| **Hot reload** | ETS is shared across all gateways on the same node. Publish takes effect instantly for all ports (:4000, :4010, :4015). |
| **Fallback (composite/DAG)** | Set `fallback_on: ["timeout", "error"]` on an adapter node in the Flow Builder. The DAG executor will mark that slot as `:all_failed` and the `ConditionalMerger` will exclude it from the response rather than crashing. |
| **Restart persistence** | Flows saved in DB are replayed into ETS at startup via `RouteTableSync`. Static `default_routes` in `config.exs` are also loaded. ETS-only inserts (not via Flow Builder) do NOT survive restarts. |

---

## Part 6 — Dispatch Strategy: Internal vs HTTP

### The deployment topology problem

The standard adapter path (Parts 1–4) works perfectly when adapter modules live in
the same BEAM node.  In production you often run services across **multiple servers**:

```
Single-deploy (umbrella, one release)     Distributed deploy (separate services)
───────────────────────────────────────   ─────────────────────────────────────────
┌─────────────────────────────────────┐   ┌──────────────┐   ┌───────────────────┐
│  BEAM node (one OS process)         │   │  Router node │   │  Risk node        │
│  ┌──────────┐  ┌──────────────────┐ │   │  mw_router   │──▶│  mw_risk          │
│  │ mw_router│──│ MwRisk.Scoring   │ │   │  :4000       │   │  :4001            │
│  │ mw_risk  │  │  Pipeline.score  │ │   └──────────────┘   └───────────────────┘
│  └──────────┘  └──────────────────┘ │       HTTP POST /api/v1/risk/score/1
└─────────────────────────────────────┘
     Direct BEAM function call
```

Rather than writing two different flows for two deployment topologies, the
**Dispatch Strategy** lets you keep the canvas identical and change only `config.exs`.

---

### How it works

```
Flow Builder canvas node: "MwRiskAdapter"
           │
           ▼
 MwRouter.DagExecutor.resolve_adapter("MwRiskAdapter")
           │
           ├─ Checks MwRouter.DispatchConfig first
           │     Found?     → {:dispatch, spec}   (config-driven path)
           │     Not found? → {:module, mod}       (standard module path)
           │
     {:dispatch, spec}
           │
           ▼
 MwRouter.DispatchRouter.call(spec, ctx, timeout_ms)
           │
     spec.dispatch = :internal  ──▶  MwRouter.Dispatch.InternalDispatch
                                        apply(module, function, [ctx])
                                        returns {:ok, normalised_map}
           │
     spec.dispatch = {:http, url} ─▶  MwRouter.Dispatch.HttpDispatch
                                        Finch POST → JSON response
                                        returns {:ok, response_map}
```

The canvas flow (business intent) **never changes**.
Only `config.exs` (infrastructure topology) changes between environments.

---

### Configuration

In `config/config.exs` or `config/runtime.exs` (preferred for prod URLs/secrets):

```elixir
config :mw_router, :dispatch_config, %{

  # ── Mode A: Internal BEAM call (single-deploy / umbrella) ───────────────
  # No network hop. Target module is resolved dynamically — no compile-time dep.
  "MwRiskAdapter" => %{
    dispatch: :internal,
    module:   MwRisk.ScoringPipeline,   # atom; loaded via Code.ensure_loaded? at runtime
    function: :score                     # must accept (%MwKernel.Context{}) → {:ok, map}
  },

  # ── Mode B: Remote HTTP POST (distributed deploy) ───────────────────────
  # Posts ctx.request.payload as JSON; expects a JSON object response.
  "MwRiskAdapter" => %{
    dispatch:   {:http, System.get_env("RISK_SERVICE_URL", "http://localhost:4001/api/v1/risk/score/1")},
    timeout_ms: 3_000,
    headers:    [{"x-internal-token", System.get_env("INTERNAL_TOKEN", "")}]
  }
}
```

> **Only one entry per adapter name** — pick Mode A or Mode B, not both.

---

### Result normalisation (internal mode)

`InternalDispatch` automatically normalises the result before it reaches the DAG:

| Raw (MwRisk.ScoringPipeline) | Normalised (Decision node sees) |
|---|---|
| `%{decision: :approve, score: 0.82}` | `%{"decision" => "approve", "score" => 0.82, "outcome" => "approve"}` |
| `%{decision: :decline, ...}` | `%{"decision" => "decline", ..., "outcome" => "decline"}` |

`"outcome"` is added as an alias for `"decision"` so Decision nodes with
`field_path: "outcome"` work without extra configuration.

**Correct Decision node config for a fraud gate:**

```
Data source: slot_result
Slot name:   fraud_check
Field path:  outcome          ← maps to ScoringPipeline's decision field
Operator:    eq
Value:       approve
```

---

### Target function contract (internal mode)

The function must accept a single `%MwKernel.Context{}` and return
`{:ok, map()} | {:error, term()}`:

```elixir
# Example — MwRisk.ScoringPipeline already satisfies this contract
@spec score(MwKernel.Context.t()) :: {:ok, scoring_result()} | {:error, term()}
def score(%MwKernel.Context{} = ctx) do
  # ... existing implementation
end
```

If the module is not loaded (e.g. `mw_risk` excluded from the release), the slot
returns `:all_failed` and the pipeline degrades gracefully — no crash.

---

### Dispatch mode badge in the Flow Builder

After selecting an adapter in the property panel, a **read-only** badge appears:

```
Adapter module   MwRiskAdapter
⚡ internal dispatch  ⓘ        ← green: internal BEAM call
```

or for HTTP mode:

```
🌐 HTTP · risk-service:4001  ⓘ   ← blue: remote HTTP call
```

The badge is informational only.  To change the mode, edit `config.exs` and redeploy.
The canvas and saved flow are **unchanged**.

---

### Switching between modes

| Scenario | Action | Canvas changes? |
|---|---|---|
| Single → Distributed deploy | Change `:internal` → `{:http, url}` in `config.exs`, redeploy | **None** |
| Distributed → Single deploy | Change `{:http, url}` → `:internal` in `config.exs`, redeploy | **None** |
| Remove dispatch override | Delete adapter entry from `dispatch_config` | **None** — resolves via module discovery |

---

### Part 6 checklist

**Internal dispatch (single-deploy):**
- [ ] Target module compiled into the same release (e.g. `{:mw_risk, in_umbrella: true}`)
- [ ] Target function accepts `%MwKernel.Context{}`, returns `{:ok, map()}`
- [ ] Entry added to `config :mw_router, :dispatch_config` with `dispatch: :internal`
- [ ] Flow Builder badge shows `⚡ internal dispatch` when adapter is selected
- [ ] Decision node: `field_path: "outcome"`, `value: "approve"` / `"decline"`

**HTTP dispatch (distributed deploy):**
- [ ] Target service reachable from the router node on the configured URL
- [ ] `RISK_SERVICE_URL` / `INTERNAL_TOKEN` env vars set in `config/runtime.exs`
- [ ] Service response JSON contains `"outcome"` key (or adjust `field_path`)
- [ ] Entry added to `config :mw_router, :dispatch_config` with `dispatch: {:http, url}`
- [ ] Flow Builder badge shows `🌐 HTTP · host:port` when adapter is selected