prop/lib/microwaveprop/workers/hrrr_fetch_worker.ex

defmodule Microwaveprop.Workers.HrrrFetchWorker do
  @moduledoc false
  use Oban.Worker,
    queue: :hrrr,
    max_attempts: 20,
    unique: [period: :infinity, states: [:available, :scheduled, :executing, :retryable]]

  alias Microwaveprop.Weather
  alias Microwaveprop.Weather.HrrrClient
  alias Microwaveprop.Weather.SoundingParams

  require Logger

  @impl Oban.Worker
  def backoff(%Oban.Job{attempt: attempt}) do
    # Exponential backoff: 2m, 4m, 8m, 16m, 32m, 1h, 2h, 4h, 6h, 6h, ...
    min(120 * Integer.pow(2, attempt - 1), _six_hours = 21_600)
  end

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"points" => points, "valid_time" => valid_time_str}}) do
    # Batch mode: download GRIB2 once, extract multiple points
    {:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str)

    point_tuples =
      points
      |> Enum.map(fn %{"lat" => lat, "lon" => lon} -> {lat, lon} end)
      |> Enum.reject(fn {lat, lon} -> Weather.has_hrrr_profile?(lat, lon, valid_time) end)

    result =
      if point_tuples == [] do
        Logger.info("HRRR batch: all #{length(points)} points already exist for #{valid_time_str}")
        :ok
      else
        Logger.info("HRRR batch: fetching #{length(point_tuples)} points for #{valid_time_str}")
        fetch_and_store_batch(point_tuples, valid_time, valid_time_str)
      end

    # Reclaim the refc binary heap held from GRIB2 ranges before the
    # Oban producer hands this process its next job. Without this each
    # worker steady-states at 100+ MiB of retained HRRR payload.
    :erlang.garbage_collect(self())
    result
  end

  def perform(%Oban.Job{args: %{"lat" => raw_lat, "lon" => raw_lon, "valid_time" => valid_time_str}}) do
    # Legacy single-point mode
    {:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str)
    {lat, lon} = Weather.round_to_hrrr_grid(raw_lat, raw_lon)

    if Weather.has_hrrr_profile?(lat, lon, valid_time) do
      Logger.info("HRRR profile already exists for #{lat},#{lon} @ #{valid_time_str}")
      :ok
    else
      Logger.info("Fetching HRRR profile for #{lat},#{lon} @ #{valid_time_str}")

      case HrrrClient.fetch_profile(lat, lon, valid_time) do
        {:ok, data} ->
          store_profile(lat, lon, valid_time, data)
          broadcast_enrichment(lat, lon, valid_time)
          :ok

        {:error, reason} ->
          handle_error(reason, "#{lat},#{lon} @ #{valid_time_str}")
      end
    end
  end

  defp fetch_and_store_batch(point_tuples, valid_time, valid_time_str) do
    case HrrrClient.fetch_grid(point_tuples, valid_time) do
      {:ok, grid_data} ->
        profiles =
          Enum.map(grid_data, fn {{lat, lon}, data} ->
            build_profile_attrs(lat, lon, valid_time, data)
          end)

        Weather.upsert_hrrr_profiles_batch(profiles, skip_existing: true)

        Enum.each(grid_data, fn {{lat, lon}, _data} ->
          broadcast_enrichment(lat, lon, valid_time)
        end)

        Logger.info("HRRR batch: saved #{map_size(grid_data)} profiles for #{valid_time_str}")
        :ok

      {:error, reason} ->
        handle_error(reason, "batch @ #{valid_time_str}")
    end
  end

  defp store_profile(lat, lon, valid_time, data) do
    attrs = build_profile_attrs(lat, lon, valid_time, data)
    Weather.upsert_hrrr_profile(attrs)
  end

  defp build_profile_attrs(lat, lon, valid_time, data) do
    params = SoundingParams.derive(data.profile)

    maybe_add_derived_params(
      %{
        valid_time: valid_time,
        lat: lat,
        lon: lon,
        run_time: data.run_time,
        profile: data.profile,
        hpbl_m: data.hpbl_m,
        pwat_mm: data.pwat_mm,
        surface_temp_c: data.surface_temp_c,
        surface_dewpoint_c: data.surface_dewpoint_c,
        surface_pressure_mb: data.surface_pressure_mb
      },
      params
    )
  end

  defp handle_error(reason, label) do
    if transient_failure?(reason) do
      Logger.error("HRRR transient error for #{label}: #{inspect(reason)}")
      {:error, reason}
    else
      Logger.warning("HRRR permanent failure for #{label}: #{inspect(reason)}")
      {:cancel, reason}
    end
  end

  # Only retry on transient network/server errors — everything else is permanent
  defp transient_failure?(%{__exception__: true}), do: true
  defp transient_failure?("HRRR idx HTTP " <> status), do: server_error?(status)
  defp transient_failure?("HRRR grib HTTP " <> status), do: server_error?(status)
  defp transient_failure?(_), do: false

  defp server_error?(status) do
    case Integer.parse(status) do
      {code, _} when code in [429, 500, 502, 503, 504] -> true
      _ -> false
    end
  end

  defp maybe_add_derived_params(attrs, nil), do: attrs

  defp maybe_add_derived_params(attrs, params) do
    Map.merge(attrs, %{
      surface_refractivity: params.surface_refractivity,
      min_refractivity_gradient: params.min_refractivity_gradient,
      ducting_detected: params.ducting_detected,
      duct_characteristics: params.duct_characteristics
    })
  end

  defp broadcast_enrichment(lat, lon, valid_time) do
    Phoenix.PubSub.broadcast(
      Microwaveprop.PubSub,
      "contact_enrichment:hrrr",
      {:hrrr_ready, %{lat: lat, lon: lon, valid_time: valid_time}}
    )
  end
end