Show aggregated propagation pipeline status on the map

Adds a small chip below the NTMS title in both the mobile and
desktop sidebars of /map that collapses PropagationGridWorker +
AsosAdjustmentWorker Oban state into one of four states —
running / idle / stale / unknown — with a plain-language label
("Updating propagation (HRRR run)…", "Up to date · 12m ago").
The chip refreshes on a 15s timer and whenever the pipeline
broadcasts propagation:updated, so a long HRRR sweep is visible
from the moment a visitor opens the page.

lib/microwaveprop/propagation/pipeline_status.ex

@@ -0,0 +1,113 @@
+defmodule Microwaveprop.Propagation.PipelineStatus do
+  @moduledoc """
+  Aggregated status for the propagation update pipeline.
+
+  The pipeline is driven by two Oban workers:
+
+    * `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.
+  """
+
+  import Ecto.Query
+
+  alias Microwaveprop.Repo
+
+  @grid_worker "Microwaveprop.Workers.PropagationGridWorker"
+  @asos_worker "Microwaveprop.Workers.AsosAdjustmentWorker"
+
+  @workers [@grid_worker, @asos_worker]
+
+  # Scores older than this flip the chip to :stale. Matches the
+  # FreshnessMonitor threshold — anything past 2h is already a gap
+  # the pipeline should have caught.
+  @stale_after_minutes 120
+
+  @type state :: :running | :idle | :stale | :unknown
+
+  @type t :: %{
+          state: state(),
+          label: String.t(),
+          last_update_at: DateTime.t() | nil
+        }
+
+  @spec current() :: t()
+  def current do
+    case running_worker() do
+      worker when is_binary(worker) ->
+        %{
+          state: :running,
+          label: running_label(worker),
+          last_update_at: latest_completed_at()
+        }
+
+      nil ->
+        build_idle_or_stale(latest_completed_at())
+    end
+  end
+
+  defp running_worker do
+    Repo.one(
+      from j in "oban_jobs",
+        where: j.state == "executing" and j.worker in ^@workers,
+        order_by: [desc: j.attempted_at],
+        limit: 1,
+        select: j.worker
+    )
+  end
+
+  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
+  end
+
+  defp build_idle_or_stale(nil) do
+    %{state: :unknown, label: "Propagation status unknown", last_update_at: nil}
+  end
+
+  defp build_idle_or_stale(%DateTime{} = last) do
+    age_minutes = max(DateTime.diff(DateTime.utc_now(), last, :minute), 0)
+
+    if age_minutes > @stale_after_minutes do
+      %{
+        state: :stale,
+        label: "Propagation data stale · last update #{format_age(age_minutes)} ago",
+        last_update_at: last
+      }
+    else
+      %{
+        state: :idle,
+        label: "Up to date · #{format_age(age_minutes)} ago",
+        last_update_at: last
+      }
+    end
+  end
+
+  defp running_label(@grid_worker), do: "Updating propagation (HRRR run)…"
+  defp running_label(@asos_worker), do: "Updating propagation (ASOS nudge)…"
+  defp running_label(_), do: "Updating propagation…"
+
+  defp format_age(0), do: "just now"
+  defp format_age(1), do: "1m"
+  defp format_age(n) when n < 60, do: "#{n}m"
+
+  defp format_age(n) do
+    hours = div(n, 60)
+    mins = rem(n, 60)
+    if mins == 0, do: "#{hours}h", else: "#{hours}h #{mins}m"
+  end
+end

lib/microwaveprop_web/live/map_live.ex

@@ -5,6 +5,7 @@ defmodule MicrowavepropWeb.MapLive do
 
   alias Microwaveprop.Propagation
   alias Microwaveprop.Propagation.BandConfig
+  alias Microwaveprop.Propagation.PipelineStatus
   alias Microwaveprop.Terrain.Viewshed
 
   require Logger
@@ -14,13 +15,22 @@ defmodule MicrowavepropWeb.MapLive do
   @default_center %{lat: 32.897, lon: -97.038}
   @default_zoom 7
 
+  # How often to re-query oban_jobs for pipeline state while the map is
+  # open. Short enough that the "Updating…" chip shows up within a page-
+  # visit for an in-progress hourly run, long enough to be cheap.
+  @pipeline_status_refresh_ms 15_000
+
   @impl true
   def mount(_params, session, socket) do
     if connected?(socket) do
       Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "propagation:updated")
+      Process.send_after(self(), :refresh_pipeline_status, @pipeline_status_refresh_ms)
     end
 
-    socket = assign(socket, :initial_utc_clock, Calendar.strftime(DateTime.utc_now(), "%H:%M UTC"))
+    socket =
+      socket
+      |> assign(:initial_utc_clock, Calendar.strftime(DateTime.utc_now(), "%H:%M UTC"))
+      |> assign(:pipeline_status, PipelineStatus.current())
 
     # LiveStash only persists the keys passed to `stash_assigns/2`
     # (selected_band + selected_time). On reconnect the recovered socket has
@@ -264,12 +274,18 @@ defmodule MicrowavepropWeb.MapLive do
     socket =
       socket
       |> assign(valid_times: valid_times, selected_time: selected_time)
+      |> assign(:pipeline_status, PipelineStatus.current())
       |> push_event("update_scores", %{scores: scores})
       |> push_timeline()
 
     {:noreply, socket}
   end
 
+  def handle_info(:refresh_pipeline_status, socket) do
+    Process.send_after(self(), :refresh_pipeline_status, @pipeline_status_refresh_ms)
+    {:noreply, assign(socket, :pipeline_status, PipelineStatus.current())}
+  end
+
   @impl true
   def handle_async(:viewshed, {:ok, result}, socket) do
     points = Enum.map(result.boundary, fn p -> %{lat: p.lat, lon: p.lon} end)
@@ -344,6 +360,38 @@ defmodule MicrowavepropWeb.MapLive do
     end
   end
 
+  attr :id, :string, required: true
+  attr :status, :map, required: true
+
+  defp pipeline_status_chip(assigns) do
+    ~H"""
+    <div
+      id={@id}
+      data-pipeline-state={@status.state}
+      class="flex items-center gap-1.5 text-xs px-1 py-1"
+      title={@status.label}
+    >
+      <%= case @status.state do %>
+        <% :running -> %>
+          <.icon
+            name="hero-arrow-path"
+            class="size-3.5 shrink-0 text-info motion-safe:animate-spin"
+          />
+          <span class="opacity-90 truncate">{@status.label}</span>
+        <% :idle -> %>
+          <span class="size-2 rounded-full bg-success inline-block shrink-0"></span>
+          <span class="opacity-70 truncate">{@status.label}</span>
+        <% :stale -> %>
+          <span class="size-2 rounded-full bg-warning inline-block shrink-0"></span>
+          <span class="opacity-80 truncate">{@status.label}</span>
+        <% :unknown -> %>
+          <span class="size-2 rounded-full bg-base-content/30 inline-block shrink-0"></span>
+          <span class="opacity-60 truncate">{@status.label}</span>
+      <% end %>
+    </div>
+    """
+  end
+
   @impl true
   def render(assigns) do
     ~H"""
@@ -405,6 +453,8 @@ defmodule MicrowavepropWeb.MapLive do
               </div>
             </div>
 
+            <.pipeline_status_chip id="pipeline-status-mobile" status={@pipeline_status} />
+
             <div class="dropdown">
               <div tabindex="0" role="button" class="btn btn-sm w-full justify-between">
                 {selected_label(@bands, @selected_band)}
@@ -558,6 +608,8 @@ defmodule MicrowavepropWeb.MapLive do
             </div>
           </div>
 
+          <.pipeline_status_chip id="pipeline-status-desktop" status={@pipeline_status} />
+
           <%!-- Band selector --%>
           <div class="dropdown">
             <div tabindex="0" role="button" class="btn btn-sm w-full justify-between">

test/microwaveprop/propagation/pipeline_status_test.exs

@@ -0,0 +1,143 @@
+defmodule Microwaveprop.Propagation.PipelineStatusTest do
+  use Microwaveprop.DataCase, async: false
+
+  alias Microwaveprop.Propagation.PipelineStatus
+  alias Microwaveprop.Repo
+
+  @grid_worker "Microwaveprop.Workers.PropagationGridWorker"
+  @asos_worker "Microwaveprop.Workers.AsosAdjustmentWorker"
+
+  # Bypass Oban's normal insert pipeline (which executes inline in test)
+  # by writing directly to the Oban.Job schema. The SQL sandbox keeps
+  # Oban's executor from seeing these rows.
+  defp insert_oban_job(overrides) do
+    now = DateTime.utc_now()
+
+    defaults = %{
+      state: "available",
+      queue: "propagation",
+      worker: @grid_worker,
+      args: %{},
+      attempt: 0,
+      max_attempts: 3,
+      inserted_at: now,
+      scheduled_at: now
+    }
+
+    attrs = Map.merge(defaults, Map.new(overrides))
+    Repo.insert!(struct!(Oban.Job, attrs))
+  end
+
+  defp minutes_ago(n) do
+    DateTime.add(DateTime.utc_now(), -n * 60, :second)
+  end
+
+  describe "current/0" do
+    test "returns :unknown when no pipeline jobs have ever run" do
+      status = PipelineStatus.current()
+
+      assert status.state == :unknown
+      assert status.last_update_at == nil
+      assert is_binary(status.label)
+    end
+
+    test "returns :running with grid worker label when PropagationGridWorker is executing" do
+      insert_oban_job(%{
+        state: "executing",
+        worker: @grid_worker,
+        attempted_at: minutes_ago(5)
+      })
+
+      status = PipelineStatus.current()
+
+      assert status.state == :running
+      assert status.label =~ "HRRR"
+    end
+
+    test "returns :running with ASOS label when AsosAdjustmentWorker is executing" do
+      insert_oban_job(%{
+        state: "executing",
+        worker: @asos_worker,
+        attempted_at: minutes_ago(1)
+      })
+
+      status = PipelineStatus.current()
+
+      assert status.state == :running
+      assert status.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
+      })
+
+      status = PipelineStatus.current()
+
+      assert status.state == :idle
+      assert status.last_update_at
+      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
+      })
+
+      status = PipelineStatus.current()
+
+      assert status.state == :stale
+      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
+      })
+
+      insert_oban_job(%{
+        state: "executing",
+        worker: @grid_worker,
+        attempted_at: minutes_ago(1)
+      })
+
+      status = PipelineStatus.current()
+
+      assert status.state == :running
+    end
+
+    test "ignores completed jobs from unrelated workers" do
+      # WeatherFetchWorker enriches individual QSOs — it's not part of
+      # the CONUS grid update pipeline, so its recent success should
+      # not make the map page show 'Up to date'.
+      completed = minutes_ago(5)
+
+      insert_oban_job(%{
+        state: "completed",
+        worker: "Microwaveprop.Workers.WeatherFetchWorker",
+        queue: "weather",
+        attempted_at: completed,
+        completed_at: completed
+      })
+
+      status = PipelineStatus.current()
+
+      assert status.state == :unknown
+    end
+  end
+end

test/microwaveprop_web/live/map_live_test.exs

@@ -97,4 +97,15 @@ defmodule MicrowavepropWeb.MapLiveTest do
       refute html =~ "checked"
     end
   end
+
+  describe "pipeline status chip" do
+    test "renders the aggregated pipeline status chip on mount", %{conn: conn} do
+      {:ok, _lv, html} = live(conn, ~p"/map")
+      # Both the mobile and desktop sidebars carry their own chip.
+      assert html =~ ~s(id="pipeline-status-mobile")
+      assert html =~ ~s(id="pipeline-status-desktop")
+      # With a clean test DB no pipeline workers have ever run.
+      assert html =~ ~s(data-pipeline-state="unknown")
+    end
+  end
 end