Hoist commercial link query + tie pipeline chip to ScoresFile

PropagationGridWorker.merge_commercial_link_data/2 called
Commercial.link_degradation_at twice per grid cell (once to count,
once to merge), and each call re-ran enabled_links plus two sample
queries per in-range link. That was ~500k SQL queries per forecast
hour — fine for the DB but catastrophic for log volume in dev
iex sessions trying to watch a chain step run.

Split into two new functions:
  * build_link_lookup/2 does all the DB work up front — one
    enabled_links query + one link_degradation per enabled link
    (~10 queries total).
  * link_degradation_from_lookup/3 is pure: takes a (lat, lon)
    and the precomputed lookup, returns the aggregate or nil.

The worker now calls build_link_lookup once per forecast hour and
the per-cell path is haversine-only. Net: 500k queries → ~10.

PipelineStatus freshness detection moves off oban_jobs.completed_at
onto ScoresFile.latest_valid_time(). The on-disk files ARE the data
the map renders from, so the chip's "Up to date · Nm ago" and the
2h stale threshold now track actual data presence. Running-state
detection still comes from oban_jobs since the chain step is still
an Oban row. Tests updated to seed a ScoresFile instead of a
completed Oban row for the idle/stale cases.

lib/microwaveprop/commercial.ex

@@ -45,43 +45,68 @@ defmodule Microwaveprop.Commercial do
   """
   @spec link_degradation_at({float(), float()}, DateTime.t(), keyword()) :: map() | nil
   def link_degradation_at({lat, lon}, valid_time, opts \\ []) do
-    radius_km = Keyword.get(opts, :radius_km, @default_radius_km)
+    lookup = build_link_lookup(valid_time, opts)
+    link_degradation_from_lookup({lat, lon}, lookup, opts)
+  end
+
+  @doc """
+  Precompute every enabled link's degradation once so a bulk caller
+  (like PropagationGridWorker's per-cell merge) can reuse the same
+  `(link, endpoint, degradation)` list across all 95k grid points
+  instead of re-running `enabled_links/0` and the per-link sample
+  queries per call. Result is fed back into
+  `link_degradation_from_lookup/3`.
+  """
+  @spec build_link_lookup(DateTime.t(), keyword()) :: [
+          {Link.t(), {float(), float()}, map() | nil}
+        ]
+  def build_link_lookup(valid_time, opts \\ []) do
     baseline_days = Keyword.get(opts, :baseline_days, @default_baseline_days)
     current_window = Keyword.get(opts, :current_window_seconds, @default_current_window_seconds)
 
-    candidates =
-      enabled_links()
-      |> Enum.map(fn link -> {link, link_endpoint(link)} end)
-      |> Enum.reject(fn {_link, endpoint} -> is_nil(endpoint) end)
-      |> Enum.filter(fn {_link, {elat, elon}} ->
-        haversine_km(lat, lon, elat, elon) <= radius_km
-      end)
-
     baseline_cutoff = DateTime.add(valid_time, -baseline_days * 24 * 3600, :second)
     current_cutoff = DateTime.add(valid_time, -current_window, :second)
 
+    enabled_links()
+    |> Enum.map(fn link -> {link, link_endpoint(link)} end)
+    |> Enum.reject(fn {_link, endpoint} -> is_nil(endpoint) end)
+    |> Enum.map(fn {link, endpoint} ->
+      degradation = link_degradation(link.id, baseline_cutoff, current_cutoff, valid_time)
+      {link, endpoint, degradation}
+    end)
+  end
+
+  @doc """
+  Pure per-cell aggregator — given a `(lat, lon)` point and a
+  precomputed lookup from `build_link_lookup/2`, returns the
+  aggregate degradation map or `nil` when no in-range link is
+  reporting usable data.
+  """
+  @spec link_degradation_from_lookup({float(), float()}, [tuple()], keyword()) :: map() | nil
+  def link_degradation_from_lookup({lat, lon}, lookup, opts \\ []) do
+    radius_km = Keyword.get(opts, :radius_km, @default_radius_km)
+
     results =
-      candidates
-      |> Enum.map(fn {link, _endpoint} ->
-        link_degradation(link.id, baseline_cutoff, current_cutoff, valid_time)
-      end)
-      |> Enum.reject(&is_nil/1)
+      for {_link, {elat, elon}, %{} = degradation} <- lookup,
+          haversine_km(lat, lon, elat, elon) <= radius_km do
+        degradation
+      end
 
-    case results do
-      [] ->
-        nil
+    aggregate_degradation(results)
+  end
 
-      list ->
-        baseline_avg = average(Enum.map(list, & &1.baseline_dbm))
-        current_avg = average(Enum.map(list, & &1.current_dbm))
+  defp aggregate_degradation([]), do: nil
 
-        %{
-          degradation_db: Float.round(baseline_avg - current_avg, 2),
-          baseline_dbm: Float.round(baseline_avg, 2),
-          current_dbm: Float.round(current_avg, 2),
-          n_links: length(list)
-        }
-    end
+  defp aggregate_degradation(list) do
+    baseline_avg = average(Enum.map(list, & &1.baseline_dbm))
+    current_avg = average(Enum.map(list, & &1.current_dbm))
+
+    %{
+      degradation_db: Float.round(baseline_avg - current_avg, 2),
+      baseline_dbm: Float.round(baseline_avg, 2),
+      current_dbm: Float.round(current_avg, 2),
+      n_links: length(list)
+    }
   end
 
   defp link_degradation(link_id, baseline_cutoff, current_cutoff, valid_time) do

lib/microwaveprop/propagation/pipeline_status.ex

@@ -2,20 +2,21 @@ defmodule Microwaveprop.Propagation.PipelineStatus do
   @moduledoc """
   Aggregated status for the propagation update pipeline.
 
-  The pipeline is driven by two Oban workers:
+  Two inputs drive the chip:
 
-    * `PropagationGridWorker` — hourly, fetches HRRR f00–f18, scores
-      the CONUS grid, and broadcasts `propagation:updated`.
-    * `AsosAdjustmentWorker` — every 10 minutes, nudges scores with
-      recent ASOS surface observations.
-
-  `current/0` returns one struct so the map page can render a single
-  status chip ("Updating…", "Up to date · 8m ago", "Stale · 4h ago")
-  that reflects the whole pipeline rather than tailing a single job.
+    * **Running detection** — queries `oban_jobs` for executing
+      PropagationGridWorker / AsosAdjustmentWorker rows so the chip
+      knows when a chain step is in flight.
+    * **Freshness detection** — reads the newest valid_time from
+      `Microwaveprop.Propagation.ScoresFile` on disk. The on-disk
+      files ARE the propagation data now, so the chip's "Up to date ·
+      Nm ago" and the 2h stale threshold track actual data presence
+      instead of Oban's `completed_at` timestamp.
   """
 
   import Ecto.Query
 
+  alias Microwaveprop.Propagation.ScoresFile
   alias Microwaveprop.Repo
 
   @grid_worker "Microwaveprop.Workers.PropagationGridWorker"
@@ -58,14 +59,14 @@ defmodule Microwaveprop.Propagation.PipelineStatus do
   def current do
     case running_workers() do
       [] ->
-        build_idle_or_stale(latest_completed_at())
+        build_idle_or_stale(latest_data_at())
 
       details ->
         %{
           state: :running,
           label: "Updating propagation",
           details: details,
-          last_update_at: latest_completed_at()
+          last_update_at: latest_data_at()
         }
     end
   end
@@ -91,18 +92,11 @@ defmodule Microwaveprop.Propagation.PipelineStatus do
   defp worker_sort_key(@asos_worker), do: 1
   defp worker_sort_key(_), do: 99
 
-  defp latest_completed_at do
-    case Repo.one(
-           from j in "oban_jobs",
-             where: j.state == "completed" and j.worker in ^@workers,
-             order_by: [desc: j.completed_at],
-             limit: 1,
-             select: j.completed_at
-         ) do
-      nil -> nil
-      %NaiveDateTime{} = naive -> DateTime.from_naive!(naive, "Etc/UTC")
-      %DateTime{} = dt -> dt
-    end
+  # Freshness is derived from the newest ScoresFile on disk. That file
+  # is the thing the map actually renders from, so its valid_time is
+  # the most honest answer to "when was the data last updated?".
+  defp latest_data_at do
+    ScoresFile.latest_valid_time()
   end
 
   defp build_idle_or_stale(nil) do

lib/microwaveprop/workers/propagation_grid_worker.ex

@@ -254,25 +254,27 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
   end
 
   defp merge_commercial_link_data(grid_data, valid_time) do
-    # Compute link degradation once per distinct link cluster and cache by
-    # point. Commercial links only cluster around DFW so most grid points see
-    # nil — cheap no-op path dominates.
-    boosted =
-      Enum.count(grid_data, fn {{lat, lon}, _profile} ->
-        degradation = Commercial.link_degradation_at({lat, lon}, valid_time)
-        not is_nil(degradation)
+    # Precompute per-link degradation once (≤10 SQL queries total).
+    # Commercial links cluster around DFW so most grid cells see nil —
+    # the per-cell path is now a pure haversine check, not a DB query.
+    lookup = Commercial.build_link_lookup(valid_time)
+
+    {merged, boosted} =
+      Enum.reduce(grid_data, {%{}, 0}, fn {{lat, lon} = point, profile}, {acc, count} ->
+        case Commercial.link_degradation_from_lookup({lat, lon}, lookup) do
+          nil ->
+            {Map.put(acc, point, profile), count}
+
+          degradation ->
+            {Map.put(acc, point, Map.put(profile, :commercial_link_degradation, degradation)), count + 1}
+        end
       end)
 
     if boosted > 0 do
       Logger.info("PropagationGrid: commercial-link degradation available for #{boosted} grid cells")
     end
 
-    Map.new(grid_data, fn {{lat, lon} = point, profile} ->
-      case Commercial.link_degradation_at({lat, lon}, valid_time) do
-        nil -> {point, profile}
-        degradation -> {point, Map.put(profile, :commercial_link_degradation, degradation)}
-      end
-    end)
+    merged
   end
 
   defp merge_nexrad_data(grid_data, valid_time) do

test/microwaveprop/propagation/pipeline_status_test.exs

@@ -2,6 +2,7 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do
   use Microwaveprop.DataCase, async: false
 
   alias Microwaveprop.Propagation.PipelineStatus
+  alias Microwaveprop.Propagation.ScoresFile
   alias Microwaveprop.Repo
 
   @grid_worker "Microwaveprop.Workers.PropagationGridWorker"
@@ -99,15 +100,12 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do
       assert asos.label =~ "ASOS"
     end
 
-    test "returns :idle with Up to date label when last grid run completed recently" do
-      completed = minutes_ago(15)
-
-      insert_oban_job(%{
-        state: "completed",
-        worker: @grid_worker,
-        attempted_at: completed,
-        completed_at: completed
-      })
+    test "returns :idle with Up to date label when a recent ScoresFile exists" do
+      ScoresFile.write!(
+        10_000,
+        minutes_ago(15),
+        [%{lat: 25.0, lon: -125.0, score: 50}]
+      )
 
       status = PipelineStatus.current()
 
@@ -116,15 +114,12 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do
       assert status.label =~ "Up to date"
     end
 
-    test "returns :stale when last completed was more than 120 minutes ago" do
-      stale = minutes_ago(180)
-
-      insert_oban_job(%{
-        state: "completed",
-        worker: @grid_worker,
-        attempted_at: stale,
-        completed_at: stale
-      })
+    test "returns :stale when the newest ScoresFile is more than 120 minutes old" do
+      ScoresFile.write!(
+        10_000,
+        minutes_ago(180),
+        [%{lat: 25.0, lon: -125.0, score: 50}]
+      )
 
       status = PipelineStatus.current()
 
@@ -132,15 +127,12 @@ defmodule Microwaveprop.Propagation.PipelineStatusTest do
       assert status.label =~ "stale"
     end
 
-    test "prefers :running over :idle when a job is executing even if a recent job completed" do
-      recent = minutes_ago(5)
-
-      insert_oban_job(%{
-        state: "completed",
-        worker: @grid_worker,
-        attempted_at: recent,
-        completed_at: recent
-      })
+    test "prefers :running over :idle when a job is executing even with a recent ScoresFile" do
+      ScoresFile.write!(
+        10_000,
+        minutes_ago(5),
+        [%{lat: 25.0, lon: -125.0, score: 50}]
+      )
 
       insert_oban_job(%{
         state: "executing",