# Phase 3 — WebSocket Gateway + Real-time Push

**Duration:** Weeks 9–10
**Status:** ✅ Complete — 2026-04-26
**Goal:** WebSocket clients receive live transaction status and system notifications.

---

## Deliverable

1. Clients connect via `wss://host/socket` with JWT authentication
2. Subscribe to `"transactions:<id>"` channel — receive live status updates
3. Subscribe to `"notifications:system"` — receive operational alerts
4. Core banking async callbacks (settlement) push to subscribed clients in real time
5. File pipeline completion events push to subscribed clients

---

## Tasks

### 1. gateway_ws — Phoenix Channels

```elixir
# apps/gateway_ws/lib/gateway_ws_web/socket.ex
defmodule GatewayWs.UserSocket do
  use Phoenix.Socket

  channel "transactions:*", GatewayWs.TransactionChannel
  channel "notifications:*", GatewayWs.NotificationChannel

  @impl true
  def connect(%{"token" => token}, socket, _connect_info) do
    case MwAuth.JWT.verify(token) do
      {:ok, claims} ->
        {:ok, socket
          |> assign(:user_id, claims["sub"])
          |> assign(:tenant_id, claims["tenant_id"])
          |> assign(:roles, claims["roles"])}
      {:error, _} -> :error
    end
  end

  @impl true
  def id(socket), do: "users_socket:#{socket.assigns.user_id}"
end
```

```elixir
defmodule GatewayWs.TransactionChannel do
  use Phoenix.Channel

  def join("transactions:" <> tx_id, _params, socket) do
    if authorized?(socket, tx_id) do
      Phoenix.PubSub.subscribe(MwCore.PubSub, "transactions:#{tx_id}")
      current_status = fetch_current_status(tx_id)
      {:ok, current_status, assign(socket, :tx_id, tx_id)}
    else
      {:error, %{reason: "unauthorized"}}
    end
  end

  def handle_info({:transaction_update, payload}, socket) do
    push(socket, "status_update", payload)
    {:noreply, socket}
  end
end
```

### 2. PubSub Wiring

When `adapter_banking` receives an async settlement callback from core banking:

```elixir
defmodule AdapterBanking.CallbackHandler do
  def handle_settlement(%{"reference" => ref, "status" => status} = payload) do
    # Update DB
    InfraRepo.Repo.update_all(
      from(t in MwCore.Transaction, where: t.cbs_reference == ^ref),
      set: [status: status]
    )
    # Push to WebSocket subscribers
    Phoenix.PubSub.broadcast(
      MwCore.PubSub,
      "transactions:#{ref}",
      {:transaction_update, %{status: status, settled_at: DateTime.utc_now()}}
    )
    # Write audit
    MwAudit.Logger.write(%{type: "settlement_callback", reference: ref, status: status})
  end
end
```

When Broadway file pipeline completes:

```elixir
defmodule InfraQueue.FilePipeline do
  # after successful batch ack:
  def on_complete(job_id, result) do
    Phoenix.PubSub.broadcast(MwCore.PubSub, "jobs:#{job_id}",
      %{event: "batch_complete", rows_processed: result.count})
  end
end
```

### 3. Notification Channel

System-wide broadcasts for operational events:

```elixir
defmodule GatewayWs.NotificationChannel do
  use Phoenix.Channel

  def join("notifications:system", _params, socket) do
    if "operator" in socket.assigns.roles or "admin" in socket.assigns.roles do
      Phoenix.PubSub.subscribe(MwCore.PubSub, "system:notifications")
      {:ok, socket}
    else
      {:error, %{reason: "insufficient_role"}}
    end
  end

  def handle_info({:system_notification, payload}, socket) do
    push(socket, "notification", payload)
    {:noreply, socket}
  end
end
```

Notifications are broadcast from:
- Circuit breaker opens: `MwRouter.CircuitBreaker` on state change
- DLQ depth threshold exceeded: `InfraQueue.FilePipeline`
- Adapter health check failure: scheduled health poller in `infra_telemetry`

### 4. Client-side Heartbeat

Channels automatically receive `phx_heartbeat` from Phoenix. Configure timeout:

```elixir
config :gateway_ws, GatewayWs.Endpoint,
  websocket: [timeout: 45_000]
```

---

## WebSocket API Reference

### Connection
```
wss://host/socket/websocket?token=<jwt>&vsn=2.0.0
```

### Channels

| Topic | Direction | Event | Payload |
|-------|-----------|-------|---------|
| `transactions:<id>` | Server→Client | `status_update` | `{status, settled_at, ...}` |
| `transactions:<id>` | Server→Client | `error` | `{code, message}` |
| `notifications:system` | Server→Client | `notification` | `{level, title, body, adapter}` |
| `jobs:<id>` | Server→Client | `batch_complete` | `{rows_processed, duration_ms}` |

---

## Acceptance Criteria

- [ ] Client connects with valid JWT and joins `transactions:<id>` channel
- [ ] Client is rejected with `:error` for invalid JWT on socket connect
- [ ] Settlement callback from banking adapter pushes `status_update` within 1s of receipt
- [ ] File pipeline completion pushes `batch_complete` to subscribed clients
- [ ] Circuit breaker state change broadcasts to `notifications:system`
- [ ] Heartbeat keeps connections alive for 60+ seconds idle
- [ ] Disconnected clients do not receive PubSub broadcasts (no memory leak)