# Broker

The **Broker** is the component that sits between the {doc}`/api` and the {doc}`/router`. It handles all API-layer concerns: client session management, middleware execution, and WebSocket subscription routing.

```mermaid
flowchart TD
    C["Client (HTTP/WS)"] --> L["Listener (ASGI process)"]
    L -->|"sends constructed Event"| B["Broker (child process)"]
    B -->|"middleware + WS clients<br/>route responses back"| RQ["Router Queue → Router → Manager"]
```

---

## What the Broker does

The Broker has two traffic directions:

### Inbound (client → Router)

1. Receives a fully-constructed API Event from the Listener.
2. Executes `process_request` on each registered middleware in order.
3. If middleware allows (`CONTINUE`), puts the event on the `router_queue` for the Router to dispatch.
4. If middleware returns `STOP`, the event is discarded (no response is sent to the client).
5. If middleware returns `REPLACE_EVENT`, the replacement event continues through the pipeline.
6. If middleware returns `REDIRECT`, the original request is dropped and the replacement event is handled as a **response** — it runs the `process_response` middleware and is delivered back to the client. (On the response path the direction is reversed: the replacement is put on the `router_queue` as a new request.)

Subscription management (`EdriRegister` / `EdriUnregister`) is answered by the
Broker itself and never reaches the Router, but it runs `process_request` the
same way. A middleware that marks the response (the standard way to deny a
request) is answered on the spot: the marked event goes back to the client and
nothing is registered or removed. Authorizing queries alone would otherwise
leave a denied client free to subscribe to the same data.

### Outbound (Router → client)

1. Receives an Event back from the Router (carrying a populated `response`).
2. Executes `process_response` on each middleware in reverse of the request order (onion model).
3. Determines the **delivery mode** from the event's `handling` type:
   * `SPECIFIC` — sends to the one client that originated the event.
   * `SUBSCRIBED` — sends to all WS clients that registered for this event/param_set.
   * `ALL` — broadcasts to all connected WebSocket clients.
4. Hands the event to the **sender pool** (below), which writes it over the pipe to the originating Listener; the Listener serializes it (JSON / HTML / WS frame) and writes the response to the client socket.

---

(non-blocking-outbound-send-sender-pool)=
## Non-blocking outbound send (sender pool)

The Broker runs a **single-threaded** event loop. Each client has its own pipe to the Listener, and writing a large response to a client whose reader has fallen behind can **block** that pipe (the OS pipe buffer fills). If the broker did that write inline, one slow client would freeze the whole loop — every other response, registration, and even trivial request would stall behind it (head-of-line blocking). A handful of large concurrent responses could wedge the entire API.

So the actual `Connection.send` is offloaded to a **bounded pool of sender threads**:

* The event loop only **submits** (non-blocking) and immediately moves on.
* Pool threads perform the blocking send. A slow client ties up **one** pool thread, not the loop — and not other clients, up to `EDRI_API_SENDER_THREADS` simultaneously stuck clients.
* **Per-client FIFO is preserved**: at most one thread services a given client at a time, so a client's events (e.g. a WebSocket stream) never reorder.
* A failed send (broken pipe / reset) is reported back to the **broker thread**, which unregisters the client there — client and subscription state is never mutated from a pool thread.

For very large payloads, prefer streaming via `File` + {doc}`/shared_memory_pipe` over a big pickled response: it moves the bytes out-of-band with backpressure and keeps them off the broker entirely. The sender pool stops a slow client from taking the API down; it does not make copying hundreds of megabytes through the pipe cheap.

---

## WebSocket client management

The Broker owns the registry of active API clients.

### Client lifecycle

| Step | Event | Broker action |
|------|-------|---------------|
| Connect | `client.Register` | Assigns a unique key to the socket, stores `Client` |
| Subscribe | `manage.EdriRegister` | Inserts client into a per-event-type subscription trie |
| Push | outbound Event | Looks up subscribers via trie, sends to each |
| Unsubscribe | `manage.EdriUnregister` | Removes client from trie for that event/param_set |
| Disconnect | `client.Unregister` | Removes client from storage, closes socket, removes from all tries |

Only `client.Register` (and the internal dead-client `Unregister` a failed send
posts from the sender pool) travels over the shared broker queue. Once a
connection is registered, every request it sends — including its own eventual
`client.Unregister` — goes over that connection's own pipe; the Broker
identifies the sender by *which pipe delivered the event*, not by a key read
off the event, and watches every registered client's pipe alongside the
router pipe in the same `wait()` loop already used for worker pipes.

### Subscription trie

The Broker stores subscriptions in a **trie keyed by `param_set`**. Each node in the trie corresponds to one `(field_name, value)` pair. Clients are stored at leaf nodes.

When a server-push event arrives, the Broker walks the trie for that event type, matching the event's field values against the stored `param_set` entries, and collects all matching clients.

This means `param_set` lookups scale with subscription depth, not with the total number of subscribers.

---

## Built-in API documentation

When `EDRI_API_ENABLE_DOCUMENTATION` is enabled, the Broker responds to the built-in `EdriDocumentation` event with a description of all registered API events, their fields, types, and validators. The event is registered at `GET /edri/docs` (REST returns the JSON below; a browser receives the rendered HTML page) and is also reachable as the WebSocket command `edri-docs`.

```json
{
  "command": "edri-docs",
  "response": {
    "data": {
      "apiResources": { "...": "..." },
      "wrapped": true
    },
    "status": "OK"
  }
}
```

`EDRI_API_ENABLE_DOCUMENTATION` defaults to enabled in development and disabled in production (it follows `ENVIRONMENT`). To force it off explicitly:

```bash
export EDRI_API_ENABLE_DOCUMENTATION=false
```

---

## Broker and middleware: what runs where

Middleware is registered on the Broker, not on the Listener. The split between Listener and Broker is:

| Step | Runs in |
|------|---------|
| Input extraction (headers, body, path, query, cookies) | Listener |
| Type coercion and `inject(...)` validation | Listener |
| `process_request` middleware | Broker |
| Routing to Manager | Router |
| Manager handler | Manager |
| `process_response` middleware | Broker |
| Serialization and response send | Broker → Listener socket |

This means middleware sees events **after validation has already succeeded**. If validation fails (400 Bad Request), the event is rejected by the Listener before it reaches the Broker.

---

## Configuration

The Broker has no dedicated environment variables. Its behavior is controlled through:

| Setting | Variable | Effect |
|---------|----------|--------|
| Response format | `EDRI_API_RESPONSE_WRAPPED` | Wraps JSON in `{"data": ...}` |
| Documentation | `EDRI_API_ENABLE_DOCUMENTATION` | Enables/disables the `/edri/docs` endpoint and the `edri-docs` command |
| Response timeout | `EDRI_API_RESPONSE_TIMEOUT` | How long the Listener waits for the response (round-tripped through the Broker) before returning 500 |
| Sender threads | `EDRI_API_SENDER_THREADS` | Size of the outbound [sender pool](#non-blocking-outbound-send-sender-pool) — how many simultaneously slow clients can be tolerated before sends queue (default 32) |

See {doc}`/env` for the full list.

---

## Starting the Broker

The Broker is started automatically when you call `edri.start_api(...)`:

```python
edri.start_api(
    middlewares=[AuthenticationMiddleware(), LoggingMiddleware()],
)
```

You do not instantiate or start the Broker directly.

---

## Error handling

If the Broker encounters an error forwarding an event (e.g. broken socket), the failure surfaces on a [sender-pool](#non-blocking-outbound-send-sender-pool) thread, which reports it back to the broker thread; the broker logs a warning and unregisters the client there. The Router is not notified — the response simply does not reach the client.

If middleware raises an unhandled exception, the Broker logs the error, sets `InternalServerErrorResponseDirective` on the response, and sends it back to the client (it then re-raises so the failure is logged at the call site too). A middleware that returns a value other than a `MiddlewareControl` is handled the same way.

---

## See also

* {doc}`/middleware` — implementing middleware, pipeline order, `MiddlewareControl`
* {doc}`/api` — API layer overview
* {doc}`/websocket` — WS commands, `register`/`unregister`, `handling` types
* {doc}`/api_event` — `@api(handling=...)`, delivery modes