Graham McIntire
4ba7269e57
Queued contacts just need reconciliation (check if data exists), while pending contacts need actual HRRR fetches. Process pending first so real work happens before reconciliation.
67 lines
1.9 KiB
Elixir
67 lines
1.9 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]
|
|
@valid_types ~w(hrrr weather terrain iemre)
|
|
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{args: %{"limit" => limit} = args}) do
|
|
types = parse_types(args)
|
|
|
|
contacts =
|
|
Contact
|
|
|> where([c], not is_nil(c.pos1) or not is_nil(c.grid1))
|
|
|> where(^type_filter(types))
|
|
|> order_by(^status_priority_order(types))
|
|
|> limit(^limit)
|
|
|> Repo.all()
|
|
|
|
count = length(contacts)
|
|
Logger.info("BackfillEnqueue: processing #{count} contacts (types: #{inspect(types)})")
|
|
|
|
Enum.each(contacts, &ContactWeatherEnqueueWorker.enqueue_for_contact(&1, types))
|
|
|
|
Phoenix.PubSub.broadcast(
|
|
Microwaveprop.PubSub,
|
|
"backfill:enqueue_complete",
|
|
{:enqueue_complete, count}
|
|
)
|
|
|
|
:ok
|
|
end
|
|
|
|
defp parse_types(%{"types" => types}) when is_list(types) do
|
|
types |> Enum.filter(&(&1 in @valid_types)) |> Enum.map(&String.to_existing_atom/1)
|
|
end
|
|
|
|
defp parse_types(_), do: [:hrrr, :weather, :terrain, :iemre]
|
|
|
|
defp status_priority_order(types) do
|
|
# Prioritize pending/failed over queued so real work happens first
|
|
field = :"#{hd(types)}_status"
|
|
|
|
[
|
|
asc: dynamic([c], fragment("CASE ? WHEN 'pending' THEN 0 WHEN 'failed' THEN 1 ELSE 2 END", field(c, ^field))),
|
|
desc: dynamic([c], c.qso_timestamp)
|
|
]
|
|
end
|
|
|
|
defp type_filter(types) do
|
|
Enum.reduce(types, dynamic(false), fn type, acc ->
|
|
field = :"#{type}_status"
|
|
dynamic([c], ^acc or field(c, ^field) in ^@enrichable)
|
|
end)
|
|
end
|
|
end
|