feat(hrrr): route per-QSO enrichment through hrrr_fetch_tasks (Stream C Elixir)

Phase 3 Stream C Elixir-side: HrrrFetchWorker is deleted; per-QSO HRRR
enrichment now writes to the new hrrr_fetch_tasks table for the Rust
hrrr-point-worker to drain.

Table shape:
- one row per valid_time (primary key) with a JSONB array of
  {lat, lon} points
- UPSERT on conflict: array-union of points, status flips back to
  queued if previously done/failed so a backfill re-scan naturally
  refills the queue for Rust

Elixir changes:
- new migration 20260419231502_create_hrrr_fetch_tasks
- new Microwaveprop.Weather.HrrrPointEnqueuer with enqueue/1 and
  enqueue_for_contacts/1. Pre-2014 contacts (NARR's territory)
  are skipped here so hrrr_status can pin them to :unavailable.
- ContactWeatherEnqueueWorker: build_hrrr_jobs/1 removed; single-
  contact path and batch perform/1 both route through
  HrrrPointEnqueuer.enqueue_for_contacts/1. A placeholder jobs-list
  is kept just to feed mark_hrrr_status!.
- contact_live/show.ex retry button enqueues via the same path.
- :hrrr queue removed from dev/config/runtime.exs
- HrrrFetchWorker module + test deleted

BackfillEnqueueWorker scans continue to flow through
ContactWeatherEnqueueWorker.enqueue_for_contact (unchanged), so the
30-min reconcile refills hrrr_fetch_tasks automatically.

4 new tests cover the routing, pre-2014 skip, UPSERT-union, and
status-reset-on-reschedule behaviour. Rust-side hrrr-point-worker
binary + k8s deployment land in the next commits.

config/config.exs

@@ -73,7 +73,6 @@ config :microwaveprop, Oban,
     weather: 20,
     enqueue: 1,
     gefs: 2,
-    hrrr: 5,
     terrain: 4,
     commercial: 2,
     iemre: 10,

config/dev.exs

@@ -79,7 +79,6 @@ config :microwaveprop, Oban,
     weather: 20,
     enqueue: 1,
     gefs: 2,
-    hrrr: 5,
     terrain: 4,
     commercial: 2,
     iemre: 10,

config/runtime.exs

@@ -143,7 +143,8 @@ if config_env() == :prod do
     # forward progress.
     weather: 1,
     gefs: 1,
-    hrrr: 2,
+    # :hrrr queue removed — per-QSO HRRR fetches flow through
+    # hrrr_fetch_tasks and the Rust hrrr-point-worker (Phase 3 Stream C).
     terrain: 3,
     iemre: 3,
     # Historical backfill for pre-2014 contacts (pre-HRRR archive).

lib/microwaveprop/weather/hrrr_point_enqueuer.ex

@@ -0,0 +1,150 @@
+defmodule Microwaveprop.Weather.HrrrPointEnqueuer do
+  @moduledoc """
+  Inserts rows into `hrrr_fetch_tasks` for the Rust hrrr-point-worker
+  to drain.
+
+  Called from `Microwaveprop.Workers.ContactWeatherEnqueueWorker` in
+  place of the legacy `HrrrFetchWorker.new/1` fan-out. One row per
+  `valid_time` with a JSONB array of `{lat, lon}` points that share
+  that hour — additional points arriving from a later backfill tick
+  union into the existing row rather than creating a second one, so
+  the Rust worker fetches each GRIB2 once regardless of how many
+  contacts feed it.
+
+  `BackfillEnqueueWorker` re-discovers contacts missing HRRR enrichment
+  every 30 minutes; each scan flows through `enqueue_for_contact` and
+  lands here, which keeps the backfill pipeline intact after the
+  cutover.
+  """
+
+  import Ecto.Query
+
+  alias Microwaveprop.Repo
+
+  require Logger
+
+  @doc """
+  Enqueue `valid_time -> [{lat, lon}, ...]` groups into
+  `hrrr_fetch_tasks`. On conflict the incoming points are array-unioned
+  into the existing row and the status is reset to `queued` so a
+  previously-done valid_time re-scheduled by backfill picks up the new
+  points without duplicating a fetch.
+  """
+  @spec enqueue(%{DateTime.t() => [{float(), float()}]}) :: {:ok, non_neg_integer()}
+  def enqueue(groups) when is_map(groups) do
+    now = DateTime.truncate(DateTime.utc_now(), :microsecond)
+
+    count =
+      Enum.reduce(groups, 0, fn {valid_time, points}, acc ->
+        valid_time = DateTime.truncate(valid_time, :second)
+        json_points = Enum.map(points, fn {lat, lon} -> %{"lat" => lat, "lon" => lon} end)
+
+        # Union with the existing row's points (JSONB array) — use a
+        # subquery with jsonb_array_elements to dedupe. Postgres
+        # handles the set math so the Elixir side stays simple.
+        #
+        # Insert uses a wrapping schema query so Postgrex's jsonb
+        # extension encodes the list directly; raw Repo.query! with
+        # a binding would fall through to `text->jsonb` cast which
+        # then stores the JSON as a jsonb *string* and breaks the
+        # || union on round-trip.
+        id = Ecto.UUID.bingenerate()
+
+        Repo.insert_all(
+          "hrrr_fetch_tasks",
+          [
+            %{
+              id: id,
+              valid_time: valid_time,
+              points: json_points,
+              status: "queued",
+              attempt: 0,
+              inserted_at: now,
+              updated_at: now
+            }
+          ],
+          on_conflict:
+            from(t in "hrrr_fetch_tasks",
+              update: [
+                set: [
+                  points:
+                    fragment(
+                      "(SELECT COALESCE(jsonb_agg(DISTINCT p), '[]'::jsonb) FROM jsonb_array_elements(? || EXCLUDED.points) AS p)",
+                      t.points
+                    ),
+                  status:
+                    fragment(
+                      "CASE WHEN ? IN ('done', 'failed') THEN 'queued' ELSE ? END",
+                      t.status,
+                      t.status
+                    ),
+                  attempt:
+                    fragment(
+                      "CASE WHEN ? IN ('done', 'failed') THEN 0 ELSE ? END",
+                      t.status,
+                      t.attempt
+                    ),
+                  updated_at: ^now
+                ]
+              ]
+            ),
+          conflict_target: [:valid_time]
+        )
+
+        acc + 1
+      end)
+
+    {:ok, count}
+  rescue
+    e ->
+      Logger.error("HrrrPointEnqueuer: enqueue failed: #{inspect(e)}")
+      {:ok, 0}
+  end
+
+  @doc """
+  Convenience: given a list of `Contact`s, extract the
+  `(valid_time, [lat, lon])` groups and enqueue in one call. Mirrors
+  the signature of the removed `ContactWeatherEnqueueWorker.build_hrrr_jobs/1`
+  so the caller swap is a one-liner.
+  """
+  @spec enqueue_for_contacts([map()]) :: {:ok, non_neg_integer()}
+  def enqueue_for_contacts(contacts) do
+    alias Microwaveprop.Radio
+    alias Microwaveprop.Weather
+    alias Microwaveprop.Weather.HrrrClient
+    alias Microwaveprop.Weather.NarrClient
+
+    groups =
+      contacts
+      |> Enum.flat_map(fn contact ->
+        cond do
+          is_nil(contact.pos1) ->
+            []
+
+          # HRRR archive starts mid-2014; NARR covers the pre-2014
+          # window. Pre-coverage contacts are NARR's responsibility.
+          NarrClient.in_coverage?(contact.qso_timestamp) ->
+            []
+
+          true ->
+            rounded_time = HrrrClient.nearest_hrrr_hour(contact.qso_timestamp)
+
+            contact
+            |> Radio.contact_path_points()
+            |> Enum.flat_map(fn {lat, lon} ->
+              {rlat, rlon} = Weather.round_to_hrrr_grid(lat, lon)
+
+              if Weather.has_hrrr_profile?(rlat, rlon, rounded_time) do
+                []
+              else
+                [{rounded_time, {rlat, rlon}}]
+              end
+            end)
+        end
+      end)
+      |> Enum.group_by(fn {time, _pt} -> time end, fn {_time, pt} -> pt end)
+      |> Map.new(fn {time, pts} -> {time, Enum.uniq(pts)} end)
+
+    enqueue(groups)
+  end
+end

lib/microwaveprop/workers/contact_weather_enqueue_worker.ex

@@ -4,10 +4,9 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
 
   alias Microwaveprop.Radio
   alias Microwaveprop.Weather
-  alias Microwaveprop.Weather.HrrrClient
+  alias Microwaveprop.Weather.HrrrPointEnqueuer
   alias Microwaveprop.Weather.NarrClient
   alias Microwaveprop.Workers.CommonVolumeRadarWorker
-  alias Microwaveprop.Workers.HrrrFetchWorker
   alias Microwaveprop.Workers.IemreFetchWorker
   alias Microwaveprop.Workers.MechanismClassifyWorker
   alias Microwaveprop.Workers.NarrFetchWorker
@@ -30,7 +29,6 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
 
     bulk_jobs =
       jobs_by_type[:weather] ++
-        jobs_by_type[:hrrr] ++
         jobs_by_type[:terrain] ++
         jobs_by_type[:iemre] ++
         jobs_by_type[:radar] ++
@@ -40,6 +38,15 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
       insert_all_chunked(bulk_jobs)
     end
 
+    # HRRR enrichment runs via the Rust hrrr-point-worker via the
+    # hrrr_fetch_tasks table (Phase 3 Stream C). Grouping by valid_time
+    # and UPSERT-union means a backfill tick that adds a new contact in
+    # an already-scheduled hour collapses into the existing row instead
+    # of creating a duplicate fetch.
+    if :hrrr in types do
+      {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([contact])
+    end
+
     # NARR jobs must go through Oban.insert/1 so the worker's unique
     # constraint is honored; Oban OSS insert_all does not check uniqueness.
     insert_unique(jobs_by_type[:narr])
@@ -52,7 +59,10 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
   defp build_jobs_by_type(contact, types) do
     %{
       weather: if(:weather in types, do: build_weather_jobs([contact]), else: []),
-      hrrr: if(:hrrr in types, do: build_hrrr_jobs([contact]), else: []),
+      # hrrr jobs are routed to hrrr_fetch_tasks (not Oban) but the
+      # status-marking path still wants to know whether any HRRR work
+      # was created for this contact, so we mirror the list-shape.
+      hrrr: if(:hrrr in types, do: hrrr_placeholder_jobs([contact]), else: []),
       terrain: if(:terrain in types, do: build_terrain_jobs([contact]), else: []),
       iemre: if(:iemre in types, do: build_iemre_jobs([contact]), else: []),
       narr: if(:narr in types, do: build_narr_jobs([contact]), else: []),
@@ -61,6 +71,22 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
     }
   end
 
+  # Placeholder for the HRRR job-list signal. Returns a truthy singleton
+  # if the contact has at least one HRRR-eligible path point so
+  # mark_hrrr_status!/3 flips hrrr_queued on the contact row. Actual
+  # scheduling happens via HrrrPointEnqueuer.enqueue_for_contacts/1
+  # above. Returns [] for pre-coverage contacts (NARR's territory) so
+  # the status stays :unavailable.
+  defp hrrr_placeholder_jobs(contacts) do
+    Enum.flat_map(contacts, fn contact ->
+      cond do
+        is_nil(contact.pos1) -> []
+        NarrClient.in_coverage?(contact.qso_timestamp) -> []
+        true -> [contact.id]
+      end
+    end)
+  end
+
   defp mark_enrichment_statuses(contact, types, jobs_by_type) do
     ids = [contact.id]
 
@@ -152,11 +178,11 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
         :ok
 
       contacts ->
-        jobs = build_hrrr_jobs(contacts)
+        # Rust Stream C path: route into hrrr_fetch_tasks instead of
-
+        # Oban. The enqueuer groups by valid_time and UPSERTs so a
-        if jobs != [] do
+        # backfill cron sweep that rediscovers the same contacts
-          insert_all_chunked(jobs)
+        # simply refreshes the existing hourly row.
-        end
+        {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts(contacts)
 
         ids = Enum.map(contacts, & &1.id)
         Radio.mark_hrrr_queued!(ids)
@@ -219,23 +245,6 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
     |> Enum.uniq_by(fn changeset -> changeset.changes.args end)
   end
 
-  def build_hrrr_jobs(contacts) do
-    contacts
-    |> Enum.flat_map(&hrrr_points_for_contact/1)
-    |> Enum.group_by(fn {_point, hour} -> hour end, fn {point, _hour} -> point end)
-    |> Enum.map(fn {hour, points} ->
-      unique_points =
-        points
-        |> Enum.uniq()
-        |> Enum.map(fn {lat, lon} -> %{"lat" => lat, "lon" => lon} end)
-
-      HrrrFetchWorker.new(%{
-        "points" => unique_points,
-        "valid_time" => DateTime.to_iso8601(hour)
-      })
-    end)
-  end
-
   def build_iemre_jobs(contacts) do
     contacts
     |> Enum.flat_map(&iemre_job_for_contact/1)
@@ -314,37 +323,6 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do
     end
   end
 
-  defp hrrr_points_for_contact(%{pos1: nil}), do: []
-
-  defp hrrr_points_for_contact(contact) do
-    # HRRR archive starts mid-2014; NARR covers the 1979 → 2014-10-02 window.
-    # Short-circuit here so build_hrrr_jobs returns nothing for pre-coverage
-    # contacts and mark_hrrr_status! can pin them to :unavailable.
-    if NarrClient.in_coverage?(contact.qso_timestamp) do
-      []
-    else
-      hrrr_points_for_path(contact)
-    end
-  end
-
-  defp hrrr_points_for_path(contact) do
-    rounded_time = HrrrClient.nearest_hrrr_hour(contact.qso_timestamp)
-
-    contact
-    |> Radio.contact_path_points()
-    |> Enum.flat_map(&hrrr_point_or_skip(&1, rounded_time))
-  end
-
-  defp hrrr_point_or_skip({lat, lon}, rounded_time) do
-    {rlat, rlon} = Weather.round_to_hrrr_grid(lat, lon)
-
-    if Weather.has_hrrr_profile?(rlat, rlon, rounded_time) do
-      []
-    else
-      [{{rlat, rlon}, rounded_time}]
-    end
-  end
-
   defp build_asos_jobs(lat, lon, timestamp) do
     start_dt = DateTime.add(timestamp, -2 * 3600, :second)
     end_dt = DateTime.add(timestamp, 2 * 3600, :second)

lib/microwaveprop/workers/hrrr_fetch_worker.ex

@@ -1,157 +0,0 @@
-defmodule Microwaveprop.Workers.HrrrFetchWorker do
-  @moduledoc false
-  use Oban.Worker,
-    queue: :hrrr,
-    max_attempts: 20,
-    unique: [period: :infinity, states: [:available, :scheduled, :executing, :retryable]]
-
-  alias Microwaveprop.Weather
-  alias Microwaveprop.Weather.HrrrClient
-  alias Microwaveprop.Weather.SoundingParams
-
-  require Logger
-
-  @impl Oban.Worker
-  def backoff(%Oban.Job{attempt: attempt}) do
-    # Exponential backoff: 2m, 4m, 8m, 16m, 32m, 1h, 2h, 4h, 6h, 6h, ...
-    min(120 * Integer.pow(2, attempt - 1), _six_hours = 21_600)
-  end
-
-  @impl Oban.Worker
-  def perform(%Oban.Job{args: %{"points" => points, "valid_time" => valid_time_str}}) do
-    # Batch mode: download GRIB2 once, extract multiple points
-    {:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str)
-
-    point_tuples =
-      points
-      |> Enum.map(fn %{"lat" => lat, "lon" => lon} -> {lat, lon} end)
-      |> Enum.reject(fn {lat, lon} -> Weather.has_hrrr_profile?(lat, lon, valid_time) end)
-
-    result =
-      if point_tuples == [] do
-        Logger.info("HRRR batch: all #{length(points)} points already exist for #{valid_time_str}")
-        :ok
-      else
-        Logger.info("HRRR batch: fetching #{length(point_tuples)} points for #{valid_time_str}")
-        fetch_and_store_batch(point_tuples, valid_time, valid_time_str)
-      end
-
-    # Reclaim the refc binary heap held from GRIB2 ranges before the
-    # Oban producer hands this process its next job. Without this each
-    # worker steady-states at 100+ MiB of retained HRRR payload.
-    :erlang.garbage_collect(self())
-    result
-  end
-
-  def perform(%Oban.Job{args: %{"lat" => raw_lat, "lon" => raw_lon, "valid_time" => valid_time_str}}) do
-    # Legacy single-point mode
-    {:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str)
-    {lat, lon} = Weather.round_to_hrrr_grid(raw_lat, raw_lon)
-
-    if Weather.has_hrrr_profile?(lat, lon, valid_time) do
-      Logger.info("HRRR profile already exists for #{lat},#{lon} @ #{valid_time_str}")
-      :ok
-    else
-      Logger.info("Fetching HRRR profile for #{lat},#{lon} @ #{valid_time_str}")
-
-      case HrrrClient.fetch_profile(lat, lon, valid_time) do
-        {:ok, data} ->
-          store_profile(lat, lon, valid_time, data)
-          broadcast_enrichment(lat, lon, valid_time)
-          :ok
-
-        {:error, reason} ->
-          handle_error(reason, "#{lat},#{lon} @ #{valid_time_str}")
-      end
-    end
-  end
-
-  defp fetch_and_store_batch(point_tuples, valid_time, valid_time_str) do
-    case HrrrClient.fetch_grid(point_tuples, valid_time) do
-      {:ok, grid_data} ->
-        profiles =
-          Enum.map(grid_data, fn {{lat, lon}, data} ->
-            build_profile_attrs(lat, lon, valid_time, data)
-          end)
-
-        Weather.upsert_hrrr_profiles_batch(profiles, skip_existing: true)
-
-        Enum.each(grid_data, fn {{lat, lon}, _data} ->
-          broadcast_enrichment(lat, lon, valid_time)
-        end)
-
-        Logger.info("HRRR batch: saved #{map_size(grid_data)} profiles for #{valid_time_str}")
-        :ok
-
-      {:error, reason} ->
-        handle_error(reason, "batch @ #{valid_time_str}")
-    end
-  end
-
-  defp store_profile(lat, lon, valid_time, data) do
-    attrs = build_profile_attrs(lat, lon, valid_time, data)
-    Weather.upsert_hrrr_profile(attrs)
-  end
-
-  defp build_profile_attrs(lat, lon, valid_time, data) do
-    params = SoundingParams.derive(data.profile)
-
-    maybe_add_derived_params(
-      %{
-        valid_time: valid_time,
-        lat: lat,
-        lon: lon,
-        run_time: data.run_time,
-        profile: data.profile,
-        hpbl_m: data.hpbl_m,
-        pwat_mm: data.pwat_mm,
-        surface_temp_c: data.surface_temp_c,
-        surface_dewpoint_c: data.surface_dewpoint_c,
-        surface_pressure_mb: data.surface_pressure_mb
-      },
-      params
-    )
-  end
-
-  defp handle_error(reason, label) do
-    if transient_failure?(reason) do
-      Logger.error("HRRR transient error for #{label}: #{inspect(reason)}")
-      {:error, reason}
-    else
-      Logger.warning("HRRR permanent failure for #{label}: #{inspect(reason)}")
-      {:cancel, reason}
-    end
-  end
-
-  # Only retry on transient network/server errors — everything else is permanent
-  defp transient_failure?(%{__exception__: true}), do: true
-  defp transient_failure?("HRRR idx HTTP " <> status), do: server_error?(status)
-  defp transient_failure?("HRRR grib HTTP " <> status), do: server_error?(status)
-  defp transient_failure?(_), do: false
-
-  defp server_error?(status) do
-    case Integer.parse(status) do
-      {code, _} when code in [429, 500, 502, 503, 504] -> true
-      _ -> false
-    end
-  end
-
-  defp maybe_add_derived_params(attrs, nil), do: attrs
-
-  defp maybe_add_derived_params(attrs, params) do
-    Map.merge(attrs, %{
-      surface_refractivity: params.surface_refractivity,
-      min_refractivity_gradient: params.min_refractivity_gradient,
-      ducting_detected: params.ducting_detected,
-      duct_characteristics: params.duct_characteristics
-    })
-  end
-
-  defp broadcast_enrichment(lat, lon, valid_time) do
-    Phoenix.PubSub.broadcast(
-      Microwaveprop.PubSub,
-      "contact_enrichment:hrrr",
-      {:hrrr_ready, %{lat: lat, lon: lon, valid_time: valid_time}}
-    )
-  end
-end

lib/microwaveprop_web/live/contact_live/show.ex

@@ -13,9 +13,9 @@ defmodule MicrowavepropWeb.ContactLive.Show do
   alias Microwaveprop.Weather
   alias Microwaveprop.Weather.HrrrClient
   alias Microwaveprop.Weather.HrrrNativeProfile
+  alias Microwaveprop.Weather.HrrrPointEnqueuer
   alias Microwaveprop.Weather.NarrClient
   alias Microwaveprop.Workers.ContactWeatherEnqueueWorker
-  alias Microwaveprop.Workers.HrrrFetchWorker
   alias Microwaveprop.Workers.NarrFetchWorker
   alias Microwaveprop.Workers.SolarIndexWorker
   alias Microwaveprop.Workers.TerrainProfileWorker
@@ -752,22 +752,11 @@ defmodule MicrowavepropWeb.ContactLive.Show do
       valid_time = HrrrClient.nearest_hrrr_hour(contact.qso_timestamp)
       {rlat, rlon} = Weather.round_to_hrrr_grid(lat, lon)
 
-      case Oban.insert(
+      # Phase 3 Stream C: route retries through hrrr_fetch_tasks so the
-             HrrrFetchWorker.new(%{
+      # Rust hrrr-point-worker picks them up alongside backfill work.
-               "lat" => rlat,
+      {:ok, _} = HrrrPointEnqueuer.enqueue(%{valid_time => [{rlat, rlon}]})
-               "lon" => rlon,
+      Radio.set_enrichment_status!([contact.id], :hrrr_status, :queued)
-               "valid_time" => DateTime.to_iso8601(valid_time)
+      {nil, %{contact | hrrr_status: :queued}}
-             })
-           ) do
-        {:ok, %{conflict?: false}} ->
-          Radio.set_enrichment_status!([contact.id], :hrrr_status, :queued)
-          {nil, %{contact | hrrr_status: :queued}}
-
-        _ ->
-          # Unique conflict — job already ran recently, mark unavailable
-          Radio.set_enrichment_status!([contact.id], :hrrr_status, :unavailable)
-          {nil, %{contact | hrrr_status: :unavailable}}
-      end
     else
       {nil, contact}
     end

priv/repo/migrations/20260419231502_create_hrrr_fetch_tasks.exs

@@ -0,0 +1,43 @@
+defmodule Microwaveprop.Repo.Migrations.CreateHrrrFetchTasks do
+  use Ecto.Migration
+
+  # Hand-off queue between the Elixir enrichment enqueuer and the Rust
+  # hrrr-point-worker (Phase 3 Stream C). One row per valid_time with a
+  # JSONB array of {lat, lon} points the workers needs to extract from
+  # that hour's HRRR cycle. Rust claims by SKIP LOCKED, fetches the
+  # GRIB2 once, extracts every point in the batch, bulk-upserts to
+  # hrrr_profiles, and marks the row done.
+  def change do
+    create table(:hrrr_fetch_tasks, primary_key: false) do
+      add :id, :binary_id, primary_key: true
+      add :valid_time, :utc_datetime, null: false
+
+      # [{"lat": f, "lon": f}, …] — additional points for the same
+      # valid_time get merged via jsonb array-union UPSERT so a
+      # backfill tick doesn't schedule a new row per contact.
+      add :points, :map, null: false, default: fragment("'[]'::jsonb")
+
+      add :status, :string, null: false, default: "queued"
+      add :attempt, :integer, null: false, default: 0
+      add :claimed_at, :utc_datetime_usec
+      add :completed_at, :utc_datetime_usec
+      add :error, :text
+
+      timestamps(type: :utc_datetime)
+    end
+
+    # One task per valid_time. Merges all points for the hour into a
+    # single fetch so a full backfill scan costs one GRIB2 download
+    # per hour, not one per contact.
+    create unique_index(:hrrr_fetch_tasks, [:valid_time])
+
+    # Hot-path claim query scans the queued subset.
+    create index(:hrrr_fetch_tasks, [:status],
+             where: "status = 'queued'",
+             name: :hrrr_fetch_tasks_queued_idx
+           )
+
+    # Prune index for retention.
+    create index(:hrrr_fetch_tasks, [:completed_at])
+  end
+end

test/microwaveprop/workers/contact_weather_enqueue_worker_test.exs

@@ -255,122 +255,61 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorkerTest do
     end
   end
 
-  describe "build_hrrr_jobs/1" do
+  describe "HRRR enrichment routing (Phase 3 Stream C → hrrr_fetch_tasks)" do
-    test "batches all path points into one job per HRRR hour" do
+    alias Microwaveprop.Weather.HrrrPointEnqueuer
+
+    test "enqueue_for_contacts writes one hrrr_fetch_tasks row per valid_time" do
+      import Ecto.Query
+
       contact = create_contact(%{pos1: %{"lat" => 32.907, "lon" => -97.038}})
+      {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([contact])
 
-      jobs = ContactWeatherEnqueueWorker.build_hrrr_jobs([contact])
+      rows =
+        Repo.all(from t in "hrrr_fetch_tasks", select: %{vt: t.valid_time, points: t.points, status: t.status})
 
-      # All 3 path points share the same HRRR hour → 1 batch job
+      assert [row] = rows
-      assert length(jobs) == 1
+      # pos1 + midpoint + pos2 → 3 distinct HRRR grid cells in the batch.
-      job = hd(jobs)
+      assert length(row.points) == 3
-      points = job.changes.args["points"]
+      assert row.status == "queued"
-      # pos1, midpoint, pos2 = 3 distinct grid points in the batch
-      assert length(points) == 3
-      lats = Enum.map(points, & &1["lat"])
-      assert 32.91 in lats
     end
 
-    test "batch contains single point when pos2 is nil" do
+    test "enqueue_for_contacts skips pre-2014 contacts (NARR's territory)" do
-      contact = create_contact(%{pos1: %{"lat" => 32.907, "lon" => -97.038}, pos2: nil})
+      import Ecto.Query
 
-      jobs = ContactWeatherEnqueueWorker.build_hrrr_jobs([contact])
-
-      assert length(jobs) == 1
-      job = hd(jobs)
-      [point] = job.changes.args["points"]
-      assert point["lat"] == 32.91
-      assert point["lon"] == -97.04
-    end
-
-    test "rounds valid_time to nearest hour" do
-      contact = create_contact(%{qso_timestamp: ~U[2026-03-28 18:45:00Z], pos2: nil})
-
-      jobs = ContactWeatherEnqueueWorker.build_hrrr_jobs([contact])
-
-      job = hd(jobs)
-      assert job.changes.args["valid_time"] == "2026-03-28T19:00:00Z"
-    end
-
-    test "deduplicates jobs across QSOs with identical path points" do
-      q1 =
-        create_contact(%{
-          station1: "A1",
-          qso_timestamp: ~U[2026-03-28 18:10:00Z],
-          pos1: %{"lat" => 32.90, "lon" => -97.04},
-          pos2: nil
-        })
-
-      q2 =
-        create_contact(%{
-          station1: "A2",
-          qso_timestamp: ~U[2026-03-28 18:20:00Z],
-          pos1: %{"lat" => 32.90, "lon" => -97.04},
-          pos2: nil
-        })
-
-      jobs = ContactWeatherEnqueueWorker.build_hrrr_jobs([q1, q2])
-
-      # Both round to same hour and same location → 1 job
-      assert length(jobs) == 1
-    end
-
-    test "deduplicates path points that round to the same HRRR grid cell" do
-      # pos1 and pos2 only 0.001° apart — all 3 points round to same HRRR cell
-      contact =
-        create_contact(%{
-          pos1: %{"lat" => 32.901, "lon" => -97.041},
-          pos2: %{"lat" => 32.902, "lon" => -97.042}
-        })
-
-      jobs = ContactWeatherEnqueueWorker.build_hrrr_jobs([contact])
-
-      # All 3 path points round to {32.90, -97.04} → 1 job
-      assert length(jobs) == 1
-    end
-
-    test "skips QSOs without pos1" do
-      contact = create_contact(%{pos1: nil})
-
-      assert ContactWeatherEnqueueWorker.build_hrrr_jobs([contact]) == []
-    end
-
-    # HRRR archive starts mid-2014; pre-coverage fetches always 404. Skipping
-    # here stops the :queued ↔ :unavailable ping-pong in BackfillEnqueueWorker
-    # where the reconcile flips pre-2014 rows to :unavailable and the same
-    # run immediately flips them back to :queued from a build_hrrr_jobs call
-    # that never had a chance of succeeding.
-    test "skips pre-2014 contacts (outside HRRR coverage — NARR's job)" do
       contact = create_contact(%{qso_timestamp: ~U[2010-06-15 18:00:00Z]})
+      {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([contact])
 
-      assert ContactWeatherEnqueueWorker.build_hrrr_jobs([contact]) == []
+      assert [] = Repo.all(from t in "hrrr_fetch_tasks", select: t.id)
     end
 
-    test "excludes existing points from batch" do
+    test "re-enqueuing the same valid_time UPSERT-unions the points" do
-      contact = create_contact(%{pos1: %{"lat" => 32.907, "lon" => -97.038}})
+      import Ecto.Query
 
-      # Insert a profile at the pos1 rounded grid point only
+      c1 = create_contact(%{station1: "A1", pos1: %{"lat" => 32.90, "lon" => -97.04}, pos2: nil})
-      Weather.upsert_hrrr_profile(%{
+      c2 = create_contact(%{station1: "A2", pos1: %{"lat" => 33.50, "lon" => -96.50}, pos2: nil})
-        valid_time: ~U[2026-03-28 18:00:00Z],
-        lat: 32.91,
-        lon: -97.04,
-        run_time: ~U[2026-03-28 18:00:00Z],
-        profile: [%{"pres" => 1000.0, "tmpc" => 25.0}],
-        hpbl_m: 1500.0,
-        pwat_mm: 25.0,
-        surface_temp_c: 25.0,
-        surface_dewpoint_c: 18.0,
-        surface_pressure_mb: 1013.0
-      })
 
-      jobs = ContactWeatherEnqueueWorker.build_hrrr_jobs([contact])
+      {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([c1])
+      {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([c2])
 
-      # 1 batch job, but pos1 excluded from points list
+      rows = Repo.all(from t in "hrrr_fetch_tasks", select: %{vt: t.valid_time, points: t.points})
-      assert length(jobs) == 1
+      assert [row] = rows
-      points = hd(jobs).changes.args["points"]
+      assert length(row.points) == 2
-      assert length(points) == 2
+    end
-      lats = Enum.map(points, & &1["lat"])
+
-      refute 32.91 in lats
+    test "re-enqueuing a 'done' row flips status back to queued" do
+      import Ecto.Query
+
+      contact = create_contact(%{pos1: %{"lat" => 32.90, "lon" => -97.04}, pos2: nil})
+      {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([contact])
+
+      Repo.update_all(
+        from(t in "hrrr_fetch_tasks"),
+        set: [status: "done", completed_at: DateTime.utc_now()]
+      )
+
+      {:ok, _} = HrrrPointEnqueuer.enqueue_for_contacts([contact])
+      assert [row] = Repo.all(from t in "hrrr_fetch_tasks", select: %{status: t.status})
+      assert row.status == "queued"
     end
   end
 

test/microwaveprop/workers/hrrr_fetch_worker_test.exs

@@ -1,152 +0,0 @@
-defmodule Microwaveprop.Workers.HrrrFetchWorkerTest do
-  use Microwaveprop.DataCase, async: true
-
-  alias Microwaveprop.Weather
-  alias Microwaveprop.Weather.HrrrProfile
-  alias Microwaveprop.Workers.HrrrFetchWorker
-
-  @sample_profile [
-    %{"pres" => 1000.0, "hght" => 110.0, "tmpc" => 25.0, "dwpc" => 18.0},
-    %{"pres" => 975.0, "hght" => 350.0, "tmpc" => 23.0, "dwpc" => 16.0},
-    %{"pres" => 950.0, "hght" => 590.0, "tmpc" => 21.0, "dwpc" => 14.0}
-  ]
-
-  @sample_client_result %{
-    surface_temp_c: 25.0,
-    surface_dewpoint_c: 18.0,
-    surface_pressure_mb: 1013.0,
-    hpbl_m: 1500.0,
-    pwat_mm: 25.0,
-    run_time: ~U[2026-03-28 18:00:00Z],
-    profile: @sample_profile
-  }
-
-  describe "perform/1" do
-    test "skips if HRRR profile already exists" do
-      Weather.upsert_hrrr_profile(%{
-        valid_time: ~U[2026-03-28 18:00:00Z],
-        lat: 32.90,
-        lon: -97.04,
-        profile: @sample_profile
-      })
-
-      job =
-        HrrrFetchWorker.new(%{
-          "lat" => 32.90,
-          "lon" => -97.04,
-          "valid_time" => "2026-03-28T18:00:00Z"
-        })
-
-      assert :ok = HrrrFetchWorker.perform(%Oban.Job{args: job.changes.args})
-    end
-
-    test "stores profile when client returns success" do
-      # We test the perform logic by mocking at the module level
-      # The actual HTTP calls are tested in HrrrClientTest
-      # Here we verify the worker's coordination logic
-
-      lat = 32.90
-      lon = -97.04
-      valid_time = ~U[2026-03-28 18:00:00Z]
-
-      # Verify no profile exists initially
-      refute Weather.has_hrrr_profile?(lat, lon, valid_time)
-
-      # Simulate what perform does after a successful fetch
-      attrs = build_profile_attrs(lat, lon, valid_time, @sample_client_result)
-      {:ok, profile} = Weather.upsert_hrrr_profile(attrs)
-
-      assert profile.lat == lat
-      assert profile.lon == lon
-      assert profile.hpbl_m == 1500.0
-      assert profile.surface_temp_c == 25.0
-      assert Weather.has_hrrr_profile?(lat, lon, valid_time)
-    end
-
-    test "rounds lat/lon before checking existence" do
-      # Profile at rounded coordinates
-      Weather.upsert_hrrr_profile(%{
-        valid_time: ~U[2026-03-28 18:00:00Z],
-        lat: 32.90,
-        lon: -97.04,
-        profile: []
-      })
-
-      # Worker called with slightly different coords that round to same
-      job =
-        HrrrFetchWorker.new(%{
-          "lat" => 32.904,
-          "lon" => -97.039,
-          "valid_time" => "2026-03-28T18:00:00Z"
-        })
-
-      # Should skip because rounded coords match
-      assert :ok = HrrrFetchWorker.perform(%Oban.Job{args: job.changes.args})
-
-      # Only one profile should exist
-      count = Repo.aggregate(HrrrProfile, :count)
-      assert count == 1
-    end
-  end
-
-  describe "perform/1 batch mode" do
-    test "skips all points when profiles already exist" do
-      Weather.upsert_hrrr_profile(%{
-        valid_time: ~U[2026-03-28 18:00:00Z],
-        lat: 32.90,
-        lon: -97.04,
-        profile: @sample_profile
-      })
-
-      job = %Oban.Job{
-        args: %{
-          "points" => [%{"lat" => 32.90, "lon" => -97.04}],
-          "valid_time" => "2026-03-28T18:00:00Z"
-        }
-      }
-
-      assert :ok = HrrrFetchWorker.perform(job)
-    end
-  end
-
-  describe "backoff/1" do
-    test "returns exponential backoff capped at 6 hours" do
-      assert HrrrFetchWorker.backoff(%Oban.Job{attempt: 1}) == 120
-      assert HrrrFetchWorker.backoff(%Oban.Job{attempt: 2}) == 240
-      assert HrrrFetchWorker.backoff(%Oban.Job{attempt: 10}) <= 21_600
-    end
-  end
-
-  describe "permanent_failure?/1" do
-    test "cancels on 404 idx" do
-      job = %Oban.Job{
-        args: %{
-          "lat" => 32.90,
-          "lon" => -97.04,
-          "valid_time" => "1995-06-15T12:00:00Z"
-        }
-      }
-
-      Req.Test.stub(Microwaveprop.Weather.HrrrClient, fn conn ->
-        Plug.Conn.send_resp(conn, 404, "not found")
-      end)
-
-      assert {:cancel, "HRRR idx HTTP 404"} = HrrrFetchWorker.perform(job)
-    end
-  end
-
-  defp build_profile_attrs(lat, lon, valid_time, client_result) do
-    %{
-      valid_time: valid_time,
-      lat: lat,
-      lon: lon,
-      run_time: client_result.run_time,
-      profile: client_result.profile,
-      hpbl_m: client_result.hpbl_m,
-      pwat_mm: client_result.pwat_mm,
-      surface_temp_c: client_result.surface_temp_c,
-      surface_dewpoint_c: client_result.surface_dewpoint_c,
-      surface_pressure_mb: client_result.surface_pressure_mb
-    }
-  end
-end