diff --git a/config/runtime.exs b/config/runtime.exs index 384d6ff8..828dbc48 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -129,7 +129,7 @@ if config_env() == :prod do ], secret_key_base: secret_key_base - # Production Oban: live scoring, polling, and on-demand QSO enrichment (no cron backfill) + # Production Oban: live scoring, polling, and on-demand QSO enrichment config :microwaveprop, Oban, # Per-pod concurrency (×3 pods = effective cluster total) queues: [propagation: 1, commercial: 1, solar: 1, weather: 3, hrrr: 5, terrain: 3, iemre: 3, backfill_enqueue: 1], @@ -140,7 +140,9 @@ if config_env() == :prod do crontab: [ {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker}, {"*/5 * * * *", Microwaveprop.Commercial.PollWorker}, - {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker} + {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, + {"*/30 * * * *", Microwaveprop.Workers.BackfillEnqueueWorker, + args: %{"limit" => 500, "types" => ["hrrr", "weather", "terrain", "iemre"]}} ]} ] diff --git a/lib/microwaveprop/workers/backfill_enqueue_worker.ex b/lib/microwaveprop/workers/backfill_enqueue_worker.ex index eeb58110..7dfdaba5 100644 --- a/lib/microwaveprop/workers/backfill_enqueue_worker.ex +++ b/lib/microwaveprop/workers/backfill_enqueue_worker.ex @@ -20,6 +20,12 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do def perform(%Oban.Job{args: %{"limit" => limit} = args}) do types = parse_types(args) + reconciled = reconcile_stale_queued(types) + + if reconciled > 0 do + Logger.info("BackfillEnqueue: reconciled #{reconciled} stale queued contacts") + end + contacts = Contact |> where([c], not is_nil(c.pos1) or not is_nil(c.grid1)) @@ -58,6 +64,28 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do ] end + # Fast SQL reconciliation for contacts stuck in "queued" where data already exists. + # Terrain profiles have a direct contact_id FK so we can reconcile in bulk. + defp reconcile_stale_queued(types) do + terrain_count = + if :terrain in types do + {count, _} = + Repo.update_all( + from(c in Contact, + where: c.terrain_status == :queued, + where: c.id in subquery(from(tp in "terrain_profiles", select: tp.contact_id)) + ), + set: [terrain_status: :complete] + ) + + count + else + 0 + end + + terrain_count + end + defp type_filter(types) do Enum.reduce(types, dynamic(false), fn type, acc -> field = :"#{type}_status"