Graham McIntire
52329cf959
Make the 500ms device deletion delay configurable via application config. Set to 10ms in test environment (vs 500ms in production). This sleep waits for in-flight polling jobs to complete before deleting a device. In tests with mocked jobs, this delay is unnecessary. Affected tests: - All device deletion tests (were 500-650ms, now ~100-150ms) - Equipment context tests - API controller tests - LiveView tests with device deletion Performance: 28.3s → 25.2s (11% faster) Total improvement from baseline: 52s → 25.2s (51.5% faster)
1170 lines
35 KiB
Elixir
1170 lines
35 KiB
Elixir
defmodule Towerops.Devices do
|
|
@moduledoc """
|
|
The Devices context.
|
|
"""
|
|
|
|
import Ecto.Query
|
|
|
|
alias Towerops.Agents
|
|
alias Towerops.Devices.CredentialResolver
|
|
alias Towerops.Devices.Device, as: DeviceSchema
|
|
alias Towerops.Devices.Event
|
|
alias Towerops.Monitoring
|
|
alias Towerops.Organizations
|
|
alias Towerops.Organizations.SubscriptionLimits
|
|
alias Towerops.Repo
|
|
alias Towerops.Sites
|
|
alias Towerops.Workers.DeviceMonitorWorker
|
|
alias Towerops.Workers.DiscoveryWorker
|
|
|
|
@doc """
|
|
Returns the list of devices for a site.
|
|
Ordered by custom display_order (if set), then alphabetically by name.
|
|
"""
|
|
@spec list_site_devices(String.t()) :: [DeviceSchema.t()]
|
|
def list_site_devices(site_id) do
|
|
Repo.all(
|
|
from(e in DeviceSchema,
|
|
where: e.site_id == ^site_id,
|
|
order_by: [asc: e.display_order, asc: e.name]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Returns the list of all devices for an organization.
|
|
Ordered by custom display_order (if set), then alphabetically by name.
|
|
|
|
Supports filtering by:
|
|
- site_id: Filter by specific site
|
|
- status: Filter by status ("up", "down", "unknown")
|
|
"""
|
|
@spec list_organization_devices(String.t(), map()) :: [DeviceSchema.t()]
|
|
def list_organization_devices(organization_id, filters \\ %{}) do
|
|
query =
|
|
from(e in DeviceSchema,
|
|
left_join: s in assoc(e, :site),
|
|
where: e.organization_id == ^organization_id,
|
|
order_by: [asc: e.display_order, asc: e.name],
|
|
preload: [site: s]
|
|
)
|
|
|
|
query =
|
|
if site_id = filters["site_id"] do
|
|
where(query, [e], e.site_id == ^site_id)
|
|
else
|
|
query
|
|
end
|
|
|
|
query =
|
|
if status = filters["status"] do
|
|
where(query, [e], e.status == ^status)
|
|
else
|
|
query
|
|
end
|
|
|
|
Repo.all(query)
|
|
end
|
|
|
|
@doc """
|
|
Returns all devices for multiple organizations (for GDPR data access).
|
|
"""
|
|
@spec list_devices_for_organizations([String.t()]) :: [DeviceSchema.t()]
|
|
def list_devices_for_organizations(organization_ids) when is_list(organization_ids) do
|
|
Repo.all(
|
|
from(e in DeviceSchema,
|
|
left_join: s in assoc(e, :site),
|
|
join: o in assoc(e, :organization),
|
|
where: e.organization_id in ^organization_ids,
|
|
order_by: [asc: e.organization_id, asc: e.name],
|
|
preload: [site: s, organization: o]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Returns the count of devices for an organization.
|
|
"""
|
|
@spec count_organization_devices(String.t()) :: integer()
|
|
def count_organization_devices(organization_id) do
|
|
Repo.aggregate(
|
|
from(e in DeviceSchema,
|
|
where: e.organization_id == ^organization_id
|
|
),
|
|
:count
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Returns the count of devices for a site.
|
|
"""
|
|
@spec count_site_devices(String.t()) :: integer()
|
|
def count_site_devices(site_id) do
|
|
Repo.aggregate(
|
|
from(e in DeviceSchema, where: e.site_id == ^site_id),
|
|
:count
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Batch count devices for multiple sites.
|
|
|
|
Returns a map of %{site_id => %{total: count, down: count}}
|
|
"""
|
|
def batch_count_site_devices(site_ids) when is_list(site_ids) do
|
|
# Single query to get both total and down counts grouped by site
|
|
from(d in DeviceSchema,
|
|
where: d.site_id in ^site_ids,
|
|
group_by: d.site_id,
|
|
select: %{
|
|
site_id: d.site_id,
|
|
total: count(d.id),
|
|
down: filter(count(d.id), d.status == :down)
|
|
}
|
|
)
|
|
|> Repo.all()
|
|
|> Map.new(fn %{site_id: site_id, total: total, down: down} ->
|
|
{site_id, %{total: total, down: down}}
|
|
end)
|
|
end
|
|
|
|
@doc """
|
|
Returns the count of devices that is down for a site.
|
|
"""
|
|
def count_site_devices_down(site_id) do
|
|
Repo.aggregate(
|
|
from(e in DeviceSchema, where: e.site_id == ^site_id and e.status == :down),
|
|
:count
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Search devices by name or IP address for an organization.
|
|
Returns up to 10 results. Requires at least 2 characters.
|
|
"""
|
|
@spec search_devices(String.t(), String.t()) :: [DeviceSchema.t()]
|
|
def search_devices(organization_id, query) do
|
|
search_term = "%#{query}%"
|
|
|
|
DeviceSchema
|
|
|> where(organization_id: ^organization_id)
|
|
|> where([d], ilike(d.name, ^search_term) or ilike(d.ip_address, ^search_term))
|
|
|> order_by(:name)
|
|
|> limit(10)
|
|
|> Repo.all()
|
|
|> Repo.preload(:site)
|
|
end
|
|
|
|
@doc """
|
|
Returns a map of device status counts for an organization.
|
|
|
|
Uses a GROUP BY query instead of loading all devices into memory.
|
|
|
|
## Examples
|
|
|
|
iex> get_device_status_counts(organization_id)
|
|
%{up: 5, down: 2, unknown: 1}
|
|
|
|
iex> get_device_status_counts(empty_org_id)
|
|
%{}
|
|
"""
|
|
@spec get_device_status_counts(String.t()) :: %{atom() => non_neg_integer()}
|
|
def get_device_status_counts(organization_id) do
|
|
from(d in DeviceSchema,
|
|
where: d.organization_id == ^organization_id,
|
|
group_by: d.status,
|
|
select: {d.status, count(d.id)}
|
|
)
|
|
|> Repo.all()
|
|
|> Map.new()
|
|
end
|
|
|
|
@doc """
|
|
Returns the list of all devices with monitoring enabled.
|
|
"""
|
|
def list_monitored_devices do
|
|
Repo.all(
|
|
from(e in DeviceSchema,
|
|
where: e.monitoring_enabled == true,
|
|
order_by: [asc: e.name]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Returns the list of all devices with SNMP enabled.
|
|
"""
|
|
def list_snmp_enabled_devices do
|
|
Repo.all(
|
|
from(e in DeviceSchema,
|
|
where: e.snmp_enabled == true,
|
|
order_by: [asc: e.name]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Lists all devices with MikroTik API enabled and configured for backup.
|
|
Returns devices that have MikroTik API enabled, credentials, and an assigned agent.
|
|
"""
|
|
def list_mikrotik_devices_with_api do
|
|
# Get all MikroTik devices with API enabled and credentials configured
|
|
# Credentials are denormalized - already resolved on device during create/update
|
|
# Agent assignment is resolved hierarchically in backup_device/1 (includes global default)
|
|
Repo.all(
|
|
from(d in DeviceSchema,
|
|
left_join: s in assoc(d, :site),
|
|
left_join: o in assoc(d, :organization),
|
|
left_join: aa in assoc(d, :agent_assignments),
|
|
where: d.mikrotik_enabled == true,
|
|
where: not is_nil(d.mikrotik_username),
|
|
order_by: [asc: d.name],
|
|
preload: [site: s, organization: o, agent_assignments: aa],
|
|
distinct: d.id
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Resolves the agent token ID for a device using hierarchical resolution:
|
|
1. Direct agent assignment (via agent_assignments table)
|
|
2. Site-level agent assignment
|
|
3. Organization default agent
|
|
4. Global default cloud poller
|
|
|
|
Returns nil if no agent is assigned.
|
|
"""
|
|
def resolve_agent_token_id(device) do
|
|
# Ensure organization, site (if present), and agent_assignments are preloaded
|
|
# Note: site: :organization is needed for fallback agent token resolution
|
|
device =
|
|
if device.site_id do
|
|
Repo.preload(device, [:organization, [site: :organization], agent_assignments: []])
|
|
else
|
|
Repo.preload(device, [:organization, agent_assignments: []])
|
|
end
|
|
|
|
# Try each level in order
|
|
get_direct_agent(device) ||
|
|
get_site_agent(device) ||
|
|
get_org_default_agent(device) ||
|
|
get_global_default_agent()
|
|
end
|
|
|
|
defp get_direct_agent(device) do
|
|
case Enum.find(device.agent_assignments, & &1.enabled) do
|
|
%{agent_token_id: id} when not is_nil(id) -> id
|
|
_ -> nil
|
|
end
|
|
end
|
|
|
|
defp get_site_agent(device) do
|
|
# Safe: only access site if it exists
|
|
if device.site_id && device.site do
|
|
device.site.agent_token_id
|
|
end
|
|
end
|
|
|
|
defp get_org_default_agent(device) do
|
|
# Use direct organization relationship instead of site.organization
|
|
# Guard against NotLoaded to prevent crashes if association isn't preloaded
|
|
case device.organization do
|
|
%Ecto.Association.NotLoaded{} ->
|
|
require Logger
|
|
|
|
Logger.error(
|
|
"Organization not preloaded for device #{device.id} in get_org_default_agent",
|
|
device_id: device.id
|
|
)
|
|
|
|
nil
|
|
|
|
organization ->
|
|
organization.default_agent_token_id
|
|
end
|
|
end
|
|
|
|
defp get_global_default_agent, do: Towerops.Settings.get_global_default_cloud_poller()
|
|
|
|
@doc """
|
|
Gets a single device by ID, returns nil if not found.
|
|
"""
|
|
def get_device(id) do
|
|
DeviceSchema
|
|
|> Repo.get(id)
|
|
|> case do
|
|
nil -> nil
|
|
device -> Repo.preload(device, [:organization, site: :organization])
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Gets a single device by ID, raises if not found.
|
|
"""
|
|
def get_device!(id) do
|
|
DeviceSchema
|
|
|> Repo.get!(id)
|
|
|> Repo.preload([:organization, site: :organization])
|
|
end
|
|
|
|
@doc """
|
|
Gets device with full details including SNMP device, interfaces, and sensors.
|
|
"""
|
|
def get_device_with_details(id) do
|
|
DeviceSchema
|
|
|> Repo.get(id)
|
|
|> case do
|
|
nil ->
|
|
nil
|
|
|
|
device ->
|
|
Repo.preload(device, [:site, :organization, snmp_device: [:interfaces, :sensors]])
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Finds a device by IP address within a site, optionally excluding a specific device ID.
|
|
Returns nil if no device found.
|
|
"""
|
|
def get_device_by_ip(ip_address, site_id, exclude_id \\ nil)
|
|
|
|
def get_device_by_ip(_ip_address, nil, _exclude_id), do: nil
|
|
|
|
def get_device_by_ip(ip_address, site_id, exclude_id) do
|
|
query =
|
|
from(d in DeviceSchema,
|
|
where: d.ip_address == ^ip_address and d.site_id == ^site_id,
|
|
preload: [:site]
|
|
)
|
|
|
|
query =
|
|
if exclude_id do
|
|
from(d in query, where: d.id != ^exclude_id)
|
|
else
|
|
query
|
|
end
|
|
|
|
Repo.one(query)
|
|
end
|
|
|
|
@doc """
|
|
Gets a single device belonging to a specific site.
|
|
"""
|
|
def get_site_device!(site_id, device_id) do
|
|
Repo.one!(from(e in DeviceSchema, where: e.id == ^device_id, where: e.site_id == ^site_id, preload: [:site]))
|
|
end
|
|
|
|
@doc """
|
|
Gets SNMP configuration for device with hierarchical fallback.
|
|
|
|
Falls back in this order:
|
|
1. Device-level configuration
|
|
2. Site-level configuration
|
|
3. Organization-level configuration
|
|
|
|
Returns a map with:
|
|
- version: SNMP version ("1", "2c", or "3")
|
|
- community: SNMP community string
|
|
- source: Where the config came from (:device, :site, or :organization)
|
|
"""
|
|
def get_snmp_config(device_id) when is_binary(device_id) do
|
|
device = Repo.get!(DeviceSchema, device_id)
|
|
get_snmp_config(device)
|
|
end
|
|
|
|
def get_snmp_config(%DeviceSchema{} = device) do
|
|
# Credentials are already denormalized on the device - just read them!
|
|
%{
|
|
version: device.snmp_version || "2c",
|
|
community: device.snmp_community,
|
|
port: device.snmp_port || 161,
|
|
transport: device.snmp_transport || "udp",
|
|
source: safe_source_to_atom(device.snmp_community_source)
|
|
}
|
|
end
|
|
|
|
# Safely convert credential source to atom with whitelist to prevent atom exhaustion
|
|
defp safe_source_to_atom("device"), do: :device
|
|
defp safe_source_to_atom("site"), do: :site
|
|
defp safe_source_to_atom("organization"), do: :organization
|
|
defp safe_source_to_atom(nil), do: :organization
|
|
defp safe_source_to_atom(_unknown), do: :organization
|
|
|
|
@doc """
|
|
Gets MikroTik API configuration for a device with hierarchical fallback.
|
|
|
|
Returns a map with username, password (decrypted), port, use_ssl, enabled, and source.
|
|
|
|
## Examples
|
|
|
|
iex> get_mikrotik_config(device)
|
|
%{
|
|
username: "admin",
|
|
password: "secret",
|
|
port: 8729,
|
|
use_ssl: true,
|
|
enabled: true,
|
|
source: :device
|
|
}
|
|
"""
|
|
def get_mikrotik_config(device_id) when is_binary(device_id) do
|
|
device = Repo.get!(DeviceSchema, device_id)
|
|
get_mikrotik_config(device)
|
|
end
|
|
|
|
def get_mikrotik_config(%DeviceSchema{} = device) do
|
|
resolve_mikrotik_config(device)
|
|
end
|
|
|
|
defp resolve_mikrotik_config(device) do
|
|
# Credentials are already denormalized on the device - just read them!
|
|
%{
|
|
username: device.mikrotik_username,
|
|
password: device.mikrotik_password,
|
|
port: device.mikrotik_port || 8729,
|
|
ssh_port: device.mikrotik_ssh_port || 22,
|
|
use_ssl: if(device.mikrotik_use_ssl == nil, do: true, else: device.mikrotik_use_ssl),
|
|
enabled: device.mikrotik_enabled || false,
|
|
source: credential_source_atom(device.mikrotik_credential_source)
|
|
}
|
|
end
|
|
|
|
@doc """
|
|
Propagates MikroTik credential changes from a site to all devices in that
|
|
site that inherit credentials from the site (credential_source = "site").
|
|
|
|
Updates username, password, port, use_ssl, and enabled fields.
|
|
"""
|
|
def propagate_site_mikrotik_change(site_id, attrs) do
|
|
# Find all devices in this site that inherit credentials from site
|
|
device_query =
|
|
from(d in DeviceSchema,
|
|
where: d.site_id == ^site_id,
|
|
where: d.mikrotik_credential_source == "site"
|
|
)
|
|
|
|
device_ids = Repo.all(from(d in device_query, select: d.id))
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
updates = attrs |> Map.new() |> Map.put(:updated_at, now) |> Map.to_list()
|
|
|
|
Repo.update_all(device_query, set: updates)
|
|
|
|
Enum.each(device_ids, fn device_id ->
|
|
# Broadcast assignment change to trigger agent job list refresh
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"device:#{device_id}:assignments",
|
|
{:assignments_changed, :mikrotik_updated}
|
|
)
|
|
end)
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc """
|
|
Propagates MikroTik credential changes from an organization to all devices
|
|
in sites that have no site-level credentials (credential_source = "site"
|
|
and site.mikrotik_username is nil).
|
|
|
|
Updates username, password, port, use_ssl, and enabled fields.
|
|
"""
|
|
def propagate_organization_mikrotik_change(organization_id, attrs) do
|
|
# With denormalized credentials, update all devices that inherit from the organization
|
|
# (devices with mikrotik_credential_source = "organization")
|
|
device_query =
|
|
from(d in DeviceSchema,
|
|
where: d.organization_id == ^organization_id,
|
|
where: d.mikrotik_credential_source == "organization"
|
|
)
|
|
|
|
device_ids = Repo.all(from(d in device_query, select: d.id))
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
updates = attrs |> Map.new() |> Map.put(:updated_at, now) |> Map.to_list()
|
|
|
|
Repo.update_all(device_query, set: updates)
|
|
|
|
Enum.each(device_ids, fn device_id ->
|
|
# Broadcast assignment change to trigger agent job list refresh
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"device:#{device_id}:assignments",
|
|
{:assignments_changed, :mikrotik_updated}
|
|
)
|
|
end)
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc """
|
|
Gets SNMPv3 configuration for a device with hierarchical fallback.
|
|
|
|
Returns a map with all v3 fields resolved, or nil if v2c/v1 is in use.
|
|
|
|
## Examples
|
|
|
|
iex> get_snmpv3_config(device)
|
|
%{
|
|
security_level: "authPriv",
|
|
username: "snmpuser",
|
|
auth_protocol: "SHA-256",
|
|
auth_password: "decrypted_auth_pass",
|
|
priv_protocol: "AES",
|
|
priv_password: "decrypted_priv_pass",
|
|
source: :device
|
|
}
|
|
"""
|
|
def get_snmpv3_config(device_id) when is_binary(device_id) do
|
|
device = Repo.get!(DeviceSchema, device_id)
|
|
get_snmpv3_config(device)
|
|
end
|
|
|
|
def get_snmpv3_config(%DeviceSchema{snmp_version: version} = _device) when version != "3" do
|
|
nil
|
|
end
|
|
|
|
def get_snmpv3_config(%DeviceSchema{} = device) do
|
|
# Credentials are already denormalized on the device - just read them!
|
|
%{
|
|
security_level: device.snmpv3_security_level,
|
|
username: device.snmpv3_username,
|
|
auth_protocol: device.snmpv3_auth_protocol || "SHA-256",
|
|
auth_password: device.snmpv3_auth_password,
|
|
priv_protocol: device.snmpv3_priv_protocol || "AES",
|
|
priv_password: device.snmpv3_priv_password,
|
|
source: credential_source_atom(device.snmpv3_credential_source)
|
|
}
|
|
end
|
|
|
|
@doc """
|
|
Propagates SNMPv3 credential changes from a site to all devices in that
|
|
site that inherit credentials from the site (credential_source = "site").
|
|
"""
|
|
def propagate_site_snmpv3_change(site_id, attrs) do
|
|
device_query =
|
|
from(d in DeviceSchema,
|
|
where: d.site_id == ^site_id,
|
|
where: d.snmpv3_credential_source == "site"
|
|
)
|
|
|
|
device_ids = Repo.all(from(d in device_query, select: d.id))
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
updates = attrs |> Map.new() |> Map.put(:updated_at, now) |> Map.to_list()
|
|
|
|
Repo.update_all(device_query, set: updates)
|
|
|
|
Enum.each(device_ids, fn device_id ->
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"device:#{device_id}:assignments",
|
|
{:assignments_changed, :snmpv3_updated}
|
|
)
|
|
end)
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc """
|
|
Propagates SNMPv3 credential changes from an organization to all devices
|
|
in sites that have no site-level credentials.
|
|
"""
|
|
def propagate_organization_snmpv3_change(organization_id, attrs) do
|
|
# With denormalized credentials, update all devices that inherit from the organization
|
|
# (devices with snmpv3_credential_source = "organization")
|
|
device_query =
|
|
from(d in DeviceSchema,
|
|
where: d.organization_id == ^organization_id,
|
|
where: d.snmpv3_credential_source == "organization"
|
|
)
|
|
|
|
device_ids = Repo.all(from(d in device_query, select: d.id))
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
updates = attrs |> Map.new() |> Map.put(:updated_at, now) |> Map.to_list()
|
|
|
|
Repo.update_all(device_query, set: updates)
|
|
|
|
Enum.each(device_ids, fn device_id ->
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"device:#{device_id}:assignments",
|
|
{:assignments_changed, :snmpv3_updated}
|
|
)
|
|
end)
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc """
|
|
Creates device.
|
|
|
|
## Options
|
|
* `:bypass_limits` - When true, bypasses subscription device limits (for superusers)
|
|
"""
|
|
def create_device(attrs, opts \\ []) do
|
|
bypass_limits = Keyword.get(opts, :bypass_limits, false)
|
|
|
|
# Check device quota before insertion (unless bypassing)
|
|
with {:ok, changeset} <- check_device_quota(attrs, bypass_limits),
|
|
{:ok, device} <- Repo.insert(changeset) do
|
|
# Start monitoring/polling if enabled
|
|
_ =
|
|
if device.monitoring_enabled do
|
|
DeviceMonitorWorker.start_monitoring(device.id)
|
|
end
|
|
|
|
# Note: Checks will be created and scheduled when discovery runs
|
|
# No need to start polling here
|
|
|
|
broadcast_device_change(device.organization_id, :device_created)
|
|
|
|
{:ok, device}
|
|
end
|
|
end
|
|
|
|
defp check_device_quota(attrs, bypass_limits) do
|
|
changeset = DeviceSchema.changeset(%DeviceSchema{}, attrs)
|
|
|
|
# Apply credential resolution
|
|
changeset = apply_credential_resolution(changeset)
|
|
|
|
if bypass_limits do
|
|
{:ok, changeset}
|
|
else
|
|
do_check_quota(changeset)
|
|
end
|
|
end
|
|
|
|
defp apply_credential_resolution(changeset) do
|
|
# Extract organization_id and site_id from changeset
|
|
organization_id = Ecto.Changeset.get_field(changeset, :organization_id)
|
|
site_id = Ecto.Changeset.get_field(changeset, :site_id)
|
|
|
|
if organization_id do
|
|
# Load organization and site (if present)
|
|
organization = Organizations.get_organization!(organization_id)
|
|
site = if site_id, do: Sites.get_site!(site_id)
|
|
|
|
# Apply credential resolution
|
|
changeset
|
|
|> CredentialResolver.resolve_snmp_credentials(organization, site)
|
|
|> CredentialResolver.resolve_snmpv3_credentials(organization, site)
|
|
|> CredentialResolver.resolve_mikrotik_credentials(organization, site)
|
|
else
|
|
# No organization_id yet, skip credential resolution
|
|
changeset
|
|
end
|
|
end
|
|
|
|
defp do_check_quota(changeset) do
|
|
case Ecto.Changeset.fetch_change(changeset, :organization_id) do
|
|
{:ok, organization_id} ->
|
|
validate_device_quota(changeset, organization_id)
|
|
|
|
:error ->
|
|
# No organization_id in changeset, let normal validation handle it
|
|
{:ok, changeset}
|
|
end
|
|
end
|
|
|
|
defp validate_device_quota(changeset, organization_id) do
|
|
organization = Organizations.get_organization!(organization_id)
|
|
|
|
case SubscriptionLimits.check_device_limit(organization) do
|
|
{:ok, :within_limit} ->
|
|
{:ok, changeset}
|
|
|
|
{:error, :at_limit, _current, max} ->
|
|
error_changeset =
|
|
Ecto.Changeset.add_error(
|
|
changeset,
|
|
:base,
|
|
"You've reached your plan limit of #{max} devices. Upgrade to add more."
|
|
)
|
|
|
|
{:error, error_changeset}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Updates device.
|
|
"""
|
|
def update_device(%DeviceSchema{} = device, attrs) do
|
|
old_monitoring = device.monitoring_enabled
|
|
old_snmp = device.snmp_enabled
|
|
old_snmp_version = device.snmp_version
|
|
old_snmp_port = device.snmp_port
|
|
|
|
changeset = DeviceSchema.changeset(device, attrs)
|
|
|
|
# Apply credential resolution
|
|
changeset = apply_credential_resolution(changeset)
|
|
|
|
# Validate MikroTik SSL requirement for cloud pollers
|
|
changeset = validate_mikrotik_ssl_with_cloud_poller(changeset, device)
|
|
|
|
case Repo.update(changeset) do
|
|
{:ok, updated_device} = result ->
|
|
_ = handle_monitoring_changes(updated_device, old_monitoring)
|
|
_ = handle_snmp_changes(updated_device, old_snmp, old_snmp_version, old_snmp_port)
|
|
broadcast_device_change(updated_device.organization_id, :device_updated)
|
|
result
|
|
|
|
error ->
|
|
error
|
|
end
|
|
end
|
|
|
|
defp validate_mikrotik_ssl_with_cloud_poller(changeset, device) do
|
|
# Check if MikroTik is enabled and SSL is being disabled
|
|
mikrotik_enabled = Ecto.Changeset.get_field(changeset, :mikrotik_enabled)
|
|
use_ssl_change = Ecto.Changeset.get_change(changeset, :mikrotik_use_ssl)
|
|
|
|
# Only validate if MikroTik is enabled and SSL is explicitly being set to false
|
|
if mikrotik_enabled && use_ssl_change == false do
|
|
validate_cloud_poller_requires_ssl(changeset, device)
|
|
else
|
|
changeset
|
|
end
|
|
end
|
|
|
|
defp validate_cloud_poller_requires_ssl(changeset, device) do
|
|
# Get effective agent to check if it's a cloud poller
|
|
device_with_site = Repo.preload(device, [:organization, site: :organization])
|
|
{effective_agent_id, _source} = Agents.get_effective_agent_token_with_source(device_with_site)
|
|
|
|
using_cloud_poller =
|
|
if effective_agent_id do
|
|
agent = Agents.get_agent_token!(effective_agent_id)
|
|
agent.is_cloud_poller
|
|
else
|
|
# No agent means cloud polling by default
|
|
true
|
|
end
|
|
|
|
if using_cloud_poller do
|
|
Ecto.Changeset.add_error(
|
|
changeset,
|
|
:mikrotik_use_ssl,
|
|
"SSL is required when using cloud pollers. Plain MikroTik API sends credentials unencrypted over the internet."
|
|
)
|
|
else
|
|
changeset
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Resolves the effective SNMP community string for a device.
|
|
|
|
Resolution order:
|
|
1. If device has its own community string (source = "device" AND community is non-empty), use it
|
|
2. Otherwise, inherit from site's community string
|
|
3. If site has no community, inherit from organization's community string
|
|
4. If no community found anywhere, return nil
|
|
|
|
Note: Even if source is "device", we fall back to inheritance if the device community is empty.
|
|
This handles cases where the source field is incorrect or the community was cleared.
|
|
"""
|
|
def resolve_snmp_community(%DeviceSchema{} = device) do
|
|
device = Repo.preload(device, [:site, :organization, site: :organization])
|
|
|
|
community =
|
|
get_device_community(device) ||
|
|
get_site_community(device) ||
|
|
get_organization_community(device)
|
|
|
|
log_empty_community(device, community)
|
|
community
|
|
end
|
|
|
|
# Get device-specific community if source is "device" and value is present
|
|
defp get_device_community(device) do
|
|
if device.snmp_community_source == "device" && present?(device.snmp_community) do
|
|
device.snmp_community
|
|
end
|
|
end
|
|
|
|
# Get community from site (handles nil site)
|
|
defp get_site_community(%{site: nil}), do: nil
|
|
defp get_site_community(%{site: site}), do: site.snmp_community
|
|
|
|
# Get community from organization
|
|
defp get_organization_community(%{organization: organization}) do
|
|
organization.snmp_community
|
|
end
|
|
|
|
# Log warning if community is empty
|
|
defp log_empty_community(device, community) when community in [nil, ""] do
|
|
require Logger
|
|
|
|
Logger.warning("Empty SNMP community resolved for device",
|
|
device_id: device.id,
|
|
device_name: device.name,
|
|
community_source: device.snmp_community_source,
|
|
device_community: device.snmp_community,
|
|
site_community: get_site_community(device),
|
|
org_community: device.organization.snmp_community,
|
|
site_id: device.site_id,
|
|
org_id: device.organization_id
|
|
)
|
|
end
|
|
|
|
defp log_empty_community(_device, _community), do: :ok
|
|
|
|
# Helper to check if a string is present (not nil and not empty)
|
|
defp present?(value) when is_binary(value), do: String.trim(value) != ""
|
|
defp present?(_), do: false
|
|
|
|
defp credential_source_atom(value) do
|
|
case value do
|
|
"device" -> :device
|
|
"site" -> :site
|
|
"organization" -> :organization
|
|
_ -> :organization
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Propagates SNMP community string changes from a site to all devices
|
|
that inherit from it (source = "site").
|
|
|
|
This updates the actual snmp_community value on devices so they have
|
|
the resolved value cached and ready for agent communication.
|
|
"""
|
|
def propagate_site_community_change(site_id, new_community) do
|
|
# Find all devices in this site that inherit community from site
|
|
device_query =
|
|
from(d in DeviceSchema, where: d.site_id == ^site_id, where: d.snmp_community_source == "site")
|
|
|
|
device_ids = Repo.all(from(d in device_query, select: d.id))
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
|
|
Repo.update_all(device_query, set: [snmp_community: new_community, updated_at: now])
|
|
|
|
Enum.each(device_ids, fn device_id ->
|
|
# Broadcast assignment change to trigger agent job list refresh
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"device:#{device_id}:assignments",
|
|
{:assignments_changed, :community_updated}
|
|
)
|
|
end)
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc """
|
|
Propagates SNMP community string changes from an organization to all
|
|
devices in sites that have no site-level community (source = "site"
|
|
and site.snmp_community is nil).
|
|
"""
|
|
def propagate_organization_community_change(organization_id, new_community) do
|
|
# With denormalized credentials, update all devices that inherit from the organization
|
|
# (devices with snmp_community_source = "organization")
|
|
device_query =
|
|
from(d in DeviceSchema,
|
|
where: d.organization_id == ^organization_id,
|
|
where: d.snmp_community_source == "organization"
|
|
)
|
|
|
|
device_ids = Repo.all(from(d in device_query, select: d.id))
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
|
|
Repo.update_all(device_query, set: [snmp_community: new_community, updated_at: now])
|
|
|
|
Enum.each(device_ids, fn device_id ->
|
|
# Broadcast assignment change to trigger agent job list refresh
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"device:#{device_id}:assignments",
|
|
{:assignments_changed, :community_updated}
|
|
)
|
|
end)
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc """
|
|
Deletes device.
|
|
"""
|
|
def delete_device(%DeviceSchema{} = device) do
|
|
organization_id = device.organization_id
|
|
|
|
# Stop monitoring and check jobs before deleting
|
|
_ = DeviceMonitorWorker.stop_monitoring(device.id)
|
|
_ = Monitoring.stop_device_checks(device.id)
|
|
|
|
# Wait briefly for any in-flight jobs to complete or detect deletion
|
|
# This reduces (but doesn't eliminate) the window for orphaned writes
|
|
# In-flight jobs check device existence via verify_polling_assignment_unchanged
|
|
# Configurable for faster tests (defaults to 500ms in prod, 10ms in test)
|
|
deletion_delay_ms = Application.get_env(:towerops, :device_deletion_delay_ms, 500)
|
|
Process.sleep(deletion_delay_ms)
|
|
|
|
# Get agent assignment before deleting (if any)
|
|
assignment = Agents.get_device_assignment(device.id)
|
|
agent_token_id = if assignment, do: assignment.agent_token_id
|
|
|
|
# Delete the device (cascades to SNMP device, sensors, interfaces, etc.)
|
|
result = Repo.delete(device)
|
|
|
|
case result do
|
|
{:ok, _} ->
|
|
# Notify agent to stop processing jobs for this device
|
|
if agent_token_id do
|
|
Agents.broadcast_assignment_change(agent_token_id, :device_deleted)
|
|
end
|
|
|
|
broadcast_device_change(organization_id, :device_deleted)
|
|
|
|
_ ->
|
|
# Notify agent even on failure (preserves original behavior)
|
|
if agent_token_id do
|
|
Agents.broadcast_assignment_change(agent_token_id, :device_deleted)
|
|
end
|
|
end
|
|
|
|
result
|
|
end
|
|
|
|
@doc """
|
|
Returns an `%Ecto.Changeset{}` for tracking device changes.
|
|
"""
|
|
def change_device(%DeviceSchema{} = device, attrs \\ %{}) do
|
|
DeviceSchema.changeset(device, attrs)
|
|
end
|
|
|
|
@doc """
|
|
Updates the status of device.
|
|
"""
|
|
def update_device_status(%DeviceSchema{} = device, new_status) do
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
status_changed = device.status != new_status
|
|
|
|
changes = %{
|
|
status: new_status,
|
|
last_checked_at: now
|
|
}
|
|
|
|
changes =
|
|
if status_changed do
|
|
Map.put(changes, :last_status_change_at, now)
|
|
else
|
|
changes
|
|
end
|
|
|
|
case device |> Ecto.Changeset.change(changes) |> Repo.update() do
|
|
{:ok, updated_device} = result ->
|
|
if status_changed do
|
|
broadcast_device_change(updated_device.organization_id, :device_status_changed)
|
|
else
|
|
broadcast_device_change(updated_device.organization_id, :device_updated)
|
|
end
|
|
|
|
result
|
|
|
|
error ->
|
|
error
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Updates the last SNMP poll timestamp for device.
|
|
Used for distributed coordination to prevent duplicate polling across pods.
|
|
Also auto-dismisses any active "device_poll_gap" insights when polling resumes.
|
|
"""
|
|
def update_snmp_poll_time(%DeviceSchema{} = device) do
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
|
|
result =
|
|
device
|
|
|> Ecto.Changeset.change(%{last_snmp_poll_at: now})
|
|
|> Repo.update()
|
|
|
|
# Auto-dismiss poll gap insights when polling resumes
|
|
case result do
|
|
{:ok, _updated_device} ->
|
|
dismiss_poll_gap_insights(device.id)
|
|
result
|
|
|
|
error ->
|
|
error
|
|
end
|
|
end
|
|
|
|
defp dismiss_poll_gap_insights(device_id) do
|
|
alias Towerops.Preseem.Insight
|
|
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
|
|
{count, _} =
|
|
Insight
|
|
|> where(device_id: ^device_id, type: "device_poll_gap", status: "active")
|
|
|> Repo.update_all(set: [status: "dismissed", dismissed_at: now])
|
|
|
|
if count > 0 do
|
|
require Logger
|
|
|
|
Logger.info("Auto-dismissed #{count} poll gap insight(s) for device #{device_id}")
|
|
end
|
|
|
|
:ok
|
|
end
|
|
|
|
# Private helpers for monitoring/polling management
|
|
|
|
defp handle_monitoring_changes(device, old_monitoring) do
|
|
cond do
|
|
device.monitoring_enabled && !old_monitoring ->
|
|
_ = DeviceMonitorWorker.start_monitoring(device.id)
|
|
# Trigger discovery when monitoring is enabled (if SNMP is also enabled)
|
|
if device.snmp_enabled, do: DiscoveryWorker.enqueue(device.id)
|
|
|
|
!device.monitoring_enabled && old_monitoring ->
|
|
_ = DeviceMonitorWorker.stop_monitoring(device.id)
|
|
|
|
true ->
|
|
:ok
|
|
end
|
|
end
|
|
|
|
defp handle_snmp_changes(device, old_snmp, old_snmp_version, old_snmp_port) do
|
|
should_discover = should_trigger_discovery?(device, old_snmp, old_snmp_version, old_snmp_port)
|
|
|
|
cond do
|
|
device.snmp_enabled && !old_snmp ->
|
|
# SNMP newly enabled - discovery will create and schedule checks
|
|
if should_discover, do: DiscoveryWorker.enqueue(device.id)
|
|
|
|
!device.snmp_enabled && old_snmp ->
|
|
# SNMP disabled - disable all checks and cancel jobs
|
|
_ = Monitoring.disable_device_checks(device.id)
|
|
|
|
should_discover ->
|
|
# SNMP settings changed - re-run discovery
|
|
_ = DiscoveryWorker.enqueue(device.id)
|
|
|
|
true ->
|
|
:ok
|
|
end
|
|
end
|
|
|
|
defp should_trigger_discovery?(device, old_snmp, old_snmp_version, old_snmp_port) do
|
|
cond do
|
|
# SNMP was just enabled
|
|
device.snmp_enabled && !old_snmp ->
|
|
true
|
|
|
|
# SNMP version changed while enabled
|
|
device.snmp_enabled && device.snmp_version != old_snmp_version ->
|
|
true
|
|
|
|
# SNMP port changed while enabled
|
|
device.snmp_enabled && device.snmp_port != old_snmp_port ->
|
|
true
|
|
|
|
# Otherwise no discovery needed
|
|
true ->
|
|
false
|
|
end
|
|
end
|
|
|
|
## Events
|
|
|
|
@doc """
|
|
Creates an device event.
|
|
"""
|
|
def create_event(attrs) do
|
|
%Event{}
|
|
|> Event.changeset(attrs)
|
|
|> Repo.insert()
|
|
end
|
|
|
|
@doc """
|
|
Returns the list of events for device, ordered by most recent first.
|
|
"""
|
|
def list_devices_events(device_id, limit \\ 100) do
|
|
Repo.all(
|
|
from(e in Event,
|
|
where: e.device_id == ^device_id,
|
|
order_by: [desc: e.occurred_at],
|
|
limit: ^limit
|
|
)
|
|
)
|
|
end
|
|
|
|
## Display Order Management
|
|
|
|
@doc """
|
|
Reorders a device to a new position within its site.
|
|
|
|
Updates display_order for all devices in the site to maintain continuous ordering (1, 2, 3, ...).
|
|
The new_position is 1-based (1 = first position).
|
|
|
|
Returns {:ok, device} on success, {:error, changeset} on failure.
|
|
"""
|
|
def reorder_device(device_id, new_position) when is_integer(new_position) and new_position > 0 do
|
|
Repo.transaction(fn ->
|
|
device = Repo.get!(DeviceSchema, device_id)
|
|
site_id = device.site_id
|
|
|
|
# Get all devices in site excluding the one being moved, in current order
|
|
other_devices =
|
|
Repo.all(
|
|
from(d in DeviceSchema,
|
|
where: d.site_id == ^site_id and d.id != ^device_id,
|
|
order_by: [asc: d.display_order, asc: d.name]
|
|
)
|
|
)
|
|
|
|
# Insert at new position (1-based index, convert to 0-based for List.insert_at)
|
|
new_order = List.insert_at(other_devices, new_position - 1, device)
|
|
|
|
{device_ids, orders} =
|
|
new_order
|
|
|> Enum.with_index(1)
|
|
|> Enum.map(fn {d, order} -> {d.id, order} end)
|
|
|> Enum.unzip()
|
|
|
|
device_ids = Enum.map(device_ids, &Ecto.UUID.dump!/1)
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
|
|
query =
|
|
from(d in DeviceSchema,
|
|
join:
|
|
data in fragment(
|
|
"SELECT * FROM unnest(?::uuid[], ?::int[]) AS data(id, ordering)",
|
|
^device_ids,
|
|
^orders
|
|
),
|
|
on: d.id == data.id,
|
|
update: [set: [display_order: field(data, :ordering), updated_at: ^now]]
|
|
)
|
|
|
|
Repo.update_all(query, [])
|
|
|
|
Repo.get!(DeviceSchema, device_id)
|
|
end)
|
|
end
|
|
|
|
@doc """
|
|
Reset all devices in organization to alphabetical order.
|
|
|
|
Clears display_order field for all devices, allowing them to fall back to alphabetical sorting.
|
|
"""
|
|
def reset_organization_device_order(organization_id) do
|
|
Repo.update_all(
|
|
from(d in DeviceSchema,
|
|
join: s in assoc(d, :site),
|
|
where: s.organization_id == ^organization_id
|
|
),
|
|
set: [display_order: nil, updated_at: DateTime.truncate(DateTime.utc_now(), :second)]
|
|
)
|
|
end
|
|
|
|
defp broadcast_device_change(organization_id, event) do
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"devices:org:#{organization_id}",
|
|
{event, organization_id}
|
|
)
|
|
end
|
|
end
|