diff --git a/lib/towerops/snmp/arp_entries.ex b/lib/towerops/snmp/arp_entries.ex new file mode 100644 index 00000000..85b29a90 --- /dev/null +++ b/lib/towerops/snmp/arp_entries.ex @@ -0,0 +1,142 @@ +defmodule Towerops.Snmp.ArpEntries do + @moduledoc """ + Read-side queries, upserts, and stale cleanup for ARP entries discovered + on SNMP devices. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Repo + alias Towerops.Snmp.ArpEntry + + require Logger + + @doc """ + Lists all ARP entries for a device. + """ + def list_arp_entries(device_id) do + ArpEntry + |> where([a], a.device_id == ^device_id) + |> order_by([a], a.ip_address) + |> preload(:interface) + |> Repo.all() + end + + @doc """ + Gets a specific ARP entry. + """ + def get_arp_entry(arp_entry_id) do + Repo.get(ArpEntry, arp_entry_id) + end + + @doc """ + Creates or updates an ARP entry. + + Uses atomic PostgreSQL ON CONFLICT to prevent race conditions when multiple + pollers try to upsert the same ARP entry simultaneously. + """ + def upsert_arp_entry(attrs) do + %ArpEntry{} + |> ArpEntry.changeset(attrs) + |> Repo.insert( + on_conflict: {:replace_all_except, [:id, :inserted_at]}, + conflict_target: [:device_id, :ip_address, :mac_address], + returning: true + ) + end + + @doc """ + Deletes stale ARP entries that haven't been seen since the given datetime. + """ + def delete_stale_arp_entries(device_id, cutoff_datetime) do + ArpEntry + |> where([a], a.device_id == ^device_id) + |> where([a], a.last_seen_at < ^cutoff_datetime) + |> Repo.delete_all() + end + + @doc """ + Atomically deletes stale ARP entries and upserts new ones in a single transaction. + Prevents race conditions where concurrent delete/upsert could remove fresh data. + """ + def delete_stale_and_upsert_arp_entries(device_id, arp_entries, interfaces, cutoff) do + Repo.transaction(fn -> + delete_stale_arp_entries(device_id, cutoff) + upsert_arp_entries(device_id, arp_entries, interfaces) + end) + end + + @doc """ + Bulk upserts ARP entries for a device. + Returns `{success_count, error_count}` tuple. + """ + def upsert_arp_entries(device_id, arp_entries, interfaces) do + interface_map = Map.new(interfaces, fn i -> {i.if_index, i.id} end) + + arp_entries + |> Enum.reduce({0, 0, []}, fn entry, {s, e, error_samples} -> + interface_id = Map.get(interface_map, entry.if_index) + + attrs = + entry + |> Map.put(:device_id, device_id) + |> Map.put(:interface_id, interface_id) + + case upsert_arp_entry(attrs) do + {:ok, _} -> + {s + 1, e, error_samples} + + {:error, changeset} -> + # Sample first 3 errors for debugging + sample = maybe_add_error_sample(e, error_samples, attrs, changeset) + {s, e + 1, sample} + end + end) + |> tap(fn {_s, errors, error_samples} -> + if errors > 0 do + samples_text = error_samples |> Enum.reverse() |> format_error_samples() + + Logger.warning( + "#{errors} ARP upsert failures for device #{device_id}#{samples_text}", + device_id: device_id, + error_count: errors + ) + end + end) + |> then(fn {s, e, _samples} -> {s, e} end) + end + + # Helper to format changeset errors for logging + defp format_changeset_errors(changeset) do + Ecto.Changeset.traverse_errors(changeset, fn {msg, _opts} -> msg end) + end + + # Helper to collect error samples (first N errors for debugging) + defp maybe_add_error_sample(error_count, samples, attrs, changeset, max_samples \\ 3) do + if error_count < max_samples do + error_details = %{ + ip: Map.get(attrs, :ip_address), + mac: Map.get(attrs, :mac_address), + errors: format_changeset_errors(changeset) + } + + [error_details | samples] + else + samples + end + end + + # Helper to format error samples into log message text + defp format_error_samples([]), do: "" + + defp format_error_samples(samples) do + formatted_samples = + Enum.map_join(samples, "\n", fn sample -> + " - IP: #{sample.ip}, MAC: #{sample.mac}, errors: #{inspect(sample.errors)}" + end) + + "\nSample errors:\n" <> formatted_samples + end +end diff --git a/lib/towerops/snmp/discovered_devices.ex b/lib/towerops/snmp/discovered_devices.ex new file mode 100644 index 00000000..cf5a9770 --- /dev/null +++ b/lib/towerops/snmp/discovered_devices.ex @@ -0,0 +1,689 @@ +defmodule Towerops.Snmp.DiscoveredDevices do + @moduledoc """ + Aggregates LLDP/CDP neighbor, ARP, and MAC-table data discovered by SNMP + pollers across an organization to surface devices that have been seen by + monitored devices but not yet added to inventory. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Devices.Device, as: DeviceSchema + alias Towerops.Devices.DeviceQuery + alias Towerops.Repo + alias Towerops.Snmp.ArpEntry + alias Towerops.Snmp.Device + alias Towerops.Snmp.Interface + alias Towerops.Snmp.MacAddress + alias Towerops.Snmp.Neighbor + + @doc """ + Lists all discovered devices across an organization that haven't been added yet. + + Aggregates data from LLDP/CDP neighbors, ARP entries, and MAC addresses. + Returns a list of discovered device structs with merged data. + + ## Returns + + List of maps with: + - `identifier`: %{type: :mac | :ip | :hostname, value: string} + - `hostname`: string | nil + - `ip_addresses`: [string] + - `mac_addresses`: [string] + - `device_type`: :router | :switch | :wireless | :server | :workstation | :unknown + - `capabilities`: [string] (from LLDP/CDP) + - `platform`: string | nil + - `discovered_by`: [%{device_id: uuid, device_name: string, interface: string}] + - `source_count`: integer + - `last_seen`: DateTime.t() + - `protocols_used`: [:lldp | :cdp | :arp | :mac] + """ + def list_discovered_devices_for_organization(organization_id) do + device_ids = query_organization_device_ids(organization_id) + + neighbors = query_neighbors_for_devices(device_ids) + arp_entries = query_arp_for_devices(device_ids) + mac_addresses = query_mac_for_devices(device_ids) + + existing_ips = get_existing_device_ips(organization_id) + existing_macs = get_existing_interface_macs(organization_id) + + discovered_map = + %{} + |> merge_neighbors_into_map(neighbors) + |> merge_arp_into_map(arp_entries) + |> merge_mac_into_map(mac_addresses) + + discovered_map + |> Enum.reject(fn {_key, entry} -> + device_already_exists?(entry, existing_ips, existing_macs) + end) + |> Enum.map(fn {_key, entry} -> build_discovered_device_struct(entry) end) + |> Enum.sort(fn a, b -> + cond do + a.source_count > b.source_count -> true + a.source_count < b.source_count -> false + true -> DateTime.after?(a.last_seen, b.last_seen) + end + end) + end + + defp query_organization_device_ids(organization_id) do + organization_id + |> DeviceQuery.for_organization() + |> select([d], d.id) + |> Repo.all() + end + + defp query_neighbors_for_devices(device_ids) do + cutoff = DateTime.add(DateTime.utc_now(), -7, :day) + + Repo.all( + from(n in Neighbor, + join: d in DeviceSchema, + on: n.device_id == d.id, + left_join: i in Interface, + on: n.interface_id == i.id, + where: n.device_id in ^device_ids, + where: n.last_discovered_at > ^cutoff, + preload: [device: d, interface: i], + order_by: [desc: n.last_discovered_at] + ) + ) + end + + defp query_arp_for_devices(device_ids) do + cutoff = DateTime.add(DateTime.utc_now(), -7, :day) + + Repo.all( + from(a in ArpEntry, + join: d in DeviceSchema, + on: a.device_id == d.id, + left_join: i in Interface, + on: a.interface_id == i.id, + where: a.device_id in ^device_ids, + where: a.last_seen_at > ^cutoff, + preload: [device: d, interface: i], + order_by: [desc: a.last_seen_at] + ) + ) + end + + defp query_mac_for_devices(device_ids) do + cutoff = DateTime.add(DateTime.utc_now(), -7, :day) + + Repo.all( + from(m in MacAddress, + join: d in DeviceSchema, + on: m.device_id == d.id, + left_join: i in Interface, + on: m.interface_id == i.id, + where: m.device_id in ^device_ids, + where: m.last_seen_at > ^cutoff, + preload: [device: d, interface: i], + order_by: [desc: m.last_seen_at] + ) + ) + end + + defp get_existing_device_ips(organization_id) do + organization_id + |> DeviceQuery.for_organization() + |> where([d], not is_nil(d.ip_address)) + |> select([d], d.ip_address) + |> Repo.all() + |> MapSet.new() + end + + defp get_existing_interface_macs(organization_id) do + organization_id + |> DeviceQuery.for_organization() + |> join(:inner, [d], sd in Device, on: sd.device_id == d.id) + |> join(:inner, [d, sd], i in Interface, on: i.snmp_device_id == sd.id) + |> where([d, sd, i], not is_nil(i.if_phys_address)) + |> select([d, sd, i], i.if_phys_address) + |> Repo.all() + |> MapSet.new(&normalize_mac/1) + end + + defp normalize_mac(nil), do: nil + + defp normalize_mac(mac) when is_binary(mac) do + mac + |> String.downcase() + |> String.replace(~r/[:-]/, "") + |> String.replace(~r/(.{2})/, "\\1:") + |> String.trim_trailing(":") + end + + defp merge_neighbors_into_map(map, neighbors) do + Enum.reduce(neighbors, map, fn neighbor, acc -> + identifiers = extract_neighbor_identifiers(neighbor) + {id_type, id_value} = choose_best_identifier_from_list(identifiers) + key = identifier_key(id_type, id_value) + + entry = Map.get(acc, key, empty_entry()) + + updated_entry = %{ + entry + | identifiers: Enum.uniq(identifiers ++ entry.identifiers), + hostnames: add_if_present(entry.hostnames, neighbor.remote_system_name), + ips: add_if_present(entry.ips, neighbor.remote_address), + macs: add_if_present(entry.macs, normalize_mac(neighbor.remote_chassis_id)), + capabilities: Enum.uniq(entry.capabilities ++ (neighbor.remote_capabilities || [])), + platforms: add_if_present(entry.platforms, neighbor.remote_platform), + discovered_by: + Enum.uniq_by( + [ + %{ + device_id: neighbor.device.id, + device_name: neighbor.device.name, + interface: interface_name(neighbor.interface), + timestamp: neighbor.last_discovered_at + } + | entry.discovered_by + ], + & &1.device_id + ), + sources: %{entry.sources | neighbors: [neighbor | entry.sources.neighbors]} + } + + Map.put(acc, key, updated_entry) + end) + end + + defp merge_arp_into_map(map, arp_entries) do + Enum.reduce(arp_entries, map, fn arp, acc -> + identifiers = extract_arp_identifiers(arp) + {id_type, id_value} = choose_best_identifier_from_list(identifiers) + key = identifier_key(id_type, id_value) + + entry = Map.get(acc, key, empty_entry()) + + updated_entry = %{ + entry + | identifiers: Enum.uniq(identifiers ++ entry.identifiers), + ips: add_if_present(entry.ips, arp.ip_address), + macs: add_if_present(entry.macs, normalize_mac(arp.mac_address)), + discovered_by: + Enum.uniq_by( + [ + %{ + device_id: arp.device.id, + device_name: arp.device.name, + interface: interface_name(arp.interface), + timestamp: arp.last_seen_at + } + | entry.discovered_by + ], + & &1.device_id + ), + sources: %{entry.sources | arp: [arp | entry.sources.arp]} + } + + Map.put(acc, key, updated_entry) + end) + end + + defp merge_mac_into_map(map, mac_addresses) do + Enum.reduce(mac_addresses, map, fn mac_entry, acc -> + mac_normalized = normalize_mac(mac_entry.mac_address) + if is_nil(mac_normalized), do: acc, else: do_merge_mac(acc, mac_entry, mac_normalized) + end) + end + + defp do_merge_mac(acc, mac_entry, mac_normalized) do + key = identifier_key(:mac, mac_normalized) + + entry = Map.get(acc, key, empty_entry()) + + updated_entry = %{ + entry + | identifiers: Enum.uniq([{:mac, mac_normalized} | entry.identifiers]), + macs: Enum.uniq([mac_normalized | entry.macs]), + discovered_by: + Enum.uniq_by( + [ + %{ + device_id: mac_entry.device.id, + device_name: mac_entry.device.name, + interface: interface_name(mac_entry.interface), + timestamp: mac_entry.last_seen_at + } + | entry.discovered_by + ], + & &1.device_id + ), + sources: %{entry.sources | mac: [mac_entry | entry.sources.mac]} + } + + Map.put(acc, key, updated_entry) + end + + defp empty_entry do + %{ + identifiers: [], + hostnames: [], + ips: [], + macs: [], + capabilities: [], + platforms: [], + discovered_by: [], + sources: %{neighbors: [], arp: [], mac: []} + } + end + + defp choose_best_identifier_from_list(identifiers) do + Enum.find(identifiers, fn {type, _value} -> type == :mac end) || + Enum.find(identifiers, fn {type, _value} -> type == :ip end) || + Enum.find(identifiers, fn {type, _value} -> type == :hostname end) || + {:unknown, "unknown"} + end + + defp extract_neighbor_identifiers(neighbor) do + Enum.reject( + [ + if(neighbor.remote_chassis_id && mac_address?(neighbor.remote_chassis_id), + do: {:mac, normalize_mac(neighbor.remote_chassis_id)} + ), + if(neighbor.remote_address, do: {:ip, neighbor.remote_address}), + if(neighbor.remote_system_name, do: {:hostname, neighbor.remote_system_name}) + ], + &is_nil/1 + ) + end + + defp mac_address?(str) when is_binary(str) do + String.match?(str, ~r/^([0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}$/) + end + + defp mac_address?(_), do: false + + defp extract_arp_identifiers(arp) do + Enum.reject( + [if(arp.mac_address, do: {:mac, normalize_mac(arp.mac_address)}), if(arp.ip_address, do: {:ip, arp.ip_address})], + &is_nil/1 + ) + end + + defp identifier_key(type, value), do: "#{type}:#{value}" + + defp add_if_present(list, nil), do: list + defp add_if_present(list, ""), do: list + defp add_if_present(list, value), do: Enum.uniq([value | list]) + + defp interface_name(nil), do: "unknown" + defp interface_name(interface), do: interface.if_name || interface.if_descr || "if#{interface.if_index}" + + defp device_already_exists?(entry, existing_ips, existing_macs) do + Enum.any?(entry.ips, &MapSet.member?(existing_ips, &1)) || + Enum.any?(entry.macs, &MapSet.member?(existing_macs, &1)) + end + + defp build_discovered_device_struct(entry) do + identifier = choose_best_identifier(entry) + device_type = infer_device_type(entry) + manufacturer = infer_manufacturer(entry) + + %{ + identifier: identifier, + hostname: List.first(entry.hostnames), + ip_addresses: Enum.uniq(entry.ips), + mac_addresses: Enum.uniq(entry.macs), + device_type: device_type, + manufacturer: manufacturer, + capabilities: Enum.uniq(entry.capabilities), + platform: List.first(entry.platforms), + discovered_by: Enum.sort_by(entry.discovered_by, & &1.timestamp, {:desc, DateTime}), + source_count: count_sources(entry), + last_seen: get_most_recent_timestamp(entry), + protocols_used: get_protocols_used(entry.sources) + } + end + + defp choose_best_identifier(entry) do + cond do + entry.macs != [] -> %{type: :mac, value: List.first(entry.macs)} + entry.ips != [] -> %{type: :ip, value: List.first(entry.ips)} + entry.hostnames != [] -> %{type: :hostname, value: List.first(entry.hostnames)} + true -> %{type: :unknown, value: "unknown"} + end + end + + defp infer_device_type(entry) do + capabilities = Enum.map(entry.capabilities, &String.downcase/1) + platform = String.downcase(List.first(entry.platforms) || "") + hostname = String.downcase(List.first(entry.hostnames) || "") + + infer_from_capabilities(capabilities) || infer_from_platform(platform) || + infer_from_hostname(hostname) || :unknown + end + + defp infer_from_capabilities(capabilities) do + cond do + "router" in capabilities -> :router + "bridge" in capabilities or "switch" in capabilities -> :switch + "wlan-ap" in capabilities or "wlan" in capabilities -> :wireless + true -> nil + end + end + + defp infer_from_platform(platform) do + cond do + String.contains?(platform, "ios") -> :router + String.contains?(platform, "catalyst") -> :switch + String.contains?(platform, "linux") or String.contains?(platform, "windows") -> :server + true -> nil + end + end + + defp infer_from_hostname(hostname) do + cond do + hostname_suggests_router?(hostname) -> :router + hostname_suggests_switch?(hostname) -> :switch + hostname_suggests_wireless?(hostname) -> :wireless + hostname_suggests_server?(hostname) -> :server + hostname_suggests_workstation?(hostname) -> :workstation + true -> nil + end + end + + defp hostname_suggests_router?(hostname) do + String.contains?(hostname, "router") or String.contains?(hostname, "rtr") or + String.contains?(hostname, "gateway") or String.contains?(hostname, "gw") + end + + defp hostname_suggests_switch?(hostname) do + String.contains?(hostname, "switch") or String.contains?(hostname, "sw") + end + + defp hostname_suggests_wireless?(hostname) do + String.contains?(hostname, "ap") or String.contains?(hostname, "access") or + String.contains?(hostname, "wireless") or String.contains?(hostname, "wifi") or + String.contains?(hostname, "wap") + end + + defp hostname_suggests_server?(hostname) do + String.contains?(hostname, "server") or String.contains?(hostname, "srv") + end + + defp hostname_suggests_workstation?(hostname) do + String.contains?(hostname, "workstation") or String.contains?(hostname, "desktop") or + String.contains?(hostname, "laptop") or String.contains?(hostname, "pc") or + String.contains?(hostname, "ws") + end + + defp infer_manufacturer(entry) do + case entry.macs do + [] -> nil + [mac | _] -> lookup_mac_oui(mac) + end + end + + defp lookup_mac_oui(mac) when is_binary(mac) do + oui = + mac + |> String.replace(":", "") + |> String.upcase() + |> String.slice(0, 6) + + Map.get(mac_oui_database(), oui) + end + + defp lookup_mac_oui(_), do: nil + + defp count_sources(entry) do + count = 0 + count = if entry.sources.neighbors == [], do: count, else: count + 1 + count = if entry.sources.arp == [], do: count, else: count + 1 + if entry.sources.mac == [], do: count, else: count + 1 + end + + defp get_most_recent_timestamp(entry) do + timestamps = + Enum.reject( + Enum.map(entry.sources.neighbors, & &1.last_discovered_at) ++ + Enum.map(entry.sources.arp, & &1.last_seen_at) ++ Enum.map(entry.sources.mac, & &1.last_seen_at), + &is_nil/1 + ) + + case timestamps do + [] -> DateTime.utc_now() + _ -> Enum.max(timestamps, DateTime) + end + end + + defp get_protocols_used(sources) do + protocols = [] + + protocols = + case sources.neighbors do + [] -> + protocols + + neighbors -> + neighbor_protocols = + neighbors + |> Enum.map(& &1.protocol) + |> Enum.map(&safe_protocol_to_atom/1) + |> Enum.uniq() + + protocols ++ neighbor_protocols + end + + protocols = if sources.arp == [], do: protocols, else: [:arp | protocols] + protocols = if sources.mac == [], do: protocols, else: [:mac | protocols] + + Enum.uniq(protocols) + end + + defp safe_protocol_to_atom("lldp"), do: :lldp + defp safe_protocol_to_atom("cdp"), do: :cdp + defp safe_protocol_to_atom(_unknown), do: :unknown + + # Common MAC OUI to manufacturer mappings. + # Subset of the IEEE OUI database focused on common network equipment vendors. + # OUI table inlined below + defp mac_oui_database do + %{ + # Cisco + "000142" => "Cisco", + "000143" => "Cisco", + "00010D" => "Cisco", + "000164" => "Cisco", + "000196" => "Cisco", + "000197" => "Cisco", + "0001C7" => "Cisco", + "0001C9" => "Cisco", + "001818" => "Cisco", + "001B0C" => "Cisco", + "001B2A" => "Cisco", + "001D45" => "Cisco", + "001D70" => "Cisco", + "001E13" => "Cisco", + "001E4A" => "Cisco", + "001E7A" => "Cisco", + "001EBD" => "Cisco", + "001F6C" => "Cisco", + "001FCA" => "Cisco", + "0022BD" => "Cisco", + "002555" => "Cisco", + "0026CA" => "Cisco", + "002713" => "Cisco", + "00D097" => "Cisco", + "00E014" => "Cisco", + "04C5A4" => "Cisco", + "0C8527" => "Cisco", + "1C6A7A" => "Cisco", + "40F4EC" => "Cisco", + "5C50F7" => "Cisco", + "6CAB31" => "Cisco", + "6C41" => "Cisco", + "74A2E6" => "Cisco", + "D4A928" => "Cisco", + "E84F25" => "Cisco", + "F46E95" => "Cisco", + + # Ubiquiti + "00156D" => "Ubiquiti", + "001B2F" => "Ubiquiti", + "001DD7" => "Ubiquiti", + "0027BA" => "Ubiquiti", + "04189A" => "Ubiquiti", + "0418D6" => "Ubiquiti", + "247F20" => "Ubiquiti", + "248A07" => "Ubiquiti", + "44D9E7" => "Ubiquiti", + "74ACB9" => "Ubiquiti", + "788A20" => "Ubiquiti", + "80EA96" => "Ubiquiti", + "A42BB0" => "Ubiquiti", + "B4FBE4" => "Ubiquiti", + "DC9FDB" => "Ubiquiti", + "E063DA" => "Ubiquiti", + "F09FC2" => "Ubiquiti", + "F4E2C6" => "Ubiquiti", + "FCECDA" => "Ubiquiti", + + # MikroTik + "000456" => "MikroTik", + "001D2E" => "MikroTik", + "00272D" => "MikroTik", + "0C424C" => "MikroTik", + "2C3952" => "MikroTik", + "48A9D2" => "MikroTik", + "4C5E0C" => "MikroTik", + "6C3B6B" => "MikroTik", + "D49443" => "MikroTik", + "D4CA6D" => "MikroTik", + "E748B8" => "MikroTik", + + # TP-Link + "000E8F" => "TP-Link", + "001DD9" => "TP-Link", + "0C8268" => "TP-Link", + "1027F5" => "TP-Link", + "148824" => "TP-Link", + "1C3BF3" => "TP-Link", + "300108" => "TP-Link", + "3C468A" => "TP-Link", + "50C7BF" => "TP-Link", + "641C67" => "TP-Link", + "7C8BCA" => "TP-Link", + "98DAC4" => "TP-Link", + "A0F3C1" => "TP-Link", + "C46E1F" => "TP-Link", + "E8DE27" => "TP-Link", + "EC086B" => "TP-Link", + "F0B429" => "TP-Link", + + # Juniper + "001517" => "Juniper", + "001B1F" => "Juniper", + "002163" => "Juniper", + "002281" => "Juniper", + "0026DE" => "Juniper", + "0026DC" => "Juniper", + "002642" => "Juniper", + "009E1E" => "Juniper", + "285024" => "Juniper", + "28A24B" => "Juniper", + "3C8AB6" => "Juniper", + "5C5E" => "Juniper", + "842B2B" => "Juniper", + + # Aruba (HPE) + "001A1E" => "Aruba", + "00242C" => "Aruba", + "04BD88" => "Aruba", + "0C5496" => "Aruba", + "200E52" => "Aruba", + "24DE52" => "Aruba", + "40E3D6" => "Aruba", + "6C72E7" => "Aruba", + "94B4C4" => "Aruba", + "9C1C12" => "Aruba", + "B8D9CE" => "Aruba", + "D8C7C8" => "Aruba", + + # Dell + "000874" => "Dell", + "000BDB" => "Dell", + "000C7B" => "Dell", + "000D56" => "Dell", + "000E0C" => "Dell", + "000F1F" => "Dell", + "001111" => "Dell", + "00123F" => "Dell", + "0013C3" => "Dell", + "001372" => "Dell", + "001485" => "Dell", + "0015C5" => "Dell", + "0016CE" => "Dell", + "001731" => "Dell", + "001871" => "Dell", + "0019B9" => "Dell", + "001A4B" => "Dell", + "001C23" => "Dell", + "001D09" => "Dell", + "001E4F" => "Dell", + "0021F6" => "Dell", + "002219" => "Dell", + "002264" => "Dell", + "0023AE" => "Dell", + "002564" => "Dell", + "5C260A" => "Dell", + + # HP/HPE + "001279" => "HP", + "001438" => "HP", + "001560" => "HP", + "0016B9" => "HP", + "001708" => "HP", + "001B78" => "HP", + "001C2E" => "HP", + "001E0B" => "HP", + "001F29" => "HP", + "00237D" => "HP", + "002481" => "HP", + "00248D" => "HP", + "002655" => "HP", + "3863BB" => "HP", + + # Netgear + "00095B" => "Netgear", + "000FB5" => "Netgear", + "001E2A" => "Netgear", + "001F33" => "Netgear", + "00223F" => "Netgear", + "00260B" => "Netgear", + "04A151" => "Netgear", + "0846A0" => "Netgear", + "2C3033" => "Netgear", + "3448ED" => "Netgear", + "4C9EFF" => "Netgear", + "74DA38" => "Netgear", + "84C9B2" => "Netgear", + "A0040A" => "Netgear", + "C43DC7" => "Netgear", + + # Fortinet + "0009819" => "Fortinet", + "001910" => "Fortinet", + "00E04C" => "Fortinet", + "080027" => "Fortinet", + "0C1D36" => "Fortinet", + "246C05" => "Fortinet", + "708BCD" => "Fortinet", + "900B29" => "Fortinet", + + # Palo Alto + "009C33" => "Palo Alto", + "001B17" => "Palo Alto", + "5C5B35" => "Palo Alto", + "8C89A5" => "Palo Alto", + "D4F46F" => "Palo Alto" + } + end +end diff --git a/lib/towerops/snmp/entity_physicals.ex b/lib/towerops/snmp/entity_physicals.ex new file mode 100644 index 00000000..2e63882a --- /dev/null +++ b/lib/towerops/snmp/entity_physicals.ex @@ -0,0 +1,103 @@ +defmodule Towerops.Snmp.EntityPhysicals do + @moduledoc """ + Read-side queries and mutations for ENTITY-MIB physical components, plus + persistence for entity physical status time-series readings. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Repo + alias Towerops.Snmp.EntityPhysical + alias Towerops.Snmp.EntityPhysicalReading + + @doc """ + Lists all entity physical components for a given SNMP device. + """ + def list_entity_physical(snmp_device_id) do + EntityPhysical + |> where([e], e.snmp_device_id == ^snmp_device_id) + |> order_by([e], e.entity_index) + |> Repo.all() + end + + @doc """ + Gets a single entity physical component by id. + """ + def get_entity_physical(id) do + case Repo.get(EntityPhysical, id) do + nil -> {:error, :not_found} + entity -> {:ok, entity} + end + end + + @doc """ + Gets entity physical components by class (chassis, module, port, etc.). + """ + def get_entity_physical_by_class(snmp_device_id, entity_class) do + EntityPhysical + |> where([e], e.snmp_device_id == ^snmp_device_id and e.entity_class == ^entity_class) + |> order_by([e], e.entity_index) + |> Repo.all() + end + + @doc """ + Gets entity physical components as a hierarchical tree. + + Returns root entities with children preloaded recursively. + """ + def get_entity_physical_tree(snmp_device_id) do + all_entities = + EntityPhysical + |> where([e], e.snmp_device_id == ^snmp_device_id) + |> order_by([e], e.entity_index) + |> Repo.all() + + build_entity_tree(all_entities) + end + + @doc """ + Updates an entity physical component with the given attributes. + """ + def update_entity_physical(entity, attrs) do + entity + |> EntityPhysical.changeset(attrs) + |> Repo.update() + end + + @doc """ + Batch inserts entity physical status readings. + """ + def create_entity_physical_readings_batch([]), do: {0, nil} + + def create_entity_physical_readings_batch(entries) when is_list(entries) do + now = Towerops.Time.now() + + rows = + Enum.map(entries, fn entry -> + entry + |> Map.put(:id, Ecto.UUID.generate()) + |> Map.put(:inserted_at, now) + |> Map.put(:updated_at, now) + end) + + Repo.insert_all(EntityPhysicalReading, rows) + end + + # Build hierarchical tree from flat list of entities + defp build_entity_tree(entities) do + children_by_parent = Enum.group_by(entities, & &1.parent_id) + root_entities = Map.get(children_by_parent, nil, []) + Enum.map(root_entities, &attach_children(&1, children_by_parent)) + end + + defp attach_children(entity, children_by_parent) do + children = + children_by_parent + |> Map.get(entity.id, []) + |> Enum.map(&attach_children(&1, children_by_parent)) + + Map.put(entity, :children, children) + end +end diff --git a/lib/towerops/snmp/mac_addresses.ex b/lib/towerops/snmp/mac_addresses.ex new file mode 100644 index 00000000..23fcc510 --- /dev/null +++ b/lib/towerops/snmp/mac_addresses.ex @@ -0,0 +1,98 @@ +defmodule Towerops.Snmp.MacAddresses do + @moduledoc """ + Read-side queries, upserts, and stale cleanup for MAC address forwarding + table entries discovered on SNMP devices. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Repo + alias Towerops.Snmp.MacAddress + + require Logger + + @doc """ + Lists all MAC address entries for a device. + """ + def list_mac_addresses(device_id) do + MacAddress + |> where([m], m.device_id == ^device_id) + |> order_by([m], [m.vlan_id, m.mac_address]) + |> preload(:interface) + |> Repo.all() + end + + @doc """ + Creates or updates a MAC address entry. + + Uses atomic PostgreSQL ON CONFLICT to prevent race conditions when multiple + pollers try to upsert the same MAC address simultaneously. The unique constraint + handles duplicate detection automatically, eliminating the need for manual cleanup. + """ + def upsert_mac_address(attrs) do + %MacAddress{} + |> MacAddress.changeset(attrs) + |> Repo.insert( + on_conflict: {:replace_all_except, [:id, :inserted_at]}, + conflict_target: [:device_id, :mac_address, :vlan_id], + returning: true + ) + end + + @doc """ + Deletes stale MAC address entries that haven't been seen since the given datetime. + """ + def delete_stale_mac_addresses(device_id, cutoff_datetime) do + MacAddress + |> where([m], m.device_id == ^device_id) + |> where([m], m.last_seen_at < ^cutoff_datetime) + |> Repo.delete_all() + end + + @doc """ + Atomically deletes stale MAC addresses and upserts new ones in a single transaction. + Prevents race conditions where concurrent delete/upsert could remove fresh data. + """ + def delete_stale_and_upsert_mac_addresses(device_id, mac_entries, interfaces, cutoff) do + Repo.transaction(fn -> + delete_stale_mac_addresses(device_id, cutoff) + upsert_mac_addresses(device_id, mac_entries, interfaces) + end) + end + + @doc """ + Bulk upserts MAC address entries for a device. + Returns `{success_count, error_count}` tuple. + """ + def upsert_mac_addresses(device_id, mac_entries, interfaces) do + # Build a map from bridge port index to interface ID + # Note: port_index from BRIDGE-MIB is not the same as if_index + # For now, we'll use a simple mapping if they're available + interface_map = Map.new(interfaces, fn i -> {i.if_index, i.id} end) + + mac_entries + |> Enum.reduce({0, 0}, fn entry, {s, e} -> + interface_id = Map.get(interface_map, entry.port_index) + + attrs = + entry + |> Map.put(:device_id, device_id) + |> Map.put(:interface_id, interface_id) + + case upsert_mac_address(attrs) do + {:ok, _} -> {s + 1, e} + {:error, _} -> {s, e + 1} + end + end) + |> tap(fn {_s, errors} -> + if errors > 0 do + Logger.warning("#{errors} MAC upsert failures for device #{device_id}", + device_id: device_id, + error_count: errors + ) + end + end) + end +end diff --git a/lib/towerops/snmp/printer_supplies.ex b/lib/towerops/snmp/printer_supplies.ex new file mode 100644 index 00000000..fd3b080e --- /dev/null +++ b/lib/towerops/snmp/printer_supplies.ex @@ -0,0 +1,68 @@ +defmodule Towerops.Snmp.PrinterSupplies do + @moduledoc """ + Read-side queries and mutations for printer supplies (toner, ink, drum, etc.) + discovered on SNMP-managed printers. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Repo + alias Towerops.Snmp.PrinterSupply + + @doc """ + Lists all printer supplies for a device. + """ + def list_printer_supplies(snmp_device_id) do + PrinterSupply + |> where([s], s.snmp_device_id == ^snmp_device_id) + |> order_by([s], s.supply_index) + |> Repo.all() + end + + @doc """ + Gets a single printer supply by id. + Returns nil if not found. + """ + def get_printer_supply(id) do + Repo.get(PrinterSupply, id) + end + + @doc """ + Gets printer supplies by type (toner, ink, drum, etc.). + """ + def get_printer_supplies_by_type(snmp_device_id, supply_type) do + PrinterSupply + |> where([s], s.snmp_device_id == ^snmp_device_id and s.supply_type == ^supply_type) + |> order_by([s], s.supply_index) + |> Repo.all() + end + + @doc """ + Gets printer supplies below a threshold percentage. + + Returns supplies where (current_level / max_capacity * 100) < threshold. + Excludes supplies without capacity data. + """ + def get_low_printer_supplies(snmp_device_id, threshold_percent) do + PrinterSupply + |> where([s], s.snmp_device_id == ^snmp_device_id) + |> where([s], not is_nil(s.max_capacity) and not is_nil(s.current_level)) + |> where([s], s.max_capacity > 0) + |> Repo.all() + |> Enum.filter(fn supply -> + percent = supply.current_level * 100 / supply.max_capacity + percent < threshold_percent + end) + end + + @doc """ + Updates a printer supply with the given attributes. + """ + def update_printer_supply(printer_supply, attrs) do + printer_supply + |> PrinterSupply.changeset(attrs) + |> Repo.update() + end +end diff --git a/lib/towerops/snmp/topology.ex b/lib/towerops/snmp/topology.ex new file mode 100644 index 00000000..5156cdc3 --- /dev/null +++ b/lib/towerops/snmp/topology.ex @@ -0,0 +1,320 @@ +defmodule Towerops.Snmp.Topology do + @moduledoc """ + Builds a network topology graph for an organization, combining inventoried + devices with discovered devices and their physical (LLDP/CDP) connections, + plus subnet groupings derived from interface IPs. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Devices.DeviceQuery + alias Towerops.Repo + alias Towerops.Snmp.Device + alias Towerops.Snmp.DiscoveredDevices + alias Towerops.Snmp.Interface + alias Towerops.Snmp.IpAddress + alias Towerops.Snmp.Neighbor + + @doc """ + Builds a network topology map for an organization showing devices and their connections. + + Returns a graph structure with: + - nodes: All devices (both added and discovered) + - edges: Physical connections between interfaces + - subnets: Logical groupings of devices on same networks + """ + def get_network_topology(organization_id) do + devices = get_devices_with_topology(organization_id) + discovered = DiscoveredDevices.list_discovered_devices_for_organization(organization_id) + + nodes = build_topology_nodes(devices, discovered) + edges = build_topology_edges(devices, nodes) + subnets = build_subnet_groups(devices) + + %{ + nodes: nodes, + edges: edges, + subnets: subnets, + stats: %{ + total_devices: length(nodes), + added_devices: Enum.count(nodes, &(!&1.discovered)), + discovered_devices: Enum.count(nodes, & &1.discovered), + total_links: length(edges), + total_subnets: length(subnets) + }, + last_updated: DateTime.utc_now() + } + end + + defp get_devices_with_topology(organization_id) do + organization_id + |> DeviceQuery.for_organization() + |> join(:inner, [d], s in assoc(d, :site)) + |> join(:left, [d, s], sd in Device, on: d.id == sd.device_id) + |> join(:left, [d, s, sd], i in Interface, on: sd.id == i.snmp_device_id) + |> join(:left, [d, s, sd, i], n in Neighbor, on: i.id == n.interface_id) + |> join(:left, [d, s, sd, i, n], ip in IpAddress, on: i.id == ip.snmp_interface_id) + |> preload( + [d, s, sd, i, n, ip], + site: s, + snmp_device: {sd, [interfaces: {i, [neighbors: n, ip_addresses: ip]}]} + ) + |> Repo.all() + end + + defp build_topology_nodes(devices, discovered) do + device_nodes = + Enum.map(devices, fn device -> + %{ + id: device.id, + label: device.name, + type: device_type_atom(device), + status: device.status, + site_id: device.site_id, + site_name: device.site.name, + ip_address: device.ip_address, + discovered: false, + snmp_enabled: device.snmp_enabled, + monitoring_enabled: device.monitoring_enabled + } + end) + + discovered_nodes = + Enum.map(discovered, fn disc -> + %{ + id: "discovered_#{disc.identifier.type}_#{disc.identifier.value}", + label: disc.hostname || disc.identifier.value, + type: disc.device_type, + status: :unknown, + site_id: nil, + site_name: nil, + ip_address: List.first(disc.ip_addresses), + discovered: true, + manufacturer: disc.manufacturer + } + end) + + device_nodes ++ discovered_nodes + end + + defp build_topology_edges(devices, nodes) do + node_lookup = build_node_lookup(devices, nodes) + + devices + |> Enum.flat_map(&build_device_edges(&1, node_lookup)) + |> Enum.uniq_by(& &1.id) + end + + defp build_node_lookup(devices, nodes) do + by_id = Map.new(nodes, &{&1.id, &1}) + + by_name = + nodes + |> Enum.filter(&(&1.label != nil)) + |> Map.new(&{String.downcase(&1.label), &1.id}) + + by_ip = Map.new(nodes, fn node -> {node.ip_address, node.id} end) + + by_mac = + devices + |> Enum.flat_map(fn device -> + case device.snmp_device do + nil -> + [] + + snmp_device -> + snmp_device.interfaces + |> Enum.filter(&(&1.if_phys_address != nil)) + |> Enum.map(&{&1.if_phys_address, device.id}) + end + end) + |> Map.new() + + %{ + by_id: by_id, + by_name: by_name, + by_ip: by_ip, + by_mac: by_mac, + node_ids: MapSet.new(Map.keys(by_id)) + } + end + + defp build_device_edges(device, node_lookup) do + case device.snmp_device do + nil -> [] + snmp_device -> Enum.flat_map(snmp_device.interfaces, &build_interface_edges(device, &1, node_lookup)) + end + end + + defp build_interface_edges(device, interface, node_lookup) do + interface.neighbors + |> Enum.map(fn neighbor -> + target_id = find_target_node_id(neighbor, node_lookup) + source_ips = format_interface_ips(interface.ip_addresses) + + edge_parts = Enum.sort([device.id, target_id || "unknown"]) + edge_id = Enum.join(edge_parts, "_") + + %{ + id: edge_id, + source: device.id, + target: target_id, + source_interface: interface.if_name || interface.if_descr, + target_interface: neighbor.remote_port_id, + source_interface_id: interface.id, + source_ips: source_ips, + target_ip: neighbor.remote_address, + protocol: neighbor.protocol, + last_discovered: neighbor.last_discovered_at + } + end) + |> Enum.filter(&(&1.target != nil)) + end + + defp format_interface_ips(ip_addresses) do + ip_addresses + |> Enum.filter(&ipv4?(&1.ip_address)) + |> Enum.map(fn ip -> + subnet = if ip.subnet_mask, do: "/#{calculate_prefix_length(ip.subnet_mask)}", else: "" + "#{ip.ip_address}#{subnet}" + end) + end + + defp ipv4?(ip_address) when is_binary(ip_address) do + case :inet.parse_address(String.to_charlist(ip_address)) do + {:ok, {_, _, _, _}} -> true + _ -> false + end + end + + defp ipv4?(_), do: false + + defp calculate_prefix_length(nil), do: "" + + defp calculate_prefix_length(subnet_mask) do + subnet_mask + |> String.split(".") + |> Enum.map(&String.to_integer/1) + |> Enum.map(&Integer.to_string(&1, 2)) + |> Enum.map_join(&String.pad_leading(&1, 8, "0")) + |> String.graphemes() + |> Enum.count(&(&1 == "1")) + end + + defp find_target_node_id(neighbor, node_lookup) do + name_match = + if neighbor.remote_system_name do + Map.get(node_lookup.by_name, String.downcase(neighbor.remote_system_name)) + end + + ip_match = + if neighbor.remote_address do + Map.get(node_lookup.by_ip, neighbor.remote_address) + end + + mac_match = + if neighbor.remote_chassis_id do + Map.get(node_lookup.by_mac, neighbor.remote_chassis_id) + end + + discovered_match = + if neighbor.remote_chassis_id do + discovered_id = "discovered_mac_#{neighbor.remote_chassis_id}" + if MapSet.member?(node_lookup.node_ids, discovered_id), do: discovered_id + end + + name_match || ip_match || mac_match || discovered_match + end + + defp build_subnet_groups(devices) do + devices + |> Enum.flat_map(fn device -> + case device.snmp_device do + nil -> + [] + + snmp_device -> + Enum.flat_map(snmp_device.interfaces, fn interface -> + Enum.map(interface.ip_addresses, fn ip -> + {device.id, ip.ip_address, ip.subnet_mask} + end) + end) + end + end) + |> Enum.group_by(fn {_device_id, ip, mask} -> + case mask do + nil -> nil + _ -> calculate_cidr(ip, mask) + end + end) + |> Enum.reject(fn {cidr, _} -> is_nil(cidr) end) + |> Enum.map(fn {cidr, entries} -> + device_ids = entries |> Enum.map(fn {device_id, _, _} -> device_id end) |> Enum.uniq() + + %{ + cidr: cidr, + device_ids: device_ids, + device_count: length(device_ids) + } + end) + |> Enum.sort_by(& &1.device_count, :desc) + end + + # Convert device type to atom with whitelist to prevent atom exhaustion + defp device_type_atom(device) when is_atom(device), do: device + defp device_type_atom("router"), do: :router + defp device_type_atom("switch"), do: :switch + defp device_type_atom("wireless"), do: :wireless + defp device_type_atom("server"), do: :server + defp device_type_atom("workstation"), do: :workstation + defp device_type_atom("firewall"), do: :firewall + defp device_type_atom("printer"), do: :printer + defp device_type_atom("phone"), do: :phone + defp device_type_atom(_), do: :unknown + + defp calculate_cidr(ip, mask) do + case {parse_ip(ip), parse_ip(mask)} do + {{:ok, _ip_tuple}, {:ok, mask_tuple}} -> + prefix_length = count_mask_bits(mask_tuple) + network = calculate_network(ip, mask) + "#{network}/#{prefix_length}" + + _ -> + nil + end + end + + defp parse_ip(ip) when is_binary(ip) do + case :inet.parse_address(String.to_charlist(ip)) do + {:ok, tuple} -> {:ok, tuple} + {:error, _} -> :error + end + end + + defp parse_ip(_), do: :error + + defp count_mask_bits({a, b, c, d}) do + [a, b, c, d] + |> Enum.map(&Integer.to_string(&1, 2)) + |> Enum.map_join(&String.pad_leading(&1, 8, "0")) + |> String.graphemes() + |> Enum.count(&(&1 == "1")) + end + + defp calculate_network(ip, mask) do + {:ok, ip_tuple} = parse_ip(ip) + {:ok, mask_tuple} = parse_ip(mask) + + {a1, b1, c1, d1} = ip_tuple + {a2, b2, c2, d2} = mask_tuple + + a = Bitwise.band(a1, a2) + b = Bitwise.band(b1, b2) + c = Bitwise.band(c1, c2) + d = Bitwise.band(d1, d2) + + "#{a}.#{b}.#{c}.#{d}" + end +end diff --git a/lib/towerops/snmp/transceivers.ex b/lib/towerops/snmp/transceivers.ex new file mode 100644 index 00000000..fab569c8 --- /dev/null +++ b/lib/towerops/snmp/transceivers.ex @@ -0,0 +1,105 @@ +defmodule Towerops.Snmp.Transceivers do + @moduledoc """ + Read-side queries and mutations for transceivers (SFPs, optics) discovered + on SNMP devices, plus persistence for transceiver DOM time-series readings. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Repo + alias Towerops.Snmp.Transceiver + alias Towerops.Snmp.TransceiverReading + + @doc """ + Lists all transceivers for a device. + """ + def list_transceivers(snmp_device_id) do + Transceiver + |> where([t], t.snmp_device_id == ^snmp_device_id) + |> order_by([t], t.port_index) + |> Repo.all() + end + + @doc """ + Gets a single transceiver by id. + """ + def get_transceiver(id) do + case Repo.get(Transceiver, id) do + nil -> {:error, :not_found} + transceiver -> {:ok, transceiver} + end + end + + @doc """ + Gets transceivers by type (SFP, SFP+, QSFP, etc.). + """ + def get_transceivers_by_type(snmp_device_id, transceiver_type) do + Transceiver + |> where([t], t.snmp_device_id == ^snmp_device_id and t.transceiver_type == ^transceiver_type) + |> order_by([t], t.port_index) + |> Repo.all() + end + + @doc """ + Lists transceivers with DOM (Digital Optical Monitoring) support. + """ + def list_transceivers_with_dom(snmp_device_id) do + Transceiver + |> where([t], t.snmp_device_id == ^snmp_device_id and t.supports_dom == true) + |> order_by([t], t.port_index) + |> Repo.all() + end + + @doc """ + Gets a transceiver with its latest DOM reading. + + Returns the transceiver with a `:latest_reading` field containing the most recent + TransceiverReading, or nil if no readings exist. + """ + def get_transceiver_with_latest_reading(id) do + case Repo.get(Transceiver, id) do + nil -> + {:error, :not_found} + + transceiver -> + latest_reading = + TransceiverReading + |> where([r], r.transceiver_id == ^id) + |> order_by([r], desc: r.measured_at) + |> limit(1) + |> Repo.one() + + {:ok, Map.put(transceiver, :latest_reading, latest_reading)} + end + end + + @doc """ + Updates a transceiver with the given attributes. + """ + def update_transceiver(transceiver, attrs) do + transceiver + |> Transceiver.changeset(attrs) + |> Repo.update() + end + + @doc """ + Creates multiple transceiver DOM readings in a single batch insert. + """ + def create_transceiver_readings_batch([]), do: {0, nil} + + def create_transceiver_readings_batch(entries) when is_list(entries) do + now = Towerops.Time.now() + + rows = + Enum.map(entries, fn entry -> + entry + |> Map.put(:id, Ecto.UUID.generate()) + |> Map.put(:inserted_at, now) + |> Map.put(:updated_at, now) + end) + + Repo.insert_all(TransceiverReading, rows) + end +end diff --git a/lib/towerops/snmp/wireless_clients.ex b/lib/towerops/snmp/wireless_clients.ex new file mode 100644 index 00000000..8347309e --- /dev/null +++ b/lib/towerops/snmp/wireless_clients.ex @@ -0,0 +1,203 @@ +defmodule Towerops.Snmp.WirelessClients do + @moduledoc """ + Read-side queries, upserts, and stale cleanup for wireless clients + associated with access-point devices, plus persistence for wireless + client time-series readings. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Devices.Device, as: DeviceSchema + alias Towerops.Repo + alias Towerops.Snmp.WirelessClient + alias Towerops.Snmp.WirelessClientReading + + @doc """ + Lists all wireless clients for a device. + """ + def list_wireless_clients(device_id) do + WirelessClient + |> where([w], w.device_id == ^device_id) + |> order_by([w], w.mac_address) + |> Repo.all() + end + + @doc """ + Gets wireless clients by MAC addresses within an organization. + """ + def get_wireless_clients_by_mac(organization_id, mac_addresses) do + WirelessClient + |> where([w], w.organization_id == ^organization_id) + |> where([w], w.mac_address in ^mac_addresses) + |> Repo.all() + end + + @doc """ + Creates or updates a wireless client entry. + """ + def upsert_wireless_client(attrs) do + %WirelessClient{} + |> WirelessClient.changeset(attrs) + |> Repo.insert( + on_conflict: {:replace_all_except, [:id, :inserted_at]}, + conflict_target: [:device_id, :mac_address], + returning: true + ) + end + + @doc """ + Upserts a list of wireless clients for a device. + Returns `{success_count, error_count}`. + """ + def upsert_wireless_clients(device_id, organization_id, clients) do + Enum.reduce(clients, {0, 0}, fn client, {s, e} -> + attrs = + client + |> Map.put(:device_id, device_id) + |> Map.put(:organization_id, organization_id) + + case upsert_wireless_client(attrs) do + {:ok, _} -> {s + 1, e} + {:error, _} -> {s, e + 1} + end + end) + end + + @doc """ + Deletes stale wireless clients that haven't been seen since the given datetime. + """ + def delete_stale_wireless_clients(device_id, cutoff_datetime) do + WirelessClient + |> where([w], w.device_id == ^device_id) + |> where([w], w.last_seen_at < ^cutoff_datetime) + |> Repo.delete_all() + end + + @doc """ + Atomically deletes stale wireless clients and upserts new ones. + """ + def delete_stale_and_upsert_wireless_clients(device_id, organization_id, clients, cutoff) do + Repo.transaction(fn -> + delete_stale_wireless_clients(device_id, cutoff) + upsert_wireless_clients(device_id, organization_id, clients) + end) + end + + @doc """ + Batch inserts multiple wireless client readings using `Repo.insert_all/3`. + + Accepts a list of attribute maps containing wireless client metrics. + Generates UUIDs and timestamps automatically. Returns `{count, nil}`. + """ + @spec create_wireless_client_readings_batch([map()]) :: {non_neg_integer(), nil} + def create_wireless_client_readings_batch([]), do: {0, nil} + + def create_wireless_client_readings_batch(entries) when is_list(entries) do + now = Towerops.Time.now() + + rows = + Enum.map(entries, fn entry -> + entry + |> Map.put(:id, Ecto.UUID.generate()) + |> Map.put(:inserted_at, now) + end) + + Repo.insert_all(WirelessClientReading, rows) + end + + @doc """ + Lists wireless clients with weak signal strength for an organization. + + Returns clients with signal strength below the threshold (default -75 dBm). + """ + @spec list_weak_signal_clients(Ecto.UUID.t(), integer()) :: [WirelessClient.t()] + def list_weak_signal_clients(organization_id, threshold_dbm \\ -75) do + Repo.all( + from(c in WirelessClient, + join: d in assoc(c, :device), + where: c.organization_id == ^organization_id, + where: not is_nil(c.signal_strength), + where: c.signal_strength < ^threshold_dbm, + preload: [device: d], + order_by: [asc: c.signal_strength] + ) + ) + end + + @doc """ + Lists wireless clients with low SNR for an organization. + + Returns clients with SNR below the threshold (default 15 dB). + """ + @spec list_low_snr_clients(Ecto.UUID.t(), integer()) :: [WirelessClient.t()] + def list_low_snr_clients(organization_id, threshold_db \\ 15) do + Repo.all( + from(c in WirelessClient, + join: d in assoc(c, :device), + where: c.organization_id == ^organization_id, + where: not is_nil(c.snr), + where: c.snr < ^threshold_db, + preload: [device: d], + order_by: [asc: c.snr] + ) + ) + end + + @doc """ + Gets wireless client count per device for an organization. + + Returns a map of device_id => client_count. + """ + @spec get_wireless_client_count_by_device(Ecto.UUID.t()) :: %{Ecto.UUID.t() => integer()} + def get_wireless_client_count_by_device(organization_id) do + from(c in WirelessClient, + where: c.organization_id == ^organization_id, + group_by: c.device_id, + select: {c.device_id, count(c.id)} + ) + |> Repo.all() + |> Map.new() + end + + @doc """ + Lists devices (access points) with client count exceeding threshold. + + Returns devices with their client counts. + """ + @spec list_overloaded_aps(Ecto.UUID.t(), integer()) :: [{DeviceSchema.t(), integer()}] + def list_overloaded_aps(organization_id, threshold \\ 50) do + Repo.all( + from(c in WirelessClient, + join: d in DeviceSchema, + on: c.device_id == d.id, + where: c.organization_id == ^organization_id, + group_by: [c.device_id, d.id], + having: count(c.id) > ^threshold, + select: {d, count(c.id)}, + order_by: [desc: count(c.id)] + ) + ) + end + + @doc """ + Gets the last seen timestamp for a wireless client by MAC address. + + Returns the most recent last_seen_at timestamp for the given MAC across all devices. + """ + @spec get_wireless_client_last_seen(Ecto.UUID.t(), String.t()) :: DateTime.t() | nil + def get_wireless_client_last_seen(organization_id, mac_address) do + mac_lower = String.downcase(mac_address) + + Repo.one( + from(c in WirelessClient, + where: c.organization_id == ^organization_id, + where: fragment("LOWER(?)", c.mac_address) == ^mac_lower, + select: c.last_seen_at, + order_by: [desc: c.last_seen_at], + limit: 1 + ) + ) + end +end