prop/lib/microwaveprop/propagation/pipeline_status.ex

defmodule Microwaveprop.Propagation.PipelineStatus do
  @moduledoc """
  Aggregated status for the propagation update pipeline.

  Two inputs drive the chip:

    * **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"
  @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 worker_tag :: :grid | :asos

  @type running_detail :: %{worker: worker_tag(), label: String.t()}

  @type t :: %{
          state: state(),
          label: String.t(),
          details: [running_detail()],
          last_update_at: DateTime.t() | nil
        }

  @doc """
  Format a forecast-progress payload (as broadcast by
  `PropagationGridWorker`) into a short detail line for the running
  chip ("through now", "through +3h", …).

  The label expresses how far the pipeline has advanced relative to
  *now*, not relative to the HRRR run time: the map can scrub the
  timeline through this instant. This is the semantic the user expects
  when watching the chip advance — HRRR lagging the wall clock by
  ~2 hours means f00's valid_time is actually "2h ago", and "+Nh" in
  worker-frame coordinates is Nh-2 in user-frame coordinates.

  Returns `nil` when the payload has no `valid_time` so callers can
  fall back to the base detail (e.g. "HRRR run") from `current/0`.
  """
  @spec running_detail(map() | nil) :: String.t() | nil
  def running_detail(%{valid_time: %DateTime{} = valid_time}) do
    diff_minutes = DateTime.diff(valid_time, DateTime.utc_now(), :minute)
    "through " <> format_offset(diff_minutes)
  end

  def running_detail(_), do: nil

  defp format_offset(diff) when diff >= -30 and diff <= 30, do: "now"

  defp format_offset(diff) when diff > 30 do
    hours = div(diff + 30, 60)
    "+#{hours}h"
  end

  defp format_offset(diff) do
    hours = div(-diff + 30, 60)
    "#{hours}h ago"
  end

  @spec current() :: t()
  def current do
    case running_workers() do
      [] ->
        build_idle_or_stale(latest_data_at())

      details ->
        %{
          state: :running,
          label: "Updating propagation",
          details: details,
          last_update_at: latest_data_at()
        }
    end
  end

  # Returns every pipeline worker currently in executing state, in a
  # stable grid-before-asos order so the chip sub-lines don't flicker
  # as jobs come and go.
  defp running_workers do
    from(j in "oban_jobs",
      where: j.state == "executing" and j.worker in ^@workers,
      distinct: j.worker,
      select: j.worker
    )
    |> Repo.all()
    |> Enum.sort_by(&worker_sort_key/1)
    |> Enum.map(&to_running_detail/1)
  end

  defp to_running_detail(@grid_worker), do: %{worker: :grid, label: "HRRR run"}
  defp to_running_detail(@asos_worker), do: %{worker: :asos, label: "ASOS nudge"}

  defp worker_sort_key(@grid_worker), do: 0
  defp worker_sort_key(@asos_worker), do: 1
  defp worker_sort_key(_), do: 99

  # 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
    %{state: :unknown, label: "Propagation status unknown", details: [], 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_with_ago(age_minutes)}",
        details: [],
        last_update_at: last
      }
    else
      %{
        state: :idle,
        label: "Up to date · #{format_age_with_ago(age_minutes)}",
        details: [],
        last_update_at: last
      }
    end
  end

  # "just now" reads fine on its own; anything older needs "ago" for
  # the duration to make sense.
  defp format_age_with_ago(0), do: "just now"
  defp format_age_with_ago(n), do: "#{format_age(n)} ago"

  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