Add subscriber-to-AP matching via ARP + Gaiia integration

Matches Gaiia subscriber accounts to TowerOps devices using:
- ARP table IP matching (high confidence)
- ARP table MAC matching (high confidence)
- Site CIDR fallback (medium confidence)

Enables subscriber impact display on alerts and device pages.

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

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

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::binary-2, b::binary-2, c::binary-2, d::binary-2, e::binary-2, f::binary-2>> ->
+        "#{a}:#{b}:#{c}:#{d}:#{e}:#{f}"
+      _ ->
+        nil
+    end
+  end
+end

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,
        %{

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
 

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()}")

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