Graham McIntire
434fca2ad1
PropagationGridWorker now broadcasts a {:propagation_pipeline_progress,
%{forecast_hour:, valid_time:}} message on the propagation:pipeline
PubSub topic at the start of each process_forecast_hour/4 call.
MapLive subscribes on mount, stashes the last message in the new
:pipeline_progress assign, and clears it on the refresh tick whenever
PipelineStatus flips out of :running.
The chip component consults a new PipelineStatus.running_label/1
helper that formats the payload as "Updating propagation · now" for
f00 and "Updating propagation · +Nh" for f01-f18, falling back to
the base running label until the first progress message arrives.
Covered by unit tests on the formatter and an integration test in
MapLiveTest that seeds an executing grid job, broadcasts progress,
and asserts the rendered chip.
127 lines
3.7 KiB
Elixir
127 lines
3.7 KiB
Elixir
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
|
||
}
|
||
|
||
@doc """
|
||
Format a forecast-progress payload (as broadcast by
|
||
`PropagationGridWorker`) into a short chip label for the map.
|
||
|
||
Returns `nil` when the payload has no `forecast_hour` so callers can
|
||
fall back to the base running label from `current/0`.
|
||
"""
|
||
@spec running_label(map() | nil) :: String.t() | nil
|
||
def running_label(%{forecast_hour: 0}), do: "Updating propagation · now"
|
||
|
||
def running_label(%{forecast_hour: fh}) when is_integer(fh) and fh > 0, do: "Updating propagation · +#{fh}h"
|
||
|
||
def running_label(_), do: nil
|
||
|
||
@spec current() :: t()
|
||
def current do
|
||
case running_worker() do
|
||
worker when is_binary(worker) ->
|
||
%{
|
||
state: :running,
|
||
label: base_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 base_running_label(@grid_worker), do: "Updating propagation (HRRR run)"
|
||
defp base_running_label(@asos_worker), do: "Updating propagation (ASOS nudge)"
|
||
defp base_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
|