Add periodic enrichment backfill cron and terrain reconciliation

Runs BackfillEnqueueWorker every 30 minutes to pick up contacts with
pending/queued/failed enrichment status and enqueue missing jobs.
Before enqueueing, reconciles terrain contacts stuck in "queued" by
checking the terrain_profiles FK directly.
This commit is contained in:
Graham McIntire 2026-04-07 07:36:45 -05:00
parent 8e39cfcc4c
commit e03b9fbc0e
2 changed files with 32 additions and 2 deletions

View file

@ -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"]}}
]}
]

View file

@ -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"