Add Microwaveprop.Workers.NarrFetchWorker

Per-contact Oban worker that fetches one NARR profile via
NarrClient.fetch_profile_at/2 and inserts it into era5_profiles.

Replaces Era5FetchWorker for pre-2014 contacts. Unlike the ERA5
router, NARR fetches are per-point so this worker does the fetch
+ insert directly — no downstream batch worker, no routing.

Args shape matches Era5FetchWorker: %{"lat", "lon", "valid_time"}.
Lat/lon get snapped to 0.25° (same dedup granularity as ERA5).
valid_time MUST already be on a 3-hourly NARR analysis slot
(00/03/06/09/12/15/18/21 UTC on the hour) — the worker raises
ArgumentError via NarrClient.url_for/1 otherwise. Caller-side
snapping (in ContactWeatherEnqueueWorker) lands in the next task.

Existence check tightened to ±15 minutes (vs ERA5's ±30) since NARR
is exact 3-hourly. On error, logs a warning and returns {:error,
reason} so Oban retries with backoff (max_attempts: 3).

The :narr Oban queue config and the ContactWeatherEnqueueWorker
swap are deliberately NOT touched in this commit — those are the
next two tasks of the plan.

lib/microwaveprop/workers/narr_fetch_worker.ex

@@ -0,0 +1,101 @@
+defmodule Microwaveprop.Workers.NarrFetchWorker do
+  @moduledoc """
+  Fetches one NCEP NARR profile from NCEI via `NarrClient.fetch_profile_at/2`
+  and inserts it into the `era5_profiles` table.
+
+  Replaces `Era5FetchWorker` for pre-2014 contacts. Unlike the ERA5 router,
+  NARR fetches are per-point (one analysis hour, one location), so this
+  worker does the fetch + insert directly — no downstream batch worker,
+  no routing.
+
+  Jobs are unique on `{lat, lon, valid_time}` so duplicate enrichment
+  requests collapse. `valid_time` MUST already be snapped to a NARR
+  3-hourly analysis slot (00/03/06/09/12/15/18/21 UTC on the hour) by the
+  caller — the worker raises `ArgumentError` otherwise via
+  `NarrClient.url_for/1`.
+
+  See `docs/plans/2026-04-15-merra2-historical-backfill.md` for the full
+  architecture. The `era5_profiles` table is reused 1:1 (rename is a
+  follow-up).
+  """
+  use Oban.Worker,
+    queue: :narr,
+    max_attempts: 3,
+    unique: [
+      period: :infinity,
+      states: [:available, :scheduled, :executing, :retryable]
+    ]
+
+  import Ecto.Query
+
+  alias Microwaveprop.Repo
+  alias Microwaveprop.Weather.Era5Profile
+  alias Microwaveprop.Weather.NarrClient
+
+  require Logger
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: %{"lat" => lat, "lon" => lon, "valid_time" => valid_time_str}}) do
+    {:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str)
+
+    rlat = Float.round(lat * 4) / 4
+    rlon = Float.round(lon * 4) / 4
+
+    if has_era5_profile?(rlat, rlon, valid_time) do
+      :ok
+    else
+      fetch_and_insert(rlat, rlon, valid_time)
+    end
+  end
+
+  defp fetch_and_insert(rlat, rlon, valid_time) do
+    case NarrClient.fetch_profile_at(valid_time, {rlat, rlon}) do
+      {:ok, attrs} ->
+        insert_profile(attrs, rlat, rlon, valid_time)
+        :ok
+
+      {:error, reason} = error ->
+        Logger.warning(
+          "NARR: fetch_profile_at failed for #{rlat},#{rlon} @ #{DateTime.to_iso8601(valid_time)}: #{inspect(reason)}"
+        )
+
+        error
+    end
+  end
+
+  defp insert_profile(attrs, rlat, rlon, valid_time) do
+    now = DateTime.truncate(DateTime.utc_now(), :second)
+
+    row =
+      attrs
+      |> Map.put(:lat, rlat)
+      |> Map.put(:lon, rlon)
+      |> Map.put(:valid_time, valid_time)
+      |> Map.put(:inserted_at, now)
+      |> Map.put(:updated_at, now)
+
+    Repo.insert_all(Era5Profile, [row],
+      on_conflict: :nothing,
+      conflict_target: [:lat, :lon, :valid_time]
+    )
+  end
+
+  defp has_era5_profile?(lat, lon, valid_time) do
+    # NARR is exact 3-hourly, so we tighten the time window to ±15 minutes
+    # compared to Era5FetchWorker's ±30-minute check. Lat/lon tolerance
+    # matches the 0.25° dedup granularity.
+    dlat = 0.13
+    dlon = 0.13
+    time_start = DateTime.add(valid_time, -900, :second)
+    time_end = DateTime.add(valid_time, 900, :second)
+
+    Era5Profile
+    |> where(
+      [p],
+      p.lat >= ^(lat - dlat) and p.lat <= ^(lat + dlat) and
+        p.lon >= ^(lon - dlon) and p.lon <= ^(lon + dlon) and
+        p.valid_time >= ^time_start and p.valid_time <= ^time_end
+    )
+    |> Repo.exists?()
+  end
+end

test/microwaveprop/workers/narr_fetch_worker_test.exs

@@ -0,0 +1,143 @@
+defmodule Microwaveprop.Workers.NarrFetchWorkerTest do
+  use Microwaveprop.DataCase, async: false
+  use Oban.Testing, repo: Microwaveprop.Repo
+
+  alias Microwaveprop.Weather.Era5Profile
+  alias Microwaveprop.Weather.NarrClient
+  alias Microwaveprop.Workers.NarrFetchWorker
+
+  @inv_fixture "test/fixtures/narr/narr-a_221_20100615_1200_000.inv"
+  @grb_fixture "test/fixtures/narr/narr_dfw_2010-06-15_12z.grb"
+  @file_size 56_221_430
+
+  # Stub NarrClient HTTP so `fetch_profile_at/2` returns the spike fixture
+  # composite without hitting NCEI.
+  defp stub_narr_success do
+    inv_body = File.read!(@inv_fixture)
+    grb_body = File.read!(@grb_fixture)
+
+    Req.Test.stub(NarrClient, fn conn ->
+      cond do
+        String.ends_with?(conn.request_path, ".inv") and conn.method == "GET" ->
+          Plug.Conn.send_resp(conn, 200, inv_body)
+
+        String.ends_with?(conn.request_path, ".grb") and conn.method == "HEAD" ->
+          conn
+          |> Plug.Conn.put_resp_header("content-length", Integer.to_string(@file_size))
+          |> Plug.Conn.send_resp(200, "")
+
+        String.ends_with?(conn.request_path, ".grb") and conn.method == "GET" ->
+          Plug.Conn.send_resp(conn, 200, grb_body)
+
+        true ->
+          Plug.Conn.send_resp(conn, 500, "unexpected #{conn.method} #{conn.request_path}")
+      end
+    end)
+  end
+
+  describe "perform/1" do
+    test "fetches a profile via NarrClient and inserts it into era5_profiles" do
+      stub_narr_success()
+
+      valid_time = ~U[2010-06-15 12:00:00Z]
+
+      args = %{
+        "lat" => 32.9,
+        "lon" => -97.0,
+        "valid_time" => DateTime.to_iso8601(valid_time)
+      }
+
+      assert :ok = NarrFetchWorker.perform(%Oban.Job{args: args})
+
+      # Row snapped to 0.25° grid (32.9 → 33.0, -97.0 → -97.0) and stored at
+      # the exact 3-hourly valid_time.
+      assert [row] = Repo.all(Era5Profile)
+      assert row.lat == 33.0
+      assert row.lon == -97.0
+      assert DateTime.compare(row.valid_time, valid_time) == :eq
+      assert_in_delta row.surface_temp_c, 23.79, 1.0
+      assert_in_delta row.surface_dewpoint_c, 21.34, 1.0
+      assert_in_delta row.surface_pressure_mb, 993.84, 2.0
+      assert_in_delta row.hpbl_m, 703.5, 10.0
+      assert_in_delta row.pwat_mm, 45.1, 2.0
+    end
+
+    test "is a no-op when an era5_profiles row already exists for the snapped lat/lon/time" do
+      valid_time = ~U[2010-06-15 12:00:00Z]
+
+      # Stub that raises if called — this test asserts no HTTP happens.
+      Req.Test.stub(NarrClient, fn _conn ->
+        raise "NarrClient should not be contacted when cache hit"
+      end)
+
+      # Pre-populate era5_profiles with a row at the snapped coordinates.
+      %Era5Profile{}
+      |> Era5Profile.changeset(%{
+        lat: 33.0,
+        lon: -97.0,
+        valid_time: valid_time,
+        profile: []
+      })
+      |> Repo.insert!()
+
+      args = %{
+        "lat" => 32.9,
+        "lon" => -97.0,
+        "valid_time" => DateTime.to_iso8601(valid_time)
+      }
+
+      assert :ok = NarrFetchWorker.perform(%Oban.Job{args: args})
+
+      # Only the pre-seeded row exists — no new insert.
+      assert Repo.aggregate(Era5Profile, :count, :id) == 1
+    end
+
+    test "returns {:error, _} when NarrClient.fetch_profile_at fails" do
+      Req.Test.stub(NarrClient, fn conn ->
+        Plug.Conn.send_resp(conn, 500, "ncei server error")
+      end)
+
+      args = %{
+        "lat" => 32.9,
+        "lon" => -97.0,
+        "valid_time" => "2010-06-15T12:00:00Z"
+      }
+
+      assert {:error, _reason} = NarrFetchWorker.perform(%Oban.Job{args: args})
+      assert Repo.aggregate(Era5Profile, :count, :id) == 0
+    end
+
+    test "raises ArgumentError when valid_time is not on a 3-hourly NARR slot" do
+      # 13:00 is NOT a NARR analysis hour (only 00/03/06/09/12/15/18/21).
+      args = %{
+        "lat" => 32.9,
+        "lon" => -97.0,
+        "valid_time" => "2010-06-15T13:00:00Z"
+      }
+
+      assert_raise ArgumentError, fn ->
+        NarrFetchWorker.perform(%Oban.Job{args: args})
+      end
+    end
+  end
+
+  describe "unique constraint" do
+    test "duplicate enqueues collapse via Oban unique" do
+      args = %{
+        "lat" => 32.9,
+        "lon" => -97.0,
+        "valid_time" => "2010-06-15T12:00:00Z"
+      }
+
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        {:ok, job1} = Oban.insert(NarrFetchWorker.new(args))
+        {:ok, job2} = Oban.insert(NarrFetchWorker.new(args))
+
+        # Second insert should hit the unique constraint and return the
+        # existing job (same id) instead of creating a new one.
+        assert job1.id == job2.id
+        assert Repo.aggregate(Oban.Job, :count, :id) == 1
+      end)
+    end
+  end
+end