prop/lib/microwaveprop/workers/asos_adjustment_worker.ex

defmodule Microwaveprop.Workers.AsosAdjustmentWorker do
  @moduledoc """
  Nudge the propagation score grid with fresh ASOS observations between
  hourly HRRR runs.

  On each cron tick (every 10 minutes) this worker:

  1. Finds the most recent `valid_time` that actually has HRRR profiles
     persisted.
  2. Pulls current ASOS observations from all state networks via
     `Microwaveprop.Weather.IemClient.fetch_current_asos/1`, keeps only
     those within 30 minutes of now.
  3. Loads every `hrrr_profiles` row on the 0.125° propagation grid for
     that valid_time into memory as plain maps.
  4. Hands both lists to `Microwaveprop.Propagation.AsosNudge.compute/3`
     — pure function, IDW bias field, upper-air fields preserved.
  5. Upserts the nudged scores onto `(lat, lon, valid_time, band_mhz)`,
     overwriting the HRRR-only values for grid cells within 250 km of a
     reporting station. Cells outside that radius are left alone.
  6. Warms the `ScoreCache` and broadcasts `propagation:updated` so every
     connected `/map` client refreshes.

  This worker owns the I/O. `AsosNudge` stays pure so its tests don't need
  Repo or network access.
  """

  use Oban.Worker, queue: :propagation, max_attempts: 3

  import Ecto.Query, only: [from: 2]

  alias Microwaveprop.Propagation
  alias Microwaveprop.Propagation.AsosNudge
  alias Microwaveprop.Propagation.BandConfig
  alias Microwaveprop.Repo
  alias Microwaveprop.Weather.HrrrProfile
  alias Microwaveprop.Weather.IemClient
  alias Microwaveprop.Weather.MrmsCache

  require Logger

  @state_networks ~w(
    AL_ASOS AK_ASOS AZ_ASOS AR_ASOS CA_ASOS CO_ASOS CT_ASOS DE_ASOS
    FL_ASOS GA_ASOS HI_ASOS ID_ASOS IL_ASOS IN_ASOS IA_ASOS KS_ASOS
    KY_ASOS LA_ASOS ME_ASOS MD_ASOS MA_ASOS MI_ASOS MN_ASOS MS_ASOS
    MO_ASOS MT_ASOS NE_ASOS NV_ASOS NH_ASOS NJ_ASOS NM_ASOS NY_ASOS
    NC_ASOS ND_ASOS OH_ASOS OK_ASOS OR_ASOS PA_ASOS RI_ASOS SC_ASOS
    SD_ASOS TN_ASOS TX_ASOS UT_ASOS VT_ASOS VA_ASOS WA_ASOS WV_ASOS
    WI_ASOS WY_ASOS DC_ASOS
  )

  @max_obs_age_seconds 1800

  @impl Oban.Worker
  def perform(%Oban.Job{}) do
    with {:ok, valid_time} <- latest_hrrr_valid_time(),
         [_ | _] = profiles <- load_hrrr_profiles(valid_time) do
      fresh = fetch_fresh_observations()
      rain_grid = mrms_rain_grid()

      if fresh == [] and map_size(rain_grid) == 0 do
        Logger.info("AsosAdjustment: no fresh ASOS obs and no MRMS rain cached, skipping")
        :ok
      else
        apply_nudge(fresh, profiles, valid_time, rain_grid)
      end
    else
      :no_hrrr ->
        Logger.info("AsosAdjustment: no HRRR profiles yet, skipping")
        :ok

      [] ->
        Logger.info("AsosAdjustment: HRRR valid_time has no grid profiles, skipping")
        :ok
    end
  end

  defp fetch_fresh_observations do
    # IemClient.fetch_current_asos/1 swallows its own per-network HTTP
    # errors and always returns `{:ok, [...]}` (possibly empty). We just
    # filter the list down to observations inside the freshness window.
    {:ok, observations} = IemClient.fetch_current_asos(@state_networks)
    filter_fresh(observations)
  end

  defp mrms_rain_grid do
    case MrmsCache.fetch() do
      {:ok, _valid_time, grid} -> grid
      :miss -> %{}
    end
  end

  defp apply_nudge(observations, profiles, valid_time, rain_grid) do
    scores = AsosNudge.compute(observations, valid_time, profiles, rain_grid)

    if scores == [] do
      Logger.info(
        "AsosAdjustment: #{length(observations)} fresh obs + #{map_size(rain_grid)} MRMS cells but no grid cells eligible"
      )

      :ok
    else
      case Propagation.upsert_scores(scores, prune: false) do
        {:ok, count} ->
          Logger.info(
            "AsosAdjustment: nudged #{count} scores from #{length(observations)} obs + #{map_size(rain_grid)} MRMS cells at #{valid_time}"
          )

          warm_and_broadcast(valid_time)
          :ok

        error ->
          Logger.error("AsosAdjustment: upsert failed — #{inspect(error)}")
          error
      end
    end
  end

  defp warm_and_broadcast(valid_time) do
    Enum.each(BandConfig.all_bands(), fn band ->
      Propagation.warm_cache_and_broadcast(band.freq_mhz, valid_time)
    end)

    Phoenix.PubSub.broadcast(
      Microwaveprop.PubSub,
      "propagation:updated",
      {:propagation_updated, valid_time}
    )
  end

  defp latest_hrrr_valid_time do
    case Repo.one(
           from(h in HrrrProfile,
             where: h.is_grid_point == true,
             select: max(h.valid_time)
           )
         ) do
      nil -> :no_hrrr
      vt -> {:ok, vt}
    end
  end

  defp filter_fresh(observations) do
    cutoff = DateTime.add(DateTime.utc_now(), -@max_obs_age_seconds, :second)

    Enum.filter(observations, fn obs ->
      obs.utc_valid && DateTime.after?(obs.utc_valid, cutoff)
    end)
  end

  # Deliberately DOES NOT select h.profile. At 92k grid points per tick that
  # column is ~120 MB of JSONB and Postgrex's per-row Jason.decode! blew past
  # the 15s pool checkout window. score_grid_point/4 already prefers the
  # persisted min_refractivity_gradient scalar over re-deriving from the
  # profile array, so leaving the array behind is functionally transparent.
  defp load_hrrr_profiles(valid_time) do
    Repo.all(
      from(h in HrrrProfile,
        where: h.valid_time == ^valid_time and h.is_grid_point == true,
        select: %{
          lat: h.lat,
          lon: h.lon,
          valid_time: h.valid_time,
          surface_temp_c: h.surface_temp_c,
          surface_dewpoint_c: h.surface_dewpoint_c,
          surface_pressure_mb: h.surface_pressure_mb,
          surface_refractivity: h.surface_refractivity,
          min_refractivity_gradient: h.min_refractivity_gradient,
          hpbl_m: h.hpbl_m,
          pwat_mm: h.pwat_mm,
          ducting_detected: h.ducting_detected,
          duct_characteristics: h.duct_characteristics
        }
      )
    )
  end
end