From c8e398ea05e9a53207e203e6b53698ab4f56b75f Mon Sep 17 00:00:00 2001
From: Graham McIntie <graham@mcintire.me>
Date: Sat, 14 Feb 2026 17:11:30 -0600
Subject: [PATCH] Add NetBox sync worker + improve Gaiia error messages

- NetBox.Sync: pull devices/sites from NetBox, match by name/IP, upsert
- NetBoxSyncWorker: Oban cron every 30min for enabled NetBox integrations
- Gaiia sync: show actual error type instead of generic 'unexpected error'
---
 config/runtime.exs                         |   2 +
 lib/towerops/gaiia/sync.ex                 |  15 +-
 lib/towerops/netbox/sync.ex                | 261 +++++++++++++++++++++
 lib/towerops/workers/netbox_sync_worker.ex |  62 +++++
 4 files changed, 339 insertions(+), 1 deletion(-)
 create mode 100644 lib/towerops/netbox/sync.ex
 create mode 100644 lib/towerops/workers/netbox_sync_worker.ex

diff --git a/config/runtime.exs b/config/runtime.exs
index 358a0ed2..1cf10d4a 100644
--- a/config/runtime.exs
+++ b/config/runtime.exs
@@ -202,6 +202,8 @@ if config_env() == :prod do
          {"30 2 * * *", Towerops.Workers.PreseemBaselineWorker},
          # Sync Gaiia data every 15 minutes
          {"*/15 * * * *", Towerops.Workers.GaiiaSyncWorker},
+         # Sync NetBox data every 30 minutes
+         {"*/30 * * * *", Towerops.Workers.NetBoxSyncWorker},
          # Device health insights nightly at 3:30 AM
          {"30 3 * * *", Towerops.Workers.DeviceHealthInsightWorker},
          # System insights (agent offline detection) every 5 minutes
diff --git a/lib/towerops/gaiia/sync.ex b/lib/towerops/gaiia/sync.ex
index b7ca9830..bd65be57 100644
--- a/lib/towerops/gaiia/sync.ex
+++ b/lib/towerops/gaiia/sync.ex
@@ -188,5 +188,18 @@ defmodule Towerops.Gaiia.Sync do
   defp humanize_sync_error(%Req.TransportError{reason: reason}), do: "Connection error: #{reason}"
 
   defp humanize_sync_error(reason) when is_binary(reason), do: reason
-  defp humanize_sync_error(_reason), do: "An unexpected error occurred during sync"
+
+  defp humanize_sync_error({:graphql_errors, errors}) when is_list(errors) do
+    messages = Enum.map_join(errors, "; ", &(&1["message"] || inspect(&1)))
+    "GraphQL errors: #{messages}"
+  end
+
+  defp humanize_sync_error({:rate_limited, retry_after}), do: "Rate limited by Gaiia — retry after #{retry_after}s"
+
+  defp humanize_sync_error({:unexpected_status, status}), do: "Gaiia returned unexpected HTTP #{status}"
+
+  defp humanize_sync_error(reason) do
+    Logger.warning("Gaiia sync failed with unexpected error: #{inspect(reason)}")
+    "An unexpected error occurred during sync: #{inspect(reason)}"
+  end
 end
diff --git a/lib/towerops/netbox/sync.ex b/lib/towerops/netbox/sync.ex
new file mode 100644
index 00000000..09b32f7f
--- /dev/null
+++ b/lib/towerops/netbox/sync.ex
@@ -0,0 +1,261 @@
+defmodule Towerops.NetBox.Sync do
+  @moduledoc """
+  Orchestrates syncing data from NetBox into the local database.
+
+  Pulls devices and sites from NetBox via the REST API, upserts them
+  into local tables, and updates integration sync status.
+  """
+
+  alias Towerops.Devices
+  alias Towerops.Integrations
+  alias Towerops.NetBox.Client
+  alias Towerops.Sites
+
+  require Logger
+
+  @doc """
+  Main entry point: syncs enabled entity types for the given integration.
+
+  Returns `{:ok, %{devices: n, sites: n}}` or `{:error, reason}`.
+  """
+  def sync_organization(%Integrations.Integration{} = integration) do
+    url = integration.credentials["url"]
+    token = integration.credentials["api_token"]
+    org_id = integration.organization_id
+
+    sync_devices? = integration.credentials["sync_devices"] in [true, "true"]
+    sync_sites? = integration.credentials["sync_sites"] in [true, "true"]
+
+    results = %{devices: 0, sites: 0}
+
+    with {:ok, results} <- maybe_sync_sites(results, sync_sites?, url, token, org_id, integration),
+         {:ok, results} <- maybe_sync_devices(results, sync_devices?, url, token, org_id, integration) do
+      message =
+        "Synced #{results.sites} sites, #{results.devices} devices from NetBox"
+
+      Integrations.update_sync_status(integration, "success", message)
+      {:ok, results}
+    else
+      {:error, reason} ->
+        Integrations.update_sync_status(integration, "failed", humanize_sync_error(reason))
+        {:error, reason}
+    end
+  end
+
+  defp maybe_sync_sites(results, false, _url, _token, _org_id, _integration), do: {:ok, results}
+
+  defp maybe_sync_sites(results, true, url, token, org_id, integration) do
+    params = build_site_params(integration.credentials)
+
+    case Client.list_sites(url, token, params) do
+      {:ok, %{"results" => sites}} ->
+        count = upsert_sites(org_id, sites)
+        Logger.info("NetBox sync: upserted #{count} sites for org #{org_id}")
+        {:ok, %{results | sites: count}}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
+  defp maybe_sync_devices(results, false, _url, _token, _org_id, _integration), do: {:ok, results}
+
+  defp maybe_sync_devices(results, true, url, token, org_id, integration) do
+    params = build_device_params(integration.credentials)
+
+    case Client.list_devices(url, token, params) do
+      {:ok, %{"results" => devices}} ->
+        count = upsert_devices(org_id, devices)
+        Logger.info("NetBox sync: upserted #{count} devices for org #{org_id}")
+        {:ok, %{results | devices: count}}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
+  # --- Sites ---
+
+  defp upsert_sites(org_id, netbox_sites) do
+    existing = Sites.list_organization_sites(org_id)
+    site_by_name = Map.new(existing, &{String.downcase(&1.name), &1})
+
+    Enum.reduce(netbox_sites, 0, fn nb_site, count ->
+      case upsert_site(org_id, nb_site, site_by_name) do
+        {:ok, _} ->
+          count + 1
+
+        {:error, reason} ->
+          Logger.warning("NetBox sync: failed to upsert site #{nb_site["name"]}: #{inspect(reason)}")
+          count
+      end
+    end)
+  end
+
+  defp upsert_site(org_id, nb_site, site_by_name) do
+    name = nb_site["name"] || ""
+    key = String.downcase(name)
+    latitude = nb_site |> get_in(["latitude"]) |> parse_decimal()
+    longitude = nb_site |> get_in(["longitude"]) |> parse_decimal()
+
+    attrs = %{
+      "name" => name,
+      "latitude" => latitude,
+      "longitude" => longitude,
+      "organization_id" => org_id
+    }
+
+    case Map.get(site_by_name, key) do
+      nil -> Sites.create_site(attrs)
+      existing -> Sites.update_site(existing, attrs)
+    end
+  end
+
+  # --- Devices ---
+
+  defp upsert_devices(org_id, netbox_devices) do
+    existing = Devices.list_organization_devices(org_id)
+    device_by_name = Map.new(existing, &{String.downcase(&1.name), &1})
+
+    device_by_ip =
+      existing
+      |> Enum.filter(& &1.ip_address)
+      |> Map.new(&{&1.ip_address, &1})
+
+    # Build site lookup for assignment
+    sites = Sites.list_organization_sites(org_id)
+    site_by_name = Map.new(sites, &{String.downcase(&1.name), &1})
+
+    Enum.reduce(netbox_devices, 0, fn nb_device, count ->
+      case upsert_device(org_id, nb_device, device_by_name, device_by_ip, site_by_name) do
+        {:ok, _} ->
+          count + 1
+
+        {:error, reason} ->
+          Logger.warning("NetBox sync: failed to upsert device #{nb_device["name"]}: #{inspect(reason)}")
+          count
+      end
+    end)
+  end
+
+  defp upsert_device(org_id, nb_device, device_by_name, device_by_ip, site_by_name) do
+    name = nb_device["name"] || ""
+    ip = extract_ip(nb_device["primary_ip4"] || nb_device["primary_ip"])
+    role = get_in(nb_device, ["role", "name"]) || get_in(nb_device, ["device_role", "name"])
+    status = get_in(nb_device, ["status", "value"]) || nb_device["status"]
+    serial = nb_device["serial"]
+    site_name = get_in(nb_device, ["site", "name"])
+
+    site_id =
+      if site_name do
+        case Map.get(site_by_name, String.downcase(site_name)) do
+          nil -> nil
+          site -> site.id
+        end
+      end
+
+    attrs = %{
+      "name" => name,
+      "ip_address" => ip,
+      "device_role" => role,
+      "status" => map_status(status),
+      "serial_number" => serial,
+      "site_id" => site_id,
+      "organization_id" => org_id
+    }
+
+    # Match by name first, then by IP
+    existing =
+      Map.get(device_by_name, String.downcase(name)) ||
+        (ip && Map.get(device_by_ip, ip))
+
+    case existing do
+      nil -> Devices.create_device(attrs)
+      device -> Devices.update_device(device, attrs)
+    end
+  end
+
+  # --- Helpers ---
+
+  defp extract_ip(nil), do: nil
+
+  defp extract_ip(%{"address" => addr}) when is_binary(addr) do
+    # NetBox returns "10.0.0.1/24" format
+    addr |> String.split("/") |> List.first()
+  end
+
+  defp extract_ip(_), do: nil
+
+  defp map_status("active"), do: "up"
+  defp map_status("offline"), do: "down"
+  defp map_status("planned"), do: "pending"
+  defp map_status("staged"), do: "pending"
+  defp map_status("decommissioning"), do: "down"
+  defp map_status(other), do: other
+
+  defp parse_decimal(nil), do: nil
+  defp parse_decimal(val) when is_float(val), do: Decimal.from_float(val)
+
+  defp parse_decimal(val) when is_binary(val) do
+    case Decimal.parse(val) do
+      {dec, _} -> dec
+      :error -> nil
+    end
+  end
+
+  defp parse_decimal(val) when is_integer(val), do: Decimal.new(val)
+  defp parse_decimal(_), do: nil
+
+  defp build_device_params(credentials) do
+    params = %{}
+
+    params =
+      case credentials["device_role_filter"] do
+        nil -> params
+        "" -> params
+        role -> Map.put(params, "role", role)
+      end
+
+    params =
+      case credentials["site_filter"] do
+        nil -> params
+        "" -> params
+        site -> Map.put(params, "site", site)
+      end
+
+    case credentials["tag_filter"] do
+      nil -> params
+      "" -> params
+      tag -> Map.put(params, "tag", tag)
+    end
+  end
+
+  defp build_site_params(credentials) do
+    case credentials["tag_filter"] do
+      nil -> %{}
+      "" -> %{}
+      tag -> %{"tag" => tag}
+    end
+  end
+
+  defp humanize_sync_error(:unauthorized), do: "Authentication failed — check your NetBox API token"
+
+  defp humanize_sync_error(:forbidden), do: "Access denied — your API token may not have sufficient permissions"
+
+  defp humanize_sync_error(:not_found), do: "API endpoint not found — check your NetBox URL"
+
+  defp humanize_sync_error(:timeout), do: "Connection timed out — NetBox may be temporarily unavailable"
+
+  defp humanize_sync_error(%Req.TransportError{reason: :econnrefused}), do: "Connection refused — unable to reach NetBox"
+
+  defp humanize_sync_error(%Req.TransportError{reason: reason}), do: "Connection error: #{reason}"
+
+  defp humanize_sync_error(reason) when is_binary(reason), do: reason
+
+  defp humanize_sync_error({:unexpected_status, status}), do: "NetBox returned unexpected HTTP #{status}"
+
+  defp humanize_sync_error(reason) do
+    Logger.warning("NetBox sync failed with unexpected error: #{inspect(reason)}")
+    "An unexpected error occurred during sync: #{inspect(reason)}"
+  end
+end
diff --git a/lib/towerops/workers/netbox_sync_worker.ex b/lib/towerops/workers/netbox_sync_worker.ex
new file mode 100644
index 00000000..dcd39d94
--- /dev/null
+++ b/lib/towerops/workers/netbox_sync_worker.ex
@@ -0,0 +1,62 @@
+defmodule Towerops.Workers.NetBoxSyncWorker do
+  @moduledoc """
+  Oban cron worker that syncs NetBox data for all enabled integrations.
+
+  Runs every 30 minutes. For each org with an enabled NetBox integration,
+  checks if enough time has elapsed since the last sync (based on the
+  integration's sync_interval_minutes), then calls NetBox.Sync.sync_organization/1.
+  """
+  use Oban.Worker, queue: :maintenance
+
+  alias Towerops.Integrations
+  alias Towerops.NetBox.Sync
+
+  require Logger
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{}) do
+    integrations = Integrations.list_enabled_integrations("netbox")
+
+    results = Enum.map(integrations, &sync_integration/1)
+
+    synced = Enum.count(results, &match?({:ok, _}, &1))
+    failed = Enum.count(results, &match?({:error, _}, &1))
+    skipped = Enum.count(results, &(&1 == :skipped))
+
+    if synced > 0 or failed > 0 do
+      Logger.info("NetBox sync batch: #{synced} synced, #{failed} failed, #{skipped} skipped")
+    end
+
+    :ok
+  end
+
+  defp sync_integration(integration) do
+    if should_sync?(integration) do
+      case Sync.sync_organization(integration) do
+        {:ok, result} ->
+          Logger.info("NetBox sync completed for org #{integration.organization_id}: #{inspect(result)}")
+
+          {:ok, result}
+
+        {:error, reason} ->
+          Logger.error("NetBox sync failed for org #{integration.organization_id}: #{inspect(reason)}")
+
+          {:error, reason}
+      end
+    else
+      :skipped
+    end
+  end
+
+  defp should_sync?(integration) do
+    case integration.last_synced_at do
+      nil ->
+        true
+
+      last_synced_at ->
+        interval_seconds = (integration.sync_interval_minutes || 30) * 60
+        elapsed = DateTime.diff(DateTime.utc_now(), last_synced_at, :second)
+        elapsed >= interval_seconds
+    end
+  end
+end