diff --git a/lib/towerops/preseem/sync.ex b/lib/towerops/preseem/sync.ex new file mode 100644 index 00000000..ffcdb138 --- /dev/null +++ b/lib/towerops/preseem/sync.ex @@ -0,0 +1,179 @@ +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 + into `preseem_access_points`, and logs each sync operation. + """ + + alias Towerops.Integrations + alias Towerops.Preseem.AccessPoint + alias Towerops.Preseem.Client + alias Towerops.Preseem.DeviceMatcher + alias Towerops.Preseem.SubscriberMetric + alias Towerops.Preseem.SyncLog + alias Towerops.Repo + + require Logger + + @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. + + Returns `{:ok, %{synced: count, matched: count}}` or `{:error, reason}`. + """ + def sync_organization(%Integrations.Integration{} = integration) do + start_time = System.monotonic_time(:millisecond) + api_key = integration.credentials["api_key"] + org_id = integration.organization_id + + opts = + case integration.credentials["base_url"] do + nil -> [] + base_url -> [base_url: base_url] + end + + case Client.list_access_points(api_key, opts) do + {:ok, ap_list} -> + handle_successful_sync(integration, org_id, ap_list, start_time) + + {:error, reason} -> + handle_failed_sync(integration, reason, start_time) + end + end + + @doc """ + Upserts a single access point from raw API response data. + + Maps 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"], + name: ap_data["name"], + mac_address: ap_data["mac_address"], + ip_address: ap_data["ip_address"], + model: ap_data["model"], + 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"], + raw_data: ap_data + } + + %AccessPoint{} + |> AccessPoint.changeset(attrs) + |> Repo.insert( + on_conflict: + {:replace_all_except, [:id, :organization_id, :preseem_id, :device_id, :match_confidence, :inserted_at]}, + conflict_target: [:organization_id, :preseem_id], + returning: true + ) + end + + @doc """ + Inserts a subscriber metric record for the given access point. + """ + def insert_metrics(access_point_id, metrics_data) do + recorded_at = + case metrics_data["recorded_at"] do + nil -> DateTime.truncate(DateTime.utc_now(), :second) + dt_string -> parse_datetime(dt_string) + end + + attrs = %{ + preseem_access_point_id: access_point_id, + avg_latency: metrics_data["avg_latency"], + avg_jitter: metrics_data["avg_jitter"], + avg_loss: metrics_data["avg_loss"], + avg_throughput: metrics_data["avg_throughput"], + p95_latency: metrics_data["p95_latency"], + subscriber_count: metrics_data["subscriber_count"], + recorded_at: recorded_at + } + + %SubscriberMetric{} + |> SubscriberMetric.changeset(attrs) + |> Repo.insert() + end + + defp handle_successful_sync(integration, org_id, ap_list, start_time) 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) + count + 1 + + {:error, changeset} -> + Logger.warning("Failed to upsert Preseem AP: #{inspect(changeset.errors)}") + count + end + end) + + matched_count = run_device_matching(org_id) + duration_ms = System.monotonic_time(:millisecond) - start_time + + status = if synced_count == length(ap_list), do: "success", else: "partial" + + log_sync(integration, status, synced_count, %{}, duration_ms) + Integrations.update_sync_status(integration, status) + + {:ok, %{synced: synced_count, matched: matched_count}} + end + + defp handle_failed_sync(integration, reason, start_time) do + duration_ms = System.monotonic_time(:millisecond) - start_time + + log_sync(integration, "failed", 0, %{error: inspect(reason)}, duration_ms) + Integrations.update_sync_status(integration, "failed") + + {: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 + {:ok, count} -> count + _ -> 0 + end + else + 0 + end + end + + defp log_sync(integration, status, records_synced, errors, duration_ms) do + %SyncLog{} + |> SyncLog.changeset(%{ + organization_id: integration.organization_id, + integration_id: integration.id, + status: status, + records_synced: records_synced, + errors: errors, + duration_ms: duration_ms, + inserted_at: DateTime.truncate(DateTime.utc_now(), :second) + }) + |> Repo.insert() + end + + defp parse_datetime(dt_string) when is_binary(dt_string) do + case DateTime.from_iso8601(dt_string) do + {:ok, dt, _offset} -> DateTime.truncate(dt, :second) + _ -> DateTime.truncate(DateTime.utc_now(), :second) + end + end +end diff --git a/test/towerops/preseem/sync_test.exs b/test/towerops/preseem/sync_test.exs new file mode 100644 index 00000000..cb6cde36 --- /dev/null +++ b/test/towerops/preseem/sync_test.exs @@ -0,0 +1,267 @@ +defmodule Towerops.Preseem.SyncTest do + use Towerops.DataCase, async: true + + import Towerops.AccountsFixtures + import Towerops.IntegrationsFixtures + import Towerops.OrganizationsFixtures + + alias Towerops.Integrations.Integration + alias Towerops.Preseem.AccessPoint + alias Towerops.Preseem.Client + alias Towerops.Preseem.SubscriberMetric + alias Towerops.Preseem.Sync + alias Towerops.Preseem.SyncLog + + setup do + user = user_fixture() + org = organization_fixture(user.id) + integration = integration_fixture(org.id) + + %{org: org, integration: integration} + end + + describe "upsert_access_point/2" do + test "creates a new access point from API data", %{org: org} do + ap_data = %{ + "id" => "preseem-ap-100", + "name" => "Tower1-AP1", + "mac_address" => "AA:BB:CC:DD:EE:FF", + "ip_address" => "10.0.1.1", + "model" => "AF60", + "firmware" => "4.3.2", + "capacity_score" => 85.0, + "qoe_score" => 92.5, + "rf_score" => 78.0, + "busy_hours" => 6, + "airtime_utilization" => 45.2, + "subscriber_count" => 30 + } + + assert {:ok, ap} = Sync.upsert_access_point(org.id, ap_data) + assert ap.preseem_id == "preseem-ap-100" + assert ap.name == "Tower1-AP1" + assert ap.mac_address == "AA:BB:CC:DD:EE:FF" + assert ap.ip_address == "10.0.1.1" + assert ap.model == "AF60" + assert ap.firmware == "4.3.2" + 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.subscriber_count == 30 + assert ap.organization_id == org.id + assert ap.match_confidence == "unmatched" + assert ap.raw_data == ap_data + end + + test "upserts an existing access point", %{org: org} do + ap_data = %{ + "id" => "preseem-ap-200", + "name" => "Tower1-AP2", + "mac_address" => "11:22:33:44:55:66", + "ip_address" => "10.0.1.2", + "model" => "AF60", + "firmware" => "4.3.1", + "capacity_score" => 70.0, + "qoe_score" => 80.0, + "rf_score" => 65.0, + "busy_hours" => 4, + "airtime_utilization" => 30.0, + "subscriber_count" => 20 + } + + assert {:ok, original} = Sync.upsert_access_point(org.id, ap_data) + + updated_data = + Map.merge(ap_data, %{ + "name" => "Tower1-AP2-Updated", + "firmware" => "4.3.2", + "qoe_score" => 88.0, + "subscriber_count" => 25 + }) + + assert {:ok, updated} = Sync.upsert_access_point(org.id, updated_data) + assert updated.id == original.id + assert updated.name == "Tower1-AP2-Updated" + assert updated.firmware == "4.3.2" + assert updated.qoe_score == 88.0 + assert updated.subscriber_count == 25 + end + + test "stores full API response as raw_data", %{org: org} do + ap_data = %{ + "id" => "preseem-ap-300", + "name" => "Tower1-AP3", + "extra_field" => "some_value", + "nested" => %{"key" => "value"} + } + + assert {:ok, ap} = Sync.upsert_access_point(org.id, ap_data) + assert ap.raw_data == ap_data + end + end + + describe "insert_metrics/2" do + test "creates a subscriber metric record", %{org: org} do + {:ok, ap} = + Sync.upsert_access_point(org.id, %{ + "id" => "preseem-ap-metrics", + "name" => "Metrics AP" + }) + + metrics_data = %{ + "avg_latency" => 12.5, + "avg_jitter" => 3.2, + "avg_loss" => 0.5, + "avg_throughput" => 150.0, + "p95_latency" => 25.0, + "subscriber_count" => 42, + "recorded_at" => "2026-02-12T10:00:00Z" + } + + assert {:ok, metric} = Sync.insert_metrics(ap.id, metrics_data) + assert metric.preseem_access_point_id == ap.id + assert metric.avg_latency == 12.5 + assert metric.avg_jitter == 3.2 + assert metric.avg_loss == 0.5 + assert metric.avg_throughput == 150.0 + assert metric.p95_latency == 25.0 + assert metric.subscriber_count == 42 + assert metric.recorded_at == ~U[2026-02-12 10:00:00Z] + end + end + + describe "sync_organization/1" do + test "syncs access points from API and creates sync log", %{ + org: org, + 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 + } + ] + }) + end) + + assert {:ok, result} = Sync.sync_organization(integration) + assert result.synced == 2 + + # Verify APs were created + aps = Repo.all(from ap in AccessPoint, where: ap.organization_id == ^org.id) + assert length(aps) == 2 + assert Enum.any?(aps, &(&1.preseem_id == "ap-sync-1")) + assert Enum.any?(aps, &(&1.preseem_id == "ap-sync-2")) + + # Verify sync log was written + [log] = Repo.all(from sl in SyncLog, where: sl.integration_id == ^integration.id) + assert log.status == "success" + assert log.records_synced == 2 + assert log.organization_id == org.id + assert log.duration_ms >= 0 + + # Verify integration sync status was updated + updated_integration = Repo.get!(Integration, integration.id) + assert updated_integration.last_sync_status == "success" + assert updated_integration.last_synced_at + end + + test "handles API errors gracefully", %{integration: integration} do + Req.Test.stub(Client, fn conn -> + conn + |> Plug.Conn.put_status(401) + |> Req.Test.json(%{"error" => "unauthorized"}) + end) + + assert {:error, :unauthorized} = Sync.sync_organization(integration) + + # Verify sync log records the failure + [log] = Repo.all(from sl in SyncLog, where: sl.integration_id == ^integration.id) + assert log.status == "failed" + assert log.records_synced == 0 + + # Verify integration sync status was updated to failed + 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 + + test "uses base_url from integration credentials", %{integration: integration} do + # Update integration with a custom base_url + integration + |> Ecto.Changeset.change(%{ + credentials: %{"api_key" => "test-key", "base_url" => "https://custom.preseem.com/v1"} + }) + |> Repo.update!() + + Req.Test.stub(Client, fn conn -> + Req.Test.json(conn, %{"data" => []}) + end) + + assert {:ok, result} = Sync.sync_organization(Repo.reload!(integration)) + assert result.synced == 0 + end + end +end