prop/lib/microwaveprop/workers/propagation_grid_worker.ex

defmodule Microwaveprop.Workers.PropagationGridWorker do
  @moduledoc """
  Oban worker that downloads HRRR data and computes propagation scores
  across the CONUS grid for all bands, one forecast hour at a time.

  Each `perform/1` processes a single forecast hour (~8–10 min of wall
  time) and enqueues the next hour in the chain. A cron fire with
  empty args seeds the chain at f00; subsequent runs carry
  `forecast_hour` + `run_time` args. At f18 the chain stops and the
  pruner cleans up old scores.

  Splitting by forecast hour is a resilience play: a full sweep takes
  ~3 hours of wall time, longer than the typical pod-restart interval
  on this deployment. Under the old "one big perform" design, any
  deploy mid-sweep killed the whole run, and max_attempts would
  exhaust without recording an error. Per-hour jobs survive deploys
  because Lifeline only needs to rescue a single 10-minute step, and
  retries re-fetch just that forecast hour.
  """

  use Oban.Worker,
    queue: :propagation,
    # Higher than the default 3 so a few DynamicLifeline rescues
    # (e.g., a rolling deploy that kills a mid-flight chain step)
    # don't exhaust the chain's retry budget and discard the whole
    # run. Legitimate scoring errors still give up after 5 attempts.
    max_attempts: 5,
    # Deduplicate identical jobs across a 1-hour window. Uniqueness
    # is over the full args set, so the seed (`%{}`) collapses with
    # itself — FreshnessMonitor's 5-minute ticks during a long outage
    # no longer stack 24 redundant f00-f18 chains — while chain steps
    # with distinct `run_time` + `forecast_hour` args remain distinct
    # from each other and from the seed. The period is the same as
    # the hourly cron interval so successive hourly cron fires move
    # past the window naturally.
    unique: [period: 3600, states: [:available, :scheduled, :executing, :retryable]]

  alias Microwaveprop.Commercial
  alias Microwaveprop.Propagation
  alias Microwaveprop.Propagation.BandConfig
  alias Microwaveprop.Propagation.Grid
  alias Microwaveprop.Propagation.ProfilesFile
  alias Microwaveprop.Propagation.ScoreCache
  alias Microwaveprop.Propagation.ScoresFile
  alias Microwaveprop.Weather
  alias Microwaveprop.Weather.GridCache
  alias Microwaveprop.Weather.HrrrClient
  alias Microwaveprop.Weather.HrrrNativeClient
  alias Microwaveprop.Weather.NexradClient

  require Logger

  # Hard ceiling for one forecast hour. A healthy step is ~8-10 min.
  # 20 min gives 2× headroom for a slow HRRR fetch or scoring batch.
  # Oban kills the executing process on timeout, which closes linked
  # ports and cascades SIGKILL to any child wgrib2 subprocess.
  @run_timeout_ms 20 * 60 * 1000

  @impl Oban.Worker
  def timeout(_job), do: @run_timeout_ms

  @max_forecast_hour 18

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"forecast_hour" => fh, "run_time" => run_time_iso}}) do
    {:ok, run_time, _} = DateTime.from_iso8601(run_time_iso)
    run_chain_step(run_time, fh)
  end

  def perform(%Oban.Job{args: args}) when args == %{} do
    seed_chain()
  end

  @doc """
  Enqueue the next chain step after a successful forecast-hour run.

  Returns `{:ok, job}` when a new step is enqueued, or `:final` when
  `fh` is already `@max_forecast_hour` so the chain has no more work.
  Public so the chain entry point and tests can both exercise the
  same enqueue path.
  """
  @spec enqueue_next_step(DateTime.t(), non_neg_integer()) :: {:ok, Oban.Job.t()} | :final
  def enqueue_next_step(_run_time, fh) when fh >= @max_forecast_hour, do: :final

  def enqueue_next_step(%DateTime{} = run_time, fh) when fh >= 0 do
    {:ok, _job} =
      %{"run_time" => DateTime.to_iso8601(run_time), "forecast_hour" => fh + 1}
      |> new()
      |> Oban.insert()
  end

  defp seed_chain do
    # HRRR takes ~45min to publish after the hour. Use 2 hours ago to
    # ensure availability.
    two_hours_ago = DateTime.add(DateTime.utc_now(), -2, :hour)
    run_time = two_hours_ago |> HrrrClient.nearest_hrrr_hour() |> DateTime.truncate(:second)

    Logger.info("PropagationGrid: seeding chain run_time=#{run_time}, f00-f#{@max_forecast_hour}")

    {:ok, _job} =
      %{"run_time" => DateTime.to_iso8601(run_time), "forecast_hour" => 0}
      |> new()
      |> Oban.insert()

    :ok
  end

  defp run_chain_step(run_time, fh) do
    t_start = System.monotonic_time(:millisecond)
    points = Grid.conus_points()
    valid_time = DateTime.add(run_time, fh * 3600, :second)

    Logger.info("PropagationGrid: chain step run_time=#{run_time} fh=#{fh} (#{length(points)} points)")

    result = process_forecast_hour(points, run_time, fh, valid_time)
    :erlang.garbage_collect()

    case result do
      :ok ->
        Phoenix.PubSub.broadcast(
          Microwaveprop.PubSub,
          "propagation:updated",
          {:propagation_updated, [valid_time]}
        )

        total_ms = System.monotonic_time(:millisecond) - t_start
        Logger.info("PropagationGrid: fh=#{fh} step finished in #{format_duration(total_ms)}")

        run_time
        |> enqueue_next_step(fh)
        |> handle_step_transition(run_time)

        :ok

      other ->
        other
    end
  end

  # `enqueue_next_step/2` returns `:final` at fh=18 (the whole chain
  # is done) or `{:ok, job}` when the next hour has been scheduled.
  # Split into its own function to keep `run_chain_step/2` under
  # credo's nesting limit.
  defp handle_step_transition(:final, run_time) do
    Weather.purge_grid_point_profiles()
    Propagation.prune_old_scores()
    # Drop any files left over from the previous chain that the new
    # chain didn't happen to overwrite (files are keyed by
    # valid_time, so an "old f00" from a prior run escapes the
    # natural last-writer-wins path).
    dropped_scores = ScoresFile.retain_window(run_time, @max_forecast_hour)
    dropped_profiles = ProfilesFile.retain_window(run_time, @max_forecast_hour)

    if dropped_scores + dropped_profiles > 0 do
      Logger.info(
        "PropagationGrid: discarded #{dropped_scores} leftover score files + " <>
          "#{dropped_profiles} profile files from prior chain"
      )
    end

    Logger.info("PropagationGrid: chain complete for run_time=#{run_time}")
  end

  defp handle_step_transition({:ok, _job}, _run_time), do: :ok

  defp process_forecast_hour(points, run_time, forecast_hour, valid_time) do
    label = "f#{String.pad_leading(Integer.to_string(forecast_hour), 2, "0")}"

    case timed(label, fn ->
           HrrrClient.fetch_grid(points, run_time, forecast_hour: forecast_hour)
         end) do
      {:ok, grid_data} ->
        # HRRR profiles used to be persisted to the hrrr_profiles table
        # here for AsosAdjustmentWorker to re-score from. That was ~12
        # minutes of JSONB inserts per chain (92k rows × 19 forecast
        # hours) with no user-visible benefit — the scores live in the
        # /data/scores files now, and AsosAdjustmentWorker is disabled.
        # Per-contact HRRR enrichment still uses HrrrFetchWorker, which
        # writes its own `is_grid_point: false` rows.

        # Fetch native duct metrics and merge into grid_data for scoring
        grid_data = merge_native_duct_data(grid_data, run_time, forecast_hour)
        :erlang.garbage_collect()

        # NEXRAD current-hour composite reflectivity catches fast-moving
        # convective cells between HRRR hourly analyses. Only useful for
        # f00 — forecast hours can't see the future radar image.
        grid_data =
          if forecast_hour == 0 do
            merge_nexrad_data(grid_data, valid_time)
          else
            grid_data
          end

        # Commercial-link inverse sensor — only meaningful for f00 because
        # the measurement is of the current atmospheric state, not a forecast.
        grid_data =
          if forecast_hour == 0 do
            merge_commercial_link_data(grid_data, valid_time)
          else
            grid_data
          end

        :erlang.garbage_collect()

        # Persist the fully-enriched grid_data for every forecast hour
        # so (a) /weather can show forecast-hour data after a pod
        # restart and (b) point_detail can rebuild the factor
        # breakdown for a clicked cell at any forecast hour by
        # re-running the scorer against the stored profile.
        persist_profiles(grid_data, valid_time)

        # Weather map only shows the analysis hour — f01..f18 are
        # forecast data that /weather doesn't render. Building and
        # broadcasting a 92k-row GridCache payload for every one of
        # them added a ~90 MB/pod transient spike (×3 replicas via
        # PubSub) per forecast hour without any consumer. Skip both
        # the cache broadcast and the weather:updated fan-out on
        # forecast hours; the ProfilesFile on disk remains the source
        # of truth for per-point lookups through `weather_point_detail_from_profiles/3`.
        if forecast_hour == 0 do
          rows = Weather.build_grid_cache_rows(grid_data, valid_time)
          GridCache.broadcast_put(valid_time, rows)

          Phoenix.PubSub.broadcast(
            Microwaveprop.PubSub,
            "weather:updated",
            {:weather_updated, valid_time}
          )
        end

        scores = compute_scores(grid_data, valid_time, forecast_hour)

        case Propagation.replace_scores(scores, valid_time) do
          {:ok, count} ->
            Logger.info("PropagationGrid: #{label} → #{count} scores for #{valid_time}")
            warm_cache(valid_time)

            # Broadcast progress *after* persistence so the map's
            # pipeline chip only advances to "through +Nh" once that
            # hour is actually readable from the scores file. Emitting
            # this before the fetch would push the chip ahead of the
            # map by the full forecast-hour wall time (~10 minutes).
            Phoenix.PubSub.broadcast(
              Microwaveprop.PubSub,
              "propagation:pipeline",
              {:propagation_pipeline_progress, %{forecast_hour: forecast_hour, valid_time: valid_time}}
            )

            :ok

          error ->
            Logger.error("PropagationGrid: #{label} replace failed: #{inspect(error)}")
            error
        end

      error ->
        Logger.warning("PropagationGrid: #{label} fetch failed: #{inspect(error)}")
        error
    end
  end

  defp persist_profiles(grid_data, valid_time) do
    timed("profiles", fn ->
      try do
        ProfilesFile.write!(valid_time, grid_data)
      rescue
        e ->
          Logger.warning("PropagationGrid: profiles write failed: #{inspect(e)}")
      end
    end)
  end

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

    ScoreCache.prune_older_than(DateTime.add(DateTime.utc_now(), -2, :hour))
  end

  defp timed(label, fun) do
    t0 = System.monotonic_time(:millisecond)
    result = fun.()
    elapsed = System.monotonic_time(:millisecond) - t0
    Logger.info("PropagationGrid: #{label} took #{format_duration(elapsed)}")
    result
  end

  defp format_duration(ms) when ms < 1000, do: "#{ms}ms"
  defp format_duration(ms), do: "#{Float.round(ms / 1000, 1)}s"

  defp merge_native_duct_data(grid_data, run_time, forecast_hour) do
    hour_dt = HrrrClient.nearest_hrrr_hour(run_time)
    date = DateTime.to_date(hour_dt)
    hour = hour_dt.hour
    grid_spec = Grid.wgrib2_grid_spec()

    case timed("native", fn ->
           HrrrNativeClient.fetch_native_duct_grid(date, hour, grid_spec, forecast_hour)
         end) do
      {:ok, duct_grid} ->
        Logger.info("PropagationGrid: merged #{map_size(duct_grid)} native duct cells")
        apply_duct_grid(grid_data, duct_grid)

      {:error, reason} ->
        Logger.warning("PropagationGrid: native duct fetch failed (continuing without): #{inspect(reason)}")
        grid_data
    end
  end

  defp apply_duct_grid(grid_data, duct_grid) do
    Map.new(grid_data, fn {point, profile} ->
      case Map.get(duct_grid, point) do
        nil -> {point, profile}
        duct -> {point, Map.merge(profile, duct)}
      end
    end)
  end

  defp merge_commercial_link_data(grid_data, valid_time) do
    # 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

    merged
  end

  defp merge_nexrad_data(grid_data, valid_time) do
    points = Map.keys(grid_data)

    case timed("nexrad", fn -> NexradClient.fetch_frame(valid_time, points) end) do
      {:ok, observations} ->
        apply_nexrad_observations(grid_data, observations)

      {:error, reason} ->
        Logger.warning("PropagationGrid: NEXRAD fetch failed (continuing without): #{inspect(reason)}")
        grid_data
    end
  end

  defp apply_nexrad_observations(grid_data, observations) do
    index = Map.new(observations, fn obs -> {{obs.lat, obs.lon}, obs.max_reflectivity_dbz} end)
    non_zero = Enum.count(index, fn {_pt, dbz} -> dbz > 0 end)
    Logger.info("PropagationGrid: NEXRAD merged (#{non_zero} cells with precip)")

    Map.new(grid_data, fn {point, profile} ->
      case Map.get(index, point) do
        nil -> {point, profile}
        dbz -> {point, Map.put(profile, :nexrad_max_reflectivity_dbz, dbz)}
      end
    end)
  end

  defp compute_scores(grid_data, valid_time, forecast_hour) do
    # Algorithm is the primary scorer. `factors` is only populated for
    # f00 (the analysis hour) — forecast hours skip the JSONB write so
    # the scoring+upsert phase can land in under a minute instead of
    # ~4-5 minutes. point_detail on forecast hours returns a nil
    # breakdown, which the UI tolerates.
    compute_scores_algorithm(grid_data, valid_time, forecast_hour == 0)
  end

  defp compute_scores_algorithm(grid_data, valid_time, include_factors?) do
    grid_data
    |> Task.async_stream(
      &score_one_point(&1, valid_time, include_factors?),
      max_concurrency: System.schedulers_online() * 2,
      timeout: 30_000
    )
    |> Stream.flat_map(fn
      {:ok, results} -> results
      {:exit, _reason} -> []
    end)
  end

  defp score_one_point({{lat, lon}, profile}, valid_time, include_factors?) do
    band_scores = Propagation.score_grid_point(profile, valid_time, lat, lon)

    Enum.map(band_scores, fn r ->
      %{
        lat: lat,
        lon: lon,
        valid_time: valid_time,
        band_mhz: r.band_mhz,
        score: r.score,
        factors: if(include_factors?, do: r.factors)
      }
    end)
  end
end