Sagents.Subscriber (Sagents v0.14.2)

Consumer-side helpers for Sagents.Publisher producers.

Captures the boilerplate of:

Subscribing to a producer by id rather than pid.
Monitoring the producer so we know when it dies.
Watching Phoenix.Presence arrivals for that id (to handle crash-restart with a new pid, or Horde migration to a new node).
Re-subscribing automatically when the producer reappears.
Cleaning up on caller exit (best-effort — the producer's monitor on the caller will clean us up anyway).

Targets two consumer shapes:

Plain GenServer / process — call subscribe_to_agent/2 and friends directly; pair with handle_publisher_down/3 and handle_presence_diff/3 from your handle_info/2.
LiveView (or anything with a socket) — use Sagents.Subscriber injects handle_info/2 clauses for :DOWN and Phoenix presence diffs that delegate to those helpers.

Subscription handle

Each active subscription is tracked in the caller's local subs map (kept on the LiveView socket under socket.private.sagents_subs, or threaded through manually for plain-process callers).

Map shape:

%{
  {:agent, agent_id} => %{
    channel: :main | :debug,
    server_pid: pid() | nil,
    monitor_ref: reference() | nil,
    state: :subscribed | :pending
  },
  {:filesystem, scope_key} => %{...}
}

States:

:subscribed — we have a live subscription on server_pid, monitored.
:pending — we want to subscribe but the producer isn't running. Will retry on the next presence arrival.

Departure vs arrival

Monitors are reliable for departure (:DOWN fires within a scheduler tick of process death, even cross-node when the connection drops). Phoenix.Presence is reliable for arrival (presence_diff with joins). We never depend on presence_diff.leaves — if it's delayed, we keep the :subscribed state until :DOWN fires, which it will.

Summary

Types

sub_entry()

sub_key()

subs()

Functions

handle_presence_diff(subs, arg2, arg3)

Handle a Phoenix.Presence diff for the agent presence topic.

handle_publisher_down(subs, ref, reason)

Handle a :DOWN from one of the producer pids we subscribed to.

presence_topic()

Returns the Phoenix.Presence topic string the agent presence layer uses.

subscribe_to_agent(subs, agent_id, channel \\ :main)

Subscribe the calling process to an agent's main channel, threading the subscription map through.

subscribe_to_agent(subs, agent_id, channel, subscriber_pid)

Subscribe subscriber_pid to an agent's channel, threading the subscription map through.

subscribe_to_filesystem(subs, scope_key, channel \\ :main)

Subscribe the calling process to filesystem change events for scope_key.

unsubscribe_from_agent(subs, agent_id)

Unsubscribe from an agent. Tears down monitor and pending state.

unsubscribe_from_filesystem(subs, scope_key)

Unsubscribe from a filesystem.

Types

sub_entry()

@type sub_entry() :: %{
  channel: atom(),
  server_pid: pid() | nil,
  monitor_ref: reference() | nil,
  state: :subscribed | :pending
}

sub_key()

@type sub_key() :: {:agent, String.t()} | {:filesystem, term()}

subs()

@type subs() :: %{required(sub_key()) => sub_entry()}

Functions

handle_presence_diff(subs, arg2, arg3)

@spec handle_presence_diff(subs(), String.t(), map()) :: subs()

Handle a Phoenix.Presence diff for the agent presence topic.

joins whose key matches a :pending agent subscription triggers a re-subscribe. Returns the (possibly updated) subs map.

Entries that cannot be re-subscribed stay :pending rather than being dropped, so a later diff can still revive them. That covers a node whose own registry is unavailable: the presence topic is cluster-wide, so this fires for agents booting anywhere in the cluster, including while this node drains and can resolve nothing.

handle_publisher_down(subs, ref, reason)

@spec handle_publisher_down(subs(), reference(), term()) ::
  {:matched, subs()} | :no_match

Handle a :DOWN from one of the producer pids we subscribed to.

Returns {:matched, new_subs} if the ref belonged to a tracked subscription (now flipped to :pending), otherwise :no_match.

presence_topic()

@spec presence_topic() :: String.t()

Returns the Phoenix.Presence topic string the agent presence layer uses.

Subscribe to this topic with Phoenix.PubSub.subscribe/2 to receive presence_diff broadcasts that drive auto-resubscription.

subscribe_to_agent(subs, agent_id, channel \\ :main)

@spec subscribe_to_agent(subs(), String.t(), :main | :debug) :: subs()

Subscribe the calling process to an agent's main channel, threading the subscription map through.

Returns the updated subs map. If the agent is not currently running, the subscription is recorded in :pending state and will become live as soon as the agent appears in Phoenix.Presence on the agent presence topic.

subscribe_to_agent(subs, agent_id, channel, subscriber_pid)

@spec subscribe_to_agent(subs(), String.t(), :main | :debug, pid()) :: subs()

Subscribe subscriber_pid to an agent's channel, threading the subscription map through.

subscribe_to_agent/3 subscribes the calling process. Use this arity when the process receiving the events is a different one, such as a host running one receiver per conversation. Main-channel events arrive as {:agent, event} and do not name the agent that sent them, so a single process holding two subscriptions cannot separate them; a receiver per agent can.

Events go to subscriber_pid. The returned subs map, and the producer monitor recorded in it, belong to the calling process, which is therefore the process that must run handle_publisher_down/3 and handle_presence_diff/3.

A revived subscription moves to the caller

An entry rests at :pending when no agent is running at subscribe time, and returns there when the producer goes down. handle_presence_diff/3 revives a pending entry by subscribing the process that calls it, and the entry does not record subscriber_pid, so a revived subscription made with this arity delivers to the caller rather than to the original receiver. Subscribe from the receiving process itself wherever that is reachable.