# South Plane — Adapters

All south-side adapters implement `MwKernel.Adapter`. The core plane never knows which
protocol an adapter uses — only the adapter knows.

---

## Adapter Behaviour Contract

```elixir
defmodule MwKernel.Adapter do
  @type state :: map()
  @type message :: MwKernel.Message.t()

  @doc "Open connection / initialise pool. Called at application start."
  @callback connect(config :: map()) :: {:ok, state()} | {:error, term()}

  @doc "Send a canonical message to the downstream system. Return canonical response."
  @callback send(state(), message()) :: {:ok, message()} | {:error, term()}

  @doc "Check if the downstream system is reachable. Used by health dashboard."
  @callback health_check(state()) :: :ok | {:error, term()}

  @doc "Gracefully close connections. Called on application shutdown."
  @callback disconnect(state()) :: :ok
end
```

---

## adapter_banking — Core Banking

**Protocol:** ISO 8583 over HTTP (or proprietary REST — configurable)
**Connection:** Finch pool, persistent, authenticated per config

### ISO 8583 Message Types Supported (Phase 1)

| MTI | Description |
|-----|-------------|
| 0200 | Financial Transaction Request |
| 0210 | Financial Transaction Response |
| 0400 | Reversal Request |
| 0410 | Reversal Response |
| 0800 | Network Management Request (sign-on/off) |

### Configuration
```elixir
config :adapter_banking, :connection, %{
  base_url: System.get_env("CBS_BASE_URL"),
  api_key: System.get_env("CBS_API_KEY"),
  pool_size: 20,
  timeout_ms: 5_000,
  protocol: :iso8583_http    # :iso8583_http | :proprietary_rest
}
```

### Async Callback Handling

Core banking systems often return `"status": "pending"` immediately and deliver a
settlement callback minutes later via:
- Webhook: `POST /internal/callbacks/banking` (gateway_api, not public)
- TCP callback socket: `AdapterBanking.CallbackServer` (GenServer, listens on port)

`AdapterBanking.CallbackHandler` processes these and broadcasts via PubSub.

---

## adapter_dw — Data Warehouse

**Protocol:** REST API or JDBC-style HTTP endpoint (configurable per DW vendor)
**Access patterns:** Bulk insert, parameterised query, streaming result set

### Streaming Adapter

`adapter_dw` implements `MwKernel.StreamingAdapter` in addition to the base behaviour:

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

  @doc "Returns a lazy stream of MwKernel.Message for large result sets."
  @callback stream(state(), query :: map()) ::
    {:ok, Enumerable.t(MwKernel.Message.t())} | {:error, term()}
end
```

Usage in `gateway_api` for large report queries:
```elixir
{:ok, stream} = AdapterDw.stream(state, %{report: "daily_summary", date: "2026-04-25"})
conn
|> put_resp_content_type("application/x-ndjson")
|> send_chunked(200)
|> then(fn conn ->
  Enum.reduce_while(stream, conn, fn record, conn ->
    case chunk(conn, Jason.encode!(record) <> "\n") do
      {:ok, conn} -> {:cont, conn}
      {:error, _} -> {:halt, conn}
    end
  end)
end)
```

### Batch Insert Rate
Target: 5,000+ rows/second via Broadway batcher with batch size 500.

---

## adapter_http — Generic HTTP/REST/SOAP

**Protocol:** HTTP/1.1 or HTTPS, JSON or XML body
**Use case:** Any internal REST or SOAP API that doesn't have a dedicated adapter

### Endpoint Configuration
```elixir
config :adapter_http, :endpoints, %{
  "payment_validation" => %{
    base_url: "https://validation.internal",
    auth: {:bearer, System.get_env("VALIDATION_TOKEN")},
    timeout_ms: 3_000,
    retries: 3,
    retry_backoff_ms: 500,
    format: :json
  },
  "fraud_check" => %{
    base_url: "https://fraud.internal",
    auth: {:basic, "user", System.get_env("FRAUD_PASS")},
    timeout_ms: 1_000,
    retries: 1,
    format: :json
  },
  "legacy_soap_system" => %{
    base_url: "https://legacy.internal/ws",
    auth: {:wsse, System.get_env("WSSE_TOKEN")},
    timeout_ms: 8_000,
    retries: 2,
    format: :soap
  }
}
```

Route rule links `message_type` → endpoint key:
```
route_rule: %{message_type: "payment.validate", adapter: "AdapterHttp",
              metadata: %{endpoint: "payment_validation"}}
```

### Retry Strategy
```elixir
defmodule AdapterHttp.Retry do
  def with_retry(fun, retries, backoff_ms) do
    Enum.reduce_while(1..retries, {:error, :not_attempted}, fn attempt, _acc ->
      case fun.() do
        {:ok, _} = ok -> {:halt, ok}
        {:error, reason} when attempt < retries ->
          Process.sleep(backoff_ms * attempt)  # linear backoff
          {:cont, {:error, reason}}
        {:error, reason} -> {:halt, {:error, reason}}
      end
    end)
  end
end
```

---

## adapter_file — File-Based Systems

**Protocol:** SFTP (primary), local filesystem (dev/test), FTP (legacy, optional)
**Formats:** CSV, XML, ISO fixed-width, proprietary flat files

### File Processing Flow

```
SFTP Server
    │ FileWatcher polls every 60s
    │ downloads new files
    ▼
CsvParser.stream(contents)        NimbleCSV streaming — no full-file buffer
    │ row by row
    ▼
Broadway.FilePipeline             back-pressure producer
    │ batch of 500 rows
    ▼
AdapterDw.BatchLoader             bulk insert to DW
    │ DLQ on failure
    ▼
PubSub: "jobs:#{job_id}"          notify gateway_ws
```

### File Cursor Management

To avoid reprocessing files on restart, the watcher maintains a cursor:
```
DB table: file_processing_log (filename, processed_at, row_count, status)
```

On startup, FileWatcher loads `file_processing_log` to rebuild `seen_files` set.

### Outbound File Generation

`adapter_file` also supports generating and uploading files to SFTP:
```elixir
AdapterFile.SFTPClient.upload(config, "settlement_20260426.csv", csv_content)
```

Triggered by route with direction `:outbound` and message type `"file.settlement_upload"`.