prop/lib/microwaveprop/workers/radar_frame_worker.ex

defmodule Microwaveprop.Workers.RadarFrameWorker do
  @moduledoc """
  Batched NEXRAD common-volume radar enrichment.

  Each job processes every contact whose QSO timestamp rounds to the
  same 5-minute NEXRAD frame — fetch + decode happens *once* per job
  and the decoded pixel buffer is walked per contact. This collapses
  the 1-job-per-contact backfill pattern (`CommonVolumeRadarWorker`)
  down by roughly 10x at current backlog depth (17k contacts / 1.7k
  distinct frames).

  `CommonVolumeRadarWorker` stays around for the submit-time single-
  contact path where batching is pointless. Both workers end in the
  same state transitions so they can coexist safely — a contact
  marked `:complete` by one is a no-op for the other.
  """

  # unique on frame_ts: backfill sweeps that re-select the same still-:queued
  # contacts each cycle would otherwise enqueue a fresh 1-job-per-contact
  # frame fetch. 9k+ wasteful duplicates observed in prod before this guard.
  use Oban.Pro.Worker,
    queue: :radar,
    max_attempts: 3,
    unique: [fields: [:args], keys: [:frame_ts], states: :incomplete, period: :infinity]

  import Ecto.Query

  alias Microwaveprop.Radio.Contact
  alias Microwaveprop.Radio.ContactCommonVolumeRadar
  alias Microwaveprop.Repo
  alias Microwaveprop.Weather.NexradClient
  alias Microwaveprop.Workers.CommonVolumeRadarWorker

  require Logger

  @terminal_statuses [:complete, :unavailable]
  @frame_window_seconds 5 * 60

  @impl Oban.Pro.Worker
  def process(%Oban.Job{args: %{"frame_ts" => frame_iso}}) do
    {:ok, frame_ts, _} = DateTime.from_iso8601(frame_iso)
    rounded = NexradClient.round_to_5min(frame_ts)
    bucket_end = DateTime.add(rounded, @frame_window_seconds, :second)

    # Query by the frame's 5-min bucket rather than a passed contact_ids
    # list. The unique-on-frame_ts guard prevents new sweeps from enqueueing
    # duplicate jobs, which means a contact that flips to :queued after the
    # job is enqueued won't land in a separate job's contact_ids — it has to
    # be picked up by the already-queued frame job instead.
    contacts =
      Contact
      |> where([c], c.qso_timestamp >= ^rounded and c.qso_timestamp < ^bucket_end)
      |> where([c], c.radar_status not in ^@terminal_statuses)
      |> Repo.all()

    case contacts do
      [] ->
        :ok

      contacts ->
        run_batch(contacts, rounded)
    end
  end

  defp run_batch(contacts, rounded) do
    case NexradClient.fetch_decoded_frame(rounded) do
      {:ok, pixels, width} ->
        process_frame(contacts, rounded, pixels, width)
        :ok

      {:error, reason} ->
        if permanent_error?(reason) do
          Enum.each(contacts, &mark_status(&1, :unavailable))

          Logger.info(
            "RadarFrameWorker: no frame for #{DateTime.to_iso8601(rounded)} (#{length(contacts)} contacts): #{inspect(reason)}"
          )

          :ok
        else
          # Transient (5xx, timeout, connrefused) — let Oban retry so the
          # contacts stay :queued and land on the next attempt rather than
          # getting pinned :unavailable after a temporary NEXRAD outage.
          Logger.warning(
            "RadarFrameWorker: transient error for #{DateTime.to_iso8601(rounded)} (#{length(contacts)} contacts): #{inspect(reason)}"
          )

          {:error, reason}
        end
    end
  end

  # Permanent: 4xx means the IEM archive does not have this frame (the
  # archive genuinely has gaps), so the frame will never exist and
  # retrying is pointless. Any other shape is treated as transient.
  defp permanent_error?("NEXRAD n0q HTTP " <> code) do
    case Integer.parse(code) do
      {status, _} when status in 400..499 -> true
      _ -> false
    end
  end

  defp permanent_error?(_), do: false

  @doc """
  Process a batch of contacts against an already-decoded frame. Exposed
  for tests and for any caller that has the pixel buffer in hand.
  """
  @spec process_frame([Contact.t()], DateTime.t(), binary(), pos_integer()) :: :ok
  def process_frame(contacts, rounded, pixels, width) do
    Enum.each(contacts, fn contact ->
      process_contact(contact, pixels, width, rounded)
    end)

    Logger.info("RadarFrameWorker: #{length(contacts)} contacts ingested for frame #{DateTime.to_iso8601(rounded)}")

    :ok
  end

  defp process_contact(%Contact{pos1: p1, pos2: p2} = contact, pixels, width, rounded) when is_map(p1) and is_map(p2) do
    pos1 = to_latlon(p1)
    pos2 = to_latlon(p2)

    stats = CommonVolumeRadarWorker.aggregate_stats(pixels, width, pos1, pos2)
    upsert_row!(contact, rounded, stats)
    mark_status(contact, :complete)
  end

  defp process_contact(contact, _pixels, _width, _rounded) do
    # Positions missing — can't compute a common volume, but we got
    # this far because the enqueue filter thought it was enrichable.
    # Mark unavailable so the backfill cron stops re-queueing it.
    mark_status(contact, :unavailable)
  end

  defp upsert_row!(%Contact{id: contact_id}, observed_at, stats) do
    %ContactCommonVolumeRadar{}
    |> ContactCommonVolumeRadar.changeset(Map.merge(stats, %{contact_id: contact_id, observed_at: observed_at}))
    |> Repo.insert(
      on_conflict: {:replace, ~w(observed_at common_volume_km2 pixel_count rain_pixel_count heavy_rain_pixel_count
            core_pixel_count max_dbz mean_dbz coverage_pct updated_at)a},
      conflict_target: :contact_id
    )
  end

  defp mark_status(%Contact{} = contact, status) do
    contact
    |> Ecto.Changeset.change(%{radar_status: status})
    |> Repo.update!()
  end

  defp to_latlon(%{"lat" => lat, "lon" => lon}) when is_number(lat) and is_number(lon) do
    {lat / 1.0, lon / 1.0}
  end
end