From 2e9f1660d07424822fe181e2b152012fecfd52a4 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Fri, 13 Feb 2026 10:52:32 -0600 Subject: [PATCH] stagger preseem sync across organizations Replace the thundering-herd cron job with a dispatcher that enqueues individual per-integration Oban jobs staggered across the 10-minute window using PollingOffset. Show last synced timestamp in user's local timezone via the <.timestamp> component. --- lib/towerops/integrations.ex | 7 + lib/towerops/preseem/client.ex | 52 ++++-- lib/towerops/preseem/sync.ex | 45 +++--- lib/towerops/workers/preseem_sync_worker.ex | 75 ++++++--- .../live/org/integrations_live.ex | 1 + .../live/org/integrations_live.html.heex | 10 +- test/towerops/preseem/client_test.exs | 103 +++++++++--- test/towerops/preseem/device_matcher_test.exs | 56 +++---- test/towerops/preseem/sync_test.exs | 148 +++++++----------- .../workers/preseem_sync_worker_test.exs | 117 +++++++++----- .../live/org/integrations_live_test.exs | 5 +- 11 files changed, 368 insertions(+), 251 deletions(-) diff --git a/lib/towerops/integrations.ex b/lib/towerops/integrations.ex index a025c2ec..0aa35cc0 100644 --- a/lib/towerops/integrations.ex +++ b/lib/towerops/integrations.ex @@ -54,6 +54,13 @@ defmodule Towerops.Integrations do Integration.changeset(integration, attrs) end + def get_integration_by_id(id) do + case Repo.get(Integration, id) do + nil -> {:error, :not_found} + integration -> {:ok, integration} + end + end + def list_enabled_integrations(provider) do Integration |> where(provider: ^provider, enabled: true) diff --git a/lib/towerops/preseem/client.ex b/lib/towerops/preseem/client.ex index ae8200b6..9f67ff18 100644 --- a/lib/towerops/preseem/client.ex +++ b/lib/towerops/preseem/client.ex @@ -3,6 +3,7 @@ defmodule Towerops.Preseem.Client do HTTP client for the Preseem Model API. Uses Req with built-in test support via `Req.Test`. + Auth: HTTP Basic (API key as username, empty password). """ require Logger @@ -10,12 +11,12 @@ defmodule Towerops.Preseem.Client do @base_url "https://api.preseem.com/model/v1" @doc """ - Tests the connection to the Preseem API by fetching account info. + Tests the connection to the Preseem API by fetching a page of access points. Returns `{:ok, body}` on success, `{:error, reason}` on failure. """ def test_connection(api_key) do - case request(:get, "#{@base_url}/account", api_key) do + case request(:get, "#{@base_url}/access_points?limit=100", api_key) do {:ok, %{status: status, body: body}} when status in 200..299 -> {:ok, body} @@ -36,14 +37,15 @@ defmodule Towerops.Preseem.Client do end @doc """ - Lists all access points from the Preseem API. + Lists all access points with scores from the Preseem API. - Returns `{:ok, [access_point]}` on success. + Fetches from `/access_point_scores` which returns rich data per AP. + Returns `{:ok, [access_point_score]}` on success. """ def list_access_points(api_key) do - case request(:get, "#{@base_url}/access_points", api_key) do - {:ok, %{status: status, body: %{"data" => data}}} when status in 200..299 -> - {:ok, data} + case request(:get, "#{@base_url}/access_point_scores", api_key) do + {:ok, %{status: status, body: body}} when status in 200..299 -> + {:ok, normalize_scores_response(body)} {:ok, %{status: 401}} -> {:error, :unauthorized} @@ -60,14 +62,27 @@ defmodule Towerops.Preseem.Client do end @doc """ - Fetches metrics for a specific access point. + Lists access points from the basic `/access_points` endpoint with pagination. - Returns `{:ok, metrics_map}` on success. + Returns `{:ok, [access_point]}` on success. """ - def get_access_point_metrics(api_key, ap_id) do - case request(:get, "#{@base_url}/access_points/#{ap_id}/metrics", api_key) do - {:ok, %{status: status, body: %{"data" => data}}} when status in 200..299 -> - {:ok, data} + def list_access_points_basic(api_key) do + fetch_all_pages(api_key, "#{@base_url}/access_points?limit=500", []) + end + + defp fetch_all_pages(api_key, url, acc) do + case request(:get, url, api_key) do + {:ok, %{status: status, body: %{"data" => data, "paginator" => paginator}}} + when status in 200..299 -> + all = acc ++ data + + if paginator["page"] < paginator["page_count"] do + next_page = paginator["page"] + 1 + next_url = "#{@base_url}/access_points?limit=500&page=#{next_page}" + fetch_all_pages(api_key, next_url, all) + else + {:ok, all} + end {:ok, %{status: 401}} -> {:error, :unauthorized} @@ -83,11 +98,20 @@ defmodule Towerops.Preseem.Client do end end + # The scores endpoint may return an array, a wrapped object with "data" key, + # or a single object. Normalize to a list. + defp normalize_scores_response(body) when is_list(body), do: body + + defp normalize_scores_response(%{"data" => data}) when is_list(data), do: data + + defp normalize_scores_response(body) when is_map(body), do: [body] + defp request(method, url, api_key) do opts = [ method: method, url: url, - headers: [{"authorization", "Bearer #{api_key}"}, {"accept", "application/json"}] + auth: {:basic, "#{api_key}:"}, + headers: [{"accept", "application/json"}] ] # Only use Req.Test plug in test environment diff --git a/lib/towerops/preseem/sync.ex b/lib/towerops/preseem/sync.ex index cab2eacc..35180c10 100644 --- a/lib/towerops/preseem/sync.ex +++ b/lib/towerops/preseem/sync.ex @@ -2,7 +2,7 @@ defmodule Towerops.Preseem.Sync do @moduledoc """ Orchestrates syncing data from the Preseem API into the local database. - Pulls access points (and optional metrics) from Preseem, upserts them + Pulls access point scores from Preseem, upserts them into `preseem_access_points`, and logs each sync operation. """ @@ -10,7 +10,6 @@ defmodule Towerops.Preseem.Sync do alias Towerops.Preseem.AccessPoint alias Towerops.Preseem.Client alias Towerops.Preseem.DeviceMatcher - alias Towerops.Preseem.SubscriberMetric alias Towerops.Preseem.SyncLog alias Towerops.Repo @@ -19,8 +18,8 @@ defmodule Towerops.Preseem.Sync do @doc """ Main entry point: syncs all access points for the given integration. - Fetches APs from the Preseem API, upserts them, optionally inserts - metrics, writes a sync log, and updates the integration's sync status. + Fetches AP scores from the Preseem API, upserts them, writes a sync log, + and updates the integration's sync status. Returns `{:ok, %{synced: count, matched: count}}` or `{:error, reason}`. """ @@ -41,24 +40,24 @@ defmodule Towerops.Preseem.Sync do @doc """ Upserts a single access point from raw API response data. - Maps API fields to schema fields and uses the unique constraint on + Maps Preseem scores API fields to schema fields and uses the unique constraint on `[:organization_id, :preseem_id]` for conflict resolution. """ def upsert_access_point(organization_id, ap_data) do attrs = %{ organization_id: organization_id, - preseem_id: ap_data["id"], + preseem_id: ap_data["element_uuid"], name: ap_data["name"], - mac_address: ap_data["mac_address"], - ip_address: ap_data["ip_address"], - model: ap_data["model"], + mac_address: ap_data["system_mac_address"], + ip_address: ap_data["management_ip"], + model: ap_data["model_name"], firmware: ap_data["firmware"], - capacity_score: ap_data["capacity_score"], - qoe_score: ap_data["qoe_score"], - rf_score: ap_data["rf_score"], - busy_hours: ap_data["busy_hours"], - airtime_utilization: ap_data["airtime_utilization"], - subscriber_count: ap_data["subscriber_count"], + capacity_score: cast_to_float(ap_data["subscriber_capacity"]), + qoe_score: cast_to_float(ap_data["ap_health_score"]), + rf_score: cast_to_float(ap_data["rf_score"]), + busy_hours: nil, + airtime_utilization: cast_to_float(ap_data["airtime_remaining_tx"]), + subscriber_count: ap_data["connections"], raw_data: ap_data } @@ -76,6 +75,8 @@ defmodule Towerops.Preseem.Sync do Inserts a subscriber metric record for the given access point. """ def insert_metrics(access_point_id, metrics_data) do + alias Towerops.Preseem.SubscriberMetric + recorded_at = case metrics_data["recorded_at"] do nil -> DateTime.truncate(DateTime.utc_now(), :second) @@ -102,8 +103,7 @@ defmodule Towerops.Preseem.Sync do synced_count = Enum.reduce(ap_list, 0, fn ap_data, count -> case upsert_access_point(org_id, ap_data) do - {:ok, ap} -> - maybe_insert_metrics(ap, ap_data) + {:ok, _ap} -> count + 1 {:error, changeset} -> @@ -132,13 +132,6 @@ defmodule Towerops.Preseem.Sync do {:error, reason} end - defp maybe_insert_metrics(ap, ap_data) do - case ap_data["metrics"] do - nil -> :ok - metrics when is_map(metrics) -> insert_metrics(ap.id, metrics) - end - end - defp run_device_matching(org_id) do if Code.ensure_loaded?(DeviceMatcher) do case DeviceMatcher.match_unmatched(org_id) do @@ -170,4 +163,8 @@ defmodule Towerops.Preseem.Sync do _ -> DateTime.truncate(DateTime.utc_now(), :second) end end + + defp cast_to_float(nil), do: nil + defp cast_to_float(val) when is_float(val), do: val + defp cast_to_float(val) when is_integer(val), do: val / 1 end diff --git a/lib/towerops/workers/preseem_sync_worker.ex b/lib/towerops/workers/preseem_sync_worker.ex index c3bf7e6d..ec3d7039 100644 --- a/lib/towerops/workers/preseem_sync_worker.ex +++ b/lib/towerops/workers/preseem_sync_worker.ex @@ -1,48 +1,77 @@ defmodule Towerops.Workers.PreseemSyncWorker do @moduledoc """ - Oban cron worker that syncs Preseem data for all enabled integrations. + Oban worker that syncs Preseem data for enabled integrations. - Runs every 10 minutes. For each org with an enabled Preseem integration, - checks if enough time has elapsed since the last sync (based on the - integration's sync_interval_minutes), then calls Preseem.Sync.sync_organization/1. + The cron job (no args) runs every 10 minutes and dispatches individual + per-integration sync jobs staggered across the 600-second window using + `PollingOffset.calculate_offset/2`. This prevents a thundering herd + against the Preseem API. + + Individual jobs receive `%{"integration_id" => id}` and perform the + actual sync via `Preseem.Sync.sync_organization/1`. """ use Oban.Worker, queue: :maintenance alias Towerops.Integrations alias Towerops.Preseem.Sync + alias Towerops.Workers.PollingOffset require Logger + @window_seconds 600 + + # Cron dispatcher: enqueues staggered individual sync jobs @impl Oban.Worker - def perform(%Oban.Job{}) do + def perform(%Oban.Job{args: args}) when args == %{} do integrations = Integrations.list_enabled_integrations("preseem") - results = Enum.map(integrations, &sync_integration/1) + eligible = Enum.filter(integrations, &should_sync?/1) - synced = Enum.count(results, &match?({:ok, _}, &1)) - failed = Enum.count(results, &match?({:error, _}, &1)) - skipped = Enum.count(results, &(&1 == :skipped)) + now = DateTime.utc_now() - if synced > 0 or failed > 0 do - Logger.info("Preseem sync batch: #{synced} synced, #{failed} failed, #{skipped} skipped") + enqueued = + Enum.count(eligible, fn integration -> + offset = PollingOffset.calculate_offset(integration.organization_id, @window_seconds) + scheduled_at = DateTime.add(now, offset, :second) + + case %{"integration_id" => integration.id} + |> new(scheduled_at: scheduled_at, unique: [period: @window_seconds]) + |> Oban.insert() do + {:ok, _job} -> true + {:error, _reason} -> false + end + end) + + if enqueued > 0 do + Logger.info("Preseem sync: dispatched #{enqueued} jobs across #{@window_seconds}s window") end :ok end - defp sync_integration(integration) do - if should_sync?(integration) do - case Sync.sync_organization(integration) do - {:ok, result} -> - Logger.info("Preseem sync completed for org #{integration.organization_id}: #{inspect(result)}") - {:ok, result} + # Individual sync: loads integration by ID and syncs + def perform(%Oban.Job{args: %{"integration_id" => id}}) do + case Integrations.get_integration_by_id(id) do + {:ok, integration} -> + sync_integration(integration) - {:error, reason} -> - Logger.error("Preseem sync failed for org #{integration.organization_id}: #{inspect(reason)}") - {:error, reason} - end - else - :skipped + {:error, :not_found} -> + Logger.warning("Preseem sync: integration #{id} not found, skipping") + {:error, :not_found} + end + end + + defp sync_integration(integration) do + case Sync.sync_organization(integration) do + {:ok, result} -> + Logger.info("Preseem sync completed for org #{integration.organization_id}: #{inspect(result)}") + + :ok + + {:error, reason} -> + Logger.error("Preseem sync failed for org #{integration.organization_id}: #{inspect(reason)}") + + :ok end end diff --git a/lib/towerops_web/live/org/integrations_live.ex b/lib/towerops_web/live/org/integrations_live.ex index 286abf80..be12d7d0 100644 --- a/lib/towerops_web/live/org/integrations_live.ex +++ b/lib/towerops_web/live/org/integrations_live.ex @@ -23,6 +23,7 @@ defmodule ToweropsWeb.Org.IntegrationsLive do {:ok, socket |> assign(:organization, organization) + |> assign(:timezone, socket.assigns.current_scope.timezone) |> assign(:providers, @providers) |> assign(:integrations, integrations) |> assign(:configuring, nil) diff --git a/lib/towerops_web/live/org/integrations_live.html.heex b/lib/towerops_web/live/org/integrations_live.html.heex index 678330b9..a1717ef2 100644 --- a/lib/towerops_web/live/org/integrations_live.html.heex +++ b/lib/towerops_web/live/org/integrations_live.html.heex @@ -51,10 +51,12 @@ <%= if integration.last_synced_at do %> - Last synced: {Calendar.strftime( - integration.last_synced_at, - "%Y-%m-%d %H:%M UTC" - )} + Last synced: + <.timestamp + datetime={integration.last_synced_at} + timezone={@timezone} + format="absolute" + /> <% end %> diff --git a/test/towerops/preseem/client_test.exs b/test/towerops/preseem/client_test.exs index 4c3a773a..8062e999 100644 --- a/test/towerops/preseem/client_test.exs +++ b/test/towerops/preseem/client_test.exs @@ -6,10 +6,16 @@ defmodule Towerops.Preseem.ClientTest do describe "test_connection/1" do test "returns ok when API responds with 200" do Req.Test.stub(Client, fn conn -> - Req.Test.json(conn, %{"status" => "ok", "account" => "test-isp"}) + assert conn.request_path == "/model/v1/access_points" + assert conn.query_string =~ "limit=100" + + Req.Test.json(conn, %{ + "data" => [%{"id" => 1, "name" => "Test AP"}], + "paginator" => %{"page" => 1, "page_count" => 1, "limit" => 100, "total_count" => 1} + }) end) - assert {:ok, %{"status" => "ok"}} = Client.test_connection("valid-api-key") + assert {:ok, _body} = Client.test_connection("valid-api-key") end test "returns error for 401 unauthorized" do @@ -34,19 +40,60 @@ defmodule Towerops.Preseem.ClientTest do end describe "list_access_points/1" do - test "returns list of access points on success" do + test "returns list of access point scores on success" do + Req.Test.stub(Client, fn conn -> + assert conn.request_path == "/model/v1/access_point_scores" + + Req.Test.json(conn, [ + %{ + "element_uuid" => "uuid-1", + "name" => "Tower1-AP1", + "system_mac_address" => "AA:BB:CC:DD:EE:FF", + "management_ip" => "10.0.0.1", + "model_name" => "AF60", + "firmware" => "4.3.2", + "rf_score" => 78.0, + "ap_health_score" => 92.5, + "subscriber_capacity" => 85, + "connections" => 30, + "airtime_remaining_tx" => 55 + }, + %{ + "element_uuid" => "uuid-2", + "name" => "Tower1-AP2", + "management_ip" => "10.0.0.2" + } + ]) + end) + + assert {:ok, [ap1, ap2]} = Client.list_access_points("valid-key") + assert ap1["element_uuid"] == "uuid-1" + assert ap2["element_uuid"] == "uuid-2" + end + + test "handles wrapped response with data key" do Req.Test.stub(Client, fn conn -> Req.Test.json(conn, %{ "data" => [ - %{"id" => "ap-1", "name" => "Tower1-AP1", "ip" => "10.0.0.1"}, - %{"id" => "ap-2", "name" => "Tower1-AP2", "ip" => "10.0.0.2"} + %{"element_uuid" => "uuid-1", "name" => "AP1"} ] }) end) - assert {:ok, [ap1, ap2]} = Client.list_access_points("valid-key") - assert ap1["id"] == "ap-1" - assert ap2["id"] == "ap-2" + assert {:ok, [ap]} = Client.list_access_points("valid-key") + assert ap["element_uuid"] == "uuid-1" + end + + test "handles single object response by wrapping in list" do + Req.Test.stub(Client, fn conn -> + Req.Test.json(conn, %{ + "element_uuid" => "uuid-1", + "name" => "Single AP" + }) + end) + + assert {:ok, [ap]} = Client.list_access_points("valid-key") + assert ap["element_uuid"] == "uuid-1" end test "returns error for 401" do @@ -58,21 +105,37 @@ defmodule Towerops.Preseem.ClientTest do end end - describe "get_access_point_metrics/2" do - test "returns metrics for an AP" do + describe "list_access_points_basic/1" do + test "fetches all pages of access points" do Req.Test.stub(Client, fn conn -> - Req.Test.json(conn, %{ - "data" => %{ - "qoe_score" => 85.5, - "capacity_score" => 72.0, - "subscriber_count" => 42, - "p95_latency" => 18.3 - } - }) + assert conn.request_path == "/model/v1/access_points" + params = Plug.Conn.fetch_query_params(conn).query_params + + case params["page"] do + nil -> + Req.Test.json(conn, %{ + "data" => [%{"id" => 1, "name" => "AP1"}], + "paginator" => %{"page" => 1, "page_count" => 2, "limit" => 500, "total_count" => 2} + }) + + "2" -> + Req.Test.json(conn, %{ + "data" => [%{"id" => 2, "name" => "AP2"}], + "paginator" => %{"page" => 2, "page_count" => 2, "limit" => 500, "total_count" => 2} + }) + end end) - assert {:ok, metrics} = Client.get_access_point_metrics("valid-key", "ap-1") - assert metrics["qoe_score"] == 85.5 + assert {:ok, aps} = Client.list_access_points_basic("valid-key") + assert length(aps) == 2 + end + + test "returns error for 401" do + Req.Test.stub(Client, fn conn -> + conn |> Plug.Conn.put_status(401) |> Req.Test.json(%{"error" => "unauthorized"}) + end) + + assert {:error, :unauthorized} = Client.list_access_points_basic("bad-key") end end end diff --git a/test/towerops/preseem/device_matcher_test.exs b/test/towerops/preseem/device_matcher_test.exs index 567b49b6..515ca87c 100644 --- a/test/towerops/preseem/device_matcher_test.exs +++ b/test/towerops/preseem/device_matcher_test.exs @@ -45,7 +45,7 @@ defmodule Towerops.Preseem.DeviceMatcherTest do ap_data = Map.merge( %{ - "id" => "preseem-#{System.unique_integer([:positive])}", + "element_uuid" => "preseem-#{System.unique_integer([:positive])}", "name" => "Test AP" }, attrs @@ -61,9 +61,9 @@ defmodule Towerops.Preseem.DeviceMatcherTest do ap = create_preseem_ap(org.id, %{ - "id" => "ap-mac-match", + "element_uuid" => "ap-mac-match", "name" => "MAC Match AP", - "mac_address" => "AA:BB:CC:DD:EE:FF" + "system_mac_address" => "AA:BB:CC:DD:EE:FF" }) assert {:ok, matched_ap} = DeviceMatcher.match_access_point(ap) @@ -76,9 +76,9 @@ defmodule Towerops.Preseem.DeviceMatcherTest do ap = create_preseem_ap(org.id, %{ - "id" => "ap-mac-normalize", + "element_uuid" => "ap-mac-normalize", "name" => "MAC Normalize AP", - "mac_address" => "11-22-33-44-55-66" + "system_mac_address" => "11-22-33-44-55-66" }) assert {:ok, matched_ap} = DeviceMatcher.match_access_point(ap) @@ -90,10 +90,10 @@ defmodule Towerops.Preseem.DeviceMatcherTest do ap = create_preseem_ap(org.id, %{ - "id" => "ap-ip-match", + "element_uuid" => "ap-ip-match", "name" => "IP Match AP", - "mac_address" => "FF:FF:FF:FF:FF:FF", - "ip_address" => "10.0.0.50" + "system_mac_address" => "FF:FF:FF:FF:FF:FF", + "management_ip" => "10.0.0.50" }) assert {:ok, matched_ap} = DeviceMatcher.match_access_point(ap) @@ -111,10 +111,10 @@ defmodule Towerops.Preseem.DeviceMatcherTest do ap = create_preseem_ap(org.id, %{ - "id" => "ap-hostname-match", + "element_uuid" => "ap-hostname-match", "name" => "Tower1-Sector-A.local", - "mac_address" => "FF:FF:FF:FF:FF:FE", - "ip_address" => "172.16.0.99" + "system_mac_address" => "FF:FF:FF:FF:FF:FE", + "management_ip" => "172.16.0.99" }) assert {:ok, matched_ap} = DeviceMatcher.match_access_point(ap) @@ -129,9 +129,9 @@ defmodule Towerops.Preseem.DeviceMatcherTest do ap = create_preseem_ap(org.id, %{ - "id" => "ap-ambiguous-mac", + "element_uuid" => "ap-ambiguous-mac", "name" => "Ambiguous MAC AP", - "mac_address" => "AA:AA:AA:AA:AA:AA" + "system_mac_address" => "AA:AA:AA:AA:AA:AA" }) assert {:ok, matched_ap} = DeviceMatcher.match_access_point(ap) @@ -148,9 +148,9 @@ defmodule Towerops.Preseem.DeviceMatcherTest do ap = create_preseem_ap(org.id, %{ - "id" => "ap-ambiguous-ip", + "element_uuid" => "ap-ambiguous-ip", "name" => "Ambiguous IP AP", - "ip_address" => "10.0.2.1" + "management_ip" => "10.0.2.1" }) assert {:ok, matched_ap} = DeviceMatcher.match_access_point(ap) @@ -163,9 +163,9 @@ defmodule Towerops.Preseem.DeviceMatcherTest do ap = create_preseem_ap(org.id, %{ - "id" => "ap-manual", + "element_uuid" => "ap-manual", "name" => "Manual AP", - "ip_address" => "10.0.3.1" + "management_ip" => "10.0.3.1" }) # Manually set the match @@ -183,10 +183,10 @@ defmodule Towerops.Preseem.DeviceMatcherTest do test "returns unmatched when no devices match", %{org: org} do ap = create_preseem_ap(org.id, %{ - "id" => "ap-no-match", + "element_uuid" => "ap-no-match", "name" => "Unmatched AP", - "mac_address" => "00:00:00:00:00:01", - "ip_address" => "172.31.255.255" + "system_mac_address" => "00:00:00:00:00:01", + "management_ip" => "172.31.255.255" }) assert {:ok, unmatched_ap} = DeviceMatcher.match_access_point(ap) @@ -202,23 +202,23 @@ defmodule Towerops.Preseem.DeviceMatcherTest do _ap1 = create_preseem_ap(org.id, %{ - "id" => "ap-bulk-1", + "element_uuid" => "ap-bulk-1", "name" => "Bulk AP 1", - "ip_address" => "10.0.10.1" + "management_ip" => "10.0.10.1" }) _ap2 = create_preseem_ap(org.id, %{ - "id" => "ap-bulk-2", + "element_uuid" => "ap-bulk-2", "name" => "Bulk AP 2", - "ip_address" => "10.0.10.2" + "management_ip" => "10.0.10.2" }) _ap3 = create_preseem_ap(org.id, %{ - "id" => "ap-bulk-3", + "element_uuid" => "ap-bulk-3", "name" => "Bulk AP 3 No Match", - "ip_address" => "172.16.0.100" + "management_ip" => "172.16.0.100" }) assert {:ok, matched_count} = DeviceMatcher.match_unmatched(org.id) @@ -235,9 +235,9 @@ defmodule Towerops.Preseem.DeviceMatcherTest do ap = create_preseem_ap(org.id, %{ - "id" => "ap-already-matched", + "element_uuid" => "ap-already-matched", "name" => "Already Matched AP", - "ip_address" => "10.0.11.1" + "management_ip" => "10.0.11.1" }) # Manually match it first diff --git a/test/towerops/preseem/sync_test.exs b/test/towerops/preseem/sync_test.exs index f93c20be..0cd5674a 100644 --- a/test/towerops/preseem/sync_test.exs +++ b/test/towerops/preseem/sync_test.exs @@ -8,7 +8,6 @@ defmodule Towerops.Preseem.SyncTest do alias Towerops.Integrations.Integration alias Towerops.Preseem.AccessPoint alias Towerops.Preseem.Client - alias Towerops.Preseem.SubscriberMetric alias Towerops.Preseem.Sync alias Towerops.Preseem.SyncLog @@ -21,20 +20,19 @@ defmodule Towerops.Preseem.SyncTest do end describe "upsert_access_point/2" do - test "creates a new access point from API data", %{org: org} do + test "creates a new access point from scores API data", %{org: org} do ap_data = %{ - "id" => "preseem-ap-100", + "element_uuid" => "preseem-ap-100", "name" => "Tower1-AP1", - "mac_address" => "AA:BB:CC:DD:EE:FF", - "ip_address" => "10.0.1.1", - "model" => "AF60", + "system_mac_address" => "AA:BB:CC:DD:EE:FF", + "management_ip" => "10.0.1.1", + "model_name" => "AF60", "firmware" => "4.3.2", - "capacity_score" => 85.0, - "qoe_score" => 92.5, + "subscriber_capacity" => 85, + "ap_health_score" => 92.5, "rf_score" => 78.0, - "busy_hours" => 6, - "airtime_utilization" => 45.2, - "subscriber_count" => 30 + "airtime_remaining_tx" => 45, + "connections" => 30 } assert {:ok, ap} = Sync.upsert_access_point(org.id, ap_data) @@ -47,8 +45,8 @@ defmodule Towerops.Preseem.SyncTest do assert ap.capacity_score == 85.0 assert ap.qoe_score == 92.5 assert ap.rf_score == 78.0 - assert ap.busy_hours == 6 - assert ap.airtime_utilization == 45.2 + assert ap.busy_hours == nil + assert ap.airtime_utilization == 45.0 assert ap.subscriber_count == 30 assert ap.organization_id == org.id assert ap.match_confidence == "unmatched" @@ -57,18 +55,17 @@ defmodule Towerops.Preseem.SyncTest do test "upserts an existing access point", %{org: org} do ap_data = %{ - "id" => "preseem-ap-200", + "element_uuid" => "preseem-ap-200", "name" => "Tower1-AP2", - "mac_address" => "11:22:33:44:55:66", - "ip_address" => "10.0.1.2", - "model" => "AF60", + "system_mac_address" => "11:22:33:44:55:66", + "management_ip" => "10.0.1.2", + "model_name" => "AF60", "firmware" => "4.3.1", - "capacity_score" => 70.0, - "qoe_score" => 80.0, + "subscriber_capacity" => 70, + "ap_health_score" => 80.0, "rf_score" => 65.0, - "busy_hours" => 4, - "airtime_utilization" => 30.0, - "subscriber_count" => 20 + "airtime_remaining_tx" => 30, + "connections" => 20 } assert {:ok, original} = Sync.upsert_access_point(org.id, ap_data) @@ -77,8 +74,8 @@ defmodule Towerops.Preseem.SyncTest do Map.merge(ap_data, %{ "name" => "Tower1-AP2-Updated", "firmware" => "4.3.2", - "qoe_score" => 88.0, - "subscriber_count" => 25 + "ap_health_score" => 88.0, + "connections" => 25 }) assert {:ok, updated} = Sync.upsert_access_point(org.id, updated_data) @@ -91,10 +88,10 @@ defmodule Towerops.Preseem.SyncTest do test "stores full API response as raw_data", %{org: org} do ap_data = %{ - "id" => "preseem-ap-300", + "element_uuid" => "preseem-ap-300", "name" => "Tower1-AP3", - "extra_field" => "some_value", - "nested" => %{"key" => "value"} + "site_name" => "Main Tower", + "status" => "online" } assert {:ok, ap} = Sync.upsert_access_point(org.id, ap_data) @@ -106,7 +103,7 @@ defmodule Towerops.Preseem.SyncTest do test "creates a subscriber metric record", %{org: org} do {:ok, ap} = Sync.upsert_access_point(org.id, %{ - "id" => "preseem-ap-metrics", + "element_uuid" => "preseem-ap-metrics", "name" => "Metrics AP" }) @@ -138,38 +135,34 @@ defmodule Towerops.Preseem.SyncTest do integration: integration } do Req.Test.stub(Client, fn conn -> - Req.Test.json(conn, %{ - "data" => [ - %{ - "id" => "ap-sync-1", - "name" => "Sync AP 1", - "mac_address" => "AA:BB:CC:11:22:33", - "ip_address" => "10.0.0.1", - "model" => "AF60", - "firmware" => "4.3.2", - "capacity_score" => 85.0, - "qoe_score" => 90.0, - "rf_score" => 75.0, - "busy_hours" => 5, - "airtime_utilization" => 40.0, - "subscriber_count" => 25 - }, - %{ - "id" => "ap-sync-2", - "name" => "Sync AP 2", - "mac_address" => "AA:BB:CC:44:55:66", - "ip_address" => "10.0.0.2", - "model" => "LTU-Rocket", - "firmware" => "2.6.1", - "capacity_score" => 70.0, - "qoe_score" => 82.0, - "rf_score" => 68.0, - "busy_hours" => 3, - "airtime_utilization" => 35.0, - "subscriber_count" => 18 - } - ] - }) + Req.Test.json(conn, [ + %{ + "element_uuid" => "ap-sync-1", + "name" => "Sync AP 1", + "system_mac_address" => "AA:BB:CC:11:22:33", + "management_ip" => "10.0.0.1", + "model_name" => "AF60", + "firmware" => "4.3.2", + "subscriber_capacity" => 85, + "ap_health_score" => 90.0, + "rf_score" => 75.0, + "airtime_remaining_tx" => 40, + "connections" => 25 + }, + %{ + "element_uuid" => "ap-sync-2", + "name" => "Sync AP 2", + "system_mac_address" => "AA:BB:CC:44:55:66", + "management_ip" => "10.0.0.2", + "model_name" => "LTU-Rocket", + "firmware" => "2.6.1", + "subscriber_capacity" => 70, + "ap_health_score" => 82.0, + "rf_score" => 68.0, + "airtime_remaining_tx" => 35, + "connections" => 18 + } + ]) end) assert {:ok, result} = Sync.sync_organization(integration) @@ -212,40 +205,5 @@ defmodule Towerops.Preseem.SyncTest do updated_integration = Repo.get!(Integration, integration.id) assert updated_integration.last_sync_status == "failed" end - - test "syncs access points with metrics when present", %{ - org: org, - integration: integration - } do - Req.Test.stub(Client, fn conn -> - Req.Test.json(conn, %{ - "data" => [ - %{ - "id" => "ap-with-metrics", - "name" => "Metrics AP", - "ip_address" => "10.0.0.5", - "metrics" => %{ - "avg_latency" => 15.0, - "avg_jitter" => 2.5, - "avg_loss" => 0.3, - "avg_throughput" => 200.0, - "p95_latency" => 30.0, - "subscriber_count" => 50, - "recorded_at" => "2026-02-12T12:00:00Z" - } - } - ] - }) - end) - - assert {:ok, result} = Sync.sync_organization(integration) - assert result.synced == 1 - - # Verify metrics were inserted - [ap] = Repo.all(from ap in AccessPoint, where: ap.organization_id == ^org.id) - metrics = Repo.all(from m in SubscriberMetric, where: m.preseem_access_point_id == ^ap.id) - assert length(metrics) == 1 - assert hd(metrics).avg_latency == 15.0 - end end end diff --git a/test/towerops/workers/preseem_sync_worker_test.exs b/test/towerops/workers/preseem_sync_worker_test.exs index 3f14c8f7..645abf04 100644 --- a/test/towerops/workers/preseem_sync_worker_test.exs +++ b/test/towerops/workers/preseem_sync_worker_test.exs @@ -15,28 +15,38 @@ defmodule Towerops.Workers.PreseemSyncWorkerTest do %{org: org} end - describe "perform/1" do - test "syncs enabled preseem integrations", %{org: org} do + describe "perform/1 dispatcher (cron)" do + test "enqueues individual sync jobs for enabled integrations", %{org: org} do integration = integration_fixture(org.id) - Req.Test.stub(Client, fn conn -> - Req.Test.json(conn, %{ - "data" => [ - %{ - "id" => "ap-1", - "name" => "Test AP", - "mac_address" => "AA:BB:CC:DD:EE:FF", - "ip_address" => "10.0.0.1" - } - ] - }) - end) - assert :ok = perform_job(PreseemSyncWorker, %{}) - updated = Repo.reload!(integration) - assert updated.last_synced_at - assert updated.last_sync_status in ["success", "partial"] + # Should have enqueued a job with this integration's ID + assert [job] = all_enqueued(worker: PreseemSyncWorker) + assert job.args == %{"integration_id" => integration.id} + assert job.scheduled_at + end + + test "staggers jobs across the 600-second window", %{org: org} do + # Create a second org with its own integration + user2 = user_fixture() + org2 = organization_fixture(user2.id) + + _int1 = integration_fixture(org.id) + _int2 = integration_fixture(org2.id) + + assert :ok = perform_job(PreseemSyncWorker, %{}) + + jobs = all_enqueued(worker: PreseemSyncWorker) + assert length(jobs) == 2 + + # Jobs should have scheduled_at times within 0-600 seconds from now + now = DateTime.utc_now() + + for job <- jobs do + diff = DateTime.diff(job.scheduled_at, now, :second) + assert diff >= 0 and diff < 600, "Expected offset in [0, 600), got #{diff}" + end end test "skips integrations synced recently", %{org: org} do @@ -46,24 +56,15 @@ defmodule Towerops.Workers.PreseemSyncWorkerTest do last_sync_status: "success" }) - # No Client stub needed - should skip without calling API assert :ok = perform_job(PreseemSyncWorker, %{}) + + # No jobs should be enqueued for recently-synced integrations + assert [] = all_enqueued(worker: PreseemSyncWorker) end test "handles no enabled integrations" do assert :ok = perform_job(PreseemSyncWorker, %{}) - end - - test "handles API errors without crashing", %{org: org} do - _integration = integration_fixture(org.id) - - Req.Test.stub(Client, fn conn -> - conn - |> Plug.Conn.put_status(401) - |> Req.Test.json(%{"error" => "unauthorized"}) - end) - - assert :ok = perform_job(PreseemSyncWorker, %{}) + assert [] = all_enqueued(worker: PreseemSyncWorker) end test "respects sync_interval_minutes", %{org: org} do @@ -80,11 +81,11 @@ defmodule Towerops.Workers.PreseemSyncWorkerTest do last_sync_status: "success" }) - # No Client stub needed - should skip without calling API assert :ok = perform_job(PreseemSyncWorker, %{}) + assert [] = all_enqueued(worker: PreseemSyncWorker) end - test "syncs when sync interval has elapsed", %{org: org} do + test "enqueues job when sync interval has elapsed", %{org: org} do # Last synced 15 minutes ago with a 10-minute interval -> should sync fifteen_minutes_ago = DateTime.utc_now() @@ -98,28 +99,60 @@ defmodule Towerops.Workers.PreseemSyncWorkerTest do last_sync_status: "success" }) - Req.Test.stub(Client, fn conn -> - Req.Test.json(conn, %{"data" => []}) - end) - assert :ok = perform_job(PreseemSyncWorker, %{}) - updated = Repo.reload!(integration) - assert DateTime.after?(updated.last_synced_at, fifteen_minutes_ago) + assert [job] = all_enqueued(worker: PreseemSyncWorker) + assert job.args == %{"integration_id" => integration.id} end - test "syncs integrations that have never been synced", %{org: org} do + test "enqueues job for integrations that have never been synced", %{org: org} do integration = integration_fixture(org.id) assert integration.last_synced_at == nil + assert :ok = perform_job(PreseemSyncWorker, %{}) + + assert [job] = all_enqueued(worker: PreseemSyncWorker) + assert job.args == %{"integration_id" => integration.id} + end + end + + describe "perform/1 individual sync" do + test "syncs the given integration", %{org: org} do + integration = integration_fixture(org.id) + Req.Test.stub(Client, fn conn -> - Req.Test.json(conn, %{"data" => []}) + Req.Test.json(conn, [ + %{ + "element_uuid" => "ap-1", + "name" => "Test AP", + "system_mac_address" => "AA:BB:CC:DD:EE:FF", + "management_ip" => "10.0.0.1" + } + ]) end) - assert :ok = perform_job(PreseemSyncWorker, %{}) + assert :ok = perform_job(PreseemSyncWorker, %{"integration_id" => integration.id}) updated = Repo.reload!(integration) assert updated.last_synced_at + assert updated.last_sync_status in ["success", "partial"] + end + + test "handles API errors without crashing", %{org: org} do + integration = integration_fixture(org.id) + + Req.Test.stub(Client, fn conn -> + conn + |> Plug.Conn.put_status(401) + |> Req.Test.json(%{"error" => "unauthorized"}) + end) + + assert :ok = perform_job(PreseemSyncWorker, %{"integration_id" => integration.id}) + end + + test "returns error for nonexistent integration" do + assert {:error, :not_found} = + perform_job(PreseemSyncWorker, %{"integration_id" => Ecto.UUID.generate()}) end end end diff --git a/test/towerops_web/live/org/integrations_live_test.exs b/test/towerops_web/live/org/integrations_live_test.exs index 0e94d330..269f0f8f 100644 --- a/test/towerops_web/live/org/integrations_live_test.exs +++ b/test/towerops_web/live/org/integrations_live_test.exs @@ -61,7 +61,10 @@ defmodule ToweropsWeb.Org.IntegrationsLiveTest do test "can test preseem connection successfully", %{conn: conn, user: user, organization: org} do Req.Test.stub(PreseemClient, fn conn -> - Req.Test.json(conn, %{"status" => "ok"}) + Req.Test.json(conn, %{ + "data" => [%{"id" => 1, "name" => "Test AP"}], + "paginator" => %{"page" => 1, "page_count" => 1, "limit" => 100, "total_count" => 1} + }) end) {:ok, view, _html} =