feat(radar): batch per NEXRAD frame instead of per contact

CommonVolumeRadarWorker was enqueuing one job per contact — every
job fetched + decoded the ~5 MB n0q PNG for its own 5-min frame.
At current backlog depth that's 17,588 contacts across just 1,747
distinct 5-min frames, i.e. ~10 contacts per frame paying for the
same PNG decode over and over.

New RadarFrameWorker takes a batch of contact_ids sharing one frame,
fetches + decodes the frame ONCE, then walks the in-memory pixel
buffer per contact. build_radar_jobs/1 in ContactWeatherEnqueueWorker
now groups the input contacts by their rounded 5-min timestamp and
emits one RadarFrameWorker job per frame instead of one
CommonVolumeRadarWorker per contact.

process_frame/4 is public so the same code path is used both by
perform/1 (production fetch) and tests (pre-decoded pixel buffer);
keeps the happy-path test hermetic without hand-crafting a PNG.

CommonVolumeRadarWorker stays in the tree — still used from the
single-contact submit path where batching is pointless and the
aggregate_stats/5 pure helper is a dependency of the new batched
worker.

Expected impact on the backfill: ~10x fewer fetch + decode cycles,
so the 17k queued-contact backlog drains in minutes instead of ~50.

lib/microwaveprop/workers/contact_weather_enqueue_worker.ex

@@ -6,10 +6,11 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
   alias Microwaveprop.Weather
   alias Microwaveprop.Weather.HrrrPointEnqueuer
   alias Microwaveprop.Weather.NarrClient
-  alias Microwaveprop.Workers.CommonVolumeRadarWorker
+  alias Microwaveprop.Weather.NexradClient
   alias Microwaveprop.Workers.IemreFetchWorker
   alias Microwaveprop.Workers.MechanismClassifyWorker
   alias Microwaveprop.Workers.NarrFetchWorker
+  alias Microwaveprop.Workers.RadarFrameWorker
   alias Microwaveprop.Workers.TerrainProfileWorker
   alias Microwaveprop.Workers.WeatherFetchWorker
 
@@ -220,14 +221,26 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
   end
 
   def build_radar_jobs(contacts) do
+    # Group contacts by their 5-min NEXRAD frame so the worker can
+    # fetch+decode the ~5 MB PNG once per frame and iterate every
+    # contact in that frame against the in-memory pixel buffer.
+    # Historical backlog averages ~10 contacts per distinct frame,
+    # collapsing the per-job cost by roughly that factor.
     contacts
     |> Enum.filter(fn contact ->
       contact.pos1 && contact.pos2 && not NarrClient.in_coverage?(contact.qso_timestamp)
     end)
-    |> Enum.map(fn contact ->
+    |> Enum.group_by(fn contact ->
-      CommonVolumeRadarWorker.new(%{"contact_id" => contact.id})
+      contact.qso_timestamp
+      |> DateTime.from_naive!("Etc/UTC")
+      |> NexradClient.round_to_5min()
+    end)
+    |> Enum.map(fn {frame_ts, batch} ->
+      RadarFrameWorker.new(%{
+        "frame_ts" => DateTime.to_iso8601(frame_ts),
+        "contact_ids" => Enum.map(batch, & &1.id)
+      })
     end)
-    |> Enum.uniq_by(fn changeset -> changeset.changes.args end)
   end
 
   def build_mechanism_jobs(contacts) do

lib/microwaveprop/workers/radar_frame_worker.ex

@@ -0,0 +1,119 @@
+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.
+  """
+
+  use Oban.Worker,
+    queue: :radar,
+    max_attempts: 3
+
+  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]
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: %{"frame_ts" => frame_iso, "contact_ids" => contact_ids}}) do
+    {:ok, frame_ts, _} = DateTime.from_iso8601(frame_iso)
+    rounded = NexradClient.round_to_5min(frame_ts)
+
+    contacts =
+      Contact
+      |> where([c], c.id in ^contact_ids)
+      |> 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)
+
+      {:error, reason} ->
+        Enum.each(contacts, &mark_status(&1, :unavailable))
+
+        Logger.info(
+          "RadarFrameWorker: no frame for #{DateTime.to_iso8601(rounded)} (#{length(contacts)} contacts): #{inspect(reason)}"
+        )
+    end
+
+    :ok
+  end
+
+  @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_all_except, [:id, :contact_id, :inserted_at]},
+      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

test/microwaveprop/workers/radar_frame_worker_test.exs

@@ -0,0 +1,107 @@
+defmodule Microwaveprop.Workers.RadarFrameWorkerTest do
+  use Microwaveprop.DataCase, async: false
+  use Oban.Testing, repo: Microwaveprop.Repo
+
+  alias Microwaveprop.Radio.Contact
+  alias Microwaveprop.Radio.ContactCommonVolumeRadar
+  alias Microwaveprop.Repo
+  alias Microwaveprop.Weather.NexradClient
+  alias Microwaveprop.Workers.RadarFrameWorker
+
+  setup do
+    Req.Test.set_req_test_from_context(%{async: false})
+    :ok
+  end
+
+  defp insert_contact(attrs \\ %{}) do
+    default = %{
+      station1: "W5XD",
+      station2: "K5TR",
+      qso_timestamp: ~U[2024-09-15 18:30:00Z],
+      mode: "CW",
+      band: Decimal.new("10000"),
+      grid1: "EM12",
+      grid2: "EM00",
+      pos1: %{"lat" => 32.9, "lon" => -97.0},
+      pos2: %{"lat" => 33.1, "lon" => -96.8},
+      distance_km: Decimal.new("30")
+    }
+
+    {:ok, contact} =
+      %Contact{}
+      |> Contact.changeset(Map.merge(default, attrs))
+      |> Repo.insert()
+
+    contact
+  end
+
+  describe "process_frame/4" do
+    test "writes a ContactCommonVolumeRadar row and marks :complete for every contact in the batch" do
+      c1 = insert_contact(%{station1: "A1"})
+      c2 = insert_contact(%{station1: "A2"})
+
+      # In-memory pixel buffer: 10x10 empty frame. The aggregator will
+      # observe zero echoes but still emit a stats row per contact.
+      pixels = :binary.copy(<<0>>, 100)
+      width = 10
+      frame_ts = ~U[2024-09-15 18:30:00Z]
+
+      assert :ok =
+               RadarFrameWorker.process_frame(
+                 [Repo.reload!(c1), Repo.reload!(c2)],
+                 frame_ts,
+                 pixels,
+                 width
+               )
+
+      assert Repo.get_by(ContactCommonVolumeRadar, contact_id: c1.id)
+      assert Repo.get_by(ContactCommonVolumeRadar, contact_id: c2.id)
+
+      assert %Contact{radar_status: :complete} = Repo.get!(Contact, c1.id)
+      assert %Contact{radar_status: :complete} = Repo.get!(Contact, c2.id)
+    end
+  end
+
+  describe "perform/1" do
+    test "marks every contact in the batch :unavailable when the frame 404s" do
+      c1 = insert_contact(%{station1: "B1"})
+      c2 = insert_contact(%{station1: "B2"})
+
+      Req.Test.stub(NexradClient, fn conn ->
+        Plug.Conn.send_resp(conn, 404, "not found")
+      end)
+
+      assert :ok =
+               perform_job(RadarFrameWorker, %{
+                 "frame_ts" => "2024-09-15T18:30:00Z",
+                 "contact_ids" => [c1.id, c2.id]
+               })
+
+      refute Repo.get_by(ContactCommonVolumeRadar, contact_id: c1.id)
+      refute Repo.get_by(ContactCommonVolumeRadar, contact_id: c2.id)
+      assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c1.id)
+      assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c2.id)
+    end
+
+    test "skips contacts already at a terminal status (idempotent replay)" do
+      c_done = insert_contact(%{station1: "DONE"})
+      {:ok, _} = c_done |> Ecto.Changeset.change(%{radar_status: :complete}) |> Repo.update()
+
+      c_new = insert_contact(%{station1: "NEW"})
+
+      Req.Test.stub(NexradClient, fn conn ->
+        Plug.Conn.send_resp(conn, 404, "not found")
+      end)
+
+      assert :ok =
+               perform_job(RadarFrameWorker, %{
+                 "frame_ts" => "2024-09-15T18:30:00Z",
+                 "contact_ids" => [c_done.id, c_new.id]
+               })
+
+      # Already-complete contact stays complete; new one is marked unavailable.
+      assert %Contact{radar_status: :complete} = Repo.get!(Contact, c_done.id)
+      assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c_new.id)
+    end
+  end
+end