9d4a5f6d81
- help_live/index.ex: 2,642→173 lines, 15 section modules + sidebar extracted MikroTik now real section, dead if false removed - agent_channel.ex: 2,506→1,751 lines, 3 helper modules extracted (heartbeat/subscriptions/job_builder), decode_and_process/4 eliminates 8 repeated handle_in patterns, guard-based size checks - antenna_catalog.ex: 1,174→47 lines, 107 specs → priv/antennas/catalog.json - topology.ex: unbounded query → batched loading (100/batch), all guard errors fixed, zero if/case/cond conditionals - proto: decoder_macros.ex + field_specs.ex infrastructure for macro-generated protobuf decoders
92 lines
2.9 KiB
Elixir
92 lines
2.9 KiB
Elixir
defmodule ToweropsWeb.AgentChannel.Subscriptions do
|
|
@moduledoc """
|
|
PubSub subscription management for agent channels.
|
|
|
|
Handles all PubSub topic subscriptions and event broadcasting for agent
|
|
lifecycle events (connection, disconnection, heartbeat).
|
|
"""
|
|
|
|
@doc """
|
|
Subscribes the current process to all PubSub topics relevant to the given
|
|
agent token.
|
|
|
|
Topics subscribed:
|
|
- `agent:{id}:assignments` — device assignment changes
|
|
- `agent:{id}:discovery` — discovery requests
|
|
- `agent:{id}:backup` — backup requests
|
|
- `agent:{id}:credential_test` — credential test requests
|
|
- `agent:{id}:live_poll` — live poll requests
|
|
- `checks:agent:{id}` — check changes
|
|
- `agent:{id}:lifecycle` — token lifecycle events
|
|
- `agent:{id}:latency_probe` — latency probes (cloud pollers only)
|
|
"""
|
|
@spec subscribe_all(map()) :: :ok
|
|
def subscribe_all(agent_token) do
|
|
id = agent_token.id
|
|
|
|
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{id}:assignments")
|
|
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{id}:discovery")
|
|
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{id}:backup")
|
|
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{id}:credential_test")
|
|
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{id}:live_poll")
|
|
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "checks:agent:#{id}")
|
|
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{id}:lifecycle")
|
|
|
|
subscribe_latency_if_cloud_poller(agent_token)
|
|
|
|
:ok
|
|
end
|
|
|
|
# Cloud pollers additionally subscribe to latency probe requests
|
|
defp subscribe_latency_if_cloud_poller(%{is_cloud_poller: true} = agent_token) do
|
|
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token.id}:latency_probe")
|
|
:ok
|
|
end
|
|
|
|
defp subscribe_latency_if_cloud_poller(_agent_token), do: :ok
|
|
|
|
@doc """
|
|
Broadcasts an agent connection event to the `agents:health` topic.
|
|
"""
|
|
@spec broadcast_connection(Ecto.UUID.t(), Ecto.UUID.t()) :: :ok
|
|
def broadcast_connection(agent_token_id, organization_id) do
|
|
_ =
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"agents:health",
|
|
{:agent_connected, agent_token_id, organization_id}
|
|
)
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc """
|
|
Broadcasts an agent disconnection event to the `agents:health` topic.
|
|
"""
|
|
@spec broadcast_disconnection(Ecto.UUID.t(), Ecto.UUID.t()) :: :ok
|
|
def broadcast_disconnection(agent_token_id, organization_id) do
|
|
_ =
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"agents:health",
|
|
{:agent_disconnected, agent_token_id, organization_id}
|
|
)
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc """
|
|
Broadcasts an agent heartbeat event to the `agents:health` topic.
|
|
"""
|
|
@spec broadcast_heartbeat(Ecto.UUID.t(), Ecto.UUID.t()) :: :ok
|
|
def broadcast_heartbeat(agent_token_id, organization_id) do
|
|
_ =
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"agents:health",
|
|
{:agent_heartbeat, agent_token_id, organization_id}
|
|
)
|
|
|
|
:ok
|
|
end
|
|
end
|