diff --git a/lib/microwaveprop/workers/narr_fetch_worker.ex b/lib/microwaveprop/workers/narr_fetch_worker.ex new file mode 100644 index 00000000..0d55d8f9 --- /dev/null +++ b/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 diff --git a/test/microwaveprop/workers/narr_fetch_worker_test.exs b/test/microwaveprop/workers/narr_fetch_worker_test.exs new file mode 100644 index 00000000..401ed835 --- /dev/null +++ b/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