feat(gefs): score extended-horizon grid and seed cron

Wires the GEFS fetch pipeline into the existing scoring machinery:

- GefsFetchWorker runs Propagation.score_grid_point over every
  fetched grid cell and persists the result via replace_scores/2, so
  Day 2-7 valid_times show up on the map through the same
  Propagation.scores_at/3 path HRRR uses
- Empty-args perform/1 seeds the chain by enqueueing f024-f168 (25
  jobs at 6-hour cadence) for the most recent GEFS run that should
  be published given NOMADS' ~3-4h publication lag
- Cron: seeds at 05:30, 11:30, 17:30, 23:30 UTC — 5 hours after each
  00/06/12/18Z run
- GefsClient.build_profile now emits :wind_u / :wind_v atom keys to
  match the scorer's input shape; DB columns keep the *_mps suffix
  for unit clarity, remapped at write time

GEFS pgrb2a lacks HPBL, native-level duct data, NEXRAD, and
commercial-link degradation — the extended-horizon score is a
rougher signal than the HRRR-driven one but covers the window HRRR
can't reach.

config/runtime.exs

@@ -210,6 +210,11 @@ if config_env() == :prod do
          # are keyed by (band, valid_time) so last-writer-wins gives
          # newer analysis data naturally.
          {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker},
+         # GEFS runs publish ~3-4h after the 00/06/12/18Z cycle; seed the
+         # Day 2-7 outlook 5h after each run to stay safely past NOMADS'
+         # publication lag. Seeder enqueues f024-f168 at 6-hour cadence
+         # (25 jobs per run, 100/day) into the :gefs queue.
+         {"30 5,11,17,23 * * *", Microwaveprop.Workers.GefsFetchWorker},
          # MRMS refreshes the PrecipRate cache that AsosAdjustmentWorker
          # overlays onto the score grid. Both must run together — an
          # ASOS nudge without a fresh MRMS frame reverts to HRRR-only

lib/microwaveprop/weather/gefs_client.ex

@@ -148,8 +148,8 @@ defmodule Microwaveprop.Weather.GefsClient do
       surface_dewpoint_c: sfc_dewpoint_c,
       surface_pressure_mb: sfc_pressure_mb,
       pwat_mm: parsed["PWAT:entire atmosphere (considered as a single layer)"],
-      wind_u_mps: parsed["UGRD:10 m above ground"],
-      wind_v_mps: parsed["VGRD:10 m above ground"],
+      wind_u: parsed["UGRD:10 m above ground"],
+      wind_v: parsed["VGRD:10 m above ground"],
       cloud_cover_pct: parsed["TCDC:entire atmosphere"],
       precip_mm: parsed["APCP:surface"],
       profile: profile

lib/microwaveprop/workers/gefs_fetch_worker.ex

@@ -16,6 +16,7 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do
     max_attempts: 5,
     unique: [period: 3600, states: [:available, :scheduled, :executing, :retryable]]
 
+  alias Microwaveprop.Propagation
   alias Microwaveprop.Weather
   alias Microwaveprop.Weather.GefsClient
   alias Microwaveprop.Weather.SoundingParams
@@ -35,9 +36,8 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do
 
       case GefsClient.fetch_grid_profiles(run_date, run_hour, fh) do
         {:ok, profiles} ->
-          attrs = Enum.map(profiles, &build_profile_attrs(run_time, fh, &1))
-          {count, _} = Weather.upsert_gefs_profiles_batch(attrs)
-          Logger.info("GefsFetch: stored #{count}/#{length(attrs)} profiles for fh=#{fh}")
+          valid_time = run_time |> DateTime.add(fh * 3600, :second) |> DateTime.truncate(:second)
+          score_and_persist(profiles, run_time, fh, valid_time)
           :ok
 
         {:error, :wgrib2_not_available} ->
@@ -52,8 +52,54 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do
     end
   end
 
+  def perform(%Oban.Job{args: args}) when args == %{} do
+    seed_extended_horizon()
+  end
+
   def perform(%Oban.Job{args: _args}), do: {:cancel, :invalid_args}
 
+  @doc """
+  Forecast hours enqueued per GEFS run: f024 through f168 at the
+  6-hour cadence GEFS publishes at. Covers Day 2-7 — the window HRRR
+  can't reach. Starts at f024 so the near-term (f00-f18) stays
+  HRRR-owned and the handoff lines up on the f018 boundary.
+  """
+  @spec extended_horizon_hours() :: [non_neg_integer()]
+  def extended_horizon_hours, do: Enum.to_list(24..168//6)
+
+  @doc """
+  Enqueues one fetch job per extended-horizon forecast hour for the
+  most recent GEFS run that should already be published on NOMADS
+  (run publication lags ~3-4 hours behind the run time; the seeder
+  targets runs 5+ hours in the past for safety).
+  """
+  @spec seed_extended_horizon() :: :ok
+  def seed_extended_horizon do
+    run_time = most_recent_available_run(DateTime.utc_now())
+
+    Logger.info("GefsFetch: seeding extended-horizon chain for run_time=#{run_time}")
+
+    Enum.each(extended_horizon_hours(), fn fh ->
+      %{"run_time" => DateTime.to_iso8601(run_time), "forecast_hour" => fh}
+      |> new()
+      |> Oban.insert()
+    end)
+
+    :ok
+  end
+
+  @doc """
+  Picks the most-recent 00/06/12/18Z GEFS run whose file set should be
+  fully published given an ~5-hour publication lag. Exposed so tests
+  can pin a deterministic "now".
+  """
+  @spec most_recent_available_run(DateTime.t()) :: DateTime.t()
+  def most_recent_available_run(%DateTime{} = now) do
+    now
+    |> DateTime.add(-5 * 3600, :second)
+    |> GefsClient.nearest_run()
+  end
+
   @doc """
   Converts a `GefsClient.build_profile/1` result (already tagged with
   `lat` and `lon`) into a full `gefs_profiles` row attr map. Exposed
@@ -75,8 +121,8 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do
       surface_dewpoint_c: profile.surface_dewpoint_c,
       surface_pressure_mb: profile.surface_pressure_mb,
       pwat_mm: profile[:pwat_mm],
-      wind_u_mps: profile[:wind_u_mps],
-      wind_v_mps: profile[:wind_v_mps],
+      wind_u_mps: profile[:wind_u] || profile[:wind_u_mps],
+      wind_v_mps: profile[:wind_v] || profile[:wind_v_mps],
       cloud_cover_pct: profile[:cloud_cover_pct],
       precip_mm: profile[:precip_mm]
     }
@@ -95,6 +141,68 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do
     end
   end
 
+  # Runs the propagation scorer across the fetched GEFS grid and
+  # persists scores to the `.ntms` file for this `valid_time` so the
+  # extended-horizon outlook picks them up via the usual
+  # `Propagation.scores_at/3` path. Also upserts raw profiles to
+  # `gefs_profiles` for future per-contact extended-horizon lookups.
+  defp score_and_persist(profiles, run_time, fh, valid_time) do
+    grid_data = profiles_to_grid(profiles)
+
+    scores =
+      grid_data
+      |> Task.async_stream(
+        fn {{lat, lon}, profile} ->
+          profile
+          |> Propagation.score_grid_point(valid_time, lat, lon)
+          |> Enum.map(fn r ->
+            %{
+              lat: lat,
+              lon: lon,
+              valid_time: valid_time,
+              band_mhz: r.band_mhz,
+              score: r.score,
+              factors: nil
+            }
+          end)
+        end,
+        max_concurrency: System.schedulers_online() * 2,
+        timeout: 30_000
+      )
+      |> Stream.flat_map(fn
+        {:ok, results} -> results
+        {:exit, _reason} -> []
+      end)
+      |> Enum.to_list()
+
+    case Propagation.replace_scores(scores, valid_time) do
+      {:ok, count} ->
+        Logger.info("GefsFetch: wrote #{count} scores for fh=#{fh} @ #{valid_time}")
+        broadcast_updated(valid_time)
+
+      error ->
+        Logger.error("GefsFetch: replace_scores failed fh=#{fh}: #{inspect(error)}")
+    end
+
+    attrs = Enum.map(profiles, &build_profile_attrs(run_time, fh, &1))
+    {count, _} = Weather.upsert_gefs_profiles_batch(attrs)
+    Logger.info("GefsFetch: upserted #{count}/#{length(attrs)} profile rows for fh=#{fh}")
+  end
+
+  defp profiles_to_grid(profiles) do
+    Map.new(profiles, fn profile ->
+      {{profile.lat, profile.lon}, profile}
+    end)
+  end
+
+  defp broadcast_updated(valid_time) do
+    Phoenix.PubSub.broadcast(
+      Microwaveprop.PubSub,
+      "propagation:updated",
+      {:propagation_updated, [valid_time]}
+    )
+  end
+
   defp validate_run_hour(hour) when hour in @valid_run_hours, do: :ok
   defp validate_run_hour(_), do: {:error, :invalid_run_hour}
 

test/microwaveprop/weather/gefs_client_test.exs

@@ -131,8 +131,8 @@ defmodule Microwaveprop.Weather.GefsClientTest do
     test "carries PWAT, winds, cloud cover, precip through untouched" do
       p = GefsClient.build_profile(@raw)
       assert p.pwat_mm == 25.4
-      assert p.wind_u_mps == 3.0
-      assert p.wind_v_mps == -1.5
+      assert p.wind_u == 3.0
+      assert p.wind_v == -1.5
       assert p.cloud_cover_pct == 62.0
       assert p.precip_mm == 0.0
     end

test/microwaveprop/workers/gefs_fetch_worker_test.exs

@@ -6,9 +6,9 @@ defmodule Microwaveprop.Workers.GefsFetchWorkerTest do
   alias Microwaveprop.Workers.GefsFetchWorker
 
   describe "perform/1" do
-    test "rejects args missing run_time or forecast_hour" do
-      job = %Oban.Job{args: %{}}
-      assert {:cancel, _} = GefsFetchWorker.perform(job)
+    test "cancels on malformed (non-empty, not a seed) args" do
+      args = %{"some_unrelated" => "garbage"}
+      assert {:cancel, _} = GefsFetchWorker.perform(%Oban.Job{args: args})
     end
 
     test "cancels when run_hour is not a valid GEFS cycle" do
@@ -71,6 +71,68 @@ defmodule Microwaveprop.Workers.GefsFetchWorkerTest do
     end
   end
 
+  describe "extended_horizon_hours/0" do
+    test "spans f024 through f168 at 6-hour cadence (25 entries)" do
+      hours = GefsFetchWorker.extended_horizon_hours()
+      assert List.first(hours) == 24
+      assert List.last(hours) == 168
+      assert length(hours) == 25
+      assert Enum.all?(hours, &(rem(&1, 6) == 0))
+    end
+  end
+
+  describe "most_recent_available_run/1" do
+    test "returns the run cycle 5 or more hours earlier" do
+      # 15:00 UTC → 10:00 UTC → snaps to 06Z
+      assert GefsFetchWorker.most_recent_available_run(~U[2026-04-18 15:00:00Z]) ==
+               ~U[2026-04-18 06:00:00Z]
+    end
+
+    test "rolls across midnight when now is near 04Z" do
+      # 04:00 UTC → 23:00 the day before → snaps to 18Z previous day
+      assert GefsFetchWorker.most_recent_available_run(~U[2026-04-18 04:00:00Z]) ==
+               ~U[2026-04-17 18:00:00Z]
+    end
+  end
+
+  describe "scoring interop" do
+    alias Microwaveprop.Propagation
+    alias Microwaveprop.Weather.GefsClient
+
+    test "a GEFS-shaped profile produces non-empty band scores" do
+      raw = %{
+        "TMP:2 m above ground" => 293.15,
+        "RH:2 m above ground" => 50.0,
+        "PRES:surface" => 101_325.0,
+        "PWAT:entire atmosphere (considered as a single layer)" => 20.0,
+        "UGRD:10 m above ground" => 3.0,
+        "VGRD:10 m above ground" => -1.5,
+        "TCDC:entire atmosphere" => 40.0,
+        "APCP:surface" => 0.0,
+        "TMP:1000 mb" => 293.15,
+        "RH:1000 mb" => 50.0,
+        "HGT:1000 mb" => 110.0,
+        "TMP:925 mb" => 288.15,
+        "RH:925 mb" => 40.0,
+        "HGT:925 mb" => 780.0,
+        "TMP:850 mb" => 283.15,
+        "RH:850 mb" => 35.0,
+        "HGT:850 mb" => 1500.0
+      }
+
+      scores = raw |> GefsClient.build_profile() |> Propagation.score_grid_point(~U[2026-04-20 18:00:00Z], 32.9, -97.0)
+
+      assert is_list(scores)
+      assert scores != []
+
+      score = hd(scores)
+      assert is_integer(score.band_mhz)
+      assert is_integer(score.score)
+      assert score.score >= 0
+      assert score.score <= 100
+    end
+  end
+
   describe "storing profiles" do
     test "round-trips a full batch through the context" do
       run_time = ~U[2026-04-18 12:00:00Z]