Graham McIntire
6a9b471d1f
- H23: introduce Towerops.Workers.SyncErrors.transient?/1 classifier; uisp/preseem/cn_maestro sync workers now retry transient failures (5xx, timeouts, rate limits) via Oban while still discarding permanent errors (401/403/404) as :ok. Stale monitoring data is the bigger risk than an extra retry on a known-bad credential. - H26: add Monitoring.get_latency_data_for_devices/2 batch counterpart (currently a stub map but called from SiteLive.Show so any future ping implementation lands in a single query, not one-per-device). - H28: Nominatim search popups in the coverage map now bind via a text node instead of an HTML string, so a compromised/poisoned response can't execute as JS in the popup. - H29: yaml_profiles dot-boundary OID prefix match (handles optional trailing dot from YAML profile keys). Stops 1.3.6.1.4.1.9 from incorrectly matching 1.3.6.1.4.1.99.
92 lines
2.9 KiB
Elixir
92 lines
2.9 KiB
Elixir
defmodule Towerops.Workers.UispSyncWorker do
|
|
@moduledoc """
|
|
Oban worker that syncs UISP data for enabled integrations.
|
|
|
|
The cron job (no args) runs every 5 minutes and dispatches individual
|
|
per-integration sync jobs staggered across the 300-second window using
|
|
`PollingOffset.calculate_offset/2`. This prevents a thundering herd
|
|
against UISP APIs.
|
|
|
|
Individual jobs receive `%{"integration_id" => id}` and perform the
|
|
actual sync via `Uisp.Sync.sync_organization/1`.
|
|
"""
|
|
use Oban.Worker, queue: :maintenance
|
|
|
|
alias Towerops.Integrations
|
|
alias Towerops.Uisp.Sync
|
|
alias Towerops.Workers.PollingOffset
|
|
alias Towerops.Workers.SyncErrors
|
|
|
|
require Logger
|
|
|
|
@window_seconds 300
|
|
|
|
# Cron dispatcher: enqueues staggered individual sync jobs
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{args: args}) when args == %{} do
|
|
integrations = Integrations.list_enabled_integrations("uisp")
|
|
|
|
eligible = Enum.filter(integrations, &should_sync?/1)
|
|
|
|
now = DateTime.utc_now()
|
|
|
|
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("UISP sync: dispatched #{enqueued} jobs across #{@window_seconds}s window")
|
|
end
|
|
|
|
:ok
|
|
end
|
|
|
|
# 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, :not_found} ->
|
|
Logger.warning("UISP 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("UISP sync completed for org #{integration.organization_id}: #{inspect(result)}")
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
Logger.error("UISP sync failed for org #{integration.organization_id}: #{inspect(reason)}")
|
|
# Distinguish permanent failures (auth, bad config) from transient
|
|
# ones (network, rate limits, 5xx) so Oban retries the latter and
|
|
# data doesn't stay stale until the next cron tick.
|
|
if SyncErrors.transient?(reason), do: {:error, reason}, else: :ok
|
|
end
|
|
end
|
|
|
|
@doc false
|
|
def should_sync?(integration) do
|
|
case integration.last_synced_at do
|
|
nil ->
|
|
true
|
|
|
|
last_synced_at ->
|
|
interval_seconds = (integration.sync_interval_minutes || 10) * 60
|
|
elapsed = DateTime.diff(DateTime.utc_now(), last_synced_at, :second)
|
|
elapsed >= interval_seconds
|
|
end
|
|
end
|
|
end
|