diff --git a/lib/towerops/gaiia.ex b/lib/towerops/gaiia.ex index 59478394..07ae732c 100644 --- a/lib/towerops/gaiia.ex +++ b/lib/towerops/gaiia.ex @@ -8,6 +8,7 @@ defmodule Towerops.Gaiia do alias Towerops.Devices.Device alias Towerops.Gaiia.Account alias Towerops.Gaiia.BillingSubscription + alias Towerops.Gaiia.DeviceSubscriberLink alias Towerops.Gaiia.InventoryItem alias Towerops.Gaiia.NetworkSite alias Towerops.Repo @@ -241,13 +242,12 @@ defmodule Towerops.Gaiia do if device_ids == [] do %{} else - # Count inventory items with assigned accounts per device as subscriber proxy - InventoryItem - |> where([ii], ii.device_id in ^device_ids and not is_nil(ii.assigned_account_gaiia_id)) - |> group_by([ii], ii.device_id) - |> select([ii], %{ - device_id: ii.device_id, - subscriber_count: count(ii.id, :distinct) + DeviceSubscriberLink + |> where([l], l.device_id in ^device_ids) + |> group_by([l], l.device_id) + |> select([l], %{ + device_id: l.device_id, + subscriber_count: count(l.gaiia_account_id, :distinct) }) |> Repo.all() |> Map.new(fn row -> @@ -264,4 +264,117 @@ defmodule Towerops.Gaiia do |> select([ns], %{account_count: ns.account_count, total_mrr: ns.total_mrr}) |> Repo.one() end + + # --- Device Impact Queries --- + + @doc """ + Get subscriber impact for a single device. + Returns `%{subscriber_count: integer, mrr: Decimal.t(), accounts: [%Account{}]}`. + """ + def get_device_impact(device_id) do + account_ids = + DeviceSubscriberLink + |> where(device_id: ^device_id) + |> select([l], l.gaiia_account_id) + |> Repo.all() + |> Enum.uniq() + + if account_ids == [] do + %{subscriber_count: 0, mrr: Decimal.new(0), accounts: []} + else + accounts = + Account + |> where([a], a.id in ^account_ids) + |> Repo.all() + + mrr = + BillingSubscription + |> join(:inner, [bs], a in Account, on: bs.account_gaiia_id == a.gaiia_id and bs.organization_id == a.organization_id) + |> where([bs, a], a.id in ^account_ids) + |> where([bs], bs.status == "ACTIVE") + |> Repo.aggregate(:sum, :mrr_amount) || Decimal.new(0) + + %{subscriber_count: length(accounts), mrr: mrr, accounts: accounts} + end + end + + @doc """ + Get subscriber impact for multiple devices (batch). + Returns `%{device_id => %{subscriber_count: integer, mrr: Decimal.t()}}`. + """ + def get_devices_impact(device_ids) when is_list(device_ids) do + if device_ids == [] do + %{} + else + # Get account links per device + links = + DeviceSubscriberLink + |> where([l], l.device_id in ^device_ids) + |> join(:inner, [l], a in Account, on: l.gaiia_account_id == a.id) + |> join(:left, [l, a], bs in BillingSubscription, + on: bs.account_gaiia_id == a.gaiia_id and bs.organization_id == a.organization_id and bs.status == "ACTIVE" + ) + |> group_by([l, a, bs], l.device_id) + |> select([l, a, bs], %{ + device_id: l.device_id, + subscriber_count: count(a.id, :distinct), + mrr: coalesce(sum(bs.mrr_amount), 0) + }) + |> Repo.all() + + Map.new(links, fn row -> + {row.device_id, %{subscriber_count: row.subscriber_count, mrr: row.mrr}} + end) + end + end + + @doc """ + Get subscriber impact for a site (all devices at site). + Returns `%{subscriber_count: integer, mrr: Decimal.t(), devices: [%{device_id, name, count, mrr}]}`. + """ + def get_site_impact(site_id) do + alias Towerops.Devices.Device + + device_ids = + Device + |> where(site_id: ^site_id) + |> select([d], d.id) + |> Repo.all() + + impact = get_devices_impact(device_ids) + + devices_with_names = + Device + |> where([d], d.id in ^device_ids) + |> select([d], {d.id, d.name}) + |> Repo.all() + |> Enum.map(fn {id, name} -> + di = Map.get(impact, id, %{subscriber_count: 0, mrr: Decimal.new(0)}) + %{device_id: id, name: name, count: di.subscriber_count, mrr: di.mrr} + end) + + total_account_ids = + DeviceSubscriberLink + |> where([l], l.device_id in ^device_ids) + |> select([l], l.gaiia_account_id) + |> Repo.all() + |> Enum.uniq() + + total_mrr = + if total_account_ids == [] do + Decimal.new(0) + else + BillingSubscription + |> join(:inner, [bs], a in Account, on: bs.account_gaiia_id == a.gaiia_id and bs.organization_id == a.organization_id) + |> where([bs, a], a.id in ^total_account_ids) + |> where([bs], bs.status == "ACTIVE") + |> Repo.aggregate(:sum, :mrr_amount) || Decimal.new(0) + end + + %{ + subscriber_count: length(total_account_ids), + mrr: total_mrr, + devices: devices_with_names + } + end end diff --git a/lib/towerops/gaiia/device_subscriber_link.ex b/lib/towerops/gaiia/device_subscriber_link.ex new file mode 100644 index 00000000..7678876e --- /dev/null +++ b/lib/towerops/gaiia/device_subscriber_link.ex @@ -0,0 +1,50 @@ +defmodule Towerops.Gaiia.DeviceSubscriberLink do + @moduledoc """ + Schema for materialized subscriber-to-device (AP) mappings. + + Links a Gaiia subscriber account to a TowerOps device based on + ARP table matching or site-level fallback. + """ + use Ecto.Schema + + import Ecto.Changeset + + @primary_key {:id, :binary_id, autogenerate: true} + @foreign_key_type :binary_id + + schema "device_subscriber_links" do + field :match_method, :string + field :confidence, :string + field :subscriber_ip, :string + field :subscriber_mac, :string + + belongs_to :organization, Towerops.Organizations.Organization + belongs_to :device, Towerops.Devices.Device + belongs_to :gaiia_account, Towerops.Gaiia.Account + belongs_to :gaiia_inventory_item, Towerops.Gaiia.InventoryItem + + timestamps(type: :utc_datetime) + end + + def changeset(link, attrs) do + link + |> cast(attrs, [ + :organization_id, + :device_id, + :gaiia_account_id, + :gaiia_inventory_item_id, + :match_method, + :confidence, + :subscriber_ip, + :subscriber_mac + ]) + |> validate_required([:organization_id, :device_id, :gaiia_account_id, :match_method, :confidence]) + |> validate_inclusion(:match_method, ~w(arp_ip arp_mac subnet site_fallback)) + |> validate_inclusion(:confidence, ~w(high medium low)) + |> unique_constraint([:device_id, :gaiia_account_id]) + |> foreign_key_constraint(:organization_id) + |> foreign_key_constraint(:device_id) + |> foreign_key_constraint(:gaiia_account_id) + |> foreign_key_constraint(:gaiia_inventory_item_id) + end +end diff --git a/lib/towerops/gaiia/subscriber_matching.ex b/lib/towerops/gaiia/subscriber_matching.ex new file mode 100644 index 00000000..94a47d9b --- /dev/null +++ b/lib/towerops/gaiia/subscriber_matching.ex @@ -0,0 +1,296 @@ +defmodule Towerops.Gaiia.SubscriberMatching do + @moduledoc """ + Matches Gaiia subscriber accounts to TowerOps devices (APs) using + ARP table entries and site-level fallback. + + Builds a materialized `device_subscriber_links` table that maps + each subscriber to the AP(s) serving them. + """ + + import Ecto.Query + + alias Towerops.Devices.Device + alias Towerops.Gaiia.Account + alias Towerops.Gaiia.DeviceSubscriberLink + alias Towerops.Gaiia.InventoryItem + alias Towerops.Gaiia.NetworkSite + alias Towerops.Integrations + alias Towerops.Repo + alias Towerops.Snmp.ArpEntry + + require Logger + + @doc """ + Rebuilds all device-subscriber links for an organization. + Deletes existing links and re-computes from ARP + inventory data. + """ + def refresh_device_links(organization_id) do + unless has_gaiia_integration?(organization_id) do + Logger.debug("Skipping subscriber matching for org #{organization_id} — no Gaiia integration") + :ok + else + do_refresh(organization_id) + end + end + + @doc """ + Lightweight refresh for a single device after ARP poll. + Rebuilds links only for the given device. + """ + def refresh_links_for_device(device_id) do + device = Repo.get(Device, device_id) + + if device do + unless has_gaiia_integration?(device.organization_id) do + :ok + else + do_refresh_for_device(device) + end + end + end + + # --- Private --- + + defp has_gaiia_integration?(organization_id) do + case Integrations.get_integration(organization_id, "gaiia") do + {:ok, _integration} -> true + {:error, :not_found} -> false + end + end + + defp do_refresh(organization_id) do + now = DateTime.truncate(DateTime.utc_now(), :second) + + # Load all inventory items with assigned accounts + items = load_assigned_items(organization_id) + # Load all ARP entries for this org's devices + arp_entries = load_arp_entries(organization_id) + + # Build lookup maps + arp_by_ip = Map.new(arp_entries, fn e -> {e.ip_address, e} end) + arp_by_mac = Map.new(arp_entries, fn e -> {normalize_mac(e.mac_address), e} end) + + # Phase 1: ARP IP match + {ip_matched, ip_links} = match_by_arp_ip(items, arp_by_ip, organization_id, now) + + # Phase 2: ARP MAC match (remaining items) + remaining_after_ip = Enum.reject(items, &MapSet.member?(ip_matched, &1.id)) + {mac_matched, mac_links} = match_by_arp_mac(remaining_after_ip, arp_by_mac, organization_id, now) + + # Phase 3: Site fallback (remaining items) + matched_ids = MapSet.union(ip_matched, mac_matched) + remaining = Enum.reject(items, &MapSet.member?(matched_ids, &1.id)) + site_links = match_by_site_fallback(remaining, organization_id, now) + + all_links = ip_links ++ mac_links ++ site_links + + # Deduplicate by {device_id, gaiia_account_id} — first match wins (highest priority) + unique_links = + all_links + |> Enum.uniq_by(fn l -> {l.device_id, l.gaiia_account_id} end) + + # Transaction: delete old + insert new + Repo.transaction(fn -> + from(d in DeviceSubscriberLink, where: d.organization_id == ^organization_id) + |> Repo.delete_all() + + if unique_links != [] do + Repo.insert_all(DeviceSubscriberLink, unique_links) + end + end) + + Logger.info( + "Subscriber matching for org #{organization_id}: " <> + "#{length(unique_links)}/#{length(items)} items matched " <> + "(#{length(ip_links)} ARP IP, #{length(mac_links)} ARP MAC, #{length(site_links)} site fallback)" + ) + + :ok + end + + defp do_refresh_for_device(device) do + now = DateTime.truncate(DateTime.utc_now(), :second) + org_id = device.organization_id + + items = load_assigned_items(org_id) + arp_entries = load_arp_entries_for_device(device.id) + + arp_by_ip = Map.new(arp_entries, fn e -> {e.ip_address, e} end) + arp_by_mac = Map.new(arp_entries, fn e -> {normalize_mac(e.mac_address), e} end) + + # Resolve account IDs from gaiia_id + account_lookup = load_account_id_lookup(org_id) + + links = + Enum.flat_map(items, fn item -> + cond do + item.ip_address && Map.has_key?(arp_by_ip, item.ip_address) -> + case Map.get(account_lookup, item.assigned_account_gaiia_id) do + nil -> [] + account_id -> + [build_link(org_id, device.id, account_id, item, "arp_ip", "high", item.ip_address, nil, now)] + end + + item.mac_address && Map.has_key?(arp_by_mac, normalize_mac(item.mac_address)) -> + case Map.get(account_lookup, item.assigned_account_gaiia_id) do + nil -> [] + account_id -> + [build_link(org_id, device.id, account_id, item, "arp_mac", "high", nil, normalize_mac(item.mac_address), now)] + end + + true -> + [] + end + end) + |> Enum.uniq_by(fn l -> {l.device_id, l.gaiia_account_id} end) + + Repo.transaction(fn -> + from(d in DeviceSubscriberLink, where: d.device_id == ^device.id) + |> Repo.delete_all() + + if links != [] do + Repo.insert_all(DeviceSubscriberLink, links) + end + end) + + :ok + end + + defp load_assigned_items(organization_id) do + InventoryItem + |> where(organization_id: ^organization_id) + |> where([ii], not is_nil(ii.assigned_account_gaiia_id)) + |> Repo.all() + end + + defp load_arp_entries(organization_id) do + ArpEntry + |> join(:inner, [a], d in Device, on: a.device_id == d.id) + |> where([a, d], d.organization_id == ^organization_id) + |> Repo.all() + end + + defp load_arp_entries_for_device(device_id) do + ArpEntry + |> where(device_id: ^device_id) + |> Repo.all() + end + + defp load_account_id_lookup(organization_id) do + Account + |> where(organization_id: ^organization_id) + |> select([a], {a.gaiia_id, a.id}) + |> Repo.all() + |> Map.new() + end + + defp match_by_arp_ip(items, arp_by_ip, org_id, now) do + account_lookup = load_account_id_lookup(org_id) + + {matched_ids, links} = + Enum.reduce(items, {MapSet.new(), []}, fn item, {matched, acc} -> + if item.ip_address && Map.has_key?(arp_by_ip, item.ip_address) do + arp = Map.get(arp_by_ip, item.ip_address) + case Map.get(account_lookup, item.assigned_account_gaiia_id) do + nil -> + {matched, acc} + account_id -> + link = build_link(org_id, arp.device_id, account_id, item, "arp_ip", "high", item.ip_address, nil, now) + {MapSet.put(matched, item.id), [link | acc]} + end + else + {matched, acc} + end + end) + + {matched_ids, Enum.reverse(links)} + end + + defp match_by_arp_mac(items, arp_by_mac, org_id, now) do + account_lookup = load_account_id_lookup(org_id) + + {matched_ids, links} = + Enum.reduce(items, {MapSet.new(), []}, fn item, {matched, acc} -> + normalized = normalize_mac(item.mac_address) + if normalized && Map.has_key?(arp_by_mac, normalized) do + arp = Map.get(arp_by_mac, normalized) + case Map.get(account_lookup, item.assigned_account_gaiia_id) do + nil -> + {matched, acc} + account_id -> + link = build_link(org_id, arp.device_id, account_id, item, "arp_mac", "high", nil, normalized, now) + {MapSet.put(matched, item.id), [link | acc]} + end + else + {matched, acc} + end + end) + + {matched_ids, Enum.reverse(links)} + end + + defp match_by_site_fallback(items, org_id, now) do + account_lookup = load_account_id_lookup(org_id) + + # Load network site -> TowerOps site mappings + site_mappings = + NetworkSite + |> where(organization_id: ^org_id) + |> where([ns], not is_nil(ns.site_id)) + |> select([ns], {ns.gaiia_id, ns.site_id}) + |> Repo.all() + |> Map.new() + + # Load devices per site + devices_by_site = + Device + |> where(organization_id: ^org_id) + |> where([d], not is_nil(d.site_id)) + |> select([d], {d.site_id, d.id}) + |> Repo.all() + |> Enum.group_by(&elem(&1, 0), &elem(&1, 1)) + + Enum.flat_map(items, fn item -> + with gaiia_site_id when not is_nil(gaiia_site_id) <- item.assigned_network_site_gaiia_id, + site_id when not is_nil(site_id) <- Map.get(site_mappings, gaiia_site_id), + device_ids when device_ids != nil <- Map.get(devices_by_site, site_id), + account_id when not is_nil(account_id) <- Map.get(account_lookup, item.assigned_account_gaiia_id) do + Enum.map(device_ids, fn device_id -> + build_link(org_id, device_id, account_id, item, "site_fallback", "medium", nil, nil, now) + end) + else + _ -> [] + end + end) + end + + defp build_link(org_id, device_id, account_id, item, method, confidence, ip, mac, now) do + %{ + id: Ecto.UUID.generate(), + organization_id: org_id, + device_id: device_id, + gaiia_account_id: account_id, + gaiia_inventory_item_id: item.id, + match_method: method, + confidence: confidence, + subscriber_ip: ip, + subscriber_mac: mac, + inserted_at: now, + updated_at: now + } + end + + defp normalize_mac(nil), do: nil + + defp normalize_mac(mac) do + mac + |> String.downcase() + |> String.replace(~r/[^0-9a-f]/, "") + |> case do + <> -> + "#{a}:#{b}:#{c}:#{d}:#{e}:#{f}" + _ -> + nil + end + end +end diff --git a/lib/towerops/gaiia/sync.ex b/lib/towerops/gaiia/sync.ex index 5c65e04c..6b5d44ba 100644 --- a/lib/towerops/gaiia/sync.ex +++ b/lib/towerops/gaiia/sync.ex @@ -39,6 +39,7 @@ defmodule Towerops.Gaiia.Sync do Integrations.update_sync_status(integration, "success", message) persist_billing_totals(integration) Gaiia.SiteAggregation.compute_and_store(org_id) + Towerops.Gaiia.SubscriberMatching.refresh_device_links(org_id) {:ok, %{ diff --git a/lib/towerops/workers/device_monitor_worker.ex b/lib/towerops/workers/device_monitor_worker.ex index 7b0619ed..5422b813 100644 --- a/lib/towerops/workers/device_monitor_worker.ex +++ b/lib/towerops/workers/device_monitor_worker.ex @@ -234,10 +234,19 @@ defmodule Towerops.Workers.DeviceMonitorWorker do end defp get_down_alert_message(device) do - if device.snmp_enabled do - "Device is not responding to SNMP" - else - "Device is not responding to ping" + base_message = + if device.snmp_enabled do + "Device is not responding to SNMP" + else + "Device is not responding to ping" + end + + case Towerops.Gaiia.get_device_impact(device.id) do + %{subscriber_count: count, mrr: mrr} when count > 0 -> + "#{device.name} is down — #{count} subscribers affected ($#{Decimal.round(mrr, 2)}/mo at risk)" + + _ -> + base_message end end diff --git a/lib/towerops/workers/device_poller_worker.ex b/lib/towerops/workers/device_poller_worker.ex index 5c7e561e..7d4c1124 100644 --- a/lib/towerops/workers/device_poller_worker.ex +++ b/lib/towerops/workers/device_poller_worker.ex @@ -356,6 +356,9 @@ defmodule Towerops.Workers.DevicePollerWorker do "device:#{device.id}", {:arp_updated, device.id} ) + + # Refresh subscriber-to-device links based on new ARP data + Towerops.Gaiia.SubscriberMatching.refresh_links_for_device(device.id) rescue error -> Logger.error("Error polling ARP table for #{device.name}: #{inspect(error)}\n#{Exception.format_stacktrace()}") diff --git a/priv/repo/migrations/20260216170000_create_device_subscriber_links.exs b/priv/repo/migrations/20260216170000_create_device_subscriber_links.exs new file mode 100644 index 00000000..23989e1c --- /dev/null +++ b/priv/repo/migrations/20260216170000_create_device_subscriber_links.exs @@ -0,0 +1,24 @@ +defmodule Towerops.Repo.Migrations.CreateDeviceSubscriberLinks do + use Ecto.Migration + + def change do + create table(:device_subscriber_links, primary_key: false) do + add :id, :binary_id, primary_key: true + add :organization_id, references(:organizations, type: :binary_id, on_delete: :delete_all), null: false + add :device_id, references(:devices, type: :binary_id, on_delete: :delete_all), null: false + add :gaiia_account_id, references(:gaiia_accounts, type: :binary_id, on_delete: :delete_all), null: false + add :gaiia_inventory_item_id, references(:gaiia_inventory_items, type: :binary_id, on_delete: :nilify_all) + add :match_method, :string, null: false + add :confidence, :string, null: false + add :subscriber_ip, :string + add :subscriber_mac, :string + + timestamps(type: :utc_datetime) + end + + create unique_index(:device_subscriber_links, [:device_id, :gaiia_account_id]) + create index(:device_subscriber_links, [:device_id]) + create index(:device_subscriber_links, [:gaiia_account_id]) + create index(:device_subscriber_links, [:organization_id]) + end +end