prop/lib/microwaveprop/workers/backfill_enqueue_worker.ex

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 era5)

  # A :queued contact whose updated_at is older than this is one the cron has
  # already tried ~144 times over 3 days without the underlying worker landing
  # data. At that point the data is effectively unreachable (dead ASOS station,
  # pre-2014 HRRR, out-of-grid IEMRE) and the cron should stop re-enqueueing.
  # set_enrichment_status!/3 is a no-op when the status is unchanged, so a
  # stuck :queued row retains a stale updated_at across cron cycles — which is
  # exactly what we need as the "given up" signal.
  @stale_queued_cutoff_seconds 3 * 86_400

  @impl Oban.Worker
  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

    stale_unavail = reconcile_stale_queued_to_unavailable(types)

    if stale_unavail > 0 do
      Logger.info("BackfillEnqueue: marked #{stale_unavail} stuck-queued contacts as unavailable")
    end

    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, :era5]

  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

  # Flip contacts that have been stuck in :queued for longer than
  # @stale_queued_cutoff_seconds to :unavailable. One Repo.update_all per type,
  # scoped to only the types passed in args so a partial cron (e.g. hrrr-only)
  # doesn't touch other statuses. Fast: hits the (<field>, updated_at) combo
  # with a simple range scan.
  defp reconcile_stale_queued_to_unavailable(types) do
    cutoff =
      DateTime.utc_now()
      |> DateTime.add(-@stale_queued_cutoff_seconds, :second)
      |> DateTime.truncate(:second)

    Enum.reduce(types, 0, fn type, acc ->
      field = :"#{type}_status"

      {n, _} =
        Contact
        |> where([c], field(c, ^field) == :queued and c.updated_at < ^cutoff)
        |> Repo.update_all(set: [{field, :unavailable}])

      acc + n
    end)
  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
      :era5, acc ->
        # ERA5 targets contacts where HRRR is unavailable (pre-2014, missing from archive)
        dynamic([c], ^acc or c.hrrr_status == :unavailable)

      type, acc ->
        field = :"#{type}_status"
        dynamic([c], ^acc or field(c, ^field) in ^@enrichable)
    end)
  end
end