Graham McIntire
eeb92bcc19
enqueue_for_contact unconditionally marked contacts as :queued even when no HRRR jobs were needed (profiles already exist). Workers never updated contact status to :complete after storing profiles. Contacts got stuck and were never re-processed. - Mark :complete when no jobs generated (data already exists) - Include :queued in backfill enrichable states for reconciliation
45 lines
1.2 KiB
Elixir
45 lines
1.2 KiB
Elixir
defmodule Microwaveprop.Workers.BackfillEnqueueWorker do
|
|
@moduledoc """
|
|
Runs backfill enrichment enqueue as an Oban job so the backfill dashboard
|
|
returns immediately instead of blocking on the enqueue loop.
|
|
"""
|
|
use Oban.Worker, queue: :backfill_enqueue, max_attempts: 1
|
|
|
|
import Ecto.Query
|
|
|
|
alias Microwaveprop.Radio.Contact
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Workers.ContactWeatherEnqueueWorker
|
|
|
|
require Logger
|
|
|
|
@enrichable [:pending, :queued, :failed]
|
|
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{args: %{"limit" => limit}}) do
|
|
contacts =
|
|
Repo.all(
|
|
from(c in Contact,
|
|
where:
|
|
(not is_nil(c.pos1) or not is_nil(c.grid1)) and
|
|
(c.hrrr_status in ^@enrichable or c.weather_status in ^@enrichable or
|
|
c.terrain_status in ^@enrichable or c.iemre_status in ^@enrichable),
|
|
order_by: [desc: c.qso_timestamp],
|
|
limit: ^limit
|
|
)
|
|
)
|
|
|
|
count = length(contacts)
|
|
Logger.info("BackfillEnqueue: processing #{count} contacts")
|
|
|
|
Enum.each(contacts, &ContactWeatherEnqueueWorker.enqueue_for_contact/1)
|
|
|
|
Phoenix.PubSub.broadcast(
|
|
Microwaveprop.PubSub,
|
|
"backfill:enqueue_complete",
|
|
{:enqueue_complete, count}
|
|
)
|
|
|
|
:ok
|
|
end
|
|
end
|