# `Anubis.Server.Transport.StreamableHTTP`

StreamableHTTP transport implementation for MCP servers.

This module manages SSE (Server-Sent Events) connections for server-to-client
communication. In the refactored architecture, request handling is done directly
by Session processes - this module only manages SSE handlers and notifications.

## Features

- SSE handler registration for server-to-client push
- Automatic handler cleanup on disconnect
- Keepalive messages to maintain connections
- Notification broadcasting to connected clients
- Cross-node delivery: SSE handlers join a `:pg` group keyed by session id,
  so `route_to_session/3` reaches a handler whose GET stream landed on a
  different node than the session owner (see "Cluster routing" below)
- Optional resumability: when an `:event_store` is configured, messages on a
  session's standalone stream are recorded with monotonic ids and replayed on
  reconnect via `Last-Event-ID`. Messages fired while no handler is attached
  are still recorded (bounded by `:stream_grace`), so they survive reconnect
  gaps. See `Anubis.Server.Transport.StreamableHTTP.EventStore`.

## Cluster routing

Each transport registers its SSE handlers in a node-local map AND in a `:pg`
group (scope derived from the transport name) keyed by session id. When
`route_to_session/3` finds no local handler, it falls back to the `:pg`
group and delivers `{:sse_message, message}` directly to the remote handler
pid, mirroring how `Anubis.Server.Registry.PG` routes inbound requests.

This removes the sticky-session requirement for server-to-client traffic:
the SSE GET may terminate on any node, regardless of where the session
process lives. Handler pids are monitored by `:pg`, so crashed or
disconnected handlers disappear from the group automatically.

Notes:

- The scope is derived from the transport's registered name, so cluster
  routing requires an atom name (the default naming used by
  `Anubis.Server.Supervisor`). Transports registered with a `:via` or
  `:global` name keep node-local delivery only.
- The broadcast path (`send_message/3` without a `:session_id`) remains
  node-local.
- Resumability across nodes additionally requires a shared
  `:event_store`; with the default node-local store, events emitted while
  the handler lives on another node are delivered live but not recorded.

## Usage

StreamableHTTP is typically started through the server supervisor:

    Anubis.Server.start_link(MyServer, [],
      transport: :streamable_http,
      streamable_http: [port: 4000]
    )

For integration with existing Phoenix/Plug applications:

    # In your router
    forward "/mcp", Anubis.Server.Transport.StreamableHTTP.Plug,
      server: MyApp.MCPServer

# `option`

```elixir
@type option() ::
  {:server, GenServer.server()} | {:name, GenServer.name()} | GenServer.option()
```

StreamableHTTP transport options

- `:server` - The server module (required)
- `:name` - Name for registering the GenServer (required)

# `t`

```elixir
@type t() :: GenServer.server()
```

# `child_spec`

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `close_session_stream`

```elixir
@spec close_session_stream(GenServer.server(), String.t()) :: :ok | {:error, term()}
```

Closes a session's resumable stream and drops its recorded events. Called on
`DELETE` (explicit session termination). No-op when resumability is disabled.

Fails closed: when the event store cannot delete the recorded events the
error is returned and the stream stays open in transport state, so a retried
`DELETE` can complete the cleanup.

# `get_schema`

# `get_sse_handler`

```elixir
@spec get_sse_handler(GenServer.server(), String.t()) :: pid() | nil
```

Returns the SSE handler pid for a session, or `nil` if none is connected.

# `handler_count`

```elixir
@spec handler_count(GenServer.server()) :: non_neg_integer()
```

Returns the number of connected SSE handlers.

# `handler_count`

```elixir
@spec handler_count(GenServer.server(), (map() -> as_boolean(term()))) ::
  non_neg_integer()
```

Returns the number of connected SSE handlers whose metadata satisfies `selector`.

`selector` receives each handler's opaque metadata map (see
`register_sse_handler/3`) and returns a truthy value to count that handler.

# `parse_options`

# `parse_options!`

# `register_sse_handler`

```elixir
@spec register_sse_handler(GenServer.server(), String.t()) :: :ok | {:error, term()}
```

Registers the calling process as the SSE handler for a session.

Called by the Plug when establishing an SSE connection. Equivalent to
`register_sse_handler/3` with empty metadata.

# `register_sse_handler`

```elixir
@spec register_sse_handler(GenServer.server(), String.t(), map()) ::
  :ok | {:error, term()}
```

Registers the calling process as the SSE handler for a session, attaching an
opaque `metadata` map.

The transport stores `metadata` verbatim and never interprets it. Hosts use it
to tag a subscriber with application-defined attributes (tenant, user, feature
scope, ...) so later `send_message_to_subscribers/4` and `handler_count/2` calls
can select on them. The Plug populates it from its `:subscriber_metadata`
callback; direct callers may pass any map.

# `resumability_config`

```elixir
@spec resumability_config(GenServer.server()) ::
  {term() | nil, non_neg_integer() | nil}
```

Returns the resumability config for this transport as `{event_store, retry}`,
where `event_store` is `{module, name}` or `nil` and `retry` is the SSE
`retry:` value in milliseconds or `nil`. Read by the Plug when opening an SSE
stream.

# `route_to_session`

```elixir
@spec route_to_session(GenServer.server(), String.t(), binary()) ::
  :ok | {:error, term()}
```

Routes a message to a specific session's SSE handler for server-to-client push.

When resumability is enabled the message is also recorded on the session's
stream so it can be replayed on reconnect. The message is recorded even if no
handler is currently attached, in which case `:ok` is still returned.

# `send_message_to_subscribers`

```elixir
@spec send_message_to_subscribers(
  GenServer.server(),
  (map() -> as_boolean(term())),
  binary(),
  keyword()
) :: :ok | {:error, term()}
```

Sends a message to every connected SSE handler whose metadata satisfies `selector`.

`selector` receives each handler's opaque metadata map (see
`register_sse_handler/3`) and returns a truthy value for the subscribers that
should receive `message`. This complements `route_to_session/3` (a single
session) and `send_message/3` (broadcast to all handlers) with delivery to an
arbitrary, application-defined subset.

`opts` accepts `:timeout` (default `5000`).

# `unregister_sse_handler`

```elixir
@spec unregister_sse_handler(GenServer.server(), String.t(), pid() | nil) :: :ok
```

Unregisters the SSE handler for a session. Called when the SSE connection closes.

---

*Consult [api-reference.md](api-reference.md) for complete listing*