add preseem sync logic for AP upsert and metrics

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

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