Routing

Topics, publish, and dispatch

A topic is a tuple. Fanout walks the ETS edges for that tuple. Storage, when asked to persist, uses the same tuple as the stream key.

Topic shapes

The whole term must fit in max_topic_bytes (512) and max_topic_depth (4). Untrusted strings are not turned into atoms. Integer ids are accepted where the validator allows an id, which is why channel:i:7 on the foreign gateway becomes {:channel, 7}.

Publish

{:ok, %{event_id: id, sent: n, cluster: cluster}} =
  Wiregrid.publish(:chat, {:channel, "general"}, %{type: :message, body: "hello"},
    session_id: session_id,
    persist: true,
    class: :durable,
    meta: %{source: "app"},
    exclude_sessions: [session_id]
  )

Publish options are :event_id, :persist, :meta, :class, :session_id, :cluster, :exclude_sessions, and :exclude_users. Class defaults to :durable. persist defaults to false. cluster defaults to true and is then combined with the instance cluster flag, so a single-node instance still reports cluster :disabled. Exclusion lists are capped at 256 entries. The event itself must fit in max_event_bytes (1,048,576). The encoded form must fit in max_encoded_event_bytes (1,572,864), and that ceiling must be at least as large as max_event_bytes.

Order inside publish/4:

1. The instance must be accepting work. A drained instance refuses the call.
2. Options, topic, and event are validated.
3. If session_id is nil, authorize_actor/6 returns :ok without calling the authorizer. That path is for trusted in-process publishers.
4. If session_id is set, it must name a live session. Authorizer.authorize/4 runs with action :publish and the topic as the resource. A missing session is :unknown_session.
5. The codec encodes the event once.
6. If persist is true, storage appends before local fanout.
7. Local fanout walks subscribers. Cluster forwarding runs when clustering is on and the call asked for it.

The result map includes the fanout counters (sent, dropped, evicted, gone, excluded, overloaded), event_id, and cluster. A batch is not a transaction: publish_batch/4 validates every event first, then reports {:ok, %{index: i, ...}} or {:error, %{index: i, reason: reason}} per event, with completed and failed totals. An empty list is {:error, :empty_batch}. The batch cap is max_batch_items (512).

Prepared events

{:ok, prepared} = Wiregrid.prepare(:chat, %{type: :notice, body: "deployed"})
{:ok, _} = Wiregrid.publish_prepared(:chat, {:channel, "general"}, prepared, persist: false)
{:ok, _} = Wiregrid.publish_topics_prepared(:chat, [{:channel, "general"}, {:document, "log"}], prepared)

prepare/2 validates and encodes once. Prepared publish, room, user, session, and dispatch functions reuse those bytes. The handle is signed for the instance that created it. Keep it inside the node. It is an in-process optimization, and it is not a credential a client should be allowed to submit.

publish_topics/4 checks every topic and every authorization decision before the first persist or fanout. The result is still per topic. publish_room/4 and publish_rooms/4 fan out along room membership rather than topic subscriptions. The authorizer action for a room publish is :publish_room.

Direct sends

Wiregrid.send_user(:chat, "bob", event, session_id: session_id, persist: true, class: :durable)
Wiregrid.send_session(:chat, other_session_id, event, session_id: session_id)
Wiregrid.send_users(:chat, ["bob", "cara"], event, session_id: session_id)
Wiregrid.send_sessions(:chat, [sid_a, sid_b], event, class: :ephemeral)

User sends accept :class, :session_id, :cluster, :event_id, :persist, and :meta. A persisted direct event uses {:user, user_id} as the stream, which is the stream inbox_history reads. Session sends accept :class and :session_id. When the option list is empty, authorize_direct/6 returns :ok without a callback. A nil session_id does the same. A present session id is checked and then authorized.

Dispatch

targets = [
  {:topic, {:channel, "general"}},
  {:room, {:room, "lobby"}},
  {:user, "bob"},
  {:session, other_session_id}
]
{:ok, _} = Wiregrid.dispatch(:chat, targets, event, session_id: session_id)
{:ok, plan} = Wiregrid.compile_dispatch(:chat, targets)
{:ok, _} = Wiregrid.dispatch_plan(:chat, plan, event, session_id: session_id)

Targets are {:topic, topic}, {:room, topic}, {:user, user_id}, or {:session, session_id}. The set is bounded by max_batch_targets (4,096) and deduplicated before side effects. Execution is still per group. compile_dispatch/2 freezes a checked topology into a plan that dispatch_plan/4 and dispatch_plan_prepared/4 can reuse. Plans are instance-local values, same as prepared events.

Internet-facing Cowboy commands inject the socket’s session id and reject a client that tries to pass :session_id itself (:transport_session_override). In-process code that omits session_id skips the authorizer. Put a real authorizer on any instance a transport can reach, and pass the acting session on every call that represents a user.